feat(pathfinder): add pathfinding utilities and SQLite integration

Cargo.toml

@@ -20,6 +20,8 @@ members = [
   "crates/testing",
   "crates/postgres",
   "crates/sqlite",
+  "crates/pathfinder",
+  "crates/starknet",
   "bins/torii-erc20",
   "bins/torii-erc20-synth",
   "bins/torii-introspect-bin",
@@ -122,6 +124,7 @@ dojo-introspect = { git = "https://github.com/dojoengine/dojo-introspect", rev =
 # Utils
 resolve-path = "0.1.0"
 itertools = "0.14.0"
+bincode = { version = "2.0.0" }
 
 # Torii core
 torii.path = "."
@@ -146,6 +149,8 @@ torii-introspect-sqlite-sink.path = "./crates/introspect-sqlite-sink"
 torii-test-utils.path = "crates/testing"
 torii-postgres.path = "crates/postgres"
 torii-sqlite.path = "crates/sqlite"
+torii-pathfinder.path = "crates/pathfinder"
+torii-starknet.path = "crates/starknet"
 
 
 [lib]
@@ -198,7 +203,11 @@ serde.workspace = true
 sqlx.workspace = true
 starknet.workspace = true
 tokio-stream.workspace = true
-tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "tls12", "ring"] }
+tokio-rustls = { version = "0.26", default-features = false, features = [
+  "logging",
+  "tls12",
+  "ring",
+] }
 tokio-util.workspace = true
 tokio.workspace = true
 tonic-reflection.workspace = true

bins/torii-arcade/Cargo.toml

@@ -23,6 +23,7 @@ torii-introspect-postgres-sink = { path = "../../crates/introspect-postgres-sink
 torii-introspect-sqlite-sink = { path = "../../crates/introspect-sqlite-sink" }
 torii-runtime-common.workspace = true
 torii-sqlite.workspace = true
+torii-pathfinder = { workspace = true, features = ["etl"] }
 
 tokio = { version = "1", features = ["full"] }
 tracing = "0.1"
@@ -31,7 +32,11 @@ starknet = "0.17"
 url = "2.5"
 clap = { version = "4.5", features = ["derive", "env"] }
 anyhow = "1.0"
-sqlx = { version = "0.8", features = ["postgres", "sqlite", "runtime-tokio-rustls"] }
+sqlx = { version = "0.8", features = [
+    "postgres",
+    "sqlite",
+    "runtime-tokio-rustls",
+] }
 tonic.workspace = true
 tonic-reflection.workspace = true
 tonic-web.workspace = true

bins/torii-arcade/src/config.rs

@@ -204,6 +204,9 @@ pub struct Config {
 
     #[arg(long, value_delimiter = ',')]
     pub historical: Vec<String>,
+
+    #[arg(long)]
+    pub pathfinder_path: Option<String>,
 }
 
 impl Config {
@@ -358,6 +361,10 @@ impl Config {
             _ => bail!("--tls-cert and --tls-key must be provided together"),
         }
     }
+
+    pub fn pathfinder_path(&self) -> Option<PathBuf> {
+        self.pathfinder_path.as_ref().map(PathBuf::from)
+    }
 }
 
 fn is_postgres_url(url: &str) -> bool {

bins/torii-arcade/src/main.rs

@@ -1,6 +1,6 @@
 mod config;
 
-use anyhow::Result;
+use anyhow::{Error, Result};
 use clap::Parser;
 use config::{Config, MetadataMode};
 use sqlx::postgres::PgPoolOptions;
@@ -52,6 +52,7 @@ use torii_erc721::{
 };
 use torii_introspect_postgres_sink::processor::IntrospectPgDb;
 use torii_introspect_sqlite_sink::processor::IntrospectSqliteDb;
+use torii_pathfinder::extractor::PathfinderCombinedExtractor;
 use torii_runtime_common::database::{
     validate_uniform_backends, DatabaseBackend, DEFAULT_SQLITE_MAX_CONNECTIONS,
 };
@@ -392,7 +393,7 @@ async fn run_indexer(config: Config) -> Result<()> {
         &erc721_addresses,
         &erc1155_addresses,
         &config,
-    );
+    )?;
 
     let (dojo_decoder, introspect_sink): (
         Arc<dyn torii::etl::Decoder>,
@@ -738,7 +739,7 @@ fn build_extractor(
     erc721_addresses: &[Felt],
     erc1155_addresses: &[Felt],
     config: &Config,
-) -> Box<dyn Extractor> {
+) -> Result<Box<dyn Extractor>> {
     let to_block = config.to_block.unwrap_or(u64::MAX);
     let mut contracts = Vec::new();
     let mut seen = HashSet::new();
@@ -771,18 +772,39 @@ fn build_extractor(
         config.from_block,
         to_block,
     );
-
-    Box::new(EventExtractor::new(
+    let event_extractor = EventExtractor::new(
         provider,
         EventExtractorConfig {
-            contracts,
+            contracts: contracts.clone(),
             chunk_size: config.event_chunk_size,
             block_batch_size: config.event_block_batch_size,
             retry_policy: RetryPolicy::default(),
             ignore_saved_state: config.ignore_saved_state,
             rpc_parallelism: config.rpc_parallelism,
         },
-    ))
+    );
+    #[allow(clippy::single_match_else)]
+    match config.pathfinder_path() {
+        Some(path) => {
+            tracing::info!(
+                "Using Pathfinder for event extraction with endpoint: {}",
+                path.display()
+            );
+            PathfinderCombinedExtractor::new(
+                path,
+                config.event_block_batch_size,
+                config.from_block,
+                config.to_block.unwrap_or(u64::MAX),
+                event_extractor,
+            )
+            .map(|e| Box::new(e) as Box<dyn Extractor>)
+            .map_err(Error::new)
+        }
+        None => {
+            tracing::info!("Using direct RPC for event extraction");
+            Ok(Box::new(event_extractor))
+        }
+    }
 }
 
 fn append_unique_contract_configs(

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

@@ -354,6 +354,10 @@ impl SyntheticIntrospectExtractor {
 
 #[async_trait]
 impl Extractor for SyntheticIntrospectExtractor {
+    fn set_start_block(&mut self, start_block: u64) {
+        self.current_block = start_block.max(self.current_block);
+    }
+
     async fn extract(
         &mut self,
         cursor: Option<String>,

crates/pathfinder/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "torii-pathfinder"
+version = "0.1.0"
+edition = "2021"
+description = "Pathfinding utilities for Torii"
+
+[dependencies]
+starknet.workspace = true
+rusqlite = { version = "0.32.1", features = ["bundled"] }
+zstd = { version = "0.13.2", features = ["experimental"] }
+serde.workspace = true
+bincode = { workspace = true, features = ["serde"] }
+thiserror.workspace = true
+rand = "0.8.5"
+torii-starknet.workspace = true
+
+# Etl dependencies
+torii = { workspace = true, optional = true }
+async-trait = { workspace = true, optional = true }
+anyhow = { workspace = true, optional = true }
+
+[lints]
+workspace = true
+
+[features]
+default = ["etl"]
+etl = ["torii", "async-trait", "anyhow", "torii-starknet/starknet"]

crates/pathfinder/src/decoding.rs

@@ -0,0 +1,165 @@
+use crate::sqlite::BlockEventsRow;
+use crate::{PFError, PFResult};
+use serde::{Deserialize, Serialize};
+use std::cell::RefCell;
+use std::fmt::{Formatter, Result as FmtResult};
+use std::io::Result as IoResult;
+use std::sync::LazyLock;
+use torii_starknet::event::Event;
+use torii_starknet::Felt;
+use zstd::bulk::Decompressor;
+
+// Taken from pathfinder-common but with some optimizations
+
+const MAX_EVENTS_UNCOMPRESSED_SIZE: usize = 128usize * 1024 * 1024;
+
+static ZSTD_EVENTS_DECODER_DICTIONARY: LazyLock<zstd::dict::DecoderDictionary<'static>> =
+    LazyLock::new(|| zstd::dict::DecoderDictionary::new(include_bytes!("assets/events.zdict")));
+
+thread_local! {
+    static EVENT_DECOMPRESSOR: RefCell<Decompressor<'static>> = RefCell::new(
+        Decompressor::with_prepared_dictionary(
+            &ZSTD_EVENTS_DECODER_DICTIONARY
+        ).expect("failed to create decompressor")
+    );
+}
+
+/// Minimally encoded Felt value.
+#[repr(transparent)]
+#[derive(Clone, Debug, PartialEq, Eq, Default)]
+pub struct MinimalFelt([u8; 32]);
+
+impl serde::Serialize for MinimalFelt {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let bytes = self.0;
+        let zeros = bytes.iter().take_while(|&&x| x == 0).count();
+        bytes[zeros..].serialize(serializer)
+    }
+}
+
+impl<'de> serde::Deserialize<'de> for MinimalFelt {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct Visitor;
+
+        impl<'de> serde::de::Visitor<'de> for Visitor {
+            type Value = MinimalFelt;
+
+            fn expecting(&self, formatter: &mut Formatter<'_>) -> FmtResult {
+                formatter.write_str("a sequence")
+            }
+
+            fn visit_seq<B>(self, mut seq: B) -> Result<Self::Value, B::Error>
+            where
+                B: serde::de::SeqAccess<'de>,
+            {
+                let len = seq.size_hint().unwrap();
+                let mut bytes = [0; 32];
+                let num_zeros = bytes.len() - len;
+                let mut i = num_zeros;
+                while let Some(value) = seq.next_element()? {
+                    bytes[i] = value;
+                    i += 1;
+                }
+                Ok(MinimalFelt(bytes))
+            }
+        }
+
+        deserializer.deserialize_seq(Visitor)
+    }
+}
+
+impl MinimalFelt {
+    #[allow(unsafe_code)]
+    fn into_felt_vec(v: Vec<Self>) -> Vec<Felt> {
+        // Safe: both repr(transparent) over [u8; 32]
+        unsafe {
+            let mut v = std::mem::ManuallyDrop::new(v);
+            Vec::from_raw_parts(v.as_mut_ptr().cast::<Felt>(), v.len(), v.capacity())
+        }
+    }
+}
+
+impl From<MinimalFelt> for Felt {
+    fn from(value: MinimalFelt) -> Self {
+        value.0.into()
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub enum EventsForBlock {
+    V0 { events: Vec<Vec<EncodedEvent>> },
+}
+
+impl EventsForBlock {
+    pub fn events(self) -> Vec<Vec<Event>> {
+        match self {
+            EventsForBlock::V0 { events } => events
+                .into_iter()
+                .map(|evs| evs.into_iter().map(Event::from).collect())
+                .collect(),
+        }
+    }
+}
+
+pub(crate) fn decompress_events(input: &[u8]) -> IoResult<Vec<u8>> {
+    EVENT_DECOMPRESSOR.with(|d| {
+        d.borrow_mut()
+            .decompress(input, MAX_EVENTS_UNCOMPRESSED_SIZE)
+    })
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct EncodedEvent {
+    pub data: Vec<MinimalFelt>,
+    pub from_address: MinimalFelt,
+    pub keys: Vec<MinimalFelt>,
+}
+
+impl From<EncodedEvent> for Event {
+    fn from(value: EncodedEvent) -> Self {
+        Self {
+            data: MinimalFelt::into_felt_vec(value.data),
+            from_address: value.from_address.into(),
+            keys: MinimalFelt::into_felt_vec(value.keys),
+        }
+    }
+}
+
+impl BlockEventsRow {
+    pub fn decompress_events(&self) -> PFResult<Vec<Vec<Event>>> {
+        match self.events {
+            None => Ok(Vec::new()),
+            Some(ref events) => {
+                let events = decompress_events(events)?;
+                let events: EventsForBlock =
+                    bincode::serde::decode_from_slice(&events, bincode::config::standard())?.0;
+                Ok(events.events())
+            }
+        }
+    }
+}
+
+impl TryFrom<BlockEventsRow> for BlockEvents {
+    type Error = PFError;
+
+    fn try_from(value: BlockEventsRow) -> Result<Self, Self::Error> {
+        Ok(Self {
+            block_number: value.block_number,
+            transactions: value.decompress_events()?,
+        })
+    }
+}
+
+#[derive(Debug)]
+pub struct BlockEvents {
+    pub block_number: u64,
+    pub transactions: Vec<Vec<Event>>,
+}