feat(torii-core): specify namespaces to exclusively index (#2687)

bin/torii/src/main.rs

@@ -126,7 +126,7 @@ async fn main() -> anyhow::Result<()> {
     let (block_tx, block_rx) = tokio::sync::mpsc::channel(100);
 
     let mut flags = IndexingFlags::empty();
-    if args.indexing.index_transactions {
+    if args.indexing.transactions {
         flags.insert(IndexingFlags::TRANSACTIONS);
     }
     if args.events.raw {
@@ -143,11 +143,12 @@ 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.index_pending,
+            index_pending: args.indexing.pending,
             polling_interval: Duration::from_millis(args.indexing.polling_interval),
             flags,
             event_processor_config: EventProcessorConfig {
-                historical_events: args.events.historical.unwrap_or_default().into_iter().collect(),
+                historical_events: args.events.historical.into_iter().collect(),
+                namespaces: args.indexing.namespaces.into_iter().collect(),
             },
         },
         shutdown_tx.clone(),

crates/torii/cli/src/args.rs

@@ -236,7 +236,7 @@ mod test {
         assert_eq!(torii_args.rpc, Url::parse("http://0.0.0.0:6060").unwrap());
         assert_eq!(torii_args.db_dir, Some(PathBuf::from("/tmp/torii-test2")));
         assert!(!torii_args.events.raw);
-        assert_eq!(torii_args.events.historical, Some(vec!["a-A".to_string()]));
+        assert_eq!(torii_args.events.historical, vec!["a-A".to_string()]);
         assert_eq!(torii_args.server, ServerOptions::default());
     }
 
@@ -261,13 +261,14 @@ mod test {
 
         [indexing]
         events_chunk_size = 9999
-        index_pending = true
+        pending = true
         max_concurrent_tasks = 1000
-        index_transactions = false
+        transactions = false
         contracts = [
             "erc20:0x1234",
             "erc721:0x5678"
         ]
+        namespaces = []
         "#;
         let path = std::env::temp_dir().join("torii-config.json");
         std::fs::write(&path, content).unwrap();
@@ -282,16 +283,13 @@ mod test {
         assert_eq!(torii_args.rpc, Url::parse("http://0.0.0.0:2222").unwrap());
         assert_eq!(torii_args.db_dir, Some(PathBuf::from("/tmp/torii-test")));
         assert!(!torii_args.events.raw);
-        assert_eq!(
-            torii_args.events.historical,
-            Some(vec!["ns-E".to_string(), "ns-EH".to_string()])
-        );
+        assert_eq!(torii_args.events.historical, vec!["ns-E".to_string(), "ns-EH".to_string()]);
         assert_eq!(torii_args.indexing.events_chunk_size, 9999);
         assert_eq!(torii_args.indexing.blocks_chunk_size, 10240);
-        assert!(torii_args.indexing.index_pending);
+        assert!(torii_args.indexing.pending);
         assert_eq!(torii_args.indexing.polling_interval, 500);
         assert_eq!(torii_args.indexing.max_concurrent_tasks, 1000);
-        assert!(!torii_args.indexing.index_transactions);
+        assert!(!torii_args.indexing.transactions);
         assert_eq!(
             torii_args.indexing.contracts,
             vec![

crates/torii/cli/src/options.rs

@@ -100,7 +100,7 @@ pub struct IndexingOptions {
 
     /// Enable indexing pending blocks
     #[arg(long = "indexing.pending", action = ArgAction::Set, default_value_t = true, help = "Whether or not to index pending blocks.")]
-    pub index_pending: bool,
+    pub pending: bool,
 
     /// Polling interval in ms
     #[arg(
@@ -127,7 +127,7 @@ pub struct IndexingOptions {
         default_value_t = false,
         help = "Whether or not to index world transactions and keep them in the database."
     )]
-    pub index_transactions: bool,
+    pub transactions: bool,
 
     /// ERC contract addresses to index
     #[arg(
@@ -138,18 +138,28 @@ pub struct IndexingOptions {
     )]
     #[serde(deserialize_with = "deserialize_contracts")]
     pub contracts: Vec<Contract>,
+
+    /// Namespaces to index
+    #[arg(
+        long = "indexing.namespaces",
+        value_delimiter = ',',
+        help = "The namespaces of the world that torii should index. If empty, all namespaces \
+                will be indexed."
+    )]
+    pub namespaces: Vec<String>,
 }
 
 impl Default for IndexingOptions {
     fn default() -> Self {
         Self {
             events_chunk_size: DEFAULT_EVENTS_CHUNK_SIZE,
             blocks_chunk_size: DEFAULT_BLOCKS_CHUNK_SIZE,
-            index_pending: true,
-            index_transactions: false,
+            pending: true,
+            transactions: false,
             contracts: vec![],
             polling_interval: DEFAULT_POLLING_INTERVAL,
             max_concurrent_tasks: DEFAULT_MAX_CONCURRENT_TASKS,
+            namespaces: vec![],
         }
     }
 }
@@ -168,12 +178,12 @@ pub struct EventsOptions {
         value_delimiter = ',',
         help = "Event messages that are going to be treated as historical during indexing."
     )]
-    pub historical: Option<Vec<String>>,
+    pub historical: Vec<String>,
 }
 
 impl Default for EventsOptions {
     fn default() -> Self {
-        Self { raw: true, historical: None }
+        Self { raw: true, historical: vec![] }
     }
 }
 

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

@@ -31,6 +31,7 @@ const ENTITY_ID_INDEX: usize = 1;
 #[derive(Clone, Debug, Default)]
 pub struct EventProcessorConfig {
     pub historical_events: HashSet<String>,
+    pub namespaces: HashSet<String>,
 }
 
 #[async_trait]

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

@@ -38,7 +38,7 @@ where
         block_timestamp: u64,
         _event_id: &str,
         event: &Event,
-        _config: &EventProcessorConfig,
+        config: &EventProcessorConfig,
     ) -> Result<(), Error> {
         // Torii version is coupled to the world version, so we can expect the event to be well
         // formed.
@@ -58,6 +58,12 @@ where
         let namespace = event.namespace.to_string().unwrap();
         let name = event.name.to_string().unwrap();
 
+        // 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) {
+            return Ok(());
+        }
+
         // Called model here by language, but it's an event. Torii rework will make clear
         // distinction.
         let model = world.model_reader(&namespace, &name).await?;

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

@@ -38,7 +38,7 @@ where
         block_timestamp: u64,
         _event_id: &str,
         event: &Event,
-        _config: &EventProcessorConfig,
+        config: &EventProcessorConfig,
     ) -> Result<(), Error> {
         // Torii version is coupled to the world version, so we can expect the event to be well
         // formed.
@@ -58,6 +58,12 @@ where
         let namespace = event.namespace.to_string().unwrap();
         let name = event.name.to_string().unwrap();
 
+        // 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) {
+            return Ok(());
+        }
+
         let model = world.model_reader(&namespace, &name).await?;
         let schema = model.schema().await?;
         let layout = model.layout().await?;

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

@@ -1,4 +1,4 @@
-use anyhow::{Error, Ok, Result};
+use anyhow::{Error, Result};
 use async_trait::async_trait;
 use dojo_world::contracts::abigen::world::Event as WorldEvent;
 use dojo_world::contracts::world::WorldContractReader;
@@ -51,7 +51,17 @@ where
             }
         };
 
-        let model = db.model(event.selector).await?;
+        // If the model does not exist, silently ignore it.
+        // This can happen if only specific namespaces are indexed.
+        let model = match db.model(event.selector).await {
+            Ok(m) => m,
+            Err(e) => {
+                if e.to_string().contains("no rows") {
+                    return Ok(());
+                }
+                return Err(e);
+            }
+        };
 
         info!(
             target: LOG_TARGET,

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

@@ -1,4 +1,4 @@
-use anyhow::{Error, Ok, Result};
+use anyhow::{Error, Result};
 use async_trait::async_trait;
 use dojo_world::contracts::abigen::world::Event as WorldEvent;
 use dojo_world::contracts::world::WorldContractReader;
@@ -52,7 +52,17 @@ where
             }
         };
 
-        let model = db.model(event.selector).await?;
+        // If the model does not exist, silently ignore it.
+        // This can happen if only specific namespaces are indexed.
+        let model = match db.model(event.selector).await {
+            Ok(m) => m,
+            Err(e) => {
+                if e.to_string().contains("no rows") {
+                    return Ok(());
+                }
+                return Err(e);
+            }
+        };
 
         info!(
             target: LOG_TARGET,

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

@@ -56,7 +56,18 @@ where
         let entity_id = event.data[ENTITY_ID_INDEX];
         let member_selector = event.data[MEMBER_INDEX];
 
-        let model = db.model(model_id).await?;
+        // If the model does not exist, silently ignore it.
+        // This can happen if only specific namespaces are indexed.
+        let model = match db.model(model_id).await {
+            Ok(m) => m,
+            Err(e) => {
+                if e.to_string().contains("no rows") {
+                    return Ok(());
+                }
+                return Err(e);
+            }
+        };
+
         let schema = model.schema;
 
         let mut member = schema

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

@@ -1,4 +1,4 @@
-use anyhow::{Error, Ok, Result};
+use anyhow::{Error, Result};
 use async_trait::async_trait;
 use dojo_types::schema::Ty;
 use dojo_world::contracts::abigen::world::Event as WorldEvent;
@@ -55,7 +55,17 @@ where
         let model_selector = event.selector;
         let entity_id = event.entity_id;
 
-        let model = db.model(model_selector).await?;
+        // If the model does not exist, silently ignore it.
+        // This can happen if only specific namespaces are indexed.
+        let model = match db.model(event.selector).await {
+            Ok(m) => m,
+            Err(e) => {
+                if e.to_string().contains("no rows") {
+                    return Ok(());
+                }
+                return Err(e);
+            }
+        };
 
         info!(
             target: LOG_TARGET,