Merge remote-tracking branch 'origin/staging' into feat/support-multiple-dbs

Author: MarcosNicolau

Makefile

@@ -240,7 +240,7 @@ reset_last_aggregated_block:
 	@echo "Resetting last aggregated block..."
 	@echo '{"last_aggregated_block":0}' > config-files/proof-aggregator.last_aggregated_block.json
 
-AGGREGATION_MODE_SOURCES = $(wildcard ./aggregation_mode/Cargo.toml) $(wildcard ./aggregation_mode/src/**) $(wildcard ./aggregation_mode/aggregation_programs/risc0/Cargo.toml) $(wildcard ./aggregation_mode/aggregation_programs/risc0/src/**) $(wildcard ./aggregation_mode/aggregation_programs/sp1/Cargo.toml) $(wildcard ./aggregation_mode/aggregation_programs/sp1/src/**)
+AGGREGATION_MODE_SOURCES = $(wildcard ./aggregation_mode/Cargo.toml) $(wildcard ./aggregation_mode/proof_aggregator/Cargo.toml) $(wildcard ./aggregation_mode/proof_aggregator/src/**) $(wildcard ./aggregation_mode/proof_aggregator/aggregation_programs/risc0/Cargo.toml) $(wildcard ./aggregation_mode/proof_aggregator/aggregation_programs/risc0/src/**) $(wildcard ./aggregation_mode/proof_aggregator/aggregation_programs/sp1/Cargo.toml) $(wildcard ./aggregation_mode/proof_aggregator/aggregation_programs/sp1/src/**)
 
 ### All Dev proof aggregator receipts with no real proving
 ./aggregation_mode/target/release/proof_aggregator_dev: $(AGGREGATION_MODE_SOURCES)
@@ -341,6 +341,15 @@ agg_mode_gateway_send_sp1_proof:
 		--vk scripts/test_files/sp1/sp1_fibonacci_5_0_0_vk.bin \
 		--private-key "0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d"
 
+agg_mode_install_cli: ## Install the aggregation mode CLI
+	@cargo install --path aggregation_mode/cli
+
+agg_mode_task_sender_start: agg_mode_install_cli ## Send proofs to agg mode gateway
+	@. scripts/.agg_mode.task_sender.env && . ./scripts/agg_mode_send_sp1_proof_interval.sh
+
+agg_mode_get_quotas:
+	curl -X GET http://127.0.0.1:8089/quotas/0x70997970C51812dc3A010C7d01b50e0d17dc79C8
+
 __AGGREGATOR__: ## ____
 
 aggregator_start: ## Start the Aggregator. Parameters: ENVIRONMENT=<devnet|testnet|mainnet>, AGG_CONFIG_FILE
@@ -994,6 +1003,10 @@ upgrade_proof_aggregator: ## Upgrade ProofAggregator contract. Parameters: NETWO
 	@echo "Upgrading ProofAggregator Contract on $(NETWORK) network..."
 	@. contracts/scripts/.env.$(NETWORK) && . contracts/scripts/upgrade_proof_aggregator.sh
 
+deploy_agg_mode_payment_service:
+	@echo "Deploying Agg Mode Payment Service contract on $(NETWORK) network..."
+	@. contracts/scripts/.env.$(NETWORK) && . contracts/scripts/deploy_agg_mode_payment_service.sh
+
 __SP1_FFI__: ##
 build_sp1_macos:
 	@cd operator/sp1/lib && cargo build $(RELEASE_FLAG)

aggregation_mode/Cargo.lock

@@ -6,7 +6,7 @@ edition = "2021"
 [dependencies]
 serde =  { workspace = true }
 tokio = { version = "1"}
-sqlx = { version = "0.8", features = [ "runtime-tokio", "postgres", "migrate" , "uuid", "bigdecimal"] }
+sqlx = { version = "0.8", features = [ "runtime-tokio", "postgres", "migrate", "chrono", "uuid", "bigdecimal"] }
 tracing = { version = "0.1", features = ["log"] }
 
 [[bin]]

aggregation_mode/db/Cargo.toml

@@ -0,0 +1 @@
+ALTER TABLE tasks add COLUMN status_updated_at TIMESTAMPTZ DEFAULT now();

aggregation_mode/db/migrations/002_task_status_updated_at.sql

@@ -1,6 +1,9 @@
 use sqlx::{
     prelude::FromRow,
-    types::{BigDecimal, Uuid},
+    types::{
+        chrono::{DateTime, Utc},
+        BigDecimal, Uuid,
+    },
     Type,
 };
 
@@ -21,6 +24,7 @@ pub struct Task {
     pub program_commitment: Vec<u8>,
     pub merkle_path: Option<Vec<u8>>,
     pub status: TaskStatus,
+    pub status_updated_at: DateTime<Utc>,
 }
 
 #[derive(Debug, Clone, FromRow)]

aggregation_mode/db/src/types.rs

@@ -4,6 +4,7 @@ use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct Config {
+    pub ip: String,
     pub port: u16,
     pub db_connection_urls: Vec<String>,
     pub network: String,

aggregation_mode/gateway/src/config.rs

@@ -1,3 +1,5 @@
+use std::time::{SystemTime, UNIX_EPOCH};
+
 pub(super) fn format_merkle_path(bytes: &[u8]) -> Result<Vec<String>, String> {
     if bytes.is_empty() {
         return Ok(vec![]);
@@ -12,3 +14,17 @@ pub(super) fn format_merkle_path(bytes: &[u8]) -> Result<Vec<String>, String> {
         .map(|chunk| format!("0x{}", hex::encode(chunk)))
         .collect())
 }
+
+pub(crate) fn get_time_left_day_formatted() -> String {
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("Error al obtener el tiempo");
+
+    let seconds_remaining = 86400 - (now.as_secs() % 86400);
+
+    let hours = seconds_remaining / 3600;
+    let minutes = (seconds_remaining % 3600) / 60;
+    let seconds = seconds_remaining % 60;
+
+    format!("{hours}:{minutes}:{seconds} UTC")
+}

aggregation_mode/gateway/src/helpers.rs

@@ -22,6 +22,7 @@ use super::{
 use crate::{
     config::Config,
     db::Db,
+    helpers::get_time_left_day_formatted,
     types::{GetReceiptsResponse, SubmitProofRequestRisc0, SubmitProofRequestSP1},
     verifiers::{verify_sp1_proof, VerificationError},
 };
@@ -52,18 +53,25 @@ impl GatewayServer {
         HttpServer::new(move || {
             App::new()
                 .app_data(Data::new(state.clone()))
+                .route("/", web::get().to(Self::get_root))
                 .route("/nonce/{address}", web::get().to(Self::get_nonce))
                 .route("/receipts", web::get().to(Self::get_receipts))
                 .route("/proof/sp1", web::post().to(Self::post_proof_sp1))
                 .route("/proof/risc0", web::post().to(Self::post_proof_risc0))
+                .route("/quotas/{address}", web::get().to(Self::get_quotas))
         })
-        .bind(("127.0.0.1", port))
+        .bind((self.config.ip.as_str(), port))
         .expect("To bind socket correctly")
         .run()
         .await
         .expect("Server to never end");
     }
 
+    // Returns an OK response (code 200), no matters what receives in the request
+    async fn get_root(_req: HttpRequest) -> impl Responder {
+        HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({})))
+    }
+
     // Returns the nonce (number of submitted tasks) for a given address
     async fn get_nonce(req: HttpRequest) -> impl Responder {
         let Some(address_raw) = req.match_info().get("address") else {
@@ -147,8 +155,13 @@ impl GatewayServer {
         };
 
         if daily_tasks_by_address >= state.config.max_daily_proofs_per_user {
+            let formatted_time_left = get_time_left_day_formatted();
+
             return HttpResponse::InternalServerError().json(AppResponse::new_unsucessfull(
-                "Request denied: Query limit exceeded.",
+                format!(
+                    "Request denied: Query limit exceeded. Quotas renew in {formatted_time_left}"
+                )
+                .as_str(),
                 400,
             ));
         }
@@ -322,4 +335,73 @@ impl GatewayServer {
                 .json(AppResponse::new_unsucessfull("Internal server error", 500)),
         }
     }
+
+    async fn get_quotas(req: HttpRequest) -> impl Responder {
+        let Some(state) = req.app_data::<Data<GatewayServer>>() else {
+            return HttpResponse::InternalServerError().json(AppResponse::new_unsucessfull(
+                "Internal server error: Failed to get app data",
+                500,
+            ));
+        };
+
+        let state = state.get_ref();
+
+        let Some(address_raw) = req.match_info().get("address") else {
+            return HttpResponse::BadRequest()
+                .json(AppResponse::new_unsucessfull("Missing address", 400));
+        };
+
+        // Check that the address is a valid ethereum address
+        if alloy::primitives::Address::from_str(address_raw.trim()).is_err() {
+            return HttpResponse::BadRequest()
+                .json(AppResponse::new_unsucessfull("Invalid address", 400));
+        }
+
+        let address = address_raw.trim().to_lowercase();
+
+        let Ok(daily_tasks_by_address) = state.db.get_daily_tasks_by_address(&address).await else {
+            return HttpResponse::InternalServerError()
+                .json(AppResponse::new_unsucessfull("Internal server error", 500));
+        };
+
+        let formatted_time_left = get_time_left_day_formatted();
+
+        let now_epoch = match SystemTime::now().duration_since(UNIX_EPOCH) {
+            Ok(duration) => duration.as_secs(),
+            Err(_) => {
+                return HttpResponse::InternalServerError()
+                    .json(AppResponse::new_unsucessfull("Internal server error", 500));
+            }
+        };
+
+        let has_payment = match state
+            .db
+            .has_active_payment_event(
+                &address,
+                // safe unwrap the number comes from a valid u64 primitive
+                BigDecimal::from_str(&now_epoch.to_string()).unwrap(),
+            )
+            .await
+        {
+            Ok(result) => result,
+            Err(_) => {
+                return HttpResponse::InternalServerError()
+                    .json(AppResponse::new_unsucessfull("Internal server error", 500));
+            }
+        };
+
+        if has_payment {
+            HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({
+                "proofs_submitted": daily_tasks_by_address,
+                "quota_limit": state.config.max_daily_proofs_per_user,
+                "quota_remaining": (state.config.max_daily_proofs_per_user - daily_tasks_by_address),
+                "quota_resets_in": formatted_time_left.as_str()
+            })))
+        } else {
+            HttpResponse::Ok().json(AppResponse::new_unsucessfull(
+                "The address doesn't have an active subscription",
+                404,
+            ))
+        }
+    }
 }

aggregation_mode/gateway/src/http.rs

@@ -1,4 +1,4 @@
-use std::{fs::File, io::Read};
+use std::{fs::File, fs::OpenOptions, io::Read, io::Write};
 
 use serde::{Deserialize, Serialize};
 
@@ -7,6 +7,12 @@ pub struct Config {
     pub db_connection_urls: Vec<String>,
     pub eth_rpc_url: String,
     pub payment_service_address: String,
+    pub last_block_fetched_filepath: String,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct LastBlockFetched {
+    pub last_block_fetched: u64,
 }
 
 impl Config {
@@ -17,4 +23,30 @@ impl Config {
         let config: Config = serde_yaml::from_str(&contents)?;
         Ok(config)
     }
+
+    pub fn get_last_block_fetched(&self) -> Result<u64, Box<dyn std::error::Error>> {
+        let mut file = File::open(&self.last_block_fetched_filepath)?;
+        let mut contents = String::new();
+        file.read_to_string(&mut contents)?;
+        let lbf_struct: LastBlockFetched = serde_json::from_str(&contents)?;
+        Ok(lbf_struct.last_block_fetched)
+    }
+
+    pub fn update_last_block_fetched(
+        &self,
+        last_block_fetched: u64,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let last_block_fetched_struct = LastBlockFetched { last_block_fetched };
+
+        let mut file = OpenOptions::new()
+            .write(true)
+            .truncate(true)
+            .create(true)
+            .open(&self.last_block_fetched_filepath)?;
+
+        let content = serde_json::to_string(&last_block_fetched_struct)?;
+        file.write_all(content.as_bytes())?;
+
+        Ok(())
+    }
 }

aggregation_mode/payments_poller/src/config.rs

@@ -30,6 +30,13 @@ async fn main() {
         .await
         .expect("db to start");
 
-    let payment_poller = PaymentsPoller::new(db, config);
-    payment_poller.start().await;
+    let payments_poller = match PaymentsPoller::new(db, config) {
+        Ok(poller) => poller,
+        Err(err) => {
+            tracing::error!("Failed to create Payments Poller: {err:?}");
+            return;
+        }
+    };
+
+    payments_poller.start().await;
 }