ethl 0.1.44

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::progress::ScanProgress;
use ethl::rpc::stream::backfill_then_watch_logs;
use ethl::rpc::stream::backfill_then_watch_logs_with_progress;
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 range = log_range_result?;
    // The returned range should cover the requested block
    assert!(
        range.from_block <= 34762687 && range.to_block >= 34762687,
        "returned range [{}, {}] should contain requested block 34762687",
        range.from_block,
        range.to_block
    );
    assert!(
        !range.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 range = log_range_result?;
    // Should cover both blocks or at least the one(s) with logs
    assert!(
        range.from_block <= 34762688 && range.to_block >= 34762687,
        "range should cover blocks 34762687-34762688"
    );
    assert!(!range.logs.is_empty(), "two-block range should return logs");

    Ok(())
}

/// G.26 — connect site: `init_to_block.unwrap_or(b - 2)` evaluated `b - 2` eagerly even
/// though a bounded fetch never uses it (`init_to_block` is `Some`), so a provider
/// answering `eth_blockNumber` with 0 or 1 panicked before a bounded fetch could start.
/// Height 0 clamps to an empty range and yields nothing (archives are 1-based); height 1
/// does fetch. Both must end without going live — a bounded fetch never does.
///
/// Mutation check: reverting `b.saturating_sub(2)` to `b - 2` at the connect site
/// (`stream.rs`, `init_to_block.unwrap_or(...)`) panics both cases below with "attempt to
/// subtract with overflow".
#[tokio::test]
async fn connect_site_low_block_number_does_not_panic_on_bounded_fetch() -> Result<()> {
    // Height 0: a bounded [0, 0] request clamps to an empty range and yields nothing,
    // but the connect site still computes (and must saturate) the tip it never uses.
    let asserter = Asserter::new();
    asserter.push_success(&serde_json::json!("0x0"));

    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);

    assert!(
        stream.next().await.is_none(),
        "a [0, 0] bounded fetch clamps to nothing, height 0 or not"
    );

    // Height 1: a bounded [1, 1] request does fetch, and must still yield its range and
    // end rather than go live.
    let logs: serde_json::Value = serde_json::from_str(include_str!(
        "fixtures/transfer_getlogs_34762687_34762688.json"
    ))?;
    let asserter = Asserter::new();
    asserter.push_success(&serde_json::json!("0x1"));
    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(1u64).to_block(1u64);
    let stream = backfill_then_watch_logs(&settings, Some(&filter)).await;
    pin_mut!(stream);

    let range = stream
        .next()
        .await
        .expect("bounded [1, 1] should yield a range")?;
    assert_eq!((range.from_block, range.to_block), (1, 1));
    assert!(!range.logs.is_empty(), "the fixture returns logs");
    assert!(
        stream.next().await.is_none(),
        "a bounded fetch ends after its range, never going live"
    );

    Ok(())
}

/// G.26 — re-poll site: `tip = b - 2` ran on every backfill pass, so a provider answering
/// `eth_blockNumber` with 0 or 1 on a re-poll panicked mid-loop, after the connect site had
/// already succeeded with a healthy height. A lead of 98 blocks over `from_block(0)` clears
/// the legacy `tip - from_block < 30` dormancy check, so one backfill pass runs before the
/// re-poll below is reached; the re-poll then hands off to the live phase, which fails
/// closed (`from_mock` configures no WS endpoint) rather than looping again.
///
/// Mutation check: reverting `b.saturating_sub(2)` to `b - 2` at the re-poll site
/// (`stream.rs`, `tip = ...`) panics after the one backfill pass below, with "attempt to
/// subtract with overflow".
#[tokio::test]
async fn repoll_site_low_block_number_does_not_panic_after_one_backfill_pass() -> Result<()> {
    let logs: serde_json::Value = serde_json::from_str(include_str!(
        "fixtures/transfer_getlogs_34762687_34762688.json"
    ))?;

    let asserter = Asserter::new();
    // Connect: height 100 -> tip 98, 98 blocks above from_block(0) -- clears the
    // legacy dormancy check, so one backfill pass runs before any re-poll.
    asserter.push_success(&serde_json::json!("0x64"));
    asserter.push_success(&logs);
    // Re-poll: the provider's height drops to 1 -- the site under test.
    asserter.push_success(&serde_json::json!("0x1"));

    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);
    let stream = backfill_then_watch_logs(&settings, Some(&filter)).await;
    pin_mut!(stream);

    let backfilled = stream
        .next()
        .await
        .expect("the one backfill pass should yield a range")?;
    assert_eq!((backfilled.from_block, backfilled.to_block), (1, 98));
    assert!(!backfilled.logs.is_empty(), "the fixture returns logs");

    let live = stream
        .next()
        .await
        .expect("the re-poll tip of 0 ends backfill and hands off to the live phase");
    assert!(
        matches!(live, Err(RpcError::AllProvidersSuspended(_))),
        "from_mock configures no WS endpoint; the live phase must fail closed, not hang"
    );

    assert!(
        stream.next().await.is_none(),
        "the stream ends after the live-phase error"
    );

    Ok(())
}

/// G.2.2 §4.1 — the load-bearing test: a bounded range with zero matching events must still
/// reach the consumer as `(from, to, [])` (`backfill.rs:419` yields on every successful fetch,
/// not just non-empty ones), and `ScanProgress::scanned_to()` must report the top of that range
/// — proof the handle records real scan activity, not just yielded data.
///
/// Mutation check (verified manually, not re-run automatically here since that would require
/// reverting shipped code): deleting the progress record at `stream.rs:98` — the `Ok(range) =>`
/// arm's `p.record_scanned_to(range.1)` — while leaving every other record site intact turns
/// this test RED with `scanned_to()` reporting `None` instead of `Some(34762688)`.
#[tokio::test]
async fn bounded_range_with_no_matching_events_still_records_scan_progress() -> Result<()> {
    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(&serde_json::json!([])); // range fetched successfully, zero matching events
    // backfill.rs's empty-batch guard re-polls eth_blockNumber to confirm the empty response
    // isn't a lagging provider silently returning [] past its own tip.
    asserter.push_success(&serde_json::json!("0x2126fc2"));

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

    let filter = Filter::new().from_block(34762687).to_block(34762688);
    let progress = ScanProgress::default();
    assert_eq!(
        progress.scanned_to(),
        None,
        "progress must be unset before the stream is driven"
    );

    let stream =
        backfill_then_watch_logs_with_progress(&settings, Some(&filter), Some(&progress)).await;
    pin_mut!(stream);

    let result = stream.next().await;
    assert!(
        result.is_some(),
        "a zero-event range must still reach the consumer as (from, to, [])"
    );
    let range = result.unwrap()?;
    assert_eq!((range.from_block, range.to_block), (34762687, 34762688));
    assert!(
        range.logs.is_empty(),
        "the fixture returns zero matching events"
    );

    assert_eq!(
        progress.scanned_to(),
        Some(34762688),
        "scanned_to() must report the top of the range even though nothing matched the filter"
    );

    Ok(())
}

/// G.2.2 §4.2 — unset: `scanned_to()` stays `None` across a `[0, 0]` bounded request. Per the
/// block-0 policy (`backfill.rs:290-293`'s `.max(1)` clamp), nothing is ever fetched for this
/// bound, so recording from the `!yielded_any` arm (`stream.rs:109-114`) would wrongly report
/// `Some(0)` for a block that was never scanned. `zero_to_zero_bounded_range_terminates` is the
/// existing vehicle for this exact bound; this test reuses its OS-thread watchdog structure
/// (see that test's doc comment for why a same-task `tokio::time::timeout` cannot substitute).
#[test]
fn zero_to_zero_bounded_range_progress_stays_unset() {
    let (tx, rx) = mpsc::channel::<(Vec<Result<(u64, u64, usize), RpcError>>, Option<u64>)>();

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

        let (collected, scanned_to) = 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; a [0, 0]
            // request never reaches eth_getLogs once the cursor clamps to 1 (empty range), so
            // this second response is queued only to keep the mock available should a mutated
            // cursor 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 progress = ScanProgress::default();
            let stream =
                backfill_then_watch_logs_with_progress(&settings, Some(&filter), Some(&progress))
                    .await;
            pin_mut!(stream);

            let mut collected = Vec::new();
            while let Some(item) = stream.next().await {
                collected.push(item.map(|r| (r.from_block, r.to_block, r.logs.len())));
            }
            (collected, progress.scanned_to())
        });

        // See `zero_to_zero_bounded_range_terminates`: dropping the send result on a timed-out
        // receiver is fine, this thread's only job past that point is to exit.
        let _ = tx.send((collected, scanned_to));
    });

    let (collected, scanned_to) = 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:?}"
    );
    assert_eq!(
        scanned_to, None,
        "block 0 is never scanned; scanned_to() must stay None, not Some(0)"
    );
}

/// 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(|r| (r.from_block, r.to_block, r.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 range = result.unwrap()?;
    assert_eq!(
        range.from_block, 1,
        "the first yielded range must start at block 1, skipping genesis"
    );

    Ok(())
}