@db-node/websocket

Type-safe WebSocket toolkit for Node — connections, rooms (IM / customer service / e-commerce), channels (multiplex), presence (office), RPC, heartbeat, JWT auth.

• ✅ TypeScript 优先，完整泛型推断
• ✅ 场景化抽象：Room（IM / 客服 / 电商 / 办公协作） + Channel（多路复用）+ Presence（在线状态）
• ✅ 内置 RPC 双向调用：Client↔Server 均可发起
• ✅ 双层心跳：ws 原生 ping/pong + 应用层 $ping/$pong
• ✅ 与 @db-node/jwt 无缝集成的 upgrade 阶段鉴权中间件
• ✅ Node 客户端：自动重连（指数退避 + 抖动）+ RPC 超时 + 订阅管理
• ✅ 优雅关闭（closeAll + wss.close()）
• ✅ 只依赖 ws（peerDependency），零多余依赖

安装

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

ws 是 peer dependency，由你自行选择版本（>=8.0.0）。

核心概念

| 概念 | 说明 | 典型场景 | |-----------------|-------------------------------------------------------------------------------|---------------------------------------------------------| | Connection | 一条底层 WebSocket 的封装，含 id / userId / scene / device / rooms / channels | 单条链接的读写、踢下线 | | Room | 多对多逻辑分组 | 客服工单、订单会话、群 IM、协作文档 | | Channel | 同一物理连接上的逻辑通道（多路复用） | 一条 WS 同时跑 chat / notify / collab | | Presence | 用户 → 连接集合的映射 | "谁在线" / 多端登录 / 新登录踢旧 | | RPC | 基于 envelope id + rpc:'req'/'res' 的请求-响应 | 加载历史消息、查询订单详情、双向调用 | | Envelope | 统一 JSON 消息壳：{ id, type, channel, room, data, ack, error, ts, seq } | 所有业务消息都建议走此结构 |

Quick Start

服务端

import http from 'node:http';
import { createWsServer, createAuthMiddleware } from '@db-node/websocket';
import { verify } from '@db-node/jwt';

const httpServer = http.createServer();

const ws = createWsServer(httpServer, {
  path: '/ws',
  heartbeatInterval: 30_000,
  onUpgrade: createAuthMiddleware({
    from: { source: 'query', key: 'token' },
    verify: (t) => verify<{ userId: string }>(t, process.env.JWT_SECRET!),
  }),
  onConnection: (conn) => {
    console.log('connected:', conn.meta.id, 'user=', conn.meta.userId);
  },
});

// 注册频道（多路复用）
ws.hub.channel({ name: 'chat' });
ws.hub.channel({ name: 'notify' });

// 业务消息 handler
ws.on<{ text: string }>('chat.send', (msg, ctx) => {
  ws.hub.toRoom(msg.room!, { type: 'chat.msg', data: msg.data }, {
    exclude: [ctx.conn.meta.id],
  });
});

// RPC handler
ws.hub.onRpc<{ page: number }, { list: any[] }>('chat.history', async (req) => {
  const list = await db.select({ table: 'chat', /* ... */ });
  return { list };
});

httpServer.listen(3000);

客户端（浏览器直接用原生 WebSocket 即可，Node 侧用本包）

import { createWsClient } from '@db-node/websocket';

const client = createWsClient({
  url: 'ws://localhost:3000/ws?token=' + jwtToken,
  onOpen: () => console.log('connected'),
  onReconnect: async (attempt) => {
    console.log('reconnecting, attempt=', attempt);
    // 可选：重连前刷新 token；返回新 url 会替换当前 url
    // return 'ws://localhost:3000/ws?token=' + newToken;
  },
});

client.subscribe('chat');
client.on('chat.msg', (msg) => console.log('recv:', msg));
client.send({ type: 'chat.send', room: 'room-1', data: { text: 'hi' } });

const { list } = await client.call('chat.history', { page: 1 });

场景 1：客服会话（1v1 / 1vN 督导）

一个工单一个房间，客户 + 客服双方入房，随时可加入督导旁听。

ws.hub.channel({ name: 'cs' });

ws.on<{ ticketId: string; role: 'customer' | 'agent' | 'supervisor' }>(
  'cs.enter',
  (msg, ctx) => {
    const roomId = `cs:${msg.data!.ticketId}`;
    ctx.conn.join(roomId, { kind: 'cs', maxSize: 8 });
    // 通知同房其他人"有人加入"
    ws.hub.toRoom(roomId, {
      type: 'cs.member.join',
      data: { userId: ctx.conn.meta.userId, role: msg.data!.role },
    }, { exclude: [ctx.conn.meta.id] });
  }
);

ws.on<{ ticketId: string; text: string }>('cs.msg', (msg, ctx) => {
  const roomId = `cs:${msg.data!.ticketId}`;
  ws.hub.toRoom(roomId, {
    type: 'cs.msg',
    data: {
      from: ctx.conn.meta.userId,
      text: msg.data!.text,
    },
    ts: Date.now(),
  });
});

// 主动关闭工单：全屋踢出
ws.on<{ ticketId: string }>('cs.close', (msg) => {
  const roomId = `cs:${msg.data!.ticketId}`;
  ws.hub.toRoom(roomId, { type: 'cs.closed', data: {} });
  ws.hub.room(roomId)?.connIds.forEach((cid) => {
    ws.hub.get(cid)?.leave(roomId);
  });
});

场景 2：电商会话（订单维度）

买家 + 商家客服 + 售后客服，围绕订单 id 组成房间；订单状态变更时服务端主动推送。

ws.hub.channel({ name: 'ec' });

ws.on<{ orderId: string }>('ec.order.enter', (_msg, ctx) => {
  ctx.conn.join(`ec:${_msg.data!.orderId}`, { kind: 'ec' });
});

// 订单支付成功后（来自其他系统的回调）
export function onOrderPaid(orderId: string, order: any) {
  ws.hub.toRoom(`ec:${orderId}`, {
    type: 'ec.order.status',
    data: { status: 'paid', order },
  });
}

// 询价：单播给该订单绑定的客服
ws.on<{ orderId: string; text: string }>('ec.ask', async (msg, ctx) => {
  const agent = await findOrderAgent(msg.data!.orderId);
  ws.hub.toUser(agent.userId, {
    type: 'ec.ask',
    data: {
      orderId: msg.data!.orderId,
      from: ctx.conn.meta.userId,
      text: msg.data!.text,
    },
  });
});

场景 3：IM 群/私聊

ws.hub.channel({ name: 'im' });

// 加入群
ws.on<{ groupId: string }>('im.group.join', (msg, ctx) => {
  ctx.conn.join(`im:g:${msg.data!.groupId}`, { kind: 'im' });
});

// 群消息
ws.on<{ groupId: string; content: string }>('im.group.send', (msg, ctx) => {
  ws.hub.toRoom(`im:g:${msg.data!.groupId}`, {
    type: 'im.group.msg',
    data: {
      from: ctx.conn.meta.userId,
      content: msg.data!.content,
    },
  });
});

// 私聊（按 userId 单播；同一 user 的多端全部收到）
ws.on<{ toUserId: string; content: string }>('im.p2p.send', (msg, ctx) => {
  ws.hub.toUser(msg.data!.toUserId, {
    type: 'im.p2p.msg',
    data: {
      from: ctx.conn.meta.userId,
      content: msg.data!.content,
    },
  });
});

场景 4：办公 Presence + 多端登录 + 新登录踢旧

// "谁在线"查询
app.get('/api/presence', (_req, res) => {
  res.json(ws.hub.presence());
  //  { users: [{ userId, conns, scenes }], totalConns, totalUsers }
});

// 新登录踢旧登录（同一 user 只允许最新一条 conn 存活）
ws.hub.onPresence((ev) => {
  if (ev.type !== 'online') return;
  const others = ws.hub.presenceMgr.otherSessions(ev.userId, ev.connId);
  for (const o of others) {
    o.close(4006, 'replaced by new login');
  }
});

// 用户上线 / 全部下线的粗粒度事件
ws.hub.onPresence((ev) => {
  if (ev.type === 'user-online') pushOnlineNotice(ev.userId);
  if (ev.type === 'user-offline') markLastSeen(ev.userId);
});

场景 5：多路复用（Channel）

同一条 WS 上跑多个业务：客户端 $sub 后才会收到该频道的广播。

// 服务端
ws.hub.channel({ name: 'chat' });
ws.hub.channel({
  name: 'notify',
  authorize: async ({ conn }) => {
    // 只有 VIP 用户才能订阅通知
    const u = await loadUser(conn.meta.userId!);
    return u.isVip;
  },
});
ws.hub.channel({
  name: 'collab',
  onMessage: (msg) => {
    // 频道级中间件：消息进入 collab 前统一加 seq
    msg.seq = nextSeq();
    return msg;
  },
});

// 定向广播到只订阅了 notify 的用户
setInterval(() => {
  ws.hub.toChannel('notify', {
    type: 'notify.ping',
    data: { at: Date.now() },
  });
}, 5000);

// 客户端
client.subscribe('chat');
client.subscribe('notify');
// 不订阅 collab，就不会收到 collab 的广播

场景 6：双向 RPC

// Client → Server
ws.hub.onRpc<{ orderId: string }, { order: any }>('order.load', async (req) => {
  const order = await db.findOne({ table: 'order', where: { id: req.orderId } });
  if (!order) throw Object.assign(new Error('not found'), { code: 'NOT_FOUND' });
  return { order };
});

const { order } = await client.call('order.load', { orderId: '123' }, { timeout: 5000 });

// Server → Client（比如让某个客户端上报本地缓存版本）
const version = await ws.hub.callRpc<{}, { v: number }>(
  connId,
  'cache.version',
  {},
  { timeout: 3000 }
);

API 详解

createWsServer(http, options)

| 选项 | 类型 | 默认 | 说明 | |-----------------------|-------------------------------------|-------------|-----------------------------------------| | path | string | /ws | HTTP upgrade 路径 | | maxPayload | number | 1MB | 单条消息最大字节数 | | heartbeatInterval | number | 30_000 | 心跳间隔（ms）；<=0 关闭 | | heartbeatTimeout | number | 60_000 | 心跳超时（ms） | | authTimeout | number | 10_000 | 鉴权超时；仍无 userId 强断 | | onUpgrade | UpgradeMiddleware \| Middleware[] | - | upgrade 前置中间件（鉴权 / 黑名单） | | onConnection | (conn, hub) => void | - | 连接建立钩子 | | onMessage | MessageMiddleware \| Middleware[] | - | 全局消息中间件 | | onClose | (conn, code, reason) => void | - | 关闭钩子 | | onError | (err, conn?) => void | - | 全局错误钩子 | | scene | string | - | 场景标签，会写入每个 conn.meta.scene | | genConnId | () => string | UUID | 自定义连接 id 生成器 |

返回 WsServerInstance。

hub 对象

| 方法 | 说明 | |------------------------------------------|------------------------------------------| | hub.broadcast(msg) | 广播到所有连接 | | hub.toUser(userId, msg) | 单播到某 user 的所有连接（多端登录） | | hub.toConn(connId, msg) | 单播到某 conn | | hub.toRoom(roomId, msg, { exclude }) | 房间广播 | | hub.toChannel(channel, msg, { exclude })| 频道广播 | | hub.channel(def) | 注册频道 | | hub.room(roomId) | 查看房间信息（成员 conn/user） | | hub.byUser(userId) | 获取某 user 的所有连接 | | hub.presence() | 快照所有在线用户 | | hub.onPresence(cb) | 订阅 presence 事件（online / offline） | | hub.presenceMgr.otherSessions(uid, cid)| 拿到同一 user 的其它在线连接（用于踢旧） | | hub.onRpc(type, handler) | 注册 RPC handler | | hub.callRpc(connId, type, data) | 主动向某 conn 发起 RPC | | hub.kick(connId, reason) | 主动踢下线 | | hub.closeAll(code, reason) | 优雅关闭所有连接 |

createWsClient(options)

| 选项 | 默认 | 说明 | |--------------------------|----------|-------------------------------------------| | url | required | ws://... | | autoReconnect | true | 自动重连 | | reconnectInitialDelay | 500 | 首次重连延迟 | | reconnectMaxDelay | 30_000 | 最大重连延迟 | | reconnectMaxAttempts | ∞ | 最大重连次数 | | heartbeatInterval | 25_000 | 应用层心跳；超时会主动 close 触发重连 | | rpcTimeout | 10_000 | RPC 默认超时 | | onReconnect(attempt) | - | 重连前钩子；返回新 url 可覆盖 |

内置消息类型

| type | 方向 | 说明 | |---------------|-------|-----------------------------------------| | $sub | C→S | 订阅频道：{ type:'$sub', channel:'x' } | | $unsub | C→S | 取消订阅 | | $ping/$pong| 双向 | 应用层心跳（补充 ws 原生 ping/pong） |

关闭码

参考 CloseCode 常量：

| 常量 | 值 | 含义 | |-------------------------|------|-----------------------------| | NORMAL | 1000 | 正常关闭 | | GOING_AWAY | 1001 | 关服 | | AUTH_FAILED | 4001 | 鉴权失败 | | AUTH_TIMEOUT | 4002 | 鉴权超时 | | RATE_LIMITED | 4003 | 限流 | | HEARTBEAT_TIMEOUT | 4004 | 心跳超时 | | KICKED_BY_SERVER | 4005 | 主动踢下线 | | REPLACED_BY_NEW_LOGIN | 4006 | 被新登录替换 |

安全建议

1. 必须开启 upgrade 阶段鉴权：createAuthMiddleware 或自定义 verifier。
2. maxPayload 收紧：默认 1MB；如只跑聊天可降至 64KB。
3. 配合限流：可在 onUpgrade 中按 IP / userId 做入连限流。
4. 不要信任 msg.to：单播目标应由服务端根据房间/权限计算，而非直接采用客户端字段。
5. 敏感消息只走 toUser / toConn：避免误广播。

License

MIT