feat(aggregation-mode): receipt endpoint in agg mode batcher

aggregation_mode/batcher/src/db.rs

@@ -14,6 +14,23 @@ pub enum DbError {
  ConnectError(String),
 }
 
+#[derive(Debug, Clone, sqlx::Type, serde::Serialize)]
+#[sqlx(type_name = "task_status")]
+#[sqlx(rename_all = "lowercase")]
+pub enum TaskStatus {
+ Pending,
+ Processing,
+ Verified,
+}
+
+#[derive(Debug, Clone, sqlx::FromRow, sqlx::Type, serde::Serialize)]
+pub struct Receipt {
+ pub status: TaskStatus,
+ pub merkle_path: Option<Vec<u8>>,
+ pub nonce: i64,
+ pub address: String,
+}
+
 impl Db {
  pub async fn try_new(connection_url: &str) -> Result<Self, DbError> {
  let pool = PgPoolOptions::new()
@@ -45,29 +62,66 @@ impl Db {
  .map(|res| res.flatten())
  }
 
+ pub async fn get_tasks_by_address_and_nonce(
+ &self,
+ address: &str,
+ nonce: i64,
+ ) -> Result<Vec<Receipt>, sqlx::Error> {
+ sqlx::query_as::<_, Receipt>(
+ "SELECT status,merkle_path,nonce,address FROM tasks
+ WHERE address = $1
+ AND nonce = $2
+ ORDER BY nonce DESC",
+ )
+ .bind(address.to_lowercase())
+ .bind(nonce)
+ .fetch_all(&self.pool)
+ .await
+ }
+
+ pub async fn get_tasks_by_address_with_limit(
+ &self,
+ address: &str,
+ limit: i64,
+ ) -> Result<Vec<Receipt>, sqlx::Error> {
+ sqlx::query_as::<_, Receipt>(
+ "SELECT status,merkle_path,nonce,address FROM tasks
+ WHERE address = $1
+ ORDER BY nonce DESC
+ LIMIT $2",
+ )
+ .bind(address.to_lowercase())
+ .bind(limit)
+ .fetch_all(&self.pool)
+ .await
+ }
+
  pub async fn insert_task(
  &self,
  address: &str,
  proving_system_id: i32,
  proof: &[u8],
  program_commitment: &[u8],
  merkle_path: Option<&[u8]>,
+ nonce: i64,
  ) -> Result<Uuid, sqlx::Error> {
  sqlx::query_scalar::<_, Uuid>(
  "INSERT INTO tasks (
  address,
  proving_system_id,
  proof,
  program_commitment,
- merkle_path
- ) VALUES ($1, $2, $3, $4, $5)
+ merkle_path,
+ nonce
+ ) VALUES ($1, $2, $3, $4, $5, $6)
  RETURNING task_id",
  )
  .bind(address.to_lowercase())
  .bind(proving_system_id)
  .bind(proof)
  .bind(program_commitment)
  .bind(merkle_path)
+ .bind(nonce)
  .fetch_one(&self.pool)
  .await
  }

aggregation_mode/batcher/src/server/http.rs

@@ -14,13 +14,13 @@ use sqlx::types::BigDecimal;
 
 use super::{
  helpers::format_merkle_path,
- types::{AppResponse, GetProofMerklePathQueryParams},
+ types::{AppResponse, GetReceiptsQueryParams},
 };
 
 use crate::{
  config::Config,
  db::Db,
- server::types::{SubmitProofRequestRisc0, SubmitProofRequestSP1},
+ server::types::{GetReceiptsResponse, SubmitProofRequestRisc0, SubmitProofRequestSP1},
  verifiers::{verify_sp1_proof, VerificationError},
 };
 
@@ -45,7 +45,7 @@ impl BatcherServer {
  App::new()
  .app_data(Data::new(state.clone()))
  .route("/nonce/{address}", web::get().to(Self::get_nonce))
- .route("/proof/merkle", web::get().to(Self::get_proof_merkle_path))
+ .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))
  })
@@ -56,20 +56,28 @@ impl BatcherServer {
  .expect("Server to never end");
  }
 
+ // Returns the nonce (number of submitted tasks) for a given address
  async fn get_nonce(req: HttpRequest) -> impl Responder {
- let Some(address) = req.match_info().get("address") else {
+ let Some(address_raw) = req.match_info().get("address") else {
  return HttpResponse::BadRequest()
  .json(AppResponse::new_unsucessfull("Missing address", 400));
  };
 
- // TODO: validate valid ethereum address
+ // 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.to_lowercase();
+
  let Some(state) = req.app_data::<Data<BatcherServer>>() else {
  return HttpResponse::InternalServerError()
  .json(AppResponse::new_unsucessfull("Internal server error", 500));
  };
 
  let state = state.get_ref();
- match state.db.count_tasks_by_address(address).await {
+ match state.db.count_tasks_by_address(&address).await {
  Ok(count) => HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!(
  {
  "nonce": count
@@ -80,6 +88,7 @@ impl BatcherServer {
  }
  }
 
+ // Posts an SP1 proof to the batcher, recovering the address from the signature
  async fn post_proof_sp1(
  req: HttpRequest,
  MultipartForm(data): MultipartForm<SubmitProofRequestSP1>,
@@ -172,6 +181,7 @@ impl BatcherServer {
  &proof_content,
  &vk_content,
  None,
+ data.nonce.0 as i64,
  )
  .await
  {
@@ -184,65 +194,89 @@ impl BatcherServer {
  }
 
  /// TODO: complete for risc0 (see `post_proof_sp1`)
+ // Posts a Risc0 proof to the batcher, recovering the address from the signature
  async fn post_proof_risc0(
  _req: HttpRequest,
  MultipartForm(_): MultipartForm<SubmitProofRequestRisc0>,
  ) -> impl Responder {
  HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({})))
  }
 
- async fn get_proof_merkle_path(
+ // Returns the last 100 receipt merkle proofs for the address received in the URL.
+ // In case of also receiving a nonce on the query param, it returns only the merkle proof for that nonce.
+ async fn get_receipts(
  req: HttpRequest,
- params: web::Query<GetProofMerklePathQueryParams>,
+ params: web::Query<GetReceiptsQueryParams>,
  ) -> impl Responder {
  let Some(state) = req.app_data::<Data<BatcherServer>>() else {
- return HttpResponse::InternalServerError()
- .json(AppResponse::new_unsucessfull("Internal server error", 500));
+ return HttpResponse::InternalServerError().json(AppResponse::new_unsucessfull(
+ "Internal server error: Failed to get app data",
+ 500,
+ ));
  };
 
  let state = state.get_ref();
 
- // TODO: maybe also accept proof commitment in query param
- let Some(id) = params.id.clone() else {
- return HttpResponse::BadRequest().json(AppResponse::new_unsucessfull(
- "Provide task `id` query param",
- 400,
- ));
- };
-
- if id.is_empty() {
- return HttpResponse::BadRequest().json(AppResponse::new_unsucessfull(
- "Proof id cannot be empty",
- 400,
- ));
+ if alloy::primitives::Address::from_str(params.address.clone().trim()).is_err() {
+ return HttpResponse::BadRequest()
+ .json(AppResponse::new_unsucessfull("Invalid address", 400));
  }
 
- let Ok(proof_id) = sqlx::types::Uuid::parse_str(&id) else {
-return HttpResponse::BadRequest()
- .json(AppResponse::new_unsucessfull("Proof id invalid uuid", 400));
+ let limit = match params.limit {
+Some(received_limit) => received_limit.min(100),
+None => 100,
  };
 
- let db_result = state.db.get_merkle_path_by_task_id(proof_id).await;
- let merkle_path = match db_result {
- Ok(Some(merkle_path)) => merkle_path,
- Ok(None) => {
- return HttpResponse::NotFound().json(AppResponse::new_unsucessfull(
- "Proof merkle path not found",
- 404,
- ))
- }
- Err(_) => {
- return HttpResponse::InternalServerError()
- .json(AppResponse::new_unsucessfull("Internal server error", 500));
- }
+ let address = params.address.to_lowercase();
+
+ let query = if let Some(nonce) = params.nonce {
+ state
+ .db
+ .get_tasks_by_address_and_nonce(&address, nonce)
+ .await
+ } else {
+ state
+ .db
+ .get_tasks_by_address_with_limit(&address, limit)
+ .await
  };
 
- match format_merkle_path(&merkle_path) {
- Ok(merkle_path) => {
- HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({
- "merkle_path": merkle_path
- })))
- }
+ let Ok(receipts) = query else {
+ return HttpResponse::InternalServerError().json(AppResponse::new_unsucessfull(
+ "Internal server error: Failed to get tasks by address and nonce",
+ 500,
+ ));
+ };
+
+ let responses: Result<Vec<GetReceiptsResponse>, String> = receipts
+ .into_iter()
+ .map(|receipt| {
+ let Some(merkle_path) = receipt.merkle_path else {
+ return Ok(GetReceiptsResponse {
+ status: receipt.status,
+ merkle_path: Vec::new(),
+ nonce: receipt.nonce,
+ address: receipt.address,
+ });
+ };
+
+ let Ok(formatted) = format_merkle_path(&merkle_path) else {
+ return Err("Error formatting merkle path".into());
+ };
+
+ Ok(GetReceiptsResponse {
+ status: receipt.status,
+ merkle_path: formatted,
+ nonce: receipt.nonce,
+ address: receipt.address,
+ })
+ })
+ .collect();
+
+ match responses {
+ Ok(resp) => HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({
+ "receipts": resp
+ }))),
  Err(_) => HttpResponse::InternalServerError()
  .json(AppResponse::new_unsucessfull("Internal server error", 500)),
  }

aggregation_mode/batcher/src/server/types.rs

@@ -2,6 +2,8 @@ use actix_multipart::form::{tempfile::TempFile, text::Text, MultipartForm};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
+use crate::db::TaskStatus;
+
 #[derive(Serialize, Deserialize)]
 pub(super) struct AppResponse {
  status: u16,
@@ -27,10 +29,14 @@ impl AppResponse {
  }
 }
 
-/// Query parameters accepted by `GET /proof/merkle`, containing an optional proof id.
+/// Query parameters accepted by `GET /receipts`. Requires an address, and accepts a nonce
+/// and a limit for the amount of tasks included in the query (the maximum value is 100).
+/// Note: The limit value will only be taken into account if nonce is None.
 #[derive(Deserialize, Clone)]
-pub(super) struct GetProofMerklePathQueryParams {
- pub id: Option<String>,
+pub(super) struct GetReceiptsQueryParams {
+ pub address: String,
+ pub nonce: Option<i64>,
+ pub limit: Option<i64>,
 }
 
 #[derive(Debug, MultipartForm)]
@@ -48,3 +54,11 @@ pub(super) struct SubmitProofRequestRisc0 {
  pub _program_image_id_hex: Text<String>,
  pub _signature_hex: Text<String>,
 }
+
+#[derive(Debug, Clone, sqlx::FromRow, sqlx::Type, serde::Serialize)]
+pub struct GetReceiptsResponse {
+ pub status: TaskStatus,
+ pub merkle_path: Vec<String>,
+ pub nonce: i64,
+ pub address: String,
+}

aggregation_mode/db/migrations/001_init.sql

@@ -7,7 +7,8 @@ CREATE TABLE tasks (
  proof BYTEA,
  program_commitment BYTEA,
  merkle_path BYTEA,
- status task_status DEFAULT 'pending'
+ status task_status DEFAULT 'pending',
+ nonce BIGINT NOT NULL
 );
 
 CREATE TABLE payment_events (