chore: dipper client

Author: pcarranzav

crates/dips/src/lib.rs

@@ -3,6 +3,7 @@
 
 use std::{str::FromStr, sync::Arc};
 
+use proto::gateway::graphprotocol::gateway::dips::dipper_service_client::DipperServiceClient;
 use thegraph_core::alloy::{
     core::primitives::Address,
     hex::ToHexExt,
@@ -19,6 +20,7 @@ pub mod store;
 
 use store::AgreementStore;
 use thiserror::Error;
+use tonic::transport::Channel;
 use uuid::Uuid;
 
 /// The Arbitrum One (mainnet) chain ID (eip155).
@@ -151,6 +153,8 @@ pub enum DipsError {
     AgreementCancelled,
     #[error("invalid voucher: {0}")]
     InvalidVoucher(String),
+    #[error("connection error: {0}")]
+    ConnectionError(#[from] tonic::transport::Error),
 }
 
 // TODO: send back messages
@@ -328,7 +332,7 @@ pub async fn validate_and_cancel_agreement(
 pub async fn collect_indexing_agreements(
     store: Arc<dyn AgreementStore>,
     allocation_id: Address,
-    _dipper_grpc_url: &str,
+    _dipper_client: &DipperServiceClient<Channel>,
 ) -> Result<(), DipsError> {
     let agreements = store
         .get_by_last_allocation_id(allocation_id.encode_hex())

crates/tap-agent/src/agent/sender_account.rs

@@ -12,7 +12,9 @@ use anyhow::Context;
 use bigdecimal::{num_bigint::ToBigInt, ToPrimitive};
 use futures::{stream, StreamExt};
 use indexer_dips::{
-    collect_indexing_agreements, database::PsqlAgreementStore, store::AgreementStore,
+    collect_indexing_agreements, database::PsqlAgreementStore,
+    proto::gateway::graphprotocol::gateway::dips::dipper_service_client::DipperServiceClient,
+    store::AgreementStore, DipsError,
 };
 use indexer_monitor::{EscrowAccounts, SubgraphClient};
 use indexer_query::{
@@ -324,6 +326,8 @@ pub struct State {
     config: &'static SenderAccountConfig,
     /// Indexing agreement store
     agreement_store: Arc<dyn AgreementStore>,
+    /// Client for the dipper grpc server
+    dipper_client: Option<DipperServiceClient<Channel>>,
 }
 
 /// Configuration derived from config.toml
@@ -836,6 +840,15 @@ impl Actor for SenderAccount {
         let agreement_store = Arc::new(PsqlAgreementStore {
             pool: pgpool.clone(),
         });
+
+        let dipper_client = if let Some(ref url) = config.dipper_grpc_url {
+            let endpoint = tonic::transport::Endpoint::from_str(url)
+                .map_err(DipsError::ConnectionError)?
+                .connect_lazy();
+            Some(DipperServiceClient::new(endpoint))
+        } else {
+            None
+        };
         let state = State {
             prefix,
             sender_fee_tracker: SenderFeeTracker::new(config.rav_request_buffer),
@@ -858,6 +871,7 @@ impl Actor for SenderAccount {
             backoff_info: BackoffInfo::default(),
             config,
             agreement_store,
+            dipper_client,
         };
 
         stream::iter(allocation_ids)
@@ -1069,12 +1083,12 @@ impl Actor for SenderAccount {
                         if let Some(sender_handle) = ActorRef::<SenderAllocationMessage>::where_is(
                             state.format_sender_allocation(&allocation_id.address()),
                         ) {
-                            if let Some(ref dipper_grpc_url) = state.config.dipper_grpc_url {
+                            if let Some(ref dipper_client) = state.dipper_client {
                                 tracing::trace!(%allocation_id, "SenderAccount checking for indexing agreements to collect");
                                 match collect_indexing_agreements(
                                     state.agreement_store.clone(),
                                     allocation_id.address(),
-                                    dipper_grpc_url,
+                                    dipper_client,
                                 )
                                 .await
                                 {