ethl 0.1.44

//! Stream decoded events from RPC providers, with an optional filter
//! Note: The filters events field will be overridden to match the provided events
//!
//! Example usage:
//! ```no_run
//! use ethl::rpc::{events::EventStreamer, config::ProviderSettings, RpcError};
//! use alloy::json_abi::Event;
//! use alloy::rpc::types::Filter;
//! use anyhow::Result;
//! use futures_util::{pin_mut, StreamExt};
//! use tracing::info;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
//!     let providers = ProviderSettings::default();
//!     let events = vec![
//!       Event::parse("Transfer(address indexed from, address indexed to, uint256 value)")?,
//!       Event::parse("Approval(address indexed owner, address indexed spender, uint256 value)")?
//!    ];
//!     let filter = Filter::new().from_block(10000000u64);
//!     let streamer = EventStreamer::new(providers, &events);
//!     // Clone the progress handle before starting the stream and hand it to a metrics
//!     // task; read it on that task's own schedule, independent of stream throughput.
//!     let progress = streamer.progress();
//!     let stream = streamer.stream(Some(&filter)).await;
//!     pin_mut!(stream);
//!     info!("scanned to: {:?}, observed tip: {:?}", progress.scanned_to(), progress.observed_tip());
//!     while let Some(result) = stream.next().await {
//!         let range = result?;
//!         info!("Blocks: {} to {}, Events: {}", range.from_block, range.to_block, range.events.len());
//!         for event in range.events {
//!             info!("  Event: {:?}", event);
//!         }
//!     }
//!     Ok(())
//! }
//! ```
use std::collections::HashMap;

use alloy::{dyn_abi::EventExt, json_abi::Event, rpc::types::Filter};
use alloy_primitives::B256;
use async_stream::stream;
use futures_util::{Stream, StreamExt, pin_mut};

use crate::{
    rpc::{
        DecodedEventRange, LogRange, RpcError, config::ProviderSettings, progress::ScanProgress,
        stream::backfill_then_watch_logs_with_progress,
    },
    storage::codec::decoder::DecodedEventWithHeader,
};

pub struct EventStreamer {
    providers: ProviderSettings,
    events: HashMap<B256, Event>,
    signatures: Vec<String>,
    progress: ScanProgress,
}

impl EventStreamer {
    pub fn new(providers: ProviderSettings, events: &[Event]) -> Self {
        let signatures = events.iter().map(|e| e.signature()).collect::<Vec<_>>();
        let events_map = events
            .iter()
            .map(|e| (e.selector(), e.clone()))
            .collect::<HashMap<_, _>>();
        Self {
            providers,
            events: events_map,
            signatures,
            progress: ScanProgress::default(),
        }
    }

    /// A cloneable handle onto this streamer's scan progress. Clone it before calling
    /// `stream()` and hand the clone to a metrics task — read `scanned_to()` / `observed_tip()`
    /// on whatever schedule that task wants; `EventStreamer` itself emits nothing.
    pub fn progress(&self) -> ScanProgress {
        self.progress.clone()
    }

    /// Decode one raw range into its events, carrying the range's block bounds
    /// and `tip_hash` across unchanged. The hash identifies the range's top
    /// block, which decoding cannot alter — a range whose blocks hold no
    /// matching events still has a canonical identity, and dropping it here
    /// would leave the consumer nothing to anchor recovery on.
    ///
    /// Extracted from the stream body so the decode hop is callable on its own;
    /// private, since it is a step of `stream`, not a second public surface.
    fn decode_range(&self, range: LogRange) -> DecodedEventRange {
        let events = range
            .logs
            .iter()
            .filter_map(|log| {
                let event = self
                    .events
                    .get(log.topic0()?)?
                    .decode_log(log.data())
                    .ok()?;
                let log_block = log.block_number?;
                let log_index = log.log_index? as u32;
                let log_address = log.address();
                Some(DecodedEventWithHeader {
                    log_block,
                    log_index,
                    log_address,
                    event,
                })
            })
            .collect::<Vec<_>>();
        DecodedEventRange {
            from_block: range.from_block,
            to_block: range.to_block,
            events,
            tip_hash: range.tip_hash,
        }
    }

    /// The returned stream is `Send` and may be spawned onto an executor (e.g. `tokio::spawn`).
    pub async fn stream(
        &self,
        filter: Option<&Filter>,
    ) -> impl Stream<Item = Result<DecodedEventRange, RpcError>> + Send {
        stream! {
            let filter = filter.cloned().unwrap_or(Filter::default()).events(&self.signatures);
            let log_stream = backfill_then_watch_logs_with_progress(
                &self.providers,
                Some(&filter),
                Some(&self.progress),
            )
            .await;
            pin_mut!(log_stream);
            while let Some(result) = log_stream.next().await {
                match result {
                    Ok(range) => yield Ok(self.decode_range(range)),
                    Err(e) => {
                        yield Err(e);
                        return;
                    }
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use alloy::json_abi::Event;
    use alloy_primitives::B256;

    use super::EventStreamer;
    use crate::rpc::{LogRange, config::ProviderSettings};

    fn streamer() -> EventStreamer {
        let event =
            Event::parse("Transfer(address indexed from, address indexed to, uint256 value)")
                .expect("the signature is a literal");
        EventStreamer::new(ProviderSettings::default(), std::slice::from_ref(&event))
    }

    /// The decode carries `tip_hash` across unchanged. `DecodedEventRange` is
    /// the surface a consumer anchors recovery on, so a hash dropped here would
    /// leave the by-hash paths' stamp invisible while `LogRange` still passed
    /// its own assertions.
    #[test]
    fn decode_carries_the_tip_hash_across_the_hop() {
        let tip_hash = B256::repeat_byte(0x2a);
        let decoded = streamer().decode_range(LogRange {
            from_block: 100,
            to_block: 101,
            logs: vec![],
            tip_hash: Some(tip_hash),
        });

        assert_eq!((decoded.from_block, decoded.to_block), (100, 101));
        assert!(
            decoded.events.is_empty(),
            "no logs in, no events out — and the range still has an identity"
        );
        assert_eq!(
            decoded.tip_hash,
            Some(tip_hash),
            "a range whose blocks decode to zero events still names its top block"
        );
    }

    /// The `None` half: backfill's ranges reach the consumer unstamped rather
    /// than acquiring a hash the decode never saw.
    #[test]
    fn decode_leaves_an_unstamped_range_unstamped() {
        let decoded = streamer().decode_range(LogRange {
            from_block: 100,
            to_block: 101,
            logs: vec![],
            tip_hash: None,
        });
        assert_eq!(decoded.tip_hash, None);
    }
}