package serving

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/go-sql-driver/mysql"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
)

func TestServingEpochRepositorySchemaReadinessAgainstTiDB(t *testing.T) {
	tests := []struct {
		name                    string
		damage                  string
		crossDatabaseForeignKey bool
	}{
		{name: "complete schema"},
		{
			name: "non-binary column collation",
			damage: `ALTER TABLE _System_ConnectorServingEpoch
  MODIFY COLUMN ReceiptProtocolVersion VARCHAR(32)
  CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL`,
		},
		{
			name: "wrong timestamp default",
			damage: `ALTER TABLE _System_ConnectorServingEpoch
  MODIFY COLUMN CreatedDate DATETIME(6) NOT NULL
  DEFAULT '2000-01-01 00:00:00.000000'`,
		},
		{
			name: "timestamp on update",
			damage: `ALTER TABLE _System_ConnectorServingEpoch
  MODIFY COLUMN LastModifiedDate DATETIME(6) NOT NULL
  DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)`,
		},
		{
			name:   "missing scope-state index",
			damage: `ALTER TABLE _System_ConnectorServingEpoch DROP INDEX idx_connector_serving_epoch_scope_state`,
		},
		{
			name:   "unexpected unique index",
			damage: `ALTER TABLE _System_ConnectorServingEpoch ADD UNIQUE KEY ux_unreviewed_org (OrgId)`,
		},
		{
			name:   "missing restrictive parent foreign key",
			damage: `ALTER TABLE _System_ConnectorServingEpochBinding DROP FOREIGN KEY fk_connector_serving_epoch_binding_epoch`,
		},
		{
			name:                    "cross-database restrictive parent foreign key",
			crossDatabaseForeignKey: true,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, cleanup := openServingEpochTestTiDB(t)
			defer cleanup()
			createServingEpochTestTables(t, db)
			repository := newServingEpochTestRepository(t, db)

			if test.damage == "" && !test.crossDatabaseForeignKey {
				if err := repository.VerifySchema(t.Context()); err != nil {
					t.Fatalf("VerifySchema() error = %v", err)
				}
				return
			}

			foreignDatabase := ""
			if test.crossDatabaseForeignKey {
				foreignDatabase = fmt.Sprintf("nexuscrm_connector_serving_foreign_%d", time.Now().UnixNano())
				quotedForeignDatabase := "`" + foreignDatabase + "`"
				if _, err := db.ExecContext(t.Context(), "CREATE DATABASE "+quotedForeignDatabase); err != nil {
					t.Fatalf("create foreign ServingEpoch database: %v", err)
				}
				defer func() {
					_, _ = db.ExecContext(context.Background(),
						`ALTER TABLE _System_ConnectorServingEpochBinding DROP FOREIGN KEY fk_connector_serving_epoch_binding_epoch`)
					_, _ = db.ExecContext(context.Background(), "DROP DATABASE IF EXISTS "+quotedForeignDatabase)
				}()
				foreignParentDDL := fmt.Sprintf(`CREATE TABLE %s._System_ConnectorServingEpoch (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PRIMARY KEY (Id)
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`, quotedForeignDatabase)
				if _, err := db.ExecContext(t.Context(), foreignParentDDL); err != nil {
					t.Fatalf("create foreign ServingEpoch parent: %v", err)
				}
				if _, err := db.ExecContext(t.Context(),
					`ALTER TABLE _System_ConnectorServingEpochBinding DROP FOREIGN KEY fk_connector_serving_epoch_binding_epoch`); err != nil {
					t.Fatalf("drop local ServingEpoch foreign key: %v", err)
				}
				crossDatabaseForeignKeyDDL := fmt.Sprintf(`ALTER TABLE _System_ConnectorServingEpochBinding
  ADD CONSTRAINT fk_connector_serving_epoch_binding_epoch
  FOREIGN KEY (ServingEpochId) REFERENCES %s._System_ConnectorServingEpoch (Id)
  ON UPDATE RESTRICT ON DELETE RESTRICT`, quotedForeignDatabase)
				if _, err := db.ExecContext(t.Context(), crossDatabaseForeignKeyDDL); err != nil {
					t.Fatalf("add cross-database ServingEpoch foreign key: %v", err)
				}
			} else if _, err := db.ExecContext(t.Context(), test.damage); err != nil {
				t.Fatalf("damage fixture schema: %v", err)
			}
			if err := repository.VerifySchema(t.Context()); !errors.Is(err, ErrSchemaUnavailable) {
				t.Fatalf("VerifySchema() error = %v, want ErrSchemaUnavailable", err)
			} else if foreignDatabase != "" && !strings.Contains(err.Error(), foreignDatabase) {
				t.Fatalf("VerifySchema() error = %v, want foreign database %s", err, foreignDatabase)
			}
		})
	}
}

func TestServingEpochRepositoryConcurrentExactPrepareAgainstTiDB(t *testing.T) {
	const workers = 12

	db, cleanup := openServingEpochTestTiDB(t)
	defer cleanup()
	createServingEpochTestTables(t, db)
	forceServingEpochConstraintSessionsOff(t, db, workers)
	repository := newServingEpochTestRepository(t, db)
	candidate := servingEpochTestCandidate(t, "epoch-concurrent-exact")

	type result struct {
		epoch     PreparedEpoch
		duplicate bool
		err       error
	}
	start := make(chan struct{})
	results := make(chan result, workers)
	var group sync.WaitGroup
	for i := 0; i < workers; i++ {
		group.Add(1)
		go func() {
			defer group.Done()
			<-start
			epoch, duplicate, err := repository.Prepare(t.Context(), candidate)
			results <- result{epoch: epoch, duplicate: duplicate, err: err}
		}()
	}
	close(start)
	group.Wait()
	close(results)

	creators := 0
	replays := 0
	for result := range results {
		if result.err != nil {
			t.Errorf("Prepare() error = %v", result.err)
			continue
		}
		if !result.epoch.equal(candidate) {
			t.Error("Prepare() returned bytes that disagree with the exact candidate")
		}
		if result.duplicate {
			replays++
		} else {
			creators++
		}
	}
	if creators != 1 || replays != workers-1 {
		t.Fatalf("Prepare() outcomes = %d creators/%d replays, want 1/%d", creators, replays, workers-1)
	}
	var restoredSessionRaw string
	if err := db.QueryRowContext(t.Context(),
		"SELECT @@SESSION.tidb_constraint_check_in_place_pessimistic").Scan(&restoredSessionRaw); err != nil {
		t.Fatalf("read pooled SESSION constraint default after Prepare: %v", err)
	}
	if restored := canonicalServingEpochTiDBBoolean(
		t, "tidb_constraint_check_in_place_pessimistic", restoredSessionRaw,
	); restored != "OFF" {
		t.Fatalf("pooled SESSION constraint default after Prepare = %s, want restored OFF", restored)
	}
	loaded, found, err := repository.LoadPrepared(t.Context(), candidate.ID())
	if err != nil || !found || !loaded.equal(candidate) {
		t.Fatalf("LoadPrepared() = equal:%t found:%t err:%v", loaded.equal(candidate), found, err)
	}
}

func forceServingEpochConstraintSessionsOff(t *testing.T, db *sql.DB, sessionCount int) {
	t.Helper()
	const variable = "tidb_constraint_check_in_place_pessimistic"

	if sessionCount < 1 {
		t.Fatalf("session count = %d, want at least 1", sessionCount)
	}
	db.SetMaxOpenConns(sessionCount)
	db.SetMaxIdleConns(0)

	held := make([]*sql.Conn, 0, sessionCount)
	defer func() {
		for _, conn := range held {
			if conn != nil {
				_ = conn.Close()
			}
		}
	}()

	// Holding every allowed physical connection prevents database/sql from
	// reusing one while the remaining sessions are still being initialized.
	for index := 0; index < sessionCount; index++ {
		conn, err := db.Conn(t.Context())
		if err != nil {
			t.Fatalf("acquire TiDB session %d: %v", index, err)
		}
		held = append(held, conn)
		if _, err := conn.ExecContext(t.Context(), "SET SESSION "+variable+" = OFF"); err != nil {
			t.Fatalf("set SESSION %s=OFF on connection %d: %v", variable, index, err)
		}
		var sessionRaw string
		if err := conn.QueryRowContext(t.Context(), "SELECT @@SESSION."+variable).Scan(&sessionRaw); err != nil {
			t.Fatalf("read SESSION %s on connection %d: %v", variable, index, err)
		}
		if session := canonicalServingEpochTiDBBoolean(t, variable, sessionRaw); session != "OFF" {
			t.Fatalf("SESSION %s on connection %d = %s, want OFF before repository use", variable, index, session)
		}
	}

	db.SetMaxIdleConns(sessionCount)
	for index := range held {
		if err := held[index].Close(); err != nil {
			t.Fatalf("return TiDB session %d to bounded idle pool: %v", index, err)
		}
		held[index] = nil
	}
}

func canonicalServingEpochTiDBBoolean(t *testing.T, name, value string) string {
	t.Helper()
	switch strings.ToUpper(strings.TrimSpace(value)) {
	case "ON", "1":
		return "ON"
	case "OFF", "0":
		return "OFF"
	default:
		t.Fatalf("unexpected %s value %q", name, value)
		return ""
	}
}

func TestServingEpochRepositoryRejectsEveryIdentityConflictAgainstTiDB(t *testing.T) {
	db, cleanup := openServingEpochTestTiDB(t)
	defer cleanup()
	createServingEpochTestTables(t, db)
	repository := newServingEpochTestRepository(t, db)

	for index, test := range candidateIdentityAxisMutations() {
		t.Run(test.name, func(t *testing.T) {
			epochID := fmt.Sprintf("epoch-conflict-%02d", index)
			view := canonicalTestBindingView()
			l2 := canonicalTestL2Snapshot(t)
			base, err := NewPreparedEpoch(epochID, view, l2)
			if err != nil {
				t.Fatal(err)
			}
			prepareServingEpochTestCandidate(t, repository, base)

			mutatedID := epochID
			mutatedView := copyIntegrationBindingView(view)
			mutatedL2 := l2.copy()
			test.mutate(&mutatedID, &mutatedView, &mutatedL2)
			mutatedView.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(mutatedView)
			mutated, err := NewPreparedEpoch(mutatedID, mutatedView, mutatedL2)
			if err != nil {
				t.Fatalf("construct mutated candidate: %v", err)
			}

			if test.changesEpochID {
				persisted, duplicate, err := repository.Prepare(t.Context(), mutated)
				if err != nil || duplicate || !persisted.equal(mutated) {
					t.Fatalf("distinct epoch ID Prepare() = equal:%t duplicate:%t err:%v", persisted.equal(mutated), duplicate, err)
				}
				return
			}

			if _, _, err := repository.Prepare(t.Context(), mutated); !errors.Is(err, ErrIdentityConflict) {
				t.Fatalf("Prepare() error = %v, want ErrIdentityConflict", err)
			}
			loaded, found, err := repository.LoadPrepared(t.Context(), epochID)
			if err != nil || !found || !loaded.equal(base) {
				t.Fatalf("conflict changed durable candidate: equal:%t found:%t err:%v", loaded.equal(base), found, err)
			}
		})
	}
}

func TestServingEpochRepositoryUsesExactBinaryIDsAgainstTiDB(t *testing.T) {
	db, cleanup := openServingEpochTestTiDB(t)
	defer cleanup()
	createServingEpochTestTables(t, db)
	repository := newServingEpochTestRepository(t, db)

	upper := servingEpochTestCandidate(t, "Epoch-Case-Exact")
	lower := servingEpochTestCandidate(t, "epoch-case-exact")
	prepareServingEpochTestCandidate(t, repository, upper)
	prepareServingEpochTestCandidate(t, repository, lower)

	for _, candidate := range []PreparedEpoch{upper, lower} {
		loaded, found, err := repository.LoadPrepared(t.Context(), candidate.ID())
		if err != nil || !found || !loaded.equal(candidate) {
			t.Fatalf("LoadPrepared(%q) = equal:%t found:%t err:%v", candidate.ID(), loaded.equal(candidate), found, err)
		}
	}
	if loaded, found, err := repository.LoadPrepared(t.Context(), "EPOCH-CASE-EXACT"); err != nil || found {
		t.Fatalf("case-folded LoadPrepared() = %#v found:%t err:%v, want absent", loaded, found, err)
	}
	var count int
	if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_ConnectorServingEpoch`).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 2 {
		t.Fatalf("case-distinct durable epoch count = %d, want 2", count)
	}
}

func TestServingEpochRepositoryRejectsCorruptStoredRowsAgainstTiDB(t *testing.T) {
	db, cleanup := openServingEpochTestTiDB(t)
	defer cleanup()
	createServingEpochTestTables(t, db)
	repository := newServingEpochTestRepository(t, db)

	tests := []struct {
		name   string
		mutate func(*testing.T, *sql.DB, PreparedEpoch)
	}{
		{
			name: "missing binding ordinal",
			mutate: func(t *testing.T, db *sql.DB, epoch PreparedEpoch) {
				execServingEpochCorruption(t, db,
					`DELETE FROM _System_ConnectorServingEpochBinding WHERE ServingEpochId = ? AND BindingOrdinal = 0`, epoch.ID())
			},
		},
		{
			name: "ordinal gap",
			mutate: func(t *testing.T, db *sql.DB, epoch PreparedEpoch) {
				execServingEpochCorruption(t, db,
					`UPDATE _System_ConnectorServingEpochBinding SET BindingOrdinal = 9 WHERE ServingEpochId = ? AND BindingOrdinal = 1`, epoch.ID())
			},
		},
		{
			name: "pack reference",
			mutate: func(t *testing.T, db *sql.DB, epoch PreparedEpoch) {
				execServingEpochCorruption(t, db,
					`UPDATE _System_ConnectorServingEpochBinding SET OwnerPack = 'wrong-pack' WHERE ServingEpochId = ? AND BindingOrdinal = 0`, epoch.ID())
			},
		},
		{
			name: "credential reference bytes",
			mutate: func(t *testing.T, db *sql.DB, epoch PreparedEpoch) {
				execServingEpochCorruption(t, db,
					`UPDATE _System_ConnectorServingEpochBinding SET CredentialRefs = ? WHERE ServingEpochId = ? AND BindingOrdinal = 0`, []byte("not-canonical"), epoch.ID())
			},
		},
		{
			name: "partial Runtime reference",
			mutate: func(t *testing.T, db *sql.DB, epoch PreparedEpoch) {
				execServingEpochCorruption(t, db,
					`UPDATE _System_ConnectorServingEpochBinding SET RuntimeDefinitionRef = NULL WHERE ServingEpochId = ? AND BindingOrdinal = 1`, epoch.ID())
			},
		},
		{
			name: "Catalog lineage",
			mutate: func(t *testing.T, db *sql.DB, epoch PreparedEpoch) {
				execServingEpochCorruption(t, db,
					`UPDATE _System_ConnectorServingEpoch SET SourceSnapshotDigest = ? WHERE Id = ?`, "sha256:"+strings.Repeat("0", 64), epoch.ID())
			},
		},
		{
			name: "L2 receipt",
			mutate: func(t *testing.T, db *sql.DB, epoch PreparedEpoch) {
				execServingEpochCorruption(t, db,
					`UPDATE _System_ConnectorServingEpoch SET L2VerificationReceiptDigest = ? WHERE Id = ?`, "sha256:"+strings.Repeat("0", 64), epoch.ID())
			},
		},
		{
			name: "non-prepared state",
			mutate: func(t *testing.T, db *sql.DB, epoch PreparedEpoch) {
				execServingEpochCorruption(t, db,
					`UPDATE _System_ConnectorServingEpoch SET State = 'active' WHERE Id = ?`, epoch.ID())
			},
		},
		{
			name: "non-NULL active scope",
			mutate: func(t *testing.T, db *sql.DB, epoch PreparedEpoch) {
				execServingEpochCorruption(t, db,
					`UPDATE _System_ConnectorServingEpoch SET ActiveScopeDigest = ? WHERE Id = ?`, strings.Repeat("a", 64), epoch.ID())
			},
		},
	}

	for index, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			candidate := servingEpochTestCandidate(t, fmt.Sprintf("epoch-corrupt-%02d", index))
			prepareServingEpochTestCandidate(t, repository, candidate)
			test.mutate(t, db, candidate)
			if _, _, err := repository.LoadPrepared(t.Context(), candidate.ID()); !errors.Is(err, ErrStoredCandidateInvalid) {
				t.Fatalf("LoadPrepared() error = %v, want ErrStoredCandidateInvalid", err)
			}
		})
	}
}

func TestServingEpochRepositoryRollsBackAllRowsOnSecondBindingCollisionAgainstTiDB(t *testing.T) {
	db, cleanup := openServingEpochTestTiDB(t)
	defer cleanup()
	createServingEpochTestTables(t, db)
	repository := newServingEpochTestRepository(t, db)

	target := servingEpochTestCandidate(t, "epoch-rollback-target")
	hostile := servingEpochTestCandidate(t, "epoch-rollback-hostile")
	prepareServingEpochTestCandidate(t, repository, hostile)
	if _, err := db.ExecContext(t.Context(), `
UPDATE _System_ConnectorServingEpochBinding
   SET Id = ?
 WHERE ServingEpochId = ? AND BindingOrdinal = 0`, target.bindings[1].id, hostile.ID()); err != nil {
		t.Fatalf("install hostile second-binding collision: %v", err)
	}

	if _, _, err := repository.Prepare(t.Context(), target); !errors.Is(err, ErrIdentityConflict) {
		t.Fatalf("Prepare() error = %v, want ErrIdentityConflict for hostile second-binding collision", err)
	}
	var parentCount, bindingCount int
	if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_ConnectorServingEpoch WHERE Id = ?`, target.ID()).Scan(&parentCount); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_ConnectorServingEpochBinding WHERE ServingEpochId = ?`, target.ID()).Scan(&bindingCount); err != nil {
		t.Fatal(err)
	}
	if parentCount != 0 || bindingCount != 0 {
		t.Fatalf("failed Prepare() retained %d parent/%d binding rows, want zero/zero", parentCount, bindingCount)
	}
	if _, found, err := repository.LoadPrepared(t.Context(), target.ID()); err != nil || found {
		t.Fatalf("LoadPrepared(failed candidate) = found:%t err:%v, want absent", found, err)
	}
}

func TestServingEpochTablesEnforcePhysicalConstraintsAgainstTiDB(t *testing.T) {
	db, cleanup := openServingEpochTestTiDB(t)
	defer cleanup()
	createServingEpochTestTables(t, db)
	repository := newServingEpochTestRepository(t, db)

	first := servingEpochTestCandidate(t, "epoch-physical-first")
	second := servingEpochTestCandidate(t, "epoch-physical-second")
	prepareServingEpochTestCandidate(t, repository, first)
	prepareServingEpochTestCandidate(t, repository, second)

	expectServingEpochConstraintFailure(t, "epoch digest uniqueness", 1062, func() error {
		_, err := db.ExecContext(t.Context(), `UPDATE _System_ConnectorServingEpoch SET EpochDigest = ? WHERE Id = ?`, first.EpochDigest(), second.ID())
		return err
	})
	activeScope := strings.Repeat("f", 64)
	if _, err := db.ExecContext(t.Context(), `UPDATE _System_ConnectorServingEpoch SET ActiveScopeDigest = ? WHERE Id = ?`, activeScope, first.ID()); err != nil {
		t.Fatalf("install first active-scope key: %v", err)
	}
	expectServingEpochConstraintFailure(t, "active-scope uniqueness", 1062, func() error {
		_, err := db.ExecContext(t.Context(), `UPDATE _System_ConnectorServingEpoch SET ActiveScopeDigest = ? WHERE Id = ?`, activeScope, second.ID())
		return err
	})
	expectServingEpochConstraintFailure(t, "binding ordinal uniqueness", 1062, func() error {
		_, err := db.ExecContext(t.Context(), `
UPDATE _System_ConnectorServingEpochBinding
   SET BindingOrdinal = 0
 WHERE ServingEpochId = ? AND BindingOrdinal = 1`, first.ID())
		return err
	})
	expectServingEpochConstraintFailure(t, "binding identity uniqueness", 1062, func() error {
		_, err := db.ExecContext(t.Context(), `
UPDATE _System_ConnectorServingEpochBinding
   SET BindingIdentityDigest = ?
 WHERE ServingEpochId = ? AND BindingOrdinal = 1`, first.bindings[0].identityDigest, first.ID())
		return err
	})
	expectServingEpochConstraintFailure(t, "orphan binding foreign key", 1452, func() error {
		_, err := db.ExecContext(t.Context(), `
INSERT INTO _System_ConnectorServingEpochBinding (
  Id, ServingEpochId, BindingOrdinal, BindingIdentityDigest, Kind,
  OwnerPack, PackInstallationId, PackPayloadDigest, BusinessKey,
  DefinitionVersion, CredentialRefs, RuntimeCommandKind,
  RuntimePayloadFormatVersion, RuntimeDefinitionRef, RuntimeDefinitionVersion,
  CreatedDate, LastModifiedDate
)
SELECT ?, ?, 99, BindingIdentityDigest, Kind,
       OwnerPack, PackInstallationId, PackPayloadDigest, BusinessKey,
       DefinitionVersion, CredentialRefs, RuntimeCommandKind,
       RuntimePayloadFormatVersion, RuntimeDefinitionRef, RuntimeDefinitionVersion,
       CreatedDate, LastModifiedDate
  FROM _System_ConnectorServingEpochBinding
 WHERE ServingEpochId = ?
 LIMIT 1`, "orphan-binding", "missing-serving-epoch", first.ID())
		return err
	})
	expectServingEpochConstraintFailure(t, "restrictive parent delete", 1451, func() error {
		_, err := db.ExecContext(t.Context(), `DELETE FROM _System_ConnectorServingEpoch WHERE Id = ?`, second.ID())
		return err
	})
}

func servingEpochTestCandidate(t *testing.T, epochID string) PreparedEpoch {
	t.Helper()
	candidate, err := NewPreparedEpoch(epochID, canonicalTestBindingView(), canonicalTestL2Snapshot(t))
	if err != nil {
		t.Fatalf("NewPreparedEpoch(%q): %v", epochID, err)
	}
	return candidate
}

func newServingEpochTestRepository(t *testing.T, db *sql.DB) *Repository {
	t.Helper()
	repository, err := NewRepository(db)
	if err != nil {
		t.Fatalf("NewRepository(): %v", err)
	}
	return repository
}

func prepareServingEpochTestCandidate(t *testing.T, repository *Repository, candidate PreparedEpoch) {
	t.Helper()
	persisted, duplicate, err := repository.Prepare(t.Context(), candidate)
	if err != nil || duplicate || !persisted.equal(candidate) {
		t.Fatalf("Prepare(%q) = equal:%t duplicate:%t err:%v", candidate.ID(), persisted.equal(candidate), duplicate, err)
	}
}

func execServingEpochCorruption(t *testing.T, db *sql.DB, statement string, args ...any) {
	t.Helper()
	result, err := db.ExecContext(t.Context(), statement, args...)
	if err != nil {
		t.Fatalf("corrupt ServingEpoch fixture: %v", err)
	}
	rows, err := result.RowsAffected()
	if err != nil {
		t.Fatalf("corrupt ServingEpoch fixture rows: %v", err)
	}
	if rows != 1 {
		t.Fatalf("corrupt ServingEpoch fixture affected %d rows, want 1", rows)
	}
}

func expectServingEpochConstraintFailure(t *testing.T, name string, mysqlCode uint16, operation func() error) {
	t.Helper()
	err := operation()
	var mysqlError *mysql.MySQLError
	if !errors.As(err, &mysqlError) || mysqlError.Number != mysqlCode {
		t.Fatalf("%s error = %v, want MySQL %d", name, err, mysqlCode)
	}
}

func createServingEpochTestTables(t *testing.T, db *sql.DB) {
	t.Helper()
	for _, ddl := range []string{
		packGenerationOwnerTestDDL,
		packGenerationBindingDefinitionSetOwnerTestDDL,
		credentialGenerationOwnerTestDDL,
		servingEpochTestDDL,
		servingEpochBindingTestDDL,
		servingEpochReproducibilityTestDDL,
		servingEpochBindingOwnerTestDDL,
		servingEpochCredentialGenerationTestDDL,
	} {
		if _, err := db.ExecContext(t.Context(), ddl); err != nil {
			t.Fatalf("create ServingEpoch fixture schema: %v", err)
		}
	}
}

// The three owner tables are deliberately minimal parent fixtures. Serving
// verifies and writes only the five catalog-owned ServingEpoch tables; these
// exact parent keys make TiDB enforce the real foreign-key and collation
// boundaries without duplicating the owner repositories in this package.
const packGenerationOwnerTestDDL = `CREATE TABLE _System_PackGeneration (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  PRIMARY KEY (Id)
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`

const packGenerationBindingDefinitionSetOwnerTestDDL = `CREATE TABLE _System_PackGenerationConnectorBindingDefinitionSet (
  PackGenerationId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  PRIMARY KEY (PackGenerationId),
  CONSTRAINT fk_pack_generation_binding_definition_set_generation
    FOREIGN KEY (PackGenerationId) REFERENCES _System_PackGeneration (Id)
    ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`

const credentialGenerationOwnerTestDDL = `CREATE TABLE _System_ConnectorCredentialGeneration (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PRIMARY KEY (Id)
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`

const servingEpochTestDDL = `CREATE TABLE _System_ConnectorServingEpoch (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  EpochDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ServingScopeDigest CHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ActiveScopeDigest CHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  State VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT 'prepared',
  OrgId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  AppInstallationId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  AuthorizedWorkloadId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  CatalogRevision DECIMAL(20,0) NOT NULL,
  SchemaGeneration DECIMAL(20,0) NOT NULL,
  EngineGeneration DECIMAL(20,0) NOT NULL,
  SourceSnapshotDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  BindingViewDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ReceiptProtocolVersion VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ReceiptCapabilityVersion VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  SecurityEpoch DECIMAL(20,0) NOT NULL,
  L2ConfigurationGenerationRef VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  L2ConfigurationDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  L2VerificationReceipt MEDIUMBLOB NOT NULL,
  L2VerificationReceiptDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  PRIMARY KEY (Id),
  UNIQUE KEY ux_connector_serving_epoch_digest (EpochDigest),
  UNIQUE KEY ux_connector_serving_epoch_active_scope (ActiveScopeDigest),
  KEY idx_connector_serving_epoch_scope_state (ServingScopeDigest, State, Id)
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`

const servingEpochBindingTestDDL = `CREATE TABLE _System_ConnectorServingEpochBinding (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ServingEpochId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  BindingOrdinal INT NOT NULL,
  BindingIdentityDigest CHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  Kind VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  OwnerPack VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PackInstallationId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PackPayloadDigest CHAR(85) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  BusinessKey VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  DefinitionVersion DECIMAL(20,0) NOT NULL,
  CredentialRefs MEDIUMBLOB NOT NULL,
  RuntimeCommandKind VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  RuntimePayloadFormatVersion BIGINT NULL,
  RuntimeDefinitionRef VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  RuntimeDefinitionVersion VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  PRIMARY KEY (Id),
  UNIQUE KEY ux_connector_epoch_binding_ordinal (ServingEpochId, BindingOrdinal),
  UNIQUE KEY ux_connector_epoch_binding_identity (ServingEpochId, BindingIdentityDigest),
  KEY idx_connector_epoch_binding_pack (ServingEpochId, PackInstallationId, BindingOrdinal),
  CONSTRAINT fk_connector_serving_epoch_binding_epoch
    FOREIGN KEY (ServingEpochId) REFERENCES _System_ConnectorServingEpoch (Id)
    ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`

const servingEpochReproducibilityTestDDL = `CREATE TABLE _System_ConnectorServingEpochReproducibility (
  ServingEpochId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ConnectorDigestProfileId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ServiceConfigurationGenerationProtocolVersion VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ServiceConfigurationReceiptProtocolVersion VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  PRIMARY KEY (ServingEpochId),
  KEY idx_connector_epoch_repro_profile (ConnectorDigestProfileId, ServingEpochId),
  CONSTRAINT fk_connector_serving_epoch_repro_epoch
    FOREIGN KEY (ServingEpochId) REFERENCES _System_ConnectorServingEpoch (Id)
    ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`

const servingEpochBindingOwnerTestDDL = `CREATE TABLE _System_ConnectorServingEpochBindingOwner (
  ServingEpochBindingId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PackGenerationId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  DefinitionEnvelopeVersion VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  DefinitionReceiptVersion VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  CanonicalDefinitionDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  DefinitionOwnerReceipt MEDIUMBLOB NOT NULL,
  DefinitionOwnerReceiptDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  PRIMARY KEY (ServingEpochBindingId),
  KEY idx_connector_epoch_binding_owner_pack (PackGenerationId, ServingEpochBindingId),
  CONSTRAINT fk_connector_serving_epoch_binding_owner_binding
    FOREIGN KEY (ServingEpochBindingId) REFERENCES _System_ConnectorServingEpochBinding (Id)
    ON UPDATE RESTRICT ON DELETE RESTRICT,
  CONSTRAINT fk_connector_serving_epoch_binding_owner_pack_generation
    FOREIGN KEY (PackGenerationId) REFERENCES _System_PackGenerationConnectorBindingDefinitionSet (PackGenerationId)
    ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`

const servingEpochCredentialGenerationTestDDL = `CREATE TABLE _System_ConnectorServingEpochCredentialGeneration (
  ServingEpochId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  CredentialOrdinal INT NOT NULL,
  CredentialGenerationId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ReceiptProtocolVersion VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  TransportConfigurationDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  OwnerReceipt MEDIUMBLOB NOT NULL,
  OwnerReceiptDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  PRIMARY KEY (ServingEpochId, CredentialOrdinal),
  UNIQUE KEY ux_connector_epoch_credential_generation (ServingEpochId, CredentialGenerationId),
  KEY idx_connector_epoch_credential_generation_ref (CredentialGenerationId, ServingEpochId),
  CONSTRAINT fk_connector_serving_epoch_credential_generation_epoch
    FOREIGN KEY (ServingEpochId) REFERENCES _System_ConnectorServingEpoch (Id)
    ON UPDATE RESTRICT ON DELETE RESTRICT,
  CONSTRAINT fk_connector_serving_epoch_credential_generation_owner
    FOREIGN KEY (CredentialGenerationId) REFERENCES _System_ConnectorCredentialGeneration (Id)
    ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`

func openServingEpochTestTiDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host := servingEpochEnvironmentOr("TIDB_HOST", "127.0.0.1")
	port := servingEpochEnvironmentOr("TIDB_PORT", "4001")
	user := servingEpochEnvironmentOr("TIDB_USER", "root")
	password := os.Getenv("TIDB_PASSWORD")
	tlsParameter := ""
	switch strings.ToLower(strings.TrimSpace(os.Getenv("TIDB_TLS"))) {
	case "1", "true", "yes", "require", "verify-full":
		tlsParameter = "&tls=true"
	}
	admin, err := sql.Open("mysql", fmt.Sprintf(
		"%s:%s@tcp(%s:%s)/?parseTime=true&loc=UTC&timeout=2s%s",
		user, password, host, port, tlsParameter,
	))
	if err != nil {
		unavailableServingEpochTiDB(t, "open admin connection: %v", err)
	}
	if err := admin.PingContext(context.Background()); err != nil {
		_ = admin.Close()
		unavailableServingEpochTiDB(t, "TiDB not reachable at %s:%s: %v", host, port, err)
	}

	databaseName := fmt.Sprintf("nexuscrm_connector_serving_%d", time.Now().UnixNano())
	if _, err := admin.ExecContext(t.Context(), "CREATE DATABASE `"+databaseName+"`"); err != nil {
		_ = admin.Close()
		unavailableServingEpochTiDB(t, "create temporary database: %v", err)
	}
	db, err := sql.Open("mysql", fmt.Sprintf(
		"%s:%s@tcp(%s:%s)/%s?parseTime=true&loc=UTC&timeout=2s%s",
		user, password, host, port, databaseName, tlsParameter,
	))
	if err != nil {
		_, _ = admin.ExecContext(context.Background(), "DROP DATABASE IF EXISTS `"+databaseName+"`")
		_ = admin.Close()
		t.Fatalf("open ServingEpoch test database: %v", err)
	}
	db.SetMaxOpenConns(32)
	if err := db.PingContext(t.Context()); err != nil {
		_ = db.Close()
		_, _ = admin.ExecContext(context.Background(), "DROP DATABASE IF EXISTS `"+databaseName+"`")
		_ = admin.Close()
		t.Fatalf("ping ServingEpoch test database: %v", err)
	}
	cleanup := func() {
		_ = db.Close()
		_, _ = admin.ExecContext(context.Background(), "DROP DATABASE IF EXISTS `"+databaseName+"`")
		_ = admin.Close()
	}
	return db, cleanup
}

func unavailableServingEpochTiDB(t *testing.T, format string, args ...any) {
	t.Helper()
	if os.Getenv("NEXUSCRM_TIDB_REQUIRED") == "1" {
		t.Fatalf("required TiDB integration unavailable: "+format, args...)
	}
	t.Skipf("skipping TiDB integration: "+format, args...)
}

func servingEpochEnvironmentOr(name, fallback string) string {
	if value := strings.TrimSpace(os.Getenv(name)); value != "" {
		return value
	}
	return fallback
}