feat(torii): index whitelisted ERC20/ERC721 tokens

bin/torii/src/main.rs

@@ -11,6 +11,7 @@
 //!   for more info.
 
 use std::net::SocketAddr;
+use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::Arc;
 
@@ -27,6 +28,9 @@
 use tokio::sync::broadcast::Sender;
 use tokio_stream::StreamExt;
 use torii_core::engine::{Engine, EngineConfig, Processors};
+use torii_core::processors::erc20_legacy_transfer::Erc20LegacyTransferProcessor;
+use torii_core::processors::erc20_transfer::Erc20TransferProcessor;
+use torii_core::processors::erc721_transfer::Erc721TransferProcessor;
 use torii_core::processors::event_message::EventMessageProcessor;
 use torii_core::processors::metadata_update::MetadataUpdateProcessor;
 use torii_core::processors::register_model::RegisterModelProcessor;
@@ -37,7 +41,7 @@
 use torii_core::processors::store_update_record::StoreUpdateRecordProcessor;
 use torii_core::simple_broker::SimpleBroker;
 use torii_core::sql::Sql;
-use torii_core::types::Model;
+use torii_core::types::{ErcContract, ErcType, Model, ToriiConfig};
 use torii_server::proxy::Proxy;
 use tracing::{error, info};
 use tracing_subscriber::{fmt, EnvFilter};
@@ -115,11 +119,39 @@
     /// Enable indexing pending blocks
     #[arg(long)]
     index_pending: bool,
+
+    /// ERC contract addresses to index
+    #[arg(long, value_parser = parse_erc_contracts)]
+    #[arg(conflicts_with = "config")]
+    erc_contracts: Option<std::vec::Vec<ErcContract>>,
+
+    /// Configuration file
+    #[arg(long)]
+    config: Option<PathBuf>,
 }
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     let args = Args::parse();
+
+    let mut start_block = args.start_block;
+
+    let mut config = if let Some(path) = args.config {
+        ToriiConfig::load_from_path(&path)?
+    } else {
+        ToriiConfig::default()
+    };
+
+    if let Some(erc_contracts) = args.erc_contracts {
+        config.erc_contracts = erc_contracts;
+    }
+
+    for address in &config.erc_contracts {
+        if address.start_block < start_block {
+            start_block = address.start_block;
+        }
+    }
+
     let filter_layer = EnvFilter::try_from_default_env()
         .unwrap_or_else(|_| EnvFilter::new("info,hyper_reverse_proxy=off"));
 
@@ -162,7 +194,12 @@
     // Get world address
     let world = WorldContractReader::new(args.world_address, &provider);
 
-    let db = Sql::new(pool.clone(), args.world_address).await?;
+    let erc_contracts = config
+        .erc_contracts
+        .iter()
+        .map(|contract| (contract.contract_address, contract.clone()))
+        .collect();
+    let db = Sql::new(pool.clone(), args.world_address, &erc_contracts).await?;
     let processors = Processors {
         event: vec![
             Box::new(RegisterModelProcessor),
@@ -172,6 +209,9 @@
             Box::new(EventMessageProcessor),
             Box::new(StoreUpdateRecordProcessor),
             Box::new(StoreUpdateMemberProcessor),
+            Box::new(Erc20TransferProcessor),
+            Box::new(Erc20LegacyTransferProcessor),
+            Box::new(Erc721TransferProcessor),
         ],
         transaction: vec![Box::new(StoreTransactionProcessor)],
         ..Processors::default()
@@ -185,13 +225,14 @@
         &provider,
         processors,
         EngineConfig {
-            start_block: args.start_block,
+            start_block,
             events_chunk_size: args.events_chunk_size,
             index_pending: args.index_pending,
             ..Default::default()
         },
         shutdown_tx.clone(),
         Some(block_tx),
+        erc_contracts,
     );
 
     let shutdown_rx = shutdown_tx.subscribe();
@@ -286,3 +327,29 @@
         }
     }
 }
+
+// Parses clap cli argument which is expected to be in the format:
+// - erc_type:address:start_block
+// - address:start_block (erc_type defaults to ERC20)
+fn parse_erc_contracts(s: &str) -> anyhow::Result<Vec<ErcContract>> {
+    let parts: Vec<&str> = s.split(',').collect();
+    let mut contracts = Vec::new();
+    for part in parts {
+        match part.split(':').collect::<Vec<&str>>().as_slice() {
+            [r#type, address, start_block] => {
+                let contract_address = Felt::from_str(address).unwrap();
+                let start_block = start_block.parse::<u64>()?;
+                let r#type = r#type.parse::<ErcType>()?;
+                contracts.push(ErcContract { contract_address, start_block, r#type });
+            }
+            [address, start_block] => {
+                let contract_address = Felt::from_str(address)?;
+                let start_block = start_block.parse::<u64>()?;
+                let r#type = ErcType::default();
+                contracts.push(ErcContract { contract_address, start_block, r#type });
+            }
+            _ => return Err(anyhow::anyhow!("Invalid ERC contract format")),
+        }
+    }
+    Ok(contracts)
+}

bin/torii/torii.toml

@@ -0,0 +1,8 @@
+# Example configuration file for Torii
+# erc_contracts = [
+#     { contract_address = "0x1234567890abcdef1234567890abcdef12345678", start_block = 0, type = "ERC20" },
+#     { contract_address = "0xabcdef1234567890abcdef1234567890abcdef12", start_block = 1, type = "ERC721" },
+# ]
+erc_contracts = [
+    { type = "ERC20", contract_address = "0x07fc13cc1f43f0b0519f84df8bf13bea4d9fd5ce2d748c3baf27bf90a565f60a", start_block = 0 },
+]

crates/katana/scripts/deploy-erc20.sh

@@ -0,0 +1 @@
+starkli deploy --account account.json --keystore signer.json --keystore-password "" 0x02a8846878b6ad1f54f6ba46f5f40e11cee755c677f130b2c4b60566c9003f1f 0x626c6f62 0x424c42 0x8 u256:10000000000 0xb3ff441a68610b30fd5e2abbf3a1548eb6ba6f3559f2862bf2dc757e5828c

crates/torii/core/Cargo.toml

@@ -37,6 +37,7 @@ thiserror.workspace = true
 tokio = { version = "1.32.0", features = [ "sync" ], default-features = true }
 tokio-stream = "0.1.11"
 tokio-util = "0.7.7"
+toml.workspace = true
 tracing.workspace = true
 
 [dev-dependencies]

crates/torii/core/src/engine.rs

@@ -1,11 +1,13 @@
-use std::collections::BTreeMap;
+use std::collections::btree_map::Entry;
+use std::collections::{BTreeMap, HashMap};
 use std::fmt::Debug;
 use std::time::Duration;
 
 use anyhow::Result;
 use dojo_world::contracts::world::WorldContractReader;
+use futures_util::future::join_all;
 use starknet::core::types::{
-    BlockId, BlockTag, Event, EventFilter, Felt, MaybePendingBlockWithTxHashes,
+    BlockId, BlockTag, Event, EventFilter, EventsPage, Felt, MaybePendingBlockWithTxHashes,
     MaybePendingBlockWithTxs, ReceiptBlock, Transaction, TransactionReceipt,
     TransactionReceiptWithBlockInfo,
 };
@@ -18,6 +20,7 @@
 
 use crate::processors::{BlockProcessor, EventProcessor, TransactionProcessor};
 use crate::sql::Sql;
+use crate::types::ErcContract;
 
 #[allow(missing_debug_implementations)]
 pub struct Processors<P: Provider + Sync> {
@@ -62,6 +65,8 @@
     config: EngineConfig,
     shutdown_tx: Sender<()>,
     block_tx: Option<BoundedSender<u64>>,
+    // ERC tokens to index
+    tokens: HashMap<Felt, ErcContract>,
 }
 
 struct UnprocessedEvent {
@@ -70,6 +75,7 @@
 }
 
 impl<P: Provider + Sync> Engine<P> {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         world: WorldContractReader<P>,
         db: Sql,
@@ -78,8 +84,18 @@
         config: EngineConfig,
         shutdown_tx: Sender<()>,
         block_tx: Option<BoundedSender<u64>>,
+        tokens: HashMap<Felt, ErcContract>,
     ) -> Self {
-        Self { world, db, provider: Box::new(provider), processors, config, shutdown_tx, block_tx }
+        Self {
+            world,
+            db,
+            provider: Box::new(provider),
+            processors,
+            config,
+            shutdown_tx,
+            block_tx,
+            tokens,
+        }
     }
 
     pub async fn start(&mut self) -> Result<()> {
@@ -221,29 +237,45 @@
         to: u64,
         pending_block_tx: Option<Felt>,
     ) -> Result<Option<Felt>> {
-        // Process all blocks from current to latest.
-        let get_events = |token: Option<String>| {
-            self.provider.get_events(
-                EventFilter {
-                    from_block: Some(BlockId::Number(from)),
-                    to_block: Some(BlockId::Number(to)),
-                    address: Some(self.world.address),
-                    keys: None,
-                },
-                token,
-                self.config.events_chunk_size,
-            )
-        };
+        // Fetch all events and collect them using `get_events` method on provider
 
         // handle next events pages
-        let mut events_pages = vec![get_events(None).await?];
+        let mut fetch_all_events_tasks = vec![];
+
+        let events_filter = EventFilter {
+            from_block: Some(BlockId::Number(from)),
+            to_block: Some(BlockId::Number(to)),
+            address: Some(self.world.address),
+            keys: None,
+        };
+
+        let world_events_pages =
+            get_all_events(&self.provider, events_filter, self.config.events_chunk_size);
+        fetch_all_events_tasks.push(world_events_pages);
+
+        for token in &self.tokens {
+            let events_filter = EventFilter {
+                from_block: Some(BlockId::Number(from)),
+                to_block: Some(BlockId::Number(to)),
+                address: Some(*token.0),
+                keys: None,
+            };
+
+            let erc20_events_pages =
+                get_all_events(&self.provider, events_filter, self.config.events_chunk_size);
+
+            fetch_all_events_tasks.push(erc20_events_pages);
+        }
 
-        while let Some(token) = &events_pages.last().unwrap().continuation_token {
-            events_pages.push(get_events(Some(token.clone())).await?);
+        // wait for all events tasks to complete
+        let task_result = join_all(fetch_all_events_tasks).await;
+
+        let mut events_pages = vec![];
+        for result in task_result {
+            events_pages.extend(result?);
         }
 
         // Transactions & blocks to process
-        let mut last_block = 0_u64;
         let mut blocks = BTreeMap::new();
 
         // Flatten events pages and events according to the pending block cursor
@@ -277,12 +309,10 @@
                     }
                 };
 
-                // Keep track of last block number and fetch block timestamp
-                if block_number > last_block {
+                // Fetch block timestamp if not already fetched and inserts into the map
+                if let Entry::Vacant(e) = blocks.entry(block_number) {
                     let block_timestamp = self.get_block_timestamp(block_number).await?;
-                    blocks.insert(block_number, block_timestamp);
-
-                    last_block = block_number;
+                    e.insert(block_timestamp);
                 }
 
                 // Then we skip all transactions until we reach the last pending processed
@@ -378,7 +408,9 @@
         let mut world_event = false;
         if let Some(events) = events {
             for (event_idx, event) in events.iter().enumerate() {
-                if event.from_address != self.world.address {
+                if event.from_address != self.world.address
+                    && !self.tokens.contains_key(&event.from_address)
+                {
                     continue;
                 }
 
@@ -501,3 +533,30 @@
         Ok(())
     }
 }
+
+async fn get_all_events<P>(
+    provider: &P,
+    events_filter: EventFilter,
+    events_chunk_size: u64,
+) -> Result<Vec<EventsPage>>
+where
+    P: Provider + Sync,
+{
+    let mut events_pages = Vec::new();
+    let mut continuation_token = None;
+
+    loop {
+        let events_page = provider
+            .get_events(events_filter.clone(), continuation_token.clone(), events_chunk_size)
+            .await?;
+
+        continuation_token = events_page.continuation_token.clone();
+        events_pages.push(events_page);
+
+        if continuation_token.is_none() {
+            break;
+        }
+    }
+
+    Ok(events_pages)
+}