import assert from "node:assert/strict";
import { generateKeyPairSync, randomUUID } from "node:crypto";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import test from "node:test";
import { signCommand, type CommandBody } from "../bus/envelope";
import { DurableFence } from "../bus/fence";
import type { ObservedStore } from "../bus/redis";
import { MediaSupervisor, type RelayJob, type RetryScheduler } from "./index";
import type { DesiredChannel, ObservedChannel } from "../types";

const base: DesiredChannel = {
  workspaceId: "workspace-1", channelId: "channel-1", workerId: "worker-1",
  assignmentEpoch: 2, desiredVersion: 1, desiredStatus: "RUNNING",
  source: { kind: "hls", url: "https://mock.local/hls/live-1/master.m3u8" },
  output: { endpointUrl: "rtmp://mock.local/live", credentialHandleId: "keyref-1" },
  profile: "720p30", mode: "normalize",
};

const store: ObservedStore = {
  async get() { return null; },
  async put(_row: ObservedChannel) {},
  async heartbeat() {},
};

test("RUNNING → RUNNING on a swappable job is an in-place reconfigure, realized only when on air", async () => {
  const dir = mkdtempSync(join(tmpdir(), "media-reconf-"));
  const keys = generateKeyPairSync("ed25519");
  const calls: string[] = [];
  let accept = true;
  let failures = 0;
  const job = (): RelayJob => ({
    async start(c) { calls.push(`start v${c.desiredVersion}`); },
    async stop() { calls.push("stop"); },
    async reconfigure(c) {
      calls.push(`swap v${c.desiredVersion}`);
      if (failures > 0) { failures -= 1; throw new Error("compositor B never went on air"); }
      return accept;
    },
  });
  // Manual retry scheduler: the test fires retries itself.
  const pending: Array<{ action: () => Promise<void>; delayMs: number; cancelled: boolean }> = [];
  const scheduler: RetryScheduler = {
    schedule(action, delayMs) { const entry = { action, delayMs, cancelled: false }; pending.push(entry); return { cancel: () => { entry.cancelled = true; } }; },
  };
  const fire = async (): Promise<number> => { const entry = pending.shift(); assert.ok(entry && !entry.cancelled, "a retry is scheduled"); await entry.action(); return entry.delayMs; };
  const worker = new MediaSupervisor("worker-1", [base], store, new DurableFence(join(dir, "fence.json")),
    new Map([["control-v1", keys.publicKey]]), job, { transition() {} }, scheduler);
  const send = (version: number, status: "RUNNING" | "STOPPED" = "RUNNING") => {
    const body: CommandBody = {
      workspaceId: base.workspaceId, channelId: base.channelId, workerId: base.workerId, assignmentEpoch: base.assignmentEpoch,
      commandId: randomUUID(), desiredVersion: version, issuedAt: new Date(Date.now() - 1_000).toISOString(),
      expiresAt: new Date(Date.now() + 60_000).toISOString(), snapshot: { ...base, desiredVersion: version, desiredStatus: status },
    };
    return worker.handle(signCommand(body, "control-v1", keys.privateKey));
  };
  const status = () => worker.status()[0]!;
  try {
    await worker.start();
    await send(1);
    assert.deepEqual(calls, ["start v1"]);
    await send(2);
    assert.deepEqual(calls, ["start v1", "swap v2"], "a TAKE never stops the job (encoder/RTMP stay up)");
    assert.equal(status().state, "RUNNING");
    assert.equal(status().realizedVersion, 2);
    assert.match(status().reason, /frame boundary/);
    accept = false;
    await send(3);
    assert.deepEqual(calls.slice(2), ["swap v3", "stop", "start v3"], "an unswappable change falls back to restart");
    accept = true;

    // A failed swap keeps the old compositor on air: still RUNNING, realized = what viewers see, retried.
    failures = 1;
    await send(4);
    assert.deepEqual(calls.slice(5), ["swap v4"]);
    assert.equal(status().state, "RUNNING");
    assert.equal(status().realizedVersion, 3, "v3 is still on air");
    assert.match(status().reason, /swap to v4 failed .*v3 stays on air; retry 1\/2/);
    assert.equal(await fire(), 1_000);
    assert.deepEqual(calls.slice(5), ["swap v4", "swap v4"], "the retry swaps; nothing restarts");
    assert.equal(status().realizedVersion, 4);

    // A newer command supersedes a pending retry.
    failures = 1;
    await send(5);
    await send(6);
    assert.deepEqual(calls.slice(7), ["swap v5", "swap v6"]);
    assert.equal(pending.shift()?.cancelled, true, "the v5 retry was cancelled by v6");
    assert.equal(status().realizedVersion, 6);

    // Out of retries: the full restart converges (and is the only path that drops the push session).
    failures = 3;
    await send(7);
    assert.equal(await fire(), 1_000);
    assert.equal(await fire(), 2_000);
    assert.deepEqual(calls.slice(9), ["swap v7", "swap v7", "swap v7", "stop", "start v7"]);
    assert.equal(status().realizedVersion, 7);
    assert.equal(status().state, "RUNNING");

    await send(8, "STOPPED");
    assert.deepEqual(calls.slice(14), ["stop"]);
    assert.equal(pending.length, 0);
    await worker.stop();
  } finally { rmSync(dir, { recursive: true, force: true }); }
});

test("worker stop waits for an in-flight swap and no retry or relaunch runs after it", async () => {
  const dir = mkdtempSync(join(tmpdir(), "media-reconf-stop-"));
  const keys = generateKeyPairSync("ed25519");
  const calls: string[] = [];
  let releaseSwap: (() => void) | null = null;
  const job = (): RelayJob => ({
    async start(c) { calls.push(`start v${c.desiredVersion}`); },
    async stop() { calls.push("stop"); },
    async reconfigure(c) {
      calls.push(`swap v${c.desiredVersion}`);
      // B is still trying to go on air when the worker shuts down; then it lags.
      await new Promise<void>((resolve) => { releaseSwap = resolve; });
      throw new Error("compositor B never went on air");
    },
  });
  const pending: Array<{ action: () => Promise<void>; cancelled: boolean }> = [];
  const scheduler: RetryScheduler = {
    schedule(action) { const entry = { action, cancelled: false }; pending.push(entry); return { cancel: () => { entry.cancelled = true; } }; },
  };
  const worker = new MediaSupervisor("worker-1", [base], store, new DurableFence(join(dir, "fence.json")),
    new Map([["control-v1", keys.publicKey]]), job, { transition() {} }, scheduler);
  const send = (version: number) => {
    const body: CommandBody = {
      workspaceId: base.workspaceId, channelId: base.channelId, workerId: base.workerId, assignmentEpoch: base.assignmentEpoch,
      commandId: randomUUID(), desiredVersion: version, issuedAt: new Date(Date.now() - 1_000).toISOString(),
      expiresAt: new Date(Date.now() + 60_000).toISOString(), snapshot: { ...base, desiredVersion: version },
    };
    return worker.handle(signCommand(body, "control-v1", keys.privateKey));
  };
  try {
    await worker.start();
    await send(1);
    const swapping = send(2);
    while (!releaseSwap) await new Promise((r) => setImmediate(r));
    let stopped = false;
    const stopping = worker.stop().then(() => { stopped = true; });
    await new Promise((r) => setImmediate(r));
    assert.equal(stopped, false, "stop waits for the in-flight swap");
    (releaseSwap as () => void)();
    await swapping;
    await stopping;
    assert.deepEqual(calls, ["start v1", "swap v2", "stop"]);
    assert.equal(worker.status()[0]?.state, "STOPPED");
    // The retry the failed swap scheduled is cancelled, and running it anyway does nothing.
    for (const entry of pending) { assert.equal(entry.cancelled, true); await entry.action(); }
    assert.deepEqual(calls, ["start v1", "swap v2", "stop"]);
  } finally { rmSync(dir, { recursive: true, force: true }); }
});