added support for usernames through ip or fingerprint, and also control based on it and restarting sessions

streaming-agent/signaling-server.js

@@ -2,7 +2,7 @@
  * Signaling Server
  *
  * Handles two tracks: video and game audio.
- * Also relays browser microphone audio to container PulseAudio via pacat.
+ * Also manages controller queue and container restart for name changes.
  * All Cloudflare API calls happen server-side.
  *
  * Run: CF_APP_ID=xxx CF_APP_TOKEN=xxx node signaling-server.js
@@ -10,34 +10,26 @@
 
 'use strict';
 
-const express   = require('express');
-const http      = require('http');
-const WebSocket = require('ws');
-const fetch     = require('node-fetch');
-const path      = require('path');
-const { spawn } = require('child_process');
+const express    = require('express');
+const http       = require('http');
+const WebSocket  = require('ws');
+const fetch      = require('node-fetch');
+const path       = require('path');
+const { spawn, execSync } = require('child_process');
 
 const PORT         = process.env.PORT         || 3000;
 const CF_APP_ID    = process.env.CF_APP_ID    || '';
 const CF_APP_TOKEN = process.env.CF_APP_TOKEN || '';
 const CF_BASE      = `https://rtc.live.cloudflare.com/v1/apps/${CF_APP_ID}`;
-
-// Container's PulseAudio sink to write mic audio into
-// pacat writes raw PCM to the null sink which virtual_mic reads from
-const PACAT_ARGS = [
-  '--playback',
-  '--device=mic_sink',
-  '--format=s16le',
-  '--rate=48000',
-  '--channels=1',
-  '--latency-msec=20',
-];
+const CONTAINER    = process.env.CONTAINER_NAME || 'css-test';
+const IDLE_TIMEOUT = 10000; // 10 seconds
 
 if (!CF_APP_ID || !CF_APP_TOKEN) {
   console.error('[sig] CF_APP_ID and CF_APP_TOKEN must be set');
   process.exit(1);
 }
 
+// ── Cloudflare API helpers ────────────────────────────────────────────────────
 async function cfPost(path, body) {
   const res = await fetch(`${CF_BASE}${path}`, {
     method: 'POST',
@@ -60,6 +52,7 @@ async function cfPut(path, body) {
   return json;
 }
 
+// ── Express + WebSocket ───────────────────────────────────────────────────────
 const app    = express();
 const server = http.createServer(app);
 const wss    = new WebSocket.Server({ server });
@@ -67,67 +60,163 @@ const wss    = new WebSocket.Server({ server });
 app.use(express.json());
 app.get('/', (req, res) => res.sendFile(path.join(__dirname, '..', 'client.html')));
 
+// ── Session registry ──────────────────────────────────────────────────────────
 const agentSessions  = new Map();
 const browserToAgent = new Map();
-
-// Per-browser pacat process for mic audio
-// browserSessionId -> pacat child process
 const pacatProcesses = new Map();
 
-function startPacat(browserSessId) {
-  // Find the container ID for this browser session
-  // pacat runs inside the Docker container via docker exec
-  const agentId = browserToAgent.get(browserSessId);
-  const entry   = agentSessions.get(agentId);
-  if (!entry) return null;
+// ── Controller queue ──────────────────────────────────────────────────────────
+let currentController   = null;
+let lastControllerName = null;
+let idleTimer           = null;
+const controlQueue      = [];
+let containerRestarting = false;
 
-  // Get container name from agent — agent sends it on register
-  const containerName = entry.containerName || 'css-test';
+function broadcastQueueState() {
+  const queueNames     = controlQueue.map(e => e.visitorName);
+  const controllerName = currentController ? currentController.visitorName : null;
+  const msg = JSON.stringify({
+    type:       'queue_state',
+    controller: controllerName,
+    queue:      queueNames,
+    restarting: containerRestarting,
+  });
+  wss.clients.forEach(ws => {
+    if (ws.readyState === WebSocket.OPEN) ws.send(msg);
+  });
+}
 
-  const pacat = spawn('docker', [
-    'exec', '-i', containerName,
-    'pacat', ...PACAT_ARGS,
-  ]);
+function resetIdleTimer(entry) {
+  if (idleTimer) clearTimeout(idleTimer);
+  idleTimer = setTimeout(() => {
+    console.log(`[sig] controller ${entry.visitorName} idle timeout`);
+    releaseControl(entry, 'idle');
+  }, IDLE_TIMEOUT);
+}
 
-  pacat.stderr.on('data', d => {
-    console.error(`[pacat:${browserSessId}]`, d.toString().trim());
+async function applyController(entry) {
+  containerRestarting = true;
+  broadcastQueueState();
+
+  if (entry.visitorName === lastControllerName) {
+    // Same person returning — no restart needed
+    containerRestarting = false;
+    currentController   = entry;
+    lastControllerName  = entry.visitorName;
+    resetIdleTimer(entry);
+    broadcastQueueState();
+    if (entry.ws.readyState === WebSocket.OPEN) {
+      entry.ws.send(JSON.stringify({ type: 'control_granted' }));
+    }
+    return;
+  }
+
+  // Different person — update rev.ini and restart
+  lastControllerName = entry.visitorName;
+
+  try {
+    const safeName = entry.visitorName.replace(/'/g, "'\\''");
+    execSync(`docker exec ${CONTAINER} sed -i "s/PlayerName=.*/PlayerName=${safeName}/" "/home/ubuntu/Counter-Strike Source/rev.ini"`);
+    console.log(`[sig] rev.ini updated with name: ${entry.visitorName}`);
+  } catch (e) {
+    console.error('[sig] failed to update rev.ini:', e.message);
+  }
+
+  try {
+    execSync(`docker restart ${CONTAINER}`);
+    console.log('[sig] container restarted');
+  } catch (e) {
+    console.error('[sig] failed to restart container:', e.message);
+  }
+
+  // Wait for agent to re-register (up to 30 seconds)
+  await new Promise(resolve => {
+    const start = Date.now();
+    const check = setInterval(() => {
+      if (agentSessions.size > 0 || Date.now() - start > 30000) {
+        clearInterval(check);
+        resolve();
+      }
+    }, 500);
   });
 
+  containerRestarting = false;
+  currentController   = entry;
+  resetIdleTimer(entry);
+  broadcastQueueState();
+
+  if (entry.ws.readyState === WebSocket.OPEN) {
+    entry.ws.send(JSON.stringify({ type: 'control_granted' }));
+  }
+}
+
+function releaseControl(entry, reason) {
+  if (idleTimer) { clearTimeout(idleTimer); idleTimer = null; }
+  if (currentController !== entry) return;
+  currentController = null;
+
+  console.log(`[sig] control released from ${entry.visitorName} (${reason})`);
+
+  if (entry.ws.readyState === WebSocket.OPEN) {
+    entry.ws.send(JSON.stringify({ type: 'control_released', reason }));
+  }
+
+  if (controlQueue.length > 0) {
+    const next = controlQueue.shift();
+    applyController(next);
+  }
+
+  broadcastQueueState();
+}
+
+function removeFromQueue(ws) {
+  const idx = controlQueue.findIndex(e => e.ws === ws);
+  if (idx !== -1) {
+    console.log(`[sig] removed ${controlQueue[idx].visitorName} from queue`);
+    controlQueue.splice(idx, 1);
+    broadcastQueueState();
+  }
+}
+
+// ── pacat for mic audio ───────────────────────────────────────────────────────
+const PACAT_ARGS = [
+  '--playback',
+  '--device=mic_sink',
+  '--format=s16le',
+  '--rate=48000',
+  '--channels=1',
+  '--latency-msec=20',
+];
+
+function startPacat(browserSessId) {
+  const pacat = spawn('docker', ['exec', '-i', CONTAINER, 'pacat', ...PACAT_ARGS]);
+  pacat.stderr.on('data', d => console.error(`[pacat:${browserSessId}]`, d.toString().trim()));
   pacat.on('close', code => {
     console.log(`[pacat:${browserSessId}] exited with code ${code}`);
     pacatProcesses.delete(browserSessId);
   });
-
   pacatProcesses.set(browserSessId, pacat);
-  console.log(`[sig] pacat started for browser session ${browserSessId}`);
   return pacat;
 }
 
 function stopPacat(browserSessId) {
   const pacat = pacatProcesses.get(browserSessId);
-  if (pacat) {
-    pacat.kill();
-    pacatProcesses.delete(browserSessId);
-    console.log(`[sig] pacat stopped for browser session ${browserSessId}`);
-  }
+  if (pacat) { pacat.kill(); pacatProcesses.delete(browserSessId); }
 }
 
+// ── WebSocket handler ─────────────────────────────────────────────────────────
 wss.on('connection', ws => {
   let role           = null;
   let agentSessionId = null;
   let browserSessId  = null;
+  let visitorName    = null;
 
   ws.on('message', async (raw, isBinary) => {
-    // Binary messages are raw PCM audio from browser microphone
     if (isBinary) {
       if (browserSessId) {
         let pacat = pacatProcesses.get(browserSessId);
-        if (!pacat) {
-          pacat = startPacat(browserSessId);
-        }
-        if (pacat && pacat.stdin.writable) {
-          pacat.stdin.write(raw);
-        }
+        if (!pacat) pacat = startPacat(browserSessId);
+        if (pacat && pacat.stdin.writable) pacat.stdin.write(raw);
       }
       return;
     }
@@ -142,17 +231,45 @@ wss.on('connection', ws => {
         agentSessionId: msg.sessionId,
         videoTrackName: msg.videoTrackName,
         audioTrackName: msg.audioTrackName,
-        containerName:  msg.containerName || 'css-test',
         agentWs:        ws,
       });
-      console.log(`[sig] agent registered — sessionId=${agentSessionId} container=${msg.containerName || 'css-test'}`);
+      console.log(`[sig] agent registered — sessionId=${agentSessionId}`);
+    }
+
+    else if (msg.type === 'request_control') {
+      visitorName = msg.visitorName;
+
+      // Reject during restart
+      if (containerRestarting) return;
+
+      // Already controlling — just reset idle timer
+      if (currentController && currentController.ws === ws) {
+        resetIdleTimer(currentController);
+        return;
+      }
+
+      // Already in queue
+      if (controlQueue.find(e => e.ws === ws)) return;
+
+      const entry = { ws, visitorName, browserSessId };
+
+      if (!currentController) {
+        applyController(entry);
+      } else {
+        controlQueue.push(entry);
+        console.log(`[sig] ${visitorName} joined queue position ${controlQueue.length}`);
+        broadcastQueueState();
+      }
+    }
+
+    else if (msg.type === 'controller_active') {
+      if (currentController && currentController.ws === ws) {
+        resetIdleTimer(currentController);
+      }
     }
 
     else if (msg.type === 'mic_stop') {
-      // Browser released PTT key — stop pacat
-      if (browserSessId) {
-        stopPacat(browserSessId);
-      }
+      if (browserSessId) stopPacat(browserSessId);
     }
 
     else if (msg.type === 'browser_connect') {
@@ -177,6 +294,8 @@ wss.on('connection', ws => {
         sessionId: browserSessId,
         sdpAnswer: newSession.sessionDescription,
       }));
+
+      broadcastQueueState();
     }
 
     else if (msg.type === 'pull_track') {
@@ -216,6 +335,10 @@ wss.on('connection', ws => {
         entry.agentWs.send(JSON.stringify(msg.event));
       }
     }
+    else if (msg.type === 'get_ip') {
+      const ip = ws._socket.remoteAddress.replace('::ffff:', '');
+      ws.send(JSON.stringify({ type: 'your_ip', ip }));
+    }
   });
 
   ws.on('close', () => {
@@ -223,7 +346,11 @@ wss.on('connection', ws => {
       agentSessions.delete(agentSessionId);
       console.log(`[sig] agent disconnected — sessionId=${agentSessionId}`);
     }
-    if (role === 'browser' && browserSessId) {
+    if (role === 'browser') {
+      if (currentController && currentController.ws === ws) {
+        releaseControl(currentController, 'disconnect');
+      }
+      removeFromQueue(ws);
       stopPacat(browserSessId);
       browserToAgent.delete(browserSessId);
     }