diff --git a/crates/qbz-library/src/database.rs b/crates/qbz-library/src/database.rs index 5c04cac02..7328a508b 100644 --- a/crates/qbz-library/src/database.rs +++ b/crates/qbz-library/src/database.rs @@ -264,6 +264,11 @@ impl LibraryDatabase { CREATE INDEX IF NOT EXISTS idx_playlist_local_tracks_playlist ON playlist_local_tracks(qobuz_playlist_id); + -- The Add-to-Playlist picker asks the inverse question — "which + -- playlists hold this track" (playlist_membership.rs). + CREATE INDEX IF NOT EXISTS idx_playlist_local_tracks_track + ON playlist_local_tracks(local_track_id, qobuz_playlist_id); + -- Plex tracks added to playlists. Kept in its own table because -- Plex tracks live on a remote server and have a TEXT rating key, -- not the i64 filesystem id used by local_tracks. No foreign key @@ -282,6 +287,9 @@ impl LibraryDatabase { CREATE INDEX IF NOT EXISTS idx_playlist_plex_tracks_playlist ON playlist_plex_tracks(qobuz_playlist_id); + CREATE INDEX IF NOT EXISTS idx_playlist_plex_tracks_key + ON playlist_plex_tracks(plex_rating_key, qobuz_playlist_id); + -- Custom track order per playlist (user-defined arrangement) CREATE TABLE IF NOT EXISTS playlist_track_custom_order ( id INTEGER PRIMARY KEY, diff --git a/crates/qbz-library/src/lib.rs b/crates/qbz-library/src/lib.rs index 5923b6931..f899e9cd0 100644 --- a/crates/qbz-library/src/lib.rs +++ b/crates/qbz-library/src/lib.rs @@ -37,6 +37,7 @@ pub mod local_playlists; mod metadata; mod models; mod mount_info; +pub mod playlist_membership; pub mod qobuz_playlist_snapshot; pub mod reachability; mod remote_tag_sidecar; diff --git a/crates/qbz-library/src/local_playlists.rs b/crates/qbz-library/src/local_playlists.rs index ff6d1e025..f3ab169a0 100644 --- a/crates/qbz-library/src/local_playlists.rs +++ b/crates/qbz-library/src/local_playlists.rs @@ -154,6 +154,19 @@ pub fn init_schema(conn: &Connection) -> Result<()> { CREATE INDEX IF NOT EXISTS idx_local_playlist_tracks_playlist ON local_playlist_tracks(playlist_id, position); + + -- Inverse direction for the Add-to-Playlist picker's containment + -- question (playlist_membership.rs). Partial: each row fills exactly + -- one of the three identity columns. + CREATE INDEX IF NOT EXISTS idx_local_playlist_tracks_qobuz + ON local_playlist_tracks(qobuz_track_id, playlist_id) + WHERE qobuz_track_id IS NOT NULL; + CREATE INDEX IF NOT EXISTS idx_local_playlist_tracks_path + ON local_playlist_tracks(local_path, playlist_id) + WHERE local_path IS NOT NULL; + CREATE INDEX IF NOT EXISTS idx_local_playlist_tracks_plex + ON local_playlist_tracks(plex_key, playlist_id) + WHERE plex_key IS NOT NULL; "#, )?; // Additive migration (B3): DBs created before the favorite/hidden diff --git a/crates/qbz-library/src/playlist_membership.rs b/crates/qbz-library/src/playlist_membership.rs new file mode 100644 index 000000000..831821f27 --- /dev/null +++ b/crates/qbz-library/src/playlist_membership.rs @@ -0,0 +1,479 @@ +//! The inverse membership question: WHICH playlists already hold this track? +//! +//! The playlist tables were all indexed forward (`playlist_id` leading); +//! the Add-to-Playlist picker asks the opposite direction. This module owns +//! the typed track reference at that boundary — replacing the loosely encoded +//! `"plex:"` / `""` strings — and the containment query across all +//! four membership stores: +//! +//! - `qobuz_playlist_snapshot_tracks` — Qobuz tracks in Qobuz playlists +//! (hydrated by the membership worker; honesty via `index_state`); +//! - `playlist_local_tracks` / `playlist_plex_tracks` — library/Plex sidecars +//! attached to Qobuz playlists (authoritative locally, complete by +//! construction: QBZ is their only writer); +//! - `local_playlist_tracks` — everything inside `local:` playlists (same: +//! locally authoritative). +//! +//! Identity notes. Jellyfin and Subsonic rows reach playlists through their +//! `local_tracks` projection, so their playlist identity IS the library row +//! (`track_id` + the stored path/key) — structurally collision-free across +//! servers because each server's items are distinct projected rows. Plex +//! sidecars store the bare rating key (the Plex merge keeps it in +//! `file_path`); rating keys are scoped to the single configured server, the +//! same assumption the sidecar tables have always made. + +use std::collections::{HashMap, HashSet}; + +use rusqlite::{params, Connection, Result}; + +/// One selected track at the picker/index boundary. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub enum PlaylistTrackRef { + /// A Qobuz catalog track. + Qobuz(u64), + /// A library-projected row: a local file, or a Jellyfin/Subsonic track + /// (their playlist storage identity is the projection). `path` is the + /// `local_tracks.file_path` value local playlists key on. + Library { + track_id: i64, + path: Option, + }, + /// A Plex track by rating key. + Plex { rating_key: String }, +} + +impl PlaylistTrackRef { + /// Build the ref for a library row the way the playlist writers key it: + /// Plex rows carry the rating key in `file_path`; every other source is + /// the projected library row itself. + pub fn from_library_row(source: &str, track_id: i64, file_path: &str) -> Self { + if source == "plex" { + Self::Plex { + rating_key: file_path.to_string(), + } + } else { + Self::Library { + track_id, + path: Some(file_path.to_string()), + } + } + } +} + +/// A playlist that contains at least one of the selected refs. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub enum ContainmentTarget { + Qobuz(u64), + Local(String), +} + +#[derive(Debug, Clone)] +pub struct PlaylistContainment { + pub target: ContainmentTarget, + /// How many DISTINCT selected refs this playlist holds (the picker's + /// "N of M"). Duplicate rows inside a playlist count once. + pub contained: u32, +} + +/// SQLite's default variable cap is 999; stay far under it. +const IN_CHUNK: usize = 400; + +fn placeholders(n: usize) -> String { + let mut s = String::with_capacity(n * 2); + for i in 0..n { + if i > 0 { + s.push(','); + } + s.push('?'); + } + s +} + +/// Every writable playlist containing at least one of `refs`, with the +/// distinct-ref count. Qobuz targets come from the membership snapshot +/// restricted to OWNED, ACTIVE playlists (the strip offers add targets, and a +/// retired header must not resurrect there) plus the sidecar tables; `local:` +/// targets from `local_playlist_tracks`. Whether the Qobuz half may be +/// presented as definitive is `qobuz_playlist_snapshot::index_state`'s call, +/// not this function's. +pub fn playlists_containing( + conn: &Connection, + refs: &[PlaylistTrackRef], +) -> Result> { + // ordinal = position in `refs`; contained counts distinct ordinals. + let mut qobuz_ids: Vec<(i64, usize)> = Vec::new(); + let mut lib_ids: Vec<(i64, usize)> = Vec::new(); + let mut lib_paths: Vec<(&str, usize)> = Vec::new(); + let mut plex_keys: Vec<(&str, usize)> = Vec::new(); + for (ordinal, r) in refs.iter().enumerate() { + match r { + PlaylistTrackRef::Qobuz(id) => qobuz_ids.push((*id as i64, ordinal)), + PlaylistTrackRef::Library { track_id, path } => { + lib_ids.push((*track_id, ordinal)); + if let Some(p) = path { + lib_paths.push((p.as_str(), ordinal)); + } + } + PlaylistTrackRef::Plex { rating_key } => plex_keys.push((rating_key.as_str(), ordinal)), + } + } + + let mut hits: HashMap> = HashMap::new(); + let mut collect_qobuz = |rows: Vec<(i64, usize)>| { + for (pid, ordinal) in rows { + hits.entry(ContainmentTarget::Qobuz(pid as u64)) + .or_default() + .insert(ordinal); + } + }; + + // Qobuz refs → snapshot membership of owned active playlists. + for chunk in qobuz_ids.chunks(IN_CHUNK) { + let sql = format!( + "SELECT t.qobuz_playlist_id, t.track_id + FROM qobuz_playlist_snapshot_tracks t + JOIN qobuz_playlist_snapshot h + ON h.qobuz_playlist_id = t.qobuz_playlist_id + WHERE h.is_owned = 1 AND h.inactive = 0 + AND t.track_id IN ({})", + placeholders(chunk.len()) + ); + let mut stmt = conn.prepare(&sql)?; + let rows = stmt.query_map( + rusqlite::params_from_iter(chunk.iter().map(|(id, _)| *id)), + |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)), + )?; + let mut found = Vec::new(); + for row in rows { + let (pid, tid) = row?; + for (id, ordinal) in chunk { + if *id == tid { + found.push((pid, *ordinal)); + } + } + } + collect_qobuz(found); + } + + // Library refs → local-track sidecars on Qobuz playlists. + for chunk in lib_ids.chunks(IN_CHUNK) { + let sql = format!( + "SELECT qobuz_playlist_id, local_track_id FROM playlist_local_tracks + WHERE local_track_id IN ({})", + placeholders(chunk.len()) + ); + let mut stmt = conn.prepare(&sql)?; + let rows = stmt.query_map( + rusqlite::params_from_iter(chunk.iter().map(|(id, _)| *id)), + |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)), + )?; + let mut found = Vec::new(); + for row in rows { + let (pid, tid) = row?; + for (id, ordinal) in chunk { + if *id == tid { + found.push((pid, *ordinal)); + } + } + } + collect_qobuz(found); + } + + // Plex refs → Plex sidecars on Qobuz playlists. + for chunk in plex_keys.chunks(IN_CHUNK) { + let sql = format!( + "SELECT qobuz_playlist_id, plex_rating_key FROM playlist_plex_tracks + WHERE plex_rating_key IN ({})", + placeholders(chunk.len()) + ); + let mut stmt = conn.prepare(&sql)?; + let rows = stmt.query_map( + rusqlite::params_from_iter(chunk.iter().map(|(key, _)| *key)), + |r| Ok((r.get::<_, i64>(0)?, r.get::<_, String>(1)?)), + )?; + let mut found = Vec::new(); + for row in rows { + let (pid, key) = row?; + for (k, ordinal) in chunk { + if *k == key { + found.push((pid, *ordinal)); + } + } + } + collect_qobuz(found); + } + + // All three identity columns of local playlists. + let mut collect_local = |rows: Vec<(String, usize)>| { + for (pid, ordinal) in rows { + hits.entry(ContainmentTarget::Local(pid)) + .or_default() + .insert(ordinal); + } + }; + for chunk in qobuz_ids.chunks(IN_CHUNK) { + let sql = format!( + "SELECT playlist_id, qobuz_track_id FROM local_playlist_tracks + WHERE qobuz_track_id IN ({})", + placeholders(chunk.len()) + ); + let mut stmt = conn.prepare(&sql)?; + let rows = stmt.query_map( + rusqlite::params_from_iter(chunk.iter().map(|(id, _)| *id)), + |r| Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?)), + )?; + let mut found = Vec::new(); + for row in rows { + let (pid, tid) = row?; + for (id, ordinal) in chunk { + if *id == tid { + found.push((pid.clone(), *ordinal)); + } + } + } + collect_local(found); + } + for chunk in lib_paths.chunks(IN_CHUNK) { + let sql = format!( + "SELECT playlist_id, local_path FROM local_playlist_tracks + WHERE local_path IN ({})", + placeholders(chunk.len()) + ); + let mut stmt = conn.prepare(&sql)?; + let rows = stmt.query_map( + rusqlite::params_from_iter(chunk.iter().map(|(p, _)| *p)), + |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)), + )?; + let mut found = Vec::new(); + for row in rows { + let (pid, path) = row?; + for (p, ordinal) in chunk { + if *p == path { + found.push((pid.clone(), *ordinal)); + } + } + } + collect_local(found); + } + for chunk in plex_keys.chunks(IN_CHUNK) { + let sql = format!( + "SELECT playlist_id, plex_key FROM local_playlist_tracks + WHERE plex_key IN ({})", + placeholders(chunk.len()) + ); + let mut stmt = conn.prepare(&sql)?; + let rows = stmt.query_map( + rusqlite::params_from_iter(chunk.iter().map(|(k, _)| *k)), + |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)), + )?; + let mut found = Vec::new(); + for row in rows { + let (pid, key) = row?; + for (k, ordinal) in chunk { + if *k == key { + found.push((pid.clone(), *ordinal)); + } + } + } + collect_local(found); + } + + let mut out: Vec = hits + .into_iter() + .map(|(target, ordinals)| PlaylistContainment { + target, + contained: ordinals.len() as u32, + }) + .collect(); + // Deterministic order: fullest first, then stable by identity. + out.sort_by(|a, b| { + b.contained.cmp(&a.contained).then_with(|| { + let key = |t: &ContainmentTarget| match t { + ContainmentTarget::Qobuz(id) => (0u8, id.to_string()), + ContainmentTarget::Local(id) => (1u8, id.clone()), + }; + key(&a.target).cmp(&key(&b.target)) + }) + }); + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::qobuz_playlist_snapshot as snapshot; + + fn conn() -> Connection { + let c = Connection::open_in_memory().unwrap(); + snapshot::init_schema(&c).unwrap(); + crate::local_playlists::init_schema(&c).unwrap(); + // Self-contained copies of the sidecar tables (the database.rs idiom: + // unit tests build only the schema slice they exercise). + c.execute_batch( + "CREATE TABLE playlist_local_tracks ( + id INTEGER PRIMARY KEY, + qobuz_playlist_id INTEGER NOT NULL, + local_track_id INTEGER NOT NULL, + position INTEGER NOT NULL, + added_at INTEGER NOT NULL, + UNIQUE(qobuz_playlist_id, local_track_id)); + CREATE TABLE playlist_plex_tracks ( + id INTEGER PRIMARY KEY, + qobuz_playlist_id INTEGER NOT NULL, + plex_rating_key TEXT NOT NULL, + position INTEGER NOT NULL, + added_at INTEGER NOT NULL, + UNIQUE(qobuz_playlist_id, plex_rating_key));", + ) + .unwrap(); + c + } + + /// ONE authoritative list naming every playlist (separate lists would + /// advance the generation and retire the earlier ones — the grace rule + /// working as intended), then a membership snapshot for each. + fn owned_playlists(c: &Connection, playlists: &[(u64, &str, &[u64])]) { + let entries: Vec = playlists + .iter() + .map(|(id, name, ids)| snapshot::AuthoritativeEntry { + qobuz_playlist_id: *id, + name: name.to_string(), + owner: None, + track_count: Some(ids.len() as u32), + remote_updated_at: None, + is_owned: true, + }) + .collect(); + snapshot::record_authoritative_list(c, &entries).unwrap(); + for (id, name, ids) in playlists { + snapshot::replace_tracks(c, *id, name, None, ids, None).unwrap(); + } + } + + fn contained(rows: &[PlaylistContainment], target: ContainmentTarget) -> Option { + rows.iter() + .find(|r| r.target == target) + .map(|r| r.contained) + } + + #[test] + fn ref_construction_routes_plex_by_stored_key() { + assert_eq!( + PlaylistTrackRef::from_library_row("plex", 99, "12345"), + PlaylistTrackRef::Plex { + rating_key: "12345".into() + } + ); + assert_eq!( + PlaylistTrackRef::from_library_row("jellyfin", 7, "/j/item"), + PlaylistTrackRef::Library { + track_id: 7, + path: Some("/j/item".into()) + } + ); + } + + #[test] + fn qobuz_containment_counts_distinct_refs() { + let c = conn(); + owned_playlists( + &c, + &[ + (1, "Both", &[10, 20, 30]), + (2, "One", &[20]), + (3, "Neither", &[99]), + ], + ); + + let rows = playlists_containing( + &c, + &[PlaylistTrackRef::Qobuz(10), PlaylistTrackRef::Qobuz(20)], + ) + .unwrap(); + assert_eq!(contained(&rows, ContainmentTarget::Qobuz(1)), Some(2)); + assert_eq!(contained(&rows, ContainmentTarget::Qobuz(2)), Some(1)); + assert_eq!(contained(&rows, ContainmentTarget::Qobuz(3)), None); + // Fullest playlist first. + assert_eq!(rows[0].target, ContainmentTarget::Qobuz(1)); + } + + #[test] + fn duplicate_membership_rows_count_once() { + let c = conn(); + owned_playlists(&c, &[(1, "Dupes", &[10, 10, 10])]); + let rows = playlists_containing(&c, &[PlaylistTrackRef::Qobuz(10)]).unwrap(); + assert_eq!(contained(&rows, ContainmentTarget::Qobuz(1)), Some(1)); + } + + #[test] + fn unowned_and_inactive_playlists_are_not_targets() { + let c = conn(); + snapshot::record_authoritative_list( + &c, + &[snapshot::AuthoritativeEntry { + qobuz_playlist_id: 5, + name: "Followed".into(), + owner: None, + track_count: Some(1), + remote_updated_at: None, + is_owned: false, + }], + ) + .unwrap(); + snapshot::replace_tracks(&c, 5, "Followed", None, &[10], None).unwrap(); + let rows = playlists_containing(&c, &[PlaylistTrackRef::Qobuz(10)]).unwrap(); + assert!(rows.is_empty()); + + owned_playlists(&c, &[(6, "Mine", &[10])]); + // Retire 6 by two authoritative lists without it. + for _ in 0..2 { + snapshot::record_authoritative_list(&c, &[]).unwrap(); + } + let rows = playlists_containing(&c, &[PlaylistTrackRef::Qobuz(10)]).unwrap(); + assert!(rows.is_empty()); + } + + #[test] + fn sidecars_and_local_playlists_answer_by_typed_identity() { + let c = conn(); + c.execute_batch( + "INSERT INTO playlist_local_tracks VALUES (NULL, 7, 42, 0, 1); + INSERT INTO playlist_plex_tracks VALUES (NULL, 7, 'rk-9', 1, 1);", + ) + .unwrap(); + let lp = crate::local_playlists::create(&c, "Local mix", None, false).unwrap(); + crate::local_playlists::add_tracks( + &c, + &lp, + &[ + crate::local_playlists::LocalPlaylistTrackInput::Qobuz(1000), + crate::local_playlists::LocalPlaylistTrackInput::Local("/music/a.flac".into()), + crate::local_playlists::LocalPlaylistTrackInput::Plex("rk-9".into()), + ], + ) + .unwrap(); + + let refs = [ + PlaylistTrackRef::Qobuz(1000), + PlaylistTrackRef::Library { + track_id: 42, + path: Some("/music/a.flac".into()), + }, + PlaylistTrackRef::Plex { + rating_key: "rk-9".into(), + }, + ]; + let rows = playlists_containing(&c, &refs).unwrap(); + // The Qobuz playlist holds the library row and the Plex key (2 of 3); + // the local playlist holds all three. + assert_eq!(contained(&rows, ContainmentTarget::Qobuz(7)), Some(2)); + assert_eq!(contained(&rows, ContainmentTarget::Local(lp.clone())), Some(3)); + assert_eq!(rows[0].target, ContainmentTarget::Local(lp)); + } + + #[test] + fn empty_refs_answer_nothing() { + let c = conn(); + assert!(playlists_containing(&c, &[]).unwrap().is_empty()); + } +} diff --git a/crates/qbz-library/src/qobuz_playlist_snapshot.rs b/crates/qbz-library/src/qobuz_playlist_snapshot.rs index 3934f2285..b46d00bfd 100644 --- a/crates/qbz-library/src/qobuz_playlist_snapshot.rs +++ b/crates/qbz-library/src/qobuz_playlist_snapshot.rs @@ -1,25 +1,35 @@ -//! Local snapshot of the user's QOBUZ playlists (offline-mode port, B7/B8). +//! Local snapshot of the user's QOBUZ playlists (offline-mode port, B7/B8; +//! membership index substrate, 2.1.1 Add-to-Playlist redesign). //! //! Spec D11 left an HONEST LIMIT: playlist names and membership live only in //! the Qobuz API, so offline a mixed playlist falls back to a synthesized //! "Playlist (N local)" name and shows zero Qobuz rows. This module stores a -//! point-in-time snapshot captured opportunistically from data the app -//! ALREADY fetches while online (no new API traffic): +//! point-in-time snapshot captured from data the app fetches while online: //! -//! - NAMES: every user-playlist list load (sidebar / playlist manager) -//! upserts id + name (+ owner, track_count) for ALL listed playlists — -//! cheap names-only rows, no track membership. +//! - HEADERS: every authoritative user-playlist list load (sidebar / playlist +//! manager) upserts id + name (+ owner, track_count, ownership) for ALL +//! listed playlists and advances the AUTHORITY GENERATION. A header absent +//! from two consecutive authoritative lists is marked `inactive` — never +//! deleted on a single miss, tolerating Qobuz post-write lag. //! - MEMBERSHIP: opening a playlist DETAIL online full-replaces its snapshot -//! track ids (the detail fetch already returns the full track list). -//! Membership is recorded ONLY for playlists already captured by the -//! names producer (the user's own list) — a merely-viewed public playlist -//! never lands in the snapshot, so the offline surfaces stay the user's. +//! track ids, and the background hydrator does the same from the +//! ids-only `playlist/get?extra=track_ids` fetch. Membership is recorded +//! ONLY for playlists already captured by a header producer (the user's own +//! list) — a merely-viewed public playlist never lands in the snapshot. +//! - MUTATIONS: successful add/remove/create through QBZ update the snapshot +//! incrementally and synchronously, so the picker's containment answer +//! moves without waiting for a re-sync. //! -//! Rows are point-in-time: offline consumers show them as-is (no staleness -//! UI in v1); `snapped_at` is stamped for the future. +//! Freshness is deliberately NOT one timestamp (that overload is exactly what +//! made an empty membership answer unreadable): `header_updated_at` says when +//! the list last named this playlist, `membership_synced_at` says when the +//! track ids were last authoritative, `membership_count` + +//! `membership_remote_updated_at` are the evidence a later list compares +//! against to decide staleness. `snapped_at` remains for the older offline +//! consumers and is stamped wherever it always was. //! -//! All functions take `&Connection` (the local_playlists idiom): no Tauri -//! state, no async runtime — testable with in-memory SQLite. +//! All functions take `&Connection` (the local_playlists idiom): no async +//! runtime — testable with in-memory SQLite. use std::collections::HashMap; @@ -31,20 +41,69 @@ pub struct SnapshotHeader { pub qobuz_playlist_id: u64, pub name: String, pub owner: Option, - /// The playlist's TOTAL Qobuz track count at snapshot time (not the + /// The playlist's TOTAL Qobuz track count at header time (not the /// offline-playable subset). pub track_count: Option, - /// Unix ms when this header was last written. + /// Unix ms when this header was last written (legacy stamp; kept for the + /// offline consumers that predate the split freshness columns). pub snapped_at: i64, + /// Unix ms when an authoritative list last named this playlist. + pub header_updated_at: Option, + /// Unix ms when the membership rows were last authoritative. NULL means + /// membership was never captured — which is NOT "empty playlist". + pub membership_synced_at: Option, + /// Number of membership rows written at the last sync. + pub membership_count: Option, + /// The playlist's API `updated_at` as reported by the authoritative list. + pub remote_updated_at: Option, + /// The API `updated_at` in force when membership was last synced. + pub membership_remote_updated_at: Option, + /// Last authority generation that listed this playlist. + pub seen_generation: i64, + /// The session user owns (can write to) this playlist. Persisted + /// explicitly instead of being re-inferred from whichever response + /// happened to be loaded. + pub is_owned: bool, + /// Absent from two consecutive authoritative lists. Kept, not deleted, so + /// a Qobuz-side hiccup cannot erase local knowledge. + pub inactive: bool, } -/// Names-producer input (one listed playlist). +/// Authoritative-list producer input (one listed playlist). #[derive(Debug, Clone)] -pub struct SnapshotNameEntry { +pub struct AuthoritativeEntry { pub qobuz_playlist_id: u64, pub name: String, pub owner: Option, pub track_count: Option, + pub remote_updated_at: Option, + pub is_owned: bool, +} + +/// Why the hydrator wants to refresh a playlist's membership. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StaleReason { + NeverSynced, + CountMismatch, + RemoteRevision, +} + +#[derive(Debug, Clone)] +pub struct HydrationCandidate { + pub qobuz_playlist_id: u64, + pub reason: StaleReason, +} + +/// The picker's honest tri-state for the containment section. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MembershipIndexState { + /// Every owned active playlist has authoritative membership. + Complete, + /// Some owned playlists still await a membership sync. + Updating { pending: u32 }, + /// No authoritative list was ever recorded (fresh profile, or the app has + /// never been online) — an empty containment answer means nothing. + Unavailable, } fn now_ms() -> i64 { @@ -55,8 +114,9 @@ fn now_ms() -> i64 { .as_millis() as i64 } -/// Create the snapshot tables. Idempotent (`IF NOT EXISTS`), run by -/// `LibraryDatabase::open` next to the rest of the schema. +/// Create the snapshot tables. Idempotent (`IF NOT EXISTS` + pragma-guarded +/// additive ALTERs, the database.rs idiom), run by `LibraryDatabase::open` +/// next to the rest of the schema. pub fn init_schema(conn: &Connection) -> Result<()> { conn.execute_batch( r#" @@ -76,80 +136,435 @@ pub fn init_schema(conn: &Connection) -> Result<()> { CREATE INDEX IF NOT EXISTS idx_qobuz_playlist_snapshot_tracks ON qobuz_playlist_snapshot_tracks(qobuz_playlist_id, position); - "#, - ) -} -/// NAMES producer: upsert a header row for every listed playlist. Never -/// touches the membership table. Stamps `snapped_at` = now on each row. -pub fn upsert_names(conn: &Connection, entries: &[SnapshotNameEntry]) -> Result<()> { - if entries.is_empty() { - return Ok(()); - } - let ts = now_ms(); - let mut stmt = conn.prepare( - "INSERT INTO qobuz_playlist_snapshot - (qobuz_playlist_id, name, owner, track_count, snapped_at) - VALUES (?1, ?2, ?3, ?4, ?5) - ON CONFLICT(qobuz_playlist_id) DO UPDATE SET - name = excluded.name, - owner = excluded.owner, - track_count = excluded.track_count, - snapped_at = excluded.snapped_at", + -- The picker's containment question runs in the opposite direction: + -- "which playlists hold THIS track". + CREATE INDEX IF NOT EXISTS idx_qobuz_playlist_snapshot_tracks_track + ON qobuz_playlist_snapshot_tracks(track_id, qobuz_playlist_id); + + CREATE TABLE IF NOT EXISTS qobuz_playlist_index_meta ( + key TEXT PRIMARY KEY, + value INTEGER NOT NULL + ); + "#, )?; - for e in entries { - stmt.execute(params![ - e.qobuz_playlist_id as i64, - e.name, - e.owner, - e.track_count, - ts - ])?; + // Additive migration (2.1.1 membership index): the split freshness + // columns. One probe guards the whole batch — they ship together. + let has_freshness: bool = conn + .query_row( + "SELECT COUNT(*) FROM pragma_table_info('qobuz_playlist_snapshot') + WHERE name = 'membership_synced_at'", + [], + |r| r.get::<_, i32>(0), + ) + .map(|count| count > 0) + .unwrap_or(false); + if !has_freshness { + conn.execute_batch( + "ALTER TABLE qobuz_playlist_snapshot ADD COLUMN header_updated_at INTEGER; + ALTER TABLE qobuz_playlist_snapshot ADD COLUMN membership_synced_at INTEGER; + ALTER TABLE qobuz_playlist_snapshot ADD COLUMN membership_count INTEGER; + ALTER TABLE qobuz_playlist_snapshot ADD COLUMN remote_updated_at INTEGER; + ALTER TABLE qobuz_playlist_snapshot ADD COLUMN membership_remote_updated_at INTEGER; + ALTER TABLE qobuz_playlist_snapshot ADD COLUMN seen_generation INTEGER NOT NULL DEFAULT 0; + ALTER TABLE qobuz_playlist_snapshot ADD COLUMN is_owned INTEGER NOT NULL DEFAULT 0; + ALTER TABLE qobuz_playlist_snapshot ADD COLUMN inactive INTEGER NOT NULL DEFAULT 0; + -- Backfill: the old single stamp was both header and membership + -- evidence; keep it as the header stamp everywhere, and as the + -- membership stamp only where membership rows actually exist. + UPDATE qobuz_playlist_snapshot + SET header_updated_at = snapped_at + WHERE header_updated_at IS NULL; + UPDATE qobuz_playlist_snapshot + SET membership_synced_at = snapped_at, + membership_count = ( + SELECT COUNT(*) FROM qobuz_playlist_snapshot_tracks t + WHERE t.qobuz_playlist_id + = qobuz_playlist_snapshot.qobuz_playlist_id) + WHERE membership_synced_at IS NULL + AND EXISTS ( + SELECT 1 FROM qobuz_playlist_snapshot_tracks t + WHERE t.qobuz_playlist_id + = qobuz_playlist_snapshot.qobuz_playlist_id);", + )?; } Ok(()) } +// ───────────────────────── Authority generations ───────────────────────── + +const GENERATION_KEY: &str = "authority_generation"; + +/// Last recorded authority generation; 0 when no authoritative list was ever +/// recorded. +pub fn current_generation(conn: &Connection) -> Result { + conn.query_row( + "SELECT value FROM qobuz_playlist_index_meta WHERE key = ?1", + params![GENERATION_KEY], + |r| r.get(0), + ) + .optional() + .map(|v| v.unwrap_or(0)) +} + +/// HEADERS producer: record one successful authoritative user-playlist list. +/// Upserts every entry, advances the authority generation, and marks headers +/// unseen for TWO consecutive generations `inactive` (the grace that +/// tolerates Qobuz post-write lag). Returns the new generation. +/// +/// An EMPTY list is still recorded (a user can genuinely delete every +/// playlist), but the caller must only invoke this for a fetch that +/// succeeded — a network failure must never masquerade as an empty library. +pub fn record_authoritative_list( + conn: &Connection, + entries: &[AuthoritativeEntry], +) -> Result { + let ts = now_ms(); + conn.execute_batch("BEGIN IMMEDIATE")?; + let result = (|| -> Result { + let generation = current_generation(conn)? + 1; + { + let mut stmt = conn.prepare( + "INSERT INTO qobuz_playlist_snapshot + (qobuz_playlist_id, name, owner, track_count, snapped_at, + header_updated_at, remote_updated_at, seen_generation, + is_owned, inactive) + VALUES (?1, ?2, ?3, ?4, ?5, ?5, ?6, ?7, ?8, 0) + ON CONFLICT(qobuz_playlist_id) DO UPDATE SET + name = excluded.name, + owner = excluded.owner, + track_count = excluded.track_count, + snapped_at = excluded.snapped_at, + header_updated_at = excluded.header_updated_at, + remote_updated_at = excluded.remote_updated_at, + seen_generation = excluded.seen_generation, + is_owned = excluded.is_owned, + inactive = 0", + )?; + for e in entries { + stmt.execute(params![ + e.qobuz_playlist_id as i64, + e.name, + e.owner, + e.track_count, + ts, + e.remote_updated_at, + generation, + e.is_owned, + ])?; + } + } + conn.execute( + "UPDATE qobuz_playlist_snapshot + SET inactive = 1 + WHERE inactive = 0 AND seen_generation <= ?1 - 2", + params![generation], + )?; + conn.execute( + "INSERT INTO qobuz_playlist_index_meta (key, value) VALUES (?1, ?2) + ON CONFLICT(key) DO UPDATE SET value = excluded.value", + params![GENERATION_KEY, generation], + )?; + Ok(generation) + })(); + match result { + Ok(generation) => { + conn.execute_batch("COMMIT")?; + Ok(generation) + } + Err(e) => { + let _ = conn.execute_batch("ROLLBACK"); + Err(e) + } + } +} + +// ───────────────────────── Membership producers ───────────────────────── + /// MEMBERSHIP producer: full-replace the snapshot track ids of ONE playlist -/// (detail load) and refresh its header (name / owner / track_count / -/// snapped_at). Returns `false` (writing NOTHING) when the playlist has no -/// header row — i.e. it was never captured by the names producer, so it is -/// not one of the user's listed playlists. +/// (detail load or hydrator fetch) and refresh its header. Stamps +/// `membership_synced_at` / `membership_count`, and records the API +/// `updated_at` in force (`remote_updated_at_evidence`) when the caller has +/// one. Returns `false` (writing NOTHING) when the playlist has no header +/// row — i.e. it was never captured by the headers producer, so it is not +/// one of the user's listed playlists. pub fn replace_tracks( conn: &Connection, qobuz_playlist_id: u64, name: &str, owner: Option<&str>, track_ids: &[u64], + remote_updated_at_evidence: Option, ) -> Result { - let updated = conn.execute( - "UPDATE qobuz_playlist_snapshot - SET name = ?2, owner = ?3, track_count = ?4, snapped_at = ?5 - WHERE qobuz_playlist_id = ?1", - params![ - qobuz_playlist_id as i64, - name, - owner, - track_ids.len() as u32, - now_ms() - ], - )?; - if updated == 0 { + let ts = now_ms(); + conn.execute_batch("BEGIN IMMEDIATE")?; + let result = (|| -> Result { + let updated = conn.execute( + "UPDATE qobuz_playlist_snapshot + SET name = ?2, owner = ?3, track_count = ?4, snapped_at = ?5, + membership_synced_at = ?5, membership_count = ?4, + membership_remote_updated_at = + COALESCE(?6, remote_updated_at) + WHERE qobuz_playlist_id = ?1", + params![ + qobuz_playlist_id as i64, + name, + owner, + track_ids.len() as u32, + ts, + remote_updated_at_evidence, + ], + )?; + if updated == 0 { + return Ok(false); + } + conn.execute( + "DELETE FROM qobuz_playlist_snapshot_tracks WHERE qobuz_playlist_id = ?1", + params![qobuz_playlist_id as i64], + )?; + let mut stmt = conn.prepare( + "INSERT INTO qobuz_playlist_snapshot_tracks (qobuz_playlist_id, position, track_id) + VALUES (?1, ?2, ?3)", + )?; + for (pos, tid) in track_ids.iter().enumerate() { + stmt.execute(params![qobuz_playlist_id as i64, pos as i64, *tid as i64])?; + } + Ok(true) + })(); + match result { + Ok(wrote) => { + conn.execute_batch("COMMIT")?; + Ok(wrote) + } + Err(e) => { + let _ = conn.execute_batch("ROLLBACK"); + Err(e) + } + } +} + +/// Incremental MUTATION producer: a successful `playlist/addTracks` appends +/// its ids to the snapshot so the containment answer moves immediately. +/// Applies only when membership was already captured (`membership_synced_at` +/// set) — an uncaptured playlist stays queued for the hydrator instead of +/// pretending an append equals a snapshot. Returns whether it applied. +pub fn apply_added_tracks( + conn: &Connection, + qobuz_playlist_id: u64, + track_ids: &[u64], +) -> Result { + if track_ids.is_empty() { return Ok(false); } - conn.execute( - "DELETE FROM qobuz_playlist_snapshot_tracks WHERE qobuz_playlist_id = ?1", - params![qobuz_playlist_id as i64], - )?; - let mut stmt = conn.prepare( - "INSERT INTO qobuz_playlist_snapshot_tracks (qobuz_playlist_id, position, track_id) - VALUES (?1, ?2, ?3)", - )?; - for (pos, tid) in track_ids.iter().enumerate() { - stmt.execute(params![qobuz_playlist_id as i64, pos as i64, *tid as i64])?; + let ts = now_ms(); + conn.execute_batch("BEGIN IMMEDIATE")?; + let result = (|| -> Result { + let captured: Option = conn + .query_row( + "SELECT membership_synced_at FROM qobuz_playlist_snapshot + WHERE qobuz_playlist_id = ?1", + params![qobuz_playlist_id as i64], + |r| r.get(0), + ) + .optional()? + .flatten(); + if captured.is_none() { + return Ok(false); + } + let next: i64 = conn.query_row( + "SELECT COALESCE(MAX(position) + 1, 0) + FROM qobuz_playlist_snapshot_tracks + WHERE qobuz_playlist_id = ?1", + params![qobuz_playlist_id as i64], + |r| r.get(0), + )?; + let mut stmt = conn.prepare( + "INSERT INTO qobuz_playlist_snapshot_tracks (qobuz_playlist_id, position, track_id) + VALUES (?1, ?2, ?3)", + )?; + for (offset, tid) in track_ids.iter().enumerate() { + stmt.execute(params![ + qobuz_playlist_id as i64, + next + offset as i64, + *tid as i64 + ])?; + } + conn.execute( + "UPDATE qobuz_playlist_snapshot + SET membership_count = ( + SELECT COUNT(*) FROM qobuz_playlist_snapshot_tracks t + WHERE t.qobuz_playlist_id = ?1), + track_count = ( + SELECT COUNT(*) FROM qobuz_playlist_snapshot_tracks t + WHERE t.qobuz_playlist_id = ?1), + membership_synced_at = ?2, snapped_at = ?2 + WHERE qobuz_playlist_id = ?1", + params![qobuz_playlist_id as i64, ts], + )?; + Ok(true) + })(); + match result { + Ok(applied) => { + conn.execute_batch("COMMIT")?; + Ok(applied) + } + Err(e) => { + let _ = conn.execute_batch("ROLLBACK"); + Err(e) + } } - Ok(true) } +/// Incremental MUTATION producer: a successful track removal drops every +/// membership row bearing the removed TRACK ids. Same capture precondition +/// as `apply_added_tracks`. +pub fn apply_removed_tracks( + conn: &Connection, + qobuz_playlist_id: u64, + track_ids: &[u64], +) -> Result { + if track_ids.is_empty() { + return Ok(false); + } + let ts = now_ms(); + conn.execute_batch("BEGIN IMMEDIATE")?; + let result = (|| -> Result { + let captured: Option = conn + .query_row( + "SELECT membership_synced_at FROM qobuz_playlist_snapshot + WHERE qobuz_playlist_id = ?1", + params![qobuz_playlist_id as i64], + |r| r.get(0), + ) + .optional()? + .flatten(); + if captured.is_none() { + return Ok(false); + } + { + let mut stmt = conn.prepare( + "DELETE FROM qobuz_playlist_snapshot_tracks + WHERE qobuz_playlist_id = ?1 AND track_id = ?2", + )?; + for tid in track_ids { + stmt.execute(params![qobuz_playlist_id as i64, *tid as i64])?; + } + } + conn.execute( + "UPDATE qobuz_playlist_snapshot + SET membership_count = ( + SELECT COUNT(*) FROM qobuz_playlist_snapshot_tracks t + WHERE t.qobuz_playlist_id = ?1), + track_count = ( + SELECT COUNT(*) FROM qobuz_playlist_snapshot_tracks t + WHERE t.qobuz_playlist_id = ?1), + membership_synced_at = ?2, snapped_at = ?2 + WHERE qobuz_playlist_id = ?1", + params![qobuz_playlist_id as i64, ts], + )?; + Ok(true) + })(); + match result { + Ok(applied) => { + conn.execute_batch("COMMIT")?; + Ok(applied) + } + Err(e) => { + let _ = conn.execute_batch("ROLLBACK"); + Err(e) + } + } +} + +/// MUTATION producer: a playlist just created through QBZ. Inserts its header +/// at the CURRENT generation (it cannot have been in the last list yet — the +/// grace rule, not a fresh fetch, is what keeps it alive) with an empty, +/// captured membership, so an immediate follow-up add applies incrementally. +pub fn record_created_playlist( + conn: &Connection, + qobuz_playlist_id: u64, + name: &str, + owner: Option<&str>, +) -> Result<()> { + let ts = now_ms(); + let generation = current_generation(conn)?; + conn.execute( + "INSERT INTO qobuz_playlist_snapshot + (qobuz_playlist_id, name, owner, track_count, snapped_at, + header_updated_at, membership_synced_at, membership_count, + seen_generation, is_owned, inactive) + VALUES (?1, ?2, ?3, 0, ?4, ?4, ?4, 0, ?5, 1, 0) + ON CONFLICT(qobuz_playlist_id) DO UPDATE SET + name = excluded.name, + owner = excluded.owner, + seen_generation = excluded.seen_generation, + is_owned = 1, + inactive = 0", + params![qobuz_playlist_id as i64, name, owner, ts, generation], + )?; + Ok(()) +} + +// ───────────────────────── Hydration planning ───────────────────────── + +/// The membership refreshes the hydrator still owes, oldest debt first: +/// never-synced playlists, then count mismatches, then remote-revision +/// mismatches. Only owned, active playlists — followed/read-only playlists +/// are informational and must not spend the rate budget. +pub fn hydration_queue(conn: &Connection) -> Result> { + let mut stmt = conn.prepare( + "SELECT qobuz_playlist_id, + membership_synced_at IS NULL AS never_synced, + membership_count IS NOT track_count AS count_mismatch, + (remote_updated_at IS NOT NULL + AND membership_remote_updated_at IS NOT remote_updated_at) + AS revision_mismatch + FROM qobuz_playlist_snapshot + WHERE is_owned = 1 AND inactive = 0 + AND (membership_synced_at IS NULL + OR membership_count IS NOT track_count + OR (remote_updated_at IS NOT NULL + AND membership_remote_updated_at IS NOT remote_updated_at)) + ORDER BY never_synced DESC, count_mismatch DESC, qobuz_playlist_id", + )?; + let mut out = Vec::new(); + for r in stmt.query_map([], |r| { + Ok(( + r.get::<_, i64>(0)? as u64, + r.get::<_, bool>(1)?, + r.get::<_, bool>(2)?, + )) + })? { + let (pid, never, count) = r?; + out.push(HydrationCandidate { + qobuz_playlist_id: pid, + reason: if never { + StaleReason::NeverSynced + } else if count { + StaleReason::CountMismatch + } else { + StaleReason::RemoteRevision + }, + }); + } + Ok(out) +} + +/// The containment section's honest tri-state. +pub fn index_state(conn: &Connection) -> Result { + if current_generation(conn)? == 0 { + return Ok(MembershipIndexState::Unavailable); + } + let pending = hydration_queue(conn)?.len() as u32; + Ok(if pending == 0 { + MembershipIndexState::Complete + } else { + MembershipIndexState::Updating { pending } + }) +} + +// ───────────────────────────── Readers ───────────────────────────── + fn row_to_header(r: &rusqlite::Row) -> Result { Ok(SnapshotHeader { qobuz_playlist_id: r.get::<_, i64>("qobuz_playlist_id")? as u64, @@ -157,26 +572,40 @@ fn row_to_header(r: &rusqlite::Row) -> Result { owner: r.get("owner")?, track_count: r.get("track_count")?, snapped_at: r.get("snapped_at")?, + header_updated_at: r.get("header_updated_at")?, + membership_synced_at: r.get("membership_synced_at")?, + membership_count: r.get("membership_count")?, + remote_updated_at: r.get("remote_updated_at")?, + membership_remote_updated_at: r.get("membership_remote_updated_at")?, + seen_generation: r.get("seen_generation")?, + is_owned: r.get("is_owned")?, + inactive: r.get("inactive")?, }) } +const HEADER_COLUMNS: &str = "qobuz_playlist_id, name, owner, track_count, snapped_at, \ + header_updated_at, membership_synced_at, membership_count, \ + remote_updated_at, membership_remote_updated_at, seen_generation, \ + is_owned, inactive"; + /// One snapshot header, or None. pub fn get_header(conn: &Connection, qobuz_playlist_id: u64) -> Result> { conn.query_row( - "SELECT qobuz_playlist_id, name, owner, track_count, snapped_at - FROM qobuz_playlist_snapshot WHERE qobuz_playlist_id = ?1", + &format!( + "SELECT {HEADER_COLUMNS} FROM qobuz_playlist_snapshot + WHERE qobuz_playlist_id = ?1" + ), params![qobuz_playlist_id as i64], row_to_header, ) .optional() } -/// All snapshot headers. +/// All snapshot headers, inactive ones included (offline consumers decide). pub fn all_headers(conn: &Connection) -> Result> { - let mut stmt = conn.prepare( - "SELECT qobuz_playlist_id, name, owner, track_count, snapped_at - FROM qobuz_playlist_snapshot", - )?; + let mut stmt = conn.prepare(&format!( + "SELECT {HEADER_COLUMNS} FROM qobuz_playlist_snapshot" + ))?; let mut out = Vec::new(); for r in stmt.query_map([], row_to_header)? { out.push(r?); @@ -226,20 +655,26 @@ mod tests { c } - fn name_entry(id: u64, name: &str, count: u32) -> SnapshotNameEntry { - SnapshotNameEntry { + fn entry(id: u64, name: &str, count: u32) -> AuthoritativeEntry { + AuthoritativeEntry { qobuz_playlist_id: id, name: name.to_string(), owner: Some("me".to_string()), track_count: Some(count), + remote_updated_at: None, + is_owned: true, } } + fn record(c: &Connection, entries: &[AuthoritativeEntry]) -> i64 { + record_authoritative_list(c, entries).unwrap() + } + #[test] fn roundtrip_header_and_tracks() { let c = conn(); - upsert_names(&c, &[name_entry(42, "Road Trip", 3)]).unwrap(); - let wrote = replace_tracks(&c, 42, "Road Trip", Some("me"), &[30, 10, 20]).unwrap(); + record(&c, &[entry(42, "Road Trip", 3)]); + let wrote = replace_tracks(&c, 42, "Road Trip", Some("me"), &[30, 10, 20], None).unwrap(); assert!(wrote); let h = get_header(&c, 42).unwrap().unwrap(); @@ -247,6 +682,10 @@ mod tests { assert_eq!(h.owner.as_deref(), Some("me")); assert_eq!(h.track_count, Some(3)); assert!(h.snapped_at > 0); + assert!(h.is_owned); + assert!(!h.inactive); + assert_eq!(h.membership_count, Some(3)); + assert!(h.membership_synced_at.is_some()); // Snapshot order preserved, not sorted by id. assert_eq!(track_ids(&c, 42).unwrap(), vec![30, 10, 20]); @@ -257,9 +696,9 @@ mod tests { #[test] fn replace_is_full_replace() { let c = conn(); - upsert_names(&c, &[name_entry(7, "Mix", 3)]).unwrap(); - replace_tracks(&c, 7, "Mix", None, &[1, 2, 3]).unwrap(); - replace_tracks(&c, 7, "Mix renamed", None, &[9]).unwrap(); + record(&c, &[entry(7, "Mix", 3)]); + replace_tracks(&c, 7, "Mix", None, &[1, 2, 3], None).unwrap(); + replace_tracks(&c, 7, "Mix renamed", None, &[9], None).unwrap(); assert_eq!(track_ids(&c, 7).unwrap(), vec![9]); let h = get_header(&c, 7).unwrap().unwrap(); @@ -279,28 +718,208 @@ mod tests { #[test] fn names_only_rows_without_tracks() { let c = conn(); - upsert_names(&c, &[name_entry(1, "A", 10), name_entry(2, "B", 0)]).unwrap(); + record(&c, &[entry(1, "A", 10), entry(2, "B", 0)]); assert_eq!(all_headers(&c).unwrap().len(), 2); assert!(track_ids(&c, 1).unwrap().is_empty()); assert!(all_track_ids(&c).unwrap().is_empty()); - // Re-upserting updates the name in place without creating track rows. - upsert_names(&c, &[name_entry(1, "A renamed", 11)]).unwrap(); + // Re-recording updates the name in place without creating track rows, + // and never touches the membership stamp. + record(&c, &[entry(1, "A renamed", 11), entry(2, "B", 0)]); let h = get_header(&c, 1).unwrap().unwrap(); assert_eq!(h.name, "A renamed"); assert_eq!(h.track_count, Some(11)); + assert!(h.membership_synced_at.is_none()); assert!(track_ids(&c, 1).unwrap().is_empty()); } #[test] fn replace_refuses_unknown_playlist() { let c = conn(); - // No names row -> the detail producer writes NOTHING (a merely + // No header row -> the detail producer writes NOTHING (a merely // viewed public playlist must not land in the snapshot). - let wrote = replace_tracks(&c, 99, "Someone's list", None, &[1, 2]).unwrap(); + let wrote = replace_tracks(&c, 99, "Someone's list", None, &[1, 2], None).unwrap(); assert!(!wrote); assert!(get_header(&c, 99).unwrap().is_none()); assert!(track_ids(&c, 99).unwrap().is_empty()); } + + #[test] + fn header_upsert_is_not_membership_evidence() { + // The 2.1.0 research finding: header refreshes used to renew the one + // timestamp that also vouched for membership. Now they must not. + let c = conn(); + record(&c, &[entry(5, "Mix", 2)]); + replace_tracks(&c, 5, "Mix", None, &[1, 2], None).unwrap(); + let synced = get_header(&c, 5).unwrap().unwrap().membership_synced_at; + + record(&c, &[entry(5, "Mix", 2)]); + let h = get_header(&c, 5).unwrap().unwrap(); + assert_eq!(h.membership_synced_at, synced); + assert_eq!(h.membership_count, Some(2)); + } + + #[test] + fn generations_and_two_generation_grace() { + let c = conn(); + assert_eq!(current_generation(&c).unwrap(), 0); + let g1 = record(&c, &[entry(1, "A", 0), entry(2, "B", 0)]); + assert_eq!(g1, 1); + + // B missing once: still active (post-write lag tolerance). + record(&c, &[entry(1, "A", 0)]); + assert!(!get_header(&c, 2).unwrap().unwrap().inactive); + + // B missing twice: inactive, but never deleted. + record(&c, &[entry(1, "A", 0)]); + let h = get_header(&c, 2).unwrap().unwrap(); + assert!(h.inactive); + + // B reappears: active again. + record(&c, &[entry(1, "A", 0), entry(2, "B", 0)]); + assert!(!get_header(&c, 2).unwrap().unwrap().inactive); + } + + #[test] + fn hydration_queue_reasons_and_order() { + let c = conn(); + let mut never = entry(1, "Never", 4); + never.remote_updated_at = Some(100); + let mut mismatch = entry(2, "Mismatch", 5); + mismatch.remote_updated_at = Some(100); + let mut revised = entry(3, "Revised", 1); + revised.remote_updated_at = Some(100); + let mut fresh = entry(4, "Fresh", 1); + fresh.remote_updated_at = Some(100); + let followed = AuthoritativeEntry { + is_owned: false, + ..entry(5, "Followed", 9) + }; + record( + &c, + &[ + never.clone(), + mismatch.clone(), + revised.clone(), + fresh.clone(), + followed, + ], + ); + // Membership for 2 (3 of 5 -> count mismatch), 3 (agrees, but synced + // under an older remote revision), 4 (agrees, current revision). + replace_tracks(&c, 2, "Mismatch", None, &[1, 2, 3], Some(100)).unwrap(); + replace_tracks(&c, 3, "Revised", None, &[7], Some(90)).unwrap(); + replace_tracks(&c, 4, "Fresh", None, &[8], Some(100)).unwrap(); + // The list reasserts the authoritative counts/revisions the entries + // declared (replace_tracks overwrote track_count with the membership + // size, which is exactly the drift the queue must notice). + record(&c, &[never, mismatch, revised, fresh]); + + let queue = hydration_queue(&c).unwrap(); + let ids: Vec = queue.iter().map(|q| q.qobuz_playlist_id).collect(); + assert_eq!(ids, vec![1, 2, 3]); + assert_eq!(queue[0].reason, StaleReason::NeverSynced); + assert_eq!(queue[1].reason, StaleReason::CountMismatch); + assert_eq!(queue[2].reason, StaleReason::RemoteRevision); + + match index_state(&c).unwrap() { + MembershipIndexState::Updating { pending } => assert_eq!(pending, 3), + other => panic!("expected Updating, got {other:?}"), + } + } + + #[test] + fn index_state_lifecycle() { + let c = conn(); + assert_eq!(index_state(&c).unwrap(), MembershipIndexState::Unavailable); + record(&c, &[entry(1, "A", 1)]); + assert!(matches!( + index_state(&c).unwrap(), + MembershipIndexState::Updating { pending: 1 } + )); + replace_tracks(&c, 1, "A", None, &[10], None).unwrap(); + assert_eq!(index_state(&c).unwrap(), MembershipIndexState::Complete); + } + + #[test] + fn incremental_add_and_remove() { + let c = conn(); + record(&c, &[entry(1, "A", 2), entry(2, "Uncaptured", 3)]); + replace_tracks(&c, 1, "A", None, &[10, 20], None).unwrap(); + + assert!(apply_added_tracks(&c, 1, &[30, 40]).unwrap()); + assert_eq!(track_ids(&c, 1).unwrap(), vec![10, 20, 30, 40]); + let h = get_header(&c, 1).unwrap().unwrap(); + assert_eq!(h.membership_count, Some(4)); + assert_eq!(h.track_count, Some(4)); + + assert!(apply_removed_tracks(&c, 1, &[20, 30]).unwrap()); + assert_eq!(track_ids(&c, 1).unwrap(), vec![10, 40]); + assert_eq!( + get_header(&c, 1).unwrap().unwrap().membership_count, + Some(2) + ); + + // A playlist whose membership was never captured refuses the + // increment — the hydrator owns it (its queue entry stays). + assert!(!apply_added_tracks(&c, 2, &[1]).unwrap()); + assert!(track_ids(&c, 2).unwrap().is_empty()); + assert!(hydration_queue(&c) + .unwrap() + .iter() + .any(|q| q.qobuz_playlist_id == 2)); + } + + #[test] + fn created_playlist_is_captured_and_survives_one_list_miss() { + let c = conn(); + record(&c, &[entry(1, "A", 0)]); + record_created_playlist(&c, 9, "Brand new", Some("me")).unwrap(); + + let h = get_header(&c, 9).unwrap().unwrap(); + assert!(h.is_owned); + assert_eq!(h.membership_count, Some(0)); + // Immediate add applies incrementally — no fetch needed. + assert!(apply_added_tracks(&c, 9, &[5]).unwrap()); + assert_eq!(track_ids(&c, 9).unwrap(), vec![5]); + + // One authoritative list that does not include it yet (Qobuz + // post-write lag): still active. + record(&c, &[entry(1, "A", 0)]); + assert!(!get_header(&c, 9).unwrap().unwrap().inactive); + } + + #[test] + fn legacy_schema_backfill() { + // A DB created by the pre-2.1.1 module: single-stamp schema. + let c = Connection::open_in_memory().unwrap(); + c.execute_batch( + "CREATE TABLE qobuz_playlist_snapshot ( + qobuz_playlist_id INTEGER PRIMARY KEY, + name TEXT NOT NULL, owner TEXT, track_count INTEGER, + snapped_at INTEGER NOT NULL); + CREATE TABLE qobuz_playlist_snapshot_tracks ( + qobuz_playlist_id INTEGER NOT NULL, + position INTEGER NOT NULL, + track_id INTEGER NOT NULL); + INSERT INTO qobuz_playlist_snapshot VALUES (1, 'With tracks', NULL, 2, 111); + INSERT INTO qobuz_playlist_snapshot VALUES (2, 'Names only', NULL, 7, 222); + INSERT INTO qobuz_playlist_snapshot_tracks VALUES (1, 0, 10), (1, 1, 20);", + ) + .unwrap(); + init_schema(&c).unwrap(); + + let with_tracks = get_header(&c, 1).unwrap().unwrap(); + assert_eq!(with_tracks.header_updated_at, Some(111)); + assert_eq!(with_tracks.membership_synced_at, Some(111)); + assert_eq!(with_tracks.membership_count, Some(2)); + + let names_only = get_header(&c, 2).unwrap().unwrap(); + assert_eq!(names_only.header_updated_at, Some(222)); + assert!(names_only.membership_synced_at.is_none()); + + // Idempotent re-run. + init_schema(&c).unwrap(); + } }