//! # `RtcSession` — the per-peer media owner (spec §3.1, §4.5)
//!
//! Owns a `str0m::Rtc` instance + an Opus decoder/encoder pair + an
//! `EchoAudioPipe` wiring inbound to outbound + the per-peer UDP socket.
//! One per WebRTC peer. The `ChannelId` (from `rutster-call-model`) is
//! the session id surfaced in the REST API.
//!
//! ## What str0m does for us (so we don't)
//!
//! str0m 0.21's `Rtc::sdp_api().accept_offer(offer)` produces the SDP
//! answer natively: DTLS fingerprint (from the cert str0m generates), ICE
//! ufrag/pwd, and codec negotiation (Opus, the only codec we registered).
//! Slice 1 does NOT hand-roll an SDP munger — str0m handles WebRTC
//! offer/answer natively. There is no first-party SIP/SDP path: under
//! [ADR-0007](../../../docs/adr/0007-trunk-rented-transport.md) the trunk is
//! rented (a CPaaS media-leg fork — raw audio, no SDP) or an out-of-tree SBC
//! (which owns any carrier SIP/SDP, outside the trust boundary). rutster's
//! WebRTC-ICE-coupled SDP lives entirely in str0m.

use std::net::{IpAddr, SocketAddr, UdpSocket};
use std::time::{Duration, Instant};

use rutster_call_model::{Channel, ChannelId, ChannelState};
use str0m::Rtc;
use thiserror::Error;

use crate::opus_codec::{OpusDecoder, OpusEncoder};
use crate::pcm::{AudioPipe, EchoAudioPipe};

/// Per-session idle timeout (spec §4.5): 60 s of no RTP from the peer
/// → close. RTC quiet periods are normal but 60 s of dead air means
/// "the browser tab is dead."
pub(crate) const IDLE_TIMEOUT: Duration = Duration::from_secs(60);

/// Cold-path errors for `RtcSession` construction and SDP negotiation.
///
/// Hot-path failures (decode, encode, UDP recv) follow the
/// match-and-continue "drop + observe" policy from spec §3.8 — they
/// never reach this enum.
#[derive(Debug, Error)]
pub enum RtcSessionError {
    /// Two-stage failure from str0m's SDP path: `SdpOffer::from_sdp_string`
    /// can fail to parse, OR `accept_offer` can reject the parsed offer.
    /// Both surface as `str0m::RtcError` / `str0m::sdp::SdpError`; we
    /// collapse them via a display-format `String` since both are
    /// display-format-only at the axum boundary (HTTP 400 in `routes.rs`).
    /// Using `String` (not `#[source]`) keeps the variants uniform and the
    /// axum layer doesn't need to downcast — it logs + 400s.
    #[error("SDP offer parse or accept failed: {0}")]
    SdpOffer(String),
    #[error("opus codec init failed: {0}")]
    Codec(#[from] opus::Error),
    #[error("UDP socket bind failed: {0}")]
    Socket(#[from] std::io::Error),
}

use str0m::change::SdpOffer;
use str0m::media::Mid;

/// Where the media socket binds vs. what address peers are told to reach.
///
/// # Why bind and advertised are SEPARATE concepts (slice-5, review B1)
///
/// Behind cloud NAT / a 1:1 elastic IP, the address a peer must send RTP
/// to is NOT the local bind address — and RTP cannot ride the HTTP load
/// balancer, so every instance must advertise its own reachable address.
/// str0m also rejects the unspecified address (`0.0.0.0`) as a candidate,
/// which forces the split at the type level: you may bind wildcard, but
/// you must then say what to advertise.
///
/// `port_range` bounds the UDP ports (inclusive) so a fleet's security
/// groups can be written; `None` keeps the OS-ephemeral behavior.
#[derive(Debug, Clone)]
pub struct MediaAddressConfig {
    pub bind_ip: IpAddr,
    pub advertised_ip: Option<IpAddr>,
    pub port_range: Option<(u16, u16)>,
}

impl Default for MediaAddressConfig {
    /// Loopback + ephemeral — byte-for-byte the pre-slice-5 behavior, so
    /// every existing test and the dev loop run unchanged.
    fn default() -> Self {
        Self {
            bind_ip: IpAddr::from([127, 0, 0, 1]),
            advertised_ip: None,
            port_range: None,
        }
    }
}

/// Bind the first free UDP port in `[lo, hi]` (inclusive).
///
/// Sequential scan — O(range) worst case, cold path only (session
/// construction). The OS is the allocator: a failed bind means "taken,"
/// so no shared allocator state is needed across sessions or threads.
fn bind_in_range(ip: IpAddr, lo: u16, hi: u16) -> std::io::Result<UdpSocket> {
    let mut last_err = None;
    for port in lo..=hi {
        match UdpSocket::bind(SocketAddr::new(ip, port)) {
            Ok(s) => return Ok(s),
            Err(e) => last_err = Some(e),
        }
    }
    Err(last_err
        .unwrap_or_else(|| std::io::Error::new(std::io::ErrorKind::AddrInUse, "empty port range")))
}

/// The per-peer media owner (spec §3.1, §4.5).
///
/// # Ownership / sharing
///
/// An `RtcSession` lives behind an `Arc<Mutex<RtcSession>>` in the
/// binary's `DashMap<ChannelId, RtcSession>` (Task 5). The mutex is
/// short-held: each tokio poll iteration locks, runs `run_poll_once`,
/// unlocks. We do NOT hold the lock across `tokio::time::sleep` — that
/// would defeat the `DashMap`'s sharded concurrency and pre-pave the
/// wrong pattern for step 4's dedicated thread.
///
/// # Why `Arc<Mutex<...>>` (not `Arc<RwLock<...>>`)
///
/// Every access of an `RtcSession` mutates it (str0m's `&mut self`
/// contract on `handle_input` + `poll_output`). `RwLock`'s read-mode
/// would be useless because str0m takes `&mut Rtc`. `Mutex` it is.
pub struct RtcSession {
    pub channel: Channel,
    pub(crate) rtc: Rtc,
    pub(crate) decoder: OpusDecoder,
    pub(crate) encoder: OpusEncoder,
    pub(crate) pipe: Box<dyn AudioPipe + Send>,
    /// Local UDP socket str0m sends `Transmit` packets out on and
    /// receives `Input::Receive` packets from. Bound to an ephemeral
    /// port at construction. The SDP candidate passed to the peer uses
    /// the separate `advertised_addr` field, not this socket's address.
    pub(crate) socket: std::net::UdpSocket,
    /// Local socket address — cached because `local_addr()` is a syscall.
    pub(crate) local_addr: SocketAddr,
    /// The address written into our ICE host candidate — advertised IP
    /// (if configured) + the actually-bound port. `local_addr` stays the
    /// real socket address (loop_driver's `Receive::new` needs it);
    /// `advertised_addr` is what the SDP answer tells the peer.
    pub(crate) advertised_addr: SocketAddr,
    /// Mid of the audio m-line we accepted. Set during `accept_offer`.
    /// Slice 1 has exactly one m-line; multi-m-line arrives with video.
    pub(crate) audio_mid: Option<Mid>,
    /// Last deadline from `Rtc::poll_output` — the next time the loop
    /// should wake the rtc with `Input::Timeout`.
    pub(crate) next_timeout: Option<Instant>,
    /// Last Instant we received an RTP packet from the peer. Used for
    /// the 60 s idle timeout (spec §4.5).
    pub(crate) last_rx: Instant,
    /// Last Instant we wrote an outbound Opus frame. Used to pace the
    /// 20 ms encode tick for the echo path (slice-1 read of spec §3.2).
    pub(crate) last_outbound_at: Instant,
    /// Outbound RTP media-time clock. For Opus audio on the wire str0m's
    /// negotiated clock is 48 kHz; a 20 ms frame advances the RTP
    /// timestamp by 48 000 × 0.020 = 960 ticks. We increment the
    /// `MediaTime` by `MediaTime::from(Duration)` per frame, since
    /// `MediaTime` has no `add(Duration)` method on str0m 0.21 — the
    /// `From<Duration>` impl interprets the duration against the
    /// underlying clock frequency.
    pub(crate) next_media_time: str0m::media::MediaTime,
}

impl RtcSession {
    /// Construct a new session — used by both the binary's `AppState`
    /// (production) and the tests. Single constructor — no `for_test` /
    /// `for_server` split; the body is identical (binding a UDP socket
    /// on `0.0.0.0:0`, constructing the `Rtc` + codecs).
    pub fn new() -> Result<Self, RtcSessionError> {
        Self::new_with_config(&MediaAddressConfig::default())
    }

    /// Construct with explicit addressing (slice-5, review B1). The
    /// binary threads `MediaThreadOpts.media_cfg` here; `new()` keeps the
    /// loopback default for tests and the dev loop.
    pub fn new_with_config(cfg: &MediaAddressConfig) -> Result<Self, RtcSessionError> {
        // Fail-fast: a wildcard bind with nothing to advertise would die
        // later inside accept_offer (str0m rejects 0.0.0.0 candidates).
        // Cold path → a real error, not an expect().
        if cfg.bind_ip.is_unspecified() && cfg.advertised_ip.is_none() {
            return Err(RtcSessionError::Socket(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "RUTSTER_MEDIA_BIND_IP is unspecified (0.0.0.0/::) — set RUTSTER_MEDIA_ADVERTISED_IP",
            )));
        }
        let socket = match cfg.port_range {
            Some((lo, hi)) => bind_in_range(cfg.bind_ip, lo, hi)?,
            None => UdpSocket::bind(SocketAddr::new(cfg.bind_ip, 0))?,
        };
        socket.set_nonblocking(true)?;
        let local_addr = socket.local_addr()?;
        let advertised_addr =
            SocketAddr::new(cfg.advertised_ip.unwrap_or(cfg.bind_ip), local_addr.port());

        let rtc = Rtc::new(Instant::now());

        Ok(Self {
            channel: Channel::new_inbound(),
            rtc,
            decoder: OpusDecoder::new()?,
            encoder: OpusEncoder::new()?,
            pipe: Box::new(EchoAudioPipe::new()),
            socket,
            local_addr,
            advertised_addr,
            audio_mid: None,
            next_timeout: None,
            last_rx: Instant::now(),
            last_outbound_at: Instant::now(),
            next_media_time: str0m::media::MediaTime::ZERO,
        })
    }

    /// Replace the default `EchoAudioPipe` with a real tap pipe (slice-2,
    /// spec §8.5 #6 — the seam test). The binary wires `TapAudioPipe`
    /// here on the `Connected` transition (spec §5.1 step 3); slice-1's
    /// unit tests + the default echo path leave `EchoAudioPipe` in place.
    ///
    /// # Why a generic `P: AudioPipe + Send + 'static`
    ///
    /// The `'static` bound is required to convert `P` into `Box<dyn
    /// Trait + Send + 'static>`; without it the trait object would
    /// carry a lifetime parameter that outlives nothing in particular,
    /// failing to coerce. The `Send` bound is required because
    /// `RtcSession` lives behind `Arc<Mutex<RtcSession>>` (sent between
    /// tokio tasks); the field type therefore must be `Send`.
    ///
    /// # Why `self`-consuming (returns `Self`)
    ///
    /// Builder-style chaining: `RtcSession::new()?.with_pipe(pipe)` reads
    /// left-to-right; an alternative `&mut self -> Result<()>` would split
    /// construction from wiring and force `mut` on every caller.
    pub fn with_pipe<P: AudioPipe + Send + 'static>(mut self, pipe: P) -> Self {
        self.set_pipe(pipe);
        self
    }

    /// In-place variant of `with_pipe` for swapping the pipe on an
    /// existing `RtcSession` (slice-2 spec §5.1 step 3 — the `Connected`
    /// transition swaps `EchoAudioPipe` for `TapAudioPipe` after the
    /// engine task is spawned). Used by the binary's `session_map`
    /// poll-task layer; `loop_driver.rs` is unaware of the swap (the
    /// seam test preserves its byte-identical call sites).
    pub fn set_pipe<P: AudioPipe + Send + 'static>(&mut self, pipe: P) {
        self.pipe = Box::new(pipe);
    }

    /// Flush the pipe's playout ring (slice-2 spec §5.3 step 4).
    ///
    /// Called by the binary's poll task when the engine task signals a
    /// flush via the side-channel `flush_tx` mpsc — i.e. on every brain
    /// disconnect / WS error. Default no-op for pipes without a buffer
    /// (slice-1's `EchoAudioPipe`); `TapAudioPipe` overrides to clear its
    /// `VecDeque`. Defers to the trait object so the binary doesn't need
    /// to know the concrete pipe type at the call site.
    pub fn clear_playout_ring(&mut self) {
        self.pipe.clear_playout_ring();
    }

    /// Returns the session's `ChannelId` — also the REST API session id
    /// surfaced at `/v1/sessions/{id}` (spec §4.5).
    pub fn channel_id(&self) -> ChannelId {
        self.channel.id
    }

    /// Returns the current signaling state (`ChannelState` from
    /// `rutster-call-model`): `New` → `Connecting` → `Connected` →
    /// `Closing` → `Closed`.
    pub fn channel_state(&self) -> ChannelState {
        self.channel.state
    }

    /// True once the channel has reached `ChannelState::Closed` — str0m
    /// resources are dropped; the loop driver returns `None` and the
    /// tokio task tears down.
    pub fn is_closed(&self) -> bool {
        matches!(self.channel.state, ChannelState::Closed)
    }

    /// Accept a browser SDP offer; return the SDP answer (spec §4.1).
    ///
    /// str0m 0.21's `sdp_api().accept_offer()` does the heavy lifting:
    /// parses the offer, picks compatible codecs (Opus, the only one we
    /// register by default), generates the DTLS fingerprint from its
    /// self-signed cert, and produces ICE ufrag/pwd. We add our local
    /// host candidate (the UDP socket we just bound) *before* calling
    /// `accept_offer` so the answer carries it.
    pub fn accept_offer(&mut self, offer_sdp: &str) -> Result<String, RtcSessionError> {
        assert!(self.audio_mid.is_none(), "accept_offer called twice");

        // Register our local UDP socket as a host candidate. str0m includes
        // this candidate's address + the ICE creds it generates in the SDP
        // answer. `add_local_candidate` returns `Option<&Candidate>` —
        // `None` means str0m rejected it (log + continue; not fatal).
        let candidate = str0m::Candidate::host(self.advertised_addr, "udp")
            .expect("host candidate from validated advertised address");
        // ^-- expect stays acceptable: new_with_config rejected the only
        // input (unspecified IP) that str0m refuses, so this is const-ish.
        if self.rtc.add_local_candidate(candidate).is_none() {
            tracing::warn!(channel_id = %self.channel.id, "str0m rejected local candidate");
        }

        // str0m's SDP API parses + accepts the offer natively. There is NO
        // `from_str_unchecked` — `SdpOffer::from_sdp_string` returns
        // `Result` and is the canonical entry point. `accept_offer` takes
        // the owned `SdpOffer` and returns the `SdpAnswer`.
        let parsed_offer = SdpOffer::from_sdp_string(offer_sdp)
            .map_err(|e| RtcSessionError::SdpOffer(format!("parse: {e}")))?;
        let answer = self
            .rtc
            .sdp_api()
            .accept_offer(parsed_offer)
            .map_err(|e| RtcSessionError::SdpOffer(format!("accept: {e}")))?;

        // The first audio mid we accepted. Used to get the Writer for
        // outbound Opus frames in `run_poll_once`. A single audio m-line
        // is slice 1's whole world; multi-m-line arrives with video.
        //
        // `SdpAnswer` has no `mid()` accessor in str0m 0.21 (the brief's
        // sketch assumed it). It derefs to `Sdp`, whose `media_lines`
        // is `Vec<MediaLine>`; each `MediaLine::mid()` returns the mid.
        // Slice 1's answer has exactly one m-line; we read its mid.
        self.audio_mid = answer.media_lines.first().map(|m| m.mid());

        self.channel.state = ChannelState::Connecting;
        Ok(answer.to_sdp_string())
    }

    /// Drive one iteration of the sans-IO poll loop (spec §3.2, §3.4).
    ///
    /// Returns the `Duration` until the next `Input::Timeout` should be
    /// fed back to str0m, or `None` if the peer is closed. The caller
    /// (Task 5's tokio task) sleeps this duration then calls again.
    ///
    /// DEV-DEVIATION: tokio polling accepted for slice 1; step 4
    /// replaces with dedicated timing thread per ARCHITECTURE.md.
    pub fn run_poll_once(&mut self, now: Instant) -> Option<Duration> {
        if self.is_closed() {
            return None;
        }
        crate::loop_driver::drive(self, now)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A captured Chrome SDP offer for an audio-only Opus m-line. Real
    /// browser-style offer with host ICE candidates — the simplest valid
    /// offer str0m 0.21 will accept and produce an answer for. Slice 1 has
    /// no video m-line; multi-m-line arrives with the escalation rung.
    ///
    /// NOTE: the brief's fixture omitted the `a=group:BUNDLE 0` line
    /// (str0m 0.21 rejects SDP without a session-level group attribute as
    /// inconsistent — `sdp::Sdp::assert_consistency`). Real Chrome offers
    /// always include BUNDLE; restored here so the offer str0m receives
    /// matches what a browser actually sends.
    const BROWSER_SDP_OFFER: &str = "\
v=0\r
o=- 4593482934 2 IN IP4 127.0.0.1\r
s=-\r
t=0 0\r
a=group:BUNDLE 0\r
m=audio 9 UDP/TLS/RTP/SAVPF 111\r
c=IN IP4 0.0.0.0\r
a=rtcp:9 IN IP4 0.0.0.0\r
a=ice-ufrag:abcd\r
a=ice-pwd:abcdefghijklmnopqrstuvwxyz0123456789\r
a=fingerprint:sha-256 AB:CD:EF:00:11:22:33:44:55:66:77:88:99:AA:BB:CC:DD:EE:FF:00:11:22:33:44:55:66:77:88:99:AA:BB:CC:DD\r
a=setup:actpass\r
a=mid:0\r
a=sendrecv\r
a=rtpmap:111 opus/48000/2\r
a=fmtp:111 minptime=10;useinbandfec=1\r
a=candidate:1 1 UDP 2113667327 192.168.1.42 50000 typ host\r
";

    #[test]
    fn accept_offer_returns_sdp_answer_with_opus() {
        let mut session = RtcSession::new().expect("session");
        let answer = session.accept_offer(BROWSER_SDP_OFFER).expect("SDP answer");
        // Answer contains an audio m-line, an Opus payload, a fingerprint,
        // and ICE credentials (str0m fills these natively in 0.21).
        assert!(answer.contains("m=audio"), "answer has an audio m-line");
        assert!(answer.contains("opus/48000"), "answer advertises Opus");
        assert!(
            answer.contains("a=fingerprint:sha-256 "),
            "DTLS fingerprint"
        );
        assert!(answer.contains("a=ice-ufrag:"), "ICE ufrag present");
        assert!(answer.contains("a=ice-pwd:"), "ICE pwd present");
    }

    #[test]
    fn channel_id_matches_session_id() {
        let session = RtcSession::new().expect("session");
        let id = session.channel_id();
        // The ChannelId IS the session id surfaced in the REST API (spec §4.5).
        assert_eq!(format!("{}", id).len(), 36);
    }

    #[test]
    fn accept_offer_transitions_channel_to_connecting() {
        // The spec §5.4 state machine: New → Connecting on offer receive.
        // This test pins the transition callers depend on; the impl sets
        // it at the end of `accept_offer`.
        let mut session = RtcSession::new().expect("session");
        assert_eq!(session.channel_state(), ChannelState::New);
        let _ = session.accept_offer(BROWSER_SDP_OFFER).expect("answer");
        assert_eq!(session.channel_state(), ChannelState::Connecting);
    }

    #[test]
    fn default_config_binds_loopback_ephemeral() {
        // Byte-for-byte behavioral compatibility: new() == new_with_config(default).
        let s = RtcSession::new_with_config(&MediaAddressConfig::default()).expect("session");
        assert!(s.local_addr.ip().is_loopback());
    }

    #[test]
    fn port_range_is_respected_and_exhaustion_errors() {
        let cfg = MediaAddressConfig {
            bind_ip: "127.0.0.1".parse().unwrap(),
            advertised_ip: None,
            port_range: Some((41500, 41501)), // room for exactly two sessions
        };
        let a = RtcSession::new_with_config(&cfg).expect("first");
        let b = RtcSession::new_with_config(&cfg).expect("second");
        assert!((41500..=41501).contains(&a.local_addr.port()));
        assert!((41500..=41501).contains(&b.local_addr.port()));
        // Range full → cold-path Socket error, not a panic.
        assert!(matches!(
            RtcSession::new_with_config(&cfg),
            Err(RtcSessionError::Socket(_))
        ));
    }

    #[test]
    fn sdp_answer_advertises_the_advertised_ip_not_the_bind_ip() {
        let cfg = MediaAddressConfig {
            bind_ip: "127.0.0.1".parse().unwrap(),
            advertised_ip: Some("203.0.113.7".parse().unwrap()), // TEST-NET-3
            port_range: None,
        };
        let mut s = RtcSession::new_with_config(&cfg).expect("session");
        let answer = s.accept_offer(BROWSER_SDP_OFFER).expect("answer");
        assert!(
            answer.contains("203.0.113.7"),
            "candidate must carry advertised IP"
        );
        assert!(
            !answer.contains("127.0.0.1 "),
            "bind IP must not leak into the candidate"
        );
    }

    #[test]
    fn unspecified_bind_without_advertised_ip_is_rejected() {
        // str0m's Candidate::host rejects 0.0.0.0 — fail at construction
        // with a config error, not at offer-accept with an expect().
        let cfg = MediaAddressConfig {
            bind_ip: "0.0.0.0".parse().unwrap(),
            advertised_ip: None,
            port_range: None,
        };
        assert!(RtcSession::new_with_config(&cfg).is_err());
    }
}
