feat: 网关 SSE 推送中心——按会话过滤扇出、订阅授权、Last-Event-ID 短窗回放、心跳

2026-08-13 11:53:06 +08:00
parent 41d58a9bb9
commit d308104c0b
8 changed files with 456 additions and 3 deletions
@@ -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<string>;
+  send(e: SseEvent): void;
+  /**
+   * D12：从 streamSSE 回调之外关掉这条流的把手。流的生命周期挂在回调里的一个
+   * 挂起 promise 上，只有 resolve 它才能收尾——所以 hub 得拿到一个由路由实现的
+   * 闭包（清心跳＋resolve）。单测手搓的连接不需要，故可选。
+   */
+  close?(): void;
+}
+export interface TaskAccessPort { allowed(taskId: string, staffSub: string): Promise<boolean> }
+
+/**
+ * 订阅失败带判别式：路由要按失败**种类**分流 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<string, Connection>();
+  private ring: SseEvent[] = [];
+  private seq = 0;
+  private streamSubs = new Map<string, Subscription>();
+  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.<sub>`。任务域的补发要求客户端
+   * 收到 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<SubscribeResult> {
+    // 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);
+    }
+  }
+}
@@ -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<void>((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.<sub>；任务域的补发要等客户端重新 /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<string, string> = {};
+    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;
+}
@@ -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=<sub>`。
+ * 走服务账号 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;
+    },
+  };
+}