refactor: actor system for tap-agent

tap-agent/Cargo.toml

@@ -50,8 +50,10 @@ tracing-subscriber = { version = "0.3", features = [
 enum-as-inner = "0.6.0"
 ethers = "2.0.13"
 typetag = "0.2.14"
+ractor = "0.9.7"
 
 [dev-dependencies]
 ethers-signers = "2.0.8"
 tempfile = "3.8.0"
 wiremock = "0.5.19"
+futures = "0.3.30"

tap-agent/src/agent.rs

@@ -3,90 +3,122 @@
 
 use std::time::Duration;
 
-use alloy_sol_types::eip712_domain;
 use indexer_common::prelude::{
     escrow_accounts, indexer_allocations, DeploymentDetails, SubgraphClient,
 };
+use ractor::concurrency::JoinHandle;
+use ractor::{Actor, ActorRef};
 
-use crate::{
-    aggregator_endpoints, config, database, tap::sender_accounts_manager::SenderAccountsManager,
+use crate::agent::sender_accounts_manager::{
+    SenderAccountsManagerArgs, SenderAccountsManagerMessage,
 };
+use crate::config::{Cli, EscrowSubgraph, Ethereum, IndexerInfrastructure, NetworkSubgraph, Tap};
+use crate::{aggregator_endpoints, database, CONFIG, EIP_712_DOMAIN};
+use sender_accounts_manager::SenderAccountsManager;
 
-pub async fn start_agent(config: &'static config::Cli) -> SenderAccountsManager {
-    let pgpool = database::connect(&config.postgres).await;
+pub mod sender_account;
+pub mod sender_accounts_manager;
+pub mod sender_allocation;
+pub mod sender_fee_tracker;
+pub mod unaggregated_receipts;
+
+/// constant graph network used in subgraphs
+const GRAPH_NETWORK_ID: u64 = 1;
+
+pub async fn start_agent() -> (ActorRef<SenderAccountsManagerMessage>, JoinHandle<()>) {
+    let Cli {
+        ethereum: Ethereum { indexer_address },
+        indexer_infrastructure:
+            IndexerInfrastructure {
+                graph_node_query_endpoint,
+                graph_node_status_endpoint,
+                ..
+            },
+        postgres,
+        network_subgraph:
+            NetworkSubgraph {
+                network_subgraph_deployment,
+                network_subgraph_endpoint,
+                allocation_syncing_interval_ms,
+            },
+        escrow_subgraph:
+            EscrowSubgraph {
+                escrow_subgraph_deployment,
+                escrow_subgraph_endpoint,
+                escrow_syncing_interval_ms,
+            },
+        tap: Tap {
+            sender_aggregator_endpoints_file,
+            ..
+        },
+        ..
+    } = &*CONFIG;
+    let pgpool = database::connect(postgres).await;
 
     let http_client = reqwest::Client::new();
 
     let network_subgraph = Box::leak(Box::new(SubgraphClient::new(
         http_client.clone(),
-        config
-            .network_subgraph
-            .network_subgraph_deployment
+        network_subgraph_deployment
             .map(|deployment| {
                 DeploymentDetails::for_graph_node(
-                    &config.indexer_infrastructure.graph_node_status_endpoint,
-                    &config.indexer_infrastructure.graph_node_query_endpoint,
+                    graph_node_status_endpoint,
+                    graph_node_query_endpoint,
                     deployment,
                 )
             })
             .transpose()
             .expect("Failed to parse graph node query endpoint and network subgraph deployment"),
-        DeploymentDetails::for_query_url(&config.network_subgraph.network_subgraph_endpoint)
+        DeploymentDetails::for_query_url(network_subgraph_endpoint)
             .expect("Failed to parse network subgraph endpoint"),
     )));
 
     let indexer_allocations = indexer_allocations(
         network_subgraph,
-        config.ethereum.indexer_address,
-        1,
-        Duration::from_millis(config.network_subgraph.allocation_syncing_interval_ms),
+        *indexer_address,
+        GRAPH_NETWORK_ID,
+        Duration::from_millis(*allocation_syncing_interval_ms),
     );
 
     let escrow_subgraph = Box::leak(Box::new(SubgraphClient::new(
         http_client.clone(),
-        config
-            .escrow_subgraph
-            .escrow_subgraph_deployment
+        escrow_subgraph_deployment
             .map(|deployment| {
                 DeploymentDetails::for_graph_node(
-                    &config.indexer_infrastructure.graph_node_status_endpoint,
-                    &config.indexer_infrastructure.graph_node_query_endpoint,
+                    graph_node_status_endpoint,
+                    graph_node_query_endpoint,
                     deployment,
                 )
             })
             .transpose()
             .expect("Failed to parse graph node query endpoint and escrow subgraph deployment"),
-        DeploymentDetails::for_query_url(&config.escrow_subgraph.escrow_subgraph_endpoint)
+        DeploymentDetails::for_query_url(escrow_subgraph_endpoint)
             .expect("Failed to parse escrow subgraph endpoint"),
     )));
 
     let escrow_accounts = escrow_accounts(
         escrow_subgraph,
-        config.ethereum.indexer_address,
-        Duration::from_millis(config.escrow_subgraph.escrow_syncing_interval_ms),
+        *indexer_address,
+        Duration::from_millis(*escrow_syncing_interval_ms),
         false,
     );
 
     // TODO: replace with a proper implementation once the gateway registry contract is ready
-    let sender_aggregator_endpoints = aggregator_endpoints::load_aggregator_endpoints(
-        config.tap.sender_aggregator_endpoints_file.clone(),
-    );
+    let sender_aggregator_endpoints =
+        aggregator_endpoints::load_aggregator_endpoints(sender_aggregator_endpoints_file.clone());
 
-    let tap_eip712_domain_separator = eip712_domain! {
-        name: "TAP",
-        version: "1",
-        chain_id: config.receipts.receipts_verifier_chain_id,
-        verifying_contract: config.receipts.receipts_verifier_address,
-    };
-
-    SenderAccountsManager::new(
-        config,
+    let args = SenderAccountsManagerArgs {
+        config: &CONFIG,
+        domain_separator: EIP_712_DOMAIN.clone(),
         pgpool,
         indexer_allocations,
         escrow_accounts,
         escrow_subgraph,
-        tap_eip712_domain_separator,
         sender_aggregator_endpoints,
-    )
-    .await
+        prefix: None,
+    };
+
+    SenderAccountsManager::spawn(None, SenderAccountsManager, args)
+        .await
+        .expect("Failed to start sender accounts manager actor.")
 }