#[cfg(test)]
use std::sync::OnceLock;
use std::{
    fmt,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
};

use bon::Builder;
use kithara_audio::{AudioWorkerHandle, EqBandConfig, generate_log_spaced_bands};
use kithara_bufpool::PcmPool;
use kithara_events::EventBus;
use kithara_platform::{Mutex, tokio::runtime::Handle as RuntimeHandle};
use portable_atomic::AtomicF32;
use ringbuf::{HeapProd, traits::Producer};
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};

use super::{
    arena_registry::ArenaRegistry,
    player_processor::PlayerCmd,
    session::{PlayerId, SessionDispatcher, session_client},
    shared_eq::SharedEq,
    shared_player_state::SharedPlayerState,
};
#[rustfmt::skip]
use crate::traits::dj::crossfade::CrossfadeConfig;
#[rustfmt::skip]
use crate::traits::engine::Engine;
use crate::{
    error::PlayError,
    events::EngineEvent,
    types::{SessionDuckingMode, SlotId},
};

/// Configuration for the audio engine.
#[derive(Clone, Builder)]
#[builder(state_mod(vis = "pub"))]
#[non_exhaustive]
pub struct EngineConfig {
    /// Master cancel token for the engine.
    pub cancel: Option<CancellationToken>,
    /// PCM buffer pool for audio-thread scratch buffers.
    pub pcm_pool: Option<PcmPool>,
    /// Pre-built audio session dispatcher.
    pub session: Option<Arc<dyn SessionDispatcher>>,
    /// EQ band layout per player. Default: 10-band log-spaced.
    #[builder(default = generate_log_spaced_bands(10))]
    pub eq_layout: Vec<EqBandConfig>,
    /// Number of output channels. Default: 2 (stereo).
    #[builder(default = 2)]
    pub channels: u16,
    /// Sample rate passed to the runtime backend as a hint. Default: 44100.
    #[builder(default = 44100)]
    pub sample_rate: u32,
    /// Maximum number of concurrent player slots. Default: 4.
    #[builder(default = 4)]
    pub max_slots: usize,
}

impl fmt::Debug for EngineConfig {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("EngineConfig")
            .field("eq_layout", &self.eq_layout)
            .field("channels", &self.channels)
            .field("sample_rate", &self.sample_rate)
            .field("max_slots", &self.max_slots)
            .field("pcm_pool", &self.pcm_pool)
            .finish_non_exhaustive()
    }
}

impl Default for EngineConfig {
    fn default() -> Self {
        Self::builder().build()
    }
}

/// Handle for a slot, providing command channel and shared state.
pub(crate) struct SlotHandle {
    pub(crate) shared_state: Arc<SharedPlayerState>,
    pub(crate) cmd_tx: HeapProd<PlayerCmd>,
    pub(crate) eq: SharedEq,
}

/// Concrete [`Engine`] implementation backed by a process-wide session.
///
/// Multiple `EngineImpl` instances share one CPAL/Firewheel stream while
/// retaining independent per-player graph controls.
pub struct EngineImpl {
    /// Audio session backend. Production paths default to the
    /// process-wide cpal-backed `SessionClient`; tests inject a
    /// per-instance offline dispatcher through [`EngineConfig::session`].
    session: Arc<dyn SessionDispatcher>,

    /// Whether this engine/player instance is currently running.
    running: AtomicBool,

    /// Master output volume for this player instance (linear 0.0 ..= 1.0).
    master_volume: AtomicF32,

    /// Shared audio worker for cooperative multi-track decoding.
    ///
    /// All tracks loaded by this engine share this single worker thread.
    worker: AudioWorkerHandle,

    config: EngineConfig,

    /// Shared event bus (passed from `PlayerImpl`).
    bus: EventBus,

    /// Per-slot tracking (owned by the main side, mirrored).
    active_slots: Mutex<Vec<SlotId>>,

    /// Session player ID allocated lazily on first start.
    player_id: Mutex<Option<PlayerId>>,

    /// Per-slot command channels and shared state.
    slot_registry: Mutex<ArenaRegistry<SlotId, SlotHandle>>,

    runtime: Option<RuntimeHandle>,
    /// Resolved PCM pool used when registering this player in the session.
    pcm_pool: PcmPool,
}

impl EngineImpl {
    /// Create a new engine with the given configuration.
    ///
    /// If `config.session` is `Some(_)` the engine binds to the supplied
    /// dispatcher (used by integration tests for offline render). If
    /// `None`, it falls back to the process-wide cpal-backed session
    /// client.
    ///
    /// The engine is created in the *stopped* state. Call [`Engine::start`]
    /// to begin audio processing.
    #[must_use]
    pub fn new(mut config: EngineConfig, bus: EventBus) -> Self {
        let session = config
            .session
            .take()
            .unwrap_or_else(|| session_client() as Arc<dyn SessionDispatcher>);
        Self::with_session(config, bus, session)
    }

    fn emit(&self, event: EngineEvent) {
        self.bus.publish(event);
    }

    fn ensure_player_id(&self) -> Result<PlayerId, PlayError> {
        let mut player_id = self.player_id.lock_sync();
        if let Some(id) = *player_id {
            return Ok(id);
        }

        let id = self
            .session
            .register_player(self.config.eq_layout.clone(), self.pcm_pool.clone())?;
        *player_id = Some(id);
        drop(player_id);
        Ok(id)
    }

    /// Runtime handle captured at engine creation.
    ///
    /// Use when building a shared
    /// [`Downloader`](kithara_stream::dl::Downloader) so its async tasks
    /// land on the same runtime the audio engine observes, then pass the
    /// downloader through [`ResourceConfig::with_downloader`](super::config::ResourceConfig::with_downloader).
    #[must_use]
    pub fn runtime(&self) -> Option<&RuntimeHandle> {
        self.runtime.as_ref()
    }

    pub(crate) fn send_slot_cmd(&self, slot: SlotId, cmd: PlayerCmd) -> Result<(), PlayError> {
        let mut slot_registry = self.slot_registry.lock_sync();
        let result = match slot_registry.get_mut(&slot) {
            Some(handle) => handle
                .cmd_tx
                .try_push(cmd)
                .map_err(|_| PlayError::Internal("slot channel full".into())),
            None => Err(PlayError::Internal("slot handle not found".into())),
        };
        drop(slot_registry);
        result
    }

    /// Process-wide session ducking mode.
    #[must_use]
    pub fn session_ducking() -> SessionDuckingMode {
        match session_client().ducking() {
            Ok(mode) => mode,
            Err(err) => {
                warn!(?err, "failed to query session ducking");
                SessionDuckingMode::Off
            }
        }
    }

    pub(crate) fn set_master_eq_gain(&self, band: usize, gain_db: f32) -> Result<(), PlayError> {
        let player_id = (*self.player_id.lock_sync()).ok_or(PlayError::EngineNotRunning)?;
        self.session.set_player_eq_gain(player_id, band, gain_db)
    }

    /// Set process-wide session ducking mode.
    pub fn set_session_ducking(mode: SessionDuckingMode) -> Result<(), PlayError> {
        session_client().set_ducking(mode)
    }

    pub(crate) fn set_slot_volume(&self, slot: SlotId, volume: f32) -> Result<(), PlayError> {
        let player_id = (*self.player_id.lock_sync()).ok_or(PlayError::EngineNotRunning)?;
        self.session
            .set_player_slot_volume(player_id, slot, volume.clamp(0.0, 1.0))
    }

    pub(crate) fn slot_eq(&self, slot: SlotId) -> Option<SharedEq> {
        self.slot_registry
            .lock_sync()
            .get(&slot)
            .map(|h| h.eq.clone())
    }

    pub(crate) fn slot_shared_state(&self, slot: SlotId) -> Option<Arc<SharedPlayerState>> {
        self.slot_registry
            .lock_sync()
            .get(&slot)
            .map(|h| Arc::clone(&h.shared_state))
    }

    pub(crate) fn tick(&self) -> Result<(), PlayError> {
        self.session.tick()
    }

    fn with_session(
        config: EngineConfig,
        bus: EventBus,
        session: Arc<dyn SessionDispatcher>,
    ) -> Self {
        let max_slots = config.max_slots;
        let resolved_pool = config
            .pcm_pool
            .clone()
            .unwrap_or_else(|| PcmPool::default().clone());
        let worker_cancel = config.cancel.clone().unwrap_or_default().child_token();

        Self {
            config,
            bus,
            session,
            active_slots: Mutex::new(Vec::new()),
            master_volume: AtomicF32::new(1.0),
            pcm_pool: resolved_pool,
            player_id: Mutex::new(None),
            running: AtomicBool::new(false),
            slot_registry: Mutex::new(ArenaRegistry::with_capacity(max_slots)),
            worker: AudioWorkerHandle::with_cancel(worker_cancel),
            runtime: RuntimeHandle::try_current().ok(),
        }
    }

    /// Shared audio worker handle for this engine.
    ///
    /// Clone and pass to [`ResourceConfig::with_worker`] so all tracks
    /// loaded through this engine share a single decode thread.
    #[must_use]
    pub fn worker(&self) -> &AudioWorkerHandle {
        &self.worker
    }
}

impl Drop for EngineImpl {
    fn drop(&mut self) {
        let player_id = *self.player_id.lock_sync();
        if let Some(player_id) = player_id
            && let Err(err) = self.session.unregister_player(player_id)
        {
            warn!(
                ?err,
                player_id, "failed to unregister player from shared session"
            );
        }

        self.worker.shutdown();
    }
}

impl Engine for EngineImpl {
    fn active_slots(&self) -> Vec<SlotId> {
        self.active_slots.lock_sync().clone()
    }

    fn allocate_slot(&self) -> Result<SlotId, PlayError> {
        if !self.running.load(Ordering::Acquire) {
            return Err(PlayError::EngineNotRunning);
        }

        {
            let slots = self.active_slots.lock_sync();
            if slots.len() >= self.config.max_slots {
                return Err(PlayError::ArenaFull);
            }
        }

        let player_id = (*self.player_id.lock_sync()).ok_or(PlayError::EngineNotRunning)?;
        let (slot_id, cmd_tx, shared_state, eq) = self.session.allocate_slot(player_id)?;

        self.active_slots.lock_sync().push(slot_id);
        self.slot_registry.lock_sync().insert(
            slot_id,
            SlotHandle {
                shared_state,
                cmd_tx,
                eq,
            },
        );

        debug!(?slot_id, player_id, "slot allocated");
        self.emit(EngineEvent::SlotAllocated { slot: slot_id });
        Ok(slot_id)
    }

    fn cancel_crossfade(&self) -> Result<(), PlayError> {
        Err(PlayError::NoCrossfade)
    }

    fn crossfade(
        &self,
        _from: SlotId,
        _to: SlotId,
        _config: CrossfadeConfig,
    ) -> Result<(), PlayError> {
        Err(PlayError::NotReady)
    }

    fn is_crossfading(&self) -> bool {
        false
    }

    fn is_running(&self) -> bool {
        self.running.load(Ordering::Acquire)
    }

    fn master_channels(&self) -> u16 {
        self.config.channels
    }

    fn master_sample_rate(&self) -> u32 {
        Self::master_sample_rate(self)
    }

    fn master_volume(&self) -> f32 {
        self.master_volume.load(Ordering::Relaxed)
    }

    fn max_slots(&self) -> usize {
        self.config.max_slots
    }

    fn release_slot(&self, slot: SlotId) -> Result<(), PlayError> {
        if !self.running.load(Ordering::Acquire) {
            return Err(PlayError::EngineNotRunning);
        }

        {
            let slots = self.active_slots.lock_sync();
            if !slots.contains(&slot) {
                return Err(PlayError::SlotNotFound(slot));
            }
        }

        let player_id = (*self.player_id.lock_sync()).ok_or(PlayError::EngineNotRunning)?;
        self.session.release_slot(player_id, slot)?;

        self.active_slots.lock_sync().retain(|s| *s != slot);
        let _ = self.slot_registry.lock_sync().remove(&slot);

        debug!(?slot, player_id, "slot released");
        self.emit(EngineEvent::SlotReleased { slot });
        Ok(())
    }

    fn set_master_volume(&self, volume: f32) {
        let clamped = volume.clamp(0.0, 1.0);
        self.master_volume.store(clamped, Ordering::Relaxed);

        if self.running.load(Ordering::Acquire)
            && let Some(player_id) = *self.player_id.lock_sync()
            && let Err(err) = self.session.set_player_master_volume(player_id, clamped)
        {
            warn!(
                ?err,
                player_id,
                volume = clamped,
                "failed to apply player master volume"
            );
        }

        self.emit(EngineEvent::MasterVolumeChanged { volume: clamped });
    }

    fn slot_count(&self) -> usize {
        self.active_slots.lock_sync().len()
    }

    fn start(&self) -> Result<(), PlayError> {
        if self.running.load(Ordering::Acquire) {
            return Err(PlayError::EngineAlreadyRunning);
        }

        let player_id = self.ensure_player_id()?;
        let master_volume = self.master_volume.load(Ordering::Relaxed);
        self.session
            .start_player(player_id, self.config.sample_rate, master_volume)?;

        self.running.store(true, Ordering::Release);

        info!(
            sample_rate = self.config.sample_rate,
            channels = self.config.channels,
            max_slots = self.config.max_slots,
            player_id,
            "engine started"
        );
        self.emit(EngineEvent::Started);
        Ok(())
    }

    fn stop(&self) -> Result<(), PlayError> {
        if !self.running.load(Ordering::Acquire) {
            return Err(PlayError::EngineNotRunning);
        }

        let player_id = (*self.player_id.lock_sync()).ok_or(PlayError::EngineNotRunning)?;
        self.session.stop_player(player_id)?;

        self.active_slots.lock_sync().clear();
        self.slot_registry.lock_sync().clear();

        self.running.store(false, Ordering::Release);
        info!(player_id, "engine stopped");
        self.emit(EngineEvent::Stopped);
        Ok(())
    }

    fn subscribe(&self) -> kithara_events::EventReceiver {
        self.bus.subscribe()
    }
}

impl EngineImpl {
    /// Effective sample rate of the audio host (from Firewheel / `CoreAudio`).
    ///
    /// Returns the config default if the engine is not running yet.
    /// Used to pre-initialise the resampler in `ResourceConfig` so that
    /// `make_sincs` runs during `Audio::new()` (off the worker thread)
    /// instead of lazily on the first `step_track()` call.
    pub fn master_sample_rate(&self) -> u32 {
        if !self.running.load(Ordering::Acquire) {
            return self.config.sample_rate;
        }
        self.session.query_sample_rate(self.config.sample_rate)
    }
}

#[cfg(test)]
pub(crate) fn ducking_test_lock() -> &'static Mutex<()> {
    static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| Mutex::new(()))
}

#[cfg(test)]
mod tests {
    use kithara_test_utils::kithara;

    use super::*;

    #[kithara::test]
    fn engine_creates_worker() {
        let engine = EngineImpl::new(EngineConfig::default(), EventBus::default());
        let _w = engine.worker();
    }

    #[kithara::test]
    fn engine_worker_is_clonable() {
        let engine = EngineImpl::new(EngineConfig::default(), EventBus::default());
        let w1 = engine.worker().clone();
        let w2 = engine.worker().clone();
        w1.wake();
        w2.wake();
    }

    #[kithara::test]
    fn engine_drop_shuts_down_worker() {
        let engine = EngineImpl::new(EngineConfig::default(), EventBus::default());
        let worker_clone = engine.worker().clone();
        drop(engine);
        worker_clone.wake();
    }
}
