use assert_matches::assert_matches;
use reth_transaction_pool::{
    noop::MockTransactionValidator,
    test_utils::{MockTransactionFactory, TestPoolBuilder},
    FullTransactionEvent, PoolTransaction, TransactionEvent, TransactionListenerKind,
    TransactionOrigin, TransactionPool,
};
use std::{future::poll_fn, task::Poll};
use tokio_stream::StreamExt;

#[tokio::test(flavor = "multi_thread")]
async fn txpool_listener_by_hash() {
    let txpool = TestPoolBuilder::default();
    let mut mock_tx_factory = MockTransactionFactory::default();
    let transaction = mock_tx_factory.create_eip1559();

    let result = txpool
        .add_transaction_and_subscribe(TransactionOrigin::External, transaction.transaction.clone())
        .await;
    assert_matches!(result, Ok(_));

    let mut events = result.unwrap();
    assert_matches!(events.next().await, Some(TransactionEvent::Pending));

    let removed_txs = txpool.remove_transactions(vec![*transaction.transaction.hash()]);
    assert_eq!(transaction.transaction.hash(), removed_txs[0].transaction.hash());

    assert_matches!(events.next().await, Some(TransactionEvent::Discarded));
}

#[tokio::test(flavor = "multi_thread")]
async fn txpool_listener_pending_promotions_propagate_only() {
    let txpool =
        TestPoolBuilder::default().with_validator(MockTransactionValidator::no_propagate_local());
    let mut mock_tx_factory = MockTransactionFactory::default();

    // Create two transactions from the same sender with nonces 0 (local, non-propagatable) and 1
    let mut tx_local = mock_tx_factory.create_eip1559();
    let mut tx_external = mock_tx_factory.create_eip1559();

    // Ensure same sender
    let sender = *tx_local.transaction.get_sender();
    tx_external.transaction.set_sender(sender);

    // Set explicit nonces to create a nonce gap scenario
    tx_local.transaction.set_nonce(0);
    tx_external.transaction.set_nonce(1);

    let hash_local = *tx_local.transaction.hash();
    let hash_external = *tx_external.transaction.hash();

    // Listeners: propagate-only and all
    let mut listener_network = txpool.pending_transactions_listener();
    let mut listener_all = txpool.pending_transactions_listener_for(TransactionListenerKind::All);

    // Insert the higher-nonce external tx first; it should be queued due to nonce gap
    let res =
        txpool.add_transaction(TransactionOrigin::External, tx_external.transaction.clone()).await;
    assert!(res.is_ok());

    // Now insert the local tx with nonce 0; it becomes pending and should promote the external tx
    let res = txpool.add_transaction(TransactionOrigin::Local, tx_local.transaction.clone()).await;
    assert!(res.is_ok());

    // All-listener should receive both pending hashes in order: inserted local first, then promoted
    // external
    let inserted_all_first = listener_all.recv().await.unwrap();
    let inserted_all_second = listener_all.recv().await.unwrap();
    assert_eq!(inserted_all_first, hash_local);
    assert_eq!(inserted_all_second, hash_external);

    // Propagate-only listener should receive only the external tx (local is non-propagatable)
    let inserted_network = listener_network.recv().await.unwrap();
    assert_eq!(inserted_network, hash_external);
}

#[tokio::test(flavor = "multi_thread")]
async fn txpool_listener_replace_event() {
    let txpool = TestPoolBuilder::default();
    let mut mock_tx_factory = MockTransactionFactory::default();
    let transaction = mock_tx_factory.create_eip1559();

    let mut all_tx_events = txpool.all_transactions_event_listener();

    let old_transaction = transaction.transaction.clone();
    let mut result = txpool
        .add_transaction_and_subscribe(TransactionOrigin::External, old_transaction.clone())
        .await;
    assert_matches!(result, Ok(_));

    let mut events = result.unwrap();
    assert_matches!(events.next().await, Some(TransactionEvent::Pending));
    assert_matches!(all_tx_events.next().await, Some(FullTransactionEvent::Pending(hash)) if hash == *old_transaction.get_hash());

    // add replace tx.
    let replace_transaction = transaction.transaction.clone().rng_hash().inc_price();
    result = txpool
        .add_transaction_and_subscribe(TransactionOrigin::External, replace_transaction.clone())
        .await;
    assert_matches!(result, Ok(_));

    let mut new_events = result.unwrap();
    assert_matches!(new_events.next().await, Some(TransactionEvent::Pending));

    // The listener of old transaction should receive replaced event.
    assert_matches!(events.next().await, Some(TransactionEvent::Replaced(hash)) if hash == *replace_transaction.get_hash());

    // The listener of all should receive one pending event of new transaction and one replaced
    // event of old transaction.
    assert_matches!(all_tx_events.next().await, Some(FullTransactionEvent::Pending(hash)) if hash == *replace_transaction.get_hash());
    assert_matches!(all_tx_events.next().await, Some(FullTransactionEvent::Replaced { transaction, replaced_by }) if *transaction.transaction.get_hash() == *old_transaction.get_hash() && replaced_by == *replace_transaction.get_hash());
}

#[tokio::test(flavor = "multi_thread")]
async fn txpool_listener_queued_event() {
    let txpool = TestPoolBuilder::default();
    let mut mock_tx_factory = MockTransactionFactory::default();
    let transaction = mock_tx_factory.create_eip1559().transaction.inc_nonce();

    let mut all_tx_events = txpool.all_transactions_event_listener();

    let result = txpool
        .add_transaction_and_subscribe(TransactionOrigin::External, transaction.clone())
        .await;
    assert_matches!(result, Ok(_));

    let mut events = result.unwrap();
    assert_matches!(events.next().await, Some(TransactionEvent::Queued));

    // The listener of all should receive queued event as well.
    assert_matches!(all_tx_events.next().await, Some(FullTransactionEvent::Queued(hash,_ )) if hash == *transaction.get_hash());
}

#[tokio::test(flavor = "multi_thread")]
async fn txpool_listener_invalid_event() {
    let txpool =
        TestPoolBuilder::default().with_validator(MockTransactionValidator::return_invalid());
    let mut mock_tx_factory = MockTransactionFactory::default();
    let transaction = mock_tx_factory.create_eip1559().transaction;

    let mut all_tx_events = txpool.all_transactions_event_listener();

    let result = txpool
        .add_transaction_and_subscribe(TransactionOrigin::External, transaction.clone())
        .await;
    assert_matches!(result, Err(_));

    // The listener of all should receive invalid event.
    assert_matches!(all_tx_events.next().await, Some(FullTransactionEvent::Invalid(hash)) if hash == *transaction.get_hash());
}

#[tokio::test(flavor = "multi_thread")]
async fn txpool_listener_all() {
    let txpool = TestPoolBuilder::default();
    let mut mock_tx_factory = MockTransactionFactory::default();
    let transaction = mock_tx_factory.create_eip1559();

    let mut all_tx_events = txpool.all_transactions_event_listener();

    let added_result =
        txpool.add_transaction(TransactionOrigin::External, transaction.transaction.clone()).await;
    assert_matches!(added_result, Ok(outcome) if outcome.hash == *transaction.transaction.get_hash());

    assert_matches!(
        all_tx_events.next().await,
        Some(FullTransactionEvent::Pending(hash)) if hash == *transaction.transaction.get_hash()
    );

    let removed_txs = txpool.remove_transactions(vec![*transaction.transaction.hash()]);

    assert_eq!(transaction.transaction.hash(), removed_txs[0].transaction.hash());

    assert_matches!(all_tx_events.next().await, Some(FullTransactionEvent::Discarded(hash)) if hash == *transaction.transaction.get_hash());
}

#[tokio::test(flavor = "multi_thread")]
async fn txpool_listener_propagate_only() {
    let txpool =
        TestPoolBuilder::default().with_validator(MockTransactionValidator::no_propagate_local());
    let mut mock_tx_factory = MockTransactionFactory::default();
    let transaction = mock_tx_factory.create_eip1559();
    let expected = *transaction.hash();
    let mut listener_network = txpool.pending_transactions_listener();
    let mut listener_all = txpool.pending_transactions_listener_for(TransactionListenerKind::All);
    let result =
        txpool.add_transaction(TransactionOrigin::Local, transaction.transaction.clone()).await;
    assert!(result.is_ok());

    let inserted = listener_all.recv().await.unwrap();
    assert_eq!(inserted, expected);

    poll_fn(|cx| {
        // no propagation
        assert!(listener_network.poll_recv(cx).is_pending());
        Poll::Ready(())
    })
    .await;
}

#[tokio::test(flavor = "multi_thread")]
async fn txpool_listener_new_propagate_only() {
    let txpool =
        TestPoolBuilder::default().with_validator(MockTransactionValidator::no_propagate_local());
    let mut mock_tx_factory = MockTransactionFactory::default();
    let transaction = mock_tx_factory.create_eip1559();
    let expected = *transaction.hash();
    let mut listener_network = txpool.new_transactions_listener();
    let mut listener_all = txpool.new_transactions_listener_for(TransactionListenerKind::All);
    let result =
        txpool.add_transaction(TransactionOrigin::Local, transaction.transaction.clone()).await;
    assert!(result.is_ok());

    let inserted = listener_all.recv().await.unwrap();
    let actual = *inserted.transaction.hash();
    assert_eq!(actual, expected);

    poll_fn(|cx| {
        // no propagation
        assert!(listener_network.poll_recv(cx).is_pending());
        Poll::Ready(())
    })
    .await;
}

#[tokio::test(flavor = "multi_thread")]
async fn txpool_listener_blob_sidecar() {
    let txpool =
        TestPoolBuilder::default().with_validator(MockTransactionValidator::no_propagate_local());
    let mut mock_tx_factory = MockTransactionFactory::default();
    let blob_transaction = mock_tx_factory.create_eip4844();
    let expected = *blob_transaction.hash();
    let mut listener_blob = txpool.blob_transaction_sidecars_listener();
    let result = txpool
        .add_transaction(TransactionOrigin::Local, blob_transaction.transaction.clone())
        .await;
    assert!(result.is_ok());

    let inserted = listener_blob.recv().await.unwrap();
    assert_eq!(*inserted.tx_hash, expected);
}
