/**
 * KHAELOR
 * File: src/anthropic/client.ts
 * Description: AnthropicModelClient — SDK streaming translated to ModelEvents, retry/backoff, AbortSignal cancellation (ADR-10).
 *
 * Author: Simon-Pierre Boucher
 * Contact: contact@spboucher.ai
 */

import Anthropic from "@anthropic-ai/sdk";
import type {
  MessageCountTokensParams,
  MessageCreateParamsStreaming,
  MessageParam,
  RawMessageStreamEvent,
  StopReason as ApiStopReason,
  ThinkingConfigParam,
  Tool,
} from "@anthropic-ai/sdk/resources/messages";
import { ModelError, classifyModelError } from "./errors.js";
import {
  buildMessagesWithCacheControl,
  buildSystemBlocks,
  planCacheBreakpoints,
} from "./caching.js";
import type { ModelClient, ModelEvent, ModelRequest, ModelUsage, StopReason } from "./types.js";

// ───────────────────────── construction options ─────────────────────────

/** Produces the raw SSE event stream — injectable so tests run without a network. */
export type RawStreamFactory = (
  params: MessageCreateParamsStreaming,
  signal?: AbortSignal,
) => Promise<AsyncIterable<RawMessageStreamEvent>>;

export interface AnthropicClientOptions {
  /** Passed in by the composition root — this module never reads config or env (§2.1 rule 6). */
  apiKey: string;
  baseUrl?: string;
  /** Retry budget for retryable errors before the first streamed event (default 3). */
  maxRetries?: number;
  /** Exponential backoff base (default 500 ms) and cap (default 8000 ms). */
  backoffBaseMs?: number;
  backoffCapMs?: number;
  /** Injectable clock / jitter / sleep for deterministic tests. */
  now?: () => number;
  random?: () => number;
  sleep?: (ms: number, signal?: AbortSignal) => Promise<void>;
  streamFactory?: RawStreamFactory;
}

/** Ceiling applied to server-provided retry-after hints. */
const RETRY_AFTER_CAP_MS = 30_000;

// ───────────────────────── stop reason mapping ─────────────────────────

/** Normalize API stop reasons onto the KHAELOR vocabulary (EVENT_MODEL.md §4). */
export function mapStopReason(raw: ApiStopReason | string | null | undefined): StopReason {
  switch (raw) {
    case "tool_use":
      return "tool_use";
    case "max_tokens":
      return "max_tokens";
    case "refusal":
      return "refusal";
    default:
      // end_turn, stop_sequence, pause_turn, model_context_window_exceeded fallback, null.
      return "end_turn";
  }
}

// ───────────────────────── SSE → ModelEvent translation ─────────────────────────

type BlockState =
  | { kind: "text"; text: string }
  | { kind: "thinking"; thinking: string; signature: string }
  | { kind: "tool_use"; toolUseId: string; toolName: string; inputJson: string }
  | { kind: "opaque" };

/**
 * Translate the raw Anthropic SSE event sequence (message_start,
 * content_block_start/delta/stop, message_delta, message_stop) into
 * ModelEvents. Pure with respect to I/O — exported for network-free tests.
 *
 * Tool input JSON is accumulated across input_json_delta chunks and parsed
 * once at content_block_stop. Usage comes only from real message_start /
 * message_delta payloads (Absolute Rule #4).
 */
export async function* translateRawStream(
  raw: AsyncIterable<RawMessageStreamEvent>,
  now: () => number = Date.now,
): AsyncGenerator<ModelEvent, void, undefined> {
  const startedAt = now();
  const blocks = new Map<number, BlockState>();
  const usage: ModelUsage = {
    inputTokens: 0,
    outputTokens: 0,
    cacheReadTokens: 0,
    cacheWriteTokens: 0,
  };
  let stopReason: StopReason | undefined;
  let sawMessageStop = false;

  for await (const event of raw) {
    switch (event.type) {
      case "message_start": {
        const u = event.message.usage;
        usage.inputTokens = u.input_tokens ?? 0;
        usage.outputTokens = u.output_tokens ?? 0;
        usage.cacheReadTokens = u.cache_read_input_tokens ?? 0;
        usage.cacheWriteTokens = u.cache_creation_input_tokens ?? 0;
        yield { type: "started", requestId: event.message.id };
        break;
      }

      case "content_block_start": {
        const block = event.content_block;
        if (block.type === "text") {
          blocks.set(event.index, { kind: "text", text: block.text ?? "" });
        } else if (block.type === "thinking") {
          blocks.set(event.index, {
            kind: "thinking",
            thinking: block.thinking ?? "",
            signature: block.signature ?? "",
          });
        } else if (block.type === "tool_use") {
          blocks.set(event.index, {
            kind: "tool_use",
            toolUseId: block.id,
            toolName: block.name,
            inputJson: "",
          });
          yield {
            type: "tool-call-started",
            blockIndex: event.index,
            toolUseId: block.id,
            toolName: block.name,
          };
        } else {
          // redacted_thinking / server tool blocks — not part of the V1 vocabulary.
          blocks.set(event.index, { kind: "opaque" });
        }
        break;
      }

      case "content_block_delta": {
        const state = blocks.get(event.index);
        const delta = event.delta;
        if (delta.type === "text_delta" && state?.kind === "text") {
          state.text += delta.text;
          yield { type: "text-delta", blockIndex: event.index, text: delta.text };
        } else if (delta.type === "thinking_delta" && state?.kind === "thinking") {
          state.thinking += delta.thinking;
          yield { type: "thinking-delta", blockIndex: event.index, text: delta.thinking };
        } else if (delta.type === "signature_delta" && state?.kind === "thinking") {
          state.signature += delta.signature;
        } else if (delta.type === "input_json_delta" && state?.kind === "tool_use") {
          state.inputJson += delta.partial_json;
          yield {
            type: "tool-input-delta",
            blockIndex: event.index,
            toolUseId: state.toolUseId,
            partialJson: delta.partial_json,
          };
        }
        // citations_delta and mismatched deltas are ignored.
        break;
      }

      case "content_block_stop": {
        const state = blocks.get(event.index);
        blocks.delete(event.index);
        if (state === undefined || state.kind === "opaque") break;
        if (state.kind === "text") {
          yield { type: "text-block-completed", blockIndex: event.index, text: state.text };
        } else if (state.kind === "thinking") {
          yield {
            type: "thinking-block-completed",
            blockIndex: event.index,
            thinking: state.thinking,
            signature: state.signature,
          };
        } else {
          let input: unknown;
          try {
            input = state.inputJson.trim() === "" ? {} : JSON.parse(state.inputJson);
          } catch (cause) {
            throw new ModelError(
              "invalid-request",
              `Malformed tool input JSON for tool "${state.toolName}" (${state.toolUseId}).`,
              { cause },
            );
          }
          yield {
            type: "tool-call-completed",
            blockIndex: event.index,
            toolUseId: state.toolUseId,
            toolName: state.toolName,
            input,
          };
        }
        break;
      }

      case "message_delta": {
        if (event.delta.stop_reason !== null) {
          stopReason = mapStopReason(event.delta.stop_reason);
        }
        const u = event.usage;
        usage.outputTokens = u.output_tokens; // cumulative
        if (u.input_tokens !== null && u.input_tokens !== undefined) {
          usage.inputTokens = u.input_tokens;
        }
        if (u.cache_read_input_tokens !== null && u.cache_read_input_tokens !== undefined) {
          usage.cacheReadTokens = u.cache_read_input_tokens;
        }
        if (u.cache_creation_input_tokens !== null && u.cache_creation_input_tokens !== undefined) {
          usage.cacheWriteTokens = u.cache_creation_input_tokens;
        }
        break;
      }

      case "message_stop": {
        sawMessageStop = true;
        yield {
          type: "completed",
          stopReason: stopReason ?? "end_turn",
          usage: { ...usage },
          durationMs: now() - startedAt,
        };
        break;
      }
    }
  }

  if (!sawMessageStop) {
    throw new ModelError("retryable", "Model stream ended without message_stop.");
  }
}

// ───────────────────────── backoff and sleep ─────────────────────────

function defaultSleep(ms: number, signal?: AbortSignal): Promise<void> {
  return new Promise((resolve, reject) => {
    if (signal?.aborted) {
      reject(new ModelError("cancelled", "Request cancelled during retry backoff."));
      return;
    }
    const onAbort = (): void => {
      clearTimeout(timer);
      reject(new ModelError("cancelled", "Request cancelled during retry backoff."));
    };
    const timer = setTimeout(() => {
      signal?.removeEventListener("abort", onAbort);
      resolve();
    }, ms);
    signal?.addEventListener("abort", onAbort, { once: true });
  });
}

/**
 * Exponential backoff with equal jitter: raw = min(cap, base·2^(attempt-1)),
 * delay = raw/2 + random·raw/2. A server retry-after hint overrides (capped).
 */
export function backoffDelayMs(
  attempt: number,
  opts: { baseMs: number; capMs: number; random: () => number; retryAfterMs?: number },
): number {
  if (opts.retryAfterMs !== undefined) {
    return Math.min(opts.retryAfterMs, RETRY_AFTER_CAP_MS);
  }
  const raw = Math.min(opts.capMs, opts.baseMs * 2 ** (attempt - 1));
  return raw / 2 + opts.random() * (raw / 2);
}

// ───────────────────────── the client ─────────────────────────

/**
 * The single V1 ModelClient implementation (CLAUDE.md §6, ADR-10).
 *
 * - Streaming only — there is no non-streaming path (ADR-5).
 * - Retries with jittered exponential backoff, ONLY for retryable errors
 *   and ONLY before the first streamed event; mid-stream failures surface.
 * - AbortSignal aborts the underlying HTTP request (SDK request option).
 * - Never logs; error messages are redacted (errors.ts).
 */
export class AnthropicModelClient implements ModelClient {
  private readonly apiKey: string;
  private readonly baseUrl: string | undefined;
  private readonly maxRetries: number;
  private readonly backoffBaseMs: number;
  private readonly backoffCapMs: number;
  private readonly now: () => number;
  private readonly random: () => number;
  private readonly sleep: (ms: number, signal?: AbortSignal) => Promise<void>;
  private readonly streamFactory: RawStreamFactory;
  private sdk: Anthropic | undefined;

  constructor(options: AnthropicClientOptions) {
    this.apiKey = options.apiKey;
    this.baseUrl = options.baseUrl;
    this.maxRetries = options.maxRetries ?? 3;
    this.backoffBaseMs = options.backoffBaseMs ?? 500;
    this.backoffCapMs = options.backoffCapMs ?? 8_000;
    this.now = options.now ?? Date.now;
    this.random = options.random ?? Math.random;
    this.sleep = options.sleep ?? defaultSleep;
    this.streamFactory = options.streamFactory ?? this.defaultStreamFactory.bind(this);
  }

  /** Lazily construct the SDK client (lazy-import discipline is handled by cli). */
  private getSdk(): Anthropic {
    if (this.sdk === undefined) {
      this.sdk = new Anthropic({
        apiKey: this.apiKey,
        maxRetries: 0, // retry policy lives here, not in the SDK
        ...(this.baseUrl !== undefined ? { baseURL: this.baseUrl } : {}),
      });
    }
    return this.sdk;
  }

  private async defaultStreamFactory(
    params: MessageCreateParamsStreaming,
    signal?: AbortSignal,
  ): Promise<AsyncIterable<RawMessageStreamEvent>> {
    return this.getSdk().messages.create(params, signal !== undefined ? { signal } : undefined);
  }

  /** Assemble the wire request — cache breakpoints planned per ADR-7 (caching.ts). */
  private buildStreamParams(request: ModelRequest): MessageCreateParamsStreaming {
    const plan = planCacheBreakpoints(request);
    const thinking: ThinkingConfigParam | undefined =
      request.thinking === undefined
        ? undefined
        : request.thinking.mode === "enabled"
          ? { type: "enabled", budget_tokens: request.thinking.budgetTokens }
          : { type: "disabled" };
    return {
      model: request.model,
      max_tokens: request.maxOutputTokens,
      system: buildSystemBlocks(request.system, plan),
      // Structurally compatible with the SDK's MessageParam vocabulary.
      messages: buildMessagesWithCacheControl(request.messages, plan) as MessageParam[],
      tools: request.tools.map(
        (t): Tool => ({ name: t.name, description: t.description, input_schema: t.inputSchema as Tool.InputSchema }),
      ),
      ...(thinking !== undefined ? { thinking } : {}),
      stream: true,
    };
  }

  async *stream(request: ModelRequest, signal?: AbortSignal): AsyncIterable<ModelEvent> {
    const params = this.buildStreamParams(request);
    let attempt = 0;

    for (;;) {
      if (signal?.aborted) {
        throw new ModelError("cancelled", "Request cancelled before start.");
      }
      let yieldedAny = false;
      try {
        const raw = await this.streamFactory(params, signal);
        for await (const event of translateRawStream(raw, this.now)) {
          yieldedAny = true;
          yield event;
        }
        return;
      } catch (err) {
        const modelError = classifyModelError(err);
        if (modelError.kind === "cancelled") throw modelError;
        if (signal?.aborted) {
          throw new ModelError("cancelled", "Request cancelled.", { cause: err });
        }
        const canRetry = modelError.retryable && !yieldedAny && attempt < this.maxRetries;
        if (!canRetry) {
          if (modelError.retryable && !yieldedAny) modelError.retriesExhausted = true;
          throw modelError;
        }
        attempt += 1;
        const delay = backoffDelayMs(attempt, {
          baseMs: this.backoffBaseMs,
          capMs: this.backoffCapMs,
          random: this.random,
          ...(modelError.retryAfterMs !== undefined
            ? { retryAfterMs: modelError.retryAfterMs }
            : {}),
        });
        await this.sleep(delay, signal);
      }
    }
  }

  /** Best-effort real token count (no retry — budgeting aid only). */
  async countTokens(request: ModelRequest): Promise<number> {
    const plan = planCacheBreakpoints(request);
    const params: MessageCountTokensParams = {
      model: request.model,
      system: buildSystemBlocks(request.system, plan),
      messages: buildMessagesWithCacheControl(request.messages, plan) as MessageParam[],
      tools: request.tools.map(
        (t): Tool => ({ name: t.name, description: t.description, input_schema: t.inputSchema as Tool.InputSchema }),
      ),
    };
    try {
      const result = await this.getSdk().messages.countTokens(params);
      return result.input_tokens;
    } catch (err) {
      throw classifyModelError(err);
    }
  }
}