diff options
| -rw-r--r-- | live/server.js | 102 | ||||
| -rw-r--r-- | tools/live-ingest-test.js | 3 |
2 files changed, 103 insertions, 2 deletions
diff --git a/live/server.js b/live/server.js index 9bba15c..dfe122e 100644 --- a/live/server.js +++ b/live/server.js @@ -20,6 +20,7 @@ // GET /live/<id>/<run>-<n>.ts a segment // GET /live/<id>/stats {state, viewers, started, seconds} // GET /live/ping (WebSocket) says pong: sockets get through +// POST /live/<id>/rtc/… WebRTC through Cloudflare's SFU for the app's viewers (below) // GET /live/turn ICE servers for a call (Cloudflare TURN, a few hours' // credentials), Authorization: Nostr <NIP-98 for GET /live/turn> // GET /live/allowed?name= may this key go live? {ok} or {ok:false, reason, until} @@ -78,6 +79,26 @@ async function turnServers() { const j = await r.json(); return Array.isArray(j.iceServers) ? j.iceServers : [j.iceServers].filter(Boolean); } +// WebRTC for the app's own viewers, through Cloudflare's SFU (Realtime): +// the broadcaster sends its camera there as well as here (HLS stays, for the +// recording and the other nostr apps), and a viewer in the app pulls it from +// there — a fraction of a second behind, where HLS is 3.5 s. This server only +// brokers it: the app's secret ($LIVE_CONFIG {sfuAppId, sfuSecret}) stays +// here, and no media comes through cs (only 443 gets in, through Cloudflare). +const SFU = CONFIG.sfuAppId && CONFIG.sfuSecret ? `https://rtc.live.cloudflare.com/v1/apps/${CONFIG.sfuAppId}` : null; +async function sfu(path, method = 'POST', body) { + const r = await fetch(SFU + path, { + method, signal: AbortSignal.timeout(10_000), + headers: { authorization: 'Bearer ' + CONFIG.sfuSecret, 'content-type': 'application/json', 'user-agent': 'coinos-live/1' }, + body: body ? JSON.stringify(body) : undefined, + }); + const j = await r.json().catch(() => ({})); + const bad = j.errorCode || (j.tracks || []).find((t) => t.errorCode); + if (!r.ok || bad) throw new Error('sfu ' + r.status + ' ' + ((bad && (bad.errorDescription || bad.errorCode)) || '')); + return j; +} +const rtcAsks = new Map(); // ip -> [times] + // The admins (the remove button, report DMs); the DMs come from config.alertSk // (the registrar's welcome key) to the admins' inbox relays. const ADMINS = Array.isArray(CONFIG.admins) ? CONFIG.admins : ['98ae4da926c471c23fd12d1ebdd5839ba82917baa618e184e0c9916d93dcf4f7']; @@ -101,7 +122,7 @@ const idOk = (id) => /^[0-9a-f]{16}$/.test(id || ''); // ---------------------------------------------------------------- streams function metaPath(id) { return `${DATA}/${id}/stream.json`; } function save(s) { - const { ws, ff, viewers, timer, ...keep } = s; + const { ws, ff, viewers, timer, rtc, watchers, ...keep } = s; try { writeFileSync(metaPath(s.id), JSON.stringify(keep)); } catch {} } function load(id) { @@ -201,6 +222,7 @@ function dropped(s) { } function end(s, why) { endRun(s); + s.rtc = null; s.watchers = null; if (s.state === 'ended') return; s.state = 'ended'; s.ended = Date.now(); @@ -309,6 +331,9 @@ export async function handle(req, server) { if (a.error) return json({ error: a.error }, 401); return json(await mayStream((url.searchParams.get('name') || '').toLowerCase(), a.pubkey)); } + const rtc = /^\/([0-9a-f]{16})\/rtc\/(publish|watch|answer)$/.exec(path); + if (rtc && req.method === 'OPTIONS') return new Response(null, { headers: { 'access-control-allow-origin': '*', 'access-control-allow-headers': 'authorization, content-type', 'access-control-allow-methods': 'POST' } }); + if (rtc && req.method === 'POST') return rtcRoute(rtc[1], rtc[2], req, url, server); const act = /^\/([0-9a-f]{16})\/(report|remove)$/.exec(path); if (act && req.method === 'OPTIONS') return new Response(null, { headers: { 'access-control-allow-origin': '*', 'access-control-allow-headers': 'authorization, content-type', 'access-control-allow-methods': 'POST' } }); if (act && req.method === 'POST') { @@ -324,7 +349,11 @@ export async function handle(req, server) { const s = load(m[1]); if (!s) return json({ error: 'no such stream' }, 404); if (s.state === 'removed') return json({ error: 'removed', state: 'removed' }, 410); - if (m[2] === 'stats') return json({ state: s.state, viewers: viewersOf(s), started: s.started, seconds: Math.round(((s.ended || Date.now()) - s.started) / 1000) }); + if (m[2] === 'stats') { + // a viewer on WebRTC never asks for the playlist: it says it's watching here + if (url.searchParams.has('watching') && s.state !== 'ended') s.viewers.set(ipOf(req, server), Date.now()); + return json({ state: s.state, viewers: viewersOf(s), started: s.started, seconds: Math.round(((s.ended || Date.now()) - s.started) / 1000), rtc: s.rtc ? s.rtc.gen : null }); + } if (m[2].endsWith('.m3u8')) { if (s.state !== 'ended') s.viewers.set(ipOf(req, server), Date.now()); return new Response(playlist(s, m[2] === 'live.m3u8'), { headers: hlsHeaders }); @@ -334,6 +363,75 @@ export async function handle(req, server) { return new Response(f, { headers: { 'content-type': 'video/mp2t', 'access-control-allow-origin': '*', 'cache-control': 'public, max-age=86400, immutable' } }); } +// ------------------------------------------------------------------ WebRTC +// POST /live/<id>/rtc/publish {sdp, tracks: [{mid, trackName}]} → {sdp}: the stream's +// sender's offer (Authorization: Nostr <NIP-98 for POST /live/<id>/rtc/publish>); +// a new publish (a reconnect) replaces the last +// POST /live/<id>/rtc/watch → {session, sdp, gen}: an offer of the stream's tracks +// POST /live/<id>/rtc/answer {session, sdp}: the viewer's answer to it +async function rtcRoute(id, what, req, url, server) { + if (!SFU) return json({ error: 'no webrtc here' }, 501); + const s = load(id); + if (!s || s.state === 'removed' || s.state === 'ended') return json({ error: 'not live' }, 404); + const body = await req.json().catch(() => ({})); + try { + if (what === 'publish') { + const a = checkAuth(url, req, `/live/${id}/rtc/publish`, 'POST'); + if (a.error) return json({ error: a.error }, 401); + if (a.pubkey !== s.pubkey) return json({ error: 'not your stream' }, 403); + const tracks = (Array.isArray(body.tracks) ? body.tracks : []).filter((t) => t && typeof t.mid === 'string' && /^(video|audio)$/.test(t.trackName)).slice(0, 2); + if (typeof body.sdp !== 'string' || !tracks.length) return json({ error: 'bad offer' }, 400); + const { sessionId } = await sfu('/sessions/new'); + const j = await sfu(`/sessions/${sessionId}/tracks/new`, 'POST', { + sessionDescription: { type: 'offer', sdp: body.sdp }, + tracks: tracks.map((t) => ({ location: 'local', mid: t.mid, trackName: t.trackName })), + }); + s.rtc = { session: sessionId, tracks: tracks.map((t) => t.trackName), gen: ((s.rtc && s.rtc.gen) || 0) + 1 }; + s.watchers = new Map(); + log('rtc publish', s.id, sessionId.slice(0, 12)); + return json({ sdp: j.sessionDescription.sdp }); + } + if (!s.rtc) return json({ error: 'no webrtc for this stream' }, 404); + if (what === 'watch') { + const ip = ipOf(req, server), now = Date.now(); + const list = (rtcAsks.get(ip) || []).filter((t) => now - t < 3600_000); + if (list.length >= 120) return json({ error: 'rate limited' }, 429); + list.push(now); rtcAsks.set(ip, list); + if (rtcAsks.size > 20000) rtcAsks.clear(); + // asked just after a publish, the SFU may not have the sender's packets + // yet (a fresh session each try: one track may have been taken) + let j, sessionId; + for (let i = 0; ; i++) { + try { + ({ sessionId } = await sfu('/sessions/new')); + j = await sfu(`/sessions/${sessionId}/tracks/new`, 'POST', { + tracks: s.rtc.tracks.map((trackName) => ({ location: 'remote', sessionId: s.rtc.session, trackName })), + }); + break; + } catch (e) { + if (i >= 3 || !/not found/i.test(e.message)) throw e; + await Bun.sleep(1000); + if (!s.rtc) throw new Error('unpublished'); + } + } + if (!j.sessionDescription) throw new Error('sfu gave no offer'); + for (const [k, at] of s.watchers) if (now - at > 60_000) s.watchers.delete(k); + if (s.watchers.size > 5000) return json({ error: 'too many watching' }, 503); + s.watchers.set(sessionId, now); + s.viewers.set(ip, now); + return json({ session: sessionId, sdp: j.sessionDescription.sdp, gen: s.rtc.gen }); + } + // answer + if (typeof body.session !== 'string' || !s.watchers.has(body.session) || typeof body.sdp !== 'string') return json({ error: 'no such session' }, 404); + s.watchers.delete(body.session); + await sfu(`/sessions/${body.session}/renegotiate`, 'PUT', { sessionDescription: { type: 'answer', sdp: body.sdp } }); + return json({ ok: true }); + } catch (e) { + log('rtc', what, s.id, e.message); + return json({ error: 'webrtc failed' }, 502); + } +} + // ----------------------------------------------------------- moderation const pool = new SimplePool(); async function inboxOf(pk) { diff --git a/tools/live-ingest-test.js b/tools/live-ingest-test.js index 3af0456..d1b15e0 100644 --- a/tools/live-ingest-test.js +++ b/tools/live-ingest-test.js @@ -86,6 +86,9 @@ try { check('...H.264 at most 1280 on its long side, AAC sound', v?.codec_name === 'h264' && Math.max(v.width, v.height) <= 1280 && v.height > v.width && au?.codec_name === 'aac', v && `${v.codec_name} ${v.width}x${v.height} ${au?.codec_name}`); const st = await (await fetch(`${HTTP}/${id}/stats`)).json(); check('the watcher who fetched the playlist is counted', st.viewers === 1 && st.state === 'live', JSON.stringify(st)); + // WebRTC (Cloudflare's SFU) needs its key in the config: here there's none, and the app keeps to HLS + check('with no SFU key, no WebRTC: stats say so, and watch/publish answer 501', st.rtc === null + && (await fetch(`${HTTP}/${id}/rtc/watch`, { method: 'POST' })).status === 501 && (await fetch(`${HTTP}/${id}/rtc/publish`, { method: 'POST', body: '{}' })).status === 501); console.log('\n[a sender that drops comes back]'); a.ws.close(); |