feat: indexer refactor - consolidate listeners and add ctx

crates/evm-helpers/Cargo.toml

@@ -14,3 +14,4 @@ futures.workspace = true
 futures-util.workspace = true
 once_cell.workspace = true
 tokio.workspace = true
+tracing.workspace = true

crates/evm-helpers/src/contracts.rs

@@ -164,8 +164,8 @@ pub trait EnclaveWrite {
 }
 
 /// Generic type to represent different provider types
-pub trait ProviderType: Send {
-    type Provider: Provider + Send + Sync + 'static;
+pub trait ProviderType: Clone + Send + Sync + 'static {
+    type Provider: Provider + Clone + Send + Sync + 'static;
 }
 
 /// Marker type for read-only provider
@@ -190,6 +190,15 @@ pub struct EnclaveContract<T: ProviderType> {
     _marker: PhantomData<T>,
 }
 
+impl<R: ProviderType> EnclaveContract<R> {
+    pub fn address(&self) -> &Address {
+        &self.contract_address
+    }
+    pub fn get_provider(&self) -> Arc<R::Provider> {
+        self.provider.clone()
+    }
+}
+
 impl EnclaveContract<ReadWrite> {
     pub async fn new(
         http_rpc_url: &str,
@@ -198,14 +207,6 @@ impl EnclaveContract<ReadWrite> {
     ) -> Result<EnclaveContract<ReadWrite>> {
         EnclaveContractFactory::create_write(http_rpc_url, contract_address, private_key).await
     }
-
-    pub fn get_provider(&self) -> Arc<EnclaveWriteProvider> {
-        self.provider.clone()
-    }
-
-    pub fn address(&self) -> &Address {
-        &self.contract_address
-    }
 }
 
 impl EnclaveContract<ReadOnly> {
@@ -215,14 +216,6 @@ impl EnclaveContract<ReadOnly> {
     ) -> Result<EnclaveContract<ReadOnly>> {
         EnclaveContractFactory::create_read(http_rpc_url, contract_address).await
     }
-
-    pub fn get_provider(&self) -> Arc<EnclaveReadOnlyProvider> {
-        self.provider.clone()
-    }
-
-    pub fn address(&self) -> &Address {
-        &self.contract_address
-    }
 }
 
 /// Type alias for read-only provider

crates/evm-helpers/src/listener.rs

@@ -15,7 +15,8 @@ use eyre::Result;
 use futures::stream::StreamExt;
 use futures_util::future::FutureExt;
 use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc};
-use tokio::{sync::RwLock, task::JoinHandle};
+use tokio::sync::RwLock;
+use tracing::info;
 
 type EventHandler =
     Box<dyn Fn(&Log) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> + Send + Sync>;
@@ -36,7 +37,7 @@ impl EventListener {
         }
     }
 
-    pub async fn add_event_handler<E, F, Fut>(&mut self, handler: F)
+    pub async fn add_event_handler<E, F, Fut>(&self, handler: F)
     where
         E: SolEvent + Send + Clone + 'static,
         F: Fn(E) -> Fut + Send + Sync + 'static,
@@ -63,7 +64,7 @@ impl EventListener {
             .push(wrapped_handler);
     }
 
-    async fn listen(&self) -> Result<()> {
+    pub async fn listen(&self) -> Result<()> {
         let mut stream = self
             .provider
             .subscribe_logs(&self.filter)
@@ -89,14 +90,14 @@ impl EventListener {
         Ok(())
     }
 
-    pub fn start(&self) -> JoinHandle<Result<()>> {
-        let this = self.clone();
-        tokio::spawn(async move { this.listen().await })
-    }
-
-    pub async fn create_contract_listener(ws_url: &str, contract_address: &str) -> Result<Self> {
+    /// Create a contract listener that will listen to events from all addresses.
+    pub async fn create_contract_listener(ws_url: &str, addresses: &[&str]) -> Result<Self> {
         let provider = Arc::new(ProviderBuilder::new().connect(ws_url).await?);
-        let address = contract_address.parse::<Address>()?;
+
+        let address = addresses
+            .iter()
+            .map(|a| a.parse::<Address>().map_err(|e| eyre::eyre!("{e}")))
+            .collect::<Result<Vec<_>>>()?;
         let filter = Filter::new()
             .address(address)
             .from_block(BlockNumberOrTag::Latest);

crates/evm-helpers/tests/integration.rs

@@ -9,7 +9,10 @@ use alloy::sol;
 use e3_evm_helpers::listener::EventListener;
 use eyre::Result;
 use helpers::setup_logs_contract;
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use std::{
+    sync::Arc,
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
 use tokio::time::sleep;
 
 sol!(
@@ -25,11 +28,13 @@ async fn test_event_listener() -> Result<()> {
     let (tx, mut rx) = tokio::sync::mpsc::channel::<String>(10);
     let (tx_addr, mut rx_addr) = tokio::sync::mpsc::channel::<String>(10);
 
-    let mut event_listener = EventListener::create_contract_listener(
-        &anvil.ws_endpoint(),
-        &contract.address().to_string(),
-    )
-    .await?;
+    let event_listener = Arc::new(
+        EventListener::create_contract_listener(
+            &anvil.ws_endpoint(),
+            &[&contract.address().to_string()],
+        )
+        .await?,
+    );
 
     event_listener
         .add_event_handler(move |event: EmitLogs::ValueChanged| {
@@ -51,7 +56,8 @@ async fn test_event_listener() -> Result<()> {
         })
         .await;
 
-    event_listener.start();
+    let spawn_event_listener = event_listener.clone();
+    let _ = tokio::spawn(async move { spawn_event_listener.listen().await });
 
     contract
         .setValue("hello".to_string())
@@ -104,11 +110,13 @@ async fn test_overlapping_listener_handlers() -> Result<()> {
     let (contract, _, _, anvil) = setup_logs_contract().await?;
     let (tx, mut rx) = tokio::sync::mpsc::channel::<String>(10);
 
-    let mut event_listener = EventListener::create_contract_listener(
-        &anvil.ws_endpoint(),
-        &contract.address().to_string(),
-    )
-    .await?;
+    let event_listener = Arc::new(
+        EventListener::create_contract_listener(
+            &anvil.ws_endpoint(),
+            &[&contract.address().to_string()],
+        )
+        .await?,
+    );
 
     let tx1 = tx.clone();
     event_listener
@@ -140,7 +148,8 @@ async fn test_overlapping_listener_handlers() -> Result<()> {
         })
         .await;
 
-    event_listener.start();
+    let spawn_event_listener = event_listener.clone();
+    let _ = tokio::spawn(async move { spawn_event_listener.listen().await });
 
     // Events should be returned roughly in this order:
     // 0ms : one

crates/indexer/Cargo.toml

@@ -15,3 +15,4 @@ eyre.workspace = true
 serde.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
+tracing.workspace = true