Upload constituent server files

Dockerfile

@@ -0,0 +1,43 @@
+# ═══════════════════════════════════════════════════════════════════════════
+# CONSTITUENT DOCKERFILE
+# HuggingFace Docker Space — Node.js + FFmpeg streaming server
+# ═══════════════════════════════════════════════════════════════════════════
+
+FROM node:20-slim
+
+# Install ffmpeg and system utilities
+RUN apt-get update && apt-get install -y \
+    ffmpeg \
+    procps \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+# Copy package files first for layer caching
+COPY package.json ./
+
+# Install dependencies
+RUN npm install --omit=dev
+
+# Copy application files
+COPY server.js ./server.js
+
+# Create required directories
+RUN mkdir -p others/temp others/songs others/hls others/data
+
+# HuggingFace Spaces runs on port 7860 by default
+EXPOSE 7860
+
+# Environment variables that MUST be set as HuggingFace Space secrets:
+#   CONSTITUENT_OWNER_ID  — the userId from your main DB who owns this space
+#   MAIN_SERVER_SECRET    — shared secret so only your main server can call add-movie
+#   TMDB_KEY              — (optional) for TMDB enrichment
+ENV PORT=7860
+ENV NODE_ENV=production
+
+# Health check — HuggingFace polls this to decide if the space is healthy
+HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \
+    CMD curl -f http://localhost:7860/constituent/health || exit 1
+
+CMD ["node", "server.js"]

package.json

@@ -0,0 +1,11 @@
+{
+  "name": "clippy",
+  "version": "1.0.0",
+  "main": "server.js",
+  "dependencies": {
+    "express": "^4.18.2",
+    "axios": "^1.6.0",
+    "fluent-ffmpeg": "^2.1.2",
+    "socket.io": "^4.7.2"
+  }
+}

server.js

@@ -0,0 +1,1200 @@
+// ═══════════════════════════════════════════════════════════════════════════
+// 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); });