From a515d20091685e27be6f06a46c4182b2c8627d19 Mon Sep 17 00:00:00 2001 From: RainbowBird Date: Fri, 27 Mar 2026 18:44:30 +0800 Subject: [PATCH] feat(server): chat broadcast utilities for message handling --- apps/server/src/routes/chat-ws.ts | 21 +---- apps/server/src/utils/chat-broadcast.ts | 78 +++++++++++++++++++ .../src/utils/tests/chat-broadcast.test.ts | 73 +++++++++++++++++ 3 files changed, 155 insertions(+), 17 deletions(-) create mode 100644 apps/server/src/utils/chat-broadcast.ts create mode 100644 apps/server/src/utils/tests/chat-broadcast.test.ts diff --git a/apps/server/src/routes/chat-ws.ts b/apps/server/src/routes/chat-ws.ts index 906102093..b598da478 100644 --- a/apps/server/src/routes/chat-ws.ts +++ b/apps/server/src/routes/chat-ws.ts @@ -9,6 +9,7 @@ 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' const log = useLogger('chat-ws').useGlobalConfig() @@ -48,20 +49,6 @@ function broadcastToLocalDevices(userId: string, excludeCtx: HonoWsInvocableEven } } -// --------------------------------------------------------------------------- -// Redis pub/sub for cross-instance broadcast -// --------------------------------------------------------------------------- - -interface BroadcastMessage { - userId: string - payload: { - chatId: string - messages: any[] - fromSeq: number - toSeq: number - } -} - export function createChatWsHandlers( chatService: ChatService, redis: Redis, @@ -72,7 +59,7 @@ export function createChatWsHandlers( sub.on('message', (_channel: string, message: string) => { try { - const data: BroadcastMessage = JSON.parse(message) + const data = parseChatBroadcastMessage(message) // Deliver to all local connections of this user (no excludeCtx since the // sender is on a different instance) broadcastToLocalDevices(data.userId, null, newMessages, data.payload) @@ -101,9 +88,9 @@ export function createChatWsHandlers( } /** Publish a broadcast message so other instances can deliver it. */ - function publishBroadcast(userId: string, payload: BroadcastMessage['payload']) { + function publishBroadcast(userId: string, payload: Parameters[1]) { const channel = userChatBroadcastRedisKey(userId) - const message: BroadcastMessage = { userId, payload } + const message = createChatBroadcastMessage(userId, payload) redis.publish(channel, JSON.stringify(message)).catch((err) => { log.withError(err).error('Failed to publish broadcast message') }) diff --git a/apps/server/src/utils/chat-broadcast.ts b/apps/server/src/utils/chat-broadcast.ts new file mode 100644 index 000000000..fbb4d91fe --- /dev/null +++ b/apps/server/src/utils/chat-broadcast.ts @@ -0,0 +1,78 @@ +export interface ChatBroadcastPayload { + chatId: string + messages: unknown[] + fromSeq: number + toSeq: number +} + +export interface ChatBroadcastMessage { + userId: string + payload: ChatBroadcastPayload +} + +function assertNonEmptyString(value: unknown, fieldName: string): string { + if (typeof value !== 'string' || value.trim().length === 0) + throw new TypeError(`${fieldName} must be a non-empty string`) + + return value +} + +function assertFiniteNumber(value: unknown, fieldName: string): number { + if (typeof value !== 'number' || !Number.isFinite(value)) + throw new TypeError(`${fieldName} must be a finite number`) + + return value +} + +export function createChatBroadcastMessage( + userId: string, + payload: ChatBroadcastPayload, +): ChatBroadcastMessage { + return { + userId: assertNonEmptyString(userId, 'chat broadcast userId'), + payload: { + chatId: assertNonEmptyString(payload.chatId, 'chat broadcast payload.chatId'), + messages: assertMessages(payload.messages), + fromSeq: assertFiniteNumber(payload.fromSeq, 'chat broadcast payload.fromSeq'), + toSeq: assertFiniteNumber(payload.toSeq, 'chat broadcast payload.toSeq'), + }, + } +} + +export function parseChatBroadcastMessage(raw: string): ChatBroadcastMessage { + let parsed: unknown + try { + parsed = JSON.parse(raw) + } + catch (error) { + throw new TypeError('chat broadcast message is not valid JSON', { cause: error }) + } + + if (!parsed || typeof parsed !== 'object') + throw new TypeError('chat broadcast message must be an object') + + const message = parsed as Record + const payload = message.payload + + if (!payload || typeof payload !== 'object') + throw new TypeError('chat broadcast payload must be an object') + + const payloadRecord = payload as Record + + return createChatBroadcastMessage( + assertNonEmptyString(message.userId, 'chat broadcast userId'), + { + chatId: assertNonEmptyString(payloadRecord.chatId, 'chat broadcast payload.chatId'), + messages: assertMessages(payloadRecord.messages), + fromSeq: assertFiniteNumber(payloadRecord.fromSeq, 'chat broadcast payload.fromSeq'), + toSeq: assertFiniteNumber(payloadRecord.toSeq, 'chat broadcast payload.toSeq'), + }, + ) +} + +function assertMessages(value: unknown): unknown[] { + if (!Array.isArray(value)) + throw new TypeError('chat broadcast payload.messages must be an array') + + return value +} diff --git a/apps/server/src/utils/tests/chat-broadcast.test.ts b/apps/server/src/utils/tests/chat-broadcast.test.ts new file mode 100644 index 000000000..56867113d --- /dev/null +++ b/apps/server/src/utils/tests/chat-broadcast.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it } from 'vitest' + +import { createChatBroadcastMessage, parseChatBroadcastMessage } from '../chat-broadcast' + +describe('chat broadcast utils', () => { + it('creates a normalized broadcast message from validated inputs', () => { + expect(createChatBroadcastMessage('user-1', { + chatId: 'chat-1', + messages: [{ id: 'msg-1' }], + fromSeq: 3, + toSeq: 4, + })).toEqual({ + userId: 'user-1', + payload: { + chatId: 'chat-1', + messages: [{ id: 'msg-1' }], + fromSeq: 3, + toSeq: 4, + }, + }) + }) + + it('rejects invalid publish-side identifiers', () => { + expect(() => createChatBroadcastMessage('', { + chatId: 'chat-1', + messages: [], + fromSeq: 1, + toSeq: 1, + })).toThrow('chat broadcast userId must be a non-empty string') + }) + + it('parses a valid broadcast message payload', () => { + expect(parseChatBroadcastMessage(JSON.stringify({ + userId: 'user-2', + payload: { + chatId: 'chat-9', + messages: ['message'], + fromSeq: 9, + toSeq: 12, + }, + }))).toEqual({ + userId: 'user-2', + payload: { + chatId: 'chat-9', + messages: ['message'], + fromSeq: 9, + toSeq: 12, + }, + }) + }) + + it('rejects invalid json and malformed payloads', () => { + expect(() => parseChatBroadcastMessage('not-json')).toThrow('chat broadcast message is not valid JSON') + expect(() => parseChatBroadcastMessage(JSON.stringify({ + userId: {}, + payload: { + chatId: 'chat-1', + messages: [], + fromSeq: 1, + toSeq: 1, + }, + }))).toThrow('chat broadcast userId must be a non-empty string') + expect(() => parseChatBroadcastMessage(JSON.stringify({ + userId: 'user-1', + payload: { + chatId: 'chat-1', + messages: {}, + fromSeq: 1, + toSeq: 1, + }, + }))).toThrow('chat broadcast payload.messages must be an array') + }) +})