use crate::cli::WaitStrategyOpt;
use crate::control_plane::{ControlPlane, EngineCtlMsg, EngineInstance};
use crate::projection::{PhaseCallback, ProjectionHandler, StatusCallback};
use crate::state::{LiveState, MarketMeta, ProjectionDiskSnapshot, ProjectionSnapshot};
use crate::ws_protocol::{MarketInfo, MarketStatus};
use crate::{PROJECTION_PATH, SNAPSHOT_FREQUENCY, WAL_PATH};
use betex::{EngineBuilder, MarketConfig, engine::consumers::ResponseCallbackHandler, prelude::*};
use disrupt_rs::wait_strategies::BlockingWaitStrategy;
use parking_lot::RwLock;
use std::{collections::HashMap, path::PathBuf, sync::Arc, time::Duration};
use uuid::Uuid;

pub(crate) fn engine_paths(engine_id: Uuid, default_engine_id: Uuid) -> (PathBuf, PathBuf) {
    if engine_id == default_engine_id {
        return (PathBuf::from(WAL_PATH), PathBuf::from(PROJECTION_PATH));
    }
    let wal_path = PathBuf::from("target/live_engines").join(format!("{engine_id}.lmdb"));
    let projection_path = wal_path.join("projection.json");
    (wal_path, projection_path)
}

#[allow(clippy::too_many_arguments)]
pub(crate) fn build_engine_instance(
    engine_id: Uuid,
    cfg: betex::config::Config,
    market_configs: Vec<MarketConfig>,
    wal_path: PathBuf,
    projection_path: PathBuf,
    status_callback: Option<StatusCallback>,
    phase_callback: Option<PhaseCallback>,
    market_id_counter: Arc<std::sync::atomic::AtomicU64>,
    wait_strategy: WaitStrategyOpt,
) -> anyhow::Result<Arc<EngineInstance>> {
    let wal_preexists = wal_path.exists();
    std::fs::create_dir_all(&wal_path)?;

    // Shared projection snapshot for queries
    let projection_snapshot = Arc::new(RwLock::new(ProjectionSnapshot::default()));
    let live_state = Arc::new(RwLock::new(LiveState::default()));
    let market_meta: Arc<RwLock<HashMap<MarketId, MarketMeta>>> =
        Arc::new(RwLock::new(HashMap::new()));
    let final_disk_snapshot: Arc<parking_lot::Mutex<Option<ProjectionDiskSnapshot>>> =
        Arc::new(parking_lot::Mutex::new(None));

    // Load from Disk Snapshot
    let disk_snapshot = ProjectionDiskSnapshot::load_postcard(&projection_path)?;

    let market_ids_from_disk = disk_snapshot.as_ref().map(|s| s.market_ids.clone());
    let market_ids_from_cfg: Vec<MarketId> = market_configs.iter().map(|c| c.market_id()).collect();
    // If the disk snapshot and config disagree on the market set, prefer config and do a full WAL replay.
    // This enables adding markets to a single engine without introducing new WAL event types.
    // Important: on normal restart, `market_ids_from_cfg` may be empty (engine starts "with no markets"),
    // but we still must recover markets from the persisted snapshot. Only force replay when the cfg is
    // non-empty and differs from the snapshot.
    let use_disk_snapshot = disk_snapshot.is_some()
        && (market_ids_from_cfg.is_empty()
            || market_ids_from_disk
                .as_ref()
                .is_some_and(|ids| *ids == market_ids_from_cfg));
    let served_market_ids = if market_ids_from_cfg.is_empty() {
        market_ids_from_disk.unwrap_or_default()
    } else if use_disk_snapshot {
        market_ids_from_disk.unwrap_or_else(|| market_ids_from_cfg.clone())
    } else {
        market_ids_from_cfg.clone()
    };
    let market_ids = Arc::new(RwLock::new(served_market_ids.clone()));

    let (projection_handler, initial_engine_root) = if use_disk_snapshot {
        if let Some(disk_snap) = disk_snapshot {
            println!(
                "[live][engine={engine_id}] Loaded disk snapshot: seq={} with {} market(s)",
                disk_snap.global_seq(),
                disk_snap.market_ids.len()
            );

            let engine_root = disk_snap.root.clone();
            let projection_handler = ProjectionHandler::from_disk_snapshot(
                disk_snap,
                Arc::clone(&market_ids),
                Arc::clone(&projection_snapshot),
                Arc::clone(&live_state),
                Arc::clone(&final_disk_snapshot),
                projection_path.clone(),
                SNAPSHOT_FREQUENCY,
                status_callback,
                phase_callback,
                Arc::clone(&market_meta),
                Arc::clone(&market_id_counter),
            );

            (projection_handler, Some(engine_root))
        } else {
            println!("[live][engine={engine_id}] No disk snapshot; starting fresh");
            let projection_handler = ProjectionHandler::new(
                cfg.clone(),
                market_configs.clone(),
                Arc::clone(&market_ids),
                Arc::clone(&projection_snapshot),
                Arc::clone(&live_state),
                Arc::clone(&final_disk_snapshot),
                projection_path.clone(),
                SNAPSHOT_FREQUENCY,
                status_callback,
                phase_callback,
                Arc::clone(&market_meta),
                Arc::clone(&market_id_counter),
            );
            (projection_handler, None)
        }
    } else {
        if disk_snapshot.is_some() {
            println!(
                "[live][engine={engine_id}] Disk snapshot market set differs from config; ignoring snapshot and replaying WAL"
            );
        }
        println!("[live][engine={engine_id}] No disk snapshot; starting fresh");
        let projection_handler = ProjectionHandler::new(
            cfg.clone(),
            market_configs.clone(),
            Arc::clone(&market_ids),
            Arc::clone(&projection_snapshot),
            Arc::clone(&live_state),
            Arc::clone(&final_disk_snapshot),
            projection_path.clone(),
            SNAPSHOT_FREQUENCY,
            status_callback,
            phase_callback,
            Arc::clone(&market_meta),
            Arc::clone(&market_id_counter),
        );
        (projection_handler, None)
    };

    let is_recovering = initial_engine_root.is_some() || wal_preexists;

    // Start Engine
    let markets: Vec<MarketConfig> = if served_market_ids == market_ids_from_cfg {
        market_configs
    } else {
        // Recovery path: rely on disk snapshot / WAL for the authoritative market definitions.
        // We still need *some* config to build the engine; use dynamic multi-runner placeholders.
        served_market_ids
            .iter()
            .copied()
            .map(|mid| {
                MarketConfig::multi_runner_dynamic(
                    mid,
                    format!("Market {}", mid.0),
                    MarketKind::InPlayCapable,
                )
            })
            .collect()
    };

    let projection_recovery = HandlerRecovery::AfterSeq(
        initial_engine_root
            .as_ref()
            .map(|root| root.version())
            .unwrap_or(0),
    );
    let builder = match initial_engine_root {
        Some(root) => EngineBuilder::from_root(root, cfg.clone(), &wal_path)?,
        None => EngineBuilder::with_markets(cfg.clone(), markets, &wal_path)?,
    };

    let engine = match wait_strategy {
        WaitStrategyOpt::BusySpin => builder
            .register_handler(
                ResponseCallbackHandler::default(),
                HandlerRecovery::LiveOnly,
                None,
            )
            .register_handler(projection_handler, projection_recovery, None)
            .build()?,
        WaitStrategyOpt::BlockingWait => builder
            .with_wait_strategy(BlockingWaitStrategy::default())
            .register_handler(
                ResponseCallbackHandler::default(),
                HandlerRecovery::LiveOnly,
                None,
            )
            .register_handler(projection_handler, projection_recovery, None)
            .build()?,
    };

    if is_recovering {
        println!("[live][engine={engine_id}] Restored from prior state");
    } else {
        println!("[live][engine={engine_id}] Fresh start");
    }

    let handler = engine.handle();
    Ok(Arc::new(EngineInstance {
        id: engine_id,
        engine: parking_lot::Mutex::new(Some(engine)),
        handler,
        projection_snapshot,
        live_state,
        market_ids,
        market_meta,
    }))
}

pub(crate) async fn run_engine_ctl_loop(
    control: ControlPlane,
    ctl_rx: flume::Receiver<EngineCtlMsg>,
    default_engine_id: Uuid,
) {
    while let Ok(msg) = ctl_rx.recv_async().await {
        match msg {
            EngineCtlMsg::CreateMarket {
                name,
                market_model,
                market_kind,
                market_phase,
                runner_ids,
                runner_labels,
                resp,
            } => {
                // Allocate a new unique market ID
                let market_id = control.inner.market_allocator.next();
                let now_ms = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_millis() as i64;

                match market_model {
                    MarketModel::ExchangeOdds => {
                        if runner_ids.len() < 2 {
                            let _ = resp.send(Err("runner_ids must have length >= 2".into()));
                            continue;
                        }
                    }
                    MarketModel::BinaryYes { .. } => {
                        if runner_ids.len() != 2 {
                            let _ =
                                resp.send(Err("BinaryYes requires exactly 2 runner_ids".into()));
                            continue;
                        }
                    }
                }
                if runner_ids.len() != runner_labels.len() {
                    let _ = resp.send(Err("runner_ids and runner_labels length mismatch".into()));
                    continue;
                }

                let engine_id = default_engine_id;
                let Some(engine) = control.get_engine(engine_id) else {
                    let _ = resp.send(Err("engine not found".into()));
                    continue;
                };

                // Persist market creation as a journaled engine event.
                let cmd = betex::book::protocol::command::Command {
                    correlation_id: None,
                    metadata: None,
                    market_id,
                    kind: betex::book::protocol::command::CommandKind::CreateMarket {
                        name: name.clone(),
                        market_model,
                        market_kind,
                        market_phase,
                        runner_ids: runner_ids.clone(),
                        runner_labels: runner_labels.clone(),
                    },
                };

                let r = match engine.handler.submit_with_response(cmd) {
                    Ok(rx) => {
                        tokio::task::spawn_blocking(move || rx.recv_timeout(Duration::from_secs(2)))
                            .await
                            .map_err(|e| format!("join error: {e}"))
                            .and_then(|x| x.map_err(|_| "engine response timeout".to_string()))
                            .and_then(|res| res.map_err(|e| format!("{e}")))
                            .and_then(|resp| match resp {
                                betex::book::protocol::response::Response::Rejected {
                                    reason,
                                    ..
                                } => Err(format!("{:?}", reason)),
                                betex::book::protocol::response::Response::Fatal { .. } => {
                                    Err("fatal".into())
                                }
                                betex::book::protocol::response::Response::Ok { .. } => Ok(()),
                            })
                    }
                    Err(e) => Err(format!("{e}")),
                };

                if let Err(e) = r {
                    let _ = resp.send(Err(e));
                    continue;
                }

                // Ensure the new market is served by the live projection/WS snapshot loops.
                {
                    let mut ids = engine.market_ids.write();
                    if !ids.contains(&market_id) {
                        ids.push(market_id);
                    }
                }

                let info = MarketInfo {
                    market_id,
                    name: name.clone(),
                    engine_id,
                    market_model,
                    market_kind,
                    market_phase,
                    runner_ids,
                    runner_labels,
                    status: MarketStatus::Active,
                    created_at: now_ms,
                };
                control
                    .inner
                    .market_infos
                    .write()
                    .insert(market_id, info.clone());

                println!(
                    "[live][engine={engine_id}] Created market: {} (id={}, runners={})",
                    name,
                    market_id.0,
                    info.runner_ids.len()
                );

                let _ = resp.send(Ok(info));
            }
            EngineCtlMsg::RemoveMarket { market_id, resp } => {
                // Get market info to find the engine_id
                let market_info = control.inner.market_infos.read().get(&market_id).cloned();

                let Some(info) = market_info else {
                    let _ = resp.send(Err("market not found".into()));
                    continue;
                };

                // Check if market is in terminal state
                if !info.status.is_terminal() {
                    let _ = resp.send(Err("market not in terminal state".into()));
                    continue;
                }

                let engine_uuid = info.engine_id;

                // Single-engine model: detaching a market removes it from the served set + read model,
                // but keeps WAL history (full removal requires WAL compaction/rotation).
                let Some(engine) = control.get_engine(engine_uuid) else {
                    let _ = resp.send(Err("engine not found".into()));
                    continue;
                };

                // Remove from served set.
                {
                    let mut ids = engine.market_ids.write();
                    ids.retain(|m| *m != market_id);
                }

                // Purge read-model state so memory can be reclaimed.
                {
                    let mut live = engine.live_state.write();
                    live.markets.remove(&market_id);
                }
                {
                    let mut snap = engine.projection_snapshot.write();
                    snap.markets.retain(|m| m.market_id != market_id);
                }
                {
                    let mut meta = engine.market_meta.write();
                    meta.remove(&market_id);
                }

                control.inner.market_infos.write().remove(&market_id);

                println!(
                    "[live][engine={engine_uuid}] Detached market: id={}",
                    market_id.0
                );

                let _ = resp.send(Ok(()));
            }
            EngineCtlMsg::ListMarkets { resp } => {
                let markets: Vec<MarketInfo> = control
                    .inner
                    .market_infos
                    .read()
                    .values()
                    .cloned()
                    .collect();
                let _ = resp.send(markets);
            }
        }
    }
}
