import { test } from "node:test";
import assert from "node:assert/strict";
import path from "node:path";
import { mkdtemp, rm, writeFile, utimes } from "node:fs/promises";
import { tmpdir } from "node:os";
import {
  decideKeep,
  reacquireConfigFor,
  refreshCues,
  type KeepInput,
} from "./backfillReacquire";
import { isCuesJsonFresh } from "./normalizeTranscript";
import {
  CUES_JSON_FILENAME,
  META_FILENAME,
  VTT_FILENAME,
} from "../lib/videoStatus";
import type { ChannelConfig } from "../lib/channelConfig";
import {
  compileLaneRoot,
  defaultChannelPriority,
  sanitizeChannelPriority,
} from "../lib/channelPriority";

// Run with:
//   pnpm --filter yt-dlp-transcript-common exec tsx --test common/controller/backfillReacquire.test.ts
//
// Only the two pure-ish surfaces: reacquireMediaFor itself needs a network and
// a disk gate, and this repo does not module-mock. The e2e in
// editor/e2e/backfill.spec.ts covers the whole path.

// ---------------------------------------------------------------------------
// reacquireConfigFor — the fix for ~16,000 caption re-fetches that landed no
// audio (eight channels, 2026-08-22 -> 08-26).

const YOUTUBE_CHANNEL: ChannelConfig = {
  handling: "youtube",
  name: "The Channel",
  platform: "youtube",
  audioFormat: "mp3",
  url: "https://www.youtube.com/@channel",
  cookiesFromBrowser: "firefox",
  cookieMode: "when-required",
  subLangs: "en",
  keepLatest: 3,
};

test("a subtitle channel re-acquires with transcribe handling, everything else intact", () => {
  const out = reacquireConfigFor(YOUTUBE_CHANNEL);
  assert.equal(out.handling, "transcribe");
  // Every OTHER field survives: the override is one key, not a synthetic
  // config. Cookies especially — a re-acquire on a members-gated channel still
  // needs them.
  for (const [key, value] of Object.entries(YOUTUBE_CHANNEL)) {
    if (key === "handling") continue;
    assert.deepEqual(
      (out as Record<string, unknown>)[key],
      value,
      `field ${key} was not carried over`,
    );
  }
});

test("the channel's stored config is never mutated", () => {
  const config: ChannelConfig = { ...YOUTUBE_CHANNEL };
  const out = reacquireConfigFor(config);
  assert.notEqual(out, config);
  // The load-bearing one: readChannelConfig hands back the object the rest of
  // the app reads, and a mutation here would turn one video's override into a
  // permanent change of what the channel downloads.
  assert.equal(config.handling, "youtube");
});

test("a transcribe channel is passed through unchanged, same object", () => {
  const config: ChannelConfig = { handling: "transcribe", name: "Other" };
  // Identity matters: reacquireMediaFor logs the override only when the object
  // changed, so a copy here would announce an override on every video.
  assert.equal(reacquireConfigFor(config), config);
});

// ---------------------------------------------------------------------------
// refreshCues — the second half of the defect. Every re-acquire rewrites
// metadata.info.json (the prefetch runs before any attempt), which leaves
// transcript.cues.json stale by mtime; that is what made 16,081 videos read
// `deferred` to the digest lane and `no-transcript` to attribution.

const CONFIG: ChannelConfig = { handling: "youtube", name: "The Channel" };

// mtimes set EXPLICITLY, never left to write order — freshness is pure mtime
// math and two files written in the same millisecond compare equal on
// filesystems with coarse timestamps. Same discipline as normalizeAll.test.ts.
async function staleVideoFixture(): Promise<{
  dir: string;
  cleanup: () => Promise<void>;
}> {
  const dir = await mkdtemp(path.join(tmpdir(), "reacquire-cues-"));
  await writeFile(
    path.join(dir, META_FILENAME),
    JSON.stringify({ id: "vid1", title: "A video", duration: 60 }),
  );
  await writeFile(
    path.join(dir, VTT_FILENAME),
    "WEBVTT\n\n00:00:00.000 --> 00:00:05.000\nhello\n",
  );
  await writeFile(
    path.join(dir, CUES_JSON_FILENAME),
    JSON.stringify({ version: 1, id: "vid1", title: "A video", cues: [] }),
  );
  const base = Date.now() / 1000 - 1000;
  // The exact shape a re-acquire leaves behind: cues.json older than the
  // metadata the fetch just rewrote.
  await utimes(path.join(dir, CUES_JSON_FILENAME), base, base);
  await utimes(path.join(dir, VTT_FILENAME), base + 5, base + 5);
  await utimes(path.join(dir, META_FILENAME), base + 10, base + 10);
  return { dir, cleanup: () => rm(dir, { recursive: true, force: true }) };
}

test("refreshCues makes a re-acquired video's cues fresh again", async () => {
  const { dir, cleanup } = await staleVideoFixture();
  try {
    assert.equal((await isCuesJsonFresh(dir)).fresh, false);
    await refreshCues(dir, "the-channel", CONFIG, "vid1", () => {});
    // The property the 16,081 lost: without this the very next lane in the same
    // sweep defers the video until an operator runs Normalize.
    assert.equal((await isCuesJsonFresh(dir)).fresh, true);
  } finally {
    await cleanup();
  }
});

test("refreshCues never throws — it runs on the download-failure path too", async () => {
  const dir = await mkdtemp(path.join(tmpdir(), "reacquire-cues-empty-"));
  try {
    const logged: string[] = [];
    // An empty dir is `skipped: no-metadata`, not an error — but the guarantee
    // being pinned is that NOTHING escapes: a normalize problem must never
    // replace the real outcome of the re-acquire.
    await refreshCues(dir, "the-channel", CONFIG, "vid1", (m) =>
      logged.push(m),
    );
    assert.equal(
      logged.some((m) => m.includes("Re-normalized")),
      false,
    );
  } finally {
    await rm(dir, { recursive: true, force: true });
  }
});

// ---------------------------------------------------------------------------
// decideKeep — keep the re-acquired audio for the transcription lane, or remove
// it? Pure, so every branch is asserted here and the call site is left holding
// nothing but thin I/O.
//
// The stake: on the "remove" side this file is the only thing standing between a
// 76,000-video sweep and a full disk; on the "keep" side, deleting an audio file
// that autoQueue.transcription has already started whisper on costs that video's
// whole run AND appends it to failed-transcriptions, which the manual
// per-channel batch honours permanently.

const GB = 1024 * 1024 * 1024;

const ALL_LEAF = {
  id: "root",
  mode: "strict" as const,
  children: [{ id: "leaf-all", match: { type: "all" as const } }],
};

// The live configuration this was written for: enabled, replaceAutoSubs on, one
// catch-all leaf.
function keepInput(over: Partial<KeepInput> = {}): KeepInput {
  return {
    doNotClean: false,
    transcribing: false,
    hasAudio: true,
    autoSubsOnly: true,
    policy: {
      enabled: true,
      replaceAutoSubs: true,
      snoozeUntil: null,
      root: ALL_LEAF,
    },
    // No channel priority set: the stored tree stands, byte for byte, which is
    // every case below except the two that name a tier.
    priority: defaultChannelPriority(),
    channel: { slug: "the-channel", platform: "youtube" },
    disk: { freeBytes: 100 * GB, minFreeDiskGB: 20, resumeMarginGB: 5 },
    ...over,
  };
}

test("decideKeep: the operator's do-not-clean outranks every policy question", () => {
  // Everything else says "remove" — no audio, not ASR-only, runner off.
  const d = decideKeep(
    keepInput({
      doNotClean: true,
      hasAudio: false,
      autoSubsOnly: false,
      policy: {
        enabled: false,
        replaceAutoSubs: false,
        snoozeUntil: null,
        root: ALL_LEAF,
      },
    }),
  );
  assert.deepEqual(d, { keep: true, reason: "do-not-clean" });
});

test("decideKeep: a running transcription keeps the file whatever the policy says", () => {
  // The in-flight veto is defense in depth: it covers the manual whisper-all
  // batch and any lane the policy check cannot predict.
  const d = decideKeep(
    keepInput({
      transcribing: true,
      autoSubsOnly: false,
      policy: {
        enabled: false,
        replaceAutoSubs: false,
        snoozeUntil: null,
        root: ALL_LEAF,
      },
    }),
  );
  assert.deepEqual(d, { keep: true, reason: "in-flight" });
});

test("decideKeep: ASR-only + an enabled policy that covers the channel is the hand-off", () => {
  assert.deepEqual(decideKeep(keepInput()), {
    keep: true,
    reason: "hand-off",
  });
});

test("decideKeep: nothing to hand over without audio, or without ASR-only captions", () => {
  assert.deepEqual(decideKeep(keepInput({ hasAudio: false })), {
    keep: false,
    reason: "no-audio",
  });
  // A manually-captioned or already-whispered video is not in the bucket, so
  // keeping its audio would leak it.
  assert.deepEqual(decideKeep(keepInput({ autoSubsOnly: false })), {
    keep: false,
    reason: "not-auto-subs-only",
  });
});

test("decideKeep: audio is not kept for a runner that is off or snoozed", () => {
  assert.deepEqual(
    decideKeep(
      keepInput({
        policy: {
          enabled: false,
          replaceAutoSubs: true,
          snoozeUntil: null,
          root: ALL_LEAF,
        },
      }),
    ),
    { keep: false, reason: "policy-off" },
  );
  // A LAPSED snooze is already normalized to null by sanitizePolicy, so a
  // non-null value here means the runner really is still idling.
  assert.deepEqual(
    decideKeep(
      keepInput({
        policy: {
          enabled: true,
          replaceAutoSubs: true,
          snoozeUntil: Date.now() + 60_000,
          root: ALL_LEAF,
        },
      }),
    ),
    { keep: false, reason: "policy-snoozed" },
  );
});

test("decideKeep: no leaf covering this channel draws the bucket", () => {
  // replaceAutoSubs is ON, but the only leaf is for another channel — the video
  // would never be claimed, so its audio would sit forever.
  const d = decideKeep(
    keepInput({
      policy: {
        enabled: true,
        replaceAutoSubs: true,
        snoozeUntil: null,
        root: {
          id: "root",
          mode: "strict",
          children: [
            { id: "l", match: { type: "channel", value: "someone-else" } },
          ],
        },
      },
    }),
  );
  assert.deepEqual(d, { keep: false, reason: "no-leaf" });
});

test("decideKeep: an explicit bucket leaf hands off with replaceAutoSubs OFF", () => {
  // The exploration's finding, pinned: `replaceAutoSubs: false` alone is NOT a
  // refusal. A leaf naming downloadedAutoSubsOnly draws it regardless, so an
  // operator who turned the flag off but kept the leaf still gets the hand-off
  // — and would otherwise get the race this whole change removes.
  const d = decideKeep(
    keepInput({
      policy: {
        enabled: true,
        replaceAutoSubs: false,
        snoozeUntil: null,
        root: {
          id: "root",
          mode: "strict",
          children: [
            {
              id: "l",
              match: { type: "all", bucket: "downloadedAutoSubsOnly" },
            },
          ],
        },
      },
    }),
  );
  assert.deepEqual(d, { keep: true, reason: "hand-off" });
});

test("decideKeep: the disk bar is the RESUME mark, not the floor", () => {
  // 22 GB free with a 20 GB floor and a 5 GB margin: above the floor, below the
  // 25 GB the download runner itself would demand before resuming. The backfill
  // must not keep audio the runner would have refused to fetch.
  assert.deepEqual(
    decideKeep(
      keepInput({
        disk: { freeBytes: 22 * GB, minFreeDiskGB: 20, resumeMarginGB: 5 },
      }),
    ),
    { keep: false, reason: "disk-low" },
  );
  // At the mark exactly, it is a hand-off.
  assert.deepEqual(
    decideKeep(
      keepInput({
        disk: { freeBytes: 25 * GB, minFreeDiskGB: 20, resumeMarginGB: 5 },
      }),
    ),
    { keep: true, reason: "hand-off" },
  );
});

test("decideKeep: a disabled disk gate never refuses the hand-off", () => {
  assert.deepEqual(
    decideKeep(
      keepInput({
        disk: { freeBytes: 1, minFreeDiskGB: 0, resumeMarginGB: 5 },
      }),
    ),
    { keep: true, reason: "hand-off" },
  );
});

test("decideKeep never mutates its input", () => {
  const input = keepInput();
  const before = JSON.stringify(input);
  decideKeep(input);
  assert.equal(JSON.stringify(input), before);
});

// --- A HOLD IS NEVER A STOP, and this is where it would have become one -----
//
// The bug this closes (the S0/S1 review, finding 1): once the priority model
// says anything, `autoQueue.transcription.root` is COMPILED from it and a
// channel paused for transcription has no leaf in that tree. Decided on the
// compiled tree with no pause branch, every such channel answers "no-leaf" and
// every re-acquired audio file it produces is UNLINKED — a pause turning into
// data loss. Decided on the STORED tree it is just as wrong the other way: the
// stored tree is not what the runner dispatches from.

test("decideKeep: a channel paused for transcription KEEPS its audio", () => {
  const priority = sanitizeChannelPriority({
    focus: { kind: "none" },
    channels: { "the-channel": { tier: "paused" } },
  });
  assert.deepEqual(decideKeep(keepInput({ priority })), {
    keep: true,
    reason: "paused",
  });
  // The compiled tree really does omit it — this is the answer the naive
  // reading would have produced, and it is why the pause is asked first.
  assert.equal(
    JSON.stringify(
      compileLaneRoot("transcription", priority, ["the-channel"], []),
    ).includes("the-channel"),
    false,
  );
});

test("decideKeep: a PER-OPERATION pause is per operation", () => {
  // What all 15 live `excludeFromSync` channels migrate to: stop syncing, keep
  // everything else. The transcription lane is untouched, so this is an
  // ordinary hand-off and not a keep-for-a-hold.
  const syncPaused = sanitizeChannelPriority({
    focus: { kind: "none" },
    channels: { "the-channel": { tier: "normal", overrides: { sync: "paused" } } },
  });
  assert.deepEqual(decideKeep(keepInput({ priority: syncPaused })), {
    keep: true,
    reason: "hand-off",
  });
  // And the inverse: paused for DOWNLOAD only leaves transcription running too.
  const downloadPaused = sanitizeChannelPriority({
    focus: { kind: "none" },
    channels: {
      "the-channel": { tier: "normal", overrides: { download: "paused" } },
    },
  });
  assert.deepEqual(decideKeep(keepInput({ priority: downloadPaused })), {
    keep: true,
    reason: "hand-off",
  });
});

test("decideKeep: a non-paused channel is decided on the COMPILED tree", () => {
  // A model that says something, and a stored root that says nothing at all.
  // The stored tree would answer "no leaf covers this channel"; the compiled
  // one gives every non-paused channel a bare leaf, and with replaceAutoSubs
  // on that leaf draws the opt-in bucket. The hand-off is the right answer
  // because the compiled tree is the one the runner dispatches from.
  const priority = sanitizeChannelPriority({
    focus: { kind: "none" },
    channels: { "someone-else": { tier: "paused" } },
  });
  const d = decideKeep(
    keepInput({
      priority,
      policy: {
        enabled: true,
        replaceAutoSubs: true,
        snoozeUntil: null,
        root: { id: "root", mode: "strict", children: [] },
      },
    }),
  );
  assert.deepEqual(d, { keep: true, reason: "hand-off" });
});

test("decideKeep: the pause does not outrank a run in progress or a missing file", () => {
  const priority = sanitizeChannelPriority({
    focus: { kind: "none" },
    channels: { "the-channel": { tier: "paused" } },
  });
  // Nothing landed: there is no audio to hold for anyone.
  assert.deepEqual(decideKeep(keepInput({ priority, hasAudio: false })), {
    keep: false,
    reason: "no-audio",
  });
  // And a paused channel whose captions are not ASR-only is not this bucket's
  // shape either — the pause keeps audio the lane would otherwise draw, not
  // audio no leaf of any tier would ever touch.
  assert.deepEqual(decideKeep(keepInput({ priority, autoSubsOnly: false })), {
    keep: false,
    reason: "not-auto-subs-only",
  });
});