psysonic/src-tauri/src/audio/commands.rs

//! Tauri commands: audio_play / chain_preload / preload + the shared
//! spawn_progress_task helper. Transport (pause/resume/stop/seek), device,
//! radio, mix-mode and AutoEQ commands live in sibling modules.
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use futures_util::StreamExt;
use ringbuf::{HeapCons, HeapRb};
use ringbuf::traits::Split;
use rodio::Player;
use rodio::Source;
use symphonia::core::io::MediaSource;
use tauri::{AppHandle, Emitter, Manager, State};

use super::decode::{build_source, build_streaming_source, SizedDecoder};
use super::engine::{audio_http_client, AudioCurrent, AudioEngine};
use super::helpers::*;
use super::ipc::{maybe_emit_normalization_state, NormalizationStatePayload};
use super::preview::preview_clear_for_new_main_playback;
use super::state::{ChainedInfo, PreloadedTrack};
use super::stream::{
    ranged_download_task, track_download_task, AudioStreamReader,
    LocalFileSource, RangedHttpSource, RADIO_READ_TIMEOUT_SECS,
    TRACK_STREAM_MAX_BUF_CAPACITY, TRACK_STREAM_MIN_BUF_CAPACITY, LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES,
};

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

/// `analysis_track_id`: Subsonic `song.id` from the UI — ties waveform/loudness
/// cache to the track when playing `psysonic-local://` (hot/offline). Optional
/// for HTTP streams (`playback_identity` is used as fallback).
///
/// `stream_format_suffix`: Subsonic `song.suffix` (e.g. m4a); `stream.view` URLs have no
/// file extension, so this helps pick a Symphonia `format_hint` for ranged HTTP.
#[tauri::command]
pub async fn audio_play(
    url: String,
    volume: f32,
    duration_hint: f64,
    replay_gain_db: Option<f32>,
    replay_gain_peak: Option<f32>,
    loudness_gain_db: 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)
    analysis_track_id: Option<String>,
    stream_format_suffix: Option<String>,
    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(());
            }
        }
    }

    // Cancel any active preview before starting fresh main playback so the
    // two sinks don't end up mixed.
    preview_clear_for_new_main_playback(&state, &app);

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

    // Pin the logical playback URL immediately so `audio_update_replay_gain` (e.g. from
    // a fast `refreshLoudness` after `playTrack`) resolves LUFS for **this** track, not
    // the previous URL still stored until the sink swap completes.
    *state.current_playback_url.lock().unwrap() = Some(url.clone());
    let logical_trim = analysis_track_id
        .as_ref()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty());
    *state.current_analysis_track_id.lock().unwrap() = logical_trim.clone();
    let cache_id_for_tasks = analysis_cache_track_id(logical_trim.as_deref(), &url);

    // Extract format hint from URL for better symphonia probing. Strip the
    // query string first so Subsonic-style URLs (`stream.view?...&v=1.16.1&...`)
    // don't latch onto random query-param substrings; only accept short
    // alphanumeric tails that look like an actual audio extension.
    let format_hint = url
        .split('?').next()
        .and_then(|path| path.rsplit('.').next())
        .filter(|ext| {
            (1..=5).contains(&ext.len())
                && ext.chars().all(|c| c.is_ascii_alphanumeric())
                && matches!(
                    ext.to_ascii_lowercase().as_str(),
                    "mp3" | "flac" | "ogg" | "oga" | "opus" | "m4a" | "mp4"
                    | "aac" | "wav" | "wave" | "ape" | "wv" | "webm" | "mka"
                )
        })
        .map(|s| s.to_lowercase());

    enum PlayInput {
        Bytes(Vec<u8>),
        /// Seekable on-demand source — `RangedHttpSource` for HTTP streams,
        /// `LocalFileSource` for `psysonic-local://` files. Goes through
        /// `build_streaming_source` (no iTunSMPB scan, since we don't have the
        /// bytes in memory; chained-track gapless trim still applies via the
        /// re-played `Bytes` path on the next start).
        SeekableMedia {
            reader: Box<dyn MediaSource>,
            format_hint: Option<String>,
            tag: &'static str,
        },
        Streaming {
            reader: AudioStreamReader,
            format_hint: Option<String>,
        },
    }

    // Data source selection:
    // 1) Reused chained bytes (manual skip onto pre-chained track)
    // 2) `psysonic-local://` (offline / hot cache hit) → LocalFileSource (instant)
    // 3) Manual uncached remote start:
    //    a) Server supports Range + Content-Length → seekable RangedHttpSource
    //    b) Server does not → legacy non-seekable AudioStreamReader fallback
    // 4) Preloaded/streamed-cache hit → in-memory bytes via fetch_data
    let play_input = if let Some(d) = reuse_chained_bytes {
        spawn_analysis_seed_from_in_memory_bytes(
            &app,
            cache_id_for_tasks.as_deref(),
            gen,
            &state.generation,
            &d,
        );
        PlayInput::Bytes(d)
    } else {
        let stream_cache_hit = {
            let streamed = state.stream_completed_cache.lock().unwrap();
            streamed
                .as_ref()
                .is_some_and(|p| same_playback_target(&p.url, &url))
        };
        let preloaded_hit = {
            let preloaded = state.preloaded.lock().unwrap();
            preloaded
                .as_ref()
                .is_some_and(|p| same_playback_target(&p.url, &url))
        };
        let is_local = url.starts_with("psysonic-local://");

        if is_local && !stream_cache_hit && !preloaded_hit {
            let path = url.strip_prefix("psysonic-local://").unwrap();
            let file = std::fs::File::open(path).map_err(|e| e.to_string())?;
            let len = file.metadata().map(|m| m.len()).unwrap_or(0);
            let local_hint = std::path::Path::new(path)
                .extension()
                .and_then(|e| e.to_str())
                .map(|s| s.to_lowercase());
            crate::app_deprintln!(
                "[stream] LocalFileSource selected — size={} KB, hint={:?}",
                len / 1024,
                local_hint
            );
            if let Some(ref seed_id) = cache_id_for_tasks {
                let skip_cpu_seed = app
                    .try_state::<crate::analysis_cache::AnalysisCache>()
                    .map(|c| c.cpu_seed_redundant_for_track(seed_id).unwrap_or(false))
                    .unwrap_or(false);
                if !skip_cpu_seed {
                    let path_owned = std::path::PathBuf::from(path);
                    let app_seed = app.clone();
                    let gen_seed = gen;
                    let gen_arc_seed = state.generation.clone();
                    let seed_id = seed_id.clone();
                    tokio::spawn(async move {
                        if gen_arc_seed.load(Ordering::SeqCst) != gen_seed {
                            return;
                        }
                        let data = match tokio::fs::read(&path_owned).await {
                            Ok(d) => d,
                            Err(_) => return,
                        };
                        if gen_arc_seed.load(Ordering::SeqCst) != gen_seed {
                            return;
                        }
                        if data.is_empty() || data.len() > LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES {
                            crate::app_deprintln!(
                                "[stream] psysonic-local: skip analysis seed track_id={} bytes={} (over {} MiB cap)",
                                seed_id,
                                data.len(),
                                LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES / (1024 * 1024)
                            );
                            return;
                        }
                        crate::app_deprintln!(
                            "[stream] psysonic-local: file read complete track_id={} size_mib={:.2} — full-track analysis (cpu-seed queue)",
                            seed_id,
                            data.len() as f64 / (1024.0 * 1024.0)
                        );
                        let high = crate::audio::engine::analysis_seed_high_priority_for_track(&app_seed, &seed_id);
                        if let Err(e) =
                            crate::submit_analysis_cpu_seed(app_seed.clone(), seed_id.clone(), data, high).await
                        {
                            crate::app_eprintln!(
                                "[analysis] local-file seed failed for {}: {}",
                                seed_id,
                                e
                            );
                        }
                    });
                }
            }
            let reader = LocalFileSource { file, len };
            PlayInput::SeekableMedia {
                reader: Box::new(reader),
                format_hint: local_hint,
                tag: "local-file",
            }
        } else if !stream_cache_hit && !preloaded_hit && !is_local {
            // `manual` must NOT gate this branch: natural track end calls `next(false)`,
            // so auto-advance must still use ranged HTTP when available. Gating used to
            // force every auto-start through `fetch_data` (full RAM buffer + extra fetch log)
            // instead of `RangedHttpSource`.
            let response = audio_http_client(&state).get(&url).send().await.map_err(|e| e.to_string())?;
            if !response.status().is_success() {
                if state.generation.load(Ordering::SeqCst) != gen {
                    return Ok(()); // superseded
                }
                let status = response.status().as_u16();
                let msg = format!("HTTP {status}");
                app.emit("audio:error", &msg).ok();
                return Err(msg);
            }

            let mut stream_hint = content_type_to_hint(
                response
                    .headers()
                    .get(reqwest::header::CONTENT_TYPE)
                    .and_then(|v| v.to_str().ok())
                    .unwrap_or(""),
            )
            .or_else(|| {
                response
                    .headers()
                    .get(reqwest::header::CONTENT_DISPOSITION)
                    .and_then(|v| v.to_str().ok())
                    .and_then(|cd| format_hint_from_content_disposition(cd))
            })
            .or_else(|| normalize_stream_suffix_for_hint(stream_format_suffix.as_deref()))
            .or_else(|| format_hint.clone());

            let supports_range = response.headers()
                .get(reqwest::header::ACCEPT_RANGES)
                .and_then(|v| v.to_str().ok())
                .is_some_and(|v| v.to_ascii_lowercase().contains("bytes"));
            let total_size = response.content_length();

            if stream_hint.is_none() && supports_range {
                if let Some(total_u64) = total_size.filter(|&t| t > 0) {
                    let last = total_u64
                        .saturating_sub(1)
                        .min((STREAM_FORMAT_SNIFF_PROBE_BYTES - 1) as u64);
                    if let Ok(pr) = audio_http_client(&state)
                        .get(&url)
                        .header(reqwest::header::RANGE, format!("bytes=0-{last}"))
                        .send()
                        .await
                    {
                        let stat = pr.status();
                        let ok = stat == reqwest::StatusCode::PARTIAL_CONTENT
                            || stat == reqwest::StatusCode::OK;
                        if ok {
                            match pr.bytes().await {
                                Ok(bytes) if !bytes.is_empty() => {
                                    stream_hint = sniff_stream_format_extension(&bytes).or(stream_hint);
                                    if stream_hint.is_some() {
                                        crate::app_deprintln!(
                                            "[stream] ranged: format sniff from {} B prefix → hint={:?}",
                                            bytes.len(),
                                            stream_hint
                                        );
                                    }
                                }
                                _ => {}
                            }
                        }
                    }
                }
            }

            // Guardrail: when format/container hint is unknown, some demuxers may
            // seek near EOF during probe. With a progressively downloaded ranged
            // source that can delay first audible samples until most/all bytes are
            // fetched. Prefer sequential streaming in that case for faster start.
            if let (true, Some(total), true) = (supports_range, total_size, stream_hint.is_some()) {
                let total_usize = total as usize;
                crate::app_deprintln!(
                    "[stream] RangedHttpSource selected — total={} KB, hint={:?}",
                    total_usize / 1024,
                    stream_hint
                );
                let buf = Arc::new(Mutex::new(vec![0u8; total_usize]));
                let downloaded_to = Arc::new(AtomicUsize::new(0));
                let done = Arc::new(AtomicBool::new(false));
                let loudness_hold_for_defer = (total_usize <= crate::audio::stream::TRACK_STREAM_PROMOTE_MAX_BYTES)
                    .then_some(state.ranged_loudness_seed_hold.clone());
                tokio::spawn(ranged_download_task(
                    gen,
                    state.generation.clone(),
                    audio_http_client(&state),
                    app.clone(),
                    duration_hint,
                    url.clone(),
                    response,
                    buf.clone(),
                    downloaded_to.clone(),
                    done.clone(),
                    state.stream_completed_cache.clone(),
                    state.normalization_engine.clone(),
                    state.normalization_target_lufs.clone(),
                    state.loudness_pre_analysis_attenuation_db.clone(),
                    cache_id_for_tasks.clone(),
                    loudness_hold_for_defer,
                ));
                let reader = RangedHttpSource {
                    buf,
                    downloaded_to,
                    total_size: total,
                    pos: 0,
                    done,
                    gen_arc: state.generation.clone(),
                    gen,
                };
                PlayInput::SeekableMedia {
                    reader: Box::new(reader),
                    format_hint: stream_hint,
                    tag: "ranged-stream",
                }
            } else {
                crate::app_deprintln!(
                    "[stream] legacy AudioStreamReader (non-seekable) — accept-ranges={}, content-length={:?}, hint={:?}",
                    supports_range, total_size, stream_hint
                );
                let buffer_cap = total_size
                    .map(|n| n as usize)
                    .unwrap_or(TRACK_STREAM_MIN_BUF_CAPACITY)
                    .clamp(TRACK_STREAM_MIN_BUF_CAPACITY, TRACK_STREAM_MAX_BUF_CAPACITY);
                let rb = HeapRb::<u8>::new(buffer_cap);
                let (prod, cons) = rb.split();
                let done = Arc::new(AtomicBool::new(false));
                tokio::spawn(track_download_task(
                    gen,
                    state.generation.clone(),
                    audio_http_client(&state),
                    app.clone(),
                    url.clone(),
                    response,
                    prod,
                    done.clone(),
                    state.stream_completed_cache.clone(),
                    state.normalization_engine.clone(),
                    state.normalization_target_lufs.clone(),
                    state.loudness_pre_analysis_attenuation_db.clone(),
                    cache_id_for_tasks.clone(),
                ));

                let (_new_cons_tx, new_cons_rx) = std::sync::mpsc::channel::<HeapCons<u8>>();
                let reader = AudioStreamReader {
                    cons: Mutex::new(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,
                    source_tag: "track-stream",
                    eof_when_empty: Some(done),
                    pos: 0,
                };
                PlayInput::Streaming {
                    reader,
                    format_hint: stream_hint,
                }
            }
        } else {
            let data = fetch_data(&url, &state, gen, &app).await?;
            let data = match data {
                Some(d) => d,
                None => return Ok(()), // superseded while downloading
            };
            spawn_analysis_seed_from_in_memory_bytes(
                &app,
                cache_id_for_tasks.as_deref(),
                gen,
                &state.generation,
                &data,
            );
            PlayInput::Bytes(data)
        }
    };

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

    let target_lufs = f32::from_bits(state.normalization_target_lufs.load(Ordering::Relaxed));
    let cache_loudness = resolve_loudness_gain_from_cache(
        &app,
        &url,
        target_lufs,
        logical_trim.as_deref(),
    );
    let resolved_loudness_gain_db = cache_loudness;
    let norm_mode = state.normalization_engine.load(Ordering::Relaxed);
    let pre_analysis_db = loudness_pre_analysis_db_for_engine(&state);
    let startup_loudness_gain_db = if norm_mode == 2 {
        loudness_gain_db_after_resolve(
            cache_loudness,
            target_lufs,
            pre_analysis_db,
            false,
            loudness_gain_db,
        )
    } else {
        cache_loudness
    };
    let (gain_linear, effective_volume) = compute_gain(
        norm_mode,
        replay_gain_db,
        replay_gain_peak,
        startup_loudness_gain_db,
        pre_gain_db,
        fallback_db,
        volume,
    );
    let current_gain_db = loudness_ui_current_gain_db(gain_linear);
    crate::app_deprintln!(
        "[normalization] audio_play track_id={:?} engine={} replay_gain_db={:?} replay_gain_peak={:?} loudness_gain_db={:?} gain_linear={:.4} current_gain_db={:?} target_lufs={:.2} volume={:.3} effective_volume={:.3}",
        playback_identity(&url),
        normalization_engine_name(norm_mode),
        replay_gain_db,
        replay_gain_peak,
        resolved_loudness_gain_db,
        gain_linear,
        current_gain_db,
        target_lufs,
        volume,
        effective_volume
    );
    maybe_emit_normalization_state(
        &app,
        NormalizationStatePayload {
            engine: normalization_engine_name(norm_mode).to_string(),
            current_gain_db,
            target_lufs,
        },
    );

    // 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;
    let mut new_is_seekable = true;
    let built = match play_input {
        PlayInput::Bytes(data) => 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,
        ),
        PlayInput::SeekableMedia { reader, format_hint, tag } => {
            let decoder = tokio::task::spawn_blocking(move || {
                SizedDecoder::new_streaming(reader, format_hint.as_deref(), tag)
            })
            .await
            .map_err(|e| e.to_string())??;

            build_streaming_source(
                decoder,
                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,
            )
        }
        PlayInput::Streaming { reader, format_hint } => {
            new_is_seekable = false;
            let decoder = tokio::task::spawn_blocking(move || {
                SizedDecoder::new_streaming(Box::new(reader), format_hint.as_deref(), "track-stream")
            })
            .await
            .map_err(|e| e.to_string())??;

            build_streaming_source(
                decoder,
                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,
            )
        }
    }.map_err(|e| { app.emit("audio:error", &e).ok(); e })?;
    state.current_is_seekable.store(new_is_seekable, Ordering::SeqCst);
    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::<Arc<rodio::MixerDeviceSink>>(0);
            let dev = state.selected_device.lock().unwrap().clone();
            if state.stream_reopen_tx.send((target_rate, hi_res_enabled, dev, 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(_) => {
                        crate::app_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 = Arc::new(Player::connect_new(state.stream_handle.lock().unwrap().mixer()));
    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.
    //
    // Use a frame-aligned `SamplesBuffer` rather than `Zero + take_duration` —
    // the latter computes its sample count via integer-nanosecond division
    // (1_000_000_000 / (sr * ch)), which at common rates leaks an odd number
    // of samples (e.g. 44103 at 44.1 kHz / 2 ch / 500 ms = 22051.5 frames).
    // The half-frame leak shifts the next source's L/R parity in the device
    // stream and can manifest as a dead channel for the rest of the track
    // (reported by users for natural-end-without-gapless transitions only).
    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 ch = source.channels();
            let sr = source.sample_rate();
            // 500 ms in whole frames, then expand to interleaved samples.
            let frames = (sr.get() / 2) as usize;
            let total_samples = frames.saturating_mul(ch.get() as usize);
            let silence = rodio::buffer::SamplesBuffer::new(ch, sr, vec![0f32; total_samples]);
            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(),
        state.current_playback_url.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>,
    loudness_gain_db: Option<f32>,
    pre_gain_db: f32,
    fallback_db: f32,
    hi_res_enabled: bool,
    analysis_track_id: Option<String>,
    app: AppHandle,
    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 = audio_http_client(&state).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 logical_trim = analysis_track_id
        .as_ref()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty());

    // Only `gain_linear` is needed — `effective_volume` is intentionally NOT
    // applied to the Sink here. `audio_chain_preload` runs ~30 s before the
    // current track ends, and `Sink::set_volume` affects the WHOLE Sink (incl.
    // the still-playing current source). Volume for the chained track is
    // applied at the gapless transition in `spawn_progress_task`, not here.
    let target_lufs = f32::from_bits(state.normalization_target_lufs.load(Ordering::Relaxed));
    let norm_mode = state.normalization_engine.load(Ordering::Relaxed);
    let pre_analysis_db = loudness_pre_analysis_db_for_engine(&state);
    let chain_loudness_db = if norm_mode == 2 {
        loudness_gain_db_or_startup(
            &app,
            &url,
            target_lufs,
            logical_trim.as_deref(),
            pre_analysis_db,
            false,
            loudness_gain_db,
        )
    } else {
        resolve_loudness_gain_from_cache(&app, &url, target_lufs, logical_trim.as_deref())
    };
    let (gain_linear, _effective_volume) = compute_gain(
        norm_mode,
        replay_gain_db,
        replay_gain_peak,
        chain_loudness_db,
        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 {
        crate::app_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.
    // Note: `set_volume` is deliberately NOT called here (see comment above).
    {
        let cur = state.current.lock().unwrap();
        match &cur.sink {
            Some(sink) => {
                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
pub(super) 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>,
    current_playback_url: Arc<Mutex<Option<String>>>,
) {
    // Keep progress aligned with audible output (ALSA/PipeWire/Pulse queue) on
    // Linux; mirrors the quantum policy used for stream open/reopen plus a small
    // scheduler/mixer cushion so the UI doesn't run ahead. Other platforms have
    // their own latency reporting paths and don't need the compensation here.
    #[cfg(target_os = "linux")]
    fn estimated_output_latency_secs(sample_rate_hz: f64) -> f64 {
        let rate = sample_rate_hz.max(1.0);
        let frames = if rate > 48_000.0 { 8192.0 } else { 4096.0 };
        (frames / rate) + 0.012
    }
    #[cfg(not(target_os = "linux"))]
    fn estimated_output_latency_secs(_sample_rate_hz: f64) -> f64 {
        0.0
    }

    // Keep near-end detection at 100 ms, but throttle progress IPC to webview.
    const PROGRESS_EMIT_MIN_MS: u64 = 1500;
    const PROGRESS_EMIT_MIN_DELTA_SECS: f64 = 0.9;

    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;
        let mut last_progress_emit_at = Instant::now() - Duration::from_millis(PROGRESS_EMIT_MIN_MS);
        let mut last_progress_emit_pos = -1.0f64;
        let mut last_progress_emit_paused = false;

        loop {
            // 100 ms tick keeps near-end detection timely for crossfade/gapless
            // handoff while frontend still interpolates smoothly via rAF.
            tokio::time::sleep(Duration::from_millis(100)).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 {
                    crate::app_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 and apply the chained track's
                    // effective volume. Deferred from `audio_chain_preload`
                    // (which runs ~30 s before the current track ends) to
                    // avoid changing loudness of the still-playing current
                    // track. `Sink::set_volume` affects the whole Sink, so it
                    // must only be called at the boundary, not at preload.
                    {
                        let mut cur = current_arc.lock().unwrap();
                        let prev_effective = (cur.base_volume * cur.replay_gain_linear * MASTER_HEADROOM).clamp(0.0, 1.0);
                        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());
                        if let Some(sink) = &cur.sink {
                            let effective = (cur.base_volume * cur.replay_gain_linear * MASTER_HEADROOM).clamp(0.0, 1.0);
                            ramp_sink_volume(Arc::clone(sink), prev_effective, effective);
                        }
                    }

                    *current_playback_url.lock().unwrap() = Some(info.url.clone());

                    // 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);

            // Read playback snapshot under a single lock to minimize contention
            // with seek/play/pause commands that also touch `current`.
            let (dur, paused_at) = {
                let cur = current_arc.lock().unwrap();
                (cur.duration_secs, cur.paused_at)
            };
            let is_paused = paused_at.is_some();

            let pos_raw = if let Some(p) = paused_at {
                p
            } else {
                (samples / divisor).min(dur.max(0.001))
            };
            let progress_latency = if is_paused {
                0.0
            } else {
                estimated_output_latency_secs(rate)
            };
            let pos = (pos_raw - progress_latency).max(0.0);

            let now = Instant::now();
            let should_emit_progress = if is_paused != last_progress_emit_paused {
                true
            } else if now.duration_since(last_progress_emit_at) >= Duration::from_millis(PROGRESS_EMIT_MIN_MS) {
                true
            } else {
                (pos - last_progress_emit_pos).abs() >= PROGRESS_EMIT_MIN_DELTA_SECS
            };
            if should_emit_progress {
                app.emit("audio:progress", ProgressPayload { current_time: pos, duration: dur }).ok();
                last_progress_emit_at = now;
                last_progress_emit_pos = pos;
                last_progress_emit_paused = is_paused;
            }

            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_raw >= 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 async fn audio_preload(
    url: String,
    duration_hint: f64,
    analysis_track_id: Option<String>,
    app: AppHandle,
    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)) {
            let _ = app.emit("audio:preload-ready", url.clone());
            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 = audio_http_client(&state).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
    let logical_trim = analysis_track_id
        .as_ref()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty());
    if let Some(track_id) = analysis_cache_track_id(logical_trim.as_deref(), &url) {
        crate::app_deprintln!(
            "[stream] audio_preload: bytes ready track_id={} size_mib={:.2} — invoking full-track analysis",
            track_id,
            data.len() as f64 / (1024.0 * 1024.0)
        );
        let high = crate::audio::engine::analysis_track_id_is_current_playback(&state, &track_id);
        if let Err(e) = crate::submit_analysis_cpu_seed(app.clone(), track_id.clone(), data.clone(), high).await {
            crate::app_eprintln!("[analysis] preload seed failed for {}: {}", track_id, e);
        }
    }
    let url_for_emit = url.clone();
    *state.preloaded.lock().unwrap() = Some(PreloadedTrack { url, data });
    let _ = app.emit("audio:preload-ready", url_for_emit);
    Ok(())
}