package audit

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	infraPersistence "github.com/nexuscrm/persistence"
)

func openAuditTiDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host := envDefault("TIDB_HOST", "127.0.0.1")
	port := envDefault("TIDB_PORT", "4001")
	user := envDefault("TIDB_USER", "root")
	password := os.Getenv("TIDB_PASSWORD")
	admin, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/?parseTime=true&loc=UTC", user, password, host, port))
	if err != nil {
		t.Skipf("open TiDB: %v", err)
	}
	pingCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	if err := admin.PingContext(pingCtx); err != nil {
		_ = admin.Close()
		t.Skipf("TiDB unavailable at %s:%s: %v", host, port, err)
	}
	databaseName := fmt.Sprintf("nexuscrm_audit_outbox_%d", time.Now().UnixNano())
	if _, err := admin.Exec("CREATE DATABASE `" + databaseName + "`"); err != nil {
		_ = admin.Close()
		t.Fatalf("create test database: %v", err)
	}
	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?parseTime=true&loc=UTC", user, password, host, port, databaseName))
	if err != nil {
		t.Fatal(err)
	}
	if err := db.Ping(); err != nil {
		t.Fatal(err)
	}
	createAuditOutboxTables(t, db)
	return db, func() {
		_ = db.Close()
		_, _ = admin.Exec("DROP DATABASE IF EXISTS `" + databaseName + "`")
		_ = admin.Close()
	}
}

func createAuditOutboxTables(t *testing.T, db *sql.DB) {
	t.Helper()
	for _, statement := range []string{
		`CREATE TABLE _System_AuditOutbox (
			Id BIGINT PRIMARY KEY AUTO_INCREMENT,
			Payload JSON NOT NULL,
			EnqueuedAt DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
			ProcessedAt DATETIME(6) NULL,
			Attempts INT NOT NULL DEFAULT 0,
			LeaseHolderId VARCHAR(64) NULL,
			LeaseExpiresAt DATETIME(6) NULL,
			OwnershipEpoch DECIMAL(20,0) NOT NULL DEFAULT 0,
			LastError VARCHAR(1024) NULL,
			PoisonClass VARCHAR(32) NULL,
			RepairCount INT NOT NULL DEFAULT 0,
			LastRepairId VARCHAR(255) NULL,
			LastRepairReason VARCHAR(1024) NULL,
			LastRepairedAt DATETIME(6) NULL,
			CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
			LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
			INDEX idx_pending (ProcessedAt, PoisonClass, EnqueuedAt),
			INDEX idx_lease (LeaseHolderId, LeaseExpiresAt)
		)`,
		`CREATE TABLE _System_AuditLog (
			Id VARCHAR(255) PRIMARY KEY,
			ObjectApiName VARCHAR(255) NOT NULL,
			RecordId VARCHAR(255) NOT NULL,
			FieldName VARCHAR(255) NOT NULL,
			OldValue LONGTEXT NULL,
			NewValue LONGTEXT NULL,
			ChangedById VARCHAR(255) NOT NULL,
			ChangedAt DATETIME NOT NULL,
			EventAt DATETIME(6) NULL,
			HistorySequence BIGINT NULL,
			FieldHistoryEligible TINYINT(1) NULL,
			CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
			LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
			Entrypoint VARCHAR(32) NULL,
			Action VARCHAR(32) NULL,
			OutboxRowId BIGINT NULL,
			Outcome VARCHAR(16) NULL,
			UNIQUE KEY uk_outbox (OutboxRowId, FieldName, ChangedAt),
			KEY idx_history_changed_at (ObjectApiName, RecordId, FieldName, ChangedAt),
			KEY idx_audit_retention_changed_at (ChangedAt, Id)
		)`,
		`CREATE TABLE _System_RuntimeWorkOwnership (
			WorkFamily VARCHAR(64) PRIMARY KEY,
			DeploymentId VARCHAR(255) NOT NULL,
			WorkerReleaseDigest CHAR(71) NOT NULL,
			OwnershipEpoch DECIMAL(20,0) NOT NULL,
			IsActive TINYINT(1) NOT NULL DEFAULT 1,
			OperationId VARCHAR(255) NOT NULL,
			CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
			LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
			KEY idx_runtime_work_ownership_deployment (DeploymentId, OwnershipEpoch)
		)`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatalf("create audit test table: %v", err)
		}
	}
}

func installAuditWorkOwnershipFixture(t *testing.T, db *sql.DB, owner WorkOwnership) {
	t.Helper()
	const operationID = "audit-outbox-fixture-install"
	if !infraPersistence.ValidRuntimeOperationID(operationID) {
		t.Fatal("invalid fixture operation ID")
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, ?, 1, ?)`, owner.WorkFamily, owner.DeploymentID, owner.WorkerReleaseDigest, owner.EpochDecimal(), operationID); err != nil {
		t.Fatalf("install ownership: %v", err)
	}
}

func TestMySQLOutboxStoreTiDBClaimExpiryAndStaleOwnerFence(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	storeA := newMySQLOutboxStore(db)
	storeB := newMySQLOutboxStore(db)
	if err := storeA.CheckReady(context.Background()); err != nil {
		t.Fatalf("CheckReady: %v", err)
	}
	id := insertOutboxIntegrationRow(t, db, 0)
	leasesA, err := storeA.AcquireLease(context.Background(), "worker-a", 1, 250*time.Millisecond, 5)
	if err != nil || len(leasesA) != 1 {
		t.Fatalf("worker A acquire leases=%v err=%v", leasesA, err)
	}
	leasesB, err := storeB.AcquireLease(context.Background(), "worker-b", 1, time.Second, 5)
	if err != nil || len(leasesB) != 0 {
		t.Fatalf("live lease stolen: leases=%v err=%v", leasesB, err)
	}
	time.Sleep(350 * time.Millisecond)
	leasesB, err = storeB.AcquireLease(context.Background(), "worker-b", 1, time.Second, 5)
	if err != nil || len(leasesB) != 1 {
		t.Fatalf("expired lease not reclaimed: leases=%v err=%v", leasesB, err)
	}
	if err := storeA.CommitFanout(context.Background(), leasesA[0], []FieldRow{testFieldRow(leasesA[0])}); !errors.Is(err, ErrOutboxLeaseLost) {
		t.Fatalf("stale fanout error = %v", err)
	}
	if _, err := storeA.BumpAttemptsAfterRollback(context.Background(), leasesA[0], 5, "stale"); !errors.Is(err, ErrOutboxLeaseLost) {
		t.Fatalf("stale disposition error = %v", err)
	}
	var holder string
	var attempts int
	if err := db.QueryRow("SELECT LeaseHolderId, Attempts FROM _System_AuditOutbox WHERE Id = ?", id).Scan(&holder, &attempts); err != nil {
		t.Fatal(err)
	}
	if holder != "worker-b" || attempts != 0 {
		t.Fatalf("new owner mutated: holder=%q attempts=%d", holder, attempts)
	}
}

func TestFencedMySQLOutboxStoreTiDBOwnershipFlipSerializesWithAtomicFanout(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	owner := testWorkOwnership()
	installAuditWorkOwnershipFixture(t, db, owner)
	store, err := newFencedMySQLOutboxStore(db, owner)
	if err != nil {
		t.Fatal(err)
	}
	if err := store.CheckReady(context.Background()); err != nil {
		t.Fatalf("CheckReady: %v", err)
	}
	insertOutboxIntegrationRow(t, db, 0)
	leases, err := store.AcquireLease(context.Background(), "worker-old", 1, 5*time.Second, 5)
	if err != nil || len(leases) != 1 || leases[0].OwnershipEpoch != owner.OwnershipEpoch {
		t.Fatalf("AcquireLease leases=%v err=%v", leases, err)
	}

	locked := make(chan struct{})
	release := make(chan struct{})
	store.afterOwnershipLockTest = func() {
		close(locked)
		<-release
	}
	commitDone := make(chan error, 1)
	go func() {
		commitDone <- store.CommitFanout(context.Background(), leases[0], []FieldRow{testFieldRow(leases[0])})
	}()
	select {
	case <-locked:
	case <-time.After(5 * time.Second):
		t.Fatal("fanout transaction did not lock work ownership")
	}

	newDigest := "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
	flipDone := make(chan error, 1)
	go func() {
		_, flipErr := db.Exec(`UPDATE _System_RuntimeWorkOwnership
SET DeploymentId = ?, WorkerReleaseDigest = ?, OwnershipEpoch = ?, OperationId = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE WorkFamily = ?`, "runtime-worker-test2-2", newDigest, "8", "audit-outbox-fixture-flip", owner.WorkFamily)
		flipDone <- flipErr
	}()
	select {
	case flipErr := <-flipDone:
		t.Fatalf("ownership flip crossed an in-flight effect fence: %v", flipErr)
	case <-time.After(150 * time.Millisecond):
		// Expected: the atomic fanout owns the authority-row lock.
	}
	close(release)
	if err := <-commitDone; err != nil {
		t.Fatalf("fanout before ownership flip: %v", err)
	}
	if err := <-flipDone; err != nil {
		t.Fatalf("ownership flip after fanout: %v", err)
	}
	store.afterOwnershipLockTest = nil

	var count int
	if err := db.QueryRow("SELECT COUNT(*) FROM _System_AuditLog WHERE OutboxRowId = ?", leases[0].Row.ID).Scan(&count); err != nil || count != 1 {
		t.Fatalf("effect count=%d err=%v", count, err)
	}
	if err := db.QueryRow("SELECT COUNT(*) FROM _System_AuditOutbox WHERE Id = ?", leases[0].Row.ID).Scan(&count); err != nil || count != 0 {
		t.Fatalf("fanout did not consume outbox row count=%d err=%v", count, err)
	}
}

func TestFencedMySQLOutboxStoreTiDBRejectsStaleLeaseBeforeEffect(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	owner := testWorkOwnership()
	installAuditWorkOwnershipFixture(t, db, owner)
	store, err := newFencedMySQLOutboxStore(db, owner)
	if err != nil {
		t.Fatal(err)
	}
	insertOutboxIntegrationRow(t, db, 0)
	leases, err := store.AcquireLease(context.Background(), "worker-old", 1, 5*time.Second, 5)
	if err != nil || len(leases) != 1 {
		t.Fatalf("AcquireLease leases=%v err=%v", leases, err)
	}
	if _, err := db.Exec(`UPDATE _System_RuntimeWorkOwnership
SET DeploymentId = 'runtime-worker-new', WorkerReleaseDigest = ?, OwnershipEpoch = 8, OperationId = ?
WHERE WorkFamily = ?`, "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "audit-outbox-fixture-flip", owner.WorkFamily); err != nil {
		t.Fatal(err)
	}
	err = store.CommitFanout(context.Background(), leases[0], []FieldRow{testFieldRow(leases[0])})
	if !errors.Is(err, ErrOutboxWorkOwnershipLost) {
		t.Fatalf("stale effect error=%v", err)
	}
	var count int
	if err := db.QueryRow("SELECT COUNT(*) FROM _System_AuditLog").Scan(&count); err != nil || count != 0 {
		t.Fatalf("stale deployment produced %d effects err=%v", count, err)
	}
}

func TestFencedMySQLOutboxStoreTiDBRepairReplaysOnlyExactPoisonState(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	owner := testWorkOwnership()
	installAuditWorkOwnershipFixture(t, db, owner)
	store, err := newFencedMySQLOutboxStore(db, owner)
	if err != nil {
		t.Fatal(err)
	}
	id := insertOutboxIntegrationRow(t, db, 5)
	if _, err := db.Exec(`UPDATE _System_AuditOutbox
SET PoisonClass = 'RETRY_EXHAUSTED', LastError = 'dependency unavailable'
WHERE Id = ?`, id); err != nil {
		t.Fatal(err)
	}
	repair := OutboxRepair{
		OutboxRowID: id, ExpectedClass: PoisonClassRetryExhausted, ExpectedAttempts: 5,
		RepairID: "repair-test2-1", Reason: "dependency recovered and operator approved replay",
	}
	if err := store.RepairPoison(context.Background(), repair); err != nil {
		t.Fatalf("RepairPoison: %v", err)
	}
	var attempts, repairCount int
	var poison, lastError sql.NullString
	var repairID, reason string
	var repairedAt sql.NullTime
	if err := db.QueryRow(`SELECT Attempts, PoisonClass, LastError, RepairCount,
LastRepairId, LastRepairReason, LastRepairedAt FROM _System_AuditOutbox WHERE Id = ?`, id).
		Scan(&attempts, &poison, &lastError, &repairCount, &repairID, &reason, &repairedAt); err != nil {
		t.Fatal(err)
	}
	if attempts != 0 || poison.Valid || lastError.Valid || repairCount != 1 || repairID != repair.RepairID || reason != repair.Reason || !repairedAt.Valid {
		t.Fatalf("repaired state attempts=%d poison=%v error=%v count=%d id=%q reason=%q at=%v",
			attempts, poison, lastError, repairCount, repairID, reason, repairedAt)
	}
	if err := store.RepairPoison(context.Background(), repair); !errors.Is(err, ErrOutboxLeaseLost) {
		t.Fatalf("stale duplicate repair error=%v", err)
	}
	leases, err := store.AcquireLease(context.Background(), "worker-after-repair", 1, time.Second, 5)
	if err != nil || len(leases) != 1 || leases[0].Row.ID != id {
		t.Fatalf("repaired work was not replayable leases=%v err=%v", leases, err)
	}
}

func TestMySQLOutboxStoreTiDBStaleObservedAttemptsCannotMutateOwner(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	store := newMySQLOutboxStore(db)
	insertOutboxIntegrationRow(t, db, 0)
	leases, err := store.AcquireLease(context.Background(), "worker-a", 1, time.Second, 5)
	if err != nil || len(leases) != 1 {
		t.Fatalf("AcquireLease leases=%v err=%v", leases, err)
	}
	if _, err := db.Exec("UPDATE _System_AuditOutbox SET Attempts = Attempts + 1 WHERE Id = ?", leases[0].Row.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := store.BumpAttemptsAfterRollback(context.Background(), leases[0], 5, "stale attempts"); !errors.Is(err, ErrOutboxLeaseLost) {
		t.Fatalf("stale observed Attempts error=%v", err)
	}
	var attempts int
	var holder string
	if err := db.QueryRow("SELECT Attempts, LeaseHolderId FROM _System_AuditOutbox WHERE Id = ?", leases[0].Row.ID).Scan(&attempts, &holder); err != nil {
		t.Fatal(err)
	}
	if attempts != 1 || holder != "worker-a" {
		t.Fatalf("stale worker mutated row attempts=%d holder=%q", attempts, holder)
	}
}

func TestMySQLOutboxStoreTiDBAtomicExhaustionAndDifferentRowIsolation(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	store := newMySQLOutboxStore(db)
	firstID := insertOutboxIntegrationRow(t, db, 4)
	secondID := insertOutboxIntegrationRow(t, db, 0)
	leases, err := store.AcquireLease(context.Background(), "worker-a", 2, time.Second, 5)
	if err != nil || len(leases) != 2 {
		t.Fatalf("AcquireLease leases=%v err=%v", leases, err)
	}
	var first, second OutboxLease
	for _, lease := range leases {
		if lease.Row.ID == firstID {
			first = lease
		} else if lease.Row.ID == secondID {
			second = lease
		}
	}
	exhausted, err := store.BumpAttemptsAfterRollback(context.Background(), first, 5, "timeout")
	if err != nil || !exhausted {
		t.Fatalf("atomic exhaustion exhausted=%v err=%v", exhausted, err)
	}
	var attempts int
	var poison sql.NullString
	var holder sql.NullString
	if err := db.QueryRow("SELECT Attempts, PoisonClass, LeaseHolderId FROM _System_AuditOutbox WHERE Id = ?", firstID).Scan(&attempts, &poison, &holder); err != nil {
		t.Fatal(err)
	}
	if attempts != 5 || !poison.Valid || poison.String != string(PoisonClassRetryExhausted) || holder.Valid {
		t.Fatalf("exhausted row attempts=%d poison=%v holder=%v", attempts, poison, holder)
	}
	if err := store.CommitFanout(context.Background(), second, []FieldRow{testFieldRow(second)}); err != nil {
		t.Fatalf("different-row fanout: %v", err)
	}
	var remaining int
	if err := db.QueryRow("SELECT COUNT(*) FROM _System_AuditOutbox WHERE Id = ?", secondID).Scan(&remaining); err != nil || remaining != 0 {
		t.Fatalf("different row payload was not durably acknowledged and deleted remaining=%d err=%v", remaining, err)
	}
}

func TestMySQLOutboxStoreTiDBEventualExhaustionAcrossWorkerRestarts(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	id := insertOutboxIntegrationRow(t, db, 0)
	for observed := 0; observed < 5; observed++ {
		store := newMySQLOutboxStore(db)
		holder := fmt.Sprintf("worker-restart-%d", observed)
		leases, err := store.AcquireLease(context.Background(), holder, 1, time.Second, 5)
		if err != nil || len(leases) != 1 || leases[0].Row.Attempts != observed {
			t.Fatalf("iteration %d leases=%v err=%v", observed, leases, err)
		}
		exhausted, err := store.BumpAttemptsAfterRollback(context.Background(), leases[0], 5, "known rollback")
		if err != nil || exhausted != (observed == 4) {
			t.Fatalf("iteration %d exhausted=%v err=%v", observed, exhausted, err)
		}
	}
	var attempts int
	var poison string
	if err := db.QueryRow("SELECT Attempts, PoisonClass FROM _System_AuditOutbox WHERE Id = ?", id).Scan(&attempts, &poison); err != nil {
		t.Fatal(err)
	}
	if attempts != 5 || poison != string(PoisonClassRetryExhausted) {
		t.Fatalf("eventual exhaustion attempts=%d poison=%q", attempts, poison)
	}
	leases, err := newMySQLOutboxStore(db).AcquireLease(context.Background(), "worker-after-exhaustion", 1, time.Second, 5)
	if err != nil || len(leases) != 0 {
		t.Fatalf("exhausted row reclaimed leases=%v err=%v", leases, err)
	}
}

func TestMySQLOutboxStoreTiDBLegacyExactReplayAndAtomicFanout(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	store := newMySQLOutboxStore(db)
	insertOutboxIntegrationRow(t, db, 0)
	leases, err := store.AcquireLease(context.Background(), "worker-a", 1, 5*time.Second, 5)
	if err != nil || len(leases) != 1 {
		t.Fatalf("AcquireLease leases=%v err=%v", leases, err)
	}
	first := testFieldRow(leases[0])
	prepared, err := prepareFieldRows(leases[0], []FieldRow{first})
	if err != nil {
		t.Fatal(err)
	}
	// This is the only legacy intermediate state: a pre-atomic worker committed
	// the effect before it consumed the outbox row. The current worker verifies
	// exact replay and atomically consumes the retained row without duplication.
	if _, err := db.Exec(insertFieldRowSQL, fieldRowArgs(prepared[0])...); err != nil {
		t.Fatalf("seed legacy effect: %v", err)
	}
	if err := newMySQLOutboxStore(db).CommitFanout(context.Background(), leases[0], []FieldRow{first}); err != nil {
		t.Fatalf("atomic legacy replay: %v", err)
	}
	var count int
	if err := db.QueryRow("SELECT COUNT(*) FROM _System_AuditLog WHERE OutboxRowId = ?", leases[0].Row.ID).Scan(&count); err != nil || count != 1 {
		t.Fatalf("exact replay count=%d err=%v", count, err)
	}
	if err := db.QueryRow("SELECT COUNT(*) FROM _System_AuditOutbox WHERE Id = ?", leases[0].Row.ID).Scan(&count); err != nil || count != 0 {
		t.Fatalf("legacy replay did not consume outbox count=%d err=%v", count, err)
	}

	secondID := insertOutboxIntegrationRow(t, db, 0)
	secondLeases, err := store.AcquireLease(context.Background(), "worker-b", 1, 5*time.Second, 5)
	if err != nil || len(secondLeases) != 1 || secondLeases[0].Row.ID != secondID {
		t.Fatalf("second AcquireLease leases=%v err=%v", secondLeases, err)
	}
	valid := testFieldRow(secondLeases[0])
	invalid := valid
	invalid.FieldName = "Description"
	invalid.ObjectApiName = string(make([]byte, 256))
	err = store.CommitFanout(context.Background(), secondLeases[0], []FieldRow{valid, invalid})
	var operationErr *OutboxOperationError
	if !errors.As(err, &operationErr) || !operationErr.Deterministic || operationErr.CommitState != OutboxCommitRolledBack {
		t.Fatalf("data violation error=%#v", err)
	}
	if err := db.QueryRow("SELECT COUNT(*) FROM _System_AuditLog WHERE OutboxRowId = ?", secondID).Scan(&count); err != nil || count != 0 {
		t.Fatalf("partial fan-out count=%d err=%v", count, err)
	}
	if err := db.QueryRow("SELECT COUNT(*) FROM _System_AuditOutbox WHERE Id = ?", secondID).Scan(&count); err != nil || count != 1 {
		t.Fatalf("rolled-back fanout consumed outbox count=%d err=%v", count, err)
	}
}

func TestMySQLOutboxStoreTiDBRejectsNonExactReplay(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	store := newMySQLOutboxStore(db)
	insertOutboxIntegrationRow(t, db, 0)
	leases, err := store.AcquireLease(context.Background(), "worker-a", 1, 5*time.Second, 5)
	if err != nil || len(leases) != 1 {
		t.Fatalf("AcquireLease leases=%v err=%v", leases, err)
	}

	row := testFieldRow(leases[0])
	prepared, err := prepareFieldRows(leases[0], []FieldRow{row})
	if err != nil {
		t.Fatal(err)
	}
	conflicting := prepared[0]
	conflicting.Outcome = OutcomeFailure
	if _, err := db.Exec(insertFieldRowSQL, fieldRowArgs(conflicting)...); err != nil {
		t.Fatalf("seed conflicting legacy effect: %v", err)
	}

	err = store.CommitFanout(context.Background(), leases[0], []FieldRow{row})
	var operationErr *OutboxOperationError
	if !errors.As(err, &operationErr) ||
		!errors.Is(err, ErrOutboxNonExactReplay) ||
		!operationErr.Deterministic ||
		operationErr.PoisonClass != PoisonClassParseFail ||
		operationErr.CommitState != OutboxCommitRolledBack {
		t.Fatalf("non-exact replay error=%#v", err)
	}

	var outcome string
	if err := db.QueryRow(`SELECT Outcome FROM _System_AuditLog WHERE OutboxRowId = ?`, leases[0].Row.ID).Scan(&outcome); err != nil {
		t.Fatalf("read conflicting legacy effect: %v", err)
	}
	if outcome != string(OutcomeFailure) {
		t.Fatalf("conflicting legacy outcome=%q, want %q", outcome, OutcomeFailure)
	}
	var retained int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_AuditOutbox WHERE Id = ? AND ProcessedAt IS NULL`, leases[0].Row.ID).Scan(&retained); err != nil {
		t.Fatalf("read retained outbox row: %v", err)
	}
	if retained != 1 {
		t.Fatalf("retained outbox rows=%d, want 1", retained)
	}
}

func TestMySQLOutboxStoreTiDBReadinessFailsClosedForIndexDriftAndCancellation(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	store := newMySQLOutboxStore(db)
	if _, err := db.Exec("ALTER TABLE _System_AuditLog DROP INDEX uk_outbox"); err != nil {
		t.Fatal(err)
	}
	if err := store.CheckReady(context.Background()); err == nil {
		t.Fatal("readiness accepted missing exact replay index")
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	if err := store.CheckReady(ctx); err == nil {
		t.Fatal("readiness ignored canceled context")
	}
}

func TestMySQLOutboxStoreTiDB1205KnownRollbackAfterLeaseExpiry(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	var databaseName string
	if err := db.QueryRow("SELECT DATABASE()").Scan(&databaseName); err != nil {
		t.Fatal(err)
	}
	host := envDefault("TIDB_HOST", "127.0.0.1")
	port := envDefault("TIDB_PORT", "4001")
	user := envDefault("TIDB_USER", "root")
	workerDB, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?parseTime=true&loc=UTC", user, os.Getenv("TIDB_PASSWORD"), host, port, databaseName))
	if err != nil {
		t.Fatal(err)
	}
	defer workerDB.Close()
	workerDB.SetMaxOpenConns(1)
	workerDB.SetMaxIdleConns(1)
	if _, err := workerDB.Exec("SET SESSION innodb_lock_wait_timeout = 1"); err != nil {
		t.Fatalf("set bounded lock wait: %v", err)
	}
	store := newMySQLOutboxStore(workerDB)
	insertOutboxIntegrationRow(t, db, 0)
	leases, err := store.AcquireLease(context.Background(), "worker-a", 1, 200*time.Millisecond, 5)
	if err != nil || len(leases) != 1 {
		t.Fatalf("AcquireLease leases=%v err=%v", leases, err)
	}
	field := testFieldRow(leases[0])
	prepared, _ := prepareFieldRows(leases[0], []FieldRow{field})
	blocker, err := db.Begin()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := blocker.Exec(insertFieldRowSQL, fieldRowArgs(prepared[0])...); err != nil {
		_ = blocker.Rollback()
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	started := time.Now()
	err = store.CommitFanout(ctx, leases[0], []FieldRow{field})
	_ = blocker.Rollback()
	var operationErr *OutboxOperationError
	if !errors.As(err, &operationErr) || operationErr.CommitState != OutboxCommitRolledBack || operationErr.Deterministic {
		t.Fatalf("1205 error=%#v", err)
	}
	if time.Since(started) <= 200*time.Millisecond {
		t.Fatalf("lock failure did not exceed the original lease")
	}
	if _, err := store.BumpAttemptsAfterRollback(context.Background(), leases[0], 5, "1205"); err != nil {
		t.Fatalf("expired old holder could not win atomic disposition: %v", err)
	}
	var attempts int
	if err := db.QueryRow("SELECT Attempts FROM _System_AuditOutbox WHERE Id = ?", leases[0].Row.ID).Scan(&attempts); err != nil || attempts != 1 {
		t.Fatalf("attempts=%d err=%v", attempts, err)
	}
}

func TestMySQLOutboxStoreTiDBKnownRollbackAfterLeaseExpiryHasSingleWinner(t *testing.T) {
	db, cleanup := openAuditTiDB(t)
	defer cleanup()
	storeA := newMySQLOutboxStore(db)
	storeB := newMySQLOutboxStore(db)
	insertOutboxIntegrationRow(t, db, 0)
	leases, err := storeA.AcquireLease(context.Background(), "worker-a", 1, 150*time.Millisecond, 5)
	if err != nil || len(leases) != 1 {
		t.Fatalf("AcquireLease leases=%v err=%v", leases, err)
	}
	time.Sleep(250 * time.Millisecond)
	if _, err := storeA.BumpAttemptsAfterRollback(context.Background(), leases[0], 5, "known rollback"); err != nil {
		t.Fatalf("old disposition should win without a reclaimer: %v", err)
	}
	var attempts int
	if err := db.QueryRow("SELECT Attempts FROM _System_AuditOutbox WHERE Id = ?", leases[0].Row.ID).Scan(&attempts); err != nil || attempts != 1 {
		t.Fatalf("attempts=%d err=%v", attempts, err)
	}
	if _, err := db.Exec("DELETE FROM _System_AuditOutbox WHERE Id = ?", leases[0].Row.ID); err != nil {
		t.Fatal(err)
	}

	secondID := insertOutboxIntegrationRow(t, db, 0)
	leases, err = storeA.AcquireLease(context.Background(), "worker-a", 1, 150*time.Millisecond, 5)
	if err != nil || len(leases) != 1 || leases[0].Row.ID != secondID {
		t.Fatalf("second A acquire leases=%v err=%v", leases, err)
	}
	time.Sleep(250 * time.Millisecond)
	newOwner, err := storeB.AcquireLease(context.Background(), "worker-b", 10, time.Second, 5)
	if err != nil || len(newOwner) == 0 {
		t.Fatalf("B reclaimer acquire leases=%v err=%v", newOwner, err)
	}
	if _, err := storeA.BumpAttemptsAfterRollback(context.Background(), leases[0], 5, "stale"); !errors.Is(err, ErrOutboxLeaseLost) {
		t.Fatalf("old disposition after reclaimer error=%v", err)
	}
	var holder string
	if err := db.QueryRow("SELECT LeaseHolderId, Attempts FROM _System_AuditOutbox WHERE Id = ?", secondID).Scan(&holder, &attempts); err != nil {
		t.Fatal(err)
	}
	if holder != "worker-b" || attempts != 0 {
		t.Fatalf("new owner state holder=%q attempts=%d", holder, attempts)
	}
}

func insertOutboxIntegrationRow(t *testing.T, db *sql.DB, attempts int) int64 {
	t.Helper()
	result, err := db.Exec(`INSERT INTO _System_AuditOutbox (Payload, EnqueuedAt, Attempts)
VALUES ('{}', CURRENT_TIMESTAMP(6), ?)`, attempts)
	if err != nil {
		t.Fatal(err)
	}
	id, err := result.LastInsertId()
	if err != nil {
		t.Fatal(err)
	}
	return id
}

func envDefault(name, fallback string) string {
	if value := os.Getenv(name); value != "" {
		return value
	}
	return fallback
}