package controlplane

import (
	"context"
	"database/sql"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	database "github.com/nexuscrm/db"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

func TestBeginApplyAllSourceRevisionAdvanceAttemptUsesFreshExactIncarnation(t *testing.T) {
	tests := []struct {
		name          string
		category      packs.Category
		pending       ApplyAllSourceRevisionAdvancePendingKind
		status        string
		wantQueue     string
		wantClearHash bool
	}{
		{
			name: "app advance", category: packs.AppPack,
			pending: ApplyAllSourceRevisionAdvancePendingAdvance,
			status:  installedPackStatusActive, wantQueue: "e.CompanionPackInstallId",
		},
		{
			name: "connector advance", category: packs.ConnectorPack,
			pending: ApplyAllSourceRevisionAdvancePendingAdvance,
			status:  installedPackStatusActive, wantQueue: "e.InstalledPackId", wantClearHash: true,
		},
		{
			name: "app failed retry", category: packs.AppPack,
			pending: ApplyAllSourceRevisionAdvancePendingRetry,
			status:  installedPackStatusFailed, wantQueue: "e.CompanionPackInstallId",
		},
		{
			name: "connector installing retry", category: packs.ConnectorPack,
			pending: ApplyAllSourceRevisionAdvancePendingRetry,
			status:  installedPackStatusInstalling, wantQueue: "e.InstalledPackId", wantClearHash: true,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			queries := []string{}
			matcher := sqlmock.QueryMatcherFunc(func(expectedSQL, actualSQL string) error {
				if err := sqlmock.QueryMatcherRegexp.Match(expectedSQL, actualSQL); err != nil {
					return err
				}
				queries = append(queries, actualSQL)
				return nil
			})
			db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(matcher))
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			db.SetMaxOpenConns(1)
			admission := testApplyAllSourceRevisionAdvanceBeginAdmission(
				t, db, test.category, test.pending, test.status,
			)
			oldRow := admission.identity.proof.entries[0].current.row
			target := admission.identity.proof.entries[0].installedCompatibility
			atom := admission.identity.proof.entries[0].atom
			applyOptions := packs.ApplyOptions{
				SkipPolymorphicDiscriminatorBackfill: true,
				RepairLegacyUniqueDuplicates:         true,
			}
			receipt := testApplyAllSourceRevisionAdvanceBeginReceipt(
				t, context.Background(), db, mock, admission, 0, applyOptions,
			)

			expectApplyAllSourceRevisionAdvanceExactIncarnationBegin(mock, oldRow, test.category, false)
			if test.category == packs.ConnectorPack {
				mock.ExpectQuery("SELECT ProtocolVersion,LegacyMaxOrdinal").WithArgs(oldRow.PackInstallID).
					WillReturnRows(sqlmock.NewRows([]string{"ProtocolVersion", "LegacyMaxOrdinal"}).AddRow(1, 0))
			}
			mock.ExpectExec("UPDATE _System_InstalledPack p").
				WillReturnResult(sqlmock.NewResult(0, 1))
			expectApplyAllSourceRevisionAdvanceExactIncarnationCommit(mock)

			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			defer cancel()
			attempt, err := beginApplyAllSourceRevisionAdvanceAttempt(
				ctx, receipt,
			)
			if err != nil {
				t.Fatalf("begin exact attempt: %v", err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
			if attempt.identity == nil || !attempt.identity.minted ||
				attempt.identity.admission.identity == nil ||
				attempt.identity.admission.identity.provenance.activeRuntimeDatabase != db ||
				attempt.identity.admission.identity.provenance.activeRuntimeSnapshot != admission.identity.provenance.activeRuntimeSnapshot ||
				attempt.identity.selectedProofIndex != 0 || attempt.identity.pendingOrdinal != 0 ||
				attempt.identity.admission.identity.proof.entries[0].installedCompatibility != target ||
				attempt.identity.admission.identity.proof.entries[0].atom != atom ||
				attempt.identity.applyOptions != applyOptions {
				t.Fatalf("attempt identity = %#v, want exact database, target, and atom", attempt.identity)
			}
			if attempt.identity.selectedAtomClaim == nil ||
				attempt.identity.selectedAtomClaim != attempt.identity.provenance.selectedAtomClaim ||
				attempt.identity.selectedAtomClaim != attempt.witness.selectedAtomClaim ||
				attempt.identity.selectedAtomClaim.claimed.Load() {
				t.Fatal("Begin did not mint one unclaimed selected-atom identity")
			}
			atomIdentity := attempt.identity.provenance.admissionProvenance.entries[0].atomIdentity
			if atomIdentity != admission.identity.provenance.entries[0].atomIdentity {
				t.Fatal("Begin reminted the proof-owned execution atom identity")
			}
			if err := packs.ValidateSourceBackedExecutionAtomIdentityTarget(
				atomIdentity, atom, target,
			); err != nil {
				t.Fatalf("attempt atom identity: %v", err)
			}
			installing := attempt.identity.installingRow
			if installing.ID != oldRow.ID || installing.Name != oldRow.Name ||
				installing.PackInstallID == "" || installing.PackInstallID == oldRow.PackInstallID ||
				installing.Status != installedPackStatusInstalling || installing.IsDeleted != 0 ||
				installing.SourceRevision.String != admission.identity.proof.entries[0].current.advance.ToRevision {
				t.Fatalf("installing row = %#v, want same row with a fresh target incarnation", installing)
			}
			if test.wantClearHash {
				if installing.AppliedPayloadHash.Valid {
					t.Fatalf("connector installing row retained applied hash: %#v", installing.AppliedPayloadHash)
				}
			} else if installing.AppliedPayloadHash != oldRow.AppliedPayloadHash {
				t.Fatalf("app installing row changed applied hash: got %#v want %#v", installing.AppliedPayloadHash, oldRow.AppliedPayloadHash)
			}

			update := testApplyAllSourceRevisionAdvanceBeginUpdateQuery(t, queries)
			for _, want := range []string{
				"PackInstallId = ?", "Status = ?", "DesiredPayloadHash = ?",
				"SourceRevision = ?", "HostRequirementsDigest = ?",
				"RuntimeObjectSpecDeclarations = ?", "RuntimeFormulaFunctionBundles = ?",
				"RuntimeStartupShallowDeclarations = ?", "LastError = NULL", "IsDeleted = 0",
				applyAllSourceRevisionAdvanceExactStringPredicate("p.Id"),
				applyAllSourceRevisionAdvanceExactStringPredicate("p.PackInstallId"),
				testApplyAllSourceRevisionAdvanceExactQueuePackMatch(test.category),
				"FROM _System_ConnectorWorkItem w",
				"ProjectionKey",
			} {
				if !strings.Contains(update, want) {
					t.Fatalf("exact begin update missing %q:\n%s", want, update)
				}
			}
			if strings.Contains(update, "INSERT INTO") {
				t.Fatalf("exact begin update contains insert fallback:\n%s", update)
			}
			clearsHash := strings.Contains(update, "AppliedPayloadHash = NULL")
			clearsAppliedDate := strings.Contains(update, "LastAppliedDate = NULL")
			if clearsHash != test.wantClearHash || clearsAppliedDate != test.wantClearHash {
				t.Fatalf("connector-only resets = hash:%t date:%t, want %t:\n%s", clearsHash, clearsAppliedDate, test.wantClearHash, update)
			}
			lock := testApplyAllSourceRevisionAdvanceBeginLockQuery(t, queries)
			for _, want := range []string{
				applyAllSourceRevisionAdvanceExactStringPredicate("p.Id"),
				applyAllSourceRevisionAdvanceExactStringPredicate("p.PackInstallId"),
				"FOR UPDATE",
			} {
				if !strings.Contains(lock, want) {
					t.Fatalf("exact begin row lock missing %q:\n%s", want, lock)
				}
			}
			queue := testApplyAllSourceRevisionAdvanceBeginQueueQuery(t, queries)
			if !strings.Contains(queue, testApplyAllSourceRevisionAdvanceExactQueuePackMatch(test.category)) ||
				!strings.Contains(queue, "ProjectionKey") {
				t.Fatalf("exact begin queue recheck is not bound to %s:\n%s", test.wantQueue, queue)
			}
		})
	}
}

func TestBeginApplyAllSourceRevisionAdvanceAttemptRequiresExactlyOneRow(t *testing.T) {
	for _, affected := range []int64{0, 2} {
		t.Run(fmt.Sprintf("%d rows", affected), func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			db.SetMaxOpenConns(1)
			admission := testApplyAllSourceRevisionAdvanceBeginAdmission(
				t, db, packs.AppPack, ApplyAllSourceRevisionAdvancePendingAdvance, installedPackStatusActive,
			)
			oldRow := admission.identity.proof.entries[0].current.row
			receipt := testApplyAllSourceRevisionAdvanceBeginReceipt(
				t, context.Background(), db, mock, admission, 0, packs.ApplyOptions{},
			)
			expectApplyAllSourceRevisionAdvanceExactIncarnationBegin(mock, oldRow, packs.AppPack, false)
			mock.ExpectExec("UPDATE _System_InstalledPack p").
				WillReturnResult(sqlmock.NewResult(0, affected))
			expectApplyAllSourceRevisionAdvanceExactIncarnationRollback(mock)

			attempt, err := beginApplyAllSourceRevisionAdvanceAttempt(
				context.Background(), receipt,
			)
			entry := admission.identity.proof.entries[0].current
			assertApplyAllSourceRevisionAdvancePreconditionError(
				t, err, ApplyAllSourceRevisionAdvancePlanState, entry.advance.index, entry.advance.Name,
			)
			if attempt.identity != nil {
				t.Fatalf("failed begin returned attempt %#v", attempt)
			}
			replayed, replayErr := beginApplyAllSourceRevisionAdvanceAttempt(
				context.Background(), receipt,
			)
			if replayErr == nil || replayed.identity != nil || !strings.Contains(replayErr.Error(), "already claimed") {
				t.Fatalf("replayed failed begin = %#v, %v, want claimed receipt error", replayed, replayErr)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

func TestBeginApplyAllSourceRevisionAdvanceAttemptRefusesQueuedOldIncarnation(t *testing.T) {
	for _, category := range []packs.Category{packs.AppPack, packs.ConnectorPack} {
		t.Run(string(category), func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			db.SetMaxOpenConns(1)
			admission := testApplyAllSourceRevisionAdvanceBeginAdmission(
				t, db, category, ApplyAllSourceRevisionAdvancePendingAdvance, installedPackStatusActive,
			)
			oldRow := admission.identity.proof.entries[0].current.row
			receipt := testApplyAllSourceRevisionAdvanceBeginReceipt(
				t, context.Background(), db, mock, admission, 0, packs.ApplyOptions{},
			)
			expectApplyAllSourceRevisionAdvanceExactIncarnationBegin(mock, oldRow, category, true)
			expectApplyAllSourceRevisionAdvanceExactIncarnationRollback(mock)

			attempt, err := beginApplyAllSourceRevisionAdvanceAttempt(
				context.Background(), receipt,
			)
			entry := admission.identity.proof.entries[0].current
			assertApplyAllSourceRevisionAdvancePreconditionError(
				t, err, ApplyAllSourceRevisionAdvancePlanState, entry.advance.index, entry.advance.Name,
			)
			if attempt.identity != nil || !strings.Contains(err.Error(), "queue is not drained") {
				t.Fatalf("queued begin = %#v, %v, want zero attempt and queue error", attempt, err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

func TestBeginApplyAllSourceRevisionAdvanceAttemptRejectsInvalidReceiptBeforeSQL(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	attempt, err := beginApplyAllSourceRevisionAdvanceAttempt(
		context.Background(), applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt{},
	)
	if err == nil || attempt.identity != nil {
		t.Fatalf("zero-receipt begin = %#v, %v, want zero attempt and error", attempt, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("invalid receipt reached SQL: %v", err)
	}
}

func TestBeginApplyAllSourceRevisionAdvanceAttemptRejectsMutatedReceiptBeforeSQL(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	admission := testApplyAllSourceRevisionAdvanceBeginAdmission(
		t, db, packs.AppPack, ApplyAllSourceRevisionAdvancePendingAdvance, installedPackStatusActive,
	)
	receipt := testApplyAllSourceRevisionAdvanceBeginReceipt(
		t, context.Background(), db, mock, admission, 0, packs.ApplyOptions{},
	)
	replacement := &applyAllSourceRevisionAdvanceSelectedLivePreflightClaim{}
	receipt.identity.claim = replacement
	receipt.identity.provenance.claim = replacement
	attempt, err := beginApplyAllSourceRevisionAdvanceAttempt(context.Background(), receipt)
	if err == nil || attempt.identity != nil {
		t.Fatalf("mutated-receipt begin = %#v, %v, want zero attempt and error", attempt, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("mutated receipt reached SQL: %v", err)
	}
}

func TestClaimApplyAllSourceRevisionAdvanceSelectedLivePreflightReceiptAllowsOneConcurrentWinner(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	admission := testApplyAllSourceRevisionAdvanceBeginAdmission(
		t, db, packs.AppPack, ApplyAllSourceRevisionAdvancePendingAdvance, installedPackStatusActive,
	)
	receipt := testApplyAllSourceRevisionAdvanceBeginReceipt(
		t, context.Background(), db, mock, admission, 0, packs.ApplyOptions{},
	)
	start := make(chan struct{})
	results := make(chan error, 2)
	var callers sync.WaitGroup
	callers.Add(2)
	for range 2 {
		go func(candidate applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt) {
			defer callers.Done()
			<-start
			_, err := claimApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(candidate)
			results <- err
		}(receipt)
	}
	close(start)
	callers.Wait()
	close(results)
	winners, replays := 0, 0
	for err := range results {
		if err == nil {
			winners++
			continue
		}
		if strings.Contains(err.Error(), "already claimed") {
			replays++
			continue
		}
		t.Fatalf("concurrent claim returned unexpected error: %v", err)
	}
	if winners != 1 || replays != 1 {
		t.Fatalf("concurrent claims = %d winners, %d replays; want one each", winners, replays)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("concurrent receipt claim reached SQL: %v", err)
	}
}

func TestBeginApplyAllSourceRevisionAdvanceAttemptConsumesReceiptBeforeSQL(t *testing.T) {
	begin := parseControlplaneFuncDecl(
		t, "apply_all_source_revision_advance_begin.go", "beginApplyAllSourceRevisionAdvanceAttempt",
	)
	claim := findFuncCallByName(begin, "claimApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt")
	transaction := findFuncCallByName(begin, "withApplyAllSourceRevisionAdvanceExactIncarnationStore")
	if claim == nil || transaction == nil || claim.Pos() >= transaction.Pos() {
		t.Fatal("Begin does not claim the selected live-preflight receipt before exact-incarnation SQL")
	}
	for _, forbidden := range []string{
		"prepareApplyAllSourceRevisionAdvanceAttempt",
		"installedPackHasRuntimeDeclarationColumns",
		"installedPackHasRuntimeStartupShallowDeclarationColumn",
		"installedPackHasRuntimeFormulaFunctionBundlesColumn",
		"installedPackHasHostRequirementsDigestColumn",
	} {
		if findFuncCallByName(begin, forbidden) != nil {
			t.Fatalf("Begin recreates receipt-owned authority through %s", forbidden)
		}
	}
}

func TestBeginApplyAllSourceRevisionAdvanceAttemptRetainsWholeAdmission(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	db.SetMaxOpenConns(1)
	proof := testApplyAllSourceRevisionAdvanceExecutionProofForDatabase(t, db, 3)
	atTarget := proof.entries[1]
	atTarget.currentTarget = atTarget.installedCompatibility
	atTarget.current = testApplyAllSourceRevisionAdvanceFrozenCurrentTargetProof(
		t, atTarget.current.advance, atTarget.currentTarget,
		installedPackStatusActive, atTarget.current.advance.Kind,
		testCurrentTargetV1Hash, nil,
	).snapshot.entries[0]
	if atTarget.current.pending != nil {
		t.Fatalf("interleaved at-target entry is pending: %#v", atTarget.current.pending)
	}
	proof.entries[1] = atTarget
	admission := testApplyAllSourceRevisionAdvanceExecutionAdmissionFromEntries(t, db, proof.entries)
	runtime := admission.identity.proof.executionRuntime
	registrations := runnerTestRegistrationSet(t)
	runtime.componentKindRegistrationSet = registrations
	runtime.provenance.componentKindRegistrationSet = registrations
	if err := validateApplyAllSourceRevisionAdvanceExecutionAdmission(admission); err != nil {
		t.Fatalf("validate nonempty-registration admission: %v", err)
	}
	selected := admission.identity.proof.entries[2]
	oldRow := selected.current.row

	options := packs.ApplyOptions{
		SkipPolymorphicDiscriminatorBackfill: true,
		RepairLegacyUniqueDuplicates:         true,
	}
	receipt := testApplyAllSourceRevisionAdvanceBeginReceipt(
		t, context.Background(), db, mock, admission, 1, options,
	)
	expectApplyAllSourceRevisionAdvanceExactIncarnationBegin(mock, oldRow, packs.AppPack, false)
	mock.ExpectExec("UPDATE _System_InstalledPack p").WillReturnResult(sqlmock.NewResult(0, 1))
	expectApplyAllSourceRevisionAdvanceExactIncarnationCommit(mock)

	attempt, err := beginApplyAllSourceRevisionAdvanceAttempt(
		context.Background(), receipt,
	)
	if err != nil {
		t.Fatalf("begin whole-tuple attempt: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
	identity := attempt.identity
	if identity == receipt.identity.prepared || receipt.identity.prepared.minted ||
		receipt.identity.prepared.installingRow != (installedPackRow{}) ||
		!receipt.identity.claim.claimed.Load() {
		t.Fatalf("Begin did not preserve immutable prepared evidence or claim the receipt")
	}
	if err := validateApplyAllSourceRevisionAdvanceAttemptIdentity(receipt.identity.prepared, false); err != nil {
		t.Fatalf("prepared receipt evidence changed after Begin: %v", err)
	}
	retained := identity.admission.identity
	if retained == nil || retained == admission.identity ||
		len(retained.proof.entries) != 3 || identity.selectedProofIndex != 2 ||
		identity.pendingOrdinal != 1 || identity.applyOptions != options ||
		retained.proof.executionRuntime != admission.identity.proof.executionRuntime ||
		identity.provenance.executionRuntime != admission.identity.proof.executionRuntime ||
		identity.installingRow != identity.provenance.installingRow {
		t.Fatalf("whole-tuple attempt identity = %#v", identity)
	}
	runtime = retained.proof.executionRuntime
	if !runtime.componentKindRegistrationSet.SameSnapshot(identity.provenance.registrations) ||
		!sameApplyAllSourceRevisionAdvanceRuntimeDeps(
			runtime.applyDeps, identity.provenance.executionRuntimeWitness.applyDeps,
		) {
		t.Fatal("attempt did not retain the exact runtime options and registration snapshot")
	}
	for index := range retained.proof.entries {
		sourceEntry := admission.identity.proof.entries[index]
		retainedEntry := retained.proof.entries[index]
		if !equalApplyAllSourceRevisionAdvanceExecutionProofEntry(retainedEntry, sourceEntry) ||
			(retainedEntry.current.pending == nil) != (sourceEntry.current.pending == nil) {
			t.Fatalf("attempt entry %d is not a detached exact admission entry", index)
		}
		if sourceEntry.current.pending != nil {
			pendingPointers := map[*PendingApplyAllSourceRevisionAdvance]struct{}{
				sourceEntry.current.pending:                                                     {},
				retainedEntry.current.pending:                                                   {},
				retained.proof.provenance.entries[index].expected.current.pending:               {},
				retained.provenance.entries[index].expected.current.pending:                     {},
				identity.provenance.admissionProvenance.entries[index].expected.current.pending: {},
			}
			if len(pendingPointers) != 5 {
				t.Fatalf("attempt entry %d pending witnesses alias", index)
			}
		}
		for _, got := range []*packs.SourceBackedPackIdentity{
			retained.proof.provenance.entries[index].currentTargetIdentity,
			retained.provenance.entries[index].currentTargetIdentity,
			identity.provenance.admissionProvenance.entries[index].currentTargetIdentity,
		} {
			if got != admission.identity.provenance.entries[index].currentTargetIdentity {
				t.Fatalf("attempt entry %d reminted its current-target identity", index)
			}
		}
		for _, got := range []*packs.SourceBackedExecutionAtomIdentity{
			retained.proof.provenance.entries[index].atomIdentity,
			retained.provenance.entries[index].atomIdentity,
			identity.provenance.admissionProvenance.entries[index].atomIdentity,
		} {
			if got != admission.identity.provenance.entries[index].atomIdentity {
				t.Fatalf("attempt entry %d reminted its execution-atom identity", index)
			}
		}
	}

	wantFirstName := retained.proof.entries[0].name
	admission.identity.proof.entries[0].name = "changed"
	admission.identity.proof.entries = admission.identity.proof.entries[:1]
	admission.identity.provenance.entries = admission.identity.provenance.entries[:1]
	if len(retained.proof.entries) != 3 || retained.proof.entries[0].name != wantFirstName ||
		validateApplyAllSourceRevisionAdvanceAttemptIdentity(identity, true) != nil {
		t.Fatal("attempt changed after source admission mutation")
	}
}

func TestApplyAllSourceRevisionAdvanceAttemptWitnessRejectsCoordinatedMutation(t *testing.T) {
	tests := []struct {
		name   string
		mutate func(*testing.T, *applyAllSourceRevisionAdvanceAttemptIdentity, *sql.DB)
	}{
		{name: "whole tuple shrink", mutate: func(_ *testing.T, identity *applyAllSourceRevisionAdvanceAttemptIdentity, _ *sql.DB) {
			retained := identity.admission.identity
			retained.proof.entries = retained.proof.entries[:2]
			retained.proof.provenance.entries = retained.proof.provenance.entries[:2]
			retained.proof.provenance.entryCount = 2
			retained.provenance.entries = retained.provenance.entries[:2]
			retained.provenance.entryCount = 2
		}},
		{name: "unselected tuple reorder", mutate: func(_ *testing.T, identity *applyAllSourceRevisionAdvanceAttemptIdentity, _ *sql.DB) {
			retained := identity.admission.identity
			retained.proof.entries[0], retained.proof.entries[1] = retained.proof.entries[1], retained.proof.entries[0]
			retained.proof.provenance.entries[0], retained.proof.provenance.entries[1] = retained.proof.provenance.entries[1], retained.proof.provenance.entries[0]
			retained.provenance.entries[0], retained.provenance.entries[1] = retained.provenance.entries[1], retained.provenance.entries[0]
		}},
		{name: "selected proof index retarget", mutate: func(_ *testing.T, identity *applyAllSourceRevisionAdvanceAttemptIdentity, _ *sql.DB) {
			identity.selectedProofIndex = 0
			identity.pendingOrdinal = 0
		}},
		{name: "execution runtime substitution", mutate: func(t *testing.T, identity *applyAllSourceRevisionAdvanceAttemptIdentity, db *sql.DB) {
			replacement, err := freezeApplyAllSourceRevisionAdvanceExecutionRuntime(
				testApplyAllSourceRevisionAdvanceExecutionProofRunner(db),
			)
			if err != nil {
				t.Fatalf("freeze replacement runtime: %v", err)
			}
			retained := identity.admission.identity
			retained.proof.executionRuntime = replacement
			retained.proof.provenance.executionRuntime = replacement
			retained.provenance.executionRuntime = replacement
		}},
		{name: "active runtime substitution", mutate: func(_ *testing.T, identity *applyAllSourceRevisionAdvanceAttemptIdentity, db *sql.DB) {
			replacement := &applyAllSourceRevisionAdvanceActiveRuntimeIdentity{
				db: db, snapshot: &packs.ActiveInstalledRuntimeSnapshot{},
			}
			retained := identity.admission.identity
			retained.proof.activeRuntimeIdentity = replacement
			retained.proof.provenance.activeRuntimeIdentity = replacement
			retained.proof.provenance.activeRuntimeDatabase = db
			retained.proof.provenance.activeRuntimeSnapshot = replacement.snapshot
			retained.provenance.activeRuntimeIdentity = replacement
			retained.provenance.activeRuntimeDatabase = db
			retained.provenance.activeRuntimeSnapshot = replacement.snapshot
		}},
		{name: "runtime apply options drift", mutate: func(_ *testing.T, identity *applyAllSourceRevisionAdvanceAttemptIdentity, _ *sql.DB) {
			runtime := identity.admission.identity.proof.executionRuntime
			runtime.applyDeps.Options.RepairLegacyUniqueDuplicates = true
			runtime.provenance.applyDeps.Options.RepairLegacyUniqueDuplicates = true
		}},
		{name: "runtime registration drift", mutate: func(t *testing.T, identity *applyAllSourceRevisionAdvanceAttemptIdentity, _ *sql.DB) {
			runtime := identity.admission.identity.proof.executionRuntime
			replacement, err := packs.NewComponentKindRegistrationSet(
				runtime.componentKindRegistrationSet.Registrations(),
			)
			if err != nil {
				t.Fatalf("rebuild equivalent registration set: %v", err)
			}
			if runtime.componentKindRegistrationSet.SameSnapshot(replacement) {
				t.Fatal("rebuilt registration set unexpectedly retained snapshot identity")
			}
			runtime.componentKindRegistrationSet = replacement
			runtime.provenance.componentKindRegistrationSet = replacement
		}},
		{name: "reminted current target identity", mutate: func(t *testing.T, identity *applyAllSourceRevisionAdvanceAttemptIdentity, _ *sql.DB) {
			retained := identity.admission.identity
			replacement, err := packs.FreezeSourceBackedPackIdentity(retained.proof.entries[0].currentTarget)
			if err != nil {
				t.Fatalf("remint current target identity: %v", err)
			}
			retained.proof.provenance.entries[0].currentTargetIdentity = replacement
			retained.provenance.entries[0].currentTargetIdentity = replacement
		}},
		{name: "reminted execution atom identity", mutate: func(t *testing.T, identity *applyAllSourceRevisionAdvanceAttemptIdentity, _ *sql.DB) {
			retained := identity.admission.identity
			replacement, err := packs.FreezeSourceBackedExecutionAtomIdentity(retained.proof.entries[0].atom)
			if err != nil {
				t.Fatalf("remint execution atom identity: %v", err)
			}
			retained.proof.provenance.entries[0].atomIdentity = replacement
			retained.provenance.entries[0].atomIdentity = replacement
		}},
		{name: "attempt apply options drift", mutate: func(_ *testing.T, identity *applyAllSourceRevisionAdvanceAttemptIdentity, _ *sql.DB) {
			identity.applyOptions.RepairLegacyUniqueDuplicates = true
		}},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db := &sql.DB{}
			proof := testApplyAllSourceRevisionAdvanceExecutionProofForDatabase(t, db, 3)
			admission := testApplyAllSourceRevisionAdvanceExecutionAdmissionFromEntries(t, db, proof.entries)
			runtime := admission.identity.proof.executionRuntime
			registrations := runnerTestRegistrationSet(t)
			runtime.componentKindRegistrationSet = registrations
			runtime.provenance.componentKindRegistrationSet = registrations
			if err := validateApplyAllSourceRevisionAdvanceExecutionAdmission(admission); err != nil {
				t.Fatalf("validate nonempty-registration admission: %v", err)
			}
			identity, err := prepareApplyAllSourceRevisionAdvanceAttempt(
				admission, 2, packs.ApplyOptions{},
			)
			if err != nil {
				t.Fatalf("prepare attempt: %v", err)
			}
			test.mutate(t, identity, db)
			if err := validateApplyAllSourceRevisionAdvanceExecutionAdmission(identity.admission); err != nil {
				t.Fatalf("coordinated mutation invalidated retained admission: %v", err)
			}
			if err := identity.admission.identity.proof.executionRuntime.validate(); err != nil {
				t.Fatalf("coordinated mutation invalidated retained runtime: %v", err)
			}
			if err := validateApplyAllSourceRevisionAdvanceAttemptIdentity(identity, false); err == nil {
				t.Fatal("coordinated mutation passed the attempt-level witness")
			}
		})
	}
}

func testApplyAllSourceRevisionAdvanceBeginAdmission(
	t *testing.T,
	db *sql.DB,
	category packs.Category,
	pendingKind ApplyAllSourceRevisionAdvancePendingKind,
	status string,
) applyAllSourceRevisionAdvanceExecutionAdmission {
	t.Helper()
	advance := testApplyAllSourceRevisionAdvance("alpha", 3)
	target := testApplyAllSourceRevisionAdvanceBeginTarget(t, advance, category, advance.ToRevision)
	currentTarget := target
	if pendingKind == ApplyAllSourceRevisionAdvancePendingAdvance {
		currentTarget = testApplyAllSourceRevisionAdvanceBeginTarget(t, advance, category, advance.FromRevision)
	}
	current := testApplyAllSourceRevisionAdvanceFrozenCurrentTargetProof(
		t, advance, currentTarget, status, advance.Kind,
		testCurrentTargetV1Hash, testCurrentTargetV1Hash,
	)
	if pendingKind == ApplyAllSourceRevisionAdvancePendingRetry {
		current.snapshot.entries[0].row.IsDeleted = 1
	}
	if current.snapshot.entries[0].pending == nil ||
		current.snapshot.entries[0].pending.Kind != pendingKind {
		t.Fatalf("fixture pending = %#v, want %s", current.snapshot.entries[0].pending, pendingKind)
	}
	atom, err := packs.FreezeSourceBackedExecutionAtom(target)
	if err != nil {
		t.Fatalf("freeze begin fixture atom: %v", err)
	}
	entry := applyAllSourceRevisionAdvanceExecutionProofEntry{
		name: advance.Name, current: current.snapshot.entries[0],
		currentTarget: currentTarget, installedCompatibility: target, atom: atom,
	}
	return testApplyAllSourceRevisionAdvanceExecutionAdmissionFromEntries(
		t, db, []applyAllSourceRevisionAdvanceExecutionProofEntry{entry},
	)
}

func testApplyAllSourceRevisionAdvanceBeginReceipt(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	mock sqlmock.Sqlmock,
	admission applyAllSourceRevisionAdvanceExecutionAdmission,
	pendingOrdinal int,
	applyOptions packs.ApplyOptions,
) applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt {
	t.Helper()
	prepared, err := prepareApplyAllSourceRevisionAdvanceAttempt(
		admission, pendingOrdinal, applyOptions,
	)
	if err != nil {
		t.Fatalf("prepare Begin live-preflight receipt: %v", err)
	}
	selected := prepared.admission.identity.proof.entries[prepared.selectedProofIndex]
	runtime := prepared.admission.identity.proof.executionRuntime
	originalSchemaProvisioner := schemaProvisioner
	schemaProvisioner = selectedLivePreflightSchemaProvisioner{
		SchemaProvisioner: originalSchemaProvisioner,
		verify: func(candidate *database.TiDBConnection) error {
			if candidate != runtime.runnerDatabase {
				return fmt.Errorf("readiness used a different TiDB connection")
			}
			return nil
		},
	}
	defer func() { schemaProvisioner = originalSchemaProvisioner }()

	if selected.installedCompatibility.Category() == packs.AppPack {
		mock.ExpectQuery(`(?s)SELECT Id.*FROM ProcessDefinition.*WHERE COALESCE\(OwnerPack, ''\) = \?`).
			WithArgs(selected.installedCompatibility.Name()).
			WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	}
	expectApplyAllSourceRevisionAdvanceAttributionInstalledPackListRows(mock)
	allowedObjects, err := AllowedL0ObjectAPINames(ctx, db)
	if err != nil {
		t.Fatalf("allowed L0 objects: %v", err)
	}
	for _, objectAPIName := range selected.installedCompatibility.OwnedObjectAPINames() {
		allowedObjects[strings.ToLower(objectAPIName)] = struct{}{}
	}
	allowedPacks := map[string]struct{}{
		strings.ToLower(selected.installedCompatibility.Name()): {},
	}
	for _, runtimePack := range packs.RuntimeRequiredPacks() {
		allowedPacks[strings.ToLower(runtimePack.Name())] = struct{}{}
	}
	expectSuccessfulApplyAllSourceRevisionAdvanceAttributionSurfaceChecks(
		mock, sortedKeys(allowedObjects), sortedKeys(allowedPacks),
	)
	expectInstalledPackRuntimeDeclarationColumns(mock, true)
	expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, true, true)
	expectApplyAllSourceRevisionAdvanceHostDigestColumn(mock, true)
	expectApplyAllSourceRevisionAdvancePreconditionRow(
		mock, selected.current.row.Name, selected.current.row,
	)
	receipt, err := preflightApplyAllSourceRevisionAdvanceSelectedTarget(ctx, prepared, nil)
	if err != nil {
		t.Fatalf("mint Begin selected live-preflight receipt: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("Begin selected live-preflight SQL: %v", err)
	}
	return receipt
}

func testApplyAllSourceRevisionAdvanceBeginTarget(
	t *testing.T,
	advance normalizedApplyAllSourceRevisionAdvance,
	category packs.Category,
	revision string,
) *packs.SourceBackedPack {
	t.Helper()
	target, err := packs.NewSourceBackedPack(&packs.ResolvedPackSource{
		Kind: advance.Kind, URI: advance.URI, Revision: revision,
		Manifest: &packs.PackManifest{
			ManifestFormatVersion: packs.SupportedManifestFormatVersion,
			Name:                  advance.Name, DisplayName: advance.Name, Category: category, Version: "1.0.0",
			PlatformCompatibility: packs.PlatformCompatibility{Min: "0.2.0", MaxExclusive: "0.3.0"},
			Components:            []packs.ManifestComponent{},
		},
		Files: map[string][]byte{}, FileRefs: []string{},
	})
	if err != nil {
		t.Fatalf("new begin target: %v", err)
	}
	return target
}

func testApplyAllSourceRevisionAdvanceBeginUpdateQuery(t *testing.T, queries []string) string {
	t.Helper()
	for _, query := range queries {
		if strings.Contains(query, "UPDATE _System_InstalledPack p") {
			return query
		}
	}
	t.Fatalf("exact begin update was not executed; queries: %#v", queries)
	return ""
}

func testApplyAllSourceRevisionAdvanceBeginLockQuery(t *testing.T, queries []string) string {
	t.Helper()
	for _, query := range queries {
		if strings.Contains(query, "FROM _System_InstalledPack p") && strings.Contains(query, "FOR UPDATE") {
			return query
		}
	}
	t.Fatalf("exact begin row lock was not executed; queries: %#v", queries)
	return ""
}

func testApplyAllSourceRevisionAdvanceBeginQueueQuery(t *testing.T, queries []string) string {
	t.Helper()
	for _, query := range queries {
		if strings.Contains(query, "SELECT 1") && strings.Contains(query, "FROM _System_ConnectorWorkItem w") && !strings.Contains(query, "UPDATE ") {
			return query
		}
	}
	t.Fatalf("exact begin queue recheck was not executed; queries: %#v", queries)
	return ""
}