// Messages API 客户端：对应 veridrop protocols/anthropic/client.py。
//
// 两层：
// - AnthropicClient：裸 fetch，应答原样返回（不做字段归一），才看得见中转站到底回了什么。
// - ThrottledClient：并发上限（信号量）、429/5xx 全局退避重试、把每个成功应答广播给被动检测项、累计用量与首字。
//
// 与 httpx 的差别：httpx 的 timeout 是「每次读之间」的超时。这里非流式请求按整个请求算（非流式应答本来就是生成完一次到齐，
// 两者差不多），流式请求按「两次收到数据之间」算，与 httpx 一致。

import { PARAM_DEPRECATIONS, normalizeModelId } from "./config.js";
import { addUsage, emptyUsage, type UsageMetrics } from "./models.js";
import type { PassiveDetector } from "./detectors/base.js";
import { cpSlice, isInt } from "./py.js";

export const ANTHROPIC_VERSION = "2023-06-01";
export const DEFAULT_TIMEOUT_S = 30;
export const DEFAULT_MAX_CONCURRENT = 3;
export const RETRYABLE_STATUS: ReadonlySet<number> = new Set([429, 500, 502, 503, 504, 529]);
export const MAX_BACKOFF_S = 30;
export const MAX_RETRIES = 4;

export type Body = Record<string, unknown>;
export type FetchLike = (input: string, init: RequestInit) => Promise<Response>;

export interface StreamEvent {
  event: string;
  data: Record<string, unknown>;
}

export interface CallResult {
  req: Body;
  resp: Record<string, unknown>;
  headers: Headers;
  latencyMs: number;
}

/** 非 2xx。message 照 Python 的 `HTTP {status}: {body[:200]}`。 */
export class AnthropicAPIError extends Error {
  constructor(
    readonly status: number,
    readonly body: string,
    readonly headers: Headers | null,
  ) {
    super(`HTTP ${status}: ${cpSlice(body, 200)}`);
    this.name = "AnthropicAPIError";
  }
}

/** 连不上、断开、超时：相当于 httpx.TimeoutException / NetworkError，会被重试。 */
export class TransportError extends Error {
  constructor(message: string, readonly timeout: boolean) {
    super(message);
    this.name = timeout ? "TimeoutException" : "NetworkError";
  }
}

/** 按型号剥掉官方会拒的参数（容忍 `claude-opus-4.7` 这种点号写法）。就地改 body。 */
export function sanitizeBody(body: Body): Body {
  const model = body["model"];
  if (typeof model !== "string") return body;
  const normalized = normalizeModelId(model);
  for (const [prefix, fields] of PARAM_DEPRECATIONS) {
    if (normalized.startsWith(prefix)) {
      for (const k of fields) delete body[k];
    }
  }
  return body;
}

/** 用户常贴带 /v1 的根地址；客户端自己拼 /v1/messages，所以去掉结尾的 /v1。 */
export function normalizeBaseUrl(baseUrl: string): string {
  let n = baseUrl.replace(/\/+$/, "");
  if (n.endsWith("/v1")) n = n.slice(0, -3).replace(/\/+$/, "");
  return n;
}

function now(): number {
  return performance.now();
}

export interface ClientOptions {
  timeoutS?: number;
  anthropicVersion?: string;
  extraHeaders?: Record<string, string>;
  fetchImpl?: FetchLike;
  /** 整轮作废（总超时）时由 runner 触发，在飞的请求一并中止。 */
  signal?: AbortSignal;
}

export class AnthropicClient {
  readonly baseUrl: string;
  private readonly headers: Record<string, string>;
  private readonly timeoutMs: number;
  private readonly fetchImpl: FetchLike;
  private readonly signal: AbortSignal | undefined;

  constructor(baseUrl: string, apiKey: string, opts: ClientOptions = {}) {
    this.baseUrl = normalizeBaseUrl(baseUrl);
    this.headers = {
      "x-api-key": apiKey,
      "anthropic-version": opts.anthropicVersion ?? ANTHROPIC_VERSION,
      "content-type": "application/json",
      ...(opts.extraHeaders ?? {}),
    };
    this.timeoutMs = (opts.timeoutS ?? DEFAULT_TIMEOUT_S) * 1000;
    this.fetchImpl = opts.fetchImpl ?? ((input, init) => fetch(input, init));
    this.signal = opts.signal;
  }

  /** 发一个 POST；超时 / 网络错误统一成 TransportError。返回时还没读 body，计时器还开着（由调用方 clear）。 */
  private async post(path: string, body: Body, timer: RequestTimer): Promise<Response> {
    timer.arm();
    try {
      return await this.fetchImpl(this.baseUrl + path, {
        method: "POST",
        headers: this.headers,
        body: JSON.stringify(body),
        signal: timer.signal,
      });
    } catch (e) {
      timer.clear();
      throw timer.wrap(e);
    }
  }

  /** 读完整个 body（受同一个超时管），读完清掉计时器。 */
  private async readAll(resp: Response, timer: RequestTimer): Promise<string> {
    try {
      return await resp.text();
    } catch (e) {
      throw timer.wrap(e);
    } finally {
      timer.clear();
    }
  }

  /** 非流式 POST /v1/messages。body 里的 request_timeout_s 覆盖默认超时。非 2xx 抛 AnthropicAPIError。 */
  async messagesCreate(input: Body): Promise<CallResult> {
    const body: Body = { ...input };
    delete body["stream"];
    const override = body["request_timeout_s"];
    delete body["request_timeout_s"];
    sanitizeBody(body);
    const timer = new RequestTimer(typeof override === "number" ? override * 1000 : this.timeoutMs, this.signal);
    const start = now();
    const resp = await this.post("/v1/messages", body, timer);
    const text = await this.readAll(resp, timer);
    const latencyMs = Math.trunc(now() - start);
    if (resp.status >= 400) throw new AnthropicAPIError(resp.status, text, resp.headers);
    return { req: body, resp: JSON.parse(text) as Record<string, unknown>, headers: resp.headers, latencyMs };
  }

  /** POST /v1/messages/count_tokens（去掉 stream / max_tokens / temperature）。 */
  async countTokens(input: Body): Promise<CallResult> {
    const body: Body = { ...input };
    delete body["stream"];
    delete body["max_tokens"];
    delete body["temperature"];
    sanitizeBody(body);
    const timer = new RequestTimer(this.timeoutMs, this.signal);
    const start = now();
    const resp = await this.post("/v1/messages/count_tokens", body, timer);
    const text = await this.readAll(resp, timer);
    const latencyMs = Math.trunc(now() - start);
    if (resp.status >= 400) throw new AnthropicAPIError(resp.status, text, resp.headers);
    return { req: body, resp: JSON.parse(text) as Record<string, unknown>, headers: resp.headers, latencyMs };
  }

  /** 流式 POST /v1/messages：逐个产出 (事件, 距请求开始的毫秒数)。onHeaders 在第一个事件之前调一次。超时按两次收到数据之间算。 */
  async *messagesStream(input: Body, onHeaders?: (h: Headers) => void): AsyncGenerator<[StreamEvent, number]> {
    const body: Body = { ...input, stream: true };
    sanitizeBody(body);
    const timer = new RequestTimer(this.timeoutMs, this.signal);
    const start = now();
    const resp = await this.post("/v1/messages", body, timer);
    try {
      if (resp.status >= 400) {
        const errBody = await this.readAll(resp, timer);
        throw new AnthropicAPIError(resp.status, errBody, resp.headers);
      }
      onHeaders?.(resp.headers);
      if (!resp.body) return;
      for await (const ev of parseSSE(readLines(resp.body, timer))) {
        yield [ev, Math.trunc(now() - start)];
      }
    } finally {
      timer.clear();
      timer.abort(); // 读完或者调用方提前不要了：把连接关掉
    }
  }
}

/** 一个请求的超时：AbortController + 可重置的计时器，外面的整轮中止信号也接进来。 */
class RequestTimer {
  private readonly ctrl = new AbortController();
  private handle: ReturnType<typeof setTimeout> | undefined;
  private timedOut = false;
  private readonly outer: AbortSignal | undefined;
  private readonly onOuter = () => this.ctrl.abort();

  constructor(private readonly ms: number, outer?: AbortSignal) {
    this.outer = outer;
    if (outer) {
      if (outer.aborted) this.ctrl.abort();
      else outer.addEventListener("abort", this.onOuter, { once: true });
    }
  }

  get signal(): AbortSignal {
    return this.ctrl.signal;
  }

  /** 开始或重新开始计时（流式每收到一块数据调一次）。 */
  arm(): void {
    if (this.handle !== undefined) clearTimeout(this.handle);
    this.handle = setTimeout(() => {
      this.timedOut = true;
      this.ctrl.abort();
    }, this.ms);
  }

  clear(): void {
    if (this.handle !== undefined) clearTimeout(this.handle);
    this.handle = undefined;
    this.outer?.removeEventListener("abort", this.onOuter);
  }

  abort(): void {
    this.ctrl.abort();
  }

  /** 把 fetch / 读 body 抛的错换成 TransportError。 */
  wrap(e: unknown): TransportError {
    if (this.timedOut) return new TransportError(`timeout after ${this.ms / 1000}s`, true);
    if (this.outer?.aborted) return new TransportError("aborted: overall timeout", true);
    return new TransportError(e instanceof Error ? `${e.name}: ${e.message}` : String(e), false);
  }
}

/** 把字节流拆成行（\n、\r\n、\r 都算行尾，与 httpx aiter_lines 一致）。每收到一块数据重置一次空闲计时。 */
async function* readLines(stream: ReadableStream<Uint8Array>, timer: RequestTimer): AsyncGenerator<string> {
  const reader = stream.getReader();
  const decoder = new TextDecoder();
  let buf = "";
  try {
    for (;;) {
      let chunk: ReadableStreamReadResult<Uint8Array>;
      try {
        chunk = await reader.read();
      } catch (e) {
        throw timer.wrap(e);
      }
      if (chunk.done) break;
      timer.arm();
      buf += decoder.decode(chunk.value, { stream: true });
      let m: RegExpExecArray | null;
      const re = /\r\n|\n|\r/g;
      let last = 0;
      // 结尾单独一个 \r 可能是 \r\n 的前半，先留着等下一块。
      while ((m = re.exec(buf)) !== null) {
        if (m[0] === "\r" && m.index === buf.length - 1) break;
        yield buf.slice(last, m.index);
        last = m.index + m[0].length;
      }
      buf = buf.slice(last);
    }
    buf += decoder.decode();
    if (buf.endsWith("\r")) buf = buf.slice(0, -1);
    if (buf !== "") yield buf;
  } finally {
    reader.releaseLock();
  }
}

/** 最小的 SSE 解析：只认 event: 与 data:（多行 data 用 \n 拼），空行结束一个事件；没有 event: 的块丢掉。 */
export async function* parseSSE(lines: AsyncIterable<string>): AsyncGenerator<StreamEvent> {
  let eventName: string | null = null;
  let dataLines: string[] = [];
  const flush = (): StreamEvent | null => {
    if (eventName === null) return null;
    const payload = dataLines.join("\n");
    let data: Record<string, unknown>;
    try {
      data = payload ? (JSON.parse(payload) as Record<string, unknown>) : {};
    } catch {
      data = { _raw: payload, _parse_error: true };
    }
    return { event: eventName, data };
  };
  for await (const line of lines) {
    if (line === "") {
      const ev = flush();
      if (ev) yield ev;
      eventName = null;
      dataLines = [];
      continue;
    }
    if (line.startsWith(":")) continue;
    if (line.startsWith("event:")) eventName = line.slice(6).trim();
    else if (line.startsWith("data:")) dataLines.push(line.slice(5).trimStart());
  }
  const ev = flush();
  if (ev) yield ev;
}

// ── 限流包装 ──────────────────────────────────────────────────────────

class Semaphore {
  private waiters: Array<() => void> = [];
  constructor(private permits: number) {}
  async acquire(): Promise<void> {
    if (this.permits > 0) {
      this.permits--;
      return;
    }
    await new Promise<void>((resolve) => this.waiters.push(resolve));
  }
  release(): void {
    const next = this.waiters.shift();
    if (next) next();
    else this.permits++;
  }
}

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

/**
 * 并发上限 + 全局退避：429 / 5xx 让所有并发中的调用一起等；成功的应答广播给被动检测项。
 * 流式请求也占信号量，但只在建连时重试，事件开始流了就不重试。
 */
export class ThrottledClient {
  requestCount = 0;
  backoffEvents = 0;
  readonly totalUsage: UsageMetrics = emptyUsage();
  /** 每个流从请求开始到第一个 content_block_delta 的毫秒数。 */
  readonly ttftSamplesMs: number[] = [];
  private readonly sema: Semaphore;
  private backoffUntil = 0; // performance.now() 毫秒

  constructor(
    private readonly base: AnthropicClient,
    private readonly passive: readonly PassiveDetector[] = [],
    maxConcurrent = DEFAULT_MAX_CONCURRENT,
    private readonly sleepImpl: (ms: number) => Promise<void> = sleep,
  ) {
    this.sema = new Semaphore(maxConcurrent);
  }

  get baseUrl(): string {
    return this.base.baseUrl;
  }

  private async waitForBackoff(): Promise<void> {
    for (;;) {
      const wait = this.backoffUntil - now();
      if (wait <= 0) return;
      await this.sleepImpl(wait);
    }
  }

  private triggerBackoff(retryAfterS: number): void {
    const until = now() + retryAfterS * 1000;
    if (until > this.backoffUntil) {
      this.backoffUntil = until;
      this.backoffEvents++;
    }
  }

  private retryAfterSeconds(e: AnthropicAPIError, attempt: number): number {
    const ra = e.headers?.get("retry-after");
    if (ra) {
      const v = Number(ra);
      if (ra.trim() !== "" && Number.isFinite(v)) return Math.min(v, MAX_BACKOFF_S);
    }
    return Math.min(2 ** attempt, MAX_BACKOFF_S);
  }

  private broadcast(request: Body, response: Record<string, unknown>, headers: Headers, latencyMs: number): void {
    for (const d of this.passive) {
      try {
        d.observe(request, response, headers, latencyMs);
      } catch {
        // 被动观察出错不能打断主流程
      }
    }
  }

  private absorbResponseUsage(response: Record<string, unknown>): void {
    const usage = response["usage"];
    if (!usage || typeof usage !== "object") return;
    const u = usage as Record<string, unknown>;
    const delta = emptyUsage();
    if (isInt(u["input_tokens"])) delta.input_tokens = u["input_tokens"];
    if (isInt(u["output_tokens"])) delta.output_tokens = u["output_tokens"];
    if (isInt(u["cache_read_input_tokens"])) delta.cache_read_input_tokens = u["cache_read_input_tokens"];
    if (isInt(u["cache_creation_input_tokens"])) delta.cache_creation_input_tokens = u["cache_creation_input_tokens"];
    addUsage(this.totalUsage, delta);
  }

  messagesCreate(body: Body): Promise<CallResult> {
    return this.withRetry(() => this.base.messagesCreate(body), true);
  }

  countTokens(body: Body): Promise<CallResult> {
    return this.withRetry(() => this.base.countTokens(body), false);
  }

  async *messagesStream(body: Body): AsyncGenerator<[StreamEvent, number]> {
    await this.waitForBackoff();
    await this.sema.acquire();
    this.requestCount++;
    const local = emptyUsage();
    const captured: Headers[] = [];
    let startMsg: Record<string, unknown> | null = null;
    let stopReason: unknown = null;
    let stopSequence: unknown = null;
    let deltaUsage: Record<string, unknown> | null = null;
    const started = now();
    let ttftRecorded = false;
    try {
      for await (const pair of this.base.messagesStream(body, (h) => captured.push(h))) {
        const [ev, elapsed] = pair;
        if (!ttftRecorded && ev.event === "content_block_delta") {
          this.ttftSamplesMs.push(elapsed);
          ttftRecorded = true;
        }
        absorbStreamEvent(ev, local);
        if (ev.event === "message_start" && startMsg === null) {
          const msg = ev.data["message"];
          if (msg && typeof msg === "object" && !Array.isArray(msg)) startMsg = msg as Record<string, unknown>;
        } else if (ev.event === "message_delta") {
          const d = (ev.data["delta"] ?? {}) as Record<string, unknown>;
          if ("stop_reason" in d) stopReason = d["stop_reason"];
          if ("stop_sequence" in d) stopSequence = d["stop_sequence"];
          const u = ev.data["usage"];
          if (u && typeof u === "object" && !Array.isArray(u)) deltaUsage = u as Record<string, unknown>;
        }
        yield pair;
      }
    } finally {
      this.sema.release();
      addUsage(this.totalUsage, local);
      if (startMsg !== null) {
        const final = synthesizeStreamResponse(startMsg, stopReason, stopSequence, deltaUsage);
        this.broadcast(body, final, captured[0] ?? new Headers(), Math.trunc(now() - started));
      }
    }
  }

  private async withRetry(op: () => Promise<CallResult>, broadcast: boolean): Promise<CallResult> {
    let lastErr: unknown = null;
    for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
      await this.waitForBackoff();
      await this.sema.acquire();
      let res: CallResult;
      try {
        this.requestCount++;
        try {
          res = await op();
        } catch (e) {
          lastErr = e;
          if (e instanceof AnthropicAPIError) {
            if (RETRYABLE_STATUS.has(e.status) && attempt < MAX_RETRIES) {
              this.triggerBackoff(this.retryAfterSeconds(e, attempt));
              continue;
            }
            throw e;
          }
          if (e instanceof TransportError && attempt < MAX_RETRIES) {
            this.triggerBackoff(Math.min(2 ** attempt, MAX_BACKOFF_S));
            continue;
          }
          throw e;
        }
      } finally {
        this.sema.release();
      }
      this.absorbResponseUsage(res.resp);
      if (broadcast) this.broadcast(res.req, res.resp, res.headers, res.latencyMs);
      return res;
    }
    throw lastErr ?? new Error("retry loop exhausted");
  }
}

/** message_start 带真的 input_tokens（与缓存两项）；message_delta 带累计的 output_tokens，取最后一个。 */
function absorbStreamEvent(ev: StreamEvent, local: UsageMetrics): void {
  if (ev.event === "message_start") {
    const msg = (ev.data["message"] ?? {}) as Record<string, unknown>;
    const usage = (msg["usage"] ?? {}) as Record<string, unknown>;
    if (isInt(usage["input_tokens"])) local.input_tokens = usage["input_tokens"];
    if (isInt(usage["cache_read_input_tokens"])) local.cache_read_input_tokens = usage["cache_read_input_tokens"];
    if (isInt(usage["cache_creation_input_tokens"])) local.cache_creation_input_tokens = usage["cache_creation_input_tokens"];
  } else if (ev.event === "message_delta") {
    const usage = (ev.data["usage"] ?? {}) as Record<string, unknown>;
    if (isInt(usage["output_tokens"])) local.output_tokens = usage["output_tokens"];
  }
}

/** 拿 message_start 的顶层字段 + message_delta 的 stop_reason / 用量拼一个非流式样子的应答，给被动检测项看（不重建 content）。 */
function synthesizeStreamResponse(
  startMsg: Record<string, unknown>,
  stopReason: unknown,
  stopSequence: unknown,
  deltaUsage: Record<string, unknown> | null,
): Record<string, unknown> {
  const final: Record<string, unknown> = { ...startMsg, stop_reason: stopReason, stop_sequence: stopSequence };
  if (deltaUsage && Object.keys(deltaUsage).length > 0) {
    const base = { ...((startMsg["usage"] ?? {}) as Record<string, unknown>) };
    Object.assign(base, deltaUsage);
    final["usage"] = base;
  }
  return final;
}