package connector_throttle

import (
	"context"
	"errors"
	"testing"
	"time"

	"golang.org/x/time/rate"
)

// --- test doubles ---

// stubCredentialResolver returns a fixed (canonicalID, namespace, active, err).
type stubCredentialResolver struct {
	canonicalID string
	namespace   string
	active      bool
	err         error
}

func (s *stubCredentialResolver) ResolveCredentialPack(_ context.Context, _ string) (string, string, bool, error) {
	return s.canonicalID, s.namespace, s.active, s.err
}

// canonicalMappingResolver maps credentialRef → canonicalID; all return the
// same namespace/active. Used for the case-variant bucket-sharing test.
type canonicalMappingResolver struct {
	mapping   map[string]string // credentialRef → canonicalID
	namespace string
	active    bool
}

func (r *canonicalMappingResolver) ResolveCredentialPack(_ context.Context, ref string) (string, string, bool, error) {
	id, ok := r.mapping[ref]
	if !ok {
		return "", "", false, nil
	}
	return id, r.namespace, r.active, nil
}

type stubLimitProvider struct {
	rps   float64
	burst int
	ok    bool
	err   error
}

func (s *stubLimitProvider) LimitFor(_ context.Context, _, _ string) (float64, int, bool, error) {
	return s.rps, s.burst, s.ok, s.err
}

// --- helpers ---

// frozenClock returns a clock that always returns t. Used to control
// time in tests so bucket refill is deterministic.
func frozenClock(t time.Time) func() time.Time {
	return func() time.Time { return t }
}

func activeResolver(canonicalID, namespace string) *stubCredentialResolver {
	return &stubCredentialResolver{canonicalID: canonicalID, namespace: namespace, active: true}
}

func inactiveResolver() *stubCredentialResolver {
	return &stubCredentialResolver{canonicalID: "some-id", namespace: "test-pack", active: false}
}

// autoMultiCredResolver returns a resolver whose ResolveCredentialPack derives
// canonicalID as ref+"-id" for any ref. Used for GC eviction tests where an
// arbitrary number of distinct credentials must each get their own bucket.
type autoMultiCredResolver struct{ namespace string }

func (r *autoMultiCredResolver) ResolveCredentialPack(_ context.Context, ref string) (string, string, bool, error) {
	return ref + "-id", r.namespace, true, nil
}

func staticLimits(rps float64, burst int) *stubLimitProvider {
	return &stubLimitProvider{rps: rps, burst: burst, ok: true}
}

func noLimits() *stubLimitProvider {
	return &stubLimitProvider{ok: false}
}

// --- tests ---

// Test 1: Bucket exhaustion — rapid calls hit cap → non-zero wait.
func TestThrottle_BucketExhaustion(t *testing.T) {
	now := time.Now()
	svc := NewService(Config{
		Credentials: activeResolver("cred-id-1", "test-pack"),
		Limits:      staticLimits(1, 2), // rps=1, burst=2
		Clock:       frozenClock(now),   // time frozen so no refill
	})

	ctx := context.Background()

	// First two calls should succeed (burst=2)
	for i := 0; i < 2; i++ {
		wait, _, err := svc.Throttle(ctx, "test-cred", "endpoint.class")
		if err != nil {
			t.Fatalf("call %d: unexpected error: %v", i, err)
		}
		if wait != 0 {
			t.Errorf("call %d: expected wait=0 within burst, got %v", i, wait)
		}
	}

	// Third call with time frozen → must wait
	wait, _, err := svc.Throttle(ctx, "test-cred", "endpoint.class")
	if err != nil {
		t.Fatalf("third call: unexpected error: %v", err)
	}
	if wait <= 0 {
		t.Errorf("third call: expected wait > 0 after burst exhausted, got %v", wait)
	}
}

// Test 2: Refill — after waiting the refill period, Throttle succeeds again.
func TestThrottle_Refill(t *testing.T) {
	base := time.Now()
	callCount := 0
	// Clock advances by 2s on each call after the first.
	svc := NewService(Config{
		Credentials: activeResolver("cred-id-1", "test-pack"),
		Limits:      staticLimits(1, 1), // rps=1, burst=1; refill = 1s per token
		Clock: func() time.Time {
			t := base
			if callCount > 0 {
				t = base.Add(time.Duration(callCount) * 2 * time.Second)
			}
			callCount++
			return t
		},
	})

	ctx := context.Background()

	// First call consumes the burst.
	_, _, err := svc.Throttle(ctx, "test-cred", "endpoint.class")
	if err != nil {
		t.Fatalf("first call: unexpected error: %v", err)
	}

	// Second call: time has advanced 2s (> 1s refill period) → should succeed with 0 wait.
	wait, _, err := svc.Throttle(ctx, "test-cred", "endpoint.class")
	if err != nil {
		t.Fatalf("second call (after refill): unexpected error: %v", err)
	}
	if wait != 0 {
		t.Errorf("after refill: expected wait=0, got %v", wait)
	}
}

// Test 3 (covered by throttler_race_test.go): concurrent caller correctness under -race.

// Test 4: Missing credential → ErrCredentialNotAvailable.
func TestThrottle_MissingCredential(t *testing.T) {
	svc := NewService(Config{
		Credentials: &stubCredentialResolver{canonicalID: "", namespace: "", active: false},
		Limits:      noLimits(),
	})

	_, _, err := svc.Throttle(context.Background(), "nonexistent", "class")
	if !errors.Is(err, ErrCredentialNotAvailable) {
		t.Fatalf("expected ErrCredentialNotAvailable, got %v", err)
	}
}

// Test 4b: CredentialResolver returns an error → propagated.
func TestThrottle_CredentialResolverError(t *testing.T) {
	resolverErr := errors.New("db timeout")
	svc := NewService(Config{
		Credentials: &stubCredentialResolver{err: resolverErr},
		Limits:      noLimits(),
	})

	_, _, err := svc.Throttle(context.Background(), "test-cred", "class")
	if !errors.Is(err, resolverErr) {
		t.Fatalf("expected resolver error to propagate, got %v", err)
	}
}

func TestThrottleResolved_UsesAdmittedIdentityWithoutCredentialLookup(t *testing.T) {
	resolverErr := errors.New("NamedCredential lookup must not run")
	limits := &recordingLimitProvider{rps: 10, burst: 20}
	svc := NewService(Config{
		Credentials: &stubCredentialResolver{err: resolverErr},
		Limits:      limits,
	})

	wait, release, err := svc.ThrottleResolved(t.Context(), ResolvedCredential{
		CanonicalID: "connected-mail/sha256:binding-generation",
		Namespace:   "mail-owner",
	}, " MAIL.Send ")
	if err != nil || wait != 0 || release == nil {
		t.Fatalf("resolved throttle wait=%v release_nil=%v error=%v", wait, release == nil, err)
	}
	release()
	if limits.gotNamespace != "mail-owner" || limits.gotEPC != "mail.send" {
		t.Fatalf("limit lookup = %q/%q", limits.gotNamespace, limits.gotEPC)
	}
}

func TestThrottleResolved_RejectsUncanonicalIdentity(t *testing.T) {
	svc := NewService(Config{Limits: noLimits()})
	for name, credential := range map[string]ResolvedCredential{
		"missing identity": {},
		"padded identity":  {CanonicalID: " binding", Namespace: "mail-owner"},
		"padded namespace": {CanonicalID: "binding", Namespace: "mail-owner "},
	} {
		t.Run(name, func(t *testing.T) {
			_, _, err := svc.ThrottleResolved(t.Context(), credential, "mail.send")
			if !errors.Is(err, ErrCredentialNotAvailable) {
				t.Fatalf("ThrottleResolved error = %v", err)
			}
		})
	}
}

// Test 5: Inactive connector pack → ErrCredentialNotAvailable.
func TestThrottle_InactivePack_FailsClosed(t *testing.T) {
	svc := NewService(Config{
		Credentials: inactiveResolver(),
		Limits:      noLimits(),
	})

	_, _, err := svc.Throttle(context.Background(), "any-cred", "any.class")
	if !errors.Is(err, ErrCredentialNotAvailable) {
		t.Fatalf("expected ErrCredentialNotAvailable for inactive pack, got %v", err)
	}
}

// Test 6: Second-connector reuse — noop-credential uses same Throttle() API
// identically to any other connector; no vendor literal in paths.
// (Vendor-literal absence is asserted by gate 3 in the CI script — this
// test covers the behavioral side: the API works generically.)
func TestThrottle_SecondConnectorReuse_NoopFixture(t *testing.T) {
	svc := NewService(Config{
		Credentials: activeResolver("noop-cred-id", "__test_noop__"),
		Limits:      staticLimits(10, 20),
	})

	ctx := context.Background()
	wait, _, err := svc.Throttle(ctx, "noop-credential", "probe")
	if err != nil {
		t.Fatalf("noop connector: unexpected error: %v", err)
	}
	// burst=20, first call → no wait.
	if wait != 0 {
		t.Errorf("noop connector first call: expected wait=0, got %v", wait)
	}
}

// Test 7 (pack-metadata SSoT): LimitProvider returns ok=false → default fallback applied.
func TestThrottle_DefaultFallback_WhenNoRow(t *testing.T) {
	now := time.Now()
	svc := NewService(Config{
		Credentials:  activeResolver("cred-id-1", "test-pack"),
		Limits:       noLimits(), // no row in DB
		DefaultRate:  5,
		DefaultBurst: 5,
		Clock:        frozenClock(now),
	})

	ctx := context.Background()
	// First 5 calls (= default burst) should succeed.
	for i := 0; i < 5; i++ {
		wait, _, err := svc.Throttle(ctx, "test-cred", "class")
		if err != nil {
			t.Fatalf("call %d: unexpected error with default burst: %v", i, err)
		}
		if wait != 0 {
			t.Errorf("call %d: expected wait=0 within default burst, got %v", i, wait)
		}
	}
	// 6th call → exhausted.
	wait, _, err := svc.Throttle(ctx, "test-cred", "class")
	if err != nil {
		t.Fatalf("6th call: unexpected error: %v", err)
	}
	if wait <= 0 {
		t.Errorf("6th call: expected wait > 0 after default burst exhausted, got %v", wait)
	}
}

// Test 8: Idle-TTL eviction — after IdleTTL of no access, a key is GC'd and
// the next access gets a fresh bucket. This is the idle-TTL replacement for
// the removed hard-LRU MaxKeys eviction.
func TestThrottle_IdleTTLEviction(t *testing.T) {
	base := time.Now()
	ttl := 5 * time.Minute
	svc := NewService(Config{
		Credentials: &autoMultiCredResolver{"test-pack"},
		Limits:      staticLimits(1, 1), // burst=1 so second call on same key waits
		IdleTTL:     ttl,
		Clock:       frozenClock(base),
	})

	ctx := context.Background()

	// Throttle "credA" once — consumes its single burst token.
	_, _, err := svc.Throttle(ctx, "credA", "endpoint")
	if err != nil {
		t.Fatalf("credA first call: %v", err)
	}

	// Second call immediately → bucket exhausted, wait > 0.
	wait, _, err := svc.Throttle(ctx, "credA", "endpoint")
	if err != nil {
		t.Fatalf("credA second call: %v", err)
	}
	if wait <= 0 {
		t.Errorf("credA second call: expected wait > 0 (burst exhausted, clock frozen), got %v", wait)
	}

	// Advance clock past IdleTTL; then touch "credB" to trigger lazy GC that evicts "credA".
	svc.clock = func() time.Time { return base.Add(ttl + time.Second) }
	_, _, err = svc.Throttle(ctx, "credB", "endpoint")
	if err != nil {
		t.Fatalf("credB (triggers GC of credA): %v", err)
	}

	// "credA" was evicted by GC; next access creates a fresh bucket → burst=1 → wait=0.
	wait, _, err = svc.Throttle(ctx, "credA", "endpoint")
	if err != nil {
		t.Fatalf("credA after idle-TTL eviction: %v", err)
	}
	if wait != 0 {
		t.Errorf("credA after idle-TTL eviction: expected fresh bucket (wait=0), got %v", wait)
	}
}

// Test 8b: DoesNotEvictWithinIdleTTL — a key accessed within the TTL window
// retains its existing limiter state (cache hit, not re-created).
// Uses a frozen clock on both calls so that refill cannot mask a spurious
// re-creation: if the bucket were incorrectly re-created, the fresh limiter
// would start full (wait=0) even with the clock frozen.
func TestThrottle_DoesNotEvictWithinIdleTTL(t *testing.T) {
	base := time.Now()
	ttl := 5 * time.Minute
	// Both calls use the same frozen time so the rate limiter cannot refill
	// between them. If the second call returns wait=0, the bucket was
	// incorrectly re-created (or the frozen-clock assertion is broken).
	svc := NewService(Config{
		Credentials: activeResolver("credA-id", "test-pack"),
		Limits:      staticLimits(1, 1), // burst=1
		IdleTTL:     ttl,
		Clock:       frozenClock(base),
	})

	ctx := context.Background()

	// First call consumes the burst token.
	_, _, err := svc.Throttle(ctx, "credA", "endpoint")
	if err != nil {
		t.Fatalf("credA first call: %v", err)
	}

	// Advance clock by a tiny amount — well within IdleTTL and well within
	// the refill window (1ms << 1s refill period for rps=1). Entry must NOT
	// be evicted; the cached limiter still has its burst consumed → wait > 0.
	svc.clock = func() time.Time { return base.Add(time.Millisecond) }

	// Second call: same cached limiter → burst already consumed → wait > 0.
	wait, _, err := svc.Throttle(ctx, "credA", "endpoint")
	if err != nil {
		t.Fatalf("credA second call within TTL: %v", err)
	}
	if wait <= 0 {
		t.Errorf("credA second call within TTL: expected wait > 0 (bucket state preserved, not re-created), got %v", wait)
	}
}

// Regression: GC must not evict a bucket whose scheduled-reservation horizon
// (debtUntil) is still in the future, even if lastUsed > IdleTTL. Otherwise a
// heavily contended key (queued reservations push timeToAct > IdleTTL) loses
// its outstanding debt on the next idle pass — codex Review PR #1519 round 8
// (HIGH). The fix is to OR the debt check with the idle check: evict only
// when BOTH are satisfied.
func TestThrottle_DebtHorizonGuardsAgainstEvictionWithOutstandingReservations(t *testing.T) {
	base := time.Now()
	ttl := 5 * time.Minute
	// rps=1 burst=1 means the second queued call has timeToAct = base + 1s.
	// We deliberately advance well past IdleTTL but before debtUntil to prove
	// that idle alone doesn't authorise eviction.
	svc := NewService(Config{
		Credentials: &autoMultiCredResolver{"test-pack"},
		Limits:      staticLimits(1, 1),
		IdleTTL:     ttl,
		Clock:       frozenClock(base),
	})
	ctx := context.Background()

	// First call consumes the only token; wait=0; debtUntil unchanged (no future-scheduled debt).
	_, _, err := svc.Throttle(ctx, "credA", "endpoint")
	if err != nil {
		t.Fatalf("credA first call: %v", err)
	}

	// Second call queues a reservation; wait > 0 → debtUntil advanced into the
	// future. With rps=1 burst=1, timeToAct ≈ base + 1s.
	wait2, _, err := svc.Throttle(ctx, "credA", "endpoint")
	if err != nil {
		t.Fatalf("credA second call: %v", err)
	}
	if wait2 <= 0 {
		t.Fatalf("credA second call: expected wait > 0 (burst exhausted), got %v", wait2)
	}

	// Stub a very long queue: many more reservations that push debtUntil far
	// out. Use direct cache poke to simulate the round-8 scenario where
	// queue depth pushes the next act-time past IdleTTL.
	svc.mu.Lock()
	if e, ok := svc.buckets.items[bucketKey{canonicalCredentialID: "credA-id", endpointClass: "endpoint"}]; ok {
		// Push debtUntil 2× IdleTTL into the future.
		e.debtUntil = base.Add(2 * ttl)
	} else {
		svc.mu.Unlock()
		t.Fatal("expected credA bucket present after two Throttle calls")
	}
	svc.mu.Unlock()

	// Advance clock by 1.5× IdleTTL (idle-only GC would fire here).
	svc.clock = func() time.Time { return base.Add(ttl + ttl/2) }

	// A new key call triggers gcLocked. With debt-aware GC, credA's entry
	// survives because debtUntil (2×ttl) > now (1.5×ttl).
	_, _, err = svc.Throttle(ctx, "credB", "endpoint")
	if err != nil {
		t.Fatalf("credB call: %v", err)
	}

	svc.mu.Lock()
	_, present := svc.buckets.items[bucketKey{canonicalCredentialID: "credA-id", endpointClass: "endpoint"}]
	svc.mu.Unlock()
	if !present {
		t.Error("credA evicted while debtUntil still in the future — GC must wait for debt to drain")
	}

	// Now advance past debtUntil (and well past idle). credA should be eligible.
	svc.clock = func() time.Time { return base.Add(3 * ttl) }

	// Trigger GC by touching the cache.
	_, _, err = svc.Throttle(ctx, "credC", "endpoint")
	if err != nil {
		t.Fatalf("credC call: %v", err)
	}

	svc.mu.Lock()
	_, stillPresent := svc.buckets.items[bucketKey{canonicalCredentialID: "credA-id", endpointClass: "endpoint"}]
	svc.mu.Unlock()
	if stillPresent {
		t.Error("credA should be evicted once debtUntil has passed and idle > ttl")
	}
}

// Regression: even when every reservation is immediate (wait=0), eviction
// must wait for the natural-refill horizon — the time the bucket would
// regenerate to full from its current balance at the configured rps. Without
// this, a low-rps + high-burst policy (e.g. burst=5, rps=1) can drain its
// burst in 5 immediate calls, sit idle for just over IdleTTL, get evicted,
// and regain a fresh full bucket while the vendor's real bucket has only
// refilled a few tokens. Codex/IANTHEREAL Review PR #1519 round 9 (P1).
func TestThrottle_RetainsBucketUntilNaturalRefillCompletes(t *testing.T) {
	base := time.Now()
	ttl := 100 * time.Millisecond // short TTL so the test isn't slow
	// burst=5, rps=1 → 5 seconds to refill from empty. IdleTTL=100ms.
	// burst/rps = 5s >> IdleTTL — exactly the IANTHEREAL counter-example.
	svc := NewService(Config{
		Credentials: &autoMultiCredResolver{"test-pack"},
		Limits:      staticLimits(1, 5),
		IdleTTL:     ttl,
		Clock:       frozenClock(base),
	})
	ctx := context.Background()

	// Drain all 5 burst tokens — every call is wait=0 (immediate).
	for i := 0; i < 5; i++ {
		wait, _, err := svc.Throttle(ctx, "credA", "endpoint")
		if err != nil {
			t.Fatalf("call %d: %v", i, err)
		}
		if wait != 0 {
			t.Errorf("call %d: expected immediate (wait=0), got %v", i, wait)
		}
	}

	// Advance clock past IdleTTL but well within the 5s refill window.
	svc.clock = func() time.Time { return base.Add(500 * time.Millisecond) }

	// Touch the cache via a different key to trigger gcLocked.
	_, _, _ = svc.Throttle(ctx, "credB", "endpoint")

	// credA must NOT be evicted — natural refill horizon (~5s) > now (500ms).
	svc.mu.Lock()
	_, present := svc.buckets.items[bucketKey{canonicalCredentialID: "credA-id", endpointClass: "endpoint"}]
	svc.mu.Unlock()
	if !present {
		t.Fatal("credA evicted before natural refill horizon — vendor bucket would still be drained")
	}

	// Advance past the natural-refill horizon. credA's cached bucket has now
	// refilled naturally to full, so eviction is semantically safe (a fresh
	// rate.NewLimiter on revisit would have the same token state).
	svc.clock = func() time.Time { return base.Add(10 * time.Second) }
	_, _, _ = svc.Throttle(ctx, "credC", "endpoint")

	svc.mu.Lock()
	_, stillPresent := svc.buckets.items[bucketKey{canonicalCredentialID: "credA-id", endpointClass: "endpoint"}]
	svc.mu.Unlock()
	if stillPresent {
		t.Error("credA should be evicted once natural refill horizon AND idle window have both passed")
	}
}

// Test: LimitProvider error propagates.
func TestThrottle_LimitProviderError(t *testing.T) {
	dbErr := errors.New("limit db unavailable")
	svc := NewService(Config{
		Credentials: activeResolver("cred-id-1", "test-pack"),
		Limits:      &stubLimitProvider{err: dbErr},
	})

	_, _, err := svc.Throttle(context.Background(), "test-cred", "class")
	if !errors.Is(err, dbErr) {
		t.Fatalf("expected limit provider error to propagate, got %v", err)
	}
}

// Regression: caller abort (ctx.Done() or other early-return) must not burn
// the reserved token. The Throttle API returns a release() function so that
// `defer release()` returns the unspent reservation to the bucket; without
// this, repeated timeouts would ratchet the bucket down to a permanently
// starved state. Codex Review on PR #1519 round 3 (HIGH).
func TestThrottle_ReleaseReturnsUnusedReservationToBucket(t *testing.T) {
	now := time.Now()
	svc := NewService(Config{
		Credentials: activeResolver("cred-id-1", "test-pack"),
		Limits:      staticLimits(1, 1), // burst=1: one token total
		Clock:       frozenClock(now),
	})

	ctx := context.Background()

	// First call consumes the only token; wait=0.
	wait1, release1, err := svc.Throttle(ctx, "test-cred", "endpoint")
	if err != nil {
		t.Fatalf("first call: %v", err)
	}
	if wait1 != 0 {
		t.Errorf("first call: expected wait=0 with burst=1, got %v", wait1)
	}

	// Second call reserves a future token; wait > 0 because frozen clock.
	wait2, release2, err := svc.Throttle(ctx, "test-cred", "endpoint")
	if err != nil {
		t.Fatalf("second call: %v", err)
	}
	if wait2 <= 0 {
		t.Errorf("second call: expected wait > 0 (burst exhausted, time frozen), got %v", wait2)
	}
	// Caller aborts before consuming the reservation — release returns the token.
	release2()

	// Third call: with the second's reservation returned, the third's wait
	// must NOT have ratcheted further into the future. Equivalent (or shorter)
	// wait proves release worked. Without release, every aborted Throttle
	// would push the bucket's timeToAct out by another 1/rps interval, eventually
	// stalling all callers.
	wait3, release3, err := svc.Throttle(ctx, "test-cred", "endpoint")
	if err != nil {
		t.Fatalf("third call: %v", err)
	}
	defer release3()
	if wait3 > wait2 {
		t.Errorf("third call after release(): wait = %v, must be <= wait2 = %v (release did not return the token)", wait3, wait2)
	}

	// Ensure release1 (after token was consumed at frozen now) is safe to call.
	release1()
}

// Regression: release() is safe to call after the reserved time slot has
// passed (caller proceeded with the request). It must NOT double-credit the
// bucket; rate.Reservation.CancelAt is a no-op when now > timeToAct (which
// is the normal production case: caller proceeds, real wall-clock advances
// past the reserved slot before defer release() runs).
func TestThrottle_ReleaseAfterProceedIsNoop(t *testing.T) {
	base := time.Now()
	callCount := 0
	// Clock advances 1ms per call. Inside Throttle, s.clock() is read twice
	// (ReserveN now + DelayFrom now); release closure reads it on invocation.
	// So by the time release1() runs, the simulated clock is past the first
	// reservation's timeToAct → CancelAt becomes a no-op.
	advancing := func() time.Time {
		t := base.Add(time.Duration(callCount) * time.Millisecond)
		callCount++
		return t
	}
	svc := NewService(Config{
		Credentials: activeResolver("cred-id-1", "test-pack"),
		Limits:      staticLimits(1, 1),
		Clock:       advancing,
	})
	ctx := context.Background()

	// Consume the burst.
	_, release1, err := svc.Throttle(ctx, "test-cred", "endpoint")
	if err != nil {
		t.Fatalf("first call: %v", err)
	}
	release1() // happy-path defer release(); clock has advanced → no-op.

	// Second Throttle: bucket should still be drained (release didn't double-credit).
	// Refill rate = 1/s; only a few ms have elapsed; second call must wait ~1s.
	wait, release2, err := svc.Throttle(ctx, "test-cred", "endpoint")
	if err != nil {
		t.Fatalf("second call: %v", err)
	}
	defer release2()
	if wait <= 0 {
		t.Errorf("second call after happy-path release(): expected wait > 0 (bucket should still be drained); got %v (release double-credited?)", wait)
	}
}

// Empirical contract test for x/time/rate v0.15.0 semantics that the throttle
// substrate relies on for tighten-without-trim correctness.
//
// Codex Review PR #1519 round 10 (HIGH) claimed: "Runtime can still spend the
// old burst immediately after a policy tighten". The empirical behavior of
// x/time/rate v0.15.0 contradicts that claim — advance() (called by every
// public read/reserve) caps tokens at the CURRENT lim.burst regardless of any
// larger preserved lim.tokens value. So no trim is needed: the next AllowN
// observes only `min(preserved-tokens, new-burst)` available.
//
// This test exists to lock that behavior in CI and document the answer for
// future readers; if a future x/time/rate version removes the cap, this test
// fails and we reintroduce explicit trimming (likely via chunked AllowN since
// the public API rejects n > burst per the same source).
func TestRateLimiterAdvanceCapsAtCurrentBurstOnTighten(t *testing.T) {
	now := time.Now()
	lim := rate.NewLimiter(rate.Limit(100), 100)
	if got := lim.TokensAt(now); got < 99 {
		t.Fatalf("fresh limiter: TokensAt(now) = %v, want ≈ 100", got)
	}

	// Tighten burst from 100 to 5. SetBurstAt preserves lim.tokens internally
	// (still 100), but advance() caps reads at the new burst.
	lim.SetBurstAt(now, 5)

	// Visible token balance is capped at new burst, NOT the preserved 100.
	if got := lim.TokensAt(now); got > 5.001 {
		t.Errorf("after SetBurstAt(now, 5): TokensAt(now) = %v, want ≤ 5 — advance() must cap", got)
	}

	// AllowN(now, 6): n > burst, rejected by reserveN's `n <= lim.burst` gate.
	if lim.AllowN(now, 6) {
		t.Error("AllowN(now, 6) succeeded — x/time/rate must reject n > burst")
	}

	// AllowN(now, 5): n == burst AND 5 tokens visibly available, succeeds.
	if !lim.AllowN(now, 5) {
		t.Error("AllowN(now, 5) failed — bucket has 5 visible tokens")
	}

	// AllowN(now, 1): bucket drained, must fail (no immediate availability).
	if lim.AllowN(now, 1) {
		t.Error("AllowN(now, 1) succeeded — bucket should be drained")
	}
}

// Regression: SetLimit + SetBurst preserve the cached limiter's current
// token balance, but x/time/rate.advance() caps the visible balance at the
// CURRENT burst on every public read. The substrate relies on that cap to
// enforce a tightened policy live; the older explicit-trim path was dead
// code that AllowN rejected silently for surplus > burst (codex round 10).
// Codex Review on PR #1519 round 5 (HIGH) + round 10 (HIGH).
func TestLimiterFor_TrimsSurplusTokensWhenBurstTightens(t *testing.T) {
	now := time.Now()
	svc := NewService(Config{
		Credentials: activeResolver("cred-id-1", "test-pack"),
		Limits:      staticLimits(100, 100), // initial: loose policy
		Clock:       frozenClock(now),
	})

	k := bucketKey{canonicalCredentialID: "cred-id-1", endpointClass: "endpoint.class"}

	// First lookup creates a full bucket at burst=100.
	lim := svc.limiterFor(k, 100, 100)
	if got, want := lim.Burst(), 100; got != want {
		t.Fatalf("initial burst = %d, want %d", got, want)
	}
	// Bucket starts full at the new-limiter default: TokensAt(now) ≈ burst.
	if tokens := lim.TokensAt(now); tokens < 99 {
		t.Fatalf("initial TokensAt(now) = %v, want ≈ 100 (fresh limiter starts full)", tokens)
	}

	// Pack tightens to burst=5. Without surplus trim, lim.Burst() would
	// say 5 but TokensAt(now) would still show ~100 → caller could emit
	// 100 immediate calls, overshooting the new quota 20x.
	lim2 := svc.limiterFor(k, 100, 5)
	if lim != lim2 {
		t.Fatalf("cache hit expected: lim (%p) != lim2 (%p)", lim, lim2)
	}
	if got, want := lim2.Burst(), 5; got != want {
		t.Errorf("after tighten: burst = %d, want %d", got, want)
	}
	// Critical assertion: surplus tokens trimmed to within the new burst.
	// Fractional tail < 1 token is acceptable (int truncation in trim).
	if tokens := lim2.TokensAt(now); tokens > float64(5)+1.0 {
		t.Errorf("after tighten: TokensAt(now) = %v, want <= 6 (new burst=5 + <1 fractional tail); pack-metadata SSoT requires the live bucket to reflect the tightened policy immediately", tokens)
	}
}

// Regression: the LimitProvider is the SSoT for limit values. When the
// authoritative source returns updated rps/burst for an existing
// (namespace, endpointClass), the cached limiter must adopt the new
// values on the next Throttle call rather than staying frozen at the
// values it was created with (otherwise the in-memory bucket becomes a
// second SSoT). Codex Review on PR #1519 round 1 (HIGH).
func TestLimiterFor_AdoptsUpdatedPackMetadataOnCacheHit(t *testing.T) {
	svc := NewService(Config{
		Credentials: activeResolver("cred-id-1", "test-pack"),
		Limits:      staticLimits(10, 20),
	})

	k := bucketKey{canonicalCredentialID: "cred-id-1", endpointClass: "endpoint.class"}

	// First lookup creates a limiter at the initial pack-authored values.
	lim1 := svc.limiterFor(k, 10, 20)
	if got, want := lim1.Limit(), rate.Limit(10); got != want {
		t.Errorf("initial limit = %v, want %v", got, want)
	}
	if got, want := lim1.Burst(), 20; got != want {
		t.Errorf("initial burst = %d, want %d", got, want)
	}

	// LimitProvider's authoritative source updates to a tighter policy.
	lim2 := svc.limiterFor(k, 1, 5)

	// Same cached instance (no eviction-and-recreate path).
	if lim1 != lim2 {
		t.Errorf("cache hit expected: lim1 (%p) != lim2 (%p)", lim1, lim2)
	}
	// New rps/burst reflected immediately, not at next restart.
	if got, want := lim2.Limit(), rate.Limit(1); got != want {
		t.Errorf("after update: limit = %v, want %v (cached bucket must adopt pack-metadata SSoT)", got, want)
	}
	if got, want := lim2.Burst(), 5; got != want {
		t.Errorf("after update: burst = %d, want %d (cached bucket must adopt pack-metadata SSoT)", got, want)
	}

	// Loosening direction also takes effect.
	lim3 := svc.limiterFor(k, 50, 100)
	if got, want := lim3.Limit(), rate.Limit(50); got != want {
		t.Errorf("after loosen: limit = %v, want %v", got, want)
	}
	if got, want := lim3.Burst(), 100; got != want {
		t.Errorf("after loosen: burst = %d, want %d", got, want)
	}
}

// Regression: bucket state must be per-canonical-credential, not per-pack-namespace.
// Two NamedCredentials owned by the same pack must NOT drain each other's
// token buckets — codex Review PR #1519 round 2 (HIGH).
func TestThrottle_SameNamespaceDifferentCredentials_HaveIndependentBuckets(t *testing.T) {
	now := time.Now()
	// Each credentialRef maps to a DIFFERENT canonicalID even though they share a pack.
	resolver := &canonicalMappingResolver{
		mapping: map[string]string{
			"credA": "id-credA",
			"credB": "id-credB",
		},
		namespace: "shared-pack",
		active:    true,
	}
	svc := NewService(Config{
		Credentials: resolver,
		Limits:      staticLimits(1, 1), // burst=1: 1 token per (cred, endpoint)
		Clock:       frozenClock(now),
	})
	ctx := context.Background()

	// credA consumes its single token.
	wait, _, err := svc.Throttle(ctx, "credA", "endpoint")
	if err != nil {
		t.Fatalf("credA: %v", err)
	}
	if wait != 0 {
		t.Errorf("credA first call: expected wait=0, got %v", wait)
	}

	// credB is a DIFFERENT credential in the SAME pack: it must have its own bucket.
	wait, _, err = svc.Throttle(ctx, "credB", "endpoint")
	if err != nil {
		t.Fatalf("credB: %v", err)
	}
	if wait != 0 {
		t.Errorf("credB first call (independent bucket): expected wait=0, got %v (would mean credA drained credB's bucket)", wait)
	}

	// credA's second call must exhaust the burst (frozen clock).
	wait, _, err = svc.Throttle(ctx, "credA", "endpoint")
	if err != nil {
		t.Fatalf("credA second call: %v", err)
	}
	if wait <= 0 {
		t.Errorf("credA second call: expected wait > 0 (burst exhausted), got %v", wait)
	}
}

// Regression: admin-installed credentials (OwnerPack="") must be throttle-
// eligible with default fallback — codex Review PR #1519 round 2 (HIGH);
// rejecting them outright made the substrate dead code in greenfield orgs.
func TestThrottle_AdminInstalledCredential_UsesDefaults(t *testing.T) {
	now := time.Now()
	svc := NewService(Config{
		// Admin-installed: canonicalID set, namespace empty, active=true.
		Credentials:  &stubCredentialResolver{canonicalID: "admin-cred-id", namespace: "", active: true},
		Limits:       noLimits(), // no pack-metadata row → LimitProvider returns ok=false
		DefaultRate:  10,
		DefaultBurst: 3,
		Clock:        frozenClock(now),
	})
	ctx := context.Background()

	// First 3 calls succeed (default burst).
	for i := 0; i < 3; i++ {
		wait, _, err := svc.Throttle(ctx, "admin-cred", "endpoint")
		if err != nil {
			t.Fatalf("call %d on admin-installed cred: %v (must be eligible, not fail-closed)", i, err)
		}
		if wait != 0 {
			t.Errorf("call %d: expected wait=0 within default burst, got %v", i, wait)
		}
	}
	// 4th call exhausted.
	wait, _, err := svc.Throttle(ctx, "admin-cred", "endpoint")
	if err != nil {
		t.Fatalf("4th call: %v", err)
	}
	if wait <= 0 {
		t.Errorf("4th call: expected wait > 0 after default burst exhausted, got %v", wait)
	}
}

// Round-11 regression: case-variant credentialRefs that resolve to the SAME
// canonical ID must share one bucket, not get independent buckets (quota bypass).
// _System_NamedCredential.DeveloperName uses utf8mb4_unicode_ci, so "MyCred"
// and "mycred" resolve to the same row (same Definition.ID = same canonicalID).
// If the throttler keyed on the raw ref, an attacker could exhaust the canonical
// bucket via "mycred" then immediately call again as "MyCred" from a fresh bucket.
func TestThrottle_CaseVariantCredentials_ShareCanonicalBucket(t *testing.T) {
	now := time.Now()
	// Both refs map to the SAME canonicalID — simulating utf8mb4_unicode_ci collation.
	resolver := &canonicalMappingResolver{
		mapping: map[string]string{
			"MyCred": "canonical-id-1", // same ID
			"mycred": "canonical-id-1", // same ID
		},
		namespace: "test-pack",
		active:    true,
	}
	svc := NewService(Config{
		Credentials: resolver,
		Limits:      staticLimits(1, 1), // burst=1: only one immediate call allowed
		Clock:       frozenClock(now),
	})
	ctx := context.Background()

	// First call using "MyCred" consumes the single burst token.
	wait1, _, err := svc.Throttle(ctx, "MyCred", "endpoint")
	if err != nil {
		t.Fatalf("MyCred first call: %v", err)
	}
	if wait1 != 0 {
		t.Errorf("MyCred first call: expected wait=0 with burst=1, got %v", wait1)
	}

	// Second call using "mycred" (case variant, same canonicalID) — must see
	// the SAME drained bucket, not a fresh one.
	wait2, _, err := svc.Throttle(ctx, "mycred", "endpoint")
	if err != nil {
		t.Fatalf("mycred second call: %v", err)
	}
	if wait2 <= 0 {
		t.Errorf("mycred second call: expected wait > 0 (shared bucket already drained by MyCred), got %v — case-variant refs must share one bucket to prevent quota bypass", wait2)
	}
}

// recordingLimitProvider records the (namespace, endpointClass) it was last
// queried with so a test can assert the throttler normalized them before the
// LimitProvider lookup.
type recordingLimitProvider struct {
	rps, burst           float64
	gotNamespace, gotEPC string
}

func (r *recordingLimitProvider) LimitFor(_ context.Context, namespace, endpointClass string) (float64, int, bool, error) {
	r.gotNamespace = namespace
	r.gotEPC = endpointClass
	return r.rps, int(r.burst), true, nil
}

// Regression: endpointClass must be normalized (lower+trim) at Throttle entry
// to match the canonical form connector_rate_limit_component.go writes to
// _System_ConnectorRateLimit. The normalized value must flow into the
// LimitProvider lookup. Codex Review PR #1537 round 1 (HIGH).
func TestThrottle_NormalizesEndpointClassForLimitLookup(t *testing.T) {
	now := time.Now()
	rec := &recordingLimitProvider{rps: 5, burst: 5}
	svc := NewService(Config{
		Credentials: activeResolver("cred-id", "test-pack"),
		Limits:      rec,
		Clock:       frozenClock(now),
	})

	_, release, err := svc.Throttle(context.Background(), "cred", "  CRM.Objects.Read  ")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	release()

	if rec.gotEPC != "crm.objects.read" {
		t.Errorf("LimitProvider got endpointClass %q, want normalized %q", rec.gotEPC, "crm.objects.read")
	}
}

// Regression: case/whitespace variants of the same endpointClass must share one
// bucket (not get separate buckets that each grant a fresh burst). Codex Review
// PR #1537 round 1 (HIGH).
func TestThrottle_EndpointClassVariants_ShareBucket(t *testing.T) {
	now := time.Now()
	svc := NewService(Config{
		Credentials: activeResolver("cred-id", "test-pack"),
		Limits:      staticLimits(1, 1), // burst=1: one token total
		Clock:       frozenClock(now),
	})
	ctx := context.Background()

	// First call drains the single token. Do NOT release (keep it consumed).
	wait1, _, err := svc.Throttle(ctx, "cred", "crm.objects.read")
	if err != nil {
		t.Fatalf("first call: %v", err)
	}
	if wait1 != 0 {
		t.Errorf("first call: expected wait=0 with burst=1, got %v", wait1)
	}

	// A messy case/whitespace variant must hit the SAME bucket (already drained)
	// → wait > 0. If it got a separate bucket it would return wait=0 (fresh burst).
	wait2, _, err := svc.Throttle(ctx, "cred", "  CRM.Objects.READ ")
	if err != nil {
		t.Fatalf("variant call: %v", err)
	}
	if wait2 <= 0 {
		t.Errorf("endpointClass variant: expected wait > 0 (shared drained bucket), got %v — case/whitespace variants must normalize to one bucket", wait2)
	}
}

// Regression: an empty/whitespace endpointClass must fail closed (not fall
// back to default quota in a separate empty-class bucket). Codex Review
// PR #1537 round 2 (HIGH).
func TestThrottle_EmptyEndpointClass_FailsClosed(t *testing.T) {
	svc := NewService(Config{
		Credentials: activeResolver("cred-id", "test-pack"),
		Limits:      staticLimits(5, 5),
	})
	for _, ec := range []string{"", "   ", "\t", "  \n "} {
		_, release, err := svc.Throttle(context.Background(), "cred", ec)
		if !errors.Is(err, ErrInvalidEndpointClass) {
			t.Errorf("endpointClass %q: expected ErrInvalidEndpointClass, got %v", ec, err)
		}
		release()
	}
}