//! # str0m poll loop (spec §3.2, §3.4)
//!
//! The heart of the media core. Drives the `str0m::Rtc` instance forward
//! on each call: drains `poll_output()` until `Output::Timeout`, handling
//! each `Output::Transmit` (send on our UDP socket) and `Output::Event`
//! (inbound `MediaData` → Opus decode → sink; inbound RTP count for the
//! idle timeout). When the drain returns `Timeout`, the caller sleeps
//! that duration and calls back with `Input::Timeout`.
//!
//! # Why this lives in a separate module
//!
//! `run_poll_once` takes `&mut RtcSession` — a single function with
//! the full poll logic would make `RtcSession::run_poll_once` 100+ lines
//! of non-trivial control flow. Splitting the loop into a module makes
//! the sans-IO pattern obvious: the loop driver takes a `&mut RtcSession`,
//! reads str0m outputs, and writes str0m inputs. Nothing else.
//!
//! # DEV-DEVIATION
//!
//! Slice 1 runs the poll on a tokio task. ARCHITECTURE.md mandates a
//! dedicated timing thread; we defer that to step 4 (barge-in) because
//! slice 1 has no reflex to time against. The poll function's shape
//! (single `&mut self`, no I/O inside) makes the step-4 swap localized.

use std::io::ErrorKind;
use std::time::{Duration, Instant};

use str0m::net::Protocol;
use str0m::{Input, Output};

use crate::rtc_session::{IDLE_TIMEOUT, RtcSession};

/// 20 ms tick for outbound encoding (matches the PCM frame size, spec §3.9:
/// 480 samples @ 24 kHz = 20 ms). On each tick, we pull one frame from the
/// source pipe and write the encoded Opus via str0m's `Writer::write`.
const OUTBOUND_TICK: Duration = Duration::from_millis(20);

/// One iteration of the str0m poll loop.
///
/// 1. Read any pending UDP packets (non-blocking) and feed each to str0m
///    as `Input::Receive`. A WouldBlock means no packets this cycle — fine.
/// 2. Drain `poll_output()` until `Timeout`:
///    - `Transmit` → send on our UDP socket.
///    - `Event::MediaData` → decode Opus → push to the echo pipe (sink).
///    - `Event::IceConnectionStateChange` → state transition + tracing.
///    - We don't break out of the drain on any of these: str0m's contract
///      is mutate → drain to `Timeout` → mutate (see str0m 0.21 lib docs).
/// 3. **Outbound encode tick:** if ≥20 ms of wallclock passed since the
///    last outbound frame, pull one `PcmFrame` from the source, encode to
///    Opus, and write via `Rtc::writer(mid)->Writer::write`. Then re-drain
///    `poll_output` (the Writer write is a mutation → must drain per str0m).
/// 4. Check the idle timeout: if `now - last_rx > IDLE_TIMEOUT`, transition
///    to `Closed`.
/// 5. Return the `Duration` to the next `Timeout`.
pub fn drive(session: &mut RtcSession, now: Instant) -> Option<Duration> {
    // === Step 1: drain our UDP socket non-blocking, feed str0m. ===
    let mut buf = [0u8; 2000];
    loop {
        match session.socket.recv_from(&mut buf) {
            Ok((n, source)) => {
                // `Receive::new` parses the raw datagram into one of
                // STUN/DTLS/RTP/RTCP (str0m's demultiplexer). It returns
                // `Err(NetError)` for things str0m can't classify; we drop
                // those, don't crash (hot-path policy, spec §3.8).
                let recv = match str0m::net::Receive::new(
                    Protocol::Udp,
                    source,
                    session.local_addr,
                    &buf[..n],
                ) {
                    Ok(r) => r,
                    Err(e) => {
                        tracing::warn!(error = ?e, "str0m datagram parse failed; dropping");
                        continue;
                    }
                };
                // str0m's `Input::Receive` carries the receive timestamp as
                // the first tuple element (str0m 0.21 API — the brief's
                // sketch had `Input::Receive(recv)` and `handle_input(now, ...)`,
                // both adjusted to the actual surface: a single `Input`
                // argument, with the Instant packed into the variant).
                if session.rtc.handle_input(Input::Receive(now, recv)).is_err() {
                    // Hot-path policy: drop + observe, don't crash.
                    tracing::warn!("str0m rejected input packet; dropping");
                }
                session.last_rx = now;
            }
            // WouldBlock (unix) / TimedOut (windows) — no packets this cycle.
            Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => break,
            Err(e) => {
                tracing::warn!(error = ?e, "UDP recv_from error; continuing");
                break;
            }
        }
    }

    // === Step 2: drain poll_output, interleaving outbound writes. ===
    // `next_timeout` is set in either the `Timeout` or `Err` exit arms of
    // the drain loop below before any read on the path forward — no
    // initial value needed. Both exit arms assign before breaking, so the
    // borrow checker is satisfied; clippy flagged the previous `= None`
    // initializer as a write-then-overwrite.
    let mut next_timeout: Option<Instant>;
    // Track whether we owe a Writer write this cycle; re-drain if so.
    // str0m's "mutate → drain to Timeout" invariant: after Writer::write,
    // poll_output must be drained to Timeout before any other mutation.
    let mut needs_redrain = false;
    loop {
        match session.rtc.poll_output() {
            Ok(Output::Timeout(t)) => {
                next_timeout = Some(t);
                if needs_redrain {
                    // We did an outbound write in the previous iteration;
                    // str0m needs to be drained again. Loop continues,
                    // but only handle Transmit/Event briefly before next Timeout.
                    needs_redrain = false;
                    continue;
                }
                break; // engine is fully drained
            }
            Ok(Output::Transmit(t)) => {
                // `Transmit.contents` is a `DatagramSend` newtype that
                // `Deref`s to `[u8]` — passing `&t.contents` deref-coerces
                // to `&[u8]` for `send_to`.
                if let Err(e) = session.socket.send_to(&t.contents, t.destination) {
                    if !matches!(e.kind(), ErrorKind::WouldBlock) {
                        tracing::warn!(error = ?e, "UDP send_to error; dropping");
                    }
                }
            }
            Ok(Output::Event(event)) => {
                handle_event(session, event, now);
                // Loop continues — mutations from inside the drain loop
                // are fine (str0m docs, "single-mutation invariant"):
                // events are observations, not external mutations.
            }
            Err(e) => {
                tracing::warn!(error = ?e, "str0m poll_output error; continuing");
                next_timeout = Some(now + OUTBOUND_TICK);
                break;
            }
        }
    }

    // === Step 3: outbound encode tick (the echo path). ===
    // If wallclock has crossed a 20 ms boundary since the last outbound
    // frame, pull a PcmFrame from the source, encode to Opus, and write
    // via Writer::write. This IS the slice-1 echo: inbound decode → pipe
    // → outbound encode.
    if now.duration_since(session.last_outbound_at) >= OUTBOUND_TICK {
        if let Some(mid) = session.audio_mid {
            if let Some(frame) = session.pipe.next_pcm_frame() {
                if let Some(opus_payload) = session.encoder.encode(&frame) {
                    // Writer::write signature (str0m 0.21, verified):
                    //   write(pt: Pt, wallclock: Instant, rtp_time: MediaTime,
                    //         data: impl Into<Arc<[u8]>>) -> Result<(), RtcError>
                    // - pt: payload type for Opus. `writer.payload_params()`
                    //   returns `impl Iterator<Item = &PayloadParams>`; the
                    //   first one's `.pt()` is our Opus PT (str0m negotiates
                    //   this in the SDP answer).
                    // - wallclock: when the sample was produced — local `now`.
                    // - rtp_time: RTP timestamp in the 48 kHz audio clock for
                    //   Opus. Increment by 960 per 20 ms (48000 * 0.020).
                    //   `MediaTime` has no `add(Duration)` method — use
                    //   `mt + MediaTime::from(duration)`.
                    //
                    // `rtc.writer(mid)` returns `Option<Writer<'_>>` — `None`
                    // if direction isn't sending (we'd be in a recvonly state).
                    if let Some(writer) = session.rtc.writer(mid) {
                        // Pull the Opus payload type out of the iterator
                        // BEFORE `writer.write(...)`, which consumes `writer`
                        // by value. Doing both inline trips E0505 because
                        // `payload_params()` borrows `writer` while
                        // `write(self, ...)` moves it — so separate the
                        // borrow from the move with a tight scope.
                        let pt = writer.payload_params().next().map(|p| p.pt());
                        if let Some(pt) = pt {
                            let rtp_time = session.next_media_time;
                            if writer
                                .write(pt, now, rtp_time, opus_payload.as_slice())
                                .is_ok()
                            {
                                // Advance media time for next 20 ms frame.
                                // `MediaTime + MediaTime::from(Duration)` —
                                // no `add()` method on `MediaTime` in str0m 0.21.
                                session.next_media_time +=
                                    str0m::media::MediaTime::from(Duration::from_millis(20));
                                needs_redrain = true;
                            }
                        }
                    }
                }
            }
            session.last_outbound_at = now;
        }
    }

    // If the outbound write happened, we owe str0m one more drain before
    // returning — Writer::write is a mutation per str0m's invariant.
    if needs_redrain {
        loop {
            match session.rtc.poll_output() {
                Ok(Output::Timeout(t)) => {
                    next_timeout = Some(t);
                    break;
                }
                Ok(Output::Transmit(t)) => {
                    let _ = session.socket.send_to(&t.contents, t.destination);
                }
                Ok(Output::Event(e)) => handle_event(session, e, now),
                Err(_) => break,
            }
        }
    }

    // === Step 4: idle timeout (spec §4.5). ===
    if now.duration_since(session.last_rx) > IDLE_TIMEOUT {
        tracing::info!(
            channel_id = %session.channel.id,
            "idle timeout (60 s no RX); closing session"
        );
        session.channel.state = rutster_call_model::ChannelState::Closed;
        return None;
    }

    session.next_timeout = next_timeout;
    next_timeout.map(|t| t.saturating_duration_since(now))
}

/// Dispatch a str0m `Event` to the audio pipe or to state bookkeeping.
fn handle_event(session: &mut RtcSession, event: str0m::Event, _now: Instant) {
    use str0m::Event;
    match event {
        Event::MediaData(media) => {
            // Inbound decoded audio frame from the peer (Frame API, spec §3.2).
            // str0m has already done RTP depacketization; `MediaData.data`
            // is the encoded Opus payload (type: `Arc<[u8]>` — passing
            // `&media.data` deref-coerces to `&[u8]` for `OpusDecoder::decode`).
            if let Some(pcm) = session.decoder.decode(&media.data) {
                session.pipe.on_pcm_frame(pcm);
            }
            // Decode failed → drop + observe (per §3.8). Don't kill the peer.
        }
        Event::IceConnectionStateChange(state) => {
            tracing::info!(
                channel_id = %session.channel.id,
                ?state,
                "ICE state change"
            );
            if state == str0m::IceConnectionState::Connected {
                session.channel.state = rutster_call_model::ChannelState::Connected;
            }
        }
        Event::EgressBitrateEstimate(_) => { /* BWE — irrelevant in slice 1 */ }
        _ => { /* str0m emits several other event variants we don't need in slice 1. */ }
    }
}
