feat(server): chat broadcast utilities for message handling

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<typeof createChatBroadcastMessage>[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')
     })

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<string, unknown>
+  const payload = message.payload
+
+  if (!payload || typeof payload !== 'object')
+    throw new TypeError('chat broadcast payload must be an object')
+
+  const payloadRecord = payload as Record<string, unknown>
+
+  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
+}

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')
+  })
+})