eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Corrections mineurs des systèmes de caches

Browse Source
This commit is contained in:
Eric Coissac
2025-12-17 15:15:10 +01:00
parent 7de0008cc8
commit 0264e0c2e1
13 changed files with 371 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -3371,6 +3371,7 @@ dependencies = [
"pmoserver",
"reqwest",
"serde",
"serde_json",
"tempfile",
"tokio",
"tracing",

13
pmoaudiocache/src/cache.rs
View File

@@ -111,18 +111,7 @@ async fn persist_transform_streaminfo(cache: Arc<Cache>, pk: &str, tmeta: &Trans
/// ```
pub async fn new_cache_with_consolidation(dir: &str, limit: usize) -> Result<Arc<Cache>> {
let cache = Arc::new(new_cache(dir, limit)?);
// Lancer la consolidation en arrière-plan pour nettoyer les fichiers incomplets
let cache_clone = cache.clone();
tokio::spawn(async move {
if let Err(e) = cache_clone.consolidate().await {
tracing::warn!("Failed to consolidate cache on startup: {}", e);
} else {
tracing::info!("Cache consolidated successfully on startup");
}
});
Ok(cache)
Ok(pmocache::Cache::with_consolidation(cache).await)
}
/// Ajoute un fichier audio local. Les FLAC sont référencés sans copie, les autres formats

8
pmoaudiocache/src/lib.rs
View File

@@ -83,9 +83,15 @@
pub mod cache;
pub mod metadata;
pub mod metadata_ext;
pub mod streaming;
pub mod track_metadata;
/// Module public pour la création de transformers FLAC streaming
///
/// Ce module expose les fonctionnalités de conversion FLAC progressive
/// pour permettre aux utilisateurs de créer des transformers custom ou
/// de réutiliser les implémentations par défaut.
pub mod streaming;
#[cfg(feature = "pmoserver")]
pub mod api;

3
pmoaudiocache/src/openapi.rs
View File

@@ -7,6 +7,9 @@ use utoipa::OpenApi;
/// L'API réutilise les handlers génériques de pmocache.
#[derive(OpenApi)]
#[openapi(
paths(
crate::api::get_cover_url,
),
components(
schemas(
pmocache::CacheEntry,

14
pmocache/src/api.rs
View File

@@ -115,7 +115,7 @@ pub struct ErrorResponse {
/// Liste tous les items en cache avec leurs statistiques
///
/// Retourne la liste complète des entrées du cache triées par nombre d'accès décroissant.
pub async fn list_items<C: CacheConfig>(State(cache): State<Arc<Cache<C>>>) -> impl IntoResponse {
pub async fn list_items<C: CacheConfig + 'static>(State(cache): State<Arc<Cache<C>>>) -> impl IntoResponse {
match cache.db.get_all(true) {
Ok(entries) => (StatusCode::OK, Json(entries)).into_response(),
Err(e) => (
@@ -132,7 +132,7 @@ pub async fn list_items<C: CacheConfig>(State(cache): State<Arc<Cache<C>>>) -> i
/// Récupère les informations d'un item spécifique
///
/// Retourne les métadonnées d'un item identifié par sa clé (pk).
pub async fn get_item_info<C: CacheConfig>(
pub async fn get_item_info<C: CacheConfig + 'static>(
State(cache): State<Arc<Cache<C>>>,
Path(pk): Path<String>,
) -> impl IntoResponse {
@@ -153,7 +153,7 @@ pub async fn get_item_info<C: CacheConfig>(
///
/// Retourne le statut actuel du téléchargement (progression, tailles, erreurs).
/// Si le téléchargement est terminé, retourne les informations du fichier.
pub async fn get_download_status<C: CacheConfig>(
pub async fn get_download_status<C: CacheConfig + 'static>(
State(cache): State<Arc<Cache<C>>>,
Path(pk): Path<String>,
) -> impl IntoResponse {
@@ -265,7 +265,7 @@ enum AddSource<'a> {
///
/// Télécharge l'item depuis l'URL fournie et l'ajoute au cache.
/// Si l'item existe déjà, il est mis à jour.
pub async fn add_item<C: CacheConfig>(
pub async fn add_item<C: CacheConfig + 'static>(
State(cache): State<Arc<Cache<C>>>,
Json(req): Json<AddItemRequest>,
) -> impl IntoResponse {
@@ -337,7 +337,7 @@ pub async fn add_item<C: CacheConfig>(
/// Supprime un item du cache
///
/// Supprime l'item et toutes ses variantes du disque et de la base de données.
pub async fn delete_item<C: CacheConfig>(
pub async fn delete_item<C: CacheConfig + 'static>(
State(cache): State<Arc<Cache<C>>>,
Path(pk): Path<String>,
) -> impl IntoResponse {
@@ -389,7 +389,7 @@ pub async fn delete_item<C: CacheConfig>(
/// Purge complètement le cache
///
/// Supprime tous les items et vide la base de données. Opération irréversible.
pub async fn purge_cache<C: CacheConfig>(State(cache): State<Arc<Cache<C>>>) -> impl IntoResponse {
pub async fn purge_cache<C: CacheConfig + 'static>(State(cache): State<Arc<Cache<C>>>) -> impl IntoResponse {
match cache.purge().await {
Ok(_) => (
StatusCode::OK,
@@ -413,7 +413,7 @@ pub async fn purge_cache<C: CacheConfig>(State(cache): State<Arc<Cache<C>>>) ->
///
/// Re-télécharge les items manquants et supprime les fichiers orphelins.
/// Utile pour réparer un cache corrompu.
pub async fn consolidate_cache<C: CacheConfig>(
pub async fn consolidate_cache<C: CacheConfig + 'static>(
State(cache): State<Arc<Cache<C>>>,
) -> impl IntoResponse {
match cache.consolidate().await {

56
pmocache/src/cache.rs
View File

@@ -152,7 +152,7 @@ pub struct Cache<C: CacheConfig> {
_phantom: std::marker::PhantomData<C>,
}
impl<C: CacheConfig> Cache<C> {
impl<C: CacheConfig + 'static> Cache<C> {
/// Retourne le chemin du fichier marker de complétion
fn get_completion_marker_path(&self, pk: &str) -> PathBuf {
self.get_file_path(pk)
@@ -424,6 +424,58 @@ impl<C: CacheConfig> Cache<C> {
})
}
/// Lance une consolidation en arrière-plan pour un cache existant
///
/// Cette fonction utilitaire lance une tâche asynchrone qui consolide le cache
/// (supprime les fichiers incomplets sans marker de complétion) et retourne
/// immédiatement le cache fourni en paramètre.
///
/// Idéal pour les crates spécialisées qui veulent offrir une fonction
/// `new_cache_with_consolidation` sans dupliquer la logique de lancement.
///
/// # Arguments
///
/// * `cache` - Instance du cache à consolider
///
/// # Returns
///
/// Le même `Arc<Cache<C>>` fourni en paramètre
///
/// # Exemple
///
/// ```rust,ignore
/// use pmocache::{Cache, CacheConfig};
/// use std::sync::Arc;
///
/// struct MyConfig;
/// impl CacheConfig for MyConfig {
/// fn file_extension() -> &'static str { "dat" }
/// }
///
/// async fn create_cache_with_cleanup() -> anyhow::Result<Arc<Cache<MyConfig>>> {
/// let cache = Arc::new(Cache::new("./cache", 1000)?);
/// Ok(Cache::with_consolidation(cache).await)
/// }
/// ```
pub async fn with_consolidation(cache: Arc<Cache<C>>) -> Arc<Cache<C>> {
let cache_clone = cache.clone();
tokio::spawn(async move {
if let Err(e) = cache_clone.consolidate().await {
tracing::warn!(
"Failed to consolidate {} cache on startup: {}",
C::cache_name(),
e
);
} else {
tracing::info!(
"{} cache consolidated successfully on startup",
C::cache_name()
);
}
});
cache
}
/// Configure la taille minimale de prébuffering
///
/// # Arguments
@@ -1559,7 +1611,7 @@ fn link_file(source: &Path, destination: &Path) -> std::io::Result<()> {
}
/// Implémentation du trait FileCache pour Cache
impl<C: CacheConfig> FileCache<C> for Cache<C> {
impl<C: CacheConfig + 'static> FileCache<C> for Cache<C> {
fn get_cache_dir(&self) -> &Path {
self.cache_dir()
}

4
pmocache/src/config_ext.rs
View File

@@ -91,7 +91,7 @@ pub trait CacheConfigExt {
/// let config = get_config();
/// let cache = config.create_cache::<AudioConfig>("audio_cache", "cache_audio", 500)?;
/// ```
fn create_cache<C: crate::CacheConfig>(
fn create_cache<C: crate::CacheConfig + 'static>(
&self,
cache_type: &str,
default_dir: &str,
@@ -121,7 +121,7 @@ impl CacheConfigExt for Config {
self.set_value(&["host", cache_type, "size"], Value::Number(n))
}
fn create_cache<C: crate::CacheConfig>(
fn create_cache<C: crate::CacheConfig + 'static>(
&self,
cache_type: &str,
default_dir: &str,

6
pmocache/src/pmoserver_ext.rs
View File

@@ -111,7 +111,7 @@ async fn get_file_with_param<C: CacheConfig + 'static>(
}
#[cfg(feature = "pmoserver")]
async fn serve_finalized_pk<C: CacheConfig>(
async fn serve_finalized_pk<C: CacheConfig + 'static>(
cache: &Arc<Cache<C>>,
pk: &str,
param: &str,
@@ -143,7 +143,7 @@ async fn serve_finalized_pk<C: CacheConfig>(
/// 5. Broadcast l'event pour PK switching
/// 6. Sert directement le fichier téléchargé
#[cfg(feature = "pmoserver")]
async fn serve_lazy_audio_file<C: CacheConfig>(
async fn serve_lazy_audio_file<C: CacheConfig + 'static>(
cache: &Arc<Cache<C>>,
lazy_pk: &str,
param: &str,
@@ -187,7 +187,7 @@ async fn serve_lazy_audio_file<C: CacheConfig>(
/// Sinon, le fichier complet est servi normalement.
/// Si le fichier n'existe pas et qu'un param_generator est fourni, tente de générer le param.
#[cfg(feature = "pmoserver")]
async fn serve_file_with_streaming<C: CacheConfig>(
async fn serve_file_with_streaming<C: CacheConfig + 'static>(
cache: &Arc<Cache<C>>,
pk: &str,
param: &str,

1
pmocovers/Cargo.toml
View File

@@ -17,6 +17,7 @@ reqwest = { version = "0.12", features = ["blocking"] }
# Utilitaires
anyhow = "1.0"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
# Async
tokio = { version = "1.0", features = ["full"] }

91
pmocovers/src/api.rs Normal file
View File

@@ -0,0 +1,91 @@
//! API REST handlers spécifiques au cache de couvertures
use crate::cache;
use crate::Cache;
use axum::{
extract::State,
http::StatusCode,
response::IntoResponse,
Json,
};
use pmocache::api::{AddItemRequest, AddItemResponse, ErrorResponse};
use std::sync::Arc;
#[derive(Clone, Copy)]
enum AddSource<'a> {
Url(&'a str),
Local(&'a str),
}
/// Handler spécialisé pour l'ajout d'images dans le cache de couvertures.
///
/// Supporte l'ajout depuis une URL (avec conversion WebP) ou depuis un fichier local
/// (avec conversion ou passthrough selon le format).
pub async fn add_cover_item(
State(cache): State<Arc<Cache>>,
Json(req): Json<AddItemRequest>,
) -> impl IntoResponse {
let mode = match (req.url.as_deref(), req.path.as_deref()) {
(Some(url), None) if !url.is_empty() => AddSource::Url(url),
(None, Some(path)) if !path.is_empty() => AddSource::Local(path),
(Some(_), Some(_)) => {
return (
StatusCode::BAD_REQUEST,
Json(ErrorResponse {
error: "INVALID_REQUEST".to_string(),
message: "Provide either 'url' or 'path', not both".to_string(),
}),
)
.into_response()
}
_ => {
return (
StatusCode::BAD_REQUEST,
Json(ErrorResponse {
error: "INVALID_REQUEST".to_string(),
message: "Either 'url' or 'path' must be provided".to_string(),
}),
)
.into_response()
}
};
let collection = req.collection.as_deref();
let add_result = match mode {
AddSource::Url(url) => cache.add_from_url(url, collection).await,
AddSource::Local(path) => cache::add_local_file(&cache, path, collection).await,
};
match add_result {
Ok(pk) => {
let origin =
cache
.db
.get_origin_url(&pk)
.ok()
.flatten()
.unwrap_or_else(|| match mode {
AddSource::Url(url) => url.to_string(),
AddSource::Local(path) => format!("file://{}", path),
});
(
StatusCode::CREATED,
Json(AddItemResponse {
pk,
url: origin,
message: "Image added successfully".to_string(),
}),
)
.into_response()
}
Err(e) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse {
error: "PROCESSING_ERROR".to_string(),
message: format!("Cannot add image: {}", e),
}),
)
.into_response(),
}
}

122
pmocovers/src/cache.rs
View File

@@ -91,13 +91,117 @@ pub fn new_cache(dir: &str, limit: usize) -> Result<Cache> {
/// des requêtes.
pub async fn new_cache_with_consolidation(dir: &str, limit: usize) -> Result<Arc<Cache>> {
let cache = Arc::new(new_cache(dir, limit)?);
let cache_clone = cache.clone();
tokio::spawn(async move {
if let Err(e) = cache_clone.consolidate().await {
tracing::warn!("Failed to consolidate cover cache on startup: {}", e);
} else {
tracing::info!("Cover cache consolidated successfully on startup");
}
});
Ok(cache)
Ok(pmocache::Cache::with_consolidation(cache).await)
}
/// Détecte si un buffer contient un fichier WebP
///
/// Le format WebP commence par "RIFF" (4 octets), suivi de la taille (4 octets),
/// puis "WEBP" (4 octets).
fn is_webp_header(buf: &[u8]) -> bool {
buf.len() >= 12 && &buf[0..4] == b"RIFF" && &buf[8..12] == b"WEBP"
}
/// Ajoute un fichier image local au cache
///
/// Les images WebP sont référencées sans copie (symlink/hardlink), les autres formats
/// sont convertis en WebP via le pipeline classique.
///
/// # Arguments
///
/// * `cache` - Instance du cache de couvertures
/// * `path` - Chemin vers le fichier image local
/// * `collection` - Collection optionnelle (ex: "album:xyz")
///
/// # Returns
///
/// Clé primaire (pk) de l'image ajoutée au cache
///
/// # Exemples
///
/// ```rust,no_run
/// use pmocovers::cache;
///
/// # async fn example() -> anyhow::Result<()> {
/// let cache = cache::new_cache("./covers", 1000)?;
/// let pk = cache::add_local_file(&cache, "/path/to/cover.webp", None).await?;
/// println!("Image ajoutée avec pk: {}", pk);
/// # Ok(())
/// # }
/// ```
pub async fn add_local_file(cache: &Cache, path: &str, collection: Option<&str>) -> Result<String> {
use pmocache::download::read_exact_or_eof;
use pmocache::pk_from_content_header;
use serde_json::json;
use tokio::io::AsyncSeekExt;
let canonical_path = std::fs::canonicalize(path)?;
let file_url = format!("file://{}", canonical_path.display());
let length = tokio::fs::metadata(&canonical_path)
.await
.ok()
.map(|m| m.len());
let mut reader = tokio::fs::File::open(&canonical_path).await?;
let header = read_exact_or_eof(&mut reader, 1024)
.await
.map_err(|e| anyhow::anyhow!("Failed to read header bytes: {}", e))?;
let pk_bytes = if header.len() >= 1024 {
&header[512..]
} else {
&header[..]
};
let pk = pk_from_content_header(pk_bytes);
// Si déjà en cache, incrémenter hit et retourner
if cache.db.get(&pk, false).is_ok() {
cache.db.update_hit(&pk)?;
return Ok(pk);
}
// Si téléchargement en cours, attendre et retourner
if let Some(download) = cache.get_download(&pk).await {
if download.finished().await {
cache.db.update_hit(&pk)?;
}
return Ok(pk);
}
let is_webp = is_webp_header(&header);
if !is_webp {
// Format non-WebP : passer par le pipeline de conversion
reader
.rewind()
.await
.map_err(|e| anyhow::anyhow!("Failed to rewind local file: {}", e))?;
return cache
.add_from_reader_with_pk(Some(&file_url), reader, length, collection, Some(pk))
.await;
}
// Format WebP : créer un lien sans copie (passthrough)
let mut metadata = vec![
("local_passthrough".to_string(), json!(true)),
(
"local_source_path".to_string(),
json!(canonical_path.to_string_lossy().to_string()),
),
];
if let Some(len) = length {
metadata.push(("source_size".to_string(), json!(len)));
}
cache
.register_local_file_reference(
&pk,
&canonical_path,
collection,
Some(&file_url),
Some(&metadata),
)
.await?;
Ok(pk)
}

89
pmocovers/src/lib.rs
View File

@@ -54,13 +54,16 @@
pub mod cache;
pub mod webp;
#[cfg(feature = "pmoserver")]
pub mod api;
#[cfg(feature = "pmoserver")]
pub mod openapi;
#[cfg(feature = "pmoconfig")]
pub mod config_ext;
pub use cache::{new_cache, new_cache_with_consolidation, Cache, CoversConfig};
pub use cache::{add_local_file, new_cache, new_cache_with_consolidation, Cache, CoversConfig};
#[cfg(feature = "pmoserver")]
pub use openapi::ApiDoc;
@@ -172,7 +175,34 @@ fn create_variant_generator() -> pmocache::pmoserver_ext::ParamGenerator<CoversC
// Handlers JPEG (transcodage à la volée)
// ========================================================================
/// Sert une image de couverture au format JPEG (transcodage depuis WebP)
///
/// Cette route transcode à la volée l'image WebP stockée en cache vers le format JPEG.
/// Utile pour la compatibilité avec les clients qui ne supportent pas WebP (ex: UPnP).
///
/// # Arguments
///
/// * `pk` - Clé primaire de l'image
///
/// # Responses
///
/// * `200 OK` - Image JPEG transcodée
/// * `404 NOT_FOUND` - Image non trouvée
/// * `500 INTERNAL_SERVER_ERROR` - Erreur de transcodage
#[cfg(feature = "pmoserver")]
#[utoipa::path(
get,
path = "/covers/jpeg/{pk}",
tag = "covers",
params(
("pk" = String, Path, description = "Clé primaire de l'image")
),
responses(
(status = 200, description = "Image JPEG", content_type = "image/jpeg"),
(status = 404, description = "Image non trouvée"),
(status = 500, description = "Erreur de transcodage"),
)
)]
async fn serve_cover_jpeg(
axum::extract::State(cache): axum::extract::State<Arc<Cache>>,
axum::extract::Path(pk): axum::extract::Path<String>,
@@ -180,7 +210,36 @@ async fn serve_cover_jpeg(
serve_jpeg_internal(cache, pk, None).await
}
/// Sert une image de couverture redimensionnée au format JPEG (transcodage depuis WebP)
///
/// Cette route transcode à la volée l'image WebP stockée en cache vers le format JPEG,
/// en la redimensionnant à la taille demandée (format carré).
///
/// # Arguments
///
/// * `pk` - Clé primaire de l'image
/// * `size` - Taille souhaitée en pixels (ex: 256 pour 256x256)
///
/// # Responses
///
/// * `200 OK` - Image JPEG redimensionnée et transcodée
/// * `404 NOT_FOUND` - Image non trouvée
/// * `500 INTERNAL_SERVER_ERROR` - Erreur de transcodage ou redimensionnement
#[cfg(feature = "pmoserver")]
#[utoipa::path(
get,
path = "/covers/jpeg/{pk}/{size}",
tag = "covers",
params(
("pk" = String, Path, description = "Clé primaire de l'image"),
("size" = String, Path, description = "Taille en pixels (ex: 256, 512)")
),
responses(
(status = 200, description = "Image JPEG redimensionnée", content_type = "image/jpeg"),
(status = 404, description = "Image non trouvée"),
(status = 500, description = "Erreur de transcodage"),
)
)]
async fn serve_cover_jpeg_with_size(
axum::extract::State(cache): axum::extract::State<Arc<Cache>>,
axum::extract::Path((pk, size)): axum::extract::Path<(String, String)>,
@@ -276,7 +335,7 @@ impl CoverCacheExt for pmoserver::Server {
cache_dir: &str,
limit: usize,
) -> anyhow::Result<Arc<Cache>> {
use pmocache::pmoserver_ext::{create_api_router, create_file_router_with_generator};
use pmocache::pmoserver_ext::create_file_router_with_generator;
let cache = Arc::new(cache::new_cache(cache_dir, limit)?);
@@ -302,9 +361,29 @@ impl CoverCacheExt for pmoserver::Server {
let combined_router = file_router.merge(jpeg_router);
self.add_router("/", combined_router).await;
// API REST générique (pmocache)
// Routes: GET/POST/DELETE /api/covers, etc.
let api_router = create_api_router(cache.clone());
// API REST (handlers génériques + POST spécialisé covers)
let api_router = axum::Router::new()
.route(
"/",
axum::routing::get(pmocache::api::list_items::<CoversConfig>)
.post(crate::api::add_cover_item)
.delete(pmocache::api::purge_cache::<CoversConfig>),
)
.route(
"/{pk}",
axum::routing::get(pmocache::api::get_item_info::<CoversConfig>)
.delete(pmocache::api::delete_item::<CoversConfig>),
)
.route(
"/{pk}/status",
axum::routing::get(pmocache::api::get_download_status::<CoversConfig>),
)
.route(
"/consolidate",
axum::routing::post(pmocache::api::consolidate_cache::<CoversConfig>),
)
.with_state(cache.clone());
let openapi = crate::ApiDoc::openapi();
self.add_openapi(api_router, openapi, "covers").await;

4
pmocovers/src/openapi.rs
View File

@@ -10,6 +10,10 @@ use utoipa::OpenApi;
/// L'API réutilise les handlers génériques de pmocache.
#[derive(OpenApi)]
#[openapi(
paths(
crate::serve_cover_jpeg,
crate::serve_cover_jpeg_with_size,
),
components(
schemas(
pmocache::CacheEntry,