From 2a0696037ae88f135ed5c8dd51ecb1e2a904060c Mon Sep 17 00:00:00 2001 From: cucadmuh <49571317+cucadmuh@users.noreply.github.com> Date: Mon, 20 Jul 2026 00:09:26 +0300 Subject: [PATCH] fix(library): align album identity partitions --- .../psysonic-library/src/browse_projection.rs | 251 +++++++++++++++--- .../psysonic-library/src/composer_scope.rs | 20 +- .../psysonic-library/src/identity/mod.rs | 3 +- .../psysonic-library/src/identity/norm.rs | 3 +- .../psysonic-library/src/identity/rebuild.rs | 59 +++- .../psysonic-library/src/repos/track.rs | 8 + .../psysonic-library/src/scope_browse.rs | 3 + 7 files changed, 301 insertions(+), 46 deletions(-) diff --git a/src-tauri/crates/psysonic-library/src/browse_projection.rs b/src-tauri/crates/psysonic-library/src/browse_projection.rs index e3f58c7d..b3a51c82 100644 --- a/src-tauri/crates/psysonic-library/src/browse_projection.rs +++ b/src-tauri/crates/psysonic-library/src/browse_projection.rs @@ -147,28 +147,11 @@ pub(crate) fn refresh_album_scopes( "UPDATE album_browse_projection SET identity_key = ?4 \ WHERE server_id = ?1 AND library_id = ?2 AND album_id = ?3", )?; - let mut identity_source = tx.prepare_cached( - "SELECT name, artist FROM album_browse_projection \ - WHERE server_id = ?1 AND library_id = ?2 AND album_id = ?3", - )?; for (server_id, library_id, album_id) in &scopes { delete.execute(params![server_id, library_id, album_id])?; insert.execute(params![server_id, library_id, album_id])?; - let source = identity_source - .query_row(params![server_id, library_id, album_id], |r| { - Ok((r.get::<_, String>(0)?, r.get::<_, Option>(1)?)) - }) - .optional()?; - if let Some((name, artist)) = source { - let identity_key = crate::identity::build_track_cluster_keys( - artist.as_deref(), - "", - &name, - artist.as_deref(), - ) - .album_key; - update_identity.execute(params![server_id, library_id, album_id, identity_key])?; - } + let identity_key = crate::identity::concrete_physical_album_key(server_id, album_id); + update_identity.execute(params![server_id, library_id, album_id, identity_key])?; } crate::composer_projection::refresh_album_scopes(tx, &scopes)?; Ok(()) @@ -196,15 +179,13 @@ pub(crate) fn rebuild_server(tx: &Transaction<'_>, server_id: &str) -> rusqlite: params![server_id], )?; let mut stmt = tx.prepare( - "SELECT library_id, album_id, name, artist FROM album_browse_projection WHERE server_id = ?1", + "SELECT library_id, album_id FROM album_browse_projection WHERE server_id = ?1", )?; let rows = stmt .query_map(params![server_id], |r| { Ok(( r.get::<_, String>(0)?, r.get::<_, String>(1)?, - r.get::<_, String>(2)?, - r.get::<_, Option>(3)?, )) })? .collect::>>()?; @@ -213,20 +194,51 @@ pub(crate) fn rebuild_server(tx: &Transaction<'_>, server_id: &str) -> rusqlite: "UPDATE album_browse_projection SET identity_key = ?4 \ WHERE server_id = ?1 AND library_id = ?2 AND album_id = ?3", )?; - for (library_id, album_id, name, artist) in rows { - let identity_key = crate::identity::build_track_cluster_keys( - artist.as_deref(), - "", - &name, - artist.as_deref(), - ) - .album_key; + for (library_id, album_id) in rows { + let identity_key = crate::identity::concrete_physical_album_key(server_id, &album_id); update.execute(params![server_id, library_id, album_id, identity_key])?; } crate::composer_projection::rebuild_scope(tx, server_id, "")?; Ok(()) } +/// Keep materialized album browse partitions aligned with the cluster sidecar. +/// Every physical album gets one unanimous cluster key or a server-qualified fallback. +pub(crate) fn reconcile_identity_keys( + tx: &Transaction<'_>, + server_id: Option<&str>, +) -> rusqlite::Result<()> { + let server_filter = if server_id.is_some() { + " AND ap.server_id = ?1" + } else { + "" + }; + let sql = format!( + "UPDATE album_browse_projection AS ap \ + SET identity_key = COALESCE(( \ + SELECT CASE \ + WHEN COUNT(*) > 0 \ + AND COUNT(*) = COUNT(ck.album_key) \ + AND COUNT(DISTINCT ck.album_key) = 1 \ + THEN MAX(ck.album_key) \ + END \ + FROM track t \ + LEFT JOIN cluster.track_cluster_key ck \ + ON ck.server_id = t.server_id AND ck.track_id = t.id \ + WHERE t.server_id = ap.server_id AND t.album_id = ap.album_id AND t.deleted = 0 \ + ), 'physical:' || length(ap.server_id) || ':' || ap.server_id || ':' || ap.album_id) \ + WHERE EXISTS ( \ + SELECT 1 FROM track t \ + WHERE t.server_id = ap.server_id AND t.album_id = ap.album_id AND t.deleted = 0 \ + ){server_filter}" + ); + match server_id { + Some(server_id) => tx.execute(&sql, params![server_id])?, + None => tx.execute(&sql, [])?, + }; + Ok(()) +} + /// Rebuild the projection rows affected by an authoritative scope mutation. /// Empty scope means every library on the server; non-empty scope is exact. pub(crate) fn rebuild_scope( @@ -462,6 +474,10 @@ fn run_backfill_impl(store: &LibraryStore, app: Option<&AppHandle>) -> Result<() #[cfg(test)] mod tests { use super::*; + use crate::dto::{ + LibraryScopeBrowseEntity, LibraryScopeBrowseRequest, LibraryScopePair, LibrarySortClause, + SortDir, + }; use crate::repos::{TrackRepository, TrackRow}; fn track(id: &str, album_id: &str, album: &str, library_id: &str) -> TrackRow { @@ -505,6 +521,58 @@ mod tests { } } + #[allow(clippy::too_many_arguments)] + fn album_track( + server_id: &str, + id: &str, + artist: &str, + artist_id: &str, + album_id: &str, + album: &str, + album_artist: &str, + library_id: &str, + ) -> TrackRow { + let mut row = track(id, album_id, album, library_id); + row.server_id = server_id.into(); + row.artist = Some(artist.into()); + row.artist_id = Some(artist_id.into()); + row.album_artist = Some(album_artist.into()); + row + } + + fn insert_artist(store: &LibraryStore, server_id: &str, artist_id: &str, name: &str) { + store + .with_conn_mut("test.browse_projection.artist", |conn| { + conn.execute( + "INSERT INTO artist(server_id, id, name, synced_at) VALUES (?1, ?2, ?3, 1)", + params![server_id, artist_id, name], + )?; + Ok(()) + }) + .unwrap(); + } + + fn browse_albums( + store: &LibraryStore, + scopes: Vec, + ) -> Vec { + crate::scope_browse::browse( + store, + &LibraryScopeBrowseRequest { + entity: LibraryScopeBrowseEntity::Album, + scopes, + sort: vec![LibrarySortClause { + field: "name".into(), + dir: SortDir::Asc, + }], + limit: 20, + cursor: None, + }, + ) + .unwrap() + .albums + } + #[test] fn ingest_refreshes_only_affected_album_projection() { let store = LibraryStore::open_in_memory(); @@ -560,4 +628,127 @@ mod tests { .unwrap(); assert_eq!(count, 1); } + + #[test] + fn ordinary_browse_reconciles_partial_keys_to_one_canonical_album_partition() { + let store = LibraryStore::open_in_memory(); + insert_artist(&store, "s1", "artist-1", "Metallica"); + insert_artist(&store, "s2", "artist-2", "Metallica"); + TrackRepository::new(&store) + .upsert_batch(&[album_track( + "s1", + "t1", + "Metallica", + "artist-1", + "album-1", + "S&M2", + "Metallica & San Francisco Symphony", + "lib-a", + )]) + .unwrap(); + crate::identity::rebuild_cluster_keys(&store, None).unwrap(); + + TrackRepository::new(&store) + .upsert_batch(&[ + album_track( + "s1", "t2", "Metallica", "artist-1", "album-1", "S&M2", + "Metallica", "lib-b", + ), + album_track( + "s2", "t3", "Metallica", "artist-2", "album-2", "S&M2", + "Metallica", "lib-c", + ), + ]) + .unwrap(); + + let albums = browse_albums( + &store, + vec![ + LibraryScopePair { + server_id: "s1".into(), + library_id: "lib-a".into(), + }, + LibraryScopePair { + server_id: "s1".into(), + library_id: "lib-b".into(), + }, + LibraryScopePair { + server_id: "s2".into(), + library_id: "lib-c".into(), + }, + ], + ); + assert_eq!(albums.len(), 1); + assert_eq!(albums[0].server_id, "s1"); + assert_eq!(albums[0].id, "album-1"); + + let keys = store + .with_read_conn(|conn| { + let mut stmt = conn.prepare( + "SELECT DISTINCT identity_key FROM album_browse_projection \ + WHERE album_id IN ('album-1', 'album-2')", + )?; + let rows = stmt + .query_map([], |row| row.get::<_, String>(0))? + .collect::>>()?; + Ok(rows) + }) + .unwrap(); + assert_eq!(keys.len(), 1); + } + + #[test] + fn ordinary_browse_keeps_ambiguous_physical_albums_separate() { + let store = LibraryStore::open_in_memory(); + for (server, artist_id, name) in [ + ("s1", "s1-a", "Artist A"), + ("s1", "s1-b", "Artist B"), + ("s2", "s2-a", "Artist A"), + ("s2", "s2-b", "Artist B"), + ] { + insert_artist(&store, server, artist_id, name); + } + TrackRepository::new(&store) + .upsert_batch(&[ + album_track( + "s1", "s1-t1", "Artist A", "s1-a", "s1-album", "Split", + "Various Artists", "lib-a", + ), + album_track( + "s1", "s1-t2", "Artist B", "s1-b", "s1-album", "Split", + "Various Artists", "lib-a", + ), + album_track( + "s2", "s2-t1", "Artist A", "s2-a", "s2-album", "Split", + "Various Artists", "lib-b", + ), + album_track( + "s2", "s2-t2", "Artist B", "s2-b", "s2-album", "Split", + "Various Artists", "lib-b", + ), + ]) + .unwrap(); + + let albums = browse_albums( + &store, + vec![ + LibraryScopePair { + server_id: "s1".into(), + library_id: "lib-a".into(), + }, + LibraryScopePair { + server_id: "s2".into(), + library_id: "lib-b".into(), + }, + ], + ); + assert_eq!(albums.len(), 2); + assert_eq!( + albums + .iter() + .map(|album| album.id.as_str()) + .collect::>(), + vec!["s1-album", "s2-album"] + ); + } } diff --git a/src-tauri/crates/psysonic-library/src/composer_scope.rs b/src-tauri/crates/psysonic-library/src/composer_scope.rs index f0a76538..2ffa900e 100644 --- a/src-tauri/crates/psysonic-library/src/composer_scope.rs +++ b/src-tauri/crates/psysonic-library/src/composer_scope.rs @@ -23,6 +23,7 @@ pub fn list_composers( request: &LibraryScopeListRequest, ) -> Result, String> { let scopes = non_empty_scopes(&request.scopes)?; + crate::scope_merge::ensure_cluster_keys_for_scopes(store, scopes)?; let limit = request.limit.unwrap_or(10_000).clamp(1, 10_000); let offset = request.offset.unwrap_or(0); let (cte, mut binds) = scope_cte_sql(scopes); @@ -150,6 +151,7 @@ pub fn composer_detail( request: &LibraryScopeComposerDetailRequest, ) -> Result { let scopes = non_empty_scopes(&request.scopes)?; + crate::scope_merge::ensure_cluster_keys_for_scopes(store, scopes)?; let server_id = request.server_id.trim(); let composer_id = request.composer_id.trim(); if server_id.is_empty() || composer_id.is_empty() { @@ -297,21 +299,23 @@ mod tests { #[test] fn browse_merges_unique_names_across_servers_and_detail_dedupes_album() { let store = LibraryStore::open_in_memory(); - let repo = TrackRepository::new(&store); - repo.upsert_batch(&[ - track("s1", "t1", "a1", "c1", "Composer"), - track("s2", "t2", "a2", "c2", "Composer"), - ]) - .unwrap(); store - .with_conn_mut("test", |conn| { + .with_conn_mut("test.composer_scope.artist", |conn| { conn.execute( - "UPDATE album_browse_projection SET identity_key = 'same-album'", + "INSERT INTO artist(server_id, id, name, synced_at) VALUES \ + ('s1', 'performer', 'Performer', 1), \ + ('s2', 'performer', 'Performer', 1)", [], )?; Ok(()) }) .unwrap(); + let repo = TrackRepository::new(&store); + let mut first = track("s1", "t1", "a1", "c1", "Composer"); + first.album = "Shared Album".into(); + let mut second = track("s2", "t2", "a2", "c2", "Composer"); + second.album = "Shared Album".into(); + repo.upsert_batch(&[first, second]).unwrap(); let composers = list_composers( &store, diff --git a/src-tauri/crates/psysonic-library/src/identity/mod.rs b/src-tauri/crates/psysonic-library/src/identity/mod.rs index 01fdea0c..13c97558 100644 --- a/src-tauri/crates/psysonic-library/src/identity/mod.rs +++ b/src-tauri/crates/psysonic-library/src/identity/mod.rs @@ -14,7 +14,8 @@ pub use norm::NORM_VERSION; pub(crate) use norm::norm_part; pub use rebuild::{cluster_rebuild_needed, ensure_cluster_keys_built, rebuild_cluster_keys}; pub(crate) use rebuild::{ - delete_cluster_keys_for_tracks, prune_cluster_keys_for_scope, refresh_library_ids_for_albums, + concrete_physical_album_key, delete_cluster_keys_for_tracks, mark_cluster_keys_dirty, + prune_cluster_keys_for_scope, refresh_library_ids_for_albums, }; pub use keys::{build_track_cluster_keys, TrackClusterKeys}; diff --git a/src-tauri/crates/psysonic-library/src/identity/norm.rs b/src-tauri/crates/psysonic-library/src/identity/norm.rs index 808e6fe5..dc55276a 100644 --- a/src-tauri/crates/psysonic-library/src/identity/norm.rs +++ b/src-tauri/crates/psysonic-library/src/identity/norm.rs @@ -23,7 +23,8 @@ pub(crate) const KEY_SEP: char = '\u{001f}'; /// v2: locale-aware folding (ß→ss, æ→ae, œ→oe, Romanian ș/ț, Cyrillic ё/й). /// v3: artist keys use the canonical artist entity name when the track has an artist id. /// v4: physical albums use one canonical or server-qualified album key. -pub const NORM_VERSION: &str = "4"; +/// v5: materialized album browse rows use the same physical-album identity partition. +pub const NORM_VERSION: &str = "5"; /// Normalize one identity field. Returns `None` when input is empty/whitespace-only /// or when normalization strips everything (punctuation-only, etc.). diff --git a/src-tauri/crates/psysonic-library/src/identity/rebuild.rs b/src-tauri/crates/psysonic-library/src/identity/rebuild.rs index 03a5cc9e..9bcbbc78 100644 --- a/src-tauri/crates/psysonic-library/src/identity/rebuild.rs +++ b/src-tauri/crates/psysonic-library/src/identity/rebuild.rs @@ -1,5 +1,6 @@ //! Batch rebuild of `cluster.track_cluster_key` from live `track` rows. +use std::collections::HashSet; use std::time::{SystemTime, UNIX_EPOCH}; use rusqlite::{params, params_from_iter, Connection, OptionalExtension, Transaction}; @@ -22,6 +23,29 @@ ON CONFLICT(server_id, track_id) DO UPDATE SET duration_sec = excluded.duration_sec "; +const DIRTY_META_PREFIX: &str = "dirty_server:"; + +fn dirty_meta_key(server_id: &str) -> String { + format!("{DIRTY_META_PREFIX}{server_id}") +} + +pub(crate) fn mark_cluster_keys_dirty<'a>( + tx: &Transaction<'_>, + server_ids: impl IntoIterator, +) -> rusqlite::Result<()> { + let mut seen = HashSet::new(); + let mut statement = tx.prepare_cached( + "INSERT INTO cluster.cluster_meta(key, value) VALUES (?1, '1') \ + ON CONFLICT(key) DO UPDATE SET value = excluded.value", + )?; + for server_id in server_ids { + if seen.insert(server_id) { + statement.execute(params![dirty_meta_key(server_id)])?; + } + } + Ok(()) +} + pub(crate) fn delete_cluster_keys_for_tracks( tx: &Transaction<'_>, server_id: &str, @@ -150,7 +174,7 @@ type SourceTrackRow = ( i64, ); -fn concrete_physical_album_key(server_id: &str, album_id: &str) -> String { +pub(crate) fn concrete_physical_album_key(server_id: &str, album_id: &str) -> String { format!("physical:{}:{server_id}:{album_id}", server_id.len()) } @@ -283,6 +307,21 @@ pub fn rebuild_cluster_keys(store: &LibraryStore, server_id: Option<&str>) -> Re [], )?; } + crate::browse_projection::reconcile_identity_keys(&tx, server_id)?; + match server_id { + Some(server_id) => { + tx.execute( + "DELETE FROM cluster.cluster_meta WHERE key = ?1", + params![dirty_meta_key(server_id)], + )?; + } + None => { + tx.execute( + "DELETE FROM cluster.cluster_meta WHERE key LIKE ?1", + params![format!("{DIRTY_META_PREFIX}%")], + )?; + } + } set_cluster_meta(&tx)?; tx.commit()?; Ok(upserted) @@ -294,7 +333,7 @@ pub fn rebuild_cluster_keys(store: &LibraryStore, server_id: Option<&str>) -> Re /// changed) — then **all** servers are rebuilt, because [`rebuild_cluster_keys`] /// stamps a single global `norm_version`; a per-server rebuild would flip the /// gate and strand every other server's stale keys; or -/// - this server has tracks but no keys yet (fresh index / newly synced server). +/// - this server has live tracks missing sidecar keys (fresh or partially synced index). pub fn ensure_cluster_keys_built(store: &LibraryStore, server_id: &str) -> Result<(), String> { let rebuild_all = store .with_read_conn(cluster_rebuild_needed) @@ -313,12 +352,20 @@ pub fn ensure_cluster_keys_built(store: &LibraryStore, server_id: &str) -> Resul if track_count == 0 { return Ok(false); } - let key_count: i64 = conn.query_row( - "SELECT COUNT(*) FROM cluster.track_cluster_key WHERE server_id = ?1", - [server_id], + let needs_rebuild: bool = conn.query_row( + "SELECT EXISTS( \ + SELECT 1 FROM track t \ + LEFT JOIN cluster.track_cluster_key ck \ + ON ck.server_id = t.server_id AND ck.track_id = t.id \ + WHERE t.server_id = ?1 AND t.deleted = 0 AND ck.track_id IS NULL \ + LIMIT 1 \ + ) OR EXISTS( \ + SELECT 1 FROM cluster.cluster_meta WHERE key = ?2 \ + )", + params![server_id, dirty_meta_key(server_id)], |r| r.get(0), )?; - Ok(key_count == 0) + Ok(needs_rebuild) }) .map_err(|e| e.to_string())?; if needs_rebuild { diff --git a/src-tauri/crates/psysonic-library/src/repos/track.rs b/src-tauri/crates/psysonic-library/src/repos/track.rs index 83ebd4f0..dca99770 100644 --- a/src-tauri/crates/psysonic-library/src/repos/track.rs +++ b/src-tauri/crates/psysonic-library/src/repos/track.rs @@ -208,6 +208,10 @@ impl<'a> TrackRepository<'a> { sync_track_genre_row(&tx, r)?; } drop(upsert); + crate::identity::mark_cluster_keys_dirty( + &tx, + rows.iter().map(|row| row.server_id.as_str()), + )?; crate::browse_projection::refresh_album_scopes(&tx, affected_album_scopes)?; tx.commit()?; Ok(()) @@ -762,6 +766,10 @@ impl<'a> TrackRepository<'a> { drop(upsert); drop(remap_lookup); + crate::identity::mark_cluster_keys_dirty( + &tx, + rows.iter().map(|row| row.server_id.as_str()), + )?; crate::browse_projection::refresh_album_scopes(&tx, affected_album_scopes)?; tx.commit()?; diff --git a/src-tauri/crates/psysonic-library/src/scope_browse.rs b/src-tauri/crates/psysonic-library/src/scope_browse.rs index 81fa4d29..7e139b22 100644 --- a/src-tauri/crates/psysonic-library/src/scope_browse.rs +++ b/src-tauri/crates/psysonic-library/src/scope_browse.rs @@ -304,6 +304,9 @@ fn browse_albums( ) -> Result { let sort = album_sort(&request.sort)?; let cursor = parse_cursor(request.cursor.as_deref(), &request.scopes, sort)?; + if cursor.is_none() { + crate::scope_merge::ensure_cluster_keys_for_scopes(store, &request.scopes)?; + } let limit = request.limit.clamp(1, 200) as usize; let candidate_limit = CANDIDATE_PAGE_SIZE.max(limit.saturating_add(1)); let mut candidates = Vec::with_capacity(request.scopes.len());