Lazy cache
This commit is contained in:
@@ -10,14 +10,55 @@ use crate::download::{
|
|||||||
};
|
};
|
||||||
use anyhow::{anyhow, bail, Result};
|
use anyhow::{anyhow, bail, Result};
|
||||||
use serde_json::{Number, Value};
|
use serde_json::{Number, Value};
|
||||||
|
use sha2::{Digest, Sha256};
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
use std::sync::atomic::{AtomicU64, Ordering};
|
use std::sync::atomic::{AtomicU64, Ordering};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use tokio::io::{AsyncRead, AsyncReadExt};
|
use tokio::io::{AsyncRead, AsyncReadExt};
|
||||||
use tokio::sync::RwLock;
|
use tokio::sync::{broadcast, RwLock};
|
||||||
use tracing;
|
use tracing;
|
||||||
|
|
||||||
|
// ============================================================================
|
||||||
|
// LAZY PK SUPPORT
|
||||||
|
// ============================================================================
|
||||||
|
|
||||||
|
/// Préfixe magique pour identifier les lazy PK
|
||||||
|
const LAZY_PK_PREFIX: &str = "L:";
|
||||||
|
|
||||||
|
/// Génère un lazy PK à partir d'une URL
|
||||||
|
///
|
||||||
|
/// Le lazy PK est calculable sans télécharger le fichier, ce qui permet
|
||||||
|
/// de créer des URLs UPnP stables avant tout téléchargement.
|
||||||
|
///
|
||||||
|
/// Format: "L:" + hex(sha256(url)[..16])
|
||||||
|
pub fn generate_lazy_pk(url: &str) -> String {
|
||||||
|
let mut hasher = Sha256::new();
|
||||||
|
hasher.update(url.as_bytes());
|
||||||
|
let hash = hasher.finalize();
|
||||||
|
format!("{}{}", LAZY_PK_PREFIX, hex::encode(&hash[..16]))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Vérifie si un PK est en mode lazy
|
||||||
|
pub fn is_lazy_pk(pk: &str) -> bool {
|
||||||
|
pk.starts_with(LAZY_PK_PREFIX)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Events émis par le cache pour notifier les changements d'état
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub enum CacheEvent {
|
||||||
|
/// Un fichier a été servi via HTTP
|
||||||
|
Served {
|
||||||
|
pk: String,
|
||||||
|
format: String,
|
||||||
|
},
|
||||||
|
/// Un fichier lazy a été téléchargé et est maintenant disponible
|
||||||
|
LazyDownloaded {
|
||||||
|
lazy_pk: String,
|
||||||
|
real_pk: String,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
/// Informations transmises lors de la diffusion d'un élément du cache via HTTP.
|
/// Informations transmises lors de la diffusion d'un élément du cache via HTTP.
|
||||||
///
|
///
|
||||||
/// - Emis uniquement quand une réponse 2xx est renvoyée par les routes HTTP générées
|
/// - Emis uniquement quand une réponse 2xx est renvoyée par les routes HTTP générées
|
||||||
@@ -99,6 +140,8 @@ pub struct Cache<C: CacheConfig> {
|
|||||||
transformer_factory: Option<Arc<dyn Fn() -> StreamTransformer + Send + Sync>>,
|
transformer_factory: Option<Arc<dyn Fn() -> StreamTransformer + Send + Sync>>,
|
||||||
/// Taille minimale de prébuffering en octets (0 = désactivé)
|
/// Taille minimale de prébuffering en octets (0 = désactivé)
|
||||||
min_prebuffer_size: u64,
|
min_prebuffer_size: u64,
|
||||||
|
/// LAZY PK SUPPORT: Channel pour broadcaster les events (lazy downloads, etc.)
|
||||||
|
served_tx: Option<broadcast::Sender<CacheEvent>>,
|
||||||
/// Phantom data pour le type de configuration
|
/// Phantom data pour le type de configuration
|
||||||
_phantom: std::marker::PhantomData<C>,
|
_phantom: std::marker::PhantomData<C>,
|
||||||
}
|
}
|
||||||
@@ -350,6 +393,9 @@ impl<C: CacheConfig> Cache<C> {
|
|||||||
std::fs::create_dir_all(&directory)?;
|
std::fs::create_dir_all(&directory)?;
|
||||||
let db = DB::init(&directory.join("cache.db"))?;
|
let db = DB::init(&directory.join("cache.db"))?;
|
||||||
|
|
||||||
|
// Créer un channel pour les events (capacité de 100 events en buffer)
|
||||||
|
let (served_tx, _) = broadcast::channel(100);
|
||||||
|
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
dir: directory,
|
dir: directory,
|
||||||
limit,
|
limit,
|
||||||
@@ -359,6 +405,7 @@ impl<C: CacheConfig> Cache<C> {
|
|||||||
subscriber_counter: AtomicU64::new(1),
|
subscriber_counter: AtomicU64::new(1),
|
||||||
transformer_factory,
|
transformer_factory,
|
||||||
min_prebuffer_size: DEFAULT_PREBUFFER_SIZE,
|
min_prebuffer_size: DEFAULT_PREBUFFER_SIZE,
|
||||||
|
served_tx: Some(served_tx),
|
||||||
_phantom: std::marker::PhantomData,
|
_phantom: std::marker::PhantomData,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -601,7 +648,7 @@ impl<C: CacheConfig> Cache<C> {
|
|||||||
pub async fn add_from_reader<R>(
|
pub async fn add_from_reader<R>(
|
||||||
&self,
|
&self,
|
||||||
source_uri: Option<&str>,
|
source_uri: Option<&str>,
|
||||||
mut reader: R,
|
reader: R,
|
||||||
length: Option<u64>,
|
length: Option<u64>,
|
||||||
collection: Option<&str>,
|
collection: Option<&str>,
|
||||||
) -> Result<String>
|
) -> Result<String>
|
||||||
@@ -1207,6 +1254,122 @@ impl<C: CacheConfig> Cache<C> {
|
|||||||
|
|
||||||
Ok(removed)
|
Ok(removed)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ============================================================================
|
||||||
|
// LAZY PK SUPPORT - Methods
|
||||||
|
// ============================================================================
|
||||||
|
|
||||||
|
/// S'abonne aux events du cache (lazy downloads, etc.)
|
||||||
|
///
|
||||||
|
/// Retourne un receiver pour écouter les events. Chaque abonné reçoit
|
||||||
|
/// une copie indépendante des events.
|
||||||
|
///
|
||||||
|
/// # Example
|
||||||
|
///
|
||||||
|
/// ```rust,no_run
|
||||||
|
/// let cache = Cache::<AudioConfig>::new("./cache", 1000)?;
|
||||||
|
/// let mut rx = cache.subscribe_events();
|
||||||
|
///
|
||||||
|
/// tokio::spawn(async move {
|
||||||
|
/// while let Ok(event) = rx.recv().await {
|
||||||
|
/// match event {
|
||||||
|
/// CacheEvent::LazyDownloaded { lazy_pk, real_pk } => {
|
||||||
|
/// println!("Lazy {} → Real {}", lazy_pk, real_pk);
|
||||||
|
/// }
|
||||||
|
/// _ => {}
|
||||||
|
/// }
|
||||||
|
/// }
|
||||||
|
/// });
|
||||||
|
/// ```
|
||||||
|
pub fn subscribe_events(&self) -> broadcast::Receiver<CacheEvent> {
|
||||||
|
self.served_tx
|
||||||
|
.as_ref()
|
||||||
|
.expect("Cache event channel not initialized")
|
||||||
|
.subscribe()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Broadcast un event quand un lazy PK est téléchargé
|
||||||
|
///
|
||||||
|
/// Cette méthode est appelée après qu'un fichier lazy a été téléchargé
|
||||||
|
/// et son real pk calculé. Elle permet aux playlists de commuter leurs PK.
|
||||||
|
pub async fn broadcast_lazy_downloaded(&self, lazy_pk: &str, real_pk: &str) {
|
||||||
|
if let Some(tx) = &self.served_tx {
|
||||||
|
let event = CacheEvent::LazyDownloaded {
|
||||||
|
lazy_pk: lazy_pk.to_string(),
|
||||||
|
real_pk: real_pk.to_string(),
|
||||||
|
};
|
||||||
|
// Ignorer l'erreur si pas d'abonnés
|
||||||
|
let _ = tx.send(event);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 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::<AudioConfig>::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
|
||||||
|
/// ```
|
||||||
|
pub async fn add_from_url_deferred(
|
||||||
|
&self,
|
||||||
|
url: &str,
|
||||||
|
collection: Option<&str>,
|
||||||
|
) -> Result<String> {
|
||||||
|
// 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...)
|
||||||
|
if let Ok(Some(_)) = self.db.get_pk_by_lazy_pk(&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)?;
|
||||||
|
|
||||||
|
tracing::debug!("Created new lazy pk {} for URL {}", lazy_pk, url);
|
||||||
|
|
||||||
|
Ok(lazy_pk)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Implémentation du trait FileCache pour Cache
|
/// Implémentation du trait FileCache pour Cache
|
||||||
|
|||||||
@@ -156,6 +156,29 @@ impl DB {
|
|||||||
[],
|
[],
|
||||||
)?;
|
)?;
|
||||||
|
|
||||||
|
// LAZY PK SUPPORT : Ajouter colonne lazy_pk pour mode deferred
|
||||||
|
// Cette colonne permet de stocker un PK temporaire calculé à partir de l'URL
|
||||||
|
// sans télécharger le fichier. Quand le fichier est téléchargé, le real pk
|
||||||
|
// est calculé et stocké, mais le lazy_pk est conservé pour compatibilité UPnP.
|
||||||
|
conn.execute(
|
||||||
|
"ALTER TABLE asset ADD COLUMN IF NOT EXISTS lazy_pk TEXT",
|
||||||
|
[],
|
||||||
|
)?;
|
||||||
|
|
||||||
|
// Index sur lazy_pk pour lookups rapides (lazy_pk → real pk)
|
||||||
|
conn.execute(
|
||||||
|
"CREATE INDEX IF NOT EXISTS idx_asset_lazy_pk ON asset (lazy_pk)",
|
||||||
|
[],
|
||||||
|
)?;
|
||||||
|
|
||||||
|
// Index unique sur lazy_pk (non-NULL) pour éviter les doublons
|
||||||
|
// Un lazy_pk ne peut pointer que vers un seul entry
|
||||||
|
conn.execute(
|
||||||
|
"CREATE UNIQUE INDEX IF NOT EXISTS idx_asset_lazy_pk_unique
|
||||||
|
ON asset (lazy_pk) WHERE lazy_pk IS NOT NULL",
|
||||||
|
[],
|
||||||
|
)?;
|
||||||
|
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
conn: Mutex::new(conn),
|
conn: Mutex::new(conn),
|
||||||
})
|
})
|
||||||
@@ -745,6 +768,219 @@ impl DB {
|
|||||||
|
|
||||||
Ok(entries)
|
Ok(entries)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ============================================================================
|
||||||
|
// LAZY PK SUPPORT
|
||||||
|
// ============================================================================
|
||||||
|
|
||||||
|
/// Ajoute une entrée en mode lazy (pk = NULL, lazy_pk rempli)
|
||||||
|
///
|
||||||
|
/// Utilisé pour créer des entries sans télécharger le fichier.
|
||||||
|
/// Le lazy_pk est calculé à partir de l'URL, le real pk sera calculé
|
||||||
|
/// lors du téléchargement effectif.
|
||||||
|
///
|
||||||
|
/// # Arguments
|
||||||
|
///
|
||||||
|
/// * `lazy_pk` - PK temporaire (format "L:" + hash(url))
|
||||||
|
/// * `id` - Identifiant optionnel
|
||||||
|
/// * `collection` - Collection optionnelle
|
||||||
|
pub fn add_lazy(
|
||||||
|
&self,
|
||||||
|
lazy_pk: &str,
|
||||||
|
id: Option<&str>,
|
||||||
|
collection: Option<&str>,
|
||||||
|
) -> rusqlite::Result<()> {
|
||||||
|
let conn = self.lock_conn("add_lazy");
|
||||||
|
|
||||||
|
// Créer une entry avec pk = NULL et lazy_pk rempli
|
||||||
|
// On utilise une astuce: insérer avec lazy_pk comme clé temporaire
|
||||||
|
// puis mettre pk à NULL
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO asset (pk, lazy_pk, id, collection, hits, last_used)
|
||||||
|
VALUES (NULL, ?1, ?2, ?3, 0, ?4)",
|
||||||
|
params![lazy_pk, id, collection, Utc::now().to_rfc3339()],
|
||||||
|
)?;
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Récupère le real pk associé à un lazy_pk
|
||||||
|
///
|
||||||
|
/// # Arguments
|
||||||
|
///
|
||||||
|
/// * `lazy_pk` - Le lazy PK à rechercher
|
||||||
|
///
|
||||||
|
/// # Returns
|
||||||
|
///
|
||||||
|
/// * `Ok(Some(pk))` - Le real pk si le fichier a été téléchargé
|
||||||
|
/// * `Ok(None)` - Pas encore téléchargé (pk = NULL) ou lazy_pk inconnu
|
||||||
|
pub fn get_pk_by_lazy_pk(&self, lazy_pk: &str) -> rusqlite::Result<Option<String>> {
|
||||||
|
let conn = self.lock_conn("get_pk_by_lazy_pk");
|
||||||
|
|
||||||
|
conn.query_row(
|
||||||
|
"SELECT pk FROM asset WHERE lazy_pk = ?1",
|
||||||
|
[lazy_pk],
|
||||||
|
|row| row.get(0),
|
||||||
|
)
|
||||||
|
.optional()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Transition d'une entry lazy vers downloaded (ajoute le real pk)
|
||||||
|
///
|
||||||
|
/// Cette méthode est appelée après le téléchargement d'un fichier lazy.
|
||||||
|
/// Elle crée une nouvelle entry avec le real pk ET garde le lazy_pk
|
||||||
|
/// pour permettre aux Control Points de continuer à utiliser l'URL lazy.
|
||||||
|
///
|
||||||
|
/// # Arguments
|
||||||
|
///
|
||||||
|
/// * `lazy_pk` - Le lazy PK de l'entry originale
|
||||||
|
/// * `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 les infos de l'entry lazy
|
||||||
|
let (collection, id): (Option<String>, Option<String>) = tx
|
||||||
|
.query_row(
|
||||||
|
"SELECT collection, id FROM asset WHERE lazy_pk = ?1 AND pk IS NULL",
|
||||||
|
[lazy_pk],
|
||||||
|
|row| Ok((row.get(0)?, row.get(1)?)),
|
||||||
|
)
|
||||||
|
.optional()?
|
||||||
|
.ok_or_else(|| Error::QueryReturnedNoRows)?;
|
||||||
|
|
||||||
|
// 2. Supprimer l'entry lazy (pk = NULL)
|
||||||
|
tx.execute(
|
||||||
|
"DELETE FROM asset WHERE lazy_pk = ?1 AND pk IS NULL",
|
||||||
|
[lazy_pk],
|
||||||
|
)?;
|
||||||
|
|
||||||
|
// 3. Créer nouvelle entry avec pk rempli ET lazy_pk
|
||||||
|
// Si le real_pk existe déjà (téléchargé via eager mode), on met juste à jour
|
||||||
|
tx.execute(
|
||||||
|
"INSERT INTO asset (pk, lazy_pk, collection, id, hits, last_used)
|
||||||
|
VALUES (?1, ?2, ?3, ?4, 1, ?5)
|
||||||
|
ON CONFLICT(pk) DO UPDATE SET
|
||||||
|
lazy_pk = excluded.lazy_pk,
|
||||||
|
last_used = excluded.last_used,
|
||||||
|
hits = hits + 1",
|
||||||
|
params![real_pk, lazy_pk, collection, id, Utc::now().to_rfc3339()],
|
||||||
|
)?;
|
||||||
|
|
||||||
|
tx.commit()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Recherche une entry par son origin_url
|
||||||
|
///
|
||||||
|
/// Retourne (pk, lazy_pk) si trouvé. Vérifie à la fois les entries
|
||||||
|
/// eager (avec pk) et lazy (avec lazy_pk).
|
||||||
|
///
|
||||||
|
/// # Arguments
|
||||||
|
///
|
||||||
|
/// * `url` - L'URL d'origine à rechercher
|
||||||
|
///
|
||||||
|
/// # Returns
|
||||||
|
///
|
||||||
|
/// * `Ok(Some((Some(pk), Some(lazy_pk))))` - Entry téléchargée (lazy→eager)
|
||||||
|
/// * `Ok(Some((Some(pk), None)))` - Entry eager (jamais lazy)
|
||||||
|
/// * `Ok(Some((None, Some(lazy_pk))))` - Entry lazy (pas encore téléchargée)
|
||||||
|
/// * `Ok(None)` - URL inconnue
|
||||||
|
pub fn get_entry_by_url(
|
||||||
|
&self,
|
||||||
|
url: &str,
|
||||||
|
) -> rusqlite::Result<Option<(Option<String>, Option<String>)>> {
|
||||||
|
let conn = self.lock_conn("get_entry_by_url");
|
||||||
|
|
||||||
|
// Chercher via origin_url dans metadata
|
||||||
|
// On joint avec asset pour récupérer pk et lazy_pk
|
||||||
|
let result: Option<(Option<String>, Option<String>)> = conn
|
||||||
|
.query_row(
|
||||||
|
"SELECT a.pk, a.lazy_pk
|
||||||
|
FROM asset a
|
||||||
|
JOIN metadata m ON (a.pk = m.pk OR a.lazy_pk = m.pk)
|
||||||
|
WHERE m.key = 'origin_url' AND m.value = ?1
|
||||||
|
LIMIT 1",
|
||||||
|
[url],
|
||||||
|
|row| Ok((row.get(0)?, row.get(1)?)),
|
||||||
|
)
|
||||||
|
.optional()?;
|
||||||
|
|
||||||
|
Ok(result)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Met à jour le compteur d'accès pour une entry lazy (pk = NULL)
|
||||||
|
///
|
||||||
|
/// # Arguments
|
||||||
|
///
|
||||||
|
/// * `lazy_pk` - Le lazy PK de l'entry
|
||||||
|
pub fn update_hit_by_lazy_pk(&self, lazy_pk: &str) -> rusqlite::Result<()> {
|
||||||
|
let conn = self.lock_conn("update_hit_by_lazy_pk");
|
||||||
|
|
||||||
|
conn.execute(
|
||||||
|
"UPDATE asset
|
||||||
|
SET hits = hits + 1, last_used = ?1
|
||||||
|
WHERE lazy_pk = ?2",
|
||||||
|
params![Utc::now().to_rfc3339(), lazy_pk],
|
||||||
|
)?;
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Enregistre l'URL d'origine pour une entry lazy
|
||||||
|
///
|
||||||
|
/// Contrairement à `set_origin_url()` qui utilise le pk, cette méthode
|
||||||
|
/// utilise le lazy_pk comme clé dans la table metadata (car pk = NULL).
|
||||||
|
///
|
||||||
|
/// # Arguments
|
||||||
|
///
|
||||||
|
/// * `lazy_pk` - Le lazy PK de l'entry
|
||||||
|
/// * `origin_url` - L'URL d'origine à stocker
|
||||||
|
pub fn set_origin_url_for_lazy(
|
||||||
|
&self,
|
||||||
|
lazy_pk: &str,
|
||||||
|
origin_url: &str,
|
||||||
|
) -> rusqlite::Result<()> {
|
||||||
|
// Pour les entries lazy, on stocke l'origin_url avec lazy_pk comme clé
|
||||||
|
// dans la table metadata (au lieu de pk qui est NULL)
|
||||||
|
self.set_a_metadata_by_key(lazy_pk, "origin_url", Value::String(origin_url.to_owned()))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Version générique de set_a_metadata qui accepte une clé arbitraire
|
||||||
|
///
|
||||||
|
/// Utilisé en interne pour stocker des métadonnées avec lazy_pk au lieu de pk
|
||||||
|
fn set_a_metadata_by_key(
|
||||||
|
&self,
|
||||||
|
key: &str,
|
||||||
|
metadata_key: &str,
|
||||||
|
value: Value,
|
||||||
|
) -> rusqlite::Result<()> {
|
||||||
|
let (value_type, value_text): (&str, Option<String>) = match value {
|
||||||
|
Value::Null => ("null", None),
|
||||||
|
Value::Bool(b) => ("boolean", Some(b.to_string())),
|
||||||
|
Value::Number(n) => ("number", Some(n.to_string())),
|
||||||
|
Value::String(s) => ("string", Some(s)),
|
||||||
|
Value::Array(arr) => ("string", Some(Value::Array(arr).to_string())),
|
||||||
|
Value::Object(map) => ("string", Some(Value::Object(map).to_string())),
|
||||||
|
};
|
||||||
|
|
||||||
|
let conn = self.lock_conn("set_a_metadata_by_key");
|
||||||
|
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO metadata (pk, key, value_type, value)
|
||||||
|
VALUES (?1, ?2, ?3, ?4)
|
||||||
|
ON CONFLICT(pk, key) DO UPDATE SET
|
||||||
|
value_type = excluded.value_type,
|
||||||
|
value = excluded.value",
|
||||||
|
params![key, metadata_key, value_type, value_text.as_deref()],
|
||||||
|
)?;
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Convertit une ligne de la table `metadata` en valeur JSON.
|
/// Convertit une ligne de la table `metadata` en valeur JSON.
|
||||||
|
|||||||
@@ -129,7 +129,10 @@ pub mod openapi;
|
|||||||
#[cfg(feature = "pmoconfig")]
|
#[cfg(feature = "pmoconfig")]
|
||||||
pub mod config_ext;
|
pub mod config_ext;
|
||||||
|
|
||||||
pub use cache::{Cache, CacheBroadcastEvent, CacheConfig, CacheSubscription};
|
pub use cache::{
|
||||||
|
generate_lazy_pk, is_lazy_pk, Cache, CacheBroadcastEvent, CacheConfig, CacheEvent,
|
||||||
|
CacheSubscription,
|
||||||
|
};
|
||||||
pub use cache_trait::{pk_from_content_header, FileCache};
|
pub use cache_trait::{pk_from_content_header, FileCache};
|
||||||
pub use db::{CacheEntry, DB};
|
pub use db::{CacheEntry, DB};
|
||||||
pub use download::{
|
pub use download::{
|
||||||
|
|||||||
@@ -110,6 +110,96 @@ async fn get_file_with_param<C: CacheConfig + 'static>(
|
|||||||
serve_file_with_streaming(&cache, &pk, ¶m, content_type, param_generator).await
|
serve_file_with_streaming(&cache, &pk, ¶m, content_type, param_generator).await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// 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
|
||||||
|
/// 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
|
||||||
|
#[cfg(feature = "pmoserver")]
|
||||||
|
async fn serve_lazy_audio_file<C: CacheConfig>(
|
||||||
|
cache: &Arc<Cache<C>>,
|
||||||
|
lazy_pk: &str,
|
||||||
|
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,
|
||||||
|
real_pk
|
||||||
|
);
|
||||||
|
|
||||||
|
// Construire l'URL de redirection
|
||||||
|
let redirect_url = if param == C::default_param() {
|
||||||
|
format!("/cache/{}/{}", C::cache_type(), real_pk)
|
||||||
|
} else {
|
||||||
|
format!("/cache/{}/{}/{}", C::cache_type(), real_pk, param)
|
||||||
|
};
|
||||||
|
|
||||||
|
return Redirect::temporary(&redirect_url).into_response();
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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 {
|
||||||
|
Ok(pk) => pk,
|
||||||
|
Err(e) => {
|
||||||
|
tracing::error!("Failed to download lazy file: {}", e);
|
||||||
|
return (
|
||||||
|
StatusCode::INTERNAL_SERVER_ERROR,
|
||||||
|
format!("Download failed: {}", e),
|
||||||
|
)
|
||||||
|
.into_response();
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// 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
|
||||||
|
cache.broadcast_lazy_downloaded(lazy_pk, &real_pk).await;
|
||||||
|
|
||||||
|
// 6. Rediriger vers la vraie URL
|
||||||
|
let redirect_url = if param == C::default_param() {
|
||||||
|
format!("/cache/{}/{}", C::cache_type(), real_pk)
|
||||||
|
} else {
|
||||||
|
format!("/cache/{}/{}/{}", 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()
|
||||||
|
}
|
||||||
|
|
||||||
/// Fonction utilitaire pour servir un fichier avec streaming progressif
|
/// Fonction utilitaire pour servir un fichier avec streaming progressif
|
||||||
///
|
///
|
||||||
/// Si le fichier est en cours de téléchargement, il est streamé au fur et à mesure.
|
/// Si le fichier est en cours de téléchargement, il est streamé au fur et à mesure.
|
||||||
@@ -123,6 +213,11 @@ async fn serve_file_with_streaming<C: CacheConfig>(
|
|||||||
content_type: &'static str,
|
content_type: &'static str,
|
||||||
param_generator: Option<ParamGenerator<C>>,
|
param_generator: Option<ParamGenerator<C>>,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
|
// LAZY PK SUPPORT: Détecter si c'est un lazy PK
|
||||||
|
if crate::cache::is_lazy_pk(pk) {
|
||||||
|
return serve_lazy_audio_file(cache, pk, param, content_type).await;
|
||||||
|
}
|
||||||
|
|
||||||
let file_path = cache.get_file_path_with_qualifier(pk, param);
|
let file_path = cache.get_file_path_with_qualifier(pk, param);
|
||||||
let qualifier = param.to_string();
|
let qualifier = param.to_string();
|
||||||
|
|
||||||
|
|||||||
@@ -325,6 +325,140 @@ impl WriteHandle {
|
|||||||
self.playlist.last_change().await
|
self.playlist.last_change().await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ============================================================================
|
||||||
|
// LAZY PK SUPPORT
|
||||||
|
// ============================================================================
|
||||||
|
|
||||||
|
/// Ajoute un track sans valider l'existence du fichier
|
||||||
|
///
|
||||||
|
/// À utiliser pour les lazy PK qui seront téléchargés on-demand.
|
||||||
|
/// Contrairement à `push()`, cette méthode ne vérifie pas si le fichier
|
||||||
|
/// existe dans le cache avant de l'ajouter.
|
||||||
|
///
|
||||||
|
/// # Arguments
|
||||||
|
///
|
||||||
|
/// * `cache_pk` - PK du fichier (peut être un lazy PK "L:...")
|
||||||
|
pub async fn push_lazy(&self, cache_pk: String) -> Result<()> {
|
||||||
|
if !self.playlist.is_alive() {
|
||||||
|
return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone()));
|
||||||
|
}
|
||||||
|
|
||||||
|
// PAS de validation is_valid_pk()
|
||||||
|
let record = Record::new(cache_pk);
|
||||||
|
let mut core = self.playlist.core.write().await;
|
||||||
|
core.push(record);
|
||||||
|
let snapshot = core.snapshot();
|
||||||
|
drop(core);
|
||||||
|
|
||||||
|
self.playlist.touch().await;
|
||||||
|
|
||||||
|
if self.playlist.persistent {
|
||||||
|
self.save_to_db().await?;
|
||||||
|
}
|
||||||
|
|
||||||
|
let manager = crate::manager::PlaylistManager();
|
||||||
|
manager.rebuild_track_index(&self.playlist.id, &snapshot).await;
|
||||||
|
manager.notify_playlist_changed(&self.playlist.id);
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Version batch pour ajouter plusieurs lazy PK
|
||||||
|
///
|
||||||
|
/// Plus efficace que push_lazy() en boucle car ne reconstruit
|
||||||
|
/// l'index qu'une seule fois à la fin.
|
||||||
|
///
|
||||||
|
/// # Arguments
|
||||||
|
///
|
||||||
|
/// * `cache_pks` - Liste de PKs à ajouter (peuvent être des lazy PK)
|
||||||
|
pub async fn push_lazy_batch(&self, cache_pks: Vec<String>) -> Result<()> {
|
||||||
|
if !self.playlist.is_alive() {
|
||||||
|
return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone()));
|
||||||
|
}
|
||||||
|
|
||||||
|
let records: Vec<Record> = cache_pks.into_iter().map(Record::new).collect();
|
||||||
|
|
||||||
|
let mut core = self.playlist.core.write().await;
|
||||||
|
core.push_all(records);
|
||||||
|
let snapshot = core.snapshot();
|
||||||
|
drop(core);
|
||||||
|
|
||||||
|
self.playlist.touch().await;
|
||||||
|
|
||||||
|
if self.playlist.persistent {
|
||||||
|
self.save_to_db().await?;
|
||||||
|
}
|
||||||
|
|
||||||
|
let manager = crate::manager::PlaylistManager();
|
||||||
|
manager.rebuild_track_index(&self.playlist.id, &snapshot).await;
|
||||||
|
manager.notify_playlist_changed(&self.playlist.id);
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Commute un cache_pk vers un nouveau PK
|
||||||
|
///
|
||||||
|
/// Utilisé quand un lazy PK est téléchargé et devient un real PK.
|
||||||
|
/// Met à jour tous les records qui utilisent l'ancien PK.
|
||||||
|
///
|
||||||
|
/// # Arguments
|
||||||
|
///
|
||||||
|
/// * `old_pk` - L'ancien PK (typiquement un lazy PK "L:...")
|
||||||
|
/// * `new_pk` - Le nouveau PK (real pk calculé après téléchargement)
|
||||||
|
///
|
||||||
|
/// # Example
|
||||||
|
///
|
||||||
|
/// ```rust,no_run
|
||||||
|
/// // Appelé quand un lazy PK est téléchargé
|
||||||
|
/// writer.update_cache_pk("L:abc123", "xyz789").await?;
|
||||||
|
/// ```
|
||||||
|
pub async fn update_cache_pk(&self, old_pk: &str, new_pk: &str) -> Result<()> {
|
||||||
|
if !self.playlist.is_alive() {
|
||||||
|
return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone()));
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut core = self.playlist.core.write().await;
|
||||||
|
let mut updated = false;
|
||||||
|
|
||||||
|
// Parcourir tous les records et recréer ceux qui correspondent
|
||||||
|
// Les records sont dans des Arc, donc on doit les recréer pour les modifier
|
||||||
|
for i in 0..core.tracks.len() {
|
||||||
|
if let Some(record_arc) = core.tracks.get(i) {
|
||||||
|
if record_arc.cache_pk == old_pk {
|
||||||
|
// Créer un nouveau record avec le nouveau PK
|
||||||
|
let mut new_record = (**record_arc).clone();
|
||||||
|
new_record.cache_pk = new_pk.to_string();
|
||||||
|
core.tracks[i] = Arc::new(new_record);
|
||||||
|
updated = true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let snapshot = core.snapshot();
|
||||||
|
drop(core);
|
||||||
|
|
||||||
|
if updated {
|
||||||
|
tracing::debug!(
|
||||||
|
"Updated {} -> {} in playlist {}",
|
||||||
|
old_pk,
|
||||||
|
new_pk,
|
||||||
|
self.playlist.id
|
||||||
|
);
|
||||||
|
|
||||||
|
self.playlist.touch().await;
|
||||||
|
|
||||||
|
if self.playlist.persistent {
|
||||||
|
self.save_to_db().await?;
|
||||||
|
}
|
||||||
|
|
||||||
|
let manager = crate::manager::PlaylistManager();
|
||||||
|
manager.rebuild_track_index(&self.playlist.id, &snapshot).await;
|
||||||
|
manager.notify_playlist_changed(&self.playlist.id);
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
// Helpers internes
|
// Helpers internes
|
||||||
|
|
||||||
async fn save_to_db(&self) -> Result<()> {
|
async fn save_to_db(&self) -> Result<()> {
|
||||||
|
|||||||
@@ -528,6 +528,137 @@ impl PlaylistManager {
|
|||||||
self.inner.persistence.as_ref()
|
self.inner.persistence.as_ref()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ============================================================================
|
||||||
|
// LAZY PK SUPPORT
|
||||||
|
// ============================================================================
|
||||||
|
|
||||||
|
/// Active le mode lazy pour une playlist
|
||||||
|
///
|
||||||
|
/// Configure l'écoute des events du cache pour :
|
||||||
|
/// 1. Commuter automatiquement les lazy PK vers real PK après téléchargement
|
||||||
|
/// 2. Prefetch intelligent des N tracks suivants
|
||||||
|
///
|
||||||
|
/// # Arguments
|
||||||
|
///
|
||||||
|
/// * `playlist_id` - ID de la playlist à gérer
|
||||||
|
/// * `lookahead` - Nombre de tracks à prefetch (recommandé: 3-5)
|
||||||
|
///
|
||||||
|
/// # Example
|
||||||
|
///
|
||||||
|
/// ```rust,no_run
|
||||||
|
/// let manager = PlaylistManager::get();
|
||||||
|
/// manager.enable_lazy_mode("qobuz-favorites-123", 5);
|
||||||
|
/// // → La playlist commute automatiquement lazy → real PK
|
||||||
|
/// // → Prefetch 5 tracks en avance pendant la lecture
|
||||||
|
/// ```
|
||||||
|
pub fn enable_lazy_mode(&self, playlist_id: &str, lookahead: usize) {
|
||||||
|
let playlist_id = playlist_id.to_string();
|
||||||
|
|
||||||
|
// Obtenir le cache audio
|
||||||
|
let cache = match audio_cache() {
|
||||||
|
Ok(c) => c,
|
||||||
|
Err(e) => {
|
||||||
|
tracing::error!("Cannot enable lazy mode: audio cache not available: {}", e);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// S'abonner aux events du cache
|
||||||
|
let mut rx = cache.subscribe_events();
|
||||||
|
let manager = self.clone();
|
||||||
|
|
||||||
|
tokio::spawn(async move {
|
||||||
|
tracing::info!("Lazy mode enabled for playlist {} (lookahead: {})", playlist_id, lookahead);
|
||||||
|
|
||||||
|
while let Ok(event) = rx.recv().await {
|
||||||
|
match event {
|
||||||
|
pmocache::CacheEvent::LazyDownloaded { lazy_pk, real_pk } => {
|
||||||
|
tracing::debug!("Received LazyDownloaded event: {} → {}", lazy_pk, real_pk);
|
||||||
|
|
||||||
|
// 1. Commuter le PK dans la playlist
|
||||||
|
if let Ok(writer) = manager.get_write_handle(playlist_id.clone()).await {
|
||||||
|
tracing::info!(
|
||||||
|
"Switching PK in playlist {}: {} -> {}",
|
||||||
|
playlist_id, lazy_pk, real_pk
|
||||||
|
);
|
||||||
|
if let Err(e) = writer.update_cache_pk(&lazy_pk, &real_pk).await {
|
||||||
|
tracing::error!("Failed to update PK in playlist: {}", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 2. Prefetch les tracks suivants
|
||||||
|
manager.prefetch_next_tracks(&playlist_id, &real_pk, lookahead).await;
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
tracing::warn!("Lazy mode listener stopped for playlist {}", playlist_id);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Prefetch les N tracks suivants après une position donnée
|
||||||
|
///
|
||||||
|
/// Cette méthode est appelée automatiquement par `enable_lazy_mode()`.
|
||||||
|
async fn prefetch_next_tracks(&self, playlist_id: &str, current_pk: &str, lookahead: usize) {
|
||||||
|
let playlist = {
|
||||||
|
let playlists = self.inner.playlists.read().await;
|
||||||
|
playlists.get(playlist_id).cloned()
|
||||||
|
};
|
||||||
|
|
||||||
|
let Some(playlist) = playlist else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
|
||||||
|
let core = playlist.core.read().await;
|
||||||
|
let tracks = core.snapshot();
|
||||||
|
|
||||||
|
// Trouver position actuelle
|
||||||
|
let Some(pos) = tracks.iter().position(|r| &r.cache_pk == current_pk) else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
|
||||||
|
// Prefetch N tracks suivants
|
||||||
|
let cache = match audio_cache() {
|
||||||
|
Ok(c) => c,
|
||||||
|
Err(_) => return,
|
||||||
|
};
|
||||||
|
|
||||||
|
for i in (pos + 1)..=(pos + lookahead).min(tracks.len() - 1) {
|
||||||
|
let next_pk = &tracks[i].cache_pk;
|
||||||
|
|
||||||
|
// Si lazy PK, déclencher download en background
|
||||||
|
if pmocache::is_lazy_pk(next_pk) {
|
||||||
|
tracing::debug!("Prefetching lazy track {}: {}", i, next_pk);
|
||||||
|
|
||||||
|
let cache = cache.clone();
|
||||||
|
let next_pk = next_pk.clone();
|
||||||
|
|
||||||
|
tokio::spawn(async move {
|
||||||
|
// Récupérer l'origin_url depuis la DB
|
||||||
|
let origin_url = match cache.db.get_origin_url(&next_pk) {
|
||||||
|
Ok(Some(url)) => url,
|
||||||
|
Ok(None) => {
|
||||||
|
tracing::warn!("No origin_url for lazy pk {}", next_pk);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
tracing::error!("Error getting origin_url for {}: {}", next_pk, e);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// Déclencher download (ne bloque pas)
|
||||||
|
if let Err(e) = cache.add_from_url(&origin_url, None).await {
|
||||||
|
tracing::error!("Failed to prefetch {}: {}", next_pk, e);
|
||||||
|
} else {
|
||||||
|
tracing::debug!("Prefetch completed for {}", next_pk);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Task d'<27>viction en background
|
/// Task d'<27>viction en background
|
||||||
async fn eviction_task(&self) {
|
async fn eviction_task(&self) {
|
||||||
loop {
|
loop {
|
||||||
|
|||||||
Reference in New Issue
Block a user