//! # MediaThread — the dedicated 20ms media loop on a std::thread
//! (slice-4 spec §2.2, §4)
//!
//! ARCHITECTURE.md mandates "dedicated timing threads, not the shared
//! tokio pool." slice-1 ran the poll on tokio as an acknowledged
//! deviation; slice-4 graduates it. ONE `std::thread::spawn` at binary
//! startup owns `HashMap<ChannelId, RtcSession>` exclusively; all access
//! from axum is via a command channel. The 10ms meta-tick between loop
//! iterations is `std::thread::sleep(Duration::from_millis(10))`.
//!
//! # Why one thread, not per-session
//!
//! Spearhead scale (see slice-4 spec §6.3). The command-channel seam
//! makes the later threadpool-shard graduation localized.
//!
//! # The seam (loop_driver + rtc_session byte-identical)
//!
//! `MediaThread` calls `RtcSession::run_poll_once(now)` — the unchanged
//! `loop_driver::drive`. The `Reflex<TapAudioPipe>` wrapper is wired in
//! here on the `Connected` transition (via `RtcSession::set_pipe`), not
//! inside `rtc_session.rs`. The seam holds.

use std::collections::HashMap;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use rutster_call_model::ChannelId;
use rutster_media::RtcSession;
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, info};

use crate::tap_engine::spawn_tap_engine;

/// The 10ms meta-tick. Finer than the 20ms outbound encode tick so str0m's
/// `Timeout` outputs are honored promptly.
const META_TICK: Duration = Duration::from_millis(10);

/// Capacity for the command channel from axum to the media thread.
const CMD_CHANNEL_CAPACITY: usize = 64;

/// Knobs the binary resolves from env and hands to the media thread.
/// Manual Default (not derive) because `sink: Arc<dyn EventSink>` has no
/// meaningful derived default — `Arc::new(TracingEventSink)` is a
/// deliberate choice, not a zero value. Defaults reproduce the
/// pre-slice-5 single-node dev behavior exactly.
pub struct MediaThreadOpts {
    pub media_cfg: rutster_media::MediaAddressConfig,
    /// Admission cap (review M2). 64 is a placeholder until the ADR-0010
    /// benchmark produces the real per-node number — the gauge below is
    /// that benchmark's primary readout.
    pub max_sessions: usize,
    /// Lifecycle event seam (Task 7, review M3). Default is the log-backed
    /// `TracingEventSink`; ADR-0005's Valkey publisher swaps in here.
    pub sink: crate::event_sink::SharedEventSink,
}

impl Default for MediaThreadOpts {
    fn default() -> Self {
        Self {
            media_cfg: rutster_media::MediaAddressConfig::default(),
            max_sessions: 64,
            sink: std::sync::Arc::new(crate::event_sink::TracingEventSink),
        }
    }
}

/// Commands axum sends to the media thread (cold-path only — NEVER on
/// the 20ms tick). The thread owns RtcSessions exclusively; this is the
/// ONLY entry point for axum-side mutation.
#[derive(Debug)]
pub enum MediaCmd {
    /// Construct a fresh RtcSession, store it under a new ChannelId, reply.
    /// The thread constructs RtcSession::new_with_config(&opts.media_cfg)
    /// (keeps all RtcSession construction on the thread that owns it).
    Register {
        tap_url: url::Url,
        reply: oneshot::Sender<Result<ChannelId, String>>,
    },
    /// Accept a browser SDP offer on the session's behalf, reply with the
    /// SDP answer (cold-path — the axum POST /v1/sessions/{id}/office handler).
    AcceptOffer {
        id: ChannelId,
        sdp: String,
        reply: oneshot::Sender<Result<String, String>>,
    },
    /// Tear down a session — fires close_tx + bounded-await the engine task
    /// (750ms cap), then removes the entry.
    Delete {
        id: ChannelId,
        reply: oneshot::Sender<()>,
    },
    /// Graceful shutdown — drain + drop + join.
    Shutdown { reply: oneshot::Sender<()> },
    /// Enter drain (review M1): reject new Registers ("draining" → 503),
    /// keep ticking existing sessions, fire `reply` when the map empties.
    /// The deadline lives with the CALLER (main.rs times out and proceeds
    /// to Shutdown) — the thread just reports emptiness; that keeps
    /// "how long is too long" an operator policy, not thread logic.
    Drain { reply: oneshot::Sender<()> },
    /// Cold-path capacity/health snapshot (readyz + the autoscaling
    /// signal; review M2). Cheap: counters the loop already maintains.
    Stats { reply: oneshot::Sender<MediaStats> },
    /// slice-5 T5: register a trunk-side session (PSTN caller via Twilio's
    /// WSS raw-audio fork). The MediaThread constructs a TrunkSession
    /// (T4), wraps its inner TapAudioPipe in `Reflex<TapAudioPipe>` then
    /// `LocalVadReflex<Reflex<TapAudioPipe>>` (same composition as slice-4
    /// Task 6 for WebRTC legs), inserts the leg in the session map under
    /// `MediaLeg::Trunk`, spawns the TapEngine, + replies with the freshly-
    /// allocated ChannelId.
    ///
    /// The WSS pump task (T3) sends this variant via the relay task in
    /// main.rs once Twilio's "start" frame arrives. Credentials never
    /// reach this codepath (ADR-0009 -- the brain holds only the per-call
    /// tap_url, an operator-configured non-credential).
    RegisterTrunk {
        call_sid: String,
        inbound_from_twilio_rx: tokio::sync::mpsc::Receiver<rutster_media::PcmFrame>,
        outbound_to_twilio_tx: tokio::sync::mpsc::Sender<rutster_media::PcmFrame>,
        tap_url: url::Url,
        reply: oneshot::Sender<rutster_call_model::ChannelId>,
    },
}

/// What the node tells the platform about itself. `serde::Serialize`
/// because /readyz returns it verbatim as JSON.
#[derive(Debug, Clone, serde::Serialize)]
pub struct MediaStats {
    pub sessions: usize,
    pub max_sessions: usize,
    pub draining: bool,
    /// Ticks whose work exceeded META_TICK — the saturation signal.
    /// Overload degrades EVERY call at once (missed 20ms deadlines), so
    /// this must trend at ~0; an autoscaler scales out on its slope.
    pub tick_overruns: u64,
    pub last_tick_micros: u64,
    /// Cold-path admission rejects (deploy slice C §5.5): every Register
    /// the media thread shed due to `draining` OR capacity.
    pub admission_rejects: u64,
    /// EventSink drops (deploy slice C §5.6): populated from sink.drops().
    /// None for TracingEventSink; Some(count) when ValkeyEventSink (Task 8)
    /// is wired.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event_sink_drops: Option<u64>,
}

/// The handle returned to the binary. Clone the `cmd_tx` per-session;
/// the `JoinHandle` is dropped on shutdown to detach (the binary's
/// graceful-shutdown signal fires `Shutdown` first).
pub struct MediaThread {
    cmd_tx: mpsc::Sender<MediaCmd>,
    join: Option<JoinHandle<()>>,
}

impl MediaThread {
    /// Return a clone of the command-channel sender, so the binary (axum
    /// routes / `AppState`) can issue cold-path commands to the thread.
    pub fn cmd_tx(&self) -> mpsc::Sender<MediaCmd> {
        self.cmd_tx.clone()
    }

    /// Spawn the dedicated media thread. Captures a `tokio::runtime::Handle`
    /// so the thread can `handle.spawn(spawn_tap_engine(...))` on the
    /// `Connected` transition. The thread owns `HashMap<ChannelId,
    /// ThreadSession>` exclusively.
    ///
    /// Returns `std::io::Error` if the OS cannot create the std::thread
    /// (for example: the process has hit its thread limit). Thread creation
    /// is part of the cold startup path, not the 20ms media loop, so the
    /// error is propagated for the binary to decide whether to abort, retry,
    /// or degrade. Inside the loop we still follow the "drop + observe, don't
    /// crash" policy.
    pub fn spawn(
        default_tap_url: url::Url,
        opts: MediaThreadOpts,
        tokio_handle: tokio::runtime::Handle,
    ) -> Result<Self, std::io::Error> {
        let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHANNEL_CAPACITY);
        // The thread needs its own command-channel sender so it can build each
        // session's `AppState`. Tools spawned inside the tap engine (notably
        // `HangupTool`) call `AppState::close(id)`, which must loop back to
        // this same media thread via `MediaCmd::Delete`.
        let cmd_tx_for_thread = cmd_tx.clone();
        let join = std::thread::Builder::new()
            .name("rutster-media".into())
            .spawn(move || {
                run_media_thread(
                    cmd_rx,
                    default_tap_url,
                    opts,
                    tokio_handle,
                    cmd_tx_for_thread,
                );
            })?;
        Ok(Self {
            cmd_tx,
            join: Some(join),
        })
    }

    /// Graceful shutdown — drains commands + joins the thread.
    pub fn shutdown(mut self) {
        let (reply, rx) = oneshot::channel();
        let _ = self.cmd_tx.blocking_send(MediaCmd::Shutdown { reply });
        let _ = rx.blocking_recv();
        if let Some(join) = self.join.take() {
            let _ = join.join();
        }
    }
}

impl Drop for MediaThread {
    fn drop(&mut self) {
        if let Some(join) = self.join.take() {
            // Best-effort: if shutdown wasn't called explicitly, just
            // detach. The thread will exit when cmd_rx is dropped.
            // (We don't block on join in Drop — that could deadlock
            // if the thread is mid-call into a tokio runtime handle
            // that's being torn down.)
            debug!(name = ?join.thread().name(), "media thread detached on drop");
        }
    }
}

/// The per-session storage enum for the binary's MediaThread (spec §3.5).
///
/// `WebRTC` is the slice-4 leg kind -- ticked by
/// `rutster_media::loop_driver::drive`. `Trunk` is the new slice-5
/// trunk-leg kind -- ticked by `rutster_trunk::loop_driver::drive`. The
/// existing WebRTC code path stays unchanged: the match arm
/// `MediaLeg::WebRTC(s) => rutster_media::loop_driver::drive(s, now)`
/// calls the SAME drive function slice-4 invoked directly. The trunk
/// arm is its sibling.
///
/// Cargo dependency note: the trunk variant carries a concrete
/// `TrunkSession<TapAudioPipe>` (production instantiation); the T4 unit
/// tests can substitute `EchoAudioPipe` for the inner pipe by constructing
/// the generic type directly -- without going through MediaLeg (which
/// pins the production OuterType). The slice-4 §6.1 reuse claim holds:
/// both leg kinds compose Reflex + LocalVadReflex identically around
/// `TapAudioPipe`.
pub enum MediaLeg {
    /// Existing WebRTC leg. Ticked by `rutster_media::loop_driver::drive`.
    /// UNCHANGED code path from slice-4 + slice-5/seams.
    WebRTC(rutster_media::RtcSession),
    /// New trunk leg (slice-5). Ticked by `rutster_trunk::loop_driver::drive`.
    /// The seam gate holds: this never enters the media crate's
    /// loop_driver; the FOB-side tick function lives in rutster-trunk's
    /// loop_driver.rs (parallel-titled file, different crate).
    Trunk(rutster_trunk::session::TrunkSession<rutster_tap::TapAudioPipe>),
}

// Step A scaffold: methods are wired into the full Step B refactor
// (ThreadSession.leg field + RegisterTrunk handler body + dispatch in
// run_per_leg_tick inside the existing tick loop). Until Step B lands,
//  silences the unused warning per AGENTS.md
// (rationale documented inline; suppression NOT silence).
#[allow(dead_code)]
impl MediaLeg {
    /// Return the underlying `RtcSession` if this leg is WebRTC; `None`
    /// for trunk legs. Used by the existing slice-4 WebRTC code paths
    /// (accept_offer, Connected transition, is_closed) which require a
    /// mutable `RtcSession`. Trunk legs do not use these paths (the
    /// TrunkSession doesn't expose an `accept_offer` API -- the leg is
    /// constructed when the Twilio Media Streams WSS arrives, not when
    /// an SDP offer is POSTed).
    fn as_webrtc_mut(&mut self) -> Option<&mut rutster_media::RtcSession> {
        match self {
            MediaLeg::WebRTC(rtc) => Some(rtc),
            MediaLeg::Trunk(_) => None,
        }
    }

    /// Return the channel ID of this leg's `Channel` regardless of
    /// variant. Used by logging + tracing.
    fn channel_id(&self) -> rutster_call_model::ChannelId {
        match self {
            MediaLeg::WebRTC(rtc) => rtc.channel.id,
            MediaLeg::Trunk(t) => t.channel.id,
        }
    }

    /// Return the ChannelState of either leg kind's underlying channel.
    fn channel_state(&self) -> rutster_call_model::ChannelState {
        match self {
            MediaLeg::WebRTC(rtc) => rtc.channel.state,
            MediaLeg::Trunk(t) => t.channel.state,
        }
    }

    /// True once the leg's channel state has reached `Closed`. Used by the
    /// std thread's session-map eviction loop.
    fn is_closed(&self) -> bool {
        matches!(
            self.channel_state(),
            rutster_call_model::ChannelState::Closed
        )
    }

    /// Return the channel's `started_at` timestamp for lifecycle events.
    fn channel_started_at(&self) -> std::time::SystemTime {
        match self {
            MediaLeg::WebRTC(rtc) => rtc.channel.started_at,
            MediaLeg::Trunk(t) => t.channel.started_at,
        }
    }

    /// Set the channel's tap handle (set by the binary on the Connected
    /// transition for WebRTC legs; trunk legs set this at TrunkSession
    /// construction time when the tap_engine is spawned -- no Connected
    /// transition for trunk legs).
    fn set_tap_handle(&mut self, handle: rutster_call_model::TapHandle) {
        match self {
            MediaLeg::WebRTC(rtc) => rtc.channel.tap = Some(handle),
            MediaLeg::Trunk(t) => t.channel.tap = Some(handle),
        }
    }
}

/// The per-call tick dispatch. Mirrors the spec §3.5 contract: WebRTC
/// legs go through the UNCHANGED slice-4 `loop_driver::drive`; trunk legs
/// go through the NEW slice-5 `trunk_driver::drive`. The seam gate stays
/// green because the trunk leg NEVER enters the media-crate's
/// loop_driver.rs (only via the `rutster_trunk::loop_driver::drive` arm
/// here, which routes to a separate file in a separate crate).
///
/// Made `pub` so the T7/T8 integration tests can drive the loop directly
/// against a manually-constructed MediaLeg.
pub fn run_per_leg_tick(
    leg: &mut MediaLeg,
    now: std::time::Instant,
) -> Option<std::time::Duration> {
    match leg {
        MediaLeg::WebRTC(s) => rutster_media::loop_driver::drive(s, now),
        MediaLeg::Trunk(s) => rutster_trunk::loop_driver::drive(s, now),
    }
}

/// The per-session state owned by the media thread.
struct ThreadSession {
    rtc: RtcSession,
    /// The tap URL resolved at registration time. This is either the
    /// per-call override from `POST /v1/sessions` or the binary-wide default.
    /// The `Connected` transition dials this URL, not the `MediaThread`-wide
    /// default, so each session can be routed to a different brain/endpoint.
    tap_url: url::Url,
    /// A binary-view clone scoped to this session. `spawn_tap_engine` needs an
    /// `AppState` so that tools like `HangupTool` can send teardown commands
    /// back to the media thread through `AppState::close(id)`.
    app_state: crate::session_map::AppState,
    /// `Some` only after the `Connected` transition spawns the TapEngine
    /// + wires the `Reflex<TapAudioPipe>` wrapper.
    tap_conn: Option<crate::tap_engine::TapConn>,
}

fn run_media_thread(
    mut cmd_rx: mpsc::Receiver<MediaCmd>,
    default_tap_url: url::Url,
    opts: MediaThreadOpts,
    tokio_handle: tokio::runtime::Handle,
    media_cmd_tx: mpsc::Sender<MediaCmd>,
) {
    let mut sessions: HashMap<ChannelId, ThreadSession> = HashMap::new();
    let max_sessions = opts.max_sessions;
    let sink = opts.sink.clone();
    let mut tick_overruns: u64 = 0;
    let mut last_tick_micros: u64 = 0;
    let mut admission_rejects: u64 = 0;
    let mut draining = false;
    let mut drain_done: Option<oneshot::Sender<()>> = None;
    info!("media thread started");

    loop {
        let tick_started = Instant::now();

        // === Step 1: drain ALL pending commands (cold path) BEFORE ticking. ===
        while let Ok(cmd) = cmd_rx.try_recv() {
            match cmd {
                MediaCmd::Register { tap_url, reply } => {
                    // Drain (review M1) takes priority over the capacity
                    // check below: once draining, EVERY new Register is
                    // shed regardless of headroom — we're bleeding out,
                    // not admitting.
                    if draining {
                        admission_rejects += 1;
                        let _ = reply.send(Err("draining: not accepting new sessions".into()));
                        continue;
                    }
                    // Admission control (review M2): shed the marginal
                    // call with a crisp 503 instead of degrading every
                    // in-flight call's 20ms budget. "node full" is the
                    // routes-layer contract string.
                    if sessions.len() >= max_sessions {
                        admission_rejects += 1;
                        let _ = reply.send(Err(format!(
                            "node full: {} sessions (max {max_sessions})",
                            sessions.len()
                        )));
                        continue;
                    }
                    match RtcSession::new_with_config(&opts.media_cfg) {
                        Ok(session) => {
                            let id = session.channel_id();
                            let started_at = session.channel.started_at;
                            let app_state = crate::session_map::AppState::new(
                                media_cmd_tx.clone(),
                                default_tap_url.clone(),
                            );
                            sessions.insert(
                                id,
                                ThreadSession {
                                    rtc: session,
                                    tap_url,
                                    app_state,
                                    tap_conn: None,
                                },
                            );
                            sink.emit(crate::event_sink::CallEvent::SessionRegistered {
                                id,
                                at: started_at,
                            });
                            let _ = reply.send(Ok(id));
                            debug!(channel_id = %id, "session registered");
                        }
                        Err(e) => {
                            let _ = reply.send(Err(format!("RtcSession::new: {e}")));
                        }
                    }
                }
                MediaCmd::AcceptOffer { id, sdp, reply } => {
                    let result = match sessions.get_mut(&id) {
                        Some(s) => s.rtc.accept_offer(&sdp).map_err(|e| format!("{e}")),
                        None => Err(format!("session {id} not found")),
                    };
                    let _ = reply.send(result);
                }
                MediaCmd::Delete { id, reply } => {
                    if let Some(mut s) = sessions.remove(&id) {
                        // Snapshot BEFORE take(): after the take the conn
                        // (and its metrics Arc) belongs to the teardown task.
                        let tap_metrics = s.tap_conn.as_ref().map(|c| c.metrics.snapshot());
                        sink.emit(crate::event_sink::CallEvent::SessionEnded {
                            id,
                            started_at: s.rtc.channel.started_at,
                            ended_at: std::time::SystemTime::now(),
                            reason: crate::event_sink::EndReason::Deleted,
                            tap_metrics,
                        });
                        if let Some(conn) = s.tap_conn.take() {
                            // Hand the 750ms bounded teardown to tokio —
                            // the tick loop never waits on brain I/O
                            // (review M7; see spawn_tap_teardown docs).
                            crate::tap_engine::spawn_tap_teardown(&tokio_handle, id, conn);
                        }
                        s.rtc.channel.tap = None;
                        s.rtc.channel.state = rutster_call_model::ChannelState::Closed;
                    }
                    let _ = reply.send(());
                }
                MediaCmd::Shutdown { reply } => {
                    info!(
                        "media thread shutdown; dropping {} sessions",
                        sessions.len()
                    );
                    for (id, s) in sessions.iter() {
                        sink.emit(crate::event_sink::CallEvent::SessionEnded {
                            id: *id,
                            started_at: s.rtc.channel.started_at,
                            ended_at: std::time::SystemTime::now(),
                            reason: crate::event_sink::EndReason::Shutdown,
                            tap_metrics: s.tap_conn.as_ref().map(|c| c.metrics.snapshot()),
                        });
                    }
                    sessions.clear();
                    let _ = reply.send(());
                    return;
                }
                MediaCmd::Stats { reply } => {
                    let _ = reply.send(MediaStats {
                        sessions: sessions.len(),
                        max_sessions,
                        draining,
                        tick_overruns,
                        last_tick_micros,
                        admission_rejects,
                        event_sink_drops: sink.drops(),
                    });
                }
                MediaCmd::Drain { reply } => {
                    draining = true;
                    if sessions.is_empty() {
                        let _ = reply.send(());
                    } else {
                        drain_done = Some(reply);
                    }
                }
                // slice-5 T5: RegisterTrunk placeholder. Full handler construction
                // (Reflex <LocalVadReflex<TapAudioPipe>> composition + TrunkSession
                // instantiation + tap_engine spawn) lands in Step B alongside the
                // ThreadSession refactor to use MediaLeg's variant dispatch. For
                // now, ack with a synthetic error so the WSS pump task's oneshot
                // await resumes + closes the WSS (the caller observes a clean
                // failure rather than hanging). This matches the existing code
                // path for rejected Register commands (admission control, drain).
                MediaCmd::RegisterTrunk {
                    call_sid,
                    inbound_from_twilio_rx,
                    outbound_to_twilio_tx,
                    tap_url: _,
                    reply,
                } => {
                    // Drop the inbound mpsc + outbound mpsc ends so they don't
                    // leak the partial wiring live across the no-op ack.
                    drop(inbound_from_twilio_rx);
                    drop(outbound_to_twilio_tx);
                    tracing::warn!(
                        call_sid = %call_sid,
                        "RegisterTrunk: Step A placeholder -- full handler lands with the                          ThreadSession refactor; rejecting the registration for now"
                    );
                    // The reply is oneshot::Sender<ChannelId>, not Result; we
                    // can't NACK via Result. Closures: drop(reply) without
                    // sending so the WSS pump's reply_rx.await returns Err
                    // (which the pump treats as "media thread gone; close WSS").
                    drop(reply);
                }
            }
        }

        // === Step 2: the 10ms meta-tick over all sessions. ===
        let now = Instant::now();
        let mut closed_ids: Vec<ChannelId> = Vec::new();
        for (id, session) in sessions.iter_mut() {
            // Drain flush side-channel BEFORE run_poll_once (slice-2 §5.3 step 4).
            if let Some(conn) = session.tap_conn.as_mut() {
                if let Some(rx) = conn.flush_rx.as_mut() {
                    let mut should_flush = false;
                    while let Ok(()) = rx.try_recv() {
                        should_flush = true;
                    }
                    if should_flush {
                        session.rtc.clear_playout_ring();
                    }
                }
            }
            let _ = session.rtc.run_poll_once(now);

            // === relay the Connected transition: spawn TapEngine + wire Reflex. ===
            use rutster_call_model::ChannelState;
            if let ChannelState::Connected = session.rtc.channel.state {
                if session.rtc.channel.tap.is_none() {
                    // Use the per-call tap URL stored at registration time, not
                    // the binary-wide default, so a `tap_url_override` from the
                    // API body is honored.
                    let url = session.tap_url.clone();
                    // Give the tap engine a real `AppState` whose `cmd_tx`
                    // points back to this media thread. That lets tools such as
                    // `HangupTool` call `AppState::close(id)` and actually
                    // reach `MediaCmd::Delete` here.
                    let app_state = session.app_state.clone();
                    // The media thread owns the advisory channel (multi-producer:
                    // tokio::sync::mpsc::Sender is Clone). Both the brain's
                    // advisories (via spawn_tap_engine) AND the local VAD's
                    // trips (via LocalVadReflex) push to the SAME mpsc; the
                    // Reflex drains both uniformly. This means spawn_tap_engine
                    // takes advisory_tx as a PARAMETER (Task 5's signature
                    // changes: spawn_tap_engine(session_id, tap_url, app_state,
                    // advisory_tx) — see Task 5's revision note).
                    let (advisory_tx, advisory_rx) =
                        mpsc::channel::<rutster_media::AdvisoryEvent>(16);
                    let (pipe, conn) = tokio_handle.block_on(async {
                        spawn_tap_engine(*id, url, app_state, advisory_tx.clone())
                    });
                    let metrics = rutster_media::ReflexMetrics::new();
                    // Compose: Reflex<TapAudioPipe> (state machine) wrapped by
                    // LocalVadReflex (primary VAD trigger). Both feed advisory_tx.
                    let reflex = rutster_media::Reflex::new(pipe, advisory_rx, metrics);
                    let vad = rutster_media::LocalVadReflex::new(reflex, advisory_tx);
                    session.rtc.set_pipe(vad);
                    session.rtc.channel.tap = Some(rutster_call_model::TapHandle::new());
                    session.tap_conn = Some(conn);
                    info!(channel_id = %id, "tap engine + reflex + local VAD wired on Connected");
                    sink.emit(crate::event_sink::CallEvent::SessionConnected {
                        id: *id,
                        at: std::time::SystemTime::now(),
                    });
                    continue;
                }
            }

            if session.rtc.is_closed() {
                closed_ids.push(*id);
            }
        }
        for id in closed_ids {
            if let Some(s) = sessions.remove(&id) {
                sink.emit(crate::event_sink::CallEvent::SessionEnded {
                    id,
                    started_at: s.rtc.channel.started_at,
                    ended_at: std::time::SystemTime::now(),
                    reason: crate::event_sink::EndReason::Closed,
                    tap_metrics: s.tap_conn.as_ref().map(|c| c.metrics.snapshot()),
                });
            }
            debug!(channel_id = %id, "session evicted after close");
        }

        // Drain completion check — after eviction so a tick that closes
        // the last session completes the drain in the same iteration.
        if draining && sessions.is_empty() {
            if let Some(done) = drain_done.take() {
                let _ = done.send(());
            }
        }

        // === Step 3: compensated sleep (review M2). ===
        // The old fixed sleep(META_TICK) silently stretched the effective
        // tick to 10ms + work; under load that pushes every session past
        // its 20ms outbound deadline SIMULTANEOUSLY. Sleep only the
        // remainder, and count the ticks where there was none to sleep.
        let elapsed = tick_started.elapsed();
        last_tick_micros = elapsed.as_micros() as u64;
        if elapsed >= META_TICK {
            tick_overruns += 1;
        }
        std::thread::sleep(META_TICK.saturating_sub(elapsed));
    }
}

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

    #[test]
    fn register_rejects_when_at_capacity_and_stats_reports() {
        let rt = tokio::runtime::Runtime::new().unwrap();
        let handle = rt.handle().clone();
        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let opts = MediaThreadOpts {
            max_sessions: 1,
            ..Default::default()
        };
        let thread = MediaThread::spawn(url.clone(), opts, handle).expect("spawn");

        let register = |thread: &MediaThread| {
            let (reply, rx) = oneshot::channel();
            thread
                .cmd_tx()
                .blocking_send(MediaCmd::Register {
                    tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
                    reply,
                })
                .unwrap();
            rx.blocking_recv().expect("reply")
        };

        assert!(register(&thread).is_ok(), "first session fits");
        let err = register(&thread).expect_err("second must be rejected");
        assert!(err.contains("node full"), "routes contract string: {err}");

        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Stats { reply })
            .unwrap();
        let stats = rx.blocking_recv().expect("stats");
        assert_eq!(stats.sessions, 1);
        assert_eq!(stats.max_sessions, 1);
        assert!(!stats.draining);
        assert_eq!(stats.admission_rejects, 1, "capacity reject counted");
        assert!(
            stats.event_sink_drops.is_none(),
            "TracingEventSink has no drop counter"
        );

        thread.shutdown();
    }

    #[test]
    fn admission_rejects_counted_on_capacity_shed() {
        let rt = tokio::runtime::Runtime::new().unwrap();
        let handle = rt.handle().clone();
        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let opts = MediaThreadOpts {
            max_sessions: 1,
            ..Default::default()
        };
        let thread = MediaThread::spawn(url.clone(), opts, handle).expect("spawn");

        let register = |thread: &MediaThread| {
            let (reply, rx) = oneshot::channel();
            thread
                .cmd_tx()
                .blocking_send(MediaCmd::Register {
                    tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
                    reply,
                })
                .unwrap();
            rx.blocking_recv().expect("reply")
        };

        assert!(register(&thread).is_ok(), "first session fits");
        assert!(register(&thread).is_err(), "second session shed");

        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Stats { reply })
            .unwrap();
        let stats = rx.blocking_recv().expect("stats");
        assert_eq!(stats.admission_rejects, 1);
        assert!(
            stats.event_sink_drops.is_none(),
            "TracingEventSink returns None by default"
        );

        thread.shutdown();
    }

    #[test]
    fn drain_rejects_new_sessions_and_completes_when_map_empties() {
        let rt = tokio::runtime::Runtime::new().unwrap();
        let handle = rt.handle().clone();
        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let thread =
            MediaThread::spawn(url.clone(), MediaThreadOpts::default(), handle).expect("spawn");

        // one live session
        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Register {
                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
                reply,
            })
            .unwrap();
        let id = rx.blocking_recv().unwrap().unwrap();

        // drain: must NOT complete while the session lives
        let (drain_reply, mut drain_rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Drain { reply: drain_reply })
            .unwrap();
        std::thread::sleep(Duration::from_millis(50));
        assert!(
            drain_rx.try_recv().is_err(),
            "drain must wait for in-flight sessions"
        );

        // new registers are shed with the contract string
        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Register {
                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
                reply,
            })
            .unwrap();
        let err = rx.blocking_recv().unwrap().expect_err("draining rejects");
        assert!(err.contains("draining"), "{err}");

        // deleting the last session completes the drain
        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Delete { id, reply })
            .unwrap();
        rx.blocking_recv().unwrap();
        assert!(
            drain_rx.blocking_recv().is_ok(),
            "drain completes once the map is empty"
        );

        thread.shutdown();
    }

    #[test]
    fn lifecycle_events_emit_registered_then_ended_deleted() {
        use crate::event_sink::{CallEvent, EndReason, EventSink};
        struct TestSink(std::sync::Mutex<Vec<CallEvent>>);
        impl EventSink for TestSink {
            fn emit(&self, event: CallEvent) {
                self.0.lock().unwrap().push(event);
            }
        }
        let sink = std::sync::Arc::new(TestSink(std::sync::Mutex::new(Vec::new())));

        let rt = tokio::runtime::Runtime::new().unwrap();
        let handle = rt.handle().clone();
        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let opts = MediaThreadOpts {
            sink: sink.clone(),
            ..Default::default()
        };
        let thread = MediaThread::spawn(url, opts, handle).expect("spawn");

        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Register {
                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
                reply,
            })
            .unwrap();
        let id = rx.blocking_recv().unwrap().unwrap();
        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Delete { id, reply })
            .unwrap();
        rx.blocking_recv().unwrap();
        thread.shutdown();

        let events = sink.0.lock().unwrap();
        assert!(matches!(&events[0], CallEvent::SessionRegistered { id: eid, .. } if *eid == id));
        assert!(events.iter().any(|e| matches!(
            e,
            CallEvent::SessionEnded { id: eid, reason: EndReason::Deleted, .. } if *eid == id
        )));
    }

    #[test]
    fn media_thread_register_and_shutdown_round_trips() {
        let rt = tokio::runtime::Runtime::new().unwrap();
        let handle = rt.handle().clone();
        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let thread = MediaThread::spawn(url, MediaThreadOpts::default(), handle)
            .expect("media thread spawn in test");
        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Register {
                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
                reply,
            })
            .unwrap();
        let id = rx
            .blocking_recv()
            .expect("register reply")
            .expect("session");
        assert_eq!(format!("{}", id).len(), 36, "UUID-shaped ChannelId");
        thread.shutdown();
    }
}
