From d308104c0b2fd08683f24249882de98009d265a4 Mon Sep 17 00:00:00 2001 From: lovefyf Date: Thu, 13 Aug 2026 11:50:44 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=BD=91=E5=85=B3=20SSE=20=E6=8E=A8?= =?UTF-8?q?=E9=80=81=E4=B8=AD=E5=BF=83=E2=80=94=E2=80=94=E6=8C=89=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E8=BF=87=E6=BB=A4=E6=89=87=E5=87=BA=E3=80=81=E8=AE=A2?= =?UTF-8?q?=E9=98=85=E6=8E=88=E6=9D=83=E3=80=81Last-Event-ID=20=E7=9F=AD?= =?UTF-8?q?=E7=AA=97=E5=9B=9E=E6=94=BE=E3=80=81=E5=BF=83=E8=B7=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/app/app.ts | 16 +++- src/app/server.ts | 30 ++++++- src/modules/sse/hub.ts | 146 +++++++++++++++++++++++++++++++++ src/modules/sse/routes.ts | 79 ++++++++++++++++++ src/modules/sse/task-access.ts | 20 +++++ src/shared/errors.ts | 11 +++ src/shared/service-token.ts | 34 ++++++++ test/integration/sse.test.ts | 123 +++++++++++++++++++++++++++ 8 files changed, 456 insertions(+), 3 deletions(-) create mode 100644 src/modules/sse/hub.ts create mode 100644 src/modules/sse/routes.ts create mode 100644 src/modules/sse/task-access.ts create mode 100644 src/shared/errors.ts create mode 100644 src/shared/service-token.ts create mode 100644 test/integration/sse.test.ts diff --git a/src/app/app.ts b/src/app/app.ts index c320c5e..4c6cadc 100644 --- a/src/app/app.ts +++ b/src/app/app.ts @@ -1,10 +1,18 @@ import { Hono } from 'hono'; import { healthRoute, registerHealthCheck } from './health'; import { checkNats } from '../shared/bus/nats'; +import { createKeycloakTokenProvider } from '../shared/service-token'; import { proxyRoutes } from '../modules/proxy/routes'; import { uploadRoutes } from '../modules/upload/routes'; +import { SseHub } from '../modules/sse/hub'; +import { sseRoutes } from '../modules/sse/routes'; +import { createHttpTaskAccess } from '../modules/sse/task-access'; -export function buildApp(): Hono { +/** + * hub 由外面传：server.ts 要拿同一个实例去 attachNats／shutdown。缺省自建一个， + * 让只关心 HTTP 面的测试照旧 buildApp() 就能用。 + */ +export function buildApp(hub: SseHub = new SseHub()): Hono { registerHealthCheck('nats', checkNats); const app = new Hono(); // 统一错误形状：任何未捕获异常都收敛成 {error,message} 的 500，不漏 Hono 默认裸文本／栈 @@ -13,7 +21,11 @@ export function buildApp(): Hono { return c.json({ error: 'internal_error', message: '服务内部错误——已记录，稍后重试或联系维护' }, 500); }); app.route('/health', healthRoute()); // 公开：不过 requireAuth - // 顺序要紧：/api/upload 必须先于 /api 代理挂载，否则被代理的 catch-all 吞掉 + // 惰性构造：createKeycloakTokenProvider 只在闭包被调用时读 KEYCLOAK_CLIENT_SECRET， + // 建 app 本身不要求环境齐备，否则任何 buildApp() 的测试都会当场炸。 + const access = createHttpTaskAccess(createKeycloakTokenProvider()); + // 顺序要紧：/api/stream 与 /api/upload 必须先于 /api 代理挂载，否则被代理的 catch-all 吞掉 + app.route('/api/stream', sseRoutes(hub, access)); app.route('/api/upload', uploadRoutes()); app.route('/api', proxyRoutes()); return app; diff --git a/src/app/server.ts b/src/app/server.ts index 729706e..b9d2a48 100644 --- a/src/app/server.ts +++ b/src/app/server.ts @@ -1,7 +1,35 @@ import { serve } from '@hono/node-server'; import { buildApp } from './app'; import { config } from '../shared/config'; +import { closeNats, getNats } from '../shared/bus/nats'; +import { SseHub } from '../modules/sse/hub'; -serve({ fetch: buildApp().fetch, port: config.port }, (info) => { +const hub = new SseHub(); +const app = buildApp(hub); + +const server = serve({ fetch: app.fetch, port: config.port }, (info) => { console.log(`${config.serviceName} listening on :${info.port}`); }); + +// NATS 不可达不该拖垮 HTTP 面：代理／上传／健康检查照常，只是没有推送（/health 会照实报 fail） +const nc = await getNats().catch(() => null); +if (nc) hub.attachNats(nc); +else console.warn(`NATS ${config.natsUrl} 不可达——SSE 推送停摆，HTTP 面照常服务`); + +/** + * D12：网关此前没有信号处理（N1 靠默认 exit 143）。现在挂着长连 SSE，那些响应永远不结束， + * server.close() 会一直等它们——必须先把连接收掉。3s 兜底强退：任何一环卡住也不赖着不走。 + */ +let closing = false; +async function shutdown(signal: string): Promise { + if (closing) return; + closing = true; + console.log(`${signal} 收到——关闭 SSE 连接后退出`); + setTimeout(() => process.exit(0), 3000).unref(); + hub.shutdown(); + server.close(); + if (nc) await closeNats(); + process.exit(0); +} +process.on('SIGINT', () => void shutdown('SIGINT')); +process.on('SIGTERM', () => void shutdown('SIGTERM')); diff --git a/src/modules/sse/hub.ts b/src/modules/sse/hub.ts new file mode 100644 index 0000000..ea14ca6 --- /dev/null +++ b/src/modules/sse/hub.ts @@ -0,0 +1,146 @@ +import type { NatsConnection, Subscription } from 'nats'; + +export interface SseEvent { seq: number; topic: string; data: string } +export interface Connection { + id: string; + staffSub: string; + isAdmin: boolean; + topics: Set; + send(e: SseEvent): void; + /** + * D12：从 streamSSE 回调之外关掉这条流的把手。流的生命周期挂在回调里的一个 + * 挂起 promise 上，只有 resolve 它才能收尾——所以 hub 得拿到一个由路由实现的 + * 闭包（清心跳＋resolve）。单测手搓的连接不需要，故可选。 + */ + close?(): void; +} +export interface TaskAccessPort { allowed(taskId: string, staffSub: string): Promise } + +/** + * 订阅失败带判别式：路由要按失败**种类**分流 HTTP 码（形状错＝422 整单驳回， + * 连接不存在／被拒＝200 里逐条报）。靠 message.includes('不可订阅') 猜种类， + * 改一个字文案就会把 422 悄悄变成 200。 + */ +export type SubscribeResult = + | { ok: true } + | { ok: false; reason: 'bad-shape' | 'no-conn' | 'denied'; message: string }; + +const SUBSCRIBABLE = /^(touch\.task\.|touch\.workflow\.|stream\.task\.)[^.]+$/; + +export class SseHub { + private conns = new Map(); + private ring: SseEvent[] = []; + private seq = 0; + private streamSubs = new Map(); + private nc: NatsConnection | null = null; + + constructor(private replayWindow = 512) {} + + attachNats(nc: NatsConnection): void { + this.nc = nc; + const sub = nc.subscribe('touch.>', { + callback: (_err, msg) => this.deliver(msg.subject, new TextDecoder().decode(msg.data)), + }); + this.streamSubs.set('touch.>', sub); + } + + /** + * D11：只登记，**不**回放。回放必须排在 hello 帧之后——客户端拿到 connectionId + * 之前收到的事件无从对账（它还不知道该拿哪个 id 去 /subscribe）。 + */ + register(conn: Connection): void { + this.conns.set(conn.id, conn); + } + + /** + * 断线重连补发：按全局 seq 补窗口内错过的、且本连接可见的事件。 + * + * 诚实的局限（D11）：EventSource 重连不带订阅状态，路由只能以空 topics 重新登记， + * 于是这里实际只补得到 `touch.system` 与 `touch.staff.`。任务域的补发要求客户端 + * 收到 hello 后先重新 /subscribe——但那时 lastEventId 这一跳已经过去了。真正的 + * 任务域断点续传得等 N3 把订阅集做成服务端持久态（按 staffSub 记住上次订阅）。 + */ + replay(conn: Connection, lastEventId?: number): void { + if (lastEventId === undefined) return; + for (const e of this.ring) { + if (e.seq > lastEventId && this.visible(conn, e.topic)) conn.send(e); + } + } + + unregister(connId: string): void { + const conn = this.conns.get(connId); + this.conns.delete(connId); + if (conn) for (const t of conn.topics) this.maybeDropStreamSub(t); + } + + async subscribe(connId: string, topic: string, access: TaskAccessPort): Promise { + // D3：形状校验排在连接查找**之前**。请求体里写了个根本不可订阅的主题，是这个请求 + // 自身不合法（422），跟连接在不在无关；反过来先查连接的话，一个失效 connectionId + // 会把形状错整个遮掉，HTTP 面只能回 200。 + if (!SUBSCRIBABLE.test(topic)) { + return { + ok: false, + reason: 'bad-shape', + message: `主题 ${topic} 不可订阅——只接受 touch.task/workflow.* 与 stream.task.*`, + }; + } + const conn = this.conns.get(connId); + if (!conn) return { ok: false, reason: 'no-conn', message: `连接 ${connId} 不存在——先建立 /api/stream` }; + const taskId = topic.split('.').pop() ?? ''; + if (!conn.isAdmin && (topic.startsWith('stream.task.') || topic.startsWith('touch.task.'))) { + if (!(await access.allowed(taskId, conn.staffSub))) { + return { ok: false, reason: 'denied', message: `无权订阅任务 ${taskId}——需要负责人/审批人/admin` }; + } + } + conn.topics.add(topic); + if (topic.startsWith('stream.task.') && this.nc && !this.streamSubs.has(topic)) { + const sub = this.nc.subscribe(topic, { + callback: (_err, msg) => this.deliver(msg.subject, new TextDecoder().decode(msg.data)), + }); + this.streamSubs.set(topic, sub); + } + return { ok: true }; + } + + unsubscribe(connId: string, topic: string): void { + this.conns.get(connId)?.topics.delete(topic); + this.maybeDropStreamSub(topic); + } + + deliver(topic: string, data: string): void { + const e: SseEvent = { seq: ++this.seq, topic, data }; + this.ring.push(e); + if (this.ring.length > this.replayWindow) this.ring.shift(); + for (const conn of this.conns.values()) { + if (this.visible(conn, topic)) conn.send(e); + } + } + + /** + * D12：优雅退出。网关此前没有 SIGTERM 处理（N1 靠默认行为 exit 143），现在挂着 + * 长连 SSE——不主动关，server.close() 会一直等这些永不结束的响应。 + * 先快照再遍历：close() 可能同步触发 unregister，边删边迭代会漏。 + */ + shutdown(): void { + for (const conn of [...this.conns.values()]) conn.close?.(); + this.conns.clear(); + for (const sub of this.streamSubs.values()) sub.unsubscribe(); + this.streamSubs.clear(); + this.nc = null; + } + + private visible(conn: Connection, topic: string): boolean { + if (topic === 'touch.system') return true; + if (topic.startsWith('touch.staff.')) return topic === `touch.staff.${conn.staffSub}`; + return conn.topics.has(topic); + } + + private maybeDropStreamSub(topic: string): void { + if (!topic.startsWith('stream.task.')) return; + const stillUsed = [...this.conns.values()].some((c) => c.topics.has(topic)); + if (!stillUsed) { + this.streamSubs.get(topic)?.unsubscribe(); + this.streamSubs.delete(topic); + } + } +} diff --git a/src/modules/sse/routes.ts b/src/modules/sse/routes.ts new file mode 100644 index 0000000..2260dc5 --- /dev/null +++ b/src/modules/sse/routes.ts @@ -0,0 +1,79 @@ +import { Hono } from 'hono'; +import { randomUUID } from 'node:crypto'; +import { streamSSE } from 'hono/streaming'; +import { requireAuth, toAuthUser, verifyAccessToken } from '../../shared/keycloak-auth'; +import type { Connection, SseHub, TaskAccessPort } from './hub'; + +export function sseRoutes(hub: SseHub, access: TaskAccessPort): Hono { + const r = new Hono(); + + r.get('/', async (c) => { + // EventSource 带不了 Authorization 头：接受 ?access_token=，也接受头（降级轮询共用） + const token = c.req.query('access_token') ?? (c.req.header('authorization') ?? '').replace(/^bearer /i, ''); + if (!token) return c.json({ error: 'unauthorized', message: '缺 token（?access_token= 或 Authorization 头）' }, 401); + let user; + try { + user = toAuthUser(await verifyAccessToken(token)); + } catch { + // TODO(N2-final): JWKS/infra 失败应 503 而非 401 + return c.json({ error: 'invalid_token', message: 'token 校验失败' }, 401); + } + const lastRaw = c.req.header('last-event-id'); + const lastEventId = lastRaw !== undefined && /^\d+$/.test(lastRaw) ? Number(lastRaw) : undefined; + const connectionId = randomUUID(); + return streamSSE(c, async (stream) => { + // D12：流的寿命＝这个 promise 的寿命。旧写法 `new Promise(() => {})` 永不 settle—— + // 客户端断开后回调仍卡在 await 上，心跳定时器与整个闭包（连接对象、user、stream） + // 一起漏着不回收。resolve 把手要两个来源都够得着：客户端 abort、服务端 shutdown。 + let endStream!: () => void; + const closed = new Promise((resolve) => { + endStream = resolve; + }); + const heartbeat = setInterval(() => void stream.writeSSE({ event: 'ping', data: '{}' }), 25_000); + const end = (): void => { + clearInterval(heartbeat); + endStream(); + }; + const conn: Connection = { + id: connectionId, + staffSub: user.id, + isAdmin: user.roles.includes('admin'), + topics: new Set(), + send: (e) => void stream.writeSSE({ id: String(e.seq), event: e.topic, data: e.data }), + close: end, // D12：hub.shutdown() 借它从外面收流 + }; + // abort 先挂上：hello 还没写完客户端就跑了也要收得住 + stream.onAbort(() => { + hub.unregister(connectionId); + end(); + }); + hub.register(conn); + // D11：hello 必须先于回放——客户端要先拿到 connectionId 才谈得上 /subscribe 与对账。 + await stream.writeSSE({ event: 'hello', data: JSON.stringify({ connectionId }) }); + // 局限：新连接的 topics 是空集（EventSource 重连不携带订阅状态），所以这里实际只补 + // touch.system 与 touch.staff.；任务域的补发要等客户端重新 /subscribe 之后才可能， + // 而那时 Last-Event-ID 这一跳已经过去。详见 hub.replay 的注释。 + hub.replay(conn, lastEventId); + await closed; + }); + }); + + r.post('/subscribe', requireAuth(), async (c) => { + const body = (await c.req.json()) as { connectionId?: string; add?: string[]; remove?: string[] }; + if (!body.connectionId) return c.json({ error: 'validation_failed', message: '缺 connectionId（hello 事件里给了）' }, 422); + const results: Record = {}; + for (const topic of body.add ?? []) { + const res = await hub.subscribe(body.connectionId, topic, access); + // D3：只有"形状不合法"是请求本身写错了 → 整单 422；连接不存在／被拒是运行时状态， + // 逐条落进 results 回 200（客户端据此决定重连还是申请权限）。判别式而非文案匹配。 + if (!res.ok && res.reason === 'bad-shape') { + return c.json({ error: 'validation_failed', message: res.message }, 422); + } + results[topic] = res.ok ? 'ok' : `denied: ${res.message}`; + } + for (const topic of body.remove ?? []) hub.unsubscribe(body.connectionId, topic); + return c.json({ results }); + }); + + return r; +} diff --git a/src/modules/sse/task-access.ts b/src/modules/sse/task-access.ts new file mode 100644 index 0000000..7ccec58 --- /dev/null +++ b/src/modules/sse/task-access.ts @@ -0,0 +1,20 @@ +import { config } from '../../shared/config'; +import type { TokenProvider } from '../../shared/service-token'; +import type { TaskAccessPort } from './hub'; + +/** + * 订阅任务域主题前问任务服务一句：`GET {TASK_SERVICE_URL}/task/:id/access?staff=`。 + * 走服务账号 token（该端点在任务侧是 requireService(['api-gateway', ...])），不是转发用户 token。 + * 非 200 一律当"无权"——推送面失败关闭：宁可少订一条，不可把别人的任务事件扇给不该看的人。 + */ +export function createHttpTaskAccess(token: TokenProvider): TaskAccessPort { + return { + async allowed(taskId, staffSub) { + const url = `${config.taskServiceUrl}/task/${encodeURIComponent(taskId)}/access?staff=${encodeURIComponent(staffSub)}`; + const res = await fetch(url, { headers: { authorization: `Bearer ${await token()}` } }); + if (res.status !== 200) return false; + const body = (await res.json()) as { allowed?: boolean }; + return body.allowed === true; + }, + }; +} diff --git a/src/shared/errors.ts b/src/shared/errors.ts new file mode 100644 index 0000000..0854387 --- /dev/null +++ b/src/shared/errors.ts @@ -0,0 +1,11 @@ +/** 跨模块的失败分类——放 shared/ 而非某个业务模块下，好让 service-token 这类通用件能整份复用到别的仓 */ + +/** + * 永久失败：重试多少次都不会好（配置缺失、凭据无效、被拒绝的鉴权）。 + * 与普通 Error 分开，是为了让 saga 能分流——永久失败落 blocked＋治理， + * 暂时失败原地等下一轮重试；两者混在一个 Error 里只能无限重试。 + * name 显式覆盖：错误要以 String(err) 落库与进治理消息，"Error: xx" 会把永久性这层信息丢掉。 + */ +export class PermanentFailure extends Error { + override name = 'PermanentFailure'; +} diff --git a/src/shared/service-token.ts b/src/shared/service-token.ts new file mode 100644 index 0000000..0108bd2 --- /dev/null +++ b/src/shared/service-token.ts @@ -0,0 +1,34 @@ +/** client_credentials 换服务账号 token（指南 8.2）；缓存到 exp-30s；secret 只从环境读 */ + +import { PermanentFailure } from './errors'; + +export type TokenProvider = () => Promise; + +export function createKeycloakTokenProvider(): TokenProvider { + let cached: { token: string; expiresAt: number } | null = null; + return async () => { + if (cached && Date.now() < cached.expiresAt) return cached.token; + const url = process.env.KEYCLOAK_URL ?? 'https://keycloak.akhr-ai.com'; + const realm = process.env.KEYCLOAK_REALM ?? 'corp'; + const clientId = process.env.KEYCLOAK_CLIENT_ID ?? 'task-dispatcher'; + const secret = process.env.KEYCLOAK_CLIENT_SECRET; + // 漏配的 secret 不会因为重试而长出来 → 永久失败 + if (!secret) throw new PermanentFailure('缺 KEYCLOAK_CLIENT_SECRET——服务间调用需要 client_credentials，检查 .env'); + const res = await fetch(`${url}/realms/${realm}/protocol/openid-connect/token`, { + method: 'POST', + headers: { 'content-type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ grant_type: 'client_credentials', client_id: clientId, client_secret: secret }), + }); + if (!res.ok) { + const text = await res.text(); + // 400/401＝凭据本身错（client 不存在 / secret 失效），重试不会自愈；其余（5xx、网关）当 Keycloak 暂时不可达 + if (res.status === 400 || res.status === 401) { + throw new PermanentFailure(`换 token 被拒 ${res.status}：${text}——client_credentials 配置有误，检查 KEYCLOAK_CLIENT_ID/KEYCLOAK_CLIENT_SECRET`); + } + throw new Error(`换 token 失败 ${res.status}：${text}`); + } + const body = (await res.json()) as { access_token: string; expires_in: number }; + cached = { token: body.access_token, expiresAt: Date.now() + (body.expires_in - 30) * 1000 }; + return cached.token; + }; +} diff --git a/test/integration/sse.test.ts b/test/integration/sse.test.ts new file mode 100644 index 0000000..cad56d2 --- /dev/null +++ b/test/integration/sse.test.ts @@ -0,0 +1,123 @@ +import { describe, expect, it, beforeAll } from 'vitest'; +import { SseHub } from '../../src/modules/sse/hub'; +import type { Connection, SseEvent } from '../../src/modules/sse/hub'; +import { buildApp } from '../../src/app/app'; +import { makeToken, setupTestJwks } from '../support/token'; + +function conn(id: string, staffSub: string, topics: string[] = []): { c: Connection; got: SseEvent[] } { + const got: SseEvent[] = []; + return { c: { id, staffSub, isAdmin: false, topics: new Set(topics), send: (e) => got.push(e) }, got }; +} + +describe('SseHub 路由', () => { + it('staff 精确投递、system 广播、task 按订阅集、回放窗口', () => { + const hub = new SseHub(8); + const a = conn('a', 'sub-a', ['touch.task.t1']); + const b = conn('b', 'sub-b'); + hub.register(a.c); + hub.register(b.c); + hub.deliver('touch.system', '{}'); + hub.deliver('touch.staff.sub-a', '{}'); + hub.deliver('touch.task.t1', '{}'); + expect(a.got.map((e) => e.topic)).toEqual(['touch.system', 'touch.staff.sub-a', 'touch.task.t1']); + expect(b.got.map((e) => e.topic)).toEqual(['touch.system']); + // 断线重连回放：c 以 lastEventId=1 上线且订阅 t1 → 补 seq>1 的可见事件。 + // D11：register 只登记、replay 单独一步——路由要把 hello 插在两者之间。 + const c = conn('c', 'sub-c', ['touch.task.t1']); + hub.register(c.c); + hub.replay(c.c, 1); + expect(c.got.map((e) => e.topic)).toEqual(['touch.task.t1']); + }); + + it('subscribe 授权：access 拒 → 不入订阅集', async () => { + const hub = new SseHub(); + const a = conn('a', 'sub-a'); + hub.register(a.c); + const denyAccess = { allowed: async () => false }; + const res = await hub.subscribe('a', 'stream.task.t9', denyAccess); + expect(res.ok).toBe(false); + expect(a.c.topics.has('stream.task.t9')).toBe(false); + }); + + it('D3：形状校验排在连接查找之前，失败带判别式 reason 供路由分流', async () => { + const hub = new SseHub(); + const denyAccess = { allowed: async () => false }; + // 连接不存在 + 形状也不对 → 必须报形状（否则 HTTP 面永远回不出 422） + expect(await hub.subscribe('nope', 'touch.staff.hack', denyAccess)).toMatchObject({ + ok: false, + reason: 'bad-shape', + }); + expect(await hub.subscribe('nope', 'touch.task.t1', denyAccess)).toMatchObject({ + ok: false, + reason: 'no-conn', + }); + const a = conn('a', 'sub-a'); + hub.register(a.c); + expect(await hub.subscribe('a', 'stream.task.t9', denyAccess)).toMatchObject({ + ok: false, + reason: 'denied', + }); + }); + + it('D12：shutdown 逐条调 close 把手并摘除连接', () => { + const hub = new SseHub(); + let closed = 0; + const a = conn('a', 'sub-a'); + hub.register({ ...a.c, close: () => (closed += 1) }); + hub.shutdown(); + expect(closed).toBe(1); + hub.deliver('touch.system', '{}'); + expect(a.got).toEqual([]); // 已摘除，不再收扇出 + }); +}); + +describe('SSE HTTP 面', () => { + beforeAll(async () => { + await setupTestJwks(); + }); + + it('subscribe 主题形状校验 422；无 token 401', async () => { + const app = buildApp(); + expect((await app.request('/api/stream/subscribe', { method: 'POST' })).status).toBe(401); + const token = await makeToken({ roles: ['user'], aud: 'api-gateway' }); + const bad = await app.request('/api/stream/subscribe', { + method: 'POST', + headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' }, + body: JSON.stringify({ connectionId: 'nope', add: ['touch.staff.hack'] }), + }); + expect(bad.status).toBe(422); + }); + + it('D11/D12：hello 先于回放；hub.shutdown() 能把在途长连收尾', async () => { + const hub = new SseHub(8); + const app = buildApp(hub); + hub.deliver('touch.system', '{"n":1}'); // seq 1 + hub.deliver('touch.staff.test-sub-1', '{"n":2}'); // seq 2 + const token = await makeToken({ roles: ['user'], aud: 'api-gateway' }); + const res = await app.request(`/api/stream?access_token=${token}`, { + headers: { 'last-event-id': '1' }, + }); + expect(res.status).toBe(200); + const body = res.body; + if (!body) throw new Error('SSE 响应缺流体'); + const reader = body.getReader(); + const decoder = new TextDecoder(); + let text = ''; + while (!text.includes('event: touch.staff.test-sub-1')) { + const { value, done } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + expect(text).toContain('event: hello'); + expect(text.indexOf('event: hello')).toBeLessThan(text.indexOf('event: touch.staff.test-sub-1')); + expect(text).not.toContain('event: touch.system'); // seq 1 ≤ lastEventId，不补 + + // D12：回调里挂住的 promise 必须能被 close 把手 settle——settle 了 streamSSE 才会 + // close 流，客户端这边才读得到 done。旧写法 `await new Promise(() => {})` 永不 settle： + // 下面这次 read 会一直吊着（用例超时），线上则是 SIGTERM 关不掉服务。 + hub.shutdown(); + let ended = false; + for (let i = 0; i < 4 && !ended; i += 1) ended = (await reader.read()).done; + expect(ended).toBe(true); + }); +});