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('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 ); `); 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 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) if (action.targetUser) { const targetUserLower = action.targetUser.toLowerCase(); if (dbMode === 'sqlite') { const existing = sqliteDb.prepare(` SELECT id 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; } } 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') .eq('action_type', action.actionType) .eq('target_user', targetUserLower) .gte('timestamp', startTimeRange) .lte('timestamp', endTimeRange) .limit(1); if (!error && existing && existing.length > 0) { 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') { 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); } } } /** * 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 || []; } } /** * 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, 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.stream_id = v.stream_id AND m.username = v.username) as message_count FROM stream_viewers v WHERE v.stream_id = ? GROUP BY v.username ORDER BY message_count DESC LIMIT ? `); return stmt.all(streamId, limit); } 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 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); } const { data, error } = await query; if (error) { console.error('[Supabase] Error getting top chatters:', error); return []; } return data || []; } } /** * 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 (?, ?)').all(...bots); } } else { let query = supabase .from('messages') .select('message, username, display_name, timestamp') .not('username', 'in', `(${bots.map(b => `"${b}"`).join(',')})`); if (streamId) { query = query.eq('stream_id', streamId); } const { data, error } = await query; if (error) { console.error('[Supabase] Error getting 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 { const { data, error } = await supabase .from('messages') .select('timestamp') .eq('stream_id', streamId) .order('timestamp', { ascending: true }); if (error) { console.error('[Supabase] Error fetching message timestamps:', error); return []; } return data || []; } } /** * 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 query = supabase .from('mod_actions') .select('moderator, action_type'); if (streamId) { query = query.eq('stream_id', streamId); } const { data, error } = await query; if (error) { console.error('[Supabase] Error fetching mod actions summary:', error); return []; } return data || []; } } /** * 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); } /** * 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) { const msgMaxRow = sqliteDb.prepare("SELECT max(timestamp) as max_time FROM messages WHERE stream_id = ?").get(stream.id); const msgMinRow = sqliteDb.prepare("SELECT min(timestamp) as min_time FROM messages WHERE stream_id = ?").get(stream.id); const voiceMaxRow = sqliteDb.prepare("SELECT max(timestamp) as max_time FROM voice_words WHERE stream_id = ?").get(stream.id); const voiceMinRow = sqliteDb.prepare("SELECT min(timestamp) as min_time FROM voice_words WHERE stream_id = ?").get(stream.id); stream.max_message_time = msgMaxRow ? msgMaxRow.max_time : null; stream.min_message_time = msgMinRow ? msgMinRow.min_time : null; stream.max_voice_time = voiceMaxRow ? voiceMaxRow.max_time : null; stream.min_voice_time = voiceMinRow ? voiceMinRow.min_time : null; } 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) { // Query max message timestamp in Supabase const { data: msgMaxData } = await supabase .from('messages') .select('timestamp') .eq('stream_id', stream.id) .order('timestamp', { ascending: false }) .limit(1); // Query min message timestamp in Supabase const { data: msgMinData } = await supabase .from('messages') .select('timestamp') .eq('stream_id', stream.id) .order('timestamp', { ascending: true }) .limit(1); // Query max voice timestamp in Supabase const { data: voiceMaxData } = await supabase .from('voice_words') .select('timestamp') .eq('stream_id', stream.id) .order('timestamp', { ascending: false }) .limit(1); // Query min voice timestamp in Supabase const { data: voiceMinData } = await supabase .from('voice_words') .select('timestamp') .eq('stream_id', stream.id) .order('timestamp', { ascending: true }) .limit(1); stream.max_message_time = msgMaxData && msgMaxData.length > 0 ? msgMaxData[0].timestamp : null; stream.min_message_time = msgMinData && msgMinData.length > 0 ? msgMinData[0].timestamp : null; stream.max_voice_time = voiceMaxData && voiceMaxData.length > 0 ? voiceMaxData[0].timestamp : null; stream.min_voice_time = voiceMinData && voiceMinData.length > 0 ? voiceMinData[0].timestamp : null; } } return streams || []; } } /** * 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); } } }