From 615917bb4543d4de22a50591d2963bf6e3be5236 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 16:43:49 +0800 Subject: [PATCH] feat(tasks,server,web): bridge outbox events to the SSE bus (U3) OutboxRelay gains a `mirror` option: listed types are claimed even without a task route and, after the claim transaction commits, published in order to the realtime bus (publish failures are logged, delivery state unaffected). The worker mirrors feed.*, decision.*, download.*, organization.* and subscription.* (feed.due excluded) with their business-id payloads, so the home page, inbox, task stream and feed tables finally receive them. Feed tables refresh on feed.fetched / feed.health_changed; the task stream also listens to feed and decision events. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/worker/index.ts | 14 ++++++++ apps/server/test/bootstrap.test.ts | 36 +++++++++++++++++++ apps/web/app/components/tasks/task-stream.tsx | 6 +++- .../app/components/workbench/feed-table.tsx | 28 +++++++++++++-- packages/tasks/src/outbox-relay.ts | 22 ++++++++++-- packages/tasks/test/outbox.test.ts | 36 +++++++++++++++++++ 6 files changed, 136 insertions(+), 6 deletions(-) diff --git a/apps/server/src/worker/index.ts b/apps/server/src/worker/index.ts index bbe53d9b..88a373cb 100644 --- a/apps/server/src/worker/index.ts +++ b/apps/server/src/worker/index.ts @@ -6,6 +6,16 @@ import { type EventBusPort, sseConnectionCount } from "../http/sse.ts"; import { type HeartbeatWriter, startHeartbeatWriter } from "../modules/system/heartbeat.ts"; import type { ServerContext, ServerModule } from "../modules/types.ts"; +/** + * outbox → SSE 桥:这些前缀的 outbox 事件在中继投递成功后镜像到实时总线(payload 原样,带业务 id), + * 首页、收件箱、任务、来源与下载页据此失效刷新。feed.due 是调度器内部节拍,不推给浏览器。 + */ +const SSE_MIRROR_PREFIXES = ["feed.", "decision.", "download.", "organization.", "subscription."] 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), +); + /** agent 组任务:整段会话占一槽数分钟,与文件复制等 io 任务分池 */ export const AGENT_TASKS = ["runAgent", "resolveDecision"] as const; @@ -146,6 +156,10 @@ export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe engine: ctx.engine, routes: collectOutboxRoutes(modules, ctx), logSink: tasksLogSink(logger), + mirror: { + types: SSE_MIRRORED_OUTBOX_TYPES, + publish: (ev) => ctx.events.publish(ev.type, ev.payload), + }, ...(runtime.outboxIntervalMs ? { intervalMs: runtime.outboxIntervalMs } : {}), }); runtime.setRelay?.(relay); diff --git a/apps/server/test/bootstrap.test.ts b/apps/server/test/bootstrap.test.ts index 9cadfab9..736db048 100644 --- a/apps/server/test/bootstrap.test.ts +++ b/apps/server/test/bootstrap.test.ts @@ -138,4 +138,40 @@ describe.skipIf(!hasPg)("bootstrap + worker(真实 PG)", () => { const rows = await sql<{ status: string }>`select status from outbox_event`.execute(db); expect(rows.rows).toEqual([{ status: "dispatched" }]); }, 40_000); + + it("outbox → SSE 桥:feed / decision / download / organization 事件提交后镜像到总线,feed.due 不推", async () => { + const got: { type: string; payload: unknown }[] = []; + const off = booted.ctx.events.subscribe((e) => { + if (e.type !== EventType.heartbeat) got.push({ type: e.type, payload: e.payload }); + }); + const decisionId = "33333333-3333-4333-8333-333333333333"; + const feedId = "44444444-4444-4444-8444-444444444444"; + await db.transaction().execute(async (trx) => { + await booted.ctx.outbox.emit(trx, EventType.feedDue, { feedId, reason: "manual" }); + await booted.ctx.outbox.emit(trx, EventType.feedFetched, { + feedId, + newItems: 2, + updatedItems: 0, + notModified: false, + }); + await booted.ctx.outbox.emit(trx, EventType.decisionCreated, { + decisionId, + kind: "identity_unresolved", + severity: "info", + title: "待识别", + }); + }); + const deadline = Date.now() + 20_000; + while (got.length < 2 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 100)); + off(); + // 同一事务内的事件 at 相同,相对顺序不保证 + got.sort((a, b) => a.type.localeCompare(b.type)); + expect(got).toEqual([ + { type: EventType.decisionCreated, payload: expect.objectContaining({ decisionId }) }, + { type: EventType.feedFetched, payload: expect.objectContaining({ feedId, newItems: 2 }) }, + ]); + // feed.due 没有路由也不镜像,保持待认领 + const due = await sql<{ status: string }>`select status from outbox_event where type = 'feed.due'`.execute(db); + expect(due.rows).toEqual([{ status: "pending" }]); + }, 30_000); }); diff --git a/apps/web/app/components/tasks/task-stream.tsx b/apps/web/app/components/tasks/task-stream.tsx index 27b5b9e0..2cb43f47 100644 --- a/apps/web/app/components/tasks/task-stream.tsx +++ b/apps/web/app/components/tasks/task-stream.tsx @@ -29,12 +29,16 @@ import { export type StreamTab = "all" | "active" | "failed"; -/** 引擎本身不发事件;业务事件足以代表任务在动,2 秒节流刷新 */ +/** 引擎本身不发事件;业务事件(outbox → SSE 桥)足以代表任务在动,2 秒节流刷新 */ const TASK_EVENTS = [ "task.created", "task.status_changed", "download.status_changed", + "download.threshold", "organization.changed", + "feed.fetched", + "decision.created", + "decision.updated", "agent.run_status", "schedule.changed", ]; diff --git a/apps/web/app/components/workbench/feed-table.tsx b/apps/web/app/components/workbench/feed-table.tsx index 743140b5..dd83aea5 100644 --- a/apps/web/app/components/workbench/feed-table.tsx +++ b/apps/web/app/components/workbench/feed-table.tsx @@ -1,8 +1,9 @@ -import type { FeedHealth, FeedSource } from "@umt/contracts"; +import { useQueryClient } from "@tanstack/react-query"; +import { EventType, type FeedHealth, type FeedSource } from "@umt/contracts"; import { RotateCcw, Rss, Zap } from "lucide-react"; -import { useState } from "react"; +import { useEffect, useRef, useState } from "react"; import { toast } from "sonner"; -import { feedGapHint, useFeedAction, useFeeds, useFetchFeedNow } from "~/api/feeds.ts"; +import { feedGapHint, feedKeys, useFeedAction, useFeeds, useFetchFeedNow } from "~/api/feeds.ts"; import { useSiteAccounts, useSites } from "~/api/sites.ts"; import { type DataColumn, DataTable } from "~/components/console/data-table.tsx"; import { DetailSection, DetailSheet } from "~/components/console/detail-sheet.tsx"; @@ -12,6 +13,7 @@ import { StatusBadge, type StatusTone } from "~/components/console/status-badge. import { Button } from "~/components/ui/button.tsx"; import { Switch } from "~/components/ui/switch.tsx"; import { errorMessage } from "~/lib/api.ts"; +import { useSse } from "~/lib/sse.ts"; import { formatDateTime } from "~/lib/utils.ts"; import { FeedFormDialog, PURPOSE_LABEL } from "./feed-form.tsx"; import { relTime } from "./release-fields.ts"; @@ -78,6 +80,25 @@ function FeedActions({ f }: { f: FeedSource }) { /** * Feed 列表 + 详情抽屉 + 新建/编辑对话框。来源健康页列全部,站点详情按 siteId 过滤。 */ +/** 抓取完成 / 健康变化(outbox → SSE 桥)到达时节流刷新 Feed 列表 */ +function useLiveFeeds() { + const qc = useQueryClient(); + const timer = useRef | null>(null); + useSse([EventType.feedFetched, EventType.feedHealthChanged], () => { + if (timer.current) return; + timer.current = setTimeout(() => { + timer.current = null; + void qc.invalidateQueries({ queryKey: feedKeys.all }); + }, 800); + }); + useEffect( + () => () => { + if (timer.current) clearTimeout(timer.current); + }, + [], + ); +} + export function FeedTable({ siteId, health, @@ -93,6 +114,7 @@ export function FeedTable({ filtered?: boolean; }) { const feeds = useFeeds({ health, siteId }); + useLiveFeeds(); const sites = useSites(); const accounts = useSiteAccounts(); const [editingFeed, setEditingFeed] = useState(null); diff --git a/packages/tasks/src/outbox-relay.ts b/packages/tasks/src/outbox-relay.ts index ca2606a7..1569e22f 100644 --- a/packages/tasks/src/outbox-relay.ts +++ b/packages/tasks/src/outbox-relay.ts @@ -34,6 +34,12 @@ export interface OutboxRelayOptions { /** 失败达到该次数后置 status='dead'(默认 10);其间按 2^attempts 秒退避,上限 5 分钟 */ maxAttempts?: number; logSink?: LogSink; + /** + * 镜像到实时总线(SSE):列出的 type 在投递成功、认领事务提交后逐条交给 publish(按 at, id 顺序)。 + * 只在 mirror 里、没有路由的 type 同样会被认领并直接标记已投递。publish 失败只记日志,不影响投递状态 + * (实时推送是尽力而为,前端断线或错过时靠重新拉快照)。 + */ + mirror?: { types: readonly string[]; publish: (event: OutboxEvent) => unknown }; } /** @@ -60,10 +66,12 @@ export class OutboxRelay { /** 处理一批;返回本批已投递与失败数 */ async pollOnce(): Promise<{ published: number; failed: number }> { - const types = Object.keys(this.opts.routes); + const mirrorTypes = new Set(this.opts.mirror?.types ?? []); + const types = [...new Set([...Object.keys(this.opts.routes), ...mirrorTypes])]; if (types.length === 0) return { published: 0, failed: 0 }; let published = 0; let failed = 0; + const toMirror: OutboxEvent[] = []; await this.opts.db.transaction().execute(async (trx) => { const rows = await sql` SELECT id, type, payload, at, idempotency_key, attempts FROM outbox_event @@ -73,7 +81,8 @@ export class OutboxRelay { FOR UPDATE SKIP LOCKED`.execute(trx); for (const ev of rows.rows) { try { - const route = this.opts.routes[ev.type]!(ev); + const routeFn = this.opts.routes[ev.type]; + const route = routeFn ? routeFn(ev) : null; if (route) { const payloadTrace = (ev.payload as { traceId?: unknown } | null)?.traceId; await this.opts.engine.enqueue(route.task, route.input, { @@ -85,6 +94,7 @@ export class OutboxRelay { await sql`UPDATE outbox_event SET status = 'dispatched', published_at = now(), attempts = attempts + 1, last_error = NULL WHERE id = ${ev.id}`.execute(trx); published++; + if (mirrorTypes.has(ev.type)) toMirror.push({ id: ev.id, type: ev.type, payload: ev.payload, at: ev.at }); } catch (err) { failed++; // 退避后重试;超过上限置 dead(需人工处理) @@ -99,6 +109,14 @@ export class OutboxRelay { } } }); + // 提交后才推送:回滚的事件不会先被浏览器看到 + for (const ev of toMirror) { + try { + await this.opts.mirror!.publish(ev); + } catch (err) { + this.log.error("outbox mirror publish failed", { eventId: ev.id, type: ev.type, err }); + } + } return { published, failed }; } diff --git a/packages/tasks/test/outbox.test.ts b/packages/tasks/test/outbox.test.ts index f11712a5..a3fe4278 100644 --- a/packages/tasks/test/outbox.test.ts +++ b/packages/tasks/test/outbox.test.ts @@ -127,4 +127,40 @@ describe.skipIf(!hasPg)("OutboxRelay", () => { }>`SELECT status FROM outbox_event WHERE type = 'echo.requested' AND published_at IS NULL`.execute(db); expect(dead.rows).toEqual([{ status: "dead" }]); }); + + it("mirror:投递成功后按顺序镜像到实时总线;只镜像不路由的 type 也认领,失败的不镜像", async () => { + await sql`DELETE FROM outbox_event`.execute(db); + await sql`INSERT INTO outbox_event (type, payload, at) VALUES + ('mirror.only', '{"recordId":"r1"}'::jsonb, now() - interval '3 second'), + ('echo.requested', '{"n":1001}'::jsonb, now() - interval '2 second'), + ('echo.requested', '{"n":"bad"}'::jsonb, now() - interval '1 second'), + ('not.mirrored', '{}'::jsonb, now())`.execute(db); + const seen: { type: string; payload: unknown }[] = []; + const relay = new OutboxRelay({ + db, + engine, + routes: { "echo.requested": (ev) => ({ task: "echo", input: ev.payload }) }, + logSink: () => {}, + mirror: { + types: ["mirror.only", "echo.requested"], + publish: (ev) => { + seen.push({ type: ev.type, payload: ev.payload }); + if (ev.type === "mirror.only") throw new Error("bus down"); // 推送失败不影响投递 + }, + }, + }); + expect(await relay.pollOnce()).toEqual({ published: 2, failed: 1 }); + expect(seen).toEqual([ + { type: "mirror.only", payload: { recordId: "r1" } }, + { type: "echo.requested", payload: { n: 1001 } }, + ]); + const status = await sql<{ type: string; status: string }>` + SELECT type, status FROM outbox_event ORDER BY at`.execute(db); + expect(status.rows.map((r) => `${r.type}:${r.status}`)).toEqual([ + "mirror.only:dispatched", + "echo.requested:dispatched", + "echo.requested:pending", + "not.mirrored:pending", + ]); + }); });