import { test } from "node:test";
import assert from "node:assert/strict";
import { newJobId, noteProgress, type JobRecord } from "../jobs/registry";
import type { JobListEntry } from "../jobs/listJobs";
import { createScheduler } from "../jobs/scheduler";
import {
  fromEntry,
  fromInFlight,
  fromRecord,
  fromSlot,
  isLive,
  mergeJobRows,
  orderLiveRows,
  reconcileSlots,
  type FromRecordContext,
} from "./jobRows";

// Run with: pnpm -C editor exec tsx --test "app/**/*.test.ts"
//
// The adapters, the order and the merge — no registry, no scheduler, no disk.
// Records are built the way common/jobs/registry.test.ts builds them: a literal
// JobRecord with a fresh id.

function rec(over: Partial<JobRecord> = {}): JobRecord {
  return {
    id: newJobId(),
    kind: "whisper-all",
    queueKey: "q",
    status: "running",
    queuedAt: 1_000,
    logPath: "",
    ...over,
  };
}

const ctx = (over: Partial<FromRecordContext> = {}): FromRecordContext => ({
  now: 10_000,
  workerNames: new Map(),
  position: -1,
  queueMax: 0,
  ...over,
});

test("fromRecord carries every field the card renderer used to read", () => {
  const j = rec({
    channelSlug: "a",
    videoId: "v1",
    startedAt: 2_000,
    background: true,
    tasks: [
      {
        id: "v1",
        label: "One",
        kind: "transcribe",
        startedAt: 3_000,
        workerId: "w1",
      },
    ],
  });
  const row = fromRecord(j, ctx({ workerNames: new Map([["w1", "Only"]]) }));
  assert.equal(row.id, j.id);
  assert.equal(row.kind, "whisper-all");
  assert.equal(row.status, "running");
  assert.equal(row.queueKey, "q");
  assert.equal(row.channelSlug, "a");
  assert.equal(row.videoId, "v1");
  assert.equal(row.background, true);
  assert.equal(row.draining, false);
  assert.equal(row.tasks?.[0].workerName, "Only");
  assert.equal(row.inRegistry, true);
  assert.equal(row.source, "registry");
  assert.equal(row.startedAt, 2_000);
  assert.equal(row.queuedAt, 1_000);
});

test("fromRecord: drainable needs running, a drainable kind and no drain", () => {
  assert.equal(fromRecord(rec(), ctx()).drainable, true);
  assert.equal(fromRecord(rec({ draining: true }), ctx()).drainable, false);
  assert.equal(fromRecord(rec({ status: "queued" }), ctx()).drainable, false);
  // check-availability has no jobKinds entry, so it is not drainable.
  assert.equal(
    fromRecord(rec({ kind: "check-availability" }), ctx()).drainable,
    false,
  );
});

test("fromRecord: reorder bounds come from position and queueMax", () => {
  const queued = (position: number, queueMax: number) =>
    fromRecord(rec({ status: "queued" }), ctx({ position, queueMax }));
  // Position 1 is the first QUEUED job (0 is the running head): nowhere up.
  assert.equal(queued(1, 3).canMoveUp, false);
  assert.equal(queued(2, 3).canMoveUp, true);
  assert.equal(queued(3, 3).canMoveDown, false);
  assert.equal(queued(2, 3).canMoveDown, true);
  // A parallel ("") job is in no queue at all.
  const parallel = fromRecord(rec({ status: "queued", queueKey: "" }), ctx());
  assert.equal(parallel.canMoveUp, false);
  assert.equal(parallel.canMoveDown, false);
  assert.equal(parallel.position, undefined);
});

test("fromRecord: no channel stat means no progress bar, and spec means replayable", () => {
  const j = rec({
    channelSlug: "a",
    progress: { metric: "transcripts", initial: 0, target: 10 },
  });
  assert.equal(fromRecord(j, ctx()).progress, undefined);
  const withStat = fromRecord(
    j,
    ctx({
      stat: {
        config: { name: "A" },
        transcriptCount: 5,
        downloadCount: 0,
      } as unknown as NonNullable<FromRecordContext["stat"]>,
    }),
  );
  assert.equal(withStat.progress?.current, 5);
  assert.equal(withStat.progress?.pct, 50);
  assert.equal(fromRecord(rec(), ctx()).replayable, false);
  assert.equal(
    fromRecord(rec({ spec: { kind: "whisper-all" } as never }), ctx())
      .replayable,
    true,
  );
});

test("fromEntry keeps the history fields, archived included", () => {
  const e: JobListEntry = {
    id: newJobId(),
    kind: "sync",
    channelSlug: "a",
    status: "archived",
    queuedAt: 1,
    endedAt: 9,
    inRegistry: false,
    replayable: true,
    logPath: "/x.log",
    logSize: 4096,
  };
  const row = fromEntry(e);
  assert.equal(row.status, "archived");
  assert.equal(row.logSize, 4096);
  assert.equal(row.inRegistry, false);
  assert.equal(row.replayable, true);
  assert.equal(row.source, "archive");
  // An old log with no sidecar has no kind at all; the cell renders "—".
  assert.equal(fromEntry({ ...e, kind: undefined }).kind, "");
  // No reason on the entry, no reason on the row (not even the key).
  assert.equal("cancelReason" in row, false);
});

test("fromEntry carries a cancelled job's cancelReason to the row", () => {
  const e: JobListEntry = {
    id: newJobId(),
    kind: "sync",
    channelSlug: "teamrcn",
    status: "cancelled",
    queuedAt: 1,
    endedAt: 9,
    inRegistry: false,
    replayable: true,
    logPath: "/x.log",
    logSize: 64,
    cancelReason: "server restarted; the scheduler re-derives syncs",
  };
  assert.equal(
    fromEntry(e).cancelReason,
    "server restarted; the scheduler re-derives syncs",
  );
});

test("orderLiveRows: running, then queued IN QUEUE ORDER, then what just ended", () => {
  const rows = orderLiveRows([
    { id: "q2", kind: "", status: "queued", queueKey: "q", position: 2, source: "registry" },
    { id: "run", kind: "", status: "running", queuedAt: 5, source: "registry" },
    { id: "q1", kind: "", status: "queued", queueKey: "q", position: 1, source: "registry" },
    { id: "done", kind: "", status: "done", endedAt: 99, source: "registry" },
  ]);
  assert.deepEqual(
    rows.map((r) => r.id),
    ["run", "q1", "q2", "done"],
  );
});

test("orderLiveRows: two queues interleave by name, then by position", () => {
  const rows = orderLiveRows([
    { id: "b1", kind: "", status: "queued", queueKey: "qB", position: 1, source: "registry" },
    { id: "a2", kind: "", status: "queued", queueKey: "qA", position: 2, source: "registry" },
    { id: "a1", kind: "", status: "queued", queueKey: "qA", position: 1, source: "registry" },
  ]);
  assert.deepEqual(
    rows.map((r) => r.id),
    ["a1", "a2", "b1"],
  );
});

test("orderLiveRows: two running jobs keep registry order (newest queued first)", () => {
  const rows = orderLiveRows([
    { id: "old", kind: "", status: "running", queuedAt: 1, source: "registry" },
    { id: "new", kind: "", status: "running", queuedAt: 2, source: "registry" },
  ]);
  assert.deepEqual(
    rows.map((r) => r.id),
    ["new", "old"],
  );
});

test("mergeJobRows: the live row wins by id, and the head stays first", () => {
  const live = [
    {
      id: "A",
      kind: "sync",
      status: "running" as const,
      progress: { metric: "transcripts" as const, initial: 0, current: 1, target: 2, pct: 50 },
      source: "registry" as const,
    },
    { id: "B", kind: "sync", status: "done" as const, endedAt: 5, source: "registry" as const },
  ];
  const history = [
    { id: "A", kind: "sync", status: "running" as const, logSize: 10, source: "archive" as const },
    { id: "B", kind: "sync", status: "done" as const, logSize: 10, source: "archive" as const },
    { id: "C", kind: "sync", status: "failed" as const, logSize: 10, source: "archive" as const },
  ];
  const rows = mergeJobRows(live, history);
  assert.deepEqual(
    rows.map((r) => r.id),
    ["A", "B", "C"],
  );
  // One <tr> per job, and the one that is drawn is the one with progress.
  assert.equal(rows[0].source, "registry");
  assert.equal(rows[0].progress?.pct, 50);
  // A recent row stays ahead of the history page that also lists it.
  assert.equal(rows[1].source, "registry");
});

test("isLive: anything non-terminal, plus a stuck or evicted slot", () => {
  assert.equal(isLive({ id: "1", kind: "", status: "running", source: "registry" }), true);
  assert.equal(isLive({ id: "1", kind: "", status: "queued", source: "registry" }), true);
  assert.equal(isLive({ id: "1", kind: "", status: "evicted", source: "slot" }), true);
  assert.equal(isLive({ id: "1", kind: "", status: "done", source: "archive" }), false);
  assert.equal(
    isLive({
      id: "1",
      kind: "",
      status: "done",
      stuck: { reason: "terminal-but-holding-slot", heldMs: 1 },
      source: "registry",
    }),
    true,
  );
});

// --- The scheduler against the registry -------------------------------------

function scheduledQueue(): {
  scheduler: ReturnType<typeof createScheduler>;
  ids: string[];
} {
  const scheduler = createScheduler();
  const ids = [newJobId(), newJobId(), newJobId()];
  for (const id of ids) {
    scheduler.submit({
      id,
      queueKey: "qFacts",
      tier: "foreground",
      concurrency: 1,
      start: () => {},
      onCancel: () => {},
    });
  }
  return { scheduler, ids };
}

test("reconcileSlots: a terminal record holding a running slot is hard-stuck", () => {
  const { scheduler, ids } = scheduledQueue();
  const [a, b, c] = ids;
  const now = 10_000_000;
  const records = new Map<string, JobRecord>([
    [a, rec({ id: a, queueKey: "qFacts", status: "done", startedAt: now - 5_000 })],
    [b, rec({ id: b, queueKey: "qFacts", status: "queued" })],
    [c, rec({ id: c, queueKey: "qFacts", status: "queued" })],
  ]);
  const facts = reconcileSlots(scheduler.queues(), (id) => records.get(id), now);
  const byId = new Map(facts.map((f) => [f.id, f]));
  assert.equal(byId.get(a)?.stuck?.reason, "terminal-but-holding-slot");
  assert.equal(byId.get(a)?.heal, true);
  assert.equal(byId.get(b)?.position, 1);
  assert.equal(byId.get(c)?.position, 2);
  assert.equal(byId.get(b)?.stuck, undefined);
  assert.equal(byId.get(c)?.stuck, undefined);
  assert.equal(facts.filter((f) => f.heal).length, 1);

  // Healing the hard cases promotes the next in line.
  facts.filter((f) => f.heal).forEach((f) => scheduler.complete(f.id));
  assert.deepEqual(scheduler.queues()[0].running, [b]);
});

test("reconcileSlots: an evicted record is hard-stuck; a quiet live one is soft", () => {
  const now = 10_000_000;

  const evicted = scheduledQueue();
  const gone = reconcileSlots(
    evicted.scheduler.queues(),
    () => undefined,
    now,
  );
  assert.equal(gone[0].stuck?.reason, "record-evicted");
  assert.equal(gone[0].heal, true);

  const stalled = scheduledQueue();
  const idA = stalled.ids[0];
  const soft = reconcileSlots(
    stalled.scheduler.queues(),
    (id) =>
      id === idA
        ? rec({
            id: idA,
            queueKey: "qFacts",
            status: "running",
            startedAt: now - 11 * 60 * 1000,
            tasks: [],
          })
        : undefined,
    now,
  );
  assert.equal(soft[0].stuck?.reason, "possibly-stalled");
  // SOFT: surfaced and force-releasable, never auto-healed.
  assert.equal(soft[0].heal, false);

  const busy = scheduledQueue();
  const idB = busy.ids[0];
  const fresh = reconcileSlots(
    busy.scheduler.queues(),
    (id) =>
      id === idB
        ? rec({
            id: idB,
            queueKey: "qFacts",
            status: "running",
            startedAt: now - 60 * 1000,
            tasks: [],
          })
        : undefined,
    now,
  );
  assert.equal(fresh[0].stuck, undefined);
  assert.equal(fresh[0].heal, false);
});

test("fromSlot builds the phantom, and a stuck row sorts above a running one", () => {
  const { scheduler } = scheduledQueue();
  const facts = reconcileSlots(scheduler.queues(), () => undefined, 10_000_000);
  const phantom = fromSlot(facts[0], "whisper-all");
  assert.equal(phantom.status, "evicted");
  assert.equal(phantom.source, "slot");
  assert.equal(phantom.stuck?.reason, "record-evicted");
  assert.equal(phantom.queueKey, "qFacts");

  const ordered = orderLiveRows([
    { id: "run", kind: "", status: "running", queuedAt: 9, source: "registry" },
    phantom,
  ]);
  assert.equal(ordered[0].id, phantom.id);
});

test("fromInFlight: a download unit carries its registry job id", () => {
  const row = fromInFlight(
    {
      videoId: "v1",
      leafId: "leaf-a",
      channelSlug: "alpha",
      startedAt: 1_000,
      jobId: "01JOBDOWNLOAD",
    },
    { kind: "download", runnerJobId: "01RUNNER", detail: "rule 1 · alpha" },
  );
  assert.equal(row.id, "01JOBDOWNLOAD");
  assert.equal(row.inRegistry, true);
  assert.equal(row.source, "runner");
  assert.equal(row.status, "running");
  assert.equal(row.background, true);
  assert.equal(row.kind, "download");
  assert.equal(row.channelSlug, "alpha");
  assert.equal(row.videoId, "v1");
  assert.equal(row.startedAt, 1_000);
  assert.equal(row.detail, "rule 1 · alpha");
});

test("fromInFlight: a task unit gets a synthetic id and no job page", () => {
  const unit = {
    videoId: "v2",
    leafId: "leaf-b",
    channelSlug: "beta",
    startedAt: 2_000,
  };
  const withRunner = fromInFlight(unit, {
    kind: "transcription",
    runnerJobId: "01RUNNER",
  });
  assert.equal(withRunner.id, "01RUNNER:v2");
  assert.equal(withRunner.inRegistry, false);
  // No runner job known (a lane between runs): the lane kind keys it.
  const bare = fromInFlight(unit, { kind: "transcription", runnerJobId: null });
  assert.equal(bare.id, "transcription:v2");
  assert.equal(bare.detail, undefined);
});

test("fromInFlight: a note replaces the detail and drops the synthetic video id", () => {
  const row = fromInFlight(
    {
      videoId: "metadata-scan alpha",
      leafId: "metadata-scan",
      channelSlug: "alpha",
      startedAt: 3_000,
      note: "metadata scan · 12 titles",
    },
    { kind: "download", detail: "rule 1 · alpha" },
  );
  assert.equal(row.detail, "metadata scan · 12 titles");
  assert.equal(row.videoId, undefined);
  assert.equal(row.id, "download:metadata-scan alpha");
});

// THE STALL LABEL MEASURES QUIET, NOT AGE (release 9, B4a). A metadata scan
// never has a task in flight and advances its count every few seconds; on
// 2026-09-25 one at ~10 videos/min read STUCK · POSSIBLY-STALLED after 10 min.
test("reconcileSlots: a long-running idle job whose progress advances is never stuck", () => {
  const { scheduler, ids } = scheduledQueue();
  const id = ids[0];
  const start = 10_000_000;
  const r = rec({
    id,
    queueKey: "qFacts",
    kind: "metadata-scan",
    status: "running",
    startedAt: start,
    tasks: [],
  });
  // One scanned video every 6 s for an hour, checked each minute.
  for (let t = 0; t <= 60 * 60 * 1000; t += 6_000) {
    noteProgress(
      r,
      { metric: "scans", initial: 0, target: 1_382, current: t / 6_000 },
      start + t,
    );
    if (t % 60_000 === 0) {
      const [fact] = reconcileSlots(scheduler.queues(), () => r, start + t);
      assert.equal(fact.stuck, undefined, `flagged at ${t / 60_000} min`);
    }
  }
});

test("reconcileSlots: idle tasks and progress frozen past 10 min is possibly-stalled", () => {
  const { scheduler, ids } = scheduledQueue();
  const id = ids[0];
  const start = 10_000_000;
  const r = rec({
    id,
    queueKey: "qFacts",
    status: "running",
    startedAt: start,
    tasks: [],
  });
  noteProgress(r, { metric: "scans", initial: 0, target: 100, current: 40 }, start + 20 * 60_000);
  // The same count re-reported is not a move.
  noteProgress(r, { metric: "scans", initial: 0, target: 100, current: 40 }, start + 25 * 60_000);
  assert.equal(r.progressAt, start + 20 * 60_000);
  const at = (min: number) =>
    reconcileSlots(scheduler.queues(), () => r, start + min * 60_000)[0];
  assert.equal(at(29).stuck, undefined);
  const late = at(31);
  assert.equal(late.stuck?.reason, "possibly-stalled");
  assert.equal(late.heal, false);
});