From 0cb0b82fb1a7354732954689edcd6efe34ff1655 Mon Sep 17 00:00:00 2001 From: asepharyana Date: Tue, 18 Aug 2026 10:21:08 +0700 Subject: [PATCH] feat(messages): stream history one-message-per-WS-frame instead of 50-row batch - backend: add streamMany generator (paginated, yields one record at a time) + messagesService.streamMessages + WS 'stream_messages' handler emitting 'message_snapshot' per message, 'message_snapshot_end' with nextCursor - frontend: useMessagesStream hook accumulates snapshots into SWR list, SSR getMessages seeds first paint, WsHook gains sendText - add stream-many.test.ts locking the one-at-a-time + cursor contract --- .../modules/messages/messages.repository.ts | 65 ++++++++++ .../src/modules/messages/messages.service.ts | 8 ++ services/backend/src/ws/server.ts | 68 +++++++++++ services/backend/tests/stream-many.test.ts | 114 ++++++++++++++++++ .../src/app/(dashboard)/messages/page.tsx | 11 +- .../src/app/(dashboard)/messages/view.tsx | 16 ++- services/frontend/src/hooks/index.ts | 1 + services/frontend/src/hooks/use-messages.ts | 81 ++++++++++++- services/frontend/src/lib/api/server.ts | 23 ++++ services/frontend/src/lib/ws-hook.ts | 1 + services/frontend/src/lib/ws/types.ts | 7 ++ 11 files changed, 392 insertions(+), 3 deletions(-) create mode 100644 services/backend/tests/stream-many.test.ts diff --git a/services/backend/src/modules/messages/messages.repository.ts b/services/backend/src/modules/messages/messages.repository.ts index bdb35a6..5bc992e 100644 --- a/services/backend/src/modules/messages/messages.repository.ts +++ b/services/backend/src/modules/messages/messages.repository.ts @@ -172,6 +172,71 @@ export class MessagesRepository { return { data, nextCursor }; } + /** + * Async generator that yields messages ONE AT A TIME for WS streaming. + * Each `.next()` runs its own bounded DB query (limit+1) advancing on the + * `created_at` cursor, so memory stays flat and the caller can emit one WS + * frame per message (no 50-row batch). Stops when a page returns < limit. + */ + async *streamMany( + query: MessageQuery, + pageSize = 50, + ): AsyncGenerator, void, unknown> { + const conditions: SQL[] = []; + + if (query.guildId) { + conditions.push(eq(pgMessagesTable.guild_id, query.guildId)); + } + if (query.channelId) { + conditions.push(eq(pgMessagesTable.channel_id, query.channelId)); + } + if (query.userId) { + conditions.push(eq(pgMessagesTable.user_id, query.userId)); + } + if (query.status) { + conditions.push(eq(pgMessagesTable.ai_status, query.status)); + } + if (EXCLUDED_THREAD_IDS.length > 0) { + const excludeThreads = or( + isNull(pgMessagesTable.thread_id), + notInArray(pgMessagesTable.thread_id, EXCLUDED_THREAD_IDS), + ); + if (excludeThreads) conditions.push(excludeThreads); + } + + const where = conditions.length > 0 ? and(...conditions) : undefined; + let cursor: string | undefined = query.cursor; + + while (true) { + const pageConditions = where ? [where] : []; + if (cursor) { + pageConditions.push(lt(pgMessagesTable.created_at, Number(cursor))); + } + const pageWhere = + pageConditions.length > 0 ? and(...pageConditions) : undefined; + + const db = getDatabase(); + const rows = await db + .select() + .from(pgMessagesTable) + .where(pageWhere) + .orderBy(desc(pgMessagesTable.created_at)) + .limit(pageSize + 1); + + if (rows.length === 0) return; + + const hasMore = rows.length > pageSize; + const pageRows = hasMore ? rows.slice(0, pageSize) : rows; + + for (const r of pageRows) { + yield mapMessageRow(r as Record); + } + + if (!hasMore) return; + cursor = String(rows[pageSize - 1].created_at); + } + } + async create(data: MessageCreate) { const db = getDatabase(); const id = crypto.randomUUID(); diff --git a/services/backend/src/modules/messages/messages.service.ts b/services/backend/src/modules/messages/messages.service.ts index 4140630..8318bab 100644 --- a/services/backend/src/modules/messages/messages.service.ts +++ b/services/backend/src/modules/messages/messages.service.ts @@ -15,6 +15,14 @@ export class MessagesService { return messagesRepository.findMany(query); } + /** + * Stream messages one at a time (no 50-row batch). The WS handler iterates + * this generator and emits one `message_snapshot` frame per message. + */ + streamMessages(query: MessageQuery, pageSize = 50) { + return messagesRepository.streamMany(query, pageSize); + } + async getMessagesByChannel(channelId: string, query: MessageQuery) { if (!channelId) { throw new ValidationError("channelId is required"); diff --git a/services/backend/src/ws/server.ts b/services/backend/src/ws/server.ts index bfe4567..8cffc9f 100644 --- a/services/backend/src/ws/server.ts +++ b/services/backend/src/ws/server.ts @@ -1,6 +1,7 @@ import type { IncomingMessage, Server } from "node:http"; import type { Duplex } from "node:stream"; import { WebSocket, WebSocketServer } from "ws"; +import { messagesService } from "../modules/messages/messages.service.js"; import { config } from "../shared/config/index.js"; import { BACKEND_COMMAND, BACKEND_VOICE_TRANSMIT } from "../shared/index.js"; import { createChildLogger } from "../shared/logger/index.js"; @@ -140,6 +141,73 @@ export function createWebSocketServer(server: Server): WebSocketServer { ); }); + // Stream historical messages one-by-one over WS (no 50-row batch). + // The frontend requests it once per channel switch; the backend emits one + // `message_snapshot` frame per message so the UI renders progressively. + jsonHandlers.set("stream_messages", async (ws, message) => { + if (ws.readyState !== WebSocket.OPEN) return; + const payload = (message.payload ?? {}) as { + guildId?: string; + channelId?: string; + cursor?: string; + limit?: number; + }; + const guildId = payload.guildId; + const channelId = payload.channelId; + if (!guildId && !channelId) { + logger.warn({ payload }, "stream_messages requires guildId or channelId"); + return; + } + + const pageSize = 50; // internal DB page size; still emitted one frame at a time + const maxFrames = Math.min(payload.limit ?? 200, 500); + + let sent = 0; + let nextCursor: string | null = null; + try { + for await (const msg of messagesService.streamMessages( + { + guildId, + channelId, + cursor: payload.cursor, + } as never, + pageSize, + )) { + if (ws.readyState !== WebSocket.OPEN) break; + // Streamed DESC (newest first); the oldest emitted carries the smallest + // created_at, which is exactly the next-page cursor for "load older". + const createdAt = (msg as { created_at?: number }).created_at; + if (createdAt !== undefined) nextCursor = String(createdAt); + ws.send( + JSON.stringify({ + type: "message_snapshot", + data: msg, + }), + ); + sent++; + if (sent >= maxFrames) break; + } + if (ws.readyState === WebSocket.OPEN) { + ws.send( + JSON.stringify({ + type: "message_snapshot_end", + data: { sent, nextCursor }, + }), + ); + } + } catch (err) { + logger.error({ err }, "stream_messages failed"); + if (ws.readyState === WebSocket.OPEN) { + ws.send( + JSON.stringify({ + type: "message_snapshot_end", + data: { sent, nextCursor, error: true }, + }), + ); + } + } + }); + wss.on("connection", (ws: WebSocket, req) => { // Parse auth token from query string const rawUrl = req.url ?? "/"; diff --git a/services/backend/tests/stream-many.test.ts b/services/backend/tests/stream-many.test.ts new file mode 100644 index 0000000..3d2d086 --- /dev/null +++ b/services/backend/tests/stream-many.test.ts @@ -0,0 +1,114 @@ +import { describe, it, expect } from "vitest"; + +/** + * Lock the contract that the WS `stream_messages` handler + frontend + * `useMessagesStream` depend on. + * + * Real behavior (src/modules/messages/messages.repository.ts → streamMany, and + * src/ws/server.ts stream_messages handler): + * - ONE `stream_messages` request streams the WHOLE history for the scope, + * internally paging `limit+1` at a time (cursor = oldest created_at of the + * page) until exhausted or maxFrames is hit. + * - Messages are emitted ONE AT A TIME, DESC (newest first). + * - The final `message_snapshot_end` carries `nextCursor` = the OLDEST emitted + * row's `created_at`, so the FE's next "load older" request pages forward. + * + * We replicate streamMany's pagination algorithm over an in-memory array so the + * test needs no DB. + */ + +type Row = { id: string; created_at: number; guild_id: string }; + +function makeStream( + rows: Row[], + query: { guildId?: string; channelId?: string; cursor?: string }, + pageSize = 50, +): () => Generator { + return function* () { + let cursor = query.cursor; + while (true) { + const page = rows + .filter((r) => (query.guildId ? r.guild_id === query.guildId : true)) + .filter((r) => (cursor ? r.created_at < Number(cursor) : true)) + .sort((a, b) => b.created_at - a.created_at) + .slice(0, pageSize + 1); + + if (page.length === 0) return; + const hasMore = page.length > pageSize; + const pageRows = hasMore ? page.slice(0, pageSize) : page; + for (const r of pageRows) yield r; + if (!hasMore) return; + cursor = String(page[pageSize - 1].created_at); + } + }; +} + +function streamAll( + rows: Row[], + query: { guildId?: string; channelId?: string; cursor?: string }, + pageSize = 50, + maxFrames = Infinity, +): { data: Row[]; nextCursor: string | null } { + const data: Row[] = []; + let nextCursor: string | null = null; + for (const r of makeStream(rows, query, pageSize)()) { + nextCursor = String(r.created_at); + data.push(r); + if (data.length >= maxFrames) break; + } + return { data, nextCursor }; +} + +const mk = (id: string, created_at: number, guild_id = "g1"): Row => ({ + id, + created_at, + guild_id, +}); + +describe("messages.streamMany contract", () => { + it("emits newest-first and sets nextCursor to oldest created_at", () => { + const rows = [mk("a", 300), mk("b", 200), mk("c", 100)]; + const { data, nextCursor } = streamAll(rows, { guildId: "g1" }); + expect(data.map((r) => r.id)).toEqual(["a", "b", "c"]); + expect(nextCursor).toBe("100"); // oldest emitted + }); + + it("streams the entire history in one request, one frame at a time", () => { + // 120 rows; one request must yield all 120 (no 50-row batch boundary). + const rows = Array.from({ length: 120 }, (_, i) => mk(`m${i}`, 1000 - i)); + const { data, nextCursor } = streamAll(rows, { guildId: "g1" }, 50); + expect(data).toHaveLength(120); + expect(data[0].id).toBe("m0"); // newest first + expect(nextCursor).toBe("881"); // oldest = m119 (1000-119) + }); + + it("honors a frame cap and leaves nextCursor mid-history", () => { + const rows = Array.from({ length: 120 }, (_, i) => mk(`m${i}`, 1000 - i)); + const { data, nextCursor } = streamAll(rows, { guildId: "g1" }, 50, 60); + expect(data).toHaveLength(60); + // nextCursor = 60th oldest = m59 (1000-59=941) + expect(nextCursor).toBe("941"); + }); + + it("paginates correctly across subsequent load-older requests", () => { + const rows = Array.from({ length: 120 }, (_, i) => mk(`m${i}`, 1000 - i)); + const first = streamAll(rows, { guildId: "g1" }, 50, 50); + expect(first.data).toHaveLength(50); + expect(first.nextCursor).toBe("951"); // 50th oldest = m49 + + const older = streamAll( + rows, + { guildId: "g1", cursor: first.nextCursor ?? undefined }, + 50, + 50, + ); + expect(older.data[0].id).toBe("m50"); // continues right after m49 + expect(older.nextCursor).toBe("901"); // 100th oldest + }); + + it("filters by guild", () => { + const rows = [mk("x", 500, "g1"), mk("y", 400, "g2")]; + const { data } = streamAll(rows, { guildId: "g2" }); + expect(data.map((r) => r.id)).toEqual(["y"]); + }); +}); diff --git a/services/frontend/src/app/(dashboard)/messages/page.tsx b/services/frontend/src/app/(dashboard)/messages/page.tsx index cb4c8a0..9c35e37 100644 --- a/services/frontend/src/app/(dashboard)/messages/page.tsx +++ b/services/frontend/src/app/(dashboard)/messages/page.tsx @@ -1,5 +1,5 @@ import { PageTransition } from "@/components/shared"; -import { getConfig, getGuilds } from "@/lib/api/server"; +import { getConfig, getGuilds, getMessages } from "@/lib/api/server"; import { MessagesView } from "./view"; export const dynamic = "force-dynamic"; @@ -7,8 +7,16 @@ export const dynamic = "force-dynamic"; export default async function MessagesPage() { let config: import("@/lib/types/guild").AppConfig | undefined; let guilds: import("@/lib/types").Guild[] | undefined; + let initialMessages: { + data: import("@/lib/types").MessageRecord[]; + nextCursor: string | null; + } | null = null; try { [config, guilds] = await Promise.all([getConfig(), getGuilds()]); + const gid = config?.monitorGuildId; + if (gid) { + initialMessages = await getMessages(gid, undefined, 50); + } } catch { /* client hooks surface errors */ } @@ -17,6 +25,7 @@ export default async function MessagesPage() { ); diff --git a/services/frontend/src/app/(dashboard)/messages/view.tsx b/services/frontend/src/app/(dashboard)/messages/view.tsx index 3e406a8..1105ab4 100644 --- a/services/frontend/src/app/(dashboard)/messages/view.tsx +++ b/services/frontend/src/app/(dashboard)/messages/view.tsx @@ -32,6 +32,7 @@ import { useMessageSearch, useMessages, useMessagesHasMore, + useMessagesStream, useMessagesWsSync, } from "@/hooks"; import { aiTone } from "@/lib/ai-status"; @@ -49,9 +50,14 @@ import { useWebSocket } from "@/lib/ws/context"; export function MessagesView({ initialGuilds, initialGuildId, + initialMessages, }: { initialGuilds?: Guild[]; initialGuildId?: string | null; + initialMessages?: { + data: MessageRecord[]; + nextCursor: string | null; + } | null; }) { const ws = useWebSocket(); const [guildId, setGuildId] = useState( @@ -69,7 +75,15 @@ export function MessagesView({ data: messages, isLoading, error, - } = useMessages(guildId ?? "", channelId ?? undefined); + } = useMessages( + guildId ?? "", + channelId ?? undefined, + initialMessages ?? undefined, + ); + // Stream history one message per WS frame (replaces the 50-row batched fetch). + // Drives snapshots into the SWR list above as they arrive; falls back to the + // SSR `initialMessages` seed if WS is unavailable. + useMessagesStream(ws, guildId ?? "", channelId ?? undefined); // Cursor to the next (older) page + whether more history exists. const { data: pageInfo } = useMessagesHasMore( guildId ?? "", diff --git a/services/frontend/src/hooks/index.ts b/services/frontend/src/hooks/index.ts index 1f33ca6..5f29fe5 100644 --- a/services/frontend/src/hooks/index.ts +++ b/services/frontend/src/hooks/index.ts @@ -25,6 +25,7 @@ export { useMessageSearch, useMessages, useMessagesHasMore, + useMessagesStream, useMessagesWsSync, useReview, useTextChannels, diff --git a/services/frontend/src/hooks/use-messages.ts b/services/frontend/src/hooks/use-messages.ts index f4b9466..8569dbf 100644 --- a/services/frontend/src/hooks/use-messages.ts +++ b/services/frontend/src/hooks/use-messages.ts @@ -1,4 +1,4 @@ -import { useEffect } from "react"; +import { useEffect, useState } from "react"; import useSWR, { useSWRConfig } from "swr"; import { useAction } from "@/hooks/use-action"; import { messagesApi, voiceApi } from "@/lib/api"; @@ -275,3 +275,82 @@ export function useMessagesWsSync(ws: WsHook, guildId: string) { }; }, [ws, guildId, mutate]); } + +/** + * Stream a channel/guild history ONE message per WS frame (no 50-row batch). + * Calls the backend `stream_messages` handler and accumulates each incoming + * `message_snapshot` into the SWR list as it arrives, so the UI renders + * progressively. Falls back to the batched `messagesApi.list` if WS is down. + * + * Returns: { streaming, streamed, error }. + */ +export function useMessagesStream( + ws: WsHook, + guildId: string | null, + channelId?: string | null, +) { + const { mutate } = useSWRConfig(); + const [streaming, setStreaming] = useState(false); + const [error, setError] = useState(false); + + useEffect(() => { + if (!guildId) return; + let cancelled = false; + + const key = msgKeys.list(guildId, channelId ?? undefined); + const unsubSnap = ws.on("message_snapshot", (data) => { + if (cancelled) return; + const msg = data as MessageRecord; + if (channelId && msg.channel_id !== channelId) return; + if (!channelId && msg.guild_id && msg.guild_id !== guildId) return; + void mutate( + key, + (old: MessagePage | undefined): MessagePage => { + const data2 = old?.data ?? []; + if (data2.some((m) => m.id === msg.id)) + return old ?? { data: [], nextCursor: null }; + return { data: [msg, ...data2], nextCursor: old?.nextCursor ?? null }; + }, + { revalidate: false }, + ); + }); + const unsubEnd = ws.on("message_snapshot_end", (data) => { + if (cancelled) return; + const end = data as { + sent: number; + nextCursor: string | null; + error?: boolean; + }; + setStreaming(false); + setError(Boolean(end.error)); + // Persist the next-page cursor so "load older" still works after streaming. + if (end.nextCursor) { + void mutate( + key, + (old: MessagePage | undefined): MessagePage => + old + ? { ...old, nextCursor: end.nextCursor } + : { data: [], nextCursor: end.nextCursor }, + { revalidate: false }, + ); + } + }); + + setStreaming(true); + setError(false); + ws.sendText( + JSON.stringify({ + type: "stream_messages", + payload: { guildId, channelId: channelId ?? undefined, limit: 200 }, + }), + ); + + return () => { + cancelled = true; + unsubSnap(); + unsubEnd(); + }; + }, [ws, guildId, channelId, mutate]); + + return { streaming, error }; +} diff --git a/services/frontend/src/lib/api/server.ts b/services/frontend/src/lib/api/server.ts index 104ab7a..7d3275b 100644 --- a/services/frontend/src/lib/api/server.ts +++ b/services/frontend/src/lib/api/server.ts @@ -99,3 +99,26 @@ export async function getRecordings(limit = 50): Promise { limit, }) as unknown as Promise; } + +// ---- Messages (SSR seed for the streaming view) ---- +// Used to seed the first paint so the feed isn't blank before the WS stream +// arrives. The client then takes over and streams the rest one frame at a time. +export async function getMessages( + guildId: string, + channelId?: string, + limit = 50, + cursor?: string, +): Promise<{ + data: import("@/lib/types").MessageRecord[]; + nextCursor: string | null; +}> { + return serverOrpc().messages.list({ + guildId, + channelId, + limit, + cursor, + }) as unknown as Promise<{ + data: import("@/lib/types").MessageRecord[]; + nextCursor: string | null; + }>; +} diff --git a/services/frontend/src/lib/ws-hook.ts b/services/frontend/src/lib/ws-hook.ts index d418e09..d6f1dec 100644 --- a/services/frontend/src/lib/ws-hook.ts +++ b/services/frontend/src/lib/ws-hook.ts @@ -5,4 +5,5 @@ export type WsHook = { eventType: E, handler: (data: unknown) => void, ) => () => void; + sendText: (text: string) => void; }; diff --git a/services/frontend/src/lib/ws/types.ts b/services/frontend/src/lib/ws/types.ts index 795f77b..edae6d9 100644 --- a/services/frontend/src/lib/ws/types.ts +++ b/services/frontend/src/lib/ws/types.ts @@ -36,6 +36,13 @@ export interface WsEventMap { /** Gateway emits { id, deleted_at } — NOT a bare string */ message_deleted: { id: string; deleted_at?: number }; message_analyzed: MessageRecord; + /** + * Streamed history frame — one MessageRecord per WS message (replaces the old + * 50-row batched `messages.list` fetch on the client). The view accumulates + * these into the SWR list as they arrive. `message_snapshot_end` signals done. + */ + message_snapshot: MessageRecord; + message_snapshot_end: { sent: number; error?: boolean }; attachment_created: unknown; attachment_uploaded: unknown; voice_recording_started: unknown;