refactor: add horizon support to tap-agent

Cargo.toml

@@ -4,6 +4,7 @@ members = [
     "crates/attestation",
     "crates/config",
     "crates/dips",
+    "crates/indexer-receipt",
     "crates/monitor",
     "crates/query",
     "crates/service",
@@ -84,12 +85,12 @@ tonic-build = "0.12.3"
 
 [patch.crates-io.tap_core]
 git = "https://github.com/semiotic-ai/timeline-aggregation-protocol"
-rev = "dbae001"
+rev = "e5546a6"
 
 [patch.crates-io.tap_aggregator]
 git = "https://github.com/semiotic-ai/timeline-aggregation-protocol"
-rev = "dbae001"
+rev = "e5546a6"
 
 [patch.crates-io.tap_graph]
 git = "https://github.com/semiotic-ai/timeline-aggregation-protocol"
-rev = "dbae001"
+rev = "e5546a6"

crates/indexer-receipt/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "indexer-receipt"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+tap_core.workspace = true
+tap_graph.workspace = true
+thegraph-core.workspace = true
+anyhow.workspace = true

crates/indexer-receipt/src/lib.rs

@@ -0,0 +1,175 @@
+// Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
+// SPDX-License-Identifier: Apache-2.0
+
+use anyhow::anyhow;
+use tap_core::{
+    receipt::{
+        rav::{Aggregate, AggregationError},
+        WithUniqueId, WithValueAndTimestamp,
+    },
+    signed_message::SignatureBytes,
+};
+use thegraph_core::alloy::{dyn_abi::Eip712Domain, primitives::Address, signers::Signature};
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum TapReceipt {
+    V1(tap_graph::SignedReceipt),
+    V2(tap_graph::v2::SignedReceipt),
+}
+
+impl Aggregate<TapReceipt> for tap_graph::ReceiptAggregateVoucher {
+    fn aggregate_receipts(
+        receipts: &[tap_core::receipt::ReceiptWithState<
+            tap_core::receipt::state::Checked,
+            TapReceipt,
+        >],
+        previous_rav: Option<tap_core::signed_message::Eip712SignedMessage<Self>>,
+    ) -> Result<Self, tap_core::receipt::rav::AggregationError> {
+        if receipts.is_empty() {
+            return Err(AggregationError::NoValidReceiptsForRavRequest);
+        }
+        let receipts: Vec<_> = receipts
+            .iter()
+            .map(|receipt| {
+                receipt
+                    .signed_receipt()
+                    .get_v1_receipt()
+                    .cloned()
+                    .ok_or(anyhow!("Receipt is not v1"))
+            })
+            .collect::<Result<_, _>>()
+            .map_err(AggregationError::Other)?;
+        let allocation_id = receipts[0].message.allocation_id;
+        tap_graph::ReceiptAggregateVoucher::aggregate_receipts(
+            allocation_id,
+            receipts.as_slice(),
+            previous_rav,
+        )
+    }
+}
+
+impl Aggregate<TapReceipt> for tap_graph::v2::ReceiptAggregateVoucher {
+    fn aggregate_receipts(
+        receipts: &[tap_core::receipt::ReceiptWithState<
+            tap_core::receipt::state::Checked,
+            TapReceipt,
+        >],
+        previous_rav: Option<tap_core::signed_message::Eip712SignedMessage<Self>>,
+    ) -> Result<Self, tap_core::receipt::rav::AggregationError> {
+        if receipts.is_empty() {
+            return Err(AggregationError::NoValidReceiptsForRavRequest);
+        }
+        let receipts: Vec<_> = receipts
+            .iter()
+            .map(|receipt| {
+                receipt
+                    .signed_receipt()
+                    .get_v2_receipt()
+                    .cloned()
+                    .ok_or(anyhow!("Receipt is not v2"))
+            })
+            .collect::<Result<_, _>>()
+            .map_err(AggregationError::Other)?;
+        let allocation_id = receipts[0].message.allocation_id;
+        let payer = receipts[0].message.payer;
+        let data_service = receipts[0].message.data_service;
+        let service_provider = receipts[0].message.service_provider;
+
+        tap_graph::v2::ReceiptAggregateVoucher::aggregate_receipts(
+            allocation_id,
+            payer,
+            data_service,
+            service_provider,
+            receipts.as_slice(),
+            previous_rav,
+        )
+    }
+}
+
+impl TapReceipt {
+    pub fn as_v1(self) -> Option<tap_graph::SignedReceipt> {
+        match self {
+            TapReceipt::V1(receipt) => Some(receipt),
+            _ => None,
+        }
+    }
+
+    pub fn as_v2(self) -> Option<tap_graph::v2::SignedReceipt> {
+        match self {
+            TapReceipt::V2(receipt) => Some(receipt),
+            _ => None,
+        }
+    }
+
+    pub fn get_v1_receipt(&self) -> Option<&tap_graph::SignedReceipt> {
+        match self {
+            TapReceipt::V1(receipt) => Some(receipt),
+            _ => None,
+        }
+    }
+
+    pub fn get_v2_receipt(&self) -> Option<&tap_graph::v2::SignedReceipt> {
+        match self {
+            TapReceipt::V2(receipt) => Some(receipt),
+            _ => None,
+        }
+    }
+
+    pub fn allocation_id(&self) -> Address {
+        match self {
+            TapReceipt::V1(receipt) => receipt.message.allocation_id,
+            TapReceipt::V2(receipt) => receipt.message.allocation_id,
+        }
+    }
+
+    pub fn signature(&self) -> Signature {
+        match self {
+            TapReceipt::V1(receipt) => receipt.signature,
+            TapReceipt::V2(receipt) => receipt.signature,
+        }
+    }
+
+    pub fn nonce(&self) -> u64 {
+        match self {
+            TapReceipt::V1(receipt) => receipt.message.nonce,
+            TapReceipt::V2(receipt) => receipt.message.nonce,
+        }
+    }
+
+    pub fn recover_signer(
+        &self,
+        domain_separator: &Eip712Domain,
+    ) -> Result<Address, tap_core::signed_message::Eip712Error> {
+        match self {
+            TapReceipt::V1(receipt) => receipt.recover_signer(domain_separator),
+            TapReceipt::V2(receipt) => receipt.recover_signer(domain_separator),
+        }
+    }
+}
+
+impl WithValueAndTimestamp for TapReceipt {
+    fn value(&self) -> u128 {
+        match self {
+            TapReceipt::V1(receipt) => receipt.value(),
+            TapReceipt::V2(receipt) => receipt.value(),
+        }
+    }
+
+    fn timestamp_ns(&self) -> u64 {
+        match self {
+            TapReceipt::V1(receipt) => receipt.timestamp_ns(),
+            TapReceipt::V2(receipt) => receipt.timestamp_ns(),
+        }
+    }
+}
+
+impl WithUniqueId for TapReceipt {
+    type Output = SignatureBytes;
+
+    fn unique_id(&self) -> Self::Output {
+        match self {
+            TapReceipt::V1(receipt) => receipt.unique_id(),
+            TapReceipt::V2(receipt) => receipt.unique_id(),
+        }
+    }
+}

crates/service/Cargo.toml

@@ -13,6 +13,7 @@ indexer-allocation = { path = "../allocation" }
 indexer-config = { path = "../config" }
 indexer-dips = { path = "../dips" }
 indexer-query = { path = "../query" }
+indexer-receipt = { path = "../indexer-receipt" }
 anyhow = { workspace = true }
 prometheus = { workspace = true }
 reqwest = { workspace = true }

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

@@ -20,7 +20,6 @@ use tap_core::{
     manager::{adapters::ReceiptStore, Manager},
     receipt::Context,
 };
-use tap_graph::ReceiptAggregateVoucher;
 use tower_http::auth::AsyncAuthorizeRequest;
 
 use crate::{
@@ -33,7 +32,7 @@ use crate::{
 ///
 /// Requires TapReceipt, MetricLabels and Arc<Context> extensions
 pub fn tap_receipt_authorize<T, B>(
-    tap_manager: Arc<Manager<T, TapReceipt, ReceiptAggregateVoucher>>,
+    tap_manager: Arc<Manager<T, TapReceipt>>,
     failed_receipt_metric: &'static prometheus::CounterVec,
 ) -> impl AsyncAuthorizeRequest<
     B,

crates/service/src/tap.rs

@@ -22,11 +22,10 @@ use crate::tap::checks::{
 };
 
 mod checks;
-mod receipt;
 mod receipt_store;
 
+pub use ::indexer_receipt::TapReceipt;
 pub use checks::value_check::AgoraQuery;
-pub use receipt::TapReceipt;
 
 pub type CheckingReceipt = ReceiptWithState<Checking, TapReceipt>;