diff --git a/Cargo.lock b/Cargo.lock index 328f7e5b..15180057 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -257,6 +257,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5b098575ebe77cb6d14fc7f32749631a6e44edbef6b796f89b020e99ba20d425" dependencies = [ "axum-core 0.5.5", + "axum-macros", "bytes", "form_urlencoded", "futures-util", @@ -336,6 +337,17 @@ dependencies = [ "tower-service", ] +[[package]] +name = "axum-macros" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "604fde5e028fea851ce1d8570bbdc034bec850d157f7569d10f347d06808c05c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.110", +] + [[package]] name = "axum-server" version = "0.7.3" @@ -3301,19 +3313,23 @@ name = "pmoplaylist" version = "0.1.0" dependencies = [ "anyhow", + "async-stream", + "axum 0.8.7", + "chrono", "once_cell", "pmoaudiocache", "pmocache", "pmoconfig", "pmodidl", "pmometadata", - "pmoupnp", "rusqlite", "serde", "serde_json", "thiserror 1.0.69", "tokio", + "tokio-stream", "tracing", + "utoipa", ] [[package]] @@ -3412,6 +3428,7 @@ dependencies = [ "pmoconfig", "pmocovers", "pmodidl", + "pmoplaylist", "pmoserver", "pmoutils", "quick-xml 0.37.5", @@ -4799,6 +4816,7 @@ dependencies = [ "futures-core", "pin-project-lite", "tokio", + "tokio-util", ] [[package]] diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md new file mode 100644 index 00000000..9f79ccda --- /dev/null +++ b/DEPENDENCIES.md @@ -0,0 +1,72 @@ +## Diagramme des dépendances PMOMusic (crates internes) + +```mermaid +graph TD + + PMOMusic --> pmoapp + PMOMusic --> pmoaudio_ext + PMOMusic --> pmoaudiocache + PMOMusic --> pmocovers + PMOMusic --> pmoconfig + PMOMusic --> pmoserver + PMOMusic --> pmosource + PMOMusic --> pmoupnp + PMOMusic --> pmomediaserver + PMOMusic --> pmomediarenderer + PMOMusic --> pmoqobuz + PMOMusic --> pmoparadise + + pmoaudio_ext --> pmoaudio + pmoaudio_ext --> pmocovers + pmoaudio_ext --> pmocache + pmoaudio_ext --> pmoaudiocache + pmoaudio_ext --> pmometadata + pmoaudio_ext --> pmoplaylist + + pmoaudiocache --> pmocache + pmoaudiocache --> pmometadata + + pmocovers --> pmocache + + pmoplaylist --> pmocache + pmoplaylist --> pmoaudiocache + pmoplaylist --> pmometadata + pmoplaylist --> pmodidl + + pmosource --> pmoaudiocache + pmosource --> pmocovers + pmosource --> pmocache + pmosource --> pmoplaylist + pmosource --> pmodidl + pmosource --> pmoconfig + pmosource --> pmoserver + pmosource --> pmoupnp + + pmoparadise --> pmosource + pmoparadise --> pmoaudiocache + pmoparadise --> pmoplaylist + pmoparadise --> pmoserver + pmoparadise --> pmoconfig + + pmoqobuz --> pmosource + pmoqobuz --> pmoaudiocache + pmoqobuz --> pmocovers + pmoqobuz --> pmoserver + pmoqobuz --> pmoconfig + + pmomediaserver --> pmoserver + pmomediaserver --> pmosource + pmomediaserver --> pmoconfig + pmomediaserver --> pmocovers + pmomediaserver --> pmoaudiocache + pmomediaserver --> pmoplaylist + + pmoupnp --> pmoserver + pmoupnp --> pmocovers + pmoupnp --> pmoaudiocache + pmoupnp --> pmoplaylist + pmoupnp --> pmocache + pmoupnp --> pmoconfig +``` + +> Flèches = “dépend de”. Dépendances externes non représentées. Cette vue correspond aux features activées par défaut dans la workspace. diff --git a/pmoapp/webapp/src/components/RadioParadiseExplorer.vue b/pmoapp/webapp/src/components/RadioParadiseExplorer.vue index 0b57f2bc..821b08c9 100644 --- a/pmoapp/webapp/src/components/RadioParadiseExplorer.vue +++ b/pmoapp/webapp/src/components/RadioParadiseExplorer.vue @@ -113,6 +113,58 @@ + +
+
+

📡 Playlist SSE (Radio Paradise)

+
+ + {{ playlistSseConnected ? 'Connected' : 'Disconnected' }} + + +
+
+
+ Watching playlists: + {{ pid }} +
+
+ ❌ {{ playlistSseError }} +
+
+ Waiting for playlist events… +
+
+
+
+ {{ event.kind }} + {{ event.timestamp ? formatTimestamp(new Date(event.timestamp)) : '—' }} + from {{ event.source_client }} +
+
+
+ Playlist + {{ event.playlist_id }} +
+
+ Cache PK + {{ event.cache_pk }} +
+
+ Qualifier + {{ event.qualifier }} +
+
+ Received + {{ formatTimestamp(event.received_at) }} +
+
+
+
+
+
@@ -600,7 +652,7 @@ @@ -2217,6 +2348,86 @@ onUnmounted(() => { color: #00d4ff; } +.playlist-events-section { + background: #141414; + border-radius: 8px; + padding: 20px; + margin: 20px 0; + border: 1px solid #333; +} + +.status-chip { + padding: 6px 10px; + border-radius: 999px; + border: 1px solid #555; + color: #bbb; + font-size: 0.8rem; +} + +.status-chip.connected { + border-color: rgba(46, 204, 113, 0.6); + color: #2ecc71; + background: rgba(46, 204, 113, 0.1); +} + +.events-list { + display: flex; + flex-direction: column; + gap: 10px; + margin-top: 14px; +} + +.event-item { + background: rgba(0, 0, 0, 0.25); + border: 1px solid rgba(255, 255, 255, 0.05); + border-radius: 8px; + padding: 12px; +} + +.event-header { + display: flex; + flex-wrap: wrap; + gap: 10px; + align-items: center; + justify-content: space-between; + margin-bottom: 10px; +} + +.event-kind { + text-transform: uppercase; + letter-spacing: 0.05em; + color: #00d4ff; + font-weight: 700; +} + +.event-time { + color: #9aa0a6; + font-size: 0.9rem; +} + +.event-source { + color: #9aa0a6; + font-style: italic; + font-size: 0.85rem; +} + +.event-body { + display: flex; + flex-direction: column; + gap: 6px; +} + +.event-row { + display: flex; + gap: 8px; + align-items: center; + flex-wrap: wrap; +} + +.event-row .label { + min-width: 80px; +} + .metadata-content { margin-top: 12px; } diff --git a/pmoaudiocache/src/openapi.rs b/pmoaudiocache/src/openapi.rs index ef05a746..aa138f27 100644 --- a/pmoaudiocache/src/openapi.rs +++ b/pmoaudiocache/src/openapi.rs @@ -7,16 +7,6 @@ use utoipa::OpenApi; /// L'API réutilise les handlers génériques de pmocache. #[derive(OpenApi)] #[openapi( - paths( - pmocache::api::list_items::, - pmocache::api::get_item_info::, - pmocache::api::get_download_status::, - pmocache::api::add_item::, - pmocache::api::delete_item::, - pmocache::api::purge_cache::, - pmocache::api::consolidate_cache::, - crate::api::get_cover_url, - ), components( schemas( pmocache::CacheEntry, diff --git a/pmocovers/src/openapi.rs b/pmocovers/src/openapi.rs index 88ef942b..e4f1b675 100644 --- a/pmocovers/src/openapi.rs +++ b/pmocovers/src/openapi.rs @@ -10,15 +10,6 @@ use utoipa::OpenApi; /// L'API réutilise les handlers génériques de pmocache. #[derive(OpenApi)] #[openapi( - paths( - pmocache::api::list_items::, - pmocache::api::get_item_info::, - pmocache::api::get_download_status::, - pmocache::api::add_item::, - pmocache::api::delete_item::, - pmocache::api::purge_cache::, - pmocache::api::consolidate_cache::, - ), components( schemas( pmocache::CacheEntry, diff --git a/pmoparadise/src/models.rs b/pmoparadise/src/models.rs index 79d7c75a..8afd6386 100644 --- a/pmoparadise/src/models.rs +++ b/pmoparadise/src/models.rs @@ -376,11 +376,11 @@ mod tests { cover: None, rating: None, extra: HashMap::new(), - gapless_url: todo!(), - sched_time_millis: todo!(), - song_id: todo!(), - artist_id: todo!(), - cover_large: todo!(), + gapless_url: Some("http://example.com/song.flac".into()), + sched_time_millis: Some(1_700_000_000_000), + song_id: Some("song-id".into()), + artist_id: Some("artist-id".into()), + cover_large: Some("cover-large.jpg".into()), }; assert_eq!(song.end_time_ms(), 6000); diff --git a/pmoparadise/src/source.rs b/pmoparadise/src/source.rs index 29c4c19f..28f8eac3 100644 --- a/pmoparadise/src/source.rs +++ b/pmoparadise/src/source.rs @@ -126,9 +126,13 @@ impl RadioParadiseSource { for pid in ids { let weak = Arc::downgrade(self); let pid_clone = pid.clone(); - let token = mgr.register_callback(move |changed_id| { + let token = mgr.register_callback(move |event| { let pid = pid_clone.clone(); - if changed_id == pid { + if event.playlist_id == pid { + // On ne réagit qu'aux mises à jour structurelles (ajout/suppression) + if !matches!(event.kind, pmoplaylist::PlaylistEventKind::Updated) { + return; + } if let Some(strong) = weak.upgrade() { tokio::spawn(async move { strong.bump_update_counter().await; @@ -663,7 +667,7 @@ impl MusicSource for RadioParadiseSource { } } - ObjectIdType::HistoryTrack { slug, pk } => { + ObjectIdType::HistoryTrack { slug: _, pk: _ } => { // Return metadata for the history track item let item = self.get_item(object_id).await?; Ok(BrowseResult::Items(vec![item])) diff --git a/pmoplaylist/Cargo.toml b/pmoplaylist/Cargo.toml index b842ea27..0c96c3f1 100644 --- a/pmoplaylist/Cargo.toml +++ b/pmoplaylist/Cargo.toml @@ -14,9 +14,6 @@ pmometadata = { path = "../pmometadata" } # DIDL-Lite pour UPnP pmodidl = { path = "../pmodidl" } -# UPnP (pour accéder au cache audio global) -pmoupnp = { path = "../pmoupnp" } - # Configuration (optionnelle) pmoconfig = { path = "../pmoconfig", optional = true } @@ -37,7 +34,13 @@ once_cell = "1.20" # Logging tracing = "0.1" +chrono = { version = "0.4", features = ["serde"] } +axum = { version = "0.8", optional = true, features = ["macros", "json"] } +tokio-stream = { version = "0.1", optional = true, features = ["sync"] } +utoipa = { version = "5.4", optional = true, features = ["axum_extras", "chrono"] } +async-stream = { version = "0.3", optional = true } [features] default = ["pmoconfig"] pmoconfig = ["dep:pmoconfig"] +pmoserver = ["dep:axum", "dep:tokio-stream", "dep:utoipa", "dep:async-stream"] diff --git a/pmoplaylist/src/handle/read.rs b/pmoplaylist/src/handle/read.rs index a01cbc12..1ca7e85a 100644 --- a/pmoplaylist/src/handle/read.rs +++ b/pmoplaylist/src/handle/read.rs @@ -60,6 +60,7 @@ impl ReadHandle { tracing::warn!("Cache entry {} missing, removing from playlist", cache_pk); let mut core = self.playlist.core.write().await; core.remove_by_cache_pk(&cache_pk); + let snapshot = core.snapshot(); drop(core); // Sauvegarder si persistante @@ -73,6 +74,11 @@ impl ReadHandle { } } + // Mettre à jour l'index pk -> playlists + crate::manager::PlaylistManager() + .rebuild_track_index(&self.playlist.id, &snapshot) + .await; + // Ne pas avancer le curseur, continuer avec la position actuelle continue; } @@ -160,7 +166,7 @@ impl ReadHandle { } let title = self.playlist.title().await; - let remaining = self.remaining().await?; + let _remaining = self.remaining().await?; Ok(Container { id: self.playlist.id.clone(), diff --git a/pmoplaylist/src/handle/write.rs b/pmoplaylist/src/handle/write.rs index c93b390e..3104e9e3 100644 --- a/pmoplaylist/src/handle/write.rs +++ b/pmoplaylist/src/handle/write.rs @@ -38,6 +38,7 @@ impl WriteHandle { 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; @@ -48,7 +49,11 @@ impl WriteHandle { } // Notifier le manager - crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + let manager = crate::manager::PlaylistManager(); + manager + .rebuild_track_index(&self.playlist.id, &snapshot) + .await; + manager.notify_playlist_changed(&self.playlist.id); Ok(()) } @@ -69,6 +74,7 @@ impl WriteHandle { let record = Record::with_ttl(cache_pk, ttl); let mut core = self.playlist.core.write().await; core.push(record); + let snapshot = core.snapshot(); drop(core); self.playlist.touch().await; @@ -78,7 +84,11 @@ impl WriteHandle { self.save_to_db().await?; } - crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + let manager = crate::manager::PlaylistManager(); + manager + .rebuild_track_index(&self.playlist.id, &snapshot) + .await; + manager.notify_playlist_changed(&self.playlist.id); Ok(()) } @@ -103,6 +113,7 @@ impl WriteHandle { // Ajouter atomiquement let mut core = self.playlist.core.write().await; core.push_all(records); + let snapshot = core.snapshot(); drop(core); self.playlist.touch().await; @@ -112,7 +123,11 @@ impl WriteHandle { self.save_to_db().await?; } - crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + let manager = crate::manager::PlaylistManager(); + manager + .rebuild_track_index(&self.playlist.id, &snapshot) + .await; + manager.notify_playlist_changed(&self.playlist.id); Ok(()) } @@ -125,6 +140,7 @@ impl WriteHandle { let mut core = self.playlist.core.write().await; core.clear(); + let snapshot = core.snapshot(); drop(core); self.playlist.touch().await; @@ -133,7 +149,11 @@ impl WriteHandle { self.save_to_db().await?; } - crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + let manager = crate::manager::PlaylistManager(); + manager + .rebuild_track_index(&self.playlist.id, &snapshot) + .await; + manager.notify_playlist_changed(&self.playlist.id); Ok(()) } @@ -148,10 +168,13 @@ impl WriteHandle { self.playlist.mark_deleted(); // Supprimer du manager + // Nettoyer les index puis supprimer du manager + let manager = crate::manager::PlaylistManager(); + manager.rebuild_track_index(&self.playlist.id, &[]).await; crate::manager::delete_playlist_internal(&self.playlist.id).await?; // Notifier la suppression - crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + manager.notify_playlist_changed(&self.playlist.id); Ok(()) } @@ -181,6 +204,7 @@ impl WriteHandle { let mut core = self.playlist.core.write().await; core.set_capacity(max_size); + let snapshot = core.snapshot(); drop(core); self.playlist.touch().await; @@ -189,7 +213,11 @@ impl WriteHandle { self.save_to_db().await?; } - crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + let manager = crate::manager::PlaylistManager(); + manager + .rebuild_track_index(&self.playlist.id, &snapshot) + .await; + manager.notify_playlist_changed(&self.playlist.id); Ok(()) } @@ -202,6 +230,7 @@ impl WriteHandle { let mut core = self.playlist.core.write().await; core.set_default_ttl(ttl); + let snapshot = core.snapshot(); drop(core); self.playlist.touch().await; @@ -210,7 +239,11 @@ impl WriteHandle { self.save_to_db().await?; } - crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + let manager = crate::manager::PlaylistManager(); + manager + .rebuild_track_index(&self.playlist.id, &snapshot) + .await; + manager.notify_playlist_changed(&self.playlist.id); Ok(()) } diff --git a/pmoplaylist/src/lib.rs b/pmoplaylist/src/lib.rs index 19d03899..ac0e69cf 100644 --- a/pmoplaylist/src/lib.rs +++ b/pmoplaylist/src/lib.rs @@ -50,6 +50,10 @@ mod manager; mod persistence; mod playlist; mod track; +#[cfg(feature = "pmoserver")] +mod sse; +#[cfg(feature = "pmoserver")] +pub mod openapi; #[cfg(feature = "pmoconfig")] mod config_ext; @@ -58,7 +62,10 @@ mod config_ext; pub use error::{Error, Result}; pub use handle::{ReadHandle, WriteHandle}; pub use manager::{register_audio_cache, PlaylistManager, PlaylistManager as Manager}; +pub use manager::{PlaylistEvent, PlaylistEventEnvelope, PlaylistEventKind, subscribe_events}; pub use track::PlaylistTrack; +#[cfg(feature = "pmoserver")] +pub use sse::playlist_events_router; #[cfg(feature = "pmoconfig")] pub use config_ext::PlaylistConfigExt; diff --git a/pmoplaylist/src/manager.rs b/pmoplaylist/src/manager.rs index 109c8a32..dd7253cc 100644 --- a/pmoplaylist/src/manager.rs +++ b/pmoplaylist/src/manager.rs @@ -5,12 +5,14 @@ use crate::persistence::PersistenceManager; use crate::playlist::core::PlaylistConfig; use crate::playlist::Playlist; use crate::Result; +use pmocache::{CacheBroadcastEvent, CacheSubscription}; use once_cell::sync::OnceCell; use std::collections::HashMap; use std::path::PathBuf; use std::sync::{Arc, atomic::{AtomicU64, Ordering}}; use std::time::Duration; use tokio::sync::RwLock; +use tokio::sync::broadcast; use std::sync::RwLock as StdRwLock; /// Singleton PlaylistManager @@ -23,8 +25,35 @@ static AUDIO_CACHE: OnceCell> = OnceCell::new(); struct ManagerInner { playlists: RwLock>>, persistence: Option>, - callbacks: StdRwLock>>, + callbacks: StdRwLock>>, cb_counter: AtomicU64, + track_index: StdRwLock>>, // cache_pk -> playlists + cache_subscriptions: StdRwLock>, + event_tx: broadcast::Sender, +} + +/// Type d'évènement émis par le PlaylistManager. +#[derive(Debug, Clone)] +pub struct PlaylistEvent { + pub playlist_id: String, + pub kind: PlaylistEventKind, +} + +/// Variantes d'évènements playlist. +#[derive(Debug, Clone)] +pub enum PlaylistEventKind { + /// La playlist a été modifiée (ajout/suppression/changement de config). + Updated, + /// Un morceau référencé par la playlist a été servi par le cache audio. + TrackPlayed { cache_pk: String, qualifier: String }, +} + +/// Evènement enrichi pour diffusion (timestamp + source client éventuel). +#[derive(Debug, Clone)] +pub struct PlaylistEventEnvelope { + pub event: PlaylistEvent, + pub timestamp: std::time::SystemTime, + pub source_client: Option, } /// Gestionnaire central de playlists @@ -52,6 +81,9 @@ impl PlaylistManager { persistence: Some(persistence.clone()), callbacks: StdRwLock::new(HashMap::new()), cb_counter: AtomicU64::new(1), + track_index: StdRwLock::new(HashMap::new()), + cache_subscriptions: StdRwLock::new(HashMap::new()), + event_tx: broadcast::channel(256).0, }), }; @@ -128,12 +160,12 @@ impl PlaylistManager { Ok(WriteHandle::new(playlist, write_token)) } - /// Enregistre un callback de modification de playlist. + /// Enregistre un callback d'évènement playlist (update, track joué). /// /// Retourne un jeton (u64) pour désenregistrer plus tard. pub fn register_callback(&self, cb: F) -> u64 where - F: Fn(&str) + Send + Sync + 'static, + F: Fn(&PlaylistEvent) + Send + Sync + 'static, { let token = self.inner.cb_counter.fetch_add(1, Ordering::Relaxed); let mut guard = self.inner.callbacks.write().unwrap(); @@ -149,9 +181,181 @@ impl PlaylistManager { /// Notifie tous les callbacks qu'une playlist a changé. pub(crate) fn notify_playlist_changed(&self, id: &str) { + self.notify_playlist_event( + id, + PlaylistEventKind::Updated, + ); + } + + /// Notifie les callbacks qu'un morceau a été joué pour une playlist donnée. + pub(crate) fn notify_playlist_track_played(&self, playlist_id: &str, cache_pk: &str, qualifier: &str) { + self.notify_playlist_event( + playlist_id, + PlaylistEventKind::TrackPlayed { + cache_pk: cache_pk.to_string(), + qualifier: qualifier.to_string(), + }, + ); + } + + fn notify_playlist_event(&self, id: &str, kind: PlaylistEventKind) { + let event = PlaylistEvent { + playlist_id: id.to_string(), + kind, + }; + let envelope = PlaylistEventEnvelope { + event: event.clone(), + timestamp: std::time::SystemTime::now(), + source_client: None, + }; + let guard = self.inner.callbacks.read().unwrap(); for cb in guard.values() { - cb(id); + cb(&event); + } + + // Diffusion via canal interne (ignoré si aucun abonné) + let _ = self.inner.event_tx.send(envelope); + } + + /// Ré-inscrit les abonnements cache pour tous les pk connus (utilisé au boot ou après enregistrement du cache audio). + async fn sync_cache_subscriptions(&self) { + let pks: Vec = { + let index = self.inner.track_index.read().unwrap(); + index.keys().cloned().collect() + }; + + if pks.is_empty() { + return; + } + + if let Ok(cache) = audio_cache() { + for pk in pks { + // Ne pas doubler les abonnements + let already = { + let subs = self.inner.cache_subscriptions.read().unwrap(); + subs.contains_key(&pk) + }; + if already { + continue; + } + + let manager = self.clone(); + let token = cache + .subscribe_broadcast(pk.clone(), move |event: &CacheBroadcastEvent| { + manager.handle_cache_broadcast(event); + let index = manager.inner.track_index.read().unwrap(); + index.contains_key(&event.pk) + }) + .await; + + self.inner + .cache_subscriptions + .write() + .unwrap() + .insert(pk, token); + } + } + } + + /// Réconcilie l'index pk→playlists et les souscriptions cache pour une playlist donnée. + pub(crate) async fn rebuild_track_index( + &self, + playlist_id: &str, + records: &[Arc], + ) { + // 1) Retirer la playlist de toutes les entrées + let mut removed_pks = Vec::new(); + { + let mut index = self.inner.track_index.write().unwrap(); + for (pk, playlists) in index.iter_mut() { + playlists.retain(|p| p != playlist_id); + if playlists.is_empty() { + removed_pks.push(pk.clone()); + } + } + for pk in &removed_pks { + index.remove(pk); + } + // 2) Ajouter les nouveaux records + for record in records { + let entry = index.entry(record.cache_pk.clone()).or_default(); + if !entry.iter().any(|p| p == playlist_id) { + entry.push(playlist_id.to_string()); + } + } + } + + // 3) Se désabonner des pk qui ne sont plus référencés + // Collecter les tokens à désinscrire sans bloquer pendant l'await + let removed_tokens: Vec = { + if removed_pks.is_empty() { + Vec::new() + } else { + let mut subs = self.inner.cache_subscriptions.write().unwrap(); + removed_pks + .into_iter() + .filter_map(|pk| subs.remove(&pk)) + .collect() + } + }; + + if !removed_tokens.is_empty() { + if let Ok(cache) = audio_cache() { + for token in removed_tokens { + cache.unsubscribe_broadcast(&token).await; + } + } + } + + // 4) S'abonner aux nouveaux pk sans souscription + let missing: Vec = { + let index = self.inner.track_index.read().unwrap(); + let subs = self.inner.cache_subscriptions.read().unwrap(); + index + .iter() + .filter_map(|(pk, playlists)| { + if playlists.contains(&playlist_id.to_string()) && !subs.contains_key(pk) { + Some(pk.clone()) + } else { + None + } + }) + .collect() + }; + + if !missing.is_empty() { + if let Ok(cache) = audio_cache() { + for pk in missing { + let manager = self.clone(); + let token = cache + .subscribe_broadcast(pk.clone(), move |event: &CacheBroadcastEvent| { + manager.handle_cache_broadcast(event); + // Garder l'abonnement tant que le pk est référencé + let index = manager.inner.track_index.read().unwrap(); + index.contains_key(&event.pk) + }) + .await; + self.inner + .cache_subscriptions + .write() + .unwrap() + .insert(pk, token); + } + } + } + } + + fn handle_cache_broadcast(&self, event: &CacheBroadcastEvent) { + let playlists = { + let index = self.inner.track_index.read().unwrap(); + index.get(&event.pk).cloned() + }; + + if let Some(playlists) = playlists { + for playlist_id in playlists { + self.notify_playlist_track_played(&playlist_id, &event.pk, &event.qualifier); + } } } @@ -220,6 +424,9 @@ impl PlaylistManager { { let mut core = playlist.core.write().await; core.tracks = tracks; + let snapshot = core.snapshot(); + drop(core); + self.rebuild_track_index(&id, &snapshot).await; } // Acquérir le write lock @@ -264,6 +471,9 @@ impl PlaylistManager { { let mut core = playlist.core.write().await; core.tracks = tracks; + let snapshot = core.snapshot(); + drop(core); + self.rebuild_track_index(id, &snapshot).await; } playlists.insert(id.to_string(), playlist.clone()); @@ -286,6 +496,9 @@ impl PlaylistManager { drop(playlists); + // Nettoyer l'index et les souscriptions + self.rebuild_track_index(id, &[]).await; + // Supprimer de la DB if let Some(persistence) = &self.inner.persistence { persistence.delete_playlist(id).await?; @@ -349,6 +562,11 @@ impl PlaylistManager { } } +/// Souscrit au flux d'évènements playlist (Updated / TrackPlayed) avec timestamp. +pub fn subscribe_events() -> broadcast::Receiver { + PlaylistManager::get().inner.event_tx.subscribe() +} + /// Helper pour supprimer une playlist (appel� depuis WriteHandle) pub(crate) async fn delete_playlist_internal(id: &str) -> Result<()> { PlaylistManager::get().delete_playlist(id).await @@ -371,6 +589,14 @@ pub(crate) async fn delete_playlist_internal(id: &str) -> Result<()> { /// ``` pub fn register_audio_cache(cache: Arc) { let _ = AUDIO_CACHE.set(cache); + + // Si le PlaylistManager est déjà initialisé, synchroniser les abonnements + if let Some(manager) = PLAYLIST_MANAGER.get() { + let manager = manager.clone(); + tokio::spawn(async move { + manager.sync_cache_subscriptions().await; + }); + } } /// Helper pour acc�der au cache audio diff --git a/pmoplaylist/src/openapi.rs b/pmoplaylist/src/openapi.rs new file mode 100644 index 00000000..b46b6df9 --- /dev/null +++ b/pmoplaylist/src/openapi.rs @@ -0,0 +1,44 @@ +//! Documentation OpenAPI pour les endpoints playlists (SSE évènements). + +#[cfg(feature = "pmoserver")] +use utoipa::OpenApi; + +/// Documentation OpenAPI pour l'API playlist (flux SSE). +#[derive(OpenApi)] +#[openapi( + paths( + crate::sse::playlist_events_sse, + ), + components( + schemas( + crate::sse::EventPayload, + crate::sse::EventsQuery, + ) + ), + tags( + (name = "playlists", description = "Suivi des playlists et des morceaux joués") + ), + info( + title = "PMO Playlist API", + version = "0.1.0", + description = r#" +# Flux d'évènements playlists + +Endpoint SSE pour suivre : +- les modifications de playlists (updated) +- les lectures de morceaux appartenant aux playlists (track_played) + +Payload JSON par évènement : +- `playlist_id` : identifiant de la playlist +- `kind` : `updated` ou `track_played` +- `cache_pk` : pk du morceau (si track_played) +- `qualifier` : qualifier de diffusion (orig/stream/etc.) +- `timestamp` : horodatage UTC +- `source_client` : client à l'origine (optionnel) + "#, + license( + name = "MIT", + ), + ) +)] +pub struct ApiDoc; diff --git a/pmoplaylist/src/sse.rs b/pmoplaylist/src/sse.rs new file mode 100644 index 00000000..dc15ae60 --- /dev/null +++ b/pmoplaylist/src/sse.rs @@ -0,0 +1,89 @@ +//! SSE pour suivre les évènements de playlists (updates + morceaux joués). +//! +//! Route type : `GET /api/playlists/events?playlist_id=foo` + +use crate::{subscribe_events, PlaylistEventKind}; +use axum::{ + extract::Query, + response::sse::{Event, KeepAlive, Sse}, + response::IntoResponse, + Router, +}; +#[cfg(feature = "pmoserver")] +use async_stream::stream; +use serde::{Deserialize, Serialize}; +#[cfg(feature = "pmoserver")] +use tokio_stream::StreamExt; + +#[derive(Debug, Default, Deserialize)] +#[cfg_attr(feature = "pmoserver", derive(utoipa::IntoParams, utoipa::ToSchema))] +pub struct EventsQuery { + /// Filtrer sur une playlist précise (optionnel). + #[serde(default)] + pub playlist_id: Option, +} + +#[derive(Debug, Serialize)] +#[cfg_attr(feature = "pmoserver", derive(utoipa::ToSchema))] +pub struct EventPayload { + pub playlist_id: String, + pub kind: String, + pub cache_pk: Option, + pub qualifier: Option, + pub timestamp: chrono::DateTime, + pub source_client: Option, +} + +/// Handler SSE : diffuse les évènements playlist enrichis. +#[utoipa::path( + get, + path = "/api/playlists/events", + tag = "playlists", + params(EventsQuery), + responses( + (status = 200, description = "Flux SSE des évènements playlists (updated, track_played)", content_type = "text/event-stream") + ) +)] +pub async fn playlist_events_sse(Query(params): Query) -> impl IntoResponse { + let mut rx = subscribe_events(); + + let stream = stream! { + while let Ok(envelope) = rx.recv().await { + if let Some(filter) = ¶ms.playlist_id { + if &envelope.event.playlist_id != filter { + continue; + } + } + + let (kind, cache_pk, qualifier) = match &envelope.event.kind { + PlaylistEventKind::Updated => ("updated", None, None), + PlaylistEventKind::TrackPlayed { cache_pk, qualifier } => { + ("track_played", Some(cache_pk.as_str()), Some(qualifier.as_str())) + } + }; + + let ts = chrono::DateTime::::from(envelope.timestamp); + let payload = EventPayload { + playlist_id: envelope.event.playlist_id.clone(), + kind: kind.to_string(), + cache_pk: cache_pk.map(|s| s.to_string()), + qualifier: qualifier.map(|s| s.to_string()), + timestamp: ts, + source_client: envelope.source_client.clone(), + }; + + if let Ok(json) = serde_json::to_string(&payload) { + yield Ok::<_, axum::Error>(Event::default().event("playlist").data(json)); + } + } + }; + + Sse::new(stream).keep_alive(KeepAlive::default()) +} + +/// Router prêt à être monté (ex: `/api/playlists/events`). +pub fn playlist_events_router() -> Router { + use axum::routing::get; + + Router::new().route("/events", get(playlist_events_sse)) +} diff --git a/pmoupnp/Cargo.toml b/pmoupnp/Cargo.toml index 6f600523..29d98e40 100644 --- a/pmoupnp/Cargo.toml +++ b/pmoupnp/Cargo.toml @@ -10,6 +10,7 @@ pmoutils = { path = "../pmoutils" } pmoserver = { path = "../pmoserver" } pmocovers = { path = "../pmocovers", features = ["pmoserver"] } pmoaudiocache = { path = "../pmoaudiocache", features = ["pmoserver"] } +pmoplaylist = { path = "../pmoplaylist", optional = true, features = ["pmoserver"] } pmocache = { path = "../pmocache" } url = "2.5.7" @@ -35,5 +36,11 @@ reqwest = "0.12.23" utoipa = { version = "5.3", features = ["axum_extras"] } socket2 = "0.5" +[features] +default = ["server"] +server = ["pmoplaylist"] +paradise = [] +qobuz = [] + [target.'cfg(unix)'.dependencies] libc = "0.2" diff --git a/pmoupnp/src/upnp_server.rs b/pmoupnp/src/upnp_server.rs index 2bb8448e..d0ea21f2 100644 --- a/pmoupnp/src/upnp_server.rs +++ b/pmoupnp/src/upnp_server.rs @@ -315,6 +315,17 @@ impl UpnpServerExt for Server { let openapi = pmoaudiocache::ApiDoc::openapi(); self.add_openapi(api_router, openapi, "audio").await; + // API playlists (SSE + OpenAPI) + #[cfg(feature = "server")] + { + use pmoplaylist::{playlist_events_router, openapi::ApiDoc}; + // SSE /api/playlists/events + self.add_router("/api/playlists", playlist_events_router()).await; + // OpenAPI pour playlists + let openapi = ApiDoc::openapi(); + self.add_openapi(axum::Router::new(), openapi, "playlists").await; + } + // Enregistrer le cache dans le registre global pmoaudiocache::register_audio_cache(cache.clone());