feat(service): use network subgraph for v2 escrow accounts

suchapalaver · suchapalaver · commit 88d69f9187fa · 2025-06-30T15:48:55.000-04:00
diff --git a/crates/service/src/middleware/sender.rs b/crates/service/src/middleware/sender.rs
@@ -6,7 +6,7 @@ use axum::{
     middleware::Next,
     response::Response,
 };
-use indexer_monitor::EscrowAccounts;
+use indexer_monitor::{EscrowAccounts, EscrowAccountsError};
 use thegraph_core::alloy::{primitives::Address, sol_types::Eip712Domain};
 use tokio::sync::watch;
 
@@ -18,9 +18,9 @@ pub struct SenderState {
     /// Used to recover the signer address
     pub domain_separator: Eip712Domain,
     /// Used to get the sender address given the signer address if v1 receipt
-    pub escrow_accounts_v1: watch::Receiver<EscrowAccounts>,
+    pub escrow_accounts_v1: Option<watch::Receiver<EscrowAccounts>>,
     /// Used to get the sender address given the signer address if v2 receipt
-    pub escrow_accounts_v2: watch::Receiver<EscrowAccounts>,
+    pub escrow_accounts_v2: Option<watch::Receiver<EscrowAccounts>>,
 }
 
 /// The current query Sender address
@@ -48,14 +48,24 @@ pub async fn sender_middleware(
     if let Some(receipt) = request.extensions().get::<TapReceipt>() {
         let signer = receipt.recover_signer(&state.domain_separator)?;
         let sender = match receipt {
-            TapReceipt::V1(_) => state
-                .escrow_accounts_v1
-                .borrow()
-                .get_sender_for_signer(&signer)?,
-            TapReceipt::V2(_) => state
-                .escrow_accounts_v2
-                .borrow()
-                .get_sender_for_signer(&signer)?,
+            TapReceipt::V1(_) => {
+                if let Some(ref escrow_accounts_v1) = state.escrow_accounts_v1 {
+                    escrow_accounts_v1.borrow().get_sender_for_signer(&signer)?
+                } else {
+                    return Err(IndexerServiceError::EscrowAccount(
+                        EscrowAccountsError::NoSenderFound { signer },
+                    ));
+                }
+            }
+            TapReceipt::V2(_) => {
+                if let Some(ref escrow_accounts_v2) = state.escrow_accounts_v2 {
+                    escrow_accounts_v2.borrow().get_sender_for_signer(&signer)?
+                } else {
+                    return Err(IndexerServiceError::EscrowAccount(
+                        EscrowAccountsError::NoSenderFound { signer },
+                    ));
+                }
+            }
         };
         request.extensions_mut().insert(Sender(sender));
     }
@@ -100,8 +110,8 @@ mod tests {
 
         let state = SenderState {
             domain_separator: test_assets::TAP_EIP712_DOMAIN.clone(),
-            escrow_accounts_v1,
-            escrow_accounts_v2,
+            escrow_accounts_v1: Some(escrow_accounts_v1),
+            escrow_accounts_v2: Some(escrow_accounts_v2),
         };
 
         let middleware = from_fn_with_state(state, sender_middleware);
diff --git a/crates/service/src/service.rs b/crates/service/src/service.rs
@@ -18,7 +18,7 @@ use indexer_dips::{
     server::{DipsServer, DipsServerContext},
     signers::EscrowSignerValidator,
 };
-use indexer_monitor::{escrow_accounts_v1, DeploymentDetails, SubgraphClient};
+use indexer_monitor::{escrow_accounts_v1, escrow_accounts_v2, DeploymentDetails, SubgraphClient};
 use release::IndexerServiceRelease;
 use reqwest::Url;
 use tap_core::tap_eip712_domain;
@@ -78,12 +78,7 @@ pub async fn run() -> anyhow::Result<()> {
     )
     .await;
 
-    let escrow_subgraph = create_subgraph_client(
-        http_client.clone(),
-        &config.graph_node,
-        &config.subgraphs.escrow.config,
-    )
-    .await;
+    // 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
@@ -105,19 +100,133 @@ pub async fn run() -> anyhow::Result<()> {
     let indexer_address = config.indexer.indexer_address;
     let ipfs_url = config.service.ipfs_url.clone();
 
-    let router = 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_subgraph(escrow_subgraph, config.subgraphs.escrow)
-        .build();
+    // Capture individual fields needed for DIPS before they get moved
+    let escrow_v1_query_url_for_dips = config.subgraphs.escrow.config.query_url.clone();
+    // 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
+            }
+        }
+    } else {
+        tracing::info!(
+            "Horizon migration support disabled in configuration - using pure legacy mode"
+        );
+        false
+    };
+
+    // 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
+            }
+            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());
 
@@ -143,14 +252,64 @@ pub async fn run() -> anyhow::Result<()> {
             Arc::new(IpfsClient::new(ipfs_url.as_str()).unwrap());
 
         // TODO: Try to re-use the same watcher for both DIPS and TAP
-        let watcher = escrow_accounts_v1(
-            escrow_subgraph,
-            indexer_address,
-            Duration::from_secs(500),
-            true,
-        )
-        .await
-        .expect("Failed to create escrow accounts watcher");
+        // DIPS is part of Horizon/v2, so use v2 escrow watcher when available
+        let dips_http_client = reqwest::Client::builder()
+            .timeout(Duration::from_secs(60))
+            .build()
+            .expect("Failed to init HTTP client");
+
+        let escrow_subgraph_for_dips = if let Some(ref escrow_v2_url) = escrow_v2_query_url_for_dips
+        {
+            tracing::info!("DIPS using v2 escrow subgraph");
+            // Create subgraph client for v2
+            Box::leak(Box::new(
+                SubgraphClient::new(
+                    dips_http_client,
+                    None, // No local deployment
+                    DeploymentDetails::for_query_url_with_token(
+                        escrow_v2_url.clone(),
+                        None, // No auth token
+                    ),
+                )
+                .await,
+            ))
+        } else {
+            tracing::info!("DIPS falling back to v1 escrow subgraph");
+            // Create subgraph client for v1
+            Box::leak(Box::new(
+                SubgraphClient::new(
+                    dips_http_client,
+                    None, // No local deployment
+                    DeploymentDetails::for_query_url_with_token(
+                        escrow_v1_query_url_for_dips,
+                        None, // No auth token
+                    ),
+                )
+                .await,
+            ))
+        };
+
+        let watcher = if escrow_v2_query_url_for_dips.is_some() {
+            // Use v2 watcher for DIPS when v2 is available
+            escrow_accounts_v2(
+                escrow_subgraph_for_dips,
+                indexer_address,
+                Duration::from_secs(500),
+                true,
+            )
+            .await
+            .expect("Failed to create escrow accounts v2 watcher for DIPS")
+        } else {
+            // Fall back to v1 watcher
+            escrow_accounts_v1(
+                escrow_subgraph_for_dips,
+                indexer_address,
+                Duration::from_secs(500),
+                true,
+            )
+            .await
+            .expect("Failed to create escrow accounts v1 watcher for DIPS")
+        };
 
         let registry = NetworksRegistry::from_latest_version().await.unwrap();
 
diff --git a/crates/service/src/service/router.rs b/crates/service/src/service/router.rs
@@ -140,33 +140,42 @@ impl ServiceRouter {
         // Monitor escrow accounts v1
         // if not provided, create monitor from subgraph
         let escrow_accounts_v1 = match (self.escrow_accounts_v1, self.escrow_subgraph.as_ref()) {
-            (Some(escrow_account), _) => escrow_account,
-            (_, Some((escrow_subgraph, escrow))) => escrow_accounts_v1(
-                escrow_subgraph,
-                indexer_address,
-                escrow.config.syncing_interval_secs,
-                true, // Reject thawing signers eagerly
-            )
-            .await
-            .expect("Error creating escrow_accounts channel"),
-            (None, None) => panic!("No escrow accounts or escrow subgraph was provided"),
+            (Some(escrow_account), _) => Some(escrow_account),
+            (_, Some((escrow_subgraph, escrow))) => Some(
+                escrow_accounts_v1(
+                    escrow_subgraph,
+                    indexer_address,
+                    escrow.config.syncing_interval_secs,
+                    true, // Reject thawing signers eagerly
+                )
+                .await
+                .expect("Error creating escrow_accounts_v1 channel"),
+            ),
+            (None, None) => None,
         };
 
         // Monitor escrow accounts v2
         // if not provided, create monitor from subgraph
         let escrow_accounts_v2 = match (self.escrow_accounts_v2, self.escrow_subgraph.as_ref()) {
-            (Some(escrow_account), _) => escrow_account,
-            (_, Some((escrow_subgraph, escrow))) => escrow_accounts_v2(
-                escrow_subgraph,
-                indexer_address,
-                escrow.config.syncing_interval_secs,
-                true, // Reject thawing signers eagerly
-            )
-            .await
-            .expect("Error creating escrow_accounts channel"),
-            (None, None) => panic!("No escrow accounts or escrow subgraph was provided"),
+            (Some(escrow_account), _) => Some(escrow_account),
+            (_, Some((escrow_subgraph, escrow))) => Some(
+                escrow_accounts_v2(
+                    escrow_subgraph,
+                    indexer_address,
+                    escrow.config.syncing_interval_secs,
+                    true, // Reject thawing signers eagerly
+                )
+                .await
+                .expect("Error creating escrow_accounts_v2 channel"),
+            ),
+            (None, None) => None,
         };
 
+        // Ensure at least one escrow accounts watcher is available
+        if escrow_accounts_v1.is_none() && escrow_accounts_v2.is_none() {
+            panic!("At least one escrow accounts watcher (v1 or v2) must be provided");
+        }
+
         // Monitor dispute manager address
         // if not provided, create monitor from subgraph
         let dispute_manager = match (self.dispute_manager, self.network_subgraph.as_ref()) {
diff --git a/crates/service/src/tap.rs b/crates/service/src/tap.rs
@@ -51,8 +51,8 @@ impl IndexerTapContext {
     pub async fn get_checks(
         pgpool: PgPool,
         indexer_allocations: Receiver<HashMap<Address, Allocation>>,
-        escrow_accounts_v1: Receiver<EscrowAccounts>,
-        escrow_accounts_v2: Receiver<EscrowAccounts>,
+        escrow_accounts_v1: Option<Receiver<EscrowAccounts>>,
+        escrow_accounts_v2: Option<Receiver<EscrowAccounts>>,
         timestamp_error_tolerance: Duration,
         receipt_max_value: u128,
     ) -> Vec<ReceiptCheck<TapReceipt>> {
diff --git a/crates/service/src/tap/checks/sender_balance_check.rs b/crates/service/src/tap/checks/sender_balance_check.rs
