From 2e3411286833e7016633fc5faf04be58d2b58073 Mon Sep 17 00:00:00 2001 From: vicrodh <103399283+vicrodh@users.noreply.github.com> Date: Wed, 2 Sep 2026 07:41:09 -0600 Subject: [PATCH] feat(app): subscription grace clock and offline purge in the shared activation The 30-day offline grace (D4) was produced by the Qt shell alone, and only when a login was REJECTED; its purge predicate had no caller in the live tree, so past the window the CMAF cache was blocked from playing but never removed (Tauri purged it; Slint recorded the gap). With member accounts now signing in, the clock has to key on the entitlement, not on a rejection, and it has to run wherever a session activates. `qbz_app::subscription_lifecycle` records the verdict from the core's session (`offline_streaming` clears the clock; a member keeps it running from its first observation) and purges the offline cache once the window has passed, through the cache a host registers with `AppRuntime::set_offline_cache`. `activate_at` runs it for every host (Qt via `activate`, qbzd directly), a daily ticker per session re-checks it, and a cache registered after activation is enforced against on registration. qbzd registers no cache and keeps the verdict. The Qt producers are gone; `offline_playback_allowed` stays the read gate. cc --- crates/qbz-app/Cargo.toml | 1 + crates/qbz-app/src/lib.rs | 1 + crates/qbz-app/src/settings/subscription.rs | 2 +- crates/qbz-app/src/shell.rs | 127 ++++- crates/qbz-app/src/subscription_lifecycle.rs | 212 +++++++++ crates/qbz-qobuz/src/client.rs | 475 +++++++++++++------ crates/qbz-qt/src/auth_qt.rs | 26 +- crates/qbz-qt/src/offline_fwd.rs | 51 -- 8 files changed, 681 insertions(+), 214 deletions(-) create mode 100644 crates/qbz-app/src/subscription_lifecycle.rs diff --git a/crates/qbz-app/Cargo.toml b/crates/qbz-app/Cargo.toml index 65f984874..55746a70f 100644 --- a/crates/qbz-app/Cargo.toml +++ b/crates/qbz-app/Cargo.toml @@ -12,6 +12,7 @@ qbz-player = { path = "../qbz-player" } qbz-core = { path = "../qbz-core" } qbz-qobuz = { path = "../qbz-qobuz" } qbz-credentials = { path = "../qbz-credentials" } +qbz-offline-cache = { path = "../qbz-offline-cache" } reqwest = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/qbz-app/src/lib.rs b/crates/qbz-app/src/lib.rs index d8d983c55..8825d27c9 100644 --- a/crates/qbz-app/src/lib.rs +++ b/crates/qbz-app/src/lib.rs @@ -11,6 +11,7 @@ pub mod runtime; pub mod scrobble_timing; pub mod session_persist; pub mod session_store; +pub mod subscription_lifecycle; pub mod shell; pub mod settings; pub mod user_data; diff --git a/crates/qbz-app/src/settings/subscription.rs b/crates/qbz-app/src/settings/subscription.rs index dd1fd4804..505bc8bf7 100644 --- a/crates/qbz-app/src/settings/subscription.rs +++ b/crates/qbz-app/src/settings/subscription.rs @@ -18,7 +18,7 @@ use std::sync::{Arc, Mutex}; /// /// The primary protection for offline files is the CMAF-at-rest cache format. /// This grace period is an additional compliance guard, not the main defense. -const GRACE_PERIOD_SECS: i64 = 30 * 24 * 60 * 60; +pub const GRACE_PERIOD_SECS: i64 = 30 * 24 * 60 * 60; #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] diff --git a/crates/qbz-app/src/shell.rs b/crates/qbz-app/src/shell.rs index 43184f0f1..6ed3fcc07 100644 --- a/crates/qbz-app/src/shell.rs +++ b/crates/qbz-app/src/shell.rs @@ -10,17 +10,26 @@ //! marker, runtime state). The shell binds its additional per-user stores as //! each session comes online. -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; +use std::time::Duration; use qbz_audio::{settings::AudioSettingsStore, AudioDiagnostic, AudioSettings, VisualizerTap}; use qbz_core::{FrontendAdapter, QbzCore}; +use qbz_offline_cache::OfflineCacheState; use qbz_player::Player; +use tokio::task::JoinHandle; use crate::runtime::RuntimeManager; use crate::session_store::SessionStore; +use crate::subscription_lifecycle; use crate::user_data::UserDataPaths; +/// How often an activated session re-checks the subscription grace clock, +/// so a long-running session without a re-login still crosses the 30-day +/// line on time. +const SUBSCRIPTION_TICK: Duration = Duration::from_secs(24 * 60 * 60); + /// The per-user stores opened for the currently active session. /// /// Minimal by design: it holds only the session store. A shell opens further @@ -28,9 +37,13 @@ use crate::user_data::UserDataPaths; /// loading the full WebKit-era store set up front. struct ActiveSession { user_id: u64, + data_dir: PathBuf, session_store: SessionStore, } +/// The offline cache a host registered, shared with the subscription ticker. +type OfflineCacheSlot = Arc>>>; + /// Composition root for a non-Tauri UI shell. /// /// Generic over the [`FrontendAdapter`] the shell supplies, so the same @@ -44,6 +57,12 @@ pub struct AppRuntime { /// the FFT producer and toggle capture. `None` for shells that do not drive /// audio visualization (the default `new`/`with_audio_settings` path). visualizer_tap: Option, + /// The host's offline cache, if it has one (Qt does, qbzd does not). + /// The subscription lifecycle purges through it once the grace window + /// has passed; without one the verdict is still recorded. + offline_cache: OfflineCacheSlot, + /// The daily subscription re-check for the active session. + subscription_ticker: Mutex>>, } impl AppRuntime { @@ -66,6 +85,8 @@ impl AppRuntime { user_paths: UserDataPaths::new(), session: Mutex::new(None), visualizer_tap: None, + offline_cache: Arc::new(Mutex::new(None)), + subscription_ticker: Mutex::new(None), } } @@ -164,19 +185,106 @@ impl AppRuntime { self.runtime.set_session_activated(true, user_id).await; - let mut guard = self - .session - .lock() - .map_err(|e| format!("session lock poisoned: {}", e))?; - *guard = Some(ActiveSession { - user_id, - session_store, - }); + { + let mut guard = self + .session + .lock() + .map_err(|e| format!("session lock poisoned: {}", e))?; + *guard = Some(ActiveSession { + user_id, + data_dir: data_dir.to_path_buf(), + session_store, + }); + } log::info!("[AppRuntime] Session activated for user"); + + // Subscription lifecycle (D4): record the login verdict from the + // core's session — a member without the offline entitlement starts + // the grace clock, a subscriber clears it — and purge the offline + // cache when the window has passed. Shared by every host through + // this one activation path; best-effort, never blocks entry. + self.run_subscription_lifecycle().await; + self.start_subscription_ticker(data_dir.to_path_buf()); Ok(()) } + /// Register (or clear) the host's offline cache. When a session is + /// already active and a cache arrives, the grace clock is enforced + /// against it right away — Qt activates its cache after the session. + pub fn set_offline_cache(self: &Arc, cache: Option>) { + let registered = cache.is_some(); + if let Ok(mut slot) = self.offline_cache.lock() { + *slot = cache; + } + if registered && self.is_session_active() { + let runtime = Arc::clone(self); + tokio::spawn(async move { runtime.run_subscription_lifecycle().await }); + } + } + + async fn run_subscription_lifecycle(&self) { + let Some(data_dir) = self.active_data_dir() else { + return; + }; + let session = { + let client_lock = self.core.client(); + let guard = client_lock.read().await; + match guard.as_ref() { + Some(client) => client.session().await, + None => None, + } + }; + let offline = self.offline_cache.lock().ok().and_then(|slot| slot.clone()); + let now = subscription_lifecycle::now_unix_secs(); + match subscription_lifecycle::run(&data_dir, session.as_ref(), offline, now).await { + Ok(v) => log::info!( + "[AppRuntime] subscription verdict: offline_entitled={} member_only={} invalid_since={:?} purged={}", + v.offline_entitled, + v.member_only, + v.invalid_since, + v.purged + ), + Err(e) => log::error!("[AppRuntime] subscription lifecycle failed: {e}"), + } + } + + fn start_subscription_ticker(&self, data_dir: PathBuf) { + let offline_cache = Arc::clone(&self.offline_cache); + let handle = tokio::spawn(async move { + let mut interval = tokio::time::interval(SUBSCRIPTION_TICK); + interval.tick().await; // the activation pass already ran + loop { + interval.tick().await; + let offline = offline_cache.lock().ok().and_then(|slot| slot.clone()); + let now = subscription_lifecycle::now_unix_secs(); + if let Err(e) = subscription_lifecycle::run(&data_dir, None, offline, now).await { + log::error!("[AppRuntime] subscription tick failed: {e}"); + } + } + }); + if let Ok(mut ticker) = self.subscription_ticker.lock() { + if let Some(previous) = ticker.replace(handle) { + previous.abort(); + } + } + } + + fn stop_subscription_ticker(&self) { + if let Ok(mut ticker) = self.subscription_ticker.lock() { + if let Some(handle) = ticker.take() { + handle.abort(); + } + } + } + + fn active_data_dir(&self) -> Option { + self.session + .lock() + .ok() + .and_then(|guard| guard.as_ref().map(|s| s.data_dir.clone())) + } + /// Activate the per-user session for `user_id`. /// /// Resolves the real per-user directories through [`UserDataPaths`], @@ -250,6 +358,7 @@ impl AppRuntime { /// `last_user_id` marker is intentionally kept on disk so a later /// offline session can still find the user's data. pub async fn deactivate(&self) -> Result<(), String> { + self.stop_subscription_ticker(); { let mut guard = self .session diff --git a/crates/qbz-app/src/subscription_lifecycle.rs b/crates/qbz-app/src/subscription_lifecycle.rs new file mode 100644 index 000000000..67a3d6ff8 --- /dev/null +++ b/crates/qbz-app/src/subscription_lifecycle.rs @@ -0,0 +1,212 @@ +//! Subscription lifecycle: the D4 grace clock and the offline-cache purge. +//! +//! Runs from the SHARED session activation (`AppRuntime::activate_at`) and +//! from a daily ticker the runtime owns, so every host — the Qt shell, the +//! daemon — gets the same verdict and the same enforcement without code of +//! its own. Before this module the verdict was produced by the Qt shell +//! alone, only when the login was REJECTED, and the purge had no caller at +//! all: past the 30-day grace the offline cache was merely blocked from +//! playing (`offline_playback_allowed`), never removed. The official Qobuz +//! app removes it; so does this. +//! +//! Semantics (member mode, 2026-09-02): +//! +//! - A login whose entitlements carry `offline_streaming` is a VALID +//! observation: the clock is cleared. +//! - A login without it — a Qobuz member with no subscription — is an +//! INVALID observation: the clock starts, or keeps running from its first +//! observation. The member still gets in; only offline is on the clock. +//! - Once the clock passes [`GRACE_PERIOD_SECS`], the cached audio is +//! purged — when a cache is registered. A host without one (qbzd) keeps +//! the verdict and the read gate, and a cache registered later is purged +//! on registration. +//! - Nothing here decides playback. `offline_playback_allowed` stays the +//! read gate the views consult. + +use std::path::Path; +use std::sync::Arc; + +use qbz_models::UserSession; +use qbz_offline_cache::OfflineCacheState; + +use crate::settings::subscription::SubscriptionStateStore; +pub use crate::settings::subscription::GRACE_PERIOD_SECS; + +/// What the lifecycle concluded for the active session. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SubscriptionVerdict { + /// The session carries the `offline_streaming` entitlement. + pub offline_entitled: bool, + /// No streaming entitlement at all: a Qobuz member without subscription. + pub member_only: bool, + /// First time the account was seen without the offline entitlement, + /// unix seconds; `None` while the clock is clear. + pub invalid_since: Option, + /// When the grace window ends (`invalid_since + GRACE_PERIOD_SECS`). + pub grace_deadline: Option, + /// The cache was purged by THIS run. + pub purged: bool, +} + +/// D4 producer: record the login verdict for this session. +pub fn observe( + store: &SubscriptionStateStore, + session: &UserSession, + now: i64, +) -> Result<(), String> { + if session.entitlements.offline_streaming { + store.mark_valid(now) + } else { + store.mark_invalid(now) + } +} + +/// D4 enforcer: purge the offline cache once the grace window has passed. +/// +/// Returns `Ok(true)` when a purge happened. With no cache registered the +/// decision is logged and left standing (not marked purged), so the next +/// registration or tick purges. Takes the directory rather than an open +/// store because the store's SQLite connection cannot be held across the +/// purge's await point on a multi-threaded runtime. +pub async fn enforce( + data_dir: &Path, + now: i64, + offline: Option<&OfflineCacheState>, +) -> Result { + let due = SubscriptionStateStore::new_at(data_dir)?.should_purge_offline_cache(now)?; + if !due { + return Ok(false); + } + let Some(cache) = offline else { + log::warn!( + "[subscription] offline cache purge is due (no offline entitlement for {} days) but no cache is registered in this host", + GRACE_PERIOD_SECS / 86_400 + ); + return Ok(false); + }; + log::warn!( + "[subscription] no offline entitlement for {} days: purging the offline cache", + GRACE_PERIOD_SECS / 86_400 + ); + qbz_offline_cache::purge_all_cached_files(cache, &cache.library_db).await?; + SubscriptionStateStore::new_at(data_dir)?.mark_offline_cache_purged(now)?; + Ok(true) +} + +/// The verdict as the host may want to show it (diagnostics, `/api/status`). +pub fn verdict( + store: &SubscriptionStateStore, + session: Option<&UserSession>, + purged: bool, +) -> Result { + let state = store.get_state()?; + let entitlements = session.map(|s| &s.entitlements); + Ok(SubscriptionVerdict { + offline_entitled: entitlements.map(|e| e.offline_streaming).unwrap_or(false), + member_only: entitlements.map(|e| e.is_member_only()).unwrap_or(false), + invalid_since: state.invalid_since, + grace_deadline: state.invalid_since.map(|t| t + GRACE_PERIOD_SECS), + purged, + }) +} + +/// One full pass for the per-user directory `data_dir`: observe the session +/// (when there is one), enforce, report. Best-effort by contract — the +/// caller logs the error and activation proceeds; a broken clock must never +/// keep a user out. +pub async fn run( + data_dir: &Path, + session: Option<&UserSession>, + offline: Option>, + now: i64, +) -> Result { + if let Some(session) = session { + observe(&SubscriptionStateStore::new_at(data_dir)?, session, now)?; + } + let purged = enforce(data_dir, now, offline.as_deref()).await?; + verdict(&SubscriptionStateStore::new_at(data_dir)?, session, purged) +} + +pub fn now_unix_secs() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + use qbz_models::Entitlements; + + fn session(offline: bool) -> UserSession { + UserSession { + user_id: 7, + entitlements: Entitlements { + lossless_streaming: offline, + offline_streaming: offline, + ..Entitlements::default() + }, + ..UserSession::default() + } + } + + #[test] + fn subscriber_login_clears_the_clock() { + let dir = tempfile::tempdir().unwrap(); + let store = SubscriptionStateStore::new_at(dir.path()).unwrap(); + store.mark_invalid(100).unwrap(); + observe(&store, &session(true), 200).unwrap(); + let v = verdict(&store, Some(&session(true)), false).unwrap(); + assert_eq!(v.invalid_since, None); + assert_eq!(v.grace_deadline, None); + assert!(v.offline_entitled); + assert!(!v.member_only); + } + + #[test] + fn member_login_starts_the_clock_once_and_keeps_its_origin() { + let dir = tempfile::tempdir().unwrap(); + let store = SubscriptionStateStore::new_at(dir.path()).unwrap(); + observe(&store, &session(false), 100).unwrap(); + observe(&store, &session(false), 500).unwrap(); + let v = verdict(&store, Some(&session(false)), false).unwrap(); + assert_eq!(v.invalid_since, Some(100)); + assert_eq!(v.grace_deadline, Some(100 + GRACE_PERIOD_SECS)); + assert!(v.member_only); + assert!(!v.offline_entitled); + } + + #[tokio::test] + async fn purge_waits_for_the_grace_window_and_needs_a_cache() { + let dir = tempfile::tempdir().unwrap(); + let store = SubscriptionStateStore::new_at(dir.path()).unwrap(); + observe(&store, &session(false), 100).unwrap(); + // Inside the window: nothing. + assert!(!enforce(dir.path(), 100 + GRACE_PERIOD_SECS - 1, None) + .await + .unwrap()); + // Past the window but no cache registered: still nothing, and the + // decision is NOT marked as purged, so a later host can act. + assert!(!enforce(dir.path(), 100 + GRACE_PERIOD_SECS, None) + .await + .unwrap()); + assert!(store + .should_purge_offline_cache(100 + GRACE_PERIOD_SECS) + .unwrap()); + } + + #[tokio::test] + async fn run_without_a_session_only_enforces() { + let dir = tempfile::tempdir().unwrap(); + let store = SubscriptionStateStore::new_at(dir.path()).unwrap(); + store.mark_invalid(100).unwrap(); + drop(store); + // An offline entry (no Qobuz session) must not clear or restart the + // clock; it only reports it. + let v = run(dir.path(), None, None, 200).await.unwrap(); + assert_eq!(v.invalid_since, Some(100)); + assert!(!v.purged); + assert!(!v.offline_entitled); + } +} diff --git a/crates/qbz-qobuz/src/client.rs b/crates/qbz-qobuz/src/client.rs index bc157a8a7..b28cf3f1a 100644 --- a/crates/qbz-qobuz/src/client.rs +++ b/crates/qbz-qobuz/src/client.rs @@ -485,6 +485,12 @@ impl QobuzClient { .and_then(|s| s.language_code.clone()) } + /// A copy of the active session (entitlements included), `None` before + /// login. The shared session lifecycle reads the verdict from here. + pub async fn session(&self) -> Option { + self.session.read().await.clone() + } + /// Exchange an OAuth code for a full user session. /// /// This implements the new Qobuz OAuth flow: @@ -792,7 +798,15 @@ impl QobuzClient { let url = endpoints::build_url(paths::ALBUM_SEARCH); let timestamp = get_timestamp(); let secret = self.secret().await?; - let signature = sign_search("albumsearch", query, limit, offset, search_type, timestamp, &secret); + let signature = sign_search( + "albumsearch", + query, + limit, + offset, + search_type, + timestamp, + &secret, + ); let limit_str = limit.to_string(); let offset_str = offset.to_string(); let ts_str = timestamp.to_string(); @@ -838,7 +852,15 @@ impl QobuzClient { let url = endpoints::build_url(paths::TRACK_SEARCH); let timestamp = get_timestamp(); let secret = self.secret().await?; - let signature = sign_search("tracksearch", query, limit, offset, search_type, timestamp, &secret); + let signature = sign_search( + "tracksearch", + query, + limit, + offset, + search_type, + timestamp, + &secret, + ); let limit_str = limit.to_string(); let offset_str = offset.to_string(); let ts_str = timestamp.to_string(); @@ -884,7 +906,15 @@ impl QobuzClient { let url = endpoints::build_url(paths::ARTIST_SEARCH); let timestamp = get_timestamp(); let secret = self.secret().await?; - let signature = sign_search("artistsearch", query, limit, offset, search_type, timestamp, &secret); + let signature = sign_search( + "artistsearch", + query, + limit, + offset, + search_type, + timestamp, + &secret, + ); let limit_str = limit.to_string(); let offset_str = offset.to_string(); let ts_str = timestamp.to_string(); @@ -924,7 +954,15 @@ impl QobuzClient { let url = endpoints::build_url(paths::CATALOG_SEARCH); let timestamp = get_timestamp(); let secret = self.secret().await?; - let signature = sign_search("catalogsearch", query, limit, offset, None, timestamp, &secret); + let signature = sign_search( + "catalogsearch", + query, + limit, + offset, + None, + timestamp, + &secret, + ); let limit_str = limit.to_string(); let offset_str = offset.to_string(); let ts_str = timestamp.to_string(); @@ -956,11 +994,15 @@ impl QobuzClient { ) -> Result> { let url = endpoints::build_url(paths::ARTIST_GET_SIMILAR); let http_response = self - .signed_get(&url, "artistgetSimilarArtists", &[ - ("artist_id", artist_id.to_string()), - ("limit", limit.to_string()), - ("offset", offset.to_string()), - ]) + .signed_get( + &url, + "artistgetSimilarArtists", + &[ + ("artist_id", artist_id.to_string()), + ("limit", limit.to_string()), + ("offset", offset.to_string()), + ], + ) .await?; log::debug!( "[API] get_similar_artists({}) status={}", @@ -987,13 +1029,17 @@ impl QobuzClient { let locale = self.locale().await; let http_response = self - .signed_get(&url, "artistget", &[ - ("artist_id", artist_id.to_string()), - ("extra", "tracks".to_string()), - ("lang", locale), - ("limit", limit.to_string()), - ("offset", offset.to_string()), - ]) + .signed_get( + &url, + "artistget", + &[ + ("artist_id", artist_id.to_string()), + ("extra", "tracks".to_string()), + ("lang", locale), + ("limit", limit.to_string()), + ("offset", offset.to_string()), + ], + ) .await?; log::debug!( "[API] get_artist_tracks({}) status={}", @@ -1061,10 +1107,9 @@ impl QobuzClient { query.push(("genre_id".to_string(), gid.to_string())); } - let params: Vec<(&str, String)> = query.iter().map(|(k, v)| (k.as_str(), v.clone())).collect(); - let http_response = self - .signed_get(&url, "albumgetFeatured", ¶ms) - .await?; + let params: Vec<(&str, String)> = + query.iter().map(|(k, v)| (k.as_str(), v.clone())).collect(); + let http_response = self.signed_get(&url, "albumgetFeatured", ¶ms).await?; log::debug!( "[API] get_featured_albums({}) status={}", featured_type, @@ -1148,10 +1193,7 @@ impl QobuzClient { if let Some(items_arr) = items_value.as_array_mut() { for item in items_arr { if let Some(obj) = item.as_object_mut() { - let needs_backfill = obj - .get("artist") - .map(|v| v.is_null()) - .unwrap_or(true); + let needs_backfill = obj.get("artist").map(|v| v.is_null()).unwrap_or(true); if needs_backfill { if let Some(first_artist) = obj .get("artists") @@ -1269,17 +1311,20 @@ impl QobuzClient { /// Qobuz radio for an artist (`/radio/artist`) — a generated track /// list. Ported from the legacy api client. pub async fn get_radio_artist(&self, artist_id: &str) -> Result { - self.get_radio(paths::RADIO_ARTIST, "artist_id", artist_id).await + self.get_radio(paths::RADIO_ARTIST, "artist_id", artist_id) + .await } /// Qobuz radio for a track (`/radio/track`). pub async fn get_radio_track(&self, track_id: &str) -> Result { - self.get_radio(paths::RADIO_TRACK, "track_id", track_id).await + self.get_radio(paths::RADIO_TRACK, "track_id", track_id) + .await } /// Qobuz radio for an album (`/radio/album`). pub async fn get_radio_album(&self, album_id: &str) -> Result { - self.get_radio(paths::RADIO_ALBUM, "album_id", album_id).await + self.get_radio(paths::RADIO_ALBUM, "album_id", album_id) + .await } async fn get_radio(&self, path: &str, key: &str, id: &str) -> Result { @@ -1312,7 +1357,14 @@ impl QobuzClient { } let http_response = self - .signed_get(&url, "genrelist", &query.iter().map(|(k, v)| (*k, v.clone())).collect::>()) + .signed_get( + &url, + "genrelist", + &query + .iter() + .map(|(k, v)| (*k, v.clone())) + .collect::>(), + ) .await?; log::debug!( "[API] get_genres(parent={:?}) status={}", @@ -1350,7 +1402,14 @@ impl QobuzClient { } let http_response = self - .signed_get_auth(&url, "discoverindex", &query.iter().map(|(k, v)| (*k, v.clone())).collect::>()) + .signed_get_auth( + &url, + "discoverindex", + &query + .iter() + .map(|(k, v)| (*k, v.clone())) + .collect::>(), + ) .await?; log::info!( "[API] get_discover_index genre_ids={:?} status={}", @@ -1404,7 +1463,14 @@ impl QobuzClient { // Derive method name from endpoint path: "/discover/newReleases" -> "discovernewReleases" let method_name = endpoint.replace('/', "").replace('.', ""); let http_response = self - .signed_get_auth(&url, &method_name, &query.iter().map(|(k, v)| (*k, v.clone())).collect::>()) + .signed_get_auth( + &url, + &method_name, + &query + .iter() + .map(|(k, v)| (*k, v.clone())) + .collect::>(), + ) .await?; log::info!( "[API] get_discover_albums({}) query={:?} status={}", @@ -1462,7 +1528,14 @@ impl QobuzClient { // First get raw JSON to debug structure let raw_response: serde_json::Value = self - .signed_get_auth(&url, "discoverplaylists", &query.iter().map(|(k, v)| (*k, v.clone())).collect::>()) + .signed_get_auth( + &url, + "discoverplaylists", + &query + .iter() + .map(|(k, v)| (*k, v.clone())) + .collect::>(), + ) .await? .json() .await?; @@ -1497,9 +1570,7 @@ impl QobuzClient { pub async fn get_playlist_tags(&self) -> Result> { let url = endpoints::build_url(paths::PLAYLIST_GET_TAGS); - let http_response = self - .signed_get_auth(&url, "playlistgetTags", &[]) - .await?; + let http_response = self.signed_get_auth(&url, "playlistgetTags", &[]).await?; log::info!("[API] get_playlist_tags status={}", http_response.status()); let raw: PlaylistTagsResponse = http_response.json().await?; @@ -1737,9 +1808,7 @@ impl QobuzClient { match serde_json::from_str::(&body) { Ok(mut doc) if doc.original.is_some() || doc.translation.is_some() => { if language.is_some() && doc.original.is_some() && doc.translation.is_none() { - doc.translation = self - .fetch_translation_fallback(track_id, &urls) - .await; + doc.translation = self.fetch_translation_fallback(track_id, &urls).await; } Ok(Some(doc)) } @@ -1811,7 +1880,14 @@ impl QobuzClient { ]; let http_response = self - .signed_get(&url, "artistget", &query.iter().map(|(k, v)| (*k, v.clone())).collect::>()) + .signed_get( + &url, + "artistget", + &query + .iter() + .map(|(k, v)| (*k, v.clone())) + .collect::>(), + ) .await?; log::debug!( "[API] get_artist_basic({}) status={}", @@ -1851,7 +1927,14 @@ impl QobuzClient { } let http_response = self - .signed_get(&url, "artistget", &query.iter().map(|(k, v)| (*k, v.clone())).collect::>()) + .signed_get( + &url, + "artistget", + &query + .iter() + .map(|(k, v)| (*k, v.clone())) + .collect::>(), + ) .await?; log::debug!( "[API] get_artist_detail({}) status={}", @@ -1902,7 +1985,14 @@ impl QobuzClient { } let http_response = self - .signed_get(&url, "artistget", &query.iter().map(|(k, v)| (*k, v.clone())).collect::>()) + .signed_get( + &url, + "artistget", + &query + .iter() + .map(|(k, v)| (*k, v.clone())) + .collect::>(), + ) .await?; log::debug!( "[API] get_artist({}, albums={}) status={}", @@ -1921,10 +2011,14 @@ impl QobuzClient { pub async fn get_playlist_track_ids(&self, playlist_id: u64) -> Result { let url = endpoints::build_url(paths::PLAYLIST_GET); let http_response = self - .signed_get(&url, "playlistget", &[ - ("playlist_id", playlist_id.to_string()), - ("extra", "track_ids".to_string()), - ]) + .signed_get( + &url, + "playlistget", + &[ + ("playlist_id", playlist_id.to_string()), + ("extra", "track_ids".to_string()), + ], + ) .await?; log::debug!( "[API] get_playlist_track_ids({}) status={}", @@ -1976,8 +2070,17 @@ impl QobuzClient { let headers = self.api_headers().await?; let timestamp = get_timestamp(); let secret = self.secret().await?; - let ids_str: String = track_ids.iter().map(|id| id.to_string()).collect::>().join(","); - let sig = sign_request("trackgetList", &[("tracks_id", &ids_str)], timestamp, &secret); + let ids_str: String = track_ids + .iter() + .map(|id| id.to_string()) + .collect::>() + .join(","); + let sig = sign_request( + "trackgetList", + &[("tracks_id", &ids_str)], + timestamp, + &secret, + ); let body = serde_json::json!({ "tracks_id": track_ids }); log::debug!("[API] get_tracks_batch POST ({} IDs)", track_ids.len()); @@ -2029,12 +2132,16 @@ impl QobuzClient { // First page — gives us metadata + total track count let http_response = self - .signed_get(&url, "playlistget", &[ - ("playlist_id", playlist_id.to_string()), - ("limit", PAGE_SIZE.to_string()), - ("offset", "0".to_string()), - ("extra", "tracks".to_string()), - ]) + .signed_get( + &url, + "playlistget", + &[ + ("playlist_id", playlist_id.to_string()), + ("limit", PAGE_SIZE.to_string()), + ("offset", "0".to_string()), + ("extra", "tracks".to_string()), + ], + ) .await?; log::debug!( "[API] get_playlist({}) status={}", @@ -2082,7 +2189,12 @@ impl QobuzClient { let ts = get_timestamp(); let sig = sign_request( "playlistget", - &[("extra", "tracks"), ("limit", &limit), ("offset", &offset_str), ("playlist_id", &pid)], + &[ + ("extra", "tracks"), + ("limit", &limit), + ("offset", &offset_str), + ("playlist_id", &pid), + ], ts, &secret, ); @@ -2165,21 +2277,18 @@ impl QobuzClient { /// Enumerate the award catalog (/award/explore). Paginated. /// Returns raw JSON items so callers can extract just (id, name). - pub async fn get_award_explore( - &self, - limit: u32, - offset: u32, - ) -> Result { + pub async fn get_award_explore(&self, limit: u32, offset: u32) -> Result { let url = endpoints::build_url(paths::AWARD_EXPLORE); - log::debug!("[API] get_award_explore(limit={}, offset={})", limit, offset); + log::debug!( + "[API] get_award_explore(limit={}, offset={})", + limit, + offset + ); let response: serde_json::Value = self .signed_get_auth( &url, "awardexplore", - &[ - ("limit", limit.to_string()), - ("offset", offset.to_string()), - ], + &[("limit", limit.to_string()), ("offset", offset.to_string())], ) .await? .json() @@ -2250,14 +2359,14 @@ impl QobuzClient { } // Current shape: V2AlbumGenericListDto {has_more, items}. - let mut items_value = response.get("items").cloned().unwrap_or(serde_json::Value::Null); + let mut items_value = response + .get("items") + .cloned() + .unwrap_or(serde_json::Value::Null); if let Some(items_arr) = items_value.as_array_mut() { for item in items_arr { if let Some(obj) = item.as_object_mut() { - let needs_backfill = obj - .get("artist") - .map(|v| v.is_null()) - .unwrap_or(true); + let needs_backfill = obj.get("artist").map(|v| v.is_null()).unwrap_or(true); if needs_backfill { if let Some(first_artist) = obj .get("artists") @@ -2313,7 +2422,11 @@ impl QobuzClient { offset ); let response: serde_json::Value = self - .signed_get(&url, "labelexplore", &[("limit", limit.to_string()), ("offset", offset.to_string())]) + .signed_get( + &url, + "labelexplore", + &[("limit", limit.to_string()), ("offset", offset.to_string())], + ) .await? .json() .await?; @@ -2342,13 +2455,28 @@ impl QobuzClient { ("limit", limit.to_string()), ("offset", offset.to_string()), ]; - if let Some(v) = sort { params.push(("sort", v.to_string())); } - if let Some(v) = order { params.push(("order", v.to_string())); } - if let Some(v) = genre_ids { params.push(("genre_ids", v.to_string())); } - if let Some(v) = from_date { params.push(("from_date", v.to_string())); } - if let Some(v) = to_date { params.push(("to_date", v.to_string())); } + if let Some(v) = sort { + params.push(("sort", v.to_string())); + } + if let Some(v) = order { + params.push(("order", v.to_string())); + } + if let Some(v) = genre_ids { + params.push(("genre_ids", v.to_string())); + } + if let Some(v) = from_date { + params.push(("from_date", v.to_string())); + } + if let Some(v) = to_date { + params.push(("to_date", v.to_string())); + } - log::debug!("[API] get_label_albums({}, limit={}, offset={})", label_id, limit, offset); + log::debug!( + "[API] get_label_albums({}, limit={}, offset={})", + label_id, + limit, + offset + ); let response: Value = self .signed_get(&url, "labelgetalbums", ¶ms) .await? @@ -2371,7 +2499,9 @@ impl QobuzClient { ("limit", limit.to_string()), ("offset", offset.to_string()), ]; - if let Some(v) = genre_ids { params.push(("genre_ids", v.to_string())); } + if let Some(v) = genre_ids { + params.push(("genre_ids", v.to_string())); + } log::debug!("[API] get_label_next_releases({})", label_id); let response: Value = self @@ -2398,9 +2528,15 @@ impl QobuzClient { ("limit", limit.to_string()), ("offset", offset.to_string()), ]; - if let Some(v) = sort { params.push(("sort", v.to_string())); } - if let Some(v) = order { params.push(("order", v.to_string())); } - if let Some(v) = genre_ids { params.push(("genre_ids", v.to_string())); } + if let Some(v) = sort { + params.push(("sort", v.to_string())); + } + if let Some(v) = order { + params.push(("order", v.to_string())); + } + if let Some(v) = genre_ids { + params.push(("genre_ids", v.to_string())); + } log::debug!("[API] get_label_awarded_releases({})", label_id); let response: Value = self @@ -2421,11 +2557,15 @@ impl QobuzClient { let url = endpoints::build_url(paths::LABEL_GET_PLAYLISTS); log::debug!("[API] get_label_playlists({})", label_id); let response: Value = self - .signed_get(&url, "labelgetplaylists", &[ - ("label_id", label_id.to_string()), - ("limit", limit.to_string()), - ("offset", offset.to_string()), - ]) + .signed_get( + &url, + "labelgetplaylists", + &[ + ("label_id", label_id.to_string()), + ("limit", limit.to_string()), + ("offset", offset.to_string()), + ], + ) .await? .json() .await?; @@ -2442,11 +2582,15 @@ impl QobuzClient { let url = endpoints::build_url(paths::LABEL_GET_TOP_ARTISTS); log::debug!("[API] get_label_top_artists({})", label_id); let response: Value = self - .signed_get(&url, "labelgettopartists", &[ - ("label_id", label_id.to_string()), - ("limit", limit.to_string()), - ("offset", offset.to_string()), - ]) + .signed_get( + &url, + "labelgettopartists", + &[ + ("label_id", label_id.to_string()), + ("limit", limit.to_string()), + ("offset", offset.to_string()), + ], + ) .await? .json() .await?; @@ -2463,11 +2607,15 @@ impl QobuzClient { let url = endpoints::build_url(paths::LABEL_STORY); log::debug!("[API] get_label_story({})", label_id); let response: Value = self - .signed_get(&url, "labelstory", &[ - ("label_id", label_id.to_string()), - ("limit", limit.to_string()), - ("offset", offset.to_string()), - ]) + .signed_get( + &url, + "labelstory", + &[ + ("label_id", label_id.to_string()), + ("limit", limit.to_string()), + ("offset", offset.to_string()), + ], + ) .await? .json() .await?; @@ -2489,7 +2637,12 @@ impl QobuzClient { .map(|id| id.to_string()) .collect::>() .join(","); - let sig = sign_request("labelgetList", &[("label_ids", &ids_str)], timestamp, &secret); + let sig = sign_request( + "labelgetList", + &[("label_ids", &ids_str)], + timestamp, + &secret, + ); let body = serde_json::json!({ "label_ids": label_ids }); log::debug!("[API] get_label_list POST ({} ids)", label_ids.len()); @@ -2669,9 +2822,11 @@ impl QobuzClient { // restriction — every quality would 403 the same way. Abort the // whole fallback loop immediately instead of firing 5 more // requests per track and feeding the storm (issue #637). - Err(e @ (ApiError::Forbidden(_) - | ApiError::ForbiddenCircuitOpen(_) - | ApiError::AuthenticationError(_))) => { + Err( + e @ (ApiError::Forbidden(_) + | ApiError::ForbiddenCircuitOpen(_) + | ApiError::AuthenticationError(_)), + ) => { log::warn!("Stream URL aborting quality fallback: {}", e); return Err(e); } @@ -2769,7 +2924,15 @@ impl QobuzClient { let url = endpoints::build_url(paths::PLAYLIST_SEARCH); let timestamp = get_timestamp(); let secret = self.secret().await?; - let signature = sign_search("playlistsearch", query, limit, offset, None, timestamp, &secret); + let signature = sign_search( + "playlistsearch", + query, + limit, + offset, + None, + timestamp, + &secret, + ); let limit_str = limit.to_string(); let offset_str = offset.to_string(); let ts_str = timestamp.to_string(); @@ -2828,8 +2991,12 @@ impl QobuzClient { pub async fn delete_playlist(&self, playlist_id: u64) -> Result<()> { let url = endpoints::build_url(paths::PLAYLIST_DELETE); - self.signed_get_auth(&url, "playlistdelete", &[("playlist_id", playlist_id.to_string())]) - .await?; + self.signed_get_auth( + &url, + "playlistdelete", + &[("playlist_id", playlist_id.to_string())], + ) + .await?; Ok(()) } @@ -2843,10 +3010,14 @@ impl QobuzClient { .collect::>() .join(","); - self.signed_get_auth(&url, "playlistaddTracks", &[ - ("playlist_id", playlist_id.to_string()), - ("track_ids", track_ids_str), - ]) + self.signed_get_auth( + &url, + "playlistaddTracks", + &[ + ("playlist_id", playlist_id.to_string()), + ("track_ids", track_ids_str), + ], + ) .await?; Ok(()) @@ -2865,10 +3036,14 @@ impl QobuzClient { .collect::>() .join(","); - self.signed_get_auth(&url, "playlistdeleteTracks", &[ - ("playlist_id", playlist_id.to_string()), - ("playlist_track_ids", track_ids_str), - ]) + self.signed_get_auth( + &url, + "playlistdeleteTracks", + &[ + ("playlist_id", playlist_id.to_string()), + ("playlist_track_ids", track_ids_str), + ], + ) .await?; Ok(()) @@ -2919,11 +3094,15 @@ impl QobuzClient { insert_before ); - self.signed_get_auth(&url, "playlistupdateTracksPosition", &[ - ("playlist_id", playlist_id.to_string()), - ("playlist_track_ids", track_ids_str), - ("insert_before", insert_before.to_string()), - ]) + self.signed_get_auth( + &url, + "playlistupdateTracksPosition", + &[ + ("playlist_id", playlist_id.to_string()), + ("playlist_track_ids", track_ids_str), + ("insert_before", insert_before.to_string()), + ], + ) .await?; Ok(()) @@ -2965,7 +3144,11 @@ impl QobuzClient { let url = endpoints::build_url(paths::PLAYLIST_SUBSCRIBE); let response = self - .signed_get_auth(&url, "playlistsubscribe", &[("playlist_id", playlist_id.to_string())]) + .signed_get_auth( + &url, + "playlistsubscribe", + &[("playlist_id", playlist_id.to_string())], + ) .await?; let status = response.status(); @@ -2985,7 +3168,11 @@ impl QobuzClient { let url = endpoints::build_url(paths::PLAYLIST_UNSUBSCRIBE); let response = self - .signed_get_auth(&url, "playlistunsubscribe", &[("playlist_id", playlist_id.to_string())]) + .signed_get_auth( + &url, + "playlistunsubscribe", + &[("playlist_id", playlist_id.to_string())], + ) .await?; let status = response.status(); @@ -3006,7 +3193,11 @@ impl QobuzClient { let type_key = format!("{}_ids", fav_type); // album_ids, track_ids, artist_ids let response = self - .signed_get_auth(&url, "favoritecreate", &[(type_key.as_str(), item_id.to_string())]) + .signed_get_auth( + &url, + "favoritecreate", + &[(type_key.as_str(), item_id.to_string())], + ) .await?; if response.status().is_success() { @@ -3025,7 +3216,11 @@ impl QobuzClient { let type_key = format!("{}_ids", fav_type); let response = self - .signed_get_auth(&url, "favoritedelete", &[(type_key.as_str(), item_id.to_string())]) + .signed_get_auth( + &url, + "favoritedelete", + &[(type_key.as_str(), item_id.to_string())], + ) .await?; if response.status().is_success() { @@ -3054,7 +3249,14 @@ impl QobuzClient { log::debug!("[API] get_artist_page({}) sort={:?}", artist_id, sort); let response: serde_json::Value = self - .signed_get(&url, "artistpage", &query.iter().map(|(k, v)| (*k, v.clone())).collect::>()) + .signed_get( + &url, + "artistpage", + &query + .iter() + .map(|(k, v)| (*k, v.clone())) + .collect::>(), + ) .await? .json() .await?; @@ -3090,7 +3292,14 @@ impl QobuzClient { offset ); let response: serde_json::Value = self - .signed_get(&url, "artistgetReleasesGrid", &query.iter().map(|(k, v)| (*k, v.clone())).collect::>()) + .signed_get( + &url, + "artistgetReleasesGrid", + &query + .iter() + .map(|(k, v)| (*k, v.clone())) + .collect::>(), + ) .await? .json() .await?; @@ -3120,7 +3329,14 @@ impl QobuzClient { limit ); let response: serde_json::Value = self - .signed_get(&url, "artiststory", &query.iter().map(|(k, v)| (*k, v.clone())).collect::>()) + .signed_get( + &url, + "artiststory", + &query + .iter() + .map(|(k, v)| (*k, v.clone())) + .collect::>(), + ) .await? .json() .await?; @@ -3230,11 +3446,7 @@ impl QobuzClient { /// /// This is the new streaming endpoint that returns encrypted CMAF segments /// instead of a direct file URL. - pub async fn get_file_url( - &self, - track_id: u64, - quality: Quality, - ) -> Result { + pub async fn get_file_url(&self, track_id: u64, quality: Quality) -> Result { let format_id = quality.id(); let url = endpoints::build_url(paths::FILE_URL); @@ -3295,11 +3507,7 @@ impl QobuzClient { let code = status.as_u16(); if status == reqwest::StatusCode::FORBIDDEN { let preview = body_preview(response).await; - log::warn!( - "[CMAF] file/url 403 for track {}{}", - track_id, - preview - ); + log::warn!("[CMAF] file/url 403 for track {}{}", track_id, preview); return Err(ApiError::Forbidden(preview)); } return Err(if code == 404 { @@ -3316,10 +3524,7 @@ impl QobuzClient { } else if status.is_server_error() { ApiError::ServerError(code) } else { - ApiError::ApiResponse(format!( - "file/url failed with status {}", - status - )) + ApiError::ApiResponse(format!("file/url failed with status {}", status)) }); } diff --git a/crates/qbz-qt/src/auth_qt.rs b/crates/qbz-qt/src/auth_qt.rs index 2d8e8bedd..1e2bb1e7e 100644 --- a/crates/qbz-qt/src/auth_qt.rs +++ b/crates/qbz-qt/src/auth_qt.rs @@ -105,14 +105,7 @@ where let client = guard.as_ref().ok_or("Qobuz client not initialized")?; match client.login_with_oauth_code(&code).await { Ok(session) => session, - Err(e) => { - // D4 producer: ONLY an explicit ineligible-account verdict - // starts the grace clock. Generic 401/network errors never do. - if matches!(e, qbz_qobuz::ApiError::IneligibleUser) { - crate::offline_fwd::subscription_mark_invalid(); - } - return Err(e.to_string()); - } + Err(e) => return Err(e.to_string()), } }; let user_id = session.user_id; @@ -160,7 +153,11 @@ where crate::playlist_qt::set_user_id(user_id); bind_per_user_stores(&dir, user_id).await; } - crate::offline_fwd::subscription_mark_valid(); + // The subscription verdict (D4) and the offline-cache purge run in the + // shared session activation (`qbz_app::subscription_lifecycle`); the + // shell only hands the runtime its cache, which activates after the + // session and is enforced against on registration. + runtime.set_offline_cache(crate::offline_qt::get().await); crate::offline_fwd::engine().set_offline_session(false); crate::media_sync_qt::resume_jellyfin_quality(); @@ -336,7 +333,7 @@ where crate::playlist_qt::set_user_id(user_id); bind_per_user_stores(&dir, user_id).await; } - crate::offline_fwd::subscription_mark_valid(); + runtime.set_offline_cache(crate::offline_qt::get().await); 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}"); @@ -348,14 +345,6 @@ where Err(e) if is_auth_rejection(&e) => { log::warn!("[qbz-qt] saved token rejected by Qobuz, clearing: {e}"); let _ = qbz_credentials::clear_oauth_token(); - // D4 producer: only the explicit ineligible verdict starts the - // grace clock; a plain 401 does not. - if matches!( - &e, - qbz_core::CoreError::Api(qbz_qobuz::ApiError::IneligibleUser) - ) { - crate::offline_fwd::subscription_mark_invalid(); - } Ok(None) } Err(e) => { @@ -487,6 +476,7 @@ where // The offline download cache (index.db/library connections + the cached // id set) — torn down so the next account never reads this one's rows. crate::offline_qt::deactivate().await; + runtime.set_offline_cache(None); runtime.deactivate().await?; log::info!("[qbz-qt] logged out"); Ok(()) diff --git a/crates/qbz-qt/src/offline_fwd.rs b/crates/qbz-qt/src/offline_fwd.rs index 91925e69c..a0ad92838 100644 --- a/crates/qbz-qt/src/offline_fwd.rs +++ b/crates/qbz-qt/src/offline_fwd.rs @@ -199,54 +199,3 @@ pub fn offline_playback_allowed() -> bool { }) .unwrap_or(true) } - -/// D4 producer: a successful login verdict. Clears any running grace clock. -pub fn subscription_mark_valid() { - let now = now_unix_secs(); - match SUBSCRIPTION.lock() { - Ok(guard) => match guard.as_ref() { - Some(store) => { - if let Err(e) = store.mark_valid(now) { - log::error!("[qbz-qt] subscription mark_valid failed: {e}"); - } - } - None => log::warn!("[qbz-qt] subscription mark_valid: no store open"), - }, - Err(e) => log::error!("[qbz-qt] subscription store lock poisoned: {e}"), - } -} - -/// D4 producer: an EXPLICIT ineligible-account login verdict -/// (`ApiError::IneligibleUser`). Generic 401/network errors must never -/// reach this — the grace clock only starts on a real verdict. -/// -/// An ineligible verdict can arrive before any session activation, so when -/// no store is open this falls back to transiently opening the LAST user's -/// store. -pub fn subscription_mark_invalid() { - let now = now_unix_secs(); - if let Ok(guard) = SUBSCRIPTION.lock() { - if let Some(store) = guard.as_ref() { - if let Err(e) = store.mark_invalid(now) { - log::error!("[qbz-qt] subscription mark_invalid failed: {e}"); - } - return; - } - } - let Some(user_id) = UserDataPaths::load_last_user_id() else { - log::warn!("[qbz-qt] subscription mark_invalid: no previous user, skipping"); - return; - }; - let Some(dir) = user_data_dir(user_id) else { - log::warn!("[qbz-qt] subscription mark_invalid: data dir unavailable"); - return; - }; - match SubscriptionStateStore::new_at(&dir) { - Ok(store) => { - if let Err(e) = store.mark_invalid(now) { - log::error!("[qbz-qt] subscription mark_invalid failed: {e}"); - } - } - Err(e) => log::error!("[qbz-qt] subscription state store open failed: {e}"), - } -}