package connector_inbound

import (
	"context"
	"database/sql"
	"log"
	"math/rand/v2"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	"golang.org/x/sync/singleflight"
)

const (
	defaultClaimLimiterDelay             = 5 * time.Second
	defaultClaimLimiterDBInUseRatio      = 0.90
	defaultClaimLimiterMaxProcessingRows = 32
	defaultClaimLimiterPressureTimeout   = 500 * time.Millisecond
	defaultClaimLimiterPressureCacheTTL  = time.Second
	defaultStaleRecoveryInterval         = time.Second

	// db-pool backpressure is softer than the fail-closed budget. One limiter is
	// shared across all shards of a pack, so returning the full 5s default froze
	// new-claim AND stale-reclaim for the whole fleet in 5s lockstep. These short
	// jittered values let shards re-probe at staggered sub-second times so forward
	// progress resumes within sub-second once pool pressure eases. They are
	// operational tuning (the DB pool is a process-wide infra concern, not pack
	// shape), so they stay as Go constants here rather than per-pack config — no
	// pack needs per-pack DB-pool tuning. This limiter's ownership is default-L1
	// per #3102 §C (no demonstrated second consumer); the constants are an
	// in-place tuning change, not an L0-standing claim.
	dbPoolThrottleBaseDelay = 300 * time.Millisecond
	dbPoolThrottleJitter    = 120 * time.Millisecond // ~40% of base
	dbPoolThrottleMinDelay  = 200 * time.Millisecond
	dbPoolThrottleMaxDelay  = 500 * time.Millisecond

	// perWorkerBudget is the queue-scope in-flight claim budget contributed by
	// each inbound worker shard above the legacy floor. See
	// MaxProcessingRowsForWorkers.
	perWorkerBudget = 4
)

// MaxProcessingRowsForWorkers returns the in-flight claim budget the supervisor
// should configure for each limiter queue scope at a pack running workerCount
// shards.
//
// The budget scales with the shard count (workerCount * perWorkerBudget). At
// the default of 8 workers, 8*4 = 32 keeps the legacy ceiling. Lower worker
// counts intentionally lower the queue pressure ceiling too; otherwise reducing
// NEXUSCRM_CONNECTOR_INBOUND_WORKERS_PER_PACK would leave the DB-facing
// processing budget unchanged.
func MaxProcessingRowsForWorkers(workerCount int) int {
	if workerCount < 1 {
		workerCount = 1
	}
	budget := workerCount * perWorkerBudget
	return budget
}

type ClaimLimiter interface {
	BeforeClaim(ctx context.Context, req ClaimLimiterRequest) (ClaimLimiterDecision, error)
}

type ClaimLimiterRequest struct {
	Namespace        string
	PackInstallID    string
	WorkerID         string
	ClaimPath        ClaimPath
	QueueClasses     []string
	BatchSize        int
	AdmissionLatency time.Duration
	EnqueueLatency   time.Duration
}

type ClaimLimiterDecision struct {
	Allow         bool
	Delay         time.Duration
	Reason        string
	MaxClaims     int
	ReservationID int64
	// StaleReclaimOnly, set on an Allow:true "limp" decision, restricts the granted
	// claim to reclaiming ONE stale-processing row: the worker routes to the
	// stale-only claim/peek path (ReclaimStaleNext / PeekStaleBatch), whose
	// candidate SELECT matches only stale-processing rows. Re-leasing a row that is
	// already 'processing' leaves ProcessingCount flat, so a limp claim cannot push
	// past maxProcessingRows even though the limp branch carries no
	// reservation/snapshot — that is why this is safe without the ceiling accounting
	// the #3082 fail-open lacked. The fan-out claim path's CAS is also stale-only, so
	// a candidate a concurrent reclaim drifts off 'processing' is simply retried. The
	// single-worker peek path's ClaimByID CAS keeps the full predicate; a rare
	// peek→claim drift may re-lease that SAME row as a failed retry — still one
	// already-counted row, so still no over-admit. This is the bounded degrade the
	// fail-closed branches fall back to so a wedged shard fleet keeps recovering
	// stranded leases.
	StaleReclaimOnly bool
}

type ClaimLimiterStatus struct {
	Allow       bool
	Delay       time.Duration
	Reason      string
	MaxClaims   int
	Reserved    int
	EvaluatedAt time.Time
	DBPool      sql.DBStats
	Queue       connector_events.ClaimPressureSnapshot
	Allocator   AdaptiveBudgetDecision
}

type DBStatsProvider interface {
	Stats() sql.DBStats
}

// ProcessingPressureProvider yields the cheap capacity inputs (total + stale
// processing counts) the limiter needs, without the claim-ordering joins that
// make the full pressure query time out under backlog. Keeping the limiter on
// this cheap query removes the death spiral at its root: there is no expensive
// probe to time out, so no fail-open degrade is needed.
type ProcessingPressureProvider interface {
	ProcessingPressure(ctx context.Context, installedPackID string, queueClasses []string) (connector_events.ClaimPressureSnapshot, error)
}

type SQLClaimLimiterConfig struct {
	DBStats                DBStatsProvider
	Pressure               ProcessingPressureProvider
	BudgetAllocator        AdaptiveBudgetAllocator
	Now                    func() time.Time
	Profile                bool
	Delay                  time.Duration
	PressureTimeout        time.Duration
	PressureCacheTTL       time.Duration
	MaxDBInUseRatio        float64
	MaxProcessingRows      int
	GlobalProcessingBudget bool
}

type SQLClaimLimiter struct {
	dbStats                DBStatsProvider
	pressure               ProcessingPressureProvider
	now                    func() time.Time
	delay                  time.Duration
	pressureTimeout        time.Duration
	pressureCacheTTL       time.Duration
	maxDBInUseRatio        float64
	maxProcessingRows      int
	globalProcessingBudget bool
	profile                bool
	budgetAllocator        AdaptiveBudgetAllocator

	mu            sync.Mutex
	lastWaitCount int64
	lastStatus    ClaimLimiterStatus
	lastLogKey    string
	nextReserveID int64
	reservations  map[int64]claimReservation
	pressureCache map[string]claimPressureCacheEntry
	staleRecovery map[string]time.Time
	pressureGroup singleflight.Group
}

type claimReservation struct {
	keys []string
	n    int
}

type claimPressureCacheEntry struct {
	snapshot  connector_events.ClaimPressureSnapshot
	expiresAt time.Time
}

func NewSQLClaimLimiter(cfg SQLClaimLimiterConfig) *SQLClaimLimiter {
	now := cfg.Now
	if now == nil {
		now = time.Now
	}
	delay := cfg.Delay
	if delay <= 0 {
		delay = defaultClaimLimiterDelay
	}
	pressureTimeout := cfg.PressureTimeout
	if pressureTimeout <= 0 {
		pressureTimeout = defaultClaimLimiterPressureTimeout
	}
	pressureCacheTTL := cfg.PressureCacheTTL
	if pressureCacheTTL <= 0 {
		pressureCacheTTL = defaultClaimLimiterPressureCacheTTL
	}
	maxDBInUseRatio := cfg.MaxDBInUseRatio
	if maxDBInUseRatio <= 0 || maxDBInUseRatio > 1 {
		maxDBInUseRatio = defaultClaimLimiterDBInUseRatio
	}
	maxProcessingRows := cfg.MaxProcessingRows
	if maxProcessingRows <= 0 {
		maxProcessingRows = defaultClaimLimiterMaxProcessingRows
	}
	budgetAllocator := cfg.BudgetAllocator
	if budgetAllocator == nil {
		budgetAllocator = DefaultAdaptiveBudgetAllocator{}
	}
	return &SQLClaimLimiter{
		dbStats:                cfg.DBStats,
		pressure:               cfg.Pressure,
		now:                    now,
		delay:                  delay,
		pressureTimeout:        pressureTimeout,
		pressureCacheTTL:       pressureCacheTTL,
		maxDBInUseRatio:        maxDBInUseRatio,
		maxProcessingRows:      maxProcessingRows,
		globalProcessingBudget: cfg.GlobalProcessingBudget,
		profile:                cfg.Profile,
		budgetAllocator:        budgetAllocator,
		reservations:           make(map[int64]claimReservation),
		pressureCache:          make(map[string]claimPressureCacheEntry),
		staleRecovery:          make(map[string]time.Time),
	}
}

func (l *SQLClaimLimiter) BeforeClaim(ctx context.Context, req ClaimLimiterRequest) (ClaimLimiterDecision, error) {
	if l == nil {
		return ClaimLimiterDecision{Allow: true}, nil
	}
	now := l.now()
	var stats sql.DBStats
	if l.dbStats != nil {
		stats = l.dbStats.Stats()
	}
	if decision, throttled := l.dbDecision(stats); throttled {
		status := ClaimLimiterStatus{Allow: decision.Allow, Delay: decision.Delay, Reason: decision.Reason, EvaluatedAt: now, DBPool: stats}
		l.recordStatus(status)
		l.logProfile(req, status, decision)
		return decision, nil
	}

	var snapshot connector_events.ClaimPressureSnapshot
	if l.pressure != nil {
		var err error
		snapshot, err = l.claimPressureSnapshot(ctx, req)
		if err != nil {
			// ProcessingPressure is the cheap capacity probe. The legacy Event
			// path stays fail-closed to stale reclaim. WorkItem workers use the
			// reservation-backed min-progress path so transient queue-pressure
			// read failures cannot starve fresh lane heads after prior completions.
			var decision ClaimLimiterDecision
			var allocatorDecision AdaptiveBudgetDecision
			if normalizeClaimPath(req.ClaimPath) == ClaimPathWorkItem {
				decision, allocatorDecision = l.limpFreshClaim(req, "queue_pressure_unavailable")
			} else {
				decision = l.limpReclaim("queue_pressure_unavailable")
				allocatorDecision = AdaptiveBudgetDecision{
					Reason:       decision.Reason,
					MaxClaims:    decision.MaxClaims,
					QueueClasses: append([]string(nil), req.QueueClasses...),
				}
			}
			status := ClaimLimiterStatus{Allow: decision.Allow, Delay: decision.Delay, Reason: decision.Reason, MaxClaims: decision.MaxClaims, Reserved: l.currentReserved(req), EvaluatedAt: now, DBPool: stats, Queue: snapshot, Allocator: allocatorDecision}
			l.recordStatus(status)
			l.logProfile(req, status, decision)
			return decision, nil
		}
	}
	decision, throttled, allocatorDecision := l.processingDecision(req, snapshot, stats)
	status := ClaimLimiterStatus{
		Allow:       decision.Allow || !throttled,
		Delay:       decision.Delay,
		Reason:      decision.Reason,
		MaxClaims:   decision.MaxClaims,
		Reserved:    l.currentReserved(req),
		EvaluatedAt: now,
		DBPool:      stats,
		Queue:       snapshot,
		Allocator:   allocatorDecision,
	}
	l.recordStatus(status)
	l.logProfile(req, status, decision)
	return decision, nil
}

func (l *SQLClaimLimiter) claimPressureSnapshot(ctx context.Context, req ClaimLimiterRequest) (connector_events.ClaimPressureSnapshot, error) {
	req = l.budgetScopeRequest(req)
	key := claimLimiterReservationKey(req)
	now := l.now()
	if snapshot, ok := l.cachedClaimPressure(key, now); ok {
		l.logPressureProbeProfile(req, "cache_hit", 0, snapshot, nil)
		return snapshot, nil
	}
	value, err, _ := l.pressureGroup.Do(key, func() (interface{}, error) {
		now := l.now()
		if snapshot, ok := l.cachedClaimPressure(key, now); ok {
			l.logPressureProbeProfile(req, "cache_hit", 0, snapshot, nil)
			return snapshot, nil
		}
		probeCtx := ctx
		cancel := func() {}
		if l.pressureTimeout > 0 {
			probeCtx, cancel = context.WithTimeout(ctx, l.pressureTimeout)
		}
		defer cancel()
		started := time.Now()
		snapshot, err := l.pressure.ProcessingPressure(probeCtx, req.PackInstallID, req.QueueClasses)
		elapsed := time.Since(started)
		if err != nil {
			l.logPressureProbeProfile(req, "query_error", elapsed, connector_events.ClaimPressureSnapshot{}, err)
			return connector_events.ClaimPressureSnapshot{}, err
		}
		l.storeClaimPressure(key, snapshot, l.now())
		l.logPressureProbeProfile(req, "query_ok", elapsed, snapshot, nil)
		return snapshot, nil
	})
	if err != nil {
		return connector_events.ClaimPressureSnapshot{}, err
	}
	snapshot, _ := value.(connector_events.ClaimPressureSnapshot)
	return snapshot, nil
}

func (l *SQLClaimLimiter) logPressureProbeProfile(req ClaimLimiterRequest, outcome string, elapsed time.Duration, snapshot connector_events.ClaimPressureSnapshot, err error) {
	if l == nil || !l.profile {
		return
	}
	queueClasses := strings.Join(req.QueueClasses, ",")
	if queueClasses == "" {
		queueClasses = "all"
	}
	if err != nil {
		log.Printf("event=connector_inbound_pressure_probe_profile elapsed_ms=%d namespace=%s installed_pack_id=%s worker_id=%s queue_classes=%s outcome=%s active_processing=%d stale_processing=%d error=%q",
			elapsed.Milliseconds(),
			req.Namespace,
			req.PackInstallID,
			req.WorkerID,
			queueClasses,
			outcome,
			snapshot.ProcessingCount,
			snapshot.StaleProcessingCount,
			err.Error(),
		)
		return
	}
	log.Printf("event=connector_inbound_pressure_probe_profile elapsed_ms=%d namespace=%s installed_pack_id=%s worker_id=%s queue_classes=%s outcome=%s active_processing=%d stale_processing=%d",
		elapsed.Milliseconds(),
		req.Namespace,
		req.PackInstallID,
		req.WorkerID,
		queueClasses,
		outcome,
		snapshot.ProcessingCount,
		snapshot.StaleProcessingCount,
	)
}

func (l *SQLClaimLimiter) ReleaseClaimPermit(decision ClaimLimiterDecision, _ int) {
	if l == nil || decision.ReservationID == 0 {
		return
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	if _, ok := l.reservations[decision.ReservationID]; !ok {
		return
	}
	delete(l.reservations, decision.ReservationID)
}

func (l *SQLClaimLimiter) Status() ClaimLimiterStatus {
	if l == nil {
		return ClaimLimiterStatus{Allow: true}
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	status := l.lastStatus
	status.Queue.QueueClasses = append([]string(nil), status.Queue.QueueClasses...)
	status.Allocator.QueueClasses = append([]string(nil), status.Allocator.QueueClasses...)
	status.Allocator.Signals = append([]string(nil), status.Allocator.Signals...)
	return status
}

func (l *SQLClaimLimiter) processingDecision(req ClaimLimiterRequest, snapshot connector_events.ClaimPressureSnapshot, stats sql.DBStats) (ClaimLimiterDecision, bool, AdaptiveBudgetDecision) {
	allocatorQueueClasses := append([]string(nil), req.QueueClasses...)
	req = l.budgetScopeRequest(req)
	if l.maxProcessingRows <= 0 {
		return ClaimLimiterDecision{Allow: true}, false, AdaptiveBudgetDecision{Reason: "disabled"}
	}
	want := req.BatchSize
	if want <= 0 || want > l.maxProcessingRows {
		want = l.maxProcessingRows
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	reservationKeys := claimLimiterReservationKeys(req)
	reserved := l.currentReservedForKeysLocked(reservationKeys)
	activeProcessing := snapshot.ProcessingCount
	if normalizeClaimPath(req.ClaimPath) == ClaimPathWorkItem {
		// An expired WorkItem lease is recovery backlog, not live execution. It must
		// not consume the global fresh-admission budget and freeze unrelated lanes.
		// Same-lane safety remains owned by the lane-head predicate, while the
		// stale-only path below keeps expired heads recoverable at fresh capacity.
		staleProcessing := min(snapshot.StaleProcessingCount, activeProcessing)
		activeProcessing -= staleProcessing
	}
	available := l.maxProcessingRows - activeProcessing - reserved
	if normalizeClaimPath(req.ClaimPath) == ClaimPathWorkItem &&
		snapshot.StaleProcessingCount > 0 && l.staleRecoveryDueLocked(req) {
		return l.limpReclaim("stale_recovery_due"), false, AdaptiveBudgetDecision{
			Reason:              "stale_recovery_due",
			Requested:           want,
			ConfiguredMaxRows:   l.maxProcessingRows,
			ActiveProcessing:    activeProcessing,
			StaleProcessing:     snapshot.StaleProcessingCount,
			Reserved:            reserved,
			AvailableBeforeTrim: available,
			MaxClaims:           1,
			QueueClasses:        allocatorQueueClasses,
		}
	}
	if available <= 0 {
		// available<=0 has two causes: (a) active processing has itself reached the
		// cap — a genuine wedge; or (b) in-flight reservations have merely booked the
		// remaining budget — those claims will complete and release it. The legacy
		// Event path keeps its at-cap stale limp. WorkItem recovery is rate-limited
		// above so one recovery attempt cannot turn the whole scheduler shape into
		// stale-only mode. In every other case, stay fail-closed.
		if normalizeClaimPath(req.ClaimPath) != ClaimPathWorkItem &&
			activeProcessing >= l.maxProcessingRows && snapshot.StaleProcessingCount > 0 {
			return l.limpReclaim("processing_capacity_limp"), false, AdaptiveBudgetDecision{
				Reason:              "stale_reclaim_limp",
				Requested:           want,
				ConfiguredMaxRows:   l.maxProcessingRows,
				ActiveProcessing:    activeProcessing,
				StaleProcessing:     snapshot.StaleProcessingCount,
				Reserved:            reserved,
				AvailableBeforeTrim: available,
				MaxClaims:           1,
				QueueClasses:        allocatorQueueClasses,
			}
		}
		decision := l.throttle("processing_capacity_full")
		if reserved > 0 && activeProcessing < l.maxProcessingRows {
			decision.Delay = jitteredDBDelay()
		}
		return decision, true, AdaptiveBudgetDecision{
			Reason:              "processing_capacity_full",
			Requested:           want,
			ConfiguredMaxRows:   l.maxProcessingRows,
			ActiveProcessing:    activeProcessing,
			StaleProcessing:     snapshot.StaleProcessingCount,
			Reserved:            reserved,
			AvailableBeforeTrim: available,
			MaxClaims:           0,
			QueueClasses:        allocatorQueueClasses,
		}
	}
	allocatorDecision := l.budgetAllocator.Allocate(AdaptiveBudgetInput{
		Requested:           want,
		ConfiguredMaxRows:   l.maxProcessingRows,
		ActiveProcessing:    activeProcessing,
		StaleProcessing:     snapshot.StaleProcessingCount,
		Reserved:            reserved,
		AvailableBeforeTrim: available,
		DBPool:              stats,
		QueueClasses:        allocatorQueueClasses,
		AdmissionLatency:    req.AdmissionLatency,
		EnqueueLatency:      req.EnqueueLatency,
	})
	if allocatorDecision.MaxClaims <= 0 {
		return l.throttle("processing_capacity_full"), true, allocatorDecision
	}
	if allocatorDecision.MaxClaims > want {
		allocatorDecision.MaxClaims = want
	}
	if allocatorDecision.MaxClaims > available {
		allocatorDecision.MaxClaims = available
	}
	if allocatorDecision.MaxClaims <= 0 {
		return l.throttle("processing_capacity_full"), true, allocatorDecision
	}
	return l.reserveLocked(reservationKeys, ClaimLimiterDecision{Allow: true, MaxClaims: allocatorDecision.MaxClaims}), false, allocatorDecision
}

func (l *SQLClaimLimiter) staleRecoveryDueLocked(req ClaimLimiterRequest) bool {
	key := claimLimiterReservationKey(req)
	now := l.now()
	if next := l.staleRecovery[key]; !next.IsZero() && now.Before(next) {
		return false
	}
	l.staleRecovery[key] = now.Add(defaultStaleRecoveryInterval)
	return true
}

func (l *SQLClaimLimiter) reserveLocked(keys []string, decision ClaimLimiterDecision) ClaimLimiterDecision {
	if decision.MaxClaims <= 0 {
		return decision
	}
	l.nextReserveID++
	decision.ReservationID = l.nextReserveID
	l.reservations[decision.ReservationID] = claimReservation{
		keys: append([]string(nil), keys...),
		n:    decision.MaxClaims,
	}
	return decision
}

func (l *SQLClaimLimiter) limpFreshClaim(req ClaimLimiterRequest, reason string) (ClaimLimiterDecision, AdaptiveBudgetDecision) {
	allocatorQueueClasses := append([]string(nil), req.QueueClasses...)
	req = l.budgetScopeRequest(req)
	if l.maxProcessingRows <= 0 {
		return ClaimLimiterDecision{Allow: true, Reason: reason}, AdaptiveBudgetDecision{Reason: reason, QueueClasses: allocatorQueueClasses}
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	reservationKeys := claimLimiterReservationKeys(req)
	reserved := l.currentReservedForKeysLocked(reservationKeys)
	available := l.maxProcessingRows - reserved
	decision := AdaptiveBudgetDecision{
		Reason:              reason,
		Requested:           1,
		ConfiguredMaxRows:   l.maxProcessingRows,
		Reserved:            reserved,
		AvailableBeforeTrim: available,
		QueueClasses:        allocatorQueueClasses,
	}
	if available <= 0 {
		decision.MaxClaims = 0
		return l.throttle("processing_capacity_full"), decision
	}
	decision.MaxClaims = 1
	return l.reserveLocked(reservationKeys, ClaimLimiterDecision{Allow: true, MaxClaims: 1, Reason: reason}), decision
}

func (l *SQLClaimLimiter) dbDecision(stats sql.DBStats) (ClaimLimiterDecision, bool) {
	if stats.MaxOpenConnections <= 0 {
		return ClaimLimiterDecision{}, false
	}
	if stats.InUse >= stats.MaxOpenConnections {
		return l.dbThrottle("db_pool_exhausted"), true
	}
	ratio := float64(stats.InUse) / float64(stats.MaxOpenConnections)
	if ratio >= l.maxDBInUseRatio && stats.Idle == 0 {
		return l.dbThrottle("db_pool_saturated"), true
	}
	l.mu.Lock()
	waitDelta := stats.WaitCount - l.lastWaitCount
	l.lastWaitCount = stats.WaitCount
	l.mu.Unlock()
	if waitDelta > 0 && stats.Idle == 0 {
		return l.dbThrottle("db_pool_waiting"), true
	}
	return ClaimLimiterDecision{}, false
}

func (l *SQLClaimLimiter) throttle(reason string) ClaimLimiterDecision {
	return ClaimLimiterDecision{Allow: false, Delay: l.delay, Reason: reason}
}

// limpReclaim is the bounded soft-degrade for the two fail-closed branches that
// would otherwise freeze the whole shard fleet for defaultClaimLimiterDelay.
// Instead of Allow:false it grants exactly one stale-reclaim-only claim: the
// worker may re-lease one stranded stale-processing row but may not admit a fresh
// pending/failed row. Because reclaiming a stale row leaves ProcessingCount flat,
// this cannot over-claim past maxProcessingRows, so unlike the #3082 fail-open it
// needs no snapshot or reservation to stay under the ceiling. No Delay is set: on
// the Allow:true path the worker proceeds straight to the (single) claim, and an
// empty stale-reclaim backs off on the worker's own PollInterval.
func (l *SQLClaimLimiter) limpReclaim(reason string) ClaimLimiterDecision {
	return ClaimLimiterDecision{Allow: true, MaxClaims: 1, StaleReclaimOnly: true, Reason: reason}
}

// dbThrottle is the softer db-pool backpressure. It carries the same Allow=false
// signal as the fail-closed throttle, but with a short jittered delay instead of
// the 5s default so the shards sharing this limiter re-probe at staggered
// sub-second times rather than waking in fleet-wide lockstep.
func (l *SQLClaimLimiter) dbThrottle(reason string) ClaimLimiterDecision {
	return ClaimLimiterDecision{Allow: false, Delay: jitteredDBDelay(), Reason: reason}
}

// jitteredDBDelay returns dbPoolThrottleBaseDelay ± dbPoolThrottleJitter, clamped
// to [dbPoolThrottleMinDelay, dbPoolThrottleMaxDelay]. With the current constants
// the effective band is [200ms, 420ms); the floor and ceiling are guards that
// keep the delay sub-second if base/jitter are re-tuned. math/rand/v2 top-level
// funcs draw from a lock-free per-goroutine source, so this never contends l.mu
// on the hot path.
func jitteredDBDelay() time.Duration {
	delay := dbPoolThrottleBaseDelay - dbPoolThrottleJitter + rand.N(2*dbPoolThrottleJitter)
	if delay < dbPoolThrottleMinDelay {
		return dbPoolThrottleMinDelay
	}
	if delay > dbPoolThrottleMaxDelay {
		return dbPoolThrottleMaxDelay
	}
	return delay
}

func (l *SQLClaimLimiter) recordStatus(status ClaimLimiterStatus) {
	l.mu.Lock()
	status.Queue.QueueClasses = append([]string(nil), status.Queue.QueueClasses...)
	status.Allocator.QueueClasses = append([]string(nil), status.Allocator.QueueClasses...)
	status.Allocator.Signals = append([]string(nil), status.Allocator.Signals...)
	l.lastStatus = status
	shouldLog := l.shouldLogAllocatorDecisionLocked(status)
	l.mu.Unlock()
	if shouldLog {
		logAllocatorDecision(status)
	}
}

func (l *SQLClaimLimiter) shouldLogAllocatorDecisionLocked(status ClaimLimiterStatus) bool {
	decision := status.Allocator
	if decision.Reason == "" || decision.Reason == "configured_capacity" {
		l.lastLogKey = ""
		return false
	}
	key := strings.Join([]string{
		status.Queue.InstalledPackID,
		decision.Reason,
		strconv.Itoa(decision.MaxClaims),
		strconv.Itoa(decision.ConfiguredMaxRows),
		strconv.Itoa(decision.ActiveProcessing),
		strconv.Itoa(decision.StaleProcessing),
		strconv.Itoa(decision.Reserved),
		strconv.Itoa(status.DBPool.InUse),
		strconv.Itoa(status.DBPool.MaxOpenConnections),
		strings.Join(decision.Signals, "+"),
		strings.Join(decision.QueueClasses, "+"),
	}, "|")
	if key == l.lastLogKey {
		return false
	}
	l.lastLogKey = key
	return true
}

func logAllocatorDecision(status ClaimLimiterStatus) {
	decision := status.Allocator
	log.Printf("connector inbound claim allocator: pack=%s queue_classes=%s reason=%s signals=%s requested=%d max_claims=%d available=%d configured_max=%d processing=%d stale=%d reserved=%d db_in_use=%d db_max_open=%d",
		status.Queue.InstalledPackID,
		strings.Join(decision.QueueClasses, ","),
		decision.Reason,
		strings.Join(decision.Signals, ","),
		decision.Requested,
		decision.MaxClaims,
		decision.AvailableBeforeTrim,
		decision.ConfiguredMaxRows,
		decision.ActiveProcessing,
		decision.StaleProcessing,
		decision.Reserved,
		status.DBPool.InUse,
		status.DBPool.MaxOpenConnections,
	)
}

func (l *SQLClaimLimiter) logProfile(req ClaimLimiterRequest, status ClaimLimiterStatus, decision ClaimLimiterDecision) {
	if l == nil || !l.profile {
		return
	}
	allocator := status.Allocator
	reason := decision.Reason
	if reason == "" {
		reason = status.Reason
	}
	if reason == "" {
		reason = allocator.Reason
	}
	if reason == "" && decision.Allow {
		reason = "allow"
	}
	if reason == "" {
		reason = "unknown"
	}
	requested := allocator.Requested
	if requested <= 0 {
		requested = req.BatchSize
	}
	maxClaims := decision.MaxClaims
	if maxClaims <= 0 {
		maxClaims = allocator.MaxClaims
	}
	queueClasses := strings.Join(req.QueueClasses, ",")
	if queueClasses == "" {
		queueClasses = "all"
	}
	log.Printf("event=connector_inbound_claim_limiter_profile elapsed_ms=0 namespace=%s installed_pack_id=%s worker_id=%s queue_classes=%s allow=%t reason=%s delay_ms=%d requested=%d max_claims=%d reservation_id=%d stale_reclaim_only=%t configured_max=%d active_processing=%d stale_processing=%d reserved=%d available=%d claimable=%d pending=%d failed_due=%d db_in_use=%d db_idle=%d db_max_open=%d global_budget=%t",
		req.Namespace,
		req.PackInstallID,
		req.WorkerID,
		queueClasses,
		decision.Allow,
		reason,
		decision.Delay.Milliseconds(),
		requested,
		maxClaims,
		decision.ReservationID,
		decision.StaleReclaimOnly,
		allocator.ConfiguredMaxRows,
		status.Queue.ProcessingCount,
		status.Queue.StaleProcessingCount,
		status.Reserved,
		allocator.AvailableBeforeTrim,
		status.Queue.ClaimableCount,
		status.Queue.PendingCount,
		status.Queue.FailedDueCount,
		status.DBPool.InUse,
		status.DBPool.Idle,
		status.DBPool.MaxOpenConnections,
		l.globalProcessingBudget,
	)
}

func (l *SQLClaimLimiter) cachedClaimPressure(key string, now time.Time) (connector_events.ClaimPressureSnapshot, bool) {
	l.mu.Lock()
	defer l.mu.Unlock()
	entry, ok := l.pressureCache[key]
	if !ok || !now.Before(entry.expiresAt) {
		if ok {
			delete(l.pressureCache, key)
		}
		return connector_events.ClaimPressureSnapshot{}, false
	}
	entry.snapshot.QueueClasses = append([]string(nil), entry.snapshot.QueueClasses...)
	return entry.snapshot, true
}

func (l *SQLClaimLimiter) storeClaimPressure(key string, snapshot connector_events.ClaimPressureSnapshot, now time.Time) {
	l.mu.Lock()
	defer l.mu.Unlock()
	snapshot.QueueClasses = append([]string(nil), snapshot.QueueClasses...)
	l.pressureCache[key] = claimPressureCacheEntry{snapshot: snapshot, expiresAt: now.Add(l.pressureCacheTTL)}
}

func (l *SQLClaimLimiter) currentReserved(req ClaimLimiterRequest) int {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.currentReservedForKeysLocked(claimLimiterReservationKeys(l.budgetScopeRequest(req)))
}

func (l *SQLClaimLimiter) budgetScopeRequest(req ClaimLimiterRequest) ClaimLimiterRequest {
	if l == nil || !l.globalProcessingBudget {
		return req
	}
	req.QueueClasses = nil
	return req
}

func (l *SQLClaimLimiter) currentReservedForKeysLocked(keys []string) int {
	reserved := 0
	for _, reservation := range l.reservations {
		if reservationKeysOverlap(keys, reservation.keys) {
			reserved += reservation.n
		}
	}
	return reserved
}

func claimLimiterReservationKey(req ClaimLimiterRequest) string {
	return strings.Join(claimLimiterReservationKeys(req), ",")
}

func claimLimiterReservationKeys(req ClaimLimiterRequest) []string {
	packID := strings.TrimSpace(req.PackInstallID)
	if packID == "" {
		packID = "*"
	}
	if len(req.QueueClasses) == 0 {
		return []string{packID + "|*"}
	}
	keys := make([]string, 0, len(req.QueueClasses))
	seen := make(map[string]struct{}, len(req.QueueClasses))
	for _, rawClass := range req.QueueClasses {
		queueClass, err := connector_inbound_routing.NormalizeQueueClass(rawClass)
		if err != nil {
			return []string{packID + "|*"}
		}
		key := packID + "|" + queueClass
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		keys = append(keys, key)
	}
	if len(keys) == 0 {
		return []string{packID + "|*"}
	}
	sort.Strings(keys)
	return keys
}

func reservationKeysOverlap(a, b []string) bool {
	if len(a) == 0 || len(b) == 0 {
		return true
	}
	for _, key := range a {
		for _, other := range b {
			if reservationKeyOverlaps(key, other) {
				return true
			}
		}
	}
	return false
}

func reservationKeyOverlaps(a, b string) bool {
	aPack, aClass := splitReservationKey(a)
	bPack, bClass := splitReservationKey(b)
	if aPack == "*" || bPack == "*" {
		return true
	}
	if aPack != bPack {
		return false
	}
	return aClass == "*" || bClass == "*" || aClass == bClass
}

func splitReservationKey(key string) (string, string) {
	if strings.TrimSpace(key) == "" || key == "*" {
		return "*", "*"
	}
	packID, queueClass, ok := strings.Cut(key, "|")
	if !ok {
		return strings.TrimSpace(key), "*"
	}
	packID = strings.TrimSpace(packID)
	queueClass = strings.TrimSpace(queueClass)
	if packID == "" {
		packID = "*"
	}
	if queueClass == "" {
		queueClass = "*"
	}
	return packID, queueClass
}

func (w *Worker) claimLimiterRequest(batchSize int) ClaimLimiterRequest {
	if batchSize <= 0 {
		batchSize = w.cfg.BatchSize
	}
	return ClaimLimiterRequest{
		Namespace:     w.cfg.Namespace,
		PackInstallID: w.cfg.PackInstallID,
		WorkerID:      w.cfg.WorkerID,
		ClaimPath:     normalizeClaimPath(w.cfg.ClaimPath),
		QueueClasses:  append([]string(nil), w.cfg.QueueClasses...),
		BatchSize:     batchSize,
	}
}

func (w *Worker) waitForClaimPermit(ctx context.Context, drain <-chan struct{}, batchSize int) (ClaimLimiterDecision, bool, error) {
	if w.cfg.ClaimLimiter == nil {
		return ClaimLimiterDecision{Allow: true}, true, nil
	}
	decision, err := w.cfg.ClaimLimiter.BeforeClaim(ctx, w.claimLimiterRequest(batchSize))
	if err != nil {
		return decision, false, err
	}
	decision = w.capClaimLimiterDecision(decision)
	if decision.Allow {
		return decision, true, nil
	}
	delay := decision.Delay
	if delay <= 0 {
		delay = w.cfg.PollInterval
	}
	timer := time.NewTimer(delay)
	defer stopTimer(timer)
	select {
	case <-ctx.Done():
		return decision, false, ctx.Err()
	case <-drain:
		return decision, false, nil
	case <-timer.C:
		return decision, false, nil
	}
}

type claimPermitReleaser interface {
	ReleaseClaimPermit(ClaimLimiterDecision, int)
}

func (w *Worker) releaseClaimPermit(decision ClaimLimiterDecision, claimed int) {
	if w.cfg.ClaimLimiter == nil || decision.ReservationID == 0 {
		return
	}
	if releaser, ok := w.cfg.ClaimLimiter.(claimPermitReleaser); ok {
		releaser.ReleaseClaimPermit(decision, claimed)
	}
}