package automation

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"

	database "github.com/nexuscrm/db"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/persistence"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

// createConnectorPushDurableCatalogTables builds the durable-store tables from
// the embedded system catalog. The integration oracle therefore follows the
// production schema owner instead of carrying another hand-copied DDL model.
func createConnectorPushDurableCatalogTables(t *testing.T, db *sql.DB, additionalTables ...string) {
	t.Helper()
	createConnectorDurableNamedCatalogTables(t, db, append([]string{
		"_System_ConnectorRemoteLink", "_System_ConnectorRemoteIdentityLane", "_System_ConnectorPushAttempt", "_System_RuntimeOutboundIntent", "_System_RuntimeOutboundIntentGeneration",
	}, additionalTables...)...)
}

func createConnectorRuntimeOwnerCatalogTables(t *testing.T, db *sql.DB) {
	t.Helper()
	createConnectorDurableNamedCatalogTables(t, db, "_System_ConnectorRemoteIdentityLane", "_System_RuntimeOutboundIntent", "_System_RuntimeOutboundIntentGeneration")
}

func createConnectorDurableNamedCatalogTables(t *testing.T, db *sql.DB, tables ...string) {
	t.Helper()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("load system table definitions: %v", err)
	}
	wanted := make(map[string]bool, len(tables))
	for _, table := range tables {
		wanted[table] = false
	}
	repository := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, ok := wanted[definition.TableName]; !ok {
			continue
		}
		if err := repository.CreatePhysicalTable(t.Context(), definition); err != nil {
			t.Fatalf("create %s from system catalog: %v", definition.TableName, err)
		}
		wanted[definition.TableName] = true
	}
	for table, found := range wanted {
		if !found {
			t.Fatalf("system catalog does not declare %s", table)
		}
	}
}

// The transfer fixtures use the same catalog schema as deployment, including
// the Runtime-owned receipt tables; no permissive copied DDL or SQL mock hides
// uniqueness, nullability, transaction, or row-lock behavior.
func connectorIntegrationTransferFixture(suffix string) ConnectorRemoteIdentityTransfer {
	source, target := connectorAuthorityLockOrderIdentities(suffix)
	return ConnectorRemoteIdentityTransfer{
		SourceObject: source.SourceObject, SourceRecordId: source.SourceRecordID,
		TargetObject: target.SourceObject, TargetRecordId: target.SourceRecordID,
		RemoteIDField: "External_Id__c", RemoteId: "remote-" + suffix,
		admissions: []connectorRemoteIdentityTransferAdmission{{
			ConnectorNamespace: source.ConnectorOwnerKey, RemoteID: "remote-" + suffix,
			SourceIdentity: source, TargetIdentity: target,
			TargetSnapshot:         connectorTransferTargetSnapshot(target, "External_Id__c", target.ConnectorOwnerKey),
			SourceIntegrationScope: connectorIntegrationOutboundScope{"test-org", "test-app", "lead-producer"},
			TargetIntegrationScope: connectorIntegrationOutboundScope{"test-org", "test-app", "contact-producer"},
		}},
	}
}

func connectorTransferTargetSnapshot(identity ConnectorPushDurableIdentity, field, namespace string) ConnectorPushDurableSnapshot {
	snapshot := connectorAuthorityLockOrderAdoption(identity, "fixture-remote", field).Snapshot
	snapshot.RequesterOwnerPack = identity.ConnectorOwnerKey
	snapshot.ConnectorNamespace = namespace
	snapshot.LastValidatedPackInstallID = snapshot.CreatedPackInstallID
	snapshot.ProfileContractHash = ConnectorPushProfileContractHash(snapshot)
	return snapshot
}

func connectorTransferFixtureSnapshots(transfer ConnectorRemoteIdentityTransfer) ConnectorRemoteIdentityTransfer {
	transfer.admissions = append([]connectorRemoteIdentityTransferAdmission(nil), transfer.admissions...)
	for i := range transfer.admissions {
		admission := &transfer.admissions[i]
		admission.TargetSnapshot = connectorTransferTargetSnapshot(admission.TargetIdentity, transfer.RemoteIDField, admission.ConnectorNamespace)
	}
	return transfer
}

func seedConnectorIntegrationTransferLink(t *testing.T, db *sql.DB, identity ConnectorPushDurableIdentity, scope connectorIntegrationOutboundScope, field, remoteID string, scoped bool) string {
	t.Helper()
	adoption := connectorAuthorityLockOrderAdoption(identity, remoteID, field)
	adoption.Snapshot.ConnectorNamespace = identity.ConnectorOwnerKey
	adoption.Snapshot.ProfileContractHash = ConnectorPushProfileContractHash(adoption.Snapshot)
	if _, err := NewSQLConnectorPushDurableStore(db).AdoptFinalizedLink(t.Context(), adoption); err != nil {
		t.Fatalf("seed finalized link: %v", err)
	}
	if !scoped {
		return adoption.LinkKey
	}
	linkKey := connectorIntegrationOutboundLinkKey(scope, identity)
	locator := connectorIntegrationOutboundRemoteLocatorHash(scope, identity, remoteID)
	if _, err := db.ExecContext(t.Context(), `UPDATE _System_ConnectorRemoteLink
 SET LinkKey = ?, RemoteLocatorHash = ?, RemoteLocatorActiveKey = ? WHERE LinkKey = ?`,
		linkKey, locator, locator, adoption.LinkKey); err != nil {
		t.Fatalf("seed scoped producer link: %v", err)
	}
	return linkKey
}

// seedConnectorIntegrationTransferCommand retains its fixture-call signature,
// but authority is seeded entirely through the Runtime repository. Integration's
// private delivery tables are deliberately absent from these databases.
func seedConnectorIntegrationTransferCommand(t *testing.T, db *sql.DB, id, _ string, state string, identity ConnectorPushDurableIdentity, scope connectorIntegrationOutboundScope, remoteID string) {
	t.Helper()
	intent := connectorRuntimeTransferIntent(t, id, identity)
	if strings.HasSuffix(state, "-delete") {
		intent.Operation = outboundintentv1.OperationDelete
		// Resource intent payloads are opaque to the neutral repository. Freeze
		// a canonical DELETE request for the already-known provider identity;
		// the refusal is settled by the real admission/result APIs below.
		payload, err := json.Marshal(struct {
			Version  uint32 `json:"version"`
			RemoteID string `json:"remoteId"`
		}{Version: 1, RemoteID: remoteID})
		if err != nil {
			t.Fatal(err)
		}
		digest := sha256.Sum256(payload)
		intent.PayloadFormatVersion, intent.CanonicalPayload, intent.PayloadDigest = 1, payload, "sha256:"+hex.EncodeToString(digest[:])
		state = strings.TrimSuffix(state, "-delete")
	}
	if err := outboundintentv1.ValidateIntent(intent); err != nil {
		t.Fatalf("invalid canonical receipt fixture intent: %v", err)
	}
	if err := writeConnectorRuntimeTransferReceipt(t.Context(), db, intent, scope, state, remoteID); err != nil {
		t.Fatalf("seed Runtime outbound receipt: %v", err)
	}
}

func connectorRuntimeTransferIntent(t *testing.T, id string, identity ConnectorPushDurableIdentity) outboundintentv1.Intent {
	t.Helper()
	intent := runtimeOutboundIntentTestIntent(runtimeOutboundIntentTestView(t))
	intent.IntentID, intent.IdempotencyKey = id, id
	intent.SourceObject, intent.SourceRecordID = identity.SourceObject, identity.SourceRecordID
	intent.ConnectorOwnerKey, intent.ActionSlotID, intent.DestinationSlotID = identity.ConnectorOwnerKey, identity.ActionSlotID, identity.DestinationSlotID
	return intent
}

func writeConnectorRuntimeTransferReceipt(ctx context.Context, db *sql.DB, intent outboundintentv1.Intent, scope connectorIntegrationOutboundScope, state, remoteID string) error {
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	repository := persistence.NewRuntimeOutboundIntentRepository(db, transactions)
	runtimeScope := persistence.RuntimeOutboundIntentScope{OrgID: scope.orgID, AppInstallationID: scope.appInstallationID, AuthorizedWorkloadID: scope.authorizedWorkloadID}
	return transactions.WithPessimisticTransaction(ctx, func(ctx context.Context) error {
		if _, _, err := repository.Enqueue(ctx, runtimeScope, intent); err != nil {
			return err
		}
		if state == "pending" {
			return nil
		}
		acknowledgement := outboundintentv1.AcknowledgeRequest{
			IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest, CatalogRevision: intent.CatalogRevision,
			Outcome: outboundintentv1.AdmissionAccepted, IntegrationCommandID: "delivery-" + intent.IntentID,
		}
		if state == "rejected" {
			acknowledgement.Outcome, acknowledgement.IntegrationCommandID, acknowledgement.RejectionCode = outboundintentv1.AdmissionRejected, "", "fixture_rejected"
		}
		if _, _, err := repository.Acknowledge(ctx, runtimeScope, acknowledgement); err != nil {
			return err
		}
		switch state {
		case "accepted", "rejected":
			return nil
		case "succeeded", "no-effect":
			request := outboundintentv1.CommitResultRequest{
				IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest, CatalogRevision: intent.CatalogRevision,
				IntegrationCommandID: acknowledgement.IntegrationCommandID, RemoteID: remoteID,
			}
			if state == "no-effect" {
				canonical, digest, err := outboundintentv1.CanonicalResourceNoEffectResultV1(outboundintentv1.ResourceNoEffectResultV1{Version: 1, Outcome: outboundintentv1.ResourceOutcomeNoEffect, Reason: outboundintentv1.ResourceNoEffectRejectedBeforeDispatch})
				if err != nil {
					return err
				}
				request.RemoteID, request.ResultFormatVersion, request.ResultDigest, request.CanonicalResult = "", outboundintentv1.ResourceNoEffectResultFormatV1, digest, canonical
			}
			_, err := repository.CommitResultWithProjection(ctx, runtimeScope, request, func(context.Context, persistence.RuntimeOutboundIntentRow) error { return nil })
			return err
		default:
			return fmt.Errorf("unsupported Runtime receipt fixture state %q", state)
		}
	})
}

func beginConnectorIntegrationTransfer(t *testing.T, db *sql.DB) (*sql.Tx, *SQLConnectorPushDurableStore) {
	t.Helper()
	tx, err := db.BeginTx(t.Context(), nil)
	if err != nil {
		t.Fatalf("begin transfer: %v", err)
	}
	t.Cleanup(func() { _ = tx.Rollback() })
	return tx, NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: tx})
}

func TestConnectorIntegrationTransferSourcesOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
	for _, kind := range []string{"legacy", "scoped", "duplicate", "different namespace"} {
		t.Run(kind, func(t *testing.T) {
			transfer := connectorIntegrationTransferFixture(kind)
			if kind == "different namespace" {
				transfer.admissions[0].TargetSnapshot = connectorTransferTargetSnapshot(
					transfer.admissions[0].TargetIdentity, transfer.RemoteIDField, "target-profile-namespace")
			}
			admission := transfer.admissions[0]
			if kind != "legacy" {
				linkKey := seedConnectorIntegrationTransferLink(t, db, admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, true)
				seedConnectorIntegrationTransferCommand(t, db, "success-"+kind, linkKey, "succeeded", admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteId)
				if kind == "scoped" {
					for _, settledDelete := range []string{"rejected-delete", "no-effect-delete"} {
						seedConnectorIntegrationTransferCommand(t, db, kind+"-"+settledDelete, linkKey, settledDelete,
							admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteId)
					}
				}
			}
			if kind != "scoped" {
				seedConnectorIntegrationTransferLink(t, db, admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, false)
			}
			tx, store := beginConnectorIntegrationTransfer(t, db)
			got, err := store.TransferInboundRemoteIdentity(t.Context(), transfer)
			if err != nil || got != transfer.RemoteId {
				t.Fatalf("transfer = %q, %v; want exact remote identity", got, err)
			}
			if err := tx.Commit(); err != nil {
				t.Fatal(err)
			}
			var retired, targetEvidence int
			if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_ConnectorRemoteLink
 WHERE SourceObject = ? AND SourceRecordId = ? AND Status = 'retired' AND RemoteLocatorActiveKey IS NULL`, transfer.SourceObject, transfer.SourceRecordId).Scan(&retired); err != nil {
				t.Fatal(err)
			}
			wantRetired := 1
			if kind == "duplicate" || kind == "different namespace" {
				wantRetired = 2
			}
			if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE SourceObject = ? AND SourceRecordId = ?`, transfer.TargetObject, transfer.TargetRecordId).Scan(&targetEvidence); err != nil {
				t.Fatal(err)
			}
			if retired != wantRetired || targetEvidence != 0 {
				t.Fatalf("retired source links = %d, fabricated target evidence = %d; want %d, 0", retired, targetEvidence, wantRetired)
			}
			// No inbound context was used for conversion. The normal later
			// inbound lookup must find the exact finalized target immediately,
			// without inventing a connector-authored write or making provider IO.
			readStore := NewSQLConnectorPushDurableStore(db)
			target, err := readStore.FindFinalizedLink(t.Context(), admission.TargetIdentity)
			if err != nil || target == nil || target.RemoteID != transfer.RemoteId ||
				!connectorPushPersistedRemoteLinkSnapshotEqual(target.Snapshot, admission.TargetSnapshot) {
				t.Fatalf("finalized target snapshot = %#v, %v", target, err)
			}
			authority, err := readStore.ResolveFinalizedRemoteLocator(t.Context(), ConnectorRemoteLocatorQuery{
				ConnectorOwnerKey: admission.TargetIdentity.ConnectorOwnerKey,
				DestinationSlotID: admission.TargetIdentity.DestinationSlotID, RemoteID: transfer.RemoteId,
			})
			if err != nil || authority == nil || authority.Identity != admission.TargetIdentity || authority.LinkKey != target.LinkKey {
				t.Fatalf("later inbound authority lookup = %#v, %v", authority, err)
			}
			var mirrorStatus string
			if err := db.QueryRowContext(t.Context(), `SELECT CompatibilityMirrorStatus FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, target.LinkKey).Scan(&mirrorStatus); err != nil || mirrorStatus != ConnectorPushCompatibilityMirrorPending {
				t.Fatalf("unverified target mirror status = %q, %v; want pending", mirrorStatus, err)
			}
			// An identical replay retains exactly one target authority.
			transfer.TargetRemoteID = transfer.RemoteId
			replayTx, replay := beginConnectorIntegrationTransfer(t, db)
			got, err = replay.TransferInboundRemoteIdentity(t.Context(), transfer)
			if err != nil || got != transfer.RemoteId {
				t.Fatalf("completed finalized-target replay = %q, %v", got, err)
			}
			_ = replayTx.Rollback()
			if kind != "scoped" {
				adoption := connectorAuthorityLockOrderAdoption(admission.SourceIdentity, transfer.RemoteId, transfer.RemoteIDField)
				adoption.Snapshot.ConnectorNamespace = admission.SourceIdentity.ConnectorOwnerKey
				adoption.Snapshot.ProfileContractHash = ConnectorPushProfileContractHash(adoption.Snapshot)
				if _, err := NewSQLConnectorPushDurableStore(db).AdoptFinalizedLink(t.Context(), adoption); !errors.Is(err, ErrConnectorPushDurableConflict) {
					t.Fatalf("retired source re-adoption = %v; want conflict", err)
				}
			}
		})
	}
}

func TestConnectorIntegrationTransferMirrorsAreNotAuthorityOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
	for _, mirrors := range []struct {
		name, source, target string
	}{
		{name: "source only", source: "mirror-only"},
		{name: "target only", target: "mirror-only"},
		{name: "matching mirrors", source: "mirror-only", target: "mirror-only"},
	} {
		t.Run(mirrors.name, func(t *testing.T) {
			transfer := connectorIntegrationTransferFixture(mirrors.name)
			// With no persisted authority, the engine has no admissible hints.
			transfer.admissions = nil
			transfer.RemoteId, transfer.TargetRemoteID = mirrors.source, mirrors.target
			tx, store := beginConnectorIntegrationTransfer(t, db)
			if remoteID, err := store.TransferInboundRemoteIdentity(t.Context(), transfer); err != nil || remoteID != "" {
				t.Fatalf("mirror-only transfer returned authority %q, %v; want empty, nil", remoteID, err)
			}
			if err := tx.Commit(); err != nil {
				t.Fatal(err)
			}
			var links, evidence int
			if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE SourceObject = ? AND SourceRecordId = ?`, transfer.TargetObject, transfer.TargetRecordId).Scan(&links); err != nil {
				t.Fatal(err)
			}
			if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_ConnectorInboundRemoteIDEvidence WHERE SourceObject = ? AND SourceRecordId = ?`, transfer.TargetObject, transfer.TargetRecordId).Scan(&evidence); err != nil {
				t.Fatal(err)
			}
			if links != 0 || evidence != 0 {
				t.Fatalf("mirror-only transfer fabricated links/evidence = %d/%d", links, evidence)
			}
		})
	}
}

func TestConnectorIntegrationTransferAtomicPublicationFailuresOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
	for _, kind := range []string{"target insert conflict", "deleted target owner", "invalid target snapshot"} {
		t.Run(kind, func(t *testing.T) {
			transfer := connectorIntegrationTransferFixture(kind)
			admission := transfer.admissions[0]
			if kind == "target insert conflict" {
				targetKey := seedConnectorIntegrationTransferLink(t, db, admission.TargetIdentity, admission.TargetIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, false)
				if _, err := db.ExecContext(t.Context(), `UPDATE _System_ConnectorRemoteLink SET Status = 'retired', RemoteLocatorActiveKey = NULL WHERE LinkKey = ?`, targetKey); err != nil {
					t.Fatal(err)
				}
			}
			sourceKey := seedConnectorIntegrationTransferLink(t, db, admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, false)
			evidence := ConnectorInboundRemoteIDEvidence{
				ID: "source-evidence-" + kind, ConnectorOwnerKey: admission.SourceIdentity.ConnectorOwnerKey,
				ConnectorNamespace: admission.ConnectorNamespace, SourceObject: transfer.SourceObject,
				SourceRecordId: transfer.SourceRecordId, RemoteIDField: transfer.RemoteIDField, RemoteId: transfer.RemoteId,
			}
			if err := NewSQLConnectorPushDurableStore(db).InsertAuthoredEvidence(t.Context(), evidence); err != nil {
				t.Fatal(err)
			}
			if kind == "deleted target owner" {
				deletedTx, _ := beginConnectorIntegrationTransfer(t, db)
				if err := markConnectorRemoteIdentityOwnerDeletedFenceTx(t.Context(), deletedTx, transfer.TargetObject, transfer.TargetRecordId); err != nil {
					t.Fatal(err)
				}
				if err := deletedTx.Commit(); err != nil {
					t.Fatal(err)
				}
			}
			if kind == "invalid target snapshot" {
				transfer.admissions[0].TargetSnapshot.RemoteIDField = "Foreign_Id__c"
			}
			tx, store := beginConnectorIntegrationTransfer(t, db)
			if _, err := store.TransferInboundRemoteIdentity(t.Context(), transfer); !errors.Is(err, ErrConnectorPushDurableConflict) {
				t.Fatalf("transfer %s = %v; want fail-closed conflict", kind, err)
			}
			if kind == "target insert conflict" {
				var status string
				if err := tx.QueryRowContext(t.Context(), `SELECT Status FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, sourceKey).Scan(&status); err != nil || status != ConnectorPushLinkStatusRetired {
					t.Fatalf("insert failure must occur after source retirement: %q, %v", status, err)
				}
			}
			if err := tx.Rollback(); err != nil {
				t.Fatal(err)
			}
			var sourceStatus, evidenceStatus string
			var targetActive int
			if err := db.QueryRowContext(t.Context(), `SELECT Status FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, sourceKey).Scan(&sourceStatus); err != nil {
				t.Fatal(err)
			}
			if err := db.QueryRowContext(t.Context(), `SELECT EvidenceStatus FROM _System_ConnectorInboundRemoteIDEvidence WHERE Id = ?`, evidence.ID).Scan(&evidenceStatus); err != nil {
				t.Fatal(err)
			}
			if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE SourceObject = ? AND SourceRecordId = ? AND RemoteLocatorActiveKey IS NOT NULL`, transfer.TargetObject, transfer.TargetRecordId).Scan(&targetActive); err != nil {
				t.Fatal(err)
			}
			if sourceStatus != ConnectorPushLinkStatusFinalized || evidenceStatus != ConnectorInboundEvidenceStatusAuthored || targetActive != 0 {
				t.Fatalf("rollback source link/evidence/target active = %s/%s/%d", sourceStatus, evidenceStatus, targetActive)
			}
		})
	}
}

func TestConnectorIntegrationTransferProofRejectionsOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
	cases := []struct {
		name, assignment string
		initialState     string
		value            any
		alterScope       func(*connectorIntegrationOutboundScope)
		extraState       string
		retriable        bool
	}{
		{name: "missing installed scope", alterScope: func(s *connectorIntegrationOutboundScope) { *s = connectorIntegrationOutboundScope{} }},
		{name: "foreign installed org", alterScope: func(s *connectorIntegrationOutboundScope) { s.orgID = "foreign" }},
		{name: "foreign installed app", alterScope: func(s *connectorIntegrationOutboundScope) { s.appInstallationID = "foreign" }},
		{name: "foreign installed workload", alterScope: func(s *connectorIntegrationOutboundScope) { s.authorizedWorkloadID = "foreign" }},
		{name: "corrupt scope", assignment: "OrgId", value: "foreign"},
		{name: "corrupt digest", assignment: "ScopeDigest", value: strings.Repeat("b", 64)},
		{name: "invalid catalog", assignment: "CatalogRevision", value: 0},
		{name: "different operation", assignment: "Operation", value: "DELETE"},
		{name: "different source", assignment: "SourceRecordId", value: "foreign"},
		{name: "different remote", assignment: "ResultRemoteId", value: "foreign"},
		{name: "no terminal proof", initialState: "rejected"},
		{name: "no effect is not remote proof", initialState: "no-effect"},
		{name: "rejected DELETE is not remote proof", initialState: "rejected-delete"},
		{name: "no-effect DELETE is not remote proof", initialState: "no-effect-delete"},
		{name: "unknown state", assignment: "AdmissionState", value: "UNKNOWN", retriable: true},
		{name: "unsupported retired state", assignment: "AdmissionState", value: "RETIRED", retriable: true},
		{name: "foreign scope tail after success", extraState: "pending", retriable: true},
	}
	for _, state := range []string{"pending", "accepted"} {
		cases = append(cases, struct {
			name, assignment string
			initialState     string
			value            any
			alterScope       func(*connectorIntegrationOutboundScope)
			extraState       string
			retriable        bool
		}{name: state + " after success", extraState: state, retriable: true})
	}
	for i, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			transfer := connectorIntegrationTransferFixture(fmt.Sprintf("reject-%d", i))
			admission := transfer.admissions[0]
			linkKey := seedConnectorIntegrationTransferLink(t, db, admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, true)
			commandID := fmt.Sprintf("proof-%d", i)
			initialState := tc.initialState
			if initialState == "" {
				initialState = "succeeded"
			}
			seedConnectorIntegrationTransferCommand(t, db, commandID, linkKey, initialState, admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteId)
			if tc.assignment != "" {
				// assignment is selected only from this test's constant case table.
				if _, err := db.ExecContext(t.Context(), "UPDATE _System_RuntimeOutboundIntent SET "+tc.assignment+" = ? WHERE Id = ?", tc.value, commandID); err != nil {
					t.Fatal(err)
				}
			}
			if tc.alterScope != nil {
				tc.alterScope(&transfer.admissions[0].SourceIntegrationScope)
			}
			if tc.extraState != "" {
				extraScope := admission.SourceIntegrationScope
				if tc.name == "foreign scope tail after success" {
					extraScope.orgID = "foreign"
				}
				seedConnectorIntegrationTransferCommand(t, db, commandID+"-new", linkKey, tc.extraState, admission.SourceIdentity, extraScope, transfer.RemoteId)
			}
			tx, store := beginConnectorIntegrationTransfer(t, db)
			_, err := store.TransferInboundRemoteIdentity(t.Context(), transfer)
			if !errors.Is(err, ErrConnectorPushDurableConflict) {
				t.Fatalf("transfer error = %v; want durable conflict", err)
			}
			var contention *apperrors.LockContentionError
			if errors.As(err, &contention) != tc.retriable {
				t.Fatalf("retry classification = %v; want retriable=%v", err, tc.retriable)
			}
			_ = tx.Rollback()
			var unchanged, targetEvidence int
			if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE LinkKey = ? AND Status = 'finalized' AND RemoteLocatorActiveKey IS NOT NULL`, linkKey).Scan(&unchanged); err != nil {
				t.Fatal(err)
			}
			if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_ConnectorInboundRemoteIDEvidence WHERE SourceObject = ? AND SourceRecordId = ?`, transfer.TargetObject, transfer.TargetRecordId).Scan(&targetEvidence); err != nil {
				t.Fatal(err)
			}
			if unchanged != 1 || targetEvidence != 0 {
				t.Fatalf("failed transfer mutated authority: source=%d, target=%d", unchanged, targetEvidence)
			}
		})
	}
}

func TestConnectorIntegrationTransferTargetReplayOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
	for _, kind := range []string{"legacy", "scoped", "scoped different profile namespace", "duplicate", "wrong target scope", "different mirror", "suspended target", "foreign legacy namespace"} {
		t.Run(kind, func(t *testing.T) {
			transfer := connectorIntegrationTransferFixture("target-" + kind)
			transfer.TargetRemoteID = transfer.RemoteId
			if kind == "scoped different profile namespace" {
				transfer.admissions[0].TargetSnapshot = connectorTransferTargetSnapshot(
					transfer.admissions[0].TargetIdentity, transfer.RemoteIDField, "target-profile-namespace")
			}
			admission := transfer.admissions[0]
			var scopedKey string
			if kind != "legacy" && kind != "foreign legacy namespace" {
				scopedKey = seedConnectorIntegrationTransferLink(t, db, admission.TargetIdentity, admission.TargetIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, true)
				seedConnectorIntegrationTransferCommand(t, db, "target-"+kind, scopedKey, "succeeded", admission.TargetIdentity, admission.TargetIntegrationScope, transfer.RemoteId)
			}
			if kind == "legacy" || kind == "duplicate" || kind == "foreign legacy namespace" {
				legacyKey := seedConnectorIntegrationTransferLink(t, db, admission.TargetIdentity, admission.TargetIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, false)
				if kind == "foreign legacy namespace" {
					if _, err := db.ExecContext(t.Context(), `UPDATE _System_ConnectorRemoteLink SET ConnectorNamespace = 'foreign' WHERE LinkKey = ?`, legacyKey); err != nil {
						t.Fatal(err)
					}
				}
			}
			switch kind {
			case "wrong target scope":
				transfer.admissions[0].TargetIntegrationScope = admission.SourceIntegrationScope
			case "different mirror":
				transfer.TargetRemoteID = "different"
			case "suspended target":
				if _, err := db.ExecContext(t.Context(), `UPDATE _System_ConnectorRemoteLink SET Status = ? WHERE LinkKey = ?`, ConnectorPushLinkStatusSuspendedDrift, scopedKey); err != nil {
					t.Fatal(err)
				}
			}
			tx, store := beginConnectorIntegrationTransfer(t, db)
			got, err := store.TransferInboundRemoteIdentity(t.Context(), transfer)
			if kind == "legacy" || kind == "scoped" || kind == "scoped different profile namespace" || kind == "duplicate" {
				if err != nil || got != transfer.RemoteId {
					t.Fatalf("target replay = %q, %v", got, err)
				}
			} else if !errors.Is(err, ErrConnectorPushDurableConflict) {
				t.Fatalf("target replay = %q, %v; want conflict", got, err)
			}
			_ = tx.Rollback()
		})
	}
}

// A new Runtime intent must be visible despite an older repeatable-read
// snapshot. A concurrently locked receipt must cause prompt retry, preserving
// result-commit lock ordering without consulting Integration delivery state.
func TestConnectorIntegrationTransferFreshCommandFenceOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
	transfer := connectorIntegrationTransferFixture("fresh-fence")
	admission := transfer.admissions[0]
	linkKey := seedConnectorIntegrationTransferLink(t, db, admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, true)
	seedConnectorIntegrationTransferCommand(t, db, "old-success", linkKey, "succeeded", admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteId)
	tx, store := beginConnectorIntegrationTransfer(t, db)
	var count int
	if err := tx.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM _System_RuntimeOutboundIntent WHERE SourceObject = ? AND SourceRecordId = ?`, transfer.SourceObject, transfer.SourceRecordId).Scan(&count); err != nil || count != 1 {
		t.Fatalf("establish old repeatable-read snapshot: %d, %v", count, err)
	}
	seedConnectorIntegrationTransferCommand(t, db, "new-pending", linkKey, "pending", admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteId)
	receiptTx, _ := beginConnectorIntegrationTransfer(t, db)
	var state string
	if err := receiptTx.QueryRowContext(t.Context(), `SELECT AdmissionState FROM _System_RuntimeOutboundIntent WHERE Id = 'new-pending' FOR UPDATE`).Scan(&state); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
	defer cancel()
	_, err := store.TransferInboundRemoteIdentity(ctx, transfer)
	var contention *apperrors.LockContentionError
	if !errors.As(err, &contention) {
		t.Fatalf("fresh locked receipt = %v; want prompt retry", err)
	}
	_ = tx.Rollback()
	_ = receiptTx.Rollback()
	seedConnectorIntegrationTransferCommand(t, db, "new-pending", linkKey, "succeeded", admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteId)
	retryTx, retry := beginConnectorIntegrationTransfer(t, db)
	got, err := retry.TransferInboundRemoteIdentity(t.Context(), transfer)
	if err != nil || got != transfer.RemoteId {
		t.Fatalf("transfer after Runtime completion = %q, %v", got, err)
	}
	_ = retryTx.Rollback()
}

func connectorIntegrationBridgeFixture(t *testing.T, db *sql.DB, suffix string) (context.Context, ConnectorRemoteIdentityTransfer, string) {
	t.Helper()
	ctx := metadata.WithRuntimeMetadataView(t.Context(), runtimeOutboundIntentTestView(t))
	contract := defaultPushConnectorContract()
	source := connectorPushDurableIdentityFromContract(contract, "bridge-source-"+suffix)
	target := source
	target.SourceRecordID = "bridge-target-" + suffix
	scope, err := resolveConnectorIntegrationOutboundScope(ctx, connectorInboundRegisterInstalledBinding{
		Identity: source, FlowDeveloperName: contract.FlowDeveloperName, FlowActionName: contract.FlowActionName,
	})
	if err != nil {
		t.Fatal(err)
	}
	remoteID := "bridge-remote-" + suffix
	adoption := connectorAuthorityLockOrderAdoption(source, remoteID, contract.RemoteIDField)
	adoption.Snapshot.ConnectorNamespace = source.ConnectorOwnerKey
	adoption.Snapshot.FlowDeveloperName = contract.FlowDeveloperName
	adoption.Snapshot.FlowActionName = contract.FlowActionName
	adoption.Snapshot.ProfileContractHash = ConnectorPushProfileContractHash(adoption.Snapshot)
	if _, err := NewSQLConnectorPushDurableStore(db).AdoptFinalizedLink(ctx, adoption); err != nil {
		t.Fatal(err)
	}
	transfer := ConnectorRemoteIdentityTransfer{
		SourceObject: source.SourceObject, SourceRecordId: source.SourceRecordID,
		TargetObject: target.SourceObject, TargetRecordId: target.SourceRecordID,
		RemoteIDField: contract.RemoteIDField, RemoteId: remoteID,
		admissions: []connectorRemoteIdentityTransferAdmission{{
			ConnectorNamespace: source.ConnectorOwnerKey, RemoteID: remoteID,
			SourceIdentity: source, TargetIdentity: target,
			SourceIntegrationScope: scope, TargetIntegrationScope: scope,
			TargetSnapshot: connectorTransferTargetSnapshot(target, contract.RemoteIDField, source.ConnectorOwnerKey),
		}},
	}
	return ctx, transfer, adoption.LinkKey
}

func performConnectorBridgeOwnerMutation(ctx context.Context, store *SQLConnectorPushDurableStore, transfer ConnectorRemoteIdentityTransfer, mutation string) error {
	if mutation == "delete" {
		return store.QuarantineFinalizedRemoteIdentitiesForDeletedOwnerInAmbientTx(ctx, transfer.SourceObject, transfer.SourceRecordId)
	}
	_, err := store.TransferInboundRemoteIdentity(ctx, transfer)
	return err
}

// A finalized V1 is already remote authority, but does not prove that a scoped
// Runtime intent has settled. Both owner mutations must preserve authority
// until every admitted request has a terminal Runtime receipt.
func TestConnectorIntegrationV1BridgeOwnerMutationFenceOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
	for _, mutation := range []string{"transfer", "delete"} {
		for _, state := range []string{"absent", "rejected", "succeeded", "no-effect", "rejected-delete", "no-effect-delete", "pending", "accepted", "wrong remote", "wrong scope", "foreign scoped key", "foreign terminal key", "no catalog"} {
			t.Run(mutation+"/"+state, func(t *testing.T) {
				ctx, transfer, legacyKey := connectorIntegrationBridgeFixture(t, db, mutation+"-"+state)
				admission := transfer.admissions[0]
				if state != "absent" {
					commandState, remoteID, scope := state, transfer.RemoteId, admission.SourceIntegrationScope
					commandKey := connectorIntegrationOutboundLinkKey(admission.SourceIntegrationScope, admission.SourceIdentity)
					switch state {
					case "wrong remote":
						commandState, remoteID = "succeeded", "other-remote"
					case "wrong scope":
						commandState, scope.orgID = "succeeded", "other-org"
					case "foreign scoped key":
						commandState, scope.authorizedWorkloadID = "accepted", "previous-workload"
						commandKey = connectorIntegrationOutboundLinkKey(scope, admission.SourceIdentity)
					case "foreign terminal key":
						commandState, scope.appInstallationID = "succeeded", "previous-installation"
						commandKey = connectorIntegrationOutboundLinkKey(scope, admission.SourceIdentity)
					case "no catalog":
						commandState = "pending"
						ctx = t.Context()
						transfer.admissions[0].SourceIntegrationScope = connectorIntegrationOutboundScope{}
					}
					seedConnectorIntegrationTransferCommand(t, db, mutation+"-"+state, commandKey, commandState,
						admission.SourceIdentity, scope, remoteID)
				}
				tx, store := beginConnectorIntegrationTransfer(t, db)
				err := performConnectorBridgeOwnerMutation(ctx, store, transfer, mutation)
				allowed := state == "absent" || state == "rejected" || state == "succeeded" || state == "no-effect" || state == "rejected-delete" || state == "no-effect-delete"
				if allowed && err != nil {
					t.Fatalf("quiescent V1 owner mutation: %v", err)
				}
				if !allowed && !errors.Is(err, ErrConnectorPushDurableConflict) {
					t.Fatalf("unsafe V1 owner mutation returned %v; want conflict", err)
				}
				var contention *apperrors.LockContentionError
				wantRetry := state == "pending" || state == "accepted" || state == "foreign scoped key" || state == "no catalog"
				if errors.As(err, &contention) != wantRetry {
					t.Fatalf("retry classification = %v, want retriable=%v", err, wantRetry)
				}
				_ = tx.Rollback()
				var status string
				if err := db.QueryRowContext(t.Context(), `SELECT Status FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, legacyKey).Scan(&status); err != nil || status != ConnectorPushLinkStatusFinalized {
					t.Fatalf("rollback lost source authority: %q, %v", status, err)
				}
			})
		}
	}
}

// Enqueue and owner mutation serialize before a RemoteLink is examined. A
// committed pending request blocks both transfer and deletion, even when the
// mutation's original transaction snapshot predates the enqueue.
func TestConnectorIntegrationV1BridgeFreshDispatchFenceOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
	for _, mutation := range []string{"transfer", "delete"} {
		t.Run(mutation, func(t *testing.T) {
			ctx, transfer, legacyKey := connectorIntegrationBridgeFixture(t, db, "race-"+mutation)
			ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
			defer cancel()
			admission := transfer.admissions[0]
			intent := connectorRuntimeTransferIntent(t, "dispatch-"+mutation, admission.SourceIdentity)
			scope := persistence.RuntimeOutboundIntentScope{OrgID: admission.SourceIntegrationScope.orgID, AppInstallationID: admission.SourceIntegrationScope.appInstallationID, AuthorizedWorkloadID: admission.SourceIntegrationScope.authorizedWorkloadID}
			tx, store := beginConnectorIntegrationTransfer(t, db)
			var count int
			if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_RuntimeOutboundIntent WHERE Id = ?`, intent.IntentID).Scan(&count); err != nil || count != 0 {
				t.Fatalf("pre-enqueue snapshot: %d, %v", count, err)
			}
			transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
			repository := persistence.NewRuntimeOutboundIntentRepository(db, transactions)
			enqueued, release := make(chan struct{}), make(chan struct{})
			enqueueDone := make(chan error, 1)
			go func() {
				enqueueDone <- transactions.WithPessimisticTransaction(ctx, func(ctx context.Context) error {
					if _, _, err := repository.Enqueue(ctx, scope, intent); err != nil {
						return err
					}
					close(enqueued)
					select {
					case <-release:
						return nil
					case <-ctx.Done():
						return ctx.Err()
					}
				})
			}()
			colocationWaitForSignal(t, ctx, enqueued, "enqueue did not acquire source owner fence")
			done := make(chan error, 1)
			go func() { done <- performConnectorBridgeOwnerMutation(ctx, store, transfer, mutation) }()
			colocationAssertStillBlocked(t, done, "owner mutation escaped an uncommitted enqueue")
			close(release)
			if err := <-enqueueDone; err != nil {
				t.Fatal(err)
			}
			select {
			case err := <-done:
				var contention *apperrors.LockContentionError
				if !errors.As(err, &contention) {
					t.Fatalf("enqueue-wins owner mutation = %v; want retriable conflict", err)
				}
			case <-ctx.Done():
				t.Fatal("owner mutation did not observe committed pending intent")
			}
			_ = tx.Rollback()
			var status string
			if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, legacyKey).Scan(&status); err != nil || status != ConnectorPushLinkStatusFinalized {
				t.Fatalf("blocked mutation changed source authority: %q, %v", status, err)
			}
		})
	}
}

// Owner deletion that commits first closes enqueue through its existing fence.
// The actual repository producer must wake, reject, and leave no durable intent.
func TestConnectorIntegrationV1BridgeOwnerMutationWinsOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
	t.Run("delete", func(t *testing.T) {
		const mutation = "delete"
		ctx, transfer, legacyKey := connectorIntegrationBridgeFixture(t, db, "owner-wins-"+mutation)
		ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
		defer cancel()
		admission := transfer.admissions[0]
		intent := connectorRuntimeTransferIntent(t, "later-"+mutation, admission.SourceIdentity)
		tx, store := beginConnectorIntegrationTransfer(t, db)
		if err := performConnectorBridgeOwnerMutation(ctx, store, transfer, mutation); err != nil {
			t.Fatal(err)
		}
		done := make(chan error, 1)
		go func() {
			done <- writeConnectorRuntimeTransferReceipt(ctx, db, intent, admission.SourceIntegrationScope, "pending", transfer.RemoteId)
		}()
		colocationAssertStillBlocked(t, done, "enqueue escaped the uncommitted owner mutation")
		if err := tx.Commit(); err != nil {
			t.Fatal(err)
		}
		select {
		case err := <-done:
			if !errors.Is(err, persistence.ErrRuntimeOutboundOwnerClosed) {
				t.Fatalf("old-source enqueue = %v; want closed owner", err)
			}
		case <-ctx.Done():
			t.Fatal("enqueue did not finish after owner mutation commit")
		}
		var intents int
		if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_RuntimeOutboundIntent WHERE IntentId = ?`, intent.IntentID).Scan(&intents); err != nil || intents != 0 {
			t.Fatalf("old-source intent count = %d, %v; want zero", intents, err)
		}
		var status string
		if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, legacyKey).Scan(&status); err != nil {
			t.Fatal(err)
		}
		want := ConnectorPushLinkStatusOwnerDeletedPendingRecovery
		if status != want {
			t.Fatalf("committed source authority = %q, want %q", status, want)
		}
	})
}

func TestConnectorIntegrationOwnerDeleteBeforeFirstLinkRequiresSettledIntentOnTiDB(t *testing.T) {
	for _, state := range []string{"pending", "accepted"} {
		t.Run(state, func(t *testing.T) {
			db, cleanup := openTiDB(t)
			defer cleanup()
			createConnectorPushDurableCatalogTables(t, db)
			transfer := connectorIntegrationTransferFixture("first-link-" + state)
			admission := transfer.admissions[0]
			intent := connectorRuntimeTransferIntent(t, "first-link-"+state, admission.SourceIdentity)
			if err := writeConnectorRuntimeTransferReceipt(t.Context(), db, intent, admission.SourceIntegrationScope, state, transfer.RemoteId); err != nil {
				t.Fatal(err)
			}
			var links int
			if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorRemoteLink`).Scan(&links); err != nil || links != 0 {
				t.Fatalf("first-link fixture has links=%d err=%v", links, err)
			}
			tx, store := beginConnectorIntegrationTransfer(t, db)
			err := store.QuarantineFinalizedRemoteIdentitiesForDeletedOwnerInAmbientTx(t.Context(), transfer.SourceObject, transfer.SourceRecordId)
			var contention *apperrors.LockContentionError
			if !errors.Is(err, ErrConnectorPushDurableConflict) || !errors.As(err, &contention) || !contention.RetriableCondition() {
				t.Fatalf("delete before first link returned %v; want retryable conflict", err)
			}
			if err := tx.Rollback(); err != nil {
				t.Fatal(err)
			}
			ownerKey, err := persistence.ConnectorRemoteIdentityOwnerFenceHash(transfer.SourceObject, transfer.SourceRecordId)
			if err != nil {
				t.Fatal(err)
			}
			var deleted bool
			if err := db.QueryRow(`SELECT OwnerDeleted FROM _System_ConnectorRemoteIdentityLane WHERE RemoteLocatorHash = ?`, ownerKey).Scan(&deleted); err != nil || deleted {
				t.Fatalf("blocked deletion changed owner fence: deleted=%t err=%v", deleted, err)
			}
			var admissionState string
			if err := db.QueryRow(`SELECT AdmissionState FROM _System_RuntimeOutboundIntent WHERE IntentId = ?`, intent.IntentID).Scan(&admissionState); err != nil || admissionState != strings.ToUpper(state) {
				t.Fatalf("blocked deletion changed intent: state=%q err=%v", admissionState, err)
			}
		})
	}
}

func TestConnectorIntegrationOwnerDeleteRequiresV2ProofOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db)
	for _, state := range []string{"succeeded", "rejected", "no-effect", "rejected-delete", "no-effect-delete", "accepted"} {
		t.Run(state, func(t *testing.T) {
			ctx, transfer, legacyKey := connectorIntegrationBridgeFixture(t, db, "delete-v2-"+state)
			admission := transfer.admissions[0]
			// Materialize the independently scoped representation with the same
			// retained snapshot, then restore a canonical V1 alongside it.
			v2Key := connectorIntegrationOutboundLinkKey(admission.SourceIntegrationScope, admission.SourceIdentity)
			locator := connectorIntegrationOutboundRemoteLocatorHash(admission.SourceIntegrationScope, admission.SourceIdentity, transfer.RemoteId)
			if _, err := db.ExecContext(ctx, `UPDATE _System_ConnectorRemoteLink SET LinkKey = ?, RemoteLocatorHash = ?, RemoteLocatorActiveKey = ? WHERE LinkKey = ?`, v2Key, locator, locator, legacyKey); err != nil {
				t.Fatal(err)
			}
			adoption := connectorAuthorityLockOrderAdoption(admission.SourceIdentity, transfer.RemoteId, transfer.RemoteIDField)
			contract := defaultPushConnectorContract()
			adoption.Snapshot.ConnectorNamespace = admission.SourceIdentity.ConnectorOwnerKey
			adoption.Snapshot.FlowDeveloperName, adoption.Snapshot.FlowActionName = contract.FlowDeveloperName, contract.FlowActionName
			adoption.Snapshot.ProfileContractHash = ConnectorPushProfileContractHash(adoption.Snapshot)
			if _, err := NewSQLConnectorPushDurableStore(db).AdoptFinalizedLink(ctx, adoption); err != nil {
				t.Fatal(err)
			}
			seedConnectorIntegrationTransferCommand(t, db, "delete-v2-"+state, v2Key, state, admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteId)
			if state == "succeeded" {
				for _, settledDelete := range []string{"rejected-delete", "no-effect-delete"} {
					seedConnectorIntegrationTransferCommand(t, db, "delete-v2-success-"+settledDelete, v2Key, settledDelete,
						admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteId)
				}
			}
			tx, store := beginConnectorIntegrationTransfer(t, db)
			err := store.QuarantineFinalizedRemoteIdentitiesForDeletedOwnerInAmbientTx(ctx, transfer.SourceObject, transfer.SourceRecordId)
			if state != "succeeded" {
				if !errors.Is(err, ErrConnectorPushDurableConflict) {
					t.Fatalf("V2 without succeeded proof was accepted: %v", err)
				}
				_ = tx.Rollback()
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			if err := tx.Commit(); err != nil {
				t.Fatal(err)
			}
			var quarantined int
			if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE LinkKey IN (?, ?) AND Status = ? AND RemoteLocatorActiveKey IS NOT NULL`,
				legacyKey, v2Key, ConnectorPushLinkStatusOwnerDeletedPendingRecovery).Scan(&quarantined); err != nil || quarantined != 2 {
				t.Fatalf("quarantined physical authorities = %d, %v; want both", quarantined, err)
			}
		})
	}
}

// TestConnectorPushDurablePointReadOutcomesOnTiDB protects the externally
// meaningful lineage rules without pinning SQL text: exact-slot source reads,
// cross-slot destination reads, and finalized-only identity reads.
func TestConnectorPushDurablePointReadOutcomesOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db)

	ctx := t.Context()
	store := NewSQLConnectorPushDurableStore(db)
	previous := connectorPushTestIdentity()
	previous.ActionSlotID = "action-slot-v1"
	previousLinkKey := mustConnectorPushTestLinkKey(previous)
	if _, err := store.AdoptFinalizedLink(ctx, ConnectorPushLinkAdoption{
		LinkKey:  previousLinkKey,
		RemoteID: "remote-v1",
		Identity: previous,
		Snapshot: connectorPushTestSnapshot(),
	}); err != nil {
		t.Fatalf("adopt previous-slot finalized link: %v", err)
	}

	current := previous
	current.ActionSlotID = "action-slot-v2"
	currentLinkKey := mustConnectorPushTestLinkKey(current)
	currentSnapshot := connectorPushTestSnapshot()
	currentSnapshot.RequestIdempotencyKey = "point-read-current"
	if _, err := seedLegacyConnectorAttempt(ctx, db, legacyConnectorAttemptFixture{
		AttemptID:      "attempt-current",
		LinkKey:        currentLinkKey,
		AttemptOrdinal: 1,
		Identity:       current,
		Snapshot:       currentSnapshot,
	}); err != nil {
		t.Fatalf("reserve current-slot create attempt: %v", err)
	}

	finalized, err := store.FindFinalizedLink(ctx, previous)
	if err != nil {
		t.Fatalf("FindFinalizedLink(previous): %v", err)
	}
	if finalized == nil || finalized.LinkKey != previousLinkKey || finalized.RemoteID != "remote-v1" || finalized.Status != ConnectorPushLinkStatusFinalized {
		t.Fatalf("FindFinalizedLink(previous) = %#v, want finalized remote-v1", finalized)
	}
	if finalized, err := store.FindFinalizedLink(ctx, current); err != nil || finalized != nil {
		t.Fatalf("FindFinalizedLink(current) = %#v, %v; want nil, nil", finalized, err)
	}

	sourceLink, sourceAttempt, err := store.FindStateBySource(ctx, ConnectorPushDurableSourceQuery{
		ConnectorOwnerKey: current.ConnectorOwnerKey,
		ActionSlotID:      current.ActionSlotID,
		ActionType:        current.ActionType,
		SourceObject:      current.SourceObject,
		SourceRecordID:    current.SourceRecordID,
	})
	if err != nil {
		t.Fatalf("FindStateBySource(current): %v", err)
	}
	if sourceLink != nil || sourceAttempt == nil || sourceAttempt.ID != "attempt-current" || sourceAttempt.LinkKey != currentLinkKey {
		t.Fatalf("source state = link %#v, attempt %#v; want current attempt only", sourceLink, sourceAttempt)
	}

	destinationLink, destinationAttempt, err := store.FindStateByDestination(ctx, ConnectorPushDurableDestinationQuery{
		ConnectorOwnerKey: current.ConnectorOwnerKey,
		ActionType:        current.ActionType,
		SourceObject:      current.SourceObject,
		SourceRecordID:    current.SourceRecordID,
		DestinationSlotID: current.DestinationSlotID,
	})
	if err != nil {
		t.Fatalf("FindStateByDestination(current): %v", err)
	}
	if destinationLink == nil || destinationLink.LinkKey != previousLinkKey || destinationLink.Identity.ActionSlotID != previous.ActionSlotID {
		t.Fatalf("destination link = %#v, want previous-slot lineage", destinationLink)
	}
	if destinationAttempt == nil || destinationAttempt.ID != "attempt-current" || destinationAttempt.LinkKey != currentLinkKey {
		t.Fatalf("destination attempt = %#v, want current-slot attempt", destinationAttempt)
	}
}

// TestConnectorPushDurableSnapshotDecodingOnTiDB writes corrupt-but-storable
// persisted values through TiDB, then proves the durable reader fails closed.
// Values rejected by the catalog's JSON columns are intentionally absent: the
// real schema already makes those states impossible.
func TestConnectorPushDurableSnapshotDecodingOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db)

	cases := []struct {
		name       string
		assignment string
		args       []any
		want       string
	}{
		{name: "dispatch lane", assignment: "DispatchLane = ?", args: []any{"legacy-unbounded"}, want: "dispatch lane"},
		{name: "allowed source fields", assignment: "AllowedSourceFieldsJSON = ?", args: []any{`{}`}, want: "allowed source fields"},
		{name: "association source ids", assignment: "AssociationSourceRecordIdsJSON = ?", args: []any{`{}`}, want: "association source record ids"},
		{name: "retry policy", assignment: "RetryPolicy = ?", args: []any{`[`}, want: "retry policy"},
		{name: "profile hash", assignment: "ProfileContractHash = ?", args: []any{"bad-profile-hash"}, want: "profile contract hash mismatch"},
		{name: "conflict patch intent", assignment: "CreateConflictPatchIntentJSON = ?", args: []any{`[]`}, want: "create conflict patch intent"},
		{name: "patch dispatch intent", assignment: "PatchDispatchIntentJSON = ?", args: []any{`[]`}, want: "PATCH dispatch intent"},
		{name: "originating session", assignment: "OriginatingUserSessionJSON = ?", args: []any{`[]`}, want: "originating user session"},
		{name: "request headers", assignment: "RequestHeaders = ?", args: []any{`[]`}, want: "request headers"},
		{name: "request body encoding", assignment: "RequestBodyBase64 = ?", args: []any{"%%%"}, want: "request body"},
		{name: "request body hash", assignment: "RequestBodyHash = ?", args: []any{"bad-request-hash"}, want: "request body hash mismatch"},
		{name: "response body hash", assignment: "ResponseBodyBase64 = ?, ResponseBodyHash = ?", args: []any{"eyJlcnJvciI6dHJ1ZX0=", "bad-response-hash"}, want: "response body hash mismatch"},
	}

	for index, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			identity := connectorPushTestIdentity()
			identity.SourceRecordID = fmt.Sprintf("003-decode-%02d", index)
			linkKey := mustConnectorPushTestLinkKey(identity)
			snapshot := connectorPushTestSnapshot()
			snapshot.RequestIdempotencyKey = fmt.Sprintf("decode-%02d", index)
			attemptID := fmt.Sprintf("attempt-decode-%02d", index)
			if _, err := seedLegacyConnectorAttempt(t.Context(), db, legacyConnectorAttemptFixture{
				AttemptID:      attemptID,
				LinkKey:        linkKey,
				AttemptOrdinal: 1,
				Identity:       identity,
				Snapshot:       snapshot,
			}); err != nil {
				t.Fatalf("reserve fixture attempt: %v", err)
			}

			args := append(append([]any(nil), tc.args...), attemptID)
			if _, err := db.ExecContext(t.Context(), "UPDATE _System_ConnectorPushAttempt SET "+tc.assignment+" WHERE Id = ?", args...); err != nil {
				t.Fatalf("persist corrupt fixture: %v", err)
			}
			_, err := NewSQLConnectorPushDurableStore(db).FindBlockingAttempt(t.Context(), linkKey)
			if err == nil || !strings.Contains(err.Error(), tc.want) {
				t.Fatalf("FindBlockingAttempt error = %v, want %q", err, tc.want)
			}
		})
	}
}