package automation

import (
	"context"
	"database/sql/driver"
	"errors"
	"reflect"
	"strings"
	"testing"
)

type connectorInboundRecoveryTxStub struct {
	active          bool
	pessimistic     bool
	calls           int
	replayCallbacks int
	callbackCalls   int
}

func (s *connectorInboundRecoveryTxStub) RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	s.calls++
	s.active = true
	s.pessimistic = true
	defer func() {
		s.active = false
		s.pessimistic = false
	}()
	attempts := s.replayCallbacks
	if attempts < 1 {
		attempts = 1
	}
	var err error
	for attempt := 0; attempt < attempts; attempt++ {
		s.callbackCalls++
		err = fn(ctx)
	}
	return err
}

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

type connectorInboundRecoveryStoreStub struct {
	evidence       *ConnectorInboundRemoteIDEvidence
	hints          []ConnectorInboundRemoteIDEvidence
	evidenceErr    error
	finalizeErr    error
	readbackErr    error
	historyErr     error
	preflightErr   error
	laneErr        error
	lockedEvidence *ConnectorInboundRemoteIDEvidence
	finalization   *ConnectorInboundRemoteIdentityFinalization
	selector       ConnectorInboundRemoteIdentityRecoverySelector
	evidenceBounds []ConnectorInboundRemoteIDEvidenceBinding
	callOrder      []string
}

func (s *connectorInboundRecoveryStoreStub) FindInboundRemoteIDEvidenceHintsInAmbientTx(
	_ context.Context,
	selector ConnectorInboundRemoteIdentityRecoverySelector,
) ([]ConnectorInboundRemoteIDEvidence, error) {
	s.callOrder = append(s.callOrder, "hint")
	s.selector = selector
	if s.evidenceErr != nil {
		return nil, s.evidenceErr
	}
	if s.hints != nil {
		return append([]ConnectorInboundRemoteIDEvidence(nil), s.hints...), nil
	}
	if s.evidence == nil {
		return nil, nil
	}
	return []ConnectorInboundRemoteIDEvidence{*s.evidence}, nil
}

func (s *connectorInboundRecoveryStoreStub) FindUniqueInboundRemoteIDEvidenceForRecoveryInAmbientTx(
	_ context.Context,
	selector ConnectorInboundRemoteIdentityRecoverySelector,
	bindings []ConnectorInboundRemoteIDEvidenceBinding,
) (*ConnectorInboundRemoteIDEvidence, error) {
	s.callOrder = append(s.callOrder, "evidence-lock")
	s.selector = selector
	s.evidenceBounds = append([]ConnectorInboundRemoteIDEvidenceBinding(nil), bindings...)
	if s.lockedEvidence != nil {
		return s.lockedEvidence, s.evidenceErr
	}
	return s.evidence, s.evidenceErr
}

func (s *connectorInboundRecoveryStoreStub) LockConnectorRemoteIdentityLaneInAmbientTx(
	_ context.Context,
	_ ConnectorPushDurableIdentity,
	_ string,
) error {
	s.callOrder = append(s.callOrder, "lane")
	return s.laneErr
}

func (s *connectorInboundRecoveryStoreStub) FinalizeInboundRemoteIdentityInAmbientTx(_ context.Context, finalization ConnectorInboundRemoteIdentityFinalization) error {
	s.callOrder = append(s.callOrder, "finalize")
	if s.finalizeErr != nil {
		return s.finalizeErr
	}
	s.finalization = &finalization
	return nil
}

func (s *connectorInboundRecoveryStoreStub) AssertExactInboundRemoteIDEvidenceRecoveryInAmbientTx(
	_ context.Context,
	_ ConnectorInboundRemoteIdentityRecoverySelector,
	_ []ConnectorInboundRemoteIDEvidenceBinding,
	_ string,
	_ string,
) error {
	s.callOrder = append(s.callOrder, "evidence-readback")
	return s.readbackErr
}

func (s *connectorInboundRecoveryStoreStub) AssertExactConnectorRemoteIdentityRecoveryHistoryInAmbientTx(
	_ context.Context,
	_ ConnectorInboundRemoteIdentityFinalization,
) error {
	s.callOrder = append(s.callOrder, "history-readback")
	return s.historyErr
}

func (s *connectorInboundRecoveryStoreStub) CheckConnectorRemoteIdentityRecoveryHistoryInAmbientTx(
	_ context.Context,
	_ ConnectorInboundRemoteIdentityFinalization,
) (bool, error) {
	s.callOrder = append(s.callOrder, "history-preflight")
	return false, s.preflightErr
}

func (s *connectorInboundRecoveryStoreStub) ReadConnectorInboundRemoteIdentityRecoveryEvidence(
	context.Context,
	string,
) (*ConnectorInboundRemoteIDEvidence, error) {
	if s.evidence == nil {
		return nil, s.evidenceErr
	}
	copy := *s.evidence
	return &copy, s.evidenceErr
}

type connectorInboundRecoveryInstalledResolverStub struct {
	result ConnectorInboundRemoteIdentityRecoveryInstalledBindings
	err    error
	calls  int
	onCall func()
}

func (s *connectorInboundRecoveryInstalledResolverStub) ResolveConnectorInboundRemoteIdentityRecoveryInstalledBindings(
	_ context.Context,
	_ ConnectorInboundRemoteIdentityRecoverySelector,
) (ConnectorInboundRemoteIdentityRecoveryInstalledBindings, error) {
	s.calls++
	if s.onCall != nil {
		s.onCall()
	}
	return s.result, s.err
}

type connectorInboundRecoverySourceValidatorStub struct {
	status ConnectorPushCompatibilityMirrorStatus
	err    error
	calls  int
}

func (s *connectorInboundRecoverySourceValidatorStub) ValidateConnectorInboundRemoteIdentityRecoverySource(
	_ context.Context,
	_ ConnectorInboundRemoteIDEvidence,
	_ ConnectorInboundRemoteIdentityRecoveryBinding,
) (ConnectorInboundRemoteIdentityRecoverySourceValidation, error) {
	s.calls++
	return ConnectorInboundRemoteIdentityRecoverySourceValidation{CompatibilityMirrorStatus: s.status}, s.err
}

func connectorInboundRecoveryTestRequest() ConnectorInboundRemoteIdentityRecoveryRequest {
	return ConnectorInboundRemoteIdentityRecoveryRequest{Selector: connectorInboundRecoveryTestSelector()}
}

func connectorInboundRecoveryTestSelector() ConnectorInboundRemoteIdentityRecoverySelector {
	return ConnectorInboundRemoteIdentityRecoverySelector{
		ConnectorOwnerKey: "owner-key", ConnectorNamespace: "hubspot",
		DestinationSlotID: "hubspot-contact", RemoteID: "12345",
	}
}

func connectorInboundRecoveryTestInstalled() *connectorInboundRecoveryInstalledResolverStub {
	identity := connectorPushTestIdentity()
	identity.SourceObject = "Lead"
	identity.SourceRecordID = ""
	identity.DestinationSlotID = "hubspot-contact"
	snapshot := connectorPushTestSnapshot()
	snapshot.ConnectorNamespace = "hubspot"
	snapshot.RemoteIDField = "HubSpot_Id__c"
	snapshot.ProfileContractHash = ""
	return &connectorInboundRecoveryInstalledResolverStub{result: ConnectorInboundRemoteIdentityRecoveryInstalledBindings{
		Generation: "installed-generation-1",
		Bindings: []ConnectorInboundRemoteIdentityRecoveryBinding{{
			EvidenceBinding: ConnectorInboundRemoteIDEvidenceBinding{
				SourceObject:  "Lead",
				RemoteIDField: "HubSpot_Id__c",
			},
			Identity:                  identity,
			Snapshot:                  snapshot,
			RecoverySourceEligibility: []string{"Archived"},
		}},
	}}
}

func connectorInboundRecoveryTestEvidence() *ConnectorInboundRemoteIDEvidence {
	return &ConnectorInboundRemoteIDEvidence{
		ID:                 "evidence-15",
		ConnectorOwnerKey:  "owner-key",
		ConnectorNamespace: "hubspot",
		SourceObject:       "Lead",
		SourceRecordId:     "00Q-source",
		RemoteIDField:      "HubSpot_Id__c",
		RemoteId:           "12345",
		EvidenceStatus:     ConnectorInboundEvidenceStatusAuthored,
		ActiveKey:          ConnectorInboundEvidenceActiveKey("owner-key", "Lead", "00Q-source", "HubSpot_Id__c"),
	}
}

func TestConnectorInboundRemoteIdentityRecoveryRejectsMalformedEvidenceDimensions(t *testing.T) {
	selector := connectorInboundRecoveryTestRequest().Selector
	for _, tc := range []struct {
		name   string
		mutate func(*ConnectorInboundRemoteIDEvidence)
	}{
		{name: "blank source object", mutate: func(e *ConnectorInboundRemoteIDEvidence) { e.SourceObject = "" }},
		{name: "blank source record id", mutate: func(e *ConnectorInboundRemoteIDEvidence) { e.SourceRecordId = "" }},
		{name: "blank remote id field", mutate: func(e *ConnectorInboundRemoteIDEvidence) { e.RemoteIDField = "" }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			evidence := connectorInboundRecoveryTestEvidence()
			tc.mutate(evidence)
			err := validateConnectorInboundRemoteIdentityRecoveryEvidence(selector, *evidence)
			if !errors.Is(err, ErrConnectorPushDurableConflict) {
				t.Fatalf("malformed evidence error = %v, want durable conflict", err)
			}
		})
	}
}

func TestConnectorInboundRemoteIdentityRecoveryUsesStrictFinalizerInsidePessimisticTransaction(t *testing.T) {
	for _, tc := range []struct {
		name   string
		status ConnectorPushCompatibilityMirrorStatus
	}{
		{name: "blank compatibility mirror remains drifted", status: ConnectorPushCompatibilityMirrorDrifted},
		{name: "exact compatibility mirror remains synced", status: ConnectorPushCompatibilityMirrorSynced},
	} {
		t.Run(tc.name, func(t *testing.T) {
			txm := &connectorInboundRecoveryTxStub{}
			store := &connectorInboundRecoveryStoreStub{evidence: connectorInboundRecoveryTestEvidence()}
			installed := connectorInboundRecoveryTestInstalled()
			validator := &connectorInboundRecoverySourceValidatorStub{status: tc.status}
			service := NewConnectorInboundRemoteIdentityRecoveryService(txm, store, installed, validator)

			result, err := service.Recover(context.Background(), connectorInboundRecoveryTestRequest())
			if err != nil {
				t.Fatalf("Recover: %v", err)
			}
			if result == nil || result.EvidenceID != "evidence-15" || result.SourceRecordID != "00Q-source" {
				t.Fatalf("Recover result = %#v", result)
			}
			if txm.calls != 1 || installed.calls != 1 || validator.calls != 1 {
				t.Fatalf("tx/installed/validator calls = %d/%d/%d, want 1/1/1", txm.calls, installed.calls, validator.calls)
			}
			if strings.Join(store.callOrder, ",") != "hint,lane,evidence-lock,history-preflight,finalize,evidence-readback,history-readback" {
				t.Fatalf("call order = %v", store.callOrder)
			}
			if store.finalization == nil {
				t.Fatal("strict finalizer was not called")
			}
			got := store.finalization
			if got.Evidence.ID != "evidence-15" || got.Adoption.Identity.SourceRecordID != "00Q-source" {
				t.Fatalf("exact evidence/owner = %q/%q", got.Evidence.ID, got.Adoption.Identity.SourceRecordID)
			}
			if got.CompatibilityMirrorStatus != tc.status {
				t.Fatalf("mirror status = %q, want %q", got.CompatibilityMirrorStatus, tc.status)
			}
			if got.Adoption.LinkKey != result.LinkKey || got.Adoption.RemoteID != "12345" {
				t.Fatalf("adoption = %#v, result = %#v", got.Adoption, result)
			}
		})
	}
}

func TestConnectorInboundRemoteIdentityRecoveryPessimisticOwnerReplaysCompleteRecoveryCallback(t *testing.T) {
	// The production owner retries its complete callback for TiDB 1205/1213.
	// Replay twice here to prove recovery does not retain an inner-only body.
	txm := &connectorInboundRecoveryTxStub{replayCallbacks: 2}
	store := &connectorInboundRecoveryStoreStub{evidence: connectorInboundRecoveryTestEvidence()}
	installed := connectorInboundRecoveryTestInstalled()
	validator := &connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorSynced}
	service := NewConnectorInboundRemoteIdentityRecoveryService(txm, store, installed, validator)

	result, err := service.Recover(context.Background(), connectorInboundRecoveryTestRequest())
	if err != nil {
		t.Fatalf("Recover replay: %v", err)
	}
	if result == nil || result.EvidenceID != "evidence-15" {
		t.Fatalf("Recover replay result = %#v", result)
	}
	if txm.calls != 1 || txm.callbackCalls != 2 || installed.calls != 2 || validator.calls != 2 {
		t.Fatalf("manager/callback/installed/validator calls = %d/%d/%d/%d, want 1/2/2/2", txm.calls, txm.callbackCalls, installed.calls, validator.calls)
	}
	wantAttempt := "hint,lane,evidence-lock,history-preflight,finalize,evidence-readback,history-readback"
	if got := strings.Join(store.callOrder, ","); got != wantAttempt+","+wantAttempt {
		t.Fatalf("replayed recovery call order = %q, want two complete attempts", got)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryRequestIsSelectorOnly(t *testing.T) {
	typ := reflect.TypeOf(ConnectorInboundRemoteIdentityRecoveryRequest{})
	if typ.NumField() != 1 || typ.Field(0).Name != "Selector" {
		t.Fatalf("recovery request fields = %v, want selector only", reflect.VisibleFields(typ))
	}
}

func TestConnectorInboundRemoteIdentityRecoveryLocksInstalledGenerationAfterHint(t *testing.T) {
	txm := &connectorInboundRecoveryTxStub{}
	store := &connectorInboundRecoveryStoreStub{evidence: connectorInboundRecoveryTestEvidence()}
	installed := connectorInboundRecoveryTestInstalled()
	installed.onCall = func() {
		if strings.Join(store.callOrder, ",") != "hint" {
			t.Fatalf("installed resolver call order = %v, want hint first", store.callOrder)
		}
	}
	validator := &connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorDrifted}
	if _, err := NewConnectorInboundRemoteIdentityRecoveryService(txm, store, installed, validator).Recover(context.Background(), connectorInboundRecoveryTestRequest()); err != nil {
		t.Fatalf("Recover: %v", err)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryRejectsGenerationDriftBeforeSourceMutation(t *testing.T) {
	txm := &connectorInboundRecoveryTxStub{}
	store := &connectorInboundRecoveryStoreStub{evidence: connectorInboundRecoveryTestEvidence()}
	installed := connectorInboundRecoveryTestInstalled()
	installed.result.Generation = ""
	validator := &connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorDrifted}
	_, err := NewConnectorInboundRemoteIdentityRecoveryService(txm, store, installed, validator).Recover(context.Background(), connectorInboundRecoveryTestRequest())
	if err == nil || !strings.Contains(err.Error(), "locked installed generation") {
		t.Fatalf("generation drift error = %v", err)
	}
	if validator.calls != 0 || store.finalization != nil {
		t.Fatalf("generation drift reached mutation: validator=%d finalization=%#v", validator.calls, store.finalization)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryRejectsEvidenceChangeAfterLane(t *testing.T) {
	txm := &connectorInboundRecoveryTxStub{}
	changed := connectorInboundRecoveryTestEvidence()
	changed.ID = "evidence-raced"
	store := &connectorInboundRecoveryStoreStub{
		evidence:       connectorInboundRecoveryTestEvidence(),
		lockedEvidence: changed,
	}
	installed := connectorInboundRecoveryTestInstalled()
	validator := &connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorDrifted}
	_, err := NewConnectorInboundRemoteIdentityRecoveryService(txm, store, installed, validator).Recover(context.Background(), connectorInboundRecoveryTestRequest())
	if !errors.Is(err, ErrConnectorPushDurableConflict) {
		t.Fatalf("evidence race error = %v, want durable conflict", err)
	}
	if store.finalization != nil {
		t.Fatalf("evidence race finalized %#v", store.finalization)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryReturnsNotFoundWithoutMutation(t *testing.T) {
	txm := &connectorInboundRecoveryTxStub{}
	store := &connectorInboundRecoveryStoreStub{}
	installed := connectorInboundRecoveryTestInstalled()
	validator := &connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorDrifted}
	service := NewConnectorInboundRemoteIdentityRecoveryService(txm, store, installed, validator)

	result, err := service.Recover(context.Background(), connectorInboundRecoveryTestRequest())
	if err != nil {
		t.Fatalf("Recover absent: %v", err)
	}
	if result != nil {
		t.Fatalf("Recover absent result = %#v, want nil", result)
	}
	if validator.calls != 0 || store.finalization != nil {
		t.Fatalf("absent recovery mutated: validator=%d finalization=%#v", validator.calls, store.finalization)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryCarriesExactConsumedEvidenceRetry(t *testing.T) {
	request := connectorInboundRecoveryTestRequest()
	installed := connectorInboundRecoveryTestInstalled()
	identity := installed.result.Bindings[0].Identity
	identity.SourceRecordID = "00Q-source"
	linkKey, err := identity.LinkKey()
	if err != nil {
		t.Fatalf("LinkKey: %v", err)
	}
	evidence := connectorInboundRecoveryTestEvidence()
	evidence.EvidenceStatus = ConnectorInboundEvidenceStatusConsumed
	evidence.ActiveKey = ""
	evidence.ConsumedByLinkKey = linkKey
	txm := &connectorInboundRecoveryTxStub{}
	store := &connectorInboundRecoveryStoreStub{evidence: evidence}
	validator := &connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorDrifted}

	result, err := NewConnectorInboundRemoteIdentityRecoveryService(txm, store, installed, validator).Recover(context.Background(), request)
	if err != nil {
		t.Fatalf("Recover consumed exact retry: %v", err)
	}
	if result == nil || result.LinkKey != linkKey || store.finalization == nil {
		t.Fatalf("consumed retry result/finalization = %#v/%#v", result, store.finalization)
	}
	if store.finalization.Evidence.ID != evidence.ID || store.finalization.Evidence.EvidenceStatus != ConnectorInboundEvidenceStatusConsumed {
		t.Fatalf("consumed retry evidence = %#v", store.finalization.Evidence)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryRejectsAmbientTransaction(t *testing.T) {
	txm := &connectorInboundRecoveryTxStub{active: true, pessimistic: true}
	store := &connectorInboundRecoveryStoreStub{evidence: connectorInboundRecoveryTestEvidence()}
	installed := connectorInboundRecoveryTestInstalled()
	validator := &connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorDrifted}

	_, err := NewConnectorInboundRemoteIdentityRecoveryService(txm, store, installed, validator).Recover(context.Background(), connectorInboundRecoveryTestRequest())
	if err == nil || !strings.Contains(err.Error(), "rejects an ambient transaction") {
		t.Fatalf("ambient recovery error = %v", err)
	}
	if txm.calls != 0 || len(store.callOrder) != 0 {
		t.Fatalf("ambient recovery opened tx or store: tx=%d calls=%v", txm.calls, store.callOrder)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryAmbientBodyDoesNotNestTransaction(t *testing.T) {
	txm := &connectorInboundRecoveryTxStub{active: true, pessimistic: true}
	store := &connectorInboundRecoveryStoreStub{evidence: connectorInboundRecoveryTestEvidence()}
	installed := connectorInboundRecoveryTestInstalled()
	validator := &connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorDrifted}
	service := NewConnectorInboundRemoteIdentityRecoveryService(txm, store, installed, validator)

	result, err := service.recoverInAmbientTx(context.Background(), connectorInboundRecoveryTestRequest())
	if err != nil {
		t.Fatalf("recoverInAmbientTx: %v", err)
	}
	if result == nil || result.EvidenceID != "evidence-15" {
		t.Fatalf("ambient recovery result = %#v", result)
	}
	if txm.calls != 0 {
		t.Fatalf("ambient recovery nested %d transaction(s), want 0", txm.calls)
	}
	if strings.Join(store.callOrder, ",") != "hint,lane,evidence-lock,history-preflight,finalize,evidence-readback,history-readback" {
		t.Fatalf("call order = %v", store.callOrder)
	}
}

func TestConnectorInboundRemoteIdentityRecoveryFailsClosedBeforeFinalization(t *testing.T) {
	tests := []struct {
		name      string
		mutate    func(*ConnectorInboundRemoteIdentityRecoveryRequest, *connectorInboundRecoveryStoreStub, *connectorInboundRecoveryInstalledResolverStub, *connectorInboundRecoverySourceValidatorStub)
		wantError string
	}{
		{
			name: "competing durable authority",
			mutate: func(_ *ConnectorInboundRemoteIdentityRecoveryRequest, store *connectorInboundRecoveryStoreStub, _ *connectorInboundRecoveryInstalledResolverStub, _ *connectorInboundRecoverySourceValidatorStub) {
				store.finalizeErr = ErrConnectorPushDurableConflict
			},
			wantError: "connector push durable conflict",
		},
		{
			name: "ambiguous evidence",
			mutate: func(_ *ConnectorInboundRemoteIdentityRecoveryRequest, store *connectorInboundRecoveryStoreStub, _ *connectorInboundRecoveryInstalledResolverStub, _ *connectorInboundRecoverySourceValidatorStub) {
				store.evidenceErr = ErrConnectorPushDurableConflict
			},
			wantError: "connector push durable conflict",
		},
		{
			name: "deleted source",
			mutate: func(_ *ConnectorInboundRemoteIdentityRecoveryRequest, _ *connectorInboundRecoveryStoreStub, _ *connectorInboundRecoveryInstalledResolverStub, validator *connectorInboundRecoverySourceValidatorStub) {
				validator.err = errors.New("source is deleted")
			},
			wantError: "source is deleted",
		},
		{
			name: "operator injected source record",
			mutate: func(_ *ConnectorInboundRemoteIdentityRecoveryRequest, _ *connectorInboundRecoveryStoreStub, installed *connectorInboundRecoveryInstalledResolverStub, _ *connectorInboundRecoverySourceValidatorStub) {
				installed.result.Bindings[0].Identity.SourceRecordID = "00Q-injected"
			},
			wantError: "SourceRecordID must be empty",
		},
		{
			name: "binding outside destination",
			mutate: func(_ *ConnectorInboundRemoteIdentityRecoveryRequest, _ *connectorInboundRecoveryStoreStub, installed *connectorInboundRecoveryInstalledResolverStub, _ *connectorInboundRecoverySourceValidatorStub) {
				installed.result.Bindings[0].Identity.DestinationSlotID = "other-destination"
			},
			wantError: "destination slot",
		},
		{
			name: "unsupported mirror result",
			mutate: func(_ *ConnectorInboundRemoteIdentityRecoveryRequest, _ *connectorInboundRecoveryStoreStub, _ *connectorInboundRecoveryInstalledResolverStub, validator *connectorInboundRecoverySourceValidatorStub) {
				validator.status = ConnectorPushCompatibilityMirrorPending
			},
			wantError: "mirror status",
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			request := connectorInboundRecoveryTestRequest()
			txm := &connectorInboundRecoveryTxStub{}
			store := &connectorInboundRecoveryStoreStub{evidence: connectorInboundRecoveryTestEvidence()}
			installed := connectorInboundRecoveryTestInstalled()
			validator := &connectorInboundRecoverySourceValidatorStub{status: ConnectorPushCompatibilityMirrorDrifted}
			tc.mutate(&request, store, installed, validator)

			_, err := NewConnectorInboundRemoteIdentityRecoveryService(txm, store, installed, validator).Recover(context.Background(), request)
			if err == nil || !strings.Contains(err.Error(), tc.wantError) {
				t.Fatalf("Recover error = %v, want %q", err, tc.wantError)
			}
			if store.finalization != nil {
				t.Fatalf("fail-closed path finalized %#v", store.finalization)
			}
		})
	}
}

func connectorInboundRegistrationHistoryColumns() []string {
	return []string{"Id", "LinkKey", "Status", "RemoteId", "RemoteLocatorHash", "RemoteLocatorActiveKey", "ConnectorOwnerKey", "ActionSlotId", "ActionType", "SourceObject", "SourceRecordId", "DestinationSlotId"}
}

func TestSQLConnectorInboundRecoveryHistoryUnionBlocksCorruptMaterialization(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	ctx := context.Background()
	remoteLinkDDL := strings.Replace(connectorRemoteLinkTestDDL, "RemoteLocatorActiveKey VARCHAR(255) NOT NULL", "RemoteLocatorActiveKey VARCHAR(255) NULL", 1)
	remoteLinkDDL = strings.Replace(remoteLinkDDL, "RemoteLocatorHash VARCHAR(255) NOT NULL", "RemoteLocatorHash VARCHAR(255) NULL", 1)
	createConnectorRuntimeOwnerCatalogTables(t, db)
	for _, ddl := range []string{
		connectorInboundEvidenceTestDDL,
		remoteLinkDDL,
		connectorPushAttemptTestDDL,
	} {
		if _, err := db.ExecContext(ctx, ddl); err != nil {
			t.Fatalf("create recovery history table: %v", err)
		}
	}

	finalization := connectorInboundTestFinalization()
	store := NewSQLConnectorPushDurableStore(db)
	if err := store.InsertAuthoredEvidence(ctx, finalization.Evidence); err != nil {
		t.Fatalf("seed authored evidence: %v", err)
	}
	finalizeTx, err := db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatalf("begin canonical finalization: %v", err)
	}
	finalizeStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: finalizeTx})
	if err := finalizeStore.LockConnectorRemoteIdentityLaneInAmbientTx(ctx, finalization.Adoption.Identity, finalization.Adoption.RemoteID); err != nil {
		_ = finalizeTx.Rollback()
		t.Fatalf("lock locator lane: %v", err)
	}
	if err := finalizeStore.FinalizeInboundRemoteIdentityInAmbientTx(ctx, finalization); err != nil {
		_ = finalizeTx.Rollback()
		t.Fatalf("finalize exact history seed: %v", err)
	}
	if err := finalizeTx.Commit(); err != nil {
		t.Fatalf("commit exact history seed: %v", err)
	}
	expectedIdentity := finalization.Adoption.Identity
	expectedRemoteID := finalization.Adoption.RemoteID
	expectedLocatorHash := ConnectorPushRemoteLocatorHash(expectedIdentity, expectedRemoteID)
	assertBlocked := func(want string) {
		t.Helper()
		tx, err := db.BeginTx(ctx, nil)
		if err != nil {
			t.Fatalf("begin strict history check: %v", err)
		}
		checkStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: tx})
		err = checkStore.AssertExactConnectorRemoteIdentityRecoveryHistoryInAmbientTx(ctx, finalization)
		_ = tx.Rollback()
		if !errors.Is(err, ErrConnectorPushDurableConflict) || !strings.Contains(err.Error(), want) {
			t.Fatalf("strict history error = %v, want durable conflict containing %q", err, want)
		}
	}
	setLogicalLocator := func(owner, destinationSlotID, remoteID string) {
		t.Helper()
		if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET ConnectorOwnerKey = ?, DestinationSlotId = ?, RemoteId = ?
 WHERE LinkKey = ?`, owner, destinationSlotID, remoteID, finalization.Adoption.LinkKey); err != nil {
			t.Fatalf("set logical locator tuple: %v", err)
		}
	}

	setLogicalLocator("corrupt-history-owner", expectedIdentity.DestinationSlotID, expectedRemoteID)
	assertBlocked("not the exact finalized recovery authority")
	setLogicalLocator(expectedIdentity.ConnectorOwnerKey, "corrupt-history-destination", expectedRemoteID)
	assertBlocked("not the exact finalized recovery authority")
	setLogicalLocator(expectedIdentity.ConnectorOwnerKey, expectedIdentity.DestinationSlotID, "corrupt-history-remote")
	assertBlocked("not the exact finalized recovery authority")
	setLogicalLocator(expectedIdentity.ConnectorOwnerKey, expectedIdentity.DestinationSlotID, expectedRemoteID)

	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET RemoteLocatorHash = NULL
 WHERE LinkKey = ?`, finalization.Adoption.LinkKey); err != nil {
		t.Fatalf("clear materialized locator hash: %v", err)
	}
	assertBlocked("not the exact finalized recovery authority")

	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET RemoteLocatorHash = ?
 WHERE LinkKey = ?`, "wrong-materialized-hash", finalization.Adoption.LinkKey); err != nil {
		t.Fatalf("corrupt materialized locator hash: %v", err)
	}
	assertBlocked("not the exact finalized recovery authority")

	// Make the original row match only the materialized-hash arm and the second
	// row match only the logical-tuple arm. The union must surface both rows and
	// block rather than choosing either as recovery authority.
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET RemoteLocatorHash = ?, ConnectorOwnerKey = ?
 WHERE LinkKey = ?`, expectedLocatorHash, "corrupt-history-union-owner", finalization.Adoption.LinkKey); err != nil {
		t.Fatalf("seed materialized-hash-only history: %v", err)
	}
	otherIdentity := finalization.Adoption.Identity
	otherIdentity.ActionSlotID = "corrupt-history-second-slot"
	otherIdentity.SourceRecordID = "003-corrupt-history-second-source"
	otherLinkKey := mustConnectorPushTestLinkKey(otherIdentity)
	insertSyncLink(t, db, "link-corrupt-history-second", otherLinkKey, ConnectorPushLinkStatusRetired, finalization.Adoption.RemoteID,
		otherIdentity.ConnectorOwnerKey, otherIdentity.ActionSlotID, otherIdentity.ActionType,
		otherIdentity.SourceObject, otherIdentity.SourceRecordID, otherIdentity.DestinationSlotID, finalization.Adoption.Snapshot.TargetObject)
	minimalProfileHash := ConnectorPushProfileContractHash(ConnectorPushDurableSnapshot{
		RetryPolicy: ConnectorPushRetryPolicy{MaxAttempts: 1},
	})
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET ProfileContractHash = ?
 WHERE Id = ?`, minimalProfileHash, "link-corrupt-history-second"); err != nil {
		t.Fatalf("seed valid logical-tuple-only history snapshot: %v", err)
	}
	assertBlocked("multiple history rows")
}

func connectorInboundRecoveryHistoryColumns() []string {
	return []string{
		"Id", "LinkKey", "Status", "RemoteId", "ConnectorOwnerKey", "ActionSlotId", "ActionType",
		"SourceObject", "SourceRecordId", "DestinationSlotId", "RequesterOwnerPack", "ProfileOwnerPack",
		"ConnectorNamespace", "FlowDeveloperName", "FlowActionName", "TargetObject", "ProfileName",
		"CreatedPackInstallId", "LastValidatedPackInstallId", "CredentialRef", "EndpointClass", "RetryPolicy",
		"ProfileContractHash", "CreatePathTemplate", "UpdatePathTemplate", "ResponseIdPath", "RemoteIdField",
		"RemoteIdProvenanceField", "MappingFingerprint", "BodyFingerprint", "RemoteLocatorHash",
		"RemoteLocatorActiveKey", "CompatibilityMirrorStatus",
	}
}

func connectorInboundRecoveryHistoryValues(link ConnectorPushRemoteLink, locatorHash, activeKey, mirrorStatus string) []driver.Value {
	snapshot := link.Snapshot
	return []driver.Value{
		link.ID, link.LinkKey, link.Status, link.RemoteID,
		link.Identity.ConnectorOwnerKey, link.Identity.ActionSlotID, link.Identity.ActionType,
		link.Identity.SourceObject, link.Identity.SourceRecordID, link.Identity.DestinationSlotID,
		snapshot.RequesterOwnerPack, snapshot.ProfileOwnerPack, snapshot.ConnectorNamespace,
		snapshot.FlowDeveloperName, snapshot.FlowActionName, snapshot.TargetObject, snapshot.ProfileName,
		snapshot.CreatedPackInstallID, snapshot.LastValidatedPackInstallID, snapshot.CredentialRef,
		snapshot.EndpointClass, mustConnectorPushJSON(snapshot.RetryPolicy), snapshot.ProfileContractHash,
		snapshot.CreatePathTemplate, snapshot.UpdatePathTemplate, snapshot.ResponseIDPath,
		snapshot.RemoteIDField, snapshot.RemoteIDProvenanceField, snapshot.MappingFingerprint,
		snapshot.BodyFingerprint, locatorHash, activeKey, mirrorStatus,
	}
}