diff --git a/crates/qbz-jellyfin/src/lib.rs b/crates/qbz-jellyfin/src/lib.rs index 5c78ab48e..47bbe254f 100644 --- a/crates/qbz-jellyfin/src/lib.rs +++ b/crates/qbz-jellyfin/src/lib.rs @@ -22,13 +22,14 @@ //! exists so a regression is caught rather than quietly resampled. QBZ never //! asks for a transcode and must never accept one. //! -//! **2. Quality arrives with the listing, but it is not free.** `BitDepth` and +//! **2. Quality is available with the listing, but it is not free.** `BitDepth` and //! `SampleRate` live in `MediaSources[].MediaStreams[]`, which requires //! `Fields=MediaSources`. Measured on a 500-item page: 0.25 s without it, 4.56 s //! with. Full sweep of 4924 tracks: **45.8 s**. `Fields=MediaStreams` trims 29 % //! of the bytes and saves nothing — the cost is server-side media-info -//! hydration, not transfer. There is no cheaper path, so the scan owns that cost -//! deliberately rather than discovering it. +//! hydration, not transfer. Catalog discovery therefore omits `MediaSources`; +//! a secondary, resumable hydration pass requests it only for bounded id +//! batches, with visible and queued rows allowed to jump ahead. //! //! **3. Artwork needs no credentials.** `/Items/{id}/Images/Primary` answers //! 200 unauthenticated. Unlike Plex, whose thumb url has to be re-tokenized on @@ -51,11 +52,15 @@ use serde::Deserialize; /// FLAC over a slow link legitimately outlives any probe budget. const HTTP_TIMEOUT: Duration = Duration::from_secs(15); -/// Page size for the library sweep. 500 is the measured sweet spot: the -/// per-request cost is dominated by server-side media-info hydration, so fewer, -/// larger pages win, and 500 items with `MediaSources` is ~1.5 MB of JSON — -/// large enough to amortise, small enough not to spike RSS. -pub const PAGE_SIZE: u32 = 500; +/// Page size for the cheap essential-metadata sweep. It stays inside the +/// catalog's initial 100–250 row budget; the expensive quality pass has its +/// own, smaller batch size. +pub const PAGE_SIZE: u32 = 250; + +/// Maximum ids in one secondary `MediaSources` request. A priority batch must +/// be small enough that newly visible/queued tracks are not stuck behind +/// several seconds of unrelated server-side probing. +pub const QUALITY_BATCH_SIZE: usize = 50; /// Cover size this client requests, app-wide. One size is deliberate, for the /// same reason `artwork_qt::PLEX_THUMB_PX` is: the cache key is the url, so a @@ -177,6 +182,20 @@ pub struct JellyfinTrack { pub server_path: Option, } +/// Quality fields hydrated independently from the essential track listing. +/// The item id is the join key; no list position is trusted because Jellyfin +/// may omit an item deleted between discovery and hydration. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct JellyfinTrackQuality { + pub id: String, + pub container: String, + pub codec: Option, + pub bit_depth: Option, + pub sample_rate_hz: Option, + pub channels: Option, + pub bitrate_bps: Option, +} + impl JellyfinTrack { /// Jellyfin counts in 100 ns ticks. fn ticks_to_ms(ticks: i64) -> u64 { @@ -277,6 +296,39 @@ struct MediaStreamDto { } impl AudioDto { + fn audio_stream(&self) -> Option<&MediaStreamDto> { + self.media_sources.first().and_then(|source| { + source + .media_streams + .iter() + .find(|stream| stream.kind == "Audio") + }) + } + + fn into_quality(self) -> JellyfinTrackQuality { + let (codec, bit_depth, sample_rate_hz, channels, bitrate_bps) = self + .audio_stream() + .map(|stream| { + ( + stream.codec.clone(), + stream.bit_depth, + stream.sample_rate, + stream.channels, + stream.bit_rate, + ) + }) + .unwrap_or_default(); + JellyfinTrackQuality { + id: self.id, + container: self.container.unwrap_or_default(), + codec, + bit_depth, + sample_rate_hz, + channels, + bitrate_bps, + } + } + fn into_track(self) -> JellyfinTrack { // The FIRST audio stream. A cover embedded as `EmbeddedImage` shows up // in the same array, which is why this filters by type rather than @@ -395,6 +447,24 @@ fn client() -> Result { .map_err(|e| JellyfinError::Transport(e.to_string())) } +/// Reqwest's Display text contains the request URL for common failures. Server +/// addresses are private configuration and authenticated stream URLs carry a +/// token, so protocol errors retain only a coarse, non-sensitive reason. +fn transport_error(error: &reqwest::Error) -> JellyfinError { + let reason = if error.is_timeout() { + "request timed out" + } else if error.is_connect() { + "connection failed" + } else if error.is_decode() { + "response decoding failed" + } else if error.is_body() { + "response body failed" + } else { + "request failed" + }; + JellyfinError::Transport(reason.to_string()) +} + fn check(status: reqwest::StatusCode) -> Result<()> { if status.is_success() { return Ok(()); @@ -415,7 +485,7 @@ pub async fn probe(base_url: &str) -> Result { .get(format!("{base}/System/Info/Public")) .send() .await - .map_err(|e| JellyfinError::Transport(e.to_string()))?; + .map_err(|e| transport_error(&e))?; check(resp.status())?; resp.json::() .await @@ -436,7 +506,7 @@ pub async fn authenticate( .json(&serde_json::json!({ "Username": username, "Pw": password })) .send() .await - .map_err(|e| JellyfinError::Transport(e.to_string()))?; + .map_err(|e| transport_error(&e))?; check(resp.status())?; let dto: AuthDto = resp .json() @@ -472,7 +542,7 @@ impl JellyfinClient { .header("X-Emby-Token", &self.token) .send() .await - .map_err(|e| JellyfinError::Transport(e.to_string()))?; + .map_err(|e| transport_error(&e))?; check(resp.status())?; resp.json::() .await @@ -516,7 +586,25 @@ impl JellyfinClient { Ok(env.total_record_count) } - /// One page of audio items, WITH quality. + /// One page of essential audio metadata, without `MediaSources`. + /// + /// This is the first useful pass. On the measured server it avoids the + /// per-track media-info probe that made the old first sync take 45.8 s. + /// The returned quality fields are intentionally absent until + /// [`track_quality`] fills them in the secondary pass. + pub async fn essential_tracks_page( + &self, + library_id: Option<&str>, + start_index: u64, + min_date_last_saved: Option<&str>, + ) -> Result<(Vec, u64)> { + self.tracks_page_with_fields(library_id, start_index, min_date_last_saved, false) + .await + } + + /// Compatibility entry point for callers that explicitly need quality in + /// the listing. The Qt catalog sync uses [`essential_tracks_page`] and the + /// bounded [`track_quality`] path instead. /// /// `Fields=MediaSources` is what carries `BitDepth` / `SampleRate`, and it /// is the expensive part (§ module docs): ~4.5 s per 500-item page against @@ -532,20 +620,24 @@ impl JellyfinClient { start_index: u64, min_date_last_saved: Option<&str>, ) -> Result<(Vec, u64)> { - let scope = library_id - .map(|id| format!("&parentId={id}")) - .unwrap_or_default(); - let delta = min_date_last_saved - .map(|d| format!("&minDateLastSaved={d}")) - .unwrap_or_default(); + self.tracks_page_with_fields(library_id, start_index, min_date_last_saved, true) + .await + } + + async fn tracks_page_with_fields( + &self, + library_id: Option<&str>, + start_index: u64, + min_date_last_saved: Option<&str>, + include_quality: bool, + ) -> Result<(Vec, u64)> { let env: ItemsEnvelope = self - .get_json(&format!( - "/Items?userId={}&IncludeItemTypes=Audio&Recursive=true\ - &Limit={PAGE_SIZE}&StartIndex={start_index}\ - &Fields=Path,MediaSources,ParentId,ProductionYear,Genres\ - &SortBy=Album,ParentIndexNumber,IndexNumber&SortOrder=Ascending\ - &EnableImages=true&ImageTypeLimit=1{scope}{delta}", - self.user_id + .get_json(&tracks_page_path( + &self.user_id, + library_id, + start_index, + min_date_last_saved, + include_quality, )) .await?; let total = env.total_record_count; @@ -555,6 +647,31 @@ impl JellyfinClient { )) } + /// Hydrate quality for a bounded set of opaque item ids. + /// + /// The response is joined by id, never by position: an item may disappear + /// between the essential pass and this request. Missing ids are simply not + /// returned so the caller can defer them without corrupting a neighbour. + pub async fn track_quality(&self, item_ids: &[String]) -> Result> { + if item_ids.is_empty() { + return Ok(Vec::new()); + } + if item_ids.len() > QUALITY_BATCH_SIZE { + return Err(JellyfinError::Decode(format!( + "quality batch exceeds {QUALITY_BATCH_SIZE} ids" + ))); + } + let ids = item_ids.join(","); + let env: ItemsEnvelope = self + .get_json(&format!( + "/Items?userId={}&Ids={ids}&IncludeItemTypes=Audio\ + &Limit={QUALITY_BATCH_SIZE}&Fields=MediaSources", + self.user_id + )) + .await?; + Ok(env.items.into_iter().map(AudioDto::into_quality).collect()) + } + /// [`stream_url`] for this client's server. pub fn stream_url(&self, item_id: &str) -> String { stream_url(&self.base, &self.token, item_id) @@ -566,6 +683,35 @@ impl JellyfinClient { } } +fn tracks_page_path( + user_id: &str, + library_id: Option<&str>, + start_index: u64, + min_date_last_saved: Option<&str>, + include_quality: bool, +) -> String { + let scope = library_id + .map(|id| format!("&parentId={id}")) + .unwrap_or_default(); + let delta = min_date_last_saved + .map(|date| format!("&minDateLastSaved={date}")) + .unwrap_or_default(); + let fields = if include_quality { + "Path,MediaSources,ParentId,ProductionYear,Genres" + } else { + // A server-side filesystem path is neither useful nor safe catalog + // metadata on a client. The compatibility path keeps exposing it, but + // the cheap persistent-library pass does not request or store it. + "ParentId,ProductionYear,Genres" + }; + format!( + "/Items?userId={user_id}&IncludeItemTypes=Audio&Recursive=true\ + &Limit={PAGE_SIZE}&StartIndex={start_index}&Fields={fields}\ + &SortBy=Album,ParentIndexNumber,IndexNumber&SortOrder=Ascending\ + &EnableImages=true&ImageTypeLimit=1{scope}{delta}" + ) +} + // --------------------------------------------------------------------------- // URL builders — PURE, and deliberately outside the client // --------------------------------------------------------------------------- @@ -635,9 +781,15 @@ mod tests { #[test] fn base_urls_are_normalised_the_way_people_type_them() { - assert_eq!(normalize_base_url("192.168.0.69:8096"), "http://192.168.0.69:8096"); + assert_eq!( + normalize_base_url("192.168.0.69:8096"), + "http://192.168.0.69:8096" + ); assert_eq!(normalize_base_url("http://host:8096/"), "http://host:8096"); - assert_eq!(normalize_base_url(" https://jf.example.com/ "), "https://jf.example.com"); + assert_eq!( + normalize_base_url(" https://jf.example.com/ "), + "https://jf.example.com" + ); assert_eq!(normalize_base_url(" "), ""); } @@ -687,8 +839,14 @@ mod tests { ] }] }); - let t = serde_json::from_value::(json).unwrap().into_track(); - assert_eq!(t.bit_depth, Some(24), "read the cover's depth, not the audio's"); + let t = serde_json::from_value::(json) + .unwrap() + .into_track(); + assert_eq!( + t.bit_depth, + Some(24), + "read the cover's depth, not the audio's" + ); assert_eq!(t.sample_rate_hz, Some(96000)); assert_eq!(t.codec.as_deref(), Some("flac")); // 3 484 800 000 ticks / 10 000 = 348 480 ms = 348.48 s. @@ -697,6 +855,110 @@ mod tests { assert_eq!(t.disc_number, Some(1)); } + #[test] + fn essential_page_omits_media_sources_but_preserves_delta_and_scope() { + let essential = tracks_page_path( + "user", + Some("library"), + 250, + Some("2026-08-23T00:00:00Z"), + false, + ); + assert!(essential.contains("Limit=250&StartIndex=250")); + assert!(essential.contains("parentId=library")); + assert!(essential.contains("minDateLastSaved=2026-08-23T00:00:00Z")); + assert!(!essential.contains("MediaSources")); + assert!(!essential.contains("Path")); + + let compatibility = tracks_page_path("user", None, 0, None, true); + assert!(compatibility.contains("MediaSources")); + } + + #[test] + fn secondary_quality_is_keyed_by_item_and_uses_only_the_audio_stream() { + let json = serde_json::json!({ + "Id": "quality-id", + "Container": "flac", + "MediaSources": [{ "MediaStreams": [ + { "Type": "EmbeddedImage", "Codec": "mjpeg", "BitDepth": 8 }, + { "Type": "Audio", "Codec": "flac", "BitDepth": 24, + "SampleRate": 192000, "Channels": 2, "BitRate": 6112000 } + ]}] + }); + let quality = serde_json::from_value::(json) + .unwrap() + .into_quality(); + assert_eq!(quality.id, "quality-id"); + assert_eq!(quality.container, "flac"); + assert_eq!(quality.codec.as_deref(), Some("flac")); + assert_eq!(quality.bit_depth, Some(24)); + assert_eq!(quality.sample_rate_hz, Some(192_000)); + assert_eq!(quality.bitrate_bps, Some(6_112_000)); + } + + #[test] + fn essential_first_pass_metric_is_bounded_and_smaller_than_quality_payloads() { + const TRACKS: u64 = 4_924; + fn page(start: u64, end: u64, quality: bool) -> String { + let mut items = Vec::new(); + for index in start..end { + let mut item = serde_json::json!({ + "Id": format!("item-{index:05}"), + "Name": format!("Track {index:05}"), + "Album": format!("Album {:04}", index / 10), + "AlbumId": format!("album-{:04}", index / 10), + "AlbumArtist": "Fixture Artist", + "Artists": ["Fixture Artist"], + "IndexNumber": index % 10 + 1, + "ParentIndexNumber": 1, + "RunTimeTicks": 1_800_000_000i64, + "ProductionYear": 2026, + "Genres": ["Fixture"], + "Container": "flac" + }); + if quality { + item["MediaSources"] = serde_json::json!([{ + "MediaStreams": [{ + "Type": "Audio", "Codec": "flac", "BitDepth": 24, + "SampleRate": 96000, "Channels": 2, "BitRate": 3120000 + }] + }]); + } + items.push(item); + } + serde_json::json!({ "Items": items, "TotalRecordCount": TRACKS }).to_string() + } + + let mut essential_pages = Vec::new(); + let mut quality_pages = Vec::new(); + for start in (0..TRACKS).step_by(PAGE_SIZE as usize) { + let end = (start + u64::from(PAGE_SIZE)).min(TRACKS); + essential_pages.push(page(start, end, false)); + quality_pages.push(page(start, end, true)); + } + let essential_bytes = essential_pages.iter().map(String::len).max().unwrap_or(0); + let quality_bytes = quality_pages.iter().map(String::len).max().unwrap_or(0); + let mut parsed = 0usize; + let essential_started = std::time::Instant::now(); + for json in &essential_pages { + let envelope: ItemsEnvelope = serde_json::from_str(json).unwrap(); + parsed += envelope.items.len(); + } + let essential_elapsed = essential_started.elapsed(); + let quality_started = std::time::Instant::now(); + for json in &quality_pages { + let _: ItemsEnvelope = serde_json::from_str(json).unwrap(); + } + let quality_elapsed = quality_started.elapsed(); + assert_eq!(parsed, TRACKS as usize); + assert!(essential_bytes < quality_bytes); + println!( + "H_JELLYFIN_METRIC tracks={TRACKS} pages=20 essential_max_bytes={essential_bytes} quality_max_bytes={quality_bytes} essential_parse_ms={} quality_parse_ms={}", + essential_elapsed.as_millis(), + quality_elapsed.as_millis(), + ); + } + /// MP3 reports NO bit depth. Defaulting it to 16 would badge every lossy /// track as CD quality, which is the one thing this app must not get wrong. #[test] @@ -707,7 +969,9 @@ mod tests { { "Type": "Audio", "Codec": "mp3", "SampleRate": 44100, "Channels": 2 } ]}] }); - let t = serde_json::from_value::(json).unwrap().into_track(); + let t = serde_json::from_value::(json) + .unwrap() + .into_track(); assert_eq!(t.bit_depth, None); assert_eq!(t.sample_rate_hz, Some(44100)); } @@ -717,7 +981,9 @@ mod tests { #[test] fn a_row_missing_its_tags_still_maps() { let json = serde_json::json!({ "Id": "bare", "Name": "Untitled" }); - let t = serde_json::from_value::(json).unwrap().into_track(); + let t = serde_json::from_value::(json) + .unwrap() + .into_track(); assert_eq!(t.id, "bare"); assert_eq!(t.artist, ""); assert_eq!(t.track_number, None); @@ -733,7 +999,9 @@ mod tests { let json = serde_json::json!({ "Id": "a", "Name": "t", "AlbumArtist": "Various Artists", "Artists": [] }); - let t = serde_json::from_value::(json).unwrap().into_track(); + let t = serde_json::from_value::(json) + .unwrap() + .into_track(); assert_eq!(t.artist, "Various Artists"); assert_eq!(t.album_artist, "Various Artists"); } @@ -741,7 +1009,10 @@ mod tests { #[test] fn image_urls_carry_no_credentials_and_bust_on_the_tag() { let with = image_url("http://h:8096", "alb", Some("deadbeef"), IMAGE_PX); - assert!(!with.contains("tok"), "a cover url must not carry the token"); + assert!( + !with.contains("tok"), + "a cover url must not carry the token" + ); assert!(with.contains("tag=deadbeef")); assert!(with.contains("maxWidth=256")); assert_eq!( @@ -755,8 +1026,14 @@ mod tests { #[test] fn the_stream_url_asks_for_the_original_bytes_and_nothing_else() { let u = stream_url("http://h:8096/", "tok", "item42"); - assert_eq!(u, "http://h:8096/Audio/item42/stream?static=true&api_key=tok"); - assert!(!u.contains("audioCodec"), "a codec parameter IS a transcode"); + assert_eq!( + u, + "http://h:8096/Audio/item42/stream?static=true&api_key=tok" + ); + assert!( + !u.contains("audioCodec"), + "a codec parameter IS a transcode" + ); assert!(!u.contains("audioBitRate")); } diff --git a/crates/qbz-media-cache/src/lib.rs b/crates/qbz-media-cache/src/lib.rs index 89ab1ba71..0cd0f2cb6 100644 --- a/crates/qbz-media-cache/src/lib.rs +++ b/crates/qbz-media-cache/src/lib.rs @@ -56,6 +56,7 @@ //! to remove, so it is not worth reintroducing for a marginally simpler write. use std::path::Path; +use std::time::Duration; use rusqlite::{params, Connection, OptionalExtension}; @@ -166,6 +167,27 @@ pub struct CachedTrack { pub size_bytes: Option, } +/// Quality-only update from a secondary server hydration pass. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct CachedTrackQuality { + pub item_id: String, + pub container: String, + pub codec: Option, + pub bit_depth: Option, + pub sample_rate_hz: Option, + pub channels: Option, + pub bitrate_kbps: Option, +} + +/// One source sync generation. The authoritative cache remains readable while +/// a newer generation is incomplete; only completion may prune older rows. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SourceSyncGeneration { + pub generation: u64, + pub observed_rows: u64, + pub resumed: bool, +} + /// One source-local artist aggregate for the Local Library rail. /// /// A row is emitted for both a track artist and a distinct album artist. The @@ -203,6 +225,8 @@ pub fn open(path: &Path) -> Result { std::fs::create_dir_all(dir).map_err(map_err("create dir"))?; } let conn = Connection::open(path).map_err(map_err("open"))?; + conn.busy_timeout(Duration::from_millis(2_500)) + .map_err(map_err("busy timeout"))?; init_schema(&conn)?; Ok(conn) } @@ -251,10 +275,21 @@ pub fn init_schema(conn: &Connection) -> Result<()> { bitrate_kbps INTEGER, artwork_token TEXT, size_bytes INTEGER, + observed_generation INTEGER NOT NULL DEFAULT 0, + quality_hydrated INTEGER NOT NULL DEFAULT 0, + quality_retry_at INTEGER NOT NULL DEFAULT 0, updated_at INTEGER NOT NULL, UNIQUE (source, item_id) ); + CREATE TABLE IF NOT EXISTS remote_cache_source_sync ( + source TEXT PRIMARY KEY, + generation INTEGER NOT NULL DEFAULT 0, + observed_rows INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'idle', + updated_at INTEGER NOT NULL DEFAULT 0 + ); + CREATE INDEX IF NOT EXISTS idx_rct_source_album ON remote_cache_tracks(source, album_id); CREATE INDEX IF NOT EXISTS idx_rct_source_library @@ -265,7 +300,30 @@ pub fn init_schema(conn: &Connection) -> Result<()> { CREATE INDEX IF NOT EXISTS idx_rct_title ON remote_cache_tracks(title); ", ) - .map_err(map_err("schema")) + .map_err(map_err("schema"))?; + + // Forward-only additive migration for caches created before incremental + // source generations and deferred Jellyfin quality existed. + for column in [ + "observed_generation INTEGER NOT NULL DEFAULT 0", + "quality_hydrated INTEGER NOT NULL DEFAULT 0", + "quality_retry_at INTEGER NOT NULL DEFAULT 0", + ] { + let statement = format!("ALTER TABLE remote_cache_tracks ADD COLUMN {column}"); + let _ = conn.execute(&statement, []); + } + let _ = conn.execute( + "ALTER TABLE remote_cache_source_sync + ADD COLUMN observed_rows INTEGER NOT NULL DEFAULT 0", + [], + ); + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_rct_quality_pending + ON remote_cache_tracks(source,quality_hydrated,quality_retry_at,updated_at)", + [], + ) + .map_err(map_err("quality index"))?; + Ok(()) } fn now() -> i64 { @@ -300,7 +358,9 @@ fn row_to_track(row: &rusqlite::Row<'_>) -> rusqlite::Result { container: row.get("container")?, codec: row.get("codec")?, bit_depth: row.get::<_, Option>("bit_depth")?.map(|v| v as u32), - sample_rate_hz: row.get::<_, Option>("sample_rate_hz")?.map(|v| v as u32), + sample_rate_hz: row + .get::<_, Option>("sample_rate_hz")? + .map(|v| v as u32), channels: row.get::<_, Option>("channels")?.map(|v| v as u32), bitrate_kbps: row.get::<_, Option>("bitrate_kbps")?.map(|v| v as u32), artwork_token: row.get("artwork_token")?, @@ -320,7 +380,11 @@ const SELECT: &str = "SELECT id, source, item_id, server_id, library_id, title, /// `session.db` row), so a re-scan must not mint a new one for a track that was /// already there. A user who re-syncs while a Jellyfin track is playing would /// otherwise find the queue pointing at nothing. -pub fn save_tracks(conn: &mut Connection, source: RemoteSource, tracks: &[CachedTrack]) -> Result { +pub fn save_tracks( + conn: &mut Connection, + source: RemoteSource, + tracks: &[CachedTrack], +) -> Result { if tracks.is_empty() { return Ok(0); } @@ -333,8 +397,8 @@ pub fn save_tracks(conn: &mut Connection, source: RemoteSource, tracks: &[Cached (source, item_id, server_id, library_id, title, artist, album_artist, album, album_id, track_number, disc_number, duration_ms, year, genre, container, codec, bit_depth, sample_rate_hz, channels, bitrate_kbps, artwork_token, - size_bytes, updated_at) - VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?22,?23) + size_bytes, quality_hydrated, quality_retry_at, updated_at) + VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?22,1,0,?23) ON CONFLICT(source, item_id) DO UPDATE SET server_id=excluded.server_id, library_id=excluded.library_id, title=excluded.title, artist=excluded.artist, @@ -345,7 +409,8 @@ pub fn save_tracks(conn: &mut Connection, source: RemoteSource, tracks: &[Cached codec=excluded.codec, bit_depth=excluded.bit_depth, sample_rate_hz=excluded.sample_rate_hz, channels=excluded.channels, bitrate_kbps=excluded.bitrate_kbps, artwork_token=excluded.artwork_token, - size_bytes=excluded.size_bytes, updated_at=excluded.updated_at", + size_bytes=excluded.size_bytes, quality_hydrated=1, + quality_retry_at=0, updated_at=excluded.updated_at", ) .map_err(map_err("prepare insert"))?; for t in tracks { @@ -381,8 +446,375 @@ pub fn save_tracks(conn: &mut Connection, source: RemoteSource, tracks: &[Cached Ok(tracks.len()) } +fn next_source_revision(conn: &Connection, source: RemoteSource) -> Result { + let current = conn + .query_row( + "SELECT COALESCE(MAX(updated_at),0) FROM remote_cache_tracks WHERE source=?1", + params![source.as_str()], + |row| row.get::<_, i64>(0), + ) + .map_err(map_err("read source revision"))?; + Ok(now().max(current.saturating_add(1))) +} + +/// Start a fresh authoritative generation for one remote source. +/// +/// A previous interrupted generation is deliberately not resumed here: the +/// Jellyfin essential pass is cheap and its offset spans multiple libraries. +/// Long-running quality hydration resumes independently from persisted pending +/// rows, while stale catalog rows remain intact until this generation finishes. +pub fn begin_source_sync( + conn: &mut Connection, + source: RemoteSource, +) -> Result { + let tx = conn.transaction().map_err(map_err("begin source sync"))?; + let previous = tx + .query_row( + "SELECT generation FROM remote_cache_source_sync WHERE source=?1", + params![source.as_str()], + |row| row.get::<_, i64>(0), + ) + .optional() + .map_err(map_err("read source sync"))? + .unwrap_or(0) + .max(0) as u64; + let generation = previous.saturating_add(1).min(i64::MAX as u64).max(1); + tx.execute( + "INSERT INTO remote_cache_source_sync(source,generation,observed_rows,status,updated_at) + VALUES (?1,?2,0,'running',?3) + ON CONFLICT(source) DO UPDATE SET generation=excluded.generation, + observed_rows=0,status='running',updated_at=excluded.updated_at", + params![source.as_str(), generation as i64, now()], + ) + .map_err(map_err("write source sync"))?; + tx.commit().map_err(map_err("commit source sync"))?; + Ok(SourceSyncGeneration { + generation, + observed_rows: 0, + resumed: false, + }) +} + +/// Upsert one cheap metadata page and mark its quality pending. +/// +/// Existing quality values remain visible until their replacement arrives; +/// `quality_hydrated=0` is the separate truth that makes even a legitimately +/// NULL lossy bit depth distinguishable from "not fetched yet". +pub fn save_essential_tracks( + conn: &mut Connection, + source: RemoteSource, + generation: u64, + tracks: &[CachedTrack], +) -> Result { + if tracks.is_empty() { + return Ok(0); + } + let tx = conn + .transaction() + .map_err(map_err("begin essential page"))?; + let state = tx + .query_row( + "SELECT status,observed_rows FROM remote_cache_source_sync + WHERE source=?1 AND generation=?2", + params![source.as_str(), generation.min(i64::MAX as u64) as i64], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)), + ) + .optional() + .map_err(map_err("read essential generation"))?; + let Some((status, observed_rows)) = state else { + return Err("media cache: essential generation is missing".to_string()); + }; + if status != "running" { + return Err("media cache: essential generation is no longer current".to_string()); + } + let revision = next_source_revision(&tx, source)?; + let generation = generation.min(i64::MAX as u64) as i64; + { + let mut stmt = tx + .prepare( + "INSERT INTO remote_cache_tracks + (source,item_id,server_id,library_id,title,artist,album_artist,album, + album_id,track_number,disc_number,duration_ms,year,genre,container, + codec,bit_depth,sample_rate_hz,channels,bitrate_kbps,artwork_token, + size_bytes,observed_generation,quality_hydrated,quality_retry_at,updated_at) + VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15, + ?16,?17,?18,?19,?20,?21,?22,?23,0,0,?24) + ON CONFLICT(source,item_id) DO UPDATE SET + server_id=excluded.server_id,library_id=excluded.library_id, + title=excluded.title,artist=excluded.artist, + album_artist=excluded.album_artist,album=excluded.album, + album_id=excluded.album_id,track_number=excluded.track_number, + disc_number=excluded.disc_number,duration_ms=excluded.duration_ms, + year=excluded.year,genre=excluded.genre, + container=COALESCE(NULLIF(excluded.container,''),remote_cache_tracks.container), + artwork_token=excluded.artwork_token,size_bytes=excluded.size_bytes, + observed_generation=excluded.observed_generation,quality_hydrated=0, + quality_retry_at=0,updated_at=excluded.updated_at", + ) + .map_err(map_err("prepare essential page"))?; + for track in tracks { + stmt.execute(params![ + source.as_str(), + track.item_id, + track.server_id, + track.library_id, + track.title, + track.artist, + track.album_artist, + track.album, + track.album_id, + track.track_number.map(|value| value as i64), + track.disc_number.map(|value| value as i64), + track.duration_ms.min(i64::MAX as u64) as i64, + track.year.map(|value| value as i64), + track.genre, + track.container, + track.codec, + track.bit_depth.map(|value| value as i64), + track.sample_rate_hz.map(|value| value as i64), + track.channels.map(|value| value as i64), + track.bitrate_kbps.map(|value| value as i64), + track.artwork_token, + track + .size_bytes + .map(|value| value.min(i64::MAX as u64) as i64), + generation, + revision, + ]) + .map_err(map_err("write essential track"))?; + } + } + let expected = (observed_rows.max(0) as u64).saturating_add(tracks.len() as u64); + let actual = tx + .query_row( + "SELECT COUNT(*) FROM remote_cache_tracks + WHERE source=?1 AND observed_generation=?2", + params![source.as_str(), generation], + |row| row.get::<_, i64>(0), + ) + .map_err(map_err("verify essential identities"))? + .max(0) as u64; + if actual != expected { + return Err("media cache: source repeated an item id across pages".to_string()); + } + tx.execute( + "UPDATE remote_cache_source_sync SET observed_rows=?3,updated_at=?4 + WHERE source=?1 AND generation=?2 AND status='running'", + params![source.as_str(), generation, actual as i64, now()], + ) + .map_err(map_err("checkpoint essential page"))?; + tx.commit().map_err(map_err("commit essential page"))?; + Ok(tracks.len()) +} + +/// Finish a source generation and, for a full sweep, atomically prune rows the +/// server did not observe. A failed/interrupted caller never reaches this gate. +pub fn complete_source_sync( + conn: &mut Connection, + source: RemoteSource, + generation: u64, + prune_old: bool, +) -> Result { + let tx = conn + .transaction() + .map_err(map_err("begin source completion"))?; + let generation = generation.min(i64::MAX as u64) as i64; + let state = tx + .query_row( + "SELECT status,observed_rows FROM remote_cache_source_sync + WHERE source=?1 AND generation=?2", + params![source.as_str(), generation], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)), + ) + .optional() + .map_err(map_err("read source completion"))?; + let Some((status, observed_rows)) = state else { + return Err("media cache: source generation is missing".to_string()); + }; + if status != "running" { + return Err("media cache: source generation cannot authorize prune".to_string()); + } + let actual = tx + .query_row( + "SELECT COUNT(*) FROM remote_cache_tracks + WHERE source=?1 AND observed_generation=?2", + params![source.as_str(), generation], + |row| row.get::<_, i64>(0), + ) + .map_err(map_err("verify source generation"))?; + if actual.max(0) != observed_rows.max(0) { + return Err("media cache: source generation identity count changed".to_string()); + } + let pruned = if prune_old { + tx.execute( + "DELETE FROM remote_cache_tracks + WHERE source=?1 AND observed_generation<>?2", + params![source.as_str(), generation], + ) + .map_err(map_err("prune source generation"))? + } else { + 0 + }; + tx.execute( + "UPDATE remote_cache_source_sync SET status='complete',updated_at=?3 + WHERE source=?1 AND generation=?2", + params![source.as_str(), generation, now()], + ) + .map_err(map_err("finish source generation"))?; + tx.commit().map_err(map_err("commit source completion"))?; + Ok(pruned) +} + +pub fn interrupt_source_sync( + conn: &Connection, + source: RemoteSource, + generation: u64, +) -> Result<()> { + conn.execute( + "UPDATE remote_cache_source_sync SET status='interrupted',updated_at=?3 + WHERE source=?1 AND generation=?2 AND status='running'", + params![ + source.as_str(), + generation.min(i64::MAX as u64) as i64, + now() + ], + ) + .map(|_| ()) + .map_err(map_err("interrupt source sync")) +} + +/// Pending quality ids in deterministic order. The retry timestamp prevents a +/// deleted item or down server from creating a hot loop. +pub fn quality_candidates( + conn: &Connection, + source: RemoteSource, + limit: usize, +) -> Result> { + let mut stmt = conn + .prepare( + "SELECT item_id FROM remote_cache_tracks + WHERE source=?1 AND quality_hydrated=0 AND quality_retry_at<=?2 + ORDER BY updated_at DESC,id LIMIT ?3", + ) + .map_err(map_err("prepare quality candidates"))?; + let rows = stmt + .query_map( + params![source.as_str(), now(), limit.min(i64::MAX as usize) as i64], + |row| row.get::<_, String>(0), + ) + .map_err(map_err("query quality candidates"))?; + rows.collect::>>() + .map_err(map_err("read quality candidates")) +} + +/// Preserve caller priority while dropping ids already hydrated or deferred. +pub fn pending_quality_ids( + conn: &Connection, + source: RemoteSource, + item_ids: &[String], +) -> Result> { + let mut stmt = conn + .prepare( + "SELECT EXISTS( + SELECT 1 FROM remote_cache_tracks + WHERE source=?1 AND item_id=?2 AND quality_hydrated=0 + AND quality_retry_at<=?3 + )", + ) + .map_err(map_err("prepare pending quality lookup"))?; + let mut pending = Vec::new(); + for item_id in item_ids { + let exists = stmt + .query_row(params![source.as_str(), item_id, now()], |row| { + row.get::<_, bool>(0) + }) + .map_err(map_err("read pending quality lookup"))?; + if exists { + pending.push(item_id.clone()); + } + } + Ok(pending) +} + +/// Apply a quality batch without rewriting metadata or changing published ids. +pub fn update_track_quality( + conn: &mut Connection, + source: RemoteSource, + updates: &[CachedTrackQuality], +) -> Result { + if updates.is_empty() { + return Ok(0); + } + let tx = conn + .transaction() + .map_err(map_err("begin quality update"))?; + let revision = next_source_revision(&tx, source)?; + let mut affected = 0usize; + { + let mut stmt = tx + .prepare( + "UPDATE remote_cache_tracks SET + container=COALESCE(NULLIF(?3,''),container),codec=?4,bit_depth=?5, + sample_rate_hz=?6,channels=?7,bitrate_kbps=?8, + quality_hydrated=1,quality_retry_at=0,updated_at=?9 + WHERE source=?1 AND item_id=?2", + ) + .map_err(map_err("prepare quality update"))?; + for update in updates { + affected += stmt + .execute(params![ + source.as_str(), + update.item_id, + update.container, + update.codec, + update.bit_depth.map(|value| value as i64), + update.sample_rate_hz.map(|value| value as i64), + update.channels.map(|value| value as i64), + update.bitrate_kbps.map(|value| value as i64), + revision, + ]) + .map_err(map_err("write quality update"))?; + } + } + tx.commit().map_err(map_err("commit quality update"))?; + Ok(affected) +} + +pub fn defer_track_quality( + conn: &mut Connection, + source: RemoteSource, + item_ids: &[String], + retry_after_secs: i64, +) -> Result { + if item_ids.is_empty() { + return Ok(0); + } + let retry_at = now().saturating_add(retry_after_secs.max(1)); + let tx = conn.transaction().map_err(map_err("begin quality defer"))?; + let mut affected = 0usize; + { + let mut stmt = tx + .prepare( + "UPDATE remote_cache_tracks SET quality_retry_at=?3 + WHERE source=?1 AND item_id=?2 AND quality_hydrated=0", + ) + .map_err(map_err("prepare quality defer"))?; + for item_id in item_ids { + affected += stmt + .execute(params![source.as_str(), item_id, retry_at]) + .map_err(map_err("write quality defer"))?; + } + } + tx.commit().map_err(map_err("commit quality defer"))?; + Ok(affected) +} + /// Replace the known libraries for one source. -pub fn save_libraries(conn: &mut Connection, source: RemoteSource, libs: &[CachedLibrary]) -> Result<()> { +pub fn save_libraries( + conn: &mut Connection, + source: RemoteSource, + libs: &[CachedLibrary], +) -> Result<()> { let ts = now(); let tx = conn.transaction().map_err(map_err("begin"))?; tx.execute( @@ -398,8 +830,14 @@ pub fn save_libraries(conn: &mut Connection, source: RemoteSource, libs: &[Cache ) .map_err(map_err("prepare library"))?; for l in libs { - stmt.execute(params![source.as_str(), l.library_id, l.name, l.server_id, ts]) - .map_err(map_err("insert library"))?; + stmt.execute(params![ + source.as_str(), + l.library_id, + l.name, + l.server_id, + ts + ]) + .map_err(map_err("insert library"))?; } } tx.commit().map_err(map_err("commit")) @@ -443,7 +881,11 @@ pub fn track_by_id(conn: &Connection, id: i64) -> Result> { } /// One track by the SERVER's own id. -pub fn track_by_item_id(conn: &Connection, source: RemoteSource, item_id: &str) -> Result> { +pub fn track_by_item_id( + conn: &Connection, + source: RemoteSource, + item_id: &str, +) -> Result> { conn.query_row( &format!("{SELECT} WHERE source = ?1 AND item_id = ?2"), params![source.as_str(), item_id], @@ -458,7 +900,11 @@ pub fn track_by_item_id(conn: &Connection, source: RemoteSource, item_id: &str) /// NULLs sort LAST rather than first: 75 of 4924 measured Jellyfin rows carry no /// track number, and letting them lead would put the untagged tracks above /// track 1 on every album that has any. -pub fn album_tracks(conn: &Connection, source: RemoteSource, album_id: &str) -> Result> { +pub fn album_tracks( + conn: &Connection, + source: RemoteSource, + album_id: &str, +) -> Result> { let mut stmt = conn .prepare(&format!( "{SELECT} WHERE source = ?1 AND album_id = ?2 \ @@ -480,7 +926,14 @@ pub fn search( needle: &str, limit: Option, ) -> Result> { - search_page(conn, source, needle, 0, limit.unwrap_or(u32::MAX) as u64, "default") + search_page( + conn, + source, + needle, + 0, + limit.unwrap_or(u32::MAX) as u64, + "default", + ) } /// One deterministic page of a remote source, in the same allowlisted sort @@ -587,6 +1040,11 @@ pub fn clear(conn: &mut Connection, source: RemoteSource) -> Result { params![source.as_str()], ) .map_err(map_err("clear libraries"))?; + tx.execute( + "DELETE FROM remote_cache_source_sync WHERE source = ?1", + params![source.as_str()], + ) + .map_err(map_err("clear source sync"))?; tx.commit().map_err(map_err("commit"))?; Ok(n) } @@ -683,15 +1141,25 @@ mod tests { #[test] fn a_saved_track_round_trips_by_both_of_its_ids() { let mut c = db(); - save_tracks(&mut c, RemoteSource::Jellyfin, &[track("srv-1", "alb", Some(1), Some(3))]).unwrap(); + save_tracks( + &mut c, + RemoteSource::Jellyfin, + &[track("srv-1", "alb", Some(1), Some(3))], + ) + .unwrap(); let by_item = track_by_item_id(&c, RemoteSource::Jellyfin, "srv-1") .unwrap() .expect("by item id"); assert_eq!(by_item.title, "t-srv-1"); assert_eq!(by_item.bit_depth, Some(24)); - assert_eq!(RemoteSource::of_id(by_item.id), Some(RemoteSource::Jellyfin)); + assert_eq!( + RemoteSource::of_id(by_item.id), + Some(RemoteSource::Jellyfin) + ); - let by_id = track_by_id(&c, by_item.id).unwrap().expect("by namespaced id"); + let by_id = track_by_id(&c, by_item.id) + .unwrap() + .expect("by namespaced id"); assert_eq!(by_id, by_item); } @@ -701,15 +1169,24 @@ mod tests { #[test] fn a_rescan_updates_a_row_without_changing_its_id() { let mut c = db(); - save_tracks(&mut c, RemoteSource::Jellyfin, &[track("srv-1", "alb", Some(1), Some(3))]).unwrap(); - let first = track_by_item_id(&c, RemoteSource::Jellyfin, "srv-1").unwrap().unwrap(); + save_tracks( + &mut c, + RemoteSource::Jellyfin, + &[track("srv-1", "alb", Some(1), Some(3))], + ) + .unwrap(); + let first = track_by_item_id(&c, RemoteSource::Jellyfin, "srv-1") + .unwrap() + .unwrap(); let mut renamed = track("srv-1", "alb", Some(1), Some(3)); renamed.title = "retagged".into(); renamed.bit_depth = Some(16); save_tracks(&mut c, RemoteSource::Jellyfin, &[renamed]).unwrap(); - let second = track_by_item_id(&c, RemoteSource::Jellyfin, "srv-1").unwrap().unwrap(); + let second = track_by_item_id(&c, RemoteSource::Jellyfin, "srv-1") + .unwrap() + .unwrap(); assert_eq!(second.id, first.id, "the row id changed under a re-scan"); assert_eq!(second.title, "retagged"); assert_eq!(second.bit_depth, Some(16)); @@ -721,12 +1198,26 @@ mod tests { #[test] fn the_same_item_id_under_two_sources_is_two_rows() { let mut c = db(); - save_tracks(&mut c, RemoteSource::Jellyfin, &[track("same", "a", None, None)]).unwrap(); - save_tracks(&mut c, RemoteSource::Subsonic, &[track("same", "a", None, None)]).unwrap(); + save_tracks( + &mut c, + RemoteSource::Jellyfin, + &[track("same", "a", None, None)], + ) + .unwrap(); + save_tracks( + &mut c, + RemoteSource::Subsonic, + &[track("same", "a", None, None)], + ) + .unwrap(); assert_eq!(count(&c, RemoteSource::Jellyfin).unwrap(), 1); assert_eq!(count(&c, RemoteSource::Subsonic).unwrap(), 1); - let j = track_by_item_id(&c, RemoteSource::Jellyfin, "same").unwrap().unwrap(); - let s = track_by_item_id(&c, RemoteSource::Subsonic, "same").unwrap().unwrap(); + let j = track_by_item_id(&c, RemoteSource::Jellyfin, "same") + .unwrap() + .unwrap(); + let s = track_by_item_id(&c, RemoteSource::Subsonic, "same") + .unwrap() + .unwrap(); assert_ne!(j.id, s.id); assert_eq!(RemoteSource::of_id(j.id), Some(RemoteSource::Jellyfin)); assert_eq!(RemoteSource::of_id(s.id), Some(RemoteSource::Subsonic)); @@ -767,15 +1258,24 @@ mod tests { for needle in ["So What", "miles", "Kind of", "so wh"] { assert_eq!( - search(&c, RemoteSource::Subsonic, needle, None).unwrap().len(), + search(&c, RemoteSource::Subsonic, needle, None) + .unwrap() + .len(), 1, "needle {needle:?} found nothing" ); } - assert_eq!(search(&c, RemoteSource::Subsonic, "", None).unwrap().len(), 1); - assert!(search(&c, RemoteSource::Subsonic, "zzz", None).unwrap().is_empty()); + assert_eq!( + search(&c, RemoteSource::Subsonic, "", None).unwrap().len(), + 1 + ); + assert!(search(&c, RemoteSource::Subsonic, "zzz", None) + .unwrap() + .is_empty()); // A search never leaks another source's rows. - assert!(search(&c, RemoteSource::Jellyfin, "", None).unwrap().is_empty()); + assert!(search(&c, RemoteSource::Jellyfin, "", None) + .unwrap() + .is_empty()); } #[test] @@ -846,8 +1346,18 @@ mod tests { #[test] fn clearing_one_source_leaves_the_other_alone() { let mut c = db(); - save_tracks(&mut c, RemoteSource::Jellyfin, &[track("j", "a", None, None)]).unwrap(); - save_tracks(&mut c, RemoteSource::Subsonic, &[track("s", "a", None, None)]).unwrap(); + save_tracks( + &mut c, + RemoteSource::Jellyfin, + &[track("j", "a", None, None)], + ) + .unwrap(); + save_tracks( + &mut c, + RemoteSource::Subsonic, + &[track("s", "a", None, None)], + ) + .unwrap(); assert_eq!(clear(&mut c, RemoteSource::Jellyfin).unwrap(), 1); assert_eq!(count(&c, RemoteSource::Jellyfin).unwrap(), 0); assert_eq!(count(&c, RemoteSource::Subsonic).unwrap(), 1); @@ -860,7 +1370,10 @@ mod tests { save_tracks( &mut c, RemoteSource::Subsonic, - &[track("keep", "a", None, None), track("gone", "a", None, None)], + &[ + track("keep", "a", None, None), + track("gone", "a", None, None), + ], ) .unwrap(); // A later sweep re-sees only `keep`. `updated_at` is whole seconds, so @@ -877,8 +1390,12 @@ mod tests { .unwrap(); assert_eq!(prune_stale(&mut c, RemoteSource::Subsonic, 150).unwrap(), 1); assert_eq!(count(&c, RemoteSource::Subsonic).unwrap(), 1); - assert!(track_by_item_id(&c, RemoteSource::Subsonic, "keep").unwrap().is_some()); - assert!(track_by_item_id(&c, RemoteSource::Subsonic, "gone").unwrap().is_none()); + assert!(track_by_item_id(&c, RemoteSource::Subsonic, "keep") + .unwrap() + .is_some()); + assert!(track_by_item_id(&c, RemoteSource::Subsonic, "gone") + .unwrap() + .is_none()); } /// AUTOINCREMENT, not plain rowid: a freed id must never be handed to a @@ -886,11 +1403,25 @@ mod tests { #[test] fn a_deleted_rows_id_is_never_reused() { let mut c = db(); - save_tracks(&mut c, RemoteSource::Jellyfin, &[track("first", "a", None, None)]).unwrap(); - let first = track_by_item_id(&c, RemoteSource::Jellyfin, "first").unwrap().unwrap(); + save_tracks( + &mut c, + RemoteSource::Jellyfin, + &[track("first", "a", None, None)], + ) + .unwrap(); + let first = track_by_item_id(&c, RemoteSource::Jellyfin, "first") + .unwrap() + .unwrap(); clear(&mut c, RemoteSource::Jellyfin).unwrap(); - save_tracks(&mut c, RemoteSource::Jellyfin, &[track("second", "a", None, None)]).unwrap(); - let second = track_by_item_id(&c, RemoteSource::Jellyfin, "second").unwrap().unwrap(); + save_tracks( + &mut c, + RemoteSource::Jellyfin, + &[track("second", "a", None, None)], + ) + .unwrap(); + let second = track_by_item_id(&c, RemoteSource::Jellyfin, "second") + .unwrap() + .unwrap(); assert_ne!( second.id, first.id, "the deleted track's id was recycled — a stale queue entry would now play this row" @@ -925,8 +1456,167 @@ mod tests { t.sample_rate_hz = Some(44100); t.container = "mp3".into(); save_tracks(&mut c, RemoteSource::Subsonic, &[t]).unwrap(); - let got = track_by_item_id(&c, RemoteSource::Subsonic, "mp3").unwrap().unwrap(); + let got = track_by_item_id(&c, RemoteSource::Subsonic, "mp3") + .unwrap() + .unwrap(); assert_eq!(got.bit_depth, None, "NULL came back as something else"); assert_eq!(got.sample_rate_hz, Some(44100)); } + + #[test] + fn essential_generation_preserves_quality_and_prunes_only_on_completion() { + let mut c = db(); + let mut keep = track("keep", "album", Some(1), Some(1)); + keep.title = "Old title".into(); + keep.container = "flac".into(); + keep.codec = Some("flac".into()); + keep.bit_depth = Some(24); + keep.sample_rate_hz = Some(96_000); + let gone = track("gone", "album", Some(1), Some(2)); + save_tracks(&mut c, RemoteSource::Jellyfin, &[keep.clone(), gone]).unwrap(); + + let first = begin_source_sync(&mut c, RemoteSource::Jellyfin).unwrap(); + let mut essential = keep.clone(); + essential.title = "Fresh title".into(); + essential.codec = None; + essential.bit_depth = None; + essential.sample_rate_hz = None; + save_essential_tracks( + &mut c, + RemoteSource::Jellyfin, + first.generation, + &[essential], + ) + .unwrap(); + assert_eq!(count(&c, RemoteSource::Jellyfin).unwrap(), 2); + let visible = track_by_item_id(&c, RemoteSource::Jellyfin, "keep") + .unwrap() + .unwrap(); + assert_eq!(visible.title, "Fresh title"); + assert_eq!(visible.bit_depth, Some(24)); + assert_eq!(visible.sample_rate_hz, Some(96_000)); + assert_eq!( + quality_candidates(&c, RemoteSource::Jellyfin, 10).unwrap(), + vec!["keep".to_string()] + ); + + interrupt_source_sync(&c, RemoteSource::Jellyfin, first.generation).unwrap(); + assert_eq!(count(&c, RemoteSource::Jellyfin).unwrap(), 2); + let second = begin_source_sync(&mut c, RemoteSource::Jellyfin).unwrap(); + save_essential_tracks(&mut c, RemoteSource::Jellyfin, second.generation, &[keep]).unwrap(); + assert_eq!( + complete_source_sync(&mut c, RemoteSource::Jellyfin, second.generation, true,).unwrap(), + 1 + ); + assert!(track_by_item_id(&c, RemoteSource::Jellyfin, "gone") + .unwrap() + .is_none()); + } + + #[test] + fn quality_hydration_is_keyed_by_id_and_clears_stale_lossless_values() { + let mut c = db(); + let mut row = track("changed", "album", Some(1), Some(1)); + row.container = "flac".into(); + row.codec = Some("flac".into()); + row.bit_depth = Some(24); + row.sample_rate_hz = Some(192_000); + save_tracks(&mut c, RemoteSource::Jellyfin, &[row.clone()]).unwrap(); + let generation = begin_source_sync(&mut c, RemoteSource::Jellyfin) + .unwrap() + .generation; + save_essential_tracks(&mut c, RemoteSource::Jellyfin, generation, &[row]).unwrap(); + + let before_revision: i64 = c + .query_row( + "SELECT updated_at FROM remote_cache_tracks WHERE item_id='changed'", + [], + |record| record.get(0), + ) + .unwrap(); + assert_eq!( + pending_quality_ids( + &c, + RemoteSource::Jellyfin, + &["missing".into(), "changed".into()], + ) + .unwrap(), + vec!["changed".to_string()] + ); + update_track_quality( + &mut c, + RemoteSource::Jellyfin, + &[CachedTrackQuality { + item_id: "changed".into(), + container: "mp3".into(), + codec: Some("mp3".into()), + bit_depth: None, + sample_rate_hz: Some(44_100), + channels: Some(2), + bitrate_kbps: Some(320), + }], + ) + .unwrap(); + let hydrated = track_by_item_id(&c, RemoteSource::Jellyfin, "changed") + .unwrap() + .unwrap(); + assert_eq!(hydrated.container, "mp3"); + assert_eq!(hydrated.bit_depth, None); + assert_eq!(hydrated.sample_rate_hz, Some(44_100)); + assert!(quality_candidates(&c, RemoteSource::Jellyfin, 10) + .unwrap() + .is_empty()); + let after_revision: i64 = c + .query_row( + "SELECT updated_at FROM remote_cache_tracks WHERE item_id='changed'", + [], + |record| record.get(0), + ) + .unwrap(); + assert!(after_revision > before_revision); + } + + #[test] + fn completed_delta_generation_keeps_unobserved_rows() { + let mut c = db(); + let keep = track("changed", "album", Some(1), Some(1)); + let untouched = track("unchanged", "album", Some(1), Some(2)); + save_tracks(&mut c, RemoteSource::Jellyfin, &[keep.clone(), untouched]).unwrap(); + let generation = begin_source_sync(&mut c, RemoteSource::Jellyfin) + .unwrap() + .generation; + save_essential_tracks(&mut c, RemoteSource::Jellyfin, generation, &[keep]).unwrap(); + assert_eq!( + complete_source_sync(&mut c, RemoteSource::Jellyfin, generation, false,).unwrap(), + 0 + ); + assert_eq!(count(&c, RemoteSource::Jellyfin).unwrap(), 2); + assert!(track_by_item_id(&c, RemoteSource::Jellyfin, "unchanged") + .unwrap() + .is_some()); + } + + #[test] + fn duplicate_item_across_essential_pages_rolls_back_the_second_page() { + let mut c = db(); + let generation = begin_source_sync(&mut c, RemoteSource::Jellyfin) + .unwrap() + .generation; + let first = track("same", "album", Some(1), Some(1)); + save_essential_tracks(&mut c, RemoteSource::Jellyfin, generation, &[first.clone()]) + .unwrap(); + let mut duplicate = first; + duplicate.title = "Wrong duplicate".into(); + assert!( + save_essential_tracks(&mut c, RemoteSource::Jellyfin, generation, &[duplicate],) + .is_err() + ); + assert_eq!( + track_by_item_id(&c, RemoteSource::Jellyfin, "same") + .unwrap() + .unwrap() + .title, + "t-same" + ); + } } diff --git a/crates/qbz-qt/src/auth_qt.rs b/crates/qbz-qt/src/auth_qt.rs index b082b6716..cf64acfaf 100644 --- a/crates/qbz-qt/src/auth_qt.rs +++ b/crates/qbz-qt/src/auth_qt.rs @@ -181,6 +181,7 @@ where } crate::offline_fwd::subscription_mark_valid(); crate::offline_fwd::engine().set_offline_session(false); + crate::media_sync_qt::resume_jellyfin_quality(); // Persist the token so the next launch restores the session silently. if let Err(e) = qbz_credentials::save_oauth_token(&token) { @@ -341,6 +342,7 @@ where } crate::offline_fwd::subscription_mark_valid(); crate::offline_fwd::engine().set_offline_session(false); + crate::media_sync_qt::resume_jellyfin_quality(); log::info!("[qbz-qt] restored saved session for user {user_id}"); Ok(Some(SessionInfo { user_id, diff --git a/crates/qbz-qt/src/local_tracks_model_qt.rs b/crates/qbz-qt/src/local_tracks_model_qt.rs index 36f0b9f71..2845bc3e6 100644 --- a/crates/qbz-qt/src/local_tracks_model_qt.rs +++ b/crates/qbz-qt/src/local_tracks_model_qt.rs @@ -584,6 +584,12 @@ fn resolve_remote_records( .filter(|record| record.track_ref.source == source_kind) .map(|record| record.track_ref.native_id.as_str()) .collect::>(); + if source_kind == SourceKind::Jellyfin && !ids.is_empty() { + crate::media_sync_qt::prioritize_jellyfin_quality( + ids.iter().map(|item_id| (*item_id).to_string()).collect(), + true, + ); + } let read = |conn: &rusqlite::Connection| { ids.iter() .filter_map(|id| { @@ -882,6 +888,15 @@ fn commit_page(loaded: LoadedPage) { ); return; }; + let jellyfin_ids = loaded + .records + .iter() + .filter(|record| record.track_ref.source == SourceKind::Jellyfin) + .map(|record| record.track_ref.native_id.clone()) + .collect::>(); + if !jellyfin_ids.is_empty() { + crate::media_sync_qt::prioritize_jellyfin_quality(jellyfin_ids, false); + } current.pending_pages.remove(&loaded.page); for (row, cursor) in &loaded.anchors { current.anchors.insert(*row, cursor.clone()); diff --git a/crates/qbz-qt/src/main.rs b/crates/qbz-qt/src/main.rs index d93f293d6..5c2504afc 100644 --- a/crates/qbz-qt/src/main.rs +++ b/crates/qbz-qt/src/main.rs @@ -3201,6 +3201,9 @@ fn main() { // A Plex page and its checkpoint are one transaction. Cancellation // prevents section prune and the next launch resumes at that page. local_plex::cancel_sync(); + // Jellyfin sync/hydration responses carry an epoch; invalidating it + // here prevents a late server response from writing after UI teardown. + media_sync_qt::cancel_all(); // The backstop for everything below AND for Qt's own teardown (engine // + QGuiApplication destructors, atexit chain): if any of it wedges, // the watchdog `_exit(0)`s the process. Idempotent — the quit paths diff --git a/crates/qbz-qt/src/media_servers_qt.rs b/crates/qbz-qt/src/media_servers_qt.rs index 3191c675c..5aff443f6 100644 --- a/crates/qbz-qt/src/media_servers_qt.rs +++ b/crates/qbz-qt/src/media_servers_qt.rs @@ -79,6 +79,9 @@ pub fn put(kind: MediaServerKind, s: &MediaServerSettings) { } pub fn disconnect(kind: MediaServerKind) { + let _jellyfin_gate = + (kind == MediaServerKind::Jellyfin).then(crate::media_sync_qt::jellyfin_state_guard); + crate::media_sync_qt::cancel(kind); STATE.disconnect(kind); invalidate_cache(); } @@ -411,6 +414,12 @@ pub fn album_tracks(prefixed_key: &str) -> Option> let rows = handle .with(|c| qbz_media_cache::album_tracks(c, source, album_id).unwrap_or_default()) .unwrap_or_default(); + if source == qbz_media_cache::RemoteSource::Jellyfin { + crate::media_sync_qt::prioritize_jellyfin_quality( + rows.iter().map(|row| row.item_id.clone()).collect(), + false, + ); + } Some(rows.into_iter().map(cached_to_local_track).collect()) } @@ -492,15 +501,19 @@ pub fn search_tracks_page( qbz_source::registry().subsonic().cache(), ), }; - handle + let rows = handle .with(|c| { qbz_media_cache::search_page(c, source, query, offset, limit, sort) .unwrap_or_default() }) - .unwrap_or_default() - .into_iter() - .map(cached_to_local_track) - .collect() + .unwrap_or_default(); + if source == qbz_media_cache::RemoteSource::Jellyfin { + crate::media_sync_qt::prioritize_jellyfin_quality( + rows.iter().map(|row| row.item_id.clone()).collect(), + false, + ); + } + rows.into_iter().map(cached_to_local_track).collect() } /// Substring search across one remote source, in the `LocalTrack` shape. diff --git a/crates/qbz-qt/src/media_sync_qt.rs b/crates/qbz-qt/src/media_sync_qt.rs index 9a53eb15a..c6631749f 100644 --- a/crates/qbz-qt/src/media_sync_qt.rs +++ b/crates/qbz-qt/src/media_sync_qt.rs @@ -19,20 +19,24 @@ //! `Fields=MediaStreams` trims 29 % of the bytes and saves nothing. Subsonic //! ships the same facts as ordinary OpenSubsonic song fields, for free. //! -//! So Jellyfin gets a progress report and a delta path; Subsonic needs neither -//! and is simply run to completion. +//! Jellyfin therefore publishes a cheap essential-metadata pass first, without +//! `MediaSources`, and hydrates quality afterwards in bounded batches. Visible, +//! queued, playing, and recent rows get priority over the catalog-wide fill. +//! Subsonic ships complete rows in its ordinary paged sweep. //! //! # Two rules the prune depends on //! -//! 1. **`prune_stale` runs only after a sweep that COMPLETED.** It deletes rows -//! the sweep did not touch, which is how a track deleted on the server -//! disappears here. A connection dropped halfway would otherwise read as -//! "the server deleted everything the sweep never got to". +//! 1. **A generation prunes only after a sweep that COMPLETED.** It deletes +//! rows the sweep did not observe, which is how a track deleted on the +//! server disappears here. A connection dropped halfway would otherwise +//! read as "the server deleted everything the sweep never got to". //! 2. **A delta sweep never prunes**, for the same reason with the sign //! flipped: it deliberately does not see unchanged rows, so every one of //! them looks stale. -use std::sync::atomic::{AtomicBool, Ordering}; +use std::collections::{HashSet, VecDeque}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{LazyLock, Mutex}; use qbz_app::settings::media_servers::{MediaServerKind, MediaServerSettings}; use qbz_media_cache::{CachedLibrary, CachedTrack, RemoteSource}; @@ -41,6 +45,79 @@ use qbz_media_cache::{CachedLibrary, CachedTrack, RemoteSource}; /// cache's write lock and double the server's load to produce the same rows. static JELLYFIN_BUSY: AtomicBool = AtomicBool::new(false); static SUBSONIC_BUSY: AtomicBool = AtomicBool::new(false); +static JELLYFIN_CANCEL: AtomicBool = AtomicBool::new(false); +static JELLYFIN_QUALITY_EPOCH: AtomicU64 = AtomicU64::new(1); +static JELLYFIN_QUALITY_RUNNING: AtomicBool = AtomicBool::new(false); +static JELLYFIN_BULK_QUALITY: AtomicBool = AtomicBool::new(false); +static JELLYFIN_QUALITY_QUEUE: LazyLock> = + LazyLock::new(|| Mutex::new(QualityQueue::default())); +static JELLYFIN_STATE_GATE: Mutex<()> = Mutex::new(()); + +const QUALITY_QUEUE_MAX: usize = 2_000; +const QUALITY_RETRY_SECS: i64 = 60; + +#[derive(Default)] +struct QualityQueue { + urgent: VecDeque, + visible: VecDeque, + present: HashSet, +} + +impl QualityQueue { + fn push(&mut self, item_ids: impl IntoIterator, urgent: bool) { + for item_id in item_ids { + if item_id.is_empty() || self.present.contains(&item_id) { + continue; + } + while self.present.len() >= QUALITY_QUEUE_MAX { + let removed = if urgent { + self.visible.pop_back().or_else(|| self.urgent.pop_back()) + } else { + // A merely visible row never displaces an urgent + // playing/queued row when the bounded queue is full. + self.visible.pop_front() + }; + if let Some(removed) = removed { + self.present.remove(&removed); + } else { + break; + } + } + if self.present.len() >= QUALITY_QUEUE_MAX { + continue; + } + if self.present.insert(item_id.clone()) { + if urgent { + self.urgent.push_back(item_id); + } else { + self.visible.push_back(item_id); + } + } + } + } + + fn pop_batch(&mut self, limit: usize) -> Vec { + let mut rows = Vec::with_capacity(limit); + while rows.len() < limit { + let Some(item_id) = self.urgent.pop_front().or_else(|| self.visible.pop_front()) else { + break; + }; + self.present.remove(&item_id); + rows.push(item_id); + } + rows + } + + fn is_empty(&self) -> bool { + self.present.is_empty() + } + + fn clear(&mut self) { + self.urgent.clear(); + self.visible.clear(); + self.present.clear(); + } +} fn busy_flag(kind: MediaServerKind) -> &'static AtomicBool { match kind { @@ -54,6 +131,62 @@ pub fn is_syncing(kind: MediaServerKind) -> bool { busy_flag(kind).load(Ordering::Relaxed) } +/// Cancel source work without waiting for an in-flight request. Every cache +/// write checks the epoch/cancel latch after the await, so a late response is +/// discarded and can never authorize prune or publish stale quality. +pub fn cancel(kind: MediaServerKind) { + match kind { + MediaServerKind::Jellyfin => { + JELLYFIN_CANCEL.store(true, Ordering::Release); + invalidate_jellyfin_quality(); + } + MediaServerKind::Subsonic => {} + } +} + +pub fn cancel_all() { + for kind in MediaServerKind::ALL { + cancel(kind); + } +} + +pub(crate) fn jellyfin_state_guard() -> std::sync::MutexGuard<'static, ()> { + JELLYFIN_STATE_GATE + .lock() + .unwrap_or_else(|error| error.into_inner()) +} + +/// Visible rows enter the secondary quality queue behind playing/queued rows +/// but ahead of the catalog-wide background fill. +pub fn prioritize_jellyfin_quality(item_ids: Vec, urgent: bool) { + if item_ids.is_empty() { + return; + } + JELLYFIN_QUALITY_QUEUE + .lock() + .unwrap_or_else(|error| error.into_inner()) + .push(item_ids, urgent); + start_jellyfin_quality_worker(); +} + +/// Resume an interrupted catalog-wide hydration after an online session bind. +/// Pending state lives in the cache, so this does not need an in-memory +/// checkpoint and exits immediately when every row is already hydrated. +pub fn resume_jellyfin_quality() { + prioritize_recent_jellyfin_quality(); + JELLYFIN_BULK_QUALITY.store(true, Ordering::Release); + start_jellyfin_quality_worker(); +} + +fn invalidate_jellyfin_quality() { + JELLYFIN_QUALITY_EPOCH.fetch_add(1, Ordering::AcqRel); + JELLYFIN_BULK_QUALITY.store(false, Ordering::Release); + JELLYFIN_QUALITY_QUEUE + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clear(); +} + /// RAII guard so an early return — or a `?` — cannot leave the flag stuck on. /// A stuck flag means the sync button never works again until a restart. struct BusyGuard(&'static AtomicBool); @@ -85,8 +218,7 @@ pub struct SyncReport { pub total: u64, } -/// Push a progress line to the UI. Cheap enough to call per page; the pages are -/// seconds apart on Jellyfin and there are 14 of them on Subsonic. +/// Push a progress line to the UI. Cheap enough to call once per bounded page. fn report(kind: MediaServerKind, done: u64, total: u64) { log::info!("[qbz-qt] {} sync: {done}/{total}", kind.as_str()); let text = format!("{done}/{total}"); @@ -119,13 +251,18 @@ fn set_syncing_ui(on: bool) { /// /// `full` forces a complete pass; otherwise a server that has been swept before /// gets a DELTA (`minDateLastSaved`), which Jellyfin honours — verified: a -/// future-dated delta returns zero rows. That turns a re-sync from 45.8 s into -/// the cost of whatever actually changed. +/// future-dated delta returns zero rows. The essential pass deliberately omits +/// expensive `MediaSources`; quality is retained while a secondary worker +/// refreshes changed rows after this function publishes the catalog. pub async fn sync_jellyfin(full: bool) -> Result { let kind = MediaServerKind::Jellyfin; let Some(_guard) = BusyGuard::acquire(kind) else { return Err("a jellyfin sync is already running".into()); }; + // A new authoritative pass supersedes an older hydration response. The + // request may still finish, but its epoch can no longer write. + invalidate_jellyfin_quality(); + JELLYFIN_CANCEL.store(false, Ordering::Release); let cfg = crate::media_servers_qt::get(kind); if !cfg.is_configured(kind) { return Err("jellyfin is not configured".into()); @@ -136,10 +273,7 @@ pub async fn sync_jellyfin(full: bool) -> Result { out } -async fn sync_jellyfin_inner( - cfg: MediaServerSettings, - full: bool, -) -> Result { +async fn sync_jellyfin_inner(cfg: MediaServerSettings, full: bool) -> Result { let kind = MediaServerKind::Jellyfin; let client = qbz_jellyfin::JellyfinClient::new(&cfg.base_url, &cfg.token, &cfg.username) @@ -151,70 +285,120 @@ async fn sync_jellyfin_inner( if libraries.is_empty() { return Err("this jellyfin server exposes no music library".into()); } - let wanted: Vec<&qbz_jellyfin::MusicLibrary> = if cfg.selected_libraries.is_empty() { - // Never chosen: take them all. Matching the Plex flow, where the first - // fetch default-selects everything rather than showing an empty grid - // until the user opens settings. - libraries.iter().collect() - } else { - libraries - .iter() - .filter(|l| cfg.selected_libraries.contains(&l.id)) - .collect() - }; + let selected: Vec<&qbz_jellyfin::MusicLibrary> = libraries + .iter() + .filter(|library| cfg.selected_libraries.contains(&library.id)) + .collect(); + let wanted: Vec<&qbz_jellyfin::MusicLibrary> = + if cfg.selected_libraries.is_empty() || selected.is_empty() { + // Never chosen: take them all. Matching the Plex flow, where the first + // fetch (and a completely stale selection) defaults to everything + // rather than authorizing an accidental empty prune. + libraries.iter().collect() + } else { + selected + }; let delta = (!full && cfg.last_sync_at > 0).then(|| iso8601(cfg.last_sync_at)); - let started = qbz_media_cache::sweep_start(); + let sync_epoch = JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire); + let generation = begin_source_sync(RemoteSource::Jellyfin)?.generation; + log::info!( + "[media-sync] source=jellyfin phase=essential generation={generation} mode={} libraries={}", + if delta.is_some() { "delta" } else { "full" }, + wanted.len(), + ); let mut saved = 0usize; - - for lib in &wanted { - let total = client - .track_count(Some(&lib.id)) - .await - .map_err(|e| e.to_string())?; - let mut offset = 0u64; - loop { - let (page, _) = client - .tracks_page(Some(&lib.id), offset, delta.as_deref()) - .await - .map_err(|e| e.to_string())?; - if page.is_empty() { - break; - } - let rows: Vec = page - .iter() - .map(|t| jellyfin_row(t, &cfg.server_id, &lib.id)) - .collect(); - saved += write_rows(RemoteSource::Jellyfin, &rows)?; - offset += page.len() as u64; - report(kind, offset.min(total), total); - if page.len() < qbz_jellyfin::PAGE_SIZE as usize { - break; + let pass = async { + for lib in &wanted { + let mut offset = 0u64; + let mut declared_total = None; + loop { + if JELLYFIN_CANCEL.load(Ordering::Acquire) { + return Err("jellyfin sync cancelled".to_string()); + } + let (page, page_total) = client + .essential_tracks_page(Some(&lib.id), offset, delta.as_deref()) + .await + .map_err(|error| error.to_string())?; + if JELLYFIN_CANCEL.load(Ordering::Acquire) { + return Err("jellyfin sync cancelled".to_string()); + } + if declared_total.is_some_and(|total| total != page_total) { + return Err("jellyfin page total changed during sync".to_string()); + } + declared_total = Some(page_total); + if page.is_empty() { + if offset != page_total { + return Err("jellyfin page ended before its declared total".to_string()); + } + break; + } + let rows: Vec = page + .iter() + .map(|track| jellyfin_row(track, &cfg.server_id, &lib.id)) + .collect(); + saved += write_essential_rows( + RemoteSource::Jellyfin, + generation, + sync_epoch, + &rows, + )?; + offset = offset.saturating_add(page.len() as u64); + if offset > page_total { + return Err("jellyfin page exceeded its declared total".to_string()); + } + report(kind, offset, page_total); + log::info!( + "[media-sync] source=jellyfin phase=essential-page generation={generation} rows={} checkpoint={offset} total={page_total} prune_authorized=false", + page.len(), + ); + if offset == page_total { + break; + } + if page.len() < qbz_jellyfin::PAGE_SIZE as usize { + return Err("jellyfin short page cannot reach its declared total".to_string()); + } } } + + write_libraries_for_epoch( + RemoteSource::Jellyfin, + sync_epoch, + &libraries + .iter() + .map(|library| CachedLibrary { + source: "jellyfin".into(), + library_id: library.id.clone(), + name: library.name.clone(), + server_id: cfg.server_id.clone(), + }) + .collect::>(), + )?; + complete_source_sync( + RemoteSource::Jellyfin, + generation, + sync_epoch, + delta.is_none(), + ) } + .await; - write_libraries( - RemoteSource::Jellyfin, - &libraries - .iter() - .map(|l| CachedLibrary { - source: "jellyfin".into(), - library_id: l.id.clone(), - name: l.name.clone(), - server_id: cfg.server_id.clone(), - }) - .collect::>(), - )?; - - // A DELTA never prunes: it did not ask about unchanged rows, so every one - // of them would look stale. - let pruned = if delta.is_none() { - prune(RemoteSource::Jellyfin, started)? - } else { - 0 + let pruned = match pass { + Ok(pruned) => pruned, + Err(error) => { + let _ = interrupt_source_sync(RemoteSource::Jellyfin, generation); + return Err(error); + } }; - finish(kind, cfg, saved, pruned, RemoteSource::Jellyfin) + let report = finish_jellyfin(cfg, saved, pruned, sync_epoch)?; + log::info!( + "[media-sync] source=jellyfin phase=essential-complete generation={generation} rows={saved} pruned={pruned} prune_authorized={}", + delta.is_none(), + ); + prioritize_recent_jellyfin_quality(); + JELLYFIN_BULK_QUALITY.store(true, Ordering::Release); + start_jellyfin_quality_worker(); + Ok(report) } fn jellyfin_row(t: &qbz_jellyfin::JellyfinTrack, server_id: &str, library_id: &str) -> CachedTrack { @@ -252,6 +436,251 @@ fn jellyfin_row(t: &qbz_jellyfin::JellyfinTrack, server_id: &str, library_id: &s } } +fn jellyfin_quality_row( + quality: qbz_jellyfin::JellyfinTrackQuality, +) -> qbz_media_cache::CachedTrackQuality { + qbz_media_cache::CachedTrackQuality { + item_id: quality.id, + container: quality.container, + codec: quality.codec, + bit_depth: quality.bit_depth, + sample_rate_hz: quality.sample_rate_hz, + channels: quality.channels, + bitrate_kbps: quality.bitrate_bps.map(|value| value / 1000), + } +} + +fn prioritize_recent_jellyfin_quality() { + let row_ids = crate::recently_qt::jellyfin_recent_track_ids(); + if row_ids.is_empty() { + return; + } + let item_ids = handle(RemoteSource::Jellyfin) + .with(|cache| { + row_ids + .iter() + .filter_map(|row_id| { + qbz_media_cache::track_by_id(cache, *row_id) + .ok() + .flatten() + .map(|track| track.item_id) + }) + .collect::>() + }) + .unwrap_or_default(); + prioritize_jellyfin_quality(item_ids, true); +} + +fn quality_queue_has_work() -> bool { + JELLYFIN_BULK_QUALITY.load(Ordering::Acquire) + || !JELLYFIN_QUALITY_QUEUE + .lock() + .unwrap_or_else(|error| error.into_inner()) + .is_empty() +} + +fn start_jellyfin_quality_worker() { + if !crate::media_servers_qt::get(MediaServerKind::Jellyfin) + .is_configured(MediaServerKind::Jellyfin) + { + return; + } + if JELLYFIN_QUALITY_RUNNING + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return; + } + let epoch = JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire); + crate::spawn(async move { + if let Err(error) = jellyfin_quality_worker(epoch).await { + log::warn!("[jellyfin-quality] phase=paused reason={error}"); + if epoch == JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) { + JELLYFIN_BULK_QUALITY.store(false, Ordering::Release); + JELLYFIN_QUALITY_QUEUE + .lock() + .unwrap_or_else(|lock_error| lock_error.into_inner()) + .clear(); + } + } + JELLYFIN_QUALITY_RUNNING.store(false, Ordering::Release); + // A new essential sync may have advanced the epoch and queued a fresh + // bulk pass while this old request was still in flight. Cancellation + // clears all work, so any work present here belongs to the current + // epoch and must get a new worker. + if quality_queue_has_work() { + start_jellyfin_quality_worker(); + } + }); +} + +async fn jellyfin_quality_worker(epoch: u64) -> Result<(), String> { + let cfg = crate::media_servers_qt::get(MediaServerKind::Jellyfin); + if !cfg.is_configured(MediaServerKind::Jellyfin) { + return Ok(()); + } + let client = qbz_jellyfin::JellyfinClient::new(&cfg.base_url, &cfg.token, &cfg.username) + .map_err(|error| error.to_string())?; + let mut hydrated = 0usize; + let mut batches = 0usize; + let mut priority_published = false; + loop { + if epoch != JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) { + return Ok(()); + } + let priority = JELLYFIN_QUALITY_QUEUE + .lock() + .unwrap_or_else(|error| error.into_inner()) + .pop_batch(qbz_jellyfin::QUALITY_BATCH_SIZE); + let was_priority = !priority.is_empty(); + let candidates = if was_priority { + pending_quality_async(priority).await? + } else if JELLYFIN_BULK_QUALITY.load(Ordering::Acquire) { + quality_candidates_async(qbz_jellyfin::QUALITY_BATCH_SIZE).await? + } else { + Vec::new() + }; + if candidates.is_empty() { + if was_priority { + continue; + } + JELLYFIN_BULK_QUALITY.store(false, Ordering::Release); + break; + } + + let started = std::time::Instant::now(); + let response = match client.track_quality(&candidates).await { + Ok(response) => response, + Err(error) => { + if epoch != JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) { + return Ok(()); + } + defer_quality_async(candidates, QUALITY_RETRY_SECS, epoch).await?; + JELLYFIN_BULK_QUALITY.store(false, Ordering::Release); + schedule_jellyfin_quality_retry(epoch); + return Err(error.to_string()); + } + }; + if epoch != JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) { + return Ok(()); + } + let returned = response + .iter() + .map(|quality| quality.id.as_str()) + .collect::>(); + let missing = candidates + .iter() + .filter(|item_id| !returned.contains(item_id.as_str())) + .cloned() + .collect::>(); + let updates = response + .into_iter() + .map(jellyfin_quality_row) + .collect::>(); + let updated = update_quality_async(updates, epoch).await?; + if !missing.is_empty() { + defer_quality_async(missing, QUALITY_RETRY_SECS, epoch).await?; + } + hydrated = hydrated.saturating_add(updated); + batches = batches.saturating_add(1); + log::info!( + "[jellyfin-quality] phase=hydrate batch={} requested={} updated={} missing={} priority={} elapsed={:?}", + batches, + candidates.len(), + updated, + candidates.len().saturating_sub(updated), + was_priority, + started.elapsed(), + ); + if was_priority && updated > 0 && !priority_published { + // Coalesced by the catalog worker. This updates a bounded visible + // page promptly while the remaining bulk fill stays silent. + crate::local_catalog_qt::request_catch_up(); + priority_published = true; + } + } + if hydrated > 0 { + crate::local_catalog_qt::request_catch_up(); + } + log::info!("[jellyfin-quality] phase=complete batches={batches} hydrated={hydrated}"); + Ok(()) +} + +fn schedule_jellyfin_quality_retry(epoch: u64) { + crate::spawn(async move { + tokio::time::sleep(std::time::Duration::from_secs(QUALITY_RETRY_SECS as u64)).await; + if epoch == JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) { + JELLYFIN_BULK_QUALITY.store(true, Ordering::Release); + start_jellyfin_quality_worker(); + } + }); +} + +async fn pending_quality_async(item_ids: Vec) -> Result, String> { + tokio::task::spawn_blocking(move || { + handle(RemoteSource::Jellyfin) + .with(|cache| { + qbz_media_cache::pending_quality_ids(cache, RemoteSource::Jellyfin, &item_ids) + }) + .ok_or_else(|| "the remote cache is not bound to a user".to_string())? + }) + .await + .map_err(|error| format!("jellyfin quality cache worker failed: {error}"))? +} + +async fn quality_candidates_async(limit: usize) -> Result, String> { + tokio::task::spawn_blocking(move || { + handle(RemoteSource::Jellyfin) + .with(|cache| qbz_media_cache::quality_candidates(cache, RemoteSource::Jellyfin, limit)) + .ok_or_else(|| "the remote cache is not bound to a user".to_string())? + }) + .await + .map_err(|error| format!("jellyfin quality candidate worker failed: {error}"))? +} + +async fn update_quality_async( + updates: Vec, + epoch: u64, +) -> Result { + tokio::task::spawn_blocking(move || { + handle(RemoteSource::Jellyfin) + .with_mut(|cache| { + if epoch != JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) { + return Ok(0); + } + qbz_media_cache::update_track_quality(cache, RemoteSource::Jellyfin, &updates) + }) + .ok_or_else(|| "the remote cache is not bound to a user".to_string())? + }) + .await + .map_err(|error| format!("jellyfin quality write worker failed: {error}"))? +} + +async fn defer_quality_async( + item_ids: Vec, + retry_secs: i64, + epoch: u64, +) -> Result<(), String> { + tokio::task::spawn_blocking(move || { + handle(RemoteSource::Jellyfin) + .with_mut(|cache| { + if epoch != JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) { + return Ok(0); + } + qbz_media_cache::defer_track_quality( + cache, + RemoteSource::Jellyfin, + &item_ids, + retry_secs, + ) + }) + .ok_or_else(|| "the remote cache is not bound to a user".to_string())? + }) + .await + .map_err(|error| format!("jellyfin quality defer worker failed: {error}"))??; + Ok(()) +} + /// Jellyfin wants `minDateLastSaved` as ISO-8601 UTC. /// /// Hand-rolled from a Unix timestamp rather than pulling in `chrono`: this is @@ -323,7 +752,10 @@ async fn sync_subsonic_inner( qbz_subsonic::SweepMode::Search3 => { let mut offset = 0u32; loop { - let page = client.search_page(offset).await.map_err(|e| e.to_string())?; + let page = client + .search_page(offset) + .await + .map_err(|e| e.to_string())?; if page.is_empty() { break; } @@ -361,7 +793,9 @@ async fn sync_subsonic_inner( let rows: Vec = t.iter().map(subsonic_row).collect(); saved += write_rows(RemoteSource::Subsonic, &rows)?; } - Err(e) => log::warn!("[qbz-qt] subsonic sync: album {id} failed ({e}) — skipped"), + Err(e) => { + log::warn!("[qbz-qt] subsonic sync: album {id} failed ({e}) — skipped") + } } if i % 25 == 0 { report(kind, i as u64, total); @@ -430,6 +864,67 @@ fn handle(source: RemoteSource) -> &'static qbz_source::CacheHandle { } } +fn begin_source_sync( + source: RemoteSource, +) -> Result { + handle(source) + .with_mut(|cache| qbz_media_cache::begin_source_sync(cache, source)) + .ok_or_else(|| "the remote cache is not bound to a user".to_string())? +} + +fn write_essential_rows( + source: RemoteSource, + generation: u64, + sync_epoch: u64, + rows: &[CachedTrack], +) -> Result { + handle(source) + .with_mut(|cache| { + if sync_epoch != JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) { + return Err("jellyfin sync cancelled".to_string()); + } + qbz_media_cache::save_essential_tracks(cache, source, generation, rows) + }) + .ok_or_else(|| "the remote cache is not bound to a user".to_string())? +} + +fn complete_source_sync( + source: RemoteSource, + generation: u64, + sync_epoch: u64, + prune_old: bool, +) -> Result { + handle(source) + .with_mut(|cache| { + if sync_epoch != JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) { + return Err("jellyfin sync cancelled".to_string()); + } + qbz_media_cache::complete_source_sync(cache, source, generation, prune_old) + }) + .ok_or_else(|| "the remote cache is not bound to a user".to_string())? +} + +fn interrupt_source_sync(source: RemoteSource, generation: u64) -> Result<(), String> { + handle(source) + .with(|cache| qbz_media_cache::interrupt_source_sync(cache, source, generation)) + .ok_or_else(|| "the remote cache is not bound to a user".to_string())? +} + +fn write_libraries_for_epoch( + source: RemoteSource, + sync_epoch: u64, + libraries: &[CachedLibrary], +) -> Result<(), String> { + handle(source) + .with_mut(|cache| { + if sync_epoch != JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) { + return Err("jellyfin sync cancelled".to_string()); + } + qbz_media_cache::save_libraries(cache, source, libraries) + }) + .ok_or_else(|| "the remote cache is not bound to a user".to_string())? +} + fn write_rows(source: RemoteSource, rows: &[CachedTrack]) -> Result { handle(source) .with_mut(|c| qbz_media_cache::save_tracks(c, source, rows)) @@ -454,6 +949,27 @@ fn count(source: RemoteSource) -> u64 { .unwrap_or(0) } +fn finish_jellyfin( + cfg: MediaServerSettings, + saved: usize, + pruned: usize, + sync_epoch: u64, +) -> Result { + let _gate = jellyfin_state_guard(); + if JELLYFIN_CANCEL.load(Ordering::Acquire) + || sync_epoch != JELLYFIN_QUALITY_EPOCH.load(Ordering::Acquire) + { + return Err("jellyfin sync cancelled".to_string()); + } + finish( + MediaServerKind::Jellyfin, + cfg, + saved, + pruned, + RemoteSource::Jellyfin, + ) +} + /// Stamp the sweep and report. /// /// `last_sync_at` is written ONLY here, at the end of a run that finished — @@ -583,4 +1099,50 @@ mod tests { }; assert_eq!(jellyfin_row(&jf, "s", "l").bitrate_kbps, Some(3120)); } + + #[test] + fn quality_queue_deduplicates_and_puts_urgent_rows_first() { + let mut queue = QualityQueue::default(); + queue.push(vec!["visible-a".into(), "same".into()], false); + queue.push(vec!["urgent".into(), "same".into()], true); + assert_eq!( + queue.pop_batch(3), + vec![ + "urgent".to_string(), + "visible-a".to_string(), + "same".to_string() + ] + ); + assert!(queue.is_empty()); + } + + #[test] + fn visible_quality_never_evicts_urgent_rows_at_the_memory_cap() { + let mut queue = QualityQueue::default(); + queue.push( + (0..QUALITY_QUEUE_MAX).map(|index| format!("urgent-{index}")), + true, + ); + queue.push(vec!["visible".into()], false); + assert_eq!(queue.present.len(), QUALITY_QUEUE_MAX); + assert!(!queue.present.contains("visible")); + assert_eq!(queue.pop_batch(1), vec!["urgent-0".to_string()]); + } + + #[test] + fn jellyfin_quality_mapping_keeps_lossy_nulls_and_normalises_bitrate() { + let mapped = jellyfin_quality_row(qbz_jellyfin::JellyfinTrackQuality { + id: "item".into(), + container: "mp3".into(), + codec: Some("mp3".into()), + bit_depth: None, + sample_rate_hz: Some(44_100), + channels: Some(2), + bitrate_bps: Some(320_999), + }); + assert_eq!(mapped.item_id, "item"); + assert_eq!(mapped.bit_depth, None); + assert_eq!(mapped.sample_rate_hz, Some(44_100)); + assert_eq!(mapped.bitrate_kbps, Some(320)); + } } diff --git a/crates/qbz-qt/src/recently_qt.rs b/crates/qbz-qt/src/recently_qt.rs index 0ba87bdf0..b0287d3a3 100644 --- a/crates/qbz-qt/src/recently_qt.rs +++ b/crates/qbz-qt/src/recently_qt.rs @@ -185,6 +185,18 @@ pub(crate) fn load_albums() -> Vec { read_store().albums } +/// Namespaced cache ids for recently played Jellyfin tracks, newest first. +/// The hydration worker resolves them back to opaque server ids through the +/// cache, so history never needs to persist a credential-bearing URL. +pub(crate) fn jellyfin_recent_track_ids() -> Vec { + read_store() + .tracks + .into_iter() + .filter(|track| track.source == "jellyfin") + .filter_map(|track| track.id.parse::().ok()) + .collect() +} + /// Drop every entry whose `album_id` is in `album_ids`; returns how many /// TRACK entries went. /// @@ -327,6 +339,11 @@ pub(crate) fn record_queue_track(track: &qbz_models::QueueTrack) { ); return; } + if track.source.as_deref() == Some("jellyfin") { + if let Some(item_id) = track.source_item_id_hint.as_ref() { + crate::media_sync_qt::prioritize_jellyfin_quality(vec![item_id.clone()], true); + } + } let tier = crate::home_qt::quality_tier_from_depth(track.bit_depth); let detail = crate::home_qt::quality_detail_from_parts(track.bit_depth, track.sample_rate); let quality_label = if tier.is_empty() {