ethl 0.1.33

use alloy::providers::{Provider, ProviderBuilder};
use alloy::rpc::types::Filter;
use alloy::transports::mock::Asserter;
use anyhow::Result;
use ethl::rpc::RpcError;
use ethl::rpc::config::ProviderSettings;
use ethl::rpc::stream::backfill_then_watch_logs;
use futures_util::pin_mut;
use futures_util::stream::StreamExt;
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
use tokio::runtime::Builder;

/// Regression test: single-block bounded range should yield at least one LogRange.
/// This tests the case where from_block == to_block, which previously yielded nothing
/// due to the loop condition being `while tip > from_block` instead of `while tip >= from_block`.
#[tokio::test]
async fn test_single_block_bounded_range() -> Result<()> {
    let logs: serde_json::Value = serde_json::from_str(include_str!(
        "fixtures/transfer_getlogs_34762687_34762688.json"
    ))?;

    let asserter = Asserter::new();
    // backfill_then_watch_logs calls eth_blockNumber before backfilling
    asserter.push_success(&serde_json::json!("0x2126fc2")); // 34762690, any block > to_block
    asserter.push_success(&logs);

    let provider = ProviderBuilder::new()
        .disable_recommended_fillers()
        .connect_mocked_client(asserter)
        .erased();
    let settings = ProviderSettings::from_mock(provider);

    // Single-block bounded range: from_block == to_block (34762687)
    let filter = Filter::new().from_block(34762687).to_block(34762687);
    let stream = backfill_then_watch_logs(&settings, Some(&filter)).await;
    pin_mut!(stream);

    // Should yield at least one result (the LogRange for block 34762687)
    let result = stream.next().await;
    assert!(
        result.is_some(),
        "single-block range should yield at least one result"
    );

    let log_range_result = result.unwrap();
    assert!(
        log_range_result.is_ok(),
        "result should be Ok, not an error"
    );

    let (from, to, logs) = log_range_result?;
    // The returned range should cover the requested block
    assert!(
        from <= 34762687 && to >= 34762687,
        "returned range [{}, {}] should contain requested block 34762687",
        from,
        to
    );
    assert!(
        !logs.is_empty(),
        "single-block range should return logs from the fixture"
    );

    Ok(())
}

/// Regression guard: two-block bounded range should still work correctly.
/// This ensures the fix doesn't break existing multi-block range behavior.
#[tokio::test]
async fn test_two_block_bounded_range() -> Result<()> {
    let logs: serde_json::Value = serde_json::from_str(include_str!(
        "fixtures/transfer_getlogs_34762687_34762688.json"
    ))?;

    let asserter = Asserter::new();
    // backfill_then_watch_logs calls eth_blockNumber before backfilling
    asserter.push_success(&serde_json::json!("0x2126fc2")); // 34762690, any block > to_block
    asserter.push_success(&logs);

    let provider = ProviderBuilder::new()
        .disable_recommended_fillers()
        .connect_mocked_client(asserter)
        .erased();
    let settings = ProviderSettings::from_mock(provider);

    // Two-block bounded range
    let filter = Filter::new().from_block(34762687).to_block(34762688);
    let stream = backfill_then_watch_logs(&settings, Some(&filter)).await;
    pin_mut!(stream);

    let result = stream.next().await;
    assert!(
        result.is_some(),
        "two-block range should yield at least one result"
    );

    let log_range_result = result.unwrap();
    assert!(log_range_result.is_ok(), "result should be Ok");

    let (from, to, logs) = log_range_result?;
    // Should cover both blocks or at least the one(s) with logs
    assert!(
        from <= 34762688 && to >= 34762687,
        "range should cover blocks 34762687-34762688"
    );
    assert!(!logs.is_empty(), "two-block range should return logs");

    Ok(())
}

/// Regression guard: unbounded filter should be unaffected by the fix.
/// An unbounded filter with no websocket provider should yield an error immediately
/// (AllProvidersSuspended) without consuming the eth_getLogs response we queued.
/// If it tries to consume eth_getLogs, the Asserter will panic, indicating the path was broken.
#[tokio::test]
async fn test_unbounded_filter_unchanged() -> Result<()> {
    let asserter = Asserter::new();
    // For unbounded filter, last_block is called and should return a value such that
    // b - 2 equals from_block (0). So last_block should return 2.
    asserter.push_success(&serde_json::json!("0x2")); // Returns 2, so b - 2 = 0
    // Do NOT push eth_getLogs; if the stream tries to call it, Asserter panics

    let provider = ProviderBuilder::new()
        .disable_recommended_fillers()
        .connect_mocked_client(asserter)
        .erased();
    let settings = ProviderSettings::from_mock(provider);

    // Unbounded filter (no to_block)
    let filter = Filter::new().from_block(0);
    let stream = backfill_then_watch_logs(&settings, Some(&filter)).await;
    pin_mut!(stream);

    let result = stream.next().await;
    assert!(
        result.is_some(),
        "unbounded stream should yield a result (error or data)"
    );

    // With no WSS providers, stream_heads_with_logs should yield an error immediately
    let item = result.unwrap();
    assert!(
        item.is_err(),
        "unbounded stream without WSS should yield AllProvidersSuspended error"
    );

    Ok(())
}

/// Regression test: a bounded `[0, 0]` range must terminate, yielding nothing.
///
/// Archives start at block 1 (deliberate policy: existing archives and
/// snapshots are 1-based, and block 0 is genesis, which carries no logs on any
/// indexed chain), so a `[0, 0]` request has no blocks in range once the
/// cursor clamps to 1 — `backfill` never enters its fetch loop and yields
/// nothing at all. Without the matching advance in `stream.rs`, that empty
/// yield would leave `from_block` unchanged, and `tip >= from_block`'s loop
/// head would re-enter with identical state forever — a synchronous,
/// zero-yield, zero-I/O spin inside a single task poll (no `.await` executes
/// before the next iteration), so a same-task `tokio::time::timeout` can never
/// fire against it. The watchdog for this test therefore has to live on a
/// separate OS thread: the drive runs on its own current-thread Tokio runtime
/// and reports back over an mpsc channel, and the test thread applies the real
/// timeout to `recv_timeout`. On a spinning implementation the channel recv
/// times out and the spinning thread leaks and dies with the test process,
/// which is expected, acceptable behavior for this watchdog rather than
/// something to engineer away. Termination is the assertion that matters.
#[test]
fn zero_to_zero_bounded_range_terminates() {
    let (tx, rx) = mpsc::channel::<Vec<Result<(u64, u64, usize), RpcError>>>();

    thread::spawn(move || {
        let runtime = Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("failed to build current-thread runtime");

        let collected = runtime.block_on(async {
            let logs: serde_json::Value = serde_json::from_str(include_str!(
                "fixtures/transfer_getlogs_34762687_34762688.json"
            ))
            .expect("fixture parses as JSON");

            let asserter = Asserter::new();
            // backfill_then_watch_logs calls eth_blockNumber before backfilling; stream.rs:45
            // evaluates `b - 2` eagerly even for a bounded filter, so this must be queued first.
            // A [0, 0] request never reaches eth_getLogs once the cursor clamps to 1 (the
            // range is empty), so this second response is queued only to keep the mock
            // available for a mutated cursor that does reach it.
            asserter.push_success(&serde_json::json!("0x2126fc2"));
            asserter.push_success(&logs);

            let provider = ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased();
            let settings = ProviderSettings::from_mock(provider);

            let filter = Filter::new().from_block(0u64).to_block(0u64);
            let stream = backfill_then_watch_logs(&settings, Some(&filter)).await;
            pin_mut!(stream);

            let mut collected = Vec::new();
            while let Some(item) = stream.next().await {
                collected.push(item.map(|(from, to, logs)| (from, to, logs.len())));
            }
            collected
        });

        // If the watchdog already timed out on the test thread, the receiver is gone;
        // dropping the send result is fine, this thread's only job past that point is
        // to exit (which it does, on the fixed code) or to keep spinning forever, which
        // is the exact spin this test exists to catch.
        let _ = tx.send(collected);
    });

    let collected = rx
        .recv_timeout(Duration::from_secs(10))
        .expect("bounded [0, 0] must terminate");

    assert_eq!(
        collected.len(),
        0,
        "bounded [0, 0] should terminate yielding zero ranges, got {collected:?}"
    );
}

/// Regression test: a bounded range starting at block 0 must fetch starting at
/// block 1, not block 0.
///
/// Archives start at block 1: existing archives, snapshots, and the python
/// tooling were produced and verified against 1-based starts, and block 0 is
/// genesis, which carries no logs on any indexed chain. Skipping it is
/// deliberate archive-layout policy, not a defect to "fix" back to block 0.
#[tokio::test]
async fn bounded_range_from_zero_starts_at_block_one() -> Result<()> {
    let logs: serde_json::Value = serde_json::from_str(include_str!(
        "fixtures/transfer_getlogs_34762687_34762688.json"
    ))?;

    let asserter = Asserter::new();
    // backfill_then_watch_logs calls eth_blockNumber before backfilling; stream.rs:45
    // evaluates `b - 2` eagerly even for a bounded filter, so this must be queued first.
    asserter.push_success(&serde_json::json!("0x2126fc2"));
    asserter.push_success(&logs);

    let provider = ProviderBuilder::new()
        .disable_recommended_fillers()
        .connect_mocked_client(asserter)
        .erased();
    let settings = ProviderSettings::from_mock(provider);

    let filter = Filter::new().from_block(0u64).to_block(2u64);
    let stream = backfill_then_watch_logs(&settings, Some(&filter)).await;
    pin_mut!(stream);

    let result = stream.next().await;
    assert!(
        result.is_some(),
        "range starting at block 0 should yield at least one result"
    );

    let (from, _to, _logs) = result.unwrap()?;
    assert_eq!(
        from, 1,
        "the first yielded range must start at block 1, skipping genesis"
    );

    Ok(())
}