arcly-stream 0.8.3

//! 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 room;
pub mod rtcp;
pub mod sdp;

pub use ice::{parse_trickle, IceCandidate};
pub use room::{DominantSpeaker, Room};
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;

/// A snapshot of one peer connection's quality, surfaced from the transport for
/// host metrics (Prometheus, dashboards) and Phase-1/2 debugging.
///
/// All fields are `Option` because a fresh connection may not have produced an
/// estimate or a feedback report yet.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct PeerStats {
    /// Egress bandwidth estimate in bits/sec (TWCC/REMB), as in
    /// [`estimated_bitrate`](DtlsSrtpTransport::estimated_bitrate).
    pub estimated_bitrate_bps: Option<u64>,
    /// Round-trip time in milliseconds, from the latest RTCP receiver report.
    pub rtt_ms: Option<f32>,
    /// Egress packet loss fraction over the last second (0.0–1.0).
    pub egress_loss: Option<f32>,
}

/// 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<()>;

    /// Receive the next (decrypted) RTCP compound packet from the peer, or `None`
    /// when the peer closes / the transport carries no RTCP.
    ///
    /// Used by the WHEP **egress** path to read viewer feedback — PLI/FIR (request
    /// a fresh keyframe), generic NACK (loss to retransmit), and Receiver Reports
    /// (loss/jitter for QoS and bandwidth estimation), decoded via
    /// [`rtcp::parse_compound`]. The default returns `None`, so a transport with no
    /// upstream RTCP keeps compiling.
    async fn recv_rtcp(&self) -> Option<Vec<u8>> {
        None
    }

    /// The transport's current egress **bandwidth estimate** in bits/sec, or
    /// `None` when the transport does not estimate bandwidth.
    ///
    /// A WebRTC stack (e.g. str0m) derives this from TWCC/REMB feedback far more
    /// accurately than the crypto-free kernel could from raw RTCP, so the kernel
    /// reads it through this seam instead of re-deriving it. It is the per-viewer
    /// input to adaptive layer selection (Phase 2). The default returns `None`.
    fn estimated_bitrate(&self) -> Option<u64> {
        None
    }

    /// A snapshot of this peer's connection quality for metrics/observability, or
    /// `None` when the transport does not collect stats. See [`PeerStats`].
    fn peer_stats(&self) -> Option<PeerStats> {
        None
    }

    /// 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 handle = session.handle().clone();
        let mut depack = H264Depacketizer::new();
        let mut needs_keyframe = true;
        // Most recent media SSRC, so a downstream-relayed keyframe request targets
        // the right stream in the PLI we send the publisher.
        let mut last_ssrc = 0u32;

        loop {
            let pkt = tokio::select! {
                pkt = self.transport.recv_rtp() => match pkt {
                    Some(p) => p,
                    None => break,
                },
                // A playback consumer (e.g. a WHEP viewer) asked for a keyframe:
                // relay it upstream as a PLI so the publisher emits a fresh IDR.
                _ = handle.keyframe_requested() => {
                    let pli = rtcp::build_pli(0, last_ssrc);
                    let _ = self.transport.send_rtcp(&pli).await;
                    continue;
                }
            };
            let Some(header) = RtpHeader::parse(&pkt) else {
                continue;
            };
            last_ssrc = header.ssrc;
            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 {
            self.key.layer(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>,
    /// Egress gate (per-app toggle + play token). `None` = open playback, the
    /// same permit-all default as RTSP `PLAY` and SRT `m=request`.
    gate: Option<crate::auth::EgressGate>,
}

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,
            gate: None,
        }
    }

    /// Gate playback (egress) requests through `gate` (per-app toggle + play
    /// token), mirroring [`RtspServer::with_gate`](crate::protocol::rtsp::RtspServer::with_gate)
    /// and the SRT egress gate. Consulted by [`accept_offer_gated`](Self::accept_offer_gated).
    pub fn with_gate(mut self, gate: crate::auth::EgressGate) -> Self {
        self.gate = Some(gate);
        self
    }

    /// Like [`accept_offer`](Self::accept_offer), but first consults the egress
    /// gate (if installed via [`with_gate`](Self::with_gate)) with `token` — the
    /// play token the viewer presented (e.g. `auth::token_from_query` over the
    /// WHEP POST URL). A denied request returns
    /// [`Unauthorized`](crate::StreamError::Unauthorized); with no gate, egress is
    /// open. Hosts that authorize at the HTTP layer can keep using `accept_offer`.
    pub async fn accept_offer_gated(
        &self,
        offer_sdp: &str,
        key: StreamKey,
        token: Option<String>,
        peer: Option<std::net::SocketAddr>,
        transport: Arc<dyn DtlsSrtpTransport>,
    ) -> Result<(WhepResource, String)> {
        if let Some(gate) = self.gate.as_ref() {
            if !gate(key.clone(), token, peer).await {
                return Err(crate::StreamError::Unauthorized(
                    "whep egress denied by gate".into(),
                ));
            }
        }
        self.accept_offer(offer_sdp, key, transport)
    }

    /// Handle a WHEP `POST`: validate the offer and mint a `sendonly` answer.
    /// Returns the resource handle (to `pump`) and the answer SDP.
    ///
    /// This does **not** consult the egress gate; gate playback either at the
    /// HTTP host layer or via [`accept_offer_gated`](Self::accept_offer_gated).
    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).
    /// Packetize `frame` stamping every RTP packet with `ts_ms` (90 kHz). The
    /// caller supplies a sanitized, strictly-monotonic timestamp (see
    /// [`MonoClock`]) rather than the frame's own — source timestamps can jump
    /// backwards (B-frame reordering, mid-stream resets) which freezes a WebRTC
    /// jitter buffer.
    fn packetize_into(&mut self, frame: &MediaFrame, ts_ms: i64, out: &mut Vec<Vec<u8>>) -> bool {
        let ts = (ts_ms.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,
        }
    }
}

/// Maps source frame timestamps onto a **strictly increasing** output clock for
/// RTP egress. A WebRTC receiver's jitter buffer treats any backward (or wildly
/// forward) RTP timestamp as a discontinuity and stalls — yet real sources emit
/// non-monotonic timestamps: B-frame reordering nudges them back by a frame,
/// keyframe-recovery replays cached frames out of band, and encoders/ingest
/// occasionally reset. This clock absorbs all of that: it advances the output by
/// the input delta when sane, by 1 ms when the input goes backwards or stalls,
/// and by the learned nominal frame interval across a large gap — so the egress
/// timeline never regresses regardless of what the source does.
struct MonoClock {
    started: bool,
    last_in: i64,
    out: i64,
    /// Learned typical inter-frame interval (ms), used to bridge discontinuities.
    nominal: i64,
}

impl MonoClock {
    fn new() -> Self {
        Self {
            started: false,
            last_in: 0,
            out: 0,
            nominal: 33, // ~30 fps until the real cadence is learned
        }
    }

    /// Map an input timestamp (ms) to the next strictly-increasing output (ms).
    fn map(&mut self, in_ms: i64) -> i64 {
        if !self.started {
            self.started = true;
            self.last_in = in_ms;
            self.out = in_ms.max(0);
            return self.out;
        }
        let delta = in_ms - self.last_in;
        self.last_in = in_ms;
        let step = if delta <= 0 {
            1 // backward / duplicate: nudge forward minimally
        } else if delta > 1_000 {
            self.nominal // discontinuity: bridge by one nominal frame
        } else {
            self.nominal = delta; // learn the real cadence
            delta
        };
        self.out += step;
        self.out
    }
}

/// Choose the best simulcast layer for a bandwidth `estimate`, with hysteresis
/// to avoid flapping between adjacent layers.
///
/// `layers` is `(key, measured_bitrate_bps)` for every available layer (the base
/// plus any `~rid` siblings); order does not matter. Rules:
///
/// * The lowest-bitrate layer is the always-available floor.
/// * With no estimate yet, keep `current` (or fall to the floor if it vanished).
/// * Otherwise target the highest layer that fits the estimate. **Up-switch**
///   needs 25% headroom over the candidate's bitrate; **down-switch** triggers
///   once the estimate drops below 95% of the current layer's bitrate (congestion
///   is urgent, so it reacts faster than it climbs).
/// * Layers with an unknown (0) bitrate are never up-switch targets — we can't
///   tell if they fit — but the base floor is always eligible.
fn select_layer(
    layers: &[(StreamKey, u64)],
    estimate: Option<u64>,
    current: &StreamKey,
) -> StreamKey {
    if layers.is_empty() {
        return current.clone();
    }
    // The floor: lowest measured bitrate (0/unknown sorts lowest, which is fine —
    // an unmeasured single layer is still the only choice).
    let floor = layers
        .iter()
        .min_by_key(|(_, bps)| *bps)
        .map(|(k, _)| k.clone())
        .unwrap();
    let current_bps = layers.iter().find(|(k, _)| k == current).map(|(_, b)| *b);
    let Some(estimate) = estimate else {
        // No estimate: stay put if the current layer still exists, else the floor.
        return if current_bps.is_some() {
            current.clone()
        } else {
            floor
        };
    };

    // Highest layer with a known bitrate that fits the estimate.
    let desired = layers
        .iter()
        .filter(|(_, bps)| *bps > 0 && *bps <= estimate)
        .max_by_key(|(_, bps)| *bps);
    let Some((desired_key, desired_bps)) = desired else {
        return floor; // nothing fits → floor
    };
    let current_bps = match current_bps {
        Some(b) => b,
        None => return desired_key.clone(), // current gone → take the fit
    };
    if *desired_bps > current_bps {
        // Up-switch only with 25% headroom over the candidate.
        if estimate >= desired_bps.saturating_mul(5) / 4 {
            return desired_key.clone();
        }
    } else if *desired_bps < current_bps {
        // Down-switch once the current layer no longer comfortably fits.
        if estimate < current_bps.saturating_mul(19) / 20 {
            return desired_key.clone();
        }
    }
    current.clone()
}

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 (mut vcfg, _) = handle.cached_configs();
        let replay = handle.replay_buffer();
        // Keep a handle clone solely to relay viewer keyframe requests upstream to
        // the publisher. A clone does not pin the bus open — `StreamHandle::close`
        // empties the shared sender cell on publish-end regardless of clones.
        // Reassigned on an adaptive-bitrate layer switch so keyframe requests
        // target the layer actually being forwarded.
        let mut kf_handle = handle.clone();
        // Release the original handle once setup is done.
        drop(handle);

        // Adaptive bitrate: the layer currently forwarded to this viewer. Starts
        // on the requested stream (the base) and may switch among simulcast
        // siblings as the transport's bandwidth estimate changes.
        let mut current_key = self.key.clone();

        // 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();

        // Strictly-monotonic egress clocks (video + audio): map every source
        // timestamp onto a never-regressing RTP timeline, so B-frame reordering,
        // keyframe-recovery replays, and source resets can't freeze the viewer.
        let mut vclock = MonoClock::new();
        let mut aclock = MonoClock::new();

        // Kept for fast local recovery: when a viewer requests a keyframe (PLI/FIR
        // over RTCP), we re-send the config + the most recent keyframe instead of
        // making the viewer wait for the next natural IDR.
        if let Some(cfg) = vcfg.as_ref() {
            self.send_frame(
                cfg,
                &mut packetizer,
                &mut audio,
                &mut pkts,
                &mut vclock,
                &mut aclock,
            )
            .await?;
        }
        let mut last_keyframe: Option<Arc<MediaFrame>> = None;
        for frame in replay {
            if frame.is_video() && frame.is_keyframe() {
                last_keyframe = Some(frame.clone());
            }
            self.send_frame(
                &frame,
                &mut packetizer,
                &mut audio,
                &mut pkts,
                &mut vclock,
                &mut aclock,
            )
            .await?;
        }

        // Poll the live subscription, the viewer's RTCP feedback, and a periodic
        // adaptive-bitrate tick together. Once `recv_rtcp` yields `None` (a
        // transport with no upstream RTCP, or a closed one) we stop polling it via
        // a never-ready future so the loop can't spin.
        let mut rtcp_open = true;
        let mut abr_tick = tokio::time::interval(std::time::Duration::from_secs(1));
        abr_tick.tick().await; // consume the immediate first tick

        // Viewer-gone detection. The ONLY reliable disconnect signal here is the
        // transport's RTCP channel closing: for the str0m WHEP egress, `recv_rtcp`
        // awaits and yields `None` exactly when the str0m driver task ends — i.e.
        // the peer is genuinely dead (str0m's own ICE/consent timeout fired) or the
        // session was cancelled. A healthy viewer does NOT send periodic RTCP we can
        // observe (str0m only surfaces *keyframe requests* over this seam, not
        // Receiver Reports), so we must NOT reap on "RTCP silence" — doing so tore
        // down every healthy viewer on a fixed timer and forced a reconnect every
        // few seconds.
        //
        // The kernel-default transport (no RTCP at all) instead returns `None`
        // *immediately* on the first poll. We distinguish the two by elapsed time:
        // a `None` within the first moment of the pump means "this transport has no
        // RTCP channel" (stop polling it, keep pumping); a `None` after the pump has
        // been alive a while means "the peer's transport closed" (end the pump).
        const NO_RTCP_GRACE: std::time::Duration = std::time::Duration::from_secs(2);
        let pump_start = std::time::Instant::now();

        // Lag recovery. If a slow link makes this viewer's bus subscription fall
        // behind and the ring overflows, `recv` resynchronizes by SKIPPING
        // frames. Forwarding the post-gap delta frames is useless — they
        // reference frames the viewer never got, so the decoder freezes until the
        // next natural IDR (the ~90s "plays then dies" symptom). Instead, after a
        // lag we drop deltas until the next keyframe and ask the publisher for a
        // fresh one, so the viewer re-primes quickly.
        let mut last_dropped = sub.dropped();
        let mut awaiting_keyframe = false;
        loop {
            let feedback = async {
                if rtcp_open {
                    self.transport.recv_rtcp().await
                } else {
                    std::future::pending().await
                }
            };
            tokio::select! {
                frame = sub.recv() => {
                    let Some(frame) = frame else { break };

                    // Detect a resync gap: the subscription dropped frames to
                    // catch up. Enter keyframe-wait so we don't forward
                    // undecodable deltas, and prod the publisher for an IDR.
                    let dropped = sub.dropped();
                    if dropped > last_dropped {
                        last_dropped = dropped;
                        awaiting_keyframe = true;
                        kf_handle.request_keyframe();
                    }

                    if frame.is_video() {
                        if frame.is_keyframe() {
                            last_keyframe = Some(frame.clone());
                            awaiting_keyframe = false;
                        } else if awaiting_keyframe {
                            // Still waiting for an IDR after a lag — skip this
                            // delta so the decoder isn't fed a dangling reference.
                            continue;
                        }
                    } else if awaiting_keyframe {
                        // Hold audio too until video re-anchors, keeping A/V from
                        // drifting during recovery.
                        continue;
                    }
                    self.send_frame(&frame, &mut packetizer, &mut audio, &mut pkts, &mut vclock, &mut aclock)
                        .await?;
                }
                rtcp = feedback => {
                    match rtcp {
                        Some(buf) => {
                            self.handle_feedback(
                                &buf,
                                &kf_handle,
                                vcfg.as_ref(),
                                last_keyframe.as_ref(),
                                &mut packetizer,
                                &mut audio,
                                &mut pkts,
                                &mut vclock,
                                &mut aclock,
                            )
                            .await?;
                        }
                        None => {
                            // `recv_rtcp` yields `None` when the peer's transport
                            // closed (str0m driver ended → peer dead/cancelled) or
                            // the transport carries no RTCP at all (kernel default,
                            // which returns `None` immediately). Discriminate by how
                            // long the pump has been alive: an early `None` is a
                            // no-RTCP transport (stop polling, keep pumping); a later
                            // `None` is a real disconnect (end the pump so its bus
                            // subscription drops and the live viewer count falls).
                            if pump_start.elapsed() > NO_RTCP_GRACE {
                                tracing::debug!(stream = %self.key, "WHEP egress: peer transport closed, ending pump");
                                break;
                            }
                            rtcp_open = false;
                        }
                    }
                }
                _ = abr_tick.tick() => {
                    // Re-evaluate which simulcast layer best fits the viewer's
                    // current bandwidth estimate; switch the subscription if so.
                    let layers = self.discover_layers();
                    let estimate = self.transport.estimated_bitrate();
                    let target = select_layer(&layers, estimate, &current_key);
                    if target != current_key {
                        if let Ok(next) = self.playback.get_stream(&target) {
                            tracing::debug!(
                                stream = %self.key, from = %current_key, to = %target,
                                estimate_bps = estimate.unwrap_or(0),
                                "WHEP egress: adaptive-bitrate layer switch",
                            );
                            sub = next.subscribe_resilient();
                            vcfg = next.cached_configs().0;
                            kf_handle = next.clone();
                            current_key = target;
                            // Resync the decoder on the new layer: re-send its
                            // config and ask its publisher for a fresh keyframe.
                            if let Some(cfg) = vcfg.as_ref() {
                                self.send_frame(cfg, &mut packetizer, &mut audio, &mut pkts, &mut vclock, &mut aclock)
                                    .await?;
                            }
                            last_keyframe = None;
                            awaiting_keyframe = true;
                            kf_handle.request_keyframe();
                        }
                    }
                }
            }
        }
        Ok(())
    }

    /// Enumerate the simulcast layers available for this viewer's stream as
    /// `(key, measured_video_bitrate_bps)` — the base (the requested
    /// `stream_id`) plus any `stream_id~rid` siblings the WHIP simulcast demux
    /// published. The per-layer bitrate is the live ~1s QoS window, the input to
    /// [`select_layer`]. When the publisher is not simulcast this returns just the
    /// base layer, so layer selection is a no-op.
    fn discover_layers(&self) -> Vec<(StreamKey, u64)> {
        let app = &self.key.app;
        let base = self.key.stream_id.as_str();
        let prefix = format!("{base}~");
        let mut out = Vec::new();
        let ids = self.playback.list_streams(app).unwrap_or_default();
        for id in ids {
            let s = id.as_str();
            if s == base || s.starts_with(&prefix) {
                let key = StreamKey::new(app.as_str(), s);
                let bitrate = self
                    .playback
                    .get_stream(&key)
                    .map(|h| h.qos().video_bitrate_bps)
                    .unwrap_or(0);
                out.push((key, bitrate));
            }
        }
        // Guarantee the base is always present even if enumeration missed it.
        if !out.iter().any(|(k, _)| k == &self.key) {
            out.push((self.key.clone(), 0));
        }
        out
    }

    /// React to a viewer's RTCP compound packet.
    ///
    /// A PLI/FIR triggers fast recovery: re-send the cached config + the most
    /// recent keyframe so the viewer paints immediately rather than waiting for
    /// the next natural IDR. Receiver Reports are logged for QoS visibility;
    /// NACK-driven retransmission is a separate (send-history) feature.
    #[allow(clippy::too_many_arguments)]
    async fn handle_feedback(
        &self,
        rtcp: &[u8],
        kf_handle: &crate::bus::StreamHandle,
        vcfg: Option<&Arc<MediaFrame>>,
        last_keyframe: Option<&Arc<MediaFrame>>,
        packetizer: &mut EgressPacketizer,
        audio: &mut Option<OpusPacketizer>,
        pkts: &mut Vec<Vec<u8>>,
        vclock: &mut MonoClock,
        aclock: &mut MonoClock,
    ) -> Result<()> {
        let mut refresh = false;
        for fb in rtcp::parse_compound(rtcp) {
            match fb {
                rtcp::RtcpFeedback::Pli { .. } | rtcp::RtcpFeedback::Fir { .. } => refresh = true,
                rtcp::RtcpFeedback::ReceiverReport {
                    fraction_lost,
                    cumulative_lost,
                    jitter,
                    ..
                } => {
                    tracing::trace!(
                        stream = %self.key,
                        fraction_lost,
                        cumulative_lost,
                        jitter,
                        "WHEP egress: viewer receiver report",
                    );
                }
                rtcp::RtcpFeedback::Nack { lost, .. } => {
                    tracing::trace!(
                        stream = %self.key,
                        lost = lost.len(),
                        "WHEP egress: viewer NACK (retransmission not yet implemented)",
                    );
                }
                rtcp::RtcpFeedback::Remb { bitrate_bps, .. } => {
                    tracing::trace!(
                        stream = %self.key,
                        bitrate_bps,
                        "WHEP egress: viewer REMB bandwidth estimate",
                    );
                }
            }
        }
        if refresh {
            // Fast local recovery: re-send the cached config + most recent
            // keyframe so the viewer paints without waiting for a natural IDR.
            // The cached frames carry old timestamps, but `send_frame` runs them
            // through `vclock`, which maps their stale time onto the current
            // monotonic output — so the keyframe lands "now" and the viewer
            // accepts it (a raw backward timestamp would be dropped as stale).
            if let Some(cfg) = vcfg {
                self.send_frame(cfg, packetizer, audio, pkts, vclock, aclock)
                    .await?;
            }
            if let Some(kf) = last_keyframe {
                self.send_frame(kf, packetizer, audio, pkts, vclock, aclock)
                    .await?;
            }
            // Also relay upstream: ask the publisher for a fresh IDR, covering the
            // case where the cache has no usable keyframe (e.g. just after join).
            kf_handle.request_keyframe();
        }
        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).
    #[allow(clippy::too_many_arguments)]
    async fn send_frame(
        &self,
        frame: &MediaFrame,
        packetizer: &mut EgressPacketizer,
        audio: &mut Option<OpusPacketizer>,
        pkts: &mut Vec<Vec<u8>>,
        vclock: &mut MonoClock,
        aclock: &mut MonoClock,
    ) -> Result<()> {
        if frame.is_audio() {
            if let Some(ap) = audio.as_mut() {
                if frame.codec == CodecId::Opus {
                    let ts = (aclock.map(frame.dts).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(());
        }
        let ts_ms = vclock.map(frame.dts);
        if packetizer.packetize_into(frame, ts_ms, 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;

    #[test]
    fn mono_clock_is_strictly_increasing_through_glitches() {
        let mut c = MonoClock::new();
        // Normal cadence, a B-frame-style backward step, a duplicate, and a large
        // forward discontinuity — the output must never regress.
        let inputs = [
            1000, 1033, 1066, 1050, // backward (reorder)
            1100, 1100, // duplicate
            1133, 50_000, // huge jump (reset)
            50_033, 50_066,
        ];
        let mut prev = i64::MIN;
        for ms in inputs {
            let out = c.map(ms);
            assert!(out > prev, "output regressed: {out} after {prev}");
            prev = out;
        }
    }

    #[test]
    fn mono_clock_passes_through_steady_cadence() {
        let mut c = MonoClock::new();
        // A clean 30 fps stream maps 1:1 (deltas preserved) after the first frame.
        assert_eq!(c.map(0), 0);
        assert_eq!(c.map(33), 33);
        assert_eq!(c.map(66), 66);
    }

    /// 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>>>,
        /// Inbound viewer RTCP script (WHEP), drained by `recv_rtcp`.
        rtcp_in: Mutex<std::collections::VecDeque<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()),
                rtcp_in: Mutex::new(Default::default()),
                keep_open: false,
            }
        }

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

        /// A kept-open transport that delivers `rtcp` packets to the egress feedback
        /// loop (then blocks), for exercising WHEP viewer-feedback handling.
        fn with_inbound_rtcp(rtcp: std::collections::VecDeque<Vec<u8>>) -> Self {
            Self {
                keep_open: true,
                rtcp_in: Mutex::new(rtcp),
                ..Self::with_packets(Default::default())
            }
        }
    }

    #[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(())
        }
        async fn recv_rtcp(&self) -> Option<Vec<u8>> {
            match self.rtcp_in.lock().await.pop_front() {
                Some(p) => Some(p),
                None if self.keep_open => std::future::pending().await,
                None => None,
            }
        }
    }

    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);
    }

    /// The WHEP egress gate denies a request whose token the gate rejects, and
    /// permits when no gate is installed (permit-all, matching RTSP/SRT egress).
    #[tokio::test]
    async fn whep_gate_denies_and_permits() {
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live"))
            .build();
        let key = StreamKey::new("live", "show");
        let offer = "v=0\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\na=rtpmap:96 H264/90000\r\n";
        let transport = || Arc::new(FakeTransport::with_packets(Default::default()));

        // Gate that allows only the token "good".
        let gate: crate::auth::EgressGate = Arc::new(|_key, token, _peer| {
            Box::pin(async move { token.as_deref() == Some("good") })
        });
        let whep = WhepEndpoint::new(engine.clone()).with_gate(gate);

        // Wrong/absent token → denied.
        assert!(whep
            .accept_offer_gated(offer, key.clone(), None, None, transport())
            .await
            .is_err());
        // Correct token → allowed.
        assert!(whep
            .accept_offer_gated(offer, key.clone(), Some("good".into()), None, transport())
            .await
            .is_ok());

        // No gate installed → permit-all even without a token.
        let open = WhepEndpoint::new(engine.clone());
        assert!(open
            .accept_offer_gated(offer, key.clone(), None, None, transport())
            .await
            .is_ok());
    }

    /// WHEP viewer feedback: a PLI arriving over `recv_rtcp` makes the egress
    /// re-send the cached config + most recent keyframe for fast recovery, so the
    /// keyframe NAL is transmitted again after the initial instant-start replay.
    #[tokio::test]
    async fn whep_egress_resends_keyframe_on_viewer_pli() {
        use crate::FrameFlags;
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live").gop_cache(8))
            .build();
        let key = StreamKey::new("live", "fb");

        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();

        // The viewer's feedback script: one PLI, delivered once the loop starts.
        let mut script = std::collections::VecDeque::new();
        script.push_back(rtcp::build_pli(0, 0));
        let transport = Arc::new(FakeTransport::with_inbound_rtcp(script));

        let whep = WhepEndpoint::new(engine.clone());
        let offer = "v=0\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\na=rtpmap:96 H264/90000\r\n";
        let (resource, _) = whep
            .accept_offer(offer, key.clone(), transport.clone())
            .unwrap();
        let pump = tokio::spawn(resource.pump());

        // Wait until the keyframe NAL (0x65) has been sent twice: once on the
        // instant-start replay, once again from the PLI-triggered refresh.
        let count_keyframes = |pkts: &[Vec<u8>]| {
            pkts.iter()
                .filter(|p| {
                    RtpHeader::parse(p)
                        .map(|h| p[h.payload_offset..].windows(1).any(|w| w[0] == 0x65))
                        .unwrap_or(false)
                })
                .count()
        };
        let mut refreshed = false;
        for _ in 0..500 {
            if count_keyframes(&transport.sent_rtp.lock().await) >= 2 {
                refreshed = true;
                break;
            }
            // A real (short) sleep lets the spawned pump task make progress and
            // its timer-driven branches advance, instead of racing a yield budget.
            tokio::time::sleep(std::time::Duration::from_millis(2)).await;
        }
        session.finish().await.unwrap();
        let _ = pump.await.unwrap();
        assert!(refreshed, "viewer PLI re-sent the keyframe");
    }

    #[test]
    fn select_layer_picks_fit_with_hysteresis() {
        let k = |s: &str| StreamKey::new("live", s);
        // Three layers: low 300k, mid 800k, high 2.5M.
        let layers = vec![
            (k("show"), 300_000u64),
            (k("show~h"), 800_000),
            (k("show~f"), 2_500_000),
        ];

        // Plenty of bandwidth, currently on low → up-switch to high (>25% headroom
        // over 2.5M needs >= 3.125M; give 4M).
        assert_eq!(
            select_layer(&layers, Some(4_000_000), &k("show")),
            k("show~f")
        );

        // Marginal bandwidth just above mid's bitrate but below the 25% headroom
        // for up-switching from low → no up-switch (stays low).
        assert_eq!(select_layer(&layers, Some(820_000), &k("show")), k("show"));

        // On high, bandwidth collapses below 95% of high → down-switch to the best
        // fit (mid at 800k fits 900k).
        assert_eq!(
            select_layer(&layers, Some(900_000), &k("show~f")),
            k("show~h")
        );

        // No estimate yet → stay on the current layer.
        assert_eq!(select_layer(&layers, None, &k("show~h")), k("show~h"));

        // Nothing fits (estimate below the floor) → the floor layer.
        assert_eq!(
            select_layer(&layers, Some(100_000), &k("show~f")),
            k("show")
        );

        // Single layer (no simulcast) → always itself.
        let one = vec![(k("solo"), 0u64)];
        assert_eq!(select_layer(&one, Some(5_000_000), &k("solo")), k("solo"));
    }

    /// WHEP layer discovery: the base stream plus its `~rid` simulcast siblings
    /// are enumerated for adaptive-bitrate selection; unrelated streams are not.
    #[tokio::test]
    async fn discover_layers_lists_base_and_simulcast_siblings() {
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live").gop_cache(4))
            .build();
        let ctx = IngestContext::new(engine.clone());
        // Publish a base layer, two simulcast siblings, and an unrelated stream.
        for id in ["show", "show~h", "show~f", "other"] {
            let s = ctx.open_publish(StreamKey::new("live", id)).await.unwrap();
            s.publish_frame(MediaFrame::new_video(
                0,
                0,
                bytes::Bytes::from_static(&[0, 0, 0, 1, 0x65]),
                CodecId::H264,
                true,
            ))
            .unwrap();
            std::mem::forget(s); // keep the streams live for the listing
        }

        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, _) = whep
            .accept_offer(offer, StreamKey::new("live", "show"), transport)
            .unwrap();

        let mut ids: Vec<String> = resource
            .discover_layers()
            .into_iter()
            .map(|(k, _)| k.stream_id.as_str().to_string())
            .collect();
        ids.sort();
        assert_eq!(ids, vec!["show", "show~f", "show~h"]);
    }

    /// The upstream keyframe backchannel: a viewer PLI on the WHEP egress calls
    /// `StreamHandle::request_keyframe`, which the publisher's ingest loop awaits
    /// via `keyframe_requested` — so a browser PLI reaches the WHIP publisher.
    #[tokio::test]
    async fn request_keyframe_signals_the_publishers_handle() {
        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("live").gop_cache(8))
            .build();
        let key = StreamKey::new("live", "kf");

        let ctx = IngestContext::new(engine.clone());
        let session = ctx.open_publish(key.clone()).await.unwrap();
        // The publisher's loop awaits a keyframe request on its handle.
        let pub_handle = session.handle().clone();
        let waiter = tokio::spawn(async move {
            tokio::time::timeout(
                std::time::Duration::from_secs(2),
                pub_handle.keyframe_requested(),
            )
            .await
        });

        // A playback consumer resolves the *same* stream and requests a keyframe.
        let view_handle = engine.get_stream(&key).unwrap();
        // Give the waiter a moment to park on `notified()` before signaling.
        tokio::task::yield_now().await;
        view_handle.request_keyframe();

        assert!(waiter.await.unwrap().is_ok(), "publisher saw the request");
    }

    /// 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);
    }
}