import { test } from "node:test";
import assert from "node:assert/strict";
import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import type { Paths } from "../lib/paths";
import {
  clearSubtitleDeferral,
  heldPlatformRefusal,
  platformCooldownRemainingMs,
  readSubtitleDeferrals,
  recordDownloadBackoff,
  recordSubtitleDeferral,
} from "./downloadBackoff";
import { PACING_DEFAULTS } from "./platformBackoff";
import { clearPlatformHold, recordPlatformClean } from "./downloadBackoff";
import { BACKOFF_BASE_MS } from "./platformBackoff";
import {
  emptyAutoQueueState,
  readAutoQueueState,
  sharedAutoQueueState,
  writeAutoQueueState,
} from "./autoQueueState";

// Run with: pnpm --filter yt-dlp-transcript-common exec tsx --test common/jobs/downloadBackoff.test.ts
//
// downloadBackoff only touches paths.autoQueueStateFile, so a stub Paths with
// just that field exercises the full read-modify-write against a temp file.

async function withTempPaths(
  fn: (paths: Paths) => Promise<void>,
): Promise<void> {
  const dir = await mkdtemp(path.join(tmpdir(), "dl-backoff-"));
  const paths = {
    autoQueueStateFile: path.join(dir, ".auto-queue", "state.json"),
  } as Paths;
  // No runner is live unless a test makes one (sharedAutoQueueState).
  globalThis.__yttAutoQueueState__ = undefined;
  try {
    await fn(paths);
  } finally {
    globalThis.__yttAutoQueueState__ = undefined;
    await rm(dir, { recursive: true, force: true });
  }
}

test("no cooldown on an untouched platform", async () => {
  await withTempPaths(async (paths) => {
    assert.equal(await platformCooldownRemainingMs("odysee", paths), 0);
  });
});

test("recordDownloadBackoff opens a cooldown window", async () => {
  await withTempPaths(async (paths) => {
    await recordDownloadBackoff("odysee", paths);
    const remaining = await platformCooldownRemainingMs("odysee", paths);
    // First failure ≈ BACKOFF_BASE_MS (±10% jitter); just assert it's positive
    // and within a sane upper bound.
    assert.ok(remaining > 0, "platform is cooling down");
    assert.ok(remaining <= BACKOFF_BASE_MS * 1.2, "within base window + jitter");
    // It is recorded under the download kind only.
    const state = JSON.parse(await readFile(paths.autoQueueStateFile, "utf8"));
    assert.ok(state.download.platformBackoff.odysee, "stored under download kind");
    assert.deepEqual(
      state.transcription.platformBackoff,
      {},
      "transcription untouched",
    );
  });
});

test("consecutive failures escalate the fails counter", async () => {
  await withTempPaths(async (paths) => {
    await recordDownloadBackoff("youtube", paths);
    await recordDownloadBackoff("youtube", paths);
    const state = JSON.parse(await readFile(paths.autoQueueStateFile, "utf8"));
    assert.equal(state.download.platformBackoff.youtube.fails, 2);
  });
});

test("remaining is 0 once the window lapses", async () => {
  await withTempPaths(async (paths) => {
    // Hand-write a lapsed entry to simulate an expired cooldown.
    const { writeAutoQueueState, emptyAutoQueueState } = await import(
      "./autoQueueState"
    );
    const s = emptyAutoQueueState();
    s.download.platformBackoff.odysee = { until: Date.now() - 1000, fails: 3 };
    await writeAutoQueueState(paths, s);
    assert.equal(await platformCooldownRemainingMs("odysee", paths), 0);
  });
});

// WITH A RUNNER LIVE, THE COOLDOWN GOES THROUGH THE SHARED OBJECT (release 8).
// Before, recordDownloadBackoff wrote the file only; the runners' in-memory
// object never saw it, so the next persist of ANY lane wrote the object over
// the file and the manual 429 cooldown was gone — and nothing merged it back
// while the download lane was paused, stopped or at capacity.

test("with a runner live, another lane's persist keeps a manual cooldown on disk", async () => {
  await withTempPaths(async (paths) => {
    // A runner holds the shared object (as every lane does from boot on).
    const shared = await sharedAutoQueueState(paths);
    // A clicked Sync hits a 429.
    await recordDownloadBackoff("youtube", paths);
    assert.ok(
      shared.download.platformBackoff.youtube,
      "the live object carries the cooldown",
    );
    // The transcription lane picks something and persists the shared object.
    shared.transcription.runtime.currentWeights = { someLeaf: 1 };
    await writeAutoQueueState(paths, shared);
    const onDisk = await readAutoQueueState(paths);
    assert.equal(onDisk.download.platformBackoff.youtube?.fails, 1);
    assert.deepEqual(onDisk.transcription.runtime.currentWeights, {
      someLeaf: 1,
    });
    assert.ok((await platformCooldownRemainingMs("youtube", paths)) > 0);
  });
});

test("with a runner live, a cooldown already on disk is merged in, not dropped", async () => {
  await withTempPaths(async (paths) => {
    const shared = await sharedAutoQueueState(paths);
    // Written to the file behind the runner's back (a later `until` than the
    // live object knows about).
    const behind = emptyAutoQueueState();
    const until = Date.now() + 10 * 60_000;
    behind.download.platformBackoff.odysee = { until, fails: 4 };
    await writeAutoQueueState(paths, behind);
    await recordDownloadBackoff("youtube", paths);
    assert.deepEqual(shared.download.platformBackoff.odysee, { until, fails: 4 });
    const onDisk = await readAutoQueueState(paths);
    assert.deepEqual(onDisk.download.platformBackoff.odysee, { until, fails: 4 });
    assert.ok(onDisk.download.platformBackoff.youtube);
  });
});

test("platformCooldownRemainingMs prefers the live object over the file", async () => {
  await withTempPaths(async (paths) => {
    const shared = await sharedAutoQueueState(paths);
    shared.download.platformBackoff.rumble = {
      until: Date.now() + 60_000,
      fails: 1,
    };
    // Not persisted yet — the file has no entry.
    assert.ok((await platformCooldownRemainingMs("rumble", paths)) > 0);
  });
});

test("with no runner live, recordDownloadBackoff stays a disk round trip and loads nothing", async () => {
  await withTempPaths(async (paths) => {
    await recordDownloadBackoff("youtube", paths);
    assert.equal(globalThis.__yttAutoQueueState__?.state ?? null, null);
    assert.equal(globalThis.__yttAutoQueueState__?.loading ?? null, null);
    const onDisk = await readAutoQueueState(paths);
    assert.equal(onDisk.download.platformBackoff.youtube?.fails, 1);
  });
});

// ── release 17, slice RL ─────────────────────────────────────────────────────

test("a manual rate limit doubles the pace; a network failure does not", async () => {
  await withTempPaths(async (paths) => {
    await recordDownloadBackoff("youtube", paths, "rate_limit", PACING_DEFAULTS);
    await recordDownloadBackoff("youtube", paths, "network", PACING_DEFAULTS);
    const onDisk = await readAutoQueueState(paths);
    assert.equal(onDisk.download.platformBackoff.youtube.fails, 2);
    const { steppedAt, ...pace } = onDisk.download.platformPace.youtube;
    assert.deepEqual(pace, { sleepRequestsSeconds: 2, baseSeconds: 1, cleanUnits: 0 });
    assert.ok(Math.abs(steppedAt - Date.now()) < 60_000);
  });
});

test("a manual 429 at the cap holds the platform, and the refusal names the next probe", async () => {
  await withTempPaths(async (paths) => {
    const state = emptyAutoQueueState();
    state.download.platformBackoff.youtube = { until: Date.now() - 1, fails: 7 };
    await writeAutoQueueState(paths, state);
    const pacing = { ...PACING_DEFAULTS, holdAfterFailsAtCap: 3, holdProbeMinutes: 60 };
    assert.equal(await heldPlatformRefusal("youtube", "Sync", paths), null);
    await recordDownloadBackoff("youtube", paths, "rate_limit", pacing);
    const onDisk = await readAutoQueueState(paths);
    assert.ok(onDisk.download.platformHolds.youtube);
    assert.equal(
      onDisk.download.platformBackoff.youtube.until,
      onDisk.download.platformHolds.youtube.probeAt,
    );
    const text = await heldPlatformRefusal("youtube", "Sync", paths);
    assert.match(text ?? "", /^youtube is held: .*\(8 failures in a row, held since \d\d:\d\d UTC\)\..* the next probe is in 60 min\. Sync will run once a probe comes back clean\.$/);
  });
});

test("subtitle deferrals count up through the file and clear on success", async () => {
  await withTempPaths(async (paths) => {
    const a = await recordSubtitleDeferral("v1", "alpha", paths, 1000);
    const b = await recordSubtitleDeferral("v1", "alpha", paths, 2000);
    assert.equal(a.count, 1);
    assert.equal(b.count, 2);
    const onDisk = await readAutoQueueState(paths);
    assert.equal(onDisk.download.subtitleDeferrals.v1.count, 2);
    // Nothing platform-wide moved.
    assert.deepEqual(onDisk.download.platformBackoff, {});
    assert.deepEqual(onDisk.download.platformPace, {});
    await clearSubtitleDeferral("v1", paths);
    assert.deepEqual(await readSubtitleDeferrals(paths), {});
  });
});

test("with a runner live, a subtitle deferral is written through the shared object", async () => {
  await withTempPaths(async (paths) => {
    const live = await sharedAutoQueueState(paths);
    await recordSubtitleDeferral("v9", "beta", paths);
    assert.equal(live.download.subtitleDeferrals.v9.count, 1);
    assert.equal((await readAutoQueueState(paths)).download.subtitleDeferrals.v9.count, 1);
  });
});

test("an overdue hold no longer refuses; a clean manual run lifts it (review H2)", async () => {
  await withTempPaths(async (paths) => {
    const state = emptyAutoQueueState();
    const now = Date.now();
    state.download.platformBackoff.youtube = { until: now - 1_000, fails: 9 };
    state.download.platformHolds.youtube = { since: now - 3_600_000, probeAt: now - 1_000, rateLimited: true };
    await writeAutoQueueState(paths, state);
    assert.equal(await heldPlatformRefusal("youtube", "Sync", paths), null);
    const line = await recordPlatformClean("youtube", paths, PACING_DEFAULTS);
    assert.match(line ?? "", /^youtube answered cleanly: its hold and backoff are cleared/);
    const after = (await readAutoQueueState(paths)).download;
    assert.deepEqual(after.platformHolds, {});
    assert.deepEqual(after.platformBackoff, {});
    // Nothing to settle: no line and no write.
    assert.equal(await recordPlatformClean("youtube", paths, PACING_DEFAULTS), null);
  });
});

test("Clear hold drops the hold, the backoff and the pace, and says so", async () => {
  await withTempPaths(async (paths) => {
    const state = emptyAutoQueueState();
    const now = Date.now();
    state.download.platformBackoff.youtube = { until: now + 3_600_000, fails: 9 };
    state.download.platformHolds.youtube = { since: now - 60_000, probeAt: now + 3_600_000, rateLimited: true };
    state.download.platformPace.youtube = { sleepRequestsSeconds: 8, baseSeconds: 1, cleanUnits: 0, steppedAt: now };
    await writeAutoQueueState(paths, state);
    const line = await clearPlatformHold("youtube", paths);
    assert.match(line ?? "", /^Cleared by hand for youtube: the hold \(since .* UTC\), the backoff \(9 failures\), the pace \(8s → base 1s\)\./);
    const after = (await readAutoQueueState(paths)).download;
    assert.deepEqual([after.platformHolds, after.platformBackoff, after.platformPace], [{}, {}, {}]);
    assert.equal(await clearPlatformHold("youtube", paths), null);
  });
});