//! Version simplifiée de stream_channel.rs utilisant RadioParadisePlaylistFeeder + PlaylistSource
//!
//! Cette version remplace l'architecture complexe RadioParadiseStreamSource par :
//! - RadioParadisePlaylistFeeder : télécharge les URLs gapless et alimente une playlist
//! - PlaylistSource::with_history() : lit la playlist et gère l'historique automatiquement

use std::{
    collections::HashMap,
    pin::Pin,
    sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
    task::{Context, Poll},
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};

use crate::{
    channels::{ChannelDescriptor, ParadiseChannelKind, ALL_CHANNELS},
    client::RadioParadiseClient,
    models::{Block, EventId},
    playlist_feeder::RadioParadisePlaylistFeeder,
};
use anyhow::{anyhow, Context as AnyhowContext, Result};
use once_cell::sync::OnceCell;
use pmoaudio::{AudioError, AudioPipelineNode};
use pmoaudio_ext::{
    FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, OggFlacStreamHandle,
    PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, StreamingSinkOptions,
    TrackBoundaryCoverNode,
};
use pmoaudiocache::{get_audio_cache, Cache as AudioCache};
use pmocovers::{get_cover_cache, Cache as CoverCache};
use pmoflac::EncoderOptions;
use pmoplaylist::PlaylistManager;
use thiserror::Error;
use tokio::io::{AsyncRead, ReadBuf};
use tokio::sync::{Mutex, Notify};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::{error, info, warn};

/// Configuration pour un canal Radio Paradise.
#[derive(Clone, Debug)]
pub struct ParadiseStreamChannelConfig {
    /// Durée maximale (en secondes) d'avance acceptée par le broadcast.
    pub max_lead_seconds: f64,
    /// Options pour le flux FLAC pur.
    pub flac_options: StreamingSinkOptions,
    /// Options pour le flux OGG-FLAC.
    pub ogg_options: StreamingSinkOptions,
    /// URL de base du serveur (pour les métadonnées, covers...)
    pub server_base_url: Option<String>,
}

impl Default for ParadiseStreamChannelConfig {
    fn default() -> Self {
        Self {
            max_lead_seconds: 3.0, // Compromis live/fluidité : assez pour absorber les transitions
            flac_options: StreamingSinkOptions::flac_defaults(),
            ogg_options: StreamingSinkOptions::ogg_defaults(),
            server_base_url: None,
        }
    }
}

/// Options pour activer l'archivage/historique d'un canal.
pub struct ParadiseHistoryOptions {
    pub audio_cache: Arc<AudioCache>,
    pub cover_cache: Arc<CoverCache>,
    pub playlist_id: String,
    pub collection: Option<String>,
    pub replay_max_lead_seconds: f64,
    pub max_history_tracks: Option<usize>,
}

/// Builder pratique pour configurer automatiquement les playlists historiques.
#[derive(Clone)]
pub struct ParadiseHistoryBuilder {
    pub audio_cache: Arc<AudioCache>,
    pub cover_cache: Arc<CoverCache>,
    pub playlist_prefix: String,
    pub playlist_title_prefix: Option<String>,
    pub max_history_tracks: Option<usize>,
    pub collection_prefix: Option<String>,
    pub replay_max_lead_seconds: f64,
}

impl ParadiseHistoryBuilder {
    pub fn new(audio_cache: Arc<AudioCache>, cover_cache: Arc<CoverCache>) -> Self {
        Self {
            audio_cache,
            cover_cache,
            playlist_prefix: "radio-paradise-history".into(),
            playlist_title_prefix: Some("Radio Paradise History".into()),
            max_history_tracks: Some(500),
            collection_prefix: Some("radio-paradise".into()),
            replay_max_lead_seconds: 3.0, // Aligné avec le live
        }
    }

    pub async fn build_for_channel(
        &self,
        descriptor: &ChannelDescriptor,
    ) -> Result<ParadiseHistoryOptions, pmoplaylist::Error> {
        let playlist_id = format!("{}-{}", self.playlist_prefix, descriptor.slug);

        let collection = self
            .collection_prefix
            .as_ref()
            .map(|prefix| format!("{}-{}", prefix, descriptor.slug));

        Ok(ParadiseHistoryOptions {
            audio_cache: self.audio_cache.clone(),
            cover_cache: self.cover_cache.clone(),
            playlist_id,
            collection,
            replay_max_lead_seconds: self.replay_max_lead_seconds,
            max_history_tracks: self.max_history_tracks,
        })
    }
}

impl Default for ParadiseHistoryBuilder {
    fn default() -> Self {
        let audio_cache = get_audio_cache()
            .expect("pmoaudiocache::register_audio_cache must be called before using ParadiseHistoryBuilder::default()");
        let cover_cache = get_cover_cache()
            .expect("pmocovers::register_cover_cache must be called before using ParadiseHistoryBuilder::default()");
        Self::new(audio_cache, cover_cache)
    }
}

#[cfg(feature = "pmoconfig")]
impl ParadiseStreamChannelConfig {
    pub fn from_config(cfg: &pmoconfig::Config, channel: ParadiseChannelKind) -> Self {
        use serde_yaml::Value;
        let path = [
            "sources",
            "radio_paradise",
            "channels",
            channel.slug(),
            "max_lead_seconds",
        ];
        match cfg.get_value(&path) {
            Ok(Value::Number(num)) => {
                if let Some(v) = num.as_f64() {
                    Self {
                        max_lead_seconds: v.max(0.1),
                        flac_options: StreamingSinkOptions::flac_defaults(),
                        ogg_options: StreamingSinkOptions::ogg_defaults(),
                        server_base_url: None,
                    }
                } else {
                    let default = Self::default();
                    let _ =
                        cfg.set_value(&path, Value::String(default.max_lead_seconds.to_string()));
                    default
                }
            }
            Ok(Value::String(s)) => {
                if let Ok(v) = s.parse::<f64>() {
                    Self {
                        max_lead_seconds: v.max(0.1),
                        flac_options: StreamingSinkOptions::flac_defaults(),
                        ogg_options: StreamingSinkOptions::ogg_defaults(),
                        server_base_url: None,
                    }
                } else {
                    let default = Self::default();
                    let _ =
                        cfg.set_value(&path, Value::String(default.max_lead_seconds.to_string()));
                    default
                }
            }
            _ => {
                let default = Self::default();
                let _ = cfg.set_value(&path, Value::String(default.max_lead_seconds.to_string()));
                default
            }
        }
    }
}

/// Stream complet (FLAC pur + OGG-FLAC) pour un canal Radio Paradise.
///
/// Version simplifiée utilisant RadioParadisePlaylistFeeder + PlaylistSource
pub struct ParadiseStreamChannel {
    descriptor: ChannelDescriptor,
    state: Arc<ChannelState>,
    pipeline_handle: JoinHandle<()>,
    feeder_handle: JoinHandle<()>,
}

impl ParadiseStreamChannel {
    /// Crée un canal avec client déjà configuré.
    pub async fn with_client(
        descriptor: ChannelDescriptor,
        client: RadioParadiseClient,
        config: ParadiseStreamChannelConfig,
        cover_cache: Option<Arc<CoverCache>>,
        history: Option<ParadiseHistoryOptions>,
    ) -> Result<Self> {
        // Propager server_base_url dans les options pour que les encoders injectent les covers du cache
        let mut config = config;
        if let Some(ref base) = config.server_base_url {
            config.flac_options = config
                .flac_options
                .clone()
                .with_server_base_url(Some(base.clone()));
            config.ogg_options = config
                .ogg_options
                .clone()
                .with_server_base_url(Some(base.clone()));
        }
        let cover_cache = cover_cache
            .or_else(|| history.as_ref().map(|opts| opts.cover_cache.clone()))
            .or_else(|| get_cover_cache());
        let manager = PlaylistManager::get();

        // 1. Créer la playlist live pour ce canal
        let live_playlist_id = format!("radio-paradise-live-{}", descriptor.slug);
        let (feeder, live_read) = if let Some(ref history_opts) = history {
            RadioParadisePlaylistFeeder::new(
                client.clone(),
                history_opts.audio_cache.clone(),
                history_opts.cover_cache.clone(),
                live_playlist_id.clone(),
                history_opts.collection.clone(),
            )
            .await?
        } else {
            // Pas d'historique, on a besoin quand même d'un cache audio basique
            return Err(anyhow!(
                "History options required for now (audio cache needed)"
            ));
        };

        let feeder = Arc::new(feeder);

        // 2. Créer/récupérer la playlist historique si activée
        let history_write = if let Some(ref history_opts) = history {
            let write = manager
                .get_persistent_write_handle(history_opts.playlist_id.clone())
                .await?;

            // Configurer la capacité
            if let Some(capacity) = history_opts.max_history_tracks {
                write.set_capacity(Some(capacity)).await?;
            }

            // Configurer le titre
            let title = format!("Radio Paradise History - {}", descriptor.display_name);
            write.set_title(title).await?;

            Some(Arc::new(write))
        } else {
            None
        };

        // 3. Créer la source playlist avec historique
        let audio_cache = history.as_ref().unwrap().audio_cache.clone();
        let mut source = if let Some(history_write) = history_write.clone() {
            PlaylistSource::with_history(live_read, audio_cache.clone(), history_write)
        } else {
            PlaylistSource::new(live_read, audio_cache.clone())
        };

        // 4. Créer les sinks de broadcast (FLAC + OGG)
        let (flac_sink, stream_handle) = StreamingFlacSink::with_options(
            EncoderOptions::default(),
            16,
            config.max_lead_seconds,
            config.flac_options.clone(),
        );
        let (ogg_sink, ogg_handle) = StreamingOggFlacSink::with_options(
            EncoderOptions::default(),
            16,
            config.max_lead_seconds,
            config.ogg_options.clone(),
        );

        let mut downstream_children: Vec<Box<dyn AudioPipelineNode>> = Vec::new();
        downstream_children.push(Box::new(flac_sink));
        downstream_children.push(Box::new(ogg_sink));

        // 5. Optionnel : ajouter le nœud de cache de covers
        if let Some(cache) = cover_cache {
            let mut cover_node = TrackBoundaryCoverNode::new(cache);
            for child in downstream_children {
                cover_node.register(child);
            }
            source.register(Box::new(cover_node));
        } else {
            for child in downstream_children {
                source.register(child);
            }
        }

        stream_handle.set_auto_stop(false);
        ogg_handle.set_auto_stop(false);

        // 6. Lancer le pipeline audio
        let stop_token = CancellationToken::new();
        let pipeline_stop = stop_token.clone();
        let channel_display_name = descriptor.display_name;

        let state = Arc::new(ChannelState {
            descriptor,
            config,
            client,
            feeder: feeder.clone(),
            stream_handle,
            ogg_handle,
            history_playlist_id: history.map(|h| h.playlist_id),
            history_audio_cache: history_write.map(|_| audio_cache),
            active_clients: AtomicUsize::new(0),
            activity_notify: Notify::new(),
            stop_token,
            current_block: Mutex::new(None),
            prefetch_lock: Mutex::new(()),
        });

        let pipeline_state = state.clone();
        let pipeline_handle = tokio::spawn(async move {
            info!(
                "RadioParadise stream pipeline started for channel {}",
                channel_display_name
            );
            if let Err(e) = Box::new(source).run(pipeline_stop).await {
                error!("Pipeline error for channel {}: {}", channel_display_name, e);
                pipeline_state.handle_pipeline_error(&e).await;
            }
        });

        // 7. Lancer le feeder qui traite les blocs
        let feeder_runner = feeder.clone();
        tokio::spawn(async move {
            if let Err(e) = feeder_runner.run().await {
                error!("RadioParadisePlaylistFeeder error: {}", e);
            }
        });

        // 8. Lancer le scheduler qui enqueue les blocs
        let feeder_state = state.clone();
        let feeder_handle = tokio::spawn(async move {
            feeder_state.run_scheduler().await;
        });

        Ok(Self {
            descriptor,
            state,
            pipeline_handle,
            feeder_handle,
        })
    }

    /// Crée un canal en construisant automatiquement le client pour ce descriptor.
    pub async fn new(
        descriptor: ChannelDescriptor,
        config: ParadiseStreamChannelConfig,
        cover_cache: Option<Arc<CoverCache>>,
        history: Option<ParadiseHistoryOptions>,
    ) -> Result<Self> {
        let client = RadioParadiseClient::builder()
            .channel(descriptor.id)
            .build()
            .await?;
        Self::with_client(descriptor, client, config, cover_cache, history).await
    }

    /// S'abonne au flux FLAC pur.
    pub fn subscribe_flac(&self) -> ChannelFlacStream {
        self.state.on_client_added();
        let inner = self.state.stream_handle.subscribe_flac();
        ChannelFlacStream::new(inner, self.state.clone())
    }

    /// S'abonne au flux FLAC + ICY metadata.
    pub fn subscribe_icy(&self) -> ChannelIcyStream {
        self.state.on_client_added();
        let inner = self.state.stream_handle.subscribe_icy();
        ChannelIcyStream::new(inner, self.state.clone())
    }

    /// S'abonne au flux OGG-FLAC.
    pub fn subscribe_ogg(&self) -> ChannelOggStream {
        self.state.on_client_added();
        let inner = self.state.ogg_handle.subscribe();
        ChannelOggStream::new(inner, self.state.clone())
    }

    /// Snapshot des métadonnées actuelles.
    pub async fn metadata(&self) -> MetadataSnapshot {
        self.state.stream_handle.get_metadata().await
    }

    /// Nombre de clients actifs.
    pub fn active_clients(&self) -> usize {
        self.state.active_clients.load(Ordering::SeqCst)
    }

    pub fn descriptor(&self) -> ChannelDescriptor {
        self.descriptor
    }

    /// Lance un pipeline dédié pour rejouer l'historique (FLAC pur) pour un client.
    pub async fn stream_history_flac(
        &self,
        client_id: &str,
    ) -> Result<HistoryFlacStream, HistoryStreamError> {
        let history_id = self
            .state
            .history_playlist_id
            .as_ref()
            .ok_or(HistoryStreamError::HistoryDisabled)?;

        let audio_cache = self
            .state
            .history_audio_cache
            .as_ref()
            .ok_or(HistoryStreamError::HistoryDisabled)?;

        tracing::info!(
            "Starting historical FLAC replay for channel {} (client_id={})",
            self.descriptor.display_name,
            client_id
        );

        let reader = pmoplaylist::PlaylistManager::get()
            .get_read_handle(history_id)
            .await
            .map_err(|e| HistoryStreamError::Playlist(e.to_string()))?;

        let mut source = PlaylistSource::new(reader, audio_cache.clone());
        let (flac_sink, handle) = StreamingFlacSink::with_max_broadcast_lead(
            EncoderOptions::default(),
            16,
            self.state.config.max_lead_seconds,
        );
        source.register(Box::new(flac_sink));
        let stop_token = CancellationToken::new();
        let stop_clone = stop_token.clone();
        let pipeline = tokio::spawn(async move {
            let _ = Box::new(source).run(stop_clone).await;
        });
        let stream = handle.subscribe_flac();
        Ok(HistoryFlacStream::new(stream, stop_token, pipeline))
    }

    /// Lance un pipeline dédié pour rejouer l'historique (OGG-FLAC) pour un client.
    pub async fn stream_history_ogg(
        &self,
        client_id: &str,
    ) -> Result<HistoryOggStream, HistoryStreamError> {
        let history_id = self
            .state
            .history_playlist_id
            .as_ref()
            .ok_or(HistoryStreamError::HistoryDisabled)?;

        let audio_cache = self
            .state
            .history_audio_cache
            .as_ref()
            .ok_or(HistoryStreamError::HistoryDisabled)?;

        tracing::info!(
            "Starting historical OGG replay for channel {} (client_id={})",
            self.descriptor.display_name,
            client_id
        );

        let reader = pmoplaylist::PlaylistManager::get()
            .get_read_handle(history_id)
            .await
            .map_err(|e| HistoryStreamError::Playlist(e.to_string()))?;

        let mut source = PlaylistSource::new(reader, audio_cache.clone());
        let (ogg_sink, handle) = StreamingOggFlacSink::with_max_broadcast_lead(
            EncoderOptions::default(),
            16,
            self.state.config.max_lead_seconds,
        );
        source.register(Box::new(ogg_sink));
        let stop_token = CancellationToken::new();
        let stop_clone = stop_token.clone();
        let pipeline = tokio::spawn(async move {
            let _ = Box::new(source).run(stop_clone).await;
        });
        let stream = handle.subscribe();
        Ok(HistoryOggStream::new(stream, stop_token, pipeline))
    }
}

impl Drop for ParadiseStreamChannel {
    fn drop(&mut self) {
        self.state.stop_token.cancel();
        self.pipeline_handle.abort();
        self.feeder_handle.abort();
    }
}

const MAX_BLOCK_LEAD: Duration = Duration::from_secs(3600);
const BLOCK_LEAD_CHECK_CHUNK: Duration = Duration::from_secs(300);
const LIVE_PREFETCH_MIN_TRACKS: usize = 5;
const LIVE_PREFETCH_TIMEOUT: Duration = Duration::from_secs(10);
const LIVE_PREFETCH_POLL_INTERVAL: Duration = Duration::from_millis(200);
const LIVE_PREFETCH_MAX_BLOCKS: usize = 4;

static GLOBAL_CHANNEL_MANAGER: OnceCell<std::sync::Weak<ParadiseChannelManager>> = OnceCell::new();

struct ChannelState {
    descriptor: ChannelDescriptor,
    config: ParadiseStreamChannelConfig,
    client: RadioParadiseClient,
    feeder: Arc<RadioParadisePlaylistFeeder>,
    stream_handle: StreamHandle,
    ogg_handle: OggFlacStreamHandle,
    history_playlist_id: Option<String>,
    history_audio_cache: Option<Arc<AudioCache>>,
    active_clients: AtomicUsize,
    activity_notify: Notify,
    stop_token: CancellationToken,
    current_block: Mutex<Option<EventId>>,
    prefetch_lock: Mutex<()>,
}

impl ChannelState {
    fn current_unix_millis() -> u64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0)
    }

    fn block_lead_delay(&self, block: &Block) -> Option<Duration> {
        let start = block.start_time_millis()?;
        let now = Self::current_unix_millis();
        let max_lead_ms = MAX_BLOCK_LEAD.as_millis() as u64;
        if start <= now + max_lead_ms {
            None
        } else {
            Some(Duration::from_millis(start - now - max_lead_ms))
        }
    }

    fn on_client_added(&self) {
        if self.active_clients.fetch_add(1, Ordering::SeqCst) == 0 {
            self.activity_notify.notify_one();
        }
    }

    fn on_client_removed(&self) {
        self.active_clients.fetch_sub(1, Ordering::SeqCst);
    }

    async fn wait_for_clients(&self) -> bool {
        while self.active_clients.load(Ordering::SeqCst) == 0 {
            tokio::select! {
                _ = self.stop_token.cancelled() => return false,
                _ = self.activity_notify.notified() => {},
            }
        }
        true
    }

    async fn wait_until_block_ready(&self, block: &Block) -> BlockReadiness {
        loop {
            if self.stop_token.is_cancelled() {
                return BlockReadiness::Stopped;
            }
            if self.active_clients.load(Ordering::SeqCst) == 0 {
                return BlockReadiness::NoClients;
            }

            if let Some(delay) = self.block_lead_delay(block) {
                let sleep_for = delay.min(BLOCK_LEAD_CHECK_CHUNK);
                let lead_secs = delay.as_secs_f64();
                info!(
                    "Block {} scheduled too far in the future ({:.1} min). Sleeping {:?} before retrying.",
                    block.event,
                    lead_secs / 60.0,
                    sleep_for
                );
                tokio::select! {
                    _ = self.stop_token.cancelled() => return BlockReadiness::Stopped,
                    _ = tokio::time::sleep(sleep_for) => {},
                }
                continue;
            }

            return BlockReadiness::Ready;
        }
    }

    fn live_playlist_id(&self) -> String {
        format!("radio-paradise-live-{}", self.descriptor.slug)
    }

    async fn prefetch_until_horizon(&self) -> Result<()> {
        let _guard = self.prefetch_lock.lock().await;
        let playlist_id = self.live_playlist_id();
        let manager = PlaylistManager::get();
        let reader = manager
            .get_read_handle(&playlist_id)
            .await
            .with_context(|| format!("Failed to get live playlist {}", playlist_id))?;
        let start = Instant::now();
        let mut next_event: Option<EventId> = None;
        let mut attempts = 0usize;

        loop {
            let available = reader
                .remaining()
                .await
                .with_context(|| format!("Failed to inspect playlist {}", playlist_id))?;
            if available >= LIVE_PREFETCH_MIN_TRACKS {
                return Ok(());
            }

            if start.elapsed() >= LIVE_PREFETCH_TIMEOUT {
                warn!(
                    "Prefetch timeout for channel {} ({} tracks available)",
                    self.descriptor.display_name, available
                );
                return Ok(());
            }

            if attempts >= LIVE_PREFETCH_MAX_BLOCKS {
                warn!(
                    "Prefetch block limit reached for channel {} ({} tracks available)",
                    self.descriptor.display_name, available
                );
                return Ok(());
            }

            match self.client.get_block(next_event).await {
                Ok(block) => {
                    attempts += 1;
                    next_event = Some(block.end_event);
                    self.feeder.push_block_id(block.event).await;
                }
                Err(e) => {
                    warn!(
                        "Failed to fetch block during prefetch for channel {}: {}",
                        self.descriptor.display_name, e
                    );
                    return Ok(());
                }
            }

            tokio::time::sleep(LIVE_PREFETCH_POLL_INTERVAL).await;
        }
    }

    async fn set_current_block(&self, event_id: EventId) {
        let mut guard = self.current_block.lock().await;
        *guard = Some(event_id);
    }

    async fn take_current_block(&self) -> Option<EventId> {
        self.current_block.lock().await.take()
    }

    async fn handle_pipeline_error(&self, err: &AudioError) {
        if let Some(event_id) = self.take_current_block().await {
            warn!(
                "Pipeline error while streaming block {} on channel {}: {}. Rescheduling block.",
                event_id, self.descriptor.display_name, err
            );
            self.feeder.retry_block(event_id).await;
        } else {
            warn!(
                "Pipeline error for channel {} but no tracked block: {}",
                self.descriptor.display_name, err
            );
        }
    }

    async fn run_scheduler(self: Arc<Self>) {
        let mut backoff = Duration::from_secs(5);
        'scheduler: loop {
            if self.stop_token.is_cancelled() {
                break;
            }

            if !self.wait_for_clients().await {
                break;
            }

            match self.client.get_block(None).await {
                Ok(block) => {
                    match self.wait_until_block_ready(&block).await {
                        BlockReadiness::Ready => {}
                        BlockReadiness::NoClients => continue,
                        BlockReadiness::Stopped => break,
                    }
                    info!(
                        "Channel {} streaming block {}",
                        self.descriptor.display_name, block.event
                    );
                    self.set_current_block(block.event).await;
                    self.feeder.push_block_id(block.event).await;
                    let mut next_event = block.end_event;

                    loop {
                        if self.stop_token.is_cancelled() {
                            return;
                        }

                        if self.active_clients.load(Ordering::SeqCst) == 0 {
                            break;
                        }

                        match self.client.get_block(Some(next_event)).await {
                            Ok(next_block) => {
                                match self.wait_until_block_ready(&next_block).await {
                                    BlockReadiness::Ready => {}
                                    BlockReadiness::NoClients => break,
                                    BlockReadiness::Stopped => break 'scheduler,
                                }
                                self.set_current_block(next_block.event).await;
                                self.feeder.push_block_id(next_block.event).await;
                                next_event = next_block.end_event;
                                backoff = Duration::from_secs(5);
                            }
                            Err(e) => {
                                warn!(
                                    "Failed to fetch next block for channel {}: {}",
                                    self.descriptor.display_name, e
                                );
                                tokio::select! {
                                    _ = self.stop_token.cancelled() => return,
                                    _ = tokio::time::sleep(backoff) => {},
                                }
                                backoff = (backoff * 2).min(Duration::from_secs(60));
                            }
                        }
                    }
                }
                Err(e) => {
                    warn!(
                        "Failed to fetch current block for channel {}: {}",
                        self.descriptor.display_name, e
                    );
                    tokio::select! {
                        _ = self.stop_token.cancelled() => break,
                        _ = tokio::time::sleep(backoff) => {},
                    }
                    backoff = (backoff * 2).min(Duration::from_secs(60));
                }
            }
        }
    }
}

enum BlockReadiness {
    Ready,
    NoClients,
    Stopped,
}

macro_rules! wrap_stream {
    ($name:ident, $inner:ty) => {
        pub struct $name {
            inner: $inner,
            state: Arc<ChannelState>,
        }

        impl $name {
            fn new(inner: $inner, state: Arc<ChannelState>) -> Self {
                Self { inner, state }
            }
        }

        impl AsyncRead for $name {
            fn poll_read(
                mut self: Pin<&mut Self>,
                cx: &mut Context<'_>,
                buf: &mut ReadBuf<'_>,
            ) -> Poll<std::io::Result<()>> {
                Pin::new(&mut self.inner).poll_read(cx, buf)
            }
        }

        impl Drop for $name {
            fn drop(&mut self) {
                self.state.on_client_removed();
            }
        }
    };
}

wrap_stream!(ChannelFlacStream, FlacClientStream);
wrap_stream!(ChannelIcyStream, IcyClientStream);
wrap_stream!(ChannelOggStream, OggFlacClientStream);

#[derive(Debug, Error)]
pub enum HistoryStreamError {
    #[error("history replay not enabled for this channel")]
    HistoryDisabled,
    #[error("playlist error: {0}")]
    Playlist(String),
}

pub struct HistoryFlacStream {
    inner: FlacClientStream,
    stop_token: CancellationToken,
    pipeline: Option<JoinHandle<()>>,
}

impl HistoryFlacStream {
    fn new(
        inner: FlacClientStream,
        stop_token: CancellationToken,
        pipeline: JoinHandle<()>,
    ) -> Self {
        Self {
            inner,
            stop_token,
            pipeline: Some(pipeline),
        }
    }
}

impl AsyncRead for HistoryFlacStream {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.inner).poll_read(cx, buf)
    }
}

impl Unpin for HistoryFlacStream {}

impl Drop for HistoryFlacStream {
    fn drop(&mut self) {
        self.stop_token.cancel();
        if let Some(handle) = self.pipeline.take() {
            handle.abort();
        }
    }
}

pub struct HistoryOggStream {
    inner: OggFlacClientStream,
    stop_token: CancellationToken,
    pipeline: Option<JoinHandle<()>>,
}

impl HistoryOggStream {
    fn new(
        inner: OggFlacClientStream,
        stop_token: CancellationToken,
        pipeline: JoinHandle<()>,
    ) -> Self {
        Self {
            inner,
            stop_token,
            pipeline: Some(pipeline),
        }
    }
}

impl AsyncRead for HistoryOggStream {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        Pin::new(&mut self.inner).poll_read(cx, buf)
    }
}

impl Unpin for HistoryOggStream {}

impl Drop for HistoryOggStream {
    fn drop(&mut self) {
        self.stop_token.cancel();
        if let Some(handle) = self.pipeline.take() {
            handle.abort();
        }
    }
}

/// Gestionnaire multi-canaux.
pub struct ParadiseChannelManager {
    channels: HashMap<u8, Arc<ParadiseStreamChannel>>,
}

impl ParadiseChannelManager {
    pub fn new(channels: HashMap<u8, Arc<ParadiseStreamChannel>>) -> Self {
        Self { channels }
    }

    pub async fn with_defaults_with_cover_cache(
        cover_cache: Option<Arc<CoverCache>>,
        history_builder: Option<ParadiseHistoryBuilder>,
        server_base_url: Option<String>,
    ) -> Result<Self> {
        tracing::info!(
            "➡️ Entering with_defaults_with_cover_cache ({} channels, base_url={:?})",
            ALL_CHANNELS.len(),
            server_base_url
        );
        let mut map = HashMap::new();
        for descriptor in ALL_CHANNELS.iter().copied() {
            let mut config = ParadiseStreamChannelConfig::default();
            config.server_base_url = server_base_url.clone();

            let start = Instant::now();
            tracing::info!(
                "⏳ Initializing Radio Paradise channel {} ({})...",
                descriptor.display_name,
                descriptor.slug
            );

            let history_opts = if let Some(builder) = &history_builder {
                tracing::debug!(
                    "  ⏳ Building history options for channel {} ({})",
                    descriptor.display_name,
                    descriptor.slug
                );
                Some(
                    builder
                        .build_for_channel(&descriptor)
                        .await
                        .map_err(|e| anyhow!("Failed to init history playlist: {}", e))?,
                )
            } else {
                None
            };
            tracing::debug!(
                "  ⏩ History options ready for channel {} ({})",
                descriptor.display_name,
                descriptor.slug
            );
            let channel = match tokio::time::timeout(
                Duration::from_secs(20),
                ParadiseStreamChannel::new(descriptor, config, cover_cache.clone(), history_opts),
            )
            .await
            {
                Ok(Ok(ch)) => {
                    tracing::info!(
                        "✅ Channel {} ({}) initialized in {:?}",
                        descriptor.display_name,
                        descriptor.slug,
                        start.elapsed()
                    );
                    ch
                }
                Ok(Err(e)) => {
                    tracing::error!(
                        "⚠️ Failed to initialize channel {} ({}): {}",
                        descriptor.display_name,
                        descriptor.slug,
                        e
                    );
                    continue;
                }
                Err(_) => {
                    tracing::error!(
                        "⚠️ Timeout initializing channel {} ({}) after 20s, skipping",
                        descriptor.display_name,
                        descriptor.slug
                    );
                    continue;
                }
            };
            map.insert(descriptor.id, Arc::new(channel));
        }
        Ok(Self { channels: map })
    }

    pub async fn with_defaults() -> Result<Self> {
        Self::with_defaults_with_cover_cache(None, None, None).await
    }

    pub fn get(&self, id: u8) -> Option<Arc<ParadiseStreamChannel>> {
        self.channels.get(&id).cloned()
    }

    pub fn iter(&self) -> impl Iterator<Item = &Arc<ParadiseStreamChannel>> {
        self.channels.values()
    }

    pub async fn prefetch_until_horizon(&self, channel_id: u8) -> Result<()> {
        let channel = self
            .get(channel_id)
            .ok_or_else(|| anyhow!("Unknown channel id {}", channel_id))?;
        channel.prefetch_until_horizon().await
    }
}

pub fn register_global_channel_manager(manager: Arc<ParadiseChannelManager>) {
    let _ = GLOBAL_CHANNEL_MANAGER.set(Arc::downgrade(&manager));
}

pub fn get_global_channel_manager() -> Option<Arc<ParadiseChannelManager>> {
    GLOBAL_CHANNEL_MANAGER.get().and_then(|weak| weak.upgrade())
}

impl ParadiseStreamChannel {
    pub async fn prefetch_until_horizon(&self) -> Result<()> {
        self.state.prefetch_until_horizon().await
    }
}
