webclient-css/streaming-agent/signaling-server.js

/**
 * Signaling Server
 *
 * Handles two tracks: video and game audio.
 * Also manages controller queue.
 * All Cloudflare API calls happen server-side.
 *
 * Run: CF_APP_ID=xxx CF_APP_TOKEN=xxx NAME_SALT=xxx CONTAINER_NAME=css-test node signaling-server.js
 */

'use strict';

// Prefix every console.log / console.error line with an ISO timestamp.
// Wrapping here means every existing log call site automatically gets a
// timestamp without needing to be touched individually.
const _origLog   = console.log;
const _origError = console.error;
console.log   = (...args) => _origLog(`[${new Date().toISOString()}]`, ...args);
console.error = (...args) => _origError(`[${new Date().toISOString()}]`, ...args);

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 dns        = require('dns').promises;

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}`;
const CONTAINER    = process.env.CONTAINER_NAME || 'css-test';
const IDLE_TIMEOUT = 60000; // 60 seconds
const NAME_SALT    = process.env.NAME_SALT || 'change-this-secret';

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',
    headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${CF_APP_TOKEN}` },
    body: JSON.stringify(body),
  });
  const json = await res.json();
  if (json.errorCode) throw new Error(`CF API error: ${json.errorDescription}`);
  return json;
}

async function cfPut(path, body) {
  const res = await fetch(`${CF_BASE}${path}`, {
    method: 'PUT',
    headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${CF_APP_TOKEN}` },
    body: JSON.stringify(body),
  });
  const json = await res.json();
  if (json.errorCode) throw new Error(`CF API error: ${json.errorDescription}`);
  return json;
}

// ── Express + WebSocket ───────────────────────────────────────────────────────
const app    = express();
const server = http.createServer(app);
const wss    = new WebSocket.Server({ server });

app.use(express.json());
app.use(express.urlencoded({ extended: false }));
app.get('/', (req, res) => res.sendFile(path.join(__dirname, '..', 'client.html')));

// ── Internal-request gate ─────────────────────────────────────────────────────
// Used by /webclient-name (POST), /webclient-current-ip (GET), and
// /webclient-force-restart (POST) — all internal-only endpoints intended for
// the SourceMod plugin (and, for restart, an optional safety-net trigger).
// Restricted to loopback / docker-bridge / private ranges, PLUS this box's
// own resolved public IP.
//
// The public-IP allowance is required because SteamWorks' embedded HTTP
// client (or the Source Engine networking layer beneath it) does NOT route
// outbound requests through true OS loopback the way a normal HTTP library
// would — even when targeting 127.0.0.1, the connection is sourced from
// whatever interface the game server process is bound to for its own
// traffic, which is the public interface. This was confirmed directly from
// logs: requests aimed at 127.0.0.1 arrived showing the box's public IP as
// the remote address. A strict "127.0.0.1 only" check would silently reject
// the SourceMod plugin's own requests.
//
// This is still safe: the TCP handshake proves the sender genuinely owns
// that IP (not spoofable over a real connection), so allowing this one
// specific, DNS-resolved address does not expose the endpoint to the public
// internet — nobody else can complete a TCP handshake claiming to be this
// server's own IP.
const SERVER_HOSTNAME = process.env.SERVER_HOSTNAME || 'ze.unloze.com';
let resolvedServerIp = null;

async function refreshServerPublicIp() {
  try {
    const { address } = await dns.lookup(SERVER_HOSTNAME);
    resolvedServerIp = address;
    console.log(`[sig] resolved ${SERVER_HOSTNAME} -> ${address}`);
  } catch (e) {
    console.error(`[sig] failed to resolve ${SERVER_HOSTNAME}:`, e.message);
  }
}

refreshServerPublicIp();

function isInternalRequest(req) {
  const ip = req.socket.remoteAddress || '';
  // Normalise IPv4-mapped IPv6 addresses (::ffff:127.0.0.1 -> 127.0.0.1)
  const normalized = ip.replace('::ffff:', '');
  return normalized === '127.0.0.1' ||
         normalized === '::1' ||
         normalized.startsWith('172.') ||  // docker bridge network range
         normalized.startsWith('10.') ||   // private network range, adjust as needed
         (resolvedServerIp && normalized === resolvedServerIp);
}

// SourceMod plugin -> signaling server: reports the name the connect
// extension generated for the currently-connected webclient player, so the
// browser queue panel and controller-restart-skip logic stay in sync with
// the actual in-game name/SteamID.
app.post('/webclient-name', (req, res) => {
  if (!isInternalRequest(req)) {
    console.log(`[sig] rejected /webclient-name from non-internal IP: ${req.socket.remoteAddress}`);
    return res.status(403).send('forbidden');
  }

  const name = (req.body && req.body.name) ? String(req.body.name).trim() : '';
  const ip   = (req.body && req.body.ip)   ? String(req.body.ip).trim()   : '';
  if (!name) {
    return res.status(400).send('missing name');
  }

  // Only apply this name if it corresponds to the CURRENTLY active
  // controller's real IP. Without this check, a delayed/stale POST from a
  // previous webclient session (the connect extension's OnClientAuthorized
  // can fire with a noticeable delay after the actual in-game reconnect)
  // can land AFTER a new person has already taken over control, silently
  // overwriting the new controller's correct name with the old one's.
  if (!currentController || !ip || currentController.ws._realIp !== ip) {
    console.log(`[sig] ignored stale /webclient-name (ip=${ip || 'none'}) — does not match current controller`);
    return res.status(200).send('ignored (stale)');
  }

  console.log(`[sig] webclient-name received: ${name}`);
  lastControllerName = name;
  currentController.visitorName = name;
  broadcastQueueState();

  res.status(200).send('ok');
});

// SourceMod plugin -> signaling server (polled): returns the current
// controller's real IP as plain text, so the plugin can detect when it
// changes and force the already-connected webclient player to reconnect
// via ClientCommand(client, "retry") — the connect extension backfills the
// real password from this same cached IP via OnClientPreConnectEx.
// Returns an empty body when nobody is currently controlling.
app.get('/webclient-current-ip', (req, res) => {
  if (!isInternalRequest(req)) {
    console.log(`[sig] rejected /webclient-current-ip from non-internal IP: ${req.socket.remoteAddress}`);
    return res.status(403).send('forbidden');
  }

  const ip = currentController ? (currentController.ws._realIp || '') : '';
  res.status(200).send(ip);
});

// Internal-only safety net: force-kill and relaunch CSS on demand. Intended
// for use by the SourceMod plugin (or any other trusted internal caller) if
// it detects the game client itself has wedged/crashed and needs a hard
// restart, separate from the manual in-browser restart button and the
// scheduled 5-hour restart.
app.post('/webclient-force-restart', (req, res) => {
  if (!isInternalRequest(req)) {
    console.log(`[sig] rejected /webclient-force-restart from non-internal IP: ${req.socket.remoteAddress}`);
    return res.status(403).send('forbidden');
  }

  console.log('[sig] /webclient-force-restart triggered');
  performCssRestart('force-restart endpoint');
  res.status(200).send('ok');
});

// ── Session registry ──────────────────────────────────────────────────────────
const agentSessions  = new Map();
const browserToAgent = new Map();
const pacatProcesses = new Map();

// ── Controller queue ──────────────────────────────────────────────────────────
let currentController   = null;
let lastControllerName  = null;
let idleTimer           = null;
const controlQueue      = [];
let containerRestarting = false;

function broadcastQueueState() {
  const controllerName = currentController ? currentController.visitorName : null;
  const queueNames = controlQueue
    .map(e => e.visitorName)
    .filter(name => name !== controllerName);
  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);
  });
}

function resetIdleTimer(entry) {
  if (idleTimer) clearTimeout(idleTimer);
  idleTimer = setTimeout(() => {
    console.log(`[sig] controller ${entry.visitorName} idle timeout`);
    releaseControl(entry, 'idle');
  }, IDLE_TIMEOUT);
}

function applyController(entry) {
  // Defensive: if this exact entry is sitting in the queue (e.g. a race
  // between request_control calls), pull it out so it's never displayed
  // as both controller and queued.
  const qIdx = controlQueue.indexOf(entry);
  if (qIdx !== -1) controlQueue.splice(qIdx, 1);

  // Granting control is now instant — no CSS restart, no GameMenu.res
  // rewrite, no automated click sequence. The SourceMod plugin picks up the
  // new controller's IP by polling GET /webclient-current-ip and forces the
  // in-game reconnect itself via ClientCommand(client, "retry").
  lastControllerName = entry.visitorName;
  currentController   = entry;
  resetIdleTimer(entry);
  broadcastQueueState();

  console.log(`[sig] applying controller: ${entry.visitorName}`);

  if (entry.ws.readyState === WebSocket.OPEN) {
    entry.ws.send(JSON.stringify({ type: 'control_granted' }));
  }
}

// ── CSS restart (shared by scheduled timer, manual button, and the internal
//    force-restart endpoint) ──────────────────────────────────────────────────
function performCssRestart(reason) {
  console.log(`[sig] CSS restart starting (${reason})`);
  containerRestarting = true;
  broadcastQueueState();

  // Use Wine's own proper teardown instead of pkill + a fixed sleep guess.
  // `wineserver -k` signals every Wine process under this prefix to exit;
  // `wineserver -w` BLOCKS until wineserver has actually fully released all
  // its shared memory / locks / sockets, rather than gambling on a fixed
  // delay being long enough. This is what previously let a fresh CSS
  // instance start before the old one had genuinely finished tearing down,
  // inheriting a wedged state and landing right back on the same stuck
  // "Connecting to server..." screen even after repeated restarts.
  // Wrapped in `timeout` so a manual restart can never hang indefinitely if
  // wineserver somehow never exits — after 15s we give up waiting and
  // proceed to relaunch anyway.
  try {
    execSync(`docker exec ${CONTAINER} bash -c "wineserver -k; timeout 15 wineserver -w || true"`);
    console.log('[sig] CSS restart: wineserver fully torn down');
  } catch (e) {
    console.error('[sig] CSS restart: wineserver teardown failed:', e.message);
  }

  // No artificial delay needed here anymore — wineserver -w above already
  // blocked until teardown was genuinely complete (or the 15s timeout was
  // hit), so we can relaunch immediately rather than adding another
  // arbitrary fixed wait on top of a real synchronization guarantee.
  (() => {
    try {
      execSync(`docker exec -d ${CONTAINER} bash -c "DISPLAY=:99 wine '/home/ubuntu/Counter-Strike Source/revLoader.exe'"`);
      console.log('[sig] CSS restart: relaunched');
    } catch (e) {
      console.error('[sig] CSS restart: failed to relaunch CSS:', e.message);
    }
    containerRestarting = false;
    broadcastQueueState();

    // Explicitly give the newly-created CSS window keyboard focus once it
    // has had time to map. Mouse clicks/movement always work after a
    // restart because they target whatever window is physically under the
    // cursor — but keyboard input goes to whichever window holds X11 input
    // focus, which is separate state that does NOT automatically transfer
    // to a freshly-relaunched window after a kill+relaunch (as opposed to a
    // genuinely fresh X session on container boot). Without this, keyboard
    // input can silently keep targeting a stale/dead window reference while
    // everything else on the new window works fine — this was very likely
    // the actual cause of "keyboard stuck, mouse/video fine" after restarts.
    setTimeout(() => {
      try {
        execSync(`docker exec ${CONTAINER} bash -c "DISPLAY=:99 WIN=\$(xdotool search --name '' | head -1) && xdotool windowfocus --sync \$WIN" || true`);
        console.log('[sig] CSS restart: focused new window');
      } catch (e) {
        console.error('[sig] CSS restart: failed to focus new window:', e.message);
      }
    }, 8000);
  })();
}

// Periodic restart: controller handoffs no longer restart the game at all,
// so CSS stays running indefinitely across many sessions. Long Wine
// sessions have historically accumulated issues (audio/keyboard corruption,
// memory growth, wedged connection state), so restart on a fixed schedule
// regardless of who's currently in control.
const CSS_RESTART_INTERVAL = 5 * 60 * 60 * 1000; // 5 hours
setInterval(() => performCssRestart('scheduled 5-hour restart'), CSS_RESTART_INTERVAL);

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

  // Tell gst_agent to release all held keys and reset X11 modifier state.
  // Without this, stuck keys (shift/ctrl/w/etc.) persist across controller
  // handoffs and eventually corrupt xdotool's input state, requiring a
  // full container restart to recover.
  const agent = [...agentSessions.values()][0];
  if (agent && agent.agentWs && agent.agentWs.readyState === WebSocket.OPEN) {
    agent.agentWs.send(JSON.stringify({ type: 'release_keys' }));
  }

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

// ── IP → Name (server side, salt never exposed to browser) ───────────────────
function hashString(str) {
  let hash = 5381;
  for (let i = 0; i < str.length; i++) {
    hash = ((hash << 5) + hash) + str.charCodeAt(i);
    hash = hash & 0x7fffffff;
  }
  return hash;
}

function ipToName(ip) {
  const adjectives = ['Fast','Slow','Red','Blue','Green','Dark','Bright','Wild',
    'Cool','Hot','Big','Small','Sharp','Brave','Swift','Bold'];
  const nouns = ['Wolf','Eagle','Tiger','Bear','Fox','Lion','Hawk','Shark',
    'Snake','Raven','Storm','Rock','Fire','Ice','Wind','Steel'];
  const salted = ip + NAME_SALT;
  const h1 = hashString(salted);
  const h2 = hashString(salted + '1');
  const h3 = hashString(salted + '2');
  return `${adjectives[h1 % adjectives.length]} ${nouns[h2 % nouns.length]} ${h3 % 9999}`;
}

// ── 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);
  });
  // Prevent EPIPE from crashing the server when pacat exits while we're
  // still writing to its stdin (e.g. mic frames arriving after disconnect).
  pacat.stdin.on('error', err => {
    if (err.code !== 'EPIPE') console.error(`[pacat:${browserSessId}] stdin error: ${err.message}`);
    pacatProcesses.delete(browserSessId);
  });
  pacatProcesses.set(browserSessId, pacat);
  return pacat;
}

function stopPacat(browserSessId) {
  const pacat = pacatProcesses.get(browserSessId);
  if (pacat) { pacat.kill(); pacatProcesses.delete(browserSessId); }
}

// ── WebSocket handler ─────────────────────────────────────────────────────────
wss.on('connection', (ws, req) => {
  // Capture the real visitor IP at connection time from nginx headers.
  // ws._socket.remoteAddress would be nginx's local IP for everyone.
  const forwarded = req.headers['x-real-ip'] ||
                    req.headers['x-forwarded-for'] || '';
  ws._realIp = forwarded.split(',')[0].trim() ||
               req.socket.remoteAddress.replace('::ffff:', '');

  let role           = null;
  let agentSessionId = null;
  let browserSessId  = null;
  let visitorName    = null;

  ws.on('message', async (raw, isBinary) => {
    if (isBinary) {
      if (browserSessId) {
        let pacat = pacatProcesses.get(browserSessId);
        if (!pacat) pacat = startPacat(browserSessId);
        if (pacat && pacat.stdin.writable) pacat.stdin.write(raw);
      }
      return;
    }

    let msg;
    try { msg = JSON.parse(raw); } catch { return; }

    if (msg.type === 'register') {
      role           = 'agent';
      agentSessionId = msg.sessionId;
      agentSessions.set(agentSessionId, {
        agentSessionId: msg.sessionId,
        videoTrackName: msg.videoTrackName,
        audioTrackName: msg.audioTrackName,
        agentWs:        ws,
      });
      console.log(`[sig] agent registered — sessionId=${agentSessionId}`);
    }

    else if (msg.type === 'request_control') {
      // Use the server-assigned name only — never trust the client-supplied one.
      // If get_ip hasn't been called yet, reject the request.
      if (!ws._assignedName) {
        console.log('[sig] request_control rejected — no server-assigned name yet');
        return;
      }
      visitorName = ws._assignedName;

      if (containerRestarting) return;

      if (currentController && currentController.ws === ws) {
        resetIdleTimer(currentController);
        return;
      }

      if (controlQueue.find(e => e.ws === ws)) return;

      // Prevent someone with the same IP as the current controller from
      // queuing behind themselves (duplicate tab or same-IP reconnect).
      if (currentController && currentController.visitorName === visitorName) {
        console.log(`[sig] ${visitorName} already controlling, ignoring duplicate request_control`);
        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 === 'restart_css') {
      // Manual restart button — only the CURRENT controller may trigger
      // this. Anyone else's message is silently ignored.
      if (currentController && currentController.ws === ws) {
        console.log(`[sig] manual CSS restart requested by ${currentController.visitorName}`);
        performCssRestart(`manual button — ${currentController.visitorName}`);
      } else {
        console.log('[sig] restart_css ignored — sender is not the current controller');
      }
    }

    else if (msg.type === 'mic_stop') {
      if (browserSessId) stopPacat(browserSessId);
    }

    else if (msg.type === 'get_ip') {
      const name = ipToName(ws._realIp);
      // Store server-assigned name on the socket so request_control can use
      // it exclusively — the client-supplied visitorName is never trusted.
      ws._assignedName = name;
      visitorName = name;
      ws.send(JSON.stringify({ type: 'your_ip', ip: name }));
    }

    else if (msg.type === 'browser_connect') {
      role = 'browser';
      const entry = [...agentSessions.values()][0];
      if (!entry) { ws.send(JSON.stringify({ type: 'waiting' })); return; }

      let newSession;
      try {
        newSession = await cfPost('/sessions/new', {
          sessionDescription: { type: 'offer', sdp: msg.sdp },
        });
      } catch (e) {
        ws.send(JSON.stringify({ type: 'error', message: e.message }));
        return;
      }

      browserSessId = newSession.sessionId;
      browserToAgent.set(browserSessId, entry.agentSessionId);
      ws.send(JSON.stringify({
        type:      'session_created',
        sessionId: browserSessId,
        sdpAnswer: newSession.sessionDescription,
      }));

      broadcastQueueState();
    }

    else if (msg.type === 'pull_track') {
      const agentId = browserToAgent.get(msg.sessionId);
      const entry   = agentSessions.get(agentId);
      if (!entry) { ws.send(JSON.stringify({ type: 'error', message: 'no agent session' })); return; }

      let trackResult;
      try {
        trackResult = await cfPost(`/sessions/${msg.sessionId}/tracks/new`, {
          tracks: [
            { location: 'remote', sessionId: entry.agentSessionId, trackName: entry.videoTrackName },
            { location: 'remote', sessionId: entry.agentSessionId, trackName: entry.audioTrackName },
          ],
        });
      } catch (e) {
        ws.send(JSON.stringify({ type: 'error', message: e.message }));
        return;
      }
      ws.send(JSON.stringify({ type: 'track_ready', trackResult }));
    }

    else if (msg.type === 'renegotiate') {
      try {
        await cfPut(`/sessions/${msg.sessionId}/renegotiate`, {
          sessionDescription: { type: 'answer', sdp: msg.sdp },
        });
      } catch (e) {
        ws.send(JSON.stringify({ type: 'error', message: e.message }));
      }
    }

    else if (msg.type === 'input') {
      const agentId = browserToAgent.get(msg.sessionId);
      const entry   = agentSessions.get(agentId);
      if (entry && entry.agentWs && entry.agentWs.readyState === WebSocket.OPEN) {
        entry.agentWs.send(JSON.stringify(msg.event));
      }
    }
  });

  ws.on('close', () => {
    if (role === 'agent' && agentSessionId) {
      agentSessions.delete(agentSessionId);
      console.log(`[sig] agent disconnected — sessionId=${agentSessionId}`);
    }
    if (role === 'browser') {
      if (currentController && currentController.ws === ws) {
        releaseControl(currentController, 'disconnect');
      }
      removeFromQueue(ws);
      stopPacat(browserSessId);
      browserToAgent.delete(browserSessId);
    }
  });
});

server.listen(PORT, () => console.log(`[sig] signaling server listening on port ${PORT}`));