peerspeak/src/audio/clip_player.rs

//! Independent playback engine for inline chat audio attachments.
//!
//! The rodio device sink stays on a dedicated OS thread and never enters iced
//! state or the call-audio pipeline. The GUI sends small commands and reads a
//! shared status snapshot at its redraw cadence.

use crate::files::AttachmentId;
use rodio::{Decoder, DeviceSinkBuilder, MixerDeviceSink, Player, Source, decoder::DecoderError};
use std::io::Cursor;
use std::sync::{Arc, Mutex, mpsc};
use std::time::Duration;

/// State published by the playback thread for the GUI.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ClipStatus {
    pub playing_id: Option<AttachmentId>,
    pub position: Duration,
    pub total: Option<Duration>,
    pub paused: bool,
    /// Set when output initialization or decoding rejects the requested clip.
    /// The app consumes this as a signal to fall back to the normal file chip.
    pub failure: Option<ClipFailure>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClipFailure {
    pub id: AttachmentId,
    pub error: String,
    /// Decoder rejection means the filename hint should fall back to a file
    /// chip. Output-device failures remain retryable as audio.
    pub invalid_audio: bool,
}

pub type SharedClipStatus = Arc<Mutex<ClipStatus>>;

#[derive(Debug)]
enum ClipCommand {
    Play(AttachmentId, Vec<u8>),
    Pause,
    Resume,
    Seek(Duration),
    Stop,
    SetVolume(f32),
}

/// Cheap, `Send` command handle for the dedicated playback thread.
pub struct ClipPlayer {
    command_tx: mpsc::Sender<ClipCommand>,
    status: SharedClipStatus,
}

impl ClipPlayer {
    /// Start the playback worker. The system output device is opened lazily on
    /// first Play, so merely launching PeerSpeak never claims another stream.
    ///
    /// `initial_volume` is the universal gain (`1.0` = unity) applied to every
    /// clip, restored from config so the level persists across sessions.
    pub fn new(initial_volume: f32) -> (Self, SharedClipStatus) {
        let (command_tx, command_rx) = mpsc::channel();
        let status = Arc::new(Mutex::new(ClipStatus::default()));
        let worker_status = Arc::clone(&status);
        std::thread::Builder::new()
            .name("peerspeak-clip-player".to_string())
            .spawn(move || playback_worker(command_rx, worker_status, initial_volume))
            .expect("failed to spawn clip playback thread");
        (
            Self {
                command_tx,
                status: Arc::clone(&status),
            },
            status,
        )
    }

    pub fn play(&self, id: AttachmentId, bytes: Vec<u8>) {
        update_status(&self.status, |status| {
            status.playing_id = Some(id);
            status.position = Duration::ZERO;
            status.total = None;
            status.paused = false;
            status.failure = None;
        });
        let _ = self.command_tx.send(ClipCommand::Play(id, bytes));
    }

    pub fn pause(&self) {
        let _ = self.command_tx.send(ClipCommand::Pause);
    }

    pub fn resume(&self) {
        let _ = self.command_tx.send(ClipCommand::Resume);
    }

    pub fn seek(&self, position: Duration) {
        let _ = self.command_tx.send(ClipCommand::Seek(position));
    }

    pub fn stop(&self) {
        let _ = self.command_tx.send(ClipCommand::Stop);
    }

    /// Set the universal playback gain (`1.0` = unity). Applies to the current
    /// clip immediately and to every clip played afterwards.
    pub fn set_volume(&self, volume: f32) {
        let _ = self.command_tx.send(ClipCommand::SetVolume(volume));
    }
}

fn playback_worker(
    command_rx: mpsc::Receiver<ClipCommand>,
    status: SharedClipStatus,
    initial_volume: f32,
) {
    let mut output: Option<MixerDeviceSink> = None;
    let mut player: Option<Player> = None;
    // Universal gain remembered across clips so a level set on one upload
    // carries to the next; reapplied to each freshly connected player.
    let mut volume = initial_volume.max(0.0);

    loop {
        match command_rx.recv_timeout(Duration::from_millis(100)) {
            Ok(ClipCommand::Play(id, bytes)) => {
                // In-memory readers do not expose file metadata to rodio. Pass
                // the known attachment length explicitly so formats without a
                // duration in their headers (notably MP3 and Vorbis) can derive
                // a total duration and support reliable seeking.
                let source = match decode_clip(bytes) {
                    Ok(source) => source,
                    Err(error) => {
                        fail(
                            &status,
                            id,
                            format!("unsupported or invalid audio: {error}"),
                            true,
                        );
                        continue;
                    }
                };
                let total = source.total_duration();

                if output.is_none() {
                    match DeviceSinkBuilder::open_default_sink() {
                        Ok(sink) => {
                            let new_player = Player::connect_new(sink.mixer());
                            new_player.set_volume(volume);
                            player = Some(new_player);
                            output = Some(sink);
                        }
                        Err(error) => {
                            fail(
                                &status,
                                id,
                                format!("audio output unavailable: {error}"),
                                false,
                            );
                            continue;
                        }
                    }
                }

                if let Some(player) = player.as_ref() {
                    player.clear();
                    player.append(source);
                    player.play();
                    update_status(&status, |s| {
                        s.playing_id = Some(id);
                        s.position = Duration::ZERO;
                        s.total = total;
                        s.paused = false;
                        s.failure = None;
                    });
                }
            }
            Ok(ClipCommand::Pause) => {
                if let Some(player) = player.as_ref() {
                    player.pause();
                    update_status(&status, |s| s.paused = true);
                }
            }
            Ok(ClipCommand::Resume) => {
                if let Some(player) = player.as_ref() {
                    player.play();
                    update_status(&status, |s| s.paused = false);
                }
            }
            Ok(ClipCommand::Seek(position)) => {
                if let Some(player) = player.as_ref()
                    && player.try_seek(position).is_ok()
                {
                    update_status(&status, |s| s.position = position);
                }
            }
            Ok(ClipCommand::Stop) => {
                if let Some(player) = player.as_ref() {
                    player.clear();
                }
                reset(&status);
            }
            Ok(ClipCommand::SetVolume(level)) => {
                volume = level.max(0.0);
                if let Some(player) = player.as_ref() {
                    player.set_volume(volume);
                }
            }
            Err(mpsc::RecvTimeoutError::Disconnected) => break,
            Err(mpsc::RecvTimeoutError::Timeout) => {}
        }

        if let Some(player) = player.as_ref() {
            let (active, failed) = status
                .lock()
                .map(|s| (s.playing_id.is_some(), s.failure.is_some()))
                .unwrap_or_default();
            if active && !failed && player.empty() {
                reset(&status);
            } else if active && !failed {
                update_status(&status, |s| {
                    s.position = player.get_pos();
                    s.paused = player.is_paused();
                });
            }
        }
    }
}

fn decode_clip(bytes: Vec<u8>) -> Result<Decoder<Cursor<Vec<u8>>>, DecoderError> {
    let byte_len = bytes.len() as u64;
    Decoder::builder()
        .with_data(Cursor::new(bytes))
        .with_byte_len(byte_len)
        .build()
}

fn fail(status: &SharedClipStatus, id: AttachmentId, error: String, invalid_audio: bool) {
    crate::log_msg(&format!("Inline audio playback failed: {error}"));
    update_status(status, |s| {
        // Keep the id active until the GUI observes the failure on its next
        // tick. This guarantees the active-only timer cannot disappear in the
        // small window between sending Play and decoder/output failure.
        s.playing_id = Some(id);
        s.position = Duration::ZERO;
        s.total = None;
        s.paused = false;
        s.failure = Some(ClipFailure {
            id,
            error,
            invalid_audio,
        });
    });
}

fn reset(status: &SharedClipStatus) {
    update_status(status, |s| *s = ClipStatus::default());
}

fn update_status(status: &SharedClipStatus, update: impl FnOnce(&mut ClipStatus)) {
    if let Ok(mut status) = status.lock() {
        update(&mut status);
    }
}

pub fn status_snapshot(status: &SharedClipStatus) -> ClipStatus {
    status.lock().map(|s| s.clone()).unwrap_or_default()
}

/// Format clip time as `mm:ss` (hours are folded into minutes).
pub fn format_time(duration: Duration) -> String {
    let seconds = duration.as_secs();
    format!("{}:{:02}", seconds / 60, seconds % 60)
}

/// Playback progress in `0.0..=1.0`; unknown and zero durations report zero.
pub fn progress(position: Duration, total: Option<Duration>) -> f32 {
    let Some(total) = total.filter(|duration| !duration.is_zero()) else {
        return 0.0;
    };
    (position.as_secs_f64() / total.as_secs_f64()).clamp(0.0, 1.0) as f32
}

/// Convert a slider fraction into a clamped position within a clip.
pub fn seek_target(fraction: f32, total: Duration) -> Duration {
    total.mul_f64(f64::from(fraction.clamp(0.0, 1.0)))
}

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

    #[test]
    fn in_memory_mp3_reports_duration() {
        // One headerless constant-bitrate MP3 frame repeated to model files
        // that do not carry an Xing/VBR duration header.
        let frame = base64::engine::general_purpose::STANDARD
            .decode("//sQxAAABIQVWVRggDCqCKiDNlAAAAGgS4BgAmTT2AQAABCxOD5d7gQOfqBAEHS4Ph/EAIRI7//0A0KBNpABgMRIDCSI04PcIFdF0PJKFgzlUf5eAoF8BRIPfh4FTvUDQl+dUi5pc0w=")
            .expect("valid test fixture");
        let bytes = frame.repeat(20);

        let decoder = decode_clip(bytes).expect("CBR MP3 should decode");
        assert!(decoder.total_duration().is_some());
    }

    #[test]
    fn formats_clip_time() {
        assert_eq!(format_time(Duration::ZERO), "0:00");
        assert_eq!(format_time(Duration::from_secs(65)), "1:05");
        assert_eq!(format_time(Duration::from_secs(3_661)), "61:01");
    }

    #[test]
    fn progress_handles_unknown_zero_and_clamps() {
        assert_eq!(progress(Duration::from_secs(1), None), 0.0);
        assert_eq!(progress(Duration::from_secs(1), Some(Duration::ZERO)), 0.0);
        assert_eq!(
            progress(Duration::from_secs(5), Some(Duration::from_secs(10))),
            0.5
        );
        assert_eq!(
            progress(Duration::from_secs(20), Some(Duration::from_secs(10))),
            1.0
        );
    }

    #[test]
    fn seek_target_clamps_fraction() {
        let total = Duration::from_secs(100);
        assert_eq!(seek_target(0.25, total), Duration::from_secs(25));
        assert_eq!(seek_target(-1.0, total), Duration::ZERO);
        assert_eq!(seek_target(2.0, total), total);
    }

    /// **The fourth playback path's exit gate (round 10, R10-2).** Drives a
    /// real [`ClipPlayer`] — the same object the app uses for chat clips, peer
    /// music and the local playlist — and asserts the node it puts on the
    /// graph carries both ownership carriers.
    ///
    /// This path was untagged through all of phase 1, which is a real echo:
    /// B broadcasts music, A tunes in, A shares their desktop, B hears their
    /// own track. It was missed because phase 1 worked from the impl plan's
    /// list of three playback sites and that list was incomplete — so this
    /// gate drives the *player*, not the tagging helper.
    ///
    /// ⚠️ **Run alone**: it sets a process-wide environment variable, which is
    /// only sound single-threaded. In production `main` does this before
    /// anything is spawned; a test binary has no such guarantee, hence
    /// `--test-threads=1`.
    ///
    /// `cargo test --lib -- --ignored --test-threads=1 clip_player_node`
    #[test]
    #[ignore = "live: requires a running PipeWire daemon and pw-dump; run with --test-threads=1"]
    fn clip_player_node_carries_both_ownership_carriers() {
        use crate::audio::ownership::{self, live_test};

        // SAFETY: `--test-threads=1` is documented above and in the ignore
        // reason; this is the same call `main` makes, exercised for real
        // rather than reimplemented, so the gate cannot pass against a
        // formatter that production never uses.
        unsafe { ownership::tag_this_process_alsa_audio() };

        let (player, _status) = ClipPlayer::new(1.0);
        // Six seconds of silence: long enough for the poll, inaudible.
        player.play([0u8; 32], live_test::silent_wav(6));

        let prefix = live_test::expected_prefix(ownership::CLIP_ROLE);
        let found = live_test::poll_for_owned_node(&prefix, Duration::from_secs(5));
        player.stop();

        let (name, owned) = found.unwrap_or_else(|| {
            panic!("no live clip-player node named {prefix:?} appeared within 5s")
        });
        assert!(
            name.starts_with(ownership::OWNED_NODE_NAME_PREFIX),
            "{name}"
        );
        assert_eq!(
            owned.as_deref(),
            Some(ownership::OWNED_PROP_VALUE),
            "carrier 1 must be on the live node, not just carrier 2"
        );
    }
}