server/eventsub.js

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;
}