feat(torii-indexer): eip 4906 update metadata processor

crates/torii/indexer/src/processors/erc4906_metadata_update.rs

@@ -0,0 +1,116 @@
+use std::hash::{DefaultHasher, Hash, Hasher};
+
+use anyhow::Error;
+use async_trait::async_trait;
+use cainome::cairo_serde::{CairoSerde, U256 as U256Cainome};
+use dojo_world::contracts::world::WorldContractReader;
+use starknet::core::types::{Event, U256};
+use starknet::providers::Provider;
+use torii_sqlite::Sql;
+use tracing::debug;
+
+use super::{EventProcessor, EventProcessorConfig};
+use crate::task_manager::{TaskId, TaskPriority};
+
+pub(crate) const LOG_TARGET: &str = "torii_indexer::processors::erc4906_metadata_update";
+
+#[derive(Default, Debug)]
+pub struct Erc4906MetadataUpdateProcessor;
+
+#[async_trait]
+impl<P> EventProcessor<P> for Erc4906MetadataUpdateProcessor
+where
+    P: Provider + Send + Sync + std::fmt::Debug,
+{
+    fn event_key(&self) -> String {
+        // We'll handle both event types in validate()
+        "MetadataUpdate".to_string()
+    }
+
+    fn validate(&self, event: &Event) -> bool {
+        // Single token metadata update: [hash(MetadataUpdate), token_id.low, token_id.high]
+        if event.keys.len() == 3 && event.data.is_empty() {
+            return true;
+        }
+
+        // Batch metadata update: [hash(BatchMetadataUpdate), from_token_id.low, from_token_id.high,
+        // to_token_id.low, to_token_id.high]
+        if event.keys.len() == 5 && event.data.is_empty() {
+            return true;
+        }
+
+        false
+    }
+
+    fn task_priority(&self) -> TaskPriority {
+        2 // Lower priority than transfers
+    }
+
+    fn task_identifier(&self, event: &Event) -> TaskId {
+        let mut hasher = DefaultHasher::new();
+        event.from_address.hash(&mut hasher); // Hash the contract address
+
+        // For single token updates
+        if event.keys.len() == 3 {
+            event.keys[1].hash(&mut hasher); // token_id.low
+            event.keys[2].hash(&mut hasher); // token_id.high
+        } else {
+            // For batch updates, we need to be more conservative
+            // Hash just the contract address to serialize all batch updates for the same contract
+            // This prevents race conditions with overlapping ranges
+        }
+
+        hasher.finish()
+    }
+
+    async fn process(
+        &self,
+        _world: &WorldContractReader<P>,
+        db: &mut Sql,
+        _block_number: u64,
+        _block_timestamp: u64,
+        _event_id: &str,
+        event: &Event,
+        _config: &EventProcessorConfig,
+    ) -> Result<(), Error> {
+        let token_address = event.from_address;
+
+        if event.keys.len() == 3 {
+            // Single token metadata update
+            let token_id = U256Cainome::cairo_deserialize(&event.keys, 1)?;
+            let token_id = U256::from_words(token_id.low, token_id.high);
+
+            db.update_erc721_metadata(token_address, token_id).await?;
+
+            debug!(
+                target: LOG_TARGET,
+                token_address = ?token_address,
+                token_id = ?token_id,
+                "ERC721 metadata updated for single token"
+            );
+        } else {
+            // Batch metadata update
+            let from_token_id = U256Cainome::cairo_deserialize(&event.keys, 1)?;
+            let from_token_id = U256::from_words(from_token_id.low, from_token_id.high);
+
+            let to_token_id = U256Cainome::cairo_deserialize(&event.keys, 3)?;
+            let to_token_id = U256::from_words(to_token_id.low, to_token_id.high);
+
+            let mut token_id = from_token_id;
+            while token_id <= to_token_id {
+                db.update_erc721_metadata(token_address, token_id).await?;
+                token_id += U256::from(1u8);
+            }
+
+            debug!(
+                target: LOG_TARGET,
+                token_address = ?token_address,
+                from_token_id = ?from_token_id,
+                to_token_id = ?to_token_id,
+                "ERC721 metadata updated for token range"
+            );
+        }
+
+        Ok(())
+    }
+}

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

@@ -12,6 +12,7 @@ use crate::task_manager::{TaskId, TaskPriority};
 pub mod controller;
 pub mod erc20_legacy_transfer;
 pub mod erc20_transfer;
+pub mod erc4906_metadata_update;
 pub mod erc721_legacy_transfer;
 pub mod erc721_transfer;
 pub mod event_message;
@@ -26,6 +27,7 @@ pub mod store_update_member;
 pub mod store_update_record;
 pub mod upgrade_event;
 pub mod upgrade_model;
+
 #[derive(Clone, Debug, Default)]
 pub struct EventProcessorConfig {
     pub historical_events: HashSet<String>,

crates/torii/sqlite/src/erc.rs

@@ -10,6 +10,7 @@
 use super::utils::{u256_to_sql_string, I256};
 use super::{Sql, SQL_FELT_DELIMITER};
 use crate::constants::TOKEN_TRANSFER_TABLE;
+use crate::executor::erc::UpdateErc721MetadataQuery;
 use crate::executor::{
     ApplyBalanceDiffQuery, Argument, QueryMessage, QueryType, RegisterErc20TokenQuery,
     RegisterErc721TokenQuery,
@@ -142,6 +143,23 @@
         Ok(())
     }
 
+    pub async fn update_erc721_metadata(
+        &mut self,
+        contract_address: Felt,
+        token_id: U256,
+    ) -> Result<()> {
+        self.executor.send(QueryMessage::new(
+            "".to_string(),
+            vec![],
+            QueryType::UpdateErc721Metadata(UpdateErc721MetadataQuery {
+                contract_address,
+                token_id,
+            }),
+        ))?;
+
+        Ok(())
+    }
+
     async fn register_erc20_token_metadata<P: Provider + Sync>(
         &mut self,
         contract_address: Felt,

crates/torii/sqlite/src/executor/erc.rs

@@ -45,6 +45,12 @@
     pub decimals: u8,
 }
 
+#[derive(Debug, Clone)]
+pub struct UpdateErc721MetadataQuery {
+    pub contract_address: Felt,
+    pub token_id: U256,
+}
+
 impl<'c, P: Provider + Sync + Send + 'static> Executor<'c, P> {
     pub async fn apply_balance_diff(
         &mut self,
@@ -180,21 +186,17 @@
         Ok(())
     }
 
-    pub async fn process_register_erc721_token_query(
-        register_erc721_token: RegisterErc721TokenQuery,
-        provider: Arc<P>,
-        name: String,
-        symbol: String,
-    ) -> Result<RegisterErc721TokenMetadata> {
+    async fn fetch_token_uri(
+        provider: &P,
+        contract_address: Felt,
+        token_id: U256,
+    ) -> Result<String> {
         let token_uri = if let Ok(token_uri) = provider
             .call(
                 FunctionCall {
-                    contract_address: register_erc721_token.contract_address,
+                    contract_address,
                     entry_point_selector: get_selector_from_name("token_uri").unwrap(),
-                    calldata: vec![
-                        register_erc721_token.actual_token_id.low().into(),
-                        register_erc721_token.actual_token_id.high().into(),
-                    ],
+                    calldata: vec![token_id.low().into(), token_id.high().into()],
                 },
                 BlockId::Tag(BlockTag::Pending),
             )
@@ -204,12 +206,9 @@
         } else if let Ok(token_uri) = provider
             .call(
                 FunctionCall {
-                    contract_address: register_erc721_token.contract_address,
+                    contract_address,
                     entry_point_selector: get_selector_from_name("tokenURI").unwrap(),
-                    calldata: vec![
-                        register_erc721_token.actual_token_id.low().into(),
-                        register_erc721_token.actual_token_id.high().into(),
-                    ],
+                    calldata: vec![token_id.low().into(), token_id.high().into()],
                 },
                 BlockId::Tag(BlockTag::Pending),
             )
@@ -218,12 +217,10 @@
             token_uri
         } else {
             warn!(
-                contract_address = format!("{:#x}", register_erc721_token.contract_address),
-                token_id = %register_erc721_token.actual_token_id,
+                contract_address = format!("{:#x}", contract_address),
+                token_id = %token_id,
                 "Error fetching token URI, empty metadata will be used instead.",
             );
-
-            // Ignoring the token URI if the contract can't return it.
             ByteArray::cairo_serialize(&"".try_into().unwrap())
         };
 
@@ -234,33 +231,82 @@
                 .iter()
                 .map(parse_cairo_short_string)
                 .collect::<Result<Vec<String>, _>>()
-                .map(|strings| strings.join(""))
-                .map_err(|_| anyhow::anyhow!("Failed parsing Array<Felt> to String"))?
+                .map(|strings| strings.join(""))?
         } else {
             return Err(anyhow::anyhow!("token_uri is neither ByteArray nor Array<Felt>"));
         };
 
-        let metadata = if token_uri.is_empty() {
-            "".to_string()
+        Ok(token_uri)
+    }
+
+    async fn fetch_token_metadata(
+        contract_address: Felt,
+        token_id: U256,
+        token_uri: &str,
+    ) -> Result<String> {
+        if token_uri.is_empty() {
+            Ok("".to_string())
         } else {
-            let metadata = Self::fetch_metadata(&token_uri).await.with_context(|| {
-                format!(
-                    "Failed to fetch metadata for token_id: {}",
-                    register_erc721_token.actual_token_id
-                )
-            });
-
-            if let Ok(metadata) = metadata {
-                serde_json::to_string(&metadata).context("Failed to serialize metadata")?
-            } else {
-                warn!(
-                contract_address = format!("{:#x}", register_erc721_token.contract_address),
-                    token_id = %register_erc721_token.actual_token_id,
-                    "Error fetching metadata, empty metadata will be used instead.",
-                );
-                "".to_string()
+            let metadata = Self::fetch_metadata(token_uri)
+                .await
+                .with_context(|| format!("Failed to fetch metadata for token_id: {}", token_id));
+
+            match metadata {
+                Ok(metadata) => {
+                    serde_json::to_string(&metadata).context("Failed to serialize metadata")
+                }
+                Err(e) => {
+                    warn!(
+                        contract_address = format!("{:#x}", contract_address),
+                        token_id = %token_id,
+                        error = ?e,
+                        "Error fetching metadata, empty metadata will be used instead.",
+                    );
+                    Ok("".to_string())
+                }
             }
-        };
+        }
+    }
+
+    pub async fn update_erc721_metadata(
+        &mut self,
+        contract_address: Felt,
+        token_id: U256,
+        provider: Arc<P>,
+    ) -> Result<()> {
+        let token_uri = Self::fetch_token_uri(&provider, contract_address, token_id).await?;
+        let metadata = Self::fetch_token_metadata(contract_address, token_id, &token_uri).await?;
+
+        // Update metadata in database
+        sqlx::query("UPDATE tokens SET metadata = ? WHERE contract_address = ? AND id LIKE ?")
+            .bind(&metadata)
+            .bind(felt_to_sql_string(&contract_address))
+            .bind(format!("%{}", u256_to_sql_string(&token_id)))
+            .execute(&mut *self.transaction)
+            .await?;
+
+        Ok(())
+    }
+
+    pub async fn process_register_erc721_token_query(
+        register_erc721_token: RegisterErc721TokenQuery,
+        provider: Arc<P>,
+        name: String,
+        symbol: String,
+    ) -> Result<RegisterErc721TokenMetadata> {
+        let token_uri = Self::fetch_token_uri(
+            &provider,
+            register_erc721_token.contract_address,
+            register_erc721_token.actual_token_id,
+        )
+        .await?;
+
+        let metadata = Self::fetch_token_metadata(
+            register_erc721_token.contract_address,
+            register_erc721_token.actual_token_id,
+            &token_uri,
+        )
+        .await?;
 
         Ok(RegisterErc721TokenMetadata { query: register_erc721_token, metadata, name, symbol })
     }
@@ -269,7 +315,6 @@
     // metadata json schema
     pub async fn fetch_metadata(token_uri: &str) -> Result<serde_json::Value> {
         // Parse the token_uri
-
         match token_uri {
             uri if uri.starts_with("http") || uri.starts_with("https") => {
                 // Fetch metadata from HTTP/HTTPS URL

crates/torii/sqlite/src/executor/mod.rs

@@ -7,6 +7,7 @@
 use anyhow::{Context, Result};
 use cainome::cairo_serde::{ByteArray, CairoSerde};
 use dojo_types::schema::{Struct, Ty};
+use erc::UpdateErc721MetadataQuery;
 use sqlx::{FromRow, Pool, Sqlite, Transaction};
 use starknet::core::types::{BlockId, BlockTag, Felt, FunctionCall};
 use starknet::core::utils::{get_selector_from_name, parse_cairo_short_string};
@@ -115,10 +116,9 @@
     TokenTransfer,
     RegisterModel,
     StoreEvent,
-    // similar to execute but doesn't create a new transaction
+    UpdateErc721Metadata(UpdateErc721MetadataQuery),
     Flush,
     Execute,
-    // rollback's the current transaction and starts a new one
     Rollback,
     Other,
 }
@@ -763,6 +763,17 @@
                     res?;
                 }
             }
+            QueryType::UpdateErc721Metadata(update_metadata) => {
+                debug!(target: LOG_TARGET, "Updating ERC721 metadata.");
+                let instant = Instant::now();
+                self.update_erc721_metadata(
+                    update_metadata.contract_address,
+                    update_metadata.token_id,
+                    self.provider.clone(),
+                )
+                .await?;
+                debug!(target: LOG_TARGET, duration = ?instant.elapsed(), "Updated ERC721 metadata.");
+            }
             QueryType::Other => {
                 query.execute(&mut **tx).await.map_err(|e| {
                     anyhow::anyhow!(