feat(torii): fetch and process erc721 metadata and image

commit-id:274aaa3a

Cargo.toml

@@ -157,13 +157,16 @@ clap_complete = "4.3"
 console = "0.15.7"
 convert_case = "0.6.0"
 crypto-bigint = { version = "0.5.3", features = [ "serde" ] }
+data-url = "0.3"
 derive_more = "0.99.17"
 flate2 = "1.0.24"
+fluent-uri = "0.3"
 futures = "0.3.30"
 futures-util = "0.3.30"
 hashlink = "0.9.1"
 hex = "0.4.3"
 http = "0.2.9"
+image = "0.25.2"
 indexmap = "2.2.5"
 indoc = "1.0.7"
 itertools = "0.12.1"

bin/torii/src/main.rs

@@ -19,6 +19,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::Context;
+use camino::Utf8PathBuf;
 use clap::{ArgAction, Parser};
 use dojo_metrics::{metrics_process, prometheus_exporter};
 use dojo_utils::parse::{parse_socket_address, parse_url};
@@ -146,6 +147,10 @@ struct Args {
     /// Configuration file
     #[arg(long)]
     config: Option<PathBuf>,
+
+    /// Path to a directory to store ERC artifacts
+    #[arg(long)]
+    artifacts_path: Utf8PathBuf,
 }
 
 #[tokio::main]
@@ -218,7 +223,8 @@ async fn main() -> anyhow::Result<()> {
         executor.run().await.unwrap();
     });
 
-    let db = Sql::new(pool.clone(), sender.clone(), &contracts).await?;
+    let db =
+        Sql::new(pool.clone(), sender.clone(), &contracts, args.artifacts_path.clone()).await?;
 
     let processors = Processors {
         transaction: vec![Box::new(StoreTransactionProcessor)],
@@ -254,10 +260,19 @@ async fn main() -> anyhow::Result<()> {
         Arc::new(contracts),
     );
 
-    let shutdown_rx = shutdown_tx.subscribe();
-    let (grpc_addr, grpc_server) =
-        torii_grpc::server::new(shutdown_rx, &pool, block_rx, world_address, Arc::clone(&provider))
-            .await?;
+    let (grpc_addr, grpc_server) = torii_grpc::server::new(
+        shutdown_tx.subscribe(),
+        &pool,
+        block_rx,
+        world_address,
+        Arc::clone(&provider),
+    )
+    .await?;
+
+    tokio::fs::create_dir_all(args.artifacts_path.clone()).await?;
+    let absolute_path = args.artifacts_path.canonicalize()?;
+    let (artifacts_addr, artifacts_server) =
+        torii_server::artifacts::new(shutdown_tx.subscribe(), absolute_path).await?;
 
     let mut libp2p_relay_server = torii_relay::server::Relay::new(
         db,
@@ -270,7 +285,13 @@ async fn main() -> anyhow::Result<()> {
     )
     .expect("Failed to start libp2p relay server");
 
-    let proxy_server = Arc::new(Proxy::new(args.addr, args.allowed_origins, Some(grpc_addr), None));
+    let proxy_server = Arc::new(Proxy::new(
+        args.addr,
+        args.allowed_origins,
+        Some(grpc_addr),
+        None,
+        Some(artifacts_addr),
+    ));
 
     let graphql_server = spawn_rebuilding_graphql_server(
         shutdown_tx.clone(),
@@ -314,6 +335,7 @@ async fn main() -> anyhow::Result<()> {
         _ = graphql_server => {},
         _ = grpc_server => {},
         _ = libp2p_relay_server.run() => {},
+        _ = artifacts_server => {},
         _ = dojo_utils::signal::wait_signals() => {},
     };
 

crates/torii/core/Cargo.toml

@@ -14,13 +14,17 @@ async-trait.workspace = true
 base64.workspace = true
 bitflags = "2.6.0"
 cainome.workspace = true
+camino.workspace = true
 chrono.workspace = true
 crypto-bigint.workspace = true
+data-url.workspace = true
 dojo-types.workspace = true
 dojo-world = { workspace = true, features = [ "contracts", "manifest" ] }
+fluent-uri.workspace = true
 futures-channel = "0.3.0"
 futures-util.workspace = true
 hashlink.workspace = true
+image.workspace = true
 num-traits.workspace = true
 once_cell.workspace = true
 reqwest.workspace = true

crates/torii/core/src/engine.rs

@@ -264,11 +264,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                                 },
                                 Err(e) => {
                                     error!(target: LOG_TARGET, error = %e, "Processing fetched data.");
-                                    erroring_out = true;
-                                    sleep(backoff_delay).await;
-                                    if backoff_delay < max_backoff_delay {
-                                        backoff_delay *= 2;
-                                    }
+                                    return Err(e);
                                 }
                             }
                             debug!(target: LOG_TARGET, duration = ?instant.elapsed(), "Processed fetched data.");
@@ -853,6 +849,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                     .await
                 {
                     error!(target: LOG_TARGET, event_name = processor.event_key(), error = ?e, "Processing event.");
+                    return Err(e);
                 }
             } else {
                 warn!(target: LOG_TARGET, event_name = processor.event_key(), "Event not validated.");

crates/torii/core/src/executor.rs

@@ -12,7 +12,8 @@ use tokio::sync::broadcast::{Receiver, Sender};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
 use tokio::sync::oneshot;
 use tokio::time::Instant;
-use tracing::{debug, error};
+use tracing::{debug, error}; /* Added import for data URI parsing // Uncommented to reuse
+                              * HTTP client */
 
 use crate::simple_broker::SimpleBroker;
 use crate::sql::utils::{felt_to_sql_string, sql_string_to_u256, u256_to_sql_string, I256};
@@ -81,6 +82,16 @@ pub struct UpdateCursorsQuery {
     pub pending_block_timestamp: u64,
 }
 
+#[derive(Debug, Clone)]
+pub struct RegisterErc721TokenQuery {
+    pub token_id: String,
+    pub contract_address: Felt,
+    pub name: String,
+    pub symbol: String,
+    pub metadata: String,
+    pub image_path: String,
+}
+
 #[derive(Debug, Clone)]
 pub enum QueryType {
     SetHead(SetHeadQuery),
@@ -90,6 +101,7 @@ pub enum QueryType {
     DeleteEntity(DeleteEntityQuery),
     EventMessage(Ty),
     ApplyBalanceDiff(ApplyBalanceDiffQuery),
+    RegisterErc721Token(RegisterErc721TokenQuery),
     RegisterModel,
     StoreEvent,
     Execute,
@@ -471,6 +483,26 @@ impl<'c> Executor<'c> {
                 self.apply_balance_diff(apply_balance_diff).await?;
                 debug!(target: LOG_TARGET, duration = ?instant.elapsed(), "Applied balance diff.");
             }
+            QueryType::RegisterErc721Token(register_erc721_token) => {
+                let query = sqlx::query(
+                    "INSERT INTO tokens (id, contract_address, name, symbol, decimals, metadata, \
+                     image_url) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                )
+                .bind(&register_erc721_token.token_id)
+                .bind(felt_to_sql_string(&register_erc721_token.contract_address))
+                .bind(&register_erc721_token.name)
+                .bind(&register_erc721_token.symbol)
+                .bind(0)
+                .bind(&register_erc721_token.metadata)
+                .bind(&register_erc721_token.image_path);
+
+                query.execute(&mut **tx).await.with_context(|| {
+                    format!(
+                        "Failed to execute RegisterErc721Token query: {:?}",
+                        register_erc721_token
+                    )
+                })?;
+            }
             QueryType::Execute => {
                 debug!(target: LOG_TARGET, "Executing query.");
                 let instant = Instant::now();