package persistence

import (
	"context"
	"database/sql"
	stderrors "errors"
	"fmt"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/db"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
)

// txContextKey is the key for storing transaction in context
type txContextKey struct{}

// pessimisticTxMarkerKey is the key for marking that the context is inside
// a pessimistic transaction. A context is safe for blocking SELECT ... FOR
// UPDATE only when it carries both the SQL transaction (txContextKey) and
// this marker. DetachTx strips both so that a detached context correctly
// signals "no active pessimistic transaction".
type pessimisticTxMarkerKey struct{}

// pessimisticTxOwnerKey binds a pessimistic transaction context to the exact
// TransactionManager that opened it. The marker alone proves transaction mode,
// but not database or manager ownership.
type pessimisticTxOwnerKey struct{}

// TransactionManager handles database transactions with retry logic for deadlocks
type TransactionManager struct {
	db                      *database.TiDBConnection
	pessimisticLockWaitSecs int
}

// NewTransactionManager creates a new TransactionManager
func NewTransactionManager(db *database.TiDBConnection) *TransactionManager {
	return &TransactionManager{db: db}
}

// SetPessimisticLockWaitTimeout configures the per-session innodb_lock_wait_timeout
// (in seconds) applied to every pessimistic transaction opened via
// WithPessimisticTransaction. A value of 0 (the default) leaves the session
// variable at its server default. Positive values are clamped to [1, 3600] by
// BeginPessimisticTx.
func (tm *TransactionManager) SetPessimisticLockWaitTimeout(secs int) {
	tm.pessimisticLockWaitSecs = secs
}

// WithTransaction executes a function within a database transaction.
// The transaction is automatically rolled back if the function returns an error or panics.
// The transaction is committed if the function returns nil.
// It injects the transaction into the context for downstream consumers.
func (tm *TransactionManager) WithTransaction(ctx context.Context, fn func(ctx context.Context) error) error {
	ctx = withTransactionTimingID(ctx)
	beginStart := time.Now()
	tx, err := tm.db.BeginTx(ctx, nil)
	logTransactionStageTiming(ctx, "WithTransaction", "begin", beginStart, err)
	if err != nil {
		return apperrors.NewInternalError("failed to begin transaction", err)
	}

	var txCtx context.Context
	// Ensure rollback on panic
	defer func() {
		if p := recover(); p != nil {
			_ = tx.Rollback()
			if txCtx != nil {
				ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRolledBack)
			}
			panic(p) // Re-throw panic after rollback
		}
	}()

	// Inject transaction into context
	txCtx = tm.InjectTx(ctx, tx)
	txCtx = ports.WithAuthorizationGrantTransactionScope(txCtx)
	txCtx = recordlocks.WithoutScope(txCtx)
	txCtx = ports.WithTransactionAbort(txCtx)
	txCtx = ports.WithAfterCommitHooks(txCtx)
	txCtx = ports.WithTransactionCompletionHooks(txCtx)
	txCtx = ports.WithHistoryTrackingAttemptState(txCtx)
	txCtx = ports.WithHistorySerializationAttemptState(txCtx)
	txCtx = withPermissionStateAttemptState(txCtx)
	// Clear any inherited pessimistic marker — this is a plain transaction,
	// not a pessimistic one. Without this, a plain transaction opened under
	// a pessimistic parent would incorrectly appear pessimistic, causing
	// nested WithPessimisticTransaction to skip BeginPessimisticTx and
	// lose FOR UPDATE blocking semantics.
	txCtx = context.WithValue(txCtx, pessimisticTxMarkerKey{}, false)
	txCtx = context.WithValue(txCtx, pessimisticTxOwnerKey{}, (*TransactionManager)(nil))

	// Execute the function
	fnStart := time.Now()
	if err := fn(txCtx); err != nil {
		logTransactionStageTiming(txCtx, "WithTransaction", "fn", fnStart, err)
		returnErr := err
		if abortErr := ports.TransactionAbortErr(txCtx); abortErr != nil {
			returnErr = abortErr
		}
		rollbackStart := time.Now()
		if rbErr := tx.Rollback(); rbErr != nil {
			logTransactionStageTiming(txCtx, "WithTransaction", "rollback", rollbackStart, rbErr)
			ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRollbackError)
			return apperrors.NewInternalError(fmt.Sprintf("transaction failed (rollback error: %v)", rbErr), returnErr)
		}
		logTransactionStageTiming(txCtx, "WithTransaction", "rollback", rollbackStart, nil)
		ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRolledBack)
		return returnErr
	}
	logTransactionStageTiming(txCtx, "WithTransaction", "fn", fnStart, nil)

	permissionStateStart := time.Now()
	if permissionStateErr := tm.advanceRequestedPermissionState(txCtx); permissionStateErr != nil {
		logTransactionStageTiming(txCtx, "WithTransaction", "permissionState", permissionStateStart, permissionStateErr)
		if ports.TransactionAbortErr(txCtx) == nil {
			ports.AbortTransaction(txCtx, permissionStateErr)
		}
	} else {
		logTransactionStageTiming(txCtx, "WithTransaction", "permissionState", permissionStateStart, nil)
	}
	if abortErr := ports.PrepareTransactionCommit(txCtx); abortErr != nil {
		rollbackStart := time.Now()
		if rbErr := tx.Rollback(); rbErr != nil {
			logTransactionStageTiming(txCtx, "WithTransaction", "rollback", rollbackStart, rbErr)
			ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRollbackError)
			return apperrors.NewInternalError(fmt.Sprintf("transaction aborted (rollback error: %v)", rbErr), abortErr)
		}
		logTransactionStageTiming(txCtx, "WithTransaction", "rollback", rollbackStart, nil)
		ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRolledBack)
		return abortErr
	}

	// Commit the transaction
	commitStart := time.Now()
	if err := tx.Commit(); err != nil {
		logTransactionStageTiming(txCtx, "WithTransaction", "commit", commitStart, err)
		ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionCommitError)
		if ports.IsHistorySerializationAttempt(txCtx) && isOptimisticWriteConflict(err) {
			return fmt.Errorf("%w: %v", ports.ErrHistorySerializationConflict, err)
		}
		if ports.IsHistoryTrackingAttempt(txCtx) && isOptimisticWriteConflict(err) {
			return fmt.Errorf("%w: %v", ports.ErrHistoryTrackingConflict, err)
		}
		return apperrors.NewInternalError("failed to commit transaction", err)
	}
	logTransactionStageTiming(txCtx, "WithTransaction", "commit", commitStart, nil)
	completionStart := time.Now()
	ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionCommitted)
	logTransactionStageTiming(txCtx, "WithTransaction", "completionHooks", completionStart, nil)
	afterCommitStart := time.Now()
	ports.RunAfterCommitHooks(txCtx)
	logTransactionStageTiming(txCtx, "WithTransaction", "afterCommitHooks", afterCommitStart, nil)

	return nil
}

// WithPessimisticTransaction executes fn inside a TiDB pessimistic-mode
// transaction. It acquires a dedicated conn via BeginPessimisticTx so that
// FOR UPDATE locks inside fn actually block rather than returning a
// snapshot. The cleanup runs after commit/rollback to reset the session
// var before the conn re-enters the pool.
//
// If the context already carries a pessimistic transaction owned by this exact
// TransactionManager, fn is executed directly without opening a second
// transaction. A foreign manager's context is rejected rather than reused.
func isOptimisticWriteConflict(err error) bool {
	if err == nil {
		return false
	}
	var mysqlErr *mysql.MySQLError
	if stderrors.As(err, &mysqlErr) {
		switch mysqlErr.Number {
		case 8028, 9007:
			return true
		}
	}
	message := strings.ToLower(err.Error())
	return strings.Contains(message, "write conflict") || strings.Contains(message, "txnlocknotfound")
}

func (tm *TransactionManager) WithPessimisticTransaction(ctx context.Context, fn func(ctx context.Context) error) error {
	// Nested pessimistic transaction: reuse the outer one.
	if tm.ownsPessimisticTx(ctx) {
		fnStart := time.Now()
		err := fn(ctx)
		logTransactionStageTiming(ctx, "WithPessimisticTransaction", "nested.fn", fnStart, err)
		return err
	}
	if tm.IsPessimisticTx(ctx) {
		return apperrors.NewInternalError(
			"WithPessimisticTransaction called with a pessimistic transaction owned by another TransactionManager",
			nil,
		)
	}

	// Fail closed: if a plain transaction is active, we cannot safely
	// open an independent pessimistic transaction (atomicity break —
	// inner work could commit while outer tx rolls back). The caller
	// must either use WithPessimisticTransaction at the outermost scope
	// or DetachTx first.
	if tm.IsTransactionActive(ctx) {
		return apperrors.NewInternalError(
			"WithPessimisticTransaction called inside a non-pessimistic transaction; "+
				"use WithPessimisticTransaction at the outermost scope or DetachTx first", nil)
	}
	ctx = withTransactionTimingID(ctx)

	if tm.pessimisticLockWaitSecs > 0 {
		ctx = withLockWaitTimeoutSecs(ctx, tm.pessimisticLockWaitSecs)
	}

	beginStart := time.Now()
	tx, cleanup, err := BeginPessimisticTx(ctx, tm.db.DB())
	logTransactionStageTiming(ctx, "WithPessimisticTransaction", "begin", beginStart, err)
	if err != nil {
		return apperrors.NewInternalError("failed to begin pessimistic transaction", err)
	}
	var txCtx context.Context
	defer func() {
		if p := recover(); p != nil {
			_ = tx.Rollback()
			_ = cleanup()
			if txCtx != nil {
				ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRolledBack)
			}
			panic(p)
		}
	}()

	// Inject both the SQL transaction and the pessimistic marker.
	txCtx = tm.InjectTx(ctx, tx)
	txCtx = ports.WithAuthorizationGrantTransactionScope(txCtx)
	txCtx = recordlocks.WithoutScope(txCtx)
	txCtx = ports.WithTransactionAbort(txCtx)
	txCtx = ports.WithAfterCommitHooks(txCtx)
	txCtx = ports.WithTransactionCompletionHooks(txCtx)
	txCtx = ports.WithHistoryTrackingAttemptState(txCtx)
	txCtx = ports.WithHistorySerializationAttemptState(txCtx)
	txCtx = withPermissionStateAttemptState(txCtx)
	txCtx = context.WithValue(txCtx, pessimisticTxMarkerKey{}, true)
	txCtx = context.WithValue(txCtx, pessimisticTxOwnerKey{}, tm)
	if readCommittedIsolationFromContext(ctx) {
		txCtx = context.WithValue(txCtx, readCommittedPessimisticTxMarkerKey{}, true)
	}

	fnStart := time.Now()
	if err := fn(txCtx); err != nil {
		logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "fn", fnStart, err)
		returnErr := err
		if abortErr := ports.TransactionAbortErr(txCtx); abortErr != nil {
			returnErr = abortErr
		}
		rollbackStart := time.Now()
		if rbErr := tx.Rollback(); rbErr != nil {
			logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "rollback", rollbackStart, rbErr)
			cleanupStart := time.Now()
			cleanupErr := cleanup()
			logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "cleanup", cleanupStart, cleanupErr)
			ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRollbackError)
			return apperrors.NewInternalError(fmt.Sprintf("pessimistic transaction failed (rollback error: %v)", rbErr), returnErr)
		}
		logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "rollback", rollbackStart, nil)
		cleanupStart := time.Now()
		cleanupErr := cleanup()
		logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "cleanup", cleanupStart, cleanupErr)
		ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRolledBack)
		return returnErr
	}
	logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "fn", fnStart, nil)

	permissionStateStart := time.Now()
	if permissionStateErr := tm.advanceRequestedPermissionState(txCtx); permissionStateErr != nil {
		logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "permissionState", permissionStateStart, permissionStateErr)
		if ports.TransactionAbortErr(txCtx) == nil {
			ports.AbortTransaction(txCtx, permissionStateErr)
		}
	} else {
		logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "permissionState", permissionStateStart, nil)
	}
	if abortErr := ports.PrepareTransactionCommit(txCtx); abortErr != nil {
		rollbackStart := time.Now()
		if rbErr := tx.Rollback(); rbErr != nil {
			logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "rollback", rollbackStart, rbErr)
			cleanupStart := time.Now()
			cleanupErr := cleanup()
			logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "cleanup", cleanupStart, cleanupErr)
			ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRollbackError)
			return apperrors.NewInternalError(fmt.Sprintf("pessimistic transaction aborted (rollback error: %v)", rbErr), abortErr)
		}
		logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "rollback", rollbackStart, nil)
		cleanupStart := time.Now()
		cleanupErr := cleanup()
		logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "cleanup", cleanupStart, cleanupErr)
		ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRolledBack)
		return abortErr
	}

	commitStart := time.Now()
	if err := tx.Commit(); err != nil {
		logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "commit", commitStart, err)
		cleanupStart := time.Now()
		cleanupErr := cleanup()
		logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "cleanup", cleanupStart, cleanupErr)
		ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionCommitError)
		if ports.IsHistorySerializationAttempt(txCtx) && isOptimisticWriteConflict(err) {
			return fmt.Errorf("%w: %v", ports.ErrHistorySerializationConflict, err)
		}
		if ports.IsHistoryTrackingAttempt(txCtx) && isOptimisticWriteConflict(err) {
			return fmt.Errorf("%w: %v", ports.ErrHistoryTrackingConflict, err)
		}
		return apperrors.NewInternalError("failed to commit pessimistic transaction", err)
	}
	logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "commit", commitStart, nil)

	cleanupStart := time.Now()
	cleanupErr := cleanup()
	logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "cleanup", cleanupStart, cleanupErr)
	completionStart := time.Now()
	ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionCommitted)
	logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "completionHooks", completionStart, nil)
	afterCommitStart := time.Now()
	ports.RunAfterCommitHooks(txCtx)
	logTransactionStageTiming(txCtx, "WithPessimisticTransaction", "afterCommitHooks", afterCommitStart, nil)
	return nil
}

// WithCurrentReadPermissionAuthoringTransaction owns the isolation contract
// required by one Permission Set authoring command. TiDB's default
// REPEATABLE READ makes plain effective-permission reads observe the
// transaction-start snapshot even after a locking Permission State read.
// READ COMMITTED plus the command's root and Permission State locks preserves
// serialization while making every dependent projection transaction-current.
//
// Nesting inside an existing pessimistic transaction is accepted only when
// that transaction already carries the same isolation guarantee. Silently
// reusing a REPEATABLE READ transaction would reintroduce stale authorization.
func (tm *TransactionManager) WithCurrentReadPermissionAuthoringTransaction(
	ctx context.Context,
	fn func(context.Context) error,
) error {
	if tm.ownsPessimisticTx(ctx) {
		currentRead, _ := ctx.Value(readCommittedPessimisticTxMarkerKey{}).(bool)
		if !currentRead {
			return apperrors.NewInternalError(
				"permission authoring requires a READ COMMITTED pessimistic transaction",
				nil,
			)
		}
		return fn(ctx)
	}
	return tm.WithPessimisticTransaction(WithReadCommittedIsolation(ctx), fn)
}

// WithRetry executes a function within a transaction with automatic retry on
// lock contention, typed history-sequence serialization conflicts, and exact
// unique-field upsert races. A growing history prelock plan converges
// independently from the bounded contention budget. Commit or rollback outcome
// ambiguity fails closed because replay is no longer safe.
func (tm *TransactionManager) WithRetry(ctx context.Context, fn func(ctx context.Context) error, maxRetries int) error {
	if maxRetries < 1 {
		maxRetries = 1
	}

	ctx = ports.EnsureHistorySequenceRetryPlan(ctx)
	var lastErr error
	forcedHistoryPrelocks := ports.HistorySequenceRetryPlanFields(ctx)
	retryCtx := ports.WithHistorySequencePrelockRetryOwner(ctx)
	transactionAttempts := 0
	retryableAttempts := 0
	for {
		transactionAttempts++
		attemptStart := time.Now()
		var completion ports.TransactionCompletionOutcome
		attemptCtx := ports.WithPlannedHistorySequenceLocks(retryCtx, forcedHistoryPrelocks)
		err := tm.WithTransaction(attemptCtx, func(txCtx context.Context) error {
			// Retrying is safe only after a proven rollback. A commit error may
			// mean COMMIT applied but its acknowledgement was lost, while a
			// rollback error leaves the attempt's outcome unknown.
			ports.RegisterTransactionCompletionHook(txCtx, func(outcome ports.TransactionCompletionOutcome) {
				completion = outcome
			})
			return fn(txCtx)
		})
		logTransactionStageTiming(ctx, "WithRetry", fmt.Sprintf("attempt.%d", transactionAttempts), attemptStart, err)
		if err == nil {
			return nil // Success
		}

		lastErr = err
		if completion == ports.TransactionCompletionCommitError ||
			completion == ports.TransactionCompletionRollbackError {
			return err
		}
		if prelocks, ok := ports.HistorySequencePrelockFields(err); ok {
			if ctxErr := ctx.Err(); ctxErr != nil {
				return ctxErr
			}
			merged, changed := ports.AugmentHistorySequenceRetryPlan(ctx, prelocks)
			if !changed {
				return err
			}
			forcedHistoryPrelocks = merged
			continue
		}

		if !IsLockContention(err) &&
			!IsHistorySerializationConflict(err) &&
			!ports.IsUniqueFieldUpsertRace(err) {
			return err
		}

		retryableAttempts++
		if retryableAttempts >= maxRetries {
			break
		}

		// Retryable contention detected; keep the existing backoff schedule.
		backoff := time.Millisecond * time.Duration(100*(1<<uint(retryableAttempts-1)))
		backoffStart := time.Now()
		time.Sleep(backoff)
		logTransactionStageTiming(ctx, "WithRetry", fmt.Sprintf("backoff.%d", retryableAttempts), backoffStart, nil)
	}

	if ports.IsUniqueFieldUpsertRace(lastErr) {
		return apperrors.NewLockContentionError(lastErr)
	}
	return apperrors.NewInternalError(fmt.Sprintf("transaction failed after %d retries", maxRetries), lastErr)
}

// InjectTx injects a transaction into the context
func (tm *TransactionManager) InjectTx(ctx context.Context, tx *sql.Tx) context.Context {
	return context.WithValue(ctx, txContextKey{}, tx)
}

// DetachTx returns a context with no active transaction and no pessimistic
// marker. The returned context will start a fresh, independent transaction
// when passed to WithTransaction / RunInTransaction. Any existing
// transaction value is overwritten with nil so that IsTransactionActive
// returns false and RunInTransaction uses WithRetry to begin a new
// independent transaction. The pessimistic marker is also cleared so that
// IsPessimisticTx returns false and RunInPessimisticTransaction opens a
// fresh pessimistic transaction. Any inherited after-commit/completion hooks
// and abort controller are also stripped; the fresh transaction will get its
// own collectors when opened.
//
// This is used by automation steps (e.g. AfterSave flows) that need to write
// to the same row the enclosing outbox transaction already holds a lock on —
// passing the original txCtx would cause a TiDB lock-wait timeout (Error 1205).
func (tm *TransactionManager) DetachTx(ctx context.Context) context.Context {
	// Store a typed nil. IsTransactionActive / ExtractTx both guard against
	// nil *sql.Tx so this correctly signals "no active transaction".
	detached := context.WithValue(ctx, txContextKey{}, (*sql.Tx)(nil))
	detached = ports.WithoutAuthorizationGrantTransactionScope(detached)
	detached = recordlocks.WithoutScope(detached)
	// Strip the pessimistic marker as well — a detached context must not
	// appear to be inside a pessimistic transaction.
	detached = context.WithValue(detached, pessimisticTxMarkerKey{}, false)
	detached = context.WithValue(detached, pessimisticTxOwnerKey{}, (*TransactionManager)(nil))
	detached = ports.WithoutAfterCommitHooks(detached)
	detached = ports.WithoutTransactionCompletionHooks(detached)
	detached = ports.WithoutTransactionAbort(detached)
	detached = ports.WithoutHistorySequencePrelockRetryOwner(detached)
	detached = ports.WithoutInitialHistorySequenceLocks(detached)
	detached = ports.WithoutInitialHistoryBusinessRowObjects(detached)
	detached = ports.WithoutPlannedHistorySequenceLocks(detached)
	detached = ports.WithoutHistorySequenceRetryPlan(detached)
	return detached
}

// ExtractTx extracts a transaction from the context.
// Returns nil (untyped) when no transaction is active or when the stored
// transaction pointer is nil (e.g. after DetachTx).
func (tm *TransactionManager) ExtractTx(ctx context.Context) interface{} {
	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); ok && tx != nil {
		return tx
	}
	return nil
}

// ReadTransactionTime returns one DB timestamp through the active transaction.
// Callers reuse it for every row belonging to one transactional event.
func (tm *TransactionManager) ReadTransactionTime(ctx context.Context) (time.Time, error) {
	tx, ok := tm.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return time.Time{}, apperrors.NewInternalError("transaction time requires an active transaction", nil)
	}
	var eventAt time.Time
	if err := tx.QueryRowContext(ctx, "SELECT UTC_TIMESTAMP(6)").Scan(&eventAt); err != nil {
		return time.Time{}, err
	}
	return eventAt, nil
}

// IsTransactionActive checks if a transaction is currently active in the context
func (tm *TransactionManager) IsTransactionActive(ctx context.Context) bool {
	return tm.ExtractTx(ctx) != nil
}

// IsPessimisticTx reports whether the context is inside a pessimistic
// transaction. A context is safe for blocking SELECT ... FOR UPDATE only
// when this returns true AND IsTransactionActive also returns true.
func (tm *TransactionManager) IsPessimisticTx(ctx context.Context) bool {
	marker, ok := ctx.Value(pessimisticTxMarkerKey{}).(bool)
	return ok && marker
}

func (tm *TransactionManager) ownsPessimisticTx(ctx context.Context) bool {
	owner, ok := ctx.Value(pessimisticTxOwnerKey{}).(*TransactionManager)
	return ok &&
		owner == tm &&
		tm.IsPessimisticTx(ctx) &&
		tm.IsTransactionActive(ctx)
}

// OwnsPessimisticTx reports whether ctx carries an active pessimistic SQL
// transaction opened by this exact manager. Transaction-bound security
// adapters use this stronger witness instead of trusting a mode marker from a
// foreign manager or a context without the matching SQL transaction.
func (tm *TransactionManager) OwnsPessimisticTx(ctx context.Context) bool {
	return tm != nil && tm.ownsPessimisticTx(ctx)
}

// WithPessimisticRetry executes fn inside a pessimistic transaction with
// automatic retry on lock contention (TiDB 1205/1213), exact unique-field
// upsert races, and convergent replay when field-history sequence locks are
// discovered late. If the context already carries a pessimistic transaction
// owned by this exact TransactionManager, fn is executed directly without
// opening a second transaction.
func (tm *TransactionManager) WithPessimisticRetry(ctx context.Context, fn func(ctx context.Context) error, maxRetries int) error {
	if maxRetries < 1 {
		maxRetries = 1
	}

	// Nested: reuse the active pessimistic transaction.
	if tm.ownsPessimisticTx(ctx) {
		return fn(ctx)
	}

	ctx = ports.EnsureHistorySequenceRetryPlan(ctx)
	var lastErr error
	forcedHistoryPrelocks := ports.HistorySequenceRetryPlanFields(ctx)
	retryCtx := ports.WithHistorySequencePrelockRetryOwner(ctx)
	transactionAttempts := 0
	retryableAttempts := 0
	for {
		transactionAttempts++
		attemptStart := time.Now()
		var completion ports.TransactionCompletionOutcome
		attemptCtx := ports.WithPlannedHistorySequenceLocks(retryCtx, forcedHistoryPrelocks)
		err := tm.WithPessimisticTransaction(attemptCtx, func(txCtx context.Context) error {
			// A commit error is outcome-ambiguous: the server may have applied
			// COMMIT and lost only the acknowledgement. A rollback error likewise
			// gives no positive rollback proof. Capture the transaction manager's
			// typed terminal outcome so neither case is ever replayed merely
			// because its wrapped database error resembles lock contention.
			ports.RegisterTransactionCompletionHook(txCtx, func(outcome ports.TransactionCompletionOutcome) {
				completion = outcome
			})
			return fn(txCtx)
		})
		logTransactionStageTiming(ctx, "WithPessimisticRetry", fmt.Sprintf("attempt.%d", transactionAttempts), attemptStart, err)
		if err == nil {
			return nil
		}

		lastErr = err
		if completion == ports.TransactionCompletionCommitError ||
			completion == ports.TransactionCompletionRollbackError {
			return err
		}
		if prelocks, ok := ports.HistorySequencePrelockFields(err); ok {
			if ctxErr := ctx.Err(); ctxErr != nil {
				return ctxErr
			}
			merged, changed := ports.AugmentHistorySequenceRetryPlan(ctx, prelocks)
			if !changed {
				return err
			}
			forcedHistoryPrelocks = merged
			continue
		}

		if !IsLockContention(err) &&
			!IsHistorySerializationConflict(err) &&
			!ports.IsUniqueFieldUpsertRace(err) {
			return err
		}

		retryableAttempts++
		if retryableAttempts >= maxRetries {
			break
		}

		// Retryable contention detected, retry with exponential backoff.
		backoff := time.Millisecond * time.Duration(100*(1<<uint(retryableAttempts-1)))
		backoffStart := time.Now()
		select {
		case <-time.After(backoff):
		case <-ctx.Done():
			return ctx.Err()
		}
		logTransactionStageTiming(ctx, "WithPessimisticRetry", fmt.Sprintf("backoff.%d", retryableAttempts), backoffStart, nil)
	}

	if IsLockContention(lastErr) || ports.IsUniqueFieldUpsertRace(lastErr) {
		return apperrors.NewLockContentionError(lastErr)
	}
	return apperrors.NewInternalError(fmt.Sprintf("pessimistic transaction failed after %d retries", maxRetries), lastErr)
}

// IsLockContention reports whether err represents retryable lock/contention
// evidence, including durable parent-lane admission contention and TiDB/MySQL
// row-lock contention (deadlock 1213 or lock-wait timeout 1205). Exported so
// retry loops and API metrics can share the same classifier.
func IsLockContention(err error) bool {
	return IsParentLaneContention(err) || IsDBRowLockContention(err)
}

// IsHistorySerializationConflict reports whether err is a typed conflict from
// a transaction that reserved field-history sequence authority.
func IsHistorySerializationConflict(err error) bool {
	return ports.IsHistorySerializationConflict(err)
}

// IsParentLaneContention reports whether err is (or wraps) the durable
// parent-lane contention sentinel. Parent-lane contention is deliberately kept
// separate from TiDB row-lock contention so request metrics can distinguish a
// busy parent-lane admission timeout from a database row-lock wait/deadlock.
func IsParentLaneContention(err error) bool {
	return err != nil && stderrors.Is(err, ErrParentLaneContention)
}

// IsDBRowLockContention reports whether err represents a TiDB/MySQL row-lock
// contention error (deadlock 1213 or lock-wait timeout 1205). Prefer typed
// *mysql.MySQLError inspection, then fall back to the legacy opaque-string
// forms emitted by older call sites and logs.
func IsDBRowLockContention(err error) bool {
	if err == nil {
		return false
	}
	var mysqlErr *mysql.MySQLError
	if stderrors.As(err, &mysqlErr) {
		return mysqlErr.Number == 1213 || mysqlErr.Number == 1205
	}

	errMsg := strings.ToLower(err.Error())
	return strings.Contains(errMsg, "deadlock") ||
		strings.Contains(errMsg, "lock wait timeout") ||
		strings.Contains(errMsg, "error "+ErrCodeDeadlock) ||
		strings.Contains(errMsg, "error "+ErrCodeLockWait)
}