From 9858b8494d5315cea28d3fb34fc10f019bc123cc Mon Sep 17 00:00:00 2001 From: root Date: Sun, 27 Sep 2026 19:30:55 +0800 Subject: [PATCH] feat(server): ServerModule skeleton with shared ServerContext, worker runs engine + outbox relay (B6) modules/{auth-settings,sources-identity,acquisition-organize,agent} placeholders listed in SERVER_MODULES; ServerModule registers routes (under /api/v1, authenticated), tasks, outbox routes, capabilities and background loops. bootstrap() builds ServerContext (db, config, logger, events, engine, outbox writer, secrets placeholder, settings reader, capability registry), fills unregistered tasks with stubs and rejects duplicate task or outbox ownership. worker starts the OpenWorkflow worker, OutboxRelay and module start(). --role migrate also migrates the engine schema; api/worker connect without migrating. tasks: TaskRegistry.has, migrateEngineSchema. Co-Authored-By: Claude Opus 5.5 --- apps/server/README.md | 60 +++++++ apps/server/src/context.ts | 167 ++++++++++++++++++ apps/server/src/http/app.ts | 37 ++-- apps/server/src/http/sse.ts | 15 +- apps/server/src/main.ts | 27 ++- .../src/modules/acquisition-organize/index.ts | 13 ++ .../src/modules/agent/capability-registry.ts | 13 ++ apps/server/src/modules/agent/index.ts | 11 ++ .../server/src/modules/auth-settings/index.ts | 12 ++ .../src/modules/auth-settings/provided.ts | 33 ++++ apps/server/src/modules/index.ts | 18 ++ .../src/modules/sources-identity/index.ts | 12 ++ apps/server/src/modules/types.ts | 94 ++++++++++ apps/server/src/worker/index.ts | 58 +++++- apps/server/test/bootstrap.test.ts | 138 +++++++++++++++ apps/server/test/mapping-compat.test.ts | 19 ++ apps/server/test/modules.test.ts | 98 ++++++++++ docs/runbook/dev.md | 4 +- packages/tasks/README.md | 3 +- packages/tasks/src/index.ts | 1 + packages/tasks/src/openworkflow-engine.ts | 11 +- packages/tasks/src/registry.ts | 5 + packages/tasks/test/registry.test.ts | 2 + 23 files changed, 829 insertions(+), 22 deletions(-) create mode 100644 apps/server/src/context.ts create mode 100644 apps/server/src/modules/acquisition-organize/index.ts create mode 100644 apps/server/src/modules/agent/capability-registry.ts create mode 100644 apps/server/src/modules/agent/index.ts create mode 100644 apps/server/src/modules/auth-settings/index.ts create mode 100644 apps/server/src/modules/auth-settings/provided.ts create mode 100644 apps/server/src/modules/index.ts create mode 100644 apps/server/src/modules/sources-identity/index.ts create mode 100644 apps/server/src/modules/types.ts create mode 100644 apps/server/test/bootstrap.test.ts create mode 100644 apps/server/test/mapping-compat.test.ts create mode 100644 apps/server/test/modules.test.ts diff --git a/apps/server/README.md b/apps/server/README.md index 75620324..dbb6d3b5 100644 --- a/apps/server/README.md +++ b/apps/server/README.md @@ -1,3 +1,63 @@ # @umt/server 服务进程(Deno):Hono HTTP API(前缀 `/api/v1`,OpenAPI 见 `/api/openapi.json`、`/api/docs`)、SSE 事件流与 worker 入口。`--role api|worker|all` 决定本进程承担哪部分,默认 `all` 单进程运行。业务规则不写在这里,只做编排、鉴权、审计与协议适配。 + +## 模块骨架(W2) + +``` +src/ + main.ts 进程入口:按 --role 装配 + context.ts bootstrap():ServerContext、任务注册表(模块 + 空壳兜底)、outbox 路由合并、连接引擎 + http/app.ts createApp():系统路由 + 逐个挂 SERVER_MODULES 的 registerRoutes + worker/index.ts startWorker():引擎 worker + OutboxRelay + 各模块 start() + modules/ + types.ts ServerModule / ServerContext / OutboxWriter / SecretStore / SettingsReader(共享,不随子任务改) + index.ts SERVER_MODULES 清单(四个模块已登记,子任务不需要再改) + auth-settings/ W2-A(provided.ts 提供 secrets / settings 实现) + sources-identity/ W2-B + acquisition-organize/ W2-C + agent/ W2-D(capability-registry.ts 负责 authorize / audit 注入) +``` + +**分工规则:各 W2 子任务只改自己的 `modules//` 目录,以及 `packages/contracts`、`packages/db` 中本域的文件**(新迁移按编号顺延,开工前在群里认领编号)。`modules/types.ts`、`modules/index.ts`、`context.ts`、`http/app.ts`、`worker/index.ts`、`main.ts` 是共享装配,确需改动时单独提交并通知其他子任务;`http/auth.ts`、`http/sse.ts` 归 W2-A。跨模块只通过 DB 表 + `outbox_event`(事件类型在 contracts/events.ts)+ contracts 类型耦合,不互相 import 对方模块。 + +一个模块长这样(`modules//index.ts` 导出,目录内自由拆 `routes.ts` / `service.ts` / `handlers/*.ts`): + +```ts +import { createRoute } from "@hono/zod-openapi"; +import { capabilitySpec, EventType } from "@umt/contracts"; +import { defineTask } from "@umt/tasks"; +import { requireScope } from "../../http/auth.ts"; +import type { ServerModule } from "../types.ts"; + +export const sourcesIdentityModule: ServerModule = { + name: "sources-identity", + // 路径相对 /api/v1,已过 API Key 鉴权;每个路由自己声明 scope + registerRoutes(app, ctx) { + app.use("/sites/*", requireScope("read")); + app.openapi(createRoute({ method: "get", path: "/sites", /* … */ }), async (c) => c.json(/* … */)); + }, + // 同名任务只能由一个模块注册;未注册的 §3.1-8 任务由 @umt/tasks 空壳兜底 + registerTasks(registry, ctx) { + registry.register(defineTask({ name: "resolveObservation", version: 1, /* … */ handler: async (t) => { + await ctx.db.transaction().execute(async (trx) => { + // …写业务表,同事务发事件(payload 按 contracts 校验) + await ctx.outbox.emit(trx, EventType.releaseResolved, payload, { idempotencyKey: `release.resolved:${id}` }); + }); + } })); + }, + // 消费别的模块发出的事件:type → 任务;同一 type 只能一个模块认领 + outboxRoutes: () => ({ "release.resolved": (ev) => ({ task: "evaluateAcquisition", input: { sourceReleaseId: (ev.payload as { sourceReleaseId: string }).sourceReleaseId } }) }), + // 能力:规格一律取 contracts,处理器调本模块服务 + registerCapabilities(registry, ctx) { + registry.register(capabilitySpec("release.search"), async (input, c) => ({ items: [], coverage: [] })); + }, + // worker 角色的后台循环(qB 轮询、Feed 调度…);引擎与中继已启动 + async start(ctx) {}, + async stop() {}, +}; +``` + +`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` 连接。 diff --git a/apps/server/src/context.ts b/apps/server/src/context.ts new file mode 100644 index 00000000..926cd03a --- /dev/null +++ b/apps/server/src/context.ts @@ -0,0 +1,167 @@ +import { type EventType, parseEventPayload } from "@umt/contracts"; +import { type DB, type Kysely, sql } from "@umt/db"; +import { + createDefaultRegistry, + type LogSink, + OpenWorkflowEngine, + type OutboxRelay, + type OutboxRouteFn, + type TaskEngine, + TaskRegistry, +} from "@umt/tasks"; +import type { Logger } from "pino"; +import type { Config } from "./config.ts"; +import type { EventBusPort } from "./http/sse.ts"; +import { createCapabilityRegistry } from "./modules/agent/capability-registry.ts"; +import { createSecretStore, createSettingsReader } from "./modules/auth-settings/provided.ts"; +import type { OutboxPort, ServerContext, ServerModule, ServerRole } from "./modules/types.ts"; + +// 进程装配:ServerContext、任务注册表、outbox 路由。api / worker / all 三种角色共用。 + +/** 引擎在任务注册表建好后才能连接;先给模块一个会转发的句柄 */ +export class DeferredTaskEngine implements TaskEngine { + private inner: TaskEngine | null = null; + + attach(engine: TaskEngine) { + this.inner = engine; + } + + get ready(): boolean { + return this.inner !== null; + } + + private get e(): TaskEngine { + if (!this.inner) throw new Error("任务引擎尚未就绪(模块不要在 register* 阶段调用引擎)"); + return this.inner; + } + + enqueue: TaskEngine["enqueue"] = (...a) => this.e.enqueue(...a); + signal: TaskEngine["signal"] = (...a) => this.e.signal(...a); + cancel: TaskEngine["cancel"] = (...a) => this.e.cancel(...a); + get: TaskEngine["get"] = (...a) => this.e.get(...a); + list: TaskEngine["list"] = (...a) => this.e.list(...a); + startWorker: TaskEngine["startWorker"] = (...a) => this.e.startWorker(...a); + stop: TaskEngine["stop"] = async () => { + await this.inner?.stop(); + }; +} + +export function createOutboxPort(): OutboxPort & { setRelay(r: OutboxRelay): void } { + let relay: OutboxRelay | null = null; + return { + get relay() { + return relay; + }, + setRelay(r) { + relay = r; + }, + async emit(executor, type, payload, opts = {}) { + if (parseEventPayload(type as EventType, payload) === null) { + throw new Error(`outbox 事件 ${type} 的 payload 不符合 contracts 定义`); + } + await sql`insert into outbox_event (type, payload, idempotency_key) + values (${type}, ${JSON.stringify(payload)}::jsonb, ${opts.idempotencyKey ?? null}) + on conflict (idempotency_key) where idempotency_key is not null do nothing`.execute(executor); + }, + }; +} + +/** 把 pino logger 接到 @umt/tasks 的 LogSink(tasks 自己已脱敏,这里再经 pino 的脱敏) */ +export function tasksLogSink(logger: Logger): LogSink { + // r.fields 里已带 component(tasks.engine / tasks.outbox) + return (r) => logger[r.level](r.fields, r.msg); +} + +/** + * 任务注册表:各模块注册自己的任务;没人注册的 §3.1-8 任务用 @umt/tasks 的空壳兜底 + * (outbox 路由到它们时不至于报"未知任务")。同名任务被两个模块注册会在这里报错。 + */ +export function buildTaskRegistry(modules: readonly ServerModule[], ctx: ServerContext): TaskRegistry { + const registry = new TaskRegistry(); + const owner = new Map(); + for (const m of modules) { + const own = new TaskRegistry(); + m.registerTasks(own, ctx); + for (const name of own.names()) { + const prev = owner.get(name); + if (prev) throw new Error(`任务 ${name} 被模块 ${prev} 与 ${m.name} 重复注册`); + owner.set(name, m.name); + } + registry.register(...own.all()); + } + for (const stub of createDefaultRegistry().all()) if (!registry.has(stub.name)) registry.register(stub); + return registry; +} + +/** 合并各模块的 outbox 路由;同一事件 type 只能由一个模块认领 */ +export function collectOutboxRoutes( + modules: readonly ServerModule[], + ctx: ServerContext, +): Record { + const routes: Record = {}; + const owner = 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; + } + } + return routes; +} + +export interface BootstrapOptions { + role: ServerRole; + config: Config; + logger: Logger; + db: Kysely; + events: EventBusPort; + modules: readonly ServerModule[]; + /** 连接引擎时是否跑引擎迁移;默认只有 all 角色跑(api/worker 分进程时由 --role migrate 预先迁移) */ + engineMigrations?: boolean; + /** 仅测试覆盖:引擎表 schema(默认 openworkflow) */ + engineSchema?: string; + /** 仅测试覆盖:不连接真实引擎 */ + engine?: TaskEngine; +} + +export interface Bootstrapped { + ctx: ServerContext; + tasks: TaskRegistry; + outbox: ReturnType; + engine: DeferredTaskEngine; +} + +/** 建 ServerContext → 模块注册任务与能力 → 连接引擎。路由与 worker 由调用方按角色启动 */ +export async function bootstrap(opts: BootstrapOptions): Promise { + const { config, logger, db } = opts; + const engine = new DeferredTaskEngine(); + const outbox = createOutboxPort(); + const ctx: ServerContext = { + role: opts.role, + db, + config, + logger, + events: opts.events, + engine, + outbox, + secrets: createSecretStore({ db, masterKey: config.SECRET_MASTER_KEY }), + settings: createSettingsReader({ db, logger }), + capabilities: createCapabilityRegistry({ logger }), + }; + const tasks = buildTaskRegistry(opts.modules, ctx); + for (const m of opts.modules) m.registerCapabilities?.(ctx.capabilities, ctx); + engine.attach( + opts.engine ?? + (await OpenWorkflowEngine.connect({ + databaseUrl: config.DATABASE_URL, + db, + registry: tasks, + logSink: tasksLogSink(logger), + runMigrations: opts.engineMigrations ?? opts.role === "all", + ...(opts.engineSchema ? { schema: opts.engineSchema } : {}), + })), + ); + return { ctx, tasks, outbox, engine }; +} diff --git a/apps/server/src/http/app.ts b/apps/server/src/http/app.ts index 7f7f8ab7..57d7fab3 100644 --- a/apps/server/src/http/app.ts +++ b/apps/server/src/http/app.ts @@ -8,17 +8,18 @@ import type { Logger } from "pino"; import type { Config } from "../config.ts"; import { isDev } from "../config.ts"; import { redactString } from "../logger.ts"; +import type { ServerContext, ServerModule } from "../modules/types.ts"; import { type ApiKeyStore, type AuthEnv, apiKeyAuth, kyselyApiKeyStore, requireScope } from "./auth.ts"; import { ApiException, errorBody } from "./errors.ts"; import { healthRoutes } from "./routes/health.ts"; import { mountOpenApi } from "./routes/openapi.ts"; -import { type EventBus, sseRoutes } from "./sse.ts"; +import { type EventBusPort, sseRoutes } from "./sse.ts"; import { mountWebRoot } from "./static.ts"; export interface AppDeps { config: Pick; logger: Logger; - bus: EventBus; + bus: EventBusPort; version: string; role: "api" | "worker" | "all"; startedAt?: number; @@ -27,6 +28,9 @@ export interface AppDeps { apiKeyStore?: ApiKeyStore; /** 单容器模式:托管 web 静态产物的目录 */ webRoot?: string; + /** 业务模块(modules/index.ts 的 SERVER_MODULES);给了 ctx 才挂载 */ + modules?: readonly ServerModule[]; + ctx?: ServerContext; } export const API_PREFIX = "/api/v1"; @@ -35,15 +39,7 @@ export function createApp(deps: AppDeps) { const { config, logger, bus, db } = deps; const dev = isDev(config); - const app = new OpenAPIHono({ - // zod 校验失败统一成 validation_failed 信封 - defaultHook: (result, c) => { - if (!result.success) { - const msg = result.error.issues.map((i) => `${i.path.join(".") || "(root)"}: ${i.message}`).join("; "); - return c.json(errorBody("validation_failed", msg), 422); - } - }, - }); + const app = newApiApp(); app.use("*", requestId()); // 访问日志:路径中的 passkey/token 参数脱敏 @@ -112,8 +108,27 @@ export function createApp(deps: AppDeps) { return c.json({ ok: true, data: { id: p.id, name: p.name, kind: p.kind, scopes: p.scopes } }); }); + // ── 业务模块:各自挂在 /api/v1 下(已鉴权),按清单顺序注册 ── + if (deps.ctx && deps.modules?.length) { + const api = newApiApp(); + for (const m of deps.modules) m.registerRoutes(api, deps.ctx); + app.route(API_PREFIX, api); + } + if (deps.webRoot) mountWebRoot(app, deps.webRoot); return app; } export type App = ReturnType; + +/** 统一的 OpenAPIHono 实例:zod 校验失败统一成 validation_failed 信封 */ +function newApiApp() { + return new OpenAPIHono({ + defaultHook: (result, c) => { + if (!result.success) { + const msg = result.error.issues.map((i) => `${i.path.join(".") || "(root)"}: ${i.message}`).join("; "); + return c.json(errorBody("validation_failed", msg), 422); + } + }, + }); +} diff --git a/apps/server/src/http/sse.ts b/apps/server/src/http/sse.ts index bffed1d1..4919ecf4 100644 --- a/apps/server/src/http/sse.ts +++ b/apps/server/src/http/sse.ts @@ -5,11 +5,22 @@ import { ApiException } from "./errors.ts"; type Listener = (e: EventEnvelope) => void; +/** + * SSE 总线端口。模块与 SSE 路由只依赖这个接口;W2-A 把实现换成 PG NOTIFY + 持久化 seq 时接口不变。 + */ +export interface EventBusPort { + readonly latestSeq: number; + publish

(type: string, payload: P): EventEnvelope

; + /** 取 seq > after 的事件;gap=true 表示起点已不可补读,客户端需重新拉快照 */ + since(after: number): { events: EventEnvelope[]; gap: boolean }; + subscribe(l: Listener): () => void; +} + /** * 进程内事件总线:序号自增,保留最近 capacity 条供断线补读。 * 多进程部署时需换成 PG LISTEN/NOTIFY 或 outbox 投影(W2)。 */ -export class EventBus { +export class EventBus implements EventBusPort { private seq = 0; private readonly buffer: EventEnvelope[] = []; private readonly listeners = new Set(); @@ -42,7 +53,7 @@ export class EventBus { } export interface SseOptions { - bus: EventBus; + bus: EventBusPort; keepaliveMs?: number; } diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index b1f421a3..c1c753a6 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -1,10 +1,13 @@ import { parseArgs } from "node:util"; import { createDb, migrateToLatest } from "@umt/db"; +import { migrateEngineSchema } from "@umt/tasks"; import denoJson from "../deno.json" with { type: "json" }; import { loadConfig } from "./config.ts"; +import { bootstrap } from "./context.ts"; import { createApp } from "./http/app.ts"; import { EventBus } from "./http/sse.ts"; import { createLogger } from "./logger.ts"; +import { SERVER_MODULES } from "./modules/index.ts"; import { runPreflight } from "./preflight.ts"; import { startWorker } from "./worker/index.ts"; @@ -52,6 +55,9 @@ if (role === "migrate" || role === "preflight") { const { results } = await migrateToLatest(db); for (const r of results ?? []) logger.info({ migration: r.migrationName, status: r.status }, "migration"); logger.info({ applied: results?.length ?? 0 }, "迁移完成"); + // 任务引擎自己的表(openworkflow schema);api/worker 进程启动时不再迁移 + await migrateEngineSchema(config.DATABASE_URL); + logger.info("任务引擎迁移完成"); } else { const problems = await runPreflight({ config, db }); for (const p of problems) logger.error({ check: p.check }, p.message); @@ -70,6 +76,17 @@ if (role === "migrate" || role === "preflight") { const bus = new EventBus(); const stops: Array<() => Promise> = []; +// 装配 ServerContext:模块注册任务/能力,连接任务引擎(api 角色也要:入队、信号、取消) +let booted: Awaited>; +try { + booted = await bootstrap({ role, config, logger, db, events: bus, modules: SERVER_MODULES }); +} catch (err) { + logger.error({ err }, "启动装配失败"); + await db.destroy(); + Deno.exit(1); +} +const { ctx, outbox, engine } = booted; + if (role === "api" || role === "all") { const app = createApp({ config, @@ -79,6 +96,8 @@ if (role === "api" || role === "all") { role, version: denoJson.version, ...(config.WEB_ROOT ? { webRoot: config.WEB_ROOT } : {}), + modules: SERVER_MODULES, + ctx, }); const ac = new AbortController(); const server = Deno.serve( @@ -101,7 +120,12 @@ if (role === "api" || role === "all") { }); } if (role === "worker" || role === "all") { - const worker = startWorker({ logger, heartbeatMs: config.WORKER_HEARTBEAT_MS, bus }); + const worker = await startWorker({ + logger, + heartbeatMs: config.WORKER_HEARTBEAT_MS, + bus, + runtime: { ctx, modules: SERVER_MODULES, setRelay: outbox.setRelay }, + }); stops.push(worker.stop); } @@ -113,6 +137,7 @@ async function shutdown(signal: Deno.Signal) { // 兜底:10 秒内没退干净就强退,避免容器 stop 超时被 SIGKILL const force = setTimeout(() => Deno.exit(1), 10_000); await Promise.allSettled(stops.map((s) => s())); + await engine.stop().catch((err) => logger.error({ err }, "任务引擎关闭失败")); await db.destroy(); clearTimeout(force); Deno.exit(0); diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts new file mode 100644 index 00000000..289d8cf4 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -0,0 +1,13 @@ +import type { ServerModule } from "../types.ts"; + +/** + * W2-C 规则/下载/入库/整理记录:subscription、acquisition_slot、policy_revision、preference_tag; + * evaluateAcquisition / submitDownload / verifyAndImport / cleanupDownloadCache / reconcileLibrary; + * qB 同步与缓存阈值、Copy/Move、path_binding、organization_record 与公开增量 API、媒体库与首页聚合。 + * 消费 release.resolved(outboxRoutes);删种带数据一律经 adapters 的 assertSafeToDeleteData。 + */ +export const acquisitionOrganizeModule: ServerModule = { + name: "acquisition-organize", + registerRoutes() {}, + registerTasks() {}, +}; diff --git a/apps/server/src/modules/agent/capability-registry.ts b/apps/server/src/modules/agent/capability-registry.ts new file mode 100644 index 00000000..df15c0cf --- /dev/null +++ b/apps/server/src/modules/agent/capability-registry.ts @@ -0,0 +1,13 @@ +import { CapabilityRegistry } from "@umt/agent"; +import type { Logger } from "pino"; + +/** + * 能力注册中心工厂(W2-D 负责接 authorize:automation_grant / approval + 自动化总开关;audit:写 audit_event)。 + * 当前未注入 authorize:read/preview 可调用,propose/approve/execute 一律拒绝(安全默认)。 + */ +export function createCapabilityRegistry(deps: { logger: Logger }): CapabilityRegistry { + const log = deps.logger.child({ component: "capabilities" }); + return new CapabilityRegistry({ + audit: (r) => log.info({ capability: r.capabilityId, outcome: r.outcome, actor: r.actorId }, "capability.invoke"), + }); +} diff --git a/apps/server/src/modules/agent/index.ts b/apps/server/src/modules/agent/index.ts new file mode 100644 index 00000000..64bfbc67 --- /dev/null +++ b/apps/server/src/modules/agent/index.ts @@ -0,0 +1,11 @@ +import type { ServerModule } from "../types.ts"; + +/** + * W2-D Agent 装配:14 个能力的 authorize/audit 注入(./capability-registry.ts)、Agent 会话 API + SSE、 + * agent_run_event 落库、resolveDecision 处理器、后台触发的有界 agent 任务。 + */ +export const agentModule: ServerModule = { + name: "agent", + registerRoutes() {}, + registerTasks() {}, +}; diff --git a/apps/server/src/modules/auth-settings/index.ts b/apps/server/src/modules/auth-settings/index.ts new file mode 100644 index 00000000..01abbeaf --- /dev/null +++ b/apps/server/src/modules/auth-settings/index.ts @@ -0,0 +1,12 @@ +import type { ServerModule } from "../types.ts"; + +/** + * W2-A 认证/设置/实例/引擎面板:管理员登录门禁、settings、api_key、secret 加密服务、 + * downloader/media_server 实例与 library_root/path_mapping、审计查询、SSE 总线改 PG 通知、引擎面板只读 API、自动化总开关。 + * secrets / settings 的实现放在 ./provided.ts(context 从这里取),SSE 总线在 http/sse.ts,鉴权在 http/auth.ts。 + */ +export const authSettingsModule: ServerModule = { + name: "auth-settings", + registerRoutes() {}, + registerTasks() {}, +}; diff --git a/apps/server/src/modules/auth-settings/provided.ts b/apps/server/src/modules/auth-settings/provided.ts new file mode 100644 index 00000000..b029e563 --- /dev/null +++ b/apps/server/src/modules/auth-settings/provided.ts @@ -0,0 +1,33 @@ +import { DEFAULT_SETTINGS, GlobalSettings, type SettingsKey } from "@umt/contracts"; +import type { DB, Kysely } from "@umt/db"; +import type { Logger } from "pino"; +import type { SecretStore, SettingsReader } from "../types.ts"; + +// auth-settings 模块向 ServerContext 提供的基础服务。W2-A 在这里替换实现,其它模块只用 ctx 上的接口。 + +/** 占位:W2-A 实现 AES-GCM 加密存取(secret 表,SECRET_MASTER_KEY) */ +export function createSecretStore(_deps: { db: Kysely; masterKey: string }): SecretStore { + const notYet = async (): Promise => { + throw new Error("SecretStore 尚未实现(W2-A)"); + }; + return { resolve: notYet, create: notYet, replace: notYet, remove: notYet }; +} + +/** settings 表(key → jsonb)按分组合并缺省值;每次读库,不缓存(设置变更立即生效) */ +export function createSettingsReader(deps: { db: Kysely; logger: Logger }): SettingsReader { + const all = async (): Promise => { + const rows = await deps.db.selectFrom("settings").select(["key", "value"]).execute(); + const stored = new Map(rows.map((r) => [r.key, r.value])); + const out: Record = {}; + for (const key of Object.keys(DEFAULT_SETTINGS) as SettingsKey[]) { + const def = DEFAULT_SETTINGS[key]; + const v = stored.get(key); + const merged = v && typeof v === "object" && !Array.isArray(v) ? { ...def, ...(v as object) } : def; + const parsed = GlobalSettings.shape[key].safeParse(merged); + if (!parsed.success) deps.logger.warn({ key }, "settings 值不合法,回退缺省"); + out[key] = parsed.success ? parsed.data : def; + } + return out as GlobalSettings; + }; + return { all, get: async (key) => (await all())[key] }; +} diff --git a/apps/server/src/modules/index.ts b/apps/server/src/modules/index.ts new file mode 100644 index 00000000..c07e33b3 --- /dev/null +++ b/apps/server/src/modules/index.ts @@ -0,0 +1,18 @@ +import { acquisitionOrganizeModule } from "./acquisition-organize/index.ts"; +import { agentModule } from "./agent/index.ts"; +import { authSettingsModule } from "./auth-settings/index.ts"; +import { sourcesIdentityModule } from "./sources-identity/index.ts"; +import type { ServerModule } from "./types.ts"; + +export type * from "./types.ts"; + +/** + * 模块清单:http/app.ts(路由)与 worker(任务/中继/后台循环)都从这里取。 + * 顺序即注册顺序;auth-settings 在前(其它模块的路由依赖它的登录门禁中间件)。 + */ +export const SERVER_MODULES: readonly ServerModule[] = [ + authSettingsModule, + sourcesIdentityModule, + acquisitionOrganizeModule, + agentModule, +]; diff --git a/apps/server/src/modules/sources-identity/index.ts b/apps/server/src/modules/sources-identity/index.ts new file mode 100644 index 00000000..10b93630 --- /dev/null +++ b/apps/server/src/modules/sources-identity/index.ts @@ -0,0 +1,12 @@ +import type { ServerModule } from "../types.ts"; + +/** + * W2-B 来源/身份/元数据:site、site_account、feed_source;ingestFeed / resolveObservation / refreshMetadata; + * observation/source_release 写入与去重;身份解析、待识别池与 decision;站点工作台 API。 + * 发出 release.resolved、decision.created(经 ctx.outbox.emit)。 + */ +export const sourcesIdentityModule: ServerModule = { + name: "sources-identity", + registerRoutes() {}, + registerTasks() {}, +}; diff --git a/apps/server/src/modules/types.ts b/apps/server/src/modules/types.ts new file mode 100644 index 00000000..9b4358f9 --- /dev/null +++ b/apps/server/src/modules/types.ts @@ -0,0 +1,94 @@ +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 { 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 { AuthEnv } from "../http/auth.ts"; +import type { EventBusPort } from "../http/sse.ts"; + +// 服务端模块骨架(W1 审查 B6;任务书 §5)。 +// 每个 W2 子任务是一个 ServerModule,只改 modules/<自己的目录>/ 与 contracts/db 中本域的文件。 +// 跨模块只通过 DB 表 + outbox_event(事件类型在 contracts/events.ts)+ contracts 类型耦合,不互相 import。 + +export type ServerRole = "api" | "worker" | "all"; + +/** + * 模块路由挂载点:已挂在 /api/v1 下、已经过 API Key 鉴权(c.var.principal 可用)。 + * 路径写相对 /api/v1 的部分(如 "/sites"),每个路由自己加 requireScope(...); + * 用 app.openapi(createRoute(...)) 注册的路由会进 /api/openapi.json。 + */ +export type ModuleApp = OpenAPIHono; + +/** 业务事务内写 outbox_event;中继(worker)再按 type 路由成任务 */ +export interface OutboxWriter { + /** + * payload 按 contracts EVENT_PAYLOADS 校验,不合法直接抛错。 + * 传 idempotencyKey 时同键重复写入会被忽略(outbox_event.idempotency_key 唯一),也作为引擎入队幂等键。 + */ + emit( + executor: Kysely | Transaction, + type: T, + payload: EventPayload, + opts?: { idempotencyKey?: string }, + ): Promise; +} + +export interface OutboxPort extends OutboxWriter { + /** 仅 worker 角色有中继实例;api 角色为 null */ + readonly relay: OutboxRelay | null; +} + +/** + * 凭证存取(secret 表 + SECRET_MASTER_KEY 加密)。W2-A 在 modules/auth-settings/provided.ts 实现, + * 当前占位实现一律抛"未实现"。其它模块只存/传 secret_ref,需要明文时现场 resolve,不缓存、不打日志。 + */ +export interface SecretStore { + resolve(secretRef: string): Promise; + create(input: { purpose: SecretPurpose; plaintext: string; label?: string }): Promise; + replace(secretRef: string, plaintext: string): Promise; + remove(secretRef: string): Promise; +} + +/** 全局设置读取:settings 表按分组合并 DEFAULT_SETTINGS 后校验;表里值非法时回退缺省并记警告 */ +export interface SettingsReader { + get(key: K): Promise; + all(): Promise; +} + +export interface ServerContext { + role: ServerRole; + db: Kysely; + config: Config; + logger: Logger; + /** SSE 事件总线 */ + events: EventBusPort; + /** 任务引擎:api 角色也可 enqueue/signal/cancel;只有 worker 角色执行任务 */ + engine: TaskEngine; + outbox: OutboxPort; + secrets: SecretStore; + settings: SettingsReader; + /** 能力注册中心(Web/Pi/CLI/MCP 共用一条 invoke 路径) */ + capabilities: CapabilityRegistry; +} + +export interface ServerModule { + /** 与目录名一致,用于日志 */ + name: string; + /** api / all 角色:挂 HTTP 路由 */ + registerRoutes(app: ModuleApp, ctx: ServerContext): void; + /** + * 所有角色:注册持久任务处理器(api 角色也要注册,enqueue 时按定义校验输入)。 + * 同名任务只能由一个模块注册;没人注册的 §3.1-8 任务由 @umt/tasks 的空壳兜底。 + */ + registerTasks(registry: TaskRegistry, ctx: ServerContext): void; + /** 所有角色:outbox 事件 type → 任务路由;同一 type 只能由一个模块认领 */ + outboxRoutes?(ctx: ServerContext): Record; + /** 所有角色:用 contracts 的 capabilitySpec(id) 注册能力处理器 */ + registerCapabilities?(registry: CapabilityRegistry, ctx: ServerContext): void; + /** worker / all 角色:启动轮询器、定时器等后台循环(引擎与中继已启动) */ + start?(ctx: ServerContext): Promise | void; + /** 退出时按注册逆序调用 */ + stop?(): Promise | void; +} diff --git a/apps/server/src/worker/index.ts b/apps/server/src/worker/index.ts index ebcfda4c..81f41079 100644 --- a/apps/server/src/worker/index.ts +++ b/apps/server/src/worker/index.ts @@ -1,16 +1,35 @@ import { EventType } from "@umt/contracts"; +import { OutboxRelay } from "@umt/tasks"; import type { Logger } from "pino"; -import type { EventBus } from "../http/sse.ts"; +import { collectOutboxRoutes, tasksLogSink } from "../context.ts"; +import type { EventBusPort } from "../http/sse.ts"; +import type { ServerContext, ServerModule } from "../modules/types.ts"; export interface WorkerDeps { logger: Logger; heartbeatMs: number; /** role=all 时与 API 共享同一总线,心跳可在 SSE 中看到 */ - bus?: EventBus; + bus?: EventBusPort; + /** 业务装配;省略时只跑心跳(测试用) */ + runtime?: { + ctx: ServerContext; + modules: readonly ServerModule[]; + /** 设置 outbox 中继实例到 ctx.outbox.relay */ + setRelay?: (relay: OutboxRelay) => void; + outboxIntervalMs?: number; + concurrency?: number; + }; } -/** worker 占位:只打印心跳。持久任务引擎在 W1(packages/tasks)接入 */ -export function startWorker({ logger, heartbeatMs, bus }: WorkerDeps) { +export interface RunningWorker { + stop(): Promise; +} + +/** + * worker 角色:心跳 + 任务引擎 worker(执行全部已注册任务)+ OutboxRelay(按模块的 outboxRoutes 路由) + * + 各模块的后台循环(start)。stop 逆序停止;引擎连接本身由进程退出时关闭。 + */ +export async function startWorker({ logger, heartbeatMs, bus, runtime }: WorkerDeps): Promise { const log = logger.child({ component: "worker" }); let n = 0; const beat = () => { @@ -18,12 +37,41 @@ export function startWorker({ logger, heartbeatMs, bus }: WorkerDeps) { log.debug({ n }, "worker heartbeat"); bus?.publish(EventType.heartbeat, { role: "worker", n }); }; - log.info({ heartbeatMs }, "worker started"); + + const started: ServerModule[] = []; + let relay: OutboxRelay | null = null; + if (runtime) { + const { ctx, modules } = runtime; + await ctx.engine.startWorker({ ...(runtime.concurrency ? { concurrency: runtime.concurrency } : {}) }); + relay = new OutboxRelay({ + db: ctx.db, + engine: ctx.engine, + routes: collectOutboxRoutes(modules, ctx), + logSink: tasksLogSink(logger), + ...(runtime.outboxIntervalMs ? { intervalMs: runtime.outboxIntervalMs } : {}), + }); + runtime.setRelay?.(relay); + relay.start(); + for (const m of modules) { + await m.start?.(ctx); + started.push(m); + } + } + + log.info({ heartbeatMs, modules: started.map((m) => m.name) }, "worker started"); beat(); const timer = setInterval(beat, heartbeatMs); return { async stop() { clearInterval(timer); + for (const m of started.reverse()) { + try { + await m.stop?.(); + } catch (err) { + log.error({ err, module: m.name }, "模块停止失败"); + } + } + await relay?.stop(); log.info("worker stopped"); }, }; diff --git a/apps/server/test/bootstrap.test.ts b/apps/server/test/bootstrap.test.ts new file mode 100644 index 00000000..5135f468 --- /dev/null +++ b/apps/server/test/bootstrap.test.ts @@ -0,0 +1,138 @@ +import { EventType } from "@umt/contracts"; +import { type DB, migrateToLatest } from "@umt/db"; +import { defineTask } from "@umt/tasks"; +import { Kysely, PostgresDialect, sql } from "kysely"; +import pg from "pg"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { z } from "zod"; +import { ConfigSchema } from "../src/config.ts"; +import { type Bootstrapped, bootstrap } from "../src/context.ts"; +import { EventBus } from "../src/http/sse.ts"; +import type { ServerModule } from "../src/modules/types.ts"; +import { type RunningWorker, startWorker } from "../src/worker/index.ts"; +import { memoryLogger } from "./helpers.ts"; + +// 真实 PG:独立业务 schema + 独立引擎 schema,验证 bootstrap → worker → outbox → 模块任务 的整条装配。 +const url = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt"; +const uniq = `${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`; +const biz = `server_test_${uniq}`; +const ow = `${biz}_ow`; + +async function pgAvailable() { + const c = new pg.Client({ connectionString: url, connectionTimeoutMillis: 2000 }); + try { + await c.connect(); + await c.end(); + return true; + } catch { + return false; + } +} +const hasPg = process.env.UMT_SKIP_DB_TESTS !== "1" && (await pgAvailable()); + +describe.skipIf(!hasPg)("bootstrap + worker(真实 PG)", () => { + let db: Kysely; + let booted: Bootstrapped; + let worker: RunningWorker; + const seen: string[] = []; + + const probe: ServerModule = { + name: "probe", + registerRoutes() {}, + registerTasks(registry, ctx) { + registry.register( + defineTask({ + name: "evaluateAcquisition", + version: 1, + inputSchema: z.object({ sourceReleaseId: z.string().optional() }), + handler: async (t) => { + const s = await ctx.settings.get("cache"); + seen.push(`${t.input.sourceReleaseId}:${s.hardThresholdPercent}`); + return null; + }, + }), + ); + }, + outboxRoutes: () => ({ + [EventType.releaseResolved]: (ev) => ({ + task: "evaluateAcquisition", + input: { sourceReleaseId: (ev.payload as { sourceReleaseId: string }).sourceReleaseId }, + }), + }), + }; + + beforeAll(async () => { + const admin = new pg.Client({ connectionString: url }); + await admin.connect(); + await admin.query(`create schema ${biz}`); + await admin.end(); + db = new Kysely({ + dialect: new PostgresDialect({ + pool: new pg.Pool({ connectionString: url, max: 5, options: `-c search_path=${biz},public` }), + }), + }); + await migrateToLatest(db, { migrationTableSchema: biz }); + await db + .insertInto("settings") + .values({ key: "cache", value: JSON.stringify({ hardThresholdPercent: 95, resumePercent: 90 }) }) + .execute(); + const config = ConfigSchema.parse({ DATABASE_URL: url, SECRET_MASTER_KEY: "test-master-key-0123456789" }); + const { logger } = memoryLogger(); + booted = await bootstrap({ + role: "all", + config, + logger, + db, + events: new EventBus(10), + modules: [probe], + engineSchema: ow, + }); + worker = await startWorker({ + logger, + heartbeatMs: 60_000, + runtime: { ctx: booted.ctx, modules: [probe], setRelay: booted.outbox.setRelay, outboxIntervalMs: 100 }, + }); + }, 60_000); + + afterAll(async () => { + await worker?.stop(); + await booted?.engine.stop(); + await db?.destroy(); + const admin = new pg.Client({ connectionString: url }); + await admin.connect(); + await admin.query(`drop schema if exists ${biz} cascade`); + await admin.query(`drop schema if exists ${ow} cascade`); + await admin.end(); + }, 60_000); + + it("settings 读取合并缺省值", async () => { + const cache = await booted.ctx.settings.get("cache"); + expect(cache).toMatchObject({ hardThresholdPercent: 95, resumePercent: 90, checkIntervalSec: 60 }); + expect((await booted.ctx.settings.all()).automation.enabled).toBe(true); + await expect(booted.ctx.secrets.resolve("x")).rejects.toThrow(/W2-A/); + }); + + it("业务事务内 emit → 中继路由 → worker 执行模块任务;同幂等键只投一次;非法 payload 拒写", async () => { + const payload = { + sourceReleaseId: "11111111-1111-4111-8111-111111111111", + observationId: null, + mediaEntityId: "22222222-2222-4222-8222-222222222222", + contentUnitIds: [], + evidenceLevel: "rule_confirmed" as const, + }; + await db.transaction().execute(async (trx) => { + await booted.ctx.outbox.emit(trx, EventType.releaseResolved, payload, { idempotencyKey: "rr:1" }); + await booted.ctx.outbox.emit(trx, EventType.releaseResolved, payload, { idempotencyKey: "rr:1" }); + }); + await expect( + booted.ctx.outbox.emit(db, EventType.releaseResolved, { ...payload, evidenceLevel: "bogus" } as never), + ).rejects.toThrow(/payload/); + expect(booted.ctx.outbox.relay).not.toBeNull(); + + const deadline = Date.now() + 30_000; + while (seen.length === 0 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 200)); + expect(seen).toEqual([`${payload.sourceReleaseId}:95`]); + const rows = await sql<{ status: string }>`select status from outbox_event`.execute(db); + expect(rows.rows).toEqual([{ status: "dispatched" }]); + }, 40_000); +}); diff --git a/apps/server/test/mapping-compat.test.ts b/apps/server/test/mapping-compat.test.ts new file mode 100644 index 00000000..ab62eb10 --- /dev/null +++ b/apps/server/test/mapping-compat.test.ts @@ -0,0 +1,19 @@ +import type { DownloaderClient, SiteRelease } from "@umt/adapters"; +import type { DomainDownloadStatus, SiteReleaseLike } from "@umt/mapping"; +import { toDomainDownloadStatus } from "@umt/mapping"; +import { describe, expect, it } from "vitest"; + +// @umt/mapping 不依赖 adapters;server 同时依赖两者,在这里做类型层兼容校验。 +type Assert = T; +export type _SiteReleaseCompatible = Assert; +// 删数据接口必须带安全检查输入(B7) +export type _DeleteRequiresSafety = Assert< + { deleteFiles: true } extends Parameters[1] ? false : true +>; + +describe("mapping ↔ adapters 兼容", () => { + it("adapters 的 SiteRelease 可直接交给 mapping", () => { + const status: DomainDownloadStatus = toDomainDownloadStatus("paused", "threshold"); + expect(status).toBe("paused_by_threshold"); + }); +}); diff --git a/apps/server/test/modules.test.ts b/apps/server/test/modules.test.ts new file mode 100644 index 00000000..bffa739b --- /dev/null +++ b/apps/server/test/modules.test.ts @@ -0,0 +1,98 @@ +import { createRoute, z } from "@hono/zod-openapi"; +import { apiResponse, CAPABILITY_IDS, capabilitySpec } from "@umt/contracts"; +import { defineTask, TASK_NAMES } from "@umt/tasks"; +import { describe, expect, it } from "vitest"; +import { buildTaskRegistry, collectOutboxRoutes, DeferredTaskEngine } from "../src/context.ts"; +import { createApp } from "../src/http/app.ts"; +import { requireScope } from "../src/http/auth.ts"; +import { EventBus } from "../src/http/sse.ts"; +import { SERVER_MODULES } from "../src/modules/index.ts"; +import type { ServerContext, ServerModule } from "../src/modules/types.ts"; +import { DEV_KEY, json, memoryLogger } from "./helpers.ts"; + +// register* 阶段模块只保存 ctx 引用,这里不需要真实依赖 +const fakeCtx = { role: "all" } as unknown as ServerContext; + +const pingRoute = createRoute({ + method: "get", + path: "/test-ping", + tags: ["test"], + responses: { + 200: { description: "ok", content: { "application/json": { schema: apiResponse(z.object({ who: z.string() })) } } }, + }, +}); + +const pingModule: ServerModule = { + name: "test-ping", + registerRoutes(app) { + app.use("/test-ping", requireScope("admin")); + app.openapi(pingRoute, (c) => c.json({ ok: true as const, data: { who: c.get("principal").name } }, 200)); + }, + registerTasks(registry) { + registry.register( + defineTask({ + name: "ingestFeed", + version: 1, + inputSchema: z.object({ feedSourceId: z.string() }), + handler: async () => "real", + }), + ); + }, + outboxRoutes: () => ({ "release.resolved": () => ({ task: "evaluateAcquisition", input: {} }) }), +}; + +describe("ServerModule 骨架", () => { + it("清单里是四个 W2 模块,名字唯一", () => { + expect(SERVER_MODULES.map((m) => m.name)).toEqual([ + "auth-settings", + "sources-identity", + "acquisition-organize", + "agent", + ]); + }); + + it("模块路由挂在 /api/v1 下,经过鉴权与 scope 检查,并进入 OpenAPI 文档", async () => { + const { logger } = memoryLogger(); + const app = createApp({ + config: { NODE_ENV: "development", DEV_ADMIN_KEY: DEV_KEY }, + logger, + bus: new EventBus(5), + role: "all", + version: "0.0.0-test", + modules: [pingModule], + ctx: fakeCtx, + }); + expect((await app.request("/api/v1/test-ping")).status).toBe(401); + const ok = await app.request("/api/v1/test-ping", { headers: { authorization: `Bearer ${DEV_KEY}` } }); + expect(ok.status).toBe(200); + expect((await json(ok)).data.who).toBe("dev-admin"); + const doc = await json(await app.request("/api/openapi.json")); + expect(doc.paths["/api/v1/test-ping"]).toBeDefined(); + // 未知路径仍是 JSON 404 + expect((await app.request("/api/v1/nope", { headers: { authorization: `Bearer ${DEV_KEY}` } })).status).toBe(404); + }); + + it("任务注册表:模块任务优先,其余 §3.1-8 任务由空壳兜底;重复注册报错", async () => { + const reg = buildTaskRegistry([...SERVER_MODULES, pingModule], fakeCtx); + expect(reg.names()).toEqual([...TASK_NAMES].sort()); + const ingest = reg.latest("ingestFeed"); + expect(await ingest.handler({ input: {} } as never)).toBe("real"); + expect(() => buildTaskRegistry([pingModule, { ...pingModule, name: "dup" }], fakeCtx)).toThrow(/重复注册/); + }); + + it("outbox 路由合并;同一事件被两个模块认领报错", () => { + expect(Object.keys(collectOutboxRoutes([...SERVER_MODULES, pingModule], fakeCtx))).toEqual(["release.resolved"]); + expect(() => collectOutboxRoutes([pingModule, { ...pingModule, name: "dup" }], fakeCtx)).toThrow(/重复认领/); + }); + + it("引擎就绪前调用给出明确错误", async () => { + const e = new DeferredTaskEngine(); + expect(e.ready).toBe(false); + expect(() => e.get("x")).toThrow(/尚未就绪/); + await e.stop(); + }); + + it("能力规格取自 contracts,可按模块注册", () => { + expect(CAPABILITY_IDS.map((id) => capabilitySpec(id).id)).toEqual([...CAPABILITY_IDS]); + }); +}); diff --git a/docs/runbook/dev.md b/docs/runbook/dev.md index f31d053a..0d8b7841 100644 --- a/docs/runbook/dev.md +++ b/docs/runbook/dev.md @@ -48,7 +48,9 @@ curl -N -H "Authorization: Bearer umt-dev-admin-key" "localhost:4380/api/v1/even `umt-server --role api|worker|all`(源码:`deno run ... apps/server/src/main.ts --role ...`)。`all` 下 worker 心跳会发布到进程内事件总线,SSE 可见;分进程部署时事件总线需改为 PG 投影(W2)。 -一次性子命令(容器入口用):`--role migrate`(迁移到最新,迁移文件已编进二进制)、`--role preflight`(启动自检:生产环境拒绝占位符密钥与 root、DB 可达、`WEB_ROOT/index.html` 存在)、`--role probe`(打本机 `/api/v1/health`,HEALTHCHECK 用)。 +启动时 `bootstrap()` 装配 `apps/server/src/modules/` 下的四个业务模块(见 apps/server/README.md),并连接任务引擎;`api` 角色也连接引擎(入队/信号/取消),`worker`/`all` 角色额外启动引擎 worker 与 OutboxRelay。数据库不可达时进程启动失败并退出。 + +一次性子命令(容器入口用):`--role migrate`(迁移到最新,迁移文件已编进二进制;随后迁移任务引擎的 openworkflow schema)、`--role preflight`(启动自检:生产环境拒绝占位符密钥与 root、DB 可达、`WEB_ROOT/index.html` 存在)、`--role probe`(打本机 `/api/v1/health`,HEALTHCHECK 用)。 设置 `WEB_ROOT` 后 API 进程在同一端口托管 web 构建产物:存在的文件直接返回,其余非 `/api` 路径回退 `index.html`。 diff --git a/packages/tasks/README.md b/packages/tasks/README.md index a9f35537..39a10c5e 100644 --- a/packages/tasks/README.md +++ b/packages/tasks/README.md @@ -15,6 +15,7 @@ | `OpenWorkflowEngine.connect({ databaseUrl, db, registry })` | `TaskEngine` 实现:`enqueue / signal / cancel / get / list / startWorker / stop` | | `defineTask` / `TaskRegistry` / `createDefaultRegistry()` | 任务定义与注册表;默认注册表里有 §3.1-8 的 9 个任务(目前是空壳) | | `OutboxRelay` | 用 `FOR UPDATE SKIP LOCKED` 认领 `outbox_event` 中 `status=pending` 且 `next_attempt_at` 已到的行,按 type 路由后 enqueue(幂等键取行的 `idempotency_key`,缺省 `outbox:`),成功标记 `dispatched` + `published_at`;失败 attempts+1 指数退避(上限 5 分钟),达 `maxAttempts` 置 `dead`;多进程并发时不会重复 | +| `migrateEngineSchema(databaseUrl, schema?)` | 只跑引擎自己的迁移后断开(部署时 `--role migrate`) | | `withAttemptToken(db, opKey, fn, { commit })` | 操作账本 fencing,表是 `tasks_op_ledger` | | 表 | `tasks_run_meta`、`tasks_fact`、`tasks_op_ledger` 由 `packages/db` 迁移 `0013_tasks_tables` 建立(类型见 `@umt/db` 的 `DB`);本包不再自带迁移,只通过注入的 Kysely 访问 | @@ -50,7 +51,7 @@ - 引擎幂等键只在 24h 内有效,而且不比较输入。长期去重靠业务表的唯一约束。 - `priority` 只记录在 `tasks_run_meta` 和 run 输入里,引擎仍按 `available_at` 先后认领,不支持真正的优先级调度。 - 取消只阻止后续步骤,不撤销已完成的外部动作。在已取消的 run 上,handler 抛出的错误会被引擎拒收,只在 worker 日志里出现。 -- `runMigrations` 默认 true(幂等)。等并入统一迁移后,启动时改成 false。 +- 引擎表(openworkflow schema)由 `migrateEngineSchema(databaseUrl)` 迁移:server 的 `--role migrate` 调用它;`--role all` 单进程连接时顺带迁移,`api`/`worker` 分进程以 `runMigrations:false` 连接。 ## 测试 diff --git a/packages/tasks/src/index.ts b/packages/tasks/src/index.ts index ada51add..fcc20859 100644 --- a/packages/tasks/src/index.ts +++ b/packages/tasks/src/index.ts @@ -36,6 +36,7 @@ export { } from "./op-ledger.ts"; export { ENGINE_SCHEMA, + migrateEngineSchema, OpenWorkflowEngine, type OpenWorkflowEngineOptions, signalName, diff --git a/packages/tasks/src/openworkflow-engine.ts b/packages/tasks/src/openworkflow-engine.ts index 2b95300d..cee64ba8 100644 --- a/packages/tasks/src/openworkflow-engine.ts +++ b/packages/tasks/src/openworkflow-engine.ts @@ -41,7 +41,7 @@ export interface OpenWorkflowEngineOptions { registry: TaskRegistry; /** 固定 "openworkflow";只允许测试覆盖 */ schema?: string; - /** 默认 true(幂等)。W2 并入统一迁移后启动时传 false */ + /** 默认 true(幂等)。部署时由 migrateEngineSchema 预先迁移,多进程启动传 false */ runMigrations?: boolean; logSink?: LogSink; /** handler 执行期间轮询取消标志的间隔,默认 1000ms */ @@ -50,6 +50,15 @@ export interface OpenWorkflowEngineOptions { export const signalName = (runId: string, name: string) => `umt:${runId}:${name}`; +/** + * 只跑引擎自己的迁移(openworkflow schema)后断开。部署时由 `--role migrate` 调用, + * 之后 api / worker 进程以 runMigrations:false 连接,避免多进程同时迁移。 + */ +export async function migrateEngineSchema(databaseUrl: string, schema = ENGINE_SCHEMA): Promise { + const backend = await BackendPostgres.connect(databaseUrl, { schema, runMigrations: true }); + await backend.stop(); +} + export class OpenWorkflowEngine implements TaskEngine { private readonly ow: OpenWorkflow; private readonly workers: Worker[] = []; diff --git a/packages/tasks/src/registry.ts b/packages/tasks/src/registry.ts index c4a0b7ce..a2ecf09b 100644 --- a/packages/tasks/src/registry.ts +++ b/packages/tasks/src/registry.ts @@ -54,6 +54,11 @@ export class TaskRegistry { return byVersion.get(Math.max(...byVersion.keys()))!; } + /** 是否已注册该任务(任一版本) */ + has(name: string): boolean { + return (this.defs.get(name)?.size ?? 0) > 0; + } + names(): string[] { return [...this.defs.keys()].sort(); } diff --git a/packages/tasks/test/registry.test.ts b/packages/tasks/test/registry.test.ts index 38a8356a..9c23f37a 100644 --- a/packages/tasks/test/registry.test.ts +++ b/packages/tasks/test/registry.test.ts @@ -7,6 +7,8 @@ describe("TaskRegistry", () => { const reg = createDefaultRegistry(); expect(reg.names()).toEqual([...TASK_NAMES].sort()); for (const n of TASK_NAMES) expect(reg.latest(n).version).toBe(1); + expect(reg.has("ingestFeed")).toBe(true); + expect(reg.has("nope")).toBe(false); }); it("latest 取最高版本;select 按 name@version / @version 过滤", () => {