gibbz/server.js

// Minimal WebRTC signaling + static file server for the Discord-clone POC.
// Load OpenTelemetry tracing and logging before HTTP / network modules
require('./tracing');

const http = require('http');
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const { WebSocketServer } = require('ws');
const { trace } = require('@opentelemetry/api');

function traceId() {
  return trace.getActiveSpan()?.spanContext().traceId;
}

const PORT = Number(process.env.PORT || 8080);
const HOST = process.env.HOST || '0.0.0.0';
const STATIC_DIR = fs.existsSync(path.join(__dirname, 'dist'))
  ? path.join(__dirname, 'dist')
  : path.join(__dirname, 'public');

// ---------------------------------------------------------------------------
// ICE configuration: env-driven, full JSON override wins.
//   STUN_URLS           comma-separated stun: urls
//   TURN_URLS           comma-separated turn:/turns: urls
//   TURN_USERNAME / TURN_CREDENTIAL  static creds for all TURN urls
//   TURN_LAN_HOST / TURN_LAN_SECRET  optional second, LAN-only TURN relay
//                       (own coturn instance + secret, reachable only from
//                       the local network via its own LoadBalancer IP with
//                       no NAT/hairpin involved). Harmless to hand out to
//                       every client: anyone not on that network simply
//                       can't reach the address and ICE ignores it.
//   ICE_SERVERS_JSON    complete {"iceServers":[...]} override
// ---------------------------------------------------------------------------
function hmacTurnCredential(secret, label, ttl) {
  const expiry = Math.floor(Date.now() / 1000) + ttl;
  const username = `${expiry}:${label}`;
  const hmac = crypto.createHmac('sha1', secret);
  hmac.update(username);
  return { username, credential: hmac.digest('base64') };
}

function iceConfig() {
  if (process.env.ICE_SERVERS_JSON) {
    try {
      const parsed = JSON.parse(process.env.ICE_SERVERS_JSON);
      if (Array.isArray(parsed.iceServers)) return parsed;
      console.error('ICE_SERVERS_JSON missing iceServers array, falling back to env vars');
    } catch (err) {
      console.error('ICE_SERVERS_JSON invalid JSON:', err.message);
    }
  }
  const urls = (s) => (s || '').split(',').map((x) => x.trim()).filter(Boolean);
  const stunUrls = urls(process.env.STUN_URLS || 'stun:turn.k6n.net:3478,stun:stun.l.google.com:19302');
  const iceServers = stunUrls.map((u) => ({ urls: u }));

  const turnUrls = urls(process.env.TURN_URLS || 'turn:turn.k6n.net:3478?transport=udp,turn:turn.k6n.net:3478?transport=tcp,turns:turn.k6n.net:5349?transport=tcp');
  if (turnUrls.length) {
    let username = process.env.TURN_USERNAME || '';
    let credential = process.env.TURN_CREDENTIAL || process.env.TURN_PASSWORD || '';

    // If a TURN secret is provided (coturn use-auth-secret / static-auth-secret), generate ephemeral HMAC credentials
    const turnSecret = process.env.TURN_SECRET || process.env.TURN_STATIC_AUTH_SECRET || '';
    const ttl = Number(process.env.TURN_TTL || 86400);
    if (turnSecret) {
      ({ username, credential } = hmacTurnCredential(turnSecret, 'gibbz', ttl));
    }

    iceServers.push({
      urls: turnUrls,
      username,
      credential,
    });

    if (process.env.TURN_LAN_HOST && process.env.TURN_LAN_SECRET) {
      const { username: lanUsername, credential: lanCredential } =
        hmacTurnCredential(process.env.TURN_LAN_SECRET, 'gibbz-lan', ttl);
      iceServers.push({
        urls: [
          `turn:${process.env.TURN_LAN_HOST}:3478?transport=udp`,
          `turn:${process.env.TURN_LAN_HOST}:3478?transport=tcp`,
        ],
        username: lanUsername,
        credential: lanCredential,
      });
    }
  }
  return { iceServers };
}

// ---------------------------------------------------------------------------
// Static files
// ---------------------------------------------------------------------------
const MIME = {
  '.html': 'text/html; charset=utf-8',
  '.js': 'text/javascript; charset=utf-8',
  '.css': 'text/css; charset=utf-8',
  '.json': 'application/json',
  '.svg': 'image/svg+xml',
  '.png': 'image/png',
  '.ico': 'image/x-icon',
};

function serveStatic(req, res) {
  const urlPath = decodeURIComponent(new URL(req.url, 'http://x').pathname);
  let rel = urlPath === '/' ? '/index.html' : urlPath;
  const file = path.normalize(path.join(STATIC_DIR, rel));
  if (!file.startsWith(STATIC_DIR)) {
    res.writeHead(403).end('forbidden');
    return;
  }
  fs.readFile(file, (err, data) => {
    if (err) {
      res.writeHead(404, { 'Content-Type': 'text/plain' }).end('not found');
      return;
    }
    res.writeHead(200, { 'Content-Type': MIME[path.extname(file)] || 'application/octet-stream', 'Cache-Control': 'no-cache' });
    res.end(data);
  });
}

const server = http.createServer((req, res) => {
  const start = process.hrtime.bigint();
  const { method, url } = req;
  res.on('close', () => {
    const ms = Number(process.hrtime.bigint() - start) / 1e6;
    // res.writableEnded false here means the client (or ingress) dropped the
    // connection before we finished writing — the smoking gun for a 502 that
    // originates on the proxy side rather than in this process.
    if (!res.writableEnded) {
      console.warn(`[http] ABORTED ${method} ${url} after ${ms.toFixed(1)}ms (status=${res.statusCode}) trace=${traceId()}`);
    } else if (ms > 1000) {
      console.warn(`[http] SLOW ${method} ${url} ${res.statusCode} ${ms.toFixed(1)}ms trace=${traceId()}`);
    }
  });

  if (req.url === '/healthz' || req.url === '/readyz' || req.url === '/health') {
    res.writeHead(200, { 'Content-Type': 'text/plain' });
    res.end('ok');
    return;
  }
  if (req.url.startsWith('/api/ice')) {
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify(iceConfig()));
    return;
  }
  if (req.url.startsWith('/api/rooms')) {
    res.writeHead(200, { 'Content-Type': 'application/json', 'Cache-Control': 'no-cache' });
    res.end(JSON.stringify(allRoomsSnapshot()));
    return;
  }
  serveStatic(req, res);
});

// Node's default keepAliveTimeout (5s) is shorter than ingress-nginx's default
// upstream keepalive idle timeout (60s). Under that mismatch, nginx can pick a
// pooled connection Node has *just* closed and forward a request onto it,
// which comes back to the browser as a 502 — invisible from inside this
// process because Node never saw the request. headersTimeout must stay above
// keepAliveTimeout (Node enforces this).
server.keepAliveTimeout = 65_000;
server.headersTimeout = 66_000;

server.on('clientError', (err, socket) => {
  console.warn(`[http] clientError: ${err.message} (code=${err.code})`);
  if (socket.writable) socket.end('HTTP/1.1 400 Bad Request\r\n\r\n');
});

// ---------------------------------------------------------------------------
// Signaling
//   client -> server: {type: offer|answer|candidate, to}  {type: pos|state}
//   server -> client: {type: hello, id, peers}  {type: joined|left, ...}
//                     relayed messages get "from" stamped
// ---------------------------------------------------------------------------
const rooms = new Map(); // roomId -> Map(peerId -> client)

const wss = new WebSocketServer({ server, path: '/ws' });

function roomPeers(roomId) {
  let room = rooms.get(roomId);
  if (!room) {
    room = new Map();
    rooms.set(roomId, room);
  }
  return room;
}

function broadcast(room, msg, exceptId) {
  const data = JSON.stringify(msg);
  for (const [id, client] of room) {
    if (id !== exceptId && client.socket.readyState === 1) client.socket.send(data);
  }
}

function allRoomsSnapshot() {
  const out = {};
  for (const [rId, r] of rooms.entries()) {
    out[rId] = [...r.values()].map(snapshot);
  }
  return out;
}

function broadcastRosters() {
  const snap = allRoomsSnapshot();
  const data = JSON.stringify({ type: 'rosters', rooms: snap });
  for (const client of wss.clients) {
    if (client.readyState === 1) {
      client.send(data);
    }
  }
}

wss.on('connection', (socket, req) => {
  const q = new URL(req.url, 'http://x').searchParams;
  const roomId = (q.get('room') || 'general').slice(0, 64);
  const name = (q.get('name') || 'anon').slice(0, 32);
  const id = crypto.randomBytes(4).toString('hex');
  const room = roomPeers(roomId);
  const client = { socket, id, name, x: 50, y: 50, state: { muted: false, deafened: false, sharing: false } };
  room.set(id, client);

  const send = (msg) => {
    if (socket.readyState === 1) socket.send(JSON.stringify(msg));
  };

  send({ type: 'hello', id, room: roomId, peers: [...room.values()].filter((p) => p.id !== id).map(snapshot) });
  broadcast(room, { type: 'joined', peer: snapshot(client) }, id);
  broadcastRosters();
  console.log(`[signaling] ${name} (${id}) joined #${roomId} — ${room.size} in room`);

  socket.on('message', (raw) => {
    let msg;
    try {
      msg = JSON.parse(raw.toString());
    } catch {
      return;
    }
    if (!msg || typeof msg.type !== 'string') return;

    switch (msg.type) {
      case 'ping': {
        send({ type: 'pong' });
        break;
      }
      case 'pos': {
        const x = Number(msg.x), y = Number(msg.y);
        if (Number.isFinite(x) && Number.isFinite(y)) {
          client.x = Math.min(100, Math.max(0, x));
          client.y = Math.min(100, Math.max(0, y));
          broadcast(room, { type: 'pos', from: id, x: client.x, y: client.y }, id);
        }
        break;
      }
      case 'state': {
        client.state = {
          muted: !!msg.state?.muted,
          deafened: !!msg.state?.deafened,
          sharing: !!msg.state?.sharing,
        };
        broadcast(room, { type: 'state', from: id, state: client.state }, id);
        broadcastRosters();
        break;
      }
      case 'offer':
      case 'answer':
      case 'candidate':
      case 'reset': {
        if (!msg.to || !room.has(msg.to)) return;
        const target = room.get(msg.to);
        if (target.socket.readyState === 1) {
          const { to: _to, ...rest } = msg;
          target.socket.send(JSON.stringify({ ...rest, from: id }));
        }
        break;
      }
      default:
        break; // ignore unknown types — forward compat
    }
  });

  socket.on('close', () => {
    room.delete(id);
    if (room.size === 0) rooms.delete(roomId);
    else broadcast(room, { type: 'left', id });
    broadcastRosters();
    console.log(`[signaling] ${name} (${id}) left #${roomId} — ${room.size} in room`);
  });
});

function snapshot(c) {
  return { id: c.id, name: c.name, x: c.x, y: c.y, state: c.state };
}

// heartbeat: drop dead sockets
const interval = setInterval(() => {
  for (const room of rooms.values()) {
    for (const [, client] of room) {
      if (client.socket.readyState === 1) {
        client.socket.ping();
      }
    }
  }
}, 30000);
wss.on('close', () => clearInterval(interval));

function startServer(port = PORT, host = HOST) {
  return new Promise((resolve) => {
    server.listen(port, host, () => {
      const ice = iceConfig();
      console.log(`voice-poc listening on http://${host}:${port}`);
      console.log(`ICE servers: ${JSON.stringify(ice.iceServers)}`);
      resolve(server);
    });
  });
}

if (require.main === module) {
  startServer();
}

module.exports = { server, wss, iceConfig, allRoomsSnapshot, interval, startServer };