diff --git a/src-tauri/src/audio.rs b/src-tauri/src/audio.rs deleted file mode 100644 index 37ebb53c..00000000 --- a/src-tauri/src/audio.rs +++ /dev/null @@ -1,5446 +0,0 @@ -use std::io::{Cursor, Read, Seek, SeekFrom}; -use std::sync::{Arc, Mutex, OnceLock, RwLock, TryLockError}; -use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering}; -use std::time::{Duration, Instant}; -#[cfg(unix)] -use libc; - -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::{CodecRegistry, 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, Manager, State}; -#[derive(Clone, Serialize)] -#[serde(rename_all = "camelCase")] -struct PartialLoudnessPayload { - track_id: Option, - gain_db: f32, - target_lufs: f32, - is_partial: bool, -} - -#[derive(Clone, Serialize)] -#[serde(rename_all = "camelCase")] -struct NormalizationStatePayload { - engine: String, - current_gain_db: Option, - target_lufs: f32, -} - -/// Last `audio:normalization-state` emit, kept so we can suppress duplicate -/// payloads. The frontend already debounces this event, but on Windows -/// (WebView2) the IPC pipe is the bottleneck — every echo we skip here is -/// renderer-thread time we don't pay. -static LAST_NORM_STATE_EMIT: OnceLock>> = OnceLock::new(); - -fn norm_state_lock() -> &'static Mutex> { - LAST_NORM_STATE_EMIT.get_or_init(|| Mutex::new(None)) -} - -fn norm_state_changed(prev: &NormalizationStatePayload, next: &NormalizationStatePayload) -> bool { - if prev.engine != next.engine { return true; } - if (prev.target_lufs - next.target_lufs).abs() >= 0.02 { return true; } - match (prev.current_gain_db, next.current_gain_db) { - (None, None) => false, - (Some(a), Some(b)) => (a - b).abs() >= 0.05, - _ => true, // None ↔ Some transition is significant - } -} - -fn maybe_emit_normalization_state(app: &AppHandle, payload: NormalizationStatePayload) { - let mut guard = norm_state_lock().lock().unwrap(); - let should_emit = match guard.as_ref() { - Some(prev) => norm_state_changed(prev, &payload), - None => true, - }; - if !should_emit { return; } - *guard = Some(payload.clone()); - drop(guard); - let _ = app.emit("audio:normalization-state", payload); -} - -/// Last `analysis:loudness-partial` gain emitted per track-identity, used to -/// suppress emits whose gain hasn't moved meaningfully (≥ 0.1 dB). The partial -/// heuristic in `emit_partial_loudness_from_bytes` and the ranged-progress curve -/// both produce values that drift by hundredths of a dB even on identical input, -/// so the time-based throttle alone is not enough to keep the loop quiet. -static LAST_PARTIAL_LOUDNESS_EMIT: OnceLock>> = OnceLock::new(); -const PARTIAL_LOUDNESS_DELTA_THRESHOLD_DB: f32 = 0.1; - -fn partial_loudness_should_emit(track_key: &str, gain_db: f32) -> bool { - let mut guard = LAST_PARTIAL_LOUDNESS_EMIT - .get_or_init(|| Mutex::new(std::collections::HashMap::new())) - .lock() - .unwrap(); - let prev = guard.get(track_key).copied(); - if let Some(p) = prev { - if (p - gain_db).abs() < PARTIAL_LOUDNESS_DELTA_THRESHOLD_DB { - return false; - } - } - guard.insert(track_key.to_string(), gain_db); - true -} - - -// ─── 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> { - inner: S, - sample_rate: u32, - channels: u16, - gains: Arc<[AtomicU32; 10]>, - enabled: Arc, - pre_gain: Arc, - filters: [[DirectForm2Transposed; 2]; 10], - current_gains: [f32; 10], - sample_counter: usize, - channel_idx: usize, -} - -impl> EqSource { - fn new(inner: S, gains: Arc<[AtomicU32; 10]>, enabled: Arc, pre_gain: Arc) -> 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::::from_params( - FilterType::PeakingEQ(0.0), - (sample_rate as f32).hz(), - freq.hz(), - EQ_Q, - ).unwrap_or_else(|_| Coefficients::::from_params( - FilterType::PeakingEQ(0.0), - (sample_rate as f32).hz(), - 1000.0f32.hz(), - EQ_Q, - ).unwrap()); - DirectForm2Transposed::::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::::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> Iterator for EqSource { - type Item = f32; - - fn next(&mut self) -> Option { - 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> Source for EqSource { - fn current_frame_len(&self) -> Option { self.inner.current_frame_len() } - fn channels(&self) -> u16 { self.channels } - fn sample_rate(&self) -> u32 { self.sample_rate } - fn total_duration(&self) -> Option { 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::::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::::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>. - -struct DynSource { - inner: Box + Send>, - channels: u16, - sample_rate: u32, -} - -impl DynSource { - fn new(src: impl Source + 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 { self.inner.next() } -} - -impl Source for DynSource { - fn current_frame_len(&self) -> Option { self.inner.current_frame_len() } - fn channels(&self) -> u16 { self.channels } - fn sample_rate(&self) -> u32 { self.sample_rate } - fn total_duration(&self) -> Option { 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> { - inner: S, - sample_count: u64, - fade_samples: u64, -} - -impl> EqualPowerFadeIn { - 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> Iterator for EqualPowerFadeIn { - type Item = f32; - fn next(&mut self) -> Option { - 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> Source for EqualPowerFadeIn { - fn current_frame_len(&self) -> Option { 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 { 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> { - inner: S, - trigger: Arc, - fade_total_samples: Arc, - fade_progress: u64, - fading: bool, - cached_total: u64, -} - -impl> TriggeredFadeOut { - fn new(inner: S, trigger: Arc, fade_total_samples: Arc) -> Self { - Self { - inner, - trigger, - fade_total_samples, - fade_progress: 0, - fading: false, - cached_total: 0, - } - } -} - -impl> Iterator for TriggeredFadeOut { - type Item = f32; - fn next(&mut self) -> Option { - // 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> Source for TriggeredFadeOut { - fn current_frame_len(&self) -> Option { 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 { 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> { - inner: S, - done: Arc, - signalled: bool, -} - -impl> NotifyingSource { - fn new(inner: S, done: Arc) -> Self { - Self { inner, done, signalled: false } - } -} - -impl> Iterator for NotifyingSource { - type Item = f32; - fn next(&mut self) -> Option { - let sample = self.inner.next(); - if sample.is_none() && !self.signalled { - self.signalled = true; - self.done.store(true, Ordering::SeqCst); - } - sample - } -} - -impl> Source for NotifyingSource { - fn current_frame_len(&self) -> Option { 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 { 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> { - inner: S, - counter: Arc, -} - -impl> CountingSource { - fn new(inner: S, counter: Arc) -> Self { - Self { inner, counter } - } -} - -impl> Iterator for CountingSource { - type Item = f32; - fn next(&mut self) -> Option { - let sample = self.inner.next(); - if sample.is_some() { - self.counter.fetch_add(1, Ordering::Relaxed); - } - sample - } -} - -impl> Source for CountingSource { - fn current_frame_len(&self) -> Option { 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 { 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 -// │ (4 MB HeapRb, lock-free) -// HeapConsumer -// │ -// 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; -/// Minimum ring buffer for on-demand track streaming starts. -const TRACK_STREAM_MIN_BUF_CAPACITY: usize = 1024 * 1024; -/// Cap ring buffer growth when content-length is known. -const TRACK_STREAM_MAX_BUF_CAPACITY: usize = 32 * 1024 * 1024; -/// Max bytes kept in memory to promote a completed streamed track for fast replay/seek recovery. -const TRACK_STREAM_PROMOTE_MAX_BYTES: usize = 64 * 1024 * 1024; -/// Hot/offline `psysonic-local://` files are read from disk for waveform/LUFS seeding — not the -/// same heap pressure as retaining a full HTTP capture. FLAC/DSD tracks often exceed 64 MiB; -/// using the stream-promote cap here skipped analysis entirely (empty seekbar). -const LOCAL_FILE_PLAYBACK_SEED_MAX_BYTES: usize = 512 * 1024 * 1024; -/// Consecutive body-stream failures tolerated for track streaming before abort. -const TRACK_STREAM_MAX_RECONNECTS: u32 = 3; -/// 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 }, -} - -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) -> Option { - let mut extracted: Option = 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 { - 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 (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, - /// 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>>, - deadline: std::time::Instant, - gen_arc: Arc, - gen: u64, - /// Diagnostic tag for logs ("radio" or "track-stream"). - source_tag: &'static str, - /// Optional completion marker: when true and the ring buffer is empty, - /// return EOF immediately (used by one-shot track streaming). - eof_when_empty: Option>, - /// 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 { - // 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> = 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 self - .eof_when_empty - .as_ref() - .is_some_and(|done| done.load(Ordering::SeqCst)) - { - return Ok(0); - } - if std::time::Instant::now() >= self.deadline { - crate::app_eprintln!( - "[{}] AudioStreamReader: {}s without data → EOF", - self.source_tag, - RADIO_READ_TIMEOUT_SECS - ); - return Err(std::io::Error::new( - std::io::ErrorKind::TimedOut, - format!("{}: no data received", self.source_tag), - )); - } - std::thread::sleep(Duration::from_millis(RADIO_YIELD_MS)); - } - } -} - -impl Seek for AudioStreamReader { - fn seek(&mut self, pos: SeekFrom) -> std::io::Result { - match pos { - SeekFrom::Current(0) => Ok(self.pos), - _ => Err(std::io::Error::new( - std::io::ErrorKind::Unsupported, - format!("{} stream is not seekable", self.source_tag), - )), - } - } -} - -impl MediaSource for AudioStreamReader { - fn is_seekable(&self) -> bool { false } - fn byte_len(&self) -> Option { None } -} - -// ── RangedHttpSource — seekable HTTP-backed MediaSource ────────────────────── -// -// Pre-allocates a Vec of total track size. A background task fills it -// linearly from offset 0 via streaming HTTP. Read blocks (with timeout) until -// requested bytes are downloaded; Seek only updates the cursor. -// -// Reports is_seekable=true so Symphonia performs time-based seeks via the -// format reader. Backward seeks: instant (data in buffer). Forward seeks -// beyond downloaded_to: Read blocks until the linear download catches up. -// -// Requires server to have responded with both Content-Length and -// `Accept-Ranges: bytes` so reconnects can resume via HTTP Range. - -struct RangedHttpSource { - /// Pre-allocated buffer of total size. Filled linearly from offset 0. - buf: Arc>>, - /// Bytes contiguously downloaded from offset 0. - downloaded_to: Arc, - total_size: u64, - pos: u64, - /// Set when the download task terminates (success or hard error). - done: Arc, - gen_arc: Arc, - gen: u64, -} - -impl Read for RangedHttpSource { - fn read(&mut self, buf: &mut [u8]) -> std::io::Result { - if self.gen_arc.load(Ordering::SeqCst) != self.gen { - return Ok(0); - } - if self.pos >= self.total_size { - return Ok(0); - } - let max_read = ((self.total_size - self.pos) as usize).min(buf.len()); - if max_read == 0 { - return Ok(0); - } - let target_end = self.pos + max_read as u64; - - let deadline = Instant::now() + Duration::from_secs(RADIO_READ_TIMEOUT_SECS); - loop { - if self.gen_arc.load(Ordering::SeqCst) != self.gen { - return Ok(0); - } - let dl = self.downloaded_to.load(Ordering::SeqCst) as u64; - if dl >= target_end { - break; - } - // Download finished but our cursor is past downloaded_to (e.g. seek - // beyond a partial download that aborted). Return what we have. - if self.done.load(Ordering::SeqCst) { - if dl > self.pos { - let avail = (dl - self.pos) as usize; - let src = self.buf.lock().unwrap(); - let start = self.pos as usize; - buf[..avail].copy_from_slice(&src[start..start + avail]); - drop(src); - self.pos += avail as u64; - return Ok(avail); - } - return Ok(0); - } - if Instant::now() >= deadline { - return Err(std::io::Error::new( - std::io::ErrorKind::TimedOut, - "ranged-http: no data within timeout", - )); - } - std::thread::sleep(Duration::from_millis(RADIO_YIELD_MS)); - } - - let src = self.buf.lock().unwrap(); - let start = self.pos as usize; - let end = start + max_read; - buf[..max_read].copy_from_slice(&src[start..end]); - drop(src); - self.pos += max_read as u64; - Ok(max_read) - } -} - -impl Seek for RangedHttpSource { - fn seek(&mut self, pos: SeekFrom) -> std::io::Result { - let new_pos: i64 = match pos { - SeekFrom::Start(p) => p as i64, - SeekFrom::Current(p) => self.pos as i64 + p, - SeekFrom::End(p) => self.total_size as i64 + p, - }; - if new_pos < 0 { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "ranged-http: seek before start", - )); - } - self.pos = (new_pos as u64).min(self.total_size); - Ok(self.pos) - } -} - -impl MediaSource for RangedHttpSource { - fn is_seekable(&self) -> bool { true } - fn byte_len(&self) -> Option { Some(self.total_size) } -} - -// ── LocalFileSource — seekable file-backed MediaSource ─────────────────────── -// -// Wraps `std::fs::File` so the decoder reads on-demand from disk instead of -// pre-loading the whole file into a Vec. Used for `psysonic-local://` URLs -// (offline library + hot playback cache hits) — gives instant track-start -// because Symphonia only needs to read ~64 KB during probe before playback -// can begin, vs the previous behaviour of `tokio::fs::read` which blocked -// until the entire file (often 100+ MB for hi-res FLAC) was in RAM. - -struct LocalFileSource { - file: std::fs::File, - len: u64, -} - -impl Read for LocalFileSource { - fn read(&mut self, buf: &mut [u8]) -> std::io::Result { - self.file.read(buf) - } -} - -impl Seek for LocalFileSource { - fn seek(&mut self, pos: SeekFrom) -> std::io::Result { - self.file.seek(pos) - } -} - -impl MediaSource for LocalFileSource { - fn is_seekable(&self) -> bool { true } - fn byte_len(&self) -> Option { Some(self.len) } -} - -// ── 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 to AudioStreamReader on reconnect. - new_cons_tx: Mutex>>, -} - -/// 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, -} - -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"] - -/// Symphonia’s default codec set for our enabled features, plus Opus via libopus. -fn psysonic_codec_registry() -> &'static CodecRegistry { - static REGISTRY: OnceLock = OnceLock::new(); - REGISTRY.get_or_init(|| { - let mut registry = CodecRegistry::new(); - symphonia::default::register_enabled_codecs(&mut registry); - registry.register_all::(); - registry - }) -} - -fn try_make_radio_decoder( - params: &symphonia::core::codecs::CodecParameters, - opts: &DecoderOptions, -) -> Result, symphonia::core::errors::Error> { - psysonic_codec_registry().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, - mut initial_response: Option, - http_client: reqwest::Client, - url: String, - mut prod: HeapProducer, - flags: Arc, - 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 = 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 { - crate::app_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() => { - crate::app_eprintln!("[radio] reconnected ({bytes_total} B so far)"); - r - } - Ok(r) => { - crate::app_eprintln!("[radio] reconnect: HTTP {} — giving up", r.status()); - break 'outer; - } - Err(e) => { - crate::app_eprintln!("[radio] reconnect error: {e} — giving up"); - break 'outer; - } - } - } - }; - - // Parse ICY metaint from each response (consistent across reconnects). - let metaint: Option = 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 = 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; - crate::app_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 { "" }; - crate::app_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; - crate::app_eprintln!("[radio] stream error: {e} → reconnecting (consecutive #{reconnect_count})"); - break 'inner; - } - None => { - reconnect_count += 1; - crate::app_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 - - crate::app_eprintln!("[radio] download task done ({bytes_total} B total)"); -} - -/// One-shot HTTP downloader for track streaming starts. -/// -/// Pushes response chunks into an SPSC ring buffer consumed by `AudioStreamReader`. -/// Terminates when: -/// - generation changes (track superseded), -/// - response stream ends, or -/// - response emits an error. -async fn track_download_task( - gen: u64, - gen_arc: Arc, - http_client: reqwest::Client, - app: AppHandle, - url: String, - initial_response: reqwest::Response, - mut prod: HeapProducer, - done: Arc, - promote_cache_slot: Arc>>, - normalization_engine: Arc, - normalization_target_lufs: Arc, - loudness_pre_analysis_attenuation_db: Arc, - cache_track_id: Option, -) { - let mut downloaded: u64 = 0; - let mut reconnects: u32 = 0; - let mut next_response: Option = Some(initial_response); - let mut capture: Vec = Vec::new(); - let mut capture_over_limit = false; - let mut last_partial_loudness_emit = Instant::now() - Duration::from_secs(5); - 'outer: loop { - let response = if let Some(r) = next_response.take() { - r - } else { - let mut req = http_client.get(&url); - if downloaded > 0 { - req = req.header(reqwest::header::RANGE, format!("bytes={downloaded}-")); - } - match req.send().await { - Ok(r) => r, - Err(err) => { - if reconnects >= TRACK_STREAM_MAX_RECONNECTS { - crate::app_eprintln!( - "[audio] streaming reconnect failed after {} attempts: {}", - reconnects, err - ); - done.store(true, Ordering::SeqCst); - return; - } - reconnects += 1; - tokio::time::sleep(Duration::from_millis(200)).await; - continue 'outer; - } - } - }; - if downloaded > 0 && response.status() != reqwest::StatusCode::PARTIAL_CONTENT { - crate::app_eprintln!( - "[audio] streaming reconnect returned {}, expected 206 for range resume", - response.status() - ); - done.store(true, Ordering::SeqCst); - return; - } - if downloaded == 0 && !response.status().is_success() { - crate::app_eprintln!("[audio] streaming HTTP {}", response.status()); - done.store(true, Ordering::SeqCst); - return; - } - - let mut byte_stream = response.bytes_stream(); - while let Some(chunk) = byte_stream.next().await { - if gen_arc.load(Ordering::SeqCst) != gen { - done.store(true, Ordering::SeqCst); - return; - } - let chunk = match chunk { - Ok(c) => c, - Err(e) => { - if reconnects >= TRACK_STREAM_MAX_RECONNECTS { - crate::app_eprintln!( - "[audio] streaming download error after {} reconnects: {}", - reconnects, e - ); - done.store(true, Ordering::SeqCst); - return; - } - reconnects += 1; - crate::app_eprintln!( - "[audio] streaming download error (attempt {}/{}): {} — reconnecting", - reconnects, - TRACK_STREAM_MAX_RECONNECTS, - e - ); - next_response = None; - continue 'outer; - } - }; - reconnects = 0; - let mut offset = 0; - while offset < chunk.len() { - if gen_arc.load(Ordering::SeqCst) != gen { - done.store(true, Ordering::SeqCst); - return; - } - let pushed = prod.push_slice(&chunk[offset..]); - if pushed == 0 { - tokio::time::sleep(Duration::from_millis(5)).await; - } else { - if !capture_over_limit { - if capture.len().saturating_add(pushed) <= TRACK_STREAM_PROMOTE_MAX_BYTES { - let from = offset; - let to = offset + pushed; - capture.extend_from_slice(&chunk[from..to]); - } else { - capture.clear(); - capture_over_limit = true; - } - } - if !capture_over_limit - && last_partial_loudness_emit.elapsed() >= Duration::from_millis(PARTIAL_LOUDNESS_EMIT_INTERVAL_MS) - { - last_partial_loudness_emit = Instant::now(); - if normalization_engine.load(Ordering::Relaxed) == 2 { - let target_lufs = f32::from_bits(normalization_target_lufs.load(Ordering::Relaxed)); - let pre_db = f32::from_bits( - loudness_pre_analysis_attenuation_db.load(Ordering::Relaxed), - ) - .clamp(-24.0, 0.0); - emit_partial_loudness_from_bytes(&app, &url, &capture, target_lufs, pre_db); - } - } - offset += pushed; - downloaded += pushed as u64; - } - } - } - if !capture_over_limit && !capture.is_empty() { - if let Some(track_id) = cache_track_id { - crate::app_deprintln!( - "[stream] legacy stream: capture complete track_id={} capture_mib={:.2} — full-track analysis (cpu-seed queue)", - track_id, - capture.len() as f64 / (1024.0 * 1024.0) - ); - let high = analysis_seed_high_priority_for_track(&app, &track_id); - if let Err(e) = - crate::submit_analysis_cpu_seed(app.clone(), track_id.clone(), capture.clone(), high).await - { - crate::app_eprintln!("[analysis] track seed failed for {}: {}", track_id, e); - } - } - *promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { - url: url.clone(), - data: capture, - }); - } - done.store(true, Ordering::SeqCst); - return; - } -} - -/// Linear downloader for `RangedHttpSource`: fills the pre-allocated buffer -/// from offset 0 to total_size. Reconnects via HTTP Range from the current -/// `downloaded` offset on transient errors. On completion (full track) the -/// data is promoted to `stream_completed_cache` for fast replay. -async fn ranged_download_task( - gen: u64, - gen_arc: Arc, - http_client: reqwest::Client, - app: AppHandle, - _duration_hint: f64, - url: String, - initial_response: reqwest::Response, - buf: Arc>>, - downloaded_to: Arc, - done: Arc, - promote_cache_slot: Arc>>, - normalization_engine: Arc, - normalization_target_lufs: Arc, - loudness_pre_analysis_attenuation_db: Arc, - cache_track_id: Option, - // When `Some`, ranged playback seeds on completion — defer HTTP backfill for that - // track; `None` for large files where ranged skips seed (needs backfill). - loudness_seed_hold: Option>>>, -) { - let _ranged_loudness_hold_clear = match (loudness_seed_hold.as_ref(), cache_track_id.as_ref()) { - (Some(slot), Some(tid)) => { - let t = tid.clone(); - { - let mut g = slot.lock().unwrap(); - *g = Some((t.clone(), gen)); - } - Some(RangedLoudnessSeedHoldClear { - slot: Arc::clone(slot), - tid: t, - gen, - }) - } - _ => None, - }; - let total_size = buf.lock().unwrap().len(); - let mut downloaded: usize = 0; - let mut reconnects: u32 = 0; - let mut next_response: Option = Some(initial_response); - let dl_started = Instant::now(); - let mut next_progress_mb: usize = 1; - let mut last_partial_loudness_emit = Instant::now() - Duration::from_secs(5); - - 'outer: loop { - let response = if let Some(r) = next_response.take() { - r - } else { - let mut req = http_client.get(&url); - if downloaded > 0 { - req = req.header(reqwest::header::RANGE, format!("bytes={downloaded}-")); - } - match req.send().await { - Ok(r) => r, - Err(err) => { - if reconnects >= TRACK_STREAM_MAX_RECONNECTS { - crate::app_eprintln!( - "[audio] ranged reconnect failed after {} attempts: {}", - reconnects, err - ); - break 'outer; - } - reconnects += 1; - tokio::time::sleep(Duration::from_millis(200)).await; - continue 'outer; - } - } - }; - if downloaded > 0 && response.status() != reqwest::StatusCode::PARTIAL_CONTENT { - crate::app_eprintln!( - "[audio] ranged reconnect returned {}, expected 206", - response.status() - ); - break 'outer; - } - if downloaded == 0 && !response.status().is_success() { - crate::app_eprintln!("[audio] ranged HTTP {}", response.status()); - break 'outer; - } - - let mut byte_stream = response.bytes_stream(); - while let Some(chunk) = byte_stream.next().await { - if gen_arc.load(Ordering::SeqCst) != gen { - done.store(true, Ordering::SeqCst); - return; - } - let chunk = match chunk { - Ok(c) => c, - Err(e) => { - if reconnects >= TRACK_STREAM_MAX_RECONNECTS { - crate::app_eprintln!( - "[audio] ranged dl error after {} reconnects: {}", - reconnects, e - ); - break 'outer; - } - reconnects += 1; - crate::app_eprintln!( - "[audio] ranged dl error (attempt {}/{}): {} — reconnecting", - reconnects, TRACK_STREAM_MAX_RECONNECTS, e - ); - next_response = None; - continue 'outer; - } - }; - reconnects = 0; - let writable = total_size.saturating_sub(downloaded); - if writable == 0 { - break; - } - let n = chunk.len().min(writable); - { - let mut b = buf.lock().unwrap(); - b[downloaded..downloaded + n].copy_from_slice(&chunk[..n]); - } - downloaded += n; - downloaded_to.store(downloaded, Ordering::SeqCst); - if downloaded >= PARTIAL_LOUDNESS_MIN_BYTES - && total_size > 0 - && last_partial_loudness_emit.elapsed() >= Duration::from_millis(PARTIAL_LOUDNESS_EMIT_INTERVAL_MS) - { - last_partial_loudness_emit = Instant::now(); - if normalization_engine.load(Ordering::Relaxed) == 2 { - let target_lufs = f32::from_bits(normalization_target_lufs.load(Ordering::Relaxed)); - let start_db = f32::from_bits(loudness_pre_analysis_attenuation_db.load(Ordering::Relaxed)) - .clamp(-24.0, 0.0); - if let Some(provisional_db) = - provisional_loudness_gain_from_progress(downloaded, total_size, target_lufs, start_db) - { - let track_key = playback_identity(&url).unwrap_or_else(|| url.clone()); - if partial_loudness_should_emit(&track_key, provisional_db) { - let _ = app.emit( - "analysis:loudness-partial", - PartialLoudnessPayload { - track_id: playback_identity(&url), - gain_db: provisional_db, - target_lufs, - is_partial: true, - }, - ); - } - } - } - } - let mb = downloaded / (1024 * 1024); - if mb >= next_progress_mb { - let pct = (downloaded as f64 / total_size as f64 * 100.0) as u32; - crate::app_deprintln!( - "[stream] dl progress: {} MB / {} MB ({}%)", - mb, - total_size / (1024 * 1024), - pct - ); - next_progress_mb = mb + 1; - } - if downloaded >= total_size { - break; - } - } - // Stream ended cleanly (or hit total_size). - break 'outer; - } - - done.store(true, Ordering::SeqCst); - - crate::app_deprintln!( - "[stream] dl done: {} / {} bytes in {:.2}s ({} reconnects)", - downloaded, - total_size, - dl_started.elapsed().as_secs_f64(), - reconnects - ); - - if downloaded == total_size && total_size > 0 && total_size <= TRACK_STREAM_PROMOTE_MAX_BYTES { - if let Some(ref tid) = cache_track_id { - crate::app_deprintln!( - "[stream] ranged: HTTP buffer full track_id={} size_mib={:.2} — cloning {} bytes then full-track analysis (cpu-seed queue; this task awaits completion)", - tid, - total_size as f64 / (1024.0 * 1024.0), - total_size - ); - } - let t_clone = Instant::now(); - let data = buf.lock().unwrap().clone(); - if total_size > 32 * 1024 * 1024 { - crate::app_deprintln!( - "[stream] ranged: buffer cloned in_ms={}", - t_clone.elapsed().as_millis() - ); - } - if let Some(track_id) = cache_track_id { - let high = analysis_seed_high_priority_for_track(&app, &track_id); - if let Err(e) = crate::submit_analysis_cpu_seed(app.clone(), track_id.clone(), data.clone(), high).await { - crate::app_eprintln!("[analysis] ranged seed failed for {}: {}", track_id, e); - } - } - *promote_cache_slot.lock().unwrap() = Some(PreloadedTrack { url, data }); - crate::app_deprintln!("[stream] promoted to stream_completed_cache for replay"); - } -} - -fn emit_partial_loudness_from_bytes( - app: &AppHandle, - url: &str, - bytes: &[u8], - target_lufs: f32, - pre_analysis_attenuation_db: f32, -) { - if bytes.len() < PARTIAL_LOUDNESS_MIN_BYTES { - crate::app_deprintln!( - "[normalization] partial-loudness skip reason=insufficient-bytes bytes={} min_bytes={}", - bytes.len(), - PARTIAL_LOUDNESS_MIN_BYTES - ); - return; - } - // Lightweight fallback based on buffered bytes count to keep CPU low. - let mb = bytes.len() as f32 / (1024.0 * 1024.0); - let pre_floor = pre_analysis_attenuation_db.clamp(-24.0, 0.0); - // Target-derived hint (e.g. -12 LUFS → -1 dB). Old `(hint).clamp(pre, 0)` left - // the hint when it lay inside [pre, 0] — e.g. -1 with pre=-6, so AAC/M4A - // streaming often sat at -1 dB until full analysis. Combine with user trim: - // stricter (more negative) pre wins; milder pre still caps vs the hint. - let heuristic_floor = (target_lufs + 11.0).clamp(-6.0, 0.0); - let floor_db = if pre_floor < heuristic_floor { - pre_floor - } else { - pre_floor.max(heuristic_floor) - }; - let gain_db = (-(mb * 0.7)).max(floor_db).min(0.0); - let track_key = playback_identity(url).unwrap_or_else(|| url.to_string()); - if !partial_loudness_should_emit(&track_key, gain_db as f32) { - crate::app_deprintln!( - "[normalization] partial-loudness skip reason=delta-below-threshold gain_db={:.2} threshold_db={:.2} track_id={:?}", - gain_db, - PARTIAL_LOUDNESS_DELTA_THRESHOLD_DB, - playback_identity(url) - ); - return; - } - crate::app_deprintln!( - "[normalization] partial-loudness emit bytes={} gain_db={:.2} target_lufs={:.2} track_id={:?}", - bytes.len(), - gain_db, - target_lufs, - playback_identity(url) - ); - let _ = app.emit( - "analysis:loudness-partial", - PartialLoudnessPayload { - track_id: playback_identity(url), - gain_db: gain_db as f32, - target_lufs, - is_partial: true, - }, - ); -} - -fn provisional_loudness_gain_from_progress( - downloaded: usize, - total_size: usize, - target_lufs: f32, - start_db_in: f32, -) -> Option { - if total_size == 0 || downloaded == 0 { - return None; - } - let progress = (downloaded as f32 / total_size as f32).clamp(0.0, 1.0); - // Move from startup attenuation toward a more realistic late-stream level. - // This avoids staying near -2 dB and then jumping hard when final LUFS lands. - let start_db = start_db_in.clamp(-24.0, 0.0).min(0.0); - let end_db = (target_lufs + 6.0).clamp(-10.0, -3.0).min(0.0); - let shaped = progress.powf(0.75); - Some(start_db + (end_db - start_db) * shaped) -} - -fn content_type_to_hint(ct: &str) -> Option { - 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 if ct.contains("wav") || ct.contains("wave") { Some("wav".into()) } - else if ct.contains("opus") { Some("opus".into()) } - else { None } -} - -// ─── SizedCursorSource — correct byte_len for seekable in-memory sources ────── -// -// rodio's internal ReadSeekSource wraps Cursor> 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>, - len: u64, -} - -impl Read for SizedCursorSource { - fn read(&mut self, buf: &mut [u8]) -> std::io::Result { - self.inner.read(buf) - } -} - -impl Seek for SizedCursorSource { - fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result { - self.inner.seek(pos) - } -} - -impl MediaSource for SizedCursorSource { - fn is_seekable(&self) -> bool { true } - fn byte_len(&self) -> Option { 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 + Source — identical interface to -// rodio::Decoder, so the rest of the source chain is unchanged. - -/// Debug logging: codec parameters in human-readable form to verify whether -/// playback is genuinely lossless. -fn log_codec_resolution( - tag: &str, - params: &symphonia::core::codecs::CodecParameters, - container_hint: Option<&str>, -) { - let codec_name = symphonia::default::get_codecs() - .get_codec(params.codec) - .map(|d| d.short_name) - .unwrap_or("?"); - let rate = params.sample_rate.map(|r| format!("{} Hz", r)).unwrap_or_else(|| "? Hz".into()); - let bits = params.bits_per_sample - .or(params.bits_per_coded_sample) - .map(|b| format!("{}-bit", b)) - .unwrap_or_else(|| "?-bit".into()); - let ch = params.channels - .map(|c| format!("{}ch", c.count())) - .unwrap_or_else(|| "?ch".into()); - let lossless = codec_name.starts_with("pcm") - || matches!( - codec_name, - "flac" | "alac" | "wavpack" | "monkeys-audio" | "tta" | "shorten" - ); - let kind = if lossless { "LOSSLESS" } else { "lossy" }; - crate::app_deprintln!( - "[stream] {tag}: codec={codec_name} ({kind}) {bits} {rate} {ch} container={}", - container_hint.unwrap_or("?") - ); -} - -/// 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, - current_frame_offset: usize, - format: Box, - total_duration: Option