refactor(torii-core): cleanups

bin/torii/src/main.rs

@@ -137,6 +137,9 @@ async fn main() -> anyhow::Result<()> {
     if args.events.raw {
         flags.insert(IndexingFlags::RAW_EVENTS);
     }
+    if args.indexing.pending {
+        flags.insert(IndexingFlags::PENDING_BLOCKS);
+    }
 
     let mut engine: Engine<Arc<JsonRpcClient<HttpTransport>>> = Engine::new(
         world,
@@ -148,7 +151,6 @@ async fn main() -> anyhow::Result<()> {
             start_block: 0,
             blocks_chunk_size: args.indexing.blocks_chunk_size,
             events_chunk_size: args.indexing.events_chunk_size,
-            index_pending: args.indexing.pending,
             polling_interval: Duration::from_millis(args.indexing.polling_interval),
             flags,
             event_processor_config: EventProcessorConfig {

crates/torii/core/src/engine.rs

@@ -136,6 +136,7 @@ bitflags! {
     pub struct IndexingFlags: u32 {
         const TRANSACTIONS = 0b00000001;
         const RAW_EVENTS = 0b00000010;
+        const PENDING_BLOCKS = 0b00000100;
     }
 }
 
@@ -145,7 +146,6 @@ pub struct EngineConfig {
     pub start_block: u64,
     pub blocks_chunk_size: u64,
     pub events_chunk_size: u64,
-    pub index_pending: bool,
     pub max_concurrent_tasks: usize,
     pub flags: IndexingFlags,
     pub event_processor_config: EventProcessorConfig,
@@ -158,7 +158,6 @@ impl Default for EngineConfig {
             start_block: 0,
             blocks_chunk_size: 10240,
             events_chunk_size: 1024,
-            index_pending: true,
             max_concurrent_tasks: 100,
             flags: IndexingFlags::empty(),
             event_processor_config: EventProcessorConfig::default(),
@@ -324,7 +323,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
             let data = self.fetch_range(from, to, &cursors.cursor_map).await?;
             debug!(target: LOG_TARGET, duration = ?instant.elapsed(), from = %from, to = %to, "Fetched data for range.");
             FetchDataResult::Range(data)
-        } else if self.config.index_pending {
+        } else if self.config.flags.contains(IndexingFlags::PENDING_BLOCKS) {
             let data =
                 self.fetch_pending(latest_block.clone(), cursors.last_pending_block_tx).await?;
             debug!(target: LOG_TARGET, duration = ?instant.elapsed(), latest_block_number = %latest_block.block_number, "Fetched pending data.");

crates/torii/core/src/processors/event_message.rs

@@ -73,8 +73,7 @@ where
         entity.deserialize(&mut keys_and_unpacked)?;
 
         // TODO: this must come from some torii's configuration.
-        let historical =
-            config.historical_events.contains(&format!("{}-{}", model.namespace, model.name));
+        let historical = config.is_historical(&format!("{}-{}", model.namespace, model.name));
         db.set_event_message(entity, event_id, block_timestamp, historical).await?;
         Ok(())
     }

crates/torii/core/src/processors/mod.rs

@@ -34,6 +34,16 @@ pub struct EventProcessorConfig {
     pub namespaces: HashSet<String>,
 }
 
+impl EventProcessorConfig {
+    pub fn is_historical(&self, tag: &str) -> bool {
+        self.historical_events.contains(tag)
+    }
+
+    pub fn should_index(&self, namespace: &str) -> bool {
+        self.namespaces.is_empty() || self.namespaces.contains(namespace)
+    }
+}
+
 #[async_trait]
 pub trait EventProcessor<P>: Send + Sync
 where

crates/torii/core/src/processors/register_event.rs

@@ -60,7 +60,7 @@ where
 
         // If the namespace is not in the list of namespaces to index, silently ignore it.
         // If our config is empty, we index all namespaces.
-        if !config.namespaces.is_empty() && !config.namespaces.contains(&namespace) {
+        if !config.should_index(&namespace) {
             return Ok(());
         }
 

crates/torii/core/src/processors/register_model.rs

@@ -60,7 +60,7 @@ where
 
         // If the namespace is not in the list of namespaces to index, silently ignore it.
         // If our config is empty, we index all namespaces.
-        if !config.namespaces.is_empty() && !config.namespaces.contains(&namespace) {
+        if !config.should_index(&namespace) {
             return Ok(());
         }
 

crates/torii/core/src/sql/mod.rs

@@ -31,7 +31,6 @@ pub const FELT_DELIMITER: &str = "/";
 
 pub mod cache;
 pub mod erc;
-pub mod query_queue;
 #[cfg(test)]
 #[path = "test.rs"]
 mod test;
@@ -830,7 +829,11 @@ impl Sql {
             Ty::Enum(e) => {
                 if e.options.iter().all(
                     |o| {
-                        if let Ty::Tuple(t) = &o.ty { t.is_empty() } else { false }
+                        if let Ty::Tuple(t) = &o.ty {
+                            t.is_empty()
+                        } else {
+                            false
+                        }
                     },
                 ) {
                     return Ok(());