import { test } from "node:test";
import assert from "node:assert/strict";
import { readFileSync, writeFileSync, existsSync } from "node:fs";
import { join } from "node:path";
import { createConnector, FIXED_LINES, RECONCILE_WINDOW_MS } from "../src/connector.mjs";
import { readOutbox, readDrops, listTurns, readInboxIds, appendOutbox, appendInbox, ensureJournal, requestStop, writeTurn, countAdmissionsOn, noticeOn } from "../src/journal.mjs";
import { makeRoot, binding, message, IDS, fakeRest, fakeGateway, fakeEngine } from "./helpers.mjs";

function setup({ bindingOverrides = {}, replies = [], outcomes = [], now, engineOverrides = {} } = {}) {
  const root = makeRoot();
  const journalDir = join(root, "journal");
  const b = binding(bindingOverrides);
  const rest = fakeRest({ outcomes });
  const gateway = fakeGateway();
  const engine = fakeEngine({ replies, ...engineOverrides });
  const logs = [];
  const clock = now || (() => Date.now());
  const connector = createConnector({ binding: b, journalDir, rest, gateway, engine, now: clock, typingIntervalMs: 5, log: (m) => logs.push(m) });
  return { root, journalDir, b, rest, gateway, engine, connector, logs };
}

test("delivery: an accepted message is in the inbox before the turn, the reply is chunked with one nonce per chunk, and the turn record is write-once", async () => {
  const long = ["para one " + "a".repeat(1000), "para two " + "b".repeat(1000), "short"].join("\n\n");
  const { journalDir, rest, engine, connector } = setup({ replies: [{ text: long }] });
  await connector.start();
  const r = await connector.handleMessage(message({ id: "300000000000000001", content: "tell me" }));
  assert.equal(r.accepted, true);
  assert.ok(readInboxIds(journalDir).has("300000000000000001"));
  assert.equal(await r.turn, "ok");
  assert.equal(engine.prompts.length, 1);
  assert.match(engine.prompts[0].text, /^\[discord server="Test Server" channel="#seat-admin" thread=none author=100000000000000100 message=300000000000000001\]\ntell me$/);
  assert.equal(rest.calls.length, 2);
  assert.deepEqual(rest.calls.map((c) => c.nonce), ["300000000000000001-0", "300000000000000001-1"]);
  assert.equal(rest.calls[0].replyTo, "300000000000000001");
  assert.equal(rest.calls[1].replyTo, null);
  assert.ok(rest.calls.every((c) => c.content.length <= 1900));
  const outbox = readOutbox(journalDir);
  assert.deepEqual([...outbox.values()].map((e) => e.status), ["confirmed", "confirmed"]);
  const turns = listTurns(journalDir);
  assert.equal(turns.length, 1);
  assert.equal(turns[0].status, "ok");
  assert.equal(turns[0].reply.chunks.length, 2);
  assert.deepEqual(turns[0].engine.usage, { input: 1, output: 1 });
  assert.throws(() => writeTurn(journalDir, "300000000000000001", {}), /already exists/);
  assert.ok(rest.typingCalls.length >= 1);
  await connector.stop();
  assert.equal(engine.stopped, true);
});

test("delivery: refused and unknown outcomes are journaled; a later chunk is not sent after a failure", async () => {
  const long = "x".repeat(1500) + "\n\n" + "y".repeat(1500) + "\n\n" + "z";
  const { journalDir, rest, connector } = setup({ replies: [{ text: long }, { text: long }], outcomes: [{ ok: false, kind: "refused" }, { ok: false, kind: "unknown" }] });
  await connector.start();
  await (await connector.handleMessage(message({ id: "300000000000000002" }))).turn;
  await (await connector.handleMessage(message({ id: "300000000000000003" }))).turn;
  const outbox = readOutbox(journalDir);
  assert.equal(outbox.get("300000000000000002-0").status, "refused");
  assert.equal(outbox.has("300000000000000002-1"), false);
  assert.equal(outbox.get("300000000000000003-0").status, "unknown");
  assert.equal(outbox.get("300000000000000003-0").content.length, 1500);
  assert.equal(rest.calls.length, 2);
  const turns = listTurns(journalDir).sort((a, b) => a.id.localeCompare(b.id));
  assert.equal(turns[0].status, "ok");
  assert.equal(turns[0].reply.chunks[0].status, "refused");
  await connector.stop();
});

test("delivery: restart with an unknown entry re-sends the same nonce once and reconciles before accepting traffic", async () => {
  const root = makeRoot();
  const journalDir = join(root, "journal");
  ensureJournal(journalDir);
  const at = new Date().toISOString();
  appendOutbox(journalDir, { nonce: "300000000000000004-0", channelId: IDS.admin, replyTo: "300000000000000004", status: "intent", content: "hello again", at });
  appendOutbox(journalDir, { nonce: "300000000000000004-0", channelId: IDS.admin, replyTo: "300000000000000004", status: "unknown", error: "socket", at });
  appendOutbox(journalDir, { nonce: "300000000000000005-0", channelId: IDS.admin, replyTo: null, status: "intent", content: "died before receipt", at });
  const rest = fakeRest({ outcomes: [{ ok: true, messageId: "existing-1" }, { ok: true, messageId: "existing-2" }] });
  const connector = createConnector({ binding: binding(), journalDir, rest, gateway: fakeGateway(), engine: fakeEngine() });
  const started = await connector.start();
  assert.equal(started.reconciled.length, 2);
  assert.deepEqual(rest.calls.map((c) => [c.nonce, c.content, c.replyTo]), [
    ["300000000000000004-0", "hello again", "300000000000000004"],
    ["300000000000000005-0", "died before receipt", null],
  ]);
  const outbox = readOutbox(journalDir);
  assert.equal(outbox.get("300000000000000004-0").status, "confirmed");
  assert.equal(outbox.get("300000000000000004-0").messageId, "existing-1");
  assert.equal(outbox.get("300000000000000005-0").reconciled, true);
  await connector.stop();
});

test("delivery: an unknown entry older than the dedupe window is marked refused, not re-sent; a still-unknown one refuses start", async () => {
  const root = makeRoot();
  const journalDir = join(root, "journal");
  ensureJournal(journalDir);
  const old = new Date(Date.now() - RECONCILE_WINDOW_MS - 1000).toISOString();
  appendOutbox(journalDir, { nonce: "old-0", channelId: IDS.admin, status: "unknown", content: "old", at: old });
  appendOutbox(journalDir, { nonce: "fresh-0", channelId: IDS.admin, status: "unknown", content: "fresh", at: new Date().toISOString() });
  const rest = fakeRest({ outcomes: [{ ok: false, kind: "unknown" }] });
  const connector = createConnector({ binding: binding(), journalDir, rest, gateway: fakeGateway(), engine: fakeEngine() });
  await assert.rejects(connector.start(), /1 unreconciled/);
  assert.deepEqual(rest.calls.map((c) => c.nonce), ["fresh-0"]);
  const outbox = readOutbox(journalDir);
  assert.equal(outbox.get("old-0").status, "refused");
  assert.match(outbox.get("old-0").error, /stale/);
  assert.equal(outbox.get("fresh-0").status, "unknown");
});

test("delivery: repeated unknown reconciliations never refresh the dedupe window; the original intent time decides", async () => {
  const root = makeRoot();
  const journalDir = join(root, "journal");
  ensureJournal(journalDir);
  const t0 = Date.parse("2026-09-13T10:00:00Z");
  appendOutbox(journalDir, { nonce: "300000000000000006-0", channelId: IDS.admin, replyTo: "300000000000000006", status: "intent", content: "once", at: new Date(t0).toISOString() });
  // First restart at T+4m: the re-send comes back unknown again.
  let t = t0 + 4 * 60 * 1000;
  let rest = fakeRest({ outcomes: [{ ok: false, kind: "unknown" }] });
  let connector = createConnector({ binding: binding(), journalDir, rest, gateway: fakeGateway(), engine: fakeEngine(), now: () => t });
  await assert.rejects(connector.start(), /1 unreconciled/);
  assert.equal(rest.calls.length, 1);
  assert.equal(readOutbox(journalDir).get("300000000000000006-0").intentAt, new Date(t0).toISOString());
  // Second restart at T+8m: eight minutes after the original send, so outside the window; not re-sent.
  t = t0 + 8 * 60 * 1000;
  rest = fakeRest({ outcomes: [{ ok: true, messageId: "must-not-happen" }] });
  connector = createConnector({ binding: binding(), journalDir, rest, gateway: fakeGateway(), engine: fakeEngine(), now: () => t });
  const started = await connector.start();
  assert.equal(rest.calls.length, 0, "a stale intent must not be re-sent");
  assert.equal(started.reconciled[0].stale, true);
  assert.equal(readOutbox(journalDir).get("300000000000000006-0").status, "refused");
  await connector.stop();
});

test("turn: a failed engine turn posts the fixed line, never model output, and writes a failed record", async () => {
  const { journalDir, rest, connector } = setup({ replies: [{ error: "provider exploded", code: "engine-error" }] });
  await connector.start();
  await (await connector.handleMessage(message({ id: "300000000000000006" }))).turn;
  assert.equal(rest.calls.length, 1);
  assert.equal(rest.calls[0].content, FIXED_LINES.failed);
  assert.equal(rest.calls[0].nonce, "300000000000000006-f-0");
  const [turn] = listTurns(journalDir);
  assert.equal(turn.status, "failed");
  assert.equal(turn.error.code, "engine-error");
  await connector.stop();
});

test("turn: a second message during a turn goes to the engine as a follow-up, both get their own reply and record", async () => {
  const { journalDir, rest, engine, connector } = setup({ replies: [{ text: "first", delayMs: 30 }, { text: "second" }] });
  await connector.start();
  const a = await connector.handleMessage(message({ id: "300000000000000007", content: "one" }));
  const b = await connector.handleMessage(message({ id: "300000000000000008", content: "two" }));
  assert.equal(connector.inFlight, 2);
  await Promise.all([a.turn, b.turn]);
  assert.equal(engine.prompts.length, 2);
  assert.deepEqual(rest.calls.map((c) => c.content), ["first", "second"]);
  assert.equal(listTurns(journalDir).length, 2);
  await connector.stop();
});

test("turn: a thread under a listed channel is answered in the thread; an unknown thread is looked up once", async () => {
  const { rest, engine, connector, journalDir } = setup({ replies: [{ text: "in thread" }] });
  rest.channels.set(IDS.threadOfAdmin, { id: IDS.threadOfAdmin, type: 11, parent_id: IDS.admin, name: "a-thread", guild_id: IDS.guild });
  await connector.start();
  const r = await connector.handleMessage(message({ id: "300000000000000009", channel_id: IDS.threadOfAdmin }));
  assert.equal(await r.turn, "ok");
  assert.equal(rest.calls[0].channelId, IDS.threadOfAdmin);
  assert.match(engine.prompts[0].text, /thread="a-thread"/);
  const [turn] = listTurns(journalDir);
  assert.equal(turn.threadId, IDS.threadOfAdmin);
  assert.equal(turn.channelId, IDS.admin);
  // GUILD_CREATE seeds the channel map so no lookup is needed.
  connector.onDispatch({ t: "GUILD_CREATE", d: { id: IDS.guild, channels: [], threads: [{ id: IDS.threadOfOther, type: 11, parent_id: IDS.other, name: "t" }] } });
  const d = await connector.handleMessage(message({ id: "300000000000000010", channel_id: IDS.threadOfOther }));
  assert.equal(d.reason, "thread-parent-unlisted");
  await connector.stop();
});

test("drop: an unlisted user gets silence and one drop line; no inbox entry, no REST call, no engine call", async () => {
  const { journalDir, rest, engine, connector } = setup();
  await connector.start();
  const r = await connector.handleMessage(message({ id: "300000000000000011", author: { id: IDS.stranger } }));
  assert.deepEqual(r, { accepted: false, reason: "user-unlisted" });
  const drops = readDrops(journalDir);
  assert.equal(drops.length, 1);
  assert.equal(drops[0].reason, "user-unlisted");
  assert.equal(drops[0].authorId, IDS.stranger);
  assert.equal(readInboxIds(journalDir).size, 0);
  assert.equal(rest.calls.length, 0);
  assert.equal(engine.prompts.length, 0);
  await connector.stop();
});

test("drop: an oversize message is accepted into the inbox, answered with the fixed line and journaled as a drop", async () => {
  const { journalDir, rest, engine, connector } = setup();
  await connector.start();
  const r = await connector.handleMessage(message({ id: "300000000000000012", content: "q".repeat(4001) }));
  assert.equal(r.reason, "oversize");
  assert.ok(readInboxIds(journalDir).has("300000000000000012"));
  assert.equal(rest.calls[0].content, FIXED_LINES.oversize);
  assert.equal(engine.prompts.length, 0);
  assert.equal(listTurns(journalDir).length, 0);
  await connector.stop();
});

test("restart: an inbox with three ids and a replay of the same three produces zero turns", async () => {
  const root = makeRoot();
  const journalDir = join(root, "journal");
  ensureJournal(journalDir);
  const ids = ["300000000000000021", "300000000000000022", "300000000000000023"];
  for (const id of ids) appendInbox(journalDir, { id, at: new Date().toISOString() });
  const rest = fakeRest();
  const engine = fakeEngine();
  const gateway = fakeGateway();
  const connector = createConnector({ binding: binding(), journalDir, rest, gateway, engine });
  const started = await connector.start();
  assert.equal(started.inbox, 3);
  for (const id of ids) gateway.emit("dispatch", { t: "MESSAGE_CREATE", d: message({ id }) });
  await new Promise((r) => setTimeout(r, 20));
  assert.equal(engine.prompts.length, 0);
  assert.equal(rest.calls.length, 0);
  assert.equal(listTurns(journalDir).length, 0);
  assert.deepEqual(readDrops(journalDir).map((d) => d.reason), ["duplicate", "duplicate", "duplicate"]);
  // A fourth, new id still works, and the inbox file has exactly four ids.
  gateway.emit("dispatch", { t: "MESSAGE_CREATE", d: message({ id: "300000000000000024" }) });
  await new Promise((r) => setTimeout(r, 20));
  assert.equal(engine.prompts.length, 1);
  assert.equal(readInboxIds(journalDir).size, 4);
  await connector.stop();
});

test("stop: STOP present refuses start; STOP written while running refuses new turns and the current one finishes", async () => {
  const { journalDir, rest, engine, connector } = setup({ replies: [{ text: "done", delayMs: 30 }] });
  requestStop(journalDir, "test");
  await assert.rejects(connector.start(), /STOP is present/);
  assert.equal(engine.started, false);
  writeFileSync(join(journalDir, "STOP"), ""); // still present; clear it
  const { unlinkSync } = await import("node:fs");
  unlinkSync(join(journalDir, "STOP"));
  await connector.start();
  const a = await connector.handleMessage(message({ id: "300000000000000031" }));
  requestStop(journalDir, "test");
  const b = await connector.handleMessage(message({ id: "300000000000000032" }));
  assert.equal(b.reason, "stopped");
  assert.equal(await a.turn, "ok");
  assert.equal(rest.calls.length, 1);
  assert.equal(listTurns(journalDir).length, 1);
  assert.ok(readInboxIds(journalDir).has("300000000000000032"));
  await connector.stop();
});

test("ceiling: the ceiling plus one is refused and journaled; one fixed line per UTC day; a new day accepts again", async () => {
  let t = Date.parse("2026-09-13T23:59:00Z");
  const now = () => t;
  const { journalDir, rest, engine, connector } = setup({ bindingOverrides: { limits: { turnsPerDay: 2 } }, now });
  await connector.start();
  await (await connector.handleMessage(message({ id: "300000000000000041" }))).turn;
  await (await connector.handleMessage(message({ id: "300000000000000042" }))).turn;
  assert.equal(listTurns(journalDir).length, 2);
  const r3 = await connector.handleMessage(message({ id: "300000000000000043" }));
  assert.equal(r3.reason, "ceiling");
  const r4 = await connector.handleMessage(message({ id: "300000000000000044" }));
  assert.equal(r4.reason, "ceiling");
  assert.equal(engine.prompts.length, 2);
  assert.equal(listTurns(journalDir).length, 2);
  const ceilingLines = rest.calls.filter((c) => c.content === FIXED_LINES.ceiling);
  assert.equal(ceilingLines.length, 1);
  assert.equal(ceilingLines[0].nonce, "300000000000000043-c-0");
  const drops = readDrops(journalDir).filter((d) => d.reason === "ceiling");
  assert.equal(drops.length, 2);
  assert.equal(drops[0].limit, 2);
  // Midnight UTC passes: accepted again.
  t = Date.parse("2026-09-14T00:00:01Z");
  const r5 = await connector.handleMessage(message({ id: "300000000000000045" }));
  assert.equal(r5.accepted, true);
  assert.equal(await r5.turn, "ok");
  await connector.stop();
});

test("ceiling: a burst arriving while turns are still running cannot queue past the ceiling", async () => {
  const now = () => Date.parse("2026-09-13T12:00:00Z");
  const { journalDir, engine, connector } = setup({ bindingOverrides: { limits: { turnsPerDay: 2 } }, now, engineOverrides: { hold: true } });
  await connector.start();
  try {
    const r1 = await connector.handleMessage(message({ id: "300000000000000046" }));
    const r2 = await connector.handleMessage(message({ id: "300000000000000047" }));
    const r3 = await connector.handleMessage(message({ id: "300000000000000048" }));
    assert.equal(r1.accepted, true);
    assert.equal(r2.accepted, true);
    assert.equal(r3.reason, "ceiling");
    assert.equal(listTurns(journalDir).length, 0, "nothing written yet: the refusal came from the in-flight count");
    assert.equal(engine.prompts.length, 2);
    engine.release();
    assert.equal(await r1.turn, "ok");
    assert.equal(await r2.turn, "ok");
    assert.equal(listTurns(journalDir).length, 2);
    const r4 = await connector.handleMessage(message({ id: "300000000000000049" }));
    assert.equal(r4.reason, "ceiling");
  } finally {
    engine.release();
    await connector.stop();
  }
});

test("ceiling: a turn interrupted by a crash still counts after restart; admissions are durable", async () => {
  const now = () => Date.parse("2026-09-13T12:00:00Z");
  const root = makeRoot();
  const journalDir = join(root, "journal");
  ensureJournal(journalDir);
  const bind = binding({ limits: { turnsPerDay: 1 } });
  const held = fakeEngine({ hold: true });
  const first = createConnector({ binding: bind, journalDir, rest: fakeRest(), gateway: fakeGateway(), engine: held, now });
  await first.start();
  const rest = fakeRest();
  const second = createConnector({ binding: bind, journalDir, rest, gateway: fakeGateway(), engine: fakeEngine(), now });
  let r1;
  try {
    r1 = await first.handleMessage(message({ id: "300000000000000061" }));
    assert.equal(r1.accepted, true);
    assert.equal(listTurns(journalDir).length, 0, "the process dies here, mid-turn: no turn record");
    assert.equal(countAdmissionsOn(journalDir, "2026-09-13"), 1);
    // "Restart": a fresh connector over the same journal, the old one never finishing.
    await second.start();
    const r2 = await second.handleMessage(message({ id: "300000000000000062" }));
    assert.equal(r2.reason, "ceiling");
    assert.equal(rest.calls.filter((c) => c.content === FIXED_LINES.ceiling).length, 1);
    assert.equal(listTurns(journalDir).length, 0);
  } finally {
    held.release();
    if (r1 && r1.turn) await r1.turn.catch(() => {});
    await second.stop();
    await first.stop();
  }
});

test("ceiling: the daily notice survives a same-day restart; one delivery attempt in total, even when the first attempt crashed mid-flight", async () => {
  const now = () => Date.parse("2026-09-13T12:00:00Z");
  const root = makeRoot();
  const journalDir = join(root, "journal");
  ensureJournal(journalDir);
  const bind = binding({ limits: { turnsPerDay: 1 } });
  const restA = fakeRest();
  const first = createConnector({ binding: bind, journalDir, rest: restA, gateway: fakeGateway(), engine: fakeEngine(), now });
  await first.start();
  await (await first.handleMessage(message({ id: "300000000000000071" }))).turn;
  const r2 = await first.handleMessage(message({ id: "300000000000000072" }));
  assert.equal(r2.reason, "ceiling");
  assert.equal(restA.calls.filter((c) => c.content === FIXED_LINES.ceiling).length, 1);
  assert.equal(noticeOn(journalDir, "ceiling", "2026-09-13"), true, "the notice decision is journaled");
  await first.stop();
  // Same UTC day, new process, same journal: no second notice.
  const restB = fakeRest();
  const second = createConnector({ binding: bind, journalDir, rest: restB, gateway: fakeGateway(), engine: fakeEngine(), now });
  await second.start();
  const r3 = await second.handleMessage(message({ id: "300000000000000073" }));
  assert.equal(r3.reason, "ceiling");
  assert.equal(restB.calls.length, 0, "no delivery attempt at all after restart");
  await second.stop();
  // The decision is recorded before the attempt: a crash during delivery leaves the record.
  const crashDir = join(root, "journal-crash");
  ensureJournal(crashDir);
  const restC = fakeRest();
  restC.createMessage = async () => { throw new Error("process died mid-delivery"); };
  const third = createConnector({ binding: bind, journalDir: crashDir, rest: restC, gateway: fakeGateway(), engine: fakeEngine(), now });
  await third.start();
  await (await third.handleMessage(message({ id: "300000000000000074" }))).turn.catch(() => {});
  await third.handleMessage(message({ id: "300000000000000075" })).catch(() => {});
  assert.equal(noticeOn(crashDir, "ceiling", "2026-09-13"), true, "recorded before the attempt");
  await third.stop();
});

test("duplicate: the same event delivered twice while the thread lookup is held yields one prompt, one admission and one reply", async () => {
  const now = () => Date.parse("2026-09-13T12:00:00Z");
  const { rest, engine, connector, journalDir } = setup({ replies: [{ text: "once" }], now });
  rest.channels.set(IDS.threadOfAdmin, { id: IDS.threadOfAdmin, type: 11, parent_id: IDS.admin, name: "a-thread", guild_id: IDS.guild });
  let release;
  rest.holdLookup = new Promise((resolve) => { release = resolve; });
  await connector.start();
  const msg = message({ id: "300000000000000081", channel_id: IDS.threadOfAdmin });
  const p1 = connector.handleMessage(msg);
  const p2 = connector.handleMessage({ ...msg });
  assert.equal(rest.lookups, 1, "the first call is parked in the lookup");
  release();
  const [r1, r2] = await Promise.all([p1, p2]);
  const accepted = [r1, r2].filter((r) => r.accepted);
  const dups = [r1, r2].filter((r) => r.reason === "duplicate");
  assert.equal(accepted.length, 1);
  assert.equal(dups.length, 1);
  assert.equal(await accepted[0].turn, "ok");
  assert.equal(rest.lookups, 1, "the duplicate never triggered a second lookup");
  assert.equal(engine.prompts.length, 1);
  assert.equal(countAdmissionsOn(journalDir, "2026-09-13"), 1);
  assert.equal(rest.calls.length, 1, "one reply");
  assert.equal(readInboxIds(journalDir).size, 1);
  assert.equal(listTurns(journalDir).length, 1);
  await connector.stop();
});

test("journal: no token-shaped string and no model output on the drop path reaches disk", async () => {
  const { journalDir, connector } = setup();
  await connector.start();
  await connector.handleMessage(message({ id: "300000000000000051", author: { id: IDS.stranger }, content: "MTAw.secret-looking-token-value-1234567890" }));
  for (const f of ["drops.jsonl", "inbox.jsonl", "outbox.jsonl"]) {
    const p = join(journalDir, f);
    if (existsSync(p)) assert.ok(!readFileSync(p, "utf8").includes("secret-looking"), f);
  }
  await connector.stop();
});