From 68e6f528e53281864c22d659817dec13cc8dd5a5 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Mon, 15 Dec 2025 15:05:35 +0100 Subject: [PATCH] debugage lazy cache --- Cargo.lock | 2 + PMOMusic/src/main.rs | 8 +- .../src/components/AudioCacheManager.vue | 165 +++++++++++-- pmoaudiocache/src/cache.rs | 1 - pmocache/Cargo.toml | 1 + pmocache/src/cache.rs | 226 ++++++++++++++---- pmocache/src/db.rs | 73 +++--- pmocache/src/lazy.rs | 42 ++++ pmocache/src/lib.rs | 2 + pmocache/src/pmoserver_ext.rs | 110 +++------ pmoqobuz/Cargo.toml | 5 +- pmoqobuz/src/lazy_provider.rs | 91 +++++++ pmoqobuz/src/lib.rs | 1 + pmoqobuz/src/source.rs | 21 +- .../__pycache__/raw.cpython-312.pyc | Bin 16676 -> 16676 bytes pmoqobuz/test_python_qobuz/test_getfileurl.py | 2 +- pmosource/Cargo.toml | 3 +- pmosource/src/cache.rs | 82 ++++++- 18 files changed, 641 insertions(+), 194 deletions(-) create mode 100644 pmocache/src/lazy.rs create mode 100644 pmoqobuz/src/lazy_provider.rs diff --git a/Cargo.lock b/Cargo.lock index 070e0c0b..707f3eb0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3281,6 +3281,7 @@ name = "pmocache" version = "0.1.0" dependencies = [ "anyhow", + "async-trait", "axum 0.8.7", "bytes", "chrono", @@ -3603,6 +3604,7 @@ dependencies = [ "futures", "lazy_static", "pmoaudiocache", + "pmocache", "pmoconfig", "pmocovers", "pmodidl", diff --git a/PMOMusic/src/main.rs b/PMOMusic/src/main.rs index 9db7cac2..9e44eb03 100644 --- a/PMOMusic/src/main.rs +++ b/PMOMusic/src/main.rs @@ -38,10 +38,10 @@ async fn main() -> Result<(), Box> { // Enregistrer les sources musicales info!("🎵 Registering music sources..."); - // // Enregistrer Qobuz - // if let Err(e) = server.write().await.register_qobuz().await { - // tracing::warn!("⚠️ Failed to register Qobuz: {}", e); - // } + // Enregistrer Qobuz pour activer les lazy providers (QOBUZ:PK) + if let Err(e) = server.write().await.register_qobuz().await { + tracing::warn!("⚠️ Failed to register Qobuz source: {}", e); + } // Initialiser les canaux de streaming Radio Paradise (pipelines + routes HTTP) info!("📻 Initializing Radio Paradise streaming channels..."); diff --git a/pmoapp/webapp/src/components/AudioCacheManager.vue b/pmoapp/webapp/src/components/AudioCacheManager.vue index cac19b94..09df4dda 100644 --- a/pmoapp/webapp/src/components/AudioCacheManager.vue +++ b/pmoapp/webapp/src/components/AudioCacheManager.vue @@ -87,7 +87,13 @@
🎵
{{ track.hits }} plays - Lazy + + {{ lazyBadgeLabel(track) }} +
@@ -102,7 +108,16 @@
{{ track.pk }} - lazy + + Lazy + + {{ lazyProviderName(track) }} + +
@@ -125,7 +140,9 @@ Last used: {{ formatDate(track.last_used) }}
- Audio not downloaded yet. First playback (or forcing download) will fetch it automatically. + Audio not downloaded yet + ({{ lazyProviderName(track) }} provider). + First playback (or forcing download) will fetch it automatically.
@@ -279,13 +296,30 @@ import { getCoverUrl, } from "../services/audioCache"; +interface LazyDisplayInfo { + prefix: string; + display: string; + className: string; + isLegacy: boolean; +} + +const LAZY_PROVIDER_LABELS: Record = { + QOBUZ: "Qobuz", +}; +const LEGACY_LAZY_INFO: LazyDisplayInfo = { + prefix: "legacy", + display: "Legacy", + className: "lazy-provider-legacy", + isLegacy: true, +}; + // --- États --- const tracks = ref([]); const selectedTrack = ref(null); const isLoading = ref(false); const sortBy = ref<"hits" | "last_used" | "recent">("hits"); const audioPlayer = ref(null); -const LAZY_PREFIX = "L:"; +const LEGACY_LAZY_PREFIX = "L:"; // Formulaire d'ajout const newTrackUrl = ref(""); @@ -531,13 +565,73 @@ function handleCoverError(pk: string) { failedCovers.value.add(pk); } +function prettifyLazyProvider(prefix: string): string { + if (LAZY_PROVIDER_LABELS[prefix]) { + return LAZY_PROVIDER_LABELS[prefix]; + } + const normalized = prefix.replace(/[^A-Za-z0-9]+/g, " ").trim(); + if (!normalized) { + return prefix.trim().toUpperCase() || "Lazy"; + } + return normalized + .split(/\s+/) + .map((segment) => segment.charAt(0).toUpperCase() + segment.slice(1).toLowerCase()) + .join(" "); +} + +function buildLazyClass(prefix: string): string { + return `lazy-provider-${prefix.toLowerCase().replace(/[^a-z0-9]+/g, "-")}`; +} + +function extractLazyInfoFromPk(pk: string | undefined | null): LazyDisplayInfo | undefined { + if (!pk) return undefined; + if (pk.startsWith(LEGACY_LAZY_PREFIX)) { + return LEGACY_LAZY_INFO; + } + const separatorIndex = pk.indexOf(":"); + if (separatorIndex <= 0) { + return undefined; + } + const prefix = pk.slice(0, separatorIndex); + if (!prefix) { + return undefined; + } + return { + prefix, + display: prettifyLazyProvider(prefix), + className: buildLazyClass(prefix), + isLegacy: false, + }; +} + +function getLazyInfo(track: AudioCacheEntry | null | undefined): LazyDisplayInfo | undefined { + if (!track?.pk) return undefined; + return extractLazyInfoFromPk(track.pk); +} + function isLazyTrack(track: AudioCacheEntry | null | undefined): boolean { - return !!track?.pk && track.pk.startsWith(LAZY_PREFIX); + return !!getLazyInfo(track); +} + +function lazyProviderName(track: AudioCacheEntry | null | undefined): string | undefined { + return getLazyInfo(track)?.display; +} + +function lazyProviderClass(track: AudioCacheEntry | null | undefined): string { + return getLazyInfo(track)?.className ?? "lazy-provider-generic"; +} + +function lazyBadgeLabel(track: AudioCacheEntry | null | undefined): string { + const info = getLazyInfo(track); + if (!info) return ""; + return info.isLegacy ? "Lazy" : `Lazy - ${info.display}`; } function trackStatusLabel(track: AudioCacheEntry | null): string { - if (isLazyTrack(track)) { - return "Lazy (audio pending download)"; + const info = getLazyInfo(track); + if (info) { + const provider = info.isLegacy ? "" : ` - ${info.display}`; + return `Lazy${provider} (audio pending download)`; } return "Cached"; } @@ -822,11 +916,14 @@ button:disabled { } .track-overlay .hits.lazy { - background: rgba(255, 152, 0, 0.85); - color: #000; + display: inline-flex; + align-items: center; + background: rgba(156, 39, 176, 0.85); + color: #fff; font-weight: bold; - padding: 0.2rem 0.5rem; + padding: 0.2rem 0.6rem; border-radius: 999px; + font-size: 0.8rem; } .track-info { @@ -869,15 +966,30 @@ button:disabled { } .lazy-tag { - display: inline-block; + display: inline-flex; + align-items: center; + gap: 0.35rem; margin-left: 0.5rem; - padding: 0.1rem 0.4rem; + padding: 0.15rem 0.6rem; border-radius: 999px; - background: #ff9800; - color: #000; - font-size: 0.65rem; + background: rgba(156, 39, 176, 0.25); + color: #f5f5f5; + font-size: 0.7rem; text-transform: uppercase; - font-weight: bold; + font-weight: 600; + border: 1px solid rgba(156, 39, 176, 0.4); +} + +.lazy-tag .lazy-label { + font-weight: 700; + letter-spacing: 0.5px; +} + +.lazy-tag .lazy-provider-name { + font-size: 0.6rem; + text-transform: none; + letter-spacing: 0.4px; + opacity: 0.9; } .meta { @@ -911,6 +1023,27 @@ button:disabled { border-radius: 6px; } +.track-overlay .hits.lazy.lazy-provider-legacy, +.lazy-tag.lazy-provider-legacy { + background: rgba(255, 152, 0, 0.85); + color: #000; + border-color: rgba(255, 193, 7, 0.8); +} + +.track-overlay .hits.lazy.lazy-provider-qobuz, +.lazy-tag.lazy-provider-qobuz { + background: rgba(76, 175, 80, 0.9); + color: #fff; + border-color: rgba(165, 214, 167, 0.9); +} + +.track-overlay .hits.lazy.lazy-provider-generic, +.lazy-tag.lazy-provider-generic { + background: rgba(103, 58, 183, 0.85); + color: #fff; + border-color: rgba(179, 157, 219, 0.8); +} + .track-actions { padding: 0 1rem 1rem; display: flex; diff --git a/pmoaudiocache/src/cache.rs b/pmoaudiocache/src/cache.rs index 3b0d7ac4..311bfc55 100644 --- a/pmoaudiocache/src/cache.rs +++ b/pmoaudiocache/src/cache.rs @@ -8,7 +8,6 @@ use crate::metadata_ext::AudioTrackMetadataExt; use anyhow::Result; use pmocache::download::TransformMetadata; use pmocache::CacheConfig; -use serde_json::Value; use std::sync::Arc; /// Configuration pour le cache audio diff --git a/pmocache/Cargo.toml b/pmocache/Cargo.toml index 8cf63f3d..b69a4337 100644 --- a/pmocache/Cargo.toml +++ b/pmocache/Cargo.toml @@ -18,6 +18,7 @@ hex = "0.4" # Utilitaires anyhow = "1.0" +async-trait = "0.1" chrono = "0.4" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" diff --git a/pmocache/src/cache.rs b/pmocache/src/cache.rs index b59125da..74f869e1 100755 --- a/pmocache/src/cache.rs +++ b/pmocache/src/cache.rs @@ -8,17 +8,23 @@ use crate::db::DB; use crate::download::{ download_with_transformer, ingest_with_transformer, Download, StreamTransformer, }; +use crate::lazy::{lazy_prefix_from_pk, LazyEntryRemoteData, LazyProvider}; use anyhow::{anyhow, bail, Result}; use serde_json::{Number, Value}; use sha2::{Digest, Sha256}; use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, RwLock as StdRwLock}; use tokio::io::{AsyncRead, AsyncReadExt}; use tokio::sync::{broadcast, RwLock}; use tracing; +enum FinalizeMode<'a> { + InsertNew, + ConvertLazy { lazy_pk: &'a str }, +} + // ============================================================================ // LAZY PK SUPPORT // ============================================================================ @@ -41,7 +47,11 @@ pub fn generate_lazy_pk(url: &str) -> String { /// VĂ©rifie si un PK est en mode lazy pub fn is_lazy_pk(pk: &str) -> bool { - pk.starts_with(LAZY_PK_PREFIX) + if pk.starts_with(LAZY_PK_PREFIX) { + return true; + } + + lazy_prefix_from_pk(pk).is_some() } /// Events Ă©mis par le cache pour notifier les changements d'Ă©tat @@ -136,6 +146,8 @@ pub struct Cache { min_prebuffer_size: u64, /// LAZY PK SUPPORT: Channel pour broadcaster les events (lazy downloads, etc.) served_tx: Option>, + /// Providers responsables de prĂ©fixes lazy spĂ©cifiques + lazy_providers: StdRwLock>>, /// Phantom data pour le type de configuration _phantom: std::marker::PhantomData, } @@ -242,6 +254,7 @@ impl Cache { download: Arc, collection: Option<&str>, origin_url: Option<&str>, + mode: FinalizeMode<'_>, ) -> Result { // Attendre le prĂ©buffering (pour le cache progressif) if self.min_prebuffer_size > 0 { @@ -256,10 +269,17 @@ impl Cache { ); } - // Ajouter Ă  la DB une fois le prĂ©buffer terminĂ© - self.db.add(pk, None, collection)?; - if let Some(url) = origin_url { - self.db.set_origin_url(pk, url)?; + // Ajouter ou commuter la DB selon le mode + match mode { + FinalizeMode::InsertNew => { + self.db.add(pk, None, collection)?; + if let Some(url) = origin_url { + self.db.set_origin_url(pk, url)?; + } + } + FinalizeMode::ConvertLazy { lazy_pk } => { + self.db.update_lazy_to_downloaded(lazy_pk, pk)?; + } } // Sauvegarder les mĂ©tadonnĂ©es techniques du transformer @@ -400,6 +420,7 @@ impl Cache { transformer_factory, min_prebuffer_size: DEFAULT_PREBUFFER_SIZE, served_tx: Some(served_tx), + lazy_providers: StdRwLock::new(HashMap::new()), _phantom: std::marker::PhantomData, }) } @@ -432,6 +453,34 @@ impl Cache { self.min_prebuffer_size } + /// Enregistre un provider responsable d'un prĂ©fixe de lazy PK. + pub fn register_lazy_provider(&self, provider: Arc) { + let prefix = provider.lazy_prefix().to_string(); + let mut guard = self + .lazy_providers + .write() + .expect("lazy provider registry poisoned"); + guard.insert(prefix, provider); + } + + /// DĂ©senregistre un provider Ă  partir de son prĂ©fixe. + pub fn unregister_lazy_provider(&self, prefix: &str) { + let mut guard = self + .lazy_providers + .write() + .expect("lazy provider registry poisoned"); + guard.remove(prefix); + } + + fn provider_for_lazy_pk(&self, lazy_pk: &str) -> Option> { + let prefix = lazy_prefix_from_pk(lazy_pk)?; + let guard = self + .lazy_providers + .read() + .expect("lazy provider registry poisoned"); + guard.get(prefix).cloned() + } + /// S'abonne aux diffusions HTTP pour un `pk` donnĂ©. /// /// La callback est appelĂ©e Ă  chaque fois qu'un Ă©lĂ©ment est servi avec succès via les routes @@ -611,10 +660,63 @@ impl Cache { } // Finaliser avec prĂ©buffering et nettoyage - self.finalize_download(&pk, download, collection, Some(url)) + self.finalize_download(&pk, download, collection, Some(url), FinalizeMode::InsertNew) .await } + /// TĂ©lĂ©charge un fichier lazy et commute l'entrĂ©e existante + pub async fn download_lazy_from_url( + &self, + lazy_pk: &str, + url: &str, + collection: Option<&str>, + ) -> Result { + // Si dĂ©jĂ  converti, retourner directement + if let Ok(Some(real_pk)) = self.db.get_pk_by_lazy_pk(lazy_pk) { + return Ok(real_pk); + } + + // Si l'URL pointe dĂ©jĂ  vers un fichier complet, commuter sans re-tĂ©lĂ©charger + if let Ok(Some(existing_pk)) = self.db.get_pk_by_origin_url(url) { + if existing_pk != lazy_pk && self.check_cached_and_complete(&existing_pk).await? { + self.db.update_lazy_to_downloaded(lazy_pk, &existing_pk)?; + return Ok(existing_pk); + } + } + + // 1. TĂ©lĂ©charger les 2048 premiers octets pour calculer le pk + let header = crate::download::peek_header(url, 2048) + .await + .map_err(|e| anyhow!("Failed to peek header: {}", e))?; + let pk = crate::cache_trait::pk_from_content_header(&header); + + // 2. Si dĂ©jĂ  en cache (complet), commuter directement + if self.check_cached_and_complete(&pk).await? { + self.db.update_lazy_to_downloaded(lazy_pk, &pk)?; + return Ok(pk); + } + + // 3. Lancer le tĂ©lĂ©chargement complet + tracing::debug!("Starting lazy download for pk {} (lazy {})", pk, lazy_pk); + let file_path = self.get_file_path(&pk); + let transformer = self.transformer_factory.as_ref().map(|f| f()); + let download = download_with_transformer(&file_path, url, transformer); + + { + let mut downloads = self.downloads.write().await; + downloads.insert(pk.clone(), download.clone()); + } + + self.finalize_download( + &pk, + download, + collection, + Some(url), + FinalizeMode::ConvertLazy { lazy_pk }, + ) + .await + } + /// Ajoute un fichier Ă  partir d'un flux asynchrone. /// /// Cette mĂ©thode utilise le mĂŞme système d'identifiants basĂ© sur le contenu que `add_from_url`. @@ -725,7 +827,7 @@ impl Cache { } // Finaliser avec prĂ©buffering et nettoyage - self.finalize_download(&pk, download, collection, source_uri) + self.finalize_download(&pk, download, collection, source_uri, FinalizeMode::InsertNew) .await } @@ -1297,71 +1399,95 @@ impl Cache { } } - /// Ajoute une URL en mode deferred (pas de download immĂ©diat) - /// - /// VĂ©rifie d'abord si l'URL existe dĂ©jĂ  en DB : - /// - Si eager (dĂ©jĂ  tĂ©lĂ©chargĂ©) : retourne le lazy_pk si existe, sinon le real pk - /// - Si lazy (pas encore tĂ©lĂ©chargĂ©) : retourne le lazy pk existant - /// - Sinon : crĂ©e nouvelle entry lazy - /// - /// # Arguments - /// - /// * `url` - URL Ă  cacher - /// * `collection` - Collection optionnelle - /// - /// # Returns - /// - /// PK (lazy ou real selon l'Ă©tat) - /// - /// # Example - /// - /// ```rust,no_run - /// let cache = Cache::::new("./cache", 1000)?; - /// let lazy_pk = cache.add_from_url_deferred("https://example.com/track.mp3", Some("qobuz")).await?; - /// // → Returns "L:abc123..." (lazy PK) - /// // Fichier pas encore tĂ©lĂ©chargĂ©, juste mĂ©tadonnĂ©es en DB - /// ``` + /// Garantit l'existence d'une entrĂ©e lazy spĂ©cifique. + pub async fn ensure_lazy_entry( + &self, + lazy_pk: &str, + collection: Option<&str>, + origin_url: Option<&str>, + ) -> Result<()> { + if let Ok(true) = self.db.has_lazy_entry(lazy_pk) { + self.db.update_hit_by_lazy_pk(lazy_pk)?; + } else { + self.db.add_lazy(lazy_pk, None, collection)?; + } + + if let Some(url) = origin_url { + self.db.set_origin_url_for_lazy(lazy_pk, url)?; + } + + Ok(()) + } + + /// RĂ©cupère auprès du provider les mĂ©tadonnĂ©es/couvertures associĂ©es. + pub async fn fetch_lazy_provider_data(&self, lazy_pk: &str) -> Result { + if let Some(provider) = self.provider_for_lazy_pk(lazy_pk) { + let metadata = provider.metadata(lazy_pk).await?; + let cover_url = provider.cover_url(lazy_pk).await?; + Ok(LazyEntryRemoteData { metadata, cover_url }) + } else { + Ok(LazyEntryRemoteData::default()) + } + } + + /// RĂ©sout l'URL d'origine pour un lazy PK, via la DB ou un provider. + pub async fn resolve_lazy_url(&self, lazy_pk: &str) -> Result { + if let Ok(Some(url)) = self.db.get_origin_url(lazy_pk) { + return Ok(url); + } + + if let Some(provider) = self.provider_for_lazy_pk(lazy_pk) { + return provider.get_url(lazy_pk).await; + } + + bail!("No origin URL or lazy provider registered for {}", lazy_pk); + } + + /// TĂ©lĂ©charge un lazy PK en rĂ©solvant automatiquement son URL. + pub async fn download_lazy(&self, lazy_pk: &str, collection: Option<&str>) -> Result { + let (existing_pk, _lazy_sec, existing_collection) = self + .db + .get_entry_by_pk_or_lazy_pk(lazy_pk)? + .ok_or_else(|| anyhow!("Lazy pk {} not found in DB", lazy_pk))?; + + let url = self.resolve_lazy_url(lazy_pk).await?; + let collection = collection.or(existing_collection.as_deref()); + + if let Some(real_pk) = existing_pk { + if real_pk != lazy_pk && self.check_cached_and_complete(&real_pk).await? { + return Ok(real_pk); + } + } + + self.download_lazy_from_url(lazy_pk, &url, collection).await + } + + /// Ajoute une URL sans lancer immĂ©diatement le tĂ©lĂ©chargement. pub async fn add_from_url_deferred( &self, url: &str, collection: Option<&str>, ) -> Result { - // 1. VĂ©rifier si URL dĂ©jĂ  en cache if let Ok(Some((pk_opt, lazy_pk_opt))) = self.db.get_entry_by_url(url) { - // URL existe dĂ©jĂ  if let Some(pk) = pk_opt { - // Fichier dĂ©jĂ  tĂ©lĂ©chargĂ© (eager ou lazy→eager) - tracing::debug!("URL {} already downloaded with pk {}", url, pk); self.db.update_hit(&pk)?; - - // Retourner lazy_pk si existe (pour compatibilitĂ© Control Point) - // sinon retourner pk if let Some(lpk) = lazy_pk_opt { return Ok(lpk); } return Ok(pk); } else if let Some(lpk) = lazy_pk_opt { - // Entry lazy existante (pas encore tĂ©lĂ©chargĂ©) - tracing::debug!("URL {} already in lazy mode with pk {}", url, lpk); self.db.update_hit_by_lazy_pk(&lpk)?; return Ok(lpk); } } - // 2. URL inconnue → crĂ©er nouvelle entry lazy - let lazy_pk = generate_lazy_pk(url); - - // VĂ©rifier si ce lazy_pk existe dĂ©jĂ  (collision improbable mais...) + let lazy_pk = format!("L:{}", generate_lazy_pk(url)); if let Ok(true) = self.db.has_lazy_entry(&lazy_pk) { bail!("Lazy PK collision for URL: {}", url); } - // 3. Ajouter en DB - self.db.add_lazy(&lazy_pk, None, collection)?; - self.db.set_origin_url_for_lazy(&lazy_pk, url)?; - + self.ensure_lazy_entry(&lazy_pk, collection, Some(url)).await?; tracing::debug!("Created new lazy pk {} for URL {}", lazy_pk, url); - Ok(lazy_pk) } } diff --git a/pmocache/src/db.rs b/pmocache/src/db.rs index 3b799141..a1a41757 100644 --- a/pmocache/src/db.rs +++ b/pmocache/src/db.rs @@ -111,6 +111,7 @@ impl DB { /// ``` pub fn init(path: &Path) -> Result { let conn = Connection::open(path)?; + conn.execute("PRAGMA foreign_keys = ON", [])?; conn.execute( "CREATE TABLE IF NOT EXISTS asset ( @@ -130,9 +131,10 @@ impl DB { value_type TEXT NOT NULL CHECK (value_type IN ('string','number','boolean','null')), value TEXT, PRIMARY KEY (pk, key), - FOREIGN KEY (pk) REFERENCES asset (pk) ON DELETE CASCADE - )" - , [])?; + FOREIGN KEY (pk) REFERENCES asset (pk) ON DELETE CASCADE ON UPDATE CASCADE + )", + [], + )?; // CrĂ©er un index sur la collection pour les requĂŞtes rapides conn.execute( @@ -856,11 +858,9 @@ impl DB { /// * `real_pk` - Le real PK calculĂ© après tĂ©lĂ©chargement pub fn update_lazy_to_downloaded(&self, lazy_pk: &str, real_pk: &str) -> rusqlite::Result<()> { let mut conn = self.lock_conn("update_lazy_to_downloaded"); - let tx = conn.transaction()?; - // 1. RĂ©cupĂ©rer l'entry lazy (pk = lazy_pk tant que pas tĂ©lĂ©chargĂ©) - let (old_pk, collection, id, hits): (String, Option, Option, i32) = tx + let (current_pk, collection, id, hits): (String, Option, Option, i32) = tx .query_row( "SELECT pk, collection, id, hits FROM asset WHERE lazy_pk = ?1", [lazy_pk], @@ -869,35 +869,40 @@ impl DB { .optional()? .ok_or_else(|| Error::QueryReturnedNoRows)?; - if old_pk == real_pk { - // Rien Ă  faire si dĂ©jĂ  commutĂ© + if current_pk == real_pk { return Ok(()); } let now = Utc::now().to_rfc3339(); let hits_to_add = if hits > 0 { hits } else { 1 }; - // 2. CrĂ©er/mettre Ă  jour l'entry avec le real pk - tx.execute( - "INSERT INTO asset (pk, lazy_pk, collection, id, hits, last_used) - VALUES (?1, ?2, ?3, ?4, ?5, ?6) - ON CONFLICT(pk) DO UPDATE SET - lazy_pk = excluded.lazy_pk, - collection = COALESCE(excluded.collection, collection), - id = COALESCE(excluded.id, id), - hits = hits + excluded.hits, - last_used = excluded.last_used", - params![real_pk, lazy_pk, collection, id, hits_to_add, now], + // Supprimer d'Ă©ventuelles mĂ©tadonnĂ©es rĂ©siduelles associĂ©es au futur pk rĂ©el + // (peut arriver si un ancien tĂ©lĂ©chargement a laissĂ© des traces sans asset correspondant). + tx.execute("DELETE FROM metadata WHERE pk = ?1", [real_pk])?; + + let updated = tx.execute( + "UPDATE asset + SET pk = ?1, + lazy_pk = ?2, + collection = COALESCE(?3, collection), + id = COALESCE(?4, id), + hits = hits + ?5, + last_used = ?6 + WHERE lazy_pk = ?7", + params![ + real_pk, + lazy_pk, + collection, + id, + hits_to_add, + now, + lazy_pk + ], )?; - // 3. Re-pointer les mĂ©tadonnĂ©es vers le real pk - tx.execute( - "UPDATE metadata SET pk = ?1 WHERE pk = ?2", - params![real_pk, old_pk], - )?; - - // 4. Supprimer l'ancienne entry lazy - tx.execute("DELETE FROM asset WHERE pk = ?1", [old_pk])?; + if updated == 0 { + return Err(Error::QueryReturnedNoRows); + } tx.commit() } @@ -952,6 +957,20 @@ impl DB { Ok(result) } + /// Retourne une entry Ă  partir d'un pk ou lazy_pk. + pub fn get_entry_by_pk_or_lazy_pk( + &self, + value: &str, + ) -> rusqlite::Result, Option, Option)>> { + let conn = self.lock_conn("get_entry_by_pk_or_lazy_pk"); + conn.query_row( + "SELECT pk, lazy_pk, collection FROM asset WHERE pk = ?1 OR lazy_pk = ?1 LIMIT 1", + [value], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .optional() + } + /// Met Ă  jour le compteur d'accès pour une entry lazy (pk = NULL) /// /// # Arguments diff --git a/pmocache/src/lazy.rs b/pmocache/src/lazy.rs new file mode 100644 index 00000000..f45214a9 --- /dev/null +++ b/pmocache/src/lazy.rs @@ -0,0 +1,42 @@ +use anyhow::Result; +use async_trait::async_trait; +use serde_json::Value; + +/// Retourne le prĂ©fixe d'un lazy PK (`PREFIX:VALUE`) +pub fn lazy_prefix_from_pk(lazy_pk: &str) -> Option<&str> { + lazy_pk.split_once(':').map(|(prefix, _)| prefix) +} + +/// DonnĂ©es optionnelles pouvant ĂŞtre fournies par un [`LazyProvider`] +#[derive(Debug, Clone, Default)] +pub struct LazyEntryRemoteData { + pub metadata: Option, + pub cover_url: Option, +} + +/// Trait gĂ©nĂ©rique dĂ©crivant un fournisseur de lazy PK. +/// +/// Chaque implĂ©mentation est responsable d'un prĂ©fixe particulier (ex: `QOBUZ`). +/// Lorsque le cache rencontre un lazy PK dont le prĂ©fixe correspond, +/// il dĂ©lègue au provider pour rĂ©soudre l'URL et rĂ©cupĂ©rer les informations +/// nĂ©cessaires (mĂ©tadonnĂ©es, couverture, etc.). +#[async_trait] +pub trait LazyProvider: Send + Sync { + /// PrĂ©fixe associĂ© (sans le `:` final). + fn lazy_prefix(&self) -> &'static str; + + /// Retourne l'URL de tĂ©lĂ©chargement actuelle pour ce lazy PK. + async fn get_url(&self, lazy_pk: &str) -> Result; + + /// MĂ©tadonnĂ©es optionnelles Ă  associer immĂ©diatement Ă  l'entrĂ©e lazy. + async fn metadata(&self, lazy_pk: &str) -> Result> { + let _ = lazy_pk; + Ok(None) + } + + /// URL de couverture Ă©ventuelle pour permettre un cache eager des jaquettes. + async fn cover_url(&self, lazy_pk: &str) -> Result> { + let _ = lazy_pk; + Ok(None) + } +} diff --git a/pmocache/src/lib.rs b/pmocache/src/lib.rs index 353a1527..57fd9c1a 100644 --- a/pmocache/src/lib.rs +++ b/pmocache/src/lib.rs @@ -115,6 +115,7 @@ pub mod cache; pub mod cache_trait; pub mod db; pub mod download; +pub mod lazy; pub mod metadata_macros; #[cfg(feature = "pmoserver")] @@ -134,6 +135,7 @@ pub use cache::{ CacheSubscription, }; pub use cache_trait::{pk_from_content_header, FileCache}; +pub use lazy::{lazy_prefix_from_pk, LazyEntryRemoteData, LazyProvider}; pub use db::{CacheEntry, DB}; pub use download::{ download, download_with_transformer, ingest_with_transformer, peek_header, peek_reader_header, diff --git a/pmocache/src/pmoserver_ext.rs b/pmocache/src/pmoserver_ext.rs index a10c8a7e..5a81f589 100644 --- a/pmocache/src/pmoserver_ext.rs +++ b/pmocache/src/pmoserver_ext.rs @@ -110,15 +110,38 @@ async fn get_file_with_param( serve_file_with_streaming(&cache, &pk, ¶m, content_type, param_generator).await } +#[cfg(feature = "pmoserver")] +async fn serve_finalized_pk( + cache: &Arc>, + pk: &str, + param: &str, + content_type: &'static str, +) -> Response { + let qualifier = param.to_string(); + let file_path = cache.get_file_path_with_qualifier(pk, param); + + if let Err(e) = cache.db.update_hit(pk) { + warn!("Error updating hit count for {}: {}", pk, e); + } + + let response = serve_complete_file(file_path, content_type).await; + + if response.status().is_success() { + cache.notify_broadcast(pk, &qualifier).await; + } + + response +} + /// Handler spĂ©cifique pour les lazy PK /// /// Gère le tĂ©lĂ©chargement on-demand des fichiers lazy : /// 1. Fast path : vĂ©rifie si dĂ©jĂ  tĂ©lĂ©chargĂ© -/// 2. RĂ©cupère l'origin_url depuis la DB +/// 2. RĂ©sout l'URL via la DB ou un provider /// 3. Lance le tĂ©lĂ©chargement et calcule le real pk /// 4. Met Ă  jour la DB (lazy → downloaded) /// 5. Broadcast l'event pour PK switching -/// 6. Redirige vers le real pk +/// 6. Sert directement le fichier tĂ©lĂ©chargĂ© #[cfg(feature = "pmoserver")] async fn serve_lazy_audio_file( cache: &Arc>, @@ -126,54 +149,20 @@ async fn serve_lazy_audio_file( param: &str, content_type: &'static str, ) -> Response { - use axum::response::Redirect; - tracing::info!("Lazy download triggered for pk: {}", lazy_pk); // 1. VĂ©rifier si dĂ©jĂ  tĂ©lĂ©chargĂ© (fast path) if let Ok(Some(real_pk)) = cache.db.get_pk_by_lazy_pk(lazy_pk) { tracing::debug!( - "Lazy PK {} already downloaded as {}, redirecting", + "Lazy PK {} already downloaded as {}, serving immediately", lazy_pk, real_pk ); - - // Construire l'URL de redirection - let redirect_url = if param == C::default_param() { - format!( - "/{}/{}/{}", - C::cache_name(), - C::cache_type(), - real_pk - ) - } else { - format!( - "/{}/{}/{}/{}", - C::cache_name(), - C::cache_type(), - real_pk, - param - ) - }; - - return Redirect::temporary(&redirect_url).into_response(); + return serve_finalized_pk(cache, &real_pk, param, content_type).await; } - // 2. RĂ©cupĂ©rer origin_url - let origin_url = match cache.db.get_origin_url(lazy_pk) { - Ok(Some(url)) => url, - Ok(None) => { - tracing::error!("Lazy PK {} has no origin_url", lazy_pk); - return (StatusCode::NOT_FOUND, "Origin URL not found").into_response(); - } - Err(e) => { - tracing::error!("Error getting origin_url for {}: {}", lazy_pk, e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Database error").into_response(); - } - }; - - // 3. Lancer download complet (cela calcule le VRAI pk basĂ© sur content[512:2048]) - let real_pk = match cache.add_from_url(&origin_url, None).await { + // 2. TĂ©lĂ©charger en rĂ©solvant l'URL via la DB ou un provider + let real_pk = match cache.download_lazy(lazy_pk, None).await { Ok(pk) => pk, Err(e) => { tracing::error!("Failed to download lazy file: {}", e); @@ -185,46 +174,11 @@ async fn serve_lazy_audio_file( } }; - // 4. Mettre Ă  jour DB : lazy_pk → real_pk mapping - if let Err(e) = cache.db.update_lazy_to_downloaded(lazy_pk, &real_pk) { - tracing::error!( - "Failed to update DB for lazy transition {} → {}: {}", - lazy_pk, - real_pk, - e - ); - // Continuer quand mĂŞme, le fichier est tĂ©lĂ©chargĂ© - } - - // 5. Broadcast event pour prefetch ET commutation PK + // 4. Broadcast event pour prefetch ET commutation PK cache.broadcast_lazy_downloaded(lazy_pk, &real_pk).await; - // 6. Rediriger vers la vraie URL - let redirect_url = if param == C::default_param() { - format!( - "/{}/{}/{}", - C::cache_name(), - C::cache_type(), - real_pk - ) - } else { - format!( - "/{}/{}/{}/{}", - C::cache_name(), - C::cache_type(), - real_pk, - param - ) - }; - - tracing::debug!( - "Lazy PK {} downloaded as {}, redirecting to {}", - lazy_pk, - real_pk, - redirect_url - ); - - Redirect::temporary(&redirect_url).into_response() + // 5. Servir directement le fichier tĂ©lĂ©chargĂ© + serve_finalized_pk(cache, &real_pk, param, content_type).await } /// Fonction utilitaire pour servir un fichier avec streaming progressif diff --git a/pmoqobuz/Cargo.toml b/pmoqobuz/Cargo.toml index ca23aafa..6cfeec7d 100644 --- a/pmoqobuz/Cargo.toml +++ b/pmoqobuz/Cargo.toml @@ -63,15 +63,17 @@ pmosource = { path = "../pmosource" } # Playlist management pmoplaylist = { path = "../pmoplaylist" } +pmocache = { path = "../pmocache" } [features] -default = [] +default = ["async-trait-support"] # Feature pour activer les extensions pmoserver pmoserver = ["dep:pmoserver", "dep:axum", "dep:utoipa"] # Feature pour activer le support serveur (cache registry) server = ["pmosource/server"] # Feature cache (deprecated - toujours actif maintenant) cache = [] +async-trait-support = ["dep:async-trait"] disk-cache = ["dep:rusqlite", "dep:async-trait"] [dev-dependencies] @@ -81,7 +83,6 @@ mockito = "1.0" tempfile = "3.0" # Pour les exemples tracing-subscriber = "0.3" -pmocache = { path = "../pmocache" } # Pour l'exemple spoofer # Specify that the with_cache example requires the cache feature diff --git a/pmoqobuz/src/lazy_provider.rs b/pmoqobuz/src/lazy_provider.rs new file mode 100644 index 00000000..c69f490a --- /dev/null +++ b/pmoqobuz/src/lazy_provider.rs @@ -0,0 +1,91 @@ +use crate::client::QobuzClient; +use crate::models::Track; +use anyhow::{anyhow, Result}; +use pmoaudiocache::AudioMetadata; +use pmocache::lazy::LazyProvider; +use serde_json::Value; +use std::sync::Arc; + +pub struct QobuzLazyProvider { + client: Arc, +} + +impl QobuzLazyProvider { + pub fn new(client: Arc) -> Self { + Self { client } + } + + fn track_id_from_lazy<'a>(&self, lazy_pk: &'a str) -> Result<&'a str> { + match lazy_pk.split_once(':') { + Some((prefix, value)) if prefix.eq_ignore_ascii_case("QOBUZ") => Ok(value), + _ => Err(anyhow!("Invalid Qobuz lazy pk {}", lazy_pk)), + } + } + + async fn fetch_track(&self, lazy_pk: &str) -> Result { + let track_id = self.track_id_from_lazy(lazy_pk)?; + self.client + .get_track(track_id) + .await + .map_err(|e| anyhow!("Failed to fetch track {}: {}", track_id, e)) + } + + fn build_metadata(track: &Track) -> AudioMetadata { + AudioMetadata { + title: Some(track.title.clone()), + artist: track.performer.as_ref().map(|p| p.name.clone()), + album: track.album.as_ref().map(|a| a.title.clone()), + duration_secs: Some(track.duration as u64), + year: track.album.as_ref().and_then(|a| { + a.release_date + .as_ref() + .and_then(|d| d.split('-').next()?.parse().ok()) + }), + track_number: Some(track.track_number), + track_total: track.album.as_ref().and_then(|a| a.tracks_count), + disc_number: Some(track.media_number), + disc_total: None, + genre: track.album.as_ref().and_then(|a| { + if !a.genres.is_empty() { + Some(a.genres.join(", ")) + } else { + None + } + }), + sample_rate: track.sample_rate, + channels: track.channels, + bitrate: None, + conversion: None, + } + } +} + +#[async_trait::async_trait] +impl LazyProvider for QobuzLazyProvider { + fn lazy_prefix(&self) -> &'static str { + "QOBUZ" + } + + async fn get_url(&self, lazy_pk: &str) -> Result { + let track_id = self.track_id_from_lazy(lazy_pk)?; + self.client + .get_stream_url(track_id) + .await + .map_err(|e| anyhow!("Failed to resolve stream URL for {}: {}", track_id, e)) + } + + async fn metadata(&self, lazy_pk: &str) -> Result> { + let track = self.fetch_track(lazy_pk).await?; + let metadata = Self::build_metadata(&track); + let value = serde_json::to_value(metadata)?; + Ok(Some(value)) + } + + async fn cover_url(&self, lazy_pk: &str) -> Result> { + let track = self.fetch_track(lazy_pk).await?; + Ok(track + .album + .and_then(|a| a.image) + .and_then(|url| if url.is_empty() { None } else { Some(url) })) + } +} diff --git a/pmoqobuz/src/lib.rs b/pmoqobuz/src/lib.rs index b2d66b06..7856d9a2 100644 --- a/pmoqobuz/src/lib.rs +++ b/pmoqobuz/src/lib.rs @@ -214,6 +214,7 @@ pub mod didl; #[cfg(feature = "disk-cache")] pub mod disk_cache; pub mod error; +mod lazy_provider; pub mod models; pub mod source; diff --git a/pmoqobuz/src/source.rs b/pmoqobuz/src/source.rs index 5b5f28ed..c93ee71b 100644 --- a/pmoqobuz/src/source.rs +++ b/pmoqobuz/src/source.rs @@ -5,6 +5,7 @@ use crate::client::QobuzClient; use crate::didl::ToDIDL; +use crate::lazy_provider::QobuzLazyProvider; use crate::models::Track; use pmoaudiocache::{AudioMetadata, Cache as AudioCache}; use pmocovers::Cache as CoverCache; @@ -71,7 +72,7 @@ pub struct QobuzSource { struct QobuzSourceInner { /// Qobuz API client - client: QobuzClient, + client: Arc, /// Cache manager (centralisĂ©) cache_manager: SourceCacheManager, @@ -103,6 +104,8 @@ impl QobuzSource { #[cfg(feature = "server")] pub fn from_registry(client: QobuzClient) -> Result { let cache_manager = SourceCacheManager::from_registry("qobuz".to_string())?; + let client = Arc::new(client); + cache_manager.register_lazy_provider(Arc::new(QobuzLazyProvider::new(client.clone()))); Ok(Self { inner: Arc::new(QobuzSourceInner { @@ -127,6 +130,8 @@ impl QobuzSource { audio_cache: Arc, ) -> Self { let cache_manager = SourceCacheManager::new("qobuz".to_string(), cover_cache, audio_cache); + let client = Arc::new(client); + cache_manager.register_lazy_provider(Arc::new(QobuzLazyProvider::new(client.clone()))); Self { inner: Arc::new(QobuzSourceInner { @@ -243,7 +248,9 @@ impl QobuzSource { pub async fn add_track_lazy(&self, track: &Track) -> Result<(String, String)> { let track_id = format!("qobuz://track/{}", track.id); - // Get streaming URL + let lazy_pk = format!("QOBUZ:{}", track.id); + + // Get streaming URL (for metadata fallback) let stream_url = self .inner .client @@ -290,15 +297,19 @@ impl QobuzSource { conversion: None, }; - // 3. Cache audio LAZILY (KEY CHANGE: use cache_audio_lazy) + // 3. Cache audio LAZILY avec un provider let cached_audio_pk = self .inner .cache_manager - .cache_audio_lazy(&stream_url, Some(metadata)) + .cache_audio_lazy_with_provider( + &lazy_pk, + Some(metadata.clone()), + cached_cover_pk.clone(), + ) .await .map_err(|e| { MusicSourceError::CacheError(format!( - "Failed to cache lazy track {}: {}", + "Failed to register lazy track {}: {}", track.title, e )) })?; diff --git a/pmoqobuz/test_python_qobuz/__pycache__/raw.cpython-312.pyc b/pmoqobuz/test_python_qobuz/__pycache__/raw.cpython-312.pyc index 0a54bd7d8bcea387182219fd051f703dc80f8384..d139e57f2a3fce6cf09d7e906084cff88558ed06 100644 GIT binary patch delta 22 ccmZ3|#JHr1k^3|+FBbz4*k7{W$Svsz07vl!uK)l5 delta 22 ccmZ3|#JHr1k^3|+FBbz4IRCQS$Svsz07&=++5i9m diff --git a/pmoqobuz/test_python_qobuz/test_getfileurl.py b/pmoqobuz/test_python_qobuz/test_getfileurl.py index fdf33a8e..c1b246d0 100755 --- a/pmoqobuz/test_python_qobuz/test_getfileurl.py +++ b/pmoqobuz/test_python_qobuz/test_getfileurl.py @@ -52,7 +52,7 @@ def main(): mime_type = file_url_data.get('mime_type', '') print(f" âś“ Success!") - print(f" URL: {url[:80]}...") + print(f" URL: {url}...") print(f" MIME type: {mime_type}") print("\n=== Test completed successfully! ===") diff --git a/pmosource/Cargo.toml b/pmosource/Cargo.toml index 5baf3e7d..b6f7fc6c 100644 --- a/pmosource/Cargo.toml +++ b/pmosource/Cargo.toml @@ -29,6 +29,7 @@ pmoplaylist = { path = "../pmoplaylist" } # Optional cache integrations pmoaudiocache = { path = "../pmoaudiocache", optional = true } pmocovers = { path = "../pmocovers", optional = true } +pmocache = { path = "../pmocache", optional = true } # Serialization (required for cache metadata) serde = { version = "1.0", features = ["derive"] } @@ -49,5 +50,5 @@ futures = { version = "0.3", optional = true } [features] default = ["cache"] -cache = ["pmoaudiocache", "pmocovers"] +cache = ["pmoaudiocache", "pmocovers", "pmocache"] server = ["pmoserver", "pmoconfig", "pmoupnp", "axum", "utoipa", "tracing", "lazy_static", "tokio-stream", "futures"] diff --git a/pmosource/src/cache.rs b/pmosource/src/cache.rs index 1a008d44..426fff56 100755 --- a/pmosource/src/cache.rs +++ b/pmosource/src/cache.rs @@ -20,6 +20,7 @@ use crate::{CacheStatus, MusicSourceError, Result}; use pmoaudiocache::{AudioMetadata, Cache as AudioCache}; +use pmocache::lazy::LazyProvider; use pmocovers::Cache as CoverCache; use serde_json::{json, Value as JsonValue}; use std::collections::HashMap; @@ -130,6 +131,11 @@ impl SourceCacheManager { } } + /// Enregistre un provider lazy supplĂ©mentaire auprès du cache audio. + pub fn register_lazy_provider(&self, provider: Arc) { + self.audio_cache.register_lazy_provider(provider); + } + /// RĂ©soudre l'URI d'une piste (prioritĂ© au cache) /// /// Retourne l'URI du fichier audio en cache si disponible, @@ -259,21 +265,79 @@ impl SourceCacheManager { .await .map_err(|e| MusicSourceError::CacheError(e.to_string()))?; - // Store metadata in lazy_pk metadata table if provided + // Seed individual metadata entries directly (no redundant JSON blob) if let Some(meta) = metadata { - // Serialize metadata to JSON and store - if let Ok(json) = serde_json::to_value(&meta) { - let _ = self - .audio_cache - .db - .set_a_metadata_by_key(&lazy_pk, "audio_metadata", json); - } self.seed_audio_metadata(&lazy_pk, &meta); } Ok(lazy_pk) } + /// CrĂ©e une entrĂ©e lazy basĂ©e sur un provider enregistrĂ©. + /// + /// # Arguments + /// + /// * `lazy_pk` - Identifiant logique (prĂ©fixe provider + valeur) + /// * `metadata` - MĂ©tadonnĂ©es optionnelles connues Ă  l'avance + /// * `cover_pk_hint` - PK Ă©ventuel d'une cover dĂ©jĂ  cachĂ©e + pub async fn cache_audio_lazy_with_provider( + &self, + lazy_pk: &str, + metadata: Option, + cover_pk_hint: Option, + ) -> Result { + self.audio_cache + .ensure_lazy_entry(lazy_pk, Some(&self.collection_id), None) + .await + .map_err(|e| MusicSourceError::CacheError(e.to_string()))?; + + let needs_provider_metadata = metadata.is_none(); + let needs_provider_cover = cover_pk_hint.is_none(); + + let provider_data = if needs_provider_metadata || needs_provider_cover { + Some( + self.audio_cache + .fetch_lazy_provider_data(lazy_pk) + .await + .map_err(|e| MusicSourceError::CacheError(e.to_string()))?, + ) + } else { + None + }; + + let mut final_metadata = metadata; + if final_metadata.is_none() { + if let Some(data) = provider_data.as_ref() { + if let Some(value) = data.metadata.as_ref() { + if let Ok(meta) = serde_json::from_value::(value.clone()) { + final_metadata = Some(meta); + } + } + } + } + + let mut final_cover_pk = cover_pk_hint; + if final_cover_pk.is_none() { + if let Some(data) = provider_data.as_ref() { + if let Some(url) = data.cover_url.as_ref() { + if let Ok(pk) = self.cache_cover(url).await { + final_cover_pk = Some(pk); + } + } + } + } + + if let Some(meta) = final_metadata.as_ref() { + self.seed_audio_metadata(lazy_pk, meta); + } + + if let Some(cover_pk) = final_cover_pk { + let _ = self.set_audio_metadata(lazy_pk, "cover_pk", json!(cover_pk)); + } + + Ok(lazy_pk.to_string()) + } + /// Cache un flux audio via un reader asynchrone pub async fn cache_audio_from_reader( &self, @@ -323,7 +387,7 @@ impl SourceCacheManager { /// PrĂ©-remplit les mĂ©tadonnĂ©es audio pour une entrĂ©e lazy fn seed_audio_metadata(&self, audio_pk: &str, metadata: &AudioMetadata) { let audio_pk = audio_pk.to_string(); - let mut store = |key: &str, value: JsonValue| { + let store = |key: &str, value: JsonValue| { if let Err(e) = self.audio_cache.db.set_a_metadata(&audio_pk, key, value) { log_metadata_warning(key, &audio_pk, &e.to_string()); }