// src/proxy/server.ts
// Thin daemon shell. Owns Bun.serve, pipelineCache, and the reload() contract.
// All request-handling logic lives in handler.ts so that `fewtok reload`
// can swap it in-process without restarting the daemon.

import { existsSync, copyFileSync, renameSync, unlinkSync } from "node:fs";
import { join } from "node:path";
import { findFreePort } from "../lifecycle/port.ts";
import { openStats } from "../stats/writer.ts";
import { anthropicProvider } from "../provider/anthropic.ts";
import { ProviderRegistry } from "../provider/registry.ts";
import { paths } from "../lifecycle/paths.ts";
import type { Provider } from "../provider/types.ts";
import type { SavingEvent } from "../layer/types.ts";
import type { RequestMeta, SavingRow } from "../stats/writer.ts";
import type { HandlerDeps, FetchHandlerBundle, buildFetchHandler as BuildFetchHandlerType } from "./handler.ts";
import { getLedger } from "../stats/ledger.ts";

const flushLedger = (): void => { try { getLedger().flush(); } catch { /* swallow */ } };

export interface ProxyOpts {
  port: number;
  upstream?: string;
  registry: ProviderRegistry;
  projectDir: string;
  onError?: (err: unknown) => void;
  onObservations?: (events: SavingEvent[]) => void;
  onRequest?: (meta: RequestMeta, events: SavingEvent[]) => void;
}

export interface ProxyHandle {
  port: number;
  stop(): Promise<void>;
  reload(): Promise<void>;
  inFlight(): number;
}

export interface StartProxyOpts {
  projectDir: string;
  upstream?: string;
  providers?: Provider[];
  port?: number;
  // Primary designation. The supervisor sets FT_PRIMARY=1 on exactly one child;
  // the primary owns shared-resource maintenance (WAL checkpoint, ctx-mode fence).
  primary?: boolean;
  // Invoked when this child is promoted to primary mid-flight (SIGUSR2 from the
  // supervisor after the old primary exits during a graceful restart / crash).
  onBecomePrimary?: () => void;
}

export interface StartProxyHandle {
  url: string;
  port: number;
  close(): Promise<void>;
}

function savingsFrom(events: SavingEvent[]): SavingRow[] {
  const rows: SavingRow[] = [];
  for (const e of events) {
    // output_rewrite events carry rawBytes=0/sentBytes=0 by spec (savings live in meta.sigils_expanded)
    if (e.rawBytes <= e.sentBytes && e.layerId !== "output_rewrite") continue;
    rows.push({
      layerId: e.layerId,
      kind: e.kind,
      rawBytes: e.rawBytes,
      sentBytes: e.sentBytes,
      ...(e.meta ? { meta: e.meta } : {}),
    });
  }
  return rows;
}

type HandlerModule = {
  buildFetchHandler: typeof BuildFetchHandlerType;
  selfCheck: () => Promise<{ ok: boolean; reason?: string }>;
  buildHash: string;
};

// dist/proxy-handler.js (live) is the bytes currently served. proxy-handler.next.js
// is a built candidate awaiting verify; proxy-handler.lkg.js is the last proven-good
// bundle, the fallback for cold-start and crashloop recovery.
//
// ALL THREE end in .js, and that is load-bearing: bun infers the OUTPUT MODULE FORMAT
// from the FINAL outfile extension. A .js name → ESM named exports (matching
// package.json "type":"module"); an unrecognized final extension (.next / .lkg)
// silently emits CJS (module.exports), which `import()` then surfaces as
// ["__esModule","default"] — the named exports HandlerModule requires are gone and
// reload throws. `--format=esm` does NOT override this (verified no-op). Hence the
// candidate/fallback are proxy-handler.next.js / proxy-handler.lkg.js, not *.js.next.
type BundleSource = "live" | "next" | "lkg";

function bundlePath(which: BundleSource): string {
  const dir = paths.distDir();
  return which === "live"
    ? join(dir, "proxy-handler.js")
    : join(dir, `proxy-handler.${which}.js`);
}

async function loadHandlerModule(which: BundleSource, bust: boolean): Promise<HandlerModule> {
  const file = bundlePath(which);
  if (existsSync(file)) {
    // Cache-bust via ?v= query so Bun re-imports the swapped bytes after a build.
    const spec = bust ? `${file}?v=${Date.now()}` : file;
    return (await import(spec)) as HandlerModule;
  }
  if (which === "live") {
    // Dev fallback: import source directly (Bun native TS).
    return (await import("./handler.ts")) as HandlerModule;
  }
  throw new Error(`bundle not found: ${file}`);
}

// Defer the OLD bundle's dispose() until its in-flight count drains to 0 (or the
// cap elapses). dispose() tears down the per-bundle SessionCaches; firing it while
// a stream is still flowing would rip those caches out from under an in-flight
// request. Non-blocking: polls on a timer so the reload returns immediately.
function drainAndDispose(old: FetchHandlerBundle, capMs: number): void {
  const start = Date.now();
  const tick = (): void => {
    let n = 0;
    try { n = old.inFlight(); } catch { n = 0; }
    if (n <= 0 || Date.now() - start >= capMs) {
      try { old.dispose(); } catch { /* ignore */ }
      if (n > 0) {
        console.error(`[fewtok reload] drain cap ${capMs}ms hit; disposed old bundle with ${n} in-flight`);
      }
      return;
    }
    setTimeout(tick, 100);
  };
  tick();
}

// Awaitable drain: resolves true the moment in-flight hits 0, false at the cap.
// Used by graceful stop() — unlike drainAndDispose (fire-and-forget on reload),
// the caller must wait so it can force-close stragglers and dispose in order.
function drainInFlight(b: FetchHandlerBundle, capMs: number): Promise<boolean> {
  return new Promise((resolve) => {
    const start = Date.now();
    const tick = (): void => {
      let n = 0;
      try { n = b.inFlight(); } catch { n = 0; }
      if (n <= 0) { resolve(true); return; }
      if (Date.now() - start >= capMs) { resolve(false); return; }
      setTimeout(tick, 100);
    };
    tick();
  });
}

export async function createProxy(opts: ProxyOpts): Promise<ProxyHandle> {
  const pipelineCache: HandlerDeps["pipelineCache"] = new Map();

  const deps: HandlerDeps = {
    registry: opts.registry,
    projectDir: opts.projectDir,
    pipelineCache,
    ...(opts.upstream !== undefined ? { upstream: opts.upstream } : {}),
    ...(opts.onError ? { onError: opts.onError } : {}),
    ...(opts.onObservations ? { onObservations: opts.onObservations } : {}),
    ...(opts.onRequest ? { onRequest: opts.onRequest } : {}),
  };

  // Hot-swap state. `bundleLiveSince` marks when `bundle` became live; a bundle
  // that has served ≥ PROVEN_MS counts as proven and may seed .lkg (see reload L1).
  // `servingMatchesDistLive` tracks L1's load-bearing invariant: the bytes ON DISK
  // at distLive == the bytes currently SERVED. True after a normal boot/reload (we
  // loaded/renamed those exact bytes); FALSE after a .lkg-fallback boot (we serve
  // lkg-in-memory while distLive on disk is still the corrupt bytes that threw).
  // Step 3's `cp live→.lkg` MUST consult this, else a fallback-boot reload copies
  // corrupt distLive over the only good .lkg before step 4 repairs distLive.
  const PROVEN_MS = 60_000;
  const DRAIN_CAP_MS = 30_000;
  // Graceful-stop drain ceiling. Generous by design: the SUPERVISOR's SIGKILL
  // escalation timer (short on a real stop, long on a graceful-restart retire) is
  // the true governor of how long an exiting child may keep draining a live SSE
  // stream — this is only the in-process upper bound before we force-close.
  const STOP_DRAIN_CAP_MS = Number(process.env.FT_DRAIN_CAP_MS ?? 600_000);
  const distLive = bundlePath("live");
  const distLkg = bundlePath("lkg");
  let bundleLiveSince = Date.now();
  let servingMatchesDistLive = true;

  // Cold-start guard (#9): a live bundle that THROWS at import (e.g. a manual
  // `bun build` straight to dist that produced bad bytes) would crash the child →
  // the supervisor respawns the SAME bytes → crashloop → bricked proxy. Fall back
  // to the last-known-good bundle so the daemon still boots and serves.
  // C1: the supervisor sets FT_USE_LKG=1 on a child it is respawning after the LIVE
  // bytes crashlooped (a manual `bun build` straight to dist that imports fine but
  // crashes under traffic — selfCheck can't catch that). Boot from .lkg directly so
  // the respawn serves proven bytes instead of re-loading the same poison. distLive
  // on disk stays the bad bytes → disarm L1 (same as the catch fallback below).
  const forceLkg = process.env.FT_USE_LKG === "1" && existsSync(distLkg);
  let bundle: FetchHandlerBundle;
  try {
    if (forceLkg) {
      const mod = await loadHandlerModule("lkg", false);
      bundle = mod.buildFetchHandler(deps);
      servingMatchesDistLive = false;
      console.error("[fewtok] FT_USE_LKG=1 — booted from .lkg (live bytes quarantined)");
    } else {
    const mod = await loadHandlerModule("live", false);
    bundle = mod.buildFetchHandler(deps);
    // Bootstrap the safety net: the bytes we just booted on ARE known-good. Seed
    // .lkg from live if absent so crashloop recovery has a fallback from boot.
    if (existsSync(distLive) && !existsSync(distLkg)) {
      try { copyFileSync(distLive, distLkg); } catch { /* best-effort */ }
    }
    }
  } catch (err) {
    console.error("[fewtok] live bundle failed to load; falling back to .lkg:", err);
    const mod = await loadHandlerModule("lkg", false);
    bundle = mod.buildFetchHandler(deps);
    // We serve lkg-in-memory, but distLive on disk is STILL the corrupt bytes that
    // threw. Disarm L1: a later reload must not copy that corrupt distLive over the
    // good .lkg. Re-armed only after step 4 renames good bytes onto distLive.
    servingMatchesDistLive = false;
    console.error("[fewtok] booted from .lkg fallback");
  }

  const server = Bun.serve({
    hostname: "127.0.0.1",
    port: opts.port,
    idleTimeout: 255,
    // SO_REUSEPORT: the kernel load-balances new connections across every live
    // listener bound to this port. Enables zero-drop full-process swaps — a new
    // child binds the same port and serves alongside the old until it retires.
    reusePort: true,
    fetch: (req: Request) => bundle.handler(req),
  });

  const handle: ProxyHandle = {
    port: server.port ?? opts.port,

    inFlight: () => { try { return bundle.inFlight(); } catch { return 0; } },

    async stop(): Promise<void> {
      // Order matters under reusePort: stop the listener FIRST so the kernel stops
      // routing NEW connections to this child (they go to any sibling listener),
      // THEN drain in-flight requests already accepted, THEN force-close any
      // stragglers past the cap, and only then dispose the bundle's caches — never
      // tear caches out from under a live stream.
      try { await server.stop(); } catch { /* already stopping */ }
      const drained = await drainInFlight(bundle, STOP_DRAIN_CAP_MS);
      if (!drained) { try { await server.stop(true); } catch { /* ignore */ } }
      try { bundle.dispose(); } catch { /* ignore */ }
    },

    async reload(): Promise<void> {
      try {
        // 1. Import candidate bytes (.next), cache-busted. A bad build that throws
        //    at import never touches live — discard and keep serving the old bundle.
        const nextFile = bundlePath("next");
        if (!existsSync(nextFile)) {
          console.error("[fewtok reload] failed: no .next bundle to promote");
          return;
        }
        let candidate: HandlerModule;
        try {
          candidate = await loadHandlerModule("next", true);
        } catch (err) {
          try { unlinkSync(nextFile); } catch { /* ignore */ }
          opts.onError?.(err);
          console.error("[fewtok reload] failed: .next import threw:", err);
          return;
        }
        // 2. Structural self-check BEFORE promote — round-trips the codec + a dry
        //    pipeline pass on an empty snapshot. A gate, not a correctness oracle
        //    (it cannot drive the real SSE/upstream path) — hence the L1 rule below.
        const verdict = await candidate.selfCheck();
        if (!verdict.ok) {
          try { unlinkSync(nextFile); } catch { /* ignore */ }
          console.error(`[fewtok reload] failed: selfCheck ${verdict.reason ?? "unknown"}`);
          return;
        }
        // 3. L1 — seed .lkg from the OUTGOING live bundle, but ONLY if it has been
        //    live long enough to count as PROVEN. selfCheck can't exercise the real
        //    serving path, so .lkg must lag proven-running bytes, never merely
        //    selfCheck-passing bytes — else a fast double-reload promotes a
        //    passing-but-broken bundle into the only crashloop fallback.
        if (servingMatchesDistLive && existsSync(distLive) && Date.now() - bundleLiveSince >= PROVEN_MS) {
          try { copyFileSync(distLive, distLkg); } catch { /* best-effort */ }
        }
        // 4. Promote: atomically rename .next → live so the cold-start and crashloop
        //    paths only ever observe a complete bundle.
        try {
          renameSync(nextFile, distLive);
          // distLive on disk now == the proven bytes about to serve. Re-arm L1.
          servingMatchesDistLive = true;
        } catch (err) {
          opts.onError?.(err);
          console.error("[fewtok reload] failed: rename .next→live:", err);
          return;
        }
        // 5. Build the new bundle and swap the closure var. Clearing the deps caches
        //    only drops Map entries; in-flight requests captured their pipeline +
        //    rewriter refs at handler entry, so they survive the clear.
        deps.pipelineCache.clear();
        const newBundle = candidate.buildFetchHandler(deps);
        const old = bundle;
        bundle = newBundle;
        bundleLiveSince = Date.now();
        // 6. Defer the OLD bundle's dispose() until its in-flight count drains (D1).
        //    No server.reload() — the Bun.serve fetch lambda closes over `bundle`
        //    (let); reassigning is sufficient (server.reload has a Bun 1.3.6 bug
        //    where it stores the options object as fetchFn).
        drainAndDispose(old, DRAIN_CAP_MS);
        console.log(`[fewtok reload] handler swapped at ${new Date().toISOString()} build=${candidate.buildHash}`);
      } catch (err) {
        opts.onError?.(err);
        console.error("[fewtok reload] failed:", err);
      }
    },
  };

  return handle;
}

export async function startProxy(opts: StartProxyOpts): Promise<StartProxyHandle> {
  const port = opts.port ?? (await findFreePort());
  const stats = openStats(paths.root());
  const registry = new ProviderRegistry();
  for (const p of opts.providers ?? [anthropicProvider]) {
    registry.register(p);
  }

  // Primary designation. Under the supervisor exactly one child carries FT_PRIMARY=1;
  // a standalone (unsupervised) proxy is always primary. Only the primary runs shared
  // single-writer maintenance — WAL checkpoint here, ctx-mode fence in start.ts — so a
  // two-child graceful-restart window never double-runs them. Per-request writeRequest
  // is append-only and stays in every child.
  let isPrimary = opts.primary ?? (process.env.FT_PRIMARY === "1" || process.env.FT_SUPERVISED !== "1");

  // Boot-time WAL truncate: collapses bloated WAL from prior runs. Primary-only.
  if (isPrimary) stats.checkpointTruncate();

  const handle = await createProxy({
    port,
    ...(opts.upstream !== undefined ? { upstream: opts.upstream } : {}),
    registry,
    projectDir: opts.projectDir,
    onError(err) {
      const name = (err as { name?: string })?.name;
      if (name === "AbortError") return;
      const msg = (err as { message?: string })?.message ?? String(err);
      console.error(`[fewtok proxy] ${name ?? "error"}: ${msg}`);
    },
    onRequest(meta, events = []) {
      stats.writeRequest(meta, savingsFrom(events));
    },
  });

  // SavingsPoller pulls process-local stats adapters — safe in EVERY child (each
  // serves its own requests), so it is owned here, not gated on primary.
  const { SavingsPoller } = await import("../stats/SavingsPoller.ts");
  const poller = new SavingsPoller(getLedger);
  poller.start();

  // Promotion to primary (supervisor SIGUSR2 after the old primary retires/dies):
  // assume the single-writer maintenance the old primary owned. Idempotent.
  const becomePrimary = (): void => {
    if (isPrimary) return;
    isPrimary = true;
    // The handler bundle's /healthz and the ledger's statusline gate read
    // process.env.FT_PRIMARY live per-request, so the promotion must be reflected
    // there too — flipping the in-closure `isPrimary` alone leaves them stale.
    process.env.FT_PRIMARY = "1";
    try { stats.checkpointTruncate(); } catch { /* best-effort */ }
    try { opts.onBecomePrimary?.(); } catch { /* best-effort */ }
  };
  process.on("SIGUSR2", () => becomePrimary());

  // Register SIGUSR1 for in-process hot-reload.
  // The signal fires in the main thread where Bun.serve lives.
  // Wrapped in try/catch — a throw in a signal handler crashes the daemon.
  process.on("SIGUSR1", () => {
    flushLedger();
    handle.reload()
      .then(() => {
        stats.applyPragmas();
      })
      .catch((err) => console.error("[fewtok SIGUSR1] reload failed:", err));
  });

  process.on("SIGHUP", () => {
    flushLedger();
    handle.reload()
      .then(() => stats.applyPragmas())
      .catch((err) => console.error("[fewtok SIGHUP] reload failed:", err));
  });

  // SIGTERM/SIGINT: graceful shutdown — dispose caches, stop server, then exit.
  const shutdown = (sig: string) => async (): Promise<void> => {
    try { await handle.stop(); } catch (err) {
      console.error(`[fewtok ${sig}] stop failed:`, err);
    }
    // Stop the poller AFTER the drain so its final tick captures just-drained
    // requests; the ledger adapters outlive bundle.dispose().
    try { poller.stop(); } catch { /* ignore */ }
    flushLedger();
    process.exit(0);
  };
  process.on("SIGTERM", () => { void shutdown("SIGTERM")(); });
  process.on("SIGINT", () => { void shutdown("SIGINT")(); });

  return {
    url: `http://127.0.0.1:${handle.port}`,
    port: handle.port,
    close: handle.stop.bind(handle),
  };
}