feat: add automatic self-healing for empty stream viewers

server/db.js

@@ -1091,6 +1091,102 @@ export async function getStreamsList() {
   }
 }
 
+/**
+ * Self-healing: Rebuild stream_viewers and chat_users for a stream from its messages
+ */
+export async function rebuildViewersForStream(streamId) {
+  if (!streamId || dbMode !== 'supabase') return;
+  console.log(`[Self-Healing] Rebuilding viewers and chat users for stream ${streamId} from messages...`);
+  
+  // Fetch messages from Supabase (only needed columns to reduce egress)
+  const { data: messages, error } = await supabase
+    .from('messages')
+    .select('username, display_name, timestamp, is_mod, is_sub')
+    .eq('stream_id', streamId);
+    
+  if (error) {
+    console.error(`[Self-Healing] Error fetching messages for stream ${streamId}:`, error);
+    return;
+  }
+  
+  if (!messages || messages.length === 0) {
+    console.log(`[Self-Healing] No messages found for stream ${streamId}.`);
+    return;
+  }
+  
+  const viewersMap = new Map();
+  const chatUsersMap = new Map();
+  
+  for (const msg of messages) {
+    const username = msg.username.toLowerCase();
+    const displayName = msg.display_name || msg.username;
+    const ts = msg.timestamp;
+    
+    // Viewer key: stream_id-username
+    const viewerKey = `${streamId}-${username}`;
+    if (!viewersMap.has(viewerKey)) {
+      viewersMap.set(viewerKey, {
+        stream_id: streamId,
+        username: username,
+        display_name: displayName,
+        has_chatted: true,
+        is_mod: msg.is_mod || false,
+        is_sub: msg.is_sub || false,
+        first_seen: ts
+      });
+    } else {
+      const viewer = viewersMap.get(viewerKey);
+      if (new Date(ts) < new Date(viewer.first_seen)) {
+        viewer.first_seen = ts;
+      }
+      if (msg.is_mod) viewer.is_mod = true;
+      if (msg.is_sub) viewer.is_sub = true;
+    }
+    
+    // Chat user profile
+    const userKey = username;
+    if (!chatUsersMap.has(userKey)) {
+      chatUsersMap.set(userKey, {
+        username: username,
+        display_name: displayName,
+        is_mod: msg.is_mod || false,
+        is_sub: msg.is_sub || false,
+        is_vip: false,
+        last_seen: ts
+      });
+    } else {
+      const user = chatUsersMap.get(userKey);
+      if (new Date(ts) > new Date(user.last_seen)) {
+        user.last_seen = ts;
+        user.display_name = displayName;
+      }
+      if (msg.is_mod) user.is_mod = true;
+      if (msg.is_sub) user.is_sub = true;
+    }
+  }
+  
+  const viewers = Array.from(viewersMap.values());
+  const chatUsers = Array.from(chatUsersMap.values());
+  
+  // Upsert chat_users
+  const { error: userErr } = await supabase
+    .from('chat_users')
+    .upsert(chatUsers, { onConflict: 'username' });
+  if (userErr) {
+    console.error(`[Self-Healing] Error upserting chat_users:`, userErr);
+  }
+  
+  // Upsert stream_viewers
+  const { error: viewErr } = await supabase
+    .from('stream_viewers')
+    .upsert(viewers, { onConflict: 'stream_id, username' });
+  if (viewErr) {
+    console.error(`[Self-Healing] Error upserting stream_viewers:`, viewErr);
+  }
+  
+  console.log(`[Self-Healing] Rebuilt and saved ${viewers.length} viewers for stream ${streamId}`);
+}
+
 /**
  * Get top active chatters
  */
@@ -1149,11 +1245,33 @@ export async function getTopChatters(streamId = null, limit = 50) {
         .order('message_count', { ascending: false })
         .limit(limit);
     }
-    const { data, error } = await query;
+    let { data, error } = await query;
     if (error) {
       console.error('[Supabase] Error getting top chatters:', error);
       return [];
     }
+
+    // Trigger self-healing if we expected chatters but got 0
+    if (streamId && (!data || data.length === 0)) {
+      const { count: msgCount } = await supabase
+        .from('messages')
+        .select('*', { count: 'exact', head: true })
+        .eq('stream_id', streamId);
+        
+      if (msgCount && msgCount > 0) {
+        await rebuildViewersForStream(streamId);
+        const requery = await supabase
+          .from('stream_chatter_stats')
+          .select('*')
+          .eq('stream_id', streamId)
+          .order('message_count', { ascending: false })
+          .limit(limit);
+        if (!requery.error && requery.data) {
+          data = requery.data;
+        }
+      }
+    }
+
     return data || [];
   }
 }
@@ -1201,10 +1319,20 @@ export async function getStatsSummary(streamId = null) {
     // Unique chatters count
     let uniqueChatters = 0;
     if (streamId) {
-      const { count } = await supabase
+      let { count } = await supabase
         .from('stream_chatter_stats')
         .select('*', { count: 'exact', head: true })
         .eq('stream_id', streamId);
+        
+      // Self-healing check if summary shows 0 viewers but we have messages
+      if ((!count || count === 0) && msgCount > 0) {
+        await rebuildViewersForStream(streamId);
+        const requery = await supabase
+          .from('stream_chatter_stats')
+          .select('*', { count: 'exact', head: true })
+          .eq('stream_id', streamId);
+        count = requery.count || 0;
+      }
       uniqueChatters = count || 0;
     } else {
       const { count } = await supabase