Optimize moderation logging: fix duplicate collisions, auto-refresh streamer OAuth tokens, and implement IRC keep-alive timeouts. Add unit tests.

local_worker/chat_mod_worker.py

@@ -76,8 +76,14 @@ def twitch_irc_listener():
             
             buffer = ""
             irc_sock.setblocking(False)
+            last_msg_time = time.time()
             
             while not stop_flag.is_set():
+                # Check for network timeout (no messages for 5 minutes)
+                if time.time() - last_msg_time > 300:
+                    print("[Twitch IRC] Network timeout (no messages for 5 minutes). Reconnecting...")
+                    break
+
                 ready = select.select([irc_sock], [], [], 1.0)
                 if not ready[0]:
                     continue
@@ -91,6 +97,7 @@ def twitch_irc_listener():
                     print("[Twitch IRC] Connection closed by remote host.")
                     break
                 
+                last_msg_time = time.time()
                 buffer += data
                 while "\r\n" in buffer:
                     line, buffer = buffer.split("\r\n", 1)

local_worker/worker.py

@@ -102,8 +102,14 @@ def twitch_chat_listener():
             
             buffer = ""
             irc_sock.setblocking(False)
+            last_msg_time = time.time()
             
             while not stop_flag.is_set():
+                # Check for network timeout (no messages for 5 minutes)
+                if time.time() - last_msg_time > 300:
+                    print("[Twitch IRC] Network timeout (no messages for 5 minutes). Reconnecting...")
+                    break
+
                 # Use select for non-blocking read with timeout to allow exit checks
                 ready = select.select([irc_sock], [], [], 1.0)
                 if not ready[0]:
@@ -118,6 +124,7 @@ def twitch_chat_listener():
                     print("[Twitch IRC] Connection closed by remote host.")
                     break
                 
+                last_msg_time = time.time()
                 buffer += data
                 while "\r\n" in buffer:
                     line, buffer = buffer.split("\r\n", 1)

server/db.js

@@ -30,7 +30,7 @@ console.log(`[Database] Initializing in Mode: ${dbMode.toUpperCase()}`);
 let sqliteDb = null;
 
 if (dbMode === 'sqlite') {
-  const dbPath = path.resolve('database.db');
+  const dbPath = path.resolve(process.env.SQLITE_DB_PATH || 'database.db');
   console.log(`[Database] SQLite file location: ${dbPath}`);
   sqliteDb = new Database(dbPath);
   
@@ -625,18 +625,23 @@ export async function logModAction(action) {
   const streamId = stream ? stream.id : null;
 
   // Prevent duplicate actions (e.g., from both EventSub and local IRC worker running together)
+  let duplicateToUpdate = null;
   if (action.targetUser) {
     const targetUserLower = action.targetUser.toLowerCase();
     if (dbMode === 'sqlite') {
       const existing = sqliteDb.prepare(`
-        SELECT id FROM mod_actions 
+        SELECT id, moderator FROM mod_actions 
         WHERE action_type = ? 
           AND target_user = ? 
           AND abs(strftime('%s', timestamp) - strftime('%s', ?)) < 15
       `).get(action.actionType, targetUserLower, time);
       if (existing) {
-        console.log(`[Database] Duplicate SQLite mod action ${action.actionType} for ${targetUserLower} ignored.`);
-        return;
+        if (existing.moderator === 'TwitchIRC' && action.moderator !== 'TwitchIRC') {
+          duplicateToUpdate = existing.id;
+        } else {
+          console.log(`[Database] Duplicate SQLite mod action ${action.actionType} for ${targetUserLower} ignored.`);
+          return;
+        }
       }
     } else {
       const timeMs = new Date(time).getTime();
@@ -645,7 +650,7 @@ export async function logModAction(action) {
       
       const { data: existing, error } = await supabase
         .from('mod_actions')
-        .select('id')
+        .select('id, moderator')
         .eq('action_type', action.actionType)
         .eq('target_user', targetUserLower)
         .gte('timestamp', startTimeRange)
@@ -653,8 +658,13 @@ export async function logModAction(action) {
         .limit(1);
         
       if (!error && existing && existing.length > 0) {
-        console.log(`[Database] Duplicate Supabase mod action ${action.actionType} for ${targetUserLower} ignored.`);
-        return;
+        const first = existing[0];
+        if (first.moderator === 'TwitchIRC' && action.moderator !== 'TwitchIRC') {
+          duplicateToUpdate = first.id;
+        } else {
+          console.log(`[Database] Duplicate Supabase mod action ${action.actionType} for ${targetUserLower} ignored.`);
+          return;
+        }
       }
     }
   }
@@ -706,36 +716,66 @@ export async function logModAction(action) {
   }
 
   if (dbMode === 'sqlite') {
-    sqliteDb.prepare(`
-      INSERT INTO mod_actions (stream_id, action_type, moderator, target_user, duration, reason, message_text, timestamp, reaction_time)
-      VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
-    `).run(
-      streamId,
-      action.actionType,
-      action.moderator,
-      action.targetUser ? action.targetUser.toLowerCase() : null,
-      action.duration || null,
-      action.reason || null,
-      action.messageText || null,
-      time,
-      reactionTime
-    );
+    if (duplicateToUpdate) {
+      sqliteDb.prepare(`
+        UPDATE mod_actions 
+        SET moderator = ?, reason = ?, reaction_time = ?
+        WHERE id = ?
+      `).run(
+        action.moderator,
+        action.reason || null,
+        reactionTime,
+        duplicateToUpdate
+      );
+      console.log(`[Database] Duplicate SQLite mod action ${action.actionType} updated with real moderator ${action.moderator}.`);
+    } else {
+      sqliteDb.prepare(`
+        INSERT INTO mod_actions (stream_id, action_type, moderator, target_user, duration, reason, message_text, timestamp, reaction_time)
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+      `).run(
+        streamId,
+        action.actionType,
+        action.moderator,
+        action.targetUser ? action.targetUser.toLowerCase() : null,
+        action.duration || null,
+        action.reason || null,
+        action.messageText || null,
+        time,
+        reactionTime
+      );
+    }
   } else {
-    const { error } = await supabase
-      .from('mod_actions')
-      .insert([{
-        stream_id: streamId,
-        action_type: action.actionType,
-        moderator: action.moderator,
-        target_user: action.targetUser ? action.targetUser.toLowerCase() : null,
-        duration: action.duration || null,
-        reason: action.reason || null,
-        message_text: action.messageText || null,
-        timestamp: time,
-        reaction_time: reactionTime
-      }]);
-    if (error) {
-      console.error('[Supabase] Error logging mod action:', error);
+    if (duplicateToUpdate) {
+      const { error } = await supabase
+        .from('mod_actions')
+        .update({
+          moderator: action.moderator,
+          reason: action.reason || null,
+          reaction_time: reactionTime
+        })
+        .eq('id', duplicateToUpdate);
+      if (error) {
+        console.error('[Supabase] Error updating mod action duplicate:', error);
+      } else {
+        console.log(`[Database] Duplicate Supabase mod action ${action.actionType} updated with real moderator ${action.moderator}.`);
+      }
+    } else {
+      const { error } = await supabase
+        .from('mod_actions')
+        .insert([{
+          stream_id: streamId,
+          action_type: action.actionType,
+          moderator: action.moderator,
+          target_user: action.targetUser ? action.targetUser.toLowerCase() : null,
+          duration: action.duration || null,
+          reason: action.reason || null,
+          message_text: action.messageText || null,
+          timestamp: time,
+          reaction_time: reactionTime
+        }]);
+      if (error) {
+        console.error('[Supabase] Error logging mod action:', error);
+      }
     }
   }
 }

server/eventsub.js

@@ -1,6 +1,6 @@
 import WebSocket from 'ws';
 import axios from 'axios';
-import { getSetting, logModAction } from './db.js';
+import { getSetting, setSetting, logModAction } from './db.js';
 
 let ws = null;
 let broadcasterUserId = null;
@@ -8,6 +8,59 @@ let keepAliveTimer = null;
 let lastKeepAlive = null;
 let reconnecting = false;
 
+/**
+ * Validate current access token and refresh it if expired/invalid
+ */
+export async function getOrRefreshAccessToken() {
+  let token = await getSetting('twitch_streamer_access_token');
+  const refreshToken = await getSetting('twitch_streamer_refresh_token');
+
+  if (!token) {
+    return null;
+  }
+
+  // Validate the current token via Twitch API
+  try {
+    await axios.get('https://id.twitch.tv/oauth2/validate', {
+      headers: {
+        'Authorization': `Bearer ${token}`
+      }
+    });
+    // Token is valid!
+    return token;
+  } catch (err) {
+    const status = err.response?.status;
+    console.log(`[EventSub] Access token validation failed (Status: ${status}). Attempting to refresh...`);
+
+    if (!refreshToken) {
+      console.error('[EventSub] Refresh token is missing in database. Cannot refresh access token.');
+      return null;
+    }
+
+    try {
+      const response = await axios.post('https://id.twitch.tv/oauth2/token', null, {
+        params: {
+          client_id: process.env.TWITCH_CLIENT_ID,
+          client_secret: process.env.TWITCH_CLIENT_SECRET,
+          grant_type: 'refresh_token',
+          refresh_token: refreshToken
+        }
+      });
+
+      const { access_token, refresh_token: newRefreshToken } = response.data;
+      await setSetting('twitch_streamer_access_token', access_token);
+      if (newRefreshToken) {
+        await setSetting('twitch_streamer_refresh_token', newRefreshToken);
+      }
+      console.log('[EventSub] Access token refreshed successfully.');
+      return access_token;
+    } catch (refreshErr) {
+      console.error('[EventSub] Token refresh request failed:', refreshErr.response?.data || refreshErr.message);
+      return null;
+    }
+  }
+}
+
 /**
  * Initialize Twitch EventSub WebSocket connection
  */
@@ -24,9 +77,9 @@ export async function initializeEventSub(streamerId = null) {
     return;
   }
 
-  const token = await getSetting('twitch_streamer_access_token');
+  const token = await getOrRefreshAccessToken();
   if (!token) {
-    console.log('[EventSub] Streamer access token is missing. Waiting for Streamer login.');
+    console.log('[EventSub] Streamer access token is missing or invalid. Waiting for Streamer login.');
     return;
   }
 
@@ -168,7 +221,7 @@ async function handleNotification(payload) {
  */
 async function subscribeToEvents(sessionId) {
   const clientId = process.env.TWITCH_CLIENT_ID;
-  const token = await getSetting('twitch_streamer_access_token');
+  const token = await getOrRefreshAccessToken();
 
   if (!clientId || !token) {
     console.error('[EventSub] Cannot subscribe: Client ID or token is missing!');

server/scratch/test_mod_duplicates.js

@@ -0,0 +1,119 @@
+// Setup testing database environment variables BEFORE any imports
+process.env.DB_MODE = 'sqlite';
+process.env.SQLITE_DB_PATH = './test_database.db';
+// Clear Supabase keys to force SQLite mode in db.js
+process.env.SUPABASE_URL = '';
+process.env.SUPABASE_KEY = '';
+process.env.SUPABASE_SERVICE_ROLE_KEY = '';
+
+import Database from 'better-sqlite3';
+import fs from 'fs';
+
+// Clean up test database
+if (fs.existsSync('./test_database.db')) {
+  fs.unlinkSync('./test_database.db');
+}
+
+const sqliteDb = new Database('./test_database.db');
+
+// Create minimal schema for testing
+sqliteDb.exec(`
+  CREATE TABLE IF NOT EXISTS streams (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    start_time TEXT,
+    end_time TEXT,
+    title TEXT,
+    twitch_stream_id TEXT UNIQUE,
+    category TEXT
+  );
+
+  CREATE TABLE IF NOT EXISTS mod_actions (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    stream_id INTEGER,
+    action_type TEXT,
+    moderator TEXT,
+    target_user TEXT,
+    duration INTEGER,
+    reason TEXT,
+    message_text TEXT,
+    timestamp TEXT,
+    reaction_time REAL
+  );
+`);
+
+// Insert a fake stream
+sqliteDb.prepare(`
+  INSERT INTO streams (id, start_time, title, twitch_stream_id, category)
+  VALUES (1, datetime('now'), 'Test Stream', 'test-stream-123', 'Testing')
+`).run();
+
+// Dynamically import db.js so it reads the env vars we just set
+const { logModAction } = await import('../db.js');
+
+async function runTest() {
+  console.log('--- STARTING DUPLICATE MERGE TEST ---');
+  
+  const timestamp = new Date().toISOString();
+  
+  // 1. Simulate IRC event (anonymous, moderator = TwitchIRC, no reason)
+  console.log('1. Logging IRC mod action (anonymous)...');
+  await logModAction({
+    actionType: 'timeout',
+    moderator: 'TwitchIRC',
+    targetUser: 'spammer_user',
+    duration: 600,
+    timestamp: timestamp
+  });
+  
+  // Verify it exists in DB
+  let record = sqliteDb.prepare('SELECT * FROM mod_actions WHERE target_user = ?').get('spammer_user');
+  console.log('Record after IRC:', record);
+  if (!record || record.moderator !== 'TwitchIRC' || record.reason !== null) {
+    console.error('FAIL: Initial IRC record is incorrect.');
+    process.exit(1);
+  }
+  
+  // 2. Simulate EventSub event (real moderator, custom reason, same timestamp)
+  console.log('2. Logging EventSub mod action (with real mod & reason)...');
+  await logModAction({
+    actionType: 'timeout',
+    moderator: 'GoodModName',
+    targetUser: 'spammer_user',
+    duration: 600,
+    reason: 'Links spamming in chat',
+    timestamp: timestamp
+  });
+  
+  // Verify it updated in DB instead of creating a duplicate or ignoring
+  const records = sqliteDb.prepare('SELECT * FROM mod_actions WHERE target_user = ?').all('spammer_user');
+  console.log('All records in DB after EventSub:', records);
+  
+  if (records.length !== 1) {
+    console.error(`FAIL: Expected exactly 1 record, found ${records.length}`);
+    process.exit(1);
+  }
+  
+  record = records[0];
+  if (record.moderator !== 'goodmodname' && record.moderator !== 'GoodModName') { // depending on casing
+    console.error(`FAIL: Expected moderator to be GoodModName/goodmodname, got: ${record.moderator}`);
+    process.exit(1);
+  }
+  
+  if (record.reason !== 'Links spamming in chat') {
+    console.error(`FAIL: Expected reason "Links spamming in chat", got: ${record.reason}`);
+    process.exit(1);
+  }
+  
+  console.log('SUCCESS: Duplicate mod action successfully merged/enriched!');
+  
+  // Clean up
+  sqliteDb.close();
+  if (fs.existsSync('./test_database.db')) {
+    fs.unlinkSync('./test_database.db');
+  }
+}
+
+runTest().catch(err => {
+  console.error('Test crashed:', err);
+  process.exit(1);
+});

server/scratch/test_token_refresh.js

@@ -0,0 +1,172 @@
+// Setup testing database environment variables BEFORE any imports
+process.env.DB_MODE = 'sqlite';
+process.env.SQLITE_DB_PATH = './test_database.db';
+// Clear Supabase keys to force SQLite mode
+process.env.SUPABASE_URL = '';
+process.env.SUPABASE_KEY = '';
+process.env.SUPABASE_SERVICE_ROLE_KEY = '';
+
+// Mock environment variables for Twitch client
+process.env.TWITCH_CLIENT_ID = 'mocked_client_id';
+process.env.TWITCH_CLIENT_SECRET = 'mocked_client_secret';
+
+import Database from 'better-sqlite3';
+import fs from 'fs';
+import axios from 'axios';
+
+// Clean up test database
+if (fs.existsSync('./test_database.db')) {
+  fs.unlinkSync('./test_database.db');
+}
+
+const sqliteDb = new Database('./test_database.db');
+
+// Create settings table
+sqliteDb.exec(`
+  CREATE TABLE IF NOT EXISTS settings (
+    key TEXT PRIMARY KEY,
+    value TEXT NOT NULL
+  );
+`);
+
+// Insert initial expired token and refresh token
+sqliteDb.prepare(`
+  INSERT INTO settings (key, value)
+  VALUES ('twitch_streamer_access_token', 'expired_token_value')
+`).run();
+
+sqliteDb.prepare(`
+  INSERT INTO settings (key, value)
+  VALUES ('twitch_streamer_refresh_token', 'valid_refresh_token_value')
+`).run();
+
+// Mock Axios methods
+let validateCalls = 0;
+let refreshCalls = 0;
+let refreshParamsReceived = null;
+
+axios.get = async (url, config) => {
+  if (url === 'https://id.twitch.tv/oauth2/validate') {
+    validateCalls++;
+    const authHeader = config?.headers?.Authorization;
+    if (authHeader === 'Bearer valid_mocked_token') {
+      // Mock success for valid token
+      return { data: { client_id: 'mocked_client_id' } };
+    } else {
+      // Mock expiration/invalid token error
+      const err = new Error('Unauthorized');
+      err.response = { status: 401, data: { message: 'invalid oauth token' } };
+      throw err;
+    }
+  }
+  throw new Error(`Unexpected GET request to ${url}`);
+};
+
+axios.post = async (url, data, config) => {
+  if (url === 'https://id.twitch.tv/oauth2/token') {
+    refreshCalls++;
+    refreshParamsReceived = config?.params;
+    return {
+      data: {
+        access_token: 'new_refreshed_access_token',
+        refresh_token: 'new_refreshed_refresh_token'
+      }
+    };
+  }
+  throw new Error(`Unexpected POST request to ${url}`);
+};
+
+// Dynamically import db.js and eventsub.js so they read correct environment variables and use the mocked axios
+const { getSetting } = await import('../db.js');
+const { getOrRefreshAccessToken } = await import('../eventsub.js');
+
+async function runTest() {
+  console.log('--- STARTING TOKEN REFRESH TEST ---');
+
+  // Scenario 1: Initial call with expired token. Should fail validate, call refresh, and return new token.
+  console.log('Scenario 1: Testing token validation failure and refresh flow...');
+  const token = await getOrRefreshAccessToken();
+
+  console.log('Returned token:', token);
+  console.log('Validate calls:', validateCalls);
+  console.log('Refresh calls:', refreshCalls);
+  console.log('Params received on refresh:', refreshParamsReceived);
+
+  // Assertions
+  if (token !== 'new_refreshed_access_token') {
+    console.error('FAIL: Expected new refreshed access token, got:', token);
+    process.exit(1);
+  }
+  if (validateCalls !== 1) {
+    console.error('FAIL: Expected 1 validate call, got:', validateCalls);
+    process.exit(1);
+  }
+  if (refreshCalls !== 1) {
+    console.error('FAIL: Expected 1 refresh call, got:', refreshCalls);
+    process.exit(1);
+  }
+  if (!refreshParamsReceived || refreshParamsReceived.refresh_token !== 'valid_refresh_token_value') {
+    console.error('FAIL: Incorrect refresh token sent in params:', refreshParamsReceived);
+    process.exit(1);
+  }
+
+  // Verify DB updated settings
+  const dbAccessToken = sqliteDb.prepare("SELECT value FROM settings WHERE key = 'twitch_streamer_access_token'").get()?.value;
+  const dbRefreshToken = sqliteDb.prepare("SELECT value FROM settings WHERE key = 'twitch_streamer_refresh_token'").get()?.value;
+
+  console.log('DB Access Token:', dbAccessToken);
+  console.log('DB Refresh Token:', dbRefreshToken);
+
+  if (dbAccessToken !== 'new_refreshed_access_token' || dbRefreshToken !== 'new_refreshed_refresh_token') {
+    console.error('FAIL: Settings not updated in database correctly');
+    process.exit(1);
+  }
+
+  // Scenario 2: Subsequent call with the now valid token. Should validate successfully and not call refresh.
+  console.log('Scenario 2: Testing valid token bypasses refresh...');
+  // Update mock token validation check to succeed for this new token
+  axios.get = async (url, config) => {
+    if (url === 'https://id.twitch.tv/oauth2/validate') {
+      validateCalls++;
+      const authHeader = config?.headers?.Authorization;
+      if (authHeader === 'Bearer new_refreshed_access_token') {
+        return { data: { client_id: 'mocked_client_id' } };
+      }
+      const err = new Error('Unauthorized');
+      err.response = { status: 401 };
+      throw err;
+    }
+    throw new Error(`Unexpected GET request to ${url}`);
+  };
+
+  const tokenScenario2 = await getOrRefreshAccessToken();
+  console.log('Scenario 2 Returned token:', tokenScenario2);
+  console.log('Scenario 2 Validate calls total:', validateCalls);
+  console.log('Scenario 2 Refresh calls total:', refreshCalls);
+
+  if (tokenScenario2 !== 'new_refreshed_access_token') {
+    console.error('FAIL: Expected token to remain new_refreshed_access_token, got:', tokenScenario2);
+    process.exit(1);
+  }
+  if (validateCalls !== 2) {
+    console.error('FAIL: Expected validate calls to increment to 2, got:', validateCalls);
+    process.exit(1);
+  }
+  if (refreshCalls !== 1) {
+    console.error('FAIL: Expected refresh calls to remain 1, got:', refreshCalls);
+    process.exit(1);
+  }
+
+  console.log('SUCCESS: Token refresh flow successfully verified and tested!');
+
+  // Clean up
+  sqliteDb.close();
+  if (fs.existsSync('./test_database.db')) {
+    fs.unlinkSync('./test_database.db');
+  }
+}
+
+runTest().catch(err => {
+  console.error('Test crashed:', err);
+  process.exit(1);
+});