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::dsp::{Chain, Setup};
use crate::audio::opus::OpusBridge;
use crate::audio::viz::VizBuffer;
use crate::config::ReplayGainMode;
use crate::player::state::{QueueItemId, RendererClock};

#[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();
        }
    }

    /// The decode thread, to wake when its ring has room. An output that
    /// drains the ring faster than it plays, as an encoder for a renderer
    /// does, would otherwise wait out the decoder's park on a full ring.
    pub fn thread(&self) -> Option<thread::Thread> {
        self.thread.as_ref().map(|t| t.thread().clone())
    }

    /// 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,
    /// The rate the ring buffer's samples play at — the source's, unless DSP
    /// resampled it to reach an impulse response.
    pub output_rate: u32,
}

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;
        let out = self.output_rate as u64;
        if ch == 0 || rate == 0 || out == 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 / out + (self.seek_samples / ch) * 1000 / rate)
    }
}

/// Where the playhead is. A play is its boundary, not its item: an item
/// repeated gaplessly has a boundary per pass, and each is a play of its own.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Playhead {
    /// The index of the boundary the playhead is past, in this session.
    pub boundary: usize,
    pub id: QueueItemId,
    pub position_ms: u64,
}

/// 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>,
    /// 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>>>,
    /// Where a renderer playing this session's stream is, in milliseconds
    /// of the stream. Set, it is the playhead in place of `samples_played`:
    /// what has been encoded runs seconds ahead of what is heard.
    clock: parking_lot::Mutex<Option<RendererClock>>,
}

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)
            .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)),
            queued: parking_lot::Mutex::new(None),
            clock: parking_lot::Mutex::new(None),
        })
    }

    /// Samples played: the engine's count, or the renderer's clock.
    fn played(&self, bounds: &[TrackBoundary]) -> u64 {
        let Some(clock) = *self.clock.lock() else {
            return self.samples_played.load(Ordering::Acquire);
        };
        let Some(first) = bounds.first() else {
            return 0;
        };
        let per_second = first.output_rate as u64 * first.info.channels as u64;
        let at = clock.now_ms() * per_second / 1000;
        // Rounded down to a whole frame, and never past what was written.
        (at - at % (first.info.channels as u64).max(1))
            .min(self.samples_written.load(Ordering::Acquire))
    }

    /// The renderer's clock, in milliseconds of the stream.
    pub fn clock(&self) -> Option<RendererClock> {
        *self.clock.lock()
    }

    pub fn set_clock(&self, clock: Option<RendererClock>) {
        *self.clock.lock() = clock;
    }

    /// The track at `samples` into the session.
    pub fn track_at(&self, samples: u64) -> Option<QueueItemId> {
        let bounds = self.boundaries.read();
        let idx = bounds.partition_point(|b| b.sample_offset <= samples);
        Some(bounds.get(idx.checked_sub(1)?)?.id)
    }

    /// How long until the playhead reaches the end of what has been written.
    pub fn until_end(&self) -> Option<std::time::Duration> {
        let bounds = self.boundaries.read();
        let played = self.played(&bounds);
        let last = bounds.last()?;
        let per_second = last.output_rate as u64 * last.info.channels as u64;
        if per_second == 0 {
            return None;
        }
        let left = self
            .samples_written
            .load(Ordering::Acquire)
            .saturating_sub(played);
        Some(std::time::Duration::from_micros(
            left.saturating_mul(1_000_000) / per_second,
        ))
    }

    /// 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 play under the playhead and how far into it, without the clones
    /// `current_playback` makes.
    pub fn playhead(&self) -> Option<Playhead> {
        let bounds = self.boundaries.read();
        let played = self.played(&bounds);
        let boundary = bounds
            .partition_point(|b| b.sample_offset <= played)
            .checked_sub(1)?;
        let current = bounds.get(boundary)?;
        Some(Playhead {
            boundary,
            id: current.id,
            position_ms: current.position_ms(played)?,
        })
    }

    /// How many boundaries the session has: the index the next one takes.
    pub fn boundary_count(&self) -> usize {
        self.boundaries.read().len()
    }

    /// How far into the play at `boundary` playback has got, stopping at its
    /// end once the playhead has moved on to the next.
    pub fn position_in(&self, boundary: usize) -> Option<u64> {
        let bounds = self.boundaries.read();
        let played = self.played(&bounds);
        let at = bounds
            .get(boundary + 1)
            .map_or(played, |next| played.min(next.sample_offset));
        bounds.get(boundary)?.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.played(&bounds);
        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.output_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 tracks the decoder has queued after the one under the playhead, in
    /// order. Committed: the ring cannot be truncated, so only a new session
    /// takes them back.
    pub fn queued_after_playhead(&self) -> Vec<QueueItemId> {
        let bounds = self.boundaries.read();
        let played = self.played(&bounds);
        let idx = bounds.partition_point(|b| b.sample_offset <= played);
        bounds[idx..].iter().map(|b| b.id).collect()
    }

    /// The decode thread's write access.
    pub fn writer(&self) -> TimelineWriter<'_> {
        TimelineWriter { timeline: self }
    }

    /// Reset for a new playback session.
    pub fn reset(&self) {
        self.boundaries.write().clear();
        self.samples_written.store(0, Ordering::Relaxed);
        self.samples_played.store(0, Ordering::Relaxed);
        *self.clock.lock() = None;
    }

    /// 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.played(&bounds);

        // 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.
///
/// Sessions never overlap: the player joins the outgoing decode thread before
/// it resets the timeline for the next, so nothing here has to tell an old
/// session's writes from a new one's.
pub struct TimelineWriter<'a> {
    timeline: &'a PlaybackTimeline,
}

impl TimelineWriter<'_> {
    /// 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) {
        self.timeline.boundaries.write().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();
        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)
// ---------------------------------------------------------------------------

/// What is done to the samples between the decoder and the ring buffer.
#[derive(Clone)]
pub struct Processing {
    pub rg_mode: ReplayGainMode,
    pub pre_amp_db: f64,
    /// The output device's DSP profile. `None` leaves the samples untouched.
    pub dsp: Option<Arc<Setup>>,
}

impl Default for Processing {
    fn default() -> Self {
        Self {
            rg_mode: ReplayGainMode::Off,
            pre_amp_db: 0.0,
            dsp: None,
        }
    }
}

impl Processing {
    /// The rate a source at `source` plays at.
    pub fn output_rate(&self, source: u32) -> u32 {
        self.dsp.as_ref().map_or(source, |d| d.output_rate(source))
    }
}

/// 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>>,
    processing: Processing,
    on_finished: F,
) -> Result<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();
    let thread = thread::Builder::new()
        .name("koan-decode".into())
        .spawn(move || {
            decode_queue_loop(
                first,
                producer,
                &stop_clone,
                seek_ms,
                &next_track,
                &timeline.writer(),
                viz_buffer.as_deref(),
                &processing,
            );
            // 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)?;

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

// ---------------------------------------------------------------------------
// 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>,
    processing: &Processing,
) 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;
    let mut chain: Option<Chain> = None;

    while let Some(entry) = pending.take() {
        if stop.load(Ordering::Relaxed) {
            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,
                processing,
                &mut chain,
                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) {
            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"),
        }
    }

    // What the chain still holds is the end of the last track.
    if let (Some(chain), Some(format)) = (chain.as_mut(), format)
        && !stop.load(Ordering::Relaxed)
    {
        write_ring(&mut producer, chain.flush(), stop, viz_buffer, format);
    }

    wait_for_drain(&producer, stop, format);
}

/// Write `samples` into the ring, blocking while it is full. False once the
/// session has been stopped.
///
/// The viz buffer is fed inside the loop so it receives samples at the rate
/// the output consumes them, not in packet-sized bursts. Without this, FLAC
/// packets (~93ms each at 44.1kHz) would update it only ~11 times a second,
/// making waveform modes visibly choppy.
fn write_ring(
    producer: &mut rtrb::Producer<f32>,
    samples: &[f32],
    stop: &AtomicBool,
    viz_buffer: Option<&VizBuffer>,
    (sample_rate, channels): PcmFormat,
) -> bool {
    let mut offset = 0;
    while offset < samples.len() {
        if stop.load(Ordering::Relaxed) {
            return false;
        }

        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() };

            if let Some(viz) = viz_buffer {
                viz.push_samples(to_write, channels, sample_rate);
            }

            offset += chunk_size;
        }
    }
    true
}

/// 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>,
    processing: &Processing,
    chain: &mut Option<Chain>,
    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,
    };

    // The ring holds the output format, which DSP may have moved off the
    // source's: two sources resampled to one impulse response's rate play
    // gaplessly on one engine.
    let out_rate = processing.output_rate(sample_rate);
    let format = (out_rate, channels);
    if let Some(expected) = expected
        && expected != format
    {
        log::info!(
            "format change at {}: {}Hz/{}ch → {}Hz/{}ch, restarting audio engine",
            path.display(),
            expected.0,
            expected.1,
            out_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();
        }
    }

    // The chain carries across a gapless boundary: resetting its filters would
    // put a transient at every track change. A resampler for the outgoing rate
    // gives up its tail first, which belongs to the track before.
    if let Some(setup) = &processing.dsp {
        match chain {
            Some(c) => {
                let tail = c.set_source_rate(sample_rate);
                if !write_ring(producer, tail, stop, viz_buffer, format) {
                    return Ok(Decoded::Complete(format));
                }
            }
            None => *chain = Some(Chain::new(setup, sample_rate, channels)),
        }
    }

    // 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,
        output_rate: out_rate,
    });

    // Read ReplayGain tags and select the active gain for this track.
    let rg_mode = processing.rg_mode;
    let pre_amp_db = processing.pre_amp_db;
    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) {
            return Ok(Decoded::Complete(format));
        }

        let packet = match reader.next_packet() {
            Ok(Some(p)) => p,
            Ok(None) => return Ok(Decoded::Complete(format)),
            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
        };

        let (samples, length) = match chain.as_mut() {
            Some(c) => c.process(samples),
            None => (samples, samples.len() as u64),
        };
        if !write_ring(producer, samples, stop, viz_buffer, format) {
            return Ok(Decoded::Complete(format));
        }
        timeline.add_written(length);
    }
}

/// 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,
            output_rate: sample_rate,
        }
    }

    fn writer(timeline: &PlaybackTimeline) -> TimelineWriter<'_> {
        timeline.writer()
    }

    // --- 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_in(0), Some(1000));
        assert_eq!(timeline.position_in(1), Some(500));
        assert_eq!(
            timeline.playhead(),
            Some(Playhead {
                boundary: 1,
                id: b,
                position_ms: 500
            })
        );
    }

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

        assert_eq!(timeline.position_in(0), Some(1000), "the first pass ended");
        assert_eq!(timeline.position_in(1), Some(500));
        assert_eq!(timeline.playhead().map(|p| p.boundary), Some(1));
    }

    #[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,
            &Processing::default(),
            &mut None,
            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> {
        run_queue_with(paths, &Processing::default()).0
    }

    /// `run_queue`, through `processing`. Also returns how many samples the
    /// consumer read and how many the timeline counted.
    fn run_queue_with(
        paths: &[PathBuf],
        processing: &Processing,
    ) -> (Vec<TrackBoundary>, u64, u64) {
        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 || {
            let mut read = 0u64;
            while !drain_flag.load(Ordering::Relaxed) {
                let n = consumer.slots();
                if n > 0
                    && let Ok(chunk) = consumer.read_chunk(n)
                {
                    read += n as u64;
                    chunk.commit_all();
                }
                std::thread::sleep(std::time::Duration::from_micros(200));
            }
            read
        });

        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,
            processing,
        );

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

        let written = timeline.samples_written.load(Ordering::Relaxed);
        (timeline.boundaries.read().clone(), read, written)
    }

    #[test]
    fn convolution_at_one_rate_plays_two_source_rates_gaplessly() {
        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.5, 16);
        crate::test_utils::generate_wav(&b, 96000, 2, 0.5, 16);
        let mut ir = vec![0.0; 129];
        ir[64] = 1.0;
        let processing = Processing {
            dsp: Some(Arc::new(Setup::new(
                vec![],
                vec![crate::audio::dsp::Impulse::from_channels(48000, vec![ir])],
            ))),
            ..Processing::default()
        };

        let (bounds, read, written) = run_queue_with(&[a, b], &processing);
        assert_eq!(bounds.len(), 2, "both resample to 48 kHz, so one session");
        assert!(bounds.iter().all(|b| b.output_rate == 48000));
        assert_eq!(bounds[0].info.sample_rate, 44100);
        assert_eq!(bounds[1].info.sample_rate, 96000);
        // Half a second of each at 48 kHz, stereo — and every sample the
        // timeline counted reached the output, delays trimmed and tails flushed.
        assert_eq!(written, 2 * 24000 * 2);
        assert_eq!(read, written);
        assert_eq!(bounds[1].sample_offset, 24000 * 2);
    }

    #[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);
    }

    // --- 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,
            &Processing::default(),
        );

        // 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,
            &Processing::default(),
            &mut None,
            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
        );
    }
}