use std::io::{Cursor, Read, Seek, SeekFrom};
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
use std::time::{Duration, Instant};

use ringbuf::{HeapConsumer, HeapProducer, HeapRb};

use biquad::{Biquad, Coefficients, DirectForm2Transposed, ToHertz, Type as FilterType};
use rodio::{Sink, Source};
use rodio::source::UniformSourceIterator;
use serde::Serialize;
use symphonia::core::{
    audio::{AudioBufferRef, SampleBuffer, SignalSpec},
    codecs::{DecoderOptions, CODEC_TYPE_NULL},
    formats::{FormatOptions, FormatReader, SeekMode, SeekTo},
    io::{MediaSource, MediaSourceStream, MediaSourceStreamOptions},
    meta::MetadataOptions,
    probe::Hint,
    units::{self, Time},
};
use futures_util::StreamExt;
use tauri::{AppHandle, Emitter, State};

// ─── 10-Band Graphic Equalizer ────────────────────────────────────────────────

const EQ_BANDS_HZ: [f32; 10] = [31.0, 62.0, 125.0, 250.0, 500.0, 1000.0, 2000.0, 4000.0, 8000.0, 16000.0];
const EQ_Q: f32 = 1.41;
const EQ_CHECK_INTERVAL: usize = 1024;

struct EqSource<S: Source<Item = f32>> {
    inner: S,
    sample_rate: u32,
    channels: u16,
    gains: Arc<[AtomicU32; 10]>,
    enabled: Arc<AtomicBool>,
    pre_gain: Arc<AtomicU32>,
    filters: [[DirectForm2Transposed<f32>; 2]; 10],
    current_gains: [f32; 10],
    sample_counter: usize,
    channel_idx: usize,
}

impl<S: Source<Item = f32>> EqSource<S> {
    fn new(inner: S, gains: Arc<[AtomicU32; 10]>, enabled: Arc<AtomicBool>, pre_gain: Arc<AtomicU32>) -> Self {
        let sample_rate = inner.sample_rate();
        let channels = inner.channels();
        let filters = std::array::from_fn(|band| {
            let freq = EQ_BANDS_HZ[band].clamp(20.0, (sample_rate as f32 / 2.0) - 100.0);
            std::array::from_fn(|_| {
                let coeffs = Coefficients::<f32>::from_params(
                    FilterType::PeakingEQ(0.0),
                    (sample_rate as f32).hz(),
                    freq.hz(),
                    EQ_Q,
                ).unwrap_or_else(|_| Coefficients::<f32>::from_params(
                    FilterType::PeakingEQ(0.0),
                    (sample_rate as f32).hz(),
                    1000.0f32.hz(),
                    EQ_Q,
                ).unwrap());
                DirectForm2Transposed::<f32>::new(coeffs)
            })
        });
        Self {
            inner, sample_rate, channels, gains, enabled, pre_gain,
            filters,
            current_gains: [0.0; 10],
            sample_counter: 0,
            channel_idx: 0,
        }
    }

    fn refresh_if_needed(&mut self) {
        for band in 0..10 {
            let gain_db = f32::from_bits(self.gains[band].load(Ordering::Relaxed));
            if (gain_db - self.current_gains[band]).abs() > 0.01 {
                self.current_gains[band] = gain_db;
                let freq = EQ_BANDS_HZ[band].clamp(20.0, (self.sample_rate as f32 / 2.0) - 100.0);
                if let Ok(coeffs) = Coefficients::<f32>::from_params(
                    FilterType::PeakingEQ(gain_db),
                    (self.sample_rate as f32).hz(),
                    freq.hz(),
                    EQ_Q,
                ) {
                    for ch in 0..2 {
                        self.filters[band][ch].update_coefficients(coeffs);
                    }
                }
            }
        }
    }
}

impl<S: Source<Item = f32>> Iterator for EqSource<S> {
    type Item = f32;

    fn next(&mut self) -> Option<f32> {
        let sample = self.inner.next()?;

        if self.sample_counter % EQ_CHECK_INTERVAL == 0 {
            self.refresh_if_needed();
        }
        self.sample_counter = self.sample_counter.wrapping_add(1);

        if !self.enabled.load(Ordering::Relaxed) {
            self.channel_idx = (self.channel_idx + 1) % self.channels as usize;
            return Some(sample);
        }

        let ch = self.channel_idx.min(1);
        self.channel_idx = (self.channel_idx + 1) % self.channels as usize;

        let pre_gain_db = f32::from_bits(self.pre_gain.load(Ordering::Relaxed));
        let pre_gain_factor = 10_f32.powf(pre_gain_db / 20.0);
        let mut s = sample * pre_gain_factor;
        for band in 0..10 {
            s = self.filters[band][ch].run(s);
        }
        Some(s.clamp(-1.0, 1.0))
    }
}

impl<S: Source<Item = f32>> Source for EqSource<S> {
    fn current_frame_len(&self) -> Option<usize> { self.inner.current_frame_len() }
    fn channels(&self) -> u16 { self.channels }
    fn sample_rate(&self) -> u32 { self.sample_rate }
    fn total_duration(&self) -> Option<Duration> { self.inner.total_duration() }

    fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> {
        // Reset biquad filter state to avoid glitches after seek.
        for band in 0..10 {
            let gain_db = f32::from_bits(self.gains[band].load(Ordering::Relaxed));
            self.current_gains[band] = gain_db;
            let freq = EQ_BANDS_HZ[band].clamp(20.0, (self.sample_rate as f32 / 2.0) - 100.0);
            if let Ok(coeffs) = Coefficients::<f32>::from_params(
                FilterType::PeakingEQ(gain_db),
                (self.sample_rate as f32).hz(),
                freq.hz(),
                EQ_Q,
            ) {
                for ch in 0..2 {
                    self.filters[band][ch] = DirectForm2Transposed::<f32>::new(coeffs);
                }
            }
        }
        self.channel_idx = 0;
        self.sample_counter = 0;
        self.inner.try_seek(pos)
    }
}

// ─── DynSource — type-erased Source wrapper ───────────────────────────────────
//
// Allows chaining differently-typed sources (with trimming applied) into a
// single concrete type accepted by EqSource<S: Source<Item=f32>>.

struct DynSource {
    inner: Box<dyn Source<Item = f32> + Send>,
    channels: u16,
    sample_rate: u32,
}

impl DynSource {
    fn new(src: impl Source<Item = f32> + Send + 'static) -> Self {
        let channels = src.channels();
        let sample_rate = src.sample_rate();
        Self { inner: Box::new(src), channels, sample_rate }
    }
}

impl Iterator for DynSource {
    type Item = f32;
    fn next(&mut self) -> Option<f32> { self.inner.next() }
}

impl Source for DynSource {
    fn current_frame_len(&self) -> Option<usize> { self.inner.current_frame_len() }
    fn channels(&self) -> u16 { self.channels }
    fn sample_rate(&self) -> u32 { self.sample_rate }
    fn total_duration(&self) -> Option<Duration> { self.inner.total_duration() }
    fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> {
        self.inner.try_seek(pos)
    }
}

// ─── EqualPowerFadeIn — per-sample sin(t·π/2) fade-in envelope ───────────────
//
// Applied to every new track:
//   • Crossfade: fade_dur = crossfade_secs  → symmetric equal-power fade-in
//   • Hard cut:  fade_dur = 5 ms            → micro-fade eliminates DC-click
//   • Gapless:   fade_dur = 0               → unity gain (no modification)
//
// gain(t) = sin(t · π/2),  t ∈ [0, 1)
// At t = 0 gain = 0, at t = 1 gain = 1.
// Equal-power property: cos²+sin² = 1 → combined with cos fade-out on Track A
// the total perceived loudness stays constant across the crossfade.

struct EqualPowerFadeIn<S: Source<Item = f32>> {
    inner: S,
    sample_count: u64,
    fade_samples: u64,
}

impl<S: Source<Item = f32>> EqualPowerFadeIn<S> {
    fn new(inner: S, fade_dur: Duration) -> Self {
        let sample_rate = inner.sample_rate();
        let channels = inner.channels() as u64;
        let fade_samples = if fade_dur.is_zero() {
            0
        } else {
            (fade_dur.as_secs_f64() * sample_rate as f64 * channels as f64) as u64
        };
        Self { inner, sample_count: 0, fade_samples }
    }
}

impl<S: Source<Item = f32>> Iterator for EqualPowerFadeIn<S> {
    type Item = f32;
    fn next(&mut self) -> Option<f32> {
        let sample = self.inner.next()?;
        let gain = if self.fade_samples == 0 || self.sample_count >= self.fade_samples {
            1.0
        } else {
            let t = self.sample_count as f32 / self.fade_samples as f32;
            (t * std::f32::consts::FRAC_PI_2).sin()
        };
        self.sample_count += 1;
        Some((sample * gain).clamp(-1.0, 1.0))
    }
}

impl<S: Source<Item = f32>> Source for EqualPowerFadeIn<S> {
    fn current_frame_len(&self) -> Option<usize> { self.inner.current_frame_len() }
    fn channels(&self) -> u16 { self.inner.channels() }
    fn sample_rate(&self) -> u32 { self.inner.sample_rate() }
    fn total_duration(&self) -> Option<Duration> { self.inner.total_duration() }
    fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> {
        // For mid-track seeks: skip straight to unity gain so the new position
        // plays at full volume immediately — no audible fade-in glitch.
        // For seeks to the very start (< 100 ms): keep the micro-fade to
        // suppress any DC-offset click from the fresh decode.
        if pos.as_millis() < 100 {
            self.sample_count = 0;
        } else {
            self.sample_count = self.fade_samples;
        }
        self.inner.try_seek(pos)
    }
}

// ─── TriggeredFadeOut — sample-level cos(t·π/2) fade-out triggered externally ─
//
// Every track source is wrapped with this. It passes through at unity gain
// until `trigger` is set to true, at which point it reads `fade_total_samples`
// and applies a cos(t·π/2) envelope:
//   gain(t) = cos(t · π/2),  t ∈ [0, 1]
//   At t = 0 gain = 1, at t = 1 gain = 0.
// After the fade completes, returns None to exhaust the source.
//
// Combined with EqualPowerFadeIn (sin curve) on Track B, this gives a
// symmetric constant-power crossfade: sin²+cos² = 1.

struct TriggeredFadeOut<S: Source<Item = f32>> {
    inner: S,
    trigger: Arc<AtomicBool>,
    fade_total_samples: Arc<AtomicU64>,
    fade_progress: u64,
    fading: bool,
    cached_total: u64,
}

impl<S: Source<Item = f32>> TriggeredFadeOut<S> {
    fn new(inner: S, trigger: Arc<AtomicBool>, fade_total_samples: Arc<AtomicU64>) -> Self {
        Self {
            inner,
            trigger,
            fade_total_samples,
            fade_progress: 0,
            fading: false,
            cached_total: 0,
        }
    }
}

impl<S: Source<Item = f32>> Iterator for TriggeredFadeOut<S> {
    type Item = f32;
    fn next(&mut self) -> Option<f32> {
        // Check trigger on first fade sample only (avoid atomic load per sample).
        if !self.fading && self.trigger.load(Ordering::Relaxed) {
            self.fading = true;
            self.cached_total = self.fade_total_samples.load(Ordering::Relaxed).max(1);
            self.fade_progress = 0;
        }

        if self.fading {
            if self.fade_progress >= self.cached_total {
                // Fade complete — exhaust the source.
                return None;
            }
            let sample = self.inner.next()?;
            let t = self.fade_progress as f32 / self.cached_total as f32;
            let gain = (t * std::f32::consts::FRAC_PI_2).cos();
            self.fade_progress += 1;
            Some((sample * gain).clamp(-1.0, 1.0))
        } else {
            self.inner.next()
        }
    }
}

impl<S: Source<Item = f32>> Source for TriggeredFadeOut<S> {
    fn current_frame_len(&self) -> Option<usize> { self.inner.current_frame_len() }
    fn channels(&self) -> u16 { self.inner.channels() }
    fn sample_rate(&self) -> u32 { self.inner.sample_rate() }
    fn total_duration(&self) -> Option<Duration> { self.inner.total_duration() }
    fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> {
        // If we seek back during a fade, cancel the fade.
        if self.fading {
            self.fading = false;
            self.trigger.store(false, Ordering::Relaxed);
        }
        self.fade_progress = 0;
        self.inner.try_seek(pos)
    }
}

// ─── NotifyingSource — sets a flag when the inner iterator is exhausted ───────
//
// This is the key mechanism for gapless: the progress task polls `done` to know
// exactly when source N has finished inside the Sink, without relying on
// wall-clock estimation or the unreliable `Sink::empty()`.

struct NotifyingSource<S: Source<Item = f32>> {
    inner: S,
    done: Arc<AtomicBool>,
    signalled: bool,
}

impl<S: Source<Item = f32>> NotifyingSource<S> {
    fn new(inner: S, done: Arc<AtomicBool>) -> Self {
        Self { inner, done, signalled: false }
    }
}

impl<S: Source<Item = f32>> Iterator for NotifyingSource<S> {
    type Item = f32;
    fn next(&mut self) -> Option<f32> {
        let sample = self.inner.next();
        if sample.is_none() && !self.signalled {
            self.signalled = true;
            self.done.store(true, Ordering::SeqCst);
        }
        sample
    }
}

impl<S: Source<Item = f32>> Source for NotifyingSource<S> {
    fn current_frame_len(&self) -> Option<usize> { self.inner.current_frame_len() }
    fn channels(&self) -> u16 { self.inner.channels() }
    fn sample_rate(&self) -> u32 { self.inner.sample_rate() }
    fn total_duration(&self) -> Option<Duration> { self.inner.total_duration() }
    fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> {
        // If we seek backwards the source is no longer exhausted.
        self.signalled = false;
        self.done.store(false, Ordering::SeqCst);
        self.inner.try_seek(pos)
    }
}

// ─── CountingSource — atomic sample counter for drift-free position tracking ─
//
// Wraps the outermost source and increments a shared AtomicU64 on every sample.
// The progress task reads this counter and divides by (sample_rate * channels)
// to get the exact playback position — no wall-clock drift.

struct CountingSource<S: Source<Item = f32>> {
    inner: S,
    counter: Arc<AtomicU64>,
}

impl<S: Source<Item = f32>> CountingSource<S> {
    fn new(inner: S, counter: Arc<AtomicU64>) -> Self {
        Self { inner, counter }
    }
}

impl<S: Source<Item = f32>> Iterator for CountingSource<S> {
    type Item = f32;
    fn next(&mut self) -> Option<f32> {
        let sample = self.inner.next();
        if sample.is_some() {
            self.counter.fetch_add(1, Ordering::Relaxed);
        }
        sample
    }
}

impl<S: Source<Item = f32>> Source for CountingSource<S> {
    fn current_frame_len(&self) -> Option<usize> { self.inner.current_frame_len() }
    fn channels(&self) -> u16 { self.inner.channels() }
    fn sample_rate(&self) -> u32 { self.inner.sample_rate() }
    fn total_duration(&self) -> Option<Duration> { self.inner.total_duration() }
    fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> {
        // Reset counter only after confirming the inner seek succeeded.
        // If we reset first and the seek fails, the counter ends up at the
        // new position while the decoder is still at the old one — causing
        // a permanent desync between displayed time and actual audio.
        let result = self.inner.try_seek(pos);
        if result.is_ok() {
            let samples = (pos.as_secs_f64() * self.inner.sample_rate() as f64
                * self.inner.channels() as f64) as u64;
            self.counter.store(samples, Ordering::Relaxed);
        }
        result
    }
}

// ─── Internet Radio v2 — Lock-Free SPSC + ICY Metadata + Hybrid Pause ────────
//
//  HTTP task (tokio)
//    └─[IcyInterceptor]─► HeapProducer<u8>
//                              │  (4 MB HeapRb, lock-free)
//                         HeapConsumer<u8>
//                              │
//                       AudioStreamReader (Read + Seek + MediaSource)
//                              │
//                       SizedDecoder (symphonia)
//                              │
//                           rodio Sink
//
// Pause modes:
//   Logical pause  — sink.pause(); download task keeps filling (time-shift).
//   Hard pause     — buffer ≥ RADIO_HARD_PAUSE_THRESH full + paused ≥ 5 s
//                    → TCP disconnect, is_hard_paused = true.
//   Resume (warm)  — sink.play(); buffer drains seamlessly.
//   Resume (cold)  — new HeapRb + new GET; consumer swapped in AudioStreamReader.
//
// New Tauri event: "radio:metadata" → String  (ICY StreamTitle)

/// 256 KB on the heap — ≈16 s at 128 kbps, ≈6 s at 320 kbps.
/// Small enough that stale audio drains within a few seconds on reconnect;
/// large enough to absorb brief network hiccups without stuttering.
const RADIO_BUF_CAPACITY: usize = 256 * 1024;
/// Seconds at stall threshold while paused before hard-disconnect.
const RADIO_HARD_PAUSE_SECS: u64 = 5;
/// AudioStreamReader timeout: if no audio bytes arrive for this long → EOF.
const RADIO_READ_TIMEOUT_SECS: u64 = 15;
/// Sleep interval when ring buffer is empty (prevents CPU spin).
const RADIO_YIELD_MS: u64 = 2;

// ── ICY Metadata State Machine ────────────────────────────────────────────────
//
// Shoutcast/Icecast embed metadata every `metaint` audio bytes:
//
//   ┌──────────────────────┬───┬─────────────┐
//   │  audio × metaint     │ N │ meta × N×16 │  (repeating)
//   └──────────────────────┴───┴─────────────┘
//
// N = 0 → no metadata this block.  Metadata bytes are stripped so only
// pure audio reaches the ring buffer and Symphonia never sees text bytes.

enum IcyState {
    /// Forwarding audio bytes; `remaining` counts down to the next boundary.
    ReadingAudio { remaining: usize },
    /// Next byte is the metadata length multiplier N.
    ReadingLengthByte,
    /// Accumulating N×16 metadata bytes.
    ReadingMetadata { remaining: usize, buf: Vec<u8> },
}

struct IcyInterceptor {
    state: IcyState,
    metaint: usize,
}

impl IcyInterceptor {
    fn new(metaint: usize) -> Self {
        Self { metaint, state: IcyState::ReadingAudio { remaining: metaint } }
    }

    /// Feed a raw HTTP chunk.
    /// Appends only audio bytes to `audio_out`.
    /// Returns `Some(IcyMeta)` when a StreamTitle is extracted.
    fn process(&mut self, input: &[u8], audio_out: &mut Vec<u8>) -> Option<IcyMeta> {
        let mut extracted: Option<IcyMeta> = None;
        let mut i = 0;
        while i < input.len() {
            match &mut self.state {
                IcyState::ReadingAudio { remaining } => {
                    let n = (input.len() - i).min(*remaining);
                    audio_out.extend_from_slice(&input[i..i + n]);
                    i += n;
                    *remaining -= n;
                    if *remaining == 0 {
                        self.state = IcyState::ReadingLengthByte;
                    }
                }
                IcyState::ReadingLengthByte => {
                    let len_n = input[i] as usize;
                    i += 1;
                    self.state = if len_n == 0 {
                        IcyState::ReadingAudio { remaining: self.metaint }
                    } else {
                        IcyState::ReadingMetadata {
                            remaining: len_n * 16,
                            buf: Vec::with_capacity(len_n * 16),
                        }
                    };
                }
                IcyState::ReadingMetadata { remaining, buf } => {
                    let n = (input.len() - i).min(*remaining);
                    buf.extend_from_slice(&input[i..i + n]);
                    i += n;
                    *remaining -= n;
                    if *remaining == 0 {
                        let bytes = std::mem::take(buf);
                        extracted = parse_icy_meta(&bytes);
                        self.state = IcyState::ReadingAudio { remaining: self.metaint };
                    }
                }
            }
        }
        extracted
    }
}

/// ICY metadata parsed from a raw metadata block.
#[derive(serde::Serialize, Clone)]
pub(crate) struct IcyMeta {
    pub title: String,
    /// `true` when `StreamUrl='0'` — indicates a CDN-injected ad/promo.
    pub is_ad: bool,
}

/// Extract `StreamTitle` and `StreamUrl` from a raw ICY metadata block.
/// Tolerates null padding and non-UTF-8 bytes (lossy conversion).
fn parse_icy_meta(raw: &[u8]) -> Option<IcyMeta> {
    let s = String::from_utf8_lossy(raw);
    let s = s.trim_end_matches('\0');

    const TITLE_TAG: &str = "StreamTitle='";
    let title_start = s.find(TITLE_TAG)? + TITLE_TAG.len();
    let title_rest = &s[title_start..];
    // find (not rfind) — rfind would skip past StreamUrl and corrupt the title
    let title_end = title_rest.find("';")?;
    let title = title_rest[..title_end].trim().to_string();
    if title.is_empty() {
        return None;
    }

    const URL_TAG: &str = "StreamUrl='";
    let stream_url = s.find(URL_TAG).map(|pos| {
        let rest = &s[pos + URL_TAG.len()..];
        let end = rest.find("';").unwrap_or(rest.len());
        rest[..end].trim().to_string()
    }).unwrap_or_default();

    Some(IcyMeta { title, is_ad: stream_url == "0" })
}

// ── AudioStreamReader — SPSC consumer → std::io::Read ────────────────────────
//
// Bridges HeapConsumer<u8> (non-blocking) into the synchronous Read interface
// that Symphonia requires.  Designed to run inside tokio::task::spawn_blocking.
//
// Empty buffer:  sleeps RADIO_YIELD_MS ms, retries. Never busy-spins.
// Timeout:       after RADIO_READ_TIMEOUT_SECS with no data → TimedOut.
// Generation:    if gen_arc != self.gen → Ok(0) (EOF; new track started).
// Reconnect:     audio_resume sends a fresh HeapConsumer via new_cons_rx.
//                On the next read() we drain the channel (keep latest) and swap.

struct AudioStreamReader {
    cons: HeapConsumer<u8>,
    /// Delivers fresh consumers on hard-pause reconnect (unbounded; drain to latest).
    /// Wrapped in Mutex so AudioStreamReader is Sync (required by symphonia::MediaSource).
    /// No real contention: only the audio thread ever calls read().
    new_cons_rx: Mutex<std::sync::mpsc::Receiver<HeapConsumer<u8>>>,
    deadline: std::time::Instant,
    gen_arc: Arc<AtomicU64>,
    gen: u64,
    /// Monotonic byte offset for SeekFrom::Current(0) "tell" (Symphonia probe).
    pos: u64,
}

impl Read for AudioStreamReader {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        // EOF guard: new track started.
        if self.gen_arc.load(Ordering::SeqCst) != self.gen {
            return Ok(0);
        }
        // Drain reconnect channel; keep only the most recently delivered consumer
        // so a double-tap of resume doesn't leave stale data in place.
        let mut newest: Option<HeapConsumer<u8>> = None;
        while let Ok(c) = self.new_cons_rx.lock().unwrap().try_recv() {
            newest = Some(c);
        }
        if let Some(c) = newest {
            self.cons = c;
            self.deadline =
                std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS);
        }
        loop {
            if self.gen_arc.load(Ordering::SeqCst) != self.gen {
                return Ok(0);
            }
            let available = self.cons.len();
            if available > 0 {
                let n = buf.len().min(available);
                let read = self.cons.pop_slice(&mut buf[..n]);
                self.pos += read as u64;
                // Reset deadline: data arrived, so connection is alive.
                self.deadline =
                    std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS);
                return Ok(read);
            }
            if std::time::Instant::now() >= self.deadline {
                eprintln!(
                    "[radio] AudioStreamReader: {}s without data → EOF",
                    RADIO_READ_TIMEOUT_SECS
                );
                return Err(std::io::Error::new(
                    std::io::ErrorKind::TimedOut,
                    "radio: no data received",
                ));
            }
            std::thread::sleep(Duration::from_millis(RADIO_YIELD_MS));
        }
    }
}

impl Seek for AudioStreamReader {
    fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
        match pos {
            SeekFrom::Current(0) => Ok(self.pos),
            _ => Err(std::io::Error::new(
                std::io::ErrorKind::Unsupported,
                "radio stream is not seekable",
            )),
        }
    }
}

impl MediaSource for AudioStreamReader {
    fn is_seekable(&self) -> bool { false }
    fn byte_len(&self) -> Option<u64> { None }
}

// ── Pause / Reconnect Coordination ───────────────────────────────────────────

pub(crate) struct RadioSharedFlags {
    /// Set by audio_pause; cleared by audio_resume.
    is_paused: AtomicBool,
    /// Set by download task on hard disconnect; cleared on resume-reconnect.
    is_hard_paused: AtomicBool,
    /// Delivers a fresh HeapConsumer<u8> to AudioStreamReader on reconnect.
    new_cons_tx: Mutex<std::sync::mpsc::Sender<HeapConsumer<u8>>>,
}

/// Live state for the current radio session, stored in AudioEngine.
/// Dropping this struct aborts the HTTP download task immediately.
pub(crate) struct RadioLiveState {
    pub url: String,
    pub gen: u64,
    pub task: tokio::task::JoinHandle<()>,
    pub flags: Arc<RadioSharedFlags>,
}

impl Drop for RadioLiveState {
    fn drop(&mut self) { self.task.abort(); }
}

// ── HE-AAC / FDK-AAC Fallback ────────────────────────────────────────────────
//
// Symphonia 0.5.x: AAC-LC only.  HE-AAC (AAC+) and HE-AACv2 lack SBR/PS →
// streams play at half speed with muffled audio.
//
// With Cargo feature "fdk-aac": FdkAacDecoder is tried first for CODEC_TYPE_AAC.
// Enable in Cargo.toml:
//   symphonia-adapter-fdk-aac = { version = "0.1", optional = true }
//   [features]
//   fdk-aac = ["dep:symphonia-adapter-fdk-aac"]

fn try_make_radio_decoder(
    params: &symphonia::core::codecs::CodecParameters,
    opts: &DecoderOptions,
) -> Result<Box<dyn symphonia::core::codecs::Decoder>, symphonia::core::errors::Error> {
    symphonia::default::get_codecs().make(params, opts)
}

// ── Async HTTP Download Task ──────────────────────────────────────────────────
//
// Lifecycle:
//   'outer loop — reconnect on TCP drop (up to MAX_RECONNECTS)
//   'inner loop — read HTTP chunks → ICY interceptor → push audio to ring buffer
//
// Hard-pause detection: if push_slice() returns 0 (buffer full) AND sink is
// paused AND that condition persists for RADIO_HARD_PAUSE_SECS → disconnect.
// Sets is_hard_paused = true so audio_resume knows it must reconnect.

async fn radio_download_task(
    gen: u64,
    gen_arc: Arc<AtomicU64>,
    mut initial_response: Option<reqwest::Response>,
    http_client: reqwest::Client,
    url: String,
    mut prod: HeapProducer<u8>,
    flags: Arc<RadioSharedFlags>,
    app: AppHandle,
) {
    let mut bytes_total: u64 = 0;
    // Counts consecutive failures (reset on each successful chunk).
    // laut.fm and similar CDNs force-reconnect every ~700 KB; this is normal.
    let mut reconnect_count: u32 = 0;
    const MAX_CONSECUTIVE_FAILURES: u32 = 5;
    let mut audio_scratch: Vec<u8> = Vec::with_capacity(65_536);

    'outer: loop {
        if gen_arc.load(Ordering::SeqCst) != gen { return; }

        // ── Obtain response (initial or reconnect) ────────────────────────────
        let response = match initial_response.take() {
            Some(r) => r,
            None => {
                if reconnect_count >= MAX_CONSECUTIVE_FAILURES {
                    eprintln!("[radio] {MAX_CONSECUTIVE_FAILURES} consecutive failures — giving up");
                    break 'outer;
                }
                tokio::time::sleep(Duration::from_millis(500)).await;
                if gen_arc.load(Ordering::SeqCst) != gen { return; }
                match http_client
                    .get(&url)
                    .header("Icy-MetaData", "1")
                    .send()
                    .await
                {
                    Ok(r) if r.status().is_success() => {
                        eprintln!("[radio] reconnected ({bytes_total} B so far)");
                        r
                    }
                    Ok(r) => {
                        eprintln!("[radio] reconnect: HTTP {} — giving up", r.status());
                        break 'outer;
                    }
                    Err(e) => {
                        eprintln!("[radio] reconnect error: {e} — giving up");
                        break 'outer;
                    }
                }
            }
        };

        // Parse ICY metaint from each response (consistent across reconnects).
        let metaint: Option<usize> = response
            .headers()
            .get("icy-metaint")
            .and_then(|v| v.to_str().ok())
            .and_then(|s| s.parse().ok());
        let mut icy = metaint.map(IcyInterceptor::new);

        let mut byte_stream = response.bytes_stream();
        // Stall timer: tracks how long push_slice() returns 0 while paused.
        let mut stall_since: Option<std::time::Instant> = None;

        'inner: loop {
            if gen_arc.load(Ordering::SeqCst) != gen { return; }

            // ── Back-pressure + hard-pause detection ──────────────────────────
            if prod.is_full() {
                if flags.is_paused.load(Ordering::Relaxed) {
                    let since = stall_since.get_or_insert(std::time::Instant::now());
                    if since.elapsed() >= Duration::from_secs(RADIO_HARD_PAUSE_SECS) {
                        let fill_pct = ((1.0
                            - prod.free_len() as f32 / RADIO_BUF_CAPACITY as f32)
                            * 100.0) as u32;
                        eprintln!(
                            "[radio] hard pause: {fill_pct}% full, \
                             paused >{RADIO_HARD_PAUSE_SECS}s → disconnecting"
                        );
                        flags.is_hard_paused.store(true, Ordering::Release);
                        return; // Drop HeapProducer → TCP connection released.
                    }
                } else {
                    stall_since = None;
                }
                tokio::time::sleep(Duration::from_millis(50)).await;
                continue 'inner;
            }
            stall_since = None;

            // ── Read HTTP chunk ───────────────────────────────────────────────
            match byte_stream.next().await {
                Some(Ok(chunk)) => {
                    bytes_total += chunk.len() as u64;
                    // Successful data → reset consecutive-failure counter.
                    reconnect_count = 0;
                    audio_scratch.clear();

                    if let Some(ref mut interceptor) = icy {
                        if let Some(meta) = interceptor.process(&chunk, &mut audio_scratch) {
                            let label = if meta.is_ad { "[Ad]" } else { "" };
                            eprintln!("[radio] ICY StreamTitle: {}{}", label, meta.title);
                            let _ = app.emit("radio:metadata", &meta);
                        }
                    } else {
                        audio_scratch.extend_from_slice(&chunk);
                    }

                    // Push with per-chunk back-pressure: yield 5 ms if full mid-chunk.
                    let mut offset = 0;
                    while offset < audio_scratch.len() {
                        if gen_arc.load(Ordering::SeqCst) != gen { return; }
                        let pushed = prod.push_slice(&audio_scratch[offset..]);
                        if pushed == 0 {
                            tokio::time::sleep(Duration::from_millis(5)).await;
                        } else {
                            offset += pushed;
                        }
                    }
                }
                Some(Err(e)) => {
                    reconnect_count += 1;
                    eprintln!("[radio] stream error: {e} → reconnecting (consecutive #{reconnect_count})");
                    break 'inner;
                }
                None => {
                    reconnect_count += 1;
                    eprintln!("[radio] stream ended cleanly → reconnecting (consecutive #{reconnect_count})");
                    break 'inner;
                }
            }
        } // 'inner

        // Do NOT swap the ring buffer here.  The remaining bytes in the buffer
        // are still valid audio and will drain naturally during reconnect.
        // Clearing it would cause an immediate underrun/glitch.
        // The buffer is kept small (RADIO_BUF_CAPACITY) so stale audio drains
        // within a few seconds rather than minutes.
    } // 'outer

    eprintln!("[radio] download task done ({bytes_total} B total)");
}

fn content_type_to_hint(ct: &str) -> Option<String> {
    let ct = ct.to_ascii_lowercase();
    if ct.contains("mpeg") || ct.contains("mp3") { Some("mp3".into()) }
    else if ct.contains("aac") || ct.contains("aacp") { Some("aac".into()) }
    else if ct.contains("ogg") { Some("ogg".into()) }
    else if ct.contains("flac") { Some("flac".into()) }
    else { None }
}

// ─── SizedCursorSource — correct byte_len for seekable in-memory sources ──────
//
// rodio's internal ReadSeekSource wraps Cursor<Vec<u8>> but hardcodes
// byte_len() → None.  This tells symphonia "stream length unknown", which
// prevents the FLAC demuxer from seeking (it validates seek offsets against
// the total stream length from byte_len).  MP3 is unaffected because its
// demuxer uses Xing/LAME headers instead.
//
// This wrapper provides the actual byte length, fixing seek for all formats.

struct SizedCursorSource {
    inner: Cursor<Vec<u8>>,
    len: u64,
}

impl Read for SizedCursorSource {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.inner.read(buf)
    }
}

impl Seek for SizedCursorSource {
    fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
        self.inner.seek(pos)
    }
}

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

// ─── SizedDecoder — symphonia decoder with correct byte_len ───────────────────
//
// Replaces rodio::Decoder::new() which wraps the source in ReadSeekSource
// (byte_len = None).  This constructs the symphonia pipeline directly,
// providing the correct byte_len via SizedCursorSource.
//
// Implements Iterator<Item = i16> + Source — identical interface to
// rodio::Decoder, so the rest of the source chain is unchanged.

/// Max retries for IO/packet-read errors (fatal — network drop, truncated file).
const DECODE_MAX_RETRIES: usize = 3;
/// Max *consecutive* DecodeErrors before giving up on a file.
/// Non-fatal errors like "invalid main_data offset" are silently dropped up to
/// this limit so a handful of corrupt MP3 frames never aborts an otherwise
/// playable track (VLC-style frame dropping).
const MAX_CONSECUTIVE_DECODE_ERRORS: usize = 100;

struct SizedDecoder {
    decoder: Box<dyn symphonia::core::codecs::Decoder>,
    current_frame_offset: usize,
    format: Box<dyn FormatReader>,
    total_duration: Option<Time>,
    buffer: SampleBuffer<i16>,
    spec: SignalSpec,
    /// Counts consecutive DecodeErrors in the hot-path. Reset to 0 on every
    /// successfully decoded frame. Used to detect fully undecodable streams.
    consecutive_decode_errors: usize,
}

impl SizedDecoder {
    fn new(data: Vec<u8>, format_hint: Option<&str>, hi_res: bool) -> Result<Self, String> {
        let data_len = data.len() as u64;
        let source = SizedCursorSource {
            inner: Cursor::new(data),
            len: data_len,
        };
        // Hi-Res: 4 MB read-ahead so Symphonia demuxes fewer Read calls for
        // high-bitrate files (88.2 kHz/24-bit FLAC ≈ 1800 kbps).
        // Standard: 512 KB is plenty for MP3/AAC — larger buffers waste allocation
        // and compete with the playback thread at track start.
        let buf_len = if hi_res { 4 * 1024 * 1024 } else { 512 * 1024 };
        let mss = MediaSourceStream::new(
            Box::new(source) as Box<dyn MediaSource>,
            MediaSourceStreamOptions { buffer_len: buf_len },
        );

        let mut hint = Hint::new();
        if let Some(ext) = format_hint {
            hint.with_extension(ext);
        }
        let format_opts = FormatOptions {
            // Disable gapless parsing — Symphonia 0.5.5 crashes on `edts` atoms
            // present in older iTunes-purchased M4A files.
            enable_gapless: false,
            ..Default::default()
        };

        let meta_opts = symphonia::core::meta::MetadataOptions {
            // Cap embedded cover art at 8 MiB so oversized MJPEG images in
            // iTunes M4A files don't choke the parser.
            limit_visual_bytes: symphonia::core::meta::Limit::Maximum(8 * 1024 * 1024),
            ..Default::default()
        };

        let probed = symphonia::default::get_probe()
            .format(&hint, mss, &format_opts, &meta_opts)
            .map_err(|e| {
                let hint_str = format_hint.unwrap_or("unknown");
                // Always print the raw Symphonia error to the terminal for diagnosis.
                eprintln!("[psysonic] probe failed (hint={hint_str}): {e}");
                if e.to_string().to_lowercase().contains("unsupported") {
                    format!("unsupported format: .{hint_str} files cannot be played (no demuxer)")
                } else {
                    format!("could not open audio stream (.{hint_str}): {e}")
                }
            })?;

        let track = probed.format
            .tracks()
            .iter()
            // Explicitly select only audio tracks: must have a valid codec and a
            // sample_rate. This skips MJPEG cover-art streams that iTunes M4A
            // files embed as a secondary video track.
            .find(|t| {
                t.codec_params.codec != CODEC_TYPE_NULL
                    && t.codec_params.sample_rate.is_some()
            })
            .ok_or_else(|| {
                eprintln!("[psysonic] no audio track found among {} tracks", probed.format.tracks().len());
                "no playable audio track found in file".to_string()
            })?;

        let track_id = track.id;
        let total_duration = track.codec_params.time_base
            .zip(track.codec_params.n_frames)
            .map(|(base, frames)| base.calc_time(frames));

        let mut decoder = symphonia::default::get_codecs()
            .make(&track.codec_params, &DecoderOptions::default())
            .map_err(|e| {
                eprintln!("[psysonic] codec init failed: {e}");
                if e.to_string().to_lowercase().contains("unsupported") {
                    "unsupported codec: no decoder available for this audio format".to_string()
                } else {
                    format!("failed to initialise audio decoder: {e}")
                }
            })?;

        let mut format = probed.format;

        // Decode the first packet to initialise spec + buffer.
        // DecodeErrors (e.g. "invalid main_data offset") are non-fatal: drop the
        // frame and try the next packet up to MAX_CONSECUTIVE_DECODE_ERRORS times.
        let mut decode_errors: usize = 0;
        let decoded = loop {
            let packet = match format.next_packet() {
                Ok(p) => p,
                Err(symphonia::core::errors::Error::IoError(_)) => {
                    break decoder.last_decoded();
                }
                Err(e) => {
                    eprintln!("[psysonic] next_packet error: {e}");
                    return Err(format!("could not read audio data: {e}"));
                }
            };
            if packet.track_id() != track_id {
                eprintln!("[psysonic] skipping packet for track {} (want {})", packet.track_id(), track_id);
                continue;
            }
            match decoder.decode(&packet) {
                Ok(decoded) => break decoded,
                Err(symphonia::core::errors::Error::DecodeError(ref msg)) => {
                    decode_errors += 1;
                    eprintln!("[psysonic] init: dropped corrupt frame #{decode_errors}: {msg}");
                    if decode_errors >= MAX_CONSECUTIVE_DECODE_ERRORS {
                        return Err("too many consecutive decode errors during init — file may be corrupt".into());
                    }
                }
                Err(e) => {
                    eprintln!("[psysonic] fatal decode error: {e}");
                    return Err(format!("audio decode error: {e}"));
                }
            }
        };

        let spec = decoded.spec().to_owned();
        let buffer = Self::make_buffer(decoded, &spec);

        Ok(SizedDecoder {
            decoder,
            current_frame_offset: 0,
            format,
            total_duration,
            buffer,
            spec,
            consecutive_decode_errors: 0,
        })
    }

    /// Build a decoder from any `MediaSource` (e.g. `RadioBuffer`).
    /// Uses `enable_gapless: false` — live streams are not seekable; gapless
    /// trimming requires seeking to read the LAME/iTunSMPB end-padding info.
    fn new_streaming(media: Box<dyn MediaSource>, format_hint: Option<&str>) -> Result<Self, String> {
        // Larger read-ahead buffer for the live radio SPSC consumer — reduces
        // read() call frequency into the ring buffer, easing I/O spikes.
        let mss = MediaSourceStream::new(media, MediaSourceStreamOptions { buffer_len: 512 * 1024 });
        let mut hint = Hint::new();
        if let Some(ext) = format_hint { hint.with_extension(ext); }
        let format_opts = FormatOptions { enable_gapless: false, ..Default::default() };
        let probed = symphonia::default::get_probe()
            .format(&hint, mss, &format_opts, &MetadataOptions::default())
            .map_err(|e| format!("radio: format probe failed: {e}"))?;

        let track = probed.format.tracks().iter()
            .find(|t| t.codec_params.codec != CODEC_TYPE_NULL)
            .ok_or_else(|| "radio: no audio track found".to_string())?;
        let track_id = track.id;
        // Live streams have no known total frame count → total_duration = None.
        let total_duration = None;
        let mut decoder = try_make_radio_decoder(&track.codec_params, &DecoderOptions::default())
            .map_err(|e| format!("radio: codec init failed: {e}"))?;
        let mut format = probed.format;

        let mut errors = 0usize;
        let decoded = loop {
            let packet = match format.next_packet() {
                Ok(p) => p,
                Err(_) => break decoder.last_decoded(),
            };
            if packet.track_id() != track_id { continue; }
            match decoder.decode(&packet) {
                Ok(d) => break d,
                Err(symphonia::core::errors::Error::DecodeError(ref msg)) => {
                    errors += 1;
                    eprintln!("[psysonic] radio init: dropped corrupt frame #{errors}: {msg}");
                    if errors >= MAX_CONSECUTIVE_DECODE_ERRORS {
                        return Err("radio: too many consecutive decode errors".into());
                    }
                }
                Err(e) => return Err(format!("radio: decode error: {e}")),
            }
        };
        let spec = decoded.spec().to_owned();
        let buffer = Self::make_buffer(decoded, &spec);
        Ok(SizedDecoder { decoder, current_frame_offset: 0, format, total_duration, buffer, spec, consecutive_decode_errors: 0 })
    }

    #[inline]
    fn make_buffer(decoded: AudioBufferRef, spec: &SignalSpec) -> SampleBuffer<i16> {
        let duration = units::Duration::from(decoded.capacity() as u64);
        let mut buffer = SampleBuffer::<i16>::new(duration, *spec);
        buffer.copy_interleaved_ref(decoded);
        buffer
    }

    /// Refine position after a coarse seek — decode packets until we reach the
    /// exact requested timestamp.
    fn refine_position(
        &mut self,
        seek_res: symphonia::core::formats::SeekedTo,
    ) -> Result<(), String> {
        let mut samples_to_pass = seek_res.required_ts - seek_res.actual_ts;
        let packet = loop {
            let candidate = self.format.next_packet()
                .map_err(|e| format!("refine seek: {e}"))?;
            if candidate.dur() > samples_to_pass {
                break candidate;
            }
            samples_to_pass -= candidate.dur();
        };

        let mut decoded = self.decoder.decode(&packet);
        for _ in 0..DECODE_MAX_RETRIES {
            if decoded.is_err() {
                let p = self.format.next_packet()
                    .map_err(|e| format!("refine retry: {e}"))?;
                decoded = self.decoder.decode(&p);
            }
        }

        let decoded = decoded.map_err(|e| format!("refine decode: {e}"))?;
        decoded.spec().clone_into(&mut self.spec);
        self.buffer = Self::make_buffer(decoded, &self.spec);
        self.current_frame_offset = samples_to_pass as usize * self.spec.channels.count();
        Ok(())
    }
}

impl Iterator for SizedDecoder {
    type Item = i16;

    #[inline]
    fn next(&mut self) -> Option<i16> {
        if self.current_frame_offset >= self.buffer.len() {
            // Loop until a decodable packet is found or the stream ends.
            // DecodeErrors (e.g. MP3 "invalid main_data offset") are non-fatal:
            // drop the frame and advance to the next packet. IO errors and a
            // clean end-of-stream both terminate the iterator normally.
            loop {
                let packet = self.format.next_packet().ok()?;
                match self.decoder.decode(&packet) {
                    Ok(decoded) => {
                        self.consecutive_decode_errors = 0;
                        decoded.spec().clone_into(&mut self.spec);
                        self.buffer = Self::make_buffer(decoded, &self.spec);
                        self.current_frame_offset = 0;
                        break;
                    }
                    Err(symphonia::core::errors::Error::DecodeError(ref msg)) => {
                        #[cfg(not(debug_assertions))]
                        let _ = msg;
                        self.consecutive_decode_errors += 1;
                        // Log sparingly: first drop, then every 10th to avoid spam.
                        #[cfg(debug_assertions)]
                        if self.consecutive_decode_errors == 1
                            || self.consecutive_decode_errors % 10 == 0
                        {
                            eprintln!(
                                "[psysonic] dropped corrupt frame #{}: {msg}",
                                self.consecutive_decode_errors
                            );
                        }
                        if self.consecutive_decode_errors >= MAX_CONSECUTIVE_DECODE_ERRORS {
                            #[cfg(debug_assertions)]
                            eprintln!(
                                "[psysonic] {MAX_CONSECUTIVE_DECODE_ERRORS} consecutive decode \
                                 failures — stream appears unrecoverable, stopping"
                            );
                            return None;
                        }
                        // continue → fetch next packet
                    }
                    Err(_) => return None, // IO error or fatal codec error → end of stream
                }
            }
        }

        let sample = *self.buffer.samples().get(self.current_frame_offset)?;
        self.current_frame_offset += 1;
        Some(sample)
    }
}

impl Source for SizedDecoder {
    #[inline]
    fn current_frame_len(&self) -> Option<usize> {
        Some(self.buffer.samples().len())
    }

    #[inline]
    fn channels(&self) -> u16 {
        self.spec.channels.count() as u16
    }

    #[inline]
    fn sample_rate(&self) -> u32 {
        self.spec.rate
    }

    #[inline]
    fn total_duration(&self) -> Option<Duration> {
        self.total_duration.map(|Time { seconds, frac }| {
            Duration::new(seconds, (frac * 1_000_000_000.0) as u32)
        })
    }

    fn try_seek(&mut self, pos: Duration) -> Result<(), rodio::source::SeekError> {
        let seek_beyond_end = self
            .total_duration()
            .is_some_and(|dur| dur.saturating_sub(pos).as_millis() < 1);

        let time: Time = if seek_beyond_end {
            let t = self.total_duration.unwrap_or(pos.as_secs_f64().into());
            // Step back a tiny bit — some demuxers can't seek to the exact end.
            let mut secs = t.seconds;
            let mut frac = t.frac - 0.0001;
            if frac < 0.0 {
                secs = secs.saturating_sub(1);
                frac = 1.0 - frac;
            }
            Time { seconds: secs, frac }
        } else {
            pos.as_secs_f64().into()
        };

        let to_skip = self.current_frame_offset % self.channels() as usize;

        let seek_res = self
            .format
            .seek(SeekMode::Accurate, SeekTo::Time { time, track_id: None })
            .map_err(|e| rodio::source::SeekError::Other(
                Box::new(std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))
            ))?;

        self.refine_position(seek_res)
            .map_err(|e| rodio::source::SeekError::Other(
                Box::new(std::io::Error::new(std::io::ErrorKind::Other, e))
            ))?;

        self.current_frame_offset += to_skip;
        Ok(())
    }
}

// ─── Encoder-gap trimming (iTunSMPB) ─────────────────────────────────────────
//
// MP3/AAC encoders prepend an "encoder delay" (typically 576–2112 silent
// samples for LAME) and append end-padding to fill the final frame.
// iTunes embeds the exact counts in an ID3v2 COMM frame with description
// "iTunSMPB". Format: " 00000000 DELAY PADDING TOTAL ..."  (space-separated hex)
//
// Parsing strategy: scan raw bytes for the ASCII marker, then extract the
// first whitespace-separated hex tokens after it.

struct GaplessInfo {
    delay_samples: u64,
    total_valid_samples: Option<u64>,
}

impl Default for GaplessInfo {
    fn default() -> Self {
        Self { delay_samples: 0, total_valid_samples: None }
    }
}

fn find_subsequence(data: &[u8], needle: &[u8]) -> Option<usize> {
    data.windows(needle.len()).position(|w| w == needle)
}

fn parse_gapless_info(data: &[u8]) -> GaplessInfo {
    let pos = match find_subsequence(data, b"iTunSMPB") {
        Some(p) => p,
        None => return GaplessInfo::default(),
    };

    // In M4A/iTunes files the key is followed by a binary 'data' atom header
    // (16 bytes: size[4] + "data"[4] + type_flags[4] + locale[4]) before the
    // actual value string. Search for the " 00000000 " sentinel that every
    // iTunSMPB value starts with to locate the true start of the text.
    let search_end = data.len().min(pos + 8 + 128);
    let search_window = &data[pos + 8..search_end];
    let value_start = find_subsequence(search_window, b" 00000000 ")
        .map(|off| pos + 8 + off)
        .unwrap_or(pos + 8);

    let tail = &data[value_start..data.len().min(value_start + 256)];
    let text: String = tail.iter()
        .map(|&b| b as char)
        .filter(|c| c.is_ascii_hexdigit() || *c == ' ')
        .collect();

    let parts: Vec<&str> = text.split_whitespace().collect();
    // parts[0] = "00000000", parts[1] = delay, parts[2] = padding, parts[3] = total
    if parts.len() < 3 {
        return GaplessInfo::default();
    }
    let delay = u64::from_str_radix(parts.get(1).unwrap_or(&"0"), 16).unwrap_or(0);
    let padding = u64::from_str_radix(parts.get(2).unwrap_or(&"0"), 16).unwrap_or(0);
    let total_raw = parts.get(3).and_then(|s| u64::from_str_radix(s, 16).ok());

    let total_valid = total_raw.map(|t| t).filter(|&t| t > 0).or_else(|| {
        // Derive from delay + padding if total not available:
        // Not possible without knowing total encoded samples, so just use None.
        let _ = padding;
        None
    });

    GaplessInfo { delay_samples: delay, total_valid_samples: total_valid }
}

/// Result of build_source: the fully-wrapped source plus metadata and control Arcs.
struct BuiltSource {
    source: CountingSource<NotifyingSource<TriggeredFadeOut<EqualPowerFadeIn<EqSource<DynSource>>>>>,
    duration_secs: f64,
    output_rate: u32,
    output_channels: u16,
    /// Trigger for the sample-level crossfade fade-out.
    fadeout_trigger: Arc<AtomicBool>,
    /// Total samples for the fade-out (set before triggering).
    fadeout_samples: Arc<AtomicU64>,
}

/// Build a fully-prepared playback source:
///   decode → trim → resample → EQ → fade-in → triggered-fade-out → notify → count
///
/// `fade_in_dur`:
///   • `Duration::ZERO`          — unity gain; used for gapless chain (no click)
///   • `Duration::from_millis(5)` — micro-fade; used for hard cuts (anti-click)
///   • `Duration::from_secs_f32(cf)` — full equal-power fade-in for crossfade
///
/// `sample_counter`: atomic counter incremented per sample for drift-free position.
/// `target_rate`: canonical output sample rate for resampling (0 = no resampling).
/// `format_hint`: optional file extension (e.g. "flac", "mp3") to help symphonia probe.
fn build_source(
    data: Vec<u8>,
    duration_hint: f64,
    eq_gains: Arc<[AtomicU32; 10]>,
    eq_enabled: Arc<AtomicBool>,
    eq_pre_gain: Arc<AtomicU32>,
    done_flag: Arc<AtomicBool>,
    fade_in_dur: Duration,
    sample_counter: Arc<AtomicU64>,
    target_rate: u32,
    format_hint: Option<&str>,
    hi_res: bool,
) -> Result<BuiltSource, String> {
    let gapless = parse_gapless_info(&data);

    let decoder = SizedDecoder::new(data, format_hint, hi_res)?;
    let sample_rate = decoder.sample_rate();
    let channels = decoder.channels();

    // Determine effective duration.
    // Prefer hint from Subsonic API (reliable) over decoder (unreliable for VBR MP3).
    let effective_dur = if duration_hint > 1.0 {
        duration_hint
    } else {
        decoder.total_duration()
            .map(|d| d.as_secs_f64())
            .unwrap_or(duration_hint)
    };

    // Apply encoder-delay trim and optional end-padding trim,
    // then resample to the canonical target rate if needed.
    let dyn_src: DynSource = if gapless.delay_samples > 0 || gapless.total_valid_samples.is_some() {
        let delay_dur = Duration::from_secs_f64(
            gapless.delay_samples as f64 / sample_rate as f64
        );
        let base = decoder.convert_samples::<f32>().skip_duration(delay_dur);

        if let Some(total) = gapless.total_valid_samples {
            let valid_dur = Duration::from_secs_f64(total as f64 / sample_rate as f64);
            let trimmed = base.take_duration(valid_dur);
            if target_rate > 0 && sample_rate != target_rate {
                DynSource::new(UniformSourceIterator::new(trimmed, channels, target_rate))
            } else {
                DynSource::new(trimmed)
            }
        } else {
            if target_rate > 0 && sample_rate != target_rate {
                DynSource::new(UniformSourceIterator::new(base, channels, target_rate))
            } else {
                DynSource::new(base)
            }
        }
    } else {
        let converted = decoder.convert_samples::<f32>();
        if target_rate > 0 && sample_rate != target_rate {
            DynSource::new(UniformSourceIterator::new(converted, channels, target_rate))
        } else {
            DynSource::new(converted)
        }
    };

    let output_rate = if target_rate > 0 && sample_rate != target_rate { target_rate } else { sample_rate };

    let fadeout_trigger = Arc::new(AtomicBool::new(false));
    let fadeout_samples = Arc::new(AtomicU64::new(0));

    let eq_src = EqSource::new(dyn_src, eq_gains, eq_enabled, eq_pre_gain);
    let fade_in = EqualPowerFadeIn::new(eq_src, fade_in_dur);
    let fade_out = TriggeredFadeOut::new(fade_in, fadeout_trigger.clone(), fadeout_samples.clone());
    let notifying = NotifyingSource::new(fade_out, done_flag);
    let counting = CountingSource::new(notifying, sample_counter);

    Ok(BuiltSource {
        source: counting,
        duration_secs: effective_dur,
        output_rate,
        output_channels: channels,
        fadeout_trigger,
        fadeout_samples,
    })
}

// ─── Engine state ─────────────────────────────────────────────────────────────

pub(crate) struct PreloadedTrack {
    url: String,
    data: Vec<u8>,
}

/// Info about the track that has been appended (chained) to the current Sink
/// but whose source has not yet started playing (gapless mode only).
pub(crate) struct ChainedInfo {
    /// The URL that was chained — used by audio_play to detect a pre-chain hit.
    url: String,
    /// Raw file bytes (shared with the chained decoder). Lets manual skip reuse
    /// them instead of re-downloading after dropping the Sink queue.
    raw_bytes: Arc<Vec<u8>>,
    duration_secs: f64,
    replay_gain_linear: f32,
    base_volume: f32,
    /// Set by NotifyingSource when this chained track's source is exhausted.
    source_done: Arc<AtomicBool>,
    /// Atomic sample counter for this chained source (swapped into
    /// samples_played on transition).
    sample_counter: Arc<AtomicU64>,
}

pub struct AudioEngine {
    pub stream_handle: Arc<std::sync::Mutex<rodio::OutputStreamHandle>>,
    /// Sample rate the output stream was last opened at (updated on every re-open).
    pub stream_sample_rate: Arc<AtomicU32>,
    /// The rate the device was opened at on cold start — used to restore the
    /// stream when Hi-Res is toggled off while a hi-res rate is active.
    pub device_default_rate: u32,
    /// Sends `(desired_rate, is_hi_res, reply_tx)` to the audio-stream thread to
    /// re-open the output device. `is_hi_res` controls thread-priority escalation.
    pub stream_reopen_tx: std::sync::mpsc::SyncSender<(u32, bool, std::sync::mpsc::SyncSender<rodio::OutputStreamHandle>)>,
    pub current: Arc<Mutex<AudioCurrent>>,
    /// Monotonically incremented on each audio_play (non-chain) / audio_stop call.
    pub generation: Arc<AtomicU64>,
    pub http_client: reqwest::Client,
    pub eq_gains: Arc<[AtomicU32; 10]>,
    pub eq_enabled: Arc<AtomicBool>,
    pub eq_pre_gain: Arc<AtomicU32>,
    pub preloaded: Arc<Mutex<Option<PreloadedTrack>>>,
    pub crossfade_enabled: Arc<AtomicBool>,
    pub crossfade_secs: Arc<AtomicU32>,
    pub fading_out_sink: Arc<Mutex<Option<Sink>>>,
    /// When true, audio_play chains sources to the existing Sink instead of
    /// creating a new one, achieving sample-accurate gapless transitions.
    pub gapless_enabled: Arc<AtomicBool>,
    /// Info about the next-up chained track (gapless mode).
    /// The progress task reads this when `current_source_done` fires.
    pub chained_info: Arc<Mutex<Option<ChainedInfo>>>,
    /// Atomic sample counter — incremented by CountingSource in the audio thread.
    /// Progress task reads this for drift-free position tracking.
    pub samples_played: Arc<AtomicU64>,
    /// Sample rate of the currently playing source (for samples → seconds).
    pub current_sample_rate: Arc<AtomicU32>,
    /// Channel count of the currently playing source.
    pub current_channels: Arc<AtomicU32>,
    /// Instant (as nanos since UNIX epoch via Instant hack) of the last gapless
    /// auto-advance. Commands arriving within 500 ms are rejected as ghost commands.
    pub gapless_switch_at: Arc<AtomicU64>,
    /// Active radio session state.  None for regular (non-radio) tracks.
    /// Dropping the value aborts the HTTP download task via RadioLiveState::Drop.
    pub radio_state: Mutex<Option<RadioLiveState>>,
}

pub struct AudioCurrent {
    pub sink: Option<Sink>,
    pub duration_secs: f64,
    pub seek_offset: f64,
    pub play_started: Option<Instant>,
    pub paused_at: Option<f64>,
    pub replay_gain_linear: f32,
    pub base_volume: f32,
    /// Crossfade: trigger for sample-level fade-out of the current source.
    pub fadeout_trigger: Option<Arc<AtomicBool>>,
    /// Crossfade: total fade samples (set before triggering).
    pub fadeout_samples: Option<Arc<AtomicU64>>,
}

impl AudioCurrent {
    pub fn position(&self) -> f64 {
        if let Some(p) = self.paused_at {
            return p;
        }
        if let Some(t) = self.play_started {
            let elapsed = t.elapsed().as_secs_f64();
            (self.seek_offset + elapsed).min(self.duration_secs.max(0.001))
        } else {
            self.seek_offset
        }
    }
}

/// Open the system default output device at `desired_rate` Hz (0 = device default).
///
/// Resolution order:
///   1. Exact rate match in the device's supported config ranges.
///   2. Highest available rate (for hardware that doesn't support the source rate).
///   3. Device default.
///   4. System default (last resort).
///
/// Returns `(OutputStream, OutputStreamHandle, actual_sample_rate)`.
fn open_stream_for_rate(desired_rate: u32) -> (rodio::OutputStream, rodio::OutputStreamHandle, u32) {
    use rodio::cpal::traits::{DeviceTrait, HostTrait};

    let host = rodio::cpal::default_host();

    if let Some(device) = host.default_output_device() {
        if desired_rate > 0 {
            if let Ok(supported) = device.supported_output_configs() {
                let configs: Vec<_> = supported.collect();

                // 1. Exact rate match — prefer more channels (stereo > mono).
                let exact = configs.iter()
                    .filter(|c| {
                        c.min_sample_rate().0 <= desired_rate
                            && desired_rate <= c.max_sample_rate().0
                    })
                    .max_by_key(|c| c.channels());

                if let Some(cfg) = exact {
                    let config = cfg.clone()
                        .with_sample_rate(rodio::cpal::SampleRate(desired_rate));
                    if let Ok((stream, handle)) =
                        rodio::OutputStream::try_from_device_config(&device, config)
                    {
                        eprintln!("[psysonic] audio stream opened at {} Hz (exact)", desired_rate);
                        return (stream, handle, desired_rate);
                    }
                }

                // 2. No exact match — use the highest supported rate.
                let best = configs.iter()
                    .max_by_key(|c| c.max_sample_rate().0);

                if let Some(cfg) = best {
                    let rate = cfg.max_sample_rate().0;
                    let config = cfg.clone()
                        .with_sample_rate(rodio::cpal::SampleRate(rate));
                    if let Ok((stream, handle)) =
                        rodio::OutputStream::try_from_device_config(&device, config)
                    {
                        eprintln!(
                            "[psysonic] audio stream opened at {} Hz (highest, wanted {})",
                            rate, desired_rate
                        );
                        return (stream, handle, rate);
                    }
                }
            }
        }

        // 3. Device default.
        if let Ok((stream, handle)) = rodio::OutputStream::try_from_device(&device) {
            let rate = device
                .default_output_config()
                .map(|c| c.sample_rate().0)
                .unwrap_or(44100);
            eprintln!("[psysonic] audio stream opened at {} Hz (device default)", rate);
            return (stream, handle, rate);
        }
    }

    // 4. Last resort: system default.
    eprintln!("[psysonic] audio stream falling back to system default");
    let (stream, handle) = rodio::OutputStream::try_default()
        .expect("cannot open any audio output device");
    let rate = rodio::cpal::default_host()
        .default_output_device()
        .and_then(|d| d.default_output_config().ok())
        .map(|c| c.sample_rate().0)
        .unwrap_or(44100);
    (stream, handle, rate)
}

pub fn create_engine() -> (AudioEngine, std::thread::JoinHandle<()>) {
    // macOS: request a smaller CoreAudio buffer to reduce output latency.
    #[cfg(target_os = "macos")]
    {
        if std::env::var("COREAUDIO_BUFFER_SIZE").is_err() {
            std::env::set_var("COREAUDIO_BUFFER_SIZE", "512");
        }
    }

    // Channels: main thread ←→ audio-stream thread.
    //   init_tx/rx : (OutputStreamHandle, actual_rate) sent once at startup.
    //   reopen_tx/rx: (desired_rate, reply_tx) — triggers a stream re-open.
    let (init_tx, init_rx) =
        std::sync::mpsc::sync_channel::<(rodio::OutputStreamHandle, u32)>(0);
    let (reopen_tx, reopen_rx) =
        std::sync::mpsc::sync_channel::<(u32, bool, std::sync::mpsc::SyncSender<rodio::OutputStreamHandle>)>(4);

    let thread = std::thread::Builder::new()
        .name("psysonic-audio-stream".into())
        .spawn(move || {
            // Set PipeWire / PulseAudio latency hints before the first open.
            #[cfg(target_os = "linux")]
            {
                if std::env::var("PIPEWIRE_LATENCY").is_err() {
                    std::env::set_var("PIPEWIRE_LATENCY", "4096/48000");
                }
                if std::env::var("PULSE_LATENCY_MSEC").is_err() {
                    std::env::set_var("PULSE_LATENCY_MSEC", "85");
                }
            }

            // Thread priority is kept at default during standard-mode playback.
            // It is escalated to Max only when a Hi-Res stream reopen is requested,
            // to prevent PipeWire underruns at high quantum sizes (8192 frames).
            let (mut _stream, handle, rate) = open_stream_for_rate(0);
            init_tx.send((handle, rate)).ok();

            // Keep the stream alive and handle sample-rate switch requests.
            while let Ok((desired_rate, is_hi_res, reply_tx)) = reopen_rx.recv() {
                // Escalate to Max for Hi-Res reopens (large PipeWire quanta need
                // real-time scheduling to avoid underruns). No escalation for
                // standard mode — the thread blocks on recv() between reopens so
                // elevated priority would only waste scheduler budget.
                if is_hi_res {
                    thread_priority::set_current_thread_priority(
                        thread_priority::ThreadPriority::Max
                    ).ok();
                }

                drop(_stream); // close old stream before opening new one

                // Scale the PipeWire quantum with the sample rate so wall-clock
                // latency stays roughly constant (≈93 ms) at all rates.
                // 8192 frames at 88200 Hz ≈ 92.9 ms (same as 4096 at 48000 Hz).
                #[cfg(target_os = "linux")]
                {
                    let frames: u32 = if desired_rate > 48_000 { 8192 } else { 4096 };
                    std::env::set_var("PIPEWIRE_LATENCY", format!("{frames}/{desired_rate}"));
                    // Keep PULSE_LATENCY_MSEC in sync so PulseAudio-based setups
                    // get the same wall-clock quantum as PipeWire.
                    let latency_ms = (frames as f64 / desired_rate as f64 * 1000.0).round() as u64;
                    std::env::set_var("PULSE_LATENCY_MSEC", latency_ms.to_string());
                }

                let (new_stream, new_handle, _actual) = open_stream_for_rate(desired_rate);
                _stream = new_stream;
                reply_tx.send(new_handle).ok();
            }
        })
        .expect("spawn audio stream thread");

    let (initial_handle, initial_rate) = init_rx.recv().expect("audio stream handle");

    let engine = AudioEngine {
        stream_handle: Arc::new(std::sync::Mutex::new(initial_handle)),
        stream_sample_rate: Arc::new(AtomicU32::new(initial_rate)),
        device_default_rate: initial_rate,
        stream_reopen_tx: reopen_tx,
        current: Arc::new(Mutex::new(AudioCurrent {
            sink: None,
            duration_secs: 0.0,
            seek_offset: 0.0,
            play_started: None,
            paused_at: None,
            replay_gain_linear: 1.0,
            base_volume: 0.8,
            fadeout_trigger: None,
            fadeout_samples: None,
        })),
        generation: Arc::new(AtomicU64::new(0)),
        http_client: reqwest::Client::builder()
            .timeout(Duration::from_secs(30))
            .use_rustls_tls()
            .user_agent(format!("psysonic/{}", env!("CARGO_PKG_VERSION")))
            .build()
            .unwrap_or_default(),
        eq_gains: Arc::new(std::array::from_fn(|_| AtomicU32::new(0f32.to_bits()))),
        eq_enabled: Arc::new(AtomicBool::new(false)),
        eq_pre_gain: Arc::new(AtomicU32::new(0f32.to_bits())),
        preloaded: Arc::new(Mutex::new(None)),
        crossfade_enabled: Arc::new(AtomicBool::new(false)),
        crossfade_secs: Arc::new(AtomicU32::new(3.0f32.to_bits())),
        fading_out_sink: Arc::new(Mutex::new(None)),
        gapless_enabled: Arc::new(AtomicBool::new(false)),
        chained_info: Arc::new(Mutex::new(None)),
        samples_played: Arc::new(AtomicU64::new(0)),
        current_sample_rate: Arc::new(AtomicU32::new(0)),
        current_channels: Arc::new(AtomicU32::new(2)),
        gapless_switch_at: Arc::new(AtomicU64::new(0)),
        radio_state: Mutex::new(None),
    };

    (engine, thread)
}

// ─── Event payloads ───────────────────────────────────────────────────────────

#[derive(Clone, Serialize)]
pub struct ProgressPayload {
    pub current_time: f64,
    pub duration: f64,
}

// ─── Helpers ──────────────────────────────────────────────────────────────────

/// Subsonic `buildStreamUrl()` uses a fresh random salt on every call, so two
/// URLs for the same track differ in `t`/`s` query params. Compare a stable key.
fn playback_identity(url: &str) -> Option<String> {
    if let Some(path) = url.strip_prefix("psysonic-local://") {
        return Some(format!("local:{path}"));
    }
    if !url.contains("stream.view") {
        return None;
    }
    let q = url.split('?').nth(1)?;
    for pair in q.split('&') {
        if let Some(v) = pair.strip_prefix("id=") {
            let v = v.split('&').next().unwrap_or(v);
            return Some(format!("stream:{v}"));
        }
    }
    None
}

fn same_playback_target(a_url: &str, b_url: &str) -> bool {
    match (playback_identity(a_url), playback_identity(b_url)) {
        (Some(a), Some(b)) => a == b,
        _ => a_url == b_url,
    }
}

/// Fetch track bytes from the preload cache or via HTTP.
async fn fetch_data(
    url: &str,
    state: &AudioEngine,
    gen: u64,
    app: &AppHandle,
) -> Result<Option<Vec<u8>>, String> {
    // Check preload cache first.
    let cached = {
        let mut preloaded = state.preloaded.lock().unwrap();
        if preloaded.as_ref().is_some_and(|p| same_playback_target(&p.url, url)) {
            preloaded.take().map(|p| p.data)
        } else {
            None
        }
    };

    if let Some(data) = cached {
        return Ok(Some(data));
    }

    // Offline cache — local file written by download_track_offline.
    if let Some(path) = url.strip_prefix("psysonic-local://") {
        let data = tokio::fs::read(path).await.map_err(|e| e.to_string())?;
        return Ok(Some(data));
    }

    let response = state.http_client.get(url).send().await.map_err(|e| e.to_string())?;
    #[cfg(debug_assertions)]
    {
        let status = response.status();
        let ct = response.headers()
            .get(reqwest::header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .unwrap_or("-");
        let server_hdr = response.headers()
            .get("server")
            .and_then(|v| v.to_str().ok())
            .unwrap_or("-");
        // Strip auth params from URL before logging.
        let safe_url = url.split('?').next().unwrap_or(url);
        eprintln!(
            "[audio] fetch {} → {} | content-type: {} | server: {}",
            safe_url, status, ct, server_hdr
        );
    }
    if !response.status().is_success() {
        if state.generation.load(Ordering::SeqCst) != gen {
            return Ok(None); // superseded
        }
        let status = response.status().as_u16();
        let msg = format!("HTTP {status}");
        app.emit("audio:error", &msg).ok();
        return Err(msg);
    }
    // Stream the body, checking gen between chunks so a rapid manual skip can
    // abort a superseded download mid-flight and free bandwidth for the new one.
    let hint = response.content_length().unwrap_or(0) as usize;
    let mut stream = response.bytes_stream();
    let mut data = Vec::with_capacity(hint);
    while let Some(chunk) = stream.next().await {
        if state.generation.load(Ordering::SeqCst) != gen {
            return Ok(None); // superseded — abort
        }
        data.extend_from_slice(&chunk.map_err(|e| e.to_string())?);
    }
    Ok(Some(data))
}

/// -1 dB headroom applied at full scale to prevent inter-sample clipping.
/// Modern masters are often at 0 dBFS; the EQ biquad chain and resampler
/// can produce inter-sample peaks slightly above ±1.0 → audible distortion.
/// 10^(-1/20) ≈ 0.891 — inaudible volume difference, eliminates clipping.
const MASTER_HEADROOM: f32 = 0.891_254;

fn compute_gain(
    replay_gain_db: Option<f32>,
    replay_gain_peak: Option<f32>,
    pre_gain_db: f32,
    fallback_db: f32,
    volume: f32,
) -> (f32, f32) {
    let gain_linear = replay_gain_db
        .map(|db| 10f32.powf((db + pre_gain_db) / 20.0))
        .unwrap_or_else(|| 10f32.powf(fallback_db / 20.0));
    let peak = replay_gain_peak.unwrap_or(1.0).max(0.001);
    let gain_linear = gain_linear.min(1.0 / peak);
    let effective = (volume.clamp(0.0, 1.0) * gain_linear * MASTER_HEADROOM).clamp(0.0, 1.0);
    (gain_linear, effective)
}

// ─── Commands ─────────────────────────────────────────────────────────────────

#[tauri::command]
pub async fn audio_play(
    url: String,
    volume: f32,
    duration_hint: f64,
    replay_gain_db: Option<f32>,
    replay_gain_peak: Option<f32>,
    pre_gain_db: f32,
    fallback_db: f32,
    manual: bool, // true = user-initiated skip → bypass crossfade, start immediately
    hi_res_enabled: bool, // false = safe 44.1 kHz mode; true = native rate (alpha)
    app: AppHandle,
    state: State<'_, AudioEngine>,
) -> Result<(), String> {
    let gapless = state.gapless_enabled.load(Ordering::Relaxed);

    // ── Ghost-command guard ───────────────────────────────────────────────────
    // After a gapless auto-advance, the frontend may fire a stale playTrack()
    // call via IPC. If we're within 500 ms of the last gapless switch AND the
    // requested URL matches the already-playing chained track, reject it.
    {
        let switch_ms = state.gapless_switch_at.load(Ordering::SeqCst);
        if switch_ms > 0 {
            let now_ms = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_millis() as u64;
            if now_ms.saturating_sub(switch_ms) < 500 {
                // Within the guard window — suppress this ghost command.
                return Ok(());
            }
        }
    }

    // ── Gapless pre-chain hit ─────────────────────────────────────────────────
    // audio_chain_preload already appended this URL to the Sink 30 s in
    // advance. The source is live in the queue — just return and let the
    // progress task handle the state transition when the previous source ends.
    //
    // Never for manual skips: the UI already jumped to this track in JS, but
    // the current source is still playing until the chain drains. User-initiated
    // play must clear the chain and start this URL immediately (standard path).
    if gapless && !manual {
        let already_chained = state.chained_info.lock().unwrap()
            .as_ref()
            .map(|c| same_playback_target(&c.url, &url))
            .unwrap_or(false);
        if already_chained {
            return Ok(());
        }
    }

    // ── Standard (new-sink) path ─────────────────────────────────────────────
    // Used for: manual skip, gapless OFF, first play, or gapless when the
    // proactive chain was not set up in time.

    // Bump generation first so the old progress task stops before we peel
    // chained_info (avoids a race where it sees current_done + empty chain).
    let gen = state.generation.fetch_add(1, Ordering::SeqCst) + 1;

    // Manual skip onto the gapless-pre-chained track: reuse raw bytes (no HTTP;
    // preload cache was already consumed when the chain was built). Otherwise
    // clear any stale chain metadata.
    let reuse_chained_bytes: Option<Vec<u8>> = if gapless && manual {
        let mut ci = state.chained_info.lock().unwrap();
        if ci.as_ref().is_some_and(|c| same_playback_target(&c.url, &url)) {
            ci.take().map(|info| {
                Arc::try_unwrap(info.raw_bytes).unwrap_or_else(|a| (*a).clone())
            })
        } else {
            *ci = None;
            None
        }
    } else {
        *state.chained_info.lock().unwrap() = None;
        None
    };

    // Stop fading-out sink from previous crossfade.
    if let Some(old) = state.fading_out_sink.lock().unwrap().take() {
        old.stop();
    }

    // Fetch bytes (preload cache) unless we reused the chained download above.
    let data = if let Some(d) = reuse_chained_bytes {
        Some(d)
    } else {
        fetch_data(&url, &state, gen, &app).await?
    };
    let data = match data {
        Some(d) => d,
        None => return Ok(()), // superseded while downloading
    };

    if state.generation.load(Ordering::SeqCst) != gen {
        return Ok(());
    }

    let (gain_linear, effective_volume) = compute_gain(replay_gain_db, replay_gain_peak, pre_gain_db, fallback_db, volume);

    // Manual skips (user-initiated) bypass crossfade — the track should start immediately.
    let crossfade_enabled = state.crossfade_enabled.load(Ordering::Relaxed) && !manual;
    let crossfade_secs_val = f32::from_bits(state.crossfade_secs.load(Ordering::Relaxed)).clamp(0.5, 12.0);

    // Measure how much audio Track A actually has left right now.
    // By the time audio_play is called, near_end_ticks (2×500ms) + IPC latency
    // have consumed ~500–800ms from Track A's tail — so its true remaining time
    // is always less than crossfade_secs_val.  Using the measured remaining time
    // for BOTH fade-out (Track A) and fade-in (Track B) keeps them in sync and
    // guarantees Track A reaches 0 exactly when its source exhausts.
    let actual_fade_secs: f32 = if crossfade_enabled {
        let cur = state.current.lock().unwrap();
        let remaining = (cur.duration_secs - cur.position()) as f32;
        remaining.clamp(0.1, crossfade_secs_val)
    } else {
        0.0
    };

    // Fade-in duration for Track B:
    //   crossfade → equal-power sin(t·π/2) over actual remaining time of Track A
    //   hard cut  → 5 ms micro-fade to suppress DC-offset click
    let fade_in_dur = if crossfade_enabled {
        Duration::from_secs_f32(actual_fade_secs)
    } else {
        Duration::from_millis(5)
    };

    // Build source: decode → trim → resample → EQ → fade-in → fade-out → notify → count.
    let done_flag = Arc::new(AtomicBool::new(false));
    // Reset sample counter for the new track.
    state.samples_played.store(0, Ordering::Relaxed);
    // Always 0 — no application-level resampling. Rodio handles conversion to
    // the output device rate internally; we let every track play at its native rate.
    let target_rate: u32 = 0;
    // Extract format hint from URL for better symphonia probing.
    let format_hint = url.rsplit('.').next()
        .and_then(|ext| ext.split('?').next())
        .map(|s| s.to_lowercase());
    let built = build_source(
        data,
        duration_hint,
        state.eq_gains.clone(),
        state.eq_enabled.clone(),
        state.eq_pre_gain.clone(),
        done_flag.clone(),
        fade_in_dur,
        state.samples_played.clone(),
        target_rate,
        format_hint.as_deref(),
        hi_res_enabled,
    ).map_err(|e| { app.emit("audio:error", &e).ok(); e })?;
    let source = built.source;
    let duration_secs = built.duration_secs;
    let output_rate = built.output_rate;
    let output_channels = built.output_channels;

    // Store the actual output rate/channels for position calculation.
    state.current_sample_rate.store(output_rate, Ordering::Relaxed);
    state.current_channels.store(output_channels as u32, Ordering::Relaxed);

    if state.generation.load(Ordering::SeqCst) != gen {
        return Ok(());
    }

    // ── Stream rate management ────────────────────────────────────────────────
    // Hi-Res ON:  open device at file's native rate (bit-perfect, no resampler).
    // Hi-Res OFF: if the stream was previously opened at a hi-res rate (e.g. the
    //             toggle was just turned off mid-session), restore the device
    //             default rate so playback is no longer at 88.2/96 kHz etc.
    //             If already at the device default — skip entirely (no IPC, no
    //             PipeWire reconfigure, no scheduler cost).
    {
        let current_stream_rate = state.stream_sample_rate.load(Ordering::Relaxed);
        let target_rate = if hi_res_enabled {
            output_rate   // native file rate
        } else {
            state.device_default_rate  // restore device default
        };
        let needs_switch = target_rate > 0 && target_rate != current_stream_rate;
        if needs_switch {
            let (reply_tx, reply_rx) = std::sync::mpsc::sync_channel::<rodio::OutputStreamHandle>(0);
            if state.stream_reopen_tx.send((target_rate, hi_res_enabled, reply_tx)).is_ok() {
                match reply_rx.recv_timeout(std::time::Duration::from_secs(5)) {
                    Ok(new_handle) => {
                        *state.stream_handle.lock().unwrap() = new_handle;
                        state.stream_sample_rate.store(target_rate, Ordering::Relaxed);
                        // Give PipeWire time to reconfigure at the new rate before
                        // we open a Sink — only needed for large hi-res quanta.
                        if hi_res_enabled && target_rate > 48_000 {
                            tokio::time::sleep(Duration::from_millis(150)).await;
                        }
                    }
                    Err(_) => {
                        eprintln!("[psysonic] stream rate switch timed out, keeping {current_stream_rate} Hz");
                    }
                }
            }
        }

        // Re-check gen: a rapid skip during the settle sleep would have bumped it.
        if state.generation.load(Ordering::SeqCst) != gen {
            return Ok(());
        }
    }

    let sink = Sink::try_new(&*state.stream_handle.lock().unwrap()).map_err(|e| e.to_string())?;
    sink.set_volume(effective_volume);

    // ── Sink pre-fill for hi-res tracks ──────────────────────────────────────
    // At sample rates > 48 kHz the hardware quantum is larger and the first
    // period demands more decoded frames than at 44.1/48 kHz.
    // Strategy: pause the sink before appending so rodio's internal mixer
    // decodes into its ring buffer ahead of the hardware. After a short delay
    // we resume — the buffer is already full and the hardware gets its frames
    // without an underrun on the very first period.
    // Standard mode: no pre-fill needed — default 44.1/48 kHz quantum is small.
    let needs_prefill = hi_res_enabled && output_rate > 48_000;
    if needs_prefill {
        sink.pause();
    }

    // Gapless OFF: prepend a short silence so tracks are clearly separated.
    // Only when this is an auto-advance (near end), not on manual skip.
    if !gapless {
        let cur_pos = {
            let cur = state.current.lock().unwrap();
            cur.position()
        };
        let cur_dur = {
            let cur = state.current.lock().unwrap();
            cur.duration_secs
        };
        let is_auto_advance = cur_dur > 3.0 && cur_pos >= cur_dur - 3.0;
        if is_auto_advance {
            let silence = rodio::source::Zero::<f32>::new(
                source.channels(),
                source.sample_rate(),
            ).take_duration(Duration::from_millis(500));
            sink.append(silence);
        }
    }

    sink.append(source);

    if needs_prefill {
        // 500 ms lets rodio decode several seconds of hi-res audio into its
        // internal buffer while the sink is paused. The hardware sees no gap
        // because the output is held — it only starts draining after sink.play().
        // 500 ms gives ~5 quanta of headroom at 8192-frame/88200 Hz quantum size,
        // absorbing scheduler jitter and PipeWire graph wake-up latency.
        tokio::time::sleep(Duration::from_millis(500)).await;
        if state.generation.load(Ordering::SeqCst) != gen {
            return Ok(()); // skipped during pre-fill — abort silently
        }
        sink.play();
    }

    // Atomically swap sinks — extract old sink + its fade-out trigger.
    let (old_sink, old_fadeout_trigger, old_fadeout_samples) = {
        let mut cur = state.current.lock().unwrap();
        let old = cur.sink.take();
        let old_fo_trigger = cur.fadeout_trigger.take();
        let old_fo_samples = cur.fadeout_samples.take();
        cur.sink = Some(sink);
        cur.duration_secs = duration_secs;
        cur.seek_offset = 0.0;
        cur.play_started = Some(Instant::now());
        cur.paused_at = None;
        cur.replay_gain_linear = gain_linear;
        cur.base_volume = volume.clamp(0.0, 1.0);
        cur.fadeout_trigger = Some(built.fadeout_trigger);
        cur.fadeout_samples = Some(built.fadeout_samples);
        (old, old_fo_trigger, old_fo_samples)
    };

    // Handle old sink: symmetric crossfade or immediate stop.
    if crossfade_enabled {
        if let Some(old) = old_sink {
            // Trigger sample-level fade-out on Track A via TriggeredFadeOut.
            // Calculate total fade samples from the measured actual_fade_secs.
            let rate = state.current_sample_rate.load(Ordering::Relaxed);
            let ch = state.current_channels.load(Ordering::Relaxed);
            let fade_total = (actual_fade_secs as f64 * rate as f64 * ch as f64) as u64;

            if let (Some(trigger), Some(samples)) = (old_fadeout_trigger, old_fadeout_samples) {
                samples.store(fade_total.max(1), Ordering::SeqCst);
                trigger.store(true, Ordering::SeqCst);
            }

            // Keep old sink alive until the fade completes + small margin,
            // then drop it. No volume stepping needed — the fade-out runs
            // at sample level inside the audio thread.
            *state.fading_out_sink.lock().unwrap() = Some(old);
            let fo_arc = state.fading_out_sink.clone();
            let cleanup_dur = Duration::from_secs_f32(actual_fade_secs + 0.5);
            tokio::spawn(async move {
                tokio::time::sleep(cleanup_dur).await;
                if let Some(s) = fo_arc.lock().unwrap().take() {
                    s.stop();
                }
            });
        }
    } else if let Some(old) = old_sink {
        old.stop();
    }

    app.emit("audio:playing", duration_secs).ok();

    // ── Progress + ended detection ────────────────────────────────────────────
    spawn_progress_task(
        gen,
        state.generation.clone(),
        state.current.clone(),
        state.chained_info.clone(),
        state.crossfade_enabled.clone(),
        state.crossfade_secs.clone(),
        done_flag,
        app,
        state.samples_played.clone(),
        state.current_sample_rate.clone(),
        state.current_channels.clone(),
        state.gapless_switch_at.clone(),
    );

    Ok(())
}

/// Proactively appends the next track to the current Sink ~30 s before the
/// current track ends. Called from JS at the same trigger point as preload.
///
/// Because this runs well before the track boundary, the IPC round-trip is
/// irrelevant — by the time the current track actually ends, the next source
/// is already live in the Sink queue and rodio transitions at sample accuracy.
///
/// audio_play() checks chained_info.url on arrival: if it matches, it returns
/// immediately without touching the Sink (pure no-op on the audio path).
#[tauri::command]
pub async fn audio_chain_preload(
    url: String,
    volume: f32,
    duration_hint: f64,
    replay_gain_db: Option<f32>,
    replay_gain_peak: Option<f32>,
    pre_gain_db: f32,
    fallback_db: f32,
    hi_res_enabled: bool,
    state: State<'_, AudioEngine>,
) -> Result<(), String> {
    // Idempotent: already chained this track → nothing to do.
    {
        let chained = state.chained_info.lock().unwrap();
        if chained.as_ref().is_some_and(|c| same_playback_target(&c.url, &url)) {
            return Ok(());
        }
    }

    // Gapless must be enabled and a sink must exist.
    if !state.gapless_enabled.load(Ordering::Relaxed) {
        return Ok(());
    }

    let snapshot_gen = state.generation.load(Ordering::SeqCst);

    // Fetch bytes — use preload cache if available, otherwise HTTP.
    let data: Vec<u8> = {
        let cached = {
            let mut preloaded = state.preloaded.lock().unwrap();
            if preloaded.as_ref().is_some_and(|p| same_playback_target(&p.url, &url)) {
                preloaded.take().map(|p| p.data)
            } else {
                None
            }
        };
        if let Some(d) = cached {
            d
        } else {
            if let Some(path) = url.strip_prefix("psysonic-local://") {
                tokio::fs::read(path).await.map_err(|e| e.to_string())?
            } else {
                let resp = state.http_client.get(&url).send().await
                    .map_err(|e| e.to_string())?;
                if !resp.status().is_success() {
                    return Ok(()); // silently fail — audio_play will retry
                }
                let hint = resp.content_length().unwrap_or(0) as usize;
                let mut stream = resp.bytes_stream();
                let mut buf = Vec::with_capacity(hint);
                while let Some(chunk) = stream.next().await {
                    if state.generation.load(Ordering::SeqCst) != snapshot_gen {
                        return Ok(()); // superseded by manual skip — abort download
                    }
                    buf.extend_from_slice(&chunk.map_err(|e| e.to_string())?);
                }
                buf
            }
        }
    };

    // Bail if the user skipped to a different track while we were downloading.
    if state.generation.load(Ordering::SeqCst) != snapshot_gen {
        return Ok(());
    }

    let raw_bytes = Arc::new(data);

    let (gain_linear, effective_volume) = compute_gain(replay_gain_db, replay_gain_peak, pre_gain_db, fallback_db, volume);

    let done_next = Arc::new(AtomicBool::new(false));
    // Use a dedicated counter for the chained source — it will be swapped into
    // samples_played when the chained track becomes active.
    let chain_counter = Arc::new(AtomicU64::new(0));
    // Always 0 — no application-level resampling (same as audio_play).
    let target_rate: u32 = 0;
    let format_hint = url.rsplit('.').next()
        .and_then(|ext| ext.split('?').next())
        .map(|s| s.to_lowercase());
    let built = build_source(
        (*raw_bytes).clone(),
        duration_hint,
        state.eq_gains.clone(),
        state.eq_enabled.clone(),
        state.eq_pre_gain.clone(),
        done_next.clone(),
        Duration::ZERO, // gapless: no fade-in — sample-accurate boundary, no click
        chain_counter.clone(),
        target_rate,
        format_hint.as_deref(),
        hi_res_enabled,
    ).map_err(|e| e.to_string())?;
    let source = built.source;
    let duration_secs = built.duration_secs;

    // Final gen check — reject if a manual skip happened during decode.
    if state.generation.load(Ordering::SeqCst) != snapshot_gen {
        return Ok(());
    }

    // In hi-res mode: if the next track's native rate differs from the current
    // output stream, we cannot chain gaplessly — audio_play will do a hard cut
    // with a stream re-open. Store raw bytes to avoid re-downloading.
    // In safe mode (44.1 kHz locked): the stream rate is always 44100, so the
    // chain proceeds and rodio resamples internally — no bail needed.
    let next_rate = if hi_res_enabled { built.output_rate } else { 44_100 };
    let stream_rate = state.stream_sample_rate.load(Ordering::Relaxed);
    if hi_res_enabled && stream_rate > 0 && next_rate != stream_rate {
        eprintln!(
            "[psysonic] gapless chain skipped: next track rate {} Hz ≠ stream {} Hz",
            next_rate, stream_rate
        );
        *state.preloaded.lock().unwrap() = Some(PreloadedTrack {
            url,
            data: Arc::try_unwrap(raw_bytes).unwrap_or_else(|a| (*a).clone()),
        });
        return Ok(());
    }

    // Append to the existing Sink. The audio hardware stream never stalls.
    {
        let cur = state.current.lock().unwrap();
        match &cur.sink {
            Some(sink) => {
                sink.set_volume(effective_volume);
                sink.append(source);
            }
            None => return Ok(()), // playback stopped — bail
        }
    }

    *state.chained_info.lock().unwrap() = Some(ChainedInfo {
        url,
        raw_bytes,
        duration_secs,
        replay_gain_linear: gain_linear,
        base_volume: volume.clamp(0.0, 1.0),
        source_done: done_next,
        sample_counter: chain_counter,
    });

    Ok(())
}

/// Spawns the per-generation progress + ended-detection task.
///
/// The task owns a local `done: Arc<AtomicBool>` reference that starts as
/// the current track's done flag. When the progress task detects that the
/// done flag is set AND `chained_info` has data, it swaps `done` to the
/// chained source's flag and transitions state — all without creating a new
/// task or changing the generation counter.
///
/// Key changes from the previous implementation:
///   • 100 ms tick (was 500 ms) — halves worst-case event latency
///   • Position from atomic sample counter (no wall-clock drift)
///   • Immediate `audio:track_switched` event at decoder boundary
///   • `audio:ended` only fires when no chained successor exists
fn spawn_progress_task(
    gen: u64,
    gen_counter: Arc<AtomicU64>,
    current_arc: Arc<Mutex<AudioCurrent>>,
    chained_arc: Arc<Mutex<Option<ChainedInfo>>>,
    crossfade_enabled_arc: Arc<AtomicBool>,
    crossfade_secs_arc: Arc<AtomicU32>,
    initial_done: Arc<AtomicBool>,
    app: AppHandle,
    samples_played: Arc<AtomicU64>,
    sample_rate_arc: Arc<AtomicU32>,
    channels_arc: Arc<AtomicU32>,
    gapless_switch_at: Arc<AtomicU64>,
) {
    tokio::spawn(async move {
        let mut near_end_ticks: u32 = 0;
        // Local done-flag reference; swapped on gapless transition.
        let mut current_done = initial_done;
        // Local sample counter; swapped to chained source's counter on transition.
        let mut samples_played = samples_played;

        loop {
            // 500 ms tick — frontend interpolates visually at 60 fps via rAF.
            tokio::time::sleep(Duration::from_millis(500)).await;

            if gen_counter.load(Ordering::SeqCst) != gen {
                break;
            }

            // ── Gapless transition detection ─────────────────────────────────
            // If the current source is exhausted AND we have a chained track
            // ready, transition seamlessly: swap tracking state, emit
            // audio:track_switched for the new track, and continue the loop.
            if current_done.load(Ordering::SeqCst) {
                // Radio (dur == 0): stream exhausted / connection dropped → stop.
                let cur_dur = current_arc.lock().unwrap().duration_secs;
                if cur_dur <= 0.0 {
                    eprintln!("[radio] current_done fired → emitting audio:ended (dur=0)");
                    gen_counter.fetch_add(1, Ordering::SeqCst);
                    app.emit("audio:ended", ()).ok();
                    break;
                }

                let chained = chained_arc.lock().unwrap().take();
                if let Some(info) = chained {
                    // Swap to the chained source's done flag.
                    current_done = info.source_done;

                    // Swap to the chained source's sample counter.
                    // The chained CountingSource increments its own Arc,
                    // so we must rebind our local reference to it —
                    // a one-time value copy would go stale immediately.
                    samples_played = info.sample_counter;

                    // Update tracking state.
                    {
                        let mut cur = current_arc.lock().unwrap();
                        cur.replay_gain_linear = info.replay_gain_linear;
                        cur.base_volume = info.base_volume;
                        cur.duration_secs = info.duration_secs;
                        cur.seek_offset = 0.0;
                        cur.play_started = Some(Instant::now());
                    }

                    // Record the gapless switch timestamp for ghost-command guard.
                    let switch_ts = std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_millis() as u64;
                    gapless_switch_at.store(switch_ts, Ordering::SeqCst);

                    // Emit the new track_switched event — this is immediate,
                    // not delayed by 500 ms like the old audio:playing was.
                    app.emit("audio:track_switched", info.duration_secs).ok();
                    near_end_ticks = 0;
                    continue;
                }
                // Current source exhausted but no chain queued — the Sink is
                // likely draining; audio:ended will fire on the next tick via
                // the near-end logic below.
            }

            // ── Position from atomic sample counter ──────────────────────────
            let rate = sample_rate_arc.load(Ordering::Relaxed) as f64;
            let ch = channels_arc.load(Ordering::Relaxed) as f64;
            let samples = samples_played.load(Ordering::Relaxed) as f64;
            let divisor = (rate * ch).max(1.0);

            let dur = {
                let cur = current_arc.lock().unwrap();
                cur.duration_secs
            };
            let is_paused = {
                let cur = current_arc.lock().unwrap();
                cur.paused_at.is_some()
            };

            let pos = if is_paused {
                let cur = current_arc.lock().unwrap();
                cur.paused_at.unwrap_or(0.0)
            } else {
                (samples / divisor).min(dur.max(0.001))
            };

            app.emit("audio:progress", ProgressPayload { current_time: pos, duration: dur }).ok();

            if is_paused {
                continue;
            }

            let cf_enabled = crossfade_enabled_arc.load(Ordering::Relaxed);
            let cf_secs = f32::from_bits(crossfade_secs_arc.load(Ordering::Relaxed)).clamp(0.5, 12.0) as f64;
            let end_threshold = if cf_enabled { cf_secs.max(1.0) } else { 1.0 };

            if dur > end_threshold && pos >= dur - end_threshold {
                near_end_ticks += 1;
                // At 100 ms ticks, 10 ticks ≈ 1 s — equivalent to the old 2×500ms.
                if near_end_ticks >= 10 {
                    // If a gapless chain is pending, the source hasn't
                    // exhausted yet — duration_hint (integer seconds from
                    // Subsonic) is shorter than the actual audio content.
                    // Don't emit audio:ended; let the gapless transition
                    // handle it when current_done fires.
                    let has_chain = chained_arc.lock().unwrap().is_some();
                    if has_chain {
                        continue;
                    }
                    gen_counter.fetch_add(1, Ordering::SeqCst);
                    app.emit("audio:ended", ()).ok();
                    break;
                }
            } else {
                near_end_ticks = 0;
            }
        }
    });
}

#[tauri::command]
pub fn audio_pause(state: State<'_, AudioEngine>) {
    let mut cur = state.current.lock().unwrap();
    if let Some(sink) = &cur.sink {
        if !sink.is_paused() {
            let pos = cur.position();
            sink.pause();
            cur.paused_at    = Some(pos);
            cur.play_started = None;
        }
    }
    // Notify the download task so it can start measuring the hard-pause stall timer.
    if let Some(rs) = state.radio_state.lock().unwrap().as_ref() {
        rs.flags.is_paused.store(true, Ordering::Release);
    }
}

/// Resume playback.
///
/// **Warm resume** (`is_hard_paused = false`): download task is still running,
/// buffer has buffered audio.  `sink.play()` suffices.
///
/// **Cold resume** (`is_hard_paused = true`): TCP was dropped.  A fresh 4 MB
/// ring buffer is created, its consumer is sent to `AudioStreamReader` (which
/// swaps it in on the next `read()`), and a new download task is spawned.
#[tauri::command]
pub async fn audio_resume(state: State<'_, AudioEngine>, app: AppHandle) -> Result<(), String> {
    // Detect radio hard-disconnect.
    let reconnect_info = {
        let guard = state.radio_state.lock().unwrap();
        guard
            .as_ref()
            .filter(|rs| rs.flags.is_hard_paused.load(Ordering::Acquire))
            .map(|rs| (rs.url.clone(), rs.gen, rs.flags.clone()))
    };

    if let Some((url, gen, flags)) = reconnect_info {
        let rb = HeapRb::<u8>::new(RADIO_BUF_CAPACITY);
        let (new_prod, new_cons) = rb.split();

        // Send new consumer to AudioStreamReader (non-blocking; unbounded channel).
        let ok = flags.new_cons_tx.lock().unwrap().send(new_cons).is_ok();

        if ok {
            let new_task = tokio::spawn(radio_download_task(
                gen,
                state.generation.clone(),
                None, // task performs its own fresh GET
                state.http_client.clone(),
                url,
                new_prod,
                flags.clone(),
                app,
            ));
            if let Some(rs) = state.radio_state.lock().unwrap().as_mut() {
                let old = std::mem::replace(&mut rs.task, new_task);
                old.abort(); // ensure any lingering old task is gone
                rs.flags.is_hard_paused.store(false, Ordering::Release);
                rs.flags.is_paused.store(false, Ordering::Release);
            }
        } else {
            eprintln!("[radio] resume: AudioStreamReader gone — skipping reconnect");
        }
    }

    // Resume the rodio Sink (works for both warm and cold resume).
    {
        let mut cur = state.current.lock().unwrap();
        if let Some(sink) = &cur.sink {
            if sink.is_paused() {
                let pos = cur.paused_at.unwrap_or(cur.seek_offset);
                sink.play();
                cur.seek_offset  = pos;
                cur.play_started = Some(Instant::now());
                cur.paused_at    = None;
            }
        }
    }
    if let Some(rs) = state.radio_state.lock().unwrap().as_ref() {
        rs.flags.is_paused.store(false, Ordering::Release);
    }
    Ok(())
}

#[tauri::command]
pub fn audio_stop(state: State<'_, AudioEngine>) {
    state.generation.fetch_add(1, Ordering::SeqCst);
    *state.chained_info.lock().unwrap() = None;
    // Drop RadioLiveState → triggers Drop → task.abort() → TCP released.
    drop(state.radio_state.lock().unwrap().take());
    let mut cur = state.current.lock().unwrap();
    if let Some(sink) = cur.sink.take() { sink.stop(); }
    cur.duration_secs = 0.0;
    cur.seek_offset   = 0.0;
    cur.play_started  = None;
    cur.paused_at     = None;
}

#[tauri::command]
pub fn audio_seek(seconds: f64, state: State<'_, AudioEngine>) -> Result<(), String> {
    // Ghost-command guard: reject seeks within 500 ms of a gapless auto-advance.
    {
        let switch_ms = state.gapless_switch_at.load(Ordering::SeqCst);
        if switch_ms > 0 {
            let now_ms = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_millis() as u64;
            if now_ms.saturating_sub(switch_ms) < 500 {
                return Ok(());
            }
        }
    }

    // Seeking back invalidates any pending gapless chain.
    let cur_pos = {
        let cur = state.current.lock().unwrap();
        cur.position()
    };
    if seconds < cur_pos - 1.0 {
        *state.chained_info.lock().unwrap() = None;
    }

    let mut cur = state.current.lock().unwrap();
    if cur.sink.is_none() { return Ok(()); }

    cur.sink.as_ref().unwrap()
        .try_seek(Duration::from_secs_f64(seconds.max(0.0)))
        .map_err(|e| e.to_string())?;

    if cur.paused_at.is_some() {
        cur.paused_at = Some(seconds);
    } else {
        cur.seek_offset = seconds;
        cur.play_started = Some(Instant::now());
    }
    Ok(())
}

#[tauri::command]
pub fn audio_set_volume(volume: f32, state: State<'_, AudioEngine>) {
    let mut cur = state.current.lock().unwrap();
    cur.base_volume = volume.clamp(0.0, 1.0);
    if let Some(sink) = &cur.sink {
        sink.set_volume((cur.base_volume * cur.replay_gain_linear * MASTER_HEADROOM).clamp(0.0, 1.0));
    }
}

#[tauri::command]
pub fn audio_update_replay_gain(
    volume: f32,
    replay_gain_db: Option<f32>,
    replay_gain_peak: Option<f32>,
    pre_gain_db: f32,
    fallback_db: f32,
    state: State<'_, AudioEngine>,
) {
    let (gain_linear, effective) = compute_gain(replay_gain_db, replay_gain_peak, pre_gain_db, fallback_db, volume);
    let mut cur = state.current.lock().unwrap();
    cur.replay_gain_linear = gain_linear;
    cur.base_volume = volume.clamp(0.0, 1.0);
    if let Some(sink) = &cur.sink {
        sink.set_volume(effective);
    }
}

/// Proxy: fetches https://autoeq.app/entries via Rust to bypass WebView CORS restrictions.
#[tauri::command]
pub async fn autoeq_entries(state: State<'_, AudioEngine>) -> Result<String, String> {
    state.http_client
        .get("https://autoeq.app/entries")
        .send().await.map_err(|e| e.to_string())?
        .text().await.map_err(|e| e.to_string())
}

/// Fetches the AutoEQ FixedBandEQ profile for a specific headphone from GitHub raw content.
///
/// Directory layout in the AutoEQ repo:
///   results/{source}/{form}/{name}/{name} FixedBandEQ.txt           (most sources)
///   results/{source}/{rig} {form}/{name}/{name} FixedBandEQ.txt     (crinacle — rig-prefixed dir)
///
/// We try the rig-prefixed path first (when rig is present), then fall back to form-only.
#[tauri::command]
pub async fn autoeq_fetch_profile(
    name: String,
    source: String,
    rig: Option<String>,
    form: String,
    state: State<'_, AudioEngine>,
) -> Result<String, String> {
    let base = "https://raw.githubusercontent.com/jaakkopasanen/AutoEq/master/results";
    let filename = format!("{} FixedBandEQ.txt", name);

    let candidates: Vec<String> = if let Some(ref r) = rig {
        vec![
            format!("{}/{}/{} {}/{}/{}", base, source, r, form, name, filename),
            format!("{}/{}/{}/{}/{}", base, source, form, name, filename),
        ]
    } else {
        vec![format!("{}/{}/{}/{}/{}", base, source, form, name, filename)]
    };

    for url in &candidates {
        let resp = state.http_client.get(url).send().await.map_err(|e| e.to_string())?;
        if resp.status().is_success() {
            return resp.text().await.map_err(|e| e.to_string());
        }
    }

    Err(format!("FixedBandEQ profile not found for '{}'", name))
}

#[tauri::command]
pub fn audio_set_eq(gains: [f32; 10], enabled: bool, pre_gain: f32, state: State<'_, AudioEngine>) {
    state.eq_enabled.store(enabled, Ordering::Relaxed);
    state.eq_pre_gain.store(pre_gain.clamp(-30.0, 6.0).to_bits(), Ordering::Relaxed);
    for (i, &gain) in gains.iter().enumerate() {
        state.eq_gains[i].store(gain.clamp(-12.0, 12.0).to_bits(), Ordering::Relaxed);
    }
}

#[tauri::command]
pub async fn audio_preload(
    url: String,
    duration_hint: f64,
    state: State<'_, AudioEngine>,
) -> Result<(), String> {
    {
        let preloaded = state.preloaded.lock().unwrap();
        if preloaded.as_ref().is_some_and(|p| same_playback_target(&p.url, &url)) {
            return Ok(());
        }
    }
    // Throttle: wait 8 s before starting the background download so it does not
    // compete with the decode + sink-feed work of the just-started current track.
    // If the user skips during the wait the generation counter changes and we abort.
    let gen_snapshot = state.generation.load(Ordering::Relaxed);
    tokio::time::sleep(Duration::from_secs(8)).await;
    if state.generation.load(Ordering::Relaxed) != gen_snapshot {
        return Ok(());
    }
    let data: Vec<u8> = if let Some(path) = url.strip_prefix("psysonic-local://") {
        tokio::fs::read(path).await.map_err(|e| e.to_string())?
    } else {
        let response = state.http_client.get(&url).send().await.map_err(|e| e.to_string())?;
        if !response.status().is_success() {
            return Ok(());
        }
        response.bytes().await.map_err(|e| e.to_string())?.into()
    };
    let _ = duration_hint; // kept in API for compatibility
    *state.preloaded.lock().unwrap() = Some(PreloadedTrack { url, data });
    Ok(())
}

/// Play a live internet radio stream.
///
/// Sends `Icy-MetaData: 1` to request inline ICY metadata.
/// Emits `audio:playing` with `duration = 0.0` (sentinel for live stream)
/// and `radio:metadata` whenever the StreamTitle changes.
#[tauri::command]
pub async fn audio_play_radio(
    url: String,
    volume: f32,
    app: AppHandle,
    state: State<'_, AudioEngine>,
) -> Result<(), String> {
    let gen = state.generation.fetch_add(1, Ordering::SeqCst) + 1;

    // Abort any previous radio task before stopping the sink.
    drop(state.radio_state.lock().unwrap().take());

    *state.chained_info.lock().unwrap() = None;
    {
        let mut cur = state.current.lock().unwrap();
        if let Some(old) = cur.sink.take() { old.stop(); }
    }
    if let Some(old) = state.fading_out_sink.lock().unwrap().take() { old.stop(); }

    // ── Open initial HTTP connection ──────────────────────────────────────────
    let response = state.http_client
        .get(&url)
        .header("Icy-MetaData", "1")
        .send()
        .await
        .map_err(|e| {
            let m = format!("radio: connection failed: {e}");
            app.emit("audio:error", &m).ok();
            m
        })?;

    if !response.status().is_success() {
        let m = format!("radio: HTTP {}", response.status());
        app.emit("audio:error", &m).ok();
        return Err(m);
    }

    let fmt_hint = content_type_to_hint(
        response.headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok())
            .unwrap_or(""),
    );

    // ── Build 4 MB lock-free SPSC ring buffer ─────────────────────────────────
    let rb = HeapRb::<u8>::new(RADIO_BUF_CAPACITY);
    let (prod, cons) = rb.split();

    let (new_cons_tx, new_cons_rx) = std::sync::mpsc::channel::<HeapConsumer<u8>>();
    let flags = Arc::new(RadioSharedFlags {
        is_paused:      AtomicBool::new(false),
        is_hard_paused: AtomicBool::new(false),
        new_cons_tx:    Mutex::new(new_cons_tx),
    });

    // ── Spawn download task ───────────────────────────────────────────────────
    let task = tokio::spawn(radio_download_task(
        gen,
        state.generation.clone(),
        Some(response),
        state.http_client.clone(),
        url.clone(),
        prod,
        flags.clone(),
        app.clone(),
    ));

    *state.radio_state.lock().unwrap() = Some(RadioLiveState {
        url:  url.clone(),
        gen,
        task,
        flags: flags.clone(),
    });

    // ── Build Symphonia decoder in a blocking thread ──────────────────────────
    let reader = AudioStreamReader {
        cons,
        new_cons_rx: Mutex::new(new_cons_rx),
        deadline: std::time::Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS),
        gen_arc:  state.generation.clone(),
        gen,
        pos: 0,
    };

    if state.generation.load(Ordering::SeqCst) != gen { return Ok(()); }

    let hint_clone = fmt_hint.clone();
    let decoder = tokio::task::spawn_blocking(move || {
        SizedDecoder::new_streaming(Box::new(reader), hint_clone.as_deref())
    })
    .await
    .map_err(|e| e.to_string())??;

    if state.generation.load(Ordering::SeqCst) != gen { return Ok(()); }

    let sample_rate     = decoder.sample_rate();
    let channels        = decoder.channels();
    let done_flag       = Arc::new(AtomicBool::new(false));
    let fadeout_trigger = Arc::new(AtomicBool::new(false));
    let fadeout_samples = Arc::new(AtomicU64::new(0));
    state.samples_played.store(0, Ordering::Relaxed);

    // Radio: no gapless trim, no ReplayGain, 5 ms fade-in to suppress click.
    let dyn_src   = DynSource::new(decoder.convert_samples::<f32>());
    let eq_src    = EqSource::new(dyn_src, state.eq_gains.clone(),
                                  state.eq_enabled.clone(), state.eq_pre_gain.clone());
    let fade_in   = EqualPowerFadeIn::new(eq_src, Duration::from_millis(5));
    let fade_out  = TriggeredFadeOut::new(fade_in, fadeout_trigger.clone(), fadeout_samples.clone());
    let notifying = NotifyingSource::new(fade_out, done_flag.clone());
    let counting  = CountingSource::new(notifying, state.samples_played.clone());

    if state.generation.load(Ordering::SeqCst) != gen { return Ok(()); }

    let sink = Sink::try_new(&*state.stream_handle.lock().unwrap()).map_err(|e| e.to_string())?;
    sink.set_volume((volume.clamp(0.0, 1.0) * MASTER_HEADROOM).clamp(0.0, 1.0));
    sink.append(counting);

    {
        let mut cur = state.current.lock().unwrap();
        if let Some(old) = cur.sink.take() { old.stop(); }
        cur.sink              = Some(sink);
        cur.duration_secs     = 0.0; // sentinel: live stream
        cur.seek_offset       = 0.0;
        cur.play_started      = Some(Instant::now());
        cur.paused_at         = None;
        cur.replay_gain_linear = 1.0;
        cur.base_volume       = volume.clamp(0.0, 1.0);
        cur.fadeout_trigger   = Some(fadeout_trigger);
        cur.fadeout_samples   = Some(fadeout_samples);
    }

    state.current_sample_rate.store(sample_rate, Ordering::Relaxed);
    state.current_channels.store(channels as u32, Ordering::Relaxed);

    app.emit("audio:playing", 0.0f64).ok();

    spawn_progress_task(
        gen,
        state.generation.clone(),
        state.current.clone(),
        state.chained_info.clone(),
        state.crossfade_enabled.clone(),
        state.crossfade_secs.clone(),
        done_flag,
        app,
        state.samples_played.clone(),
        state.current_sample_rate.clone(),
        state.current_channels.clone(),
        state.gapless_switch_at.clone(),
    );

    Ok(())
}

#[tauri::command]
pub fn audio_set_crossfade(enabled: bool, secs: f32, state: State<'_, AudioEngine>) {
    state.crossfade_enabled.store(enabled, Ordering::Relaxed);
    state.crossfade_secs.store(secs.clamp(0.1, 12.0).to_bits(), Ordering::Relaxed);
}

#[tauri::command]
pub fn audio_set_gapless(enabled: bool, state: State<'_, AudioEngine>) {
    state.gapless_enabled.store(enabled, Ordering::Relaxed);
}

// ─── Device-change watcher ────────────────────────────────────────────────────
//
// Polls the OS default output device every 3 s.  When it changes (Bluetooth
// headphones connecting, USB DAC plugging in, etc.) the stream is reopened on
// the new device and `audio:device-changed` is emitted so the frontend can
// restart playback.  The old Sink is dropped here — it was bound to the
// now-closed OutputStream and can no longer produce audio on any device.

pub fn start_device_watcher(engine: &AudioEngine, app: tauri::AppHandle) {
    let reopen_tx     = engine.stream_reopen_tx.clone();
    let stream_handle = engine.stream_handle.clone();
    let stream_rate   = engine.stream_sample_rate.clone();
    let current       = engine.current.clone();
    let fading_out    = engine.fading_out_sink.clone();

    tauri::async_runtime::spawn(async move {
        let mut last_name: Option<String> = tauri::async_runtime::spawn_blocking(|| {
            use rodio::cpal::traits::{DeviceTrait, HostTrait};
            rodio::cpal::default_host()
                .default_output_device()
                .and_then(|d| d.name().ok())
        }).await.unwrap_or(None);

        loop {
            tokio::time::sleep(Duration::from_secs(3)).await;

            let current_name: Option<String> = tauri::async_runtime::spawn_blocking(|| {
                use rodio::cpal::traits::{DeviceTrait, HostTrait};
                rodio::cpal::default_host()
                    .default_output_device()
                    .and_then(|d| d.name().ok())
            }).await.unwrap_or(None);

            if current_name == last_name {
                continue;
            }

            last_name = current_name.clone();

            // Only act if there is actually a device to open.
            let Some(_new_name) = current_name else { continue };

            // Debounce: give the OS time to finish configuring the new device.
            tokio::time::sleep(Duration::from_millis(500)).await;

            let rate = stream_rate.load(Ordering::Relaxed);
            let reopen_tx2 = reopen_tx.clone();
            let new_handle = tauri::async_runtime::spawn_blocking(move || {
                let (reply_tx, reply_rx) =
                    std::sync::mpsc::sync_channel::<rodio::OutputStreamHandle>(0);
                if reopen_tx2.send((rate, false, reply_tx)).is_err() {
                    return None; // audio thread exited
                }
                reply_rx.recv_timeout(Duration::from_secs(5)).ok()
            }).await.unwrap_or(None);

            let Some(handle) = new_handle else {
                eprintln!("[psysonic] device-watcher: stream reopen timed out");
                continue;
            };

            *stream_handle.lock().unwrap() = handle;

            // Drop the old Sink — it was bound to the now-closed OutputStream.
            if let Some(s) = current.lock().unwrap().sink.take() { s.stop(); }
            if let Some(s) = fading_out.lock().unwrap().take()   { s.stop(); }

            app.emit("audio:device-changed", ()).ok();
        }
    });
}