crabidy/audio-player/src/player_engine.rs

use std::fs::File;
use std::io::BufReader;
use std::path::Path;
use std::time::Duration;

use anyhow::{anyhow, Context, Result};
use flume::{Receiver, RecvTimeoutError, Sender};
use rodio::source::EmptyCallback;
use rodio::stream::{DeviceSinkBuilder, MixerDeviceSink};
use rodio::{Decoder, Source};
use stream_download::storage::temp::TempStorageProvider;
use stream_download::{Settings, StreamDownload};

use crate::windowed_http::{WindowedHttpParams, WindowedHttpStream};
use thiserror::Error;
use tracing::{debug, info, instrument, trace, warn};
use url::Url;

/// How long we wait for the initial prefetch of a network stream.
const STREAM_OPEN_TIMEOUT: Duration = Duration::from_secs(30);

/// What playback logs for a source: local paths verbatim, network URLs
/// reduced to scheme and host — stream URLs embed access tokens
/// (googlevideo `sig`, tidal tokens) and must never reach the log.
fn display_source(source_str: &str) -> String {
    match Url::parse(source_str) {
        Ok(url) if matches!(url.scheme(), "http" | "https") => {
            format!("{}://{}/…", url.scheme(), url.host_str().unwrap_or("?"))
        }
        _ => source_str.to_string(),
    }
}
/// Interval between elapsed-position updates while playing.
const TICK_INTERVAL: Duration = Duration::from_millis(250);

pub enum PlayerEngineCommand {
    Play(String, Sender<Result<MediaInfo>>),
    SetVolume(f32, Sender<f32>),
    Pause(Sender<Result<()>>),
    Unpause(Sender<Result<()>>),
    TogglePlay(Sender<Result<bool>>),
    Restart(Sender<Result<MediaInfo>>),
    Stop(Sender<Result<()>>),
    GetDuration(Sender<Result<Duration>>),
    GetElapsed(Sender<Result<Duration>>),
    SeekTo(Duration, Sender<Result<Duration>>),
    GetVolume(Sender<f32>),
    GetPaused(Sender<Result<bool>>),
    /// End of stream for the source started by the given generation.
    /// Stale generations are ignored so an old track finishing can never
    /// interfere with a newly started one.
    Eos(u64),
}

pub enum PlayerMessage {
    Duration {
        duration: Duration,
    },
    Elapsed {
        duration: Duration,
        elapsed: Duration,
    },
    Stopped,
    Paused,
    Playing,
    EndOfStream,
}

#[derive(Clone, Debug)]
pub struct MediaInfo {
    pub duration: Option<Duration>,
}

#[derive(Debug, Error)]
pub enum PlayerEngineError {
    #[error("Sink is not playing")]
    NotPlaying,
}

pub struct PlayerEngine {
    current_source: Option<String>,
    media_info: Option<MediaInfo>,
    /// Monotonically increasing id for the currently playing source. Used to
    /// discard end-of-stream callbacks from sources that were replaced.
    generation: u64,
    sink: rodio::Player,
    // We need to keep the device sink around; audio stops when it's dropped.
    _stream: MixerDeviceSink,
    tx_engine: Sender<PlayerEngineCommand>,
    tx_player: Sender<PlayerMessage>,
    runtime: tokio::runtime::Handle,
    // Present when the engine had to bring its own runtime because the
    // creating thread was not inside a tokio context.
    _owned_runtime: Option<tokio::runtime::Runtime>,
    /// Shared client for windowed network streams.
    http: reqwest::Client,
}

impl PlayerEngine {
    pub fn init(
        tx_engine: Sender<PlayerEngineCommand>,
        tx_player: Sender<PlayerMessage>,
        runtime: Option<tokio::runtime::Handle>,
    ) -> Result<Self> {
        let stream =
            DeviceSinkBuilder::open_default_sink().context("failed to open audio output device")?;
        let sink = rodio::Player::connect_new(stream.mixer());
        let (runtime, owned_runtime) = match runtime {
            Some(handle) => (handle, None),
            None => {
                let rt = tokio::runtime::Builder::new_multi_thread()
                    .worker_threads(1)
                    .enable_all()
                    .build()
                    .context("failed to create tokio runtime for the player engine")?;
                (rt.handle().clone(), Some(rt))
            }
        };
        info!("audio output device opened");
        let http = reqwest::Client::builder()
            .connect_timeout(Duration::from_secs(30))
            .build()
            .context("failed to build the http client")?;
        Ok(Self {
            current_source: None,
            media_info: None,
            generation: 0,
            sink,
            _stream: stream,
            tx_engine,
            tx_player,
            runtime,
            _owned_runtime: owned_runtime,
            http,
        })
    }

    /// Drives the engine until all command senders are dropped.
    pub fn run(mut self, rx_engine: Receiver<PlayerEngineCommand>) {
        loop {
            match rx_engine.recv_timeout(TICK_INTERVAL) {
                Ok(command) => self.handle_command(command),
                Err(RecvTimeoutError::Timeout) => self.tick(),
                Err(RecvTimeoutError::Disconnected) => {
                    debug!("player engine channel closed, shutting down");
                    break;
                }
            }
        }
    }

    fn handle_command(&mut self, command: PlayerEngineCommand) {
        match command {
            PlayerEngineCommand::Play(source_str, tx) => {
                send_reply(tx, self.play(&source_str));
            }
            PlayerEngineCommand::Pause(tx) => send_reply(tx, self.pause()),
            PlayerEngineCommand::Unpause(tx) => send_reply(tx, self.unpause()),
            PlayerEngineCommand::Stop(tx) => send_reply(tx, self.stop()),
            PlayerEngineCommand::TogglePlay(tx) => send_reply(tx, self.toggle_play()),
            PlayerEngineCommand::Restart(tx) => send_reply(tx, self.restart()),
            PlayerEngineCommand::GetDuration(tx) => send_reply(tx, self.duration()),
            PlayerEngineCommand::GetElapsed(tx) => send_reply(tx, self.elapsed()),
            PlayerEngineCommand::SeekTo(time, tx) => send_reply(tx, self.seek_to(time)),
            PlayerEngineCommand::SetVolume(volume, tx) => {
                send_reply(tx, self.set_volume(volume));
            }
            PlayerEngineCommand::GetVolume(tx) => send_reply(tx, self.volume()),
            PlayerEngineCommand::GetPaused(tx) => send_reply(tx, self.is_paused()),
            PlayerEngineCommand::Eos(generation) => self.handle_eos(generation),
        }
    }

    /// Emits an elapsed-position update while a source is playing.
    fn tick(&self) {
        if self.sink.empty() || self.sink.is_paused() {
            return;
        }
        let duration = self
            .media_info
            .as_ref()
            .and_then(|m| m.duration)
            .unwrap_or_default();
        // Dropping a tick when the channel is full is harmless.
        let _ = self.tx_player.try_send(PlayerMessage::Elapsed {
            duration,
            elapsed: self.sink.get_pos(),
        });
    }

    #[instrument(skip_all, fields(source = %display_source(source_str)))]
    pub fn play(&mut self, source_str: &str) -> Result<MediaInfo> {
        self.reset();

        let duration = self.start_source(source_str)?;
        let media_info = MediaInfo { duration };

        self.media_info = Some(media_info.clone());
        self.current_source = Some(source_str.to_string());

        self.notify(PlayerMessage::Duration {
            duration: duration.unwrap_or_default(),
        });

        self.sink.play();
        self.notify(PlayerMessage::Playing);
        debug!(duration = ?duration, "started playback");

        Ok(media_info)
    }

    /// Decodes the source and appends it (plus an end-of-stream callback) to
    /// the sink. Returns the total duration if known.
    fn start_source(&mut self, source_str: &str) -> Result<Option<Duration>> {
        self.generation += 1;
        let duration = match Url::parse(source_str) {
            Ok(url) if matches!(url.scheme(), "http" | "https") => {
                trace!(
                    host = url.host_str().unwrap_or("?"),
                    "opening network stream"
                );
                // Windowed fetching: some CDNs (googlevideo) reject plain
                // and open-ended requests with 403 and only serve bounded
                // ranges (see audio-player/src/windowed_http.rs).
                let params = WindowedHttpParams::new(url.clone(), self.http.clone());
                let reader = self.runtime.block_on(async {
                    tokio::time::timeout(
                        STREAM_OPEN_TIMEOUT,
                        StreamDownload::new::<WindowedHttpStream>(
                            params,
                            TempStorageProvider::new(),
                            Settings::default(),
                        ),
                    )
                    .await
                    .map_err(|_| anyhow!("timed out opening stream after {STREAM_OPEN_TIMEOUT:?}"))?
                    .context("failed to open http stream")
                })?;
                // Symphonia probes the container length during init; without a
                // known byte length it seeks from the end, which rodio 0.22 turns
                // into an `unreachable!` panic on a streamed source. Handing it the
                // content length up front avoids that seek entirely.
                let byte_len = reader.content_length();
                let mut builder = Decoder::builder().with_data(reader).with_seekable(true);
                if let Some(len) = byte_len {
                    builder = builder.with_byte_len(len);
                }
                if let Some(extension) = Path::new(url.path()).extension().and_then(|e| e.to_str())
                {
                    builder = builder.with_hint(extension);
                }
                let decoder = builder.build().context("failed to decode http stream")?;
                let duration = decoder.total_duration();
                self.sink.append(decoder);
                duration
            }
            Ok(url) => return Err(anyhow!("Not a valid URL scheme: {}", url.scheme())),
            Err(_) => {
                trace!(path = source_str, "opening local file");
                let file = File::open(source_str)
                    .with_context(|| format!("failed to open file {source_str}"))?;
                let byte_len = file.metadata().ok().map(|m| m.len());
                let mut builder = Decoder::builder()
                    .with_data(BufReader::new(file))
                    .with_seekable(true);
                if let Some(len) = byte_len {
                    builder = builder.with_byte_len(len);
                }
                if let Some(extension) = Path::new(source_str).extension().and_then(|e| e.to_str())
                {
                    builder = builder.with_hint(extension);
                }
                let decoder = builder.build().context("failed to decode file")?;
                let duration = decoder.total_duration();
                self.sink.append(decoder);
                duration
            }
        };

        // Fires only when the decoder ahead of it finished naturally; a
        // stop/replace clears the queue before this source is ever played.
        let tx_engine = self.tx_engine.clone();
        let generation = self.generation;
        self.sink.append(EmptyCallback::new(Box::new(move || {
            if let Err(err) = tx_engine.try_send(PlayerEngineCommand::Eos(generation)) {
                warn!("failed to send end-of-stream signal: {err}");
            }
        })));

        Ok(duration)
    }

    pub fn restart(&mut self) -> Result<MediaInfo> {
        if let Some(source) = self.current_source.clone() {
            return self.play(&source);
        }
        Err(PlayerEngineError::NotPlaying.into())
    }

    pub fn pause(&mut self) -> Result<()> {
        if self.is_stopped() {
            return Err(PlayerEngineError::NotPlaying.into());
        }
        self.sink.pause();
        self.notify(PlayerMessage::Paused);
        Ok(())
    }

    pub fn unpause(&mut self) -> Result<()> {
        if self.is_stopped() {
            return Err(PlayerEngineError::NotPlaying.into());
        }
        self.sink.play();
        self.notify(PlayerMessage::Playing);
        Ok(())
    }

    pub fn toggle_play(&mut self) -> Result<bool> {
        if self.is_stopped() {
            return Err(PlayerEngineError::NotPlaying.into());
        }
        if self.sink.is_paused() {
            self.sink.play();
            self.notify(PlayerMessage::Playing);
            Ok(true)
        } else {
            self.sink.pause();
            self.notify(PlayerMessage::Paused);
            Ok(false)
        }
    }

    pub fn stop(&mut self) -> Result<()> {
        if self.is_stopped() {
            return Err(PlayerEngineError::NotPlaying.into());
        }
        self.reset();
        self.notify(PlayerMessage::Stopped);
        Ok(())
    }

    pub fn is_paused(&self) -> Result<bool> {
        if self.is_stopped() {
            return Err(PlayerEngineError::NotPlaying.into());
        }
        Ok(self.sink.is_paused())
    }

    pub fn is_stopped(&self) -> bool {
        self.sink.empty()
    }

    pub fn duration(&self) -> Result<Duration> {
        self.media_info
            .as_ref()
            .map_or(Err(PlayerEngineError::NotPlaying.into()), |m| {
                Ok(m.duration.unwrap_or_default())
            })
    }

    pub fn elapsed(&self) -> Result<Duration> {
        if self.is_stopped() {
            return Err(PlayerEngineError::NotPlaying.into());
        }
        Ok(self.sink.get_pos())
    }

    pub fn seek_to(&self, time: Duration) -> Result<Duration> {
        let duration = self.duration().unwrap_or_else(|_| self.sink.get_pos());
        let time = time.clamp(Duration::from_secs(1), duration);
        self.sink
            .try_seek(time)
            .map_err(|err| anyhow!("seek failed: {err}"))?;
        Ok(self.sink.get_pos())
    }

    pub fn volume(&self) -> f32 {
        self.sink.volume()
    }

    pub fn set_volume(&mut self, volume: f32) -> f32 {
        self.sink.set_volume(volume.clamp(0.0, 1.1));
        self.sink.volume()
    }

    fn handle_eos(&mut self, generation: u64) {
        if generation != self.generation {
            debug!(
                stale = generation,
                current = self.generation,
                "ignoring end-of-stream from replaced source"
            );
            return;
        }
        debug!("end of stream");
        self.reset();
        self.notify(PlayerMessage::EndOfStream);
    }

    fn reset(&mut self) {
        self.current_source = None;
        self.media_info = None;
        self.generation += 1;
        self.sink.stop();
    }

    fn notify(&self, message: PlayerMessage) {
        self.tx_player
            .send(message)
            .unwrap_or_else(|e| warn!("Send error {}", e));
    }
}

fn send_reply<T>(tx: Sender<T>, value: T) {
    if tx.send(value).is_err() {
        warn!("player engine reply receiver dropped");
    }
}

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

    #[test]
    fn logged_sources_never_carry_url_tokens() {
        // Stream URLs embed access tokens; only scheme and host may be
        // logged. Local paths pass through verbatim.
        assert_eq!(
            display_source("https://rr1.googlevideo.com/videoplayback?sig=SECRET&x=1"),
            "https://rr1.googlevideo.com/…"
        );
        assert_eq!(
            display_source("/home/user/music/song.m4a"),
            "/home/user/music/song.m4a"
        );
    }
}