/**
 * lib/mailbox's concurrency, pinned.
 *
 * vectors/mailbox.json pins the BYTES this file puts on the wire. It said
 * nothing about time or re-entrancy, and that is the gap two production outages
 * came through: rooms would not open at all, and offline delivery had never once
 * worked, with a green 45-case and then 54-case suite either side of both fixes.
 *
 * vectors/mailbox-effects.json pins the other half — one interleaved transcript
 * of clock ops and side effects per scenario (helpers/mailbox-effects.mjs has
 * the rationale for each), compared with a single deepEqual. Exact counts
 * everywhere; nothing here says "at least one".
 */
import test from "node:test";
import assert from "node:assert/strict";
import { installDom, localStorageStub } from "./harness/dom.mjs";
import { mods, readVector, priv } from "./helpers/load.mjs";

installDom();
const M = await mods();
const { flushReentrancy, crashSafetyOrder, backoffLadder, replayThrottle } =
  await import("./helpers/mailbox-effects.mjs");

const V = readVector("mailbox-effects");
const env = { MailboxSync: M.MailboxSync, localStorageStub, priv };

/** first index of a transcript row matching `pred`, or -1 */
const at = (t, pred) => t.findIndex(pred);
const count = (t, pred) => t.filter(pred).length;
const isDeposit = (e) => e[0] === "client.deposit";

test("_flushing: a second enqueue in flight starts no second deposit", async () => {
  const got = await flushReentrancy(env);
  assert.deepEqual(got.transcript, V.flushReentrancy.transcript);

  // and, said again as invariants rather than as a snapshot:
  assert.equal(count(got.transcript, isDeposit), 2, "exactly two deposits, never three, never concurrent");
  const [d1, d2] = got.transcript.map((e, i) => [e, i]).filter(([e]) => isDeposit(e)).map(([, i]) => i);
  const releaseA = at(got.transcript, (e) => e[0] === "script" && e[1] === "release A");
  assert.ok(d1 < releaseA, "A's deposit was in flight before it was released");
  assert.ok(d2 > releaseA, "B's deposit only started after A's resolved");
  // the queue drains completely and strands no timer — the consequence of the
  // tail re-check at the bottom of flush!
  assert.equal(got.queued, 0);
  assert.deepEqual(got.pending, []);
  assert.deepEqual(got.sent, V.flushReentrancy.sent);
});

test("crash safety: add-retry < cursor advance < ingestEntry", async () => {
  const got = await crashSafetyOrder(env);
  assert.deepEqual(got.transcript, V.crashSafetyOrder.transcript);

  const t = got.transcript;
  const iRetry = at(t, (e) => e[0] === "storage.setItem" && e[1] === "K.retry");
  const iCursor = at(t, (e) => e[0] === "storage.setItem" && e[1] === "K.cursor");
  const iIngest = at(t, (e) => e[0] === "log.ingestEntry");
  const iPut = at(t, (e) => e[0] === "log.putEntryBlock");
  assert.notEqual(iPut, -1);
  assert.notEqual(iRetry, -1);
  assert.notEqual(iCursor, -1);
  assert.notEqual(iIngest, -1);
  // THE invariant lib/mailbox states in a comment and nothing pinned: the entry
  // is on the retry list from the moment its block is stored, so a crash
  // between store and join cannot strand it behind an advanced cursor.
  assert.ok(iPut < iRetry, "the block is stored before it is listed for retry");
  assert.ok(iRetry < iCursor, "add-retry precedes the cursor advance");
  assert.ok(iCursor < iIngest, "the cursor advance precedes the first join attempt");
  assert.deepEqual(got.retry, V.crashSafetyOrder.retry, "a deferred entry stays on the retry list");
  assert.equal(got.cursor, V.crashSafetyOrder.cursor);
  assert.deepEqual(got.pending, []);
});

test("backoff: 1000·2ⁿ capped at 300000, reset to 1000 after a success", async () => {
  const got = await backoffLadder(env);
  assert.deepEqual(got.transcript, V.backoffLadder.transcript);

  assert.deepEqual(got.delays, [1000, 2000, 4000, 8000, 16000, 32000, 64000, 128000, 256000, 300000, 300000]);
  assert.equal(got.attemptAfterSuccess, 0, "_attempt is reset by the first successful deposit");
  assert.equal(got.restartDelay, 1000, "...so the next failure restarts the ladder at 1000, not at the cap");
  // start() finds a live retry timer and must not arm a second one
  assert.deepEqual(got.beforeKick.map((p) => p.id), [got.clearedByKick]);
  // the wake kick clears it, and leaves exactly one fresh timer — no orphan
  assert.deepEqual(
    got.transcript.filter((e) => e[0] === "clearTimeout"),
    V.backoffLadder.transcript.filter((e) => e[0] === "clearTimeout"),
  );
  assert.ok(
    got.transcript.some((e) => e[0] === "clearTimeout" && e[1] === got.clearedByKick),
    "the live retry timer is cleared by the kick",
  );
  assert.equal(got.afterKick.length, 1);
  assert.notEqual(got.afterKick[0].id, got.clearedByKick, "and the cleared one is gone, not re-armed");
  assert.equal(got.afterKick[0].delay, 1000);
});

test("replay throttle: two inside the window replay once; force bypasses it", async () => {
  const got = await replayThrottle(env);
  assert.deepEqual(got.transcript, V.replayThrottle.transcript);

  const t = got.transcript;
  assert.equal(count(t, (e) => e[0] === "client.replay"), 3,
    "four _replay calls, three replays: the second is inside the 30s window");
  const label = (s) => at(t, (e) => e[0] === "script" && e[1] === s);
  const replaysBetween = (a, b) => t.slice(a, b).filter((e) => e[0] === "client.replay").length;
  assert.equal(replaysBetween(label("_replay(false) — no previous replay, so it runs"),
                              label("_replay(false) again, inside the 30s window")), 1);
  assert.equal(replaysBetween(label("_replay(false) again, inside the 30s window"),
                              label("_replay(true) — force bypasses the window")), 0);
  assert.equal(replaysBetween(label("_replay(true) — force bypasses the window"),
                              label("advance 30s")), 1);
  assert.equal(replaysBetween(label("_replay(false) — the window has passed"), t.length), 1);
  assert.deepEqual(got.pending, []);
});

test("the mailbox-effects vectors are not vacuous", () => {
  // Perturb each recorded transcript the way the corresponding defect would,
  // and assert the comparison notices. This proves deepEqual over these arrays
  // is SENSITIVE; the commit message records the matching source mutations,
  // which is what proves the code path is reached at all.
  const clone = (x) => JSON.parse(JSON.stringify(x));

  // a second concurrent deposit
  const twice = clone(V.flushReentrancy.transcript);
  twice.splice(at(twice, isDeposit) + 1, 0, ["client.deposit", 40]);
  assert.notDeepEqual(twice, V.flushReentrancy.transcript);

  // add-retry moved AFTER ingestEntry — the crash-safety defect exactly
  const t = clone(V.crashSafetyOrder.transcript);
  const iRetry = at(t, (e) => e[0] === "storage.setItem" && e[1] === "K.retry");
  const iIngest = at(t, (e) => e[0] === "log.ingestEntry");
  assert.ok(iRetry !== -1 && iIngest !== -1 && iRetry < iIngest);
  const [row] = t.splice(iRetry, 1);
  t.splice(at(t, (e) => e[0] === "log.ingestEntry") + 1, 0, row);
  assert.notDeepEqual(t, V.crashSafetyOrder.transcript);

  // a halved backoff step, and a deleted clearTimeout
  const halved = clone(V.backoffLadder.transcript);
  const iSet = at(halved, (e) => e[0] === "setTimeout" && e[1] === 4000);
  assert.notEqual(iSet, -1);
  halved[iSet][1] = 2000;
  assert.notDeepEqual(halved, V.backoffLadder.transcript);

  const noClear = clone(V.backoffLadder.transcript).filter((e) => e[0] !== "clearTimeout");
  assert.notEqual(noClear.length, V.backoffLadder.transcript.length);
  assert.notDeepEqual(noClear, V.backoffLadder.transcript);

  // a throttle that let the second call through
  const extra = clone(V.replayThrottle.transcript);
  extra.splice(at(extra, (e) => e[0] === "client.replay") + 1, 0, ["client.replay", null, 100]);
  assert.notDeepEqual(extra, V.replayThrottle.transcript);

  // ...and the recorded transcripts are real, not empty
  assert.equal(count(V.flushReentrancy.transcript, isDeposit), 2);
  assert.equal(count(V.backoffLadder.transcript, (e) => e[0] === "setTimeout"), 14);
  assert.equal(count(V.crashSafetyOrder.transcript, (e) => e[0] === "log.ingestEntry"), 2);
  assert.equal(count(V.replayThrottle.transcript, (e) => e[0] === "client.replay"), 3);
});