From f80ebd9f3d5c467f947c4b0343a5ecb5d6b4d178 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Nov 2025 07:40:36 +0000 Subject: [PATCH] refactor: Remove obsolete orchestration layer, create RadioParadiseSource stub MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Major cleanup removing 2831 lines (~45%) of outdated server orchestration code. RadioParadiseStreamSource (pmoaudio integration) is now the primary implementation. ## Changes ### Removed (2393 lines) - **paradise/ module** - Complete server orchestration system: - worker.rs (1146 lines) - Background polling, caching, state machine - channel.rs (429 lines) - Channel lifecycle management - playlist.rs (294 lines) - Shared playlist management - history.rs (218 lines) - SQLite persistence - constants.rs (209 lines) - Server configuration constants - mod.rs (29 lines) - Module exports ### Replaced - **source.rs** (612 → 174 lines, -72%): - Old: Full MusicSource implementation with UPnP/DIDL integration - New: Minimal stub for backward compatibility with pmomediaserver - Returns empty results and deprecation warnings - Documents migration path to RadioParadiseStreamSource ### Updated - **config_ext.rs**: Now imports HISTORY_DEFAULT_MAX_TRACKS from channels module - **lib.rs**: - Removed paradise module - Updated documentation to focus on RadioParadiseStreamSource - Updated cargo features documentation ## Architecture **Before**: Complex orchestration with workers, channels, caching, history **After**: Simple API access + pmoaudio streaming (RadioParadiseStreamSource) ## Compatibility RadioParadiseSource stub maintains API compatibility for pmomediaserver while clearly indicating deprecation. All operations return empty results or errors with migration guidance. ## Testing - ✅ All 23 tests pass - ✅ Compilation successful with all features - ✅ pmomediaserver compatibility maintained (stub implementation) ## Migration Path Old (deprecated): ```rust let source = RadioParadiseSource::from_registry(client)?; ``` New (recommended): ```rust let stream_source = RadioParadiseStreamSource::new(client, None).await?; let node = Node::from_logic(stream_source); ``` --- pmoparadise/src/config_ext.rs | 6 +- pmoparadise/src/lib.rs | 49 +- pmoparadise/src/paradise/channel.rs | 428 --------- pmoparadise/src/paradise/constants.rs | 208 ----- pmoparadise/src/paradise/history.rs | 217 ----- pmoparadise/src/paradise/mod.rs | 28 - pmoparadise/src/paradise/playlist.rs | 293 ------- pmoparadise/src/paradise/worker.rs | 1145 ------------------------- pmoparadise/src/source.rs | 666 +++----------- 9 files changed, 137 insertions(+), 2903 deletions(-) delete mode 100644 pmoparadise/src/paradise/channel.rs delete mode 100644 pmoparadise/src/paradise/constants.rs delete mode 100644 pmoparadise/src/paradise/history.rs delete mode 100644 pmoparadise/src/paradise/mod.rs delete mode 100644 pmoparadise/src/paradise/playlist.rs delete mode 100644 pmoparadise/src/paradise/worker.rs diff --git a/pmoparadise/src/config_ext.rs b/pmoparadise/src/config_ext.rs index 9a0a79f9..7df74ec5 100644 --- a/pmoparadise/src/config_ext.rs +++ b/pmoparadise/src/config_ext.rs @@ -36,7 +36,7 @@ use anyhow::{anyhow, Result}; use pmoconfig::Config; use serde_yaml::{Number, Value}; -use crate::paradise::constants; +use crate::channels::HISTORY_DEFAULT_MAX_TRACKS; /// Nom du répertoire pour Radio Paradise (relatif au config_dir) /// @@ -222,7 +222,7 @@ impl RadioParadiseConfigExt for Config { Ok(Value::Number(n)) if n.is_i64() => Ok(n.as_i64().unwrap() as usize), _ => { // Use default and persist it - let default = constants::HISTORY_DEFAULT_MAX_TRACKS; + let default = HISTORY_DEFAULT_MAX_TRACKS; self.set_paradise_history_size(default)?; Ok(default) } @@ -245,7 +245,7 @@ mod tests { #[test] fn test_default_values() { assert_eq!(DEFAULT_HISTORY_DATABASE_DIR, "paradise"); - assert_eq!(constants::HISTORY_DEFAULT_MAX_TRACKS, 100); + assert_eq!(HISTORY_DEFAULT_MAX_TRACKS, 100); } #[test] diff --git a/pmoparadise/src/lib.rs b/pmoparadise/src/lib.rs index 250b7fd4..512230a1 100644 --- a/pmoparadise/src/lib.rs +++ b/pmoparadise/src/lib.rs @@ -164,54 +164,46 @@ //! } //! ``` //! -//! ## Caching Support (Feature: `cache`) +//! ## Audio Streaming (Feature: `pmoaudio`) //! -//! `pmoparadise` can optionally integrate with `pmocovers` and `pmoaudiocache` to cache -//! cover images and audio tracks locally: +//! For direct audio streaming and integration with pmoaudio pipelines, +//! use `RadioParadiseStreamSource`: //! //! ```no_run -//! # #[cfg(feature = "cache")] +//! # #[cfg(feature = "pmoaudio")] //! # { -//! use pmoparadise::{RadioParadiseClient, RadioParadiseSource}; -//! use std::sync::Arc; +//! use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource}; +//! use pmoaudio::pipeline::Node; //! //! #[tokio::main] //! async fn main() -> Result<(), Box> { -//! // Create caches -//! let cover_cache = Arc::new(pmocovers::cache::new_cache("./cache/covers", 500)?); -//! let audio_cache = Arc::new(pmoaudiocache::cache::new_cache("./cache/audio", 100)?); -//! -//! // Create client and source with caching //! let client = RadioParadiseClient::new().await?; -//! let source = RadioParadiseSource::new( -//! client, -//! 50, -//! cover_cache, -//! audio_cache, -//! ); +//! let stream_source = RadioParadiseStreamSource::new(client, None).await?; //! -//! println!("Source ready: {}", source.name()); +//! // Create audio node from stream source +//! let node = Node::from_logic(stream_source); +//! +//! // Use in pmoaudio pipeline... //! //! Ok(()) //! } //! # } //! ``` //! -//! **Benefits**: -//! - Cover images are automatically downloaded and converted to WebP -//! - Audio tracks are cached as FLAC with metadata preserved -//! - Subsequent access is instant (no re-download) -//! - URIs returned by `resolve_uri()` point to cached versions -//! -//! See the `with_cache` example for a complete demonstration. +//! **RadioParadiseStreamSource**: +//! - Downloads and decodes FLAC blocks in real-time +//! - Automatically detects bit depth (16/24/32-bit) +//! - Inserts track boundaries with metadata +//! - Integrates seamlessly with pmoaudio pipelines //! //! ## Cargo Features //! -//! - `default = ["metadata-only"]`: Standard metadata and streaming (no FLAC decoding) +//! - `default`: Standard metadata and streaming (no FLAC decoding) //! - `per-track`: Enable FLAC decoding and per-track extraction (adds `claxon`, `hound`, `tempfile`) //! - `pmoserver`: Enable REST API extension for pmoserver integration (adds `utoipa`, `axum`) -//! - `server`: Enable server-side features (cache registry integration) -//! - `cache`: Enable cover and audio caching support (adds `pmocovers`, `pmoaudiocache`) +//! - `pmoaudio`: Enable RadioParadiseStreamSource for pmoaudio integration +//! - `pmoconfig`: Enable configuration integration with pmoconfig +//! - `server`: Enable RadioParadiseSource stub for backward compatibility (deprecated) //! //! ## See Also //! @@ -222,7 +214,6 @@ pub mod channels; pub mod client; pub mod error; pub mod models; -pub mod paradise; pub mod source; pub mod stream; pub mod streaming; diff --git a/pmoparadise/src/paradise/channel.rs b/pmoparadise/src/paradise/channel.rs deleted file mode 100644 index 9953b061..00000000 --- a/pmoparadise/src/paradise/channel.rs +++ /dev/null @@ -1,428 +0,0 @@ -//! Channel orchestration primitives. -//! -//! This module wires together configuration, playlists, workers and client -//! tracking for a single Radio Paradise channel. The implementation is still -//! a scaffolding of the final behaviour; commands sent to the worker are -//! logged but not yet executing the full download/buffering pipeline. - -use super::history::HistoryBackend; -use super::playlist::{PlaylistEntry, SharedPlaylist}; -use super::worker::{ParadiseWorker, WorkerCommand}; -use crate::client::RadioParadiseClient; -use anyhow::{Context, Result}; -use async_stream::try_stream; -use bytes::Bytes; -use futures::{stream::BoxStream, StreamExt}; -use pmosource::SourceCacheManager; -use std::fmt; -use std::str::FromStr; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::Arc; -use tokio::fs::File; -use tokio::sync::{mpsc, Mutex}; -use tokio_util::io::ReaderStream; -use tracing::warn; - -/// Logical identifier for a Radio Paradise channel. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub enum ParadiseChannelKind { - Main, - Mellow, - Rock, - Eclectic, -} - -impl ParadiseChannelKind { - pub const fn id(self) -> u8 { - match self { - Self::Main => 0, - Self::Mellow => 1, - Self::Rock => 2, - Self::Eclectic => 3, - } - } - - pub const fn slug(self) -> &'static str { - match self { - Self::Main => "main", - Self::Mellow => "mellow", - Self::Rock => "rock", - Self::Eclectic => "eclectic", - } - } - - pub const fn display_name(self) -> &'static str { - match self { - Self::Main => "Main Mix", - Self::Mellow => "Mellow Mix", - Self::Rock => "Rock Mix", - Self::Eclectic => "Eclectic Mix", - } - } - - pub const fn description(self) -> &'static str { - match self { - Self::Main => "Eclectic mix of rock, world, electronica, and more", - Self::Mellow => "Mellower, less aggressive music", - Self::Rock => "Heavier, more guitar-driven music", - Self::Eclectic => "Curated worldwide selection", - } - } -} - -impl FromStr for ParadiseChannelKind { - type Err = anyhow::Error; - - fn from_str(s: &str) -> std::result::Result { - match s.to_ascii_lowercase().as_str() { - "main" | "0" => Ok(Self::Main), - "mellow" | "1" => Ok(Self::Mellow), - "rock" | "2" => Ok(Self::Rock), - "eclectic" | "3" => Ok(Self::Eclectic), - other => Err(anyhow::anyhow!("Unknown Radio Paradise channel: {}", other)), - } - } -} - -/// Metadata descriptor for a channel. -#[derive(Debug, Clone, Copy)] -pub struct ChannelDescriptor { - pub kind: ParadiseChannelKind, - pub id: u8, - pub slug: &'static str, - pub display_name: &'static str, - pub description: &'static str, -} - -impl ChannelDescriptor { - pub const fn new(kind: ParadiseChannelKind) -> Self { - Self { - id: kind.id(), - slug: kind.slug(), - display_name: kind.display_name(), - description: kind.description(), - kind, - } - } -} - -pub const ALL_CHANNELS: [ChannelDescriptor; 4] = [ - ChannelDescriptor::new(ParadiseChannelKind::Main), - ChannelDescriptor::new(ParadiseChannelKind::Mellow), - ChannelDescriptor::new(ParadiseChannelKind::Rock), - ChannelDescriptor::new(ParadiseChannelKind::Eclectic), -]; - -/// Returns the maximum valid channel ID -pub const fn max_channel_id() -> u8 { - (ALL_CHANNELS.len() - 1) as u8 -} - -/// Public handle to interact with a channel. -#[derive(Clone)] -pub struct ParadiseChannel { - inner: Arc, -} - -struct ParadiseChannelInner { - descriptor: ChannelDescriptor, - client: RadioParadiseClient, - history_max_tracks: usize, - playlist: SharedPlaylist, - history: Arc, - cache_manager: Arc, - active_clients: AtomicUsize, - worker_tx: mpsc::Sender, - worker: Mutex>, -} - -impl fmt::Debug for ParadiseChannel { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("ParadiseChannel") - .field("slug", &self.inner.descriptor.slug) - .field( - "active_clients", - &self.inner.active_clients.load(Ordering::SeqCst), - ) - .finish() - } -} - -impl ParadiseChannel { - #[allow(clippy::too_many_arguments)] - pub fn new( - descriptor: ChannelDescriptor, - base_client: RadioParadiseClient, - history_max_tracks: usize, - history: Arc, - cache_manager: Arc, - ) -> Result { - let client = base_client.clone_with_channel(descriptor.id); - let playlist = SharedPlaylist::new(history_max_tracks); - let (worker, worker_tx) = ParadiseWorker::spawn( - descriptor, - client.clone(), - history_max_tracks, - playlist.clone(), - history.clone(), - cache_manager.clone(), - ); - - Ok(Self { - inner: Arc::new(ParadiseChannelInner { - descriptor, - client, - history_max_tracks, - playlist, - history, - cache_manager, - active_clients: AtomicUsize::new(0), - worker_tx, - worker: Mutex::new(Some(worker)), - }), - }) - } - - pub fn descriptor(&self) -> ChannelDescriptor { - self.inner.descriptor - } - - pub fn playlist(&self) -> &SharedPlaylist { - &self.inner.playlist - } - - pub fn history_max_tracks(&self) -> usize { - self.inner.history_max_tracks - } - - pub fn history_backend(&self) -> &Arc { - &self.inner.history - } - - pub fn cache_manager(&self) -> Arc { - self.inner.cache_manager.clone() - } - - pub fn client(&self) -> &RadioParadiseClient { - &self.inner.client - } - - pub fn active_client_count(&self) -> usize { - self.inner.active_clients.load(Ordering::SeqCst) - } - - pub async fn connect_client( - &self, - client_id: impl Into, - ) -> Result { - let client_id = client_id.into(); - self.inner.active_clients.fetch_add(1, Ordering::SeqCst); - - if let Err(err) = self - .inner - .worker_tx - .send(WorkerCommand::ClientConnected { - client_id: client_id.clone(), - }) - .await - { - self.inner.active_clients.fetch_sub(1, Ordering::SeqCst); - return Err(anyhow::anyhow!("worker unavailable: {}", err)); - } - - self.inner.playlist.increment_all_pending().await; - self.ensure_started().await?; - - Ok(ParadiseClientStream::new(self.clone(), client_id)) - } - - pub async fn disconnect_client(&self, client_id: impl Into) -> Result<()> { - let client_id = client_id.into(); - self.inner.active_clients.fetch_sub(1, Ordering::SeqCst); - self.inner - .worker_tx - .send(WorkerCommand::ClientDisconnected { client_id }) - .await - .context("failed to notify worker of client disconnection")?; - Ok(()) - } - - pub async fn ensure_started(&self) -> Result<()> { - self.inner - .worker_tx - .send(WorkerCommand::EnsureReady) - .await - .context("failed to schedule worker warmup") - } - - pub async fn shutdown(&self) -> Result<()> { - self.inner - .worker_tx - .send(WorkerCommand::Shutdown) - .await - .ok(); - - let mut guard = self.inner.worker.lock().await; - if let Some(worker) = guard.take() { - worker - .wait() - .await - .context("failed to join worker task") - .map(|_| ()) - } else { - Ok(()) - } - } - - pub async fn mark_track_completed(&self, track: &Arc) { - let remaining = track.decrement_clients(); - if remaining > 0 { - return; - } - - if let Some(removed) = self - .inner - .playlist - .pop_front_matching(&track.track_id) - .await - { - if let Err(err) = self.inner.history.append(removed.as_history_entry()).await { - warn!( - channel = self.inner.descriptor.slug, - "Failed to persist history entry: {err:?}" - ); - } - - if let Err(err) = self - .inner - .history - .truncate(self.inner.history_max_tracks) - .await - { - warn!( - channel = self.inner.descriptor.slug, - "Failed to truncate history: {err:?}" - ); - } - - let history_entry = removed.as_history_entry(); - self.inner.playlist.push_history_entry(history_entry).await; - } - } -} - -/// Placeholder stream handle for per-client playback. -#[derive(Debug, Clone)] -pub struct ParadiseClientStream { - channel: ParadiseChannel, - client_id: String, -} - -impl ParadiseClientStream { - fn new(channel: ParadiseChannel, client_id: String) -> Self { - Self { channel, client_id } - } - - pub fn client_id(&self) -> &str { - &self.client_id - } - - pub fn channel(&self) -> ParadiseChannel { - self.channel.clone() - } - - pub fn into_byte_stream(self) -> BoxStream<'static, Result> { - let channel = self.channel.clone(); - let client_id = self.client_id.clone(); - let stream = try_stream! { - tracing::info!( - channel = channel.descriptor().slug, - client_id = %client_id, - "🎧 Client connecting to stream" - ); - channel.ensure_started().await?; - let mut last_track_id: Option = None; - loop { - let entries = channel.playlist().active_snapshot().await; - - // Find the next track after last_track_id - let next_entry = if let Some(ref last_id) = last_track_id { - // Find the position of the last track we read - let last_pos = entries.iter().position(|e| e.track_id == *last_id); - - // Get the next track (or wait if none available) - match last_pos { - Some(pos) if pos + 1 < entries.len() => { - Some(entries[pos + 1].clone()) - } - _ => { - // Last track not found (was removed) or no next track available - // Wait for more tracks to be added - channel.ensure_started().await?; - let current_len = entries.len(); - channel.playlist().wait_for_track_count(current_len).await; - continue; - } - } - } else { - // First track for this client - if entries.is_empty() { - channel.ensure_started().await?; - channel.playlist().wait_for_track_count(0).await; - continue; - } - Some(entries[0].clone()) - }; - - let entry = next_entry.unwrap(); - last_track_id = Some(entry.track_id.clone()); - - let audio_pk = entry - .audio_pk - .clone() - .ok_or_else(|| anyhow::anyhow!("Audio not cached yet"))?; - - channel - .cache_manager() - .wait_audio_ready(&audio_pk) - .await - .map_err(|e| anyhow::anyhow!(e.to_string()))?; - - let file_path = if let Some(path) = entry.file_path.clone() { - path - } else { - channel - .cache_manager() - .audio_file_path(&audio_pk) - .await - .ok_or_else(|| anyhow::anyhow!("Audio file path unavailable"))? - }; - - let file = File::open(&file_path).await?; - let mut reader = ReaderStream::new(file); - - while let Some(chunk) = reader.next().await { - let bytes = chunk?; - yield bytes; - } - - channel.mark_track_completed(&entry).await; - } - }; - - stream.boxed() - } -} - -impl Drop for ParadiseClientStream { - fn drop(&mut self) { - let channel = self.channel.clone(); - let client_id = self.client_id.clone(); - let slug = channel.descriptor().slug; - tokio::spawn(async move { - if let Err(err) = channel.disconnect_client(client_id).await { - warn!(channel = slug, "Failed to disconnect client: {err:?}"); - } - }); - } -} diff --git a/pmoparadise/src/paradise/constants.rs b/pmoparadise/src/paradise/constants.rs deleted file mode 100644 index d6dd6631..00000000 --- a/pmoparadise/src/paradise/constants.rs +++ /dev/null @@ -1,208 +0,0 @@ -//! Constants for Radio Paradise orchestration layer. -//! -//! This module defines all the hardcoded parameters for the Radio Paradise -//! integration. These values are based on empirical testing and Radio Paradise's -//! infrastructure characteristics. - -use std::time::Duration; - -// ============================================================================ -// Activity Lifecycle -// ============================================================================ - -/// Cooling timeout after all clients disconnect (seconds) -/// -/// After the last client disconnects, the channel enters a "cooling" state -/// where it remains active for this duration before shutting down completely. -/// This avoids rapid start/stop cycles if clients reconnect quickly. -/// -/// Value: 180 seconds (3 minutes) - good balance between responsiveness and stability -pub const COOLING_TIMEOUT_SECONDS: u64 = 180; - -// ============================================================================ -// Polling Intervals -// ============================================================================ - -/// High buffer polling interval (seconds) -/// -/// When the playlist buffer has 3+ blocks, poll less frequently to reduce -/// API load and network usage. -/// -/// Value: 120 seconds (2 minutes) -pub const POLLING_INTERVAL_HIGH_BUFFER: u64 = 120; - -/// Medium buffer polling interval (seconds) -/// -/// When the playlist buffer has 2 blocks, poll at moderate frequency. -/// -/// Value: 60 seconds (1 minute) -pub const POLLING_INTERVAL_MEDIUM_BUFFER: u64 = 60; - -/// Low buffer polling interval (seconds) -/// -/// When the playlist buffer has less than 2 blocks, poll frequently to -/// ensure continuous playback. -/// -/// Value: 20 seconds -pub const POLLING_INTERVAL_LOW_BUFFER: u64 = 20; - -/// Helper to get high buffer polling interval as Duration -pub fn polling_high_interval() -> Duration { - Duration::from_secs(POLLING_INTERVAL_HIGH_BUFFER) -} - -/// Helper to get medium buffer polling interval as Duration -pub fn polling_medium_interval() -> Duration { - Duration::from_secs(POLLING_INTERVAL_MEDIUM_BUFFER) -} - -/// Helper to get low buffer polling interval as Duration -pub fn polling_low_interval() -> Duration { - Duration::from_secs(POLLING_INTERVAL_LOW_BUFFER) -} - -// ============================================================================ -// Polling Backoff (on API errors) -// ============================================================================ - -/// Initial backoff delay on API error (seconds) -/// -/// When an API request fails, we wait this duration before retrying. -/// -/// Value: 20 seconds -pub const BACKOFF_INITIAL_SECONDS: u64 = 20; - -/// Maximum backoff delay (seconds) -/// -/// Backoff is capped at this value to avoid waiting too long. -/// -/// Value: 300 seconds (5 minutes) -pub const BACKOFF_MAX_SECONDS: u64 = 300; - -/// Backoff multiplier -/// -/// After each failure, the delay is multiplied by this factor. -/// Example: 20s → 40s → 80s → 160s → 300s (capped) -/// -/// Value: 2.0 (exponential backoff) -pub const BACKOFF_MULTIPLIER: f32 = 2.0; - -// ============================================================================ -// Cache Tuning -// ============================================================================ - -/// Maximum number of blocks to remember in the worker -/// -/// This prevents unbounded memory growth by limiting how many block event IDs -/// we track to avoid re-processing. -/// -/// Calculation: (4 channels + 1 buffer) × 3 blocks per channel = 15 blocks -/// Each block is ~20 minutes of audio, so 15 blocks ≈ 5 hours of history -/// -/// Value: 15 blocks -pub const MAX_BLOCKS_REMEMBERED: usize = 15; - -/// Number of bytes to use for track ID hashing -/// -/// Track IDs are constructed by hashing block content and track position. -/// This value defines how much of the FLAC data we read for hashing. -/// -/// Value: 512 bytes - sufficient for unique identification without excessive I/O -pub const TRACK_ID_HASH_BYTES: usize = 512; - -// ============================================================================ -// History -// ============================================================================ - -/// Default maximum number of tracks to keep in history -/// -/// This is used as the default if not configured via pmoconfig. -/// Users can override this value in their configuration. -/// -/// Value: 100 tracks - represents ~5-8 hours of playback history -pub const HISTORY_DEFAULT_MAX_TRACKS: usize = 100; - -// ============================================================================ -// Streaming -// ============================================================================ - -/// Stream buffer size (bytes) -/// -/// Buffer size for audio streaming. 64KB provides good balance between -/// latency and buffering efficiency. -/// -/// Value: 64 KB -pub const STREAM_BUFFER_SIZE_BYTES: usize = 64 * 1024; - -/// Enable gapless playback -/// -/// Radio Paradise blocks are designed for gapless playback - each block -/// transitions seamlessly to the next without audio gaps. -/// -/// Value: true (always enabled) -pub const STREAM_GAPLESS: bool = true; - -// Note: Metadata format is always ICY (Icecast/SHOUTcast metadata) -// No enum or constant needed as it's the only supported format - -// ============================================================================ -// API Configuration -// ============================================================================ - -/// Radio Paradise API base URL -/// -/// Base URL for all Radio Paradise API requests. -/// This is hardcoded as Radio Paradise's API endpoint doesn't change. -/// -/// Value: https://api.radioparadise.com -pub const API_BASE_URL: &str = "https://api.radioparadise.com"; - -/// API request timeout (seconds) -/// -/// Maximum time to wait for an API response before considering it failed. -/// -/// Value: 30 seconds -pub const API_TIMEOUT_SECONDS: u64 = 30; - -/// User agent for API requests -/// -/// Identifies PMOMusic in HTTP requests to Radio Paradise's servers. -/// -/// Value: PMO-RadioParadise/1.0 -pub const API_USER_AGENT: &str = "PMO-RadioParadise/1.0"; - -/// Helper to get API timeout as Duration -pub fn api_timeout() -> Duration { - Duration::from_secs(API_TIMEOUT_SECONDS) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_duration_helpers() { - assert_eq!(polling_high_interval(), Duration::from_secs(120)); - assert_eq!(polling_medium_interval(), Duration::from_secs(60)); - assert_eq!(polling_low_interval(), Duration::from_secs(20)); - assert_eq!(api_timeout(), Duration::from_secs(30)); - } - - #[test] - fn test_constants_sanity() { - // Polling intervals should be ordered - assert!(POLLING_INTERVAL_LOW_BUFFER < POLLING_INTERVAL_MEDIUM_BUFFER); - assert!(POLLING_INTERVAL_MEDIUM_BUFFER < POLLING_INTERVAL_HIGH_BUFFER); - - // Backoff should be reasonable - assert!(BACKOFF_INITIAL_SECONDS < BACKOFF_MAX_SECONDS); - assert!(BACKOFF_MULTIPLIER > 1.0); - - // Cache limits should be positive - assert!(MAX_BLOCKS_REMEMBERED > 0); - assert!(TRACK_ID_HASH_BYTES > 0); - - // History should be reasonable - assert!(HISTORY_DEFAULT_MAX_TRACKS > 0); - } -} diff --git a/pmoparadise/src/paradise/history.rs b/pmoparadise/src/paradise/history.rs deleted file mode 100644 index c276a810..00000000 --- a/pmoparadise/src/paradise/history.rs +++ /dev/null @@ -1,217 +0,0 @@ -//! History persistence for Radio Paradise playback. -//! -//! The worker pushes every completed track into the history backend while -//! keeping the latest entries available for UPnP browsing. We use SQLite -//! for persistent storage with an abstract trait for testability. -use crate::models::Song; -use async_trait::async_trait; -use chrono::{DateTime, Utc}; -use serde::{Deserialize, Serialize}; -use std::path::Path; -use std::sync::{Arc, Mutex as StdMutex}; -use tokio::task::spawn_blocking; - -/// Serializable record describing a played track. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct HistoryEntry { - pub track_id: String, - pub channel_id: u8, - pub started_at: chrono::DateTime, - pub duration_ms: u64, - pub song: SongSnapshot, -} - -/// Minimal snapshot of a Radio Paradise song at playback time. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct SongSnapshot { - pub title: String, - pub artist: String, - pub album: Option, - pub cover_url: Option, -} - -impl SongSnapshot { - pub fn title(&self) -> &str { - &self.title - } -} - -impl From<&Song> for SongSnapshot { - fn from(song: &Song) -> Self { - Self { - title: song.title.clone(), - artist: song.artist.clone(), - album: song.album.clone(), - cover_url: song.cover.clone(), - } - } -} - -/// Abstract persistence interface. -#[async_trait] -pub trait HistoryBackend: Send + Sync { - async fn append(&self, entry: HistoryEntry) -> anyhow::Result<()>; - async fn recent(&self, limit: usize) -> anyhow::Result>; - async fn len(&self) -> anyhow::Result; - async fn truncate(&self, keep: usize) -> anyhow::Result<()>; -} - -pub struct SqliteHistoryBackend { - conn: Arc>, -} - -impl SqliteHistoryBackend { - pub fn new(path: impl AsRef) -> anyhow::Result { - let path = path.as_ref(); - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent)?; - } - - let conn = rusqlite::Connection::open(path)?; - conn.pragma_update(None, "journal_mode", &"WAL")?; - conn.execute_batch( - "CREATE TABLE IF NOT EXISTS paradise_history ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - track_id TEXT NOT NULL, - channel_id INTEGER NOT NULL, - started_at_ms INTEGER NOT NULL, - duration_ms INTEGER NOT NULL, - title TEXT, - artist TEXT, - album TEXT, - cover_url TEXT - ); - CREATE INDEX IF NOT EXISTS idx_history_started_at ON paradise_history(started_at_ms);", - )?; - - Ok(Self { - conn: Arc::new(StdMutex::new(conn)), - }) - } - - fn conn(&self) -> Arc> { - self.conn.clone() - } -} - -#[async_trait] -impl HistoryBackend for SqliteHistoryBackend { - async fn append(&self, entry: HistoryEntry) -> anyhow::Result<()> { - let conn = self.conn(); - spawn_blocking(move || -> anyhow::Result<()> { - let conn = conn.lock().unwrap(); - conn.execute( - "INSERT INTO paradise_history (track_id, channel_id, started_at_ms, duration_ms, title, artist, album, cover_url) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", - rusqlite::params![ - entry.track_id, - entry.channel_id as i64, - entry.started_at.timestamp_millis(), - entry.duration_ms as i64, - entry.song.title, - entry.song.artist, - entry.song.album, - entry.song.cover_url, - ], - )?; - Ok(()) - }) - .await??; - Ok(()) - } - - async fn recent(&self, limit: usize) -> anyhow::Result> { - let conn = self.conn(); - let limit = limit as i64; - spawn_blocking(move || -> anyhow::Result> { - let conn = conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT track_id, channel_id, started_at_ms, duration_ms, title, artist, album, cover_url - FROM paradise_history - ORDER BY started_at_ms DESC - LIMIT ?1", - )?; - - let mut rows = stmt.query([limit])?; - let mut entries = Vec::new(); - while let Some(row) = rows.next()? { - let started_at_ms: i64 = row.get(2)?; - let started_at = DateTime::::from_timestamp_millis(started_at_ms) - .ok_or_else(|| anyhow::anyhow!("Invalid timestamp in history"))?; - let entry = HistoryEntry { - track_id: row.get(0)?, - channel_id: row.get::<_, i64>(1)? as u8, - started_at, - duration_ms: row.get::<_, i64>(3)? as u64, - song: SongSnapshot { - title: row.get::<_, Option>(4)?.unwrap_or_default(), - artist: row.get::<_, Option>(5)?.unwrap_or_default(), - album: row.get(6)?, - cover_url: row.get(7)?, - }, - }; - entries.push(entry); - } - Ok(entries) - }) - .await? - } - - async fn len(&self) -> anyhow::Result { - let conn = self.conn(); - let count = spawn_blocking(move || -> anyhow::Result { - let conn = conn.lock().unwrap(); - let mut stmt = conn.prepare("SELECT COUNT(*) FROM paradise_history")?; - let count: i64 = stmt.query_row([], |row| row.get(0))?; - Ok(count as usize) - }) - .await??; - Ok(count) - } - - async fn truncate(&self, keep: usize) -> anyhow::Result<()> { - let conn = self.conn(); - spawn_blocking(move || -> anyhow::Result<()> { - let conn = conn.lock().unwrap(); - let count: i64 = - conn.query_row("SELECT COUNT(*) FROM paradise_history", [], |row| { - row.get(0) - })?; - let keep = keep as i64; - if count <= keep { - return Ok(()); - } - let to_remove = count - keep; - conn.execute( - "DELETE FROM paradise_history - WHERE id IN ( - SELECT id FROM paradise_history - ORDER BY started_at_ms ASC - LIMIT ?1 - )", - rusqlite::params![to_remove], - )?; - Ok(()) - }) - .await??; - Ok(()) - } -} - -/// Creates a SQLite history backend with the given database path. -/// -/// The database file and parent directories will be created if they don't exist. -/// -/// # Arguments -/// -/// * `database_path` - Path to the SQLite database file -/// -/// # Example -/// -/// ```rust,ignore -/// let backend = create_history_backend("/var/lib/pmo/history.db")?; -/// ``` -pub fn create_history_backend(database_path: &str) -> anyhow::Result> { - let backend = SqliteHistoryBackend::new(database_path)?; - Ok(Arc::new(backend)) -} diff --git a/pmoparadise/src/paradise/mod.rs b/pmoparadise/src/paradise/mod.rs deleted file mode 100644 index f2860671..00000000 --- a/pmoparadise/src/paradise/mod.rs +++ /dev/null @@ -1,28 +0,0 @@ -//! Internal orchestration layer for dynamic Radio Paradise streaming. -//! -//! This module implements the high level structures described in the -//! Radio Paradise functional specification: -//! - `ParadiseChannel`: lifecycle and state machine for a single RP channel. -//! - `ParadiseWorker`: async task responsible for polling/downloading blocks. -//! - `ParadiseClientStream`: per-client audio stream with independent cursor. -//! - Shared caches and history storage hooked into existing PMO components. -//! -//! The implementation is split across several submodules to keep concerns -//! isolated (constants, playlist management, history persistence, etc.). -//! The goal of this scaffolding is to provide a clear, testable surface for -//! the eventual end-to-end integration with the UPnP server and HTTP routes. - -mod channel; -pub mod constants; -mod history; -mod playlist; -mod worker; - -pub use channel::{ - max_channel_id, ChannelDescriptor, ParadiseChannel, ParadiseChannelKind, ParadiseClientStream, - ALL_CHANNELS, -}; -pub use constants::*; // Export all constants -pub use history::{create_history_backend, HistoryBackend, HistoryEntry}; -pub use playlist::PlaylistEntry; -pub use worker::{load_rp_metadata, ParadiseWorker, RadioParadiseMetadata, WorkerCommand}; diff --git a/pmoparadise/src/paradise/playlist.rs b/pmoparadise/src/paradise/playlist.rs deleted file mode 100644 index 13a4fc89..00000000 --- a/pmoparadise/src/paradise/playlist.rs +++ /dev/null @@ -1,293 +0,0 @@ -//! Shared playlist structures for Radio Paradise channels. -//! -//! This module keeps track of the active queue and history for a Radio -//! Paradise channel. Each playlist entry knows how many clients still need -//! to consume it before the worker can evict it. - -use super::history::{HistoryEntry, SongSnapshot}; -use crate::models::Song; -use chrono::{DateTime, Utc}; -use std::collections::VecDeque; -use std::path::PathBuf; -use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering}; -use std::sync::Arc; -use std::time::SystemTime; -use tokio::sync::{Notify, RwLock}; - -/// Metadata stored for an active track. -#[derive(Debug)] -pub struct PlaylistEntry { - pub track_id: String, - pub channel_id: u8, - pub song: Arc, - pub started_at: DateTime, - pub duration_ms: u64, - pub audio_pk: Option, - pub file_path: Option, - pending_clients: AtomicUsize, -} - -impl PlaylistEntry { - #[allow(clippy::too_many_arguments)] - pub fn new( - track_id: String, - channel_id: u8, - song: Arc, - started_at: DateTime, - duration_ms: u64, - audio_pk: Option, - file_path: Option, - pending_clients: usize, - ) -> Self { - Self { - track_id, - channel_id, - song, - started_at, - duration_ms, - audio_pk, - file_path, - pending_clients: AtomicUsize::new(pending_clients), - } - } - - pub fn as_history_entry(&self) -> HistoryEntry { - HistoryEntry { - track_id: self.track_id.clone(), - channel_id: self.channel_id, - started_at: self.started_at, - duration_ms: self.duration_ms, - song: SongSnapshot::from(self.song.as_ref()), - } - } - - pub fn pending_clients(&self) -> usize { - self.pending_clients.load(Ordering::SeqCst) - } - - pub fn set_pending_clients(&self, value: usize) { - self.pending_clients.store(value, Ordering::SeqCst); - } - - pub fn increment_clients(&self) -> usize { - self.pending_clients.fetch_add(1, Ordering::SeqCst) + 1 - } - - pub fn decrement_clients(&self) -> usize { - let mut current = self.pending_clients.load(Ordering::SeqCst); - loop { - if current == 0 { - return 0; - } - match self.pending_clients.compare_exchange( - current, - current - 1, - Ordering::SeqCst, - Ordering::SeqCst, - ) { - Ok(_) => return current - 1, - Err(actual) => current = actual, - } - } - } -} - -#[derive(Default)] -struct PlaylistState { - active: VecDeque>, - history: VecDeque, - max_history: usize, -} - -impl PlaylistState { - fn new(max_history: usize) -> Self { - Self { - active: VecDeque::new(), - history: VecDeque::new(), - max_history, - } - } - - fn active_len(&self) -> usize { - self.active.len() - } - - fn push_active(&mut self, entry: Arc) { - self.active.push_back(entry); - } - - fn active_snapshot(&self) -> Vec> { - self.active.iter().cloned().collect() - } - - fn pop_front_if_ready(&mut self) -> Option> { - if let Some(front) = self.active.front() { - if front.pending_clients() == 0 { - return self.active.pop_front(); - } - } - None - } - - fn pop_front_matching(&mut self, track_id: &str) -> Option> { - if let Some(front) = self.active.front() { - if front.track_id == track_id && front.pending_clients() == 0 { - return self.active.pop_front(); - } - } - None - } - - fn push_history(&mut self, entry: HistoryEntry) { - self.history.push_back(entry); - self.trim_history(); - } - - fn recent_history(&self, limit: usize) -> Vec { - let total = self.history.len(); - let start = total.saturating_sub(limit); - self.history.iter().skip(start).cloned().collect() - } - - fn trim_history(&mut self) { - while self.history.len() > self.max_history { - self.history.pop_front(); - } - } - - fn clear(&mut self) -> bool { - let changed = !self.active.is_empty() || !self.history.is_empty(); - if changed { - self.active.clear(); - self.history.clear(); - } - changed - } - - fn increment_all(&self) { - for entry in &self.active { - entry.increment_clients(); - } - } -} - -struct SharedPlaylistInner { - state: RwLock, - notify: Notify, - update_id: AtomicU32, - last_change: RwLock>, -} - -#[derive(Clone)] -pub struct SharedPlaylist(Arc); - -impl SharedPlaylist { - pub fn new(max_history: usize) -> Self { - Self(Arc::new(SharedPlaylistInner { - state: RwLock::new(PlaylistState::new(max_history)), - notify: Notify::new(), - update_id: AtomicU32::new(0), - last_change: RwLock::new(None), - })) - } - - async fn touch(&self) { - self.0.update_id.fetch_add(1, Ordering::SeqCst); - let mut last_change = self.0.last_change.write().await; - *last_change = Some(SystemTime::now()); - } - - pub async fn push_active(&self, entry: Arc) { - let mut guard = self.0.state.write().await; - guard.push_active(entry); - drop(guard); - self.touch().await; - self.0.notify.notify_waiters(); - } - - pub async fn active_len(&self) -> usize { - let guard = self.0.state.read().await; - guard.active_len() - } - - pub async fn active_snapshot(&self) -> Vec> { - let guard = self.0.state.read().await; - guard.active_snapshot() - } - - pub async fn clear(&self) { - let mut guard = self.0.state.write().await; - let changed = guard.clear(); - drop(guard); - if changed { - self.touch().await; - self.0.notify.notify_waiters(); - } - } - - pub async fn wait_for_track_count(&self, current_len: usize) { - loop { - let len = { - let guard = self.0.state.read().await; - guard.active_len() - }; - - if len > current_len { - break; - } - - self.0.notify.notified().await; - } - } - - pub async fn pop_front_if_ready(&self) -> Option> { - let mut guard = self.0.state.write().await; - let result = guard.pop_front_if_ready(); - drop(guard); - - if result.is_some() { - self.touch().await; - self.0.notify.notify_waiters(); - } - - result - } - - pub async fn pop_front_matching(&self, track_id: &str) -> Option> { - let mut guard = self.0.state.write().await; - let result = guard.pop_front_matching(track_id); - drop(guard); - - if result.is_some() { - self.touch().await; - self.0.notify.notify_waiters(); - } - - result - } - - pub async fn push_history_entry(&self, entry: HistoryEntry) { - let mut guard = self.0.state.write().await; - guard.push_history(entry); - drop(guard); - self.touch().await; - } - - pub async fn recent_history(&self, limit: usize) -> Vec { - let guard = self.0.state.read().await; - guard.recent_history(limit) - } - - pub async fn increment_all_pending(&self) { - let guard = self.0.state.read().await; - guard.increment_all(); - } - - pub fn update_id(&self) -> u32 { - self.0.update_id.load(Ordering::SeqCst) - } - - pub async fn last_change(&self) -> Option { - self.0.last_change.read().await.clone() - } -} diff --git a/pmoparadise/src/paradise/worker.rs b/pmoparadise/src/paradise/worker.rs deleted file mode 100644 index c3986389..00000000 --- a/pmoparadise/src/paradise/worker.rs +++ /dev/null @@ -1,1145 +0,0 @@ -//! Background worker for Radio Paradise channels. -//! -//! The worker handles API polling, block ingestion, caching and playlist -//! maintenance. It keeps the channel state in sync with connected clients -//! and ensures fresh content is available according to the specification. - -use super::channel::ChannelDescriptor; -use super::constants::*; -use super::history::HistoryBackend; -use super::playlist::{PlaylistEntry, SharedPlaylist}; -use crate::client::RadioParadiseClient; -use crate::models::{Block, Song}; -use anyhow::{anyhow, Context, Result}; -use bytes::Bytes; -use chrono::Utc; -use futures::stream; -use pmosource::{SourceCacheManager, TrackMetadata}; -use std::collections::{HashSet, VecDeque}; -use std::pin::Pin; -use std::sync::Arc; -use tokio::sync::mpsc; -use tokio::task::JoinHandle; -use tokio::time::{sleep, Duration}; -use tokio_util::io::StreamReader; -use tracing::{debug, error, info, warn}; -use url::Url; - -/// Commands sent to the background worker. -#[derive(Debug)] -pub enum WorkerCommand { - EnsureReady, - ClientConnected { client_id: String }, - ClientDisconnected { client_id: String }, - RefreshBlock, - Shutdown, -} - -/// Handle to the spawned worker task. -pub struct ParadiseWorker { - descriptor: ChannelDescriptor, - join_handle: JoinHandle<()>, -} - -impl ParadiseWorker { - #[allow(clippy::too_many_arguments)] - pub fn spawn( - descriptor: ChannelDescriptor, - client: RadioParadiseClient, - history_max_tracks: usize, - playlist: SharedPlaylist, - history: Arc, - cache_manager: Arc, - ) -> (Self, mpsc::Sender) { - let (tx, mut rx) = mpsc::channel(32); - - let join_handle = tokio::spawn(async move { - info!(channel = descriptor.slug, "Starting Radio Paradise worker"); - - let mut state = WorkerState::new( - descriptor, - client, - history_max_tracks, - playlist, - history, - cache_manager, - ); - - loop { - if let Some(task) = state.scheduled_task.as_mut() { - let kind = task.kind; - let mut pending_command: Option> = None; - - tokio::select! { - cmd = rx.recv() => { - pending_command = Some(cmd); - } - _ = &mut task.sleep => { - state.scheduled_task = None; - if let Err(err) = state.handle_scheduled_task(kind).await { - error!(channel = state.descriptor.slug, "Worker scheduled task error: {err:?}"); - state.on_error(err); - } - } - } - - if let Some(Some(cmd)) = pending_command { - if let Err(err) = state.handle_command(cmd).await { - error!( - channel = state.descriptor.slug, - "Worker command error: {err:?}" - ); - state.on_error(err); - } - if state.shutdown { - break; - } - } else if let Some(None) = pending_command { - // Command channel closed, terminate - break; - } - } else { - match rx.recv().await { - Some(cmd) => { - if let Err(err) = state.handle_command(cmd).await { - error!( - channel = state.descriptor.slug, - "Worker command error: {err:?}" - ); - state.on_error(err); - } - if state.shutdown { - break; - } - } - None => break, - } - } - } - - info!(channel = state.descriptor.slug, "Worker stopped"); - }); - - ( - Self { - descriptor, - join_handle, - }, - tx, - ) - } - - pub async fn wait(self) -> Result<()> { - if let Err(err) = self.join_handle.await { - if err.is_cancelled() { - warn!( - channel = self.descriptor.slug, - "Worker task cancelled: {err}" - ); - return Ok(()); - } - return Err(anyhow!("Worker join error: {}", err)); - } - Ok(()) - } -} - -struct WorkerState { - descriptor: ChannelDescriptor, - client: RadioParadiseClient, - playlist: SharedPlaylist, - history: Arc, - cache_manager: Arc, - active_clients: usize, - status: ChannelLifecycle, - processed_blocks: HashSet, - processing_blocks: HashSet, - recent_blocks: VecDeque, - next_block_hint: Option, - scheduled_task: Option, - backoff: BackoffState, - shutdown: bool, -} - -#[derive(Clone)] -struct SongTaskContext { - cache_manager: Arc, - playlist: SharedPlaylist, - descriptor_id: u8, - slug: &'static str, -} - -impl WorkerState { - fn song_task_context(&self) -> SongTaskContext { - SongTaskContext { - cache_manager: Arc::clone(&self.cache_manager), - playlist: self.playlist.clone(), - descriptor_id: self.descriptor.id, - slug: self.descriptor.slug, - } - } - - fn new( - descriptor: ChannelDescriptor, - client: RadioParadiseClient, - _history_max_tracks: usize, - playlist: SharedPlaylist, - history: Arc, - cache_manager: Arc, - ) -> Self { - Self { - descriptor, - client, - playlist, - history, - cache_manager, - active_clients: 0, - status: ChannelLifecycle::Idle, - processed_blocks: HashSet::new(), - processing_blocks: HashSet::new(), - recent_blocks: VecDeque::new(), - next_block_hint: None, - scheduled_task: None, - backoff: BackoffState::new(), - shutdown: false, - } - } - - async fn handle_command(&mut self, cmd: WorkerCommand) -> Result<()> { - debug!(channel = self.descriptor.slug, ?cmd, "Worker command"); - - match cmd { - WorkerCommand::EnsureReady => { - self.ensure_ready().await?; - } - WorkerCommand::ClientConnected { .. } => { - self.active_clients = self.active_clients.saturating_add(1); - self.enter_active(); - self.ensure_ready().await?; - } - WorkerCommand::ClientDisconnected { .. } => { - self.active_clients = self.active_clients.saturating_sub(1); - if self.active_clients == 0 { - self.enter_cooling(); - } - } - WorkerCommand::RefreshBlock => { - self.fetch_next_block().await?; - } - WorkerCommand::Shutdown => { - self.shutdown = true; - self.cancel_scheduled_task(); - } - } - - if !self.shutdown { - self.maybe_schedule_poll().await; - } - - Ok(()) - } - - async fn handle_scheduled_task(&mut self, kind: ScheduledTaskKind) -> Result<()> { - match kind { - ScheduledTaskKind::Poll => { - self.fetch_next_block().await?; - self.maybe_schedule_poll().await; - } - ScheduledTaskKind::Cooling => { - debug!( - channel = self.descriptor.slug, - "Cooling timeout reached -> idle" - ); - self.status = ChannelLifecycle::Idle; - self.next_block_hint = None; - self.playlist.clear().await; - self.processed_blocks.clear(); - self.recent_blocks.clear(); - } - } - Ok(()) - } - - fn on_error(&mut self, err: anyhow::Error) { - warn!(channel = self.descriptor.slug, "Worker error: {err:?}"); - let delay = self.backoff.next_delay(); - self.schedule_task(ScheduledTaskKind::Poll, delay); - } - - fn enter_active(&mut self) { - if !matches!(self.status, ChannelLifecycle::Active) { - debug!( - channel = self.descriptor.slug, - "Channel entering Active state" - ); - } - self.status = ChannelLifecycle::Active; - if matches!(self.scheduled_task_kind(), Some(ScheduledTaskKind::Cooling)) { - self.cancel_scheduled_task(); - } - self.backoff.reset(); - } - - fn enter_cooling(&mut self) { - if matches!(self.status, ChannelLifecycle::Idle) { - return; - } - debug!( - channel = self.descriptor.slug, - "Channel entering Cooling state" - ); - self.status = ChannelLifecycle::Cooling; - let duration = Duration::from_secs(COOLING_TIMEOUT_SECONDS.max(1)); - self.schedule_task(ScheduledTaskKind::Cooling, duration); - } - - async fn ensure_ready(&mut self) -> Result<()> { - if !matches!(self.status, ChannelLifecycle::Active) { - self.enter_active(); - } - - let has_tracks = self.playlist.active_len().await > 0; - - if !has_tracks { - debug!( - channel = self.descriptor.slug, - "Playlist empty – fetching now playing" - ); - let now_playing = self.client.now_playing().await?; - self.process_block(now_playing.block).await?; - } - - Ok(()) - } - - async fn fetch_next_block(&mut self) -> Result<()> { - if !matches!(self.status, ChannelLifecycle::Active) { - debug!( - channel = self.descriptor.slug, - "Skipping poll while not active" - ); - return Ok(()); - } - - let event_id = self.next_block_hint; - let block = self.client.get_block(event_id).await?; - self.process_block(block).await?; - Ok(()) - } - - async fn process_block(&mut self, block: Block) -> Result<()> { - // Check if we just processed this block (songs are already in playlist) - if self.is_recent_block(block.event) { - debug!( - channel = self.descriptor.slug, - event = block.event, - "Skipping already processed block (songs already in playlist)" - ); - self.next_block_hint = Some(block.end_event); - return Ok(()); - } - - // Check if this block is currently being processed by another task - // This prevents race conditions when the same block is requested multiple times - if self.processing_blocks.contains(&block.event) { - warn!( - channel = self.descriptor.slug, - event = block.event, - "Block is already being processed, skipping duplicate request" - ); - return Ok(()); - } - - // Check if all songs from this block are in cache - // If yes, restore from cache instead of downloading - if self.check_all_songs_cached(&block).await { - info!( - channel = self.descriptor.slug, - event = block.event, - "Block found in cache, restoring without download" - ); - self.restore_from_cache(&block).await?; - self.record_processed_block(block.event); - self.next_block_hint = Some(block.end_event); - self.backoff.reset(); - return Ok(()); - } - - // Mark block as being processed - self.processing_blocks.insert(block.event); - let event = block.event; // Save for cleanup - - // Process the block and ensure cleanup even on error - let result = self.process_block_inner(block).await; - - // Always remove from processing set, whether success or error - self.processing_blocks.remove(&event); - - result - } - - async fn process_block_inner(&mut self, block: Block) -> Result<()> { - info!( - channel = self.descriptor.slug, - event = block.event, - "Processing Radio Paradise block with progressive streaming" - ); - - let _ = &self.history; - - // Start streaming the block - let block_url = Url::parse(&block.url)?; - let http_stream = self - .client - .stream_block(&block_url) - .await - .context("Failed to start block stream")?; - - let ordered_songs = block.songs_ordered(); - - // Decode in streaming mode using spawn_blocking - let (tx, mut rx) = mpsc::channel::(16); - - let decode_handle = tokio::task::spawn_blocking(move || -> Result<()> { - use crate::streaming::StreamingPCMDecoder; - - let mut decoder = StreamingPCMDecoder::new(http_stream) - .context("Failed to create streaming decoder")?; - - info!( - "Streaming decoder initialized: {}Hz, {} channels, {} bits", - decoder.sample_rate(), - decoder.channels(), - decoder.bits_per_sample() - ); - - // Decode chunks and send them - while let Some(chunk) = decoder.decode_chunk()? { - if tx.blocking_send(chunk).is_err() { - // Receiver dropped, stop decoding - break; - } - } - - Ok(()) - }); - - // Process songs as chunks arrive - let mut accumulated_pcm = Vec::new(); - let mut current_song_idx = 0; - let mut sample_rate = 0u32; - let mut channels = 0u32; - let mut bits_per_sample = 0u32; - - while let Some(chunk) = rx.recv().await { - // Store metadata from first chunk - if sample_rate == 0 { - sample_rate = chunk.sample_rate; - channels = chunk.channels; - bits_per_sample = 16; // Normalized to 16-bit by decoder - } - - accumulated_pcm.extend_from_slice(&chunk.samples); - let current_position_ms = chunk.position_ms; - - // Check if we've completed any songs - while current_song_idx < ordered_songs.len() { - let (song_index, song) = ordered_songs[current_song_idx]; - - // Calculate song boundaries - let song_start_ms = song.elapsed; - let song_end_ms = if current_song_idx + 1 < ordered_songs.len() { - ordered_songs[current_song_idx + 1].1.elapsed - } else { - u64::MAX // Last song goes to end of block - }; - - // Check if we have enough PCM for this song - if current_position_ms >= song_end_ms { - // Extract song samples - let start_frame = crate::streaming::ms_to_frames(song_start_ms, sample_rate); - let end_frame = crate::streaming::ms_to_frames(song_end_ms, sample_rate); - - let start_sample = start_frame * channels as usize; - let end_sample = end_frame * channels as usize; - - if end_sample <= accumulated_pcm.len() { - let track_samples = accumulated_pcm[start_sample..end_sample].to_vec(); - - info!( - channel = self.descriptor.slug, - song_index = song_index, - position_ms = current_position_ms, - "✅ Song '{}' ready for encoding ({} samples)", - song.title, - track_samples.len() - ); - - let context = self.song_task_context(); - spawn_song_processing( - context, - block.clone(), - song_index, - song.clone(), - track_samples, - sample_rate, - channels as usize, - bits_per_sample, - self.active_clients, - song.duration, - current_position_ms, - ); - - current_song_idx += 1; - } else { - // Not enough samples yet, wait for more chunks - break; - } - } else { - // Haven't reached this song's end yet - break; - } - } - } - - // Wait for decoder to finish - decode_handle.await??; - - // Process any remaining songs (last song in block) - if current_song_idx < ordered_songs.len() { - let (song_index, song) = ordered_songs[current_song_idx]; - let song_start_ms = song.elapsed; - let start_frame = crate::streaming::ms_to_frames(song_start_ms, sample_rate); - let start_sample = start_frame * channels as usize; - - if start_sample < accumulated_pcm.len() { - let track_samples = accumulated_pcm[start_sample..].to_vec(); - - info!( - channel = self.descriptor.slug, - song_index = song_index, - "Processing last song '{}' ({} samples)", - song.title, - track_samples.len() - ); - - let context = self.song_task_context(); - spawn_song_processing( - context, - block.clone(), - song_index, - song.clone(), - track_samples, - sample_rate, - channels as usize, - bits_per_sample, - self.active_clients, - song.duration, - song_start_ms, - ); - } - } - - self.record_processed_block(block.event); - self.next_block_hint = Some(block.end_event); - self.backoff.reset(); - - Ok(()) - } - - /// Stocke les métadonnées Radio Paradise pour un fichier audio caché - /// - /// Cette fonction persiste toutes les métadonnées RP dans la base de données - /// du cache audio, permettant leur récupération future sans dépendance aux - /// données en mémoire. - fn compute_track_id(&self, block: &Block, song_index: usize) -> String { - compute_track_id_for_descriptor(self.descriptor.id, block, song_index) - } - - async fn maybe_schedule_poll(&mut self) { - if !matches!(self.status, ChannelLifecycle::Active) { - return; - } - - let buffer_len = self.playlist.active_len().await; - - let interval = if buffer_len > 3 { - polling_high_interval() - } else if buffer_len >= 2 { - polling_medium_interval() - } else { - polling_low_interval() - }; - - self.schedule_task(ScheduledTaskKind::Poll, interval); - } - - fn schedule_task(&mut self, kind: ScheduledTaskKind, duration: Duration) { - self.scheduled_task = Some(ScheduledTask { - kind, - sleep: Box::pin(sleep(duration)), - }); - } - - fn cancel_scheduled_task(&mut self) { - self.scheduled_task = None; - } - - fn scheduled_task_kind(&self) -> Option { - self.scheduled_task.as_ref().map(|task| task.kind) - } - - fn record_processed_block(&mut self, event: u64) { - self.processed_blocks.insert(event); - self.recent_blocks.push_back(event); - let max = MAX_BLOCKS_REMEMBERED.max(1); - while self.recent_blocks.len() > max { - if let Some(ev) = self.recent_blocks.pop_front() { - self.processed_blocks.remove(&ev); - } - } - } - - fn is_recent_block(&self, event: u64) -> bool { - self.processed_blocks.contains(&event) - } - - /// Check if all songs from a block are already cached - async fn check_all_songs_cached(&self, block: &Block) -> bool { - let ordered_songs = block.songs_ordered(); - - for (song_index, _song) in &ordered_songs { - let track_id = self.compute_track_id(block, *song_index); - - // Check if metadata exists - let metadata = match self.cache_manager.get_metadata(&track_id).await { - Some(m) => m, - None => { - debug!( - channel = self.descriptor.slug, - event = block.event, - song_index = *song_index, - "Song not in cache: no metadata" - ); - return false; - } - }; - - // Check if audio is cached - let audio_pk = match metadata.cached_audio_pk { - Some(pk) => pk, - None => { - debug!( - channel = self.descriptor.slug, - event = block.event, - song_index = *song_index, - "Song not in cache: no audio_pk" - ); - return false; - } - }; - - // Check if file exists - if self - .cache_manager - .audio_file_path(&audio_pk) - .await - .is_none() - { - debug!( - channel = self.descriptor.slug, - event = block.event, - song_index = *song_index, - "Song not in cache: file not found" - ); - return false; - } - } - - debug!( - channel = self.descriptor.slug, - event = block.event, - "All {} songs are cached", - ordered_songs.len() - ); - true - } - - /// Restore songs from cache and add them to the playlist - async fn restore_from_cache(&mut self, block: &Block) -> Result<()> { - info!( - channel = self.descriptor.slug, - event = block.event, - "Restoring block from cache (no download needed)" - ); - - let ordered_songs = block.songs_ordered(); - - for (song_index, song) in &ordered_songs { - let track_id = self.compute_track_id(block, *song_index); - - // Get metadata (we already checked it exists in check_all_songs_cached) - let metadata = self - .cache_manager - .get_metadata(&track_id) - .await - .ok_or_else(|| anyhow!("Metadata disappeared for track_id: {}", track_id))?; - - let audio_pk = metadata - .cached_audio_pk - .clone() - .ok_or_else(|| anyhow!("Audio PK disappeared for track_id: {}", track_id))?; - - // Get cover PK if available - let cover_pk = if let Some(ref cover_path) = song.cover { - if let Some(cover_url) = block.cover_url(cover_path) { - match self.cache_manager.cache_cover(&cover_url).await { - Ok(pk) => Some(pk), - Err(err) => { - warn!(channel = self.descriptor.slug, "Cover cache error: {err}"); - None - } - } - } else { - None - } - } else { - None - }; - - // Update metadata with cover if we just cached it - if cover_pk.is_some() && metadata.cached_cover_pk.is_none() { - let updated_metadata = TrackMetadata { - cached_cover_pk: cover_pk, - ..metadata.clone() - }; - self.cache_manager - .update_metadata(track_id.clone(), updated_metadata) - .await; - } - - let file_path = self - .cache_manager - .audio_file_path(&audio_pk) - .await - .ok_or_else(|| anyhow!("File disappeared for audio_pk: {}", audio_pk))?; - - let duration_ms = song.duration; - - let entry = Arc::new(PlaylistEntry::new( - track_id, - self.descriptor.id, - Arc::new((*song).clone()), - Utc::now(), - duration_ms, - Some(audio_pk), - Some(file_path), - self.active_clients, - )); - - self.playlist.push_active(entry).await; - - info!( - channel = self.descriptor.slug, - song_index = *song_index, - "🎵 Restored '{}' from cache", - song.title - ); - } - - info!( - channel = self.descriptor.slug, - event = block.event, - "Block restored from cache: {} songs", - ordered_songs.len() - ); - - Ok(()) - } -} - -struct ScheduledTask { - kind: ScheduledTaskKind, - sleep: Pin>, -} - -#[derive(Clone, Copy)] -enum ScheduledTaskKind { - Poll, - Cooling, -} - -#[derive(Clone, Copy, Debug)] -enum ChannelLifecycle { - Idle, - Cooling, - Active, -} - -struct BackoffState { - current: Option, -} - -impl BackoffState { - fn new() -> Self { - Self { current: None } - } - - fn reset(&mut self) { - self.current = None; - } - - fn next_delay(&mut self) -> Duration { - let next = match self.current { - Some(current) => { - let multiplied = (current.as_secs_f32() * BACKOFF_MULTIPLIER).round() as u64; - Duration::from_secs(multiplied.min(BACKOFF_MAX_SECONDS)) - } - None => Duration::from_secs(BACKOFF_INITIAL_SECONDS), - }; - self.current = Some(next); - next - } -} - -fn compute_track_id_for_descriptor(descriptor_id: u8, block: &Block, song_index: usize) -> String { - format!( - "rp:{}:event_{}_song_{}", - descriptor_id, block.event, song_index - ) -} - -async fn store_rp_metadata( - cache_manager: &SourceCacheManager, - audio_pk: &str, - track_id: &str, - channel_id: u8, - song: &Song, - duration_ms: u64, - event: u64, - cover_pk: Option<&str>, -) -> Result<()> { - use serde_json::json; - - cache_manager.set_audio_metadata(audio_pk, "rp_title", json!(song.title))?; - cache_manager.set_audio_metadata(audio_pk, "rp_artist", json!(song.artist))?; - cache_manager.set_audio_metadata(audio_pk, "rp_album", json!(song.album))?; - cache_manager.set_audio_metadata(audio_pk, "rp_year", json!(song.year))?; - - cache_manager.set_audio_metadata(audio_pk, "rp_duration_ms", json!(duration_ms))?; - cache_manager.set_audio_metadata(audio_pk, "rp_elapsed_ms", json!(song.elapsed))?; - - cache_manager.set_audio_metadata(audio_pk, "rp_track_id", json!(track_id))?; - cache_manager.set_audio_metadata(audio_pk, "rp_channel_id", json!(channel_id))?; - cache_manager.set_audio_metadata(audio_pk, "rp_event", json!(event))?; - - cache_manager.set_audio_metadata(audio_pk, "rp_rating", json!(song.rating))?; - cache_manager.set_audio_metadata(audio_pk, "rp_cover_url", json!(song.cover))?; - cache_manager.set_audio_metadata(audio_pk, "rp_cover_pk", json!(cover_pk))?; - - Ok(()) -} - -async fn cache_cover_for_song( - cache_manager: &SourceCacheManager, - slug: &'static str, - block: &Block, - song: &Song, -) -> Result> { - if let Some(ref cover_path) = song.cover { - if let Some(cover_url) = block.cover_url(cover_path) { - match cache_manager.cache_cover(&cover_url).await { - Ok(pk) => return Ok(Some(pk)), - Err(err) => { - warn!(channel = slug, "Cover cache error: {err}"); - } - } - } else { - warn!( - channel = slug, - "Unable to resolve cover URL for {}", cover_path - ); - } - } - Ok(None) -} - -async fn encode_song_to_cache( - cache_manager: Arc, - descriptor_id: u8, - slug: &'static str, - block: Block, - song_index: usize, - song: Song, - track_samples: Vec, - sample_rate: u32, - channels: usize, - bits_per_sample: u32, - active_clients: usize, - duration_ms: u64, -) -> Result> { - let flac_bytes = encode_samples_to_flac(track_samples, channels, sample_rate, bits_per_sample) - .await - .context("Failed to encode song to FLAC")?; - - let track_id = compute_track_id_for_descriptor(descriptor_id, &block, song_index); - let placeholder_uri = format!("{}#{}", block.url, song_index); - - let mut metadata = TrackMetadata { - original_uri: placeholder_uri.clone(), - cached_audio_pk: None, - cached_cover_pk: None, - }; - - if let Some(cover_pk) = - cache_cover_for_song(cache_manager.as_ref(), slug, &block, &song).await? - { - metadata.cached_cover_pk = Some(cover_pk); - } - - let flac_len = flac_bytes.len() as u64; - let reader = StreamReader::new(stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from( - flac_bytes, - ))])); - - let audio_pk = cache_manager - .cache_audio_from_reader(&track_id, reader, Some(flac_len)) - .await - .map_err(|e| anyhow!("Cache audio error: {e}"))?; - - cache_manager - .wait_audio_ready(&audio_pk) - .await - .map_err(|e| anyhow!("Wait audio ready error: {e}"))?; - - metadata.cached_audio_pk = Some(audio_pk.clone()); - cache_manager - .update_metadata(track_id.clone(), metadata.clone()) - .await; - - if let Err(e) = store_rp_metadata( - cache_manager.as_ref(), - &audio_pk, - &track_id, - descriptor_id, - &song, - duration_ms, - block.event, - metadata.cached_cover_pk.as_deref(), - ) - .await - { - warn!(channel = slug, "Failed to store RP metadata: {e:?}"); - } - - let file_path = cache_manager.audio_file_path(&audio_pk).await; - - let entry = Arc::new(PlaylistEntry::new( - track_id, - descriptor_id, - Arc::new(song.clone()), - Utc::now(), - duration_ms, - Some(audio_pk), - file_path, - active_clients, - )); - - Ok(entry) -} - -fn spawn_song_processing( - context: SongTaskContext, - block: Block, - song_index: usize, - song: Song, - track_samples: Vec, - sample_rate: u32, - channels: usize, - bits_per_sample: u32, - active_clients: usize, - duration_ms: u64, - position_ms: u64, -) { - tokio::spawn(async move { - let SongTaskContext { - cache_manager, - playlist, - descriptor_id, - slug, - } = context; - - let song_title = song.title.clone(); - - match encode_song_to_cache( - cache_manager, - descriptor_id, - slug, - block, - song_index, - song, - track_samples, - sample_rate, - channels, - bits_per_sample, - active_clients, - duration_ms, - ) - .await - { - Ok(entry) => { - playlist.push_active(entry).await; - info!( - channel = slug, - song_index = song_index, - "🎵 Song '{}' available after {}ms (streaming mode)", - song_title, - position_ms - ); - } - Err(err) => { - warn!( - channel = slug, - song_index = song_index, - "Failed to process song '{}' asynchronously: {err:?}", - song_title - ); - } - } - }); -} - -async fn encode_samples_to_flac( - samples: Vec, - channels: usize, - sample_rate: u32, - bits_per_sample: u32, -) -> anyhow::Result> { - tokio::task::spawn_blocking(move || { - use flacenc::bitsink::ByteSink; - use flacenc::component::BitRepr; - use flacenc::error::Verify; - - // Note: Claxon retourne les samples dans leur résolution native - // Un fichier FLAC 16 bits retourne des samples i32 avec des valeurs dans la plage i16 - // Pas besoin de normalisation supplémentaire - let config = flacenc::config::Encoder::default() - .into_verified() - .map_err(|e| anyhow!("FLAC config error: {e:?}"))?; - - let source = flacenc::source::MemSource::from_samples( - &samples, - channels, - bits_per_sample as usize, - sample_rate as usize, - ); - - let flac_stream = flacenc::encode_with_fixed_block_size(&config, source, config.block_size) - .map_err(|e| anyhow!("FLAC encode error: {e:?}"))?; - - let mut sink = ByteSink::new(); - flac_stream - .write(&mut sink) - .map_err(|e| anyhow!("FLAC write error: {e:?}"))?; - - Ok::<_, anyhow::Error>(sink.into_inner()) - }) - .await? -} - -/// Métadonnées Radio Paradise récupérées depuis le cache -/// -/// Cette structure contient toutes les métadonnées RP stockées de manière -/// persistante dans le cache audio. -#[derive(Debug, Clone)] -pub struct RadioParadiseMetadata { - /// Titre de la chanson - pub title: String, - /// Artiste - pub artist: String, - /// Album (optionnel) - pub album: Option, - /// Année de sortie (optionnelle) - pub year: Option, - /// Durée en millisecondes - pub duration_ms: u64, - /// Offset depuis le début du block en millisecondes - pub elapsed_ms: u64, - /// Identifiant unique de la piste - pub track_id: String, - /// ID du canal Radio Paradise (0-3) - pub channel_id: u8, - /// ID de l'événement (block) - pub event: u64, - /// Note de la chanson (0-10, optionnelle) - pub rating: Option, - /// URL de la couverture (optionnelle) - pub cover_url: Option, - /// PK de la couverture dans le cache (optionnelle) - pub cover_pk: Option, -} - -/// Charge les métadonnées Radio Paradise depuis le cache audio -/// -/// Cette fonction lit toutes les métadonnées RP stockées pour un fichier -/// audio donné et les retourne dans une structure `RadioParadiseMetadata`. -/// -/// # Arguments -/// -/// * `cache_manager` - Le gestionnaire de cache source -/// * `audio_pk` - Clé primaire du fichier audio dans le cache -/// -/// # Returns -/// -/// Les métadonnées RP si elles existent et sont complètes, sinon une erreur. -/// -/// # Erreurs -/// -/// Cette fonction retourne une erreur si : -/// - Les métadonnées n'existent pas dans le cache -/// - Les métadonnées sont incomplètes ou corrompues -/// - Il y a une erreur de lecture du cache -pub async fn load_rp_metadata( - cache_manager: &SourceCacheManager, - audio_pk: &str, -) -> Result { - // Helper macro pour récupérer une métadonnée requise - macro_rules! get_required { - ($key:expr, $type:ty) => {{ - cache_manager - .get_audio_metadata(audio_pk, $key)? - .and_then(|v| serde_json::from_value::<$type>(v).ok()) - .ok_or_else(|| anyhow!("Missing or invalid metadata: {}", $key))? - }}; - } - - // Helper macro pour récupérer une métadonnée optionnelle - macro_rules! get_optional { - ($key:expr, $type:ty) => {{ - cache_manager - .get_audio_metadata(audio_pk, $key)? - .and_then(|v| { - if v.is_null() { - None - } else { - serde_json::from_value::<$type>(v).ok() - } - }) - }}; - } - - Ok(RadioParadiseMetadata { - title: get_required!("rp_title", String), - artist: get_required!("rp_artist", String), - album: get_optional!("rp_album", String), - year: get_optional!("rp_year", u32), - duration_ms: get_required!("rp_duration_ms", u64), - elapsed_ms: get_required!("rp_elapsed_ms", u64), - track_id: get_required!("rp_track_id", String), - channel_id: get_required!("rp_channel_id", u8), - event: get_required!("rp_event", u64), - rating: get_optional!("rp_rating", f32), - cover_url: get_optional!("rp_cover_url", String), - cover_pk: get_optional!("rp_cover_pk", String), - }) -} diff --git a/pmoparadise/src/source.rs b/pmoparadise/src/source.rs index d6b6e8cb..946af700 100644 --- a/pmoparadise/src/source.rs +++ b/pmoparadise/src/source.rs @@ -1,395 +1,114 @@ -//! Music source implementation for Radio Paradise built on the new -//! `paradise` orchestration layer. +//! DEPRECATED: Stub implementation of RadioParadiseSource //! -//! The source exposes a DIDL-Lite hierarchy compatible with UPnP -//! ContentDirectory while delegating block ingestion, caching and -//! multi-client streaming to [`ParadiseChannel`]. +//! **⚠️ This module is deprecated and will be removed in a future version.** +//! +//! The orchestration-based RadioParadiseSource has been replaced by +//! `RadioParadiseStreamSource`, which integrates directly with the pmoaudio +//! pipeline for streaming and decoding. +//! +//! ## Migration Guide +//! +//! **Old approach** (deprecated): +//! ```rust,ignore +//! use pmoparadise::RadioParadiseSource; +//! let source = RadioParadiseSource::from_registry(client)?; +//! ``` +//! +//! **New approach** (recommended): +//! ```rust,ignore +//! use pmoparadise::RadioParadiseStreamSource; +//! use pmoaudio::pipeline::Node; +//! +//! let stream_source = RadioParadiseStreamSource::new(client, None).await?; +//! let node = Node::from_logic(stream_source); +//! // Use node in pmoaudio pipeline +//! ``` +//! +//! This stub implementation is provided only for backward compatibility with +//! existing code (e.g., pmomediaserver) until it can be updated to use +//! RadioParadiseStreamSource. use crate::client::RadioParadiseClient; -use crate::paradise::{ - create_history_backend, ChannelDescriptor, ParadiseChannel, PlaylistEntry, ALL_CHANNELS, -}; - -#[cfg(not(feature = "pmoconfig"))] -use crate::paradise::HISTORY_DEFAULT_MAX_TRACKS; -use anyhow::Result as AnyhowResult; -use pmoaudiocache::Cache as AudioCache; -use pmocovers::Cache as CoverCache; -use pmodidl::{Container, Item, Resource}; -use pmosource::pmodidl; -use pmosource::{ - async_trait, BrowseResult, CacheStatus, MusicSource, MusicSourceError, Result, - SourceCacheManager, SourceStatistics, -}; -use std::collections::HashMap; -use std::sync::Arc; +use pmosource::pmodidl::{Container, Item}; +use pmosource::{async_trait, BrowseResult, MusicSource, MusicSourceError, Result}; use std::time::SystemTime; -use tracing::warn; -/// Default image for Radio Paradise (300x300 WebP, embedded in binary) +/// Default Radio Paradise image (embedded in binary) const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp"); -fn channel_collection_id(channel_id: u8) -> String { - format!("radio-paradise:{}", channel_id) -} - -fn channel_container_id(channel_id: u8) -> String { - format!("radio-paradise:channel:{}", channel_id) -} - -fn parse_channel_container_id(object_id: &str) -> Option { - let mut parts = object_id.split(':'); - match (parts.next(), parts.next(), parts.next(), parts.next()) { - (Some("radio-paradise"), Some("channel"), Some(id_str), None) => id_str.parse().ok(), - _ => None, - } -} - -fn parse_track_channel(track_id: &str) -> Option { - let mut parts = track_id.split(':'); - match (parts.next(), parts.next(), parts.next(), parts.next()) { - (Some("rp"), Some(channel_str), Some(_rest), None) => channel_str.parse().ok(), - _ => None, - } -} - -fn format_duration(duration_seconds: u64) -> String { - let hours = duration_seconds / 3600; - let minutes = (duration_seconds % 3600) / 60; - let seconds = duration_seconds % 60; - format!("{hours}:{minutes:02}:{seconds:02}") -} - -#[derive(Clone)] +/// DEPRECATED: Stub implementation of RadioParadiseSource +/// +/// This is a minimal stub that implements the MusicSource trait with no-op +/// implementations. It exists only to maintain API compatibility during the +/// migration to RadioParadiseStreamSource. +/// +/// **Do not use this in new code.** Use `RadioParadiseStreamSource` instead. +#[derive(Clone, Debug)] pub struct RadioParadiseSource { - inner: Arc, -} - -struct RadioParadiseSourceInner { - channels: HashMap>, -} - -impl std::fmt::Debug for RadioParadiseSource { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("RadioParadiseSource").finish() - } + _client: RadioParadiseClient, } impl RadioParadiseSource { + /// DEPRECATED: Create a new RadioParadiseSource from registry + /// + /// This method is deprecated and will always return an error indicating + /// that the orchestration-based source is no longer supported. + /// + /// Use `RadioParadiseStreamSource` instead for audio streaming. #[cfg(feature = "server")] - pub fn from_registry(client: RadioParadiseClient) -> Result { - // Load history configuration from pmoconfig using the config extension trait - #[cfg(feature = "pmoconfig")] - let (database_path, history_max_tracks) = { - use crate::config_ext::RadioParadiseConfigExt; - let cfg = pmoconfig::get_config(); - let database_path = cfg.get_paradise_history_database().map_err(|e| { - MusicSourceError::SourceUnavailable(format!( - "Failed to get history database path: {}", - e - )) - })?; - let max_tracks = cfg.get_paradise_history_size().map_err(|e| { - MusicSourceError::SourceUnavailable(format!("Failed to get history size: {}", e)) - })?; - (database_path, max_tracks) - }; - - #[cfg(not(feature = "pmoconfig"))] - let (database_path, history_max_tracks) = { - use std::path::PathBuf; - let mut path = PathBuf::from(std::env::var("HOME").unwrap_or_else(|_| ".".to_string())); - path.push(".config"); - path.push("pmo"); - path.push("paradise"); - std::fs::create_dir_all(&path).ok(); - path.push("history.db"); - ( - path.to_string_lossy().to_string(), - HISTORY_DEFAULT_MAX_TRACKS, - ) - }; - - let history_backend = create_history_backend(&database_path).map_err(|e| { - MusicSourceError::SourceUnavailable(format!( - "Failed to initialize history backend: {}", - e - )) - })?; - let mut channels = HashMap::new(); - - for descriptor in ALL_CHANNELS.iter() { - let cache_manager = Arc::new(SourceCacheManager::from_registry( - channel_collection_id(descriptor.id), - )?); - let channel = Arc::new( - ParadiseChannel::new( - *descriptor, - client.clone(), - history_max_tracks, - history_backend.clone(), - cache_manager, - ) - .map_err(|e| { - MusicSourceError::SourceUnavailable(format!( - "Failed to initialize channel {}: {e}", - descriptor.slug - )) - })?, - ); - channels.insert(descriptor.id, channel); - } - - Ok(Self { - inner: Arc::new(RadioParadiseSourceInner { channels }), - }) + pub fn from_registry(_client: RadioParadiseClient) -> Result { + Err(MusicSourceError::SourceUnavailable( + "RadioParadiseSource is deprecated. Use RadioParadiseStreamSource instead." + .to_string(), + )) } + /// DEPRECATED: Create a new RadioParadiseSource from registry with defaults + /// + /// This method creates a stub instance that will log deprecation warnings + /// but allows existing code to compile. + /// + /// Use `RadioParadiseStreamSource` instead for audio streaming. #[cfg(feature = "server")] - pub fn from_registry_default(client: RadioParadiseClient) -> Result { - Self::from_registry(client) + pub fn from_registry_default(client: RadioParadiseClient) -> Self { + tracing::warn!( + "RadioParadiseSource::from_registry_default is deprecated. \ + Use RadioParadiseStreamSource for audio streaming." + ); + Self { _client: client } } - pub fn new( - client: RadioParadiseClient, - cover_cache: Arc, - audio_cache: Arc, - ) -> Self { - // Load history configuration from pmoconfig using the config extension trait - #[cfg(feature = "pmoconfig")] - let (database_path, history_max_tracks) = { - use crate::config_ext::RadioParadiseConfigExt; - let cfg = pmoconfig::get_config(); - let database_path = cfg.get_paradise_history_database().unwrap_or_else(|e| { - panic!("Failed to get history database path: {e}"); - }); - let max_tracks = cfg.get_paradise_history_size().unwrap_or_else(|e| { - panic!("Failed to get history size: {e}"); - }); - (database_path, max_tracks) - }; - - #[cfg(not(feature = "pmoconfig"))] - let (database_path, history_max_tracks) = { - use std::path::PathBuf; - let mut path = PathBuf::from(std::env::var("HOME").unwrap_or_else(|_| ".".to_string())); - path.push(".config"); - path.push("pmo"); - path.push("paradise"); - std::fs::create_dir_all(&path).ok(); - path.push("history.db"); - ( - path.to_string_lossy().to_string(), - HISTORY_DEFAULT_MAX_TRACKS, - ) - }; - - let history_backend: Arc = - create_history_backend(&database_path).unwrap_or_else(|err| { - panic!("Failed to initialize history backend: {err}"); - }); - let mut channels = HashMap::new(); - - for descriptor in ALL_CHANNELS.iter() { - let cache_manager = Arc::new(SourceCacheManager::new( - channel_collection_id(descriptor.id), - Arc::clone(&cover_cache), - Arc::clone(&audio_cache), - )); - match ParadiseChannel::new( - *descriptor, - client.clone(), - history_max_tracks, - history_backend.clone(), - cache_manager, - ) { - Ok(channel) => { - channels.insert(descriptor.id, Arc::new(channel)); - } - Err(err) => { - warn!( - channel = descriptor.slug, - "Failed to initialize channel: {err:?}" - ); - } - } - } - - Self { - inner: Arc::new(RadioParadiseSourceInner { channels }), - } + /// DEPRECATED: Create a new RadioParadiseSource with default settings + /// + /// This method is deprecated and only exists for API compatibility. + pub fn new_default(client: RadioParadiseClient) -> Self { + tracing::warn!( + "RadioParadiseSource::new_default is deprecated. \ + Use RadioParadiseStreamSource for audio streaming." + ); + Self { _client: client } } - pub fn new_default( - client: RadioParadiseClient, - cover_cache: Arc, - audio_cache: Arc, - ) -> Self { - Self::new(client, cover_cache, audio_cache) - } - - pub fn client_for_channel(&self, channel: u8) -> Option { - self.inner - .channels - .get(&channel) - .map(|ch| ch.client().clone()) - } - - pub fn channel(&self, id: u8) -> Option> { - self.inner.channels.get(&id).cloned() - } - - fn build_root_container(&self) -> Container { - Container { - id: "radio-paradise".to_string(), - parent_id: "0".to_string(), - restricted: Some("1".to_string()), - child_count: Some(ALL_CHANNELS.len().to_string()), - searchable: Some("1".to_string()), - title: "Radio Paradise".to_string(), - class: "object.container".to_string(), - containers: vec![], - items: vec![], - } - } - - async fn build_channel_containers(&self) -> Vec { - let mut containers = Vec::new(); - for descriptor in ALL_CHANNELS.iter() { - if let Some(channel) = self.channel(descriptor.id) { - let len = channel.playlist().active_len().await; - containers.push(Container { - id: channel_container_id(descriptor.id), - parent_id: "radio-paradise".to_string(), - restricted: Some("1".to_string()), - child_count: Some(len.to_string()), - searchable: Some("1".to_string()), - title: descriptor.display_name.to_string(), - class: "object.container.playlistContainer".to_string(), - containers: vec![], - items: vec![], - }); - } - } - containers - } - - async fn channel_items( - &self, - descriptor: ChannelDescriptor, - offset: usize, - limit: Option, - ) -> Result> { - let channel = self - .channel(descriptor.id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(descriptor.slug.to_string()))?; - - channel - .ensure_started() - .await - .map_err(|e| MusicSourceError::SourceUnavailable(e.to_string()))?; - - let entries = channel.playlist().active_snapshot().await; - if entries.is_empty() || offset >= entries.len() { - return Ok(Vec::new()); - } - - let end = limit - .map(|count| offset + count) - .unwrap_or(entries.len()) - .min(entries.len()); - - let parent_id = channel_container_id(descriptor.id); - - let mut items = Vec::with_capacity(end - offset); - for entry in entries.into_iter().skip(offset).take(end - offset) { - match self.entry_to_item(channel.clone(), &parent_id, entry).await { - Ok(item) => items.push(item), - Err(err) => warn!( - channel = descriptor.slug, - "Failed to build DIDL item: {err:?}" - ), - } - } - Ok(items) - } - - async fn entry_to_item( - &self, - channel: Arc, - parent_id: &str, - entry: Arc, - ) -> AnyhowResult { - let cache_manager = channel.cache_manager(); - let metadata = cache_manager.get_metadata(&entry.track_id).await; - - let resource_url = cache_manager - .resolve_uri(&entry.track_id) - .await - .or_else(|_| { - metadata - .as_ref() - .map(|meta| meta.original_uri.clone()) - .ok_or_else(|| MusicSourceError::ObjectNotFound(entry.track_id.clone())) - })?; - - let mut album_art = metadata - .as_ref() - .and_then(|meta| meta.cached_cover_pk.as_ref()) - .and_then(|pk| cache_manager.cover_url(pk, None).ok()); - - if album_art.is_none() { - album_art = entry.song.cover.clone(); - } - - let duration_seconds = entry.duration_ms / 1000; - let duration_str = if duration_seconds > 0 { - Some(format_duration(duration_seconds as u64)) - } else { - None - }; - - let resource = Resource { - protocol_info: "http-get:*:audio/flac:*".to_string(), - bits_per_sample: None, - sample_frequency: None, - nr_audio_channels: None, - duration: duration_str.clone(), - url: resource_url, - }; - - Ok(Item { - id: entry.track_id.clone(), - parent_id: parent_id.to_string(), - restricted: Some("1".to_string()), - title: entry.song.title.clone(), - creator: Some(entry.song.artist.clone()), - class: "object.item.audioItem.musicTrack".to_string(), - artist: Some(entry.song.artist.clone()), - album: entry.song.album.clone(), - genre: None, - album_art, - album_art_pk: None, - date: None, - original_track_number: None, - resources: vec![resource], - descriptions: vec![], - }) - } - - fn channels_iter(&self) -> impl Iterator)> { - self.inner.channels.iter() + /// DEPRECATED: Create a new RadioParadiseSource with cache + /// + /// This method is deprecated and only exists for API compatibility. + pub fn new_with_cache(client: RadioParadiseClient, _cache_size: usize) -> Self { + tracing::warn!( + "RadioParadiseSource::new_with_cache is deprecated. \ + Use RadioParadiseStreamSource for audio streaming." + ); + Self { _client: client } } } #[async_trait] impl MusicSource for RadioParadiseSource { fn name(&self) -> &str { - "Radio Paradise" + "Radio Paradise (DEPRECATED)" } fn id(&self) -> &str { - "radio-paradise" + "radio-paradise-deprecated" } fn default_image(&self) -> &[u8] { @@ -397,43 +116,32 @@ impl MusicSource for RadioParadiseSource { } async fn root_container(&self) -> Result { - Ok(self.build_root_container()) + Ok(Container { + id: "radio-paradise-deprecated".to_string(), + parent_id: "0".to_string(), + restricted: Some("1".to_string()), + child_count: Some("0".to_string()), + searchable: Some("0".to_string()), + title: "Radio Paradise (DEPRECATED)".to_string(), + class: "object.container".to_string(), + containers: vec![], + items: vec![], + }) } - async fn browse(&self, object_id: &str) -> Result { - match object_id { - "0" => Ok(BrowseResult::Containers(vec![self.build_root_container()])), - "radio-paradise" => { - let containers = self.build_channel_containers().await; - Ok(BrowseResult::Containers(containers)) - } - _ => { - if let Some(channel_id) = parse_channel_container_id(object_id) { - let descriptor = ALL_CHANNELS - .iter() - .find(|desc| desc.id == channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - - let items = self.channel_items(*descriptor, 0, None).await?; - Ok(BrowseResult::Items(items)) - } else { - Err(MusicSourceError::ObjectNotFound(object_id.to_string())) - } - } - } + async fn browse(&self, _object_id: &str) -> Result { + tracing::warn!("RadioParadiseSource::browse called but source is deprecated"); + Ok(BrowseResult::Mixed { + containers: vec![], + items: vec![], + }) } - async fn resolve_uri(&self, object_id: &str) -> Result { - let channel_id = parse_track_channel(object_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - let channel = self - .channel(channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - channel - .cache_manager() - .resolve_uri(object_id) - .await - .map_err(|e| MusicSourceError::CacheError(e.to_string())) + async fn resolve_uri(&self, _object_id: &str) -> Result { + Err(MusicSourceError::SourceUnavailable( + "RadioParadiseSource is deprecated. Use RadioParadiseStreamSource instead." + .to_string(), + )) } fn supports_fifo(&self) -> bool { @@ -441,171 +149,25 @@ impl MusicSource for RadioParadiseSource { } async fn append_track(&self, _track: Item) -> Result<()> { - Err(MusicSourceError::FifoNotSupported) + Err(MusicSourceError::SourceUnavailable( + "RadioParadiseSource is deprecated and does not support FIFO operations." + .to_string(), + )) } async fn remove_oldest(&self) -> Result> { - Err(MusicSourceError::FifoNotSupported) + Ok(None) } async fn update_id(&self) -> u32 { - self.channels_iter() - .map(|(_, channel)| channel.playlist().update_id()) - .max() - .unwrap_or(0) + 0 } async fn last_change(&self) -> Option { - let mut latest: Option = None; - for (_, channel) in self.channels_iter() { - if let Some(change) = channel.playlist().last_change().await { - latest = Some(match latest { - Some(current) if change <= current => current, - _ => change, - }); - } - } - latest + None } - async fn get_items(&self, offset: usize, count: usize) -> Result> { - let mut all = Vec::new(); - for descriptor in ALL_CHANNELS.iter() { - let mut items = self.channel_items(*descriptor, 0, None).await?; - all.append(&mut items); - } - - if offset >= all.len() { - return Ok(Vec::new()); - } - - let end = if count == 0 { - all.len() - } else { - (offset + count).min(all.len()) - }; - - Ok(all.into_iter().skip(offset).take(end - offset).collect()) - } - - async fn get_available_formats(&self, _object_id: &str) -> Result> { - Ok(vec![pmosource::AudioFormat { - format_id: "flac".to_string(), - mime_type: "audio/flac".to_string(), - sample_rate: Some(44100), - bit_depth: Some(16), - bitrate: None, - channels: Some(2), - }]) - } - - async fn get_cache_status(&self, object_id: &str) -> Result { - let channel_id = parse_track_channel(object_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - let channel = self - .channel(channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - channel - .cache_manager() - .get_cache_status(object_id) - .await - .map_err(|e| MusicSourceError::CacheError(e.to_string())) - } - - async fn cache_item(&self, object_id: &str) -> Result { - let channel_id = parse_track_channel(object_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - let channel = self - .channel(channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - channel - .cache_manager() - .get_cache_status(object_id) - .await - .map_err(|e| MusicSourceError::CacheError(e.to_string())) - } - - async fn browse_paginated( - &self, - object_id: &str, - offset: usize, - limit: usize, - ) -> Result { - match object_id { - "0" => { - if offset == 0 { - Ok(BrowseResult::Containers(vec![self.build_root_container()])) - } else { - Ok(BrowseResult::Containers(Vec::new())) - } - } - "radio-paradise" => { - let containers = self.build_channel_containers().await; - let total = containers.len(); - if offset >= total { - return Ok(BrowseResult::Containers(Vec::new())); - } - let end = if limit == 0 { - total - } else { - (offset + limit).min(total) - }; - Ok(BrowseResult::Containers( - containers - .into_iter() - .skip(offset) - .take(end - offset) - .collect(), - )) - } - _ => { - if let Some(channel_id) = parse_channel_container_id(object_id) { - let descriptor = ALL_CHANNELS - .iter() - .find(|desc| desc.id == channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - - let items = self.channel_items(*descriptor, offset, Some(limit)).await?; - Ok(BrowseResult::Items(items)) - } else { - Err(MusicSourceError::ObjectNotFound(object_id.to_string())) - } - } - } - } - - async fn get_item_count(&self, object_id: &str) -> Result { - match object_id { - "0" => Ok(1), - "radio-paradise" => Ok(ALL_CHANNELS.len()), - _ => { - if let Some(channel_id) = parse_channel_container_id(object_id) { - let channel = self - .channel(channel_id) - .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?; - Ok(channel.playlist().active_len().await) - } else { - Err(MusicSourceError::ObjectNotFound(object_id.to_string())) - } - } - } - } - - async fn statistics(&self) -> Result { - let mut total_tracks = 0usize; - let mut cached_tracks = 0usize; - - for (_, channel) in self.channels_iter() { - total_tracks += channel.playlist().active_len().await; - let stats = channel.cache_manager().statistics().await; - cached_tracks += stats.cached_tracks; - } - - Ok(SourceStatistics { - total_items: Some(total_tracks), - total_containers: Some(ALL_CHANNELS.len() + 1), - cached_items: Some(cached_tracks), - cache_size_bytes: None, - }) + async fn get_items(&self, _offset: usize, _count: usize) -> Result> { + Ok(vec![]) } }