fix(monitor): implement v2 escrow accounts

Author: suchapalaver

crates/monitor/src/escrow_accounts.rs

@@ -9,6 +9,9 @@ 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;
@@ -112,13 +115,76 @@ pub async fn escrow_accounts_v2(
     .await
 }
 
-// TODO implement escrow accounts v2 query
 async fn get_escrow_accounts_v2(
-    _escrow_subgraph: &'static SubgraphClient,
-    _indexer_address: Address,
-    _reject_thawing_signers: bool,
+    escrow_subgraph: &'static SubgraphClient,
+    indexer_address: Address,
+    reject_thawing_signers: bool,
 ) -> anyhow::Result<EscrowAccounts> {
-    Ok(EscrowAccounts::new(HashMap::new(), HashMap::new()))
+    // 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
+
+    let response = escrow_subgraph
+        .query::<EscrowAccountQueryV2, _>(escrow_account_v2::Variables {
+            indexer: format!("{:x?}", indexer_address),
+            thaw_end_timestamp: if reject_thawing_signers {
+                U256::ZERO.to_string()
+            } else {
+                U256::MAX.to_string()
+            },
+        })
+        .await?;
+
+    let response = response?;
+
+    tracing::trace!("V2 Escrow accounts response: {:?}", response);
+
+    let senders_balances: HashMap<Address, U256> = response
+        .escrow_accounts
+        .iter()
+        .map(|account| {
+            let balance = U256::checked_sub(
+                U256::from_str(&account.balance)?,
+                U256::from_str(&account.total_amount_thawing)?,
+            )
+            .unwrap_or_else(|| {
+                tracing::warn!(
+                    "Balance minus total amount thawing underflowed for V2 account {}. \
+                                 Setting balance to 0, no V2 queries will be served for this payer.",
+                    account.payer.id
+                );
+                U256::from(0)
+            });
+
+            Ok((Address::from_str(&account.payer.id)?, balance))
+        })
+        .collect::<Result<HashMap<_, _>, anyhow::Error>>()?;
+
+    let senders_to_signers = response
+        .escrow_accounts
+        .into_iter()
+        .map(|account| {
+            let payer = Address::from_str(&account.payer.id)?;
+            let signers = account
+                .payer
+                .signers
+                .iter()
+                .map(|signer| Address::from_str(&signer.id))
+                .collect::<Result<Vec<_>, _>>()?;
+            Ok((payer, signers))
+        })
+        .collect::<Result<HashMap<_, _>, anyhow::Error>>()?;
+
+    Ok(EscrowAccounts::new(senders_balances, senders_to_signers))
 }
 
 async fn get_escrow_accounts_v1(
@@ -262,4 +328,54 @@ mod tests {
             )
         );
     }
+
+    #[test(tokio::test)]
+    async fn test_current_accounts_v2() {
+        // Set up a mock escrow subgraph for V2 with payer fields
+        let mock_server = MockServer::start().await;
+        let escrow_subgraph = Box::leak(Box::new(
+            SubgraphClient::new(
+                reqwest::Client::new(),
+                None,
+                DeploymentDetails::for_query_url(&format!(
+                    "{}/subgraphs/id/{}",
+                    &mock_server.uri(),
+                    test_assets::ESCROW_SUBGRAPH_DEPLOYMENT
+                ))
+                .unwrap(),
+            )
+            .await,
+        ));
+
+        let mock = Mock::given(method("POST"))
+            .and(path(format!(
+                "/subgraphs/id/{}",
+                test_assets::ESCROW_SUBGRAPH_DEPLOYMENT
+            )))
+            .respond_with(
+                ResponseTemplate::new(200)
+                    .set_body_raw(test_assets::ESCROW_QUERY_RESPONSE_V2, "application/json"),
+            );
+        mock_server.register(mock).await;
+
+        // Test V2 escrow accounts watcher
+        let mut accounts = escrow_accounts_v2(
+            escrow_subgraph,
+            test_assets::INDEXER_ADDRESS,
+            Duration::from_secs(60),
+            true,
+        )
+        .await
+        .unwrap();
+        accounts.changed().await.unwrap();
+
+        // V2 should produce identical results to V1 since they query the same data
+        assert_eq!(
+            accounts.borrow().clone(),
+            EscrowAccounts::new(
+                ESCROW_ACCOUNTS_BALANCES.to_owned(),
+                ESCROW_ACCOUNTS_SENDERS_TO_SIGNERS.to_owned(),
+            )
+        );
+    }
 }

crates/monitor/src/horizon_detection.rs

@@ -0,0 +1,48 @@
+// Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Horizon (V2) contract detection utilities
+//!
+//! This module provides functionality to detect if Horizon (V2) contracts are active
+//! in the network by querying the network subgraph for PaymentsEscrow accounts.
+
+use anyhow::Result;
+use indexer_query::horizon_detection::{self, HorizonDetectionQuery};
+
+use crate::client::SubgraphClient;
+
+/// Detects if Horizon (V2) contracts are active in the network.
+///
+/// This function queries the network subgraph to check if any PaymentsEscrow accounts exist.
+/// If they do, it indicates that Horizon contracts are deployed and active.
+///
+/// # Arguments
+/// * `network_subgraph` - The network subgraph client to query
+///
+/// # Returns
+/// * `Ok(true)` if Horizon contracts are active
+/// * `Ok(false)` if only legacy (V1) contracts are active
+/// * `Err(...)` if there was an error querying the subgraph
+pub async fn is_horizon_active(network_subgraph: &SubgraphClient) -> Result<bool> {
+    tracing::debug!("Checking if Horizon (V2) contracts are active in the network");
+
+    let response = network_subgraph
+        .query::<HorizonDetectionQuery, _>(horizon_detection::Variables {})
+        .await?;
+
+    let response = response?;
+
+    // If we find any PaymentsEscrow accounts, Horizon is active
+    let horizon_active = !response.payments_escrow_accounts.is_empty();
+
+    if horizon_active {
+        tracing::info!(
+            "Horizon (V2) contracts detected - found {} PaymentsEscrow accounts",
+            response.payments_escrow_accounts.len()
+        );
+    } else {
+        tracing::info!("No Horizon (V2) contracts found - using legacy (V1) mode");
+    }
+
+    Ok(horizon_active)
+}

crates/monitor/src/lib.rs

@@ -7,6 +7,7 @@ mod client;
 mod deployment_to_allocation;
 mod dispute_manager;
 mod escrow_accounts;
+mod horizon_detection;
 
 pub use crate::{
     allocations::{indexer_allocations, AllocationWatcher},
@@ -18,4 +19,5 @@ pub use crate::{
         escrow_accounts_v1, escrow_accounts_v2, EscrowAccounts, EscrowAccountsError,
         EscrowAccountsWatcher,
     },
+    horizon_detection::is_horizon_active,
 };