@db-node/sse

Type-safe Server-Sent Events toolkit for Node — 主动推送、多路复用（Topic）、Last-Event-Id 断线续传、Presence、心跳保活、JWT 鉴权。

• ✅ TypeScript 优先，事件 / 主题 / 上下文全部有类型
• ✅ 多路复用：一条 SSE 连接同时订阅 notify / order / chat 多主题
• ✅ 断线续传：客户端 EventSource 自动带 Last-Event-Id，服务端 replay 未收到的事件
• ✅ 主动推送：hub.toUser / toConn / toTopic / broadcast 四种投递
• ✅ 多端登录：Presence 记录 user → 连接集合，toUser 天然覆盖所有端
• ✅ 心跳保活：定期下发 : keepalive\n\n 穿透 nginx / cloudflare / 移动网关
• ✅ 鉴权中间件：与 @db-node/jwt 无缝集成
• ✅ 零外部依赖：只用 Node 内建 http，Express/Koa/原生 http.Server 通吃
• ✅ 优雅关闭：sse.close() 统一释放心跳 + 所有连接

安装

npm i @db-node/sse
# 若使用 jwt 鉴权
npm i @db-node/jwt

核心概念

| 概念 | 说明 | 典型场景 | |-----------------|---------------------------------------------------------------------|-------------------------------------------------------| | Connection | 一条 SSE 长连接（http.ServerResponse 的封装） | 单条链接的推送、踢下线 | | Topic | 主题订阅（多路复用；一条连接可订阅多 topic） | order:123 / notify:user42 / chat:room7 | | Hub | 全局中心调度器：广播 / 单播 / topic 播 / user 播 | 主动上报订单状态 / 系统公告 / AI 流式输出 | | ReplayStore | 每个 topic 独立的 FIFO 事件缓冲 | Last-Event-Id 断线续传 | | Presence | userId → connIds 的多端映射 | "谁在线" / 多端同步 / 强制下线 |

Quick Start

服务端（Express）

import express from 'express';
import { createSseServer, createSseAuthMiddleware } from '@db-node/sse';
import { verify } from '@db-node/jwt';

const app = express();

const sse = createSseServer({
  heartbeatInterval: 25_000, // 25s 一个 keepalive
  defaultReplaySize: 200,    // 每 topic 保留最近 200 条带 id 事件
  onUpgrade: createSseAuthMiddleware({
    from: { source: 'query', key: 'token' },
    verify: (t) => verify<{ userId: string }>(t, process.env.JWT_SECRET!),
  }),
  onConnection: (conn) => {
    // 建连时按业务默认订阅几个 topic
    conn.subscribe([`notify:${conn.meta.userId}`, 'sys.broadcast']);
    // 也可以显式订阅 URL query 里传来的 topic
    // conn.subscribe(conn.meta.initialTopics as string[]);
  },
});

app.get('/sse', sse.handler);
app.listen(3000);

// 业务：任何异步事件都可以主动推
setInterval(() => {
  sse.hub.toUser('u42', {
    event: 'notify',
    data: { title: '你有一条新消息', ts: Date.now() },
    id: `n-${Date.now()}`,
  });
}, 5_000);

服务端（原生 http.Server）

import http from 'node:http';
import { createSseServer } from '@db-node/sse';

const sse = createSseServer();
const server = http.createServer((req, res) => {
  if (req.url?.startsWith('/sse')) return sse.handler(req, res);
  res.writeHead(404).end();
});
server.listen(3000);

浏览器客户端（原生 EventSource）

const es = new EventSource('/sse?token=' + jwt);

// 通用 message（不带 event: 名字的事件）
es.onmessage = (e) => console.log('msg:', JSON.parse(e.data));

// 具名事件
es.addEventListener('notify', (e) => {
  const payload = JSON.parse(e.data);
  toast(payload.title);
});

es.addEventListener('order.paid', (e) => {
  const order = JSON.parse(e.data);
  updateOrderUI(order);
});

EventSource 会自动重连，并在 Last-Event-Id header 中带上最后收到的事件 id，服务端 replay 缓冲会自动补齐。

场景示例

场景 1：客服 —— 服务端主动上报订单/工单状态

const sse = createSseServer({ onUpgrade: authMw });

// 客户端建连后订阅当前工单
sse.hub.topic({
  name: 'ticket',
  authorize: ({ conn, topic }) => hasTicketAccess(conn.meta.userId, topic),
  replaySize: 500,
});

// 座席状态变化 → 主动上报
kafkaConsume('ticket-events', (msg) => {
  sse.hub.toTopic(`ticket:${msg.ticketId}`, {
    event: 'ticket.update',
    data: msg,
    id: msg.eventId, // 关键：带 id 才会进 replay 缓冲
  });
});

场景 2：电商 —— 订单支付/发货实时推送

// 用户下单成功后，前端订阅 order:xxx
sse.hub.toUser(userId, {
  event: 'order.paid',
  data: { orderId, amount, paidAt },
  id: `${orderId}:paid`,
});

// 同时按 topic 广播给"物流大屏"这样的第三方消费者
sse.hub.toTopic(`order:${orderId}`, {
  event: 'order.shipped',
  data: { trackingNo },
  id: `${orderId}:shipped`,
});

场景 3：办公 —— 通知中心（多路复用）

一条 SSE 连接同时订阅 notify / mention / sys 三类事件：

sse.hub.topic({ name: 'notify' });
sse.hub.topic({ name: 'mention' });
sse.hub.topic({ name: 'sys' });

// 建连时按 user 订阅个人 topic
sse.handler; // 内部会调用 onConnection：
// onConnection: (conn) => conn.subscribe([
//   `notify:${conn.meta.userId}`,
//   `mention:${conn.meta.userId}`,
//   'sys',
// ]);

// @ 提醒
sse.hub.toTopic(`mention:${targetUserId}`, {
  event: 'mention',
  data: { from, docId, snippet },
  id: mentionId,
});

场景 4：断线续传（Last-Event-Id 自动 replay）

// 事件必须带 id 才能进 replay 缓冲
for (const item of stream) {
  sse.hub.toTopic('feed', {
    event: 'feed.item',
    data: item,
    id: item.seq.toString(), // 关键
  });
}

// 客户端网络抖动断开后：
// 1) EventSource 自动重连
// 2) 请求 header 带 Last-Event-Id: <上一次收到的 seq>
// 3) createSseServer 内部：conn.subscribe('feed') → hub.replayStore.since('feed', lastId) → 逐条 replay

配置每 topic 的缓冲大小：

sse.hub.topic({ name: 'feed', replaySize: 1000 });     // 保留 1000 条
sse.hub.topic({ name: 'high-freq', replaySize: 0 });   // 关闭该 topic 的 replay

场景 5：AI 流式输出（chunk 逐条推）

app.post('/chat/:sid', async (req, res) => {
  const { sid } = req.params;
  // 客户端另开一个 EventSource 订阅 chat:<sid>
  for await (const chunk of aiStream(req.body.prompt)) {
    sse.hub.toTopic(`chat:${sid}`, {
      event: 'chat.chunk',
      data: { text: chunk.text, done: chunk.done },
      id: `${sid}:${chunk.seq}`,
    });
  }
  res.json({ ok: true });
});

场景 6：Presence 事件 + 强制下线

sse.hub.onPresence((ev) => {
  if (ev.type === 'user-online') metrics.online.inc();
  if (ev.type === 'user-offline') metrics.online.dec();
});

// 后台管理踢用户
for (const c of sse.hub.byUser('spamUser')) c.close();

API 速览

createSseServer(options?)

| Option | 类型 | 说明 | |---------------------|-----------------------------------|--------------------------------------------------------------| | heartbeatInterval | number（ms，默认 25000，≤0 关）| SSE 注释心跳间隔 | | clientRetry | number（ms，默认 3000） | 建连时告诉客户端断线后 X ms 重连 | | defaultReplaySize | number（默认 100） | 每 topic 默认 replay 缓冲条数 | | onUpgrade | 中间件数组 | 建连阶段鉴权 / 限流 / 黑名单 | | onConnection | (conn, hub) => void \| Promise | 建连后钩子；一般在这里 conn.subscribe(...) | | onClose | (conn, reason) => void | 断开钩子 | | onError | (err, conn?) => void | 兜底错误 | | cors | {origin, credentials, headers} | 需要跨域时开启 | | scene | string | 打到每条 conn.meta.scene 上（如 'office'） |

SseHub

hub.broadcast(ev)                          // 全体
hub.toUser(userId | userId[], ev)          // 单/多 user（覆盖多端）
hub.toConn(connId | connId[], ev)          // 精确到连接
hub.toTopic(topic | topic[], ev, {exclude?}) // 主题（会走 replay 缓冲）
hub.topic({ name, authorize?, onEvent?, replaySize? }) // 注册主题
hub.kick(connId)                           // 强踢
hub.presence()                             // 在线快照
hub.byUser(userId) / hub.byTopic(topic)    // 反向查询
hub.closeAll()                             // 优雅关闭所有连接

SseConnection

conn.subscribe(topic | topic[])
conn.unsubscribe(topic | topic[])
conn.send({ event, data, id?, retry? })
conn.sendRaw(chunk)
conn.patchMeta({ userId, ... })
conn.close()

与 @db-node/websocket 的选择

| 需求 | 建议 | |----------------------------------------|--------------------------| | 服务端 → 客户端 单向 推送 | ✅ SSE，最简单省心 | | 客户端 → 服务端 也需要频繁上行 | ✅ WebSocket | | 强防火墙 / 公司代理环境 | ✅ SSE（走 HTTP） | | RPC 双向请求-响应 | ✅ WebSocket | | 只发通知/大屏/流式输出/进度上报 | ✅ SSE | | IM / 协作文档 / 在线游戏 | ✅ WebSocket |

两者可以在同一进程内共存。

License

MIT