slop

2026-07-31 20:20:48 +01:00
parent d4b77bd717
commit c1435a70fb
12 changed files with 1009 additions and 168 deletions
@@ -1,17 +1,561 @@
-use std::sync::Arc;
+use std::{net::SocketAddr, sync::Arc, time::Instant};

-use axum::{extract::State, response::IntoResponse};
-use str0m::change::SdpOffer;
-use tracing::info;
+use async_broadcast::broadcast;
+use axum::{
+    extract::{ConnectInfo, Path, State},
+    http::{HeaderMap, StatusCode, header},
+    response::IntoResponse,
+};
+use bytes::Bytes;
+use chrono::{Local, Utc};
+use dashmap::DashMap;
+use entity::{stream_key, stream_session};
+use sea_orm::{DatabaseConnection, IntoActiveModel};
+use str0m::{
+    Candidate, Event, Input, Output, Rtc,
+    change::SdpOffer,
+    media::{MediaKind, Mid},
+    net::{Protocol, Receive},
+};
+use tokio::{net::UdpSocket, sync::mpsc::Receiver};
+use tracing::{debug, error, info, warn};

-use crate::http::HttpServer;
+use crate::{
+    StreamSession, audio::OpusAudioFrame, codec::VideoFrame, http::HttpServer,
+    http_error::HttpError,
+};
+
+pub async fn handle_whip_injest_delete(
+    State(state): State<Arc<HttpServer>>,
+    ConnectInfo(remote): ConnectInfo<SocketAddr>,
+    headers: HeaderMap,
+) -> Result<impl IntoResponse, HttpError> {
+    let token = headers
+        .get(header::AUTHORIZATION)
+        .and_then(|v| v.to_str().ok())
+        .and_then(|v| v.strip_prefix("Bearer "))
+        .map(|s| s.trim().to_string());
+    let Some(token) = token else {
+        warn!("Whip: missing bearer token from {}", remote);
+        return Err(HttpError::Unauthorized);
+    };
+
+    let key = stream_key::Entity::find_by_key(&state.db, &token)
+        .await?
+        .ok_or(HttpError::Unauthorized)?;
+
+    let active_session = stream_session::Model::get_active_by_stream_key_id(&state.db, key.id)
+        .await?
+        .ok_or(HttpError::NotFound)?;
+
+    active_session
+        .into_active_model()
+        .finish_stream_session(&state.db, Utc::now())
+        .await?;
+
+    Ok(StatusCode::OK)
+}
+
+/// PATCH /api/whip/{id} — Trickle ICE candidate delivery.
+pub async fn handle_whip_injest_patch(
+    State(state): State<Arc<HttpServer>>,
+    Path(slug): Path<String>,
+    headers: HeaderMap,
+    body: String,
+) -> Result<impl IntoResponse, HttpError> {
+    // The slug is the stream_key_id (set by the POST handler's Location header).
+    let stream_key_id: i32 = slug
+        .parse()
+        .map_err(|e| {
+warn!(%slug, "Whip PATCH: bad slug: {:?}", e);
+HttpError::NotFound
+        })?;
+
+    info!(stream_key_id, ct = ?headers.get(header::CONTENT_TYPE), "Whip PATCH: trickle candidate");
+
+    // Look up the trickle sender for this session.
+    let trickle_map = state
+        .appstate
+        .lock()
+        .await
+        .webrtc_proxy
+        .trickle_tx
+        .clone();
+    let tx = trickle_map
+        .get(&stream_key_id)
+        .ok_or_else(|| {
+warn!(stream_key_id, "Whip PATCH: no trickle channel for session");
+HttpError::NotFound
+        })?;
+
+    debug!(stream_key_id, %body, "Whip PATCH: forwarding trickle candidate");
+    tx.send(body).ok();
+
+    Ok(StatusCode::NO_CONTENT)
+}

 pub async fn handle_whip_injest(
    State(state): State<Arc<HttpServer>>,
+    ConnectInfo(remote): ConnectInfo<SocketAddr>,
+    headers: HeaderMap,
    offer: String,
-) -> impl IntoResponse {
-    let sdp_offer = SdpOffer::from_sdp_string(&offer).unwrap();
-    for x in &sdp_offer.media_lines {
-        info!("{}", x);
+) -> axum::response::Response {
+    let err = |status, msg: &str| -> axum::response::Response {
+        (
+            status,
+            [(header::CONTENT_TYPE, "text/plain")],
+            msg.to_string(),
+        )
+            .into_response()
+    };
+    let public_addr = state.appstate.lock().await.webrtc_proxy.public_addr();
+
+    let token = headers
+        .get(header::AUTHORIZATION)
+        .and_then(|v| v.to_str().ok())
+        .and_then(|v| v.strip_prefix("Bearer "))
+        .map(|s| s.trim().to_string());
+    let Some(token) = token else {
+        warn!("Whip: missing bearer token from {}", remote);
+        return err(StatusCode::UNAUTHORIZED, "missing bearer token");
+    };
+
+    let key = match stream_key::Entity::find_by_key(&state.db, &token).await {
+        Ok(Some(key)) => key,
+        Ok(None) => {
+            warn!("Whip: stream key not found, rejecting {}", remote);
+            return err(StatusCode::UNAUTHORIZED, "invalid stream key");
+        }
+        Err(e) => {
+            error!("Whip: DB error looking up stream key: {:?}", e);
+            return err(StatusCode::INTERNAL_SERVER_ERROR, "db error");
+        }
+    };
+
+    match stream_session::Model::get_active_by_stream_key_id(&state.db, key.id).await {
+        Ok(Some(_)) => {
+            warn!(stream_key_id = key.id, label = %key.label, "Whip: stream key already live, rejecting duplicate publish from {}", remote);
+            return err(StatusCode::CONFLICT, "stream already live");
+        }
+        Ok(None) => {}
+        Err(e) => {
+            error!("Whip: DB error checking active stream: {:?}", e);
+            return err(StatusCode::INTERNAL_SERVER_ERROR, "db error");
+        }
+    };
+    info!(stream_key_id = key.id, label = %key.label, "Whip authenticated stream key from {}", remote);
+
+        // Parse the SDP offer first so we can detect the codec and
+        // configure the Rtc before accepting.
+        let sdp_offer = match SdpOffer::from_sdp_string(&offer) {
+            Ok(o) => o,
+            Err(e) => {
+                error!("Whip: cant parse offer from {}: {:?}", remote, e);
+                return err(StatusCode::BAD_REQUEST, "invalid SDP offer");
+            }
+        };
+        info!(
+            "Whip offer from {} — {} media line(s):",
+            remote,
+            sdp_offer.media_lines.len()
+        );
+        for x in &sdp_offer.media_lines {
+            info!("  {}", x);
+        }
+
+        // Detect codec from the offer so we can enable matching codecs.
+        let stream_codec = video_codec_from_sdp_offer(&sdp_offer);
+        info!(?stream_codec, "detected codec from WHIP offer");
+
+        // Build Rtc with ICE-Lite (required by WHIP RFC 9728 §4.1) and
+        // matching codecs enabled.
+        // Not using ICE-Lite: full ICE lets the server initiate checks
+        // when OBS hasn't sent its candidates yet (Trickle ICE without PATCH).
+        let mut builder = Rtc::builder();
+        {
+            let cc = builder.codec_config();
+            cc.clear();
+            cc.enable_opus(true);
+            match &stream_codec {
+                Some(crate::StreamCodec::H264) => {
+                    info!("Whip: enabling H.264 codec");
+                    cc.enable_h264(true);
+                }
+                Some(crate::StreamCodec::H265) => {
+                    info!("Whip: enabling H.265 codec");
+                    cc.enable_h265(true);
+                }
+                Some(crate::StreamCodec::AV1) => {
+                    info!("Whip: enabling AV1 codec");
+                    cc.enable_av1(true);
+                }
+                None => {
+                    warn!("Whip: no video codec detected in offer, enabling H.264 as fallback");
+                    cc.enable_h264(true);
+                }
+            }
+        }
+        let mut rtc = builder.build(Instant::now());
+        let candidate = Candidate::host(public_addr, Protocol::Udp).unwrap();
+        rtc.add_local_candidate(candidate);
+        info!(%public_addr, "Whip: added local ICE candidate, accepting offer…");
+
+        let offer_answe = match rtc.sdp_api().accept_offer(sdp_offer) {
+            Ok(a) => a,
+            Err(e) => {
+                error!("cant accept inject offer: {:?}", e);
+                return err(StatusCode::BAD_REQUEST, "could not accept offer");
+            }
+        };
+
+        // OBS sends no a=candidate: lines (disableAutoGathering). Derive a
+        // remote host candidate from the HTTP source address so the server
+        // has somewhere to send STUN checks.
+        if let Ok(c) = Candidate::host(remote, Protocol::Udp) {
+            info!(%remote, "Whip: no candidates in offer, adding HTTP-derived remote host candidate");
+            rtc.add_remote_candidate(c);
+        }
+
+        // Extract negotiated codec, PT, and profile from the answer
+        // so WHEP viewers can use the exact same codec config.
+        let (negotiated_codec, video_pt, video_profile) =
+            extract_negotiated_codec_info(&offer_answe);
+        info!(
+            ?negotiated_codec,
+            video_pt,
+            ?video_profile,
+            "negotiated codec from WHIP answer"
+        );
+        if negotiated_codec.is_none() || video_pt.is_none() {
+            warn!("Whip: no common video codec negotiated, rejecting");
+            return err(StatusCode::NOT_ACCEPTABLE, "no common video codec");
+        }
+
+        let answer_sdp = offer_answe.to_sdp_string()
+            // Strip a=ice-options:trickle so OBS starts ICE immediately.
+            .replace("a=ice-options:trickle\r\n", "")
+            .replace("a=ice-options:trickle\n", "");
+        // Fix up the answer for libdatachannel (OBS WHIP):
+        // 1. Strip a=group:BUNDLE — OBS doesn't negotiate it.
+        // 2. Add the host candidate to the video m= line.
+        let answer_sdp = answer_sdp
+            .replace("a=group:BUNDLE 0 1\r\n", "")
+            .replace("a=group:BUNDLE 0 1\n", "");
+        let answer_sdp = if let Some(cand_line) = answer_sdp
+            .lines()
+            .find(|l| l.starts_with("a=candidate:"))
+        {
+            // Insert the candidate line after the video m= line.
+            let cand_replacement = format!("m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 127.0.0.1\r\n{}\r\n", cand_line);
+            answer_sdp.replace("m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 0.0.0.0\r\n", &cand_replacement)
+        } else {
+            answer_sdp
+        };
+        info!("Serving Whip SDP answer to {}:\n{}", remote, answer_sdp);
+
+    let ufrag = answer_sdp
+        .lines()
+        .find(|l| l.starts_with("a=ice-ufrag:"))
+        .and_then(|l| l.strip_prefix("a=ice-ufrag:"))
+        .map(|s| s.trim().to_string());
+    let Some(ufrag) = ufrag else {
+        error!("Whip: answer has no a=ice-ufrag");
+        return err(StatusCode::INTERNAL_SERVER_ERROR, "no ice-ufrag in answer");
+    };
+    let ice_pwd = answer_sdp
+        .lines()
+        .find(|l| l.starts_with("a=ice-pwd:"))
+        .and_then(|l| l.strip_prefix("a=ice-pwd:"))
+        .map(|s| s.trim().to_string());
+    info!(?ufrag, ?ice_pwd, "Whip: registering ICE credentials with proxy");
+    let (socket, rx) = state.appstate.lock().await.webrtc_proxy.add_client(ufrag);
+
+    let stream_sessions = state.appstate.lock().await.stream_sessions.clone();
+    let (video_tx, video_rx) = broadcast::<Arc<VideoFrame>>(4);
+    let (audio_tx, audio_rx) = broadcast::<Arc<OpusAudioFrame>>(4);
+
+    stream_sessions.insert(
+        key.id,
+        StreamSession {
+            stream_key_id: key.id,
+            stream_key_label: key.label,
+            frame_channel: video_tx,
+            audio_channel: audio_tx,
+            codec: negotiated_codec,
+            started_at: Utc::now(),
+            active_clients: 0.into(),
+            video_pt,
+            video_profile_level_id: video_profile,
+        },
+    );
+    info!(stream_key_id = key.id, "Whip: StreamSession inserted, spawning detach task");
+
+    // Trickle-ICE channel: OBS can send candidates via PATCH after the
+    // initial offer. We forward them to the Rtc task for add_remote_candidate.
+    let (trickle_tx, trickle_rx) = tokio::sync::mpsc::unbounded_channel();
+    let trickle_map = state.appstate.lock().await.webrtc_proxy.trickle_tx.clone();
+    trickle_map.insert(key.id, trickle_tx);
+
+    let db = state.db.clone();
+    tokio::spawn(async move {
+        detach_inject_rtc(
+            stream_sessions,
+            key.id,
+            db,
+            socket,
+            rx,
+            trickle_rx,
+            rtc,
+            public_addr,
+            video_rx,
+            audio_rx,
+        )
+        .await;
+        // Clean up trickle channel when done.
+        trickle_map.remove(&key.id);
+    });
+
+    stream_session::Model::create_stream_session(&state.db, key.id, Local::now().into())
+        .await
+        .ok();
+
+    let location = format!("/api/whip/{}", key.id);
+    axum::response::Response::builder()
+        .status(StatusCode::CREATED)
+        .header(header::CONTENT_TYPE, "application/sdp")
+        .header(header::LOCATION, &location)
+        // Provide a STUN server via Link header so OBS can gather
+        // ICE candidates even without explicit STUN configuration.
+        .header(header::LINK, "<stun:stun.l.google.com:19302>; rel=\"ice-server\"")
+        .body(axum::body::Body::from(answer_sdp))
+        .unwrap()
+}
+
+async fn detach_inject_rtc(
+    sessions_ref: Arc<DashMap<i32, StreamSession>>,
+    stream_key_id: i32,
+    db: DatabaseConnection,
+    socket: Arc<UdpSocket>,
+    mut rx: Receiver<(Bytes, SocketAddr)>,
+    mut trickle_rx: tokio::sync::mpsc::UnboundedReceiver<String>,
+    mut rtc: Rtc,
+    local_addr: SocketAddr,
+    video_rx: async_broadcast::Receiver<Arc<VideoFrame>>,
+    audio_rx: async_broadcast::Receiver<Arc<OpusAudioFrame>>,
+) {
+    let cleanup = || {
+        let sessions_ref = sessions_ref.clone();
+        let db = db.clone();
+        async move {
+            sessions_ref.remove(&stream_key_id);
+            if let Ok(Some(s)) =
+                stream_session::Model::get_active_by_stream_key_id(&db, stream_key_id).await
+            {
+                s.into_active_model()
+                    .finish_stream_session(&db, Local::now().into())
+                    .await
+                    .ok();
+            }
+        }
+    };
+    let mut video_mid: Option<Mid> = None;
+    let mut audio_mid: Option<Mid> = None;
+    let mut video_tx: Option<async_broadcast::Sender<Arc<VideoFrame>>> = None;
+    let mut audio_tx: Option<async_broadcast::Sender<Arc<OpusAudioFrame>>> = None;
+    let mut _connected = false;
+
+    loop {
+        // Blankly using them so they don't drop (like RTMP).
+        let _ = video_rx.is_closed();
+        let _ = audio_rx.is_closed();
+
+        let deadline = loop {
+            match rtc.poll_output() {
+                Ok(Output::Timeout(t)) => break t,
+                    Ok(Output::Transmit(t)) => {
+                        debug!(
+                            "Whip TX: {} bytes → {}:{}",
+                            t.contents.len(),
+                            t.destination.ip(),
+                            t.destination.port()
+                        );
+                        if let Err(e) = socket.send_to(&t.contents, t.destination).await {
+                        warn!("Whip UDP send error: {:?}", e);
+                        cleanup().await;
+                        return;
+                    }
+                }
+                Ok(Output::Event(e)) => match e {
+                    Event::MediaAdded(ma) => {
+                        info!(stream_key_id, kind = ?ma.kind, mid = ?ma.mid, "Whip MediaAdded");
+                        if ma.kind == MediaKind::Video {
+                            video_mid = Some(ma.mid);
+                        }
+                        if ma.kind == MediaKind::Audio {
+                            audio_mid = Some(ma.mid);
+                        }
+                    }
+                    Event::MediaData(md) => {
+                        // str0m depacketizes RTP into full codec frames (no decode).
+                        // H.264/H.265 → Annex-B, AV1 → OBU, Opus → raw Opus packets.
+                        // Same format as our RTMP codec parsers produce — push directly.
+                        if Some(md.mid) == video_mid {
+                            let ts_ms = (md.time.as_seconds() * 1000.0) as u32;
+                            let frame = VideoFrame {
+                                data: Bytes::copy_from_slice(&md.data),
+                                is_keyframe: md.is_keyframe(),
+                                timestamp_ms: ts_ms,
+                            };
+                            if let Some(tx) = &video_tx {
+                                tx.broadcast(Arc::new(frame)).await.ok();
+                            }
+                        }
+                        if Some(md.mid) == audio_mid {
+                            let ts_ms = (md.time.as_seconds() * 1000.0) as u32;
+                            let frame = OpusAudioFrame {
+                                data: Bytes::copy_from_slice(&md.data),
+                                timestamp_ms: ts_ms,
+                            };
+                            if let Some(tx) = &audio_tx {
+                                tx.broadcast(Arc::new(frame)).await.ok();
+                            }
+                        }
+                    }
+                    Event::IceConnectionStateChange(state) => {
+                        info!(stream_key_id, ?state, "Whip ICE state change");
+                        if matches!(state, str0m::IceConnectionState::Disconnected) {
+                            info!("Whip ICE disconnected, closing connection");
+                            cleanup().await;
+                            return;
+                        }
+                    }
+                    Event::Connected => {
+                        info!(stream_key_id, "Whip DTLS+ICE connected, wiring broadcast channels");
+                        if let Some(session) = sessions_ref.get(&stream_key_id) {
+                            video_tx = Some(session.frame_channel.clone());
+                            audio_tx = Some(session.audio_channel.clone());
+                        }
+                        _connected = true;
+                    }
+                    _ => {}
+                },
+                Err(e) => {
+                    error!("Whip poll_output error (closing): {:?}", e);
+                    cleanup().await;
+                    return;
+                }
+            }
+        };
+
+        let sleep = tokio::time::sleep_until(deadline.max(Instant::now()).into());
+            tokio::select! {
+                _ = sleep => {
+                    if let Err(e) = rtc.handle_input(Input::Timeout(Instant::now())) {
+                        error!(stream_key_id, "Whip handle_input(Timeout) error: {:?}", e);
+                        cleanup().await;
+                        return;
+                    }
+                }
+                // Trickle-ICE candidates from PATCH /api/whip/{id}
+                Some(candidate_line) = trickle_rx.recv() => {
+                    info!(stream_key_id, %candidate_line, "Whip: received trickle candidate");
+                    // Parse the candidate string (without "a=candidate:" prefix if present).
+                    let cand_str = candidate_line
+                        .strip_prefix("a=candidate:")
+                        .unwrap_or(&candidate_line);
+                    match str0m::Candidate::from_sdp_string(cand_str) {
+                        Ok(c) => {
+                            info!(stream_key_id, "Whip: adding remote candidate");
+                            rtc.add_remote_candidate(c);
+                        }
+                        Err(e) => {
+                            warn!(stream_key_id, %cand_str, "Whip: bad trickle candidate: {:?}", e);
+                        }
+                    }
+                }
+                result = rx.recv() => {
+                let Some((data, from)) = result else {
+                    info!(stream_key_id, "Whip proxy channel closed, cleaning up");
+                    cleanup().await;
+                    return;
+                };
+                debug!(
+                    stream_key_id,
+                    len = data.len(),
+                    %from,
+                    "Whip RX: {} bytes",
+                    data.len()
+                );
+                if let Ok(contents) = (&data[..]).try_into()
+                    && let Err(e) = rtc.handle_input(Input::Receive(
+                        Instant::now(),
+                        Receive {
+                            proto: Protocol::Udp,
+                            source: from,
+                            destination: local_addr,
+                            contents,
+                        },
+                    )) {
+                        error!("Whip handle_input(Receive) error: {:?}", e);
+                        cleanup().await;
+                        return;
+                    }
+            }
+        }
    }
 }
+
+/// Extract the negotiated video codec, PT, and profile_level_id from an SDP answer.
+///
+/// Walks the answer's media lines, finds the first video m-line with
+/// negotiated rtp_params, and returns the codec + PT + H.264 profile.
+pub fn extract_negotiated_codec_info(
+answer: &str0m::change::SdpAnswer,
+) -> (Option<crate::StreamCodec>, Option<u8>, Option<u32>) {
+use str0m::format::Codec;
+
+for line in answer.media_lines.iter() {
+// Iterate rtp_params on every m-line; video check via
+// codec.is_video() avoids needing str0m's private MediaType.
+for p in line.rtp_params() {
+if p.spec().codec.is_video() {
+let pt = Some(*p.pt());
+let profile = p.spec().format.profile_level_id;
+let codec = match p.spec().codec {
+Codec::H264 => Some(crate::StreamCodec::H264),
+Codec::H265 => Some(crate::StreamCodec::H265),
+Codec::Av1 => Some(crate::StreamCodec::AV1),
+_ => None,
+};
+return (codec, pt, profile);
+}
+}
+}
+(None, None, None)
+}
+
+/// Extract the video codec from an SDP offer's media lines.
+///
+/// Walks every m= line's `rtp_params()`, maps str0m's `Codec` enum
+/// to our `StreamCodec`. Returns the first video codec found.
+pub fn video_codec_from_sdp_offer(
+    sdp_offer: &str0m::change::SdpOffer,
+) -> Option<crate::StreamCodec> {
+    use str0m::format::Codec;
+
+    // SdpOffer derefs to Sdp, which has pub media_lines.
+    // MediaLine is pub(crate) in str0m, but we can call its pub methods.
+    for line in sdp_offer.media_lines.iter() {
+        for p in line.rtp_params() {
+            if p.spec().codec.is_video() {
+                return match p.spec().codec {
+                    Codec::H264 => Some(crate::StreamCodec::H264),
+                    Codec::H265 => Some(crate::StreamCodec::H265),
+                    Codec::Av1 => Some(crate::StreamCodec::AV1),
+                    _ => None,
+                };
+            }
+        }
+    }
+    None
+}