use crate::state::MarketIdAllocator;
use crate::ws_protocol::{MarketInfo, ServerMsg};
use betex::{Engine, engine::core::EngineHandler, prelude::*};
use parking_lot::{Mutex, RwLock};
use std::{collections::HashMap, sync::Arc};
use tokio::sync::{broadcast, oneshot, watch};
use uuid::Uuid;

// ==========================================================================
// App State
// ==========================================================================

#[derive(Clone)]
pub(crate) struct AppState {
    pub(crate) control: ControlPlane,
    pub(crate) default_engine_id: Uuid,
    pub(crate) shutdown_rx: watch::Receiver<bool>,
}

#[derive(Clone)]
pub(crate) struct ControlPlane {
    pub(crate) inner: Arc<ControlPlaneInner>,
}

pub(crate) struct ControlPlaneInner {
    pub(crate) engines: RwLock<HashMap<Uuid, Arc<EngineInstance>>>,
    pub(crate) market_infos: RwLock<HashMap<MarketId, MarketInfo>>,
    pub(crate) market_allocator: MarketIdAllocator,
    pub(crate) cfg_unpinned: betex::config::Config,
    pub(crate) ctl_tx: flume::Sender<EngineCtlMsg>,
    /// Broadcast channel for status changes to notify all WebSocket clients
    pub(crate) status_broadcast_tx: broadcast::Sender<ServerMsg>,
}

pub(crate) struct EngineInstance {
    pub(crate) id: Uuid,
    pub(crate) engine: Mutex<Option<Engine>>,
    pub(crate) handler: EngineHandler,
    pub(crate) projection_snapshot: Arc<RwLock<crate::state::ProjectionSnapshot>>,
    pub(crate) live_state: Arc<RwLock<crate::state::LiveState>>,
    pub(crate) market_ids: Arc<RwLock<Vec<MarketId>>>,
    pub(crate) market_meta: Arc<RwLock<HashMap<MarketId, crate::state::MarketMeta>>>,
}

#[derive(Debug)]
pub(crate) enum EngineCtlMsg {
    CreateMarket {
        name: String,
        market_model: MarketModel,
        market_kind: MarketKind,
        market_phase: MarketPhase,
        runner_ids: Vec<RunnerId>,
        runner_labels: Vec<String>,
        resp: oneshot::Sender<Result<MarketInfo, String>>,
    },
    RemoveMarket {
        market_id: MarketId,
        resp: oneshot::Sender<Result<(), String>>,
    },
    ListMarkets {
        resp: oneshot::Sender<Vec<MarketInfo>>,
    },
}

impl ControlPlane {
    pub(crate) fn new(
        cfg_unpinned: betex::config::Config,
    ) -> (Self, flume::Receiver<EngineCtlMsg>) {
        let (ctl_tx, ctl_rx) = flume::bounded::<EngineCtlMsg>(1024);
        let (status_broadcast_tx, _) = broadcast::channel::<ServerMsg>(256);
        (
            Self {
                inner: Arc::new(ControlPlaneInner {
                    engines: RwLock::new(HashMap::new()),
                    market_infos: RwLock::new(HashMap::new()),
                    market_allocator: MarketIdAllocator::new(),
                    cfg_unpinned,
                    ctl_tx,
                    status_broadcast_tx,
                }),
            },
            ctl_rx,
        )
    }

    pub(crate) fn insert_engine(&self, engine: Arc<EngineInstance>) {
        self.inner.engines.write().insert(engine.id, engine);
    }

    pub(crate) fn get_engine(&self, engine_id: Uuid) -> Option<Arc<EngineInstance>> {
        self.inner.engines.read().get(&engine_id).cloned()
    }

    pub(crate) async fn create_market(
        &self,
        name: String,
        market_model: MarketModel,
        market_kind: MarketKind,
        market_phase: MarketPhase,
        runner_ids: Vec<RunnerId>,
        runner_labels: Vec<String>,
    ) -> Result<MarketInfo, String> {
        let (resp_tx, resp_rx) = oneshot::channel();
        self.inner
            .ctl_tx
            .send(EngineCtlMsg::CreateMarket {
                name,
                market_model,
                market_kind,
                market_phase,
                runner_ids,
                runner_labels,
                resp: resp_tx,
            })
            .map_err(|_| "control plane stopped".to_string())?;
        resp_rx
            .await
            .map_err(|_| "control plane dropped response".to_string())?
    }

    pub(crate) async fn remove_market(&self, market_id: MarketId) -> Result<(), String> {
        let (resp_tx, resp_rx) = oneshot::channel();
        self.inner
            .ctl_tx
            .send(EngineCtlMsg::RemoveMarket {
                market_id,
                resp: resp_tx,
            })
            .map_err(|_| "control plane stopped".to_string())?;
        resp_rx
            .await
            .map_err(|_| "control plane dropped response".to_string())?
    }

    pub(crate) async fn list_markets(&self) -> Vec<MarketInfo> {
        let (resp_tx, resp_rx) = oneshot::channel();
        if self
            .inner
            .ctl_tx
            .send(EngineCtlMsg::ListMarkets { resp: resp_tx })
            .is_err()
        {
            return vec![];
        }
        resp_rx.await.unwrap_or_default()
    }

    pub(crate) fn stop_all_blocking(&self) -> anyhow::Result<()> {
        let engines: Vec<_> = self.inner.engines.read().values().cloned().collect();
        for engine in engines {
            let mut guard = engine.engine.lock();
            if let Some(mut e) = guard.take() {
                e.stop()?;
            }
        }
        Ok(())
    }
}
