Capture team chat and expose moderation match archives

src/lib/server/fleet/chat.ts

@@ -212,10 +212,14 @@ interface ChatReadState {
     entries: ContextualChatEntry[];
     contextVersion?: string;
     wasOccupied: boolean;
+    needsDrain: boolean;
+    bytesRead: number;
 }
 const readers = new Map<string, ChatReadState>();
 const failures = new Map<string, number>();
 const CHAT_READ_MIN_AGE_MS = 15_000;
+const MAX_IDLE_STALLED_READS = 3;
+const idleStalls = new Map<string, number>();
 type LastChatRead = { completedAt: number } & ({ missing: boolean } | { error: unknown });
 const lastReads = new Map<string, LastChatRead>();
 
@@ -249,6 +253,7 @@ export function collectServerChat(
     const key = chatReadKey(server);
     const existing = pending.get(key);
     if (existing) return existing;
+    let progressed = false;
     const work = (async () => {
         const path = resolveLogPath(server.name, gameLogKey(server.name));
         if (!path) throw new Error('Invalid server log path');
@@ -263,7 +268,9 @@ export function collectServerChat(
                   partial: Buffer.alloc(0),
                   skipLine: false,
                   entries: [],
-                  wasOccupied: false
+                  wasOccupied: false,
+                  needsDrain: false,
+                  bytesRead: 0
               };
         const tail = await readLogChunk(path, state.cursor);
         const current = getServer(server.name);
@@ -290,7 +297,10 @@ export function collectServerChat(
                 state.entries = [];
             }
             state.cursor = tail.cursor;
-            const bytes = Buffer.concat([state.partial, Buffer.from(tail.contentBase64, 'base64')]);
+            const content = Buffer.from(tail.contentBase64, 'base64');
+            progressed = content.length > 0;
+            state.bytesRead += content.length;
+            const bytes = Buffer.concat([state.partial, content]);
             const end = bytes.lastIndexOf(10) + 1;
             if (end) {
                 const lines = bytes
@@ -318,14 +328,17 @@ export function collectServerChat(
                 state.skipLine = true;
             }
             // File mtime anchors the final line only after the backlog has been read.
-            if (tail.cursor.offset < tail.size || state.partial.length) {
+            state.needsDrain = tail.cursor.offset < tail.size || state.partial.length > 0;
+            if (state.needsDrain) {
                 state.signature = undefined;
                 readers.set(server.name, state);
                 return { entries: currentChatEntries(current, state.entries), missing: false };
             }
         } else {
             // Older agents remain readable during a rolling update.
+            state.needsDrain = false;
             changed = state.signature !== `${tail.size}:${tail.mtime}`;
+            progressed = changed;
             if (changed) state.lines = tail.lines;
         }
         const signature = `${tail.size}:${tail.mtime}`;
@@ -354,6 +367,9 @@ export function collectServerChat(
     })()
         .then(
             (result) => {
+                const state = readers.get(server.name);
+                if (progressed || result.missing || (state?.key === key && !state.needsDrain))
+                    idleStalls.delete(key);
                 lastReads.set(key, { completedAt: Date.now(), missing: result.missing });
                 return result;
             },
@@ -377,25 +393,53 @@ export async function collectFleetChat(): Promise<void> {
     for (const name of readers.keys()) if (!names.has(name)) readers.delete(name);
     for (const name of failures.keys()) if (!names.has(name)) failures.delete(name);
     for (const key of lastReads.keys()) if (!keys.has(key)) lastReads.delete(key);
+    for (const key of idleStalls.keys()) if (!keys.has(key)) idleStalls.delete(key);
     const servers = live.filter((server) => {
         if (server.host_id && !hosts.get(server.host_id)?.connected) return false;
         if (server.container_state && !['running', 'unknown'].includes(server.container_state))
             return false;
         const players = observe(server).last_players ?? server.last_players;
-        // One final read after players leave retains messages written at match end.
-        return players !== 0 || readers.get(server.name)?.wasOccupied;
+        const key = chatReadKey(server);
+        if (players !== 0) {
+            idleStalls.delete(key);
+            return true;
+        }
+        const stalls = idleStalls.get(key) ?? 0;
+        if (stalls >= MAX_IDLE_STALLED_READS) return false;
+        const last = lastReads.get(key);
+        if (stalls > 0 && last && Date.now() - last.completedAt < CHAT_READ_MIN_AGE_MS)
+            return false;
+        const state = readers.get(server.name);
+        // Pick up surviving final logs after a collector restart, then finish any
+        // unread chunks even after players leave. Errors and incomplete final lines
+        // have bounded idle retries rather than polling an empty server forever.
+        return (
+            !last ||
+            'error' in last ||
+            (state?.key === key && (state.wasOccupied || state.needsDrain))
+        );
     });
     // Bound agent traffic and isolate failures to their own server.
     async function collect() {
         for (let server = servers.shift(); server; server = servers.shift()) {
+            const key = chatReadKey(server);
+            const before = readers.get(server.name);
             try {
                 await collectServerChat(server);
                 const state = readers.get(server.name);
-                if (state)
+                if (state?.key === key) {
                     state.wasOccupied =
                         (observe(server).last_players ?? server.last_players ?? 0) > 0;
+                    if (!state.wasOccupied && state.needsDrain) {
+                        const progressed =
+                            before?.key !== key || state.bytesRead > before.bytesRead;
+                        idleStalls.set(key, progressed ? 0 : (idleStalls.get(key) ?? 0) + 1);
+                    } else idleStalls.delete(key);
+                }
                 failures.delete(server.name);
             } catch (error) {
+                if ((observe(server).last_players ?? server.last_players) === 0)
+                    idleStalls.set(key, (idleStalls.get(key) ?? 0) + 1);
                 if (Date.now() - (failures.get(server.name) ?? -Infinity) >= 5 * 60_000) {
                     console.error(`[chat] Could not collect chat for ${server.name}`, error);
                     failures.set(server.name, Date.now());