UHD_Media_Tool/apps/server/test/acquisition-organize/background-loop.test.ts

import { sql } from "@umt/db";
import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
import { submitDownload, syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
import { createSchedulesModule } from "../../src/modules/schedules/index.ts";
import {
  createHarness,
  fakeDisk,
  type Harness,
  pgAvailable,
  seedBase,
  seedEntity,
  seedRelease,
  seedSlot,
} from "./harness.ts";

// W4-D P0:后台循环只编排(W7-C 起由 schedules 调度器驱动模块登记的定时任务)。补入库在巡检里只入队 verifyAndImport;一个 50MB 的入库(复制 + sha256 + 慢 ffprobe)
// 进行中时,qB 同步定时器仍按周期打到 mock qB(用 mock qB 的请求时间戳断言)。

const hasPg = await pgAvailable();
const SIZE = 50 * 1024 * 1024;
const SYNC_MS = 200;
const PROBE_DELAY_MS = 2500;

describe.skipIf(!hasPg)("acquisition-organize 后台循环不被入库阻塞", () => {
  let h: Harness;
  let qb: MockQb;
  let lib: TempTree;
  let cache: TempTree;
  const disk = fakeDisk(10 * 1024 ** 3);
  const torrents = new Map<string, Uint8Array>();
  let base: Awaited<ReturnType<typeof seedBase>>;
  const schedules = createSchedulesModule({ builtinChecks: false, scheduler: { tickMs: 50, lockRetryMs: 100 } });

  beforeAll(async () => {
    qb = await createMockQb({ addVisibleAfterPolls: 0 });
    lib = await createTempTree({}, { sentinel: true, prefix: "umt-lib-" });
    cache = await createTempTree({}, { prefix: "umt-cache-" });
    disk.state.watch = [cache.root];
    disk.setPct(40);
    h = await createHarness("bgloop", {
      diskUsage: disk.fn,
      torrentSource: async ({ release }) => ({ kind: "file", bytes: torrents.get(release.id)! }),
      // 慢 ffprobe:让一次入库稳定地持续数秒
      probe: async () => {
        await new Promise((r) => setTimeout(r, PROBE_DELAY_MS));
        return { status: "unknown", reason: "ffprobe_unavailable" };
      },
      options: { backgroundLoops: true, syncIntervalMs: SYNC_MS, planIntervalMs: SYNC_MS, reconcileIntervalMs: 0 },
    });
    base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url });
    await h.setSettings("retention", { seedRetentionDays: 0 });
  }, 60_000);

  afterAll(async () => {
    await schedules.stop?.();
    await h?.module.stop?.();
    await h?.close();
    await qb?.close();
    await lib?.cleanup();
    await cache?.cleanup();
  });

  it("巡检只入队 verifyAndImport;50MB 入库进行中 qB 同步仍按时执行", async () => {
    const entityId = await seedEntity(h.db, { kind: "movie", title: "长片", year: 2023 });
    await seedSlot(h.db, { entityId });
    const t = makeTorrent({ name: "Long.Movie.2023.1080p.BluRay.x264-GRP.mkv", length: SIZE, pieceSeed: 7 });
    const relId = await seedRelease(h.db, {
      siteId: base.siteId,
      entityId,
      title: "Long.Movie.2023.1080p.BluRay.x264-GRP",
      size: SIZE,
      hash: t.hash,
    });
    torrents.set(relId, t.bytes);
    const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relId });
    const sub = await submitDownload(h.d, ev.plans[0]!);
    expect(sub.status).toBe("submitted");
    await qb.complete(t.hash, { writeFiles: true });
    expect((await syncDownloads(h.d)).completed).toContain(sub.downloadTaskId);
    const dtId = sub.downloadTaskId!;
    // 完成超过 5 分钟且没有整理记录 → 巡检对象
    await sql`update download_task set completed_at = now() - interval '10 minutes' where id = ${dtId}`.execute(h.db);

    await h.module.start?.(h.ctx);
    await schedules.start?.(h.ctx);
    const deadline = Date.now() + 10_000;
    while (!h.engine.enqueued.some((e) => e.task === "verifyAndImport") && Date.now() < deadline) {
      await new Promise((r) => setTimeout(r, 50));
    }
    const enq = h.engine.enqueued.find((e) => e.task === "verifyAndImport");
    expect(enq?.input).toMatchObject({ downloadTaskId: dtId });
    expect(enq?.key).toMatch(new RegExp(`^verifyAndImport:import:${dtId}:\\d+$`));
    // 循环没有自己执行入库
    const recs = await h.db
      .selectFrom("organization_record")
      .select("id")
      .where("download_task_id", "=", dtId)
      .execute();
    expect(recs).toHaveLength(0);
    // 同一时间槽重复巡检用同一个引擎幂等键(真实引擎去重;假引擎只记录)
    await new Promise((r) => setTimeout(r, SYNC_MS * 3));
    const keys = new Set(h.engine.enqueued.filter((e) => e.task === "verifyAndImport").map((e) => e.key));
    expect(keys.size).toBe(1);

    // 模拟 worker 执行这次 verifyAndImport(同进程,与同步循环并发)
    const t0 = Date.now();
    const plan = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" });
    expect(plan.status).toBe("running");
    const ex = await executeRecord(h.d, plan.recordId!);
    const t1 = Date.now();
    expect(ex).toMatchObject({ status: "succeeded", succeeded: 1 });
    expect(t1 - t0).toBeGreaterThanOrEqual(PROBE_DELAY_MS);

    const stamps = qb
      .callsTo("torrents/info")
      .map((c) => c.at)
      .filter((at) => at >= t0 && at <= t1);
    // 每 200ms 一轮(同步本身几十 ms);入库期间至少有 1/3 的轮次、最大间隔不超过 1s
    expect(stamps.length).toBeGreaterThanOrEqual(Math.floor((t1 - t0) / SYNC_MS / 3));
    const gaps = [t0, ...stamps, t1].slice(1).map((v, i) => v - [t0, ...stamps][i]!);
    expect(Math.max(...gaps)).toBeLessThan(1000);
  }, 90_000);
});