feat(torii): add support for erc1155

crates/torii/indexer/src/engine.rs

@@ -28,6 +28,8 @@ use torii_sqlite::{Cursors, Sql};
 use tracing::{debug, error, info, trace, warn};
 
 use crate::constants::LOG_TARGET;
+use crate::processors::erc1155_transfer_batch::Erc1155TransferBatchProcessor;
+use crate::processors::erc1155_transfer_single::Erc1155TransferSingleProcessor;
 use crate::processors::erc20_legacy_transfer::Erc20LegacyTransferProcessor;
 use crate::processors::erc20_transfer::Erc20TransferProcessor;
 use crate::processors::erc721_legacy_transfer::Erc721LegacyTransferProcessor;
@@ -105,6 +107,13 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Processors<P> {
                     Box::new(Erc721LegacyTransferProcessor) as Box<dyn EventProcessor<P>>,
                 ],
             ),
+            (
+                ContractType::ERC1155,
+                vec![
+                    Box::new(Erc1155TransferBatchProcessor) as Box<dyn EventProcessor<P>>,
+                    Box::new(Erc1155TransferSingleProcessor) as Box<dyn EventProcessor<P>>,
+                ],
+            ),
         ];
 
         for (contract_type, processors) in event_processors {

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

@@ -0,0 +1,113 @@
+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::{self, TaskId, TaskPriority};
+
+pub(crate) const LOG_TARGET: &str = "torii_indexer::processors::erc1155_transfer_batch";
+
+#[derive(Default, Debug)]
+pub struct Erc1155TransferBatchProcessor;
+
+#[async_trait]
+impl<P> EventProcessor<P> for Erc1155TransferBatchProcessor
+where
+    P: Provider + Send + Sync + std::fmt::Debug,
+{
+    fn event_key(&self) -> String {
+        "TransferBatch".to_string()
+    }
+
+    fn validate(&self, event: &Event) -> bool {
+        // key: [hash(TransferBatch), operator, from, to]
+        // data: [ids_len, ids[0].low, ids[0].high, ..., values_len, values[0].low, values[0].high,
+        // ...]
+        event.keys.len() == 4 && !event.data.is_empty()
+    }
+
+    fn task_priority(&self) -> TaskPriority {
+        1
+    }
+
+    fn task_identifier(&self, _event: &Event) -> TaskId {
+        task_manager::TASK_ID_SEQUENTIAL
+    }
+
+    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;
+        let from = event.keys[2];
+        let to = event.keys[3];
+
+        // ERC1155 TransferBatch event data format:
+        // - ids_len: felt (first element)
+        // - ids: U256[] (each element stored as 2 felts: [low, high])
+        // - values_len: felt
+        // - values: U256[] (each element stored as 2 felts: [low, high])
+        // Spec reference: https://eips.ethereum.org/EIPS/eip-1155#transferbatch
+        let ids_len = event.data[0].try_into().unwrap_or(0u64) as usize;
+        let mut current_idx = 1;
+
+        // First pass: read all token IDs
+        let mut token_ids = Vec::with_capacity(ids_len);
+        for _ in 0..ids_len {
+            if current_idx + 1 >= event.data.len() {
+                break;
+            }
+            let token_id = U256Cainome::cairo_deserialize(&event.data, current_idx)?;
+            token_ids.push(U256::from_words(token_id.low, token_id.high));
+            current_idx += 2;
+        }
+
+        // Move index to values array
+        let values_len = event.data[current_idx].try_into().unwrap_or(0u64) as usize;
+        current_idx += 1;
+
+        // Second pass: read and process amounts
+        for (idx, token_id) in token_ids.iter().enumerate() {
+            if idx >= values_len || current_idx + (idx * 2) + 1 >= event.data.len() {
+                break;
+            }
+
+            let amount = U256Cainome::cairo_deserialize(&event.data, current_idx + (idx * 2))?;
+            let amount = U256::from_words(amount.low, amount.high);
+
+            db.handle_nft_transfer(
+                token_address,
+                from,
+                to,
+                *token_id,
+                amount,
+                block_timestamp,
+                event_id,
+                block_number,
+            )
+            .await?;
+
+            debug!(
+                target: LOG_TARGET,
+                from = ?from,
+                to = ?to,
+                token_id = ?token_id,
+                amount = ?amount,
+                "ERC1155 TransferBatch"
+            );
+        }
+
+        Ok(())
+    }
+}

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

@@ -0,0 +1,76 @@
+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::{self, TaskId, TaskPriority};
+
+pub(crate) const LOG_TARGET: &str = "torii_indexer::processors::erc1155_transfer_single";
+
+#[derive(Default, Debug)]
+pub struct Erc1155TransferSingleProcessor;
+
+#[async_trait]
+impl<P> EventProcessor<P> for Erc1155TransferSingleProcessor
+where
+    P: Provider + Send + Sync + std::fmt::Debug,
+{
+    fn event_key(&self) -> String {
+        "TransferSingle".to_string()
+    }
+
+    fn validate(&self, event: &Event) -> bool {
+        // key: [hash(TransferSingle), operator, from, to]
+        // data: [id.low, id.high, value.low, value.high]
+        event.keys.len() == 4 && event.data.len() == 4
+    }
+
+    fn task_priority(&self) -> TaskPriority {
+        1
+    }
+
+    fn task_identifier(&self, _event: &Event) -> TaskId {
+        task_manager::TASK_ID_SEQUENTIAL
+    }
+
+    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;
+        let from = event.keys[2];
+        let to = event.keys[3];
+
+        let token_id = U256Cainome::cairo_deserialize(&event.data, 0)?;
+        let token_id = U256::from_words(token_id.low, token_id.high);
+
+        let amount = U256Cainome::cairo_deserialize(&event.data, 2)?;
+        let amount = U256::from_words(amount.low, amount.high);
+
+        db.handle_nft_transfer(
+            token_address,
+            from,
+            to,
+            token_id,
+            amount,
+            block_timestamp,
+            event_id,
+            block_number,
+        )
+        .await?;
+        debug!(target: LOG_TARGET, from = ?from, to = ?to, token_id = ?token_id, amount = ?amount, "ERC1155 TransferSingle");
+
+        Ok(())
+    }
+}

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

@@ -80,11 +80,12 @@
         let token_id = U256Cainome::cairo_deserialize(&event.data, 2)?;
         let token_id = U256::from_words(token_id.low, token_id.high);
 
-        db.handle_erc721_transfer(
+        db.handle_nft_transfer(
             token_address,
             from,
             to,
             token_id,
+            U256::from(1u8),
             block_timestamp,
             event_id,
             block_number,

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

@@ -80,11 +80,12 @@
         let token_id = U256Cainome::cairo_deserialize(&event.keys, 3)?;
         let token_id = U256::from_words(token_id.low, token_id.high);
 
-        db.handle_erc721_transfer(
+        db.handle_nft_transfer(
             token_address,
             from,
             to,
             token_id,
+            U256::from(1u8),
             block_timestamp,
             event_id,
             block_number,

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

@@ -9,6 +9,8 @@ use torii_sqlite::Sql;
 
 use crate::task_manager::{TaskId, TaskPriority};
 
+pub mod erc1155_transfer_batch;
+pub mod erc1155_transfer_single;
 pub mod erc20_legacy_transfer;
 pub mod erc20_transfer;
 pub mod erc721_legacy_transfer;

crates/torii/sqlite/src/erc.rs

@@ -10,9 +10,9 @@
 use super::utils::{u256_to_sql_string, I256};
 use super::{Sql, SQL_FELT_DELIMITER};
 use crate::constants::TOKEN_TRANSFER_TABLE;
+use crate::executor::erc::RegisterNftTokenQuery;
 use crate::executor::{
     ApplyBalanceDiffQuery, Argument, QueryMessage, QueryType, RegisterErc20TokenQuery,
-    RegisterErc721TokenQuery,
 };
 use crate::types::ContractType;
 use crate::utils::{
@@ -79,12 +79,13 @@
     }
 
     #[allow(clippy::too_many_arguments)]
-    pub async fn handle_erc721_transfer(
+    pub async fn handle_nft_transfer(
         &mut self,
         contract_address: Felt,
         from_address: Felt,
         to_address: Felt,
         token_id: U256,
+        amount: U256,
         block_timestamp: u64,
         event_id: &str,
         block_number: u64,
@@ -95,15 +96,14 @@
         let token_exists: bool = self.local_cache.contains_token_id(&token_id).await;
 
         if !token_exists {
-            self.register_erc721_token_metadata(contract_address, &token_id, actual_token_id)
-                .await?;
+            self.register_nft_token_metadata(contract_address, &token_id, actual_token_id).await?;
         }
 
         self.store_erc_transfer_event(
             contract_address,
             from_address,
             to_address,
-            U256::from(1u8),
+            amount,
             &token_id,
             block_timestamp,
             event_id,
@@ -120,15 +120,15 @@
                 );
                 let from_balance =
                     erc_cache.entry((ContractType::ERC721, from_balance_id)).or_default();
-                *from_balance -= I256::from(1u8);
+                *from_balance -= I256::from(amount);
             }
 
             if to_address != Felt::ZERO {
                 let to_balance_id =
                     format!("{}{SQL_FELT_DELIMITER}{}", felt_to_sql_string(&to_address), &token_id);
                 let to_balance =
                     erc_cache.entry((ContractType::ERC721, to_balance_id)).or_default();
-                *to_balance += I256::from(1u8);
+                *to_balance += I256::from(amount);
             }
         }
 
@@ -220,7 +220,7 @@
         Ok(())
     }
 
-    async fn register_erc721_token_metadata(
+    async fn register_nft_token_metadata(
         &mut self,
         contract_address: Felt,
         token_id: &str,
@@ -229,7 +229,7 @@
         self.executor.send(QueryMessage::new(
             "".to_string(),
             vec![],
-            QueryType::RegisterErc721Token(RegisterErc721TokenQuery {
+            QueryType::RegisterNftToken(RegisterNftTokenQuery {
                 token_id: token_id.to_string(),
                 contract_address,
                 actual_token_id,