import { test } from "node:test";
import assert from "node:assert/strict";
import { mkdtemp, readFile, rm } from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { getRegistry, type JobRecord } from "../jobs/registry";
import { getJobKind, jobKindLabel } from "../jobs/jobKinds";
import { readJobMeta } from "../jobs/jobMeta";
import { getPaths, type Paths } from "../lib/paths";
import {
  PUBLISH_QUEUE,
  enqueuePublishRun,
  enqueueStage,
  findQueuedStage,
  newPublishRunId,
  stageIdentity,
  stageRequestFromSpec,
  stageSpec,
} from "./publishStages";
import { STAGES, STAGE_KINDS, type StageRequest } from "./stages";
import type { PublishPlan } from "./publishPlan";

// Run with: pnpm --filter yt-dlp-transcript-common exec tsx --test publish/publishStages.test.ts
//
// Enqueueing publish stages (release 18): the spec is the request, a duplicate
// is refused with the existing id, and a run's preconditions ride on its specs.
// One test spawns the real stage child — with a request its own parser refuses
// (exit 2), so it touches nothing — to prove the job's shape end to end.

test("the seven stages' job kinds are jobKinds.ts's, with the stages' labels", () => {
  for (const kind of STAGE_KINDS) {
    const jobKind = STAGES[kind].jobKind;
    assert.ok(getJobKind(jobKind), `${jobKind} is registered`);
    assert.equal(jobKindLabel(jobKind), STAGES[kind].label, `${jobKind} label`);
    assert.equal(getJobKind(jobKind)?.replayable, true);
  }
});

test("a stage's spec is its request, and reads back to the same request", () => {
  const reqs: StageRequest[] = [
    { kind: "update-index", target: "_index", runId: "run-1" },
    { kind: "build-site", target: "jeralyzer", runId: "run-1", indexAfter: 123, force: true, skipArchives: true },
    { kind: "build-site", target: "_all", runId: "run-1", runner: "docker" },
    { kind: "deploy-site", target: "jeralyzer", runId: "run-1", preview: "r18", builtAfter: 456 },
    { kind: "deploy-site", target: "jeralyzer", runId: "run-1", to: "local" },
    { kind: "build-hub", target: "_hub", runId: "run-1", allowMissingMedia: true },
    { kind: "deploy-homepage", target: "_homepage", runId: "run-1", preview: "smoke" },
  ];
  for (const req of reqs) {
    const spec = stageSpec(req);
    assert.equal(spec.kind, `publish-${req.kind}`);
    assert.equal(spec.slug, req.target);
    assert.equal(spec.params?.runId, "run-1");
    assert.deepEqual(stageRequestFromSpec(JSON.parse(JSON.stringify(spec))), req);
  }
  // Not a stage, or a request the stage row would refuse.
  assert.equal(stageRequestFromSpec({ kind: "sync", slug: "x" }), null);
  assert.equal(stageRequestFromSpec({ kind: "publish-update-index", slug: "_index", params: {} }), null, "no run id");
  assert.equal(stageRequestFromSpec({ kind: "publish-update-index", slug: "jeralyzer", params: { runId: "r" } }), null);
  assert.equal(
    stageRequestFromSpec({ kind: "publish-build-site", slug: "a", params: { runId: "r", kind: "deploy-site" } }),
    null,
    "the params' kind must agree with the job kind",
  );
  assert.match(newPublishRunId("lane", 1_700_000_000_000), /^lane-[0-9a-z]{9}-[0-9a-f]{8}$/);
});

function rec(over: Partial<JobRecord>): JobRecord {
  return {
    id: "J",
    kind: "publish-build-site",
    queueKey: PUBLISH_QUEUE,
    status: "queued",
    queuedAt: 1,
    logPath: "/dev/null",
    ...over,
  };
}

test("a duplicate is the same kind + target (+ destination) queued or running on `publish`", () => {
  const build: StageRequest = { kind: "build-site", target: "alpha", runId: "r2" };
  const records = [rec({ id: "B1", spec: stageSpec({ ...build, runId: "r1" }) })];
  assert.equal(findQueuedStage(records, build)?.id, "B1", "another run's queued build is the same stage");
  assert.equal(findQueuedStage([{ ...records[0], status: "running" }], build)?.id, "B1");
  for (const other of [
    { ...records[0], status: "done" as const },
    { ...records[0], queueKey: "build" },
    { ...records[0], spec: stageSpec({ ...build, target: "beta" }) },
  ]) {
    assert.equal(findQueuedStage([other], build), undefined);
  }
  // A production deploy is not a preview deploy of the same site.
  const prod: StageRequest = { kind: "deploy-site", target: "alpha", runId: "r" };
  const prev: StageRequest = { ...prod, preview: "preview" };
  const deploys = [rec({ id: "D1", kind: "publish-deploy-site", spec: stageSpec(prev) })];
  assert.equal(findQueuedStage(deploys, prod), undefined);
  assert.equal(findQueuedStage(deploys, prev)?.id, "D1");
  assert.notEqual(stageIdentity(prod), stageIdentity(prev));
  const forced: StageRequest = { ...build, runId: "zzz", force: true };
  assert.equal(stageIdentity(build), stageIdentity(forced), "another run, forced: still the same stage");
});

test("enqueueStage refuses a duplicate with info and the existing job's id, spawning nothing", async () => {
  const registry = getRegistry();
  const existing = rec({
    id: "01EXISTINGPUBLISHSTAGE0001",
    kind: "publish-update-index",
    spec: stageSpec({ kind: "update-index", target: "_index", runId: "run-a" }),
  });
  registry.register(existing);
  try {
    const res = await enqueueStage(getPaths(), { kind: "update-index", target: "_index", runId: "run-b" });
    assert.equal(res.ok, false);
    assert.ok(!res.ok && res.info === true);
    assert.ok(!res.ok && res.jobId === existing.id);
    assert.match(!res.ok ? res.error : "", /Update the index _index is already queued/);
  } finally {
    existing.status = "cancelled";
  }
});

test("enqueuePublishRun enqueues the plan in order under one run id, carrying each step's preconditions", async () => {
  const plan: PublishPlan = {
    runStart: 1000,
    steps: [
      { kind: "update-index", target: "_index", reason: "stale" },
      { kind: "build-site", target: "alpha", indexAfter: 1000, reason: "1 channel changed" },
      { kind: "deploy-site", target: "alpha", preview: "preview", builtAfter: 1000, reason: "its build", previewUrl: "https://preview.alpha.pages.dev" },
    ],
    skipped: [{ kind: "build-site", target: "beta", reason: "stale, and its policy is off" }],
  };
  const seen: StageRequest[] = [];
  const res = await enqueuePublishRun(getPaths(), plan, {
    runId: "run-test",
    enqueue: async (_paths, req) => {
      seen.push(req);
      if (req.kind === "build-site") return { ok: false, info: true, jobId: "OLD", error: "already queued" };
      return {
        ok: true,
        jobId: `J-${req.kind}`,
        stream: new ReadableStream<string>(),
        done: new Promise(() => {}),
      };
    },
  });
  assert.deepEqual(
    seen.map((r) => [r.kind, r.target, r.runId, r.indexAfter, r.builtAfter, r.preview]),
    [
      ["update-index", "_index", "run-test", undefined, undefined, undefined],
      ["build-site", "alpha", "run-test", 1000, undefined, undefined],
      ["deploy-site", "alpha", "run-test", undefined, 1000, "preview"],
    ],
  );
  for (const r of seen) assert.equal("reason" in r, false, "a request carries no plan words");
  assert.equal(res.runId, "run-test");
  assert.deepEqual(res.jobs, [
    { kind: "update-index", target: "_index", jobId: "J-update-index" },
    { kind: "build-site", target: "alpha", jobId: "OLD", existing: true },
    { kind: "deploy-site", target: "alpha", jobId: "J-deploy-site", previewUrl: "https://preview.alpha.pages.dev" },
  ]);
  assert.deepEqual(res.skipped, plan.skipped);
});

test("a real stage job: runManagedCommand on the `publish` queue, the spec on its meta, the child's exit code kept", async () => {
  const jobsDir = await mkdtemp(path.join(os.tmpdir(), "publish-stage-job-"));
  try {
    const paths: Paths = { ...getPaths(), jobsDir };
    // An empty run id: the stage row refuses it (exit 2) before any lock or
    // file is touched.
    const req: StageRequest = { kind: "update-index", target: "_index", runId: "" };
    const res = await enqueueStage(paths, req, { background: true });
    assert.ok(res.ok, "enqueued");
    if (!res.ok) return;
    void res.stream.cancel();
    const live = getRegistry().get(res.jobId);
    assert.equal(live?.queueKey, PUBLISH_QUEUE);
    assert.equal(live?.kind, "publish-update-index");
    assert.equal(live?.background, true);
    assert.equal(live?.channelSlug, undefined);
    const term = await res.done;
    assert.equal(term.status, "failed");
    assert.equal(getRegistry().get(res.jobId)?.exitCode, 2);
    // The meta is written after the end; give the serial writer a moment.
    let meta = await readJobMeta(paths, res.jobId);
    for (let i = 0; i < 50 && meta?.status !== "failed"; i++) {
      await new Promise((r) => setTimeout(r, 20));
      meta = await readJobMeta(paths, res.jobId);
    }
    assert.equal(meta?.status, "failed");
    assert.deepEqual(meta?.spec, stageSpec(req));
    assert.match(await readFile(path.join(jobsDir, `${res.jobId}.log`), "utf8"), /--run-id is required/);
  } finally {
    await rm(jobsDir, { recursive: true, force: true });
  }
});