feat(indexer)!: migrate to REST api + UTXO streaming

Cargo.toml

@@ -238,9 +238,9 @@ syn = "2.0"
 tempfile = "3.3.0"
 thiserror = "1.0.59"
 time = "0.3.15"
-tokio = { version = "1.44.1", default-features = false }
-tokio-stream = "0.1.7"
-tokio-util = "0.7.10"
+tokio = { version = "1.47.1", default-features = false }
+tokio-stream = "0.1.17"
+tokio-util = "0.7.16"
 tonic = "0.13.1"
 tower = "0.5"
 tower-http = { version = "0.6", default-features = false }
@@ -255,6 +255,8 @@ webrtc = "0.9.0"
 zeroize = "1"
 webauthn-rs = { version = "0.5.1", features = ["default"] }
 webauthn-rs-proto = { version = "0.5.1", features = ["default"] }
+utoipa = "5.4.0"
+utoipa-swagger-ui = { version = "9.0.2" }
 uuid = "1.15.0"
 
 # Shutdown when panicking so we can see the error, specifically for the wallet

applications/tari_app_utilities/config_presets/d_indexer.toml

@@ -22,8 +22,8 @@
 # The relative path to store persistent data (default = "data/indexer")
 #data_dir = "data/indexer"
 
-# JSON-RPC listener address (default = "127.0.0.1:18300")
-#json_rpc_address = "127.0.0.1:18300"
+# API listener address (default = "127.0.0.1:18300")
+#api_listen_address = "127.0.0.1:18300"
 
 # HTTP UI listener address (default = "127.0.0.1:15000")
 #web_ui_address = "127.0.0.1:15000"

applications/tari_app_utilities/config_presets/e_wallet_daemon.toml

@@ -12,8 +12,8 @@
 # signaling_server_addr = "127.0.0.1:9100"
 
 [localnet.ootle_wallet_daemon]
-# Indexer JSON-RPC url (default: http://127.0.0.1:18200/json_rpc)
-# indexer_json_rpc_url = "http://127.0.0.1:18200/json_rpc"
+# Indexer API url (default: http://127.0.0.1:18200)
+# indexer_api_url = "http://127.0.0.1:18200"
 [igor.ootle_wallet_daemon]
-# Indexer JSON-RPC url for Igor.
-indexer_json_rpc_url = "http://18.217.22.26:12500/json_rpc"
+# Indexer API url for Igor.
+indexer_api_url = "http://18.217.22.26:12500"

applications/tari_indexer/Cargo.toml

@@ -20,7 +20,7 @@ tari_engine = { workspace = true }
 tari_engine_types = { workspace = true }
 tari_epoch_manager = { workspace = true }
 tari_epoch_oracles = { workspace = true, features = ["base_layer"] }
-tari_indexer_client = { workspace = true, default-features = false }
+tari_indexer_client = { workspace = true, default-features = false, features = ["utoipa"] }
 tari_indexer_lib = { workspace = true }
 tari_networking = { workspace = true }
 tari_ootle_app_utilities = { workspace = true }
@@ -37,10 +37,10 @@ tari_validator_node_rpc = { workspace = true }
 tari_ootle_wallet_sdk = { workspace = true }
 
 anyhow = { workspace = true }
-axum = { workspace = true }
+axum = { workspace = true, features = ["query", "matched-path"] }
 async-graphql = { workspace = true }
 async-graphql-axum = { workspace = true }
-axum-jrpc = { workspace = true, features = ["anyhow_error"] }
+bytes = { workspace = true }
 bincode = { workspace = true, features = ["serde"] }
 clap = { workspace = true, features = ["derive", "env"] }
 config = { workspace = true }
@@ -56,13 +56,15 @@ include_dir = { workspace = true, optional = true }
 libp2p = { workspace = true }
 log = { workspace = true, features = ["std"] }
 hex = { workspace = true }
+headers-accept = "0.2.1"
 log4rs = { workspace = true, features = [
     "rolling_file_appender",
     "compound_policy",
     "size_trigger",
     "fixed_window_roller",
 ] }
 mime_guess = { workspace = true }
+mediatype = "0.20.0"
 serde = { workspace = true, features = ["default", "derive"] }
 serde_json = { workspace = true }
 serde_with = { workspace = true, features = ["indexmap_2"] }
@@ -75,9 +77,12 @@ tokio = { workspace = true, features = [
     "sync",
     "rt-multi-thread",
 ] }
-tower-http = { workspace = true, features = ["default", "cors"] }
+tower-http = { workspace = true, features = ["default", "cors", "limit"] }
 url = { workspace = true, features = ["serde"] }
 cacache = { version = "13.1.0", default-features = false, features = ["mmap", "tokio-runtime"] }
+prost = { workspace = true }
+utoipa = { workspace = true }
+utoipa-swagger-ui = { workspace = true, features = ["axum"] }
 
 [package.metadata.cargo-machete]
 ignored = [

applications/tari_indexer/src/bootstrap.rs

@@ -33,6 +33,7 @@ use tari_common::{
 };
 use tari_consensus::consensus_constants::ConsensusConstants;
 use tari_crypto::tari_utilities::ByteArray;
+use tari_engine::transaction::TransactionProcessorConfig;
 use tari_engine_types::ToByteType;
 use tari_epoch_manager::service::{EpochManagerConfig, EpochManagerHandle};
 use tari_epoch_oracles::{
@@ -42,8 +43,10 @@ use tari_epoch_oracles::{
     store::EpochOracleStore,
     EpochOracle,
 };
+use tari_indexer_lib::substate_scanner::SubstateScanner;
 use tari_networking::{MessagingMode, NetworkingHandle, RelayCircuitLimits, RelayReservationLimits, SwarmConfig};
 use tari_ootle_app_utilities::{
+    claim_burn_proof_verifier::TariClaimBurnProofVerifier,
     common::verify_correct_network,
     configuration::convert_network_to_l1_network,
     epoch_oracle_config::EpochOracleType,
@@ -61,6 +64,7 @@ use tari_template_manager::{implementation::TemplateManager, interface::Template
 use tari_validator_node_rpc::client::TariValidatorNodeRpcClientFactory;
 
 use crate::{
+    dry_run::processor::DryRunTransactionProcessor,
     network_client::TariNetworkClient,
     network_state_sync,
     network_state_sync::NetworkWideStateSyncConfig,
@@ -71,6 +75,9 @@ use crate::{
         IndexerStoreWriteTransaction,
         SqliteIndexerStore,
     },
+    substate_file_cache::SubstateFileCache,
+    substate_manager::SubstateManager,
+    transaction_manager::TransactionManager,
     ApplicationConfig,
     IndexerEpochManagerSpec,
     Noop,
@@ -216,6 +223,29 @@ pub async fn spawn_services(
     )
     .spawn(shutdown.clone());
 
+    let substate_cache_dir = config.common.base_path.join("substate_cache");
+    let substate_cache = SubstateFileCache::new(substate_cache_dir)
+        .map_err(|e| ExitError::new(ExitCode::ConfigError, format!("Substate cache error: {}", e)))?;
+
+    let substate_scanner = SubstateScanner::new(
+        epoch_manager.clone(),
+        validator_node_client_factory.clone(),
+        substate_cache,
+    );
+
+    let substate_manager = SubstateManager::new(substate_scanner.clone(), store.clone());
+    let transaction_manager = TransactionManager::new(network_client.clone(), store.clone());
+
+    // dry run
+    let dry_run_transaction_processor = DryRunTransactionProcessor::new(
+        TransactionProcessorConfig::new(config.network)
+            .with_template_binary_max_size_bytes(consensus_constants.template_binary_max_size_bytes),
+        epoch_manager.clone(),
+        validator_node_client_factory.clone(),
+        template_manager.clone(),
+        TariClaimBurnProofVerifier::new(config.network, global_db.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.
     save_identities(config, &keypair)?;
@@ -224,11 +254,13 @@ pub async fn spawn_services(
         networking,
         epoch_manager,
         validator_node_client_factory,
-        network_client,
         store,
         global_db,
         template_manager,
         _template_manager_service: template_manager_service,
+        substate_manager,
+        transaction_manager,
+        dry_run_transaction_processor,
     })
 }
 
@@ -238,10 +270,13 @@ pub struct Services {
     pub epoch_manager: EpochManagerHandle<PeerAddress>,
     pub validator_node_client_factory: TariValidatorNodeRpcClientFactory,
     pub store: SqliteIndexerStore,
-    pub network_client: TariNetworkClient<EpochManagerHandle<PeerAddress>, TariValidatorNodeRpcClientFactory>,
     pub global_db: GlobalDb<SqliteGlobalDbAdapter<PeerAddress>>,
     pub template_manager: TemplateManager<PeerAddress>,
     pub _template_manager_service: TemplateManagerHandle,
+    pub substate_manager: SubstateManager,
+    pub transaction_manager:
+        TransactionManager<EpochManagerHandle<PeerAddress>, TariValidatorNodeRpcClientFactory, SqliteIndexerStore>,
+    pub dry_run_transaction_processor: DryRunTransactionProcessor,
 }
 
 fn ensure_directories_exist(config: &ApplicationConfig) -> io::Result<()> {

applications/tari_indexer/src/cli.rs

@@ -43,9 +43,9 @@ pub struct Cli {
     #[clap(long, short = 'i')]
     pub scanning_interval: Option<u64>,
 
-    /// Bind address for JSON-rpc server
-    #[clap(long, short = 'r', alias = "rpc-address")]
-    pub json_rpc_address: Option<SocketAddr>,
+    /// Bind address for API server
+    #[clap(long, short = 'r', alias = "api-address")]
+    pub api_listen_address: Option<SocketAddr>,
     #[clap(long, alias = "node-grpc", short = 'g', env = "TARI_INDEXER_MINOTARI_NODE_GRPC_URL")]
     pub epoch_oracle_minotari_node_grpc_url: Option<Url>,
     #[clap(long, alias = "oracle-config")]
@@ -61,9 +61,9 @@ pub struct Cli {
     pub reachability: Option<ReachabilityMode>,
     #[clap(long)]
     pub disable_mdns: bool,
-    /// Public address of the JSON RPC
-    #[clap(long, env = "TARI_INDEXER_WEB_UI_PUBLIC_JSON_RPC_URL")]
-    pub web_ui_public_json_rpc_url: Option<String>,
+    /// Public address of the API
+    #[clap(long, env = "TARI_INDEXER_WEB_UI_PUBLIC_API_URL")]
+    pub web_ui_public_api_url: Option<String>,
     /// Public address of the GraphQL API
     #[clap(long, env = "TARI_INDEXER_WEB_UI_PUBLIC_GRAPHQL_URL")]
     pub web_ui_public_graphql_url: Option<String>,
@@ -86,8 +86,8 @@ impl ConfigOverrideProvider for Cli {
         overrides.push(("indexer.override_from".to_string(), network.to_string()));
         overrides.push(("p2p.seeds.override_from".to_string(), network.to_string()));
 
-        if let Some(ref addr) = self.json_rpc_address {
-            overrides.push(("indexer.json_rpc_address".to_string(), addr.to_string()));
+        if let Some(ref addr) = self.api_listen_address {
+            overrides.push(("indexer.api_listen_address".to_string(), addr.to_string()));
         }
 
         if !self.peer_seeds.is_empty() {
@@ -99,11 +99,8 @@ impl ConfigOverrideProvider for Cli {
         if let Some(seconds) = self.scanning_interval {
             overrides.push(("indexer.scanning_interval".to_string(), seconds.to_string()));
         }
-        if let Some(ref json_rpc_url) = self.web_ui_public_json_rpc_url {
-            overrides.push((
-                "indexer.web_ui_public_json_rpc_url".to_string(),
-                json_rpc_url.to_string(),
-            ));
+        if let Some(ref api_url) = self.web_ui_public_api_url {
+            overrides.push(("indexer.web_ui_public_api_url".to_string(), api_url.to_string()));
         }
         if let Some(ref graphql_url) = self.web_ui_public_graphql_url {
             overrides.push(("indexer.web_ui_public_graphql_url".to_string(), graphql_url.to_string()));

applications/tari_indexer/src/config.rs

@@ -80,15 +80,15 @@ pub struct IndexerConfig {
     pub data_dir: PathBuf,
     /// The p2p configuration settings
     pub p2p: P2pConfig,
-    /// JSON-RPC address of the indexer application
-    pub json_rpc_address: Option<SocketAddr>,
+    /// Listening address for the indexer API server
+    pub api_listen_address: Option<SocketAddr>,
     /// GraphQL port of the indexer application
     pub graphql_address: Option<SocketAddr>,
     /// The address of the Web UI
     pub web_ui_address: Option<SocketAddr>,
-    /// The jrpc address where the UI should connect to the JSON RPC (it can be the same as the json_rpc_address, but
-    /// doesn't have to be), if this will be None, then the listen_addr will be used.
-    pub web_ui_public_json_rpc_url: Option<String>,
+    /// The publicly-accessible URL that the UI uses to connect to the API.
+    /// If this is None, then the api_listen_address will be used.
+    pub web_ui_public_api_url: Option<String>,
     /// The jrpc address where the UI should connect to the GraphQL API(it can be the same as the json_rpc_address, but
     /// doesn't have to be), if this will be None, then the listen_addr will be used.
     pub web_ui_public_graphql_url: Option<String>,
@@ -135,10 +135,10 @@ impl Default for IndexerConfig {
             tor_identity_file: PathBuf::from("indexer_tor_id.json"),
             data_dir: PathBuf::from("data/indexer"),
             p2p: P2pConfig::default(),
-            json_rpc_address: Some("127.0.0.1:18300".parse().unwrap()),
+            api_listen_address: Some("127.0.0.1:18300".parse().unwrap()),
             graphql_address: Some("127.0.0.1:18301".parse().unwrap()),
             web_ui_address: Some("127.0.0.1:15000".parse().unwrap()),
-            web_ui_public_json_rpc_url: None,
+            web_ui_public_api_url: None,
             web_ui_public_graphql_url: None,
             block_scanning_interval: Duration::from_secs(10),
             state_scanning_interval: Duration::from_secs(60),

applications/tari_indexer/src/dry_run/processor.rs

@@ -49,6 +49,7 @@ use crate::dry_run::error::DryRunTransactionProcessorError;
 
 const LOG_TARGET: &str = "tari::indexer::dry_run_transaction_processor";
 
+#[derive(Clone)]
 pub struct DryRunTransactionProcessor {
     config: TransactionProcessorConfig,
     epoch_manager: EpochManagerHandle<PeerAddress>,

applications/tari_indexer/src/http_ui/server.rs

@@ -37,13 +37,13 @@ const LOG_TARGET: &str = "tari::indexer::web_ui::server";
 
 pub async fn run_http_ui_server(
     address: SocketAddr,
-    json_rpc_address: Url,
+    api_address: Url,
     graphql_address: Option<Url>,
 ) -> Result<(), anyhow::Error> {
-    let json_rpc_address = Arc::new(json_rpc_address);
+    let api_address = Arc::new(api_address);
 
     let router = Router::new()
-        .route("/json_rpc_address", get(|| async move { json_rpc_address.to_string() }))
+        .route("/rest_api_address", get(|| async move { api_address.to_string() }))
         .route(
             "/graphql_address",
             get(|| async move { graphql_address.map(|a| a.to_string()).unwrap_or_default() }),