package store

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"sort"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
)

const (
	admissionWaiterLeaseTTL = 15 * time.Second
	admissionNotifyChannel  = "gateway_task_admission"
	admissionNotifyTimeout  = 2 * time.Second
	executionSlotWaitMax    = 24 * time.Hour
)

var ErrQueueTimeout = errors.New("task admission queue wait timed out")

type QueueTimeoutError struct {
	TaskID string
}

func (e *QueueTimeoutError) Error() string {
	return ErrQueueTimeout.Error()
}

func (e *QueueTimeoutError) ErrorCode() string {
	return "queue_timeout"
}

func (e *QueueTimeoutError) Unwrap() error {
	return ErrQueueTimeout
}

type AdmissionScope struct {
	ScopeType       string
	ScopeKey        string
	ScopeName       string
	ScopeMetadata   map[string]any
	ConcurrentLimit float64
	Amount          float64
	LeaseTTLSeconds int
	QueueLimit      int
	MaxWaitSeconds  int
	Policy          map[string]any
}

type TaskAdmissionInput struct {
	TaskID          string
	PlatformID      string
	PlatformModelID string
	UserGroupID     string
	QueueKey        string
	Mode            string
	Priority        int
	WaiterID        string
	Scopes          []AdmissionScope
}

type TaskAdmission struct {
	TaskID                string
	PlatformID            string
	PlatformModelID       string
	UserGroupID           string
	QueueKey              string
	Mode                  string
	Status                string
	Priority              int
	EnqueuedAt            time.Time
	WaitDeadlineAt        time.Time
	AdmittedAt            time.Time
	WaiterID              string
	WaiterLeaseExpires    time.Time
	Scopes                []AdmissionScope
	ReselectRequestedAt   time.Time
	DispatchNextAt        time.Time
	DispatchFailureCount  int
	DispatchLastOutcome   string
	DispatchLastErrorCode string
	DispatchLastErrorAt   time.Time
	SnapshotInvalid       bool
}

type TaskAdmissionBlocker struct {
	Reason    string
	ScopeType string
	ScopeKey  string
	RetryAt   time.Time
}

type TaskAdmissionResult struct {
	Admission     TaskAdmission
	Admitted      bool
	NewlyAdmitted bool
	Leases        []ConcurrencyLease
	Blockers      []TaskAdmissionBlocker
}

type TaskAdmissionBatchOutcome struct {
	TaskID string
	Result TaskAdmissionResult
	Err    error
}

type TaskAdmissionMetricsSnapshot struct {
	QueueDepth                    int
	WaitingSync                   int
	WaitingAsync                  int
	DeferredAsync                 int
	PendingReselect               int
	OldestWaitSeconds             float64
	OldestDispatchableWaitSeconds float64
	ExpiredWaiterBacklog          int
	ExpiredDeadlineBacklog        int
}

type TaskAdmissionReapResult struct {
	ExpiredWaiters   int
	ExpiredDeadlines int
}

func (s *Store) TryTaskAdmission(ctx context.Context, input TaskAdmissionInput) (TaskAdmissionResult, error) {
	return s.tryTaskAdmission(ctx, input, nil)
}

// TryTaskAdmissionWithAdmittedHook runs onAdmitted in the same PostgreSQL
// transaction that changes the admission to admitted. Asynchronous callers use
// this hook to insert the unique River job without leaving a crash window
// between capacity admission and durable job creation.
func (s *Store) TryTaskAdmissionWithAdmittedHook(
	ctx context.Context,
	input TaskAdmissionInput,
	onAdmitted func(pgx.Tx) error,
) (TaskAdmissionResult, error) {
	return s.tryTaskAdmission(ctx, input, onAdmitted)
}

// QueueTaskAdmissionWithHook durably registers an asynchronous FIFO waiter and
// runs onQueued in the same transaction. It intentionally does not reserve
// concurrency leases: the Worker dispatcher obtains business and execution
// capacity only when the task reaches the head of every applicable queue.
func (s *Store) QueueTaskAdmissionWithHook(
	ctx context.Context,
	input TaskAdmissionInput,
	onQueued func(pgx.Tx) error,
) (TaskAdmission, error) {
	if err := validateTaskAdmissionInput(input); err != nil {
		return TaskAdmission{}, err
	}
	if input.Mode != "async" {
		return TaskAdmission{}, errors.New("queued admission registration requires asynchronous mode")
	}
	return retryAdmissionOperation(ctx, queuedAdmissionOperationLockKeys(input), func() (TaskAdmission, error) {
		return s.queueTaskAdmissionOnce(ctx, input, onQueued)
	})
}

func (s *Store) queueTaskAdmissionOnce(
	ctx context.Context,
	input TaskAdmissionInput,
	onQueued func(pgx.Tx) error,
) (TaskAdmission, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return TaskAdmission{}, err
	}
	defer rollbackTransaction(tx)
	// The task itself was synchronously committed before this derived FIFO
	// registration. Do not serialize a large media burst on one cross-region
	// synchronous commit per waiting row: PostgreSQL still writes this WAL
	// locally, and generic queued-task recovery reconstructs a registration
	// after failover if the small asynchronous-commit window is hit. Admission,
	// River job, execution lease, task state, and billing commits stay fully
	// synchronous.
	if _, err := tx.Exec(ctx, `SET LOCAL synchronous_commit = off`); err != nil {
		return TaskAdmission{}, err
	}

	if err := tryAdmissionTransactionLock(ctx, tx, "task-admission:"+input.TaskID); err != nil {
		return TaskAdmission{}, err
	}
	var taskActive bool
	if err := tx.QueryRow(ctx, `
SELECT status IN ('queued', 'running')
FROM gateway_tasks
WHERE id = $1::uuid`, input.TaskID).Scan(&taskActive); err != nil {
		return TaskAdmission{}, err
	}
	if !taskActive {
		return TaskAdmission{}, ErrTaskExecutionFinished
	}

	admission, found, err := loadTaskAdmissionTx(ctx, tx, input.TaskID)
	if err != nil {
		return TaskAdmission{}, err
	}
	scopes := normalizedAdmissionScopes(input.Scopes)
	lockScopes := append([]AdmissionScope{}, scopes...)
	if found {
		lockScopes = append(lockScopes,
			AdmissionScope{ScopeType: "platform_model", ScopeKey: admission.PlatformModelID},
			AdmissionScope{ScopeType: "user_group", ScopeKey: admission.UserGroupID},
		)
	}
	// Waiting registration has its own cross-process lock domain. Registrations
	// still serialize with each other for exact queue-limit checks, but they do
	// not wait behind a Worker admission transaction holding execution scope
	// locks through synchronous replication. Moving one task from waiting to
	// active preserves active+waiting, so concurrent dispatch cannot make this
	// capacity check over-admit.
	for _, scope := range normalizedAdmissionScopes(lockScopes) {
		if err := tryAdmissionTransactionLock(ctx, tx, admissionQueueRegistrationLockKey(scope)); err != nil {
			return TaskAdmission{}, err
		}
	}

	if found && admission.Status == "admitted" {
		activeLeases, leaseErr := activeTaskAdmissionLeaseCountTx(ctx, tx, input.TaskID)
		if leaseErr != nil {
			return TaskAdmission{}, leaseErr
		}
		if activeLeases == 0 {
			admission, err = resetTaskAdmissionToWaitingTx(ctx, tx, input)
			if err != nil {
				return TaskAdmission{}, err
			}
		}
	}
	if found && admission.Status == "waiting" && !admission.WaitDeadlineAt.After(time.Now()) {
		if err := expireTaskAdmissionTx(ctx, tx, input.TaskID); err != nil {
			return TaskAdmission{}, err
		}
		if err := tx.Commit(ctx); err != nil {
			return TaskAdmission{}, err
		}
		s.notifyTaskAdmissionBestEffort(ctx, "*")
		return TaskAdmission{}, &QueueTimeoutError{TaskID: input.TaskID}
	}
	if !found {
		scopeStates, stateErr := admissionScopeStatesTx(ctx, tx, scopes)
		if stateErr != nil {
			return TaskAdmission{}, stateErr
		}
		if admissionErr := validateQueuedAdmissionCapacity(scopeStates); admissionErr != nil {
			return TaskAdmission{}, admissionErr
		}
		admission, err = insertTaskAdmissionTx(ctx, tx, input, admissionDeadline(scopes))
		if err != nil {
			return TaskAdmission{}, err
		}
		found = true
	}
	if onQueued != nil {
		if err := onQueued(tx); err != nil {
			return TaskAdmission{}, err
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return TaskAdmission{}, err
	}
	s.notifyTaskAdmissionBestEffort(ctx, input.TaskID)
	return admission, nil
}

func (s *Store) tryTaskAdmission(
	ctx context.Context,
	input TaskAdmissionInput,
	onAdmitted func(pgx.Tx) error,
) (TaskAdmissionResult, error) {
	if err := validateTaskAdmissionInput(input); err != nil {
		return TaskAdmissionResult{}, err
	}
	return retryAdmissionOperation(ctx, admissionOperationLockKeys(input), func() (TaskAdmissionResult, error) {
		return s.tryTaskAdmissionOnce(ctx, input, onAdmitted)
	})
}

func (s *Store) tryTaskAdmissionOnce(
	ctx context.Context,
	input TaskAdmissionInput,
	onAdmitted func(pgx.Tx) error,
) (TaskAdmissionResult, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return TaskAdmissionResult{}, err
	}
	defer rollbackTransaction(tx)

	outcome, err := s.tryTaskAdmissionTx(ctx, tx, input, onAdmitted)
	if err != nil {
		return TaskAdmissionResult{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return TaskAdmissionResult{}, err
	}
	s.notifyTaskAdmissionBestEffort(ctx, outcome.NotifyTaskID)
	return outcome.Result, outcome.PostCommitErr
}

type taskAdmissionTxOutcome struct {
	Result        TaskAdmissionResult
	PostCommitErr error
	NotifyTaskID  string
}

func (s *Store) tryTaskAdmissionTx(
	ctx context.Context,
	tx pgx.Tx,
	input TaskAdmissionInput,
	onAdmitted func(pgx.Tx) error,
) (taskAdmissionTxOutcome, error) {
	if err := tryAdmissionTransactionLock(ctx, tx, "task-admission:"+input.TaskID); err != nil {
		return taskAdmissionTxOutcome{}, err
	}
	var taskActive bool
	if err := tx.QueryRow(ctx, `
SELECT status IN ('queued', 'running')
FROM gateway_tasks
WHERE id = $1::uuid`, input.TaskID).Scan(&taskActive); err != nil {
		return taskAdmissionTxOutcome{}, err
	}
	if !taskActive {
		return taskAdmissionTxOutcome{}, ErrTaskExecutionFinished
	}

	admission, found, err := loadTaskAdmissionTx(ctx, tx, input.TaskID)
	if err != nil {
		return taskAdmissionTxOutcome{}, err
	}
	scopes := normalizedAdmissionScopes(input.Scopes)
	lockScopes := append([]AdmissionScope{}, scopes...)
	if found {
		lockScopes = append(lockScopes,
			AdmissionScope{ScopeType: "platform_model", ScopeKey: admission.PlatformModelID},
			AdmissionScope{ScopeType: "user_group", ScopeKey: admission.UserGroupID},
		)
	}
	for _, scope := range normalizedAdmissionScopes(lockScopes) {
		if err := tryAdmissionTransactionLock(ctx, tx, admissionLockKey(scope)); err != nil {
			return taskAdmissionTxOutcome{}, err
		}
	}
	if found && admission.Status == "admitted" {
		activeLeases, leaseErr := activeTaskAdmissionLeaseCountTx(ctx, tx, input.TaskID)
		if leaseErr != nil {
			return taskAdmissionTxOutcome{}, leaseErr
		}
		bindingChanged := admission.PlatformID != input.PlatformID ||
			admission.PlatformModelID != input.PlatformModelID ||
			admission.UserGroupID != input.UserGroupID
		if activeLeases > 0 && !bindingChanged {
			result := TaskAdmissionResult{Admission: admission, Admitted: true}
			if onAdmitted != nil {
				if err := onAdmitted(tx); err != nil {
					return taskAdmissionTxOutcome{}, err
				}
			}
			return taskAdmissionTxOutcome{Result: result}, nil
		}
		if bindingChanged || activeLeases == 0 {
			targetStates, stateErr := admissionScopeStatesTx(ctx, tx, scopes)
			if stateErr != nil {
				return taskAdmissionTxOutcome{}, stateErr
			}
			if admissionErr := validateNewAdmissionCapacity(targetStates); admissionErr != nil {
				return taskAdmissionTxOutcome{}, admissionErr
			}
		}
		admission, err = resetTaskAdmissionToWaitingTx(ctx, tx, input)
		if err != nil {
			return taskAdmissionTxOutcome{}, err
		}
	}
	if found && !admission.WaitDeadlineAt.After(time.Now()) {
		if err := expireTaskAdmissionTx(ctx, tx, input.TaskID); err != nil {
			return taskAdmissionTxOutcome{}, err
		}
		return taskAdmissionTxOutcome{
			PostCommitErr: &QueueTimeoutError{TaskID: input.TaskID},
			NotifyTaskID:  "*",
		}, nil
	}

	scopeStates, err := admissionScopeStatesTx(ctx, tx, scopes)
	if err != nil {
		return taskAdmissionTxOutcome{}, err
	}
	if !found {
		if admissionErr := validateNewAdmissionCapacity(scopeStates); admissionErr != nil {
			return taskAdmissionTxOutcome{}, admissionErr
		}
		deadline := admissionDeadline(scopes)
		admission, err = insertTaskAdmissionTx(ctx, tx, input, deadline)
		if err != nil {
			return taskAdmissionTxOutcome{}, err
		}
		found = true
	} else if input.Mode == "sync" && strings.TrimSpace(input.WaiterID) != "" {
		admission, err = renewTaskAdmissionWaiterTx(ctx, tx, input.TaskID, input.WaiterID)
		if err != nil {
			return taskAdmissionTxOutcome{}, err
		}
	}

	head, headBlockers, err := isTaskAdmissionHeadTx(ctx, tx, admission, scopes)
	if err != nil {
		return taskAdmissionTxOutcome{}, err
	}
	if !head {
		return taskAdmissionTxOutcome{
			Result: TaskAdmissionResult{Admission: admission, Blockers: headBlockers},
		}, nil
	}
	blockers := make([]TaskAdmissionBlocker, 0, len(scopeStates))
	for _, state := range scopeStates {
		if state.Saturated {
			blockers = append(blockers, TaskAdmissionBlocker{
				Reason:    "saturated",
				ScopeType: state.Scope.ScopeType,
				ScopeKey:  state.Scope.ScopeKey,
				RetryAt:   state.NextLeaseExpiration,
			})
		}
	}
	if len(blockers) > 0 {
		return taskAdmissionTxOutcome{
			Result: TaskAdmissionResult{Admission: admission, Blockers: blockers},
		}, nil
	}

	leases := make([]ConcurrencyLease, 0, len(scopes))
	for _, scope := range scopes {
		if scope.ConcurrentLimit <= 0 || scope.Amount <= 0 {
			continue
		}
		lease, err := reserveConcurrencyLease(ctx, tx, input.TaskID, "", RateLimitReservation{
			ScopeType:       scope.ScopeType,
			ScopeKey:        scope.ScopeKey,
			ScopeName:       scope.ScopeName,
			ScopeMetadata:   scope.ScopeMetadata,
			Metric:          "concurrent",
			Limit:           scope.ConcurrentLimit,
			Amount:          scope.Amount,
			LeaseTTLSeconds: scope.LeaseTTLSeconds,
			Policy:          scope.Policy,
		})
		if err != nil {
			if errors.Is(err, ErrRateLimited) {
				return taskAdmissionTxOutcome{
					Result: TaskAdmissionResult{Admission: admission},
				}, nil
			}
			return taskAdmissionTxOutcome{}, err
		}
		leases = append(leases, lease)
	}
	admission, err = markTaskAdmissionAdmittedTx(ctx, tx, input.TaskID)
	if err != nil {
		return taskAdmissionTxOutcome{}, err
	}
	// Continue the FIFO chain after one task is admitted. Every API process
	// only probes the global head it owns, so free capacity is filled without
	// broadcasting an advisory-lock attempt to every waiter.
	if onAdmitted != nil {
		if err := onAdmitted(tx); err != nil {
			return taskAdmissionTxOutcome{}, err
		}
	}
	return taskAdmissionTxOutcome{
		Result: TaskAdmissionResult{
			Admission:     admission,
			Admitted:      true,
			NewlyAdmitted: true,
			Leases:        leases,
		},
		NotifyTaskID: "*",
	}, nil
}

// TryTaskAdmissionBatchWithAdmittedHook fills a FIFO capacity window with one
// transaction per task. Each transaction still atomically reserves admission
// and inserts its unique River job, but a large global capacity window no
// longer holds every task row and scope lock until the entire batch commits.
// This is deliberately a batch at the dispatcher boundary, not a database
// transaction boundary.
func (s *Store) TryTaskAdmissionBatchWithAdmittedHook(
	ctx context.Context,
	inputs []TaskAdmissionInput,
	onAdmitted func(pgx.Tx, TaskAdmissionInput) error,
) ([]TaskAdmissionBatchOutcome, error) {
	if len(inputs) == 0 {
		return nil, nil
	}
	for _, input := range inputs {
		if err := validateTaskAdmissionInput(input); err != nil {
			return nil, err
		}
	}
	outcomes := make([]TaskAdmissionBatchOutcome, 0, len(inputs))
	for _, input := range inputs {
		hook := func(tx pgx.Tx) error {
			if onAdmitted == nil {
				return nil
			}
			return onAdmitted(tx, input)
		}
		result, admissionErr := s.TryTaskAdmissionWithAdmittedHook(ctx, input, hook)
		outcomes = append(outcomes, TaskAdmissionBatchOutcome{
			TaskID: input.TaskID,
			Result: result,
			Err:    admissionErr,
		})
		if errors.Is(admissionErr, ErrTaskExecutionFinished) ||
			errors.Is(admissionErr, ErrQueueTimeout) {
			continue
		}
		if admissionErr != nil {
			return outcomes, admissionErr
		}
	}
	return outcomes, nil
}

// TryTaskAdmissionAtomicBatchWithAdmittedHook admits a deliberately small
// group of tasks in one transaction. The caller bounds the group size so the
// global capacity lock spans one synchronous commit without returning to the
// old failure mode where a full capacity window locked dozens of task rows.
// Admission leases and unique River jobs either commit together for the whole
// microbatch or roll back together.
func (s *Store) TryTaskAdmissionAtomicBatchWithAdmittedHook(
	ctx context.Context,
	inputs []TaskAdmissionInput,
	onAdmitted func(pgx.Tx, TaskAdmissionInput) error,
) ([]TaskAdmissionBatchOutcome, error) {
	if len(inputs) == 0 {
		return nil, nil
	}
	for _, input := range inputs {
		if err := validateTaskAdmissionInput(input); err != nil {
			return nil, err
		}
	}
	lockKeys := make([]string, 0, len(inputs)*3)
	for _, input := range inputs {
		lockKeys = append(lockKeys, admissionOperationLockKeys(input)...)
	}
	return retryAdmissionOperation(ctx, lockKeys, func() ([]TaskAdmissionBatchOutcome, error) {
		return s.tryTaskAdmissionAtomicBatchOnce(ctx, inputs, onAdmitted, lockKeys)
	})
}

func (s *Store) tryTaskAdmissionAtomicBatchOnce(
	ctx context.Context,
	inputs []TaskAdmissionInput,
	onAdmitted func(pgx.Tx, TaskAdmissionInput) error,
	lockKeys []string,
) ([]TaskAdmissionBatchOutcome, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer rollbackTransaction(tx)
	// Lock every task and scope in deterministic order. Without this pre-lock,
	// two dispatchers could each hold a different task lock while waiting for
	// the shared worker-capacity lock.
	for _, key := range normalizedAdmissionLockKeys(lockKeys) {
		if err := tryAdmissionTransactionLock(ctx, tx, key); err != nil {
			return nil, err
		}
	}

	outcomes := make([]TaskAdmissionBatchOutcome, 0, len(inputs))
	notify := false
	for _, input := range inputs {
		hook := func(tx pgx.Tx) error {
			if onAdmitted == nil {
				return nil
			}
			return onAdmitted(tx, input)
		}
		outcome, admissionErr := s.tryTaskAdmissionTx(ctx, tx, input, hook)
		if errors.Is(admissionErr, ErrTaskExecutionFinished) {
			outcomes = append(outcomes, TaskAdmissionBatchOutcome{
				TaskID: input.TaskID,
				Err:    admissionErr,
			})
			continue
		}
		if admissionErr != nil {
			outcomes = append(outcomes, TaskAdmissionBatchOutcome{
				TaskID: input.TaskID,
				Err:    admissionErr,
			})
			return outcomes, admissionErr
		}
		outcomes = append(outcomes, TaskAdmissionBatchOutcome{
			TaskID: input.TaskID,
			Result: outcome.Result,
			Err:    outcome.PostCommitErr,
		})
		if outcome.NotifyTaskID != "" {
			notify = true
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return outcomes, err
	}
	if notify {
		s.notifyTaskAdmissionBestEffort(ctx, "*")
	}
	return outcomes, nil
}

func validateNewAdmissionCapacity(states []admissionScopeState) error {
	mustWait := false
	for _, state := range states {
		if !state.Saturated {
			continue
		}
		mustWait = true
		if state.Scope.QueueLimit <= 0 {
			return concurrencyAdmissionError(state)
		}
	}
	if !mustWait {
		return nil
	}
	for _, state := range states {
		if state.Scope.QueueLimit > 0 && state.Waiting >= state.Scope.QueueLimit {
			return queueFullAdmissionError(state)
		}
	}
	return nil
}

func validateQueuedAdmissionCapacity(states []admissionScopeState) error {
	// Waiting asynchronous tasks have not acquired leases yet. Count them as
	// virtual capacity reservations so a burst cannot bypass concurrent and
	// queue limits before the dispatcher creates River jobs.
	for _, state := range states {
		if state.Scope.ConcurrentLimit <= 0 {
			continue
		}
		virtualUsage := state.Active + float64(state.Waiting)*state.Scope.Amount + state.Scope.Amount
		if virtualUsage <= state.Scope.ConcurrentLimit {
			continue
		}
		if state.Scope.QueueLimit <= 0 {
			return concurrencyAdmissionError(state)
		}
		virtualQueueDepth := int(math.Ceil(virtualUsage - state.Scope.ConcurrentLimit))
		if virtualQueueDepth > state.Scope.QueueLimit {
			return queueFullAdmissionError(state)
		}
	}
	return nil
}

// RebindWaitingTaskAdmission migrates a queued task to a newly selected
// candidate without changing its FIFO age or priority. Capacity checks happen
// while both the old and new queue locks are held.
func (s *Store) RebindWaitingTaskAdmission(ctx context.Context, input TaskAdmissionInput) (TaskAdmission, error) {
	if err := validateTaskAdmissionInput(input); err != nil {
		return TaskAdmission{}, err
	}
	return retryAdmissionOperation(ctx, queuedAdmissionOperationLockKeys(input), func() (TaskAdmission, error) {
		return s.rebindWaitingTaskAdmissionOnce(ctx, input)
	})
}

func (s *Store) rebindWaitingTaskAdmissionOnce(ctx context.Context, input TaskAdmissionInput) (TaskAdmission, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return TaskAdmission{}, err
	}
	defer rollbackTransaction(tx)
	if err := tryAdmissionTransactionLock(ctx, tx, "task-admission:"+input.TaskID); err != nil {
		return TaskAdmission{}, err
	}
	current, found, err := loadTaskAdmissionTx(ctx, tx, input.TaskID)
	if err != nil {
		return TaskAdmission{}, err
	}
	if !found {
		return TaskAdmission{}, pgx.ErrNoRows
	}
	if current.Status != "waiting" {
		return current, nil
	}
	scopes := normalizedAdmissionScopes(input.Scopes)
	lockScopes := append([]AdmissionScope{}, scopes...)
	lockScopes = append(lockScopes,
		AdmissionScope{ScopeType: "platform_model", ScopeKey: current.PlatformModelID},
		AdmissionScope{ScopeType: "user_group", ScopeKey: current.UserGroupID},
	)
	for _, scope := range normalizedAdmissionScopes(lockScopes) {
		if err := tryAdmissionTransactionLock(ctx, tx, admissionQueueRegistrationLockKey(scope)); err != nil {
			return TaskAdmission{}, err
		}
	}
	states, err := admissionScopeStatesTx(ctx, tx, scopes)
	if err != nil {
		return TaskAdmission{}, err
	}
	for index := range states {
		sameScope := (states[index].Scope.ScopeType == "platform_model" && states[index].Scope.ScopeKey == current.PlatformModelID) ||
			(states[index].Scope.ScopeType == "user_group" && states[index].Scope.ScopeKey == current.UserGroupID)
		if sameScope && states[index].Waiting > 0 {
			states[index].Waiting--
		}
	}
	if admissionErr := validateNewAdmissionCapacity(states); admissionErr != nil {
		return TaskAdmission{}, admissionErr
	}
	deadline := admissionDeadline(scopes)
	scopeSnapshot, err := json.Marshal(scopes)
	if err != nil {
		return TaskAdmission{}, fmt.Errorf("marshal rebound task admission scopes: %w", err)
	}
	row := tx.QueryRow(ctx, `
UPDATE gateway_task_admissions
SET platform_id = $2::uuid,
    platform_model_id = $3::uuid,
    user_group_id = NULLIF($4, '')::uuid,
    queue_key = $5,
    wait_deadline_at = LEAST(wait_deadline_at, $6),
    scope_snapshot = $7::jsonb,
    reselect_requested_at = NULL,
    dispatch_next_at = now(),
    dispatch_failure_count = 0,
    dispatch_last_outcome = 'candidate_migrated',
    dispatch_last_error_code = NULL,
    dispatch_last_error_at = NULL,
    updated_at = now()
WHERE task_id = $1::uuid
  AND status = 'waiting'
RETURNING task_id::text, platform_id::text, platform_model_id::text,
          COALESCE(user_group_id::text, ''), queue_key, mode, status, priority,
          enqueued_at, wait_deadline_at, admitted_at,
          COALESCE(waiter_id, ''), waiter_lease_expires_at,
          scope_snapshot, reselect_requested_at, dispatch_next_at,
          dispatch_failure_count, COALESCE(dispatch_last_outcome, ''),
          COALESCE(dispatch_last_error_code, ''), dispatch_last_error_at`,
		input.TaskID,
		input.PlatformID,
		input.PlatformModelID,
		input.UserGroupID,
		input.QueueKey,
		deadline,
		scopeSnapshot,
	)
	migrated, err := scanTaskAdmission(row)
	if err != nil {
		return TaskAdmission{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return TaskAdmission{}, err
	}
	s.notifyTaskAdmissionBestEffort(ctx, input.TaskID)
	return migrated, nil
}

func activeTaskAdmissionLeaseCountTx(ctx context.Context, tx pgx.Tx, taskID string) (int, error) {
	var count int
	err := tx.QueryRow(ctx, `
SELECT COUNT(*)
FROM gateway_concurrency_leases
WHERE task_id = $1::uuid
  AND released_at IS NULL
  AND expires_at > statement_timestamp()`, taskID).Scan(&count)
	return count, err
}

func resetTaskAdmissionToWaitingTx(ctx context.Context, tx pgx.Tx, input TaskAdmissionInput) (TaskAdmission, error) {
	if _, err := tx.Exec(ctx, `
UPDATE gateway_concurrency_leases
SET released_at = statement_timestamp()
WHERE task_id = $1::uuid
  AND released_at IS NULL`, input.TaskID); err != nil {
		return TaskAdmission{}, err
	}
	var waiterLease any
	if input.Mode == "sync" {
		waiterLease = time.Now().Add(admissionWaiterLeaseTTL)
	}
	scopeSnapshot, err := json.Marshal(normalizedAdmissionScopes(input.Scopes))
	if err != nil {
		return TaskAdmission{}, fmt.Errorf("marshal reset task admission scopes: %w", err)
	}
	row := tx.QueryRow(ctx, `
UPDATE gateway_task_admissions
SET status = 'waiting',
    admitted_at = NULL,
    platform_id = $2::uuid,
    platform_model_id = $3::uuid,
    user_group_id = NULLIF($4, '')::uuid,
    queue_key = $5,
    wait_deadline_at = LEAST(wait_deadline_at, $6),
    waiter_id = NULLIF($7, ''),
    waiter_lease_expires_at = $8,
    scope_snapshot = $9::jsonb,
    dispatch_next_at = now(),
    dispatch_failure_count = 0,
    dispatch_last_outcome = NULL,
    dispatch_last_error_code = NULL,
    dispatch_last_error_at = NULL,
    updated_at = now()
WHERE task_id = $1::uuid
RETURNING task_id::text, platform_id::text, platform_model_id::text,
          COALESCE(user_group_id::text, ''), queue_key, mode, status, priority,
          enqueued_at, wait_deadline_at, admitted_at,
          COALESCE(waiter_id, ''), waiter_lease_expires_at,
          scope_snapshot, reselect_requested_at, dispatch_next_at,
          dispatch_failure_count, COALESCE(dispatch_last_outcome, ''),
          COALESCE(dispatch_last_error_code, ''), dispatch_last_error_at`,
		input.TaskID,
		input.PlatformID,
		input.PlatformModelID,
		input.UserGroupID,
		input.QueueKey,
		admissionDeadline(input.Scopes),
		strings.TrimSpace(input.WaiterID),
		waiterLease,
		scopeSnapshot,
	)
	return scanTaskAdmission(row)
}

func (s *Store) RenewTaskAdmissionWaiter(ctx context.Context, taskID string, waiterID string) error {
	result, err := s.pool.Exec(ctx, `
UPDATE gateway_task_admissions
SET waiter_lease_expires_at = now() + $3::interval,
    updated_at = now()
WHERE task_id = $1::uuid
  AND mode = 'sync'
  AND status = 'waiting'
  AND waiter_id = $2`,
		taskID, waiterID, admissionWaiterLeaseTTL.String())
	if err != nil {
		return err
	}
	if result.RowsAffected() == 0 {
		return pgx.ErrNoRows
	}
	return nil
}

// RenewTaskAdmissionWaiters refreshes all synchronous waiters owned by one API
// process in a single statement. This keeps the 15-second crash-recovery lease
// without running a full advisory-lock admission transaction per waiter.
func (s *Store) RenewTaskAdmissionWaiters(ctx context.Context, taskIDs []string, waiterIDs []string) error {
	if len(taskIDs) == 0 {
		return nil
	}
	if len(taskIDs) != len(waiterIDs) {
		return errors.New("task admission waiter renewal requires matching task and waiter IDs")
	}
	_, err := s.pool.Exec(ctx, `
UPDATE gateway_task_admissions admission
SET waiter_lease_expires_at = now() + $3::interval,
    updated_at = now()
FROM unnest($1::uuid[], $2::text[]) AS owned(task_id, waiter_id)
WHERE admission.task_id = owned.task_id
  AND admission.waiter_id = owned.waiter_id
  AND admission.mode = 'sync'
  AND admission.status = 'waiting'`,
		taskIDs,
		waiterIDs,
		admissionWaiterLeaseTTL.String(),
	)
	return err
}

func (s *Store) DeleteTaskAdmission(ctx context.Context, taskID string) error {
	_, err := retryAdmissionOperation(ctx, []string{"task-admission:" + taskID}, func() (struct{}, error) {
		return struct{}{}, s.deleteTaskAdmissionOnce(ctx, taskID)
	})
	return err
}

func (s *Store) deleteTaskAdmissionOnce(ctx context.Context, taskID string) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer rollbackTransaction(tx)
	if err := tryAdmissionTransactionLock(ctx, tx, "task-admission:"+taskID); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `
UPDATE gateway_concurrency_leases
SET released_at = statement_timestamp()
WHERE task_id = $1::uuid
  AND released_at IS NULL`, taskID); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `DELETE FROM gateway_task_admissions WHERE task_id = $1::uuid`, taskID); err != nil {
		return err
	}
	if err := tx.Commit(ctx); err != nil {
		return err
	}
	s.notifyTaskAdmissionBestEffort(ctx, "*")
	return nil
}

func (s *Store) ActiveTaskAdmissionLeases(ctx context.Context, taskID string) ([]ConcurrencyLease, error) {
	rows, err := s.pool.Query(ctx, `
SELECT id::text,
       GREATEST(EXTRACT(EPOCH FROM expires_at - acquired_at), 1)::float8
FROM gateway_concurrency_leases
WHERE task_id = $1::uuid
  AND released_at IS NULL
  AND expires_at > statement_timestamp()
ORDER BY scope_type, scope_key, id`, taskID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	leases := make([]ConcurrencyLease, 0)
	for rows.Next() {
		var lease ConcurrencyLease
		var ttlSeconds float64
		if err := rows.Scan(&lease.ID, &ttlSeconds); err != nil {
			return nil, err
		}
		lease.TTL = time.Duration(ttlSeconds * float64(time.Second))
		leases = append(leases, lease)
	}
	return leases, rows.Err()
}

func (s *Store) ListWaitingAsyncAdmissionTaskIDs(ctx context.Context, limit int) ([]string, error) {
	if limit <= 0 || limit > 1000 {
		limit = 100
	}
	rows, err := s.pool.Query(ctx, `
SELECT admission.task_id::text
FROM gateway_task_admissions admission
JOIN gateway_tasks task ON task.id = admission.task_id
LEFT JOIN river_job job ON job.id = task.river_job_id
WHERE admission.mode = 'async'
  AND task.status = 'queued'
  AND task.next_run_at <= now()
  AND admission.dispatch_next_at <= now()
  AND (
    admission.status = 'waiting'
    OR (
      admission.status = 'admitted'
      AND NOT EXISTS (
        SELECT 1
        FROM gateway_concurrency_leases lease
        WHERE lease.task_id = admission.task_id
          AND lease.released_at IS NULL
          AND lease.expires_at > statement_timestamp()
      )
    )
  )
  AND (
    task.river_job_id IS NULL
    OR job.id IS NULL
    OR job.state <> 'running'
  )
ORDER BY admission.priority ASC, admission.enqueued_at ASC, admission.task_id ASC
LIMIT $1`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	taskIDs := make([]string, 0)
	for rows.Next() {
		var taskID string
		if err := rows.Scan(&taskID); err != nil {
			return nil, err
		}
		taskIDs = append(taskIDs, taskID)
	}
	return taskIDs, rows.Err()
}

// ListWaitingAsyncAdmissions returns the durable routing and admission-scope
// snapshots needed by the Worker dispatcher in one round trip. Normal queued
// tasks must not be fully rehydrated and routed again merely to reserve an
// execution slot; only rows explicitly marked for reselection take that slower
// path in the runner.
func (s *Store) ListWaitingAsyncAdmissions(ctx context.Context, limit int) ([]TaskAdmission, error) {
	if limit <= 0 || limit > 1000 {
		limit = 100
	}
	rows, err := s.pool.Query(ctx, `
SELECT admission.task_id::text, admission.platform_id::text, admission.platform_model_id::text,
       COALESCE(admission.user_group_id::text, ''), admission.queue_key, admission.mode,
       admission.status, admission.priority, admission.enqueued_at, admission.wait_deadline_at,
       admission.admitted_at, COALESCE(admission.waiter_id, ''), admission.waiter_lease_expires_at,
       admission.scope_snapshot, admission.reselect_requested_at, admission.dispatch_next_at,
       admission.dispatch_failure_count, COALESCE(admission.dispatch_last_outcome, ''),
       COALESCE(admission.dispatch_last_error_code, ''), admission.dispatch_last_error_at
FROM gateway_task_admissions admission
JOIN gateway_tasks task ON task.id = admission.task_id
LEFT JOIN river_job job ON job.id = task.river_job_id
WHERE admission.mode = 'async'
  AND task.status = 'queued'
  AND task.next_run_at <= now()
  AND admission.dispatch_next_at <= now()
  AND (
    admission.status = 'waiting'
    OR (
      admission.status = 'admitted'
      AND NOT EXISTS (
        SELECT 1
        FROM gateway_concurrency_leases lease
        WHERE lease.task_id = admission.task_id
          AND lease.released_at IS NULL
          AND lease.expires_at > statement_timestamp()
      )
    )
  )
  AND (
    task.river_job_id IS NULL
    OR job.id IS NULL
    OR job.state <> 'running'
  )
ORDER BY admission.priority ASC, admission.enqueued_at ASC, admission.task_id ASC
LIMIT $1`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	admissions := make([]TaskAdmission, 0, limit)
	for rows.Next() {
		admission, scanErr := scanTaskAdmission(rows)
		if scanErr != nil {
			return nil, scanErr
		}
		admissions = append(admissions, admission)
	}
	return admissions, rows.Err()
}

// RequestTaskAdmissionReselect marks one waiting task for candidate
// reselection. The expected platform-model binding makes the operation a CAS:
// a concurrent dispatcher that already migrated or admitted the task wins.
func (s *Store) RequestTaskAdmissionReselect(
	ctx context.Context,
	taskID string,
	expectedPlatformModelID string,
) (TaskAdmission, bool, error) {
	row := s.pool.QueryRow(ctx, `
UPDATE gateway_task_admissions admission
SET reselect_requested_at = now(),
    dispatch_next_at = now(),
    dispatch_last_outcome = 'reselect',
    dispatch_last_error_code = NULL,
    dispatch_last_error_at = NULL,
    updated_at = now()
FROM gateway_tasks task
WHERE admission.task_id = $1::uuid
  AND admission.task_id = task.id
  AND admission.platform_model_id = $2::uuid
  AND admission.mode = 'async'
  AND admission.status = 'waiting'
  AND task.status = 'queued'
  AND task.next_run_at <= now()
  AND admission.reselect_requested_at IS NULL
RETURNING admission.task_id::text, admission.platform_id::text, admission.platform_model_id::text,
          COALESCE(admission.user_group_id::text, ''), admission.queue_key, admission.mode,
          admission.status, admission.priority, admission.enqueued_at, admission.wait_deadline_at,
          admission.admitted_at, COALESCE(admission.waiter_id, ''), admission.waiter_lease_expires_at,
          admission.scope_snapshot, admission.reselect_requested_at, admission.dispatch_next_at,
          admission.dispatch_failure_count, COALESCE(admission.dispatch_last_outcome, ''),
          COALESCE(admission.dispatch_last_error_code, ''), admission.dispatch_last_error_at`,
		taskID,
		expectedPlatformModelID,
	)
	admission, err := scanTaskAdmission(row)
	if errors.Is(err, pgx.ErrNoRows) {
		return TaskAdmission{}, false, nil
	}
	if err != nil {
		return TaskAdmission{}, false, err
	}
	s.notifyTaskAdmissionBestEffort(ctx, taskID)
	return admission, true, nil
}

// DeferWaitingTaskAdmission quarantines one task-specific dispatcher failure
// until retryAt. Deferred tasks are omitted from FIFO head selection so a
// corrupt or temporarily unroutable row cannot strand healthy followers.
func (s *Store) DeferWaitingTaskAdmission(
	ctx context.Context,
	taskID string,
	expectedPlatformModelID string,
	retryAt time.Time,
	outcome string,
	errorCode string,
) (TaskAdmission, bool, error) {
	if retryAt.Before(time.Now().Add(time.Second)) {
		retryAt = time.Now().Add(time.Second)
	}
	row := s.pool.QueryRow(ctx, `
UPDATE gateway_task_admissions admission
SET dispatch_next_at = $3,
    dispatch_failure_count = dispatch_failure_count + 1,
    dispatch_last_outcome = NULLIF($4, ''),
    dispatch_last_error_code = NULLIF($5, ''),
    dispatch_last_error_at = now(),
    updated_at = now()
FROM gateway_tasks task
WHERE admission.task_id = $1::uuid
  AND admission.task_id = task.id
  AND (NULLIF($2, '') IS NULL OR admission.platform_model_id = NULLIF($2, '')::uuid)
  AND admission.mode = 'async'
  AND admission.status = 'waiting'
  AND task.status = 'queued'
  AND task.next_run_at <= now()
  AND admission.dispatch_next_at <= now()
RETURNING admission.task_id::text, admission.platform_id::text, admission.platform_model_id::text,
          COALESCE(admission.user_group_id::text, ''), admission.queue_key, admission.mode,
          admission.status, admission.priority, admission.enqueued_at, admission.wait_deadline_at,
          admission.admitted_at, COALESCE(admission.waiter_id, ''), admission.waiter_lease_expires_at,
          admission.scope_snapshot, admission.reselect_requested_at, admission.dispatch_next_at,
          admission.dispatch_failure_count, COALESCE(admission.dispatch_last_outcome, ''),
          COALESCE(admission.dispatch_last_error_code, ''), admission.dispatch_last_error_at`,
		taskID,
		strings.TrimSpace(expectedPlatformModelID),
		retryAt,
		strings.TrimSpace(outcome),
		strings.TrimSpace(errorCode),
	)
	admission, err := scanTaskAdmission(row)
	if errors.Is(err, pgx.ErrNoRows) {
		return TaskAdmission{}, false, nil
	}
	if err != nil {
		return TaskAdmission{}, false, err
	}
	s.notifyTaskAdmissionBestEffort(ctx, taskID)
	return admission, true, nil
}

// ListWaitingTaskAdmissionIDs returns one FIFO leader for every independent
// platform-model queue, additionally requiring the task to lead its user-group
// queue when one exists. It is used after capacity is released so API
// processes can refill every candidate without waking every synchronous
// waiter or allowing one saturated candidate to block another.
func (s *Store) ListWaitingTaskAdmissionIDs(ctx context.Context, limit int) ([]string, error) {
	if limit <= 0 {
		limit = 256
	}
	rows, err := s.pool.Query(ctx, `
WITH ranked AS (
  SELECT task_id,
         user_group_id,
         priority,
         enqueued_at,
         row_number() OVER (
           PARTITION BY platform_model_id
           ORDER BY priority ASC, enqueued_at ASC, task_id ASC
         ) AS platform_rank,
         row_number() OVER (
           PARTITION BY user_group_id
           ORDER BY priority ASC, enqueued_at ASC, task_id ASC
         ) AS group_rank
  FROM gateway_task_admissions
  WHERE status = 'waiting'
    AND mode = 'sync'
)
SELECT task_id::text
FROM ranked
WHERE platform_rank = 1
  AND (user_group_id IS NULL OR group_rank = 1)
ORDER BY priority ASC, enqueued_at ASC, task_id ASC
LIMIT $1`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	taskIDs := make([]string, 0, limit)
	for rows.Next() {
		var taskID string
		if err := rows.Scan(&taskID); err != nil {
			return nil, err
		}
		taskIDs = append(taskIDs, taskID)
	}
	return taskIDs, rows.Err()
}

func (s *Store) GetTaskAdmission(ctx context.Context, taskID string) (TaskAdmission, error) {
	admission, found, err := loadTaskAdmission(ctx, s.pool, taskID)
	if err != nil {
		return TaskAdmission{}, err
	}
	if !found {
		return TaskAdmission{}, pgx.ErrNoRows
	}
	return admission, nil
}

func (s *Store) TaskAdmissionMetrics(ctx context.Context) (TaskAdmissionMetricsSnapshot, error) {
	var snapshot TaskAdmissionMetricsSnapshot
	err := s.pool.QueryRow(ctx, `
SELECT COUNT(*) FILTER (WHERE status = 'waiting')::int,
       COUNT(*) FILTER (WHERE status = 'waiting' AND mode = 'sync')::int,
       COUNT(*) FILTER (WHERE status = 'waiting' AND mode = 'async')::int,
       COUNT(*) FILTER (
         WHERE status = 'waiting' AND mode = 'async' AND dispatch_next_at > now()
       )::int,
       COUNT(*) FILTER (
         WHERE status = 'waiting' AND mode = 'async' AND reselect_requested_at IS NOT NULL
       )::int,
       COALESCE(EXTRACT(EPOCH FROM now() - MIN(enqueued_at) FILTER (WHERE status = 'waiting')), 0)::float8,
       COALESCE(EXTRACT(EPOCH FROM now() - MIN(enqueued_at) FILTER (
         WHERE status = 'waiting' AND mode = 'async' AND dispatch_next_at <= now()
       )), 0)::float8,
       COUNT(*) FILTER (
         WHERE status = 'waiting' AND mode = 'sync' AND waiter_lease_expires_at <= now()
       )::int,
       COUNT(*) FILTER (
         WHERE status = 'waiting' AND wait_deadline_at <= now()
       )::int
FROM gateway_task_admissions`).Scan(
		&snapshot.QueueDepth,
		&snapshot.WaitingSync,
		&snapshot.WaitingAsync,
		&snapshot.DeferredAsync,
		&snapshot.PendingReselect,
		&snapshot.OldestWaitSeconds,
		&snapshot.OldestDispatchableWaitSeconds,
		&snapshot.ExpiredWaiterBacklog,
		&snapshot.ExpiredDeadlineBacklog,
	)
	return snapshot, err
}

func (s *Store) ReapExpiredTaskAdmissions(
	ctx context.Context,
	limit int,
	defaultCallbackURLs ...string,
) (TaskAdmissionReapResult, error) {
	if limit <= 0 || limit > 1000 {
		limit = 500
	}
	defaultCallbackURL := ""
	if len(defaultCallbackURLs) > 0 {
		defaultCallbackURL = strings.TrimSpace(defaultCallbackURLs[0])
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return TaskAdmissionReapResult{}, err
	}
	defer rollbackTransaction(tx)
	rows, err := tx.Query(ctx, `
SELECT task_id::text
FROM gateway_task_admissions
WHERE status = 'waiting'
  AND (
    wait_deadline_at <= now()
    OR (mode = 'sync' AND waiter_lease_expires_at <= now())
  )
ORDER BY wait_deadline_at ASC, task_id ASC
LIMIT $1`, limit)
	if err != nil {
		return TaskAdmissionReapResult{}, err
	}
	expired := make([]string, 0)
	for rows.Next() {
		var taskID string
		if err := rows.Scan(&taskID); err != nil {
			rows.Close()
			return TaskAdmissionReapResult{}, err
		}
		expired = append(expired, taskID)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return TaskAdmissionReapResult{}, err
	}
	rows.Close()
	result := TaskAdmissionReapResult{}
	reaped := 0
	for _, taskID := range expired {
		if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, "task-admission:"+taskID); err != nil {
			return TaskAdmissionReapResult{}, err
		}
		var deadlineExpired bool
		var dispatchFailureCount int
		err := tx.QueryRow(ctx, `
DELETE FROM gateway_task_admissions
WHERE task_id = $1::uuid
  AND status = 'waiting'
  AND (
    wait_deadline_at <= now()
    OR (mode = 'sync' AND waiter_lease_expires_at <= now())
  )
RETURNING wait_deadline_at <= now(), dispatch_failure_count`, taskID).Scan(&deadlineExpired, &dispatchFailureCount)
		if errors.Is(err, pgx.ErrNoRows) {
			continue
		}
		if err != nil {
			return TaskAdmissionReapResult{}, err
		}
		code := "client_disconnected"
		message := "synchronous queue waiter disconnected"
		status := "cancelled"
		if deadlineExpired {
			code = "queue_timeout"
			message = ErrQueueTimeout.Error()
			status = "failed"
			if dispatchFailureCount > 0 {
				code = "admission_dispatch_failed"
				message = "task admission dispatcher could not prepare the task before its queue deadline"
			}
			result.ExpiredDeadlines++
		} else {
			result.ExpiredWaiters++
		}
		publicErrorJSON := encodePublicErrorSnapshot(code, message, 0, true, "", taskID)
		tag, err := tx.Exec(ctx, `
UPDATE gateway_tasks
SET status = $2,
    error = NULL,
    error_code = $4,
    error_message = $3,
    public_error = $5::jsonb,
    billing_status = CASE
      WHEN run_mode NOT IN ('production', 'acceptance', 'acceptance_canary') OR gateway_user_id IS NULL THEN 'not_required'
      WHEN reservation_amount > 0 THEN 'pending'
      ELSE 'released'
    END,
    billing_updated_at = now(),
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    finished_at = now(),
    updated_at = now()
WHERE id = $1::uuid
  AND status = 'queued'`, taskID, status, message, code, string(publicErrorJSON))
		if err != nil {
			return TaskAdmissionReapResult{}, err
		}
		if tag.RowsAffected() == 0 {
			continue
		}
		if _, err := tx.Exec(ctx, `
UPDATE gateway_concurrency_leases
SET released_at = statement_timestamp()
WHERE task_id = $1::uuid
  AND released_at IS NULL`, taskID); err != nil {
			return TaskAdmissionReapResult{}, err
		}
		if _, err := releaseRuntimeRateReservationsTx(ctx, tx, []string{taskID}); err != nil {
			return TaskAdmissionReapResult{}, err
		}
		payloadJSON, _ := json.Marshal(map[string]any{"taskId": taskID, "reason": code})
		if _, err := tx.Exec(ctx, `
INSERT INTO settlement_outbox (
  task_id, event_type, action, amount, currency, pricing_snapshot, payload, status, next_attempt_at
)
SELECT id, 'task.billing.release', 'release', reservation_amount, billing_currency,
       pricing_snapshot, $2::jsonb, 'pending', now()
FROM gateway_tasks
WHERE id = $1::uuid
  AND run_mode IN ('production', 'acceptance', 'acceptance_canary')
  AND gateway_user_id IS NOT NULL
  AND reservation_amount > 0
ON CONFLICT (task_id, event_type) DO NOTHING`, taskID, string(payloadJSON)); err != nil {
			return TaskAdmissionReapResult{}, err
		}
		eventType := "task.cancelled"
		if status == "failed" {
			eventType = "task.failed"
		}
		var eventID string
		var eventSeq int64
		if err := tx.QueryRow(ctx, `
WITH next_seq AS (
  SELECT COALESCE(MAX(seq), 0) + 1 AS seq
  FROM gateway_task_events
  WHERE task_id = $1::uuid
)
INSERT INTO gateway_task_events (
  task_id, seq, event_type, status, phase, progress, message, payload, simulated
)
SELECT $1::uuid, next_seq.seq, $2, $3, 'admission', 0, $4, '{}'::jsonb,
       task.run_mode = 'simulation'
FROM next_seq
JOIN gateway_tasks task ON task.id = $1::uuid
RETURNING id::text, seq`, taskID, eventType, status, message).Scan(&eventID, &eventSeq); err != nil {
			return TaskAdmissionReapResult{}, err
		}
		var acceptanceCallbackURL string
		if err := tx.QueryRow(ctx, `
SELECT COALESCE((
  SELECT run.callback_url
  FROM gateway_tasks task
  JOIN gateway_acceptance_runs run ON run.id = task.acceptance_run_id
  WHERE task.id = $1::uuid
), '')`, taskID).Scan(&acceptanceCallbackURL); err != nil {
			return TaskAdmissionReapResult{}, err
		}
		callbackURL := defaultCallbackURL
		if strings.TrimSpace(acceptanceCallbackURL) != "" {
			callbackURL = strings.TrimSpace(acceptanceCallbackURL)
		}
		if callbackURL != "" {
			if _, err := tx.Exec(ctx, `
INSERT INTO gateway_task_callback_outbox (task_id, event_id, seq, callback_url, payload)
VALUES ($1::uuid, $2::uuid, $3, $4, '{}'::jsonb)
ON CONFLICT (task_id, seq, callback_url) DO NOTHING`, taskID, eventID, eventSeq, callbackURL); err != nil {
				return TaskAdmissionReapResult{}, err
			}
		}
		reaped++
	}
	if err := tx.Commit(ctx); err != nil {
		return TaskAdmissionReapResult{}, err
	}
	if reaped > 0 {
		s.notifyTaskAdmissionBestEffort(ctx, "*")
	}
	return result, nil
}

// ListenTaskAdmissionNotifications holds one pooled PostgreSQL connection for
// the process and forwards NOTIFY payloads as wake-up hints. Callers must still
// retain periodic polling because LISTEN notifications are not durable.
func (s *Store) ListenTaskAdmissionNotifications(ctx context.Context, onNotification func(string)) error {
	connection, err := s.pool.Acquire(ctx)
	if err != nil {
		return err
	}
	defer connection.Release()
	if _, err := connection.Exec(ctx, `LISTEN gateway_task_admission`); err != nil {
		return err
	}
	for {
		notification, err := connection.Conn().WaitForNotification(ctx)
		if err != nil {
			return err
		}
		if onNotification != nil {
			onNotification(notification.Payload)
		}
	}
}

func validateTaskAdmissionInput(input TaskAdmissionInput) error {
	if strings.TrimSpace(input.TaskID) == "" || strings.TrimSpace(input.PlatformID) == "" || strings.TrimSpace(input.PlatformModelID) == "" {
		return errors.New("task admission requires task, platform, and platform model")
	}
	if input.Mode != "sync" && input.Mode != "async" {
		return errors.New("task admission mode must be sync or async")
	}
	if input.Mode == "sync" && strings.TrimSpace(input.WaiterID) == "" {
		return errors.New("synchronous task admission requires a waiter ID")
	}
	return nil
}

func normalizedAdmissionScopes(scopes []AdmissionScope) []AdmissionScope {
	out := make([]AdmissionScope, 0, len(scopes))
	seen := make(map[string]struct{})
	for _, scope := range scopes {
		scope.ScopeType = strings.TrimSpace(scope.ScopeType)
		scope.ScopeKey = strings.TrimSpace(scope.ScopeKey)
		if scope.ScopeType == "" || scope.ScopeKey == "" {
			continue
		}
		if scope.Amount <= 0 {
			scope.Amount = 1
		}
		if scope.LeaseTTLSeconds <= 0 {
			scope.LeaseTTLSeconds = 120
		}
		key := admissionLockKey(scope)
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		out = append(out, scope)
	}
	sort.Slice(out, func(i, j int) bool {
		return admissionLockKey(out[i]) < admissionLockKey(out[j])
	})
	return out
}

func admissionLockKey(scope AdmissionScope) string {
	return fmt.Sprintf("task-admission:%d:%s:%d:%s", len(scope.ScopeType), scope.ScopeType, len(scope.ScopeKey), scope.ScopeKey)
}

func admissionQueueRegistrationLockKey(scope AdmissionScope) string {
	return fmt.Sprintf("task-admission-queue:%d:%s:%d:%s", len(scope.ScopeType), scope.ScopeType, len(scope.ScopeKey), scope.ScopeKey)
}

type admissionScopeState struct {
	Scope               AdmissionScope
	Active              float64
	Waiting             int
	Saturated           bool
	NextLeaseExpiration time.Time
}

func admissionScopeStatesTx(ctx context.Context, tx pgx.Tx, scopes []AdmissionScope) ([]admissionScopeState, error) {
	states := make([]admissionScopeState, 0, len(scopes))
	for _, scope := range scopes {
		state := admissionScopeState{Scope: scope}
		if scope.ConcurrentLimit > 0 {
			if err := tx.QueryRow(ctx, `
SELECT COALESCE(SUM(lease_value), 0)::float8,
       COALESCE(MIN(expires_at), statement_timestamp() + interval '1 second')
FROM gateway_concurrency_leases
WHERE scope_type = $1
  AND scope_key = $2
  AND released_at IS NULL
  AND expires_at > statement_timestamp()`, scope.ScopeType, scope.ScopeKey).Scan(&state.Active, &state.NextLeaseExpiration); err != nil {
				return nil, err
			}
			state.Saturated = state.Active+scope.Amount > scope.ConcurrentLimit
		}
		waiting, err := admissionScopeWaitingTx(ctx, tx, scope)
		if err != nil {
			return nil, err
		}
		state.Waiting = waiting
		states = append(states, state)
	}
	return states, nil
}

func admissionScopeWaitingTx(ctx context.Context, tx pgx.Tx, scope AdmissionScope) (int, error) {
	var waiting int
	switch scope.ScopeType {
	case "platform_model":
		err := tx.QueryRow(ctx, `
SELECT COUNT(*)
FROM gateway_task_admissions
WHERE status = 'waiting' AND platform_model_id::text = $1`, scope.ScopeKey).Scan(&waiting)
		return waiting, err
	case "user_group":
		err := tx.QueryRow(ctx, `
SELECT COUNT(*)
FROM gateway_task_admissions
WHERE status = 'waiting' AND user_group_id::text = $1`, scope.ScopeKey).Scan(&waiting)
		return waiting, err
	case "worker_capacity":
		err := tx.QueryRow(ctx, `
SELECT COUNT(*)
FROM gateway_task_admissions
WHERE status = 'waiting' AND mode = 'async'`).Scan(&waiting)
		return waiting, err
	default:
		return 0, nil
	}
}

func admissionDeadline(scopes []AdmissionScope) time.Time {
	maxWait := 0
	for _, scope := range scopes {
		if scope.QueueLimit <= 0 {
			continue
		}
		wait := scope.MaxWaitSeconds
		if wait <= 0 {
			wait = DefaultQueueMaxWaitSeconds
		}
		if maxWait == 0 || wait < maxWait {
			maxWait = wait
		}
	}
	if maxWait == 0 {
		// A task may be registered before a River execution slot is available
		// even when business queueing is disabled. Keep that infrastructure wait
		// separate from the policy's default 600-second queue timeout.
		return time.Now().Add(executionSlotWaitMax)
	}
	return time.Now().Add(time.Duration(maxWait) * time.Second)
}

func concurrencyAdmissionError(state admissionScopeState) error {
	return &RateLimitExceededError{
		ScopeType:     state.Scope.ScopeType,
		ScopeKey:      state.Scope.ScopeKey,
		ScopeName:     state.Scope.ScopeName,
		ScopeMetadata: state.Scope.ScopeMetadata,
		Metric:        "concurrent",
		Limit:         state.Scope.ConcurrentLimit,
		Amount:        state.Scope.Amount,
		Current:       state.Active,
		Used:          state.Active,
		Projected:     state.Active + state.Scope.Amount,
		ResetAt:       state.NextLeaseExpiration,
		Policy:        state.Scope.Policy,
		Message:       "concurrency limit is saturated and queueing is disabled",
		RetryAfter:    queueRetryAfter(state.NextLeaseExpiration),
		Retryable:     true,
	}
}

func queueFullAdmissionError(state admissionScopeState) error {
	return &RateLimitExceededError{
		ScopeType:     state.Scope.ScopeType,
		ScopeKey:      state.Scope.ScopeKey,
		ScopeName:     state.Scope.ScopeName,
		ScopeMetadata: state.Scope.ScopeMetadata,
		Metric:        "queue_size",
		Reason:        "queue_full",
		Limit:         float64(state.Scope.QueueLimit),
		Amount:        1,
		Current:       float64(state.Waiting),
		Used:          float64(state.Waiting),
		Projected:     float64(state.Waiting + 1),
		ResetAt:       state.NextLeaseExpiration,
		Policy:        state.Scope.Policy,
		Message:       "task admission queue is full",
		RetryAfter:    queueRetryAfter(state.NextLeaseExpiration),
		Retryable:     true,
		QueueDepth:    state.Waiting,
		QueueLimit:    state.Scope.QueueLimit,
	}
}

func queueRetryAfter(next time.Time) time.Duration {
	if next.IsZero() {
		return time.Second
	}
	wait := time.Until(next)
	if wait < time.Second {
		return time.Second
	}
	if wait > 30*time.Second {
		return 30 * time.Second
	}
	return wait
}

func insertTaskAdmissionTx(ctx context.Context, tx pgx.Tx, input TaskAdmissionInput, deadline time.Time) (TaskAdmission, error) {
	waiterID := strings.TrimSpace(input.WaiterID)
	scopeSnapshot, err := json.Marshal(normalizedAdmissionScopes(input.Scopes))
	if err != nil {
		return TaskAdmission{}, fmt.Errorf("marshal task admission scopes: %w", err)
	}
	var waiterLease any
	if input.Mode == "sync" {
		waiterLease = time.Now().Add(admissionWaiterLeaseTTL)
	}
	row := tx.QueryRow(ctx, `
INSERT INTO gateway_task_admissions (
  task_id, platform_id, platform_model_id, user_group_id, queue_key, mode, status,
  priority, wait_deadline_at, waiter_id, waiter_lease_expires_at, scope_snapshot
)
VALUES (
  $1::uuid, $2::uuid, $3::uuid, NULLIF($4, '')::uuid, $5, $6, 'waiting',
  $7, $8, NULLIF($9, ''), $10, $11::jsonb
)
ON CONFLICT (task_id) DO UPDATE
SET waiter_id = CASE WHEN gateway_task_admissions.mode = 'sync' THEN EXCLUDED.waiter_id ELSE gateway_task_admissions.waiter_id END,
    waiter_lease_expires_at = CASE WHEN gateway_task_admissions.mode = 'sync' THEN EXCLUDED.waiter_lease_expires_at ELSE gateway_task_admissions.waiter_lease_expires_at END,
    scope_snapshot = CASE
      WHEN gateway_task_admissions.scope_snapshot IS NULL OR gateway_task_admissions.scope_snapshot = '[]'::jsonb THEN EXCLUDED.scope_snapshot
      ELSE gateway_task_admissions.scope_snapshot
    END,
    updated_at = now()
RETURNING task_id::text, platform_id::text, platform_model_id::text,
          COALESCE(user_group_id::text, ''), queue_key, mode, status, priority,
          enqueued_at, wait_deadline_at, admitted_at,
          COALESCE(waiter_id, ''), waiter_lease_expires_at,
          scope_snapshot, reselect_requested_at, dispatch_next_at,
          dispatch_failure_count, COALESCE(dispatch_last_outcome, ''),
          COALESCE(dispatch_last_error_code, ''), dispatch_last_error_at`,
		input.TaskID, input.PlatformID, input.PlatformModelID, input.UserGroupID,
		input.QueueKey, input.Mode, input.Priority, deadline, waiterID, waiterLease, scopeSnapshot)
	return scanTaskAdmission(row)
}

func renewTaskAdmissionWaiterTx(ctx context.Context, tx pgx.Tx, taskID string, waiterID string) (TaskAdmission, error) {
	row := tx.QueryRow(ctx, `
UPDATE gateway_task_admissions
SET waiter_id = $2,
    waiter_lease_expires_at = now() + $3::interval,
    updated_at = now()
WHERE task_id = $1::uuid
  AND mode = 'sync'
  AND status = 'waiting'
RETURNING task_id::text, platform_id::text, platform_model_id::text,
          COALESCE(user_group_id::text, ''), queue_key, mode, status, priority,
          enqueued_at, wait_deadline_at, admitted_at,
          COALESCE(waiter_id, ''), waiter_lease_expires_at,
          scope_snapshot, reselect_requested_at, dispatch_next_at,
          dispatch_failure_count, COALESCE(dispatch_last_outcome, ''),
          COALESCE(dispatch_last_error_code, ''), dispatch_last_error_at`,
		taskID, waiterID, admissionWaiterLeaseTTL.String())
	return scanTaskAdmission(row)
}

func markTaskAdmissionAdmittedTx(ctx context.Context, tx pgx.Tx, taskID string) (TaskAdmission, error) {
	row := tx.QueryRow(ctx, `
UPDATE gateway_task_admissions
SET status = 'admitted',
    admitted_at = now(),
    waiter_lease_expires_at = NULL,
    dispatch_failure_count = 0,
    dispatch_last_outcome = 'admitted',
    dispatch_last_error_code = NULL,
    dispatch_last_error_at = NULL,
    updated_at = now()
WHERE task_id = $1::uuid AND status = 'waiting'
RETURNING task_id::text, platform_id::text, platform_model_id::text,
          COALESCE(user_group_id::text, ''), queue_key, mode, status, priority,
          enqueued_at, wait_deadline_at, admitted_at,
          COALESCE(waiter_id, ''), waiter_lease_expires_at,
          scope_snapshot, reselect_requested_at, dispatch_next_at,
          dispatch_failure_count, COALESCE(dispatch_last_outcome, ''),
          COALESCE(dispatch_last_error_code, ''), dispatch_last_error_at`,
		taskID)
	return scanTaskAdmission(row)
}

func loadTaskAdmissionTx(ctx context.Context, tx pgx.Tx, taskID string) (TaskAdmission, bool, error) {
	return loadTaskAdmission(ctx, tx, taskID)
}

type admissionQuerier interface {
	QueryRow(context.Context, string, ...any) pgx.Row
}

func loadTaskAdmission(ctx context.Context, q admissionQuerier, taskID string) (TaskAdmission, bool, error) {
	row := q.QueryRow(ctx, `
SELECT task_id::text, platform_id::text, platform_model_id::text,
       COALESCE(user_group_id::text, ''), queue_key, mode, status, priority,
       enqueued_at, wait_deadline_at, admitted_at,
       COALESCE(waiter_id, ''), waiter_lease_expires_at,
       scope_snapshot, reselect_requested_at, dispatch_next_at,
       dispatch_failure_count, COALESCE(dispatch_last_outcome, ''),
       COALESCE(dispatch_last_error_code, ''), dispatch_last_error_at
FROM gateway_task_admissions
WHERE task_id = $1::uuid`, taskID)
	admission, err := scanTaskAdmission(row)
	if errors.Is(err, pgx.ErrNoRows) {
		return TaskAdmission{}, false, nil
	}
	return admission, err == nil, err
}

func scanTaskAdmission(row pgx.Row) (TaskAdmission, error) {
	var admission TaskAdmission
	var admittedAt sql.NullTime
	var waiterLeaseExpires sql.NullTime
	var reselectRequestedAt sql.NullTime
	var dispatchLastErrorAt sql.NullTime
	var scopeSnapshot []byte
	err := row.Scan(
		&admission.TaskID,
		&admission.PlatformID,
		&admission.PlatformModelID,
		&admission.UserGroupID,
		&admission.QueueKey,
		&admission.Mode,
		&admission.Status,
		&admission.Priority,
		&admission.EnqueuedAt,
		&admission.WaitDeadlineAt,
		&admittedAt,
		&admission.WaiterID,
		&waiterLeaseExpires,
		&scopeSnapshot,
		&reselectRequestedAt,
		&admission.DispatchNextAt,
		&admission.DispatchFailureCount,
		&admission.DispatchLastOutcome,
		&admission.DispatchLastErrorCode,
		&dispatchLastErrorAt,
	)
	if admittedAt.Valid {
		admission.AdmittedAt = admittedAt.Time
	}
	if waiterLeaseExpires.Valid {
		admission.WaiterLeaseExpires = waiterLeaseExpires.Time
	}
	if reselectRequestedAt.Valid {
		admission.ReselectRequestedAt = reselectRequestedAt.Time
	}
	if dispatchLastErrorAt.Valid {
		admission.DispatchLastErrorAt = dispatchLastErrorAt.Time
	}
	if err == nil && len(scopeSnapshot) > 0 {
		if unmarshalErr := json.Unmarshal(scopeSnapshot, &admission.Scopes); unmarshalErr != nil {
			admission.SnapshotInvalid = true
		}
	}
	return admission, err
}

func isTaskAdmissionHeadTx(ctx context.Context, tx pgx.Tx, admission TaskAdmission, scopes []AdmissionScope) (bool, []TaskAdmissionBlocker, error) {
	for _, scope := range scopes {
		var head string
		switch scope.ScopeType {
		case "platform_model":
			if err := tx.QueryRow(ctx, `
SELECT task_id::text
FROM gateway_task_admissions
WHERE platform_model_id = $1::uuid
  AND status = 'waiting'
  AND (mode <> 'async' OR dispatch_next_at <= statement_timestamp())
ORDER BY priority ASC, enqueued_at ASC, task_id ASC
LIMIT 1`, admission.PlatformModelID).Scan(&head); err != nil {
				return false, nil, err
			}
		case "user_group":
			if admission.UserGroupID == "" {
				continue
			}
			if err := tx.QueryRow(ctx, `
SELECT task_id::text
FROM gateway_task_admissions
WHERE user_group_id = $1::uuid
  AND status = 'waiting'
  AND (mode <> 'async' OR dispatch_next_at <= statement_timestamp())
ORDER BY priority ASC, enqueued_at ASC, task_id ASC
LIMIT 1`, admission.UserGroupID).Scan(&head); err != nil {
				return false, nil, err
			}
		default:
			continue
		}
		if head != admission.TaskID {
			return false, []TaskAdmissionBlocker{{
				Reason:    "not_queue_head",
				ScopeType: scope.ScopeType,
				ScopeKey:  scope.ScopeKey,
			}}, nil
		}
	}
	return true, nil, nil
}

func expireTaskAdmissionTx(ctx context.Context, tx pgx.Tx, taskID string) error {
	if _, err := tx.Exec(ctx, `DELETE FROM gateway_task_admissions WHERE task_id = $1::uuid`, taskID); err != nil {
		return err
	}
	_, err := tx.Exec(ctx, `
UPDATE gateway_tasks
SET status = 'failed',
    error = 'task admission queue wait timed out',
    error_code = 'queue_timeout',
    error_message = 'task admission queue wait timed out',
    finished_at = now(),
    updated_at = now()
WHERE id = $1::uuid AND status = 'queued'`, taskID)
	return err
}

// notifyTaskAdmission publishes a non-durable wake-up hint only after the
// corresponding state transaction has committed. Keeping NOTIFY out of
// synchronous replication transactions prevents its database object lock from
// serializing unrelated admission, execution, and release commits.
func (s *Store) notifyTaskAdmission(ctx context.Context, taskID string) error {
	taskID = strings.TrimSpace(taskID)
	if taskID == "" {
		return nil
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer rollbackTransaction(tx)
	if _, err := tx.Exec(ctx, `SET LOCAL synchronous_commit = off`); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `SELECT pg_notify($1, $2)`, admissionNotifyChannel, taskID); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (s *Store) notifyTaskAdmissionBestEffort(ctx context.Context, taskID string) {
	if strings.TrimSpace(taskID) == "" {
		return
	}
	notifyCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), admissionNotifyTimeout)
	defer cancel()
	_ = s.notifyTaskAdmission(notifyCtx, taskID)
}
