rutster/crates/rutster-tap/src/tap_audio_pipe.rs

//! # TapAudioPipe — the seam object (spec §4.1)
//!
//! The sync object `RtcSession` holds and `loop_driver` calls via the
//! `AudioSource`/`AudioSink` trait seam. It's a thin wrapper over two
//! `tokio::sync::mpsc` channels + a bounded `VecDeque` playout ring:
//!
//! ```text
//!   peer mic → Opus decode → on_pcm_frame() → tx_pcm_in → TapClient → audio_in (WS)
//!   brain audio_out (WS) → TapClient → rx_audio_out → [playout ring] → next_pcm_frame() → Opus encode → peer
//! ```
//!
//! # Why `VecDeque` for the playout ring (not `mpsc`?)
//! The playout ring has a specific policy on overflow (drop *oldest*, not
//! newest — spec §4.1). `mpsc` drops the *newest* on `try_send` when
//! full (the receiver would see a gap). For a real-time media path, we
//! want stale (old) frames dropped, not late (new) frames — drop-oldest
//! keeps the buffer at-or-behind real-time. A `VecDeque` under our
//! manual `push_back_bounded` policy is the smallest structure that fits.

use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::atomic::Ordering;

use rutster_media::{AudioSink, AudioSource, PcmFrame};
use tokio::sync::mpsc;
use tracing::{debug, trace};

use crate::metrics::{TAP_PLAYOUT_FRAMES, TapMetrics};

pub struct TapAudioPipe {
    // Core → brain (inbound decoded PCM from peer):
    tx_pcm_in: mpsc::Sender<PcmFrame>,
    // Brain → core (playout buffer):
    rx_audio_out: mpsc::Receiver<PcmFrame>,
    // Bounded ring between rx_audio_out and next_pcm_frame (spec §4.1).
    playout_ring: VecDeque<PcmFrame>,
    metrics: Arc<TapMetrics>,
}

impl TapAudioPipe {
    pub fn new(
        tx_pcm_in: mpsc::Sender<PcmFrame>,
        rx_audio_out: mpsc::Receiver<PcmFrame>,
        metrics: Arc<TapMetrics>,
    ) -> Self {
        Self {
            tx_pcm_in,
            rx_audio_out,
            playout_ring: VecDeque::with_capacity(TAP_PLAYOUT_FRAMES),
            metrics,
        }
    }

    /// Drain the inbound mpsc into the playout ring. Called by
    /// `next_pcm_frame` (one drain per source tick). Returns the count
    /// drained. **Hot-path policy:** every drop is counted (drop + observe).
    fn drain_inbound(&mut self) -> usize {
        let mut drained = 0;
        loop {
            match self.rx_audio_out.try_recv() {
                Ok(frame) => {
                    // push_back_bounded: drop oldest if full (drop-oldest policy).
                    if self.playout_ring.len() >= TAP_PLAYOUT_FRAMES {
                        self.playout_ring.pop_front();
                        self.metrics
                            .playout_overflow
                            .fetch_add(1, Ordering::Relaxed);
                        debug!(overflow = true, "playout ring overflow; dropped oldest");
                    }
                    self.playout_ring.push_back(frame);
                    drained += 1;
                }
                Err(mpsc::error::TryRecvError::Empty) => break,
                Err(mpsc::error::TryRecvError::Disconnected) => {
                    // Engine task gone. Silence until TapClient reconnects
                    // via a fresh mpsc (handled by TapEngine's reconnect loop).
                    trace!("rx_audio_out disconnected; silence until reconnect");
                    break;
                }
            }
        }
        drained
    }
}

impl AudioSource for TapAudioPipe {
    /// Take the next brain-proposed PCM frame to send to the peer.
    /// `None` = silence (loop_driver emits Opus silence on None — slice-1).
    fn next_pcm_frame(&mut self) -> Option<PcmFrame> {
        // First drain any frames the engine task has queued; then pop one.
        self.drain_inbound();
        match self.playout_ring.pop_front() {
            Some(frame) => Some(frame),
            None => {
                // Underflow: brain slower than the 20ms tick. Silence.
                self.metrics
                    .playout_underflow
                    .fetch_add(1, Ordering::Relaxed);
                None
            }
        }
    }
}

impl AudioSink for TapAudioPipe {
    /// Receive a decoded PCM frame from the peer. Must not block
    /// (slice-1 §3.3 contract — `on_pcm_frame` runs in the 20ms loop).
    /// `try_send` (not `send`) — if the channel is full (engine task slow
    /// or gone), drop + count (hot-path "drop + observe, don't crash").
    fn on_pcm_frame(&mut self, frame: PcmFrame) {
        if self.tx_pcm_in.try_send(frame).is_err() {
            self.metrics.inbound_dropped.fetch_add(1, Ordering::Relaxed);
            trace!("inbound PCM dropped (channel full)");
        }
    }
}

impl rutster_media::AudioPipe for TapAudioPipe {
    /// slice-2 spec §5.3 step 4: on brain disconnect, the engine task
    /// signals a flush via the side-channel `flush_tx` mpsc; the binary's
    /// poll task drains that signal and calls this. Without the flush,
    /// stale brain-proposed frames in `playout_ring` would keep playing
    /// after the WS is dead — the user would hear ~100 ms of audio from
    /// the *previous* brain session right after the reconnect, which is
    /// exactly the "stale bleed-through" the spec warns against.
    fn clear_playout_ring(&mut self) {
        let cleared = self.playout_ring.len();
        self.playout_ring.clear();
        if cleared > 0 {
            debug!(cleared, "playout ring flushed on brain disconnect");
        }
    }

    /// slice-4 spec §3.3 — barge-in flush: clear the playout ring AND
    /// drain `rx_audio_out` of any frames queued before the barge. Without
    /// this drain, a stale brain frame in the mpsc would un-mute
    /// immediately on the next tick — defeating the "first fresh audio_out"
    /// resume condition. Hot-path: try_recv loop, bounded, no blocking.
    fn barge_in_flush(&mut self) {
        let cleared = self.playout_ring.len();
        self.playout_ring.clear();
        if cleared > 0 {
            debug!(cleared, "playout ring flushed on barge-in");
        }

        let mut drained = 0usize;
        while self.rx_audio_out.try_recv().is_ok() {
            drained += 1;
        }
        if drained > 0 {
            self.metrics
                .barge_drained_inflight
                .fetch_add(drained as u64, Ordering::Relaxed);
            debug!(drained, "in-flight brain frames drained on barge-in");
        }
    }
}

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

    #[allow(clippy::type_complexity)] // test helper: 5-tuple of channel ends; not worth a struct.
    fn channels() -> (
        mpsc::Sender<PcmFrame>,
        mpsc::Receiver<PcmFrame>,
        mpsc::Sender<PcmFrame>,
        mpsc::Receiver<PcmFrame>,
        Arc<TapMetrics>,
    ) {
        // tx_pcm_in (inbound PCM → engine), rx_audio_out (engine → playout ring)
        let (tx_pcm_in, rx_pcm_in) = mpsc::channel(64);
        let (tx_audio_out, rx_audio_out) = mpsc::channel(64);
        let metrics = TapMetrics::new();
        (tx_pcm_in, rx_pcm_in, tx_audio_out, rx_audio_out, metrics)
    }

    #[test]
    fn source_returns_none_on_empty_ring() {
        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics);
        assert!(pipe.next_pcm_frame().is_none());
    }

    #[test]
    fn source_returns_frame_pushed_by_engine() {
        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone());
        // Engine task (simulated): push an audio_out frame.
        let mut frame = PcmFrame::zeroed();
        frame.samples[0] = 42;
        tx_audio_out.blocking_send(frame).unwrap();
        // Source should now drain + return it.
        let got = pipe.next_pcm_frame().expect("frame present");
        assert_eq!(got.samples[0], 42);
        assert_eq!(metrics.playout_underflow.load(Ordering::Relaxed), 0);
    }

    #[test]
    fn overflow_drops_oldest_not_newest() {
        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone());
        // Push TAP_PLAYOUT_FRAMES + 3 frames. The first 3 should be dropped
        // (oldest), keeping the last TAP_PLAYOUT_FRAMES.
        for i in 0..(TAP_PLAYOUT_FRAMES + 3) {
            let mut f = PcmFrame::zeroed();
            f.samples[0] = i as i16;
            tx_audio_out.blocking_send(f).unwrap();
        }
        // Drain into the ring via next_pcm_frame's drain step (returns the
        // first popped, so we call N+1 times to fully drain + assert).
        let first = pipe.next_pcm_frame().expect("first frame after drain");
        assert_eq!(first.samples[0], 3); // first 3 (0,1,2) were dropped
        assert_eq!(metrics.playout_overflow.load(Ordering::Relaxed), 3);
    }

    #[test]
    fn underflow_increments_counter() {
        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics.clone());
        pipe.next_pcm_frame(); // None -> underflow
        pipe.next_pcm_frame(); // None -> underflow
        assert_eq!(metrics.playout_underflow.load(Ordering::Relaxed), 2);
    }

    #[test]
    fn sink_full_drops_and_counts() {
        // Tiny inbound channel to force overflow.
        let (tx_pcm_in, _rx_pcm_in, _tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, metrics.clone());
        // Channel capacity 64 — push 100 frames.
        for _ in 0..100 {
            pipe.on_pcm_frame(PcmFrame::zeroed());
        }
        assert!(metrics.inbound_dropped.load(Ordering::Relaxed) > 0);
    }

    #[test]
    fn disconnected_engine_returns_none() {
        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics);
        // Drop the sender → rx_audio_out sees Disconnected → None (silence).
        // (Implicit drop at scope exit here; explicit next_pcm_frame below
        // reads Disconnected.)
        assert!(pipe.next_pcm_frame().is_none());
    }

    #[test]
    fn disconnected_engine_returns_none_after_close() {
        let (tx_pcm_in, _rx_pcm_in, tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, metrics);
        drop(tx_audio_out); // close the engine→playout-ring direction
        // Now next_pcm_frame should return None (silence) — Disconnected path.
        assert!(pipe.next_pcm_frame().is_none());
    }

    /// slice-4 §3.3: barge_in_flush clears the playout ring AND drains the
    /// inbound `rx_audio_out` mpsc of any frames queued before the barge.
    /// Without draining the mpsc, a stale pre-barge frame would un-mute
    /// immediately on the next tick — defeating the "first fresh audio_out"
    /// resume condition.
    #[test]
    fn barge_in_flush_clears_ring_and_drains_rx_audio_out() {
        let (_tx_pcm_in, _rx_pcm_in, tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone());

        // Put some frames into the playout ring first (simulating steady-state
        // playout that already drained from the mpsc).
        for i in 0..2 {
            let mut f = PcmFrame::zeroed();
            f.samples[0] = i as i16;
            tx_audio_out.blocking_send(f).unwrap();
        }
        let _ = pipe.next_pcm_frame(); // drains mpsc into ring, pops first

        // Now queue MORE frames into rx_audio_out that have NOT been pulled
        // into the ring yet — these are the "in-flight" stale frames that
        // must be drained during a barge-in to keep resume race-free.
        for i in 0..3 {
            let mut f = PcmFrame::zeroed();
            f.samples[0] = (10 + i) as i16;
            tx_audio_out.blocking_send(f).unwrap();
        }

        // Barge-in: ring should clear + the 3 mpsc frames drain.
        pipe.barge_in_flush();

        assert!(pipe.next_pcm_frame().is_none());
        assert_eq!(
            metrics.barge_drained_inflight.load(Ordering::Relaxed),
            3,
            "three in-flight mpsc frames should be drained on barge-in"
        );
    }

    /// slice-4 §3.3: barge_in_flush when empty is a no-op and leaves the
    /// counter at zero.
    #[test]
    fn barge_in_flush_when_already_empty_is_noop() {
        let (_tx_pcm_in, _rx_pcm_in, _tx_audio_out, rx_audio_out, metrics) = channels();
        let mut pipe = TapAudioPipe::new(_tx_pcm_in, rx_audio_out, metrics.clone());
        pipe.barge_in_flush();
        assert_eq!(metrics.barge_drained_inflight.load(Ordering::Relaxed), 0);
    }
}