arcly-stream 0.8.3

//! RTSP **egress server** — serve a live stream to RTSP players (VLC, ffmpeg).
//!
//! The serving counterpart to the client-pull [ingest handler](super::RtspHandler):
//! [`RtspServer`] accepts TCP connections, runs the
//! `OPTIONS → DESCRIBE → SETUP → PLAY → TEARDOWN` state machine, and streams the
//! requested stream's H.264 access units as RTP over the TCP-interleaved
//! transport (RFC 2326 §10.12), reusing the shared
//! [`RtpPacketizer`](crate::protocol::rtp::RtpPacketizer).
//!
//! Interleaved (RTP-over-TCP) transport only — the universally-supported path
//! that needs no separate UDP ports. The stream is selected from the request
//! URI's `/app/stream` path.

use std::net::SocketAddr;
use std::ops::ControlFlow;
use std::sync::Arc;

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream, UdpSocket};
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};

use super::message::{InterleavedFrame, RtspRequest};
use super::sdp::Sdp;
use crate::auth::{Credentials, EgressGate};
use crate::bus::PlaybackRegistry;
use crate::inbound::IngestContext;
use crate::protocol::rtp::{
    AccessUnit, DepacketizeError, H264Depacketizer, H265Depacketizer, RtpHeader, RtpPacketizer,
};
use crate::{CodecId, MediaFrame, Result, StreamKey};

/// After this many consecutive RTP packets fail to depacketize, the ingest is
/// treated as a codec/format mismatch and the connection is dropped — instead of
/// flooding the log and feeding garbage forever.
const MAX_CONSECUTIVE_DEPACK_FAILURES: u32 = 64;

/// Publish one depacketized video access unit, synthesizing the out-of-band
/// **CONFIG** frame that container packagers (HLS/DASH fMP4 & MPEG-TS) need.
///
/// RTP delivers parameter sets (SPS/PPS for H.264, VPS/SPS/PPS for H.265) only
/// in-band, prepended to each IDR access unit. An RTSP *player* recovers them
/// from the bitstream, but a fMP4/TS muxer must build its init segment from a
/// dedicated `FrameFlags::CONFIG` access unit (see `PackagerCore::ensure_init`)
/// — without one, no init segment is produced and browser HLS/DASH preview
/// never starts. So on every keyframe we extract the parameter sets and emit a
/// CONFIG frame first; the bus caches it for late-joining packagers/subscribers.
fn publish_au(
    session: &crate::inbound::PublishSession,
    codec: CodecId,
    au: AccessUnit,
    sdp_config: &Option<bytes::Bytes>,
) -> Result<()> {
    let pts = (au.timestamp / 90) as i64; // 90 kHz → ms
    if au.keyframe {
        // Prefer parameter sets carried in-band on this IDR; otherwise fall back
        // to the SDP `sprop-*` sets. Emitting the CONFIG *inline before each
        // keyframe* (not just once at RECORD) is what lets a late-subscribing
        // packager — which joins after RECORD and replays nothing — build its
        // init segment from the first keyframe it sees, mirroring RTMP/MPEG-TS.
        let cfg =
            crate::codec::dispatch::parameter_sets(codec, &au.data).or_else(|| sdp_config.clone());
        if let Some(params) = cfg {
            let mut frame = MediaFrame::new_video(pts, pts, params, codec, true);
            frame.flags |= crate::FrameFlags::CONFIG;
            session.publish_frame(frame)?;
        }
    }
    let mf = MediaFrame::new_video(pts, pts, au.data, codec, au.keyframe);
    session.publish_frame(mf)?;
    Ok(())
}

/// Publish a video CONFIG frame carrying `annexb` parameter sets (SPS/PPS, plus
/// VPS for H.265). Sent once at `RECORD` from the `ANNOUNCE` SDP's `sprop-*`
/// fields so the bus caches a decoder config for fMP4/TS packagers (HLS/DASH)
/// even when the encoder never repeats parameter sets in-band.
fn publish_config(
    session: &crate::inbound::PublishSession,
    codec: CodecId,
    annexb: bytes::Bytes,
) -> Result<()> {
    let mut cfg = MediaFrame::new_video(0, 0, annexb, codec, true);
    cfg.flags |= crate::FrameFlags::CONFIG;
    session.publish_frame(cfg)?;
    Ok(())
}

/// Video depacketizer chosen from the publisher's announced codec. H.264 and
/// H.265 share the same RTP push API and `AccessUnit` shape, so the ingest loop
/// is codec-agnostic past this enum.
enum VideoDepack {
    H264(H264Depacketizer),
    H265(H265Depacketizer),
}

impl VideoDepack {
    fn new(codec: CodecId) -> Self {
        match codec {
            CodecId::H265 => VideoDepack::H265(H265Depacketizer::new()),
            _ => VideoDepack::H264(H264Depacketizer::new()),
        }
    }
    fn codec(&self) -> CodecId {
        match self {
            VideoDepack::H264(_) => CodecId::H264,
            VideoDepack::H265(_) => CodecId::H265,
        }
    }
    fn push(
        &mut self,
        payload: &[u8],
        marker: bool,
        timestamp: u32,
        sequence: u16,
    ) -> std::result::Result<Option<AccessUnit>, DepacketizeError> {
        match self {
            VideoDepack::H264(d) => d.push(payload, marker, timestamp, sequence),
            VideoDepack::H265(d) => d.push(payload, marker, timestamp, sequence),
        }
    }
}

/// The video codec announced in an `ANNOUNCE` SDP body (defaults to H.264).
fn video_codec_from_sdp(body: &[u8]) -> CodecId {
    let sdp = Sdp::parse(&String::from_utf8_lossy(body));
    let enc = sdp
        .media
        .iter()
        .find(|m| m.media == "video")
        .and_then(|m| m.encoding.as_deref())
        .map(|e| e.to_ascii_uppercase());
    match enc.as_deref() {
        Some("H265") | Some("HEVC") => CodecId::H265,
        _ => CodecId::H264,
    }
}

/// A negotiated UDP media transport for one RTSP session: our bound RTP socket
/// and the player/publisher's RTP address (`peer ip : client_port`).
struct UdpMedia {
    rtp: Arc<UdpSocket>,
    client_rtp: SocketAddr,
    server_rtp_port: u16,
}

/// Parse `client_port=A-B` from a SETUP `Transport` header.
fn parse_client_ports(transport: &str) -> Option<(u16, u16)> {
    let v = transport
        .split(';')
        .find_map(|p| p.trim().strip_prefix("client_port="))?;
    let (a, b) = v.split_once('-')?;
    Some((a.trim().parse().ok()?, b.trim().parse().ok()?))
}

/// A unified RTSP **server**: serves players (`OPTIONS/DESCRIBE/SETUP/PLAY`) and
/// ingests publishers (`ANNOUNCE/SETUP/RECORD`) on one TCP port, mirroring a
/// real RTSP server (e.g. MediaMTX). Stream is selected by the request URI's
/// `/app/stream` path.
pub struct RtspServer {
    playback: Arc<dyn PlaybackRegistry>,
    ingest: IngestContext,
    bind: SocketAddr,
    /// Egress gate (per-app toggle + play token). `None` = open playback.
    gate: Option<EgressGate>,
}

impl RtspServer {
    /// Serve streams from `playback` and ingest publishers via `ingest`,
    /// listening on `bind`.
    pub fn new(
        playback: Arc<dyn PlaybackRegistry>,
        ingest: IngestContext,
        bind: SocketAddr,
    ) -> Self {
        Self {
            playback,
            ingest,
            bind,
            gate: None,
        }
    }

    /// Gate playback (egress) requests through `gate` (per-app toggle + token).
    pub fn with_gate(mut self, gate: EgressGate) -> Self {
        self.gate = Some(gate);
        self
    }

    /// Accept and serve RTSP connections until `shutdown` fires.
    pub async fn run(self, shutdown: CancellationToken) -> Result<()> {
        let listener = TcpListener::bind(self.bind).await?;
        info!(bind = %self.bind, "rtsp server listening");
        loop {
            tokio::select! {
                _ = shutdown.cancelled() => break,
                accepted = listener.accept() => {
                    let (sock, peer) = match accepted {
                        Ok(v) => v,
                        Err(e) => { warn!(error = %e, "rtsp accept failed"); continue; }
                    };
                    let playback = Arc::clone(&self.playback);
                    let ingest = self.ingest.clone();
                    let gate = self.gate.clone();
                    let shutdown = shutdown.clone();
                    tokio::spawn(async move {
                        if let Err(e) = serve_connection(sock, playback, ingest, gate, shutdown).await {
                            debug!(%peer, error = %e, "rtsp connection ended");
                        }
                    });
                }
            }
        }
        Ok(())
    }
}

/// The `token` query parameter from an RTSP URI, if present.
fn uri_token(uri: &str) -> Option<String> {
    crate::auth::token_from_query(uri)
}

/// The RTP payload type advertised for H.264 (must match the SDP `rtpmap`).
const PT_H264: u8 = 96;
/// The interleaved channel for RTP (RTCP would be channel 1).
const RTP_CHANNEL: u8 = 0;

async fn serve_connection(
    mut sock: TcpStream,
    playback: Arc<dyn PlaybackRegistry>,
    ingest: IngestContext,
    gate: Option<EgressGate>,
    shutdown: CancellationToken,
) -> Result<()> {
    let mut buf = Vec::with_capacity(2048);
    // A distinct session id per connection (a real server tracks state per id).
    let session = new_session_id();
    // Set by ANNOUNCE: this connection is a publisher; RECORD will ingest it.
    let mut announce_key: Option<StreamKey> = None;
    // Video codec from the ANNOUNCE SDP (H.264 unless it says H.265/HEVC).
    let mut announce_codec = CodecId::H264;
    // Out-of-band parameter sets from the ANNOUNCE SDP's `sprop-*` fields, used
    // to seed the decoder config (CONFIG frame) at RECORD.
    let mut announce_config: Option<bytes::Bytes> = None;
    // The TCP peer: its IP is the UDP media destination when UDP is negotiated,
    // and the player address handed to the egress gate for token IP-binding.
    let peer = sock.peer_addr().ok();
    let peer_ip = peer.map(|a| a.ip());
    // Set by a UDP SETUP; consumed by RECORD/PLAY. `None` = TCP-interleaved.
    let mut udp: Option<UdpMedia> = None;
    loop {
        let Some((req, body)) = read_request(&mut sock, &mut buf).await? else {
            return Ok(()); // peer closed
        };
        let key = stream_key_from_uri(&req.uri);
        match req.method.as_str() {
            "OPTIONS" => {
                sock.write_all(options_response(req.cseq).as_bytes())
                    .await?
            }
            // ── Publisher (ingest) path: ANNOUNCE → SETUP → RECORD ──
            "ANNOUNCE" => match &key {
                Some(k) => {
                    announce_key = Some(k.clone());
                    let sdp = Sdp::parse(&String::from_utf8_lossy(&body));
                    announce_codec = video_codec_from_sdp(&body);
                    announce_config = sdp.video_config_annexb();
                    sock.write_all(simple_ok(req.cseq, &session).as_bytes())
                        .await?;
                }
                None => sock.write_all(not_found(req.cseq).as_bytes()).await?,
            },
            "RECORD" => match announce_key.take().or_else(|| key.clone()) {
                Some(k) => {
                    sock.write_all(simple_ok(req.cseq, &session).as_bytes())
                        .await?;
                    // Admission tags `proto=rtsp` so the per-app RTSP ingress
                    // toggle / publish token are enforced; token rides the URI.
                    let mut creds = Credentials::default();
                    creds.params.push(("proto".into(), "rtsp".into()));
                    creds.token = uri_token(&req.uri);
                    // Carry the peer so admission can bind a signed publish token
                    // to the publisher's IP when the app enables it.
                    creds.addr = peer;
                    let sess = ingest.open_publish_checked(k.clone(), &creds).await?;
                    // Seed the decoder config from the ANNOUNCE SDP so HLS/DASH
                    // packagers can build an init segment immediately, before the
                    // first keyframe (encoders rarely repeat parameter sets
                    // in-band over RTP). In-band sets, when present, still flow
                    // through `publish_au` as additional CONFIG frames.
                    // Decoder config from the ANNOUNCE SDP's `sprop-*` fields, if
                    // present. Seeded into the bus cache now and re-emitted inline
                    // before each keyframe (see `publish_au`) so packagers that
                    // subscribe after RECORD still get it.
                    let config = announce_config.take();
                    if let Some(cfg) = &config {
                        publish_config(&sess, announce_codec, cfg.clone())?;
                    }
                    match udp.take() {
                        Some(u) => {
                            info!(stream = %k, codec = ?announce_codec, "rtsp ingest (RECORD/UDP) started");
                            return record_udp(sock, sess, u, announce_codec, config, shutdown)
                                .await;
                        }
                        None => {
                            info!(stream = %k, codec = ?announce_codec, "rtsp ingest (RECORD/TCP) started");
                            return record(sock, sess, buf, announce_codec, config, shutdown).await;
                        }
                    }
                }
                None => sock.write_all(not_found(req.cseq).as_bytes()).await?,
            },
            // ── Player (egress) path: DESCRIBE → SETUP → PLAY ──
            "DESCRIBE" => match key.as_ref().and_then(|k| playback.get_stream(k).ok()) {
                Some(handle) => {
                    // Advertise the stream's real codec (H.264 or H.265) so the
                    // player decodes it correctly.
                    let codec = stream_video_codec(&handle);
                    let sdp = build_sdp(codec);
                    sock.write_all(describe_response(req.cseq, &req.uri, &sdp).as_bytes())
                        .await?;
                }
                None => sock.write_all(not_found(req.cseq).as_bytes()).await?,
            },
            "SETUP" => {
                let transport = req
                    .headers
                    .iter()
                    .find(|(n, _)| n.eq_ignore_ascii_case("transport"))
                    .map(|(_, v)| v.as_str())
                    .unwrap_or("");
                let is_tcp = transport.is_empty()
                    || transport.contains("TCP")
                    || transport.contains("interleaved");
                if is_tcp {
                    // TCP-interleaved: echo the client's channels + record mode.
                    udp = None;
                    sock.write_all(setup_response(req.cseq, &session, transport).as_bytes())
                        .await?;
                } else if let Some((cl_rtp, cl_rtcp)) = parse_client_ports(transport) {
                    // UDP: bind a server RTP socket and advertise its port.
                    match (peer_ip, UdpSocket::bind("0.0.0.0:0").await.ok()) {
                        (Some(ip), Some(rtp_sock)) => {
                            let server_rtp_port =
                                rtp_sock.local_addr().map(|a| a.port()).unwrap_or(0);
                            udp = Some(UdpMedia {
                                rtp: Arc::new(rtp_sock),
                                client_rtp: SocketAddr::new(ip, cl_rtp),
                                server_rtp_port,
                            });
                            let record = transport.to_ascii_lowercase().contains("mode=record");
                            sock.write_all(
                                setup_response_udp(
                                    req.cseq,
                                    &session,
                                    cl_rtp,
                                    cl_rtcp,
                                    server_rtp_port,
                                    server_rtp_port.wrapping_add(1),
                                    record,
                                )
                                .as_bytes(),
                            )
                            .await?;
                        }
                        _ => {
                            sock.write_all(unsupported_transport(req.cseq).as_bytes())
                                .await?;
                        }
                    }
                } else {
                    sock.write_all(unsupported_transport(req.cseq).as_bytes())
                        .await?;
                }
            }
            "PLAY" => {
                let live = key.filter(|k| playback.get_stream(k).is_ok());
                match live {
                    Some(key) => {
                        // Egress gate: per-app toggle + play token.
                        let allowed = match gate.as_ref() {
                            Some(g) => g(key.clone(), uri_token(&req.uri), peer).await,
                            None => true,
                        };
                        if !allowed {
                            sock.write_all(unauthorized(req.cseq).as_bytes()).await?;
                            continue;
                        }
                        sock.write_all(play_response(req.cseq, &session).as_bytes())
                            .await?;
                        return match udp.take() {
                            Some(u) => play_udp(sock, &playback, key, u, shutdown).await,
                            None => play(sock, &playback, key, shutdown).await,
                        };
                    }
                    None => sock.write_all(not_found(req.cseq).as_bytes()).await?,
                }
            }
            "TEARDOWN" => {
                sock.write_all(simple_ok(req.cseq, &session).as_bytes())
                    .await?;
                return Ok(());
            }
            other => {
                debug!(method = other, "rtsp: unsupported method");
                sock.write_all(not_implemented(req.cseq).as_bytes()).await?;
            }
        }
    }
}

/// Ingest loop: read TCP-interleaved RTP from a publisher, depacketize channel-0
/// H.264 into access units, and publish them until the publisher disconnects or
/// `shutdown` fires. `carry` is any bytes already buffered after RECORD.
async fn record(
    mut sock: TcpStream,
    session: crate::inbound::PublishSession,
    mut carry: Vec<u8>,
    codec: CodecId,
    config: Option<bytes::Bytes>,
    shutdown: CancellationToken,
) -> Result<()> {
    let mut depack = VideoDepack::new(codec);
    let mut fails = 0u32;
    drain_record(&mut carry, &mut depack, &session, &mut fails, &config)?;
    let mut read = [0u8; 16 * 1024];
    loop {
        tokio::select! {
            _ = shutdown.cancelled() => break,
            n = sock.read(&mut read) => {
                let n = n?;
                if n == 0 { break; }
                carry.extend_from_slice(&read[..n]);
                drain_record(&mut carry, &mut depack, &session, &mut fails, &config)?;
            }
        }
    }
    session.finish().await
}

/// Ingest loop for **UDP** transport: each datagram on our RTP socket is one RTP
/// packet — depacketize H.264 and publish. Ends on the publisher's RTSP control
/// connection closing (TEARDOWN / EOF) or `shutdown`.
async fn record_udp(
    mut sock: TcpStream,
    session: crate::inbound::PublishSession,
    udp: UdpMedia,
    codec: CodecId,
    config: Option<bytes::Bytes>,
    shutdown: CancellationToken,
) -> Result<()> {
    let mut depack = VideoDepack::new(codec);
    let mut dgram = vec![0u8; 64 * 1024];
    let mut ctl = [0u8; 4096];
    let mut fails = 0u32;
    loop {
        tokio::select! {
            _ = shutdown.cancelled() => break,
            // Publisher closed the RTSP control connection.
            n = sock.read(&mut ctl) => {
                if n.unwrap_or(0) == 0 { break; }
            }
            r = udp.rtp.recv_from(&mut dgram) => {
                let Ok((n, _from)) = r else { continue };
                let Some(header) = RtpHeader::parse(&dgram[..n]) else { continue };
                let payload = &dgram[header.payload_offset..n];
                match depack.push(payload, header.marker, header.timestamp, header.sequence) {
                    Ok(Some(au)) => {
                        fails = 0;
                        publish_au(&session, codec, au, &config)?;
                    }
                    Ok(None) => fails = 0,
                    Err(_) => {
                        fails += 1;
                        if fails >= MAX_CONSECUTIVE_DEPACK_FAILURES {
                            warn!(?codec, "rtsp udp ingest: repeated depacketize \
                                  failures — aborting (codec mismatch?)");
                            break;
                        }
                    }
                }
            }
        }
    }
    session.finish().await
}

/// Egress for **UDP** transport: packetize the stream's H.264 access units to RTP
/// and send each packet as a datagram to the player's `client_port`. Stops when
/// the stream ends, the player closes the RTSP control connection, or shutdown.
async fn play_udp(
    mut sock: TcpStream,
    playback: &Arc<dyn PlaybackRegistry>,
    key: StreamKey,
    udp: UdpMedia,
    shutdown: CancellationToken,
) -> Result<()> {
    let handle = playback.get_stream(&key)?;
    let codec = stream_video_codec(&handle);
    // Replay the GOP/instant-start (parameter sets + a keyframe), then live.
    let mut sub = handle.subscribe_resilient();
    let mut packetizer = packetizer_for(codec);
    let mut pkts: Vec<Vec<u8>> = Vec::new();
    let mut ctl = [0u8; 4096];
    loop {
        tokio::select! {
            _ = shutdown.cancelled() => break,
            n = sock.read(&mut ctl) => {
                if n.unwrap_or(0) == 0 { break; } // player gone / TEARDOWN
            }
            frame = sub.recv() => {
                let Some(frame) = frame else { break };
                if !frame.is_video() || frame.codec != codec {
                    continue;
                }
                let timestamp = (frame.pts.max(0) as u64).wrapping_mul(90) as u32;
                packetizer.packetize_into(&frame.data, timestamp, &mut pkts);
                for pkt in pkts.iter() {
                    if udp.rtp.send_to(pkt, udp.client_rtp).await.is_err() {
                        return Ok(());
                    }
                }
            }
        }
    }
    let _ = udp.server_rtp_port; // (advertised in SETUP)
    Ok(())
}

/// Consume whole interleaved frames from `buf`, depacketizing channel-0 video
/// RTP into access units and publishing them. Leaves any partial frame in `buf`.
fn drain_record(
    buf: &mut Vec<u8>,
    depack: &mut VideoDepack,
    session: &crate::inbound::PublishSession,
    fails: &mut u32,
    config: &Option<bytes::Bytes>,
) -> Result<()> {
    // Desync guard: interleaved data must begin with `$`. Anything else means
    // the peer isn't speaking TCP-interleaved RTP (e.g. it negotiated UDP but
    // dumped bytes here) — bail instead of mis-framing on stray `$` bytes and
    // flooding logs with bogus NAL types.
    if buf.first().is_some_and(|&b| b != b'$') {
        return Err(crate::StreamError::protocol(
            "rtsp ingest: non-interleaved data on control connection",
        ));
    }
    let codec = depack.codec();
    let mut consumed = 0;
    while let Some((frame, len)) = InterleavedFrame::parse(&buf[consumed..]) {
        consumed += len;
        // Even channel = RTP media, odd = RTCP (ignored). Video on channel 0.
        if frame.channel != 0 {
            continue;
        }
        let Some(header) = RtpHeader::parse(frame.payload) else {
            continue;
        };
        let payload = &frame.payload[header.payload_offset..];
        match depack.push(payload, header.marker, header.timestamp, header.sequence) {
            Ok(Some(au)) => {
                *fails = 0;
                publish_au(session, codec, au, config)?;
            }
            Ok(None) => *fails = 0,
            Err(_) => {
                *fails += 1;
                if *fails >= MAX_CONSECUTIVE_DEPACK_FAILURES {
                    warn!(
                        ?codec,
                        "rtsp ingest: repeated depacketize failures — \
                          aborting (codec mismatch or corrupt RTP?)"
                    );
                    return Err(crate::StreamError::protocol(
                        "rtsp ingest: depacketize failure threshold",
                    ));
                }
            }
        }
    }
    buf.drain(..consumed);
    Ok(())
}

/// Stream `key`'s H.264 access units as interleaved RTP until the stream ends,
/// the client disconnects, or `shutdown` fires.
async fn play(
    mut sock: TcpStream,
    playback: &Arc<dyn PlaybackRegistry>,
    key: StreamKey,
    shutdown: CancellationToken,
) -> Result<()> {
    let handle = playback.get_stream(&key)?;
    let codec = stream_video_codec(&handle);
    let packetizer = packetizer_for(codec);

    // Replay the instant-start buffer then forward live frames; a write error
    // means the player disconnected, which ends the session gracefully.
    let mut sink = RtspSink {
        sock: &mut sock,
        packetizer,
        codec,
        pkts: Vec::new(),
        wbuf: Vec::with_capacity(1500),
    };
    handle.drive_to(&shutdown, &mut sink).await
}

/// Streams a stream's frames to one RTSP player as interleaved RTP-over-TCP.
struct RtspSink<'a> {
    sock: &'a mut TcpStream,
    packetizer: RtpPacketizer,
    /// The negotiated video codec; frames of any other codec are skipped.
    codec: CodecId,
    /// Reused across frames: the per-frame RTP packet buffers.
    pkts: Vec<Vec<u8>>,
    /// Reused across frames: the interleaved-framing write buffer.
    wbuf: Vec<u8>,
}

#[async_trait::async_trait]
impl crate::bus::FrameSink for RtspSink<'_> {
    async fn send(&mut self, frame: Arc<MediaFrame>) -> Result<ControlFlow<()>> {
        // A write error means the player disconnected — stop the drive cleanly.
        match send_frame(
            self.sock,
            &mut self.packetizer,
            self.codec,
            &mut self.pkts,
            &mut self.wbuf,
            &frame,
        )
        .await
        {
            Ok(()) => Ok(ControlFlow::Continue(())),
            Err(_) => Ok(ControlFlow::Break(())),
        }
    }
}

async fn send_frame(
    sock: &mut TcpStream,
    packetizer: &mut RtpPacketizer,
    codec: CodecId,
    pkts: &mut Vec<Vec<u8>>,
    wbuf: &mut Vec<u8>,
    frame: &MediaFrame,
) -> std::io::Result<()> {
    if !frame.is_video() || frame.codec != codec {
        return Ok(()); // only the negotiated video codec is served
    }
    // Clamp to 0 before scaling: a negative PTS cast straight to u64 would become
    // an enormous value and make the RTP timestamp jump wildly.
    let timestamp = (frame.pts.max(0) as u64).wrapping_mul(90) as u32; // ms → 90 kHz
    packetizer.packetize_into(&frame.data, timestamp, pkts);
    // Coalesce every interleaved RTP packet of this access unit into one write
    // (and one reused buffer) rather than a syscall + allocation per packet.
    wbuf.clear();
    for pkt in pkts.iter() {
        frame_interleaved(RTP_CHANNEL, pkt, wbuf);
    }
    sock.write_all(wbuf).await
}

/// Append an RTP packet framed for the RTSP TCP-interleaved transport to `out`:
/// `$` + 1-byte channel + 2-byte big-endian length + the RTP packet.
fn frame_interleaved(channel: u8, rtp: &[u8], out: &mut Vec<u8>) {
    out.push(b'$');
    out.push(channel);
    out.extend_from_slice(&(rtp.len() as u16).to_be_bytes());
    out.extend_from_slice(rtp);
}

/// Read one RTSP request (headers + any `Content-Length` body — e.g. the SDP of
/// an `ANNOUNCE`) from `sock`, returning `None` on a clean EOF. Bytes past the
/// request (e.g. interleaved RTP after `RECORD`) stay in `buf`.
async fn read_request(
    sock: &mut TcpStream,
    buf: &mut Vec<u8>,
) -> Result<Option<(RtspRequest, Vec<u8>)>> {
    let mut tmp = [0u8; 1024];
    loop {
        if let Some(end) = find_double_crlf(buf) {
            let head = String::from_utf8_lossy(&buf[..end]).into_owned();
            let Some(req) = RtspRequest::parse(&head) else {
                buf.drain(..end);
                return Ok(None);
            };
            let want = content_length(&req);
            // Wait for the full body to arrive.
            while buf.len() < end + want {
                let n = sock.read(&mut tmp).await?;
                if n == 0 {
                    return Ok(None);
                }
                buf.extend_from_slice(&tmp[..n]);
            }
            let body = buf[end..end + want].to_vec();
            buf.drain(..end + want);
            return Ok(Some((req, body)));
        }
        let n = sock.read(&mut tmp).await?;
        if n == 0 {
            return Ok(None);
        }
        buf.extend_from_slice(&tmp[..n]);
        if buf.len() > 64 * 1024 {
            return Err(crate::StreamError::protocol("rtsp request too large"));
        }
    }
}

/// The request's `Content-Length`, or 0.
fn content_length(req: &RtspRequest) -> usize {
    req.headers
        .iter()
        .find(|(n, _)| n.eq_ignore_ascii_case("content-length"))
        .and_then(|(_, v)| v.parse().ok())
        .unwrap_or(0)
}

fn find_double_crlf(buf: &[u8]) -> Option<usize> {
    buf.windows(4).position(|w| w == b"\r\n\r\n").map(|p| p + 4)
}

/// Map an RTSP URI (`rtsp://host[:port]/app/stream[/trackID][?q]`) to a stream.
fn stream_key_from_uri(uri: &str) -> Option<StreamKey> {
    let rest = uri.strip_prefix("rtsp://")?;
    let path = rest.split_once('/').map(|(_, p)| p)?;
    let path = path.split(['?', ';']).next().unwrap_or(path);
    let mut segs = path.split('/').filter(|s| !s.is_empty());
    let app = segs.next()?;
    let stream = segs.next()?;
    Some(StreamKey::new(app, stream))
}

// ── Response builders ────────────────────────────────────────────────────────

fn options_response(cseq: u32) -> String {
    format!(
        "RTSP/1.0 200 OK\r\nCSeq: {cseq}\r\n\
         Public: OPTIONS, DESCRIBE, SETUP, PLAY, TEARDOWN\r\n\r\n"
    )
}

fn build_sdp(codec: CodecId) -> String {
    // Minimal elementary description; the player learns parameter sets in-band.
    let enc = if codec == CodecId::H265 {
        "H265"
    } else {
        "H264"
    };
    format!(
        "v=0\r\n\
         o=- 0 0 IN IP4 0.0.0.0\r\n\
         s=arcly-stream\r\n\
         t=0 0\r\n\
         m=video 0 RTP/AVP 96\r\n\
         a=rtpmap:96 {enc}/90000\r\n\
         a=control:streamid=0\r\n"
    )
}

/// The stream's video codec, read from its cached CONFIG frame — synchronous, no
/// subscription and no wait (the bus caches the config via `ArcSwap`). Defaults
/// to H.264 when no video config has been cached yet.
fn stream_video_codec(handle: &crate::bus::StreamHandle) -> CodecId {
    handle
        .cached_configs()
        .0
        .map(|f| f.codec)
        .unwrap_or(CodecId::H264)
}

/// A packetizer for `codec` with the RTSP server's SSRC and MTU.
fn packetizer_for(codec: CodecId) -> RtpPacketizer {
    if codec == CodecId::H265 {
        RtpPacketizer::new_h265(PT_H264, 0x5254_5350, 1400)
    } else {
        RtpPacketizer::new(PT_H264, 0x5254_5350, 1400)
    }
}

fn describe_response(cseq: u32, uri: &str, sdp: &str) -> String {
    format!(
        "RTSP/1.0 200 OK\r\nCSeq: {cseq}\r\n\
         Content-Base: {uri}\r\nContent-Type: application/sdp\r\n\
         Content-Length: {}\r\n\r\n{sdp}",
        sdp.len()
    )
}

fn setup_response(cseq: u32, session: &str, client_transport: &str) -> String {
    // Echo the client's interleaved channel pair (default 0-1) and preserve
    // `mode=record` for publishers, so the negotiated transport matches exactly.
    let interleaved = client_transport
        .split(';')
        .find_map(|p| p.trim().strip_prefix("interleaved="))
        .unwrap_or("0-1");
    let mode =
        if client_transport.contains("mode=record") || client_transport.contains("mode=RECORD") {
            ";mode=record"
        } else {
            ""
        };
    format!(
        "RTSP/1.0 200 OK\r\nCSeq: {cseq}\r\n\
         Transport: RTP/AVP/TCP;unicast;interleaved={interleaved}{mode}\r\n\
         Session: {session}\r\n\r\n"
    )
}

#[allow(clippy::too_many_arguments)]
fn setup_response_udp(
    cseq: u32,
    session: &str,
    client_rtp: u16,
    client_rtcp: u16,
    server_rtp: u16,
    server_rtcp: u16,
    record: bool,
) -> String {
    let mode = if record { ";mode=record" } else { "" };
    format!(
        "RTSP/1.0 200 OK\r\nCSeq: {cseq}\r\n\
         Transport: RTP/AVP;unicast;client_port={client_rtp}-{client_rtcp};\
         server_port={server_rtp}-{server_rtcp}{mode}\r\n\
         Session: {session}\r\n\r\n"
    )
}

fn unsupported_transport(cseq: u32) -> String {
    format!("RTSP/1.0 461 Unsupported Transport\r\nCSeq: {cseq}\r\n\r\n")
}

fn play_response(cseq: u32, session: &str) -> String {
    format!("RTSP/1.0 200 OK\r\nCSeq: {cseq}\r\nSession: {session}\r\nRange: npt=0.000-\r\n\r\n")
}

fn simple_ok(cseq: u32, session: &str) -> String {
    format!("RTSP/1.0 200 OK\r\nCSeq: {cseq}\r\nSession: {session}\r\n\r\n")
}

fn not_implemented(cseq: u32) -> String {
    format!("RTSP/1.0 501 Not Implemented\r\nCSeq: {cseq}\r\n\r\n")
}

fn not_found(cseq: u32) -> String {
    format!("RTSP/1.0 404 Not Found\r\nCSeq: {cseq}\r\n\r\n")
}

fn unauthorized(cseq: u32) -> String {
    format!("RTSP/1.0 401 Unauthorized\r\nCSeq: {cseq}\r\n\r\n")
}

/// A fresh, hard-to-guess RTSP session id (hex), unique per connection. Seeded
/// from the OS-backed `RandomState`, so no RNG dependency is pulled in.
fn new_session_id() -> String {
    use std::collections::hash_map::RandomState;
    use std::hash::{BuildHasher, Hasher};
    use std::time::{SystemTime, UNIX_EPOCH};
    let mut h = RandomState::new().build_hasher();
    h.write_u128(
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0),
    );
    format!("{:016X}", h.finish())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::protocol::rtsp::message::{InterleavedFrame, RtspResponse};

    #[test]
    fn interleaved_frame_round_trips() {
        let rtp = [0x80u8, 0xE0, 0x00, 0x01, 0xAA, 0xBB];
        let mut framed = Vec::new();
        frame_interleaved(0, &rtp, &mut framed);
        assert_eq!(framed[0], b'$');
        let (f, used) = InterleavedFrame::parse(&framed).expect("parse");
        assert_eq!(used, framed.len());
        assert_eq!(f.channel, 0);
        assert_eq!(f.payload, &rtp);
    }

    #[test]
    fn stream_key_parsed_from_uri_variants() {
        let k = stream_key_from_uri("rtsp://host:554/live/cam").unwrap();
        assert_eq!((k.app.as_str(), k.stream_id.as_str()), ("live", "cam"));
        // trailing track/control + query are stripped.
        let k = stream_key_from_uri("rtsp://h/live/cam/streamid=0?x=1").unwrap();
        assert_eq!((k.app.as_str(), k.stream_id.as_str()), ("live", "cam"));
        assert!(stream_key_from_uri("rtsp://host/onlyapp").is_none());
    }

    #[test]
    fn uri_token_extracted() {
        assert_eq!(
            uri_token("rtsp://h/live/cam?token=abc").as_deref(),
            Some("abc")
        );
        assert_eq!(uri_token("rtsp://h/live/cam").as_deref(), None);
    }

    #[test]
    fn announce_request_parses_with_body_len() {
        let req = RtspRequest::parse(
            "ANNOUNCE rtsp://h/live/cam RTSP/1.0\r\nCSeq: 2\r\nContent-Length: 5\r\n",
        )
        .unwrap();
        assert_eq!(req.method, "ANNOUNCE");
        assert_eq!(content_length(&req), 5);
        assert_eq!(
            stream_key_from_uri(&req.uri).map(|k| (k.app.to_string(), k.stream_id.to_string())),
            Some(("live".into(), "cam".into()))
        );
    }

    #[test]
    fn responses_are_well_formed() {
        assert!(options_response(2).contains("Public: OPTIONS"));
        let d = describe_response(3, "rtsp://h/live/cam", &build_sdp(CodecId::H264));
        assert!(build_sdp(CodecId::H265).contains("H265/90000"));
        let parsed = RtspResponse::parse(
            d.split("\r\n\r\n").next().unwrap(),
            d.split("\r\n\r\n").nth(1).unwrap_or("").to_string(),
        )
        .expect("response parses");
        assert_eq!(parsed.header("Content-Type"), Some("application/sdp"));
        let setup = setup_response(4, "DEADBEEF", "RTP/AVP/TCP;unicast;interleaved=0-1");
        assert!(setup.contains("interleaved=0-1"));
        assert!(setup.contains("Session: DEADBEEF"));
        // A record-mode client gets mode=record echoed back.
        let rec = setup_response(5, "S", "RTP/AVP/TCP;unicast;interleaved=0-1;mode=record");
        assert!(rec.contains("mode=record"));
        // UDP-only request is refused with 461 (so clients retry over TCP).
        assert!(unsupported_transport(6).starts_with("RTSP/1.0 461"));
        assert!(not_found(7).starts_with("RTSP/1.0 404"));
    }
}