feat: gRPC service implementation

Author: pedrohba1

tap_aggregator/Cargo.toml

@@ -35,7 +35,14 @@ axum = { version = "0.7.5", features = [
 futures-util = "0.3.28"
 lazy_static = "1.4.0"
 ruint = "1.10.1"
-tower = { version = "0.4", features = ["util"] }
+tower = { version = "0.4", features = ["util", "steer"] }
+tonic = { version = "0.12.3", features = ["transport", "zstd"] }
+prost = "0.13.3"
+hyper = { version = "1", features = ["full"] }
+
+[build-dependencies]
+tonic-build = "0.12.3"
+
 
 [dev-dependencies]
 jsonrpsee = { workspace = true, features = ["http-client", "jsonrpsee-core"] }

tap_aggregator/build.rs

@@ -0,0 +1,11 @@
+// Copyright 2023-, Semiotic AI, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    println!("Running build.rs...");
+    let out_dir = std::env::var("OUT_DIR").expect("OUT_DIR not set by Cargo");
+    println!("OUT_DIR: {}", out_dir); // This should print the output directory
+
+    tonic_build::compile_protos("./proto/tap_aggregator.proto")?;
+    Ok(())
+}

tap_aggregator/proto/tap_aggregator.proto

@@ -1,3 +1,6 @@
+// Copyright 2023-, Semiotic AI, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
 syntax = "proto3";
 package tap_aggregator.v1;
 
@@ -42,4 +45,4 @@ message Uint128 {
   uint64 high = 1;
   // Lowest 64 bits of a 128 bit number.
   uint64 low = 2;
-}
+}

tap_aggregator/src/tap_aggregator.rs renamed to tap_aggregator/src/grpc.rs

@@ -1,3 +1,6 @@
+// Copyright 2023-, Semiotic AI, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
 use anyhow::anyhow;
 use tap_core::signed_message::EIP712SignedMessage;
 
@@ -28,6 +31,26 @@ impl TryFrom<SignedReceipt> for tap_core::receipt::SignedReceipt {
     }
 }
 
+impl From<tap_core::receipt::Receipt> for Receipt {
+    fn from(value: tap_core::receipt::Receipt) -> Self {
+        Self {
+            allocation_id: value.allocation_id.as_slice().to_vec(),
+            timestamp_ns: value.timestamp_ns,
+            nonce: value.nonce,
+            value: Some(value.value.into()),
+        }
+    }
+}
+
+impl From<tap_core::receipt::SignedReceipt> for SignedReceipt {
+    fn from(value: tap_core::receipt::SignedReceipt) -> Self {
+        Self {
+            message: Some(value.message.into()),
+            signature: value.signature.as_bytes().to_vec(),
+        }
+    }
+}
+
 impl TryFrom<SignedRav> for EIP712SignedMessage<tap_core::rav::ReceiptAggregateVoucher> {
     type Error = anyhow::Error;
     fn try_from(voucher: SignedRav) -> Result<Self, Self::Error> {
@@ -87,3 +110,26 @@ impl From<u128> for Uint128 {
         Self { high, low }
     }
 }
+
+impl RavRequest {
+    pub fn new(
+        receipts: Vec<tap_core::receipt::SignedReceipt>,
+        previous_rav: Option<tap_core::rav::SignedRAV>,
+    ) -> Self {
+        Self {
+            receipts: receipts.into_iter().map(Into::into).collect(),
+            previous_rav: previous_rav.map(Into::into),
+        }
+    }
+}
+
+impl RavResponse {
+    pub fn signed_rav(mut self) -> anyhow::Result<tap_core::rav::SignedRAV> {
+        let signed_rav: tap_core::rav::SignedRAV = self
+            .rav
+            .take()
+            .ok_or(anyhow!("Couldn't find rav"))?
+            .try_into()?;
+        Ok(signed_rav)
+    }
+}

tap_aggregator/src/lib.rs

@@ -4,6 +4,7 @@
 pub mod aggregator;
 pub mod api_versioning;
 pub mod error_codes;
+pub mod grpc;
 pub mod jsonrpsee_helpers;
 pub mod metrics;
 pub mod server;

tap_aggregator/src/main.rs

@@ -3,22 +3,15 @@
 
 #![doc = include_str!("../README.md")]
 
-use std::borrow::Cow;
-use std::collections::HashSet;
-use std::str::FromStr;
-
-use alloy::dyn_abi::Eip712Domain;
-use alloy::primitives::Address;
-use alloy::primitives::FixedBytes;
-use alloy::signers::local::PrivateKeySigner;
+use std::{collections::HashSet, str::FromStr};
+
+use alloy::{dyn_abi::Eip712Domain, primitives::Address, signers::local::PrivateKeySigner};
 use anyhow::Result;
 use clap::Parser;
-use ruint::aliases::U256;
-use tokio::signal::unix::{signal, SignalKind};
-
 use log::{debug, info};
-use tap_aggregator::metrics;
-use tap_aggregator::server;
+use tap_core::tap_eip712_domain;
+
+use tap_aggregator::{metrics, server};
 
 #[derive(Parser, Debug)]
 #[command(author, version, about, long_about = None)]
@@ -126,37 +119,22 @@ async fn main() -> Result<()> {
     .await?;
     info!("Server started. Listening on port {}.", args.port);
 
-    // Have tokio wait for SIGTERM or SIGINT.
-    let mut signal_sigint = signal(SignalKind::interrupt())?;
-    let mut signal_sigterm = signal(SignalKind::terminate())?;
-    tokio::select! {
-        _ = signal_sigint.recv() => debug!("Received SIGINT."),
-        _ = signal_sigterm.recv() => debug!("Received SIGTERM."),
-    }
+    let _ = handle.await;
 
     // If we're here, we've received a signal to exit.
     info!("Shutting down...");
-
-    // Stop the server and wait for it to finish gracefully.
-    handle.stop()?;
-    handle.stopped().await;
-
-    debug!("Goodbye!");
     Ok(())
 }
 
 fn create_eip712_domain(args: &Args) -> Result<Eip712Domain> {
     // Transfrom the args into the types expected by Eip712Domain::new().
 
     // Transform optional strings into optional Cow<str>.
-    let name = args.domain_name.clone().map(Cow::Owned);
-    let version = args.domain_version.clone().map(Cow::Owned);
-
     // Transform optional strings into optional U256.
     if args.domain_chain_id.is_some() {
         debug!("Parsing domain chain ID...");
     }
-    let chain_id: Option<U256> = args
+    let chain_id: Option<u64> = args
         .domain_chain_id
         .as_ref()
         .map(|s| s.parse())
@@ -165,17 +143,13 @@ fn create_eip712_domain(args: &Args) -> Result<Eip712Domain> {
     if args.domain_salt.is_some() {
         debug!("Parsing domain salt...");
     }
-    let salt: Option<FixedBytes<32>> = args.domain_salt.as_ref().map(|s| s.parse()).transpose()?;
 
     // Transform optional strings into optional Address.
     let verifying_contract: Option<Address> = args.domain_verifying_contract;
 
     // Create the EIP-712 domain separator.
-    Ok(Eip712Domain::new(
-        name,
-        version,
-        chain_id,
-        verifying_contract,
-        salt,
+    Ok(tap_eip712_domain(
+        chain_id.unwrap_or(1),
+        verifying_contract.unwrap_or_default(),
     ))
 }