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

426 lines
15 KiB
JavaScript

/**
* Signaling Server
*
* Handles two tracks: video and game audio.
* Also manages controller queue and CSS process restart for name changes.
* 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';
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}`;
const CONTAINER = process.env.CONTAINER_NAME || 'css-test';
const IDLE_TIMEOUT = 60000; // 60 seconds
const NAME_SALT = process.env.NAME_SALT || 'change-this-secret';
// How long to wait after relaunching CSS before granting control
const CSS_LAUNCH_WAIT = 15000;
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.get('/', (req, res) => res.sendFile(path.join(__dirname, '..', 'client.html')));
// ── 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);
}
async 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);
containerRestarting = true;
broadcastQueueState();
// Same person returning — no restart needed
if (entry.visitorName === lastControllerName) {
console.log(`[sig] same controller returning, skipping restart`);
containerRestarting = false;
currentController = entry;
resetIdleTimer(entry);
broadcastQueueState();
if (entry.ws.readyState === WebSocket.OPEN) {
entry.ws.send(JSON.stringify({ type: 'control_granted' }));
}
return;
}
// Different person — update rev.ini and relaunch CSS only
lastControllerName = entry.visitorName;
console.log(`[sig] applying controller: ${entry.visitorName}`);
// Update rev.ini
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);
}
// Kill CSS and revLoader (leave Xvfb, PulseAudio, gst_agent running)
try {
execSync(`docker exec ${CONTAINER} pkill -f "cstrike_win64.exe" || true`);
execSync(`docker exec ${CONTAINER} pkill -f "revLoader.exe" || true`);
execSync(`docker exec ${CONTAINER} pkill -f "wineserver" || true`);
execSync(`docker exec ${CONTAINER} pkill -f "winedevice.exe" || true`);
console.log('[sig] CSS processes killed');
} catch (e) {
console.error('[sig] failed to kill CSS:', e.message);
}
// Wait for Wine to clean up
await new Promise(r => setTimeout(r, 2000));
// Relaunch CSS
try {
execSync(`docker exec -d ${CONTAINER} bash -c "DISPLAY=:99 wine '/home/ubuntu/Counter-Strike Source/revLoader.exe'"`);
console.log('[sig] CSS relaunched');
} catch (e) {
console.error('[sig] failed to relaunch CSS:', e.message);
}
// Wait for CSS to launch and connect to server
await new Promise(r => setTimeout(r, CSS_LAUNCH_WAIT));
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();
}
}
// ── 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);
});
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 === '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}`));