use betex::{
    book::protocol::command::{Command, CommandKind, Persistence, Side, TimeInForce},
    config::Config,
    engine::{
        consumers::ResponseCallbackHandler,
        core::{EngineBuilder, EngineEvent},
    },
    types::*,
};
mod common;
use common::{is_lmdb_eprem, two_runner_market_config};
use parking_lot::Mutex;
use std::{collections::HashMap, sync::Arc, time::Duration};

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

struct CaptureHandler {
    events: Arc<Mutex<Vec<(u64, MarketId, u64)>>>,
}

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.seq, event.payload.market_id, event.payload.market_seq));
    }
}

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

#[test]
fn market_sequences_are_contiguous_per_market_when_events_interleave() -> anyhow::Result<()> {
    let wal_path = temp_wal_path("market-seq-interleaving");
    let _ = std::fs::remove_dir_all(&wal_path);

    let cfg = Config::default();
    let m1 = two_runner_market_config(MarketId(1));
    let m2 = two_runner_market_config(MarketId(2));

    let seen: Arc<Mutex<Vec<(u64, MarketId, u64)>>> = Arc::new(Mutex::new(Vec::new()));

    let capture = CaptureHandler {
        events: Arc::clone(&seen),
    };

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

    let handle = engine.handle();
    for cmd in [
        place(1, MarketId(1), AccountId::from(10)),
        place(2, MarketId(2), AccountId::from(20)),
        place(3, MarketId(1), AccountId::from(11)),
        place(4, MarketId(2), AccountId::from(21)),
    ] {
        let rx = handle.submit_with_response(cmd)?;
        let _ = rx.recv_timeout(Duration::from_secs(2))?;
    }

    engine.stop()?;

    let seen = seen.lock().clone();
    assert!(!seen.is_empty());

    for pair in seen.windows(2) {
        assert_eq!(pair[1].0, pair[0].0.saturating_add(1));
    }

    let mut per_market: HashMap<MarketId, Vec<u64>> = HashMap::new();
    for (_seq, market_id, market_seq) in seen {
        per_market.entry(market_id).or_default().push(market_seq);
    }

    let market1 = per_market.get(&MarketId(1)).cloned().unwrap_or_default();
    let market2 = per_market.get(&MarketId(2)).cloned().unwrap_or_default();
    assert!(!market1.is_empty());
    assert!(!market2.is_empty());

    for (ix, seq) in market1.iter().enumerate() {
        assert_eq!(*seq, ix as u64 + 1);
    }
    for (ix, seq) in market2.iter().enumerate() {
        assert_eq!(*seq, ix as u64 + 1);
    }

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