import {
  createMockRssServer,
  type MockFeedItem,
  type MockRssServer,
  TMDB_FIXTURES,
  type TmdbFixture,
} from "@umt/testkit";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { TORRENT_SIZE, torrentFor } from "./fixtures/modules.ts";
import { completeDownload, createE2E, type E2E, pgAvailable, seedBase, waitFor } from "./harness.ts";

// E9 连续更新剧集(W5-C,对照 MP 缺集 − note 与批量下载):全部经真实任务引擎 + outbox 中继。
// 季共 3 集(mock TMDB),新作品策略按 once 订阅。E01 还在下载(Slot 忙)时 E02、E03 与 E02 的另一个版本先后到达:
//   · E02/E03 与在途的 E01 单元不相交 → 各自得到计划并同时下载(Slot×单元预留),不再因 slot_busy 丢集;
//   · E02 另一版本与在途 E02 单元冲突 → 延后(候选评估 deferred;订阅接收即完成后延后行清掉),不重复下载;
//   · refreshMetadata(编排循环按周期入队)把 TMDB 季集数写进季单元 → 三集全部入库后 once 订阅完成并写历史。

const hasPg = await pgAvailable();
const BASE = "https://audiences.example";
const EP = (n: number, grp = "x264-GRP") => `Three.Eps.S01E0${n}.1080p.WEB-DL.${grp}`;
const ALT = EP(2, "x265-ALT");

const SHOW: TmdbFixture = {
  type: "tv",
  id: 990001,
  original_title: "Three Eps",
  original_language: "en",
  origin_country: ["US"],
  date: "2026-09-01",
  titles: { "zh-CN": "三集剧", "en-US": "Three Eps" },
  popularity: 50,
  alternative_titles: [],
  external_ids: {},
  seasons: [
    {
      season_number: 1,
      air_date: "2026-09-01",
      names: { "zh-CN": "第 1 季", "en-US": "Season 1" },
      episodes: [1, 2, 3].map((i) => ({
        episode_number: i,
        air_date: `2026-09-0${i}`,
        names: { "zh-CN": `第 ${i} 集`, "en-US": `Episode ${i}` },
      })),
    },
  ],
};

const item = (id: number, title: string): MockFeedItem => ({
  guid: `aud-${id}`,
  title,
  link: `${BASE}/details.php?id=${id}&hit=1`,
  enclosureUrl: `${BASE}/download.php?id=${id}&passkey=E9PASSKEY`,
  size: TORRENT_SIZE,
  pubDate: new Date(),
  categories: ["剧集"],
});

describe.skipIf(!hasPg)("E9 连续更新剧集:Slot 忙时到达的集全部入库", () => {
  let e: E2E;
  let rss: MockRssServer;
  let feedId: string;
  let entityId: string;
  let slotId: string;
  let subId: string;

  let lastSlot = -1;
  const fetchFeed = async () => {
    const slotOf = () => Math.floor(Date.now() / 10_000);
    if (slotOf() === lastSlot) await new Promise((r) => setTimeout(r, (lastSlot + 1) * 10_000 - Date.now() + 50));
    lastSlot = slotOf();
    const r = await e.api("POST", `/feed-sources/${feedId}/fetch`);
    expect(r.status).toBe(200);
  };
  const dtOf = (title: string) =>
    e.db.selectFrom("download_task").selectAll().where("torrent_hash", "=", torrentFor(title).hash).executeTakeFirst();
  const importedFor = async (title: string) => {
    const dt = await waitFor(async () => dtOf(title), 60_000, `${title} download_task`);
    return waitFor(
      async () =>
        e.db
          .selectFrom("organization_record")
          .selectAll()
          .where("download_task_id", "=", dt.id)
          .where("status", "in", ["succeeded", "failed", "partial"])
          .executeTakeFirst(),
      90_000,
      `${title} 整理记录结束`,
    );
  };

  beforeAll(async () => {
    e = await createE2E("e9", { tmdbFixtures: [...TMDB_FIXTURES, SHOW] });
    const base = await seedBase(e, { baseUrl: `${BASE}/` });
    rss = await createMockRssServer({ passkey: "E9PASSKEY", items: [item(930001, EP(1))] });
    const urlRef = await e.booted.ctx.secrets.create({ purpose: "feed_url", plaintext: rss.feedUrl });
    feedId = (
      await e.db
        .insertInto("feed_source")
        .values({
          site_id: base.siteId,
          name: "观众-剧集",
          purpose: "series",
          url_secret_ref: urlRef,
          url_display: rss.feedUrl.replace(/passkey=[^&]+/, "passkey=[REDACTED]"),
          poll_interval_sec: 3600,
        })
        .returning("id")
        .executeTakeFirstOrThrow()
    ).id;
    const created = await e.api("POST", "/policy-revisions", {
      policyKey: "series",
      kind: "acquisition",
      title: "剧集 1080p",
      scope: { mediaKinds: ["series"] },
      rules: {
        newTitle: {
          enabled: true,
          windowDays: 3650,
          mediaKinds: ["series"],
          slots: [{ name: "1080p", hardConditions: { resolutions: ["1080p"] } }],
        },
      },
    });
    expect(created.status).toBe(201);
    const pub = await e.api("POST", `/policy-revisions/${created.body.data.id}/publish`, {
      expectedRevision: created.body.data.revision,
    });
    expect(pub.status).toBe(200);
  }, 120_000);

  afterAll(async () => {
    await rss?.close();
    await e?.close();
  }, 60_000);

  it("E01 下载中;TMDB 季集数写入季单元(总集数 3)", async () => {
    await fetchFeed();
    const rel = await waitFor(
      async () =>
        e.db
          .selectFrom("source_release")
          .selectAll()
          .where("title", "=", EP(1))
          .where("resolution_status", "=", "resolved")
          .executeTakeFirst(),
      60_000,
      "E01 识别完成",
    );
    entityId = rel.media_entity_id!;
    const sub = await waitFor(
      async () =>
        e.db.selectFrom("subscription").selectAll().where("media_entity_id", "=", entityId).executeTakeFirst(),
      30_000,
      "新作品订阅",
    );
    subId = sub.id;
    expect(sub).toMatchObject({ mode: "once", status: "active" });
    slotId = (
      await e.db
        .selectFrom("acquisition_slot")
        .select("id")
        .where("media_entity_id", "=", entityId)
        .executeTakeFirstOrThrow()
    ).id;
    await waitFor(async () => e.qb.get(torrentFor(EP(1)).hash), 60_000, "qB 收到 E01");
    const season = await waitFor(
      async () =>
        e.db
          .selectFrom("content_unit")
          .select(["episode_total", "episode_total_source"])
          .where("media_entity_id", "=", entityId)
          .where("kind", "=", "season")
          .where("episode_total", "is not", null)
          .executeTakeFirst(),
      60_000,
      "季总集数刷新",
    );
    expect(season).toEqual({ episode_total: 3, episode_total_source: "tmdb" });
  }, 180_000);

  it("Slot 忙时 E02、E03 与 E02 另一版本到达:E02/E03 同时在途,另一版本延后", async () => {
    rss.addItem(item(930002, EP(2)));
    rss.addItem(item(930003, EP(3)));
    rss.addItem(item(930004, ALT));
    await fetchFeed();
    // E02(任一版本)与 E03 都进了 qB;E01 仍未完成 → 三个在途计划并存
    await waitFor(async () => e.qb.get(torrentFor(EP(3)).hash), 90_000, "qB 收到 E03");
    const e02 = await waitFor(
      async () => e.qb.get(torrentFor(EP(2)).hash) ?? e.qb.get(torrentFor(ALT).hash),
      60_000,
      "qB 收到 E02",
    );
    const inflight = await e.db
      .selectFrom("acquisition_plan")
      .select(["unit_keys", "status"])
      .where("slot_id", "=", slotId)
      .where("status", "in", ["reserved", "submitted"])
      .execute();
    expect(inflight.map((p) => p.unit_keys[0]).sort()).toEqual(["s1e1", "s1e2", "s1e3"]);
    const slot = await e.db
      .selectFrom("acquisition_slot")
      .selectAll()
      .where("id", "=", slotId)
      .executeTakeFirstOrThrow();
    expect(slot.status).toBe("downloading");
    // 另一版本 E02:延后而不是丢弃
    const loser = e02.name.includes("x265-ALT") ? EP(2) : ALT;
    const loserRel = await waitFor(
      async () =>
        e.db
          .selectFrom("source_release")
          .select("id")
          .where("title", "=", loser)
          .where("resolution_status", "=", "resolved")
          .executeTakeFirst(),
      60_000,
      "另一版本识别",
    );
    // 三集都被下载器接收后 once 订阅即完成,完成时延后行随之清掉;以候选评估记录为准
    await waitFor(
      async () =>
        (
          await e.db
            .selectFrom("artifact")
            .select("content")
            .where("title", "=", "candidate_evaluation")
            .where("subject_id", "=", slotId)
            .execute()
        ).find((r) => {
          const c = r.content as { sourceReleaseId: string; decision: string; reasons: string[] };
          return (
            c.sourceReleaseId === loserRel.id &&
            c.decision === "deferred" &&
            c.reasons.includes("deferred:units_inflight")
          );
        }),
      60_000,
      "另一版本延后",
    );
  }, 180_000);

  it("三集依次完成全部入库;once 订阅完成并写历史;延后的版本重评后不重复下载", async () => {
    const e02Title = e.qb.get(torrentFor(EP(2)).hash) ? EP(2) : ALT;
    for (const t of [EP(3), e02Title, EP(1)]) {
      await completeDownload(e, t);
      const rec = await importedFor(t);
      expect(rec.status).toBe("succeeded");
    }
    const cov = await e.db
      .selectFrom("asset_unit_coverage as c")
      .innerJoin("content_unit as u", "u.id", "c.content_unit_id")
      .innerJoin("media_asset as a", "a.id", "c.media_asset_id")
      .select(["u.episode_number"])
      .where("a.media_entity_id", "=", entityId)
      .where("a.status", "=", "active")
      .orderBy("u.episode_number")
      .execute();
    expect(cov.map((c) => c.episode_number)).toEqual([1, 2, 3]);
    const sub = await waitFor(
      async () => {
        const s = await e.db.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirstOrThrow();
        return s.status === "completed" ? s : null;
      },
      30_000,
      "订阅完成",
    );
    expect(sub.status_reason).toBe("download_accepted");
    const slot = await e.db
      .selectFrom("acquisition_slot")
      .selectAll()
      .where("id", "=", slotId)
      .executeTakeFirstOrThrow();
    expect(slot.status).toBe("satisfied");
    const hist = await e.api("GET", `/subscription-history?subscriptionId=${subId}`);
    expect(hist.body.data.items).toHaveLength(1);
    expect(hist.body.data.items[0]).toMatchObject({ reason: "download_accepted", mediaEntityId: entityId });
    // 进度 API:三集已覆盖,总集数已知
    const prog = await e.api("GET", `/subscriptions/${subId}/progress`);
    expect(prog.body.data.seasons[0]).toMatchObject({ episodeTotal: 3, totalKnown: true, complete: true });
    // 延后的另一版本:计划结束后重评,E02 已覆盖 → 不下载;三个种子各提交一次
    await new Promise((r) => setTimeout(r, 1500));
    expect(e.qb.callsTo("torrents/add")).toHaveLength(3);
  }, 300_000);
});