refactor: use generic manager

Signed-off-by: Gustavo Inacio <[email protected]>

Author: gusinacio

Cargo.lock

@@ -24,7 +24,7 @@ axum = { version = "0.7.9", default-features = false, features = [
     "http1",
     "http2",
 ] }
-tokio = "1.40"
+tokio = "1"
 prometheus = "0.13.3"
 anyhow = { version = "1.0.72" }
 thiserror = "1.0.49"
@@ -52,22 +52,22 @@ uuid = { version = "1.11.0", features = ["v7"] }
 tracing = { version = "0.1.40", default-features = false }
 bigdecimal = "0.4.3"
 build-info = "0.0.39"
-tap_core = { git = "https://github.com/semiotic-ai/timeline-aggregation-protocol", rev = "6af1add", default-features = false }
-tap_aggregator = { git = "https://github.com/semiotic-ai/timeline-aggregation-protocol", rev = "6af1add", default-features = false }
+tap_core = { git = "https://github.com/semiotic-ai/timeline-aggregation-protocol", rev = "1fc51a3", default-features = false }
+tap_aggregator = { git = "https://github.com/semiotic-ai/timeline-aggregation-protocol", rev = "1fc51a3", default-features = false }
 tracing-subscriber = { version = "0.3", features = [
     "json",
     "env-filter",
     "ansi",
 ], default-features = false }
-thegraph-core = { version = "0.10.0", features = [
+thegraph-core = { git = "https://github.com/edgeandnode/toolshed", rev = "a1d0509", features = [
     "attestation",
     "alloy-eip712",
     "alloy-sol-types",
     "alloy-rlp",
     "alloy-signers",
     "alloy-signer-local",
     "alloy-signer-mnemonic",
-    "serde"
+    "serde",
 ] }
 thegraph-graphql-http = { version = "0.3.2", features = ["reqwest"] }
 graphql_client = { version = "0.14.0", features = ["reqwest-rustls"] }

Cargo.toml

@@ -22,9 +22,8 @@ pub mod server;
 pub mod store;
 
 use store::AgreementStore;
-use uuid::Uuid;
-
 use thiserror::Error;
+use uuid::Uuid;
 
 sol! {
     // EIP712 encoded bytes, ABI - ethers
@@ -264,9 +263,10 @@ pub async fn validate_and_cancel_agreement(
 
 #[cfg(test)]
 mod test {
-    use std::sync::Arc;
-
-    use std::time::{Duration, SystemTime, UNIX_EPOCH};
+    use std::{
+        sync::Arc,
+        time::{Duration, SystemTime, UNIX_EPOCH},
+    };
 
     use thegraph_core::{
         alloy::{
@@ -276,12 +276,12 @@ mod test {
         },
         attestation::eip712_domain,
     };
-
-    use crate::{CancellationRequest, DipsError};
-    use crate::{IndexingAgreementVoucher, SubgraphIndexingVoucherMetadata};
     use uuid::Uuid;
 
     pub use crate::store::{AgreementStore, InMemoryAgreementStore};
+    use crate::{
+        CancellationRequest, DipsError, IndexingAgreementVoucher, SubgraphIndexingVoucherMetadata,
+    };
 
     #[tokio::test]
     async fn test_validate_and_create_agreement() -> anyhow::Result<()> {

crates/dips/src/lib.rs

@@ -3,14 +3,15 @@
 
 use std::{str::FromStr, sync::Arc, time::Duration};
 
+use anyhow::anyhow;
+use async_trait::async_trait;
+use thegraph_core::alloy::{primitives::Address, sol_types::Eip712Domain};
+use uuid::Uuid;
+
 use crate::{
     proto::graphprotocol::indexer::dips::*, store::AgreementStore, validate_and_cancel_agreement,
     validate_and_create_agreement, DipsError,
 };
-use anyhow::anyhow;
-use async_trait::async_trait;
-use thegraph_core::alloy::{dyn_abi::Eip712Domain, primitives::Address};
-use uuid::Uuid;
 
 #[derive(Debug)]
 pub struct DipsServer {

crates/dips/src/server.rs

@@ -4,11 +4,9 @@
 use anyhow::bail;
 use axum::async_trait;
 use build_info::chrono::Utc;
-
 use indexer_dips::{
     store::AgreementStore, SignedCancellationRequest, SignedIndexingAgreementVoucher,
 };
-
 use sqlx::PgPool;
 use thegraph_core::alloy::rlp::Decodable;
 use uuid::Uuid;

crates/service/src/database/dips.rs

@@ -18,6 +18,7 @@ use axum::{
 };
 use tap_core::{
     manager::{adapters::ReceiptStore, Manager},
+    rav::ReceiptAggregateVoucher,
     receipt::{Context, SignedReceipt},
 };
 use tower_http::auth::AsyncAuthorizeRequest;
@@ -30,7 +31,7 @@ use crate::{error::IndexerServiceError, middleware::prometheus_metrics::MetricLa
 ///
 /// Requires SignedReceipt, MetricLabels and Arc<Context> extensions
 pub fn tap_receipt_authorize<T, B>(
-    tap_manager: Arc<Manager<T>>,
+    tap_manager: Arc<Manager<T, SignedReceipt, ReceiptAggregateVoucher>>,
     failed_receipt_metric: &'static prometheus::CounterVec,
 ) -> impl AsyncAuthorizeRequest<
     B,
@@ -40,7 +41,7 @@ pub fn tap_receipt_authorize<T, B>(
 > + Clone
        + Send
 where
-    T: ReceiptStore + Sync + Send + 'static,
+    T: ReceiptStore<SignedReceipt> + Sync + Send + 'static,
     B: Send,
 {
     move |request: Request<B>| {
@@ -91,7 +92,7 @@ mod tests {
         receipt::{
             checks::{Check, CheckError, CheckList, CheckResult},
             state::Checking,
-            ReceiptWithState,
+            ReceiptWithState, SignedReceipt,
         },
     };
     use test_assets::{
@@ -133,11 +134,11 @@ mod tests {
 
         struct MyCheck;
         #[async_trait::async_trait]
-        impl Check for MyCheck {
+        impl Check<SignedReceipt> for MyCheck {
             async fn check(
                 &self,
                 _: &tap_core::receipt::Context,
-                receipt: &ReceiptWithState<Checking>,
+                receipt: &ReceiptWithState<Checking, SignedReceipt>,
             ) -> CheckResult {
                 if receipt.signed_receipt().message.nonce == FAILED_NONCE {
                     Err(CheckError::Failed(anyhow::anyhow!("Failed")))

crates/service/src/middleware/auth/tap.rs

@@ -6,6 +6,7 @@ use std::{net::SocketAddr, sync::Arc, time::Duration};
 
 use anyhow::anyhow;
 use axum::{extract::Request, serve, ServiceExt};
+use clap::Parser;
 use indexer_config::{Config, DipsConfig, GraphNodeConfig, SubgraphConfig};
 use indexer_dips::{
     proto::graphprotocol::indexer::dips::agreement_service_server::{
@@ -19,14 +20,13 @@ use reqwest::Url;
 use tap_core::tap_eip712_domain;
 use tokio::{net::TcpListener, signal};
 use tower_http::normalize_path::NormalizePath;
+use tracing::info;
 
 use crate::{
     cli::Cli,
     database::{self, dips::PsqlAgreementStore},
     metrics::serve_metrics,
 };
-use clap::Parser;
-use tracing::info;
 
 mod release;
 mod router;

crates/service/src/service.rs

@@ -7,7 +7,7 @@ use indexer_allocation::Allocation;
 use indexer_monitor::EscrowAccounts;
 use receipt_store::{DatabaseReceipt, InnerContext};
 use sqlx::PgPool;
-use tap_core::receipt::checks::ReceiptCheck;
+use tap_core::receipt::{checks::ReceiptCheck, SignedReceipt};
 use thegraph_core::alloy::{primitives::Address, sol_types::Eip712Domain};
 use tokio::sync::{
     mpsc::{self, Sender},
@@ -48,7 +48,7 @@ impl IndexerTapContext {
         escrow_accounts: Receiver<EscrowAccounts>,
         timestamp_error_tolerance: Duration,
         receipt_max_value: u128,
-    ) -> Vec<ReceiptCheck> {
+    ) -> Vec<ReceiptCheck<SignedReceipt>> {
         vec![
             Arc::new(AllocationEligible::new(indexer_allocations)),
             Arc::new(SenderBalanceCheck::new(escrow_accounts)),

crates/service/src/tap.rs

@@ -8,7 +8,7 @@ use indexer_allocation::Allocation;
 use tap_core::receipt::{
     checks::{Check, CheckError, CheckResult},
     state::Checking,
-    ReceiptWithState,
+    ReceiptWithState, SignedReceipt,
 };
 use thegraph_core::alloy::primitives::Address;
 use tokio::sync::watch::Receiver;
@@ -25,11 +25,11 @@ impl AllocationEligible {
     }
 }
 #[async_trait::async_trait]
-impl Check for AllocationEligible {
+impl Check<SignedReceipt> for AllocationEligible {
     async fn check(
         &self,
         _: &tap_core::receipt::Context,
-        receipt: &ReceiptWithState<Checking>,
+        receipt: &ReceiptWithState<Checking, SignedReceipt>,
     ) -> CheckResult {
         let allocation_id = receipt.signed_receipt().message.allocation_id;
         if !self

crates/service/src/tap/checks/allocation_eligible.rs

@@ -11,7 +11,7 @@ use sqlx::{postgres::PgListener, PgPool};
 use tap_core::receipt::{
     checks::{Check, CheckError, CheckResult},
     state::Checking,
-    ReceiptWithState,
+    ReceiptWithState, SignedReceipt,
 };
 use thegraph_core::alloy::primitives::Address;
 
@@ -153,11 +153,11 @@ impl DenyListCheck {
 }
 
 #[async_trait::async_trait]
-impl Check for DenyListCheck {
+impl Check<SignedReceipt> for DenyListCheck {
     async fn check(
         &self,
         ctx: &tap_core::receipt::Context,
-        _: &ReceiptWithState<Checking>,
+        _: &ReceiptWithState<Checking, SignedReceipt>,
     ) -> CheckResult {
         let Sender(receipt_sender) = ctx
             .get::<Sender>()