qobuz-qt/rust/src/player/decoder.rs

use anyhow::Result;
use std::io::{self, Read, Seek, SeekFrom};
use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc,
};

use symphonia::core::{
    codecs::DecoderOptions,
    errors::Error as SymphoniaError,
    formats::{FormatOptions, SeekMode, SeekTo},
    io::{MediaSource, MediaSourceStream},
    meta::MetadataOptions,
    probe::Hint,
    units::Time,
};

use super::{output::AudioOutput, PlayerCommand, PlayerStatus, TrackInfo};

/// First 512 KiB of stream kept in memory to support backward seeks during probing.
const HEAD_SIZE: usize = 512 * 1024;

struct HttpStreamSource {
    reader: reqwest::blocking::Response,
    head: Vec<u8>,
    reader_pos: u64,
    pos: u64,
    content_length: Option<u64>,
}

impl HttpStreamSource {
    fn new(response: reqwest::blocking::Response, content_length: Option<u64>) -> Self {
        Self {
            reader: response,
            head: Vec::new(),
            reader_pos: 0,
            pos: 0,
            content_length,
        }
    }
}

impl Read for HttpStreamSource {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let pos = self.pos as usize;
        if pos < self.head.len() {
            let avail = self.head.len() - pos;
            let n = buf.len().min(avail);
            buf[..n].copy_from_slice(&self.head[pos..pos + n]);
            self.pos += n as u64;
            return Ok(n);
        }

        let n = self.reader.read(buf)?;
        if n > 0 {
            if self.reader_pos < HEAD_SIZE as u64 {
                let capacity = HEAD_SIZE.saturating_sub(self.head.len());
                let to_buf = n.min(capacity);
                if to_buf > 0 {
                    self.head.extend_from_slice(&buf[..to_buf]);
                }
            }
            self.reader_pos += n as u64;
            self.pos += n as u64;
        }
        Ok(n)
    }
}

impl Seek for HttpStreamSource {
    fn seek(&mut self, from: SeekFrom) -> io::Result<u64> {
        let cl = self.content_length.unwrap_or(u64::MAX);
        let target: u64 = match from {
            SeekFrom::Start(n) => n,
            SeekFrom::End(n) if n < 0 => cl.saturating_sub((-n) as u64),
            SeekFrom::End(_) => cl,
            SeekFrom::Current(n) if n >= 0 => self.pos.saturating_add(n as u64),
            SeekFrom::Current(n) => self.pos.saturating_sub((-n) as u64),
        };

        if target == self.pos {
            return Ok(self.pos);
        }

        if target < self.reader_pos {
            if target < self.head.len() as u64 {
                self.pos = target;
                return Ok(self.pos);
            }
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "backward seek past head buffer",
            ));
        }

        let mut remaining = target - self.reader_pos;
        while remaining > 0 {
            let mut discard = [0u8; 8192];
            let want = (remaining as usize).min(discard.len());
            match self.reader.read(&mut discard[..want]) {
                Ok(0) => break,
                Ok(n) => {
                    if self.reader_pos < HEAD_SIZE as u64 {
                        let capacity = HEAD_SIZE.saturating_sub(self.head.len());
                        let to_buf = n.min(capacity);
                        if to_buf > 0 {
                            self.head.extend_from_slice(&discard[..to_buf]);
                        }
                    }
                    self.reader_pos += n as u64;
                    remaining -= n as u64;
                }
                Err(e) => return Err(e),
            }
        }
        self.pos = self.reader_pos;
        Ok(self.pos)
    }
}

impl MediaSource for HttpStreamSource {
    fn is_seekable(&self) -> bool {
        true
    }
    fn byte_len(&self) -> Option<u64> {
        self.content_length
    }
}

/// Decode and play `url` inline on the calling thread (the player loop).
///
/// `audio_output` is reused across calls if the sample rate and channel count match,
/// keeping the CPAL stream open between tracks for gapless playback.
///
/// Returns:
/// - `Ok(Some(TrackInfo))` — a new Play command arrived; start that track next.
/// - `Ok(None)`            — track finished naturally or was stopped.
/// - `Err(_)`              — unrecoverable playback error.
pub fn play_track_inline(
    url: &str,
    status: &PlayerStatus,
    paused: &Arc<AtomicBool>,
    audio_output: &mut Option<AudioOutput>,
    cmd_rx: &std::sync::mpsc::Receiver<PlayerCommand>,
) -> Result<Option<TrackInfo>> {
    let response = reqwest::blocking::get(url)?;
    let content_length = response.content_length();
    let source = HttpStreamSource::new(response, content_length);
    let mss = MediaSourceStream::new(Box::new(source), Default::default());

    let hint = Hint::new();
    let probed = symphonia::default::get_probe()
        .format(&hint, mss, &FormatOptions::default(), &MetadataOptions::default())
        .map_err(|e| anyhow::anyhow!("probe failed: {e}"))?;

    let mut format = probed.format;
    let track = format
        .tracks()
        .iter()
        .find(|t| t.codec_params.codec != symphonia::core::codecs::CODEC_TYPE_NULL)
        .ok_or_else(|| anyhow::anyhow!("no audio tracks"))?
        .clone();

    let track_id = track.id;
    let sample_rate = track.codec_params.sample_rate.unwrap_or(44100);
    let channels = track.codec_params.channels.map(|c| c.count()).unwrap_or(2);

    let mut decoder = symphonia::default::get_codecs()
        .make(&track.codec_params, &DecoderOptions::default())
        .map_err(|e| anyhow::anyhow!("decoder init failed: {e}"))?;

    // Reuse existing audio output if format matches; rebuild only on format change.
    if let Some(ao) = audio_output.as_ref() {
        if ao.sample_rate != sample_rate || ao.channels != channels {
            *audio_output = None; // will be recreated below
        }
    }
    if audio_output.is_none() {
        *audio_output = Some(AudioOutput::try_open(sample_rate, channels)?);
    }
    let ao = audio_output.as_mut().unwrap();

    let mut stopped = false;
    let mut next_track: Option<TrackInfo> = None;

    'decode: loop {
        // Non-blocking command check — handle Pause/Resume/Seek/Stop/Play
        loop {
            match cmd_rx.try_recv() {
                Ok(PlayerCommand::Pause) => {
                    paused.store(true, Ordering::SeqCst);
                    *status.state.lock().unwrap() = super::PlayerState::Paused;
                }
                Ok(PlayerCommand::Resume) => {
                    paused.store(false, Ordering::SeqCst);
                    *status.state.lock().unwrap() = super::PlayerState::Playing;
                }
                Ok(PlayerCommand::Seek(s)) => {
                    status.seek_target_secs.store(s, Ordering::Relaxed);
                    status.seek_requested.load(Ordering::SeqCst); // read-side fence
                    status.seek_requested.store(true, Ordering::SeqCst);
                }
                Ok(PlayerCommand::SetVolume(v)) => {
                    status.volume.store(v, Ordering::Relaxed);
                }
                Ok(PlayerCommand::Stop) => {
                    paused.store(false, Ordering::SeqCst);
                    *status.state.lock().unwrap() = super::PlayerState::Idle;
                    *status.current_track.lock().unwrap() = None;
                    status.position_secs.store(0, Ordering::Relaxed);
                    status.duration_secs.store(0, Ordering::Relaxed);
                    stopped = true;
                    break 'decode;
                }
                Ok(PlayerCommand::Play(info)) => {
                    // New track requested — stop current and return it
                    next_track = Some(info);
                    break 'decode;
                }
                Err(std::sync::mpsc::TryRecvError::Empty) => break,
                Err(std::sync::mpsc::TryRecvError::Disconnected) => {
                    stopped = true;
                    break 'decode;
                }
            }
        }

        // Spin while paused, but keep checking for commands
        while paused.load(Ordering::SeqCst) {
            std::thread::sleep(std::time::Duration::from_millis(10));
            // Still check for Stop/Play while paused
            match cmd_rx.try_recv() {
                Ok(PlayerCommand::Resume) => {
                    paused.store(false, Ordering::SeqCst);
                    *status.state.lock().unwrap() = super::PlayerState::Playing;
                }
                Ok(PlayerCommand::Stop) => {
                    paused.store(false, Ordering::SeqCst);
                    stopped = true;
                    break;
                }
                Ok(PlayerCommand::Play(info)) => {
                    paused.store(false, Ordering::SeqCst);
                    next_track = Some(info);
                    break 'decode;
                }
                Ok(PlayerCommand::SetVolume(v)) => {
                    status.volume.store(v, Ordering::Relaxed);
                }
                _ => {}
            }
            if stopped { break 'decode; }
        }
        if stopped { break; }

        // Handle seek
        if status.seek_requested.load(Ordering::SeqCst) {
            status.seek_requested.store(false, Ordering::SeqCst);
            let target = status.seek_target_secs.load(Ordering::Relaxed);
            let seeked = format.seek(
                SeekMode::Coarse,
                SeekTo::Time { time: Time::from(target), track_id: None },
            );
            if let Ok(s) = seeked {
                let actual = s.actual_ts / sample_rate as u64;
                status.position_secs.store(actual, Ordering::Relaxed);
            }
            decoder.reset();
            continue;
        }

        let packet = match format.next_packet() {
            Ok(p) => p,
            Err(SymphoniaError::IoError(e))
                if e.kind() == std::io::ErrorKind::UnexpectedEof =>
            {
                break; // natural end of track
            }
            Err(SymphoniaError::ResetRequired) => {
                decoder.reset();
                continue;
            }
            Err(e) => return Err(anyhow::anyhow!("format error: {e}")),
        };

        if packet.track_id() != track_id {
            continue;
        }

        if let Some(ts) = packet.ts().checked_div(sample_rate as u64) {
            status.position_secs.store(ts, Ordering::Relaxed);
        }

        match decoder.decode(&packet) {
            Ok(decoded) => {
                let volume = status.volume.load(Ordering::Relaxed) as f32 / 100.0;
                let rg = *status.replaygain_gain.lock().unwrap();
                // Use a stop flag tied to new-track-incoming so write doesn't block
                let dummy_stop = Arc::new(AtomicBool::new(false));
                ao.write(decoded, (volume * rg).min(1.0), &dummy_stop)?;
            }
            Err(SymphoniaError::IoError(_)) => break,
            Err(SymphoniaError::DecodeError(e)) => eprintln!("decode error: {e}"),
            Err(e) => return Err(anyhow::anyhow!("decode error: {e}")),
        }
    }

    if stopped {
        // On explicit stop, drop the audio output to silence immediately
        *audio_output = None;
    }

    Ok(next_track)
}