//! Live soccer market simulation server.
//!
//! This example demonstrates:
//! - Engine with WAL-based durability
//! - Projection as a disruptor consumer
//! - Disk-based projection snapshots for crash recovery
//! - 3-runner soccer market (Home/Draw/Away)
//! - Axum HTTP/WebSocket server for snapshot queries
//!
//! Run with: `cargo run --example live`
//!
//! Then run the TUI client (with embedded bots):
//! - `cargo run --example live_tui`
//!
//! Or query directly:
//! - `curl http://127.0.0.1:3000/snapshot`

use crate::handlers::{
    completed_orders_handler, health_handler, live_counts_handler, metrics_handler, risk_handler,
    snapshot_handler, ws_ctl_handler, ws_handler,
};
use betex::{config::ThreadPinning, prelude::*};
use clap::Parser;
use std::{net::SocketAddr, sync::Arc};
use uuid::Uuid;

mod cli;
mod control_plane;
mod engine_builder;
mod handlers;
mod projection;
mod state;

#[path = "../shared/ws_protocol.rs"]
mod ws_protocol;

use cli::Cli;
use control_plane::{AppState, ControlPlane};
use engine_builder::{build_engine_instance, engine_paths, run_engine_ctl_loop};
use projection::{PhaseCallback, StatusCallback};

const MARKET_ID: MarketId = MarketId(1);

// Storage paths
const WAL_PATH: &str = "target/live.lmdb";
const PROJECTION_PATH: &str = "target/live.lmdb/projection.bin";
const SNAPSHOT_FREQUENCY: u64 = 1_000_000;

fn main() -> anyhow::Result<()> {
    let cli = Cli::parse();
    let ring_size_pow2 = cli.ring_size_pow2;
    let wait_strategy = cli.wait_strategy;

    let (pinning, tokio_core) = {
        let cores = core_affinity::get_core_ids().unwrap_or_default();
        if cores.len() < 4 {
            println!(
                "[live] core_affinity: need >=4 cores to pin (engine/wal/projection/tokio); have {}; running unpinned",
                cores.len()
            );
            (None, None)
        } else {
            (
                Some(ThreadPinning {
                    engine_core: Some(cores[0].id),
                    wal_poller_core: Some(cores[1].id),
                    handler_cores: vec![cores[2].id],
                }),
                Some(cores[3].id),
            )
        }
    };

    let cfg = betex::config::Config {
        ring_size_pow2,
        enforce_binary_yes_max_price_ticks: Some(cli.binary_yes_max_price_ticks),
        pinning,
        ..Default::default()
    };
    let mut cfg_unpinned = cfg.clone();
    cfg_unpinned.pinning = None;
    println!("[live] ring_size_pow2={}", cfg.ring_size_pow2);
    println!("[live] wait_strategy={:?}", wait_strategy);

    let default_engine_id = Uuid::nil();
    let (control, ctl_rx) = ControlPlane::new(cfg_unpinned);
    println!(
        "[live] Starting engine={default_engine_id} with 0 markets (create via control plane)."
    );

    // Start the default engine immediately so default HTTP/WS endpoints work.
    {
        let control_inner = Arc::clone(&control.inner);
        let status_callback: Option<StatusCallback> = Some(Arc::new(move |mid, status| {
            if let Some(info) = control_inner.market_infos.write().get_mut(&mid) {
                info.status = status;
            }
            let _ = control_inner.status_broadcast_tx.send(
                ws_protocol::ServerMsg::MarketStatusChanged {
                    market_id: mid,
                    status,
                },
            );
        }));
        let control_inner = Arc::clone(&control.inner);
        let phase_callback: Option<PhaseCallback> = Some(Arc::new(move |mid, phase| {
            if let Some(info) = control_inner.market_infos.write().get_mut(&mid) {
                info.market_phase = phase;
            }
            let _ = control_inner.status_broadcast_tx.send(
                ws_protocol::ServerMsg::MarketPhaseChanged {
                    market_id: mid,
                    phase,
                },
            );
        }));

        let cfg = control.inner.cfg_unpinned.clone();
        let (wal_path, projection_path) = engine_paths(default_engine_id, default_engine_id);
        let engine = build_engine_instance(
            default_engine_id,
            cfg,
            Vec::new(),
            wal_path,
            projection_path,
            status_callback,
            phase_callback,
            control.inner.market_allocator.counter(),
            wait_strategy,
        )?;
        control.insert_engine(engine);
    }

    // Seed control-plane market registry from projection state after recovery catch-up.
    {
        if let Some(engine) = control.get_engine(default_engine_id) {
            let meta = engine.market_meta.read().clone();
            let snap = engine.projection_snapshot.read().clone();

            // Keep served set aligned.
            {
                let mut ids = engine.market_ids.write();
                for mid in meta.keys().copied() {
                    if !ids.contains(&mid) {
                        ids.push(mid);
                    }
                }
            }

            let mut infos = control.inner.market_infos.write();
            for (mid, m) in meta {
                let status = snap
                    .markets
                    .iter()
                    .find(|s| s.market_id == mid)
                    .map(|s| match s.state {
                        betex::book::BookMarketState::Open
                        | betex::book::BookMarketState::Suspended => {
                            ws_protocol::MarketStatus::Active
                        }
                        betex::book::BookMarketState::Closed => ws_protocol::MarketStatus::Closed,
                        betex::book::BookMarketState::Settled => ws_protocol::MarketStatus::Settled,
                        betex::book::BookMarketState::Voided => ws_protocol::MarketStatus::Voided,
                        betex::book::BookMarketState::Halted => ws_protocol::MarketStatus::Active,
                    })
                    .unwrap_or(ws_protocol::MarketStatus::Active);

                infos.insert(
                    mid,
                    ws_protocol::MarketInfo {
                        market_id: mid,
                        engine_id: default_engine_id,
                        name: m.name,
                        market_model: m.market_model,
                        market_kind: m.market_kind,
                        market_phase: snap
                            .markets
                            .iter()
                            .find(|s| s.market_id == mid)
                            .map(|s| s.phase)
                            .unwrap_or(MarketPhase::Pre),
                        runner_ids: m.runner_ids,
                        runner_labels: m.runner_labels,
                        status,
                        created_at: m.created_at_ms,
                    },
                );
            }
        }
    }

    // HTTP Server
    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
    let state = AppState {
        control: control.clone(),
        default_engine_id,
        shutdown_rx,
    };

    let app = axum::Router::new()
        .route("/snapshot", axum::routing::get(snapshot_handler))
        .route(
            "/completed_orders",
            axum::routing::get(completed_orders_handler),
        )
        .route("/risk", axum::routing::get(risk_handler))
        .route(
            "/debug/live_counts",
            axum::routing::get(live_counts_handler),
        )
        .route("/health", axum::routing::get(health_handler))
        .route("/metrics", axum::routing::get(metrics_handler))
        .route("/ws", axum::routing::get(ws_handler))
        .route("/ws_ctl", axum::routing::get(ws_ctl_handler))
        .with_state(state);

    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    println!("\n[live] Server running:");
    println!("  Snapshot:   http://{}/snapshot", addr);
    println!("  Health:     http://{}/health", addr);
    println!("  Metrics:    http://{}/metrics", addr);
    println!("  WS:         ws://{}/ws?depth=10[&market_id=1]", addr);
    println!("  WS+Ctrl:    ws://{}/ws_ctl?depth=10[&market_id=1]", addr);
    println!("\n[live] Run TUI client: cargo run --example live_tui");
    println!("[live] Press Ctrl+C to stop\n");

    let control_for_server = control.clone();
    let server_thread = std::thread::spawn(move || -> anyhow::Result<()> {
        if let Some(id) = tokio_core {
            let _ = core_affinity::set_for_current(core_affinity::CoreId { id });
        }

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()?;

        rt.block_on(async move {
            tokio::spawn(run_engine_ctl_loop(
                control_for_server.clone(),
                ctl_rx,
                default_engine_id,
            ));
            let listener = tokio::net::TcpListener::bind(addr).await?;
            axum::serve(listener, app)
                .with_graceful_shutdown(async move {
                    #[cfg(unix)]
                    {
                        use tokio::signal::unix::{SignalKind, signal};
                        let mut term =
                            signal(SignalKind::terminate()).expect("install SIGTERM handler");
                        tokio::select! {
                            _ = tokio::signal::ctrl_c() => {},
                            _ = term.recv() => {},
                        }
                    }
                    #[cfg(not(unix))]
                    {
                        let _ = tokio::signal::ctrl_c().await;
                    }
                    let _ = shutdown_tx.send(true);
                })
                .await?;
            Ok::<_, anyhow::Error>(())
        })?;

        Ok(())
    });

    server_thread
        .join()
        .map_err(|_| anyhow::anyhow!("server thread panicked"))??;

    println!("\n[live] Shutting down...");
    control.stop_all_blocking()?;

    println!("[live] Done (WAL preserved for next restart)");
    Ok(())
}
