From f5b051d9690138eff28a6383bbdbc82b7627c67f Mon Sep 17 00:00:00 2001 From: RainbowBird Date: Wed, 3 Jun 2026 00:45:50 +0800 Subject: [PATCH] refactor(server): split chat ws runtime --- apps/server/src/routes/chat-ws/broadcast.ts | 97 ++++++++++ .../src/routes/chat-ws/connection-registry.ts | 78 +++++++++ apps/server/src/routes/chat-ws/index.ts | 165 ++++-------------- apps/server/src/routes/chat-ws/rpc.ts | 74 ++++++++ 4 files changed, 279 insertions(+), 135 deletions(-) create mode 100644 apps/server/src/routes/chat-ws/broadcast.ts create mode 100644 apps/server/src/routes/chat-ws/connection-registry.ts create mode 100644 apps/server/src/routes/chat-ws/rpc.ts diff --git a/apps/server/src/routes/chat-ws/broadcast.ts b/apps/server/src/routes/chat-ws/broadcast.ts new file mode 100644 index 000000000..85142ab1a --- /dev/null +++ b/apps/server/src/routes/chat-ws/broadcast.ts @@ -0,0 +1,97 @@ +import type Redis from 'ioredis' + +import type { ChatBroadcastPayload } from '../../utils/chat-broadcast' +import type { ChatConnectionRegistry } from './connection-registry' + +import { useLogger } from '@guiiai/logg' + +import { createChatBroadcastMessage, parseChatBroadcastMessage } from '../../utils/chat-broadcast' +import { userChatBroadcastRedisKey } from '../../utils/redis-keys' + +const log = useLogger('chat-ws').useGlobalConfig() + +/** + * Cross-instance chat broadcast coordinator. + */ +export interface ChatBroadcastCoordinator { + /** Subscribes this process to the user's Redis channel. */ + ensureSubscribed: (userId: string) => void + /** Unsubscribes once this process has no local devices for the user. */ + maybeUnsubscribe: (userId: string) => void + /** Publishes a validated notification for other instances to fan out locally. */ + publish: (userId: string, payload: ChatBroadcastPayload) => void +} + +export interface ChatBroadcastCoordinatorOptions { + /** Redis connection used for publish and duplicate subscriber creation. */ + redis: Redis + /** Local registry that receives messages from other instances. */ + registry: ChatConnectionRegistry + /** Stable per-process id used to skip self-published messages. */ + instanceId: string +} + +/** + * Creates the Redis Pub/Sub coordinator for chat websocket notifications. + * + * Use when: + * - Local device fanout must also notify devices connected to other API instances. + * + * Expects: + * - Redis Pub/Sub is used only as a best-effort notification channel. + * - Durable chat truth remains in `ChatService` and clients can recover with `pullMessages`. + * + * Returns: + * - A small coordinator for subscribe, unsubscribe, and publish operations. + */ +export function createChatBroadcastCoordinator(options: ChatBroadcastCoordinatorOptions): ChatBroadcastCoordinator { + // Dedicated subscriber connection (ioredis requires a separate connection for subscribe mode). + const sub = options.redis.duplicate() + + sub.on('message', (_channel: string, message: string) => { + try { + const data = parseChatBroadcastMessage(message) + // Skip messages we ourselves published. ioredis pub/sub delivers to + // every subscriber, including the publishing connection — without + // this filter the publisher's local peers would receive each message + // twice (once via in-process broadcastToLocalDevices, once via the + // sub callback) and the sender's own ctx would receive an unwanted + // echo. + if (data.originInstanceId === options.instanceId) + return + // Cross-instance delivery: hand off to local peers of this user. + // No excludeCtx because the sender lives on a different instance. + options.registry.emitNewMessages(data.userId, null, data.payload) + } + catch (err) { + log.withError(err).error('Failed to parse broadcast message') + } + }) + + return { + ensureSubscribed(userId) { + const channel = userChatBroadcastRedisKey(userId) + sub.subscribe(channel).catch((err) => { + log.withError(err).error('Failed to subscribe to broadcast channel') + }) + }, + + maybeUnsubscribe(userId) { + if (options.registry.hasUser(userId)) + return + + const channel = userChatBroadcastRedisKey(userId) + sub.unsubscribe(channel).catch((err) => { + log.withError(err).error('Failed to unsubscribe from broadcast channel') + }) + }, + + publish(userId, payload) { + const channel = userChatBroadcastRedisKey(userId) + const message = createChatBroadcastMessage(userId, payload, options.instanceId) + options.redis.publish(channel, JSON.stringify(message)).catch((err) => { + log.withError(err).error('Failed to publish broadcast message') + }) + }, + } +} diff --git a/apps/server/src/routes/chat-ws/connection-registry.ts b/apps/server/src/routes/chat-ws/connection-registry.ts new file mode 100644 index 000000000..1fb288336 --- /dev/null +++ b/apps/server/src/routes/chat-ws/connection-registry.ts @@ -0,0 +1,78 @@ +import type { HonoWsInvocableEventContext } from '../../libs/eventa-hono-adapter' +import type { ChatBroadcastPayload } from '../../utils/chat-broadcast' + +import { newMessages } from '@proj-airi/server-sdk-shared' + +/** + * In-process websocket connection registry keyed by authenticated user id. + */ +export interface ChatConnectionRegistry { + /** Adds one websocket Eventa context for the user. */ + add: (userId: string, ctx: HonoWsInvocableEventContext) => void + /** Removes one websocket Eventa context and deletes the user bucket when empty. */ + remove: (userId: string, ctx: HonoWsInvocableEventContext) => void + /** Returns whether this process still has local connections for the user. */ + hasUser: (userId: string) => boolean + /** Counts all local websocket connections across users for metrics export. */ + activeCount: () => number + /** Emits `chat:new-messages` to all local user devices except an optional sender context. */ + emitNewMessages: (userId: string, excludeCtx: HonoWsInvocableEventContext | null, payload: ChatBroadcastPayload) => void +} + +/** + * Creates a local connection registry for chat websocket peers. + * + * Use when: + * - A chat websocket runtime needs local device fanout. + * - Engagement metrics need an active connection count. + * + * Expects: + * - Contexts belong to the same process and are removed on disconnect. + * + * Returns: + * - A mutable registry scoped to one chat websocket runtime. + */ +export function createChatConnectionRegistry(): ChatConnectionRegistry { + const userConnections = new Map>() + + return { + add(userId, ctx) { + let conns = userConnections.get(userId) + if (!conns) { + conns = new Set() + userConnections.set(userId, conns) + } + conns.add(ctx) + }, + + remove(userId, ctx) { + const conns = userConnections.get(userId) + if (!conns) + return + conns.delete(ctx) + if (conns.size === 0) + userConnections.delete(userId) + }, + + hasUser(userId) { + return userConnections.has(userId) + }, + + activeCount() { + let total = 0 + for (const conns of userConnections.values()) + total += conns.size + return total + }, + + emitNewMessages(userId, excludeCtx, payload) { + const conns = userConnections.get(userId) + if (!conns) + return + for (const ctx of conns) { + if (ctx !== excludeCtx) + ctx.emit(newMessages, payload) + } + }, + } +} diff --git a/apps/server/src/routes/chat-ws/index.ts b/apps/server/src/routes/chat-ws/index.ts index 932fcaa5d..eb700bc57 100644 --- a/apps/server/src/routes/chat-ws/index.ts +++ b/apps/server/src/routes/chat-ws/index.ts @@ -1,172 +1,67 @@ import type Redis from 'ioredis' -import type { HonoWsInvocableEventContext } from '../../libs/eventa-hono-adapter' import type { EngagementMetrics } from '../../otel' import type { ChatService } from '../../services/domain/chats' import { useLogger } from '@guiiai/logg' -import { defineInvokeHandler } from '@moeru/eventa' -import { newMessages, pullMessages, sendMessages } from '@proj-airi/server-sdk-shared' import { createPeerHooks, wsDisconnectedEvent } from '../../libs/eventa-hono-adapter' -import { createChatBroadcastMessage, parseChatBroadcastMessage } from '../../utils/chat-broadcast' -import { userChatBroadcastRedisKey } from '../../utils/redis-keys' +import { createChatBroadcastCoordinator } from './broadcast' +import { createChatConnectionRegistry } from './connection-registry' +import { registerChatRpcHandlers } from './rpc' const log = useLogger('chat-ws').useGlobalConfig() -// --------------------------------------------------------------------------- -// Local connection registry (per-process) -// --------------------------------------------------------------------------- - -const userConnections = new Map>() - -function addConnection(userId: string, ctx: HonoWsInvocableEventContext) { - let conns = userConnections.get(userId) - if (!conns) { - conns = new Set() - userConnections.set(userId, conns) - } - conns.add(ctx) -} - -function removeConnection(userId: string, ctx: HonoWsInvocableEventContext) { - const conns = userConnections.get(userId) - if (conns) { - conns.delete(ctx) - if (conns.size === 0) - userConnections.delete(userId) - } -} - -function broadcastToLocalDevices(userId: string, excludeCtx: HonoWsInvocableEventContext | null, event: any, payload: any) { - const conns = userConnections.get(userId) - if (!conns) - return - for (const ctx of conns) { - if (ctx !== excludeCtx) { - ctx.emit(event, payload) - } - } -} - +/** + * Creates websocket handlers for chat sync RPC and message fanout. + * + * Use when: + * - Mounting `/ws/chat` after bearer-token auth has resolved a user id. + * + * Expects: + * - `instanceId` is stable for this process so Redis echo suppression works. + * - Redis Pub/Sub is used only for best-effort cross-instance notification. + * + * Returns: + * - A per-user Hono websocket setup function. + */ export function createChatWsHandlers( chatService: ChatService, redis: Redis, instanceId: string, metrics?: EngagementMetrics | null, ) { - // TODO: Separate connection lifecycle, cross-instance broadcast, and RPC orchestration into smaller modules. - // This file is still acting as both transport adapter and chat delivery coordinator. - // Dedicated subscriber connection (ioredis requires a separate connection for subscribe mode) - const sub = redis.duplicate() + const registry = createChatConnectionRegistry() + const broadcast = createChatBroadcastCoordinator({ redis, registry, instanceId }) // Pull-based active-connection gauge: walk the local registry on each // export interval and report the actual live count. Registered exactly // once per process here (factory runs once via injeca); duplicate // registration would double-count. metrics?.wsConnectionsActive.addCallback((result) => { - let total = 0 - for (const conns of userConnections.values()) - total += conns.size - result.observe(total) + result.observe(registry.activeCount()) }) - sub.on('message', (_channel: string, message: string) => { - try { - const data = parseChatBroadcastMessage(message) - // Skip messages we ourselves published. ioredis pub/sub delivers to - // every subscriber, including the publishing connection — without - // this filter the publisher's local peers would receive each message - // twice (once via in-process broadcastToLocalDevices, once via the - // sub callback) and the sender's own ctx would receive an unwanted - // echo. - if (data.originInstanceId === instanceId) - return - // Cross-instance delivery: hand off to local peers of this user. - // No excludeCtx because the sender lives on a different instance. - broadcastToLocalDevices(data.userId, null, newMessages, data.payload) - } - catch (err) { - log.withError(err).error('Failed to parse broadcast message') - } - }) - - /** Subscribe to a user's broadcast channel when they first connect on this instance. */ - function ensureSubscribed(userId: string) { - const channel = userChatBroadcastRedisKey(userId) - sub.subscribe(channel).catch((err) => { - log.withError(err).error('Failed to subscribe to broadcast channel') - }) - } - - /** Unsubscribe when the user has no more connections on this instance. */ - function maybeUnsubscribe(userId: string) { - if (!userConnections.has(userId)) { - const channel = userChatBroadcastRedisKey(userId) - sub.unsubscribe(channel).catch((err) => { - log.withError(err).error('Failed to unsubscribe from broadcast channel') - }) - } - } - - /** Publish a broadcast message so other instances can deliver it. */ - function publishBroadcast(userId: string, payload: Parameters[1]) { - const channel = userChatBroadcastRedisKey(userId) - const message = createChatBroadcastMessage(userId, payload, instanceId) - redis.publish(channel, JSON.stringify(message)).catch((err) => { - log.withError(err).error('Failed to publish broadcast message') - }) - } - return function setupPeer(userId: string) { const { hooks } = createPeerHooks({ onContext: (ctx) => { - addConnection(userId, ctx) - ensureSubscribed(userId) + registry.add(userId, ctx) + broadcast.ensureSubscribed(userId) log.withFields({ userId }).log('WS connected') ctx.on(wsDisconnectedEvent, () => { - removeConnection(userId, ctx) - maybeUnsubscribe(userId) + registry.remove(userId, ctx) + broadcast.maybeUnsubscribe(userId) log.withFields({ userId }).log('WS disconnected') }) - // RPC: send messages - defineInvokeHandler(ctx, sendMessages, async (req) => { - log.withFields({ userId, chatId: req!.chatId, count: req!.messages.length }).log('sendMessages') - const result = await chatService.pushMessages(userId, req!.chatId, req!.messages) - - // Fetch the wire messages for broadcast - const wireMessages = await chatService.pullMessages(userId, req!.chatId, result.fromSeq - 1, result.toSeq - result.fromSeq + 1) - const broadcastPayload = { - chatId: req!.chatId, - messages: wireMessages.messages, - fromSeq: result.fromSeq, - toSeq: result.toSeq, - } - - // Broadcast to all chat members (not just the sender) - const members = await chatService.getMembers(req!.chatId) - const memberUserIds = members - .filter(m => m.memberType === 'user' && m.userId != null) - .map(m => m.userId!) - - for (const memberUserId of memberUserIds) { - // For the sender, exclude the current connection - const excludeCtx = memberUserId === userId ? ctx : null - broadcastToLocalDevices(memberUserId, excludeCtx, newMessages, broadcastPayload) - // Cross-instance broadcast via Redis pub/sub - publishBroadcast(memberUserId, broadcastPayload) - } - - metrics?.wsMessagesSent.add(wireMessages.messages.length) - return { seq: result.seq } - }) - - // RPC: pull messages - defineInvokeHandler(ctx, pullMessages, async (req) => { - log.withFields({ userId, chatId: req!.chatId, afterSeq: req!.afterSeq }).log('pullMessages') - return chatService.pullMessages(userId, req!.chatId, req!.afterSeq, req!.limit) + registerChatRpcHandlers({ + ctx, + userId, + chatService, + registry, + broadcast, + metrics, }) }, }) diff --git a/apps/server/src/routes/chat-ws/rpc.ts b/apps/server/src/routes/chat-ws/rpc.ts new file mode 100644 index 000000000..367a38b8a --- /dev/null +++ b/apps/server/src/routes/chat-ws/rpc.ts @@ -0,0 +1,74 @@ +import type { HonoWsInvocableEventContext } from '../../libs/eventa-hono-adapter' +import type { EngagementMetrics } from '../../otel' +import type { ChatService } from '../../services/domain/chats' +import type { ChatBroadcastCoordinator } from './broadcast' +import type { ChatConnectionRegistry } from './connection-registry' + +import { useLogger } from '@guiiai/logg' +import { defineInvokeHandler } from '@moeru/eventa' +import { pullMessages, sendMessages } from '@proj-airi/server-sdk-shared' + +const log = useLogger('chat-ws').useGlobalConfig() + +export interface RegisterChatRpcHandlersOptions { + /** Eventa websocket context for the connected peer. */ + ctx: HonoWsInvocableEventContext + /** Authenticated user that owns this websocket connection. */ + userId: string + /** Domain service that persists and reads chat messages. */ + chatService: ChatService + /** Local websocket registry for same-instance fanout. */ + registry: ChatConnectionRegistry + /** Redis coordinator for cross-instance fanout. */ + broadcast: ChatBroadcastCoordinator + /** Optional engagement metrics. */ + metrics?: EngagementMetrics | null +} + +/** + * Registers chat Eventa RPC handlers on one websocket context. + * + * Use when: + * - A peer context has just been created by the Hono Eventa adapter. + * + * Expects: + * - `chatService` enforces membership and message sequencing. + * + * Returns: + * - Nothing; handlers are attached to the provided context. + */ +export function registerChatRpcHandlers(options: RegisterChatRpcHandlersOptions): void { + const { ctx, userId, chatService, registry, broadcast, metrics } = options + + defineInvokeHandler(ctx, sendMessages, async (req) => { + log.withFields({ userId, chatId: req!.chatId, count: req!.messages.length }).log('sendMessages') + const result = await chatService.pushMessages(userId, req!.chatId, req!.messages) + + const wireMessages = await chatService.pullMessages(userId, req!.chatId, result.fromSeq - 1, result.toSeq - result.fromSeq + 1) + const broadcastPayload = { + chatId: req!.chatId, + messages: wireMessages.messages, + fromSeq: result.fromSeq, + toSeq: result.toSeq, + } + + const members = await chatService.getMembers(req!.chatId) + const memberUserIds = members + .filter(m => m.memberType === 'user' && m.userId != null) + .map(m => m.userId!) + + for (const memberUserId of memberUserIds) { + const excludeCtx = memberUserId === userId ? ctx : null + registry.emitNewMessages(memberUserId, excludeCtx, broadcastPayload) + broadcast.publish(memberUserId, broadcastPayload) + } + + metrics?.wsMessagesSent.add(wireMessages.messages.length) + return { seq: result.seq } + }) + + defineInvokeHandler(ctx, pullMessages, async (req) => { + log.withFields({ userId, chatId: req!.chatId, afterSeq: req!.afterSeq }).log('pullMessages') + return chatService.pullMessages(userId, req!.chatId, req!.afterSeq, req!.limit) + }) +}