package automation

// T4 co-location artifact for issue #3102 §6 (case 3, item 3:
// "remote-id-provenance Fragment-A T4 atomicity").
//
// The L0/L1 standard (docs/design/l0-l1-standard.md, T4 at :233-248) admits a
// single-consumer mechanism into L0 only when a *checkable artifact* proves the
// co-location is load-bearing: "a test that fails when the logic is moved
// outside the named boundary" (:242-245). A prose claim is unverified and routes
// to §6. #3102 §6 left the connector remote-id PROVENANCE consume — the
// in-transaction consume of inbound authorship evidence inside AdoptFinalizedLink
// — classified default-L1 *pending exactly this artifact*.
//
// NAMED BOUNDARY UNDER TEST: production AdoptFinalizedLink
// (connector_push_durable_store.go) opens ONE BeginTx that spans BOTH the
// finalized remote-link INSERT AND consumeEvidenceTx(true) (which FOR UPDATEs the
// authored evidence row and flips it consumed) BEFORE the single tx.Commit. The
// claim: the link INSERT and the evidence consume MUST commit together.
//
// THE T4 GUARD (TestConnectorPushDurableProvenanceConsumeGuardsProductionPath):
// runs the REAL AdoptFinalizedLink and drives it through the BETWEEN-COMMITS
// window via a narrow production test seam (SQLConnectorPushDurableStore's
// adoptAfterInsertTestHook, nil in production — the same *ForTest seam idiom as
// scheduler_service.go). The seam retires the authored evidence AFTER the
// finalized-link INSERT and BEFORE the in-transaction consume. In the production
// single commit the consume then reads stale evidence, fails closed, and the
// deferred rollback erases the INSERT, so NO finalized link survives. If
// AdoptFinalizedLink is refactored to commit the link INSERT before consuming the
// evidence — i.e. the consume is moved OUTSIDE the named boundary — the committed
// link survives the failed consume as an orphan and this assertion goes RED.
// Verified by mutation: splitting the production consume into a second
// transaction turns this guard red. Because the staleness is injected INSIDE the
// window (not before the call), the guard also catches a "split + early
// stale-evidence precheck" refactor, which a stale-before-call check would miss.
//
// WHY THIS IS NOT REDUNDANT with the pre-existing atomic test
// (connector_push_durable_store_test.go, TestSQLConnectorAdoptFinalizedLinkConsumesEvidenceAtomically):
// that test's stale-adopt fixture reuses the already-finalized link's
// (ConnectorOwnerKey, DestinationSlotId, RemoteId), so the second INSERT collides
// on uniq_connector_remote_locator_active and conflicts BEFORE it could orphan.
// It therefore stays green even when production splits the consume out — it does
// not actually guard this boundary. This guard uses isolated fixtures (distinct
// RemoteId per case) so the assertion exercises the consume/commit boundary, not
// the remote-locator unique index. Verified by mutation: under a split-consume
// production, this guard goes red while that pre-existing test stays green.
//
// THE NUANCE (why the bad outcome is stale-evidence-adopt, NOT double-adopt):
// the evidence ActiveKey = hash(ConnectorOwnerKey, SourceObject, SourceRecordId,
// RemoteIDField) (ConnectorInboundEvidenceActiveKey) is coarser than the LinkKey.
// Two distinct LinkKeys can share one ActiveKey only with the SAME
// DestinationSlotId (distinct destinations are required to use distinct remote-id
// fields, connector_push_durable_types.go :69-74), which makes their
// RemoteLocatorActiveKey collide — so uniq_connector_remote_locator_active, NOT
// the consume atomicity, blocks the second finalize. The consume atomicity guards
// the per-link provenance binding: a finalized link exists IFF the authored
// evidence it claims was consumed FOR IT, atomically. Splitting the consume out
// makes an ORPHAN reachable: a finalized link whose authored-evidence binding
// never completed.
//
// TestConnectorPushDurableProvenanceConsumeSplitVariantReachability is supporting
// reachability evidence only: it exhibits the concrete orphan against an isolated
// test-local fork of AdoptFinalizedLink, WITHOUT mutating production. The
// production-path guard above is the actual T4 gate.
//
// Both tests require a live TiDB (openTiDB; default 127.0.0.1:4001, override
// TIDB_PORT). They are skipped when TiDB is unreachable; the go-ci.yml
// l0-neutrality-gate job provisions TiDB and runs them so the artifact is
// automation-enforced, not skip-only.

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"testing"
)

const (
	provGuardOwner = "owner-pack"
	provGuardNS    = "connector_ns"
	provGuardObj   = "Contact"
	provGuardField = "HubSpot_Id__c"
)

// provGuardHarness wires the three tables, the store, and the shared fixtures.
type provGuardHarness struct {
	db       *sql.DB
	store    *SQLConnectorPushDurableStore
	snapshot ConnectorPushDurableSnapshot
}

func newProvGuardHarness(t *testing.T) (*provGuardHarness, func()) {
	t.Helper()
	db, cleanup := openTiDB(t)
	ctx := context.Background()
	createConnectorRuntimeOwnerCatalogTables(t, db)
	for _, ddl := range []string{connectorRemoteLinkTestDDL, connectorPushAttemptTestDDL, connectorInboundEvidenceTestDDL} {
		if _, err := db.ExecContext(ctx, ddl); err != nil {
			cleanup()
			t.Fatalf("create table: %v", err)
		}
	}
	snapshot := connectorPushTestSnapshot()
	snapshot.ConnectorNamespace = provGuardNS
	return &provGuardHarness{db: db, store: NewSQLConnectorPushDurableStore(db), snapshot: snapshot}, cleanup
}

func (h *provGuardHarness) identity(rec, slot, dest string) ConnectorPushDurableIdentity {
	return ConnectorPushDurableIdentity{
		ConnectorOwnerKey: provGuardOwner, ActionSlotID: slot, ActionType: "PushToConnector",
		SourceObject: provGuardObj, SourceRecordID: rec, DestinationSlotID: dest,
	}
}

func (h *provGuardHarness) adoption(id ConnectorPushDurableIdentity, activeKey, remoteID string) ConnectorPushLinkAdoption {
	return ConnectorPushLinkAdoption{RemoteID: remoteID, Identity: id, Snapshot: h.snapshot, EvidenceActiveKey: activeKey}
}

func (h *provGuardHarness) linkKey(t *testing.T, id ConnectorPushDurableIdentity) string {
	t.Helper()
	k, err := id.LinkKey()
	if err != nil {
		t.Fatalf("LinkKey: %v", err)
	}
	return k
}

func (h *provGuardHarness) seedAuthoredEvidence(t *testing.T, id, rec, remoteID string) {
	t.Helper()
	if err := h.store.InsertAuthoredEvidence(context.Background(), ConnectorInboundRemoteIDEvidence{
		ID: id, ConnectorOwnerKey: provGuardOwner, ConnectorNamespace: provGuardNS,
		SourceObject: provGuardObj, SourceRecordId: rec, RemoteIDField: provGuardField, RemoteId: remoteID,
	}); err != nil {
		t.Fatalf("seed authored evidence %s: %v", id, err)
	}
}

func (h *provGuardHarness) finalizedLinkExists(t *testing.T, linkKey string) bool {
	t.Helper()
	var n int
	if err := h.db.QueryRowContext(context.Background(),
		`SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE LinkKey = ? AND Status = ?`,
		linkKey, ConnectorPushLinkStatusFinalized).Scan(&n); err != nil {
		t.Fatalf("count finalized link %s: %v", linkKey, err)
	}
	return n > 0
}

func (h *provGuardHarness) readEvidence(t *testing.T, evID string) (status, consumedBy, activeKey string) {
	t.Helper()
	if err := h.db.QueryRowContext(context.Background(),
		`SELECT EvidenceStatus, COALESCE(ConsumedByLinkKey,''), COALESCE(ActiveKey,'') FROM _System_ConnectorInboundRemoteIDEvidence WHERE Id = ?`,
		evID).Scan(&status, &consumedBy, &activeKey); err != nil {
		t.Fatalf("read evidence %s: %v", evID, err)
	}
	return status, consumedBy, activeKey
}

// TestConnectorPushDurableProvenanceConsumeGuardsProductionPath is the #3102 §6
// case-3 item-3 T4 gate. It exercises the REAL AdoptFinalizedLink and goes red if
// the evidence consume is moved outside the link-INSERT transaction.
func TestConnectorPushDurableProvenanceConsumeGuardsProductionPath(t *testing.T) {
	h, cleanup := newProvGuardHarness(t)
	defer cleanup()
	ctx := context.Background()

	// (a) Success: a real adopt consumes the evidence atomically with the INSERT.
	{
		const rec, evID, remoteID = "003-prov-ok", "ev-prov-ok", "remote-prov-ok"
		id := h.identity(rec, "slot-A", "dest-A")
		linkKey := h.linkKey(t, id)
		activeKey := ConnectorInboundEvidenceActiveKey(provGuardOwner, provGuardObj, rec, provGuardField)
		h.seedAuthoredEvidence(t, evID, rec, remoteID)

		link, err := h.store.AdoptFinalizedLink(ctx, h.adoption(id, activeKey, remoteID))
		if err != nil {
			t.Fatalf("adopt: %v", err)
		}
		if link == nil || link.Status != ConnectorPushLinkStatusFinalized || link.RemoteID != remoteID {
			t.Fatalf("adopted link = %#v, want finalized/%s", link, remoteID)
		}
		if status, consumedBy, activeKeyAfter := h.readEvidence(t, evID); status != ConnectorInboundEvidenceStatusConsumed || consumedBy != linkKey || activeKeyAfter != "" {
			t.Fatalf("after adopt: evidence status=%q consumedBy=%q activeKey=%q, want consumed/%s/empty", status, consumedBy, activeKeyAfter, linkKey)
		}

		// A retry sees a consistent state: the link is returned idempotently and
		// the evidence binding does not change.
		retry, err := h.store.AdoptFinalizedLink(ctx, h.adoption(id, activeKey, remoteID))
		if err != nil {
			t.Fatalf("idempotent retry adopt: %v", err)
		}
		if retry == nil || retry.Status != ConnectorPushLinkStatusFinalized {
			t.Fatalf("retry link = %#v, want finalized", retry)
		}
		if status, consumedBy, _ := h.readEvidence(t, evID); status != ConnectorInboundEvidenceStatusConsumed || consumedBy != linkKey {
			t.Fatalf("after retry: evidence status=%q consumedBy=%q, want consumed/%s (binding must not change)", status, consumedBy, linkKey)
		}
	}

	// (b) THE GUARD: drive the REAL AdoptFinalizedLink through the BETWEEN-COMMITS
	// window using the production test seam. The evidence is authored when adopt
	// starts; the seam retires it AFTER the finalized-link INSERT and BEFORE the
	// in-transaction consume. In the production single commit the consume then reads
	// stale evidence, fails closed, and the deferred rollback erases the INSERT — no
	// orphan. If the consume is moved outside the named boundary (the INSERT
	// committed before the consume), the committed link survives the failed consume
	// as an orphan and this assertion goes RED.
	//
	// Retiring INSIDE the window — not before the call — is what makes this airtight
	// against a "split + early stale-evidence precheck" refactor: such a precheck
	// runs before the INSERT, sees still-authored evidence, and passes; only the
	// post-INSERT consume observes the staleness, so the orphan still surfaces and
	// the guard still goes red. Verified by mutation: splitting the production
	// consume into a second transaction turns this guard red while the pre-existing
	// atomic test stays green.
	//
	// Isolated fixtures (a RemoteId distinct from every other case here) keep the
	// link INSERT off the uniq_connector_remote_locator_active key, so it is the
	// consume/commit boundary — not the remote-locator unique index — that this
	// assertion exercises. This is the gap that lets the pre-existing atomic test
	// stay green under a split-consume production.
	{
		const rec, evID, remoteID = "003-prov-guard", "ev-prov-guard", "remote-prov-guard"
		id := h.identity(rec, "slot-A", "dest-A")
		activeKey := ConnectorInboundEvidenceActiveKey(provGuardOwner, provGuardObj, rec, provGuardField)
		h.seedAuthoredEvidence(t, evID, rec, remoteID)

		var hookRan bool
		var retireErr error
		h.store.adoptAfterInsertTestHook = func() {
			hookRan = true
			// A separate connection retires the evidence in the window. The adopt
			// transaction has not yet FOR UPDATE-locked the evidence row (the consume
			// runs after this hook), so the retire commits without contention.
			retireErr = h.store.RetireEvidence(context.Background(), evID)
		}
		_, err := h.store.AdoptFinalizedLink(ctx, h.adoption(id, activeKey, remoteID))
		h.store.adoptAfterInsertTestHook = nil

		if !hookRan {
			t.Fatalf("between-commits seam never fired — the guard did not exercise the window")
		}
		if retireErr != nil {
			t.Fatalf("retire evidence in the between-commits window: %v", retireErr)
		}
		if !errors.Is(err, ErrConnectorPushDurableConflict) {
			t.Fatalf("adopt with evidence retired in the window: err = %v, want ErrConnectorPushDurableConflict", err)
		}
		if h.finalizedLinkExists(t, h.linkKey(t, id)) {
			t.Fatalf("BOUNDARY VIOLATED: a finalized link survives a failed evidence consume — the INSERT must roll back with the consume in one transaction")
		}
		t.Logf("production guard: evidence retired in the between-commits window -> adopt fails closed AND no orphaned link (consume is co-located with the INSERT commit)")
	}
}

// TestConnectorPushDurableProvenanceConsumeSplitVariantReachability is supporting
// reachability evidence (NOT the T4 gate; the production-path guard above is). It
// reproduces the concrete orphan against an isolated test-local fork of
// AdoptFinalizedLink whose only difference from production is the transaction
// boundary, without mutating production.
func TestConnectorPushDurableProvenanceConsumeSplitVariantReachability(t *testing.T) {
	h, cleanup := newProvGuardHarness(t)
	defer cleanup()
	ctx := context.Background()

	const rec, evID, remoteID = "003-prov-split", "ev-prov-split", "remote-prov-split"
	id := h.identity(rec, "slot-A", "dest-A")
	linkKey := h.linkKey(t, id)
	activeKey := ConnectorInboundEvidenceActiveKey(provGuardOwner, provGuardObj, rec, provGuardField)
	h.seedAuthoredEvidence(t, evID, rec, remoteID)

	// Split-adopt: the link INSERT is committed first; in the window before the
	// (now separate) consume runs, a real RetireEvidence makes the evidence stale.
	var retireErr error
	committed, consumeErr := adoptFinalizedLinkSplitConsume(ctx, h.db, h.adoption(id, activeKey, remoteID), func() {
		retireErr = h.store.RetireEvidence(ctx, evID)
	})

	if !committed {
		t.Fatalf("split variant must commit the link before the window; committed=false, consumeErr=%v", consumeErr)
	}
	if retireErr != nil {
		t.Fatalf("windowed RetireEvidence: %v", retireErr)
	}
	if !errors.Is(consumeErr, ErrConnectorPushDurableConflict) {
		t.Fatalf("split variant consume err = %v, want ErrConnectorPushDurableConflict (evidence retired in the window)", consumeErr)
	}

	// The orphan the production single commit forecloses: a finalized link whose
	// authored-evidence binding never completed.
	if !h.finalizedLinkExists(t, linkKey) {
		t.Fatalf("link must be durably committed (the orphan)")
	}
	if status, consumedBy, _ := h.readEvidence(t, evID); status != ConnectorInboundEvidenceStatusRetired || consumedBy != "" {
		t.Fatalf("evidence status=%q consumedBy=%q, want retired/empty — the orphaned link was never bound to it", status, consumedBy)
	}
	t.Logf("split commit reachability: link %s is finalized but its authored evidence (id=%s) is retired/unconsumed (orphaned provenance)", linkKey, evID)
}

// adoptFinalizedLinkSplitConsume is a TEST-LOCAL fork of the production
// SQLConnectorPushDurableStore.AdoptFinalizedLink. Its ONLY difference from
// production is the transaction boundary: production runs the finalized-link
// INSERT and consumeEvidenceTx inside ONE BeginTx and commits them together; this
// fork COMMITS the INSERT in tx1, then runs the evidence consume in a SEPARATE
// tx2. Every gate, the INSERT column set/values, and the consume SQL are copied
// verbatim from production so the sole observable difference is WHERE the commit
// lands. It backs the reachability test only; the production-path guard is the
// authoritative T4 gate.
//
// inWindow runs in the gap between the two commits (the crash/interleave that the
// production single commit closes). It returns whether the link was committed and
// the error from the (separate-transaction) consume.
func adoptFinalizedLinkSplitConsume(ctx context.Context, db *sql.DB, adoption ConnectorPushLinkAdoption, inWindow func()) (linkCommitted bool, consumeErr error) {
	adoption, err := normalizeConnectorPushLinkAdoption(adoption)
	if err != nil {
		return false, err
	}

	// --- tx1: gates + finalized-link INSERT, committed on its own. ---
	tx1, err := db.BeginTx(ctx, nil)
	if err != nil {
		return false, err
	}
	rollback1 := true
	defer func() {
		if rollback1 {
			_ = tx1.Rollback()
		}
	}()

	// Verbatim from production: attempt-occupancy FOR UPDATE gate.
	blockingStatus := ""
	err = tx1.QueryRowContext(ctx, `
SELECT Status
  FROM _System_ConnectorPushAttempt
 WHERE AttemptOccupancyKey = ?
 LIMIT 1
 FOR UPDATE`, adoption.LinkKey).Scan(&blockingStatus)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return false, err
	}
	if strings.TrimSpace(blockingStatus) != "" {
		return false, fmt.Errorf("%w: create attempt %s already occupies link key", ErrConnectorPushDurableConflict, strings.TrimSpace(blockingStatus))
	}

	// Verbatim from production: existing-link FOR UPDATE gate (the idempotent
	// sub-path is unreachable here — the fork is only driven for a fresh key — so
	// it is omitted to keep the fork to the boundary under test).
	var existingID, existingStatus, existingRemoteID string
	err = tx1.QueryRowContext(ctx, `
SELECT Id, Status, RemoteId
  FROM _System_ConnectorRemoteLink
 WHERE LinkKey = ?
 LIMIT 1
 FOR UPDATE`, adoption.LinkKey).Scan(&existingID, &existingStatus, &existingRemoteID)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return false, err
	}
	if strings.TrimSpace(existingID) != "" {
		return false, fmt.Errorf("%w: remote link %s already owns link key", ErrConnectorPushDurableConflict, strings.TrimSpace(existingStatus))
	}

	// Verbatim from production: the finalized-link INSERT (same columns/values).
	linkID := "splitlink-" + adoption.LinkKey
	remoteLocatorHash := ConnectorPushRemoteLocatorHash(adoption.Identity, adoption.RemoteID)
	_, err = tx1.ExecContext(ctx, `
INSERT INTO _System_ConnectorRemoteLink
  (Id, LinkKey, Status, RemoteId, RemoteLocatorHash, RemoteLocatorActiveKey, 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,
   CompatibilityMirrorStatus)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		linkID, adoption.LinkKey, ConnectorPushLinkStatusFinalized, adoption.RemoteID,
		remoteLocatorHash, remoteLocatorHash,
		adoption.Identity.ConnectorOwnerKey, adoption.Identity.ActionSlotID, adoption.Identity.ActionType,
		adoption.Identity.SourceObject, adoption.Identity.SourceRecordID, adoption.Identity.DestinationSlotID,
		adoption.Snapshot.RequesterOwnerPack, adoption.Snapshot.ProfileOwnerPack, adoption.Snapshot.ConnectorNamespace,
		adoption.Snapshot.FlowDeveloperName, adoption.Snapshot.FlowActionName, adoption.Snapshot.TargetObject,
		adoption.Snapshot.ProfileName, adoption.Snapshot.CreatedPackInstallID, adoption.Snapshot.CreatedPackInstallID,
		adoption.Snapshot.CredentialRef, adoption.Snapshot.EndpointClass, mustConnectorPushJSON(adoption.Snapshot.RetryPolicy),
		adoption.Snapshot.ProfileContractHash, adoption.Snapshot.CreatePathTemplate,
		adoption.Snapshot.UpdatePathTemplate, adoption.Snapshot.ResponseIDPath,
		nullableConnectorPushString(adoption.Snapshot.RemoteIDField),
		nullableConnectorPushString(adoption.Snapshot.RemoteIDProvenanceField),
		adoption.Snapshot.MappingFingerprint, adoption.Snapshot.BodyFingerprint,
		ConnectorPushCompatibilityMirrorSynced,
	)
	if err != nil {
		if conflictErr := connectorPushDuplicateKeyConflict(err); conflictErr != nil {
			return false, conflictErr
		}
		return false, err
	}
	if err := tx1.Commit(); err != nil {
		return false, err
	}
	rollback1 = false
	linkCommitted = true

	// --- WINDOW: the link is durable; the evidence is still active (not consumed). ---
	if inWindow != nil {
		inWindow()
	}

	// --- tx2: evidence consume, SEPARATE transaction (the moved boundary). ---
	// Verbatim from production consumeEvidenceTx(requireMatch=true), only relocated
	// out of tx1.
	tx2, err := db.BeginTx(ctx, nil)
	if err != nil {
		return linkCommitted, err
	}
	defer func() { _ = tx2.Rollback() }()

	key := strings.TrimSpace(adoption.EvidenceActiveKey)
	if key == "" {
		return linkCommitted, tx2.Commit()
	}
	var evID, evStatus, evRemoteID string
	err = tx2.QueryRowContext(ctx, `
SELECT Id, EvidenceStatus, RemoteId
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE ActiveKey = ?
 LIMIT 1
 FOR UPDATE`, key).Scan(&evID, &evStatus, &evRemoteID)
	if errors.Is(err, sql.ErrNoRows) {
		return linkCommitted, fmt.Errorf("%w: inbound authorship evidence is no longer active for adopt", ErrConnectorPushDurableConflict)
	}
	if err != nil {
		return linkCommitted, err
	}
	if strings.TrimSpace(evStatus) != ConnectorInboundEvidenceStatusAuthored || strings.TrimSpace(evRemoteID) != adoption.RemoteID {
		return linkCommitted, fmt.Errorf("%w: inbound authorship evidence changed before adopt", ErrConnectorPushDurableConflict)
	}
	if _, err := tx2.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRemoteIDEvidence
   SET EvidenceStatus = ?, ConsumedByLinkKey = ?, ActiveKey = NULL, LastModifiedDate = CURRENT_TIMESTAMP
 WHERE Id = ? AND EvidenceStatus = ?`,
		ConnectorInboundEvidenceStatusConsumed, adoption.LinkKey, evID, ConnectorInboundEvidenceStatusAuthored); err != nil {
		return linkCommitted, err
	}
	return linkCommitted, tx2.Commit()
}