live: WebRTC through Cloudflare's SFU (Realtime) for the app's viewers, a fraction of a second behind the broadcaster where HLS is 3.5 s. The server only brokers it, and the app's secret stays here ($LIVE_CONFIG sfuAppId/sfuSecret):

- POST /live/<id>/rtc/publish (the sender, NIP-98) pushes its tracks.
- POST /live/<id>/rtc/watch hands a viewer an offer of them, and /rtc/answer takes its answer.
/stats says which publish is current (rtc), and ?watching=1 counts a WebRTC viewer, who never fetches the playlist. HLS carries on as before, for the recording and the other nostr apps.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_013ZDf1cyCQLkfjftYhymZAg

1 files changed, 100 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) {