package records

import (
	"context"

	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/persistence"
)

// RunInTransaction executes a function within a transaction (new or existing) with retry support
func (ps *PersistenceService) RunInTransaction(
	ctx context.Context,
	fn func(ctx context.Context) error,
) error {
	// If already in a transaction, execute directly
	if ps.txManager.IsTransactionActive(ctx) {
		return fn(ctx)
	}

	// Otherwise start new transaction with retry
	return ps.txManager.WithRetry(ctx, func(txCtx context.Context) error {
		return ps.runWithRecordLifecycleBatchAttempt(txCtx, fn)
	}, 3)
}

// RunInPessimisticTransaction executes a function within a pessimistic
// transaction (new or existing) with retry support for lock contention
// (TiDB 1205/1213). If the context already carries a pessimistic marker
// (i.e. is already inside a WithPessimisticTransaction call), fn is
// executed directly without opening a second transaction.
// If a plain (non-pessimistic) transaction is active, this fails closed
// with an error — opening an independent pessimistic transaction inside
// a plain one would break atomicity. The caller must either use
// WithPessimisticTransaction at the outermost scope or DetachTx first.
func (ps *PersistenceService) RunInPessimisticTransaction(
	ctx context.Context,
	fn func(ctx context.Context) error,
) error {
	_, err := ps.RunInPessimisticTransactionWithOutcome(ctx, fn)
	return err
}

// RunInPessimisticTransactionWithOutcome is the maintenance-grade variant of
// RunInPessimisticTransaction. In addition to the callback error it returns the
// transaction manager's typed terminal outcome. Callers may claim rollback
// only from TransactionCompletionRolledBack; error ancestry is not proof that
// rollback itself completed.
func (ps *PersistenceService) RunInPessimisticTransactionWithOutcome(
	ctx context.Context,
	fn func(ctx context.Context) error,
) (ports.TransactionCompletionOutcome, error) {
	// If already in a pessimistic transaction, execute directly
	if ps.txManager.IsPessimisticTx(ctx) {
		return "", fn(ctx)
	}

	// Run the record-write pessimistic transaction at READ COMMITTED. Combined
	// with the parent-row lock acquired before each rollup recompute, RC lets
	// the recompute read the latest committed siblings with a plain SELECT
	// instead of a child FOR UPDATE that locks every child row of the parent
	// (the #2936 deadlock/lockwait hot spot). Scoped here so direct
	// txManager.WithPessimisticTransaction callers (outbox/metadata/approval
	// lifecycle) keep REPEATABLE READ semantics.
	ctx = persistence.WithReadCommittedIsolation(ctx)
	ctx = ports.EnsureHistorySequenceRetryPlan(ctx)

	// ExclusiveSiblingFields can discover a parent lock only after before-save
	// final-image mutation. When this method owns the pessimistic boundary,
	// restart the whole transaction with those parent locks forced up front so
	// the earlier child lock is released before the retry.
	forcedParentPrelocks := parentLockForcedPrelocksFromContext(ctx)
	var err error
	var completion ports.TransactionCompletionOutcome
	for attempt := 0; attempt < 3; attempt++ {
		completion = ""
		retryCtx := withParentLockForcedPrelocks(ctx, forcedParentPrelocks)
		// Delegate to WithPessimisticRetry, which fails closed if a plain
		// transaction is active (atomicity break).
		err = ps.txManager.WithPessimisticRetry(retryCtx, func(txCtx context.Context) error {
			if !ports.RegisterTransactionCompletionHook(txCtx, func(outcome ports.TransactionCompletionOutcome) {
				completion = outcome
			}) {
				completion = ""
			}
			return ps.runWithRecordLifecycleBatchAttempt(txCtx, func(batchCtx context.Context) error {
				if len(forcedParentPrelocks) == 0 {
					return fn(batchCtx)
				}
				// The transaction boundary owns replay, so it must also consume the
				// parent intents learned by the previous attempt before any nested
				// write can establish a later lock high-water mark. Merely carrying
				// the intents on context makes correctness depend on every callback's
				// first planner remembering to consume them; a callback that first
				// locks another parent repeats the same out-of-order failure forever.
				return ps.runParentLockCoordinatorIfActive(batchCtx, func(lockCtx context.Context) error {
					if _, lockErr := ps.LockParentRows(lockCtx, forcedParentPrelocks); lockErr != nil {
						return lockErr
					}
					return fn(lockCtx)
				})
			})
		}, 3)
		retryKeys, shouldRetry := parentLockPrelockRetryKeys(err)
		if !shouldRetry {
			return completion, err
		}
		var changed bool
		forcedParentPrelocks, changed = mergeParentLockKeys(forcedParentPrelocks, retryKeys)
		if !changed {
			return completion, err
		}
	}
	return completion, err
}

// IsTransactionActive reports whether ctx already carries an open database
// transaction. Automation uses this to avoid opening a nested pessimistic
// wrapper while some caller families still invoke the flow engine from a
// plain outer transaction.
func (ps *PersistenceService) IsTransactionActive(ctx context.Context) bool {
	if ps == nil || ps.txManager == nil {
		return false
	}
	return ps.txManager.IsTransactionActive(ctx)
}