From 1db689d9a716e3fae709f3cd47d5a1a9b0d8f705 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 6 Oct 2026 19:27:02 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F?= =?UTF-8?q?=E5=B9=BF=E6=92=AD=E4=BA=8B=E4=BB=B6=E4=B8=8E=E9=93=BE=E5=BC=8F?= =?UTF-8?q?=E9=92=A9=E5=AD=90=E8=AE=BE=E6=96=BD=EF=BC=88ADR-006=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - contracts:补齐生命周期广播事件(订阅、下载、整理结束、刮削、字幕、作品关注 / 元数据 / 季集结构、站点可达性) 的类型与 payload schema,LIFECYCLE_EVENTS 清单与 lifecycleKey 幂等键;HOOK_POINTS 链式钩子挂点; FEATURES 功能清单与 SLOT_POSITIONS 前端插槽位置 - outbox 中继支持一条事件路由给多个订阅方(其余订阅方用 <键>#<任务名>);生命周期事件允许多模块订阅, 没有订阅方时也被中继认领并镜像到 SSE - 服务端 HookRegistry:按注册顺序执行、rewrite / intercept 两类、单钩子超时与异常隔离、无注册零开销; ServerModule 增加 manifest 与 registerHooks,bootstrap 装配 ctx.hooks - ADR-006 与 server README「如何写一个功能模块」 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- apps/server/README.md | 29 +++ apps/server/src/context.ts | 61 ++++- apps/server/src/hooks/index.ts | 44 ++++ apps/server/src/hooks/points.ts | 81 +++++++ apps/server/src/hooks/registry.ts | 126 +++++++++++ apps/server/src/modules/types.ts | 24 +- apps/server/src/worker/index.ts | 14 +- apps/server/test/hooks/hook-registry.test.ts | 175 +++++++++++++++ apps/server/vitest.config.ts | 2 + docs/adr/ADR-006-lifecycle-events.md | 93 ++++++++ packages/contracts/src/events.ts | 224 ++++++++++++++++++- packages/contracts/src/features.ts | 119 ++++++++++ packages/contracts/src/hooks.ts | 44 ++++ packages/contracts/src/index.ts | 2 + packages/tasks/src/outbox-relay.ts | 17 +- packages/tasks/test/outbox-broadcast.test.ts | 88 ++++++++ 16 files changed, 1121 insertions(+), 22 deletions(-) create mode 100644 apps/server/src/hooks/index.ts create mode 100644 apps/server/src/hooks/points.ts create mode 100644 apps/server/src/hooks/registry.ts create mode 100644 apps/server/test/hooks/hook-registry.test.ts create mode 100644 docs/adr/ADR-006-lifecycle-events.md create mode 100644 packages/contracts/src/features.ts create mode 100644 packages/contracts/src/hooks.ts create mode 100644 packages/tasks/test/outbox-broadcast.test.ts diff --git a/apps/server/README.md b/apps/server/README.md index d32662ae..0fa90cfa 100644 --- a/apps/server/README.md +++ b/apps/server/README.md @@ -10,6 +10,7 @@ src/ context.ts bootstrap():ServerContext、任务注册表(模块 + 空壳兜底)、outbox 路由合并、连接引擎 http/app.ts createApp():系统路由 + 逐个挂 SERVER_MODULES 的 registerRoutes worker/index.ts startWorker():引擎 worker + OutboxRelay + 各模块 start() + hooks/ 链式钩子注册表与调用入口(ADR-006) modules/ types.ts ServerModule / ServerContext / OutboxWriter / SecretStore / SettingsReader(共享,不随子任务改) index.ts SERVER_MODULES 清单 @@ -64,3 +65,31 @@ export const sourcesIdentityModule: ServerModule = { `ServerContext` 字段:`db`(Kysely)、`config`、`logger`(pino,自动脱敏)、`events`(SSE 总线 `EventBusPort`)、`engine`(TaskEngine:enqueue/signal/cancel/get/list;register* 阶段不要调用)、`outbox`(`emit(trx, type, payload, {idempotencyKey})` + worker 的 `relay`)、`secrets`(SecretStore,W2-A 实现前调用会抛错)、`settings`(读 settings 表并合并缺省)、`capabilities`(CapabilityRegistry)。domain ↔ contracts 的取值转换一律用 `@umt/mapping`。 角色:`all` 单进程时连接引擎会顺带跑引擎迁移;`api` / `worker` 分进程部署时由 `--role migrate` 预先迁移(db + openworkflow),两个进程都以 `runMigrations:false` 连接。 + +## 如何写一个功能模块(ADR-006) + +外围功能(补字幕这类)按下面的约定接进来,不做插件系统、不做开关。样板是 `modules/subtitles/`。 + +1. **登记 manifest**:在 `packages/contracts/src/features.ts` 的 `FEATURES` 里加一份(名称、说明、怎么用、订阅的事件、能力、页面入口、插槽、钩子)。它会进智能体系统提示「能力地图」末尾的功能清单;`test/modules.test.ts` 核对模块与清单一一对应。 +2. **建模块**:`modules//index.ts` 导出 `ServerModule`,`name` 与 manifest id 相同,写 `manifest: featureManifest("")`,加进 `modules/index.ts`。 +3. **订阅广播事件**(事后通知,异步、至少一次):`outboxRoutes` 里写 `"<事件>": (ev) => 任务 | 任务数组 | null`。生命周期事件(`LIFECYCLE_EVENTS`)可以多个模块同时订阅;payload 用 `parseEventPayload` 取,按 ID 回查业务表;订阅任务要幂等(同一事件可能投递多次)。 + ```ts + outboxRoutes: (ctx) => ({ + "organization.finished": (ev) => { + const p = parseEventPayload("organization.finished", ev.payload); + return p?.outcome === "succeeded" ? { task: "myFollowUp", input: { recordId: p.recordId } } : null; + }, + }), + ``` +4. **注册链式钩子**(同步改写 / 拦截,挂点见 `packages/contracts/src/hooks.ts`): + ```ts + registerHooks(hooks) { + hooks.rewrite("organize.rename", (name, c) => (c.mediaType === "movie" ? name.replace(/ - 1080p/, "") : undefined)); + hooks.intercept("organize.intercept", (c) => (c.files.some((f) => f.dst.includes("/Sample/")) ? "目标在 Sample 目录" : null)); + }, + ``` + 钩子可能在业务事务里执行:只做内存里的判断,不发外部请求、不写库;缺省 2 秒超时,抛错 / 超时只跳过自己。 +5. **自己发事件**:业务节点在事务里 `ctx.outbox.emit(trx, "<事件>", payload, { idempotencyKey: lifecycleKey("<事件>", id) })`;新事件先在 `contracts/events.ts` 加类型、payload schema,并登记进 `LIFECYCLE_EVENTS`。 +6. **前端入口**:在 `apps/web/app/features/.tsx` 用插槽注册(作品页操作区 / 面板区、整理记录行操作、侧栏、设置分区),在 `features/index.ts` 调一次;默认直接显示。 +7. **智能体能力**(可选):照常在 contracts 定义能力、在 `registerCapabilities` 挂处理器,并登记 Skill 与能力地图(见仓库根 CLAUDE.md)。 + diff --git a/apps/server/src/context.ts b/apps/server/src/context.ts index d895fea1..bd184e0f 100644 --- a/apps/server/src/context.ts +++ b/apps/server/src/context.ts @@ -1,16 +1,18 @@ -import { type EventType, parseEventPayload } from "@umt/contracts"; +import { type EventType, LIFECYCLE_EVENTS, parseEventPayload } from "@umt/contracts"; import { type DB, type Kysely, sql } from "@umt/db"; import { createDefaultRegistry, type LogSink, OpenWorkflowEngine, type OutboxRelay, + type OutboxRoute, type OutboxRouteFn, type TaskEngine, TaskRegistry, } from "@umt/tasks"; import type { Logger } from "pino"; import type { Config } from "./config.ts"; +import { HookRegistry } from "./hooks/registry.ts"; import type { EventBusPort } from "./http/sse.ts"; import { assertCapabilityCount, createCapabilityRegistry } from "./modules/agent/capability-registry.ts"; import { createSecretStore, createSettingsReader } from "./modules/auth-settings/provided.ts"; @@ -97,24 +99,66 @@ export function buildTaskRegistry(modules: readonly ServerModule[], ctx: ServerC return registry; } -/** 合并各模块的 outbox 路由;同一事件 type 只能由一个模块认领 */ +const BROADCAST_TYPES = new Set(LIFECYCLE_EVENTS.map((e) => e.type)); + +/** + * 合并各模块的 outbox 路由。生命周期广播事件(LIFECYCLE_EVENTS,ADR-006)可以有多个订阅方,按模块顺序合成一个路由: + * 各订阅方各自返回任务(或 null),合起来交给中继逐个入队;某个订阅方的路由函数抛错只记日志,不连累别人。 + * 其它事件(内部流程,如 release.observed → resolveObservation)仍只能由一个模块认领。 + */ export function collectOutboxRoutes( modules: readonly ServerModule[], ctx: ServerContext, ): Record { - const routes: Record = {}; - const owner = new Map(); + const subscribers = new Map(); for (const m of modules) { for (const [type, fn] of Object.entries(m.outboxRoutes?.(ctx) ?? {})) { - const prev = owner.get(type); - if (prev) throw new Error(`outbox 事件 ${type} 被模块 ${prev} 与 ${m.name} 重复认领`); - owner.set(type, m.name); - routes[type] = fn; + const prev = subscribers.get(type) ?? []; + if (prev.length && !BROADCAST_TYPES.has(type)) + throw new Error(`outbox 事件 ${type} 被模块 ${prev[0]!.module} 与 ${m.name} 重复认领`); + subscribers.set(type, [...prev, { module: m.name, fn }]); } } + const routes: Record = {}; + for (const [type, subs] of subscribers) { + if (subs.length === 1) { + routes[type] = subs[0]!.fn; + continue; + } + routes[type] = (ev) => { + const out: OutboxRoute[] = []; + for (const s of subs) { + let r: ReturnType; + try { + r = s.fn(ev); + } catch (err) { + ctx.logger.warn( + { + type, + module: s.module, + eventId: ev.id, + err: { message: err instanceof Error ? err.message : String(err) }, + }, + "outbox 订阅方路由出错，已跳过", + ); + continue; + } + if (r === null) continue; + out.push(...((Array.isArray(r) ? r : [r]) as OutboxRoute[])); + } + return out.length ? out : null; + }; + } return routes; } +/** 链式钩子注册表:按模块顺序调 registerHooks,执行顺序即注册顺序(ADR-006) */ +export function buildHookRegistry(modules: readonly ServerModule[], ctx: ServerContext): HookRegistry { + const hooks = new HookRegistry(ctx.logger?.child({ component: "hooks" })); + for (const m of modules) m.registerHooks?.(hooks.forModule(m.name), ctx); + return hooks; +} + export interface BootstrapOptions { role: ServerRole; config: Config; @@ -158,6 +202,7 @@ export async function bootstrap(opts: BootstrapOptions): Promise { }; const tasks = buildTaskRegistry(opts.modules, ctx); for (const m of opts.modules) m.registerCapabilities?.(ctx.capabilities, ctx); + ctx.hooks = buildHookRegistry(opts.modules, ctx); // 装配了 agent 模块(即完整模块清单)时,首批 14 个能力必须全部挂上处理器 if (opts.modules.some((m) => m.name === "agent")) assertCapabilityCount(ctx.capabilities); engine.attach( diff --git a/apps/server/src/hooks/index.ts b/apps/server/src/hooks/index.ts new file mode 100644 index 00000000..84d7733f --- /dev/null +++ b/apps/server/src/hooks/index.ts @@ -0,0 +1,44 @@ +import type { HookPoint } from "@umt/contracts"; +import type { ServerContext } from "../modules/types.ts"; +import type { HookContext, HookValue, InterceptPoint, RewritePoint } from "./points.ts"; +import type { HookRejection } from "./registry.ts"; + +export type * from "./points.ts"; +export { + DEFAULT_HOOK_TIMEOUT_MS, + type HookOptions, + type HookRegistrar, + HookRegistry, + type HookRejection, +} from "./registry.ts"; + +// 业务节点调用钩子的入口。ctx.hooks 由 bootstrap 装配;测试里手搭的上下文没有它,等同于没有注册。 + +/** 这个挂点有没有钩子(上下文构造代价大时先问一下) */ +export function hasHook(ctx: Pick, point: HookPoint): boolean { + return ctx.hooks?.has(point) ?? false; +} + +/** 改写:没有钩子时原样返回同一个值;accept 判定单个钩子的结果能不能用(不能用的跳过) */ +export async function rewriteHook

( + ctx: Pick, + point: P, + value: HookValue

, + context: HookContext

| (() => HookContext

), + accept?: (v: HookValue

) => boolean, +): Promise> { + const h = ctx.hooks; + if (!h?.has(point)) return value; + return h.runRewrite(point, value, typeof context === "function" ? context() : context, accept); +} + +/** 拦截:返回第一个拒绝,没有钩子或都放行时返回 null */ +export async function interceptHook

( + ctx: Pick, + point: P, + context: HookContext

| (() => HookContext

), +): Promise { + const h = ctx.hooks; + if (!h?.has(point)) return null; + return h.runIntercept(point, typeof context === "function" ? context() : context); +} diff --git a/apps/server/src/hooks/points.ts b/apps/server/src/hooks/points.ts new file mode 100644 index 00000000..c49c8846 --- /dev/null +++ b/apps/server/src/hooks/points.ts @@ -0,0 +1,81 @@ +import type { HOOK_POINTS, HookPoint } from "@umt/contracts"; +import type { ParsedRelease } from "@umt/domain"; + +// 链式钩子挂点的入参 / 返回值类型(挂点清单与语义在 contracts/hooks.ts,ADR-006)。 +// rewrite 挂点:value 是要改写的值,context 只读;intercept 挂点只有 context。 + +/** 资源选择的候选:只能重排或过滤,不能加入新候选(按 sourceReleaseId 映射回原资源) */ +export interface SelectionCandidate { + sourceReleaseId: string; + title: string; + parsed: ParsedRelease; +} + +export interface HookPointTypes { + "name.recognize": { + value: ParsedRelease; + context: { + title: string; + subtitle: string | null; + /** 来源资源;按名称识别整理文件时为 null */ + sourceReleaseId: string | null; + trigger: string; + }; + }; + "organize.rename": { + /** 目标文件名(不含目录);返回值不能含路径分隔符 */ + value: string; + context: { + targetPath: string; + sourcePath: string; + mediaEntityId: string; + mediaType: "movie" | "tv"; + season: number | null; + episodes: { start: number; end: number } | null; + }; + }; + "organize.intercept": { + value: never; + context: { + mediaEntityId: string; + mediaType: "movie" | "tv"; + action: string; + trigger: string; + downloadTaskId: string | null; + files: { src: string; dst: string; kind: string; role: "main" | "sidecar" }[]; + }; + }; + "resource.selection": { + value: SelectionCandidate[]; + context: { mediaEntityId: string; subscriptionId: string | null; slotId: string | null }; + }; + "subscription.completion.check": { + value: never; + context: { + subscriptionId: string; + mediaEntityId: string | null; + mediaKind: string; + /** 本次判定完成的原因(slot_satisfied / download_accepted) */ + reason: string; + progress: Record; + }; + }; +} + +type KindOf

= (typeof HOOK_POINTS)[P]["kind"]; +export type RewritePoint = { [P in HookPoint]: KindOf

extends "rewrite" ? P : never }[HookPoint]; +export type InterceptPoint = { [P in HookPoint]: KindOf

extends "intercept" ? P : never }[HookPoint]; + +export type HookValue

= HookPointTypes[P]["value"]; +export type HookContext

= Readonly; + +type MaybePromise = T | Promise; +/** 返回新值;返回 undefined 表示不改 */ +export type RewriteHook

= ( + value: HookValue

, + context: HookContext

, +) => MaybePromise | undefined>; +/** 返回拒绝原因;返回 null / undefined 表示放行 */ +export type InterceptHook

= ( + context: HookContext

, +) => MaybePromise; diff --git a/apps/server/src/hooks/registry.ts b/apps/server/src/hooks/registry.ts new file mode 100644 index 00000000..5da11441 --- /dev/null +++ b/apps/server/src/hooks/registry.ts @@ -0,0 +1,126 @@ +import { HOOK_POINTS, type HookPoint } from "@umt/contracts"; +import type { Logger } from "pino"; +import type { HookContext, HookValue, InterceptHook, InterceptPoint, RewriteHook, RewritePoint } from "./points.ts"; + +// 链式钩子(ADR-006,参考 MoviePilot ChainEventType):模块启动时注册,业务节点同步调用。 +// - 按注册顺序执行(模块顺序 = SERVER_MODULES 顺序,同一模块内按调用顺序); +// - 每个钩子单独限时(缺省 2 秒),超时或抛错只记日志、跳过这个钩子,不影响主流程; +// - 没有注册时 has() 为 false,调用方直接用原值(零开销)。 +// 钩子可能在业务事务里被调用:只做内存里的判断与改写,不要发长耗时外部请求、不要写库。 + +export const DEFAULT_HOOK_TIMEOUT_MS = 2_000; + +export interface HookOptions { + /** 单次执行上限,缺省 DEFAULT_HOOK_TIMEOUT_MS */ + timeoutMs?: number; + /** 日志里显示的名字,缺省 <模块>#<序号> */ + name?: string; +} + +export interface HookRegistrar { + rewrite

(point: P, fn: RewriteHook

, opts?: HookOptions): void; + intercept

(point: P, fn: InterceptHook

, opts?: HookOptions): void; +} + +interface Entry { + owner: string; + name: string; + timeoutMs: number; + // biome-ignore lint/suspicious/noExplicitAny: 按挂点分表存,调用时由 run* 的类型参数约束 + fn: (...args: any[]) => unknown; +} + +const TIMEOUT = Symbol("timeout"); +const FAILED = Symbol("failed"); + +export interface HookRejection { + reason: string; + /** 拒绝的钩子名 */ + by: string; +} + +export class HookRegistry { + private readonly hooks = new Map(); + + constructor(private readonly logger?: Pick) {} + + /** 给某个模块的注册入口(owner 进日志) */ + forModule(owner: string): HookRegistrar { + const add = (point: HookPoint, kind: "rewrite" | "intercept", fn: Entry["fn"], opts: HookOptions = {}) => { + if (HOOK_POINTS[point].kind !== kind) + throw new Error(`钩子挂点 ${point} 是 ${HOOK_POINTS[point].kind} 类，不能用 ${kind} 注册`); + const list = this.hooks.get(point) ?? []; + list.push({ + owner, + name: opts.name ?? `${owner}#${list.filter((e) => e.owner === owner).length + 1}`, + timeoutMs: opts.timeoutMs ?? DEFAULT_HOOK_TIMEOUT_MS, + fn, + }); + this.hooks.set(point, list); + }; + return { + rewrite: (point, fn, opts) => add(point, "rewrite", fn, opts), + intercept: (point, fn, opts) => add(point, "intercept", fn, opts), + }; + } + + has(point: HookPoint): boolean { + return (this.hooks.get(point)?.length ?? 0) > 0; + } + + /** 已注册的钩子(挂点 → 钩子名),给功能清单与诊断用 */ + list(): Record { + return Object.fromEntries([...this.hooks].map(([p, l]) => [p, l.map((e) => e.name)])); + } + + /** 依次改写;某个钩子超时 / 抛错 / 结果不被 accept 接受时跳过它,沿用上一个值 */ + async runRewrite

( + point: P, + value: HookValue

, + context: HookContext

, + accept?: (v: HookValue

) => boolean, + ): Promise> { + let cur = value; + for (const e of this.hooks.get(point) ?? []) { + const r = await this.call(point, e, [cur, context]); + if (r === undefined || r === TIMEOUT || r === FAILED) continue; + if (accept && !accept(r as HookValue

)) { + this.logger?.warn({ point, hook: e.name }, "钩子返回的结果不可用，已跳过"); + continue; + } + cur = r as HookValue

; + } + return cur; + } + + /** 依次判定;第一个返回原因的钩子拒绝,后面的不再执行 */ + async runIntercept

(point: P, context: HookContext

): Promise { + for (const e of this.hooks.get(point) ?? []) { + const r = await this.call(point, e, [context]); + if (typeof r === "string" && r.trim()) return { reason: r.trim(), by: e.name }; + } + return null; + } + + private async call(point: HookPoint, e: Entry, args: unknown[]): Promise { + let timer: ReturnType | undefined; + try { + const r = await Promise.race([ + Promise.resolve().then(() => e.fn(...args)), + new Promise((resolve) => { + timer = setTimeout(() => resolve(TIMEOUT), e.timeoutMs); + }), + ]); + if (r === TIMEOUT) this.logger?.warn({ point, hook: e.name, timeoutMs: e.timeoutMs }, "钩子超时，已跳过"); + return r; + } catch (err) { + this.logger?.warn( + { point, hook: e.name, err: { message: err instanceof Error ? err.message : String(err) } }, + "钩子出错，已跳过", + ); + return FAILED; + } finally { + clearTimeout(timer); + } + } +} diff --git a/apps/server/src/modules/types.ts b/apps/server/src/modules/types.ts index 2c5951ca..df6838a6 100644 --- a/apps/server/src/modules/types.ts +++ b/apps/server/src/modules/types.ts @@ -1,10 +1,18 @@ import type { OpenAPIHono } from "@hono/zod-openapi"; import type { CapabilityRegistry } from "@umt/agent"; -import type { EventPayload, EventType, GlobalSettings, SecretPurpose, SettingsKey } from "@umt/contracts"; +import type { + EventPayload, + EventType, + FeatureManifest, + GlobalSettings, + SecretPurpose, + SettingsKey, +} from "@umt/contracts"; import type { DB, Kysely, Transaction } from "@umt/db"; import type { OutboxRelay, OutboxRouteFn, TaskEngine, TaskRegistry } from "@umt/tasks"; import type { Logger } from "pino"; import type { Config } from "../config.ts"; +import type { HookRegistrar, HookRegistry } from "../hooks/registry.ts"; import type { AuthEnv } from "../http/auth.ts"; import type { EventBusPort } from "../http/sse.ts"; @@ -72,6 +80,11 @@ export interface ServerContext { settings: SettingsReader; /** 能力注册中心(Web/Pi/CLI/MCP 共用一条 invoke 路径) */ capabilities: CapabilityRegistry; + /** + * 链式钩子(ADR-006):bootstrap 装配,模块在 registerHooks 里注册;业务节点经 hooks/index.ts 的 + * rewriteHook / interceptHook 调用。测试里手搭的上下文可以没有(等同于没有注册)。 + */ + hooks?: HookRegistry; } export interface ServerModule { @@ -84,8 +97,15 @@ export interface ServerModule { * 同名任务只能由一个模块注册;没人注册的 §3.1-8 任务由 @umt/tasks 的空壳兜底。 */ registerTasks(registry: TaskRegistry, ctx: ServerContext): void; - /** 所有角色:outbox 事件 type → 任务路由;同一 type 只能由一个模块认领 */ + /** + * 所有角色:outbox 事件 type → 任务路由。生命周期广播事件(ADR-006)可以被多个模块订阅, + * 中继按模块顺序把每个订阅方的路由都投递一遍;返回 null 表示这条事件本模块不处理。 + */ outboxRoutes?(ctx: ServerContext): Record; + /** 外围功能的说明(功能清单只给智能体看,不做开关);取 contracts 的 featureManifest(id) */ + manifest?: FeatureManifest; + /** 所有角色:注册链式钩子(同步、按注册顺序、可改写或拦截;见 hooks/registry.ts) */ + registerHooks?(hooks: HookRegistrar, ctx: ServerContext): void; /** 所有角色:用 contracts 的 capabilitySpec(id) 注册能力处理器 */ registerCapabilities?(registry: CapabilityRegistry, ctx: ServerContext): void; /** worker / all 角色:启动轮询器、定时器等后台循环(引擎与中继已启动) */ diff --git a/apps/server/src/worker/index.ts b/apps/server/src/worker/index.ts index 14761bcc..26aa9823 100644 --- a/apps/server/src/worker/index.ts +++ b/apps/server/src/worker/index.ts @@ -18,8 +18,20 @@ import { /** * outbox → SSE 桥:这些前缀的 outbox 事件在中继投递成功后镜像到实时总线(payload 原样,带业务 id), * 首页、收件箱、任务、来源与下载页据此失效刷新。feed.due 是调度器内部节拍,不推给浏览器。 + * 镜像列表同时决定中继认领哪些没有路由的事件:生命周期广播事件都在这里,否则没有订阅方时会一直挂在 pending。 */ -const SSE_MIRROR_PREFIXES = ["feed.", "decision.", "download.", "organization.", "subscription."] as const; +const SSE_MIRROR_PREFIXES = [ + "feed.", + "decision.", + "download.", + "organization.", + "subscription.", + // 生命周期广播事件(ADR-006):没有订阅方时也要被中继认领(标记已投递),顺带推给浏览器 + "scrape.", + "subtitle.", + "media.", + "site.", +] as const; const SSE_MIRROR_EXCLUDED = new Set([EventType.feedDue]); export const SSE_MIRRORED_OUTBOX_TYPES: readonly string[] = Object.values(EventType).filter( (t) => SSE_MIRROR_PREFIXES.some((p) => t.startsWith(p)) && !SSE_MIRROR_EXCLUDED.has(t), diff --git a/apps/server/test/hooks/hook-registry.test.ts b/apps/server/test/hooks/hook-registry.test.ts new file mode 100644 index 00000000..3d6d84af --- /dev/null +++ b/apps/server/test/hooks/hook-registry.test.ts @@ -0,0 +1,175 @@ +import type { ParsedRelease } from "@umt/domain"; +import { describe, expect, it, vi } from "vitest"; +import { HookRegistry, hasHook, interceptHook, rewriteHook } from "../../src/hooks/index.ts"; + +// 链式钩子(ADR-006):注册顺序、改写、拦截、抛错隔离、超时、没有注册时零开销。 + +const warnings = () => { + const warn = vi.fn(); + return { logger: { warn } as never, warn }; +}; +const parsed = (title: string) => ({ title, raw: title, explicitIds: [] }) as unknown as ParsedRelease; +const recognizeCtx = { title: "x", subtitle: null, sourceReleaseId: null, trigger: "observed" }; +const completionCtx = { + subscriptionId: "s", + mediaEntityId: null, + mediaKind: "movie", + reason: "slot_satisfied", + progress: {}, +}; + +describe("链式钩子注册表", () => { + it("按注册顺序执行(bootstrap 按模块顺序注册),每个钩子拿到上一个的结果;缺省名 <模块>#<序号>", async () => { + const reg = new HookRegistry(); + const seen: string[] = []; + const a = reg.forModule("a"); + const b = reg.forModule("b"); + a.rewrite("organize.rename", (name) => { + seen.push(`a1:${name}`); + return `${name}-a1`; + }); + b.rewrite("organize.rename", (name) => { + seen.push(`b:${name}`); + return `${name}-b`; + }); + a.rewrite("organize.rename", (name) => { + seen.push(`a2:${name}`); + return undefined; // 不改 + }); + const ctx = { hooks: reg }; + const out = await rewriteHook(ctx, "organize.rename", "X.mkv", { + targetPath: "/lib/X.mkv", + sourcePath: "/src/X.mkv", + mediaEntityId: "e", + mediaType: "movie", + season: null, + episodes: null, + }); + expect(out).toBe("X.mkv-a1-b"); + expect(seen).toEqual(["a1:X.mkv", "b:X.mkv-a1", "a2:X.mkv-a1-b"]); + expect(reg.list()).toEqual({ "organize.rename": ["a#1", "b#1", "a#2"] }); + }); + + it("改写生效:异步钩子改写解析结果", async () => { + const reg = new HookRegistry(); + reg.forModule("m").rewrite("name.recognize", async (p) => ({ ...p, title: "改写后" })); + const out = await rewriteHook({ hooks: reg }, "name.recognize", parsed("原名"), recognizeCtx); + expect(out.title).toBe("改写后"); + }); + + it("拦截生效:第一个返回原因的钩子拒绝,后面的不再执行;都放行返回 null", async () => { + const reg = new HookRegistry(); + const later = vi.fn(() => "不该执行"); + const m = reg.forModule("m"); + m.intercept("subscription.completion.check", () => null); + m.intercept("subscription.completion.check", () => " 还缺特别篇 ", { name: "缺特别篇" }); + m.intercept("subscription.completion.check", later); + expect(await interceptHook({ hooks: reg }, "subscription.completion.check", completionCtx)).toEqual({ + reason: "还缺特别篇", + by: "缺特别篇", + }); + expect(later).not.toHaveBeenCalled(); + + const pass = new HookRegistry(); + pass.forModule("m").intercept("subscription.completion.check", () => undefined); + expect(await interceptHook({ hooks: pass }, "subscription.completion.check", completionCtx)).toBeNull(); + }); + + it("抛错隔离:出错的钩子跳过并记日志,前后的钩子照常生效", async () => { + const { logger, warn } = warnings(); + const reg = new HookRegistry(logger); + const m = reg.forModule("m"); + m.rewrite("name.recognize", (p) => ({ ...p, title: `${p.title}+1` })); + m.rewrite("name.recognize", () => { + throw new Error("插件坏了"); + }); + m.rewrite("name.recognize", async () => { + throw new Error("异步也坏了"); + }); + m.rewrite("name.recognize", (p) => ({ ...p, title: `${p.title}+2` })); + const out = await rewriteHook({ hooks: reg }, "name.recognize", parsed("t"), recognizeCtx); + expect(out.title).toBe("t+1+2"); + expect(warn).toHaveBeenCalledTimes(2); + expect(warn.mock.calls[0]![0]).toMatchObject({ + point: "name.recognize", + hook: "m#2", + err: { message: "插件坏了" }, + }); + + // 拦截类抛错按放行处理 + const r2 = new HookRegistry(logger); + r2.forModule("m").intercept("organize.intercept", () => { + throw new Error("x"); + }); + expect( + await interceptHook({ hooks: r2 }, "organize.intercept", { + mediaEntityId: "e", + mediaType: "movie", + action: "copy", + trigger: "auto", + downloadTaskId: null, + files: [], + }), + ).toBeNull(); + }); + + it("超时:超过上限的钩子跳过(沿用上一个值),不等它跑完", async () => { + const { logger, warn } = warnings(); + const reg = new HookRegistry(logger); + const m = reg.forModule("slow"); + m.rewrite("organize.rename", () => new Promise((r) => setTimeout(() => r("迟到.mkv"), 5_000)), { + timeoutMs: 30, + }); + m.rewrite("organize.rename", (n) => `${n}.ok`); + const t0 = Date.now(); + const out = await reg.runRewrite("organize.rename", "A.mkv", { + targetPath: "/l/A.mkv", + sourcePath: "/s/A.mkv", + mediaEntityId: "e", + mediaType: "movie", + season: null, + episodes: null, + }); + expect(out).toBe("A.mkv.ok"); + expect(Date.now() - t0).toBeLessThan(2_000); + expect(warn.mock.calls[0]![0]).toMatchObject({ point: "organize.rename", hook: "slow#1", timeoutMs: 30 }); + }); + + it("没有注册时零开销:不构造上下文、原样返回同一个对象;没有注册表(手搭上下文)同样", async () => { + const ctxBuilder = vi.fn(() => recognizeCtx); + const p = parsed("t"); + const empty = new HookRegistry(); + // 别的挂点有钩子不影响这里 + empty.forModule("m").intercept("organize.intercept", () => "x"); + expect(hasHook({ hooks: empty }, "name.recognize")).toBe(false); + expect(await rewriteHook({ hooks: empty }, "name.recognize", p, ctxBuilder)).toBe(p); + expect(await rewriteHook({}, "name.recognize", p, ctxBuilder)).toBe(p); + expect(await interceptHook({}, "subscription.completion.check", () => completionCtx)).toBeNull(); + expect(ctxBuilder).not.toHaveBeenCalled(); + }); + + it("调用方给了 accept 时,结果不可用的钩子跳过,后面的钩子拿到的是上一个可用值", async () => { + const { logger, warn } = warnings(); + const reg = new HookRegistry(logger); + const m = reg.forModule("m"); + m.rewrite("organize.rename", (n) => `../${n}`); + m.rewrite("organize.rename", (n) => `${n}.ok`); + const c = { + targetPath: "/a", + sourcePath: "/b", + mediaEntityId: "e", + mediaType: "movie" as const, + season: null, + episodes: null, + }; + expect(await rewriteHook({ hooks: reg }, "organize.rename", "A.mkv", c, (v) => !v.includes("/"))).toBe("A.mkv.ok"); + expect(warn.mock.calls[0]![0]).toMatchObject({ point: "organize.rename", hook: "m#1" }); + }); + + it("用错挂点类型注册直接报错", () => { + const reg = new HookRegistry(); + // biome-ignore lint/suspicious/noExplicitAny: 故意绕过类型检查 + const m = reg.forModule("m") as any; + expect(() => m.rewrite("organize.intercept", () => "x")).toThrow(/intercept 类/); + }); +}); diff --git a/apps/server/vitest.config.ts b/apps/server/vitest.config.ts index 7c6b5034..75b2c5d2 100644 --- a/apps/server/vitest.config.ts +++ b/apps/server/vitest.config.ts @@ -14,6 +14,8 @@ const UNIT = [ "test/search/work-match.test.ts", "test/auth.test.ts", "test/health.test.ts", + "test/hooks/hook-registry.test.ts", + "test/hooks/resource-selection.test.ts", "test/sources-identity/identity-differential.test.ts", "test/sources-identity/enrich-sources.test.ts", "test/invoke-limits.test.ts", diff --git a/docs/adr/ADR-006-lifecycle-events.md b/docs/adr/ADR-006-lifecycle-events.md new file mode 100644 index 00000000..ac4cbb1a --- /dev/null +++ b/docs/adr/ADR-006-lifecycle-events.md @@ -0,0 +1,93 @@ +# ADR-006：生命周期事件、链式钩子与功能模块约定 + +- 状态：采纳（2026-10-06，用户确认方向）。 +- 关联：ADR-002（outbox 与任务引擎）、`packages/contracts/src/events.ts`（`LIFECYCLE_EVENTS`）、`packages/contracts/src/hooks.ts`（`HOOK_POINTS`）、`packages/contracts/src/features.ts`（`FEATURES`、`SLOT_POSITIONS`）、`apps/server/src/hooks/`、`apps/web/app/lib/slots.ts`。 +- 参照：MoviePilot v3 `app/schemas/types.py` 的 `EventType`（广播）与 `ChainEventType`（链式），分发在 `app/runtime/events.py`。 + +## 背景 + +外围功能（补字幕、以后的入库后通知、自动补字幕等）需要在业务节点上「知道发生了什么」或「插一手」。MoviePilot 用插件系统 + 两类事件解决。我们是内部工具，用户定了： + +- 不做插件系统、插件市场、启停开关； +- 做「功能模块」约定：功能清单只给智能体知道；补齐生命周期事件（参考 MP 的两类事件）；前端插槽要做、默认直接显示。 + +## 决定 + +### 1. 两类事件 + +| | 广播事件 | 链式钩子 | +| --- | --- | --- | +| 对照 MP | `EventType`（transfer.complete、subscribe.added …） | `ChainEventType`（name.recognize、transfer.rename …） | +| 时机 | 事后通知：事情已经发生并提交 | 事中：业务流程走到挂点时同步调用 | +| 传递 | 业务事务里写 `outbox_event`，中继投递成任务（异步、至少一次） | 进程内直接调用，按注册顺序 | +| 能做什么 | 订阅方自己建任务做后续动作 | 改写一个值（rewrite）或拒绝（intercept） | +| 失败 | 订阅任务按引擎重试；路由函数抛错只跳过该订阅方 | 钩子抛错 / 超时只记日志、跳过该钩子，主流程照常 | +| 去重 | outbox 幂等键（`/<部分…>`），订阅任务用引擎幂等键 | 不需要 | + +**何时用哪类**：只想在「X 发生之后」做点事（发通知、补字幕、预热缓存）→ 订阅广播事件；要改变正在进行的决定（换个名字、换个候选、不让整理、晚点完成）→ 注册链式钩子。拿不准就用广播：它不在主流程里，坏了也不影响主流程。 + +**事务边界**： + +- 广播事件在业务事务**里**写 outbox（`ctx.outbox.emit(trx, …)`），与业务写入一起提交或回滚；payload 只放 ID 和少量判断用的事实，订阅方按 ID 回查业务表（事件可能晚到、重复到）。 +- 钩子可能在业务事务里被调用（订阅完成判定就在行锁里）：只做内存里的判断与改写，**不要发外部请求、不要写库**；缺省每个钩子 2 秒上限。需要查外部的，改成订阅广播事件后异步做。 + +### 2. 广播事件清单 + +payload schema 都在 `EVENT_PAYLOADS`（`emit` 时校验，不合法直接抛错）。已有事件保持原样；`download.status_changed` / `organization.changed` 仍是状态流（SSE 与增量游标用，每次变化都发），下表是给功能模块订阅的生命周期事件。 + +| 事件 | 含义 | 发出位置 | 幂等键 | +| --- | --- | --- | --- | +| `release.resolved` | 资源识别完成（已有） | sources-identity `resolve.ts` 等 | 已有 | +| `subscription.created` | 订阅已建 | 订阅 API 新建、策略自动建（`acquisition.ts`） | `subscription.created/` | +| `subscription.updated` | 订阅已改（参数、暂停、恢复、重新订阅；`changes` 列字段） | 订阅 API 修改、状态动作 | 无（每次改动一条） | +| `subscription.completed` | 订阅已完成 | `completeSubscription`、停止日期到达 | `…//<历史行 id>`、`…//stop_after:<日期>` | +| `subscription.cancelled` | 订阅已取消（删除即取消） | 订阅 API 删除 / 改状态为 cancelled | 无 | +| `download.submitted` | 下载已提交（已有） | `downloads.ts` 提交 | 已有 | +| `download.completed` / `download.failed` / `download.removed` | 下载完成 / 失败 / 移除 | `downloads.ts` `emitStatus`（与 `download.status_changed` 同事务） | `/<任务 id>/g<代数>` | +| `organization.finished` | 整理结束，`outcome` = succeeded / partial / failed（统一 `organization.changed` 的终态） | `emitRecord`（所有整理记录写入都经过它） | `organization.finished/<记录 id>/` | +| `scrape.completed` / `scrape.failed` | 刮削完成（含部分 / 跳过）/ 失败 | `scrape.ts` `runScrapeRecord` 收尾 | `/<记录 id>/<结束时间>` | +| `subtitle.written` | 字幕已写入（每个文件一条） | 补字幕 `applySubtitles` | `subtitle.written/<写入 id>` | +| `media.watched` | 作品进入关注（建订阅、开始下载、用户搜索 / 手动认定） | 调用方：订阅 API、`submitDownload`、`resolveRelease` | `media.watched/<作品>/<原因>/<触发对象 id>` | +| `media.metadata_refreshed` | 作品元数据已刷新 | `refreshMetadata` 任务处理器（调用方） | `…/<作品>/<任务 run id>` | +| `media.episode_structure_changed` | 季集结构已切换（剧集组） | `setEpisodeGroup` 事务内 | `…/<作品>/<时间>` | +| `site.unreachable` / `site.recovered` | 站点不可达 / 恢复 | 列表抓取连续失败到阈值（与待处理同口径）、登录失效、站点检测结果跨越 | 列表抓取带 `…/<站点>/<快照 id>`，其余只在跨越时发 | + +幂等键用斜杠分隔，与已有的 `:` 键分开（`download.completed:` 已被 `download.status_changed` 占用，outbox 幂等键全表唯一）。 + +**订阅**：模块在 `outboxRoutes` 里写 `type → 路由函数`。生命周期事件可以有多个订阅方（`collectOutboxRoutes` 按模块顺序合成），其它事件（内部流程，如 `release.observed → resolveObservation`）仍只能一个模块认领。路由函数返回任务、任务数组或 `null`（不建任务）；中继对第一个任务用事件的幂等键、其余用 `<键>#<任务名>`。没有订阅方的生命周期事件也会被中继认领、标记已投递，并镜像到 SSE（`worker/index.ts` 的镜像前缀）。 + +### 3. 链式钩子清单 + +挂点名称与语义在 `HOOK_POINTS`，入参 / 返回值类型在 `apps/server/src/hooks/points.ts`。 + +| 挂点 | 类型 | 位置 | 能做什么 | +| --- | --- | --- | --- | +| `name.recognize` | rewrite | `resolveRelease`（标题解析后、召回前）、`identifyParsed`（整理 / 接管按名称识别） | 改写解析结果（标题、年份、季集、外部 ID） | +| `organize.rename` | rewrite | `planImport`，模板命名生成目标之后（保留原名 / 原地入库不调） | 改写主视频目标文件名；同主干的附属文件跟着改；结果带路径分隔符或空名的钩子跳过；改名后撞名转 `duplicate_target` | +| `organize.intercept` | intercept | `planImport`，写整理记录之前（预演也调） | 返回原因即拒绝：记录为失败，`reason = intercepted:<原因>` | +| `resource.selection` | rewrite | 订阅 / 收录版本重评（`reevaluate`）择优排序之后 | 重排或过滤候选；不认识的 ID 忽略，不能加入新候选 | +| `subscription.completion.check` | intercept | `completeSubscription`，标记完成之前 | 追加完成条件：返回原因即这次不完成，订阅继续追踪 | + +语义： + +- 注册在模块启动时完成：`ServerModule.registerHooks(hooks, ctx)`，`bootstrap` 按 `SERVER_MODULES` 顺序调用；执行顺序即注册顺序（不设优先级数字，需要先后就调整模块顺序）。 +- rewrite：每个钩子拿到上一个钩子的结果，返回新值或 `undefined`（不改）。intercept：第一个返回非空原因的钩子拒绝，后面不再执行。 +- 每个钩子单独限时（缺省 2 秒，可在注册时改），超时或抛错记 warn 日志、跳过这个钩子；拦截类出错按放行处理。 +- 没有注册时调用方直接用原值（`hasHook` 为 false，不构造上下文）。 + +MP 的其它链式事件（`transfer.overwrite.check`、`resource.download`、`discover.source`、`recommend.source` 等）等有真实需求再加：在 `HOOK_POINTS` 登记、在 `points.ts` 写类型、在业务节点调用。 + +### 4. 功能模块约定 + +一个外围功能 = 服务端一个 `ServerModule` + contracts 里一份 manifest + （可选）前端插槽注册： + +- **manifest**（`FEATURES`）：名称、说明、怎么用、订阅的事件、提供的能力、页面入口、插槽位置、钩子。服务端模块 `manifest: featureManifest("")`；`modules.test.ts` 核对模块与清单一一对应、订阅的事件与 `outboxRoutes` 一致。 +- **功能清单只给智能体**：系统提示「能力地图」一节末尾列出外围功能（`renderFeatures`），能力只列本次可用的。**不做开关**：功能都直接启用。 +- **前端插槽**（`apps/web/app/lib/slots.ts`）：位置见 `SLOT_POSITIONS`（侧栏入口、作品页操作区、作品页面板区、整理记录行操作、设置分区）。动作类位置注册 `{ label, icon, visible?, Dialog }`，页面渲染成按钮 / 菜单项、点了打开对话框；面板类位置注册组件，挂载点按顺序渲染。注册写在 `app/features/<功能>.tsx`，`app/features/index.ts` 汇总。**默认直接显示，不做开关**。已挂载：作品页操作区、作品页面板区（`library.$id.tsx`）、整理记录行菜单（`organize.tsx`）；侧栏与设置分区先只定义位置。 +- 样板：补字幕（`modules/subtitles`）——manifest、订阅 `organization.finished`（认出值得补字幕的整理，自动补字幕还没做，只记调试日志不建任务）、写入后发 `subtitle.written`、前端在作品页操作区和整理记录行菜单挂「补字幕」。 + +## 不做 + +- 插件系统、热加载、插件市场、功能开关。 +- 钩子优先级数字、钩子里做外部请求。 +- 把广播事件放进内存队列（MP 做法）：我们已有 outbox，事件要能跨进程、跨重启。 diff --git a/packages/contracts/src/events.ts b/packages/contracts/src/events.ts index 92fb7a08..dbf81f15 100644 --- a/packages/contracts/src/events.ts +++ b/packages/contracts/src/events.ts @@ -4,8 +4,11 @@ import { Id, NonNegInt, Timestamp } from "./common.ts"; import { DownloadStatus, PauseReason, TaskKind, TaskStatus } from "./execution.ts"; import { IdentifyCompletedPayload } from "./files.ts"; import { EvidenceLevel } from "./identity.ts"; -import { OrganizationStatus } from "./organization.ts"; +import { OrganizationAction, OrganizationStatus, OrganizationTrigger } from "./organization.ts"; +import { SubscriptionMediaKind, SubscriptionOrigin, SubscriptionStatus } from "./policy.ts"; +import { ScrapeStatus, ScrapeTrigger } from "./scrape.ts"; import { FeedDuePayload, FeedGapSuspectedPayload, FeedHealth, ReleaseObservedPayload } from "./sources.ts"; +import { SubtitleLanguage, SubtitleSource } from "./subtitles.ts"; /** SSE 事件信封:seq 在进程内单调递增,客户端断线后用 ?after= 补读 */ export const EventEnvelope = z.object({ @@ -24,10 +27,11 @@ export type EventEnvelope

= { /** * 已知事件类型:`域.动作`。新增事件先在这里加类型与 payload schema。 * - * 没有单独的 download.completed / organization.completed 事件,约定用现有事件表达(W2 收口): - * - "下载完成" = `download.status_changed` 且 `to === "completed"`(outbox 幂等键 `download.completed:`,只发一次; - * acquisition-organize 据此入队 verifyAndImport); - * - "入库完成" = `organization.changed` 且 `status === "succeeded"`(或 partial);`changeSeq` 与公开增量 API 游标一致。 + * 状态流事件(`download.status_changed`、`organization.changed`)每次状态变化都发,给 SSE 与增量游标用; + * 生命周期广播事件(LIFECYCLE_EVENTS,ADR-006)只在业务节点发一次,给功能模块订阅: + * - 下载完成 = `download.completed`(与 `download.status_changed → completed` 同事务发出;acquisition-organize + * 仍按后者入队 verifyAndImport,幂等键 `download.completed:` 属于后者); + * - 整理结束 = `organization.finished`,`outcome` 区分成功 / 部分成功 / 失败。 */ export const EventType = { heartbeat: "system.heartbeat", @@ -59,6 +63,24 @@ export const EventType = { decisionUpdated: "decision.updated", identifyCompleted: "identify.completed", + + // ── 生命周期广播事件(ADR-006):业务节点事后通知,经 outbox,至少一次 ── + subscriptionCreated: "subscription.created", + subscriptionUpdated: "subscription.updated", + subscriptionCompleted: "subscription.completed", + subscriptionCancelled: "subscription.cancelled", + downloadCompleted: "download.completed", + downloadFailed: "download.failed", + downloadRemoved: "download.removed", + organizationFinished: "organization.finished", + scrapeCompleted: "scrape.completed", + scrapeFailed: "scrape.failed", + subtitleWritten: "subtitle.written", + mediaWatched: "media.watched", + mediaMetadataRefreshed: "media.metadata_refreshed", + mediaEpisodeStructureChanged: "media.episode_structure_changed", + siteUnreachable: "site.unreachable", + siteRecovered: "site.recovered", } as const; export type EventType = (typeof EventType)[keyof typeof EventType]; @@ -172,6 +194,99 @@ export const DecisionUpdatedPayload = z.object({ status: DecisionStatus, }); +// ── 生命周期广播事件 payload(ADR-006)── +// payload 只放业务 ID 与订阅方做判断要用的少量事实;详情由订阅方按 ID 查业务表(事件可能晚到、重复到)。 + +const SubscriptionRef = z.object({ + subscriptionId: Id, + mediaEntityId: Id.nullable(), + mediaKind: SubscriptionMediaKind, +}); +/** 订阅已建(用户 / 智能体 / 策略) */ +export const SubscriptionCreatedPayload = SubscriptionRef.extend({ + origin: SubscriptionOrigin, + status: SubscriptionStatus, +}); +/** 订阅已改:参数或状态(暂停 / 恢复 / 重新订阅);changes 是改动的字段名 */ +export const SubscriptionUpdatedPayload = SubscriptionRef.extend({ + status: SubscriptionStatus, + changes: z.array(z.string()), +}); +/** 订阅已完成:reason 如 slot_satisfied / download_accepted / stop_after_reached */ +export const SubscriptionCompletedPayload = SubscriptionRef.extend({ reason: z.string() }); +/** 订阅已取消(删除订阅即取消,行保留) */ +export const SubscriptionCancelledPayload = SubscriptionRef.extend({ reason: z.string() }); + +const DownloadRef = z.object({ + downloadTaskId: Id, + planId: Id.nullable(), + mediaEntityId: Id.nullable(), + name: z.string(), + /** 任务行代数:removed 行复活后 +1 */ + generation: NonNegInt, +}); +/** 下载已完成(种子下完,尚未整理) */ +export const DownloadCompletedPayload = DownloadRef; +/** 下载失败 */ +export const DownloadFailedPayload = DownloadRef.extend({ from: DownloadStatus.nullable() }); +/** 下载已移除(用户删除 / 清理 / 下载器里找不到) */ +export const DownloadRemovedPayload = DownloadRef.extend({ from: DownloadStatus.nullable() }); + +/** 整理结束:一条整理记录进入 succeeded / partial / failed(统一 organization.changed 的终态) */ +export const OrganizationFinishedPayload = z.object({ + recordId: Id, + outcome: z.enum(["succeeded", "partial", "failed"]), + action: OrganizationAction, + trigger: OrganizationTrigger, + mediaEntityId: Id.nullable(), + downloadTaskId: Id.nullable(), + reason: z.string().nullable(), + changeSeq: z.string().regex(/^\d+$/), +}); + +const ScrapeRef = z.object({ + scrapeRecordId: Id, + mediaEntityId: Id.nullable(), + organizationRecordId: Id.nullable(), + trigger: ScrapeTrigger, + status: ScrapeStatus, + written: NonNegInt, + failed: NonNegInt, +}); +/** 刮削完成(succeeded / partial / skipped) */ +export const ScrapeCompletedPayload = ScrapeRef; +/** 刮削失败 */ +export const ScrapeFailedPayload = ScrapeRef.extend({ reason: z.string().nullable() }); + +/** 字幕已写入媒体库(一条写入历史一条事件) */ +export const SubtitleWrittenPayload = z.object({ + writeId: Id, + mediaEntityId: Id, + videoLocationId: Id, + language: SubtitleLanguage, + source: SubtitleSource, +}); + +/** 作品进入关注(建订阅、开始下载、用户搜索 / 手动认定);reason 为触发来源 */ +export const MediaWatchedPayload = z.object({ mediaEntityId: Id, reason: z.string() }); +/** 作品元数据已刷新;changedFields 为本次变化的字段 */ +export const MediaMetadataRefreshedPayload = z.object({ + mediaEntityId: Id, + full: z.boolean(), + changedFields: z.array(z.string()), +}); +/** 季集结构已切换(TMDB 默认季 ⇄ 剧集组);null 表示默认季 */ +export const MediaEpisodeStructureChangedPayload = z.object({ + mediaEntityId: Id, + fromGroupId: z.string().nullable(), + toGroupId: z.string().nullable(), +}); + +/** 站点不可达:登录失效、列表抓取连续失败或检测失败;reason 为原因码 */ +export const SiteUnreachablePayload = z.object({ siteId: Id, reason: z.string(), message: z.string().nullable() }); +/** 站点恢复 */ +export const SiteRecoveredPayload = z.object({ siteId: Id, reason: z.string() }); + /** 事件类型 → payload schema */ export const EVENT_PAYLOADS = { "system.heartbeat": HeartbeatPayload, @@ -195,6 +310,22 @@ export const EVENT_PAYLOADS = { "decision.created": DecisionCreatedPayload, "decision.updated": DecisionUpdatedPayload, "identify.completed": IdentifyCompletedPayload, + "subscription.created": SubscriptionCreatedPayload, + "subscription.updated": SubscriptionUpdatedPayload, + "subscription.completed": SubscriptionCompletedPayload, + "subscription.cancelled": SubscriptionCancelledPayload, + "download.completed": DownloadCompletedPayload, + "download.failed": DownloadFailedPayload, + "download.removed": DownloadRemovedPayload, + "organization.finished": OrganizationFinishedPayload, + "scrape.completed": ScrapeCompletedPayload, + "scrape.failed": ScrapeFailedPayload, + "subtitle.written": SubtitleWrittenPayload, + "media.watched": MediaWatchedPayload, + "media.metadata_refreshed": MediaMetadataRefreshedPayload, + "media.episode_structure_changed": MediaEpisodeStructureChangedPayload, + "site.unreachable": SiteUnreachablePayload, + "site.recovered": SiteRecoveredPayload, } as const satisfies Record; export type EventPayload = z.infer<(typeof EVENT_PAYLOADS)[T]>; @@ -210,6 +341,89 @@ export const TypedEventEnvelope = z.discriminatedUnion( ) as unknown as [z.ZodObject, ...z.ZodObject[]], ); +/** + * 生命周期广播事件清单(ADR-006):给功能清单、智能体提示与文档用。 + * 已有的 release.resolved 也算一条(资源识别完成)。emittedBy 是发出的模块,说明写在什么时候发。 + */ +export const LIFECYCLE_EVENTS: readonly { type: EventType; label: string; emittedBy: string; when: string }[] = [ + { type: "release.resolved", label: "资源识别完成", emittedBy: "sources-identity", when: "来源资源认到本地作品" }, + { + type: "subscription.created", + label: "订阅已建", + emittedBy: "acquisition-organize", + when: "新建订阅（含策略自动建）", + }, + { + type: "subscription.updated", + label: "订阅已改", + emittedBy: "acquisition-organize", + when: "改参数、暂停、恢复、重新订阅", + }, + { + type: "subscription.completed", + label: "订阅已完成", + emittedBy: "acquisition-organize", + when: "收录版本全部满足或到达停止日期", + }, + { type: "subscription.cancelled", label: "订阅已取消", emittedBy: "acquisition-organize", when: "删除订阅" }, + { type: "download.submitted", label: "下载已提交", emittedBy: "acquisition-organize", when: "种子交给下载器" }, + { type: "download.completed", label: "下载已完成", emittedBy: "acquisition-organize", when: "下载器报告下完" }, + { type: "download.failed", label: "下载失败", emittedBy: "acquisition-organize", when: "下载任务进入失败" }, + { + type: "download.removed", + label: "下载已移除", + emittedBy: "acquisition-organize", + when: "任务被删除或在下载器里消失", + }, + { + type: "organization.finished", + label: "整理结束", + emittedBy: "acquisition-organize", + when: "整理记录成功、部分成功或失败", + }, + { + type: "scrape.completed", + label: "刮削完成", + emittedBy: "acquisition-organize", + when: "刮削记录成功 / 部分成功 / 跳过", + }, + { type: "scrape.failed", label: "刮削失败", emittedBy: "acquisition-organize", when: "刮削记录失败" }, + { type: "subtitle.written", label: "字幕已写入", emittedBy: "subtitles", when: "补字幕写入一个字幕文件" }, + { + type: "media.watched", + label: "作品进入关注", + emittedBy: "sources-identity / acquisition-organize", + when: "建订阅、开始下载、搜索认定作品", + }, + { + type: "media.metadata_refreshed", + label: "作品元数据已刷新", + emittedBy: "sources-identity", + when: "刷新元数据任务完成", + }, + { + type: "media.episode_structure_changed", + label: "季集结构已切换", + emittedBy: "sources-identity", + when: "作品切换剧集组", + }, + { + type: "site.unreachable", + label: "站点不可达", + emittedBy: "sources-identity", + when: "登录失效、列表连续抓取失败或检测失败", + }, + { type: "site.recovered", label: "站点恢复", emittedBy: "sources-identity", when: "不可达的站点再次正常" }, +]; + +/** + * 生命周期事件的 outbox 幂等键:`/<部分…>`。用斜杠与已有的 `:` 键分开 + * (`download.completed:` 已被 download.status_changed 占用,outbox 幂等键全表唯一)。 + */ +export function lifecycleKey(type: EventType, ...parts: (string | number)[]): string { + return [type, ...parts].join("/"); +} + export function isEventType(type: string): type is EventType { return Object.hasOwn(EVENT_PAYLOADS, type); } diff --git a/packages/contracts/src/features.ts b/packages/contracts/src/features.ts new file mode 100644 index 00000000..7c671fb5 --- /dev/null +++ b/packages/contracts/src/features.ts @@ -0,0 +1,119 @@ +import type { KnownCapabilityId } from "./capability.ts"; +import type { EventType } from "./events.ts"; +import type { HookPoint } from "./hooks.ts"; + +// 功能模块约定(ADR-006):外围功能各自声明一份 manifest,汇总成功能清单。 +// 功能清单只给智能体看(系统提示的「功能清单」一节),不做开关;前端插槽默认直接显示。 + +/** 前端插槽位置(apps/web/app/lib/slots.ts 按这些位置注册组件) */ +export const SLOT_POSITIONS = { + "sidebar.entry": "侧栏入口", + "entity.actions": "作品页操作区", + "entity.panels": "作品页面板区", + "organize.record.actions": "整理记录行操作", + "settings.sections": "设置分区", +} as const; +export type SlotPosition = keyof typeof SLOT_POSITIONS; + +export interface FeatureManifest { + /** 与服务端模块名一致(modules//) */ + id: string; + name: string; + /** 一句话说明这个功能做什么 */ + summary: string; + /** 怎么用:入口在哪、智能体能做什么、不能做什么 */ + usage: string; + /** 订阅的生命周期事件 */ + subscribes: readonly EventType[]; + /** 提供的能力(智能体可调用) */ + capabilities: readonly KnownCapabilityId[]; + /** 页面入口(界面上的位置) */ + pages: readonly string[]; + /** 挂在哪些前端插槽 */ + slots: readonly SlotPosition[]; + /** 注册的链式钩子 */ + hooks: readonly HookPoint[]; +} + +/** 功能清单。新增外围功能:在这里加一份,服务端模块 manifest 指向它(featureManifest(id)) */ +export const FEATURES = [ + { + id: "subtitles", + name: "补字幕", + summary: "给媒体库里的视频补字幕：上传或从 SubHD / 字幕库在线搜索下载，预览配对后写到视频同目录，可看历史、撤销。", + usage: + "入口在作品页「补字幕」和整理记录行菜单「补字幕」；字幕站在站点页单列一组。智能体没有补字幕能力，用户问到时指引去作品页操作；整理结束事件已订阅，整理后自动补字幕还没做。", + subscribes: ["organization.finished"], + capabilities: [], + pages: ["作品页「补字幕」", "整理记录行菜单「补字幕」", "站点页「字幕站」"], + slots: ["entity.actions", "organize.record.actions"], + hooks: [], + }, + { + id: "browser", + name: "原站视图", + summary: "在服务端浏览器里打开站点原页面：人工接管、过验证、从页面下载种子或字幕。", + usage: "入口在站点页「原站视图」；智能体可以读原站页面、点击、下载种子，挑战页走视觉子流程。", + subscribes: [], + capabilities: [ + "browser.snapshot", + "browser.navigate", + "browser.interact", + "browser.download", + "browser.solve_challenge", + ], + pages: ["站点页「原站视图」"], + slots: [], + hooks: [], + }, + { + id: "desktop", + name: "桌面模式", + summary: "原站视图的远程桌面：直接操作浏览器容器的桌面，处理复杂验证。", + usage: "入口在原站视图的桌面模式；只给人用，智能体不操作。", + subscribes: [], + capabilities: [], + pages: ["原站视图「桌面模式」"], + slots: [], + hooks: [], + }, + { + id: "websearch", + name: "网页搜索", + summary: "用 Brave 搜索站外公开信息，并读取搜索结果的正文（结果是不可信数据）。", + usage: "在设置 → 智能体 → 网页搜索里填 Key 后可用；智能体先搜索，再读其中一条结果的正文。", + subscribes: [], + capabilities: ["web.search", "web.fetch"], + pages: ["设置 → 智能体 → 网页搜索"], + slots: [], + hooks: [], + }, + { + id: "mcp", + name: "MCP 服务", + summary: "把能力以 MCP 工具对外提供，外部客户端用 API Key 接入，授权与智能体相同。", + usage: "在设置 → 智能体 → 工具与 MCP 启用并选允许的 API Key；智能体自己不需要经 MCP 调用。", + subscribes: [], + capabilities: [], + pages: ["设置 → 智能体 → 工具与 MCP"], + slots: [], + hooks: [], + }, + { + id: "schedules", + name: "定时任务与巡检", + summary: "各模块的周期任务统一调度；内置智能体巡检模板（默认关闭）。", + usage: "在日志 → 定时任务里查看、启停；智能体只能起草未启用的巡检，由用户检查后启用。", + subscribes: [], + capabilities: ["schedule.draft_check"], + pages: ["日志 → 定时任务"], + slots: [], + hooks: [], + }, +] as const satisfies readonly FeatureManifest[]; + +export type FeatureId = (typeof FEATURES)[number]["id"]; + +export function featureManifest(id: FeatureId): FeatureManifest { + return FEATURES.find((f) => f.id === id)!; +} diff --git a/packages/contracts/src/hooks.ts b/packages/contracts/src/hooks.ts new file mode 100644 index 00000000..c4ebc160 --- /dev/null +++ b/packages/contracts/src/hooks.ts @@ -0,0 +1,44 @@ +// 链式钩子挂点(ADR-006,参考 MoviePilot ChainEventType):同步、按注册顺序执行、可改写或拦截。 +// 这里只登记挂点名称与语义(给功能清单、智能体提示与文档用);入参 / 返回值类型在服务端 +// apps/server/src/hooks/points.ts,因为它们用到 domain 的解析结果等类型。 + +/** + * - rewrite:每个钩子拿到上一个钩子的结果,返回新值或 undefined(不改); + * - intercept:任一钩子返回拒绝原因即停止,后面的钩子不再执行。 + */ +export type HookKind = "rewrite" | "intercept"; + +export const HOOK_POINTS = { + "name.recognize": { + kind: "rewrite", + label: "名称识别", + when: "资源标题解析之后、按名称召回作品之前", + effect: "改写解析结果（标题、年份、季集、外部 ID）", + }, + "organize.rename": { + kind: "rewrite", + label: "整理改名", + when: "按命名模板生成目标路径之后", + effect: "改写主视频的目标文件名（同主干的字幕等附属文件跟着改）", + }, + "organize.intercept": { + kind: "intercept", + label: "整理拦截", + when: "整理计划生成后、写整理记录之前", + effect: "返回原因即拒绝本次整理（记录为失败，原因 intercepted:<原因>）", + }, + "resource.selection": { + kind: "rewrite", + label: "资源选择", + when: "订阅 / 收录版本重评时候选排好序之后、逐条评估之前", + effect: "调整候选顺序或过滤掉一部分（不能加入新候选）", + }, + "subscription.completion.check": { + kind: "intercept", + label: "订阅完成判定", + when: "订阅满足完成条件、标记完成之前", + effect: "追加完成条件：返回原因即本次不完成，订阅继续追踪", + }, +} as const satisfies Record; + +export type HookPoint = keyof typeof HOOK_POINTS; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 23409032..b0733398 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -14,6 +14,8 @@ export * from "./download-board.ts"; export * from "./episode-mapping.ts"; export * from "./events.ts"; export * from "./execution.ts"; +export * from "./features.ts"; +export * from "./hooks.ts"; export * from "./identity.ts"; export * from "./instances.ts"; export * from "./library-versions.ts"; diff --git a/packages/tasks/src/outbox-relay.ts b/packages/tasks/src/outbox-relay.ts index 1569e22f..c21676cd 100644 --- a/packages/tasks/src/outbox-relay.ts +++ b/packages/tasks/src/outbox-relay.ts @@ -21,8 +21,11 @@ export interface OutboxRoute { priority?: number; } -/** 返回 null 表示该事件无需建任务,直接标记已投递 */ -export type OutboxRouteFn = (event: OutboxEvent) => OutboxRoute | null; +/** + * 返回 null 表示该事件无需建任务,直接标记已投递。 + * 返回数组 = 广播给多个订阅方(ADR-006):第一个任务用事件的幂等键,其余用 `<键>#<任务名>`,重投时由引擎幂等去重。 + */ +export type OutboxRouteFn = (event: OutboxEvent) => OutboxRoute | readonly OutboxRoute[] | null; export interface OutboxRelayOptions { db: AnyDb; @@ -82,11 +85,13 @@ export class OutboxRelay { for (const ev of rows.rows) { try { const routeFn = this.opts.routes[ev.type]; - const route = routeFn ? routeFn(ev) : null; - if (route) { - const payloadTrace = (ev.payload as { traceId?: unknown } | null)?.traceId; + const routed = routeFn ? routeFn(ev) : null; + const routes = (routed === null ? [] : Array.isArray(routed) ? routed : [routed]) as readonly OutboxRoute[]; + const baseKey = ev.idempotency_key ?? OutboxRelay.idempotencyKey(ev.id); + const payloadTrace = (ev.payload as { traceId?: unknown } | null)?.traceId; + for (const [i, route] of routes.entries()) { await this.opts.engine.enqueue(route.task, route.input, { - idempotencyKey: ev.idempotency_key ?? OutboxRelay.idempotencyKey(ev.id), + idempotencyKey: i === 0 ? baseKey : `${baseKey}#${route.task}`, traceId: route.traceId ?? (typeof payloadTrace === "string" ? payloadTrace : randomUUID()), ...(route.priority === undefined ? {} : { priority: route.priority }), }); diff --git a/packages/tasks/test/outbox-broadcast.test.ts b/packages/tasks/test/outbox-broadcast.test.ts new file mode 100644 index 00000000..6b2aa697 --- /dev/null +++ b/packages/tasks/test/outbox-broadcast.test.ts @@ -0,0 +1,88 @@ +import { sql } from "kysely"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { type EnqueueOptions, OutboxRelay, type TaskEngine } from "../src/index.ts"; +import { + connectDb, + dropSchemas, + migrateBizSchema, + pgAvailable, + resetSchemas, + type TestDb, + testSchemas, +} from "./helpers.ts"; + +// 生命周期广播(ADR-006):一条 outbox 事件路由给多个订阅方任务。第一个任务用事件的幂等键, +// 其余用 <键>#<任务名>;失败重投时已入队的由引擎幂等键去重,没入队的补上。 + +const schemas = testSchemas("outbox_bc"); +const hasPg = await pgAvailable(); + +describe.skipIf(!hasPg)("OutboxRelay 广播给多个订阅方", () => { + let db: TestDb; + + beforeAll(async () => { + await resetSchemas(schemas); + db = connectDb(schemas.biz); + await migrateBizSchema(db, schemas.biz); + }, 30_000); + afterAll(async () => { + await db?.destroy(); + await dropSchemas(schemas); + }, 30_000); + + it("每个订阅方各入队一次,幂等键按订阅方区分;中途失败重投时同键去重", async () => { + await sql`INSERT INTO outbox_event (type, payload, idempotency_key) + VALUES ('thing.done', '{"id":"t1"}'::jsonb, 'thing.done/t1')`.execute(db); + const keys = new Map(); + let failB = true; + const engine: Pick = { + enqueue: async (task: string, _input: unknown, opts: EnqueueOptions) => { + if (task === "b" && failB) { + failB = false; + throw new Error("引擎暂时不可用"); + } + keys.set(opts.idempotencyKey, (keys.get(opts.idempotencyKey) ?? 0) + 1); + return { runId: opts.idempotencyKey, deduplicated: false }; + }, + }; + const relay = new OutboxRelay({ + db, + engine, + routes: { + "thing.done": (ev) => [ + { task: "a", input: ev.payload }, + { task: "b", input: ev.payload }, + ], + }, + }); + const first = await relay.pollOnce(); + expect(first).toEqual({ published: 0, failed: 1 }); + // 退避期过了再投:a 用同一把键再入一次(真实引擎按幂等键返回已有 run),b 补上 + await sql`UPDATE outbox_event SET next_attempt_at = now()`.execute(db); + expect(await relay.pollOnce()).toEqual({ published: 1, failed: 0 }); + expect([...keys.keys()].sort()).toEqual(["thing.done/t1", "thing.done/t1#b"]); + expect(keys.get("thing.done/t1")).toBe(2); + expect(keys.get("thing.done/t1#b")).toBe(1); + const row = await sql<{ status: string }>`SELECT status FROM outbox_event`.execute(db); + expect(row.rows).toEqual([{ status: "dispatched" }]); + }); + + it("单个订阅方保持原来的键;返回空数组 / null 直接标记已投递", async () => { + await sql`DELETE FROM outbox_event`.execute(db); + await sql`INSERT INTO outbox_event (type, payload, idempotency_key) VALUES + ('one.sub', '{}'::jsonb, 'k-one'), ('no.sub', '{}'::jsonb, 'k-none')`.execute(db); + const keys: string[] = []; + const relay = new OutboxRelay({ + db, + engine: { + enqueue: async (_t: string, _i: unknown, o: EnqueueOptions) => { + keys.push(o.idempotencyKey); + return { runId: "x", deduplicated: false }; + }, + }, + routes: { "one.sub": () => [{ task: "a", input: {} }], "no.sub": () => [] }, + }); + expect(await relay.pollOnce()).toEqual({ published: 2, failed: 0 }); + expect(keys).toEqual(["k-one"]); + }); +});