package automation

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"testing"

	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/persistence"
)

type connectorInboundRecoveryBatchTxStub struct {
	active        bool
	callbacks     int
	replaySuccess bool
	completion    ports.TransactionCompletionOutcome
	returnErr     error
	afterCallback func()
}

func (s *connectorInboundRecoveryBatchTxStub) RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	runs := 1
	if s.replaySuccess {
		runs = 2
	}
	for i := 0; i < runs; i++ {
		s.callbacks++
		s.active = true
		err := fn(ctx)
		s.active = false
		if err != nil {
			return err
		}
	}
	return nil
}

func (s *connectorInboundRecoveryBatchTxStub) IsTransactionActive(context.Context) bool {
	return s.active
}

func (s *connectorInboundRecoveryBatchTxStub) RunInPessimisticTransactionWithOutcome(
	ctx context.Context,
	fn func(context.Context) error,
) (ports.TransactionCompletionOutcome, error) {
	err := s.RunInPessimisticTransaction(ctx, fn)
	if s.afterCallback != nil {
		s.afterCallback()
	}
	if s.returnErr != nil {
		err = s.returnErr
	}
	if s.completion != "" {
		return s.completion, err
	}
	if err != nil {
		return ports.TransactionCompletionRolledBack, err
	}
	return ports.TransactionCompletionCommitted, nil
}

type connectorInboundRecoveryBatchStoreStub struct {
	evidenceByRemote  map[string]ConnectorInboundRemoteIDEvidence
	authorityByRemote map[string]ConnectorRemoteIdentityAuthority
	recoveryOrder     []string
	resolveCalls      int
	resolveErr        error
}

func newConnectorInboundRecoveryBatchStoreStub(items []ConnectorInboundRemoteIdentityRecoveryBatchItem) *connectorInboundRecoveryBatchStoreStub {
	store := &connectorInboundRecoveryBatchStoreStub{
		evidenceByRemote:  make(map[string]ConnectorInboundRemoteIDEvidence, len(items)),
		authorityByRemote: make(map[string]ConnectorRemoteIdentityAuthority, len(items)),
	}
	for i, item := range items {
		evidence := *connectorInboundRecoveryTestEvidence()
		evidence.ID = fmt.Sprintf("evidence-%02d", i+1)
		evidence.RemoteId = item.RemoteID
		evidence.SourceObject = item.ExpectedSourceObject
		evidence.SourceRecordId = item.ExpectedSourceRecordID
		evidence.ActiveKey = ConnectorInboundEvidenceActiveKey(
			evidence.ConnectorOwnerKey,
			evidence.SourceObject,
			evidence.SourceRecordId,
			evidence.RemoteIDField,
		)
		store.evidenceByRemote[item.RemoteID] = evidence
	}
	return store
}

func (s *connectorInboundRecoveryBatchStoreStub) FindInboundRemoteIDEvidenceHintsInAmbientTx(
	_ context.Context,
	selector ConnectorInboundRemoteIdentityRecoverySelector,
) ([]ConnectorInboundRemoteIDEvidence, error) {
	evidence, ok := s.evidenceByRemote[selector.RemoteID]
	if !ok {
		return nil, nil
	}
	return []ConnectorInboundRemoteIDEvidence{evidence}, nil
}

func (s *connectorInboundRecoveryBatchStoreStub) FindUniqueInboundRemoteIDEvidenceForRecoveryInAmbientTx(
	_ context.Context,
	selector ConnectorInboundRemoteIdentityRecoverySelector,
	_ []ConnectorInboundRemoteIDEvidenceBinding,
) (*ConnectorInboundRemoteIDEvidence, error) {
	evidence, ok := s.evidenceByRemote[selector.RemoteID]
	if !ok {
		return nil, nil
	}
	copy := evidence
	return &copy, nil
}

func (s *connectorInboundRecoveryBatchStoreStub) LockConnectorRemoteIdentityLaneInAmbientTx(
	_ context.Context,
	_ ConnectorPushDurableIdentity,
	remoteID string,
) error {
	s.recoveryOrder = append(s.recoveryOrder, remoteID)
	return nil
}

func (s *connectorInboundRecoveryBatchStoreStub) FinalizeInboundRemoteIdentityInAmbientTx(
	_ context.Context,
	finalization ConnectorInboundRemoteIdentityFinalization,
) error {
	identity := finalization.Adoption.Identity
	remoteID := finalization.Adoption.RemoteID
	s.authorityByRemote[remoteID] = ConnectorRemoteIdentityAuthority{
		LinkID:                    "link-row-" + remoteID,
		LinkKey:                   finalization.Adoption.LinkKey,
		LinkStatus:                ConnectorPushLinkStatusFinalized,
		RemoteID:                  remoteID,
		RemoteLocatorHash:         ConnectorPushRemoteLocatorHash(identity, remoteID),
		Identity:                  identity,
		RemoteIDField:             finalization.Evidence.RemoteIDField,
		CompatibilityMirrorStatus: string(finalization.CompatibilityMirrorStatus),
	}
	evidence := s.evidenceByRemote[remoteID]
	evidence.EvidenceStatus = ConnectorInboundEvidenceStatusConsumed
	evidence.ConsumedByLinkKey = finalization.Adoption.LinkKey
	evidence.ActiveKey = ""
	s.evidenceByRemote[remoteID] = evidence
	return nil
}

func (*connectorInboundRecoveryBatchStoreStub) AssertExactInboundRemoteIDEvidenceRecoveryInAmbientTx(
	context.Context,
	ConnectorInboundRemoteIdentityRecoverySelector,
	[]ConnectorInboundRemoteIDEvidenceBinding,
	string,
	string,
) error {
	return nil
}

func (*connectorInboundRecoveryBatchStoreStub) CheckConnectorRemoteIdentityRecoveryHistoryInAmbientTx(
	context.Context,
	ConnectorInboundRemoteIdentityFinalization,
) (bool, error) {
	return false, nil
}

func (*connectorInboundRecoveryBatchStoreStub) AssertExactConnectorRemoteIdentityRecoveryHistoryInAmbientTx(
	context.Context,
	ConnectorInboundRemoteIdentityFinalization,
) error {
	return nil
}

func (s *connectorInboundRecoveryBatchStoreStub) ResolveFinalizedRemoteLocator(
	_ context.Context,
	query ConnectorRemoteLocatorQuery,
) (*ConnectorRemoteIdentityAuthority, error) {
	s.resolveCalls++
	if s.resolveErr != nil {
		return nil, s.resolveErr
	}
	authority, ok := s.authorityByRemote[query.RemoteID]
	if !ok {
		return nil, nil
	}
	copy := authority
	return &copy, nil
}

func (s *connectorInboundRecoveryBatchStoreStub) ReadConnectorInboundRemoteIdentityRecoveryEvidence(
	_ context.Context,
	evidenceID string,
) (*ConnectorInboundRemoteIDEvidence, error) {
	for _, evidence := range s.evidenceByRemote {
		if evidence.ID == evidenceID {
			copy := evidence
			return &copy, nil
		}
	}
	return nil, nil
}

func (*connectorInboundRecoveryBatchStoreStub) RecordCompatibilityMirrorReadback(context.Context, string, string) error {
	return nil
}

func (s *connectorInboundRecoveryBatchStoreStub) restoreAuthored(remoteIDs ...string) {
	for _, remoteID := range remoteIDs {
		delete(s.authorityByRemote, remoteID)
		evidence := s.evidenceByRemote[remoteID]
		evidence.EvidenceStatus = ConnectorInboundEvidenceStatusAuthored
		evidence.ConsumedByLinkKey = ""
		evidence.ActiveKey = ConnectorInboundEvidenceActiveKey(
			evidence.ConnectorOwnerKey,
			evidence.SourceObject,
			evidence.SourceRecordId,
			evidence.RemoteIDField,
		)
		s.evidenceByRemote[remoteID] = evidence
	}
}

func connectorInboundRecoveryBatchTestItems(n int) []ConnectorInboundRemoteIdentityRecoveryBatchItem {
	items := make([]ConnectorInboundRemoteIdentityRecoveryBatchItem, 0, n)
	for i := 0; i < n; i++ {
		items = append(items, ConnectorInboundRemoteIdentityRecoveryBatchItem{
			RemoteID:               fmt.Sprintf("remote-%02d", i+1),
			ExpectedSourceObject:   "Lead",
			ExpectedSourceRecordID: fmt.Sprintf("00Q-source-%02d", i+1),
		})
	}
	return items
}

func connectorInboundRecoveryBatchTestRequest(items []ConnectorInboundRemoteIdentityRecoveryBatchItem) ConnectorInboundRemoteIdentityRecoveryBatchRequest {
	return ConnectorInboundRemoteIdentityRecoveryBatchRequest{
		Selector: ConnectorInboundRemoteIdentityRecoveryBatchSelector{
			ConnectorOwnerKey:  "owner-key",
			ConnectorNamespace: "hubspot",
			DestinationSlotID:  "hubspot-contact",
		},
		Items: items,
	}
}

func newConnectorInboundRecoveryBatchTestService(
	tx *connectorInboundRecoveryBatchTxStub,
	store *connectorInboundRecoveryBatchStoreStub,
	installed *connectorInboundRecoveryInstalledResolverStub,
) *ConnectorInboundRemoteIdentityRecoveryService {
	return NewConnectorInboundRemoteIdentityRecoveryService(
		tx,
		store,
		installed,
		&connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorDrifted},
	)
}

func TestConnectorInboundRemoteIdentityRecoveryBatchSortsLocatorLanesAndVerifiesReadback(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(15)
	for left, right := 0, len(items)-1; left < right; left, right = left+1, right-1 {
		items[left], items[right] = items[right], items[left]
	}
	tx := &connectorInboundRecoveryBatchTxStub{}
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	result, err := newConnectorInboundRecoveryBatchTestService(tx, store, connectorInboundRecoveryTestInstalled()).RecoverBatch(
		context.Background(),
		connectorInboundRecoveryBatchTestRequest(items),
	)
	if err != nil {
		t.Fatalf("RecoverBatch: %v", err)
	}
	if result.Outcome != ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedVerified ||
		!result.Committed || !result.ReadbackVerified || !result.EnrollmentAllowed ||
		result.RequestedCount != 15 || result.RecoveredCount != 15 || len(result.Items) != 15 {
		t.Fatalf("batch result = %#v", result)
	}
	if tx.callbacks != 1 || store.resolveCalls != 15 {
		t.Fatalf("transaction/readback calls = %d/%d, want 1/15", tx.callbacks, store.resolveCalls)
	}
	expectedOrder := append([]string(nil), store.recoveryOrder...)
	sort.Slice(expectedOrder, func(i, j int) bool {
		identity := ConnectorPushDurableIdentity{ConnectorOwnerKey: "owner-key", DestinationSlotID: "hubspot-contact"}
		return ConnectorPushRemoteLocatorHash(identity, expectedOrder[i]) < ConnectorPushRemoteLocatorHash(identity, expectedOrder[j])
	})
	if !reflect.DeepEqual(store.recoveryOrder, expectedOrder) {
		t.Fatalf("recovery order = %v, want locator-hash order %v", store.recoveryOrder, expectedOrder)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchDryRunUsesMutationPathAndSkipsPostRollbackResolver(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(15)
	tx := &connectorInboundRecoveryBatchTxStub{}
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	result, err := newConnectorInboundRecoveryBatchTestService(tx, store, connectorInboundRecoveryTestInstalled()).DryRunBatch(
		context.Background(),
		connectorInboundRecoveryBatchTestRequest(items),
	)
	if err != nil {
		t.Fatalf("DryRunBatch: %v", err)
	}
	if result.Outcome != ConnectorInboundRemoteIdentityRecoveryBatchOutcomeDryRunVerified || result.Committed ||
		result.ReadbackVerified || result.EnrollmentAllowed || result.RecoveredCount != 15 || len(store.recoveryOrder) != 15 {
		t.Fatalf("dry-run result/order = %#v/%v", result, store.recoveryOrder)
	}
	if store.resolveCalls != 0 {
		t.Fatalf("dry-run resolver calls = %d, want 0 after sentinel rollback", store.resolveCalls)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchRejectsDuplicateBeforeTransaction(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(2)
	items[1].RemoteID = items[0].RemoteID
	tx := &connectorInboundRecoveryBatchTxStub{}
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	result, err := newConnectorInboundRecoveryBatchTestService(tx, store, connectorInboundRecoveryTestInstalled()).RecoverBatch(
		context.Background(),
		connectorInboundRecoveryBatchTestRequest(items),
	)
	if err == nil || !strings.Contains(err.Error(), "duplicate remote ID") {
		t.Fatalf("duplicate error = %v", err)
	}
	if result.Outcome != ConnectorInboundRemoteIdentityRecoveryBatchOutcomeRolledBackBlocked || tx.callbacks != 0 || len(store.recoveryOrder) != 0 {
		t.Fatalf("duplicate result/tx/order = %#v/%d/%v", result, tx.callbacks, store.recoveryOrder)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchExpectedPairIsPostSelectionGuard(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(15)
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	request := connectorInboundRecoveryBatchTestRequest(items)
	request.Items[14].ExpectedSourceRecordID = "00Q-wrong-guard"
	result, err := newConnectorInboundRecoveryBatchTestService(
		&connectorInboundRecoveryBatchTxStub{},
		store,
		connectorInboundRecoveryTestInstalled(),
	).RecoverBatch(context.Background(), request)
	if err == nil || !strings.Contains(err.Error(), "does not match expected source") {
		t.Fatalf("expected-pair guard error = %v", err)
	}
	if result.Committed || result.RecoveredCount != 0 || store.resolveCalls != 0 || len(store.recoveryOrder) == 0 {
		t.Fatalf("expected-pair guard did not run after actual selection: result=%#v resolver=%d order=%v", result, store.resolveCalls, store.recoveryOrder)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchRetryReplacesAttemptResults(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(3)
	tx := &connectorInboundRecoveryBatchTxStub{replaySuccess: true}
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	result, err := newConnectorInboundRecoveryBatchTestService(tx, store, connectorInboundRecoveryTestInstalled()).RecoverBatch(
		context.Background(),
		connectorInboundRecoveryBatchTestRequest(items),
	)
	if err != nil {
		t.Fatalf("RecoverBatch replay: %v", err)
	}
	if tx.callbacks != 2 || len(store.recoveryOrder) != 6 || len(result.Items) != 3 || result.RecoveredCount != 3 || store.resolveCalls != 3 {
		t.Fatalf("replay tx/order/result/readback = %d/%d/%#v/%d", tx.callbacks, len(store.recoveryOrder), result, store.resolveCalls)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchGenerationDriftBlocksBeforeCommit(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(4)
	installed := connectorInboundRecoveryTestInstalled()
	installed.onCall = func() {
		if installed.calls == 3 {
			installed.result.Generation = ""
		}
	}
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	result, err := newConnectorInboundRecoveryBatchTestService(
		&connectorInboundRecoveryBatchTxStub{},
		store,
		installed,
	).RecoverBatch(context.Background(), connectorInboundRecoveryBatchTestRequest(items))
	if err == nil || !strings.Contains(err.Error(), "locked installed generation") {
		t.Fatalf("generation drift error = %v", err)
	}
	if result.Committed || result.RecoveredCount != 0 || store.resolveCalls != 0 {
		t.Fatalf("generation drift result/resolver = %#v/%d", result, store.resolveCalls)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchPostCommitReadbackFailureIsNotRollback(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(2)
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	store.resolveErr = errors.New("resolver unavailable")
	result, err := newConnectorInboundRecoveryBatchTestService(
		&connectorInboundRecoveryBatchTxStub{},
		store,
		connectorInboundRecoveryTestInstalled(),
	).RecoverBatch(context.Background(), connectorInboundRecoveryBatchTestRequest(items))
	if !errors.Is(err, ErrConnectorInboundRemoteIdentityRecoveryBatchReadbackBlocked) {
		t.Fatalf("readback error = %v", err)
	}
	if result.Outcome != ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedReadbackBlocked ||
		!result.Committed || result.ReadbackVerified || result.EnrollmentAllowed || result.RecoveredCount != 2 {
		t.Fatalf("committed readback-blocked result = %#v", result)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchCommitAppliedAcknowledgementLostUsesExactReadback(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(3)
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	tx := &connectorInboundRecoveryBatchTxStub{
		completion: ports.TransactionCompletionCommitError,
		returnErr:  errors.New("commit acknowledgement lost"),
	}
	result, err := newConnectorInboundRecoveryBatchTestService(tx, store, connectorInboundRecoveryTestInstalled()).RecoverBatch(
		context.Background(), connectorInboundRecoveryBatchTestRequest(items),
	)
	if err != nil {
		t.Fatalf("ambiguous commit with exact applied readback: %v", err)
	}
	if result.Outcome != ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedVerified ||
		!result.Committed || !result.ReadbackVerified || !result.EnrollmentAllowed ||
		result.RecoveredCount != len(items) || len(result.Items) != len(items) {
		t.Fatalf("applied acknowledgement-lost result = %#v", result)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchCommitErrorZeroSnapshotRemainsCommitUnknown(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(3)
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	remoteIDs := make([]string, 0, len(items))
	for _, item := range items {
		remoteIDs = append(remoteIDs, item.RemoteID)
	}
	tx := &connectorInboundRecoveryBatchTxStub{
		completion: ports.TransactionCompletionCommitError,
		returnErr:  errors.New("commit rejected"),
		afterCallback: func() {
			store.restoreAuthored(remoteIDs...)
		},
	}
	result, err := newConnectorInboundRecoveryBatchTestService(tx, store, connectorInboundRecoveryTestInstalled()).RecoverBatch(
		context.Background(), connectorInboundRecoveryBatchTestRequest(items),
	)
	if !errors.Is(err, ErrConnectorInboundRemoteIdentityRecoveryBatchReadbackBlocked) {
		t.Fatalf("zero-snapshot commit error = %v", err)
	}
	if result.Outcome != ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedReadbackBlocked ||
		!result.Committed || result.ReadbackVerified || result.EnrollmentAllowed ||
		result.RecoveredCount != 0 || len(result.Items) != 0 {
		t.Fatalf("zero-snapshot commit-unknown result = %#v", result)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchAmbiguousPartialReadbackBlocksEnrollment(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(3)
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	tx := &connectorInboundRecoveryBatchTxStub{
		completion: ports.TransactionCompletionCommitError,
		returnErr:  errors.New("commit outcome unknown"),
		afterCallback: func() {
			store.restoreAuthored(items[1].RemoteID, items[2].RemoteID)
		},
	}
	result, err := newConnectorInboundRecoveryBatchTestService(tx, store, connectorInboundRecoveryTestInstalled()).RecoverBatch(
		context.Background(), connectorInboundRecoveryBatchTestRequest(items),
	)
	if !errors.Is(err, ErrConnectorInboundRemoteIdentityRecoveryBatchReadbackBlocked) {
		t.Fatalf("partial readback error = %v", err)
	}
	if result.Outcome != ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedReadbackBlocked ||
		!result.Committed || result.ReadbackVerified || result.EnrollmentAllowed ||
		result.RecoveredCount != 1 || len(result.Items) != 1 {
		t.Fatalf("partial readback result = %#v", result)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchDryRunRollbackFailureNeverVerifies(t *testing.T) {
	items := connectorInboundRecoveryBatchTestItems(3)
	store := newConnectorInboundRecoveryBatchStoreStub(items)
	tx := &connectorInboundRecoveryBatchTxStub{
		completion: ports.TransactionCompletionRollbackError,
		returnErr:  fmt.Errorf("rollback failed after callback: %w", errConnectorInboundRemoteIdentityRecoveryDryRunRollback),
	}
	result, err := newConnectorInboundRecoveryBatchTestService(tx, store, connectorInboundRecoveryTestInstalled()).DryRunBatch(
		context.Background(), connectorInboundRecoveryBatchTestRequest(items),
	)
	if !errors.Is(err, ErrConnectorInboundRemoteIdentityRecoveryBatchReadbackBlocked) {
		t.Fatalf("rollback failure error = %v", err)
	}
	if result.Outcome == ConnectorInboundRemoteIdentityRecoveryBatchOutcomeDryRunVerified ||
		!result.Committed || result.ReadbackVerified || result.EnrollmentAllowed {
		t.Fatalf("rollback failure must remain blocked: %#v", result)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryBatchItemHasGuardOnlyInputs(t *testing.T) {
	typ := reflect.TypeOf(ConnectorInboundRemoteIdentityRecoveryBatchItem{})
	want := []string{"RemoteID", "ExpectedSourceObject", "ExpectedSourceRecordID"}
	if typ.NumField() != len(want) {
		t.Fatalf("batch item fields = %v", reflect.VisibleFields(typ))
	}
	for i, name := range want {
		if typ.Field(i).Name != name {
			t.Fatalf("batch item field[%d] = %s, want %s", i, typ.Field(i).Name, name)
		}
	}
}

type connectorInboundRecoveryBatchTiDBTxManager struct {
	inner *persistence.TransactionManager
}

func (m connectorInboundRecoveryBatchTiDBTxManager) RunInPessimisticTransaction(
	ctx context.Context,
	fn func(context.Context) error,
) error {
	return m.inner.WithPessimisticRetry(ctx, fn, 3)
}

func (m connectorInboundRecoveryBatchTiDBTxManager) RunInPessimisticTransactionWithOutcome(
	ctx context.Context,
	fn func(context.Context) error,
) (ports.TransactionCompletionOutcome, error) {
	var completion ports.TransactionCompletionOutcome
	err := m.inner.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		ports.RegisterTransactionCompletionHook(txCtx, func(outcome ports.TransactionCompletionOutcome) {
			completion = outcome
		})
		return fn(txCtx)
	}, 3)
	return completion, err
}

func (m connectorInboundRecoveryBatchTiDBTxManager) IsTransactionActive(ctx context.Context) bool {
	return m.inner.IsTransactionActive(ctx)
}

func (m connectorInboundRecoveryBatchTiDBTxManager) ExtractTx(ctx context.Context) interface{} {
	return m.inner.ExtractTx(ctx)
}

func TestConnectorInboundRemoteIdentityRecoveryBatchRealTiDBAtomicity(t *testing.T) {
	for _, tc := range []struct {
		name         string
		dryRun       bool
		mismatch15   bool
		wantLinks    int
		wantConsumed int
	}{
		{name: "fifteenth mismatch rolls back all", mismatch15: true, wantLinks: 0, wantConsumed: 0},
		{name: "dry run rolls back all", dryRun: true, wantLinks: 0, wantConsumed: 0},
		{name: "fifteen success commit and resolve", wantLinks: 15, wantConsumed: 15},
	} {
		t.Run(tc.name, func(t *testing.T) {
			db, cleanup := openTiDB(t)
			defer cleanup()
			db.SetMaxOpenConns(8)
			ctx := context.Background()
			createConnectorRuntimeOwnerCatalogTables(t, db)
			for _, ddl := range []string{
				connectorRemoteLinkTestDDL,
				connectorPushAttemptTestDDL,
				connectorInboundEvidenceTestDDL,
			} {
				if _, err := db.ExecContext(ctx, ddl); err != nil {
					t.Fatalf("create connector recovery table: %v", err)
				}
			}

			items := connectorInboundRecoveryBatchTestItems(15)
			identity := ConnectorPushDurableIdentity{ConnectorOwnerKey: "owner-key", DestinationSlotID: "hubspot-contact"}
			sort.Slice(items, func(i, j int) bool {
				return ConnectorPushRemoteLocatorHash(identity, items[i].RemoteID) < ConnectorPushRemoteLocatorHash(identity, items[j].RemoteID)
			})
			plainStore := NewSQLConnectorPushDurableStore(db)
			for i, item := range items {
				evidence := ConnectorInboundRemoteIDEvidence{
					ID:                 fmt.Sprintf("tidb-batch-evidence-%02d", i+1),
					ConnectorOwnerKey:  "owner-key",
					ConnectorNamespace: "hubspot",
					SourceObject:       item.ExpectedSourceObject,
					SourceRecordId:     item.ExpectedSourceRecordID,
					RemoteIDField:      "HubSpot_Id__c",
					RemoteId:           item.RemoteID,
				}
				if err := plainStore.InsertAuthoredEvidence(ctx, evidence); err != nil {
					t.Fatalf("seed evidence %d: %v", i, err)
				}
			}

			requestItems := append([]ConnectorInboundRemoteIdentityRecoveryBatchItem(nil), items...)
			if tc.mismatch15 {
				requestItems[14].ExpectedSourceRecordID = "00Q-deliberate-mismatch"
			}
			innerTx := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
			innerTx.SetPessimisticLockWaitTimeout(10)
			txManager := connectorInboundRecoveryBatchTiDBTxManager{inner: innerTx}
			txStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, txManager)
			service := NewConnectorInboundRemoteIdentityRecoveryService(
				txManager,
				txStore,
				connectorInboundRecoveryTestInstalled(),
				&connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorDrifted},
			)

			var result ConnectorInboundRemoteIdentityRecoveryBatchResult
			var err error
			if tc.dryRun {
				result, err = service.DryRunBatch(ctx, connectorInboundRecoveryBatchTestRequest(requestItems))
			} else {
				result, err = service.RecoverBatch(ctx, connectorInboundRecoveryBatchTestRequest(requestItems))
			}
			if tc.mismatch15 {
				if err == nil || !strings.Contains(err.Error(), "does not match expected source") {
					t.Fatalf("fifteenth mismatch error = %v", err)
				}
				if result.Committed {
					t.Fatalf("fifteenth mismatch reported committed result %#v", result)
				}
			} else if err != nil {
				t.Fatalf("batch operation: %v", err)
			}

			links, consumed, authored := connectorInboundRecoveryBatchTiDBCounts(t, db)
			if links != tc.wantLinks || consumed != tc.wantConsumed || authored != 15-tc.wantConsumed {
				t.Fatalf("durable batch state links=%d consumed=%d authored=%d, want %d/%d/%d", links, consumed, authored, tc.wantLinks, tc.wantConsumed, 15-tc.wantConsumed)
			}
			if !tc.dryRun && !tc.mismatch15 && (!result.Committed || !result.ReadbackVerified || !result.EnrollmentAllowed || result.RecoveredCount != 15) {
				t.Fatalf("success result = %#v", result)
			}
			if tc.dryRun && (result.Outcome != ConnectorInboundRemoteIdentityRecoveryBatchOutcomeDryRunVerified || result.Committed || result.RecoveredCount != 15) {
				t.Fatalf("dry-run result = %#v", result)
			}
		})
	}
}

func connectorInboundRecoveryBatchTiDBCounts(t *testing.T, db *sql.DB) (links, consumed, authored int) {
	t.Helper()
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE Status = ?`, ConnectorPushLinkStatusFinalized).Scan(&links); err != nil {
		t.Fatalf("count finalized links: %v", err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorInboundRemoteIDEvidence WHERE EvidenceStatus = ?`, ConnectorInboundEvidenceStatusConsumed).Scan(&consumed); err != nil {
		t.Fatalf("count consumed evidence: %v", err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorInboundRemoteIDEvidence WHERE EvidenceStatus = ?`, ConnectorInboundEvidenceStatusAuthored).Scan(&authored); err != nil {
		t.Fatalf("count authored evidence: %v", err)
	}
	return links, consumed, authored
}