Update server.js

server.js

@@ -72,6 +72,11 @@ async function connectDB() {
     { 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 });
@@ -1201,6 +1206,54 @@ async function resolveStreamId(id) {
   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) => {
@@ -1208,37 +1261,17 @@ streamApp.post('/joined/:streamId', async (req, res) => {
   const streamId = await resolveStreamId(rawId);
   const { name, id } = req.body;
   if (!name || !id) return res.status(400).json({ error: 'Name and id are required' });
-  const stream = streams[streamId];
 
-  // ── If stream not in main pool, check constituent servers ─────────────────
-  if (!stream) {
-    try {
-      const constituents = await db.collection('constituents').find({ userId: streamId }).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: 5000 }
-          );
-          if (r.data && r.data.success) {
-            // Constituent has this stream — proxy the /joined call to it
-            const proxyRes = await axios.post(
-              `${c.spaceUrl}/joined/${streamId}`,
-              { name, id, isGuest: req.body.isGuest },
-              { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 5000 }
-            );
-            return res.json({ ...proxyRes.data, proxiedToConstituent: c.spaceName });
-          }
-        } catch {
-          // This constituent unreachable or doesn't have the stream — try next
-        }
-      }
-    } catch (err) {
-      console.error('/joined constituent fallback error:', err.message);
-    }
-    return res.status(404).json({ error: 'Stream not found' });
-  }
+  // 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_');
@@ -1258,33 +1291,16 @@ streamApp.post('/left/:streamId', async (req, res) => {
   const streamId = await resolveStreamId(rawId);
   const { id } = req.body;
   if (!id) return res.status(400).json({ error: 'User id is required' });
-  const stream = streams[streamId];
 
-  // ── If stream not in main pool, proxy to constituent ──────────────────────
-  if (!stream) {
-    try {
-      const constituents = await db.collection('constituents').find({ userId: streamId }).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: 5000 }
-          );
-          if (r.data && r.data.success) {
-            const proxyRes = await axios.post(
-              `${c.spaceUrl}/left/${streamId}`,
-              { id },
-              { headers: { 'x-constituent-secret': MAIN_SERVER_SECRET }, timeout: 5000 }
-            );
-            return res.json({ ...proxyRes.data, proxiedToConstituent: c.spaceName });
-          }
-        } catch { /* try next */ }
-      }
-    } catch (err) {
-      console.error('/left constituent fallback error:', err.message);
-    }
-    return res.status(404).json({ error: 'Stream not found' });
-  }
+  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' });
@@ -1304,6 +1320,14 @@ streamApp.post('/left/:streamId', async (req, res) => {
 // 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
@@ -1316,6 +1340,14 @@ streamApp.get('/listeners/:streamId', async (req, res) => {
 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();
@@ -1323,23 +1355,48 @@ streamApp.get('/list/:streamId', async (req, res) => {
 });
 
 // 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) { user.lastHeartbeat = new Date().toISOString(); }
+  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 });
 });
@@ -1347,6 +1404,14 @@ streamApp.post('/heartbeat/:streamId', async (req, res) => {
 // 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];
@@ -1372,6 +1437,14 @@ streamApp.get('/stream/:streamId/currentTrack', async (req, res) => {
 // 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];
@@ -1721,6 +1794,7 @@ streamApp.post('/api/play', requireRegistered, async (req, res) => {
 
     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) {
@@ -1758,6 +1832,7 @@ streamApp.post('/api/vplay', requireRegistered, async (req, res) => {
 
     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) {
@@ -1977,6 +2052,8 @@ async function showplayEnqueueLink(streamId, pendingLink, pendingTitle, thumbnai
   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".
@@ -2207,6 +2284,14 @@ streamApp.post('/api/skip', requireRegistered, (req, res) => {
 // 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' });
 
@@ -2256,6 +2341,60 @@ streamApp.post('/api/internal/track-song', async (req, res) => {
   }
 });
 
+// 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;
@@ -3055,6 +3194,9 @@ streamApp.post('/api/constituent/add-movie', requireRegistered, async (req, res)
       { 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',
@@ -3101,6 +3243,9 @@ streamApp.post('/api/constituent/add-song', requireRegistered, async (req, res)
       { 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',