feat: allocation id to collection id for horizon

Author: suchapalaver

crates/indexer-receipt/src/lib.rs

@@ -9,7 +9,11 @@ use tap_core::{
     },
     signed_message::SignatureBytes,
 };
-use thegraph_core::alloy::{dyn_abi::Eip712Domain, primitives::Address, signers::Signature};
+use thegraph_core::alloy::{
+    dyn_abi::Eip712Domain,
+    primitives::{Address, FixedBytes},
+    signers::Signature,
+};
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum TapReceipt {
@@ -70,13 +74,13 @@ impl Aggregate<TapReceipt> for tap_graph::v2::ReceiptAggregateVoucher {
             })
             .collect::<Result<_, _>>()
             .map_err(AggregationError::Other)?;
-        let allocation_id = receipts[0].message.allocation_id;
+        let collection_id = receipts[0].message.collection_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,
+            collection_id,
             payer,
             data_service,
             service_provider,
@@ -115,10 +119,17 @@ impl TapReceipt {
         }
     }
 
-    pub fn allocation_id(&self) -> Address {
+    pub fn allocation_id(&self) -> Option<Address> {
         match self {
-            TapReceipt::V1(receipt) => receipt.message.allocation_id,
-            TapReceipt::V2(receipt) => receipt.message.allocation_id,
+            TapReceipt::V1(receipt) => Some(receipt.message.allocation_id),
+            _ => None,
+        }
+    }
+
+    pub fn collection_id(&self) -> Option<FixedBytes<32>> {
+        match self {
+            TapReceipt::V2(receipt) => Some(receipt.message.collection_id),
+            _ => None,
         }
     }
 

crates/profiler/src/lib.rs

@@ -1,13 +1,17 @@
 // Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::fs::{self, File};
-use std::io::Write;
-use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::Arc;
-use std::thread;
-use std::time::{Duration, SystemTime};
+use std::{
+    fs::{self, File},
+    io::Write,
+    path::{Path, PathBuf},
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
+    thread,
+    time::{Duration, SystemTime},
+};
 
 use chrono::{DateTime, Utc};
 use pprof::protos::Message;

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

@@ -6,7 +6,7 @@ use std::collections::HashMap;
 use anyhow::anyhow;
 use indexer_allocation::Allocation;
 use tap_core::receipt::checks::{Check, CheckError, CheckResult};
-use thegraph_core::alloy::primitives::Address;
+use thegraph_core::{alloy::primitives::Address, CollectionId};
 use tokio::sync::watch::Receiver;
 
 use crate::tap::{CheckingReceipt, TapReceipt};
@@ -29,7 +29,12 @@ impl Check<TapReceipt> for AllocationEligible {
         _: &tap_core::receipt::Context,
         receipt: &CheckingReceipt,
     ) -> CheckResult {
-        let allocation_id = receipt.signed_receipt().allocation_id();
+        let allocation_id = match receipt.signed_receipt() {
+            TapReceipt::V1(receipt) => receipt.message.allocation_id,
+            TapReceipt::V2(receipt) => {
+                CollectionId::from(receipt.message.collection_id).as_address()
+            }
+        };
         if !self
             .indexer_allocations
             .borrow()

crates/service/src/tap/receipt_store.rs

@@ -7,7 +7,10 @@ use itertools::{Either, Itertools};
 use sqlx::{types::BigDecimal, PgPool};
 use tap_core::{manager::adapters::ReceiptStore, receipt::WithValueAndTimestamp};
 use thegraph_core::alloy::{hex::ToHexExt, sol_types::Eip712Domain};
-use tokio::{sync::mpsc::Receiver, sync::oneshot::Sender as OneShotSender, task::JoinHandle};
+use tokio::{
+    sync::{mpsc::Receiver, oneshot::Sender as OneShotSender},
+    task::JoinHandle,
+};
 use tokio_util::sync::CancellationToken;
 
 use super::{AdapterError, CheckingReceipt, IndexerTapContext, TapReceipt};
@@ -164,7 +167,7 @@ impl InnerContext {
         let receipts_len = receipts.len();
         let mut signers = Vec::with_capacity(receipts_len);
         let mut signatures = Vec::with_capacity(receipts_len);
-        let mut allocation_ids = Vec::with_capacity(receipts_len);
+        let mut collection_ids = Vec::with_capacity(receipts_len);
         let mut payers = Vec::with_capacity(receipts_len);
         let mut data_services = Vec::with_capacity(receipts_len);
         let mut service_providers = Vec::with_capacity(receipts_len);
@@ -175,7 +178,7 @@ impl InnerContext {
         for receipt in receipts {
             signers.push(receipt.signer_address);
             signatures.push(receipt.signature);
-            allocation_ids.push(receipt.allocation_id);
+            collection_ids.push(receipt.collection_id);
             payers.push(receipt.payer);
             data_services.push(receipt.data_service);
             service_providers.push(receipt.service_provider);
@@ -187,7 +190,7 @@ impl InnerContext {
             r#"INSERT INTO tap_horizon_receipts (
                 signer_address,
                 signature,
-                allocation_id,
+                collection_id,
                 payer,
                 data_service,
                 service_provider,
@@ -207,7 +210,7 @@ impl InnerContext {
             )"#,
             &signers,
             &signatures,
-            &allocation_ids,
+            &collection_ids,
             &payers,
             &data_services,
             &service_providers,
@@ -328,7 +331,7 @@ impl DbReceiptV1 {
 pub struct DbReceiptV2 {
     signer_address: String,
     signature: Vec<u8>,
-    allocation_id: String,
+    collection_id: String,
     payer: String,
     data_service: String,
     service_provider: String,
@@ -342,7 +345,7 @@ impl DbReceiptV2 {
         receipt: &tap_graph::v2::SignedReceipt,
         separator: &Eip712Domain,
     ) -> anyhow::Result<Self> {
-        let allocation_id = receipt.message.allocation_id.encode_hex();
+        let collection_id = receipt.message.collection_id.encode_hex();
         let payer = receipt.message.payer.encode_hex();
         let data_service = receipt.message.data_service.encode_hex();
         let service_provider = receipt.message.service_provider.encode_hex();
@@ -360,7 +363,7 @@ impl DbReceiptV2 {
         let nonce = BigDecimal::from(receipt.message.nonce);
         let value = BigDecimal::from(BigInt::from(receipt.value()));
         Ok(Self {
-            allocation_id,
+            collection_id,
             payer,
             data_service,
             service_provider,