import { createClient } from '@supabase/supabase-js'; import Database from 'better-sqlite3'; import dotenv from 'dotenv'; import path from 'path'; import fs from 'fs'; import WebSocket from 'ws'; // Polyfill WebSocket for Supabase Realtime in Node.js < 22 globalThis.WebSocket = WebSocket; dotenv.config(); // Detect mode const supabaseUrl = process.env.SUPABASE_URL; const supabaseKey = process.env.SUPABASE_SERVICE_ROLE_KEY || process.env.SUPABASE_KEY; // We use Supabase if keys are provided and don't look like default placeholders const useSupabase = supabaseUrl && supabaseKey && !supabaseUrl.includes('YOUR_PROJECT_ID') && !supabaseKey.includes('your-supabase'); export const dbMode = useSupabase ? 'supabase' : 'sqlite'; console.log(`[Database] Initializing in Mode: ${dbMode.toUpperCase()}`); // ========================================================================= // SQLITE SETUP (Local development fallback) // ========================================================================= let sqliteDb = null; if (dbMode === 'sqlite') { const dbPath = path.resolve(process.env.SQLITE_DB_PATH || 'database.db'); console.log(`[Database] SQLite file location: ${dbPath}`); sqliteDb = new Database(dbPath); // Create tables in SQLite if they don't exist sqliteDb.exec(` CREATE TABLE IF NOT EXISTS streams ( id INTEGER PRIMARY KEY AUTOINCREMENT, start_time TEXT DEFAULT CURRENT_TIMESTAMP, end_time TEXT, title TEXT, twitch_stream_id TEXT UNIQUE, category TEXT, backfill_status TEXT DEFAULT 'pending', twitch_vod_id TEXT ); CREATE TABLE IF NOT EXISTS messages ( id TEXT PRIMARY KEY, stream_id INTEGER, username TEXT NOT NULL, display_name TEXT, message TEXT NOT NULL, timestamp TEXT DEFAULT CURRENT_TIMESTAMP, is_streamer INTEGER DEFAULT 0, is_mod INTEGER DEFAULT 0, is_sub INTEGER DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_messages_stream_id ON messages(stream_id); CREATE INDEX IF NOT EXISTS idx_messages_username ON messages(username); CREATE INDEX IF NOT EXISTS idx_messages_timestamp ON messages(timestamp); CREATE TABLE IF NOT EXISTS voice_words ( id INTEGER PRIMARY KEY AUTOINCREMENT, stream_id INTEGER, word TEXT NOT NULL, timestamp TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_voice_words_stream_id ON voice_words(stream_id); CREATE INDEX IF NOT EXISTS idx_voice_words_word ON voice_words(word); CREATE TABLE IF NOT EXISTS mod_actions ( id INTEGER PRIMARY KEY AUTOINCREMENT, stream_id INTEGER, action_type TEXT NOT NULL, moderator TEXT NOT NULL, target_user TEXT, duration INTEGER, reason TEXT, message_text TEXT, timestamp TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_mod_actions_stream_id ON mod_actions(stream_id); CREATE INDEX IF NOT EXISTS idx_mod_actions_timestamp ON mod_actions(timestamp); CREATE TABLE IF NOT EXISTS settings ( 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.'); // Migrations for SQLite try { const columnsStreams = sqliteDb.prepare("PRAGMA table_info(streams)").all(); if (!columnsStreams.some(c => c.name === 'twitch_stream_id')) { sqliteDb.exec("ALTER TABLE streams ADD COLUMN twitch_stream_id TEXT"); try { sqliteDb.exec("CREATE UNIQUE INDEX IF NOT EXISTS idx_streams_twitch_stream_id ON streams(twitch_stream_id)"); } catch (idxErr) { // ignore } console.log("[Database] SQLite Migrated: Added twitch_stream_id to streams"); } if (!columnsStreams.some(c => c.name === 'category')) { sqliteDb.exec("ALTER TABLE streams ADD COLUMN category TEXT"); console.log("[Database] SQLite Migrated: Added category to streams"); } if (!columnsStreams.some(c => c.name === 'backfill_status')) { sqliteDb.exec("ALTER TABLE streams ADD COLUMN backfill_status TEXT DEFAULT 'pending'"); console.log("[Database] SQLite Migrated: Added backfill_status to streams"); } if (!columnsStreams.some(c => c.name === 'twitch_vod_id')) { sqliteDb.exec("ALTER TABLE streams ADD COLUMN twitch_vod_id TEXT"); console.log("[Database] SQLite Migrated: Added twitch_vod_id to streams"); } const columnsModActions = sqliteDb.prepare("PRAGMA table_info(mod_actions)").all(); if (!columnsModActions.some(c => c.name === 'reaction_time')) { sqliteDb.exec("ALTER TABLE mod_actions ADD COLUMN reaction_time REAL"); console.log("[Database] SQLite Migrated: Added reaction_time to mod_actions"); } } catch (err) { console.error("[Database] SQLite Migration error:", err.message); } } // ========================================================================= // SUPABASE CLIENT (Cloud production mode) // ========================================================================= export const supabase = useSupabase ? createClient(supabaseUrl, supabaseKey) : null; // Gaps larger than 2 hours (in ms) will trigger a new stream session automatically const STREAM_SESSION_GAP = 2 * 60 * 60 * 1000; let activeStreamCache = null; let lastMessageTime = null; // ========================================================================= // HELPER METHODS (Dually implemented for SQLite & Supabase) // ========================================================================= /** * Get active stream or auto-start a new one */ export async function getOrStartActiveStream() { const now = new Date(); // 1. Check cache first if (activeStreamCache) { if (lastMessageTime && (now - lastMessageTime < STREAM_SESSION_GAP)) { lastMessageTime = now; return activeStreamCache; } else if (lastMessageTime) { console.log('[Database] Stream session gap exceeded. Closing old stream.'); await endActiveStream(activeStreamCache.id); activeStreamCache = null; } } // 2. Fetch from DB if (dbMode === 'sqlite') { const row = sqliteDb.prepare('SELECT * FROM streams WHERE end_time IS NULL ORDER BY start_time DESC LIMIT 1').get(); if (row) { activeStreamCache = row; lastMessageTime = now; return activeStreamCache; } // Auto-create new stream in SQLite console.log('[Database] No active stream found. Auto-starting new stream in SQLite.'); const title = `Stream ${now.toLocaleDateString('ru-RU')} - ${now.toLocaleTimeString('ru-RU', { hour: '2-digit', minute: '2-digit' })}`; const info = sqliteDb.prepare("INSERT INTO streams (title, start_time, backfill_status) VALUES (?, ?, 'live')").run(title, now.toISOString()); activeStreamCache = { id: info.lastInsertRowid, title, start_time: now.toISOString(), end_time: null, twitch_stream_id: null, category: null, backfill_status: 'live' }; lastMessageTime = now; return activeStreamCache; } else { // Supabase Mode const { data: activeStreams, error } = await supabase .from('streams') .select('*') .is('end_time', null) .order('start_time', { ascending: false }) .limit(1); if (error) { console.error('[Supabase] Error fetching active stream:', error); return null; } if (activeStreams && activeStreams.length > 0) { activeStreamCache = activeStreams[0]; lastMessageTime = now; return activeStreamCache; } // Auto-create new stream in Supabase console.log('[Database] No active stream found. Auto-starting new stream in Supabase.'); const title = `Stream ${now.toLocaleDateString('ru-RU')} - ${now.toLocaleTimeString('ru-RU', { hour: '2-digit', minute: '2-digit' })}`; const { data: newStream, error: createError } = await supabase .from('streams') .insert([{ title, start_time: now.toISOString(), backfill_status: 'live' }]) .select() .single(); if (createError) { console.error('[Supabase] Error starting stream:', createError); return null; } activeStreamCache = newStream; lastMessageTime = now; return activeStreamCache; } } /** * Get the current active stream without starting a new one */ export async function getActiveStream() { if (activeStreamCache) { return activeStreamCache; } if (dbMode === 'sqlite') { const row = sqliteDb.prepare('SELECT * FROM streams WHERE end_time IS NULL ORDER BY start_time DESC LIMIT 1').get(); if (row) { activeStreamCache = row; } return row || null; } else { const { data: activeStreams, error } = await supabase .from('streams') .select('*') .is('end_time', null) .order('start_time', { ascending: false }) .limit(1); if (error) { console.error('[Supabase] Error fetching active stream:', error); return null; } if (activeStreams && activeStreams.length > 0) { activeStreamCache = activeStreams[0]; return activeStreamCache; } return null; } } /** * Check if a stream with given twitch_stream_id already exists in the database */ export async function checkIfStreamExists(twitchStreamId) { if (dbMode === 'sqlite') { const row = sqliteDb.prepare('SELECT id FROM streams WHERE twitch_stream_id = ?').get(twitchStreamId); return !!row; } else { const { data, error } = await supabase .from('streams') .select('id') .eq('twitch_stream_id', twitchStreamId) .limit(1); if (error) { console.error('[Supabase] Error checking if stream exists:', error); return false; } return data && data.length > 0; } } /** * Sync active stream session with Twitch API data (status, category, started_at) */ export async function syncActiveStream(twitchStreamId, title, category, startTimeIso) { if (dbMode === 'sqlite') { // Check if stream with this twitch_stream_id exists let stream = sqliteDb.prepare('SELECT * FROM streams WHERE twitch_stream_id = ?').get(twitchStreamId); if (stream) { if (stream.end_time) { sqliteDb.prepare("UPDATE streams SET end_time = NULL, category = ?, title = ?, backfill_status = 'live' WHERE id = ?").run(category, title, stream.id); stream.end_time = null; stream.category = category; stream.title = title; } else { sqliteDb.prepare("UPDATE streams SET category = ?, title = ?, backfill_status = 'live' WHERE id = ?").run(category, title, stream.id); stream.category = category; stream.title = title; } activeStreamCache = stream; lastMessageTime = new Date(); return stream; } // If it doesn't exist, check if there is an active stream with no twitch_stream_id let activeStream = sqliteDb.prepare('SELECT * FROM streams WHERE end_time IS NULL AND twitch_stream_id IS NULL ORDER BY start_time DESC LIMIT 1').get(); if (activeStream) { sqliteDb.prepare("UPDATE streams SET twitch_stream_id = ?, category = ?, title = ?, backfill_status = 'live' WHERE id = ?").run(twitchStreamId, category, title, activeStream.id); activeStream.twitch_stream_id = twitchStreamId; activeStream.category = category; activeStream.title = title; activeStreamCache = activeStream; lastMessageTime = new Date(); return activeStream; } // Otherwise, create a new stream session const info = sqliteDb.prepare("INSERT INTO streams (twitch_stream_id, title, category, start_time, backfill_status) VALUES (?, ?, ?, ?, 'live')").run(twitchStreamId, title, category, startTimeIso || new Date().toISOString()); activeStreamCache = { id: info.lastInsertRowid, twitch_stream_id: twitchStreamId, title, category, start_time: startTimeIso || new Date().toISOString(), end_time: null, backfill_status: 'live' }; lastMessageTime = new Date(); return activeStreamCache; } else { // Supabase Mode let { data: existingStreams, error: fetchErr } = await supabase .from('streams') .select('*') .eq('twitch_stream_id', twitchStreamId) .limit(1); if (existingStreams && existingStreams.length > 0) { let stream = existingStreams[0]; if (stream.end_time) { const { data: updated, error } = await supabase .from('streams') .update({ end_time: null, category, title, backfill_status: 'live' }) .eq('id', stream.id) .select() .single(); if (!error) stream = updated; } else { const { data: updated, error } = await supabase .from('streams') .update({ category, title, backfill_status: 'live' }) .eq('id', stream.id) .select() .single(); if (!error) stream = updated; } activeStreamCache = stream; lastMessageTime = new Date(); return stream; } let { data: activeStreams } = await supabase .from('streams') .select('*') .is('end_time', null) .is('twitch_stream_id', null) .order('start_time', { ascending: false }) .limit(1); if (activeStreams && activeStreams.length > 0) { let activeStream = activeStreams[0]; const { data: updated, error } = await supabase .from('streams') .update({ twitch_stream_id: twitchStreamId, category, title, backfill_status: 'live' }) .eq('id', activeStream.id) .select() .single(); if (!error) activeStream = updated; activeStreamCache = activeStream; lastMessageTime = new Date(); return activeStream; } const { data: newStream, error: insertErr } = await supabase .from('streams') .insert([{ twitch_stream_id: twitchStreamId, title, category, start_time: startTimeIso || new Date().toISOString(), backfill_status: 'live' }]) .select() .single(); if (insertErr) { console.error('[Supabase] Error syncing stream insertion:', insertErr); return null; } activeStreamCache = newStream; lastMessageTime = new Date(); return activeStreamCache; } } /** * End an active stream */ export async function endActiveStream(streamId = null, endTime = null, twitchVodId = null) { const idToClose = streamId || (activeStreamCache ? activeStreamCache.id : null); if (!idToClose) return; const endTimeIso = endTime || new Date().toISOString(); if (dbMode === 'sqlite') { if (twitchVodId) { sqliteDb.prepare("UPDATE streams SET end_time = ?, backfill_status = 'pending', twitch_vod_id = ? WHERE id = ?").run(endTimeIso, twitchVodId, idToClose); } else { sqliteDb.prepare("UPDATE streams SET end_time = ?, backfill_status = 'pending' WHERE id = ?").run(endTimeIso, idToClose); } console.log(`[Database] SQLite Stream ID ${idToClose} ended. VOD: ${twitchVodId || 'None'}`); } else { const updateData = { end_time: endTimeIso, backfill_status: 'pending' }; if (twitchVodId) updateData.twitch_vod_id = twitchVodId; const { error } = await supabase .from('streams') .update(updateData) .eq('id', idToClose); if (error) { console.error('[Supabase] Error ending stream:', error); } else { console.log(`[Database] Supabase Stream ID ${idToClose} ended. VOD: ${twitchVodId || 'None'}`); } } if (activeStreamCache && activeStreamCache.id === idToClose) { activeStreamCache = null; lastMessageTime = null; } } /** * Find stream active at a specific timestamp */ export async function getStreamAtTimestamp(timestampIso) { const time = new Date(timestampIso); if (dbMode === 'sqlite') { const stream = sqliteDb.prepare(` SELECT * FROM streams WHERE datetime(start_time) <= datetime(?) AND (end_time IS NULL OR datetime(end_time) >= datetime(?)) ORDER BY start_time DESC LIMIT 1 `).get(timestampIso, timestampIso); return stream || null; } else { const { data, error } = await supabase .from('streams') .select('*') .lte('start_time', timestampIso) .order('start_time', { ascending: false }); if (error) { console.error('[Supabase] Error finding stream at timestamp:', error); return null; } const stream = data.find(s => !s.end_time || new Date(s.end_time) >= time); return stream || null; } } /** * Log chat message */ export async function logChatMessage(msg) { const time = msg.timestamp || new Date().toISOString(); let stream = null; if (msg.timestamp) { stream = await getStreamAtTimestamp(msg.timestamp); } if (!stream) { stream = await getActiveStream(); } const streamId = stream ? stream.id : null; if (dbMode === 'sqlite') { try { sqliteDb.prepare(` INSERT INTO messages (id, stream_id, username, display_name, message, timestamp, is_streamer, is_mod, is_sub) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) `).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) { // Ignore duplicate keys if (!err.message.includes('UNIQUE constraint failed')) { console.error('[Database] SQLite Error saving message:', err.message); } } // Update stream_viewers table try { 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 `).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 Error updating stream_viewers:', err.message); } } else { const { error } = await supabase .from('messages') .insert([{ 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 }]); if (error && error.code !== '23505') { console.error('[Supabase] Error logging message:', error); } // Update stream_viewers const { error: viewerError } = await supabase .from('stream_viewers') .upsert({ 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() }, { onConflict: 'stream_id, username' }); if (viewerError) { console.error('[Supabase] Error updating stream_viewers:', viewerError); } } } /** * 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) 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 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); try { insertMsg.run( msg.id, streamId, msg.username.toLowerCase(), msg.displayName || msg.username, msg.message, msg.timestamp || new Date().toISOString(), msg.isStreamer ? 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); } 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); } } }); runTransaction(messages); } else { // 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: username, display_name: msg.displayName || msg.username, message: msg.message, timestamp: ts, is_streamer: msg.isStreamer || false }); const viewerKey = `${streamId || 'null'}-${username}`; 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: 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 }); } // 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); } // 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()])); // Always update roles (is_mod/is_sub/is_vip), but only advance last_seen if newer const toUpsert = uniqueChatUsers.map(u => { const oldTime = existingMap.get(u.username) || 0; const newTime = new Date(u.last_seen).getTime(); return { ...u, // Keep last_seen as the newer of the two last_seen: newTime > oldTime ? u.last_seen : (existingUsers || []).find(e => e.username === u.username)?.last_seen || u.last_seen }; }); 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); } } } } /** * Log voice words (bulk insert) */ export async function logVoiceWords(words, timestamp) { if (!words || words.length === 0) return; const time = timestamp || new Date().toISOString(); let stream = null; if (timestamp) { stream = await getStreamAtTimestamp(timestamp); } if (!stream) { stream = await getActiveStream(); } const clientId = process.env.TWITCH_CLIENT_ID; const clientSecret = process.env.TWITCH_CLIENT_SECRET; const twitchConfigured = !!(clientId && clientSecret && !clientId.includes('YOUR_') && !clientSecret.includes('YOUR_')); if (!stream && !twitchConfigured) { stream = await getOrStartActiveStream(); } const streamId = stream ? stream.id : null; const cleanWords = words.map(w => w.toLowerCase().trim().replace(/[.,\/#!$%\^&\*;:{}=\-_`~()?]/g,"")).filter(w => w.length > 0); if (cleanWords.length === 0) return; if (dbMode === 'sqlite') { const insert = sqliteDb.prepare('INSERT INTO voice_words (stream_id, word, timestamp) VALUES (?, ?, ?)'); const transaction = sqliteDb.transaction((recs) => { for (const word of recs) { insert.run(streamId, word, time); } }); transaction(cleanWords); } else { const records = cleanWords.map(word => ({ stream_id: streamId, word, timestamp: time })); const { error } = await supabase.from('voice_words').insert(records); if (error) { console.error('[Supabase] Error logging voice words:', error); } } } /** * Log moderator actions */ export async function logModAction(action) { const time = action.timestamp || new Date().toISOString(); let stream = null; if (action.timestamp) { stream = await getStreamAtTimestamp(action.timestamp); } if (!stream) { stream = await getActiveStream(); } 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, 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) { 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(); const startTimeRange = new Date(timeMs - 15000).toISOString(); const endTimeRange = new Date(timeMs + 15000).toISOString(); const { data: existing, error } = await supabase .from('mod_actions') .select('id, moderator') .eq('action_type', action.actionType) .eq('target_user', targetUserLower) .gte('timestamp', startTimeRange) .lte('timestamp', endTimeRange) .limit(1); if (!error && existing && existing.length > 0) { 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; } } } } // Try to auto-calculate reaction time if target user and message exist let reactionTime = null; if (action.targetUser) { const targetUserLower = action.targetUser.toLowerCase(); let originalMsg = null; try { if (action.actionType === 'delete' && action.messageText) { if (dbMode === 'sqlite') { originalMsg = sqliteDb.prepare('SELECT timestamp FROM messages WHERE username = ? AND message = ? ORDER BY timestamp DESC LIMIT 1').get(targetUserLower, action.messageText); } else { const { data } = await supabase .from('messages') .select('timestamp') .eq('username', targetUserLower) .eq('message', action.messageText) .order('timestamp', { ascending: false }) .limit(1); if (data && data.length > 0) originalMsg = data[0]; } } else { if (dbMode === 'sqlite') { originalMsg = sqliteDb.prepare('SELECT timestamp FROM messages WHERE username = ? ORDER BY timestamp DESC LIMIT 1').get(targetUserLower); } else { const { data } = await supabase .from('messages') .select('timestamp') .eq('username', targetUserLower) .order('timestamp', { ascending: false }) .limit(1); if (data && data.length > 0) originalMsg = data[0]; } } if (originalMsg) { const origTime = new Date(originalMsg.timestamp); const actionTime = new Date(time); const diffSec = (actionTime - origTime) / 1000; if (diffSec >= 0 && diffSec < 1800) { // must be positive and within 30 minutes reactionTime = diffSec; } } } catch (e) { console.error('[Database] Failed to calculate reaction time:', e.message); } } if (dbMode === 'sqlite') { 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 { 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); } } } } /** * Get list of all streams */ export async function getStreamsList() { if (dbMode === 'sqlite') { return sqliteDb.prepare('SELECT * FROM streams ORDER BY start_time DESC').all(); } else { const { data, error } = await supabase .from('streams') .select('*') .order('start_time', { ascending: false }); if (error) { console.error('[Supabase] Error getting streams:', error); return []; } return data || []; } } /** * 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 */ export async function getTopChatters(streamId = null, limit = 50) { if (dbMode === 'sqlite') { let stmt; if (streamId) { stmt = sqliteDb.prepare(` SELECT v.username, 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 LIMIT ? `); return stmt.all(streamId, limit); } else { stmt = sqliteDb.prepare(` 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 ORDER BY message_count DESC LIMIT ? `); return stmt.all(limit); } } else { // Supabase Mode queries the views we created let query; if (streamId) { query = supabase .from('stream_chatter_stats') .select('*') .eq('stream_id', streamId) .order('message_count', { ascending: false }) .limit(limit); } else { query = supabase .from('global_chatter_stats') .select('*') .order('message_count', { ascending: false }) .limit(limit); } 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 || []; } } /** * Get overall summary statistics (true total metrics) */ export async function getStatsSummary(streamId = null) { if (dbMode === 'sqlite') { let msgCount, uniqueChatters, voiceCount, topChatter; if (streamId) { msgCount = sqliteDb.prepare('SELECT COUNT(*) as count FROM messages WHERE stream_id = ?').get(streamId).count; uniqueChatters = sqliteDb.prepare('SELECT COUNT(DISTINCT username) as count FROM messages WHERE stream_id = ?').get(streamId).count; voiceCount = sqliteDb.prepare('SELECT COUNT(*) as count FROM voice_words WHERE stream_id = ?').get(streamId).count; const row = sqliteDb.prepare('SELECT display_name FROM stream_viewers WHERE stream_id = ? AND has_chatted = 1 ORDER BY (SELECT COUNT(*) FROM messages m WHERE m.stream_id = ? AND m.username = stream_viewers.username) DESC LIMIT 1').get(streamId, streamId); topChatter = row ? row.display_name : '—'; } else { msgCount = sqliteDb.prepare('SELECT COUNT(*) as count FROM messages').get().count; uniqueChatters = sqliteDb.prepare('SELECT COUNT(DISTINCT username) as count FROM messages').get().count; voiceCount = sqliteDb.prepare('SELECT COUNT(*) as count FROM voice_words').get().count; const row = sqliteDb.prepare('SELECT username FROM messages GROUP BY username ORDER BY COUNT(*) DESC LIMIT 1').get(); topChatter = row ? row.username : '—'; } return { messages: msgCount, uniqueChatters: uniqueChatters, mostActiveChatter: topChatter, voiceWordsCount: voiceCount }; } else { // Supabase Mode let msgCountQuery = supabase.from('messages').select('*', { count: 'exact', head: true }); let voiceCountQuery = supabase.from('voice_words').select('*', { count: 'exact', head: true }); if (streamId) { msgCountQuery = msgCountQuery.eq('stream_id', streamId); voiceCountQuery = voiceCountQuery.eq('stream_id', streamId); } const { count: msgCount } = await msgCountQuery; const { count: voiceCount } = await voiceCountQuery; // Unique chatters count let uniqueChatters = 0; if (streamId) { 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 .from('global_chatter_stats') .select('*', { count: 'exact', head: true }); uniqueChatters = count || 0; } // Top active chatter let topChatter = '—'; if (streamId) { const { data } = await supabase .from('stream_chatter_stats') .select('display_name') .eq('stream_id', streamId) .order('message_count', { ascending: false }) .limit(1); if (data && data.length > 0) topChatter = data[0].display_name; } else { const { data } = await supabase .from('global_chatter_stats') .select('username') .order('message_count', { ascending: false }) .limit(1); if (data && data.length > 0) topChatter = data[0].username; } return { messages: msgCount || 0, uniqueChatters: uniqueChatters, mostActiveChatter: topChatter, voiceWordsCount: voiceCount || 0 }; } } /** * Get top spoken words */ export async function getSpokenWords(streamId = null, limit = 100) { if (dbMode === 'sqlite') { let stmt; if (streamId) { stmt = sqliteDb.prepare(` SELECT word, count(*) as word_count FROM voice_words WHERE stream_id = ? GROUP BY word ORDER BY word_count DESC LIMIT ? `); return stmt.all(streamId, limit); } else { stmt = sqliteDb.prepare(` SELECT word, count(*) as word_count FROM voice_words GROUP BY word ORDER BY word_count DESC LIMIT ? `); return stmt.all(limit); } } else { let query; if (streamId) { query = supabase .from('stream_voice_word_stats') .select('word, word_count') .eq('stream_id', streamId) .order('word_count', { ascending: false }) .limit(limit); } else { query = supabase .from('global_voice_word_stats') .select('word, word_count') .order('word_count', { ascending: false }) .limit(limit); } const { data, error } = await query; if (error) { console.error('[Supabase] Error getting spoken words:', error); return []; } return data || []; } } /** * Get all messages written in chat (for client-side word count in JS) */ export async function getChatMessages(streamId = null) { const bots = ['streamelements', 'nightbot']; if (dbMode === 'sqlite') { if (streamId) { return sqliteDb.prepare('SELECT message, username, display_name, timestamp FROM messages WHERE stream_id = ? AND username NOT IN (?, ?)').all(streamId, ...bots); } else { return sqliteDb.prepare('SELECT message, username, display_name, timestamp FROM messages WHERE username NOT IN (?, ?) ORDER BY timestamp DESC LIMIT 10000').all(...bots); } } else { if (streamId) { let allData = []; let from = 0; const limit = 1000; while(true) { const { data, error } = await supabase .from('messages') .select('message, username, display_name, timestamp') .eq('stream_id', streamId) .not('username', 'in', `(${bots.map(b => `"${b}"`).join(',')})`) .range(from, from + limit - 1); if (error) { console.error('[Supabase] Error getting chat messages:', error); break; } if (data && data.length > 0) { allData = allData.concat(data); if (data.length < limit) break; // Fetched the last incomplete page from += limit; } else { break; // Empty page } } return allData; } else { // Global query (All streams) - limit to 10,000 most recent messages to prevent huge egress const { data, error } = await supabase .from('messages') .select('message, username, display_name, timestamp') .not('username', 'in', `(${bots.map(b => `"${b}"`).join(',')})`) .order('timestamp', { ascending: false }) .limit(10000); if (error) { console.error('[Supabase] Error getting global chat messages:', error); return []; } return data || []; } } } /** * Get log of moderator actions */ export async function getModActionsList(streamId = null, limit = 100) { if (dbMode === 'sqlite') { if (streamId) { return sqliteDb.prepare('SELECT * FROM mod_actions WHERE stream_id = ? ORDER BY timestamp DESC LIMIT ?').all(streamId, limit); } else { return sqliteDb.prepare('SELECT * FROM mod_actions ORDER BY timestamp DESC LIMIT ?').all(limit); } } else { let query = supabase .from('mod_actions') .select('*') .order('timestamp', { ascending: false }) .limit(limit); if (streamId) { query = query.eq('stream_id', streamId); } const { data, error } = await query; if (error) { console.error('[Supabase] Error getting mod actions:', error); return []; } return data || []; } } /** * Settings Get Helper */ export async function getSetting(key) { if (dbMode === 'sqlite') { const row = sqliteDb.prepare('SELECT value FROM settings WHERE key = ?').get(key); return row ? row.value : null; } else { const { data, error } = await supabase .from('settings') .select('value') .eq('key', key) .single(); if (error) { if (error.code !== 'PGRST116') { console.error(`[Supabase] Error getting setting ${key}:`, error); } return null; } return data ? data.value : null; } } /** * Settings Set Helper */ export async function setSetting(key, value) { if (dbMode === 'sqlite') { sqliteDb.prepare('INSERT INTO settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = ?').run(key, String(value), String(value)); } else { const { error } = await supabase .from('settings') .upsert([{ key, value: String(value) }]); if (error) { console.error(`[Supabase] Error saving setting ${key}:`, error); } } } /** * Get all message timestamps for a stream (for activity chart) */ export async function getStreamMessageTimestamps(streamId) { if (dbMode === 'sqlite') { return sqliteDb.prepare('SELECT timestamp FROM messages WHERE stream_id = ? ORDER BY timestamp ASC').all(streamId); } else { let allData = []; let from = 0; const limit = 1000; while(true) { const { data, error } = await supabase .from('messages') .select('timestamp') .eq('stream_id', streamId) .order('timestamp', { ascending: true }) .range(from, from + limit - 1); if (error) { console.error('[Supabase] Error fetching message timestamps:', error); break; } if (data && data.length > 0) { allData = allData.concat(data); if (data.length < limit) break; from += limit; } else { break; } } return allData; } } /** * Get raw moderator actions for summary aggregation */ export async function getModActionsSummaryRaw(streamId = null) { if (dbMode === 'sqlite') { if (streamId) { return sqliteDb.prepare('SELECT moderator, action_type FROM mod_actions WHERE stream_id = ?').all(streamId); } else { return sqliteDb.prepare('SELECT moderator, action_type FROM mod_actions').all(); } } else { let allData = []; let from = 0; const limit = 1000; while(true) { let query = supabase.from('mod_actions').select('moderator, action_type').range(from, from + limit - 1); if (streamId) { query = query.eq('stream_id', streamId); } const { data, error } = await query; if (error) { console.error('[Supabase] Error getting mod actions summary:', error); break; } if (data && data.length > 0) { allData = allData.concat(data); if (data.length < limit) break; from += limit; } else { break; } } return allData; } } /** * Get aggregated moderator stats and radar scores for profiles */ export async function getModeratorProfilesData(streamId = null) { let actions = []; if (dbMode === 'sqlite') { if (streamId) { actions = sqliteDb.prepare('SELECT * FROM mod_actions WHERE stream_id = ?').all(streamId); } else { actions = sqliteDb.prepare('SELECT * FROM mod_actions').all(); } } else { let query = supabase.from('mod_actions').select('*'); if (streamId) { query = query.eq('stream_id', streamId); } const { data, error } = await query; if (error) { console.error('[Supabase] Error fetching mod actions for profiles:', error); return []; } actions = data || []; } const modData = {}; for (const act of actions) { const mod = act.moderator; if (!modData[mod]) { modData[mod] = { moderator: mod, total_actions: 0, reaction_times: [], bans_count: 0, timeouts_count: 0, deletions_count: 0, unbans_count: 0 }; } const stats = modData[mod]; stats.total_actions += 1; if (act.action_type === 'ban') stats.bans_count += 1; else if (act.action_type === 'timeout') stats.timeouts_count += 1; else if (act.action_type === 'delete') stats.deletions_count += 1; else if (act.action_type === 'unban') stats.unbans_count += 1; if (act.reaction_time !== null && act.reaction_time !== undefined) { stats.reaction_times.push(act.reaction_time); } } const profiles = Object.values(modData).map(stats => { const totalReactionTime = stats.reaction_times.reduce((sum, val) => sum + val, 0); const avgReactionTime = stats.reaction_times.length > 0 ? totalReactionTime / stats.reaction_times.length : null; // Normalizing scores (0 to 100) let scoreSpeed = 50; if (avgReactionTime !== null) { if (avgReactionTime <= 2) scoreSpeed = 100; else if (avgReactionTime >= 30) scoreSpeed = 10; else scoreSpeed = Math.round(100 - ((avgReactionTime - 2) * (90 / 28))); } const scoreActivity = Math.min(100, Math.round((stats.total_actions / 50) * 100)); const strictCount = stats.bans_count + stats.timeouts_count; const scoreHarshness = stats.total_actions > 0 ? Math.round((strictCount / stats.total_actions) * 100) : 0; const scoreWatchfulness = stats.total_actions > 0 ? Math.round((stats.deletions_count / stats.total_actions) * 100) : 0; return { moderator: stats.moderator, total_actions: stats.total_actions, reaction_time_avg: avgReactionTime ? parseFloat(avgReactionTime.toFixed(2)) : null, bans_count: stats.bans_count, timeouts_count: stats.timeouts_count, deletions_count: stats.deletions_count, unbans_count: stats.unbans_count, scores: { speed: scoreSpeed, activity: scoreActivity, harshness: scoreHarshness, watchfulness: scoreWatchfulness } }; }); return profiles.sort((a, b) => b.total_actions - a.total_actions); } // ============================================================ // COVERAGE WINDOWS — track which time ranges were captured // ============================================================ /** * Start a new coverage window (worker session started) */ export async function startCoverageWindow(streamId, source = 'live', coveredFrom = null, coveredTo = null) { const fromTime = coveredFrom || new Date().toISOString(); if (dbMode === 'sqlite') { const result = sqliteDb.prepare( 'INSERT INTO stream_coverage (stream_id, covered_from, covered_to, source) VALUES (?, ?, ?, ?)' ).run(streamId, fromTime, coveredTo, source); return result.lastInsertRowid; } else { const { data, error } = await supabase .from('stream_coverage') .insert({ stream_id: streamId, covered_from: fromTime, covered_to: coveredTo, source }) .select('id') .single(); if (error) console.error('[Supabase] startCoverageWindow error:', error.message); return data?.id || null; } } /** * Update coverage heartbeat (worker still running) */ export async function updateCoverageHeartbeat(coverageId) { const now = new Date().toISOString(); if (dbMode === 'sqlite') { sqliteDb.prepare('UPDATE stream_coverage SET covered_to = ? WHERE id = ?').run(now, coverageId); } else { await supabase.from('stream_coverage').update({ covered_to: now }).eq('id', coverageId); } } /** * End a coverage window (worker session stopped) */ export async function endCoverageWindow(coverageId, coveredTo = null) { const toTime = coveredTo || new Date().toISOString(); if (dbMode === 'sqlite') { sqliteDb.prepare('UPDATE stream_coverage SET covered_to = ? WHERE id = ?').run(toTime, coverageId); } else { await supabase.from('stream_coverage').update({ covered_to: toTime }).eq('id', coverageId); } } /** * Fetch VOD duration in seconds via Twitch GQL */ async function getVodDurationSeconds(vodId) { try { const res = await fetch('https://gql.twitch.tv/gql', { method: 'POST', headers: { 'Client-Id': 'kimne78kx3ncx6brgo4mv6wki5h1ko', 'Content-Type': 'application/json' }, body: JSON.stringify({ query: `query { video(id: "${vodId}") { lengthSeconds } }` }) }); const data = await res.json(); return data?.data?.video?.lengthSeconds || null; } catch (e) { console.error('[GQL] getVodDurationSeconds error:', e.message); return null; } } /** * Compute uncovered gaps for a stream given its coverage windows. * Returns array of { from_offset, to_offset } in seconds from stream start. */ export async function getCoverageGaps(streamId, vodId, streamStartTime) { const startEpoch = new Date(streamStartTime).getTime() / 1000; // 1. Get VOD duration const durationSeconds = vodId ? await getVodDurationSeconds(vodId) : null; const vodEnd = durationSeconds != null ? startEpoch + durationSeconds : null; // 2. Get all coverage windows for this stream let windows = []; if (dbMode === 'sqlite') { windows = sqliteDb.prepare( 'SELECT covered_from, covered_to FROM stream_coverage WHERE stream_id = ? AND covered_to IS NOT NULL ORDER BY covered_from ASC' ).all(streamId); } else { const { data } = await supabase .from('stream_coverage') .select('covered_from, covered_to') .eq('stream_id', streamId) .not('covered_to', 'is', null) .order('covered_from', { ascending: true }); windows = data || []; } if (windows.length === 0) { // No coverage at all — backfill entire VOD if (vodEnd) return [{ from_offset: 0, to_offset: durationSeconds }]; return [{ from_offset: 0, to_offset: null }]; } // 3. Convert to [start_offset, end_offset] pairs (seconds from stream start) const intervals = windows.map(w => ({ s: Math.max(0, Math.round(new Date(w.covered_from).getTime() / 1000 - startEpoch)), e: Math.round(new Date(w.covered_to).getTime() / 1000 - startEpoch) })).filter(w => w.e > w.s); // 4. Sort and merge overlapping intervals intervals.sort((a, b) => a.s - b.s); const merged = []; for (const iv of intervals) { if (merged.length === 0 || iv.s > merged[merged.length - 1].e + 30) { merged.push({ ...iv }); } else { merged[merged.length - 1].e = Math.max(merged[merged.length - 1].e, iv.e); } } // 5. Find gaps const gaps = []; const THRESHOLD = 60; // ignore gaps < 60 seconds // Gap before first coverage window if (merged[0].s > THRESHOLD) { gaps.push({ from_offset: 0, to_offset: merged[0].s }); } // Gaps between windows for (let i = 0; i < merged.length - 1; i++) { const gapSize = merged[i + 1].s - merged[i].e; if (gapSize > THRESHOLD) { gaps.push({ from_offset: merged[i].e, to_offset: merged[i + 1].s }); } } // Gap after last coverage window (to end of VOD) const lastEnd = merged[merged.length - 1].e; if (vodEnd) { const remainingSeconds = durationSeconds - lastEnd; if (remainingSeconds > THRESHOLD) { gaps.push({ from_offset: lastEnd, to_offset: durationSeconds }); } } else { // Unknown VOD duration — backfill from last known point to end gaps.push({ from_offset: lastEnd, to_offset: null }); } return gaps; } /** * Get list of VOD streams waiting for backfill */ export async function getPendingBackfillStreams() { if (dbMode === 'sqlite') { const streams = sqliteDb.prepare("SELECT * FROM streams WHERE backfill_status = 'pending' AND twitch_vod_id IS NOT NULL ORDER BY start_time ASC").all(); for (const stream of streams) { stream.gaps = await getCoverageGaps(stream.id, stream.twitch_vod_id, stream.start_time); } return streams; } else { const { data: streams, error } = await supabase .from('streams') .select('*') .eq('backfill_status', 'pending') .not('twitch_vod_id', 'is', null) .order('start_time', { ascending: true }); if (error) { console.error('[Supabase] Error getting pending backfill streams:', error); return []; } if (streams) { for (const stream of streams) { stream.gaps = await getCoverageGaps(stream.id, stream.twitch_vod_id, stream.start_time); } } return streams || []; } } /** * Get list of pending streams that are missing their twitch_vod_id */ export async function getStreamsMissingVod() { if (dbMode === 'sqlite') { return sqliteDb.prepare("SELECT * FROM streams WHERE backfill_status = 'pending' AND twitch_vod_id IS NULL ORDER BY start_time ASC").all(); } else { const { data: streams, error } = await supabase .from('streams') .select('*') .eq('backfill_status', 'pending') .is('twitch_vod_id', null) .order('start_time', { ascending: true }); if (error) { console.error('[Supabase] Error getting streams missing VOD:', error); return []; } return streams || []; } } /** * Update the twitch_vod_id of a stream */ export async function updateStreamVodId(streamId, twitchVodId) { if (dbMode === 'sqlite') { sqliteDb.prepare("UPDATE streams SET twitch_vod_id = ? WHERE id = ?").run(twitchVodId, streamId); console.log(`[Database] SQLite Stream ID ${streamId} updated twitch_vod_id to ${twitchVodId}`); } else { const { error } = await supabase .from('streams') .update({ twitch_vod_id: twitchVodId }) .eq('id', streamId); if (error) { console.error('[Supabase] Error updating twitch_vod_id:', error); } else { console.log(`[Database] Supabase Stream ID ${streamId} updated twitch_vod_id to ${twitchVodId}`); } } } /** * Mark a stream as backfilled and completed */ export async function markStreamBackfilled(streamId = null, twitchStreamId = null) { if (dbMode === 'sqlite') { if (streamId) { sqliteDb.prepare("UPDATE streams SET backfill_status = 'completed' WHERE id = ?").run(streamId); } else if (twitchStreamId) { sqliteDb.prepare("UPDATE streams SET backfill_status = 'completed' WHERE twitch_stream_id = ?").run(twitchStreamId); } } else { const query = supabase.from('streams').update({ backfill_status: 'completed' }); let res; if (streamId) { res = await query.eq('id', streamId); } else if (twitchStreamId) { res = await query.eq('twitch_stream_id', twitchStreamId); } if (res && res.error) { console.error('[Supabase] Error marking stream as backfilled:', res.error); } } } /** * Reset backfill status and optionally delete existing data for rebuild */ export async function resetStreamBackfill(streamId, mode = 'gap_fill') { if (dbMode === 'sqlite') { if (mode === 'full_rebuild') { sqliteDb.prepare("DELETE FROM messages WHERE stream_id = ?").run(streamId); sqliteDb.prepare("DELETE FROM voice_words WHERE stream_id = ?").run(streamId); console.log(`[Database] SQLite Stream ID ${streamId} data cleared for full rebuild.`); } sqliteDb.prepare("UPDATE streams SET backfill_status = 'pending' WHERE id = ?").run(streamId); console.log(`[Database] SQLite Stream ID ${streamId} reset to pending.`); } else { if (mode === 'full_rebuild') { const { error: msgErr } = await supabase.from('messages').delete().eq('stream_id', streamId); const { error: voiceErr } = await supabase.from('voice_words').delete().eq('stream_id', streamId); if (msgErr || voiceErr) { console.error('[Supabase] Error clearing data for rebuild:', msgErr, voiceErr); } else { console.log(`[Database] Supabase Stream ID ${streamId} data cleared for full rebuild.`); } } const { error: streamErr } = await supabase.from('streams').update({ backfill_status: 'pending' }).eq('id', streamId); if (streamErr) { console.error('[Supabase] Error resetting stream to pending:', streamErr); } else { console.log(`[Database] Supabase Stream ID ${streamId} reset to pending.`); } } } /** * Delete stream and all its messages, words, and mod actions */ export async function deleteStream(streamId) { if (dbMode === 'sqlite') { sqliteDb.prepare("DELETE FROM messages WHERE stream_id = ?").run(streamId); sqliteDb.prepare("DELETE FROM voice_words WHERE stream_id = ?").run(streamId); sqliteDb.prepare("DELETE FROM mod_actions WHERE stream_id = ?").run(streamId); sqliteDb.prepare("DELETE FROM streams WHERE id = ?").run(streamId); console.log(`[Database] SQLite Stream ID ${streamId} fully deleted.`); } else { const { error: msgErr } = await supabase.from('messages').delete().eq('stream_id', streamId); const { error: voiceErr } = await supabase.from('voice_words').delete().eq('stream_id', streamId); const { error: modErr } = await supabase.from('mod_actions').delete().eq('stream_id', streamId); const { error: streamErr } = await supabase.from('streams').delete().eq('id', streamId); if (msgErr || voiceErr || modErr || streamErr) { console.error('[Supabase] Error deleting stream:', msgErr, voiceErr, modErr, streamErr); } else { console.log(`[Database] Supabase Stream ID ${streamId} fully deleted.`); } } } /** * Update metadata (title/category) of a stream */ export async function updateStreamMetadata(streamId, title, category) { if (dbMode === 'sqlite') { sqliteDb.prepare("UPDATE streams SET title = ?, category = ? WHERE id = ?").run(title, category, streamId); console.log(`[Database] SQLite Stream ID ${streamId} updated: Title="${title}", Category="${category}"`); } else { const { error } = await supabase .from('streams') .update({ title, category }) .eq('id', streamId); if (error) { console.error('[Supabase] Error updating stream metadata:', error); } else { console.log(`[Database] Supabase Stream ID ${streamId} updated: Title="${title}", Category="${category}"`); } } } /** * Get system statistics (size, count of records) */ export async function getSystemStats() { let stats = { dbMode, totalStreams: 0, totalMessages: 0, totalVoiceWords: 0, dbSizeMb: 0 }; if (dbMode === 'sqlite') { stats.totalStreams = sqliteDb.prepare("SELECT count(*) as count FROM streams").get().count; stats.totalMessages = sqliteDb.prepare("SELECT count(*) as count FROM messages").get().count; stats.totalVoiceWords = sqliteDb.prepare("SELECT count(*) as count FROM voice_words").get().count; try { const dbPath = path.resolve('database.db'); if (fs.existsSync(dbPath)) { const fileStats = fs.statSync(dbPath); stats.dbSizeMb = parseFloat((fileStats.size / (1024 * 1024)).toFixed(2)); } } catch (err) { console.error('[Database] Error reading SQLite file size:', err.message); } } else { // Supabase count queries const { count: streamsCount, error: err1 } = await supabase.from('streams').select('*', { count: 'exact', head: true }); const { count: messagesCount, error: err2 } = await supabase.from('messages').select('*', { count: 'exact', head: true }); const { count: voiceWordsCount, error: err3 } = await supabase.from('voice_words').select('*', { count: 'exact', head: true }); if (err1 || err2 || err3) { console.error('[Supabase] Error fetching system stats:', err1, err2, err3); } stats.totalStreams = streamsCount || 0; stats.totalMessages = messagesCount || 0; stats.totalVoiceWords = voiceWordsCount || 0; } return stats; } /** * Clean up streams with 0 messages AND 0 voice words */ export async function cleanupGhostStreams() { if (dbMode === 'sqlite') { const info = sqliteDb.prepare(` DELETE FROM streams WHERE id NOT IN (SELECT DISTINCT stream_id FROM messages WHERE stream_id IS NOT NULL) AND id NOT IN (SELECT DISTINCT stream_id FROM voice_words WHERE stream_id IS NOT NULL) `).run(); console.log(`[Database] SQLite Ghost streams cleaned up. Deleted ${info.changes} streams.`); return info.changes; } else { // Supabase mode - fetch all streams and check manually (safer for postgres count limitations) const { data: streams, error: fetchErr } = await supabase.from('streams').select('id'); if (fetchErr || !streams) { console.error('[Supabase] Error fetching streams for cleanup:', fetchErr); return 0; } let deletedCount = 0; for (const stream of streams) { const { count: msgCount } = await supabase.from('messages').select('*', { count: 'exact', head: true }).eq('stream_id', stream.id); const { count: voiceCount } = await supabase.from('voice_words').select('*', { count: 'exact', head: true }).eq('stream_id', stream.id); if ((msgCount || 0) === 0 && (voiceCount || 0) === 0) { const { error: delErr } = await supabase.from('streams').delete().eq('id', stream.id); if (delErr) { console.error(`[Supabase] Error deleting empty stream ID ${stream.id}:`, delErr); } else { deletedCount++; } } } console.log(`[Database] Supabase Ghost streams cleaned up. Deleted ${deletedCount} streams.`); return deletedCount; } } /** * Log viewer join event (Lurker) */ export async function logViewerJoin(username) { const stream = await getActiveStream(); if (!stream) return; const streamId = stream.id; const time = new Date().toISOString(); if (dbMode === 'sqlite') { try { sqliteDb.prepare(` INSERT INTO stream_viewers (stream_id, username, display_name, has_chatted, is_mod, is_sub, first_seen) VALUES (?, ?, ?, 0, 0, 0, ?) ON CONFLICT(stream_id, username) DO NOTHING `).run(streamId, username.toLowerCase(), username, time); } catch (err) { console.error('[Database] SQLite Error logging viewer join:', err.message); } } else { const { error } = await supabase .from('stream_viewers') .insert([{ stream_id: streamId, username: username.toLowerCase(), display_name: username, has_chatted: false, is_mod: false, is_sub: false, first_seen: time }], { onConflict: 'stream_id, username' }); if (error && error.code !== '23505') { console.error('[Supabase] Error logging viewer join:', error); } } } /** * Get backfill status for a specific stream by its twitch_stream_id */ export async function getStreamBackfillStatus(twitchStreamId) { let stream = null; if (dbMode === 'sqlite') { stream = sqliteDb.prepare('SELECT * FROM streams WHERE twitch_stream_id = ?').get(twitchStreamId); } else { const { data, error } = await supabase .from('streams') .select('*') .eq('twitch_stream_id', twitchStreamId) .limit(1); if (data && data.length > 0) { stream = data[0]; } } if (!stream) return null; let latestMessageTimestamp = null; let latestVoiceTimestamp = null; let messageCount = 0; let voiceCount = 0; if (dbMode === 'sqlite') { const msgStats = sqliteDb.prepare('SELECT COUNT(*) as count, MAX(timestamp) as latest FROM messages WHERE stream_id = ?').get(stream.id); messageCount = msgStats.count; latestMessageTimestamp = msgStats.latest; const voiceStats = sqliteDb.prepare('SELECT COUNT(*) as count, MAX(timestamp) as latest FROM voice_words WHERE stream_id = ?').get(stream.id); voiceCount = voiceStats.count; latestVoiceTimestamp = voiceStats.latest; } else { // Supabase const { count: msgCount } = await supabase .from('messages') .select('*', { count: 'exact', head: true }) .eq('stream_id', stream.id); messageCount = msgCount || 0; if (messageCount > 0) { const { data: latestMsg } = await supabase .from('messages') .select('timestamp') .eq('stream_id', stream.id) .order('timestamp', { ascending: false }) .limit(1); if (latestMsg && latestMsg.length > 0) { latestMessageTimestamp = latestMsg[0].timestamp; } } const { count: vCount } = await supabase .from('voice_words') .select('*', { count: 'exact', head: true }) .eq('stream_id', stream.id); voiceCount = vCount || 0; if (voiceCount > 0) { const { data: latestVoice } = await supabase .from('voice_words') .select('timestamp') .eq('stream_id', stream.id) .order('timestamp', { ascending: false }) .limit(1); if (latestVoice && latestVoice.length > 0) { latestVoiceTimestamp = latestVoice[0].timestamp; } } } return { streamId: stream.id, twitchStreamId: stream.twitch_stream_id, messageCount, latestMessageTimestamp, voiceCount, latestVoiceTimestamp }; }