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

385 lines
13 KiB
JavaScript

/**
* Signaling Server
*
* Handles two tracks: video and game audio.
* 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
*/
'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 = 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',
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 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);
});
}
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) {
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);
return pacat;
}
function stopPacat(browserSessId) {
const pacat = pacatProcesses.get(browserSessId);
if (pacat) { pacat.kill(); pacatProcesses.delete(browserSessId); }
}
const NAME_SALT = process.env.NAME_SALT || 'change-this-secret';
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}`;
}
// ── WebSocket handler ─────────────────────────────────────────────────────────
wss.on('connection', ws => {
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') {
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') {
if (browserSessId) stopPacat(browserSessId);
}
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));
}
}
else if (msg.type === 'get_ip') {
const ip = ws._socket.remoteAddress.replace('::ffff:', '');
const name = ipToName(ip);
ws.send(JSON.stringify({ type: 'your_ip', ip: name }));
}
});
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}`));