diff --git a/Cargo.lock b/Cargo.lock index 59b613e..a7522dd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1750,6 +1750,12 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hyper" version = "1.10.1" @@ -2398,6 +2404,7 @@ name = "metadata" version = "0.1.0" dependencies = [ "futures", + "httpdate", "image", "log", "reqwest 0.12.28", diff --git a/crates/metadata/Cargo.toml b/crates/metadata/Cargo.toml index ab96f30..76c71e7 100644 --- a/crates/metadata/Cargo.toml +++ b/crates/metadata/Cargo.toml @@ -13,3 +13,5 @@ serde_json.workspace = true tokio.workspace = true thiserror.workspace = true log.workspace = true +# RFC-7231 Date parsing for preview-lease math (tiny, no deps of its own). +httpdate = "1" diff --git a/crates/metadata/src/lib.rs b/crates/metadata/src/lib.rs index 50fc02c..05b340f 100644 --- a/crates/metadata/src/lib.rs +++ b/crates/metadata/src/lib.rs @@ -918,6 +918,49 @@ impl MetadataClient { self.cached_json(url, false).await } + /// Fetch one Deezer track FRESH, bypassing the catalog cache entirely. + /// + /// This exists for preview-lease minting and must not be used for page + /// data: every `/track/{id}` response mints a new signed preview URL that + /// expires in ~13–15 minutes (Phase 0), so a cached body is a stale lease + /// by construction. Pacing and the compliant UA still apply; two attempts, + /// then the caller classifies the failure. The generic cache is also + /// UPDATED with the fresh body so later page reads can't resurrect an + /// older, already-expired preview URL for this track. + pub async fn deezer_track_uncached(&self, id: u64) -> Option { + let url = format!("https://api.deezer.com/track/{id}"); + for attempt in 0..2u32 { + self.deezer_throttle().await; + let resp = self.http.get(&url).send().await; + match resp { + Ok(r) if r.status().is_success() => { + let Ok(body) = r.json::().await else { + return None; + }; + // Deezer signals quota inside a 200 body; honor the same + // retry the cached path uses before giving up. + let quota = body + .pointer("/error/code") + .and_then(|c| c.as_u64()) + .is_some_and(|c| c == 4); + if quota && attempt == 0 { + continue; + } + let mut cache = self.catalog_cache.lock().await; + cache.insert(url.clone(), (Instant::now(), body.clone())); + return Some(body); + } + _ => { + if attempt == 0 { + continue; + } + return None; + } + } + } + None + } + /// GET a MusicBrainz endpoint through the same cache/single-flight layer. /// The one-request-per-second gate belongs inside the cache-miss /// initializer: putting it before the cache lookup made two completely warm @@ -6235,6 +6278,146 @@ struct CoverCandidate { url: String, } +// -------------------------------------------------------------------------- +// Preview media fetch — the bounded, allowlisted transfer behind the preview +// resolver (scrollable-track-previews Phase 2). Kept separate from the shared +// catalog client because it needs a CUSTOM REDIRECT POLICY: every hop is +// re-validated against the media allowlist, where the shared client follows +// redirects freely (fine for JSON APIs, not for media bytes we will decode). +// -------------------------------------------------------------------------- + +/// Hosts preview media may be fetched from. HTTPS-only, exact match — one +/// stable CDN host served 71/71 Phase 0 samples. Mirrors the frontend +/// allowlist in `src/lib/preview/redact.js`; a new provider host is a +/// reviewed change in both places. +pub const PREVIEW_MEDIA_HOSTS: [&str; 1] = ["cdnt-preview.dzcdn.net"]; + +/// Hard cap on a preview transfer. Real clips are ~480 KB (Phase 0, zero +/// variance); 2 MB tolerates provider drift while still refusing anything +/// that plainly is not a 30-second clip. +pub const PREVIEW_MEDIA_MAX_BYTES: usize = 2_000_000; + +pub struct PreviewMediaResponse { + pub bytes: Vec, + pub content_type: Option, + /// Provider `Date` header as epoch ms, for provider-clock lease math. + pub date_ms: Option, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum PreviewFetchError { + /// Initial URL or a redirect target fell outside the https+allowlist gate. + DisallowedUrl, + /// The CDN answered with a non-success status (403 = stale lease). + Status(u16), + /// Body exceeded [`PREVIEW_MEDIA_MAX_BYTES`]. + TooLarge, + /// Transport-level failure (timeout, DNS, connection). + Network, +} + +pub fn preview_url_allowed(url: &str) -> bool { + let Ok(u) = reqwest::Url::parse(url) else { + return false; + }; + u.scheme() == "https" + && u.host_str() + .is_some_and(|h| PREVIEW_MEDIA_HOSTS.contains(&h)) +} + +fn preview_media_client() -> &'static reqwest::Client { + use std::sync::OnceLock; + static CLIENT: OnceLock = OnceLock::new(); + CLIENT.get_or_init(|| { + reqwest::Client::builder() + .user_agent(USER_AGENT) + .timeout(Duration::from_secs(15)) + .redirect(reqwest::redirect::Policy::custom(|attempt| { + // Every hop passes the same gate as the initial URL; a + // redirect off the allowlist is a hard stop, never a follow. + let ok = attempt.url().scheme() == "https" + && attempt + .url() + .host_str() + .is_some_and(|h| PREVIEW_MEDIA_HOSTS.contains(&h)); + if !ok { + return attempt.error("redirect target off the preview media allowlist"); + } + if attempt.previous().len() > 4 { + return attempt.error("too many redirects"); + } + attempt.follow() + })) + .build() + .expect("build preview media client") + }) +} + +/// Fetch preview bytes with the full host/response policy applied. Callers +/// never see the URL again after this returns — bytes, MIME and provider +/// time are all a resolver needs. +pub async fn fetch_preview_media(url: &str) -> Result { + if !preview_url_allowed(url) { + return Err(PreviewFetchError::DisallowedUrl); + } + let resp = preview_media_client() + .get(url) + .send() + .await + .map_err(|e| { + // A redirect rejected by the policy surfaces as a generic + // reqwest error; classify it as the policy violation it is. + if e.is_redirect() { + PreviewFetchError::DisallowedUrl + } else { + PreviewFetchError::Network + } + })?; + let status = resp.status(); + if !status.is_success() { + return Err(PreviewFetchError::Status(status.as_u16())); + } + if let Some(len) = resp.content_length() { + if len as usize > PREVIEW_MEDIA_MAX_BYTES { + return Err(PreviewFetchError::TooLarge); + } + } + let content_type = resp + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .map(String::from); + let date_ms = resp + .headers() + .get(reqwest::header::DATE) + .and_then(|v| v.to_str().ok()) + .and_then(|s| httpdate::parse_http_date(s).ok()) + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| d.as_millis() as i64); + + // Stream with the cap enforced even when Content-Length lied or was + // absent — the count is on received bytes, not the header's promise. + let mut bytes = Vec::with_capacity(512 * 1024); + let mut stream = resp; + loop { + match stream.chunk().await { + Ok(Some(chunk)) => { + if bytes.len() + chunk.len() > PREVIEW_MEDIA_MAX_BYTES { + return Err(PreviewFetchError::TooLarge); + } + bytes.extend_from_slice(&chunk); + } + Ok(None) => break, + Err(_) => return Err(PreviewFetchError::Network), + } + } + Ok(PreviewMediaResponse { + bytes, + content_type, + date_ms, + }) +} + /// Decode candidate bytes and keep only plausible front covers: square-ish /// (grids/banners/booklet scans are not covers) and at least 300px. fn decode_cover( diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index d4a150c..618d547 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -5,6 +5,7 @@ mod hero_crop; mod hero_detector; mod pipeline; mod playback; +mod preview; mod state; #[cfg_attr(mobile, tauri::mobile_entry_point)] @@ -29,6 +30,7 @@ pub fn run() { hero_detector::warm_up(app.handle().clone()); playback::init(app.handle()); playback::prune_cache(app.handle()); + preview::init(app.handle()); if let Err(e) = pipeline::init(app.handle()) { log::error!("library init failed: {e}"); } @@ -101,6 +103,11 @@ pub fn run() { commands::preload_stream, commands::preload_cached_stream, commands::stream_waveform, + preview::preview_resolve, + preview::preview_load, + preview::preview_control, + preview::preview_dispose, + preview::player_pause_acked, ]) .run({ #[allow(unused_mut)] diff --git a/src-tauri/src/preview.rs b/src-tauri/src/preview.rs new file mode 100644 index 0000000..9ca0e60 --- /dev/null +++ b/src-tauri/src/preview.rs @@ -0,0 +1,510 @@ +//! Preview playback backend — scrollable-track-previews Phase 2. +//! +//! A wholly separate audio path from the ordinary player: its own `Player` +//! instance, its own event channel (`soulstream:preview`), its own bounded +//! on-disk clip cache. Nothing here touches the queue, history, stream +//! caches, or the Soulseek resolver — that isolation is the feature's +//! non-negotiable boundary. +//! +//! The resolver contract mirrors `src/lib/preview/resolver.js`: stable track +//! id in, `{ ok, lease | failure }` out, with the failure classes Phase 0 +//! actually observed. Signed URLs never cross the IPC boundary — the +//! renderer receives an opaque lease id; bytes live in a temp file the +//! renderer never sees either. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use serde::Serialize; +use tauri::{Emitter, Manager, State}; + +use player::{Player, PlayerEvent}; + +use crate::pipeline::LibraryState; +use crate::playback::PlayerState; + +/// Position events at ~10/s: the modal's canvas scrubber wants smoother +/// motion than the main bar's 200 ms cadence, and one small window is cheap. +const PREVIEW_POSITION_INTERVAL: Duration = Duration::from_millis(100); + +/// Lease-lifetime constants, mirroring `src/lib/preview/lease.js` — the two +/// sides MUST agree or the renderer would trust a lease the backend already +/// considers dead. Values are ADRs backed by the Phase 0 soak. +const SAFETY_MARGIN_MS: i64 = 20_000; +const FALLBACK_SAFE_MS: i64 = 60_000; +const MAX_SAFE_MS: i64 = 30 * 60_000; + +/// Clip files older than this are swept on init and on each resolve — a +/// preview session never legitimately needs an hour-old clip. +const CLIP_MAX_AGE: Duration = Duration::from_secs(60 * 60); + +pub struct PreviewState { + pub player: Player, + /// opaque lease id → clip file. Entries die with the session (dispose) + /// or the sweep; the map is small by construction (active + prefetch). + leases: Mutex>, + clip_dir: PathBuf, + lease_counter: AtomicU64, +} + +// ------------------------------------------------------------------ init -- + +pub fn init(app: &tauri::AppHandle) { + // Same configured output device as the main player, so previews follow + // the user's routing. (Device CHANGES while a preview session is open + // are part of the Phase 2 focus matrix, handled at the coordinator.) + let output_device = crate::config::load(app).audio_output_device; + let (player, events) = Player::spawn_with_device(output_device); + + let clip_dir = app + .path() + .app_cache_dir() + .map(|d| d.join("preview-clips")) + .unwrap_or_else(|_| std::env::temp_dir().join("soulstream-preview-clips")); + let _ = std::fs::create_dir_all(&clip_dir); + sweep_old_clips(&clip_dir); + + app.manage(PreviewState { + player, + leases: Mutex::new(HashMap::new()), + clip_dir, + lease_counter: AtomicU64::new(1), + }); + + let handle = app.clone(); + std::thread::Builder::new() + .name("preview-events".into()) + .spawn(move || { + let mut last_position = Instant::now() - PREVIEW_POSITION_INTERVAL; + for ev in events { + let ui = match ev { + PlayerEvent::Buffering => PreviewUiEvent::Buffering, + PlayerEvent::Playing => PreviewUiEvent::Playing, + PlayerEvent::Paused => PreviewUiEvent::Paused, + PlayerEvent::Stopped => PreviewUiEvent::Stopped, + PlayerEvent::TrackEnded { .. } => PreviewUiEvent::Ended, + PlayerEvent::Position { secs, duration } => { + let now = Instant::now(); + if now.duration_since(last_position) < PREVIEW_POSITION_INTERVAL { + continue; + } + last_position = now; + PreviewUiEvent::Position { secs, duration } + } + PlayerEvent::Error(message) => PreviewUiEvent::Error { message }, + }; + let _ = handle.emit("soulstream:preview", ui); + } + }) + .expect("spawn preview event forwarder"); +} + +#[derive(Serialize, Clone)] +#[serde(tag = "kind", rename_all = "kebab-case")] +enum PreviewUiEvent { + Buffering, + Playing, + Paused, + Stopped, + Ended, + Position { secs: f64, duration: Option }, + Error { message: String }, +} + +fn sweep_old_clips(dir: &std::path::Path) { + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + for entry in entries.flatten() { + let stale = entry + .metadata() + .and_then(|m| m.modified()) + .map(|t| t.elapsed().unwrap_or_default() > CLIP_MAX_AGE) + .unwrap_or(true); + if stale { + let _ = std::fs::remove_file(entry.path()); + } + } +} + +// ------------------------------------------------------------- resolver --- + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct PreviewLeaseDto { + pub opaque_lease_id: String, + pub safe_for_ms: i64, + pub byte_length: usize, + pub mime_type: String, + /// Hint only; the transport reports the decoder's real duration. + pub clip_duration_ms: i64, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct PreviewResolveResult { + pub ok: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub lease: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub failure: Option<&'static str>, + pub retryable: bool, +} + +fn fail(failure: &'static str, retryable: bool) -> PreviewResolveResult { + PreviewResolveResult { + ok: false, + lease: None, + failure: Some(failure), + retryable, + } +} + +/// Resolve a stable Deezer track id into a playable lease. Every call mints +/// fresh — /track responses carry per-request signed URLs (Phase 0), so a +/// cached body is a stale lease by construction. The single-retry cap on +/// failures lives with the caller (renderer), matching the fixture resolver. +#[tauri::command] +pub async fn preview_resolve( + library: State<'_, LibraryState>, + state: State<'_, PreviewState>, + track_id: u64, +) -> Result { + sweep_old_clips(&state.clip_dir); + + let Some(body) = library.metadata.deezer_track_uncached(track_id).await else { + return Ok(fail("network", true)); + }; + if let Some(err) = body.get("error") { + let class = err.get("type").and_then(|t| t.as_str()).unwrap_or(""); + return Ok(match class { + // Deezer's 200-with-error-body shapes, from Phase 0. + "DataException" | "InvalidQueryException" => fail("not_found", false), + _ => fail("temporarily_failed", true), + }); + } + let preview_url = body + .get("preview") + .and_then(|p| p.as_str()) + .unwrap_or("") + .to_string(); + if preview_url.is_empty() { + return Ok(fail("not_provided", false)); + } + + let media = match metadata::fetch_preview_media(&preview_url).await { + Ok(m) => m, + Err(metadata::PreviewFetchError::Status(403)) => return Ok(fail("stale_lease", true)), + Err(metadata::PreviewFetchError::Status(_)) => return Ok(fail("temporarily_failed", true)), + Err(metadata::PreviewFetchError::DisallowedUrl) => return Ok(fail("invalid_media", false)), + Err(metadata::PreviewFetchError::TooLarge) => return Ok(fail("invalid_media", false)), + Err(metadata::PreviewFetchError::Network) => return Ok(fail("network", true)), + }; + + // MIME gate first (the stale-lease body is an HTML error page), then the + // container transform every Deezer clip needs: skip the universal empty + // ID3v2.4 shell that crashes the symphonia entry (Phase 0.3), then prove + // the remainder starts on an MPEG sync before it goes anywhere near a + // decoder. + let mime_ok = media + .content_type + .as_deref() + .is_some_and(|m| m.starts_with("audio/mpeg")); + if !mime_ok { + return Ok(fail("invalid_media", false)); + } + let audio = strip_id3v2(&media.bytes); + if !starts_with_mpeg_sync(audio) { + return Ok(fail("invalid_media", false)); + } + + let nonce = state.lease_counter.fetch_add(1, Ordering::Relaxed); + let lease_id = format!("pv-{track_id}-{nonce}"); + let path = state.clip_dir.join(format!("{lease_id}.mp3")); + if std::fs::write(&path, audio).is_err() { + return Ok(fail("temporarily_failed", true)); + } + state + .leases + .lock() + .expect("preview lease lock") + .insert(lease_id.clone(), path); + + let safe_for_ms = compute_safe_for_ms( + parse_exp_ms(&preview_url), + media.date_ms, + now_epoch_ms(), + ); + + Ok(PreviewResolveResult { + ok: true, + lease: Some(PreviewLeaseDto { + opaque_lease_id: lease_id, + safe_for_ms, + byte_length: audio.len(), + mime_type: "audio/mpeg".into(), + clip_duration_ms: 30_000, + }), + failure: None, + retryable: false, + }) +} + +// ------------------------------------------------------------ transport --- + +/// Load a resolved clip into the preview player. `start_paused` uses the +/// desired-paused latch (Phase 0.4): the decoder comes up parked, so opening +/// against a still-negotiating focus transfer can never leak audio. +#[tauri::command] +pub fn preview_load( + state: State<'_, PreviewState>, + lease_id: String, + start_paused: bool, +) -> Result<(), String> { + let path = state + .leases + .lock() + .expect("preview lease lock") + .get(&lease_id) + .cloned() + .ok_or("unknown or expired lease")?; + state.player.play_local(&path, Some(30.0)); + if start_paused { + // Ack receipt matters less than the latch being set before any + // decoder can finish; a timeout here still leaves the latch armed. + let _ = state + .player + .pause_acked() + .recv_timeout(Duration::from_millis(750)); + } + Ok(()) +} + +#[tauri::command] +pub fn preview_control( + state: State<'_, PreviewState>, + action: String, + value: Option, +) -> Result<(), String> { + match action.as_str() { + "pause" => state.player.pause(), + "resume" => state.player.resume(), + "stop" => state.player.stop(), + "seek" => state.player.seek(value.unwrap_or(0.0)), + "volume" => state.player.set_volume(value.unwrap_or(0.7) as f32), + other => return Err(format!("unknown preview action {other}")), + } + Ok(()) +} + +/// Session teardown: silence the transport and drop every clip file. Safe to +/// call repeatedly (close, navigation, hot reload all funnel here). +#[tauri::command] +pub fn preview_dispose(state: State<'_, PreviewState>) -> Result<(), String> { + state.player.stop(); + let mut leases = state.leases.lock().expect("preview lease lock"); + for (_, path) in leases.drain() { + let _ = std::fs::remove_file(path); + } + Ok(()) +} + +// ----------------------------------------------------- main-player ack ---- + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct PauseAckDto { + pub generation: u64, + pub sink_was_active: bool, + pub latched: bool, +} + +/// The focus coordinator's silence request against the MAIN player: pause +/// with a generation-bound acknowledgement (Phase 0.4 primitive). A timeout +/// is an error — the caller keeps preview parked; it never assumes silence. +#[tauri::command] +pub async fn player_pause_acked(players: State<'_, PlayerState>) -> Result { + let rx = players.player.pause_acked(); + tauri::async_runtime::spawn_blocking(move || { + rx.recv_timeout(Duration::from_millis(1_500)) + .map(|ack| PauseAckDto { + generation: ack.generation, + sink_was_active: ack.sink_was_active, + latched: ack.latched, + }) + .map_err(|_| "silence acknowledgement timed out".to_string()) + }) + .await + .map_err(|e| e.to_string())? +} + +// ------------------------------------------------------- pure helpers ----- + +fn now_epoch_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} + +/// `exp=` out of an Akamai `hdnea` token (or bare query). +fn parse_exp_ms(url: &str) -> Option { + let idx = url.find("exp=")?; + let digits: String = url[idx + 4..] + .chars() + .take_while(|c| c.is_ascii_digit()) + .collect(); + if !(9..=11).contains(&digits.len()) { + return None; + } + digits.parse::().ok().map(|s| s * 1000) +} + +/// Mirror of `computeSafeForMs` in src/lib/preview/lease.js — see that file +/// for the reasoning; the two implementations are kept behaviorally +/// identical and both covered by tests. +fn compute_safe_for_ms(exp_ms: Option, date_ms: Option, local_now_ms: i64) -> i64 { + let Some(exp) = exp_ms else { + return FALLBACK_SAFE_MS; + }; + if exp <= 0 { + return FALLBACK_SAFE_MS; + } + let mut remaining: Option = None; + if let Some(date) = date_ms { + if date > 0 { + let r = exp - date; + if r <= 0 || r > 2 * MAX_SAFE_MS { + return FALLBACK_SAFE_MS; + } + remaining = Some(r); + } + } + let local_view = exp - local_now_ms; + if local_view <= 0 && remaining.is_none() { + return FALLBACK_SAFE_MS; + } + let usable = match remaining { + None => local_view, + Some(r) => { + if local_view > 0 { + r.min(local_view) + } else { + r + } + } + }; + (usable.min(MAX_SAFE_MS) - SAFETY_MARGIN_MS).max(0) +} + +/// Skip a leading ID3v2 tag. Every Phase 0 sample carries a 10-byte v2.4 +/// shell with tag size 0, but the parser handles real non-zero tags too — +/// the syncsafe size field is authoritative. +fn strip_id3v2(bytes: &[u8]) -> &[u8] { + if bytes.len() < 10 || &bytes[0..3] != b"ID3" { + return bytes; + } + let size = ((bytes[6] & 0x7f) as usize) << 21 + | ((bytes[7] & 0x7f) as usize) << 14 + | ((bytes[8] & 0x7f) as usize) << 7 + | (bytes[9] & 0x7f) as usize; + let end = 10usize.saturating_add(size); + if end >= bytes.len() { + return bytes; // tag claims to swallow the file: leave it alone + } + &bytes[end..] +} + +fn starts_with_mpeg_sync(bytes: &[u8]) -> bool { + bytes.len() >= 2 && bytes[0] == 0xFF && (bytes[1] & 0xE0) == 0xE0 +} + +// ------------------------------------------------------------------ tests - + +#[cfg(test)] +mod tests { + use super::*; + + const MIN: i64 = 60_000; + const NOW: i64 = 1_784_760_000_000; + + #[test] + fn safe_for_ms_valid_uses_provider_clock() { + assert_eq!( + compute_safe_for_ms(Some(NOW + 14 * MIN), Some(NOW), NOW), + 14 * MIN - SAFETY_MARGIN_MS + ); + } + + #[test] + fn safe_for_ms_local_clock_ahead_shrinks() { + assert_eq!( + compute_safe_for_ms(Some(NOW + 14 * MIN), Some(NOW), NOW + 10 * MIN), + 4 * MIN - SAFETY_MARGIN_MS + ); + } + + #[test] + fn safe_for_ms_local_clock_behind_cannot_extend() { + assert_eq!( + compute_safe_for_ms(Some(NOW + 14 * MIN), Some(NOW), NOW - 60 * MIN), + 14 * MIN - SAFETY_MARGIN_MS + ); + } + + #[test] + fn safe_for_ms_missing_or_contradictory_falls_back() { + assert_eq!(compute_safe_for_ms(None, Some(NOW), NOW), FALLBACK_SAFE_MS); + assert_eq!( + compute_safe_for_ms(Some(NOW - MIN), Some(NOW), NOW), + FALLBACK_SAFE_MS + ); + assert_eq!( + compute_safe_for_ms(Some(NOW + 6 * 60 * MIN), Some(NOW), NOW), + FALLBACK_SAFE_MS + ); + } + + #[test] + fn exp_parses_from_hdnea_token() { + let url = "https://cdnt-preview.dzcdn.net/api/x.mp3?hdnea=exp=1784761537~acl=/x*~hmac=aa"; + assert_eq!(parse_exp_ms(url), Some(1_784_761_537_000)); + assert_eq!(parse_exp_ms("https://h/x"), None); + assert_eq!(parse_exp_ms("https://h/x?exp=12"), None); // too short + } + + #[test] + fn id3_empty_shell_is_skipped() { + // The exact Phase 0 shape: v2.4 header, flags 0, size 0, then sync. + let mut f = b"ID3\x04\x00\x00\x00\x00\x00\x00".to_vec(); + f.extend_from_slice(&[0xFF, 0xFB, 0x92, 0x64, 1, 2, 3]); + let out = strip_id3v2(&f); + assert!(starts_with_mpeg_sync(out)); + assert_eq!(out.len(), 7); + } + + #[test] + fn id3_nonzero_tag_is_skipped_by_syncsafe_size() { + let mut f = b"ID3\x03\x00\x00\x00\x00\x00\x0A".to_vec(); // size 10 + f.extend_from_slice(&[0u8; 10]); // tag body + f.extend_from_slice(&[0xFF, 0xFB, 0x00, 0x00]); + assert!(starts_with_mpeg_sync(strip_id3v2(&f))); + } + + #[test] + fn id3_overlong_tag_leaves_bytes_untouched() { + let f = b"ID3\x04\x00\x00\x7f\x7f\x7f\x7f rest".to_vec(); + assert_eq!(strip_id3v2(&f), &f[..]); + } + + #[test] + fn plain_frames_pass_through() { + let f = [0xFF, 0xFB, 0x92, 0x64]; + assert_eq!(strip_id3v2(&f), &f[..]); + assert!(starts_with_mpeg_sync(&f)); + assert!(!starts_with_mpeg_sync(b"")); + } +} diff --git a/src/lib/api.js b/src/lib/api.js index 20fdc85..d09ba64 100644 --- a/src/lib/api.js +++ b/src/lib/api.js @@ -132,6 +132,19 @@ export const preloadCachedStream = (username, path, size, duration) => export const streamWaveform = (username, path, size, buckets) => invoke("stream_waveform", { username, path, size, buckets }); +// Preview tracks (scrollable-track-previews): a fully separate audio path. +// previewResolve returns { ok, lease | failure, retryable } — the signed URL +// never crosses this boundary, only an opaque lease id does. +export const previewResolve = (trackId) => invoke("preview_resolve", { trackId }); +export const previewLoad = (leaseId, startPaused) => + invoke("preview_load", { leaseId, startPaused }); +export const previewControl = (action, value = null) => + invoke("preview_control", { action, value }); +export const previewDispose = () => invoke("preview_dispose"); +// Main-player silence acknowledgement (audio-focus contract). Throws on +// timeout — callers keep preview parked; they never assume silence. +export const playerPauseAcked = () => invoke("player_pause_acked"); + // Window chrome (the app draws its own titlebar; decorations are off). import { getCurrentWindow } from "@tauri-apps/api/window"; export const win = { diff --git a/src/lib/preview/focusBridge.js b/src/lib/preview/focusBridge.js new file mode 100644 index 0000000..30aa1d2 --- /dev/null +++ b/src/lib/preview/focusBridge.js @@ -0,0 +1,71 @@ +// Production wiring for the Phase 1 focus coordinator: the real main player +// on one side, the preview session on the other. +// +// One honest deviation from the pure model, documented here because it is a +// real design decision: the renderer cannot know the NATIVE playback +// generation ahead of time (it only travels in the PauseAck itself). The +// production binding therefore replaces "ack.generation === snapshot's +// native generation" with an equivalent renderer-side invariant: +// +// granted ⇔ ack received (never assumed on timeout) +// ∧ no renderer PLAYBACK generation change during negotiation +// (a new Buffering = a new track start = a new generation) +// ∧ no revision-gateway change during negotiation +// +// Every native source change reaches the renderer as an event that bumps the +// gateway (wired in state.svelte.js), so a native-side surprise invalidates +// the grant exactly as a user command does. The native ack generation is +// kept on the token for diagnostics. + +import * as api from "../api.js"; +import { createFocusCoordinator, createRevisionGateway } from "./focus.js"; + +/** + * @param {object} deps + * @param {() => {status: string, track: object|null, position: number}} deps.readPlayer + * Reads the live main-player store (state.svelte.js `player`). + * @param {() => number} deps.readPlaybackGeneration + * Renderer-side monotonic counter, bumped on every main "buffering". + * @param {(snapshot: object) => void} deps.resumeMain + * Ordinary resume path, invoked focus-internally. + */ +export function createProductionFocus({ readPlayer, readPlaybackGeneration, resumeMain }) { + const gateway = createRevisionGateway(); + + const coordinator = createFocusCoordinator({ + gateway, + mainStatus: () => { + const p = readPlayer(); + return { + status: p.status, + trackKey: p.track ? `${p.track.source}:${p.track.sub ?? ""}:${p.track.title ?? ""}` : null, + positionMs: Math.round((p.position ?? 0) * 1000), + // Audible covers buffering too: a decode in flight will BECOME + // audible, which is exactly the auto-start window the + // desired-paused latch exists to close. + audible: p.status === "playing" || p.status === "buffering", + playbackGeneration: readPlaybackGeneration(), + }; + }, + requestSilence: async (generationAtSnapshot) => { + const ack = await api.playerPauseAcked(); // throws on timeout + const generationNow = readPlaybackGeneration(); + return { + // Applied only if no new main track started mid-negotiation; + // the returned generation makes the coordinator's equality + // check express exactly that. + applied: true, + generation: + generationNow === generationAtSnapshot ? generationAtSnapshot : generationNow, + nativeGeneration: ack.generation, + latched: ack.latched, + }; + }, + resumeMain: (snapshot, opts) => { + gateway.userPlay(opts); // focus-internal: bumps nothing + resumeMain(snapshot); + }, + }); + + return { coordinator, gateway }; +} diff --git a/src/lib/preview/nativeTransport.js b/src/lib/preview/nativeTransport.js new file mode 100644 index 0000000..c7b38e1 --- /dev/null +++ b/src/lib/preview/nativeTransport.js @@ -0,0 +1,108 @@ +// Native preview transport — the production implementation of the transport +// contract in transport.js, over the Tauri preview backend. +// +// The backend's event stream (`soulstream:preview`) knows nothing about +// sessions or generations; THIS wrapper is what makes those events safe. It +// stamps every incoming event with the {sessionId, generation, sessionKey} +// of the load that is CURRENTLY active on this side of the boundary, and the +// reducer's guard does the rest: after a step, events from the superseded +// clip carry the old generation and land inert. stop()/dispose() clear the +// context, so a straggler event after teardown dispatches nothing at all. + +import * as api from "../api.js"; +import { isValidLease } from "./lease.js"; + +export function createNativeTransport(dispatch) { + let ctx = null; // { sessionId, generation, sessionKey } + let unlisten = null; + let disposed = false; + + const stamp = (type, extra = {}) => { + if (!ctx) return; + dispatch({ type, ...ctx, ...extra }); + }; + + async function ensureSubscribed() { + if (unlisten || disposed) return; + unlisten = await api.onEvent("soulstream:preview", (ev) => { + switch (ev.kind) { + case "playing": + stamp("MEDIA_PLAYING"); + break; + case "paused": + stamp("MEDIA_PAUSED"); + break; + case "position": + stamp("MEDIA_POSITION", { positionMs: Math.round(ev.secs * 1000) }); + break; + case "ended": + stamp("MEDIA_ENDED"); + break; + case "error": + stamp("ITEM_FAILED", { + sessionKey: ctx?.sessionKey, + errorClass: "transport", + }); + break; + // "buffering"/"stopped" carry no session-level meaning: the + // reducer derives buffering from LEASE_READY + desiredPlaying, + // and stops are always consequences of our own commands. + default: + break; + } + }); + } + + return { + /** Accept a native lease and hand its clip to the preview player. + * `startPaused` maps to the backend's desired-paused latch, so a + * parked open can never leak audio (Phase 0.4). */ + async load(lease, { sessionId, generation, sessionKey }, { startPaused = false } = {}) { + if (disposed) throw new Error("transport disposed"); + await ensureSubscribed(); + if (!isValidLease(lease) || lease.transport !== "native") { + ctx = { sessionId, generation, sessionKey }; + stamp("ITEM_FAILED", { sessionKey, errorClass: "invalid_media" }); + ctx = null; + return; + } + ctx = { sessionId, generation, sessionKey }; + await api.previewLoad(lease.opaqueLeaseId, startPaused); + }, + async play() { + if (!ctx) return; + await api.previewControl("resume"); + }, + async pause() { + if (!ctx) return; + await api.previewControl("pause"); + }, + async seek(positionMs) { + if (!ctx) return; + await api.previewControl("seek", positionMs / 1000); + }, + async setVolume(v) { + await api.previewControl("volume", v); + }, + async stop() { + ctx = null; // events from the dying clip stamp nothing + await api.previewControl("stop"); + }, + /** Idempotent teardown: silences the backend, deletes clip files, + * detaches the event listener. Close, navigation and hot reload all + * funnel here. */ + async dispose() { + if (disposed) return; + disposed = true; + ctx = null; + try { + await api.previewDispose(); + } finally { + if (unlisten) { + unlisten(); + unlisten = null; + } + } + }, + }; +} diff --git a/src/lib/preview/productionResolver.js b/src/lib/preview/productionResolver.js new file mode 100644 index 0000000..dfecc9e --- /dev/null +++ b/src/lib/preview/productionResolver.js @@ -0,0 +1,51 @@ +// Production PreviewResolver — the renderer half of the native resolver. +// +// Same contract as createFixtureResolver: stable source ref in, discriminated +// result out. The heavy lifting (fresh mint, host policy, bounded fetch, MIME +// gate, ID3 strip, safeForMs) happens in Rust; this wrapper only reshapes the +// DTO into a MediaLease and keeps the failure classes intact. `bypassCache` +// is accepted for contract parity — the backend mints fresh on every call by +// design, so the flag only matters to the caller's single-retry accounting. + +import * as api from "../api.js"; +import { makeNativeLease } from "./lease.js"; + +export function createProductionResolver() { + return { + async resolve(sourceRef, { generation } = {}) { + if (sourceRef.provider !== "deezer") { + // Apple identity plumbing is the Phase 2 backend extension + // tracked in the plan; until it lands, non-Deezer sources + // classify as permanently unresolvable rather than guessing. + return { ok: false, failure: "not_found", generation, retryable: false }; + } + let dto; + try { + dto = await api.previewResolve(Number(sourceRef.providerTrackId)); + } catch (e) { + return { ok: false, failure: "network", generation, retryable: true }; + } + if (!dto?.ok || !dto.lease) { + return { + ok: false, + failure: dto?.failure ?? "temporarily_failed", + generation, + retryable: !!dto?.retryable, + }; + } + return { + ok: true, + generation, + lease: makeNativeLease({ + source: sourceRef, + opaqueLeaseId: dto.lease.opaqueLeaseId, + safeForMs: dto.lease.safeForMs, + mimeType: dto.lease.mimeType, + clipDurationMs: dto.lease.clipDurationMs, + byteLength: dto.lease.byteLength, + supportsRanges: true, + }), + }; + }, + }; +} diff --git a/src/lib/preview/service.js b/src/lib/preview/service.js new file mode 100644 index 0000000..a3caf42 --- /dev/null +++ b/src/lib/preview/service.js @@ -0,0 +1,282 @@ +// Preview session service — the effectful orchestrator over the pure parts. +// +// The reducer decides WHAT happens; this service decides WHEN: the settle +// window after navigation, the resolve→load→play pipeline, the single +// cache-bypass retry, one direction-aware lease prefetch, and the focus +// handshake at the session edges. Everything time-like is injected +// (scheduler, monotonic clock) so the whole loop runs under node:test with +// manual ticks — no real timers anywhere in the tests. +// +// Staleness discipline: every async continuation re-checks BOTH the session +// id and the request generation before dispatching or touching the +// transport. The reducer would drop a stale dispatch anyway; checking here +// too keeps the transport from even being commanded on behalf of a dead +// generation (the "transport boundary" half of the Phase 2 exit gate). + +import { reduce, createInitialState, playableManifestIndices, playableCursor, PHASE } from "./session.js"; + +const defaultScheduler = { + set: (fn, ms) => setTimeout(fn, ms), + clear: (id) => clearTimeout(id), +}; + +export function createPreviewService({ + resolver, + // The service CONSTRUCTS the transport, binding it to its own dispatch — + // the transport's events (MEDIA_PLAYING, MEDIA_ENDED, …) flow through the + // same generation-guarded reducer as everything else. Passing an already- + // built transport is how the original draft silently lost that wiring. + transportFactory, + focus, // { coordinator } from focusBridge, or a test double + settleMs = 150, + scheduler = defaultScheduler, + monotonicNow = () => (globalThis.performance ?? Date).now(), +}) { + let state = null; + let listeners = new Set(); + let settleTimer = null; + let focusToken = null; + let focusWaiting = false; + // sessionKey -> { lease, receiptAt } — the one direction-aware prefetch. + let prefetched = new Map(); + + const emit = () => { + for (const l of listeners) l(state, { focusWaiting }); + }; + const dispatch = (action) => { + if (!state) return; + const before = state; + state = reduce(state, action); + if (state !== before) { + afterChange(before); + emit(); + } + }; + const transport = transportFactory(dispatch); + + // ------------------------------------------------------ media pipeline -- + + function afterChange(before) { + if (!state.manifest) return; + const generationMoved = state.requestGeneration !== before.requestGeneration; + if (generationMoved && state.phase === PHASE.RESOLVING_MEDIA) { + scheduleSettle(); + } + } + + function scheduleSettle() { + if (settleTimer != null) scheduler.clear(settleTimer); + const sessionId = state.sessionId; + const generation = state.requestGeneration; + // Stop the outgoing clip immediately — silence between cards is + // correct; audio under the wrong artwork never is. + void transport.stop(); + settleTimer = scheduler.set(() => { + settleTimer = null; + if (!state || state.sessionId !== sessionId || state.requestGeneration !== generation) return; + void runPipeline(sessionId, generation); + }, settleMs); + } + + async function runPipeline(sessionId, generation) { + const fresh = () => + state && state.sessionId === sessionId && state.requestGeneration === generation; + if (!fresh() || state.activeItemKey == null) return; + const item = state.manifest.items[state.activeManifestIndex]; + const source = item.sources[0]; + if (!source) { + dispatch({ type: "ITEM_FAILED", sessionId, generation, sessionKey: item.sessionKey, errorClass: "no_source" }); + return; + } + dispatch({ type: "RESOLVE_STARTED", sessionId, generation, sessionKey: item.sessionKey }); + + // Prefetched lease still inside its safe window → skip the resolve. + let lease = null; + const pre = prefetched.get(item.sessionKey); + if (pre && monotonicNow() - pre.receiptAt < pre.lease.safeForMs * 0.8) { + lease = pre.lease; + } + prefetched.delete(item.sessionKey); + + if (!lease) { + let result = await resolver.resolve(source, { generation }); + if (!fresh()) return; + if (!result.ok && result.retryable) { + // The capped retry: exactly one cache-bypass re-mint per + // user intent, per the resolver contract. Never a loop. + result = await resolver.resolve(source, { generation, bypassCache: true }); + if (!fresh()) return; + } + if (!result.ok) { + dispatch({ + type: "ITEM_FAILED", + sessionId, + generation, + sessionKey: item.sessionKey, + errorClass: result.failure, + availability: result.failure === "not_provided" ? "not_provided" : undefined, + }); + return; + } + lease = result.lease; + } + + dispatch({ + type: "LEASE_READY", + sessionId, + generation, + sessionKey: item.sessionKey, + clipDurationMs: lease.clipDurationMs, + }); + if (!fresh()) return; + + const audible = state.desiredPlaying && focusToken != null && !state.muted; + await transport.load(lease, { sessionId, generation, sessionKey: item.sessionKey }, { startPaused: !audible }); + if (!fresh()) { + void transport.stop(); + return; + } + if (audible) await transport.play(); + + prefetchNeighbor(sessionId, generation); + } + + /** Resolve the NEXT playable item's lease in the background so stepping + * forward usually skips the network entirely. One item, forward only — + * the bounded mechanism the plan allows without byte prefetch. */ + function prefetchNeighbor(sessionId, generation) { + if (!state || state.sessionId !== sessionId || state.requestGeneration !== generation) return; + const playable = playableManifestIndices(state); + const cur = playableCursor(state); + if (cur == null || cur + 1 >= playable.length) return; + const next = state.manifest.items[playable[cur + 1]]; + if (prefetched.has(next.sessionKey) || !next.sources[0]) return; + void resolver + .resolve(next.sources[0], { generation }) + .then((r) => { + if (r.ok && state && state.sessionId === sessionId) { + prefetched.set(next.sessionKey, { lease: r.lease, receiptAt: monotonicNow() }); + // bounded: keep at most one prefetched lease around + if (prefetched.size > 1) { + const oldest = prefetched.keys().next().value; + if (oldest !== next.sessionKey) prefetched.delete(oldest); + } + } + }) + .catch(() => {}); + } + + // -------------------------------------------------------------- focus -- + + async function ensureFocus() { + if (focusToken) return true; + focusWaiting = true; + emit(); + try { + const { granted, token } = await focus.coordinator.acquirePreview(); + focusToken = granted ? token : null; + return granted; + } catch { + focusToken = null; + return false; + } finally { + focusWaiting = false; + emit(); + } + } + + // ----------------------------------------------------------- public ---- + + return { + subscribe(fn) { + listeners.add(fn); + if (state) fn(state, { focusWaiting }); + return () => listeners.delete(fn); + }, + getState: () => state, + isFocusWaiting: () => focusWaiting, + + /** Open a session: focus first (take-audio-on-open), then manifest. + * If the acknowledgement fails, the session still opens — parked, + * with the transient wait surfaced — and Play retries acquisition. */ + async open({ sessionId, manifest, startAtManifestIndex }) { + state = createInitialState(sessionId); + emit(); + const granted = await ensureFocus(); + dispatch({ type: "MANIFEST_READY", sessionId, manifest, startAtManifestIndex }); + if (!granted && state.phase !== PHASE.EMPTY) { + dispatch({ type: "SET_DESIRED_PLAYING", sessionId, value: false }); + } + }, + + step: (dir) => dispatch({ type: "STEP", sessionId: state?.sessionId, dir }), + activateKey: (sessionKey) => dispatch({ type: "ACTIVATE_KEY", sessionId: state?.sessionId, sessionKey }), + replay: () => dispatch({ type: "REPLAY", sessionId: state?.sessionId }), + seek: (positionMs) => { + if (!state) return; + void transport.seek?.(positionMs); + dispatch({ + type: "MEDIA_POSITION", + sessionId: state.sessionId, + generation: state.requestGeneration, + positionMs, + }); + }, + setMuted(value) { + if (!state) return; + dispatch({ type: "SET_MUTED", sessionId: state.sessionId, value }); + if (value && state.phase === PHASE.PLAYING) void transport.pause(); + }, + + async togglePlay() { + if (!state) return; + const sid = state.sessionId; + if (state.phase === PHASE.PLAYING) { + dispatch({ type: "SET_DESIRED_PLAYING", sessionId: sid, value: false }); + await transport.pause(); + return; + } + if (state.phase === PHASE.ENDED) { + dispatch({ type: "REPLAY", sessionId: sid }); + return; + } + dispatch({ type: "SET_DESIRED_PLAYING", sessionId: sid, value: true }); + const granted = await ensureFocus(); + if (!granted) return; // parked; wait state already surfaced + if (state.phase === PHASE.READY || state.phase === PHASE.PAUSED) { + await transport.play(); + } + // RESOLVING/BUFFERING: the pipeline observes desiredPlaying when + // the lease lands; nothing else to do here. + }, + + /** Play full track: hand ownership to ordinary playback. The old + * main snapshot is intentionally discarded (never restored). */ + async playFullTrack(startMainPlayback) { + if (!state) return; + await transport.stop(); + focus.coordinator.handoffToMain(); + focusToken = null; + dispatch({ type: "SET_DESIRED_PLAYING", sessionId: state.sessionId, value: false }); + await startMainPlayback(state.manifest.items[state.activeManifestIndex]); + }, + + /** Close: terminal for the session; transport torn down; main + * restored only if we paused it and nothing changed since. */ + async close() { + if (!state) return { restored: false }; + if (settleTimer != null) { + scheduler.clear(settleTimer); + settleTimer = null; + } + dispatch({ type: "CLOSE", sessionId: state.sessionId }); + prefetched.clear(); + await transport.stop(); + await transport.dispose(); + const token = focusToken; + focusToken = null; + const restored = token ? focus.coordinator.release(token) : { restored: false }; + return restored; + }, + }; +} diff --git a/tests/preview/service.test.mjs b/tests/preview/service.test.mjs new file mode 100644 index 0000000..c873cac --- /dev/null +++ b/tests/preview/service.test.mjs @@ -0,0 +1,302 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; + +import { AVAILABILITY } from "../../src/lib/preview/contracts.js"; +import { buildFixtureManifest } from "../../src/lib/preview/manifest.js"; +import { createFixtureResolver } from "../../src/lib/preview/resolver.js"; +import { createPreviewService } from "../../src/lib/preview/service.js"; +import { PHASE } from "../../src/lib/preview/session.js"; + +const T0 = 1_700_000_000_000; +const SID = "svc-1"; + +function manifest(n = 3) { + return buildFixtureManifest({ + rows: Array.from({ length: n }, (_, i) => ({ + providerTrackId: String(i + 1), + title: `T${i + 1}`, + initialAvailability: AVAILABILITY.AVAILABLE, + })), + createdAtMs: T0, + }); +} + +/** Manual scheduler: settle windows fire only when the test says so. */ +function makeScheduler() { + let queue = new Map(); + let nextId = 1; + return { + set(fn, ms) { + const id = nextId++; + queue.set(id, fn); + return id; + }, + clear(id) { + queue.delete(id); + }, + async flush() { + const fns = [...queue.values()]; + queue.clear(); + for (const fn of fns) await fn(); + await drainMicrotasks(); + }, + pending: () => queue.size, + }; +} +const drainMicrotasks = () => new Promise((r) => setTimeout(r, 0)); + +/** Transport double: records every command AND behaves like the real one — + * play/pause dispatch their media events stamped with the load's context, + * which is exactly the wiring the service's transportFactory provides. */ +function makeTransport() { + const log = []; + let dispatch = null; + let ctx = null; + const t = { + log, + async load(lease, c, opts) { + ctx = c; + log.push(["load", lease.opaqueLeaseId, c.generation, !!opts?.startPaused]); + }, + async play() { + log.push(["play"]); + if (ctx) dispatch({ type: "MEDIA_PLAYING", ...ctx }); + }, + async pause() { + log.push(["pause"]); + if (ctx) dispatch({ type: "MEDIA_PAUSED", ...ctx }); + }, + async seek(ms) { + log.push(["seek", ms]); + }, + async stop() { + ctx = null; + log.push(["stop"]); + }, + async dispose() { + log.push(["dispose"]); + }, + end() { + if (ctx) dispatch({ type: "MEDIA_ENDED", ...ctx }); + }, + }; + return { + transport: t, + factory: (d) => { + dispatch = d; + return t; + }, + }; +} + +function makeFocus({ grant = true, pausedByPreview = true } = {}) { + const calls = { acquire: 0, release: 0, handoff: 0 }; + let live = null; + return { + calls, + released: [], + coordinator: { + async acquirePreview() { + calls.acquire += 1; + if (!grant) return { granted: false, token: null }; + live = { id: `t${calls.acquire}`, pausedByPreview }; + return { granted: true, token: live }; + }, + release(token) { + calls.release += 1; + const ok = token === live; + live = null; + return { restored: ok && pausedByPreview }; + }, + handoffToMain() { + calls.handoff += 1; + live = null; + return { restored: false, reason: "handoff" }; + }, + }, + }; +} + +function makeService({ plan, grant = true, n = 3 } = {}) { + const scheduler = makeScheduler(); + const { transport, factory } = makeTransport(); + const focus = makeFocus({ grant }); + let clock = 0; + const resolver = createFixtureResolver( + plan ?? { 1: { opaqueLeaseId: "op-1" }, 2: { opaqueLeaseId: "op-2" }, 3: { opaqueLeaseId: "op-3" } } + ); + const service = createPreviewService({ + resolver, + transportFactory: factory, + focus, + scheduler, + monotonicNow: () => clock, + }); + return { service, scheduler, transport, focus, resolver, advance: (ms) => (clock += ms), manifest: manifest(n) }; +} + +// ------------------------------------------------------------------ open -- + +test("open: focus, settle, resolve, load audible, play", async () => { + const { service, scheduler, transport, focus, manifest } = makeService(); + await service.open({ sessionId: SID, manifest }); + assert.equal(focus.calls.acquire, 1); + assert.equal(service.getState().phase, PHASE.RESOLVING_MEDIA); + assert.equal(scheduler.pending(), 1); // settle armed + + await scheduler.flush(); + const st = service.getState(); + assert.equal(st.phase, PHASE.PLAYING); // lease landed, played, confirmed + const load = transport.log.find((e) => e[0] === "load"); + assert.deepEqual(load, ["load", "op-1", st.requestGeneration, false]); // audible → not parked + assert.ok(transport.log.some((e) => e[0] === "play")); +}); + +test("open with focus denied: session opens PARKED, wait state surfaced", async () => { + const { service, scheduler, transport, manifest } = makeService({ grant: false }); + await service.open({ sessionId: SID, manifest }); + assert.equal(service.getState().desiredPlaying, false); + await scheduler.flush(); + const load = transport.log.find((e) => e[0] === "load"); + assert.equal(load[3], true, "must load parked without focus"); + assert.ok(!transport.log.some((e) => e[0] === "play"), "no audio without focus"); +}); + +// ------------------------------------------------------------- stepping --- + +test("rapid steps: only the LAST generation ever reaches the transport", async () => { + const { service, scheduler, transport, manifest } = makeService(); + await service.open({ sessionId: SID, manifest }); + service.step(1); // supersedes the open's pipeline before its settle fires + service.step(1); + const finalGen = service.getState().requestGeneration; + await scheduler.flush(); + await scheduler.flush(); + const loads = transport.log.filter((e) => e[0] === "load"); + assert.equal(loads.length, 1, "superseded generations never loaded"); + assert.equal(loads[0][2], finalGen); + assert.equal(loads[0][1], "op-3"); // stepped 1 → 2 → 3 +}); + +test("failed resolve retries exactly once, then fails the item in place", async () => { + const { service, scheduler, resolver, manifest } = makeService({ + plan: { + 1: (n) => (n === 1 ? { failure: "stale_lease", retryable: true } : { opaqueLeaseId: "op-fresh" }), + 2: { failure: "network", retryable: true }, // fails on retry too + 3: { opaqueLeaseId: "op-3" }, + }, + }); + await service.open({ sessionId: SID, manifest }); + await scheduler.flush(); + await drainMicrotasks(); // background neighbor prefetch (item 2) settles + assert.equal(service.getState().phase, PHASE.PLAYING); // retry rescued item 1 + assert.equal(resolver.callCount("1"), 2); + assert.equal(resolver.callCount("2"), 1, "prefetch tried item 2 once — prefetch never retries"); + + service.step(1); + await scheduler.flush(); + const st = service.getState(); + assert.equal(st.phase, PHASE.ITEM_FAILED); + assert.equal(st.errorClass, "network"); + // 1 prefetch + 1 pipeline resolve + exactly 1 capped retry = 3, then stop + assert.equal(resolver.callCount("2"), 3, "one capped retry in the pipeline, never a loop"); + // the failed card stays; stepping onward still works + service.step(1); + await scheduler.flush(); + assert.equal(service.getState().phase, PHASE.PLAYING); + assert.equal(resolver.callCount("2"), 3, "failure is settled — no more calls"); +}); + +// ------------------------------------------------------------- prefetch --- + +test("neighbor lease prefetch: stepping forward skips the resolve", async () => { + const { service, scheduler, resolver, manifest } = makeService(); + await service.open({ sessionId: SID, manifest }); + await scheduler.flush(); // pipeline + prefetch of item 2 kicked + await drainMicrotasks(); + assert.equal(resolver.callCount("2"), 1, "prefetched in the background"); + service.step(1); + await scheduler.flush(); + assert.equal(resolver.callCount("2"), 1, "no second resolve on activation"); + assert.equal(service.getState().phase, PHASE.PLAYING); +}); + +test("a prefetched lease past 80% of its safe window is not trusted", async () => { + const { service, scheduler, resolver, advance, manifest } = makeService({ + plan: { + 1: { opaqueLeaseId: "op-1", safeForMs: 10_000 }, + 2: { opaqueLeaseId: "op-2", safeForMs: 10_000 }, + 3: { opaqueLeaseId: "op-3" }, + }, + }); + await service.open({ sessionId: SID, manifest }); + await scheduler.flush(); + await drainMicrotasks(); + advance(9_000); // beyond 80% of the 10s safe window + service.step(1); + await scheduler.flush(); + assert.equal(resolver.callCount("2"), 2, "stale prefetch re-resolved"); +}); + +// ------------------------------------------------------ close/handoff ----- + +test("close during resolve: terminal, transport torn down, focus released once", async () => { + const { service, scheduler, transport, focus, manifest } = makeService(); + await service.open({ sessionId: SID, manifest }); + // close BEFORE the settle window fires — mid-"resolving" phase + const out = await service.close(); + assert.equal(service.getState().phase, PHASE.CLOSING); + assert.equal(scheduler.pending(), 0, "settle timer cancelled"); + assert.ok(transport.log.some((e) => e[0] === "dispose")); + assert.equal(focus.calls.release, 1); + assert.ok(out.restored); + // straggler flush does nothing + await scheduler.flush(); + assert.ok(!transport.log.some((e) => e[0] === "load")); +}); + +test("play-full handoff: transport stopped, no restoration, main starts with the active item", async () => { + const { service, scheduler, transport, focus, manifest } = makeService(); + await service.open({ sessionId: SID, manifest }); + await scheduler.flush(); + let started = null; + await service.playFullTrack(async (item) => { + started = item; + }); + assert.equal(focus.calls.handoff, 1); + assert.equal(started.title, "T1"); + assert.ok(transport.log.some((e) => e[0] === "stop")); + // closing afterwards must NOT restore the pre-preview main track + const out = await service.close(); + assert.equal(out.restored, false); +}); + +// -------------------------------------------------------------- controls -- + +test("togglePlay: pause from playing, resume from paused, replay from ended", async () => { + const { service, scheduler, transport, manifest } = makeService(); + await service.open({ sessionId: SID, manifest }); + await scheduler.flush(); + assert.equal(service.getState().phase, PHASE.PLAYING); + + await service.togglePlay(); + assert.equal(service.getState().phase, PHASE.PAUSED); + assert.equal(service.getState().desiredPlaying, false); + assert.ok(transport.log.some((e) => e[0] === "pause")); + + await service.togglePlay(); + assert.equal(service.getState().phase, PHASE.PLAYING); + assert.equal(service.getState().desiredPlaying, true); + + // ride to the LAST playable item, end its clip, and replay from ENDED + service.step(1); + await scheduler.flush(); + service.step(1); + await scheduler.flush(); + assert.equal(service.getState().phase, PHASE.PLAYING); + transport.end(); + assert.equal(service.getState().phase, PHASE.ENDED); + await service.togglePlay(); // replay + assert.equal(service.getState().phase, PHASE.RESOLVING_MEDIA); + assert.equal(service.getState().desiredPlaying, true); +});