chore: progress towards collecting payments

Author: pcarranzav

.sqlx/query-f36803bd99ba0b93b312950e8d88c94a80335ea34bf7200aef5d18e6983a2cb1.json

@@ -153,3 +153,4 @@ max_receipts_per_request = 10000
 host = "0.0.0.0"
 port = "7601"
 allowed_payers = ["0x3333333333333333333333333333333333333333"]
+dipper_grpc_url = "https://dipper.thegraph.com/"

Cargo.lock

@@ -390,6 +390,7 @@ pub struct DipsConfig {
     pub host: String,
     pub port: String,
     pub allowed_payers: Vec<Address>,
+    pub dipper_grpc_url: String,
 }
 
 impl Default for DipsConfig {
@@ -398,6 +399,7 @@ impl Default for DipsConfig {
             host: "0.0.0.0".to_string(),
             port: "7601".to_string(),
             allowed_payers: vec![],
+            dipper_grpc_url: "".to_string(),
         }
     }
 }
@@ -462,6 +464,7 @@ mod tests {
             allowed_payers: vec![Address(
                 FixedBytes::<20>::from_str("0x3333333333333333333333333333333333333333").unwrap(),
             )],
+            dipper_grpc_url: "https://dipper.thegraph.com/".to_string(),
             ..Default::default()
         });
 

crates/config/maximal-config-example.toml

@@ -4,6 +4,8 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+axum.workspace = true
+build-info.workspace = true
 thiserror.workspace = true
 anyhow.workspace = true
 alloy-rlp = "0.3.10"
@@ -15,6 +17,7 @@ prost-types.workspace = true
 uuid.workspace = true
 base64.workspace = true
 tokio.workspace = true
+sqlx.workspace = true
 
 [build-dependencies]
 tonic-build = { workspace = true }

crates/config/src/config.rs

@@ -6,15 +6,13 @@ fn main() {
     tonic_build::configure()
         .out_dir("src/proto")
         .include_file("indexer.rs")
-        .build_client(false)
         .protoc_arg("--experimental_allow_proto3_optional")
         .compile_protos(&["proto/indexer.proto"], &["proto/"])
         .expect("Failed to compile DIPs indexer RPC proto(s)");
 
     tonic_build::configure()
         .out_dir("src/proto")
         .include_file("gateway.rs")
-        .build_server(false)
         .protoc_arg("--experimental_allow_proto3_optional")
         .compile_protos(&["proto/gateway.proto"], &["proto"])
         .expect("Failed to compile DIPs gateway RPC proto(s)");

crates/dips/Cargo.toml

@@ -2,7 +2,7 @@ syntax = "proto3";
 
 package graphprotocol.gateway.dips;
 
-service DipsService {
+service DipperService {
   /**
    * Cancel an _indexing agreement_.
    *

crates/dips/build.rs

@@ -2,7 +2,7 @@ syntax = "proto3";
 
 package graphprotocol.indexer.dips;
 
-service DipsService {
+service IndexerDipsService {
   /**
    * Propose a new _indexing agreement_ to an _indexer_.
    *

crates/dips/proto/gateway.proto

@@ -5,14 +5,15 @@ use std::str::FromStr;
 
 use axum::async_trait;
 use build_info::chrono::{DateTime, Utc};
-use indexer_dips::{
+use sqlx::{types::BigDecimal, PgPool};
+use thegraph_core::alloy::{core::primitives::U256 as uint256, hex::ToHexExt, sol_types::SolType};
+use uuid::Uuid;
+
+use crate::{
     store::{AgreementStore, StoredIndexingAgreement},
     DipsError, SignedCancellationRequest, SignedIndexingAgreementVoucher,
     SubgraphIndexingVoucherMetadata,
 };
-use sqlx::{types::BigDecimal, PgPool};
-use thegraph_core::alloy::{core::primitives::U256 as uint256, hex::ToHexExt, sol_types::SolType};
-use uuid::Uuid;
 
 #[derive(Debug)]
 pub struct PsqlAgreementStore {
@@ -51,6 +52,47 @@ impl AgreementStore for PsqlAgreementStore {
             last_allocation_id: item.last_allocation_id,
         }))
     }
+    async fn get_by_last_allocation_id(
+        &self,
+        allocation_id: String,
+    ) -> Result<Vec<StoredIndexingAgreement>, DipsError> {
+        let items = sqlx::query!(
+            "SELECT * FROM indexing_agreements WHERE last_allocation_id=$1",
+            allocation_id,
+        )
+        .fetch_all(&self.pool)
+        .await;
+
+        let items = match items {
+            Ok(items) => items,
+            Err(sqlx::Error::RowNotFound) => return Ok(vec![]),
+            Err(err) => return Err(DipsError::UnknownError(err.into())),
+        };
+
+        // Note: we discard any agreements that fail to decode
+        let agreements = items
+            .into_iter()
+            .map(|item| {
+                let signed =
+                    SignedIndexingAgreementVoucher::abi_decode(item.signed_payload.as_ref(), true)
+                        .map_err(|e| DipsError::AbiDecoding(e.to_string()))?;
+                let metadata = SubgraphIndexingVoucherMetadata::abi_decode(
+                    signed.voucher.metadata.as_ref(),
+                    true,
+                )
+                .map_err(|e| DipsError::AbiDecoding(e.to_string()))?;
+                Ok(StoredIndexingAgreement {
+                    voucher: signed,
+                    metadata,
+                    cancelled: item.cancelled_at.is_some(),
+                    current_allocation_id: item.current_allocation_id,
+                    last_allocation_id: item.last_allocation_id,
+                })
+            })
+            .filter_map(|agreement: Result<StoredIndexingAgreement, DipsError>| agreement.ok())
+            .collect();
+        Ok(agreements)
+    }
     async fn create_agreement(
         &self,
         agreement: SignedIndexingAgreementVoucher,
@@ -133,7 +175,6 @@ pub(crate) mod test {
     use std::sync::Arc;
 
     use build_info::chrono::Duration;
-    use indexer_dips::{CancellationRequest, IndexingAgreementVoucher};
     use sqlx::PgPool;
     use thegraph_core::alloy::{
         primitives::{ruint::aliases::U256, Address},
@@ -142,6 +183,7 @@ pub(crate) mod test {
     use uuid::Uuid;
 
     use super::*;
+    use crate::{CancellationRequest, IndexingAgreementVoucher};
 
     #[sqlx::test(migrations = "../../migrations")]
     async fn test_store_agreement(pool: PgPool) {

crates/dips/proto/indexer.proto

@@ -5,12 +5,14 @@ use std::{str::FromStr, sync::Arc};
 
 use thegraph_core::alloy::{
     core::primitives::Address,
+    hex::ToHexExt,
     primitives::{b256, ChainId, PrimitiveSignature as Signature, B256},
     signers::SignerSync,
     sol,
     sol_types::{eip712_domain, Eip712Domain, SolStruct, SolValue},
 };
 
+pub mod database;
 pub mod proto;
 pub mod server;
 pub mod store;
@@ -323,6 +325,25 @@ pub async fn validate_and_cancel_agreement(
     Ok(id)
 }
 
+pub async fn collect_indexing_agreements(
+    store: Arc<dyn AgreementStore>,
+    allocation_id: Address,
+    _dipper_grpc_url: &str,
+) -> Result<(), DipsError> {
+    let agreements = store
+        .get_by_last_allocation_id(allocation_id.encode_hex())
+        .await?;
+
+    for _agreement in agreements {
+        // TODO get the entities count for the deployment from graph node
+        // TODO create collection request
+        // TODO sign collection request
+        // TODO send collection request to the dipper via grpc
+    }
+
+    Ok(())
+}
+
 #[cfg(test)]
 mod test {
     use std::{