package connector_inbound

import (
	"context"
	"database/sql"
	"errors"
	"testing"
	"time"

	connector_events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
)

type fakeDBStatsProvider struct {
	stats sql.DBStats
}

func (f fakeDBStatsProvider) Stats() sql.DBStats {
	return f.stats
}

type fakePressureProvider struct {
	snapshot     connector_events.ClaimPressureSnapshot
	err          error
	calls        int
	queueClasses [][]string
}

func (f *fakePressureProvider) ProcessingPressure(_ context.Context, _ string, queueClasses []string) (connector_events.ClaimPressureSnapshot, error) {
	f.calls++
	f.queueClasses = append(f.queueClasses, append([]string(nil), queueClasses...))
	if f.err != nil {
		return connector_events.ClaimPressureSnapshot{}, f.err
	}
	return f.snapshot, nil
}

type pressureProviderWithExpensiveClaimPressure struct {
	fakePressureProvider
	claimPressureCalls int
}

func (f *pressureProviderWithExpensiveClaimPressure) ClaimPressure(context.Context, string, []string) (connector_events.ClaimPressureSnapshot, error) {
	f.claimPressureCalls++
	return connector_events.ClaimPressureSnapshot{}, errors.New("expensive ClaimPressure must not be called from the limiter hot path")
}

type fixedAdaptiveBudgetAllocator struct {
	decision AdaptiveBudgetDecision
	inputs   []AdaptiveBudgetInput
}

func (f *fixedAdaptiveBudgetAllocator) Allocate(input AdaptiveBudgetInput) AdaptiveBudgetDecision {
	f.inputs = append(f.inputs, input)
	return f.decision
}

func containsString(values []string, want string) bool {
	for _, value := range values {
		if value == want {
			return true
		}
	}
	return false
}

func TestSQLClaimLimiterLimpsOnProbeError(t *testing.T) {
	// A probe error (timeout / transient DB failure) must not admit fresh rows:
	// without a live ProcessingCount snapshot, a per-process reservation cannot
	// enforce maxProcessingRows across pods. The fallback may only reclaim stale
	// processing rows, which keeps ProcessingCount flat.
	pressure := &fakePressureProvider{err: errors.New("context deadline exceeded")}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		DBStats:           fakeDBStatsProvider{stats: sql.DBStats{MaxOpenConnections: 100, InUse: 2, Idle: 98}},
		Pressure:          pressure,
		MaxProcessingRows: 1,
	})
	decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: 10})
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if !decision.Allow {
		t.Fatalf("decision.Allow = false reason=%q, want stale reclaim to remain available", decision.Reason)
	}
	if decision.Reason != "queue_pressure_unavailable" {
		t.Fatalf("decision.Reason = %q, want queue_pressure_unavailable", decision.Reason)
	}
	if !decision.StaleReclaimOnly {
		t.Fatal("decision.StaleReclaimOnly = false, want probe failures to reject fresh claims")
	}
	if decision.MaxClaims != 1 {
		t.Fatalf("decision.MaxClaims = %d, want 1 stale reclaim", decision.MaxClaims)
	}
	if decision.ReservationID != 0 {
		t.Fatalf("decision.ReservationID = %d, want stale reclaim fallback to avoid fresh-cap reservation", decision.ReservationID)
	}
}

func TestSQLClaimLimiterWorkItemProbeErrorAllowsReservedMinProgress(t *testing.T) {
	pressure := &fakePressureProvider{err: errors.New("context deadline exceeded")}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		DBStats:           fakeDBStatsProvider{stats: sql.DBStats{MaxOpenConnections: 100, InUse: 2, Idle: 98}},
		Pressure:          pressure,
		MaxProcessingRows: 1,
	})
	decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{
		PackInstallID: "pack-1",
		ClaimPath:     ClaimPathWorkItem,
		BatchSize:     1,
	})
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if !decision.Allow || decision.Reason != "queue_pressure_unavailable" {
		t.Fatalf("decision = %+v, want queue_pressure_unavailable min-progress allow", decision)
	}
	if decision.StaleReclaimOnly {
		t.Fatalf("decision.StaleReclaimOnly = true, want fresh WorkItem min-progress")
	}
	if decision.MaxClaims != 1 {
		t.Fatalf("decision.MaxClaims = %d, want 1", decision.MaxClaims)
	}
	if decision.ReservationID == 0 {
		t.Fatalf("decision.ReservationID = 0, want reserved fresh min-progress permit")
	}
	limiter.ReleaseClaimPermit(decision, 1)
}

func TestSQLClaimLimiterAdaptiveDownshiftsUnderProcessingAndDBPressure(t *testing.T) {
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID: "pack-1",
			ProcessingCount: 24,
			QueueClasses:    []string{connector_inbound_routing.QueueClassInteractive, connector_inbound_routing.QueueClassDefault},
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		DBStats:           fakeDBStatsProvider{stats: sql.DBStats{MaxOpenConnections: 10, InUse: 8, Idle: 0}},
		Pressure:          pressure,
		MaxProcessingRows: 32,
		MaxDBInUseRatio:   0.90,
	})

	decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{
		PackInstallID: "pack-1",
		QueueClasses:  []string{connector_inbound_routing.QueueClassInteractive, connector_inbound_routing.QueueClassDefault},
		BatchSize:     16,
	})
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if !decision.Allow {
		t.Fatalf("decision.Allow = false reason=%q, want adaptive downshift, not throttle", decision.Reason)
	}
	if decision.MaxClaims != 2 {
		t.Fatalf("decision.MaxClaims = %d, want 2 after processing+DB adaptive downshift", decision.MaxClaims)
	}
	status := limiter.Status()
	if status.Allocator.Reason != "adaptive_downshift" {
		t.Fatalf("allocator reason = %q, want adaptive_downshift", status.Allocator.Reason)
	}
	if !containsString(status.Allocator.Signals, "processing_high") || !containsString(status.Allocator.Signals, "db_pool_high") {
		t.Fatalf("allocator signals = %v, want processing_high and db_pool_high", status.Allocator.Signals)
	}
	if status.Allocator.AvailableBeforeTrim != 8 {
		t.Fatalf("allocator available = %d, want 8", status.Allocator.AvailableBeforeTrim)
	}
}

func TestSQLClaimLimiterAdaptivePreservesBulkQueueClassWithGlobalBudget(t *testing.T) {
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID: "pack-1",
			ProcessingCount: 24,
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		DBStats:                fakeDBStatsProvider{stats: sql.DBStats{MaxOpenConnections: 10, InUse: 8, Idle: 0}},
		Pressure:               pressure,
		MaxProcessingRows:      32,
		MaxDBInUseRatio:        0.90,
		GlobalProcessingBudget: true,
	})

	decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{
		PackInstallID: "pack-1",
		QueueClasses:  []string{connector_inbound_routing.QueueClassBulk},
		BatchSize:     100,
	})
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if !decision.Allow {
		t.Fatalf("decision.Allow = false reason=%q, want adaptive downshift", decision.Reason)
	}
	if decision.MaxClaims != 1 {
		t.Fatalf("decision.MaxClaims = %d, want 1 after global-budget bulk downshift", decision.MaxClaims)
	}
	status := limiter.Status()
	if !containsString(status.Allocator.Signals, "bulk_queue_pressure") {
		t.Fatalf("allocator signals = %v, want bulk_queue_pressure", status.Allocator.Signals)
	}
	if len(status.Allocator.QueueClasses) != 1 || status.Allocator.QueueClasses[0] != connector_inbound_routing.QueueClassBulk {
		t.Fatalf("allocator queue classes = %v, want original bulk queue class", status.Allocator.QueueClasses)
	}
	if len(pressure.queueClasses) != 1 || len(pressure.queueClasses[0]) != 0 {
		t.Fatalf("pressure queue classes = %v, want pack-global pressure probe", pressure.queueClasses)
	}
}

func TestSQLClaimLimiterClampsCustomAllocatorOutput(t *testing.T) {
	tests := []struct {
		name       string
		maxRows    int
		processing int
		batchSize  int
		want       int
	}{
		{name: "available capacity", maxRows: 10, processing: 8, batchSize: 100, want: 2},
		{name: "requested batch", maxRows: 32, processing: 0, batchSize: 5, want: 5},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			allocator := &fixedAdaptiveBudgetAllocator{
				decision: AdaptiveBudgetDecision{Reason: "custom", MaxClaims: 50},
			}
			limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
				DBStats: fakeDBStatsProvider{stats: sql.DBStats{MaxOpenConnections: 100, InUse: 2, Idle: 98}},
				Pressure: &fakePressureProvider{snapshot: connector_events.ClaimPressureSnapshot{
					InstalledPackID: "pack-1",
					ProcessingCount: tt.processing,
				}},
				MaxProcessingRows: tt.maxRows,
				BudgetAllocator:   allocator,
			})

			decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: tt.batchSize})
			if err != nil {
				t.Fatalf("BeforeClaim: %v", err)
			}
			if !decision.Allow {
				t.Fatalf("decision.Allow = false reason=%q, want clamped allow", decision.Reason)
			}
			if decision.MaxClaims != tt.want {
				t.Fatalf("decision.MaxClaims = %d, want %d", decision.MaxClaims, tt.want)
			}
			if decision.ReservationID == 0 {
				t.Fatalf("decision.ReservationID = 0, want clamped reservation")
			}
			status := limiter.Status()
			if status.Allocator.MaxClaims != tt.want {
				t.Fatalf("status.Allocator.MaxClaims = %d, want %d", status.Allocator.MaxClaims, tt.want)
			}
			if len(allocator.inputs) != 1 {
				t.Fatalf("allocator inputs = %d, want 1", len(allocator.inputs))
			}
		})
	}
}

func TestSQLClaimLimiterAdaptiveDoesNotStarveStaleReclaimAtCapacity(t *testing.T) {
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID:      "pack-1",
			ProcessingCount:      32,
			StaleProcessingCount: 3,
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		Pressure:          pressure,
		MaxProcessingRows: 32,
	})

	decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: 10})
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if !decision.Allow || !decision.StaleReclaimOnly || decision.MaxClaims != 1 {
		t.Fatalf("decision = %+v, want one stale-reclaim-only permit at full capacity", decision)
	}
	status := limiter.Status()
	if status.Allocator.Reason != "stale_reclaim_limp" {
		t.Fatalf("allocator reason = %q, want stale_reclaim_limp", status.Allocator.Reason)
	}
}

func TestSQLClaimLimiterAdaptiveHotPathDoesNotCallClaimPressure(t *testing.T) {
	pressure := &pressureProviderWithExpensiveClaimPressure{
		fakePressureProvider: fakePressureProvider{
			snapshot: connector_events.ClaimPressureSnapshot{
				InstalledPackID: "pack-1",
				ProcessingCount: 4,
			},
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		Pressure:          pressure,
		MaxProcessingRows: 32,
	})

	decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: 10})
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if !decision.Allow || decision.MaxClaims != 10 {
		t.Fatalf("decision = %+v, want configured capacity from ProcessingPressure", decision)
	}
	if pressure.fakePressureProvider.calls != 1 {
		t.Fatalf("ProcessingPressure calls = %d, want 1", pressure.fakePressureProvider.calls)
	}
	if pressure.claimPressureCalls != 0 {
		t.Fatalf("ClaimPressure calls = %d, want 0 on limiter hot path", pressure.claimPressureCalls)
	}
}

func TestMaxProcessingRowsForWorkers(t *testing.T) {
	// The budget scales with shard count in both directions. The default
	// 8-worker case stays at the legacy 32-row ceiling, but lowering worker count
	// must lower DB-facing processing pressure too.
	cases := []struct {
		workerCount int
		want        int
	}{
		{workerCount: 0, want: 4},
		{workerCount: 1, want: 4},
		{workerCount: 2, want: 8},
		{workerCount: 3, want: 12},
		{workerCount: 7, want: 28},
		{workerCount: 8, want: 32},
		{workerCount: 16, want: 64},
		{workerCount: 32, want: 128},
	}
	for _, tc := range cases {
		if got := MaxProcessingRowsForWorkers(tc.workerCount); got != tc.want {
			t.Fatalf("MaxProcessingRowsForWorkers(%d) = %d, want %d", tc.workerCount, got, tc.want)
		}
	}
	// A limiter built with the helper output exposes that budget unchanged.
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		MaxProcessingRows: MaxProcessingRowsForWorkers(16),
	})
	if got, want := limiter.maxProcessingRows, 64; got != want {
		t.Fatalf("maxProcessingRows = %d, want %d", got, want)
	}
	// Behavior preservation: default 8-worker case still equals the legacy default.
	if got := MaxProcessingRowsForWorkers(8); got != defaultClaimLimiterMaxProcessingRows {
		t.Fatalf("MaxProcessingRowsForWorkers(8) = %d, want %d (default behavior must be unchanged)", got, defaultClaimLimiterMaxProcessingRows)
	}
}

func TestSQLClaimLimiterWorkItemExpiredLeasesDoNotConsumeFreshCapacity(t *testing.T) {
	// A crashed/restarted pack can leave maxProcessingRows rows stuck in
	// 'processing'. On the WorkItem path those expired leases are recovery backlog,
	// not active capacity. Lane-head ordering prevents same-lane bypass, so unrelated
	// fresh lane heads must remain admissible.
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID:      "pack-1",
			ProcessingCount:      32,
			StaleProcessingCount: 32,
		},
	}
	now := time.Date(2026, 7, 11, 14, 0, 0, 0, time.UTC)
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		DBStats:           fakeDBStatsProvider{stats: sql.DBStats{MaxOpenConnections: 100, InUse: 2, Idle: 98}},
		Pressure:          pressure,
		MaxProcessingRows: 32,
		Now:               func() time.Time { return now },
	})
	req := ClaimLimiterRequest{
		PackInstallID: "pack-1",
		ClaimPath:     ClaimPathWorkItem,
		BatchSize:     10,
	}
	recovery, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if !recovery.Allow || !recovery.StaleReclaimOnly || recovery.MaxClaims != 1 || recovery.Reason != "stale_recovery_due" {
		t.Fatalf("recovery decision = %+v, want one bounded stale recovery claim", recovery)
	}
	decision, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim fresh: %v", err)
	}
	if !decision.Allow {
		t.Fatalf("decision.Allow = false reason=%q, want fresh capacity", decision.Reason)
	}
	if decision.StaleReclaimOnly || decision.MaxClaims != 10 {
		t.Fatalf("decision = %+v, want ten fresh claims", decision)
	}
	if decision.ReservationID == 0 {
		t.Fatal("decision.ReservationID = 0, want fresh-capacity reservation")
	}
	status := limiter.Status()
	if status.Allocator.ActiveProcessing != 0 || status.Allocator.StaleProcessing != 32 {
		t.Fatalf("allocator = %+v, want active=0 stale=32", status.Allocator)
	}
	now = now.Add(defaultStaleRecoveryInterval)
	recovery, err = limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim recovery interval: %v", err)
	}
	if !recovery.StaleReclaimOnly || recovery.Reason != "stale_recovery_due" {
		t.Fatalf("recovery after interval = %+v, want stale recovery due", recovery)
	}
}

func TestSQLClaimLimiterWorkItemClampsStalePressureBeforeSubtracting(t *testing.T) {
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID:      "pack-1",
			ProcessingCount:      3,
			StaleProcessingCount: 8,
		},
	}
	now := time.Date(2026, 7, 11, 14, 0, 0, 0, time.UTC)
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		Pressure:          pressure,
		MaxProcessingRows: 4,
		Now:               func() time.Time { return now },
	})
	req := ClaimLimiterRequest{
		PackInstallID: "pack-1",
		ClaimPath:     ClaimPathWorkItem,
		BatchSize:     4,
	}
	if recovery, err := limiter.BeforeClaim(context.Background(), req); err != nil || !recovery.StaleReclaimOnly {
		t.Fatalf("initial recovery decision = %+v err=%v, want stale recovery", recovery, err)
	}
	decision, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if !decision.Allow || decision.MaxClaims != 4 || decision.StaleReclaimOnly {
		t.Fatalf("decision = %+v, want four fresh claims after clamped subtraction", decision)
	}
}

func TestSQLClaimLimiterThrottlesAtProcessingCapacity(t *testing.T) {
	// Genuinely active (non-stale) processing at the cap must throttle.
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID: "pack-1",
			ProcessingCount: 32, // all active, none stale
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		DBStats:           fakeDBStatsProvider{stats: sql.DBStats{MaxOpenConnections: 100, InUse: 2, Idle: 98}},
		Pressure:          pressure,
		MaxProcessingRows: 32,
	})
	decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: 10})
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if decision.Allow {
		t.Fatalf("decision.Allow = true reason=%q, want processing_capacity_full throttle", decision.Reason)
	}
	if decision.Reason != "processing_capacity_full" {
		t.Fatalf("decision.Reason = %q, want processing_capacity_full", decision.Reason)
	}
}

func TestSQLClaimLimiterLimpsAtCapacityWithReclaimableStale(t *testing.T) {
	// Capacity is full, and stale rows are present. Because stale pressure is only
	// a hint, the limiter must attempt exactly one stale-only reclaim instead of
	// discounting those stale rows from fresh capacity.
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID:      "pack-1",
			ProcessingCount:      40,
			StaleProcessingCount: 8,
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		DBStats:           fakeDBStatsProvider{stats: sql.DBStats{MaxOpenConnections: 100, InUse: 2, Idle: 98}},
		Pressure:          pressure,
		MaxProcessingRows: 32,
	})
	decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: 10})
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if !decision.Allow {
		t.Fatalf("decision.Allow = false reason=%q, want limp so stale rows are reclaimed at capacity", decision.Reason)
	}
	if decision.Reason != "processing_capacity_limp" {
		t.Fatalf("decision.Reason = %q, want processing_capacity_limp", decision.Reason)
	}
	if !decision.StaleReclaimOnly || decision.MaxClaims != 1 {
		t.Fatalf("decision = %+v, want StaleReclaimOnly with MaxClaims 1", decision)
	}
	// Limp at capacity must not reserve ceiling budget either.
	if decision.ReservationID != 0 {
		t.Fatalf("decision.ReservationID = %d, want 0 (stale reclaim consumes no ceiling budget)", decision.ReservationID)
	}
}

func TestSQLClaimLimiterThrottlesWhenDBPoolExhausted(t *testing.T) {
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		DBStats: fakeDBStatsProvider{stats: sql.DBStats{
			MaxOpenConnections: 10,
			InUse:              10,
			Idle:               0,
		}},
		Delay: time.Second,
	})

	decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{
		PackInstallID: "pack-1",
		BatchSize:     10,
	})
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if decision.Allow {
		t.Fatal("decision.Allow = true, want throttled")
	}
	if decision.Reason != "db_pool_exhausted" {
		t.Fatalf("decision.Reason = %q, want db_pool_exhausted", decision.Reason)
	}
}

func TestSQLClaimLimiterDBPoolThrottleUsesShortJitteredDelay(t *testing.T) {
	// One SQLClaimLimiter is shared across all 8 shards of a pack. A db-pool
	// throttle that returned the full 5s default froze new-claim AND stale-reclaim
	// for the whole fleet in 5s lockstep. The db-pool branches must instead return
	// a short sub-second delay so shards re-probe quickly once pool pressure eases.
	cases := []struct {
		name   string
		stats  sql.DBStats
		reason string
	}{
		{
			name:   "exhausted",
			stats:  sql.DBStats{MaxOpenConnections: 10, InUse: 10, Idle: 0},
			reason: "db_pool_exhausted",
		},
		{
			name:   "saturated",
			stats:  sql.DBStats{MaxOpenConnections: 10, InUse: 9, Idle: 0},
			reason: "db_pool_saturated",
		},
		{
			name:   "waiting",
			stats:  sql.DBStats{MaxOpenConnections: 10, InUse: 5, Idle: 0, WaitCount: 1},
			reason: "db_pool_waiting",
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
				DBStats: fakeDBStatsProvider{stats: tc.stats},
			})
			decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: 10})
			if err != nil {
				t.Fatalf("BeforeClaim: %v", err)
			}
			if decision.Allow {
				t.Fatalf("decision.Allow = true reason=%q, want db-pool throttle", decision.Reason)
			}
			if decision.Reason != tc.reason {
				t.Fatalf("decision.Reason = %q, want %q", decision.Reason, tc.reason)
			}
			// Short jittered delay, NOT the 5s lockstep stall. Floor 200ms, and well
			// under the old 5s default (a generous 500ms ceiling proves the change).
			if decision.Delay < 200*time.Millisecond || decision.Delay > 500*time.Millisecond {
				t.Fatalf("decision.Delay = %v, want short jittered delay in [200ms, 500ms] (NOT the 5s lockstep stall)", decision.Delay)
			}
		})
	}
}

func TestSQLClaimLimiterDBPoolThrottleDelaysDesync(t *testing.T) {
	// The point of the jitter: shards sharing one limiter must not wake in
	// lockstep. Many calls with identical saturated stats must return a spread of
	// delays, each within the clamped [200ms, ~420ms] band.
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		DBStats: fakeDBStatsProvider{stats: sql.DBStats{MaxOpenConnections: 10, InUse: 10, Idle: 0}},
	})
	seen := make(map[time.Duration]struct{})
	const samples = 64
	for i := 0; i < samples; i++ {
		decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: 10})
		if err != nil {
			t.Fatalf("BeforeClaim: %v", err)
		}
		if decision.Reason != "db_pool_exhausted" {
			t.Fatalf("decision.Reason = %q, want db_pool_exhausted", decision.Reason)
		}
		if decision.Delay < 200*time.Millisecond || decision.Delay > 420*time.Millisecond {
			t.Fatalf("decision.Delay = %v, want within clamped band [200ms, 420ms]", decision.Delay)
		}
		seen[decision.Delay] = struct{}{}
	}
	if len(seen) < 2 {
		t.Fatalf("got %d distinct delay(s) across %d samples, want jittered (non-lockstep) delays", len(seen), samples)
	}
}

func TestSQLClaimLimiterCapsClaimsByProcessingPressure(t *testing.T) {
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID:      "pack-1",
			ProcessingCount:      8,
			StaleProcessingCount: 2,
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		Pressure:          pressure,
		MaxProcessingRows: 10,
	})

	req := ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: 10}
	decision, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if !decision.Allow {
		t.Fatalf("decision.Allow = false reason=%q", decision.Reason)
	}
	if decision.MaxClaims != 2 {
		t.Fatalf("decision.MaxClaims = %d, want 2", decision.MaxClaims)
	}
	if decision.ReservationID == 0 {
		t.Fatal("decision.ReservationID = 0, want reservation")
	}

	second, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim second: %v", err)
	}
	if second.Allow {
		t.Fatalf("second.Allow = true, want capacity throttle")
	}
	if second.Reason != "processing_capacity_full" {
		t.Fatalf("second.Reason = %q, want processing_capacity_full", second.Reason)
	}
	if second.Delay < 200*time.Millisecond || second.Delay > 500*time.Millisecond {
		t.Fatalf("second.Delay = %v, want short reservation-only retry delay in [200ms, 500ms]", second.Delay)
	}

	limiter.ReleaseClaimPermit(decision, 1)
	pressure.snapshot = connector_events.ClaimPressureSnapshot{
		InstalledPackID:      "pack-1",
		ProcessingCount:      9,
		StaleProcessingCount: 2,
	}
	third, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim third: %v", err)
	}
	if !third.Allow || third.MaxClaims != 2 {
		t.Fatalf("third decision = %+v, want allow with completed permits released", third)
	}
	if pressure.calls != 1 {
		t.Fatalf("pressure calls = %d, want cached pressure reused after successful claim", pressure.calls)
	}
}

func TestSQLClaimLimiterProcessingCapacityFullUsesDefaultDelayForActiveProcessing(t *testing.T) {
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID: "pack-1",
			ProcessingCount: 10,
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		Pressure:          pressure,
		MaxProcessingRows: 10,
	})

	decision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: 1})
	if err != nil {
		t.Fatalf("BeforeClaim: %v", err)
	}
	if decision.Allow || decision.Reason != "processing_capacity_full" {
		t.Fatalf("decision = %+v, want processing_capacity_full throttle", decision)
	}
	if decision.Delay != defaultClaimLimiterDelay {
		t.Fatalf("decision.Delay = %v, want default active-processing delay %v", decision.Delay, defaultClaimLimiterDelay)
	}
}

func TestSQLClaimLimiterKeepsReservationsUntilExplicitRelease(t *testing.T) {
	now := time.Date(2026, 7, 7, 10, 0, 0, 0, time.UTC)
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID: "pack-1",
			ProcessingCount: 0,
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		Now:                    func() time.Time { return now },
		Pressure:               pressure,
		PressureCacheTTL:       time.Hour,
		MaxProcessingRows:      1,
		GlobalProcessingBudget: true,
	})
	req := ClaimLimiterRequest{
		PackInstallID: "pack-1",
		QueueClasses:  []string{connector_inbound_routing.QueueClassBulk},
		BatchSize:     1,
	}

	first, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim first: %v", err)
	}
	if !first.Allow || first.ReservationID == 0 {
		t.Fatalf("first decision = %+v, want one reserved claim", first)
	}

	second, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim second: %v", err)
	}
	if second.Allow || second.Reason != "processing_capacity_full" {
		t.Fatalf("second decision = %+v, want reservation to consume the only slot", second)
	}
	status := limiter.Status()
	if status.Allocator.Reserved != 1 {
		t.Fatalf("reserved = %d, want 1", status.Allocator.Reserved)
	}

	now = now.Add(10 * time.Minute)
	third, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim third: %v", err)
	}
	if third.Allow || third.Reason != "processing_capacity_full" {
		t.Fatalf("third decision = %+v, want reservation to remain held after time advances", third)
	}
	status = limiter.Status()
	if status.Allocator.Reserved != 1 {
		t.Fatalf("allocator reserved after time advance = %d, want 1", status.Allocator.Reserved)
	}
	limiter.ReleaseClaimPermit(first, 0)
	fourth, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim fourth: %v", err)
	}
	if !fourth.Allow || fourth.ReservationID == 0 || fourth.ReservationID == first.ReservationID {
		t.Fatalf("fourth decision = %+v, want refill after explicit reservation release", fourth)
	}
	if got := limiter.currentReserved(req); got != 1 {
		t.Fatalf("current reserved after fourth = %d, want only the new reservation", got)
	}
	if pressure.calls != 1 {
		t.Fatalf("pressure calls = %d, want cached pressure reused while reservation accounting stays local", pressure.calls)
	}
}

func TestSQLClaimLimiterScopesBudgetByQueueClass(t *testing.T) {
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID: "pack-1",
			ProcessingCount: 0,
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		Pressure:          pressure,
		PressureCacheTTL:  time.Minute,
		MaxProcessingRows: 10,
	})

	bulkReq := ClaimLimiterRequest{
		PackInstallID: "pack-1",
		QueueClasses:  []string{connector_inbound_routing.QueueClassBulk},
		BatchSize:     10,
	}
	bulkDecision, err := limiter.BeforeClaim(context.Background(), bulkReq)
	if err != nil {
		t.Fatalf("BeforeClaim bulk: %v", err)
	}
	if !bulkDecision.Allow || bulkDecision.MaxClaims != 10 {
		t.Fatalf("bulk decision = %+v, want full reservation", bulkDecision)
	}

	latencyReq := ClaimLimiterRequest{
		PackInstallID: "pack-1",
		QueueClasses:  []string{connector_inbound_routing.QueueClassInteractive, connector_inbound_routing.QueueClassDefault},
		BatchSize:     5,
	}
	latencyDecision, err := limiter.BeforeClaim(context.Background(), latencyReq)
	if err != nil {
		t.Fatalf("BeforeClaim latency: %v", err)
	}
	if !latencyDecision.Allow || latencyDecision.MaxClaims != 5 {
		t.Fatalf("latency decision = %+v, want independent queue-class budget", latencyDecision)
	}
	sharedDecision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{
		PackInstallID: "pack-1",
		QueueClasses: []string{
			connector_inbound_routing.QueueClassInteractive,
			connector_inbound_routing.QueueClassDefault,
			connector_inbound_routing.QueueClassBulk,
		},
		BatchSize: 10,
	})
	if err != nil {
		t.Fatalf("BeforeClaim shared: %v", err)
	}
	if sharedDecision.Allow {
		t.Fatalf("shared decision = %+v, want capacity throttle because it overlaps bulk and latency reservations", sharedDecision)
	}
	if sharedDecision.Reason != "processing_capacity_full" {
		t.Fatalf("shared reason = %q, want processing_capacity_full", sharedDecision.Reason)
	}
	if pressure.calls != 3 {
		t.Fatalf("pressure calls = %d, want one pressure probe per queue-class budget", pressure.calls)
	}
	if got := pressure.queueClasses[0]; len(got) != 1 || got[0] != connector_inbound_routing.QueueClassBulk {
		t.Fatalf("bulk pressure queue classes = %v", got)
	}
	if got := pressure.queueClasses[1]; len(got) != 2 || got[0] != connector_inbound_routing.QueueClassInteractive || got[1] != connector_inbound_routing.QueueClassDefault {
		t.Fatalf("latency pressure queue classes = %v", got)
	}
}

func TestSQLClaimLimiterGlobalProcessingBudgetSharesQueueClassCapacity(t *testing.T) {
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID: "pack-1",
			ProcessingCount: 0,
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		Pressure:               pressure,
		PressureCacheTTL:       time.Minute,
		MaxProcessingRows:      10,
		GlobalProcessingBudget: true,
	})

	bulkDecision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{
		PackInstallID: "pack-1",
		QueueClasses:  []string{connector_inbound_routing.QueueClassBulk},
		BatchSize:     10,
	})
	if err != nil {
		t.Fatalf("BeforeClaim bulk: %v", err)
	}
	if !bulkDecision.Allow || bulkDecision.MaxClaims != 10 {
		t.Fatalf("bulk decision = %+v, want full global reservation", bulkDecision)
	}

	latencyDecision, err := limiter.BeforeClaim(context.Background(), ClaimLimiterRequest{
		PackInstallID: "pack-1",
		QueueClasses:  []string{connector_inbound_routing.QueueClassInteractive, connector_inbound_routing.QueueClassDefault},
		BatchSize:     5,
	})
	if err != nil {
		t.Fatalf("BeforeClaim latency: %v", err)
	}
	if latencyDecision.Allow {
		t.Fatalf("latency decision = %+v, want global capacity throttle", latencyDecision)
	}
	if latencyDecision.Reason != "processing_capacity_full" {
		t.Fatalf("latency reason = %q, want processing_capacity_full", latencyDecision.Reason)
	}
	if pressure.calls != 1 {
		t.Fatalf("pressure calls = %d, want one global pressure probe", pressure.calls)
	}
	if got := pressure.queueClasses[0]; len(got) != 0 {
		t.Fatalf("global pressure queue classes = %v, want pack-global probe", got)
	}
}

func TestSQLClaimLimiterCachesRefreshAfterCompletedClaim(t *testing.T) {
	now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
	pressure := &fakePressureProvider{
		snapshot: connector_events.ClaimPressureSnapshot{
			InstalledPackID: "pack-1",
			ProcessingCount: 8,
		},
	}
	limiter := NewSQLClaimLimiter(SQLClaimLimiterConfig{
		Pressure:          pressure,
		Now:               func() time.Time { return now },
		PressureCacheTTL:  time.Second,
		MaxProcessingRows: 10,
	})
	req := ClaimLimiterRequest{PackInstallID: "pack-1", BatchSize: 10}

	first, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim first: %v", err)
	}
	limiter.ReleaseClaimPermit(first, 1)

	now = now.Add(2 * time.Second)
	pressure.snapshot = connector_events.ClaimPressureSnapshot{
		InstalledPackID: "pack-1",
		ProcessingCount: 4,
	}
	second, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim second: %v", err)
	}
	if !second.Allow || second.MaxClaims != 6 {
		t.Fatalf("second decision = %+v, want allow with refreshed pressure", second)
	}
	if pressure.calls != 2 {
		t.Fatalf("pressure calls after refresh = %d, want 2", pressure.calls)
	}
	limiter.ReleaseClaimPermit(second, 0)

	third, err := limiter.BeforeClaim(context.Background(), req)
	if err != nil {
		t.Fatalf("BeforeClaim third: %v", err)
	}
	if !third.Allow || third.MaxClaims != 6 {
		t.Fatalf("third decision = %+v, want cached refreshed pressure", third)
	}
	if pressure.calls != 2 {
		t.Fatalf("pressure calls after cached refresh = %d, want 2", pressure.calls)
	}
}

func TestWorkerClaimLimiterRequestUsesActualClaimUnit(t *testing.T) {
	worker := &Worker{cfg: WorkerConfig{
		Namespace:     "ns",
		PackInstallID: "pack-1",
		WorkerID:      "worker-1",
		BatchSize:     100,
	}}

	batchReq := worker.claimLimiterRequest(worker.cfg.BatchSize)
	if batchReq.BatchSize != 100 {
		t.Fatalf("batch request BatchSize = %d, want 100", batchReq.BatchSize)
	}

	claimFirstReq := worker.claimLimiterRequest(1)
	if claimFirstReq.BatchSize != 1 {
		t.Fatalf("claim-first request BatchSize = %d, want 1", claimFirstReq.BatchSize)
	}
}