fix(monitor): implement v2 escrow accounts

Author: suchapalaver

contrib/indexer-service/start.sh

@@ -30,6 +30,15 @@ ESCROW_DEPLOYMENT=$(curl -s --max-time 10 "http://graph-node:8000/subgraphs/name
     -d '{"query": "{ _meta { deployment } }"}' | jq -r '.data._meta.deployment' 2>/dev/null)
 stdbuf -oL echo "🔍 DEBUG: Escrow deployment result: $ESCROW_DEPLOYMENT"
 
+# Handle null deployment IDs by removing the lines entirely
+if [ "$NETWORK_DEPLOYMENT" = "null" ] || [ -z "$NETWORK_DEPLOYMENT" ]; then
+    NETWORK_DEPLOYMENT=""
+fi
+
+if [ "$ESCROW_DEPLOYMENT" = "null" ] || [ -z "$ESCROW_DEPLOYMENT" ]; then
+    ESCROW_DEPLOYMENT=""
+fi
+
 stdbuf -oL echo "Escrow subgraph deployment ID: $ESCROW_DEPLOYMENT"
 stdbuf -oL echo "Using test Network subgraph deployment ID: $NETWORK_DEPLOYMENT"
 stdbuf -oL echo "Using test Verifier address: $VERIFIER_ADDRESS"
@@ -41,8 +50,19 @@ stdbuf -oL echo "Using test Account0 address: $ACCOUNT0_ADDRESS"
 cp /opt/config/config.toml /opt/config.toml
 
 # Replace the placeholders with actual values
-sed -i "s/NETWORK_DEPLOYMENT_PLACEHOLDER/$NETWORK_DEPLOYMENT/g" /opt/config.toml
-sed -i "s/ESCROW_DEPLOYMENT_PLACEHOLDER/$ESCROW_DEPLOYMENT/g" /opt/config.toml
+if [ -n "$NETWORK_DEPLOYMENT" ]; then
+    sed -i "s/NETWORK_DEPLOYMENT_PLACEHOLDER/$NETWORK_DEPLOYMENT/g" /opt/config.toml
+else
+    # Remove the deployment_id line entirely for network subgraph
+    sed -i '/deployment_id = "NETWORK_DEPLOYMENT_PLACEHOLDER"/d' /opt/config.toml
+fi
+
+if [ -n "$ESCROW_DEPLOYMENT" ]; then
+    sed -i "s/ESCROW_DEPLOYMENT_PLACEHOLDER/$ESCROW_DEPLOYMENT/g" /opt/config.toml
+else
+    # Remove the deployment_id line entirely for escrow subgraph
+    sed -i '/deployment_id = "ESCROW_DEPLOYMENT_PLACEHOLDER"/d' /opt/config.toml
+fi
 sed -i "s/VERIFIER_ADDRESS_PLACEHOLDER/$VERIFIER_ADDRESS/g" /opt/config.toml
 sed -i "s/INDEXER_ADDRESS_PLACEHOLDER/$RECEIVER_ADDRESS/g" /opt/config.toml
 sed -i "s/INDEXER_MNEMONIC_PLACEHOLDER/$INDEXER_MNEMONIC/g" /opt/config.toml

crates/monitor/src/escrow_accounts.rs

@@ -112,13 +112,77 @@ 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::<EscrowAccountQuery, _>(escrow_account::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 sender.",
+                    account.sender.id
+                );
+                U256::from(0)
+            });
+
+            Ok((Address::from_str(&account.sender.id)?, balance))
+        })
+        .collect::<Result<HashMap<_, _>, anyhow::Error>>()?;
+
+    let senders_to_signers = response
+        .escrow_accounts
+        .into_iter()
+        .map(|account| {
+            let sender = Address::from_str(&account.sender.id)?;
+            let signers = account
+                .sender
+                .signers
+                .ok_or(anyhow!("Could not find any signers for V2 sender {sender}"))?
+                .iter()
+                .map(|signer| Address::from_str(&signer.id))
+                .collect::<Result<Vec<_>, _>>()?;
+            Ok((sender, signers))
+        })
+        .collect::<Result<HashMap<_, _>, anyhow::Error>>()?;
+
+    Ok(EscrowAccounts::new(senders_balances, senders_to_signers))
 }
 
 async fn get_escrow_accounts_v1(
@@ -262,4 +326,54 @@ mod tests {
             )
         );
     }
+
+    #[test(tokio::test)]
+    async fn test_current_accounts_v2() {
+        // Set up a mock escrow subgraph - V2 uses the same subgraph as V1
+        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, "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/tap-agent/src/agent/sender_account.rs

@@ -869,6 +869,7 @@ impl Actor for SenderAccount {
                     // Depends on Escrow Subgraph Schema
                     SenderType::Horizon => {
                         if config.horizon_enabled {
+                            // TODO: Implement query for unfinalized v2 transactions, It depends on Escrow Subgraph Schema
                             todo!("Implement query for unfinalized v2 transactions, It depends on Escrow Subgraph Schema")
                         }
                         // if we have any problems, we don't want to filter out