UHD_Media_Tool/apps/server/test/acquisition-organize/blocked-reevaluate.test.ts

import { createMockQb, type MockQb } from "@umt/testkit";
import { sql } from "kysely";
import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
import {
  completeSubscription,
  GATE_RETRY_BASE_MS,
  sweepDueSlots,
} from "../../src/modules/acquisition-organize/services/subscriptions.ts";
import { createHarness, fakeDisk, type Harness, pgAvailable, seedBase, seedEntity, seedRelease } from "./harness.ts";

// 获取环节被挡住 / 失败后的重评(T1-3):闸门不过写延后并排重评(电影、已满足剧集收录版本);
// 并发预算不计 stalled;失败 / 取消的旧计划不挡同一资源重建计划;订阅完成 / 资源过期清延后行。

const hasPg = await pgAvailable();
const SIZE = 300_000_000;

describe.skipIf(!hasPg)("被挡住与失败后的重评", () => {
  let h: Harness;
  let qb: MockQb;
  let base: Awaited<ReturnType<typeof seedBase>>;
  const disk = fakeDisk(10_000_000_000);

  beforeAll(async () => {
    qb = await createMockQb({ addVisibleAfterPolls: 0 });
    disk.setPct(10);
    h = await createHarness("blocked_reevaluate", { diskUsage: disk.fn });
    base = await seedBase(h.db, { libraryPath: "/tmp/umt-none-lib", cachePath: "/tmp/umt-none-cache", qbUrl: qb.url });
  }, 60_000);

  afterAll(async () => {
    await h?.close();
    await qb?.close();
  });

  const db = () => h.db;
  // 用例之间不共享在途:上一个用例留下的预留计划会占并发预算
  beforeEach(async () => {
    await db().updateTable("acquisition_plan").set({ status: "cancelled" }).where("status", "=", "reserved").execute();
  });
  const hash = () => Math.random().toString(16).slice(2).padEnd(40, "0").slice(0, 40);

  async function movieSlot(title: string) {
    const entityId = await seedEntity(db(), { kind: "movie", title, year: 2024 });
    const sub = await db()
      .insertInto("subscription")
      .values({ media_entity_id: entityId, media_kind: "movie", mode: "once", status: "active" })
      .returning("id")
      .executeTakeFirstOrThrow();
    const slot = await db()
      .insertInto("acquisition_slot")
      .values({
        media_entity_id: entityId,
        subscription_id: sub.id,
        name: "1080p",
        hard_conditions: JSON.stringify({ resolutions: ["1080p"] }),
        upgrade_threshold: JSON.stringify({}),
      })
      .returning("id")
      .executeTakeFirstOrThrow();
    return { entityId, subId: sub.id, slotId: slot.id };
  }

  async function seriesSlot(title: string, status: "empty" | "satisfied" = "empty") {
    const entityId = await seedEntity(db(), { kind: "series", title, year: 2024 });
    const sub = await db()
      .insertInto("subscription")
      .values({ media_entity_id: entityId, media_kind: "series", mode: "continuous", status: "active" })
      .returning("id")
      .executeTakeFirstOrThrow();
    const slot = await db()
      .insertInto("acquisition_slot")
      .values({
        media_entity_id: entityId,
        subscription_id: sub.id,
        name: "1080p",
        status,
        hard_conditions: JSON.stringify({ resolutions: ["1080p"] }),
        upgrade_threshold: JSON.stringify({}),
      })
      .returning("id")
      .executeTakeFirstOrThrow();
    return { entityId, subId: sub.id, slotId: slot.id };
  }

  const release = (entityId: string, title: string) =>
    seedRelease(db(), { siteId: base.siteId, entityId, title, size: SIZE, hash: hash() });
  const deferrals = (slotId: string) =>
    db().selectFrom("slot_deferral").selectAll().where("slot_id", "=", slotId).execute();
  const slotRow = (slotId: string) =>
    db().selectFrom("acquisition_slot").selectAll().where("id", "=", slotId).executeTakeFirstOrThrow();
  const candidate = async (slotId: string, releaseId: string) =>
    (
      await sql<{ content: { decision: string; reasons: string[] } }>`
        select content from artifact where title = 'candidate_evaluation' and subject_id = ${slotId}
           and content->>'sourceReleaseId' = ${releaseId}`.execute(db())
    ).rows[0]?.content;
  /** 模拟计划结束后收录版本释放(applySlotEvent release 的效果) */
  const releaseSlot = (slotId: string) =>
    db()
      .updateTable("acquisition_slot")
      .set({ status: "empty", reserved_plan_id: null })
      .where("id", "=", slotId)
      .execute();

  /** 缓存卷超阈值挡住 → 延后 + 排重评;恢复后到期重评能建计划 */
  async function gateThenRecover(slotId: string, releaseId: string) {
    disk.setPct(95);
    const t0 = Date.now();
    const ev = await evaluateAcquisition(h.d, { sourceReleaseId: releaseId });
    expect(ev.plans).toHaveLength(0);
    expect(ev.skipped[0]?.reason).toMatch(/^cache:/);
    expect(ev.deferred).toContainEqual({ sourceReleaseId: releaseId, slotId, reason: "gate:cache" });
    const rows = await deferrals(slotId);
    expect(rows.map((r) => [r.source_release_id, r.reason])).toEqual([[releaseId, "gate:cache"]]);
    expect((await candidate(slotId, releaseId))?.decision).toBe("deferred");
    const due = (await slotRow(slotId)).next_evaluate_at;
    expect(due).not.toBeNull();
    const delay = due!.getTime() - t0;
    expect(delay).toBeGreaterThan(GATE_RETRY_BASE_MS - 60_000);
    expect(delay).toBeLessThanOrEqual(GATE_RETRY_BASE_MS + 60_000);

    // 恢复后到期:巡检入队按收录版本重评,先评延后的资源
    disk.setPct(10);
    await db()
      .updateTable("acquisition_slot")
      .set({ next_evaluate_at: new Date(Date.now() - 1000) })
      .where("id", "=", slotId)
      .execute();
    h.engine.enqueued.length = 0;
    expect(await sweepDueSlots(h.d)).toBeGreaterThanOrEqual(1);
    const job = h.engine.enqueued.find(
      (e) =>
        e.task === "evaluateAcquisition" && (e.input as { acquisitionSlotId?: string }).acquisitionSlotId === slotId,
    );
    expect(job).toBeDefined();
    const ev2 = await evaluateAcquisition(h.d, job!.input as { acquisitionSlotId: string });
    expect(ev2.plans, JSON.stringify(ev2)).toHaveLength(1);
    expect(await deferrals(slotId)).toHaveLength(0);
    expect((await candidate(slotId, releaseId))?.decision).toBe("accept");
  }

  it("电影收录版本:缓存卷超阈值写延后 gate:cache 并排重评,恢复后重评建计划", async () => {
    const { entityId, slotId } = await movieSlot("Gate Movie");
    const rel = await release(entityId, "Gate.Movie.2024.1080p.WEB-DL.x264-GRP");
    await gateThenRecover(slotId, rel);
  });

  it("已满足的剧集收录版本:新集被挡住同样写延后并重评", async () => {
    const { entityId, slotId } = await seriesSlot("Gate Series", "satisfied");
    const rel = await release(entityId, "Gate.Series.S01E05.1080p.WEB-DL.x264-GRP");
    await gateThenRecover(slotId, rel);
  });

  it("再次被挡住按次数退避(第 2 次 60 分钟)", async () => {
    const { entityId, slotId } = await movieSlot("Backoff Movie");
    const rel = await release(entityId, "Backoff.Movie.2024.1080p.WEB-DL.x264-GRP");
    await h.setSettings("automation", { enabled: false });
    try {
      await evaluateAcquisition(h.d, { sourceReleaseId: rel });
      await db().updateTable("acquisition_slot").set({ next_evaluate_at: null }).where("id", "=", slotId).execute();
      const t0 = Date.now();
      await evaluateAcquisition(h.d, { sourceReleaseId: rel });
      const [row] = await deferrals(slotId);
      expect(row).toMatchObject({ reason: "gate:automation", attempts: 2 });
      const delay = (await slotRow(slotId)).next_evaluate_at!.getTime() - t0;
      expect(delay).toBeGreaterThan(2 * GATE_RETRY_BASE_MS - 60_000);
      expect(delay).toBeLessThanOrEqual(2 * GATE_RETRY_BASE_MS + 60_000);
    } finally {
      await h.setSettings("automation", { enabled: true });
    }
  });

  it("预算并发:stalled 任务不占并发名额,在跑的任务占", async () => {
    await h.setSettings("budget", { maxConcurrentDownloads: 1 });
    try {
      const busyTask = await db()
        .insertInto("download_task")
        .values({
          instance_id: base.instanceId,
          torrent_hash: hash(),
          name: "stalled-one",
          size_bytes: 1_000_000,
          status: "stalled",
          managed: true,
        })
        .returning("id")
        .executeTakeFirstOrThrow();
      const a = await movieSlot("Budget Movie A");
      const relA = await release(a.entityId, "Budget.Movie.A.2024.1080p.WEB-DL.x264-GRP");
      const evA = await evaluateAcquisition(h.d, { sourceReleaseId: relA });
      expect(evA.plans, JSON.stringify(evA)).toHaveLength(1);
      // 刚预留、还没建下载任务的计划算一个名额
      const b = await movieSlot("Budget Movie B");
      const relB = await release(b.entityId, "Budget.Movie.B.2024.1080p.WEB-DL.x264-GRP");
      const evB = await evaluateAcquisition(h.d, { sourceReleaseId: relB });
      expect(evB.plans).toHaveLength(0);
      expect(evB.skipped[0]?.reason).toBe("budget_exceeded");
      expect((await deferrals(b.slotId))[0]?.reason).toBe("gate:budget");
      // A 结束、stalled 任务变成下载中:仍占满
      await db().updateTable("acquisition_plan").set({ status: "cancelled" }).where("id", "=", evA.plans[0]!).execute();
      await db().updateTable("download_task").set({ status: "downloading" }).where("id", "=", busyTask.id).execute();
      expect((await evaluateAcquisition(h.d, { sourceReleaseId: relB })).plans).toHaveLength(0);
      await db().updateTable("download_task").set({ status: "stalled" }).where("id", "=", busyTask.id).execute();
      const evB2 = await evaluateAcquisition(h.d, { sourceReleaseId: relB });
      expect(evB2.plans).toHaveLength(1);
      expect(await deferrals(b.slotId)).toHaveLength(0);
      await db().deleteFrom("download_task").where("id", "=", busyTask.id).execute();
      await db()
        .updateTable("acquisition_plan")
        .set({ status: "cancelled" })
        .where("id", "=", evB2.plans[0]!)
        .execute();
    } finally {
      await h.setSettings("budget", {});
    }
  });

  it("计划失败后同一资源能重建计划(冷却后);在途 / 已完成 / 用户移除的仍挡住", async () => {
    const { entityId, slotId } = await movieSlot("Retry Movie");
    const rel = await release(entityId, "Retry.Movie.2024.1080p.WEB-DL.x264-GRP");
    const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev1.plans, JSON.stringify(ev1)).toHaveLength(1);
    const p1 = ev1.plans[0]!;
    // 失败时间 = updated_at;触发器会把它改成 now(),回拨时在事务里临时停用本 schema 的触发器
    const fail = (id: string, reason: string, agoMs: number) =>
      db()
        .transaction()
        .execute(async (trx) => {
          await sql`alter table acquisition_plan disable trigger user`.execute(trx);
          await trx
            .updateTable("acquisition_plan")
            .set({ status: "failed", status_reason: reason, updated_at: new Date(Date.now() - agoMs) })
            .where("id", "=", id)
            .execute();
          await sql`alter table acquisition_plan enable trigger user`.execute(trx);
        });

    // 刚失败:冷却中 → 延后 retry_cooldown,排到冷却结束
    await fail(p1, "source_failed: cookie expired", 0);
    await releaseSlot(slotId);
    const ev2 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev2.plans).toHaveLength(0);
    expect(ev2.deferred).toContainEqual({ sourceReleaseId: rel, slotId, reason: "retry_cooldown" });
    expect((await deferrals(slotId))[0]?.reason).toBe("retry_cooldown");
    expect((await slotRow(slotId)).next_evaluate_at).not.toBeNull();

    // 冷却过后:建新计划,旧计划的幂等键让位
    await fail(p1, "source_failed: cookie expired", GATE_RETRY_BASE_MS + 60_000);
    const ev3 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev3.plans, JSON.stringify(ev3)).toHaveLength(1);
    const p2 = ev3.plans[0]!;
    expect(p2).not.toBe(p1);
    const keys = await db()
      .selectFrom("acquisition_plan")
      .select(["id", "idempotency_key"])
      .where("id", "in", [p1, p2])
      .execute();
    expect(keys.find((k) => k.id === p2)?.idempotency_key).toBe(`acq:${slotId}:${rel}`);
    expect(keys.find((k) => k.id === p1)?.idempotency_key).toBe(`acq:${slotId}:${rel}:retired:${p1}`);
    expect(await deferrals(slotId)).toHaveLength(0);

    // 第二次失败:冷却翻倍(30 分钟前失败仍在冷却)
    await fail(p2, "rejected", GATE_RETRY_BASE_MS + 60_000);
    await releaseSlot(slotId);
    expect((await evaluateAcquisition(h.d, { sourceReleaseId: rel })).plans).toHaveLength(0);
    await fail(p2, "rejected", 2 * GATE_RETRY_BASE_MS + 60_000);
    const ev4 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev4.plans).toHaveLength(1);
    const p3 = ev4.plans[0]!;

    // 已完成:挡住
    await db().updateTable("acquisition_plan").set({ status: "completed" }).where("id", "=", p3).execute();
    await releaseSlot(slotId);
    const ev5 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev5.plans).toHaveLength(0);
    expect((await candidate(slotId, rel))?.reasons).toContain("slot_changed_or_reserved");

    // 用户在下载器里移除:不自动重下
    await fail(p3, "download_removed", 24 * 3600_000);
    const ev6 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev6.plans).toHaveLength(0);
  });

  it("取消的旧计划:重置订阅取消的立即让位,用户手动取消的仍挡住", async () => {
    const { entityId, slotId } = await movieSlot("Cancel Movie");
    const rel = await release(entityId, "Cancel.Movie.2024.1080p.WEB-DL.x264-GRP");
    const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev1.plans).toHaveLength(1);
    const cancel = (id: string, reason: string) =>
      db()
        .updateTable("acquisition_plan")
        .set({ status: "cancelled", status_reason: reason })
        .where("id", "=", id)
        .execute();
    await cancel(ev1.plans[0]!, "subscription_reset");
    await releaseSlot(slotId);
    const ev2 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev2.plans, JSON.stringify(ev2)).toHaveLength(1);
    await cancel(ev2.plans[0]!, "cancelled_by_user");
    await releaseSlot(slotId);
    expect((await evaluateAcquisition(h.d, { sourceReleaseId: rel })).plans).toHaveLength(0);
  });

  it("同一种子已被别的计划占用而取消的不让位(避免取消 → 重评 → 再撞上同一任务的空转)", async () => {
    const { entityId, slotId } = await movieSlot("InUse Movie");
    const rel = await release(entityId, "InUse.Movie.2024.1080p.WEB-DL.x264-GRP");
    const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev1.plans).toHaveLength(1);
    await db()
      .updateTable("acquisition_plan")
      .set({ status: "cancelled", status_reason: "torrent_in_use_by_plan:00000000-0000-0000-0000-000000000000" })
      .where("id", "=", ev1.plans[0]!)
      .execute();
    await releaseSlot(slotId);
    expect((await evaluateAcquisition(h.d, { sourceReleaseId: rel })).plans).toHaveLength(0);
  });

  it("在途计划(已提交)仍挡住同一资源", async () => {
    const { entityId } = await seriesSlot("Inflight Series");
    const rel = await release(entityId, "Inflight.Series.S01E01.1080p.WEB-DL.x264-GRP");
    const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev1.plans, JSON.stringify(ev1)).toHaveLength(1);
    await db().updateTable("acquisition_plan").set({ status: "submitted" }).where("id", "=", ev1.plans[0]!).execute();
    const ev2 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
    expect(ev2.plans).toHaveLength(0);
    expect(ev2.deferred[0]?.reason).toBe("units_inflight");
  });

  it("订阅完成、订阅暂停、资源过期时清掉延后行", async () => {
    disk.setPct(95);
    try {
      // 完成
      const a = await movieSlot("Cleanup Movie A");
      const relA = await release(a.entityId, "Cleanup.Movie.A.2024.1080p.WEB-DL.x264-GRP");
      await evaluateAcquisition(h.d, { sourceReleaseId: relA });
      expect(await deferrals(a.slotId)).toHaveLength(1);
      const subA = await db()
        .selectFrom("subscription")
        .selectAll()
        .where("id", "=", a.subId)
        .executeTakeFirstOrThrow();
      expect(await completeSubscription(h.d, db(), subA, "test")).toBe(true);
      expect(await deferrals(a.slotId)).toHaveLength(0);

      // 暂停:到期重评时跳过并清掉
      const b = await movieSlot("Cleanup Movie B");
      const relB = await release(b.entityId, "Cleanup.Movie.B.2024.1080p.WEB-DL.x264-GRP");
      await evaluateAcquisition(h.d, { sourceReleaseId: relB });
      expect(await deferrals(b.slotId)).toHaveLength(1);
      await db().updateTable("subscription").set({ status: "paused" }).where("id", "=", b.subId).execute();
      await evaluateAcquisition(h.d, { acquisitionSlotId: b.slotId });
      expect(await deferrals(b.slotId)).toHaveLength(0);

      // 资源过期
      const c = await movieSlot("Cleanup Movie C");
      const relC = await release(c.entityId, "Cleanup.Movie.C.2024.1080p.WEB-DL.x264-GRP");
      await evaluateAcquisition(h.d, { sourceReleaseId: relC });
      expect(await deferrals(c.slotId)).toHaveLength(1);
      await db().updateTable("source_release").set({ availability: "expired" }).where("id", "=", relC).execute();
      const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relC });
      expect(ev.skipped[0]?.reason).toBe("release_expired");
      expect(await deferrals(c.slotId)).toHaveLength(0);
    } finally {
      disk.setPct(10);
    }
  });
});