use betex::{
    book::{
        BookEvent, BookEventEnvelope, BookMarketState, CancelCause,
        protocol::{
            command::{Command, CommandKind, MarketState, Persistence, Side, TimeInForce},
            response::Response,
        },
    },
    config::Config,
    engine::{
        consumers::ResponseCallbackHandler,
        core::{EngineBuilder, EngineEvent},
    },
    types::*,
};
use parking_lot::Mutex;
use std::{sync::Arc, time::Duration};

mod common;
use common::{is_lmdb_eprem, two_runner_market_config};

fn temp_wal_path(test_name: &str) -> std::path::PathBuf {
    common::unique_path_under_target("engine-batch-scheduler", test_name)
}

struct CaptureHandler {
    events: Arc<Mutex<Vec<BookEventEnvelope>>>,
}

impl disrupt_rs::EventHandler<EngineEvent> for CaptureHandler {
    fn on_event(
        &mut self,
        event: &EngineEvent,
        _sequence: disrupt_rs::Sequence,
        _end_of_batch: bool,
    ) {
        self.events.lock().push(event.payload.clone());
    }
}

fn place_cmd(correlation_id: u64, market_id: MarketId, account_id: u64) -> Command {
    place_cmd_with_persistence(correlation_id, market_id, account_id, Persistence::Persist)
}

fn lapse_place_cmd(correlation_id: u64, market_id: MarketId, account_id: u64) -> Command {
    place_cmd_with_persistence(correlation_id, market_id, account_id, Persistence::Lapse)
}

fn place_cmd_with_persistence(
    correlation_id: u64,
    market_id: MarketId,
    account_id: u64,
    persistence: Persistence,
) -> Command {
    Command {
        correlation_id: Some(CorrelationId(correlation_id.to_string())),
        metadata: None,
        market_id,
        kind: CommandKind::PlaceOrder {
            runner_id: RunnerId(1),
            account_id: AccountId::from(account_id),
            client_order_id: None,
            side: Side::Yes,
            odds: OddsX10000(20_000),
            stake: Money(100),
            persistence,
            time_in_force: TimeInForce::Gtc,
        },
    }
}

fn close_cmd(market_id: MarketId) -> Command {
    Command {
        correlation_id: Some(CorrelationId(format!("close-{}", market_id.0))),
        metadata: None,
        market_id,
        kind: CommandKind::CloseMarket {
            batch_max_events: betex::book::close_process::MIN_CLOSE_BATCH_EVENTS,
        },
    }
}

fn set_market_state_cmd(market_id: MarketId, state: MarketState) -> Command {
    Command {
        correlation_id: Some(CorrelationId(format!("state-{}-{state:?}", market_id.0))),
        metadata: None,
        market_id,
        kind: CommandKind::SetMarketState { state },
    }
}

fn set_market_phase_cmd(market_id: MarketId, phase: MarketPhase) -> Command {
    Command {
        correlation_id: Some(CorrelationId(format!("phase-{}-{phase:?}", market_id.0))),
        metadata: None,
        market_id,
        kind: CommandKind::SetMarketPhase { phase },
    }
}

fn await_ok(
    rx: flume::Receiver<Result<Response, betex::engine::error::EngineError>>,
) -> anyhow::Result<Response> {
    let response = rx.recv_timeout(Duration::from_secs(2))??;
    assert!(
        matches!(response, Response::Ok { .. }),
        "expected Response::Ok, got {response:?}"
    );
    Ok(response)
}

fn assert_batch_process_completed(
    events: &[BookEventEnvelope],
    market_id: MarketId,
    cancel_cause: CancelCause,
) {
    assert!(
        events.iter().any(|env| {
            env.market_id == market_id
                && matches!(
                    env.event,
                    BookEvent::BatchProcessCompleted {
                        cancel_cause: observed_cause,
                    } if observed_cause == cancel_cause
                )
        }),
        "expected batch process completion for {cancel_cause:?}; events={events:?}"
    );
}

#[test]
fn busy_external_flow_forces_internal_burst_before_more_external_commands() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("busy-quota-interleave");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config {
        engine_poll_timeout_ms: 1_000,
        engine_max_external_before_internal_burst: 5,
        engine_internal_burst_base_commands: 1,
        engine_internal_burst_base_elapsed_ms: 20,
        ..Default::default()
    };

    let captured: Arc<Mutex<Vec<BookEventEnvelope>>> = Arc::new(Mutex::new(Vec::new()));
    let capture = CaptureHandler {
        events: Arc::clone(&captured),
    };

    let market1 = MarketId(1);
    let market2 = MarketId(2);
    let mut engine = match EngineBuilder::with_markets(
        cfg,
        [
            two_runner_market_config(market1),
            two_runner_market_config(market2),
        ],
        &wal_path,
    ) {
        Ok(builder) => builder
            .register_handler(
                ResponseCallbackHandler::default(),
                betex::HandlerRecovery::LiveOnly,
                None,
            )
            .register_handler(capture, betex::HandlerRecovery::LiveOnly, None)
            .build()?,
        Err(err) if is_lmdb_eprem(&err) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(err) => return Err(err),
    };

    let handle = engine.handle();
    for offset in 0..3 {
        await_ok(handle.submit_with_response(place_cmd(offset + 1, market1, 10_000 + offset))?)?;
    }

    await_ok(handle.submit_with_response(close_cmd(market1))?)?;

    let rx_suspend =
        handle.submit_with_response(set_market_state_cmd(market2, MarketState::Suspended))?;
    let rx_open = handle.submit_with_response(set_market_state_cmd(market2, MarketState::Open))?;
    await_ok(rx_suspend)?;
    await_ok(rx_open)?;

    engine.stop()?;

    let events = captured.lock().clone();
    let suspend_ix = events
        .iter()
        .position(|env| {
            env.market_id == market2
                && matches!(
                    env.event,
                    BookEvent::MarketStateChanged {
                        to: BookMarketState::Suspended,
                        ..
                    }
                )
        })
        .expect("suspend event should be captured");
    let open_ix = events
        .iter()
        .position(|env| {
            env.market_id == market2
                && matches!(
                    env.event,
                    BookEvent::MarketStateChanged {
                        to: BookMarketState::Open,
                        ..
                    }
                )
        })
        .expect("open event should be captured");
    let close_batch_positions: Vec<_> = events
        .iter()
        .enumerate()
        .filter_map(|(ix, env)| {
            (env.market_id == market1
                && matches!(
                    env.event,
                    BookEvent::OrderCancelledBatched {
                        cancel_cause: CancelCause::CloseCancel,
                        ..
                    }
                ))
            .then_some(ix)
        })
        .collect();

    assert!(
        close_batch_positions.len() >= 2,
        "expected at least two close-batch events, got {close_batch_positions:?}"
    );
    assert!(suspend_ix < close_batch_positions[1]);
    assert!(close_batch_positions[1] < open_ix);

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn suspend_lapse_batch_continues_without_follow_up_commands() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("suspend-lapse-regression");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config {
        engine_poll_timeout_ms: 5,
        engine_internal_burst_base_commands: 1,
        engine_internal_burst_base_elapsed_ms: 20,
        ..Default::default()
    };

    let captured: Arc<Mutex<Vec<BookEventEnvelope>>> = Arc::new(Mutex::new(Vec::new()));
    let capture = CaptureHandler {
        events: Arc::clone(&captured),
    };
    let market_id = MarketId(10);

    let mut engine =
        match EngineBuilder::with_markets(cfg, [two_runner_market_config(market_id)], &wal_path) {
            Ok(builder) => builder
                .register_handler(
                    ResponseCallbackHandler::default(),
                    betex::HandlerRecovery::LiveOnly,
                    None,
                )
                .register_handler(capture, betex::HandlerRecovery::LiveOnly, None)
                .build()?,
            Err(err) if is_lmdb_eprem(&err) => {
                eprintln!("[skip] lmdb open not permitted in this environment");
                return Ok(());
            }
            Err(err) => return Err(err),
        };

    let handle = engine.handle();
    for offset in 0..3 {
        await_ok(handle.submit_with_response(lapse_place_cmd(
            offset + 1,
            market_id,
            30_000 + offset,
        ))?)?;
    }

    await_ok(
        handle.submit_with_response(set_market_state_cmd(market_id, MarketState::Suspended))?,
    )?;

    std::thread::sleep(Duration::from_millis(100));
    engine.stop()?;

    let events = captured.lock().clone();
    assert_batch_process_completed(&events, market_id, CancelCause::SuspendLapse);

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn pre_to_live_lapse_batch_continues_without_follow_up_commands() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("pre-to-live-lapse");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config {
        engine_poll_timeout_ms: 5,
        engine_internal_burst_base_commands: 1,
        engine_internal_burst_base_elapsed_ms: 20,
        ..Default::default()
    };

    let captured: Arc<Mutex<Vec<BookEventEnvelope>>> = Arc::new(Mutex::new(Vec::new()));
    let capture = CaptureHandler {
        events: Arc::clone(&captured),
    };
    let market_id = MarketId(12);

    let mut engine =
        match EngineBuilder::with_markets(cfg, [two_runner_market_config(market_id)], &wal_path) {
            Ok(builder) => builder
                .register_handler(
                    ResponseCallbackHandler::default(),
                    betex::HandlerRecovery::LiveOnly,
                    None,
                )
                .register_handler(capture, betex::HandlerRecovery::LiveOnly, None)
                .build()?,
            Err(err) if is_lmdb_eprem(&err) => {
                eprintln!("[skip] lmdb open not permitted in this environment");
                return Ok(());
            }
            Err(err) => return Err(err),
        };

    let handle = engine.handle();
    for offset in 0..3 {
        await_ok(handle.submit_with_response(lapse_place_cmd(
            offset + 1,
            market_id,
            40_000 + offset,
        ))?)?;
    }

    await_ok(handle.submit_with_response(set_market_phase_cmd(market_id, MarketPhase::Live))?)?;

    std::thread::sleep(Duration::from_millis(100));
    engine.stop()?;

    let events = captured.lock().clone();
    assert_batch_process_completed(&events, market_id, CancelCause::InPlayLapse);

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}

#[test]
fn recovery_seed_resumes_close_without_external_commands() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("recovery-seed-close");
    let _ = std::fs::remove_dir_all(&wal_path);

    let initial_cfg = Config {
        engine_poll_timeout_ms: 10_000,
        engine_internal_burst_base_commands: 1,
        engine_internal_burst_base_elapsed_ms: 20,
        ..Default::default()
    };
    let market_id = MarketId(11);

    let mut engine = match EngineBuilder::with_markets(
        initial_cfg,
        [two_runner_market_config(market_id)],
        &wal_path,
    ) {
        Ok(builder) => builder
            .register_handler(
                ResponseCallbackHandler::default(),
                betex::HandlerRecovery::LiveOnly,
                None,
            )
            .build()?,
        Err(err) if is_lmdb_eprem(&err) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(err) => return Err(err),
    };

    let handle = engine.handle();
    for offset in 0..3 {
        await_ok(handle.submit_with_response(place_cmd(
            offset + 1,
            market_id,
            20_000 + offset,
        ))?)?;
    }
    await_ok(handle.submit_with_response(close_cmd(market_id))?)?;
    engine.stop()?;

    let resume_cfg = Config {
        engine_poll_timeout_ms: 5,
        engine_internal_burst_base_commands: 1,
        engine_internal_burst_base_elapsed_ms: 20,
        ..Default::default()
    };
    let captured: Arc<Mutex<Vec<BookEventEnvelope>>> = Arc::new(Mutex::new(Vec::new()));
    let capture = CaptureHandler {
        events: Arc::clone(&captured),
    };

    let mut resumed = match EngineBuilder::with_markets(
        resume_cfg,
        [two_runner_market_config(market_id)],
        &wal_path,
    ) {
        Ok(builder) => builder
            .register_handler(capture, betex::HandlerRecovery::LiveOnly, None)
            .build()?,
        Err(err) if is_lmdb_eprem(&err) => {
            eprintln!("[skip] lmdb open not permitted in this environment");
            return Ok(());
        }
        Err(err) => return Err(err),
    };

    std::thread::sleep(Duration::from_millis(100));
    resumed.stop()?;

    let events = captured.lock().clone();
    assert_batch_process_completed(&events, market_id, CancelCause::CloseCancel);

    let _ = std::fs::remove_dir_all(&wal_path);
    Ok(())
}
