feat: global roles migration, VIP badges, role backfiller script

client/src/App.jsx

@@ -1074,7 +1074,8 @@ export default function App() {
                                       <span className="badge badge-streamer">Стример</span>
                                     )}
                                     {c.is_mod ? <span className="badge badge-mod">Мод</span> : null}
-                                    {c.is_sub ? <span className="badge badge-sub">Саб</span> : null}
+                                    {c.is_vip && !c.is_mod ? <span className="badge badge-vip">VIP</span> : null}
+                                    {c.is_sub && !c.is_mod && !c.is_vip ? <span className="badge badge-sub">Саб</span> : null}
                                   </div>
                                 </td>
                                 <td style={{ fontWeight: 700, color: 'var(--color-text-accent)' }}>

local_worker/role_backfiller.py

@@ -0,0 +1,157 @@
+import os
+import sys
+import time
+import requests
+from dotenv import load_dotenv
+
+# Load local environment settings
+load_dotenv()
+
+API_URL = os.getenv("API_URL", "http://localhost:3000")
+API_KEY = os.getenv("API_KEY", "")
+
+if not API_KEY:
+    print("[Error] API_KEY is missing in .env! Cannot push data to backend.")
+    sys.exit(1)
+
+headers = {
+    "x-api-key": API_KEY,
+    "Content-Type": "application/json"
+}
+
+def download_vod_roles(vod_id):
+    """Download chat log of a past VOD and extract only roles"""
+    roles_dict = {}  # Store latest role per user
+    url = "https://gql.twitch.tv/gql"
+    gql_headers = {
+        "Client-Id": "kimne78kx3ncx6brgo4mv6wki5h1ko",
+        "User-Agent": "Mozilla/5.0"
+    }
+    
+    current_offset = 0
+    print(f"\n[Roles] Запуск сверхбыстрого парсера ролей для VOD {vod_id}...")
+    
+    total_messages = 0
+
+    while True:
+        payload = {
+            "operationName": "VideoCommentsByOffsetOrCursor",
+            "variables": {
+                "videoID": str(vod_id),
+                "contentOffsetSeconds": current_offset
+            },
+            "extensions": {
+                "persistedQuery": {
+                    "version": 1,
+                    "sha256Hash": "b70a3591ff0f4e0313d126c6a1502d79a1c02baebb288227c582044aa76adf6a"
+                }
+            }
+        }
+            
+        try:
+            res = requests.post(url, json=payload, headers=gql_headers, timeout=10)
+            if res.status_code != 200:
+                print(f"[Roles] Ошибка GQL: {res.status_code}")
+                break
+                
+            data = res.json()
+            if isinstance(data, list):
+                data = data[0]
+                
+            video = data.get("data", {}).get("video", {})
+            if not video:
+                break
+                
+            comments_edge = video.get("comments") or {}
+            edges = comments_edge.get("edges") or []
+            if not edges:
+                break
+            
+            for edge in edges:
+                if not edge:
+                    continue
+                node = edge.get("node")
+                if not node:
+                    continue
+                    
+                commenter = node.get("commenter")
+                if not commenter:
+                    continue
+                
+                user = commenter.get("login")
+                if not user:
+                    continue
+                    
+                display_name = commenter.get("displayName", user)
+                
+                message = node.get("message")
+                if not message:
+                    continue
+                
+                timestamp = node.get("createdAt")
+                
+                # Badges parse
+                user_badges = message.get("userBadges") or []
+                badges = [b.get("setID") for b in user_badges if b]
+                is_mod = "moderator" in badges or "broadcaster" in badges
+                is_sub = "subscriber" in badges or "founder" in badges
+                is_vip = "vip" in badges
+                
+                # We always overwrite with the latest role from the VOD
+                # because the chat goes forward in time
+                roles_dict[user.lower()] = {
+                    "username": user,
+                    "displayName": display_name,
+                    "isMod": is_mod,
+                    "isSub": is_sub,
+                    "isVip": is_vip,
+                    "timestamp": timestamp
+                }
+                total_messages += 1
+                
+            print(f"-> Обработано {total_messages} сообщений. Найдено уникальных пользователей: {len(roles_dict)}...", end="\r")
+            
+            # Progress offset
+            last_offset = edges[-1].get("node", {}).get("contentOffsetSeconds")
+            if last_offset is not None:
+                next_offset = last_offset + 1
+                if next_offset <= current_offset:
+                    next_offset = current_offset + 30
+                current_offset = next_offset
+            else:
+                break
+        except Exception as e:
+            print(f"\n[Roles] Ошибка при загрузке: {e}")
+            break
+            
+    print(f"\n[Roles] Парсинг завершен. Всего уникальных пользователей: {len(roles_dict)}")
+    return list(roles_dict.values())
+
+if __name__ == "__main__":
+    print("=========================================================")
+    print("      Ultra-Fast Role Backfiller                        ")
+    print("=========================================================")
+    
+    vod_id = input("Введите ID Twitch VOD (например, 2154382910): ").strip()
+    if not vod_id:
+        print("ID VOD не введен. Выход.")
+        sys.exit(0)
+        
+    roles = download_vod_roles(vod_id)
+    if roles:
+        print("\n[Sync] Отправка ролей на сервер (пакетами по 500 пользователей)...")
+        batch_size = 500
+        for i in range(0, len(roles), batch_size):
+            batch = roles[i:i+batch_size]
+            try:
+                url = f"{API_URL}/api/log/roles"
+                res = requests.post(url, json={"roles": batch}, headers=headers, timeout=10)
+                if res.status_code == 200:
+                    print(f"-> Обновлено пользователей: {i + len(batch)} / {len(roles)}", end="\r")
+                else:
+                    print(f"\n-> Ошибка отправки пакета: {res.status_code}")
+            except Exception as e:
+                print(f"\n-> Ошибка сети при отправке пакета: {e}")
+        print("\n-> Восстановление ролей успешно завершено!")
+    else:
+        print("\n-> Нет ролей для обновления.")

local_worker/vod_backfiller.py

@@ -162,10 +162,11 @@ def download_vod_chat(vod_id, start_offset=0, end_offset=None):
                 timestamp = node.get("createdAt")
                 
                 # Badges parse
-                user_badges = node.get("userBadges") or []
-                badges = [b.get("id") for b in user_badges if b]
+                user_badges = message.get("userBadges") or []
+                badges = [b.get("setID") for b in user_badges if b]
                 is_mod = "moderator" in badges or "broadcaster" in badges
                 is_sub = "subscriber" in badges or "founder" in badges
+                is_vip = "vip" in badges
                 is_streamer = (user.lower() == TWITCH_CHANNEL)
                 
                 seen_ids.add(msg_id)
@@ -177,7 +178,8 @@ def download_vod_chat(vod_id, start_offset=0, end_offset=None):
                     "timestamp": timestamp,
                     "isStreamer": is_streamer,
                     "isMod": is_mod,
-                    "isSub": is_sub
+                    "isSub": is_sub,
+                    "isVip": is_vip
                 })
                 
             print(f"-> Загружено {len(comments)} комментариев...", end="\r")

local_worker/worker.py

@@ -163,6 +163,7 @@ def twitch_chat_listener():
                         is_streamer = (user.lower() == TWITCH_CHANNEL)
                         is_mod = "moderator" in badges or "broadcaster" in badges
                         is_sub = "subscriber" in badges or "founder" in badges
+                        is_vip = "vip" in badges
                         
                         parsed_msg = {
                             "id": msg_id,
@@ -172,7 +173,8 @@ def twitch_chat_listener():
                             "timestamp": new_iso_timestamp(),
                             "isStreamer": is_streamer,
                             "isMod": is_mod,
-                            "isSub": is_sub
+                            "isSub": is_sub,
+                            "isVip": is_vip
                         }
                         
                         # Add to queue

server/db.js

@@ -92,6 +92,26 @@ if (dbMode === 'sqlite') {
       key TEXT PRIMARY KEY,
       value TEXT NOT NULL
     );
+
+    CREATE TABLE IF NOT EXISTS stream_viewers (
+      stream_id INTEGER,
+      username TEXT NOT NULL,
+      display_name TEXT,
+      has_chatted INTEGER DEFAULT 0,
+      is_mod INTEGER DEFAULT 0,
+      is_sub INTEGER DEFAULT 0,
+      first_seen TEXT,
+      PRIMARY KEY (stream_id, username)
+    );
+
+    CREATE TABLE IF NOT EXISTS chat_users (
+      username TEXT PRIMARY KEY,
+      display_name TEXT,
+      is_mod INTEGER DEFAULT 0,
+      is_sub INTEGER DEFAULT 0,
+      is_vip INTEGER DEFAULT 0,
+      last_seen TEXT NOT NULL
+    );
   `);
   console.log('[Database] SQLite Tables & Indexes initialized.');
 
@@ -589,8 +609,8 @@ export async function logChatMessagesBatch(messages) {
 
   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 (?, ?, ?, ?, ?, ?, ?, ?, ?)
+      INSERT INTO messages (id, stream_id, username, display_name, message, timestamp, is_streamer)
+      VALUES (?, ?, ?, ?, ?, ?, ?)
     `);
 
     const insertViewer = sqliteDb.prepare(`
@@ -602,6 +622,18 @@ export async function logChatMessagesBatch(messages) {
         is_sub = excluded.is_sub
     `);
 
+    const insertChatUser = sqliteDb.prepare(`
+      INSERT INTO chat_users (username, display_name, is_mod, is_sub, is_vip, last_seen)
+      VALUES (?, ?, ?, ?, ?, ?)
+      ON CONFLICT(username) DO UPDATE SET
+        display_name = excluded.display_name,
+        is_mod = excluded.is_mod,
+        is_sub = excluded.is_sub,
+        is_vip = excluded.is_vip,
+        last_seen = excluded.last_seen
+      WHERE excluded.last_seen > chat_users.last_seen
+    `);
+
     const runTransaction = sqliteDb.transaction((msgs) => {
       for (const msg of msgs) {
         const streamId = getStreamIdForTime(msg.timestamp);
@@ -613,9 +645,7 @@ export async function logChatMessagesBatch(messages) {
             msg.displayName || msg.username,
             msg.message,
             msg.timestamp || new Date().toISOString(),
-            msg.isStreamer ? 1 : 0,
-            msg.isMod ? 1 : 0,
-            msg.isSub ? 1 : 0
+            msg.isStreamer ? 1 : 0
           );
         } catch (err) {
           if (!err.message.includes('UNIQUE constraint failed')) {
@@ -635,6 +665,19 @@ export async function logChatMessagesBatch(messages) {
         } catch (err) {
           console.error('[Database] SQLite Batch Error updating viewer:', err.message);
         }
+
+        try {
+          insertChatUser.run(
+            msg.username.toLowerCase(),
+            msg.displayName || msg.username,
+            msg.isMod ? 1 : 0,
+            msg.isSub ? 1 : 0,
+            msg.isVip ? 1 : 0,
+            msg.timestamp || new Date().toISOString()
+          );
+        } catch (err) {
+          console.error('[Database] SQLite Batch Error updating chat_user:', err.message);
+        }
       }
     });
 
@@ -643,23 +686,24 @@ export async function logChatMessagesBatch(messages) {
     // Supabase Mode: Build batch lists
     const messagesToInsert = [];
     const viewersToUpsert = new Map(); // Use Map to unique-fy viewers by stream_id + username
+    const chatUsersToUpsert = new Map();
 
     for (const msg of messages) {
       const streamId = getStreamIdForTime(msg.timestamp);
+      const ts = msg.timestamp || new Date().toISOString();
+      const username = msg.username.toLowerCase();
       
       messagesToInsert.push({
         id: msg.id,
         stream_id: streamId,
-        username: msg.username.toLowerCase(),
+        username: username,
         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
+        timestamp: ts,
+        is_streamer: msg.isStreamer || false
       });
 
-      const viewerKey = `${streamId || 'null'}-${msg.username.toLowerCase()}`;
+      const viewerKey = `${streamId || 'null'}-${username}`;
       viewersToUpsert.set(viewerKey, {
         stream_id: streamId,
         username: msg.username.toLowerCase(),
@@ -667,7 +711,17 @@ export async function logChatMessagesBatch(messages) {
         has_chatted: true,
         is_mod: msg.isMod || false,
         is_sub: msg.isSub || false,
-        first_seen: msg.timestamp || new Date().toISOString()
+        first_seen: ts
+      });
+
+      const userKey = username;
+      chatUsersToUpsert.set(userKey, {
+        username: username,
+        display_name: msg.displayName || msg.username,
+        is_mod: msg.isMod || false,
+        is_sub: msg.isSub || false,
+        is_vip: msg.isVip || false,
+        last_seen: ts
       });
     }
 
@@ -689,6 +743,110 @@ export async function logChatMessagesBatch(messages) {
     if (viewerError) {
       console.error('[Supabase] Bulk viewers upsert error:', viewerError.message);
     }
+
+    // 3. Bulk upsert global chat_users safely based on last_seen
+    const uniqueChatUsers = Array.from(chatUsersToUpsert.values());
+    if (uniqueChatUsers.length > 0) {
+      const { data: existingUsers } = await supabase
+        .from('chat_users')
+        .select('username, last_seen')
+        .in('username', uniqueChatUsers.map(u => u.username));
+      
+      const existingMap = new Map((existingUsers || []).map(u => [u.username, new Date(u.last_seen).getTime()]));
+      const toUpsert = uniqueChatUsers.filter(u => {
+         const oldTime = existingMap.get(u.username) || 0;
+         const newTime = new Date(u.last_seen).getTime();
+         return newTime > oldTime;
+      });
+
+      if (toUpsert.length > 0) {
+         const { error: chatUserError } = await supabase
+            .from('chat_users')
+            .upsert(toUpsert, { onConflict: 'username' });
+         if (chatUserError) {
+            console.error('[Supabase] Bulk chat_users upsert error:', chatUserError.message);
+         }
+      }
+    }
+  }
+}
+
+/**
+ * Log only roles (used by role backfiller)
+ */
+export async function logRolesBatch(users) {
+  if (!users || users.length === 0) return;
+
+  const chatUsersToUpsert = new Map();
+  for (const user of users) {
+    const ts = user.timestamp || new Date().toISOString();
+    const username = user.username.toLowerCase();
+    
+    chatUsersToUpsert.set(username, {
+      username: username,
+      display_name: user.displayName || user.username,
+      is_mod: user.isMod || false,
+      is_sub: user.isSub || false,
+      is_vip: user.isVip || false,
+      last_seen: ts
+    });
+  }
+
+  const uniqueChatUsers = Array.from(chatUsersToUpsert.values());
+
+  if (dbMode === 'sqlite') {
+    const insertChatUser = sqliteDb.prepare(`
+      INSERT INTO chat_users (username, display_name, is_mod, is_sub, is_vip, last_seen)
+      VALUES (?, ?, ?, ?, ?, ?)
+      ON CONFLICT(username) DO UPDATE SET
+        display_name = excluded.display_name,
+        is_mod = excluded.is_mod,
+        is_sub = excluded.is_sub,
+        is_vip = excluded.is_vip,
+        last_seen = excluded.last_seen
+      WHERE excluded.last_seen > chat_users.last_seen
+    `);
+
+    const runTransaction = sqliteDb.transaction((usersToInsert) => {
+      for (const u of usersToInsert) {
+        try {
+          insertChatUser.run(
+            u.username,
+            u.display_name,
+            u.is_mod ? 1 : 0,
+            u.is_sub ? 1 : 0,
+            u.is_vip ? 1 : 0,
+            u.last_seen
+          );
+        } catch (err) {
+          console.error('[Database] SQLite Error updating role for user:', u.username, err.message);
+        }
+      }
+    });
+
+    runTransaction(uniqueChatUsers);
+  } else {
+    // Supabase Mode
+    const { data: existingUsers } = await supabase
+      .from('chat_users')
+      .select('username, last_seen')
+      .in('username', uniqueChatUsers.map(u => u.username));
+    
+    const existingMap = new Map((existingUsers || []).map(u => [u.username, new Date(u.last_seen).getTime()]));
+    const toUpsert = uniqueChatUsers.filter(u => {
+       const oldTime = existingMap.get(u.username) || 0;
+       const newTime = new Date(u.last_seen).getTime();
+       return newTime > oldTime;
+    });
+
+    if (toUpsert.length > 0) {
+       const { error: chatUserError } = await supabase
+          .from('chat_users')
+          .upsert(toUpsert, { onConflict: 'username' });
+       if (chatUserError) {
+          console.error('[Supabase] Bulk roles upsert error:', chatUserError.message);
+       }
+    }
   }
 }
 
@@ -937,12 +1095,14 @@ export async function getTopChatters(streamId = null, limit = 50) {
       stmt = sqliteDb.prepare(`
         SELECT 
           v.username, 
-          max(v.display_name) as display_name, 
-          max(v.is_mod) as is_mod, 
-          max(v.is_sub) as is_sub, 
-          max(v.has_chatted) as has_chatted,
+          u.display_name, 
+          u.is_mod, 
+          u.is_sub,
+          u.is_vip,
+          v.has_chatted,
           (SELECT COUNT(*) FROM messages m WHERE m.stream_id = v.stream_id AND m.username = v.username) as message_count
         FROM stream_viewers v
+        LEFT JOIN chat_users u ON v.username = u.username
         WHERE v.stream_id = ? 
         GROUP BY v.username 
         ORDER BY message_count DESC 
@@ -952,14 +1112,15 @@ export async function getTopChatters(streamId = null, limit = 50) {
     } else {
       stmt = sqliteDb.prepare(`
         SELECT 
-          v.username, 
-          max(v.display_name) as display_name, 
-          max(v.is_mod) as is_mod, 
-          max(v.is_sub) as is_sub, 
-          max(v.has_chatted) as has_chatted,
-          (SELECT COUNT(*) FROM messages m WHERE m.username = v.username) as message_count
-        FROM stream_viewers v
-        GROUP BY v.username 
+          m.username, 
+          u.display_name, 
+          u.is_mod, 
+          u.is_sub,
+          u.is_vip,
+          COUNT(*) as message_count
+        FROM messages m
+        LEFT JOIN chat_users u ON m.username = u.username
+        GROUP BY m.username 
         ORDER BY message_count DESC 
         LIMIT ?
       `);

server/init_db.sql

@@ -44,6 +44,16 @@ CREATE TABLE IF NOT EXISTS stream_viewers (
 );
 CREATE INDEX IF NOT EXISTS idx_stream_viewers_stream_id ON stream_viewers(stream_id);
 
+-- 2.6 CHAT USERS TABLE
+CREATE TABLE IF NOT EXISTS chat_users (
+    username TEXT PRIMARY KEY,
+    display_name TEXT,
+    is_mod BOOLEAN DEFAULT FALSE,
+    is_sub BOOLEAN DEFAULT FALSE,
+    is_vip BOOLEAN DEFAULT FALSE,
+    last_seen TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
+);
+
 -- 3. VOICE WORDS TABLE (transcribed spoken words from streamer)
 CREATE TABLE IF NOT EXISTS voice_words (
     id SERIAL PRIMARY KEY,
@@ -98,27 +108,34 @@ CREATE OR REPLACE TRIGGER update_settings_modtime
 -- =========================================================================
 
 -- View for chatter statistics per stream
-CREATE OR REPLACE VIEW stream_chatter_stats AS
+DROP VIEW IF EXISTS stream_chatter_stats;
+CREATE VIEW stream_chatter_stats AS
 SELECT 
-    stream_id,
-    username,
-    max(display_name) as display_name,
-    bool_or(is_mod) as is_mod,
-    bool_or(is_sub) as is_sub,
-    count(*) as message_count
-FROM messages
-GROUP BY stream_id, username;
+    v.stream_id,
+    v.username,
+    u.display_name,
+    u.is_mod,
+    u.is_sub,
+    u.is_vip,
+    v.has_chatted,
+    v.first_seen,
+    (SELECT COUNT(*) FROM messages m WHERE m.stream_id = v.stream_id AND m.username = v.username) as message_count
+FROM stream_viewers v
+LEFT JOIN chat_users u ON v.username = u.username;
 
 -- View for chatter statistics overall (all-time)
-CREATE OR REPLACE VIEW global_chatter_stats AS
+DROP VIEW IF EXISTS global_chatter_stats;
+CREATE VIEW global_chatter_stats AS
 SELECT 
-    username,
-    max(display_name) as display_name,
-    bool_or(is_mod) as is_mod,
-    bool_or(is_sub) as is_sub,
+    m.username,
+    u.display_name,
+    u.is_mod,
+    u.is_sub,
+    u.is_vip,
     count(*) as message_count
-FROM messages
-GROUP BY username;
+FROM messages m
+LEFT JOIN chat_users u ON m.username = u.username
+GROUP BY m.username, u.display_name, u.is_mod, u.is_sub, u.is_vip;
 
 -- View for spoken word statistics per stream
 CREATE OR REPLACE VIEW stream_voice_word_stats AS

server/server.js

@@ -8,6 +8,7 @@ import fs from 'fs';
 import { 
   logChatMessage, 
   logChatMessagesBatch,
+  logRolesBatch,
   logVoiceWords, 
   logModAction, 
   logViewerJoin,
@@ -187,6 +188,23 @@ app.post('/api/log/messages', authenticateWorker, async (req, res) => {
   res.json({ success: true, count: messages.length });
 });
 
+// Ingest chat roles ONLY (used by role_backfiller)
+app.post('/api/log/roles', authenticateWorker, async (req, res) => {
+  const { roles } = req.body;
+  if (!roles || !Array.isArray(roles)) {
+    return res.status(400).json({ error: 'Invalid input. Expected array of roles.' });
+  }
+
+  try {
+    await logRolesBatch(roles);
+  } catch (err) {
+    console.error('Error logging batch roles:', err);
+    return res.status(500).json({ error: 'Failed to log roles' });
+  }
+
+  res.json({ success: true, count: roles.length });
+});
+
 // Endpoint for the Python worker to log viewer joins
 app.post('/api/log/viewers/join', authenticateWorker, async (req, res) => {
   const username = req.body.username;

server/supabase_migration_roles.sql

@@ -0,0 +1,57 @@
+-- 1. Create the new global users table
+CREATE TABLE IF NOT EXISTS chat_users (
+    username TEXT PRIMARY KEY,
+    display_name TEXT,
+    is_mod BOOLEAN DEFAULT FALSE,
+    is_sub BOOLEAN DEFAULT FALSE,
+    is_vip BOOLEAN DEFAULT FALSE,
+    last_seen TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
+);
+
+-- 2. Backfill existing user roles into chat_users
+-- We find the most recent message for each user and take their roles.
+-- Since the old DB only had is_mod and is_sub, we use those.
+INSERT INTO chat_users (username, display_name, is_mod, is_sub, is_vip, last_seen)
+SELECT 
+    username, 
+    max(display_name), 
+    bool_or(is_mod), 
+    bool_or(is_sub), 
+    FALSE, 
+    max(first_seen)
+FROM stream_viewers
+GROUP BY username
+ON CONFLICT (username) DO NOTHING;
+
+-- 3. We can optionally drop is_mod and is_sub from messages and stream_viewers, 
+-- but it's safer to just leave them (or drop them later). For now, we leave them to not break things.
+
+-- 4. Recreate the global_chatter_stats view to use the new global roles
+DROP VIEW IF EXISTS global_chatter_stats;
+CREATE VIEW global_chatter_stats AS
+SELECT 
+    m.username,
+    u.display_name,
+    u.is_mod,
+    u.is_sub,
+    u.is_vip,
+    count(*) as message_count
+FROM messages m
+LEFT JOIN chat_users u ON m.username = u.username
+GROUP BY m.username, u.display_name, u.is_mod, u.is_sub, u.is_vip;
+
+-- 5. Recreate the stream_chatter_stats view to use the new global roles
+DROP VIEW IF EXISTS stream_chatter_stats;
+CREATE VIEW stream_chatter_stats AS
+SELECT 
+    v.stream_id,
+    v.username,
+    u.display_name,
+    u.is_mod,
+    u.is_sub,
+    u.is_vip,
+    v.has_chatted,
+    v.first_seen,
+    (SELECT COUNT(*) FROM messages m WHERE m.stream_id = v.stream_id AND m.username = v.username) as message_count
+FROM stream_viewers v
+LEFT JOIN chat_users u ON v.username = u.username;