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

/**
 * Signaling Server
 *
 * Handles two tracks: video and game audio.
 * 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 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}`;

if (!CF_APP_ID || !CF_APP_TOKEN) {
  console.error('[sig] CF_APP_ID and CF_APP_TOKEN must be set');
  process.exit(1);
}

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

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

const agentSessions  = new Map();
const browserToAgent = new Map();

wss.on('connection', ws => {
  let role           = null;
  let agentSessionId = null;
  let browserSessId  = null;

  ws.on('message', async raw => {
    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} video=${msg.videoTrackName} audio=${msg.audioTrackName}`);
    }

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

    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' && browserSessId) {
      browserToAgent.delete(browserSessId);
    }
  });
});

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