import { test } from "node:test";
import assert from "node:assert/strict";
import {
  BACKOFF_BASE_MS,
  FAILS_TO_REACH_CAP,
  PACING_DEFAULTS,
  PACE_TIME_DECAY_MS,
  SUBTITLE_HOLD_MS,
  VIDEO_RATE_LIMIT_DEFER_MS,
  effectivePaceSeconds,
  heldPlatformSentence,
  isVideoDeferred,
} from "./platformBackoff";
import { type UnitOutcomeState, applyUnitOutcome } from "./unitOutcome";

// Run with: pnpm --filter yt-dlp-transcript-common exec tsx --test jobs/unitOutcome.test.ts

const noJitter = () => 0.5;
const empty = (): UnitOutcomeState => ({
  platformBackoff: {},
  videoDeferrals: {},
  platformPace: {},
  platformHolds: {},
  subtitleDeferrals: {},
});
const unit = (
  videoId: string,
  over: Partial<Parameters<typeof applyUnitOutcome>[1]> = {},
) => ({
  platform: "youtube",
  videoId,
  channelSlug: "alpha",
  outcome: "failed" as const,
  ...over,
});

test("rate_limit backs the platform off AND defers the video 6h", () => {
  const s = empty();
  const now = 1_000_000;
  const r = applyUnitOutcome(s, unit("v1", { failureClass: "rate_limit" }), now, noJitter);
  assert.equal(r.markCompleted, false);
  assert.deepEqual(s.platformBackoff.youtube, { until: now + BACKOFF_BASE_MS, fails: 1 });
  assert.deepEqual(s.videoDeferrals.v1, {
    until: now + VIDEO_RATE_LIMIT_DEFER_MS,
    channelSlug: "alpha",
  });
  assert.equal(
    r.line,
    "Auto-download: youtube rate_limit — backing off 60s (attempt 1). v1 deferred 6h; next video after cooldown. Pace now 2s between requests.",
  );
});

test("network backs off only — no deferral, today's line", () => {
  const s = empty();
  const r = applyUnitOutcome(s, unit("v1", { failureClass: "network" }), 0, noJitter);
  assert.equal(r.markCompleted, false);
  assert.equal(s.platformBackoff.youtube.fails, 1);
  assert.deepEqual(s.videoDeferrals, {});
  assert.equal(
    r.line,
    "Auto-download: youtube network — backing off 60s (attempt 1). v1 will retry after cooldown.",
  );
});

test("transcribed clears the platform cooldown and retires the video", () => {
  const s = empty();
  s.platformBackoff.youtube = { until: 10, fails: 3 };
  const r = applyUnitOutcome(s, unit("v1", { outcome: "transcribed" }), 0, noJitter);
  assert.deepEqual(r, { markCompleted: true, line: null });
  assert.deepEqual(s.platformBackoff, {});
});

test("any other failure retires the video and leaves the cooldown alone", () => {
  const s = empty();
  s.platformBackoff.youtube = { until: 10, fails: 3 };
  const r = applyUnitOutcome(s, unit("v1", { failureClass: "per_video" }), 0, noJitter);
  assert.deepEqual(r, { markCompleted: true, line: null });
  assert.deepEqual(s.platformBackoff.youtube, { until: 10, fails: 3 });
});

test("a non-download unit (no platform) touches neither map", () => {
  const s = empty();
  const r = applyUnitOutcome(
    s,
    unit("v1", { platform: null, failureClass: "rate_limit" }),
    0,
    noJitter,
  );
  assert.deepEqual(r, { markCompleted: true, line: null });
  assert.deepEqual(s, empty());
});

test("every branch prunes lapsed deferrals", () => {
  const s = empty();
  s.videoDeferrals = {
    old: { until: 5, channelSlug: "a" },
    live: { until: 50, channelSlug: "a" },
  };
  applyUnitOutcome(s, unit("v1", { outcome: "skipped" }), 10, noJitter);
  assert.deepEqual(Object.keys(s.videoDeferrals), ["live"]);
});

test("fails climbs only across DISTINCT ids: the deferred video is not re-picked", () => {
  // The runner's pick skips a deferred id, so after a cooldown the next 429 is
  // necessarily a different video — the escalation now measures the platform,
  // not one stubborn video.
  const s = empty();
  let now = 0;
  const ids = ["v1", "v2", "v3"];
  for (const id of ids) {
    assert.equal(isVideoDeferred(s.videoDeferrals, id, now), false);
    applyUnitOutcome(s, unit(id, { failureClass: "rate_limit" }), now, noJitter);
    now = s.platformBackoff.youtube.until; // the cooldown lapses
  }
  assert.equal(s.platformBackoff.youtube.fails, 3);
  for (const id of ids) assert.equal(isVideoDeferred(s.videoDeferrals, id, now), true);
});

// ── release 17, slice RL: the subtitle deferral, the pace, the hold ─────────

test("subs_rate_limit retires the video, defers its subtitles, and touches no platform state", () => {
  const s = empty();
  s.platformBackoff.youtube = { until: 10, fails: 3 };
  s.platformPace.youtube = { sleepRequestsSeconds: 4, baseSeconds: 1, cleanUnits: 2, steppedAt: 1_000_000 };
  const before = structuredClone({
    platformBackoff: s.platformBackoff,
    platformPace: s.platformPace,
    platformHolds: s.platformHolds,
  });
  const now = 1_000_000;
  const r = applyUnitOutcome(
    s,
    unit("v1", { outcome: "transcribed", failureClass: "subs_rate_limit" }),
    now,
    noJitter,
  );
  assert.equal(r.markCompleted, true);
  assert.equal(
    r.line,
    "Auto-download: v1 downloaded; its subtitles were rate-limited (1×) — deferred 6h for download-missing-subs. youtube is not backed off.",
  );
  assert.deepEqual(s.subtitleDeferrals.v1, {
    count: 1,
    lastAt: now,
    until: now + VIDEO_RATE_LIMIT_DEFER_MS,
    channelSlug: "alpha",
  });
  // Not cleared like a success, not escalated like a failure, no video deferral.
  assert.deepEqual(
    { platformBackoff: s.platformBackoff, platformPace: s.platformPace, platformHolds: s.platformHolds },
    before,
  );
  assert.deepEqual(s.videoDeferrals, {});
});

test("the third subtitle deferral leaves the subtitles alone for 7 days", () => {
  const s = empty();
  let now = 0;
  for (let i = 1; i <= 3; i++) {
    const r = applyUnitOutcome(
      s,
      unit("v1", { outcome: "transcribed", failureClass: "subs_rate_limit" }),
      now,
      noJitter,
    );
    assert.equal(s.subtitleDeferrals.v1.count, i);
    if (i < 3) {
      assert.equal(s.subtitleDeferrals.v1.until, now + VIDEO_RATE_LIMIT_DEFER_MS);
      now += VIDEO_RATE_LIMIT_DEFER_MS + 1; // the 6 h window lapses; the count stays
    } else {
      assert.equal(s.subtitleDeferrals.v1.until, now + SUBTITLE_HOLD_MS);
      assert.match(r.line ?? "", /\(3×\) — left alone for 7 days/);
    }
  }
  // A week after the hold lapsed the record is pruned, count and all.
  applyUnitOutcome(s, unit("v2", { outcome: "skipped" }), now + 2 * SUBTITLE_HOLD_MS + 1, noJitter);
  assert.deepEqual(s.subtitleDeferrals, {});
});

test("the pace doubles on each rate limit, never on network, and stops at the cap", () => {
  const s = empty();
  const pacing = { ...PACING_DEFAULTS, sleepRequestsCapSeconds: 8 };
  let now = 0;
  const paces: number[] = [];
  for (let i = 0; i < 5; i++) {
    applyUnitOutcome(s, unit(`v${i}`, { failureClass: "rate_limit" }), now, noJitter, pacing);
    paces.push(s.platformPace.youtube.sleepRequestsSeconds);
    now = s.platformBackoff.youtube.until;
  }
  assert.deepEqual(paces, [2, 4, 8, 8, 8]);
  applyUnitOutcome(s, unit("vn", { failureClass: "network" }), now, noJitter, pacing);
  assert.equal(s.platformPace.youtube.sleepRequestsSeconds, 8);
  // A platform with no static pace starts from nothing and goes to 1 s.
  applyUnitOutcome(s, unit("o1", { platform: "odysee", failureClass: "rate_limit" }), now, noJitter, pacing);
  assert.deepEqual(s.platformPace.odysee, { sleepRequestsSeconds: 1, baseSeconds: 0, cleanUnits: 0, steppedAt: now });
});

test("the pace eases one step per N clean units, and an entry back at its base is gone", () => {
  const s = empty();
  const pacing = { ...PACING_DEFAULTS, decayAfterCleanUnits: 2 };
  s.platformPace.youtube = { sleepRequestsSeconds: 4, baseSeconds: 1, cleanUnits: 0, steppedAt: 0 };
  const lines: (string | null)[] = [];
  for (let i = 0; i < 4; i++) {
    lines.push(applyUnitOutcome(s, unit(`c${i}`, { outcome: "transcribed" }), 0, noJitter, pacing).line);
  }
  assert.deepEqual(lines, [
    null,
    "Auto-download: youtube pace eased to 2s between requests.",
    null,
    "Auto-download: youtube pace eased to 1s between requests.",
  ]);
  assert.deepEqual(s.platformPace, {});
  // A subtitle-429 unit is not clean: it does not count toward the decay.
  s.platformPace.youtube = { sleepRequestsSeconds: 4, baseSeconds: 1, cleanUnits: 1, steppedAt: 0 };
  applyUnitOutcome(s, unit("x", { outcome: "transcribed", failureClass: "subs_rate_limit" }), 0, noJitter, pacing);
  assert.equal(s.platformPace.youtube.cleanUnits, 1);
});

test("a backoff failing at the cap N times holds the platform: one probe per holdProbeMinutes", () => {
  const s = empty();
  const pacing = { ...PACING_DEFAULTS, holdAfterFailsAtCap: 3, holdProbeMinutes: 60 };
  let now = 0;
  const lines: string[] = [];
  // Climb to the cap, then two failures AT it: no hold yet.
  for (let i = 0; i < FAILS_TO_REACH_CAP + 1; i++) {
    lines.push(applyUnitOutcome(s, unit(`v${i}`, { failureClass: "rate_limit" }), now, noJitter, pacing).line ?? "");
    now = s.platformBackoff.youtube.until;
  }
  assert.deepEqual(s.platformHolds, {});
  assert.equal(s.platformBackoff.youtube.fails, FAILS_TO_REACH_CAP + 1);
  // The third at the cap: held, and the next try is a probe an hour away.
  const r = applyUnitOutcome(s, unit("h1", { failureClass: "rate_limit" }), now, noJitter, pacing);
  assert.deepEqual(s.platformHolds.youtube, { since: now, probeAt: now + 60 * 60_000, rateLimited: true });
  assert.equal(s.platformBackoff.youtube.until, now + 60 * 60_000);
  assert.match(r.line ?? "", /^Auto-download: youtube rate_limit — held after 8 failures in a row; next probe in 3600s\./);
  // A failed probe keeps the hold (and its `since`) and re-arms the hour.
  const since = now;
  now = s.platformBackoff.youtube.until;
  const r2 = applyUnitOutcome(s, unit("h2", { failureClass: "network" }), now, noJitter, pacing);
  assert.deepEqual(s.platformHolds.youtube, { since, probeAt: now + 60 * 60_000, rateLimited: true });
  assert.match(r2.line ?? "", /still held/);
  // A clean probe clears the hold AND the backoff.
  const r3 = applyUnitOutcome(s, unit("h3", { outcome: "transcribed" }), now + 60 * 60_000, noJitter, pacing);
  assert.deepEqual(s.platformHolds, {});
  assert.equal(s.platformBackoff.youtube, undefined);
  assert.equal(
    r3.line,
    "Auto-download: youtube probe came back clean — the hold and the backoff are cleared.",
  );
});

test("a probe whose media came down but whose subtitles 429'd lifts the hold, without easing the pace (review L1)", () => {
  const s = empty();
  s.platformBackoff.youtube = { until: 100, fails: 9 };
  s.platformHolds.youtube = { since: 0, probeAt: 100, rateLimited: true };
  s.platformPace.youtube = { sleepRequestsSeconds: 4, baseSeconds: 1, cleanUnits: 0, steppedAt: 100 };
  const r = applyUnitOutcome(
    s,
    unit("p", { outcome: "transcribed", failureClass: "subs_rate_limit" }),
    100,
    noJitter,
  );
  assert.deepEqual(s.platformHolds, {});
  assert.equal(s.platformBackoff.youtube, undefined);
  assert.equal(s.platformPace.youtube.cleanUnits, 0);
  assert.equal(s.subtitleDeferrals.p.count, 1);
  assert.match(r.line ?? "", /The youtube probe's media came down: the hold and the backoff are cleared\.$/);
});

test("the pace eases one step per hour with no rate limit (the one time-based rule, review L2)", () => {
  const s = empty();
  const t0 = 10 * PACE_TIME_DECAY_MS;
  s.platformPace.youtube = { sleepRequestsSeconds: 8, baseSeconds: 1, cleanUnits: 0, steppedAt: t0 };
  applyUnitOutcome(s, unit("x", { outcome: "skipped" }), t0 + PACE_TIME_DECAY_MS - 1, noJitter);
  assert.equal(s.platformPace.youtube.sleepRequestsSeconds, 8);
  applyUnitOutcome(s, unit("x", { outcome: "skipped" }), t0 + 2 * PACE_TIME_DECAY_MS, noJitter);
  assert.deepEqual(s.platformPace.youtube, {
    sleepRequestsSeconds: 2,
    baseSeconds: 1,
    cleanUnits: 0,
    steppedAt: t0 + 2 * PACE_TIME_DECAY_MS,
  });
  // The readers see the same easing without a write.
  assert.equal(effectivePaceSeconds(s.platformPace.youtube, t0 + 3 * PACE_TIME_DECAY_MS), 1);
  applyUnitOutcome(s, unit("x", { outcome: "skipped" }), t0 + 9 * PACE_TIME_DECAY_MS, noJitter);
  assert.deepEqual(s.platformPace, {});
});

test("a hold reached through network failures alone is worded as failing, not rate-limited (review L5)", () => {
  const s = empty();
  const pacing = { ...PACING_DEFAULTS, holdAfterFailsAtCap: 1 };
  s.platformBackoff.youtube = { until: 0, fails: 5 };
  applyUnitOutcome(s, unit("n", { failureClass: "network" }), 10, noJitter, pacing);
  assert.equal(s.platformHolds.youtube.rateLimited, false);
  assert.match(
    heldPlatformSentence("youtube", s.platformHolds.youtube, 6, 10, { probeMinutes: 60 }),
    /^youtube is held: it kept failing \(network errors\) past the cooldown cap/,
  );
  // A rate limit while held makes it a rate-limited hold.
  applyUnitOutcome(s, unit("r", { failureClass: "rate_limit" }), 20, noJitter, pacing);
  assert.equal(s.platformHolds.youtube.rateLimited, true);
});