feat: use network subgraph for v2 escrow accounts

Author: suchapalaver

crates/monitor/src/escrow_accounts.rs

@@ -9,9 +9,6 @@ use std::{
 
 use anyhow::anyhow;
 use indexer_query::escrow_account::{self, EscrowAccountQuery};
-use indexer_query::escrow_account_v2::{
-    self as escrow_account_v2, EscrowAccountQuery as EscrowAccountQueryV2,
-};
 use thegraph_core::alloy::primitives::{Address, U256};
 use thiserror::Error;
 use tokio::sync::watch::Receiver;
@@ -91,6 +88,12 @@ impl EscrowAccounts {
 
 pub type EscrowAccountsWatcher = Receiver<EscrowAccounts>;
 
+pub fn empty_escrow_accounts_watcher() -> EscrowAccountsWatcher {
+    let (_, receiver) =
+        tokio::sync::watch::channel(EscrowAccounts::new(HashMap::new(), HashMap::new()));
+    receiver
+}
+
 pub async fn escrow_accounts_v1(
     escrow_subgraph: &'static SubgraphClient,
     indexer_address: Address,
@@ -120,22 +123,16 @@ async fn get_escrow_accounts_v2(
     indexer_address: Address,
     reject_thawing_signers: bool,
 ) -> anyhow::Result<EscrowAccounts> {
-    // V2 TAP receipts use different field names (payer/service_provider) but the underlying
-    // escrow account model is identical to V1. Both V1 and V2 receipts reference the same
-    // sender addresses and the same escrow relationships.
-    //
-    // The separation of V1/V2 escrow account watchers allows for potential future differences
-    // in escrow models, but currently both query the same subgraph data with identical logic.
-    //
-    // V2 receipt flow:
-    // 1. V2 receipt contains payer address (equivalent to V1 sender)
-    // 2. Receipt is signed by a signer authorized by the payer
-    // 3. Escrow accounts map: signer -> payer (sender) -> balance
-    // 4. Service provider (indexer) receives payments from payer's escrow
+    // Query V2 escrow accounts from the network subgraph which tracks PaymentsEscrow
+    // and GraphTallyCollector contract events.
+
+    use indexer_query::network_escrow_account_v2::{
+        self as network_escrow_account_v2, NetworkEscrowAccountQueryV2,
+    };
 
     let response = escrow_subgraph
-        .query::<EscrowAccountQueryV2, _>(escrow_account_v2::Variables {
-            indexer: format!("{:x?}", indexer_address),
+        .query::<NetworkEscrowAccountQueryV2, _>(network_escrow_account_v2::Variables {
+            receiver: format!("{:x?}", indexer_address),
             thaw_end_timestamp: if reject_thawing_signers {
                 U256::ZERO.to_string()
             } else {
@@ -146,10 +143,23 @@ async fn get_escrow_accounts_v2(
 
     let response = response?;
 
-    tracing::trace!("V2 Escrow accounts response: {:?}", response);
+    tracing::trace!("Network V2 Escrow accounts response: {:?}", response);
+
+    // V2 TAP receipts use different field names (payer/service_provider) but the underlying
+    // escrow account model is identical to V1. Both V1 and V2 receipts reference the same
+    // sender addresses and the same escrow relationships.
+    //
+    // V1 queries the TAP subgraph while V2 queries the network subgraph, but both return
+    // the same escrow account structure for processing.
+    //
+    // V2 receipt flow:
+    // 1. V2 receipt contains payer address (equivalent to V1 sender)
+    // 2. Receipt is signed by a signer authorized by the payer
+    // 3. Escrow accounts map: signer -> payer (sender) -> balance
+    // 4. Service provider (indexer) receives payments from payer's escrow
 
     let senders_balances: HashMap<Address, U256> = response
-        .escrow_accounts
+        .payments_escrow_accounts
         .iter()
         .map(|account| {
             let balance = U256::checked_sub(
@@ -170,7 +180,7 @@ async fn get_escrow_accounts_v2(
         .collect::<Result<HashMap<_, _>, anyhow::Error>>()?;
 
     let senders_to_signers = response
-        .escrow_accounts
+        .payments_escrow_accounts
         .into_iter()
         .map(|account| {
             let payer = Address::from_str(&account.payer.id)?;

crates/monitor/src/lib.rs

@@ -16,8 +16,8 @@ pub use crate::{
     deployment_to_allocation::{deployment_to_allocation, DeploymentToAllocationWatcher},
     dispute_manager::{dispute_manager, DisputeManagerWatcher},
     escrow_accounts::{
-        escrow_accounts_v1, escrow_accounts_v2, EscrowAccounts, EscrowAccountsError,
-        EscrowAccountsWatcher,
+        empty_escrow_accounts_watcher, escrow_accounts_v1, escrow_accounts_v2, EscrowAccounts,
+        EscrowAccountsError, EscrowAccountsWatcher,
     },
     horizon_detection::is_horizon_active,
 };

crates/service/src/service.rs

@@ -78,18 +78,7 @@ pub async fn run() -> anyhow::Result<()> {
     )
     .await;
 
-    let escrow_subgraph_v2 = if let Some(ref escrow_v2_config) = config.subgraphs.escrow_v2 {
-        Some(
-            create_subgraph_client(
-                http_client.clone(),
-                &config.graph_node,
-                &escrow_v2_config.config,
-            )
-            .await,
-        )
-    } else {
-        None
-    };
+    // V2 escrow accounts are in the network subgraph, not a separate escrow_v2 subgraph
 
     // Establish Database connection necessary for serving indexer management
     // requests with defined schema
@@ -113,158 +102,130 @@ pub async fn run() -> anyhow::Result<()> {
 
     // Capture individual fields needed for DIPS before they get moved
     let escrow_v1_query_url_for_dips = config.subgraphs.escrow.config.query_url.clone();
-    let escrow_v2_query_url_for_dips = config
-        .subgraphs
-        .escrow_v2
-        .as_ref()
-        .map(|c| c.config.query_url.clone());
-
-    // Configure router with escrow watchers based on Horizon mode
-    use indexer_config::HorizonMode;
-    let router = match config.horizon.mode {
-        HorizonMode::Legacy => {
-            tracing::info!("Horizon mode: Legacy - using escrow accounts v1 only");
-            // Only create v1 watcher for legacy mode
-            let escrow_subgraph_v1 = create_subgraph_client(
-                http_client.clone(),
-                &config.graph_node,
-                &config.subgraphs.escrow.config,
-            )
-            .await;
-
-            let v1_watcher = indexer_monitor::escrow_accounts_v1(
-                escrow_subgraph_v1,
-                indexer_address,
-                config.subgraphs.escrow.config.syncing_interval_secs,
-                true, // Reject thawing signers eagerly
-            )
-            .await
-            .expect("Error creating escrow_accounts_v1 channel");
-
-            ServiceRouter::builder()
-                .database(database.clone())
-                .domain_separator(domain_separator.clone())
-                .graph_node(config.graph_node)
-                .http_client(http_client)
-                .release(release)
-                .indexer(config.indexer)
-                .service(config.service)
-                .blockchain(config.blockchain)
-                .timestamp_buffer_secs(config.tap.rav_request.timestamp_buffer_secs)
-                .network_subgraph(network_subgraph, config.subgraphs.network)
-                .escrow_accounts_v1(v1_watcher)
-                .build()
+    // V2 escrow accounts are in the network subgraph
+    let escrow_v2_query_url_for_dips = Some(config.subgraphs.network.config.query_url.clone());
+
+    // Determine if we should check for Horizon contracts and potentially enable hybrid mode:
+    // - If horizon.enabled = false: Pure legacy mode, no Horizon detection
+    // - If horizon.enabled = true: Check if Horizon contracts are active in the network
+    let is_horizon_active = if config.horizon.enabled {
+        tracing::info!("Horizon migration support enabled - checking if Horizon contracts are active in the network");
+        match indexer_monitor::is_horizon_active(network_subgraph).await {
+            Ok(active) => {
+                if active {
+                    tracing::info!("Horizon contracts detected in network subgraph - enabling hybrid migration mode");
+                    tracing::info!("Mode: Accept new V2 receipts only, continue processing existing V1 receipts for RAVs");
+                } else {
+                    tracing::info!("Horizon contracts not yet active in network subgraph - remaining in legacy mode");
+                }
+                active
+            }
+            Err(e) => {
+                tracing::warn!(
+                    "Failed to detect Horizon contracts: {}. Remaining in legacy mode.",
+                    e
+                );
+                false
+            }
         }
-        HorizonMode::Transition => {
-            tracing::info!("Horizon mode: Transition - using both escrow accounts v1 and v2");
-            // Create both watchers for transition mode using separate subgraph clients
-            let escrow_subgraph_v1 = create_subgraph_client(
-                http_client.clone(),
-                &config.graph_node,
-                &config.subgraphs.escrow.config,
-            )
-            .await;
+    } else {
+        tracing::info!(
+            "Horizon migration support disabled in configuration - using pure legacy mode"
+        );
+        false
+    };
 
-            let v1_watcher = indexer_monitor::escrow_accounts_v1(
-                escrow_subgraph_v1,
-                indexer_address,
-                config.subgraphs.escrow.config.syncing_interval_secs,
-                true, // Reject thawing signers eagerly
-            )
-            .await
-            .expect("Error creating escrow_accounts_v1 channel");
-
-            if let Some(escrow_v2_subgraph) = escrow_subgraph_v2 {
-                let v2_watcher = indexer_monitor::escrow_accounts_v2(
-                    escrow_v2_subgraph,
-                    indexer_address,
-                    config
-                        .subgraphs
-                        .escrow_v2
-                        .as_ref()
-                        .unwrap()
-                        .config
-                        .syncing_interval_secs,
-                    true, // Reject thawing signers eagerly
-                )
-                .await
-                .expect("Error creating escrow_accounts_v2 channel");
-
-                ServiceRouter::builder()
-                    .database(database.clone())
-                    .domain_separator(domain_separator.clone())
-                    .graph_node(config.graph_node)
-                    .http_client(http_client)
-                    .release(release)
-                    .indexer(config.indexer)
-                    .service(config.service)
-                    .blockchain(config.blockchain)
-                    .timestamp_buffer_secs(config.tap.rav_request.timestamp_buffer_secs)
-                    .network_subgraph(network_subgraph, config.subgraphs.network)
-                    .escrow_accounts_v1(v1_watcher)
-                    .escrow_accounts_v2(v2_watcher)
-                    .build()
-            } else {
-                tracing::warn!("Horizon mode is Transition but no escrow_v2 configuration provided, falling back to v1 only");
-                ServiceRouter::builder()
-                    .database(database.clone())
-                    .domain_separator(domain_separator.clone())
-                    .graph_node(config.graph_node)
-                    .http_client(http_client)
-                    .release(release)
-                    .indexer(config.indexer)
-                    .service(config.service)
-                    .blockchain(config.blockchain)
-                    .timestamp_buffer_secs(config.tap.rav_request.timestamp_buffer_secs)
-                    .network_subgraph(network_subgraph, config.subgraphs.network)
-                    .escrow_accounts_v1(v1_watcher)
-                    .build()
+    // Configure router with escrow watchers based on automatic Horizon detection
+    let router = if is_horizon_active {
+        tracing::info!("Horizon contracts detected - using Horizon migration mode: V2 receipts only, but processing existing V1 receipts");
+
+        // Create V1 escrow watcher for processing existing receipts
+        let escrow_subgraph_v1 = create_subgraph_client(
+            http_client.clone(),
+            &config.graph_node,
+            &config.subgraphs.escrow.config,
+        )
+        .await;
+
+        let v1_watcher = indexer_monitor::escrow_accounts_v1(
+            escrow_subgraph_v1,
+            indexer_address,
+            config.subgraphs.escrow.config.syncing_interval_secs,
+            true, // Reject thawing signers eagerly
+        )
+        .await
+        .expect("Error creating escrow_accounts_v1 channel");
+
+        // Create V2 escrow watcher for new receipts (V2 escrow accounts are in the network subgraph)
+        let v2_watcher = match indexer_monitor::escrow_accounts_v2(
+            network_subgraph,
+            indexer_address,
+            config.subgraphs.network.config.syncing_interval_secs,
+            true, // Reject thawing signers eagerly
+        )
+        .await
+        {
+            Ok(watcher) => {
+                tracing::info!("V2 escrow accounts successfully initialized from network subgraph");
+                watcher
             }
-        }
-        HorizonMode::Full => {
-            tracing::info!("Horizon mode: Full - using escrow accounts v2 only");
-            // Only create v2 watcher for full Horizon mode
-            let v2_subgraph = if let Some(escrow_v2_subgraph) = escrow_subgraph_v2 {
-                escrow_v2_subgraph
-            } else {
-                tracing::warn!("Horizon mode is Full but no escrow_v2 configuration provided, falling back to escrow v1 endpoint for v2 queries");
-                create_subgraph_client(
-                    http_client.clone(),
-                    &config.graph_node,
-                    &config.subgraphs.escrow.config,
-                )
-                .await
-            };
-            let v2_config = config
-                .subgraphs
-                .escrow_v2
-                .as_ref()
-                .map(|c| &c.config)
-                .unwrap_or(&config.subgraphs.escrow.config);
-
-            let v2_watcher = indexer_monitor::escrow_accounts_v2(
-                v2_subgraph,
-                indexer_address,
-                v2_config.syncing_interval_secs,
-                true, // Reject thawing signers eagerly
-            )
-            .await
-            .expect("Error creating escrow_accounts_v2 channel");
-
-            ServiceRouter::builder()
-                .database(database.clone())
-                .domain_separator(domain_separator.clone())
-                .graph_node(config.graph_node)
-                .http_client(http_client)
-                .release(release)
-                .indexer(config.indexer)
-                .service(config.service)
-                .blockchain(config.blockchain)
-                .timestamp_buffer_secs(config.tap.rav_request.timestamp_buffer_secs)
-                .network_subgraph(network_subgraph, config.subgraphs.network)
-                .escrow_accounts_v2(v2_watcher)
-                .build()
-        }
+            Err(e) => {
+                tracing::error!(
+                    "Failed to initialize V2 escrow accounts: {}. Service cannot continue.",
+                    e
+                );
+                std::process::exit(1);
+            }
+        };
+
+        ServiceRouter::builder()
+            .database(database.clone())
+            .domain_separator(domain_separator.clone())
+            .graph_node(config.graph_node)
+            .http_client(http_client)
+            .release(release)
+            .indexer(config.indexer)
+            .service(config.service)
+            .blockchain(config.blockchain)
+            .timestamp_buffer_secs(config.tap.rav_request.timestamp_buffer_secs)
+            .network_subgraph(network_subgraph, config.subgraphs.network)
+            .escrow_accounts_v1(v1_watcher)
+            .escrow_accounts_v2(v2_watcher)
+            .build()
+    } else {
+        tracing::info!(
+            "No Horizon contracts detected - using Legacy (V1) mode with escrow accounts v1 only"
+        );
+        // Only create v1 watcher for legacy mode
+        let escrow_subgraph_v1 = create_subgraph_client(
+            http_client.clone(),
+            &config.graph_node,
+            &config.subgraphs.escrow.config,
+        )
+        .await;
+
+        let v1_watcher = indexer_monitor::escrow_accounts_v1(
+            escrow_subgraph_v1,
+            indexer_address,
+            config.subgraphs.escrow.config.syncing_interval_secs,
+            true, // Reject thawing signers eagerly
+        )
+        .await
+        .expect("Error creating escrow_accounts_v1 channel");
+
+        ServiceRouter::builder()
+            .database(database.clone())
+            .domain_separator(domain_separator.clone())
+            .graph_node(config.graph_node)
+            .http_client(http_client)
+            .release(release)
+            .indexer(config.indexer)
+            .service(config.service)
+            .blockchain(config.blockchain)
+            .timestamp_buffer_secs(config.tap.rav_request.timestamp_buffer_secs)
+            .network_subgraph(network_subgraph, config.subgraphs.network)
+            .escrow_accounts_v1(v1_watcher)
+            .build()
     };
 
     serve_metrics(config.metrics.get_socket_addr());