import { gzipSync } from 'node:zlib'; import type { PairedTurnAttemptRecord, PairedTurnRecord } from './db.js'; import type { StatusSnapshot } from './status-dashboard.js'; import type { NewMessage, PairedTask, PairedTurnOutput } from './types.js'; import { buildWebDashboardRoomActivity, type WebDashboardRoomActivity, } from './web-dashboard-data.js'; type JsonResponse = ( value: unknown, init?: ResponseInit, request?: Request, ) => Response; export interface RoomsTimelineRouteDependencies { statusMaxAgeMs: number; readSnapshots: (maxAgeMs: number) => StatusSnapshot[]; loadLatestPairedTaskForChat: ( chatJid: string, ) => PairedTask | null | undefined; loadPairedTurnsForTask: (taskId: string) => PairedTurnRecord[]; loadLatestPairedTurnForTask: ( taskId: string, ) => PairedTurnRecord | null | undefined; loadPairedTurnAttempts: (turnId: string) => PairedTurnAttemptRecord[]; loadPairedTurnOutputs: (taskId: string) => PairedTurnOutput[]; loadRecentPairedTurnOutputsForChat: ( chatJid: string, limit: number, ) => PairedTurnOutput[]; loadRecentChatMessages: (chatJid: string, limit?: number) => NewMessage[]; } interface RoomsTimelineRouteContext extends RoomsTimelineRouteDependencies { url: URL; request: Request; jsonResponse: JsonResponse; } interface RoomsTimelineCache { key: string; builtAt: number; rawJson: string; gzipBuffer: Uint8Array; } const ROOMS_TIMELINE_BG_INTERVAL_MS = 2000; let roomsTimelineCache: RoomsTimelineCache | null = null; function parseRoomTimelinePath(pathname: string): string | null { const match = pathname.match(/^\/api\/rooms\/([^/]+)\/timeline$/); if (!match) return null; try { return decodeURIComponent(match[1]); } catch { return null; } } function isReadMethod(method: string): boolean { return method === 'GET' || method === 'HEAD'; } function latestSnapshotEntry( snapshots: StatusSnapshot[], jid: string, ): | { snapshot: StatusSnapshot; entry: StatusSnapshot['entries'][number] } | undefined { return snapshots .flatMap((snapshot) => snapshot.entries .filter((entry) => entry.jid === jid) .map((entry) => ({ snapshot, entry })), ) .sort((a, b) => b.snapshot.updatedAt.localeCompare(a.snapshot.updatedAt)) .at(0); } function buildRoomActivity( deps: RoomsTimelineRouteDependencies, snapshot: StatusSnapshot, entry: StatusSnapshot['entries'][number], ): WebDashboardRoomActivity { const pairedTask = deps.loadLatestPairedTaskForChat(entry.jid) ?? null; const turns = pairedTask ? deps.loadPairedTurnsForTask(pairedTask.id) : []; const attempts = turns.flatMap((turn) => deps.loadPairedTurnAttempts(turn.turn_id), ); return buildWebDashboardRoomActivity({ serviceId: snapshot.serviceId, entry, pairedTask, turns, attempts, outputs: pairedTask ? deps.loadPairedTurnOutputs(pairedTask.id) : [], messages: deps.loadRecentChatMessages(entry.jid, 8), progressMessages: deps.loadRecentChatMessages(entry.jid, 40), }); } export function buildRoomsTimelineResult( deps: RoomsTimelineRouteDependencies, ): Record { const snapshots = deps.readSnapshots(deps.statusMaxAgeMs); const uniqueByJid = new Map< string, { snapshot: StatusSnapshot; entry: StatusSnapshot['entries'][number]; } >(); for (const snapshot of snapshots) { for (const entry of snapshot.entries) { const existing = uniqueByJid.get(entry.jid); if ( !existing || existing.snapshot.updatedAt.localeCompare(snapshot.updatedAt) < 0 ) { uniqueByJid.set(entry.jid, { snapshot, entry }); } } } const result: Record = {}; for (const [jid, { snapshot, entry }] of uniqueByJid) { const pairedTask = deps.loadLatestPairedTaskForChat(jid) ?? null; const messages = deps.loadRecentChatMessages(jid, 8); const progressMessages = pairedTask ? deps.loadRecentChatMessages(jid, 40) : messages; const outputs = deps.loadRecentPairedTurnOutputsForChat(jid, 8); if (!pairedTask && messages.length === 0 && outputs.length === 0) continue; const latestTurn = pairedTask ? deps.loadLatestPairedTurnForTask(pairedTask.id) : null; const turns = latestTurn ? [latestTurn] : []; result[jid] = buildWebDashboardRoomActivity({ serviceId: snapshot.serviceId, entry, pairedTask, turns, attempts: [], outputs, messages, progressMessages, outputLimit: 8, }); } return result; } function computeRoomsCacheKey(deps: RoomsTimelineRouteDependencies): string { return deps .readSnapshots(deps.statusMaxAgeMs) .map((s) => s.updatedAt) .sort() .join('|'); } export function ensureRoomsTimelineCache( deps: RoomsTimelineRouteDependencies, ): RoomsTimelineCache { const key = computeRoomsCacheKey(deps); if (roomsTimelineCache && roomsTimelineCache.key === key) { return roomsTimelineCache; } const result = buildRoomsTimelineResult(deps); const rawJson = JSON.stringify(result); const gzipBuffer = gzipSync(new TextEncoder().encode(rawJson)); roomsTimelineCache = { key, builtAt: Date.now(), rawJson, gzipBuffer, }; return roomsTimelineCache; } export function startRoomsTimelineCacheRefresh( deps: RoomsTimelineRouteDependencies, ): void { setTimeout(() => { try { ensureRoomsTimelineCache(deps); } catch { /* warm-up failure is non-fatal */ } }, 0); setInterval(() => { try { ensureRoomsTimelineCache(deps); } catch { /* refresh failure is non-fatal */ } }, ROOMS_TIMELINE_BG_INTERVAL_MS).unref(); } function createRoomsTimelineStream( deps: RoomsTimelineRouteDependencies, request: Request, ): ReadableStream { const encoder = new TextEncoder(); return new ReadableStream({ start(controller) { let lastBuiltAt = 0; let closed = false; const enqueue = (chunk: string) => { if (closed) return; try { controller.enqueue(encoder.encode(chunk)); } catch { closed = true; } }; enqueue(`retry: 3000\n\n`); try { const cache = ensureRoomsTimelineCache(deps); lastBuiltAt = cache.builtAt; enqueue(`event: rooms-timeline\ndata: ${cache.rawJson}\n\n`); } catch { /* warm-up failure is non-fatal */ } const tick = () => { if (closed) return; try { const cache = ensureRoomsTimelineCache(deps); if (cache.builtAt !== lastBuiltAt) { lastBuiltAt = cache.builtAt; enqueue(`event: rooms-timeline\ndata: ${cache.rawJson}\n\n`); } else { enqueue(`: ping ${Date.now()}\n\n`); } } catch { /* skip this tick */ } }; const interval = setInterval(tick, 1500); const close = () => { if (closed) return; closed = true; clearInterval(interval); try { controller.close(); } catch { /* already closed */ } }; request.signal.addEventListener('abort', close); }, }); } function handleRoomsStream( deps: RoomsTimelineRouteDependencies, request: Request, jsonResponse: JsonResponse, ): Response { if (!isReadMethod(request.method)) { return jsonResponse({ error: 'Method not allowed' }, { status: 405 }); } return new Response(createRoomsTimelineStream(deps, request), { headers: { 'content-type': 'text/event-stream; charset=utf-8', 'cache-control': 'no-cache, no-transform', connection: 'keep-alive', 'x-accel-buffering': 'no', }, }); } function handleRoomsTimelineSnapshot( deps: RoomsTimelineRouteDependencies, request: Request, jsonResponse: JsonResponse, ): Response { if (!isReadMethod(request.method)) { return jsonResponse({ error: 'Method not allowed' }, { status: 405 }); } const cache = ensureRoomsTimelineCache(deps); const acceptsGzip = request.headers.get('accept-encoding')?.includes('gzip') ?? false; if (acceptsGzip) { return new Response(cache.gzipBuffer, { headers: { 'content-type': 'application/json; charset=utf-8', 'content-encoding': 'gzip', vary: 'accept-encoding', 'x-cache-age': String(Date.now() - cache.builtAt), }, }); } return new Response(cache.rawJson, { headers: { 'content-type': 'application/json; charset=utf-8', 'x-cache-age': String(Date.now() - cache.builtAt), }, }); } function handleSingleRoomTimeline({ request, jsonResponse, ...deps }: RoomsTimelineRouteContext & { timelineRoomJid: string }): Response { if (!isReadMethod(request.method)) { return jsonResponse({ error: 'Method not allowed' }, { status: 405 }); } const snapshots = deps.readSnapshots(deps.statusMaxAgeMs); const matched = latestSnapshotEntry(snapshots, deps.timelineRoomJid); if (!matched) { return jsonResponse({ error: 'Room timeline not found' }, { status: 404 }); } return jsonResponse(buildRoomActivity(deps, matched.snapshot, matched.entry)); } export function handleRoomTimelineRoute( context: RoomsTimelineRouteContext, ): Response | null { const timelineRoomJid = parseRoomTimelinePath(context.url.pathname); if (timelineRoomJid) { return handleSingleRoomTimeline({ ...context, timelineRoomJid }); } if (context.url.pathname === '/api/stream') { return handleRoomsStream(context, context.request, context.jsonResponse); } if (context.url.pathname === '/api/rooms-timeline') { return handleRoomsTimelineSnapshot( context, context.request, context.jsonResponse, ); } return null; }