feat(library): persist distinct purchase copies
cx
This commit is contained in:
4 files changed
+826
-2
No files matched your search
@@ -379,6 +379,8 @@ impl LibraryDatabase {
|
||||
)
|
||||
.map_err(|e| LibraryError::Database(format!("Failed to create schema: {}", e)))?;
|
||||
|
||||
crate::purchase_copies::init_schema(&self.conn)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -974,6 +976,8 @@ impl LibraryDatabase {
|
||||
})?;
|
||||
}
|
||||
|
||||
crate::purchase_copies::migrate_legacy_registry(&self.conn)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -37,6 +37,7 @@ pub mod local_playlists;
|
||||
mod metadata;
|
||||
mod models;
|
||||
mod mount_info;
|
||||
mod purchase_copies;
|
||||
pub mod playlist_membership;
|
||||
pub mod qobuz_playlist_snapshot;
|
||||
pub mod reachability;
|
||||
@@ -60,7 +61,13 @@ pub use errors::LibraryError;
|
||||
pub use metadata::MetadataExtractor;
|
||||
pub use models::*;
|
||||
pub use mount_info::{is_network_path, network_fs_label};
|
||||
pub use reachability::{probe, probe_default, Reach, DEFAULT_PROBE};
|
||||
pub use purchase_copies::{
|
||||
PurchaseAlbumPlaybackPreference, PurchaseCopyHealth, PurchaseCopySnapshot,
|
||||
PurchaseDownloadCopy, PurchaseDownloadCopyTrack, PurchasePlaybackMode, PurchaseTrackHealth,
|
||||
};
|
||||
pub use reachability::{
|
||||
probe, probe_default, probe_file, probe_file_default, FileProbe, Reach, DEFAULT_PROBE,
|
||||
};
|
||||
pub use sacd::{SacdImageImport, SacdImportResult};
|
||||
pub use sacd_scan::{build_image_rows, SacdImageRows, SacdLabels, SacdScanSummary};
|
||||
pub use scan::{scan_with_progress, ScanEvent};
|
||||
|
||||
@@ -0,0 +1,737 @@
|
||||
//! Persistent, additive inventory for DRM-free Qobuz purchase copies.
|
||||
//!
|
||||
//! `downloaded_purchases` remains a compatibility pointer for old UI and the
|
||||
//! scanner. This module is the authority for N physical copies: coverage is
|
||||
//! always evaluated inside one `copy_id`, never by combining folders.
|
||||
|
||||
use crate::{FileProbe, LibraryDatabase, LibraryError};
|
||||
use rusqlite::{params, Connection, OptionalExtension};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::{BTreeMap, HashSet};
|
||||
use std::path::Path;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use uuid::Uuid;
|
||||
|
||||
const LEGACY_MIGRATION_KEY: &str = "purchase_copies_legacy_migration_v1";
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum PurchasePlaybackMode {
|
||||
Qobuz,
|
||||
Purchase,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct PurchaseAlbumPlaybackPreference {
|
||||
pub album_id: String,
|
||||
pub mode: PurchasePlaybackMode,
|
||||
pub format_id: Option<i64>,
|
||||
pub updated_at: i64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct PurchaseDownloadCopy {
|
||||
pub copy_id: String,
|
||||
pub album_id: String,
|
||||
pub format_id: i64,
|
||||
pub resolved_album_folder: String,
|
||||
pub created_at: i64,
|
||||
pub updated_at: i64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct PurchaseDownloadCopyTrack {
|
||||
pub copy_id: String,
|
||||
pub track_id: i64,
|
||||
pub file_path: String,
|
||||
pub mime_type: Option<String>,
|
||||
pub container: Option<String>,
|
||||
pub sample_rate_hz: Option<u32>,
|
||||
pub bit_depth: Option<u32>,
|
||||
pub channels: Option<u16>,
|
||||
pub file_size: u64,
|
||||
pub modified_ns: Option<u128>,
|
||||
pub last_verified_at: Option<i64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "kebab-case")]
|
||||
pub enum PurchaseTrackHealth {
|
||||
Healthy,
|
||||
Missing,
|
||||
Changed,
|
||||
Unreachable,
|
||||
Unreadable,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "kebab-case")]
|
||||
pub enum PurchaseCopyHealth {
|
||||
CompleteHealthy,
|
||||
Partial,
|
||||
Missing,
|
||||
Changed,
|
||||
Unreachable,
|
||||
Unreadable,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct PurchaseCopySnapshot {
|
||||
pub copy: PurchaseDownloadCopy,
|
||||
pub health: PurchaseCopyHealth,
|
||||
pub downloaded_tracks: u32,
|
||||
pub healthy_tracks: u32,
|
||||
pub total_tracks: u32,
|
||||
pub tracks: Vec<(PurchaseDownloadCopyTrack, PurchaseTrackHealth)>,
|
||||
}
|
||||
|
||||
pub(crate) fn init_schema(conn: &Connection) -> Result<(), LibraryError> {
|
||||
conn.execute_batch(
|
||||
r#"
|
||||
CREATE TABLE IF NOT EXISTS purchase_download_copies (
|
||||
copy_id TEXT PRIMARY KEY,
|
||||
album_id TEXT NOT NULL,
|
||||
format_id INTEGER NOT NULL,
|
||||
resolved_album_folder TEXT NOT NULL,
|
||||
created_at INTEGER NOT NULL,
|
||||
updated_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_purchase_copies_album
|
||||
ON purchase_download_copies(album_id, format_id, updated_at DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS purchase_download_copy_tracks (
|
||||
copy_id TEXT NOT NULL,
|
||||
track_id INTEGER NOT NULL,
|
||||
file_path TEXT NOT NULL,
|
||||
mime_type TEXT,
|
||||
container TEXT,
|
||||
sample_rate_hz INTEGER,
|
||||
bit_depth INTEGER,
|
||||
channels INTEGER,
|
||||
file_size INTEGER NOT NULL DEFAULT 0,
|
||||
modified_ns TEXT,
|
||||
last_verified_at INTEGER,
|
||||
PRIMARY KEY(copy_id, track_id),
|
||||
FOREIGN KEY(copy_id) REFERENCES purchase_download_copies(copy_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_purchase_copy_tracks_track
|
||||
ON purchase_download_copy_tracks(track_id, copy_id);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS purchase_album_playback_preferences (
|
||||
album_id TEXT PRIMARY KEY,
|
||||
mode TEXT NOT NULL CHECK(mode IN ('qobuz','purchase')),
|
||||
format_id INTEGER NULL,
|
||||
updated_at INTEGER NOT NULL,
|
||||
CHECK((mode = 'qobuz' AND format_id IS NULL)
|
||||
OR (mode = 'purchase' AND format_id IS NOT NULL))
|
||||
);
|
||||
"#,
|
||||
)
|
||||
.map_err(|e| LibraryError::Database(format!("purchase copies schema: {e}")))
|
||||
}
|
||||
|
||||
/// One-time, filesystem-free import of the compatibility registry. Grouping by
|
||||
/// album + format + parent folder is what prevents an old split registry from
|
||||
/// becoming one fabricated complete copy.
|
||||
pub(crate) fn migrate_legacy_registry(conn: &Connection) -> Result<(), LibraryError> {
|
||||
let done = conn
|
||||
.query_row(
|
||||
"SELECT value FROM library_kv WHERE key = ?1",
|
||||
[LEGACY_MIGRATION_KEY],
|
||||
|row| row.get::<_, String>(0),
|
||||
)
|
||||
.optional()
|
||||
.map_err(|e| LibraryError::Database(format!("read purchase migration marker: {e}")))?
|
||||
.is_some();
|
||||
if done {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let mut stmt = conn
|
||||
.prepare(
|
||||
"SELECT track_id, format_id, album_id, file_path
|
||||
FROM downloaded_purchases WHERE album_id IS NOT NULL
|
||||
ORDER BY album_id, format_id, file_path",
|
||||
)
|
||||
.map_err(|e| LibraryError::Database(format!("read legacy purchase registry: {e}")))?;
|
||||
let rows = stmt
|
||||
.query_map([], |row| {
|
||||
Ok((
|
||||
row.get::<_, i64>(0)?,
|
||||
row.get::<_, i64>(1)?,
|
||||
row.get::<_, String>(2)?,
|
||||
row.get::<_, String>(3)?,
|
||||
))
|
||||
})
|
||||
.map_err(|e| LibraryError::Database(format!("query legacy purchase registry: {e}")))?
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| LibraryError::Database(format!("collect legacy purchase registry: {e}")))?;
|
||||
drop(stmt);
|
||||
|
||||
let mut groups: BTreeMap<(String, i64, String), Vec<(i64, String)>> = BTreeMap::new();
|
||||
for (track_id, format_id, album_id, file_path) in rows {
|
||||
let Some(folder) = Path::new(&file_path).parent() else {
|
||||
continue;
|
||||
};
|
||||
groups
|
||||
.entry((album_id, format_id, folder.to_string_lossy().into_owned()))
|
||||
.or_default()
|
||||
.push((track_id, file_path));
|
||||
}
|
||||
|
||||
let now = unix_now();
|
||||
let tx = conn
|
||||
.unchecked_transaction()
|
||||
.map_err(|e| LibraryError::Database(format!("begin purchase migration: {e}")))?;
|
||||
for ((album_id, format_id, folder), tracks) in groups {
|
||||
let copy_id = Uuid::new_v4().to_string();
|
||||
tx.execute(
|
||||
"INSERT INTO purchase_download_copies
|
||||
(copy_id, album_id, format_id, resolved_album_folder, created_at, updated_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?5)",
|
||||
params![copy_id, album_id, format_id, folder, now],
|
||||
)
|
||||
.map_err(|e| LibraryError::Database(format!("migrate purchase copy: {e}")))?;
|
||||
for (track_id, file_path) in tracks {
|
||||
let container = Path::new(&file_path)
|
||||
.extension()
|
||||
.and_then(|value| value.to_str())
|
||||
.map(|value| value.to_ascii_lowercase());
|
||||
tx.execute(
|
||||
"INSERT INTO purchase_download_copy_tracks
|
||||
(copy_id, track_id, file_path, container, file_size)
|
||||
VALUES (?1, ?2, ?3, ?4, 0)",
|
||||
params![copy_id, track_id, file_path, container],
|
||||
)
|
||||
.map_err(|e| LibraryError::Database(format!("migrate purchase track: {e}")))?;
|
||||
}
|
||||
}
|
||||
tx.execute(
|
||||
"INSERT OR REPLACE INTO library_kv(key, value) VALUES (?1, '1')",
|
||||
[LEGACY_MIGRATION_KEY],
|
||||
)
|
||||
.map_err(|e| LibraryError::Database(format!("write purchase migration marker: {e}")))?;
|
||||
tx.commit()
|
||||
.map_err(|e| LibraryError::Database(format!("commit purchase migration: {e}")))
|
||||
}
|
||||
|
||||
impl LibraryDatabase {
|
||||
pub fn create_purchase_copy(
|
||||
&self,
|
||||
album_id: &str,
|
||||
format_id: i64,
|
||||
resolved_album_folder: &Path,
|
||||
) -> Result<PurchaseDownloadCopy, LibraryError> {
|
||||
let copy = PurchaseDownloadCopy {
|
||||
copy_id: Uuid::new_v4().to_string(),
|
||||
album_id: album_id.to_string(),
|
||||
format_id,
|
||||
resolved_album_folder: resolved_album_folder.to_string_lossy().into_owned(),
|
||||
created_at: unix_now(),
|
||||
updated_at: unix_now(),
|
||||
};
|
||||
self.with_connection(|conn| insert_copy(conn, ©))?;
|
||||
Ok(copy)
|
||||
}
|
||||
|
||||
pub fn upsert_purchase_copy_track(
|
||||
&self,
|
||||
track: &PurchaseDownloadCopyTrack,
|
||||
) -> Result<(), LibraryError> {
|
||||
self.with_connection(|conn| {
|
||||
conn.execute(
|
||||
"INSERT INTO purchase_download_copy_tracks
|
||||
(copy_id, track_id, file_path, mime_type, container, sample_rate_hz,
|
||||
bit_depth, channels, file_size, modified_ns, last_verified_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
|
||||
ON CONFLICT(copy_id, track_id) DO UPDATE SET
|
||||
file_path=excluded.file_path, mime_type=excluded.mime_type,
|
||||
container=excluded.container, sample_rate_hz=excluded.sample_rate_hz,
|
||||
bit_depth=excluded.bit_depth, channels=excluded.channels,
|
||||
file_size=excluded.file_size, modified_ns=excluded.modified_ns,
|
||||
last_verified_at=excluded.last_verified_at",
|
||||
params![
|
||||
track.copy_id,
|
||||
track.track_id,
|
||||
track.file_path,
|
||||
track.mime_type,
|
||||
track.container,
|
||||
track.sample_rate_hz,
|
||||
track.bit_depth,
|
||||
track.channels,
|
||||
u64_to_i64(track.file_size),
|
||||
track.modified_ns.map(|value| value.to_string()),
|
||||
track.last_verified_at,
|
||||
],
|
||||
)
|
||||
.and_then(|_| {
|
||||
conn.execute(
|
||||
"UPDATE purchase_download_copies SET updated_at = ?2 WHERE copy_id = ?1",
|
||||
params![track.copy_id, unix_now()],
|
||||
)
|
||||
})
|
||||
.map(|_| ())
|
||||
.map_err(|e| LibraryError::Database(format!("upsert purchase copy track: {e}")))
|
||||
})
|
||||
}
|
||||
|
||||
pub fn get_purchase_copy(
|
||||
&self,
|
||||
copy_id: &str,
|
||||
) -> Result<Option<PurchaseDownloadCopy>, LibraryError> {
|
||||
self.with_connection(|conn| select_copy(conn, copy_id))
|
||||
}
|
||||
|
||||
pub fn get_purchase_copies_for_album(
|
||||
&self,
|
||||
album_id: &str,
|
||||
) -> Result<Vec<PurchaseDownloadCopy>, LibraryError> {
|
||||
self.with_connection(|conn| {
|
||||
let mut stmt = conn
|
||||
.prepare(
|
||||
"SELECT copy_id, album_id, format_id, resolved_album_folder,
|
||||
created_at, updated_at
|
||||
FROM purchase_download_copies WHERE album_id = ?1
|
||||
ORDER BY updated_at DESC, copy_id",
|
||||
)
|
||||
.map_err(db_error)?;
|
||||
let rows = stmt
|
||||
.query_map([album_id], copy_from_row)
|
||||
.map_err(db_error)?
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(db_error)?;
|
||||
Ok(rows)
|
||||
})
|
||||
}
|
||||
|
||||
pub fn get_purchase_copy_tracks(
|
||||
&self,
|
||||
copy_id: &str,
|
||||
) -> Result<Vec<PurchaseDownloadCopyTrack>, LibraryError> {
|
||||
self.with_connection(|conn| select_copy_tracks(conn, copy_id))
|
||||
}
|
||||
|
||||
/// Atomically point the compatibility registry at `copy_id`, but only if
|
||||
/// that one copy covers every expected album track.
|
||||
pub fn promote_complete_purchase_copy(
|
||||
&self,
|
||||
copy_id: &str,
|
||||
expected_track_ids: &[i64],
|
||||
) -> Result<bool, LibraryError> {
|
||||
self.with_connection(|conn| {
|
||||
let Some(copy) = select_copy(conn, copy_id)? else {
|
||||
return Ok(false);
|
||||
};
|
||||
let tracks = select_copy_tracks(conn, copy_id)?;
|
||||
let present: HashSet<i64> = tracks.iter().map(|track| track.track_id).collect();
|
||||
if expected_track_ids.is_empty()
|
||||
|| !expected_track_ids
|
||||
.iter()
|
||||
.all(|track_id| present.contains(track_id))
|
||||
{
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
let tx = conn.unchecked_transaction().map_err(db_error)?;
|
||||
for track in tracks
|
||||
.iter()
|
||||
.filter(|track| expected_track_ids.contains(&track.track_id))
|
||||
{
|
||||
tx.execute(
|
||||
"INSERT OR REPLACE INTO downloaded_purchases
|
||||
(track_id, format_id, album_id, file_path, downloaded_at)
|
||||
VALUES (?1, ?2, ?3, ?4, datetime('now'))",
|
||||
params![
|
||||
track.track_id,
|
||||
copy.format_id,
|
||||
copy.album_id,
|
||||
track.file_path
|
||||
],
|
||||
)
|
||||
.map_err(db_error)?;
|
||||
}
|
||||
tx.commit().map_err(db_error)?;
|
||||
Ok(true)
|
||||
})
|
||||
}
|
||||
|
||||
pub fn purchase_playback_preference(
|
||||
&self,
|
||||
album_id: &str,
|
||||
) -> Result<PurchaseAlbumPlaybackPreference, LibraryError> {
|
||||
self.with_connection(|conn| {
|
||||
conn.query_row(
|
||||
"SELECT album_id, mode, format_id, updated_at
|
||||
FROM purchase_album_playback_preferences WHERE album_id = ?1",
|
||||
[album_id],
|
||||
|row| {
|
||||
let mode = match row.get::<_, String>(1)?.as_str() {
|
||||
"purchase" => PurchasePlaybackMode::Purchase,
|
||||
_ => PurchasePlaybackMode::Qobuz,
|
||||
};
|
||||
Ok(PurchaseAlbumPlaybackPreference {
|
||||
album_id: row.get(0)?,
|
||||
mode,
|
||||
format_id: row.get(2)?,
|
||||
updated_at: row.get(3)?,
|
||||
})
|
||||
},
|
||||
)
|
||||
.optional()
|
||||
.map_err(db_error)
|
||||
.map(|value| {
|
||||
value.unwrap_or(PurchaseAlbumPlaybackPreference {
|
||||
album_id: album_id.to_string(),
|
||||
mode: PurchasePlaybackMode::Qobuz,
|
||||
format_id: None,
|
||||
updated_at: 0,
|
||||
})
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
pub fn set_purchase_playback_preference(
|
||||
&self,
|
||||
album_id: &str,
|
||||
mode: PurchasePlaybackMode,
|
||||
format_id: Option<i64>,
|
||||
) -> Result<PurchaseAlbumPlaybackPreference, LibraryError> {
|
||||
let normalized_format = match mode {
|
||||
PurchasePlaybackMode::Qobuz => None,
|
||||
PurchasePlaybackMode::Purchase => format_id,
|
||||
};
|
||||
if mode == PurchasePlaybackMode::Purchase && normalized_format.is_none() {
|
||||
return Err(LibraryError::Database(
|
||||
"purchase playback preference requires a format_id".to_string(),
|
||||
));
|
||||
}
|
||||
let value = PurchaseAlbumPlaybackPreference {
|
||||
album_id: album_id.to_string(),
|
||||
mode,
|
||||
format_id: normalized_format,
|
||||
updated_at: unix_now(),
|
||||
};
|
||||
self.with_connection(|conn| {
|
||||
conn.execute(
|
||||
"INSERT INTO purchase_album_playback_preferences
|
||||
(album_id, mode, format_id, updated_at) VALUES (?1, ?2, ?3, ?4)
|
||||
ON CONFLICT(album_id) DO UPDATE SET mode=excluded.mode,
|
||||
format_id=excluded.format_id, updated_at=excluded.updated_at",
|
||||
params![
|
||||
value.album_id,
|
||||
mode_word(value.mode),
|
||||
value.format_id,
|
||||
value.updated_at
|
||||
],
|
||||
)
|
||||
.map_err(db_error)
|
||||
})?;
|
||||
Ok(value)
|
||||
}
|
||||
|
||||
/// Snapshot all copies of an album. This performs bounded probes and must
|
||||
/// therefore be called away from a UI thread.
|
||||
pub fn inspect_purchase_copies(
|
||||
&self,
|
||||
album_id: &str,
|
||||
expected_track_ids: &[i64],
|
||||
) -> Result<Vec<PurchaseCopySnapshot>, LibraryError> {
|
||||
let expected: HashSet<i64> = expected_track_ids.iter().copied().collect();
|
||||
let mut result = Vec::new();
|
||||
for copy in self.get_purchase_copies_for_album(album_id)? {
|
||||
let rows = self.get_purchase_copy_tracks(©.copy_id)?;
|
||||
let mut tracks = Vec::with_capacity(rows.len());
|
||||
let mut healthy_ids = HashSet::new();
|
||||
let mut saw_missing = false;
|
||||
let mut saw_changed = false;
|
||||
let mut saw_unreachable = false;
|
||||
let mut saw_unreadable = false;
|
||||
for row in rows {
|
||||
let health = track_health(&row);
|
||||
match health {
|
||||
PurchaseTrackHealth::Healthy => {
|
||||
healthy_ids.insert(row.track_id);
|
||||
}
|
||||
PurchaseTrackHealth::Missing => saw_missing = true,
|
||||
PurchaseTrackHealth::Changed => saw_changed = true,
|
||||
PurchaseTrackHealth::Unreachable => saw_unreachable = true,
|
||||
PurchaseTrackHealth::Unreadable => saw_unreadable = true,
|
||||
}
|
||||
tracks.push((row, health));
|
||||
}
|
||||
let complete = !expected.is_empty()
|
||||
&& expected
|
||||
.iter()
|
||||
.all(|track_id| healthy_ids.contains(track_id));
|
||||
let health = if complete {
|
||||
PurchaseCopyHealth::CompleteHealthy
|
||||
} else if saw_unreachable {
|
||||
PurchaseCopyHealth::Unreachable
|
||||
} else if saw_changed {
|
||||
PurchaseCopyHealth::Changed
|
||||
} else if saw_unreadable {
|
||||
PurchaseCopyHealth::Unreadable
|
||||
} else if saw_missing {
|
||||
PurchaseCopyHealth::Missing
|
||||
} else {
|
||||
PurchaseCopyHealth::Partial
|
||||
};
|
||||
result.push(PurchaseCopySnapshot {
|
||||
copy,
|
||||
health,
|
||||
downloaded_tracks: tracks
|
||||
.iter()
|
||||
.filter(|(row, _)| expected.contains(&row.track_id))
|
||||
.count() as u32,
|
||||
healthy_tracks: healthy_ids.intersection(&expected).count() as u32,
|
||||
total_tracks: expected.len() as u32,
|
||||
tracks,
|
||||
});
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
}
|
||||
|
||||
fn insert_copy(conn: &Connection, copy: &PurchaseDownloadCopy) -> Result<(), LibraryError> {
|
||||
conn.execute(
|
||||
"INSERT INTO purchase_download_copies
|
||||
(copy_id, album_id, format_id, resolved_album_folder, created_at, updated_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
|
||||
params![
|
||||
copy.copy_id,
|
||||
copy.album_id,
|
||||
copy.format_id,
|
||||
copy.resolved_album_folder,
|
||||
copy.created_at,
|
||||
copy.updated_at
|
||||
],
|
||||
)
|
||||
.map(|_| ())
|
||||
.map_err(db_error)
|
||||
}
|
||||
|
||||
fn select_copy(
|
||||
conn: &Connection,
|
||||
copy_id: &str,
|
||||
) -> Result<Option<PurchaseDownloadCopy>, LibraryError> {
|
||||
conn.query_row(
|
||||
"SELECT copy_id, album_id, format_id, resolved_album_folder, created_at, updated_at
|
||||
FROM purchase_download_copies WHERE copy_id = ?1",
|
||||
[copy_id],
|
||||
copy_from_row,
|
||||
)
|
||||
.optional()
|
||||
.map_err(db_error)
|
||||
}
|
||||
|
||||
fn copy_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<PurchaseDownloadCopy> {
|
||||
Ok(PurchaseDownloadCopy {
|
||||
copy_id: row.get(0)?,
|
||||
album_id: row.get(1)?,
|
||||
format_id: row.get(2)?,
|
||||
resolved_album_folder: row.get(3)?,
|
||||
created_at: row.get(4)?,
|
||||
updated_at: row.get(5)?,
|
||||
})
|
||||
}
|
||||
|
||||
fn select_copy_tracks(
|
||||
conn: &Connection,
|
||||
copy_id: &str,
|
||||
) -> Result<Vec<PurchaseDownloadCopyTrack>, LibraryError> {
|
||||
let mut stmt = conn
|
||||
.prepare(
|
||||
"SELECT copy_id, track_id, file_path, mime_type, container, sample_rate_hz,
|
||||
bit_depth, channels, file_size, modified_ns, last_verified_at
|
||||
FROM purchase_download_copy_tracks WHERE copy_id = ?1 ORDER BY track_id",
|
||||
)
|
||||
.map_err(db_error)?;
|
||||
let rows = stmt
|
||||
.query_map([copy_id], |row| {
|
||||
let modified_ns = row
|
||||
.get::<_, Option<String>>(9)?
|
||||
.and_then(|value| value.parse::<u128>().ok());
|
||||
Ok(PurchaseDownloadCopyTrack {
|
||||
copy_id: row.get(0)?,
|
||||
track_id: row.get(1)?,
|
||||
file_path: row.get(2)?,
|
||||
mime_type: row.get(3)?,
|
||||
container: row.get(4)?,
|
||||
sample_rate_hz: row.get(5)?,
|
||||
bit_depth: row.get(6)?,
|
||||
channels: row.get(7)?,
|
||||
file_size: row.get::<_, i64>(8)?.max(0) as u64,
|
||||
modified_ns,
|
||||
last_verified_at: row.get(10)?,
|
||||
})
|
||||
})
|
||||
.map_err(db_error)?
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(db_error)?;
|
||||
Ok(rows)
|
||||
}
|
||||
|
||||
fn track_health(track: &PurchaseDownloadCopyTrack) -> PurchaseTrackHealth {
|
||||
match crate::probe_file_default(Path::new(&track.file_path)) {
|
||||
FileProbe::Missing => PurchaseTrackHealth::Missing,
|
||||
FileProbe::Unreachable => PurchaseTrackHealth::Unreachable,
|
||||
FileProbe::Present(metadata) if !metadata.is_file() => PurchaseTrackHealth::Unreadable,
|
||||
FileProbe::Present(metadata) => {
|
||||
if track.file_size > 0 && metadata.len() != track.file_size {
|
||||
return PurchaseTrackHealth::Changed;
|
||||
}
|
||||
if let Some(expected) = track.modified_ns {
|
||||
let observed = metadata
|
||||
.modified()
|
||||
.ok()
|
||||
.and_then(|value| value.duration_since(UNIX_EPOCH).ok())
|
||||
.map(|value| value.as_nanos());
|
||||
if observed != Some(expected) {
|
||||
return PurchaseTrackHealth::Changed;
|
||||
}
|
||||
}
|
||||
PurchaseTrackHealth::Healthy
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn mode_word(mode: PurchasePlaybackMode) -> &'static str {
|
||||
match mode {
|
||||
PurchasePlaybackMode::Qobuz => "qobuz",
|
||||
PurchasePlaybackMode::Purchase => "purchase",
|
||||
}
|
||||
}
|
||||
|
||||
fn unix_now() -> i64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs()
|
||||
.min(i64::MAX as u64) as i64
|
||||
}
|
||||
|
||||
fn u64_to_i64(value: u64) -> i64 {
|
||||
value.min(i64::MAX as u64) as i64
|
||||
}
|
||||
|
||||
fn db_error(error: rusqlite::Error) -> LibraryError {
|
||||
LibraryError::Database(error.to_string())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn track(copy_id: &str, id: i64, path: &Path) -> PurchaseDownloadCopyTrack {
|
||||
let metadata = std::fs::metadata(path).unwrap();
|
||||
PurchaseDownloadCopyTrack {
|
||||
copy_id: copy_id.to_string(),
|
||||
track_id: id,
|
||||
file_path: path.to_string_lossy().into_owned(),
|
||||
mime_type: Some("audio/flac".to_string()),
|
||||
container: Some("flac".to_string()),
|
||||
sample_rate_hz: Some(96_000),
|
||||
bit_depth: Some(24),
|
||||
channels: Some(2),
|
||||
file_size: metadata.len(),
|
||||
modified_ns: None,
|
||||
last_verified_at: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn copies_never_combine_coverage_across_folders() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let db = LibraryDatabase::open(&root.path().join("library.db")).unwrap();
|
||||
let first_dir = root.path().join("first");
|
||||
let second_dir = root.path().join("second");
|
||||
std::fs::create_dir_all(&first_dir).unwrap();
|
||||
std::fs::create_dir_all(&second_dir).unwrap();
|
||||
let first = db.create_purchase_copy("album", 7, &first_dir).unwrap();
|
||||
let second = db.create_purchase_copy("album", 7, &second_dir).unwrap();
|
||||
let one = first_dir.join("01.flac");
|
||||
let two = second_dir.join("02.flac");
|
||||
std::fs::write(&one, b"one").unwrap();
|
||||
std::fs::write(&two, b"two").unwrap();
|
||||
db.upsert_purchase_copy_track(&track(&first.copy_id, 1, &one))
|
||||
.unwrap();
|
||||
db.upsert_purchase_copy_track(&track(&second.copy_id, 2, &two))
|
||||
.unwrap();
|
||||
|
||||
let snapshots = db.inspect_purchase_copies("album", &[1, 2]).unwrap();
|
||||
assert_eq!(snapshots.len(), 2);
|
||||
assert!(snapshots
|
||||
.iter()
|
||||
.all(|copy| copy.health == PurchaseCopyHealth::Partial));
|
||||
assert!(!db
|
||||
.promote_complete_purchase_copy(&first.copy_id, &[1, 2])
|
||||
.unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_complete_copy_promotes_legacy_atomically() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let db = LibraryDatabase::open(&root.path().join("library.db")).unwrap();
|
||||
let copy = db.create_purchase_copy("album", 55, root.path()).unwrap();
|
||||
for id in [1, 2] {
|
||||
let path = root.path().join(format!("{id}.dsf"));
|
||||
std::fs::write(&path, b"dsd").unwrap();
|
||||
db.upsert_purchase_copy_track(&track(©.copy_id, id, &path))
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
assert!(db
|
||||
.promote_complete_purchase_copy(©.copy_id, &[1, 2])
|
||||
.unwrap());
|
||||
assert_eq!(
|
||||
db.get_downloaded_purchase_formats().unwrap(),
|
||||
vec![(1, 55), (2, 55)]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn preference_defaults_to_qobuz_and_round_trips_exact_format() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let db = LibraryDatabase::open(&root.path().join("library.db")).unwrap();
|
||||
let default = db.purchase_playback_preference("album").unwrap();
|
||||
assert_eq!(default.mode, PurchasePlaybackMode::Qobuz);
|
||||
assert_eq!(default.format_id, None);
|
||||
|
||||
db.set_purchase_playback_preference("album", PurchasePlaybackMode::Purchase, Some(55))
|
||||
.unwrap();
|
||||
let stored = db.purchase_playback_preference("album").unwrap();
|
||||
assert_eq!(stored.mode, PurchasePlaybackMode::Purchase);
|
||||
assert_eq!(stored.format_id, Some(55));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn legacy_migration_groups_by_folder() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let db_path = root.path().join("library.db");
|
||||
{
|
||||
let db = LibraryDatabase::open(&db_path).unwrap();
|
||||
db.with_connection(|conn| {
|
||||
conn.execute(
|
||||
"DELETE FROM library_kv WHERE key = ?1",
|
||||
[LEGACY_MIGRATION_KEY],
|
||||
)
|
||||
.unwrap();
|
||||
});
|
||||
db.mark_purchase_downloaded(1, Some("album"), "/a/01.flac", 7)
|
||||
.unwrap();
|
||||
db.mark_purchase_downloaded(2, Some("album"), "/b/02.flac", 7)
|
||||
.unwrap();
|
||||
}
|
||||
let db = LibraryDatabase::open(&db_path).unwrap();
|
||||
let copies = db.get_purchase_copies_for_album("album").unwrap();
|
||||
assert_eq!(copies.len(), 2);
|
||||
assert_ne!(
|
||||
copies[0].resolved_album_folder,
|
||||
copies[1].resolved_album_folder
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -35,6 +35,8 @@
|
||||
//! into the outage it was meant to prevent.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::fs::Metadata;
|
||||
use std::io::ErrorKind;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
use std::time::{Duration, Instant};
|
||||
@@ -63,6 +65,15 @@ pub enum Reach {
|
||||
Unreachable,
|
||||
}
|
||||
|
||||
/// A bounded metadata probe for callers that need to validate a regular file
|
||||
/// without doing a second, potentially blocking filesystem syscall.
|
||||
#[derive(Debug)]
|
||||
pub enum FileProbe {
|
||||
Present(Metadata),
|
||||
Missing,
|
||||
Unreachable,
|
||||
}
|
||||
|
||||
impl Reach {
|
||||
/// Can playback proceed?
|
||||
pub fn is_playable(self) -> bool {
|
||||
@@ -167,6 +178,51 @@ pub fn probe_default(path: &Path) -> Reach {
|
||||
probe(path, DEFAULT_PROBE)
|
||||
}
|
||||
|
||||
/// Read metadata for `path`, bounded by `deadline`.
|
||||
///
|
||||
/// Only `NotFound` is classified as [`FileProbe::Missing`]. Permission errors,
|
||||
/// I/O failures, symlink loops, and timeouts preserve durable registry state by
|
||||
/// returning [`FileProbe::Unreachable`].
|
||||
pub fn probe_file(path: &Path, deadline: Duration) -> FileProbe {
|
||||
if is_cooling_down(path) {
|
||||
return FileProbe::Unreachable;
|
||||
}
|
||||
let (tx, rx) = std::sync::mpsc::channel();
|
||||
let probe_path = path.to_path_buf();
|
||||
std::thread::Builder::new()
|
||||
.name("qbz-file-probe".to_string())
|
||||
.spawn(move || {
|
||||
let _ = tx.send(std::fs::metadata(probe_path));
|
||||
})
|
||||
.ok();
|
||||
|
||||
match rx.recv_timeout(deadline) {
|
||||
Ok(Ok(metadata)) => FileProbe::Present(metadata),
|
||||
Ok(Err(error)) if error.kind() == ErrorKind::NotFound => FileProbe::Missing,
|
||||
Ok(Err(error)) => {
|
||||
log::warn!(
|
||||
"[reach] file metadata returned an I/O error — treating it as unreachable: {}",
|
||||
error
|
||||
);
|
||||
mark_unreachable(path);
|
||||
FileProbe::Unreachable
|
||||
}
|
||||
Err(_) => {
|
||||
log::warn!(
|
||||
"[reach] file metadata did not answer in {:?} — treating it as unreachable for {:?}",
|
||||
deadline,
|
||||
COOLDOWN
|
||||
);
|
||||
mark_unreachable(path);
|
||||
FileProbe::Unreachable
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn probe_file_default(path: &Path) -> FileProbe {
|
||||
probe_file(path, DEFAULT_PROBE)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -193,7 +249,9 @@ mod tests {
|
||||
mark_unreachable(&p);
|
||||
assert!(is_cooling_down(&p));
|
||||
// A sibling under the SAME mount is covered by the one probe.
|
||||
assert!(is_cooling_down(Path::new("/mnt/qbz-test-dead/other/x.flac")));
|
||||
assert!(is_cooling_down(Path::new(
|
||||
"/mnt/qbz-test-dead/other/x.flac"
|
||||
)));
|
||||
assert_eq!(probe_default(&p), Reach::Unreachable);
|
||||
reset_cooldowns();
|
||||
assert!(!is_cooling_down(&p));
|
||||
@@ -223,4 +281,22 @@ mod tests {
|
||||
assert_eq!(probe_default(&loop_path), Reach::Unreachable);
|
||||
reset_cooldowns();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn file_probe_returns_metadata_and_only_not_found_is_missing() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let file = root.path().join("track.flac");
|
||||
std::fs::write(&file, b"audio").unwrap();
|
||||
match probe_file_default(&file) {
|
||||
FileProbe::Present(metadata) => {
|
||||
assert!(metadata.is_file());
|
||||
assert_eq!(metadata.len(), 5);
|
||||
}
|
||||
other => panic!("unexpected probe: {other:?}"),
|
||||
}
|
||||
assert!(matches!(
|
||||
probe_file_default(&root.path().join("absent")),
|
||||
FileProbe::Missing
|
||||
));
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user