package persistence

import (
	"context"
	"database/sql"
	"errors"
	"strings"
	"testing"

	"github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
)

func TestIntegration_ConnectorV2ServingSchemaEnforcesEpochAndGateIdentity(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()

	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	byName := make(map[string]schema.TableDefinition, len(definitions))
	for _, definition := range definitions {
		byName[definition.TableName] = definition
	}

	repository := NewSchemaRepository(db, nil)
	tableNames := []string{
		"_System_ConnectorServingEpoch",
		"_System_ConnectorServingEpochBinding",
		"_System_ConnectorAccountServingGate",
	}
	// Run the idempotent path too: the old FK normalizer rewrote only the
	// already-existing parent on pass one and then rewrote the child on pass two.
	for pass := 1; pass <= 2; pass++ {
		for _, tableName := range tableNames {
			definition, ok := byName[tableName]
			if !ok {
				t.Fatalf("%s absent from canonical schema catalog", tableName)
			}
			if err := repository.CreatePhysicalTable(context.Background(), definition); err != nil {
				t.Fatalf("CreatePhysicalTable(%s), pass %d: %v", tableName, pass, err)
			}
		}
	}
	for _, participant := range []struct{ table, column string }{
		{"_System_ConnectorServingEpoch", "Id"},
		{"_System_ConnectorServingEpochBinding", "ServingEpochId"},
		{"_System_ConnectorAccountServingGate", "ServingEpochId"},
	} {
		requireConnectorServingFKColumnShape(t, db, participant.table, participant.column)
	}

	const (
		epochID = "epoch-1"
		scope   = "1111111111111111111111111111111111111111111111111111111111111111"
	)
	insertEpoch := `INSERT INTO _System_ConnectorServingEpoch (
		Id, EpochDigest, ServingScopeDigest, ActiveScopeDigest, State,
		OrgId, AppInstallationId, AuthorizedWorkloadId,
		CatalogRevision, SchemaGeneration, EngineGeneration,
		SourceSnapshotDigest, BindingViewDigest,
		ReceiptProtocolVersion, ReceiptCapabilityVersion, SecurityEpoch,
		L2ConfigurationGenerationRef, L2ConfigurationDigest,
		L2VerificationReceipt, L2VerificationReceiptDigest
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	digest := func(value string) string {
		return "sha256:" + strings.Repeat(value, 64)
	}
	packPayloadDigest := "nexus-pack-sha256-v1:" + strings.Repeat("1", 64)
	epochArgs := func(id, epochDigest string, activeScope any) []any {
		return []any{
			id, epochDigest, scope, activeScope, "active",
			"org-1", "installation-1", "workload-1",
			uint64(11), uint64(12), uint64(13),
			digest("a"), digest("b"),
			"1.0", "catalog.consumer.integration-binding-view.v1", uint64(14),
			"l2-generation-1", digest("c"),
			[]byte("l2-verification-receipt-v1"), digest("d"),
		}
	}
	if _, err := db.Exec(insertEpoch, epochArgs(epochID, digest("e"), scope)...); err != nil {
		t.Fatalf("insert first active ServingEpoch: %v", err)
	}
	if _, err := db.Exec(insertEpoch, epochArgs("Epoch-1", digest("g"), nil)...); err != nil {
		t.Fatalf("insert byte-distinct ServingEpoch Id: %v", err)
	}
	if _, err := db.Exec(insertEpoch, epochArgs("epoch-same-digest", digest("e"), nil)...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate EpochDigest error = %v, want duplicate-key rejection", err)
	}
	if _, err := db.Exec(insertEpoch, epochArgs("epoch-second-active", digest("f"), scope)...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate ActiveScopeDigest error = %v, want duplicate-key rejection", err)
	}

	insertBinding := `INSERT INTO _System_ConnectorServingEpochBinding (
		Id, ServingEpochId, BindingOrdinal, BindingIdentityDigest,
		Kind, OwnerPack, PackInstallationId, PackPayloadDigest,
		BusinessKey, DefinitionVersion, CredentialRefs
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	bindingArgs := func(id, servingEpochID string, ordinal int, identityDigest string) []any {
		return []any{
			id, servingEpochID, ordinal, identityDigest,
			"CONNECTOR_INBOUND_ROUTE", "connector-pack", "pack-installation-1", packPayloadDigest,
			"route-1", uint64(1), []byte("credential-a\x00credential-b"),
		}
	}
	if _, err := db.Exec(insertBinding, bindingArgs("binding-1", epochID, 0, strings.Repeat("2", 64))...); err != nil {
		t.Fatalf("insert first ServingEpoch binding: %v", err)
	}
	var storedPackPayloadDigest string
	var storedPackPayloadDigestBytes int
	if err := db.QueryRow(
		`SELECT PackPayloadDigest, OCTET_LENGTH(PackPayloadDigest)
		 FROM _System_ConnectorServingEpochBinding WHERE Id = ?`,
		"binding-1",
	).Scan(&storedPackPayloadDigest, &storedPackPayloadDigestBytes); err != nil {
		t.Fatalf("read exact Catalog pack payload digest: %v", err)
	}
	if storedPackPayloadDigest != packPayloadDigest || storedPackPayloadDigestBytes != 85 {
		t.Fatalf(
			"stored PackPayloadDigest = %q/%d bytes, want %q/85 bytes",
			storedPackPayloadDigest,
			storedPackPayloadDigestBytes,
			packPayloadDigest,
		)
	}
	if _, err := db.Exec(insertBinding, bindingArgs("binding-case-distinct-epoch", "Epoch-1", 0, strings.Repeat("6", 64))...); err != nil {
		t.Fatalf("insert exact binding for byte-distinct ServingEpoch: %v", err)
	}
	if _, err := db.Exec(insertBinding, bindingArgs("binding-ordinal-conflict", epochID, 0, strings.Repeat("3", 64))...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate binding ordinal error = %v, want duplicate-key rejection", err)
	}
	if _, err := db.Exec(insertBinding, bindingArgs("binding-identity-conflict", epochID, 1, strings.Repeat("2", 64))...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate binding identity error = %v, want duplicate-key rejection", err)
	}
	_, err = db.Exec(insertBinding, bindingArgs("binding-missing-epoch", "missing-epoch", 0, strings.Repeat("4", 64))...)
	requireConnectorServingForeignKeyError(t, err, "binding for a missing ServingEpoch")
	_, err = db.Exec(insertBinding, bindingArgs("binding-case-mismatch", "EPOCH-1", 0, strings.Repeat("5", 64))...)
	requireConnectorServingForeignKeyError(t, err, "binding for an unpersisted case variant")

	insertGate := `INSERT INTO _System_ConnectorAccountServingGate (
		Id, ServingScopeDigest, ConnectorAccountId, ServingEpochId, ServingEpochDigest
	) VALUES (?, ?, ?, ?, ?)`
	if _, err := db.Exec(insertGate, "gate-1", scope, "account-1", epochID, digest("e")); err != nil {
		t.Fatalf("insert first AccountServingGate: %v", err)
	}
	if _, err := db.Exec(insertGate, "gate-duplicate-account", scope, "account-1", epochID, digest("e")); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate scoped account gate error = %v, want duplicate-key rejection", err)
	}
	if _, err := db.Exec(insertGate, "gate-case-distinct-account", scope, "Account-1", epochID, digest("e")); err != nil {
		t.Fatalf("byte-distinct scoped account gate was collapsed by collation: %v", err)
	}
	_, err = db.Exec(insertGate, "gate-missing-epoch", scope, "account-2", "missing-epoch", digest("f"))
	requireConnectorServingForeignKeyError(t, err, "gate for a missing ServingEpoch")
}

func requireConnectorServingFKColumnShape(t *testing.T, db interface {
	QueryRow(query string, args ...any) *sql.Row
}, tableName, columnName string) {
	t.Helper()

	var columnType, collation string
	if err := db.QueryRow(
		`SELECT COLUMN_TYPE, COLLATION_NAME
		 FROM INFORMATION_SCHEMA.COLUMNS
		 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`,
		tableName,
		columnName,
	).Scan(&columnType, &collation); err != nil {
		t.Fatalf("inspect %s.%s physical shape: %v", tableName, columnName, err)
	}
	if columnType != "varchar(255)" || collation != "utf8mb4_bin" {
		t.Fatalf("%s.%s physical shape = %s/%s, want varchar(255)/utf8mb4_bin", tableName, columnName, columnType, collation)
	}
}

func requireConnectorServingForeignKeyError(t *testing.T, err error, operation string) {
	t.Helper()
	if err == nil {
		t.Fatalf("%s succeeded, want foreign-key rejection", operation)
	}
	var mysqlErr *mysql.MySQLError
	if !errors.As(err, &mysqlErr) || mysqlErr.Number != 1452 {
		t.Fatalf("%s error = %v, want MySQL 1452 foreign-key rejection", operation, err)
	}
}