/**
 * @factory/ai — AiGateway (ai-architecture §2.3–§2.4, TASK-0121).
 *
 * The only place provider types are imported. Capability surface:
 *
 *   - `chat(params)` → `AsyncIterable<ChatChunk>` — tool calling is a
 *     parameter of chat, not a separate method
 *   - `chatCollected(params)` → `Promise<ChatResult>` (collects the stream)
 *   - `embed(params)` → `EmbedResult` (Anthropic is routed to the fallback
 *     provider — it has no embedding API)
 *   - `resolve(workspaceId)` → the resolved config (for `POST /ai/ping`)
 *
 * Reliability (arch §2.4):
 *   - retries: 429/5xx/network, exponential backoff 1 s → 4 s → 16 s, max 3 —
 *     **only if no output has been yielded yet** (re-streaming partial output
 *     would duplicate content)
 *   - fallback: after retries exhausted → one attempt on the fallback
 *     provider/model (usage metered to the fallback provider)
 *   - circuit breaker: per provider, 5 consecutive failures → open 30 s
 *     (half-open probe)
 *   - budgets: submit-time check via `BudgetService` (fast fail, 402/429)
 *   - usage: every completed or failed-after-consume call appends an
 *     `ai_usage` record via the injected ledger (arch §7.1)
 *
 * **Recorded decision (spec conflict, rule 29):** TASK-0121 names the codes
 * `AI_BUDGET_EXCEEDED`/`SERVICE_UNAVAILABLE`, which the frozen
 * `@factory/shared` `ErrorCodes` does not define. The gateway uses the shared
 * envelope (`AI_ERROR`, and `RATE_LIMITED` where the shared code exists) and
 * carries the task's code in `details.code` (see `usage.ts` module doc); the
 * API layer maps `details.code` onto the canonical table.
 */

import { ApiError, ErrorCodes } from "@factory/shared";

import type { ResolvedAiConfig } from "./config.js";
import { ProviderHttpError } from "./providers/http.js";
import type {
  ChatChunk,
  ChatMessage,
  ChatRequest,
  ChatResult,
  EmbedResult,
  ProviderAdapter,
  ProviderId,
  ToolSpec,
} from "./types.js";
import { validateTools } from "./tools.js";
import type { Budgets, UsageLedger } from "./usage.js";import { BudgetService } from "./usage.js";

export interface ChatParams {
  workspaceId: string;
  userId?: string;
  capability?: string;
  messages: ChatMessage[];
  model?: string;
  tools?: ToolSpec[];
  toolChoice?: ChatRequest["toolChoice"];
  temperature?: number;
  maxTokens?: number;
  stream?: boolean;
  metadata?: Record<string, string>;
  /**
   * TASK-0122: caller cancellation (the SSE endpoint passes the request's
   * AbortSignal). Aborting mid-stream stops the provider call and ends the
   * stream with a non-retryable `ai_aborted` error (never retried or
   * re-routed to the fallback — the caller already went away).
   */
  signal?: AbortSignal;
}

export interface EmbedParams {
  workspaceId: string;
  userId?: string;
  texts: string[];
  model?: string;
  metadata?: Record<string, string>;
}

/** App layer provides adapters from config + decrypted provider secrets. */
export interface ProviderRegistry {
  get(providerId: ProviderId, config: ResolvedAiConfig): ProviderAdapter;
}

export interface AiGatewayOptions {
  resolveConfig: (workspaceId: string) => ResolvedAiConfig;
  registry: ProviderRegistry;
  budgets: BudgetService;
  ledger: UsageLedger;
  now?: () => number;
  /** Injectable sleep for backoff (tests pass a no-op). */
  sleep?: (ms: number) => Promise<void>;
  /** Max retries after the first attempt (arch §2.4: 3). */
  maxRetries?: number;
}

const BACKOFFS_MS: [number, number, number] = [1_000, 4_000, 16_000];

/** Per-provider circuit breaker: N consecutive failures → open for openMs. */
export class CircuitBreaker {
  private failures = 0;
  private openUntil = 0;
  private readonly now: () => number;
  private readonly threshold: number;
  private readonly openMs: number;
  constructor(now: () => number, threshold = 5, openMs = 30_000) {
    this.now = now;
    this.threshold = threshold;
    this.openMs = openMs;
  }
  get state(): "closed" | "open" | "half-open" {
    if (this.failures >= this.threshold) {
      return this.now() < this.openUntil ? "open" : "half-open";
    }
    return "closed";
  }
  failure(): void {
    this.failures += 1;
    if (this.failures >= this.threshold) this.openUntil = this.now() + this.openMs;
  }
  success(): void {
    this.failures = 0;
    this.openUntil = 0;
  }
}

/**
 * Retryable at the gateway level per arch §2.4: provider 429 / 5xx /
 * network. The final (post-retries) failure is then normalized — a provider
 * 429 surfaces as `RATE_LIMITED`, 5xx/transport as `AI_ERROR` +
 * `SERVICE_UNAVAILABLE` (module doc).
 */
/** TASK-0122: true when the error is a caller-initiated stream abort. */
function isAborted(err: unknown, signal?: AbortSignal): boolean {
  if (signal?.aborted) return true;
  if (err instanceof Error) return err.name === "AbortError" || err.message === "The operation was aborted";
  return false;
}

function isRetryable(err: unknown, signal?: AbortSignal): boolean {
  // A caller-initiated abort is never retryable (the caller went away —
  // retrying would re-spend tokens for no one, and a fallback re-route
  // would stream a duplicate answer into the dead socket).
  if (isAborted(err, signal)) return false;
  if (err instanceof ProviderHttpError) return err.retryable;
  if (err instanceof ApiError) return err.code === ErrorCodes.TIMEOUT;
  return true; // transport-level failures (e.g. "fetch failed") are retryable
}

/**
 * Normalize a provider failure to the canonical error envelope (module doc
 * explains the recorded code decision): 401/403 → AI_ERROR +
 * `details.code = SERVICE_UNAVAILABLE {service:"ai", reason:"auth"}`;
 * 429 → RATE_LIMITED; timeout/transport → AI_ERROR +
 * `{reason:"ai_timeout", retryable:true}`. Raw provider bodies never leak
 * (only a truncated status + message is kept server-side).
 */
export function normalizeProviderError(err: unknown, provider: ProviderId): ApiError {
  if (err instanceof ApiError) return err;
  if (err instanceof ProviderHttpError) {
    if (err.status === 401 || err.status === 403) {
      return new ApiError(ErrorCodes.AI_ERROR, {
        message: "AI provider authentication failed. Check the provider key configuration.",
        status: 502,
        details: { code: "SERVICE_UNAVAILABLE", service: "ai", reason: "auth", provider },
      });
    }
    if (err.status === 429) {
      return new ApiError(ErrorCodes.RATE_LIMITED, {
        message: "AI provider rate limit reached. Try again shortly.",
        details: { provider },
      });
    }
    return new ApiError(ErrorCodes.AI_ERROR, {
      message: "AI provider request failed.",
      status: 502,
      details: {
        code: "SERVICE_UNAVAILABLE",
        service: "ai",
        reason: err.status === 504 ? "ai_timeout" : "provider_error",
        retryable: err.retryable,
        provider,
        httpStatus: err.status,
      },
    });
  }
  return new ApiError(ErrorCodes.AI_ERROR, {
    message: "AI provider request failed.",
    status: 502,
    details: {
      code: "SERVICE_UNAVAILABLE",
      service: "ai",
      reason: "ai_timeout",
      retryable: true,
      provider,
    },
  });
}

function metadataFor(params: {
  workspaceId: string;
  userId?: string;
  capability?: string;
  metadata?: Record<string, string>;
}): Record<string, string> {
  return {
    capability: params.capability ?? "chat",
    workspaceId: params.workspaceId,
    ...(params.userId ? { userId: params.userId } : {}),
    ...params.metadata,
  };
}

export class AiGateway {
  private readonly options: AiGatewayOptions;
  private readonly now: () => number;
  private readonly sleep: (ms: number) => Promise<void>;
  private readonly maxRetries: number;
  private readonly breakers = new Map<ProviderId, CircuitBreaker>();

  constructor(options: AiGatewayOptions) {
    this.options = options;
    this.now = options.now ?? Date.now;
    this.sleep = options.sleep ?? ((ms) => new Promise((r) => setTimeout(r, ms)));
    this.maxRetries = options.maxRetries ?? 3;
  }

  /** Resolved config for a workspace (drives `POST /ai/ping`). */
  resolve(workspaceId: string): ResolvedAiConfig {
    const config = this.options.resolveConfig(workspaceId);
    if (!config.enabled) {
      throw new ApiError(ErrorCodes.NOT_IMPLEMENTED, {
        message: "AI is disabled for this workspace.",
        details: { code: "AI_DISABLED" },
      });
    }
    return config;
  }

  breaker(provider: ProviderId): CircuitBreaker {
    let b = this.breakers.get(provider);
    if (!b) {
      b = new CircuitBreaker(this.now);
      this.breakers.set(provider, b);
    }
    return b;
  }

  private async adapterFor(config: ResolvedAiConfig, providerId: ProviderId): Promise<ProviderAdapter> {
    const state = this.breaker(providerId).state;
    if (state === "open") {
      throw new ApiError(ErrorCodes.AI_ERROR, {
        message: "AI provider is temporarily unavailable (circuit open).",
        status: 503,
        details: { code: "SERVICE_UNAVAILABLE", service: "ai", reason: "circuit_open", provider: providerId },
      });
    }
    const adapter = this.options.registry.get(providerId, config);
    if (adapter.id !== providerId) {
      throw new ApiError(ErrorCodes.INTERNAL, {
        message: `registry returned adapter "${adapter.id}", expected "${providerId}"`,
      });
    }
    return adapter;
  }

  /**
   * Streaming chat (the base primitive). Retries + fallback happen
   * pre-output only; once a chunk is yielded the caller owns the stream.
   */
  async *chat(params: ChatParams): AsyncIterable<ChatChunk> {
    const config = this.resolve(params.workspaceId);
    const model = params.model ?? config.chatModel;

    // tool schema validation (strict subset) — reject before any spend
    if (params.tools && params.tools.length > 0) {
      const result = validateTools(params.tools);
      if (!result.ok) {
        throw new ApiError(ErrorCodes.VALIDATION_ERROR, { message: `invalid tool schema: ${result.error}` });
      }
    }

    // maxOutputTokens clamp (silent, arch §7.2)
    const maxTokens = this.options.budgets.clampMaxTokens(params.maxTokens, config.budgets);

    // submit-time budget check (fast fail; estimate = 2× maxTokens)
    this.options.budgets.check(params.workspaceId, config.budgets, 2 * maxTokens);

    const baseRequest: ChatRequest = {
      model,
      messages: params.messages,
      stream: true,
      metadata: metadataFor(params),
      ...(params.signal ? { signal: params.signal } : {}),
    };
    if (params.temperature !== undefined) baseRequest.temperature = params.temperature;
    if (maxTokens !== undefined) baseRequest.maxTokens = maxTokens;
    if (params.tools && params.tools.length > 0) {
      baseRequest.tools = params.tools;
      if (params.toolChoice !== undefined) baseRequest.toolChoice = params.toolChoice;
    }

    let adapter = await this.adapterFor(config, config.provider);
    let request = baseRequest;
    let attempts = 0;
    let usedFallback = false;

    for (;;) {
      let yielded = false;
      let usage = { in: 0, out: 0, total: 0 };
      let finishReason: ChatResult["finishReason"] = "stop";
      const started = this.now();
      try {
        // A pre-stream abort (the caller cancelled before any provider
        // output) maps to the same non-retryable `ai_aborted` path.
        if (params.signal?.aborted) {
          const preAbort = new Error("The operation was aborted");
          preAbort.name = "AbortError";
          throw preAbort;
        }
        for await (const chunk of adapter.chat(request)) {
          // Per-chunk abort check (TASK-0122): adapters backed by real
          // fetch streams honor the signal themselves, but plain-JS
          // generators (the mock) yield back-to-back — without this check a
          // mid-stream abort would only surface at the adapter's end.
          if (params.signal?.aborted) {
            const abortErr = new Error("The operation was aborted");
            abortErr.name = "AbortError";
            throw abortErr;
          }
          yielded = true;
          if (chunk.done) {
            if (chunk.usage) usage = chunk.usage;
            if (chunk.finishReason) finishReason = chunk.finishReason;
          }
          yield chunk;
        }
        this.breaker(adapter.id).success();
        void this.recordChat(params, adapter.id, request.model, usage, this.now() - started, "ok");
        return;
      } catch (err) {
        if (isAborted(err, params.signal)) {
          // Caller-initiated cancel (TASK-0122): record what was consumed,
          // never retry/re-route (the caller is gone), and surface a
          // non-retryable `ai_aborted` error the service distinguishes from
          // a provider failure. Deliberately NOT counted as a breaker
          // failure — the provider did nothing wrong (a "Stop" click must
          // not trip the circuit after enough cancellations).
          void this.recordChat(params, adapter.id, request.model, usage, this.now() - started, "error");
          throw new ApiError(ErrorCodes.AI_ERROR, {
            message: "AI stream aborted by the caller.",
            status: 502,
            details: { code: "SERVICE_UNAVAILABLE", service: "ai", reason: "ai_aborted", provider: adapter.id },
          });
        }
        this.breaker(adapter.id).failure();
        if (yielded) {
          // mid-stream failure: never re-stream (would duplicate content)
          void this.recordChat(params, adapter.id, request.model, usage, this.now() - started, "error");
          throw normalizeProviderError(err, adapter.id);
        }
        const retryable = isRetryable(err, params.signal);
        if (retryable && attempts < this.maxRetries) {
          attempts += 1;
          await this.sleep(BACKOFFS_MS[Math.min(attempts - 1, BACKOFFS_MS.length - 1)]!);
          continue;
        }
        if (!usedFallback && config.fallback && config.fallback.provider !== adapter.id) {
          usedFallback = true;
          try {
            adapter = await this.adapterFor(config, config.fallback.provider);
            request = { ...baseRequest, model: config.fallback.chatModel };
            continue;
          } catch {
            // fall through to normalizing the primary failure
          }
        }
        void this.recordChat(params, adapter.id, request.model, usage, this.now() - started, "error");
        throw normalizeProviderError(err, adapter.id);
      }
    }
  }

  /** Non-streaming chat: collects the chunk stream into a ChatResult. */
  async chatCollected(params: ChatParams): Promise<ChatResult> {
    const config = this.resolve(params.workspaceId);
    const model = params.model ?? config.chatModel;
    const started = this.now();
    let content = "";
    const toolCalls: ChatResult["toolCalls"] = [];
    let usage = { in: 0, out: 0, total: 0 };
    let finishReason: ChatResult["finishReason"] = "stop";
    for await (const chunk of this.chat(params)) {
      if (chunk.delta) content += chunk.delta;
      if (chunk.toolCallDelta) toolCalls.push(chunk.toolCallDelta);
      if (chunk.done) {
        if (chunk.usage) usage = chunk.usage;
        if (chunk.finishReason) finishReason = chunk.finishReason;
      }
    }
    return {
      content,
      toolCalls,
      usage,
      model,
      latencyMs: this.now() - started,
      provider: config.provider,
      finishReason,
    };
  }

  /**
   * Embeddings. If the primary provider has no embedding API (Anthropic) the
   * gateway routes to `embedFallbackProvider` (default `openai`, arch §2.2).
   * Retries on 429/5xx with the same backoff schedule.
   */
  async embed(params: EmbedParams): Promise<EmbedResult> {
    const config = this.resolve(params.workspaceId);
    const model = params.model ?? config.embedModel;
    const started = this.now();

    // Anthropic has no embed API: route straight to the fallback provider.
    let providerId: ProviderId =
      config.provider === "anthropic" && config.embedFallbackProvider ? config.embedFallbackProvider : config.provider;
    let adapter = await this.adapterFor(config, providerId);
    let attempts = 0;
    let usedEmbedFallback = false;

    const record = (status: "ok" | "error", result?: EmbedResult): Promise<unknown> =>
      this.options.ledger.record({
        workspaceId: params.workspaceId,
        ...(params.userId ? { userId: params.userId } : {}),
        ...(params.metadata?.capability ? { capability: params.metadata.capability } : {}),
        kind: "embed",
        model,
        provider: adapter.id,
        tokensIn: 0,
        tokensOut: result?.usage.tokens ?? estimateTokens(params.texts),
        latencyMs: result?.latencyMs ?? this.now() - started,
        costUsd: 0, // embed prices are tracked per-model in the app layer
        requestKind: "batch",
        status,
      });

    for (;;) {
      try {
        const result = await adapter.embed({
          model,
          texts: params.texts,
          metadata: { workspaceId: params.workspaceId, ...(params.userId ? { userId: params.userId } : {}), ...params.metadata },
        });
        this.breaker(adapter.id).success();
        void record("ok", result);
        return result;
      } catch (err) {
        this.breaker(adapter.id).failure();
        const retryable = isRetryable(err);
        if (retryable && attempts < this.maxRetries) {
          attempts += 1;
          await this.sleep(BACKOFFS_MS[Math.min(attempts - 1, BACKOFFS_MS.length - 1)]!);
          continue;
        }
        if (
          !usedEmbedFallback &&
          config.embedFallbackProvider &&
          config.embedFallbackProvider !== adapter.id
        ) {
          usedEmbedFallback = true;
          try {
            adapter = await this.adapterFor(config, config.embedFallbackProvider);
            continue;
          } catch {
            // fall through to normalizing the primary failure
          }
        }
        void record("error");
        throw normalizeProviderError(err, adapter.id);
      }
    }
  }

  /** Ping/diagnostic: resolved provider/model + budget state (no tokens). */
  async ping(workspaceId: string): Promise<{
    provider: ProviderId;
    chatModel: string;
    embedModel: string;
    budget: { limit: number; used: number };
  }> {
    const config = this.resolve(workspaceId);
    const monthStart = new Date(this.now());
    const used = this.options.ledger.tokensSince(workspaceId, new Date(monthStart.getFullYear(), monthStart.getMonth(), 1).getTime());
    return {
      provider: config.provider,
      chatModel: config.chatModel,
      embedModel: config.embedModel,
      budget: { limit: config.budgets.tokensPerMonth, used },
    };
  }

  private async recordChat(
    params: ChatParams,
    provider: ProviderId,
    model: string,
    usage: { in: number; out: number; total: number },
    latencyMs: number,
    status: "ok" | "error" | "timeout",
  ): Promise<void> {
    await this.options.ledger.record({
      workspaceId: params.workspaceId,
      ...(params.userId ? { userId: params.userId } : {}),
      ...(params.capability ? { capability: params.capability } : {}),
      kind: "chat",
      model,
      provider,
      tokensIn: usage.in,
      tokensOut: usage.out,
      latencyMs,
      costUsd: this.costFor(model, usage.in, usage.out),
      requestKind: "stream",
      status,
    });
  }

  /** Cost in USD from the static price table (0 for OSS/unknown models). */
  private costFor(model: string, tokensIn: number, tokensOut: number): number {
    const price = MODEL_PRICES[model];
    if (!price) return 0;
    return (tokensIn / 1000) * price.in + (tokensOut / 1000) * price.out;
  }
}

/** Static price table (USD per 1k tokens). The app layer can extend this. */
export const MODEL_PRICES: Record<string, { in: number; out: number }> = {
  "gpt-4o": { in: 2.5, out: 10 },
  "gpt-4o-mini": { in: 0.15, out: 0.6 },
  "text-embedding-3-small": { in: 0.02, out: 0 },
  "text-embedding-3-large": { in: 0.13, out: 0 },
  "claude-sonnet-4-20250514": { in: 3, out: 15 },
  "claude-opus-4-20250514": { in: 15, out: 75 },
  "claude-3-5-haiku-20241022": { in: 0.8, out: 4 },
  "gemini-2.0-flash": { in: 0.1, out: 0.4 },
  "text-embedding-004": { in: 0.025, out: 0 },
};

function estimateTokens(texts: string[]): number {
  return Math.ceil(texts.join(" ").length / 4);
}