feat(library): add derived local catalog

Introduce a frontend-agnostic generation schema with stable source identities, materialized album and artist structures, trigram FTS, matching sort indices, and descriptor-bound keyset queries.

Add deterministic SQLite coverage and an explicit one-million-track query gate.

cx
This commit is contained in:
vicrodh committed 2026-08-23 06:39:00 -06:00
1 parent 1edae11e52
commit a0d8a7fc94
8 files changed
+2033

No files matched your search

+10
View File
@@ -6535,6 +6535,16 @@ dependencies = [
"walkdir",
]
[[package]]
name = "qbz-local-catalog"
version = "0.1.0"
dependencies = [
"rusqlite",
"serde",
"tempfile",
"thiserror 2.0.18",
]
[[package]]
name = "qbz-log"
version = "2.0.2"
+3
View File
@@ -13,6 +13,9 @@ members = [
"qbz-integrations",
"qbz-cache",
"qbz-library",
# Frontend-agnostic, derived Local Library read projection. It owns no
# source APIs and no Qt types; source caches project into this boundary.
"qbz-local-catalog",
"qbz-cast",
"qbz-cmaf",
"qbz-dsd",
+14
View File
@@ -0,0 +1,14 @@
[package]
name = "qbz-local-catalog"
version = "0.1.0"
edition.workspace = true
license.workspace = true
repository.workspace = true
[dependencies]
rusqlite = { version = "0.31", features = ["bundled"] }
serde.workspace = true
thiserror.workspace = true
[dev-dependencies]
tempfile = "3"
+837
View File
@@ -0,0 +1,837 @@
use std::collections::HashSet;
use std::path::Path;
use std::time::{Duration, Instant};
use rusqlite::types::Value;
use rusqlite::{params, params_from_iter, Connection, OptionalExtension, Row, Statement};
use crate::model::{
ProjectedTrack, QueryDescriptor, SourceKey, SourceKind, TrackCursor, TrackGroup, TrackPage,
TrackRecord, TrackRef, TrackSort,
};
use crate::schema;
use crate::{CatalogError, Result, SCHEMA_VERSION};
const MAX_PAGE_SIZE: usize = 500;
pub struct Catalog {
conn: Connection,
generation: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CatalogStats {
pub schema_version: u32,
pub generation: u64,
pub track_count: u64,
pub source_counts: Vec<(SourceKey, u64)>,
pub page_size_bytes: u64,
pub page_count: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IntegrityReport {
pub sqlite_ok: bool,
pub foreign_key_violations: u64,
pub fts_ok: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct QueryMetrics {
pub sql_time: Duration,
pub rows: usize,
}
struct RowWithCursor {
record: TrackRecord,
cursor: TrackCursor,
}
impl Catalog {
pub fn open(path: &Path, generation: u64) -> Result<Self> {
let mut conn = Connection::open(path)?;
schema::init(&mut conn, generation)?;
Ok(Self { conn, generation })
}
pub fn open_in_memory(generation: u64) -> Result<Self> {
let mut conn = Connection::open_in_memory()?;
schema::init(&mut conn, generation)?;
Ok(Self { conn, generation })
}
pub fn generation(&self) -> u64 {
self.generation
}
/// Idempotent batch projection. Source-native identity owns the upsert;
/// the internal catalog rowid is retained and never becomes product state.
pub fn upsert_tracks(&mut self, tracks: &[ProjectedTrack]) -> Result<usize> {
let tx = self.conn.transaction()?;
{
let mut upsert = tx.prepare_cached(UPSERT_TRACK_SQL)?;
let mut clear_credits =
tx.prepare_cached("DELETE FROM artist_credits WHERE catalog_id = ?1")?;
let mut insert_credit = tx.prepare_cached(
"INSERT OR IGNORE INTO artist_credits(
catalog_id, artist_key, display_name, role, ordinal
) VALUES (?1,?2,?3,?4,?5)",
)?;
for track in tracks {
validate_track(track)?;
upsert_track(&mut upsert, &mut clear_credits, &mut insert_credit, track)?;
}
}
tx.commit()?;
Ok(tracks.len())
}
pub fn remove_track(&mut self, track_ref: &TrackRef) -> Result<bool> {
validate_ref(track_ref)?;
Ok(self.conn.execute(
"DELETE FROM tracks
WHERE source_kind = ?1 AND source_instance = ?2 AND native_track_id = ?3",
params![
track_ref.source.as_str(),
track_ref.source_instance,
track_ref.native_id
],
)? > 0)
}
pub fn resolve(&self, track_ref: &TrackRef) -> Result<Option<TrackRecord>> {
validate_ref(track_ref)?;
let sql = format!(
"SELECT {TRACK_COLUMNS}
FROM tracks t
WHERE t.source_kind = ?1
AND t.source_instance = ?2
AND t.native_track_id = ?3"
);
self.conn
.query_row(
&sql,
params![
track_ref.source.as_str(),
track_ref.source_instance,
track_ref.native_id
],
map_row,
)
.optional()
.map(|row| row.map(|value| value.record))
.map_err(Into::into)
}
pub fn count_tracks(&self, descriptor: &QueryDescriptor) -> Result<u64> {
validate_tracks_descriptor(descriptor)?;
let (sql, values) = count_query(descriptor)?;
let count: i64 = self
.conn
.query_row(&sql, params_from_iter(values), |row| row.get(0))?;
Ok(count as u64)
}
pub fn query_tracks(
&self,
descriptor: &QueryDescriptor,
cursor: Option<&TrackCursor>,
page_size: usize,
) -> Result<TrackPage> {
self.query_tracks_timed(descriptor, cursor, page_size)
.map(|(page, _)| page)
}
pub fn query_tracks_timed(
&self,
descriptor: &QueryDescriptor,
cursor: Option<&TrackCursor>,
page_size: usize,
) -> Result<(TrackPage, QueryMetrics)> {
validate_tracks_descriptor(descriptor)?;
let limit = page_size.clamp(1, MAX_PAGE_SIZE);
let (sql, values, sort) = page_query(descriptor, cursor, limit)?;
let descriptor_key = descriptor_key(descriptor);
let started = Instant::now();
let mut stmt = self.conn.prepare_cached(&sql)?;
let mapped = stmt.query_map(params_from_iter(values), map_row)?;
let mut values = mapped.collect::<rusqlite::Result<Vec<_>>>()?;
for value in &mut values {
value.cursor.sort = sort;
value.cursor.descriptor_key = descriptor_key.clone();
}
let sql_time = started.elapsed();
let has_more = values.len() > limit;
values.truncate(limit);
let next_cursor = has_more.then(|| values.last().expect("non-empty page").cursor.clone());
let rows = values
.into_iter()
.map(|value| value.record)
.collect::<Vec<_>>();
let metrics = QueryMetrics {
sql_time,
rows: rows.len(),
};
Ok((
TrackPage {
rows,
next_cursor,
has_more,
},
metrics,
))
}
pub fn stats(&self) -> Result<CatalogStats> {
let track_count: i64 = self
.conn
.query_row("SELECT COUNT(*) FROM tracks", [], |row| row.get(0))?;
let mut stmt = self.conn.prepare(
"SELECT source_kind, source_instance, COUNT(*)
FROM tracks
GROUP BY source_kind, source_instance
ORDER BY source_kind, source_instance",
)?;
let rows = stmt.query_map([], |row| {
let word: String = row.get(0)?;
Ok((
SourceKey {
source: SourceKind::from_str(&word).expect("schema source CHECK"),
source_instance: row.get(1)?,
},
row.get::<_, i64>(2)? as u64,
))
})?;
let source_counts = rows.collect::<rusqlite::Result<Vec<_>>>()?;
let page_size_bytes: i64 = self
.conn
.query_row("PRAGMA page_size", [], |row| row.get(0))?;
let page_count: i64 = self
.conn
.query_row("PRAGMA page_count", [], |row| row.get(0))?;
Ok(CatalogStats {
schema_version: SCHEMA_VERSION,
generation: self.generation,
track_count: track_count as u64,
source_counts,
page_size_bytes: page_size_bytes as u64,
page_count: page_count as u64,
})
}
pub fn integrity_check(&self) -> Result<IntegrityReport> {
let result: String = self
.conn
.query_row("PRAGMA quick_check", [], |row| row.get(0))?;
let foreign_key_violations: i64 =
self.conn
.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| {
row.get(0)
})?;
let fts_ok = self
.conn
.execute(
"INSERT INTO tracks_fts(tracks_fts) VALUES ('integrity-check')",
[],
)
.is_ok();
Ok(IntegrityReport {
sqlite_ok: result == "ok",
foreign_key_violations: foreign_key_violations as u64,
fts_ok,
})
}
#[cfg(test)]
pub(crate) fn connection(&self) -> &Connection {
&self.conn
}
}
fn validate_ref(track_ref: &TrackRef) -> Result<()> {
if track_ref.source_instance.trim().is_empty() {
return Err(CatalogError::InvalidInput(
"source instance must not be empty".to_string(),
));
}
if track_ref.native_id.trim().is_empty() {
return Err(CatalogError::InvalidInput(
"native track id must not be empty".to_string(),
));
}
Ok(())
}
fn validate_track(track: &ProjectedTrack) -> Result<()> {
validate_ref(&track.track_ref)?;
if track.duration_ms > i64::MAX as u64 {
return Err(CatalogError::InvalidInput(
"duration does not fit SQLite INTEGER".to_string(),
));
}
Ok(())
}
fn validate_tracks_descriptor(descriptor: &QueryDescriptor) -> Result<()> {
if descriptor.surface() != crate::QuerySurface::Tracks {
return Err(CatalogError::InvalidInput(
"a non-Tracks descriptor was passed to a Tracks query".to_string(),
));
}
Ok(())
}
fn upsert_track(
upsert: &mut Statement<'_>,
clear_credits: &mut Statement<'_>,
insert_credit: &mut Statement<'_>,
track: &ProjectedTrack,
) -> Result<i64> {
let sort_title = normalize_sort_key(&track.title);
let sort_artist = normalize_sort_key(if track.album_artist.trim().is_empty() {
&track.artist
} else {
&track.album_artist
});
let sort_album = normalize_sort_key(&track.album);
let year_missing = i64::from(track.year.is_none());
let year_value = track.year.unwrap_or(0) as i64;
let disc_sort = track.disc_number.unwrap_or(0) as i64;
let track_sort = track.track_number.unwrap_or(0) as i64;
let mut seen_credits = HashSet::new();
let credits = track
.credits
.iter()
.filter_map(|credit| {
let name = credit.display_name.trim();
(!name.is_empty() && seen_credits.insert(name.to_lowercase())).then(|| name.to_string())
})
.collect::<Vec<_>>()
.join(" \u{1f} ");
let catalog_id: i64 = upsert.query_row(
params![
track.track_ref.source.as_str(),
track.track_ref.source_instance,
track.track_ref.native_id,
track.local_track_id,
track.local_path,
track.source_copy_id,
track.title,
sort_title,
track.artist,
sort_artist,
track.album_artist,
track.album,
sort_album,
credits,
track.duration_ms as i64,
track.year.map(i64::from),
year_missing,
year_value,
track.disc_number.map(i64::from),
disc_sort,
track.track_number.map(i64::from),
track_sort,
track.format.trim().to_ascii_lowercase(),
track.bit_depth.map(i64::from),
track.sample_rate_hz.map(i64::from),
track.artwork_token,
track.isrc,
track.musicbrainz_recording_id,
track.added_at,
i64::from(track.available),
track.observed_generation,
],
|row| row.get(0),
)?;
clear_credits.execute(params![catalog_id])?;
for credit in &track.credits {
let artist_key = normalize_artist_key(&credit.display_name);
if artist_key.is_empty() {
continue;
}
insert_credit.execute(params![
catalog_id,
artist_key,
credit.display_name.trim(),
credit.role.as_str(),
i64::from(credit.ordinal),
])?;
}
Ok(catalog_id)
}
const UPSERT_TRACK_SQL: &str = "INSERT INTO tracks (
source_kind, source_instance, native_track_id, local_track_id, local_path,
source_copy_id, title, sort_title, artist, sort_artist, album_artist,
album, sort_album, credits, duration_ms, year, year_missing, year_value,
disc_number, disc_sort, track_number, track_sort, format, bit_depth,
sample_rate_hz, artwork_token, isrc, musicbrainz_recording_id, added_at,
available, last_observed_generation
) VALUES (
?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,
?19,?20,?21,?22,?23,?24,?25,?26,?27,?28,?29,?30,?31
)
ON CONFLICT(source_kind, source_instance, native_track_id) DO UPDATE SET
local_track_id=excluded.local_track_id,
local_path=excluded.local_path,
source_copy_id=excluded.source_copy_id,
title=excluded.title,
sort_title=excluded.sort_title,
artist=excluded.artist,
sort_artist=excluded.sort_artist,
album_artist=excluded.album_artist,
album=excluded.album,
sort_album=excluded.sort_album,
credits=excluded.credits,
duration_ms=excluded.duration_ms,
year=excluded.year,
year_missing=excluded.year_missing,
year_value=excluded.year_value,
disc_number=excluded.disc_number,
disc_sort=excluded.disc_sort,
track_number=excluded.track_number,
track_sort=excluded.track_sort,
format=excluded.format,
bit_depth=excluded.bit_depth,
sample_rate_hz=excluded.sample_rate_hz,
artwork_token=excluded.artwork_token,
isrc=excluded.isrc,
musicbrainz_recording_id=excluded.musicbrainz_recording_id,
added_at=excluded.added_at,
available=excluded.available,
last_observed_generation=excluded.last_observed_generation
RETURNING catalog_id";
struct QueryParts {
where_sql: String,
params: Vec<Value>,
sort: TrackSort,
}
fn filter_parts(
descriptor: &QueryDescriptor,
cursor: Option<&TrackCursor>,
alias: &str,
) -> Result<QueryParts> {
let sort = effective_sort(descriptor);
if let Some(cursor) = cursor {
if cursor.sort != sort || cursor.descriptor_key != descriptor_key(descriptor) {
return Err(CatalogError::CursorDescriptorMismatch);
}
}
let mut predicates = Vec::new();
let mut values = Vec::new();
if descriptor.available_only() {
predicates.push(format!("{alias}.available = ?"));
values.push(Value::Integer(1));
}
if !descriptor.sources().is_empty() {
let mut source_predicates = Vec::new();
for source in descriptor.sources() {
if source.source_instance.trim().is_empty() {
return Err(CatalogError::InvalidInput(
"source filter instance must not be empty".to_string(),
));
}
source_predicates.push(format!(
"({alias}.source_kind = ? AND {alias}.source_instance = ?)"
));
values.push(Value::Text(source.source.as_str().to_string()));
values.push(Value::Text(source.source_instance.clone()));
}
predicates.push(format!("({})", source_predicates.join(" OR ")));
}
if !descriptor.formats().is_empty() {
predicates.push(format!(
"{alias}.format IN ({})",
std::iter::repeat("?")
.take(descriptor.formats().len())
.collect::<Vec<_>>()
.join(",")
));
values.extend(descriptor.formats().iter().cloned().map(Value::Text));
}
if let Some(cursor) = cursor {
let (predicate, cursor_values) = cursor_predicate(cursor, alias);
predicates.push(predicate);
values.extend(cursor_values);
}
if predicates.is_empty() {
predicates.push("1".to_string());
}
Ok(QueryParts {
where_sql: predicates.join(" AND "),
params: values,
sort,
})
}
fn count_query(descriptor: &QueryDescriptor) -> Result<(String, Vec<Value>)> {
if descriptor.search().is_empty() {
let parts = filter_parts(descriptor, None, "t")?;
return Ok((
format!("SELECT COUNT(*) FROM tracks t WHERE {}", parts.where_sql),
parts.params,
));
}
let match_value = fts_match_value(descriptor.search())?;
let parts = filter_parts(descriptor, None, "t")?;
let sql = format!(
"SELECT COUNT(*)
FROM tracks_fts
CROSS JOIN tracks t NOT INDEXED
WHERE tracks_fts MATCH ?
AND t.catalog_id = tracks_fts.rowid
AND {}",
parts.where_sql
);
let mut values = vec![match_value];
values.extend(parts.params);
Ok((sql, values))
}
fn page_query(
descriptor: &QueryDescriptor,
cursor: Option<&TrackCursor>,
limit: usize,
) -> Result<(String, Vec<Value>, TrackSort)> {
if descriptor.search().is_empty() {
let parts = filter_parts(descriptor, cursor, "t")?;
let sql = format!(
"SELECT {TRACK_COLUMNS}
FROM tracks t
WHERE {}
ORDER BY {}
LIMIT {}",
parts.where_sql,
order_clause(parts.sort, "t"),
limit + 1
);
return Ok((sql, parts.params, parts.sort));
}
let match_value = fts_match_value(descriptor.search())?;
let parts = filter_parts(descriptor, cursor, "t")?;
let sql = format!(
"SELECT {TRACK_COLUMNS}
FROM tracks_fts
CROSS JOIN tracks t NOT INDEXED
WHERE tracks_fts MATCH ?
AND t.catalog_id = tracks_fts.rowid
AND {}
ORDER BY {}
LIMIT {}",
parts.where_sql,
order_clause(parts.sort, "t"),
limit + 1,
);
let mut values = vec![match_value];
values.extend(parts.params);
Ok((sql, values, parts.sort))
}
fn fts_match_value(search: &str) -> Result<Value> {
if search.chars().count() < 3 {
return Err(CatalogError::SearchTooShort);
}
let escaped = search.replace('"', "\"\"");
Ok(Value::Text(format!("\"{escaped}\"")))
}
fn effective_sort(descriptor: &QueryDescriptor) -> TrackSort {
match descriptor.group() {
TrackGroup::Album => TrackSort::Default,
TrackGroup::Artist => TrackSort::ArtistAsc,
TrackGroup::Name => TrackSort::TitleAsc,
TrackGroup::Off => descriptor.sort(),
}
}
fn order_clause(sort: TrackSort, alias: &str) -> String {
let columns = match sort {
TrackSort::Default => {
"sort_album, sort_artist, disc_sort, track_sort, sort_title, catalog_id"
}
TrackSort::TitleAsc => "sort_title, sort_artist, catalog_id",
TrackSort::TitleDesc => "sort_title DESC, sort_artist, catalog_id",
TrackSort::ArtistAsc => "sort_artist, sort_album, disc_sort, track_sort, catalog_id",
TrackSort::ArtistDesc => "sort_artist DESC, sort_album, disc_sort, track_sort, catalog_id",
TrackSort::YearAsc => {
"year_missing, year_value, sort_album, disc_sort, track_sort, catalog_id"
}
TrackSort::YearDesc => {
"year_missing, year_value DESC, sort_album, disc_sort, track_sort, catalog_id"
}
TrackSort::AddedDesc => "added_at DESC, sort_album, disc_sort, track_sort, catalog_id",
};
columns
.split(", ")
.map(|column| {
let (name, suffix) = column.split_once(' ').unwrap_or((column, ""));
if suffix.is_empty() {
format!("{alias}.{name}")
} else {
format!("{alias}.{name} {suffix}")
}
})
.collect::<Vec<_>>()
.join(", ")
}
fn cursor_predicate(cursor: &TrackCursor, alias: &str) -> (String, Vec<Value>) {
let text = |value: &str| Value::Text(value.to_string());
let integer = Value::Integer;
let (predicate, values) = match cursor.sort {
TrackSort::Default => (
"(t.sort_album,t.sort_artist,t.disc_sort,t.track_sort,t.sort_title,t.catalog_id) > (?,?,?,?,?,?)",
vec![
text(&cursor.sort_album),
text(&cursor.sort_artist),
integer(cursor.disc_sort),
integer(cursor.track_sort),
text(&cursor.sort_title),
integer(cursor.row_id),
],
),
TrackSort::TitleAsc => (
"(t.sort_title,t.sort_artist,t.catalog_id) > (?,?,?)",
vec![
text(&cursor.sort_title),
text(&cursor.sort_artist),
integer(cursor.row_id),
],
),
TrackSort::TitleDesc => (
"(t.sort_title < ? OR (t.sort_title = ? AND (t.sort_artist,t.catalog_id) > (?,?)))",
vec![
text(&cursor.sort_title),
text(&cursor.sort_title),
text(&cursor.sort_artist),
integer(cursor.row_id),
],
),
TrackSort::ArtistAsc => (
"(t.sort_artist,t.sort_album,t.disc_sort,t.track_sort,t.catalog_id) > (?,?,?,?,?)",
vec![
text(&cursor.sort_artist),
text(&cursor.sort_album),
integer(cursor.disc_sort),
integer(cursor.track_sort),
integer(cursor.row_id),
],
),
TrackSort::ArtistDesc => (
"(t.sort_artist < ? OR (t.sort_artist = ? AND (t.sort_album,t.disc_sort,t.track_sort,t.catalog_id) > (?,?,?,?)))",
vec![
text(&cursor.sort_artist),
text(&cursor.sort_artist),
text(&cursor.sort_album),
integer(cursor.disc_sort),
integer(cursor.track_sort),
integer(cursor.row_id),
],
),
TrackSort::YearAsc => (
"(t.year_missing,t.year_value,t.sort_album,t.disc_sort,t.track_sort,t.catalog_id) > (?,?,?,?,?,?)",
vec![
integer(cursor.year_missing),
integer(cursor.year_value),
text(&cursor.sort_album),
integer(cursor.disc_sort),
integer(cursor.track_sort),
integer(cursor.row_id),
],
),
TrackSort::YearDesc => (
"(t.year_missing > ? OR (t.year_missing = ? AND (t.year_value < ? OR (t.year_value = ? AND (t.sort_album,t.disc_sort,t.track_sort,t.catalog_id) > (?,?,?,?)))))",
vec![
integer(cursor.year_missing),
integer(cursor.year_missing),
integer(cursor.year_value),
integer(cursor.year_value),
text(&cursor.sort_album),
integer(cursor.disc_sort),
integer(cursor.track_sort),
integer(cursor.row_id),
],
),
TrackSort::AddedDesc => (
"(t.added_at < ? OR (t.added_at = ? AND (t.sort_album,t.disc_sort,t.track_sort,t.catalog_id) > (?,?,?,?)))",
vec![
integer(cursor.added_at),
integer(cursor.added_at),
text(&cursor.sort_album),
integer(cursor.disc_sort),
integer(cursor.track_sort),
integer(cursor.row_id),
],
),
};
(predicate.replace("t.", &format!("{alias}.")), values)
}
fn map_row(row: &Row<'_>) -> rusqlite::Result<RowWithCursor> {
let source_word: String = row.get(0)?;
let source = SourceKind::from_str(&source_word).expect("schema source CHECK");
let sort_title = row.get(18)?;
let sort_artist = row.get(19)?;
let sort_album = row.get(20)?;
let year_missing = row.get(21)?;
let year_value = row.get(22)?;
let disc_sort = row.get(23)?;
let track_sort = row.get(24)?;
let added_at = row.get(25)?;
let row_id = row.get(26)?;
Ok(RowWithCursor {
record: TrackRecord {
track_ref: TrackRef {
source,
source_instance: row.get(1)?,
native_id: row.get(2)?,
},
local_track_id: row.get(3)?,
local_path: row.get(4)?,
title: row.get(5)?,
artist: row.get(6)?,
album_artist: row.get(7)?,
album: row.get(8)?,
duration_ms: row.get::<_, i64>(9)?.max(0) as u64,
year: row.get::<_, Option<i64>>(10)?.map(|value| value as u32),
disc_number: row.get::<_, Option<i64>>(11)?.map(|value| value as u32),
track_number: row.get::<_, Option<i64>>(12)?.map(|value| value as u32),
format: row.get(13)?,
bit_depth: row.get::<_, Option<i64>>(14)?.map(|value| value as u32),
sample_rate_hz: row.get::<_, Option<i64>>(15)?.map(|value| value as u32),
artwork_token: row.get(16)?,
available: row.get::<_, i64>(17)? != 0,
},
cursor: TrackCursor {
// Filled by query_tracks_timed after the effective sort is known.
sort: TrackSort::Default,
descriptor_key: String::new(),
sort_title,
sort_artist,
sort_album,
year_missing,
year_value,
disc_sort,
track_sort,
added_at,
row_id,
},
})
}
const TRACK_COLUMNS: &str = "
t.source_kind, t.source_instance, t.native_track_id, t.local_track_id, t.local_path,
t.title, t.artist, t.album_artist, t.album, t.duration_ms, t.year, t.disc_number,
t.track_number, t.format, t.bit_depth, t.sample_rate_hz, t.artwork_token, t.available,
t.sort_title, t.sort_artist, t.sort_album, t.year_missing, t.year_value, t.disc_sort,
t.track_sort, t.added_at, t.catalog_id";
/// Unicode-aware display sort key shared by ingest and query indices.
pub fn normalize_sort_key(value: &str) -> String {
value
.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
.to_lowercase()
}
/// Stable artist key: lowercased, common Latin diacritics folded, and runs of
/// punctuation collapsed while preserving non-Latin alphanumerics.
pub fn normalize_artist_key(value: &str) -> String {
let mut out = String::with_capacity(value.len());
let mut pending_space = false;
for character in value.to_lowercase().chars() {
let folded = fold_diacritic(character);
if folded.is_alphanumeric() {
if pending_space && !out.is_empty() {
out.push(' ');
}
out.push(folded);
pending_space = false;
} else {
pending_space = true;
}
}
out
}
fn fold_diacritic(character: char) -> char {
match character {
'á' | 'à' | 'â' | 'ä' | 'ã' | 'å' | 'ā' => 'a',
'é' | 'è' | 'ê' | 'ë' | 'ē' | 'ė' => 'e',
'í' | 'ì' | 'î' | 'ï' | 'ī' => 'i',
'ó' | 'ò' | 'ô' | 'ö' | 'õ' | 'ō' | 'ø' => 'o',
'ú' | 'ù' | 'û' | 'ü' | 'ū' => 'u',
'ñ' => 'n',
'ç' => 'c',
'ý' | 'ÿ' => 'y',
'ß' => 's',
_ => character,
}
}
pub(crate) fn descriptor_key(descriptor: &QueryDescriptor) -> String {
let mut key = String::new();
push_key_part(
&mut key,
match descriptor.surface() {
crate::QuerySurface::Tracks => "tracks",
crate::QuerySurface::Albums => "albums",
crate::QuerySurface::Artists => "artists",
},
);
push_key_part(&mut key, descriptor.search());
push_key_part(
&mut key,
if descriptor.available_only() {
"1"
} else {
"0"
},
);
push_key_part(&mut key, sort_word(effective_sort(descriptor)));
push_key_part(
&mut key,
match descriptor.group() {
TrackGroup::Off => "off",
TrackGroup::Album => "album",
TrackGroup::Artist => "artist",
TrackGroup::Name => "name",
},
);
for source in descriptor.sources() {
push_key_part(&mut key, source.source.as_str());
push_key_part(&mut key, &source.source_instance);
}
for format in descriptor.formats() {
push_key_part(&mut key, format);
}
key
}
fn push_key_part(key: &mut String, value: &str) {
key.push_str(&value.len().to_string());
key.push(':');
key.push_str(value);
key.push('|');
}
fn sort_word(sort: TrackSort) -> &'static str {
match sort {
TrackSort::Default => "default",
TrackSort::TitleAsc => "title-asc",
TrackSort::TitleDesc => "title-desc",
TrackSort::ArtistAsc => "artist-asc",
TrackSort::ArtistDesc => "artist-desc",
TrackSort::YearAsc => "year-asc",
TrackSort::YearDesc => "year-desc",
TrackSort::AddedDesc => "added-desc",
}
}
+39
View File
@@ -0,0 +1,39 @@
//! Frontend-agnostic derived catalog for Local Library.
//!
//! Source databases remain authoritative. This crate owns only a rebuildable
//! read projection and query primitives; it has no dependency on Qt, QML,
//! playback, or any source protocol/API.
mod catalog;
mod model;
mod schema;
pub use catalog::{
normalize_artist_key, normalize_sort_key, Catalog, CatalogStats, IntegrityReport, QueryMetrics,
};
pub use model::{
ArtistCredit, CreditRole, ProjectedTrack, QueryDescriptor, QuerySurface, SourceKey, SourceKind,
TrackCursor, TrackGroup, TrackPage, TrackRecord, TrackRef, TrackSort,
};
pub use schema::SCHEMA_VERSION;
#[derive(Debug, thiserror::Error)]
pub enum CatalogError {
#[error("catalog sqlite error: {0}")]
Sqlite(#[from] rusqlite::Error),
#[error("catalog schema {found} is not supported (expected {expected})")]
UnsupportedSchema { found: u32, expected: u32 },
#[error("refusing to open a non-catalog SQLite database")]
NotCatalog,
#[error("invalid catalog input: {0}")]
InvalidInput(String),
#[error("search terms shorter than three characters are deferred")]
SearchTooShort,
#[error("a cursor from a different query descriptor cannot be reused")]
CursorDescriptorMismatch,
}
pub type Result<T> = std::result::Result<T, CatalogError>;
#[cfg(test)]
mod tests;
+297
View File
@@ -0,0 +1,297 @@
use serde::{Deserialize, Serialize};
/// Authoritative origin of a projected row. Wire values are deliberately the
/// same words the existing source/cache layers use.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SourceKind {
Local,
Offline,
Plex,
Jellyfin,
Subsonic,
}
impl SourceKind {
pub const ALL: [Self; 5] = [
Self::Local,
Self::Offline,
Self::Plex,
Self::Jellyfin,
Self::Subsonic,
];
pub const fn as_str(self) -> &'static str {
match self {
Self::Local => "local",
Self::Offline => "offline",
Self::Plex => "plex",
Self::Jellyfin => "jellyfin",
Self::Subsonic => "subsonic",
}
}
pub fn from_str(value: &str) -> Option<Self> {
match value {
"local" => Some(Self::Local),
"offline" => Some(Self::Offline),
"plex" => Some(Self::Plex),
"jellyfin" => Some(Self::Jellyfin),
"subsonic" => Some(Self::Subsonic),
_ => None,
}
}
}
/// Stable product identity. The catalog's integer rowid never leaves through
/// this type; playback and user data continue to use source-native identity.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct TrackRef {
pub source: SourceKind,
pub source_instance: String,
pub native_id: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
pub struct SourceKey {
pub source: SourceKind,
pub source_instance: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CreditRole {
TrackArtist,
AlbumArtist,
Composer,
Performer,
Featured,
}
impl CreditRole {
pub const fn as_str(self) -> &'static str {
match self {
Self::TrackArtist => "track_artist",
Self::AlbumArtist => "album_artist",
Self::Composer => "composer",
Self::Performer => "performer",
Self::Featured => "featured",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArtistCredit {
pub display_name: String,
pub role: CreditRole,
pub ordinal: u32,
}
/// Source-neutral write shape. C/H adapters translate authoritative rows into
/// this type; the catalog itself never calls a source API.
#[derive(Debug, Clone)]
pub struct ProjectedTrack {
pub track_ref: TrackRef,
pub local_track_id: Option<i64>,
pub local_path: Option<String>,
pub source_copy_id: Option<i64>,
pub title: String,
pub artist: String,
pub album_artist: String,
pub album: String,
pub duration_ms: u64,
pub year: Option<u32>,
pub disc_number: Option<u32>,
pub track_number: Option<u32>,
pub format: String,
pub bit_depth: Option<u32>,
pub sample_rate_hz: Option<u32>,
pub artwork_token: Option<String>,
pub isrc: Option<String>,
pub musicbrainz_recording_id: Option<String>,
pub added_at: i64,
pub available: bool,
pub observed_generation: i64,
pub credits: Vec<ArtistCredit>,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum TrackSort {
#[default]
Default,
TitleAsc,
TitleDesc,
ArtistAsc,
ArtistDesc,
YearAsc,
YearDesc,
AddedDesc,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TrackGroup {
#[default]
Off,
Album,
Artist,
Name,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum QuerySurface {
Tracks,
Albums,
Artists,
}
/// Immutable query snapshot. Builders consume and return a new descriptor so
/// an async request cannot observe later UI mutations through shared state.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QueryDescriptor {
surface: QuerySurface,
search: String,
sources: Vec<SourceKey>,
formats: Vec<String>,
sort: TrackSort,
group: TrackGroup,
available_only: bool,
}
impl QueryDescriptor {
pub fn tracks() -> Self {
Self::for_surface(QuerySurface::Tracks)
}
pub fn albums() -> Self {
Self::for_surface(QuerySurface::Albums)
}
pub fn artists() -> Self {
Self::for_surface(QuerySurface::Artists)
}
pub fn for_surface(surface: QuerySurface) -> Self {
Self {
surface,
search: String::new(),
sources: Vec::new(),
formats: Vec::new(),
sort: TrackSort::Default,
group: TrackGroup::Off,
available_only: true,
}
}
pub fn with_search(mut self, search: impl Into<String>) -> Self {
self.search = search.into().trim().to_string();
self
}
pub fn with_sources(mut self, sources: Vec<SourceKey>) -> Self {
self.sources = sources;
self.sources.sort();
self.sources.dedup();
self
}
pub fn with_formats(mut self, formats: Vec<String>) -> Self {
self.formats = formats
.into_iter()
.map(|value| value.trim().to_ascii_lowercase())
.filter(|value| !value.is_empty())
.collect();
self.formats.sort();
self.formats.dedup();
self
}
pub fn with_sort(mut self, sort: TrackSort) -> Self {
self.sort = sort;
self
}
pub fn with_group(mut self, group: TrackGroup) -> Self {
self.group = group;
self
}
pub fn including_unavailable(mut self) -> Self {
self.available_only = false;
self
}
pub fn surface(&self) -> QuerySurface {
self.surface
}
pub fn search(&self) -> &str {
&self.search
}
pub fn sources(&self) -> &[SourceKey] {
&self.sources
}
pub fn formats(&self) -> &[String] {
&self.formats
}
pub fn sort(&self) -> TrackSort {
self.sort
}
pub fn group(&self) -> TrackGroup {
self.group
}
pub fn available_only(&self) -> bool {
self.available_only
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TrackRecord {
pub track_ref: TrackRef,
pub local_track_id: Option<i64>,
pub local_path: Option<String>,
pub title: String,
pub artist: String,
pub album_artist: String,
pub album: String,
pub duration_ms: u64,
pub year: Option<u32>,
pub disc_number: Option<u32>,
pub track_number: Option<u32>,
pub format: String,
pub bit_depth: Option<u32>,
pub sample_rate_hz: Option<u32>,
pub artwork_token: Option<String>,
pub available: bool,
}
/// Opaque keyset cursor. It carries the exact normalized ORDER BY values plus
/// the catalog rowid tie-breaker, but exposes none as product identity.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TrackCursor {
pub(crate) sort: TrackSort,
pub(crate) descriptor_key: String,
pub(crate) sort_title: String,
pub(crate) sort_artist: String,
pub(crate) sort_album: String,
pub(crate) year_missing: i64,
pub(crate) year_value: i64,
pub(crate) disc_sort: i64,
pub(crate) track_sort: i64,
pub(crate) added_at: i64,
pub(crate) row_id: i64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TrackPage {
pub rows: Vec<TrackRecord>,
pub next_cursor: Option<TrackCursor>,
pub has_more: bool,
}
+294
View File
@@ -0,0 +1,294 @@
use rusqlite::{params, Connection};
use crate::{CatalogError, Result};
pub const SCHEMA_VERSION: u32 = 1;
pub const APPLICATION_ID: i64 = 0x5142_5A43; // "QBZC"
pub(crate) fn configure(conn: &Connection) -> Result<()> {
conn.busy_timeout(std::time::Duration::from_millis(2_500))?;
conn.execute_batch(&format!(
"PRAGMA foreign_keys=ON;
PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;
PRAGMA temp_store=MEMORY;
PRAGMA cache_size=-32768;
PRAGMA mmap_size=268435456;
PRAGMA application_id={APPLICATION_ID};"
))?;
Ok(())
}
pub(crate) fn init(conn: &mut Connection, generation: u64) -> Result<()> {
let application_id: i64 = conn.query_row("PRAGMA application_id", [], |row| row.get(0))?;
if application_id != 0 && application_id != APPLICATION_ID {
return Err(CatalogError::NotCatalog);
}
if application_id == 0 {
let existing_tables: i64 = conn.query_row(
"SELECT COUNT(*) FROM sqlite_master
WHERE type = 'table' AND name NOT LIKE 'sqlite_%'",
[],
|row| row.get(0),
)?;
if existing_tables > 0 {
return Err(CatalogError::NotCatalog);
}
}
configure(conn)?;
let found: u32 = conn.query_row("PRAGMA user_version", [], |row| row.get(0))?;
if found != 0 && found != SCHEMA_VERSION {
return Err(CatalogError::UnsupportedSchema {
found,
expected: SCHEMA_VERSION,
});
}
let tx = conn.transaction()?;
tx.execute_batch(SCHEMA_SQL)?;
tx.execute(
"INSERT INTO catalog_meta(key, value) VALUES ('schema_version', ?1)
ON CONFLICT(key) DO NOTHING",
params![SCHEMA_VERSION.to_string()],
)?;
tx.execute(
"INSERT INTO catalog_meta(key, value) VALUES ('generation', ?1)
ON CONFLICT(key) DO NOTHING",
params![generation.to_string()],
)?;
tx.pragma_update(None, "user_version", SCHEMA_VERSION)?;
tx.commit()?;
let stored_generation: String = conn.query_row(
"SELECT value FROM catalog_meta WHERE key = 'generation'",
[],
|row| row.get(0),
)?;
if stored_generation != generation.to_string() {
return Err(CatalogError::InvalidInput(format!(
"requested generation {generation}, database contains generation {stored_generation}"
)));
}
Ok(())
}
const SCHEMA_SQL: &str = r#"
CREATE TABLE IF NOT EXISTS catalog_meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
) STRICT;
CREATE TABLE IF NOT EXISTS source_state (
source_kind TEXT NOT NULL CHECK (
source_kind IN ('local','offline','plex','jellyfin','subsonic')
),
source_instance TEXT NOT NULL,
available INTEGER NOT NULL DEFAULT 1 CHECK (available IN (0,1)),
last_observed_at INTEGER NOT NULL DEFAULT 0,
watermark TEXT NOT NULL DEFAULT '',
complete_generation INTEGER NOT NULL DEFAULT 0,
checkpoint_cursor TEXT NOT NULL DEFAULT '',
checkpoint_rows INTEGER NOT NULL DEFAULT 0,
checkpoint_version TEXT NOT NULL DEFAULT '',
PRIMARY KEY (source_kind, source_instance)
) STRICT;
CREATE TABLE IF NOT EXISTS logical_albums (
logical_album_id INTEGER PRIMARY KEY,
stable_key TEXT NOT NULL UNIQUE,
display_title TEXT NOT NULL,
sort_title TEXT NOT NULL,
display_artist TEXT NOT NULL,
sort_artist TEXT NOT NULL,
association_strength TEXT NOT NULL DEFAULT 'source_native'
CHECK (association_strength IN ('source_native','isrc','musicbrainz','manual')),
association_evidence TEXT NOT NULL DEFAULT ''
) STRICT;
CREATE TABLE IF NOT EXISTS editions (
edition_id INTEGER PRIMARY KEY,
logical_album_id INTEGER NOT NULL REFERENCES logical_albums(logical_album_id),
edition_key TEXT NOT NULL UNIQUE,
display_title TEXT NOT NULL,
display_artist TEXT NOT NULL,
release_year INTEGER,
musicbrainz_release_id TEXT,
provider_release_id TEXT,
evidence_kind TEXT NOT NULL DEFAULT 'source_native',
evidence_value TEXT NOT NULL DEFAULT ''
) STRICT;
CREATE INDEX IF NOT EXISTS idx_editions_logical
ON editions(logical_album_id, release_year, edition_id);
CREATE TABLE IF NOT EXISTS source_copies (
source_copy_id INTEGER PRIMARY KEY,
edition_id INTEGER NOT NULL REFERENCES editions(edition_id),
source_kind TEXT NOT NULL CHECK (
source_kind IN ('local','offline','plex','jellyfin','subsonic')
),
source_instance TEXT NOT NULL,
native_album_id TEXT NOT NULL,
local_directory TEXT,
available INTEGER NOT NULL DEFAULT 1 CHECK (available IN (0,1)),
last_observed_generation INTEGER NOT NULL DEFAULT 0,
UNIQUE (source_kind, source_instance, native_album_id)
) STRICT;
CREATE INDEX IF NOT EXISTS idx_source_copies_edition
ON source_copies(edition_id, source_kind, source_instance);
CREATE TABLE IF NOT EXISTS tracks (
catalog_id INTEGER PRIMARY KEY,
source_kind TEXT NOT NULL CHECK (
source_kind IN ('local','offline','plex','jellyfin','subsonic')
),
source_instance TEXT NOT NULL,
native_track_id TEXT NOT NULL,
local_track_id INTEGER,
local_path TEXT,
source_copy_id INTEGER REFERENCES source_copies(source_copy_id),
title TEXT NOT NULL,
sort_title TEXT NOT NULL,
artist TEXT NOT NULL,
sort_artist TEXT NOT NULL,
album_artist TEXT NOT NULL,
album TEXT NOT NULL,
sort_album TEXT NOT NULL,
credits TEXT NOT NULL DEFAULT '',
duration_ms INTEGER NOT NULL DEFAULT 0,
year INTEGER,
year_missing INTEGER NOT NULL CHECK (year_missing IN (0,1)),
year_value INTEGER NOT NULL,
disc_number INTEGER,
disc_sort INTEGER NOT NULL,
track_number INTEGER,
track_sort INTEGER NOT NULL,
format TEXT NOT NULL,
bit_depth INTEGER,
sample_rate_hz INTEGER,
artwork_token TEXT,
isrc TEXT,
musicbrainz_recording_id TEXT,
added_at INTEGER NOT NULL DEFAULT 0,
available INTEGER NOT NULL DEFAULT 1 CHECK (available IN (0,1)),
last_observed_generation INTEGER NOT NULL DEFAULT 0,
UNIQUE (source_kind, source_instance, native_track_id)
) STRICT;
CREATE INDEX IF NOT EXISTS idx_tracks_source_copy
ON tracks(source_copy_id, disc_sort, track_sort, catalog_id);
CREATE INDEX IF NOT EXISTS idx_tracks_local_id
ON tracks(local_track_id) WHERE local_track_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_tracks_isrc
ON tracks(isrc) WHERE isrc IS NOT NULL AND isrc != '';
CREATE INDEX IF NOT EXISTS idx_tracks_mbid
ON tracks(musicbrainz_recording_id)
WHERE musicbrainz_recording_id IS NOT NULL AND musicbrainz_recording_id != '';
-- Every index begins with the default availability predicate and ends in the
-- same catalog_id tie-breaker used by keyset cursors. Query SQL must not drift
-- from these expressions/collations.
CREATE INDEX IF NOT EXISTS idx_tracks_default
ON tracks(available, sort_album, sort_artist, disc_sort, track_sort, sort_title, catalog_id);
CREATE INDEX IF NOT EXISTS idx_tracks_title_asc
ON tracks(available, sort_title, sort_artist, catalog_id);
CREATE INDEX IF NOT EXISTS idx_tracks_title_desc
ON tracks(available, sort_title DESC, sort_artist, catalog_id);
CREATE INDEX IF NOT EXISTS idx_tracks_artist_asc
ON tracks(available, sort_artist, sort_album, disc_sort, track_sort, catalog_id);
CREATE INDEX IF NOT EXISTS idx_tracks_artist_desc
ON tracks(available, sort_artist DESC, sort_album, disc_sort, track_sort, catalog_id);
CREATE INDEX IF NOT EXISTS idx_tracks_year_asc
ON tracks(available, year_missing, year_value, sort_album, disc_sort, track_sort, catalog_id);
CREATE INDEX IF NOT EXISTS idx_tracks_year_desc
ON tracks(available, year_missing, year_value DESC, sort_album, disc_sort, track_sort, catalog_id);
CREATE INDEX IF NOT EXISTS idx_tracks_added_desc
ON tracks(available, added_at DESC, sort_album, disc_sort, track_sort, catalog_id);
CREATE TABLE IF NOT EXISTS artist_credits (
catalog_id INTEGER NOT NULL REFERENCES tracks(catalog_id) ON DELETE CASCADE,
artist_key TEXT NOT NULL,
display_name TEXT NOT NULL,
role TEXT NOT NULL CHECK (
role IN ('track_artist','album_artist','composer','performer','featured')
),
ordinal INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (catalog_id, role, ordinal, artist_key)
) STRICT;
CREATE INDEX IF NOT EXISTS idx_artist_credits_artist
ON artist_credits(artist_key, role, catalog_id);
CREATE TABLE IF NOT EXISTS albums_materialized (
edition_id INTEGER PRIMARY KEY REFERENCES editions(edition_id) ON DELETE CASCADE,
logical_album_id INTEGER NOT NULL REFERENCES logical_albums(logical_album_id),
title TEXT NOT NULL,
sort_title TEXT NOT NULL,
artist TEXT NOT NULL,
sort_artist TEXT NOT NULL,
year INTEGER,
track_count INTEGER NOT NULL DEFAULT 0,
total_duration_ms INTEGER NOT NULL DEFAULT 0,
source_count INTEGER NOT NULL DEFAULT 0,
available INTEGER NOT NULL DEFAULT 1 CHECK (available IN (0,1)),
artwork_source TEXT NOT NULL DEFAULT '',
artwork_token TEXT NOT NULL DEFAULT ''
) STRICT;
CREATE INDEX IF NOT EXISTS idx_albums_materialized_artist
ON albums_materialized(available, sort_artist, year, sort_title, edition_id);
CREATE INDEX IF NOT EXISTS idx_albums_materialized_title
ON albums_materialized(available, sort_title, sort_artist, edition_id);
CREATE TABLE IF NOT EXISTS artists_materialized (
artist_key TEXT PRIMARY KEY,
display_name TEXT NOT NULL,
sort_name TEXT NOT NULL,
album_count INTEGER NOT NULL DEFAULT 0,
track_count INTEGER NOT NULL DEFAULT 0,
available INTEGER NOT NULL DEFAULT 1 CHECK (available IN (0,1)),
artwork_source TEXT NOT NULL DEFAULT '',
artwork_token TEXT NOT NULL DEFAULT ''
) STRICT;
CREATE INDEX IF NOT EXISTS idx_artists_materialized_name
ON artists_materialized(available, sort_name, artist_key);
CREATE TABLE IF NOT EXISTS edition_artists (
edition_id INTEGER NOT NULL REFERENCES editions(edition_id) ON DELETE CASCADE,
artist_key TEXT NOT NULL,
role TEXT NOT NULL,
PRIMARY KEY (edition_id, artist_key, role)
) STRICT;
CREATE INDEX IF NOT EXISTS idx_edition_artists_artist
ON edition_artists(artist_key, role, edition_id);
CREATE VIRTUAL TABLE IF NOT EXISTS tracks_fts USING fts5(
title,
album,
artist,
credits,
content='tracks',
content_rowid='catalog_id',
tokenize='trigram case_sensitive 0'
);
CREATE TRIGGER IF NOT EXISTS tracks_fts_insert AFTER INSERT ON tracks BEGIN
INSERT INTO tracks_fts(rowid, title, album, artist, credits)
VALUES (new.catalog_id, new.title, new.album, new.artist, new.credits);
END;
CREATE TRIGGER IF NOT EXISTS tracks_fts_delete AFTER DELETE ON tracks BEGIN
INSERT INTO tracks_fts(tracks_fts, rowid, title, album, artist, credits)
VALUES ('delete', old.catalog_id, old.title, old.album, old.artist, old.credits);
END;
CREATE TRIGGER IF NOT EXISTS tracks_fts_update AFTER UPDATE ON tracks BEGIN
INSERT INTO tracks_fts(tracks_fts, rowid, title, album, artist, credits)
VALUES ('delete', old.catalog_id, old.title, old.album, old.artist, old.credits);
INSERT INTO tracks_fts(rowid, title, album, artist, credits)
VALUES (new.catalog_id, new.title, new.album, new.artist, new.credits);
END;
"#;
+539
View File
@@ -0,0 +1,539 @@
use std::collections::HashSet;
use std::time::{Duration, Instant};
use crate::*;
fn source_for(index: usize) -> SourceKind {
match index % 100 {
0..=59 => SourceKind::Local,
60..=79 => SourceKind::Plex,
80..=89 => SourceKind::Jellyfin,
90..=97 => SourceKind::Subsonic,
_ => SourceKind::Offline,
}
}
fn source_instance(source: SourceKind) -> &'static str {
match source {
SourceKind::Local => "local-main",
SourceKind::Offline => "qobuz-user-7",
SourceKind::Plex => "plex-server-a",
SourceKind::Jellyfin => "jellyfin-server-b",
SourceKind::Subsonic => "subsonic-server-c",
}
}
fn projected(index: usize, source: SourceKind) -> ProjectedTrack {
let artist_number = index % 50_000;
let album_number = index % 80_000;
let format = match index % 10 {
0..=5 => "flac",
6..=7 => "mp3",
8 => "alac",
_ => "dsf",
};
let artist = format!("Artist {artist_number:05}");
ProjectedTrack {
track_ref: TrackRef {
source,
source_instance: source_instance(source).to_string(),
native_id: index.to_string(),
},
local_track_id: (source == SourceKind::Local).then_some(index as i64 + 1),
local_path: (source == SourceKind::Local)
.then(|| format!("/fixture/artist-{artist_number:05}/track-{index:07}.{format}")),
source_copy_id: None,
title: format!("Track {index:07} Signal {:03}", index % 997),
artist: artist.clone(),
album_artist: artist.clone(),
album: format!("Album {album_number:05}"),
duration_ms: 120_000 + (index % 480_000) as u64,
year: (index % 23 != 0).then_some(1950 + (index % 77) as u32),
disc_number: Some((index % 3 + 1) as u32),
track_number: Some((index % 24 + 1) as u32),
format: format.to_string(),
bit_depth: (format != "mp3").then_some(if index % 3 == 0 { 24 } else { 16 }),
sample_rate_hz: Some(if index % 5 == 0 { 96_000 } else { 44_100 }),
artwork_token: Some(format!("art-{:05}", album_number)),
isrc: (index % 11 == 0).then(|| format!("ISRC{index:08}")),
musicbrainz_recording_id: (index % 101 == 0).then(|| format!("mbid-{index:08}")),
added_at: 1_700_000_000 + index as i64,
available: index % 97 != 0,
observed_generation: 1,
credits: vec![ArtistCredit {
display_name: artist,
role: CreditRole::TrackArtist,
ordinal: 0,
}],
}
}
fn insert_fixture(catalog: &mut Catalog, count: usize) {
for start in (0..count).step_by(2_000) {
let end = (start + 2_000).min(count);
let batch = (start..end)
.map(|index| projected(index, source_for(index)))
.collect::<Vec<_>>();
catalog.upsert_tracks(&batch).unwrap();
}
}
fn collect_all(catalog: &Catalog, descriptor: &QueryDescriptor) -> Vec<TrackRecord> {
let mut cursor = None;
let mut rows = Vec::new();
loop {
let page = catalog
.query_tracks(descriptor, cursor.as_ref(), 137)
.unwrap();
let done = !page.has_more;
cursor = page.next_cursor;
rows.extend(page.rows);
if done {
break;
}
}
rows
}
#[test]
fn schema_is_versioned_frontend_agnostic_and_fts5_enabled() {
let catalog = Catalog::open_in_memory(7).unwrap();
let stats = catalog.stats().unwrap();
assert_eq!(stats.schema_version, 1);
assert_eq!(stats.generation, 7);
assert_eq!(stats.track_count, 0);
let fts5: i64 = catalog
.connection()
.query_row(
"SELECT sqlite_compileoption_used('ENABLE_FTS5')",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(fts5, 1);
let tables: HashSet<String> = catalog
.connection()
.prepare("SELECT name FROM sqlite_master WHERE type IN ('table','view')")
.unwrap()
.query_map([], |row| row.get(0))
.unwrap()
.collect::<rusqlite::Result<_>>()
.unwrap();
for required in [
"source_state",
"logical_albums",
"editions",
"source_copies",
"tracks",
"artist_credits",
"albums_materialized",
"artists_materialized",
"edition_artists",
"tracks_fts",
] {
assert!(tables.contains(required), "missing {required}");
}
}
#[test]
fn opening_an_unrelated_sqlite_file_is_refused_without_mutating_it() {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("authoritative.db");
let connection = rusqlite::Connection::open(&path).unwrap();
connection
.execute("CREATE TABLE authoritative_secret(value TEXT)", [])
.unwrap();
drop(connection);
assert!(matches!(
Catalog::open(&path, 1),
Err(CatalogError::NotCatalog)
));
let connection = rusqlite::Connection::open(&path).unwrap();
let catalog_tables: i64 = connection
.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE name = 'catalog_meta'",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(catalog_tables, 0);
}
#[test]
fn identical_native_ids_from_distinct_sources_never_collide() {
let mut catalog = Catalog::open_in_memory(1).unwrap();
let mut local = projected(42, SourceKind::Local);
let mut plex = projected(43, SourceKind::Plex);
local.track_ref.native_id = "same".to_string();
plex.track_ref.native_id = "same".to_string();
catalog
.upsert_tracks(&[local.clone(), plex.clone()])
.unwrap();
assert_eq!(catalog.stats().unwrap().track_count, 2);
assert_eq!(
catalog
.resolve(&local.track_ref)
.unwrap()
.unwrap()
.track_ref,
local.track_ref
);
assert_eq!(
catalog.resolve(&plex.track_ref).unwrap().unwrap().track_ref,
plex.track_ref
);
}
#[test]
fn upsert_retains_identity_and_fts_tracks_updates_and_deletes() {
let mut catalog = Catalog::open_in_memory(1).unwrap();
let mut track = projected(1, SourceKind::Jellyfin);
track.title = "Old Needle".to_string();
catalog.upsert_tracks(&[track.clone()]).unwrap();
assert_eq!(
catalog
.count_tracks(&QueryDescriptor::tracks().with_search("Old Needle"))
.unwrap(),
1
);
track.title = "New Signal".to_string();
catalog.upsert_tracks(&[track.clone()]).unwrap();
assert_eq!(catalog.stats().unwrap().track_count, 1);
assert_eq!(
catalog
.count_tracks(&QueryDescriptor::tracks().with_search("Old Needle"))
.unwrap(),
0
);
assert_eq!(
catalog
.count_tracks(&QueryDescriptor::tracks().with_search("New Signal"))
.unwrap(),
1
);
assert!(catalog.remove_track(&track.track_ref).unwrap());
assert_eq!(catalog.stats().unwrap().track_count, 0);
let integrity = catalog.integrity_check().unwrap();
assert!(integrity.sqlite_ok && integrity.fts_ok);
assert_eq!(integrity.foreign_key_violations, 0);
}
#[test]
fn artist_keys_preserve_roles_fold_diacritics_and_keep_non_latin_names() {
assert_eq!(normalize_artist_key("Beyoncé & Co."), "beyonce co");
assert_eq!(normalize_artist_key("宇多田 ヒカル"), "宇多田 ヒカル");
let mut catalog = Catalog::open_in_memory(1).unwrap();
let mut track = projected(3, SourceKind::Subsonic);
track.credits = vec![
ArtistCredit {
display_name: "Beyoncé".to_string(),
role: CreditRole::TrackArtist,
ordinal: 0,
},
ArtistCredit {
display_name: "Guest".to_string(),
role: CreditRole::Featured,
ordinal: 1,
},
];
catalog.upsert_tracks(&[track]).unwrap();
let credits: Vec<(String, String)> = catalog
.connection()
.prepare("SELECT artist_key, role FROM artist_credits ORDER BY ordinal")
.unwrap()
.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
.unwrap()
.collect::<rusqlite::Result<_>>()
.unwrap();
assert_eq!(
credits,
vec![
("beyonce".to_string(), "track_artist".to_string()),
("guest".to_string(), "featured".to_string())
]
);
}
#[test]
fn every_sort_uses_keysets_without_omissions_or_duplicates() {
let mut catalog = Catalog::open_in_memory(1).unwrap();
insert_fixture(&mut catalog, 1_701);
let expected = catalog.count_tracks(&QueryDescriptor::tracks()).unwrap() as usize;
for sort in [
TrackSort::Default,
TrackSort::TitleAsc,
TrackSort::TitleDesc,
TrackSort::ArtistAsc,
TrackSort::ArtistDesc,
TrackSort::YearAsc,
TrackSort::YearDesc,
TrackSort::AddedDesc,
] {
let rows = collect_all(&catalog, &QueryDescriptor::tracks().with_sort(sort));
let ids: HashSet<TrackRef> = rows.iter().map(|row| row.track_ref.clone()).collect();
assert_eq!(rows.len(), expected, "sort {sort:?}");
assert_eq!(ids.len(), rows.len(), "sort {sort:?}");
}
}
#[test]
fn search_source_format_availability_and_group_are_descriptor_scoped() {
let mut catalog = Catalog::open_in_memory(1).unwrap();
insert_fixture(&mut catalog, 2_500);
let search = QueryDescriptor::tracks().with_search("Signal 042");
assert!(catalog.count_tracks(&search).unwrap() > 0);
assert!(matches!(
catalog.count_tracks(&QueryDescriptor::tracks().with_search("Si")),
Err(CatalogError::SearchTooShort)
));
let plex_flac = QueryDescriptor::tracks()
.with_sources(vec![SourceKey {
source: SourceKind::Plex,
source_instance: source_instance(SourceKind::Plex).to_string(),
}])
.with_formats(vec!["FLAC".to_string()]);
let rows = collect_all(&catalog, &plex_flac);
assert!(!rows.is_empty());
assert!(rows.iter().all(|row| {
row.track_ref.source == SourceKind::Plex && row.format == "flac" && row.available
}));
let available = catalog.count_tracks(&QueryDescriptor::tracks()).unwrap();
let all = catalog
.count_tracks(&QueryDescriptor::tracks().including_unavailable())
.unwrap();
assert!(all > available);
let grouped = collect_all(
&catalog,
&QueryDescriptor::tracks()
.with_sort(TrackSort::AddedDesc)
.with_group(TrackGroup::Artist),
);
assert_eq!(grouped.len(), available as usize);
let first = catalog
.query_tracks(&QueryDescriptor::tracks(), None, 50)
.unwrap();
let cursor = first.next_cursor.expect("fixture has another page");
assert!(matches!(
catalog.query_tracks(
&QueryDescriptor::tracks().with_formats(vec!["flac".to_string()]),
Some(&cursor),
50,
),
Err(CatalogError::CursorDescriptorMismatch)
));
assert!(matches!(
catalog.count_tracks(&QueryDescriptor::albums()),
Err(CatalogError::InvalidInput(_))
));
}
#[test]
fn album_hierarchy_keeps_weak_same_name_matches_reversible() {
let catalog = Catalog::open_in_memory(1).unwrap();
catalog
.connection()
.execute_batch(
"INSERT INTO logical_albums
(stable_key,display_title,sort_title,display_artist,sort_artist,
association_strength,association_evidence)
VALUES
('plex:a','Album','album','Artist','artist','source_native','plex:a'),
('jellyfin:b','Album','album','Artist','artist','source_native','jellyfin:b');
INSERT INTO editions
(logical_album_id,edition_key,display_title,display_artist,evidence_kind,evidence_value)
SELECT logical_album_id, stable_key || ':edition', display_title, display_artist,
'source_native', stable_key
FROM logical_albums;
INSERT INTO source_copies
(edition_id,source_kind,source_instance,native_album_id)
SELECT edition_id,
CASE WHEN edition_key LIKE 'plex:%' THEN 'plex' ELSE 'jellyfin' END,
CASE WHEN edition_key LIKE 'plex:%' THEN 'plex-a' ELSE 'jellyfin-b' END,
edition_key
FROM editions;",
)
.unwrap();
let counts: (i64, i64, i64) = catalog
.connection()
.query_row(
"SELECT
(SELECT COUNT(*) FROM logical_albums),
(SELECT COUNT(*) FROM editions),
(SELECT COUNT(*) FROM source_copies)",
[],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.unwrap();
assert_eq!(counts, (2, 2, 2));
}
#[test]
fn sort_indices_match_the_actual_order_by_without_temp_sort() {
let mut catalog = Catalog::open_in_memory(1).unwrap();
insert_fixture(&mut catalog, 100);
for (index, order) in [
(
"idx_tracks_default",
"sort_album,sort_artist,disc_sort,track_sort,sort_title,catalog_id",
),
(
"idx_tracks_title_desc",
"sort_title DESC,sort_artist,catalog_id",
),
(
"idx_tracks_year_desc",
"year_missing,year_value DESC,sort_album,disc_sort,track_sort,catalog_id",
),
] {
let sql = format!(
"EXPLAIN QUERY PLAN SELECT catalog_id FROM tracks
WHERE available=1 ORDER BY {order} LIMIT 250"
);
let details = catalog
.connection()
.prepare(&sql)
.unwrap()
.query_map([], |row| row.get::<_, String>(3))
.unwrap()
.collect::<rusqlite::Result<Vec<_>>>()
.unwrap()
.join(" | ");
assert!(details.contains(index), "{details}");
assert!(!details.contains("TEMP B-TREE"), "{details}");
}
let fts_plan = catalog
.connection()
.prepare(
"EXPLAIN QUERY PLAN
SELECT t.catalog_id
FROM tracks_fts
CROSS JOIN tracks t NOT INDEXED
WHERE tracks_fts MATCH '\"Signal 042\"'
AND t.catalog_id=tracks_fts.rowid
AND t.available=1
ORDER BY t.sort_album,t.sort_artist,t.disc_sort,t.track_sort,
t.sort_title,t.catalog_id
LIMIT 250",
)
.unwrap()
.query_map([], |row| row.get::<_, String>(3))
.unwrap()
.collect::<rusqlite::Result<Vec<_>>>()
.unwrap()
.join(" | ");
assert!(fts_plan.contains("VIRTUAL TABLE INDEX"), "{fts_plan}");
assert!(fts_plan.contains("INTEGER PRIMARY KEY"), "{fts_plan}");
}
/// The contract's reproducible scale gate. It is ignored in ordinary unit
/// runs because it intentionally creates and removes a several-hundred-MiB
/// database; the implementation handoff records an explicit execution.
#[test]
#[ignore = "explicit 1M catalog fixture and query benchmark"]
fn fixture_one_million_tracks_meets_the_query_shape_gate() {
const TRACKS: usize = 1_000_000;
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("local_catalog-v1-g1.db.building");
let mut catalog = Catalog::open(&path, 1).unwrap();
let insert_started = Instant::now();
for start in (0..TRACKS).step_by(5_000) {
let end = (start + 5_000).min(TRACKS);
let batch = (start..end)
.map(|index| projected(index, source_for(index)))
.collect::<Vec<_>>();
catalog.upsert_tracks(&batch).unwrap();
if end % 100_000 == 0 {
println!("[catalog-1m] projected={end}");
}
}
let insert_time = insert_started.elapsed();
catalog
.connection()
.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")
.unwrap();
let stats = catalog.stats().unwrap();
assert_eq!(stats.track_count, TRACKS as u64);
let integrity = catalog.integrity_check().unwrap();
assert!(integrity.sqlite_ok && integrity.fts_ok);
assert_eq!(integrity.foreign_key_violations, 0);
let default = QueryDescriptor::tracks();
let (_, first_warmup) = catalog.query_tracks_timed(&default, None, 250).unwrap();
let (_, first_warm) = catalog.query_tracks_timed(&default, None, 250).unwrap();
let broad_fts = QueryDescriptor::tracks().with_search("Signal 731");
let (broad_page, broad_fts_cold) = catalog.query_tracks_timed(&broad_fts, None, 250).unwrap();
assert_eq!(broad_page.rows.len(), 250);
let (_, broad_fts_warm) = catalog.query_tracks_timed(&broad_fts, None, 250).unwrap();
let selective_fts = QueryDescriptor::tracks().with_search("0731042");
let (selective_page, selective_fts_cold) = catalog
.query_tracks_timed(&selective_fts, None, 250)
.unwrap();
assert_eq!(selective_page.rows.len(), 1);
let (_, selective_fts_warm) = catalog
.query_tracks_timed(&selective_fts, None, 250)
.unwrap();
let deep_cursor = catalog
.connection()
.query_row(
"SELECT sort_title,sort_artist,sort_album,year_missing,year_value,
disc_sort,track_sort,added_at,catalog_id
FROM tracks WHERE available=1
ORDER BY sort_title,sort_artist,catalog_id
LIMIT 1 OFFSET 800000",
[],
|row| {
Ok(TrackCursor {
sort: TrackSort::TitleAsc,
descriptor_key: crate::catalog::descriptor_key(
&QueryDescriptor::tracks().with_sort(TrackSort::TitleAsc),
),
sort_title: row.get(0)?,
sort_artist: row.get(1)?,
sort_album: row.get(2)?,
year_missing: row.get(3)?,
year_value: row.get(4)?,
disc_sort: row.get(5)?,
track_sort: row.get(6)?,
added_at: row.get(7)?,
row_id: row.get(8)?,
})
},
)
.unwrap();
let title = QueryDescriptor::tracks().with_sort(TrackSort::TitleAsc);
let (_, deep_metrics) = catalog
.query_tracks_timed(&title, Some(&deep_cursor), 250)
.unwrap();
drop(catalog);
let reopened = Catalog::open(&path, 1).unwrap();
let (_, connection_cold) = reopened.query_tracks_timed(&default, None, 250).unwrap();
let file_bytes = std::fs::metadata(&path).unwrap().len();
println!(
"[catalog-1m] rows={TRACKS} db_bytes={file_bytes} insert_s={:.3} first_warmup_ms={:.3} first_warm_ms={:.3} broad_fts_cold_ms={:.3} broad_fts_warm_ms={:.3} selective_fts_cold_ms={:.3} selective_fts_warm_ms={:.3} deep_keyset_ms={:.3} connection_cold_ms={:.3}",
insert_time.as_secs_f64(),
first_warmup.sql_time.as_secs_f64() * 1000.0,
first_warm.sql_time.as_secs_f64() * 1000.0,
broad_fts_cold.sql_time.as_secs_f64() * 1000.0,
broad_fts_warm.sql_time.as_secs_f64() * 1000.0,
selective_fts_cold.sql_time.as_secs_f64() * 1000.0,
selective_fts_warm.sql_time.as_secs_f64() * 1000.0,
deep_metrics.sql_time.as_secs_f64() * 1000.0,
connection_cold.sql_time.as_secs_f64() * 1000.0,
);
assert!(first_warm.sql_time <= Duration::from_millis(50));
assert!(broad_fts_warm.sql_time <= Duration::from_millis(250));
assert!(selective_fts_warm.sql_time <= Duration::from_millis(50));
assert!(deep_metrics.sql_time <= Duration::from_millis(100));
assert!(connection_cold.sql_time <= Duration::from_millis(250));
}