package connector_workqueue

import (
	"context"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

func identityConvergenceDependencyForTest(t *testing.T, objectAPIName, recordID, remoteID string) contractmodels.FlowExecutionDependency {
	t.Helper()
	request, err := contractmodels.NewFlowIdentityConvergenceDependencyRequest(
		objectAPIName, recordID, "pingcap-hubspot-companion", "hubspot-contact",
	)
	if err != nil {
		t.Fatalf("NewFlowIdentityConvergenceDependencyRequest: %v", err)
	}
	dependency, err := contractmodels.BindFlowIdentityConvergenceDependency(request, "hubspot", "contacts", remoteID)
	if err != nil {
		t.Fatalf("BindFlowIdentityConvergenceDependency: %v", err)
	}
	return dependency
}

func identityConvergenceWaitColumnsForTest() []string {
	return []string{
		"WorkId", "InstalledPackId", "DependencyBlockedAt", "Namespace", "SourceObjectKind", "SourceObjectID",
		"DependencyKind", "DependencyObjectAPIName", "DependencyRecordId", "DependencyKeyHash", "DependencyConnectorOwnerKey",
		"DependencyConnectorNamespace", "DependencyConnectorSourceObject", "DependencyDestinationSlotId", "DependencyConnectorRemoteId",
		"DependencyProbeAfter",
	}
}

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

	blockedAt := time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC)
	valid := identityConvergenceDependencyForTest(t, "Lead", "00Q000000000001AAA", "contact-1")
	tampered := valid
	tampered.KeyHash = strings.Repeat("f", 64)
	mock.ExpectQuery(`(?s)JOIN _System_ConnectorEvent e.*ORDER BY w.DependencyBlockedAt ASC, w.WorkId ASC.*LIMIT \?`).
		WithArgs(WorkStateDependencyBlocked, 10).
		WillReturnRows(sqlmock.NewRows(identityConvergenceWaitColumnsForTest()).
			AddRow(
				"work-tampered", "pack-1", blockedAt, "hubspot", "contacts", "contact-1",
				tampered.Kind, tampered.ObjectAPIName, tampered.RecordID, tampered.KeyHash, tampered.ConnectorOwnerKey,
				tampered.ConnectorNamespace, tampered.ConnectorSourceObject, tampered.DestinationSlotID, tampered.ConnectorRemoteID,
				blockedAt,
			).
			AddRow(
				"work-valid", "pack-1", blockedAt.Add(2*time.Second), "hubspot", "contacts", "contact-1",
				valid.Kind, valid.ObjectAPIName, valid.RecordID, valid.KeyHash, valid.ConnectorOwnerKey,
				valid.ConnectorNamespace, valid.ConnectorSourceObject, valid.DestinationSlotID, valid.ConnectorRemoteID,
				blockedAt,
			))

	waits, err := NewRepository(db, nil, nil).ListIdentityConvergenceWaits(context.Background(), nil, 10)
	if err != nil {
		t.Fatalf("ListIdentityConvergenceWaits: %v", err)
	}
	if len(waits) != 2 || waits[0].WorkID != "work-tampered" || !IsIdentityConvergenceWaitInvalid(waits[0]) || waits[1].WorkID != "work-valid" {
		t.Fatalf("reconcile waits = %#v, want isolated tampered row plus work-valid", waits)
	}
	mock.ExpectQuery(`(?s)LEFT JOIN _System_ConnectorEvent e.*e.Id IS NULL.*ORDER BY w.WorkId ASC.*LIMIT \?`).
		WithArgs(WorkStateDependencyBlocked, 10).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).
			AddRow("work-orphan"))
	invalid, err := NewRepository(db, nil, nil).ListInvalidIdentityConvergenceWaitIDs(context.Background(), 10)
	if err != nil {
		t.Fatalf("ListInvalidIdentityConvergenceWaitIDs: %v", err)
	}
	if len(invalid) != 1 || invalid[0] != "work-orphan" {
		t.Fatalf("invalid waits = %#v, want isolated work-orphan", invalid)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 22, 8, 0, 0, 0, time.UTC)
	dependency := identityConvergenceDependencyForTest(t, "Lead", "00Q000000000001AAA", "contact-1")
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = \?.*DependencyKind = \?.*DependencyObjectAPIName = \?.*DependencyRecordId = \?.*DependencyKeyHash = \?.*DependencyConnectorOwnerKey = \?.*DependencyConnectorNamespace = \?.*DependencyConnectorSourceObject = \?.*DependencyDestinationSlotId = \?.*DependencyConnectorRemoteId = \?.*DependencyBlockedAt = \?.*DependencyProbeAfter = \?.*LeaseOwner = NULL.*WHERE WorkId = \?.*EventId = \?.*State = \?.*LeaseToken = \?`).
		WithArgs(
			WorkStateDependencyBlocked,
			dependency.Kind,
			dependency.ObjectAPIName,
			dependency.RecordID,
			dependency.KeyHash,
			dependency.ConnectorOwnerKey,
			dependency.ConnectorNamespace,
			dependency.ConnectorSourceObject,
			dependency.DestinationSlotID,
			dependency.ConnectorRemoteID,
			now,
			now,
			now,
			"work-1",
			"event-1",
			WorkStateProcessing,
			"lease-1",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	blocked, err := NewRepository(db, nil, nil).BlockWorkItem(context.Background(), BlockWorkItemInput{
		WorkID:     "work-1",
		EventID:    "event-1",
		LeaseToken: "lease-1",
		Dependency: dependency,
		Now:        now,
	})
	if err != nil {
		t.Fatalf("BlockWorkItem: %v", err)
	}
	if !blocked {
		t.Fatal("BlockWorkItem blocked=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	blockedAt := time.Date(2026, 7, 22, 8, 0, 0, 0, time.UTC)
	now := blockedAt.Add(30 * time.Minute)
	dependency := identityConvergenceDependencyForTest(t, "Lead", "00Q000000000001AAA", "contact-1")
	wait := IdentityConvergenceWait{
		WorkID:              "work-1",
		DependencyBlockedAt: blockedAt,
		Dependency:          dependency,
	}
	repo := NewRepository(db, func() string { return "expiry-token" }, nil)

	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*LeaseOwner = \?.*DependencyKind = NULL.*DependencyProbeAfter = NULL.*WHERE WorkId = \?.*DependencyKind = \?.*DependencyKeyHash = \?.*DependencyBlockedAt = \?.*DependencyProbeAfter <=> \?.*State = \?.*LeaseOwner IS NULL`).
		WithArgs(
			"identity-convergence-expirer",
			"identity-convergence-expiry:expiry-token",
			now.Add(time.Minute),
			now,
			"work-1",
			contractmodels.FlowDependencyKindIdentityConvergence,
			dependency.KeyHash,
			blockedAt,
			nil,
			WorkStateProcessing,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`(?s)SELECT EventId.*FROM _System_ConnectorWorkItem.*WHERE WorkId = \?.*State = \?.*LeaseToken = \?.*LIMIT 1`).
		WithArgs("work-1", WorkStateProcessing, "identity-convergence-expiry:expiry-token").
		WillReturnRows(sqlmock.NewRows([]string{"EventId"}).AddRow("event-1"))
	mock.ExpectQuery(`(?s)SELECT w.EventId, w.FailureCount.*FROM _System_ConnectorWorkItem w.*WHERE w.WorkId = \?.*w.EventId = \?.*w.State = \?.*w.LeaseToken = \?.*FOR UPDATE`).
		WithArgs("work-1", "event-1", WorkStateProcessing, "identity-convergence-expiry:expiry-token").
		WillReturnRows(sqlmock.NewRows([]string{"EventId", "FailureCount"}).AddRow("event-1", 0))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*FailureCount = \?.*FailureClass = \?.*TerminalAt = \?.*LeaseToken = NULL.*WHERE WorkId = \?.*EventId = \?.*State = \?.*LeaseToken = \?.*FailureCount = \?`).
		WithArgs(
			WorkStateDead,
			now,
			1,
			FailureClassIdentityConvergenceExpired,
			FailureSummaryIdentityConvergenceExpired,
			now,
			now,
			now,
			"work-1",
			"event-1",
			WorkStateProcessing,
			"identity-convergence-expiry:expiry-token",
			0,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-1").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(1), "", ""))
	mock.ExpectCommit()
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))

	expired, err := repo.ExpireIdentityConvergenceWait(context.Background(), wait, now)
	if err != nil {
		t.Fatalf("ExpireIdentityConvergenceWait: %v", err)
	}
	if !expired {
		t.Fatal("ExpireIdentityConvergenceWait expired=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 29, 12, 30, 0, 0, time.UTC)
	observed := now.Add(-time.Minute)
	dependency := identityConvergenceDependencyForTest(t, "Lead", "00Q000000000001AAA", "contact-1")
	repo := NewRepository(db, func() string { return "invalid-token" }, nil)

	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
	mock.ExpectQuery(`(?s)SELECT w.InstalledPackId,.*w.LastModifiedDate.*LEFT JOIN _System_ConnectorEvent e.*WHERE w.WorkId = \?.*w.DependencyKind = \?.*w.State = \?.*FOR UPDATE`).
		WithArgs("work-invalid", contractmodels.FlowDependencyKindIdentityConvergence, WorkStateProcessing).
		WillReturnRows(sqlmock.NewRows([]string{
			"InstalledPackId", "DependencyBlockedAt", "Namespace", "SourceObjectKind", "SourceObjectID",
			"DependencyKind", "DependencyObjectAPIName", "DependencyRecordId", "DependencyKeyHash", "DependencyConnectorOwnerKey",
			"DependencyConnectorNamespace", "DependencyConnectorSourceObject", "DependencyDestinationSlotId", "DependencyConnectorRemoteId",
			"LastModifiedDate",
		}).AddRow(
			"pack-1", now.Add(-2*time.Minute), nil, nil, nil,
			dependency.Kind, dependency.ObjectAPIName, dependency.RecordID, dependency.KeyHash, dependency.ConnectorOwnerKey,
			dependency.ConnectorNamespace, dependency.ConnectorSourceObject, dependency.DestinationSlotID, dependency.ConnectorRemoteID,
			observed,
		))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*LeaseOwner = \?.*DependencyKind = NULL.*WHERE WorkId = \?.*LastModifiedDate = \?.*DependencyKind = \?.*State = \?.*LeaseOwner IS NULL`).
		WithArgs(
			"identity-convergence-invalid-terminalizer",
			"identity-convergence-invalid:invalid-token",
			now.Add(time.Minute),
			now,
			"work-invalid",
			observed,
			contractmodels.FlowDependencyKindIdentityConvergence,
			WorkStateProcessing,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`(?s)SELECT EventId.*FROM _System_ConnectorWorkItem.*WHERE WorkId = \?.*State = \?.*LeaseToken = \?.*LIMIT 1`).
		WithArgs("work-invalid", WorkStateProcessing, "identity-convergence-invalid:invalid-token").
		WillReturnRows(sqlmock.NewRows([]string{"EventId"}).AddRow("event-missing"))
	mock.ExpectQuery(`(?s)SELECT w.EventId, w.FailureCount.*FROM _System_ConnectorWorkItem w.*WHERE w.WorkId = \?.*w.EventId = \?.*w.State = \?.*w.LeaseToken = \?.*FOR UPDATE`).
		WithArgs("work-invalid", "event-missing", WorkStateProcessing, "identity-convergence-invalid:invalid-token").
		WillReturnRows(sqlmock.NewRows([]string{"EventId", "FailureCount"}).AddRow("event-missing", 0))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*FailureClass = \?.*TerminalAt = \?.*WHERE WorkId = \?.*EventId = \?.*State = \?.*LeaseToken = \?`).
		WithArgs(
			WorkStateDead,
			now,
			1,
			FailureClassIdentityConvergenceInvalid,
			FailureSummaryIdentityConvergenceInvalid,
			now,
			now,
			now,
			"work-invalid",
			"event-missing",
			WorkStateProcessing,
			"identity-convergence-invalid:invalid-token",
			0,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-invalid").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(1), "", ""))
	mock.ExpectCommit()
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))

	terminalized, err := repo.TerminalizeInvalidIdentityConvergenceWait(context.Background(), "work-invalid", now)
	if err != nil {
		t.Fatalf("TerminalizeInvalidIdentityConvergenceWait: %v", err)
	}
	if !terminalized {
		t.Fatal("TerminalizeInvalidIdentityConvergenceWait terminalized=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 29, 12, 45, 0, 0, time.UTC)
	observed := now.Add(-time.Minute)
	tampered := identityConvergenceDependencyForTest(t, "Lead", "00Q000000000001AAA", "contact-1")
	tampered.KeyHash = strings.Repeat("f", 64)
	repo := NewRepository(db, func() string { return "invalid-token" }, nil)

	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
	mock.ExpectQuery(`(?s)SELECT w.InstalledPackId,.*w.LastModifiedDate.*LEFT JOIN _System_ConnectorEvent e.*WHERE w.WorkId = \?.*w.DependencyKind = \?.*w.State = \?.*FOR UPDATE`).
		WithArgs("work-tampered", contractmodels.FlowDependencyKindIdentityConvergence, WorkStateProcessing).
		WillReturnRows(sqlmock.NewRows([]string{
			"InstalledPackId", "DependencyBlockedAt", "Namespace", "SourceObjectKind", "SourceObjectID",
			"DependencyKind", "DependencyObjectAPIName", "DependencyRecordId", "DependencyKeyHash", "DependencyConnectorOwnerKey",
			"DependencyConnectorNamespace", "DependencyConnectorSourceObject", "DependencyDestinationSlotId", "DependencyConnectorRemoteId",
			"LastModifiedDate",
		}).AddRow(
			"pack-1", now.Add(-2*time.Minute), "hubspot", "contacts", "contact-1",
			tampered.Kind, tampered.ObjectAPIName, tampered.RecordID, tampered.KeyHash, tampered.ConnectorOwnerKey,
			tampered.ConnectorNamespace, tampered.ConnectorSourceObject, tampered.DestinationSlotID, tampered.ConnectorRemoteID,
			observed,
		))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*LeaseOwner = \?.*DependencyKind = NULL.*WHERE WorkId = \?.*LastModifiedDate = \?.*DependencyKind = \?.*State = \?.*LeaseOwner IS NULL`).
		WithArgs(
			"identity-convergence-invalid-terminalizer",
			"identity-convergence-invalid:invalid-token",
			now.Add(time.Minute),
			now,
			"work-tampered",
			observed,
			contractmodels.FlowDependencyKindIdentityConvergence,
			WorkStateProcessing,
		).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectRollback()
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))

	terminalized, err := repo.TerminalizeInvalidIdentityConvergenceWait(context.Background(), "work-tampered", now)
	if err != nil {
		t.Fatalf("TerminalizeInvalidIdentityConvergenceWait: %v", err)
	}
	if terminalized {
		t.Fatal("TerminalizeInvalidIdentityConvergenceWait terminalized=true after lost CAS, want false")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	blockedAt := time.Date(2026, 7, 22, 8, 0, 0, 0, time.UTC)
	observedProbeAfter := blockedAt.Add(29 * time.Minute)
	now := blockedAt.Add(30 * time.Minute)
	dependency := identityConvergenceDependencyForTest(t, "Lead", "00Q000000000001AAA", "contact-1")
	wait := IdentityConvergenceWait{
		WorkID:               "work-1",
		DependencyBlockedAt:  blockedAt,
		DependencyProbeAfter: &observedProbeAfter,
		Dependency:           dependency,
	}
	repo := NewRepository(db, func() string { return "expiry-token" }, nil)

	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*DependencyKeyHash = \?.*DependencyBlockedAt = \?.*DependencyProbeAfter <=> \?.*State = \?`).
		WithArgs(
			"identity-convergence-expirer",
			"identity-convergence-expiry:expiry-token",
			now.Add(time.Minute),
			now,
			"work-1",
			contractmodels.FlowDependencyKindIdentityConvergence,
			dependency.KeyHash,
			blockedAt,
			observedProbeAfter,
			WorkStateProcessing,
		).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectRollback()
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))

	expired, err := repo.ExpireIdentityConvergenceWait(context.Background(), wait, now)
	if err != nil {
		t.Fatalf("ExpireIdentityConvergenceWait: %v", err)
	}
	if expired {
		t.Fatal("ExpireIdentityConvergenceWait expired=true after concurrent probe reschedule, want false")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	observedProbeAfter := time.Date(2026, 7, 23, 12, 0, 0, 0, time.UTC)
	now := observedProbeAfter.Add(time.Minute)
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET DependencyProbeAfter = \?.*WHERE WorkId = \?.*DependencyKeyHash = \?.*DependencyProbeAfter = \?.*State = \?`).
		WithArgs(now.Add(identityConvergenceProbeRetryDelay), now, "work-1", "dependency-hash", observedProbeAfter, WorkStateDependencyBlocked).
		WillReturnResult(sqlmock.NewResult(0, 0))

	repo := NewRepository(db, nil, nil)
	if err := repo.RescheduleIdentityConvergenceProbeRetry(context.Background(), "work-1", "dependency-hash", observedProbeAfter, now); err != nil {
		t.Fatalf("RescheduleIdentityConvergenceProbeRetry: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	observedProbeAfter := time.Date(2026, 7, 23, 12, 0, 0, 0, time.UTC)
	now := observedProbeAfter.Add(time.Minute)
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*DependencyProbeAfter = NULL.*WHERE WorkId = \?.*DependencyKeyHash = \?.*DependencyProbeAfter = \?.*State = \?`).
		WithArgs(now, "work-1", "dependency-hash", observedProbeAfter, WorkStateDependencyBlocked).
		WillReturnResult(sqlmock.NewResult(0, 1))

	repo := NewRepository(db, nil, nil)
	if err := repo.ClearIdentityConvergenceProbeRetry(context.Background(), "work-1", "dependency-hash", observedProbeAfter, now); err != nil {
		t.Fatalf("ClearIdentityConvergenceProbeRetry: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 22, 8, 1, 0, 0, time.UTC)
	dependency := identityConvergenceDependencyForTest(t, "Lead", "00Q000000000001AAA", "contact-1")
	pattern := `(?s)UPDATE _System_ConnectorWorkItem.*State = \?.*RunAfter = \?.*DependencyKind = NULL.*DependencyRecordId = NULL.*WHERE DependencyKind = \?.*DependencyObjectAPIName = \?.*DependencyRecordId = \?.*State = \?`
	mock.ExpectExec(pattern).
		WithArgs(WorkStatePending, now, now, dependency.Kind, dependency.ObjectAPIName, dependency.RecordID, WorkStateDependencyBlocked).
		WillReturnResult(sqlmock.NewResult(0, 2))
	mock.ExpectExec(pattern).
		WithArgs(WorkStatePending, now, now, dependency.Kind, dependency.ObjectAPIName, dependency.RecordID, WorkStateDependencyBlocked).
		WillReturnResult(sqlmock.NewResult(0, 0))

	repo := NewRepository(db, nil, nil)
	for attempt, want := range []int64{2, 0} {
		got, err := repo.WakeRecordDependency(context.Background(), dependency.ObjectAPIName, dependency.RecordID, now)
		if err != nil {
			t.Fatalf("WakeRecordDependency attempt %d: %v", attempt+1, err)
		}
		if got != want {
			t.Fatalf("WakeRecordDependency attempt %d = %d, want %d", attempt+1, got, want)
		}
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	dependency := identityConvergenceDependencyForTest(t, "Lead", "00Q000000000001AAA", "contact-1")
	mock.ExpectQuery(`(?s)SELECT DependencyKind, DependencyObjectAPIName, DependencyRecordId, DependencyKeyHash.*WHERE State = \?.*ORDER BY DependencyBlockedAt ASC, WorkId ASC.*LIMIT \?`).
		WithArgs(WorkStateDependencyBlocked, 100).
		WillReturnRows(sqlmock.NewRows([]string{
			"DependencyKind", "DependencyObjectAPIName", "DependencyRecordId", "DependencyKeyHash",
			"DependencyConnectorOwnerKey", "DependencyConnectorNamespace", "DependencyConnectorSourceObject",
			"DependencyDestinationSlotId", "DependencyConnectorRemoteId",
		}).
			AddRow(dependency.Kind, dependency.ObjectAPIName, dependency.RecordID, dependency.KeyHash, dependency.ConnectorOwnerKey, dependency.ConnectorNamespace, dependency.ConnectorSourceObject, dependency.DestinationSlotID, dependency.ConnectorRemoteID).
			AddRow(dependency.Kind, dependency.ObjectAPIName, dependency.RecordID, dependency.KeyHash, dependency.ConnectorOwnerKey, dependency.ConnectorNamespace, dependency.ConnectorSourceObject, dependency.DestinationSlotID, dependency.ConnectorRemoteID))

	got, err := NewRepository(db, nil, nil).ListRecordDependencies(context.Background(), 100)
	if err != nil {
		t.Fatalf("ListRecordDependencies: %v", err)
	}
	if len(got) != 1 || got[0] != dependency {
		t.Fatalf("ListRecordDependencies = %#v, want one deduplicated dependency", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 22, 8, 2, 0, 0, time.UTC)
	dependency := identityConvergenceDependencyForTest(t, "Lead", "00Q000000000002AAA", "236750625181")
	// A finalized remote link is keyed without Event source-object provenance.
	// Returning two affected rows models two independently hydrated source-object
	// waiters for the same authority tuple; both must wake and revalidate.
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*DependencyKind = NULL.*WHERE DependencyKind = \?.*DependencyConnectorOwnerKey = \?.*DependencyConnectorNamespace = \?\s+AND DependencyDestinationSlotId = \?.*DependencyConnectorRemoteId = \?.*State = \?`).
		WithArgs(WorkStatePending, now, now, dependency.Kind, dependency.ConnectorOwnerKey, dependency.ConnectorNamespace, dependency.DestinationSlotID, dependency.ConnectorRemoteID, WorkStateDependencyBlocked).
		WillReturnResult(sqlmock.NewResult(0, 2))

	got, err := NewRepository(db, nil, nil).WakeRemoteIdentity(context.Background(), IdentityConvergenceRemoteSignal{
		ConnectorOwnerKey:  dependency.ConnectorOwnerKey,
		ConnectorNamespace: dependency.ConnectorNamespace,
		DestinationSlotID:  dependency.DestinationSlotID,
		ConnectorRemoteID:  dependency.ConnectorRemoteID,
	}, now)
	if err != nil {
		t.Fatalf("WakeRemoteIdentity: %v", err)
	}
	if got != 2 {
		t.Fatalf("WakeRemoteIdentity = %d, want 2", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	dependency := identityConvergenceDependencyForTest(t, "Lead", "00Q000000000002AAA", "236750625181")
	blockedAt := time.Date(2026, time.July, 22, 12, 0, 0, 0, time.UTC)
	mock.ExpectQuery(`(?s)SELECT w.WorkId, w.InstalledPackId, w.DependencyBlockedAt, e.Namespace, e.SourceObjectKind, e.SourceObjectID,.*w.DependencyKind.*FROM _System_ConnectorWorkItem w.*JOIN _System_ConnectorEvent e.*e.InstalledPackId = w.InstalledPackId.*WHERE w.State = \?.*ORDER BY w.DependencyBlockedAt ASC, w.WorkId ASC.*LIMIT \?`).
		WithArgs(WorkStateDependencyBlocked, 100).
		WillReturnRows(sqlmock.NewRows([]string{
			"WorkId", "InstalledPackId", "DependencyBlockedAt", "Namespace", "SourceObjectKind", "SourceObjectID",
			"DependencyKind", "DependencyObjectAPIName", "DependencyRecordId", "DependencyKeyHash", "DependencyConnectorOwnerKey",
			"DependencyConnectorNamespace", "DependencyConnectorSourceObject", "DependencyDestinationSlotId", "DependencyConnectorRemoteId", "DependencyProbeAfter",
		}).AddRow(
			"work-1", "pack-1", blockedAt, "hubspot", "contacts", "236750625181",
			dependency.Kind, dependency.ObjectAPIName, dependency.RecordID, dependency.KeyHash, dependency.ConnectorOwnerKey,
			dependency.ConnectorNamespace, dependency.ConnectorSourceObject, dependency.DestinationSlotID, dependency.ConnectorRemoteID, nil,
		))

	got, err := NewRepository(db, nil, nil).ListIdentityConvergenceWaits(context.Background(), nil, 100)
	if err != nil {
		t.Fatalf("ListIdentityConvergenceWaits: %v", err)
	}
	if len(got) != 1 || got[0].WorkID != "work-1" || got[0].InstalledPackID != "pack-1" || !got[0].DependencyBlockedAt.Equal(blockedAt) || got[0].Namespace != "hubspot" || got[0].SourceID != "236750625181" || got[0].Dependency != dependency {
		t.Fatalf("ListIdentityConvergenceWaits = %#v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	blockedAt := time.Date(2026, time.July, 22, 12, 0, 0, 0, time.UTC)
	cursor := &IdentityConvergenceCursor{DependencyBlockedAt: blockedAt, WorkID: "work-1000"}
	mock.ExpectQuery(`(?s)JOIN _System_ConnectorEvent e.*e.InstalledPackId = w.InstalledPackId.*WHERE w.State = \?.*w.DependencyBlockedAt > \? OR \(w.DependencyBlockedAt = \? AND w.WorkId > \?\).*ORDER BY w.DependencyBlockedAt ASC, w.WorkId ASC.*LIMIT \?`).
		WithArgs(WorkStateDependencyBlocked, blockedAt, blockedAt, "work-1000", 100).
		WillReturnRows(sqlmock.NewRows([]string{
			"WorkId", "InstalledPackId", "DependencyBlockedAt", "Namespace", "SourceObjectKind", "SourceObjectID",
			"DependencyKind", "DependencyObjectAPIName", "DependencyRecordId", "DependencyKeyHash", "DependencyConnectorOwnerKey",
			"DependencyConnectorNamespace", "DependencyConnectorSourceObject", "DependencyDestinationSlotId", "DependencyConnectorRemoteId", "DependencyProbeAfter",
		}))

	got, err := NewRepository(db, nil, nil).ListIdentityConvergenceWaits(context.Background(), cursor, 100)
	if err != nil {
		t.Fatalf("ListIdentityConvergenceWaits: %v", err)
	}
	if len(got) != 0 {
		t.Fatalf("ListIdentityConvergenceWaits = %#v, want empty page", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	mock.ExpectQuery(`(?s)SELECT LaneOrdinal, InputStartOrdinal, OwnershipEpoch, FailureCount.*ProjectionKey, ProjectionActiveKey.*DependencyKind, DependencyObjectAPIName, DependencyRecordId.*DependencyKeyHash, DependencyConnectorOwnerKey.*DependencyConnectorNamespace, DependencyConnectorSourceObject.*DependencyDestinationSlotId, DependencyConnectorRemoteId.*DependencyBlockedAt, DependencyProbeAfter.*FROM _System_ConnectorWorkItem.*WHERE 1 = 0`).
		WillReturnRows(sqlmock.NewRows([]string{
			"LaneOrdinal", "OwnershipEpoch", "FailureCount", "FailureClass", "FailureSummary",
			"LastFailureAt", "TerminalAt", "ReplayedAt", "ReplayOfWorkId", "ProjectionKey", "ProjectionActiveKey",
			"DependencyKind", "DependencyObjectAPIName", "DependencyRecordId", "DependencyKeyHash",
			"DependencyConnectorOwnerKey", "DependencyConnectorNamespace", "DependencyConnectorSourceObject",
			"DependencyDestinationSlotId", "DependencyConnectorRemoteId", "DependencyBlockedAt", "DependencyProbeAfter",
		}))
	if err := NewRepository(db, nil, nil).VerifyWorkItemSchema(context.Background()); err != nil {
		t.Fatalf("VerifyWorkItemSchema: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestDependencySuspensionUsesLeaseFreeProcessingCompatibilityFence(t *testing.T) {
	predicate := activeWorkItemBlockerPredicate("older")
	if strings.Count(predicate, "older.State = ?") != 1 {
		t.Fatalf("active blocker predicate must retain the processing compatibility fence: %s", predicate)
	}
	args := laneHeadWorkItemBlockerArgs(time.Now().UTC())
	if len(args) < 1 || args[0] != WorkStateProcessing {
		t.Fatalf("lane blocker args = %#v, want processing compatibility fence", args)
	}
	if WorkStateDependencyBlocked != WorkStateProcessing {
		t.Fatalf("dependency compatibility alias = %q, want processing", WorkStateDependencyBlocked)
	}
	for _, forbidden := range []string{"DependencyKind", "LeaseOwner", "LeaseToken", "LeaseUntil"} {
		if strings.Contains(predicate, forbidden) {
			t.Fatalf("baseline claim predicate must not require dependency schema column %q: %s", forbidden, predicate)
		}
	}
	staleClaim := claimCandidateWorkItemPredicate("w", true)
	if !strings.Contains(staleClaim, "w.LeaseUntil IS NOT NULL") || !strings.Contains(staleClaim, "w.LeaseUntil <= ?") {
		t.Fatalf("legacy stale-claim predicate must require an expired non-null lease: %s", staleClaim)
	}

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()
	now := time.Date(2026, time.July, 23, 1, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(time.Minute)
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorSchedulerLease.*LeaseUntil <= \?`).
		WithArgs("owner-1", "fence-1", leaseUntil, now, now, "pack-1", int64(7), now).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorSchedulerLease l.*NOT EXISTS.*w.State = \?.*w.LeaseUntil > \?`).
		WithArgs("owner-1", "fence-1", leaseUntil, now, now, "pack-1", int64(7), now, WorkStateProcessing, now).
		WillReturnResult(sqlmock.NewResult(0, 1))
	acquired, err := NewRepository(db, nil, nil).AcquireSchedulerLease(context.Background(), SchedulerLeaseInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil || !acquired {
		t.Fatalf("AcquireSchedulerLease = (%v, %v), want lease-free dependency suspension not to block scheduler", acquired, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestNonTerminalWorkStateSQLListIncludesEveryRecoverableState(t *testing.T) {
	for _, state := range []string{
		WorkStatePending,
		WorkStateRetry,
		WorkStateBlocked,
		WorkStateBlockedRetry,
		WorkStateDependencyBlocked,
		WorkStateProcessing,
	} {
		if !strings.Contains(NonTerminalWorkStateSQLList, "'"+state+"'") {
			t.Fatalf("NonTerminalWorkStateSQLList = %q, missing %q", NonTerminalWorkStateSQLList, state)
		}
	}
	for _, state := range []string{WorkStateProcessed, WorkStateSkipped, WorkStateDead} {
		if strings.Contains(NonTerminalWorkStateSQLList, "'"+state+"'") {
			t.Fatalf("NonTerminalWorkStateSQLList = %q, contains terminal state %q", NonTerminalWorkStateSQLList, state)
		}
	}
}