package audit

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/binary"
	"errors"
	"fmt"
	"strings"
	"time"

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

const (
	auditOutboxTable = "_System_AuditOutbox"
	auditLogTable    = "_System_AuditLog"
)

var requiredOutboxColumns = map[string]string{
	"Id":               "bigint",
	"Payload":          "json",
	"EnqueuedAt":       "datetime(6)",
	"ProcessedAt":      "datetime(6)",
	"Attempts":         "int",
	"LeaseHolderId":    "varchar(64)",
	"LeaseExpiresAt":   "datetime(6)",
	"OwnershipEpoch":   "decimal(20,0)",
	"LastError":        "varchar(1024)",
	"PoisonClass":      "varchar(32)",
	"RepairCount":      "int",
	"LastRepairId":     "varchar(255)",
	"LastRepairReason": "varchar(1024)",
	"LastRepairedAt":   "datetime(6)",
}

var requiredAuditLogColumns = map[string]string{
	"Id":                   "varchar(255)",
	"ObjectApiName":        "varchar(255)",
	"RecordId":             "varchar(255)",
	"FieldName":            "varchar(255)",
	"OldValue":             "longtext",
	"NewValue":             "longtext",
	"ChangedById":          "varchar(255)",
	"ChangedAt":            "datetime",
	"EventAt":              "datetime(6)",
	"HistorySequence":      "bigint",
	"FieldHistoryEligible": "tinyint(1)",
	"Entrypoint":           "varchar(32)",
	"Action":               "varchar(32)",
	"OutboxRowId":          "bigint",
	"Outcome":              "varchar(16)",
}

type mysqlOutboxStore struct {
	db                     *sql.DB
	ownership              *WorkOwnership
	workOwnership          *persistence.RuntimeWorkOwnershipRepository
	afterOwnershipLockTest func()
}

func newMySQLOutboxStore(db *sql.DB) *mysqlOutboxStore {
	return &mysqlOutboxStore{db: db}
}

func newFencedMySQLOutboxStore(db *sql.DB, ownership WorkOwnership) (*mysqlOutboxStore, error) {
	repository, err := persistence.NewRuntimeWorkOwnershipRepository(db, ownership)
	if err != nil {
		return nil, err
	}
	return &mysqlOutboxStore{db: db, ownership: &ownership, workOwnership: repository}, nil
}

func (s *mysqlOutboxStore) CheckReady(ctx context.Context) error {
	if s == nil || s.db == nil {
		return errors.New("audit outbox: nil database")
	}
	if err := s.db.PingContext(ctx); err != nil {
		return fmt.Errorf("audit outbox readiness ping: %w", err)
	}
	if err := s.checkColumns(ctx, auditOutboxTable, requiredOutboxColumns); err != nil {
		return err
	}
	if err := s.checkColumns(ctx, auditLogTable, requiredAuditLogColumns); err != nil {
		return err
	}
	if s.ownership != nil {
		if s.workOwnership == nil {
			return errors.New("audit outbox: work ownership repository is unavailable")
		}
		if err := s.workOwnership.CheckReady(ctx); err != nil {
			return err
		}
	}
	if err := s.checkExactIndex(ctx, auditLogTable, "uk_outbox", true, []string{"OutboxRowId", "FieldName", "ChangedAt"}); err != nil {
		return err
	}
	if err := s.checkIndexColumns(ctx, auditLogTable, false, []string{"ObjectApiName", "RecordId", "FieldName", "ChangedAt"}); err != nil {
		return err
	}
	if err := s.checkExactIndex(ctx, auditLogTable, "idx_audit_retention_changed_at", false, []string{"ChangedAt", "Id"}); err != nil {
		return err
	}

	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return fmt.Errorf("audit outbox readiness claim probe begin: %w", err)
	}
	defer func() { _ = cleanup() }()
	if err := s.lockActiveOwnership(ctx, tx); err != nil {
		_ = tx.Rollback()
		return fmt.Errorf("audit outbox readiness ownership probe: %w", err)
	}
	result, execErr := tx.ExecContext(ctx, s.acquireLeaseUpdateSQL(),
		s.acquireLeaseUpdateArgs("readiness-probe", int64(time.Second/time.Microsecond), int64(-1), 0, 1)...)
	if execErr == nil {
		var affected int64
		affected, execErr = result.RowsAffected()
		if execErr == nil && affected != 0 {
			execErr = fmt.Errorf("impossible claim probe affected %d rows", affected)
		}
	}
	if execErr == nil {
		result, execErr = tx.ExecContext(ctx, deleteAuditRowsBeforeSQL, time.Date(1000, 1, 1, 0, 0, 0, 0, time.UTC), auditRetentionDeleteBatchSize)
		if execErr == nil {
			var affected int64
			affected, execErr = result.RowsAffected()
			if execErr == nil && affected != 0 {
				execErr = fmt.Errorf("impossible retention probe affected %d rows", affected)
			}
		}
	}
	rollbackErr := tx.Rollback()
	if execErr != nil {
		return fmt.Errorf("audit outbox readiness claim probe: %w", execErr)
	}
	if rollbackErr != nil {
		return fmt.Errorf("audit outbox readiness claim probe rollback: %w", rollbackErr)
	}
	return nil
}

func (s *mysqlOutboxStore) lockActiveOwnership(ctx context.Context, tx *sql.Tx) error {
	if s.ownership == nil {
		return nil
	}
	if s.workOwnership == nil {
		return errors.New("audit outbox: work ownership repository is unavailable")
	}
	if err := s.workOwnership.LockActive(ctx, tx); err != nil {
		return err
	}
	if s.afterOwnershipLockTest != nil {
		s.afterOwnershipLockTest()
	}
	return nil
}

func (s *mysqlOutboxStore) checkColumns(ctx context.Context, table string, required map[string]string) error {
	if err := persistence.VerifyRequiredTableColumns(ctx, s.db, table, required); err != nil {
		return fmt.Errorf("audit outbox readiness: %w", err)
	}
	return nil
}

func (s *mysqlOutboxStore) checkExactIndex(ctx context.Context, table, index string, unique bool, want []string) error {
	if err := persistence.VerifyExactTableIndex(ctx, s.db, table, index, unique, want); err != nil {
		return fmt.Errorf("audit outbox readiness: %w", err)
	}
	return nil
}

func (s *mysqlOutboxStore) checkIndexColumns(ctx context.Context, table string, unique bool, want []string) error {
	rows, err := s.db.QueryContext(ctx, `SELECT NON_UNIQUE, GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX)
FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
GROUP BY INDEX_NAME, NON_UNIQUE`, table)
	if err != nil {
		return fmt.Errorf("audit outbox readiness indexes for %s: %w", table, err)
	}
	defer rows.Close()
	wantColumns := strings.ToLower(strings.Join(want, ","))
	for rows.Next() {
		var nonUnique int
		var columns string
		if err := rows.Scan(&nonUnique, &columns); err != nil {
			return fmt.Errorf("audit outbox readiness scan indexes for %s: %w", table, err)
		}
		if (nonUnique == 0) == unique && strings.EqualFold(columns, wantColumns) {
			return nil
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("audit outbox readiness iterate indexes for %s: %w", table, err)
	}
	return fmt.Errorf("audit outbox readiness: %s requires index columns %v unique=%t", table, want, unique)
}

const acquireLeaseCandidatesSQL = `SELECT Id, Payload, EnqueuedAt, Attempts
FROM _System_AuditOutbox
WHERE ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND Attempts < ?
  AND (LeaseHolderId IS NULL OR LeaseExpiresAt IS NULL OR LeaseExpiresAt <= CURRENT_TIMESTAMP(6))
ORDER BY EnqueuedAt, Id
LIMIT ?`

const observeQueueSQL = `SELECT
       COALESCE(SUM(CASE WHEN PoisonClass IS NULL THEN 1 ELSE 0 END), 0),
       COALESCE(SUM(CASE WHEN PoisonClass IS NOT NULL THEN 1 ELSE 0 END), 0),
       COALESCE(TIMESTAMPDIFF(MICROSECOND, MIN(CASE WHEN PoisonClass IS NULL THEN EnqueuedAt END), CURRENT_TIMESTAMP(6)), 0)
FROM _System_AuditOutbox
WHERE ProcessedAt IS NULL`

func (s *mysqlOutboxStore) ObserveQueue(ctx context.Context) (OutboxQueueState, error) {
	if s == nil || s.db == nil {
		return OutboxQueueState{}, errors.New("audit outbox: nil database")
	}
	if s.ownership != nil {
		if s.workOwnership == nil {
			return OutboxQueueState{}, errors.New("audit outbox: work ownership repository is unavailable")
		}
		if err := s.workOwnership.VerifyActive(ctx); err != nil {
			return OutboxQueueState{}, err
		}
	}
	var depth, poisonDepth, oldestMicros int64
	if err := s.db.QueryRowContext(ctx, observeQueueSQL).Scan(&depth, &poisonDepth, &oldestMicros); err != nil {
		return OutboxQueueState{}, fmt.Errorf("audit outbox: observe queue: %w", err)
	}
	if depth < 0 || poisonDepth < 0 || oldestMicros < 0 {
		return OutboxQueueState{}, errors.New("audit outbox: invalid queue aggregate")
	}
	return OutboxQueueState{Depth: depth, PoisonDepth: poisonDepth, OldestAge: time.Duration(oldestMicros) * time.Microsecond}, nil
}

const acquireLeaseUpdateSQL = `UPDATE _System_AuditOutbox
SET LeaseHolderId = ?,
    LeaseExpiresAt = TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)),
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND (LeaseHolderId IS NULL OR LeaseExpiresAt IS NULL OR LeaseExpiresAt <= CURRENT_TIMESTAMP(6))
  AND Attempts = ?
  AND Attempts < ?`

const acquireFencedLeaseUpdateSQL = `UPDATE _System_AuditOutbox
SET LeaseHolderId = ?,
    LeaseExpiresAt = TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)),
    OwnershipEpoch = ?,
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND (LeaseHolderId IS NULL OR LeaseExpiresAt IS NULL OR LeaseExpiresAt <= CURRENT_TIMESTAMP(6))
  AND Attempts = ?
  AND Attempts < ?`

func (s *mysqlOutboxStore) acquireLeaseUpdateSQL() string {
	if s.ownership != nil {
		return acquireFencedLeaseUpdateSQL
	}
	return acquireLeaseUpdateSQL
}

func (s *mysqlOutboxStore) acquireLeaseUpdateArgs(holderID string, leaseMicros, rowID int64, attempts, maxAttempts int) []any {
	if s.ownership != nil {
		return []any{holderID, leaseMicros, s.ownership.EpochDecimal(), rowID, attempts, maxAttempts}
	}
	return []any{holderID, leaseMicros, rowID, attempts, maxAttempts}
}

func (s *mysqlOutboxStore) AcquireLease(ctx context.Context, holderID string, batchSize int, leaseDuration time.Duration, maxAttempts int) ([]OutboxLease, error) {
	if s == nil || s.db == nil {
		return nil, errors.New("audit outbox: nil database")
	}
	if len(holderID) < 1 || len(holderID) > 64 {
		return nil, fmt.Errorf("audit outbox: holder ID length must be 1..64")
	}
	if batchSize <= 0 || maxAttempts <= 0 || leaseDuration <= 0 {
		return nil, fmt.Errorf("audit outbox: invalid acquire bounds")
	}

	rows, err := s.db.QueryContext(ctx, acquireLeaseCandidatesSQL, maxAttempts, batchSize)
	if err != nil {
		return nil, err
	}
	candidates := make([]OutboxRow, 0, batchSize)
	for rows.Next() {
		var row OutboxRow
		if err := rows.Scan(&row.ID, &row.Payload, &row.EnqueuedAt, &row.Attempts); err != nil {
			rows.Close()
			return nil, err
		}
		candidates = append(candidates, row)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}

	leaseMicros := leaseDuration.Microseconds()
	leases := make([]OutboxLease, 0, len(candidates))
	for _, candidate := range candidates {
		lease, acquired, err := s.acquireCandidate(ctx, holderID, leaseMicros, candidate, maxAttempts)
		if err != nil {
			return nil, err
		}
		if acquired {
			leases = append(leases, lease)
		}
	}
	return leases, nil
}

func (s *mysqlOutboxStore) acquireCandidate(ctx context.Context, holderID string, leaseMicros int64, candidate OutboxRow, maxAttempts int) (OutboxLease, bool, error) {
	if s.ownership == nil {
		result, err := s.db.ExecContext(ctx, acquireLeaseUpdateSQL,
			holderID, leaseMicros, candidate.ID, candidate.Attempts, maxAttempts)
		if err != nil {
			return OutboxLease{}, false, err
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return OutboxLease{}, false, err
		}
		if affected != 1 {
			return OutboxLease{}, false, nil
		}
		var expiresAt time.Time
		err = s.db.QueryRowContext(ctx, `SELECT LeaseExpiresAt
FROM _System_AuditOutbox
WHERE Id = ? AND ProcessedAt IS NULL AND PoisonClass IS NULL
  AND LeaseHolderId = ? AND Attempts = ?`, candidate.ID, holderID, candidate.Attempts).Scan(&expiresAt)
		if err != nil {
			return OutboxLease{}, false, err
		}
		return OutboxLease{Row: candidate, HolderID: holderID, LeaseExpiresAt: expiresAt}, true, nil
	}

	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return OutboxLease{}, false, err
	}
	defer func() { _ = cleanup() }()
	if err := s.lockActiveOwnership(ctx, tx); err != nil {
		_ = tx.Rollback()
		return OutboxLease{}, false, err
	}
	result, err := tx.ExecContext(ctx, acquireFencedLeaseUpdateSQL,
		holderID, leaseMicros, s.ownership.EpochDecimal(), candidate.ID, candidate.Attempts, maxAttempts)
	if err != nil {
		_ = tx.Rollback()
		return OutboxLease{}, false, err
	}
	affected, err := result.RowsAffected()
	if err != nil {
		_ = tx.Rollback()
		return OutboxLease{}, false, err
	}
	if affected != 1 {
		if err := tx.Rollback(); err != nil {
			return OutboxLease{}, false, err
		}
		return OutboxLease{}, false, nil
	}
	var expiresAt time.Time
	var epoch string
	err = tx.QueryRowContext(ctx, `SELECT LeaseExpiresAt, CAST(OwnershipEpoch AS CHAR)
FROM _System_AuditOutbox
WHERE Id = ? AND ProcessedAt IS NULL AND PoisonClass IS NULL
  AND LeaseHolderId = ? AND Attempts = ?`, candidate.ID, holderID, candidate.Attempts).Scan(&expiresAt, &epoch)
	if err != nil {
		_ = tx.Rollback()
		return OutboxLease{}, false, err
	}
	if epoch != s.ownership.EpochDecimal() {
		_ = tx.Rollback()
		return OutboxLease{}, false, persistence.RuntimeWorkOwnershipLost(errors.New("claimed row has a different ownership epoch"))
	}
	if err := tx.Commit(); err != nil {
		return OutboxLease{}, false, &OutboxOperationError{CommitState: OutboxCommitUnknown, Err: fmt.Errorf("commit audit lease claim: %w", err)}
	}
	return OutboxLease{
		Row: candidate, HolderID: holderID, LeaseExpiresAt: expiresAt,
		OwnershipEpoch: s.ownership.OwnershipEpoch,
	}, true, nil
}

const ownershipLockSQL = `SELECT LeaseExpiresAt
FROM _System_AuditOutbox
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND LeaseHolderId = ?
  AND Attempts = ?
  AND LeaseExpiresAt > CURRENT_TIMESTAMP(6)
FOR UPDATE`

const fencedOwnershipLockSQL = `SELECT LeaseExpiresAt, CAST(OwnershipEpoch AS CHAR)
FROM _System_AuditOutbox
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND LeaseHolderId = ?
  AND Attempts = ?
  AND OwnershipEpoch = ?
  AND LeaseExpiresAt > CURRENT_TIMESTAMP(6)
FOR UPDATE`

const insertFieldRowSQL = `INSERT INTO _System_AuditLog (
    Id, ObjectApiName, RecordId, FieldName, OldValue, NewValue,
    ChangedById, ChangedAt, EventAt, HistorySequence, FieldHistoryEligible,
    Entrypoint, Action, OutboxRowId, Outcome
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`

const readExactFieldRowSQL = `SELECT Id, ObjectApiName, RecordId, FieldName, OldValue, NewValue,
       ChangedById, ChangedAt, EventAt, HistorySequence, FieldHistoryEligible,
       Entrypoint, Action, OutboxRowId, Outcome
FROM _System_AuditLog
WHERE OutboxRowId = ? AND FieldName = ? AND ChangedAt = ?`

// CommitFanout makes the audit effect and durable acknowledgement one outcome.
// A previous release could leave exact field rows behind before consuming the
// outbox row; duplicate-key verification below deliberately permits that
// recovery shape, but every new successful attempt commits both facts together.
func (s *mysqlOutboxStore) CommitFanout(ctx context.Context, lease OutboxLease, rows []FieldRow) error {
	prepared, err := prepareFieldRows(lease, rows)
	if err != nil {
		return &OutboxOperationError{CommitState: OutboxCommitNoEffect, Deterministic: true, PoisonClass: PoisonClassParseFail, Err: err}
	}

	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return &OutboxOperationError{CommitState: OutboxCommitNoEffect, Err: fmt.Errorf("begin field fan-out: %w", err)}
	}
	defer func() { _ = cleanup() }()
	if err := s.lockActiveOwnership(ctx, tx); err != nil {
		return rollbackOperation(tx, err, false, PoisonClassNone)
	}
	if err := s.lockOutboxLease(ctx, tx, lease); err != nil {
		return rollbackOperation(tx, err, false, PoisonClassNone)
	}

	for _, row := range prepared {
		_, err := tx.ExecContext(ctx, insertFieldRowSQL, fieldRowArgs(row)...)
		if err == nil {
			continue
		}
		if isMySQLErrorNumber(err, 1062) {
			exact, readErr := exactFieldRowExists(ctx, tx, row)
			if readErr != nil {
				return rollbackOperation(tx, fmt.Errorf("read exact field replay: %w", readErr), false, PoisonClassNone)
			}
			if !exact {
				return rollbackOperation(tx, ErrOutboxNonExactReplay, true, PoisonClassParseFail)
			}
			continue
		}
		class := ClassifyError(err)
		deterministic := class == PoisonClassParseFail || class == PoisonClassFKViolation
		return rollbackOperation(tx, fmt.Errorf("insert audit field row: %w", err), deterministic, class)
	}
	finalizeSQL, finalizeArgs, err := s.finalizeLeaseStatement(lease)
	if err != nil {
		return rollbackOperation(tx, err, false, PoisonClassNone)
	}
	result, err := tx.ExecContext(ctx, finalizeSQL, finalizeArgs...)
	if err != nil {
		return rollbackOperation(tx, fmt.Errorf("consume audit outbox row: %w", err), false, PoisonClassNone)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return rollbackOperation(tx, fmt.Errorf("count consumed audit outbox row: %w", err), false, PoisonClassNone)
	}
	if affected != 1 {
		return rollbackOperation(tx, ErrOutboxLeaseLost, false, PoisonClassNone)
	}
	if err := tx.Commit(); err != nil {
		return &OutboxOperationError{CommitState: OutboxCommitUnknown, Err: fmt.Errorf("commit audit outbox fanout: %w", err)}
	}
	return nil
}

func (s *mysqlOutboxStore) lockOutboxLease(ctx context.Context, tx *sql.Tx, lease OutboxLease) error {
	var lockedExpiry time.Time
	if s.ownership == nil {
		err := tx.QueryRowContext(ctx, ownershipLockSQL,
			lease.Row.ID, lease.HolderID, lease.Row.Attempts).Scan(&lockedExpiry)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrOutboxLeaseLost
		}
		if err != nil {
			return fmt.Errorf("lock outbox ownership: %w", err)
		}
		return nil
	}
	if lease.OwnershipEpoch != s.ownership.OwnershipEpoch {
		return persistence.RuntimeWorkOwnershipLost(errors.New("lease epoch does not match configured authority"))
	}
	var epoch string
	err := tx.QueryRowContext(ctx, fencedOwnershipLockSQL,
		lease.Row.ID, lease.HolderID, lease.Row.Attempts, s.ownership.EpochDecimal()).Scan(&lockedExpiry, &epoch)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrOutboxLeaseLost
	}
	if err != nil {
		return fmt.Errorf("lock fenced outbox ownership: %w", err)
	}
	if epoch != s.ownership.EpochDecimal() {
		return persistence.RuntimeWorkOwnershipLost(errors.New("outbox row epoch changed"))
	}
	return nil
}

func rollbackOperation(tx *sql.Tx, cause error, deterministic bool, class PoisonClass) error {
	if err := tx.Rollback(); err != nil {
		return &OutboxOperationError{
			CommitState: OutboxCommitUnknown,
			Err:         errors.Join(cause, fmt.Errorf("rollback audit outbox fanout: %w", err)),
		}
	}
	return &OutboxOperationError{
		CommitState:   OutboxCommitRolledBack,
		Deterministic: deterministic,
		PoisonClass:   class,
		Err:           cause,
	}
}

type preparedFieldRow struct {
	FieldRow
	ID string
}

func prepareFieldRows(lease OutboxLease, rows []FieldRow) ([]preparedFieldRow, error) {
	seen := make(map[string]struct{}, len(rows))
	prepared := make([]preparedFieldRow, 0, len(rows))
	for i, row := range rows {
		if row.OutboxRowID != lease.Row.ID {
			return nil, fmt.Errorf("%w: row %d outbox ID %d does not match lease %d", ErrOutboxInvalidFieldRows, i, row.OutboxRowID, lease.Row.ID)
		}
		if strings.TrimSpace(row.FieldName) == "" || strings.TrimSpace(string(row.Outcome)) == "" {
			return nil, fmt.Errorf("%w: row %d is missing FieldName or Outcome", ErrOutboxInvalidFieldRows, i)
		}
		row.ChangedAt = normalizeChangedAt(row.ChangedAt)
		row.EventAt = normalizeEventAt(row.EventAt)
		key := semanticFieldKey(row.OutboxRowID, row.FieldName, row.ChangedAt)
		if _, exists := seen[key]; exists {
			return nil, fmt.Errorf("%w: duplicate semantic key at row %d", ErrOutboxInvalidFieldRows, i)
		}
		seen[key] = struct{}{}
		prepared = append(prepared, preparedFieldRow{FieldRow: row, ID: stableFieldRowID(row)})
	}
	return prepared, nil
}

func semanticFieldKey(outboxID int64, fieldName string, changedAt time.Time) string {
	return fmt.Sprintf("%d\x00%s\x00%s", outboxID, fieldName, normalizeChangedAt(changedAt).Format(time.RFC3339))
}

func stableFieldRowID(row FieldRow) string {
	h := sha256.New()
	h.Write([]byte("nexuscrm-audit-field-row:v1\x00"))
	var id [8]byte
	binary.BigEndian.PutUint64(id[:], uint64(row.OutboxRowID))
	h.Write(id[:])
	writeCanonicalPart(h, row.FieldName)
	writeCanonicalPart(h, normalizeChangedAt(row.ChangedAt).Format(time.RFC3339))
	return fmt.Sprintf("audf_%x", h.Sum(nil))
}

type canonicalWriter interface {
	Write([]byte) (int, error)
}

func writeCanonicalPart(w canonicalWriter, value string) {
	var size [8]byte
	binary.BigEndian.PutUint64(size[:], uint64(len(value)))
	_, _ = w.Write(size[:])
	_, _ = w.Write([]byte(value))
}

func normalizeChangedAt(value time.Time) time.Time {
	return value.UTC().Truncate(time.Second)
}

func normalizeEventAt(value time.Time) time.Time {
	if value.IsZero() {
		return time.Time{}
	}
	return value.UTC().Truncate(time.Microsecond)
}

func fieldRowArgs(row preparedFieldRow) []interface{} {
	var eventAt interface{}
	if !row.EventAt.IsZero() {
		eventAt = row.EventAt
	}
	var sequence interface{}
	if row.HistorySequence != nil {
		sequence = *row.HistorySequence
	}
	var fieldHistoryEligible interface{}
	if row.FieldHistoryEligible != nil {
		fieldHistoryEligible = *row.FieldHistoryEligible
	}
	return []interface{}{
		row.ID, row.ObjectApiName, row.RecordId, row.FieldName,
		row.OldValue, row.NewValue, row.ChangedById, row.ChangedAt,
		eventAt, sequence, fieldHistoryEligible, string(row.Entrypoint), string(row.Action),
		row.OutboxRowID, string(row.Outcome),
	}
}

type exactFieldRow struct {
	ID                   string
	ObjectApiName        string
	RecordID             string
	FieldName            string
	OldValue             sql.NullString
	NewValue             sql.NullString
	ChangedByID          string
	ChangedAt            time.Time
	EventAt              sql.NullTime
	HistorySequence      sql.NullInt64
	FieldHistoryEligible sql.NullBool
	Entrypoint           sql.NullString
	Action               sql.NullString
	OutboxRowID          sql.NullInt64
	Outcome              sql.NullString
}

func exactFieldRowExists(ctx context.Context, tx *sql.Tx, expected preparedFieldRow) (bool, error) {
	var got exactFieldRow
	err := tx.QueryRowContext(ctx, readExactFieldRowSQL,
		expected.OutboxRowID, expected.FieldName, expected.ChangedAt).Scan(
		&got.ID, &got.ObjectApiName, &got.RecordID, &got.FieldName,
		&got.OldValue, &got.NewValue, &got.ChangedByID, &got.ChangedAt,
		&got.EventAt, &got.HistorySequence, &got.FieldHistoryEligible, &got.Entrypoint, &got.Action,
		&got.OutboxRowID, &got.Outcome,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return exactFieldRowMatches(got, expected), nil
}

func exactFieldRowMatches(got exactFieldRow, expected preparedFieldRow) bool {
	return got.ID == expected.ID && got.ObjectApiName == expected.ObjectApiName &&
		got.RecordID == expected.RecordId && got.FieldName == expected.FieldName &&
		exactString(got.OldValue, expected.OldValue) && exactString(got.NewValue, expected.NewValue) &&
		got.ChangedByID == expected.ChangedById &&
		normalizeChangedAt(got.ChangedAt).Equal(expected.ChangedAt) &&
		exactEventAt(got.EventAt, expected.EventAt) &&
		exactSequence(got.HistorySequence, expected.HistorySequence) &&
		exactBool(got.FieldHistoryEligible, expected.FieldHistoryEligible) &&
		exactString(got.Entrypoint, string(expected.Entrypoint)) &&
		exactString(got.Action, string(expected.Action)) &&
		got.OutboxRowID.Valid && got.OutboxRowID.Int64 == expected.OutboxRowID &&
		exactString(got.Outcome, string(expected.Outcome))
}

func exactString(got sql.NullString, expected string) bool {
	return got.Valid && got.String == expected
}

func exactEventAt(got sql.NullTime, expected time.Time) bool {
	if expected.IsZero() {
		return !got.Valid
	}
	return got.Valid && normalizeEventAt(got.Time).Equal(expected)
}

func exactSequence(got sql.NullInt64, expected *int64) bool {
	if expected == nil {
		return !got.Valid
	}
	return got.Valid && got.Int64 == *expected
}

func exactBool(got sql.NullBool, expected *bool) bool {
	if expected == nil {
		return !got.Valid
	}
	return got.Valid && got.Bool == *expected
}

const finalizeSuccessSQL = `DELETE FROM _System_AuditOutbox
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND LeaseHolderId = ?
  AND Attempts = ?
  AND LeaseExpiresAt > CURRENT_TIMESTAMP(6)`

const finalizeFencedSuccessSQL = `DELETE FROM _System_AuditOutbox
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND LeaseHolderId = ?
  AND Attempts = ?
  AND OwnershipEpoch = ?
  AND LeaseExpiresAt > CURRENT_TIMESTAMP(6)`

func (s *mysqlOutboxStore) finalizeLeaseStatement(lease OutboxLease) (string, []interface{}, error) {
	if s.ownership == nil {
		return finalizeSuccessSQL, []interface{}{lease.Row.ID, lease.HolderID, lease.Row.Attempts}, nil
	}
	if lease.OwnershipEpoch != s.ownership.OwnershipEpoch {
		return "", nil, persistence.RuntimeWorkOwnershipLost(errors.New("lease epoch does not match configured authority"))
	}
	return finalizeFencedSuccessSQL, []interface{}{
		lease.Row.ID, lease.HolderID, lease.Row.Attempts, s.ownership.EpochDecimal(),
	}, nil
}

const bumpAttemptsAfterRollbackSQL = `UPDATE _System_AuditOutbox
SET PoisonClass = CASE WHEN Attempts + 1 >= ? THEN 'RETRY_EXHAUSTED' ELSE NULL END,
    Attempts = Attempts + 1,
    LastError = ?,
    LeaseHolderId = NULL,
    LeaseExpiresAt = NULL,
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND LeaseHolderId = ?
  AND Attempts = ?`

const bumpAttemptsAfterRollbackFencedSQL = `UPDATE _System_AuditOutbox
SET PoisonClass = CASE WHEN Attempts + 1 >= ? THEN 'RETRY_EXHAUSTED' ELSE NULL END,
    Attempts = Attempts + 1,
    LastError = ?,
    LeaseHolderId = NULL,
    LeaseExpiresAt = NULL,
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND LeaseHolderId = ?
  AND Attempts = ?
  AND OwnershipEpoch = ?`

func (s *mysqlOutboxStore) BumpAttemptsAfterRollback(ctx context.Context, lease OutboxLease, maxAttempts int, lastErr string) (bool, error) {
	if s.ownership == nil {
		result, err := s.db.ExecContext(ctx, bumpAttemptsAfterRollbackSQL,
			maxAttempts, lastErr, lease.Row.ID, lease.HolderID, lease.Row.Attempts)
		if err != nil {
			return false, &OutboxOperationError{CommitState: OutboxCommitUnknown, Err: err}
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return false, &OutboxOperationError{CommitState: OutboxCommitUnknown, Err: err}
		}
		if affected != 1 {
			return false, ErrOutboxLeaseLost
		}
		return lease.Row.Attempts+1 >= maxAttempts, nil
	}
	if lease.OwnershipEpoch != s.ownership.OwnershipEpoch {
		return false, persistence.RuntimeWorkOwnershipLost(errors.New("lease epoch does not match configured authority"))
	}
	err := s.executeFencedStateUpdate(ctx, bumpAttemptsAfterRollbackFencedSQL,
		maxAttempts, lastErr, lease.Row.ID, lease.HolderID, lease.Row.Attempts, s.ownership.EpochDecimal())
	if err != nil {
		return false, err
	}
	return lease.Row.Attempts+1 >= maxAttempts, nil
}

const markPoisonAfterRollbackSQL = `UPDATE _System_AuditOutbox
SET PoisonClass = ?,
    LastError = ?,
    LeaseHolderId = NULL,
    LeaseExpiresAt = NULL,
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND LeaseHolderId = ?
  AND Attempts = ?`

const markPoisonAfterRollbackFencedSQL = `UPDATE _System_AuditOutbox
SET PoisonClass = ?,
    LastError = ?,
    LeaseHolderId = NULL,
    LeaseExpiresAt = NULL,
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass IS NULL
  AND LeaseHolderId = ?
  AND Attempts = ?
  AND OwnershipEpoch = ?`

func (s *mysqlOutboxStore) MarkPoisonAfterRollback(ctx context.Context, lease OutboxLease, class PoisonClass, lastErr string) error {
	if class != PoisonClassParseFail && class != PoisonClassFKViolation {
		return fmt.Errorf("audit outbox: invalid deterministic poison class %q", class)
	}
	if s.ownership == nil {
		result, err := s.db.ExecContext(ctx, markPoisonAfterRollbackSQL,
			string(class), lastErr, lease.Row.ID, lease.HolderID, lease.Row.Attempts)
		if err != nil {
			return &OutboxOperationError{CommitState: OutboxCommitUnknown, Err: err}
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return &OutboxOperationError{CommitState: OutboxCommitUnknown, Err: err}
		}
		if affected != 1 {
			return ErrOutboxLeaseLost
		}
		return nil
	}
	if lease.OwnershipEpoch != s.ownership.OwnershipEpoch {
		return persistence.RuntimeWorkOwnershipLost(errors.New("lease epoch does not match configured authority"))
	}
	return s.executeFencedStateUpdate(ctx, markPoisonAfterRollbackFencedSQL,
		string(class), lastErr, lease.Row.ID, lease.HolderID, lease.Row.Attempts, s.ownership.EpochDecimal())
}

const repairPoisonSQL = `UPDATE _System_AuditOutbox
SET PoisonClass = NULL,
    Attempts = 0,
    LastError = NULL,
    LeaseHolderId = NULL,
    LeaseExpiresAt = NULL,
    OwnershipEpoch = ?,
    RepairCount = RepairCount + 1,
    LastRepairId = ?,
    LastRepairReason = ?,
    LastRepairedAt = CURRENT_TIMESTAMP(6),
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?
  AND ProcessedAt IS NULL
  AND PoisonClass = ?
  AND Attempts = ?
  AND LeaseHolderId IS NULL`

func (s *mysqlOutboxStore) RepairPoison(ctx context.Context, repair OutboxRepair) error {
	if s == nil || s.db == nil {
		return errors.New("audit outbox: nil database")
	}
	if s.ownership == nil {
		return errors.New("audit outbox: repair requires fenced work ownership")
	}
	if repair.OutboxRowID <= 0 || repair.ExpectedAttempts < 0 ||
		(repair.ExpectedClass != PoisonClassParseFail && repair.ExpectedClass != PoisonClassFKViolation && repair.ExpectedClass != PoisonClassRetryExhausted) ||
		!persistence.ValidRuntimeOperationID(repair.RepairID) || strings.TrimSpace(repair.Reason) == "" || len(repair.Reason) > lastErrorMaxLen {
		return errors.New("audit outbox: invalid repair request")
	}
	return s.executeFencedStateUpdate(ctx, repairPoisonSQL,
		s.ownership.EpochDecimal(), repair.RepairID, sanitizeRepairReason(repair.Reason), repair.OutboxRowID,
		string(repair.ExpectedClass), repair.ExpectedAttempts)
}

func sanitizeRepairReason(reason string) string {
	reason = strings.Map(func(r rune) rune {
		if r < 0x20 || r == 0x7f {
			return -1
		}
		return r
	}, reason)
	return strings.TrimSpace(reason)
}

func (s *mysqlOutboxStore) executeFencedStateUpdate(ctx context.Context, statement string, args ...any) error {
	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return &OutboxOperationError{CommitState: OutboxCommitNoEffect, Err: err}
	}
	defer func() { _ = cleanup() }()
	if err := s.lockActiveOwnership(ctx, tx); err != nil {
		_ = tx.Rollback()
		return err
	}
	result, err := tx.ExecContext(ctx, statement, args...)
	if err != nil {
		return rollbackOperation(tx, err, false, PoisonClassNone)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return rollbackOperation(tx, err, false, PoisonClassNone)
	}
	if affected != 1 {
		if err := tx.Rollback(); err != nil {
			return &OutboxOperationError{CommitState: OutboxCommitUnknown, Err: errors.Join(ErrOutboxLeaseLost, err)}
		}
		return ErrOutboxLeaseLost
	}
	if err := tx.Commit(); err != nil {
		return &OutboxOperationError{CommitState: OutboxCommitUnknown, Err: err}
	}
	return nil
}

func isMySQLErrorNumber(err error, number uint16) bool {
	var mysqlErr *mysql.MySQLError
	return errors.As(err, &mysqlErr) && mysqlErr.Number == number
}