import test from "node:test";
import assert from "node:assert/strict";
import { defaultSiteSettings, type SiteSettings } from "../lib/settings";
import { defaultChannelPriority, compileLaneRoot } from "../lib/channelPriority";
import { LANES, type AutoQueueKind } from "../lib/autoQueueTypes";
import { emptyAutoQueueState } from "../jobs/autoQueueState";
import type { AutoRunnerStatus, LeafPending } from "../controller/autoRunner";
import { buildAutoQueueLanes } from "./autoQueueLanes";
import type { PriorityView } from "./channelPriority";
import {
  AUTO_QUEUE_STATUS_MEMO_MS,
  buildAutoQueueStatusPayload,
  singleFlightMemo,
  type AutoQueueStatusInputs,
} from "./autoQueueStatus";

// THE FOUR LANES' CONSOLE, as a fold over state that is handed to it. Every
// reader this used to do itself — the auto-queue document, the pending counts,
// the runner status, the clock — is an argument now, which is what makes the
// three rules below testable at all.

const runnerFor = (kind: AutoQueueKind): AutoRunnerStatus => ({
  kind,
  running: false,
  jobId: null,
  startedAt: null,
  inFlight: [],
  activeByNode: {},
  idleReason: "stopped",
});

const pendingFor = (counts: Record<string, number> = {}): LeafPending => ({
  counts,
  head: {},
  owner: {},
  recency: {},
  nextUp: null,
});

const byLane = <T,>(make: (kind: AutoQueueKind) => T): Record<AutoQueueKind, T> =>
  Object.fromEntries(LANES.map((l) => [l, make(l)])) as Record<AutoQueueKind, T>;

const NOW = 1_700_000_000_000;

function inputs(over: Partial<AutoQueueStatusInputs> = {}): AutoQueueStatusInputs {
  const settings = defaultSiteSettings();
  const priority: PriorityView = {
    model: defaultChannelPriority(),
    compiled: false,
    slugs: [],
    focusSlugs: [],
    name: null,
  };
  return {
    settings,
    pool: {
      summary: () => [],
      isPaused: () => false,
      canStopPartial: () => false,
    },
    now: () => NOW,
    priority,
    runner: byLane(runnerFor),
    state: emptyAutoQueueState(),
    pending: byLane(() => pendingFor()),
    lanes: buildAutoQueueLanes({ settings, briefs: [], workerSummary: [] }),
    ...over,
  };
}

test("every lane in LANES gets an entry, and the rail rides along", () => {
  const i = inputs();
  const payload = buildAutoQueueStatusPayload(i);
  for (const lane of LANES) assert.equal(payload[lane].kind, lane);
  assert.equal(payload.lanes, i.lanes);
});

test("a cooldown is filtered by the injected clock, newest first", () => {
  const state = emptyAutoQueueState();
  state.download.platformBackoff = {
    // Lapsed a second ago: the operator is not waiting on it and it must not
    // show as a reason the lane is quiet.
    youtube: { until: NOW - 1_000, fails: 9 },
    rumble: { until: NOW + 60_000, fails: 2 },
    twitch: { until: NOW + 600_000, fails: 5 },
    // The boundary is strictly greater-than: a cooldown lapsing exactly now has
    // lapsed.
    kick: { until: NOW, fails: 1 },
  };
  const payload = buildAutoQueueStatusPayload(inputs({ state }));
  assert.deepEqual(payload.download.cooldowns, [
    { platform: "twitch", untilMs: NOW + 600_000, fails: 5, hold: null },
    { platform: "rumble", untilMs: NOW + 60_000, fails: 2, hold: null },
  ]);
  // The clock is a function, so the same state read later says something else.
  assert.deepEqual(
    buildAutoQueueStatusPayload(inputs({ state, now: () => NOW + 120_000 }))
      .download.cooldowns,
    [{ platform: "twitch", untilMs: NOW + 600_000, fails: 5, hold: null }],
  );
});

test("a hold, the pace and the subtitle deferrals reach the download lane (release 17, RL)", () => {
  const state = emptyAutoQueueState();
  state.download.platformBackoff = { youtube: { until: NOW + 3_600_000, fails: 8 } };
  state.download.platformHolds = { youtube: { since: NOW - 60_000, probeAt: NOW + 3_600_000, rateLimited: true } };
  state.download.platformPace = {
    youtube: { sleepRequestsSeconds: 4, baseSeconds: 1, cleanUnits: 0, steppedAt: NOW },
  };
  state.download.subtitleDeferrals = {
    b: { count: 3, lastAt: NOW - 1, until: NOW + 7 * 86_400_000, channelSlug: "alpha" },
    a: { count: 1, lastAt: NOW - 1, until: NOW + 3_600_000, channelSlug: "alpha" },
    gone: { count: 1, lastAt: NOW - 10, until: NOW - 1, channelSlug: "alpha" },
  };
  const d = buildAutoQueueStatusPayload(inputs({ state })).download;
  assert.deepEqual(d.cooldowns, [
    { platform: "youtube", untilMs: NOW + 3_600_000, fails: 8, hold: { sinceMs: NOW - 60_000, rateLimited: true } },
  ]);
  assert.deepEqual(d.pace, [{ platform: "youtube", sleepRequestsSeconds: 4, baseSeconds: 1 }]);
  assert.deepEqual(
    d.subtitleDeferred.map((x) => [x.videoId, x.count, x.held]),
    [
      ["a", 1, false],
      ["b", 3, true],
    ],
  );
});

test("deferred videos: live only, soonest first, ties by id, clock-driven", () => {
  const state = emptyAutoQueueState();
  // Inserted out of order on purpose; the strip must not depend on key order.
  state.download.videoDeferrals = {
    late: { until: NOW + 6 * 3_600_000, channelSlug: "beta" },
    tieB: { until: NOW + 60_000, channelSlug: "alpha" },
    lapsed: { until: NOW - 1_000, channelSlug: "alpha" },
    // Strictly greater-than, like the cooldowns: lapsing exactly now has lapsed.
    boundary: { until: NOW, channelSlug: "alpha" },
    tieA: { until: NOW + 60_000, channelSlug: "gamma" },
  };
  const payload = buildAutoQueueStatusPayload(inputs({ state }));
  assert.deepEqual(payload.download.deferred, [
    { videoId: "tieA", channelSlug: "gamma", untilMs: NOW + 60_000 },
    { videoId: "tieB", channelSlug: "alpha", untilMs: NOW + 60_000 },
    { videoId: "late", channelSlug: "beta", untilMs: NOW + 6 * 3_600_000 },
  ]);
  assert.deepEqual(payload.transcription.deferred, []);
  // Past the tied pair, only the 6 h one is left.
  assert.deepEqual(
    buildAutoQueueStatusPayload(inputs({ state, now: () => NOW + 60_000 }))
      .download.deferred,
    [{ videoId: "late", channelSlug: "beta", untilMs: NOW + 6 * 3_600_000 }],
  );
});

test("transcription's hold is the pool's, every other lane's is the gate", () => {
  // The asymmetry is deliberate and is the reason `pool` is on these inputs at
  // all: transcription's hold is LIVE on the worker pool, while every other
  // lane's stored flag IS its gate.
  const paused = buildAutoQueueStatusPayload(
    inputs({
      pool: {
        summary: () => [],
        isPaused: () => true,
        canStopPartial: () => false,
      },
    }),
  );
  assert.equal(paused.transcription.held, true);
  assert.equal(paused.download.held, false);
  assert.equal(paused.digest.held, false);

  const settings: SiteSettings = defaultSiteSettings();
  settings.autoQueue = {
    ...settings.autoQueue,
    download: { ...settings.autoQueue.download, held: true },
    transcription: { ...settings.autoQueue.transcription, held: true },
  };
  const gated = buildAutoQueueStatusPayload(inputs({ settings }));
  assert.equal(gated.download.held, true);
  // Settings say held; the live pool says otherwise, and the pool wins.
  assert.equal(gated.transcription.held, false);
});

test("policy.root is the tree the lane DISPATCHES from, stored or compiled", () => {
  const stored = buildAutoQueueStatusPayload(inputs());
  const settings = defaultSiteSettings();
  assert.equal(stored.digest.policyCompiled, false);
  assert.deepEqual(stored.digest.policy.root, settings.autoQueue.digest.root);
  // Everything else on the policy comes through the spread untouched.
  assert.equal(stored.digest.policy.enabled, settings.autoQueue.digest.enabled);
  assert.equal(stored.digest.policy.order, settings.autoQueue.digest.order);

  const model = defaultChannelPriority();
  model.channels = { alpha: { tier: "low" } };
  const priority: PriorityView = {
    model,
    compiled: true,
    slugs: ["alpha", "beta"],
    focusSlugs: ["beta"],
    name: "beta",
  };
  const compiled = buildAutoQueueStatusPayload(inputs({ priority }));
  assert.equal(compiled.digest.policyCompiled, true);
  assert.deepEqual(
    compiled.digest.policy.root,
    compileLaneRoot("digest", model, priority.slugs, priority.focusSlugs),
  );
  assert.notDeepEqual(compiled.digest.policy.root, settings.autoQueue.digest.root);
  assert.equal(compiled.digest.focusName, "beta");
});

test("the pending fold is per lane and reaches the focus banner", () => {
  const pending = byLane(() => pendingFor());
  pending.digest = pendingFor({ "prio-focus-beta": 3, "prio-normal": 7 });
  const model = defaultChannelPriority();
  const priority: PriorityView = {
    model: { ...model, focus: { kind: "channels", slugs: ["beta"] } },
    compiled: true,
    slugs: ["alpha", "beta"],
    focusSlugs: ["beta"],
    name: "beta",
  };
  const payload = buildAutoQueueStatusPayload(inputs({ pending, priority }));
  assert.deepEqual(payload.digest.pendingByLeaf, {
    "prio-focus-beta": 3,
    "prio-normal": 7,
  });
  assert.equal(payload.digest.focus.active, true);
  assert.equal(payload.digest.focus.focusPending, 3);
  assert.equal(payload.digest.focus.otherPending, 7);
  // A lane with nothing pending is still an entry, with the same shape.
  assert.deepEqual(payload.download.pendingByLeaf, {});
});

// THE POLL'S MEMO (slice D0, release 17): N pollers of the 3 s poll cost one
// fold of the snapshots, concurrent callers share the one in flight, and time
// is the only thing that expires it.

function deferred<T>() {
  let resolve!: (v: T) => void;
  let reject!: (e: unknown) => void;
  const promise = new Promise<T>((res, rej) => {
    resolve = res;
    reject = rej;
  });
  return { promise, resolve, reject };
}

test("memo: concurrent callers share the computation in flight", async () => {
  const memo = singleFlightMemo<number>({ now: () => 0 });
  const d = deferred<number>();
  let calls = 0;
  const compute = () => {
    calls++;
    return d.promise;
  };
  const all = Promise.all([memo.get("k", compute), memo.get("k", compute), memo.get("k", compute)]);
  d.resolve(42);
  assert.deepEqual(await all, [42, 42, 42]);
  assert.equal(calls, 1);
});

test("memo: a landed value is reused for the window, then recomputed", async () => {
  let t = 1_000;
  const memo = singleFlightMemo<number>({ now: () => t });
  let calls = 0;
  const compute = async () => ++calls;
  assert.equal(await memo.get("k", compute), 1);
  t += AUTO_QUEUE_STATUS_MEMO_MS - 1;
  assert.equal(await memo.get("k", compute), 1, "inside the window: the memo");
  t += 1;
  assert.equal(await memo.get("k", compute), 2, "at the window's end: computed again");
  assert.equal(calls, 2);
});

test("memo: the window runs from when the value LANDED, not when it started", async () => {
  let t = 0;
  const memo = singleFlightMemo<number>({ ttlMs: 3_000, now: () => t });
  const d = deferred<number>();
  const first = memo.get("k", () => d.promise);
  t = 10_000; // a slow fold: ten seconds
  d.resolve(7);
  assert.equal(await first, 7);
  t = 12_000;
  assert.equal(await memo.get("k", async () => 8), 7);
});

test("memo: a rejection is shared by its waiters and never memoized", async () => {
  const memo = singleFlightMemo<number>({ now: () => 0 });
  const d = deferred<number>();
  const a = memo.get("k", () => d.promise);
  const b = memo.get("k", async () => 99);
  d.reject(new Error("drive not answering"));
  await assert.rejects(a, /drive not answering/);
  await assert.rejects(b, /drive not answering/);
  assert.equal(await memo.get("k", async () => 5), 5);
});

test("memo: clear() drops the value and detaches the computation in flight", async () => {
  const memo = singleFlightMemo<string>({ now: () => 0 });
  assert.equal(await memo.get("k", async () => "old fixture"), "old fixture");
  memo.clear();
  const slow = deferred<string>();
  const before = memo.get("k", () => slow.promise);
  memo.clear();
  // A caller after the clear does not join the detached computation…
  assert.equal(await memo.get("k", async () => "new fixture"), "new fixture");
  // …and when it lands, it neither answers the memo nor evicts the new value.
  slow.resolve("stale");
  assert.equal(await before, "stale");
  assert.equal(await memo.get("k", async () => "unused"), "new fixture");
});

test("memo: a different key misses the memo and the computation in flight", async () => {
  const memo = singleFlightMemo<string>({ now: () => 0 });
  assert.equal(await memo.get("tree A", async () => "counts under A"), "counts under A");
  // A rule added, a focus set: the settings the fold reads changed.
  assert.equal(await memo.get("tree B", async () => "counts under B"), "counts under B");
  assert.equal(await memo.get("tree B", async () => "unused"), "counts under B");
  // Asking under A again is a miss too — the stored value is B's.
  assert.equal(await memo.get("tree A", async () => "A again"), "A again");

  // In flight: a caller with another key does not join it, and the older
  // computation landing late does not overwrite the newer key's value.
  const slowA = deferred<string>();
  const a = memo.get("tree A2", () => slowA.promise);
  assert.equal(await memo.get("tree C", async () => "C"), "C");
  slowA.resolve("late A2");
  assert.equal(await a, "late A2");
  assert.equal(await memo.get("tree C", async () => "unused"), "C");
});

test("a held platform stays listed after its probe time passed (review H2)", () => {
  const state = emptyAutoQueueState();
  state.download.platformBackoff = { youtube: { until: NOW - 5_000, fails: 9 }, rumble: { until: NOW - 5_000, fails: 2 } };
  state.download.platformHolds = { youtube: { since: NOW - 7_200_000, probeAt: NOW - 5_000, rateLimited: false } };
  const d = buildAutoQueueStatusPayload(inputs({ state })).download;
  assert.deepEqual(d.cooldowns, [
    { platform: "youtube", untilMs: NOW - 5_000, fails: 9, hold: { sinceMs: NOW - 7_200_000, rateLimited: false } },
  ]);
});