use crate::state::{
    LiveOrder, LiveState, LiveTrade, MarketMeta, PendingTradeView, ProjectionDiskSnapshot,
    ProjectionSnapshot, TerminalEventKind,
};
use crate::ws_protocol::MarketStatus;
use betex::{
    EngineEvent, MarketConfig,
    book::{BookEventEnvelope, TradeRole},
    config::Config,
    engine::{root::EngineRoot, state::EngineState},
    prelude::*,
    snapshot::MarketSnapshot,
};
use parking_lot::RwLock;
use std::{collections::HashMap, path::PathBuf, sync::Arc, time::Instant};

// ==========================================================================
// Projection Handler
// ==========================================================================

#[derive(Debug, Clone, Copy)]
struct CloseLogState {
    cancelled_total: u64,
}

/// Callback for notifying status changes to the control plane
pub(crate) type StatusCallback = Arc<dyn Fn(MarketId, MarketStatus) + Send + Sync>;
/// Callback for notifying phase changes to the control plane.
pub(crate) type PhaseCallback = Arc<dyn Fn(MarketId, MarketPhase) + Send + Sync>;

pub(crate) struct ProjectionHandler {
    root: EngineRoot,
    market_ids: Arc<RwLock<Vec<MarketId>>>,
    projection_snapshot: Arc<RwLock<ProjectionSnapshot>>,
    live_state: Arc<RwLock<LiveState>>,
    final_disk_snapshot: Arc<parking_lot::Mutex<Option<ProjectionDiskSnapshot>>>,
    projection_path: PathBuf,
    snapshot_frequency: u64,
    last_snapshot_seq: u64,
    pending_seq: u64,
    close_log: HashMap<MarketId, CloseLogState>,
    status_callback: Option<StatusCallback>,
    phase_callback: Option<PhaseCallback>,
    market_meta: Arc<RwLock<HashMap<MarketId, MarketMeta>>>,
    market_id_counter: Arc<std::sync::atomic::AtomicU64>,
}

impl ProjectionHandler {
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        cfg: Config,
        market_configs: Vec<MarketConfig>,
        market_ids: Arc<RwLock<Vec<MarketId>>>,
        projection_snapshot: Arc<RwLock<ProjectionSnapshot>>,
        live_state: Arc<RwLock<LiveState>>,
        final_disk_snapshot: Arc<parking_lot::Mutex<Option<ProjectionDiskSnapshot>>>,
        projection_path: PathBuf,
        snapshot_frequency: u64,
        status_callback: Option<StatusCallback>,
        phase_callback: Option<PhaseCallback>,
        market_meta: Arc<RwLock<HashMap<MarketId, MarketMeta>>>,
        market_id_counter: Arc<std::sync::atomic::AtomicU64>,
    ) -> Self {
        let next_market_id = market_configs
            .iter()
            .map(|mc| mc.market_id().0)
            .max()
            .unwrap_or(0)
            .saturating_add(1);
        let mut state = EngineState::new(cfg);
        for mc in market_configs {
            let mid = mc.market_id();
            state.markets.insert(
                mid,
                betex::engine::state::Market::from_config_with_cfg(&state.cfg, mc),
            );
        }
        let root = EngineRoot::new_with(state, 0);

        let current_market_id = market_id_counter.load(std::sync::atomic::Ordering::Relaxed);
        if next_market_id > current_market_id {
            market_id_counter.store(next_market_id, std::sync::atomic::Ordering::Relaxed);
        }

        let handler = Self {
            root,
            market_ids,
            projection_snapshot,
            live_state,
            final_disk_snapshot,
            projection_path,
            snapshot_frequency,
            last_snapshot_seq: 0,
            pending_seq: 0,
            close_log: HashMap::new(),
            status_callback,
            phase_callback,
            market_meta,
            market_id_counter,
        };

        // Populate initial snapshot immediately so HTTP/WS clients see state before any events.
        handler.update_shared_snapshot();

        handler
    }

    #[allow(clippy::too_many_arguments)]
    pub(crate) fn from_disk_snapshot(
        disk_snap: ProjectionDiskSnapshot,
        market_ids: Arc<RwLock<Vec<MarketId>>>,
        projection_snapshot: Arc<RwLock<ProjectionSnapshot>>,
        live_state: Arc<RwLock<LiveState>>,
        final_disk_snapshot: Arc<parking_lot::Mutex<Option<ProjectionDiskSnapshot>>>,
        projection_path: PathBuf,
        snapshot_frequency: u64,
        status_callback: Option<StatusCallback>,
        phase_callback: Option<PhaseCallback>,
        market_meta: Arc<RwLock<HashMap<MarketId, MarketMeta>>>,
        market_id_counter: Arc<std::sync::atomic::AtomicU64>,
    ) -> Self {
        let seq = disk_snap.global_seq();
        {
            let mut mm = market_meta.write();
            *mm = disk_snap.market_meta.clone();
        }
        let counter_base = if disk_snap.market_id_counter > 0 {
            disk_snap.market_id_counter
        } else {
            disk_snap
                .market_ids
                .iter()
                .map(|mid| mid.0)
                .max()
                .unwrap_or(0)
                .saturating_add(1)
        };
        market_id_counter.store(counter_base, std::sync::atomic::Ordering::Relaxed);

        // Restore live state so completed/risk queries work immediately after recovery.
        {
            let mut live = live_state.write();
            *live = disk_snap.live.clone();
            live.global_seq = seq;
        }

        let handler = Self {
            root: disk_snap.root,
            market_ids,
            projection_snapshot,
            live_state,
            final_disk_snapshot,
            projection_path,
            snapshot_frequency,
            last_snapshot_seq: seq,
            pending_seq: seq,
            close_log: HashMap::new(),
            status_callback,
            phase_callback,
            market_meta,
            market_id_counter,
        };

        // Populate the shared snapshot immediately so queries work after recovery
        handler.update_shared_snapshot();

        handler
    }

    fn update_shared_snapshot(&self) {
        let market_ids = self.market_ids.read().clone();
        let markets: Vec<MarketSnapshot> = market_ids
            .into_iter()
            .filter_map(|mid| self.root.market_snapshot(mid, 50))
            .collect();

        let mut snap = self.projection_snapshot.write();
        snap.global_seq = self.pending_seq;
        snap.markets = markets;
    }

    fn notify_status_change(&self, market_id: MarketId, status: MarketStatus) {
        if let Some(ref cb) = self.status_callback {
            cb(market_id, status);
        }
    }

    fn notify_phase_change(&self, market_id: MarketId, phase: MarketPhase) {
        if let Some(ref cb) = self.phase_callback {
            cb(market_id, phase);
        }
    }

    fn trade_pair_key(order_id: OrderId, counter_party: OrderId) -> (OrderId, OrderId) {
        if order_id <= counter_party {
            (order_id, counter_party)
        } else {
            (counter_party, order_id)
        }
    }

    fn apply_live_event(&self, env: &BookEventEnvelope) {
        let ts_ms = env.timestamp.timestamp_millis();
        let market_id = env.market_id;
        if !self.market_ids.read().contains(&market_id) {
            return;
        }

        let mut live = self.live_state.write();
        let market = live.markets.entry(market_id).or_default();

        match env.event.clone() {
            betex::book::BookEvent::OrderAccepted {
                order_id,
                account_id,
                runner_id,
                side,
                price,
                stake,
                ..
            } => {
                market.orders.insert(
                    order_id,
                    LiveOrder {
                        order_id,
                        account_id,
                        runner_id,
                        side,
                        odds: price,
                        stake,
                        matched: Money::zero(),
                        state: betex::book::BookOrderState::ExecutableUnmatched,
                        created_at: ts_ms,
                        last_updated_at: ts_ms,
                    },
                );
                market.open_orders.insert(order_id);
            }
            betex::book::BookEvent::MarketStateChanged { to, .. } => {
                market.current_state = to;
            }
            betex::book::BookEvent::OrderCancelled {
                cancelled_order,
                cancel_cause,
                cause_detail,
            } => {
                let kind = match cancel_cause {
                    betex::book::CancelCause::SuspendLapse
                    | betex::book::CancelCause::InPlayLapse => TerminalEventKind::Lapsed,
                    betex::book::CancelCause::MarketVoid => TerminalEventKind::Voided,
                    _ => TerminalEventKind::Cancelled,
                };
                market.mark_terminal(
                    market_id,
                    ts_ms,
                    cancelled_order.order_id,
                    kind,
                    cause_detail.clone(),
                );
            }
            betex::book::BookEvent::OrderCancelledBatched {
                cancelled_orders,
                cancel_cause,
                cause_detail,
                ..
            } => {
                let kind = match cancel_cause {
                    betex::book::CancelCause::SuspendLapse
                    | betex::book::CancelCause::InPlayLapse => TerminalEventKind::Lapsed,
                    betex::book::CancelCause::MarketVoid => TerminalEventKind::Voided,
                    _ => TerminalEventKind::Cancelled,
                };
                for cancelled_order in cancelled_orders {
                    market.mark_terminal(
                        market_id,
                        ts_ms,
                        cancelled_order.order_id,
                        kind.clone(),
                        cause_detail.clone(),
                    );
                }
            }
            betex::book::BookEvent::TradeMatched {
                order_id,
                role,
                runner_id,
                side,
                price,
                stake,
                counter_party,
                remaining_stake,
                matched_delta,
                ..
            } => {
                let account_id = market
                    .orders
                    .get(&order_id)
                    .map(|o| o.account_id.clone())
                    .unwrap_or(AccountId::from(0u64));
                if let Some(order) = market.orders.get_mut(&order_id) {
                    order.matched = Money(order.matched.0.saturating_add(matched_delta.0));
                    order.last_updated_at = ts_ms;
                    if remaining_stake.0 == 0 {
                        order.matched = order.stake;
                        market.mark_terminal(
                            market_id,
                            ts_ms,
                            order_id,
                            TerminalEventKind::ExecutionComplete,
                            None,
                        );
                    } else {
                        order.state = betex::book::BookOrderState::ExecutablePartiallyMatched;
                    }
                }

                let current = PendingTradeView {
                    order_id,
                    role,
                    runner_id,
                    side,
                    price,
                    stake,
                    matched_delta,
                    account_id,
                };
                let key = Self::trade_pair_key(order_id, counter_party);
                if let Some(first) = market.pending_trades.remove(&key) {
                    if first.order_id != counter_party {
                        return;
                    }
                    let (maker_view, taker_view) = match (first.role, current.role) {
                        (TradeRole::Maker, TradeRole::Taker) => (first, current),
                        (TradeRole::Taker, TradeRole::Maker) => (current, first),
                        _ => return,
                    };
                    market.trades.entry(key).or_insert(LiveTrade {
                        maker_account_id: maker_view.account_id,
                        maker_order_id: maker_view.order_id,
                        maker_side: maker_view.side,
                        taker_account_id: taker_view.account_id,
                        taker_order_id: taker_view.order_id,
                        taker_side: taker_view.side,
                        runner_id: taker_view.runner_id,
                        price: taker_view.price,
                        stake: taker_view.stake,
                        matched_at: ts_ms,
                        voided: false,
                    });
                } else {
                    market.pending_trades.insert(key, current);
                }
            }
            betex::book::BookEvent::VoidTradesFromTime { .. } => {}
            _ => {}
        }

        live.global_seq = self.pending_seq;
    }

    fn maybe_disk_snapshot(&mut self) {
        if self.pending_seq - self.last_snapshot_seq < self.snapshot_frequency {
            return;
        }

        betex::metrics_stage!("projection_disk");
        betex::metric!({ inflight: 1, in_total: 1 });
        let t0 = Instant::now();

        let live = self.live_state.read();
        let market_ids = self.market_ids.read();
        let market_meta = self.market_meta.read();
        if let Err(e) = ProjectionDiskSnapshot::save_postcard_ref(
            &self.projection_path,
            &market_ids,
            &self.root,
            &live,
            &market_meta,
            self.market_id_counter
                .load(std::sync::atomic::Ordering::Relaxed),
        ) {
            eprintln!("[projection] Failed to save disk snapshot: {e}");
            betex::metric!({
                duration_ns: t0.elapsed().as_nanos() as u64,
                err_total: 1,
                out_total: 0,
                inflight: 0
            });
        } else {
            self.last_snapshot_seq = self.pending_seq;
            betex::metric!({
                duration_ns: t0.elapsed().as_nanos() as u64,
                out_total: 1,
                inflight: 0
            });
        }
    }
}

impl Drop for ProjectionHandler {
    fn drop(&mut self) {
        let live = self.live_state.read();
        // Avoid storing a potentially huge clone in-memory; we only need the on-disk snapshot.
        *self.final_disk_snapshot.lock() = None;

        let market_ids = self.market_ids.read();
        let market_meta = self.market_meta.read();
        if let Err(e) = ProjectionDiskSnapshot::save_postcard_ref(
            &self.projection_path,
            &market_ids,
            &self.root,
            &live,
            &market_meta,
            self.market_id_counter
                .load(std::sync::atomic::Ordering::Relaxed),
        ) {
            eprintln!("[projection] Failed to save final disk snapshot: {e}");
        } else {
            println!(
                "[projection] Final snapshot saved at seq={}",
                self.pending_seq
            );
        }
    }
}

impl disrupt_rs::EventHandler<EngineEvent> for ProjectionHandler {
    fn on_event(
        &mut self,
        event: &EngineEvent,
        sequence: disrupt_rs::Sequence,
        end_of_batch: bool,
    ) {
        let sequence_for_debug = sequence;
        debug_assert_eq!(sequence_for_debug, sequence);

        let t_apply = Instant::now();
        betex::metrics_stage!("projection_apply");
        betex::metric!({ inflight: 1, in_total: 1 });

        match &event.payload.event {
            betex::book::BookEvent::MarketCreated {
                name,
                market_model,
                market_kind,
                runner_ids,
                runner_labels,
                ..
            } => {
                // Ensure the market is included in the served set and metadata registry.
                {
                    let mut ids = self.market_ids.write();
                    if !ids.contains(&event.payload.market_id) {
                        ids.push(event.payload.market_id);
                    }
                }
                {
                    let mut mm = self.market_meta.write();
                    mm.insert(
                        event.payload.market_id,
                        MarketMeta {
                            name: name.clone(),
                            market_model: *market_model,
                            market_kind: *market_kind,
                            runner_ids: runner_ids.clone(),
                            runner_labels: runner_labels.clone(),
                            created_at_ms: event.payload.timestamp.timestamp_millis(),
                        },
                    );
                }
            }
            betex::book::BookEvent::OrderCancelledBatched {
                cancelled_orders,
                cancel_cause,
                ..
            } => {
                if *cancel_cause == betex::book::CancelCause::CloseCancel {
                    let prev = self
                        .close_log
                        .get(&event.payload.market_id)
                        .copied()
                        .unwrap_or(CloseLogState { cancelled_total: 0 });

                    let cancelled_in_chunk = cancelled_orders.len() as u64;
                    let cancelled_total = prev.cancelled_total.saturating_add(cancelled_in_chunk);

                    println!(
                        "[live][close] progress market_id={} cancelled_total={cancelled_total} orders_cancelled={}",
                        event.payload.market_id.0,
                        cancelled_orders.len()
                    );
                    self.close_log
                        .insert(event.payload.market_id, CloseLogState { cancelled_total });
                }
            }
            betex::book::BookEvent::BatchProcessCompleted { cancel_cause } => {
                if *cancel_cause == betex::book::CancelCause::CloseCancel {
                    let cancelled_total = self
                        .close_log
                        .get(&event.payload.market_id)
                        .copied()
                        .unwrap_or(CloseLogState { cancelled_total: 0 })
                        .cancelled_total;
                    println!(
                        "[live][close] done market_id={} cancelled_total={cancelled_total}",
                        event.payload.market_id.0
                    );
                    self.close_log.remove(&event.payload.market_id);
                    self.notify_status_change(event.payload.market_id, MarketStatus::Closed);
                }
            }
            betex::book::BookEvent::MarketSettled { .. } => {
                // Market has been settled (terminal)
                self.notify_status_change(event.payload.market_id, MarketStatus::Settled);
            }
            betex::book::BookEvent::MarketStateChanged { to, .. } => {
                // Map BookMarketState to MarketStatus
                let status = match to {
                    betex::book::BookMarketState::Open
                    | betex::book::BookMarketState::Suspended => MarketStatus::Active,
                    betex::book::BookMarketState::Closed => MarketStatus::Closed,
                    betex::book::BookMarketState::Settled => MarketStatus::Settled,
                    betex::book::BookMarketState::Voided => MarketStatus::Voided,
                    betex::book::BookMarketState::Halted => MarketStatus::Active,
                };
                self.notify_status_change(event.payload.market_id, status);
            }
            betex::book::BookEvent::MarketPhaseChanged { to, .. } => {
                self.notify_phase_change(event.payload.market_id, *to);
            }
            _ => {}
        }

        let expected = self.root.version().saturating_add(1);
        assert!(
            event.seq == expected,
            "projection seq gap: expected seq={}, got seq={}",
            expected,
            event.seq
        );
        self.root.assert_next_seq(event.seq);
        self.root.apply(std::slice::from_ref(&event.payload));
        debug_assert_eq!(self.root.version(), event.seq);
        self.pending_seq = self.root.version();

        betex::metric!({
            duration_ns: t_apply.elapsed().as_nanos() as u64,
            out_total: 1,
            inflight: 0
        });

        let t_live = Instant::now();
        betex::metrics_stage!("projection_live");
        betex::metric!({ inflight: 1, in_total: 1 });
        self.apply_live_event(&event.payload);
        betex::metric!({
            duration_ns: t_live.elapsed().as_nanos() as u64,
            out_total: 1,
            inflight: 0
        });

        if end_of_batch {
            let t_snap = Instant::now();
            betex::metrics_stage!("projection_snapshot");
            betex::metric!({ inflight: 1, in_total: 1 });
            self.update_shared_snapshot();
            betex::metric!({
                duration_ns: t_snap.elapsed().as_nanos() as u64,
                out_total: 1,
                inflight: 0
            });
            self.maybe_disk_snapshot();
        }
    }
}
