Novu Cloud 实时通道基于 Cloudflare Workers 与 Durable Objects 的 novu/socket-worker 架构与本地开发指南【免费下载链接】novuThe open-source communication infrastructure for agents and products项目地址: https://gitcode.com/GitHub_Trending/no/novu本篇文章围绕 Novu 仓库中 enterprise/workers/socket/README.md 展开系统讲解novu/socket-worker——一个承载 Novu Cloud WebSocketsPartySocket实时通道的 Cloudflare Worker Durable Object 服务。你将掌握它的整体架构Hono 路由、Durable Object 房间模型、EU 数据驻留、本地联调方法8787 端口、.dev.vars、与 API/Worker/Playground 的环境接线以及 JWT 认证、内部 API 鉴权、WebSocket Hibernation 与 contextKeys 精确匹配等核心实现原理可直接上手在本地跑通整套实时链路。一、socket-worker 在 Novu 实时体系中的角色Novu 的开源实时通道由apps/wsNode.js 网关基于 Socket.IO/PartySocket 生态与本次要讲的novu/socket-worker组成。后者定位为Novu Cloud 的 WebSocket 实时路径当NOVU_ENTERPRISEtrue时Cloud 环境下的实时消息不再走传统 Node 网关而是由 Cloudflare 边缘上的 Worker Durable Object 承担 WebSocket 升级与消息投递也就是 README 中标注的 Cloudflare Worker Durable Object for Novu Cloud WebSockets (PartySocket)。从 package.json 可以看到该 Worker 的运行时依赖非常精简honoHTTP 路由框架处理 WebSocket 升级、内部消息接口与健康检查ws/types/jsonwebtokenWebSocket 与 JWT 相关类型支撑tsndr/cloudflare-worker-jwt在 Worker 运行时内完成 JWT 签名校验Cloudflare Workers 无 Node 原生crypto完整 API故使用专为 Worker 设计的 JWT 库wrangler^4.49.0本地开发与多环境部署 CLI。二、整体架构与请求路径Worker 的入口是 src/index.ts一个典型的 Hono 应用对外暴露三条路由路由方法中间件职责/GETauthenticateJWT携带?token发起 WebSocket 升级/sendPOSTauthenticateInternalAPI内部服务向指定用户房间推送事件/healthGET无健康检查返回OK未命中路由统一返回 404Not found应用级错误经app.onError记录日志并返回 500。2.1 WebSocket 升级链路当客户端如 playground/web-chat 中的 PartySocket 客户端发起连接时请求经过 middleware/auth.ts 校验?token中的 JWT然后进入 handlers/websocket.ts 的handleWebSocketUpgrade从 JWT payload 与请求中取出userId、subscriberId、organizationId、environmentId、contextKeys计算房间 IDroomId ${environmentId}:${userId}即每个用户在其环境内拥有一个专属房间根据REGION变量决定是否使用 EU 数据驻留命名空间WEBSOCKET_ROOM.jurisdiction(eu)通过idFromName(roomId)定位 Durable Object 实例并stub.fetch(...)把用户信息以X-User-Id、X-Subscriber-Id、X-Organization-Id、X-Environment-Id、X-JWT-Token、X-Context-Keys等自定义头透传给 DO。2.2 消息推送链路API/Worker 需要给在线用户推送实时事件时向/send发起 POST请求体结构由handleSendMessage校验逻辑确认{ userId: 用户 ID字符串, environmentId: 环境 ID字符串, event: 事件名如 notification.inbox_received, data: 任意业务负载, contextKeys: [可选, 上下文键数组] }handleSendMessage会依次校验userId与event必填、environmentId必填、三者必须为字符串随后同样按environmentId:userId定位 Durable Object并通过context.executionCtx.waitUntil(stub.sendToUser(...))异步投递接口立即返回{ success: true, roomId, timestamp }。三、本地开发环境搭建该包属于 pnpm workspace见根目录 pnpm-workspace.yaml因此依赖安装统一在仓库根目录执行pnpm install启动开发服务器有两种等价方式# 方式一仓库根目录运行利用 workspace filter pnpm dev:socket-worker # 方式二进入本包目录直接运行 cd enterprise/workers/socket pnpm run dev根目录 package.json 中dev:socket-worker定义为pnpm --filter novu/socket-worker dev即pnpm run dev执行的是wrangler dev --env local。端口约定务必遵守socket-worker 固定运行在http://127.0.0.1:8787而本地thalamus-observer另一 Cloudflare Worker见 scripts/dev-environment-setup.sh 相关脚本使用8788两者互不冲突。若修改 socket-worker 端口需同步修改下游所有指向它的环境变量。3.1 首次运行前的密钥配置.dev.vars被 gitignore首次需要从模板复制并填写cd enterprise/workers/socket cp .dev.vars.example .dev.vars然后参考 .dev.vars.example 中的注释从apps/api/src/.env取值JWT_SECRETapps/api/src/.env 中的 JWT_SECRET INTERNAL_API_KEYapps/api/src/.env 中的 INTERNAL_SERVICES_API_KEY关键约束INTERNAL_API_KEY必须与 API 侧的INTERNAL_SERVICES_API_KEY完全一致因为/send接口的调用方API/Worker 内部服务正是用它作为 Bearer 凭证不一致将导致推送被 401 拒绝。JWT_SECRET则用于校验客户端 WebSocket 升级时携带的?token签名必须与签发 token 的 API 侧共享同一密钥。localwrangler 环境会将API_URL默认设置为http://127.0.0.1:3000见 wrangler.jsonc用于 Durable Object 回拨 API 上报用户在线状态。四、打通 API / Worker / Playground 的完整环境接线README 给出了三条链路的联调配置。要让本地整套系统API Worker Playground都走 Cloudflare socket需要1) API 与 Worker 侧apps/api/src/.env和apps/worker/src/.envSOCKET_WORKER_URLhttp://127.0.0.1:8787 NOVU_ENTERPRISEtrue # INTERNAL_SERVICES_API_KEY 保持与 .dev.vars 的 INTERNAL_API_KEY 相同2) Playground 侧playground/web-chatNEXT_PUBLIC_NOVU_SOCKET_URLhttp://127.0.0.1:8787 NEXT_PUBLIC_NOVU_SOCKET_TYPEcloud其中NEXT_PUBLIC_NOVU_SOCKET_TYPEcloud指示 Playground 走 Cloudflare 云端 socket 路径而非本地 Node 网关——这正是 README 强调的 Locally, Cloudflare sockets are the realtime path 的含义。配置完成后Playground 发起的 WebSocket 升级请求会携带 JWT 直连 8787 端口的 Worker。五、wrangler 多环境配置与部署wrangler.jsonc 定义了四个环境差异点集中在名称、路由域名、Durable Object 绑定与变量环境Worker 名称自定义域名API_URLREGIONlocalsocket-worker-local无workers_devhttp://127.0.0.1:3000globalstagingsocket-worker-stagingsocket.novu-staging.cohttps://api.novu-staging.coglobalproduction-ussocket-worker-production-ussocket.novu.cohttps://api.novu.coglobalproduction-eusocket-worker-production-eueu.socket.novu.cohttps://eu.api.novu.coeu所有环境都声明了同一个 Durable Object 绑定WEBSOCKET_ROOM → WebSocketRoom并在migrations中以new_sqlite_classes: [WebSocketRoom]tagv1注册SQLite 后端 DO。此外observability.enabled: true开启了 Cloudflare 观测。对应 package.json 中的部署脚本pnpm run deploy # wrangler deploy默认环境 pnpm run deploy:staging pnpm run deploy:production-us pnpm run deploy:production-eu pnpm run deploy:local pnpm run cf-typegen # 从 wrangler 配置生成 Worker 类型声明REGIONeu在生产 EU 环境中的作用非常关键DO 命名空间会调用jurisdiction(eu)将连接与数据锁定在欧盟境内以满足数据驻留要求见下一节源码说明。六、核心实现原理WebSocketRoom Durable ObjectDurable Object 的实现集中在 src/durable-objects/websocket-room.ts类WebSocketRoom是整条实时链路的房间载体。以下是几个值得深入理解的设计点。6.1 基于 Hibernation API 的连接管理构造函数中通过this.ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair(ping, pong))配置自动心跳应答使 Worker 可在空闲时休眠而连接保持存活。WebSocket 接受使用hibernation 兼容方式const tags [user:${userId}, env:${environmentId}]; this.ctx.acceptWebSocket(server, tags); server.serializeAttachment({ jwtToken, connectedAt: Date.now(), contextKeys });tags给连接打上user:与env:标签后续可定向按标签检索连接serializeAttachment把 JWT、连接时间、contextKeys 持久化到连接附件上附件上限 2KBJWT 通常 1KB这样 DO 休眠唤醒后依然能恢复连接元数据——这正是代码注释强调No need to store JWT tokens in memory的原因。运行时通过三个钩子驱动生命周期webSocketMessage收到客户端消息此处仅校验元数据存在性、webSocketClose关闭连接并触发下线上报、webSocketError记录错误日志。6.2 房间容量与并发保护每个 DO 实例设定了MAX_CONNECTIONS 100的硬上限。fetch在升级前检查this.ctx.getWebSockets().length达到上限返回503 Retry-After: 60让客户端 60 秒后重试。此外还暴露了三个统计方法getActiveConnectionsForUser、getTotalActiveConnections、getConnectionCapacity返回{ current, max, available }接口契约定义在 src/types/index.ts。6.3 contextKeys 精确匹配多上下文隔离投递一个用户可能同时打开多个上下文例如不同的工作区页面/send携带的contextKeys与连接附件中的contextKeys做精确匹配后才投递。isExactMatch的规则是消息 contextKeys 为空数组 → 仅投递给 contextKeys 也为空的连接长度不一致 → 不投递否则逐个成员比较every(key inboxContextKeys.includes(key))全部命中才投递。代码注释明确指出这套逻辑与ws.gateway.ts保持一致保证了新旧实时通道在语义上的兼容。投递过程对消息体只做一次JSON.stringify预序列化{ event, data, timestamp }再对命中连接并行发送并用Promise.allSettled容错。6.4 在线状态回拨 API连接建立与断开时DO 都会通过notifySubscriberOnlineState向API_URL的POST /v1/internal/subscriber-online-state上报请求体含subscriberId、environmentId、isOnline、organizationId、timestamp以Bearer ${jwtToken}鉴权。该调用使用ctx.waitUntil包裹确保 DO 可立即进入休眠而不阻塞连接建立只有所有同用户连接都断开剩余连接数 ≤ 0时才上报离线。七、安全模型双层认证7.1 客户端侧JWT 校验middleware/auth.ts 中authenticateJWT负责升级请求认证从?token查询参数取 JWT缺失返回 401用JWT_SECRET通过tsndr/cloudflare-worker-jwt验签并解码从 payload 提取_id作为 userId、subscriberId缺省回退为 userId、organizationId、environmentId、contextKeys任一缺失返回 401认证信息通过 Honocontext.set注入后续处理。7.2 内部侧常量时间比较middleware/internal-auth.ts 保护/send请求头需携带Authorization: Bearer key与INTERNAL_API_KEY做常量时间比较constantTimeEquals逐字符异或累加长度不等直接失败从实现层面抵御时序侧信道攻击。若服务端未配置INTERNAL_API_KEY则返回 500。八、类型契约与可观测性src/types/index.ts 定义了完整的环境与元数据契约IEnvWEBSOCKET_ROOMDO 命名空间绑定、JWT_SECRET、INTERNAL_API_KEY必填API_URL、REGION可选IConnectionMetadatauserId、environmentId、connectedAt、jwtToken、contextKeys——即 serializeAttachment 持久化的全部字段IWebSocketRoomsendToUser、连接数查询与容量查询接口。运维层面/health提供存活探针wrangler 配置开启了observabilityapp.onError统一记录应用错误。日常排障可结合 Cloudflare Dashboard 的 Worker 日志查看[Internal API] Routing message to room: ...等关键日志行。九、小结novu/socket-worker展示了如何用 Cloudflare Workers Durable Objects 构建生产级实时通道边缘就近升级 WebSocket、以environmentId:userId为粒度分房、Hibernation API 降本增效、JWT 内部 API Key 双层鉴权、EU 数据驻留按区域隔离。对于希望深度定制 Novu Cloud 实时链路或自建同类基础设施的开发者建议从 enterprise/workers/socket/src/index.ts路由入口→ src/handlers/websocket.ts升级与推送→ src/durable-objects/websocket-room.ts房间核心这条调用链入手阅读再结合 wrangler.jsonc 完成本地与多环境部署验证。【免费下载链接】novuThe open-source communication infrastructure for agents and products项目地址: https://gitcode.com/GitHub_Trending/no/novu创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考