Scope live chat to the current match and save match history

src/lib/server/fleet/chat.ts

@@ -1,14 +1,15 @@
+import { createHash } from 'node:crypto';
 import { db } from '../db/index.js';
 import { enrichGameChatTimeline, type ChatMatch, type ChatRoundReport, type ContextualChatEntry } from '../../game-chat.js';
 import { parseGameEvent } from '../../telemetry.js';
 import { telemetryRetention, telemetryCutoff } from './telemetry-requests.js';
-import type { ServerRow } from './servers.js';
-import type { TailResult } from './logs.js';
+import { getServer, listServers, type ServerRow } from './servers.js';
+import { gameLogKey, resolveLogPath, tailLog, type TailResult } from './logs.js';
 
 /** Historical context is scoped to this registration, including round-one matches. */
 export function serverChatTimeline(server: ServerRow, tail: TailResult): ContextualChatEntry[] {
     const handle = db();
-    const matches = handle.prepare(`SELECT started_at AS startedAt, ended_at AS endedAt,
+    const matches = handle.prepare(`SELECT id, started_at AS startedAt, ended_at AS endedAt,
         last_received_at AS lastReceivedAt, status, start_observed AS startObserved
         FROM leaderboard_matches WHERE server = ? AND server_deleted_at IS NULL
         ORDER BY id DESC LIMIT 2000`).all(server.name) as Array<Omit<ChatMatch, 'startObserved'> & { startObserved: number }>;
@@ -47,3 +48,89 @@ export function serverChatTimeline(server: ServerRow, tail: TailResult): Context
         reports
     });
 }
+
+export function matchChatTimeline(matchId: number): ContextualChatEntry[] {
+    return (db().prepare('SELECT entry FROM match_chat WHERE match_id = ? ORDER BY rowid')
+        .all(matchId) as { entry: string }[]).map(row => JSON.parse(row.entry));
+}
+
+/** Save before filtering the live view so completed matches retain their messages. */
+function archiveChat(entries: ContextualChatEntry[]): void {
+    const handle = db();
+    const insert = handle.prepare(`INSERT INTO match_chat (match_id, message_key, occurrence, entry)
+        VALUES (?, ?, ?, ?) ON CONFLICT DO NOTHING`);
+    const occurrences = new Map<string, number>();
+    handle.transaction(() => {
+        for (const entry of entries) {
+            if (entry.type !== 'message' || entry.matchId === undefined) continue;
+            // Engine time is stable across reads; estimated wall time is not.
+            // Count identical messages separately, including repeats in one second.
+            const key = createHash('sha256').update(JSON.stringify(entry.message)).digest('hex');
+            const scope = `${entry.matchId}:${key}`;
+            const occurrence = (occurrences.get(scope) ?? 0) + 1;
+            occurrences.set(scope, occurrence);
+            insert.run(entry.matchId, key, occurrence, JSON.stringify(entry));
+        }
+    })();
+}
+
+export function currentMatchChat(server: ServerRow, tail: TailResult | null): ContextualChatEntry[] {
+    const entries = tail ? serverChatTimeline(server, tail) : [];
+    archiveChat(entries);
+    const match = db().prepare(`SELECT id, started_at AS startedAt FROM leaderboard_matches
+        WHERE server = ? AND server_deleted_at IS NULL ORDER BY id DESC LIMIT 1`)
+        .get(server.name) as { id: number; startedAt: string } | undefined;
+    let startIndex = entries.length - 1;
+    while (startIndex >= 0) {
+        const entry = entries[startIndex];
+        if (entry.type === 'boundary' && entry.event !== 'end') break;
+        startIndex--;
+    }
+    const current = entries.slice(Math.max(0, startIndex));
+    if (!match) {
+        // Servers without gameplay telemetry still have log-defined sessions.
+        return current.filter(entry => entry.context.realTime !== null
+            && Date.parse(entry.context.realTime) >= Date.parse(server.created_at));
+    }
+    const start = entries[startIndex];
+    // A log restart can arrive before game_start telemetry. Clear immediately
+    // instead of falling back to messages from the previous recorded match.
+    if (start && !current.some(entry => entry.matchId === match.id)
+        && (!start.context.realTime || Date.parse(start.context.realTime) > Date.parse(match.startedAt) + 1000)) return [];
+    return matchChatTimeline(match.id);
+}
+
+const pending = new Map<string, Promise<{ entries: ContextualChatEntry[]; missing: boolean }>>();
+
+/** Shared by the live API and the worker; collection does not require a viewer. */
+export function collectServerChat(server: ServerRow): Promise<{ entries: ContextualChatEntry[]; missing: boolean }> {
+    const key = `${server.name}:${server.created_at}:${server.native_instance_id}:${server.instance_generation}`;
+    const existing = pending.get(key);
+    if (existing) return existing;
+    const work = (async () => {
+        const path = resolveLogPath(server.name, gameLogKey(server.name));
+        if (!path) throw new Error('Invalid server log path');
+        const tail = await tailLog(path, 2000);
+        const current = getServer(server.name);
+        // A remote read may finish after destruction or replacement.
+        if (!current || current.destroyed_at || current.created_at !== server.created_at
+            || current.native_instance_id !== server.native_instance_id || current.instance_generation !== server.instance_generation) {
+            return { entries: [], missing: true };
+        }
+        return { entries: currentMatchChat(current, tail), missing: tail === null };
+    })().finally(() => pending.delete(key));
+    pending.set(key, work);
+    return work;
+}
+
+export async function collectFleetChat(): Promise<void> {
+    const servers = listServers().filter(server => !server.destroyed_at);
+    // Bound agent traffic and isolate failures to their own server.
+    async function collect() {
+        for (let server = servers.shift(); server; server = servers.shift()) {
+            try { await collectServerChat(server); }
+            catch (error) { console.error(`[chat] Could not collect chat for ${server.name}`, error); }
+        }
+    }
+    await Promise.all(Array.from({ length: Math.min(4, servers.length) }, collect));
+}