SoulStream/src-tauri/src/playback.rs

//! Playback orchestration: bridges the audio thread, the streaming cache,
//! and the webview.

use serde::Serialize;
use std::collections::{HashMap, VecDeque};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

static CACHE_PRUNE_LOCK: Mutex<()> = Mutex::new(());
use std::time::{Duration, Instant};
use tauri::{Emitter, Manager};

use player::{Player, PlayerEvent, StreamProgress};

// The audio thread keeps its responsive 60 ms command/end-detection loop, but
// the webview does not need 16-17 IPC/reactivity updates per second to move a
// progress indicator. Roughly four to five updates per second remain visually
// smooth while letting Chromium spend most playback time asleep.
const UI_POSITION_INTERVAL: Duration = Duration::from_millis(200);
const WAVEFORM_CACHE_MAX_ENTRIES: usize = 256;
const PROGRESSIVE_WAVEFORM_CACHE_MAX_ENTRIES: usize = 64;
const WAVEFORM_JOB_MAX_IDLE_ENTRIES: usize = 256;

/// Tiny insertion-order LRU used for decoded waveform data. Waveforms are
/// cheap individually, but a path-keyed session cache must not grow with every
/// track ever visited.
struct WaveformLru<T> {
    entries: HashMap<String, T>,
    order: VecDeque<String>,
    capacity: usize,
}

impl<T: Clone> WaveformLru<T> {
    fn new(capacity: usize) -> Self {
        Self {
            entries: HashMap::new(),
            order: VecDeque::new(),
            capacity,
        }
    }

    fn get(&mut self, key: &str) -> Option<T> {
        let value = self.entries.get(key)?.clone();
        self.touch(key);
        Some(value)
    }

    fn insert(&mut self, key: String, value: T) {
        self.entries.insert(key.clone(), value);
        self.touch(&key);
        while self.entries.len() > self.capacity {
            let Some(oldest) = self.order.pop_front() else {
                break;
            };
            self.entries.remove(&oldest);
        }
    }

    fn remove(&mut self, key: &str) {
        self.entries.remove(key);
        if let Some(index) = self.order.iter().position(|candidate| candidate == key) {
            self.order.remove(index);
        }
    }

    fn touch(&mut self, key: &str) {
        if let Some(index) = self.order.iter().position(|candidate| candidate == key) {
            self.order.remove(index);
        }
        self.order.push_back(key.to_string());
    }
}

#[derive(Clone)]
pub(crate) struct ProgressiveWaveform {
    decoded_bytes: u64,
    next_decode_bytes: u64,
    peaks: Vec<f32>,
}

impl ProgressiveWaveform {
    pub(crate) fn new(decoded_bytes: u64, total_bytes: u64, peaks: Vec<f32>) -> Self {
        Self {
            decoded_bytes,
            next_decode_bytes: next_progressive_waveform_threshold(decoded_bytes, total_bytes),
            peaks,
        }
    }

    /// Recompute only after the growing file has at least doubled. If a
    /// transfer restarted and the file shrank, force a fresh decode instead of
    /// returning peaks from bytes that no longer exist.
    pub(crate) fn should_recompute(&self, available_bytes: u64) -> bool {
        available_bytes < self.decoded_bytes
            || (available_bytes > self.decoded_bytes && available_bytes >= self.next_decode_bytes)
    }

    pub(crate) fn peaks(&self) -> Vec<f32> {
        self.peaks.clone()
    }
}

/// When the next partial re-decode is due. Early on the file doubles between
/// decodes — they're cheap and the shape fills in fast. But pure doubling made
/// the LAST gap span half the transfer (50% → 100%), so the drawn wave sat up
/// to 2× behind the buffer and the playhead visibly walked off its end
/// (decoded audio is always ahead of playback, the DRAWING wasn't). Past the
/// eighth-of-file mark the schedule switches to fixed total/8 steps: the wave
/// is never more than 12.5% behind the transfer, for at most ~7 extra decodes
/// spread across the download.
fn next_progressive_waveform_threshold(decoded_bytes: u64, total_bytes: u64) -> u64 {
    let doubled = decoded_bytes
        .saturating_mul(2)
        .max(decoded_bytes.saturating_add(1));
    let linear_step = decoded_bytes.saturating_add((total_bytes / 8).max(1));
    doubled
        .min(linear_step)
        .min(total_bytes.max(decoded_bytes.saturating_add(1)))
}

pub struct PlayerState {
    pub player: Player,
    /// Active stream downloads: download id → progress shared with the player.
    pub streams: Mutex<HashMap<u64, Arc<StreamProgress>>>,
    /// Cache target for each active stream. This lets a library save adopt an
    /// in-flight stream instead of requesting the same peer/path again.
    pub stream_caches: Mutex<HashMap<u64, PathBuf>>,
    /// Bounded completed-waveform LRU, keyed by file path + bucket count.
    waveforms: Mutex<WaveformLru<Vec<f32>>>,
    /// Last useful partial decode per active/recent stream. A growing file is
    /// decoded again only after its byte length doubles.
    progressive_waveforms: Mutex<WaveformLru<ProgressiveWaveform>>,
    /// Per-key single-flight locks. The owned guard moves into the blocking
    /// decoder so cancellation of the IPC future cannot expose overlapping
    /// work while that decoder is still running.
    waveform_jobs: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
    /// Pre-buffered next-track downloads: download id → cache target.
    /// Their events are swallowed (no UI, no library pipeline); play_stream
    /// attaches to them instead of restarting the transfer.
    pub preloads: Mutex<HashMap<u64, PathBuf>>,
    /// Transfers whose completed cache file should also be organized into the
    /// library. A save can subscribe to a stream/preload already in progress.
    pub deferred_saves: Mutex<HashMap<u64, PathBuf>>,
}

#[derive(Serialize, Clone)]
#[serde(tag = "kind", rename_all = "kebab-case")]
enum PlayerUiEvent {
    Buffering,
    Playing,
    Paused,
    Stopped,
    TrackEnded { gapless: bool },
    Position { secs: f64, duration: Option<f64> },
    Error { message: String },
}

/// Create the audio thread and forward its events to the webview.
pub fn init(app: &tauri::AppHandle) {
    let output_device = crate::config::load(app).audio_output_device;
    let (player, events) = Player::spawn_with_device(output_device);
    app.manage(PlayerState {
        player,
        streams: Mutex::new(HashMap::new()),
        stream_caches: Mutex::new(HashMap::new()),
        waveforms: Mutex::new(WaveformLru::new(WAVEFORM_CACHE_MAX_ENTRIES)),
        progressive_waveforms: Mutex::new(WaveformLru::new(PROGRESSIVE_WAVEFORM_CACHE_MAX_ENTRIES)),
        waveform_jobs: Mutex::new(HashMap::new()),
        preloads: Mutex::new(HashMap::new()),
        deferred_saves: Mutex::new(HashMap::new()),
    });

    let handle = app.clone();
    std::thread::Builder::new()
        .name("player-events".into())
        .spawn(move || {
            let mut last_position_emit = Instant::now() - UI_POSITION_INTERVAL;
            for ev in events {
                let ui = match ev {
                    PlayerEvent::Buffering => {
                        last_position_emit = Instant::now() - UI_POSITION_INTERVAL;
                        PlayerUiEvent::Buffering
                    }
                    PlayerEvent::Playing => {
                        last_position_emit = Instant::now() - UI_POSITION_INTERVAL;
                        PlayerUiEvent::Playing
                    }
                    PlayerEvent::Paused => PlayerUiEvent::Paused,
                    PlayerEvent::Stopped => PlayerUiEvent::Stopped,
                    PlayerEvent::TrackEnded { gapless } => {
                        last_position_emit = Instant::now() - UI_POSITION_INTERVAL;
                        PlayerUiEvent::TrackEnded { gapless }
                    }
                    PlayerEvent::Position { secs, duration } => {
                        let now = Instant::now();
                        if now.duration_since(last_position_emit) < UI_POSITION_INTERVAL {
                            continue;
                        }
                        last_position_emit = now;
                        PlayerUiEvent::Position { secs, duration }
                    }
                    PlayerEvent::Error(message) => PlayerUiEvent::Error { message },
                };
                let _ = handle.emit("soulseek:player", ui);
            }
        })
        .expect("spawn player event forwarder");
}

impl PlayerState {
    pub(crate) fn cached_waveform(&self, key: &str) -> Option<Vec<f32>> {
        self.waveforms.lock().unwrap().get(key)
    }

    pub(crate) fn cache_waveform(&self, key: String, peaks: Vec<f32>) {
        self.waveforms.lock().unwrap().insert(key, peaks);
    }

    pub(crate) fn remove_cached_waveform(&self, key: &str) {
        self.waveforms.lock().unwrap().remove(key);
    }

    pub(crate) fn progressive_waveform(&self, key: &str) -> Option<ProgressiveWaveform> {
        self.progressive_waveforms.lock().unwrap().get(key)
    }

    pub(crate) fn cache_progressive_waveform(&self, key: String, waveform: ProgressiveWaveform) {
        self.progressive_waveforms
            .lock()
            .unwrap()
            .insert(key, waveform);
    }

    pub(crate) fn remove_progressive_waveform(&self, key: &str) {
        self.progressive_waveforms.lock().unwrap().remove(key);
    }

    pub(crate) fn waveform_job(&self, key: &str) -> Arc<tokio::sync::Mutex<()>> {
        let mut jobs = self.waveform_jobs.lock().unwrap();
        if jobs.len() >= WAVEFORM_JOB_MAX_IDLE_ENTRIES {
            // An entry with only the map's Arc is idle and safe to forget.
            // Active/waiting jobs may temporarily exceed the soft bound, but
            // they must keep their shared lock until their decoder exits.
            let idle: Vec<String> = jobs
                .iter()
                .filter(|(_, job)| Arc::strong_count(job) == 1)
                .map(|(key, _)| key.clone())
                .collect();
            for idle_key in idle {
                if jobs.len() < WAVEFORM_JOB_MAX_IDLE_ENTRIES {
                    break;
                }
                jobs.remove(&idle_key);
            }
        }
        jobs.entry(key.to_string())
            .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
            .clone()
    }

    pub(crate) fn release_waveform_job(&self, key: &str, job: &Arc<tokio::sync::Mutex<()>>) {
        let mut jobs = self.waveform_jobs.lock().unwrap();
        if jobs
            .get(key)
            .is_some_and(|current| Arc::ptr_eq(current, job) && Arc::strong_count(current) == 2)
        {
            jobs.remove(key);
        }
    }
}

/// Stable cache filename for a remote file: short hash + original name.
pub fn cache_path(app: &tauri::AppHandle, username: &str, remote_path: &str) -> Option<PathBuf> {
    let dir = app.path().app_data_dir().ok()?.join("stream-cache");
    std::fs::create_dir_all(&dir).ok()?;

    let mut hash: u64 = 0xcbf29ce484222325;
    for b in username.bytes().chain(remote_path.bytes()) {
        hash ^= b as u64;
        hash = hash.wrapping_mul(0x100000001b3);
    }
    let name = remote_path.rsplit('\\').next().unwrap_or(remote_path);
    let safe: String = name
        .chars()
        .map(|c| match c {
            '<' | '>' | ':' | '"' | '/' | '\\' | '|' | '?' | '*' => '_',
            c if (c as u32) < 0x20 => '_',
            c => c,
        })
        .collect();
    Some(dir.join(format!("{hash:016x} {safe}")))
}

/// The in-flight download target for a cache file: `<name>.<ext>.partial`.
pub fn partial_path(cache: &std::path::Path) -> PathBuf {
    cache.with_extension(format!(
        "{}partial",
        cache
            .extension()
            .map(|e| format!("{}.", e.to_string_lossy()))
            .unwrap_or_default()
    ))
}

/// Enforce the configured streamed-track retention on the cache: drop files
/// not played within the retention window, then oldest-first beyond the size
/// cap. The playing track's file is held open by the OS, so its delete fails
/// and it is simply kept.
pub fn prune_cache(app: &tauri::AppHandle) {
    // Completion/settings paths can request pruning concurrently. Serialize
    // their directory snapshots so a stale running total cannot over-delete.
    let _prune_guard = CACHE_PRUNE_LOCK
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    let cfg = crate::config::load(app);
    let max_bytes = crate::config::stream_cache_max_mb(&cfg) * 1024 * 1024; // 0 = unlimited
    let retention = std::time::Duration::from_secs(
        crate::config::stream_retention_days(&cfg) as u64 * 24 * 60 * 60,
    );
    // In keep-forever mode streams land in the library, so the cache is only
    // a transfer buffer — age expiry off, size cap still applies.
    let age_prune = !crate::config::stream_keep_forever(&cfg);

    let Ok(dir) = app.path().app_data_dir().map(|d| d.join("stream-cache")) else {
        return;
    };
    let Ok(entries) = std::fs::read_dir(&dir) else {
        return;
    };

    let mut files: Vec<(std::time::SystemTime, u64, PathBuf)> = entries
        .flatten()
        .filter_map(|e| {
            let path = e.path();
            if path
                .extension()
                .is_some_and(|ext| ext.to_string_lossy().eq_ignore_ascii_case("partial"))
            {
                // Active downloads are transfer buffers, not evictable cache
                // entries. Deleting one can make its final rename fail.
                return None;
            }
            let md = e.metadata().ok()?;
            if !md.is_file() {
                return None;
            }
            Some((md.modified().ok()?, md.len(), path))
        })
        .collect();
    files.sort_by_key(|(t, ..)| *t);

    let now = std::time::SystemTime::now();
    let mut total: u64 = files.iter().map(|(_, len, _)| len).sum();
    let mut kept = Vec::new();
    for (modified, len, path) in files {
        let expired = age_prune
            && now
                .duration_since(modified)
                .map(|age| age > retention)
                .unwrap_or(false);
        if expired && std::fs::remove_file(&path).is_ok() {
            total -= len;
            continue;
        }
        kept.push((len, path));
    }

    if max_bytes == 0 {
        return;
    }
    for (len, path) in kept {
        if total <= max_bytes {
            break;
        }
        if std::fs::remove_file(&path).is_ok() {
            total -= len;
        }
    }
}

/// Mark a cache file as freshly used so retention counts from the last play,
/// not the first stream.
pub fn touch(path: &std::path::Path) {
    if let Ok(f) = std::fs::File::options().write(true).open(path) {
        let _ = f.set_modified(std::time::SystemTime::now());
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn waveform_lru_evicts_oldest_and_refreshes_hits() {
        let mut cache = WaveformLru::new(2);
        cache.insert("a".into(), vec![1.0]);
        cache.insert("b".into(), vec![2.0]);
        assert_eq!(cache.get("a"), Some(vec![1.0]));

        cache.insert("c".into(), vec![3.0]);
        assert_eq!(cache.get("b"), None);
        assert_eq!(cache.get("a"), Some(vec![1.0]));
        assert_eq!(cache.get("c"), Some(vec![3.0]));
        assert_eq!(cache.entries.len(), 2);
    }

    #[test]
    fn progressive_waveform_waits_for_doubling_but_restarts_after_shrink() {
        let waveform = ProgressiveWaveform::new(10, 100, vec![1.0]);
        assert!(!waveform.should_recompute(10));
        assert!(!waveform.should_recompute(19));
        assert!(waveform.should_recompute(20));
        assert!(waveform.should_recompute(9));
    }

    #[test]
    fn progressive_thresholds_bound_lag_and_total_decode_cost() {
        let total: u64 = 256 * 1024 * 1024;
        let mut decoded: u64 = 1024 * 1024;
        let mut cumulative = 0u64;
        let mut steps = 0;
        while decoded < total {
            cumulative += decoded;
            let next = next_progressive_waveform_threshold(decoded, total);
            assert!(next > decoded, "thresholds must advance");
            // The drawn wave may never fall more than an eighth of the file
            // behind the transfer — the doubling-only schedule's half-file
            // final gap is the bug this replaced (playhead walking off the
            // stale wave's end mid-song).
            assert!(
                next - decoded <= total / 8,
                "gap {} exceeds total/8",
                next - decoded
            );
            decoded = next;
            steps += 1;
        }
        assert_eq!(decoded, total);
        // Early doubling + eighth-of-file steps: the whole progressive series
        // costs a small constant number of full-file decodes, spread across
        // the download.
        assert!(steps <= 16, "too many recomputes: {steps}");
        assert!(cumulative <= total * 4, "cumulative redecode too large");
    }
}