koan_core/audio/

buffer.rs

use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::thread;

use symphonia::core::codecs::audio::well_known::{
    CODEC_ID_AAC, CODEC_ID_ALAC, CODEC_ID_FLAC, CODEC_ID_MP3, CODEC_ID_OPUS, CODEC_ID_PCM_F32LE,
    CODEC_ID_PCM_S16LE, CODEC_ID_PCM_S24LE, CODEC_ID_PCM_S32LE, CODEC_ID_VORBIS,
};
use symphonia::core::codecs::audio::{AudioCodecId, AudioCodecParameters, AudioDecoderOptions};
use symphonia::core::formats::probe::Hint;
use symphonia::core::formats::{FormatOptions, FormatReader, SeekMode, SeekTo, Track, TrackType};
use symphonia::core::io::MediaSourceStream;
use symphonia::core::meta::MetadataOptions;
use symphonia::core::units::{Duration, Time, TimeBase, Timestamp};
use thiserror::Error;

use crate::audio::opus::OpusBridge;
use crate::audio::viz::VizBuffer;
use crate::config::ReplayGainMode;
use crate::player::state::QueueItemId;

#[derive(Debug, Error)]
pub enum DecodeError {
    #[error("failed to open file: {0}")]
    Io(#[from] std::io::Error),
    #[error("no supported audio track found")]
    NoTrack,
    #[error("unsupported codec")]
    UnsupportedCodec,
    #[error("decode error: {0}")]
    Decode(String),
}

/// Info about the decoded audio stream, extracted before decoding starts.
#[derive(Debug, Clone)]
pub struct StreamInfo {
    pub codec: String,
    pub sample_rate: u32,
    pub channels: u16,
    pub bit_depth: Option<u16>,
    /// Bitrate in kbps. Meaningful for lossy codecs, None for lossless.
    pub bitrate_kbps: Option<u32>,
    pub duration_ms: u64,
}

/// Handle to a running decode thread. Drop to stop it.
pub struct DecodeHandle {
    stop: Arc<AtomicBool>,
    thread: Option<thread::JoinHandle<()>>,
}

impl DecodeHandle {
    /// Signal the decode thread to stop without waiting for it to exit.
    /// Unparks it too: a full ring has it parked for as long as the output
    /// takes to drain half of it.
    pub fn signal_stop(&self) {
        self.stop.store(true, Ordering::Relaxed);
        if let Some(handle) = &self.thread {
            handle.thread().unpark();
        }
    }

    /// Create a DecodeHandle with no real thread (for tests only).
    #[cfg(test)]
    pub fn new_for_test(stop: Arc<AtomicBool>) -> Self {
        Self { stop, thread: None }
    }

    /// Signal the decode thread to stop and wait for it.
    pub fn stop(&mut self) {
        self.signal_stop();
        if let Some(handle) = self.thread.take()
            && let Err(payload) = handle.join()
        {
            let msg = payload
                .downcast_ref::<String>()
                .map(|s| s.as_str())
                .or_else(|| payload.downcast_ref::<&str>().copied())
                .unwrap_or("unknown");
            log::error!("decode thread panicked: {}", msg);
        }
    }
}

impl Drop for DecodeHandle {
    fn drop(&mut self) {
        self.stop();
    }
}

// --- Playback timeline: the source of truth for "what's playing" ---

/// A track boundary in the playback stream. At `sample_offset` cumulative
/// samples written to the ring buffer, this track starts.
#[derive(Debug, Clone)]
pub struct TrackBoundary {
    pub id: QueueItemId,
    pub path: PathBuf,
    pub info: StreamInfo,
    /// Cumulative interleaved samples written to the ring buffer when this
    /// track's first sample was pushed. For the first track this is 0
    /// (or seek_samples if seeking).
    pub sample_offset: u64,
    /// Samples of this track's audio written to ring buffer so far.
    /// Updated as decode progresses. At EOF, equals total decoded samples.
    pub samples_written: u64,
    /// The seek offset in samples for this track (non-zero only if user seeked).
    pub seek_samples: u64,
}

impl TrackBoundary {
    /// Where in this track the playhead is when `played` samples have played.
    fn position_ms(&self, played: u64) -> Option<u64> {
        let ch = self.info.channels as u64;
        let rate = self.info.sample_rate as u64;
        if ch == 0 || rate == 0 {
            return None;
        }
        // Add seek offset since that's where playback started within the track.
        let track_samples = played.saturating_sub(self.sample_offset);
        Some((track_samples / ch) * 1000 / rate + (self.seek_samples / ch) * 1000 / rate)
    }
}

/// Shared timeline that the decode thread writes and the UI reads.
/// The decode thread appends boundaries; the UI reads them + samples_played
/// to derive current track and position.
pub struct PlaybackTimeline {
    boundaries: parking_lot::RwLock<Vec<TrackBoundary>>,
    /// Total interleaved samples written to the ring buffer across all tracks.
    samples_written: AtomicU64,
    /// Total interleaved samples consumed (played) by the audio engine.
    /// Written by the audio render callback, read by UI.
    pub samples_played: Arc<AtomicU64>,
    /// Incremented by every `reset()`. A decode thread writes only while the
    /// generation it started in is still the current one.
    generation: AtomicU64,
    /// Told when the decoder queues another track, which is when the moment
    /// the playhead reaches it becomes known.
    queued: parking_lot::Mutex<Option<Box<dyn Fn() + Send + Sync>>>,
}

impl std::fmt::Debug for PlaybackTimeline {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PlaybackTimeline")
            .field("samples_played", &self.samples_played)
            .field("generation", &self.generation)
            .finish_non_exhaustive()
    }
}

impl PlaybackTimeline {
    pub fn new() -> Arc<Self> {
        Arc::new(Self {
            boundaries: parking_lot::RwLock::new(Vec::new()),
            samples_written: AtomicU64::new(0),
            samples_played: Arc::new(AtomicU64::new(0)),
            generation: AtomicU64::new(0),
            queued: parking_lot::Mutex::new(None),
        })
    }

    /// Call `f` whenever the decoder queues another track.
    pub fn on_queued(&self, f: impl Fn() + Send + Sync + 'static) {
        *self.queued.lock() = Some(Box::new(f));
    }

    /// The track under the playhead and how far into it, without the clones
    /// `current_playback` makes.
    pub fn playhead(&self) -> Option<(QueueItemId, u64)> {
        let bounds = self.boundaries.read();
        let played = self.samples_played.load(Ordering::Acquire);
        let idx = bounds.partition_point(|b| b.sample_offset <= played);
        let current = bounds.get(idx.checked_sub(1)?)?;
        Some((current.id, current.position_ms(played)?))
    }

    /// How far into `id` playback has got, stopping at its end once the
    /// playhead has moved on to the next track.
    pub fn position_of(&self, id: QueueItemId) -> Option<u64> {
        let bounds = self.boundaries.read();
        let played = self.samples_played.load(Ordering::Acquire);
        let idx = bounds.iter().rposition(|b| b.id == id)?;
        let at = bounds
            .get(idx + 1)
            .map_or(played, |next| played.min(next.sample_offset));
        bounds[idx].position_ms(at)
    }

    /// How long until the playhead reaches the track queued after the one
    /// playing, at the rate this one plays. None with nothing queued.
    pub fn until_next_track(&self) -> Option<std::time::Duration> {
        let bounds = self.boundaries.read();
        let played = self.samples_played.load(Ordering::Acquire);
        let idx = bounds.partition_point(|b| b.sample_offset <= played);
        let next = bounds.get(idx)?;
        let current = bounds.get(idx.checked_sub(1)?)?;
        let per_second = current.info.sample_rate as u64 * current.info.channels as u64;
        if per_second == 0 {
            return None;
        }
        let left = next.sample_offset - played;
        Some(std::time::Duration::from_micros(
            left.saturating_mul(1_000_000) / per_second,
        ))
    }

    /// The current session's generation, to be handed to `writer`.
    ///
    /// Read on the player thread between `reset()` and spawning the decode
    /// thread, so a session can never capture a generation newer than its own.
    pub fn generation(&self) -> u64 {
        self.generation.load(Ordering::Acquire)
    }

    /// Open a write handle for the session identified by `generation`.
    pub fn writer(&self, generation: u64) -> TimelineWriter<'_> {
        TimelineWriter {
            timeline: self,
            generation,
        }
    }

    /// Reset for a new playback session.
    pub fn reset(&self) {
        // The generation bump happens under the boundary lock, which is the
        // same lock every guarded write takes — so a write either lands wholly
        // before this reset or sees the new generation and is dropped.
        let mut bounds = self.boundaries.write();
        self.generation.fetch_add(1, Ordering::AcqRel);
        bounds.clear();
        self.samples_written.store(0, Ordering::Relaxed);
        self.samples_played.store(0, Ordering::Relaxed);
    }

    /// Get a clone of the samples_played Arc for the audio engine.
    pub fn samples_played_counter(&self) -> Arc<AtomicU64> {
        self.samples_played.clone()
    }

    /// Derive current track info and position from the playback head.
    /// Returns (id, path, stream_info, position_ms).
    ///
    /// Acquires the boundaries read lock BEFORE reading `samples_played` so
    /// channels/sample_rate/boundaries are all from a consistent snapshot.
    /// Without this ordering, a track transition could update the atomics
    /// after we read `samples_played` but before we read the boundary list.
    pub fn current_playback(&self) -> Option<(QueueItemId, PathBuf, StreamInfo, u64)> {
        // Lock first — ensures we see boundaries consistent with the atomic read.
        let bounds = self.boundaries.read();

        if bounds.is_empty() {
            return None;
        }

        // Read samples_played while holding the lock. This guarantees we
        // never observe a stale boundary list with a newer samples_played
        // (or vice versa).
        let played = self.samples_played.load(Ordering::Acquire);

        // Find which track the playback head is in via binary search.
        // partition_point returns first index where offset > played;
        // the track we want is one before that.
        let idx = bounds.partition_point(|b| b.sample_offset <= played);
        let current = if idx > 0 {
            &bounds[idx - 1]
        } else {
            return None;
        };

        let position_ms = current.position_ms(played)?;

        Some((
            current.id,
            current.path.clone(),
            current.info.clone(),
            position_ms,
        ))
    }
}

/// One decode session's write access to the timeline.
///
/// `stop_engine` signals the decode thread and hands the join to a cleanup
/// thread, so the outgoing thread can still be mid-packet when `reset()` runs
/// and the next session starts. Without a guard its final `add_written` lands
/// on the fresh timeline and the first boundary of the new track gets stamped
/// at that offset instead of 0 — `current_playback()` then finds nothing at
/// `samples_played = 0` and the transport goes blank for ~50-100ms on every
/// skip and seek. A late `push_boundary` is worse: the wrong track's metadata
/// for the rest of the session.
///
/// Carrying the generation in the handle rather than passing it per call means
/// a write cannot be made with the wrong one.
pub struct TimelineWriter<'a> {
    timeline: &'a PlaybackTimeline,
    generation: u64,
}

impl TimelineWriter<'_> {
    /// False once another session has started. The decode thread polls this
    /// alongside its stop flag as a second abort signal.
    pub fn is_current(&self) -> bool {
        self.timeline.generation.load(Ordering::Acquire) == self.generation
    }

    /// Cumulative samples written to the ring buffer this session.
    fn samples_written(&self) -> u64 {
        self.timeline.samples_written.load(Ordering::Relaxed)
    }

    /// Called by decode thread when starting a new track.
    fn push_boundary(&self, boundary: TrackBoundary) {
        {
            let mut bounds = self.timeline.boundaries.write();
            if !self.is_current() {
                return;
            }
            bounds.push(boundary);
        }
        if let Some(queued) = self.timeline.queued.lock().as_ref() {
            queued();
        }
    }

    /// Called by decode thread after pushing samples.
    fn add_written(&self, count: u64) {
        let mut bounds = self.timeline.boundaries.write();
        if !self.is_current() {
            return;
        }
        self.timeline
            .samples_written
            .fetch_add(count, Ordering::Relaxed);
        // Also update the last boundary's samples_written.
        if let Some(last) = bounds.last_mut() {
            last.samples_written += count;
        }
    }
}

// ---------------------------------------------------------------------------
// Source abstraction
// ---------------------------------------------------------------------------

/// A source entry for the generic decode queue.
///
/// Each entry provides an ID, a display path (for logging/timeline),
/// a format hint, and a factory that constructs a fresh `MediaSourceStream`.
pub struct SourceEntry {
    pub id: QueueItemId,
    /// Path used for logging and `TrackBoundary`. Need not be a real FS path.
    pub path: PathBuf,
    /// Format hint for Symphonia (e.g. file extension).
    pub hint: Hint,
    /// Factory that creates the `MediaSourceStream`. Called exactly once per track.
    pub make_mss: Box<dyn FnOnce() -> std::io::Result<MediaSourceStream<'static>> + Send>,
}

impl SourceEntry {
    /// Convenience: build a `SourceEntry` from a local file path.
    pub fn from_file(id: QueueItemId, path: PathBuf) -> Self {
        let ext = path
            .extension()
            .and_then(|e| e.to_str())
            .unwrap_or("")
            .to_string();
        let path_clone = path.clone();
        let mut hint = Hint::new();
        if !ext.is_empty() {
            hint.with_extension(&ext);
        }
        Self {
            id,
            path,
            hint,
            make_mss: Box::new(move || {
                let file = File::open(&path_clone)?;
                Ok(MediaSourceStream::new(Box::new(file), Default::default()))
            }),
        }
    }
}

// ---------------------------------------------------------------------------
// Probe API
// ---------------------------------------------------------------------------

/// Probe a `MediaSourceStream` (with hint) and return stream info without decoding.
pub fn probe_source(mss: MediaSourceStream<'_>, hint: &Hint) -> Result<StreamInfo, DecodeError> {
    probe_mss(mss, hint)
}

/// Probe a file and return stream info without decoding.
pub fn probe_file(path: &Path) -> Result<StreamInfo, DecodeError> {
    let file_size = std::fs::metadata(path).ok().map(|m| m.len());
    let file = File::open(path)?;
    let mss = MediaSourceStream::new(Box::new(file), Default::default());
    let mut hint = Hint::new();
    if let Some(ext) = path.extension().and_then(|e| e.to_str()) {
        hint.with_extension(ext);
    }
    let mut info = probe_mss(mss, &hint)?;
    // For Opus (and other lossy codecs where symphonia couldn't give us a
    // bitrate), estimate from file size / duration when both are available.
    if info.bitrate_kbps.is_none()
        && info.bit_depth.is_none()
        && let Some(size) = file_size
        && info.duration_ms > 0
    {
        info.bitrate_kbps = Some((size * 8 / info.duration_ms) as u32);
    }
    Ok(info)
}

/// Probe a `MediaSourceStream` with a hint.
fn probe_mss(mss: MediaSourceStream<'_>, hint: &Hint) -> Result<StreamInfo, DecodeError> {
    let reader = symphonia::default::get_probe()
        .probe(
            hint,
            mss,
            FormatOptions::default(),
            MetadataOptions::default(),
        )
        .map_err(|e| match e {
            // Kept whole rather than flattened to a string: a probe against a
            // partial file fails by running out of bytes, and the caller has to
            // tell that apart from a file it cannot make sense of.
            symphonia::core::errors::Error::IoError(io) => DecodeError::Io(io),
            other => DecodeError::Decode(other.to_string()),
        })?;

    let track = reader
        .default_track(TrackType::Audio)
        .ok_or(DecodeError::NoTrack)?;
    let codec_params = track
        .codec_params
        .as_ref()
        .and_then(|p| p.audio())
        .ok_or(DecodeError::NoTrack)?;
    let is_opus = codec_params.codec == CODEC_ID_OPUS;
    // Opus always decodes to 48 kHz regardless of the input sample rate.
    let sample_rate = if is_opus {
        48000
    } else {
        codec_params.sample_rate.unwrap_or(44100)
    };
    let channels = codec_params
        .channels
        .as_ref()
        .map(|c| c.count() as u16)
        .unwrap_or(2);
    let bit_depth = if is_opus {
        None
    } else {
        Some(codec_params.bits_per_sample.unwrap_or(16) as u16)
    };
    let duration_ms = track_duration_ms(&*reader, track, sample_rate);
    let codec = codec_name(codec_params.codec);

    // Symphonia doesn't expose bitrate directly. For lossy codecs we can
    // estimate from bits_per_coded_sample when the demuxer provides it.
    // Opus estimation from file size is handled in probe_file() where we
    // have the path; here we only have a MediaSourceStream.
    let bitrate_kbps = estimate_bitrate_from_codec_params(codec_params);

    Ok(StreamInfo {
        codec,
        sample_rate,
        channels,
        bit_depth,
        bitrate_kbps,
        duration_ms,
    })
}

// ---------------------------------------------------------------------------
// Generic decode API (SourceEntry-based)
// ---------------------------------------------------------------------------

/// Start decoding from a `SourceEntry` into the ring buffer.
///
/// `first`      — the first track's source entry.
/// `seek_ms`    — if > 0, seek to this position before decoding the first track.
/// `next_track` — closure returning the next `SourceEntry` for gapless playback.
///                Called on EOF. Returns None when the playlist is exhausted.
#[allow(clippy::too_many_arguments)]
pub fn start_decode<N, F>(
    first: SourceEntry,
    producer: rtrb::Producer<f32>,
    seek_ms: u64,
    next_track: N,
    timeline: Arc<PlaybackTimeline>,
    viz_buffer: Option<Arc<VizBuffer>>,
    rg_mode: ReplayGainMode,
    pre_amp_db: f64,
    on_finished: F,
) -> Result<(StreamInfo, DecodeHandle), DecodeError>
where
    N: Fn() -> Option<SourceEntry> + Send + 'static,
    F: FnOnce() + Send + 'static,
{
    let stop = Arc::new(AtomicBool::new(false));
    let stop_clone = stop.clone();
    // Captured here rather than on the decode thread: the player has just
    // reset the timeline, and reading it before the spawn means this session
    // cannot pick up a generation belonging to a later one.
    let generation = timeline.generation();

    let thread = thread::Builder::new()
        .name("koan-decode".into())
        .spawn(move || {
            decode_queue_loop(
                first,
                producer,
                &stop_clone,
                seek_ms,
                &next_track,
                &timeline.writer(generation),
                viz_buffer.as_deref(),
                rg_mode,
                pre_amp_db,
            );
            // Notify the player that the decode loop finished (playlist
            // exhausted or error). Only fire if we weren't explicitly stopped
            // (i.e. this is a natural end, not a seek/skip teardown).
            if !stop_clone.load(Ordering::Relaxed) {
                on_finished();
            }
        })
        .map_err(DecodeError::Io)?;

    // Return a placeholder StreamInfo — the real info is pushed to the timeline
    // by the decode thread immediately after probing the source.
    let placeholder = StreamInfo {
        codec: String::from("?"),
        sample_rate: 44100,
        channels: 2,
        bit_depth: Some(16),
        bitrate_kbps: None,
        duration_ms: 0,
    };

    Ok((
        placeholder,
        DecodeHandle {
            stop,
            thread: Some(thread),
        },
    ))
}

// ---------------------------------------------------------------------------
// File-based convenience wrapper
// ---------------------------------------------------------------------------

/// Start decoding a file into the ring buffer (convenience wrapper).
///
/// `initial_id` — the QueueItemId of the first track.
/// `seek_ms` — if > 0, seek to this position before decoding the first track.
/// `next_track` — closure returning the next (id, path) for gapless playback.
#[allow(clippy::too_many_arguments)]
pub fn start_decode_file<N, F>(
    initial_id: QueueItemId,
    path: &Path,
    producer: rtrb::Producer<f32>,
    seek_ms: u64,
    next_track: N,
    timeline: Arc<PlaybackTimeline>,
    viz_buffer: Option<Arc<VizBuffer>>,
    rg_mode: ReplayGainMode,
    pre_amp_db: f64,
    on_finished: F,
) -> Result<(StreamInfo, DecodeHandle), DecodeError>
where
    N: Fn() -> Option<(QueueItemId, PathBuf)> + Send + 'static,
    F: FnOnce() + Send + 'static,
{
    let info = probe_file(path)?;
    let first = SourceEntry::from_file(initial_id, path.to_path_buf());
    let (_, handle) = start_decode(
        first,
        producer,
        seek_ms,
        move || {
            let (id, p) = next_track()?;
            Some(SourceEntry::from_file(id, p))
        },
        timeline,
        viz_buffer,
        rg_mode,
        pre_amp_db,
        on_finished,
    )?;
    Ok((info, handle))
}

// ---------------------------------------------------------------------------
// Internal decode loop
// ---------------------------------------------------------------------------

/// Sources that may fail to open or decode in a row before the session is
/// abandoned. One bad file must not end the queue, but a queue of nothing but
/// bad files still has to terminate.
const MAX_CONSECUTIVE_FAILURES: u32 = 32;

/// Gapless decode loop: decode first entry, then call next_track on EOF.
///
/// Every track in a session shares one ring buffer, and therefore the audio
/// engine configured for it. A track whose PCM format differs from the first
/// ends the session rather than being written at the wrong format; the player
/// restarts it on a correctly configured engine.
///
/// An unreadable source is skipped, not fatal — the decode head runs up to a
/// full ring buffer ahead of the DAC, so tearing down here would truncate the
/// track still being heard as well as dropping the rest of the queue.
#[allow(clippy::too_many_arguments)]
fn decode_queue_loop<N>(
    first: SourceEntry,
    mut producer: rtrb::Producer<f32>,
    stop: &AtomicBool,
    initial_seek_ms: u64,
    next_track: &N,
    timeline: &TimelineWriter<'_>,
    viz_buffer: Option<&VizBuffer>,
    rg_mode: ReplayGainMode,
    pre_amp_db: f64,
) where
    N: Fn() -> Option<SourceEntry>,
{
    // The delay line is indexed against the engine's played counter, which the
    // player has just reset for this session.
    if let Some(viz) = viz_buffer {
        viz.reset();
    }

    let mut pending = Some(first);
    let mut seek_ms = initial_seek_ms;
    let mut format: Option<PcmFormat> = None;
    let mut failures: u32 = 0;

    while let Some(entry) = pending.take() {
        if stop.load(Ordering::Relaxed) || !timeline.is_current() {
            break;
        }

        let SourceEntry {
            id,
            path,
            hint,
            make_mss,
        } = entry;

        let outcome = make_mss().map_err(DecodeError::Io).and_then(|mss| {
            decode_single(
                id,
                &path,
                &hint,
                mss,
                &mut producer,
                stop,
                seek_ms,
                timeline,
                viz_buffer,
                rg_mode,
                pre_amp_db,
                format,
            )
        });

        match outcome {
            Ok(Decoded::Complete(decoded_format)) => {
                format = Some(decoded_format);
                failures = 0;
            }
            Ok(Decoded::FormatMismatch) => break,
            Err(e) => {
                if stop.load(Ordering::Relaxed) {
                    break;
                }
                failures += 1;
                log::error!("skipping {}: {}", path.display(), e);
                if failures >= MAX_CONSECUTIVE_FAILURES {
                    log::error!(
                        "{} sources failed in a row, decode thread giving up",
                        failures
                    );
                    break;
                }
            }
        }
        // A stop ends the session wherever it lands; looking ahead would peek
        // the playlist and log a transition that never happens.
        if stop.load(Ordering::Relaxed) || !timeline.is_current() {
            break;
        }

        seek_ms = 0;
        pending = (next_track)();
        match pending {
            Some(ref next) => log::info!("gapless transition → {}", next.path.display()),
            None => log::info!("playlist exhausted, decode thread finishing"),
        }
    }

    wait_for_drain(&producer, stop, format);
}

/// How long the output takes to play `samples` interleaved samples. Without a
/// format to go by, a step short enough to poll with.
fn time_to_play(samples: usize, format: Option<PcmFormat>) -> std::time::Duration {
    match format {
        Some((rate, channels)) if rate > 0 && channels > 0 => std::time::Duration::from_micros(
            samples as u64 * 1_000_000 / (rate as u64 * channels as u64),
        ),
        _ => std::time::Duration::from_millis(2),
    }
}

/// Block until the audio engine has consumed everything in the ring buffer.
///
/// A session ends only once its audio has been heard, so the player can tear
/// the engine down without clipping the tail of the last track decoded.
/// Returns early if playback is torn down underneath us.
fn wait_for_drain(producer: &rtrb::Producer<f32>, stop: &AtomicBool, format: Option<PcmFormat>) {
    let capacity = producer.buffer().capacity();
    while !stop.load(Ordering::Relaxed) && !producer.is_abandoned() {
        let left = capacity.saturating_sub(producer.slots());
        if left == 0 {
            return;
        }
        thread::park_timeout(time_to_play(left, format));
    }
}

// ---------------------------------------------------------------------------
// Core decode single track
// ---------------------------------------------------------------------------

/// The PCM format of a decoded stream: sample rate in Hz and channel count.
/// The audio engine is configured from this, so the ring buffer may only ever
/// hold samples of one such format at a time.
type PcmFormat = (u32, u16);

/// Outcome of decoding one source.
enum Decoded {
    /// Decoded to EOF, in the given format.
    Complete(PcmFormat),
    /// The source's format differs from the stream already in the ring buffer.
    /// Nothing further was written — the engine must be reconfigured first.
    FormatMismatch,
}

/// Decode a single source into the producer. Returns on clean EOF.
///
/// `expected` — the format already in the ring buffer, if any. A source that
/// does not match it is rejected without writing samples or pushing a boundary.
#[allow(clippy::too_many_arguments)]
fn decode_single(
    queue_item_id: QueueItemId,
    path: &Path,
    hint: &Hint,
    mss: MediaSourceStream<'_>,
    producer: &mut rtrb::Producer<f32>,
    stop: &AtomicBool,
    seek_ms: u64,
    timeline: &TimelineWriter<'_>,
    viz_buffer: Option<&VizBuffer>,
    rg_mode: ReplayGainMode,
    pre_amp_db: f64,
    expected: Option<PcmFormat>,
) -> Result<Decoded, DecodeError> {
    let mut reader = symphonia::default::get_probe()
        .probe(
            hint,
            mss,
            FormatOptions::default(),
            MetadataOptions::default(),
        )
        .map_err(|e| DecodeError::Decode(e.to_string()))?;

    let track = reader
        .default_track(TrackType::Audio)
        .ok_or(DecodeError::NoTrack)?;
    let track_id = track.id;
    let time_base = track.time_base;
    let codec_params = track
        .codec_params
        .as_ref()
        .and_then(|p| p.audio())
        .ok_or(DecodeError::NoTrack)?;
    let is_opus_codec = codec_params.codec == CODEC_ID_OPUS;

    // Opus always decodes to 48 kHz regardless of the internal rate.
    let sample_rate = if is_opus_codec {
        48000
    } else {
        codec_params.sample_rate.unwrap_or(44100)
    };
    let channels = codec_params
        .channels
        .as_ref()
        .map(|c| c.count() as u16)
        .unwrap_or(2);

    let duration_ms = track_duration_ms(&*reader, track, sample_rate);

    // Try codec_params first; fall back to file-size estimation for Opus/lossy.
    let mut bitrate_kbps = estimate_bitrate_from_codec_params(codec_params);
    if bitrate_kbps.is_none()
        && is_opus_codec
        && let Ok(meta) = std::fs::metadata(path)
        && duration_ms > 0
    {
        bitrate_kbps = Some((meta.len() * 8 / duration_ms) as u32);
    }

    let info = StreamInfo {
        codec: codec_name(codec_params.codec),
        sample_rate,
        channels,
        bit_depth: if is_opus_codec {
            None
        } else {
            Some(codec_params.bits_per_sample.unwrap_or(16) as u16)
        },
        bitrate_kbps,
        duration_ms,
    };

    if let Some(expected) = expected
        && expected != (sample_rate, channels)
    {
        log::info!(
            "format change at {}: {}Hz/{}ch → {}Hz/{}ch, restarting audio engine",
            path.display(),
            expected.0,
            expected.1,
            sample_rate,
            channels
        );
        return Ok(Decoded::FormatMismatch);
    }

    // Build either a Symphonia decoder or our Opus bridge.
    let mut symphonia_decoder = if is_opus_codec {
        None
    } else {
        Some(
            symphonia::default::get_codecs()
                .make_audio_decoder(codec_params, &AudioDecoderOptions::default())
                .map_err(|_| DecodeError::UnsupportedCodec)?,
        )
    };
    let mut opus_bridge = if is_opus_codec {
        Some(OpusBridge::new(codec_params).map_err(|e| DecodeError::Decode(e.to_string()))?)
    } else {
        None
    };

    // Seek if requested (only for the first track usually).
    //
    // Accurate rather than coarse: a coarse seek picks a byte offset by
    // interpolating linearly over the file and then derives its reported
    // timestamp from that same guess, so on VBR MP3 it lands seconds from the
    // request *and* reports a position it never reached. Accurate mode walks
    // frame headers and is truthful about both, for 1.5-3ms on files up to
    // 79MB. The timeline records where playback actually resumed, so the
    // transport shows the position being heard.
    let mut seek_samples = 0;
    if seek_ms > 0 {
        let seeked = reader
            .seek(
                SeekMode::Accurate,
                SeekTo::Time {
                    time: Time::from_millis_u64(seek_ms),
                    track_id: Some(track_id),
                },
            )
            .map_err(|e| DecodeError::Decode(format!("seek failed: {}", e)))?;
        seek_samples = landing_samples(time_base, seeked.actual_ts, sample_rate, channels)
            .unwrap_or(seek_ms * sample_rate as u64 * channels as u64 / 1000);
        if let Some(ref mut dec) = symphonia_decoder {
            dec.reset();
        }
        if let Some(ref mut opus) = opus_bridge {
            opus.reset();
        }
    }

    // Record this track's boundary in the timeline.
    let write_offset = timeline.samples_written();
    timeline.push_boundary(TrackBoundary {
        id: queue_item_id,
        path: path.to_path_buf(),
        info,
        sample_offset: write_offset,
        samples_written: 0,
        seek_samples,
    });

    // Read ReplayGain tags and select the active gain for this track.
    let rg_gain = if rg_mode != ReplayGainMode::Off {
        match crate::audio::replaygain::read_tags(path) {
            Ok(rg_info) => {
                let selected = crate::audio::replaygain::select_gain(&rg_info, rg_mode);
                if let Some((gain_db, _)) = selected {
                    log::info!(
                        "replaygain: applying {:.2} dB ({:?}) to {}",
                        gain_db,
                        rg_mode,
                        path.display()
                    );
                }
                selected
            }
            Err(e) => {
                log::debug!("replaygain: no tags for {}: {}", path.display(), e);
                None
            }
        }
    } else {
        None
    };
    let mut rg_scratch: Vec<f32> = Vec::new();

    let mut sample_buf: Vec<f32> = Vec::new();

    loop {
        if stop.load(Ordering::Relaxed) || !timeline.is_current() {
            return Ok(Decoded::Complete((sample_rate, channels)));
        }

        let packet = match reader.next_packet() {
            Ok(Some(p)) => p,
            Ok(None) => return Ok(Decoded::Complete((sample_rate, channels))),
            Err(e) => return Err(DecodeError::Decode(e.to_string())),
        };

        if packet.track_id != track_id {
            continue;
        }

        // Decode the packet — either via Opus bridge or Symphonia codec.
        let samples: &[f32] = if let Some(ref mut opus) = opus_bridge {
            match opus.decode_packet(&packet.data) {
                Ok(s) => s,
                Err(e) => {
                    log::warn!("opus decode error (skipping packet): {}", e);
                    continue;
                }
            }
        } else {
            let decoder = symphonia_decoder.as_mut().unwrap();
            let decoded = match decoder.decode(&packet) {
                Ok(d) => d,
                Err(symphonia::core::errors::Error::DecodeError(e)) => {
                    log::warn!("decode error (skipping packet): {}", e);
                    continue;
                }
                Err(e) => return Err(DecodeError::Decode(e.to_string())),
            };

            let spec = decoded.spec();
            let (decoded_rate, decoded_channels) = (spec.rate(), spec.channels().count() as u16);
            // The engine is configured from the probed format. PCM that
            // disagrees with it would play at the wrong speed, so end the
            // session instead and let the player reconfigure.
            if (decoded_rate, decoded_channels) != (sample_rate, channels) {
                log::warn!(
                    "{}: decoded {}Hz/{}ch but stream declares {}Hz/{}ch, restarting audio engine",
                    path.display(),
                    decoded_rate,
                    decoded_channels,
                    sample_rate,
                    channels
                );
                return Ok(Decoded::FormatMismatch);
            }
            decoded.copy_to_vec_interleaved(&mut sample_buf);
            &sample_buf[..]
        };

        if samples.is_empty() {
            continue;
        }

        // Apply ReplayGain if active. Uses a reusable scratch buffer to avoid
        // allocating per packet. Zero overhead when RG is off.
        let samples = if let Some((gain_db, peak)) = rg_gain {
            rg_scratch.clear();
            rg_scratch.extend_from_slice(samples);
            crate::audio::replaygain::apply_gain(&mut rg_scratch, gain_db, peak, pre_amp_db);
            &rg_scratch[..]
        } else {
            samples
        };

        // Push samples into ring buffer, blocking if full.
        // VizBuffer is updated incrementally inside this loop so it receives
        // samples at the real-time audio consumption rate (paced by the audio
        // callback draining the rtrb consumer), not in packet-sized bursts.
        // Without this, FLAC packets (~93ms each at 44.1kHz) would update the
        // viz buffer only ~11 times/sec, making waveform modes visibly choppy.
        let mut offset = 0;
        while offset < samples.len() {
            // Also drops out on a stale generation, which keeps a dying thread
            // from pushing into the viz delay line the next session just reset.
            if stop.load(Ordering::Relaxed) || !timeline.is_current() {
                return Ok(Decoded::Complete((sample_rate, channels)));
            }

            let slots = producer.slots();
            if slots == 0 {
                // A full ring is the steady state, so this is the wait playback
                // spends nearly all its time in: until the output has played
                // half of it, which leaves the other half — a second or more at
                // any rate koan plays — to refill against. The viz delay line
                // is read at the playhead, so a refill arriving in one burst
                // reads the same as one trickling in. A stop unparks it.
                let half = producer.buffer().capacity() / 2;
                thread::park_timeout(time_to_play(half, Some((sample_rate, channels))));
                continue;
            }

            let chunk_size = slots.min(samples.len() - offset);
            if let Ok(mut chunk) = producer.write_chunk_uninit(chunk_size) {
                let to_write = &samples[offset..offset + chunk_size];
                let (first, second) = chunk.as_mut_slices();
                let first_len = first.len().min(to_write.len());
                for (slot, &val) in first.iter_mut().zip(&to_write[..first_len]) {
                    slot.write(val);
                }
                if first_len < to_write.len() {
                    for (slot, &val) in second.iter_mut().zip(&to_write[first_len..]) {
                        slot.write(val);
                    }
                }
                // SAFETY: All slots in the chunk have been initialized by the
                // two loops above — first.len() + second.len() == chunk_size,
                // and every slot is written via MaybeUninit::write().
                unsafe { chunk.commit_all() };

                // Feed viz buffer at the same rate as rtrb consumption.
                if let Some(viz) = viz_buffer {
                    viz.push_samples(to_write, channels, sample_rate);
                }

                offset += chunk_size;
            }
        }

        timeline.add_written(samples.len() as u64);
    }
}

/// Interleaved sample offset of a seek's landing point.
///
/// `actual_ts` is in the track's timebase, which is not always the reciprocal
/// of the sample rate (Matroska ticks in milliseconds), so it is converted
/// through `Time` rather than assumed to be a frame count.
fn landing_samples(
    time_base: Option<TimeBase>,
    actual_ts: Timestamp,
    sample_rate: u32,
    channels: u16,
) -> Option<u64> {
    let (seconds, nanos) = time_base?.calc_time(actual_ts)?.parts();
    let rate = sample_rate as u64;
    let frames = seconds.max(0) as u64 * rate + (nanos as u64 * rate) / 1_000_000_000;
    Some(frames * channels as u64)
}

/// Duration of a track in milliseconds.
///
/// The container's stated duration is authoritative because a track's timebase
/// is not always the reciprocal of the sample rate — Matroska ticks in
/// milliseconds, and states its duration at media level rather than per track.
/// Falls back to the playable frame count when no duration is stated at all.
pub(crate) fn track_duration_ms(
    reader: &(impl FormatReader + ?Sized),
    track: &Track,
    sample_rate: u32,
) -> u64 {
    fn to_ms(time_base: Option<TimeBase>, duration: Option<Duration>) -> Option<u64> {
        let time = time_base?.calc_duration(duration?)?;
        Some(time.as_millis().max(0) as u64)
    }

    let media = reader.media_info();
    to_ms(track.time_base, track.duration)
        .or_else(|| to_ms(media.time_base, media.duration))
        .or_else(|| {
            track
                .num_frames
                .map(|frames| frames * 1000 / sample_rate as u64)
        })
        .unwrap_or(0)
}

/// Estimate bitrate (kbps) from Symphonia codec parameters.
///
/// Symphonia doesn't expose a `bit_rate` field. For lossy codecs like MP3/AAC
/// we can derive it from `bits_per_coded_sample` when the demuxer populates it.
/// Returns `None` for lossless codecs or when the info isn't available.
fn estimate_bitrate_from_codec_params(params: &AudioCodecParameters) -> Option<u32> {
    let is_lossy = matches!(
        params.codec,
        CODEC_ID_MP3 | CODEC_ID_AAC | CODEC_ID_VORBIS | CODEC_ID_OPUS
    );
    if !is_lossy {
        return None;
    }

    // bits_per_coded_sample * sample_rate / 1000 gives kbps for CBR streams.
    // Few demuxers fill this in, but it's our best shot without file size.
    let bpcs = params.bits_per_coded_sample?;
    let sr = params.sample_rate?;
    let channels = params
        .channels
        .as_ref()
        .map(|c| c.count() as u32)
        .unwrap_or(2);
    Some(bpcs * sr * channels / 1000)
}

pub fn codec_name(codec: AudioCodecId) -> String {
    match codec {
        CODEC_ID_FLAC => "FLAC",
        CODEC_ID_MP3 => "MP3",
        CODEC_ID_AAC => "AAC",
        CODEC_ID_VORBIS => "Vorbis",
        CODEC_ID_OPUS => "Opus",
        CODEC_ID_ALAC => "ALAC",
        CODEC_ID_PCM_S16LE => "PCM/16",
        CODEC_ID_PCM_S24LE => "PCM/24",
        CODEC_ID_PCM_S32LE => "PCM/32",
        CODEC_ID_PCM_F32LE => "PCM/f32",
        other => return format!("Unknown({:?})", other),
    }
    .to_string()
}

#[cfg(test)]
mod tests {
    use std::path::PathBuf;
    use std::sync::atomic::Ordering;

    use super::*;
    use crate::player::state::QueueItemId;

    fn make_info(sample_rate: u32, channels: u16) -> StreamInfo {
        StreamInfo {
            codec: "FLAC".to_string(),
            sample_rate,
            channels,
            bit_depth: Some(16),
            bitrate_kbps: None,
            duration_ms: 10_000,
        }
    }

    fn make_boundary(
        id: QueueItemId,
        sample_offset: u64,
        seek_samples: u64,
        channels: u16,
        sample_rate: u32,
    ) -> TrackBoundary {
        TrackBoundary {
            id,
            path: PathBuf::from("/music/track.flac"),
            info: make_info(sample_rate, channels),
            sample_offset,
            samples_written: 0,
            seek_samples,
        }
    }

    /// A writer for the timeline's current generation.
    fn writer(timeline: &PlaybackTimeline) -> TimelineWriter<'_> {
        timeline.writer(timeline.generation())
    }

    // --- PlaybackTimeline tests ---

    #[test]
    fn test_timeline_single_track() {
        // Push one boundary at offset 0 with stereo 44100 Hz audio.
        // After simulating 44100 frames (88200 interleaved samples) played,
        // current_playback() should report track index 0 at position 1000 ms.
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let id = QueueItemId::new();
        // sample_offset=0, seek_samples=0, channels=2, sample_rate=44100
        tl.push_boundary(make_boundary(id, 0, 0, 2, 44100));
        tl.add_written(88200); // 1 second of audio

        // Simulate 1 second played: 44100 frames * 2 channels = 88200 interleaved samples
        timeline.samples_played.store(88200, Ordering::Relaxed);

        let result = timeline.current_playback();
        assert!(
            result.is_some(),
            "expected Some for single track with samples played"
        );
        let (result_id, _path, _info, position_ms) = result.unwrap();
        assert_eq!(result_id, id);
        assert_eq!(
            position_ms, 1000,
            "1 second of 44100 Hz stereo should be 1000 ms"
        );
    }

    #[test]
    fn test_timeline_gapless_transition() {
        // Two tracks in gapless sequence. Track 1 ends at sample 88200 (1 sec stereo 44100 Hz).
        // Track 2 begins at sample_offset 88200. When playback head is at 100000 (past the boundary),
        // current_playback() should report track 2.
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let id1 = QueueItemId::new();
        let id2 = QueueItemId::new();

        // Track 1: starts at offset 0
        tl.push_boundary(make_boundary(id1, 0, 0, 2, 44100));
        tl.add_written(88200);

        // Track 2: starts at offset 88200 (immediately after track 1's samples)
        tl.push_boundary(make_boundary(id2, 88200, 0, 2, 44100));
        tl.add_written(44100); // half a second of track 2

        // Set playback head past the track 1/2 boundary
        timeline.samples_played.store(90000, Ordering::Relaxed);

        let result = timeline.current_playback();
        assert!(result.is_some());
        let (result_id, _path, _info, position_ms) = result.unwrap();
        assert_eq!(
            result_id, id2,
            "playback head past boundary should report second track"
        );
        // (90000 - 88200) / 2 channels * 1000 / 44100 = 900 / 44100 ≈ 20 ms
        assert_eq!(position_ms, 20, "position within track 2 should be ~20 ms");
    }

    #[test]
    fn test_timeline_zero_samples() {
        // With 0 samples played and a boundary at offset 0, current_playback() should
        // still return the first track at position 0 ms.
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let id = QueueItemId::new();
        tl.push_boundary(make_boundary(id, 0, 0, 2, 44100));
        tl.add_written(1000);
        timeline.samples_played.store(0, Ordering::Relaxed);

        let result = timeline.current_playback();
        assert!(
            result.is_some(),
            "expected Some at 0 samples played with a boundary at offset 0"
        );
        let (result_id, _path, _info, position_ms) = result.unwrap();
        assert_eq!(result_id, id);
        assert_eq!(position_ms, 0);
    }

    #[test]
    fn test_timeline_past_all_boundaries() {
        // When samples_played exceeds all boundaries, the last track should be reported.
        // The binary search finds the last boundary whose sample_offset <= played.
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let id1 = QueueItemId::new();
        let id2 = QueueItemId::new();

        tl.push_boundary(make_boundary(id1, 0, 0, 2, 44100));
        tl.add_written(88200);
        tl.push_boundary(make_boundary(id2, 88200, 0, 2, 44100));
        tl.add_written(88200);

        // Simulate playback far past both tracks
        timeline
            .samples_played
            .store(999_999_999, Ordering::Relaxed);

        let result = timeline.current_playback();
        assert!(result.is_some());
        let (result_id, _path, _info, _position_ms) = result.unwrap();
        assert_eq!(
            result_id, id2,
            "samples past all boundaries should report the last track"
        );
    }

    #[test]
    fn test_timeline_seek_offset() {
        // When a seek offset is set, position_ms should include the seek position.
        // seek_samples = 88200 means playback started 1 second into the track.
        // With 0 additional samples played past the boundary, position should be 1000 ms.
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let id = QueueItemId::new();
        let seek_samples = 88200u64; // 1 second at 44100 Hz stereo
        tl.push_boundary(make_boundary(id, 0, seek_samples, 2, 44100));
        tl.add_written(44100); // half a second written so far
        // samples_played at the track boundary (0 frames past the track start)
        timeline.samples_played.store(0, Ordering::Relaxed);

        let result = timeline.current_playback();
        assert!(result.is_some());
        let (_result_id, _path, _info, position_ms) = result.unwrap();
        // track_samples = 0 - 0 = 0; seek contribution = (88200/2)*1000/44100 = 1000 ms
        assert_eq!(
            position_ms, 1000,
            "position should include seek offset of 1000 ms"
        );
    }

    #[test]
    fn the_next_track_is_due_when_the_current_one_has_played_out() {
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let (a, b) = (QueueItemId::new(), QueueItemId::new());
        // One second of 44.1kHz stereo before b starts.
        tl.push_boundary(make_boundary(a, 0, 0, 2, 44100));
        assert_eq!(timeline.until_next_track(), None, "nothing queued after it");
        tl.push_boundary(make_boundary(b, 88200, 0, 2, 44100));

        timeline.samples_played.store(44100, Ordering::Relaxed);
        assert_eq!(
            timeline.until_next_track(),
            Some(std::time::Duration::from_millis(500))
        );
        timeline.samples_played.store(88200, Ordering::Relaxed);
        assert_eq!(timeline.until_next_track(), None, "b is playing now");
    }

    #[test]
    fn the_queued_callback_hears_each_track() {
        let timeline = PlaybackTimeline::new();
        let heard = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let counter = heard.clone();
        timeline.on_queued(move || {
            counter.fetch_add(1, Ordering::Relaxed);
        });
        let tl = writer(&timeline);
        tl.push_boundary(make_boundary(QueueItemId::new(), 0, 0, 2, 44100));
        tl.push_boundary(make_boundary(QueueItemId::new(), 88200, 0, 2, 44100));
        assert_eq!(heard.load(Ordering::Relaxed), 2);
    }

    #[test]
    fn a_track_played_past_reads_as_played_to_its_end() {
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let (a, b) = (QueueItemId::new(), QueueItemId::new());
        tl.push_boundary(make_boundary(a, 0, 0, 2, 44100));
        tl.push_boundary(make_boundary(b, 88200, 0, 2, 44100));
        timeline
            .samples_played
            .store(88200 + 44100, Ordering::Relaxed);

        assert_eq!(timeline.position_of(a), Some(1000));
        assert_eq!(timeline.position_of(b), Some(500));
        assert_eq!(timeline.playhead(), Some((b, 500)));
    }

    #[test]
    fn test_timeline_reset() {
        // After reset(), current_playback() returns None and all counters are cleared.
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let id = QueueItemId::new();
        tl.push_boundary(make_boundary(id, 0, 0, 2, 44100));
        tl.add_written(88200);
        timeline.samples_played.store(44100, Ordering::Relaxed);

        // Sanity check: playback is live before reset
        assert!(timeline.current_playback().is_some());

        timeline.reset();

        assert!(
            timeline.current_playback().is_none(),
            "after reset, current_playback should return None"
        );
        assert_eq!(
            timeline.samples_played.load(Ordering::Relaxed),
            0,
            "samples_played should be 0 after reset"
        );
        assert_eq!(
            timeline.samples_written.load(Ordering::Relaxed),
            0,
            "samples_written should be 0 after reset"
        );
    }

    // --- Probe and decode integration tests ---

    #[test]
    fn probe_file_extracts_stream_info() {
        let dir = tempfile::tempdir().unwrap();
        let wav_path = dir.path().join("probe_test.wav");
        crate::test_utils::generate_wav(&wav_path, 44100, 2, 1.0, 16);

        let info = probe_file(&wav_path).expect("probe_file should succeed on a valid WAV");
        assert_eq!(info.sample_rate, 44100, "sample rate mismatch");
        assert_eq!(info.channels, 2, "channel count mismatch");
        assert_eq!(info.bit_depth, Some(16), "bit depth mismatch");
        assert!(
            info.duration_ms > 900 && info.duration_ms < 1100,
            "duration should be ~1000ms, got {}",
            info.duration_ms
        );
        assert!(
            info.codec.contains("PCM"),
            "codec should be PCM variant, got {}",
            info.codec
        );
    }

    #[test]
    fn decode_single_produces_samples() {
        let dir = tempfile::tempdir().unwrap();
        let wav_path = dir.path().join("tone.wav");
        // 440 Hz sine, mono, 0.1s — enough to verify non-zero decode output.
        crate::test_utils::generate_wav_tone(&wav_path, 44100, 440.0, 0.1);

        // Set up rtrb ring buffer.
        let (mut producer, mut consumer) = rtrb::RingBuffer::new(44100 * 2);

        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let stop = Arc::new(AtomicBool::new(false));

        let id = QueueItemId::new();
        let entry = SourceEntry::from_file(id, wav_path.clone());
        let hint = entry.hint.clone();
        let mss = (entry.make_mss)().expect("should open WAV file");

        let result = decode_single(
            id,
            &wav_path,
            &hint,
            mss,
            &mut producer,
            &stop,
            0,
            &tl,
            None,
            crate::config::ReplayGainMode::Off,
            0.0,
            None,
        );
        assert!(
            matches!(result, Ok(Decoded::Complete((44100, 1)))),
            "decode_single should complete at the source format"
        );

        // Read samples from the consumer side.
        let available = consumer.slots();
        assert!(available > 0, "expected samples in ring buffer, got 0");

        // Verify at least some samples are non-zero (it's a sine wave, not silence).
        let mut found_nonzero = false;
        while consumer.slots() > 0 {
            if let Ok(chunk) = consumer.read_chunk(consumer.slots().min(1024)) {
                let (first, second) = chunk.as_slices();
                for &s in first.iter().chain(second.iter()) {
                    if s.abs() > 0.001 {
                        found_nonzero = true;
                        break;
                    }
                }
                chunk.commit_all();
            }
            if found_nonzero {
                break;
            }
        }
        assert!(
            found_nonzero,
            "expected non-zero samples from 440Hz sine decode"
        );
    }

    // --- Ring buffer format contract ---

    /// Decode a queue of files through `decode_queue_loop` with a consumer
    /// draining in the background. Returns the boundaries the decode thread
    /// pushed onto the timeline.
    fn run_queue(paths: &[PathBuf]) -> Vec<TrackBoundary> {
        let (producer, mut consumer) = rtrb::RingBuffer::new(1 << 16);
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let stop = Arc::new(AtomicBool::new(false));

        let drain_stop = Arc::new(AtomicBool::new(false));
        let drain_flag = drain_stop.clone();
        let drainer = std::thread::spawn(move || {
            while !drain_flag.load(Ordering::Relaxed) {
                let n = consumer.slots();
                if n > 0
                    && let Ok(chunk) = consumer.read_chunk(n)
                {
                    chunk.commit_all();
                }
                std::thread::sleep(std::time::Duration::from_micros(200));
            }
        });

        let rest: std::sync::Mutex<Vec<PathBuf>> = std::sync::Mutex::new(paths[1..].to_vec());
        let next_track = move || {
            let mut rest = rest.lock().ok()?;
            if rest.is_empty() {
                return None;
            }
            Some(SourceEntry::from_file(QueueItemId::new(), rest.remove(0)))
        };

        decode_queue_loop(
            SourceEntry::from_file(QueueItemId::new(), paths[0].clone()),
            producer,
            &stop,
            0,
            &next_track,
            &tl,
            None,
            crate::config::ReplayGainMode::Off,
            0.0,
        );

        drain_stop.store(true, Ordering::Relaxed);
        drainer.join().unwrap();

        timeline.boundaries.read().clone()
    }

    #[test]
    fn gapless_continues_when_format_matches() {
        let dir = tempfile::tempdir().unwrap();
        let a = dir.path().join("a.wav");
        let b = dir.path().join("b.wav");
        crate::test_utils::generate_wav(&a, 44100, 2, 0.1, 16);
        crate::test_utils::generate_wav(&b, 44100, 2, 0.1, 16);

        let bounds = run_queue(&[a, b]);
        assert_eq!(
            bounds.len(),
            2,
            "same-format tracks should decode gaplessly"
        );
    }

    #[test]
    fn gapless_stops_at_sample_rate_change() {
        let dir = tempfile::tempdir().unwrap();
        let a = dir.path().join("a.wav");
        let b = dir.path().join("b.wav");
        crate::test_utils::generate_wav(&a, 44100, 2, 0.1, 16);
        crate::test_utils::generate_wav(&b, 48000, 2, 0.1, 16);

        let bounds = run_queue(&[a, b]);
        assert_eq!(
            bounds.len(),
            1,
            "a 48kHz track must not join a 44.1kHz ring buffer"
        );
        assert_eq!(bounds[0].info.sample_rate, 44100);
    }

    #[test]
    fn gapless_stops_at_channel_change() {
        let dir = tempfile::tempdir().unwrap();
        let a = dir.path().join("a.wav");
        let b = dir.path().join("b.wav");
        crate::test_utils::generate_wav(&a, 44100, 2, 0.1, 16);
        crate::test_utils::generate_wav(&b, 44100, 1, 0.1, 16);

        let bounds = run_queue(&[a, b]);
        assert_eq!(
            bounds.len(),
            1,
            "a mono track must not join a stereo ring buffer"
        );
        assert_eq!(bounds[0].info.channels, 2);
    }

    #[test]
    fn drain_waits_for_the_consumer() {
        let (mut producer, mut consumer) = rtrb::RingBuffer::new(64);
        for _ in 0..64 {
            producer.push(0.0).unwrap();
        }
        let stop = Arc::new(AtomicBool::new(false));

        let reader = std::thread::spawn(move || {
            std::thread::sleep(std::time::Duration::from_millis(20));
            let chunk = consumer.read_chunk(64).unwrap();
            chunk.commit_all();
            consumer
        });

        wait_for_drain(&producer, &stop, None);
        assert_eq!(producer.slots(), 64, "drain must wait for an empty buffer");
        drop(reader.join().unwrap());
    }

    #[test]
    fn drain_returns_when_playback_is_torn_down() {
        let (producer, consumer) = rtrb::RingBuffer::<f32>::new(64);
        let stop = Arc::new(AtomicBool::new(true));
        wait_for_drain(&producer, &stop, None);
        drop(consumer);
    }

    // --- Generation guard on timeline writes ---

    #[test]
    fn a_writer_knows_when_its_session_has_ended() {
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        assert!(tl.is_current());

        timeline.reset();
        assert!(!tl.is_current(), "reset must retire the outgoing writer");
        assert!(writer(&timeline).is_current());
    }

    #[test]
    fn stale_writes_are_dropped_after_reset() {
        let timeline = PlaybackTimeline::new();
        let dying = writer(&timeline);
        dying.push_boundary(make_boundary(QueueItemId::new(), 0, 0, 2, 44100));
        dying.add_written(88200);

        timeline.reset();

        // The outgoing decode thread checks `stop` only at the top of its chunk
        // loop, so its last packet lands after the reset.
        dying.add_written(4608);
        dying.push_boundary(make_boundary(QueueItemId::new(), 0, 0, 2, 44100));

        assert_eq!(timeline.samples_written.load(Ordering::Relaxed), 0);
        assert!(timeline.boundaries.read().is_empty());
    }

    #[test]
    fn a_dying_decode_thread_cannot_blank_the_transport() {
        let timeline = PlaybackTimeline::new();
        let dying = writer(&timeline);
        dying.push_boundary(make_boundary(QueueItemId::new(), 0, 0, 2, 44100));
        dying.add_written(88200);

        // A skip: reset, then the old thread's final packet, then the new
        // session stamps its first boundary at whatever the counter now says.
        timeline.reset();
        dying.add_written(4608);

        let fresh = writer(&timeline);
        let id = QueueItemId::new();
        let write_offset = fresh.samples_written();
        fresh.push_boundary(make_boundary(id, write_offset, 0, 2, 44100));

        assert_eq!(write_offset, 0, "first boundary must start at 0");
        let (playing, _, _, position_ms) = timeline
            .current_playback()
            .expect("transport must not go blank at samples_played = 0");
        assert_eq!(playing, id);
        assert_eq!(position_ms, 0);
    }

    // --- Failure handling in the gapless queue ---

    /// A file that exists and has an audio extension but no audio in it.
    fn write_garbage(path: &Path) {
        std::fs::write(path, b"this is not a wav file").unwrap();
    }

    #[test]
    fn an_unreadable_track_is_skipped_and_the_queue_continues() {
        let dir = tempfile::tempdir().unwrap();
        let a = dir.path().join("a.wav");
        let bad = dir.path().join("bad.wav");
        let c = dir.path().join("c.wav");
        crate::test_utils::generate_wav(&a, 44100, 2, 0.1, 16);
        write_garbage(&bad);
        crate::test_utils::generate_wav(&c, 44100, 2, 0.1, 16);

        let bounds = run_queue(&[a.clone(), bad, c.clone()]);
        let decoded: Vec<_> = bounds.iter().map(|b| b.path.clone()).collect();
        assert_eq!(
            decoded,
            vec![a, c],
            "one bad file must not take the rest of the queue with it"
        );
    }

    #[test]
    fn a_missing_track_is_skipped_and_the_queue_continues() {
        let dir = tempfile::tempdir().unwrap();
        let missing = dir.path().join("gone.wav");
        let b = dir.path().join("b.wav");
        crate::test_utils::generate_wav(&b, 44100, 2, 0.1, 16);

        let bounds = run_queue(&[missing, b.clone()]);
        assert_eq!(bounds.len(), 1);
        assert_eq!(
            bounds[0].path, b,
            "a bad first track must not end the session"
        );
    }

    #[test]
    fn an_entirely_unreadable_queue_terminates() {
        let dir = tempfile::tempdir().unwrap();
        let bad = dir.path().join("bad.wav");
        write_garbage(&bad);

        // Nothing decodes, so the ring stays empty; the consumer only has to
        // outlive the producer.
        let (producer, _consumer) = rtrb::RingBuffer::<f32>::new(1 << 12);
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);
        let stop = Arc::new(AtomicBool::new(false));

        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let counter = calls.clone();
        let bad_path = bad.clone();
        let next_track = move || {
            counter.fetch_add(1, Ordering::Relaxed);
            Some(SourceEntry::from_file(QueueItemId::new(), bad_path.clone()))
        };

        decode_queue_loop(
            SourceEntry::from_file(QueueItemId::new(), bad),
            producer,
            &stop,
            0,
            &next_track,
            &tl,
            None,
            crate::config::ReplayGainMode::Off,
            0.0,
        );

        // The first source plus one per next_track call, capped.
        assert_eq!(
            calls.load(Ordering::Relaxed) + 1,
            MAX_CONSECUTIVE_FAILURES as usize
        );
        assert!(timeline.boundaries.read().is_empty());
    }

    // --- Seek landing position ---

    #[test]
    fn landing_samples_converts_frame_timebases() {
        // The usual audio case: one tick per frame.
        let tb = TimeBase::try_from_recip(44100).unwrap();
        assert_eq!(
            landing_samples(Some(tb), Timestamp::from(44100u32), 44100, 2),
            Some(88_200)
        );
    }

    #[test]
    fn landing_samples_converts_millisecond_timebases() {
        // Matroska ticks in milliseconds, not frames.
        let tb = TimeBase::try_new(1, 1000).unwrap();
        assert_eq!(
            landing_samples(Some(tb), Timestamp::from(1500u32), 48000, 2),
            Some(48000 * 3 / 2 * 2)
        );
    }

    #[test]
    fn landing_samples_needs_a_timebase() {
        assert_eq!(
            landing_samples(None, Timestamp::from(1000u32), 44100, 2),
            None
        );
    }

    /// Build a 300s VBR MP3: 30s of near-silence then a loud tone, which gives
    /// lame a wide enough bitrate spread to make coarse seeking miss badly.
    #[cfg(test)]
    fn make_vbr_mp3(dir: &Path) -> PathBuf {
        let wav = dir.join("source.wav");
        let mp3 = dir.join("source.mp3");
        let ok = std::process::Command::new("sox")
            .args(["-n", "-r", "44100", "-c", "2"])
            .arg(&wav)
            .args([
                "synth", "30", "sine", "200", "vol", "0.02", ":", "synth", "270", "sine", "880",
                "vol", "0.9",
            ])
            .status()
            .expect("sox not installed")
            .success();
        assert!(ok, "sox failed");
        let ok = std::process::Command::new("lame")
            .args(["-V", "2", "--quiet"])
            .arg(&wav)
            .arg(&mp3)
            .status()
            .expect("lame not installed")
            .success();
        assert!(ok, "lame failed");
        mp3
    }

    /// A VBR seek must report where playback actually resumed: the reported
    /// start plus the audio that actually followed has to add back up to the
    /// file's duration. Under a coarse seek this file lands 3.7s late while
    /// reporting 83ms early — a 3.8s lie for the rest of the track.
    #[test]
    #[ignore = "generates a fixture with sox + lame; run with cargo test -- --ignored"]
    fn seek_on_vbr_reports_where_it_landed() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_vbr_mp3(dir.path());
        let info = probe_file(&path).unwrap();
        let channels = info.channels as u64;
        let rate = info.sample_rate as u64;

        let seek_ms = 150_000u64;
        let (mut producer, mut consumer) = rtrb::RingBuffer::<f32>::new(1 << 16);
        let stop = Arc::new(AtomicBool::new(false));
        let timeline = PlaybackTimeline::new();
        let tl = writer(&timeline);

        let drain_stop = stop.clone();
        let drained = std::thread::spawn(move || {
            let mut total = 0u64;
            while !drain_stop.load(Ordering::Relaxed) {
                let slots = consumer.slots();
                if slots == 0 {
                    std::thread::sleep(std::time::Duration::from_micros(200));
                    continue;
                }
                let chunk = consumer.read_chunk(slots).unwrap();
                total += slots as u64;
                chunk.commit_all();
            }
            total
        });

        let file = File::open(&path).unwrap();
        let mss = MediaSourceStream::new(Box::new(file), Default::default());
        let mut hint = Hint::new();
        hint.with_extension("mp3");
        decode_single(
            QueueItemId::new(),
            &path,
            &hint,
            mss,
            &mut producer,
            &stop,
            seek_ms,
            &tl,
            None,
            ReplayGainMode::Off,
            0.0,
            None,
        )
        .unwrap();

        let written = timeline.samples_written.load(Ordering::Relaxed);
        stop.store(true, Ordering::Relaxed);
        drained.join().unwrap();

        let reported_start_ms = {
            let bounds = timeline.boundaries.read();
            (bounds[0].seek_samples / channels) * 1000 / rate
        };
        let decoded_ms = (written / channels) * 1000 / rate;

        // Where playback started plus how much audio followed is the whole file.
        let total_ms = reported_start_ms + decoded_ms;
        assert!(
            total_ms.abs_diff(info.duration_ms) < 500,
            "reported start {}ms + {}ms decoded = {}ms, but the file is {}ms",
            reported_start_ms,
            decoded_ms,
            total_ms,
            info.duration_ms
        );
        // Landing is frame-granular, never sample-exact.
        assert!(
            reported_start_ms.abs_diff(seek_ms) < 100,
            "seek to {}ms reported {}ms",
            seek_ms,
            reported_start_ms
        );
    }
}