diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 00000000..81600d5d --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,52 @@ +# Changelog + +版本号遵循 [SemVer](https://semver.org/lang/zh-CN/)。所有成员包(`apps/*`、`packages/*`)与根 `deno.json` 使用同一个版本号;`/api/v1/health` 与备份清单里的版本取自 `apps/server/deno.json`。 + +## [0.1.0] - 2026-09-27 + +第一个可用版本:REQ v0.7 的 M0 + M1 纵切。全链路在没有真实 PT 账号的情况下用 mock(RSS / 站点 / TMDB / qBittorrent / Emby)验证。范围与验收依据见 `docs/plan/00-first-version-plan.md` §3.1 和 `docs/plan/04-v0.1-acceptance.md`。 + +### W0 底座 + +- Deno workspace 单仓(`apps/server`、`apps/web`、`packages/{adapters,agent,contracts,db,domain,mapping,tasks,testkit}`),迁入 uhd_web 设计系统。 +- OpenWorkflow 与 Pi SDK 兼容性 spike,结论写入 ADR-002 / ADR-003;Deno 2.9.7 + TypeScript 7 spike 写入 ADR-004。 +- MoviePilot / MovieClaw 迁移对照账本(识别、刮削、转移入库、站点框架)。 + +### W1 领域与适配 + +- `@umt/domain`:发布标题解析、身份判定、Slot 状态机与完成条件、收录模式判定、路径模板与路径绑定、整理计划、下载状态机,全部是纯函数。 +- `@umt/adapters`:qBittorrent(多实例、管理标签、带数据删除前的安全检查)、文件系统(Copy/Move、跨卷 Copy→核验→删源、EXDEV 不静默降级、全量 sha256)、TMDB、Emby、RSS、PT 站点(NexusPHP 观众、馒头 API,移植 MoviePilot 开放站点框架)。`@umt/testkit` 提供对应 mock。 +- `@umt/db`:36 张表的全量迁移;`@umt/contracts`:共享 zod 契约与 API 注册表;`@umt/mapping`:domain 与 contracts 之间的映射集中在一处。 +- `@umt/tasks`:基于 OpenWorkflow 的持久任务引擎、任务注册表、outbox 中继、操作账本 fencing(旧 attempt 的提交会被拒绝)。 +- `@umt/agent`:能力注册中心、Pi AgentRuntimePort、fake 模型、输出脱敏。 + +### W2 服务编排 + +- 认证与设置:管理员登录门禁(argon2id、会话 cookie、CSRF 双提交、登录限速、可选 TOTP)、API Key scope(read / organize.read / organize.execute / admin)、密钥加密存储、实例与路径映射、审计、PG 通知驱动的 SSE(带序号,可补读)、自动化总开关与只读对账模式、工作流引擎面板 API。 +- 来源与身份:站点与 Feed 管理、多 RSS 抓取与多层去重、站点快照与实时补查、TMDB 识别与本地实体优先、待识别池、人工纠错映射。 +- 规则、下载与入库:订阅、Slot、策略版本(发布即生成持续授权)、确定性 evaluateAcquisition、每日预算与并发、qB 同步与标签接管、缓存卷硬阈值与恢复水位、verifyAndImport(Copy/Move)、路径绑定固定、整理记录与公开增量 API、媒体库与首页聚合。 +- Agent:14 个能力接到服务,服务端授权(持续授权 / 一次性审批),会话 API + SSE 增量,后台事件触发有界调查任务。 + +### W3 控制台 + +- 20 个路由:首页、媒体库、订阅与规则、偏好标签、待处理中心、身份与元数据、PT 站点工作台、来源 / Feed / 快照、下载与任务、整理记录、Agent 会话、系统设置、工作流引擎面板、审计。 + +### W4 集成验收与修复 + +- 单容器打包(API + worker + 静态前端,compose 只有 app 和 pg),`DATA_DIR`,备份与恢复脚本,runbook。 +- 控制台接入真实端点;Playwright 全路由冒烟(`deno task test:web:smoke`)。 +- BrowserProvider 接口与 camofox-browser 适配骨架(只落接口)。 +- 主循环解耦:后台循环只做编排,清理 / 对账 / 入库一律入队到任务引擎;io / agent worker 分组;外部 IO 能力调用限流;事件循环延迟探针。 +- 安全修复:审批只能由管理员会话决定,共享一份脱敏表,TMDB 错误信息不带 api_key,生产构建不能打开 Agent mock。 +- 授权内异常调查与恢复:有持续授权时,调查结论可在授权范围内自动重试并继续。 +- 手动整理守卫(`POST /organization-records`):`sourcePaths` 必须在已登记的媒体库根或下载器路径映射之下(realpath 后按目录前缀匹配,拒绝 `..` 与符号链接逃逸,dryRun 同样受限);Move 需要管理员会话,或 API Key 同时有覆盖目标作品的持续授权 / 与本次请求一致的一次性审批;来源是纳管下载任务且做种义务未满足时返回 409 并建待处理,不改成 Copy。 +- 剧集完成判定:总集数未知(集单元都是临时单元)时 Slot 不判满足,订阅继续追踪;覆盖按实际文件登记,整季包不因包名覆盖整季;已在库的相同文件算作核验到的内容。 +- 工作流引擎面板支持重跑:`POST /engine/runs/{id}/retry`(仅 admin,新幂等键 `retry::<序号>`,写审计),run 详情页有"重跑"按钮。 +- 新增 e2e:E7 常规剧集纵切、E8 管理标签移除与 qB 任务消失(AT24)。 + +### 已知限制 + +- v0.1 明确不做的内容见 `docs/plan/00-first-version-plan.md` §3.2:国产综艺分段、双版本 require_distinct_asset 组合校验、浏览器接入与原站视图、HHClub / CHD、旧库接管、Agent 主观选片与规则学习、媒体级主动整理 UI、多用户。 +- 验收报告 `docs/plan/04-v0.1-acceptance.md` §1.2 的应修项和 §1.3 的建议项里,未标"已修"的条目留在 v0.1.x 处理。 +- 剧集总集数目前没有来源会把集单元确认为 `confirmed`,所以持续追踪的剧集 Slot 在 v0.1 不会进入 satisfied。 +- 恢复流程只有人工演练记录,没有自动化测试。 diff --git a/apps/server/deno.json b/apps/server/deno.json index aadb22f3..23850dbc 100644 --- a/apps/server/deno.json +++ b/apps/server/deno.json @@ -1,6 +1,6 @@ { "name": "@umt/server", - "version": "0.0.0", + "version": "0.1.0", "exports": { ".": "./src/main.ts" }, "tasks": { "dev": "deno run -A --watch --env-file=../../.env src/main.ts", diff --git a/apps/server/src/modules/acquisition-organize/__tests__/organize-guard.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/organize-guard.test.ts new file mode 100644 index 00000000..4b3d66c9 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/__tests__/organize-guard.test.ts @@ -0,0 +1,275 @@ +import { existsSync } from "node:fs"; +import { mkdir, symlink, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +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 "../services/acquisition.ts"; +import { submitDownload, syncDownloads } from "../services/downloads.ts"; +import { + createHarness, + EXEC_KEY, + fakeDisk, + type Harness, + pgAvailable, + seedBase, + seedEntity, + seedRelease, + seedSlot, +} from "./harness.ts"; + +// 验收 R4:POST /organization-records 的三道闸。 +// 路径守卫(400):根外路径 / `..` / 符号链接逃逸(顶层与目录内)/ 不存在;dryRun 同样受限。 +// Move 授权(403):API Key 无授权、授权范围不含目标作品 → 拒绝且不留记录;覆盖作品的持续授权、 +// 与本次请求一致的一次性审批(用后 consumed)、管理员 → 放行;Copy 与 dryRun 不要求授权。 +// 做种义务(409):来源是纳管下载任务(按 ID 或按路径命中)且 H&R 未满足 → 冲突 + 待处理,不改 Copy、不动 qB。 + +const hasPg = await pgAvailable(); +const SIZE = 150_000; + +describe.skipIf(!hasPg)("手动整理守卫:路径 / Move 授权 / 做种义务", () => { + let h: Harness; + let qb: MockQb; + let lib: TempTree; + let cache: TempTree; + let outside: TempTree; + const disk = fakeDisk(100_000_000); + const torrents = new Map(); + let base: Awaited>; + let movie: string; + + const post = (body: Record, key?: string) => h.api("POST", "/organization-records", body, key); + const records = async (entityId: string) => + h.db.selectFrom("organization_record").select(["id", "action"]).where("media_entity_id", "=", entityId).execute(); + const file = async (rel: string) => cache.write(rel, SIZE); + + beforeAll(async () => { + qb = await createMockQb({ addVisibleAfterPolls: 0 }); + lib = await createTempTree({}, { sentinel: true, prefix: "umt-glib-" }); + cache = await createTempTree({}, { prefix: "umt-gcache-" }); + outside = await createTempTree( + { "Secret.2020.1080p.BluRay-X.mkv": SIZE, "etc/passwd": 100 }, + { prefix: "umt-gout-" }, + ); + disk.state.watch = [cache.root]; + disk.setPct(10); + h = await createHarness("guard", { + 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 }); + movie = await seedEntity(h.db, { kind: "movie", title: "守卫片", year: 2020 }); + }, 60_000); + + afterAll(async () => { + await h?.close(); + await qb?.close(); + for (const t of [lib, cache, outside]) await t?.cleanup(); + }); + + it("路径守卫:根外 / `..` / 符号链接逃逸 / 不存在 → 400,dryRun 也不例外,且不留记录", async () => { + const outFile = outside.path("Secret.2020.1080p.BluRay-X.mkv"); + for (const dryRun of [true, false]) { + const r = await post({ sourcePaths: [outFile], mediaEntityId: movie, action: "copy", dryRun }); + expect(r.status).toBe(400); + expect(r.body.error.message).toContain("不在已登记的媒体库根或下载器路径映射之下"); + } + // `..` 即使规范化后仍在根内也拒绝 + const inside = await file("Dots.2020.1080p.BluRay-X.mkv"); + const dotted = `${cache.root}/sub/../Dots.2020.1080p.BluRay-X.mkv`; + expect((await post({ sourcePaths: [dotted], mediaEntityId: movie, action: "copy" })).status).toBe(400); + // 顶层符号链接:根内的链接指向根外文件 + const link = join(cache.root, "Link.2020.1080p.BluRay-X.mkv"); + await symlink(outFile, link); + const r1 = await post({ sourcePaths: [link], mediaEntityId: movie, action: "copy", dryRun: true }); + expect(r1.status).toBe(400); + // 目录内的符号链接:遍历用 lstat,不跟随链接(文件链接、目录链接都不进计划),根外内容不会被列出 + const dir = join(cache.root, "Pack.2020.1080p.BluRay-X"); + await mkdir(dir); + await writeFile(join(dir, "Pack.2020.1080p.BluRay-X.mkv"), Buffer.alloc(SIZE, 1)); + await symlink(outside.root, join(dir, "escape")); + await symlink(outFile, join(dir, "Pack.2020.1080p.BluRay-X.sample.mkv")); + const r2 = await post({ sourcePaths: [dir], mediaEntityId: movie, action: "copy", dryRun: true }); + expect(r2.status).toBe(201); + const planArt = await h.db + .selectFrom("artifact") + .select("content") + .where("id", "=", r2.body.data.planId) + .executeTakeFirstOrThrow(); + const srcs = (planArt.content as { plan: { organize: { files: { src: string }[] } } }).plan.organize.files.map( + (f) => f.src, + ); + expect(srcs).toEqual([join(dir, "Pack.2020.1080p.BluRay-X.mkv")]); + expect(JSON.stringify(planArt.content)).not.toContain(outside.root); + // 不存在 + expect( + (await post({ sourcePaths: [join(cache.root, "nope.mkv")], mediaEntityId: movie, action: "copy" })).status, + ).toBe(400); + expect(await records(movie)).toHaveLength(0); + expect(existsSync(outFile)).toBe(true); + // 根内正常路径:dryRun 通过 + const ok = await post({ sourcePaths: [inside], mediaEntityId: movie, action: "copy", dryRun: true }, EXEC_KEY); + expect(ok.status).toBe(201); + expect(ok.body.data.planId).toBeTruthy(); + }); + + it("Move 授权:API Key 无授权 / 范围不含作品 → 403 不留记录;dryRun 与 Copy 不要求授权", async () => { + const src = await file("Auth.2020.1080p.BluRay-X.mkv"); + const denied = await post({ sourcePaths: [src], mediaEntityId: movie, action: "move" }, EXEC_KEY); + expect(denied.status).toBe(403); + expect(denied.body.error.message).toContain("需要"); + expect(await records(movie)).toHaveLength(0); + expect(existsSync(src)).toBe(true); + + // 授权范围只含别的作品 + const other = await seedEntity(h.db, { kind: "movie", title: "别的片", year: 2019 }); + const g = await h.db + .insertInto("automation_grant") + .values({ action: "organize.execute", scope: JSON.stringify({ mediaEntityIds: [other] }), granted_by: "test" }) + .returning("id") + .executeTakeFirstOrThrow(); + const scoped = await post({ sourcePaths: [src], mediaEntityId: movie, action: "move" }, EXEC_KEY); + expect(scoped.status).toBe(403); + expect(scoped.body.error.message).toContain("对象范围不含目标作品"); + // 没有目标作品时持续授权无从核对 + expect((await post({ sourcePaths: [src], action: "move" }, EXEC_KEY)).status).toBe(403); + + // dryRun Move / Copy:只要 organize.execute + expect( + (await post({ sourcePaths: [src], mediaEntityId: movie, action: "move", dryRun: true }, EXEC_KEY)).status, + ).toBe(201); + await h.db.updateTable("automation_grant").set({ status: "revoked" }).where("id", "=", g.id).execute(); + const audits = await sql<{ n: number }>` + select count(*)::int as n from audit_event where action = 'organize.move_denied'`.execute(h.db); + expect(audits.rows[0]!.n).toBe(3); + }); + + it("Move 授权:覆盖作品的持续授权放行;策略版本非发布状态不算", async () => { + const e = await seedEntity(h.db, { kind: "movie", title: "授权片", year: 2021 }); + const src = await file("Granted.2021.1080p.BluRay-X.mkv"); + const pol = await h.db + .insertInto("policy_revision") + .values({ policy_key: "g", revision: 1, kind: "acquisition", title: "g", status: "draft" }) + .returning("id") + .executeTakeFirstOrThrow(); + await h.db + .insertInto("automation_grant") + .values({ + action: "policy.acquisition", + policy_revision_id: pol.id, + scope: JSON.stringify({ mediaKinds: ["movie"] }), + granted_by: "policy:g", + }) + .execute(); + const draft = await post({ sourcePaths: [src], mediaEntityId: e, action: "move" }, EXEC_KEY); + expect(draft.status).toBe(403); + expect(draft.body.error.message).toContain("不是发布状态"); + const g = await h.db + .insertInto("automation_grant") + .values({ action: "organize.execute", scope: JSON.stringify({ mediaEntityIds: [e] }), granted_by: "test" }) + .returning("id") + .executeTakeFirstOrThrow(); + const ok = await post({ sourcePaths: [src], mediaEntityId: e, action: "move" }, EXEC_KEY); + expect(ok.status).toBe(201); + expect((await records(e)).map((r) => r.action)).toEqual(["move"]); + await h.db.updateTable("automation_grant").set({ status: "revoked" }).where("id", "=", g.id).execute(); + }); + + it("Move 授权:一次性审批绑定 dryRun 计划,请求一致才放行,用后 consumed 不可复用", async () => { + const e = await seedEntity(h.db, { kind: "movie", title: "审批片", year: 2022 }); + const src = await file("Approved.2022.1080p.BluRay-X.mkv"); + const req = { sourcePaths: [src], mediaEntityId: e, action: "move" as const }; + const dry = await post({ ...req, dryRun: true }, EXEC_KEY); + expect(dry.status).toBe(201); + const art = await h.db + .selectFrom("artifact") + .select(["id", "content_hash", "content"]) + .where("id", "=", dry.body.data.planId) + .executeTakeFirstOrThrow(); + expect(art.content).toMatchObject({ mediaEntityId: e, mediaKind: "movie" }); + const approve = (hash: string) => + h.db + .insertInto("approval") + .values({ + plan_type: "organize_plan", + plan_id: art.id, + payload_hash: hash, + expected_revision: 1, + status: "approved", + requested_by: "api_key:x", + decided_by: "admin_session:a", + }) + .returning("id") + .executeTakeFirstOrThrow(); + // 请求与计划不一致(换了目标库根字段)→ 不算 + const a = await approve(art.content_hash); + const mismatch = await post({ ...req, action: "move", targetLibraryRootId: base.rootId }, EXEC_KEY); + expect(mismatch.status).toBe(403); + const ok = await post(req, EXEC_KEY); + expect(ok.status).toBe(201); + const used = await h.db.selectFrom("approval").select("status").where("id", "=", a.id).executeTakeFirstOrThrow(); + expect(used.status).toBe("consumed"); + // 用过的审批不能再放行第二次 + const again = await post(req, EXEC_KEY); + expect(again.status).toBe(403); + // payloadHash 与计划内容不符的审批无效 + await approve("tampered"); + expect((await post(req, EXEC_KEY)).status).toBe(403); + }); + + it("做种义务:纳管下载任务 H&R 未满足 → 409 + 待处理,不改 Copy、不动 qB;管理员 Copy 可以", async () => { + const e = await seedEntity(h.db, { kind: "movie", title: "做种片", year: 2023 }); + await seedSlot(h.db, { entityId: e }); + const t = makeTorrent({ name: "Seeding.2023.1080p.BluRay-HR.mkv", length: SIZE, pieceSeed: 11 }); + const relId = await seedRelease(h.db, { + siteId: base.siteId, + entityId: e, + title: "Seeding.2023.1080p.BluRay-HR", + size: SIZE, + hash: t.hash, + hr: true, + }); + torrents.set(relId, t.bytes); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relId }); + const sub = await submitDownload(h.d, ev.plans[0]!); + await qb.complete(t.hash, { writeFiles: true }); + await syncDownloads(h.d); + const dtId = sub.downloadTaskId!; + const deletesBefore = qb.callsTo("torrents/delete").length; + + // 按下载任务 ID(管理员也一样被做种义务拦下) + const byId = await post({ downloadTaskId: dtId, action: "move" }); + expect(byId.status).toBe(409); + expect(byId.body.error.message).toContain("未改为 Copy"); + // 按路径命中纳管任务的内容 + const content = join(cache.root, "Seeding.2023.1080p.BluRay-HR.mkv"); + expect(existsSync(content)).toBe(true); + const byPath = await post({ sourcePaths: [content], mediaEntityId: e, action: "move" }); + expect(byPath.status).toBe(409); + const decisions = await h.db + .selectFrom("decision") + .select(["kind", "status", "attempts"]) + .where("dedupe_key", "=", `manual_move_hnr:${dtId}`) + .execute(); + expect(decisions).toMatchObject([{ kind: "approval_required", status: "open" }]); + expect(Number(decisions[0]!.attempts)).toBeGreaterThanOrEqual(1); + expect(await records(e)).toHaveLength(0); + expect(qb.callsTo("torrents/delete")).toHaveLength(deletesBefore); + expect(qb.get(t.hash)).toBeDefined(); + expect(existsSync(content)).toBe(true); + // 同一来源 Copy 不受做种义务限制 + const copy = await post({ downloadTaskId: dtId, action: "copy" }); + expect(copy.status).toBe(201); + expect((await records(e)).map((r) => r.action)).toEqual(["copy"]); + }, 60_000); + + it("管理员会话(dev admin)按路径 Move:放行并记审计", async () => { + const e = await seedEntity(h.db, { kind: "movie", title: "管理员片", year: 2024 }); + const src = await file("Admin.2024.1080p.BluRay-X.mkv"); + const r = await post({ sourcePaths: [src], mediaEntityId: e, action: "move" }); + expect(r.status).toBe(201); + const a = await sql<{ payload: { via: string } }>` + select payload from audit_event where action = 'organize.move_authorized' and target_id = ${e}`.execute(h.db); + expect(a.rows.map((x) => x.payload.via)).toEqual(["admin_session"]); + }); +}); diff --git a/apps/server/src/modules/acquisition-organize/__tests__/organize-move.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/organize-move.test.ts index d2d2e0af..152f9e3c 100644 --- a/apps/server/src/modules/acquisition-organize/__tests__/organize-move.test.ts +++ b/apps/server/src/modules/acquisition-organize/__tests__/organize-move.test.ts @@ -56,7 +56,8 @@ describe.skipIf(!hasPg)("整理:跨卷 Move 恢复 / 路径绑定固定 / 手动 "Show.S02E01.1080p.WEB-DL.x264-GRP.mkv": SIZE, "Show.S01E01.1080p.WEB-DL.x264-GRP.chs.srt": 2_000, }, - { prefix: "umt-src-" }, + // 手动整理的源路径必须在已登记的根下(验收 R4):放在下载器保存路径里 + { prefix: "umt-src-", base: cache.root }, ); disk.state.watch = [cache.root]; disk.setPct(30); diff --git a/apps/server/src/modules/acquisition-organize/routes/organization.ts b/apps/server/src/modules/acquisition-organize/routes/organization.ts index 074ade31..8285a7de 100644 --- a/apps/server/src/modules/acquisition-organize/routes/organization.ts +++ b/apps/server/src/modules/acquisition-organize/routes/organization.ts @@ -5,7 +5,16 @@ import { ApiException } from "../../../http/errors.ts"; import type { ModuleApp } from "../../types.ts"; import { type AoDeps, ensureTaskRow, markTask } from "../services/common.ts"; import { loadRecordsPublic, planImport } from "../services/organize.ts"; -import { actorOf, errMessage, jsonb, notFound } from "../util.ts"; +import { + authorizeMove, + hnrBlockForMove, + managedTasksUnder, + PathGuardError, + requestEntityId, + type SourceGuard, + sourceGuard, +} from "../services/organize-guard.ts"; +import { actorOf, errMessage, hashJson, jsonb, notFound } from "../util.ts"; import { mount, paginate } from "./helpers.ts"; // 整理记录公开 API(organize.read / organize.execute)。changes 必须先于 {id} 注册。 @@ -80,13 +89,65 @@ export function registerOrganizationRoutes(app: ModuleApp, d: AoDeps) { { tag: TAG }, ); - // 手动发起整理:与自动整理走同一冻结计划流程;返回 record/task ID,由任务页追踪 + // 手动发起整理:与自动整理走同一冻结计划流程;返回 record/task ID,由任务页追踪。 + // 先过三道闸(organize-guard.ts):路径守卫(400)→ Move 授权(403)→ 纳管来源的做种义务(409 + 待处理)。 mount( app, Api.organization.create, async (c, { body }) => { const principal = c.get("principal"); const actorType = principal?.kind === "api_key" ? "api_key" : "user"; + + // ① 路径守卫:dryRun 也受限(否则可借计划枚举任意目录) + let guard: SourceGuard | null = null; + const realPaths: string[] = []; + if (body.sourcePaths) { + guard = await sourceGuard(db()); + try { + for (const p of body.sourcePaths) realPaths.push(await guard.check(p)); + } catch (err) { + if (err instanceof PathGuardError) throw new ApiException("bad_request", err.message); + throw err; + } + } + if (body.downloadTaskId) { + const dt = await db() + .selectFrom("download_task") + .select(["status"]) + .where("id", "=", body.downloadTaskId) + .executeTakeFirst(); + if (!dt) throw new ApiException("not_found", "下载任务不存在"); + if (!["completed", "importing", "imported"].includes(dt.status)) + throw new ApiException("precondition_failed", `下载任务尚未完成(${dt.status})`); + } + + // ② / ③ Move:删源 + 移除 qB 任务记录,必须有授权;纳管来源还要满足做种义务 + if (!body.dryRun && body.action === "move") { + const managed = body.downloadTaskId ? [body.downloadTaskId] : await managedTasksUnder(db(), realPaths); + const entityId = await requestEntityId(db(), { + mediaEntityId: body.mediaEntityId, + downloadTaskIds: managed, + }); + const auth = await authorizeMove(d, principal, body, entityId); + if (!auth.allowed) { + await audit(d, actorOf(c), "organize.move_denied", { reason: auth.reason, entityId }); + throw new ApiException("forbidden", auth.reason); + } + const block = await hnrBlockForMove(d, managed, { entityId, actor: actorOf(c) }); + if (block) { + throw new ApiException( + "conflict", + `来源下载任务 ${block.downloadTaskId} 的做种义务${block.hnr === "unknown" ? "未知" : "未满足"},` + + `拒绝 Move(未改为 Copy);已建待处理 ${block.decisionId}`, + ); + } + await audit(d, actorOf(c), "organize.move_authorized", { + via: auth.via, + ...(auth.id ? { id: auth.id } : {}), + entityId, + }); + } + const idem = body.downloadTaskId && !body.dryRun ? `manual:${body.downloadTaskId}:${randomUUID()}` @@ -99,16 +160,6 @@ export function registerOrganizationRoutes(app: ModuleApp, d: AoDeps) { subjectId: body.downloadTaskId ?? null, triggeredBy: actorOf(c), }); - if (body.downloadTaskId) { - const dt = await db() - .selectFrom("download_task") - .select(["status"]) - .where("id", "=", body.downloadTaskId) - .executeTakeFirst(); - if (!dt) throw new ApiException("not_found", "下载任务不存在"); - if (!["completed", "importing", "imported"].includes(dt.status)) - throw new ApiException("precondition_failed", `下载任务尚未完成(${dt.status})`); - } let plan: Awaited>; try { plan = await planImport(d, { @@ -123,13 +174,31 @@ export function registerOrganizationRoutes(app: ModuleApp, d: AoDeps) { actorId: actorOf(c), taskId, dryRun: body.dryRun, + ...(guard ? { pathGuard: guard.check } : {}), }); } catch (err) { await markTask(db(), taskId, "failed", { error: errMessage(err) }); throw new ApiException("bad_request", errMessage(err)); } if (body.dryRun) { - // 只生成计划:冻结为 artifact(kind=plan),planId 供 organize.execute 使用 + // 只生成计划:冻结为 artifact(kind=plan),planId 供一次性审批 / organize.execute 使用。 + // content_hash 由内容确定(审批绑定 payloadHash);mediaEntityId/mediaKind 供授权范围核对。 + const entity = plan.plan + ? await db() + .selectFrom("media_entity") + .select(["kind"]) + .where("id", "=", plan.plan.entityId) + .executeTakeFirst() + : undefined; + const content = { + request: body, + plan: plan.plan ?? null, + reason: plan.reason, + status: plan.status, + truncated: plan.truncated === true, + mediaEntityId: plan.plan?.entityId ?? body.mediaEntityId ?? null, + mediaKind: entity?.kind ?? null, + }; const art = await db() .insertInto("artifact") .values({ @@ -137,14 +206,8 @@ export function registerOrganizationRoutes(app: ModuleApp, d: AoDeps) { task_id: taskId, subject_type: "organize_request", title: "organize_plan", - content: jsonb({ - request: body, - plan: plan.plan ?? null, - reason: plan.reason, - status: plan.status, - truncated: plan.truncated === true, - }), - content_hash: randomUUID(), + content: jsonb(content), + content_hash: hashJson({ ...content, taskId }), }) .returning("id") .executeTakeFirstOrThrow(); @@ -175,3 +238,17 @@ export function registerOrganizationRoutes(app: ModuleApp, d: AoDeps) { { tag: TAG, status: 201 }, ); } + +async function audit(d: AoDeps, actor: string, action: string, detail: Record) { + await d.ctx.db + .insertInto("audit_event") + .values({ + actor_type: actor.startsWith("api_key:") ? "api_key" : "user", + actor_id: actor, + action, + target_type: "organize_request", + target_id: typeof detail.entityId === "string" ? detail.entityId : null, + payload: jsonb(detail), + }) + .execute(); +} diff --git a/apps/server/src/modules/acquisition-organize/services/organize-guard.ts b/apps/server/src/modules/acquisition-organize/services/organize-guard.ts new file mode 100644 index 00000000..e306670a --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/organize-guard.ts @@ -0,0 +1,297 @@ +import { realpath } from "node:fs/promises"; +import { isAbsolute, resolve, sep } from "node:path"; +import type { Principal } from "../../../http/auth.ts"; +import { canonicalJson } from "../util.ts"; +import { type AoDeps, type Exec, hostPathOf, loadInstances, upsertDecision } from "./common.ts"; +import { hnrSatisfied } from "./retention.ts"; + +// 手动整理(POST /organization-records)的三道闸(验收 R4,REQ §2.1 原则 5、§8.3): +// 1) 路径守卫:sourcePaths 必须落在已登记的 library_root 或下载器路径映射的本机路径之下 +// (realpath 规范化后按目录前缀匹配;拒绝 `..` 段与符号链接逃逸,遍历时逐个文件复核)。dryRun 同样受限。 +// 2) 授权:Move 会删源、会移除 qB 任务记录,必须是管理员会话,或 API Key 同时有覆盖目标作品的 +// automation_grant(发布策略生成)/ 与本次请求一致的一次性 approval;Copy 只需 organize.execute scope。 +// 3) 做种义务:来源是纳管下载任务时,Move 前按 verifyAndImport 同一判定检查 H&R;未满足 → 409 + 待处理, +// 不静默改成 Copy。 + +export class PathGuardError extends Error { + override name = "PathGuardError"; + constructor( + message: string, + readonly path: string, + ) { + super(message); + } +} + +const within = (p: string, root: string) => p === root || p.startsWith(root.endsWith(sep) ? root : `${root}${sep}`); + +async function real(p: string): Promise { + try { + return await realpath(p); + } catch { + return resolve(p); + } +} + +/** 允许作为整理来源的本机根:媒体库根 + 下载器路径映射的本机前缀(无映射实例 = 同机同路径,取保存路径) */ +export async function allowedSourceRoots(exec: Exec): Promise { + const roots = new Set(); + for (const r of await exec.selectFrom("library_root").select("local_path").execute()) roots.add(r.local_path); + for (const t of await loadInstances(exec, { enabledOnly: false })) { + for (const m of t.mappings) roots.add(m.hostPath); + if (t.mappings.length === 0 && t.localSavePath) roots.add(t.localSavePath); + } + const out: string[] = []; + for (const r of roots) if (r && isAbsolute(r)) out.push(await real(r.replace(/\/+$/, "") || "/")); + // 根目录本身("/")不能作为白名单,否则等于不设防 + return [...new Set(out)].filter((r) => r !== "/"); +} + +export interface SourceGuard { + roots: string[]; + /** 返回 realpath;不在任何根下抛 PathGuardError */ + check(p: string): Promise; +} + +export async function sourceGuard(exec: Exec): Promise { + const roots = await allowedSourceRoots(exec); + return { + roots, + async check(p) { + if (!isAbsolute(p)) throw new PathGuardError(`源路径必须是绝对路径: ${p}`, p); + if (p.split("/").some((seg) => seg === "..")) throw new PathGuardError(`源路径不允许包含 "..": ${p}`, p); + let r: string; + try { + r = await realpath(p); + } catch { + throw new PathGuardError(`源路径不存在或不可访问: ${p}`, p); + } + if (!roots.some((root) => within(r, root))) + throw new PathGuardError(`源路径不在已登记的媒体库根或下载器路径映射之下: ${p}`, p); + return r; + }, + }; +} + +// ── 目标作品与来源下载任务 ── + +/** 手动请求的目标作品:显式 mediaEntityId,否则取下载任务的计划 Slot / 来源资源的作品 */ +export async function requestEntityId( + exec: Exec, + req: { mediaEntityId?: string | undefined; downloadTaskIds: string[] }, +): Promise { + if (req.mediaEntityId) return req.mediaEntityId; + for (const id of req.downloadTaskIds) { + const r = await exec + .selectFrom("download_task as dt") + .leftJoin("acquisition_plan as p", "p.id", "dt.acquisition_plan_id") + .leftJoin("acquisition_slot as s", "s.id", "p.slot_id") + .leftJoin("source_release as sr", "sr.id", "dt.source_release_id") + .select(["s.media_entity_id as slot_entity", "sr.media_entity_id as release_entity"]) + .where("dt.id", "=", id) + .executeTakeFirst(); + const e = r?.slot_entity ?? r?.release_entity; + if (e) return e; + } + return null; +} + +/** 与这些源路径重叠的纳管下载任务(源路径在任务内容里,或任务内容在源路径下) */ +export async function managedTasksUnder(exec: Exec, realPaths: string[]): Promise { + const rows = await exec + .selectFrom("download_task") + .select(["id", "instance_id", "content_path", "save_path", "name"]) + .where("managed", "=", true) + .where("removed_at", "is", null) + .where("status", "!=", "removed") + .execute(); + if (rows.length === 0) return []; + const topo = new Map((await loadInstances(exec, { enabledOnly: false })).map((t) => [t.inst.id, t])); + const hit: string[] = []; + for (const r of rows) { + const t = topo.get(r.instance_id); + if (!t) continue; + const content = r.content_path ?? (r.save_path && r.name ? `${r.save_path.replace(/\/+$/, "")}/${r.name}` : null); + const host = content ? hostPathOf(t, content) : null; + if (!host) continue; + const h = await real(host); + if (realPaths.some((p) => within(p, h) || within(h, p))) hit.push(r.id); + } + return hit; +} + +// ── 授权 ── + +const GRANT_ACTIONS = ["organize.execute", "policy.acquisition", "policy.organize", "*"]; + +function kindCovered(kinds: readonly string[], entityKind: string): boolean { + if (kinds.includes(entityKind)) return true; + return entityKind !== "movie" && entityKind !== "variety" && kinds.includes("series"); +} + +/** 手动请求里参与"是否同一请求"比对的字段(不含 dryRun) */ +export function requestKey(body: { + downloadTaskId?: string | undefined; + sourcePaths?: string[] | undefined; + mediaEntityId?: string | undefined; + contentUnitId?: string | undefined; + action: "copy" | "move"; + targetLibraryRootId?: string | undefined; +}): string { + return canonicalJson({ + downloadTaskId: body.downloadTaskId, + sourcePaths: body.sourcePaths, + mediaEntityId: body.mediaEntityId, + contentUnitId: body.contentUnitId, + action: body.action, + targetLibraryRootId: body.targetLibraryRootId, + }); +} + +export type MoveAuthorization = + | { allowed: true; via: "admin_session" | "grant" | "approval"; id?: string } + | { allowed: false; reason: string }; + +/** + * Move 授权:管理员会话直接放行;API Key 需要覆盖目标作品的持续授权,或一次性审批—— + * 审批绑定的是 dryRun 生成的整理计划(artifact kind=plan),计划里冻结的请求须与本次请求一致;用后标 consumed。 + */ +export async function authorizeMove( + d: AoDeps, + principal: Principal | undefined, + body: Parameters[0], + entityId: string | null, +): Promise { + if (principal?.kind === "admin_session" || principal?.kind === "dev_admin") + return { allowed: true, via: "admin_session" }; + const db = d.ctx.db; + const now = d.ports.now(); + const entity = entityId + ? await db.selectFrom("media_entity").select(["id", "kind"]).where("id", "=", entityId).executeTakeFirst() + : undefined; + const reasons: string[] = []; + + // 1) 一次性审批 + const key = requestKey(body); + const approvals = await db + .selectFrom("approval as a") + .innerJoin("artifact as art", "art.id", "a.plan_id") + .select(["a.id", "a.payload_hash", "a.expires_at", "art.content_hash", "art.content"]) + .where("a.plan_type", "=", "organize_plan") + .where("a.status", "=", "approved") + .where("art.kind", "=", "plan") + .where((eb) => eb.or([eb("a.expires_at", "is", null), eb("a.expires_at", ">", now)])) + .orderBy("a.created_at", "desc") + .execute(); + for (const a of approvals) { + const req = ((a.content ?? {}) as { request?: Parameters[0] }).request; + if (!req || requestKey(req) !== key) continue; + if (a.payload_hash !== a.content_hash) { + reasons.push(`审批 ${a.id} 绑定的计划内容已变化`); + continue; + } + const used = await db + .updateTable("approval") + .set({ status: "consumed" }) + .where("id", "=", a.id) + .where("status", "=", "approved") + .executeTakeFirst(); + if (Number(used.numUpdatedRows ?? 0n) === 1) return { allowed: true, via: "approval", id: a.id }; + } + + // 2) 持续授权 + if (!entity) { + reasons.push("无法确定目标作品,持续授权无从核对范围"); + } else { + const grants = await db + .selectFrom("automation_grant as g") + .leftJoin("policy_revision as p", "p.id", "g.policy_revision_id") + .select(["g.id", "g.scope", "g.policy_revision_id", "p.status as policy_status"]) + .where("g.status", "=", "active") + .where("g.action", "in", GRANT_ACTIONS) + .where((eb) => eb.or([eb("g.expires_at", "is", null), eb("g.expires_at", ">", now)])) + .orderBy("g.created_at", "desc") + .execute(); + for (const g of grants) { + if (g.policy_revision_id && g.policy_status !== "published") { + reasons.push(`授权 ${g.id} 绑定的策略版本已不是发布状态`); + continue; + } + const scope = (g.scope ?? {}) as { mediaEntityIds?: string[]; mediaKinds?: string[]; capabilities?: string[] }; + if (scope.capabilities?.length && !scope.capabilities.includes("organize.execute")) { + reasons.push(`授权 ${g.id} 不含 organize.execute`); + continue; + } + if (scope.mediaEntityIds?.length && !scope.mediaEntityIds.includes(entity.id)) { + reasons.push(`授权 ${g.id} 的对象范围不含目标作品`); + continue; + } + if (scope.mediaKinds?.length && !kindCovered(scope.mediaKinds, entity.kind)) { + reasons.push(`授权 ${g.id} 的媒体类型范围不含 ${entity.kind}`); + continue; + } + return { allowed: true, via: "grant", id: g.id }; + } + } + return { + allowed: false, + reason: + reasons.length > 0 + ? `Move 需要授权:${reasons.join(";")}` + : "Move 会删除源文件并移除下载器任务记录,需要管理员会话,或覆盖目标作品的持续授权 / 一次性审批", + }; +} + +// ── 做种义务 ── + +export interface HnrBlock { + downloadTaskId: string; + hnr: false | "unknown"; + decisionId: string; +} + +/** 纳管下载任务的 H&R / 保留要求:未满足(含未知)→ 建待处理并返回阻止信息 */ +export async function hnrBlockForMove( + d: AoDeps, + downloadTaskIds: string[], + ctx: { entityId: string | null; actor: string }, +): Promise { + const db = d.ctx.db; + for (const id of downloadTaskIds) { + const dt = await db.selectFrom("download_task").selectAll().where("id", "=", id).executeTakeFirst(); + if (!dt) continue; + const rel = dt.source_release_id + ? await db.selectFrom("source_release").selectAll().where("id", "=", dt.source_release_id).executeTakeFirst() + : undefined; + const site = rel + ? await db.selectFrom("site").selectAll().where("id", "=", rel.site_id).executeTakeFirst() + : undefined; + const hnr = hnrSatisfied(dt, rel, site); + if (hnr === true) continue; + const decisionId = await upsertDecision(d, db, { + kind: "approval_required", + title: `手动 Move 被做种义务阻止:${dt.name ?? dt.torrent_hash}`, + summary: + hnr === false + ? "来源是纳管下载任务,H&R / 做种要求尚未满足;Move 会删源并移除任务,已拒绝(未改为 Copy)" + : "来源是纳管下载任务,H&R 要求未知,按未满足处理;Move 已拒绝(未改为 Copy)", + subjectType: "download_task", + subjectId: dt.id, + dedupeKey: `manual_move_hnr:${dt.id}`, + evidence: { + downloadTaskId: dt.id, + hnr, + seedingTimeSec: dt.seeding_time_sec, + ratio: dt.ratio, + mediaEntityId: ctx.entityId, + requestedBy: ctx.actor, + }, + options: [ + { key: "wait", label: "等待做种义务满足后再 Move" }, + { key: "copy", label: "改用 Copy 入库(需重新发起)" }, + ], + }); + return { downloadTaskId: dt.id, hnr, decisionId }; + } + return null; +} diff --git a/apps/server/src/modules/acquisition-organize/services/organize.ts b/apps/server/src/modules/acquisition-organize/services/organize.ts index 43513c68..98cc3cb4 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize.ts @@ -37,6 +37,7 @@ import { errMessage, hashJson, jsonb, toApi } from "../util.ts"; import { publishedAcquisitionPolicies, unitKeysOf } from "./acquisition.ts"; import { type AoDeps, type Exec, hostPathOf, loadInstances, upsertDecision } from "./common.ts"; import { applySlotEvent } from "./downloads.ts"; +import { PathGuardError } from "./organize-guard.ts"; import { hnrSatisfied } from "./retention.ts"; // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→ @@ -56,6 +57,8 @@ export interface ImportRequest { taskId?: string | null; /** 只生成计划,不写记录、不建待处理 */ dryRun?: boolean; + /** 手动按路径整理:遍历到的每个文件都经它复核(realpath 在允许的根下),越界抛 PathGuardError */ + pathGuard?: ((p: string) => Promise) | undefined; } /** organization_record.plan 的冻结内容 */ @@ -111,10 +114,17 @@ function segmentsWithin(path: string, root: string | null) { } /** 递归列出文件(跳过隐藏项),最多 limit 个;调用方传 上限+1 判断是否截断 */ -async function walk(p: string, out: string[], limit: number): Promise { +async function walk( + p: string, + out: string[], + limit: number, + guard?: ((p: string) => Promise) | undefined, +): Promise { const st = await statSafe(p); if (st.kind !== "ok") throw new FsError("ENOENT", `源路径不可访问: ${p} (${st.kind})`, { path: p }); if (st.facts.isFile) { + // 目录里的符号链接可能指向根外:逐个文件复核 realpath + if (guard) await guard(p); out.push(p); return; } @@ -123,7 +133,7 @@ async function walk(p: string, out: string[], limit: number): Promise { for (const n of names) { if (out.length >= limit) return; if (n.startsWith(".")) continue; - await walk(join(p, n), out, limit); + await walk(join(p, n), out, limit, guard); } } @@ -235,8 +245,9 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise max) { @@ -298,7 +309,8 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise [f.src, f])); const assetBySrc = new Map(); const mainAssets: string[] = []; + const presentAssets: string[] = []; for (const f of files) { + // 目标已在库且全量 sha256 相同(already_in_library):不重复登记,但已登记的资产算作本次核验到的内容 + // (例如整季包里与单集入库相同的文件),参与 Slot 覆盖判断 + if (f.status === "skipped" && f.skip_reason === "already_in_library" && f.role === "main" && f.target_path) { + const have = await trx + .selectFrom("asset_location as l") + .innerJoin("media_asset as a", "a.id", "l.media_asset_id") + .select("a.id") + .where("l.library_root_id", "=", frozen.libraryRootId) + .where("l.relative_path", "=", f.target_path) + .where("l.status", "!=", "removed") + .where("a.status", "=", "active") + .executeTakeFirst(); + if (have) { + presentAssets.push(have.id); + await trx + .updateTable("organization_file_result") + .set({ media_asset_id: have.id }) + .where("id", "=", f.id) + .execute(); + } + continue; + } if (f.status !== "succeeded" || !f.target_path) continue; const pf = planFiles.get(f.source_path); const isMain = f.role === "main"; @@ -1018,16 +1053,9 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise=", fx.episodes.start) - .where("episode_number", "<=", fx.episodes.end) - .execute(); + if (isMain && fx?.episodes && frozen.mediaType === "tv") { + // 覆盖按实际文件:整季包里的每个文件按文件名解析出的集号登记;单元未建档的建临时单元(provisional) + const units = await ensureEpisodeUnits(trx, frozen.entityId, fx.season ?? 1, fx.episodes); for (const u of units) { await trx .insertInto("asset_unit_coverage") @@ -1058,7 +1086,8 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise s === "failed").length, file_skipped: statuses.filter((s) => s === "skipped").length, }; - const okAny = mainAssets.length > 0; + const verifiedAssets = [...mainAssets, ...presentAssets]; + const okAny = verifiedAssets.length > 0; await trx .updateTable("organization_record") .set({ @@ -1096,16 +1125,16 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise u.id); if (required.length === 0) return false; + // 总集数只有在单元都已确认(元数据/人工)时才算已知;由发布标题或文件名临时建出的单元(provisional) + // 只说明"见过这些集",不说明"只有这些集":总集数未知时不判完整,订阅继续追踪(REQ §7.1.1) + const totalKnown = units.every((u) => u.status === "confirmed"); + if (!totalKnown) return false; const cov = await sql<{ asset_id: string; unit_id: string }>` select c.media_asset_id as asset_id, c.content_unit_id as unit_id from asset_unit_coverage c join media_asset a on a.id = c.media_asset_id @@ -1200,10 +1234,57 @@ async function slotComplete( return unitCompletion( required, [...byAsset].map(([assetId, fileUnits]) => ({ assetId, verified: true, fileUnits })), - { totalKnown: true }, + { totalKnown }, ).complete; } +/** 按季号与集号区间找或建集单元(临时 provisional,挂在季单元下) */ +async function ensureEpisodeUnits( + exec: Exec, + entityId: string, + season: number, + eps: { start: number; end: number }, +): Promise<{ id: string }[]> { + const find = (kind: "season" | "episode", ep: number | null) => { + let q = exec + .selectFrom("content_unit") + .select("id") + .where("media_entity_id", "=", entityId) + .where("kind", "=", kind) + .where("season_number", "=", season) + .where("status", "!=", "retired"); + q = ep === null ? q.where("episode_number", "is", null) : q.where("episode_number", "=", ep); + return q.orderBy("created_at").executeTakeFirst(); + }; + const create = async (kind: "season" | "episode", ep: number | null, parent: string | null) => + exec + .insertInto("content_unit") + .values({ + media_entity_id: entityId, + parent_unit_id: parent, + kind, + season_number: season, + episode_number: ep, + status: "provisional", + display_order: ep ?? season, + }) + .returning("id") + .executeTakeFirstOrThrow(); + let seasonUnit: { id: string } | undefined; + const out: { id: string }[] = []; + const end = Math.min(eps.end, eps.start + 199); + for (let e = eps.start; e <= end; e++) { + const hit = await find("episode", e); + if (hit) { + out.push(hit); + continue; + } + seasonUnit ??= (await find("season", null)) ?? (await create("season", null, null)); + out.push(await create("episode", e, seasonUnit.id)); + } + return out; +} + // ── 公开记录映射 ── export async function loadRecordsPublic(exec: Exec, rows: OrganizationRecordRow[], includeFiles: boolean) { diff --git a/apps/server/src/modules/auth-settings/engine-routes.ts b/apps/server/src/modules/auth-settings/engine-routes.ts index ce1cb364..85c741af 100644 --- a/apps/server/src/modules/auth-settings/engine-routes.ts +++ b/apps/server/src/modules/auth-settings/engine-routes.ts @@ -56,8 +56,20 @@ const D = { response: EngineCancelResult, tags, }, + retry: { + method: "post", + path: "/engine/runs/{id}/retry", + scope: "admin", + summary: "按原任务名与输入重新入队(新幂等键 retry::<序号>)", + params: RunIdParam, + response: z.object({ runId: z.string(), retryOf: z.string(), attempt: z.number().int().positive() }), + tags, + }, } as const satisfies Record; +/** 可重跑的终态(运行中/待执行的 run 先取消再重跑) */ +const RETRYABLE = new Set(["completed", "succeeded", "failed", "canceled"]); + const NAMESPACE = "default"; const iso = (d: Date | null) => (d ? d.toISOString() : null); @@ -257,4 +269,39 @@ export function registerEngineRoutes(app: ModuleApp, ctx: ServerContext, opts: { ); return { canceled }; }); + // 重跑:读原 run 的任务名与信封里的原始输入(不经脱敏),以新幂等键重新入队;业务侧幂等仍由各任务的 op 账本保证 + endpoint(app, D.retry, async ({ params, actor }) => { + const t = sql.raw(`"${opts.schema ?? "openworkflow"}".workflow_runs`); + const r = await sql<{ workflow_name: string; status: string; input: unknown }>` + select workflow_name, status, input from ${t} where namespace_id = ${NAMESPACE} and id = ${params.id}`.execute( + ctx.db, + ); + const run = r.rows[0]; + if (!run) throw new ApiException("not_found", `run ${params.id} 不存在`); + if (!RETRYABLE.has(run.status)) + throw new ApiException("conflict", `run 状态为 ${run.status},只能重跑已结束的 run(可先取消)`); + const env = (run.input ?? {}) as { input?: unknown; traceId?: string; priority?: number }; + const prefix = `retry:${params.id}:`; + const prev = await sql<{ n: number }>` + select count(*)::int as n from tasks_run_meta where idempotency_key like ${`${prefix}%`}`.execute(ctx.db); + const attempt = (prev.rows[0]?.n ?? 0) + 1; + let runId: string; + try { + ({ runId } = await ctx.engine.enqueue(run.workflow_name, "input" in env ? env.input : {}, { + idempotencyKey: `${prefix}${attempt}`, + traceId: env.traceId ?? `retry:${params.id}`, + ...(typeof env.priority === "number" ? { priority: env.priority } : {}), + })); + } catch (err) { + throw new ApiException("precondition_failed", `无法重新入队:${(err as Error).message}`); + } + await audit( + ctx, + actor, + "engine.run_retry", + { type: "engine_run", id: params.id }, + { workflowName: run.workflow_name, newRunId: runId, attempt }, + ); + return { runId, retryOf: params.id, attempt }; + }); } diff --git a/apps/server/test/auth-settings/engine-retry.test.ts b/apps/server/test/auth-settings/engine-retry.test.ts new file mode 100644 index 00000000..9af8a82c --- /dev/null +++ b/apps/server/test/auth-settings/engine-retry.test.ts @@ -0,0 +1,64 @@ +import { sql } from "@umt/db"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { DEV_KEY, json } from "../helpers.ts"; +import { bearer, createHarness, type Harness, pgAvailable } from "./harness.ts"; + +// 引擎面板 retry(验收 S8):仅 admin;按原任务名与信封里的原始输入重新入队,幂等键 retry::<序号>; +// 运行中/待执行的 run 不能重跑(409);未知 run 404;写审计。 + +const hasPg = await pgAvailable(); + +describe.skipIf(!hasPg)("POST /engine/runs/{id}/retry", () => { + let h: Harness; + let admin: ReturnType; + let n = 0; + const enqueue = vi.fn(async (_task: string, _input: unknown, _opts: { idempotencyKey: string }) => { + n++; + await sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key) + values (${`new-${n}`}, ${_task}, 1, ${_opts.idempotencyKey})`.execute(h.db); + return { runId: `new-${n}`, deduplicated: false }; + }); + + beforeAll(async () => { + h = await createHarness({ engine: true }); + admin = bearer(h.app, DEV_KEY); + (h.ctx.engine as unknown as { enqueue: typeof enqueue }).enqueue = enqueue; + const t = sql.raw(`"${h.engineSchema}".workflow_runs`); + const run = (id: string, status: string, input: unknown) => + sql`insert into ${t} (namespace_id, id, workflow_name, version, status, config, context, input, attempts, + created_at, updated_at) + values ('default', ${id}, 'verifyAndImport', 'v1', ${status}, '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 3, + now(), now())`.execute(h.db); + // 信封里的原始输入含敏感字段:重跑用原值(详情展示才脱敏) + await run("f1", "failed", { input: { downloadTaskId: "dt-1", password: "p@ss" }, traceId: "trace-1", priority: 5 }); + await run("p1", "pending", { input: { downloadTaskId: "dt-2" }, traceId: "trace-2", priority: 0 }); + }, 60_000); + afterAll(async () => { + await h?.close(); + }); + + it("失败的 run:按原任务名与原始输入重新入队,序号递增,写审计", async () => { + const r1 = await json(await admin("POST", "/engine/runs/f1/retry")); + expect(r1.data).toEqual({ runId: "new-1", retryOf: "f1", attempt: 1 }); + expect(enqueue).toHaveBeenLastCalledWith( + "verifyAndImport", + { downloadTaskId: "dt-1", password: "p@ss" }, + { idempotencyKey: "retry:f1:1", traceId: "trace-1", priority: 5 }, + ); + const r2 = await json(await admin("POST", "/engine/runs/f1/retry")); + expect(r2.data).toMatchObject({ runId: "new-2", attempt: 2 }); + expect(enqueue.mock.calls.at(-1)![2].idempotencyKey).toBe("retry:f1:2"); + const a = await sql<{ n: number }>` + select count(*)::int as n from audit_event where action = 'engine.run_retry'`.execute(h.db); + expect(a.rows[0]!.n).toBe(2); + }); + + it("未结束的 run 409,未知 run 404,非 admin 403", async () => { + expect((await admin("POST", "/engine/runs/p1/retry")).status).toBe(409); + expect((await admin("POST", "/engine/runs/nope/retry")).status).toBe(404); + const created = await json(await admin("POST", "/api-keys", { name: "exec", scopes: ["organize.execute"] })); + const other = bearer(h.app, created.data.plaintextKey); + expect((await other("POST", "/engine/runs/f1/retry")).status).toBe(403); + expect(n).toBe(2); + }); +}); diff --git a/apps/server/test/e2e/e7-tv-series.test.ts b/apps/server/test/e2e/e7-tv-series.test.ts new file mode 100644 index 00000000..4d8b0aae --- /dev/null +++ b/apps/server/test/e2e/e7-tv-series.test.ts @@ -0,0 +1,308 @@ +import { existsSync } from "node:fs"; +import { copyFile, mkdir } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { createMockRssServer, type MockFeedItem, type MockRssServer } 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"; + +// E7 常规剧集纵切(验收 S5,REQ §7.1.1 / §8.4):全部经真实任务引擎 + outbox 中继。 +// mock RSS 依次出 S01E01、S01E02 单集,最后出一个整季包;已发布收录策略(剧集 1080p Slot),订阅改为持续追踪。 +// 断言: +// · 两集分别入库到同一个已绑定的季目录(季绑定只建一次,之后固定); +// · content_unit 覆盖按实际文件计算(E01→E01、E02→E02); +// · 整季包"声称"整季,但实际文件只有 E01/E02(与库内相同)→ 不新增覆盖,Slot 不因包名判满足; +// · 第三集从未出现:Slot 仍未 satisfied,订阅保持 active/continuous(总集数未知不结束)。 + +const hasPg = await pgAvailable(); +const BASE = "https://audiences.example"; +const E01 = "The.Bear.S01E01.1080p.WEB-DL.x264-GRP"; +const E02 = "The.Bear.S01E02.1080p.WEB-DL.x264-GRP"; +const PACK = "The.Bear.S01.1080p.WEB-DL.x264-GRP"; + +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=E7PASSKEY`, + size: TORRENT_SIZE, + pubDate: new Date(), + categories: ["剧集"], +}); + +describe.skipIf(!hasPg)("E7 常规剧集纵切", () => { + let e: E2E; + let rss: MockRssServer; + let base: Awaited>; + let feedId: string; + let entityId: string; + let slotId: string; + let subId: string; + + // 手动抓取按 10 秒时间槽去重(ingestFeed:manual::<槽>):同一槽内的第二次请求会被合并,等到下一个槽再发 + 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 importedFor = async (title: string) => { + const hash = torrentFor(title).hash; + const dt = await waitFor( + async () => e.db.selectFrom("download_task").selectAll().where("torrent_hash", "=", hash).executeTakeFirst(), + 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} 整理记录结束`, + ); + }; + const slot = () => e.db.selectFrom("acquisition_slot").selectAll().where("id", "=", slotId).executeTakeFirstOrThrow(); + const sub = () => e.db.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirstOrThrow(); + const coverage = async () => { + const r = 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(["a.id as asset", "u.season_number", "u.episode_number"]) + .where("a.media_entity_id", "=", entityId) + .where("a.status", "=", "active") + .orderBy("u.episode_number") + .execute(); + return r.map((x) => ({ asset: x.asset, unit: `S${x.season_number}E${x.episode_number}` })); + }; + + beforeAll(async () => { + e = await createE2E("e7"); + base = await seedBase(e, { baseUrl: `${BASE}/` }); + rss = await createMockRssServer({ passkey: "E7PASSKEY", items: [item(910001, E01)] }); + const urlRef = await e.booted.ctx.secrets.create({ purpose: "feed_url", plaintext: rss.feedUrl }); + const feed = 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(); + feedId = feed.id; + }, 120_000); + + afterAll(async () => { + await rss?.close(); + await e?.close(); + }, 60_000); + + it("发布剧集 1080p 收录策略 → 持续授权", async () => { + 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); + }); + + it("S01E01:识别为剧集 → 订阅改为持续追踪 → 下载 → 入库到季目录;总集数未知,Slot 不判满足", async () => { + await fetchFeed(); + const rel = await waitFor( + async () => + e.db + .selectFrom("source_release") + .selectAll() + .where("title", "=", E01) + .where("resolution_status", "=", "resolved") + .executeTakeFirst(), + 60_000, + "S01E01 识别完成", + ); + entityId = rel.media_entity_id!; + const entity = await e.db + .selectFrom("media_entity") + .selectAll() + .where("id", "=", entityId) + .executeTakeFirstOrThrow(); + expect(entity.kind).toBe("series"); + const s = await waitFor( + async () => + e.db.selectFrom("subscription").selectAll().where("media_entity_id", "=", entityId).executeTakeFirst(), + 30_000, + "新作品订阅", + ); + subId = s.id; + const patched = await e.api("PATCH", `/subscriptions/${subId}`, { mode: "continuous" }); + expect(patched.status).toBe(200); + slotId = ( + await e.db + .selectFrom("acquisition_slot") + .select("id") + .where("media_entity_id", "=", entityId) + .executeTakeFirstOrThrow() + ).id; + + await completeDownload(e, E01); + const rec = await importedFor(E01); + expect(rec).toMatchObject({ status: "succeeded", action: "copy", trigger: "auto" }); + expect((await coverage()).map((c) => c.unit)).toEqual(["S1E1"]); + expect((await slot()).status).toBe("empty"); + expect(await sub()).toMatchObject({ status: "active", mode: "continuous" }); + }, 180_000); + + let seasonDir: string; + it("S01E02:第二集进同一个已绑定的季目录,覆盖逐集计算,Slot 仍未满足", async () => { + const bindings0 = await e.db + .selectFrom("path_binding") + .select(["binding_kind", "season_number", "relative_path"]) + .where("media_entity_id", "=", entityId) + .orderBy("binding_kind") + .execute(); + expect(bindings0.map((b) => `${b.binding_kind}:${b.season_number ?? "-"}`)).toEqual(["entity:-", "season:1"]); + seasonDir = bindings0.find((b) => b.binding_kind === "season")!.relative_path; + + rss.addItem(item(910002, E02)); + await fetchFeed(); + await completeDownload(e, E02); + const rec = await importedFor(E02); + expect(rec.status).toBe("succeeded"); + + const files = await e.db + .selectFrom("organization_file_result as f") + .innerJoin("organization_record as r", "r.id", "f.organization_record_id") + .select(["f.target_path", "f.status"]) + .where("r.media_entity_id", "=", entityId) + .where("f.role", "=", "main") + .orderBy("f.target_path") + .execute(); + expect(files).toHaveLength(2); + for (const f of files) { + expect(f.status).toBe("succeeded"); + expect(dirname(f.target_path!)).toBe(seasonDir); + expect(existsSync(join(e.lib.root, f.target_path!))).toBe(true); + } + expect(files.map((f) => f.target_path!)).toEqual([ + expect.stringMatching(/S01E01\.mkv$/), + expect.stringMatching(/S01E02\.mkv$/), + ]); + // 路径绑定固定:没有因为第二集新建任何绑定 + const bindings1 = await e.db + .selectFrom("path_binding") + .select(["binding_kind", "season_number", "relative_path"]) + .where("media_entity_id", "=", entityId) + .orderBy("binding_kind") + .execute(); + expect(bindings1).toEqual(bindings0); + + const cov = await coverage(); + expect(cov.map((c) => c.unit)).toEqual(["S1E1", "S1E2"]); + expect(new Set(cov.map((c) => c.asset)).size).toBe(2); + expect((await slot()).status).toBe("empty"); + expect(await sub()).toMatchObject({ status: "active", mode: "continuous" }); + }, 180_000); + + it("整季包:实际文件只有 E01/E02(与库内相同)→ 不新增覆盖,不因包名判满足;第三集未出现,订阅继续", async () => { + const libFiles = await e.db + .selectFrom("organization_file_result as f") + .innerJoin("organization_record as r", "r.id", "f.organization_record_id") + .select(["f.target_path"]) + .where("r.media_entity_id", "=", entityId) + .where("f.role", "=", "main") + .where("f.status", "=", "succeeded") + .orderBy("f.target_path") + .execute(); + const assetsBefore = await e.db + .selectFrom("media_asset") + .select("id") + .where("media_entity_id", "=", entityId) + .execute(); + + rss.addItem(item(910003, PACK)); + await fetchFeed(); + const packRel = await waitFor( + async () => + e.db + .selectFrom("source_release") + .selectAll() + .where("title", "=", PACK) + .where("resolution_status", "=", "resolved") + .executeTakeFirst(), + 60_000, + "整季包识别完成", + ); + expect(packRel.media_entity_id).toBe(entityId); + const hash = torrentFor(PACK).hash; + await waitFor(async () => e.qb.get(hash), 60_000, "qB 收到整季包"); + + // 包内真实文件清单:只有 E01、E02,内容与已入库的相同(常见的"同组合集") + const pack = [E01, E02].map((t, i) => ({ + path: `${PACK}/${t}.mkv`, + src: join(e.lib.root, libFiles[i]!.target_path!), + })); + for (const p of pack) { + const dst = join(e.cache.root, p.path); + await mkdir(dirname(dst), { recursive: true }); + await copyFile(p.src, dst); + } + await e.qb.complete(hash, { files: pack.map((p) => ({ path: p.path, size: TORRENT_SIZE })), writeFiles: false }); + const rec = await importedFor(PACK); + expect(rec.status).toBe("succeeded"); + const packFiles = await e.db + .selectFrom("organization_file_result") + .select(["status", "skip_reason", "media_asset_id"]) + .where("organization_record_id", "=", rec.id) + .where("role", "=", "main") + .execute(); + expect(packFiles).toHaveLength(2); + for (const f of packFiles) { + expect(f).toMatchObject({ status: "skipped", skip_reason: "already_in_library" }); + expect(assetsBefore.map((a) => a.id)).toContain(f.media_asset_id); + } + // 没有新资产、没有新覆盖;季单元的"整季"声明不算覆盖,也没有凭空出现第三集 + expect( + await e.db.selectFrom("media_asset").select("id").where("media_entity_id", "=", entityId).execute(), + ).toHaveLength(assetsBefore.length); + expect((await coverage()).map((c) => c.unit)).toEqual(["S1E1", "S1E2"]); + const eps = await e.db + .selectFrom("content_unit") + .select(["episode_number", "status"]) + .where("media_entity_id", "=", entityId) + .where("kind", "=", "episode") + .orderBy("episode_number") + .execute(); + expect(eps).toEqual([ + { episode_number: 1, status: "provisional" }, + { episode_number: 2, status: "provisional" }, + ]); + // 第三集没出现、总集数未知:Slot 不满足,订阅继续追踪 + const s = await slot(); + expect(s.status).not.toBe("satisfied"); + expect(await sub()).toMatchObject({ status: "active", mode: "continuous" }); + // 三个种子各提交一次 + expect(e.qb.callsTo("torrents/add")).toHaveLength(3); + }, 180_000); +}); diff --git a/apps/server/test/e2e/e8-tag-removal.test.ts b/apps/server/test/e2e/e8-tag-removal.test.ts new file mode 100644 index 00000000..2ecd70fd --- /dev/null +++ b/apps/server/test/e2e/e8-tag-removal.test.ts @@ -0,0 +1,259 @@ +import { createQbClient } from "@umt/adapters"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import type { createAcquisitionOrganizeModule } from "../../src/modules/acquisition-organize/index.ts"; +import { syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts"; +import { torrentFor } from "./fixtures/modules.ts"; +import { createE2E, type E2E, pgAvailable, seedBase, waitFor } from "./harness.ts"; + +// E8 AT24(REQ §8.4):管理标签被移除 / qB 任务消失。真实任务引擎 + worker + 后台同步循环 + mock qB。 +// A(已入库)、B(下载中)被用户在 qB 里去掉管理标签 → managed=false:B 之后完成也不入库,A 不再被缓存清理; +// 历史(download_task 行、状态事件)与资产保留。 +// D(已入库)在 qB 里被外部删除 → download_task=removed,media_asset / asset_location 不变,Slot 仍 satisfied,不重新下载。 +// C(已入库、仍纳管)作对照:同一轮缓存清理里被删,证明清理确实跑过。 +// 断言以 mock qB 调用日志为准:add 只有最初 4 次;系统发出的 delete 只针对 C。 + +const hasPg = await pgAvailable(); +const T = { + A: "Alpha.Tag.2020.1080p.BluRay.x264-GRP", + B: "Beta.Tag.2020.1080p.BluRay.x264-GRP", + C: "Gamma.Tag.2020.1080p.BluRay.x264-GRP", + D: "Delta.Tag.2020.1080p.BluRay.x264-GRP", +} as const; +type K = keyof typeof T; + +describe.skipIf(!hasPg)("E8 AT24 标签移除 / 任务消失", () => { + let e: E2E; + let base: Awaited>; + const ids = {} as Record; + const deps = () => + ( + e.modules.find((m) => m.name === "acquisition-organize") as ReturnType + ).deps(); + const dt = (k: K) => + e.db.selectFrom("download_task").selectAll().where("id", "=", ids[k].dtId!).executeTakeFirstOrThrow(); + const hashesOf = (endpoint: string) => + e.qb.callsTo(endpoint).flatMap((c) => (c.form.hashes ?? "").split("|").filter(Boolean)); + + beforeAll(async () => { + e = await createE2E("e8"); + base = await seedBase(e); + let n = 0; + for (const k of Object.keys(T) as K[]) { + const ent = await e.db + .insertInto("media_entity") + .values({ kind: "movie", title: `${k} 标签片`, year: 2020, status: "active" }) + .returning("id") + .executeTakeFirstOrThrow(); + const sub = await e.db + .insertInto("subscription") + .values({ media_entity_id: ent.id, media_kind: "movie", mode: "once" }) + .returning("id") + .executeTakeFirstOrThrow(); + const slot = await e.db + .insertInto("acquisition_slot") + .values({ + media_entity_id: ent.id, + subscription_id: sub.id, + name: "1080p", + hard_conditions: JSON.stringify({ resolutions: ["1080p"] }), + }) + .returning("id") + .executeTakeFirstOrThrow(); + const rel = await e.db + .insertInto("source_release") + .values({ + site_id: base.siteId, + site_release_id: String(770000 + ++n), + title: T[k], + size_bytes: 300_000, + media_entity_id: ent.id, + resolution_status: "resolved", + raw: JSON.stringify({ _site: { hr: false } }), + }) + .returning("id") + .executeTakeFirstOrThrow(); + ids[k] = { entityId: ent.id, slotId: slot.id, releaseId: rel.id, hash: torrentFor(T[k]).hash }; + } + }, 120_000); + + afterAll(async () => { + await e?.close(); + }, 60_000); + + const submit = async (k: K) => { + const r = await e.api("POST", "/acquisition-plans", { sourceReleaseId: ids[k].releaseId }); + expect(r.status, JSON.stringify(r.body)).toBe(200); + expect(r.body.data.submitted).toBe(true); + await waitFor(async () => e.qb.get(ids[k].hash), 60_000, `qB 收到 ${k}`); + const row = await waitFor( + async () => + e.db.selectFrom("download_task").select("id").where("torrent_hash", "=", ids[k].hash).executeTakeFirst(), + 30_000, + `${k} download_task`, + ); + ids[k].dtId = row.id; + }; + const imported = async (k: K) => { + await e.qb.complete(ids[k].hash, { writeFiles: true }); + await waitFor( + async () => + e.db + .selectFrom("organization_record") + .select("id") + .where("download_task_id", "=", ids[k].dtId!) + .where("status", "=", "succeeded") + .executeTakeFirst(), + 90_000, + `${k} 入库`, + ); + }; + + it("准备:A/C/D 下载完成并 Copy 入库,B 仍在下载", async () => { + for (const k of ["A", "C", "D"] as K[]) { + await submit(k); + await imported(k); + } + await submit("B"); + expect((await dt("B")).status).not.toBe("completed"); + for (const k of ["A", "C", "D"] as K[]) { + expect(await dt(k)).toMatchObject({ status: "imported", managed: true }); + const slot = await e.db + .selectFrom("acquisition_slot") + .select("status") + .where("id", "=", ids[k].slotId) + .executeTakeFirstOrThrow(); + expect(slot.status).toBe("satisfied"); + } + expect(e.qb.callsTo("torrents/add")).toHaveLength(4); + }, 300_000); + + it("移除管理标签 → managed=false;之后完成也不入库、不再被自动动作触碰,历史保留", async () => { + const tag = ( + await e.db + .selectFrom("downloader_instance") + .select("managed_tag") + .where("id", "=", base.instanceId) + .executeTakeFirstOrThrow() + ).managed_tag; + for (const k of ["A", "B"] as K[]) { + const t = e.qb.get(ids[k].hash)!; + expect(t.tags).toContain(tag); + t.tags = t.tags.filter((x) => x !== tag); + } + await syncDownloads(deps()); + for (const k of ["A", "B"] as K[]) { + const row = await dt(k); + expect(row.managed).toBe(false); + expect(row.removed_at).toBeNull(); + expect(row.tags).not.toContain(tag); + } + const bStatus = (await dt("B")).status; + const statusEventsA = (await e.db.selectFrom("outbox_event").select(["type", "payload"]).execute()).filter( + (x) => + x.type === "download.status_changed" && + (x.payload as { downloadTaskId?: string }).downloadTaskId === ids.A.dtId, + ); + expect(statusEventsA.length).toBeGreaterThan(0); + + // B 在 qB 里完成:不再同步状态、不发完成事件、不入库 + await e.qb.complete(ids.B.hash, { writeFiles: true }); + for (let i = 0; i < 3; i++) await syncDownloads(deps()); + await new Promise((r) => setTimeout(r, 1_500)); + expect((await dt("B")).status).toBe(bStatus); + expect( + await e.db.selectFrom("organization_record").select("id").where("download_task_id", "=", ids.B.dtId!).execute(), + ).toHaveLength(0); + const bImports = await e.db + .selectFrom("task") + .select("id") + .where("kind", "=", "verifyAndImport") + .where("subject_id", "=", ids.B.dtId!) + .execute(); + expect(bImports).toHaveLength(0); + // A 的历史与资产原样 + const assetsA = await e.db + .selectFrom("media_asset as a") + .innerJoin("asset_location as l", "l.media_asset_id", "a.id") + .select(["a.status", "l.status as loc"]) + .where("a.download_task_id", "=", ids.A.dtId!) + .execute(); + expect(assetsA).toEqual([{ status: "active", loc: "available" }]); + const after = (await e.db.selectFrom("outbox_event").select(["type", "payload"]).execute()).filter( + (x) => + x.type === "download.status_changed" && + (x.payload as { downloadTaskId?: string }).downloadTaskId === ids.A.dtId, + ); + expect(after.length).toBe(statusEventsA.length); + }, 120_000); + + it("qB 任务消失 → removed;资产与位置不变,Slot 仍 satisfied,不重新下载", async () => { + const before = await e.db + .selectFrom("media_asset as a") + .innerJoin("asset_location as l", "l.media_asset_id", "a.id") + .select(["a.id", "a.status", "l.id as loc_id", "l.status as loc", "l.relative_path"]) + .where("a.download_task_id", "=", ids.D.dtId!) + .execute(); + expect(before).toHaveLength(1); + // 外部(用户在 qB 里)删掉 D;这一条 delete 是测试发出的,不是系统 + const ext = createQbClient({ + instanceId: base.instanceId, + baseUrl: e.qb.url, + username: "admin", + password: "adminadmin", + }); + await ext.delete([ids.D.hash], { deleteFiles: false }); + const externalDeletes = e.qb.callsTo("torrents/delete").length; + await syncDownloads(deps()); + const row = await dt("D"); + expect(row.status).toBe("removed"); + expect(row.removed_at).not.toBeNull(); + const after = await e.db + .selectFrom("media_asset as a") + .innerJoin("asset_location as l", "l.media_asset_id", "a.id") + .select(["a.id", "a.status", "l.id as loc_id", "l.status as loc", "l.relative_path"]) + .where("a.download_task_id", "=", ids.D.dtId!) + .execute(); + expect(after).toEqual(before); + // 几轮后台同步 + 编排之后:没有新计划、没有新 add + await new Promise((r) => setTimeout(r, 1_500)); + const slot = await e.db + .selectFrom("acquisition_slot") + .select("status") + .where("id", "=", ids.D.slotId) + .executeTakeFirstOrThrow(); + expect(slot.status).toBe("satisfied"); + const plans = await e.db.selectFrom("acquisition_plan").select("id").where("slot_id", "=", ids.D.slotId).execute(); + expect(plans).toHaveLength(1); + expect(e.qb.callsTo("torrents/add")).toHaveLength(4); + expect(e.qb.callsTo("torrents/delete")).toHaveLength(externalDeletes); + }, 60_000); + + it("缓存清理:纳管的 C 被清,去标签的 A 不碰(qB 调用日志)", async () => { + await e.settings({ cache: { hardThresholdPercent: 30, resumePercent: 25, cleanupTargetPercent: 20 } }); + const r = await e.api("POST", "/download-cache/cleanup", {}); + expect(r.status).toBe(202); + await waitFor( + async () => { + const t = await e.db + .selectFrom("task") + .select("status") + .where("id", "=", r.body.data.taskId) + .executeTakeFirst(); + return t?.status === "succeeded" ? t : null; + }, + 60_000, + "cleanup 任务完成", + ); + expect((await dt("C")).status).toBe("removed"); + const deleted = hashesOf("torrents/delete"); + expect(deleted).toContain(ids.C.hash); + expect(deleted).not.toContain(ids.A.hash); + expect(deleted).not.toContain(ids.B.hash); + // 系统发出的带数据删除只有 C;D 那一条是外部 deleteFiles=false + const withData = e.qb.callsTo("torrents/delete").filter((c) => c.form.deleteFiles === "true"); + expect(withData.flatMap((c) => c.form.hashes!.split("|"))).toEqual([ids.C.hash]); + expect(e.qb.get(ids.A.hash)).toBeDefined(); + expect(await dt("A")).toMatchObject({ managed: false, status: "imported", removed_at: null }); + expect(e.qb.callsTo("torrents/add")).toHaveLength(4); + }, 90_000); +}); diff --git a/apps/web/app/api/engine.ts b/apps/web/app/api/engine.ts index 93f69043..02d16e75 100644 --- a/apps/web/app/api/engine.ts +++ b/apps/web/app/api/engine.ts @@ -48,3 +48,16 @@ export function useCancelEngineRun() { onSettled: () => qc.invalidateQueries({ queryKey: qk.all("engine") }), }); } + +/** 重跑已结束的 run:服务端按原任务名与原始输入重新入队,返回新 runId */ +export function useRetryEngineRun() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: ({ id }: { id: string }) => + apiData<{ runId: string; retryOf: string; attempt: number }>("/engine/runs/{id}/retry", { + method: "POST", + pathParams: { id }, + }), + onSettled: () => qc.invalidateQueries({ queryKey: qk.all("engine") }), + }); +} diff --git a/apps/web/app/routes/console/engine.$runId.tsx b/apps/web/app/routes/console/engine.$runId.tsx index 8db6ebb8..4b7c34b5 100644 --- a/apps/web/app/routes/console/engine.$runId.tsx +++ b/apps/web/app/routes/console/engine.$runId.tsx @@ -1,8 +1,8 @@ import type { EngineStepAttempt } from "@umt/contracts"; -import { ArrowLeft, Ban, RotateCw } from "lucide-react"; -import { Link, useParams } from "react-router"; +import { ArrowLeft, Ban, Repeat, RotateCw } from "lucide-react"; +import { Link, useNavigate, useParams } from "react-router"; import { toast } from "sonner"; -import { ENGINE_TERMINAL, useCancelEngineRun, useEngineRun } from "~/api/engine.ts"; +import { ENGINE_TERMINAL, useCancelEngineRun, useEngineRun, useRetryEngineRun } from "~/api/engine.ts"; import { ConfirmDialog } from "~/components/console/confirm-dialog.tsx"; import { DetailSection } from "~/components/console/detail-sheet.tsx"; import { EmptyState, ErrorState, LoadingState } from "~/components/console/empty-state.tsx"; @@ -93,6 +93,8 @@ export default function EngineRunPage() { const { runId } = useParams(); const q = useEngineRun(runId); const cancel = useCancelEngineRun(); + const retry = useRetryEngineRun(); + const navigate = useNavigate(); const back = ( - ) : null} + ) : ( + + + 重跑 + + } + title="重跑这个 run？" + description="按原任务名和原始输入新建一个 run（新的幂等键），原 run 保持不变。已完成的文件操作由任务自己的操作账本去重。" + confirmLabel="重跑" + cancelLabel="不重跑" + onConfirm={async () => { + try { + const res = await retry.mutateAsync({ id: r.id }); + toast.success(`已重新入队（第 ${res.attempt} 次重跑）`); + navigate(consolePath("engine", res.runId)); + } catch (err) { + toast.error(errorMessage(err, "重跑失败")); + throw err; + } + }} + /> + )} } /> diff --git a/apps/web/package.json b/apps/web/package.json index 7447fee6..85775dd1 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -1,6 +1,6 @@ { "name": "@umt/web", - "version": "0.0.0", + "version": "0.1.0", "private": true, "type": "module", "scripts": { diff --git a/deno.json b/deno.json index 5a4ee0da..4314baaa 100644 --- a/deno.json +++ b/deno.json @@ -1,4 +1,5 @@ { + "version": "0.1.0", "workspace": [ "./packages/adapters", "./packages/agent", diff --git a/packages/adapters/deno.json b/packages/adapters/deno.json index 11d860de..5e6de606 100644 --- a/packages/adapters/deno.json +++ b/packages/adapters/deno.json @@ -1,6 +1,6 @@ { "name": "@umt/adapters", - "version": "0.0.0", + "version": "0.1.0", "exports": { ".": "./src/index.ts" }, "tasks": { "test": "deno run -A npm:vitest run", diff --git a/packages/agent/deno.json b/packages/agent/deno.json index 7e2f2bbb..881c2b48 100644 --- a/packages/agent/deno.json +++ b/packages/agent/deno.json @@ -1,6 +1,6 @@ { "name": "@umt/agent", - "version": "0.0.0", + "version": "0.1.0", "exports": { ".": "./src/index.ts" }, "tasks": { "test": "deno run -A npm:vitest run", diff --git a/packages/contracts/deno.json b/packages/contracts/deno.json index 5e9afceb..f50d86fd 100644 --- a/packages/contracts/deno.json +++ b/packages/contracts/deno.json @@ -1,6 +1,6 @@ { "name": "@umt/contracts", - "version": "0.0.0", + "version": "0.1.0", "exports": { ".": "./src/index.ts" }, "tasks": { "test": "deno run -A npm:vitest run", diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index 9d68b95f..99aed1f7 100644 --- a/packages/contracts/src/api.ts +++ b/packages/contracts/src/api.ts @@ -1292,6 +1292,14 @@ export const Api = { body: ops.EngineCancelRequest, response: ops.EngineCancelResult, }), + retry: ep({ + method: "post", + path: "/engine/runs/{id}/retry", + scope: "admin", + summary: "按原任务名与输入重新入队(新幂等键 retry::<序号>)", + params: EngineRunIdParam, + response: z.object({ runId: z.string(), retryOf: z.string(), attempt: z.number().int().positive() }), + }), }, // ── 待识别池 ── diff --git a/packages/contracts/src/auth.ts b/packages/contracts/src/auth.ts index 34c006ae..3308b8d9 100644 --- a/packages/contracts/src/auth.ts +++ b/packages/contracts/src/auth.ts @@ -139,5 +139,6 @@ export const AuditAction = z.enum([ "path_mapping.verify", "site_account.credentials_set", "engine.run_cancel", + "engine.run_retry", ]); export type AuditAction = z.infer; diff --git a/packages/db/deno.json b/packages/db/deno.json index f3baa771..9d39563d 100644 --- a/packages/db/deno.json +++ b/packages/db/deno.json @@ -1,6 +1,6 @@ { "name": "@umt/db", - "version": "0.0.0", + "version": "0.1.0", "exports": { ".": "./src/index.ts" }, "tasks": { "test": "deno run -A npm:vitest run", diff --git a/packages/domain/deno.json b/packages/domain/deno.json index b5360999..a89a7c10 100644 --- a/packages/domain/deno.json +++ b/packages/domain/deno.json @@ -1,6 +1,6 @@ { "name": "@umt/domain", - "version": "0.0.0", + "version": "0.1.0", "exports": { ".": "./src/index.ts" }, "tasks": { "test": "deno run -A npm:vitest run", diff --git a/packages/mapping/deno.json b/packages/mapping/deno.json index f897d6e1..40cae034 100644 --- a/packages/mapping/deno.json +++ b/packages/mapping/deno.json @@ -1,6 +1,6 @@ { "name": "@umt/mapping", - "version": "0.0.0", + "version": "0.1.0", "exports": { ".": "./src/index.ts" }, "tasks": { "test": "deno run -A npm:vitest run", diff --git a/packages/tasks/deno.json b/packages/tasks/deno.json index 9437a212..6a5104ac 100644 --- a/packages/tasks/deno.json +++ b/packages/tasks/deno.json @@ -1,6 +1,6 @@ { "name": "@umt/tasks", - "version": "0.0.0", + "version": "0.1.0", "exports": { ".": "./src/index.ts" }, "tasks": { "test": "deno run -A npm:vitest run", diff --git a/packages/testkit/deno.json b/packages/testkit/deno.json index 8d5ef137..59c97e68 100644 --- a/packages/testkit/deno.json +++ b/packages/testkit/deno.json @@ -1,6 +1,6 @@ { "name": "@umt/testkit", - "version": "0.0.0", + "version": "0.1.0", "exports": { ".": "./src/index.ts" }, "tasks": { "test": "deno run -A npm:vitest run",