rootandClaude Opus 5.5
c9e8120fc6
- 新增 apps/server/test/support/{pg,engine}.ts:测试库地址、可用性探测、schema 咨询锁、
建 schema + 加锁迁移 + 删除、可选去重的假引擎;6 份 harness 改用它(acquisition-organize 与
file-manager 原先迁移不加锁,并行下有冲突隐患,一并修正)
- sse-pg 起子进程,并行下子进程启动超时,归入 serial
- 修正 b61335e 挪动 / 改名测试后遗留的 import 排序与格式(当时 lint 检查经管道被吞掉退出码)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
351 lines
17 KiB
TypeScript
351 lines
17 KiB
TypeScript
import { createQbClient } from "@umt/adapters";
|
||
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 {
|
||
applyEpisodeTotal,
|
||
subscriptionProgress,
|
||
sweepDueSlots,
|
||
} from "../../src/modules/acquisition-organize/services/subscriptions.ts";
|
||
import { createHarness, fakeDisk, type Harness, pgAvailable, seedBase, seedEntity, seedRelease } from "./harness.ts";
|
||
|
||
// W5-C 订阅增强集成:Slot×单元多在途 + 忙时延后重评 + 移除释放;季包按缺集选文件(mock qB files/filePrio);
|
||
// 总集数未知不结束(疑似缺集);once 完成写历史;pending 不完成;分集洗版只替换能提升的单元。
|
||
|
||
const hasPg = await pgAvailable();
|
||
const SIZE = 300_000;
|
||
|
||
describe.skipIf(!hasPg)("订阅增强(真实 PG + mock qB + 临时目录)", () => {
|
||
let h: Harness;
|
||
let qb: MockQb;
|
||
let lib: TempTree;
|
||
let cache: TempTree;
|
||
const disk = fakeDisk(10_000_000_000);
|
||
const torrents = new Map<string, Uint8Array>();
|
||
let base: Awaited<ReturnType<typeof seedBase>>;
|
||
|
||
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(10);
|
||
h = await createHarness("subscription", {
|
||
diskUsage: disk.fn,
|
||
torrentSource: async ({ release }) => ({ kind: "file", bytes: torrents.get(release.id)! }),
|
||
});
|
||
base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url });
|
||
}, 60_000);
|
||
|
||
afterAll(async () => {
|
||
await h?.close();
|
||
await qb?.close();
|
||
await lib?.cleanup();
|
||
await cache?.cleanup();
|
||
});
|
||
|
||
const db = () => h.db;
|
||
|
||
async function seriesSlot(
|
||
title: string,
|
||
opts: { mode?: "once" | "continuous"; status?: "active" | "pending"; hard?: object; upgrade?: object } = {},
|
||
) {
|
||
const entityId = await seedEntity(h.db, { kind: "series", title, year: 2024 });
|
||
const sub = await db()
|
||
.insertInto("subscription")
|
||
.values({
|
||
media_entity_id: entityId,
|
||
media_kind: "series",
|
||
mode: opts.mode ?? "continuous",
|
||
status: opts.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(opts.hard ?? { resolutions: ["1080p"] }),
|
||
upgrade_threshold: JSON.stringify(opts.upgrade ?? {}),
|
||
})
|
||
.returning("id")
|
||
.executeTakeFirstOrThrow();
|
||
return { entityId, subId: sub.id, slotId: slot.id };
|
||
}
|
||
|
||
async function release(entityId: string, title: string, files?: { path: string; length: number }[]) {
|
||
const t = files
|
||
? makeTorrent({ name: title, files, pieceSeed: Math.floor(Math.random() * 1e6) })
|
||
: makeTorrent({ name: `${title}.mkv`, length: SIZE, pieceSeed: Math.floor(Math.random() * 1e6) });
|
||
const id = await seedRelease(h.db, { siteId: base.siteId, entityId, title, size: SIZE, hash: t.hash });
|
||
torrents.set(id, t.bytes);
|
||
return { id, hash: t.hash, name: t.name };
|
||
}
|
||
|
||
const slotRow = (id: string) =>
|
||
db().selectFrom("acquisition_slot").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
|
||
const planRow = (id: string) =>
|
||
db().selectFrom("acquisition_plan").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
|
||
const subRow = (id: string) =>
|
||
db().selectFrom("subscription").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
|
||
|
||
async function planFor(relId: string) {
|
||
const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relId });
|
||
expect(ev.plans, JSON.stringify(ev)).toHaveLength(1);
|
||
return ev.plans[0]!;
|
||
}
|
||
async function submit(planId: string) {
|
||
const s = await submitDownload(h.d, planId);
|
||
expect(s.status).toBe("submitted");
|
||
return s.downloadTaskId!;
|
||
}
|
||
async function completeAndImport(hash: string, dtId: string) {
|
||
await qb.complete(hash, { writeFiles: true });
|
||
const sync = await syncDownloads(h.d);
|
||
expect(sync.completed).toContain(dtId);
|
||
const p = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" });
|
||
expect(p.status, JSON.stringify(p)).toBe("running");
|
||
const ex = await executeRecord(h.d, p.recordId!);
|
||
expect(ex.status).toBe("succeeded");
|
||
}
|
||
|
||
it("Slot×单元多在途:两集同时在途;重叠单元的发布延后;下载被移除即释放预留,重评后补上", async () => {
|
||
const { entityId, slotId } = await seriesSlot("熊家餐馆");
|
||
const e1 = await release(entityId, "The.Bear.S01E01.1080p.WEB-DL.x264-GRP");
|
||
const e2 = await release(entityId, "The.Bear.S01E02.1080p.WEB-DL.x264-GRP");
|
||
const p1 = await planFor(e1.id);
|
||
const p2 = await planFor(e2.id); // Slot 已 reserved,单元不相交 → 第二个在途计划
|
||
expect((await planRow(p1)).unit_keys).toEqual(["s1e1"]);
|
||
expect((await planRow(p2)).unit_keys).toEqual(["s1e2"]);
|
||
expect((await slotRow(slotId)).status).toBe("reserved");
|
||
|
||
// 同一集的另一个版本:单元在途 → 延后(deferred),不丢
|
||
const e2b = await release(entityId, "The.Bear.S01E02.1080p.WEB-DL.x265-OTHER");
|
||
const ev = await evaluateAcquisition(h.d, { sourceReleaseId: e2b.id });
|
||
expect(ev.plans).toHaveLength(0);
|
||
expect(ev.deferred).toEqual([{ sourceReleaseId: e2b.id, slotId, reason: "units_inflight" }]);
|
||
expect(
|
||
await db().selectFrom("slot_deferral").select(["unit_keys"]).where("slot_id", "=", slotId).execute(),
|
||
).toEqual([{ unit_keys: ["s1e2"] }]);
|
||
|
||
const dt1 = await submit(p1);
|
||
await submit(p2);
|
||
expect((await slotRow(slotId)).status).toBe("downloading");
|
||
await completeAndImport(e1.hash, dt1);
|
||
// E01 入库,但 E02 仍在途:Slot 显示在途阶段而不是静止态;分集账本登记 E01
|
||
const s1 = await slotRow(slotId);
|
||
expect(s1.status).toBe("downloading");
|
||
expect(s1.current_asset_id).not.toBeNull();
|
||
expect(s1.next_evaluate_at).not.toBeNull();
|
||
const ledger = await db()
|
||
.selectFrom("slot_unit as su")
|
||
.innerJoin("content_unit as u", "u.id", "su.content_unit_id")
|
||
.select(["u.episode_number", "su.plan_id"])
|
||
.where("su.slot_id", "=", slotId)
|
||
.execute();
|
||
expect(ledger).toEqual([{ episode_number: 1, plan_id: p1 }]);
|
||
|
||
// 用户在 qB 里删掉 E02:确认移除 → 计划失败释放单元 → Slot 到期重评 → 延后的 E02 另一版本得到计划
|
||
const qc = createQbClient({ instanceId: "t", baseUrl: qb.url, username: "admin", password: "adminadmin" });
|
||
await qc.delete([e2.hash], { deleteFiles: false });
|
||
await syncDownloads(h.d);
|
||
expect(await planRow(p2)).toMatchObject({ status: "failed", status_reason: "download_removed" });
|
||
expect((await slotRow(slotId)).status).toBe("attention");
|
||
h.engine.enqueued.length = 0;
|
||
expect(await sweepDueSlots(h.d)).toBeGreaterThanOrEqual(1);
|
||
expect(h.engine.enqueued).toContainEqual(
|
||
expect.objectContaining({ task: "evaluateAcquisition", input: { acquisitionSlotId: slotId } }),
|
||
);
|
||
const re = await evaluateAcquisition(h.d, { acquisitionSlotId: slotId });
|
||
expect(re.plans).toHaveLength(1);
|
||
expect(await planRow(re.plans[0]!)).toMatchObject({ source_release_id: e2b.id, unit_keys: ["s1e2"] });
|
||
// e2b 的延后已消化;被移除的原 E02 此时单元在途(e2b)→ 转为延后,等 e2b 结束后再判
|
||
const left = await db()
|
||
.selectFrom("slot_deferral")
|
||
.select("source_release_id")
|
||
.where("slot_id", "=", slotId)
|
||
.execute();
|
||
expect(left.map((x) => x.source_release_id)).toEqual([e2.id]);
|
||
// 候选评估记录带 trace / unitKeys
|
||
const art = await db()
|
||
.selectFrom("artifact")
|
||
.select("content")
|
||
.where("subject_id", "=", slotId)
|
||
.where("title", "=", "candidate_evaluation")
|
||
.execute();
|
||
const byRel = new Map(art.map((a) => [(a.content as { sourceReleaseId: string }).sourceReleaseId, a.content]));
|
||
expect(byRel.get(e2b.id)).toMatchObject({ decision: "accept", unitKeys: ["s1e2"], pri: 100 });
|
||
}, 60_000);
|
||
|
||
it("季包:总集数已知 → 只预留缺集;提交后读文件清单,已覆盖/在途的集文件优先级置 0", async () => {
|
||
const { entityId, slotId } = await seriesSlot("选集剧");
|
||
await db()
|
||
.transaction()
|
||
.execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 4, now: new Date() }));
|
||
const e1 = await release(entityId, "Pick.Show.S01E01.1080p.WEB-DL.x264-GRP");
|
||
await completeAndImport(e1.hash, await submit(await planFor(e1.id)));
|
||
const e2 = await release(entityId, "Pick.Show.S01E02.1080p.WEB-DL.x264-GRP");
|
||
const p2 = await planFor(e2.id); // E02 在途(未提交)
|
||
|
||
const packTitle = "Pick.Show.S01.1080p.WEB-DL.x264-GRP";
|
||
const files = [1, 2, 3, 4].map((e) => ({ path: `Pick.Show.S01E0${e}.1080p.WEB-DL.x264-GRP.mkv`, length: SIZE }));
|
||
const pack = await release(entityId, packTitle, [
|
||
...files,
|
||
{ path: "Pick.Show.S01E01.chs.srt", length: 100 },
|
||
{ path: "Pick.Show.S01E03.chs.srt", length: 100 },
|
||
]);
|
||
const pp = await planFor(pack.id);
|
||
expect((await planRow(pp)).unit_keys).toEqual(["s1e3", "s1e4"]);
|
||
const before = qb.callsTo("torrents/filePrio").length;
|
||
await submit(pp);
|
||
const prio = qb.callsTo("torrents/filePrio").slice(before);
|
||
expect(prio).toHaveLength(1);
|
||
const t = qb.get(pack.hash)!;
|
||
const off = t.files.filter((f) => f.priority === 0).map((f) => f.name.split("/").pop());
|
||
expect(off.sort()).toEqual(
|
||
[
|
||
"Pick.Show.S01E01.1080p.WEB-DL.x264-GRP.mkv",
|
||
"Pick.Show.S01E01.chs.srt",
|
||
"Pick.Show.S01E02.1080p.WEB-DL.x264-GRP.mkv",
|
||
].sort(),
|
||
);
|
||
expect((await planRow(pp)).file_selection).toMatchObject({ mode: "selected", selectedUnits: ["s1e3", "s1e4"] });
|
||
// 进度:E01 已覆盖,E02–E04 在途
|
||
const prog = await subscriptionProgress(db(), (await slotRow(slotId)).subscription_id!);
|
||
expect(prog!.seasons[0]).toMatchObject({ seasonNumber: 1, episodeTotal: 4, totalKnown: true, missing: [] });
|
||
expect(prog!.seasons[0]!.units.map((u) => u.state)).toEqual(["covered", "inflight", "inflight", "inflight"]);
|
||
expect(prog!.slots[0]!.inflightPlans.map((x) => x.planId).sort()).toEqual([p2, pp].sort());
|
||
|
||
// 认不出集号的整包:整包下载并记理由
|
||
const { entityId: e2Id } = await seriesSlot("整包剧");
|
||
const odd = await release(e2Id, "Odd.Show.S01.1080p.WEB-DL.x264-GRP", [
|
||
{ path: "disc1.mkv", length: SIZE },
|
||
{ path: "disc2.mkv", length: SIZE },
|
||
]);
|
||
const po = await planFor(odd.id);
|
||
expect((await planRow(po)).unit_keys).toEqual(["s1*"]);
|
||
const n0 = qb.callsTo("torrents/filePrio").length;
|
||
await submit(po);
|
||
expect(qb.callsTo("torrents/filePrio").length).toBe(n0);
|
||
expect((await planRow(po)).file_selection).toMatchObject({ mode: "whole", reason: "unmapped_video_file" });
|
||
}, 60_000);
|
||
|
||
it("总集数未知:入库后不结束、记疑似缺集;总集数就绪后补齐 → once 订阅完成并写历史", async () => {
|
||
const { entityId, slotId, subId } = await seriesSlot("未知集数剧", { mode: "once" });
|
||
for (const ep of [1, 3]) {
|
||
const r = await release(entityId, `Unknown.Total.S01E0${ep}.1080p.WEB-DL.x264-GRP`);
|
||
await completeAndImport(r.hash, await submit(await planFor(r.id)));
|
||
}
|
||
expect((await slotRow(slotId)).status).toBe("empty");
|
||
expect(await subRow(subId)).toMatchObject({ status: "active" });
|
||
const dec = await db()
|
||
.selectFrom("decision")
|
||
.selectAll()
|
||
.where("dedupe_key", "=", `suspected_missing:${slotId}`)
|
||
.executeTakeFirstOrThrow();
|
||
expect(dec.evidence).toMatchObject({ suspectedMissing: [{ season: 1, episodes: [2] }] });
|
||
|
||
await db()
|
||
.transaction()
|
||
.execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 3, now: new Date() }));
|
||
const r2 = await release(entityId, "Unknown.Total.S01E02.1080p.WEB-DL.x264-GRP");
|
||
const acceptedTask = await submit(await planFor(r2.id));
|
||
expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "download_accepted" });
|
||
expect((await slotRow(slotId)).status).toBe("downloading");
|
||
await completeAndImport(r2.hash, acceptedTask);
|
||
expect((await slotRow(slotId)).status).toBe("satisfied");
|
||
expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "download_accepted" });
|
||
const hist = await db()
|
||
.selectFrom("subscription_history")
|
||
.selectAll()
|
||
.where("subscription_id", "=", subId)
|
||
.execute();
|
||
expect(hist).toHaveLength(1);
|
||
expect(hist[0]).toMatchObject({ reason: "download_accepted", media_entity_id: entityId, mode: "once" });
|
||
expect((hist[0]!.progress as { seasons: { complete: boolean }[] }).seasons[0]!.complete).toBe(true);
|
||
// 历史 API + 重新订阅(改回 active)
|
||
const api = await h.api("GET", `/subscription-history?subscriptionId=${subId}`);
|
||
expect(api.body.data.items).toHaveLength(1);
|
||
const re = await h.api("PATCH", `/subscriptions/${subId}`, { status: "active" });
|
||
expect(re.body.data.status).toBe("active");
|
||
}, 90_000);
|
||
|
||
it("多个目标版本按各自已接收单元完成，单版本入库不提前完成订阅", async () => {
|
||
const { entityId, subId } = await seriesSlot("双版本剧", { mode: "once" });
|
||
await db()
|
||
.insertInto("acquisition_slot")
|
||
.values({
|
||
media_entity_id: entityId,
|
||
subscription_id: subId,
|
||
name: "720p",
|
||
hard_conditions: JSON.stringify({ resolutions: ["720p"] }),
|
||
})
|
||
.execute();
|
||
await db()
|
||
.transaction()
|
||
.execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 1, now: new Date() }));
|
||
const first = await release(entityId, "Two.Versions.S01E01.1080p.WEB-DL.x264-GRP");
|
||
const firstTask = await submit(await planFor(first.id));
|
||
expect((await subRow(subId)).status).toBe("active");
|
||
await completeAndImport(first.hash, firstTask);
|
||
expect((await subRow(subId)).status).toBe("active");
|
||
const second = await release(entityId, "Two.Versions.S01E01.720p.WEB-DL.x264-GRP");
|
||
await submit(await planFor(second.id));
|
||
expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "download_accepted" });
|
||
await qb.complete(second.hash, { writeFiles: true });
|
||
await syncDownloads(h.d);
|
||
}, 60_000);
|
||
|
||
it("pending(TMDB 未就绪)照常下载但不完成;暂停的订阅不评估", async () => {
|
||
const { entityId, slotId, subId } = await seriesSlot("待定剧", { mode: "once", status: "pending" });
|
||
await db()
|
||
.transaction()
|
||
.execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 1, now: new Date() }));
|
||
const r = await release(entityId, "Pending.Show.S01E01.1080p.WEB-DL.x264-GRP");
|
||
await completeAndImport(r.hash, await submit(await planFor(r.id)));
|
||
expect((await slotRow(slotId)).status).toBe("empty");
|
||
expect((await subRow(subId)).status).toBe("pending");
|
||
|
||
const paused = await seriesSlot("暂停剧", { status: "active" });
|
||
await db().updateTable("subscription").set({ status: "paused" }).where("id", "=", paused.subId).execute();
|
||
const pr = await release(paused.entityId, "Paused.Show.S01E01.1080p.WEB-DL.x264-GRP");
|
||
const ev = await evaluateAcquisition(h.d, { sourceReleaseId: pr.id });
|
||
expect(ev).toMatchObject({ evaluated: 0, plans: [] });
|
||
}, 60_000);
|
||
|
||
it("分集洗版:只替换能提升的单元;同档不换(trace 可读)", async () => {
|
||
const { entityId, slotId } = await seriesSlot("洗版剧", {
|
||
hard: { resolutions: ["720p", "1080p"] },
|
||
upgrade: { enabled: true, allowResolutionUpgrade: true },
|
||
});
|
||
const r = await release(entityId, "Wash.Show.S01E01.720p.WEB-DL.x264-GRP");
|
||
await completeAndImport(r.hash, await submit(await planFor(r.id)));
|
||
const same = await release(entityId, "Wash.Show.S01E01.720p.WEB-DL.x264-OTHER");
|
||
const ev = await evaluateAcquisition(h.d, { sourceReleaseId: same.id });
|
||
expect(ev.plans).toHaveLength(0);
|
||
const up = await release(entityId, "Wash.Show.S01E01-E02.1080p.WEB-DL.x264-GRP");
|
||
// E02 还缺 → 追更(fill);只有 E01 覆盖时的纯升级场景见下一条
|
||
const fill = await planFor(up.id);
|
||
expect(await planRow(fill)).toMatchObject({ purpose: "fill", unit_keys: ["s1e2"] });
|
||
await db().updateTable("acquisition_plan").set({ status: "cancelled" }).where("id", "=", fill).execute();
|
||
await db().updateTable("acquisition_slot").set({ status: "empty" }).where("id", "=", slotId).execute();
|
||
const up1 = await release(entityId, "Wash.Show.S01E01.1080p.WEB-DL.x264-GRP");
|
||
const pu = await planFor(up1.id);
|
||
expect(await planRow(pu)).toMatchObject({ purpose: "upgrade", unit_keys: ["s1e1"] });
|
||
const art = await db()
|
||
.selectFrom("artifact")
|
||
.select("content")
|
||
.where("subject_id", "=", slotId)
|
||
.where("title", "=", "candidate_evaluation")
|
||
.execute();
|
||
const sameEval = art
|
||
.map((a) => a.content as { sourceReleaseId: string; trace?: string[] })
|
||
.find((a) => a.sourceReleaseId === same.id);
|
||
expect(sameEval?.trace?.[0]).toBe("与当前版本同档,不反复更换");
|
||
}, 60_000);
|
||
});
|