SoulStream/src-tauri/src/headless_slsk.rs

//! A Soulseek session for a process with no page — so a row the phone does
//! not hold yet can be found and streamed from the car.
//!
//! "Needs the app" was the honest label for what the cold service could not
//! do: the resolver and the Soulseek session live in the page, and a car
//! that starts the app cold has no page. Kyle's answer to that label was the
//! right one — a tap should find a copy and play it, in the background —
//! and this is the smallest thing that does it: the stored account signs
//! in from here, one search, the best copy by rules close to the page's
//! (`src/lib/sources.js` songCandidates), a download into the same stream
//! cache the page uses, and the engine plays the bytes as they land, the
//! way `play_stream` does for the page.
//!
//! **One session per process, and the page adopts it.** When the app opens
//! over a car session, `commands::login_inner` takes this client instead of
//! logging in again (two logins from one address in quick succession is
//! what the server throttles), and every event from here on is forwarded to
//! the page's own pump through [`subscribe`]. The download in flight keeps
//! going: dropping a `Client` handle does not tear the connection down.
//!
//! Its own tokio runtime, because there is no Tauri one in a process the
//! car started; the two coexist when the app comes up.

use player::StreamProgress;
use soulseek_client::{Client, Config, DownloadState, Event};
use soulseek_proto::types::SearchResponse;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};
use tokio::sync::mpsc;

use crate::auto_browse::warn;

static RT: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
/// The handle, until the page adopts it.
static CLIENT: Mutex<Option<Client>> = Mutex::new(None);
/// Signed in, from this process's point of view.
static CONNECTED: Mutex<Option<String>> = Mutex::new(None);
/// token → the responses that came back for it.
static SEARCHES: Mutex<Vec<(u32, Vec<SearchResponse>)>> = Mutex::new(Vec::new());
/// download id → the progress the engine is reading.
static DOWNLOADS: Mutex<Vec<(u64, Arc<StreamProgress>)>> = Mutex::new(Vec::new());
/// The server's excluded search phrases, lowercased — a query containing
/// one returns nothing network-wide.
static EXCLUDED: Mutex<Vec<String>> = Mutex::new(Vec::new());
/// Where the page's pump listens once it exists.
static SUBSCRIBER: Mutex<Option<mpsc::UnboundedSender<Event>>> = Mutex::new(None);

pub(crate) fn rt() -> &'static tokio::runtime::Runtime {
    RT.get_or_init(|| {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .thread_name("slsk-cold")
            .build()
            .expect("a tokio runtime for the cold Soulseek session")
    })
}

fn lock<T>(m: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
    m.lock().unwrap_or_else(|e| e.into_inner())
}

/// True once this process has signed in (and the page has not yet adopted
/// the session — after adoption the page's own state is the truth).
pub fn connected() -> bool {
    lock(&CONNECTED).is_some() && lock(&CLIENT).is_some()
}

/// The library root, the way auto_browse derives it: `download_dir` from
/// config.json, else `<data_dir>/Library`.
fn library_dir(data_dir: &Path) -> PathBuf {
    std::fs::read_to_string(data_dir.join("config.json"))
        .ok()
        .and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok())
        .and_then(|v| v.get("download_dir").and_then(|d| d.as_str()).map(str::to_owned))
        .filter(|s| !s.is_empty())
        .map_or_else(|| data_dir.join("Library"), PathBuf::from)
}

/// Sign in with the stored account, once. Blocks the caller — a car tap
/// arrives on a worker thread that may wait — for up to ~20 s.
pub fn ensure_session(data_dir: &Path) -> Result<(), String> {
    if connected() {
        return Ok(());
    }
    crate::config::remember_secrets_dir_path(data_dir);
    let cfg_json = std::fs::read_to_string(data_dir.join("config.json"))
        .ok()
        .and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok())
        .unwrap_or_default();
    let username = cfg_json
        .get("username")
        .and_then(|u| u.as_str())
        .map(str::to_owned)
        .filter(|u| !u.is_empty())
        .ok_or("no Soulseek account on this phone")?;
    let password = crate::config::load_password(&username)
        .ok_or("the Soulseek account is not signed in on this phone")?;

    let mut cfg = Config::default();
    if let Some(p) = cfg_json.get("listen_port").and_then(|p| p.as_u64()) {
        cfg.listen_port = p as u16;
    }
    let downloads = library_dir(data_dir);
    cfg.downloads_dir = downloads.clone();
    cfg.shared_folders = vec![downloads];
    cfg.description = "Shared with SoulStream".into();

    let (client, events) = rt()
        .block_on(async {
            let (client, events) =
                tokio::time::timeout(Duration::from_secs(15), Client::connect(cfg))
                    .await
                    .map_err(|_| "could not reach the Soulseek server in time".to_string())?
                    .map_err(|e| format!("could not reach the Soulseek server: {e}"))?;
            tokio::time::timeout(Duration::from_secs(15), client.login(&username, &password))
                .await
                .map_err(|_| "Soulseek did not answer the sign-in in time".to_string())?
                .map_err(|e| format!("Soulseek refused the sign-in: {e}"))?;
            Ok::<_, String>((client, events))
        })?;
    warn(&format!("cold soulseek: signed in as {username}"));
    *lock(&CLIENT) = Some(client);
    *lock(&CONNECTED) = Some(username);
    rt().spawn(consume(events));
    Ok(())
}

/// The one consumer of the session's events: search replies and download
/// progress land in the tables above, and everything is forwarded to the
/// page's pump once one has subscribed.
async fn consume(mut events: mpsc::UnboundedReceiver<Event>) {
    while let Some(ev) = events.recv().await {
        match &ev {
            Event::SearchResult { token, response } => {
                let mut s = lock(&SEARCHES);
                if let Some((_, list)) = s.iter_mut().find(|(t, _)| t == token) {
                    list.push(response.clone());
                }
            }
            Event::Download(u) => {
                let progress = lock(&DOWNLOADS)
                    .iter()
                    .find(|(id, _)| *id == u.id)
                    .map(|(_, p)| p.clone());
                if let Some(p) = progress {
                    let (done, failed) = match &u.state {
                        DownloadState::Complete { .. } => (true, None),
                        DownloadState::Failed { reason } => (false, Some(reason.clone())),
                        DownloadState::Cancelled => (false, Some("stream cancelled".to_string())),
                        _ => (false, None),
                    };
                    p.update(u.received, done, failed);
                    if done {
                        lock(&DOWNLOADS).retain(|(id, _)| *id != u.id);
                    }
                }
            }
            Event::ExcludedSearchPhrases(p) => {
                *lock(&EXCLUDED) = p.iter().map(|s| s.to_lowercase()).collect();
            }
            Event::Disconnected { reason } => {
                warn(&format!("cold soulseek: disconnected ({reason})"));
                *lock(&CONNECTED) = None;
            }
            _ => {}
        }
        if let Some(tx) = lock(&SUBSCRIBER).as_ref() {
            let _ = tx.send(ev);
        }
    }
}

/// The page taking the session over: the client handle, and the account.
/// `None` when there is nothing to adopt (the page then signs in itself).
pub fn adopt() -> Option<(Client, String)> {
    let client = lock(&CLIENT).take()?;
    let user = lock(&CONNECTED).clone()?;
    Some((client, user))
}

/// Where the page's pump reads this session's events from after adoption.
pub fn subscribe() -> mpsc::UnboundedReceiver<Event> {
    let (tx, rx) = mpsc::unbounded_channel();
    *lock(&SUBSCRIBER) = Some(tx);
    rx
}

/* ── finding a copy ───────────────────────────────────────────────────── */

/// Streamable, the page's list (RESOLVER_STREAMABLE).
const STREAMABLE: [&str; 8] = ["flac", "mp3", "m4a", "ogg", "opus", "wav", "aiff", "aif"];
const LOSSLESS: [&str; 4] = ["flac", "wav", "aiff", "aif"];

/// Lowercased, punctuation folded to spaces, whitespace collapsed — the
/// page's normText, near enough for a contains() test.
fn norm(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    let mut space = true;
    for c in s.chars().flat_map(|c| c.to_lowercase()) {
        if c.is_alphanumeric() {
            out.push(c);
            space = false;
        } else if !space {
            out.push(' ');
            space = true;
        }
    }
    out.trim().to_owned()
}

/// The page's `simplifyTitle`, for the query: the title up to the first
/// `; : ( [`, and its first six words. The full title still has to be in
/// the filename.
fn simplify(title: &str) -> String {
    let first = title
        .split(|c| c == ';' || c == ':' || c == '(' || c == '[')
        .next()
        .unwrap_or("")
        .trim();
    let base = if first.is_empty() { title } else { first };
    let words: Vec<&str> = base.split_whitespace().take(6).collect();
    let out = words.join(" ");
    if out.is_empty() {
        title.to_owned()
    } else {
        out
    }
}

#[derive(Clone, Debug)]
pub struct Pick {
    pub username: String,
    pub path: String,
    pub size: u64,
    pub duration: Option<f64>,
    pub score: i64,
    pub slot: bool,
}

/// The best copies of a track on the network, best first. With
/// `require_identity` — the page's `requireIdentityEvidence`, for a rung
/// that searched by title alone — the artist has to appear somewhere in
/// the peer's path, or the copy is another artist's song of that name.
fn candidates(
    responses: &[SearchResponse],
    title: &str,
    artist: &str,
    expect_secs: Option<f64>,
    require_identity: bool,
) -> Vec<Pick> {
    let want = norm(title);
    let want_short = norm(&simplify(title));
    let artist_key = norm(artist);
    let mut out: Vec<Pick> = Vec::new();
    for r in responses {
        for f in &r.files {
            let name = f.filename.rsplit('\\').next().unwrap_or(&f.filename);
            let stem = name.rsplit_once('.').map(|(s, _)| s).unwrap_or(name);
            let nname = norm(stem);
            if !(nname.contains(&want) || (!want_short.is_empty() && nname.contains(&want_short))) {
                continue;
            }
            let ext = f.extension.to_lowercase();
            if !STREAMABLE.contains(&ext.as_str()) {
                continue;
            }
            if f.size == 0 || f.size > 400 * 1024 * 1024 {
                continue;
            }
            let bitrate = f.attributes.iter().find(|(k, _)| *k == 0).map(|(_, v)| *v);
            let duration = f.attributes.iter().find(|(k, _)| *k == 1).map(|(_, v)| *v as f64);
            if let (Some(want), Some(got)) = (expect_secs, duration) {
                // the page's durationMismatch: a copy that is not the same
                // recording — a radio edit, a live take — is not this song
                if (want - got).abs() > (want * 0.08).max(8.0) {
                    continue;
                }
            }
            let lossless = LOSSLESS.contains(&ext.as_str());
            let mut score: i64 = if lossless { 3000 } else { 1000 };
            if let Some(b) = bitrate {
                score += (b.min(400) as i64) * 2;
            }
            if f.size < 800 * 1024 {
                score -= 2000; // a snippet, not the song
            }
            let path_n = norm(&f.filename);
            let artist_evidence = !artist_key.is_empty() && path_n.contains(&artist_key);
            if require_identity && !artist_key.is_empty() && !artist_evidence {
                continue;
            }
            if artist_evidence {
                score += 400;
            }
            if nname.ends_with(&want) {
                score += 250;
            }
            if r.queue_length > 0 {
                score -= (r.queue_length.min(50) as i64) * 20;
            }
            out.push(Pick {
                username: r.username.clone(),
                path: f.filename.clone(),
                size: f.size,
                duration,
                score,
                slot: r.has_free_upload_slot,
            });
        }
    }
    out.sort_by(|a, b| b.score.cmp(&a.score));
    // Streaming needs bytes NOW: peers with a free slot first, best within
    // each half (the page's rule, learned on "Da Funk" 2026-08-03).
    let (free, busy): (Vec<Pick>, Vec<Pick>) = out.into_iter().partition(|p| p.slot);
    free.into_iter().chain(busy).collect()
}

fn strip_excluded(query: &str) -> String {
    let excluded = lock(&EXCLUDED).clone();
    let mut words: Vec<&str> = query.split_whitespace().collect();
    for phrase in &excluded {
        let lower = query.to_lowercase();
        if lower.contains(phrase.as_str()) {
            let bad: Vec<&str> = phrase.split_whitespace().collect();
            words.retain(|w| !bad.iter().any(|b| w.eq_ignore_ascii_case(b)));
        }
    }
    words.join(" ")
}

/// Every word of the artist still in the query, after the server's
/// excluded phrases came out of it.
fn keeps_artist(query: &str, artist: &str) -> bool {
    let q = norm(query);
    let words: Vec<&str> = q.split(' ').collect();
    norm(artist)
        .split(' ')
        .filter(|w| !w.is_empty())
        .all(|w| words.contains(&w))
}

/// The page's ladder for a song (`streamSong`): artist + simplified title,
/// then the simplified title alone. Each rung is one search, a few seconds
/// of replies, the best copies; the first rung that produces any wins. A
/// rung without the artist in it — the second, or the first once the
/// server's excluded phrases took the artist out — matches by title
/// alone, so the artist has to appear in the peer's path instead.
pub fn find(title: &str, artist: &str, expect_secs: Option<f64>) -> Result<Vec<Pick>, String> {
    let simple = simplify(title);
    let rungs = [format!("{artist} {simple}"), simple.clone()];
    for raw in rungs {
        let stripped = strip_excluded(&raw);
        let query = if stripped.trim().is_empty() { raw.clone() } else { stripped };
        if query.trim().is_empty() {
            continue;
        }
        let require_identity = !keeps_artist(&query, artist);
        // the client may spawn onto its runtime from any call; this thread
        // is not on it (see `fetch`)
        let _rt = rt().enter();
        let token = {
            let guard = lock(&CLIENT);
            let client = guard.as_ref().ok_or("not signed in to Soulseek")?;
            client.search(&query).map_err(|e| e.to_string())?
        };
        lock(&SEARCHES).push((token, Vec::new()));
        let started = Instant::now();
        let mut best: Vec<Pick> = Vec::new();
        let mut files = 0usize;
        // Up to eight seconds; done early once a dozen usable copies are in.
        while started.elapsed() < Duration::from_secs(8) {
            std::thread::sleep(Duration::from_millis(400));
            let responses = lock(&SEARCHES)
                .iter()
                .find(|(t, _)| *t == token)
                .map(|(_, r)| r.clone())
                .unwrap_or_default();
            files = responses.iter().map(|r| r.files.len()).sum();
            best = candidates(&responses, title, artist, expect_secs, require_identity);
            if best.len() >= 12 && started.elapsed() > Duration::from_millis(2500) {
                break;
            }
        }
        if let Some(client) = lock(&CLIENT).as_ref() {
            client.stop_search(token);
        }
        lock(&SEARCHES).retain(|(t, _)| *t != token);
        warn(&format!("cold soulseek: '{query}' → {files} files, {} copies", best.len()));
        if !best.is_empty() {
            return Ok(best);
        }
    }
    Ok(Vec::new())
}

/// Start fetching a peer's copy into the stream cache. Returns the cache
/// path, the partial path the bytes land in, the progress the engine reads,
/// and the download id.
pub fn fetch(
    data_dir: &Path,
    username: &str,
    path: &str,
    size: u64,
) -> Result<(PathBuf, PathBuf, Arc<StreamProgress>, u64), String> {
    let dir = crate::playback::cache_dir_in(data_dir);
    std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
    let cache = crate::playback::cache_file_in(&dir, username, path);
    let partial = crate::playback::partial_path(&cache);
    let _ = std::fs::remove_file(&partial);
    let progress = StreamProgress::new();
    let id = {
        // `download_to` spawns the transfer task with `tokio::spawn`, which
        // needs a runtime on the CALLING thread — and the caller here is
        // the engine's resolver thread, not a Tauri command. Without this
        // the spawn panics, the thread dies without a word (a cold process
        // has no stderr in logcat), and the card sits on BUFFERING for as
        // long as anyone looks (2026-09-18, the first cold fetch).
        let _rt = rt().enter();
        let guard = lock(&CLIENT);
        let client = guard.as_ref().ok_or("not signed in to Soulseek")?;
        client.download_to(username, path, size, cache.clone())
    };
    lock(&DOWNLOADS).push((id, progress.clone()));
    Ok((cache, partial, progress, id))
}

/// Give up on a fetch that never delivered.
pub fn cancel(id: u64) {
    let _rt = rt().enter();
    if let Some(client) = lock(&CLIENT).as_ref() {
        client.cancel_download(id);
    }
    lock(&DOWNLOADS).retain(|(i, _)| *i != id);
}