/**
 * AAO Gateway Phase 2b — sliding-window rate limiter.
 *
 * Per-key sliding window of 60 seconds. The implementation is:
 *  - in-memory only (process-local; multi-instance setups would need
 *    Redis — explicitly out of scope until Phase 3)
 *  - Map<keyId, number[]> where the value is an append-only list of
 *    request timestamps (ms since epoch)
 *  - on `tryAdmit`: sweep entries outside the window, then compare
 *    length to the rpm cap; either push & admit, or refuse
 *  - pending request counters (`requests` only — no token deltas) are
 *    accumulated separately and flushed to the DB every 30s by
 *    `flushToDb`, batched so the DB sees one UPSERT per (key, period)
 *    instead of one per request
 *
 * Lossy by design: a crash between two flushes drops at most 30s of
 * request counters. The counter table is for analytics + the GET
 * /usage response — not for billing or load-bearing logic — so the
 * trade-off is fine.
 *
 * Memory cap: the inner timestamp array is bounded at 2× the rate cap
 * (oldest entries are evicted FIFO on each call). Without this a misuse
 * pattern (admit-without-purge) would let the array grow unboundedly.
 */
import type { Request, Response, NextFunction, RequestHandler } from 'express';
import type { Repository } from '../db/repository.js';
import type { AuthenticatedRequest } from './auth.js';
import { currentPeriodYearMonth } from './period.js';
import { logger } from '../logger.js';

export const RATE_LIMIT_WINDOW_MS = 60_000;
/** Default cadence at which pending request counters are flushed to the DB. */
export const DEFAULT_FLUSH_INTERVAL_MS = 30_000;
/** Hard cap on the number of distinct keys we track. Prevents memory growth on key churn. */
export const MAX_TRACKED_KEYS = 10_000;

export interface RateLimitAdmitResult {
  admitted: boolean;
  /** Slots remaining in the current window (0 if at the cap). */
  remaining: number;
  /**
   * Wall-clock time when the OLDEST in-window timestamp falls outside
   * the window — i.e., the moment one slot opens up. Caller uses this
   * for the Retry-After header. Equals nowMs when there's room.
   */
  resetAtMs: number;
  /** Count of timestamps currently in the window (for observability). */
  recentRequests: number;
}

export interface RateLimiter {
  /**
   * Window-only admission check. Pushes the timestamp into the per-key
   * window iff admitted, but does NOT increment the pending-counter
   * snapshot used by flushToDb. Use `recordSuccess` from the response
   * lifecycle to record an actually-served request — that way upstream
   * 5xx / abort / client-disconnect don't inflate the analytics counter.
   *
   * The window timestamp IS pushed eagerly on admit (rather than on
   * success) so the rate limit still bounds peak in-flight load even
   * when a key's requests are all failing — otherwise a misbehaving
   * upstream would let the same key hammer the gateway uncapped.
   */
  tryAdmit(keyId: string, rpmLimit: number | null, nowMs?: number): RateLimitAdmitResult;
  /**
   * Record one served request against the pending counter. Called from
   * the stream-proxy's finally block (via the bootstrap usage hook) on
   * any upstream response, including non-2xx — the request consumed
   * backend capacity so it should appear in the usage history. NOT
   * called for pre-upstream rejects (auth 401, budget 402, this
   * limiter's own 429) or for upstream fetch failures (502 from us).
   */
  recordSuccess(keyId: string): void;
  /** Snapshot pending requests + flush to the DB. Safe to call at any cadence. */
  flushToDb(): Promise<void>;
  /** Begin a setInterval that calls flushToDb every `intervalMs` (default 30s). */
  startFlushInterval(intervalMs?: number): void;
  /** Stop the flush interval and clear in-memory state. */
  stop(): void;
  /** Test-only: returns the current count of distinct tracked keys. */
  trackedKeyCount(): number;
}

export interface RateLimiterDeps {
  /** Repository slice used to flush request counts. */
  repo: Pick<Repository, 'incrementGatewayKeyUsage'>;
  /** Test hook: how to read the current UTC period (defaults to live clock). */
  period?: () => string;
}

export function createRateLimiter(deps: RateLimiterDeps): RateLimiter {
  // Per-key window of timestamps.
  const windows = new Map<string, number[]>();
  // Pending request counters since the last DB flush.
  let pending = new Map<string, number>();
  let flushTimer: ReturnType<typeof setInterval> | null = null;
  const periodFn = deps.period ?? ((): string => currentPeriodYearMonth());

  function sweepInPlace(arr: number[], cutoffMs: number): void {
    // arr is roughly time-ordered (we always push at the tail) but can
    // be lightly out of order if `nowMs` is passed by tests in arbitrary
    // order. Find the first index whose timestamp is still in-window and
    // splice everything before it. This is O(n) worst case, but n is
    // bounded by 2 * rpmLimit so the cost is negligible.
    let dropTo = 0;
    while (dropTo < arr.length && arr[dropTo]! <= cutoffMs) dropTo++;
    if (dropTo > 0) arr.splice(0, dropTo);
  }

  function tryAdmit(keyId: string, rpmLimit: number | null, nowMs?: number): RateLimitAdmitResult {
    const now = nowMs ?? Date.now();
    const cutoff = now - RATE_LIMIT_WINDOW_MS;

    if (rpmLimit === null || rpmLimit <= 0) {
      // Unlimited: skip the in-memory window so we don't waste memory
      // on the array. The pending counter is bumped by recordSuccess
      // when the response actually completes.
      return { admitted: true, remaining: Number.POSITIVE_INFINITY, resetAtMs: now, recentRequests: 0 };
    }

    // Memory cap: evict the LEAST-RECENTLY-ADMITTED tracked key so the
    // new key can be tracked with real enforcement. The pre-F11
    // behavior admitted-without-tracking, which let a key churn pattern
    // silently disable rate limiting for every new key once we hit the
    // cap — opposite of the safe default.
    //
    // True LRU (not FIFO): each tryAdmit on an existing key re-bumps
    // it to the tail of `windows` (see the else branch below). Without
    // that re-bump, an attacker could spray 10k throwaway keys to
    // wipe a hot key's window — the hot key's counter would reset and
    // rate-limit enforcement would silently fail for the next 60s.
    // Same pattern as the touch-deduper in auth.ts.
    if (!windows.has(keyId) && windows.size >= MAX_TRACKED_KEYS) {
      const oldest = windows.keys().next().value;
      if (oldest !== undefined) {
        windows.delete(oldest);
        logger.debug(
          `[gateway-rate] evicted oldest tracked key=${oldest} to track key=${keyId} at MAX_TRACKED_KEYS=${MAX_TRACKED_KEYS}`,
        );
      }
    }

    let arr = windows.get(keyId);
    if (!arr) {
      arr = [];
      windows.set(keyId, arr);
    } else {
      // LRU re-bump: move the existing entry to the tail so churn
      // can't evict a frequently-admitted (hot) key. Map preserves
      // insertion order, so delete+set is the cheapest way to move.
      windows.delete(keyId);
      windows.set(keyId, arr);
    }
    sweepInPlace(arr, cutoff);

    if (arr.length >= rpmLimit) {
      // resetAtMs = the oldest in-window timestamp + RATE_LIMIT_WINDOW_MS.
      // That's the moment it ages out and a slot opens. arr[0] is the
      // oldest because we always append at the tail.
      const oldest = arr[0]!;
      const resetAtMs = oldest + RATE_LIMIT_WINDOW_MS;
      return { admitted: false, remaining: 0, resetAtMs, recentRequests: arr.length };
    }

    arr.push(now);
    // Bound the array even when rpmLimit is unbounded(ish). We keep
    // up to 2× the cap to allow for chunked bursts at the edge of the
    // window without truncating a still-in-window timestamp.
    if (arr.length > rpmLimit * 2) {
      arr.splice(0, arr.length - rpmLimit * 2);
    }
    return { admitted: true, remaining: rpmLimit - arr.length, resetAtMs: now, recentRequests: arr.length };
  }

  function recordSuccess(keyId: string): void {
    pending.set(keyId, (pending.get(keyId) ?? 0) + 1);
  }

  async function flushToDb(): Promise<void> {
    if (pending.size === 0) return;
    // Snapshot + reset so admissions during the flush land in the next batch.
    const snapshot = pending;
    pending = new Map<string, number>();
    const period = periodFn();
    for (const [keyId, count] of snapshot) {
      try {
        deps.repo.incrementGatewayKeyUsage({ keyId, period, requests: count });
      } catch (e) {
        // Best-effort — re-queue would risk runaway memory if the DB is
        // permanently down. Log and drop; the next request will create
        // a fresh pending entry anyway.
        logger.warn(
          `[gateway-rate] flushToDb failed for key=${keyId} count=${count} err=${e instanceof Error ? e.message : String(e)}`,
        );
      }
    }
  }

  function startFlushInterval(intervalMs?: number): void {
    if (flushTimer) return; // idempotent: subsequent calls are no-ops
    const ms = intervalMs ?? DEFAULT_FLUSH_INTERVAL_MS;
    flushTimer = setInterval(() => {
      // Fire-and-forget: rejection in flushToDb is caught + logged
      // internally so the timer never sees an unhandled error.
      void flushToDb();
    }, ms);
    // unref so the timer doesn't keep the process alive in tests
    if (typeof flushTimer.unref === 'function') flushTimer.unref();
  }

  function stop(): void {
    if (flushTimer) {
      clearInterval(flushTimer);
      flushTimer = null;
    }
    windows.clear();
    pending.clear();
  }

  function trackedKeyCount(): number {
    return windows.size;
  }

  return { tryAdmit, recordSuccess, flushToDb, startFlushInterval, stop, trackedKeyCount };
}

export interface RateLimitMiddlewareDeps {
  limiter: RateLimiter;
  /** Lookup of the current effective rate limit for a key. Returns null for unlimited. */
  getRateLimit: (keyId: string) => number | null;
  /**
   * Phase 3b: optional metrics handle. When set, every 429 reject path
   * increments `rateLimitRejectionsTotal{team}` AND
   * `requestsTotal{status="rate_limited"}` so Grafana can show both the
   * rejection rate and its proportion of total traffic per team.
   */
  metrics?: import('../metrics/gateway-metrics.js').GatewayMetrics;
}

/**
 * Express middleware that enforces the per-key rpm cap. Mounted after
 * the auth + budget chain in gateway/server.ts.
 *
 * Skips silently when:
 *  - No gatewayAuth on the request (defensive; auth would have 401'd).
 *  - The matched key came from config-only (no keyId, no DB-tracked limit).
 *  - getRateLimit returns null (= unlimited for this key).
 *
 * On reject responds with HTTP 429, a structured `detail`, and the
 * standard `Retry-After` header in seconds (rounded UP so the client
 * doesn't retry early).
 */
export function buildRateLimitMiddleware(deps: RateLimitMiddlewareDeps): RequestHandler {
  return function gatewayRateLimitMiddleware(req: Request, res: Response, next: NextFunction): void {
    const auth = (req as AuthenticatedRequest).gatewayAuth;
    if (!auth || auth.source !== 'db' || !auth.keyId) {
      next();
      return;
    }
    const rpmLimit = deps.getRateLimit(auth.keyId);
    const result = deps.limiter.tryAdmit(auth.keyId, rpmLimit);
    if (result.admitted) {
      next();
      return;
    }
    const retryAfterSec = Math.max(1, Math.ceil((result.resetAtMs - Date.now()) / 1000));
    res.setHeader('Retry-After', String(retryAfterSec));
    res.status(429).json({
      error: 'rate_limited',
      detail:
        `team=${auth.team} limit_rpm=${rpmLimit} reset_at=${new Date(result.resetAtMs).toISOString()}`,
    });
    if (deps.metrics) {
      try {
        deps.metrics.rateLimitRejectionsTotal.labels({ team: auth.team }).inc();
        // Also emit on requests_total so the per-team request total
        // includes rejected attempts — operators routinely query
        // sum(requests_total) by team for traffic share.
        deps.metrics.requestsTotal
          .labels({ team: auth.team, backend: 'none', model: 'none', status: 'rate_limited' })
          .inc();
      } catch {
        /* metrics never affect the response */
      }
    }
  };
}
