import WebSocket from 'ws'; import axios from 'axios'; import { getSetting, setSetting, logModAction } from './db.js'; let ws = null; let broadcasterUserId = null; let keepAliveTimer = null; let lastKeepAlive = null; let reconnecting = false; /** * Validate current access token and refresh it if expired/invalid */ export async function getOrRefreshAccessToken() { let token = await getSetting('twitch_streamer_access_token'); const refreshToken = await getSetting('twitch_streamer_refresh_token'); if (!token) { return null; } // Validate the current token via Twitch API try { await axios.get('https://id.twitch.tv/oauth2/validate', { headers: { 'Authorization': `Bearer ${token}` } }); // Token is valid! return token; } catch (err) { const status = err.response?.status; console.log(`[EventSub] Access token validation failed (Status: ${status}). Attempting to refresh...`); if (!refreshToken) { console.error('[EventSub] Refresh token is missing in database. Cannot refresh access token.'); return null; } try { const response = await axios.post('https://id.twitch.tv/oauth2/token', null, { params: { client_id: process.env.TWITCH_CLIENT_ID, client_secret: process.env.TWITCH_CLIENT_SECRET, grant_type: 'refresh_token', refresh_token: refreshToken } }); const { access_token, refresh_token: newRefreshToken } = response.data; await setSetting('twitch_streamer_access_token', access_token); if (newRefreshToken) { await setSetting('twitch_streamer_refresh_token', newRefreshToken); } console.log('[EventSub] Access token refreshed successfully.'); return access_token; } catch (refreshErr) { console.error('[EventSub] Token refresh request failed:', refreshErr.response?.data || refreshErr.message); return null; } } } /** * Initialize Twitch EventSub WebSocket connection */ export async function initializeEventSub(streamerId = null) { // Try to load broadcaster ID from database if not passed if (!streamerId) { broadcasterUserId = await getSetting('twitch_broadcaster_id'); } else { broadcasterUserId = streamerId; } if (!broadcasterUserId) { console.log('[EventSub] Broadcaster ID is not configured yet. Waiting for Streamer login.'); return; } const token = await getOrRefreshAccessToken(); if (!token) { console.log('[EventSub] Streamer access token is missing or invalid. Waiting for Streamer login.'); return; } if (ws) { console.log('[EventSub] Connection already exists, closing it first...'); try { ws.close(); } catch (e) {} } connect('wss://eventsub.wss.twitch.tv/ws'); } /** * Connect to EventSub WebSocket */ function connect(url) { console.log(`[EventSub] Connecting to ${url}...`); ws = new WebSocket(url); reconnecting = false; ws.on('open', () => { console.log('[EventSub] WebSocket connection opened.'); lastKeepAlive = Date.now(); startKeepAliveCheck(); }); ws.on('message', async (data) => { try { const message = JSON.parse(data.toString()); await handleMessage(message); } catch (err) { console.error('[EventSub] Error parsing message:', err); } }); ws.on('close', (code, reason) => { console.log(`[EventSub] WebSocket connection closed: Code ${code}, Reason: ${reason}`); cleanup(); if (!reconnecting) { // Reconnect after 5 seconds if not a controlled migration console.log('[EventSub] Reconnecting in 5 seconds...'); setTimeout(() => initializeEventSub(), 5000); } }); ws.on('error', (err) => { console.error('[EventSub] WebSocket error:', err); }); } /** * Handle incoming EventSub WebSocket messages */ async function handleMessage(message) { const { metadata, payload } = message; const messageType = metadata.message_type; lastKeepAlive = Date.now(); switch (messageType) { case 'session_welcome': { const sessionId = payload.session.id; console.log(`[EventSub] Welcome received. Session ID: ${sessionId}`); await subscribeToEvents(sessionId); break; } case 'session_keepalive': // Just updating lastKeepAlive is enough break; case 'session_reconnect': { const reconnectUrl = payload.session.reconnect_url; console.log(`[EventSub] Reconnect requested. Migrating connection to: ${reconnectUrl}`); reconnecting = true; connect(reconnectUrl); break; } case 'notification': { await handleNotification(payload); break; } case 'revocation': console.warn('[EventSub] Subscription revoked:', payload.subscription.type); break; default: console.log('[EventSub] Unknown message type:', messageType); } } /** * Handle incoming event notifications */ async function handleNotification(payload) { const { subscription, event } = payload; const type = subscription.type; console.log(`[EventSub] Event received: ${type}`); switch (type) { case 'channel.ban': { // Event triggers on ban or timeout const isTimeout = event.ends_at !== null; await logModAction({ actionType: isTimeout ? 'timeout' : 'ban', moderator: event.moderator_user_name, targetUser: event.user_name, duration: isTimeout ? Math.round((new Date(event.ends_at) - new Date(event.banned_at)) / 1000) : null, reason: event.reason, timestamp: event.banned_at }); break; } case 'channel.unban': { await logModAction({ actionType: 'unban', moderator: event.moderator_user_name, targetUser: event.user_name, timestamp: new Date().toISOString() }); break; } case 'channel.chat.message_delete': { await logModAction({ actionType: 'delete', moderator: event.moderator_user_name, targetUser: event.target_user_name, messageText: event.message_body, timestamp: new Date().toISOString() }); break; } default: console.log('[EventSub] Unhandled event type:', type); } } /** * Register EventSub subscriptions via Twitch Helix API */ async function subscribeToEvents(sessionId) { const clientId = process.env.TWITCH_CLIENT_ID; const token = await getOrRefreshAccessToken(); if (!clientId || !token) { console.error('[EventSub] Cannot subscribe: Client ID or token is missing!'); return; } const subscriptions = [ { type: 'channel.ban', version: '1' }, { type: 'channel.unban', version: '1' }, { type: 'channel.chat.message_delete', version: '1' } ]; const headers = { 'Client-ID': clientId, 'Authorization': `Bearer ${token}`, 'Content-Type': 'application/json' }; for (const sub of subscriptions) { try { const body = { type: sub.type, version: sub.version, condition: { broadcaster_user_id: broadcasterUserId }, transport: { method: 'websocket', session_id: sessionId } }; const response = await axios.post('https://api.twitch.tv/helix/eventsub/subscriptions', body, { headers }); console.log(`[EventSub] Subscribed to ${sub.type}. Status: ${response.status}`); } catch (err) { console.error(`[EventSub] Subscription failed for ${sub.type}:`, err.response?.data || err.message); } } } /** * Keep WebSocket connection healthy */ function startKeepAliveCheck() { cleanupKeepAlive(); keepAliveTimer = setInterval(() => { // Twitch sends keepalives every 10 seconds. If no message for 20 seconds, reconnect. if (Date.now() - lastKeepAlive > 20000) { console.warn('[EventSub] Keepalive timeout. Reconnecting WebSocket...'); cleanupKeepAlive(); if (ws) { try { ws.terminate(); } catch (e) {} } initializeEventSub(); } }, 10000); } function cleanupKeepAlive() { if (keepAliveTimer) { clearInterval(keepAliveTimer); keepAliveTimer = null; } } function cleanup() { cleanupKeepAlive(); ws = null; }