clippy / server.js
txtorg's picture
Upload constituent server files
4e3b382 verified
Raw
History Blame Contribute Delete
56.8 kB
// ═══════════════════════════════════════════════════════════════════════════
// CONSTITUENT SERVER
// Runs inside a HuggingFace Docker Space.
// Handles all HLS/FFmpeg streaming logic + constituent-specific APIs.
// Main web server communicates with this via HTTP only.
// config.json is auto-created on first boot storing the constituent owner id.
// ═══════════════════════════════════════════════════════════════════════════
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const https = require('https');
const ffmpeg = require('fluent-ffmpeg');
const axios = require('axios');
const express = require('express');
const http = require('http');
const { Server: SocketIOServer } = require('socket.io');
const os = require('os');
// ── Config ────────────────────────────────────────────────────────────────────
// CONSTITUENT_OWNER_ID must be set as a HuggingFace Space secret/env var.
// It is the userId from your main database that "owns" this constituent.
const CONSTITUENT_OWNER_ID = process.env.CONSTITUENT_OWNER_ID;
const MAIN_SERVER_SECRET = process.env.MAIN_SERVER_SECRET || 'mysecretkeyforogudupaogeuwuwuhdg'; // shared secret to authenticate main server calls
const PORT = parseInt(process.env.PORT || '7860', 10);
const TMDB_KEY = process.env.TMDB_KEY || null;
const TMDB_BASE = 'https://api.themoviedb.org/3';
const TMDB_IMG = 'https://image.tmdb.org/t/p/w500';
if (!CONSTITUENT_OWNER_ID) {
console.error('CONSTITUENT_OWNER_ID env var is required. Set it as a HuggingFace Space secret.');
process.exit(1);
}
console.log(`πŸ”‘ MAIN_SERVER_SECRET: ${process.env.MAIN_SERVER_SECRET ? 'loaded from env' : 'using built-in default'}`);
// ── Auto-create config.json ───────────────────────────────────────────────────
const CONFIG_PATH = path.join(__dirname, 'config.json');
let constituentConfig = {};
if (fs.existsSync(CONFIG_PATH)) {
try { constituentConfig = JSON.parse(fs.readFileSync(CONFIG_PATH, 'utf8')); }
catch { constituentConfig = {}; }
}
if (!constituentConfig.ownerId) {
constituentConfig.ownerId = CONSTITUENT_OWNER_ID;
constituentConfig.createdAt = new Date().toISOString();
fs.writeFileSync(CONFIG_PATH, JSON.stringify(constituentConfig, null, 2));
console.log(`βœ… config.json created for owner: ${CONSTITUENT_OWNER_ID}`);
}
// ── Dirs & constants ──────────────────────────────────────────────────────────
const TEMP_DIR = path.join(__dirname, 'others', 'temp');
const SONGS_DIR = path.join(__dirname, 'others', 'songs');
const HLS_DIR = path.join(__dirname, 'others', 'hls');
const DATA_DIR = path.join(__dirname, 'others', 'data');
fs.mkdirSync(TEMP_DIR, { recursive: true });
fs.mkdirSync(SONGS_DIR, { recursive: true });
fs.mkdirSync(HLS_DIR, { recursive: true });
fs.mkdirSync(DATA_DIR, { recursive: true });
const SHOWPLAY_MAX_FILE_SIZE = 2 * 1024 * 1024 * 1024; // 2 GB
const SHOWPLAY_MAX_DURATION = 6 * 60 * 60; // 6 hrs
const MAX_FILE_SIZE = 50 * 1024 * 1024; // 50 MB (audio)
const MAX_DURATION = 15 * 60; // 15 min (audio)
const STREAM_CLEANUP_INTERVAL = 30 * 60 * 1000;
const DEFAULT_ARTWORK = 'https://touchio.vercel.app/tf14k0.jpeg';
const HLS_PLAYLIST_WINDOW = 6;
const HLS_MAX_SEGMENTS = 800;
// ── SSL agent ─────────────────────────────────────────────────────────────────
const httpsAgentNoVerify = new https.Agent({ rejectUnauthorized: false });
axios.defaults.httpsAgent = httpsAgentNoVerify;
// ── Express + Socket.IO ───────────────────────────────────────────────────────
const app = express();
const server = http.createServer(app);
const io = new SocketIOServer(server, {
cors: { origin: true, credentials: true, methods: ['GET', 'POST'] },
transports: ['websocket', 'polling']
});
app.use(express.json());
app.use('/hls', express.static(HLS_DIR, {
setHeaders: (res, filePath) => {
if (filePath.endsWith('.m3u8')) {
res.setHeader('Content-Type', 'application/vnd.apple.mpegurl');
res.setHeader('Cache-Control', 'no-cache, no-store');
res.setHeader('Access-Control-Allow-Origin', '*');
}
if (filePath.endsWith('.ts')) {
res.setHeader('Content-Type', 'video/MP2T');
res.setHeader('Cache-Control', 'public, max-age=3600');
res.setHeader('Access-Control-Allow-Origin', '*');
}
}
}));
app.use('/songs', express.static(SONGS_DIR));
// ── In-memory streaming state ─────────────────────────────────────────────────
const streams = {};
const hlsState = {};
const hlsMutex = {};
const hlsGeneration = {};
const activeFFmpeg = {};
// ── Auth middleware for main-server calls ─────────────────────────────────────
function requireMainServer(req, res, next) {
const secret = req.headers['x-constituent-secret'];
if (!secret || secret !== MAIN_SERVER_SECRET) {
return res.status(403).json({ success: false, error: 'Forbidden: invalid or missing secret' });
}
next();
}
// ═══════════════════════════════════════════════════════════════════════════
// HEALTH / STATUS API
// Called by main server to check if this constituent is alive and ready.
// ═══════════════════════════════════════════════════════════════════════════
app.get('/constituent/health', (req, res) => {
const totalMem = os.totalmem();
const freeMem = os.freemem();
const usedMem = totalMem - freeMem;
const cpuLoad = os.loadavg()[0]; // 1-min average
// Disk usage via df (Linux only β€” fine for HF Docker)
let diskTotal = null, diskUsed = null, diskFree = null;
try {
const { execSync } = require('child_process');
const dfOut = execSync("df -k / | tail -1").toString().trim().split(/\s+/);
diskTotal = parseInt(dfOut[1]) * 1024;
diskUsed = parseInt(dfOut[2]) * 1024;
diskFree = parseInt(dfOut[3]) * 1024;
} catch {}
const activeStreamCount = Object.keys(streams).filter(id => streams[id]?.isActive).length;
res.json({
success: true,
status: 'running',
ownerId: constituentConfig.ownerId,
createdAt: constituentConfig.createdAt,
uptime: process.uptime(),
memory: {
totalMB: Math.round(totalMem / 1024 / 1024),
usedMB: Math.round(usedMem / 1024 / 1024),
freeMB: Math.round(freeMem / 1024 / 1024),
usedPct: Math.round((usedMem / totalMem) * 100),
},
cpu: { loadAvg1min: cpuLoad.toFixed(2) },
disk: diskTotal ? {
totalGB: (diskTotal / 1024 ** 3).toFixed(1),
usedGB: (diskUsed / 1024 ** 3).toFixed(1),
freeGB: (diskFree / 1024 ** 3).toFixed(1),
usedPct: Math.round((diskUsed / diskTotal) * 100),
} : null,
streams: {
active: activeStreamCount,
total: Object.keys(streams).length,
},
});
});
// ═══════════════════════════════════════════════════════════════════════════
// SHOWPLAY API β€” search by title name (no raw link needed)
// Called by main server when a user (who owns this constituent) adds a movie or episode.
// Only the constituent's owner can trigger this.
//
// POST /constituent/add-movie β€” body: { streamId, title }
// Searches iktracks for the title, picks the first movie result, downloads it.
//
// POST /constituent/add-episode β€” body: { streamId, title, season, episode }
// Searches iktracks for the series, finds the matching S/E, downloads it.
// ═══════════════════════════════════════════════════════════════════════════
const IKTRACKS_BASE = 'https://iktracks.vercel.app';
function spSeriesName(title) {
return (title || '').replace(/\s*\(?\d{4}\)?\s*$/, '').trim() || title;
}
function extractAllEpisodes(details) {
const allEps = [];
for (const season of (details.seasons || [])) {
for (const ep of (season.episodes || [])) {
if (ep && ep.downloadLink) {
allEps.push({ season: season.season, episode: ep.episode, downloadLink: ep.downloadLink });
}
}
}
return allEps;
}
app.post('/constituent/add-movie', requireMainServer, async (req, res) => {
// Supports two modes:
// 1. { streamId, movieLink, movieTitle, thumbnail?, tmdbInfo? } β€” direct link from server.js
// 2. { streamId, title } β€” search by name (legacy / direct constituent use)
const { streamId, movieLink, movieTitle, title: titleOnly, thumbnail, tmdbInfo } = req.body;
if (!streamId) {
return res.status(400).json({ success: false, error: 'streamId is required' });
}
if (streamId !== constituentConfig.ownerId) {
return res.status(403).json({ success: false, error: 'Only the constituent owner can add movies to this server' });
}
// ── Mode 1: direct link provided ───────────────────────────────────────────
if (movieLink && movieTitle) {
res.json({ success: true, message: 'Movie queued for download and encoding', streamId, title: movieTitle });
setImmediate(async () => {
try {
const result = await showplayEnqueueLink(streamId, movieLink, movieTitle, thumbnail || DEFAULT_ARTWORK, tmdbInfo || null);
console.log(`βœ… Movie added to stream ${streamId}: ${result.title}`);
} catch (err) {
console.error(`❌ Failed to add movie to stream ${streamId}:`, err.message);
}
});
return;
}
// ── Mode 2: search by title ────────────────────────────────────────────────
const title = titleOnly || movieTitle;
if (!title) {
return res.status(400).json({ success: false, error: 'Either (movieLink + movieTitle) or title is required' });
}
// Search for the title
let searchResults;
try {
const r = await axios.get(`${IKTRACKS_BASE}/search?query=${encodeURIComponent(title)}`, { timeout: 15000 });
searchResults = (r.data?.results || []).filter(r => r && r.link);
} catch (err) {
return res.status(500).json({ success: false, error: `Search failed: ${err.message}` });
}
if (!searchResults.length) {
return res.status(404).json({ success: false, error: `No results found for "${title}"` });
}
// Pick the first movie result (prefer type==='movie', fall back to first result)
const movieResult = searchResults.find(r => r.type === 'movie') || searchResults[0];
// Fetch details to get the download link
let details;
try {
const r = await axios.get(`${IKTRACKS_BASE}/details?url=${encodeURIComponent(movieResult.link)}`, { timeout: 15000 });
details = r.data;
if (!details) throw new Error('Empty details response');
} catch (err) {
return res.status(500).json({ success: false, error: `Details fetch failed: ${err.message}` });
}
if (details.type === 'series') {
return res.status(400).json({ success: false, error: 'This title is a series. Use /constituent/add-episode instead.' });
}
const link = details.downloadLinks?.[0]?.downloadLink;
if (!link) {
return res.status(404).json({ success: false, error: 'No download link found for this title' });
}
const pendingTitle = details.title || movieResult.title || title;
const pendingThumb = details.thumbnail || movieResult.thumbnail || DEFAULT_ARTWORK;
res.json({ success: true, message: 'Movie queued for download and encoding', streamId, title: pendingTitle });
setImmediate(async () => {
try {
const result = await showplayEnqueueLink(streamId, link, pendingTitle, pendingThumb, tmdbInfo || null);
console.log(`βœ… Movie added to stream ${streamId}: ${result.title}`);
} catch (err) {
console.error(`❌ Failed to add movie to stream ${streamId}:`, err.message);
}
});
});
// POST /constituent/add-episode β€” body: { streamId, title, season, episode }
app.post('/constituent/add-episode', requireMainServer, async (req, res) => {
const { streamId, title, season, episode } = req.body;
if (!streamId || !title) {
return res.status(400).json({ success: false, error: 'streamId and title are required' });
}
if (season == null || episode == null) {
return res.status(400).json({ success: false, error: 'season and episode are required' });
}
if (streamId !== constituentConfig.ownerId) {
return res.status(403).json({ success: false, error: 'Only the constituent owner can add episodes to this server' });
}
// Search for the series
let searchResults;
try {
const r = await axios.get(`${IKTRACKS_BASE}/search?query=${encodeURIComponent(title)}`, { timeout: 15000 });
searchResults = (r.data?.results || []).filter(r => r && r.link);
} catch (err) {
return res.status(500).json({ success: false, error: `Search failed: ${err.message}` });
}
if (!searchResults.length) {
return res.status(404).json({ success: false, error: `No results found for "${title}"` });
}
// Pick best series result
const seriesResult = searchResults.find(r => r.type === 'series') || searchResults[0];
// Fetch details
let details;
try {
const r = await axios.get(`${IKTRACKS_BASE}/details?url=${encodeURIComponent(seriesResult.link)}`, { timeout: 15000 });
details = r.data;
if (!details) throw new Error('Empty details response');
} catch (err) {
return res.status(500).json({ success: false, error: `Details fetch failed: ${err.message}` });
}
const allEps = extractAllEpisodes(details);
if (!allEps.length) {
return res.status(404).json({ success: false, error: 'No downloadable episodes found for this title' });
}
const ep = allEps.find(e => String(e.season) === String(season) && String(e.episode) === String(episode));
if (!ep) {
return res.status(404).json({ success: false, error: `Episode S${season}E${episode} not found` });
}
const seriesName = spSeriesName(details.title || seriesResult.title || title);
const epLabel = `S${String(ep.season).padStart(2,'0')} E${String(ep.episode).padStart(2,'0')}`;
const pendingTitle = `${seriesName} β€’ ${epLabel}`;
const thumbnail = details.thumbnail || seriesResult.thumbnail || DEFAULT_ARTWORK;
res.json({ success: true, message: 'Episode queued for download and encoding', streamId, title: pendingTitle });
setImmediate(async () => {
try {
const result = await showplayEnqueueLink(streamId, ep.downloadLink, pendingTitle, thumbnail, null);
console.log(`βœ… Episode added to stream ${streamId}: ${result.title}`);
} catch (err) {
console.error(`❌ Failed to add episode to stream ${streamId}:`, err.message);
}
});
});
// POST /constituent/add-song β€” body: { streamId, songUrl, title, thumbnail? }
// Accepts a direct audio URL + title, downloads and enqueues without searching.
app.post('/constituent/add-song', requireMainServer, async (req, res) => {
const { streamId, songUrl, title, thumbnail } = req.body;
if (!streamId || !songUrl || !title) {
return res.status(400).json({ success: false, error: 'streamId, songUrl, and title are required' });
}
if (streamId !== constituentConfig.ownerId) {
return res.status(403).json({ success: false, error: 'Only the constituent owner can add songs to this server' });
}
res.json({ success: true, message: 'Song queued for download and encoding', streamId, title });
setImmediate(async () => {
try {
if (!streams[streamId]) {
streams[streamId] = { queue: [], songStartTime: null, streamTimeOffset: 0, users: new Map(), ownerId: streamId, lastActivity: Date.now(), isActive: false };
}
// Download the audio
const fileName = crypto.randomUUID() + '.mp3';
const filePath = require('path').join(SONGS_DIR, fileName);
const writer = require('fs').createWriteStream(filePath);
const response = await axios({ url: songUrl, method: 'GET', responseType: 'stream', httpsAgent: httpsAgentNoVerify });
response.data.pipe(writer);
await new Promise((resolve, reject) => {
writer.on('finish', resolve);
writer.on('error', (e) => { writer.destroy(); reject(e); });
response.data.on('error', reject);
});
const mediaMeta = await getAudioMeta(filePath);
const songInfo = {
fileName,
meta: {
title,
thumbnail: thumbnail || DEFAULT_ARTWORK,
duration: mediaMeta.duration || 0,
views: 'N/A',
published: 'N/A',
source: songUrl,
videoUrl: null,
},
};
enqueueToStream(streamId, songInfo);
console.log(`βœ… Song added to stream ${streamId}: ${title}`);
} catch (err) {
console.error(`❌ Failed to add song to stream ${streamId}:`, err.message);
}
});
});
// ─── Queue status for a stream ────────────────────────────────────────────────
app.get('/constituent/queue/:streamId', requireMainServer, (req, res) => {
const { streamId } = req.params;
const stream = streams[streamId];
if (!stream) return res.json({ success: true, streamId, queue: [], isActive: false });
const queue = stream.queue.map(s => ({
_sid: s._sid,
title: s.meta.title,
thumbnail: s.meta.thumbnail,
duration: s.meta.duration,
isVideo: !!s.meta.videoUrl,
hlsReady: !!(s._hlsPregened && typeof s._hlsStart === 'number'),
}));
res.json({
success: true,
streamId,
isActive: stream.isActive,
queue,
hlsUrl: stream.isActive ? `/stream-hls/${streamId}/live.m3u8` : null,
showplayInProgress: stream._showplayInProgress || 0,
});
});
// ═══════════════════════════════════════════════════════════════════════════
// HLS PLAYLIST ENDPOINT
// ═══════════════════════════════════════════════════════════════════════════
app.get('/stream-hls/:streamId/live.m3u8', async (req, res) => {
const streamId = req.params.streamId;
const POLL_MS = 300;
const TIMEOUT_MS = 30000;
let waited = 0;
while (waited < TIMEOUT_MS) {
const state = hlsState[streamId];
if (state && state.segments.length > 0) break;
if (!streams[streamId]) return res.status(404).send('Stream not found');
if (state && !state.generating) {
return res.status(500).send('HLS generation failed');
}
await new Promise(r => setTimeout(r, POLL_MS));
waited += POLL_MS;
}
const state = hlsState[streamId];
if (!state || state.segments.length === 0) {
return res.status(503).set('Retry-After', '3').send('HLS generation timed out, retry shortly');
}
const stream = streams[streamId];
let elapsed = 0;
if (stream && stream.songStartTime) {
const current = stream.queue[0];
const withinSong = (Date.now() - stream.songStartTime) / 1000;
const hlsStart = (current && current._hlsStart !== undefined) ? current._hlsStart : (stream.streamTimeOffset || 0);
elapsed = hlsStart + withinSong;
}
pruneOldSegments(streamId, elapsed);
const playlist = buildLivePlaylistAt(streamId, elapsed);
if (!playlist) return res.status(503).set('Retry-After', '2').send('Segments not ready yet');
res.setHeader('Content-Type', 'application/vnd.apple.mpegurl');
res.setHeader('Cache-Control', 'no-cache, no-store');
res.setHeader('Access-Control-Allow-Origin', '*');
res.send(playlist);
});
// ─── Current track ────────────────────────────────────────────────────────────
app.get('/stream/:streamId/currentTrack', (req, res) => {
const streamId = req.params.streamId;
const stream = streams[streamId];
if (!stream) return res.status(404).json({ error: 'Stream not found' });
const current = stream.queue[0];
if (!current) return res.json({ queue: [], currentIndex: 0, elapsed: 0, withinSong: 0, hlsUrl: null });
const songDuration = (current.meta.duration > 0) ? current.meta.duration : (typeof current._hlsEnd === 'number' && typeof current._hlsStart === 'number') ? (current._hlsEnd - current._hlsStart) : 0;
const hlsStartOfSong = current._hlsStart !== undefined ? current._hlsStart : (stream.streamTimeOffset || 0);
const rawWithin = stream.songStartTime ? (Date.now() - stream.songStartTime) / 1000 : 0;
const withinSong = Math.max(0, Math.min(rawWithin, songDuration));
const elapsed = hlsStartOfSong + withinSong;
const hlsStateNow = hlsState[streamId];
const hlsReady = !!(hlsStateNow && hlsStateNow.segments.length > 0 && !hlsStateNow.generating);
res.json({ queue: stream.queue.map(s => ({ _sid: s._sid, meta: s.meta, tmdb: s.tmdb || null })), currentIndex: 0, elapsed, withinSong, streamTimeOffset: hlsStartOfSong, hlsUrl: `/stream-hls/${streamId}/live.m3u8`, isVideo: !!current.meta.videoUrl, hlsReady, songId: current._sid || null, tmdb: current.tmdb || null });
});
// ─── HLS status ───────────────────────────────────────────────────────────────
app.get('/stream/:streamId/hlsStatus', (req, res) => {
const streamId = req.params.streamId;
const stream = streams[streamId];
if (!stream) return res.status(404).json({ error: 'Stream not found' });
const state = hlsState[streamId];
const current = stream.queue[0];
const generating = !!(state && state.generating);
const ready = !!(state && state.segments.length > 0);
res.json({ ready, generating, segmentsReady: ready, totalSegments: state ? state.segments.length : 0, currentSong: current ? current.meta.title : null, hlsUrl: ready ? `/stream-hls/${streamId}/live.m3u8` : null });
});
// POST /constituent/stop/:streamId β€” stop the stream entirely (owner only via main server)
app.post('/constituent/stop/:streamId', requireMainServer, (req, res) => {
const { streamId } = req.params;
const stream = streams[streamId];
if (!stream) return res.json({ success: true, message: 'No active stream' });
killActiveFFmpeg(streamId);
for (const song of stream.queue) {
const fp = path.join(SONGS_DIR, song.fileName);
if (fs.existsSync(fp)) { try { fs.unlinkSync(fp); } catch {} }
}
stream.queue = [];
stream.isActive = false;
stream.songStartTime = null;
stream.streamTimeOffset = 0;
if (hlsState[streamId]) {
const hlsStreamDir = path.join(HLS_DIR, streamId);
if (fs.existsSync(hlsStreamDir)) {
try { const files = fs.readdirSync(hlsStreamDir); for (const f of files) { try { fs.unlinkSync(path.join(hlsStreamDir, f)); } catch {} } } catch {}
}
delete hlsState[streamId]; delete hlsMutex[streamId];
}
hlsGeneration[streamId] = (hlsGeneration[streamId] || 0) + 1;
io.to(`stream:${streamId}`).emit('message', { type: 'stream_ended', message: 'Stream stopped by owner.' });
res.json({ success: true, message: 'Stream stopped.' });
});
// POST /constituent/skip/:streamId
app.post('/constituent/skip/:streamId', requireMainServer, (req, res) => {
const { streamId } = req.params;
const stream = streams[streamId];
if (!stream) return res.status(404).json({ success: false, error: 'Stream not found' });
advanceToNextSong(streamId, false);
res.json({ success: true });
});
// ═══════════════════════════════════════════════════════════════════════════
// SOCKET.IO β€” real-time updates for stream viewers
// ═══════════════════════════════════════════════════════════════════════════
io.on('connection', (socket) => {
const streamId = socket.handshake.query.streamId;
if (!streamId) { socket.emit('message', { type: 'error', message: 'streamId required' }); socket.disconnect(); return; }
socket.join(`stream:${streamId}`);
// Initialize stream entry if missing (e.g. adding media not started yet)
if (!streams[streamId]) {
streams[streamId] = { queue: [], songStartTime: null, streamTimeOffset: 0, users: new Map(), ownerId: streamId, lastActivity: Date.now(), isActive: false };
}
const stream = streams[streamId];
stream.lastActivity = Date.now();
if (stream.queue.length > 0) sendStreamUpdate(streamId, socket);
// If media is being added, immediately notify this socket
if (stream._showplayInProgress) {
socket.emit('message', { type: 'showplay_progress', stage: 'processing', title: stream.queue[0]?.meta?.title || 'media' });
}
socket.on('join-stream', (data) => {
const sid = data?.streamId || streamId;
socket.join(`stream:${sid}`);
if (streams[sid]) {
streams[sid].lastActivity = Date.now();
sendStreamUpdate(sid, socket);
}
});
socket.on('heartbeat', (data) => {
const sid = data?.streamId || streamId;
if (streams[sid]) streams[sid].lastActivity = Date.now();
});
socket.on('disconnect', () => { console.log(`Socket disconnected from stream ${streamId}`); });
});
function sendStreamUpdate(streamId, specificSocket = null) {
const stream = streams[streamId];
if (!stream || (!stream.isActive && stream.queue.length === 0)) return;
const current = stream.queue[0];
if (!current) return;
const hlsStateNow = hlsState[streamId];
const hlsReady = !!(hlsStateNow && hlsStateNow.segments.length > 0 && !hlsStateNow.generating);
const songDuration = (current.meta.duration > 0) ? current.meta.duration : (typeof current._hlsEnd === 'number' && typeof current._hlsStart === 'number') ? (current._hlsEnd - current._hlsStart) : 0;
const hlsStartOfSong = current._hlsStart !== undefined ? current._hlsStart : (stream.streamTimeOffset || 0);
const rawWithin = stream.songStartTime ? (Date.now() - stream.songStartTime) / 1000 : 0;
const withinSong = Math.max(0, Math.min(rawWithin, songDuration));
const absoluteElapsed = hlsStartOfSong + withinSong;
const nextSong = stream.queue.length > 1 ? stream.queue[1] : null;
const payload = {
type: 'update',
elapsed: absoluteElapsed,
withinSong,
streamTimeOffset: hlsStartOfSong,
currentIndex: 0,
hlsReady,
current: { file: `/songs/${current.fileName}`, meta: current.meta, isVideo: !!current.meta.videoUrl, _sid: current._sid, tmdb: current.tmdb || null },
songId: current._sid,
next: nextSong ? { file: `/songs/${nextSong.fileName}`, meta: nextSong.meta, isVideo: !!nextSong.meta.videoUrl, tmdb: nextSong.tmdb || null } : null,
queue: stream.queue,
queueLength: stream.queue.length,
hlsUrl: `/stream-hls/${streamId}/live.m3u8`
};
if (specificSocket) specificSocket.emit('message', payload);
else io.to(`stream:${streamId}`).emit('message', payload);
}
// ═══════════════════════════════════════════════════════════════════════════
// HLS ENGINE (exact logic from main server)
// ═══════════════════════════════════════════════════════════════════════════
function ensureHlsDir(streamId) {
const dir = path.join(HLS_DIR, streamId);
if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });
return dir;
}
function killActiveFFmpeg(streamId) {
hlsGeneration[streamId] = (hlsGeneration[streamId] || 0) + 1;
const cmd = activeFFmpeg[streamId];
if (cmd) {
try { cmd.kill('SIGKILL'); } catch {}
delete activeFFmpeg[streamId];
console.log(`πŸ”ͺ FFmpeg killed for stream ${streamId}`);
}
hlsMutex[streamId] = Promise.resolve();
if (hlsState[streamId]) hlsState[streamId].generating = false;
}
function buildLivePlaylistAt(streamId, elapsed) {
const state = hlsState[streamId];
if (!state || !state.segments.length) return null;
const segs = state.segments;
let startIdx = -1;
for (let i = 0; i < segs.length; i++) {
if (segs[i].streamEnd > elapsed) { startIdx = i; break; }
}
if (startIdx === -1) return null;
const window = segs.slice(startIdx, startIdx + HLS_PLAYLIST_WINDOW);
const mediaSeq = state.mediaSeq + startIdx;
const lines = ['#EXTM3U','#EXT-X-VERSION:3','#EXT-X-TARGETDURATION:10',`#EXT-X-MEDIA-SEQUENCE:${mediaSeq}`];
let prevSid = null;
for (const seg of window) {
if (prevSid !== null && seg.ownerSid && seg.ownerSid !== prevSid) {
lines.push('#EXT-X-DISCONTINUITY');
}
prevSid = seg.ownerSid || prevSid;
lines.push(`#EXTINF:${seg.duration.toFixed(6)},`);
lines.push(seg.uri);
}
return lines.join('\n') + '\n';
}
function pruneOldSegments(streamId, elapsed) {
const state = hlsState[streamId];
if (!state) return;
const dropBefore = elapsed - HLS_PLAYLIST_WINDOW * 10 * 3;
let dropped = 0;
while (state.segments.length > HLS_MAX_SEGMENTS && state.segments[0].streamEnd < dropBefore) {
const seg = state.segments.shift();
dropped++;
const dir = ensureHlsDir(streamId);
const file = path.join(dir, path.basename(seg.uri));
try { if (fs.existsSync(file)) fs.unlinkSync(file); } catch {}
}
if (dropped > 0) console.log(`πŸ—‘οΈ Pruned ${dropped} segments for stream ${streamId}`);
}
function parseM3u8Durations(playlistPath) {
if (!fs.existsSync(playlistPath)) return [];
const lines = fs.readFileSync(playlistPath, 'utf8').split('\n');
const entries = [];
for (let i = 0; i < lines.length; i++) {
if (lines[i].startsWith('#EXTINF:')) {
const dur = parseFloat(lines[i].replace('#EXTINF:', ''));
const file = (lines[i + 1] || '').trim();
if (file && !file.startsWith('#')) entries.push({ file, dur });
}
}
return entries;
}
function watchForSegments(streamId, dir, segPrefix, songHlsStart, onFirstSeg, ownerSid, state) {
let cursor = songHlsStart, firstFlushed = false;
const stitched = new Set();
const playlistPath = path.join(dir, segPrefix + '.m3u8');
let pollCount = 0;
console.log(`πŸ‘ watchForSegments created: ownerSid=${ownerSid?.slice(0,8)} songHlsStart=${songHlsStart} playlistPath=${playlistPath}`);
const flush = () => {
pollCount++;
const entries = parseM3u8Durations(playlistPath);
if (pollCount <= 3 || entries.length > 0) {
console.log(`πŸ‘ watch poll #${pollCount} [${ownerSid?.slice(0,8)}]: playlist=${fs.existsSync(playlistPath)} entries=${entries.length} stitched=${stitched.size} firstFlushed=${firstFlushed}`);
}
for (const { file, dur } of entries) {
if (stitched.has(file)) continue;
const segPath = path.join(dir, file);
try { if (fs.statSync(segPath).size < 188) continue; } catch { continue; }
stitched.add(file);
const seg = { uri: `/hls/${streamId}/${file}`, _path: segPath, streamStart: cursor, streamEnd: cursor + dur, duration: dur, ownerSid };
cursor += dur;
state.segments.push(seg);
state.totalDuration = cursor;
if (!firstFlushed) {
firstFlushed = true;
state.generating = false;
if (streams[streamId]?.queue.length > 0) {
const q0 = streams[streamId].queue[0];
const sidMatch = ownerSid ? q0._sid === ownerSid : true;
const startMatch = typeof q0._hlsStart === 'number' && songHlsStart === q0._hlsStart;
console.log(`πŸ”‘ Ownership check: sid=${q0._sid?.slice(0,8)}==${ownerSid?.slice(0,8)}:${sidMatch} hlsStart=${q0._hlsStart}==${songHlsStart}:${startMatch}`);
if (sidMatch && startMatch) {
streams[streamId].songStartTime = Date.now();
console.log(`⏱️ songStartTime reset for "${q0.meta.title}" [${q0._sid}] (first segment ready)`);
} else if (sidMatch && q0._hlsStart === undefined) {
// brief window before _hlsStart is set β€” harmless
} else {
console.log(`⚠️ watchForSegments ownership mismatch β€” skipping songStartTime reset. watcher=[${ownerSid}@${songHlsStart}] queue[0]=[${q0._sid}@${q0._hlsStart}]`);
}
}
if (onFirstSeg) onFirstSeg();
}
}
};
let lastEntryCount = -1;
let stablePolls = 0;
const STABLE_NEEDED = 3;
const iv = setInterval(() => {
flush();
const entries = parseM3u8Durations(playlistPath);
if (entries.length === lastEntryCount && !activeFFmpeg[streamId]) {
stablePolls++;
if (stablePolls >= STABLE_NEEDED) {
console.log(`πŸ›‘ watchForSegments auto-stop [${ownerSid?.slice(0,8)}]: stable for ${STABLE_NEEDED} polls, FFmpeg done`);
clearInterval(iv);
}
} else {
stablePolls = 0;
lastEntryCount = entries.length;
}
}, 800);
const markDone = () => { flush(); clearInterval(iv); return cursor; };
return { stop: () => clearInterval(iv), markDone };
}
async function generateSegmentsForSong(streamId, songInfo, isVideo, state) {
const dir = ensureHlsDir(streamId);
const songPath = path.join(SONGS_DIR, songInfo.fileName);
const segPrefix = `seg_${streamId}_${Date.now()}`;
console.log(`🎬 FFmpeg starting: ${songPath} isVideo=${isVideo}`);
if (!fs.existsSync(songPath)) throw new Error(`Source file missing: ${songPath}`);
const fileStat = fs.statSync(songPath);
if (fileStat.size === 0) throw new Error('Source file is empty');
console.log(`πŸ“ Source file: ${(fileStat.size / 1024 / 1024).toFixed(1)}MB`);
const segPattern = path.join(dir, segPrefix + '_%03d.ts');
const playlistPath = path.join(dir, segPrefix + '.m3u8');
const songHlsStart = state.totalDuration;
console.log(`🎯 segPrefix=${segPrefix} songHlsStart=${songHlsStart} ownerSid=${songInfo._sid?.slice(0,8)}`);
return new Promise((resolve, reject) => {
const cmd = ffmpeg(songPath);
if (isVideo) {
cmd.outputOptions([
'-map','0:v:0','-map','0:a:0',
'-c:v','libx264','-preset','ultrafast','-crf','28',
'-profile:v','main','-level','3.1','-pix_fmt','yuv420p',
'-vf','scale=854:480',
'-c:a','aac','-b:a','128k',
'-f','segment','-segment_time','8',
'-segment_list',playlistPath,'-segment_list_flags','+live','-segment_format','mpegts',
]);
} else {
cmd.outputOptions(['-vn','-c:a','aac','-b:a','128k','-f','segment','-segment_time','8','-segment_list',playlistPath,'-segment_list_flags','+live','-segment_format','mpegts']);
}
let watcher = null;
cmd.output(segPattern)
.on('start', () => {
activeFFmpeg[streamId] = cmd;
console.log(`🎬 FFmpeg process started [${streamId}] gen=${hlsGeneration[streamId]}`);
watcher = watchForSegments(streamId, dir, segPrefix, songHlsStart, () => {
console.log(`⚑ First segment ready for stream ${streamId}`);
sendStreamUpdate(streamId);
}, songInfo._sid, state);
})
.on('stderr', line => {
if (line.includes('Error') || line.includes('error') || line.includes('Invalid')) {
console.error(`FFmpeg stderr: ${line}`);
}
})
.on('end', () => {
console.log(`βœ… FFmpeg done for ${streamId}`);
delete activeFFmpeg[streamId];
if (!watcher) { resolve(0); return; }
const finalCursor = watcher.markDone();
console.log(`πŸ“ Final cursor from playlist: ${finalCursor.toFixed(3)}s`);
state.totalDuration = finalCursor;
try { fs.unlinkSync(playlistPath); } catch {}
resolve(finalCursor);
})
.on('error', (err) => {
console.log(`πŸ’₯ FFmpeg error for ${streamId}: ${err.message}`);
delete activeFFmpeg[streamId];
if (err.message && (err.message.includes('SIGKILL') || err.message.includes('killed'))) {
console.log(`⚑ FFmpeg killed cleanly for ${streamId} (skip)`);
if (watcher) watcher.stop();
resolve(0);
return;
}
console.error(`❌ FFmpeg error for ${streamId}:`, err.message);
if (watcher) watcher.stop();
reject(err);
})
.run();
});
}
async function appendSongToHls(streamId, songInfo) {
if (!hlsState[streamId]) {
hlsState[streamId] = { mediaSeq: 0, segments: [], totalDuration: 0, generating: true };
console.log(`πŸ“¦ appendSongToHls: created fresh hlsState for ${streamId}`);
}
const myGeneration = hlsGeneration[streamId] || 0;
const prev = hlsMutex[streamId] || Promise.resolve();
console.log(`πŸ“Œ appendSongToHls queued: "${songInfo.meta.title}" [${songInfo._sid?.slice(0,8)}] gen=${myGeneration}`);
const next = prev.then(async () => {
const currentGen = hlsGeneration[streamId] || 0;
if (currentGen !== myGeneration) {
console.log(`⏩ Skipping stale appendSongToHls for "${songInfo.meta.title}" (gen ${myGeneration} vs ${currentGen})`);
return;
}
const isVideo = !!(songInfo.meta && songInfo.meta.videoUrl);
const state = hlsState[streamId];
if (!state) {
console.log(`⏩ Skipping appendSongToHls for "${songInfo.meta.title}" β€” hlsState gone`);
return;
}
state.generating = true;
songInfo._hlsStart = state.totalDuration;
console.log(`πŸ“ _hlsStart set to ${songInfo._hlsStart.toFixed(2)}s for "${songInfo.meta.title}"`);
try {
const finalCursor = await generateSegmentsForSong(streamId, songInfo, isVideo, state);
if (typeof finalCursor === 'number' && finalCursor > 0) {
songInfo._hlsEnd = finalCursor;
const actualDuration = finalCursor - songInfo._hlsStart;
if (actualDuration > 0 && Math.abs(actualDuration - (songInfo.meta.duration || 0)) > 30) {
console.log(`πŸ“ Correcting meta.duration for "${songInfo.meta.title}": ${(songInfo.meta.duration || 0).toFixed(1)}s β†’ ${actualDuration.toFixed(1)}s`);
songInfo.meta.duration = actualDuration;
}
songInfo._hlsDurationTrusted = true;
} else {
songInfo._hlsEnd = state.totalDuration;
console.log(`⚑ Encode killed for "${songInfo.meta.title}" β€” hlsEnd set to ${songInfo._hlsEnd?.toFixed(2)}s`);
}
state.generating = false;
console.log(`πŸ“Ί HLS done for "${songInfo.meta.title}": hlsStart=${songInfo._hlsStart?.toFixed(2)}s hlsEnd=${songInfo._hlsEnd?.toFixed(2)}s segs=${state.segments.length}`);
const finalGen = hlsGeneration[streamId] || 0;
const liveStream = streams[streamId];
if (finalGen === myGeneration && liveStream && liveStream.queue[0]?._sid === songInfo._sid) {
if (!liveStream.songStartTime) {
liveStream.songStartTime = Date.now();
console.log(`⏱️ songStartTime set post-encode for "${songInfo.meta.title}" [${songInfo._sid}]`);
sendStreamUpdate(streamId);
}
preGenerateNextSong(streamId).catch(console.error);
}
} catch (err) {
console.error(`HLS generation failed for stream ${streamId}:`, err);
if (hlsState[streamId]) hlsState[streamId].generating = false;
}
});
hlsMutex[streamId] = next;
return next;
}
async function preGenerateNextSong(streamId) {
const stream = streams[streamId];
if (!stream || stream.queue.length < 2) return;
const nextSong = stream.queue[1];
if (!nextSong || nextSong._hlsPregened || nextSong._hlsPregenInProgress) return;
nextSong._hlsPregenInProgress = true;
const sid = nextSong._sid;
console.log(`πŸ”„ Pre-generating HLS for next: ${nextSong.meta.title} [${sid}]`);
try {
await appendSongToHls(streamId, nextSong);
} catch (err) {
nextSong._hlsPregenInProgress = false;
console.error(`Pre-gen failed for "${nextSong.meta.title}":`, err.message);
return;
}
const streamNow = streams[streamId];
const stillQueued = streamNow && streamNow.queue.some(s => s._sid === sid);
const encodingFinished = typeof nextSong._hlsEnd === 'number' && typeof nextSong._hlsStart === 'number' && nextSong._hlsEnd > nextSong._hlsStart;
if (stillQueued && encodingFinished) {
nextSong._hlsPregened = true;
console.log(`βœ… Pre-gen confirmed for "${nextSong.meta.title}" [${sid}]: hlsStart=${nextSong._hlsStart.toFixed(2)}s hlsEnd=${nextSong._hlsEnd.toFixed(2)}s`);
} else {
nextSong._hlsPregened = false;
nextSong._hlsPregenInProgress = false;
delete nextSong._hlsStart;
delete nextSong._hlsEnd;
console.log(`⚠️ Pre-gen invalidated for "${nextSong.meta.title}" [${sid}]`);
}
}
function advanceToNextSong(streamId, autoAdvance = false) {
const stream = streams[streamId];
if (!stream) return false;
if (autoAdvance) stream._notifyOnStart = true;
else delete stream._notifyOnStart;
killActiveFFmpeg(streamId);
const finishedSong = stream.queue.shift();
const filePath = path.join(SONGS_DIR, finishedSong.fileName);
if (fs.existsSync(filePath)) { try { fs.unlinkSync(filePath); } catch {} }
if (stream.queue.length === 0) {
stream.isActive = false;
stream.streamTimeOffset = 0;
stream.songStartTime = null;
if (hlsState[streamId]) {
const hlsDir = path.join(HLS_DIR, streamId);
if (fs.existsSync(hlsDir)) {
try { const files = fs.readdirSync(hlsDir); for (const f of files) { try { fs.unlinkSync(path.join(hlsDir, f)); } catch {} } } catch {}
}
delete hlsState[streamId]; delete hlsMutex[streamId];
}
hlsGeneration[streamId] = (hlsGeneration[streamId] || 0) + 1;
console.log(`πŸ”„ Stream ${streamId} queue empty β€” HLS state reset for fresh start`);
io.to(`stream:${streamId}`).emit('message', { type: 'stream_ended', message: 'Queue is empty.' });
return false;
}
const nextSong = stream.queue[0];
const pregenIsValid = nextSong._hlsPregened && typeof nextSong._hlsStart === 'number' && typeof nextSong._hlsEnd === 'number' && nextSong._hlsEnd > nextSong._hlsStart;
if (pregenIsValid) {
stream.streamTimeOffset = nextSong._hlsStart;
stream.songStartTime = Date.now();
stream.lastActivity = Date.now();
stream.isActive = true;
delete stream._notifyOnStart;
sendStreamUpdate(streamId);
preGenerateNextSong(streamId).catch(console.error);
} else {
nextSong._hlsPregened = nextSong._hlsPregenInProgress = false;
delete nextSong._hlsStart; delete nextSong._hlsEnd;
if (hlsState[streamId]) {
const hlsDir = path.join(HLS_DIR, streamId);
if (fs.existsSync(hlsDir)) {
try { const files = fs.readdirSync(hlsDir); for (const f of files) { try { fs.unlinkSync(path.join(hlsDir, f)); } catch {} } } catch {}
}
delete hlsState[streamId]; delete hlsMutex[streamId];
}
hlsGeneration[streamId] = (hlsGeneration[streamId] || 0) + 1;
stream.songStartTime = null;
stream.lastActivity = Date.now();
stream.isActive = true;
appendSongToHls(streamId, nextSong).then(() => {
sendStreamUpdate(streamId);
preGenerateNextSong(streamId).catch(console.error);
}).catch(console.error);
}
return true;
}
function enqueueToStream(streamId, songInfo) {
if (!streams[streamId]) {
songInfo._sid = crypto.randomUUID();
streams[streamId] = { queue: [songInfo], songStartTime: null, streamTimeOffset: 0, users: new Map(), ownerId: streamId, lastActivity: Date.now(), isActive: true };
appendSongToHls(streamId, songInfo).then(() => sendStreamUpdate(streamId)).catch(console.error);
return { songInfo, position: 1, started: true };
}
const stream = streams[streamId];
songInfo._sid = crypto.randomUUID();
stream.queue.push(songInfo);
stream.lastActivity = Date.now();
const position = stream.queue.length;
if (!stream.isActive && position === 1 && !stream._showplayInProgress) {
// Stream was idle/ended β€” ensure HLS state is fresh so this song starts at t=0.
if (!hlsState[streamId] || hlsState[streamId].totalDuration > 0) {
if (hlsState[streamId]) {
const hlsDir = path.join(HLS_DIR, streamId);
if (fs.existsSync(hlsDir)) {
try { const files = fs.readdirSync(hlsDir); for (const f of files) { try { fs.unlinkSync(path.join(hlsDir, f)); } catch {} } } catch {}
}
delete hlsState[streamId]; delete hlsMutex[streamId];
}
hlsGeneration[streamId] = (hlsGeneration[streamId] || 0) + 1;
}
stream.streamTimeOffset = 0;
stream.songStartTime = null;
stream.isActive = true;
appendSongToHls(streamId, songInfo).then(() => sendStreamUpdate(streamId)).catch(console.error);
return { songInfo, position, started: true };
}
if (stream.isActive && position >= 2) preGenerateNextSong(streamId).catch(console.error);
sendStreamUpdate(streamId);
return { songInfo, position, started: false };
}
// ═══════════════════════════════════════════════════════════════════════════
// DOWNLOAD HELPERS
// ═══════════════════════════════════════════════════════════════════════════
const DIRECT_VIDEO_EXTS = /\.(mkv|mp4|mov|avi|webm|m4v|flv|wmv|ts)(\?.*)?$/i;
function isDirectVideoUrl(url) {
if (!url) return false;
try { return DIRECT_VIDEO_EXTS.test(new URL(url).pathname); } catch { return DIRECT_VIDEO_EXTS.test(url); }
}
function getAudioMeta(filePath) {
return new Promise((resolve, reject) => {
ffmpeg.ffprobe(filePath, (err, metadata) => {
if (err) return reject(err);
function parseDurationTag(tag) {
if (!tag || typeof tag !== 'string') return 0;
const m = tag.match(/^(\d+):(\d+):(\d+(?:\.\d+)?)$/);
if (!m) return 0;
return parseInt(m[1], 10) * 3600 + parseInt(m[2], 10) * 60 + parseFloat(m[3]);
}
const candidates = [
parseFloat(metadata.format?.duration) || 0,
...(metadata.streams || []).flatMap(s => [
parseFloat(s.duration) || 0,
parseDurationTag(s.tags?.DURATION),
parseDurationTag(s.tags?.duration),
]),
];
const duration = Math.max(...candidates.filter(n => isFinite(n) && n > 0), 0);
resolve({ duration, size: metadata.format.size, bit_rate: metadata.format.bit_rate });
});
});
}
async function downloadVideoFile(downloadUrl) {
const fileName = crypto.randomUUID() + '.mp4';
const filePath = path.join(SONGS_DIR, fileName);
const writer = fs.createWriteStream(filePath);
try {
const response = await axios({ url: downloadUrl, method: 'GET', responseType: 'stream', httpsAgent: httpsAgentNoVerify });
const contentLength = parseInt(response.headers['content-length'] || '0', 10);
let bytesWritten = 0;
response.data.on('data', chunk => { bytesWritten += chunk.length; });
response.data.pipe(writer);
await new Promise((resolve, reject) => {
writer.on('finish', resolve);
writer.on('error', reject);
response.data.on('error', reject);
});
if (contentLength > 0 && bytesWritten < contentLength * 0.95) {
throw new Error(`Download truncated: got ${bytesWritten} of ${contentLength} bytes`);
}
} catch (err) {
writer.destroy();
if (fs.existsSync(filePath)) { try { fs.unlinkSync(filePath); } catch {} }
throw err;
}
return { fileName, filePath };
}
async function showplayEnqueueLink(streamId, pendingLink, pendingTitle, thumbnail, tmdbInfo = null) {
if (!streams[streamId]) {
streams[streamId] = { queue: [], songStartTime: null, streamTimeOffset: 0, users: new Map(), ownerId: streamId, lastActivity: Date.now(), isActive: false };
}
streams[streamId]._showplayInProgress = (streams[streamId]._showplayInProgress || 0) + 1;
// Notify listeners that download is starting
io.to(`stream:${streamId}`).emit('message', { type: 'showplay_progress', stage: 'downloading', title: pendingTitle });
console.log(`πŸ“₯ Downloading: ${pendingTitle}`);
let directUrl;
if (isDirectVideoUrl(pendingLink)) {
directUrl = pendingLink;
} else {
let extractRes;
try {
extractRes = await axios.get(`https://downw.vercel.app/extract?url=${encodeURIComponent(pendingLink)}`, { timeout: 60000, httpsAgent: httpsAgentNoVerify });
} catch (err) { throw new Error(`Extract API failed: ${err.message}`); }
directUrl = extractRes.data?.downloadUrl;
if (!directUrl) throw new Error('No download URL returned by extractor');
}
let fileName, filePath;
try { ({ fileName, filePath } = await downloadVideoFile(directUrl)); }
catch (err) { throw new Error(`Download failed: ${err.message}`); }
// Notify listeners that encoding is starting
io.to(`stream:${streamId}`).emit('message', { type: 'showplay_progress', stage: 'encoding', title: pendingTitle });
console.log(`βš™οΈ Encoding: ${pendingTitle}`);
let mediaMeta;
try { mediaMeta = await getAudioMeta(filePath); }
catch (e) { try { fs.unlinkSync(filePath); } catch {} throw new Error('ffprobe could not read the video file'); }
if (mediaMeta.size > SHOWPLAY_MAX_FILE_SIZE) {
try { fs.unlinkSync(filePath); } catch {}
throw new Error(`File too large (${(mediaMeta.size / (1024 ** 3)).toFixed(2)} GB). Max 2 GB.`);
}
if (mediaMeta.duration > SHOWPLAY_MAX_DURATION) {
try { fs.unlinkSync(filePath); } catch {}
throw new Error(`Video too long. Max 6 hours.`);
}
if (streams[streamId]) streams[streamId]._showplayInProgress = Math.max(0, (streams[streamId]._showplayInProgress || 1) - 1);
const effectivePoster = tmdbInfo?.poster || thumbnail || DEFAULT_ARTWORK;
const songInfo = {
fileName,
meta: {
title: tmdbInfo?.title || pendingTitle || 'Unknown',
thumbnail: effectivePoster,
duration: mediaMeta.duration || 0,
views: 'N/A',
published: tmdbInfo?.releaseDate || 'N/A',
source: pendingLink,
videoUrl: pendingLink || 'showplay',
},
tmdb: tmdbInfo || null,
isShowplay: true,
};
enqueueToStream(streamId, songInfo);
return { title: songInfo.meta.title, duration: mediaMeta.duration, thumbnail: effectivePoster };
}
// ═══════════════════════════════════════════════════════════════════════════
// BACKGROUND TIMERS
// ═══════════════════════════════════════════════════════════════════════════
// Auto-advance
setInterval(async () => {
try {
for (const streamId in streams) {
const stream = streams[streamId];
if (!stream.isActive) continue;
const current = stream.queue[0];
if (!current) continue;
let songDuration;
if (current._hlsDurationTrusted && typeof current._hlsEnd === 'number' && typeof current._hlsStart === 'number') {
songDuration = current._hlsEnd - current._hlsStart;
} else if (current.meta.duration > 0) {
songDuration = current.meta.duration;
} else if (typeof current._hlsEnd === 'number' && typeof current._hlsStart === 'number') {
songDuration = current._hlsEnd - current._hlsStart;
} else continue;
if (songDuration < 5 || !stream.songStartTime) continue;
const elapsed = (Date.now() - stream.songStartTime) / 1000;
if (elapsed >= songDuration + 3) {
if (stream._advancingFromSid === current._sid) continue;
stream._advancingFromSid = current._sid;
console.log(`⏭️ Auto-advance "${current.meta.title}": elapsed=${elapsed.toFixed(1)}s duration=${songDuration.toFixed(1)}s`);
advanceToNextSong(streamId, true);
if (stream._advancingFromSid === current._sid) delete stream._advancingFromSid;
}
}
} catch (err) { console.error('Auto-advance error:', err); }
}, 1000);
// Segment pruning
setInterval(() => {
for (const streamId in streams) {
const stream = streams[streamId];
if (!stream.isActive || !stream.songStartTime) continue;
const current = stream.queue[0];
if (!current) continue;
const hlsStart = current._hlsStart !== undefined ? current._hlsStart : (stream.streamTimeOffset || 0);
const withinSong = (Date.now() - stream.songStartTime) / 1000;
pruneOldSegments(streamId, hlsStart + withinSong);
}
}, 30 * 1000);
// Inactivity cleanup
setInterval(() => {
const now = Date.now(), toDelete = [];
for (const streamId in streams) {
const stream = streams[streamId];
if ((!stream.users || stream.users.size === 0) && (now - (stream.lastActivity || 0)) > STREAM_CLEANUP_INTERVAL) {
toDelete.push(streamId);
}
}
for (const streamId of toDelete) {
const stream = streams[streamId];
killActiveFFmpeg(streamId);
for (const song of stream.queue) { const fp = path.join(SONGS_DIR, song.fileName); if (fs.existsSync(fp)) { try { fs.unlinkSync(fp); } catch {} } }
const hlsStreamDir = path.join(HLS_DIR, streamId);
if (fs.existsSync(hlsStreamDir)) { try { fs.rmSync(hlsStreamDir, { recursive: true }); } catch {} }
delete streams[streamId]; delete hlsState[streamId]; delete hlsMutex[streamId]; delete hlsGeneration[streamId];
console.log(`🧹 Cleaned up stream: ${streamId}`);
}
}, 10 * 60 * 1000);
// ═══════════════════════════════════════════════════════════════════════════
// LAUNCH
// ═══════════════════════════════════════════════════════════════════════════
server.listen(PORT, async () => {
console.log(`πŸš€ Constituent server running on port ${PORT}`);
console.log(`πŸ‘€ Owner ID: ${CONSTITUENT_OWNER_ID}`);
try {
const ipRes = await axios.get('https://api.ipify.org?format=json', { timeout: 5000 });
console.log(`🌐 Public IP: ${ipRes.data.ip}`);
} catch {}
});
process.once('SIGINT', () => { server.close(); process.exit(0); });
process.once('SIGTERM', () => { server.close(); process.exit(0); });