Optimize message ingestion with logChatMessagesBatch: bulk upsert to database.

server/db.js

@@ -565,6 +565,134 @@ export async function logChatMessage(msg) {
   }
 }
 
+/**
+ * Log chat messages in batch (performance optimized)
+ */
+export async function logChatMessagesBatch(messages) {
+  if (!messages || messages.length === 0) return;
+
+  // 1. Fetch all streams to search in-memory instead of making DB queries for each message
+  const streams = await getStreamsList();
+  const activeStream = await getActiveStream();
+
+  const getStreamIdForTime = (timestampIso) => {
+    if (!timestampIso) return activeStream ? activeStream.id : null;
+    const time = new Date(timestampIso);
+    const matchedStream = streams.find(s => {
+      const start = new Date(s.start_time);
+      const isAfterStart = start <= time;
+      const isBeforeEnd = !s.end_time || new Date(s.end_time) >= time;
+      return isAfterStart && isBeforeEnd;
+    });
+    return matchedStream ? matchedStream.id : (activeStream ? activeStream.id : null);
+  };
+
+  if (dbMode === 'sqlite') {
+    const insertMsg = sqliteDb.prepare(`
+      INSERT INTO messages (id, stream_id, username, display_name, message, timestamp, is_streamer, is_mod, is_sub)
+      VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+    `);
+
+    const insertViewer = sqliteDb.prepare(`
+      INSERT INTO stream_viewers (stream_id, username, display_name, has_chatted, is_mod, is_sub, first_seen)
+      VALUES (?, ?, ?, 1, ?, ?, ?)
+      ON CONFLICT(stream_id, username) DO UPDATE SET 
+        has_chatted = 1,
+        is_mod = excluded.is_mod,
+        is_sub = excluded.is_sub
+    `);
+
+    const runTransaction = sqliteDb.transaction((msgs) => {
+      for (const msg of msgs) {
+        const streamId = getStreamIdForTime(msg.timestamp);
+        try {
+          insertMsg.run(
+            msg.id,
+            streamId,
+            msg.username.toLowerCase(),
+            msg.displayName || msg.username,
+            msg.message,
+            msg.timestamp || new Date().toISOString(),
+            msg.isStreamer ? 1 : 0,
+            msg.isMod ? 1 : 0,
+            msg.isSub ? 1 : 0
+          );
+        } catch (err) {
+          if (!err.message.includes('UNIQUE constraint failed')) {
+            console.error('[Database] SQLite Batch Error saving message:', err.message);
+          }
+        }
+
+        try {
+          insertViewer.run(
+            streamId,
+            msg.username.toLowerCase(),
+            msg.displayName || msg.username,
+            msg.isMod ? 1 : 0,
+            msg.isSub ? 1 : 0,
+            msg.timestamp || new Date().toISOString()
+          );
+        } catch (err) {
+          console.error('[Database] SQLite Batch Error updating viewer:', err.message);
+        }
+      }
+    });
+
+    runTransaction(messages);
+  } else {
+    // Supabase Mode: Build batch lists
+    const messagesToInsert = [];
+    const viewersToUpsert = new Map(); // Use Map to unique-fy viewers by stream_id + username
+
+    for (const msg of messages) {
+      const streamId = getStreamIdForTime(msg.timestamp);
+      
+      messagesToInsert.push({
+        id: msg.id,
+        stream_id: streamId,
+        username: msg.username.toLowerCase(),
+        display_name: msg.displayName || msg.username,
+        message: msg.message,
+        timestamp: msg.timestamp || new Date().toISOString(),
+        is_streamer: msg.isStreamer || false,
+        is_mod: msg.isMod || false,
+        is_sub: msg.isSub || false
+      });
+
+      const viewerKey = `${streamId || 'null'}-${msg.username.toLowerCase()}`;
+      viewersToUpsert.set(viewerKey, {
+        stream_id: streamId,
+        username: msg.username.toLowerCase(),
+        display_name: msg.displayName || msg.username,
+        has_chatted: true,
+        is_mod: msg.isMod || false,
+        is_sub: msg.isSub || false,
+        first_seen: msg.timestamp || new Date().toISOString()
+      });
+    }
+
+    // 1. Bulk upsert messages
+    const { error: msgError } = await supabase
+      .from('messages')
+      .upsert(messagesToInsert, { onConflict: 'id' });
+
+    if (msgError) {
+      console.error('[Supabase] Bulk message upsert error:', msgError.message);
+    }
+
+    // 2. Bulk upsert viewers
+    const uniqueViewers = Array.from(viewersToUpsert.values());
+    const { error: viewerError } = await supabase
+      .from('stream_viewers')
+      .upsert(uniqueViewers, { onConflict: 'stream_id, username' });
+
+    if (viewerError) {
+      console.error('[Supabase] Bulk viewers upsert error:', viewerError.message);
+    }
+  }
+}
+
+
 /**
  * Log voice words (bulk insert)
  */

server/server.js

@@ -7,6 +7,7 @@ import path from 'path';
 import fs from 'fs';
 import { 
   logChatMessage, 
+  logChatMessagesBatch,
   logVoiceWords, 
   logModAction, 
   logViewerJoin,
@@ -175,12 +176,11 @@ app.post('/api/log/messages', authenticateWorker, async (req, res) => {
     return res.status(400).json({ error: 'Invalid input. Expected array of messages.' });
   }
 
-  for (const msg of messages) {
-    try {
-      await logChatMessage(msg);
-    } catch (e) {
-      console.error('Error logging batch message:', e);
-    }
+  try {
+    await logChatMessagesBatch(messages);
+  } catch (err) {
+    console.error('Error logging batch messages:', err);
+    return res.status(500).json({ error: 'Failed to log messages' });
   }
 
   res.json({ success: true, count: messages.length });