logger

commit c7844ea92bd1005ee709bbe0816564df9c23d657
parent 5ee74af1e78649c8b72605877ced9ea75f4e50da
Author: notamitgamer-bot <bot@users.noreply.github.com>
Date:   Sat, 29 Aug 2026 01:21:40 +0000

Add global /api/sync/stream so all chats sync at once

Previously, new messages were only pushed to a client if it had an
active SSE connection open for that specific chatId. Since the
frontend only opened a messages stream when a chat was clicked,
messages for chats you hadn't opened never arrived until you opened
them and did a fresh pull.

This adds a single global sync endpoint that streams message updates
for every chat over one connection, matching how the frontend is
being updated to sync all conversations on reconnect instead of
per-chat-on-open.

- sseManager.js: new clients.sync client set
- cache.js: broadcast every message change to clients.sync, stamped
  with chatId (not present on the raw Firestore doc otherwise)
- routes/api.js: new GET /api/sync/stream?since=<ts> route, mirrors
  /api/messages/stream but covers all chats in one connection

Diffstat:
M
src/cache.js
 | 
8
++++++++
M
src/routes/api.js
 | 
44
++++++++++++++++++++++++++++++++++++++++++++
M
src/sseManager.js
 | 
3
++-

3 files changed, 54 insertions(+), 1 deletion(-)
diff --git a/src/cache.js b/src/cache.js
@@ -73,6 +73,14 @@ function startPermanentListeners() {
                 const payload = `event: update\ndata: ${JSON.stringify([{ type: change.type, doc: data }])}\n\n`;
                 chatClients.forEach(res => { try { res.write(payload); } catch (e) {} });
             }
+
+            // Broadcast to global sync clients too, stamped with chatId since
+            // the raw Firestore doc doesn't carry it (it's derived from the doc path).
+            if (clients.sync.size > 0) {
+                const syncData = { ...data, chatId };
+                const syncPayload = `event: update\ndata: ${JSON.stringify([{ type: change.type, doc: syncData }])}\n\n`;
+                clients.sync.forEach(res => { try { res.write(syncPayload); } catch (e) {} });
+            }
         });
     });
 }
diff --git a/src/routes/api.js b/src/routes/api.js
@@ -106,6 +106,50 @@ router.get('/api/messages/stream', verifyApiToken, async (req, res) => {
     }
 });
 
+// Global sync stream: one connection covers messages for ALL chats, so the
+// client doesn't need to open a new SSE connection per chat it opens.
+router.get('/api/sync/stream', verifyApiToken, async (req, res) => {
+    if (!isCacheReady()) return res.status(503).json({ error: 'Cache still warming, retry shortly' });
+
+    const { since } = req.query;
+
+    res.writeHead(200, {
+        'Content-Type': 'text/event-stream',
+        'Cache-Control': 'no-cache',
+        'Connection': 'keep-alive',
+        'X-Accel-Buffering': 'no'
+    });
+    if (res.flushHeaders) res.flushHeaders();
+    if (res.socket) res.socket.setNoDelay(true);
+    res.write('\n');
+
+    const cleanup = () => {
+        clients.sync.delete(res);
+    };
+
+    enforceConnectionCeiling(req, res, cleanup);
+    clients.sync.add(res);
+
+    try {
+        const sinceTs = since ? parseInt(since, 10) : 0;
+        const initialMessages = [];
+
+        messagesCache.forEach((msgMap, chatId) => {
+            for (const msg of msgMap.values()) {
+                if (msg.timestamp > sinceTs) {
+                    initialMessages.push({ ...msg, chatId });
+                }
+            }
+        });
+
+        initialMessages.sort((a, b) => a.timestamp - b.timestamp);
+
+        res.write(`event: initial\ndata: ${JSON.stringify(initialMessages)}\n\n`);
+    } catch (e) {
+        console.error("Error sending initial sync payload:", e);
+    }
+});
+
 // --- API: Standard Actions ---
 router.post('/api/rename', verifyApiToken, async (req, res) => {
     if (!isCacheReady()) return res.status(503).json({ error: 'Cache still warming, retry shortly' });
diff --git a/src/sseManager.js b/src/sseManager.js
@@ -4,7 +4,8 @@ const { MAX_CONNECTIONS_PER_TOKEN } = require('./config');
 const activeConnections = [];
 const clients = {
     chats: new Set(),
-    messages: new Map()
+    messages: new Map(),
+    sync: new Set() // Global message-sync clients (one connection covers all chats)
 };
 
 function enforceConnectionCeiling(req, res, cleanupFunction) {