Merge branch 'dl2'

Co-Authored-By: Claude Fable 5.1 <[email protected]>

# Conflicts:
#	apps/web/app/routes/console/downloads.tsx
#	docs/plan/00-first-version-plan.md
This commit is contained in:
root committed 2026-09-29 21:45:15 +08:00
commit 8315ece4d1
17 files changed
+1113 -93

No files matched your search

@@ -0,0 +1,197 @@
import { createMockQb, type MockQb } from "@umt/testkit";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { syncDownloads } from "../services/downloads.ts";
import { createHarness, type Harness, pgAvailable, seedEntity } from "./harness.ts";
// DL-2 下载面板:进行中 / 已完成分组与计数、每个种子的实时速度 / 剩余时间 / 连接数、下载器总速度、
// 实时数据短 TTL 缓存(轮询不直打 qB)、下载器不可达时退化为只看数据库。
const hasPg = await pgAvailable();
const TTL = 400;
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const h40 = (n: number) => n.toString(16).padStart(40, "b");
describe.skipIf(!hasPg)("DL-2 下载面板(真实 PG + mock qB)", () => {
let h: Harness;
let qb: MockQb;
let instId: string;
beforeAll(async () => {
qb = await createMockQb({ addVisibleAfterPolls: 0 });
h = await createHarness("dlboard", { options: { liveTtlMs: TTL } });
instId = (
await h.db
.insertInto("downloader_instance")
.values({ name: "qb-main", base_url: qb.url, username: "admin", settings: JSON.stringify({}) })
.returning("id")
.executeTakeFirstOrThrow()
).id;
// 进行中:下载中(有速度与 eta)、暂停、排队、出错;已完成:做种、完成暂停;管理标签外的任务不纳入
const dl = qb.seed({ hash: h40(1), name: "Dune.Part.Two.2024.2160p", tags: ["umt"], progress: 0.4 });
Object.assign(dl, {
dlspeed: 12_900_000,
upspeed: 1_200_000,
eta: 620,
num_seeds: 12,
num_leechs: 3,
num_complete: 80,
num_incomplete: 9,
});
qb.seed({ hash: h40(2), name: "Paused.Show.S01", tags: ["umt"], progress: 0.2, state: "pausedDL" });
qb.seed({ hash: h40(3), name: "Queued.Movie.2020", tags: ["umt"], progress: 0, state: "queuedDL" });
qb.seed({ hash: h40(4), name: "Broken.Movie.2019", tags: ["umt"], progress: 0.1, state: "error" });
const seeding = qb.seed({ hash: h40(5), name: "Heat.1995.1080p", tags: ["umt"], progress: 1 });
Object.assign(seeding, { upspeed: 300_000, ratio: 1.25 });
qb.seed({ hash: h40(6), name: "Done.Paused.2018", tags: ["umt"], progress: 1, state: "pausedUP" });
const other = qb.seed({ hash: h40(7), name: "Not.Managed", tags: ["other"], progress: 0.5 });
other.dlspeed = 100_000;
await syncDownloads(h.d);
const heat = await seedEntity(h.db, { kind: "movie", title: "盗火线", year: 1995 });
const t5 = await h.db
.updateTable("download_task")
.set({ media_entity_id: heat })
.where("torrent_hash", "=", h40(5))
.returning("id")
.executeTakeFirstOrThrow();
await h.db
.insertInto("organization_record")
.values({
trigger: "auto",
actor_type: "system",
action: "copy",
status: "succeeded",
download_task_id: t5.id,
media_entity_id: heat,
})
.execute();
// 完成暂停的那个:最近一次整理失败 + 未关闭的待处理事项(下载任务维度)
const t6 = await h.db
.selectFrom("download_task")
.select("id")
.where("torrent_hash", "=", h40(6))
.executeTakeFirstOrThrow();
await h.db
.insertInto("organization_record")
.values({
trigger: "auto",
actor_type: "system",
action: "copy",
status: "failed",
reason: "identity_unknown",
download_task_id: t6.id,
})
.execute();
await h.db
.insertInto("decision")
.values({
kind: "identity_unresolved",
title: "没识别出作品",
subject_type: "download_task",
subject_id: t6.id,
dedupe_key: `test:${t6.id}`,
})
.execute();
}, 60_000);
afterAll(async () => {
await h?.close();
await qb?.close();
});
it("默认进行中:只含未完成任务,两组各带计数,附实时速度与剩余时间", async () => {
await sleep(TTL + 50);
const r = await h.api("GET", "/download-tasks/board");
expect(r.status).toBe(200);
const b = r.body.data;
expect(b.counts).toEqual({ active: 4, completed: 2 });
const names = b.items.map((i: { name: string }) => i.name).sort();
expect(names).toEqual(["Broken.Movie.2019", "Dune.Part.Two.2024.2160p", "Paused.Show.S01", "Queued.Movie.2020"]);
const dune = b.items.find((i: { torrentHash: string }) => i.torrentHash === h40(1));
expect(dune.live).toMatchObject({
dlSpeed: 12_900_000,
upSpeed: 1_200_000,
etaSec: 620,
seeds: 12,
leechs: 3,
seedsTotal: 80,
leechsTotal: 9,
rawState: "downloading",
});
expect(dune.live.progress).toBeCloseTo(0.4);
// eta 8640000(qB 的无穷)记为 null;num_complete -1 记为未知
const queued = b.items.find((i: { torrentHash: string }) => i.torrentHash === h40(3));
expect(queued.live.etaSec).toBeNull();
expect(queued.live.seedsTotal).toBeNull();
expect(queued.live.dlSpeed).toBe(0);
// 总速度来自 transfer/info:整个下载器(含管理标签外的任务)
expect(b.transfer.dlSpeed).toBe(12_900_000 + 100_000);
expect(b.transfer.upSpeed).toBe(1_200_000 + 300_000);
expect(b.transfer.instances).toEqual([
{ instanceId: instId, name: "qb-main", reachable: true, dlSpeed: 13_000_000, upSpeed: 1_500_000 },
]);
expect(b.liveAt).toBeTruthy();
});
it("已完成:含做种与完成后暂停,带作品与最近整理记录", async () => {
const b = (await h.api("GET", "/download-tasks/board?state=completed")).body.data;
expect(b.counts).toEqual({ active: 4, completed: 2 });
expect(b.items).toHaveLength(2);
const heat = b.items.find((i: { torrentHash: string }) => i.torrentHash === h40(5));
expect(heat.status).toBe("completed");
expect(heat.media).toMatchObject({ title: "盗火线", year: 1995 });
expect(heat.organize).toMatchObject({ status: "succeeded", reason: null, decision: null });
expect(heat.live).toMatchObject({ upSpeed: 300_000, ratio: 1.25, etaSec: null });
expect(heat.live.completedAt).toBeTruthy();
const paused = b.items.find((i: { torrentHash: string }) => i.torrentHash === h40(6));
expect(paused.organize).toMatchObject({
status: "failed",
reason: "identity_unknown",
decision: { kind: "identity_unresolved" },
});
expect(paused.media).toBeNull();
});
it("TTL 内的轮询共用一次 qB 查询,并发请求合并;过期后重新查询", async () => {
await sleep(TTL + 50);
const before = qb.callsTo("torrents/info").length;
const beforeT = qb.callsTo("transfer/info").length;
await Promise.all([
h.api("GET", "/download-tasks/board"),
h.api("GET", "/download-tasks/board?state=completed"),
h.api("GET", "/download-tasks/board"),
]);
await h.api("GET", "/download-tasks/board");
expect(qb.callsTo("torrents/info").length - before).toBe(1);
expect(qb.callsTo("transfer/info").length - beforeT).toBe(1);
// 缓存期间速度变化不可见,过期后可见
const dl = qb.get(h40(1));
if (dl) dl.dlspeed = 5_000_000;
const cached = (await h.api("GET", "/download-tasks/board")).body.data;
expect(cached.items.find((i: { torrentHash: string }) => i.torrentHash === h40(1)).live.dlSpeed).toBe(12_900_000);
await sleep(TTL + 50);
const fresh = (await h.api("GET", "/download-tasks/board")).body.data;
expect(fresh.items.find((i: { torrentHash: string }) => i.torrentHash === h40(1)).live.dlSpeed).toBe(5_000_000);
expect(qb.callsTo("torrents/info").length - before).toBe(2);
});
it("下载器不可达:列表照常返回,实时数据为空,实例标记不可达", async () => {
await sleep(TTL + 50);
qb.fault("torrents/info", { status: 500, times: 10 });
const b = (await h.api("GET", "/download-tasks/board")).body.data;
qb.clearFaults();
expect(b.counts.active).toBe(4);
expect(b.items.every((i: { live: unknown }) => i.live === null)).toBe(true);
expect(b.transfer).toMatchObject({ dlSpeed: 0, upSpeed: 0 });
expect(b.transfer.instances[0].reachable).toBe(false);
expect(b.liveAt).toBeNull();
});
it("按实例过滤;非法 state 返回 422", async () => {
const other = await h.api("GET", `/download-tasks/board?instanceId=${crypto.randomUUID()}`);
expect(other.body.data).toMatchObject({ items: [], counts: { active: 0, completed: 0 } });
expect(other.body.data.transfer.instances).toEqual([]);
expect((await h.api("GET", "/download-tasks/board?state=done")).status).toBe(422);
});
});
@@ -56,6 +56,8 @@ export interface AoOptions {
slotReevaluateIntervalMs?: number;
/** 追踪中剧集的元数据(总集数)刷新周期,缺省 6h;0 = 关 */
metadataRefreshIntervalMs?: number;
/** 下载面板实时数据(qB torrents/info + transfer/info)的缓存时长,缺省 2000ms;0 = 不缓存(仍合并并发请求) */
liveTtlMs?: number;
}
export interface AoPorts {
@@ -3,6 +3,7 @@ import { z } from "zod";
import { ApiException } from "../../../http/errors.ts";
import type { ModuleApp } from "../../types.ts";
import { type AoDeps, enqueueTracked, timeSlot } from "../services/common.ts";
import { downloadBoard } from "../services/download-board.ts";
import { userPause, userResume } from "../services/downloads.ts";
import { actorOf, notFound, toApi, toApiList } from "../util.ts";
import { mount, paginate } from "./helpers.ts";
@@ -49,6 +50,8 @@ export function registerDownloadRoutes(app: ModuleApp, d: AoDeps) {
},
{ tag: TAG },
);
// 在 /download-tasks/{id} 之前注册,避免 board 被当成 id
mount(app, Api.downloadTasks.board, async (_c, { query }) => downloadBoard(d, query), { tag: TAG });
mount(
app,
Api.downloadTasks.get,
@@ -0,0 +1,222 @@
import type { QbTorrent } from "@umt/adapters";
import {
type DecisionKind,
DOWNLOAD_BOARD_STATUSES,
type DownloadBoard,
type DownloadBoardItem,
type DownloadBoardQuery,
type DownloadLive,
type DownloadTask,
type OrganizationStatus,
} from "@umt/contracts";
import { type DownloaderInstanceRow, sql } from "@umt/db";
import { paginate } from "../routes/helpers.ts";
import { errMessage, toApi } from "../util.ts";
import type { AoDeps } from "./common.ts";
// 下载面板(DL-2):数据库里的下载任务 + qB 实时数据。
// 实时数据按实例缓存 liveTtlMs(缺省 2 s)并合并并发请求:多个标签页 / 多人每 3–5 s 轮询,
// 同一实例在一个 TTL 内只打一次 torrents/info(管理标签)+ transfer/info。
// 对照 MP:下载中心前端 3 s 轮询、页面不活跃时停;后端每次直查下载器。这里多一层短缓存。
/** qB 的 eta 无穷值(8640000 = 100 天) */
const QB_MAX_ETA = 8_640_000;
const DEFAULT_TTL_MS = 2000;
export interface InstanceLive {
instanceId: string;
name: string;
reachable: boolean;
dlSpeed: number;
upSpeed: number;
/** hash → 实时数据 */
torrents: Map<string, DownloadLive>;
at: Date;
}
interface Entry {
stamp: string;
at: number;
done: boolean;
p: Promise<InstanceLive>;
}
const caches = new WeakMap<AoDeps, Map<string, Entry>>();
const nonNeg = (v: unknown): number => {
const n = typeof v === "number" && Number.isFinite(v) ? v : 0;
return Math.max(0, Math.round(n));
};
export function liveOf(t: QbTorrent): DownloadLive {
const eta = t.eta !== null && t.eta >= 0 && t.eta < QB_MAX_ETA && !t.complete ? Math.round(t.eta) : null;
return {
dlSpeed: nonNeg(t.dlspeed),
upSpeed: nonNeg(t.upspeed),
etaSec: eta,
progress: Math.min(1, Math.max(0, t.progress)),
rawState: t.rawState,
seeds: nonNeg(t.numSeeds),
leechs: nonNeg(t.numLeechs),
seedsTotal: t.numComplete === null ? null : nonNeg(t.numComplete),
leechsTotal: t.numIncomplete === null ? null : nonNeg(t.numIncomplete),
ratio: t.ratio,
completedAt: t.completionOn ? new Date(t.completionOn * 1000).toISOString() : null,
};
}
async function fetchLive(d: AoDeps, inst: DownloaderInstanceRow): Promise<InstanceLive> {
const base = { instanceId: inst.id, name: inst.name };
let client: Awaited<ReturnType<AoDeps["ports"]["downloader"]>>;
try {
client = await d.ports.downloader(inst);
} catch (err) {
d.ctx.logger.warn({ instanceId: inst.id, err: { message: errMessage(err) } }, "下载面板:下载器客户端创建失败");
return { ...base, reachable: false, dlSpeed: 0, upSpeed: 0, torrents: new Map(), at: new Date() };
}
const [list, transfer] = await Promise.allSettled([client.list({ tag: inst.managed_tag }), client.transferInfo()]);
if (list.status === "rejected") {
d.ctx.logger.debug({ instanceId: inst.id, err: { message: errMessage(list.reason) } }, "下载面板:qB 不可达");
return { ...base, reachable: false, dlSpeed: 0, upSpeed: 0, torrents: new Map(), at: new Date() };
}
const torrents = new Map(list.value.map((t) => [t.hash, liveOf(t)]));
// transfer/info 是整个下载器的速度(含非本系统管理的任务);取不到时退回管理标签内任务之和
const sum = (k: "dlSpeed" | "upSpeed") => [...torrents.values()].reduce((a, t) => a + t[k], 0);
const info = transfer.status === "fulfilled" ? transfer.value : null;
return {
...base,
reachable: true,
dlSpeed: info ? nonNeg(info.dl_info_speed) : sum("dlSpeed"),
upSpeed: info ? nonNeg(info.up_info_speed) : sum("upSpeed"),
torrents,
at: new Date(),
};
}
/** 实例实时数据(短 TTL 缓存 + 合并在途请求;实例配置变更后作废) */
export function instanceLive(d: AoDeps, inst: DownloaderInstanceRow): Promise<InstanceLive> {
const ttl = d.ports.options.liveTtlMs ?? DEFAULT_TTL_MS;
let m = caches.get(d);
if (!m) {
m = new Map();
caches.set(d, m);
}
const stamp = `${inst.updated_at.toISOString()}|${inst.base_url}|${inst.managed_tag}`;
const hit = m.get(inst.id);
if (hit && hit.stamp === stamp && (!hit.done || Date.now() - hit.at < ttl)) return hit.p;
const entry: Entry = { stamp, at: Date.now(), done: false, p: fetchLive(d, inst) };
entry.p.then(
() => {
entry.done = true;
entry.at = Date.now();
},
() => {
entry.done = true;
entry.at = 0;
},
);
m.set(inst.id, entry);
return entry.p;
}
export async function downloadBoard(d: AoDeps, query: DownloadBoardQuery): Promise<DownloadBoard> {
const db = d.ctx.db;
let q = db
.selectFrom("download_task")
.selectAll()
.where("removed_at", "is", null)
.where("status", "in", [...DOWNLOAD_BOARD_STATUSES[query.state]]);
if (query.instanceId) q = q.where("instance_id", "=", query.instanceId);
let insts = db.selectFrom("downloader_instance").selectAll().where("enabled", "=", true);
if (query.instanceId) insts = insts.where("id", "=", query.instanceId);
const [page, counts, instances] = await Promise.all([
paginate(q, query),
sql<{ active: string; completed: string }>`
select count(*) filter (where status in (${sql.join([...DOWNLOAD_BOARD_STATUSES.active])})) as active,
count(*) filter (where status in (${sql.join([...DOWNLOAD_BOARD_STATUSES.completed])})) as completed
from download_task
where removed_at is null ${query.instanceId ? sql`and instance_id = ${query.instanceId}` : sql``}`.execute(db),
insts.orderBy("name").execute(),
]);
const rows = page.rows;
const entityIds = [...new Set(rows.map((r) => r.media_entity_id).filter((x): x is string => !!x))];
const taskIds = rows.map((r) => r.id);
const [entities, records, lives] = await Promise.all([
entityIds.length
? db.selectFrom("media_entity").select(["id", "title", "year"]).where("id", "in", entityIds).execute()
: Promise.resolve([]),
taskIds.length
? sql<{
id: string;
download_task_id: string;
status: OrganizationStatus;
reason: string | null;
decision_id: string | null;
decision_kind: DecisionKind | null;
}>`
with last as (
select distinct on (download_task_id) id, download_task_id, status, reason
from organization_record
where download_task_id in (${sql.join(taskIds)})
order by download_task_id, created_at desc
)
select last.*, dc.id as decision_id, dc.kind as decision_kind
from last
left join lateral (
select id, kind from decision
where status in ('open', 'investigating', 'awaiting_user')
and kind not in ('download_stalled', 'download_failed')
and ((subject_type = 'download_task' and subject_id = last.download_task_id)
or (subject_type = 'organization_record' and subject_id = last.id))
order by created_at desc limit 1
) dc on true`
.execute(db)
.then((r) => r.rows)
: Promise.resolve([]),
Promise.all(instances.map((i) => instanceLive(d, i))),
]);
const entityById = new Map(entities.map((e) => [e.id, e]));
const recordByTask = new Map(records.map((r) => [r.download_task_id, r]));
const liveByInst = new Map(lives.map((l) => [l.instanceId, l]));
const items: DownloadBoardItem[] = rows.map((r) => {
const e = r.media_entity_id ? entityById.get(r.media_entity_id) : undefined;
const rec = recordByTask.get(r.id);
return {
...toApi<DownloadTask>(r),
live: liveByInst.get(r.instance_id)?.torrents.get(r.torrent_hash) ?? null,
media: e ? { id: e.id, title: e.title, year: e.year } : null,
organize: rec
? {
recordId: rec.id,
status: rec.status,
reason: rec.reason,
decision: rec.decision_id && rec.decision_kind ? { id: rec.decision_id, kind: rec.decision_kind } : null,
}
: null,
};
});
const c = counts.rows[0];
const reachable = lives.filter((l) => l.reachable);
const oldest = reachable.length ? Math.min(...reachable.map((l) => l.at.getTime())) : null;
return {
items,
nextCursor: page.nextCursor,
counts: { active: Number(c?.active ?? 0), completed: Number(c?.completed ?? 0) },
transfer: {
dlSpeed: reachable.reduce((a, l) => a + l.dlSpeed, 0),
upSpeed: reachable.reduce((a, l) => a + l.upSpeed, 0),
instances: lives.map((l) => ({
instanceId: l.instanceId,
name: l.name,
reachable: l.reachable,
dlSpeed: l.dlSpeed,
upSpeed: l.upSpeed,
})),
},
liveAt: oldest === null ? null : new Date(oldest).toISOString(),
};
}