import { test } from "node:test";
import assert from "node:assert/strict";
import {
  channelsForOperation,
  compileLaneRoot,
  focusSummary,
  isChannelPaused,
  prioLeafId,
  sanitizeChannelPriority,
} from "../lib/channelPriority";
import type { ChannelPriority } from "../lib/channelPriority";
import {
  type AutoQueueGroup,
  type ChannelWork,
  buildPendingByLeaf,
  emptyAutoQueueRuntime,
  selectNextWork,
} from "./autoQueuePolicy";

// THE DESIGN, ASSERTED THROUGH THE REAL ENGINE.
//
// `plans/channel-priority.md`'s decision is COMPILE, not consult: the model
// becomes an ordinary `AutoQueueGroup` and dispatch runs the engine it always
// ran. So the claim "a focus holds the rest until its work is done" is not a
// claim about `channelPriority.ts` at all — it is a claim about what
// `buildPendingByLeaf` + `selectNextWork` do to a COMPILED tree, and it can
// only be tested by driving both.
//
// IN jobs/, NOT lib/, for the reason `laneMigration.test.ts` is: this file must
// import `jobs/autoQueuePolicy`, and `../architecture.test.ts` forbids
// `lib/ -> jobs/` from a test file like any other. It must also stay out of the
// controller layer's reach — `jobs/ -> controller/` is forbidden too — which is
// why the runner-side halves (the `listChannelMeta` filter as the runner calls
// it, and the once-per-transition log line) are asserted in
// `controller/autoRunner.test.ts` instead.
//
// The sanitizer round-trip half lives in `channelPrioritySanitize.test.ts` (S0).

// One bucket, spelled once. Which bucket a lane draws is `defaultDrawsForPolicy`'s
// business and autoQueuePolicy.test.ts's; nothing here depends on the name.
const BUCKET = "downloadedNoTranscript";

function work(slug: string, ids: string[]): ChannelWork {
  return { slug, platform: "youtube", buckets: { [BUCKET]: [...ids] } };
}

function model(value: unknown): ChannelPriority {
  return sanitizeChannelPriority(value);
}

// EXACTLY WHAT THE RUNNER DOES, in the runner's order: `listChannelMeta` filters
// the channel list by the lane's effective tier, then `laneDispatchRoot`
// compiles the root from what survived. Both halves are here so a test that
// says "paused is never drawn" is testing the pair the runner actually runs,
// catch-all included.
function dispatch(
  lane: "transcription" | "download" | "digest" | "backfill",
  priority: ChannelPriority,
  channels: ChannelWork[],
  focusSlugs: readonly string[] = [],
): { root: AutoQueueGroup; channels: ChannelWork[] } {
  const slugs = channelsForOperation(
    priority,
    channels.map((c) => c.slug),
    lane,
  );
  const live = channels.filter((c) => slugs.includes(c.slug));
  return {
    root: compileLaneRoot(lane, priority, slugs, focusSlugs),
    channels: live,
  };
}

// Drain a compiled tree the way the runner does: re-ask the policy after every
// grant, consuming the chosen id. Returns the owning channel of each pick, in
// served order — the focus question is "whose video came next", not "which
// leaf".
function drainOwners(
  root: AutoQueueGroup,
  pending: Record<string, string[]>,
  max = 100,
): string[] {
  const runtime = emptyAutoQueueRuntime();
  const owners: string[] = [];
  for (let i = 0; i < max; i++) {
    const pick = selectNextWork(root, pending, runtime);
    if (!pick) break;
    owners.push(pick.leafId);
    const ids = pending[pick.leafId];
    assert.equal(ids[0], pick.videoId, "pick should be head of leaf queue");
    ids.shift();
  }
  return owners;
}

test("a focus holds a non-focus channel with work, then releases it when the focus is exhausted", () => {
  const priority = model({ focus: { kind: "channels", slugs: ["slow-a"] } });
  const channels = [work("slow-a", ["a1", "a2"]), work("slow-b", ["b1", "b2"])];
  const { root, channels: live } = dispatch(
    "transcription",
    priority,
    channels,
    ["slow-a"],
  );
  const pending = buildPendingByLeaf(root, live, [BUCKET]);

  // Both channels have work RIGHT NOW — the hold is not "b has nothing".
  assert.deepEqual(pending[prioLeafId("focus", "slow-a")], ["a1", "a2"]);
  assert.deepEqual(pending[prioLeafId("normal", "slow-b")], ["b1", "b2"]);

  // Strict descent serves every one of A's before it ever reaches B's group.
  assert.deepEqual(drainOwners(root, pending), [
    prioLeafId("focus", "slow-a"),
    prioLeafId("focus", "slow-a"),
    prioLeafId("normal", "slow-b"),
    prioLeafId("normal", "slow-b"),
  ]);
});

test("new focus work retakes the lane on the very next pick", () => {
  const priority = model({ focus: { kind: "channels", slugs: ["slow-a"] } });
  const channels = [work("slow-a", []), work("slow-b", ["b1", "b2"])];
  const { root, channels: live } = dispatch(
    "transcription",
    priority,
    channels,
    ["slow-a"],
  );
  const pending = buildPendingByLeaf(root, live, [BUCKET]);
  const runtime = emptyAutoQueueRuntime();

  // Mid-"slow-b batch": the focus group is empty, so strict descent skips it.
  let pick = selectNextWork(root, pending, runtime);
  assert.equal(pick?.leafId, prioLeafId("normal", "slow-b"));
  pending[pick!.leafId].shift();

  // A snapshot regen gives the focus channel a video. The runner rebuilds
  // `pending` every tick and re-asks, so the very next pick is the focus's —
  // no second mechanism, no preemption of the unit already running.
  pending[prioLeafId("focus", "slow-a")].push("a1");
  pick = selectNextWork(root, pending, runtime);
  assert.equal(pick?.leafId, prioLeafId("focus", "slow-a"));
});

test("a focus that resolves to nothing compiles no focus group and holds no one", () => {
  // An unknown siteId is the live case: `resolveFocusSlugs` answers [], so a
  // typo must leave the tree exactly as it would be with no focus at all.
  const priority = model({ focus: { kind: "site", siteId: "no-such-site" } });
  const channels = [work("slow-a", ["a1"]), work("slow-b", ["b1"])];
  const { root, channels: live } = dispatch(
    "transcription",
    priority,
    channels,
    [],
  );
  assert.equal(
    root.children.some((c) => c.id === "prio-focus"),
    false,
  );
  const pending = buildPendingByLeaf(root, live, [BUCKET]);
  assert.deepEqual(drainOwners(root, pending).sort(), [
    prioLeafId("normal", "slow-a"),
    prioLeafId("normal", "slow-b"),
  ]);
});

test("a paused channel is drawn by no leaf, the catch-all included", () => {
  const priority = model({ channels: { gone: { tier: "paused" } } });
  const channels = [work("kept", ["k1"]), work("gone", ["g1", "g2"])];
  const { root, channels: live } = dispatch("download", priority, channels);

  // The FILTER is what does it: a tree cannot express exclusion, so the paused
  // channel never reaches the projection at all.
  assert.deepEqual(
    live.map((c) => c.slug),
    ["kept"],
  );
  const pending = buildPendingByLeaf(root, live, [BUCKET]);
  const everyId = Object.values(pending).flat();
  assert.deepEqual(everyId, ["k1"]);
  assert.deepEqual(pending["prio-all"], []);
  assert.equal(
    Object.keys(pending).includes(prioLeafId("paused", "gone")),
    false,
  );
});

test("sync-only pause leaves the download lane drawing the channel", () => {
  // What all 15 live `excludeFromSync` channels migrate to. The base tier and
  // the rank stand; only the `sync` operation loses the channel, so no lane's
  // membership moves — which is what makes that migration lossless.
  const priority = model({
    channels: {
      omnivods: { tier: "normal", rank: 7, overrides: { sync: "paused" } },
    },
  });
  assert.equal(isChannelPaused(priority, "omnivods", "sync"), true);
  assert.equal(isChannelPaused(priority, "omnivods", "download"), false);

  const channels = [work("omnivods", ["o1"])];
  const { root, channels: live } = dispatch("download", priority, channels);
  const pending = buildPendingByLeaf(root, live, [BUCKET]);
  assert.deepEqual(pending[prioLeafId("normal", "omnivods")], ["o1"]);
  assert.deepEqual(channelsForOperation(priority, ["omnivods"], "sync"), []);
});

test("a per-lane override pauses one lane and leaves the others drawing", () => {
  const priority = model({
    channels: { noisy: { tier: "normal", overrides: { download: "paused" } } },
  });
  const channels = [work("noisy", ["n1"])];

  const dl = dispatch("download", priority, channels);
  assert.deepEqual(dl.channels, []);
  assert.deepEqual(
    Object.values(buildPendingByLeaf(dl.root, dl.channels, [BUCKET])).flat(),
    [],
  );

  const tr = dispatch("transcription", priority, channels);
  const pending = buildPendingByLeaf(tr.root, tr.channels, [BUCKET]);
  assert.deepEqual(pending[prioLeafId("normal", "noisy")], ["n1"]);
});

test("a focused channel paused for that lane is not drawn by it", () => {
  // Focus wins over the stored tier; paused wins over focus, per operation.
  const priority = model({
    focus: { kind: "channels", slugs: ["star"] },
    channels: { star: { tier: "normal", overrides: { digest: "paused" } } },
  });
  const channels = [work("star", ["s1"]), work("other", ["o1"])];

  const digest = dispatch("digest", priority, channels, ["star"]);
  assert.deepEqual(
    digest.channels.map((c) => c.slug),
    ["other"],
  );
  assert.equal(
    digest.root.children.some((c) => c.id === "prio-focus"),
    false,
  );

  const backfill = dispatch("backfill", priority, channels, ["star"]);
  const pending = buildPendingByLeaf(backfill.root, backfill.channels, [
    BUCKET,
  ]);
  assert.deepEqual(pending[prioLeafId("focus", "star")], ["s1"]);
});

test("focusSummary reads the hold off the same pending map the pick used", () => {
  const priority = model({ focus: { kind: "channels", slugs: ["slow-a"] } });
  const channels = [work("slow-a", ["a1"]), work("slow-b", ["b1", "b2"])];
  const { root, channels: live } = dispatch(
    "digest",
    priority,
    channels,
    ["slow-a"],
  );
  const pending = buildPendingByLeaf(root, live, [BUCKET]);

  const holding = focusSummary(priority, ["slow-a"], pending);
  assert.equal(holding.active, true);
  assert.equal(holding.holding, true);
  assert.equal(holding.focusPending, 1);
  assert.equal(holding.otherPending, 2);
  assert.equal(holding.heldChannels, 1);

  // Drain the focus's only unit: the lane is released, and the summary says so
  // off the same map — no second read, no new clock.
  pending[prioLeafId("focus", "slow-a")].shift();
  const released = focusSummary(priority, ["slow-a"], pending);
  assert.equal(released.holding, false);
  assert.equal(released.heldChannels, 0);
  assert.equal(released.otherPending, 2);
});