From 1cd523d3e6a4a144a1db5524255aa19a568e5245 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 27 Sep 2026 19:59:54 +0800 Subject: [PATCH] feat(server): wire PgEventBus and event-driven settings cache into process assembly main.ts starts the PG-backed SSE bus (api/worker processes now see each other's events) and stops it on shutdown; bootstrap passes ctx.events to the settings reader so settings.changed invalidates every process's cache. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/context.ts | 2 +- apps/server/src/main.ts | 17 +++++++++++++++-- 2 files changed, 16 insertions(+), 3 deletions(-) diff --git a/apps/server/src/context.ts b/apps/server/src/context.ts index 926cd03a..948a6a6e 100644 --- a/apps/server/src/context.ts +++ b/apps/server/src/context.ts @@ -147,7 +147,7 @@ export async function bootstrap(opts: BootstrapOptions): Promise { engine, outbox, secrets: createSecretStore({ db, masterKey: config.SECRET_MASTER_KEY }), - settings: createSettingsReader({ db, logger }), + settings: createSettingsReader({ db, logger, events: opts.events }), capabilities: createCapabilityRegistry({ logger }), }; const tasks = buildTaskRegistry(opts.modules, ctx); diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index c1c753a6..eb144934 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -5,7 +5,7 @@ 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 { PgEventBus } from "./http/sse-pg.ts"; import { createLogger } from "./logger.ts"; import { SERVER_MODULES } from "./modules/index.ts"; import { runPreflight } from "./preflight.ts"; @@ -73,7 +73,19 @@ if (role === "migrate" || role === "preflight") { Deno.exit(code); } -const bus = new EventBus(); +// SSE 总线:sse_event 表 + LISTEN/NOTIFY,api 与 worker 分进程时事件互相可见 +let bus: PgEventBus; +try { + bus = await PgEventBus.start({ + db, + connectionString: config.DATABASE_URL, + logger: logger.child({ component: "sse-bus" }), + }); +} catch (err) { + logger.error({ err }, "SSE 总线启动失败"); + await db.destroy(); + Deno.exit(1); +} const stops: Array<() => Promise> = []; // 装配 ServerContext:模块注册任务/能力,连接任务引擎(api 角色也要:入队、信号、取消) @@ -138,6 +150,7 @@ async function shutdown(signal: Deno.Signal) { const force = setTimeout(() => Deno.exit(1), 10_000); await Promise.allSettled(stops.map((s) => s())); await engine.stop().catch((err) => logger.error({ err }, "任务引擎关闭失败")); + await bus.stop().catch(() => {}); await db.destroy(); clearTimeout(force); Deno.exit(0);