/**
 * Reproducible terminal latency benchmark (Node-only, isolated daemon).
 *
 * Boots its OWN isolated daemon subprocess (fresh mkdtemp PASEO_HOME, random
 * port) — it NEVER touches the developer daemon on port 6767 — and measures:
 *   A) terminal echo latency  (single-byte input -> first echoed output frame)
 *   B) terminal output jitter (inter-frame gaps while a command drains ~2MB)
 *   C) RPC ping RTT at 10Hz (proxy for daemon main-loop delay)
 * across load levels L0..L3 where load is generated by dev-only `mock` agents
 * that stream agent updates over the same socket the benchmark client uses.
 *
 * Run:
 *   npx tsx scripts/benchmark-terminal-latency.ts
 *
 * Optional:
 *   BENCH_INCLUDE_L3=1   include the L3 level (L2 + a second noisy terminal)
 *
 * Output: pretty table to stdout + JSON to /tmp/paseo-terminal-bench/<ts>.json
 *
 * Requires built client/protocol dist (packages/client/dist). Build with:
 *   npm run build:client
 */

import { spawn, execSync, type ChildProcess } from "node:child_process";
import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises";
import net from "node:net";
import os from "node:os";
import path from "node:path";
import { performance } from "node:perf_hooks";
import { fileURLToPath, pathToFileURL } from "node:url";
import { WebSocket } from "ws";

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const REPO_ROOT = path.resolve(__dirname, "..");
const SERVER_DIR = path.join(REPO_ROOT, "packages", "server");

// --- minimal structural view of the dist DaemonClient we depend on ----------

interface TerminalStreamEvent {
  terminalId: string;
  type: "output" | "snapshot" | "restore";
  data?: Uint8Array;
}

interface DaemonClientLike {
  connect(): Promise<void>;
  close(): Promise<void>;
  ping(params?: { timeoutMs?: number }): Promise<{ rttMs: number }>;
  listTerminals(cwd?: string): Promise<unknown>;
  openProject(cwd: string): Promise<{ workspace: { id: string } | null; error: string | null }>;
  createTerminal(
    cwd: string,
    name?: string,
    requestId?: string,
    options?: { workspaceId?: string },
  ): Promise<{ terminal: { id: string } | null; error: string | null }>;
  subscribeTerminal(terminalId: string): Promise<{ error: string | null }>;
  killTerminal(terminalId: string): Promise<unknown>;
  sendTerminalInput(
    terminalId: string,
    message: { type: "input"; data: string } | { type: "resize"; rows: number; cols: number },
  ): void;
  onTerminalStreamEvent(handler: (event: TerminalStreamEvent) => void): () => void;
  createAgent(options: {
    provider: string;
    cwd: string;
    workspaceId?: string;
    model?: string;
    title?: string;
  }): Promise<{ id: string }>;
  sendAgentMessage(agentId: string, text: string): Promise<void>;
  cancelAgent?(agentId: string): Promise<unknown>;
  deleteAgent?(agentId: string): Promise<unknown>;
}

interface DaemonClientCtor {
  new (config: {
    url: string;
    clientId: string;
    clientType: "cli";
    appVersion?: string;
    webSocketFactory: (url: string, options?: { headers?: Record<string, string> }) => unknown;
  }): DaemonClientLike;
}

// --- stats helpers ----------------------------------------------------------

interface Stats {
  count: number;
  min: number;
  p50: number;
  p95: number;
  max: number;
  avg: number;
}

function summarize(values: number[]): Stats {
  if (values.length === 0) {
    return { count: 0, min: 0, p50: 0, p95: 0, max: 0, avg: 0 };
  }
  const sorted = [...values].sort((a, b) => a - b);
  const pick = (q: number) => sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * q))];
  return {
    count: sorted.length,
    min: round(sorted[0]),
    p50: round(pick(0.5)),
    p95: round(pick(0.95)),
    max: round(sorted[sorted.length - 1]),
    avg: round(values.reduce((a, b) => a + b, 0) / values.length),
  };
}

function round(n: number): number {
  return Math.round(n * 100) / 100;
}

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

// --- daemon lifecycle -------------------------------------------------------

async function getFreePort(): Promise<number> {
  return new Promise((resolve, reject) => {
    const srv = net.createServer();
    srv.on("error", reject);
    srv.listen(0, "127.0.0.1", () => {
      const addr = srv.address();
      if (addr === null || typeof addr === "string") {
        srv.close();
        reject(new Error("Failed to acquire a free port"));
        return;
      }
      const { port } = addr;
      srv.close(() => resolve(port));
    });
  });
}

async function waitForPort(port: number, child: ChildProcess, timeoutMs: number): Promise<void> {
  const start = Date.now();
  let lastError: unknown = null;
  while (Date.now() - start < timeoutMs) {
    if (child.exitCode !== null) {
      throw new Error(`Daemon exited before listening (code ${child.exitCode})`);
    }
    try {
      await new Promise<void>((resolve, reject) => {
        const socket = net.connect(port, "127.0.0.1", () => {
          socket.end();
          resolve();
        });
        socket.setTimeout(1000, () => {
          socket.destroy();
          reject(new Error("connect timeout"));
        });
        socket.on("error", reject);
      });
      return;
    } catch (error) {
      lastError = error;
      await sleep(150);
    }
  }
  throw new Error(
    `Daemon did not listen on 127.0.0.1:${port} within ${timeoutMs}ms (${String(lastError)})`,
  );
}

interface BootedDaemon {
  child: ChildProcess;
  port: number;
  paseoHome: string;
  pid: number;
}

async function bootDaemon(): Promise<BootedDaemon> {
  const port = await getFreePort();
  if (port === 6767) {
    throw new Error("Refusing to use port 6767 (the developer daemon)");
  }
  const paseoHome = await mkdtemp(path.join(os.tmpdir(), "paseo-bench-home-"));
  const tsxBin = execSync("which tsx").toString().trim();

  const child = spawn(tsxBin, ["scripts/supervisor-entrypoint.ts", "--dev"], {
    cwd: SERVER_DIR,
    env: {
      ...process.env,
      PASEO_HOME: paseoHome,
      PASEO_SERVER_ID: "srv_terminal_bench",
      PASEO_LISTEN: `127.0.0.1:${port}`,
      PASEO_NODE_ENV: "development",
      NODE_ENV: "development",
    },
    stdio: ["ignore", "pipe", "pipe"],
    detached: false,
  });

  child.stdout?.on("data", (data: Buffer) => {
    for (const line of data.toString().split("\n")) {
      const trimmed = line.trim();
      if (trimmed) console.log(`[daemon] ${trimmed}`);
    }
  });
  child.stderr?.on("data", (data: Buffer) => {
    for (const line of data.toString().split("\n")) {
      const trimmed = line.trim();
      if (trimmed) console.error(`[daemon] ${trimmed}`);
    }
  });

  await waitForPort(port, child, 30_000);
  return { child, port, paseoHome, pid: child.pid ?? -1 };
}

async function loadDaemonClientCtor(): Promise<DaemonClientCtor> {
  const moduleUrl = pathToFileURL(
    path.join(REPO_ROOT, "packages", "client", "dist", "daemon-client.js"),
  ).href;
  const mod = (await import(moduleUrl)) as { DaemonClient: DaemonClientCtor };
  return mod.DaemonClient;
}

async function connectClient(ctor: DaemonClientCtor, port: number): Promise<DaemonClientLike> {
  const client = new ctor({
    url: `ws://127.0.0.1:${port}/ws`,
    clientId: `bench-${Math.random().toString(36).slice(2)}`,
    clientType: "cli",
    webSocketFactory: (url, options) => new WebSocket(url, { headers: options?.headers }),
  });
  await client.connect();
  return client;
}

// --- measurement primitives -------------------------------------------------

const ECHO_SAMPLES = 100;
const ECHO_INTERVAL_MS = 50;
const ECHO_FRAME_TIMEOUT_MS = 2_000;
// Printable ASCII the shell echoes back verbatim (skip control chars/space).
const ECHO_ALPHABET = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";

interface EchoResult {
  samples: number;
  hits: number;
  latencyMs: Stats;
}

/**
 * Sends single distinct bytes spaced out in time; each byte's latency is the
 * gap between send and the first output frame whose decoded text contains it.
 */
async function measureEcho(client: DaemonClientLike, terminalId: string): Promise<EchoResult> {
  const decoder = new TextDecoder();
  // The stream handler records the arrival time of the awaited echo byte; the
  // sender loop polls this between sends. A plain shared slot avoids juggling a
  // promise resolver across the handler/timeout boundary.
  // Mutated by the async stream handler and read by the sender loop; a holder
  // object keeps the cross-callback mutation explicit.
  const slot: { awaitedChar: string | null; arrivalAt: number | null } = {
    awaitedChar: null,
    arrivalAt: null,
  };

  const unsubscribe = client.onTerminalStreamEvent((event) => {
    if (event.terminalId !== terminalId || event.type !== "output" || !event.data) {
      return;
    }
    if (slot.awaitedChar === null || slot.arrivalAt !== null) {
      return;
    }
    if (decoder.decode(event.data).includes(slot.awaitedChar)) {
      slot.arrivalAt = performance.now();
    }
  });

  const latencies: number[] = [];
  let hits = 0;
  try {
    for (let i = 0; i < ECHO_SAMPLES; i += 1) {
      const char = ECHO_ALPHABET[i % ECHO_ALPHABET.length];
      slot.awaitedChar = char;
      slot.arrivalAt = null;
      const sentAt = performance.now();
      client.sendTerminalInput(terminalId, { type: "input", data: char });

      const deadline = sentAt + ECHO_FRAME_TIMEOUT_MS;
      while (slot.arrivalAt === null && performance.now() < deadline) {
        await sleep(1);
      }
      if (slot.arrivalAt !== null) {
        latencies.push(slot.arrivalAt - sentAt);
        hits += 1;
      }
      slot.awaitedChar = null;
      await sleep(ECHO_INTERVAL_MS);
    }
  } finally {
    unsubscribe();
  }

  return { samples: ECHO_SAMPLES, hits, latencyMs: summarize(latencies) };
}

const BURST_TARGET_BYTES = 2_000_000;
const BURST_CHUNK = 16_384;
const BURST_DRAIN_TIMEOUT_MS = 20_000;
const BURST_QUIET_MS = 400;

interface BurstResult {
  bytesSeen: number;
  outputFrames: number;
  snapshotFrames: number;
  restoreFrames: number;
  drainMs: number;
  interFrameGapMs: Stats;
}

/**
 * Runs a node one-liner in the terminal that prints ~2MB rapidly, then records
 * inter-output-frame gaps until the stream goes quiet. Snapshot/restore frames
 * during the burst indicate the snapshot-churn path firing.
 */
async function measureBurst(client: DaemonClientLike, terminalId: string): Promise<BurstResult> {
  let bytesSeen = 0;
  let outputFrames = 0;
  let snapshotFrames = 0;
  let restoreFrames = 0;
  const gaps: number[] = [];
  let lastBurstFrameAt = 0;
  let firstBurstFrameAt = 0;
  let lastActivityAt = performance.now();

  // Frames smaller than this are the echoed command / prompt noise, not burst
  // payload — gap timing only starts once real drain data is flowing, so the
  // node startup latency doesn't pollute the inter-frame jitter.
  const BURST_FRAME_MIN_BYTES = 1_024;

  const unsubscribe = client.onTerminalStreamEvent((event) => {
    if (event.terminalId !== terminalId) {
      return;
    }
    const now = performance.now();
    lastActivityAt = now;
    if (event.type === "snapshot") {
      snapshotFrames += 1;
      return;
    }
    if (event.type === "restore") {
      restoreFrames += 1;
      return;
    }
    if (event.type !== "output" || !event.data) {
      return;
    }
    outputFrames += 1;
    bytesSeen += event.data.byteLength;
    if (event.data.byteLength < BURST_FRAME_MIN_BYTES && firstBurstFrameAt === 0) {
      return;
    }
    if (firstBurstFrameAt === 0) {
      firstBurstFrameAt = now;
    } else {
      gaps.push(now - lastBurstFrameAt);
    }
    lastBurstFrameAt = now;
  });

  // Clear whatever the echo measurement left on the shell input line (Ctrl-C),
  // then let the fresh prompt settle before sending the burst command.
  client.sendTerminalInput(terminalId, { type: "input", data: "" });
  await sleep(300);

  const startedAt = performance.now();
  try {
    const chunks = Math.ceil(BURST_TARGET_BYTES / BURST_CHUNK);
    // Print BURST_TARGET_BYTES of 'x' in chunks, no trailing newline spam.
    const oneLiner =
      `const c='x'.repeat(${BURST_CHUNK});` +
      `for(let i=0;i<${chunks};i++)process.stdout.write(c);` +
      `process.stdout.write('\\nBURST_DONE\\n')`;
    client.sendTerminalInput(terminalId, {
      type: "input",
      data: `node -e "${oneLiner}"\r`,
    });

    const deadline = performance.now() + BURST_DRAIN_TIMEOUT_MS;
    while (performance.now() < deadline) {
      await sleep(50);
      const quietFor = performance.now() - lastActivityAt;
      // The daemon coalesces a large pty write into a snapshot plus a handful of
      // frames, so the client rarely sees the full BURST_TARGET_BYTES. Exit once
      // real burst data has arrived and the stream has gone quiet.
      if (firstBurstFrameAt > 0 && quietFor >= BURST_QUIET_MS) {
        break;
      }
    }
  } finally {
    unsubscribe();
  }

  return {
    bytesSeen,
    outputFrames,
    snapshotFrames,
    restoreFrames,
    drainMs: round((lastBurstFrameAt || startedAt) - startedAt),
    interFrameGapMs: summarize(gaps),
  };
}

const PING_INTERVAL_MS = 100;

/**
 * Samples RPC RTT at 10Hz for `durationMs`. Uses `ping` if available, else a
 * cheap listTerminals request. Returns RTT stats — a proxy for main-loop delay.
 */
async function measurePing(client: DaemonClientLike, durationMs: number): Promise<Stats> {
  const rtts: number[] = [];
  const deadline = Date.now() + durationMs;
  while (Date.now() < deadline) {
    const start = performance.now();
    try {
      const result = await client.ping({ timeoutMs: 5_000 });
      rtts.push(typeof result?.rttMs === "number" ? result.rttMs : performance.now() - start);
    } catch {
      const fallbackStart = performance.now();
      await client.listTerminals().catch(() => undefined);
      rtts.push(performance.now() - fallbackStart);
    }
    await sleep(PING_INTERVAL_MS);
  }
  return summarize(rtts);
}

// --- load generation --------------------------------------------------------

interface LoadController {
  agentIds: string[];
  noisyTerminalId: string | null;
  stop(): Promise<void>;
}

interface LevelSpec {
  id: string;
  description: string;
  mockAgents: number;
  bigDiffEvery2s: boolean;
  secondNoisyTerminal: boolean;
}

const LEVELS: LevelSpec[] = [
  {
    id: "L0",
    description: "idle",
    mockAgents: 0,
    bigDiffEvery2s: false,
    secondNoisyTerminal: false,
  },
  {
    id: "L1",
    description: "2 mock agents streaming",
    mockAgents: 2,
    bigDiffEvery2s: false,
    secondNoisyTerminal: false,
  },
  {
    id: "L2",
    description: "4 mock agents + 200KB diff bursts every 2s",
    mockAgents: 4,
    bigDiffEvery2s: true,
    secondNoisyTerminal: false,
  },
];

if (process.env.BENCH_INCLUDE_L3 === "1") {
  LEVELS.push({
    id: "L3",
    description: "L2 + a second terminal continuously emitting",
    mockAgents: 4,
    bigDiffEvery2s: true,
    secondNoisyTerminal: true,
  });
}

async function startLoad(
  client: DaemonClientLike,
  cwd: string,
  workspaceId: string,
  spec: LevelSpec,
): Promise<LoadController> {
  const agentIds: string[] = [];
  for (let i = 0; i < spec.mockAgents; i += 1) {
    const agent = await client.createAgent({
      provider: "mock",
      cwd,
      workspaceId,
      model: "five-minute-stream",
      title: `${spec.id}-load-${i}`,
    });
    agentIds.push(agent.id);
    // Kick off the sustained stream.
    await client.sendAgentMessage(agent.id, "start streaming").catch(() => undefined);
  }

  let bigDiffTimer: ReturnType<typeof setInterval> | null = null;
  if (spec.bigDiffEvery2s && agentIds.length > 0) {
    bigDiffTimer = setInterval(() => {
      const target = agentIds[0];
      void client
        .sendAgentMessage(target, "emit 200000 byte large diff agent stream update")
        .catch(() => undefined);
    }, 2_000);
    bigDiffTimer.unref?.();
  }

  let noisyTerminalId: string | null = null;
  if (spec.secondNoisyTerminal) {
    const created = await client.createTerminal(cwd, `${spec.id}-noisy`, undefined, {
      workspaceId,
    });
    if (created.terminal) {
      noisyTerminalId = created.terminal.id;
      await client.subscribeTerminal(noisyTerminalId);
      // Continuous infinite emit loop in the shell.
      client.sendTerminalInput(noisyTerminalId, {
        type: "input",
        data: `node -e "setInterval(()=>process.stdout.write('y'.repeat(4096)),20)"\r`,
      });
    }
  }

  // Let the streams warm up before we start measuring.
  await sleep(500);

  return {
    agentIds,
    noisyTerminalId,
    async stop() {
      if (bigDiffTimer) {
        clearInterval(bigDiffTimer);
      }
      if (noisyTerminalId) {
        // Ctrl-C then kill the terminal.
        client.sendTerminalInput(noisyTerminalId, { type: "input", data: "" });
        await client.killTerminal(noisyTerminalId).catch(() => undefined);
      }
      for (const agentId of agentIds) {
        if (client.cancelAgent) {
          await client.cancelAgent(agentId).catch(() => undefined);
        }
        if (client.deleteAgent) {
          await client.deleteAgent(agentId).catch(() => undefined);
        }
      }
    },
  };
}

// --- per-level run ----------------------------------------------------------

interface LevelResult {
  id: string;
  description: string;
  echo: EchoResult;
  burst: BurstResult;
  pingMs: Stats;
}

async function runLevel(
  client: DaemonClientLike,
  cwd: string,
  workspaceId: string,
  spec: LevelSpec,
  measuredTerminalId: string,
): Promise<LevelResult> {
  console.log(`\n=== ${spec.id} (${spec.description}) ===`);
  const load = await startLoad(client, cwd, workspaceId, spec);
  try {
    // Ping sampling runs concurrently with the echo + burst measurements so it
    // observes the daemon under the same load the latency primitives see.
    let pingMs: Stats = summarize([]);
    const pingDone = (async () => {
      pingMs = await measurePing(client, 14_000);
    })();

    console.log(`  measuring echo latency (${ECHO_SAMPLES} samples)...`);
    const echo = await measureEcho(client, measuredTerminalId);
    console.log(
      `    echo p50=${echo.latencyMs.p50}ms p95=${echo.latencyMs.p95}ms max=${echo.latencyMs.max}ms (${echo.hits}/${echo.samples} hits)`,
    );

    console.log("  measuring output burst/jitter (~2MB)...");
    const burst = await measureBurst(client, measuredTerminalId);
    console.log(
      `    burst drain=${burst.drainMs}ms frames=${burst.outputFrames} gap_p95=${burst.interFrameGapMs.p95}ms snap=${burst.snapshotFrames} restore=${burst.restoreFrames}`,
    );

    await pingDone;
    console.log(`    ping p50=${pingMs.p50}ms p95=${pingMs.p95}ms max=${pingMs.max}ms`);

    return { id: spec.id, description: spec.description, echo, burst, pingMs };
  } finally {
    await load.stop();
  }
}

// --- output -----------------------------------------------------------------

function pad(value: string | number, width: number): string {
  return String(value).padStart(width);
}

function printTable(results: LevelResult[]): void {
  console.log("\n========== TERMINAL LATENCY BENCHMARK ==========\n");
  const header = [
    pad("level", 6),
    pad("echo_p50", 9),
    pad("echo_p95", 9),
    pad("echo_max", 9),
    pad("ping_p50", 9),
    pad("ping_p95", 9),
    pad("gap_p95", 8),
    pad("gap_max", 8),
    pad("drain", 8),
    pad("frames", 7),
    pad("snap", 5),
    pad("restore", 8),
  ].join(" ");
  console.log(header);
  console.log("-".repeat(header.length));
  for (const r of results) {
    console.log(
      [
        pad(r.id, 6),
        pad(r.echo.latencyMs.p50, 9),
        pad(r.echo.latencyMs.p95, 9),
        pad(r.echo.latencyMs.max, 9),
        pad(r.pingMs.p50, 9),
        pad(r.pingMs.p95, 9),
        pad(r.burst.interFrameGapMs.p95, 8),
        pad(r.burst.interFrameGapMs.max, 8),
        pad(r.burst.drainMs, 8),
        pad(r.burst.outputFrames, 7),
        pad(r.burst.snapshotFrames, 5),
        pad(r.burst.restoreFrames, 8),
      ].join(" "),
    );
  }
  console.log("");
  for (const r of results) {
    console.log(`  ${r.id}: ${r.description}`);
  }
  console.log("");
}

function getCommitHash(): string {
  try {
    return execSync("git rev-parse --short HEAD", { cwd: REPO_ROOT }).toString().trim();
  } catch {
    return "unknown";
  }
}

// --- main -------------------------------------------------------------------

async function main(): Promise<void> {
  const commit = getCommitHash();
  console.log(`Terminal latency benchmark — commit ${commit}`);
  console.log("Booting isolated daemon (random port, fresh PASEO_HOME)...");

  let daemon: BootedDaemon | null = null;
  let client: DaemonClientLike | null = null;
  let workspaceDir: string | null = null;
  let measuredTerminalId: string | null = null;

  try {
    daemon = await bootDaemon();
    console.log(`Daemon ready: pid=${daemon.pid} port=${daemon.port} home=${daemon.paseoHome}`);

    const ctor = await loadDaemonClientCtor();
    client = await connectClient(ctor, daemon.port);

    workspaceDir = await mkdtemp(path.join(os.tmpdir(), "paseo-bench-ws-"));
    const opened = await client.openProject(workspaceDir);
    if (!opened.workspace) {
      throw new Error(`Failed to open project: ${opened.error}`);
    }

    const workspaceId = opened.workspace.id;

    const created = await client.createTerminal(workspaceDir, "measured", undefined, {
      workspaceId,
    });
    if (!created.terminal) {
      throw new Error(`Failed to create terminal: ${created.error}`);
    }
    measuredTerminalId = created.terminal.id;
    await client.subscribeTerminal(measuredTerminalId);

    // Let the shell print its prompt and settle before measuring.
    await sleep(750);

    const results: LevelResult[] = [];
    for (const spec of LEVELS) {
      const result = await runLevel(client, workspaceDir, workspaceId, spec, measuredTerminalId);
      results.push(result);
      // Settle between levels so a previous level's load fully drains.
      await sleep(2_000);
    }

    printTable(results);

    const outDir = "/tmp/paseo-terminal-bench";
    await mkdir(outDir, { recursive: true });
    const timestamp = new Date().toISOString().replace(/[:.]/g, "-");
    const outPath = path.join(outDir, `${timestamp}.json`);
    await writeFile(
      outPath,
      JSON.stringify(
        {
          commit,
          daemonPid: daemon.pid,
          port: daemon.port,
          paseoHome: daemon.paseoHome,
          node: process.version,
          platform: `${os.platform()} ${os.arch()}`,
          createdAt: new Date().toISOString(),
          levels: results,
        },
        null,
        2,
      ),
    );
    console.log(`Wrote results to ${outPath}`);
  } finally {
    if (client && measuredTerminalId) {
      await client.killTerminal(measuredTerminalId).catch(() => undefined);
    }
    if (client) {
      await client.close().catch(() => undefined);
    }
    if (daemon) {
      daemon.child.kill("SIGTERM");
      await Promise.race([
        new Promise<void>((resolve) => daemon!.child.once("exit", () => resolve())),
        sleep(5_000),
      ]);
      if (daemon.child.exitCode === null && daemon.child.signalCode === null) {
        daemon.child.kill("SIGKILL");
      }
      await rm(daemon.paseoHome, { recursive: true, force: true }).catch(() => undefined);
    }
    if (workspaceDir) {
      await rm(workspaceDir, { recursive: true, force: true }).catch(() => undefined);
    }
  }
}

main().catch((error) => {
  console.error(error);
  process.exit(1);
});