import express from 'express'; import cors from 'cors'; import cookieSession from 'cookie-session'; import dotenv from 'dotenv'; import axios from 'axios'; import path from 'path'; import fs from 'fs'; import { logChatMessage, logChatMessagesBatch, logRolesBatch, logVoiceWords, logModAction, logViewerJoin, getStreamsList, getTopChatters, getSpokenWords, getChatMessages, getModActionsList, getSetting, setSetting, endActiveStream, getStreamMessageTimestamps, getModActionsSummaryRaw, syncActiveStream, getOrStartActiveStream, getStatsSummary, getModeratorProfilesData, getActiveStream, getPendingBackfillStreams, getStreamsMissingVod, updateStreamVodId, markStreamBackfilled, checkIfStreamExists, resetStreamBackfill, deleteStream, updateStreamMetadata, getSystemStats, cleanupGhostStreams, startCoverageWindow, updateCoverageHeartbeat, endCoverageWindow, getStreamBackfillStatus } from './db.js'; // import { initializeEventSub } from './eventsub.js'; import { cache, cacheMiddleware } from './cache.js'; dotenv.config(); const app = express(); app.set('trust proxy', true); // Safe trust proxy behind Cloudflare and Hugging Face proxies const PORT = process.env.PORT || 3000; // Configure CORS to allow frontend connections const allowedOrigins = [ 'http://localhost:5173', // Vite React local dev 'http://localhost:3000', // Express local serving 'https://mimic42.qzz.io', // Custom production domain 'https://winx-prinx-statistics.pages.dev', // Cloudflare Pages default domain ]; if (process.env.ALLOWED_ORIGINS) { const envOrigins = process.env.ALLOWED_ORIGINS.split(',').map(o => o.trim()); allowedOrigins.push(...envOrigins); } app.use(cors({ origin: function (origin, callback) { if (!origin || allowedOrigins.indexOf(origin) !== -1) { callback(null, true); } else { callback(new Error('Not allowed by CORS')); } }, credentials: true })); app.use(express.json()); // Session setup const isProduction = process.env.NODE_ENV === 'production'; app.use(cookieSession({ name: 'twitch-analytics-session', keys: [process.env.SESSION_SECRET || 'fallback_secret_key_98765'], maxAge: 30 * 24 * 60 * 60 * 1000, // 30 days secure: isProduction, // Requires HTTPS to allow sameSite: 'none' sameSite: isProduction ? 'none' : 'lax' })); // Initialize EventSub connection if we already have the streamer token in DB // initializeEventSub().catch(err => { // console.error('[EventSub] Startup initialization failed:', err); // }); // ========================================================================= // MIDDLEWARES // ========================================================================= // API Key authentication for the local worker const authenticateWorker = (req, res, next) => { const apiKey = req.headers['x-api-key']; if (!apiKey || apiKey !== process.env.API_KEY) { return res.status(401).json({ error: 'Unauthorized. Invalid API Key.' }); } next(); }; // Authentication check for dashboard users (streamer / mods) const requireModeratorRole = (req, res, next) => { 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 (!twitchConfigured) { return next(); // Bypass authentication check if Twitch login is not configured } if (!req.session || !req.session.user) { return res.status(401).json({ error: 'Unauthorized. Please log in.' }); } const role = req.session.user.role; if (role !== 'streamer' && role !== 'moderator' && role !== 'admin') { return res.status(403).json({ error: 'Forbidden. Access restricted to streamer, admin and moderators.' }); } next(); }; // Authentication check for dashboard administrators (streamer / admins) const requireAdminRole = (req, res, next) => { 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 (!twitchConfigured) { return next(); // Bypass authentication check if Twitch login is not configured } if (!req.session || !req.session.user) { return res.status(401).json({ error: 'Unauthorized. Please log in.' }); } const role = req.session.user.role; if (role !== 'streamer' && role !== 'admin') { return res.status(403).json({ error: 'Forbidden. Access restricted to streamer and admins.' }); } next(); }; // ========================================================================= // PUBLIC UTILITY ENDPOINTS // ========================================================================= // Server health check for frontend connection status app.get('/api/health', (req, res) => { res.status(200).json({ status: 'ok', uptime: process.uptime() }); }); // Avatar proxy to bypass CORS issues when fetching from decapi.me app.get('/api/avatar/:channel', async (req, res) => { try { const { channel } = req.params; // decapi.me returns the URL of the avatar as plain text const urlResp = await axios.get(`https://decapi.me/twitch/avatar/${channel}`, { timeout: 5000 }); const avatarUrl = urlResp.data.trim(); if (!avatarUrl.startsWith('http')) { return res.status(404).send('Avatar not found'); } // Fetch the actual image and stream it back const imgResp = await axios.get(avatarUrl, { responseType: 'arraybuffer', timeout: 8000 }); res.set('Content-Type', imgResp.headers['content-type'] || 'image/png'); res.set('Cache-Control', 'public, max-age=3600'); res.send(imgResp.data); } catch (err) { console.error('[Avatar] Error fetching avatar:', err.message); res.status(500).send('Error fetching avatar'); } }); // ========================================================================= // WORKER INGESTION ENDPOINTS (Local Worker -> Server) // ========================================================================= // Ingest chat messages app.post('/api/log/messages', authenticateWorker, async (req, res) => { const { messages } = req.body; if (!messages || !Array.isArray(messages)) { return res.status(400).json({ error: 'Invalid input. Expected array of messages.' }); } try { await logChatMessagesBatch(messages); } catch (err) { console.error('Error logging batch messages:', err); return res.status(500).json({ error: 'Failed to log messages' }); } res.json({ success: true, count: messages.length }); }); // Ingest chat roles ONLY (used by role_backfiller) app.post('/api/log/roles', authenticateWorker, async (req, res) => { const { roles } = req.body; if (!roles || !Array.isArray(roles)) { return res.status(400).json({ error: 'Invalid input. Expected array of roles.' }); } try { await logRolesBatch(roles); } catch (err) { console.error('Error logging batch roles:', err); return res.status(500).json({ error: 'Failed to log roles' }); } res.json({ success: true, count: roles.length }); }); // Endpoint for the Python worker to log viewer joins app.post('/api/log/viewers/join', authenticateWorker, async (req, res) => { const username = req.body.username; if (!username) return res.status(400).json({ error: 'No username' }); await logViewerJoin(username); // We can broadcast this if we want live updates of lurkers, // but since there's 1000s, let's just log it. res.json({ success: true }); }); // Ingest voice words app.post('/api/log/voice', authenticateWorker, async (req, res) => { const { words, timestamp } = req.body; if (!words || !Array.isArray(words)) { return res.status(400).json({ error: 'Invalid input. Expected array of words.' }); } try { await logVoiceWords(words, timestamp); res.json({ success: true, count: words.length }); } catch (err) { console.error('[API Error] /api/log/voice:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // Manual/fallback mod action logging app.post('/api/log/mod-action', authenticateWorker, async (req, res) => { const action = req.body; try { await logModAction(action); res.json({ success: true }); } catch (err) { console.error('[API Error] /api/log/mod-action:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // Explicit start stream command app.post('/api/log/stream-start', authenticateWorker, async (req, res) => { try { const stream = await getOrStartActiveStream(); res.json({ success: true, streamId: stream ? stream.id : null }); } catch (err) { console.error('[API Error] /api/log/stream-start:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // Explicit end stream command app.post('/api/log/stream-end', authenticateWorker, async (req, res) => { try { const { twitchVodId } = req.body || {}; await endActiveStream(null, null, twitchVodId); res.json({ success: true }); } catch (err) { console.error('[API Error] /api/log/stream-end:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // GET pending backfills list app.get('/api/streams/pending-backfill', authenticateWorker, async (req, res) => { try { const streams = await getPendingBackfillStreams(); res.json({ success: true, streams }); } catch (err) { console.error('[API Error] /api/streams/pending-backfill:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // GET backfill status for a specific stream (chat messages count & latest timestamp, voice words count & latest timestamp) app.get('/api/streams/backfill-status', authenticateWorker, async (req, res) => { const { twitchStreamId } = req.query; if (!twitchStreamId) { return res.status(400).json({ error: 'Missing twitchStreamId query parameter' }); } try { const status = await getStreamBackfillStatus(twitchStreamId); if (!status) { return res.status(404).json({ error: 'Stream not found' }); } res.json(status); } catch (err) { console.error('[API Error] /api/streams/backfill-status:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // GET list of streams missing twitch_vod_id app.get('/api/streams/missing-vod', authenticateWorker, async (req, res) => { try { const streams = await getStreamsMissingVod(); res.json({ success: true, streams }); } catch (err) { console.error('[API Error] /api/streams/missing-vod:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // POST to update twitch_vod_id for a stream app.post('/api/streams/:id/resolve-vod', authenticateWorker, async (req, res) => { const { id } = req.params; const { twitchVodId } = req.body; try { await updateStreamVodId(id, twitchVodId); res.json({ success: true }); } catch (err) { console.error('[API Error] /api/streams/resolve-vod:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // POST start coverage window app.post('/api/log/coverage/start', authenticateWorker, async (req, res) => { const { streamId, source, coveredFrom, coveredTo } = req.body; if (!streamId) return res.status(400).json({ error: 'Missing streamId' }); try { const coverageId = await startCoverageWindow(streamId, source || 'live', coveredFrom, coveredTo); res.json({ success: true, coverageId }); } catch (err) { console.error('[API Error] /api/log/coverage/start:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // PATCH heartbeat coverage window app.patch('/api/log/coverage/:id/heartbeat', authenticateWorker, async (req, res) => { const coverageId = parseInt(req.params.id); try { await updateCoverageHeartbeat(coverageId); res.json({ success: true }); } catch (err) { console.error('[API Error] /api/log/coverage/heartbeat:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // POST end coverage window app.post('/api/log/coverage/end', authenticateWorker, async (req, res) => { const { coverageId, coveredTo } = req.body; if (!coverageId) return res.status(400).json({ error: 'Missing coverageId' }); try { await endCoverageWindow(coverageId, coveredTo); res.json({ success: true }); } catch (err) { console.error('[API Error] /api/log/coverage/end:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // POST mark stream as backfilled app.post('/api/streams/mark-backfilled', authenticateWorker, async (req, res) => { const { streamId, twitchStreamId } = req.body; try { await markStreamBackfilled(streamId, twitchStreamId); res.json({ success: true }); } catch (err) { console.error('[API Error] /api/streams/mark-backfilled:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // POST reset stream backfill status (admin/moderator controls) app.post('/api/streams/reset-backfill', requireModeratorRole, async (req, res) => { const { streamId, mode } = req.body; if (!streamId) { return res.status(400).json({ error: 'Missing streamId parameter.' }); } try { await resetStreamBackfill(parseInt(streamId), mode || 'gap_fill'); cache.flush(); res.json({ success: true }); } catch (err) { console.error('[API Error] /api/streams/reset-backfill:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // DELETE stream (admin controls) app.delete('/api/streams/:id', requireAdminRole, async (req, res) => { const streamId = parseInt(req.params.id); if (isNaN(streamId)) { return res.status(400).json({ error: 'Invalid stream ID.' }); } try { await deleteStream(streamId); cache.flush(); res.json({ success: true }); } catch (err) { console.error('[API Error] /api/streams/:id (DELETE):', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // PUT stream metadata (admin controls) app.put('/api/streams/:id', requireAdminRole, async (req, res) => { const streamId = parseInt(req.params.id); const { title, category } = req.body; if (isNaN(streamId)) { return res.status(400).json({ error: 'Invalid stream ID.' }); } if (!title) { return res.status(400).json({ error: 'Title is required.' }); } try { await updateStreamMetadata(streamId, title, category || ''); cache.flush(); res.json({ success: true }); } catch (err) { console.error('[API Error] /api/streams/:id (PUT):', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // POST clean up empty streams (admin controls) app.post('/api/admin/cleanup', requireAdminRole, async (req, res) => { try { const deletedCount = await cleanupGhostStreams(); cache.flush(); res.json({ success: true, count: deletedCount }); } catch (err) { console.error('[API Error] /api/admin/cleanup:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // GET system statistics (admin controls) app.get('/api/admin/stats', requireAdminRole, async (req, res) => { try { const stats = await getSystemStats(); const cacheStats = cache.getStats(); res.json({ success: true, stats, cacheStats }); } catch (err) { console.error('[API Error] /api/admin/stats:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // ========================================================================= // TWITCH OAUTH ENDPOINTS // ========================================================================= // Redirect to Twitch Authorization Screen app.get('/api/auth/twitch', (req, res) => { const clientId = process.env.TWITCH_CLIENT_ID; const redirectUri = process.env.TWITCH_REDIRECT_URI; // Scopes required: // - user:read:email (basic user identity) // - channel:moderate (to listen to event sub) // - moderation:read (to read moderators lists) const scopes = 'user:read:email channel:moderate moderation:read'; const twitchAuthUrl = `https://id.twitch.tv/oauth2/authorize` + `?client_id=${clientId}` + `&redirect_uri=${encodeURIComponent(redirectUri)}` + `&response_type=code` + `&scope=${encodeURIComponent(scopes)}`; res.redirect(twitchAuthUrl); }); // Twitch Auth Callback app.get('/api/auth/twitch/callback', async (req, res) => { const { code, error } = req.query; let frontendRedirect = '/'; if (process.env.NODE_ENV === 'production') { if (process.env.TWITCH_REDIRECT_URI) { try { const redirectUrl = new URL(process.env.TWITCH_REDIRECT_URI); frontendRedirect = `${redirectUrl.protocol}//${redirectUrl.host}/`; } catch (e) { frontendRedirect = 'https://mimic42.qzz.io/'; } } else { frontendRedirect = 'https://mimic42.qzz.io/'; } } else { frontendRedirect = 'http://localhost:5173/'; } if (error) { console.error('Twitch OAuth error callback:', error); return res.redirect(`${frontendRedirect}auth-failed`); } try { // Exchange Auth Code for Access Token const tokenResponse = 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, code, grant_type: 'authorization_code', redirect_uri: process.env.TWITCH_REDIRECT_URI } }); const { access_token, refresh_token } = tokenResponse.data; // Fetch User Info from Twitch API const userResponse = await axios.get('https://api.twitch.tv/helix/users', { headers: { 'Client-ID': process.env.TWITCH_CLIENT_ID, 'Authorization': `Bearer ${access_token}` } }); const twitchUser = userResponse.data.data[0]; const username = twitchUser.login.toLowerCase(); const displayName = twitchUser.display_name; const userId = twitchUser.id; const targetChannel = process.env.TWITCH_CHANNEL.toLowerCase(); let role = 'viewer'; const adminUsernames = (process.env.ADMIN_USERNAMES || '') .toLowerCase() .split(',') .map(name => name.trim()) .filter(Boolean); if (adminUsernames.includes(username)) { role = 'admin'; } else if (username === targetChannel) { // User is the Streamer role = 'streamer'; // Save Streamer's tokens and ID to the DB settings await setSetting('twitch_broadcaster_id', userId); await setSetting('twitch_streamer_access_token', access_token); await setSetting('twitch_streamer_refresh_token', refresh_token); console.log(`[Auth] Streamer ${displayName} logged in. (EventSub subscription disabled)`); // Start/Restart EventSub using this session // await initializeEventSub(userId); } else { // Check if user is a moderator by querying Twitch API via Streamer's token const streamerToken = await getSetting('twitch_streamer_access_token'); const broadcasterId = await getSetting('twitch_broadcaster_id'); if (streamerToken && broadcasterId) { try { const modCheckResponse = await axios.get('https://api.twitch.tv/helix/moderators', { headers: { 'Client-ID': process.env.TWITCH_CLIENT_ID, 'Authorization': `Bearer ${streamerToken}` }, params: { broadcaster_id: broadcasterId, user_id: userId } }); const isMod = modCheckResponse.data.data.length > 0; if (isMod) { role = 'moderator'; } } catch (modErr) { console.error('[Auth] Error checking moderator status:', modErr.response?.data || modErr.message); } } else { console.warn('[Auth] Streamer tokens not available in DB. Cannot verify moderator status.'); } } // Save in session cookie req.session.user = { id: userId, username, displayName, role }; console.log(`[Auth] User ${displayName} logged in. Role: ${role}`); res.redirect(frontendRedirect); } catch (err) { console.error('Error during Twitch OAuth callback:', err.response?.data || err.message); res.status(500).send('Authentication failed.'); } }); // Auth Status Endpoint app.get('/api/auth/status', (req, res) => { 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 (req.session && req.session.user) { res.json({ loggedIn: true, user: req.session.user, twitchConfigured }); } else { res.json({ loggedIn: false, twitchConfigured }); } }); // Logout Endpoint app.get('/api/auth/logout', (req, res) => { req.session = null; res.json({ success: true }); }); // ========================================================================= // PUBLIC STATISTICS API ENDPOINTS // ========================================================================= // Get list of streams app.get('/api/streams', cacheMiddleware(120), async (req, res) => { const streams = await getStreamsList(); res.json(streams); }); // Get top active chatters app.get('/api/stats/chatters', cacheMiddleware(300), async (req, res) => { const streamId = req.query.stream_id ? parseInt(req.query.stream_id) : null; const limit = req.query.limit ? parseInt(req.query.limit) : 50; const chatters = await getTopChatters(streamId, limit); res.json(chatters); }); // Get true total summary statistics app.get('/api/stats/summary', cacheMiddleware(120), async (req, res) => { const streamId = req.query.stream_id ? parseInt(req.query.stream_id) : null; try { const summary = await getStatsSummary(streamId); res.json(summary); } catch (err) { console.error('[API Error] /api/stats/summary:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // Stop words arrays (Russian and English) to filter out common articles, prepositions const STOP_WORDS = new Set([ // Russian 'и', 'в', 'во', 'не', 'что', 'он', 'на', 'я', 'с', 'со', 'как', 'а', 'то', 'все', 'она', 'так', 'его', 'но', 'да', 'ты', 'к', 'у', 'же', 'вы', 'за', 'бы', 'по', 'только', 'ее', 'мне', 'было', 'вот', 'от', 'меня', 'еще', 'нет', 'о', 'из', 'ему', 'же', 'им', 'имхо', 'это', 'этот', 'эта', 'эти', 'этого', 'себя', 'свой', 'свои', 'или', 'был', 'была', 'были', 'быть', 'когда', 'кто', 'который', 'для', 'чтобы', 'если', 'через', 'после', 'даже', 'же', 'ни', 'там', 'тут', 'где', // Extra Russian non-independent words and pronouns (length >= 3) 'тебя', 'тебе', 'себе', 'меня', 'этого', 'этому', 'этом', 'этой', 'этими', 'него', 'нее', 'ними', 'будет', 'будут', 'было', 'были', 'была', 'быть', 'своих', 'своим', 'своей', 'тоже', 'очень', 'вообще', 'какой', 'какая', 'какое', 'какие', 'чтото', 'что-то', 'хочу', 'хочешь', 'хочет', 'хотят', 'буду', 'будешь', 'будем', 'будете', 'знаю', 'знаешь', 'знает', 'знают', 'эту', 'эти', 'этим', 'этих', 'здесь', 'туда', 'оттуда', 'сюда', 'потому', 'почему', 'зачем', 'какойто', 'какой-то', 'под', 'про', 'без', 'вас', 'вам', 'нас', 'нам', 'ими', 'они', 'ней', 'нем', 'ваши', 'наши', 'своем', 'этого', // English 'the', 'a', 'an', 'and', 'or', 'but', 'to', 'of', 'in', 'on', 'at', 'by', 'for', 'with', 'about', 'against', 'between', 'into', 'through', 'during', 'before', 'after', 'above', 'below', 'from', 'up', 'down', 'is', 'are', 'was', 'were', 'be', 'been', 'being', 'have', 'has', 'had', 'having', 'do', 'does', 'did', 'doing', 'i', 'you', 'he', 'she', 'it', 'we', 'they', 'me', 'him', 'her', 'us', 'them', 'my', 'your', 'his', 'their', 'this', 'that', 'these', 'those', 'then', 'there', 'here', 'whats', 'what', 'which', 'who', 'whom', 'whose', 'why', 'how', 'can', 'will', 'just' ]); // Word mappings/synonyms to group similar pronunciations or typos const WORD_MAPPINGS = { 'блядь': 'блять', 'петерка': 'пятерка', 'питерка': 'пятерка', 'петёрка': 'пятерка', 'пётерка': 'пятерка', 'кримбал': 'кримбр', 'кримбл': 'кримбр', 'кримбор': 'кримбр', 'кримбол': 'кримбр', 'кримбалу': 'кримбр', 'кримбру': 'кримбр', 'кримб': 'кримбр' }; // Get word frequencies (Voice vs Chat messages from streamer) app.get('/api/stats/words', cacheMiddleware(600), async (req, res) => { const streamId = req.query.stream_id ? parseInt(req.query.stream_id) : null; const limit = req.query.limit ? parseInt(req.query.limit) : 50; const type = req.query.type || 'voice'; // 'voice' or 'chat' if (type === 'voice') { // Spoken words from voice database table. Get extra words first to allow filtering stop words. const rawWords = await getSpokenWords(streamId, limit * 10); const wordCounts = {}; for (const w of rawWords) { if (!w || !w.word || typeof w.word !== 'string') continue; let clean = w.word.toLowerCase().trim(); if (WORD_MAPPINGS[clean]) { clean = WORD_MAPPINGS[clean]; } const isAllowedLength = clean.length >= 3 || clean === 'ну'; if (isAllowedLength && !STOP_WORDS.has(clean)) { wordCounts[clean] = (wordCounts[clean] || 0) + w.word_count; } } const sortedWords = Object.entries(wordCounts) .map(([word, count]) => ({ word, word_count: count })) .sort((a, b) => b.word_count - a.word_count) .slice(0, limit); return res.json(sortedWords); } else if (type === 'chat') { // Written words in chat by viewers const messages = await getChatMessages(streamId); // Build a set of all user names from chatters in this stream (nickname filter) const chatters = new Set(); const targetChannel = (process.env.TWITCH_CHANNEL || 'winx_prinx').toLowerCase(); chatters.add(targetChannel); chatters.add('winx'); chatters.add('prinx'); for (const msg of messages) { if (msg.username) { chatters.add(msg.username.toLowerCase().trim()); } if (msg.display_name) { chatters.add(msg.display_name.toLowerCase().trim()); } if (msg.displayName) { chatters.add(msg.displayName.toLowerCase().trim()); } } // Process words in JS const wordCounts = {}; for (const msg of messages) { if (!msg || !msg.message || typeof msg.message !== 'string') continue; const tokens = msg.message .toLowerCase() .replace(/[.,\/#!$%\^&\*;:{}=\-_`~()?]/g, ' ') // replace punctuation with space .split(/\s+/); for (let word of tokens) { word = word.trim(); // Remove mention @ prefix if present if (word.startsWith('@')) { word = word.slice(1); } if (!word) continue; if (WORD_MAPPINGS[word]) { word = WORD_MAPPINGS[word]; } const isAllowedLength = word.length >= 3 || word === 'ну'; const isNickname = chatters.has(word); if (isAllowedLength && !isNickname && !STOP_WORDS.has(word) && !word.startsWith('http') && !word.startsWith('www')) { wordCounts[word] = (wordCounts[word] || 0) + 1; } } } const sortedWords = Object.entries(wordCounts) .map(([word, count]) => ({ word, word_count: count })) .sort((a, b) => b.word_count - a.word_count) .slice(0, limit); return res.json(sortedWords); } else { res.status(400).json({ error: "Invalid type. Must be 'voice' or 'chat'." }); } }); // Get chat activity (messages count grouped by 5-minute intervals) app.get('/api/stats/activity', cacheMiddleware(300), async (req, res) => { const streamId = req.query.stream_id ? parseInt(req.query.stream_id) : null; if (!streamId) { return res.status(400).json({ error: 'Missing stream_id query parameter.' }); } // Get timestamps of all messages in this stream const messages = await getStreamMessageTimestamps(streamId); if (!messages || messages.length === 0) { return res.json([]); } // Group into 5-minute buckets const intervalMs = 5 * 60 * 1000; const buckets = {}; messages.forEach(msg => { const timeMs = new Date(msg.timestamp).getTime(); const roundedTime = new Date(Math.floor(timeMs / intervalMs) * intervalMs); const bucketKey = roundedTime.toISOString(); buckets[bucketKey] = (buckets[bucketKey] || 0) + 1; }); const chartData = Object.entries(buckets).map(([timestamp, count]) => ({ timestamp, message_count: count })); res.json(chartData); }); // ========================================================================= // PROTECTED MODERATOR API ENDPOINTS (Require logged in mod/streamer) // ========================================================================= // Get log of moderator actions app.get('/api/stats/moderators', requireModeratorRole, cacheMiddleware(300), async (req, res) => { const streamId = req.query.stream_id ? parseInt(req.query.stream_id) : null; const limit = req.query.limit ? parseInt(req.query.limit) : 100; const actions = await getModActionsList(streamId, limit); res.json(actions); }); // Get summary counts of actions per moderator app.get('/api/stats/moderators/summary', requireModeratorRole, cacheMiddleware(300), async (req, res) => { const streamId = req.query.stream_id ? parseInt(req.query.stream_id) : null; const data = await getModActionsSummaryRaw(streamId); // Aggregate in JS: { moderator_name: { ban: X, timeout: Y, delete: Z, total: T } } const summary = {}; data.forEach(act => { const mod = act.moderator; const type = act.action_type; if (!summary[mod]) { summary[mod] = { ban: 0, timeout: 0, delete: 0, unban: 0, total: 0 }; } if (type in summary[mod]) { summary[mod][type]++; } summary[mod].total++; }); const responseData = Object.entries(summary).map(([moderator, stats]) => ({ moderator, ...stats })).sort((a, b) => b.total - a.total); res.json(responseData); }); // GET aggregated profiles with scores for moderators app.get('/api/stats/moderators/profiles', cacheMiddleware(300), async (req, res) => { const streamId = req.query.stream_id ? parseInt(req.query.stream_id) : null; const profiles = await getModeratorProfilesData(streamId); res.json(profiles); }); // POST endpoint to sync past VODs from Twitch app.post('/api/streams/sync-vods', async (req, res) => { const clientId = process.env.TWITCH_CLIENT_ID; const clientSecret = process.env.TWITCH_CLIENT_SECRET; const channelName = process.env.TWITCH_CHANNEL || 'winx_prinx'; if (!clientId || !clientSecret || clientId.includes('YOUR_') || clientSecret.includes('YOUR_')) { return res.status(400).json({ error: 'Twitch Developer credentials are not configured in бэкенд .env' }); } try { // 1. Get App Access Token const tokenRes = await axios.post(`https://id.twitch.tv/oauth2/token?client_id=${clientId}&client_secret=${clientSecret}&grant_type=client_credentials`); const accessToken = tokenRes.data.access_token; // 2. Get user ID const userRes = await axios.get(`https://api.twitch.tv/helix/users?login=${channelName}`, { headers: { 'Client-ID': clientId, 'Authorization': `Bearer ${accessToken}` } }); if (!userRes.data.data || userRes.data.data.length === 0) { return res.status(404).json({ error: 'User not found on Twitch.' }); } const userId = userRes.data.data[0].id; // 3. Get VOD list const vodRes = await axios.get(`https://api.twitch.tv/helix/videos?user_id=${userId}&type=archive&first=20`, { headers: { 'Client-ID': clientId, 'Authorization': `Bearer ${accessToken}` } }); const vods = vodRes.data.data || []; const syncedStreams = []; for (const vod of vods) { const durationStr = vod.duration; let durationMs = 0; const hoursMatch = durationStr.match(/(\d+)h/); const minsMatch = durationStr.match(/(\d+)m/); const secsMatch = durationStr.match(/(\d+)s/); if (hoursMatch) durationMs += parseInt(hoursMatch[1]) * 60 * 60 * 1000; if (minsMatch) durationMs += parseInt(minsMatch[1]) * 60 * 1000; if (secsMatch) durationMs += parseInt(secsMatch[1]) * 1000; const startTime = new Date(vod.created_at); const endTime = new Date(startTime.getTime() + durationMs).toISOString(); // Creates or restores the stream const stream = await syncActiveStream(vod.stream_id || `vod-${vod.id}`, vod.title, '', vod.created_at); if (stream) { await endActiveStream(stream.id, endTime, vod.id); syncedStreams.push({ id: stream.id, twitch_stream_id: vod.stream_id || `vod-${vod.id}`, title: vod.title, start_time: vod.created_at, end_time: endTime, twitch_vod_url: vod.url }); } } res.json({ success: true, count: syncedStreams.length, streams: syncedStreams }); } catch (err) { console.error('[Twitch API] Error syncing VODs:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // POST endpoint for local worker to sync VOD details directly app.post('/api/streams/sync-vod-direct', authenticateWorker, async (req, res) => { const { twitchStreamId, title, category, startTime, endTime } = req.body; try { const stream = await syncActiveStream(twitchStreamId, title, category || 'Архив', startTime); if (stream) { const numericVodId = twitchStreamId ? twitchStreamId.replace('vod-', '') : null; await endActiveStream(stream.id, endTime, numericVodId); return res.json({ success: true, streamId: stream.id }); } res.status(500).json({ error: 'Failed to sync stream session' }); } catch (err) { console.error('[API Error] /api/streams/sync-vod-direct:', err); res.status(500).json({ error: 'Internal Server Error' }); } }); // Outgoing Webhook receiver from Better Stack to forward alerts to Telegram bot app.post('/api/betterstack-webhook', async (req, res) => { console.log('[Better Stack Webhook] Received webhook payload:', req.body); const botToken = process.env.TELEGRAM_BOT_TOKEN; const chatId = process.env.TELEGRAM_CHAT_ID; if (!botToken || !chatId) { console.warn('[Better Stack Webhook] Missing TELEGRAM_BOT_TOKEN or TELEGRAM_CHAT_ID in environment'); return res.status(200).json({ success: false, error: 'Telegram credentials missing' }); } try { const event = req.body.event || 'incident.updated'; const data = req.body.data || {}; const attributes = data.attributes || {}; const monitorName = attributes.name || 'Unknown Monitor'; const url = attributes.url || ''; const cause = attributes.cause || 'No cause provided'; const status = attributes.status || ''; let statusEmoji = '⚠️'; let statusText = 'UPDATED'; if (status === 'ongoing') { statusEmoji = '🔴'; statusText = 'DOWN (CRITICAL)'; } else if (status === 'resolved') { statusEmoji = '🟢'; statusText = 'RESOLVED (UP)'; } const message = `🚨 *[Better Stack Alert]*\n\n` + `🖥️ *Monitor:* ${monitorName}\n` + `📈 *Status:* ${statusEmoji} ${statusText}\n` + `❓ *Cause:* ${cause}\n` + `🔗 *URL:* ${url || 'N/A'}\n` + `🔔 *Event:* \`${event}\``; // Send to Telegram Bot API await axios.post(`https://api.telegram.org/bot${botToken}/sendMessage`, { chat_id: chatId, text: message, parse_mode: 'Markdown' }); console.log('[Better Stack Webhook] Successfully forwarded alert to Telegram'); return res.json({ success: true }); } catch (err) { console.error('[Better Stack Webhook] Error sending message to Telegram:', err.message); // Return 200 to Better Stack so it doesn't retry infinitely on transport failures, but log the error return res.status(200).json({ success: false, error: err.message }); } }); // Periodic Twitch Stream Status Polling async function pollTwitchStreamStatus() { const clientId = process.env.TWITCH_CLIENT_ID; const clientSecret = process.env.TWITCH_CLIENT_SECRET; const channelName = process.env.TWITCH_CHANNEL || 'winx_prinx'; if (!clientId || !clientSecret || clientId.includes('YOUR_') || clientSecret.includes('YOUR_')) { // Silent skip in local test mode return; } try { // 1. Get App Access Token const tokenRes = await axios.post(`https://id.twitch.tv/oauth2/token?client_id=${clientId}&client_secret=${clientSecret}&grant_type=client_credentials`); const accessToken = tokenRes.data.access_token; // 2. Query Stream Status const streamRes = await axios.get(`https://api.twitch.tv/helix/streams?user_login=${channelName}`, { headers: { 'Client-ID': clientId, 'Authorization': `Bearer ${accessToken}` } }); const streamData = streamRes.data.data && streamRes.data.data[0]; if (streamData) { const twitchStreamId = streamData.id; const title = streamData.title; const category = streamData.game_name; const startedAt = streamData.started_at; await syncActiveStream(twitchStreamId, title, category, startedAt); } else { // Offline: close active stream if we have one with twitch_stream_id const active = await getActiveStream(); let userId = null; let userFetched = false; if (active && active.twitch_stream_id) { let endTime = new Date().toISOString(); try { // Find user ID const userRes = await axios.get(`https://api.twitch.tv/helix/users?login=${channelName}`, { headers: { 'Client-ID': clientId, 'Authorization': `Bearer ${accessToken}` } }); userId = userRes.data.data[0].id; userFetched = true; // Find VOD list to compute duration const vodRes = await axios.get(`https://api.twitch.tv/helix/videos?user_id=${userId}&type=archive`, { headers: { 'Client-ID': clientId, 'Authorization': `Bearer ${accessToken}` } }); let lastVodId = null; const lastVod = vodRes.data.data && vodRes.data.data[0]; if (lastVod && lastVod.stream_id === active.twitch_stream_id) { lastVodId = lastVod.id; const durationStr = lastVod.duration; let durationMs = 0; const hoursMatch = durationStr.match(/(\d+)h/); const minsMatch = durationStr.match(/(\d+)m/); const secsMatch = durationStr.match(/(\d+)s/); if (hoursMatch) durationMs += parseInt(hoursMatch[1]) * 60 * 60 * 1000; if (minsMatch) durationMs += parseInt(minsMatch[1]) * 60 * 1000; if (secsMatch) durationMs += parseInt(secsMatch[1]) * 1000; const start = new Date(active.start_time); endTime = new Date(start.getTime() + durationMs).toISOString(); } await endActiveStream(active.id, endTime, lastVodId); } catch (vodErr) { await endActiveStream(active.id, endTime); } } // Auto-scan recent VODs to find any streams we might have missed (e.g. backend was down) try { if (!userFetched) { const userRes = await axios.get(`https://api.twitch.tv/helix/users?login=${channelName}`, { headers: { 'Client-ID': clientId, 'Authorization': `Bearer ${accessToken}` } }); if (userRes.data.data && userRes.data.data.length > 0) { userId = userRes.data.data[0].id; userFetched = true; } } if (userId) { const vodRes = await axios.get(`https://api.twitch.tv/helix/videos?user_id=${userId}&type=archive&first=10`, { headers: { 'Client-ID': clientId, 'Authorization': `Bearer ${accessToken}` } }); const vods = vodRes.data.data || []; for (const vod of vods) { const streamIdToCheck = vod.stream_id || `vod-${vod.id}`; const exists = await checkIfStreamExists(streamIdToCheck); if (!exists) { console.log(`[Twitch Sync] Auto-discovered missing stream VOD: ${vod.title} (${streamIdToCheck})`); // Calculate duration and end time const durationStr = vod.duration; let durationMs = 0; const hoursMatch = durationStr.match(/(\d+)h/); const minsMatch = durationStr.match(/(\d+)m/); const secsMatch = durationStr.match(/(\d+)s/); if (hoursMatch) durationMs += parseInt(hoursMatch[1]) * 60 * 60 * 1000; if (minsMatch) durationMs += parseInt(minsMatch[1]) * 60 * 1000; if (secsMatch) durationMs += parseInt(secsMatch[1]) * 1000; const startTime = new Date(vod.created_at); const endTime = new Date(startTime.getTime() + durationMs).toISOString(); // Register and immediately end it as 'pending' for local backfiller to pick up const stream = await syncActiveStream(streamIdToCheck, vod.title, '', vod.created_at); if (stream) { await endActiveStream(stream.id, endTime, vod.id); } } } } } catch (scanErr) { console.error('[Twitch API] Error auto-syncing offline VODs:', scanErr.message); } } } catch (err) { console.error('[Twitch API] pollTwitchStreamStatus error:', err.message); } } // Serve static frontend files in production if (process.env.NODE_ENV === 'production') { const __dirname = new URL('.', import.meta.url).pathname; const clientDistPath = path.resolve(__dirname, '../client/dist'); if (fs.existsSync(clientDistPath)) { // Express static client serve app.use(express.static(clientDistPath)); app.get('*', (req, res) => { res.sendFile(path.resolve(clientDistPath, 'index.html')); }); } else { // In decoupled mode (Hugging Face API-only), serve a simple JSON status page on root app.get('/', (req, res) => { res.json({ status: 'online', message: 'winx_prinx API Server is running', environment: 'production' }); }); } } // Start Server app.listen(PORT, () => { console.log(`[Server] Running on http://localhost:${PORT}`); console.log(`[Server] Secure API Key: ${process.env.API_KEY ? 'CONFIGURED' : 'MISSING'}`); console.log(`[Server] Environment: ${process.env.NODE_ENV || 'development'}`); // Start stream status polling task (every 2 minutes) // pollTwitchStreamStatus(); // setInterval(pollTwitchStreamStatus, 2 * 60 * 1000); });