use async_trait::async_trait;
use std::{
    convert::Infallible,
    sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    },
    time::Duration,
};

use betex::{
    book::protocol::command::{Command, CommandKind, Persistence, Side, TimeInForce},
    idempotency::{
        BetexCommandFingerprint, DashMapIdempotencyStore, FingerprintPolicy, IdempotencyConfig,
        IdempotencyError, IdempotencyHelper, IdempotencyOutcome, IdempotencyStore, ReserveOutcome,
        fingerprint_command,
    },
    types::{AccountId, CorrelationId, IdempotencyKey, MarketId, Money, OddsX10000, RunnerId},
};

fn key(value: &str) -> IdempotencyKey {
    IdempotencyKey(value.to_string())
}

#[tokio::test]
async fn concurrent_same_key_and_fingerprint_joins_single_submission() {
    let helper = IdempotencyHelper::with_dashmap_store(
        BetexCommandFingerprint,
        IdempotencyConfig::default(),
    );
    let key = key("11111111-1111-1111-1111-111111111111");
    let submissions = Arc::new(AtomicUsize::new(0));
    let barrier = Arc::new(tokio::sync::Barrier::new(2));

    let helper_a = helper.clone();
    let submissions_a = Arc::clone(&submissions);
    let barrier_a = Arc::clone(&barrier);
    let key_a = key.clone();
    let task_a = tokio::spawn(async move {
        barrier_a.wait().await;
        helper_a
            .execute_with_fingerprint(key_a, 7, (), move |_| {
                let submissions = Arc::clone(&submissions_a);
                async move {
                    submissions.fetch_add(1, Ordering::SeqCst);
                    tokio::time::sleep(Duration::from_millis(40)).await;
                    Ok::<u64, &'static str>(42)
                }
            })
            .await
    });

    let helper_b = helper.clone();
    let submissions_b = Arc::clone(&submissions);
    let barrier_b = Arc::clone(&barrier);
    let key_b = key.clone();
    let task_b = tokio::spawn(async move {
        barrier_b.wait().await;
        helper_b
            .execute_with_fingerprint(key_b, 7, (), move |_| {
                let submissions = Arc::clone(&submissions_b);
                async move {
                    submissions.fetch_add(1, Ordering::SeqCst);
                    tokio::time::sleep(Duration::from_millis(40)).await;
                    Ok::<u64, &'static str>(42)
                }
            })
            .await
    });

    let result_a = task_a.await.expect("join task a");
    let result_b = task_b.await.expect("join task b");

    assert_eq!(submissions.load(Ordering::SeqCst), 1);
    assert_eq!(result_a, Ok(IdempotencyOutcome::Executed(42)));
    assert_eq!(result_b, Ok(IdempotencyOutcome::Executed(42)));
}

#[tokio::test]
async fn same_key_different_fingerprint_returns_conflict() {
    let helper = IdempotencyHelper::with_dashmap_store(
        BetexCommandFingerprint,
        IdempotencyConfig::default(),
    );
    let key = key("22222222-2222-2222-2222-222222222222");

    let (started_tx, started_rx) = tokio::sync::oneshot::channel::<()>();
    let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>();

    let helper_a = helper.clone();
    let key_a = key.clone();
    let task_a = tokio::spawn(async move {
        helper_a
            .execute_with_fingerprint(key_a, 100, (), move |_| async move {
                let _ = started_tx.send(());
                let _ = release_rx.await;
                Ok::<u64, &'static str>(1)
            })
            .await
    });

    started_rx.await.expect("first submission started");

    let result_b = helper
        .execute_with_fingerprint(key.clone(), 101, (), |_| async {
            Ok::<u64, &'static str>(2)
        })
        .await;

    assert_eq!(result_b, Err(IdempotencyError::Conflict));

    let _ = release_tx.send(());
    let result_a = task_a.await.expect("join task a");
    assert_eq!(result_a, Ok(IdempotencyOutcome::Executed(1)));
}

#[tokio::test]
async fn completed_response_replays_within_ttl() {
    let helper = IdempotencyHelper::with_dashmap_store(
        BetexCommandFingerprint,
        IdempotencyConfig::default(),
    );
    let key = key("33333333-3333-3333-3333-333333333333");
    let submissions = Arc::new(AtomicUsize::new(0));

    let first = helper
        .execute_with_fingerprint(key.clone(), 22, (), {
            let submissions = Arc::clone(&submissions);
            move |_| async move {
                submissions.fetch_add(1, Ordering::SeqCst);
                Ok::<u64, &'static str>(9)
            }
        })
        .await;
    assert_eq!(first, Ok(IdempotencyOutcome::Executed(9)));

    let second = helper
        .execute_with_fingerprint(key, 22, (), {
            let submissions = Arc::clone(&submissions);
            move |_| async move {
                submissions.fetch_add(1, Ordering::SeqCst);
                Ok::<u64, &'static str>(10)
            }
        })
        .await;

    assert_eq!(submissions.load(Ordering::SeqCst), 1);
    assert_eq!(second, Ok(IdempotencyOutcome::Replayed(9)));
}

#[tokio::test]
async fn ttl_expiry_causes_reexecution() {
    let helper = IdempotencyHelper::with_dashmap_store(
        BetexCommandFingerprint,
        IdempotencyConfig {
            ttl: Duration::from_millis(25),
        },
    );
    let key = key("44444444-4444-4444-4444-444444444444");
    let submissions = Arc::new(AtomicUsize::new(0));

    let first = helper
        .execute_with_fingerprint(key.clone(), 33, (), {
            let submissions = Arc::clone(&submissions);
            move |_| async move {
                submissions.fetch_add(1, Ordering::SeqCst);
                Ok::<u64, &'static str>(11)
            }
        })
        .await;
    assert_eq!(first, Ok(IdempotencyOutcome::Executed(11)));

    tokio::time::sleep(Duration::from_millis(40)).await;

    let second = helper
        .execute_with_fingerprint(key, 33, (), {
            let submissions = Arc::clone(&submissions);
            move |_| async move {
                submissions.fetch_add(1, Ordering::SeqCst);
                Ok::<u64, &'static str>(12)
            }
        })
        .await;

    assert_eq!(submissions.load(Ordering::SeqCst), 2);
    assert_eq!(second, Ok(IdempotencyOutcome::Executed(12)));
}

#[tokio::test]
async fn submit_failure_releases_inflight_and_allows_retry() {
    let helper = IdempotencyHelper::with_dashmap_store(
        BetexCommandFingerprint,
        IdempotencyConfig::default(),
    );
    let key = key("55555555-5555-5555-5555-555555555555");
    let submissions = Arc::new(AtomicUsize::new(0));

    let first = helper
        .execute_with_fingerprint(key.clone(), 44, (), {
            let submissions = Arc::clone(&submissions);
            move |_| async move {
                submissions.fetch_add(1, Ordering::SeqCst);
                Err::<u64, &'static str>("boom")
            }
        })
        .await;
    assert_eq!(first, Err(IdempotencyError::Submit("boom")));

    let second = helper
        .execute_with_fingerprint(key, 44, (), {
            let submissions = Arc::clone(&submissions);
            move |_| async move {
                submissions.fetch_add(1, Ordering::SeqCst);
                Ok::<u64, &'static str>(13)
            }
        })
        .await;

    assert_eq!(submissions.load(Ordering::SeqCst), 2);
    assert_eq!(second, Ok(IdempotencyOutcome::Executed(13)));
}

#[derive(Clone)]
struct AlwaysInFlightStore;

#[async_trait]
impl IdempotencyStore<u64> for AlwaysInFlightStore {
    type Error = Infallible;

    async fn reserve(
        &self,
        _key: &IdempotencyKey,
        _fingerprint: u64,
        _expires_at_ms: i64,
    ) -> Result<ReserveOutcome<u64>, Self::Error> {
        Ok(ReserveOutcome::InFlight)
    }

    async fn complete(
        &self,
        _key: &IdempotencyKey,
        _fingerprint: u64,
        _response: u64,
        _expires_at_ms: i64,
    ) -> Result<(), Self::Error> {
        Ok(())
    }

    async fn release_inflight(
        &self,
        _key: &IdempotencyKey,
        _fingerprint: u64,
    ) -> Result<(), Self::Error> {
        Ok(())
    }
}

#[derive(Clone, Copy)]
struct U64Fingerprint;

impl FingerprintPolicy<u64> for U64Fingerprint {
    fn fingerprint(&self, payload: &u64) -> u64 {
        *payload
    }
}

#[tokio::test]
async fn inflight_elsewhere_is_returned_when_no_local_waiter_exists() {
    let helper = IdempotencyHelper::new(
        AlwaysInFlightStore,
        U64Fingerprint,
        IdempotencyConfig::default(),
    );
    let key = key("66666666-6666-6666-6666-666666666666");

    let result = helper
        .execute_with_fingerprint(key, 77, 77_u64, |_| async { Ok::<u64, &'static str>(77) })
        .await;

    assert_eq!(result, Err(IdempotencyError::InFlightElsewhere));
}

#[tokio::test]
async fn expired_inflight_entry_is_not_reclaimed_by_reserve() {
    let store = DashMapIdempotencyStore::<u64>::new();
    let key = key("77777777-7777-7777-7777-777777777777");

    let first = store
        .reserve(&key, 501, 0)
        .await
        .expect("reserve initial in-flight");
    assert_eq!(first, ReserveOutcome::Reserved);

    let same = store
        .reserve(&key, 501, 0)
        .await
        .expect("reserve same fingerprint");
    assert_eq!(same, ReserveOutcome::InFlight);

    let conflict = store
        .reserve(&key, 502, 0)
        .await
        .expect("reserve conflicting fingerprint");
    assert_eq!(conflict, ReserveOutcome::Conflict);

    store
        .release_inflight(&key, 501)
        .await
        .expect("release in-flight");

    let after_release = store
        .reserve(&key, 501, 0)
        .await
        .expect("reserve after releasing in-flight");
    assert_eq!(after_release, ReserveOutcome::Reserved);
}

#[tokio::test]
async fn pruner_does_not_evict_inflight_entries() {
    let store = DashMapIdempotencyStore::<u64>::with_prune_interval(Duration::from_millis(5));
    let key = key("88888888-8888-8888-8888-888888888888");

    let first = store
        .reserve(&key, 601, 0)
        .await
        .expect("reserve initial in-flight");
    assert_eq!(first, ReserveOutcome::Reserved);

    tokio::time::sleep(Duration::from_millis(25)).await;

    let second = store
        .reserve(&key, 601, 0)
        .await
        .expect("reserve after prune cycles");
    assert_eq!(second, ReserveOutcome::InFlight);
}

#[test]
fn betex_command_fingerprint_is_stable_for_same_command_input() {
    let cmd_a = Command {
        correlation_id: Some(CorrelationId(1.to_string())),
        metadata: None,
        market_id: MarketId(9),
        kind: CommandKind::PlaceOrder {
            runner_id: RunnerId(1),
            account_id: AccountId::from(10),
            client_order_id: None,
            side: Side::Yes,
            odds: OddsX10000(20_000),
            stake: Money(1_000),
            persistence: Persistence::Persist,
            time_in_force: TimeInForce::Gtc,
        },
    };

    let mut cmd_b = cmd_a.clone();
    cmd_b.correlation_id = Some(CorrelationId(2.to_string()));

    let mut cmd_c = cmd_a.clone();
    if let CommandKind::PlaceOrder { stake, .. } = &mut cmd_c.kind {
        *stake = Money(2_000);
    }

    let policy = BetexCommandFingerprint;
    let fp_a = policy.fingerprint(&cmd_a);
    let fp_b = policy.fingerprint(&cmd_b);
    let fp_c = policy.fingerprint(&cmd_c);

    assert_eq!(fp_a, fp_b);
    assert_ne!(fp_a, fp_c);
    assert_eq!(fp_a, fingerprint_command(&cmd_a));
}
