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
This commit is contained in:
vicrodh committed 2026-09-02 07:41:09 -06:00
1 parent 2b7d804f09
commit 2e34112868
8 files changed
+681 -214

No files matched your search

+1
View File
@@ -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 }
+1
View File
@@ -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;
+1 -1
View File
@@ -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")]
+118 -9
View File
@@ -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<Mutex<Option<Arc<OfflineCacheState>>>>;
/// 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<A: FrontendAdapter + Send + Sync + 'static> {
/// 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<VisualizerTap>,
/// 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<Option<JoinHandle<()>>>,
}
impl<A: FrontendAdapter + Send + Sync + 'static> AppRuntime<A> {
@@ -66,6 +85,8 @@ impl<A: FrontendAdapter + Send + Sync + 'static> AppRuntime<A> {
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<A: FrontendAdapter + Send + Sync + 'static> AppRuntime<A> {
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<Self>, cache: Option<Arc<OfflineCacheState>>) {
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<PathBuf> {
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<A: FrontendAdapter + Send + Sync + 'static> AppRuntime<A> {
/// `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
@@ -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<i64>,
/// When the grace window ends (`invalid_since + GRACE_PERIOD_SECS`).
pub grace_deadline: Option<i64>,
/// 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<bool, String> {
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<SubscriptionVerdict, String> {
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<Arc<OfflineCacheState>>,
now: i64,
) -> Result<SubscriptionVerdict, String> {
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);
}
}
+340 -135
View File
@@ -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<UserSession> {
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<SearchResultsPage<Artist>> {
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", &params)
.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", &params).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<RadioResponse> {
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<RadioResponse> {
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<RadioResponse> {
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<RadioResponse> {
@@ -1312,7 +1357,14 @@ impl QobuzClient {
}
let http_response = self
.signed_get(&url, "genrelist", &query.iter().map(|(k, v)| (*k, v.clone())).collect::<Vec<_>>())
.signed_get(
&url,
"genrelist",
&query
.iter()
.map(|(k, v)| (*k, v.clone()))
.collect::<Vec<_>>(),
)
.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::<Vec<_>>())
.signed_get_auth(
&url,
"discoverindex",
&query
.iter()
.map(|(k, v)| (*k, v.clone()))
.collect::<Vec<_>>(),
)
.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::<Vec<_>>())
.signed_get_auth(
&url,
&method_name,
&query
.iter()
.map(|(k, v)| (*k, v.clone()))
.collect::<Vec<_>>(),
)
.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::<Vec<_>>())
.signed_get_auth(
&url,
"discoverplaylists",
&query
.iter()
.map(|(k, v)| (*k, v.clone()))
.collect::<Vec<_>>(),
)
.await?
.json()
.await?;
@@ -1497,9 +1570,7 @@ impl QobuzClient {
pub async fn get_playlist_tags(&self) -> Result<Vec<PlaylistTag>> {
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::<QobuzLyricsDocument>(&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::<Vec<_>>())
.signed_get(
&url,
"artistget",
&query
.iter()
.map(|(k, v)| (*k, v.clone()))
.collect::<Vec<_>>(),
)
.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::<Vec<_>>())
.signed_get(
&url,
"artistget",
&query
.iter()
.map(|(k, v)| (*k, v.clone()))
.collect::<Vec<_>>(),
)
.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::<Vec<_>>())
.signed_get(
&url,
"artistget",
&query
.iter()
.map(|(k, v)| (*k, v.clone()))
.collect::<Vec<_>>(),
)
.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<PlaylistWithTrackIds> {
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::<Vec<_>>().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::<Vec<_>>()
.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<serde_json::Value> {
pub async fn get_award_explore(&self, limit: u32, offset: u32) -> Result<serde_json::Value> {
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", &params)
.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::<Vec<_>>()
.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::<Vec<_>>()
.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::<Vec<_>>()
.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::<Vec<_>>())
.signed_get(
&url,
"artistpage",
&query
.iter()
.map(|(k, v)| (*k, v.clone()))
.collect::<Vec<_>>(),
)
.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::<Vec<_>>())
.signed_get(
&url,
"artistgetReleasesGrid",
&query
.iter()
.map(|(k, v)| (*k, v.clone()))
.collect::<Vec<_>>(),
)
.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::<Vec<_>>())
.signed_get(
&url,
"artiststory",
&query
.iter()
.map(|(k, v)| (*k, v.clone()))
.collect::<Vec<_>>(),
)
.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<TrackFileUrl> {
pub async fn get_file_url(&self, track_id: u64, quality: Quality) -> Result<TrackFileUrl> {
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))
});
}
+8 -18
View File
@@ -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(())
-51
View File
@@ -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}"),
}
}