import { spawn, type ChildProcess } from "node:child_process";
import { closeSync, mkdirSync, openSync, readFileSync, writeFileSync } from "node:fs";
import { dirname, join } from "node:path";
import { DatabaseSync } from "node:sqlite";
import { fileURLToPath, pathToFileURL } from "node:url";
import { expect, it } from "vitest";

import { launchVerificationServer, runControlOmb } from "../scripts/control-omb.ts";
import { waitForExit } from "./testing/cleanup.ts";

it("survives a real server crash: queued sends keep receipts, cancellation and uncertain dispatch never replay", async () => {
  const fixture = await launchVerificationServer();
  const { url, dataDir, logPath } = fixture.info;
  let restarted: ChildProcess | undefined;
  const evidence: unknown[] = [];
  const api = async (method: string, path: string, body?: unknown, status = 200) => {
    const response = await fetch(`${url}${path}`, {
      method, headers: { "content-type": "application/json" },
      ...(body === undefined ? {} : { body: JSON.stringify(body) }), signal: AbortSignal.timeout(5_000),
    });
    const result = await response.json() as any;
    if (method !== "GET") evidence.push({ method, path, body, status: response.status, result });
    expect(response.status, JSON.stringify(result)).toBe(status);
    return result;
  };
  // A second person on a paired device: what they queue stays theirs through the wait, the crash and the restart. A row
  // written straight to the database below keeps the older name-only shape.
  const PAIRED = { name: "Safari on Mac", id: expect.stringMatching(/^p_[\w-]{22}$/) };
  const PAIRED_ROW = { name: "Safari on Mac" };
  let pairedToken = "";
  const asPairedPerson = async (path: string, body: unknown) => {
    const response = await fetch(`${url}${path}`, {
      method: "POST", headers: { "content-type": "application/json", authorization: `Bearer ${pairedToken}` },
      body: JSON.stringify(body), signal: AbortSignal.timeout(5_000),
    });
    const result = await response.json() as any;
    evidence.push({ method: "POST", as: PAIRED.name, path, body, status: response.status, result });
    expect(response.status, JSON.stringify(result)).toBe(202);
    return result;
  };
  const messages = async (thread: string) => (await api("GET", `/api/threads/${thread}/messages?limit=100`)).messages as any[];
  const prompts = (thread: string): any[] => {
    try { return readFileSync(join(dataDir, `${thread}.prompts`), "utf8").trim().split("\n").map((line) => JSON.parse(line)); }
    catch (error) {
      if (error instanceof SyntaxError || (error as NodeJS.ErrnoException).code === "ENOENT") return [];
      throw error;
    }
  };
  const journal = () => {
    const db = new DatabaseSync(join(dataDir, "messages.db"), { readOnly: true });
    try { return db.prepare("SELECT id, status FROM chat_followups ORDER BY rowid").all(); }
    finally { db.close(); }
  };
  const restart = async () => {
    const env: NodeJS.ProcessEnv = {};
    for (const [key, value] of Object.entries(process.env)) {
      if (["SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT", "LANG", "LC_ALL", "TZ"].includes(key.toUpperCase()) && value) env[key.toUpperCase()] = value;
    }
    Object.assign(env, {
      HOME: dataDir, USERPROFILE: dataDir, OMB_DATA_DIR: dataDir,
      APPDATA: join(dataDir, "AppData", "Roaming"), LOCALAPPDATA: join(dataDir, "AppData", "Local"),
      XDG_CONFIG_HOME: join(dataDir, ".config"), XDG_CACHE_HOME: join(dataDir, ".cache"),
      XDG_DATA_HOME: join(dataDir, ".local", "share"), HERMES_HOME: join(dataDir, ".hermes"),
      TEMP: join(dataDir, "tmp"), TMP: join(dataDir, "tmp"), TMPDIR: join(dataDir, "tmp"),
      OMB_PORT: new URL(url).port, OMB_WEBHOOK_PORT: String(Number(new URL(url).port) + 1), PATH: dirname(process.execPath),
    });
    const log = openSync(logPath, "a", 0o600);
    restarted = spawn(process.execPath, ["--experimental-strip-types", fileURLToPath(new URL("./index.ts", import.meta.url))], {
      cwd: fileURLToPath(new URL("..", import.meta.url)), env, stdio: ["ignore", log, log],
    });
    closeSync(log);
    await expect.poll(async () => {
      expect(restarted?.exitCode, `see ${logPath}`).toBeNull();
      try { return (await fetch(`${url}/api/health`, { signal: AbortSignal.timeout(1_000) })).ok; }
      catch { return false; }
    }, { timeout: 20_000 }).toBe(true);
  };
  try {
    const wrapper = join(dataDir, "queued-claude.mjs");
    writeFileSync(wrapper, [
      "#!/usr/bin/env node",
      'import { existsSync } from "node:fs";',
      'import { basename, join } from "node:path";',
      'const thread = basename(process.cwd());',
      `const home = ${JSON.stringify(dataDir)};`,
      'process.env.FAKE_CLAUDE_MODE = existsSync(join(home, "restarted")) ? "happy" : "slow";',
      'process.env.FAKE_CLAUDE_SLOW_FINISH_GATE = join(home, thread + ".gate");',
      'process.env.FAKE_CLAUDE_PROMPTS = join(home, thread + ".prompts");',
      // Crash the server, not unrelated processes: a fixture provider exits
      // when its parent's pipe closes, even if its fake finish gate is shut.
      'process.stdin.on("end", () => process.exit(0));',
      `await import(${JSON.stringify(pathToFileURL(fileURLToPath(new URL("./testing/fake-claude-cli.ts", import.meta.url))).href)});`,
    ].join("\n"), { mode: 0o700 });
    await api("PATCH", "/api/instances/claude", { cli: wrapper });
    await api("PATCH", "/api/config", { threads: { maxConcurrentPerBot: 1 } });
    const pairing = await fetch(`${url}/api/auth/pair`, {
      method: "POST",
      headers: { "content-type": "application/json", "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0) Safari/605.1" },
      body: JSON.stringify({ code: (await api("POST", "/api/auth/pairing", {})).code }),
    });
    const paired = await pairing.json() as any;
    expect(pairing.status, JSON.stringify(paired)).toBe(200);
    expect(paired.session.label).toBe(PAIRED.name);
    pairedToken = paired.token;
    const bot = (await api("POST", "/api/bots", { name: "Durable follow-ups" }, 201)).bot;
    const uncertain = (await api("POST", "/api/bots", { name: "Uncertain follow-up" }, 201)).bot;
    const later = (await api("POST", `/api/bots/${uncertain.id}/tasks`, { title: "In-flight receipt" }, 201)).task;
    const worker = (await api("POST", "/api/bots", { name: "Channel worker" }, 201)).bot;
    const channel = (await api("POST", "/api/groups", {
      name: "Durable channel", memberIds: [worker.id],
      setup: { bulletin: "", defaultResponder: { kind: "member", botId: worker.id } },
    }, 201)).group;

    await api("POST", `/api/bots/${bot.id}/messages`, { text: "Working before restart", threadId: bot.threadId }, 202);
    await expect.poll(() => prompts(bot.threadId).length, { timeout: 15_000 }).toBe(1);
    const replyToId = (await messages(bot.threadId)).find((message) => message.role === "user").id;
    // Claude takes pictures into a running turn, so words for this busy
    // thread would be steered in. Let the turn finish and have a second task
    // hold the bot's one slot: the attachment then waits in the durable queue.
    writeFileSync(join(dataDir, `${bot.threadId}.gate`), "finish the first task");
    expect(await runControlOmb(["wait", "--bot", bot.id, "--task", bot.threadId, "--url", url])).toMatchObject({ status: "settled" });
    const holder = (await api("POST", `/api/bots/${bot.id}/tasks`, { title: "Hold the slot" }, 201)).task;
    await api("POST", `/api/bots/${bot.id}/messages`, { text: "Hold the slot", threadId: holder.threadId }, 202);
    await expect.poll(() => prompts(holder.threadId).length, { timeout: 15_000 }).toBe(1);
    const attachments = join(dataDir, "attachments");
    mkdirSync(attachments, { recursive: true });
    const image = join(attachments, "123e4567-e89b-42d3-a456-426614174000.png");
    writeFileSync(image, Buffer.from("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+j7f8AAAAASUVORK5CYII=", "base64"));
    const text = `Inspect screenshot after the first task\n\n<attached-image path="${image}" name="fixture.png" />`;
    const body = { text, threadId: bot.threadId, replyToId, sendId: "durable_bot_send_123456" };
    const queued = await asPairedPerson(`/api/bots/${bot.id}/messages`, body);
    expect(queued).toMatchObject({ queued: true, queueId: expect.any(String), reason: "capacity" });
    const cancelledBody = { ...body, text: `${text}\n\nCancelled task`, sendId: "cancelled_bot_send_123456" };
    const cancelled = await api("POST", `/api/bots/${bot.id}/messages`, cancelledBody, 202);
    await api("DELETE", `/api/bots/${bot.id}/queue/${cancelled.queueId}`, { threadId: bot.threadId });

    await api("POST", `/api/bots/${uncertain.id}/messages`, { threadId: uncertain.threadId, text: "Hold capacity" }, 202);
    const uncertainBody = { threadId: later.threadId, text: "A possibly executed action", sendId: "uncertain_bot_send_123456" };
    const claimed = await api("POST", `/api/bots/${uncertain.id}/messages`, uncertainBody, 202);
    expect(claimed).toMatchObject({ queued: true, reason: "capacity" });
    writeFileSync(join(dataDir, `${uncertain.threadId}.gate`), "finish first task only");
    await expect.poll(() => prompts(later.threadId).length, { timeout: 15_000 }).toBe(1);

    // Stop revokes provider credentials before the provider reports completion;
    // it must still retire the already-dispatched queue receipt exactly once.
    const stopped = (await api("POST", "/api/bots", { name: "Stopped follow-up" }, 201)).bot;
    const stoppedTask = (await api("POST", `/api/bots/${stopped.id}/tasks`, { title: "Stop this follow-up" }, 201)).task;
    await api("POST", `/api/bots/${stopped.id}/messages`, { threadId: stopped.threadId, text: "Hold capacity before Stop" }, 202);
    const stoppedBody = { threadId: stoppedTask.threadId, text: "Stop this dispatched follow-up", sendId: "stopped_bot_send_123456" };
    const stoppedReceipt = await api("POST", `/api/bots/${stopped.id}/messages`, stoppedBody, 202);
    expect(stoppedReceipt).toMatchObject({ queued: true, reason: "capacity" });
    writeFileSync(join(dataDir, `${stopped.threadId}.gate`), "release capacity for the stopped follow-up");
    await expect.poll(() => prompts(stoppedTask.threadId).length, { timeout: 15_000 }).toBe(1);
    expect(journal()).toContainEqual({ id: stoppedReceipt.queueId, status: "dispatching" });
    await api("POST", `/api/bots/${stopped.id}/interrupt`, { threadId: stoppedTask.threadId });
    await expect.poll(() => journal().some((row) => row.id === stoppedReceipt.queueId), { timeout: 5_000 }).toBe(false);
    expect((await api("POST", `/api/bots/${stopped.id}/messages`, stoppedBody, 202)).message.queueId).toBe(stoppedReceipt.queueId);

    const initial = await api("POST", `/api/groups/${channel.id}/messages`, { text: "Working channel" }, 202);
    const channelBody = { text: "Channel follow-up after restart", threadId: channel.threadId,
      replyToId: initial.message.id, sendId: "durable_channel_send_123456", mode: "chat" };
    const channelQueued = await api("POST", `/api/groups/${channel.id}/messages`, channelBody, 202);
    expect(channelQueued).toMatchObject({ queued: true, queueId: expect.any(String) });
    const channelCancelledBody = { ...channelBody, text: "Never run cancelled channel task", sendId: "cancelled_channel_send_123456" };
    const channelCancelled = await api("POST", `/api/groups/${channel.id}/messages`, channelCancelledBody, 202);
    await api("DELETE", `/api/groups/${channel.id}/queue/${channelCancelled.queueId}`);
    // A queued room reply whose target is gone by the time it can run: the
    // drain cannot start the turn, and keeps the person's words in their name.
    const lostWorker = (await api("POST", "/api/bots", { name: "Lost reply worker" }, 201)).bot;
    const lost = (await api("POST", "/api/groups", {
      name: "Lost reply channel", memberIds: [lostWorker.id],
      setup: { bulletin: "", defaultResponder: { kind: "member", botId: lostWorker.id } },
    }, 201)).group;
    const lostTarget = await api("POST", `/api/groups/${lost.id}/messages`, { text: "Working before the reply target is lost" }, 202);
    const lostQueued = await asPairedPerson(`/api/groups/${lost.id}/messages`, {
      text: "Reply to a message that disappears", threadId: lost.threadId, replyToId: lostTarget.message.id, sendId: "lost_reply_channel_send_123456",
    });
    expect(lostQueued).toMatchObject({ queued: true, queueId: expect.any(String) });
    const unappended = (await api("POST", "/api/bots", { name: "Claimed before append" }, 201)).bot;
    expect(journal()).toEqual(expect.arrayContaining([
      { id: queued.queueId, status: "pending" }, { id: cancelled.queueId, status: "cancelled" },
      { id: claimed.queueId, status: "dispatching" }, { id: channelQueued.queueId, status: "pending" },
    ]));

    await waitForExit(fixture.child, { signal: "SIGKILL" });
    writeFileSync(join(dataDir, "restarted"), "allow restored fake turns to finish");
    const crashed = new DatabaseSync(join(dataDir, "messages.db"));
    try {
      // The narrowest crash window: the dispatch claim reached disk, the transcript line did not. One row names its sender, one
      // is from a build that kept none, one was spoken in a Live call, and one typed row carries a stray call id.
      const claim = crashed.prepare(
        "INSERT INTO chat_followups(id, kind, owner_id, thread_id, send_id, status, payload) VALUES (?, 'bot', ?, ?, NULL, 'dispatching', ?)",
      );
      claim.run("claimed_before_append_named", unappended.id, unappended.threadId, JSON.stringify({ text: "Claimed, never appended", sender: PAIRED_ROW }));
      claim.run("claimed_before_append_legacy", unappended.id, unappended.threadId, JSON.stringify({ text: "Claimed by an older build" }));
      claim.run("claimed_before_append_call", unappended.id, unappended.threadId, JSON.stringify({ text: "Claimed from a call", via: "call", callId: "call-restart-fixture" }));
      claim.run("claimed_before_append_stray", unappended.id, unappended.threadId, JSON.stringify({ text: "Typed, with a stray call id", callId: "call-stray" }));
      crashed.prepare("UPDATE messages SET json = json_set(json, '$.text', '') WHERE thread_id = ? AND id = ?")
        .run(lost.threadId, lostTarget.message.id);
    } finally { crashed.close(); }
    await restart();
    expect((await messages(stoppedTask.threadId)).some((message) => message.queueId === stoppedReceipt.queueId && message.kind === "activity")).toBe(false);
    expect(prompts(stoppedTask.threadId)).toHaveLength(1);
    await expect.poll(async () => (await messages(bot.threadId)).filter((message) => message.sendId === body.sendId).length, { timeout: 15_000 }).toBe(1);
    const settled = await runControlOmb(["wait", "--bot", bot.id, "--task", bot.threadId, "--url", url]);
    expect(settled).toMatchObject({ status: "settled" });
    expect((await api("POST", `/api/bots/${bot.id}/messages`, body, 202)).message).toMatchObject({
      sendId: body.sendId, queueId: queued.queueId, replyToId, text, sender: PAIRED,
    });
    expect((await messages(unappended.threadId)).filter((message) => message.role === "user").map((message) => [message.queueId, message.text, message.sender, message.via, message.callId])).toEqual([
      ["claimed_before_append_named", "Claimed, never appended", PAIRED_ROW, undefined, undefined],
      ["claimed_before_append_legacy", "Claimed by an older build", undefined, undefined, undefined],
      ["claimed_before_append_call", "Claimed from a call", undefined, "call", "call-restart-fixture"],
      ["claimed_before_append_stray", "Typed, with a stray call id", undefined, undefined, undefined],
    ]);
    await expect.poll(async () => (await messages(lost.threadId)).some((message) => message.tool?.name?.includes("queued channel message could not start")), { timeout: 15_000 }).toBe(true);
    expect((await messages(lost.threadId)).filter((message) => message.queueId === lostQueued.queueId)).toEqual([
      expect.objectContaining({ role: "user", text: "Reply to a message that disappears", sender: PAIRED }),
    ]);
    expect((await api("POST", `/api/bots/${bot.id}/messages`, cancelledBody, 409)).error).toContain("cancelled");
    expect(prompts(bot.threadId)).toHaveLength(2);
    expect(prompts(bot.threadId)[1].message.content).toEqual(expect.arrayContaining([expect.objectContaining({ type: "image" })]));

    await expect.poll(async () => (await messages(channel.threadId)).filter((message) => message.sendId === channelBody.sendId).length, { timeout: 15_000 }).toBe(1);
    expect((await api("POST", `/api/groups/${channel.id}/messages`, channelBody, 202)).message).toMatchObject({
      text: channelBody.text, queueId: channelQueued.queueId, replyToId: initial.message.id, channelMode: "chat", via: "api",
    });
    expect((await api("POST", `/api/groups/${channel.id}/messages`, channelCancelledBody, 409)).error).toContain("cancelled");
    const channelSettled = await runControlOmb(["wait", "--channel", channel.id, "--url", url]);
    expect(channelSettled).toMatchObject({ status: "settled" });
    await expect.poll(() => journal().some((row) => row.id === queued.queueId || row.id === channelQueued.queueId)).toBe(false);
    const interrupted = await messages(later.threadId);
    expect(interrupted.filter((message) => message.queueId === claimed.queueId)).toEqual([
      expect.objectContaining({ role: "user", text: uncertainBody.text, sendId: uncertainBody.sendId }),
      expect.objectContaining({ kind: "activity", tool: { name: expect.stringContaining("Review the result"), ok: false, nameCode: "chat.notice.followupInterrupted" } }),
    ]);
    expect((await api("POST", `/api/bots/${uncertain.id}/messages`, uncertainBody, 202)).message.queueId).toBe(claimed.queueId);
    expect(prompts(later.threadId)).toHaveLength(1);
    await waitForExit(restarted, { signal: "SIGTERM" });
    await restart();
    expect((await messages(later.threadId)).filter((message) => message.queueId === claimed.queueId)).toEqual(interrupted.filter((message) => message.queueId === claimed.queueId));
    expect(prompts(later.threadId)).toHaveLength(1);
    evidence.push({ settled, channelSettled, journal: journal(), recoveredBot: await messages(bot.threadId), recoveredChannel: await messages(channel.threadId), interrupted });
  } finally {
    await waitForExit(restarted, { signal: "SIGTERM" });
    const evidencePath = `${logPath}.chat-followups-restart.json`;
    writeFileSync(evidencePath, JSON.stringify({ fixture: fixture.info, evidence }, null, 2));
    console.log(JSON.stringify({ logPath, evidencePath }));
    await fixture.close();
  }
}, 90_000);