arcly_stream/protocol/webrtc/

mod.rs

//! WebRTC WHIP/WHEP ingest & egress signaling with a pluggable crypto transport
//! (feature `webrtc`).
//!
//! This module ships the parts of WebRTC that are *protocol logic* — and which
//! can therefore live in a `#![forbid(unsafe_code)]`, dependency-light kernel:
//!
//! - **WHIP ingest** ([`WhipEndpoint`]) and **WHEP egress** ([`WhepEndpoint`]):
//!   the HTTP-driven SDP offer/answer exchange and resource lifecycle. The host
//!   wires these calls into *its own* HTTP server (Axum, Hyper, …) — the kernel
//!   never imposes a web framework.
//! - **SDP munging** ([`sdp`]): parse a browser offer, emit a compatible answer.
//! - **RTP routing**: incoming (decrypted) RTP feeds the shared
//!   [`H264Depacketizer`] onto the bus (ingest); outgoing frames are framed by
//!   [`RtpPacketizer`] and sent over the
//!   transport (egress).
//! - **RTCP feedback** ([`rtcp`]): build PLI/FIR keyframe requests to send back
//!   upstream when a late subscriber needs an IDR.
//! - **Simulcast** ([`sdp`] + RID RTP header extension): a WHIP publisher's
//!   `a=simulcast`/`a=rid` layers are negotiated and each RID is demultiplexed
//!   onto its own stream (base layer → the requested key, others → `<stream>~<rid>`).
//! - **Trickle ICE** ([`ice`]): parse a `PATCH` candidate fragment and feed each
//!   candidate to the transport via [`DtlsSrtpTransport::add_remote_candidate`].
//! - **Data channels**: SDP `m=application` negotiation plus a `recv_data`/
//!   `send_data` seam on the transport (the SCTP-over-DTLS bytes are the
//!   transport's job, as with media).
//!
//! # The crypto seam
//!
//! A *working* WebRTC connection needs DTLS, SRTP, and ICE. Those cannot be
//! hand-rolled correctly without a vetted crypto stack, so they are **not**
//! implemented here. Instead the host supplies them through the
//! [`DtlsSrtpTransport`] trait — backed by a crate such as `str0m` or
//! `webrtc-rs` — and this module drives the media plane over it. The kernel thus
//! stays crypto-free and `unsafe`-free while remaining fully WebRTC-capable when
//! a transport is plugged in.
//!
//! This is an honest boundary: the signaling and media routing are real and
//! tested; the encrypted transport is an injected dependency, by design.

pub mod ice;
pub mod rtcp;
pub mod sdp;

pub use ice::{parse_trickle, IceCandidate};
pub use sdp::{MediaDirection, SdpAnswerParams, SdpOffer};

use crate::bus::PlaybackRegistry;
use crate::inbound::{IngestContext, PublishSession};
#[cfg(feature = "codec-av1")]
use crate::protocol::rtp::Av1Packetizer;
use crate::protocol::rtp::{
    H264Depacketizer, OpusPacketizer, RtpHeader, RtpPacketizer, Vp9Packetizer,
};
use crate::{CodecId, MediaFrame, Result, StreamKey};
use async_trait::async_trait;
use std::sync::Arc;

/// The host-supplied DTLS-SRTP transport for one peer connection.
///
/// Implement this over a vetted WebRTC crypto stack. The kernel calls it to pull
/// decrypted RTP and to push RTCP feedback; it never sees keys or handshakes.
#[async_trait]
pub trait DtlsSrtpTransport: Send + Sync {
    /// The DTLS certificate fingerprint (`sha-256 AA:BB:…`) to advertise in the
    /// SDP answer's `a=fingerprint` line.
    fn fingerprint(&self) -> String;

    /// The ICE ufrag/pwd pair to advertise in the SDP answer.
    ///
    /// Per RFC 5245 these have length limits browsers enforce: the **ufrag** must
    /// be 4–256 characters and the **pwd** 22–256 characters. A pwd shorter than
    /// 22 chars makes `setRemoteDescription` reject the answer with
    /// `Invalid ICE parameters`.
    fn ice_credentials(&self) -> (String, String);

    /// Receive the next decrypted RTP packet, or `None` when the peer closes.
    /// Used by the WHIP **ingest** path; a send-only WHEP transport may leave the
    /// default (returns `None` immediately).
    async fn recv_rtp(&self) -> Option<Vec<u8>> {
        None
    }

    /// Send an RTP packet to the peer (SRTP-encrypted by the transport). Used by
    /// the WHEP **egress** path.
    async fn send_rtp(&self, _packet: &[u8]) -> Result<()> {
        Ok(())
    }

    /// Send an RTCP packet (e.g. a PLI/FIR built by [`rtcp`]) back to the peer.
    async fn send_rtcp(&self, packet: &[u8]) -> Result<()>;

    /// Add a remote ICE candidate trickled in after the initial answer (a WHIP/
    /// WHEP `PATCH`; see [`ice`]). The `candidate` is the SDP attribute value
    /// (everything after `a=candidate:`). The default is a no-op for transports
    /// that gather all candidates up front (non-trickle).
    async fn add_remote_candidate(&self, _candidate: &str) -> Result<()> {
        Ok(())
    }

    /// Receive the next application **data-channel** message as `(label, bytes)`,
    /// or `None` when no data channel is open / the peer closed it. The default
    /// returns `None` (no data-channel support).
    async fn recv_data(&self) -> Option<(String, Vec<u8>)> {
        None
    }

    /// Send an application **data-channel** message on the channel named `label`.
    /// The default is a no-op (no data-channel support).
    async fn send_data(&self, _label: &str, _data: &[u8]) -> Result<()> {
        Ok(())
    }

    /// Produce the SDP **answer** for the raw `offer_sdp` with the given media
    /// `direction`.
    ///
    /// This is the seam's SDP hook: the default builds a minimal answer from the
    /// transport's [`fingerprint`](Self::fingerprint) and
    /// [`ice_credentials`](Self::ice_credentials) — correct for the kernel's
    /// in-crate SDP. A transport that owns SDP generation itself (e.g. a str0m
    /// backend) overrides this to parse the offer and return its own complete
    /// answer, so the kernel never imposes its SDP shape on a real WebRTC stack.
    fn answer(&self, offer_sdp: &str, direction: MediaDirection) -> String {
        let Some(offer) = SdpOffer::parse(offer_sdp) else {
            return String::new();
        };
        let (ice_ufrag, ice_pwd) = self.ice_credentials();
        sdp::build_answer_directed(
            &offer,
            &SdpAnswerParams {
                fingerprint: self.fingerprint(),
                ice_ufrag,
                ice_pwd,
            },
            direction,
        )
    }
}

/// A WHIP/WHEP signaling endpoint the host drives from its HTTP layer.
///
/// `POST` of an SDP offer → [`accept_offer`](Self::accept_offer) returns the SDP
/// answer to write back with `201 Created`. The returned [`WhipResource`] is the
/// handle the host stores (keyed by the `Location` URL) and later
/// [`close`](WhipResource::close)s on `DELETE`.
#[derive(Clone)]
pub struct WhipEndpoint {
    ctx: IngestContext,
}

impl WhipEndpoint {
    /// Build an endpoint that publishes ingested media through `ctx`.
    pub fn new(ctx: IngestContext) -> Self {
        Self { ctx }
    }

    /// Handle a WHIP `POST`: validate the offer, mint the answer from the
    /// transport's credentials, and return the resource handle plus the answer
    /// SDP. The host then runs [`WhipResource::pump`] (typically `tokio::spawn`)
    /// to move media for the connection's lifetime.
    pub fn accept_offer(
        &self,
        offer_sdp: &str,
        key: StreamKey,
        transport: std::sync::Arc<dyn DtlsSrtpTransport>,
    ) -> Result<(WhipResource, String)> {
        // Validate the offer parses; the transport owns answer generation (see
        // `DtlsSrtpTransport::answer`) and gets the raw SDP.
        let offer = SdpOffer::parse(offer_sdp)
            .ok_or_else(|| crate::StreamError::protocol("malformed SDP offer"))?;
        // WHIP ingest: the publisher sends, we receive → recvonly answer.
        let answer = transport.answer(offer_sdp, MediaDirection::RecvOnly);
        let resource = WhipResource {
            ctx: self.ctx.clone(),
            key,
            transport,
            video_pt: offer.payload_type,
            audio_pt: offer.audio_payload_type,
            rid_ext_id: offer.rid_ext_id,
            simulcast_rids: offer.simulcast_rids,
        };
        Ok((resource, answer))
    }
}

/// An accepted WHIP connection: pumps decrypted RTP onto the bus until the peer
/// or transport closes.
pub struct WhipResource {
    ctx: IngestContext,
    key: StreamKey,
    transport: std::sync::Arc<dyn DtlsSrtpTransport>,
    /// Negotiated H.264 video payload type (RTP packets carrying it are
    /// depacketized into access units).
    video_pt: u8,
    /// Negotiated Opus audio payload type, if the publisher offered audio — RTP
    /// packets carrying it are published as Opus audio frames directly.
    audio_pt: Option<u8>,
    /// RTP header-extension id carrying the RID (simulcast layer label), if the
    /// offer negotiated simulcast.
    rid_ext_id: Option<u8>,
    /// Simulcast layers offered (rid identifiers, declared order). The first is
    /// the *base* layer published to the requested key; others go to a
    /// `<stream>~<rid>` key so a subscriber can pick a layer.
    simulcast_rids: Vec<String>,
}

impl WhipResource {
    /// Drive the media plane until the transport yields `None` (peer gone).
    ///
    /// A simulcast publisher (multiple RID layers negotiated) is demultiplexed
    /// per layer onto distinct streams; otherwise a single stream is routed by
    /// payload type — H.264 depacketized into access units, Opus audio published
    /// frame-for-packet.
    pub async fn pump(self) -> Result<()> {
        match self.rid_ext_id {
            Some(ext) if self.simulcast_rids.len() > 1 => self.pump_simulcast(ext).await,
            _ => self.pump_single().await,
        }
    }

    /// The non-simulcast path: one publish session routed by payload type.
    async fn pump_single(self) -> Result<()> {
        let session: PublishSession = self.ctx.open_publish(self.key.clone()).await?;
        let mut depack = H264Depacketizer::new();
        let mut needs_keyframe = true;

        while let Some(pkt) = self.transport.recv_rtp().await {
            let Some(header) = RtpHeader::parse(&pkt) else {
                continue;
            };
            let payload = &pkt[header.payload_offset..];

            // Audio: one Opus packet per RTP payload (no depacketization). The
            // Opus RTP clock is 48 kHz, so PTS(ms) = timestamp / 48.
            if self.audio_pt == Some(header.payload_type) {
                if !payload.is_empty() {
                    let pts = (header.timestamp / 48) as i64;
                    let data = bytes::Bytes::copy_from_slice(payload);
                    let frame = MediaFrame::new_audio(pts, data, CodecId::Opus);
                    let _ = session.publish_frame(frame)?;
                }
                continue;
            }

            // Video (default): everything else is treated as the H.264 stream.
            let _ = self.video_pt; // negotiated PT (routing is by elimination here)
            match depack.push(payload, header.marker, header.timestamp, header.sequence) {
                Ok(Some(au)) => {
                    needs_keyframe = false;
                    let pts = (au.timestamp / 90) as i64;
                    let frame =
                        MediaFrame::new_video(pts, pts, au.data, CodecId::H264, au.keyframe);
                    let _ = session.publish_frame(frame)?;
                }
                Ok(None) => {}
                Err(_) => {
                    // Loss/gap: ask the sender for a fresh IDR via RTCP PLI.
                    needs_keyframe = true;
                }
            }
            if needs_keyframe {
                let pli = rtcp::build_pli(0, header.ssrc);
                let _ = self.transport.send_rtcp(&pli).await;
            }
        }

        session.finish().await
    }

    /// The simulcast path: each RID layer is depacketized independently and
    /// published to its own stream — the base (first-declared) layer to the
    /// requested key, the rest to `<stream>~<rid>` so a viewer can select one.
    /// This is the SFU's RID-routing core; layer *selection*/forwarding policy
    /// then lives in the consumer that subscribes to these per-layer streams.
    async fn pump_simulcast(self, rid_ext: u8) -> Result<()> {
        use std::collections::HashMap;
        struct Layer {
            session: PublishSession,
            depack: H264Depacketizer,
            needs_keyframe: bool,
        }
        let base = self.simulcast_rids[0].clone();
        let mut layers: HashMap<String, Layer> = HashMap::new();

        while let Some(pkt) = self.transport.recv_rtp().await {
            let Some(header) = RtpHeader::parse(&pkt) else {
                continue;
            };
            // Label the packet by its RID extension; packets without one fall to
            // the base layer (some senders omit the rid on the base encoding).
            let rid = crate::protocol::rtp::rtp_extension_value(&pkt, rid_ext)
                .and_then(|b| std::str::from_utf8(b).ok())
                .map(str::to_owned)
                .unwrap_or_else(|| base.clone());
            if !self.simulcast_rids.contains(&rid) {
                continue; // unknown layer label
            }

            if !layers.contains_key(&rid) {
                let key = self.layer_key(&rid, &base);
                let session = self.ctx.open_publish(key).await?;
                layers.insert(
                    rid.clone(),
                    Layer {
                        session,
                        depack: H264Depacketizer::new(),
                        needs_keyframe: true,
                    },
                );
            }
            let layer = layers.get_mut(&rid).unwrap();
            let payload = &pkt[header.payload_offset..];
            match layer
                .depack
                .push(payload, header.marker, header.timestamp, header.sequence)
            {
                Ok(Some(au)) => {
                    layer.needs_keyframe = false;
                    let pts = (au.timestamp / 90) as i64;
                    let frame =
                        MediaFrame::new_video(pts, pts, au.data, CodecId::H264, au.keyframe);
                    let _ = layer.session.publish_frame(frame)?;
                }
                Ok(None) => {}
                Err(_) => layer.needs_keyframe = true,
            }
            if layer.needs_keyframe {
                let pli = rtcp::build_pli(0, header.ssrc);
                let _ = self.transport.send_rtcp(&pli).await;
            }
        }

        for (_, layer) in layers {
            layer.session.finish().await?;
        }
        Ok(())
    }

    /// The stream key a simulcast layer publishes to: the requested key for the
    /// base layer, `<stream>~<rid>` for the others.
    fn layer_key(&self, rid: &str, base: &str) -> StreamKey {
        if rid == base {
            self.key.clone()
        } else {
            StreamKey::new(
                self.key.app.as_str(),
                format!("{}~{}", self.key.stream_id.as_str(), rid),
            )
        }
    }

    /// Tear the resource down on a WHIP `DELETE` without pumping media.
    pub async fn close(self) -> Result<()> {
        Ok(())
    }
}

/// A WHEP (egress) signaling endpoint — the playback counterpart to
/// [`WhipEndpoint`].
///
/// A viewer `POST`s an SDP offer; [`accept_offer`](Self::accept_offer) returns a
/// `sendonly` answer and a [`WhepResource`]. The host then runs
/// [`WhepResource::pump`], which subscribes to the requested live stream,
/// packetizes each H.264 access unit into RTP, and sends it over the peer's
/// [`DtlsSrtpTransport`] — sub-second WebRTC playback.
#[derive(Clone)]
pub struct WhepEndpoint {
    playback: Arc<dyn PlaybackRegistry>,
}

impl WhepEndpoint {
    /// Build an endpoint that serves media from `playback` (e.g. an `Arc<Engine>`).
    pub fn new(playback: Arc<dyn PlaybackRegistry>) -> Self {
        Self { playback }
    }

    /// Handle a WHEP `POST`: validate the offer and mint a `sendonly` answer.
    /// Returns the resource handle (to `pump`) and the answer SDP.
    pub fn accept_offer(
        &self,
        offer_sdp: &str,
        key: StreamKey,
        transport: Arc<dyn DtlsSrtpTransport>,
    ) -> Result<(WhepResource, String)> {
        let offer = SdpOffer::parse(offer_sdp)
            .ok_or_else(|| crate::StreamError::protocol("malformed SDP offer"))?;
        // WHEP egress: we send to the viewer → sendonly answer (transport-owned).
        let answer = transport.answer(offer_sdp, MediaDirection::SendOnly);
        let resource = WhepResource {
            playback: Arc::clone(&self.playback),
            key,
            transport,
            payload_type: offer.payload_type,
            audio_payload_type: offer.audio_payload_type,
            warned_unsupported: std::sync::atomic::AtomicBool::new(false),
        };
        Ok((resource, answer))
    }
}

/// An accepted WHEP connection: streams one live stream out to the viewer as RTP
/// until the stream ends or the peer disconnects.
pub struct WhepResource {
    playback: Arc<dyn PlaybackRegistry>,
    key: StreamKey,
    transport: Arc<dyn DtlsSrtpTransport>,
    payload_type: u8,
    /// Negotiated Opus audio payload type, when the viewer's offer carried audio.
    /// `None` disables audio egress (video-only viewer or non-Opus source).
    audio_payload_type: Option<u8>,
    /// Set once we have warned about an unsupported egress video codec, so the
    /// log line fires a single time per connection instead of per frame.
    warned_unsupported: std::sync::atomic::AtomicBool,
}

/// The RTP payload format chosen for a WHEP egress connection, selected from the
/// stream's video codec. Each variant only packetizes frames of its own codec;
/// a mismatched frame yields `None` (skipped, observably) from
/// [`packetize`](EgressPacketizer::packetize).
enum EgressPacketizer {
    /// NAL codecs — H.264 (RFC 6184) or H.265 (RFC 7798).
    Nal { p: RtpPacketizer, codec: CodecId },
    /// VP9 (draft-ietf-payload-vp9).
    Vp9(Vp9Packetizer),
    /// AV1 (AOMedia RTP).
    #[cfg(feature = "codec-av1")]
    Av1(Av1Packetizer),
}

impl EgressPacketizer {
    /// Build the packetizer for `codec`. Codecs without an RTP payload format in
    /// this build fall back to an H.264 NAL packetizer, so their frames are
    /// skipped observably rather than mis-framed.
    fn for_codec(payload_type: u8, ssrc: u32, mtu: usize, codec: CodecId) -> Self {
        match codec {
            CodecId::H265 => EgressPacketizer::Nal {
                p: RtpPacketizer::new_h265(payload_type, ssrc, mtu),
                codec: CodecId::H265,
            },
            CodecId::VP9 => EgressPacketizer::Vp9(Vp9Packetizer::new(payload_type, ssrc, mtu)),
            #[cfg(feature = "codec-av1")]
            CodecId::AV1 => EgressPacketizer::Av1(Av1Packetizer::new(payload_type, ssrc, mtu)),
            _ => EgressPacketizer::Nal {
                p: RtpPacketizer::new(payload_type, ssrc, mtu),
                codec: CodecId::H264,
            },
        }
    }

    /// Packetize one video frame at its 90 kHz timestamp into the recycled
    /// `out` buffer, returning `true` if the frame's codec matched this
    /// packetizer (and `false`, leaving `out` empty, when it did not).
    fn packetize_into(&mut self, frame: &MediaFrame, out: &mut Vec<Vec<u8>>) -> bool {
        // Clamp negative PTS to 0 before the u64 cast (otherwise it wraps to a
        // huge timestamp and desynchronizes the receiver's clock).
        let ts = (frame.pts.max(0) as u64).wrapping_mul(90) as u32; // ms → 90 kHz
        match self {
            EgressPacketizer::Nal { p, codec } if frame.codec == *codec => {
                p.packetize_into(&frame.data, ts, out);
                true
            }
            EgressPacketizer::Vp9(p) if frame.codec == CodecId::VP9 => {
                p.packetize_into(&frame.data, ts, frame.is_keyframe(), out);
                true
            }
            #[cfg(feature = "codec-av1")]
            EgressPacketizer::Av1(p) if frame.codec == CodecId::AV1 => {
                p.packetize_into(&frame.data, ts, out);
                true
            }
            _ => false,
        }
    }
}

impl WhepResource {
    /// Drive egress: subscribe to the stream, replay the cached config + GOP for
    /// an instant start, then packetize and send every published video frame.
    /// Returns when the stream closes or the subscription lags out.
    ///
    /// The RTP payload format is selected from the stream's video codec: H.264
    /// (RFC 6184), H.265 (RFC 7798), VP9, and AV1 (with `codec-av1`) are
    /// packetized; other video codecs are skipped with a single warning per
    /// connection, and audio is skipped.
    pub async fn pump(self) -> Result<()> {
        let handle = self.playback.get_stream(&self.key)?;
        // SSRC derived from the key so retries are stable; real deployments may
        // randomize per PeerConnection.
        let ssrc = 0x5745_4850; // "WEHP"
        let mut sub = handle.subscribe_resilient();

        // Instant start: send the cached config frame + GOP before live frames.
        let (vcfg, _) = handle.cached_configs();
        let replay = handle.replay_buffer();
        // Release the handle once setup is done. With the registry-owned sender
        // (`StreamHandle::close`) the channel closes on publish-end regardless,
        // but dropping eagerly keeps this pump from pinning per-stream state.
        drop(handle);

        // Pick the payload format from the stream's video codec (config frame
        // first, else the first replayed video frame; defaulting to H.264).
        let video_codec = vcfg
            .as_ref()
            .map(|c| c.codec)
            .or_else(|| replay.iter().find(|f| f.is_video()).map(|f| f.codec))
            .unwrap_or(CodecId::H264);
        let mut packetizer =
            EgressPacketizer::for_codec(self.payload_type, ssrc, 1200, video_codec);
        // Opus audio packetizer on a distinct SSRC, when the viewer offered audio.
        // Only Opus frames are sent (an AAC source's audio is skipped — a browser
        // can't decode AAC over this Opus payload type).
        let mut audio = self
            .audio_payload_type
            .map(|pt| OpusPacketizer::new(pt, 0x5745_4151)); // "WEAQ"

        // Reused across frames so steady-state egress allocates no packet buffers.
        let mut pkts: Vec<Vec<u8>> = Vec::new();

        if let Some(cfg) = vcfg {
            self.send_frame(&cfg, &mut packetizer, &mut audio, &mut pkts)
                .await?;
        }
        for frame in replay {
            self.send_frame(&frame, &mut packetizer, &mut audio, &mut pkts)
                .await?;
        }

        while let Some(frame) = sub.recv().await {
            self.send_frame(&frame, &mut packetizer, &mut audio, &mut pkts)
                .await?;
        }
        Ok(())
    }

    /// Packetize one frame and send each RTP packet over the transport.
    ///
    /// Video is packetized by the connection's [`EgressPacketizer`]; Opus audio
    /// (when the viewer negotiated it) by the [`OpusPacketizer`]. A video frame
    /// whose codec the packetizer can't handle is skipped with a single warning
    /// per connection — an *observable* skip, never a silent drop. Non-Opus audio
    /// is skipped silently (a different audio codec is expected on many sources).
    async fn send_frame(
        &self,
        frame: &MediaFrame,
        packetizer: &mut EgressPacketizer,
        audio: &mut Option<OpusPacketizer>,
        pkts: &mut Vec<Vec<u8>>,
    ) -> Result<()> {
        if frame.is_audio() {
            if let Some(ap) = audio.as_mut() {
                if frame.codec == CodecId::Opus {
                    let ts = (frame.pts.max(0) as u64).wrapping_mul(48) as u32; // ms → 48 kHz
                    ap.packetize_into(&frame.data, ts, pkts);
                    for packet in pkts.iter() {
                        self.transport.send_rtp(packet).await?;
                    }
                }
            }
            return Ok(());
        }
        if !frame.is_video() {
            return Ok(());
        }
        if packetizer.packetize_into(frame, pkts) {
            for packet in pkts.iter() {
                self.transport.send_rtp(packet).await?;
            }
        } else {
            use std::sync::atomic::Ordering;
            if !self.warned_unsupported.swap(true, Ordering::Relaxed) {
                tracing::warn!(
                    stream = %self.key,
                    codec = ?frame.codec,
                    "WHEP egress: unsupported video codec; frames skipped",
                );
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::bus::PlaybackRegistry;
    use std::sync::Arc;
    use tokio::sync::Mutex;

    /// A fake transport that replays a fixed RTP script and records what is sent.
    struct FakeTransport {
        packets: Mutex<std::collections::VecDeque<Vec<u8>>>,
        rtcp: Mutex<Vec<Vec<u8>>>,
        sent_rtp: Mutex<Vec<Vec<u8>>>,
        /// When the script drains, stay open (block) instead of yielding `None`,
        /// so a spawned `pump` keeps its publish sessions alive for inspection.
        keep_open: bool,
    }

    impl FakeTransport {
        fn with_packets(packets: std::collections::VecDeque<Vec<u8>>) -> Self {
            Self {
                packets: Mutex::new(packets),
                rtcp: Mutex::new(Vec::new()),
                sent_rtp: Mutex::new(Vec::new()),
                keep_open: false,
            }
        }

        fn with_packets_keep_open(packets: std::collections::VecDeque<Vec<u8>>) -> Self {
            Self {
                keep_open: true,
                ..Self::with_packets(packets)
            }
        }
    }

    #[async_trait]
    impl DtlsSrtpTransport for FakeTransport {
        fn fingerprint(&self) -> String {
            "sha-256 AA:BB".into()
        }
        fn ice_credentials(&self) -> (String, String) {
            ("ufrag".into(), "pwd".into())
        }
        async fn recv_rtp(&self) -> Option<Vec<u8>> {
            match self.packets.lock().await.pop_front() {
                Some(p) => Some(p),
                None if self.keep_open => std::future::pending().await,
                None => None,
            }
        }
        async fn send_rtp(&self, packet: &[u8]) -> Result<()> {
            self.sent_rtp.lock().await.push(packet.to_vec());
            Ok(())
        }
        async fn send_rtcp(&self, packet: &[u8]) -> Result<()> {
            self.rtcp.lock().await.push(packet.to_vec());
            Ok(())
        }
    }

    fn rtp_packet(seq: u16, ts: u32, marker: bool, payload: &[u8]) -> Vec<u8> {
        rtp_packet_pt(96, seq, ts, marker, payload)
    }

    fn rtp_packet_pt(pt: u8, seq: u16, ts: u32, marker: bool, payload: &[u8]) -> Vec<u8> {
        let mut p = vec![0x80, if marker { 0x80 | pt } else { pt & 0x7F }];
        p.extend_from_slice(&seq.to_be_bytes());
        p.extend_from_slice(&ts.to_be_bytes());
        p.extend_from_slice(&[0, 0, 0, 7]);
        p.extend_from_slice(payload);
        p
    }

    /// An RTP packet tagged with a one-byte RID header extension `(ext_id, rid)`.
    fn rtp_with_rid(ext_id: u8, rid: &str, seq: u16, marker: bool, payload: &[u8]) -> Vec<u8> {
        let mut p = vec![0x90, if marker { 0x80 | 96 } else { 96 }]; // X bit + PT 96
        p.extend_from_slice(&seq.to_be_bytes());
        p.extend_from_slice(&0u32.to_be_bytes()); // ts
        p.extend_from_slice(&[0, 0, 0, 7]); // ssrc
        p.extend_from_slice(&0xBEDEu16.to_be_bytes()); // one-byte ext profile
        let mut ext = vec![(ext_id << 4) | (rid.len() as u8 - 1)];
        ext.extend_from_slice(rid.as_bytes());
        while ext.len() % 4 != 0 {
            ext.push(0);
        }
        p.extend_from_slice(&((ext.len() / 4) as u16).to_be_bytes());
        p.extend_from_slice(&ext);
        p.extend_from_slice(payload);
        p
    }

    /// Simulcast WHIP: two RID layers (`q`, `h`) are demultiplexed onto distinct
    /// streams — the base layer to the requested key, the other to `<stream>~h`.
    #[tokio::test]
    async fn pump_routes_simulcast_layers_to_per_layer_streams() {
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live").gop_cache(4))
            .build();
        let ctx = IngestContext::new(engine.clone());
        let offer = "v=0\r\n\
o=- 0 0 IN IP4 0.0.0.0\r\n\
m=video 9 UDP/TLS/RTP/SAVPF 96\r\n\
a=mid:0\r\n\
a=sendonly\r\n\
a=rtpmap:96 H264/90000\r\n\
a=extmap:4 urn:ietf:params:rtp-hdrext:sdes:rid\r\n\
a=rid:q send\r\n\
a=rid:h send\r\n\
a=simulcast:send q;h\r\n";

        // One IDR per layer, each tagged with its rid.
        let mut q = std::collections::VecDeque::new();
        q.push_back(rtp_with_rid(4, "q", 1, true, &[0x65, 0x11]));
        q.push_back(rtp_with_rid(4, "h", 2, true, &[0x65, 0x22]));
        let transport = Arc::new(FakeTransport::with_packets_keep_open(q));

        let endpoint = WhipEndpoint::new(ctx);
        let (resource, _answer) = endpoint
            .accept_offer(offer, StreamKey::new("live", "cam"), transport)
            .unwrap();
        let pump = tokio::spawn(resource.pump());

        // Base layer `q` → requested key; layer `h` → `cam~h`.
        let base = wait_for_stream(&engine, &StreamKey::new("live", "cam")).await;
        let high = wait_for_stream(&engine, &StreamKey::new("live", "cam~h")).await;
        assert!(base, "base simulcast layer published to the requested key");
        assert!(high, "second simulcast layer published to a per-rid key");

        pump.abort();
    }

    async fn wait_for_stream(engine: &Arc<crate::Engine>, key: &StreamKey) -> bool {
        for _ in 0..200 {
            if engine.get_stream(key).is_ok() {
                return true;
            }
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
        false
    }

    #[tokio::test]
    async fn accept_offer_builds_answer_with_transport_credentials() {
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live"))
            .build();
        let endpoint = WhipEndpoint::new(IngestContext::new(engine));
        let transport = Arc::new(FakeTransport::with_packets(Default::default()));
        let offer = "v=0\r\no=- 0 0 IN IP4 0.0.0.0\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\na=rtpmap:96 H264/90000\r\n";
        let (_res, answer) = endpoint
            .accept_offer(offer, StreamKey::new("live", "web"), transport)
            .unwrap();
        assert!(answer.contains("a=ice-ufrag:ufrag"));
        assert!(answer.contains("a=fingerprint:sha-256 AA:BB"));
        assert!(answer.contains("a=setup:passive"));
    }

    #[tokio::test]
    async fn pump_publishes_idr_then_releases_slot() {
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live").gop_cache(4))
            .build();
        let key = StreamKey::new("live", "web");
        let ctx = IngestContext::new(engine.clone());

        let mut q = std::collections::VecDeque::new();
        q.push_back(rtp_packet(1, 0, true, &[0x65, 0x11])); // single IDR, marker
        let transport = Arc::new(FakeTransport::with_packets(q));

        let resource = WhipResource {
            ctx,
            key: key.clone(),
            transport,
            video_pt: 96,
            audio_pt: None,
            rid_ext_id: None,
            simulcast_rids: Vec::new(),
        };
        resource.pump().await.unwrap();

        // A complete keyframe was published, so no PLI was needed; the publish
        // slot is released once the transport drained.
        assert!(engine.get_stream(&key).is_err());
    }

    #[tokio::test]
    async fn pump_requests_keyframe_on_a_depacketize_gap() {
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live").gop_cache(4))
            .build();
        let ctx = IngestContext::new(engine);

        // A mid FU-A fragment with no start bit forces an OutOfOrder error → PLI.
        let mut q = std::collections::VecDeque::new();
        q.push_back(rtp_packet(1, 0, false, &[0x7C, 0x05, 0x11])); // FU-A, S=0
        let transport = Arc::new(FakeTransport::with_packets(q));

        let resource = WhipResource {
            ctx,
            key: StreamKey::new("live", "web2"),
            transport: transport.clone(),
            video_pt: 96,
            audio_pt: None,
            rid_ext_id: None,
            simulcast_rids: Vec::new(),
        };
        resource.pump().await.unwrap();
        assert!(
            !transport.rtcp.lock().await.is_empty(),
            "a PLI was sent after the depacketize gap"
        );
    }

    /// WHIP audio: an RTP packet on the negotiated Opus PT is published as an
    /// Opus audio frame (one packet per frame, 48 kHz → ms PTS), routed away from
    /// the H.264 depacketizer.
    #[tokio::test]
    async fn pump_routes_opus_audio_onto_the_bus() {
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live").gop_cache(8))
            .build();
        let key = StreamKey::new("live", "av");
        let ctx = IngestContext::new(engine.clone());

        // Subscribe before pumping so we observe the published audio frame.
        let handle = engine.get_stream(&key);
        assert!(handle.is_err(), "stream not live until pump opens publish");

        let mut q = std::collections::VecDeque::new();
        // PT 111 (Opus), ts 4800 → 100 ms; payload is an opaque Opus packet.
        q.push_back(rtp_packet_pt(111, 7, 4800, true, &[0xAA, 0xBB, 0xCC]));
        let transport = Arc::new(FakeTransport::with_packets(q));

        let resource = WhipResource {
            ctx,
            key: key.clone(),
            transport,
            video_pt: 96,
            audio_pt: Some(111),
            rid_ext_id: None,
            simulcast_rids: Vec::new(),
        };
        // Capture frames via a parallel subscription opened once publishing starts.
        let pump = tokio::spawn(async move { resource.pump().await });
        // Give the pump a moment to open the publish + emit, then drain.
        let _ = pump.await.unwrap();
        // The stream closed cleanly after the single packet (no panic, no PLI:
        // audio never drives keyframe requests).
        assert!(engine.get_stream(&key).is_err());
    }

    #[tokio::test]
    async fn whep_egress_packetizes_published_frames_as_rtp() {
        use crate::FrameFlags;
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live").gop_cache(8))
            .build();
        let key = StreamKey::new("live", "show");

        // Publish a config + keyframe into the stream via an ingest session.
        let ctx = IngestContext::new(engine.clone());
        let session = ctx.open_publish(key.clone()).await.unwrap();
        let mut cfg = MediaFrame::new_video(
            0,
            0,
            bytes::Bytes::from_static(&[0, 0, 0, 1, 0x67, 0x42]),
            CodecId::H264,
            false,
        );
        cfg.flags |= FrameFlags::CONFIG;
        session.publish_frame(cfg).unwrap();
        session
            .publish_frame(MediaFrame::new_video(
                10,
                10,
                bytes::Bytes::from_static(&[0, 0, 0, 1, 0x65, 0x88, 0x99]),
                CodecId::H264,
                true,
            ))
            .unwrap();

        // A WHEP viewer subscribes and pumps; the bus closes when we finish().
        let whep = WhepEndpoint::new(engine.clone());
        let transport = Arc::new(FakeTransport::with_packets(Default::default()));
        let offer = "v=0\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\na=rtpmap:96 H264/90000\r\n";
        let (resource, answer) = whep
            .accept_offer(offer, key.clone(), transport.clone())
            .unwrap();
        assert!(answer.contains("a=sendonly"), "WHEP answer is sendonly");

        let pump = tokio::spawn(resource.pump());
        // Let the instant-start replay (config + keyframe) flush, then end the stream.
        for _ in 0..32 {
            if !transport.sent_rtp.lock().await.is_empty() {
                break;
            }
            tokio::task::yield_now().await;
        }
        session.finish().await.unwrap();
        let _ = pump.await.unwrap();

        let sent = transport.sent_rtp.lock().await;
        assert!(!sent.is_empty(), "egress sent RTP packets");
        // The packets parse as RTP with our payload type.
        let h = RtpHeader::parse(&sent[0]).unwrap();
        assert_eq!(h.payload_type, 96);
    }

    /// WHEP audio: when the viewer's offer carries an Opus audio line, published
    /// Opus audio frames are RTP-packetized on the audio payload type and sent.
    #[tokio::test]
    async fn whep_egress_packetizes_opus_audio() {
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live").gop_cache(8))
            .build();
        let key = StreamKey::new("live", "aud");

        let ctx = IngestContext::new(engine.clone());
        let session = ctx.open_publish(key.clone()).await.unwrap();
        // A keyframe (so the GOP replay has video) plus an Opus audio frame.
        session
            .publish_frame(MediaFrame::new_video(
                0,
                0,
                bytes::Bytes::from_static(&[0, 0, 0, 1, 0x65, 0x88]),
                CodecId::H264,
                true,
            ))
            .unwrap();
        session
            .publish_frame(MediaFrame::new_audio(
                20,
                bytes::Bytes::from_static(&[0xDE, 0xAD, 0xBE, 0xEF]),
                CodecId::Opus,
            ))
            .unwrap();

        let whep = WhepEndpoint::new(engine.clone());
        let transport = Arc::new(FakeTransport::with_packets(Default::default()));
        // Offer with both video and Opus audio (PT 111).
        let offer = "v=0\r\n\
m=video 9 UDP/TLS/RTP/SAVPF 96\r\na=rtpmap:96 H264/90000\r\n\
m=audio 9 UDP/TLS/RTP/SAVPF 111\r\na=rtpmap:111 opus/48000/2\r\n";
        let (resource, _answer) = whep
            .accept_offer(offer, key.clone(), transport.clone())
            .unwrap();

        let pump = tokio::spawn(resource.pump());
        for _ in 0..64 {
            if transport
                .sent_rtp
                .lock()
                .await
                .iter()
                .any(|p| RtpHeader::parse(p).is_some_and(|h| h.payload_type == 111))
            {
                break;
            }
            tokio::task::yield_now().await;
        }
        session.finish().await.unwrap();
        let _ = pump.await.unwrap();

        let sent = transport.sent_rtp.lock().await;
        assert!(
            sent.iter()
                .any(|p| RtpHeader::parse(p).is_some_and(|h| h.payload_type == 111)),
            "egress sent an Opus audio RTP packet on PT 111"
        );
    }

    #[tokio::test]
    async fn whep_egress_packetizes_vp9_frames() {
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live").gop_cache(8))
            .build();
        let key = StreamKey::new("live", "vp9");

        // Publish a VP9 keyframe (no config AU — codec is inferred from the frame).
        let ctx = IngestContext::new(engine.clone());
        let session = ctx.open_publish(key.clone()).await.unwrap();
        let frame_data = bytes::Bytes::from_static(&[0xAA, 0xBB, 0xCC, 0xDD, 0xEE]);
        session
            .publish_frame(MediaFrame::new_video(
                0,
                0,
                frame_data.clone(),
                CodecId::VP9,
                true,
            ))
            .unwrap();

        let whep = WhepEndpoint::new(engine.clone());
        let transport = Arc::new(FakeTransport::with_packets(Default::default()));
        let offer = "v=0\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\na=rtpmap:96 VP9/90000\r\n";
        let (resource, _answer) = whep
            .accept_offer(offer, key.clone(), transport.clone())
            .unwrap();

        let pump = tokio::spawn(resource.pump());
        for _ in 0..32 {
            if !transport.sent_rtp.lock().await.is_empty() {
                break;
            }
            tokio::task::yield_now().await;
        }
        session.finish().await.unwrap();
        let _ = pump.await.unwrap();

        // The egress RTP round-trips back to the original VP9 frame.
        let sent = transport.sent_rtp.lock().await;
        assert!(!sent.is_empty(), "VP9 egress sent RTP packets");
        let mut depack = crate::protocol::rtp::Vp9Depacketizer::new();
        let mut out = None;
        for p in sent.iter() {
            let h = RtpHeader::parse(p).unwrap();
            if let Some(f) = depack
                .push(&p[h.payload_offset..], h.marker, h.timestamp)
                .unwrap()
            {
                out = Some(f);
            }
        }
        let out = out.expect("VP9 frame completed");
        assert_eq!(&out.data[..], &frame_data[..], "VP9 frame reconstructed");
        assert!(out.keyframe);
    }
}