//! # TrunkSession -- the per-trunk-leg session (spec §3.3)
//!
//! Parallels slice-1's `RtcSession` minus the str0m/Opus/UDP-socket footprint.
//! The wrapped pipe stack is REUSED from slice-4 verbatim:
//! `LocalVadReflex<Reflex<P>>`, where `P: AudioPipe + Send` is the inner pipe
//! (production: `TapAudioPipe`; tests: `EchoAudioPipe`). The composition site
//! (MediaThread's `RegisterTrunk` handler -- T5) constructs the full stack.

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

use rutster_call_model::Channel;
use rutster_media::PcmFrame;
use tokio::sync::mpsc;

/// 60 s of silence (no inbound PCM from Twilio's "media" frames) means
/// "the call leg is dead" -- the WSS pump task has likely crashed or the
/// Twilio Media Streams fork has been silently closed by the cloud. Same
/// threshold as slice-1's RtcSession idle timeout (spec §4.5).
pub const IDLE_TIMEOUT: Duration = Duration::from_secs(60);

/// 20 ms outbound encode tick (matches slice-1's PCM frame size: 480 samples
/// @ 24 kHz = 20 ms). The trunk driver pulls a single PCM frame from the
/// pipe's `next_pcm_frame` source on each tick onset + pushes to the
/// outbound WSS mpsc. Mirrors `rutster_media::loop_driver::OUTBOUND_TICK`.
pub const OUTBOUND_TICK: Duration = Duration::from_millis(20);

/// The per-trunk-leg session. Generic over `P: AudioPipe + Send` so tests
/// substitute a stub inner pipe (e.g. `EchoAudioPipe`) without constructing
/// the full TapEngine wiring (TapConn, WS handshake, mpsc pairs).
///
/// # Production type instantiation
///
/// `TrunkSession<TapAudioPipe>` -- MediaThread::RegisterTrunk (T5)
/// composes:
///   1. `let (tap_pipe, tap_conn) = spawn_tap_engine(...)` (slice-2).
///   2. `let reflex = Reflex::new(tap_pipe, advisory_rx, metrics)` (slice-4).
///   3. `let wrapped = LocalVadReflex::new(reflex, advisory_tx)` (slice-4).
///   4. `TrunkSession { pipe: wrapped, ... }`.
///
/// The `(P: AudioPipe + Send)` bound mirrors the slice-4 trait constraints:
/// the std thread holds the session behind `&mut TrunkSession` exclusively,
/// but Send is required when the binary moves sessions across thread-task
/// boundaries for teardown or migration.
pub struct TrunkSession<P: rutster_media::AudioPipe + Send> {
    /// The call-channel the FOB created for this trunk leg.
    pub channel: Channel,
    /// The wrapped reflex stack: `LocalVadReflex<Reflex<P>>`. Constructed
    /// by MediaThread::RegisterTrunk (T5), REUSED slice-4 task-6 composition.
    pub pipe: rutster_media::LocalVadReflex<rutster_media::Reflex<P>>,
    /// Caller PCM from Twilio (decoded at 24 kHz by the WSS pump task).
    pub inbound_from_twilio_rx: mpsc::Receiver<PcmFrame>,
    /// FOB outbound PCM for Twilio (the WSS pump encodes µ-law + sends back).
    pub outbound_to_twilio_tx: mpsc::Sender<PcmFrame>,
    /// Last Instant a caller PCM frame was received. Used to detect the
    /// 60 s idle timeout (WSS pump gone silent -> close).
    pub last_idle_rx: Instant,
    /// Cached next-deadline Instant (mirrors `RtcSession::next_timeout`).
    pub next_timeout: Option<Instant>,
    /// Last Instant we wrote an outbound PCM frame (mirrors
    /// `RtcSession::last_outbound_at`). Paces the 20 ms outbound tick.
    pub last_outbound_at: Instant,
}

impl<P: rutster_media::AudioPipe + Send> TrunkSession<P> {
    /// Construct a session with the provided wrapped pipe + mpsc pair. The
    /// MediaThread::RegisterTrunk (T5) handler owns this construction site
    /// in production; tests call this directly with stub pipes.
    pub fn new(
        channel: Channel,
        pipe: rutster_media::LocalVadReflex<rutster_media::Reflex<P>>,
        inbound_from_twilio_rx: mpsc::Receiver<PcmFrame>,
        outbound_to_twilio_tx: mpsc::Sender<PcmFrame>,
        now: Instant,
    ) -> Self {
        Self {
            channel,
            pipe,
            inbound_from_twilio_rx,
            outbound_to_twilio_tx,
            last_idle_rx: now,
            next_timeout: None,
            last_outbound_at: now,
        }
    }

    /// Slice-4's `ChannelState::Closed` predicate (mirrors the slice-1
    /// `RtcSession::is_closed` API). Used by the std thread's tick loop to
    /// evict closed sessions from the session map.
    pub fn is_closed(&self) -> bool {
        matches!(self.channel.state, rutster_call_model::ChannelState::Closed)
    }
}

#[cfg(test)]
pub mod tests {
    use super::*;
    use rutster_media::{EchoAudioPipe, LocalVadReflex, Reflex, ReflexMetrics};
    use tokio::sync::mpsc;

    /// Build a `TrunkSession<EchoAudioPipe>` with no real brain wiring
    /// (no TapEngine, no WS endpoint). The materialized wrapped pipe is the
    /// same composition the FOB constructs in production:
    /// `LocalVadReflex<Reflex<EchoAudioPipe>>`. The advisory mpsc pair is
    /// fresh and never receives events in these tests.
    pub fn trunk_session_with_echo_pipe(now: Instant) -> TrunkSession<EchoAudioPipe> {
        let (advisory_tx, advisory_rx) = mpsc::channel(16);
        let metrics = ReflexMetrics::new();
        let reflex = Reflex::new(EchoAudioPipe::new(), advisory_rx, metrics);
        let wrapped = LocalVadReflex::new(reflex, advisory_tx);
        let (_inbound_tx, inbound_rx) = mpsc::channel(16);
        let (outbound_tx, _outbound_rx) = mpsc::channel(16);
        let channel = Channel::new_inbound();
        TrunkSession::new(channel, wrapped, inbound_rx, outbound_tx, now)
    }

    /// EchoAudioPipe ownership partner of [`trunk_session_with_echo_pipe`]:
    /// returns the sender + receiver ends separately for test scenarios.
    pub fn trunk_session_with_echo_pipe_and_mpsc(
        now: Instant,
    ) -> (
        TrunkSession<EchoAudioPipe>,
        mpsc::Sender<PcmFrame>,
        mpsc::Receiver<PcmFrame>,
    ) {
        let (advisory_tx, advisory_rx) = mpsc::channel(16);
        let metrics = ReflexMetrics::new();
        let reflex = Reflex::new(EchoAudioPipe::new(), advisory_rx, metrics);
        let wrapped = LocalVadReflex::new(reflex, advisory_tx);
        let (inbound_tx, inbound_rx) = mpsc::channel(16);
        let (outbound_tx, outbound_rx) = mpsc::channel(16);
        let channel = Channel::new_inbound();
        let session = TrunkSession::new(channel, wrapped, inbound_rx, outbound_tx, now);
        (session, inbound_tx, outbound_rx)
    }

    #[test]
    fn new_session_starts_in_new_state() {
        let now = Instant::now();
        let session = trunk_session_with_echo_pipe(now);
        assert!(matches!(
            session.channel.state,
            rutster_call_model::ChannelState::New
        ));
    }

    #[test]
    fn is_closed_false_for_fresh_session() {
        let now = Instant::now();
        let session = trunk_session_with_echo_pipe(now);
        assert!(!session.is_closed());
    }
}
