server.js

require('dotenv').config();
const fs                    = require('fs');
const path                  = require('path');
const crypto                = require('crypto');
const https                 = require('https');
const ffmpeg                = require('fluent-ffmpeg');
const jwt                   = require('jsonwebtoken');
const axios                 = require('axios');
const { MongoClient }       = require('mongodb');
const express               = require('express');
const http                  = require('http');
const cookieParser          = require('cookie-parser');
const bcrypt                = require('bcryptjs');
const { Server: SocketIOServer } = require('socket.io');

// ── Config ────────────────────────────────────────────────────────────────────
const JWT_SECRET  = process.env.JWT_SECRET  || crypto.randomBytes(64).toString('hex');
const MONGO_URI   = process.env.MONGO_URI;
const OTHERS_DIR  = process.env.OTHERS_DIR  || 'others';
const TMDB_KEY    = process.env.TMDB_KEY;    // TMDB v3 API key (set TMDB_KEY in env)
const TMDB_BASE   = 'https://api.themoviedb.org/3';
const TMDB_IMG    = 'https://image.tmdb.org/t/p/w500';

if (!MONGO_URI) { console.error('MONGO_URI environment variable is required'); process.exit(1); }

// ── MongoDB ───────────────────────────────────────────────────────────────────
let db;
async function connectDB() {
  const client = new MongoClient(MONGO_URI);

  await client.connect();
  db = client.db('redactilexyy');

  const users = db.collection('users');

  // Users
  await users.createIndex({ username: 1 }, { unique: true });
  await users.createIndex({ email: 1 }, { unique: true });

  // FIXED mirrors index (REMOVE sparse)
  await users.createIndex(
  { mirrors: 1 },
  {
    unique: true,
    partialFilterExpression: {
      "mirrors.0": { $exists: true }
    }
  }
);

  // Guests
  const guests = db.collection('guests');
  await guests.createIndex({ guestId: 1 }, { unique: true });
  await guests.createIndex(
    { lastSeen: 1 },
    { expireAfterSeconds: 7 * 24 * 60 * 60 } // 7 days TTL
  );

  // Stream queue
  const streamQueue = db.collection('stream_queue');
  await streamQueue.createIndex({ streamId: 1 });
  await streamQueue.createIndex({ _sid: 1 }, { unique: true });

  // HF keys
  await db.collection('hf_keys').createIndex({ userId: 1 }, { unique: true });

  // Constituents
  const constituents = db.collection('constituents');
  await constituents.createIndex({ userId: 1 });
  await constituents.createIndex(
    { spaceName: 1, hfUsername: 1 },
    { unique: true }
  );

  // Stream location registry — tracks where each userId's stream is hosted
  // source: 'pool' | 'constituent'
  const streamLoc = db.collection('stream_locations');
  await streamLoc.createIndex({ userId: 1 }, { unique: true });

  // Rooms, notifications, social indexes
  try {
    await db.collection('rooms').createIndex({ id: 1 }, { unique: true });
    await db.collection('rooms').createIndex({ 'members.userId': 1 });
    await db.collection('rooms').createIndex({ ownerId: 1 });
    await db.collection('notifications').createIndex({ userId: 1, read: 1 });
    await db.collection('notifications').createIndex(
      { createdAt: 1 }, { expireAfterSeconds: 30 * 24 * 60 * 60 }
    );
    await db.collection('friendships').createIndex(
      { requesterId: 1, addresseeId: 1 }, { unique: true }
    );
    await db.collection('friendships').createIndex({ addresseeId: 1 });
  } catch (e) { console.warn('Optional index creation warning:', e.message); }

  console.log('MongoDB connected.');
}

// ── 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 MAX_FILE_SIZE          = 50  * 1024 * 1024;        // 50 MB  — play/vplay
const SHOWPLAY_MAX_FILE_SIZE = 2   * 1024 * 1024 * 1024; // 2  GB  — showplay
const MAX_DURATION           = 15  * 60;                  // 15 min — play/vplay
const SHOWPLAY_MAX_DURATION  = 6   * 60 * 60;             // 6  hrs — showplay
const STREAM_CLEANUP_INTERVAL = 30 * 60 * 1000;

// ── Main-server pool limits ────────────────────────────────────────────────
// Each user on the MAIN server (no constituent) is capped at 200 MB total
// across all files currently in their stream queue.
const MAIN_POOL_MAX_SIZE_BYTES = 200 * 1024 * 1024;  // 200 MB per user

// Only 5 movies can be actively downloading/encoding on the main server at
// once (across ALL users). If the pool is full the user is told to wait or
// create a constituent server.
const MAIN_POOL_MAX_SLOTS = 5;

// Hard cap on total disk space used by ALL free-pool movies combined (SONGS_DIR).
// No new movie is accepted once this limit is reached, regardless of per-user quota.
const MAIN_POOL_DISK_CAP_BYTES = 5 * 1024 * 1024 * 1024; // 5 GB global disk cap

// Shared secret used by the main server to authenticate calls to constituents
const MAIN_SERVER_SECRET = process.env.MAIN_SERVER_SECRET || 'mysecretkeyforogudupaogeuwuwuhdg';
console.log(`🔑 MAIN_SERVER_SECRET: ${process.env.MAIN_SERVER_SECRET ? 'loaded from env' : 'using built-in default'}`);
const DEFAULT_CHANNELS = [
  {
    streamId: 'filmrise-anime',
    name: 'FilmRise Anime',
    description: 'Your home for the best anime titles, streaming live around the clock.',
    thumbnail: 'https://touchio.vercel.app/5z6v4u.webp',
    hlsUrl: 'https://dvu7aia8rjlfm.cloudfront.net/master.m3u8',
    listeners: 0,
  },
];

const MAX_CONSTITUENTS = 3;
const HF_API           = 'https://huggingface.co/api';
const DEFAULT_ARTWORK        = 'https://touchio.vercel.app/tf14k0.jpeg';

const HLS_PLAYLIST_WINDOW = 6;   // 6 segments = ~48s lookahead
const HLS_MAX_SEGMENTS    = 800; // ~106min at 8s/seg

// ── SSL agent ─────────────────────────────────────────────────────────────────
const httpsAgentNoVerify = new https.Agent({ rejectUnauthorized: false });
axios.defaults.httpsAgent = httpsAgentNoVerify;

// ── Express + Socket.IO ───────────────────────────────────────────────────────
const streamApp    = express();
const streamServer = http.createServer(streamApp);
const io           = new SocketIOServer(streamServer, {
  cors: { origin: true, credentials: true, methods: ['GET', 'POST'] },
  transports: ['websocket', 'polling']
});

streamApp.use(express.json());
streamApp.use(cookieParser());

// ── Static routes ─────────────────────────────────────────────────────────────
streamApp.use(express.static(path.join(__dirname, OTHERS_DIR, 'public')));
streamApp.get('/join', (req, res) => res.sendFile(path.join(__dirname, OTHERS_DIR, 'public', 'join.html')));
// /watch/:streamId — SPA route; sends the dashboard which handles #watch/streamId client-side
streamApp.get('/watch/:streamId', (req, res) => res.sendFile(path.join(__dirname, OTHERS_DIR, 'public', 'index.html')));
streamApp.use('/songs', express.static(SONGS_DIR));
streamApp.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', '*');
    }
  }
}));

// ── In-memory state ───────────────────────────────────────────────────────────
const streams      = {};
const userStats    = {};
const hlsState      = {};
const hlsMutex      = {};
const hlsGeneration = {};
const activeFFmpeg  = {};

// ── Active viewer tracking (userId/tempId -> streamId) ────────────────────────
const activeViewers = {}; // id -> streamId

// ── Main-server movie pool state ──────────────────────────────────────────
// Tracks how many bytes each user currently has queued on the main server
// and the total number of active showplay slots across all users.
const userQueueBytes  = {};   // userId -> total bytes of files in their queue
let   poolActiveSlots = 0;    // global counter: how many movies are mid-process

// ═══════════════════════════════════════════════════════════════════════════
// USER ID GENERATOR
// Starts at 9 digits, rolls over to 10 when 9-digit space fills up, etc.
// ═══════════════════════════════════════════════════════════════════════════

async function generateUserId() {
  // Find the highest existing numeric userId
  const last = await db.collection('users').find({}, { projection: { userId: 1 } })
    .sort({ userId: -1 }).limit(1).toArray();

  if (!last.length) return '100000000'; // first user: 9 digits

  const prev = String(last[0].userId || '100000000');
  const next = (BigInt(prev) + 1n).toString();
  return next; // naturally expands from 9 → 10 → ... digits as needed
}

// ═══════════════════════════════════════════════════════════════════════════
// AUTH MIDDLEWARE
// ═══════════════════════════════════════════════════════════════════════════

function requireAuth(req, res, next) {
  const token = req.cookies?.token;
  if (!token) return res.status(401).json({ success: false, error: 'Not authenticated' });
  try {
    const payload = jwt.verify(token, JWT_SECRET);
    req.user = payload;
    // Unified id field: guests use guestId, registered users use userId
    req.user.id = payload.isGuest ? payload.guestId : payload.userId;
    next();
  } catch {
    res.status(401).json({ success: false, error: 'Invalid or expired token' });
  }
}

// Like requireAuth but only for registered (non-guest) users
function requireRegistered(req, res, next) {
  const token = req.cookies?.token;
  if (!token) return res.status(401).json({ success: false, error: 'Not authenticated' });
  try {
    const payload = jwt.verify(token, JWT_SECRET);
    if (payload.isGuest) return res.status(403).json({ success: false, error: 'Guests cannot perform this action' });
    req.user = payload;
    req.user.id = payload.userId;
    next();
  } catch {
    res.status(401).json({ success: false, error: 'Invalid or expired token' });
  }
}

// ═══════════════════════════════════════════════════════════════════════════
// USERNAME / DISPLAY NAME VALIDATORS
// ═══════════════════════════════════════════════════════════════════════════

// Only plain ASCII printable chars — no Unicode fancy letters/symbols
const FANCY_UNICODE_RE = /[^\x20-\x7E]/;

function validateUsername(u) {
  if (!u || typeof u !== 'string') return 'Username is required';
  if (u.length < 5)  return 'Username must be at least 5 characters';
  if (u.length > 32) return 'Username must be at most 32 characters';
  if (!/^[a-zA-Z0-9_.-]+$/.test(u)) return 'Username may only contain letters, numbers, underscores, dots, and hyphens';
  return null;
}

function validateDisplayName(d) {
  if (!d || typeof d !== 'string') return 'Display name is required';
  const trimmed = d.trim();
  if (trimmed.length < 3)  return 'Display name must be at least 3 characters';
  if (trimmed.length > 15) return 'Display name must be at most 15 characters';
  if (FANCY_UNICODE_RE.test(trimmed)) return 'Display name may not contain special/fancy characters';
  return null;
}

function validateEmail(e) {
  if (!e || typeof e !== 'string') return 'Email is required';
  const trimmed = e.trim().toLowerCase();
  if (!/^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(trimmed)) return 'Invalid email address';
  if (trimmed.length > 254) return 'Email address is too long';
  return null;
}

function validatePassword(p) {
  if (!p || typeof p !== 'string') return 'Password is required';
  if (p.length < 8)  return 'Password must be at least 8 characters';
  if (p.length > 128) return 'Password must be at most 128 characters';
  return null;
}

// ═══════════════════════════════════════════════════════════════════════════
// AUTH ROUTES
// ═══════════════════════════════════════════════════════════════════════════

// POST /auth/register  { username, displayName, email, password }
streamApp.post('/auth/register', async (req, res) => {
  const { username, displayName, email, password } = req.body;

  const uErr = validateUsername(username);
  if (uErr) return res.status(400).json({ success: false, error: uErr });

  const dErr = validateDisplayName(displayName);
  if (dErr) return res.status(400).json({ success: false, error: dErr });

  const eErr = validateEmail(email);
  if (eErr) return res.status(400).json({ success: false, error: eErr });

  const pErr = validatePassword(password);
  if (pErr) return res.status(400).json({ success: false, error: pErr });

  try {
    const userId      = await generateUserId();
    const now         = new Date().toISOString();
    const passwordHash = await bcrypt.hash(password, 12);

    await db.collection('users').insertOne({
      userId,
      username:    username.toLowerCase(),
      displayName: displayName.trim(),
      email:       email.trim().toLowerCase(),
      passwordHash,
      coins:       0,
      mirrors:     [],
      createdAt:   now,
    });

    const token = jwt.sign(
      { userId, username: username.toLowerCase(), displayName: displayName.trim() },
      JWT_SECRET,
      { expiresIn: '30d' }
    );

    res.cookie('token', token, {
      httpOnly: true,
      sameSite: 'lax',
      secure:   process.env.NODE_ENV === 'production',
      maxAge:   30 * 24 * 60 * 60 * 1000, // 30 days
    });

    res.json({ success: true, userId, username: username.toLowerCase(), displayName: displayName.trim() });
  } catch (err) {
    if (err.code === 11000) {
      // Distinguish which unique field collided
      const key = Object.keys(err.keyPattern || {})[0] || '';
      const field = key.includes('email') ? 'Email' : key.includes('username') ? 'Username' : null;
      if (!field) {
        console.error('/auth/register unexpected duplicate key collision on field:', key, err.keyPattern);
        return res.status(500).json({ success: false, error: 'Registration failed. Please try again.' });
      }
      return res.status(409).json({ success: false, error: `${field} already taken` });
    }
    console.error('/auth/register error:', err.message);
    res.status(500).json({ success: false, error: 'Registration failed. Please try again.' });
  }
});

// POST /auth/login  { email, password }
streamApp.post('/auth/login', async (req, res) => {
  const { email, password } = req.body;

  const eErr = validateEmail(email);
  if (eErr) return res.status(400).json({ success: false, error: eErr });

  if (!password) return res.status(400).json({ success: false, error: 'Password is required' });

  try {
    const user = await db.collection('users').findOne({ email: email.trim().toLowerCase() });
    // Use a constant-time compare even on "not found" to avoid timing attacks
    const hashToCheck = user?.passwordHash || '$2a$12$invalidhashfortimingprotection000000000000000000000000';
    const match       = await bcrypt.compare(password, hashToCheck);

    if (!user || !match) {
      return res.status(401).json({ success: false, error: 'Invalid email or password' });
    }

    const token = jwt.sign(
      { userId: user.userId, username: user.username, displayName: user.displayName },
      JWT_SECRET,
      { expiresIn: '30d' }
    );

    res.cookie('token', token, {
      httpOnly: true,
      sameSite: 'lax',
      secure:   process.env.NODE_ENV === 'production',
      maxAge:   30 * 24 * 60 * 60 * 1000,
    });

    res.json({ success: true, userId: user.userId, username: user.username, displayName: user.displayName });
  } catch (err) {
    console.error('/auth/login error:', err.message);
    res.status(500).json({ success: false, error: 'Login failed. Please try again.' });
  }
});

// POST /auth/logout
streamApp.post('/auth/logout', (req, res) => {
  res.clearCookie('token');
  res.json({ success: true });
});

// GET /auth/me — return current user info from cookie (works for guests too)
streamApp.get('/auth/me', requireAuth, async (req, res) => {
  try {
    if (req.user.isGuest) {
      const guest = await db.collection('guests').findOneAndUpdate(
        { guestId: req.user.guestId },
        { $set: { lastSeen: new Date() } },
        { returnDocument: 'after' }
      );
      if (!guest) return res.status(404).json({ success: false, error: 'Guest session not found or expired' });
      const currentlyWatching = activeViewers[req.user.guestId] || null;
      return res.json({ success: true, user: { guestId: guest.guestId, displayName: guest.displayName, isGuest: true, createdAt: guest.createdAt, currentlyWatching } });
    }
    const user = await db.collection('users').findOne(
      { userId: req.user.userId },
      { projection: { _id: 0, userId: 1, username: 1, displayName: 1, createdAt: 1, coins: 1, mirrors: 1 } }
    );
    if (!user) return res.status(404).json({ success: false, error: 'User not found' });
    const currentlyWatching = activeViewers[req.user.userId] || null;
    res.json({ success: true, user: { ...user, coins: user.coins ?? 0, mirrors: user.mirrors || [], currentlyWatching } });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// ═══════════════════════════════════════════════════════════════════════════
// GUEST SESSION
// POST /auth/guest
//   No body  → create new guest, returns { guestId, displayName, isNew: true }
//   Body: { guestId } → restore existing guest, sets httpOnly cookie
// Guest accounts live in MongoDB with a 7-day inactivity TTL.
// ═══════════════════════════════════════════════════════════════════════════

streamApp.post('/auth/guest', async (req, res) => {
  const { guestId, displayName } = req.body || {};

  // ── Restore existing guest ────────────────────────────────────────────────
  if (guestId) {
    if (typeof guestId !== 'string' || !guestId.startsWith('g_')) {
      return res.status(400).json({ success: false, error: 'Invalid guestId' });
    }
    try {
      const guest = await db.collection('guests').findOneAndUpdate(
        { guestId },
        { $set: { lastSeen: new Date() } },
        { returnDocument: 'after' }
      );
      if (!guest) return res.status(404).json({ success: false, error: 'Guest session not found or expired' });

      const token = jwt.sign(
        { guestId: guest.guestId, displayName: guest.displayName, isGuest: true },
        JWT_SECRET,
        { expiresIn: '7d' }
      );
      res.cookie('token', token, {
        httpOnly: true, sameSite: 'lax',
        secure: process.env.NODE_ENV === 'production',
        maxAge: 7 * 24 * 60 * 60 * 1000,
      });
      return res.json({ success: true, guestId: guest.guestId, displayName: guest.displayName, isNew: false });
    } catch (err) {
      console.error('/auth/guest restore error:', err.message);
      return res.status(500).json({ success: false, error: 'Failed to restore guest session' });
    }
  }

  // ── Create new guest ──────────────────────────────────────────────────────
  if (!displayName || typeof displayName !== 'string') {
    return res.status(400).json({ success: false, error: 'displayName is required for new guest sessions' });
  }
  const trimmed = displayName.trim();
  if (trimmed.length < 1 || trimmed.length > 30) {
    return res.status(400).json({ success: false, error: 'displayName must be 1–30 characters' });
  }

  try {
    const newGuestId = 'g_' + crypto.randomBytes(16).toString('hex');
    const now = new Date();
    await db.collection('guests').insertOne({
      guestId:     newGuestId,
      displayName: trimmed,
      createdAt:   now,
      lastSeen:    now,
    });

    const token = jwt.sign(
      { guestId: newGuestId, displayName: trimmed, isGuest: true },
      JWT_SECRET,
      { expiresIn: '7d' }
    );
    res.cookie('token', token, {
      httpOnly: true, sameSite: 'lax',
      secure: process.env.NODE_ENV === 'production',
      maxAge: 7 * 24 * 60 * 60 * 1000,
    });
    return res.json({ success: true, guestId: newGuestId, displayName: trimmed, isNew: true });
  } catch (err) {
    console.error('/auth/guest create error:', err.message);
    return res.status(500).json({ success: false, error: 'Failed to create guest session' });
  }
});

// ═══════════════════════════════════════════════════════════════════════════
// MIRROR APIs
// GET  /api/mirrors          — list caller's mirrors (auth required)
// POST /api/mirrors/add      — add a mirror   { mirror }
// POST /api/mirrors/remove   — remove a mirror { mirror }
// A mirror is a 3–10 char alphanumeric string that also resolves as a streamId
// Max 5 mirrors per user. Must be globally unique.
// ═══════════════════════════════════════════════════════════════════════════

const MIRROR_RE = /^[a-zA-Z0-9]{3,10}$/;

streamApp.get('/api/mirrors', requireRegistered, async (req, res) => {
  try {
    const user = await db.collection('users').findOne({ userId: req.user.userId }, { projection: { _id: 0, mirrors: 1 } });
    res.json({ success: true, mirrors: user?.mirrors || [] });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

streamApp.post('/api/mirrors/add', requireRegistered, async (req, res) => {
  const { mirror } = req.body;
  if (!mirror || !MIRROR_RE.test(mirror)) return res.status(400).json({ success: false, error: 'mirror must be 3–10 alphanumeric characters' });
  const m = mirror.toLowerCase();
  try {
    const user = await db.collection('users').findOne({ userId: req.user.userId }, { projection: { mirrors: 1 } });
    const current = user?.mirrors || [];
    if (current.length >= 5) return res.status(400).json({ success: false, error: 'Maximum of 5 mirrors reached' });
    if (current.includes(m)) return res.status(409).json({ success: false, error: 'You already have this mirror' });
    // Check global uniqueness
    const taken = await db.collection('users').findOne({ mirrors: m });
    if (taken) return res.status(409).json({ success: false, error: 'Mirror already taken' });
    await db.collection('users').updateOne({ userId: req.user.userId }, { $addToSet: { mirrors: m } });
    res.json({ success: true, mirror: m, mirrors: [...current, m] });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

streamApp.post('/api/mirrors/remove', requireRegistered, async (req, res) => {
  const { mirror } = req.body;
  if (!mirror) return res.status(400).json({ success: false, error: 'mirror is required' });
  const m = mirror.toLowerCase();
  try {
    const result = await db.collection('users').updateOne({ userId: req.user.userId }, { $pull: { mirrors: m } });
    if (result.modifiedCount === 0) return res.status(404).json({ success: false, error: 'Mirror not found on your account' });
    const updated = await db.collection('users').findOne({ userId: req.user.userId }, { projection: { mirrors: 1 } });
    res.json({ success: true, removed: m, mirrors: updated?.mirrors || [] });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// ═══════════════════════════════════════════════════════════════════════════
// MONGODB STAT HELPERS
// ═══════════════════════════════════════════════════════════════════════════

async function loadUserStats(userId) {
  try {
    const stats = await db.collection('stream_user_stats').findOne({ userId });
    if (stats) {
      userStats[userId] = {
        totalListeningTime: stats.totalListeningTime || 0,
        sessions:           stats.sessions           || [],
        songCount:          stats.songCount          || 0,
        lastUpdated:        stats.lastUpdated
      };
    }
  } catch (err) { console.error('Error loading user stats:', err); }
}

async function saveUserStats(userId) {
  try {
    await db.collection('stream_user_stats').updateOne(
      { userId },
      { $set: { userId, ...userStats[userId] } },
      { upsert: true }
    );
  } catch (err) { console.error('Error saving user stats:', err); }
}

async function initUserStats(userId) {
  if (!userStats[userId]) {
    await loadUserStats(userId);
    if (!userStats[userId]) {
      userStats[userId] = { totalListeningTime: 0, sessions: [], songCount: 0, lastUpdated: new Date().toISOString() };
      await saveUserStats(userId);
    }
  }
}

async function trackListeningSession(userId, duration) {
  await initUserStats(userId);
  const now = new Date();
  userStats[userId].sessions.push({ timestamp: now.toISOString(), duration });
  userStats[userId].totalListeningTime += duration;
  userStats[userId].lastUpdated = now.toISOString();
  const NINETY_DAYS_AGO = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000).toISOString();
  userStats[userId].sessions = userStats[userId].sessions.filter(s => s.timestamp >= NINETY_DAYS_AGO);
  if (userStats[userId].sessions.length > 500) {
    userStats[userId].sessions = userStats[userId].sessions.slice(-500);
  }
  await saveUserStats(userId);
}

// Periodic eviction of cold user caches (not accessed in 2 hours)
setInterval(() => {
  const cutoff = Date.now() - 2 * 60 * 60 * 1000;
  for (const userId of Object.keys(userStats)) {
    const last = userStats[userId].lastUpdated ? new Date(userStats[userId].lastUpdated).getTime() : 0;
    if (last < cutoff) delete userStats[userId];
  }
}, 30 * 60 * 1000);

async function trackSongListened(userId) {
  await initUserStats(userId);
  userStats[userId].songCount += 1;
  await saveUserStats(userId);
}

async function getUserStats(userId) {
  await initUserStats(userId);
  const stats = userStats[userId];
  const now   = new Date();
  const todayStart = new Date(now.getFullYear(), now.getMonth(), now.getDate());
  const weekStart  = new Date(now); weekStart.setDate(now.getDate() - now.getDay()); weekStart.setHours(0,0,0,0);
  const monthStart = new Date(now.getFullYear(), now.getMonth(), 1);
  let todayTime = 0, weekTime = 0, monthTime = 0;
  stats.sessions.forEach(session => {
    const d = new Date(session.timestamp);
    if (d >= todayStart) todayTime += session.duration;
    if (d >= weekStart)  weekTime  += session.duration;
    if (d >= monthStart) monthTime += session.duration;
  });
  return { totalTime: stats.totalListeningTime, todayTime, weekTime, monthTime, songCount: stats.songCount };
}

// ═══════════════════════════════════════════════════════════════════════════
// MAIN-SERVER POOL HELPERS
// ═══════════════════════════════════════════════════════════════════════════

/**
 * Returns the total bytes currently held in a user's stream queue.
 * Reads actual file sizes on disk so the number stays accurate even
 * after files are deleted by advanceToNextSong.
 */
function getUserQueueBytes(streamId) {
  const stream = streams[streamId];
  if (!stream || !stream.queue.length) return 0;
  let total = 0;
  for (const song of stream.queue) {
    const fp = path.join(SONGS_DIR, song.fileName);
    try { total += fs.statSync(fp).size; } catch {}
  }
  userQueueBytes[streamId] = total;
  return total;
}

/**
 * Returns the total bytes of all files currently sitting in SONGS_DIR.
 * This is the true global pool disk usage across every user on the main server.
 */
function getPoolDiskBytes() {
  let total = 0;
  try {
    for (const f of fs.readdirSync(SONGS_DIR)) {
      try { total += fs.statSync(path.join(SONGS_DIR, f)).size; } catch {}
    }
  } catch {}
  return total;
}

/**
 * Check whether the main-server pool can accept a new showplay job for streamId.
 * Optionally pass expectedBytes (estimated file size from Content-Length) to
 * pre-check whether the incoming movie would push the global disk cap over the limit.
 * Returns { allowed: bool, reason?: string, diskFull?: bool, diskUsedGB?: string, diskCapGB?: string }
 */
function checkPoolAvailability(streamId, expectedBytes = 0) {
  if (poolActiveSlots >= MAIN_POOL_MAX_SLOTS) {
    return {
      allowed: false,
      reason:  `The main server pool is full (${MAIN_POOL_MAX_SLOTS} movies are already processing). ` +
               `Please wait for a slot to free up, or create a constituent server to get your own dedicated capacity.`,
    };
  }

  // ── Global 5 GB disk cap (enforced before per-user check) ─────────────────
  const poolDiskBytes  = getPoolDiskBytes();
  const projectedBytes = poolDiskBytes + expectedBytes;
  const capGB          = (MAIN_POOL_DISK_CAP_BYTES / 1024 ** 3).toFixed(0);

  if (poolDiskBytes >= MAIN_POOL_DISK_CAP_BYTES) {
    const usedGB = (poolDiskBytes / 1024 ** 3).toFixed(2);
    return {
      allowed:    false,
      diskFull:   true,
      diskUsedGB: usedGB,
      diskCapGB:  capGB,
      reason:     `The free pool is out of disk space (${usedGB} GB / ${capGB} GB used). ` +
                  `Movies will free up space as they finish playing. ` +
                  `Use a constituent server for your own dedicated storage.`,
    };
  }
  if (expectedBytes > 0 && projectedBytes > MAIN_POOL_DISK_CAP_BYTES) {
    const usedGB     = (poolDiskBytes  / 1024 ** 3).toFixed(2);
    const freeGB     = ((MAIN_POOL_DISK_CAP_BYTES - poolDiskBytes) / 1024 ** 3).toFixed(2);
    const expectedGB = (expectedBytes  / 1024 ** 3).toFixed(2);
    return {
      allowed:    false,
      diskFull:   true,
      diskUsedGB: usedGB,
      diskCapGB:  capGB,
      reason:     `This movie (~${expectedGB} GB) would exceed the free pool disk cap of ${capGB} GB ` +
                  `(only ${freeGB} GB remaining). Wait for movies to finish playing or use a constituent server.`,
    };
  }

  // ── Per-user 200 MB quota ─────────────────────────────────────────────────
  const usedBytes = getUserQueueBytes(streamId);
  if (usedBytes >= MAIN_POOL_MAX_SIZE_BYTES) {
    const usedMB = (usedBytes / 1024 / 1024).toFixed(1);
    return {
      allowed: false,
      reason:  `You are using ${usedMB} MB of your 200 MB main-server quota. ` +
               `Wait for a movie to finish playing (it will be freed automatically) ` +
               `or use a constituent server for unlimited capacity.`,
    };
  }

  return {
    allowed:    true,
    diskUsedGB: (poolDiskBytes / 1024 ** 3).toFixed(2),
    diskCapGB:  capGB,
    diskFreeGB: ((MAIN_POOL_DISK_CAP_BYTES - poolDiskBytes) / 1024 ** 3).toFixed(2),
  };
}

// ═══════════════════════════════════════════════════════════════════════════
// HLS ENGINE
// ═══════════════════════════════════════════════════════════════════════════

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} (gen ${hlsGeneration[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) {
    // Insert a discontinuity marker whenever the owning song changes.
    // This tells HLS players (hls.js, AVPlayer, etc.) to flush their buffer
    // and reset decoder state — preventing the tail of one song playing under
    // the next song's metadata, or the new song starting mid-buffer.
    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}`);
    }
    if (entries.length > 0 && !firstFlushed) {
      console.log(`🔍 watchForSegments flush: ${entries.length} entries, ownerSid=[${ownerSid?.slice(0,8)}] songHlsStart=${songHlsStart}`);
    }
    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];
          // Reset songStartTime only when this watcher owns the current track.
          // Guard with both exact _hlsStart match AND _sid identity so a stale
          // watcher from a killed pregen can never claim ownership of a different song.
          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)`);
            if (streams[streamId]._notifyOnStart) {
              delete streams[streamId]._notifyOnStart;
            }
          } else if (sidMatch && q0._hlsStart === undefined) {
            // Same song but _hlsStart not yet assigned — this watcher fired in the
            // brief window between advanceToNextSong deleting _hlsStart and the new
            // appendSongToHls mutex job setting it. Harmless; the new encode's watcher
            // will fire again once _hlsStart is set and handle it correctly.
          } 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; // 3 × 800ms = 2.4s with no new entries → auto-stop

  const iv = setInterval(() => {
    flush();
    // Auto-stop: if entry count hasn't changed for STABLE_NEEDED consecutive polls
    // AND ffmpeg already finished (no activeFFmpeg), the interval is stale and leaking.
    // This prevents runaway polling on long files (e.g. 1h+ TV episodes) that fills RAM
    // and eventually crashes the process.
    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}`);
  console.log(`🔍 generateSegmentsForSong: dir=${dir} exists=${fs.existsSync(dir)} songExists=${fs.existsSync(songPath)} stateSegs=${state.segments.length} stateTotalDur=${state.totalDuration}`);

  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', (cmdLine) => {
        activeFFmpeg[streamId] = cmd;
        console.log(`🎬 FFmpeg process started [${streamId}] gen=${hlsGeneration[streamId]} activeFFmpegSet=true`);
        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} — activeFFmpeg=${activeFFmpeg[streamId] ? 'still set' : 'already cleared'}`);
        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, stdout, stderr) => {
        console.log(`💥 FFmpeg error for ${streamId}: ${err.message} activeFFmpeg=${activeFFmpeg[streamId] ? 'still set' : 'already cleared'}`);
        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} hlsStatExists=${!!hlsState[streamId]}`);
  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;
    }
    console.log(`▶️  appendSongToHls mutex running for "${songInfo.meta.title}" [${songInfo._sid}] gen=${myGeneration}`);
    const isVideo = !!(songInfo.meta && songInfo.meta.videoUrl);
    const state   = hlsState[streamId];
    if (!state) {
      console.log(`⏩ Skipping appendSongToHls for "${songInfo.meta.title}" — hlsState gone (gen=${myGeneration})`);
      return;
    }
    console.log(`📊 State at encode start: segs=${state.segments.length} totalDuration=${state.totalDuration.toFixed(2)}s generating=${state.generating}`);
    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);
      console.log(`📊 generateSegmentsForSong returned: finalCursor=${finalCursor} for "${songInfo.meta.title}" gen=${myGeneration} currentGen=${hlsGeneration[streamId]}`);
      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];
      console.log(`📊 Post-encode check: finalGen=${finalGen} myGen=${myGeneration} queue[0]._sid=${liveStream?.queue[0]?._sid?.slice(0,8)} song._sid=${songInfo._sid?.slice(0,8)} songStartTime=${liveStream?.songStartTime ? 'set' : 'null'}`);
      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;

  // Mark in-progress but do NOT set _hlsPregened=true yet — encoding hasn't finished.
  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;
  }

  // Only mark pregened if:
  //  1. This exact song object is still in queue (wasn't skipped away)
  //  2. Encoding produced valid _hlsStart and _hlsEnd
  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}]: stillQueued=${stillQueued} encodingFinished=${encodingFinished}`);
  }
}

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 {} }
  // Remove finished item from DB queue (showplay items only, but harmless for others)
  if (finishedSong._sid) removeQueueItem(finishedSong._sid).catch(console.error);
  // Refresh user's byte quota after deletion
  getUserQueueBytes(streamId);

  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`);
    // Purge any remaining DB queue entries (safety net)
    db.collection('stream_queue').deleteMany({ streamId }).catch(console.error);
    io.to(`stream:${streamId}`).emit('message', { type: 'stream_ended', message: 'Queue is empty. Add more songs to continue!' });
    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;
    console.log(`⏱️  songStartTime set immediately for pregened "${nextSong.meta.title}"`);
    delete stream._notifyOnStart;
    sendStreamUpdate(streamId);
    preGenerateNextSong(streamId).catch(console.error);
  } else {
    console.log(`🆕 Fresh-start encode for "${nextSong.meta.title}"`);
    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;
}

// ═══════════════════════════════════════════════════════════════════════════
// HLS PLAYLIST ENDPOINT
// ═══════════════════════════════════════════════════════════════════════════

streamApp.get('/stream-hls/:streamId/live.m3u8', async (req, res) => {
  const streamId = await resolveStreamId(req.params.streamId);
  const POLL_MS = 300, 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) || !state) {
      await new Promise(r => setTimeout(r, POLL_MS));
      waited += POLL_MS;
      continue;
    }
    return res.status(500).send('HLS generation failed — check server logs');
  }
  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, retry shortly');
  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);
});

// ═══════════════════════════════════════════════════════════════════════════
// EXPRESS REST ROUTES — stream viewer endpoints
// ═══════════════════════════════════════════════════════════════════════════

streamApp.get('/stream/:streamId', (req, res) => {
  res.sendFile(path.join(__dirname, OTHERS_DIR, 'public', 'stream.html'));
});

// ── Mirror resolution helper ──────────────────────────────────────────────────
// Resolves a streamId that may be a mirror alias or a userId.
// Returns the real streamId (userId) or null if not found.
async function resolveStreamId(id) {
  if (!id) return null;
  // Check if it looks like a mirror (3-10 alphanumeric)
  if (MIRROR_RE.test(id)) {
    const owner = await db.collection('users').findOne({ mirrors: id.toLowerCase() }, { projection: { userId: 1 } });
    if (owner) return owner.userId;
  }
  // Otherwise treat as direct userId
  return id;
}

// ── Stream location registry ──────────────────────────────────────────────────
// Records where a user's stream is hosted so viewer-facing endpoints can proxy
// correctly when the stream lives on a constituent rather than the main pool.
//
// source : 'pool' | 'constituent'
// spaceUrl / spaceName are set only for constituent sources.

async function setStreamLocation(userId, source, spaceUrl = null, spaceName = null) {
  try {
    await db.collection('stream_locations').updateOne(
      { userId },
      { $set: { userId, source, spaceUrl, spaceName, updatedAt: new Date() } },
      { upsert: true }
    );
  } catch (err) {
    console.error('setStreamLocation error:', err.message);
  }
}

// Returns the location record or null.
// { source:'pool' } or { source:'constituent', spaceUrl, spaceName }
async function getStreamLocation(userId) {
  try {
    return await db.collection('stream_locations').findOne({ userId });
  } catch { return null; }
}

// For viewer-facing endpoints that receive a *stream owner's* streamId:
// If the stream is not in the local pool, attempt to proxy to its constituent.
// `proxyFn` receives `(spaceUrl, streamId)` and should call the constituent API
// and settle `res` itself.  Returns true if proxied, false if not applicable.
async function proxyToConstituentIfNeeded(streamId, res, proxyFn) {
  if (streams[streamId]) return false; // present in local pool — caller handles it

  const loc = await getStreamLocation(streamId);
  if (!loc || loc.source !== 'constituent' || !loc.spaceUrl) return false;

  try {
    await proxyFn(loc.spaceUrl, streamId);
    return true;
  } catch (err) {
    const body = err.response?.data;
    const status = err.response?.status || 502;
    res.status(status).json(body || { error: err.message });
    return true;
  }
}

// Viewer joined
// Body: { name, id }  where id is userId, tempId (guest), passed explicitly
streamApp.post('/joined/:streamId', async (req, res) => {
  const rawId = req.params.streamId;
  const streamId = await resolveStreamId(rawId);
  const { name, id } = req.body;
  if (!name || !id) return res.status(400).json({ error: 'Name and id are required' });

  // If stream not in local pool, proxy to its constituent
  const proxied = await proxyToConstituentIfNeeded(streamId, res, async (spaceUrl) => {
    const r = await axios.post(`${spaceUrl}/joined/${streamId}`, { name, id },
      { headers: { 'Content-Type': 'application/json' }, timeout: 8000 });
    res.json(r.data);
  });
  if (proxied) return;

  const stream = streams[streamId];
  if (!stream) return res.status(404).json({ error: 'Stream not found' });
  if (!stream.users) stream.users = new Map();
  stream.lastActivity = Date.now();
  const isGuest = id.startsWith('g_');
  if (isGuest) {
    db.collection('guests').updateOne({ guestId: id }, { $set: { lastSeen: new Date() } }).catch(console.error);
  } else {
    activeViewers[id] = streamId;
  }
  stream.users.set(id, { id, name, isGuest, joinedAt: new Date().toISOString(), lastHeartbeat: new Date().toISOString(), connectionStart: Date.now() });
  broadcastUserUpdate(streamId, 'user_joined', { id, name, isGuest }, true);
  res.json({ success: true, message: `User ${name} joined stream ${streamId}`, totalUsers: stream.users.size, resolvedStreamId: streamId });
});

// Viewer left
streamApp.post('/left/:streamId', async (req, res) => {
  const rawId = req.params.streamId;
  const streamId = await resolveStreamId(rawId);
  const { id } = req.body;
  if (!id) return res.status(400).json({ error: 'User id is required' });

  const proxied = await proxyToConstituentIfNeeded(streamId, res, async (spaceUrl) => {
    const r = await axios.post(`${spaceUrl}/left/${streamId}`, { id },
      { headers: { 'Content-Type': 'application/json' }, timeout: 8000 });
    res.json(r.data);
  });
  if (proxied) return;

  const stream = streams[streamId];
  if (!stream) return res.status(404).json({ error: 'Stream not found' });
  if (!stream.users) stream.users = new Map();
  const user = stream.users.get(id);
  if (!user) return res.status(404).json({ error: 'User not found in stream' });
  const listenDuration = (Date.now() - user.connectionStart) / 1000;
  if (!user.isGuest) await trackListeningSession(id, listenDuration);
  stream.lastActivity = Date.now();
  stream.users.delete(id);
  if (user.isGuest) {
    db.collection('guests').updateOne({ guestId: id }, { $set: { lastSeen: new Date() } }).catch(console.error);
  } else {
    delete activeViewers[id];
  }
  broadcastUserUpdate(streamId, 'user_left', { id, name: user.name, isGuest: user.isGuest }, true);
  res.json({ success: true, message: `User ${user.name} left stream ${streamId}`, totalUsers: stream.users.size });
});

// GET /listeners/:streamId — used by frontend watch page to show who's watching
streamApp.get('/listeners/:streamId', async (req, res) => {
  const streamId = await resolveStreamId(req.params.streamId);

  const proxied = await proxyToConstituentIfNeeded(streamId, res, async (spaceUrl) => {
    const r = await axios.get(`${spaceUrl}/listeners/${streamId}`,
      { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 8000 });
    res.json(r.data);
  });
  if (proxied) return;

  const stream = streams[streamId];
  if (!stream) return res.json({ success: true, listeners: [] });
  const list = stream.users
    ? Array.from(stream.users.values()).map(u => ({ name: u.name, isGuest: u.isGuest }))
    : [];
  res.json({ success: true, listeners: list });
});

// Viewer list
streamApp.get('/list/:streamId', async (req, res) => {
  const rawId = req.params.streamId;
  const streamId = await resolveStreamId(rawId);

  const proxied = await proxyToConstituentIfNeeded(streamId, res, async (spaceUrl) => {
    const r = await axios.get(`${spaceUrl}/list/${streamId}`,
      { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 8000 });
    res.json(r.data);
  });
  if (proxied) return;

  const stream = streams[streamId];
  if (!stream) return res.status(404).json({ error: 'Stream not found' });
  if (!stream.users) stream.users = new Map();
  res.json({ streamId, totalUsers: stream.users.size, users: Array.from(stream.users.values()) });
});

// Heartbeat — accepts userId (registered) or tempId (guest)
// For registered users on the main pool: tracks listening session time so
// stats stay current without waiting for an explicit /left event.
streamApp.post('/heartbeat/:streamId', async (req, res) => {
  const rawId = req.params.streamId;
  const streamId = await resolveStreamId(rawId);
  const { userId } = req.body;
  if (!userId) return res.status(400).json({ error: 'userId is required' });

  const proxied = await proxyToConstituentIfNeeded(streamId, res, async (spaceUrl) => {
    const r = await axios.post(`${spaceUrl}/heartbeat/${streamId}`, { userId },
      { headers: { 'Content-Type': 'application/json' }, timeout: 8000 });
    res.json(r.data);
  });
  if (proxied) return;

  const stream = streams[streamId];
  if (!stream) return res.json({ success: true }); // stream not initialized yet — no-op
  if (!stream.users) stream.users = new Map();
  stream.lastActivity = Date.now();

  const isGuest = userId.startsWith('g_');
  if (isGuest) {
    db.collection('guests').updateOne({ guestId: userId }, { $set: { lastSeen: new Date() } }).catch(console.error);
  } else {
    activeViewers[userId] = streamId;
  }

  const user = stream.users.get(userId);
  if (user) {
    const now     = new Date();
    const prev    = new Date(user.lastHeartbeat || user.joinedAt);
    const elapsed = Math.max(0, (now - prev) / 1000); // seconds since last heartbeat

    user.lastHeartbeat = now.toISOString();

    // Accumulate listening time on every heartbeat for registered users.
    // Clamp to 120 s to avoid crediting huge gaps from reconnections.
    if (!isGuest && elapsed > 0 && elapsed < 120) {
      trackListeningSession(userId, elapsed).catch(console.error);
    }
  }

  // Return success regardless — constituent-stream viewers aren't in this map
  res.json({ success: true });
});

// Current track info
streamApp.get('/stream/:streamId/currentTrack', async (req, res) => {
  const streamId = await resolveStreamId(req.params.streamId);

  const proxied = await proxyToConstituentIfNeeded(streamId, res, async (spaceUrl) => {
    const r = await axios.get(`${spaceUrl}/stream/${streamId}/currentTrack`,
      { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 8000 });
    res.json(r.data);
  });
  if (proxied) return;

  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);
  const enrichedQueue = stream.queue.map(s => ({
    _sid:     s._sid,
    meta:     s.meta,
    tmdb:     s.tmdb || null,
    isShowplay: s.isShowplay || false,
    _hlsStart: s._hlsStart,
    _hlsEnd:   s._hlsEnd,
  }));
  res.json({ queue: enrichedQueue, 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
streamApp.get('/stream/:streamId/hlsStatus', async (req, res) => {
  const streamId = await resolveStreamId(req.params.streamId);

  const proxied = await proxyToConstituentIfNeeded(streamId, res, async (spaceUrl) => {
    const r = await axios.get(`${spaceUrl}/stream/${streamId}/hlsStatus`,
      { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 8000 });
    res.json(r.data);
  });
  if (proxied) return;

  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 segmentsReady = !!(state && state.segments.length > 0);
  const ready         = segmentsReady;
  res.json({ ready, generating, segmentsReady, totalSegments: state ? state.segments.length : 0, currentSong: current ? current.meta.title : null, hlsUrl: ready ? `/stream-hls/${streamId}/live.m3u8` : null });
});

// All active streams (paginated)
streamApp.get('/api/streams', (req, res) => {
  const page    = parseInt(req.query.page) || 1;
  const perPage = 30;
  const filter  = req.query.filter || 'all';
  try {
    let activeStreams = Object.entries(streams)
      .map(([streamId, stream]) => {
        const current         = stream.queue[0];
        const next            = stream.queue.length > 1 ? stream.queue[1] : null;
        const activeListeners = stream.users?.size || 0;
        return {
          streamId,
          currentSong:  current ? { title: current.meta.title, thumbnail: current.meta.thumbnail, duration: current.meta.duration } : null,
          nextSong:     next    ? { title: next.meta.title,    thumbnail: next.meta.thumbnail,    duration: next.meta.duration }    : null,
          activeListeners, queueLength: stream.queue.length, isPlaying: stream.isActive && current !== null,
          hlsReady: !!(hlsState[streamId] && hlsState[streamId].segments.length > 0 && !hlsState[streamId].generating),
        };
      });
    if (filter === 'active')  activeStreams.sort((a, b) => b.activeListeners - a.activeListeners);
    if (filter === 'playing') activeStreams = activeStreams.filter(g => g.isPlaying);
    const total     = activeStreams.length;
    const totalPages = Math.ceil(total / perPage);
    const cur       = Math.min(Math.max(1, page), totalPages || 1);
    const start     = (cur - 1) * perPage;
    res.json({ success: true, data: activeStreams.slice(start, start + perPage), pagination: { currentPage: cur, totalPages, total, perPage }, filter });
  } catch (err) {
    res.status(500).json({ success: false, error: 'Failed to fetch streams' });
  }
});

// Stream status for a specific user stream
streamApp.get('/api/stream-status', requireRegistered, (req, res) => {
  const streamId = req.user.userId;
  const stream   = streams[streamId];
  if (!stream || !stream.queue || stream.queue.length === 0) {
    return res.json({ success: true, streamId, nowPlaying: null, upNext: null });
  }
  const nowPlaying = stream.queue[0];
  const upNext     = stream.queue.length > 1 ? stream.queue[1] : null;
  res.json({
    success: true, streamId,
    nowPlaying: { title: nowPlaying.meta.title, duration: nowPlaying.meta.duration, thumbnail: nowPlaying.meta.thumbnail, isVideo: !!nowPlaying.meta.videoUrl, tmdb: nowPlaying.tmdb || null, isShowplay: nowPlaying.isShowplay || false },
    upNext: upNext ? { title: upNext.meta.title, duration: upNext.meta.duration, thumbnail: upNext.meta.thumbnail, isVideo: !!upNext.meta.videoUrl, tmdb: upNext.tmdb || null, isShowplay: upNext.isShowplay || false } : null,
    hlsUrl: `/stream-hls/${streamId}/live.m3u8`,
    pool: {
      slotsUsed:    poolActiveSlots,
      slotsMax:     MAIN_POOL_MAX_SLOTS,
      userMBUsed:   parseFloat((getUserQueueBytes(streamId) / 1024 / 1024).toFixed(1)),
      userMBMax:    200,
      diskUsedGB:   parseFloat((getPoolDiskBytes() / 1024 ** 3).toFixed(2)),
      diskCapGB:    parseFloat((MAIN_POOL_DISK_CAP_BYTES / 1024 ** 3).toFixed(0)),
      diskFreeGB:   parseFloat(((MAIN_POOL_DISK_CAP_BYTES - getPoolDiskBytes()) / 1024 ** 3).toFixed(2)),
    },
  });
});

// Stream listeners
streamApp.get('/api/stream-listeners', requireRegistered, (req, res) => {
  const streamId = req.user.userId;
  const stream   = streams[streamId];
  if (!stream) return res.json({ success: true, streamId, activeListeners: [], totalActive: 0 });
  const listeners = stream.users ? Array.from(stream.users.values()).map(u => ({ id: u.id, name: u.name, joinedAt: u.joinedAt, listeningDuration: Math.floor((Date.now() - u.connectionStart) / 1000) })) : [];
  res.json({ success: true, streamId, activeListeners: listeners, totalActive: listeners.length });
});


// GET /api/my-streams — returns ALL active streams (constituents + pool) for the user
// Used when a user has multiple constituents possibly playing simultaneously
streamApp.get('/api/my-streams', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  const result = { sources: [], pool: null };

  try {
    const constituents = await db.collection('constituents').find({ userId }).toArray();
    await Promise.all(constituents.map(async (c) => {
      try {
        const r = await axios.get(
          `${c.spaceUrl}/constituent/queue/${userId}`,
          { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 8000 }
        );
        const data = r.data;
        if (!data.success) return;
        const current = data.queue?.[0] || null;
        result.sources.push({
          type:         'constituent',
          spaceName:    c.spaceName,
          spaceUrl:     c.spaceUrl,
          isActive:     data.isActive,
          addingMedia:  !!(data.showplayInProgress),
          nowPlaying:   current ? { title: current.title, thumbnail: current.thumbnail, duration: current.duration, isVideo: current.isVideo } : null,
          queue:        data.queue || [],
          hlsUrl:       data.isActive && data.hlsUrl ? `${c.spaceUrl}${data.hlsUrl}` : null,
          socketUrl:    c.spaceUrl,
          streamId:     userId,
          currentTrackUrl: `${c.spaceUrl}/stream/${userId}/currentTrack`,
          hlsStatusUrl:    `${c.spaceUrl}/stream/${userId}/hlsStatus`,
        });
      } catch { /* unreachable constituent */ }
    }));
  } catch (err) {
    console.error('/api/my-streams constituent check error:', err.message);
  }

  // Pool stream
  const stream = streams[userId];
  if (stream && (stream.isActive || stream._showplayInProgress)) {
    const current = stream.queue[0] || null;
    result.pool = {
      type:        'pool',
      isActive:    stream.isActive,
      addingMedia: !!(stream._showplayInProgress),
      nowPlaying:  current ? { title: current.meta.title, thumbnail: current.meta.thumbnail, duration: current.meta.duration, isVideo: !!current.meta.videoUrl } : null,
      queue:       stream.queue.map(s => ({ title: s.meta.title, thumbnail: s.meta.thumbnail, duration: s.meta.duration, isVideo: !!s.meta.videoUrl })),
      hlsUrl:      stream.isActive ? `/stream-hls/${userId}/live.m3u8` : null,
      streamId:    userId,
      currentTrackUrl: `/stream/${userId}/currentTrack`,
      hlsStatusUrl:    `/stream/${userId}/hlsStatus`,
      pool: {
        slotsUsed: poolActiveSlots, slotsMax: MAIN_POOL_MAX_SLOTS,
        userMBUsed: parseFloat((getUserQueueBytes(userId) / 1024 / 1024).toFixed(1)), userMBMax: 200,
      },
    };
  }

  res.json({ success: true, ...result });
});

// GET /api/my-stream-source
// Returns the resolved HLS URL and source details for the authenticated user's stream.
// Checks constituent servers first, then falls back to main pool.
// Handles: constituent playing, pool playing, no stream active, no HF key.
streamApp.get('/api/my-stream-source', requireRegistered, async (req, res) => {
  const userId = req.user.userId;

  // ── 1. Check if user has a constituent serving their stream ──────────────
  try {
    const constituents = await db.collection('constituents').find({ userId }).toArray();
    for (const c of constituents) {
      try {
        const r = await axios.get(
          `${c.spaceUrl}/constituent/queue/${userId}`,
          { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 8000 }
        );
        const data = r.data;
        if (data.success && (data.isActive && data.hlsUrl)) {
          // This constituent is actively playing
          const current = data.queue?.[0] || null;
          return res.json({
            success:    true,
            source:     'constituent',
            spaceUrl:   c.spaceUrl,
            spaceName:  c.spaceName,
            hlsUrl:     `${c.spaceUrl}${data.hlsUrl}`,
            currentTrackUrl: `${c.spaceUrl}/stream/:streamId/currentTrack`.replace(':streamId', userId),
            hlsStatusUrl:    `${c.spaceUrl}/stream/:streamId/hlsStatus`.replace(':streamId', userId),
            socketUrl:  c.spaceUrl,
            streamId:   userId,
            nowPlaying: current ? { title: current.title, thumbnail: current.thumbnail, duration: current.duration, isVideo: current.isVideo } : null,
            queue:      data.queue || [],
            addingMedia: !!(data.showplayInProgress),
          });
        }
        // Constituent reachable but not yet playing — check if it's adding media
        if (data.success && data.showplayInProgress) {
          return res.json({
            success:      true,
            source:       'constituent',
            spaceUrl:     c.spaceUrl,
            spaceName:    c.spaceName,
            hlsUrl:       null,
            currentTrackUrl: null,
            hlsStatusUrl: null,
            socketUrl:    c.spaceUrl,
            streamId:     userId,
            nowPlaying:   null,
            queue:        [],
            addingMedia:  true,
          });
        }
      } catch {
        // Constituent unreachable — skip and try next
      }
    }
  } catch (err) {
    console.error('/api/my-stream-source constituent check error:', err.message);
  }

  // ── 2. Fall back to main pool stream ──────────────────────────────────────
  const stream  = streams[userId];
  const hasKey  = !!(await db.collection('hf_keys').findOne({ userId }).catch(() => null));
  const poolAddingMedia = !!(stream && stream._showplayInProgress > 0);

  if (!stream || !stream.isActive || stream.queue.length === 0) {
    return res.json({
      success:      true,
      source:       poolAddingMedia ? 'pool' : 'none',
      hlsUrl:       null,
      nowPlaying:   null,
      hasConstituent: false,
      hasHfKey:     hasKey,
      addingMedia:  poolAddingMedia,
      pool: {
        slotsUsed:  poolActiveSlots,
        slotsMax:   MAIN_POOL_MAX_SLOTS,
        userMBUsed: parseFloat((getUserQueueBytes(userId) / 1024 / 1024).toFixed(1)),
        userMBMax:  200,
      },
    });
  }

  const current = stream.queue[0];
  const hlsStateNow = hlsState[userId];
  const hlsReady    = !!(hlsStateNow && hlsStateNow.segments.length > 0 && !hlsStateNow.generating);

  return res.json({
    success:    true,
    source:     'pool',
    hlsUrl:     `/stream-hls/${userId}/live.m3u8`,
    currentTrackUrl: `/stream/${userId}/currentTrack`,
    hlsStatusUrl:    `/stream/${userId}/hlsStatus`,
    socketUrl:  null, // same origin
    streamId:   userId,
    hlsReady,
    nowPlaying: current ? {
      title:     current.meta.title,
      thumbnail: current.meta.thumbnail,
      duration:  current.meta.duration,
      isVideo:   !!current.meta.videoUrl,
    } : null,
    queue: stream.queue.map(s => ({
      title:     s.meta.title,
      thumbnail: s.meta.thumbnail,
      duration:  s.meta.duration,
      isVideo:   !!s.meta.videoUrl,
      hlsReady:  !!(s._hlsPregened && typeof s._hlsStart === 'number'),
    })),
    hasConstituent: false,
    hasHfKey:   hasKey,
    pool: {
      slotsUsed:  poolActiveSlots,
      slotsMax:   MAIN_POOL_MAX_SLOTS,
      userMBUsed: parseFloat((getUserQueueBytes(userId) / 1024 / 1024).toFixed(1)),
      userMBMax:  200,
    },
  });
});

// GET /api/pool-status — global main-server pool info for all users
streamApp.get('/api/pool-status', requireRegistered, (req, res) => {
  const streamId = req.user.userId;
  const poolDiskBytes = getPoolDiskBytes();
  res.json({
    success: true,
    pool: {
      slotsUsed:  poolActiveSlots,
      slotsMax:   MAIN_POOL_MAX_SLOTS,
      slotsFree:  Math.max(0, MAIN_POOL_MAX_SLOTS - poolActiveSlots),
      diskUsedGB: parseFloat((poolDiskBytes / 1024 ** 3).toFixed(2)),
      diskCapGB:  parseFloat((MAIN_POOL_DISK_CAP_BYTES / 1024 ** 3).toFixed(0)),
      diskFreeGB: parseFloat(((MAIN_POOL_DISK_CAP_BYTES - poolDiskBytes) / 1024 ** 3).toFixed(2)),
      diskFull:   poolDiskBytes >= MAIN_POOL_DISK_CAP_BYTES,
    },
    user: {
      userMBUsed: parseFloat((getUserQueueBytes(streamId) / 1024 / 1024).toFixed(1)),
      userMBMax:  200,
      userMBFree: parseFloat(((MAIN_POOL_MAX_SIZE_BYTES - getUserQueueBytes(streamId)) / 1024 / 1024).toFixed(1)),
    },
  });
});

// ═══════════════════════════════════════════════════════════════════════════
// API ROUTES — search / details / play / vplay / showplay
// All playback routes require auth; userId from JWT is the streamId
// ═══════════════════════════════════════════════════════════════════════════

// GET /api/search?query=<show name>
// Returns show/movie results. Each result's `id` (= link URL) is used by /api/details.
streamApp.get('/api/search', async (req, res) => {
  const { query } = req.query;
  if (!query) return res.status(400).json({ success: false, error: 'query is required' });
  try {
    const r       = await axios.get(`https://iktracks.vercel.app/search?query=${encodeURIComponent(query)}`, { timeout: 15000 });
    const results = (r.data?.results || []).filter(r => r && r.link).map(r => ({
      id:        r.link,
      title:     r.title,
      type:      r.type,        // 'movie' | 'series'
      thumbnail: r.thumbnail || DEFAULT_ARTWORK,
      year:      r.year || null,
    }));
    res.json({ success: true, results });
  } catch (err) {
    console.error('/api/search error:', err.message);
    res.status(500).json({ success: false, error: 'Search failed. Please try again.' });
  }
});

// GET /api/details?id=<show link URL>
// For movies: returns downloadLinks[]. For series: returns seasons[].episodes[].
streamApp.get('/api/details', async (req, res) => {
  const { id } = req.query;
  if (!id) return res.status(400).json({ success: false, error: 'id (show link) is required' });
  try {
    const r       = await axios.get(`https://iktracks.vercel.app/details?url=${encodeURIComponent(id)}`, { timeout: 15000 });
    const details = r.data;
    if (!details) return res.status(404).json({ success: false, error: 'No details returned' });
    res.json({ success: true, details });
  } catch (err) {
    console.error('/api/details error:', err.message);
    res.status(500).json({ success: false, error: 'Failed to fetch details. Please try again.' });
  }
});

// POST /api/play  — body: { songName }
// Requires auth cookie. User's own userId is the streamId.
streamApp.post('/api/play', requireRegistered, async (req, res) => {
  const { songName } = req.body;
  if (!songName) return res.status(400).json({ success: false, error: 'songName is required' });
  const streamId = req.user.userId;

  let filePath;
  try {
    const apiRes   = await axios.get(`https://iktracks.vercel.app/play?query=${encodeURIComponent(songName)}`, { timeout: 15000 });
    const songData = apiRes.data?.result;
    if (!songData?.download_url) return res.status(404).json({ success: false, error: 'No audio result found for that song name' });

    const fileName = crypto.randomUUID() + '.mp3';
    filePath       = path.join(SONGS_DIR, fileName);
    const writer   = fs.createWriteStream(filePath);
    const response = await axios({ url: songData.download_url, method: 'GET', responseType: 'stream' });
    response.data.pipe(writer);
    await new Promise((resolve, reject) => { writer.on('finish', resolve); writer.on('error', (e) => { writer.destroy(); reject(e); }); });

    const audioMeta = await getAudioMeta(filePath);
    if (audioMeta.size > MAX_FILE_SIZE)    { fs.unlinkSync(filePath); return res.status(400).json({ success: false, error: 'File too large (max 50 MB)' }); }
    if (audioMeta.duration > MAX_DURATION) { fs.unlinkSync(filePath); return res.status(400).json({ success: false, error: 'Song too long (max 15 min)' }); }

    const songInfo = { fileName, meta: { title: songData.title, thumbnail: songData.thumbnail || DEFAULT_ARTWORK, duration: audioMeta.duration, views: songData.views, published: songData.published, source: songData.video_url } };
    const { position, started } = enqueueToStream(streamId, songInfo, streamId);
    setStreamLocation(streamId, 'pool').catch(console.error);

    res.json({ success: true, started, position, title: songData.title, duration: audioMeta.duration, thumbnail: songData.thumbnail || DEFAULT_ARTWORK, hlsUrl: `/stream-hls/${streamId}/live.m3u8` });
  } catch (err) {
    console.error('/api/play error:', err.message);
    if (filePath && fs.existsSync(filePath)) { try { fs.unlinkSync(filePath); } catch {} }
    res.status(500).json({ success: false, error: 'Failed to add song. Please try again.' });
  }
});

// POST /api/vplay  — body: { songName }
// Requires auth cookie. Searches YouTube, downloads ≤480p video (max 15 min).
streamApp.post('/api/vplay', requireRegistered, async (req, res) => {
  const { songName } = req.body;
  if (!songName) return res.status(400).json({ success: false, error: 'songName is required' });
  const streamId = req.user.userId;

  let filePath;
  try {
    const videoData = await fetchVideoData(songName);
    if (!videoData.videoDownloadUrl) return res.status(404).json({ success: false, error: 'No video format found for that query' });

    let dlResult;
    try { dlResult = await downloadVideoFile(videoData.videoDownloadUrl); }
    catch (err) { return res.status(500).json({ success: false, error: `Download failed: ${err.message}` }); }
    filePath = dlResult.filePath;

    let mediaMeta;
    try { mediaMeta = await getAudioMeta(filePath); }
    catch (e) { fs.unlinkSync(filePath); return res.status(500).json({ success: false, error: 'Failed to process video file' }); }

    if (mediaMeta.duration > MAX_DURATION) {
      fs.unlinkSync(filePath);
      return res.status(400).json({ success: false, error: 'Video too long (max 15 minutes)' });
    }

    const songInfo = { fileName: dlResult.fileName, meta: { title: videoData.title, thumbnail: videoData.thumbnail, duration: mediaMeta.duration, views: videoData.views, published: videoData.published, source: videoData.videoUrl, videoUrl: videoData.videoUrl } };
    const { position, started } = enqueueToStream(streamId, songInfo, streamId);
    setStreamLocation(streamId, 'pool').catch(console.error);

    res.json({ success: true, started, position, title: videoData.title, duration: mediaMeta.duration, thumbnail: videoData.thumbnail, hlsUrl: `/stream-hls/${streamId}/live.m3u8` });
  } catch (err) {
    console.error('/api/vplay error:', err.message);
    if (filePath && fs.existsSync(filePath)) { try { fs.unlinkSync(filePath); } catch {} }
    res.status(500).json({ success: false, error: 'Failed to add video. Please try again.' });
  }
});

// ─── Showplay helpers (shared logic, identical to original bot) ────────────

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 spSeriesName(title) { if (!title) return 'Episode'; return title.replace(/\s*\|.*$/, '').replace(/\s*\(.*$/, '').trim() || title; }

// Extract all episodes from details (same logic as bot sppl handler)
function extractAllEpisodes(details) {
  if (!Array.isArray(details.seasons) || !details.seasons.length) return [];
  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;
}

// ═══════════════════════════════════════════════════════════════════════════
// TMDB HELPERS
// ═══════════════════════════════════════════════════════════════════════════

// Search TMDB for a movie or TV show by name, return enriched metadata.
// Falls back gracefully if TMDB_KEY is not set or the request fails.
async function fetchTmdbInfo(title, type = 'multi') {
  if (!TMDB_KEY) return null;
  try {
    const endpoint = type === 'movie'  ? `${TMDB_BASE}/search/movie`
                   : type === 'tv'     ? `${TMDB_BASE}/search/tv`
                   :                    `${TMDB_BASE}/search/multi`;

    const searchRes = await axios.get(endpoint, {
      params: { api_key: TMDB_KEY, query: title, language: 'en-US', page: 1 },
      timeout: 8000,
    });

    const results = searchRes.data?.results;
    if (!results?.length) return null;
    const hit = results[0];

    // Fetch full details for the top result
    const mediaType = hit.media_type || (hit.first_air_date ? 'tv' : 'movie');
    const detailsRes = await axios.get(`${TMDB_BASE}/${mediaType}/${hit.id}`, {
      params: { api_key: TMDB_KEY, language: 'en-US', append_to_response: 'credits,videos' },
      timeout: 8000,
    });
    const d = detailsRes.data;

    const poster   = d.poster_path   ? `${TMDB_IMG}${d.poster_path}`   : null;
    const backdrop = d.backdrop_path ? `https://image.tmdb.org/t/p/w1280${d.backdrop_path}` : null;
    const trailer  = d.videos?.results?.find(v => v.type === 'Trailer' && v.site === 'YouTube');

    return {
      tmdbId:       d.id,
      tmdbType:     mediaType,
      title:        d.title || d.name || title,
      originalTitle: d.original_title || d.original_name || null,
      overview:     d.overview      || null,
      tagline:      d.tagline       || null,
      releaseDate:  d.release_date  || d.first_air_date || null,
      runtime:      d.runtime       || (d.episode_run_time?.[0]) || null,
      genres:       (d.genres || []).map(g => g.name),
      voteAverage:  d.vote_average  || null,
      voteCount:    d.vote_count    || null,
      popularity:   d.popularity    || null,
      poster,
      backdrop,
      trailerKey:   trailer?.key    || null,
      cast:         (d.credits?.cast || []).slice(0, 10).map(c => ({ name: c.name, character: c.character, photo: c.profile_path ? `${TMDB_IMG}${c.profile_path}` : null })),
      director:     (d.credits?.crew || []).find(c => c.job === 'Director')?.name || null,
      status:       d.status        || null,
      language:     d.original_language || null,
      tmdbUrl:      `https://www.themoviedb.org/${mediaType}/${d.id}`,
    };
  } catch (err) {
    console.warn(`TMDB fetch failed for "${title}":`, err.message);
    return null;
  }
}

// ── Stream queue persistence helpers ─────────────────────────────────────────

async function saveQueueItem(streamId, songInfo) {
  try {
    await db.collection('stream_queue').updateOne(
      { _sid: songInfo._sid },
      { $set: {
          _sid:      songInfo._sid,
          streamId,
          fileName:  songInfo.fileName,
          meta:      songInfo.meta,
          tmdb:      songInfo.tmdb || null,
          addedAt:   new Date(),
        }
      },
      { upsert: true }
    );
  } catch (err) { console.error('saveQueueItem error:', err.message); }
}

async function removeQueueItem(sid) {
  if (!sid) return;
  try { await db.collection('stream_queue').deleteOne({ _sid: sid }); }
  catch (err) { console.error('removeQueueItem error:', err.message); }
}

// getPersistedQueue — reads DB queue for a stream (useful for admin/debug)
async function getPersistedQueue(streamId) {
  try {
    return await db.collection('stream_queue')
      .find({ streamId })
      .sort({ addedAt: 1 })
      .toArray();
  } catch (err) { console.error('getQueueFromDB error:', err.message); return []; }
}

// Probe the Content-Length of a URL without downloading the file body.
// Returns the file size in bytes, or 0 if the server doesn't expose it.
async function probeContentLength(url) {
  try {
    const res = await axios.head(url, { timeout: 10000, httpsAgent: httpsAgentNoVerify, maxRedirects: 5 });
    const cl = parseInt(res.headers['content-length'] || '0', 10);
    return isFinite(cl) && cl > 0 ? cl : 0;
  } catch { return 0; }
}

// Core download-and-enqueue for showplay (same as original sppl handler)
// fromConstituent = true skips pool checks (constituents have no limits on main server)
async function showplayEnqueueLink(streamId, pendingLink, pendingTitle, thumbnail, tmdbInfo = null, fromConstituent = false) {
  if (!streams[streamId]) {
    streams[streamId] = { queue: [], songStartTime: null, streamTimeOffset: 0, users: new Map(), ownerId: streamId, lastActivity: Date.now(), isActive: false, isGroup: false };
  }
  streams[streamId]._showplayInProgress = (streams[streamId]._showplayInProgress || 0) + 1;
  // Keep the stream warm so the inactivity cleanup never evicts it mid-download
  streams[streamId].lastActivity = Date.now();
  if (!fromConstituent) poolActiveSlots++;

  // Notify listeners that download is starting
  io.to(`stream:${streamId}`).emit('message', { type: 'showplay_progress', stage: 'downloading', title: 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');
  }

  // ── Pre-download disk cap check with Content-Length ───────────────────────
  // Now that we have a direct URL, probe its size and recheck the global disk
  // cap before committing to a potentially large download.
  if (!fromConstituent) {
    const expectedBytes = await probeContentLength(directUrl);
    if (expectedBytes > 0) {
      const diskCheck = checkPoolAvailability(streamId, expectedBytes);
      if (!diskCheck.allowed) {
        // Roll back the slot we incremented above before throwing
        if (streams[streamId]) streams[streamId]._showplayInProgress = Math.max(0, (streams[streamId]._showplayInProgress || 1) - 1);
        poolActiveSlots = Math.max(0, poolActiveSlots - 1);
        const err = new Error(diskCheck.reason);
        err.diskFull   = diskCheck.diskFull || false;
        err.diskUsedGB = diskCheck.diskUsedGB;
        err.diskCapGB  = diskCheck.diskCapGB;
        throw err;
      }
    }
  }

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

  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 is 2 GB.`);
  }
  if (mediaMeta.duration > SHOWPLAY_MAX_DURATION) {
    try { fs.unlinkSync(filePath); } catch {}
    throw new Error(`Video too long (${Math.round(mediaMeta.duration / 60)} min). Max is ${SHOWPLAY_MAX_DURATION / 3600} hours.`);
  }

  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, streamId);
  // Persist to DB after _sid is assigned by enqueueToStream
  saveQueueItem(streamId, songInfo).catch(console.error);
  // Record that this stream is on the main pool (not a constituent)
  if (!fromConstituent) setStreamLocation(streamId, 'pool').catch(console.error);

  // Decrement AFTER enqueue so the inactivity cleanup never sees a momentary
  // zero-count window between "file ready" and "item in queue".
  if (streams[streamId]) streams[streamId]._showplayInProgress = Math.max(0, (streams[streamId]._showplayInProgress || 1) - 1);
  if (!fromConstituent) poolActiveSlots = Math.max(0, poolActiveSlots - 1);

  return {
    title:         songInfo.meta.title,
    duration:      mediaMeta.duration,
    thumbnail:     effectivePoster,
    queuePosition: streams[streamId]?.queue.length || 1,
    tmdb:          tmdbInfo || null,
  };
}

// POST /api/showplay/movie  — body: { showId }
// showId is the `id` (link) from /api/search. Automatically picks first downloadLink (movie).
streamApp.post('/api/showplay/movie', requireRegistered, async (req, res) => {
  const { showId } = req.body;
  if (!showId) return res.status(400).json({ success: false, error: 'showId is required' });
  const streamId = req.user.userId;

  // ── Pool availability check (slots + global disk cap + per-user quota) ─────
  const poolCheck = checkPoolAvailability(streamId);
  if (!poolCheck.allowed) {
    return res.status(429).json({
      success:    false,
      error:      poolCheck.reason,
      poolFull:   true,
      diskFull:   poolCheck.diskFull   || false,
      diskUsedGB: poolCheck.diskUsedGB || null,
      diskCapGB:  poolCheck.diskCapGB  || null,
      poolSlots:  { used: poolActiveSlots, max: MAIN_POOL_MAX_SLOTS },
      userBytes:  { usedMB: (getUserQueueBytes(streamId) / 1024 / 1024).toFixed(1), maxMB: 200 },
    });
  }

  try {
    const r       = await axios.get(`https://iktracks.vercel.app/details?url=${encodeURIComponent(showId)}`, { timeout: 15000 });
    const details = r.data;
    if (!details) return res.status(404).json({ success: false, error: 'No details returned for this show' });

    if (details.type === 'series') return res.status(400).json({ success: false, error: 'This is a series. Use /api/showplay/episode instead.' });

    const link = details.downloadLinks?.[0]?.downloadLink;
    if (!link) return res.status(404).json({ success: false, error: 'No download link found for this movie' });

    const thumbnail = details.thumbnail || DEFAULT_ARTWORK;
    const title     = details.title || 'Movie';

    // Fetch TMDB enrichment (non-blocking on failure)
    const tmdbInfo = await fetchTmdbInfo(title, 'movie');

    const result = await showplayEnqueueLink(streamId, link, title, thumbnail, tmdbInfo);
    res.json({ success: true, ...result, hlsUrl: `/stream-hls/${streamId}/live.m3u8` });
  } catch (err) {
    if (streams[streamId]) streams[streamId]._showplayInProgress = Math.max(0, (streams[streamId]._showplayInProgress || 1) - 1);
    console.error('/api/showplay/movie error:', err.message);
    if (err.diskFull) {
      return res.status(507).json({ success: false, error: err.message, diskFull: true, diskUsedGB: err.diskUsedGB, diskCapGB: err.diskCapGB });
    }
    res.status(500).json({ success: false, error: err.message || 'Failed to queue movie. Please try again.' });
  }
});

// POST /api/showplay/episode  — body: { showId, season, episode }
// season and episode are numbers (e.g. season: 1, episode: 3).
streamApp.post('/api/showplay/episode', requireRegistered, async (req, res) => {
  const { showId, season, episode } = req.body;
  if (!showId)  return res.status(400).json({ success: false, error: 'showId is required' });
  if (season  == null) return res.status(400).json({ success: false, error: 'season is required' });
  if (episode == null) return res.status(400).json({ success: false, error: 'episode is required' });
  const streamId = req.user.userId;

  // ── Pool availability check (slots + global disk cap + per-user quota) ─────
  const poolCheck = checkPoolAvailability(streamId);
  if (!poolCheck.allowed) {
    return res.status(429).json({
      success:    false,
      error:      poolCheck.reason,
      poolFull:   true,
      diskFull:   poolCheck.diskFull   || false,
      diskUsedGB: poolCheck.diskUsedGB || null,
      diskCapGB:  poolCheck.diskCapGB  || null,
      poolSlots:  { used: poolActiveSlots, max: MAIN_POOL_MAX_SLOTS },
      userBytes:  { usedMB: (getUserQueueBytes(streamId) / 1024 / 1024).toFixed(1), maxMB: 200 },
    });
  }

  try {
    const r       = await axios.get(`https://iktracks.vercel.app/details?url=${encodeURIComponent(showId)}`, { timeout: 15000 });
    const details = r.data;
    if (!details) return res.status(404).json({ success: false, error: 'No details returned for this show' });

    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 || 'Series');
    const epLabel     = `S${String(ep.season).padStart(2,'0')} E${String(ep.episode).padStart(2,'0')}`;
    const pendingTitle = `${seriesName} • ${epLabel}`;
    const thumbnail   = details.thumbnail || DEFAULT_ARTWORK;

    // Fetch TMDB enrichment using the series name (non-blocking on failure)
    const tmdbInfo = await fetchTmdbInfo(spSeriesName(details.title || seriesName), 'tv');

    const result = await showplayEnqueueLink(streamId, ep.downloadLink, pendingTitle, thumbnail, tmdbInfo);
    res.json({ success: true, ...result, hlsUrl: `/stream-hls/${streamId}/live.m3u8` });
  } catch (err) {
    if (streams[streamId]) streams[streamId]._showplayInProgress = Math.max(0, (streams[streamId]._showplayInProgress || 1) - 1);
    console.error('/api/showplay/episode error:', err.message);
    if (err.diskFull) {
      return res.status(507).json({ success: false, error: err.message, diskFull: true, diskUsedGB: err.diskUsedGB, diskCapGB: err.diskCapGB });
    }
    res.status(500).json({ success: false, error: err.message || 'Failed to queue episode. Please try again.' });
  }
});

// GET /api/showplay/episodes?showId=<id>  — list all available episodes for a series
streamApp.get('/api/showplay/episodes', async (req, res) => {
  const { showId } = req.query;
  if (!showId) return res.status(400).json({ success: false, error: 'showId is required' });
  try {
    const r       = await axios.get(`https://iktracks.vercel.app/details?url=${encodeURIComponent(showId)}`, { timeout: 15000 });
    const details = r.data;
    if (!details) return res.status(404).json({ success: false, error: 'No details returned' });

    const allEps = extractAllEpisodes(details);
    res.json({ success: true, title: details.title, type: details.type, thumbnail: details.thumbnail || DEFAULT_ARTWORK, multiSeason: (details.seasons?.length || 0) > 1, episodes: allEps });
  } catch (err) {
    console.error('/api/showplay/episodes error:', err.message);
    res.status(500).json({ success: false, error: 'Failed to fetch episodes. Please try again.' });
  }
});

// POST /api/stop  — stop the stream entirely (clears queue, kills FFmpeg)
streamApp.post('/api/stop', requireRegistered, (req, res) => {
  const streamId = req.user.userId;
  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 {} }
    if (song._sid) removeQueueItem(song._sid).catch(console.error);
  }
  stream.queue           = [];
  stream.isActive        = false;
  stream.songStartTime   = null;
  stream.streamTimeOffset = 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;
  db.collection('stream_queue').deleteMany({ streamId }).catch(console.error);
  io.to(`stream:${streamId}`).emit('message', { type: 'stream_ended', message: 'Stream stopped by owner.' });
  res.json({ success: true, message: 'Stream stopped.' });
});

// POST /api/constituent/stop/:spaceName — stop a constituent's stream
streamApp.post('/api/constituent/stop/:spaceName', requireRegistered, async (req, res) => {
  const { spaceName } = req.params;
  const userId = req.user.userId;
  try {
    const constituent = await db.collection('constituents').findOne({ userId, spaceName });
    if (!constituent) return res.status(404).json({ success: false, error: 'Constituent not found' });
    await axios.post(
      `${constituent.spaceUrl}/constituent/stop/${userId}`,
      {},
      { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 10000 }
    );
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.response?.data?.error || err.message });
  }
});

// Returns how many HLS segments have been encoded for a given song (_sid) in a stream.
function countSegmentsForSong(streamId, sid) {
  const state = hlsState[streamId];
  if (!state || !sid) return 0;
  return state.segments.filter(s => s.ownerSid === sid).length;
}

// Minimum segments that must be ready before next/skip is allowed
const MIN_SEGMENTS_TO_SKIP = 3;

// POST /api/skip  — skip current track (auth required, must be stream owner)
streamApp.post('/api/skip', requireRegistered, (req, res) => {
  const streamId = req.user.userId;
  const stream   = streams[streamId];
  if (!stream) return res.status(404).json({ success: false, error: 'No active stream' });
  if (stream.queue.length <= 1) return res.status(400).json({ success: false, error: 'No more songs in queue' });

  const upcoming = stream.queue[1];
  const segsReady = countSegmentsForSong(streamId, upcoming?._sid);
  const isReady   = (upcoming?._hlsPregened &&
    typeof upcoming._hlsStart === 'number' &&
    typeof upcoming._hlsEnd   === 'number' &&
    upcoming._hlsEnd > upcoming._hlsStart) ||
    segsReady >= MIN_SEGMENTS_TO_SKIP;

  if (!isReady && upcoming) {
    return res.status(409).json({
      success:       false,
      error:         `Next track is buffering (${segsReady}/${MIN_SEGMENTS_TO_SKIP} segments ready). Try again shortly.`,
      nextTitle:     upcoming.meta.title,
      segmentsReady: segsReady,
      segmentsNeeded: MIN_SEGMENTS_TO_SKIP,
    });
  }

  if (stream.users) { for (const [uid] of stream.users) { trackSongListened(uid).catch(console.error); } }
  if (!stream.users || !stream.users.has(streamId)) { trackSongListened(streamId).catch(console.error); }
  delete stream._advancingFromSid;
  advanceToNextSong(streamId);
  const nowPlaying = stream.queue[0]?.meta?.title || '(queue empty)';
  res.json({ success: true, nowPlaying });
});

// GET /api/queue/:streamId  — full enriched queue for a stream (resolves mirrors)
streamApp.get('/api/queue/:streamId', async (req, res) => {
  const streamId = await resolveStreamId(req.params.streamId);

  const proxied = await proxyToConstituentIfNeeded(streamId, res, async (spaceUrl) => {
    const r = await axios.get(`${spaceUrl}/api/queue/${streamId}`,
      { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 8000 });
    res.json(r.data);
  });
  if (proxied) return;

  const stream   = streams[streamId];
  if (!stream) return res.status(404).json({ success: false, error: 'Stream not found' });

  const current = stream.queue[0] || null;
  const queue   = stream.queue.map((s, idx) => ({
    position:   idx,
    _sid:       s._sid,
    title:      s.meta.title,
    thumbnail:  s.meta.thumbnail,
    duration:   s.meta.duration,
    isVideo:    !!s.meta.videoUrl,
    isShowplay: s.isShowplay || false,
    tmdb:       s.tmdb || null,
    hlsReady:   !!(s._hlsPregened && typeof s._hlsStart === 'number'),
  }));

  const hlsStateNow = hlsState[streamId];
  const withinSong  = stream.songStartTime ? (Date.now() - stream.songStartTime) / 1000 : 0;

  res.json({
    success: true,
    streamId,
    isActive:    stream.isActive,
    totalQueued: queue.length,
    nowPlaying:  queue[0]   || null,
    upNext:      queue[1]   || null,
    queue,
    withinSong:  Math.max(0, withinSong),
    hlsUrl:      stream.isActive ? `/stream-hls/${streamId}/live.m3u8` : null,
  });
});

// POST /api/internal/track-song — called by constituent servers when a song auto-advances.
// Increments the stream owner's songCount stat. Authenticated with the shared secret.
streamApp.post('/api/internal/track-song', async (req, res) => {
  const secret = req.headers['x-constituent-secret'];
  if (!secret || secret !== MAIN_SERVER_SECRET) {
    return res.status(403).json({ success: false, error: 'Forbidden' });
  }
  const { userId } = req.body;
  if (!userId) return res.status(400).json({ success: false, error: 'userId is required' });
  try {
    await trackSongListened(userId);
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/internal/track-listening
// Called by constituent servers when a viewer fires /left so the main server
// can persist the listening session duration to the user's stats.
streamApp.post('/api/internal/track-listening', async (req, res) => {
  const secret = req.headers['x-constituent-secret'];
  if (!secret || secret !== MAIN_SERVER_SECRET) {
    return res.status(403).json({ success: false, error: 'Forbidden' });
  }
  const { userId, duration } = req.body;
  if (!userId || duration == null) return res.status(400).json({ success: false, error: 'userId and duration are required' });
  try {
    await trackListeningSession(userId, duration);
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/internal/track-heartbeat
// Called by constituent servers on each viewer heartbeat so the main server
// can accumulate listening time incrementally via trackListeningSession.
// The constituent passes the elapsed seconds since the last heartbeat.
streamApp.post('/api/internal/track-heartbeat', async (req, res) => {
  const secret = req.headers['x-constituent-secret'];
  if (!secret || secret !== MAIN_SERVER_SECRET) {
    return res.status(403).json({ success: false, error: 'Forbidden' });
  }
  const { userId, streamId } = req.body;
  if (!userId) return res.status(400).json({ success: false, error: 'userId is required' });

  const isGuest = (userId || '').startsWith('g_');
  if (isGuest) return res.json({ success: true }); // no stats for guests

  try {
    // We use a small in-memory tracker keyed by userId to measure the exact
    // delta since the constituent last pinged us, then record it as a session.
    if (!constituentHeartbeatTs) constituentHeartbeatTs = {};
    const now  = Date.now();
    const prev = constituentHeartbeatTs[userId] || now;
    const elapsed = Math.max(0, (now - prev) / 1000);
    constituentHeartbeatTs[userId] = now;

    if (elapsed > 0 && elapsed < 120) {
      await trackListeningSession(userId, elapsed);
    }
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// In-memory map to track the last heartbeat timestamp per userId coming from constituents
const constituentHeartbeatTs = {};

// GET /api/profile  — listening stats for the authenticated user
streamApp.get('/api/profile', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const stats = await getUserStats(userId);
    const stream = streams[userId];
    const current = stream?.queue[0] || null;
    const sessionDuration = current && stream.users?.has(userId)
      ? Math.floor((Date.now() - (stream.users.get(userId)?.connectionStart || Date.now())) / 1000)
      : 0;
    res.json({
      success: true,
      userId,
      username:    req.user.username,
      displayName: req.user.displayName,
      stats,
      currentlyPlaying: current ? { title: current.meta.title, isVideo: !!current.meta.videoUrl, sessionDuration } : null,
    });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// ═══════════════════════════════════════════════════════════════════════════
// SOCKET.IO — real-time updates
// ═══════════════════════════════════════════════════════════════════════════

function broadcastUserUpdate(streamId, type, userData, debounce = false) {
  const stream = streams[streamId];
  if (!stream) return;
  const payload = { type, data: userData, timestamp: new Date().toISOString() };
  const doSend  = () => io.to(`stream:${streamId}`).emit('message', payload);
  if (debounce) {
    if (stream.userUpdateTimeout) clearTimeout(stream.userUpdateTimeout);
    stream.userUpdateTimeout = setTimeout(doSend, 500);
  } else { doSend(); }
}

io.on('connection', async (socket) => {
  const rawId = socket.handshake.query.streamId;
  // Room-only connections (no streamId query param) are handled by room:join event below.
  // Only process stream socket logic when streamId is provided.
  if (!rawId) return;
  const streamId = await resolveStreamId(rawId);

  // Lazily initialise the stream entry so that heartbeat / joined / list
  // calls never 404 just because no media has been added yet.
  if (!streams[streamId]) {
    streams[streamId] = {
      queue: [], songStartTime: null, streamTimeOffset: 0,
      users: new Map(), ownerId: streamId,
      lastActivity: Date.now(), isActive: false, isGroup: false,
    };
  }

  const stream = streams[streamId];
  socket.join(`stream:${streamId}`);
  stream.lastActivity = Date.now();

  // Only send an update if there is actually something playing.
  if (stream.isActive || stream.queue.length > 0) {
    sendStreamUpdate(streamId, socket);
  }

  // If media is being processed, notify the new socket so the UI can show a
  // loading state rather than a blank screen.
  if (stream._showplayInProgress) {
    socket.emit('message', { type: 'showplay_progress', stage: 'processing', title: stream.queue[0]?.meta?.title || 'media' });
  }

  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) return;
  if (!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, isShowplay: current.isShowplay || false },
    songId: current._sid,
    next: nextSong ? { file: `/songs/${nextSong.fileName}`, meta: nextSong.meta, isVideo: !!nextSong.meta.videoUrl, tmdb: nextSong.tmdb || null, isShowplay: nextSong.isShowplay || false } : 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);
}

// ═══════════════════════════════════════════════════════════════════════════
// HELPERS
// ═══════════════════════════════════════════════════════════════════════════

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

function fmtDur(secs) {
  const m = Math.floor(secs / 60), s = Math.floor(secs % 60);
  return `${m}:${String(s).padStart(2, '0')}`;
}

async function extractAudioMetadata(filePath, originalName) {
  try {
    const audioMeta = await getAudioMeta(filePath);
    const title     = originalName.replace(/\.(mp3|m4a|wav|ogg|flac|mp4|webm)$/i, '');
    return { title, thumbnail: DEFAULT_ARTWORK, duration: audioMeta.duration, views: 'N/A', published: 'Uploaded', source: 'File Upload' };
  } catch (err) { throw new Error('Failed to extract audio metadata'); }
}

async function fetchVideoData(query) {
  const searchRes = await axios.get(`https://apis.prexzyvilla.site/search/youtube?q=${encodeURIComponent(query)}`, { timeout: 15000 });
  const results   = searchRes.data.data || [];
  if (!results.length) throw new Error('No search results found');
  const first    = results[0];
  const videoUrl = first.link;
  const dlRes    = await axios.get(`https://apis.prexzyvilla.site/download/ytdl?url=${encodeURIComponent(videoUrl)}`, { timeout: 30000 });
  const formats  = dlRes.data.formats || [];
  const info     = dlRes.data.info    || {};
  const video    = formats.find(f => f.type === 'video' && f.quality === '480p') ||
                   formats.find(f => f.type === 'video' && f.quality === '360p') ||
                   formats.find(f => f.type === 'video');
  let durationSecs = 0;
  if (typeof info.duration === 'number') durationSecs = info.duration;
  else if (typeof info.duration === 'string' && info.duration.includes(':')) {
    const p = info.duration.split(':').map(Number);
    durationSecs = p.length === 2 ? p[0] * 60 + p[1] : p[0] * 3600 + p[1] * 60 + p[2];
  } else if (first.duration) {
    const p = first.duration.split(':').map(Number);
    durationSecs = p.length === 2 ? p[0] * 60 + p[1] : p[0] * 3600 + p[1] * 60 + p[2];
  }
  return { title: info.title || first.title, thumbnail: info.thumbnail || first.imageUrl || DEFAULT_ARTWORK, duration: durationSecs, videoDownloadUrl: video?.url, videoUrl, views: info.view_count || 'N/A', published: info.upload_date || 'N/A' };
}

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

// ═══════════════════════════════════════════════════════════════════════════
// enqueueToStream — main entry point for adding a song
// ═══════════════════════════════════════════════════════════════════════════

function enqueueToStream(streamId, songInfo, ownerId = null) {
  if (!streams[streamId]) {
    songInfo._sid = crypto.randomUUID();
    streams[streamId] = { queue: [songInfo], songStartTime: null, streamTimeOffset: 0, users: new Map(), ownerId, lastActivity: Date.now(), isActive: true, isGroup: false };
    appendSongToHls(streamId, songInfo).then(() => sendStreamUpdate(streamId)).catch(console.error);
    return { songInfo, position: 1, started: true };
  }
  const stream = streams[streamId];
  songInfo._sid = crypto.randomUUID();

  // ── Replace-not-queue logic ───────────────────────────────────────────────
  // Non-showplay items (short audio/video via /api/play or /api/vplay) replace
  // whatever is currently playing so the user gets instant feedback.
  // Showplay items (movies/episodes) are ALWAYS appended to the queue — they
  // must never evict content that took minutes to download and encode.
  if (!songInfo.isShowplay && stream.isActive && stream.queue.length > 0) {
    killActiveFFmpeg(streamId);
    for (const old of stream.queue) {
      const fp = path.join(SONGS_DIR, old.fileName);
      if (fs.existsSync(fp)) { try { fs.unlinkSync(fp); } catch {} }
      if (old._sid) removeQueueItem(old._sid).catch(console.error);
    }
    stream.queue = [];
    stream.songStartTime = null;
    stream.streamTimeOffset = 0;
    // Reset HLS state for fresh start
    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.queue.push(songInfo);
    stream.lastActivity = Date.now();
    stream.isActive = true;
    appendSongToHls(streamId, songInfo).then(() => sendStreamUpdate(streamId)).catch(console.error);
    // Notify listeners that the track changed
    io.to(`stream:${streamId}`).emit('message', { type: 'track_replaced', title: songInfo.meta.title });
    return { songInfo, position: 1, started: true };
  }

  stream.queue.push(songInfo);
  stream.lastActivity = Date.now();
  const position = stream.queue.length;

  // Start encoding immediately if the stream is idle.
  // Also handles the showplay case: stream.isActive may be false (no prior content)
  // or the queue was empty (prior content finished) — either way start fresh.
  if (!stream.isActive && position === 1) {
    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 };
}

// ═══════════════════════════════════════════════════════════════════════════
// HF API HELPERS
// ═══════════════════════════════════════════════════════════════════════════

async function validateHfToken(token) {
  try {
    const res = await axios.get(`${HF_API}/whoami-v2`, {
      headers: { Authorization: `Bearer ${token}` },
      timeout: 10000,
    });
    return { valid: true, username: res.data.name };
  } catch (err) {
    return { valid: false, error: err.response?.data?.error || err.message };
  }
}

async function createHfSpace(token, hfUsername, spaceName) {
  const res = await axios.post(`${HF_API}/repos/create`, {
    type: 'space', name: spaceName, sdk: 'docker', private: false,
  }, {
    headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' },
    timeout: 20000,
  });
  // Response: { id: "username/spaceName", author: "username", ... }
  // Resolve the real hfUsername from the response if we didn't have it
  const resolvedUsername = res.data?.author || res.data?.id?.split('/')?.[0] || hfUsername;
  return { ...res.data, resolvedUsername };
}

// Fetch the current HEAD sha — required as parentCommit by the HF commit API.
async function getHfSpaceLatestCommitOid(token, hfUsername, spaceName) {
  const url = `https://huggingface.co/api/spaces/${hfUsername}/${spaceName}`;
  console.log(`[HF] GET ${url}`);
  try {
    const res = await axios.get(url, {
      headers: { Authorization: `Bearer ${token}` },
      timeout: 15000,
    });
    console.log(`[HF] space info status=${res.status} sha=${res.data?.sha}`);
    const oid = res.data?.sha || null;
    if (!oid) throw new Error(`No sha in space info response: ${JSON.stringify(res.data)}`);
    return oid;
  } catch (err) {
    const detail = err.response
      ? `status=${err.response.status} body=${JSON.stringify(err.response.data)}`
      : err.message;
    console.error(`[HF] getHfSpaceLatestCommitOid FAILED — ${detail}`);
    throw err;
  }
}

// Upload files to a HF Space in one atomic commit.
// files: Array of { path: string, content: Buffer|string }
async function uploadAllFilesToHfSpace(token, hfUsername, spaceName, files) {
  const parentCommit = await getHfSpaceLatestCommitOid(token, hfUsername, spaceName);

  const fileEntries = files.map(f => {
    if (Buffer.isBuffer(f.content)) {
      return { path: f.path, content: f.content.toString('base64'), encoding: 'base64' };
    }
    return { path: f.path, content: f.content.toString(), encoding: 'utf-8' };
  });

  const payload = {
    summary:      'Upload constituent server files',
    description:  '',
    parentCommit,
    files: fileEntries,
  };

  const url = `https://huggingface.co/api/spaces/${hfUsername}/${spaceName}/commit/main`;
  console.log(`[HF] POST ${url}  parentCommit=${parentCommit}  files=[${files.map(f => f.path).join(', ')}]`);

  try {
    const res = await axios.post(url, payload, {
      headers: {
        Authorization:  `Bearer ${token}`,
        'Content-Type': 'application/json',
      },
      timeout: 120000,
      maxBodyLength:    Infinity,
      maxContentLength: Infinity,
    });
    console.log(`[HF] commit success status=${res.status} commitOid=${res.data?.commitOid}`);
    return res.data;
  } catch (err) {
    const detail = err.response
      ? `status=${err.response.status} body=${JSON.stringify(err.response.data)}`
      : err.message;
    console.error(`[HF] uploadAllFilesToHfSpace FAILED — ${detail}`);
    throw err;
  }
}

// Thin wrapper for single-file callers.
async function uploadFileToHfSpace(token, hfUsername, spaceName, filePath, fileContent) {
  return uploadAllFilesToHfSpace(token, hfUsername, spaceName, [
    { path: filePath, content: fileContent },
  ]);
}

async function setHfSpaceSecret(token, hfUsername, spaceName, key, value) {
  await axios.post(
    `${HF_API}/spaces/${hfUsername}/${spaceName}/secrets`,
    { key, value },
    {
      headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' },
      timeout: 15000,
    }
  );
}

async function deleteHfSpace(token, hfUsername, spaceName) {
  await axios.delete(`${HF_API}/repos/delete`, {
    data:    { type: 'space', name: `${hfUsername}/${spaceName}` },
    headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' },
    timeout: 20000,
  });
}

async function getHfSpaceRuntime(hfUsername, spaceName) {
  const res = await axios.get(`${HF_API}/spaces/${hfUsername}/${spaceName}/runtime`, { timeout: 10000 });
  return res.data;
}

function hfStageToHealth(stage) {
  if (!stage) return 'unknown';
  if (stage === 'RUNNING')          return 'running';
  if (stage.startsWith('RUNNING_')) return 'building';
  if (stage === 'STOPPED' || stage === 'PAUSED') return 'paused';
  if (stage.includes('ERROR'))      return 'error';
  if (stage.includes('BUILD'))      return 'building';
  return 'building';
}

async function getConstituentStats(spaceUrl) {
  try {
    const res = await axios.get(`${spaceUrl}/constituent/health`, {
      headers: { 'x-constituent-secret': MAIN_SERVER_SECRET },
      timeout: 8000,
    });
    return res.data;
  } catch { return null; }
}

// ═══════════════════════════════════════════════════════════════════════════
// CONSTITUENT ROUTES
// ═══════════════════════════════════════════════════════════════════════════

// POST /api/constituent/set-hf-key  { hfToken, tmdbKey? }
// tmdbKey is optional — if provided it overrides the server-wide TMDB_KEY
// for this user's constituent spaces only.
// NOTE: We skip the /whoami validation call intentionally — HuggingFace rate-limits
// the server IP aggressively. The token is validated implicitly when it is first used
// to create a Space (any auth failure there surfaces a clear error to the user).
streamApp.post('/api/constituent/set-hf-key', requireRegistered, async (req, res) => {
  const { hfToken, tmdbKey } = req.body;
  if (!hfToken || typeof hfToken !== 'string' || !hfToken.startsWith('hf_'))
    return res.status(400).json({ success: false, error: 'A valid HuggingFace token (starting with hf_) is required' });

  try {
    // Resolve the HF username via whoami only if not already stored; fall back
    // to a silent save with hfUsername=null so the token is usable right away.
    let hfUsername = null;
    try {
      const whoami = await validateHfToken(hfToken);
      if (whoami.valid) hfUsername = whoami.username;
    } catch { /* rate-limited or unreachable — store token without username */ }

    const update = { userId: req.user.userId, hfToken, hfUsername, updatedAt: new Date() };
    if (tmdbKey && typeof tmdbKey === 'string' && tmdbKey.trim()) update.tmdbKey = tmdbKey.trim();
    await db.collection('hf_keys').updateOne({ userId: req.user.userId }, { $set: update }, { upsert: true });
    res.json({
      success:    true,
      message:    'HuggingFace API key saved' + (hfUsername ? '' : ' (username will be resolved on first use)'),
      hfUsername: hfUsername || '(pending)',
      tmdbKeySet: !!update.tmdbKey,
    });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// GET /api/constituent/check-hf-key — returns whether the user has a saved HF key
// Returns: { success, hasKey, hfUsername?, tmdbKeySet? }
streamApp.get('/api/constituent/check-hf-key', requireRegistered, async (req, res) => {
  try {
    const keyDoc = await db.collection('hf_keys').findOne(
      { userId: req.user.userId },
      { projection: { _id: 0, hfUsername: 1, tmdbKey: 1, updatedAt: 1 } }
    );
    if (!keyDoc) return res.json({ success: true, hasKey: false });
    res.json({
      success:    true,
      hasKey:     true,
      hfUsername: keyDoc.hfUsername,
      tmdbKeySet: !!keyDoc.tmdbKey,
      updatedAt:  keyDoc.updatedAt,
    });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// DELETE /api/constituent/remove-hf-key — removes token AND deletes all spaces
streamApp.delete('/api/constituent/remove-hf-key', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const keyDoc = await db.collection('hf_keys').findOne({ userId });
    if (!keyDoc) return res.status(404).json({ success: false, error: 'No HuggingFace key found' });

    const constituents   = await db.collection('constituents').find({ userId }).toArray();
    const deleteResults  = [];
    for (const c of constituents) {
      try {
        await deleteHfSpace(keyDoc.hfToken, keyDoc.hfUsername, c.spaceName);
        deleteResults.push({ spaceName: c.spaceName, deleted: true });
      } catch (err) {
        deleteResults.push({ spaceName: c.spaceName, deleted: false, error: err.message });
      }
    }
    await db.collection('constituents').deleteMany({ userId });
    await db.collection('hf_keys').deleteOne({ userId });
    res.json({ success: true, message: 'HF key removed and constituent spaces deleted', spaces: deleteResults });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// PUT /api/constituent/update-hf-key  { hfToken, tmdbKey? }
// Updates the token without touching existing spaces.
streamApp.put('/api/constituent/update-hf-key', requireRegistered, async (req, res) => {
  const { hfToken, tmdbKey } = req.body;
  if (!hfToken || !hfToken.startsWith('hf_'))
    return res.status(400).json({ success: false, error: 'Valid HuggingFace token required' });

  try {
    // Best-effort whoami — don't hard-fail on rate-limit
    let hfUsername = null;
    try {
      const whoami = await validateHfToken(hfToken);
      if (whoami.valid) hfUsername = whoami.username;
    } catch { /* rate-limited — store token without username */ }

    const update = { hfToken, hfUsername, updatedAt: new Date() };
    if (tmdbKey && typeof tmdbKey === 'string' && tmdbKey.trim()) update.tmdbKey = tmdbKey.trim();
    const result = await db.collection('hf_keys').updateOne({ userId: req.user.userId }, { $set: update });
    if (result.matchedCount === 0)
      return res.status(404).json({ success: false, error: 'No HF key record found. Use set-hf-key first.' });
    res.json({
      success:    true,
      message:    'HuggingFace API key updated',
      hfUsername: hfUsername || '(pending)',
    });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/constituent/create  { spaceName }
// Creates a new HF Docker Space, uploads files, sets secrets. Max 3 per user.
streamApp.post('/api/constituent/create', requireRegistered, async (req, res) => {
  const { spaceName } = req.body;
  if (!spaceName || !/^[a-zA-Z0-9_-]{3,40}$/.test(spaceName))
    return res.status(400).json({ success: false, error: 'spaceName must be 3–40 alphanumeric/hyphen/underscore characters' });

  const userId = req.user.userId;
  try {
    const keyDoc = await db.collection('hf_keys').findOne({ userId });
    if (!keyDoc) return res.status(400).json({ success: false, error: 'Set a HuggingFace API key first via POST /api/constituent/set-hf-key' });

    // Step 1 — Always re-verify hfUsername via whoami-v2.
    // We never trust the DB-cached value because a previous bug stored the
    // numeric user ID (e.g. "69db2dbea12738d676a94b50") instead of the real
    // username (e.g. "Reaperxxxx"). whoami-v2 always returns the correct name.
    try {
      const whoamiRes = await axios.get(`${HF_API}/whoami-v2`, {
        headers: { Authorization: `Bearer ${keyDoc.hfToken}` },
        timeout: 10000,
      });
      const freshUsername = whoamiRes.data?.name || null;
      console.log(`[HF] whoami-v2 → name=${freshUsername}  (DB had: ${keyDoc.hfUsername})`);
      if (!freshUsername) {
        return res.status(400).json({ success: false, error: 'Could not determine HuggingFace username from token.' });
      }
      // Repair DB if the cached value was wrong
      if (freshUsername !== keyDoc.hfUsername) {
        console.log(`[HF] Correcting stored hfUsername: "${keyDoc.hfUsername}" → "${freshUsername}"`);
        await db.collection('hf_keys').updateOne({ userId }, { $set: { hfUsername: freshUsername } });
      }
      keyDoc.hfUsername = freshUsername;
    } catch (err) {
      const msg = err.response?.data?.error || err.message || 'unknown';
      return res.status(400).json({ success: false, error: `Could not verify HuggingFace token: ${msg}` });
    }

    const existingCount = await db.collection('constituents').countDocuments({ userId });
    if (existingCount >= MAX_CONSTITUENTS)
      return res.status(400).json({ success: false, error: `You can only have up to ${MAX_CONSTITUENTS} constituent servers` });

    const nameExists = await db.collection('constituents').findOne({ userId, spaceName });
    if (nameExists) return res.status(409).json({ success: false, error: 'You already have a constituent with that space name' });

    // Step 2 — Create the HF Space.
    // hfUsername is now authoritative (verified via whoami above) — no override needed.
    await createHfSpace(keyDoc.hfToken, keyDoc.hfUsername, spaceName);
    const spaceUrl = `https://${keyDoc.hfUsername}-${spaceName}.hf.space`.toLowerCase();

    // Read constituent files from disk (they live next to this server file)
    const constituentServerCode = fs.readFileSync(path.join(__dirname, 'files', 'constituent-server.js'));
    const dockerfileCode        = fs.readFileSync(path.join(__dirname, 'files', 'Dockerfile'));
    const packageJson = JSON.stringify({
      name: spaceName, 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',
      }
    }, null, 2);

    // Upload all 3 files in one atomic commit — one rebuild, no race window
    await uploadAllFilesToHfSpace(keyDoc.hfToken, keyDoc.hfUsername, spaceName, [
      { path: 'server.js',    content: constituentServerCode },
      { path: 'Dockerfile',   content: dockerfileCode },
      { path: 'package.json', content: Buffer.from(packageJson) },
    ]);

    // Secrets — user's own tmdbKey overrides the server default if provided
    const effectiveTmdbKey = keyDoc.tmdbKey || TMDB_KEY || '';
    await setHfSpaceSecret(keyDoc.hfToken, keyDoc.hfUsername, spaceName, 'CONSTITUENT_OWNER_ID', userId);
    await setHfSpaceSecret(keyDoc.hfToken, keyDoc.hfUsername, spaceName, 'MAIN_SERVER_SECRET',   MAIN_SERVER_SECRET || '');
    if (effectiveTmdbKey) {
      await setHfSpaceSecret(keyDoc.hfToken, keyDoc.hfUsername, spaceName, 'TMDB_KEY', effectiveTmdbKey);
    }

    const constituentDoc = {
      userId, spaceName,
      hfUsername: keyDoc.hfUsername,
      spaceUrl,
      repoUrl:   `https://huggingface.co/spaces/${keyDoc.hfUsername}/${spaceName}`,
      createdAt: new Date(),
    };
    await db.collection('constituents').insertOne(constituentDoc);

    res.json({
      success:  true,
      message:  'Constituent Space created. HuggingFace is now building it — check status with GET /api/constituent/status/:spaceName',
      spaceName, spaceUrl, repoUrl: constituentDoc.repoUrl,
    });
  } catch (err) {
    const hfStatus = err.response?.status;
    const hfBody   = err.response?.data;
    console.error('/api/constituent/create error:', err.message);
    if (hfStatus) console.error(`[HF] response status=${hfStatus} body=${JSON.stringify(hfBody)}`);
    res.status(500).json({
      success:  false,
      error:    err.message,
      hfStatus: hfStatus || null,
      hfError:  hfBody   || null,
    });
  }
});

// GET /api/constituent/status/:spaceName
// 206 = still booting (tell user to wait), 200 = running and ready.
streamApp.get('/api/constituent/status/:spaceName', requireRegistered, async (req, res) => {
  const { spaceName } = req.params;
  const userId = req.user.userId;
  try {
    const constituent = await db.collection('constituents').findOne({ userId, spaceName });
    if (!constituent) return res.status(404).json({ success: false, error: 'Constituent not found' });

    let hfRuntime = null;
    try { hfRuntime = await getHfSpaceRuntime(constituent.hfUsername, spaceName); } catch {}
    const hfStage  = hfRuntime?.stage || 'UNKNOWN';
    const health   = hfStageToHealth(hfStage);

    let constituentStats = null;
    if (health === 'running') constituentStats = await getConstituentStats(constituent.spaceUrl);
    const isReady = health === 'running' && constituentStats !== null;

    res.status(isReady ? 200 : 206).json({
      success: true, spaceName,
      spaceUrl: constituent.spaceUrl, repoUrl: constituent.repoUrl,
      hfStage, health, isReady,
      message: isReady
        ? 'Constituent is running and ready to accept movies'
        : health === 'paused'
          ? 'Constituent is paused (HuggingFace free tier timeout). It will wake on next request.'
          : 'HuggingFace is still booting this constituent — please wait a moment',
      stats:   constituentStats,
      domains: hfRuntime?.domains || [],
    });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// GET /api/constituent/list — all constituents with live RAM/CPU/disk stats
streamApp.get('/api/constituent/list', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const constituents = await db.collection('constituents').find({ userId }).toArray();
    const results = await Promise.all(constituents.map(async (c) => {
      let hfRuntime = null;
      try { hfRuntime = await getHfSpaceRuntime(c.hfUsername, c.spaceName); } catch {}
      const hfStage = hfRuntime?.stage || 'UNKNOWN';
      const health  = hfStageToHealth(hfStage);
      let stats = null;
      if (health === 'running') stats = await getConstituentStats(c.spaceUrl);
      return {
        spaceName: c.spaceName, hfUsername: c.hfUsername,
        spaceUrl: c.spaceUrl, repoUrl: c.repoUrl, createdAt: c.createdAt,
        hfStage, health, isReady: health === 'running' && stats !== null,
        stats: stats ? { memory: stats.memory, cpu: stats.cpu, disk: stats.disk, streams: stats.streams, uptime: stats.uptime } : null,
      };
    }));
    res.json({ success: true, total: results.length, maxAllowed: MAX_CONSTITUENTS, constituents: results });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// DELETE /api/constituent/delete/:spaceName — delete one space (keeps HF key)
streamApp.delete('/api/constituent/delete/:spaceName', requireRegistered, async (req, res) => {
  const { spaceName } = req.params;
  const userId = req.user.userId;
  try {
    const keyDoc      = await db.collection('hf_keys').findOne({ userId });
    const constituent = await db.collection('constituents').findOne({ userId, spaceName });
    if (!constituent) return res.status(404).json({ success: false, error: 'Constituent not found' });
    let hfDeleted = false;
    if (keyDoc) {
      try { await deleteHfSpace(keyDoc.hfToken, keyDoc.hfUsername, spaceName); hfDeleted = true; } catch (err) {
        console.warn(`Could not delete HF Space ${spaceName}:`, err.message);
      }
    }
    await db.collection('constituents').deleteOne({ userId, spaceName });
    res.json({ success: true, message: `Constituent ${spaceName} deleted`, hfDeleted });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/constituent/add-movie  { spaceName, movieLink, movieTitle, thumbnail? }
// OR   { spaceName, showId, movieTitle }  — showId is an iktracks link; server fetches the download link
// User picks a constituent; main server proxies the add-movie call to it.
streamApp.post('/api/constituent/add-movie', requireRegistered, async (req, res) => {
  const { spaceName, movieLink: rawMovieLink, movieTitle, showId, thumbnail } = req.body;
  if (!spaceName || !movieTitle)
    return res.status(400).json({ success: false, error: 'spaceName and movieTitle are required' });

  const userId = req.user.userId;
  try {
    const constituent = await db.collection('constituents').findOne({ userId, spaceName });
    if (!constituent) return res.status(404).json({ success: false, error: 'Constituent not found or not yours' });

    // Health check
    let hfRuntime = null;
    try { hfRuntime = await getHfSpaceRuntime(constituent.hfUsername, spaceName); } catch {}
    const health = hfStageToHealth(hfRuntime?.stage);
    if (health !== 'running') {
      return res.status(503).json({
        success: false,
        error: health === 'paused'
          ? 'This constituent is paused. It will wake shortly — try again in ~1 minute.'
          : 'This constituent is not running yet. Check status with GET /api/constituent/status/:spaceName',
        health,
      });
    }

    // If showId given (iktracks link) but no direct movieLink, fetch the download link
    let movieLink = rawMovieLink;
    let resolvedThumbnail = thumbnail;
    if (!movieLink && showId) {
      try {
        const r       = await axios.get(`https://iktracks.vercel.app/details?url=${encodeURIComponent(showId)}`, { timeout: 15000 });
        const details = r.data;
        if (!details || details.type === 'series') {
          return res.status(400).json({ success: false, error: 'showId must point to a movie, not a series' });
        }
        const link = details.downloadLinks?.[0]?.downloadLink;
        if (!link) return res.status(404).json({ success: false, error: 'No download link found for this movie' });
        movieLink = link;
        if (!resolvedThumbnail && details.thumbnail) resolvedThumbnail = details.thumbnail;
      } catch (err) {
        return res.status(500).json({ success: false, error: `Failed to resolve showId: ${err.message}` });
      }
    }

    if (!movieLink) {
      return res.status(400).json({ success: false, error: 'Either movieLink or showId is required' });
    }

    // Optional TMDB enrichment — use user's key if they have one, else server default
    const keyDoc = await db.collection('hf_keys').findOne({ userId });
    const effectiveTmdbKey = keyDoc?.tmdbKey || TMDB_KEY;
    let tmdbInfo = null;
    if (effectiveTmdbKey) {
      try {
        const tmdbRes = await axios.get(`${TMDB_BASE}/search/movie`, {
          params: { api_key: effectiveTmdbKey, query: movieTitle, language: 'en-US', page: 1 },
          timeout: 6000,
        });
        const hit = tmdbRes.data?.results?.[0];
        if (hit) tmdbInfo = {
          tmdbId: hit.id, title: hit.title || movieTitle,
          overview: hit.overview || null, releaseDate: hit.release_date || null,
          poster: hit.poster_path ? `${TMDB_IMG}${hit.poster_path}` : null,
          voteAverage: hit.vote_average || null,
        };
      } catch {}
    }

    const proxyRes = await axios.post(
      `${constituent.spaceUrl}/constituent/add-movie`,
      { streamId: userId, movieLink, movieTitle, thumbnail: resolvedThumbnail || null, tmdbInfo },
      { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET, 'Content-Type': 'application/json' }, timeout: 15000 }
    );

    // Record that this user's stream is now hosted on this constituent
    setStreamLocation(userId, 'constituent', constituent.spaceUrl, spaceName).catch(console.error);

    res.json({
      success: true,
      message: proxyRes.data.message || 'Movie queued on constituent',
      spaceName, spaceUrl: constituent.spaceUrl,
      title: movieTitle,
      hlsUrl: `${constituent.spaceUrl}/stream-hls/${userId}/live.m3u8`,
      tmdb: tmdbInfo,
    });
  } catch (err) {
    console.error('/api/constituent/add-movie error:', err.message);
    res.status(500).json({ success: false, error: err.response?.data?.error || err.message });
  }
});

// POST /api/constituent/add-song  { spaceName, songUrl, title, thumbnail? }
// Proxies an audio-track add request to the user's chosen constituent server.
streamApp.post('/api/constituent/add-song', requireRegistered, async (req, res) => {
  const { spaceName, songUrl, title, thumbnail } = req.body;
  if (!spaceName || !songUrl || !title)
    return res.status(400).json({ success: false, error: 'spaceName, songUrl, and title are required' });

  const userId = req.user.userId;
  try {
    const constituent = await db.collection('constituents').findOne({ userId, spaceName });
    if (!constituent) return res.status(404).json({ success: false, error: 'Constituent not found or not yours' });

    // Health check — reject if the space isn't running
    let hfRuntime = null;
    try { hfRuntime = await getHfSpaceRuntime(constituent.hfUsername, spaceName); } catch {}
    const health = hfStageToHealth(hfRuntime?.stage);
    if (health !== 'running') {
      return res.status(503).json({
        success: false,
        error: health === 'paused'
          ? 'This constituent is paused. It will wake shortly — try again in ~1 minute.'
          : 'This constituent is not running yet. Check status with GET /api/constituent/status/:spaceName',
        health,
      });
    }

    const proxyRes = await axios.post(
      `${constituent.spaceUrl}/constituent/add-song`,
      { streamId: userId, songUrl, title, thumbnail: thumbnail || null },
      { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET, 'Content-Type': 'application/json' }, timeout: 15000 }
    );

    // Record that this user's stream is now hosted on this constituent
    setStreamLocation(userId, 'constituent', constituent.spaceUrl, spaceName).catch(console.error);

    res.json({
      success:  true,
      message:  proxyRes.data.message || 'Song queued on constituent',
      spaceName,
      spaceUrl: constituent.spaceUrl,
      title,
      hlsUrl:   `${constituent.spaceUrl}/stream-hls/${userId}/live.m3u8`,
    });
  } catch (err) {
    console.error('/api/constituent/add-song error:', err.message);
    res.status(500).json({ success: false, error: err.response?.data?.error || err.message });
  }
});

// GET /api/constituent/queue/:spaceName
streamApp.get('/api/constituent/queue/:spaceName', requireRegistered, async (req, res) => {
  const { spaceName } = req.params;
  const userId = req.user.userId;
  try {
    const constituent = await db.collection('constituents').findOne({ userId, spaceName });
    if (!constituent) return res.status(404).json({ success: false, error: 'Constituent not found' });
    const proxyRes = await axios.get(
      `${constituent.spaceUrl}/constituent/queue/${userId}`,
      { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 10000 }
    );
    res.json(proxyRes.data);
  } catch (err) {
    res.status(500).json({ success: false, error: err.response?.data?.error || err.message });
  }
});

// POST /api/constituent/skip/:spaceName
streamApp.post('/api/constituent/skip/:spaceName', requireRegistered, async (req, res) => {
  const { spaceName } = req.params;
  const userId = req.user.userId;
  try {
    const constituent = await db.collection('constituents').findOne({ userId, spaceName });
    if (!constituent) return res.status(404).json({ success: false, error: 'Constituent not found' });
    await axios.post(
      `${constituent.spaceUrl}/constituent/skip/${userId}`,
      {},
      { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 10000 }
    );
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.response?.data?.error || err.message });
  }
});

// ═══════════════════════════════════════════════════════════════════════════
// STREAM SOURCE — resolve any user's HLS source (for watch picker)
// GET /api/stream-source/:streamId
// ═══════════════════════════════════════════════════════════════════════════

streamApp.get('/api/stream-source/:streamId', async (req, res) => {
  const raw             = req.params.streamId;
  const streamId        = await resolveStreamId(raw);
  const wantConstituent = req.query.constituentName || null; // optional filter

  // 1. Check constituents owned by this userId
  try {
    const query = wantConstituent
      ? { userId: streamId, spaceName: wantConstituent }
      : { userId: streamId };
    const constituents = await db.collection('constituents').find(query).toArray();
    for (const c of constituents) {
      try {
        const r = await axios.get(
          `${c.spaceUrl}/constituent/queue/${streamId}`,
          { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 6000 }
        );
        const data = r.data;
        if (data.success && data.isActive && data.hlsUrl) {
          const current = data.queue?.[0] || null;
          return res.json({
            success:   true,
            source:    'constituent',
            hlsUrl:    `${c.spaceUrl}${data.hlsUrl}`,
            socketUrl: c.spaceUrl,
            streamId,
            spaceName: c.spaceName,
            nowPlaying: current ? { title: current.title, thumbnail: current.thumbnail, duration: current.duration } : null,
            currentTrackUrl: `${c.spaceUrl}/stream/${streamId}/currentTrack`,
          });
        }
      } catch { /* unreachable */ }
    }
  } catch { /* no constituents */ }

  // 2. Fall back to main pool
  const stream = streams[streamId];
  if (stream && stream.isActive && stream.queue.length > 0) {
    const current = stream.queue[0];
    return res.json({
      success:   true,
      source:    'pool',
      hlsUrl:    `/stream-hls/${streamId}/live.m3u8`,
      socketUrl: null,
      streamId,
      nowPlaying: { title: current.meta.title, thumbnail: current.meta.thumbnail, duration: current.meta.duration },
      currentTrackUrl: `/stream/${streamId}/currentTrack`,
    });
  }

  res.json({ success: true, source: 'none', hlsUrl: null, streamId });
});

// ═══════════════════════════════════════════════════════════════════════════
// ROOMS
// MongoDB collection: rooms
//   { _id, id, name, ownerId, constituentName?, members: [{userId,username,displayName}],
//     messages: [{id,userId,displayName,body,createdAt,replyTo?}],
//     nowPlaying: {title,thumbnail}?, createdAt }
// ═══════════════════════════════════════════════════════════════════════════

function makeRoomId() { return 'room_' + crypto.randomBytes(8).toString('hex'); }

// Ensure rooms index
async function ensureRoomsIndexes() {
  const col = db.collection('rooms');
  await col.createIndex({ id: 1 }, { unique: true });
  await col.createIndex({ 'members.userId': 1 });
  await col.createIndex({ ownerId: 1 });
}

// GET /api/rooms — list rooms the user owns or is a member of
streamApp.get('/api/rooms', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const rooms = await db.collection('rooms')
      .find({ 'members.userId': userId }, { projection: { messages: 0 } })
      .sort({ createdAt: -1 })
      .toArray();
    res.json({ success: true, rooms: rooms.map(r => ({ ...r, _id: undefined })) });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/rooms/create  { name, constituentName? }
streamApp.post('/api/rooms/create', requireRegistered, async (req, res) => {
  const { name, constituentName } = req.body;
  if (!name || typeof name !== 'string' || !name.trim())
    return res.status(400).json({ success: false, error: 'Room name is required' });
  if (name.trim().length > 40)
    return res.status(400).json({ success: false, error: 'Room name must be at most 40 characters' });

  const userId = req.user.userId;
  try {
    const existing = await db.collection('rooms').countDocuments({ ownerId: userId });
    if (existing >= 3)
      return res.status(400).json({ success: false, error: 'You can only have up to 3 rooms' });

    const user = await db.collection('users').findOne(
      { userId },
      { projection: { username: 1, displayName: 1 } }
    );

    const roomId = makeRoomId();
    const now    = new Date().toISOString();
    const room   = {
      id:              roomId,
      name:            name.trim(),
      ownerId:         userId,
      constituentName: constituentName || null,
      members:         [{ userId, username: user.username, displayName: user.displayName }],
      messages:        [],
      nowPlaying:      null,
      createdAt:       now,
    };
    await db.collection('rooms').insertOne(room);
    res.json({ success: true, room: { ...room, _id: undefined } });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// GET /api/rooms/:roomId
streamApp.get('/api/rooms/:roomId', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const room = await db.collection('rooms').findOne(
      { id: req.params.roomId },
      { projection: { messages: 0 } }
    );
    if (!room) return res.status(404).json({ success: false, error: 'Room not found' });
    const isMember = room.members.some(m => m.userId === userId);
    if (!isMember && room.ownerId !== userId)
      return res.status(403).json({ success: false, error: 'You are not a member of this room' });
    res.json({ success: true, room: { ...room, _id: undefined } });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// DELETE /api/rooms/:roomId — owner only
streamApp.delete('/api/rooms/:roomId', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const room = await db.collection('rooms').findOne({ id: req.params.roomId });
    if (!room) return res.status(404).json({ success: false, error: 'Room not found' });
    if (room.ownerId !== userId)
      return res.status(403).json({ success: false, error: 'Only the owner can delete this room' });
    await db.collection('rooms').deleteOne({ id: req.params.roomId });
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/rooms/:roomId/leave
streamApp.post('/api/rooms/:roomId/leave', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const room = await db.collection('rooms').findOne({ id: req.params.roomId });
    if (!room) return res.status(404).json({ success: false, error: 'Room not found' });
    if (room.ownerId === userId)
      return res.status(400).json({ success: false, error: 'Owner cannot leave. Delete the room instead.' });
    await db.collection('rooms').updateOne(
      { id: req.params.roomId },
      { $pull: { members: { userId } } }
    );
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/rooms/:roomId/members/add  { username }  — owner only
streamApp.post('/api/rooms/:roomId/members/add', requireRegistered, async (req, res) => {
  const ownerId  = req.user.userId;
  const { username } = req.body;
  if (!username) return res.status(400).json({ success: false, error: 'username is required' });
  try {
    const room = await db.collection('rooms').findOne({ id: req.params.roomId });
    if (!room) return res.status(404).json({ success: false, error: 'Room not found' });
    if (room.ownerId !== ownerId)
      return res.status(403).json({ success: false, error: 'Only the owner can add members' });

    const target = await db.collection('users').findOne(
      { username: username.toLowerCase().trim() },
      { projection: { userId: 1, username: 1, displayName: 1 } }
    );
    if (!target) return res.status(404).json({ success: false, error: 'User not found' });

    const alreadyMember = room.members.some(m => m.userId === target.userId);
    if (alreadyMember)
      return res.status(409).json({ success: false, error: 'User is already a member' });

    await db.collection('rooms').updateOne(
      { id: req.params.roomId },
      { $push: { members: { userId: target.userId, username: target.username, displayName: target.displayName } } }
    );

    // Create notification for the invited user
    await createNotification(target.userId, 'room_invite', {
      fromUserId: ownerId,
      fromDisplayName: req.user.displayName,
      roomId: room.id,
      roomName: room.name,
    });

    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/rooms/:roomId/members/remove  { userId }  — owner only
streamApp.post('/api/rooms/:roomId/members/remove', requireRegistered, async (req, res) => {
  const ownerId  = req.user.userId;
  const { userId: targetId } = req.body;
  if (!targetId) return res.status(400).json({ success: false, error: 'userId is required' });
  try {
    const room = await db.collection('rooms').findOne({ id: req.params.roomId });
    if (!room) return res.status(404).json({ success: false, error: 'Room not found' });
    if (room.ownerId !== ownerId)
      return res.status(403).json({ success: false, error: 'Only the owner can remove members' });
    if (targetId === ownerId)
      return res.status(400).json({ success: false, error: 'Cannot remove yourself as owner' });
    await db.collection('rooms').updateOne(
      { id: req.params.roomId },
      { $pull: { members: { userId: targetId } } }
    );
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// GET /api/rooms/:roomId/messages — last 100 messages
streamApp.get('/api/rooms/:roomId/messages', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const room = await db.collection('rooms').findOne(
      { id: req.params.roomId },
      { projection: { members: 1, ownerId: 1, messages: { $slice: -100 } } }
    );
    if (!room) return res.status(404).json({ success: false, error: 'Room not found' });
    const isMember = room.members.some(m => m.userId === userId) || room.ownerId === userId;
    if (!isMember)
      return res.status(403).json({ success: false, error: 'Not a member of this room' });
    res.json({ success: true, messages: room.messages || [] });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/rooms/join-request/respond
// Accepts: { roomId, requesterId, accept } OR { roomId, userId, action: 'accept'|'decline' }
streamApp.post('/api/rooms/join-request/respond', requireRegistered, async (req, res) => {
  const ownerId = req.user.userId;
  const { roomId, requesterId: _rid, userId: _uid, accept, action } = req.body;
  const requesterId  = _rid || _uid;
  const shouldAccept = accept !== undefined ? !!accept : (action === 'accept');
  if (!roomId || !requesterId) return res.status(400).json({ success: false, error: 'roomId and requesterId/userId are required' });
  try {
    const room = await db.collection('rooms').findOne({ id: roomId });
    if (!room) return res.status(404).json({ success: false, error: 'Room not found' });
    if (room.ownerId !== ownerId) return res.status(403).json({ success: false, error: 'Not the room owner' });

    if (shouldAccept) {
      const target = await db.collection('users').findOne({ userId: requesterId }, { projection: { userId: 1, username: 1, displayName: 1 } });
      if (target) {
        const alreadyMember = room.members.some(m => m.userId === requesterId);
        if (!alreadyMember) {
          await db.collection('rooms').updateOne(
            { id: roomId },
            { $push: { members: { userId: target.userId, username: target.username, displayName: target.displayName } } }
          );
        }
      }
    }
    res.json({ success: true, accepted: shouldAccept });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/rooms/queue-channel  { roomId, constituentName, streamId, name, hlsUrl? }
streamApp.post('/api/rooms/queue-channel', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  const { roomId, constituentName, streamId: channelStreamId, name, hlsUrl } = req.body;
  if (!roomId || !constituentName)
    return res.status(400).json({ success: false, error: 'roomId and constituentName are required' });
  try {
    const room = await db.collection('rooms').findOne({ id: roomId });
    if (!room) return res.status(404).json({ success: false, error: 'Room not found' });
    const isMember = room.members.some(m => m.userId === userId) || room.ownerId === userId;
    if (!isMember) return res.status(403).json({ success: false, error: 'Not a member of this room' });

    const constituent = await db.collection('constituents').findOne({ spaceName: constituentName });
    if (!constituent) return res.status(404).json({ success: false, error: 'Constituent not found' });

    // Update room nowPlaying
    await db.collection('rooms').updateOne(
      { id: roomId },
      { $set: { nowPlaying: { title: name, thumbnail: null, streamId: channelStreamId } } }
    );

    // Notify room members via socket
    const roomSocketKey = `room:${roomId}`;
    io.to(roomSocketKey).emit('room:now_playing', { title: name, streamId: channelStreamId, hlsUrl });

    res.json({ success: true, name });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// ═══════════════════════════════════════════════════════════════════════════
// DEFAULT CHANNELS
// GET /api/default-channels — returns list of public live channels/streams
// These are users who have made their stream public or admin-defined channels
// ═══════════════════════════════════════════════════════════════════════════

streamApp.get('/api/default-channels', requireRegistered, async (req, res) => {
  try {
    // Return active main-pool streams as channels
    const liveChannels = Object.entries(streams)
      .filter(([, s]) => s.isActive && s.queue.length > 0)
      .map(([streamId, s]) => {
        const current = s.queue[0];
        return {
          streamId,
          name:        current.meta.title || streamId,
          description: `Live stream`,
          hlsUrl:      `/stream-hls/${streamId}/live.m3u8`,
          thumbnail:   current.meta.thumbnail || null,
          listeners:   s.users?.size || 0,
        };
      })
      .slice(0, 20);
    // Also include any admin-defined channels from DB (if collection exists)
    let adminChannels = [];
    try {
      adminChannels = await db.collection('default_channels').find({}).toArray();
    } catch { /* collection may not exist yet */ }
    const channels = [
      ...DEFAULT_CHANNELS,
      ...adminChannels.map(c => ({ streamId: c.streamId, name: c.name, description: c.description || '', hlsUrl: c.hlsUrl || null, thumbnail: c.thumbnail || null, listeners: 0 })),
      ...liveChannels,
    ];
    res.json({ success: true, channels });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// ═══════════════════════════════════════════════════════════════════════════
// NOTIFICATIONS
// MongoDB collection: notifications
//   { id, userId, type, data, read, createdAt }
// ═══════════════════════════════════════════════════════════════════════════

function makeNotifId() { return 'notif_' + crypto.randomBytes(6).toString('hex'); }

function buildNotifMessage(type, data) {
  switch (type) {
    case 'friend_request':  return `${data.fromDisplayName || 'Someone'} sent you a friend request.`;
    case 'friend_accepted': return `${data.fromDisplayName || 'Someone'} accepted your friend request.`;
    case 'watch_invite':    return `${data.fromDisplayName || 'Someone'} invited you to watch together.`;
    case 'room_invite':     return `${data.fromDisplayName || 'Someone'} added you to room ${data.roomName || ''}.`;
    case 'room_join_request': return `${data.fromDisplayName || 'Someone'} wants to join your room ${data.roomName || ''}.`;
    default:                return type;
  }
}

async function createNotification(userId, type, data = {}) {
  try {
    const notif = {
      id:          makeNotifId(),
      userId,
      type,
      data,
      // Flat fields the frontend reads directly
      message:     buildNotifMessage(type, data),
      status:      'pending',
      read:        false,
      fromUserId:  data.fromUserId  || null,
      roomId:      data.roomId      || null,
      watchUrl:    data.watchUrl    || null,
      createdAt:   new Date().toISOString(),
    };
    await db.collection('notifications').insertOne(notif);
    io.to(`user:${userId}`).emit('notification', { ...notif, _id: undefined });
    return notif;
  } catch (err) {
    console.error('createNotification error:', err.message);
  }
}

// GET /api/notifications
streamApp.get('/api/notifications', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const notifications = await db.collection('notifications')
      .find({ userId })
      .sort({ createdAt: -1 })
      .limit(50)
      .toArray();
    res.json({ success: true, notifications: notifications.map(n => ({ ...n, _id: undefined })) });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// GET /api/notifications/unread-count
streamApp.get('/api/notifications/unread-count', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const count = await db.collection('notifications').countDocuments({ userId, read: false });
    res.json({ success: true, count });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/notifications/read-all
streamApp.post('/api/notifications/read-all', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    await db.collection('notifications').updateMany({ userId, read: false }, { $set: { read: true } });
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/notifications/:id/read
streamApp.post('/api/notifications/:id/read', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const result = await db.collection('notifications').updateOne(
      { id: req.params.id, userId },
      { $set: { read: true } }
    );
    if (result.matchedCount === 0)
      return res.status(404).json({ success: false, error: 'Notification not found' });
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// ═══════════════════════════════════════════════════════════════════════════
// SOCIAL — Friends system
// MongoDB collection: friendships
//   { id, requesterId, addresseeId, status: 'pending'|'accepted', createdAt, updatedAt }
// ═══════════════════════════════════════════════════════════════════════════

async function ensureSocialIndexes() {
  const col = db.collection('friendships');
  await col.createIndex({ requesterId: 1, addresseeId: 1 }, { unique: true });
  await col.createIndex({ addresseeId: 1 });
  await col.createIndex({ requesterId: 1 });
}

function makeFriendshipId() { return 'fs_' + crypto.randomBytes(8).toString('hex'); }

// GET /api/social/friends
streamApp.get('/api/social/friends', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const friendships = await db.collection('friendships')
      .find({
        $or: [{ requesterId: userId }, { addresseeId: userId }],
        status: 'accepted',
      })
      .toArray();

    const friendIds = friendships.map(f =>
      f.requesterId === userId ? f.addresseeId : f.requesterId
    );

    const users = await db.collection('users')
      .find({ userId: { $in: friendIds } }, { projection: { _id: 0, userId: 1, username: 1, displayName: 1 } })
      .toArray();

    const friends = users.map(u => {
      const watchingStreamId = activeViewers[u.userId] || null;
      let nowPlaying = null;
      if (watchingStreamId) {
        const ws = streams[watchingStreamId];
        nowPlaying = ws && ws.queue && ws.queue[0] ? ws.queue[0].meta.title : null;
      }
      return {
        ...u,
        isOnline:          !!(watchingStreamId),
        isWatching:        !!(watchingStreamId),
        currentlyWatching: watchingStreamId,
        streamId:          watchingStreamId,
        nowPlaying,
      };
    });

    res.json({ success: true, friends });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// GET /api/social/requests — incoming (sent to me) + outgoing (sent by me)
streamApp.get('/api/social/requests', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  try {
    const [incomingFs, outgoingFs] = await Promise.all([
      db.collection('friendships').find({ addresseeId: userId, status: 'pending' }).toArray(),
      db.collection('friendships').find({ requesterId: userId, status: 'pending' }).toArray(),
    ]);

    const allIds = [
      ...incomingFs.map(f => f.requesterId),
      ...outgoingFs.map(f => f.addresseeId),
    ];
    const users = await db.collection('users')
      .find({ userId: { $in: allIds } }, { projection: { _id: 0, userId: 1, username: 1, displayName: 1 } })
      .toArray();
    const umap = {};
    users.forEach(u => { umap[u.userId] = u; });

    const incoming = incomingFs.map(f => ({
      friendshipId: f.id,
      userId:       f.requesterId,
      ...(umap[f.requesterId] || { userId: f.requesterId }),
      createdAt:    f.createdAt,
    }));
    const outgoing = outgoingFs.map(f => ({
      friendshipId: f.id,
      userId:       f.addresseeId,
      ...(umap[f.addresseeId] || { userId: f.addresseeId }),
      createdAt:    f.createdAt,
    }));

    res.json({ success: true, incoming, outgoing, requests: incoming }); // requests kept for back-compat
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/social/friend-request  { username }
streamApp.post('/api/social/friend-request', requireRegistered, async (req, res) => {
  const requesterId = req.user.userId;
  const { username }  = req.body;
  if (!username) return res.status(400).json({ success: false, error: 'username is required' });
  try {
    const target = await db.collection('users').findOne(
      { username: username.toLowerCase().trim() },
      { projection: { userId: 1, username: 1, displayName: 1 } }
    );
    if (!target) return res.status(404).json({ success: false, error: 'User not found' });
    if (target.userId === requesterId)
      return res.status(400).json({ success: false, error: 'You cannot send a friend request to yourself' });

    // Check no existing friendship / request
    const existing = await db.collection('friendships').findOne({
      $or: [
        { requesterId, addresseeId: target.userId },
        { requesterId: target.userId, addresseeId: requesterId },
      ],
    });
    if (existing) {
      if (existing.status === 'accepted') return res.status(409).json({ success: false, error: 'Already friends' });
      return res.status(409).json({ success: false, error: 'Friend request already sent' });
    }

    const now = new Date().toISOString();
    await db.collection('friendships').insertOne({
      id:          makeFriendshipId(),
      requesterId,
      addresseeId: target.userId,
      status:      'pending',
      createdAt:   now,
      updatedAt:   now,
    });

    await createNotification(target.userId, 'friend_request', {
      fromUserId:      requesterId,
      fromDisplayName: req.user.displayName,
    });

    res.json({ success: true, message: `Friend request sent to ${target.displayName}` });
  } catch (err) {
    if (err.code === 11000) return res.status(409).json({ success: false, error: 'Friend request already exists' });
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/social/friend-request/cancel  { targetUserId }
streamApp.post('/api/social/friend-request/cancel', requireRegistered, async (req, res) => {
  const requesterId   = req.user.userId;
  const { targetUserId: _tuid, userId: _uid } = req.body;
  const targetUserId = _tuid || _uid;
  if (!targetUserId) return res.status(400).json({ success: false, error: 'targetUserId is required' });
  try {
    const result = await db.collection('friendships').deleteOne({
      requesterId,
      addresseeId: targetUserId,
      status:      'pending',
    });
    if (result.deletedCount === 0)
      return res.status(404).json({ success: false, error: 'Pending request not found' });
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// POST /api/social/friend-request/respond
// Accepts: { friendshipId, accept } OR { userId, action: 'accept'|'decline' }
streamApp.post('/api/social/friend-request/respond', requireRegistered, async (req, res) => {
  const addresseeId = req.user.userId;
  const { friendshipId, accept, userId: requesterUserId, action } = req.body;
  // Normalise both calling conventions
  const shouldAccept = accept !== undefined ? !!accept : (action === 'accept');
  try {
    let friendship;
    if (friendshipId) {
      friendship = await db.collection('friendships').findOne({ id: friendshipId, addresseeId, status: 'pending' });
    } else if (requesterUserId) {
      friendship = await db.collection('friendships').findOne({ requesterId: requesterUserId, addresseeId, status: 'pending' });
    }
    if (!friendship) return res.status(404).json({ success: false, error: 'Friend request not found' });

    if (shouldAccept) {
      await db.collection('friendships').updateOne(
        { id: friendship.id },
        { $set: { status: 'accepted', updatedAt: new Date().toISOString() } }
      );
      await createNotification(friendship.requesterId, 'friend_accepted', {
        fromUserId:      addresseeId,
        fromDisplayName: req.user.displayName,
      });
    } else {
      await db.collection('friendships').deleteOne({ id: friendship.id });
    }

    res.json({ success: true, accepted: shouldAccept });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// DELETE /api/social/friends/remove and POST alias — { friendUserId } or { userId }
async function handleRemoveFriend(req, res) {
  const myId = req.user.userId;
  const { friendUserId, userId: _uid } = req.body;
  const targetId = friendUserId || _uid;
  if (!targetId) return res.status(400).json({ success: false, error: 'friendUserId is required' });
  try {
    const result = await db.collection('friendships').deleteOne({
      $or: [
        { requesterId: myId,    addresseeId: targetId, status: 'accepted' },
        { requesterId: targetId, addresseeId: myId,   status: 'accepted' },
      ],
    });
    if (result.deletedCount === 0)
      return res.status(404).json({ success: false, error: 'Friendship not found' });
    res.json({ success: true });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
}
streamApp.delete('/api/social/friends/remove', requireRegistered, handleRemoveFriend);
streamApp.post('/api/social/friends/remove',   requireRegistered, handleRemoveFriend);

// POST /api/social/invite-watch  { friendUserId | username, streamId? }
streamApp.post('/api/social/invite-watch', requireRegistered, async (req, res) => {
  const userId = req.user.userId;
  const { friendUserId: _fid, username, streamId: inviteStreamId } = req.body;
  try {
    let friendUserId = _fid;
    // Resolve username → userId if needed
    if (!friendUserId && username) {
      const target = await db.collection('users').findOne(
        { username: username.toLowerCase().trim() },
        { projection: { userId: 1 } }
      );
      if (!target) return res.status(404).json({ success: false, error: 'User not found' });
      friendUserId = target.userId;
    }
    if (!friendUserId) return res.status(400).json({ success: false, error: 'friendUserId or username is required' });

    // Verify they are friends
    const fs = await db.collection('friendships').findOne({
      $or: [
        { requesterId: userId, addresseeId: friendUserId },
        { requesterId: friendUserId, addresseeId: userId },
      ],
      status: 'accepted',
    });
    if (!fs) return res.status(403).json({ success: false, error: 'You can only invite friends to watch' });

    const targetStreamId = inviteStreamId || userId;
    await createNotification(friendUserId, 'watch_invite', {
      fromUserId:      userId,
      fromDisplayName: req.user.displayName,
      streamId:        targetStreamId,
      watchUrl:        `/watch/${targetStreamId}`,
    });

    res.json({ success: true, message: 'Watch invite sent' });
  } catch (err) {
    res.status(500).json({ success: false, error: err.message });
  }
});

// ═══════════════════════════════════════════════════════════════════════════
// SOCKET.IO — ROOM CHAT
// Rooms: users join socket room `room:<roomId>` and `user:<userId>`
// Events: room:join, room:message
// Broadcasts: room:message, room:user_joined, room:user_left, room:now_playing
// ═══════════════════════════════════════════════════════════════════════════

// Track connected sockets per room: roomId -> Map<socketId, {userId, displayName}>
const roomSockets = {};

io.on('connection', (socket) => {

  // ── room:join ────────────────────────────────────────────────────────────
  socket.on('room:join', async ({ roomId, userId, displayName }) => {
    if (!roomId || !userId) return;
    try {
      // Verify the user is a member (best-effort — skip guests)
      const isGuest = String(userId).startsWith('g_');
      if (!isGuest) {
        const room = await db.collection('rooms').findOne(
          { id: roomId },
          { projection: { members: 1, ownerId: 1 } }
        );
        if (!room) return;
        const isMember = room.ownerId === userId || room.members.some(m => m.userId === userId);
        if (!isMember) return;
      }

      // Join the socket room
      socket.join(`room:${roomId}`);
      socket.join(`user:${userId}`);

      // Track membership
      if (!roomSockets[roomId]) roomSockets[roomId] = new Map();
      roomSockets[roomId].set(socket.id, { userId, displayName: displayName || userId });

      // Broadcast join event to all OTHER members
      socket.to(`room:${roomId}`).emit('room:user_joined', { userId, displayName: displayName || userId });

      // Store on socket for cleanup
      socket._roomId  = roomId;
      socket._userId  = userId;
      socket._dName   = displayName || userId;

    } catch (err) {
      console.error('room:join error:', err.message);
    }
  });

  // ── room:message ─────────────────────────────────────────────────────────
  socket.on('room:message', async (msg) => {
    if (!msg || !msg.roomId || !msg.userId || !msg.body) return;
    if (typeof msg.body !== 'string' || msg.body.length > 300) return;

    try {
      const storedMsg = {
        id:          msg.id || ('msg_' + crypto.randomBytes(6).toString('hex')),
        userId:      msg.userId,
        displayName: msg.displayName || msg.userId,
        body:        msg.body.trim(),
        replyTo:     msg.replyTo || null,
        createdAt:   msg.createdAt || new Date().toISOString(),
      };

      // Persist message to DB (keep last 200 per room)
      await db.collection('rooms').updateOne(
        { id: msg.roomId },
        {
          $push: {
            messages: {
              $each:  [storedMsg],
              $slice: -200,
            },
          },
        }
      );

      // Broadcast to all members in the room (including sender for confirmation)
      io.to(`room:${msg.roomId}`).emit('room:message', storedMsg);

    } catch (err) {
      console.error('room:message error:', err.message);
    }
  });

  // ── disconnect cleanup ────────────────────────────────────────────────────
  socket.on('disconnect', () => {
    const roomId = socket._roomId;
    const userId = socket._userId;
    const dName  = socket._dName;
    if (roomId && userId) {
      if (roomSockets[roomId]) {
        roomSockets[roomId].delete(socket.id);
        if (roomSockets[roomId].size === 0) delete roomSockets[roomId];
      }
      socket.to(`room:${roomId}`).emit('room:user_left', { userId, displayName: dName });
    }
  });
});

// ═══════════════════════════════════════════════════════════════════════════
// BACKGROUND TIMERS
// ═══════════════════════════════════════════════════════════════════════════

// Auto-advance: check every second if current song has finished
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;
      // 1. Trusted HLS duration (_hlsDurationTrusted) — set after a clean full encode,
      //    most accurate for actual playback length
      if (current._hlsDurationTrusted && typeof current._hlsEnd === 'number' && typeof current._hlsStart === 'number') {
        songDuration = current._hlsEnd - current._hlsStart;
      // 2. meta.duration from ffprobe — reliable for most containers
      } else if (current.meta.duration > 0) {
        songDuration = current.meta.duration;
      // 3. Raw _hlsEnd - _hlsStart — only as last resort, may be truncated if FFmpeg
      //    was killed early
      } 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) {
        // Guard: only advance once per song. Check _advancingFromSid so two concurrent
        // ticks can't both shift the queue at the same moment, and also guard against a
        // tick firing after advanceToNextSong already moved to a different song (so queue[0]
        // is now different). If a concurrent tick or /next already shifted this song off
        // the queue (so queue[0] is now different), skip — it was already handled.
        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`);
        if (stream.users) { for (const [userId] of stream.users) { trackSongListened(userId).catch(console.error); } }
        // Always track for the stream owner (streamId === ownerId === userId) even if they
        // aren't in stream.users (e.g. they are watching via a constituent server)
        if (!stream.users || !stream.users.has(streamId)) {
          trackSongListened(streamId).catch(console.error);
        }
        advanceToNextSong(streamId, true);
        if (stream._advancingFromSid === current._sid) delete stream._advancingFromSid;
      }
    }
  } catch (err) { console.error('Auto-advance timer error:', err); }
}, 1000);

// Background segment pruning (every 30s)
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);

// Heartbeat timeout — evict users with no heartbeat in 30s (check every 15s)
setInterval(async () => {
  try {
    const now = new Date();
    for (const streamId in streams) {
      const stream = streams[streamId];
      if (!stream.users) continue;
      const usersToRemove = [];
      for (const [userId, user] of stream.users) {
        const timeSinceHeartbeat = (now - new Date(user.lastHeartbeat || user.joinedAt)) / 1000;
        if (timeSinceHeartbeat > 30) {
          const listenDuration = (Date.now() - user.connectionStart) / 1000;
          if (!user.isGuest) await trackListeningSession(userId, listenDuration);
          usersToRemove.push({ userId, user });
        }
      }
      for (const { userId, user } of usersToRemove) {
        stream.users.delete(userId);
        if (!user.isGuest) delete activeViewers[userId];
        broadcastUserUpdate(streamId, 'user_left', { id: userId, name: user.name, reason: 'timeout' }, true);
      }
    }
  } catch (err) { console.error('Heartbeat timer error:', err); }
}, 15000);

// Stream inactivity cleanup (every 10 min)
setInterval(() => {
  const now = Date.now(), toDelete = [];
  for (const streamId in streams) {
    const stream = streams[streamId];
    // Never evict a stream that is still actively downloading/encoding media
    if (stream._showplayInProgress > 0) continue;
    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 {} }
    // Purge any remaining DB queue entries for this stream
    db.collection('stream_queue').deleteMany({ streamId }).catch(console.error);
    delete streams[streamId]; delete hlsState[streamId]; delete hlsMutex[streamId]; delete hlsGeneration[streamId];
    console.log(`🧹 Cleaned up stream: ${streamId}`);
  }
}, 10 * 60 * 1000);

// ═══════════════════════════════════════════════════════════════════════════
// LAUNCH
// ═══════════════════════════════════════════════════════════════════════════

connectDB().then(async () => {
  // On startup, clear any stale stream_queue entries left from a previous run.
  // In-memory queue is always empty at boot so any DB entries are orphaned.
  try {
    const purged = await db.collection('stream_queue').deleteMany({});
    if (purged.deletedCount > 0) console.log(`🗑️  Purged ${purged.deletedCount} stale stream_queue entries from previous run`);
  } catch (e) { console.warn('Could not purge stale stream_queue:', e.message); }

  const PORT = parseInt(process.env.PORT || '7860', 10);
  streamServer.listen(PORT, async () => {
    console.log(`🚀 Stream server + WS running on port ${PORT}`);
    try {
      const ipRes = await axios.get('https://api.ipify.org?format=json', { timeout: 5000 });
      console.log(`🌐 Public IP: ${ipRes.data.ip}`);
      console.log(`🔗 Server reachable at http://${ipRes.data.ip}:${PORT}`);
    } catch (e) {
      console.warn('⚠️  Could not fetch public IP:', e.message);
    }
  });
}).catch((err) => {
  console.error('Failed to connect to MongoDB:', err);
  process.exit(1);
});

process.once('SIGINT',  () => { streamServer.close(); process.exit(0); });
process.once('SIGTERM', () => { streamServer.close(); process.exit(0); });