use crate::MARKET_ID;
use crate::control_plane::{AppState, ControlPlane, EngineInstance};
use crate::ws_protocol::{ClientMsg as WsClientMsg, ServerMsg as WsServerMsg};
use axum::{
    extract::{
        Query, State,
        ws::{Message, WebSocket, WebSocketUpgrade},
    },
    http::StatusCode,
    response::IntoResponse,
};
use betex::{
    book::RunnerResult,
    book::protocol::{
        command::{BinaryTimeInForce, Command, CommandKind, Persistence, Side, TimeInForce},
        response::Response,
    },
    prelude::*,
};
use futures::{SinkExt, StreamExt};
use serde::Deserialize;
use std::{collections::HashMap, sync::Arc, time::Duration};
use tokio::sync::{Semaphore, mpsc, watch};
use uuid::Uuid;

#[derive(Debug, Deserialize)]
pub(crate) struct WsParams {
    pub(crate) depth: Option<usize>,
    pub(crate) market_id: Option<u64>,
    /// If non-zero, suppress Ack/OrderResponse messages (snapshots still stream).
    pub(crate) no_ack: Option<u8>,
}

pub(crate) async fn health_handler() -> &'static str {
    "ok"
}

pub(crate) async fn metrics_handler() -> impl IntoResponse {
    axum::Json(betex::metrics::globals::metrics().snapshot()).into_response()
}

#[derive(Debug, Deserialize)]
pub(crate) struct SnapshotParams {
    market_id: Option<u64>,
    /// Depth per side (defaults to 5).
    depth: Option<usize>,
}

fn snapshot_response(
    engine: &EngineInstance,
    market_id: MarketId,
    depth: usize,
) -> axum::response::Response {
    let snapshot = engine.projection_snapshot.read();

    if let Some(market) = snapshot.markets.iter().find(|m| m.market_id == market_id) {
        let mut truncated = market.clone();
        truncated.truncate_depth(depth);
        axum::Json(truncated).into_response()
    } else {
        (
            StatusCode::NOT_FOUND,
            axum::Json(serde_json::json!({"error": "market not found"})),
        )
            .into_response()
    }
}

pub(crate) async fn snapshot_handler(
    State(state): State<AppState>,
    Query(params): Query<SnapshotParams>,
) -> impl IntoResponse {
    let engine_id = state.default_engine_id;
    let Some(engine) = state.control.get_engine(engine_id) else {
        return (
            StatusCode::NOT_FOUND,
            axum::Json(serde_json::json!({ "error": "engine not found" })),
        )
            .into_response();
    };
    let market_id = MarketId(params.market_id.unwrap_or(MARKET_ID.0));
    let depth = params.depth.unwrap_or(5).clamp(1, 50);
    snapshot_response(&engine, market_id, depth)
}

#[derive(Debug, Deserialize)]
pub(crate) struct CompletedOrdersParams {
    market_id: Option<u64>,
    account_id: Option<String>,
    limit: Option<usize>,
}

pub(crate) async fn completed_orders_handler(
    State(state): State<AppState>,
    Query(params): Query<CompletedOrdersParams>,
) -> impl IntoResponse {
    completed_orders_for_engine(state.default_engine_id, state, params).await
}

async fn completed_orders_for_engine(
    engine_id: Uuid,
    state: AppState,
    params: CompletedOrdersParams,
) -> axum::response::Response {
    let Some(engine) = state.control.get_engine(engine_id) else {
        return (
            StatusCode::NOT_FOUND,
            axum::Json(serde_json::json!({ "error": "engine not found" })),
        )
            .into_response();
    };
    let market_id = MarketId(params.market_id.unwrap_or(MARKET_ID.0));
    let account_id = params.account_id.map(AccountId::from);
    let limit = params.limit.unwrap_or(100).clamp(1, 10_000);

    let live = engine.live_state.read();
    let Some(market) = live.markets.get(&market_id) else {
        return axum::Json(serde_json::json!({"error": "market not found"})).into_response();
    };

    let mut out = Vec::new();
    for o in market.completed_orders.iter().rev() {
        if let Some(a) = account_id.as_ref()
            && &o.account_id != a
        {
            continue;
        }
        out.push(o);
        if out.len() >= limit {
            break;
        }
    }

    axum::Json(serde_json::json!({
        "market_id": market_id.0,
        "seq": live.global_seq,
        "count": out.len(),
        "orders": out,
    }))
    .into_response()
}

#[derive(Debug, Deserialize)]
pub(crate) struct RiskParams {
    market_id: Option<u64>,
    account_id: Option<String>,
}

fn profit_for_win(odds: OddsX10000, stake: Money) -> Money {
    let diff = odds.0.saturating_sub(10_000) as i128;
    let s = stake.0 as i128;
    Money(((s * diff) / 10_000).clamp(i64::MIN as i128, i64::MAX as i128) as i64)
}

fn apply_bet_pnl(
    pnl: &mut HashMap<RunnerId, i64>,
    runner_ids: &[RunnerId],
    bet_runner: RunnerId,
    side: Side,
    odds: OddsX10000,
    stake: Money,
) {
    if stake.0 <= 0 {
        return;
    }
    let win_profit = profit_for_win(odds, stake).0;
    for &winner in runner_ids {
        let delta = if winner == bet_runner {
            match side {
                Side::Back => win_profit,
                Side::Lay => -win_profit,
            }
        } else {
            match side {
                Side::Back => -stake.0,
                Side::Lay => stake.0,
            }
        };
        *pnl.entry(winner).or_insert(0) += delta;
    }
}

pub(crate) async fn risk_handler(
    State(state): State<AppState>,
    Query(params): Query<RiskParams>,
) -> impl IntoResponse {
    risk_for_engine(state.default_engine_id, state, params).await
}

async fn risk_for_engine(
    engine_id: Uuid,
    state: AppState,
    params: RiskParams,
) -> axum::response::Response {
    let Some(engine) = state.control.get_engine(engine_id) else {
        return (
            StatusCode::NOT_FOUND,
            axum::Json(serde_json::json!({ "error": "engine not found" })),
        )
            .into_response();
    };
    let market_id = MarketId(params.market_id.unwrap_or(MARKET_ID.0));
    let filter_account = params.account_id.map(AccountId::from);

    let runner_ids: Vec<RunnerId> = {
        let snap = engine.projection_snapshot.read();
        snap.markets
            .iter()
            .find(|m| m.market_id == market_id)
            .map(|m| m.runner_ids())
            .unwrap_or_default()
    };

    if runner_ids.is_empty() {
        return axum::Json(serde_json::json!({"error": "market not found"})).into_response();
    }

    let live = engine.live_state.read();
    let Some(market) = live.markets.get(&market_id) else {
        return axum::Json(serde_json::json!({"error": "market not found"})).into_response();
    };

    let mut matched_pnl_by_acct: HashMap<AccountId, HashMap<RunnerId, i64>> = HashMap::new();

    // Matched PnL from trades
    for trade in market.trades.values() {
        if trade.voided {
            continue;
        }
        let maker_map = matched_pnl_by_acct
            .entry(trade.maker_account_id.clone())
            .or_default();
        apply_bet_pnl(
            maker_map,
            &runner_ids,
            trade.runner_id,
            trade.maker_side,
            trade.price,
            trade.stake,
        );
        let taker_map = matched_pnl_by_acct
            .entry(trade.taker_account_id.clone())
            .or_default();
        apply_bet_pnl(
            taker_map,
            &runner_ids,
            trade.runner_id,
            trade.taker_side,
            trade.price,
            trade.stake,
        );
    }

    // Start totals as matched.
    let mut total_pnl_by_acct: HashMap<AccountId, HashMap<RunnerId, i64>> =
        matched_pnl_by_acct.clone();

    // Potential PnL from unmatched remainder (open orders only).
    for oid in market.open_orders.iter() {
        let Some(order) = market.orders.get(oid) else {
            continue;
        };
        let remaining = order.remaining();
        if remaining.0 <= 0 {
            continue;
        }
        let map = total_pnl_by_acct
            .entry(order.account_id.clone())
            .or_default();
        apply_bet_pnl(
            map,
            &runner_ids,
            order.runner_id,
            order.side,
            order.odds,
            remaining,
        );
    }

    let mut account_ids: Vec<AccountId> = total_pnl_by_acct.keys().cloned().collect();
    account_ids.sort_by(|a, b| a.0.cmp(&b.0));

    let mut accounts = Vec::new();
    for account_id in account_ids {
        if let Some(filter) = filter_account.as_ref()
            && &account_id != filter
        {
            continue;
        }
        let matched_map = matched_pnl_by_acct
            .get(&account_id)
            .cloned()
            .unwrap_or_default();
        let total_map = total_pnl_by_acct
            .get(&account_id)
            .cloned()
            .unwrap_or_default();

        let worst_case_loss_matched = runner_ids
            .iter()
            .map(|rid| matched_map.get(rid).copied().unwrap_or(0))
            .min()
            .unwrap_or(0)
            .min(0)
            .unsigned_abs();
        let worst_case_loss_if_filled = runner_ids
            .iter()
            .map(|rid| total_map.get(rid).copied().unwrap_or(0))
            .min()
            .unwrap_or(0)
            .min(0)
            .unsigned_abs();

        let matched_json: serde_json::Value = runner_ids
            .iter()
            .map(|rid| {
                (
                    rid.0.to_string(),
                    serde_json::Value::from(matched_map.get(rid).copied().unwrap_or(0)),
                )
            })
            .collect::<serde_json::Map<String, serde_json::Value>>()
            .into();
        let total_json: serde_json::Value = runner_ids
            .iter()
            .map(|rid| {
                (
                    rid.0.to_string(),
                    serde_json::Value::from(total_map.get(rid).copied().unwrap_or(0)),
                )
            })
            .collect::<serde_json::Map<String, serde_json::Value>>()
            .into();

        accounts.push(serde_json::json!({
            "account_id": account_id.0,
            "pnl_matched_if_winner": matched_json,
            "pnl_if_filled_if_winner": total_json,
            "worst_case_loss_matched": worst_case_loss_matched,
            "worst_case_loss_if_filled": worst_case_loss_if_filled,
        }));
    }

    axum::Json(serde_json::json!({
        "market_id": market_id.0,
        "seq": live.global_seq,
        "accounts": accounts,
    }))
    .into_response()
}

pub(crate) async fn live_counts_handler(State(state): State<AppState>) -> impl IntoResponse {
    live_counts_for_engine(state.default_engine_id, state).await
}

async fn live_counts_for_engine(engine_id: Uuid, state: AppState) -> axum::response::Response {
    let Some(engine) = state.control.get_engine(engine_id) else {
        return (
            StatusCode::NOT_FOUND,
            axum::Json(serde_json::json!({ "error": "engine not found" })),
        )
            .into_response();
    };

    let live = engine.live_state.read();
    let mut markets: Vec<_> = live
        .markets
        .iter()
        .map(|(mid, m)| {
            serde_json::json!({
                "market_id": mid.0,
                "orders_len": m.orders.len(),
                "open_orders_len": m.open_orders.len(),
                "completed_orders_len": m.completed_orders.len(),
                "trades_len": m.trades.len(),
            })
        })
        .collect();
    markets.sort_by_key(|v| v.get("market_id").and_then(|x| x.as_u64()).unwrap_or(0));

    axum::Json(serde_json::json!({
        "seq": live.global_seq,
        "market_count": markets.len(),
        "markets": markets,
    }))
    .into_response()
}

pub(crate) async fn ws_handler(
    ws: WebSocketUpgrade,
    State(state): State<AppState>,
    Query(params): Query<WsParams>,
) -> impl IntoResponse {
    let engine_id = state.default_engine_id;
    ws_for_engine(ws, state, engine_id, params)
}

fn ws_for_engine(
    ws: WebSocketUpgrade,
    state: AppState,
    engine_id: Uuid,
    params: WsParams,
) -> axum::response::Response {
    let depth = params.depth.unwrap_or(10).clamp(1, 50);
    let market_id = MarketId(params.market_id.unwrap_or(MARKET_ID.0));
    let shutdown_rx = state.shutdown_rx.clone();
    let Some(engine) = state.control.get_engine(engine_id) else {
        return (
            StatusCode::NOT_FOUND,
            axum::Json(serde_json::json!({ "error": "engine not found" })),
        )
            .into_response();
    };
    ws.on_upgrade(move |socket| ws_stream(socket, engine, shutdown_rx, market_id, depth))
        .into_response()
}

async fn ws_stream(
    mut socket: WebSocket,
    engine: Arc<EngineInstance>,
    mut shutdown_rx: watch::Receiver<bool>,
    market_id: MarketId,
    depth: usize,
) {
    let mut interval = tokio::time::interval(Duration::from_millis(200));

    loop {
        tokio::select! {
            _ = shutdown_rx.changed() => {
                if *shutdown_rx.borrow() {
                    break;
                }
            }
            _ = interval.tick() => {
                // Clone the snapshot while holding the lock briefly
                let truncated = {
                    let snapshot = engine.projection_snapshot.read();
                    snapshot.markets.iter().find(|m| m.market_id == market_id).map(|market| {
                        let mut truncated = market.clone();
                        truncated.truncate_depth(depth);
                        truncated
                    })
                };

                if let Some(truncated) = truncated {
                    let msg = WsServerMsg::Snapshot { snapshot: truncated };
                    if let Ok(text) = crate::ws_protocol::encode_msg(&msg)
                        && socket.send(Message::Text(text.into())).await.is_err()
                    {
                        break;
                    }
                }
            }
            incoming = socket.recv() => {
                match incoming {
                    Some(Ok(Message::Close(_))) | None => break,
                    _ => {}
                }
            }
        }
    }
}

pub(crate) async fn ws_ctl_handler(
    ws: WebSocketUpgrade,
    State(state): State<AppState>,
    Query(params): Query<WsParams>,
) -> impl IntoResponse {
    // Allow connection even without a default engine for control-plane operations
    let depth = params.depth.unwrap_or(10).clamp(1, 50);
    let market_id = MarketId(params.market_id.unwrap_or(MARKET_ID.0));
    let no_trade_ack = params.no_ack.unwrap_or(0) != 0;
    let shutdown_rx = state.shutdown_rx.clone();
    let control = state.control.clone();

    ws.on_upgrade(move |socket| {
        ws_ctl_stream(
            socket,
            Some(state.default_engine_id),
            control,
            shutdown_rx,
            market_id,
            depth,
            no_trade_ack,
        )
    })
    .into_response()
}

async fn ws_ctl_stream(
    socket: WebSocket,
    engine_id: Option<Uuid>,
    control: ControlPlane,
    mut shutdown_rx: watch::Receiver<bool>,
    market_id: MarketId,
    depth: usize,
    no_trade_ack: bool,
) {
    let mut snapshot_interval = tokio::time::interval(Duration::from_millis(200));
    let mut status_interval = tokio::time::interval(Duration::from_secs(1));

    let (mut ws_tx, mut ws_rx) = socket.split();

    // Outbound queue so we don't block command ingestion on websocket writes.
    let (out_tx, mut out_rx) = mpsc::channel::<WsServerMsg>(4096);

    // Subscribe to broadcast channel for status updates
    let mut status_rx = control.inner.status_broadcast_tx.subscribe();

    // Cap how many engine responses we await concurrently; beyond this, acks become lossy.
    let ack_sem = Arc::new(Semaphore::new(4096));

    let writer = tokio::spawn(async move {
        while let Some(msg) = out_rx.recv().await {
            if let Ok(text) = crate::ws_protocol::encode_msg(&msg)
                && ws_tx.send(Message::Text(text.into())).await.is_err()
            {
                break;
            }
        }
    });

    let mut selected_engine_id = engine_id;
    let mut selected_market_id = market_id;

    loop {
        tokio::select! {
            _ = shutdown_rx.changed() => {
                if *shutdown_rx.borrow() {
                    break;
                }
            }
            _ = snapshot_interval.tick() => {
                let Some(engine_id) = selected_engine_id else {
                    continue;
                };
                let Some(engine) = control.get_engine(engine_id) else {
                    continue;
                };
                // Clone the snapshot while holding the lock briefly
                let truncated = {
                    let snapshot = engine.projection_snapshot.read();
                    snapshot.markets.iter().find(|m| m.market_id == selected_market_id).map(|market| {
                        let mut truncated = market.clone();
                        truncated.truncate_depth(depth);
                        truncated
                    })
                };

                if let Some(truncated) = truncated {
                    let _ = out_tx.try_send(WsServerMsg::Snapshot { snapshot: truncated });
                }
            }
            _ = status_interval.tick() => {
                let Some(engine_id) = selected_engine_id else {
                    continue;
                };
                let Some(engine) = control.get_engine(engine_id) else {
                    continue;
                };
                // Clone status info while holding lock briefly
                let (global_seq, market_count) = {
                    let snapshot = engine.projection_snapshot.read();
                    (snapshot.global_seq, snapshot.markets.len())
                };
                let _ = out_tx.try_send(WsServerMsg::Status { global_seq, market_count });
            }
            // Forward broadcast status changes to WebSocket
            broadcast_msg = status_rx.recv() => {
                if let Ok(msg) = broadcast_msg {
                    let _ = out_tx.try_send(msg);
                }
            }
            incoming = ws_rx.next() => {
                match incoming {
                    Some(Ok(Message::Text(text))) => {
                        match crate::ws_protocol::decode_msg::<WsClientMsg>(text.as_bytes()) {
                            Ok(cmd) => {
                                handle_ws_command(
                                    &out_tx,
                                    &mut selected_engine_id,
                                    &control,
                                    &mut selected_market_id,
                                    cmd,
                                    no_trade_ack,
                                    Arc::clone(&ack_sem),
                                )
                                .await;
                            }
                            Err(e) => {
                                eprintln!("[live][ws] decode error: text bytes={} err={e}", text.len());
                            }
                        }
                    }
                    Some(Ok(Message::Binary(bytes))) => {
                        match crate::ws_protocol::decode_msg::<WsClientMsg>(bytes.as_ref()) {
                            Ok(cmd) => {
                                // Handy when debugging client issues (market creation, etc.)
                                // Keep this lightweight: one line per inbound client command.
                                // println!("[live][ws] inbound: {cmd:?}");
                                handle_ws_command(
                                    &out_tx,
                                    &mut selected_engine_id,
                                    &control,
                                    &mut selected_market_id,
                                    cmd,
                                    no_trade_ack,
                                    Arc::clone(&ack_sem),
                                )
                                .await;
                            }
                            Err(e) => {
                                eprintln!(
                                    "[live][ws] decode error: {} bytes: {e}",
                                    bytes.len()
                                );
                            }
                        }
                    }
                    Some(Ok(Message::Close(_))) | None => break,
                    _ => {}
                }
            }
        }
    }

    drop(out_tx);
    let _ = writer.await;
}

async fn handle_ws_command(
    out_tx: &mpsc::Sender<WsServerMsg>,
    selected_engine_id: &mut Option<Uuid>,
    control: &ControlPlane,
    default_market_id: &mut MarketId,
    cmd: WsClientMsg,
    no_trade_ack: bool,
    ack_sem: Arc<Semaphore>,
) {
    async fn await_engine_response(
        rx: flume::Receiver<Result<Response, betex::engine::error::EngineError>>,
    ) -> Result<Response, String> {
        let join = tokio::task::spawn_blocking(move || rx.recv_timeout(Duration::from_secs(2)));
        match join.await {
            Ok(Ok(Ok(resp))) => Ok(resp),
            Ok(Ok(Err(e))) => Err(format!("{e}")),
            Ok(Err(_)) => Err("engine response timeout".into()),
            Err(e) => Err(format!("join error: {e}")),
        }
    }

    // Helper macro to require engine for a command
    macro_rules! require_engine {
        ($out_tx:expr) => {{
            let Some(engine_id) = selected_engine_id.as_ref().copied() else {
                let _ = $out_tx
                    .send(WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some("no engine available - create a market first".into()),
                    })
                    .await;
                return;
            };
            let Some(e) = control.get_engine(engine_id) else {
                let _ = $out_tx
                    .send(WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some("engine not found".into()),
                    })
                    .await;
                return;
            };
            e
        }};
    }

    match cmd {
        // Admin commands - return Ack (require engine)
        WsClientMsg::SetMarketState {
            market_id,
            state: desired,
        } => {
            let engine = require_engine!(out_tx);
            let market_id = market_id.unwrap_or(*default_market_id);
            let cmd = Command {
                correlation_id: None,
                metadata: None,
                market_id,
                kind: CommandKind::SetMarketState { state: desired },
            };
            let msg = match engine.handler.submit_with_response(cmd) {
                Ok(rx) => match await_engine_response(rx).await {
                    Ok(response) => {
                        if !matches!(response, Response::Rejected { .. }) {
                            println!(
                                "[live][accept] SET_MARKET_STATE market_id={} state={desired:?}",
                                market_id.0
                            );
                        }
                        WsServerMsg::Ack {
                            ok: !matches!(response, Response::Rejected { .. }),
                            response: Some(response.clone()),
                            error: None,
                        }
                    }
                    Err(e) => WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some(e),
                    },
                },
                Err(e) => WsServerMsg::Ack {
                    ok: false,
                    response: None,
                    error: Some(format!("{e}")),
                },
            };
            let _ = out_tx.send(msg).await;
        }
        WsClientMsg::SetMarketPhase {
            market_id,
            phase: desired,
        } => {
            let engine = require_engine!(out_tx);
            let market_id = market_id.unwrap_or(*default_market_id);
            let cmd = Command {
                correlation_id: None,
                metadata: None,
                market_id,
                kind: CommandKind::SetMarketPhase { phase: desired },
            };
            let msg = match engine.handler.submit_with_response(cmd) {
                Ok(rx) => match await_engine_response(rx).await {
                    Ok(response) => {
                        if !matches!(response, Response::Rejected { .. }) {
                            println!(
                                "[live][accept] SET_MARKET_PHASE market_id={} phase={desired:?}",
                                market_id.0
                            );
                        }
                        WsServerMsg::Ack {
                            ok: !matches!(response, Response::Rejected { .. }),
                            response: Some(response.clone()),
                            error: None,
                        }
                    }
                    Err(e) => WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some(e),
                    },
                },
                Err(e) => WsServerMsg::Ack {
                    ok: false,
                    response: None,
                    error: Some(format!("{e}")),
                },
            };
            let _ = out_tx.send(msg).await;
        }
        WsClientMsg::CloseMarket {
            market_id,
            batch_max_events,
        } => {
            let engine = require_engine!(out_tx);
            let market_id = market_id.unwrap_or(*default_market_id);
            let cmd = Command {
                correlation_id: None,
                metadata: None,
                market_id,
                kind: CommandKind::CloseMarket { batch_max_events },
            };
            let msg = match engine.handler.submit_with_response(cmd) {
                Ok(rx) => match await_engine_response(rx).await {
                    Ok(response) => {
                        if !matches!(response, Response::Rejected { .. }) {
                            println!(
                                "[live][accept] CLOSE_MARKET market_id={} batch_max_events={batch_max_events}",
                                market_id.0
                            );
                        }
                        WsServerMsg::Ack {
                            ok: !matches!(response, Response::Rejected { .. }),
                            response: Some(response.clone()),
                            error: None,
                        }
                    }
                    Err(e) => WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some(e),
                    },
                },
                Err(e) => WsServerMsg::Ack {
                    ok: false,
                    response: None,
                    error: Some(format!("{e}")),
                },
            };
            let _ = out_tx.send(msg).await;
        }
        WsClientMsg::BatchCancelOrders {
            correlation_id,
            market_id,
            from_created_at_inclusive_ms,
            to_created_at_inclusive_ms,
            account_id,
            runner_id,
            batch_max_events,
            reason,
            metadata,
        } => {
            let engine = require_engine!(out_tx);
            let market_id = market_id.unwrap_or(*default_market_id);
            let cmd = Command {
                correlation_id,
                metadata,
                market_id,
                kind: CommandKind::BatchCancelOrders {
                    from_created_at_inclusive: from_created_at_inclusive_ms
                        .map(betex::types::from_unix_timestamp_ms),
                    to_created_at_inclusive: to_created_at_inclusive_ms
                        .map(betex::types::from_unix_timestamp_ms),
                    account_id,
                    runner_id,
                    batch_max_events,
                    reason,
                },
            };
            let msg = match engine.handler.submit_with_response(cmd) {
                Ok(rx) => match await_engine_response(rx).await {
                    Ok(response) => {
                        if !matches!(response, Response::Rejected { .. }) {
                            println!(
                                "[live][accept] BATCH_CANCEL_ORDERS market_id={} batch_max_events={batch_max_events}",
                                market_id.0
                            );
                        }
                        WsServerMsg::Ack {
                            ok: !matches!(response, Response::Rejected { .. }),
                            response: Some(response.clone()),
                            error: None,
                        }
                    }
                    Err(e) => WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some(e),
                    },
                },
                Err(e) => WsServerMsg::Ack {
                    ok: false,
                    response: None,
                    error: Some(format!("{e}")),
                },
            };
            let _ = out_tx.send(msg).await;
        }
        WsClientMsg::ContinueBatchCancelOrders {
            correlation_id,
            market_id,
        } => {
            let engine = require_engine!(out_tx);
            let market_id = market_id.unwrap_or(*default_market_id);
            let cmd = Command {
                correlation_id,
                metadata: None,
                market_id,
                kind: CommandKind::ContinueBatchCancelOrders,
            };
            let msg = match engine.handler.submit_with_response(cmd) {
                Ok(rx) => match await_engine_response(rx).await {
                    Ok(response) => WsServerMsg::Ack {
                        ok: !matches!(response, Response::Rejected { .. }),
                        response: Some(response.clone()),
                        error: None,
                    },
                    Err(e) => WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some(e),
                    },
                },
                Err(e) => WsServerMsg::Ack {
                    ok: false,
                    response: None,
                    error: Some(format!("{e}")),
                },
            };
            let _ = out_tx.send(msg).await;
        }
        WsClientMsg::SettleMarket {
            market_id,
            winner_runner_id,
        } => {
            let engine = require_engine!(out_tx);
            let market_id = market_id.unwrap_or(*default_market_id);
            let runner_ids: Vec<RunnerId> = {
                let snapshot = engine.projection_snapshot.read();
                let from_snapshot = snapshot
                    .markets
                    .iter()
                    .find(|m| m.market_id == market_id)
                    .map(|m| m.runner_ids())
                    .unwrap_or_default();

                if !from_snapshot.is_empty() {
                    from_snapshot
                } else {
                    let meta = engine.market_meta.read();
                    meta.get(&market_id)
                        .map(|m| m.runner_ids.clone())
                        .unwrap_or_default()
                }
            };

            let winner = winner_runner_id;
            let runner_results: Vec<(RunnerId, RunnerResult)> = runner_ids
                .into_iter()
                .map(|rid| {
                    if rid == winner {
                        (rid, RunnerResult::Winner)
                    } else {
                        (rid, RunnerResult::Loser)
                    }
                })
                .collect();

            let msg = if runner_results.is_empty() {
                WsServerMsg::Ack {
                    ok: false,
                    response: None,
                    error: Some("no runners available for market (snapshot/meta empty)".into()),
                }
            } else if !runner_results.iter().any(|(rid, _)| *rid == winner) {
                WsServerMsg::Ack {
                    ok: false,
                    response: None,
                    error: Some(format!("winner runner_id {:?} not found", winner_runner_id)),
                }
            } else {
                let cmd = Command {
                    correlation_id: None,
                    metadata: None,
                    market_id,
                    kind: CommandKind::SettleMarket {
                        runner_results,
                        dead_heat_divisor: None,
                    },
                };
                match engine.handler.submit_with_response(cmd) {
                    Ok(rx) => match await_engine_response(rx).await {
                        Ok(response) => {
                            if !matches!(response, Response::Rejected { .. }) {
                                println!(
                                    "[live][accept] SETTLE_MARKET market_id={} winner_runner_id={:?}",
                                    market_id.0, winner_runner_id
                                );
                            }
                            WsServerMsg::Ack {
                                ok: !matches!(response, Response::Rejected { .. }),
                                response: Some(response.clone()),
                                error: None,
                            }
                        }
                        Err(e) => WsServerMsg::Ack {
                            ok: false,
                            response: None,
                            error: Some(e),
                        },
                    },
                    Err(e) => WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some(format!("{e}")),
                    },
                }
            };

            let _ = out_tx.send(msg).await;
        }
        WsClientMsg::VoidMarket { market_id, reason } => {
            let engine = require_engine!(out_tx);
            let market_id = market_id.unwrap_or(*default_market_id);
            let cmd = Command {
                correlation_id: None,
                metadata: None,
                market_id,
                kind: CommandKind::VoidMarket {
                    reason: reason.unwrap_or_else(|| "void via live_tui".into()),
                },
            };
            let msg = match engine.handler.submit_with_response(cmd) {
                Ok(rx) => match await_engine_response(rx).await {
                    Ok(response) => {
                        if !matches!(response, Response::Rejected { .. }) {
                            println!("[live][accept] VOID_MARKET market_id={}", market_id.0);
                        }
                        WsServerMsg::Ack {
                            ok: !matches!(response, Response::Rejected { .. }),
                            response: Some(response.clone()),
                            error: None,
                        }
                    }
                    Err(e) => WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some(e),
                    },
                },
                Err(e) => WsServerMsg::Ack {
                    ok: false,
                    response: None,
                    error: Some(format!("{e}")),
                },
            };
            let _ = out_tx.send(msg).await;
        }
        WsClientMsg::HaltMarket { market_id, reason } => {
            let engine = require_engine!(out_tx);
            let market_id = market_id.unwrap_or(*default_market_id);
            let cmd = Command {
                correlation_id: None,
                metadata: None,
                market_id,
                kind: CommandKind::HaltMarket { reason },
            };
            let msg = match engine.handler.submit_with_response(cmd) {
                Ok(rx) => match await_engine_response(rx).await {
                    Ok(response) => {
                        if !matches!(response, Response::Rejected { .. }) {
                            println!(
                                "[live][accept] HALT_MARKET market_id={} reason={reason}",
                                market_id.0
                            );
                        }
                        WsServerMsg::Ack {
                            ok: !matches!(response, Response::Rejected { .. }),
                            response: Some(response.clone()),
                            error: None,
                        }
                    }
                    Err(e) => WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some(e),
                    },
                },
                Err(e) => WsServerMsg::Ack {
                    ok: false,
                    response: None,
                    error: Some(format!("{e}")),
                },
            };
            let _ = out_tx.send(msg).await;
        }
        WsClientMsg::ResumeMarket { market_id } => {
            let engine = require_engine!(out_tx);
            let market_id = market_id.unwrap_or(*default_market_id);
            let cmd = Command {
                correlation_id: None,
                metadata: None,
                market_id,
                kind: CommandKind::ResumeMarket,
            };
            let msg = match engine.handler.submit_with_response(cmd) {
                Ok(rx) => match await_engine_response(rx).await {
                    Ok(response) => {
                        if !matches!(response, Response::Rejected { .. }) {
                            println!("[live][accept] RESUME_MARKET market_id={}", market_id.0);
                        }
                        WsServerMsg::Ack {
                            ok: !matches!(response, Response::Rejected { .. }),
                            response: Some(response.clone()),
                            error: None,
                        }
                    }
                    Err(e) => WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some(e),
                    },
                },
                Err(e) => WsServerMsg::Ack {
                    ok: false,
                    response: None,
                    error: Some(format!("{e}")),
                },
            };
            let _ = out_tx.send(msg).await;
        }
        WsClientMsg::Ping => {
            let msg = WsServerMsg::Ack {
                ok: true,
                response: None,
                error: None,
            };
            let _ = out_tx.send(msg).await;
        }

        // Trading commands - return OrderResponse
        WsClientMsg::PlaceOrder {
            correlation_id,
            order_id,
            market_id,
            runner_id,
            account_id,
            side,
            odds,
            stake,
            persistence,
            time_in_force,
        } => {
            let engine = require_engine!(out_tx);
            let response_correlation_id = correlation_id.clone();
            let cmd = Command {
                correlation_id: correlation_id.clone(),
                metadata: None,
                market_id,
                kind: CommandKind::PlaceOrder {
                    runner_id,
                    account_id,
                    client_order_id: Some(ClientOrderId(order_id.0)),
                    side,
                    odds,
                    stake: Money(stake),
                    persistence: persistence.unwrap_or(Persistence::Persist),
                    time_in_force: time_in_force.unwrap_or(TimeInForce::Gtc),
                },
            };
            if no_trade_ack {
                // Fire-and-forget: this is for hammering throughput and intentionally avoids
                // per-command response channels / websockets replies.
                let _ = engine.handler.submit(cmd);
                return;
            }

            let Ok(permit) = ack_sem.clone().try_acquire_owned() else {
                // Under load, prioritize throughput and accept lossy acks.
                let _ = engine.handler.submit(cmd);
                return;
            };

            match engine.handler.submit_with_response(cmd) {
                Ok(rx) => {
                    let out_tx = out_tx.clone();
                    let correlation_id = response_correlation_id.clone();
                    tokio::spawn(async move {
                        let _permit = permit;
                        let msg = match await_engine_response(rx).await {
                            Ok(Response::Ok {
                                order_id, trades, ..
                            }) => {
                                let _trades_n = trades.len();
                                // println!(
                                //     "[live][accept] PLACE_ORDER corr_id={correlation_id} market_id={market_id} runner_id={runner_id} account_id={account_id} side={side:?} odds={odds} stake={stake} trades={_trades_n}"
                                // );
                                WsServerMsg::OrderResponse {
                                    correlation_id,
                                    ok: true,
                                    order_id,
                                    trades: trades.to_vec(),
                                    error: None,
                                }
                            }
                            Ok(resp) if matches!(resp, Response::Rejected { .. }) => {
                                let Response::Rejected { reason, .. } = resp else {
                                    unreachable!();
                                };
                                WsServerMsg::OrderResponse {
                                    correlation_id,
                                    ok: false,
                                    order_id: None,
                                    trades: vec![],
                                    error: Some(format!("{:?}", reason)),
                                }
                            }
                            Ok(_) => WsServerMsg::OrderResponse {
                                correlation_id,
                                ok: false,
                                order_id: None,
                                trades: vec![],
                                error: Some("unexpected response".into()),
                            },
                            Err(e) => WsServerMsg::OrderResponse {
                                correlation_id,
                                ok: false,
                                order_id: None,
                                trades: vec![],
                                error: Some(e),
                            },
                        };
                        let _ = out_tx.send(msg).await;
                    });
                }
                Err(e) => {
                    let _ = out_tx
                        .send(WsServerMsg::OrderResponse {
                            correlation_id: response_correlation_id,
                            ok: false,
                            order_id: None,
                            trades: vec![],
                            error: Some(format!("{e}")),
                        })
                        .await;
                }
            }
        }
        WsClientMsg::PlaceBinaryOrder {
            correlation_id,
            order_id,
            market_id,
            account_id,
            side,
            price_ticks,
            qty_shares,
            time_in_force,
        } => {
            let engine = require_engine!(out_tx);
            let response_correlation_id = correlation_id.clone();
            let cmd = Command {
                correlation_id: correlation_id.clone(),
                metadata: None,
                market_id,
                kind: CommandKind::PlaceBinaryOrder {
                    account_id,
                    client_order_id: Some(ClientOrderId(order_id.0)),
                    side,
                    price_ticks,
                    qty_shares,
                    time_in_force: time_in_force.unwrap_or(BinaryTimeInForce::Gtc),
                },
            };
            if no_trade_ack {
                let _ = engine.handler.submit(cmd);
                return;
            }

            let Ok(permit) = ack_sem.clone().try_acquire_owned() else {
                let _ = engine.handler.submit(cmd);
                return;
            };

            match engine.handler.submit_with_response(cmd) {
                Ok(rx) => {
                    let out_tx = out_tx.clone();
                    let correlation_id = response_correlation_id.clone();
                    tokio::spawn(async move {
                        let _permit = permit;
                        let msg = match await_engine_response(rx).await {
                            Ok(Response::Ok {
                                order_id, trades, ..
                            }) => WsServerMsg::OrderResponse {
                                correlation_id,
                                ok: true,
                                order_id,
                                trades: trades.to_vec(),
                                error: None,
                            },
                            Ok(resp) if matches!(resp, Response::Rejected { .. }) => {
                                let Response::Rejected { reason, .. } = resp else {
                                    unreachable!();
                                };
                                WsServerMsg::OrderResponse {
                                    correlation_id,
                                    ok: false,
                                    order_id: None,
                                    trades: vec![],
                                    error: Some(format!("{:?}", reason)),
                                }
                            }
                            Ok(_) => WsServerMsg::OrderResponse {
                                correlation_id,
                                ok: false,
                                order_id: None,
                                trades: vec![],
                                error: Some("unexpected response".into()),
                            },
                            Err(e) => WsServerMsg::OrderResponse {
                                correlation_id,
                                ok: false,
                                order_id: None,
                                trades: vec![],
                                error: Some(e),
                            },
                        };
                        let _ = out_tx.send(msg).await;
                    });
                }
                Err(e) => {
                    let _ = out_tx
                        .send(WsServerMsg::OrderResponse {
                            correlation_id: response_correlation_id,
                            ok: false,
                            order_id: None,
                            trades: vec![],
                            error: Some(format!("{e}")),
                        })
                        .await;
                }
            }
        }
        WsClientMsg::CancelOrder {
            correlation_id,
            market_id,
            account_id,
            order_id,
        } => {
            let engine = require_engine!(out_tx);
            let response_correlation_id = correlation_id.clone();
            let cmd = Command {
                correlation_id: correlation_id.clone(),
                metadata: None,
                market_id,
                kind: CommandKind::CancelOrder {
                    account_id,
                    order_id,
                },
            };
            if no_trade_ack {
                let _ = engine.handler.submit(cmd);
                return;
            }

            let Ok(permit) = ack_sem.clone().try_acquire_owned() else {
                let _ = engine.handler.submit(cmd);
                return;
            };

            match engine.handler.submit_with_response(cmd) {
                Ok(rx) => {
                    let out_tx = out_tx.clone();
                    let correlation_id = response_correlation_id.clone();
                    tokio::spawn(async move {
                        let _permit = permit;
                        let msg = match await_engine_response(rx).await {
                            Ok(Response::Ok {
                                order_id, trades, ..
                            }) => WsServerMsg::OrderResponse {
                                correlation_id,
                                ok: true,
                                order_id,
                                trades: trades.to_vec(),
                                error: None,
                            },
                            Ok(resp) if matches!(resp, Response::Rejected { .. }) => {
                                let Response::Rejected { reason, .. } = resp else {
                                    unreachable!();
                                };
                                WsServerMsg::OrderResponse {
                                    correlation_id,
                                    ok: false,
                                    order_id: None,
                                    trades: vec![],
                                    error: Some(format!("{:?}", reason)),
                                }
                            }
                            Ok(_) => WsServerMsg::OrderResponse {
                                correlation_id,
                                ok: false,
                                order_id: None,
                                trades: vec![],
                                error: Some("unexpected response".into()),
                            },
                            Err(e) => WsServerMsg::OrderResponse {
                                correlation_id,
                                ok: false,
                                order_id: None,
                                trades: vec![],
                                error: Some(e),
                            },
                        };
                        let _ = out_tx.send(msg).await;
                    });
                }
                Err(e) => {
                    let _ = out_tx
                        .send(WsServerMsg::OrderResponse {
                            correlation_id: response_correlation_id,
                            ok: false,
                            order_id: None,
                            trades: vec![],
                            error: Some(format!("{e}")),
                        })
                        .await;
                }
            }
        }
        WsClientMsg::CreateMarket {
            name,
            market_model,
            market_kind,
            market_phase,
            runner_ids,
            runner_labels,
        } => {
            println!(
                "[live][ws] CREATE_MARKET name=\"{}\" model={:?} kind={:?} phase={:?} runners={}",
                name,
                market_model,
                market_kind,
                market_phase,
                runner_ids.len()
            );
            match control
                .create_market(
                    name,
                    market_model,
                    market_kind,
                    market_phase,
                    runner_ids,
                    runner_labels,
                )
                .await
            {
                Ok(info) => {
                    // Auto-subscribe creator to the new market.
                    *selected_engine_id = Some(info.engine_id);
                    *default_market_id = info.market_id;
                    let _ = out_tx
                        .send(WsServerMsg::MarketCreated { market: info })
                        .await;
                }
                Err(e) => {
                    let _ = out_tx
                        .send(WsServerMsg::Ack {
                            ok: false,
                            response: None,
                            error: Some(e),
                        })
                        .await;
                }
            }
        }
        WsClientMsg::RemoveMarket { market_id } => match control.remove_market(market_id).await {
            Ok(()) => {
                let _ = out_tx.send(WsServerMsg::MarketRemoved { market_id }).await;
            }
            Err(e) => {
                let _ = out_tx
                    .send(WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some(e),
                    })
                    .await;
            }
        },
        WsClientMsg::ListMarkets => {
            let markets = control.list_markets().await;
            let _ = out_tx.send(WsServerMsg::MarketList { markets }).await;
        }
        WsClientMsg::Subscribe { market_id } => {
            let info = control.inner.market_infos.read().get(&market_id).cloned();
            let Some(info) = info else {
                let _ = out_tx
                    .send(WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some("market not found".into()),
                    })
                    .await;
                return;
            };
            if control.get_engine(info.engine_id).is_none() {
                let _ = out_tx
                    .send(WsServerMsg::Ack {
                        ok: false,
                        response: None,
                        error: Some("engine not found".into()),
                    })
                    .await;
                return;
            };
            *selected_engine_id = Some(info.engine_id);
            *default_market_id = market_id;
            let _ = out_tx
                .send(WsServerMsg::Ack {
                    ok: true,
                    response: None,
                    error: None,
                })
                .await;
        }
    }
}
