From 079c3eafe9453d545feed8cd06bfa6979b00602a Mon Sep 17 00:00:00 2001 From: root Date: Tue, 6 Oct 2026 19:27:14 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=B8=9A=E5=8A=A1=E8=8A=82=E7=82=B9?= =?UTF-8?q?=E5=8F=91=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=E5=B9=B6=E8=B0=83=E7=94=A8=E9=93=BE=E5=BC=8F=E9=92=A9=E5=AD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 订阅已建 / 已改 / 已完成 / 已取消,下载完成 / 失败 / 移除,整理结束(统一 organization.changed 终态), 刮削完成 / 失败,作品进入关注,元数据已刷新(任务调用方),季集结构切换,站点不可达 / 恢复: 都在调用方事务内写 outbox,带幂等键 - 钩子挂点:name.recognize(resolveRelease / identifyParsed)、organize.rename 与 organize.intercept(planImport)、 resource.selection(订阅重评)、subscription.completion.check(completeSubscription) - 测试:各节点事件 payload 与去重、整理改名 / 拦截 / 出错不影响整理、名称识别改写、资源选择、订阅完成拦截 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../routes/subscriptions.ts | 47 ++- .../services/acquisition.ts | 51 ++- .../services/downloads.ts | 9 +- .../services/lifecycle-events.ts | 140 ++++++++ .../services/organize-hooks.ts | 89 +++++ .../services/organize-plan.ts | 17 +- .../acquisition-organize/services/scrape.ts | 2 + .../services/subscriptions.ts | 31 +- .../sources-identity/service/account.ts | 9 + .../service/episode-groups.ts | 20 +- .../service/lifecycle-events.ts | 55 +++ .../sources-identity/service/resolve.ts | 29 +- .../sources-identity/service/snapshot.ts | 26 +- .../src/modules/sources-identity/tasks.ts | 14 +- .../test/hooks/resource-selection.test.ts | 45 +++ .../test/lifecycle/lifecycle-events.test.ts | 314 ++++++++++++++++++ .../lifecycle/name-recognize-hook.test.ts | 68 ++++ .../test/lifecycle/organize-hooks.test.ts | 112 +++++++ 18 files changed, 1049 insertions(+), 29 deletions(-) create mode 100644 apps/server/src/modules/acquisition-organize/services/lifecycle-events.ts create mode 100644 apps/server/src/modules/acquisition-organize/services/organize-hooks.ts create mode 100644 apps/server/src/modules/sources-identity/service/lifecycle-events.ts create mode 100644 apps/server/test/hooks/resource-selection.test.ts create mode 100644 apps/server/test/lifecycle/lifecycle-events.test.ts create mode 100644 apps/server/test/lifecycle/name-recognize-hook.test.ts create mode 100644 apps/server/test/lifecycle/organize-hooks.test.ts diff --git a/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts b/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts index 9b5bcb49..a0a62ea9 100644 --- a/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts +++ b/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts @@ -14,6 +14,7 @@ import type { SubscriptionRow } from "@umt/db"; import type { Context } from "hono"; import type { AuthEnv } from "../../../http/auth.ts"; import { ApiException } from "../../../http/errors.ts"; +import { emitMediaWatched } from "../../sources-identity/service/lifecycle-events.ts"; import { enqueueWatchedRefresh } from "../../sources-identity/service/watched.ts"; import type { ModuleApp } from "../../types.ts"; import { ensureSubscriptionSlot } from "../services/acquisition.ts"; @@ -29,6 +30,7 @@ import { slotsOfSub, subscriptionCoverage, } from "../services/coverage.ts"; +import { emitSubscriptionEvent } from "../services/lifecycle-events.ts"; import { loadRuleCatalog } from "../services/rules.ts"; import { CALENDAR_MAX_DAYS, subscriptionCalendar } from "../services/subscription-calendar.ts"; import { resetSubscription } from "../services/subscription-reset.ts"; @@ -376,6 +378,9 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) { // 订阅总有自己的收录版本:策略模板,没有就建一个默认版本(质量预设随后由 applyDerived 写入) await ensureSubscriptionSlot(trx, r); await applyDerived(trx, r, norm, now); + // 生命周期广播(ADR-006):订阅已建;关联作品 = 作品进入关注 + await emitSubscriptionEvent(d, trx, "subscription.created", r); + if (r.media_entity_id) await emitMediaWatched(d.ctx, trx, r.media_entity_id, "subscription", r.id); return r; }); if (row.media_entity_id) { @@ -436,6 +441,7 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) { if (r.status !== "cancelled") await ensureSubscriptionSlot(trx, r); if (norm) await applyDerived(trx, r, norm, d.ports.now(), subParams(cur)); else if (body.status !== undefined) await syncPrivateRuleGroup(trx, r, d.ports.now()); + await emitSubscriptionChange(trx, cur, r, Object.keys(body)); return r; }); invalidateCoverage(r.id); @@ -446,6 +452,21 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) { { tag: TAG }, ); + /** 生命周期广播(ADR-006):改成 cancelled 发「已取消」,其余改动发「已改」 */ + const emitSubscriptionChange = async ( + trx: Exec, + before: SubscriptionRow, + after: SubscriptionRow, + changes: string[], + reason?: string, + ) => { + if (after.status === "cancelled" && before.status !== "cancelled") + await emitSubscriptionEvent(d, trx, "subscription.cancelled", after, { + reason: reason ?? after.status_reason ?? "", + }); + else if (changes.length) await emitSubscriptionEvent(d, trx, "subscription.updated", after, { changes }); + }; + // ── 状态动作 ── const setStatus = async ( sub: SubscriptionRow, @@ -456,16 +477,22 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) { const now = d.ports.now(); const searchOn = (subParams(sub).searchIntervalHours ?? DEFAULT_SEARCH_INTERVAL_HOURS) > 0; const r = await db() - .updateTable("subscription") - .set({ - status: to, - status_reason: reason, - ...(to === "paused" || to === "cancelled" ? { next_search_at: null } : {}), - ...(extra.nextSearchNow && searchOn ? { next_search_at: now } : {}), - }) - .where("id", "=", sub.id) - .returningAll() - .executeTakeFirstOrThrow(); + .transaction() + .execute(async (trx) => { + const r = await trx + .updateTable("subscription") + .set({ + status: to, + status_reason: reason, + ...(to === "paused" || to === "cancelled" ? { next_search_at: null } : {}), + ...(extra.nextSearchNow && searchOn ? { next_search_at: now } : {}), + }) + .where("id", "=", sub.id) + .returningAll() + .executeTakeFirstOrThrow(); + await emitSubscriptionChange(trx, sub, r, ["status"], reason); + return r; + }); // 私有规则组随订阅删除(取消),重新订阅时按 params 重建 if (to === "cancelled" || sub.status === "cancelled") await syncPrivateRuleGroup(db(), r, now); invalidateCoverage(sub.id); diff --git a/apps/server/src/modules/acquisition-organize/services/acquisition.ts b/apps/server/src/modules/acquisition-organize/services/acquisition.ts index 7688565b..73686d7c 100644 --- a/apps/server/src/modules/acquisition-organize/services/acquisition.ts +++ b/apps/server/src/modules/acquisition-organize/services/acquisition.ts @@ -1,4 +1,10 @@ -import type { AcquisitionPolicyRules, AcquisitionPriority, CandidateEvaluation, SlotTemplate } from "@umt/contracts"; +import { + type AcquisitionPolicyRules, + type AcquisitionPriority, + type CandidateEvaluation, + lifecycleKey, + type SlotTemplate, +} from "@umt/contracts"; import type { AcquisitionSlotRow, MediaEntityRow, SourceReleaseRow, SubscriptionRow } from "@umt/db"; import { sql } from "@umt/db"; import { @@ -23,6 +29,7 @@ import { unitsOverlap, } from "@umt/domain"; import { toContractSlotEvaluation } from "@umt/mapping"; +import { hasHook, rewriteHook } from "../../../hooks/index.ts"; import { hashJson, jsonb, startOfDay } from "../util.ts"; import { type AoDeps, @@ -38,6 +45,7 @@ import { recordCandidate, } from "./common.ts"; import { autoFillCheck, type ForceFill } from "./evaluate-autofill.ts"; +import { emitSubscriptionEvent } from "./lifecycle-events.ts"; import { contextOf, episodeResolutionOf, parsedOf, unitKeysOf } from "./release-facts.ts"; import { entityOriginalLanguage, loadRuleCatalog, type RuleCatalog } from "./rules.ts"; import { @@ -430,7 +438,12 @@ async function reevaluate(c: Ctx, input: EvaluateInput): Promise !seen.has(x.id)), ); - const ordered = await rankReleases(c, entity, sub, focus[0], [...deferred, ...fresh]); + const ranked = await rankReleases(c, entity, sub, focus[0], [...deferred, ...fresh]); + const ordered = await selectResources(c, ranked, { + mediaEntityId: entity.id, + subscriptionId: sub?.id ?? null, + slotId: input.acquisitionSlotId ?? null, + }); const scope = input.acquisitionSlotId ? { slotId: input.acquisitionSlotId } : sub ? { subscriptionId: sub.id } : {}; for (const r of ordered) await evaluateRelease(c, r, [], scope); return c.out; @@ -500,6 +513,33 @@ async function rankReleases( })).map((x) => byParsed.get(x.candidate)!); } +/** + * 链式钩子 resource.selection(ADR-006):在择优排序之后调整顺序或过滤。钩子只能重排 / 删减, + * 返回里不认识的 ID 忽略、重复的只取第一次;没有钩子时原样返回。 + */ +export async function selectResources( + c: Ctx, + rels: SourceReleaseRow[], + context: { mediaEntityId: string; subscriptionId: string | null; slotId: string | null }, +): Promise { + if (!hasHook(c.d.ctx, "resource.selection") || rels.length === 0) return rels; + const byId = new Map(rels.map((r) => [r.id, r])); + const picked = await rewriteHook( + c.d.ctx, + "resource.selection", + rels.map((r) => ({ sourceReleaseId: r.id, title: r.title, parsed: parsedOf(r) })), + context, + ); + const out: SourceReleaseRow[] = []; + for (const p of picked) { + const r = byId.get(p.sourceReleaseId); + if (!r) continue; + byId.delete(p.sourceReleaseId); + out.push(r); + } + return out; +} + /** 评估范围:只评某个收录版本,或只评某个订阅的收录版本;缺省评作品的全部收录版本 */ interface EvalScope { slotId?: string | undefined; @@ -600,8 +640,9 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon policy_revision_id: pol.row.id, created_by: `policy:${pol.row.policy_key}`, }) - .returning("id") + .returningAll() .executeTakeFirstOrThrow(); + await emitSubscriptionEvent(c.d, trx, "subscription.created", sub); await createSlotsFromTemplates(trx, { entityId: entity.id, subscriptionId: sub.id, @@ -634,6 +675,10 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon .where("id", "=", s.id) .execute(); await clearSubscriptionDeferrals(db, s.id); + await emitSubscriptionEvent(c.d, db, "subscription.completed", s, { + reason: "stop_after_reached", + key: lifecycleKey("subscription.completed", s.id, `stop_after:${stopAfter}`), + }); s.status = "completed"; } } diff --git a/apps/server/src/modules/acquisition-organize/services/downloads.ts b/apps/server/src/modules/acquisition-organize/services/downloads.ts index aa572fbf..3876cf6b 100644 --- a/apps/server/src/modules/acquisition-organize/services/downloads.ts +++ b/apps/server/src/modules/acquisition-organize/services/downloads.ts @@ -35,6 +35,7 @@ import { import { mergeObservedDownloadState, toDomainDownloadStatus } from "@umt/mapping"; import { withAttemptToken } from "@umt/tasks"; import { createEntityEpisodeMapper, publishedAtOf } from "../../sources-identity/service/episode-resolution.ts"; +import { emitMediaWatched } from "../../sources-identity/service/lifecycle-events.ts"; import { enqueueWatchedRefresh } from "../../sources-identity/service/watched.ts"; import { errMessage, jsonb } from "../util.ts"; import { parsedOf } from "./acquisition.ts"; @@ -52,6 +53,7 @@ import { timeSlot, upsertDecision, } from "./common.ts"; +import { emitDownloadLifecycle } from "./lifecycle-events.ts"; import { matchesQbSource, qbSourceTags } from "./organize-config.ts"; import { hnrSatisfied, retentionElapsed } from "./retention.ts"; import { @@ -212,6 +214,8 @@ async function emitStatus( ? { idempotencyKey: generation > 0 ? `download.completed:${id}:g${generation}` : `download.completed:${id}` } : {}, ); + // 生命周期广播(ADR-006):下载已完成 / 失败 / 已移除 + await emitDownloadLifecycle(d, exec, id, from, to); } // ── 提交 ── @@ -269,7 +273,10 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: .executeTakeFirstOrThrow(); // 开始下载 = 作品进入关注范围:从没刷新过元数据的(自动建档时不在关注范围)补排一次 full 刷新(集数两层,见 watched.ts) const watchedEntity = ev.mediaEntityId ?? rel.media_entity_id; - if (watchedEntity) await enqueueWatchedRefresh(d.ctx, watchedEntity, "download"); + if (watchedEntity) { + await enqueueWatchedRefresh(d.ctx, watchedEntity, "download"); + await emitMediaWatched(d.ctx, db, watchedEntity, "download", plan.id); + } const site = await db.selectFrom("site").selectAll().where("id", "=", rel.site_id).executeTakeFirst(); const client = await d.ports.downloader(topo.inst); const tag = topo.inst.managed_tag; diff --git a/apps/server/src/modules/acquisition-organize/services/lifecycle-events.ts b/apps/server/src/modules/acquisition-organize/services/lifecycle-events.ts new file mode 100644 index 00000000..62726a39 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/lifecycle-events.ts @@ -0,0 +1,140 @@ +import { type DownloadStatus, lifecycleKey, type OrganizationStatus } from "@umt/contracts"; +import type { ScrapeRecordRow, SubscriptionRow } from "@umt/db"; +import type { AoDeps, Exec } from "./common.ts"; + +// 生命周期广播事件(ADR-006)在本模块的发出点:都在调用方的业务事务里写 outbox,订阅方按 ID 回查业务表。 + +type SubEvent = "subscription.created" | "subscription.updated" | "subscription.completed" | "subscription.cancelled"; + +type SubRef = Pick; + +/** 订阅已建 / 已改 / 已完成 / 已取消 */ +export async function emitSubscriptionEvent( + d: AoDeps, + exec: Exec, + type: SubEvent, + sub: SubRef, + extra: { reason?: string; changes?: string[]; key?: string } = {}, +): Promise { + const ref = { subscriptionId: sub.id, mediaEntityId: sub.media_entity_id, mediaKind: sub.media_kind }; + const opts = extra.key ? { idempotencyKey: extra.key } : {}; + switch (type) { + case "subscription.created": + return d.ctx.outbox.emit( + exec, + type, + { ...ref, origin: sub.origin, status: sub.status }, + { idempotencyKey: lifecycleKey(type, sub.id) }, + ); + case "subscription.updated": + return d.ctx.outbox.emit(exec, type, { ...ref, status: sub.status, changes: extra.changes ?? [] }, opts); + case "subscription.completed": + case "subscription.cancelled": + return d.ctx.outbox.emit(exec, type, { ...ref, reason: extra.reason ?? "" }, opts); + } +} + +const DOWNLOAD_END: Partial> = { + completed: "download.completed", + failed: "download.failed", + removed: "download.removed", +}; + +/** 下载状态进入 completed / failed / removed 时发对应的生命周期事件(每代每种只发一次) */ +export async function emitDownloadLifecycle( + d: AoDeps, + exec: Exec, + id: string, + from: DownloadStatus | null, + to: DownloadStatus, +): Promise { + const type = DOWNLOAD_END[to]; + if (!type || from === to) return; + const row = await exec + .selectFrom("download_task as t") + .leftJoin("acquisition_plan as p", "p.id", "t.acquisition_plan_id") + .leftJoin("acquisition_slot as s", "s.id", "p.slot_id") + .select([ + "t.name", + "t.generation", + "t.acquisition_plan_id", + "t.media_entity_id", + "s.media_entity_id as slot_entity", + ]) + .where("t.id", "=", id) + .executeTakeFirst(); + if (!row) return; + const ref = { + downloadTaskId: id, + planId: row.acquisition_plan_id, + mediaEntityId: row.media_entity_id ?? row.slot_entity ?? null, + name: row.name, + generation: row.generation, + }; + const key = { idempotencyKey: lifecycleKey(type, id, `g${row.generation}`) }; + if (type === "download.completed") await d.ctx.outbox.emit(exec, type, ref, key); + else await d.ctx.outbox.emit(exec, type, { ...ref, from }, key); +} + +const FINISHED: readonly OrganizationStatus[] = ["succeeded", "partial", "failed"]; + +/** 整理记录进入终态时发 organization.finished(按 change_seq 去重:同一次状态变化只发一次) */ +export async function emitOrganizationFinished( + d: AoDeps, + exec: Exec, + r: { + id: string; + status: OrganizationStatus; + action: string; + trigger: string; + media_entity_id: string | null; + download_task_id: string | null; + reason: string | null; + change_seq: number | string; + }, +): Promise { + if (!FINISHED.includes(r.status)) return; + await d.ctx.outbox.emit( + exec, + "organization.finished", + { + recordId: r.id, + outcome: r.status as "succeeded" | "partial" | "failed", + action: r.action as never, + trigger: r.trigger as never, + mediaEntityId: r.media_entity_id, + downloadTaskId: r.download_task_id, + reason: r.reason, + changeSeq: String(r.change_seq), + }, + { idempotencyKey: lifecycleKey("organization.finished", r.id, String(r.change_seq)) }, + ); +} + +/** 刮削记录结束:failed → scrape.failed;succeeded / partial / skipped → scrape.completed */ +export async function emitScrapeFinished(d: AoDeps, row: ScrapeRecordRow): Promise { + if (row.status === "queued" || row.status === "running") return; + const ref = { + scrapeRecordId: row.id, + mediaEntityId: row.media_entity_id, + organizationRecordId: row.organization_record_id, + trigger: row.trigger, + status: row.status, + written: row.written, + failed: row.failed, + }; + const at = row.finished_at?.getTime() ?? 0; + if (row.status === "failed") + await d.ctx.outbox.emit( + d.ctx.db, + "scrape.failed", + { ...ref, reason: row.reason }, + { + idempotencyKey: lifecycleKey("scrape.failed", row.id, at), + }, + ); + else + await d.ctx.outbox.emit(d.ctx.db, "scrape.completed", ref, { + idempotencyKey: lifecycleKey("scrape.completed", row.id, at), + }); +} diff --git a/apps/server/src/modules/acquisition-organize/services/organize-hooks.ts b/apps/server/src/modules/acquisition-organize/services/organize-hooks.ts new file mode 100644 index 00000000..7c3348d9 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/organize-hooks.ts @@ -0,0 +1,89 @@ +import { basename, dirname, extname, join } from "node:path"; +import type { OrganizePlan } from "@umt/domain"; +import { hasHook, interceptHook, rewriteHook } from "../../../hooks/index.ts"; +import type { ServerContext } from "../../types.ts"; + +// 整理计划上的链式钩子(ADR-006):organize.rename 改写主视频目标文件名、organize.intercept 拒绝整理。 +// 都在计划生成之后、写整理记录之前调用;没有注册钩子时什么都不做。 + +const BAD_NAME = /[\\/\0]/; +const usableName = (n: unknown): boolean => + typeof n === "string" && n.trim() !== "" && !BAD_NAME.test(n) && n !== "." && n !== ".."; + +/** + * organize.rename:逐个主视频把模板生成的目标文件名交给钩子改写,同主干的附属文件(字幕、NFO 等)跟着换主干。 + * 只改文件名不改目录;返回带路径分隔符或空名的钩子结果跳过。改名后撞名的项转 hold + duplicate_target(与保留原名同一处理)。 + * 改名后的目标是否已存在由执行阶段按冲突处理(计划期的存在性探测在改名之前)。 + */ +export async function applyRenameHook( + ctx: Pick, + plan: OrganizePlan, + base: { mediaEntityId: string; mediaType: "movie" | "tv" }, +): Promise { + if (!hasHook(ctx, "organize.rename")) return 0; + const renamed = new Map(); + for (const main of plan.files.filter((f) => f.role === "main")) { + const name = basename(main.dst); + // 返回带路径分隔符或空名的钩子结果跳过(记日志),其余钩子照常 + const next = await rewriteHook( + ctx, + "organize.rename", + name, + { + targetPath: main.dst, + sourcePath: main.src, + mediaEntityId: base.mediaEntityId, + mediaType: base.mediaType, + season: main.season ?? null, + episodes: main.episodes ?? null, + }, + usableName, + ); + if (next === name) continue; + const oldStem = name.slice(0, name.length - extname(name).length); + const newStem = next.slice(0, next.length - extname(next).length); + const to = join(dirname(main.dst), next); + renamed.set(main.dst, to); + main.dst = to; + for (const f of plan.files) { + if (f.mainSrc !== main.src) continue; + const side = basename(f.dst); + if (!side.startsWith(oldStem)) continue; + const nd = join(dirname(f.dst), newStem + side.slice(oldStem.length)); + renamed.set(f.dst, nd); + f.dst = nd; + } + } + if (renamed.size === 0) return 0; + for (const c of plan.conflicts) if (c.dst && renamed.has(c.dst)) c.dst = renamed.get(c.dst)!; + for (const pc of plan.preconditions) + if (pc.kind === "target_absent" || pc.kind === "verify_after_copy") + pc.paths = pc.paths.map((p) => renamed.get(p) ?? p); + const byDst = new Map(); + for (const f of plan.files) byDst.set(f.dst, [...(byDst.get(f.dst) ?? []), f]); + for (const [dst, fs] of byDst) { + if (fs.length < 2 || plan.conflicts.some((c) => c.kind === "duplicate_target" && c.dst === dst)) continue; + for (const f of fs) f.action = "hold"; + plan.conflicts.push({ kind: "duplicate_target", blocking: true, dst, detail: fs.map((f) => f.src).join(" | ") }); + } + plan.requiresApproval = plan.conflicts.some((c) => c.blocking); + return renamed.size; +} + +/** organize.intercept:返回拒绝原因(含拒绝的钩子名),放行返回 null */ +export async function organizeIntercept( + ctx: Pick, + plan: OrganizePlan, + base: { + mediaEntityId: string; + mediaType: "movie" | "tv"; + action: string; + trigger: string; + downloadTaskId: string | null; + }, +) { + return interceptHook(ctx, "organize.intercept", () => ({ + ...base, + files: plan.files.map((f) => ({ src: f.src, dst: f.dst, kind: f.kind, role: f.role })), + })); +} diff --git a/apps/server/src/modules/acquisition-organize/services/organize-plan.ts b/apps/server/src/modules/acquisition-organize/services/organize-plan.ts index b2d5871b..3710dbe2 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-plan.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-plan.ts @@ -42,10 +42,12 @@ import { publishedAcquisitionPolicies, unitKeysOf } from "./acquisition.ts"; import { classifyEntity } from "./classification.ts"; import { type AoDeps, type Exec, hostPathOf, loadInstances, upsertDecision } from "./common.ts"; import { applySlotEvent } from "./downloads.ts"; +import { emitOrganizationFinished } from "./lifecycle-events.ts"; import { effectiveNaming, ensureNamingVersion, loadNamingEntity, namingByVersion } from "./naming.ts"; import { resolveOrganizeSource } from "./organize-config.ts"; import { PathGuardError } from "./organize-guard.ts"; import { successfulOrganizeCounts } from "./organize-history.ts"; +import { applyRenameHook, organizeIntercept } from "./organize-hooks.ts"; import { planTransfer } from "./organize-transfer.ts"; // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→ // 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。 @@ -925,8 +927,19 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise { const r = await exec .selectFrom("organization_record") - .select(["status", "change_seq", "media_entity_id"]) + .select(["id", "status", "change_seq", "media_entity_id", "action", "trigger", "download_task_id", "reason"]) .where("id", "=", recordId) .executeTakeFirstOrThrow(); await d.ctx.outbox.emit(exec, "organization.changed", { @@ -1297,6 +1310,8 @@ export async function emitRecord(d: AoDeps, exec: Exec, recordId: string): Promi changeSeq: String(r.change_seq), mediaEntityId: r.media_entity_id, }); + // 终态另发生命周期广播(ADR-006):整理成功 / 部分成功 / 失败 + await emitOrganizationFinished(d, exec, r); } /** diff --git a/apps/server/src/modules/acquisition-organize/services/scrape.ts b/apps/server/src/modules/acquisition-organize/services/scrape.ts index 95d9ff26..69951c1f 100644 --- a/apps/server/src/modules/acquisition-organize/services/scrape.ts +++ b/apps/server/src/modules/acquisition-organize/services/scrape.ts @@ -30,6 +30,7 @@ import { } from "../../sources-identity/service/episode-resolution.ts"; import { errMessage, jsonb } from "../util.ts"; import type { AoDeps } from "./common.ts"; +import { emitScrapeFinished } from "./lifecycle-events.ts"; import { type EntityForNfo, episodeMetadataMissing, @@ -930,6 +931,7 @@ export async function runScrapeRecord( .execute() .catch(() => {}); if (dirs.size) void d.ctx.events.publish("files.changed", { rootId: rec.root_id, dirs: [...dirs] }); + await emitScrapeFinished(d, row); return scrapeRecordView(row); }; const fail = (reason: string) => finish({ status: "failed", reason, files: [] }); diff --git a/apps/server/src/modules/acquisition-organize/services/subscriptions.ts b/apps/server/src/modules/acquisition-organize/services/subscriptions.ts index d1857262..1c300bc8 100644 --- a/apps/server/src/modules/acquisition-organize/services/subscriptions.ts +++ b/apps/server/src/modules/acquisition-organize/services/subscriptions.ts @@ -1,4 +1,10 @@ -import type { AcquisitionSlot, SubscriptionParams, SubscriptionProgress, UnitProgressState } from "@umt/contracts"; +import { + type AcquisitionSlot, + lifecycleKey, + type SubscriptionParams, + type SubscriptionProgress, + type UnitProgressState, +} from "@umt/contracts"; import type { AcquisitionSlotRow, SourceReleaseRow, SubscriptionRow } from "@umt/db"; import { sql } from "@umt/db"; import { @@ -20,8 +26,10 @@ import { type UnitRange, } from "@umt/domain"; import { slotSpecFromRow } from "@umt/mapping"; +import { interceptHook } from "../../../hooks/index.ts"; import { errMessage, jsonb, toApi } from "../util.ts"; import { type AoDeps, type Exec, upsertDecision } from "./common.ts"; +import { emitSubscriptionEvent } from "./lifecycle-events.ts"; import { contextOf, parsedOf, unitKeysOf } from "./release-facts.ts"; import { entityOriginalLanguage, loadRuleCatalog, mergeHard, type RuleCatalog, resolveRuleChain } from "./rules.ts"; @@ -868,6 +876,18 @@ export async function completeSubscription( reason: string, progress: Record = {}, ): Promise { + // 链式钩子 subscription.completion.check(ADR-006):追加完成条件,拒绝时订阅继续追踪 + const veto = await interceptHook(d.ctx, "subscription.completion.check", () => ({ + subscriptionId: sub.id, + mediaEntityId: sub.media_entity_id, + mediaKind: sub.media_kind, + reason, + progress, + })); + if (veto) { + d.ctx.logger.info({ subscriptionId: sub.id, reason, veto }, "订阅完成被钩子拦下，继续追踪"); + return false; + } const done = await trx .updateTable("subscription") .set({ status: "completed", status_reason: reason, next_search_at: null }) @@ -877,7 +897,7 @@ export async function completeSubscription( .executeTakeFirst(); if (!done) return false; await clearSubscriptionDeferrals(trx, done.id); - await trx + const history = await trx .insertInto("subscription_history") .values({ subscription_id: done.id, @@ -890,7 +910,12 @@ export async function completeSubscription( progress: jsonb(progress), completed_at: d.ports.now(), }) - .execute(); + .returning("id") + .executeTakeFirstOrThrow(); + await emitSubscriptionEvent(d, trx, "subscription.completed", done, { + reason, + key: lifecycleKey("subscription.completed", done.id, history.id), + }); d.ctx.logger.info({ subscriptionId: sub.id, reason }, "订阅完成，已写历史"); return true; } diff --git a/apps/server/src/modules/sources-identity/service/account.ts b/apps/server/src/modules/sources-identity/service/account.ts index cb3327c5..2292ce3d 100644 --- a/apps/server/src/modules/sources-identity/service/account.ts +++ b/apps/server/src/modules/sources-identity/service/account.ts @@ -30,6 +30,7 @@ import type { SourcesRuntime } from "../runtime.ts"; import { capabilitiesOf, LOGIN_KINDS } from "../site-kinds.ts"; import { type Actor, audit, closeDecisions, type Exec, openDecision, SYSTEM } from "./common.ts"; import { cookieFromBrowser, writeBrowserCookie } from "./credentials.ts"; +import { emitSiteReachability } from "./lifecycle-events.ts"; import { browserOnlyCookiesOf, resolveUserAgent, type UserAgentDefaults } from "./user-agent.ts"; // W9-A 站点中心(docs/design/site-center.md):凭证写入、刷新登录(账密 takelogin / Cookie / API Key 校验)、 @@ -865,12 +866,20 @@ export async function checkSiteHealth( ? (HEALTH_OF[status] ?? "degraded") : site.health_status; const last = { ok, status, latencyMs, message, detail }; + // 生命周期广播(ADR-006):登录失效 / 访问不了 ⇄ 正常,只在跨越时发 + const bad = (h: string, st: unknown) => h === "failing" || st === "unreachable"; + const wasBad = bad(site.health_status, (site.last_health as { status?: unknown } | null)?.status); + const isBad = site.enabled && bad(healthStatus, status); await rt.ctx.db.transaction().execute(async (trx) => { await trx .updateTable("site") .set({ last_health: jsonb(last), last_health_at: now, health_status: healthStatus }) .where("id", "=", site.id) .execute(); + if (isBad && !wasBad) + await emitSiteReachability(rt.ctx, trx, site.id, { reachable: false, reason: status, message }); + else if (wasBad && healthStatus === "healthy") + await emitSiteReachability(rt.ctx, trx, site.id, { reachable: true, reason: "health_check" }); // 首页显示已登录:顺手记上次登录 if (ok) { await trx diff --git a/apps/server/src/modules/sources-identity/service/episode-groups.ts b/apps/server/src/modules/sources-identity/service/episode-groups.ts index 4d642228..18b34644 100644 --- a/apps/server/src/modules/sources-identity/service/episode-groups.ts +++ b/apps/server/src/modules/sources-identity/service/episode-groups.ts @@ -8,12 +8,13 @@ import { TmdbError, type TmdbRichDetail, } from "@umt/adapters"; -import type { - EpisodeGroupList, - EpisodeGroupSetResult, - EpisodeStructureCurrent, - EpisodeStructureSeason, - EpisodeStructureSeasons, +import { + type EpisodeGroupList, + type EpisodeGroupSetResult, + type EpisodeStructureCurrent, + type EpisodeStructureSeason, + type EpisodeStructureSeasons, + lifecycleKey, } from "@umt/contracts"; import { sql } from "@umt/db"; import { ApiException } from "../../../http/errors.ts"; @@ -357,6 +358,13 @@ export async function setEpisodeGroup( }, ); const reresolve = await reresolveEntityEpisodes(rt.ctx, trx, tv.id); + // 生命周期广播(ADR-006):季集结构已切换 + await rt.ctx.outbox.emit( + trx, + "media.episode_structure_changed", + { mediaEntityId: tv.id, fromGroupId: fromId, toGroupId: groupId }, + { idempotencyKey: lifecycleKey("media.episode_structure_changed", tv.id, now.toISOString()) }, + ); return { ...r, reresolve }; }); let refresh = "refreshed"; diff --git a/apps/server/src/modules/sources-identity/service/lifecycle-events.ts b/apps/server/src/modules/sources-identity/service/lifecycle-events.ts new file mode 100644 index 00000000..5964d907 --- /dev/null +++ b/apps/server/src/modules/sources-identity/service/lifecycle-events.ts @@ -0,0 +1,55 @@ +import { lifecycleKey } from "@umt/contracts"; +import type { DB, Kysely, Transaction } from "@umt/db"; +import type { ServerContext } from "../../types.ts"; + +// 生命周期广播事件(ADR-006)在来源与识别侧的发出点。元数据刷新、进入关注的判定本身在 metadata.ts / watched.ts, +// 这里只在调用方发事件,不改那两处。 + +type Exec = Kysely | Transaction; + +/** + * 作品进入关注:建订阅、开始下载、用户搜索 / 手动认定时由调用方发。 + * 同一作品同一触发对象(订阅、下载计划、资源)只发一次;订阅方自己判断作品是否已经处理过。 + */ +export async function emitMediaWatched( + ctx: Pick, + exec: Exec, + mediaEntityId: string, + reason: string, + subjectId: string, +): Promise { + await ctx.outbox.emit( + exec, + "media.watched", + { mediaEntityId, reason }, + { idempotencyKey: lifecycleKey("media.watched", mediaEntityId, reason, subjectId) }, + ); +} + +/** 作品元数据刷新完成(刷新任务的调用方发;按任务 run 去重) */ +export async function emitMetadataRefreshed( + ctx: Pick, + exec: Exec, + input: { mediaEntityId: string; full: boolean; changedFields: string[]; runId: string }, +): Promise { + await ctx.outbox.emit( + exec, + "media.metadata_refreshed", + { mediaEntityId: input.mediaEntityId, full: input.full, changedFields: input.changedFields }, + { idempotencyKey: lifecycleKey("media.metadata_refreshed", input.mediaEntityId, input.runId) }, + ); +} + +/** 站点不可达 / 恢复。只在状态跨越时由调用方发(连续失败到阈值、登录失效、检测结果变化) */ +export async function emitSiteReachability( + ctx: Pick, + exec: Exec, + siteId: string, + change: { reachable: false; reason: string; message: string | null } | { reachable: true; reason: string }, + key?: string, +): Promise { + const opts = key ? { idempotencyKey: key } : {}; + if (change.reachable) await ctx.outbox.emit(exec, "site.recovered", { siteId, reason: change.reason }, opts); + else + await ctx.outbox.emit(exec, "site.unreachable", { siteId, reason: change.reason, message: change.message }, opts); +} diff --git a/apps/server/src/modules/sources-identity/service/resolve.ts b/apps/server/src/modules/sources-identity/service/resolve.ts index 280a60cf..ea776c5b 100644 --- a/apps/server/src/modules/sources-identity/service/resolve.ts +++ b/apps/server/src/modules/sources-identity/service/resolve.ts @@ -21,6 +21,7 @@ import { toParsedColumns, toReleaseResolutionStatus, } from "@umt/mapping"; +import { rewriteHook } from "../../../hooks/index.ts"; import { jsonb } from "../rows.ts"; import type { SourcesRuntime } from "../runtime.ts"; import { addSiteAliases, type SiteAliasOutcome } from "./alias-guard.ts"; @@ -52,6 +53,7 @@ import { openGroupDecision, refreshGroupDecision, } from "./identity-decisions.ts"; +import { emitMediaWatched } from "./lifecycle-events.ts"; import { loadSubject, searchBangumi } from "./meta-subjects.ts"; import { tmdbRecallInput, tmdbToCandidate } from "./tmdb-candidate.ts"; import { loadRecognizeCache, type RecognizeMeta, recognizeMetaOf, saveRecognizeCache } from "./tmdb-recognize-cache.ts"; @@ -480,7 +482,18 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P mediaEntityId: release.media_entity_id, }; } - const parsed = parseRelease(release.title, release.subtitle ? { subtitle: release.subtitle } : {}); + // 链式钩子 name.recognize(ADR-006):可改写解析结果;没有钩子时原样 + const parsed = await rewriteHook( + ctx, + "name.recognize", + parseRelease(release.title, release.subtitle ? { subtitle: release.subtitle } : {}), + () => ({ + title: release.title, + subtitle: release.subtitle ?? null, + sourceReleaseId: release.id, + trigger: input.trigger ?? (input.force ? "review" : "observed"), + }), + ); const cols = toParsedColumns(parsed); // 站点给出的外部 ID(列表 / 详情页的 IMDb、豆瓣、TMDB 链接)作为显式 ID 证据参与打分 const siteIds = siteExplicitIds(release.raw); @@ -909,6 +922,9 @@ async function bindAuto( } else if (userAsked && result.mediaEntityId) { await enqueueWatchedRefresh(ctx, result.mediaEntityId, `lookup:${lookup}`); } + // 用户搜索 / 手动认定 = 作品进入关注(ADR-006 生命周期广播) + if (userAsked && result.mediaEntityId) + await emitMediaWatched(ctx, ctx.db, result.mediaEntityId, `lookup:${lookup}`, release.id); return result; } catch (e) { if (e instanceof IdentityConflictError && attempt < 2) continue; @@ -1160,13 +1176,20 @@ export interface NameIdentifyResult { */ export async function identifyParsed( rt: SourcesRuntime, - parsed: ParsedRelease, + input: ParsedRelease, opts: { trigger?: ResolveTrigger; evidence?: Record } = {}, ): Promise { const { ctx } = rt; const none = { mediaEntityId: null, contentUnitIds: [] as string[], candidates: [], created: false }; - if (!parsed.title || parsed.auxiliaryOnly) return { outcome: "unparseable", reason: "unparseable_title", ...none }; const trigger = opts.trigger ?? "takeover"; + // 链式钩子 name.recognize(ADR-006):整理 / 接管按名称识别时同样可改写解析结果 + const parsed = await rewriteHook(ctx, "name.recognize", input, () => ({ + title: input.raw, + subtitle: null, + sourceReleaseId: null, + trigger, + })); + if (!parsed.title || parsed.auxiliaryOnly) return { outcome: "unparseable", reason: "unparseable_title", ...none }; const local = await localCandidates(ctx.db, parsed, null); const lookup = await lookupReason(rt, parsed, trigger, local.length > 0, null); let remote: MetadataCandidate[] = []; diff --git a/apps/server/src/modules/sources-identity/service/snapshot.ts b/apps/server/src/modules/sources-identity/service/snapshot.ts index d0dbee3d..e315e00a 100644 --- a/apps/server/src/modules/sources-identity/service/snapshot.ts +++ b/apps/server/src/modules/sources-identity/service/snapshot.ts @@ -1,6 +1,6 @@ import { createHash } from "node:crypto"; import { type SiteDetailPage, SiteError, type SitePage, type SiteRelease } from "@umt/adapters"; -import type { SiteSnapshotKind } from "@umt/contracts"; +import { lifecycleKey, type SiteSnapshotKind } from "@umt/contracts"; import { type SiteRow, sql } from "@umt/db"; import { parseRelease } from "@umt/domain"; import { siteReleaseToSourceRelease } from "@umt/mapping"; @@ -10,6 +10,7 @@ import { siteCan } from "../site-kinds.ts"; import { enqueueRelogin } from "./account.ts"; import { closeDecisions, type Exec, openDecision } from "./common.ts"; import { hasDetailPage, saveDetailPage } from "./detail-page.ts"; +import { emitSiteReachability } from "./lifecycle-events.ts"; // 站点快照(周期列表)、补漏(Feed 疑似缺口,有页数预算)、实时补查(工作台搜索)。 // 所有请求经站点共享限流器;结果按 site+站点资源 ID upsert 到 source_release,并记录 site_snapshot 覆盖范围。 @@ -114,6 +115,12 @@ async function onAuthError(rt: SourcesRuntime, trx: Exec, site: SiteRow, e: Site .where("status", "in", ["unknown", "active"]) .execute(); await trx.updateTable("site").set({ health_status: "failing" }).where("id", "=", site.id).execute(); + if (site.health_status !== "failing") + await emitSiteReachability(rt.ctx, trx, site.id, { + reachable: false, + reason: e.kind, + message: e.message.slice(0, 300), + }); await openDecision(rt.ctx, trx, { kind: "source_unreachable", severity: "warning", @@ -247,6 +254,23 @@ export async function snapshotSite(rt: SourcesRuntime, input: SnapshotInput): Pr const prevStreak = Number((site.snapshot_cursor as { failStreak?: unknown } | null)?.failStreak) || 0; const failStreak = status === "failed" ? prevStreak + 1 : 0; await syncSnapshotFailureDecision(rt, trx, site, failStreak, errJson); + // 生命周期广播(ADR-006):连续失败到开待处理的阈值 = 不可达;从阈值以上恢复 = 恢复 + if (failStreak === SNAPSHOT_FAIL_DECISION_AT) + await emitSiteReachability( + rt.ctx, + trx, + site.id, + { reachable: false, reason: "snapshot_failed", message: errJson?.message.slice(0, 300) ?? null }, + lifecycleKey("site.unreachable", site.id, snap.id), + ); + else if (failStreak === 0 && prevStreak >= SNAPSHOT_FAIL_DECISION_AT) + await emitSiteReachability( + rt.ctx, + trx, + site.id, + { reachable: true, reason: "snapshot_ok" }, + lifecycleKey("site.recovered", site.id, snap.id), + ); await trx .updateTable("site") .set({ diff --git a/apps/server/src/modules/sources-identity/tasks.ts b/apps/server/src/modules/sources-identity/tasks.ts index f6287bc6..7c943a7d 100644 --- a/apps/server/src/modules/sources-identity/tasks.ts +++ b/apps/server/src/modules/sources-identity/tasks.ts @@ -5,6 +5,7 @@ import type { SourcesRuntime } from "./runtime.ts"; import { refreshSiteAccount } from "./service/account.ts"; import { AUTO_ENRICH_TASK, runAutoEnrich } from "./service/enrich.ts"; import { ingestFeed } from "./service/feeds.ts"; +import { emitMetadataRefreshed } from "./service/lifecycle-events.ts"; import { refreshMetadata } from "./service/metadata.ts"; import { resolveRelease } from "./service/resolve.ts"; import { snapshotSite } from "./service/snapshot.ts"; @@ -75,7 +76,18 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu inputSchema: z.object({ mediaEntityId: id, full: z.boolean().optional() }), retryPolicy: NETWORK_RETRY, handler: (t) => - t.step.run("refresh", () => refreshMetadata(rt(), t.input.mediaEntityId, { full: !!t.input.full })), + t.step.run("refresh", async () => { + const r = await refreshMetadata(rt(), t.input.mediaEntityId, { full: !!t.input.full }); + // 生命周期广播(ADR-006):元数据已刷新(在调用方发,刷新本身在 metadata.ts) + if (r.status === "refreshed") + await emitMetadataRefreshed(rt().ctx, rt().ctx.db, { + mediaEntityId: t.input.mediaEntityId, + full: !!t.input.full, + changedFields: r.changedFields, + runId: t.runId, + }); + return r; + }), }), defineTask({ name: "searchSubscription", diff --git a/apps/server/test/hooks/resource-selection.test.ts b/apps/server/test/hooks/resource-selection.test.ts new file mode 100644 index 00000000..f1429e52 --- /dev/null +++ b/apps/server/test/hooks/resource-selection.test.ts @@ -0,0 +1,45 @@ +import type { SourceReleaseRow } from "@umt/db"; +import { describe, expect, it } from "vitest"; +import { HookRegistry } from "../../src/hooks/index.ts"; +import { selectResources } from "../../src/modules/acquisition-organize/services/acquisition.ts"; + +// 链式钩子 resource.selection(ADR-006,对照 MP ChainEventType.ResourceSelection):订阅重评时调整候选顺序 / 过滤, +// 只能重排与删减,不能加入新候选。 + +const rel = (id: string, title: string) => ({ id, title, parsed: null }) as unknown as SourceReleaseRow; +const rels = [ + rel("r1", "Show.S01E01.1080p.WEB-DL"), + rel("r2", "Show.S01E01.2160p.WEB-DL"), + rel("r3", "Show.S01E01.720p"), +]; +const context = { mediaEntityId: "e", subscriptionId: "s", slotId: null }; +const ctxOf = (hooks?: HookRegistry) => ({ d: { ctx: { hooks } } }) as never; + +describe("resource.selection 钩子", () => { + it("没有钩子时原样返回同一个数组", async () => { + expect(await selectResources(ctxOf(), rels, context)).toBe(rels); + expect(await selectResources(ctxOf(new HookRegistry()), rels, context)).toBe(rels); + }); + + it("重排与过滤生效;不认识的 ID 忽略、重复的只取一次", async () => { + const hooks = new HookRegistry(); + const seen: unknown[] = []; + hooks.forModule("test").rewrite("resource.selection", (cands, c) => { + seen.push({ ids: cands.map((x) => x.sourceReleaseId), c }); + const by = new Map(cands.map((x) => [x.sourceReleaseId, x])); + // 2160p 优先、去掉 720p,夹带一个不存在的候选和一个重复 + return [by.get("r2")!, { ...by.get("r1")!, sourceReleaseId: "intruder" }, by.get("r1")!, by.get("r2")!]; + }); + const out = await selectResources(ctxOf(hooks), rels, context); + expect(out.map((r) => r.id)).toEqual(["r2", "r1"]); + expect(seen).toEqual([{ ids: ["r1", "r2", "r3"], c: context }]); + }); + + it("钩子出错时沿用原顺序", async () => { + const hooks = new HookRegistry(); + hooks.forModule("test").rewrite("resource.selection", () => { + throw new Error("坏了"); + }); + expect((await selectResources(ctxOf(hooks), rels, context)).map((r) => r.id)).toEqual(["r1", "r2", "r3"]); + }); +}); diff --git a/apps/server/test/lifecycle/lifecycle-events.test.ts b/apps/server/test/lifecycle/lifecycle-events.test.ts new file mode 100644 index 00000000..4d81ca09 --- /dev/null +++ b/apps/server/test/lifecycle/lifecycle-events.test.ts @@ -0,0 +1,314 @@ +import { type EventType, isEventType, LIFECYCLE_EVENTS, parseEventPayload } from "@umt/contracts"; +import { OutboxRelay } from "@umt/tasks"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { collectOutboxRoutes } from "../../src/context.ts"; +import { HookRegistry } from "../../src/hooks/index.ts"; +import { + emitDownloadLifecycle, + emitScrapeFinished, +} from "../../src/modules/acquisition-organize/services/lifecycle-events.ts"; +import { emitRecord } from "../../src/modules/acquisition-organize/services/organize.ts"; +import { completeSubscription } from "../../src/modules/acquisition-organize/services/subscriptions.ts"; +import { SERVER_MODULES } from "../../src/modules/index.ts"; +import { + emitMetadataRefreshed, + emitSiteReachability, +} from "../../src/modules/sources-identity/service/lifecycle-events.ts"; +import type { ServerContext, ServerModule } from "../../src/modules/types.ts"; +import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "../acquisition-organize/harness.ts"; + +// 生命周期广播事件(ADR-006):各业务节点在事务内写 outbox,payload 过 contracts schema,幂等键防重复。 + +const hasPg = await pgAvailable(); + +describe.skipIf(!hasPg)("生命周期广播事件(真实 PG)", () => { + let h: Harness; + let base: Awaited>; + + /** 某类事件的全部行;顺带确认 payload 过 schema */ + const events = async (type: EventType) => { + const rows = await h.outbox(type); + for (const r of rows) expect(parseEventPayload(type, r.payload), `${type} payload`).not.toBeNull(); + return rows; + }; + + beforeAll(async () => { + h = await createHarness("lifecycle"); + base = await seedBase(h.db, { + libraryPath: "/tmp/umt-lc-lib", + cachePath: "/tmp/umt-lc-dl", + qbUrl: "http://127.0.0.1:9", + }); + }, 60_000); + afterAll(async () => { + await h?.close(); + }); + + it("清单里的事件都是已登记类型", () => { + for (const e of LIFECYCLE_EVENTS) expect(isEventType(e.type)).toBe(true); + }); + + it("订阅:已建(并发作品进入关注)/ 已改 / 暂停也算已改 / 删除发已取消", async () => { + const e = await seedEntity(h.db, { kind: "movie", title: "订阅事件片", year: 2024 }); + const created = await h.api("POST", "/subscriptions", { mediaEntityId: e, mediaKind: "movie" }); + expect(created.status).toBe(201); + const id = created.body.data.id as string; + expect((await events("subscription.created")).filter((r) => r.payload.subscriptionId === id)).toEqual([ + expect.objectContaining({ + payload: { subscriptionId: id, mediaEntityId: e, mediaKind: "movie", origin: "user", status: "active" }, + idempotency_key: `subscription.created/${id}`, + }), + ]); + expect((await events("media.watched")).find((r) => r.payload.mediaEntityId === e)).toMatchObject({ + payload: { mediaEntityId: e, reason: "subscription" }, + idempotency_key: `media.watched/${e}/subscription/${id}`, + }); + + expect((await h.api("PATCH", `/subscriptions/${id}`, { priority: "new_work" })).status).toBe(200); + expect((await h.api("POST", `/subscriptions/${id}/pause`)).status).toBe(200); + const updated = (await events("subscription.updated")).filter((r) => r.payload.subscriptionId === id); + expect(updated.map((r) => [r.payload.changes, r.payload.status])).toEqual([ + [["priority"], "active"], + [["status"], "paused"], + ]); + + expect((await h.api("POST", "/subscriptions/batch", { ids: [id], action: "delete" })).status).toBe(200); + expect((await events("subscription.cancelled")).filter((r) => r.payload.subscriptionId === id)).toEqual([ + expect.objectContaining({ payload: expect.objectContaining({ reason: "user_deleted" }) }), + ]); + }); + + it("订阅完成:发已完成(按历史行去重);钩子 subscription.completion.check 拒绝时不完成、不发事件", async () => { + const e = await seedEntity(h.db, { kind: "movie", title: "完成判定片", year: 2023 }); + const sub = await h.db + .insertInto("subscription") + .values({ media_entity_id: e, media_kind: "movie", status: "active" }) + .returningAll() + .executeTakeFirstOrThrow(); + + const hooks = new HookRegistry(); + const seen: unknown[] = []; + hooks.forModule("test").intercept("subscription.completion.check", (c) => { + seen.push(c); + return "还要等导演剪辑版"; + }); + h.ctx.hooks = hooks; + try { + const done = await h.db.transaction().execute((trx) => completeSubscription(h.d, trx, sub, "slot_satisfied")); + expect(done).toBe(false); + } finally { + h.ctx.hooks = undefined; + } + expect(seen).toEqual([ + { subscriptionId: sub.id, mediaEntityId: e, mediaKind: "movie", reason: "slot_satisfied", progress: {} }, + ]); + const still = await h.db.selectFrom("subscription").select("status").where("id", "=", sub.id).executeTakeFirst(); + expect(still?.status).toBe("active"); + expect((await events("subscription.completed")).filter((r) => r.payload.subscriptionId === sub.id)).toEqual([]); + + expect(await h.db.transaction().execute((trx) => completeSubscription(h.d, trx, sub, "slot_satisfied"))).toBe(true); + // 已完成的再判一次不会重复发 + expect(await h.db.transaction().execute((trx) => completeSubscription(h.d, trx, sub, "slot_satisfied"))).toBe( + false, + ); + const completed = (await events("subscription.completed")).filter((r) => r.payload.subscriptionId === sub.id); + expect(completed).toHaveLength(1); + expect(completed[0]!.payload.reason).toBe("slot_satisfied"); + expect(completed[0]!.idempotency_key).toMatch(new RegExp(`^subscription\\.completed/${sub.id}/`)); + }); + + it("整理结束:终态才发 organization.finished,同一次状态变化只发一次;进行中不发", async () => { + const e = await seedEntity(h.db, { kind: "movie", title: "整理事件片", year: 2022 }); + const rec = await h.db + .insertInto("organization_record") + .values({ trigger: "auto", actor_type: "system", action: "copy", status: "running", media_entity_id: e }) + .returning("id") + .executeTakeFirstOrThrow(); + await emitRecord(h.d, h.db, rec.id); + expect((await events("organization.finished")).filter((r) => r.payload.recordId === rec.id)).toEqual([]); + + await h.db.updateTable("organization_record").set({ status: "succeeded" }).where("id", "=", rec.id).execute(); + await emitRecord(h.d, h.db, rec.id); + await emitRecord(h.d, h.db, rec.id); // 重放:同一 change_seq + const fin = (await events("organization.finished")).filter((r) => r.payload.recordId === rec.id); + expect(fin).toHaveLength(1); + expect(fin[0]!.payload).toMatchObject({ + outcome: "succeeded", + action: "copy", + trigger: "auto", + mediaEntityId: e, + downloadTaskId: null, + }); + expect(fin[0]!.idempotency_key).toBe(`organization.finished/${rec.id}/${fin[0]!.payload.changeSeq}`); + // organization.changed 照旧每次都发 + expect((await events("organization.changed")).filter((r) => r.payload.recordId === rec.id)).toHaveLength(3); + + // 重新整理后失败:新的状态变化再发一条 + await h.db + .updateTable("organization_record") + .set({ status: "failed", reason: "target_exists" }) + .where("id", "=", rec.id) + .execute(); + await emitRecord(h.d, h.db, rec.id); + const again = (await events("organization.finished")).filter((r) => r.payload.recordId === rec.id); + expect(again.map((r) => [r.payload.outcome, r.payload.reason])).toEqual([ + ["succeeded", null], + ["failed", "target_exists"], + ]); + }); + + it("下载:完成 / 失败 / 移除各发一次(按任务行代数去重),其它状态不发", async () => { + const dt = await h.db + .insertInto("download_task") + .values({ instance_id: base.instanceId, torrent_hash: "a".repeat(40), name: "Some.Movie.2024.mkv" }) + .returning("id") + .executeTakeFirstOrThrow(); + await emitDownloadLifecycle(h.d, h.db, dt.id, "queued", "downloading"); + await emitDownloadLifecycle(h.d, h.db, dt.id, "downloading", "completed"); + await emitDownloadLifecycle(h.d, h.db, dt.id, "downloading", "completed"); + await emitDownloadLifecycle(h.d, h.db, dt.id, "completed", "removed"); + const done = (await events("download.completed")).filter((r) => r.payload.downloadTaskId === dt.id); + expect(done).toEqual([ + expect.objectContaining({ + payload: { + downloadTaskId: dt.id, + planId: null, + mediaEntityId: null, + name: "Some.Movie.2024.mkv", + generation: 0, + }, + idempotency_key: `download.completed/${dt.id}/g0`, + }), + ]); + expect((await events("download.removed")).filter((r) => r.payload.downloadTaskId === dt.id)).toEqual([ + expect.objectContaining({ payload: expect.objectContaining({ from: "completed" }) }), + ]); + // 复活后(代数 +1)再完成会再发 + await h.db.updateTable("download_task").set({ generation: 1 }).where("id", "=", dt.id).execute(); + await emitDownloadLifecycle(h.d, h.db, dt.id, "downloading", "completed"); + await emitDownloadLifecycle(h.d, h.db, dt.id, "downloading", "failed"); + expect((await events("download.completed")).filter((r) => r.payload.downloadTaskId === dt.id)).toHaveLength(2); + expect((await events("download.failed")).filter((r) => r.payload.downloadTaskId === dt.id)).toHaveLength(1); + }); + + it("刮削:失败发 scrape.failed,成功发 scrape.completed", async () => { + const lib = await h.db.selectFrom("library_root").select("id").executeTakeFirstOrThrow(); + const mk = async (status: "failed" | "succeeded") => + h.db + .insertInto("scrape_record") + .values({ + root_id: lib.id, + root_path: "/tmp/umt-lc-lib", + status, + reason: status === "failed" ? "scrape_no_metadata" : null, + written: status === "failed" ? 0 : 3, + finished_at: new Date(), + }) + .returningAll() + .executeTakeFirstOrThrow(); + const bad = await mk("failed"); + const ok = await mk("succeeded"); + await emitScrapeFinished(h.d, bad); + await emitScrapeFinished(h.d, bad); + await emitScrapeFinished(h.d, ok); + expect((await events("scrape.failed")).map((r) => r.payload)).toEqual([ + expect.objectContaining({ scrapeRecordId: bad.id, reason: "scrape_no_metadata", status: "failed" }), + ]); + expect((await events("scrape.completed")).map((r) => r.payload)).toEqual([ + expect.objectContaining({ scrapeRecordId: ok.id, written: 3, status: "succeeded" }), + ]); + }); + + it("站点不可达 / 恢复、元数据已刷新:payload 合法,带键的按键去重", async () => { + await emitSiteReachability(h.ctx, h.db, base.siteId, { + reachable: false, + reason: "auth_expired", + message: "登录失效", + }); + await emitSiteReachability( + h.ctx, + h.db, + base.siteId, + { reachable: true, reason: "health_check" }, + "site.recovered/x", + ); + await emitSiteReachability( + h.ctx, + h.db, + base.siteId, + { reachable: true, reason: "health_check" }, + "site.recovered/x", + ); + expect((await events("site.unreachable")).map((r) => r.payload)).toEqual([ + { siteId: base.siteId, reason: "auth_expired", message: "登录失效" }, + ]); + expect(await events("site.recovered")).toHaveLength(1); + + const e = await seedEntity(h.db, { kind: "series", title: "刷新剧", year: 2021 }); + for (let i = 0; i < 2; i++) + await emitMetadataRefreshed(h.ctx, h.db, { + mediaEntityId: e, + full: true, + changedFields: ["overview"], + runId: "run-1", + }); + expect((await events("media.metadata_refreshed")).map((r) => r.payload)).toEqual([ + { mediaEntityId: e, full: true, changedFields: ["overview"] }, + ]); + }); + + it("补字幕订阅整理结束:中继投递时补字幕与其它订阅方都收到,入库类整理被认作补字幕候选", async () => { + await h.db.deleteFrom("outbox_event").execute(); + const e = await seedEntity(h.db, { kind: "movie", title: "字幕订阅片", year: 2020 }); + const recs = []; + for (const action of ["copy", "trash"] as const) { + const r = await h.db + .insertInto("organization_record") + .values({ trigger: "auto", actor_type: "system", action, status: "succeeded", media_entity_id: e }) + .returning("id") + .executeTakeFirstOrThrow(); + await emitRecord(h.d, h.db, r.id); + recs.push(r.id); + } + const debug = vi.fn(); + const ctx = { ...h.ctx, logger: { debug, warn: vi.fn() } } as unknown as ServerContext; + const other: ServerModule = { + name: "test-other", + registerRoutes() {}, + registerTasks() {}, + outboxRoutes: () => ({ + "organization.finished": (ev) => ({ + task: "otherTask", + input: { id: (ev.payload as { recordId: string }).recordId }, + }), + }), + }; + const enqueued: { task: string; key: string }[] = []; + const relay = new OutboxRelay({ + db: h.db, + engine: { + enqueue: async (task, _input, o) => { + enqueued.push({ task, key: o.idempotencyKey }); + return { runId: o.idempotencyKey, deduplicated: false }; + }, + }, + routes: collectOutboxRoutes([...SERVER_MODULES, other], ctx), + }); + await relay.pollOnce(); + // 补字幕只认复制 / 移动 / 原地入库(进回收站不算),自动补字幕还没做,不建任务 + expect(debug.mock.calls.map((c) => c[0])).toEqual([{ recordId: recs[0], mediaEntityId: e }]); + expect(enqueued.map((x) => x.task)).toEqual(["otherTask", "otherTask"]); + const pending = await h.db + .selectFrom("outbox_event") + .select(["type", "status"]) + .where("type", "=", "organization.finished") + .execute(); + expect(pending.map((p) => p.status)).toEqual(["dispatched", "dispatched"]); + }); + + it("payload 不合法直接抛错,不写 outbox", async () => { + await expect(h.ctx.outbox.emit(h.db, "organization.finished", { recordId: "x" } as never)).rejects.toThrow( + /不符合 contracts/, + ); + }); +}); diff --git a/apps/server/test/lifecycle/name-recognize-hook.test.ts b/apps/server/test/lifecycle/name-recognize-hook.test.ts new file mode 100644 index 00000000..a85c8f16 --- /dev/null +++ b/apps/server/test/lifecycle/name-recognize-hook.test.ts @@ -0,0 +1,68 @@ +import { parseRelease } from "@umt/domain"; +import { createMockTmdb, type MockTmdb, TMDB_FIXTURES } from "@umt/testkit"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { HookRegistry } from "../../src/hooks/index.ts"; +import { resolveRelease } from "../../src/modules/sources-identity/service/resolve.ts"; +import { createHarness, type Harness, pgAvailable, seedSite } from "../sources-identity/harness.ts"; + +// 链式钩子 name.recognize(ADR-006,对照 MP ChainEventType.NameRecognize):资源标题解析之后、召回之前改写解析结果。 + +const up = await pgAvailable(); +const GARBLED = "Zhuo.Wu.Ming.2023.2160p.UHD.BluRay.REMUX.HDR.HEVC-FGT"; + +describe.skipIf(!up)("name.recognize 钩子改写解析结果", () => { + let h: Harness; + let tmdb: MockTmdb; + let siteId: string; + let n = 0; + const release = async (title: string) => + h.db + .insertInto("source_release") + .values({ site_id: siteId, site_release_id: String(910000 + ++n), title }) + .returningAll() + .executeTakeFirstOrThrow(); + + beforeAll(async () => { + tmdb = await createMockTmdb({ fixtures: TMDB_FIXTURES }); + h = await createHarness("name_hook", { tmdbUrl: tmdb.url }); + siteId = (await seedSite(h, { baseUrl: "https://hook.example/" })).site.id; + await h.db + .insertInto("settings") + .values({ key: "identity", value: JSON.stringify({ eagerTmdbLookup: true }) } as never) + .execute(); + }); + afterEach(() => { + h.ctx.hooks = undefined; + }); + afterAll(async () => { + await h?.close(); + await tmdb?.close(); + }); + + it("没有钩子:认不出的标题照常不自动识别", async () => { + const r = await release(GARBLED); + const res = await resolveRelease(h.rt, { sourceReleaseId: r.id }); + expect(res.outcome).not.toBe("auto"); + }); + + it("钩子按站点别名改写解析结果后自动识别;上下文带原标题与来源资源", async () => { + const seen: unknown[] = []; + const hooks = new HookRegistry(); + hooks.forModule("test").rewrite("name.recognize", (p, c) => { + seen.push({ title: c.title, sourceReleaseId: c.sourceReleaseId, trigger: c.trigger }); + return c.title.startsWith("Zhuo.Wu.Ming") ? parseRelease(c.title.replace("Zhuo.Wu.Ming", "Oppenheimer")) : p; + }); + h.ctx.hooks = hooks; + const r = await release(GARBLED); + const res = await resolveRelease(h.rt, { sourceReleaseId: r.id }); + expect(res.outcome).toBe("auto"); + const e = await h.db + .selectFrom("external_identity") + .select("external_id") + .where("media_entity_id", "=", res.mediaEntityId!) + .where("provider", "=", "tmdb") + .executeTakeFirstOrThrow(); + expect(e.external_id).toBe("872585"); + expect(seen).toEqual([{ title: GARBLED, sourceReleaseId: r.id, trigger: "observed" }]); + }); +}); diff --git a/apps/server/test/lifecycle/organize-hooks.test.ts b/apps/server/test/lifecycle/organize-hooks.test.ts new file mode 100644 index 00000000..b2008dac --- /dev/null +++ b/apps/server/test/lifecycle/organize-hooks.test.ts @@ -0,0 +1,112 @@ +import { existsSync } from "node:fs"; +import { createTempTree, type TempTree } from "@umt/testkit"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { HookRegistry } from "../../src/hooks/index.ts"; +import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts"; +import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "../acquisition-organize/harness.ts"; + +// 整理上的链式钩子(ADR-006):organize.rename 改写目标文件名(字幕跟着改)、organize.intercept 拒绝整理; +// 钩子出错不影响整理。整理结束发 organization.finished。 + +const hasPg = await pgAvailable(); +const SIZE = 200_000; +const DIR = "Inception.2010.1080p.BluRay.x264-GRP"; +const VIDEO = `${DIR}/Inception.2010.1080p.BluRay.x264-GRP.mkv`; +const SUB = `${DIR}/Inception.2010.1080p.BluRay.x264-GRP.chs.srt`; + +describe.skipIf(!hasPg)("整理钩子:改名与拦截(真实 PG + 临时目录)", () => { + let h: Harness; + let lib: TempTree; + let src: TempTree; + let entity = ""; + const req = () => ({ + sourcePaths: [src.path(DIR)], + mediaEntityId: entity, + action: "copy" as const, + trigger: "manual" as const, + actorType: "user" as const, + }); + + beforeAll(async () => { + lib = await createTempTree({}, { sentinel: true, prefix: "umt-hklib-" }); + src = await createTempTree({ [VIDEO]: SIZE, [SUB]: 2_000 }, { prefix: "umt-hksrc-" }); + h = await createHarness("organize_hooks"); + await seedBase(h.db, { libraryPath: lib.root, cachePath: src.root, qbUrl: "http://127.0.0.1:9" }); + await h.setSettings("organize", { scrapeAfterImport: false }); + entity = await seedEntity(h.db, { kind: "movie", title: "盗梦空间", year: 2010 }); + await h.db.updateTable("media_entity").set({ original_title: "Inception" }).where("id", "=", entity).execute(); + }, 60_000); + afterEach(() => { + h.ctx.hooks = undefined; + }); + afterAll(async () => { + await h?.close(); + for (const t of [lib, src]) await t?.cleanup(); + }); + + it("拦截:钩子返回原因 → 记录失败(intercepted:<原因>),不动文件;整理结束事件 outcome=failed", async () => { + const hooks = new HookRegistry(); + const seen: string[][] = []; + hooks.forModule("test").intercept("organize.intercept", (c) => { + seen.push(c.files.map((f) => `${f.role}:${f.kind}`)); + return c.action === "copy" ? "演示：不许复制" : null; + }); + h.ctx.hooks = hooks; + const plan = await planImport(h.d, req()); + expect(plan).toMatchObject({ status: "failed", reason: "intercepted:演示：不许复制" }); + expect(seen).toEqual([["main:video", "sidecar:subtitle"]]); + const rec = await h.db + .selectFrom("organization_record") + .select(["status", "error"]) + .where("id", "=", plan.recordId!) + .executeTakeFirstOrThrow(); + expect(rec).toMatchObject({ status: "failed", error: { hook: "test#1", reason: "演示：不许复制" } }); + const fin = (await h.outbox("organization.finished")).filter((r) => r.payload.recordId === plan.recordId); + expect(fin.map((r) => [r.payload.outcome, r.payload.reason])).toEqual([["failed", "intercepted:演示：不许复制"]]); + }); + + it("改名:主视频按钩子返回的文件名入库,同主干的字幕跟着改;成功后发 organization.finished", async () => { + const hooks = new HookRegistry(); + const ctxs: unknown[] = []; + hooks.forModule("test").rewrite("organize.rename", (name, c) => { + ctxs.push({ name, mediaType: c.mediaType, mediaEntityId: c.mediaEntityId }); + return name.replace(/\.mkv$/, " [自定义].mkv"); + }); + // 返回带路径分隔符的结果会被忽略(排在后面,拿到的是上一个钩子的结果) + hooks.forModule("bad").rewrite("organize.rename", (name) => `../${name}`); + h.ctx.hooks = hooks; + const plan = await planImport(h.d, req()); + expect(plan.status).toBe("running"); + const ex = await executeRecord(h.d, plan.recordId!); + expect(ex.status).toBe("succeeded"); + expect(ctxs).toEqual([{ name: expect.stringMatching(/\.mkv$/), mediaType: "movie", mediaEntityId: entity }]); + const files = await h.db + .selectFrom("organization_file_result") + .select(["target_path", "status"]) + .where("organization_record_id", "=", plan.recordId!) + .execute(); + const targets = files.map((f) => f.target_path ?? "").sort(); + expect(targets).toHaveLength(2); + const video = targets.find((t) => t.endsWith(".mkv"))!; + const sub = targets.find((t) => t.endsWith(".srt"))!; + expect(video).toMatch(/ \[自定义\]\.mkv$/); + const stem = video.slice(0, -".mkv".length); + expect(sub.startsWith(stem)).toBe(true); + expect(existsSync(`${lib.root}/${video}`) || existsSync(video)).toBe(true); + const fin = (await h.outbox("organization.finished")).filter((r) => r.payload.recordId === plan.recordId); + expect(fin.map((r) => r.payload.outcome)).toEqual(["succeeded"]); + }); + + it("钩子出错 / 超时不影响整理:按模板名照常生成计划", async () => { + const hooks = new HookRegistry(); + hooks.forModule("broken").rewrite("organize.rename", () => { + throw new Error("坏了"); + }); + hooks.forModule("broken").intercept("organize.intercept", () => new Promise(() => {}), { timeoutMs: 20 }); + h.ctx.hooks = hooks; + const plan = await planImport(h.d, { ...req(), dryRun: true, action: "move" }); + expect(plan.status).toBe("queued"); + const main = plan.plan!.organize.files.find((f) => f.role === "main")!; + expect(main.dst).not.toContain("自定义"); + }); +});