332 lines
12 KiB
JavaScript
332 lines
12 KiB
JavaScript
// 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
|
|
// ICE_TRANSPORT_POLICY optional WebRTC policy (all, relay, or none)
|
|
// 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,
|
|
});
|
|
}
|
|
}
|
|
const config = { iceServers };
|
|
const iceTransportPolicy = process.env.ICE_TRANSPORT_POLICY;
|
|
if (iceTransportPolicy && ['all', 'relay', 'none'].includes(iceTransportPolicy)) {
|
|
config.iceTransportPolicy = iceTransportPolicy;
|
|
}
|
|
return config;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 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',
|
|
'.webmanifest': 'application/manifest+json',
|
|
'.svg': 'image/svg+xml',
|
|
'.png': 'image/png',
|
|
'.jpg': 'image/jpeg',
|
|
'.jpeg': 'image/jpeg',
|
|
'.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 };
|
|
|