/**
 * AI module — `AiGateway` provider + budget guard (Phase 12, TASK-0121).
 *
 * Wires the `@factory/ai` gateway into the API:
 *
 *   - **config resolution** (per workspace, per call): the workspace's
 *     `settings.ai` (Zod-validated `AiConfigSchema`) over the platform env
 *     defaults (`AI_PROVIDER`/`AI_BASE_URL`/`AI_API_KEY`/`AI_MODEL`) with
 *     per-field fallback. Provider selection is configuration-only: swapping
 *     `AI_PROVIDER` (or a workspace's `settings.ai.provider`) changes zero
 *     application code (ROADMAP Phase 12 exit criterion).
 *   - **usage ledger**: every completed (or failed-after-consume) gateway
 *     call appends an `ai_messages` row — in the caller's conversation when
 *     the feature has one (chat), else a hidden per-call conversation
 *     (`title: "ai-usage:<capability>"`). The monthly rollup (the budget
 *     guard + the `usage.rollup` worker job) sums `ai_messages` per
 *     workspace since month-start — one ledger, no second table.
 *   - **budget guard** ({@link createAiBudgetGuard}): the service-level
 *     guard every AI endpoint awaits BEFORE calling the gateway — a
 *     Postgres-backed month-to-date total vs
 *     `workspace.usageLimits.aiMonthlyTokens` (unset/0 = unlimited), a
 *     Redis fixed-window rate cap, and a 24 h-throttled admin notification
 *     (Redis `SET … NX EX` slot — globally consistent across replicas) +
 *     the `ai.budget_exceeded` audit event. The gateway's own
 *     submit-time check (per-process in-memory ledger) stays as a fast
 *     backstop.
 *   - **error translation** ({@link mapAiError}): the gateway throws the
 *     `@factory/shared` envelope; it is re-raised as the API's `ApiError`
 *     with the canonical code — the budget 402 surfaces as a TOP-LEVEL
 *     `AI_BUDGET_EXCEEDED` (the shared package carries it in
 *     `details.code`; the API's error table is extensible by design —
 *     recorded decision in `@factory/ai/src/usage.ts` +
 *     `docs/api/authentication.md` §9.1).
 *
 * **Mock provider (CI/dev, recorded decision):** with `AI_PROVIDER=mock`
 * the registry returns a deterministic `MockProvider` — the CI capability
 * suite (TASK-0121) and local dev run the full gateway surface offline.
 * The mock is dev/test configuration, never a production provider value.
 */
import {
  AiConfigSchema,
  AiGateway,
  AnthropicAdapter,
  BudgetService,
  CustomAdapter,
  GoogleAdapter,
  MockProvider,
  OpenAIAdapter,
  OssAdapter,
  resolveAiConfig,
  resolveKeyRef,
  type ProviderAdapter,
  type ProviderId,
  type ResolvedAiConfig,
  UsageLedger,
  type UsageRecord,
} from "@factory/ai";
import { ApiError as SharedApiError, ErrorCodes as SharedErrorCodes } from "@factory/shared";
import { z } from "zod";

import type { Request } from "express";
import type { PrismaClient } from "@prisma/client";
import type Redis from "ioredis";
import { ApiError } from "../../common/errors.js";
import { logger } from "../../common/logger.js";
import { loadEnv } from "../../config/env.js";

/** DI token for the gateway (the module constructs it once at boot). */
export const AI_GATEWAY = "AI_GATEWAY";
/** DI token for the service-level budget guard. */
export const AI_BUDGET_GUARD = "AI_BUDGET_GUARD";

/**
 * The dimension the OpenSearch content index's vector field uses. The
 * gateway's embed result is checked against it (fail fast with a config
 * error, not a corrupt index at query time — TASK-0121 acceptance). The
 * default matches the search package's `DEFAULT_VECTOR_DIMENSION`; the
 * capability suite sets it to the mock's dimension.
 */
export function embedDimension(): number {
  const raw = process.env.SEARCH_VECTOR_DIMENSION || process.env.AI_EMBEDDING_DIMENSION;
  const n = raw ? Number.parseInt(raw, 10) : NaN;
  return Number.isFinite(n) && n > 0 ? n : 1536;
}

export interface AiGatewayDeps {
  /**
   * Budget-exceeded hook (invoked by the gateway's in-memory backstop when
   * it trips; the module no-ops it — the service-level guard is the
   * enforcement + notification point). At most once per workspace per 24 h
   * (the `BudgetService` throttle).
   */
  onBudgetExceeded: (workspaceId: string, used: number, limit: number) => Promise<void> | void;
  /**
   * Usage sink: persists a completed/failed gateway call. The module
   * implements this (ai_messages ledger-row upsert + conversation
   * bookkeeping).
   */
  ledgerSink: (record: UsageRecord) => Promise<void>;
}

const envAiProviderSchema = z.enum(["openai", "anthropic", "google", "oss", "custom", "mock"]);

/**
 * TASK-0122 (E2E): the mock-provider outage arming (single-shot, PER WORKSPACE).
 *
 * The mock is created fresh on every gateway call (the registry factory), so
 * a RUNNING API process cannot be switched into "fail the next call" via
 * `process.env` (the E2E suite talks to a live server, unlike the in-process
 * integration suite). Instead the test-seam endpoint
 * (`POST /internal/workspaces/:workspaceId/ai/mock-fail`, service-token
 * guarded) arms the workspace, and the NEXT mock the registry builds FOR THAT
 * WORKSPACE consumes it (single-shot per workspace).
 *
 * Why per-workspace (not process-global): Playwright runs its desktop +
 * mobile projects IN PARALLEL against the same API process, and each E2E test
 * uses its own `freshWorkspace`. A process-global arming would be consumed by
 * whichever request landed first — a happy-path test from the other project
 * could eat this test's arming (or worse, a happy test could receive the
 * error). Scoping by workspace makes the arming deterministic regardless of
 * project concurrency.
 *
 * Only meaningful when `AI_PROVIDER=mock`; a no-op for real providers.
 */
const mockOutageArmings = new Map<string, { status: number; deltas: number }>();

/**
 * Test seam: make the NEXT mock chat call FOR `workspaceId` fail MID-stream
 * after `deltas` deltas (with a retryable `status`). Exactly ONE request in
 * that workspace fails — the SSE stream carries the in-band `error` delta +
 * terminal `done` — and the next request (the UI's Retry button, a fresh
 * gateway call in the same workspace) succeeds. Returns the armed status.
 */
export function armMockFailures(workspaceId: string, status = 500, deltas = 2): number {
  mockOutageArmings.set(workspaceId, { status, deltas });
  return status;
}

/** Drop any pending arming for a workspace (called by `resetWorkspaceCache`). */
export function clearMockOutageArming(workspaceId: string): void {
  mockOutageArmings.delete(workspaceId);
}

// TASK-0123 (test seam): make the NEXT mock WRITING call for `workspaceId`
// append a blockId the prompt never contained — the deterministic seam for
// the server-side "unknown ID → drop + note, never applied" validation.
const mockUnknownIdArmings = new Map<string, string>();

/** Arm the next mock writing call for `workspaceId` to emit `unknownId`. */
export function armMockUnknownBlockId(workspaceId: string, unknownId: string): void {
  mockUnknownIdArmings.set(workspaceId, unknownId);
}

/** Drop any pending arming for a workspace (called by `resetWorkspaceCache`). */
export function clearMockUnknownBlockId(workspaceId: string): void {
  mockUnknownIdArmings.delete(workspaceId);
}

// TASK-0124 (test seam): make the NEXT mock DATABASE-GENERATION call for
// `workspaceId` emit a deliberately INVALID tool call (a non-generatable
// property type / an unknown propertyId / an unparseable formula) — the
// deterministic seam for the "zod/engine validation fails → retry once →
// 422, no partial write" path.
const mockDbInvalidArmings = new Map<string, string>();

/** Arm the next mock database call for `workspaceId` to emit an invalid call. */
export function armMockDbInvalidStructure(workspaceId: string, action: "generate" | "extract" | "formula" = "generate"): void {
  mockDbInvalidArmings.set(workspaceId, action);
}

/** Drop any pending arming for a workspace (called by `resetWorkspaceCache`). */
export function clearMockDbInvalidStructure(workspaceId: string): void {
  mockDbInvalidArmings.delete(workspaceId);
}

// TASK-0126 (test seam): make the NEXT mock PAGE-GENERATION call for
// `workspaceId` emit a failing plan step (`failStep`, a 1-based step number
// whose injected step references an unknown ref → the executor's rollback
// path) and/or fetch the plan slowly (`slowMs`, so the service's total-timeout
// watchdog fires → the `ai_timeout` path). Both are armed via the internal
// test-seam endpoint (single-shot PER WORKSPACE, consumed by the next generate
// chat call built for that workspace).
const mockGenerateArmings = new Map<string, { failStep?: string; slowMs?: string }>();

/** Arm the next mock generate call for `workspaceId` (the timeout/rollback seams). */
export function armMockGenerateSeam(
  workspaceId: string,
  opts: { failStep?: number; slowMs?: number },
): { failStep?: string; slowMs?: string } {
  const arming = mockGenerateArmings.get(workspaceId) ?? {};
  if (opts.failStep !== undefined) arming.failStep = String(opts.failStep);
  if (opts.slowMs !== undefined) arming.slowMs = String(opts.slowMs);
  mockGenerateArmings.set(workspaceId, arming);
  return arming;
}

/** Drop any pending arming for a workspace (called by `resetWorkspaceCache`). */
export function clearMockGenerateSeam(workspaceId: string): void {
  mockGenerateArmings.delete(workspaceId);
}

// T-172…T-175 (test seam): make the assistant mock calls for `workspaceId`
// emit a deliberately INVALID tool call (`hostile` → the "validate → retry
// once → 422, no write" path) and/or a forged candidate id (`forgeId` → the
// "no fabricated refs → drop, never use" path). PERSISTENT per workspace (like
// the database-invalid arming: the retry-once makes two chat calls for the
// same capability, so the arming must survive the first). Cleared by
// `resetWorkspaceCache`.
const mockAssistantArmings = new Map<string, { hostile: boolean; forgeId: boolean }>();

/** Arm the assistant mock for `workspaceId` (hostile and/or forgeId). */
export function armMockAssistantSeam(workspaceId: string, opts: { hostile?: boolean; forgeId?: boolean }): void {
  const arming = mockAssistantArmings.get(workspaceId) ?? { hostile: false, forgeId: false };
  if (opts.hostile !== undefined) arming.hostile = opts.hostile;
  if (opts.forgeId !== undefined) arming.forgeId = opts.forgeId;
  mockAssistantArmings.set(workspaceId, arming);
}

/** Drop any pending arming for a workspace (called by `resetWorkspaceCache`). */
export function clearMockAssistantSeam(workspaceId: string): void {
  mockAssistantArmings.delete(workspaceId);
}

/**
 * The platform-level `maxOutputTokens` default (env `AI_MAX_OUTPUT_TOKENS`).
 * Unset → the gateway's 4096 default (no behavior change for other
 * deployments). See the recorded decision in `resolveConfig`.
 */
function platformMaxOutputTokens(): number {
  const n = Number(process.env.AI_MAX_OUTPUT_TOKENS);
  return Number.isFinite(n) && n > 0 ? Math.floor(n) : 4096;
}

/**
 * Adapter stream timeouts (env-overridable; defaults = the adapter's built-in
 * values, so unset env changes nothing):
 *
 *   - `AI_STREAM_IDLE_MS` — the max gap between streamed BYTES before the
 *     stream aborts (adapter default 30 s). Recorded decision (2025): the OSS
 *     reasoning model streams a long SILENT thinking phase (no tool-call
 *     deltas until the plan JSON starts) — a 39 s gap was observed between the
 *     last thinking byte and the first plan byte, which the 30 s default turns
 *     into a mid-stream 504 the gateway cannot retry (partial output
 *     already yielded). 90 s covers the observed silence with margin without
 *     hiding a genuinely dead stream.
 *   - `AI_STREAM_TIMEOUT_MS` — the fetch-level cap (adapter default 60 s;
 *     covers the response phase, not the whole stream).
 */
function adapterTimeouts(): { timeoutMs?: number; idleMs?: number } {
  const t: { timeoutMs?: number; idleMs?: number } = {};
  const to = Number(process.env.AI_STREAM_TIMEOUT_MS);
  if (Number.isFinite(to) && to > 0) t.timeoutMs = Math.floor(to);
  const idle = Number(process.env.AI_STREAM_IDLE_MS);
  if (Number.isFinite(idle) && idle > 0) t.idleMs = Math.floor(idle);
  return t;
}

/** The env platform defaults (the per-field fallback floor). */
function envDefaults(): {
  provider: "openai" | "anthropic" | "google" | "oss" | "custom";
  chatModel: string;
  embedModel: string;
  baseUrl?: string;
  apiKeyRef?: string;
} {
  const env = loadEnv();
  const parsed = envAiProviderSchema.safeParse(env.AI_PROVIDER);
  // "mock" resolves to the "custom" family; the registry inspects the raw
  // env to swap in the MockProvider (configuration, not code).
  const provider = !parsed.success || parsed.data === "mock" ? "custom" : parsed.data;
  return {
    provider,
    chatModel: env.AI_MODEL || "gpt-4o",
    embedModel: process.env.AI_EMBEDDING_MODEL || "text-embedding-3-small",
    ...(env.AI_BASE_URL ? { baseUrl: env.AI_BASE_URL } : {}),
    ...(env.AI_API_KEY ? { apiKeyRef: "env:AI_API_KEY" } : {}),
  };
}

/**
 * Resolve the API key for a resolved config. `secretsRef` is a NAME, never a
 * literal: `env:NAME` resolves from the process environment; a bare name
 * resolves against the provider's env key (this phase's secret store is the
 * environment — the encrypted `provider_secrets` store is enterprise
 * scope). Raw key values never appear in config, logs, or audits.
 */
function apiKeyFor(config: ResolvedAiConfig): string | undefined {
  const ref = config.secretsRef;
  if (!ref) return undefined;
  if (ref.startsWith("env:")) return resolveKeyRef(ref, process.env as Record<string, string | undefined>);
  return process.env.AI_API_KEY;
}

export interface WorkspaceAiRow {
  settings: unknown;
  usageLimits: unknown;
  billingPlan: string;
}

/**
 * The guard's budget for a workspace: `usageLimits.aiMonthlyTokens` (the
 * admin override — the task's `aiMonthlyTokens` field) wins; the workspace
 * `settings.ai.budgets.tokensPerMonth` is the config-level default.
 * Unset / 0 = unlimited (all current plans).
 */
export function tokensPerMonthFor(row: WorkspaceAiRow, settingsAi: { budgets?: { tokensPerMonth?: number } } | null): number {
  const usageLimits = (row.usageLimits ?? {}) as { aiMonthlyTokens?: unknown };
  const fromLimits =
    typeof usageLimits.aiMonthlyTokens === "number" && Number.isFinite(usageLimits.aiMonthlyTokens)
      ? Math.max(0, usageLimits.aiMonthlyTokens)
      : 0;
  const fromSettings =
    typeof settingsAi?.budgets?.tokensPerMonth === "number" ? Math.max(0, settingsAi.budgets.tokensPerMonth) : 0;
  return Math.max(fromLimits, fromSettings);
}

/**
 * Build the gateway for the module. The gateway's own submit-time guard
 * uses its (per-process) in-memory ledger as a fast backstop; the
 * Postgres-backed, replica-consistent enforcement is the service-level
 * {@link AiBudgetGuard} every AI endpoint awaits before the gateway call.
 */
export function buildAiGateway(deps: AiGatewayDeps): AiGateway {
  const ledger = new UsageLedger({
    sink: async (record: UsageRecord) => {
      try {
        await deps.ledgerSink(record);
      } catch (err) {
        // The ledger must never take a request down: usage loss is logged,
        // the nightly rollup reconciles.
        logger.warn(`ai usage sink failed (${record.workspaceId}): ${(err as Error).message}`);
      }
    },
  });

  const budgets = new BudgetService(ledger);

  const resolveConfig = (workspaceId: string): ResolvedAiConfig => {
    const row = cachedWorkspaceRow(workspaceId);
    if (!row) {
      // Services prime the cache before any gateway call (a cold miss means
      // the workspace vanished between prime and call — 404).
      throw new SharedApiError(SharedErrorCodes.NOT_FOUND, { message: "Workspace not found.", status: 404 });
    }
    const settings = (row.settings ?? {}) as { ai?: unknown };
    const parsed = AiConfigSchema.safeParse(settings.ai ?? {});
    const workspaceCfg = parsed.success ? parsed.data : {};
    const config = resolveAiConfig(workspaceCfg, null, envDefaults());
    // The registry reads `workspaceId` to scope the per-workspace mock-outage
    // arming (test seam) — see `mockOutageArmings` above.
    config.workspaceId = workspaceId;
    config.budgets = {
      ...config.budgets,
      tokensPerMonth: tokensPerMonthFor(row, parsed.success ? parsed.data : null),
      // Platform-level output-token default (env `AI_MAX_OUTPUT_TOKENS`).
      //
      // Recorded decision (2025, root cause of a silent plan truncation): the
      // gateway clamps every chat request to `budgets.maxOutputTokens`
      // (default 4096). The OSS reasoning model this deployment runs
      // (qwen3.8-27B via vLLM) emits ~9 k REASONING tokens before a complex
      // generation plan of ~6 k more — a 4096 clamp cuts the plan JSON
      // mid-stream, the arguments no longer parse, and the run dies. The env
      // var raises the platform default for deployments that need it (unset →
      // the 4096 default, so other deployments are untouched); a workspace's
      // own explicit `settings.ai.budgets.maxOutputTokens` still wins.
      maxOutputTokens:
        parsed.success &&
        (parsed.data as { budgets?: { maxOutputTokens?: number } }).budgets?.maxOutputTokens != null
          ? (config.budgets.maxOutputTokens)
          : platformMaxOutputTokens(),
    };
    return config;
  };

  const registry = {
    get(providerId: ProviderId, config: ResolvedAiConfig): ProviderAdapter {
      const env = loadEnv();
      const key = apiKeyFor(config);
      const baseUrl = config.endpoint?.baseUrl ?? env.AI_BASE_URL ?? "https://api.openai.com/v1";
      // CI/dev mock (recorded decision): the capability suite runs the whole
      // gateway surface offline. Deviates from the production provider list
      // on purpose — it is a test adapter, selected by test config only.
      if (env.AI_PROVIDER === "mock") {
        const mock = new MockProvider(providerId === "anthropic" ? "anthropic" : "custom", embedDimension());
        // TASK-0122: optional per-delta pacing so mid-stream-cancellation
        // tests can land an abort BETWEEN deltas (0 = instant, the default).
        const delay = Number(process.env.AI_MOCK_DELTA_DELAY_MS);
        if (Number.isFinite(delay) && delay > 0) mock.deltaDelayMs = delay;
        // TASK-0122 (E2E): arm a deterministic provider outage FOR THIS
        // WORKSPACE. The mock is created PER gateway call (the registry
        // factory), so a RUNNING API process cannot be flipped via
        // process.env — the test-seam endpoint
        // (`POST /internal/workspaces/:workspaceId/ai/mock-fail`) arms the
        // workspace and the next mock built for it consumes the arming
        // (single-shot PER WORKSPACE — see `mockOutageArmings`). Scoping by
        // workspace (via `config.workspaceId`) keeps it deterministic even
        // when Playwright's desktop + mobile projects hit the API in parallel.
        //
        // The failure is MID-stream (a few deltas, then a retryable 5xx):
        // the gateway never re-streams partial output (it would duplicate
        // content), so the SSE endpoint emits an in-band `error` delta +
        // terminal `done {reason:"error"}` → the panel's retryable banner.
        // A PRE-stream failure would be useless here: the gateway's retry
        // loop reuses the same adapter instance, so it auto-recovers inside
        // the request and the user never sees an error.
        if (config.workspaceId) {
          const armed = mockOutageArmings.get(config.workspaceId);
          if (armed && armed.deltas > 0) {
            mock.failAfterDeltas = armed.deltas;
            mock.failStatus = armed.status;
            mockOutageArmings.delete(config.workspaceId); // single-shot
          }
          // TASK-0123 seam: the next WRITING call emits a forged blockId
          // (single-shot per workspace, like the outage arming).
          const unknownId = mockUnknownIdArmings.get(config.workspaceId);
          if (unknownId) {
            mockUnknownIdArmings.delete(config.workspaceId); // single-shot
            const originalChat = mock.chat.bind(mock);
            mock.chat = (req: Parameters<typeof originalChat>[0]) =>
              originalChat({
                ...req,
                metadata: { ...req.metadata, ...(req.metadata?.mode === "writing" ? { forceUnknownId: unknownId } : {}) },
              });
          }
          // TASK-0124 seam: the DATABASE-GENERATION calls for `workspaceId`
          // emit a deliberately INVALID tool call — the "validation fails →
          // retry once → 422, no partial write" path.
          //
          // PERSISTENT (unlike the single-shot outage / unknown-id armings):
          // the retry-once makes TWO chat calls for the SAME action, so the
          // arming must survive the first call for the 422 to actually fire.
          // It is scoped to the armed ACTION (generate/extract/formula), so a
          // hostile `generate` does not poison a later `extract` in the same
          // workspace. Tests clear it via `resetWorkspaceCache` / a fresh
          // workspace.
          const dbInvalidAction = mockDbInvalidArmings.get(config.workspaceId);
          if (dbInvalidAction) {
            const originalChat2 = mock.chat.bind(mock);
            mock.chat = (req: Parameters<typeof originalChat2>[0]) =>
              originalChat2({
                ...req,
                metadata: { ...req.metadata, ...(req.metadata?.mode === "database" ? { hostile: dbInvalidAction } : {}) },
              });
          }
          // TASK-0126 seam: the next PAGE-GENERATION call emits the armed
          // failing plan step and/or the armed plan latency (single-shot per
          // workspace, like the outage arming — generate is one chat call per
          // run, so no persistence is needed).
          const genSeam = mockGenerateArmings.get(config.workspaceId);
          if (genSeam) {
            mockGenerateArmings.delete(config.workspaceId); // single-shot
            const originalChat3 = mock.chat.bind(mock);
            mock.chat = (req: Parameters<typeof originalChat3>[0]) =>
              originalChat3({
                ...req,
                metadata: {
                  ...req.metadata,
                  ...(req.metadata?.mode === "generate"
                    ? { ...(genSeam.failStep ? { generateFailStep: genSeam.failStep } : {}), ...(genSeam.slowMs ? { generateSlowMs: genSeam.slowMs } : {}) }
                    : {}),
                },
              });
          }
          // T-172…T-175 seam: the ASSISTANT calls for `config.workspaceId`
          // emit a hostile (invalid) tool call and/or a forged candidate id —
          // persistent, like the database-invalid arming (retry-once).
          const assistantSeam = mockAssistantArmings.get(config.workspaceId);
          if (assistantSeam) {
            const originalChat4 = mock.chat.bind(mock);
            mock.chat = (req: Parameters<typeof originalChat4>[0]) =>
              originalChat4({
                ...req,
                metadata: {
                  ...req.metadata,
                  ...(req.metadata?.mode === "assistant"
                    ? {
                        ...(assistantSeam.hostile ? { hostile: "assistant" } : {}),
                        ...(assistantSeam.forgeId ? { forgeId: "1" } : {}),
                      }
                    : {}),
                },
              });
          }
        }
        return mock;
      }
      // Env-tunable stream timeouts for every real adapter (see
      // `adapterTimeouts` — defaults unchanged when the env is unset).
      const timeouts = adapterTimeouts();
      switch (providerId) {
        case "openai":
          return new OpenAIAdapter({ baseUrl, ...timeouts, ...(key ? { apiKey: key } : {}) });
        case "anthropic":
          return new AnthropicAdapter({ ...timeouts, ...(key ? { apiKey: key } : { apiKey: "" }) });
        case "google":
          return new GoogleAdapter({ ...timeouts, ...(key ? { apiKey: key } : { apiKey: "" }) });
        case "oss":
          return new OssAdapter({ baseUrl, ...timeouts, ...(key ? { apiKey: key } : {}) });
        case "custom":
          return new CustomAdapter({ baseUrl, ...timeouts, ...(key ? { apiKey: key } : {}), extraHeaders: config.endpoint?.extraHeaders });
      }
    },
  };

  return new AiGateway({ resolveConfig, registry, budgets, ledger });
}

// ---------------------------------------------------------------------------
// Workspace-row microtask cache (the sync resolveConfig contract)
// ---------------------------------------------------------------------------

const workspaceRowCache = new Map<string, WorkspaceAiRow>();
const cacheTimestamps = new Map<string, number>();
const WORKSPACE_CACHE_TTL_MS = 30_000;

/**
 * The synchronous `resolveConfig` callback needs the workspace row without an
 * await: every AI service primes this cache (30 s TTL) before its gateway
 * call. A cold miss in `resolveConfig` → 404 (the service already checked
 * membership, so a cold miss means the workspace vanished mid-request).
 */
export async function primeWorkspaceCache(prisma: PrismaClient, workspaceId: string): Promise<WorkspaceAiRow> {
  const hit = workspaceRowCache.get(workspaceId);
  if (hit && Date.now() - (cacheTimestamps.get(workspaceId) ?? 0) < WORKSPACE_CACHE_TTL_MS) return hit;
  const row = await prisma.workspace.findUnique({
    where: { id: workspaceId },
    select: { settings: true, usageLimits: true, billingPlan: true },
  });
  if (!row) throw ApiError.notFound("Workspace not found.");
  const wsRow = row as unknown as WorkspaceAiRow;
  workspaceRowCache.set(workspaceId, wsRow);
  cacheTimestamps.set(workspaceId, Date.now());
  return wsRow;
}

/** Test hook: drop the cache (new budget/settings take effect immediately). */
export function resetWorkspaceCache(): void {
  workspaceRowCache.clear();
  cacheTimestamps.clear();
  mockOutageArmings.clear();
  mockUnknownIdArmings.clear();
  mockDbInvalidArmings.clear();
  mockGenerateArmings.clear();
  mockAssistantArmings.clear();
}

function cachedWorkspaceRow(workspaceId: string): WorkspaceAiRow | null {
  if (Date.now() - (cacheTimestamps.get(workspaceId) ?? 0) >= WORKSPACE_CACHE_TTL_MS) {
    workspaceRowCache.delete(workspaceId);
    cacheTimestamps.delete(workspaceId);
  }
  return workspaceRowCache.get(workspaceId) ?? null;
}

// ---------------------------------------------------------------------------
// Service-level budget guard (Postgres-backed, replica-consistent)
// ---------------------------------------------------------------------------

export interface AiBudgetGuard {
  /**
   * Await BEFORE any gateway call. Throws (shared envelope, mapped by
   * {@link mapAiError} at the service boundary):
   *   - `RATE_LIMITED` (429) — per-workspace 60 s fixed window over Redis;
   *   - `AI_BUDGET_EXCEEDED` (402) — month-to-date ≥ the monthly budget
   *     (estimating this call's output at 2×maxOutputTokens, arch §7.2).
   * On budget exhaustion the admin notification + audit fire (throttled to
   * one per workspace per 24 h via a Redis `SET NX EX` slot).
   */
  check(workspaceId: string, maxOutputTokens?: number, req?: Request): Promise<void>;
  /** Month-to-date tokens (for the ping diagnostic + dashboards). */
  usedTokens(workspaceId: string): Promise<number>;
  /** The resolved monthly budget for the workspace (0 = unlimited). */
  monthlyBudget(workspaceId: string): Promise<number>;
}

export function createAiBudgetGuard(opts: {
  prisma: PrismaClient;
  redis: Redis;
  onBudgetExceeded: (workspaceId: string, used: number, limit: number, req?: Request) => Promise<void> | void;
}): AiBudgetGuard {
  const { prisma, redis, onBudgetExceeded } = opts;

  function monthStart(): Date {
    const d = new Date();
    return new Date(d.getFullYear(), d.getMonth(), 1);
  }

  async function usedTokens(workspaceId: string): Promise<number> {
    const rows = await prisma.aiMessage.aggregate({
      where: { conversation: { workspaceId }, createdAt: { gte: monthStart() } },
      _sum: { tokensIn: true, tokensOut: true },
    });
    return (rows._sum.tokensIn ?? 0) + (rows._sum.tokensOut ?? 0);
  }

  async function monthlyBudget(workspaceId: string): Promise<number> {
    const row = await primeWorkspaceCache(prisma, workspaceId);
    const settings = (row.settings ?? {}) as { ai?: unknown };
    const parsed = AiConfigSchema.safeParse(settings.ai ?? {});
    return tokensPerMonthFor(row, parsed.success ? parsed.data : null);
  }

  async function check(workspaceId: string, maxOutputTokens = 4096, req?: Request): Promise<void> {
    const row = await primeWorkspaceCache(prisma, workspaceId);
    const settings = (row.settings ?? {}) as { ai?: unknown };
    const parsed = AiConfigSchema.safeParse(settings.ai ?? {});
    const cfg = parsed.success ? parsed.data : null;
    const limit = tokensPerMonthFor(row, cfg);
    const requestsPerMinute = typeof cfg?.budgets?.requestsPerMinute === "number" ? cfg.budgets.requestsPerMinute : 60;

    // 1) rate window (per-workspace 60 s fixed window over Redis; the
    //    per-USER 30 msg/min edge cap is the AI limiter — TASK-0122).
    if (requestsPerMinute > 0) {
      const minute = Math.floor(Date.now() / 60_000);
      const key = `ai:rl:ws:${workspaceId}:${minute}`;
      let count = 0;
      try {
        count = (await redis.incr(key)) as number;
        if (count === 1) await redis.expire(key, 120);
      } catch (err) {
        // Redis blip: degrade open (the per-process gateway backstop still
        // rate-limits; the API never crashes on a Redis failure).
        logger.warn(`ai rate window unreadable (${workspaceId}): ${(err as Error).message}`);
      }
      if (count > requestsPerMinute) {
        throw new SharedApiError(SharedErrorCodes.RATE_LIMITED, {
          message: "AI rate limit reached. Try again shortly.",
          details: { code: "RATE_LIMITED", limit: requestsPerMinute, windowSeconds: 60 },
        });
      }
    }

    // 2) monthly token budget (0 = unlimited). The estimate (2×
    //    maxOutputTokens, arch §7.2) keeps a huge generation from crossing
    //    the line before the stream starts.
    if (limit > 0) {
      const used = await usedTokens(workspaceId);
      if (used + 2 * maxOutputTokens >= limit) {
        if (await acquireNotificationSlot(redis, workspaceId)) {
          try {
            await onBudgetExceeded(workspaceId, used, limit, req);
          } catch (err) {
            logger.warn(`ai budget notification failed (${workspaceId}): ${(err as Error).message}`);
          }
        }
        throw new SharedApiError(SharedErrorCodes.AI_ERROR, {
          status: 402,
          message: "Monthly AI token budget reached. Ask an admin.",
          details: { code: "AI_BUDGET_EXCEEDED", reason: "monthly_token_budget_exceeded", used, limit },
        });
      }
    }
  }

  return { check, usedTokens, monthlyBudget };
}

/**
 * The 24 h notification throttle: a Redis `SET NX EX 86400` slot (atomic,
 * replica-consistent). Redis failure → allow the notification (fail open:
 * a duplicate email is preferable to a missed budget alert).
 */
async function acquireNotificationSlot(redis: Redis, workspaceId: string): Promise<boolean> {
  const key = `ai:budget:notified:${workspaceId}`;
  try {
    const ok = await redis.set(key, "1", "EX", 24 * 3600, "NX");
    return ok === "OK";
  } catch {
    return true;
  }
}

// ---------------------------------------------------------------------------
// Shared-envelope → API ApiError translation
// ---------------------------------------------------------------------------

const SHARED_CODE_TO_API: Record<string, string> = {
  VALIDATION_ERROR: "VALIDATION_ERROR",
  UNAUTHORIZED: "UNAUTHORIZED",
  FORBIDDEN: "FORBIDDEN",
  NOT_FOUND: "NOT_FOUND",
  CONFLICT: "CONFLICT",
  RATE_LIMITED: "RATE_LIMITED",
  NOT_IMPLEMENTED: "NOT_IMPLEMENTED",
  SEARCH_UNAVAILABLE: "SERVICE_UNAVAILABLE",
  AI_ERROR: "SERVICE_UNAVAILABLE",
  TIMEOUT: "SERVICE_UNAVAILABLE",
  INTERNAL: "INTERNAL_ERROR",
};

/**
 * Translate a gateway/guard throw into the API's ApiError. The budget 402
 * arrives as shared `AI_ERROR` + `details.code = AI_BUDGET_EXCEEDED` (the
 * frozen shared enum has no 402 code) — it is re-raised as a TOP-LEVEL
 * `AI_BUDGET_EXCEEDED` (the API's canonical table, extensible by design).
 */
export function mapAiError(err: unknown): ApiError {
  if (err instanceof ApiError) return err;
  const shared = err as Partial<SharedApiError> | null;
  if (shared && typeof shared.code === "string" && typeof shared.status === "number") {
    const details =
      shared.details && typeof shared.details === "object" && !Array.isArray(shared.details)
        ? [shared.details as Record<string, unknown>]
        : undefined;
    const budget = details?.find((d) => d?.code === "AI_BUDGET_EXCEEDED") as
      | { code: string; reason?: string; used?: number; limit?: number }
      | undefined;
    if (budget) {
      return new ApiError("AI_BUDGET_EXCEEDED", 402, shared.message ?? "AI budget exceeded.", [
        { reason: budget.reason ?? "monthly_token_budget_exceeded", used: budget.used ?? 0, limit: budget.limit ?? 0 },
      ]);
    }
    const code = SHARED_CODE_TO_API[shared.code];
    if (!code) {
      logger.warn(`unmapped AI gateway error code: ${shared.code}`);
      return ApiError.internal("AI request failed.");
    }
    return new ApiError(code as ApiError["code"], shared.status, shared.message ?? "AI request failed.", details);
  }
  logger.warn(`unmapped AI gateway error: ${err instanceof Error ? err.message : String(err)}`);
  return ApiError.internal("AI request failed.");
}