Merge pull request #70 from bilinearlabs/async_trait

Refactor the Storage trait and implementors to use async_trait #66

Author: felipet

Cargo.lock

@@ -15,6 +15,7 @@ test-utils = ["dep:fake"]
 [dependencies]
 alloy = { version = "1.4.0", features = ["rpc-types", "json", "rpc-client", "json-rpc"] }
 anyhow = "1.0.100"
+async-trait = "0.1"
 axum = { version = "0.8.7", features = ["default", "http2"] }
 hex = "0.4"
 clap = { version = "4.5.51", features = ["derive"] }

Cargo.toml

@@ -7,41 +7,30 @@ use crate::{
 };
 use alloy::{json_abi::Event, primitives::B256, rpc::types::Log};
 use anyhow::Result;
+use async_trait::async_trait;
 use serde_json::Value;
-use std::{any::Any, future::Future, pin::Pin};
+use std::any::Any;
 
 /// Trait that defines the API between the producer task and the storage.
+#[async_trait]
 pub trait Storage: Send + Sync + 'static + Any {
     /// Adds a list of events to the storage for a specific chain.
-    fn add_events(
-        &self,
-        chain_id: u64,
-        events: &[Log],
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>>;
+    async fn add_events(&self, chain_id: u64, events: &[Log]) -> Result<()>;
 
     /// Lists the events that are registered in the storage along their indexing status.
-    fn list_indexed_events(
-        &self,
-    ) -> Pin<Box<dyn Future<Output = Result<Vec<EventDescriptorDb>>> + Send + '_>>;
+    async fn list_indexed_events(&self) -> Result<Vec<EventDescriptorDb>>;
 
     /// Get the status of an event in the database for a specific chain.
-    fn event_index_status(
-        &self,
-        chain_id: u64,
-        event: &Event,
-    ) -> Pin<Box<dyn Future<Output = Result<Option<EventStatus>>> + Send + '_>>;
+    async fn event_index_status(&self, chain_id: u64, event: &Event)
+    -> Result<Option<EventStatus>>;
 
     /// Includes a list of events in the storage for a specific chain.
     ///
     /// # Description
     ///
     /// This method shall be used once at the beginning of the application to include the events that
     /// are to going be indexed.
-    fn include_events(
-        &self,
-        chain_id: u64,
-        events: &[Event],
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>>;
+    async fn include_events(&self, chain_id: u64, events: &[Event]) -> Result<()>;
 
     /// Gets the full signature of an event type by its hash.
     ///
@@ -52,32 +41,16 @@ pub trait Storage: Send + Sync + 'static + Any {
     /// "Transfer(address indexed from,address indexed to,uint256 value)".
     ///
     /// This signature is used to build a an event object.
-    fn get_event_signature(
-        &self,
-        event_hash: &str,
-    ) -> Pin<Box<dyn Future<Output = Result<String>> + Send + '_>>;
+    async fn get_event_signature(&self, event_hash: &str) -> Result<String>;
 
     /// Gets the latest block number that has been indexed for a specific chain.
-    fn last_block(
-        &self,
-        chain_id: u64,
-        event: &Event,
-    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + '_>>;
+    async fn last_block(&self, chain_id: u64, event: &Event) -> Result<u64>;
 
     /// Gets the first block number that has been indexed for a specific chain.
-    fn first_block(
-        &self,
-        chain_id: u64,
-        event: &Event,
-    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + '_>>;
+    async fn first_block(&self, chain_id: u64, event: &Event) -> Result<u64>;
 
     /// Sets the first block number for an event on a specific chain.
-    fn set_first_block(
-        &self,
-        chain_id: u64,
-        event: &Event,
-        block_number: u64,
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>>;
+    async fn set_first_block(&self, chain_id: u64, event: &Event, block_number: u64) -> Result<()>;
 
     /// Sets the last block number for the specified events on a specific chain.
     ///
@@ -93,26 +66,21 @@ pub trait Storage: Send + Sync + 'static + Any {
     /// * `event_selectors` - The list of event selectors (topic0 hashes) to update.
     /// * `last_processed` - The block number to set as `last_block`. If `None`, uses the
     ///   maximum `last_block` among the specified events.
-    fn synchronize_events(
+    async fn synchronize_events(
         &self,
         chain_id: u64,
         event_selectors: &[B256],
         last_processed: Option<u64>,
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>>;
+    ) -> Result<()>;
 
     /// Sends a raw SQL query to the storage and returns a JSON value.
-    fn send_raw_query(
-        &self,
-        query: &str,
-    ) -> Pin<Box<dyn Future<Output = Result<Value>> + Send + '_>>;
+    async fn send_raw_query(&self, query: &str) -> Result<Value>;
 
     /// Lists the contracts indexed in the storage.
-    fn list_contracts(
-        &self,
-    ) -> Pin<Box<dyn Future<Output = Result<Vec<ContractDescriptorDb>>> + Send + '_>>;
+    async fn list_contracts(&self) -> Result<Vec<ContractDescriptorDb>>;
 
     /// Returns the database schema including all tables and their column definitions.
-    fn describe_database(&self) -> Pin<Box<dyn Future<Output = Result<Value>> + Send + '_>>;
+    async fn describe_database(&self) -> Result<Value>;
 }
 
 /// Trait for creating storage instances.

src/storage/storage_api.rs

@@ -16,12 +16,11 @@ use alloy::{
     rpc::types::Log,
 };
 use anyhow::{Context, Result};
+use async_trait::async_trait;
 use duckdb::{Connection, OptionalExt, types::ValueRef};
 use serde_json::{Map, Number, Value, json};
 use std::{
     collections::{HashMap, HashSet},
-    future::Future,
-    pin::Pin,
     string::ToString,
     sync::{Mutex, RwLock},
 };
@@ -61,107 +60,63 @@ impl Clone for DuckDBStorage {
     }
 }
 
+#[async_trait]
 impl Storage for DuckDBStorage {
-    fn add_events(
-        &self,
-        chain_id: u64,
-        events: &[Log],
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>> {
-        Box::pin(std::future::ready(self.add_events_sync(chain_id, events)))
+    async fn add_events(&self, chain_id: u64, events: &[Log]) -> Result<()> {
+        self.add_events_sync(chain_id, events)
     }
 
-    fn list_indexed_events(
-        &self,
-    ) -> Pin<Box<dyn Future<Output = Result<Vec<EventDescriptorDb>>> + Send + '_>> {
-        Box::pin(std::future::ready(self.list_indexed_events_sync()))
+    async fn list_indexed_events(&self) -> Result<Vec<EventDescriptorDb>> {
+        self.list_indexed_events_sync()
     }
 
-    fn last_block(
-        &self,
-        chain_id: u64,
-        event: &Event,
-    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + '_>> {
-        Box::pin(std::future::ready(self.last_block_sync(chain_id, event)))
+    async fn last_block(&self, chain_id: u64, event: &Event) -> Result<u64> {
+        self.last_block_sync(chain_id, event)
     }
 
-    fn first_block(
-        &self,
-        chain_id: u64,
-        event: &Event,
-    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + '_>> {
-        Box::pin(std::future::ready(self.first_block_sync(chain_id, event)))
+    async fn first_block(&self, chain_id: u64, event: &Event) -> Result<u64> {
+        self.first_block_sync(chain_id, event)
     }
 
-    fn include_events(
-        &self,
-        chain_id: u64,
-        events: &[Event],
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>> {
-        Box::pin(std::future::ready(
-            self.include_events_sync(chain_id, events),
-        ))
+    async fn include_events(&self, chain_id: u64, events: &[Event]) -> Result<()> {
+        self.include_events_sync(chain_id, events)
     }
 
-    fn get_event_signature(
-        &self,
-        event_hash: &str,
-    ) -> Pin<Box<dyn Future<Output = Result<String>> + Send + '_>> {
-        Box::pin(std::future::ready(
-            self.get_event_signature_sync(event_hash),
-        ))
+    async fn get_event_signature(&self, event_hash: &str) -> Result<String> {
+        self.get_event_signature_sync(event_hash)
     }
 
-    fn event_index_status(
+    async fn event_index_status(
         &self,
         chain_id: u64,
         event: &Event,
-    ) -> Pin<Box<dyn Future<Output = Result<Option<EventStatus>>> + Send + '_>> {
-        Box::pin(std::future::ready(
-            self.event_index_status_sync(chain_id, event),
-        ))
+    ) -> Result<Option<EventStatus>> {
+        self.event_index_status_sync(chain_id, event)
     }
 
-    fn synchronize_events(
+    async fn synchronize_events(
         &self,
         chain_id: u64,
         event_selectors: &[B256],
         last_processed: Option<u64>,
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>> {
-        Box::pin(std::future::ready(self.synchronize_events_sync(
-            chain_id,
-            event_selectors,
-            last_processed,
-        )))
+    ) -> Result<()> {
+        self.synchronize_events_sync(chain_id, event_selectors, last_processed)
     }
 
-    fn send_raw_query(
-        &self,
-        query: &str,
-    ) -> Pin<Box<dyn Future<Output = Result<Value>> + Send + '_>> {
-        Box::pin(std::future::ready(self.send_raw_query_sync(query)))
+    async fn send_raw_query(&self, query: &str) -> Result<Value> {
+        self.send_raw_query_sync(query)
     }
 
-    fn list_contracts(
-        &self,
-    ) -> Pin<Box<dyn Future<Output = Result<Vec<ContractDescriptorDb>>> + Send + '_>> {
-        Box::pin(std::future::ready(self.list_contracts_sync()))
+    async fn list_contracts(&self) -> Result<Vec<ContractDescriptorDb>> {
+        self.list_contracts_sync()
     }
 
-    fn set_first_block(
-        &self,
-        chain_id: u64,
-        event: &Event,
-        block_number: u64,
-    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>> {
-        Box::pin(std::future::ready(self.set_first_block_sync(
-            chain_id,
-            event,
-            block_number,
-        )))
+    async fn set_first_block(&self, chain_id: u64, event: &Event, block_number: u64) -> Result<()> {
+        self.set_first_block_sync(chain_id, event, block_number)
     }
 
-    fn describe_database(&self) -> Pin<Box<dyn Future<Output = Result<Value>> + Send + '_>> {
-        Box::pin(std::future::ready(self.describe_database_sync()))
+    async fn describe_database(&self) -> Result<Value> {
+        self.describe_database_sync()
     }
 }