feat(library): playlist membership index substrate
Split the snapshot's single timestamp into honest freshness columns (header_updated_at vs membership_synced_at, plus membership_count and remote revision evidence), add authoritative-list generations with a two-generation retirement grace, persist ownership explicitly, and add incremental add/remove/create producers so successful mutations move the snapshot synchronously. New playlist_membership module answers the inverse question — which writable playlists already hold these tracks — over the snapshot, both sidecar tables and local playlists, through a typed PlaylistTrackRef instead of the plex:/id string mangling. All membership tables gain reverse-direction indexes; migrations are probe-then-ALTER with a legacy backfill test. cc
This commit is contained in:
5 files changed
+1214
-94
No files matched your search
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:<key>"` / `"<i64>"` 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<String>,
|
||||
},
|
||||
/// 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<Vec<PlaylistContainment>> {
|
||||
// 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<ContainmentTarget, HashSet<usize>> = 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<PlaylistContainment> = 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<snapshot::AuthoritativeEntry> = 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<u32> {
|
||||
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());
|
||||
}
|
||||
}
|
||||
@@ -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<String>,
|
||||
/// 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<u32>,
|
||||
/// 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<i64>,
|
||||
/// 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<i64>,
|
||||
/// Number of membership rows written at the last sync.
|
||||
pub membership_count: Option<u32>,
|
||||
/// The playlist's API `updated_at` as reported by the authoritative list.
|
||||
pub remote_updated_at: Option<i64>,
|
||||
/// The API `updated_at` in force when membership was last synced.
|
||||
pub membership_remote_updated_at: Option<i64>,
|
||||
/// 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<String>,
|
||||
pub track_count: Option<u32>,
|
||||
pub remote_updated_at: Option<i64>,
|
||||
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<i64> {
|
||||
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<i64> {
|
||||
let ts = now_ms();
|
||||
conn.execute_batch("BEGIN IMMEDIATE")?;
|
||||
let result = (|| -> Result<i64> {
|
||||
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<i64>,
|
||||
) -> Result<bool> {
|
||||
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<bool> {
|
||||
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<bool> {
|
||||
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<bool> {
|
||||
let captured: Option<i64> = 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<bool> {
|
||||
if track_ids.is_empty() {
|
||||
return Ok(false);
|
||||
}
|
||||
let ts = now_ms();
|
||||
conn.execute_batch("BEGIN IMMEDIATE")?;
|
||||
let result = (|| -> Result<bool> {
|
||||
let captured: Option<i64> = 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<Vec<HydrationCandidate>> {
|
||||
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<MembershipIndexState> {
|
||||
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<SnapshotHeader> {
|
||||
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<SnapshotHeader> {
|
||||
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<Option<SnapshotHeader>> {
|
||||
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<Vec<SnapshotHeader>> {
|
||||
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<u64> = 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();
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user