path: root/src/audio.rs

/*
 * This file is a part of RSPlayer
 * Licensed under the GPL-3.0 License
 * Copyright (C) 2026 Salmonik @ salmonik@tuta.io
 * https://codeberg.org/salmonik/RSPlayer
*/

use crate::notification;
use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, LazyLock, Mutex};
use std::thread;
use std::time::{Duration, Instant};
use symphonia::core::codecs::registry::CodecRegistry;
use symphonia_adapter_libopus::OpusDecoder;

static CODEC_REGISTRY: LazyLock<CodecRegistry> = LazyLock::new(|| {
    let mut registry = CodecRegistry::new();
    symphonia::default::register_enabled_codecs(&mut registry);
    registry.register_audio_decoder::<OpusDecoder>();
    registry
});

struct HttpMediaSource {
    reader: Mutex<Box<dyn Read + Send>>,
    buffer: Vec<u8>,
    position: usize,
    eof: bool,
}

impl HttpMediaSource {
    fn new(reader: Box<dyn Read + Send>) -> Self {
        HttpMediaSource {
            reader: Mutex::new(reader),
            buffer: Vec::new(),
            position: 0,
            eof: false,
        }
    }
}

impl Read for HttpMediaSource {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        if self.position < self.buffer.len() {
            let available = self.buffer.len() - self.position;
            let to_copy = buf.len().min(available);
            buf[..to_copy].copy_from_slice(&self.buffer[self.position..self.position + to_copy]);
            self.position += to_copy;
            return Ok(to_copy);
        }
        if self.eof {
            return Ok(0);
        }
        let n = self.reader.lock().unwrap().read(buf)?;
        if n > 0 {
            self.buffer.extend_from_slice(&buf[..n]);
            self.position += n;
        } else {
            self.eof = true;
        }
        Ok(n)
    }
}

impl Seek for HttpMediaSource {
    fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
        match pos {
            SeekFrom::Start(offset) => {
                let offset = offset as usize;
                while offset > self.buffer.len() && !self.eof {
                    let mut temp = [0u8; 8192];
                    let n = self.reader.lock().unwrap().read(&mut temp)?;
                    if n == 0 {
                        self.eof = true;
                    } else {
                        self.buffer.extend_from_slice(&temp[..n]);
                    }
                }
                self.position = offset.min(self.buffer.len());
                Ok(self.position as u64)
            }
            SeekFrom::End(_) => Err(std::io::Error::new(
                std::io::ErrorKind::Unsupported,
                "cannot seek from end on a stream",
            )),
            SeekFrom::Current(offset) => {
                let new_pos = if offset >= 0 {
                    self.position.saturating_add(offset as usize)
                } else {
                    self.position.saturating_sub((-offset) as usize)
                };
                while new_pos > self.buffer.len() && !self.eof {
                    let mut temp = [0u8; 8192];
                    let n = self.reader.lock().unwrap().read(&mut temp)?;
                    if n == 0 {
                        self.eof = true;
                    } else {
                        self.buffer.extend_from_slice(&temp[..n]);
                    }
                }
                self.position = new_pos.min(self.buffer.len());
                Ok(self.position as u64)
            }
        }
    }
}

impl symphonia::core::io::MediaSource for HttpMediaSource {
    fn is_seekable(&self) -> bool {
        true
    }

    fn byte_len(&self) -> Option<u64> {
        None
    }
}

#[allow(dead_code)]
pub enum AudioCommand {
    Play(PathBuf),
    Pause,
    Resume,
    Stop,
    Seek(f64),
    SetVolume(f32),
}

pub struct AudioEngine {
    command_tx: Sender<AudioCommand>,
    is_playing: Arc<AtomicBool>,
    pub current_track: Arc<Mutex<Option<std::path::PathBuf>>>,
    track_ended: Arc<AtomicBool>,
    volume: Arc<Mutex<f32>>,
    position: Arc<Mutex<f64>>,
}

#[allow(clippy::new_without_default)]
impl AudioEngine {
    pub fn new() -> Self {
        let (tx, rx) = mpsc::channel();
        let is_playing = Arc::new(AtomicBool::new(false));
        let is_playing_clone = Arc::clone(&is_playing);
        let current_track = Arc::new(Mutex::new(None));
        let current_track_clone = Arc::clone(&current_track);
        let track_ended = Arc::new(AtomicBool::new(false));
        let track_ended_clone = Arc::clone(&track_ended);

        let volume = Arc::new(Mutex::new(1.0f32));
        let volume_clone = Arc::clone(&volume);

        let position = Arc::new(Mutex::new(0.0));
        let position_clone = Arc::clone(&position);

        thread::spawn(move || {
            audio_thread(
                rx,
                is_playing_clone,
                current_track_clone,
                volume_clone,
                track_ended_clone,
                position_clone,
            );
        });

        AudioEngine {
            command_tx: tx,
            is_playing,
            current_track,
            track_ended,
            volume,
            position,
        }
    }

    pub fn play(&self, path: PathBuf) {
        self.track_ended.store(false, Ordering::SeqCst);
        if let Ok(mut track) = self.current_track.lock() {
            *track = Some(path.clone());
        }
        let _ = self.command_tx.send(AudioCommand::Play(path));
    }

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

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

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

    /// Jump to `position` seconds into the current track.
    pub fn seek(&self, position: f64) {
        let _ = self.command_tx.send(AudioCommand::Seek(position));
    }

    pub fn set_volume(&self, volume: f32) {
        let _ = self.command_tx.send(AudioCommand::SetVolume(volume));
    }

    pub fn get_volume(&self) -> u8 {
        self.volume
            .lock()
            .map(|v| (*v * 100.0) as u8)
            .unwrap_or(100)
    }

    pub fn is_playing(&self) -> bool {
        self.is_playing.load(Ordering::SeqCst)
    }

    pub fn current_track(&self) -> Option<std::path::PathBuf> {
        self.current_track.lock().ok().and_then(|g| g.clone())
    }

    pub fn poll_track_ended(&self) -> bool {
        self.track_ended.swap(false, Ordering::SeqCst)
    }

    pub fn position(&self) -> f64 {
        self.position.lock().map(|p| *p).unwrap_or(0.0)
    }
}

struct AudioData {
    format: Box<dyn symphonia::core::formats::FormatReader>,
    decoder: Box<dyn symphonia::core::codecs::audio::AudioDecoder>,
    sample_rate: u32,
    channels: u16,
    sample_buffer: Vec<f32>,
    buffer_pos: usize,
    resample_ratio: Option<f64>,
}

fn resample_linear(input: &[f32], ratio: f64, channels: usize) -> Vec<f32> {
    if ratio == 1.0 {
        return input.to_vec();
    }
    let input_frames = input.len() / channels;
    let output_frames = (input_frames as f64 * ratio).ceil() as usize;
    let mut output = Vec::with_capacity(output_frames * channels);
    for i in 0..output_frames {
        let input_pos = i as f64 / ratio;
        let idx = input_pos.floor() as usize;
        let frac = input_pos.fract() as f32;
        for ch in 0..channels {
            let cur = if idx * channels + ch < input.len() {
                input[idx * channels + ch]
            } else {
                0.0
            };
            let next = if (idx + 1) * channels + ch < input.len() {
                input[(idx + 1) * channels + ch]
            } else {
                cur
            };
            output.push(cur + frac * (next - cur));
        }
    }
    output
}

const WATCHDOG_INTERVAL: Duration = Duration::from_millis(250);
/// A playing stream must fill a buffer every few milliseconds, so a gap this
/// long means the backend stopped driving the callback entirely.
const STALL_TIMEOUT: Duration = Duration::from_secs(2);
/// A stream that stayed up at least this long is considered healthy: its
/// failure does not count against the retry budget.
const HEALTHY_STREAM: Duration = Duration::from_secs(15);
const MAX_RECOVERIES: usize = 3;
const RECOVERY_WINDOW: Duration = Duration::from_secs(60);
/// Delay before rebuilding after a stream died almost immediately. Doubled per
/// consecutive failure, so a device that is really gone is not hammered.
const RETRY_BACKOFF: Duration = Duration::from_millis(250);
const MAX_BACKOFF: Duration = Duration::from_secs(2);

struct StreamContext {
    audio_data: Arc<Mutex<Option<AudioData>>>,
    volume: Arc<Mutex<f32>>,
    track_ended: Arc<AtomicBool>,
    is_playing: Arc<AtomicBool>,
    position: Arc<Mutex<f64>>,
    failed: Arc<AtomicBool>,
    callbacks: Arc<AtomicU64>,
    /// Error callbacks seen on the current stream. A dead PCM fires the error
    /// callback hundreds of thousands of times per second, so it is only
    /// counted; the tally is reported once, when the stream is torn down.
    stream_errors: Arc<AtomicU64>,
    /// Message of the first error on the current stream, formatted once.
    first_error: Arc<Mutex<Option<String>>>,
}

fn make_data_callback(
    ctx: &StreamContext,
    device_sample_rate: u32,
) -> impl FnMut(&mut [f32], &cpal::OutputCallbackInfo) + Send + 'static {
    let audio_data = Arc::clone(&ctx.audio_data);
    let volume = Arc::clone(&ctx.volume);
    let track_ended = Arc::clone(&ctx.track_ended);
    let is_playing = Arc::clone(&ctx.is_playing);
    let position = Arc::clone(&ctx.position);
    let callbacks = Arc::clone(&ctx.callbacks);

    move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
        // Liveness marker for the watchdog: a relaxed increment is enough and
        // avoids a clock read in the realtime callback.
        callbacks.fetch_add(1, Ordering::Relaxed);

        let vol = volume.lock().map(|v| *v).unwrap_or(1.0);

        if let Ok(mut audio_data) = audio_data.lock() {
            let mut should_clear = false;

            if let Some(ref mut state) = *audio_data {
                let channels = state.channels as usize;
                let mut offset = 0usize;

                // Keep decoding until we fill the buffer or run out of data
                while offset < data.len() {
                    // First, copy any buffered samples
                    if state.buffer_pos < state.sample_buffer.len() {
                        let remaining = &state.sample_buffer[state.buffer_pos..];
                        let to_copy = remaining.len().min(data.len() - offset);
                        data[offset..offset + to_copy].copy_from_slice(&remaining[..to_copy]);
                        state.buffer_pos += to_copy;
                        offset += to_copy;

                        // If we've consumed all buffered samples, clear the buffer
                        if state.buffer_pos >= state.sample_buffer.len() {
                            state.sample_buffer.clear();
                            state.buffer_pos = 0;
                        }
                        continue; // Try to fill more of the buffer
                    }

                    // Decode more data
                    match state.format.next_packet() {
                        Ok(Some(packet)) => {
                            match state.decoder.decode(&packet) {
                                Ok(decoded) => {
                                    let num_samples = decoded.samples_interleaved();
                                    let mut samples = vec![0.0f32; num_samples];
                                    decoded.copy_to_slice_interleaved(&mut samples);

                                    // Resample if needed
                                    let resampled = if let Some(ratio) = state.resample_ratio {
                                        resample_linear(&samples, ratio, channels)
                                    } else {
                                        samples
                                    };

                                    // Copy what we can to output
                                    let to_copy = resampled.len().min(data.len() - offset);
                                    data[offset..offset + to_copy]
                                        .copy_from_slice(&resampled[..to_copy]);
                                    offset += to_copy;

                                    // Buffer the rest
                                    if to_copy < resampled.len() {
                                        state.sample_buffer.clear();
                                        state
                                            .sample_buffer
                                            .extend_from_slice(&resampled[to_copy..]);
                                        state.buffer_pos = 0;
                                    }
                                }
                                Err(_) => break,
                            }
                        }
                        _ => break,
                    }
                }

                // Apply volume
                for sample in data.iter_mut().take(offset) {
                    *sample *= vol;
                }

                // Fill remaining with silence
                for sample in data.iter_mut().skip(offset) {
                    *sample = 0.0;
                }

                // Track playback position
                if offset > 0
                    && let Ok(mut pos) = position.lock()
                {
                    *pos += offset as f64 / (device_sample_rate as f64 * channels as f64);
                }

                // If no more data and offset is 0, track ended
                if offset == 0 && state.sample_buffer.is_empty() {
                    track_ended.store(true, Ordering::SeqCst);
                    is_playing.store(false, Ordering::SeqCst);
                    should_clear = true;
                }
            } else {
                for sample in data.iter_mut() {
                    *sample = 0.0;
                }
            }

            if should_clear {
                *audio_data = None;
            }
        }
    }
}

fn build_output_stream(
    device: &cpal::Device,
    config: cpal::SupportedStreamConfig,
    ctx: &StreamContext,
) -> Result<cpal::Stream, String> {
    use cpal::traits::{DeviceTrait, StreamTrait};

    let device_sample_rate = config.sample_rate();
    let stream_config: cpal::StreamConfig = config.into();
    let failed = Arc::clone(&ctx.failed);
    let errors = Arc::clone(&ctx.stream_errors);
    let first_error = Arc::clone(&ctx.first_error);

    // Every stream gets a fresh tally, so a report only covers this one.
    errors.store(0, Ordering::Relaxed);
    if let Ok(mut slot) = first_error.lock() {
        *slot = None;
    }
    let error_seen = AtomicBool::new(false);

    let stream = device
        .build_output_stream(
            &stream_config,
            make_data_callback(ctx, device_sample_rate),
            move |err: cpal::StreamError| {
                // Buffer underruns are reported but already recovered by the
                // backend, so they must not tear down a working stream.
                if matches!(err, cpal::StreamError::BufferUnderrun) {
                    if errors.fetch_add(1, Ordering::Relaxed) == 0 {
                        eprintln!("Audio buffer underrun");
                    }
                    return;
                }
                failed.store(true, Ordering::SeqCst);
                // A broken PCM spins on `poll()` and reports the same error
                // again immediately: keep this path free of allocation and
                // logging, and format the message only for the first one.
                errors.fetch_add(1, Ordering::Relaxed);
                if !error_seen.swap(true, Ordering::Relaxed)
                    && let Ok(mut slot) = first_error.lock()
                {
                    *slot = Some(err.to_string());
                }
            },
            None,
        )
        .map_err(|e| e.to_string())?;

    stream.play().map_err(|e| e.to_string())?;

    Ok(stream)
}

/// Reports the error that tore the current stream down.
///
/// The error callback only counts, because a dead backend calls it hundreds of
/// thousands of times per second; the message and the tally are printed here,
/// once, from the watchdog thread.
fn log_stream_error(ctx: &StreamContext) {
    let count = ctx.stream_errors.swap(0, Ordering::Relaxed);
    let Ok(mut slot) = ctx.first_error.lock() else {
        return;
    };
    let Some(message) = slot.take() else {
        return;
    };
    if count > 1 {
        eprintln!("Stream error: {} ({} callbacks)", message, count);
    } else {
        eprintln!("Stream error: {}", message);
    }
}

/// Stops playback of the current track after the output device stopped working,
/// so the queue can move on instead of reporting "Playing" in silence.
fn abandon_playback(ctx: &StreamContext) {
    if let Ok(mut d) = ctx.audio_data.lock() {
        *d = None;
    }
    ctx.is_playing.store(false, Ordering::SeqCst);
    ctx.track_ended.store(true, Ordering::SeqCst);
    notification::notify_error("Audio output unavailable");
}

/// Stops trying once the retry budget is spent.
fn give_up(recovery: &mut RecoveryState, ctx: &StreamContext) {
    eprintln!(
        "Audio stream failed {} times within {}s, giving up",
        MAX_RECOVERIES,
        RECOVERY_WINDOW.as_secs()
    );
    recovery.retry_at = None;
    abandon_playback(ctx);
}

/// Decides how the audio thread recovers from a stream the backend broke.
///
/// cpal's ALSA backend cannot revive a dead PCM: `poll()` keeps returning
/// `POLLERR`, the data callback is never called again, and the worker thread
/// spins at full speed. The only way out is to drop the stream and build a new
/// one; the decoded audio lives in [`StreamContext::audio_data`], so playback
/// continues from where it stopped.
///
/// Rebuilding is cheap, but a device that is really gone would be hammered in a
/// loop, so failures are budgeted:
///
/// * the first failure of a healthy stream is retried immediately,
/// * consecutive failures back off further each time,
/// * a stream that played for [`HEALTHY_STREAM`] clears the budget, and
/// * [`MAX_RECOVERIES`] failures inside [`RECOVERY_WINDOW`] give up.
#[derive(Default)]
struct RecoveryState {
    /// When the current stream was built.
    stream_started: Option<Instant>,
    /// Failures inside [`RECOVERY_WINDOW`].
    failures: Vec<Instant>,
    /// Failures since the last stream that played long enough to count.
    consecutive: usize,
    /// Earliest instant at which the next rebuild may start.
    retry_at: Option<Instant>,
}

impl RecoveryState {
    /// Forgets the whole budget, for a new track or an explicit stop.
    fn clear(&mut self) {
        *self = Self::default();
    }

    fn stream_built(&mut self, now: Instant) {
        self.stream_started = Some(now);
    }

    /// Records a failure and returns how long to wait before rebuilding, or
    /// `None` when the backend has failed too often to keep trying.
    fn failed(&mut self, now: Instant) -> Option<Duration> {
        let uptime = self
            .stream_started
            .map_or(Duration::ZERO, |t| now.saturating_duration_since(t));
        self.failures
            .retain(|t| now.saturating_duration_since(*t) < RECOVERY_WINDOW);
        if uptime >= HEALTHY_STREAM {
            self.failures.clear();
            self.consecutive = 0;
        }
        if self.failures.len() >= MAX_RECOVERIES {
            return None;
        }
        self.failures.push(now);
        // The first retry is immediate: the outage is already audible, and a
        // healthy device comes back without a gap. Everything after that backs
        // off, because retrying straight away clearly did not help.
        let delay = if self.consecutive == 0 {
            Duration::ZERO
        } else {
            let shift = (self.consecutive as u32 - 1).min(4);
            (RETRY_BACKOFF * 2u32.pow(shift)).min(MAX_BACKOFF)
        };
        self.consecutive += 1;
        self.retry_at = Some(now + delay);
        Some(delay)
    }
}

/// Rebuilds the output stream after the backend broke it.
///
/// The default output can change while the daemon runs (a re-plugged DAC, a
/// sink switch, a rate change), which leaves the cached device handle pointing
/// at a PCM that no longer exists, so it is looked up again.
fn rebuild_stream(
    host: &cpal::Host,
    device: &mut cpal::Device,
    ctx: &StreamContext,
) -> Result<cpal::Stream, String> {
    use cpal::traits::{DeviceTrait, HostTrait};

    if let Some(fresh) = host.default_output_device() {
        *device = fresh;
    }

    let config = device
        .default_output_config()
        .map_err(|e| format!("Failed to get output config while rebuilding: {}", e))?;
    let device_sample_rate = config.sample_rate();

    // The device may report a different rate now than when the track was
    // opened, so re-target the resampler before resuming.
    if let Ok(mut d) = ctx.audio_data.lock()
        && let Some(state) = d.as_mut()
        && state.sample_rate != device_sample_rate
    {
        crate::debug_eprintln(format_args!(
            "Retargeting resampler from {} Hz to {} Hz after recovery",
            state.sample_rate, device_sample_rate
        ));
        state.resample_ratio = Some(device_sample_rate as f64 / state.sample_rate as f64);
    }

    build_output_stream(device, config, ctx)
}

fn audio_thread(
    rx: Receiver<AudioCommand>,
    is_playing: Arc<AtomicBool>,
    current_track: Arc<Mutex<Option<std::path::PathBuf>>>,
    volume: Arc<Mutex<f32>>,
    track_ended: Arc<AtomicBool>,
    position: Arc<Mutex<f64>>,
) {
    use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};

    let host = cpal::default_host();
    let mut device = match host.default_output_device() {
        Some(d) => d,
        None => {
            eprintln!("No output device available");
            return;
        }
    };

    let mut current_stream: Option<cpal::Stream> = None;
    let ctx = StreamContext {
        audio_data: Arc::new(Mutex::new(None)),
        volume: Arc::clone(&volume),
        track_ended: Arc::clone(&track_ended),
        is_playing: Arc::clone(&is_playing),
        position: Arc::clone(&position),
        failed: Arc::new(AtomicBool::new(false)),
        callbacks: Arc::new(AtomicU64::new(0)),
        stream_errors: Arc::new(AtomicU64::new(0)),
        first_error: Arc::new(Mutex::new(None)),
    };

    let mut recovery = RecoveryState::default();
    let mut last_callbacks = 0u64;
    let mut last_progress = Instant::now();

    loop {
        // Wake up early when a rebuild is due, so the backoff does not add
        // jitter on top of the watchdog tick.
        let wait = if current_stream.is_none() && is_playing.load(Ordering::SeqCst) {
            match recovery.retry_at {
                Some(at) => WATCHDOG_INTERVAL.min(at.saturating_duration_since(Instant::now())),
                None => WATCHDOG_INTERVAL,
            }
        } else {
            WATCHDOG_INTERVAL
        };

        let cmd = match rx.recv_timeout(wait) {
            Ok(cmd) => Some(cmd),
            Err(mpsc::RecvTimeoutError::Timeout) => None,
            Err(mpsc::RecvTimeoutError::Disconnected) => break,
        };

        if let Some(cmd) = cmd {
            match cmd {
                AudioCommand::Play(path) => {
                    is_playing.store(true, Ordering::SeqCst);
                    if let Ok(mut track) = current_track.lock() {
                        *track = Some(path.clone());
                    }
                    if let Ok(mut pos) = position.lock() {
                        *pos = 0.0;
                    }

                    // Drop the current stream before creating a new one
                    current_stream = None;
                    recovery.clear();
                    ctx.failed.store(false, Ordering::SeqCst);

                    // Query the config every time instead of caching it at
                    // startup: the default device or its rate can change while
                    // the daemon runs.
                    let config = match device.default_output_config() {
                        Ok(c) => c,
                        Err(e) => {
                            eprintln!("Failed to get output config: {}", e);
                            notification::notify_error("Failed to query audio output");
                            is_playing.store(false, Ordering::SeqCst);
                            continue;
                        }
                    };
                    let device_sample_rate = config.sample_rate();
                    crate::debug_eprintln(format_args!(
                        "Default output: {} Hz, {} channels",
                        device_sample_rate,
                        config.channels()
                    ));

                    match open_audio_file(&path) {
                        Ok(mut data) => {
                            crate::debug_eprintln(format_args!(
                                "File: {} Hz, {} channels",
                                data.sample_rate, data.channels
                            ));

                            if data.sample_rate != device_sample_rate {
                                crate::debug_eprintln(format_args!(
                                    "Resampling from {} Hz to {} Hz",
                                    data.sample_rate, device_sample_rate
                                ));
                                data.resample_ratio =
                                    Some(device_sample_rate as f64 / data.sample_rate as f64);
                            }

                            // Pre-decode some samples before starting playback to prevent buffer underrun
                            let channels = data.channels as usize;
                            for _ in 0..100 {
                                match data.format.next_packet() {
                                    Ok(Some(packet)) => match data.decoder.decode(&packet) {
                                        Ok(decoded) => {
                                            let num_samples = decoded.samples_interleaved();
                                            let mut samples = vec![0.0f32; num_samples];
                                            decoded.copy_to_slice_interleaved(&mut samples);

                                            let resampled = if let Some(ratio) = data.resample_ratio
                                            {
                                                resample_linear(&samples, ratio, channels)
                                            } else {
                                                samples
                                            };

                                            data.sample_buffer.extend(resampled);
                                        }
                                        Err(_) => break,
                                    },
                                    _ => break,
                                }
                            }
                            data.buffer_pos = 0;

                            if let Ok(mut d) = ctx.audio_data.lock() {
                                *d = Some(data);
                            }

                            match build_output_stream(&device, config, &ctx) {
                                Ok(stream) => {
                                    current_stream = Some(stream);
                                    recovery.stream_built(Instant::now());
                                }
                                Err(e) => {
                                    eprintln!("Failed to create stream: {}", e);
                                    notification::notify_error("Failed to create audio stream");
                                    is_playing.store(false, Ordering::SeqCst);
                                }
                            }
                        }
                        Err(e) => {
                            eprintln!("Failed to open audio: {}", e);
                            notification::notify_error("Failed to open audio file");
                            is_playing.store(false, Ordering::SeqCst);
                        }
                    }

                    // Opening a file (or an http stream) can take a while, so
                    // restart the stall window once the stream is up.
                    last_progress = Instant::now();
                }
                AudioCommand::Pause => {
                    if let Some(ref stream) = current_stream {
                        if let Err(e) = stream.pause() {
                            eprintln!("Failed to pause stream: {}", e);
                        }
                        is_playing.store(false, Ordering::SeqCst);
                    }
                }
                AudioCommand::Resume => {
                    if let Some(ref stream) = current_stream {
                        if let Err(e) = stream.play() {
                            eprintln!("Failed to resume stream: {}", e);
                        } else {
                            is_playing.store(true, Ordering::SeqCst);
                            last_progress = Instant::now();
                        }
                    } else if ctx.audio_data.lock().is_ok_and(|d| d.is_some()) {
                        // The stream was dropped while paused because the
                        // backend broke it; let the recovery path rebuild it
                        // instead of resuming into a dead stream.
                        is_playing.store(true, Ordering::SeqCst);
                        recovery.retry_at = Some(Instant::now());
                    }
                }
                AudioCommand::Stop => {
                    current_stream = None;
                    recovery.clear();
                    if let Ok(mut d) = ctx.audio_data.lock() {
                        *d = None;
                    }
                    is_playing.store(false, Ordering::SeqCst);
                    if let Ok(mut track) = current_track.lock() {
                        *track = None;
                    }
                }
                AudioCommand::Seek(seconds) => {
                    let target = seconds.max(0.0);
                    if let Ok(mut d) = ctx.audio_data.lock()
                        && let Some(data) = d.as_mut()
                    {
                        match seek_audio(data, target) {
                            Ok(()) => {
                                if let Ok(mut pos) = position.lock() {
                                    *pos = target;
                                }
                                crate::debug_eprintln(format_args!("Seeked to {:.1}s", target));
                            }
                            Err(e) => eprintln!("Failed to seek: {}", e),
                        }
                    }
                }
                AudioCommand::SetVolume(vol) => {
                    if let Ok(mut volume_ref) = volume.lock() {
                        *volume_ref = vol;
                    }
                }
            }
        }

        // Watchdog: the backend never brings a dead stream back on its own.
        // It either reports a fatal error and then stops feeding the callback
        // (`POLLERR` and friends), or - when the audio server is restarted or
        // changed underneath it - silently stops scheduling the callback.
        // Both look the same from the outside: playback is dead while the
        // daemon still reports "Playing", so react to either one at once.
        let failed = ctx.failed.swap(false, Ordering::SeqCst);
        let callbacks = ctx.callbacks.load(Ordering::Relaxed);
        let advanced = callbacks != last_callbacks;
        if advanced {
            last_callbacks = callbacks;
            last_progress = Instant::now();
        }
        let stalled = current_stream.is_some()
            && !advanced
            && is_playing.load(Ordering::SeqCst)
            && last_progress.elapsed() > STALL_TIMEOUT;

        if current_stream.is_some() && (failed || stalled) {
            if stalled {
                eprintln!("Audio stream stalled, rebuilding it");
            }
            // Dropping the stream is what stops the backend spinning on the
            // dead PCM, so do that before deciding whether to retry.
            current_stream = None;
            ctx.failed.store(false, Ordering::SeqCst);
            last_progress = Instant::now();
            log_stream_error(&ctx);

            if recovery.failed(Instant::now()).is_none() {
                give_up(&mut recovery, &ctx);
            }
        }

        // Rebuild once the backoff has elapsed. Playback has to be wanted and
        // the stream gone, otherwise a paused daemon would start audio.
        let retry_due = current_stream.is_none()
            && is_playing.load(Ordering::SeqCst)
            && recovery.retry_at.is_some_and(|at| at <= Instant::now());
        if retry_due {
            recovery.retry_at = None;
            match rebuild_stream(&host, &mut device, &ctx) {
                Ok(stream) => {
                    current_stream = Some(stream);
                    recovery.stream_built(Instant::now());
                    last_progress = Instant::now();
                }
                Err(e) => {
                    eprintln!("Failed to rebuild audio stream: {}", e);
                    if recovery.failed(Instant::now()).is_none() {
                        give_up(&mut recovery, &ctx);
                    }
                }
            }
        }
    }
}

/// Moves the decoder to `seconds` into the current track.
///
/// Symphonia's accurate seek lands on the packet *before* the requested
/// position, so packets short of the required timestamp are dropped before
/// playback resumes. The cpal stream itself keeps running: only the decoded
/// buffers are flushed, so there is no gap beyond the dropped packet.
fn seek_audio(data: &mut AudioData, seconds: f64) -> Result<(), String> {
    use symphonia::core::formats::{SeekMode, SeekTo};
    use symphonia::core::units::Time;

    let time = Time::try_from_secs_f64(seconds).ok_or("invalid seek position")?;
    let seeked = data
        .format
        .seek(
            SeekMode::Accurate,
            SeekTo::Time {
                time,
                track_id: None,
            },
        )
        .map_err(|e| e.to_string())?;

    data.decoder.reset();
    data.sample_buffer.clear();
    data.buffer_pos = 0;
    while let Ok(Some(packet)) = data.format.next_packet() {
        if packet.pts >= seeked.required_ts {
            break;
        }
    }

    Ok(())
}

fn open_audio_file(path: &PathBuf) -> Result<AudioData, Box<dyn std::error::Error>> {
    use symphonia::core::audio::layouts::CHANNEL_LAYOUT_STEREO;
    use symphonia::core::codecs::audio::AudioDecoderOptions;
    use symphonia::core::formats::probe::Hint;
    use symphonia::core::io::MediaSourceStream;

    let path_str = path.to_string_lossy();
    let is_url = path_str.starts_with("http://") || path_str.starts_with("https://");

    let mss: MediaSourceStream = if is_url {
        let client = reqwest::blocking::Client::builder()
            .user_agent("RSPlayer/0.8.2")
            .connect_timeout(std::time::Duration::from_secs(10))
            .build()
            .map_err(|e| format!("Failed to create HTTP client: {}", e))?;
        let response = client
            .get(path_str.as_ref())
            .send()
            .map_err(|e| format!("HTTP request failed: {}", e))?;
        let status = response.status();
        if !status.is_success() {
            return Err(format!("HTTP error {} for URL", status).into());
        }
        let reader = HttpMediaSource::new(Box::new(response));
        MediaSourceStream::new(Box::new(reader), Default::default())
    } else {
        let file = std::fs::File::open(path)?;
        MediaSourceStream::new(Box::new(file), Default::default())
    };

    let mut hint = Hint::new();
    if is_url
        && let Some(ext) = Path::new(path_str.as_ref())
            .extension()
            .and_then(|e| e.to_str())
    {
        hint.with_extension(ext);
    }
    let format = symphonia::default::get_probe().probe(
        &hint,
        mss,
        Default::default(),
        Default::default(),
    )?;

    let track = format
        .tracks()
        .iter()
        .find(|t| {
            t.codec_params
                .as_ref()
                .and_then(|p| p.audio())
                .and_then(|a| a.sample_rate)
                .is_some()
        })
        .ok_or("No suitable track found")?;

    let codec_params = track
        .codec_params
        .as_ref()
        .ok_or("No codec parameters")?
        .audio()
        .ok_or("Not an audio track")?;

    let sample_rate = codec_params.sample_rate.unwrap_or(44100);
    let channels = codec_params
        .channels
        .clone()
        .unwrap_or(CHANNEL_LAYOUT_STEREO)
        .count() as u16;

    let decoder =
        CODEC_REGISTRY.make_audio_decoder(codec_params, &AudioDecoderOptions::default())?;

    Ok(AudioData {
        format,
        decoder,
        sample_rate,
        channels,
        sample_buffer: Vec::new(),
        buffer_pos: 0,
        resample_ratio: None,
    })
}

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

    fn test_context() -> StreamContext {
        StreamContext {
            audio_data: Arc::new(Mutex::new(None)),
            volume: Arc::new(Mutex::new(1.0)),
            track_ended: Arc::new(AtomicBool::new(false)),
            is_playing: Arc::new(AtomicBool::new(true)),
            position: Arc::new(Mutex::new(0.0)),
            failed: Arc::new(AtomicBool::new(false)),
            callbacks: Arc::new(AtomicU64::new(0)),
            stream_errors: Arc::new(AtomicU64::new(0)),
            first_error: Arc::new(Mutex::new(None)),
        }
    }

    #[test]
    fn first_failure_of_a_healthy_stream_retries_immediately() {
        let mut state = RecoveryState::default();
        let start = Instant::now();
        state.stream_built(start);

        let delay = state.failed(start + Duration::from_secs(30));

        assert_eq!(delay, Some(Duration::ZERO), "retry without adding silence");
        assert_eq!(state.retry_at, Some(start + Duration::from_secs(30)));
    }

    #[test]
    fn fast_failures_back_off_then_give_up() {
        let mut state = RecoveryState::default();
        let start = Instant::now();
        let mut delays = Vec::new();

        for i in 0..4 {
            let at = start + Duration::from_millis(100 * i);
            state.stream_built(at);
            delays.push(state.failed(at + Duration::from_millis(50)));
        }

        assert_eq!(
            delays,
            vec![
                Some(Duration::ZERO),
                Some(RETRY_BACKOFF),
                Some(RETRY_BACKOFF * 2),
                None,
            ],
            "consecutive fast failures back off, then stop trying"
        );
    }

    #[test]
    fn healthy_stream_clears_the_budget() {
        let mut state = RecoveryState::default();
        let start = Instant::now();

        for i in 0..3 {
            state.stream_built(start);
            assert!(state.failed(start + Duration::from_secs(1 + i)).is_some());
        }

        // A stream that survived long enough must not inherit the strikes.
        let healthy = start + Duration::from_secs(10);
        state.stream_built(healthy);
        let delay = state.failed(healthy + HEALTHY_STREAM + Duration::from_secs(1));

        assert_eq!(delay, Some(Duration::ZERO));
        assert_eq!(state.failures.len(), 1);
        assert_eq!(state.consecutive, 1);
    }

    #[test]
    fn stale_failures_do_not_count() {
        let mut state = RecoveryState::default();
        let start = Instant::now();
        state.stream_built(start);
        state.failures = vec![
            start - RECOVERY_WINDOW - Duration::from_secs(5),
            start - RECOVERY_WINDOW - Duration::from_secs(4),
            start - RECOVERY_WINDOW - Duration::from_secs(3),
        ];

        assert!(state.failed(start).is_some());
        assert_eq!(state.failures.len(), 1);
    }

    #[test]
    fn gives_up_once_the_budget_is_spent() {
        let ctx = test_context();
        ctx.stream_errors.store(42, Ordering::Relaxed);
        if let Ok(mut slot) = ctx.first_error.lock() {
            *slot = Some("boom".to_string());
        }

        abandon_playback(&ctx);

        assert!(!ctx.is_playing.load(Ordering::SeqCst));
        assert!(ctx.track_ended.load(Ordering::SeqCst));
        assert!(ctx.audio_data.lock().expect("lock").is_none());

        // The tally is reported exactly once and does not leak into the next
        // stream.
        log_stream_error(&ctx);
        assert_eq!(ctx.stream_errors.load(Ordering::Relaxed), 0);
        assert!(ctx.first_error.lock().expect("lock").is_none());
    }

    #[test]
    fn rebuilds_stream_after_error() {
        use cpal::traits::{HostTrait, StreamTrait};

        let host = cpal::default_host();
        let Some(mut device) = host.default_output_device() else {
            eprintln!("skipped: no output device available");
            return;
        };

        let ctx = test_context();
        let Ok(stream) = rebuild_stream(&host, &mut device, &ctx) else {
            eprintln!("skipped: cannot build output stream");
            return;
        };

        stream.pause().expect("rebuilt stream is usable");
    }
}