Introspect/bin init

Cargo.toml

@@ -6,6 +6,7 @@ members = [
   "crates/torii-runtime-common",
   "crates/torii-sql-sink",
   "crates/torii-log-sink",
+  "crates/torii-ecs-sink",
   "crates/torii-erc20",
   "crates/torii-erc721",
   "crates/torii-erc1155",
@@ -126,6 +127,7 @@ torii-runtime-common = { path = "crates/torii-runtime-common" }
 # Internal crates
 torii-sql-sink.path = "crates/torii-sql-sink"
 torii-log-sink.path = "crates/torii-log-sink"
+torii-ecs-sink.path = "crates/torii-ecs-sink"
 torii-erc20.path = "crates/torii-erc20"
 torii-erc721.path = "crates/torii-erc721"
 torii-erc1155.path = "crates/torii-erc1155"

bins/torii-introspect-bin/Cargo.toml

@@ -12,6 +12,7 @@ path = "src/main.rs"
 torii = { path = "../../" }
 torii-config-common.workspace = true
 torii-dojo = { path = "../../crates/dojo" }
+torii-ecs-sink = { path = "../../crates/torii-ecs-sink" }
 torii-introspect-postgres-sink = { path = "../../crates/introspect-postgres-sink" }
 torii-introspect-sqlite-sink = { path = "../../crates/introspect-sqlite-sink" }
 torii-sqlite = { path = "../../crates/sqlite" }
@@ -24,6 +25,9 @@ url = "2.5"
 clap = { version = "4.5", features = ["derive", "env"] }
 anyhow = "1.0"
 sqlx = { version = "0.8", features = ["postgres", "sqlite", "runtime-tokio-rustls", "any"] }
+tonic.workspace = true
+tonic-reflection.workspace = true
+tonic-web.workspace = true
 
 [lints]
 workspace = true

bins/torii-introspect-bin/src/config.rs

@@ -81,9 +81,6 @@ pub struct Config {
     pub max_db_connections: Option<u32>,
 
     /// Ignore persisted extractor state and force extraction from `from_block`.
-    ///
-    /// This binary no longer supports historical schema bootstrap, so resumed
-    /// runs must opt into a fresh extraction window explicitly.
     #[arg(long)]
     pub ignore_saved_state: bool,
 }
@@ -133,31 +130,18 @@ mod tests {
 
     #[test]
     fn observability_defaults_to_disabled() {
-        let cfg = Config::parse_from([
-            "torii-introspect",
-            "--contracts",
-            "0x1",
-            "--storage-database-url",
-            "postgres://localhost/torii",
-        ]);
+        let cfg = Config::parse_from(["torii-introspect", "--contracts", "0x1"]);
         assert!(!cfg.observability);
     }
 
     #[test]
     fn observability_flag_enables_metrics() {
-        let cfg = Config::parse_from([
-            "torii-introspect",
-            "--contracts",
-            "0x1",
-            "--storage-database-url",
-            "postgres://localhost/torii",
-            "--observability",
-        ]);
+        let cfg = Config::parse_from(["torii-introspect", "--contracts", "0x1", "--observability"]);
         assert!(cfg.observability);
     }
 
     #[test]
-    fn storage_database_url_must_be_postgres() {
+    fn storage_database_url_accepts_sqlite() {
         let cfg = Config::parse_from([
             "torii-introspect",
             "--contracts",
@@ -182,13 +166,7 @@ mod tests {
 
     #[test]
     fn contract_addresses_parse_from_hex() {
-        let cfg = Config::parse_from([
-            "torii-introspect",
-            "--contracts",
-            "0x1,0x2",
-            "--storage-database-url",
-            "postgres://localhost/torii",
-        ]);
+        let cfg = Config::parse_from(["torii-introspect", "--contracts", "0x1,0x2"]);
 
         let contracts = cfg.contract_addresses().unwrap();
         assert_eq!(contracts.len(), 2);

bins/torii-introspect-bin/src/main.rs

@@ -2,26 +2,48 @@
 
 mod config;
 
-use anyhow::{bail, Result};
+use anyhow::Result;
 use clap::Parser;
 use config::{Config, StorageBackend};
 use sqlx::postgres::PgPoolOptions;
-use sqlx::sqlite::SqlitePoolOptions;
+use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
 use starknet::core::types::Felt;
 use std::path::Path;
+use std::str::FromStr;
 use std::sync::Arc;
 use torii::etl::decoder::DecoderId;
-use torii::etl::engine_db::{EngineDb, EngineDbConfig};
 use torii::etl::extractor::{
     ContractEventConfig, EventExtractor, EventExtractorConfig, RetryPolicy,
 };
 use torii_config_common::apply_observability_env;
 use torii_dojo::decoder::DojoDecoder;
 use torii_dojo::store::postgres::PgStore;
 use torii_dojo::store::sqlite::SqliteStore;
+use torii_ecs_sink::proto::world::world_server::WorldServer;
+use torii_ecs_sink::{EcsSink, FILE_DESCRIPTOR_SET as ECS_DESCRIPTOR_SET};
 use torii_introspect_postgres_sink::processor::IntrospectPgDb;
 use torii_introspect_sqlite_sink::processor::IntrospectSqliteDb;
-use torii_sqlite::{is_sqlite_memory_path, sqlite_connect_options};
+use torii_sqlite::is_sqlite_memory_path;
+
+fn sqlite_connect_options(path: &str) -> Result<SqliteConnectOptions> {
+    if path == ":memory:" || path == "sqlite::memory:" {
+        return SqliteConnectOptions::from_str("sqlite::memory:")
+            .map_err(|err| anyhow::anyhow!("Failed to parse sqlite URL: {err}"));
+    }
+
+    let options = if path.starts_with("sqlite:") {
+        SqliteConnectOptions::from_str(path)
+            .map_err(|err| anyhow::anyhow!("Failed to parse sqlite URL {path}: {err}"))?
+    } else {
+        SqliteConnectOptions::new().filename(path)
+    };
+
+    if path.starts_with("sqlite:") && path.contains("mode=") {
+        Ok(options)
+    } else {
+        Ok(options.create_if_missing(true))
+    }
+}
 
 #[tokio::main]
 async fn main() -> Result<()> {
@@ -59,12 +81,13 @@ async fn run_indexer(config: Config) -> Result<()> {
     tracing::info!("Storage backend: {:?}", backend);
     tracing::info!("Engine database URL: {}", engine_database_url);
     tracing::info!("Storage database URL: {}", storage_database_url);
+    tracing::info!("Database backend: {:?}", backend);
     tracing::info!(
         "Saved state handling: {}",
         if config.ignore_saved_state {
             "ignoring persisted extractor state"
         } else {
-            "failing closed if persisted extractor state exists"
+            "resuming from persisted extractor state when available"
         }
     );
     tracing::info!(
@@ -106,12 +129,6 @@ async fn run_indexer(config: Config) -> Result<()> {
         tokio::fs::create_dir_all(db_dir).await?;
     }
 
-    let engine_db = EngineDb::new(EngineDbConfig {
-        path: engine_database_url.clone(),
-    })
-    .await?;
-    ensure_no_saved_state(&engine_db, &contracts, config.ignore_saved_state).await?;
-
     match backend {
         StorageBackend::Postgres => {
             run_with_postgres(
@@ -160,17 +177,39 @@ async fn run_with_postgres(
     );
 
     let decoder = DojoDecoder::<PgStore<_>, _>::new(pool.clone(), provider);
+    let sink = IntrospectPgDb::new(pool.clone(), ());
     decoder.store.initialize().await?;
+    sink.initialize_introspect_pg_sink().await?;
     decoder.load_tables(&[]).await?;
+    sink.load_tables_no_commit(decoder.get_tables()?)?;
 
     let decoder: Arc<dyn torii::etl::Decoder> = Arc::new(decoder);
 
+    let ecs_sink = EcsSink::new(storage_database_url, config.max_db_connections).await?;
+    let ecs_grpc_service = ecs_sink.get_grpc_service_impl();
+
+    let reflection = tonic_reflection::server::Builder::configure()
+        .register_encoded_file_descriptor_set(torii::TORII_DESCRIPTOR_SET)
+        .register_encoded_file_descriptor_set(ECS_DESCRIPTOR_SET)
+        .build_v1()
+        .expect("failed to build ECS reflection service");
+
+    let grpc_router = tonic::transport::Server::builder()
+        .accept_http1(true)
+        .add_service(tonic_web::enable(WorldServer::new(
+            (*ecs_grpc_service).clone(),
+        )))
+        .add_service(tonic_web::enable(reflection));
+
     let mut torii_config = torii::ToriiConfig::builder()
         .port(config.port)
+        .with_grpc_router(grpc_router)
+        .with_custom_reflection(true)
         .engine_database_url(engine_database_url)
         .with_extractor(extractor)
         .add_decoder(decoder)
-        .add_sink_boxed(Box::new(IntrospectPgDb::new(pool.clone(), ())));
+        .add_sink_boxed(Box::new(ecs_sink))
+        .add_sink_boxed(Box::new(sink));
 
     let decoder_id = DecoderId::new("dojo-introspect");
     for contract in contracts {
@@ -181,6 +220,7 @@ async fn run_with_postgres(
     tracing::info!("Torii configured, starting ETL pipeline...");
     tracing::info!("gRPC service available on port {}", config.port);
     tracing::info!("  - torii.Torii (core subscriptions and metrics endpoint)");
+    tracing::info!("  - world.World (legacy ECS gRPC service)");
 
     torii::run(torii_config.build())
         .await
@@ -227,11 +267,30 @@ async fn run_with_sqlite(
 
     let decoder: Arc<dyn torii::etl::Decoder> = Arc::new(decoder);
 
+    let ecs_sink = EcsSink::new(storage_database_url, config.max_db_connections).await?;
+    let ecs_grpc_service = ecs_sink.get_grpc_service_impl();
+
+    let reflection = tonic_reflection::server::Builder::configure()
+        .register_encoded_file_descriptor_set(torii::TORII_DESCRIPTOR_SET)
+        .register_encoded_file_descriptor_set(ECS_DESCRIPTOR_SET)
+        .build_v1()
+        .expect("failed to build ECS reflection service");
+
+    let grpc_router = tonic::transport::Server::builder()
+        .accept_http1(true)
+        .add_service(tonic_web::enable(WorldServer::new(
+            (*ecs_grpc_service).clone(),
+        )))
+        .add_service(tonic_web::enable(reflection));
+
     let mut torii_config = torii::ToriiConfig::builder()
         .port(config.port)
+        .with_grpc_router(grpc_router)
+        .with_custom_reflection(true)
         .engine_database_url(engine_database_url)
         .with_extractor(extractor)
         .add_decoder(decoder)
+        .add_sink_boxed(Box::new(ecs_sink))
         .add_sink_boxed(Box::new(IntrospectSqliteDb::new(pool.clone(), ())));
 
     let decoder_id = DecoderId::new("dojo-introspect");
@@ -243,46 +302,10 @@ async fn run_with_sqlite(
     tracing::info!("Torii configured, starting ETL pipeline...");
     tracing::info!("gRPC service available on port {}", config.port);
     tracing::info!("  - torii.Torii (core subscriptions and metrics endpoint)");
+    tracing::info!("  - world.World (legacy ECS gRPC service)");
 
     torii::run(torii_config.build())
         .await
         .map_err(|e| anyhow::anyhow!("Torii error: {e}"))?;
     Ok(())
 }
-
-async fn ensure_no_saved_state(
-    engine_db: &EngineDb,
-    contracts: &[Felt],
-    ignore_saved_state: bool,
-) -> Result<()> {
-    if ignore_saved_state {
-        return Ok(());
-    }
-
-    let mut owners_with_saved_state = Vec::new();
-    for owner in contracts {
-        let state_key = format!("{owner:#x}");
-        if engine_db
-            .get_extractor_state("event", &state_key)
-            .await?
-            .is_some()
-        {
-            owners_with_saved_state.push(*owner);
-        }
-    }
-
-    if owners_with_saved_state.is_empty() {
-        return Ok(());
-    }
-
-    bail!(
-        "persisted extractor state exists for contract(s): {}. \
-         Historical schema bootstrap is not supported on this branch; \
-         re-run with --ignore-saved-state or use a fresh database",
-        owners_with_saved_state
-            .iter()
-            .map(|owner| format!("{owner:#x}"))
-            .collect::<Vec<_>>()
-            .join(", ")
-    )
-}