From d3b0ed8a1d8794e3bd13acc9a521d99233c9118a Mon Sep 17 00:00:00 2001 From: archipelago Date: Tue, 4 Aug 2026 11:09:18 -0400 Subject: [PATCH] feat(13-07): music.* RPC surface behind one dispatcher arm MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - api/rpc/music.rs: handle_music prefix sub-dispatcher (assistant_chat.rs shape) with list-albums / list-artists / list-tracks / status / reindex - one guarded dispatcher.rs arm for the whole music. prefix, adjacent to the content.* block; the only registration point for the surface - list-tracks: optional album_id filter, limit/offset pagination, limit clamped to [1,500] (default 100) — out-of-range degrades, never errors (T-13-41) - reindex spawns the scan and returns immediately; second call while one runs reports already-running with the last stats; optional incremental:true routes to refresh_incremental so a changed library is reflected without a full re-extraction - newer-schema index served as an empty library, never overwritten or reinterpreted by readers (13-MUSIC-MODEL.md downgrade contract) - nothing music.* in UNAUTHENTICATED_METHODS — the surface rides the session/CSRF/RBAC gate; asserted by music_methods_require_session (T-13-40) - 7 tests, one per Task 2 behavior bullet plus the incremental mode Co-Authored-By: Claude Fable 5 --- core/archipelago/src/api/rpc/dispatcher.rs | 7 + core/archipelago/src/api/rpc/mod.rs | 1 + core/archipelago/src/api/rpc/music.rs | 521 +++++++++++++++++++++ 3 files changed, 529 insertions(+) create mode 100644 core/archipelago/src/api/rpc/music.rs diff --git a/core/archipelago/src/api/rpc/dispatcher.rs b/core/archipelago/src/api/rpc/dispatcher.rs index d76b859c..817c8514 100644 --- a/core/archipelago/src/api/rpc/dispatcher.rs +++ b/core/archipelago/src/api/rpc/dispatcher.rs @@ -324,6 +324,13 @@ impl RpcHandler { } "content.preview-peer" => self.handle_content_preview_peer(params).await, + // Music library (13-07, D-13): the whole `music.*` surface + // lives in music.rs, not as new arms here — this single + // guarded arm is the ONLY dispatcher.rs registration point + // for it, mirroring the `assistant.` arm below. Adjacent to + // the content.* block so the media surfaces read together. + m if m.starts_with("music.") => self.handle_music(m, params).await, + // DWN (Decentralized Web Node) "dwn.status" => self.handle_dwn_status().await, "dwn.sync" => self.handle_dwn_sync().await, diff --git a/core/archipelago/src/api/rpc/mod.rs b/core/archipelago/src/api/rpc/mod.rs index 32508161..276e8fe0 100644 --- a/core/archipelago/src/api/rpc/mod.rs +++ b/core/archipelago/src/api/rpc/mod.rs @@ -22,6 +22,7 @@ mod marketplace; mod mesh; mod middleware; mod monitoring; +mod music; mod names; mod network; mod node; diff --git a/core/archipelago/src/api/rpc/music.rs b/core/archipelago/src/api/rpc/music.rs new file mode 100644 index 00000000..6d8a2ae7 --- /dev/null +++ b/core/archipelago/src/api/rpc/music.rs @@ -0,0 +1,521 @@ +//! `music.*` RPC surface (13-07 Task 2) — the library index behind +//! `SongGrid`. Mirrors `assistant_chat.rs`'s shape: every `music.*` method +//! is added inside this file's own match; `dispatcher.rs` registers exactly +//! one guarded arm for the whole `music.` prefix +//! (`grep -c 'starts_with("music.")' dispatcher.rs` == 1). +//! +//! Reached only after the session-cookie + CSRF + `role.can_access()` gate +//! in `api/rpc/mod.rs` — no bespoke auth here, and no `music.*` method is +//! ever added to `UNAUTHENTICATED_METHODS` (T-13-40, asserted by +//! `music_methods_require_session` below). + +use super::RpcHandler; +use crate::music::index::{self, IndexError, MusicIndex, ReindexOutcome, ReindexState}; +use crate::music::{media_roots, AlbumId, MUSIC_SCHEMA_VERSION}; +use anyhow::Result; +use std::path::{Path, PathBuf}; + +/// `music.list-tracks` page-size cap: a huge library cannot be pulled in +/// one response (T-13-41). An out-of-range `limit` is clamped, not +/// rejected, so a UI bug degrades to a smaller page rather than an error. +const MAX_TRACK_LIMIT: u64 = 500; +const DEFAULT_TRACK_LIMIT: u64 = 100; + +impl RpcHandler { + /// Prefix sub-dispatcher for `music.*` — the only entry point + /// `dispatcher.rs` knows about, mirroring 13-01's `assistant.` arm. + pub(in crate::api::rpc) async fn handle_music( + &self, + method: &str, + params: Option, + ) -> Result { + music_dispatch( + index::shared_state(), + &self.config.data_dir, + media_roots(&self.config), + method, + params, + ) + .await + } +} + +/// The actual dispatch, parameterized on the scan state and paths so tests +/// can drive it against tempdirs and their own (leaked) states without +/// constructing an `RpcHandler`. +async fn music_dispatch( + state: &'static ReindexState, + data_dir: &Path, + roots: Vec, + method: &str, + params: Option, +) -> Result { + match method { + "music.list-albums" => handle_music_list_albums(data_dir).await, + "music.list-artists" => handle_music_list_artists(data_dir).await, + "music.list-tracks" => handle_music_list_tracks(data_dir, params).await, + "music.status" => handle_music_status(state, data_dir).await, + "music.reindex" => handle_music_reindex(state, data_dir.to_path_buf(), roots, params).await, + other => anyhow::bail!("no such music method: {other}"), + } +} + +/// Read the index for serving. A newer-schema index is treated as absent +/// (empty library) per `13-MUSIC-MODEL.md`'s downgrade contract — readers +/// never overwrite or reinterpret it; only an explicit `music.reindex` +/// rebuilds. +fn load_for_read(data_dir: &Path) -> Result { + match index::load(data_dir) { + Ok(idx) => Ok(idx), + Err(IndexError::NewerSchema { found, supported }) => { + tracing::warn!( + "music index schema {found} > supported {supported}; \ + serving an empty library until an explicit reindex" + ); + Ok(MusicIndex::empty()) + } + Err(e) => Err(e.into()), + } +} + +/// music.list-albums — all albums in the deterministic library order, with +/// `scanned_at` and a total count. An empty library is empty arrays plus a +/// timestamp, never null and never an error. +async fn handle_music_list_albums(data_dir: &Path) -> Result { + let snapshot = load_for_read(data_dir)?.snapshot(); + let albums: Vec = snapshot + .albums + .iter() + .map(|album| { + serde_json::json!({ + "id": album.id, + "album": album.id.album, + "album_artist": album.id.album_artist, + "track_count": album.track_ids.len(), + "track_ids": album.track_ids, + }) + }) + .collect(); + Ok(serde_json::json!({ + "albums": albums, + "total": snapshot.albums.len(), + "scanned_at": snapshot.scanned_at, + })) +} + +/// music.list-artists — all artists in deterministic order. +async fn handle_music_list_artists(data_dir: &Path) -> Result { + let snapshot = load_for_read(data_dir)?.snapshot(); + let artists: Vec = snapshot + .artists + .iter() + .map(|artist| { + serde_json::json!({ + "id": artist.id, + "name": artist.id.0, + "track_count": artist.track_ids.len(), + "track_ids": artist.track_ids, + }) + }) + .collect(); + Ok(serde_json::json!({ + "artists": artists, + "total": snapshot.artists.len(), + "scanned_at": snapshot.scanned_at, + })) +} + +/// music.list-tracks — paginated tracks, optionally filtered by `album_id` +/// (`{ "album": string, "album_artist": string|null }`). `limit` defaults +/// to 100 and is clamped to [1, 500]; `total` is the filtered count before +/// pagination so the UI can page honestly. +async fn handle_music_list_tracks( + data_dir: &Path, + params: Option, +) -> Result { + let params = params.unwrap_or_default(); + let album_id: Option = match params.get("album_id") { + None | Some(serde_json::Value::Null) => None, + Some(value) => Some(serde_json::from_value(value.clone()).map_err(|_| { + anyhow::anyhow!("Invalid album_id: expected {{ album, album_artist }}") + })?), + }; + let limit = params + .get("limit") + .and_then(|v| v.as_u64()) + .unwrap_or(DEFAULT_TRACK_LIMIT) + .clamp(1, MAX_TRACK_LIMIT) as usize; + let offset = params.get("offset").and_then(|v| v.as_u64()).unwrap_or(0) as usize; + + let snapshot = load_for_read(data_dir)?.snapshot(); + let filtered: Vec<&crate::music::Track> = snapshot + .tracks + .iter() + .filter(|track| match &album_id { + None => true, + Some(id) => { + track.album.as_deref() == Some(id.album.as_str()) + && track.album_artist == id.album_artist + } + }) + .collect(); + let total = filtered.len(); + let page: Vec<&crate::music::Track> = filtered.into_iter().skip(offset).take(limit).collect(); + + Ok(serde_json::json!({ + "tracks": page, + "total": total, + "limit": limit, + "offset": offset, + "scanned_at": snapshot.scanned_at, + })) +} + +/// music.status — the last scan's stats, whether a scan is running, and +/// the schema version (both the binary's and the on-disk index's). +async fn handle_music_status(state: &ReindexState, data_dir: &Path) -> Result { + let index = load_for_read(data_dir)?; + Ok(serde_json::json!({ + "running": state.is_running(), + "last_stats": state.last_stats(), + "schema_version": MUSIC_SCHEMA_VERSION, + "index_schema_version": index.schema_version, + "scanned_at": index.scanned_at, + "track_count": index.entries.len(), + })) +} + +/// music.reindex — start a scan and return immediately; the walk runs in a +/// spawned task (a synchronous reindex would hold an RPC connection for +/// the length of a library walk) and holds no shared lock across it. +/// Params: `{ "incremental": bool }` (default false). The default is the +/// on-demand full rebuild; `incremental: true` runs the cheap +/// stat-comparison refresh so a file added/changed/removed since the last +/// scan is reflected without re-extracting the whole library. +/// A second call while one is running reports already-running with the +/// previous scan's stats instead of queueing a duplicate (T-13-41) — the +/// in-progress flag inside `index::run_scan` is the authoritative guard. +async fn handle_music_reindex( + state: &'static ReindexState, + data_dir: PathBuf, + roots: Vec, + params: Option, +) -> Result { + let incremental = params + .as_ref() + .and_then(|p| p.get("incremental")) + .and_then(|v| v.as_bool()) + .unwrap_or(false); + if state.is_running() { + return Ok(serde_json::json!({ + "started": false, + "already_running": true, + "running": true, + "stats": state.last_stats(), + })); + } + tokio::spawn(async move { + let outcome = if incremental { + index::refresh_incremental(state, &data_dir, &roots).await + } else { + index::reindex(state, &data_dir, &roots).await + }; + match outcome { + Ok(ReindexOutcome::Completed(stats)) => { + tracing::info!( + "music {} scan complete: {} scanned, {} extracted, {} skipped, {} removed in {}ms", + if incremental { "incremental" } else { "full" }, + stats.scanned, stats.extracted, stats.skipped, stats.removed, stats.elapsed_ms + ); + } + Ok(ReindexOutcome::AlreadyRunning(prev)) => { + tracing::info!( + "music reindex: a scan was already running (previous: {prev:?}); not duplicating" + ); + } + Err(e) => tracing::warn!("music reindex failed: {e:#}"), + } + }); + Ok(serde_json::json!({ + "started": true, + "already_running": false, + "running": true, + })) +} + +// --------------------------------------------------------------------- +// Tests — one per 13-07 Task 2 bullet, driving music_dispatch +// against tempdirs and leaked per-test ReindexStates (never the process- +// wide shared_state, so parallel tests can't interfere). +// --------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::music::index::{reindex, ReindexState}; + + fn leaked_state() -> &'static ReindexState { + Box::leak(Box::new(ReindexState::default())) + } + + /// Minimal tagged-FLAC fixture (same byte-level builder family as + /// music/tags.rs and music/index.rs tests — no committed binaries). + fn build_flac(comments: &[(&str, &str)]) -> Vec { + let info_bits: u32 = (44100 << 12) | (1 << 9) | (15 << 4); + let mut streaminfo = Vec::with_capacity(34); + streaminfo.extend_from_slice(&[0, 0, 0, 0]); + streaminfo.extend_from_slice(&[0, 0, 0, 0, 0, 0]); + streaminfo.extend_from_slice(&info_bits.to_be_bytes()); + streaminfo.extend_from_slice(&(44100u32 * 3).to_be_bytes()); + streaminfo.extend_from_slice(&[0u8; 16]); + + let block = |block_type: u8, is_last: bool, content: &[u8]| { + let mut b = Vec::with_capacity(4 + content.len()); + b.push(((is_last as u8) << 7) | (block_type & 0x7F)); + let len = content.len() as u32; + b.push(((len >> 16) & 0xFF) as u8); + b.push(((len >> 8) & 0xFF) as u8); + b.push((len & 0xFF) as u8); + b.extend_from_slice(content); + b + }; + + let vendor = b"test-vendor"; + let mut vorbis = Vec::new(); + vorbis.extend_from_slice(&(vendor.len() as u32).to_le_bytes()); + vorbis.extend_from_slice(vendor); + vorbis.extend_from_slice(&(comments.len() as u32).to_le_bytes()); + for (key, value) in comments { + let field = format!("{key}={value}"); + vorbis.extend_from_slice(&(field.len() as u32).to_le_bytes()); + vorbis.extend_from_slice(field.as_bytes()); + } + + let mut file = Vec::new(); + file.extend_from_slice(b"fLaC"); + file.extend_from_slice(&block(0, false, &streaminfo)); + file.extend_from_slice(&block(4, true, &vorbis)); + file + } + + fn write_track(dir: &Path, name: &str, title: &str, artist: &str, album: &str, no: u32) { + let bytes = build_flac(&[ + ("TITLE", title), + ("ARTIST", artist), + ("ALBUM", album), + ("ALBUMARTIST", artist), + ("TRACKNUMBER", &no.to_string()), + ("DISCNUMBER", "1"), + ("DATE", "2024"), + ]); + std::fs::write(dir.join(name), bytes).expect("write track fixture"); + } + + /// data_dir + media root with 2 albums / 3 tracks, already indexed. + async fn indexed_library() -> (tempfile::TempDir, tempfile::TempDir, Vec) { + let data_dir = tempfile::tempdir().unwrap(); + let media = tempfile::tempdir().unwrap(); + write_track(media.path(), "a1.flac", "Dawn", "X", "Alpha", 1); + write_track(media.path(), "a2.flac", "Noon", "X", "Alpha", 2); + write_track(media.path(), "b1.flac", "Dusk", "Y", "Beta", 1); + let roots = vec![media.path().to_path_buf()]; + reindex(&ReindexState::default(), data_dir.path(), &roots) + .await + .expect("fixture reindex succeeds"); + (data_dir, media, roots) + } + + async fn dispatch( + state: &'static ReindexState, + data_dir: &Path, + roots: &[PathBuf], + method: &str, + params: Option, + ) -> serde_json::Value { + music_dispatch(state, data_dir, roots.to_vec(), method, params) + .await + .unwrap_or_else(|e| panic!("{method} failed: {e:#}")) + } + + #[test] + fn music_methods_require_session() { + // Mirrors assistant_methods_require_session: the whole surface + // rides the normal session/CSRF/RBAC gate; nothing music.* may + // ever appear in the unauthenticated allowlist (T-13-40). + let has_music_method = crate::api::rpc::UNAUTHENTICATED_METHODS + .iter() + .any(|m| m.starts_with("music.")); + assert!( + !has_music_method, + "music.* must never be added to UNAUTHENTICATED_METHODS" + ); + } + + #[tokio::test] + async fn list_albums_is_deterministic_with_scanned_at_and_total() { + let (data_dir, _media, roots) = indexed_library().await; + let state = leaked_state(); + + let first = dispatch(state, data_dir.path(), &roots, "music.list-albums", None).await; + let second = dispatch(state, data_dir.path(), &roots, "music.list-albums", None).await; + assert_eq!(first, second, "ordering is stable across repeated calls"); + + assert_eq!(first["total"], 2); + assert!(first["scanned_at"].as_str().is_some_and(|s| !s.is_empty())); + let albums = first["albums"].as_array().unwrap(); + assert_eq!(albums[0]["album"], "Alpha"); + assert_eq!(albums[0]["track_count"], 2); + assert_eq!(albums[1]["album"], "Beta"); + } + + #[tokio::test] + async fn list_tracks_filters_by_album_and_paginates_with_clamped_limit() { + let (data_dir, _media, roots) = indexed_library().await; + let state = leaked_state(); + + // album_id filter. + let alpha = dispatch( + state, + data_dir.path(), + &roots, + "music.list-tracks", + Some(serde_json::json!({ "album_id": { "album": "Alpha", "album_artist": "X" } })), + ) + .await; + assert_eq!(alpha["total"], 2); + assert_eq!(alpha["tracks"].as_array().unwrap().len(), 2); + + // Pagination: limit/offset slice the filtered set; total is the + // pre-pagination count. + let page = dispatch( + state, + data_dir.path(), + &roots, + "music.list-tracks", + Some(serde_json::json!({ "limit": 1, "offset": 1 })), + ) + .await; + assert_eq!(page["total"], 3); + assert_eq!(page["tracks"].as_array().unwrap().len(), 1); + assert_eq!(page["limit"], 1); + assert_eq!(page["offset"], 1); + + // A huge limit is clamped, not rejected. + let clamped = dispatch( + state, + data_dir.path(), + &roots, + "music.list-tracks", + Some(serde_json::json!({ "limit": 1_000_000 })), + ) + .await; + assert_eq!(clamped["limit"], 500, "limit clamps to the 500 cap"); + // And a zero limit degrades to the smallest page, still no error. + let floor = dispatch( + state, + data_dir.path(), + &roots, + "music.list-tracks", + Some(serde_json::json!({ "limit": 0 })), + ) + .await; + assert_eq!(floor["limit"], 1); + } + + #[tokio::test] + async fn status_reports_stats_running_flag_and_schema_version() { + let (data_dir, _media, roots) = indexed_library().await; + let state = leaked_state(); + + // Run a scan through THIS state so last_stats is populated. + reindex(state, data_dir.path(), &roots).await.unwrap(); + + let status = dispatch(state, data_dir.path(), &roots, "music.status", None).await; + assert_eq!(status["running"], false); + assert_eq!(status["schema_version"], MUSIC_SCHEMA_VERSION); + assert_eq!(status["last_stats"]["scanned"], 3); + assert_eq!(status["track_count"], 3); + assert!(status["scanned_at"].as_str().is_some_and(|s| !s.is_empty())); + } + + #[tokio::test] + async fn reindex_while_running_reports_already_running_not_a_duplicate() { + let (data_dir, _media, roots) = indexed_library().await; + let state = leaked_state(); + + // Simulate a scan in flight by holding the in-progress guard. + let guard = state.try_begin().expect("guard acquired"); + let response = dispatch(state, data_dir.path(), &roots, "music.reindex", None).await; + assert_eq!(response["already_running"], true); + assert_eq!(response["started"], false); + assert_eq!(response["running"], true); + drop(guard); + + // Free again: a reindex starts and returns immediately. + let response = dispatch(state, data_dir.path(), &roots, "music.reindex", None).await; + assert_eq!(response["started"], true); + } + + #[tokio::test] + async fn reindex_incremental_mode_refreshes_without_full_reextraction() { + let (data_dir, media, roots) = indexed_library().await; + let state = leaked_state(); + + write_track(media.path(), "b2.flac", "Night", "Y", "Beta", 2); + let response = dispatch( + state, + data_dir.path(), + &roots, + "music.reindex", + Some(serde_json::json!({ "incremental": true })), + ) + .await; + assert_eq!(response["started"], true); + + // The scan runs in a spawned task; poll until it records stats. + let mut stats = None; + for _ in 0..200 { + if !state.is_running() { + if let Some(s) = state.last_stats() { + stats = Some(s); + break; + } + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + let stats = stats.expect("incremental scan completed"); + assert_eq!(stats.scanned, 4); + assert_eq!( + stats.extracted, 1, + "only the new file is extracted — no full rebuild" + ); + } + + #[tokio::test] + async fn empty_library_returns_empty_arrays_with_scanned_at() { + // Fresh data_dir: no index has ever been written. + let data_dir = tempfile::tempdir().unwrap(); + let media = tempfile::tempdir().unwrap(); + let roots = vec![media.path().to_path_buf()]; + let state = leaked_state(); + + for (method, key) in [ + ("music.list-albums", "albums"), + ("music.list-artists", "artists"), + ("music.list-tracks", "tracks"), + ] { + let response = dispatch(state, data_dir.path(), &roots, method, None).await; + let items = response[key] + .as_array() + .unwrap_or_else(|| panic!("{method}: {key} is an array, never null")); + assert!(items.is_empty()); + assert_eq!(response["total"], 0); + assert!( + response["scanned_at"] + .as_str() + .is_some_and(|s| !s.is_empty()), + "{method}: scanned_at is populated even for an empty library" + ); + } + } +}