refactor(server): split chat ws runtime
This commit is contained in:
RainbowBird
@@ -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')
|
||||
})
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -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<string, Set<HonoWsInvocableEventContext>>()
|
||||
|
||||
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)
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -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<string, Set<HonoWsInvocableEventContext>>()
|
||||
|
||||
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<typeof createChatBroadcastMessage>[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,
|
||||
})
|
||||
},
|
||||
})
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user