Live streams over WebRTC. Going live also sends the camera to Cloudflare's SFU, which the live server brokers, and a viewer in the app watches from there: 0.1 s behind the broadcaster in go-live-test LATENCY=1, against 3.5 s on HLS.

- Switching to the screen and back swaps the track in place.
- A reconnect that makes a new stream publishes again, and its viewers follow.
- A viewer falls back to HLS if there's no WebRTC for the stream or no connection in 8 s.
HLS is still sent, 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, 107 insertions, 3 deletions

diff --git a/src/features/messages.js b/src/features/messages.js
index 6692ddc..969f0c1 100644
--- a/src/features/messages.js
+++ b/src/features/messages.js
@@ -8856,9 +8856,56 @@ export function messagesFeature(ctx) {
   function stopLivePlayer(w = liveWatch) {
     if (!w) return;
     if (w.hls) { try { w.hls.destroy(); } catch {} w.hls = null; }
+    if (w.pc) { try { w.pc.close(); } catch {} w.pc = null; }
+    clearInterval(w.rtcBeat);
     const v = w.video;
     w.video = null;
-    if (v) { try { v.pause(); v.removeAttribute('src'); v.load(); } catch {} }
+    if (v) { try { v.pause(); v.srcObject = null; v.removeAttribute('src'); v.load(); } catch {} }
+  }
+  // A stream of ours as WebRTC: the live server gets Cloudflare's SFU to offer
+  // the sender's tracks, we answer. Resolves once connected (rejects after
+  // 8 s, or if there's no WebRTC for it); after that, a lost connection or a
+  // new publish (the sender reconnected) mounts the player again.
+  async function rtcWatch(w, el, src, go) {
+    const base = src.replace(/live\.m3u8$/, '');
+    const pc = w.pc = new RTCPeerConnection({ iceServers: [{ urls: 'stun:stun.cloudflare.com:3478' }], bundlePolicy: 'max-bundle' });
+    const media = new MediaStream();
+    pc.ontrack = (e) => {
+      media.addTrack(e.track);
+      if (w.video === el && el.srcObject !== media) { el.srcObject = media; go(); }
+    };
+    const r = await fetch(base + 'rtc/watch', { method: 'POST', signal: AbortSignal.timeout(8000) });
+    if (!r.ok) throw new Error('watch ' + r.status);
+    const j = await r.json();
+    if (w.pc !== pc) return;
+    await pc.setRemoteDescription({ type: 'offer', sdp: j.sdp });
+    await pc.setLocalDescription(await pc.createAnswer());
+    const r2 = await fetch(base + 'rtc/answer', { method: 'POST', headers: { 'content-type': 'application/json' },
+      body: JSON.stringify({ session: j.session, sdp: pc.localDescription.sdp }), signal: AbortSignal.timeout(8000) });
+    if (!r2.ok) throw new Error('answer ' + r2.status);
+    await new Promise((resolve, reject) => {
+      const t = setTimeout(() => reject(new Error('rtc timeout')), 8000);
+      const on = () => {
+        if (pc.connectionState === 'connected') { clearTimeout(t); resolve(); }
+        else if (pc.connectionState === 'failed' || pc.connectionState === 'closed') { clearTimeout(t); reject(new Error('rtc ' + pc.connectionState)); }
+      };
+      pc.addEventListener('connectionstatechange', on);
+      on();
+    });
+    if (w.pc !== pc) return;
+    w.rtcFails = 0;
+    const again = (giveUp) => {
+      if (w.pc !== pc || w.video !== el) return;
+      if (giveUp || ++w.rtcFails > 3) w.noRtc = true;
+      stopLivePlayer(w);
+      scheduleRepaint();
+    };
+    pc.addEventListener('connectionstatechange', () => { if (pc.connectionState === 'failed') again(); });
+    // counted as watching (no playlist is asked for), and told of a new publish
+    w.rtcBeat = setInterval(() => {
+      fetch(base + 'stats?watching=1', { cache: 'no-store' }).then((x) => x.json())
+        .then((st) => { if (st.state === 'live' && st.rtc !== j.gen) again(st.rtc == null); }).catch(() => {});
+    }, 10_000);
   }
   function stopLiveWatch() {
     const w = liveWatch;
@@ -8910,6 +8957,17 @@ export function messagesFeature(ctx) {
     // still takes recordings, and live streams where hls.js can't run (iOS).
     // Measured (go-live-test LATENCY=1, one machine): 2 s segments on
     // Chrome's player 8.6 s behind the broadcaster; 1 s segments 4.5 s.
+    // Ours, on: WebRTC first (a fraction of a second behind), HLS if it
+    // can't be had — no WebRTC for the stream, or no connection in 8 s
+    if (src === liveStream(ev) && OUR_LIVE.test(src) && typeof RTCPeerConnection !== 'undefined' && !w.noRtc) {
+      rtcWatch(w, el, src, go).catch(() => {
+        if (w.video !== el) return;
+        w.noRtc = true;
+        stopLivePlayer(w);
+        scheduleRepaint(); // mounted again, as HLS
+      });
+      return;
+    }
     const live = src === liveStream(ev) && isHls(src) && typeof MediaSource !== 'undefined';
     if (!isHls(src) || (canHls() && !live)) { el.src = src; go(); return; }
     import('../hls-player.js').then((m) => m.attachHls(el, src, { live })).then((hls) => {
@@ -9159,8 +9217,9 @@ export function messagesFeature(ctx) {
       if (bc !== b) { for (const tr of media.getTracks()) tr.stop(); return; }
       if (live) bcStopMedia();
       b.media = media;
-      // a switch mid-stream: the recorder carries on with the new picture
-      if (b.phase === 'live') bcRecord(b, true);
+      // a switch mid-stream: the recorder carries on with the new picture,
+      // and the WebRTC copy swaps it in where it is
+      if (b.phase === 'live') { bcRecord(b, true); bcRtcSwap(b); }
     } catch (e) {
       if (bc !== b) return;
       if (live) b.source = was; // no screen after all: the camera goes on
@@ -9230,6 +9289,7 @@ export function messagesFeature(ctx) {
           toast(t('goLiveRestarted'));
           const old = { ...b, recording: null };
           Object.assign(b, { id: info.id, liveUrl: info.live, recording: info.recording, a: LIVE_KIND + ':' + b.pk + ':' + info.id });
+          bcRtc(b).catch(() => {});
           bcAnnounce(old, 'ended').catch(() => {});
           bcPlaying(b).then(() => (bc === b && b.phase === 'live' ? bcAnnounce(b, 'live') : null)).then(() => {
             // the streamer's own view of it follows to the new address
@@ -9244,6 +9304,48 @@ export function messagesFeature(ctx) {
     b.reconnecting = false;
     if (bc === b && b.phase === 'live') { toast(t('goLiveLost')); endBroadcast(); }
   }
+  // The stream as WebRTC, for the app's own viewers: the picture also goes to
+  // Cloudflare's SFU (the live server brokers it), and a viewer pulls it from
+  // there a fraction of a second behind — HLS is seconds behind. Best effort:
+  // the HLS above is the stream either way, and the recording.
+  async function bcRtc(b, tries = 0) {
+    if (typeof RTCPeerConnection === 'undefined' || bc !== b || b.phase !== 'live' || !b.id || !b.media) return;
+    bcRtcStop(b);
+    const id = b.id;
+    const pc = b.pc = new RTCPeerConnection({ iceServers: [{ urls: 'stun:stun.cloudflare.com:3478' }], bundlePolicy: 'max-bundle' });
+    const again = () => {
+      if (bc !== b || b.pc !== pc || b.phase !== 'live' || tries >= 5) return;
+      bcRtcStop(b);
+      setTimeout(() => bcRtc(b, tries + 1).catch(() => {}), 3000 * (tries + 1));
+    };
+    pc.onconnectionstatechange = () => { if (pc.connectionState === 'failed') again(); else if (pc.connectionState === 'connected') tries = 0; };
+    try {
+      const vt = b.media.getVideoTracks()[0], at = b.media.getAudioTracks()[0];
+      const tv = pc.addTransceiver(vt || 'video', { direction: 'sendonly', sendEncodings: [{ maxBitrate: 2_500_000 }] });
+      const ta = pc.addTransceiver(at || 'audio', { direction: 'sendonly' });
+      await pc.setLocalDescription(await pc.createOffer());
+      const u = liveBase() + '/live/' + id + '/rtc/publish';
+      const auth = await signAsMe({ kind: 27235, created_at: Math.floor(Date.now() / 1000), content: '', tags: [['u', APP_BASE + '/live/' + id + '/rtc/publish'], ['method', 'POST']] });
+      const r = await fetch(u, { method: 'POST', headers: { authorization: 'Nostr ' + btoa(JSON.stringify(auth)), 'content-type': 'application/json' },
+        body: JSON.stringify({ sdp: pc.localDescription.sdp, tracks: [{ mid: tv.mid, trackName: 'video' }, { mid: ta.mid, trackName: 'audio' }] }) });
+      if (r.status === 501 || r.status === 403) { bcRtcStop(b); return; } // no WebRTC on this server
+      if (!r.ok) throw new Error('publish ' + r.status);
+      const j = await r.json();
+      if (b.pc !== pc) return;
+      await pc.setRemoteDescription({ type: 'answer', sdp: j.sdp });
+    } catch { again(); }
+  }
+  function bcRtcSwap(b) {
+    if (!b.pc || !b.media) return;
+    for (const tr of b.pc.getTransceivers()) {
+      const kind = tr.receiver && tr.receiver.track && tr.receiver.track.kind;
+      const next = kind === 'video' ? b.media.getVideoTracks()[0] : kind === 'audio' ? b.media.getAudioTracks()[0] : null;
+      if (kind) tr.sender.replaceTrack(next || null).catch(() => {});
+    }
+  }
+  function bcRtcStop(b) {
+    if (b && b.pc) { try { b.pc.close(); } catch {} b.pc = null; }
+  }
   // the announcement: your 30311, and its copies as the stream goes on
   async function bcAnnounce(b, status) {
     const id = await requireIdentity();
@@ -9287,6 +9389,7 @@ export function messagesFeature(ctx) {
       b.phase = 'live';
       b.a = LIVE_KIND + ':' + b.pk + ':' + b.id;
       bcRecord(b);
+      bcRtc(b).catch(() => {});
       try { b.lock = await navigator.wakeLock?.request('screen'); } catch {}
       // the stream is a few seconds behind the camera: announced once it has
       // three segments — Chrome's own player won't start a live playlist with
@@ -9322,6 +9425,7 @@ export function messagesFeature(ctx) {
     b.phase = 'ended';
     clearInterval(b.tick);
     try { b.rec && b.rec.stop(); } catch {}
+    bcRtcStop(b);
     try { if (b.ws && b.ws.readyState === 1) b.ws.send('end'); } catch {}
     setTimeout(() => { try { b.ws && b.ws.close(1000, 'ended'); } catch {} }, 500);
     bcStopMedia();