package controlplane

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

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/pingcap/tidb/pkg/parser"
)

func TestVerifyConnectorWorkItemOnlyExecutionCutover_AlreadyAppliedSkips(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverReceiptAttested(mock)

	if err := VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db); err != nil {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover (already applied) = error %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestWorkItemCutoverReceiptUsesDistinctAttestedRevision(t *testing.T) {
	if migrationIDVerifyConnectorWorkItemOnlyExecutionCutover == "20260724_verify_connector_workitem_only_execution_cutover" {
		t.Fatal("cutover receipt must not reuse the pre-terminalization migration id")
	}
	if migrationVersionVerifyConnectorWorkItemOnlyExecutionCutover != 2 {
		t.Fatalf("receipt version = %d, want 2", migrationVersionVerifyConnectorWorkItemOnlyExecutionCutover)
	}
	if !strings.Contains(migrationDescriptionVerifyConnectorWorkItemOnlyExecutionCutover, "terminalizing Event-owned backlog") {
		t.Fatalf("receipt description = %q, want terminalization attestation", migrationDescriptionVerifyConnectorWorkItemOnlyExecutionCutover)
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_RejectsStaleReceiptBeforeSchemaCheck(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	mock.ExpectQuery(`(?s)SELECT Version, Description\s+FROM _System_Migration\s+WHERE Id = \?\s+LIMIT 1`).
		WithArgs(migrationIDVerifyConnectorWorkItemOnlyExecutionCutover).
		WillReturnRows(sqlmock.NewRows([]string{"Version", "Description"}).AddRow(
			migrationVersionVerifyConnectorWorkItemOnlyExecutionCutover-1,
			migrationDescriptionVerifyConnectorWorkItemOnlyExecutionCutover,
		))

	err = VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db)
	if err == nil || !strings.Contains(err.Error(), "stale or malformed handoff receipt") {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover error = %v, want stale receipt rejection", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestRecordWorkItemCutoverMigrationApplied_RejectsMismatchedReadback(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	mock.ExpectBegin()
	tx, err := db.Begin()
	if err != nil {
		t.Fatalf("db.Begin: %v", err)
	}
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_Migration")).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectQuery(`(?s)SELECT Version, Description\s+FROM _System_Migration\s+WHERE Id = \?\s+LIMIT 1`).
		WithArgs(migrationIDVerifyConnectorWorkItemOnlyExecutionCutover).
		WillReturnRows(sqlmock.NewRows([]string{"Version", "Description"}).AddRow(
			migrationVersionVerifyConnectorWorkItemOnlyExecutionCutover-1,
			migrationDescriptionVerifyConnectorWorkItemOnlyExecutionCutover,
		))
	mock.ExpectRollback()

	err = recordWorkItemCutoverMigrationApplied(context.Background(), tx)
	if err == nil || !strings.Contains(err.Error(), "stale or malformed handoff receipt") {
		t.Fatalf("recordWorkItemCutoverMigrationApplied error = %v, want mismatched receipt rejection", err)
	}
	if err := tx.Rollback(); err != nil {
		t.Fatalf("tx.Rollback: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestConnectorIngressPackCutoverLockIncludesRetiredRoutes(t *testing.T) {
	if strings.Contains(connectorIngressPackCutoverLockSQL, "IsCurrent") {
		t.Fatal("cutover pack lock must include non-deleted retired routes so their Event-owned backlog is terminalized before receipt")
	}
	if !strings.Contains(connectorIngressPackCutoverLockSQL, "FROM _System_ConnectorEvent e") {
		t.Fatal("cutover pack lock must include retained Event namespaces even when a route was deleted")
	}
	if !strings.Contains(connectorIngressPackCutoverLockSQL, "FROM _System_ConnectorWorkItem w") || !strings.Contains(connectorIngressPackCutoverLockSQL, "w.InstalledPackId = p.PackInstallId") || !strings.Contains(connectorIngressPackCutoverLockSQL, "w.Attempt > 0") {
		t.Fatal("cutover pack lock must include route-less positive-attempt legacy WorkItem namespaces")
	}
}

func TestConnectorIngressPackCutoverLockIncludesEveryNamespaceIncarnation(t *testing.T) {
	if strings.Contains(connectorIngressPackCutoverLockSQL, "r.InstalledPackId = p.PackInstallId") {
		t.Fatal("cutover pack lock must not exclude stale nonblank route incarnations from Event retirement")
	}
	if !strings.Contains(connectorIngressPackCutoverLockSQL, "LOWER(TRIM(r.Namespace)) = LOWER(TRIM(p.Name))") {
		t.Fatal("cutover pack lock must select all non-deleted routes in the pack namespace")
	}
	for _, query := range []string{connectorLegacyEventCutoverLockSQL, connectorLegacyEventCutoverRetireSQL} {
		if strings.Contains(query, "InstalledPackId = ?") {
			t.Fatal("cutover Event query must retire stale nonblank as well as blank install identities")
		}
	}
}

func TestConnectorIngressPackCutoverLockIncludesNonActiveLifecycleRows(t *testing.T) {
	if strings.Contains(connectorIngressPackCutoverLockSQL, "p.Status = 'active'") {
		t.Fatal("cutover pack lock must drain installing/removing/failed namespaces before the global receipt")
	}
	if !strings.Contains(connectorIngressPackCutoverLockSQL, "SELECT p.Name, p.PackInstallId, p.Status") {
		t.Fatal("cutover pack lock must retain lifecycle status so only active packs require ingress fencing")
	}
}

func TestConnectorIngressControlCutoverLockCoalescesNullableInstallID(t *testing.T) {
	if !strings.Contains(connectorIngressControlCutoverLockSQL, "SELECT COALESCE(PackInstallId, '')") {
		t.Fatal("cutover fence scan must not decode a nullable PackInstallId into a string")
	}
	if !strings.Contains(connectorIngressControlCutoverLockSQL, "LOWER(TRIM(Namespace)) = LOWER(TRIM(?))") {
		t.Fatal("cutover fence must find mixed-case pack namespaces after normalized pack selection")
	}
}

func TestConnectorLegacyWorkItemCutoverLockFencesPositiveAttemptExecutors(t *testing.T) {
	if strings.Contains(connectorLegacyWorkItemCutoverLockSQL, "w.Attempt > 0") {
		t.Fatal("cutover must not terminalize a positive-attempt WorkItem when its Event is terminal or otherwise unusual")
	}
	if !strings.Contains(connectorLegacyWorkItemCutoverLockSQL, "COALESCE(w.Attempt, 0) = 0") || !strings.Contains(connectorLegacyWorkItemCutoverRetireSQL, "COALESCE(w.Attempt, 0) = 0") {
		t.Fatal("cutover may retire only attempt-zero Event compatibility mirrors; a positive attempt is native WorkItem authority")
	}
	if !strings.Contains(connectorLegacyWorkItemCutoverLockSQL, "w.WorkType IN ('connector_event', 'workflow_action_apply')") {
		t.Fatal("cutover must not terminalize non-connector WorkItem types")
	}
	if !strings.Contains(connectorLegacyWorkItemCutoverLockSQL, "e.ExecutionOwner = 'event'") {
		t.Fatal("cutover must preserve already WorkItem-owned execution")
	}
	if strings.Contains(connectorLegacyWorkItemCutoverLockSQL, "AND e.InstalledPackId = w.InstalledPackId") {
		t.Fatal("cutover must join mirrors by Event ID so stale or blank WorkItem identities are retired before their Event is terminalized")
	}
	if !strings.Contains(connectorLegacyWorkItemCutoverLockSQL, "COALESCE(w.State, '') NOT IN ('processed', 'skipped', 'dead')") {
		t.Fatal("cutover must match every nonterminal WorkItem state, not a partial state list")
	}
	if !strings.Contains(connectorLegacyWorkItemCutoverLockSQL, "LOWER(TRIM(e.Namespace)) = LOWER(TRIM(?))") || !strings.Contains(connectorLegacyWorkItemCutoverLockSQL, "e.Status IN ('pending', 'failed')") {
		t.Fatal("cutover must drain attempt-zero mirrors before changing their Event owner")
	}
	if !strings.Contains(connectorLegacyWorkItemCutoverRetireSQL, "w.State = 'skipped'") {
		t.Fatal("cutover must terminalize drained legacy WorkItems before Event retirement")
	}
}

func TestConnectorCutoverEventAuthorityProofIsExhaustive(t *testing.T) {
	for _, fragment := range []string{
		connectorNonterminalEventPredicate("e"),
		"COALESCE(e.ExecutionOwner, '') <> 'work_item'",
		"LEFT JOIN _System_InstalledPack p",
		"p.PackInstallId = e.InstalledPackId",
		"COALESCE(p.IsDeleted, 0) = 0",
		"NULLIF(TRIM(COALESCE(e.InstalledPackId, '')), '') IS NULL",
		"p.PackInstallId IS NULL",
		"w.EventId = e.Id",
		"w.InstalledPackId = e.InstalledPackId",
		"w.WorkType IN ('connector_event', 'workflow_action_apply')",
		connectorNonterminalWorkItemPredicate("w"),
	} {
		if !strings.Contains(connectorEventAuthorityResidualCutoverLockSQL, fragment) {
			t.Fatalf("global Event authority proof is incomplete; missing %q", fragment)
		}
	}
	if strings.Contains(connectorEventAuthorityResidualCutoverLockSQL, "WHERE e.ExecutionOwner =") {
		t.Fatal("global Event authority proof must not omit NULL or unknown owners with an exact-owner WHERE clause")
	}
}

func TestConnectorCutoverWorkItemAuthorityProofMatchesReapplyAuthority(t *testing.T) {
	authority := connectorWorkItemExecutionAuthorityPredicate("w", "e")
	for _, fragment := range []string{
		authority,
		connectorNonterminalWorkItemPredicate("w"),
		"e.Id = w.EventId",
		"e.InstalledPackId = w.InstalledPackId",
		"p.PackInstallId = w.InstalledPackId",
		"NULLIF(TRIM(COALESCE(w.InstalledPackId, '')), '') IS NULL",
		"p.PackInstallId IS NULL",
		"e.Id IS NULL",
		"COALESCE(e.ExecutionOwner, '') <> 'work_item'",
	} {
		if !strings.Contains(connectorWorkItemAuthorityResidualCutoverLockSQL, fragment) {
			t.Fatalf("global WorkItem authority proof is incomplete; missing %q", fragment)
		}
	}
	reapply := connectorInboundReapplyBlockerPredicate(connectorPackInboundReapplyScope, true)
	if !strings.Contains(reapply, connectorWorkItemExecutionAuthorityPredicate("w", "queued_event")) {
		t.Fatal("cutover and reapply must share the same WorkItem authority definition")
	}
}

func TestConnectorCutoverAuthorityProofQueriesParseWithTiDBParser(t *testing.T) {
	p := parser.New()
	for name, query := range map[string]string{
		"event authority":    connectorEventAuthorityResidualCutoverLockSQL,
		"workitem authority": connectorWorkItemAuthorityResidualCutoverLockSQL,
	} {
		if _, _, err := p.Parse(query, "", ""); err != nil {
			t.Fatalf("%s proof does not parse: %v\nSQL:\n%s", name, err, query)
		}
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_StampsAtomicIngressFencedHandoff(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverInitialReceiptMissing(mock)
	expectWorkItemCutoverPessimisticBegin(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressPackCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"Name", "PackInstallId", "Status"}).AddRow("hubspot-crm-connector", "install-1", "active"))
	expectWorkItemCutoverLockedReceiptMissing(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressControlCutoverLockSQL)).
		WithArgs("hubspot-crm-connector", "install-1").
		WillReturnRows(sqlmock.NewRows([]string{"PackInstallId", "Channel", "Mode"}).AddRow("", "*", "quiescing"))
	mock.ExpectQuery(regexp.QuoteMeta(connectorLegacyEventCutoverLockSQL)).
		WithArgs("hubspot-crm-connector").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "Status"}).
			AddRow("event-pending", "pending").
			AddRow("event-failed", "failed"))
	expectNoLegacyWorkItemExecutors(mock, "install-1", "hubspot-crm-connector")
	mock.ExpectExec(regexp.QuoteMeta(connectorLegacyEventCutoverRetireSQL)).
		WithArgs("hubspot-crm-connector").
		WillReturnResult(sqlmock.NewResult(0, 3))
	expectNoResidualConnectorEventAuthority(mock)
	expectNoResidualConnectorWorkItemAuthority(mock)
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_Migration")).
		WillReturnResult(sqlmock.NewResult(1, 1))
	expectWorkItemCutoverReceiptAttested(mock)
	expectWorkItemCutoverCommit(mock)

	if err := VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db); err != nil {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover = error %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_RejectsMissingWorkItemSchemaBeforeEventRetirement(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverReceiptMissing(mock)
	mock.ExpectQuery(`SELECT COUNT\(DISTINCT COLUMN_NAME\)[\s\S]*FROM INFORMATION_SCHEMA\.COLUMNS`).
		WillReturnRows(sqlmock.NewRows([]string{"COUNT(DISTINCT COLUMN_NAME)"}).AddRow(len(connectorWorkItemCutoverRequiredColumns) - 1))

	err = VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db)
	if err == nil || !strings.Contains(err.Error(), "execution schema is not ready") {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover error = %v, want schema-readiness failure", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_RetiresFailedPackBacklogBeforeReceipt(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverInitialReceiptMissing(mock)
	expectWorkItemCutoverPessimisticBegin(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressPackCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"Name", "PackInstallId", "Status"}).AddRow("hubspot-crm-connector", "install-failed", installedPackStatusFailed))
	expectWorkItemCutoverLockedReceiptMissing(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorLegacyEventCutoverLockSQL)).
		WithArgs("hubspot-crm-connector").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "Status"}).AddRow("failed-pack-event", "pending"))
	expectNoLegacyWorkItemExecutors(mock, "install-failed", "hubspot-crm-connector")
	mock.ExpectExec(regexp.QuoteMeta(connectorLegacyEventCutoverRetireSQL)).
		WithArgs("hubspot-crm-connector").
		WillReturnResult(sqlmock.NewResult(0, 1))
	expectNoResidualConnectorEventAuthority(mock)
	expectNoResidualConnectorWorkItemAuthority(mock)
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_Migration")).
		WillReturnResult(sqlmock.NewResult(1, 1))
	expectWorkItemCutoverReceiptAttested(mock)
	expectWorkItemCutoverCommit(mock)

	if err := VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db); err != nil {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_RejectsInFlightLegacyWorkItemWithoutStamp(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverInitialReceiptMissing(mock)
	expectWorkItemCutoverPessimisticBegin(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressPackCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"Name", "PackInstallId", "Status"}).AddRow("hubspot-crm-connector", "install-1", installedPackStatusActive))
	expectWorkItemCutoverLockedReceiptMissing(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressControlCutoverLockSQL)).
		WithArgs("hubspot-crm-connector", "install-1").
		WillReturnRows(sqlmock.NewRows([]string{"PackInstallId", "Channel", "Mode"}).AddRow("", "*", "quiescing"))
	mock.ExpectQuery(regexp.QuoteMeta(connectorLegacyEventCutoverLockSQL)).
		WithArgs("hubspot-crm-connector").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "Status"}).AddRow("event-pending", "pending"))
	mock.ExpectQuery(regexp.QuoteMeta(connectorLegacyWorkItemCutoverLockSQL)).
		WithArgs("install-1", "hubspot-crm-connector").
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "State"}).AddRow("work-processing", "processing"))
	expectWorkItemCutoverRollback(mock)

	err = VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db)
	if err == nil || !strings.Contains(err.Error(), "WorkItem executor(s) are still processing") {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover error = %v, want in-flight WorkItem rejection", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_RejectsResidualOrphanEventsBeforeReceipt(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverInitialReceiptMissing(mock)
	expectWorkItemCutoverPessimisticBegin(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressPackCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"Name", "PackInstallId", "Status"}))
	expectWorkItemCutoverLockedReceiptMissing(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorEventAuthorityResidualCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "Namespace"}).AddRow("orphan-event", "orphan-connector"))
	expectWorkItemCutoverRollback(mock)

	err = VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db)
	if err == nil || !strings.Contains(err.Error(), "lack exact WorkItem-only authority") {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover error = %v, want residual orphan rejection", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_RejectsResidualOrphanWorkItemsBeforeReceipt(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverInitialReceiptMissing(mock)
	expectWorkItemCutoverPessimisticBegin(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressPackCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"Name", "PackInstallId", "Status"}))
	expectWorkItemCutoverLockedReceiptMissing(mock)
	expectNoResidualConnectorEventAuthority(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorWorkItemAuthorityResidualCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "Namespace"}).AddRow("orphan-work", "orphan-connector"))
	expectWorkItemCutoverRollback(mock)

	err = VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db)
	if err == nil || !strings.Contains(err.Error(), "lack an exact Event or installed-pack owner") {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover error = %v, want residual orphan WorkItem rejection", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_RejectsInFlightLegacyEventWithoutStamp(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverInitialReceiptMissing(mock)
	expectWorkItemCutoverPessimisticBegin(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressPackCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"Name", "PackInstallId", "Status"}).AddRow("hubspot-crm-connector", "install-1", "active"))
	expectWorkItemCutoverLockedReceiptMissing(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressControlCutoverLockSQL)).
		WithArgs("hubspot-crm-connector", "install-1").
		WillReturnRows(sqlmock.NewRows([]string{"PackInstallId", "Channel", "Mode"}).AddRow("", "*", "quiescing"))
	mock.ExpectQuery(regexp.QuoteMeta(connectorLegacyEventCutoverLockSQL)).
		WithArgs("hubspot-crm-connector").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "Status"}).
			AddRow("event-processing", "processing"))
	expectWorkItemCutoverRollback(mock)

	err = VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db)
	if err == nil || !strings.Contains(err.Error(), "still processing") {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover error = %v, want in-flight legacy Event rejection", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_RejectsMissingIngressFenceWithoutStamp(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverInitialReceiptMissing(mock)
	expectWorkItemCutoverPessimisticBegin(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressPackCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"Name", "PackInstallId", "Status"}).AddRow("hubspot-crm-connector", "install-1", "active"))
	expectWorkItemCutoverLockedReceiptMissing(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressControlCutoverLockSQL)).
		WithArgs("hubspot-crm-connector", "install-1").
		WillReturnRows(sqlmock.NewRows([]string{"PackInstallId", "Channel", "Mode"}))
	expectWorkItemCutoverRollback(mock)

	err = VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db)
	if err == nil || !strings.Contains(err.Error(), "inbound ingress is not quiesced") {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover error = %v, want ingress-fence failure", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_RejectsActiveIngressOverrideWithoutStamp(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverInitialReceiptMissing(mock)
	expectWorkItemCutoverPessimisticBegin(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressPackCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"Name", "PackInstallId", "Status"}).AddRow("hubspot-crm-connector", "install-1", "active"))
	expectWorkItemCutoverLockedReceiptMissing(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressControlCutoverLockSQL)).
		WithArgs("hubspot-crm-connector", "install-1").
		WillReturnRows(sqlmock.NewRows([]string{"PackInstallId", "Channel", "Mode"}).
			AddRow("", "*", "quiescing").
			AddRow("install-1", "webhook", "active"))
	expectWorkItemCutoverRollback(mock)

	err = VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db)
	if err == nil || !strings.Contains(err.Error(), "active admission override") {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover error = %v, want active override failure", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorWorkItemOnlyExecutionCutover_PackLockErrorPropagates(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectWorkItemCutoverInitialReceiptMissing(mock)
	expectWorkItemCutoverPessimisticBegin(mock)
	mock.ExpectQuery(regexp.QuoteMeta(connectorIngressPackCutoverLockSQL)).
		WillReturnError(sql.ErrConnDone)
	expectWorkItemCutoverRollback(mock)

	err = VerifyConnectorWorkItemOnlyExecutionCutover(context.Background(), db)
	if !errors.Is(err, sql.ErrConnDone) {
		t.Fatalf("VerifyConnectorWorkItemOnlyExecutionCutover error = %v, want %v", err, sql.ErrConnDone)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func expectWorkItemCutoverInitialReceiptMissing(mock sqlmock.Sqlmock) {
	expectWorkItemCutoverReceiptMissing(mock)
	mock.ExpectQuery(`SELECT COUNT\(DISTINCT COLUMN_NAME\)[\s\S]*FROM INFORMATION_SCHEMA\.COLUMNS`).
		WillReturnRows(sqlmock.NewRows([]string{"COUNT(DISTINCT COLUMN_NAME)"}).AddRow(len(connectorWorkItemCutoverRequiredColumns)))
}

func expectWorkItemCutoverLockedReceiptMissing(mock sqlmock.Sqlmock) {
	expectWorkItemCutoverReceiptMissing(mock)
}

func expectWorkItemCutoverReceiptMissing(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(`(?s)SELECT Version, Description\s+FROM _System_Migration\s+WHERE Id = \?\s+LIMIT 1`).
		WithArgs(migrationIDVerifyConnectorWorkItemOnlyExecutionCutover).
		WillReturnRows(sqlmock.NewRows([]string{"Version", "Description"}))
}

func expectWorkItemCutoverReceiptAttested(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(`(?s)SELECT Version, Description\s+FROM _System_Migration\s+WHERE Id = \?\s+LIMIT 1`).
		WithArgs(migrationIDVerifyConnectorWorkItemOnlyExecutionCutover).
		WillReturnRows(sqlmock.NewRows([]string{"Version", "Description"}).AddRow(
			migrationVersionVerifyConnectorWorkItemOnlyExecutionCutover,
			migrationDescriptionVerifyConnectorWorkItemOnlyExecutionCutover,
		))
}

func expectNoResidualConnectorEventAuthority(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(regexp.QuoteMeta(connectorEventAuthorityResidualCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "Namespace"}))
}

func expectNoResidualConnectorWorkItemAuthority(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(regexp.QuoteMeta(connectorWorkItemAuthorityResidualCutoverLockSQL)).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "Namespace"}))
}

func expectNoLegacyWorkItemExecutors(mock sqlmock.Sqlmock, packInstallID, namespace string) {
	mock.ExpectQuery(regexp.QuoteMeta(connectorLegacyWorkItemCutoverLockSQL)).
		WithArgs(packInstallID, namespace).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "State"}))
	mock.ExpectExec(regexp.QuoteMeta(connectorLegacyWorkItemCutoverRetireSQL)).
		WithArgs(packInstallID, namespace).
		WillReturnResult(sqlmock.NewResult(0, 0))
}

func expectWorkItemCutoverPessimisticBegin(mock sqlmock.Sqlmock) {
	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
}

func expectWorkItemCutoverCommit(mock sqlmock.Sqlmock) {
	mock.ExpectCommit()
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))
}

func expectWorkItemCutoverRollback(mock sqlmock.Sqlmock) {
	mock.ExpectRollback()
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))
}