fix(torii-core): engine update cursor when world event

crates/torii/core/src/engine.rs

@@ -18,6 +18,9 @@
 
 use crate::processors::{BlockProcessor, EventProcessor, TransactionProcessor};
 use crate::sql::Sql;
+
+type HasWorldEvent = bool;
+
 #[allow(missing_debug_implementations)]
 pub struct Processors<P: Provider + Sync> {
     pub block: Vec<Box<dyn BlockProcessor<P>>>,
@@ -189,12 +192,14 @@
                         }
                     }
                 }
+                Ok(true) => {
+                    pending_block_tx = Some(*transaction.transaction_hash());
+                    info!(target: LOG_TARGET, transaction_hash = %format!("{:#x}", transaction.transaction_hash()), "Processed pending world transaction.");
+                }
                 Ok(_) => {
                     info!(target: LOG_TARGET, transaction_hash = %format!("{:#x}", transaction.transaction_hash()), "Processed pending transaction.")
                 }
             }
-
-            pending_block_tx = Some(*transaction.transaction_hash());
         }
 
         // Set the head to the last processed pending transaction
@@ -305,16 +310,17 @@
         }
 
         // Process blocks
-        for (block_number, block_timestamp) in blocks.iter() {
-            if let Some(ref block_tx) = self.block_tx {
-                block_tx.send(*block_number).await?;
-            }
+        // for (block_number, block_timestamp) in blocks.iter() {
+        //     if let Some(ref block_tx) = self.block_tx {
+        //         block_tx.send(*block_number).await?;
+        //     }
 
-            self.process_block(*block_number, *block_timestamp).await?;
-            info!(target: LOG_TARGET, block_number = %block_number, "Processed block.");
-        }
+        //     self.process_block(*block_number, *block_timestamp).await?;
+        //
+        // }
 
         // Process all transactions
+        let mut last_block = 0;
         for (block_number, transaction_hash) in transactions {
             // Process transaction
             let transaction = self.provider.get_transaction_by_hash(transaction_hash).await?;
@@ -326,6 +332,16 @@
                 blocks[&block_number],
             )
             .await?;
+
+            // Process block
+            if block_number > last_block {
+                if let Some(ref block_tx) = self.block_tx {
+                    block_tx.send(block_number).await?;
+                }
+
+                self.process_block(block_number, blocks[&block_number]).await?;
+                last_block = block_number;
+            }
         }
 
         // We return None for the pending_block_tx because our sync_range
@@ -354,16 +370,16 @@
         transaction: &Transaction,
         block_number: u64,
         block_timestamp: u64,
-    ) -> Result<()> {
+    ) -> Result<HasWorldEvent> {
         let receipt = self.provider.get_transaction_receipt(transaction_hash).await?;
         let events = match &receipt.receipt {
             TransactionReceipt::Invoke(receipt) => Some(&receipt.events),
             TransactionReceipt::L1Handler(receipt) => Some(&receipt.events),
             _ => None,
         };
 
+        let mut world_event = false;
         if let Some(events) = events {
-            let mut world_event = false;
             for (event_idx, event) in events.iter().enumerate() {
                 if event.from_address != self.world.address {
                     continue;
@@ -397,7 +413,7 @@
             }
         }
 
-        Ok(())
+        Ok(world_event)
     }
 
     async fn process_block(&mut self, block_number: u64, block_timestamp: u64) -> Result<()> {
@@ -406,6 +422,8 @@
                 .process(&mut self.db, self.provider.as_ref(), block_number, block_timestamp)
                 .await?
         }
+
+        info!(target: LOG_TARGET, block_number = %block_number, "Processed block.");
         Ok(())
     }