Torii integration test work

.vscode/settings.json

@@ -0,0 +1,6 @@
+{
+    "rust-analyzer.linkedProjects": [
+        "./crates/sozo/Cargo.toml",
+        "./crates/sozo/Cargo.toml"
+    ]
+}

crates/dojo-test-utils/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod compiler;
+pub mod migration;
 pub mod rpc;
 pub mod sequencer;
 

crates/dojo-test-utils/src/migration.rs

@@ -0,0 +1,15 @@
+use std::path::PathBuf;
+
+use anyhow::Result;
+use camino::Utf8PathBuf;
+use dojo_world::manifest::Manifest;
+use dojo_world::migration::strategy::{prepare_for_migration, MigrationStrategy};
+use dojo_world::migration::world::WorldDiff;
+use starknet::macros::felt;
+
+pub fn prepare_migration(path: PathBuf) -> Result<MigrationStrategy> {
+    let target_dir = Utf8PathBuf::from_path_buf(path).unwrap();
+    let manifest = Manifest::load_from_path(target_dir.join("manifest.json")).unwrap();
+    let world = WorldDiff::compute(manifest, None);
+    prepare_for_migration(None, Some(felt!("0x12345")), target_dir, world)
+}

crates/dojo-types/src/component.rs

@@ -25,7 +25,9 @@ impl Ty {
     }
 
     pub fn flatten(&self) -> Vec<Ty> {
-        Ty::flatten_ty(self.clone())
+        let mut flattened = Ty::flatten_ty(self.clone());
+        flattened.reverse();
+        flattened
     }
 
     fn flatten_ty(ty: Ty) -> Vec<Ty> {

crates/sozo/src/commands/completions.rs

@@ -4,7 +4,7 @@ use anyhow::Result;
 use clap::{Args, CommandFactory};
 use clap_complete::{generate, Shell};
 
-use crate::SozoArgs;
+use crate::args::SozoArgs;
 
 #[derive(Args, Debug)]
 pub struct CompletionsArgs {

crates/sozo/src/lib.rs

@@ -0,0 +1,3 @@
+pub mod args;
+pub mod commands;
+pub mod ops;

crates/sozo/src/main.rs

@@ -8,12 +8,7 @@ use dojo_lang::plugin::CairoPluginRepository;
 use scarb::compiler::CompilerRepository;
 use scarb::core::Config;
 use scarb_ui::{OutputFormat, Ui};
-
-mod args;
-mod commands;
-mod ops;
-
-use args::{Commands, SozoArgs};
+use sozo::args::{Commands, SozoArgs};
 
 fn main() {
     let args = SozoArgs::parse();
@@ -46,5 +41,5 @@ fn cli_main(args: SozoArgs) -> Result<()> {
         .compilers(compilers)
         .build()?;
 
-    commands::run(args.command, &config)
+    sozo::commands::run(args.command, &config)
 }

crates/sozo/src/ops/migration/migration_test.rs

@@ -1,10 +1,10 @@
-use anyhow::Result;
 use camino::Utf8PathBuf;
+use dojo_test_utils::migration::prepare_migration;
 use dojo_test_utils::sequencer::{
     get_default_test_starknet_config, SequencerConfig, StarknetConfig, TestSequencer,
 };
 use dojo_world::manifest::Manifest;
-use dojo_world::migration::strategy::{prepare_for_migration, MigrationStrategy};
+use dojo_world::migration::strategy::prepare_for_migration;
 use dojo_world::migration::world::WorldDiff;
 use scarb_ui::{OutputFormat, Ui, Verbosity};
 use starknet::accounts::{ExecutionEncoding, SingleOwnerAccount};
@@ -18,17 +18,10 @@ use starknet::signers::{LocalWallet, SigningKey};
 use crate::commands::options::transaction::TransactionOptions;
 use crate::ops::migration::execute_strategy;
 
-pub fn prepare_example_ecs_migration() -> Result<MigrationStrategy> {
-    let target_dir = Utf8PathBuf::from_path_buf("../../examples/ecs/target/dev".into()).unwrap();
-    let manifest = Manifest::load_from_path(target_dir.join("manifest.json")).unwrap();
-    let world = WorldDiff::compute(manifest, None);
-    prepare_for_migration(None, Some(felt!("0x12345")), target_dir, world)
-}
-
 #[tokio::test(flavor = "multi_thread")]
 async fn migrate_with_auto_mine() {
     let ui = Ui::new(Verbosity::Verbose, OutputFormat::Text);
-    let migration = prepare_example_ecs_migration().unwrap();
+    let migration = prepare_migration("../../examples/ecs/target/dev".into()).unwrap();
 
     let sequencer =
         TestSequencer::start(SequencerConfig::default(), get_default_test_starknet_config()).await;
@@ -37,13 +30,14 @@ async fn migrate_with_auto_mine() {
     account.set_block_id(BlockId::Tag(BlockTag::Pending));
 
     execute_strategy(&migration, &account, &ui, None).await.unwrap();
+
     sequencer.stop().unwrap();
 }
 
 #[tokio::test(flavor = "multi_thread")]
 async fn migrate_with_block_time() {
     let ui = Ui::new(Verbosity::Verbose, OutputFormat::Text);
-    let migration = prepare_example_ecs_migration().unwrap();
+    let migration = prepare_migration("../../examples/ecs/target/dev".into()).unwrap();
 
     let sequencer = TestSequencer::start(
         SequencerConfig { block_time: Some(1000), ..Default::default() },
@@ -61,7 +55,7 @@ async fn migrate_with_block_time() {
 #[tokio::test(flavor = "multi_thread")]
 async fn migrate_with_small_fee_multiplier_will_fail() {
     let ui = Ui::new(Verbosity::Verbose, OutputFormat::Text);
-    let migration = prepare_example_ecs_migration().unwrap();
+    let migration = prepare_migration("../../examples/ecs/target/dev".into()).unwrap();
 
     let sequencer = TestSequencer::start(
         Default::default(),
@@ -79,14 +73,16 @@ async fn migrate_with_small_fee_multiplier_will_fail() {
         ExecutionEncoding::Legacy,
     );
 
-    assert!(execute_strategy(
-        &migration,
-        &account,
-        &ui,
-        Some(TransactionOptions { fee_estimate_multiplier: Some(0.2f64) }),
-    )
-    .await
-    .is_err());
+    assert!(
+        execute_strategy(
+            &migration,
+            &account,
+            &ui,
+            Some(TransactionOptions { fee_estimate_multiplier: Some(0.2f64) }),
+        )
+        .await
+        .is_err()
+    );
     sequencer.stop().unwrap();
 }
 

crates/sozo/src/ops/migration/mod.rs

@@ -240,7 +240,7 @@ where
 
 // returns the Some(block number) at which migration world is deployed, returns none if world was
 // not redeployed
-async fn execute_strategy<P, S>(
+pub async fn execute_strategy<P, S>(
     strategy: &MigrationStrategy,
     migrator: &SingleOwnerAccount<P, S>,
     ui: &Ui,

crates/torii/core/src/engine.rs

@@ -7,10 +7,9 @@ use starknet::core::types::{
 };
 use starknet::core::utils::get_selector_from_name;
 use starknet::providers::Provider;
-use starknet_crypto::FieldElement;
 use tokio::time::sleep;
+use tokio_util::sync::CancellationToken;
 use torii_client::contract::world::WorldContractReader;
-
 use tracing::{error, info, warn};
 
 use crate::processors::{BlockProcessor, EventProcessor, TransactionProcessor};
@@ -31,29 +30,30 @@ impl<P: Provider + Sync + Send> Default for Processors<P> {
 #[derive(Debug)]
 pub struct EngineConfig {
     pub block_time: Duration,
-    pub world_address: FieldElement,
     pub start_block: u64,
 }
 
 impl Default for EngineConfig {
     fn default() -> Self {
-        Self {
-            block_time: Duration::from_secs(1),
-            world_address: FieldElement::ZERO,
-            start_block: 0,
-        }
+        Self { block_time: Duration::from_secs(1), start_block: 0 }
     }
 }
 
-pub struct Engine<'a, P: Provider + Sync + Send> {
+pub struct Engine<'a, P: Provider + Sync + Send>
+where
+    P::Error: 'static,
+{
     world: &'a WorldContractReader<'a, P>,
     db: &'a Sql,
     provider: &'a P,
     processors: Processors<P>,
     config: EngineConfig,
 }
 
-impl<'a, P: Provider + Sync + Send> Engine<'a, P> {
+impl<'a, P: Provider + Sync + Send> Engine<'a, P>
+where
+    P::Error: 'static,
+{
     pub fn new(
         world: &'a WorldContractReader<'a, P>,
         db: &'a Sql,
@@ -64,10 +64,10 @@ impl<'a, P: Provider + Sync + Send> Engine<'a, P> {
         Self { world, db, provider, processors, config }
     }
 
-    pub async fn start(&self) -> Result<(), Box<dyn Error>> {
+    pub async fn start(&self, cts: CancellationToken) -> Result<(), Box<dyn Error>> {
         let db_head = self.db.head().await?;
 
-        let mut current_block_number = match db_head {
+        let current_block_number = match db_head {
             0 => self.config.start_block,
             _ => {
                 if self.config.start_block != 0 {
@@ -78,45 +78,57 @@ impl<'a, P: Provider + Sync + Send> Engine<'a, P> {
         };
 
         loop {
+            if cts.is_cancelled() {
+                break Ok(());
+            }
+
             sleep(self.config.block_time).await;
+            match self.sync_to_head(current_block_number).await {
+                Ok(block_with_txs) => block_with_txs,
+                Err(e) => {
+                    error!("getting  block: {}", e);
+                    continue;
+                }
+            };
+        }
+    }
 
-            let latest_block_with_txs =
-                match self.provider.get_block_with_txs(BlockId::Tag(BlockTag::Latest)).await {
-                    Ok(block_with_txs) => block_with_txs,
-                    Err(e) => {
-                        error!("getting  block: {}", e);
-                        continue;
-                    }
-                };
+    pub async fn sync_to_head(&self, from: u64) -> Result<u64, Box<dyn Error>> {
+        let latest_block_with_txs =
+            self.provider.get_block_with_txs(BlockId::Tag(BlockTag::Latest)).await?;
+
+        let latest_block_number = match latest_block_with_txs {
+            MaybePendingBlockWithTxs::Block(latest_block_with_txs) => {
+                latest_block_with_txs.block_number
+            }
+            _ => return Err(anyhow::anyhow!("Getting latest block number").into()),
+        };
+
+        self.sync_range(from, latest_block_number).await?;
+
+        Ok(latest_block_number)
+    }
 
-            let latest_block_number = match latest_block_with_txs {
-                MaybePendingBlockWithTxs::Block(latest_block_with_txs) => {
-                    latest_block_with_txs.block_number
+    pub async fn sync_range(&self, mut from: u64, to: u64) -> Result<(), Box<dyn Error>> {
+        // Process all blocks from current to latest.
+        while from <= to {
+            let block_with_txs = match self.provider.get_block_with_txs(BlockId::Number(from)).await
+            {
+                Ok(block_with_txs) => block_with_txs,
+                Err(e) => {
+                    error!("getting block: {}", e);
+                    continue;
                 }
-                _ => continue,
             };
 
-            // Process all blocks from current to latest.
-            while current_block_number <= latest_block_number {
-                let block_with_txs = match self
-                    .provider
-                    .get_block_with_txs(BlockId::Number(current_block_number))
-                    .await
-                {
-                    Ok(block_with_txs) => block_with_txs,
-                    Err(e) => {
-                        error!("getting block: {}", e);
-                        continue;
-                    }
-                };
-
-                self.process(block_with_txs).await?;
+            self.process(block_with_txs).await?;
 
-                self.db.set_head(current_block_number).await?;
-                self.db.execute().await?;
-                current_block_number += 1;
-            }
+            self.db.set_head(from).await?;
+            self.db.execute().await?;
+            from += 1;
         }
+
+        Ok(())
     }
 
     async fn process(&self, block: MaybePendingBlockWithTxs) -> Result<(), Box<dyn Error>> {
@@ -154,7 +166,7 @@ impl<'a, P: Provider + Sync + Send> Engine<'a, P> {
 
             if let TransactionReceipt::Invoke(invoke_receipt) = receipt.clone() {
                 for (event_idx, event) in invoke_receipt.events.iter().enumerate() {
-                    if event.from_address != self.config.world_address {
+                    if event.from_address != self.world.address {
                         continue;
                     }
 
@@ -188,7 +200,7 @@ impl<'a, P: Provider + Sync + Send> Engine<'a, P> {
     }
 }
 
-async fn process_block<P: Provider + Sync>(
+async fn process_block<P: Provider + Sync + Send>(
     db: &Sql,
     provider: &P,
     processors: &[Box<dyn BlockProcessor<P>>],
@@ -200,7 +212,7 @@ async fn process_block<P: Provider + Sync>(
     Ok(())
 }
 
-async fn process_transaction<P: Provider + Sync>(
+async fn process_transaction<P: Provider + Sync + Send>(
     db: &Sql,
     provider: &P,
     processors: &[Box<dyn TransactionProcessor<P>>],
@@ -215,7 +227,7 @@ async fn process_transaction<P: Provider + Sync>(
 }
 
 #[allow(clippy::too_many_arguments)]
-async fn process_event<P: Provider + Sync>(
+async fn process_event<P: Provider + Sync + Send>(
     world: &WorldContractReader<'_, P>,
     db: &Sql,
     provider: &P,

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

@@ -9,7 +9,7 @@ use crate::sql::Sql;
 pub mod register_model;
 pub mod register_system;
 pub mod store_set_record;
-pub mod store_system_call;
+// pub mod store_system_call;
 
 #[async_trait]
 pub trait EventProcessor<P: Provider + Sync> {