feat(indexer)!: implement state sync and utxo scan api

Cargo.toml

@@ -160,7 +160,7 @@ axum-jrpc = "0.3.2"
 base64 = "0.20.0"
 bigdecimal = "0.4.1"
 bincode = "2.0.0-rc.3"
-bitflags = "2.4"
+bitflags = "2.9.2"
 blake2 = "0.10.6"
 borsh = { version = "1.5", default-features = false }
 bytes = "1.10.0"
@@ -228,6 +228,7 @@ rocksdb = "0.23.0"
 semver = "1.0"
 serde = { version = "1.0", default-features = false }
 serde_json = "1.0"
+serde_with = "3.14.0"
 sha2 = "0.10.8"
 smallvec = "2.0.0-alpha.11"
 std-semaphore = "0.1.0"

applications/tari_indexer/Cargo.toml

@@ -23,7 +23,7 @@ tari_ootle_storage = { workspace = true }
 tari_ootle_storage_sqlite = { workspace = true }
 tari_epoch_manager = { workspace = true }
 tari_engine_types = { workspace = true }
-tari_indexer_client = { workspace = true }
+tari_indexer_client = { workspace = true, default-features = false }
 tari_indexer_lib = { workspace = true }
 tari_template_lib = { workspace = true }
 tari_template_manager = { workspace = true }
@@ -44,7 +44,7 @@ config = { workspace = true }
 diesel = { workspace = true, default-features = false, features = [
     "sqlite",
     "returning_clauses_for_sqlite_3_35",
-    "chrono",
+    "time",
 ] }
 diesel_migrations = { workspace = true }
 futures = { workspace = true }
@@ -61,6 +61,7 @@ log4rs = { workspace = true, features = [
 mime_guess = { workspace = true }
 serde = { workspace = true, features = ["default", "derive"] }
 serde_json = { workspace = true }
+serde_with = { workspace = true, features = ["indexmap"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = [
     "default",

applications/tari_indexer/src/bootstrap.rs

@@ -20,9 +20,9 @@
 //   WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 //   USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use std::{fs, io, str::FromStr};
+use std::{fs, io, str::FromStr, sync::Arc};
 
-use anyhow::Context;
+use anyhow::{anyhow, Context};
 use libp2p::identity;
 use log::warn;
 use minotari_app_utilities::identity_management;
@@ -51,7 +51,7 @@ use tari_ootle_app_utilities::{
     seed_peer::SeedPeer,
     template_download_queue::TemplateDownloadQueue,
 };
-use tari_ootle_common_types::PeerAddress;
+use tari_ootle_common_types::{optional::Optional, Network, PeerAddress};
 use tari_ootle_p2p::TariMessagingSpec;
 use tari_ootle_storage::global::GlobalDb;
 use tari_ootle_storage_sqlite::global::SqliteGlobalDbAdapter;
@@ -62,7 +62,12 @@ use tari_validator_node_rpc::client::TariValidatorNodeRpcClientFactory;
 
 use crate::{
     network_client::TariNetworkClient,
-    storage_sqlite::store_factory::SqliteIndexerStore,
+    network_state_sync,
+    network_state_sync::NetworkWideStateSyncConfig,
+    storage_sqlite::{
+        models::Key,
+        store_factory::{IndexerStore, IndexerStoreReadTransaction, IndexerStoreWriteTransaction, SqliteIndexerStore},
+    },
     ApplicationConfig,
     IndexerEpochManagerSpec,
     Noop,
@@ -151,6 +156,7 @@ pub async fn spawn_services(
 
     // Connect to substate db
     let store = SqliteIndexerStore::try_create(config.indexer.state_db_path())?;
+    check_store(config, &store)?;
 
     // Epoch event oracle
     let epoch_event_oracle = create_epoch_oracle(config, global_db.clone(), &consensus_constants).await?;
@@ -196,6 +202,17 @@ pub async fn spawn_services(
         template_queue_receiver,
         shutdown.clone(),
     );
+    network_state_sync::NetworkWideStateSync::new(
+        epoch_manager.clone(),
+        networking.clone(),
+        store.clone(),
+        template_manager_service.clone(),
+        NetworkWideStateSyncConfig {
+            event_filters: Arc::from(config.indexer.event_filters.clone()),
+            ..Default::default()
+        },
+    )
+    .spawn(shutdown.clone());
 
     // Save final node identity after comms has initialized. This is required because the public_address can be
     // changed by comms during initialization when using tor.
@@ -209,7 +226,7 @@ pub async fn spawn_services(
         store,
         global_db,
         template_manager,
-        template_manager_service,
+        _template_manager_service: template_manager_service,
     })
 }
 
@@ -222,7 +239,7 @@ pub struct Services {
     pub network_client: TariNetworkClient<EpochManagerHandle<PeerAddress>, TariValidatorNodeRpcClientFactory>,
     pub global_db: GlobalDb<SqliteGlobalDbAdapter<PeerAddress>>,
     pub template_manager: TemplateManager<PeerAddress>,
-    pub template_manager_service: TemplateManagerHandle,
+    pub _template_manager_service: TemplateManagerHandle,
 }
 
 fn ensure_directories_exist(config: &ApplicationConfig) -> io::Result<()> {
@@ -319,3 +336,25 @@ async fn create_hybrid_epoch_oracle<TStore: EpochOracleStore + Clone + Send + 's
     let configured_oracle = ConfiguredEpochOracle::with_custom_ticker(oracle_config, store, ticker);
     Ok(HybridEpochOracle::new(configured_oracle, base_layer_oracle, trigger))
 }
+
+fn check_store<TStore: IndexerStore>(config: &ApplicationConfig, store: &TStore) -> anyhow::Result<()> {
+    store.with_write_tx(|tx| {
+        match tx.key_value_get_value::<_, Network>(Key::Network).optional()? {
+            Some(network) => {
+                if network != config.network {
+                    return Err(anyhow!(
+                        "The network in the database ({}) does not match the configured network ({})",
+                        network,
+                        config.network
+                    ));
+                }
+                Ok(())
+            },
+            None => {
+                // If the network is not set, we can assume this is a new store and we can set it
+                tx.key_value_set(Key::Network, config.network)
+                    .map_err(|e| anyhow!("Failed to set network in the store: {}", e))
+            },
+        }
+    })
+}

applications/tari_indexer/src/config.rs

@@ -42,6 +42,8 @@ use tari_ootle_app_utilities::{
 use tari_ootle_common_types::Network;
 use tari_template_manager::implementation::TemplateConfig;
 
+use crate::network_state_sync::EventFilter;
+
 #[derive(Debug, Clone)]
 pub struct ApplicationConfig {
     pub common: CommonConfig,
@@ -102,7 +104,7 @@ pub struct IndexerConfig {
     /// The burnt utxos sidechain id
     pub burnt_utxo_sidechain_id: Option<RistrettoPublicKey>,
     /// The event filtering configuration
-    pub event_filters: Vec<EventFilterConfig>,
+    pub event_filters: Vec<EventFilter>,
 }
 
 impl IndexerConfig {
@@ -151,11 +153,3 @@ impl SubConfigPath for IndexerConfig {
         "indexer"
     }
 }
-
-#[derive(Default, Debug, Serialize, Deserialize, Clone)]
-pub struct EventFilterConfig {
-    pub topic: Option<String>,
-    pub entity_id: Option<String>,
-    pub substate_id: Option<String>,
-    pub template_address: Option<String>,
-}

applications/tari_indexer/src/event_manager.rs

@@ -20,71 +20,24 @@
 //  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 //  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use std::{collections::BTreeMap, str::FromStr, sync::Arc};
+use std::{collections::BTreeMap, str::FromStr};
 
 use log::*;
-use tari_crypto::tari_utilities::message_format::MessageFormat;
 use tari_engine_types::{events::Event, substate::SubstateId};
-use tari_epoch_manager::service::EpochManagerHandle;
-use tari_indexer_lib::substate_scanner::SubstateScanner;
-use tari_ootle_app_utilities::substate_file_cache::SubstateFileCache;
-use tari_ootle_common_types::PeerAddress;
-use tari_template_lib::{
-    models::Metadata,
-    types::{Hash, TemplateAddress},
-};
-use tari_transaction::TransactionId;
-use tari_validator_node_rpc::client::TariValidatorNodeRpcClientFactory;
+use tari_template_lib::{models::Metadata, types::Hash};
 
-use crate::storage_sqlite::{
-    models::events::NewEvent,
-    store_factory::{IndexerStore, IndexerStoreReadTransaction, IndexerStoreWriteTransaction, SqliteIndexerStore},
-};
+use crate::storage_sqlite::store_factory::{IndexerStore, IndexerStoreReadTransaction, SqliteIndexerStore};
 
 const LOG_TARGET: &str = "tari::indexer::event_manager";
 
+#[derive(Debug, Clone)]
 pub struct EventManager {
     substate_store: SqliteIndexerStore,
-    _substate_scanner:
-        Arc<SubstateScanner<EpochManagerHandle<PeerAddress>, TariValidatorNodeRpcClientFactory, SubstateFileCache>>,
 }
 
 impl EventManager {
-    pub fn new(
-        substate_store: SqliteIndexerStore,
-        substate_scanner: Arc<
-            SubstateScanner<EpochManagerHandle<PeerAddress>, TariValidatorNodeRpcClientFactory, SubstateFileCache>,
-        >,
-    ) -> Self {
-        Self {
-            substate_store,
-            _substate_scanner: substate_scanner,
-        }
-    }
-
-    pub fn save_event_to_db(
-        &self,
-        substate_id: &SubstateId,
-        template_address: TemplateAddress,
-        tx_hash: TransactionId,
-        topic: String,
-        payload: &Metadata,
-        version: u64,
-        timestamp: u64,
-    ) -> Result<(), anyhow::Error> {
-        self.substate_store.with_write_tx(|tx| {
-            let new_event = NewEvent {
-                substate_id: Some(substate_id.to_string()),
-                template_address: template_address.to_string(),
-                tx_hash: tx_hash.to_string(),
-                topic,
-                payload: payload.to_json().expect("Failed to convert to JSON"),
-                version: version as i32,
-                timestamp: timestamp as i64,
-            };
-            tx.save_event(new_event)
-        })?;
-        Ok(())
+    pub fn new(substate_store: SqliteIndexerStore) -> Self {
+        Self { substate_store }
     }
 
     pub async fn get_events_from_db(