// src/proxy/handler.ts
// Extracted fetch handler — imported dynamically by server.ts so that
// `fewtok reload` can swap handler logic in-process without restarting
// the daemon. server.ts owns Bun.serve; this file owns all request logic.

import { join } from "node:path";
import { existsSync, mkdirSync, writeFileSync, chmodSync, readFileSync } from "node:fs";
import { homedir } from "node:os";
import { createHash } from "node:crypto";
import { fileURLToPath } from "node:url";
import { handleControl } from "./controlEndpoint.ts";
import { paths } from "../lifecycle/paths.ts";
import { forwardToUpstream } from "./forward.ts";
import { wrapSseForUsage, type SseUsageAcc } from "./sse-usage.ts";
import { runOutbound, runInbound, collectObservations, makePipeline } from "../layer/pipeline.ts";
import type { SavingEvent, Layer, SessionId } from "../layer/types.ts";
import { asSessionId } from "../layer/types.ts";
import { openReadCache } from "../cache/ReadCache.ts";
import type { ProviderRegistry, ResolvedProvider } from "../provider/registry.ts";
import type { AdapterRegistry } from "../provider/AdapterRegistry.ts";
import type { Provider } from "../provider/types.ts";
import type { Pipeline } from "../layer/pipeline.ts";
import type { RequestMeta } from "../stats/writer.ts";
import { OutputRewriteLayer } from "../layer/OutputRewriteLayer.ts";
import { KbToolInjectLayer } from "../layer/KbToolInjectLayer.ts";
import { KbResponseInterceptor } from "../kb/KbResponseInterceptor.ts";
import { KbStore } from "../kb/KbStore.ts";
import { ExecToolInjectLayer } from "../layer/ExecToolInjectLayer.ts";
import { ExecResponseInterceptor } from "../layer/ExecResponseInterceptor.ts";
import { sessionState } from "./sessionState.ts";
import { SessionCache } from "./sessionCache.ts";
import { makeReadCacheLayer } from "../layer/ReadCacheLayer.ts";
import type { ContinuationFn } from "../kb/types.ts";
import { RtkRouteLayer } from "../layer/RtkRouteLayer.ts";
import { spawnSync } from "node:child_process";
import { sanitizeRequest, detectCtxModeCommand, ctxModeBlockedResponse, type AnthropicRequest } from "./sanitize.ts";
import { ToolResultAdapterLayer } from "../layer/ToolResultAdapterLayer.ts";
import { AdapterRegistry as ToolResultAdapterRegistry } from "../layer/adapters/index.ts";
import { registerTier1 } from "../layer/adapters/registerTier1.ts";

// ---------------------------------------------------------------------------
// Hot-swap support: content-addressed build hash + structural self-check.
// ---------------------------------------------------------------------------

function computeBuildHash(): string {
  try {
    const self = fileURLToPath(import.meta.url.split("?")[0] ?? import.meta.url);
    return createHash("sha256").update(readFileSync(self)).digest("hex").slice(0, 12);
  } catch {
    return "unknown";
  }
}
export const buildHash: string = computeBuildHash();

export async function selfCheck(): Promise<{ ok: boolean; buildHash: string; reason?: string }> {
  try {
    const pipe = await makePipeline();
    const probe = {
      model: "claude-selfcheck",
      system: "selfcheck",
      messages: [{ role: "user", content: "ping roundtrip" }],
    };
    const out = await pipe.outbound(probe);
    const back = await pipe.inbound(out);
    if (JSON.stringify(back) !== JSON.stringify(probe)) {
      return { ok: false, buildHash, reason: "pipeline round-trip mismatch" };
    }
    return { ok: true, buildHash };
  } catch (err) {
    return { ok: false, buildHash, reason: err instanceof Error ? err.message : String(err) };
  }
}

export interface HandlerDeps {
  registry: ProviderRegistry;
  adapterRegistry?: AdapterRegistry;
  projectDir: string;
  upstream?: string;
  pipelineCache: Map<string, { pipeline: Pipeline; rewriter: OutputRewriteLayer }>;
  onError?: (err: unknown) => void;
  onObservations?: (events: SavingEvent[]) => void;
  onRequest?: (meta: RequestMeta, events: SavingEvent[]) => void;
}

const SESSION_ID_RE = /^[A-Za-z0-9_-]{1,128}$/;
const MAX_JSON_BODY_BYTES = 32 * 1024 * 1024;
const UPSTREAM_ERROR_LOG_MAX = 200;
const SAFE_OUTBOUND_HEADERS = new Set([
  "accept",
  "accept-encoding",
  "anthropic-beta",
  "anthropic-dangerous-direct-browser-access",
  "anthropic-version",
  "content-type",
]);

function validateSessionId(raw: string | null | undefined): string | null {
  if (!raw || raw.includes("\0") || raw.includes("/") || raw.includes("\\") || raw.includes("..")) {
    return null;
  }
  return SESSION_ID_RE.test(raw) ? raw : null;
}

async function readBodyWithLimit(req: Request, maxBytes: number): Promise<string | "too-large"> {
  const contentLength = req.headers.get("content-length");
  if (contentLength) {
    const len = Number.parseInt(contentLength, 10);
    if (!Number.isNaN(len) && len > maxBytes) return "too-large";
  }
  const buf = await req.arrayBuffer();
  if (buf.byteLength > maxBytes) return "too-large";
  return new TextDecoder().decode(buf);
}

function redactErrorBodyForLog(body: unknown): string {
  if (body == null) return String(body);
  if (typeof body !== "object") return String(body).slice(0, UPSTREAM_ERROR_LOG_MAX);
  const copy: Record<string, unknown> = { ...(body as Record<string, unknown>) };
  for (const key of Object.keys(copy)) {
    if (/auth|authorization|api[-_]?key|token|secret|cookie|password|credential/i.test(key)) {
      copy[key] = "[redacted]";
    }
  }
  return JSON.stringify(copy).slice(0, UPSTREAM_ERROR_LOG_MAX);
}

function filterSafeOutboundHeaders(headers: Headers): Record<string, string> {
  const out: Record<string, string> = {};
  for (const [k, v] of headers.entries()) {
    if (SAFE_OUTBOUND_HEADERS.has(k.toLowerCase())) out[k] = v;
  }
  return out;
}

function enrichUpstreamError(
  body: unknown,
  res: Response,
  onError?: (err: unknown) => void,
): unknown {
  const status = res.status;
  const rl: Record<string, string> = {};
  for (const [k, v] of res.headers.entries()) {
    const lk = k.toLowerCase();
    if (
      lk.startsWith("anthropic-ratelimit-") ||
      lk === "retry-after" ||
      lk === "anthropic-priority-input-tokens-server-tokens-left"
    ) {
      rl[k] = v;
    }
  }

  let enriched: unknown = body;
  if (Object.keys(rl).length > 0 && body != null && typeof body === "object") {
    enriched = { ...(body as Record<string, unknown>), _ratelimit: rl };
  }

  if (status >= 500) {
    onError?.(new Error(`upstream ${status}: ${redactErrorBodyForLog(body)}`));
  } else if (status === 429) {
    const ra = rl["retry-after"];
    if (ra) onError?.(new Error(`rate limited; retry-after: ${ra}`));
  }

  return enriched;
}

export interface FetchHandlerBundle {
  (req: Request): Promise<Response>;
  handler: (req: Request) => Promise<Response>;
  dispose(): void;
  inFlight(): number;
}

export function buildFetchHandler(
  deps: HandlerDeps,
): FetchHandlerBundle {
  const { registry, adapterRegistry, projectDir, upstream, pipelineCache, onError, onObservations, onRequest } =
    deps;

  let inFlightCount = 0;
  const inFlight = (): number => inFlightCount;

  const kbStore = new KbStore(projectDir);
  kbStore.maintain();

  const execInjectLayer = new ExecToolInjectLayer();
  const execInterceptor = new ExecResponseInterceptor(execInjectLayer.available);

  const readCacheLayerMap = new SessionCache<Layer>({
    maxEntries: 1000,
    idleMs: 60 * 60 * 1000,
    onEvict: (_k, layer) => { try { layer.dispose(); } catch { /* ignore */ } },
  });
  const readCache = openReadCache(projectDir);

  const toolResultRegistry = new ToolResultAdapterRegistry();
  registerTier1(toolResultRegistry);
  const adapterLayerCache = new SessionCache<ToolResultAdapterLayer>({
    maxEntries: 1000,
    idleMs: 60 * 60 * 1000,
    onEvict: (_k, layer) => { try { layer.dispose(); } catch { /* ignore */ } },
  });
  const frozenBytesCache = new SessionCache<Map<string, string>>({
    maxEntries: 1000,
    idleMs: 60 * 60 * 1000,
  });

  function rtkInstalled(): boolean {
    return spawnSync("which", ["rtk"], { encoding: "utf8" }).status === 0;
  }
  const rtkLayer = rtkInstalled() ? new RtkRouteLayer() : null;

  function getOrBuildReadCacheLayer(sid: string, provider: Provider): Layer {
    const existing = readCacheLayerMap.get(sid);
    if (existing) return existing;
    const layer = makeReadCacheLayer({ cache: readCache, sessionId: sid as SessionId, provider });
    layer.init();
    readCacheLayerMap.set(sid, layer);
    return layer;
  }

  async function getOrBuildPipeline(provider: Provider): Promise<{ pipeline: Pipeline; rewriter: OutputRewriteLayer }> {
    const cacheKey = provider.id;
    const cached = pipelineCache.get(cacheKey);
    if (cached) return cached;

    const kbInjectLayer = new KbToolInjectLayer();
    const pipeline = await makePipeline({ provider, extraLayers: [kbInjectLayer, execInjectLayer] });
    const rewriter = new OutputRewriteLayer(readCache);
    rewriter.init();
    const entry = { pipeline, rewriter };
    pipelineCache.set(cacheKey, entry);
    return entry;
  }

  function getOrBuildAdapterLayer(sid: string, provider: Provider, store: KbStore): ToolResultAdapterLayer {
    const existing = adapterLayerCache.get(sid);
    if (existing) return existing;
    let frozenBytes = frozenBytesCache.get(sid);
    if (!frozenBytes) {
      frozenBytes = new Map<string, string>();
      frozenBytesCache.set(sid, frozenBytes);
    }
    const layer = new ToolResultAdapterLayer({
      sessionId: sid,
      provider,
      store,
      registry: toolResultRegistry,
      frozenBytes,
    });
    adapterLayerCache.set(sid, layer);
    return layer;
  }

  const handler = async function fetchHandler(req: Request): Promise<Response> {
    const signal = req.signal;
    const startedAt = Date.now();
    const onStreamError = (err: unknown): void => onError?.(err);

    let counted = false;
    let handoff = false;
    let released = false;
    const release = (): void => {
      if (counted && !released) {
        released = true;
        inFlightCount--;
      }
    };

    try {
      const url = new URL(req.url);

      if (url.pathname.startsWith("/_ft/")) {
        return handleControl(req);
      }

      if (url.pathname === "/healthz") {
        return Response.json({
          ok: true,
          buildHash,
          primary: process.env.FT_PRIMARY !== "0",
          pid: process.pid,
        });
      }

      inFlightCount++;
      counted = true;

      const isJsonBody =
        req.method === "POST" && /json/i.test(req.headers.get("content-type") ?? "");

      if (!isJsonBody) {
        return await forwardToUpstream({
          upstreamUrl: upstream ?? "",
          method: req.method,
          path: url.pathname + url.search,
          headers: req.headers,
          body: req.body,
          signal,
          onStreamError,
        });
      }

      const resolved: ResolvedProvider | null = registry.resolve(
        req.method,
        url.pathname,
        req.headers,
      );

      if (!resolved) {
        return new Response("no provider for this route", { status: 404 });
      }

      const adapter = adapterRegistry?.resolve(req) ?? null;

      const upstreamBase =
        upstream ?? resolved.provider.defaultUpstreamUrl ?? "https://api.anthropic.com";

      const provider = resolved.provider;

      const text = await readBodyWithLimit(req, MAX_JSON_BODY_BYTES);
      if (text === "too-large") {
        return new Response("payload too large", { status: 413 });
      }
      let json: unknown;
      try {
        json = JSON.parse(text);
      } catch {
        return new Response("bad json", { status: 400 });
      }

      if (detectCtxModeCommand(json as AnthropicRequest)) {
        return ctxModeBlockedResponse();
      }

      const sanitized: AnthropicRequest = sanitizeRequest(json as AnthropicRequest);

      const model = (json as { model?: string })?.model ?? "unknown";

      const rawSessionHeader = req.headers.get("x-claude-code-session-id");
      const validatedSessionId = validateSessionId(rawSessionHeader);
      const sessionId = validatedSessionId
        ? asSessionId(validatedSessionId)
        : null;

      if (sessionId && existsSync(paths.bypassFile(String(sessionId)))) {
        return await forwardToUpstream({
          upstreamUrl: upstreamBase,
          method: req.method,
          path: url.pathname + url.search,
          headers: req.headers,
          body: text,
          signal,
          onStreamError,
        });
      }

      const { pipeline, rewriter: capturedOutputRewriter } = await getOrBuildPipeline(provider);

      const rcLayer =
        sessionId && provider.supportsCacheMarkers
          ? getOrBuildReadCacheLayer(sessionId as string, provider)
          : null;

      const adapterLayer = sessionId
        ? getOrBuildAdapterLayer(sessionId as string, provider, kbStore)
        : null;
      if (adapterLayer) adapterLayer.init();

      const requestLayers: Layer[] = [
        ...(rcLayer ? [rcLayer] : []),
        ...pipeline.layers,
        ...(adapterLayer ? [adapterLayer] : []),
      ];

      if (sessionId) {
        const messagesForKb = Array.isArray((json as Record<string, unknown>).messages)
          ? [...((json as Record<string, unknown>).messages as unknown[])]
          : [];
        sessionState.setMessages(sessionId as string, messagesForKb);
      }

      let outBody: unknown;
      try {
        outBody = await runOutbound(sanitized, requestLayers);
      } catch (err) {
        onError?.(err);
        outBody = sanitized;
      }

      const outHeaders = new Headers(req.headers);
      outHeaders.delete("content-length");

      const BYTES_PER_TOKEN = 4.5;

      const emitRequest = async (inJson: unknown, events: SavingEvent[] = []): Promise<void> => {
        if (!onRequest) return;
        const usage = (inJson as { usage?: Record<string, number> })?.usage ?? {};
        const compressedToks =
          (usage["input_tokens"] ?? 0) +
          (usage["cache_creation_input_tokens"] ?? 0) +
          (usage["cache_read_input_tokens"] ?? 0);
        let bytesSaved = 0;
        for (const ev of events) bytesSaved += (ev.rawBytes - ev.sentBytes);
        const tokenDelta = Math.round(bytesSaved / BYTES_PER_TOKEN);
        const rawTokens = compressedToks > 0 ? compressedToks + tokenDelta : 0;
        onRequest({
          at: startedAt,
          provider: provider.id,
          model,
          sessionId: validatedSessionId,
          project: projectDir,
          rawInputTokens: rawTokens,
          compressedInputTokens: compressedToks,
          rawOutputTokens: usage["output_tokens"] ?? 0,
          cacheHitInputTokens: usage["cache_read_input_tokens"] ?? 0,
          cacheCreationInputTokens: usage["cache_creation_input_tokens"] ?? 0,
          latencyMs: Date.now() - startedAt,
        }, events);
      };

      const forwardPromise = forwardToUpstream({
        upstreamUrl: upstreamBase,
        method: req.method,
        path: url.pathname + url.search,
        headers: outHeaders,
        body: JSON.stringify(outBody),
        signal,
      });
      const upstreamRes = await forwardPromise;

      const ct = upstreamRes.headers.get("content-type") ?? "";

      if (!upstreamRes.ok) {
        const inJson = await upstreamRes.json().catch(() => null);
        const enriched = enrichUpstreamError(inJson, upstreamRes, onError);
        try {
          if (process.env.FEWTOK_DEBUG === "1" && upstreamRes.status === 400) {
            const errMsg = JSON.stringify(inJson ?? {});
            if (/thinking|redacted_thinking/i.test(errMsg)) {
              const logsDir = join(homedir(), ".fewtok", "logs");
              mkdirSync(logsDir, { recursive: true, mode: 0o700 });
              chmodSync(logsDir, 0o700);
              const dumpDir = join(logsDir, "thinking-400");
              mkdirSync(dumpDir, { recursive: true, mode: 0o700 });
              chmodSync(dumpDir, 0o700);
              const ts = new Date().toISOString().replace(/[:.]/g, "-");
              const sidPart = validatedSessionId ?? "no-sid";
              const fname = join(dumpDir, `${ts}_${sidPart}.json`);
              const dump = {
                ts,
                sessionId: sidPart,
                upstreamUrl: upstreamBase,
                status: upstreamRes.status,
                errorResponse: inJson,
                outboundHeaders: filterSafeOutboundHeaders(outHeaders),
                sanitizedBody: sanitized,
                outBody,
              };
              writeFileSync(fname, JSON.stringify(dump, null, 2), { mode: 0o600 });
              onError?.(new Error(`[thinking-400-dump] wrote ${fname}`));
            }
          }
        } catch (dumpErr) { onError?.(dumpErr); }
        setImmediate(() => {
          emitRequest(enriched).catch(err => onError?.(err));
        });
        return new Response(JSON.stringify(enriched), {
          status: upstreamRes.status,
          headers: upstreamRes.headers,
        });
      }

      if (ct.includes("event-stream")) {
        const observations = collectObservations(requestLayers);
        if (observations.length > 0) onObservations?.(observations);

        if (!upstreamRes.body) {
          return new Response(null, { status: upstreamRes.status, headers: upstreamRes.headers });
        }

        const acc: SseUsageAcc = {
          input_tokens: 0,
          output_tokens: 0,
          cache_read_input_tokens: 0,
          cache_creation_input_tokens: 0,
          deltas_seen: 0,
          was_cancelled: false,
        };

        const outputRewriter = capturedOutputRewriter;

        const kbInterceptor = sessionId
          ? new KbResponseInterceptor(kbStore, sessionId as string)
          : null;

        const execContinuationFn: ContinuationFn = async (assistantContent, execToolUseId, toolResultContent) => {
          const origMessages = sessionId ? (sessionState.getMessages(sessionId as string) ?? []) : [];
          const contMessages = [
            ...origMessages,
            { role: "assistant" as const, content: assistantContent },
            { role: "user" as const, content: [{ type: "tool_result" as const, tool_use_id: execToolUseId, content: toolResultContent }] },
          ];
          const outBodyRec = outBody as Record<string, unknown>;
          const contTools = Array.isArray(outBodyRec.tools)
            ? (outBodyRec.tools as unknown[]).filter(
                t => typeof t !== "object" || t === null || (t as Record<string, unknown>).name !== "exec"
              )
            : outBodyRec.tools;
          const contBody = {
            ...outBodyRec,
            messages: contMessages,
            tools: contTools,
            stream: true as const,
          };
          const result = await forwardToUpstream({
            upstreamUrl: upstreamBase,
            method: req.method,
            path: url.pathname + url.search,
            headers: outHeaders,
            body: JSON.stringify(contBody),
            signal,
            onStreamError,
          });
          if (sessionId) sessionState.clear(sessionId as string);
          return result;
        };

        const continuationFn: ContinuationFn = async (assistantContent, kbToolUseId, toolResultContent) => {
          const origMessages = sessionId ? (sessionState.getMessages(sessionId as string) ?? []) : [];
          const contMessages = [
            ...origMessages,
            { role: "assistant" as const, content: assistantContent },
            { role: "user" as const, content: [{ type: "tool_result" as const, tool_use_id: kbToolUseId, content: toolResultContent }] },
          ];
          const outBodyRec = outBody as Record<string, unknown>;
          const contTools = Array.isArray(outBodyRec.tools)
            ? (outBodyRec.tools as unknown[]).filter(
                t => typeof t !== "object" || t === null || (t as Record<string, unknown>).name !== "kb"
              )
            : outBodyRec.tools;
          const contBody = {
            ...outBodyRec,
            messages: contMessages,
            tools: contTools,
            stream: true as const,
          };
          const result = await forwardToUpstream({
            upstreamUrl: upstreamBase,
            method: req.method,
            path: url.pathname + url.search,
            headers: outHeaders,
            body: JSON.stringify(contBody),
            signal,
            onStreamError,
          });
          if (sessionId) sessionState.clear(sessionId as string);
          return result;
        };

        const execWrapped = execInterceptor.wrapResponse(upstreamRes, execContinuationFn);
        const kbWrapped = kbInterceptor
          ? kbInterceptor.wrapResponse(execWrapped, continuationFn)
          : execWrapped;
        const rtkWrapped = rtkLayer ? rtkLayer.wrapResponse(kbWrapped) : kbWrapped;
        const sseResponse = outputRewriter
          ? outputRewriter.wrapResponse(rtkWrapped)
          : rtkWrapped;

        const stream = wrapSseForUsage(sseResponse.body!, acc, (captured) => {
          release();
          if (sessionId) sessionState.clear(sessionId as string);

          const events = [...observations];
          if (outputRewriter) {
            const rwEvents = outputRewriter.observe();
            if (rwEvents.length > 0) onObservations?.(rwEvents);
            events.push(...rwEvents);
          }
          const suspicious = captured && acc.output_tokens < 10 && acc.input_tokens > 100;

          setImmediate(() => {
            if (suspicious) {
              console.warn(
                `[sse-suspicious] low output_tokens captured: path=${url.pathname}` +
                ` input=${acc.input_tokens} output=${acc.output_tokens}` +
                ` deltas_seen=${acc.deltas_seen} was_cancelled=${acc.was_cancelled}`,
              );
            }
            if (captured) emitRequest({ usage: acc }, events).catch(err => onError?.(err));
          });
        }, adapter, onStreamError);

        const sseResp = new Response(stream, {
          status: sseResponse.status,
          headers: sseResponse.headers,
        });
        handoff = true;
        return sseResp;
      }

      // non-SSE response
      const inJson = await upstreamRes.json().catch(() => null);
      let expanded: unknown = inJson;
      try {
        expanded = await runInbound(inJson, requestLayers);
      } catch (err) {
        onError?.(err);
      }
      const observations = collectObservations(requestLayers);
      if (observations.length > 0) onObservations?.(observations);
      setImmediate(() => {
        emitRequest(inJson, observations).catch(err => onError?.(err));
      });

      return new Response(JSON.stringify(expanded), {
        status: upstreamRes.status,
        headers: upstreamRes.headers,
      });
    } catch (err) {
      if ((err as { name?: string })?.name === "AbortError") {
        return new Response("client disconnected", { status: 499 });
      }
      onError?.(err);
      return new Response("proxy error", { status: 502 });
    } finally {
      if (!handoff) release();
    }
  };

  const dispose = (): void => {
    readCacheLayerMap.dispose();
    adapterLayerCache.dispose();
    frozenBytesCache.dispose();
  };

  return Object.assign(handler, { handler, dispose, inFlight });
}