// Package-level note: the bucket cache is a plain map with per-entry
// lastUsed timestamps (idle-TTL eviction) to avoid the correctness hazard
// of hard-LRU eviction — if a key were evicted while an outstanding
// rate.Reservation was still held by a caller, the new bucket created on
// revisit would silently diverge from those in-flight handles.
// golang.org/x/time/rate is already in go.mod and is the chosen bucket
// algorithm (lazy refill, no goroutine, race-tested, consistent with
// backend/internal/interfaces/middleware/rate_limit.go).
package connector_throttle

import (
	"context"
	"strings"
	"sync"
	"time"

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

// Throttler reserves capacity for an outbound connector call. Returns:
//   - wait: how long the caller should wait before issuing the call
//   - release: caller MUST `defer release()` so that if the caller aborts
//     (e.g. on ctx.Done()) before consuming the reserved capacity, the
//     token is returned to the bucket instead of being silently burned.
//     release() is safe to call always — it is a no-op once the reserved
//     time slot has passed (i.e. the caller proceeded with the request).
//   - err: ErrCredentialNotAvailable for missing/inactive credential;
//     LimitProvider / CredentialResolver IO errors propagate as-is.
//
// v1 LIMITATION (acknowledged): bucket state is in-process only. Multi-replica
// deployments multiply the effective quota by replica count. Cross-process
// aggregation is tracked as a named follow-up; until that follow-up ships, this
// substrate is NOT promoted to "the L0 quota SSoT" in NexusCRM docs.
type Throttler interface {
	Throttle(ctx context.Context, credentialRef, endpointClass string) (wait time.Duration, release func(), err error)
}

// ResolvedCredential is a secret-free credential identity already admitted by
// an authoritative runtime binding. CanonicalID owns the token bucket;
// Namespace selects the pack-authored endpoint policy. Callers must not derive
// either value from request input.
type ResolvedCredential struct {
	CanonicalID string
	Namespace   string
}

// ResolvedThrottler reserves capacity for a credential whose active identity
// has already been established by another exact authority, such as an
// immutable Catalog binding. It deliberately has no credentialRef lookup.
type ResolvedThrottler interface {
	ThrottleResolved(
		ctx context.Context,
		credential ResolvedCredential,
		endpointClass string,
	) (wait time.Duration, release func(), err error)
}

// noopRelease is returned for error paths where no reservation was made;
// callers can always `defer release()` without nil-checking.
var noopRelease = func() {}

// bucketKey is the in-process map key. Bucket state is keyed by
// canonicalCredentialID (the stable Definition.ID, not the raw credentialRef)
// so that case-variant DeveloperNames (e.g. "MyCred"/"mycred") that resolve
// to the same definition row share one bucket and cannot bypass per-credential
// quota. The pack namespace is used ONLY for LimitProvider policy lookup
// (multiple credentials in the same pack share the policy but not the bucket
// state). Round-11 codex HIGH fix.
type bucketKey struct {
	canonicalCredentialID string
	endpointClass         string
}

// ttlBucketEntry pairs a rate limiter with its last-access timestamp AND the
// latest scheduled reservation act-time. Entries are evicted by
// bucketCache.gcLocked only when BOTH:
//   - lastUsed is older than ttl (idle), AND
//   - debtUntil has passed (no future-scheduled reservations remain).
//
// debtUntil is necessary because rate.Limiter.ReserveN can queue a reservation
// arbitrarily far into the future when the bucket is heavily contended.
// Evicting on lastUsed alone (codex Review PR #1519 round 8 HIGH) would drop
// the bucket while in-flight reservations still have unspent token debt; a
// revisit would create a fresh full limiter and overshoot the vendor quota.
type ttlBucketEntry struct {
	limiter   *rate.Limiter
	lastUsed  time.Time
	debtUntil time.Time
}

// bucketCache is an unbounded map of *rate.Limiter keyed by bucketKey with
// debt-aware idle-TTL eviction. Correctness argument for TTL-based (not
// cap-based) eviction:
//
//   - GC requires BOTH idleness (now - lastUsed > ttl) AND drained debt
//     (now >= debtUntil). debtUntil tracks the latest scheduled-reservation
//     act-time; while in-flight reservations remain, the entry survives even
//     past the lastUsed-based idle window. This is what guards against the
//     round-8 hazard where heavy contention pushes wait > IdleTTL, the
//     idle-only GC drops the bucket, and release() handles orphan against
//     a fresh limiter.
//
//   - The vendor's refill window is also well under 1 hour for any reasonable
//     rps/burst setting, so rate.NewLimiter on a revisit of an evicted key
//     is semantically correct: the vendor's bucket is fully refilled by then.
//
// Memory bound: O(active-working-set). At NexusCRM's expected cardinality
// (credential count × endpoint-class count), this is well under 100K entries
// (~10 MB), comparable to the old hard-LRU cap of 1024 in practice.
type bucketCache struct {
	ttl   time.Duration
	items map[bucketKey]*ttlBucketEntry
}

// getOrCreate returns the limiter for k, creating one if absent.
// It also runs a lazy GC pass to evict idle entries before the lookup.
// Caller must hold the enclosing Service.mu.
func (c *bucketCache) getOrCreate(k bucketKey, lim *rate.Limiter, now time.Time) *rate.Limiter {
	c.gcLocked(now)
	if e, ok := c.items[k]; ok {
		e.lastUsed = now
		return e.limiter
	}
	c.items[k] = &ttlBucketEntry{limiter: lim, lastUsed: now}
	return lim
}

// recordReservation extends the bucket's scheduled-debt horizon to actAt.
// Called by Throttle after a successful ReserveN whose timeToAct lies in the
// future. Caller must hold s.mu. No-op if the key has already been evicted
// (e.g. by a race against another GC pass — the caller's release() closure
// still owns the underlying rate.Reservation, which is safe to cancel
// independently).
func (c *bucketCache) recordReservation(k bucketKey, actAt time.Time) {
	if e, ok := c.items[k]; ok && actAt.After(e.debtUntil) {
		e.debtUntil = actAt
	}
}

// gcLocked evicts entries that have BOTH been idle for longer than ttl AND
// have no scheduled debt remaining (now >= debtUntil). The debt-horizon guard
// is what makes TTL eviction safe even under heavy contention that pushes
// reservation timeToAct far past the lastUsed timestamp. Caller must hold s.mu.
func (c *bucketCache) gcLocked(now time.Time) {
	for k, e := range c.items {
		if now.Before(e.debtUntil) {
			continue
		}
		if now.Sub(e.lastUsed) > c.ttl {
			delete(c.items, k)
		}
	}
}

// Config is the constructor configuration for Service.
type Config struct {
	// Credentials resolves a credentialRef to (canonicalID, namespace, active).
	Credentials CredentialResolver
	// Limits looks up per-(namespace, endpointClass) policy rows.
	Limits LimitProvider
	// DefaultRate is the fallback rps when LimitProvider returns ok=false.
	// Hardcoded constant; NOT env-var-tunable in v1.
	DefaultRate float64
	// DefaultBurst is the fallback burst when LimitProvider returns ok=false.
	// Hardcoded constant; NOT env-var-tunable in v1.
	DefaultBurst int
	// IdleTTL is the duration after which an idle bucket entry is evicted.
	// 0 → 1 hour. Increasing this reduces GC churn at the cost of memory;
	// decreasing it is safe only if IdleTTL >> max(reservation wait window).
	IdleTTL time.Duration
	// Clock is injectable for tests; nil → time.Now.
	Clock func() time.Time
}

// Service implements Throttler.
type Service struct {
	creds   CredentialResolver
	limits  LimitProvider
	cfg     Config
	mu      sync.Mutex
	buckets *bucketCache
	clock   func() time.Time
}

// NewService constructs a Service from cfg. DefaultRate and DefaultBurst are
// hardcoded Go constants (10 rps, 20 burst); they are not env-var-tunable in
// v1 — the pack-metadata row is the only authorized override.
func NewService(cfg Config) *Service {
	if cfg.DefaultRate <= 0 {
		cfg.DefaultRate = 10
	}
	if cfg.DefaultBurst <= 0 {
		cfg.DefaultBurst = 20
	}
	if cfg.IdleTTL <= 0 {
		cfg.IdleTTL = time.Hour
	}
	lim := cfg.Limits
	if lim == nil {
		lim = NewNoopLimitProvider()
	}
	clk := cfg.Clock
	if clk == nil {
		clk = time.Now
	}
	return &Service{
		creds:  cfg.Credentials,
		limits: lim,
		cfg:    cfg,
		buckets: &bucketCache{
			ttl:   cfg.IdleTTL,
			items: make(map[bucketKey]*ttlBucketEntry),
		},
		clock: clk,
	}
}

// Throttle implements Throttler. It:
//  1. Resolves credentialRef → (canonicalID, namespace, active) via CredentialResolver.
//     Returns ErrCredentialNotAvailable if missing or inactive.
//  2. Looks up (namespace, endpointClass) policy in LimitProvider.
//     Falls back to DefaultRate/DefaultBurst when no row exists.
//  3. Reserves one token from the in-process bucket and returns:
//     - wait: how long the caller must wait before proceeding
//     - release: caller MUST `defer release()` so that ctx.Done() abort
//     paths return the reserved token to the bucket. Calling release()
//     after consuming the reserved time slot is a no-op (rate.Reservation
//     semantics), so `defer release()` is correct for both happy-path
//     and abort-path callers.
//
// The bucket key is (canonicalID, endpointClass), NOT (credentialRef, endpointClass).
// This ensures case-variant credentialRefs (utf8mb4_unicode_ci) that resolve to
// the same definition row share one bucket (round-11 codex HIGH fix).
func (s *Service) Throttle(ctx context.Context, credentialRef, endpointClass string) (time.Duration, func(), error) {
	if s == nil || s.creds == nil {
		return 0, noopRelease, ErrCredentialNotAvailable
	}
	canonicalID, namespace, active, err := s.creds.ResolveCredentialPack(ctx, credentialRef)
	if err != nil {
		return 0, noopRelease, err
	}
	if !active {
		return 0, noopRelease, ErrCredentialNotAvailable
	}
	if canonicalID == "" {
		canonicalID = credentialRef // active legacy resolvers should always return an ID
	}
	return s.ThrottleResolved(ctx, ResolvedCredential{
		CanonicalID: canonicalID,
		Namespace:   namespace,
	}, endpointClass)
}

// ThrottleResolved implements ResolvedThrottler. Unlike Throttle, it does not
// consult CredentialResolver: the caller has already proved that the exact
// credential identity is active. This keeps Catalog-bound credentials on their
// Catalog authority instead of requiring a duplicate NamedCredential row.
func (s *Service) ThrottleResolved(
	ctx context.Context,
	credential ResolvedCredential,
	endpointClass string,
) (time.Duration, func(), error) {
	if s == nil || credential.CanonicalID == "" ||
		credential.CanonicalID != strings.TrimSpace(credential.CanonicalID) ||
		credential.Namespace != strings.TrimSpace(credential.Namespace) {
		return 0, noopRelease, ErrCredentialNotAvailable
	}

	// Normalize endpointClass to the canonical form the pack-installer writes
	// to _System_ConnectorRateLimit (connector_rate_limit_component.go lowers +
	// trims on storage). Without this, a case/whitespace variant misses the
	// authored row, silently falls back to the default quota, and gets a
	// separate bucket — breaking the pack-metadata SSoT. Use the normalized
	// value for BOTH the LimitProvider lookup AND the bucket key (codex
	// Review PR #1537 round 1 HIGH).
	endpointClass = strings.ToLower(strings.TrimSpace(endpointClass))
	// Fail closed on a blank endpointClass: a "" / whitespace caller bug would
	// otherwise skip every pack-authored row, fall back to the default quota,
	// and get a separate empty-class bucket — bypassing the pack-metadata SSoT
	// (codex Review PR #1537 round 2 HIGH).
	if endpointClass == "" {
		return 0, noopRelease, ErrInvalidEndpointClass
	}

	rps, burst, ok, err := s.limits.LimitFor(ctx, credential.Namespace, endpointClass)
	if err != nil {
		return 0, noopRelease, err
	}
	if !ok {
		rps = s.cfg.DefaultRate
		burst = s.cfg.DefaultBurst
	}

	// Use the authority-provided canonical identity, never a raw credential
	// reference, so aliases cannot create independent buckets.
	k := bucketKey{canonicalCredentialID: credential.CanonicalID, endpointClass: endpointClass}
	lim := s.limiterFor(k, rps, burst)
	now := s.clock()
	r := lim.ReserveN(now, 1)
	if !r.OK() {
		r.Cancel()
		return 0, noopRelease, ErrCredentialNotAvailable
	}
	wait := r.DelayFrom(now)
	// Extend the bucket's retention horizon to whichever is later: the
	// reservation's act-time (round 8: protects queued reservations under
	// contention) OR the natural refill horizon — the time at which the
	// bucket would regenerate to full from its current token balance at the
	// configured rps. Without the refill horizon, an immediate reservation
	// (wait=0) on a low-rps + high-burst policy could leave debtUntil at zero,
	// let IdleTTL expire, evict the entry, and recreate a fresh full limiter
	// even though the vendor's real bucket has only partially refilled
	// (codex/IANTHEREAL Review PR #1519 round 9 P1).
	horizon := s.naturalRefillCompleteAt(lim, rps, burst, now)
	if act := now.Add(wait); wait > 0 && act.After(horizon) {
		horizon = act
	}
	if horizon.After(now) {
		s.mu.Lock()
		s.buckets.recordReservation(k, horizon)
		s.mu.Unlock()
	}
	// CancelAt(now) returns the unspent reservation to the bucket iff the
	// caller has NOT yet consumed the reserved time slot. The clock closure
	// captures s.clock at call time so tests with frozen / advanced clocks
	// see deterministic behavior.
	release := func() { r.CancelAt(s.clock()) }
	return wait, release, nil
}

// naturalRefillCompleteAt returns the time at which lim's current token
// balance would naturally regenerate to burst, given the configured rps.
// Used to extend the bucket's retention horizon so eviction never recreates
// a fuller limiter than the vendor's real bucket has refilled. Returns now
// when rps <= 0 (cannot refill — eviction is safe) or when the bucket is
// already at or above burst (eviction creates an equivalent bucket).
func (s *Service) naturalRefillCompleteAt(lim *rate.Limiter, rps float64, burst int, now time.Time) time.Time {
	if rps <= 0 {
		return now
	}
	tokens := lim.TokensAt(now)
	toRefill := float64(burst) - tokens
	if toRefill <= 0 {
		return now
	}
	seconds := toRefill / rps
	return now.Add(time.Duration(seconds * float64(time.Second)))
}

// limiterFor returns the existing limiter for k, or creates and caches one.
// On a cache hit the limiter's rate and burst are reconciled to (rps, burst)
// so that the LimitProvider remains the single source of truth: when a future
// pack-installer (PR-B) updates the authoritative policy, the cached limiter
// adopts the new rps/burst on the next Throttle call without waiting for
// process restart or TTL eviction.
//
// Tightening (new burst < preserved token balance) is enforced live by
// x/time/rate's advance() function, which caps the visible token balance at
// the CURRENT burst on every read (TokensAt / AllowN / ReserveN). After
// SetBurstAt(now, smaller), the next public read returns at most the new
// burst regardless of any larger internal lim.tokens value preserved by
// SetBurstAt. No explicit surplus trim is needed — and indeed not possible
// via the public API, because AllowN/ReserveN reject n > burst by contract.
// Verified empirically in TestRateLimiterAdvanceCapsAtCurrentBurstOnTighten.
// (Earlier rounds 5/10 of codex review here.)
//
// Caller must NOT hold s.mu when calling this (it acquires internally).
func (s *Service) limiterFor(k bucketKey, rps float64, burst int) *rate.Limiter {
	s.mu.Lock()
	defer s.mu.Unlock()
	want := rate.Limit(rps)
	now := s.clock()
	newLim := rate.NewLimiter(want, burst)
	lim := s.buckets.getOrCreate(k, newLim, now)
	if lim != newLim {
		// Cache hit: reconcile rps/burst to the current policy. advance()
		// inside x/time/rate enforces the new burst as the per-call ceiling
		// on the next public read.
		if lim.Limit() != want {
			lim.SetLimitAt(now, want)
		}
		if lim.Burst() != burst {
			lim.SetBurstAt(now, burst)
		}
	}
	return lim
}

var _ Throttler = (*Service)(nil)
var _ ResolvedThrottler = (*Service)(nil)