diff --git a/README.it.md b/README.it.md index 63ef951..e66c4ad 100644 --- a/README.it.md +++ b/README.it.md @@ -105,9 +105,14 @@ I browser consentono la videocamera solo su pagine sicure (HTTPS), quindi la fin - **I link internet** usano lo strumento gratuito di Cloudflare [`cloudflared`](https://developers.cloudflare.com/cloudflare-one/connections/connect-networks/downloads/): - va installato una volta: `brew install cloudflared` su macOS, `winget install --id Cloudflare.cloudflared` su Windows; - - l'indirizzo cambia a ogni riavvio di streamFree; - - attraverso Cloudflare passano solo la pagina ospite e la segnalazione; audio e video viaggiano direttamente tra te e gli ospiti. -- **Ospiti che non riescono a connettersi** (reti aziendali o mobili restrittive): aggiungi un server TURN in **Impostazioni → Ospiti → Server ICE**. + - l'indirizzo cambia a ogni riavvio di streamFree (e se il link cade, streamFree ne crea uno nuovo e te lo segnala). +- **Come viaggiano audio e video.** Audio e video passano direttamente tra te e ogni ospite (WebRTC) quando le due reti lo permettono. + Reti aziendali restrittive, rete mobile e CGNAT degli operatori spesso non lo permettono. Dopo circa 10 secondi senza connessione diretta, + streamFree passa automaticamente quell'ospite al suo **relay**: lo stesso audio e video, compressi dal browser e trasportati + da streamFree stesso, sullo stesso link che l'ospite ha aperto. Non servono account, server o configurazioni; il ritardo è un po' maggiore. + Privacy: le connessioni dirette sono cifrate end-to-end tra te e l'ospite. Anche il media del relay viaggia cifrato, + ma con un link Internet passa dalla rete di Cloudflare, che decifra e ricifra il traffico ai suoi nodi. +- **Un percorso più veloce per quegli ospiti:** aggiungi un server TURN in **Impostazioni → Ospiti → Server ICE**. **Impostazioni → Ospiti → Connessione** può anche far passare tutti gli ospiti dal relay. - **Un tuo proxy HTTPS:** avvia streamFree con `--public-url https://studio.example.com`. ## Riga di comando @@ -162,17 +167,19 @@ streamFree è giovane. Cosa è stato verificato, e come: | Registrazione, due destinazioni RTMP in contemporanea, sincronia audio/video | Test automatici nel browser (Chrome vero, videocamera simulata) a ogni push | | Ospiti: tre insieme, ognuno sente tutti tranne se stesso, chat del backstage nei due sensi, rimozione di un ospite, uscita e rientro, limite di 16 ospiti, sostituzione del link di invito | Test automatici nel browser | | Ospiti dalla rete locale (HTTPS) e da internet (tunnel Cloudflare reale) | Test automatici nel browser, eseguiti in locale (in CI manca cloudflared) | +| Ospiti la cui rete non permette connessioni dirette (simulata: ICE solo relay, senza TURN): passaggio automatico al relay, video e audio nei due sensi, mix-minus, registrazione con l'ospite sul palco, anche attraverso un tunnel Cloudflare reale | Test automatici nel browser | +| Un ospite la cui connessione cade torna sul palco da solo | Test automatici nel browser | | QR code, pulsanti di condivisione, interfaccia in italiano, ogni stringa tradotta | Test automatici nel browser | | Uscita RTMP e file MP4 | FFmpeg come verificatore indipendente nella suite di test | | App macOS: finestra nativa, 30 fps con la finestra ridotta a icona, audio sincronizzato | Smoke test integrato (`streamfree --smoke-test 8`) | -| App macOS che ospita un ospite (host WebKit, ospite Chrome, programma ricevuto a 15 fps) | `npm run test:native` | +| App macOS che ospita un ospite (host WebKit, ospite Chrome, programma ricevuto a 15 fps), più un ospite tramite relay | `npm run test:native` | | App macOS: videocamera, microfono e condivisione schermo | Test manuale | | I server di YouTube, Twitch, Facebook (RTMPS) e Kick (RTMPS) accettano la connessione di streamFree, e una chiave sbagliata dà un errore chiaro | Pulsante **Prova** e `cargo run --example ingest_probe`, con una chiave fittizia | Non ancora verificato: - **Una diretta con una chiave vera.** I controlli di protocollo qui sopra passano, ma nessuna diretta reale è stata ancora fatta. Segnala cosa funziona per te. -- **Ospiti da telefono e da altri browser.** Nessun ospite è ancora entrato da un telefono in rete mobile, né da Safari o Firefox. +- **Ospiti da telefono e da altri browser.** Nessun ospite è ancora entrato da un telefono in rete mobile, né da Safari o Firefox. Il relay è testato con una rete restrittiva simulata, non ancora con una reale. - **Windows.** App e installer si compilano e passano i test in CI, ma non sono ancora stati eseguiti su un PC Windows. - **Condurre da Linux.** Chrome su Linux sembra non avere un encoder AAC, quindi su Linux streamFree potrebbe funzionare solo per gli ospiti; in quel caso mostra un errore chiaro. diff --git a/README.md b/README.md index 1175f2b..62e1354 100644 --- a/README.md +++ b/README.md @@ -105,9 +105,14 @@ Browsers only allow cameras on secure (HTTPS) pages, so the Invite window asks w - **Internet links** use Cloudflare's free [`cloudflared`](https://developers.cloudflare.com/cloudflare-one/connections/connect-networks/downloads/) tool: - install it once: `brew install cloudflared` on macOS, `winget install --id Cloudflare.cloudflared` on Windows; - - the address changes every time streamFree restarts; - - only the guest page and its signaling go through Cloudflare; audio and video travel directly between you and your guests. -- **Guests who cannot connect** (strict company or mobile networks): add a TURN server in **Settings → Guests → ICE servers**. + - the address changes every time streamFree restarts (and if the link drops, streamFree makes a new one and tells you). +- **How media travels.** Audio and video go directly between you and each guest (WebRTC) whenever the two networks allow it. + Strict company networks, mobile data and carrier-grade NAT often don't. After about 10 seconds without a direct connection, + streamFree switches that guest automatically to its **relay**: the same audio and video, compressed by the browser and carried + through streamFree itself, over the same link the guest opened. No account, server or setup is needed; there is a little more delay. + Privacy: direct connections are encrypted end to end between you and the guest. Relayed media is encrypted on the way too, + but with an Internet link it passes through Cloudflare's network, which decrypts and re-encrypts traffic at its edge. +- **A faster path for those guests:** add a TURN server in **Settings → Guests → ICE servers**. **Settings → Guests → Connection** can also send every guest through the relay. - **Your own HTTPS proxy:** start streamFree with `--public-url https://studio.example.com`. ## Command line @@ -162,17 +167,19 @@ streamFree is young. What has been checked, and how: | Recording, two simultaneous RTMP destinations, audio/video sync | Automated browser tests (real Chrome, fake camera) on every push | | Guests: three at once, each hearing everyone but themselves, backstage chat both ways, removing a guest, leaving and rejoining, the 16-guest limit, replacing the invite link | Automated browser tests | | Guests over the local network (HTTPS) and over the internet (a real Cloudflare tunnel) | Automated browser tests, run locally (CI has no cloudflared) | +| Guests whose network allows no direct connection (simulated: relay-only ICE, no TURN): automatic switch to the relay, video and audio both ways, mix-minus, recording with them on stage, also through a real Cloudflare tunnel | Automated browser tests | +| A guest whose connection drops comes back on stage by itself | Automated browser tests | | QR code, share buttons, Italian interface, every string translated | Automated browser tests | | RTMP output and MP4 files | FFmpeg as an independent oracle in the test suite | | macOS app: native window, 30 fps with the window minimized, audio in sync | Built-in smoke test (`streamfree --smoke-test 8`) | -| macOS app hosting a guest (WebKit host, Chrome guest, program received at 15 fps) | `npm run test:native` | +| macOS app hosting a guest (WebKit host, Chrome guest, program received at 15 fps), plus a relayed guest | `npm run test:native` | | macOS app: camera, microphone and screen sharing | Manual test | | YouTube, Twitch, Facebook (RTMPS) and Kick (RTMPS) ingest servers accept streamFree's connection, and a wrong key gets a clear error | **Test** button and `cargo run --example ingest_probe`, with a dummy key | Not verified yet: - **Going live with a real stream key.** The protocol checks above pass, but no real broadcast has been made yet. Please report what works for you. -- **Phones and other browsers as guests.** No guest has yet joined from a phone on mobile data, or from Safari or Firefox. +- **Phones and other browsers as guests.** No guest has yet joined from a phone on mobile data, or from Safari or Firefox. The relay is tested with a simulated strict network, not yet with a real one. - **Windows.** The app and installer build and pass the test suite in CI, but have not been run on a Windows PC yet. - **Linux hosting.** Chrome on Linux appears to have no AAC encoder, so on Linux streamFree may only work for guests; it shows a clear error if so. diff --git a/docs/architecture.md b/docs/architecture.md index 8ba6f07..031fce5 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -143,6 +143,25 @@ The server answers with `started`, then a `status` message every second answers with its camera and microphone on the same transceivers. - The Rust server only relays signaling messages between the host and guests; guests cannot address each other. +- **Relay fallback.** When a guest's peer connection is not `connected` about + 10 s after the offer (or turns `failed`, or stays `disconnected`), the host + closes it and sends `relay-start`. Both pages then exchange media over two + more WebSockets through the server (`/ws/relay/guest`, authenticated with a + per-guest key sent only in that guest's `welcome`, and `/ws/relay/host`), + which work wherever the guest page loads, the Cloudflare tunnel included. + Media is encoded with WebCodecs (`web/js/relay.js`): H.264 (Annex B) or VP8 + video, at most 640 px on the long side so portrait phones stay portrait, and + Opus or 16-bit PCM audio, each side choosing what the other can decode. Packets + carry sequence numbers; a gap makes the receiver ask for a keyframe. Audio + plays through a ~100 ms jitter buffer that skips ahead past 250 ms, so delay + does not grow over a long call (`relay-worklet.js`). The server forwards + packets without decoding them and drops them when a receiver's queue is full. + A relayed guest enters the mixer like any other guest, with its own + mix-minus bus, and the host encodes the return feed once for all relayed guests. +- Every signaling and relay socket is pinged every 20 s, because proxies such as + Cloudflare close silent WebSockets. A guest whose socket drops reconnects by + itself for about a minute (unless the host removed them, close code 4001) and, + identified by a per-tab client id, goes back on stage if it was on. - Cameras require a secure context, so guests need HTTPS unless they are on `localhost`. The Invite window sets the **sharing mode** (`src/sharing.rs`), which can change while streamFree runs and is remembered in the settings: @@ -150,8 +169,10 @@ The server answers with `started`, then a `status` message every second - *Local network*: an extra HTTPS listener on port + 1 with a self-signed certificate made on first use (`src/net.rs`), stopped when the mode changes. - *Internet*: a Cloudflare quick tunnel (`cloudflared`), started in the - background; the studio is notified when the public address is ready, if it - fails, or if the tunnel stops later. cloudflared is looked up in PATH and in + background. The address is announced only once cloudflared reports a + registered edge connection (before that, guests get an error page). A tunnel + that fails to start or stops later is restarted up to four times, and the + studio is told about the new address. cloudflared is looked up in PATH and in the usual Homebrew and Windows install folders, because apps started from Finder or the Start menu do not inherit the shell's PATH. `--lan`, `--tunnel` and `--public-url` (own reverse proxy) override the mode diff --git a/src/net.rs b/src/net.rs index 5f407fb..94eb1fa 100644 --- a/src/net.rs +++ b/src/net.rs @@ -137,9 +137,11 @@ impl Tunnel { } } -/// Starts `cloudflared tunnel --url http://127.0.0.1:` and waits for the -/// public `https://*.trycloudflare.com` URL it prints. `on_exit` runs if the -/// tunnel process ends later on its own. +/// Starts `cloudflared tunnel --url http://127.0.0.1:` and waits until +/// the public `https://*.trycloudflare.com` address it prints actually works: +/// cloudflared prints the address before the edge has registered the tunnel, +/// and a guest who opens the link in that window gets an error page. +/// `on_exit` runs if the tunnel process ends later on its own. pub async fn start_tunnel(port: u16, on_exit: impl FnOnce() + Send + 'static) -> Result { let program = find_cloudflared().ok_or_else(|| anyhow!("cloudflared is not installed"))?; let mut child = Command::new(program) @@ -171,6 +173,20 @@ pub async fn start_tunnel(port: u16, on_exit: impl FnOnce() + Send + 'static) -> Ok(Err(e)) => return Err(e.into()), Err(_) => bail!("cloudflared did not print an address within 45 s"), }; + // Then wait for the first registered edge connection. Older cloudflared + // versions may word it differently, so a timeout is not an error. + let registered = tokio::time::timeout(Duration::from_secs(30), async { + while let Some(line) = lines.next_line().await? { + if is_registered(&line) { + return Ok::<_, std::io::Error>(true); + } + } + Ok(false) + }) + .await; + if let Ok(Ok(false)) = registered { + bail!("cloudflared exited before the internet link was ready"); + } // Keep draining stderr so cloudflared never blocks on a full pipe; when it // closes, the process has ended. tokio::spawn(async move { @@ -180,6 +196,10 @@ pub async fn start_tunnel(port: u16, on_exit: impl FnOnce() + Send + 'static) -> Ok(Tunnel { url, child }) } +fn is_registered(line: &str) -> bool { + line.contains("Registered tunnel connection") +} + fn find_tunnel_url(line: &str) -> Option { let start = line.find("https://")?; let url: String = line[start..] @@ -206,6 +226,16 @@ mod tests { ); } + #[test] + fn detects_registered_connection() { + assert!(is_registered( + "2026-10-14T10:00:02Z INF Registered tunnel connection connIndex=0 connection=abc event=0 ip=198.41.200.13 location=mxp01 protocol=quic" + )); + assert!(!is_registered( + "INF Starting metrics server on 127.0.0.1:20241/metrics" + )); + } + #[test] fn cloudflared_override_is_used() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/server.rs b/src/server.rs index 159ea92..c5bc11f 100644 --- a/src/server.rs +++ b/src/server.rs @@ -6,6 +6,12 @@ //! * the host holds `host_token` (in the URL the app opens) and may do anything; //! * guests hold `invite_token` (in the invite link) and may only join the //! signaling room. Stream keys never leave this process. +//! +//! Guests normally exchange media with the host directly over WebRTC. When no +//! direct path exists (strict NATs, firewalls, mobile networks), the host +//! switches that guest to the relay: encoded media then travels over two +//! WebSockets through this server (`/ws/relay/guest`, `/ws/relay/host`), which +//! works wherever the guest page itself loads. use crate::chat; use crate::live::{DestinationSpec, Session, SessionOptions}; @@ -34,12 +40,34 @@ struct Web; const MAX_GUESTS: usize = 16; const MAX_ASSET_BYTES: usize = 512 * 1024 * 1024; +/// Proxies (Cloudflare among them) close WebSockets that stay silent for +/// about 100 s; a ping keeps backstage guests connected. +const KEEPALIVE: Duration = Duration::from_secs(20); +/// Queued to a guest's signaling channel when the host removes them, so the +/// guest page can tell a removal from a dropped connection. +const KICKED: &str = "\u{0}kicked"; +/// WebSocket close code sent to a removed guest. +const CLOSE_REMOVED: u16 = 4001; +/// Largest relay message (one encoded frame plus its header). +const MAX_RELAY_MESSAGE: usize = 2 * 1024 * 1024; struct Guest { name: String, + /// Chosen by the guest page and kept across its reconnections, so the + /// host can put a guest who dropped for a moment back where they were. + cid: String, + /// Lets this guest, and only this guest, open its relay socket. + relay_key: String, tx: mpsc::Sender, } +/// Where relayed media goes: the host page's relay socket and each guest's. +#[derive(Default)] +struct RelayRoutes { + host: Option>, + guests: HashMap>, +} + pub struct AppState { pub store: Mutex, pub recordings_dir: PathBuf, @@ -54,6 +82,7 @@ pub struct AppState { router: std::sync::OnceLock, host_tx: Mutex>>, guests: Mutex>, + relay: Mutex, events: broadcast::Sender, live: Mutex, /// Open studio pages, and when the count last changed (app mode quits @@ -78,6 +107,7 @@ impl AppState { router: std::sync::OnceLock::new(), host_tx: Mutex::new(None), guests: Mutex::new(HashMap::new()), + relay: Mutex::new(RelayRoutes::default()), events, live: Mutex::new(false), studio_pages: std::sync::Mutex::new((0, Instant::now())), @@ -176,6 +206,8 @@ pub fn router(state: Shared) -> Router { .route("/api/recordings/open", post(open_recordings)) .route("/ws/host", get(host_ws)) .route("/ws/guest", get(guest_ws)) + .route("/ws/relay/host", get(relay_host_ws)) + .route("/ws/relay/guest", get(relay_guest_ws)) .route("/ws/ingest", get(ingest_ws)) .layer(axum::middleware::map_response(security_headers)) .with_state(state) @@ -683,6 +715,16 @@ async fn host_session(state: Shared, socket: WebSocket) { state.studio_page_delta(false); } +fn keepalive() -> tokio::time::Interval { + let mut t = tokio::time::interval_at(tokio::time::Instant::now() + KEEPALIVE, KEEPALIVE); + t.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + t +} + +fn guest_summary(id: &str, g: &Guest) -> Value { + json!({"id": id, "name": g.name, "cid": g.cid}) +} + async fn host_session_inner(state: &Shared, mut socket: WebSocket) { let (tx, mut rx) = mpsc::channel::(256); *state.host_tx.lock().await = Some(tx.clone()); @@ -692,7 +734,7 @@ async fn host_session_inner(state: &Shared, mut socket: WebSocket) { .lock() .await .iter() - .map(|(id, g)| json!({"id": id, "name": g.name})) + .map(|(id, g)| guest_summary(id, g)) .collect(); let hello = json!({"type": "hello", "guests": snapshot}); if socket @@ -702,6 +744,7 @@ async fn host_session_inner(state: &Shared, mut socket: WebSocket) { { return; } + let mut ping = keepalive(); loop { tokio::select! { incoming = socket.recv() => { @@ -710,8 +753,7 @@ async fn host_session_inner(state: &Shared, mut socket: WebSocket) { let Ok(mut value) = serde_json::from_str::(&text) else { continue }; let Some(to) = value.get("to").and_then(Value::as_str).map(str::to_owned) else { continue }; if value.get("type").and_then(Value::as_str) == Some("kick") { - state.guests.lock().await.remove(&to); - state.to_host(json!({"type": "guest-left", "id": to})).await; + remove_guest(state, &to, true).await; continue; } value["from"] = "host".into(); @@ -733,6 +775,9 @@ async fn host_session_inner(state: &Shared, mut socket: WebSocket) { Err(broadcast::error::RecvError::Closed) => break, } } + _ = ping.tick() => { + if socket.send(Message::Ping(Bytes::new())).await.is_err() { break } + } } } let mut slot = state.host_tx.lock().await; @@ -744,11 +789,35 @@ async fn host_session_inner(state: &Shared, mut socket: WebSocket) { } } +/// Removes a guest: their signaling and relay sockets close and the host is +/// told. `kicked` tells the guest page it was removed on purpose. +async fn remove_guest(state: &Shared, id: &str, kicked: bool) { + let removed = state.guests.lock().await.remove(id); + if let Some(guest) = removed { + if kicked { + let _ = guest.tx.try_send(KICKED.to_owned()); + } + state.relay.lock().await.guests.remove(id); + state.to_host(json!({"type": "guest-left", "id": id})).await; + } +} + #[derive(Deserialize)] struct GuestQuery { invite: String, #[serde(default)] name: String, + #[serde(default)] + cid: String, +} + +/// Keeps `s` only if it is a short run of ASCII letters, digits and dashes. +fn clean_id(s: &str) -> String { + if !s.is_empty() && s.len() <= 40 && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '-') { + s.to_owned() + } else { + String::new() + } } async fn guest_ws( @@ -775,29 +844,34 @@ async fn guest_ws( } else { name }; - ws.on_upgrade(move |socket| guest_session(state, socket, name)) + let cid = clean_id(&q.cid); + ws.on_upgrade(move |socket| guest_session(state, socket, name, cid)) } -async fn guest_session(state: Shared, mut socket: WebSocket, name: String) { +async fn guest_session(state: Shared, mut socket: WebSocket, name: String, cid: String) { let id = random_token()[..12].to_owned(); + let relay_key = random_token(); let (tx, mut rx) = mpsc::channel::(256); - state.guests.lock().await.insert( - id.clone(), - Guest { - name: name.clone(), - tx, - }, - ); + let guest = Guest { + name: name.clone(), + cid: cid.clone(), + relay_key: relay_key.clone(), + tx, + }; + let summary = guest_summary(&id, &guest); + state.guests.lock().await.insert(id.clone(), guest); let ice = state.store.lock().await.settings.ice_servers.clone(); - let welcome = json!({"type": "welcome", "id": id, "iceServers": ice}); + let welcome = json!({"type": "welcome", "id": id, "iceServers": ice, "relayKey": relay_key}); + let mut kicked = false; if socket .send(Message::Text(welcome.to_string().into())) .await .is_ok() { - state - .to_host(json!({"type": "guest-joined", "id": id, "name": name})) - .await; + let mut joined = summary; + joined["type"] = "guest-joined".into(); + state.to_host(joined).await; + let mut ping = keepalive(); loop { tokio::select! { incoming = socket.recv() => { @@ -810,15 +884,195 @@ async fn guest_session(state: Shared, mut socket: WebSocket, name: String) { state.to_host(value).await; } outgoing = rx.recv() => { - // The host kicked this guest when the sender is dropped. + // The sender is dropped when the guest is removed. let Some(text) = outgoing else { break }; + if text == KICKED { + kicked = true; + break; + } if socket.send(Message::Text(text.into())).await.is_err() { break } } + _ = ping.tick() => { + if socket.send(Message::Ping(Bytes::new())).await.is_err() { break } + } } } } - if state.guests.lock().await.remove(&id).is_some() { - state.to_host(json!({"type": "guest-left", "id": id})).await; + if kicked { + let frame = axum::extract::ws::CloseFrame { + code: CLOSE_REMOVED, + reason: "removed by the host".into(), + }; + let _ = socket.send(Message::Close(Some(frame))).await; + } + remove_guest(&state, &id, false).await; +} + +// ---------------------------------------------------------------- media relay + +#[derive(Deserialize)] +struct RelayGuestQuery { + invite: String, + id: String, + key: String, +} + +async fn relay_host_ws( + State(state): State, + headers: HeaderMap, + Query(q): Query, + ws: WebSocketUpgrade, +) -> Response { + require_host!(state, headers, q); + ws.max_message_size(MAX_RELAY_MESSAGE) + .on_upgrade(move |socket| relay_host_session(state, socket)) +} + +/// The host page's relay socket. Binary messages are `[id length][guest +/// id][payload]` in both directions; text messages are JSON with `to` +/// (from the host) or `from` (to the host). +async fn relay_host_session(state: Shared, mut socket: WebSocket) { + let (tx, mut rx) = mpsc::channel::(512); + state.relay.lock().await.host = Some(tx.clone()); + let mut ping = keepalive(); + loop { + tokio::select! { + incoming = socket.recv() => { + let Some(Ok(msg)) = incoming else { break }; + match msg { + Message::Binary(data) => { + let Some((id, payload)) = split_relay(&data) else { continue }; + let relay = state.relay.lock().await; + if let Some(guest) = relay.guests.get(id) { + // A full queue means the guest's link is slower than the + // stream: drop the frame, the receiver asks for a keyframe. + let _ = guest.try_send(Message::Binary(Bytes::copy_from_slice(payload))); + } + } + Message::Text(text) => { + let Ok(mut value) = serde_json::from_str::(&text) else { continue }; + let Some(to) = value.get("to").and_then(Value::as_str).map(str::to_owned) else { continue }; + if let Some(obj) = value.as_object_mut() { obj.remove("to"); } + let relay = state.relay.lock().await; + if let Some(guest) = relay.guests.get(&to) { + let _ = guest.try_send(Message::Text(value.to_string().into())); + } + } + _ => {} + } + } + outgoing = rx.recv() => { + let Some(msg) = outgoing else { break }; + if socket.send(msg).await.is_err() { break } + } + _ = ping.tick() => { + if socket.send(Message::Ping(Bytes::new())).await.is_err() { break } + } + } + } + let mut relay = state.relay.lock().await; + if relay.host.as_ref().is_some_and(|h| h.same_channel(&tx)) { + relay.host = None; + } +} + +/// `[id length][id][payload]` -> (id, payload). +fn split_relay(data: &[u8]) -> Option<(&str, &[u8])> { + let len = *data.first()? as usize; + let id = std::str::from_utf8(data.get(1..1 + len)?).ok()?; + Some((id, data.get(1 + len..)?)) +} + +fn join_relay(id: &str, payload: &[u8]) -> Bytes { + let mut out = Vec::with_capacity(1 + id.len() + payload.len()); + out.push(id.len() as u8); + out.extend_from_slice(id.as_bytes()); + out.extend_from_slice(payload); + out.into() +} + +async fn relay_guest_ws( + State(state): State, + Query(q): Query, + ws: WebSocketUpgrade, +) -> Response { + let expected = state.store.lock().await.settings.invite_token.clone(); + if !constant_eq(&q.invite, &expected) { + return (StatusCode::FORBIDDEN, "invalid invite").into_response(); + } + let allowed = state + .guests + .lock() + .await + .get(&q.id) + .is_some_and(|g| constant_eq(&q.key, &g.relay_key)); + if !allowed { + return (StatusCode::FORBIDDEN, "unknown guest").into_response(); + } + ws.max_message_size(MAX_RELAY_MESSAGE) + .on_upgrade(move |socket| relay_guest_session(state, socket, q.id)) +} + +async fn relay_guest_session(state: Shared, mut socket: WebSocket, id: String) { + let (tx, mut rx) = mpsc::channel::(128); + state + .relay + .lock() + .await + .guests + .insert(id.clone(), tx.clone()); + to_relay_host( + &state, + Message::Text(json!({"type": "relay-open", "from": id}).to_string().into()), + ) + .await; + let mut ping = keepalive(); + loop { + tokio::select! { + incoming = socket.recv() => { + let Some(Ok(msg)) = incoming else { break }; + match msg { + Message::Binary(data) => { + to_relay_host(&state, Message::Binary(join_relay(&id, &data))).await; + } + Message::Text(text) => { + if text.len() > 64 * 1024 { continue } + let Ok(mut value) = serde_json::from_str::(&text) else { continue }; + value["from"] = id.clone().into(); + to_relay_host(&state, Message::Text(value.to_string().into())).await; + } + _ => {} + } + } + outgoing = rx.recv() => { + // The route is removed (and this sender dropped) when the guest leaves. + let Some(msg) = outgoing else { break }; + if socket.send(msg).await.is_err() { break } + } + _ = ping.tick() => { + if socket.send(Message::Ping(Bytes::new())).await.is_err() { break } + } + } + } + let mut relay = state.relay.lock().await; + if relay.guests.get(&id).is_some_and(|g| g.same_channel(&tx)) { + relay.guests.remove(&id); + } + drop(relay); + to_relay_host( + &state, + Message::Text( + json!({"type": "relay-closed", "from": id}) + .to_string() + .into(), + ), + ) + .await; +} + +async fn to_relay_host(state: &Shared, msg: Message) { + if let Some(host) = state.relay.lock().await.host.as_ref() { + let _ = host.try_send(msg); } } @@ -1091,6 +1345,21 @@ mod tests { assert!(safe_file_stem("../..").starts_with("streamfree-")); } + #[test] + fn relay_framing_round_trips() { + let framed = join_relay("abc123", &[1, 2, 3]); + assert_eq!(split_relay(&framed), Some(("abc123", &[1u8, 2, 3][..]))); + assert_eq!(split_relay(&[]), None); + assert_eq!(split_relay(&[9, b'a']), None); + } + + #[test] + fn client_ids_are_sanitized() { + assert_eq!(clean_id("a1-B2"), "a1-B2"); + assert_eq!(clean_id("../x"), ""); + assert_eq!(clean_id(&"a".repeat(41)), ""); + } + #[test] fn constant_time_compare() { assert!(constant_eq("abc", "abc")); diff --git a/src/sharing.rs b/src/sharing.rs index 30881f7..6796d08 100644 --- a/src/sharing.rs +++ b/src/sharing.rs @@ -122,7 +122,7 @@ pub async fn set_mode(state: &Shared, mode: SharingMode) { if mode == SharingMode::Internet { if s.tunnel.is_none() && s.tunnel_state != TunnelState::Starting { s.tunnel_state = TunnelState::Starting; - tokio::spawn(start_tunnel(state.clone(), generation)); + tokio::spawn(start_tunnel(state.clone(), generation, 1)); } } else { if let Some(tunnel) = s.tunnel.take() { @@ -134,10 +134,24 @@ pub async fn set_mode(state: &Shared, mode: SharingMode) { notify(state).await; } -async fn start_tunnel(state: Shared, generation: u64) { +/// How many times in a row a failed or dropped tunnel is started again before +/// the studio shows the error. +const TUNNEL_ATTEMPTS: u32 = 4; + +fn retry_tunnel(state: Shared, generation: u64, attempt: u32) { + let next: std::pin::Pin + Send>> = + Box::pin(async move { + tokio::time::sleep(std::time::Duration::from_secs(2 * u64::from(attempt))).await; + start_tunnel(state, generation, attempt).await; + }); + tokio::spawn(next); +} + +async fn start_tunnel(state: Shared, generation: u64, attempt: u32) { let watcher = state.clone(); let result = net::start_tunnel(state.port, move || { - // The tunnel process ended on its own (network change, crash). + // The tunnel process ended on its own (network change, crash): start + // a new one. Its address differs, and the studio is told. tokio::spawn(async move { { let mut s = watcher.sharing.lock().await; @@ -146,12 +160,11 @@ async fn start_tunnel(state: Shared, generation: u64) { return; } s.tunnel = None; - s.tunnel_state = TunnelState::Error( - "The internet link stopped working. Choose Internet again to get a new link." - .into(), - ); + s.tunnel_state = TunnelState::Starting; } + tracing::warn!("the internet link stopped; starting a new one"); notify(&watcher).await; + retry_tunnel(watcher, generation, 1); }); }) .await; @@ -166,6 +179,16 @@ async fn start_tunnel(state: Shared, generation: u64) { s.tunnel_state = TunnelState::On; } Ok(tunnel) => tunnel.stop().await, + Err(e) + if still_wanted + && attempt < TUNNEL_ATTEMPTS + && net::find_cloudflared().is_some() => + { + tracing::warn!("internet link attempt {attempt} failed: {e:#}"); + drop(s); + retry_tunnel(state.clone(), generation, attempt + 1); + return; + } Err(e) if still_wanted => s.tunnel_state = TunnelState::Error(format!("{e:#}")), Err(_) => {} } diff --git a/src/store.rs b/src/store.rs index 1ffbbc9..c32952d 100644 --- a/src/store.rs +++ b/src/store.rs @@ -87,12 +87,31 @@ pub struct Settings { pub studio: serde_json::Value, } -fn default_ice() -> Vec { - vec![IceServer { - urls: "stun:stun.l.google.com:19302".into(), +fn stun(urls: &str) -> IceServer { + IceServer { + urls: urls.into(), username: None, credential: None, - }] + } +} + +/// Two independent public STUN services, so one being blocked or down does +/// not stop guests from finding a direct path. +fn default_ice() -> Vec { + vec![ + stun("stun:stun.l.google.com:19302"), + stun("stun:stun.cloudflare.com:3478"), + ] +} + +/// Settings saved by older versions hold only the first STUN server; give +/// them the current defaults. A list the user edited is left alone. +fn upgrade_ice(servers: &mut Vec) -> bool { + if *servers == [stun("stun:stun.l.google.com:19302")] { + *servers = default_ice(); + return true; + } + false } pub fn random_token() -> String { @@ -130,7 +149,8 @@ impl Store { Err(e) if e.kind() == std::io::ErrorKind::NotFound => Settings::default(), Err(e) => return Err(e).with_context(|| format!("read {}", path.display())), }; - let store = Self { dir, settings }; + let mut store = Self { dir, settings }; + upgrade_ice(&mut store.settings.ice_servers); store.save()?; Ok(store) } @@ -215,6 +235,16 @@ mod tests { } } + #[test] + fn old_default_ice_list_is_upgraded() { + let mut old = vec![stun("stun:stun.l.google.com:19302")]; + assert!(upgrade_ice(&mut old)); + assert_eq!(old, default_ice()); + let mut custom = vec![stun("stun:example.org:3478")]; + assert!(!upgrade_ice(&mut custom)); + assert_eq!(custom.len(), 1); + } + #[test] fn public_view_hides_key() { let d = Destination { diff --git a/tests/e2e/native.mjs b/tests/e2e/native.mjs index 29b4f3d..864c6b8 100644 --- a/tests/e2e/native.mjs +++ b/tests/e2e/native.mjs @@ -1,15 +1,18 @@ // Release check for macOS/Windows: the native app (system WebView) hosts a -// 24-second smoke recording while a Chrome guest joins, goes on stage and -// receives the program. +// 40-second smoke recording while a Chrome guest joins over WebRTC, goes on +// stage and receives the program, and a second guest with no direct path +// joins through the relay. // // cargo build && (cd tests/e2e && npm run test:native) import { chromium } from 'playwright-core'; import { spawn } from 'node:child_process'; -import { mkdtempSync } from 'node:fs'; +import { mkdtempSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; const tmp = mkdtempSync(join(tmpdir(), 'streamfree-native-')); -const child = spawn('../../target/debug/streamfree', ['--smoke-test', '24', '--port', '8960', '--data-dir', `${tmp}/data`, '--recordings-dir', `${tmp}/rec`], { stdio: ['ignore', 'pipe', 'pipe'] }); +const child = spawn('../../target/debug/streamfree', ['--smoke-test', '40', '--port', '8960', '--data-dir', `${tmp}/data`, '--recordings-dir', `${tmp}/rec`], { stdio: ['ignore', 'pipe', 'pipe'] }); +let err = ''; child.stderr.on('data', (d) => (err += d)); +process.on('exit', () => writeFileSync(`${tmp}/app-stderr.log`, err)); let out = ''; child.stdout.on('data', (d) => (out += d)); const exited = new Promise((r) => child.on('exit', r)); for (let i = 0; i < 100 && !out.includes('Invite link:'); i++) await new Promise((r) => setTimeout(r, 100)); @@ -32,7 +35,33 @@ const frames = await g.evaluate(async () => { const toasts = await g.evaluate(() => [...document.querySelectorAll('.toast')].map((t) => t.textContent)); console.log('guest connected + on stage; program frames received in 3 s:', frames, '| errors:', JSON.stringify(errors), '| toasts:', JSON.stringify(toasts)); if (frames < 20 || errors.length) { console.log('FAIL'); process.exitCode = 1; } + +// A guest whose network allows no direct connection: the WebView host must +// switch it to the relay, decode its video (it confirms with relay-ok) and +// send it the program. +const rctx = await b.newContext({ permissions: ['camera', 'microphone'], locale: 'en-US' }); +await rctx.addInitScript(() => { + const Orig = window.RTCPeerConnection; + window.RTCPeerConnection = function (config = {}) { return new Orig({ ...config, iceTransportPolicy: 'relay' }); }; + window.RTCPeerConnection.prototype = Orig.prototype; +}); +const r = await rctx.newPage(); +const rErrors = []; r.on('pageerror', (e) => rErrors.push(e.message)); +await r.goto(invite); +await r.fill('input[aria-label="Your name"]', 'Relay guest'); +await r.click('button[type=submit]'); +let relay = null; +try { + await r.waitForFunction(() => { const s = window.streamFreeGuest.relay; return s && s.videoFrames > 20 && s.hostReceiving; }, null, { timeout: 25000 }); +} catch { + // reported below +} +relay = await r.evaluate(() => window.streamFreeGuest.relay); +console.log('relay guest:', JSON.stringify(relay), '| errors:', JSON.stringify(rErrors)); +if (!relay || relay.videoFrames < 20 || !relay.hostReceiving || rErrors.length) { console.log('FAIL (relay)'); process.exitCode = 1; } await exited; console.log(out.split('\n').filter((l) => l.includes('smoke')).join('\n')); +const code = await exited.then((c) => c); +if (err.includes('smoke test:')) console.log(err.split('\n').filter((l) => l.includes('smoke')).join('\n')); console.log(`recording in ${tmp}/rec`); await b.close(); diff --git a/tests/e2e/run.mjs b/tests/e2e/run.mjs index 9ffa58d..17dba23 100644 --- a/tests/e2e/run.mjs +++ b/tests/e2e/run.mjs @@ -276,7 +276,7 @@ await test('the host can remove a guest; a guest can leave and come back', async await receivesProgram(g.page); const id = await guestId('Kicked'); await studio.evaluate((i) => window.streamFree.kick(i), id); - await waitFor(g.page, () => document.body.textContent.includes('connection to the studio ended')); + await waitFor(g.page, () => document.body.textContent.includes('removed you from the studio')); await waitFor(studio, (i) => !window.streamFree.people.has(i), id); // Leave, then rejoin from the same page. await g.page.click('text=Join again'); @@ -290,6 +290,122 @@ await test('the host can remove a guest; a guest can leave and come back', async await g.ctx.close(); }); +// A guest on a network that allows no direct connection (strict NATs, +// firewalls, many mobile networks): only relay candidates are allowed and no +// TURN server is configured, so WebRTC cannot connect and the host must +// switch the guest to the streamFree relay. +async function joinRelayOnlyGuest(url, name) { + const gctx = await browser.newContext({ permissions: ['camera', 'microphone'], locale: 'en-US' }); + await gctx.addInitScript(() => { + const Orig = window.RTCPeerConnection; + window.RTCPeerConnection = function (config = {}) { + return new Orig({ ...config, iceTransportPolicy: 'relay' }); + }; + window.RTCPeerConnection.prototype = Orig.prototype; + }); + const page = await gctx.newPage(); + const errors = []; + page.on('pageerror', (e) => errors.push(e.message)); + await page.goto(url); + await page.fill('input[aria-label="Your name"]', name); + await page.click('button[type=submit]'); + return { ctx: gctx, page, errors }; +} + +// The guest gets the program and its mix-minus, and the host decodes the +// guest's video (it confirms with relay-ok). +async function receivesRelay(page, timeout = 30000) { + await waitFor(page, () => { + const r = window.streamFreeGuest.relay; + return r && r.videoFrames > 20 && r.audioPackets > 20 && r.hostReceiving; + }, null, timeout); +} + +function newestRecording() { + const dir = join(work, 'recordings'); + const newest = readdirSync(dir).filter((f) => f.endsWith('.mp4')) + .sort((a, b) => statSync(join(dir, a)).mtimeMs - statSync(join(dir, b)).mtimeMs).at(-1); + return join(dir, newest); +} + +await test('a guest with no direct path joins through the relay', async () => { + const { inviteUrl } = await sharingStatus(); + const g = await joinRelayOnlyGuest(inviteUrl, 'Faraway'); + await receivesRelay(g.page); + const id = await guestId('Faraway'); + await waitFor(studio, (i) => { + const s = window.streamFree.relayStats()[i]; + return s && s.transport === 'relay' && s.videoFrames > 20 && s.audioPackets > 20 && s.track === 'live'; + }, id, 30000); + // The fake microphones beep once a second: both directions carry sound. + const peak = await studio.evaluate(async (i) => { + let max = 0; + const end = performance.now() + 3000; + while (performance.now() < end) { + max = Math.max(max, window.streamFree.relayStats()[i]?.level || 0); + await new Promise((r) => setTimeout(r, 15)); + } + return max; + }, id); + assert.ok(peak > 0.01, `host hears the relayed guest (peak ${peak})`); + assert.ok(await g.page.evaluate(() => window.streamFreeGuest.relay.audioPeak) > 0.01, 'the relayed guest hears the studio'); + await studio.evaluate((i) => window.streamFree.toggleStage(i), id); + await waitFor(g.page, () => window.streamFreeGuest.onStage === true); + const routing = await studio.evaluate(() => window.streamFree.routing()); + assert.ok(routing[id].includes('program'), 'the relayed guest is in the program'); + assert.ok(!routing[id].includes(`guest:${id}`), 'the relayed guest does not hear themselves'); + assert.ok(routing.host.includes(`guest:${id}`), 'the relayed guest hears the host'); + // A recording with the relayed guest on stage is as good as any other. + await studio.evaluate(() => window.streamFree.startBroadcast('record', [])); + await waitFor(studio, () => window.streamFree.live.state === 'live'); + await sleep(SECONDS * 1000); + await studio.evaluate(() => window.streamFree.stopBroadcast()); + await waitFor(studio, () => window.streamFree.live.state === 'idle', null, 30000); + checkMedia(newestRecording(), 'recording with a relayed guest'); + assert.deepEqual(g.errors, []); + await g.ctx.close(); + await waitFor(studio, (i) => !window.streamFree.people.has(i), id); +}); + +await test('a guest who loses the connection comes back on stage', async () => { + const { inviteUrl } = await sharingStatus(); + const g = await joinGuest(inviteUrl, 'Wobbly'); + await receivesProgram(g.page); + const id = await guestId('Wobbly'); + await studio.evaluate((i) => window.streamFree.toggleStage(i), id); + await waitFor(g.page, () => window.streamFreeGuest.onStage === true); + await g.page.evaluate(() => window.streamFreeGuest.dropConnection()); + await waitFor(studio, (old) => { + const p = [...window.streamFree.people.values()].find((x) => x.name === 'Wobbly'); + return p && p.id !== old && window.streamFree.stage.includes(p.id); + }, id, 20000); + await waitFor(g.page, () => window.streamFreeGuest.onStage === true); + await receivesProgram(g.page); + assert.deepEqual(g.errors, []); + await g.ctx.close(); +}); + +await test('"Always through streamFree" sends a guest to the relay at once', async () => { + await studio.evaluate(() => window.streamFree.setGuestTransport('relay')); + try { + const { inviteUrl } = await sharingStatus(); + const t0 = Date.now(); + const g = await joinGuest(inviteUrl, 'Relayed'); + await receivesRelay(g.page, 15000); + assert.ok(Date.now() - t0 < 9000, `the relay took ${Date.now() - t0} ms`); + assert.equal(await g.page.evaluate(() => window.streamFreeGuest.pc), null, 'no WebRTC connection'); + const id = await guestId('Relayed'); + await waitFor(studio, (i) => { + const s = window.streamFree.relayStats()[i]; + return s && s.transport === 'relay' && s.videoFrames > 20 && s.audioPackets > 20; + }, id); + assert.deepEqual(g.errors, []); + await g.ctx.close(); + } finally { + await studio.evaluate(() => window.streamFree.setGuestTransport('auto')); + } +}); + await test('no more than 16 guests can join', async () => { const { inviteUrl } = await sharingStatus(); const invite = inviteUrl.split('/join/')[1]; @@ -358,6 +474,11 @@ await test('guests on the internet join through the tunnel', async () => { const g = await joinGuest(status.inviteUrl, 'Remote'); await receivesProgram(g.page, 40000); await g.ctx.close(); + // And a guest with no direct path, whose media must cross the tunnel. + const r = await joinRelayOnlyGuest(status.inviteUrl, 'Remote relay'); + await receivesRelay(r.page, 45000); + assert.deepEqual(r.errors, []); + await r.ctx.close(); const back = await api('PUT', '/api/sharing', { mode: 'local' }); assert.equal(back.tunnel.state, 'off'); }); diff --git a/web/css/app.css b/web/css/app.css index 363f11d..44e0044 100644 --- a/web/css/app.css +++ b/web/css/app.css @@ -126,6 +126,8 @@ textarea { font-family: ui-monospace, Menlo, monospace; font-size: 12px; resize: aspect-ratio: 16 / 9; background: #000; border-radius: 12px; overflow: hidden; box-shadow: var(--shadow); } .preview-wrap canvas, .preview-wrap > video:not(.self) { width: 100%; height: 100%; display: block; object-fit: contain; background: #000; } +.preview-wrap [hidden] { display: none !important; } +.preview-wrap .tap-to-play { position: absolute; left: 50%; top: 50%; transform: translate(-50%, -50%); z-index: 2; } .live-badge { position: absolute; top: 12px; left: 12px; padding: 3px 10px; border-radius: 6px; background: var(--live); color: #fff; font-weight: 800; font-size: 12px; letter-spacing: 0.06em; diff --git a/web/js/audio.js b/web/js/audio.js index 937a3a9..35bf60f 100644 --- a/web/js/audio.js +++ b/web/js/audio.js @@ -26,7 +26,7 @@ export class Mixer { this.monitor.connect(this.ctx.destination); this.monitorEnabled = true; this.sources = new Map(); // id -> { input, gain, analyser, kind, element } - this.guestBuses = new Map(); // guest id -> MediaStreamAudioDestinationNode + this.guestBuses = new Map(); // guest id -> { sum: GainNode, dest: MediaStreamAudioDestinationNode } this.onStage = new Set(); this.featured = null; this.muted = new Set(); @@ -75,21 +75,34 @@ export class Mixer { this.capture?.port.postMessage({ active: false }); } - /** The return mix for one guest; created on first use. */ - guestTrack(guestId) { + guestBus(guestId) { let bus = this.guestBuses.get(guestId); if (!bus) { - bus = this.ctx.createMediaStreamDestination(); + const sum = this.ctx.createGain(); + const dest = this.ctx.createMediaStreamDestination(); + sum.connect(dest); + bus = { sum, dest }; this.guestBuses.set(guestId, bus); this.route(); } - return bus.stream.getAudioTracks()[0]; + return bus; + } + + /** The return mix for one guest, as a track for WebRTC; created on first use. */ + guestTrack(guestId) { + return this.guestBus(guestId).dest.stream.getAudioTracks()[0]; + } + + /** The same return mix as an audio node, for the relay. */ + guestBusNode(guestId) { + return this.guestBus(guestId).sum; } removeGuestBus(guestId) { const bus = this.guestBuses.get(guestId); if (!bus) return; - bus.stream.getTracks().forEach((t) => t.stop()); + bus.sum.disconnect(); + bus.dest.stream.getTracks().forEach((t) => t.stop()); this.guestBuses.delete(guestId); this.route(); } @@ -97,13 +110,16 @@ export class Mixer { /** * @param {string} id participant or source id * @param {'host'|'guest'|'screen'|'media'} kind - * @param {MediaStream|HTMLMediaElement} input + * @param {MediaStream|HTMLMediaElement|AudioNode} input (an AudioNode: a + * relayed guest's player) */ add(id, kind, input) { this.remove(id); let node; let element = null; - if (input instanceof HTMLMediaElement) { + if (input instanceof AudioNode) { + node = input; + } else if (input instanceof HTMLMediaElement) { node = this.ctx.createMediaElementSource(input); element = input; } else { @@ -153,7 +169,7 @@ export class Mixer { if (inProgram) s.gain.connect(this.program); for (const [guestId, bus] of this.guestBuses) { if (guestId === id) continue; - if (s.kind === 'screen' || s.kind === 'media' ? inProgram : true) s.gain.connect(bus); + if (s.kind === 'screen' || s.kind === 'media' ? inProgram : true) s.gain.connect(bus.sum); } const hearIt = s.kind === 'host' ? false : s.kind === 'guest' || inProgram; if (hearIt && this.monitorEnabled) s.gain.connect(this.monitor); diff --git a/web/js/guest.js b/web/js/guest.js index 8c6cb58..e7a0965 100644 --- a/web/js/guest.js +++ b/web/js/guest.js @@ -1,14 +1,24 @@ -// Guest page: preview camera and microphone, join the studio over WebRTC, -// watch the program and talk to the host. Works in any modern browser. +// Guest page: preview camera and microphone, join the studio, watch the +// program and talk to the host. Works in any modern browser. +// +// Media normally flows over WebRTC, directly between this browser and the +// host. When the host cannot reach us directly it sends `relay-start` and we +// switch to the relay (relay.js): the same media, encoded here and carried +// over a WebSocket through the streamFree server. import { h, icon, toast, applyTheme } from './util.js'; import { t, setLang, detectLang } from './i18n.js'; +import { + VIDEO, AUDIO, relayCaps, pickVideoCodec, parsePacket, loadRelayWorklets, + VideoSender, VideoReceiver, AudioSender, AudioReceiver, +} from './relay.js'; applyTheme(); // Guests see the page in their own browser's language. setLang(detectLang()); document.title = t('Join streamFree'); +const CLOSE_REMOVED = 4001; const invite = decodeURIComponent(location.pathname.split('/').pop() || ''); const app = document.querySelector('#app'); let local = null; @@ -22,6 +32,14 @@ let onStage = false; let chat = []; let queue = Promise.resolve(); let leaving = false; +let myName = ''; +let myId = ''; +let relayKey = ''; +let iceServers = []; +let audioCtx = null; +let connectedOnce = false; +let retries = 0; +let relay = null; // { ws, videoSend, audioSend, video, audio, frames } function savedName() { try { @@ -31,6 +49,20 @@ function savedName() { } } +/** A random id kept for this tab, so the host recognizes us after a reconnect. */ +function clientId() { + try { + let id = sessionStorage.getItem('streamfree-guest-cid'); + if (!id) { + id = crypto.randomUUID(); + sessionStorage.setItem('streamfree-guest-cid', id); + } + return id; + } catch { + return ''; + } +} + async function getMedia(camId, micId) { local?.getTracks().forEach((t) => t.stop()); const constraints = { @@ -83,6 +115,9 @@ async function lobby() { h('form', { class: 'stack', onsubmit: (e) => { e.preventDefault(); + // Safari only lets audio start from a user gesture: create the + // audio context now, before anything asynchronous happens. + startAudio(); try { localStorage.setItem('streamfree-guest-name', name.value.trim()); } catch { @@ -99,6 +134,17 @@ async function lobby() { await refresh(); } +function startAudio() { + if (audioCtx) return; + const Ctx = window.AudioContext || window.webkitAudioContext; + try { + audioCtx = new Ctx({ sampleRate: 48000, latencyHint: 'interactive' }); + } catch { + audioCtx = new Ctx(); + } + audioCtx.resume().catch(() => {}); +} + function sendSignal(msg) { if (ws?.readyState === WebSocket.OPEN) ws.send(JSON.stringify(msg)); } @@ -108,56 +154,101 @@ async function join(name) { // even if the guest clicks while the permission prompt is still open. await mediaReady; if (!local) local = await getMedia('', ''); + myName = name; + renderRoom(name); + connect(); +} + +function connect() { const proto = location.protocol === 'https:' ? 'wss:' : 'ws:'; - ws = new WebSocket(`${proto}//${location.host}/ws/guest?invite=${encodeURIComponent(invite)}&name=${encodeURIComponent(name)}`); - ws.onmessage = (e) => { + const params = new URLSearchParams({ invite, name: myName, cid: clientId() }); + const socket = new WebSocket(`${proto}//${location.host}/ws/guest?${params}`); + ws = socket; + socket.onmessage = (e) => { const msg = JSON.parse(e.data); // Handle messages strictly in order: ICE candidates must wait for the offer. queue = queue.then(() => onMessage(msg)).catch((err) => toast(err.message, 'error')); }; - ws.onclose = () => { - pc?.close(); - local?.getTracks().forEach((t) => t.stop()); - if (!leaving) ended(t('The connection to the studio ended. The host may have removed you or closed the studio.')); + socket.onclose = (e) => { + if (ws !== socket) return; + closeMedia(); + if (leaving) return; + if (e.code === CLOSE_REMOVED) { + endSession(t('The host removed you from the studio.')); + return; + } + // A dropped connection (network change, a proxy hiccup): try again for + // about a minute. The host puts us back on stage if we were on it. + if (connectedOnce && retries < 8) { + retries++; + setStatus(t('Reconnecting…')); + setTimeout(() => !leaving && connect(), Math.min(8000, 1000 * retries)); + return; + } + endSession(t('The connection to the studio ended. The host may have closed the studio.')); }; - renderRoom(name); } -let iceServers = []; +function closeMedia() { + pc?.close(); + pc = null; + stopRelay(); +} + +function endSession(message) { + ws = null; + local?.getTracks().forEach((t) => t.stop()); + ended(message); +} async function onMessage(m) { switch (m.type) { case 'welcome': + connectedOnce = true; + retries = 0; + myId = m.id; + relayKey = m.relayKey || ''; iceServers = m.iceServers || []; sendSignal({ type: 'state', micOn, camOn }); + setStatus(); break; case 'offer': { + stopRelay(); pc?.close(); - pc = new RTCPeerConnection({ iceServers }); - pc.onicecandidate = (e) => e.candidate && sendSignal({ type: 'candidate', candidate: e.candidate }); - pc.ontrack = (e) => { - const program = document.querySelector('#program'); - const stream = program.srcObject || new MediaStream(); + const conn = new RTCPeerConnection({ iceServers }); + pc = conn; + const program = document.querySelector('#program'); + const stream = new MediaStream(); + program.srcObject = stream; + showProgram('video'); + conn.onicecandidate = (e) => e.candidate && sendSignal({ type: 'candidate', candidate: e.candidate }); + conn.ontrack = (e) => { stream.addTrack(e.track); - program.srcObject = stream; - program.play().catch(() => {}); + program.play().catch(() => askToPlay()); }; - pc.onconnectionstatechange = () => setStatus(); - await pc.setRemoteDescription({ type: 'offer', sdp: m.sdp }); - for (const t of pc.getTransceivers()) { - const kind = t.receiver.track.kind; + conn.onconnectionstatechange = () => setStatus(); + await conn.setRemoteDescription({ type: 'offer', sdp: m.sdp }); + for (const tr of conn.getTransceivers()) { + const kind = tr.receiver.track.kind; const track = kind === 'video' ? local.getVideoTracks()[0] : local.getAudioTracks()[0]; - if (track) await t.sender.replaceTrack(track); - t.direction = 'sendrecv'; + if (track) await tr.sender.replaceTrack(track); + tr.direction = 'sendrecv'; } - const answer = await pc.createAnswer(); - await pc.setLocalDescription(answer); + const answer = await conn.createAnswer(); + await conn.setLocalDescription(answer); sendSignal({ type: 'answer', sdp: answer.sdp }); break; } case 'candidate': if (pc && m.candidate) await pc.addIceCandidate(m.candidate).catch(() => {}); break; + case 'relay-start': + await startRelay(m.caps || {}); + break; + case 'relay-ok': + if (relay) relay.hostReceiving = true; + setStatus(); + break; case 'stage': onStage = !!m.onStage; setStatus(); @@ -170,15 +261,149 @@ async function onMessage(m) { } } -function setStatus() { +// ---------------------------------------------------------------- relay + +async function startRelay(hostDecode) { + pc?.close(); + pc = null; + stopRelay(); + startAudio(); + await loadRelayWorklets(audioCtx); + const mine = await relayCaps(); + const proto = location.protocol === 'https:' ? 'wss:' : 'ws:'; + const params = new URLSearchParams({ invite, id: myId, key: relayKey }); + const socket = new WebSocket(`${proto}//${location.host}/ws/relay/guest?${params}`); + socket.binaryType = 'arraybuffer'; + const r = { ws: socket, frames: 0 }; + relay = r; + const sendBin = (buf) => socket.readyState === WebSocket.OPEN && socket.send(buf); + socket.onopen = () => { + if (relay !== r) return; + const camera = local.getVideoTracks()[0]; + const codec = pickVideoCodec(mine, hostDecode.video); + if (camera && codec) { + r.videoSend = new VideoSender({ send: sendBin, codec, maxSide: 640, fps: 20, canSend: () => socket.bufferedAmount < 1_500_000 }); + r.videoSend.setTrack(camera); + } + if (local.getAudioTracks().length) { + r.micSource = audioCtx.createMediaStreamSource(local); + r.audioSend = new AudioSender({ ctx: audioCtx, source: r.micSource, send: sendBin, opus: mine.encode.opus && hostDecode.opus }); + } + r.video = new VideoReceiver({ + onFrame: (frame) => drawProgram(r, frame), + requestKey: () => socket.readyState === WebSocket.OPEN && socket.send(JSON.stringify({ type: 'key' })), + }); + r.audio = new AudioReceiver({ ctx: audioCtx }); + r.audio.node.connect(audioCtx.destination); + if (audioCtx.state !== 'running') audioCtx.resume().catch(() => askToPlay()); + showProgram('canvas'); + sendSignal({ type: 'relay-ready', caps: mine.decode }); + setStatus(); + }; + socket.onmessage = (e) => { + if (relay !== r) return; + if (typeof e.data === 'string') { + // The host lost some of our video and needs a keyframe. + if (JSON.parse(e.data).type === 'key') r.videoSend?.requestKey(); + return; + } + let p; + try { + p = parsePacket(e.data); + } catch { + return; + } + if (p.kind === VIDEO) r.video?.push(p); + else if (p.kind === AUDIO) r.audio?.push(p); + }; + socket.onclose = () => { + if (relay !== r) return; + stopRelay(); + // The signaling socket decides whether we are still in the studio. + if (ws?.readyState === WebSocket.OPEN) setTimeout(() => ws?.readyState === WebSocket.OPEN && startRelay(hostDecode), 1000); + }; + setStatus(); +} + +function stopRelay() { + const r = relay; + if (!r) return; + relay = null; + r.videoSend?.stop(); + r.audioSend?.stop(); + r.micSource?.disconnect(); + r.video?.close(); + r.audio?.close(); + try { + r.ws.close(); + } catch { + // already closed + } +} + +function drawProgram(r, frame) { + const canvas = document.querySelector('#program-canvas'); + if (!canvas || relay !== r) { + frame.close(); + return; + } + if (canvas.width !== frame.displayWidth || canvas.height !== frame.displayHeight) { + canvas.width = frame.displayWidth; + canvas.height = frame.displayHeight; + } + canvas.getContext('2d').drawImage(frame, 0, 0); + frame.close(); + if (r.frames++ === 0) setStatus(); +} + +/** Shows the program as a