package packs

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"errors"
	"fmt"
	"io"
	"reflect"
	"regexp"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/ports"
	domainschema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/persistencewiring"
)

const (
	captureObjectInsert = "capture-object-insert"
	captureFieldInsert  = "capture-field-insert"
	objectParamWidth    = 35
	fieldParamWidth     = 42
	historyPrefetchSize = 50
)

func TestObjectSpecTrackHistoryMapsToFieldMetadata(t *testing.T) {
	specs, err := parseObjectSpecsJSON([]byte(`[
		{"apiName":"Account","keyPrefix":"001","fields":[
			{"apiName":"HistoryOn__c","label":"On","type":"string","trackHistory":true},
			{"apiName":"HistoryOff__c","label":"Off","type":"string","trackHistory":false}]},
		{"apiName":"Opportunity","keyPrefix":"006","fields":[
			{"apiName":"HistoryDefault__c","label":"Default","type":"string"}]}
	]`), "track-history fixture")
	if err != nil {
		t.Fatalf("parseObjectSpecsJSON() error = %v", err)
	}

	wantPrefixes := []string{"001", "006"}
	wantHistory := [][]bool{{true, false}, {false}}
	for objectIndex, spec := range specs {
		if spec.KeyPrefix != wantPrefixes[objectIndex] {
			t.Fatalf("spec %s key prefix = %q, want %q", spec.APIName, spec.KeyPrefix, wantPrefixes[objectIndex])
		}
		for fieldIndex, fieldSpec := range spec.Fields {
			field := buildPackFieldSpecMetadata(fieldSpec, "", objectSpecFieldResolver{})
			if field.TrackHistory != wantHistory[objectIndex][fieldIndex] {
				t.Fatalf("%s.%s TrackHistory = %t, want %t", spec.APIName, fieldSpec.APIName, field.TrackHistory, wantHistory[objectIndex][fieldIndex])
			}
		}
	}
}

func TestObjectSpecInlineHelpTextMapsToFieldMetadata(t *testing.T) {
	specs, err := parseObjectSpecsJSON([]byte(`[
		{"apiName":"Lead","fields":[
			{"apiName":"Region__c","label":"Region","type":"string","inlineHelpText":"Assigned from the address country"}]}
	]`), "inline-help-text fixture")
	if err != nil {
		t.Fatalf("parseObjectSpecsJSON() error = %v", err)
	}

	writeField := buildPackFieldSpecMetadata(specs[0].Fields[0], "", objectSpecFieldResolver{})
	if writeField.HelpText == nil || *writeField.HelpText != "Assigned from the address country" {
		t.Fatalf("pack write HelpText = %#v, want %q", writeField.HelpText, "Assigned from the address country")
	}

	objectField := buildObjectMetadata(specs)[0].Fields[0]
	if objectField.HelpText == nil || *objectField.HelpText != "Assigned from the address country" {
		t.Fatalf("object metadata HelpText = %#v, want %q", objectField.HelpText, "Assigned from the address country")
	}
}

func TestPrepareFreshHistoryTrackingFields_StagesOnlyMissingTrackedFields(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	fields := []metadata.FieldWithContext{
		metadataFieldFixture("Account", "obj_account", "fld_z", "Z__c", true),
		metadataFieldFixture("Account", "obj_account", "fld_existing", "Existing__c", true),
		metadataFieldFixture("Account", "obj_account", "fld_a", "A__c", true),
		metadataFieldFixture("Account", "obj_account", "fld_off", "Off__c", false),
	}
	expectHistoryFieldExistence(mock, "fld_z", false)
	expectHistoryFieldExistence(mock, "fld_existing", true)
	expectHistoryFieldExistence(mock, "fld_a", false)

	prepared, pending, err := prepareFreshHistoryTrackingFields(context.Background(), db, fields)
	if err != nil {
		t.Fatalf("prepareFreshHistoryTrackingFields() error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
	if got, want := pending, []string{"fld_a", "fld_z"}; !reflect.DeepEqual(got, want) {
		t.Fatalf("pending IDs = %v, want %v", got, want)
	}
	if prepared[0].Field.TrackHistory || !prepared[1].Field.TrackHistory || prepared[2].Field.TrackHistory || prepared[3].Field.TrackHistory {
		t.Fatalf("prepared TrackHistory states = [%t %t %t %t], want [false true false false]",
			prepared[0].Field.TrackHistory, prepared[1].Field.TrackHistory, prepared[2].Field.TrackHistory, prepared[3].Field.TrackHistory)
	}
	if !fields[0].Field.TrackHistory || !fields[1].Field.TrackHistory || !fields[2].Field.TrackHistory {
		t.Fatal("preparation mutated caller-owned field metadata")
	}
}

func TestObjectFieldMetadataAttempt_AccountOpportunityTrackHistoryReapplyIsAtomicAndIdempotent(t *testing.T) {
	objects, fields := accountOpportunityMetadataFixture()
	fields[1].Field.TrackHistory = true
	db, mock, matcher := newAttemptSQLMock(t)
	defer db.Close()
	deps := attemptDeps(t, db)

	var objectArgs, fieldArgs [][]driver.Value
	for replay := 0; replay < 2; replay++ {
		expectPessimisticBegin(mock)
		expectTrackedFieldClassification(mock, fields, replay > 0)
		expectObjectPermissionStateProjection(mock, objects, replay > 0, true)
		expectObjectAccessPolicyReads(mock, objects)
		objectArgs = append(objectArgs, expectCapturedExec(mock, captureObjectInsert, len(objects)*objectParamWidth, nil))
		expectObjectPermissionStateProjection(mock, objects, true, false)
		expectBatchHistoryStatePrefetch(mock, fields, replay > 0)
		expectPhysicalPreload(mock, fields)
		if replay > 0 {
			expectBatchHistoryNoOpClaims(mock, fields)
		}
		fieldArgs = append(fieldArgs, expectCapturedExec(mock, captureFieldInsert, len(fields)*fieldParamWidth, nil))
		if replay == 0 {
			expectPermissionStateAdvance(mock)
		}
		expectPessimisticCommit(mock)
		if replay == 0 {
			expectFreshHistoryActivation(mock, fields[0].FieldID, fields[1].FieldID)
		}
		if err := applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt(context.Background(), deps, "history-pack", nil, objects, fields, false); err != nil {
			t.Fatalf("replay %d: applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt() error = %v", replay+1, err)
		}
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
	objectQueries := matcher.queries[captureObjectInsert]
	fieldQueries := matcher.queries[captureFieldInsert]
	if len(objectQueries) != 2 || len(fieldQueries) != 2 {
		t.Fatalf("mutation inventory: object=%d field=%d, want 2 and 2", len(objectQueries), len(fieldQueries))
	}
	assertReplayStable(t, objectQueries[0], objectArgs[0], objectQueries[1], objectArgs[1], objectParamWidth, 5, 21)
	assertReplayStable(t, fieldQueries[0], fieldArgs[0], fieldQueries[1], fieldArgs[1], fieldParamWidth, 4, 18)
	assertDuplicateUpdateUsesNow(t, objectQueries[0])
	assertDuplicateUpdateUsesNow(t, fieldQueries[0])

	objectColumns := insertColumns(t, objectQueries[0], constants.TableObject)
	assertRowParam(t, objectColumns, objectArgs[0], objectParamWidth, 0, constants.FieldID, "obj_account")
	assertRowParam(t, objectColumns, objectArgs[0], objectParamWidth, 0, constants.FieldAPIName, "Account")
	assertRowParam(t, objectColumns, objectArgs[0], objectParamWidth, 0, constants.FieldSysObject_KeyPrefix, "001")
	assertRowParam(t, objectColumns, objectArgs[0], objectParamWidth, 1, constants.FieldAPIName, "Opportunity")
	assertRowParam(t, objectColumns, objectArgs[0], objectParamWidth, 1, constants.FieldSysObject_KeyPrefix, "006")

	fieldColumns := insertColumns(t, fieldQueries[0], constants.TableField)
	for row, field := range fields {
		assertRowParam(t, fieldColumns, fieldArgs[0], fieldParamWidth, row, constants.FieldID, field.FieldID)
		assertRowParam(t, fieldColumns, fieldArgs[0], fieldParamWidth, row, constants.FieldObjectID, field.ObjectID)
		assertRowParam(t, fieldColumns, fieldArgs[0], fieldParamWidth, row, constants.FieldAPIName, field.Field.APIName)
		assertRowParam(t, fieldColumns, fieldArgs[0], fieldParamWidth, row, constants.FieldSysField_TrackHistory, false)
		assertRowParam(t, fieldColumns, fieldArgs[0], fieldParamWidth, row, constants.FieldSysField_OwnerKind, constants.FieldOwnerKindPack)
		assertRowParam(t, fieldColumns, fieldArgs[0], fieldParamWidth, row, constants.FieldSysField_OwnerPack, "history-pack")
	}
	if !fields[0].Field.TrackHistory || !fields[1].Field.TrackHistory {
		t.Fatal("fresh-field staging mutated the caller's authored TrackHistory value")
	}
}

func TestObjectFieldMetadataAttempt_FreshHistoryActivationFailureRollsBackAllPendingFields(t *testing.T) {
	activationFailure := errors.New("second history activation failed")
	objects, fields := accountOpportunityMetadataFixture()
	fields[1].Field.TrackHistory = true
	db, mock, _ := newAttemptSQLMock(t)
	defer db.Close()

	expectPessimisticBegin(mock)
	expectTrackedFieldClassification(mock, fields, false)
	expectObjectPermissionStateProjection(mock, objects, false, true)
	expectObjectAccessPolicyReads(mock, objects)
	expectCapturedExec(mock, captureObjectInsert, len(objects)*objectParamWidth, nil)
	expectObjectPermissionStateProjection(mock, objects, true, false)
	expectBatchHistoryStatePrefetch(mock, fields, false)
	expectPhysicalPreload(mock, fields)
	expectCapturedExec(mock, captureFieldInsert, len(fields)*fieldParamWidth, nil)
	expectPermissionStateAdvance(mock)
	expectPessimisticCommit(mock)
	mock.ExpectBegin()
	expectFreshHistoryActivationMutation(mock, fields[0].FieldID)
	mock.ExpectQuery(`(?s)SELECT TrackHistory, HistoryTrackingRevision, IsCustomized, IsDeleted\s+FROM _System_Field\s+WHERE Id = \?`).
		WithArgs(fields[1].FieldID).
		WillReturnError(activationFailure)
	mock.ExpectRollback()

	err := applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt(context.Background(), attemptDeps(t, db), "history-pack", nil, objects, fields, false)
	if err == nil || !strings.Contains(err.Error(), activationFailure.Error()) {
		t.Fatalf("apply error = %v, want wrapped %q", err, activationFailure)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func TestObjectFieldMetadataAttempt_SecondFieldBatchFailureRollsBackEverything(t *testing.T) {
	secondBatchFailure := errors.New("second field batch failed")
	rollbackFailure := errors.New("rollback failed")
	for _, test := range []struct {
		name        string
		rollbackErr error
	}{
		{name: "rollback_succeeds"},
		{name: "rollback_error_fails_closed", rollbackErr: rollbackFailure},
	} {
		t.Run(test.name, func(t *testing.T) {
			objects, fields := largeMetadataFixture(51)
			db, mock, _ := newAttemptSQLMock(t)
			defer db.Close()

			expectPessimisticBegin(mock)
			expectTrackedFieldClassification(mock, fields, false)
			expectObjectPermissionStateProjection(mock, objects, false, true)
			expectObjectAccessPolicyReads(mock, objects)
			expectCapturedExec(mock, captureObjectInsert, len(objects)*objectParamWidth, nil)
			expectObjectPermissionStateProjection(mock, objects, true, false)
			expectBatchHistoryStatePrefetch(mock, fields, false)
			expectPhysicalPreload(mock, fields)
			expectCapturedExec(mock, captureFieldInsert, 50*fieldParamWidth, nil)
			expectCapturedExec(mock, captureFieldInsert, fieldParamWidth, secondBatchFailure)
			rollback := mock.ExpectRollback()
			if test.rollbackErr != nil {
				rollback.WillReturnError(test.rollbackErr)
			}
			mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
				WillReturnResult(sqlmock.NewResult(0, 0))

			err := applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt(context.Background(), attemptDeps(t, db), "history-pack", nil, objects, fields, false)
			if err == nil || !strings.Contains(err.Error(), secondBatchFailure.Error()) {
				t.Fatalf("apply error = %v, want wrapped %q", err, secondBatchFailure)
			}
			if test.rollbackErr != nil && !strings.Contains(err.Error(), test.rollbackErr.Error()) {
				t.Fatalf("apply error = %v, want fail-closed rollback error %q", err, test.rollbackErr)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

func TestObjectFieldMetadataAttempt_RequiresValidatedExecutor(t *testing.T) {
	typedNilManager := (*attemptTransactionManager)(nil)
	typedNilExecutor := (*nilExecutor)(nil)
	for _, test := range []struct {
		name string
		tm   ports.TransactionManager
	}{
		{name: "nil_manager"},
		{name: "typed_nil_manager", tm: typedNilManager},
		{name: "nil_extracted_executor", tm: &attemptTransactionManager{}},
		{name: "typed_nil_extracted_executor", tm: &attemptTransactionManager{tx: typedNilExecutor}},
		{name: "non_executor", tm: &attemptTransactionManager{tx: struct{}{}}},
	} {
		t.Run(test.name, func(t *testing.T) {
			pessimisticCalls := 0
			if tm, ok := test.tm.(*attemptTransactionManager); ok && tm != nil {
				tm.withPessimisticTransactionCalls = &pessimisticCalls
			}
			err := applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt(context.Background(), ApplyDeps{TxManager: test.tm}, "history-pack", nil, nil, nil, false)
			if err == nil {
				t.Fatal("applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt() error = nil, want fail-closed validation")
			}
			if test.name == "nil_manager" || test.name == "typed_nil_manager" {
				if pessimisticCalls != 0 {
					t.Fatalf("WithPessimisticTransaction calls = %d, want 0", pessimisticCalls)
				}
			} else if pessimisticCalls != 1 {
				t.Fatalf("WithPessimisticTransaction calls = %d, want 1", pessimisticCalls)
			}
		})
	}

	objects, fields := accountOpportunityMetadataFixture()
	db, mock, _ := newAttemptSQLMock(t)
	defer db.Close()
	expectPessimisticBegin(mock)
	expectObjectPermissionStateProjection(mock, objects[1:], false, true)
	expectObjectAccessPolicyReads(mock, objects[1:])
	expectCapturedExec(mock, captureObjectInsert, objectParamWidth, nil)
	expectObjectPermissionStateProjection(mock, objects[1:], true, false)
	expectBatchHistoryStatePrefetch(mock, fields[1:], false)
	expectPhysicalPreload(mock, fields[1:])
	expectCapturedExec(mock, captureFieldInsert, fieldParamWidth, nil)
	expectPermissionStateAdvance(mock)
	expectPessimisticCommit(mock)
	if err := applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt(context.Background(), attemptDeps(t, db), "history-pack", nil, objects[1:], fields[1:], false); err != nil {
		t.Fatalf("valid *sql.Tx executor rejected: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}

	t.Run("object_spec_rejects_invalid_manager_before_mutation", func(t *testing.T) {
		specs, err := parseObjectSpecsJSON([]byte(`[
			{"apiName":"Account","keyPrefix":"001","fields":[
				{"apiName":"History__c","label":"History","type":"string","trackHistory":true}]}
		]`), "transaction-manager preflight fixture")
		if err != nil {
			t.Fatalf("parseObjectSpecsJSON() error = %v", err)
		}

		typedNilManager := (*attemptTransactionManager)(nil)
		typedNilExecutor := (*nilExecutor)(nil)
		for _, test := range []struct {
			name                    string
			tm                      ports.TransactionManager
			wantWithTransactionCall int
			wantError               string
		}{
			{name: "nil_manager", wantError: "metadata apply for history-pack requires a transaction manager"},
			{name: "typed_nil_manager", tm: typedNilManager, wantError: "metadata apply for history-pack requires a transaction manager"},
			{name: "nil_extracted_executor", tm: &attemptTransactionManager{}, wantWithTransactionCall: 1, wantError: "extract transaction executor for metadata apply history-pack"},
			{name: "typed_nil_extracted_executor", tm: &attemptTransactionManager{tx: typedNilExecutor}, wantWithTransactionCall: 1, wantError: "extract transaction executor for metadata apply history-pack"},
			{name: "non_executor", tm: &attemptTransactionManager{tx: struct{}{}}, wantWithTransactionCall: 1, wantError: "extract transaction executor for metadata apply history-pack"},
		} {
			t.Run(test.name, func(t *testing.T) {
				transactionCalls := 0
				if tm, ok := test.tm.(*attemptTransactionManager); ok && tm != nil {
					tm.withTransactionCalls = &transactionCalls
				}
				mutationCalls := &attemptMutationCalls{}
				db := sql.OpenDB(attemptMutationConnector{calls: mutationCalls})
				defer db.Close()
				deps := ApplyDeps{
					DB:        db,
					Schema:    metadata.NewSchemaManager(persistence.NewSchemaRepository(db, nil)),
					TxManager: test.tm,
				}

				var panicValue interface{}
				var applyErr error
				func() {
					defer func() { panicValue = recover() }()
					applyErr = applyObjectSpecsWithResolverAndValidation(
						context.Background(), deps, "history-pack", specs, false, false,
						newObjectSpecFieldResolver(specs), packFieldAuthoringCompat{}, nil,
					)
				}()
				if panicValue != nil {
					t.Fatalf("applyObjectSpecsWithResolverAndValidation() panic = %v; mutation dependency was reached before TxManager validation", panicValue)
				}
				if applyErr == nil || applyErr.Error() != test.wantError {
					t.Fatalf("apply error = %v, want early transaction-manager validation", applyErr)
				}
				if transactionCalls != test.wantWithTransactionCall {
					t.Fatalf("WithTransaction calls = %d, want %d", transactionCalls, test.wantWithTransactionCall)
				}
				if *mutationCalls != (attemptMutationCalls{}) {
					t.Fatalf("mutation calls = %+v, want schema create/register/reconcile/index/metadata and all DB calls to remain 0", *mutationCalls)
				}
			})
		}
	})

	t.Run("source_backed_outer_entries_reject_invalid_manager_before_work", func(t *testing.T) {
		specs, err := parseObjectSpecsJSON([]byte(`[
			{"apiName":"Account","keyPrefix":"001","fields":[
				{"apiName":"LegacyOwned__c","label":"Legacy owned","type":"string"},
				{"apiName":"ManagedMissing__c","label":"Managed missing","type":"string","createable":false,"updateable":false}]}
		]`), "source-backed transaction-manager fixture")
		if err != nil {
			t.Fatalf("parseObjectSpecsJSON() error = %v", err)
		}
		component := &sourceBackedObjectSpecComponent{
			packName: "history-pack",
			manifestComponent: ManifestComponent{
				BusinessKey:   "history-pack:account",
				Manageability: ManageabilityLocked,
			},
			specs:                   specs,
			packObjectSpecs:         specs,
			allowOwnerlessObjectAPI: map[string]bool{"account": true},
		}
		typedNilManager := (*attemptTransactionManager)(nil)
		typedNilExecutor := (*nilExecutor)(nil)
		for _, entry := range []struct {
			name  string
			apply func(context.Context, ApplyDeps) error
		}{
			{name: "apply_blocks_legacy_ownership_adoption", apply: component.Apply},
			{name: "repair_blocks_repairable_missing_column", apply: component.repairSubstrate},
		} {
			t.Run(entry.name, func(t *testing.T) {
				for _, manager := range []struct {
					name                    string
					tm                      ports.TransactionManager
					wantWithTransactionCall int
					wantError               string
				}{
					{name: "nil_manager", wantError: "metadata apply for history-pack requires a transaction manager"},
					{name: "typed_nil_manager", tm: typedNilManager, wantError: "metadata apply for history-pack requires a transaction manager"},
					{name: "nil_extracted_executor", tm: &attemptTransactionManager{}, wantWithTransactionCall: 1, wantError: "extract transaction executor for metadata apply history-pack"},
					{name: "typed_nil_extracted_executor", tm: &attemptTransactionManager{tx: typedNilExecutor}, wantWithTransactionCall: 1, wantError: "extract transaction executor for metadata apply history-pack"},
					{name: "non_executor", tm: &attemptTransactionManager{tx: struct{}{}}, wantWithTransactionCall: 1, wantError: "extract transaction executor for metadata apply history-pack"},
				} {
					t.Run(manager.name, func(t *testing.T) {
						transactionCalls := 0
						if tm, ok := manager.tm.(*attemptTransactionManager); ok && tm != nil {
							tm.withTransactionCalls = &transactionCalls
						}
						mutationCalls := &attemptMutationCalls{}
						db := sql.OpenDB(attemptMutationConnector{calls: mutationCalls})
						defer db.Close()
						deps := ApplyDeps{
							DB:        db,
							Schema:    metadata.NewSchemaManager(persistence.NewSchemaRepository(db, nil)),
							TxManager: manager.tm,
						}

						err := entry.apply(context.Background(), deps)
						if err == nil || err.Error() != manager.wantError {
							t.Fatalf("outer entry error = %v, want early transaction-manager validation", err)
						}
						if transactionCalls != manager.wantWithTransactionCall {
							t.Fatalf("WithTransaction calls = %d, want %d", transactionCalls, manager.wantWithTransactionCall)
						}
						if *mutationCalls != (attemptMutationCalls{}) {
							t.Fatalf("pre-validation work/mutation calls = %+v, want resolver, legacy ownership, missing-column repair, reconciliation, and metadata calls all 0", *mutationCalls)
						}
					})
				}
			})
		}
	})
}

func TestObjectFieldMetadataAttempt_CommitUnknownFullReplayConverges(t *testing.T) {
	commitUnknown := errors.New("commit outcome unknown")
	objects, fields := accountOpportunityMetadataFixture()
	fields[0].Field.TrackHistory = false
	db, mock, matcher := newAttemptSQLMock(t)
	defer db.Close()
	deps := attemptDeps(t, db)

	var objectArgs, fieldArgs [][]driver.Value
	for attempt := 0; attempt < 2; attempt++ {
		expectPessimisticBegin(mock)
		expectObjectPermissionStateProjection(mock, objects, attempt > 0, true)
		expectObjectAccessPolicyReads(mock, objects)
		objectArgs = append(objectArgs, expectCapturedExec(mock, captureObjectInsert, len(objects)*objectParamWidth, nil))
		expectObjectPermissionStateProjection(mock, objects, true, false)
		expectBatchHistoryStatePrefetch(mock, fields, false)
		expectPhysicalPreload(mock, fields)
		fieldArgs = append(fieldArgs, expectCapturedExec(mock, captureFieldInsert, len(fields)*fieldParamWidth, nil))
		if attempt == 0 {
			expectPermissionStateAdvance(mock)
		}
		if attempt == 0 {
			mock.ExpectCommit().WillReturnError(commitUnknown)
		} else {
			mock.ExpectCommit()
		}
		expectPessimisticCleanup(mock)
		err := applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt(context.Background(), deps, "history-pack", nil, objects, fields, false)
		if attempt == 0 {
			if err == nil || !strings.Contains(err.Error(), commitUnknown.Error()) {
				t.Fatalf("first apply error = %v, want commit-unknown error", err)
			}
		} else if err != nil {
			t.Fatalf("full replay error = %v", err)
		}
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
	objectQueries := matcher.queries[captureObjectInsert]
	fieldQueries := matcher.queries[captureFieldInsert]
	if len(objectQueries) != 2 || len(fieldQueries) != 2 {
		t.Fatalf("internal retry detected or replay missing: object=%d field=%d", len(objectQueries), len(fieldQueries))
	}
	assertReplayStable(t, objectQueries[0], objectArgs[0], objectQueries[1], objectArgs[1], objectParamWidth, 5, 21)
	assertReplayStable(t, fieldQueries[0], fieldArgs[0], fieldQueries[1], fieldArgs[1], fieldParamWidth, 4, 18)
}

func TestObjectFieldMetadataAttempt_FreshHistoryCommitUnknownFullReplayConverges(t *testing.T) {
	commitUnknown := errors.New("commit outcome unknown")
	for _, test := range []struct {
		name         string
		unknownPhase int
	}{
		{name: "phase_one", unknownPhase: 1},
		{name: "phase_two", unknownPhase: 2},
	} {
		t.Run(test.name, func(t *testing.T) {
			objects, fields := accountOpportunityMetadataFixture()
			db, mock, _ := newAttemptSQLMock(t)
			defer db.Close()

			expectPessimisticBegin(mock)
			expectTrackedFieldClassification(mock, fields, false)
			expectObjectPermissionStateProjection(mock, objects, false, true)
			expectObjectAccessPolicyReads(mock, objects)
			expectCapturedExec(mock, captureObjectInsert, len(objects)*objectParamWidth, nil)
			expectObjectPermissionStateProjection(mock, objects, true, false)
			expectBatchHistoryStatePrefetch(mock, fields, false)
			expectPhysicalPreload(mock, fields)
			expectCapturedExec(mock, captureFieldInsert, len(fields)*fieldParamWidth, nil)
			expectPermissionStateAdvance(mock)
			if test.unknownPhase == 1 {
				mock.ExpectCommit().WillReturnError(commitUnknown)
				expectPessimisticCleanup(mock)
			} else {
				mock.ExpectCommit()
				expectPessimisticCleanup(mock)
				mock.ExpectBegin()
				expectFreshHistoryActivationMutation(mock, fields[0].FieldID)
				mock.ExpectCommit().WillReturnError(commitUnknown)
			}

			err := applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt(context.Background(), attemptDeps(t, db), "history-pack", nil, objects, fields, false)
			if err == nil || !strings.Contains(err.Error(), commitUnknown.Error()) {
				t.Fatalf("first apply error = %v, want commit-unknown error", err)
			}

			expectPessimisticBegin(mock)
			expectTrackedFieldClassification(mock, fields, true)
			expectObjectPermissionStateProjection(mock, objects, true, true)
			expectObjectAccessPolicyReads(mock, objects)
			expectCapturedExec(mock, captureObjectInsert, len(objects)*objectParamWidth, nil)
			expectObjectPermissionStateProjection(mock, objects, true, false)
			actualTracked := map[string]bool(nil)
			if test.unknownPhase == 1 {
				actualTracked = map[string]bool{
					fields[0].FieldID: false,
					fields[1].FieldID: false,
				}
			}
			expectBatchHistoryStatePrefetchWithActual(mock, fields, true, actualTracked)
			expectPhysicalPreload(mock, fields)
			expectBatchHistoryNoOpClaimsWithActual(mock, fields, actualTracked)
			expectCapturedExec(mock, captureFieldInsert, len(fields)*fieldParamWidth, nil)
			if test.unknownPhase == 1 {
				expectFreshHistoryActivationMutation(mock, fields[0].FieldID)
			}
			expectPessimisticCommit(mock)

			if err := applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt(context.Background(), attemptDeps(t, db), "history-pack", nil, objects, fields, false); err != nil {
				t.Fatalf("full replay error = %v", err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

type capturingQueryMatcher struct {
	queries map[string][]string
}

func (m *capturingQueryMatcher) Match(expectedSQL, actualSQL string) error {
	switch expectedSQL {
	case captureObjectInsert:
		if !strings.HasPrefix(actualSQL, "INSERT INTO `"+constants.TableObject+"`") {
			return fmt.Errorf("expected object insert, got %s", actualSQL)
		}
		m.queries[expectedSQL] = append(m.queries[expectedSQL], actualSQL)
		return nil
	case captureFieldInsert:
		if !strings.HasPrefix(actualSQL, "INSERT INTO `"+constants.TableField+"`") {
			return fmt.Errorf("expected field insert, got %s", actualSQL)
		}
		m.queries[expectedSQL] = append(m.queries[expectedSQL], actualSQL)
		return nil
	default:
		return sqlmock.QueryMatcherRegexp.Match(expectedSQL, actualSQL)
	}
}

type captureArgument struct {
	destination *driver.Value
}

func (a captureArgument) Match(value driver.Value) bool {
	*a.destination = value
	return true
}

func newAttemptSQLMock(t *testing.T) (*sql.DB, sqlmock.Sqlmock, *capturingQueryMatcher) {
	t.Helper()
	matcher := &capturingQueryMatcher{queries: make(map[string][]string)}
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(matcher))
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	return db, mock, matcher
}

func attemptDeps(t *testing.T, db *sql.DB) ApplyDeps {
	t.Helper()
	txManager := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	directWriters, err := persistencewiring.NewPermissionStateDirectWritersForTransactionManager(txManager)
	if err != nil {
		t.Fatalf("construct permission-state direct writers: %v", err)
	}
	schemaRepository := persistence.NewSchemaRepository(db, nil)
	if err := directWriters.BindSchema(schemaRepository); err != nil {
		t.Fatalf("bind schema permission-state direct writer: %v", err)
	}
	return ApplyDeps{
		DB:        db,
		TxManager: txManager,
		Schema:    metadata.NewSchemaManager(schemaRepository),
	}
}

func expectPessimisticBegin(mock sqlmock.Sqlmock) {
	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
}

func expectPessimisticCommit(mock sqlmock.Sqlmock) {
	mock.ExpectCommit()
	expectPessimisticCleanup(mock)
}

func expectPessimisticCleanup(mock sqlmock.Sqlmock) {
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))
}

func expectCapturedExec(mock sqlmock.Sqlmock, marker string, argCount int, execErr error) []driver.Value {
	captured := make([]driver.Value, argCount)
	matchers := make([]driver.Value, argCount)
	for i := range captured {
		matchers[i] = captureArgument{destination: &captured[i]}
	}
	expectation := mock.ExpectExec(marker).WithArgs(matchers...)
	if execErr != nil {
		expectation.WillReturnError(execErr)
	} else {
		expectation.WillReturnResult(sqlmock.NewResult(0, 1))
	}
	return captured
}

func expectObjectAccessPolicyReads(mock sqlmock.Sqlmock, objects []*models.ObjectMetadata) {
	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(objects)), ",")
	identityQuery := regexp.QuoteMeta(
		"SELECT `Id`, `ApiName` FROM `_System_Object` WHERE `Id` IN (" +
			placeholders + ") OR LOWER(`ApiName`) IN (" + placeholders + ") FOR UPDATE",
	)
	identityArgs := make([]driver.Value, 0, len(objects)*2)
	for _, object := range objects {
		identityArgs = append(identityArgs, object.ID)
	}
	for _, object := range objects {
		identityArgs = append(identityArgs, strings.ToLower(object.APIName))
	}
	mock.ExpectQuery(identityQuery).
		WithArgs(identityArgs...).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "ApiName"}))

	claimArgs := make([]driver.Value, len(objects)*objectParamWidth)
	for index := range claimArgs {
		claimArgs[index] = sqlmock.AnyArg()
	}
	mock.ExpectExec("(?s)INSERT INTO `_System_Object` .* ON DUPLICATE KEY UPDATE `Id` = `Id`").
		WithArgs(claimArgs...).
		WillReturnResult(sqlmock.NewResult(0, int64(len(objects))))

	claimedRows := sqlmock.NewRows([]string{"Id", "ApiName"})
	for _, object := range objects {
		claimedRows.AddRow(object.ID, object.APIName)
	}
	mock.ExpectQuery(identityQuery).
		WithArgs(identityArgs...).
		WillReturnRows(claimedRows)

	query := "SELECT .* FROM `_System_Object` WHERE LOWER\\(`ApiName`\\) = \\? AND COALESCE\\(`IsDeleted`, 0\\) = 0 LIMIT 1"
	for _, object := range objects {
		mock.ExpectQuery(query).
			WithArgs(strings.ToLower(object.APIName)).
			WillReturnRows(sqlmock.NewRows([]string{
				"AccessPolicy", "AccessPolicyRecordMutationPackName", "AccessPolicyRecordMutationPackValue",
				"AccessPolicyLifecyclePackName", "AccessPolicyLifecyclePackValue",
			}))
	}
}

func expectObjectPermissionStateProjection(
	mock sqlmock.Sqlmock,
	objects []*models.ObjectMetadata,
	existing bool,
	lock bool,
) {
	placeholders := make([]string, len(objects))
	args := make([]driver.Value, len(objects))
	rows := sqlmock.NewRows([]string{constants.FieldAPIName, constants.FieldIsDeleted})
	for index, object := range objects {
		placeholders[index] = "?"
		args[index] = object.APIName
		if existing {
			rows.AddRow(object.APIName, false)
		}
	}
	query := fmt.Sprintf(
		"SELECT `%s`, `%s` FROM `%s` WHERE `%s` IN (%s) ORDER BY `%s`, `%s`",
		constants.FieldAPIName,
		constants.FieldIsDeleted,
		constants.TableObject,
		constants.FieldAPIName,
		strings.Join(placeholders, ", "),
		constants.FieldAPIName,
		constants.FieldIsDeleted,
	)
	if lock {
		query += " FOR UPDATE"
	}
	mock.ExpectQuery(regexp.QuoteMeta(query)).
		WithArgs(args...).
		WillReturnRows(rows)
}

func expectPermissionStateAdvance(mock sqlmock.Sqlmock) {
	lineage := []byte("launch-proof-001")
	now := time.Unix(1, 0).UTC()
	mock.ExpectQuery(regexp.QuoteMeta(
		"SELECT SingletonKey, Lineage, CAST(Revision AS CHAR), EnforcementMode, CreatedDate, LastModifiedDate\n" +
			"FROM _System_PermissionStateVersion\n" +
			"ORDER BY SingletonKey\n" +
			"FOR UPDATE",
	)).WillReturnRows(sqlmock.NewRows([]string{
		"SingletonKey",
		"Lineage",
		"Revision",
		"EnforcementMode",
		"CreatedDate",
		"LastModifiedDate",
	}).AddRow(1, lineage, "1", "dark", now, now))
	mock.ExpectExec(regexp.QuoteMeta(
		"UPDATE _System_PermissionStateVersion\n"+
			"SET Revision = Revision + 1, LastModifiedDate = CURRENT_TIMESTAMP(6)\n"+
			"WHERE SingletonKey = 1 AND Lineage = ? AND Revision = ? AND EnforcementMode = ?",
	)).WithArgs(lineage, "1", "dark").WillReturnResult(sqlmock.NewResult(0, 1))
}

func expectTrackedFieldClassification(mock sqlmock.Sqlmock, fields []metadata.FieldWithContext, existing bool) {
	for _, field := range fields {
		if field.Field == nil || !field.Field.TrackHistory {
			continue
		}
		expectHistoryFieldExistence(mock, field.FieldID, existing)
	}
}

func expectHistoryFieldExistence(mock sqlmock.Sqlmock, fieldID string, existing bool) {
	rows := sqlmock.NewRows([]string{"1"})
	if existing {
		rows.AddRow(1)
	}
	mock.ExpectQuery(regexp.QuoteMeta("SELECT 1 FROM _System_Field WHERE Id = ?")).
		WithArgs(fieldID).
		WillReturnRows(rows)
}

func expectBatchHistoryStatePrefetch(mock sqlmock.Sqlmock, fields []metadata.FieldWithContext, existing bool) {
	expectBatchHistoryStatePrefetchWithActual(mock, fields, existing, nil)
}

func expectBatchHistoryStatePrefetchWithActual(mock sqlmock.Sqlmock, fields []metadata.FieldWithContext, existing bool, actualTracked map[string]bool) {
	sortedFields := append([]metadata.FieldWithContext(nil), fields...)
	sort.Slice(sortedFields, func(i, j int) bool {
		return sortedFields[i].FieldID < sortedFields[j].FieldID
	})
	for start := 0; start < len(sortedFields); start += historyPrefetchSize {
		end := start + historyPrefetchSize
		if end > len(sortedFields) {
			end = len(sortedFields)
		}
		chunk := sortedFields[start:end]
		args := make([]driver.Value, len(chunk))
		stateRows := sqlmock.NewRows([]string{"Id", "TrackHistory", "HistoryTrackingRevision", "IsCustomized", "IsDeleted"})
		for i, field := range chunk {
			args[i] = field.FieldID
			if existing {
				tracked := field.Field != nil && field.Field.TrackHistory
				if actual, ok := actualTracked[field.FieldID]; ok {
					tracked = actual
				}
				revision := int64(0)
				if tracked {
					revision = 1
				}
				stateRows.AddRow(field.FieldID, tracked, revision, false, false)
			}
		}
		mock.ExpectQuery(`(?s)SELECT Id, TrackHistory, HistoryTrackingRevision, IsCustomized, IsDeleted.*WHERE Id IN \([^)]*\)\s*$`).
			WithArgs(args...).
			WillReturnRows(stateRows)
		if !existing {
			continue
		}

		intervalArgs := make([]driver.Value, 0, len(args)*2)
		intervalArgs = append(intervalArgs, args...)
		intervalArgs = append(intervalArgs, args...)
		intervalRows := sqlmock.NewRows([]string{
			"FieldId", "count", "open", "valid_open", "stale_open_key",
			"invalid_bounds", "overlaps", "latest_start", "latest_end",
		})
		for _, field := range chunk {
			tracked := field.Field != nil && field.Field.TrackHistory
			if actual, ok := actualTracked[field.FieldID]; ok {
				tracked = actual
			}
			if tracked {
				intervalRows.AddRow(field.FieldID, 1, 1, 1, 0, 0, 0, int64(1), nil)
			}
		}
		mock.ExpectQuery(`(?s)SELECT intervals\.FieldId,.*FROM _System_FieldHistoryTrackingInterval intervals`).
			WithArgs(intervalArgs...).
			WillReturnRows(intervalRows)

	}
}

func expectBatchHistoryNoOpClaims(mock sqlmock.Sqlmock, fields []metadata.FieldWithContext) {
	expectBatchHistoryNoOpClaimsWithActual(mock, fields, nil)
}

func expectBatchHistoryNoOpClaimsWithActual(mock sqlmock.Sqlmock, fields []metadata.FieldWithContext, actualTracked map[string]bool) {
	sortedFields := append([]metadata.FieldWithContext(nil), fields...)
	sort.Slice(sortedFields, func(i, j int) bool {
		return sortedFields[i].FieldID < sortedFields[j].FieldID
	})
	claims := make([]metadata.FieldWithContext, 0, len(sortedFields))
	for _, field := range sortedFields {
		desired := field.Field != nil && field.Field.TrackHistory
		actual := desired
		if tracked, ok := actualTracked[field.FieldID]; ok {
			actual = tracked
		}
		if actual == desired {
			claims = append(claims, field)
		}
	}
	for start := 0; start < len(claims); start += historyPrefetchSize {
		end := start + historyPrefetchSize
		if end > len(claims) {
			end = len(claims)
		}
		chunk := claims[start:end]
		claimArgs := make([]driver.Value, 0, len(chunk)*4)
		for _, field := range chunk {
			tracked := field.Field != nil && field.Field.TrackHistory
			if actual, ok := actualTracked[field.FieldID]; ok {
				tracked = actual
			}
			revision := int64(0)
			if tracked {
				revision = 1
			}
			claimArgs = append(claimArgs, field.FieldID, tracked, revision, false)
		}
		mock.ExpectExec(`(?s)UPDATE _System_Field.*LastModifiedDate.*HistoryTrackingRevision`).
			WithArgs(claimArgs...).
			WillReturnResult(sqlmock.NewResult(0, int64(len(chunk))))
	}
}

func expectHistoryTrackingState(mock sqlmock.Sqlmock, fieldID string, tracked bool, revision, intervals, open int64, latestStart, latestEnd interface{}) {
	mock.ExpectQuery(`(?s)SELECT TrackHistory, HistoryTrackingRevision, IsCustomized, IsDeleted\s+FROM _System_Field\s+WHERE Id = \?`).
		WithArgs(fieldID).
		WillReturnRows(sqlmock.NewRows([]string{"TrackHistory", "HistoryTrackingRevision", "IsCustomized", "IsDeleted"}).
			AddRow(tracked, revision, false, false))
	mock.ExpectQuery(`(?s)SELECT COUNT\(\*\),.*FROM _System_FieldHistoryTrackingInterval.*WHERE FieldId = \?`).
		WithArgs(fieldID, fieldID).
		WillReturnRows(sqlmock.NewRows([]string{
			"intervals", "open", "validOpen", "staleOpenKey", "invalidBounds", "overlaps", "latestStart", "latestEnd",
		}).AddRow(intervals, open, open, 0, 0, 0, latestStart, latestEnd))
}

func expectFreshHistoryActivation(mock sqlmock.Sqlmock, fieldIDs ...string) {
	mock.ExpectBegin()
	for _, fieldID := range fieldIDs {
		expectFreshHistoryActivationMutation(mock, fieldID)
	}
	mock.ExpectCommit()
}

func expectFreshHistoryActivationMutation(mock sqlmock.Sqlmock, fieldID string) {
	expectHistoryTrackingState(mock, fieldID, false, 0, 0, 0, nil, nil)
	mock.ExpectExec(`(?s)INSERT INTO _System_FieldHistorySequence.*ON DUPLICATE KEY UPDATE`).
		WithArgs(fieldID).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectQuery(`(?s)SELECT CurrentSequence\s+FROM _System_FieldHistorySequence\s+WHERE FieldId = \?`).
		WithArgs(fieldID).
		WillReturnRows(sqlmock.NewRows([]string{"CurrentSequence"}).AddRow(1))
	boundary := time.Now().UTC()
	mock.ExpectQuery(regexp.QuoteMeta("SELECT UTC_TIMESTAMP(6)")).
		WillReturnRows(sqlmock.NewRows([]string{"UTC_TIMESTAMP(6)"}).AddRow(boundary))
	mock.ExpectExec(`(?s)UPDATE _System_Field\s+SET TrackHistory = \?,.*WHERE Id = \?`).
		WithArgs(true, false, boundary, fieldID, false, int64(0), false).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)INSERT INTO _System_FieldHistoryTrackingInterval.*VALUES \(\?, \?, \?, \?, NULL, \?, NULL\)`).
		WithArgs(sqlmock.AnyArg(), fieldID, fieldID, int64(1), boundary).
		WillReturnResult(sqlmock.NewResult(1, 1))
}

func expectPhysicalPreload(mock sqlmock.Sqlmock, fields []metadata.FieldWithContext) {
	tableSet := make(map[string]struct{})
	for _, field := range fields {
		tableSet[field.TableName] = struct{}{}
	}
	tables := make([]string, 0, len(tableSet))
	for tableName := range tableSet {
		tables = append(tables, tableName)
	}
	sort.Strings(tables)

	placeholders := make([]string, len(tables))
	args := make([]driver.Value, len(tables))
	for i, tableName := range tables {
		placeholders[i] = "?"
		args[i] = tableName
	}
	query := fmt.Sprintf(`
		SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, EXTRA, COLUMN_COMMENT, COLLATION_NAME
		FROM INFORMATION_SCHEMA.COLUMNS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME IN (%s)
	`, strings.Join(placeholders, ", "))
	rows := sqlmock.NewRows([]string{
		"TABLE_NAME", "COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE",
		"COLUMN_DEFAULT", "EXTRA", "COLUMN_COMMENT", "COLLATION_NAME",
	})
	for _, field := range fields {
		rows.AddRow(field.TableName, field.ColumnDefinition.Name, "varchar(255)", "YES", nil, nil, nil, "utf8mb4_bin")
	}
	mock.ExpectQuery(regexp.QuoteMeta(query)).WithArgs(args...).WillReturnRows(rows)
}

func accountOpportunityMetadataFixture() ([]*models.ObjectMetadata, []metadata.FieldWithContext) {
	accountPrefix := "001"
	opportunityPrefix := "006"
	objects := []*models.ObjectMetadata{
		{ID: "obj_account", APIName: "Account", KeyPrefix: &accountPrefix, Label: "Account", PluralLabel: "Accounts"},
		{ID: "obj_opportunity", APIName: "Opportunity", KeyPrefix: &opportunityPrefix, Label: "Opportunity", PluralLabel: "Opportunities"},
	}
	fields := []metadata.FieldWithContext{
		metadataFieldFixture("Account", "obj_account", "fld_account_history", "AccountHistory__c", true),
		metadataFieldFixture("Opportunity", "obj_opportunity", "fld_opportunity_history", "OpportunityHistory__c", false),
	}
	return objects, fields
}

func largeMetadataFixture(fieldCount int) ([]*models.ObjectMetadata, []metadata.FieldWithContext) {
	objects, _ := accountOpportunityMetadataFixture()
	fields := make([]metadata.FieldWithContext, 0, fieldCount)
	for i := 0; i < fieldCount; i++ {
		tableName, objectID := "Account", "obj_account"
		if i == fieldCount-1 {
			tableName, objectID = "Opportunity", "obj_opportunity"
		}
		fields = append(fields, metadataFieldFixture(tableName, objectID, fmt.Sprintf("fld_history_%02d", i), fmt.Sprintf("History%02d__c", i), i%2 == 0))
	}
	return objects, fields
}

func metadataFieldFixture(tableName, objectID, fieldID, apiName string, trackHistory bool) metadata.FieldWithContext {
	return metadata.FieldWithContext{
		TableName: tableName,
		ObjectID:  objectID,
		FieldID:   fieldID,
		ColumnDefinition: domainschema.ColumnDefinition{
			Name: apiName, Type: "VARCHAR(255)", LogicalType: string(constants.FieldTypeString), Nullable: true,
		},
		Field: &models.FieldMetadata{
			APIName: apiName, PhysicalName: apiName, Label: apiName, Type: constants.FieldTypeString,
			Createable: true, Updateable: true, TrackHistory: trackHistory,
		},
		OwnerKind: constants.FieldOwnerKindPack, OwnerPack: "history-pack", CapabilitiesDeclared: true,
	}
}

func assertReplayStable(t *testing.T, firstQuery string, firstArgs []driver.Value, secondQuery string, secondArgs []driver.Value, width int, timestampOffsets ...int) {
	t.Helper()
	if firstQuery != secondQuery {
		t.Fatalf("replay SQL changed\nfirst:  %s\nsecond: %s", firstQuery, secondQuery)
	}
	if len(firstArgs) != len(secondArgs) || len(firstArgs)%width != 0 {
		t.Fatalf("replay parameter shape changed: first=%d second=%d width=%d", len(firstArgs), len(secondArgs), width)
	}
	timestampSlots := make(map[int]bool, len(timestampOffsets))
	for _, offset := range timestampOffsets {
		timestampSlots[offset] = true
	}
	for i := range firstArgs {
		if timestampSlots[i%width] {
			if value, ok := firstArgs[i].(time.Time); !ok || value.IsZero() {
				t.Fatalf("first replay timestamp slot %d = %#v, want non-zero time.Time", i, firstArgs[i])
			}
			if value, ok := secondArgs[i].(time.Time); !ok || value.IsZero() {
				t.Fatalf("second replay timestamp slot %d = %#v, want non-zero time.Time", i, secondArgs[i])
			}
			continue
		}
		if !reflect.DeepEqual(firstArgs[i], secondArgs[i]) {
			t.Fatalf("stable replay parameter %d changed: first=%#v second=%#v", i, firstArgs[i], secondArgs[i])
		}
	}
}

func assertDuplicateUpdateUsesNow(t *testing.T, query string) {
	t.Helper()
	if !strings.Contains(query, "ON DUPLICATE KEY UPDATE") || !strings.Contains(query, "`"+constants.FieldLastModifiedDate+"` = NOW()") {
		t.Fatalf("upsert lost duplicate LastModifiedDate=NOW() contract: %s", query)
	}
}

func insertColumns(t *testing.T, query, table string) []string {
	t.Helper()
	prefix := "INSERT INTO `" + table + "` ("
	if !strings.HasPrefix(query, prefix) {
		t.Fatalf("unexpected insert prefix: %s", query)
	}
	remainder := strings.TrimPrefix(query, prefix)
	end := strings.Index(remainder, ") VALUES ")
	if end < 0 {
		t.Fatalf("insert missing VALUES boundary: %s", query)
	}
	rawColumns := strings.Split(remainder[:end], ", ")
	columns := make([]string, len(rawColumns))
	for i, raw := range rawColumns {
		columns[i] = strings.Trim(raw, "`")
	}
	return columns
}

func assertRowParam(t *testing.T, columns []string, args []driver.Value, width, row int, column string, want driver.Value) {
	t.Helper()
	for offset, candidate := range columns {
		if candidate == column {
			index := row*width + offset
			if index >= len(args) || !reflect.DeepEqual(args[index], want) {
				t.Fatalf("row %d %s param = %#v, want %#v", row, column, args[index], want)
			}
			return
		}
	}
	t.Fatalf("insert columns missing %s", column)
}

type attemptTransactionManager struct {
	tx                              interface{}
	withTransactionCalls            *int
	withPessimisticTransactionCalls *int
}

func (m *attemptTransactionManager) WithTransaction(ctx context.Context, fn func(context.Context) error) error {
	if m.withTransactionCalls != nil {
		*m.withTransactionCalls++
	}
	return fn(ctx)
}
func (m *attemptTransactionManager) WithPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	if m.withPessimisticTransactionCalls != nil {
		*m.withPessimisticTransactionCalls++
	}
	return fn(ctx)
}
func (m *attemptTransactionManager) WithRetry(ctx context.Context, fn func(context.Context) error, _ int) error {
	return fn(ctx)
}
func (m *attemptTransactionManager) WithPessimisticRetry(ctx context.Context, fn func(context.Context) error, _ int) error {
	return fn(ctx)
}
func (m *attemptTransactionManager) IsTransactionActive(context.Context) bool { return true }
func (m *attemptTransactionManager) IsPessimisticTx(context.Context) bool     { return false }
func (m *attemptTransactionManager) ExtractTx(context.Context) interface{}    { return m.tx }
func (m *attemptTransactionManager) DetachTx(ctx context.Context) context.Context {
	return ctx
}

type nilExecutor struct{}

func (*nilExecutor) Exec(string, ...interface{}) (sql.Result, error) { panic("unreachable") }
func (*nilExecutor) Query(string, ...interface{}) (*sql.Rows, error) { panic("unreachable") }
func (*nilExecutor) QueryRow(string, ...interface{}) *sql.Row        { panic("unreachable") }
func (*nilExecutor) ExecContext(context.Context, string, ...interface{}) (sql.Result, error) {
	panic("unreachable")
}
func (*nilExecutor) QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error) {
	panic("unreachable")
}
func (*nilExecutor) QueryRowContext(context.Context, string, ...interface{}) *sql.Row {
	panic("unreachable")
}

type attemptMutationCalls struct {
	totalDB         int
	schemaCreate    int
	tableRegister   int
	columnReconcile int
	indexReconcile  int
	metadataWriter  int
}

type attemptMutationConnector struct {
	calls *attemptMutationCalls
}

func (c attemptMutationConnector) Connect(context.Context) (driver.Conn, error) {
	return &attemptMutationConn{calls: c.calls}, nil
}

func (c attemptMutationConnector) Driver() driver.Driver {
	return attemptMutationDriver{calls: c.calls}
}

type attemptMutationDriver struct {
	calls *attemptMutationCalls
}

func (d attemptMutationDriver) Open(string) (driver.Conn, error) {
	return &attemptMutationConn{calls: d.calls}, nil
}

type attemptMutationConn struct {
	calls *attemptMutationCalls
}

func (c *attemptMutationConn) Prepare(query string) (driver.Stmt, error) {
	c.record(query)
	return nil, errors.New("unexpected mutation prepare")
}

func (*attemptMutationConn) Close() error { return nil }

func (c *attemptMutationConn) Begin() (driver.Tx, error) {
	c.calls.totalDB++
	c.calls.metadataWriter++
	return nil, errors.New("unexpected mutation transaction")
}

func (c *attemptMutationConn) ExecContext(_ context.Context, query string, _ []driver.NamedValue) (driver.Result, error) {
	c.record(query)
	return nil, errors.New("unexpected mutation exec")
}

func (c *attemptMutationConn) QueryContext(_ context.Context, query string, _ []driver.NamedValue) (driver.Rows, error) {
	c.record(query)
	return attemptMutationRows{}, errors.New("unexpected mutation query")
}

func (c *attemptMutationConn) record(query string) {
	c.calls.totalDB++
	upper := strings.ToUpper(query)
	switch {
	case strings.Contains(upper, "CREATE TABLE"):
		c.calls.schemaCreate++
	case strings.Contains(query, constants.TableTable):
		c.calls.tableRegister++
	case strings.Contains(upper, "INFORMATION_SCHEMA.COLUMNS") || strings.Contains(upper, "ALTER TABLE"):
		c.calls.columnReconcile++
	case strings.Contains(upper, "INDEX"):
		c.calls.indexReconcile++
	case strings.Contains(query, constants.TableObject) || strings.Contains(query, constants.TableField):
		c.calls.metadataWriter++
	}
}

type attemptMutationRows struct{}

func (attemptMutationRows) Columns() []string         { return nil }
func (attemptMutationRows) Close() error              { return nil }
func (attemptMutationRows) Next([]driver.Value) error { return io.EOF }