feat: db types and tasks selection

Author: MarcosNicolau

aggregation_mode/Cargo.lock

@@ -13,6 +13,7 @@ serde_yaml = "0.9"
 alloy = { version = "1.1.1", features = ["default", "signer-keystore", "kzg"] }
 bincode = "1.3.3"
 aligned-sdk = { path = "../crates/sdk/" }
+db = { path = "./db" }
 sp1-sdk = "5.0.0"
 risc0-zkvm = { version = "3.0.3" }
 

aggregation_mode/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "agg_mode_db"
+name = "db"
 version = "0.1.0"
 edition = "2021"
 

aggregation_mode/db/Cargo.toml

@@ -0,0 +1 @@
+pub mod types;

aggregation_mode/db/src/lib.rs

@@ -0,0 +1,35 @@
+use sqlx::{
+    prelude::FromRow,
+    types::{BigDecimal, Uuid},
+    Type,
+};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Type)]
+#[sqlx(type_name = "task_status", rename_all = "lowercase")]
+pub enum TaskStatus {
+    Pending,
+    Running,
+    Done,
+    Failed,
+}
+
+#[derive(Debug, Clone, FromRow)]
+pub struct Task {
+    pub task_id: Uuid,
+    pub address: String,
+    pub proving_system_id: i32,
+    pub proof: Vec<u8>,
+    pub program_commitment: Vec<u8>,
+    pub merkle_path: Vec<u8>,
+    pub status: TaskStatus,
+}
+
+#[derive(Debug, Clone, FromRow)]
+pub struct Payment {
+    pub payment_event_id: Uuid,
+    pub address: String,
+    pub amount: i32,
+    pub started_at: BigDecimal,
+    pub valid_until: BigDecimal,
+    pub tx_hash: String,
+}

aggregation_mode/db/src/types.rs

@@ -10,6 +10,7 @@ serde_yaml = { workspace = true }
 alloy = { workspace = true }
 bincode = { workspace = true }
 aligned-sdk = { workspace = true }
+db = { workspace = true }
 
 tracing = { version = "0.1", features = ["log"] }
 tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }

aggregation_mode/proof_aggregator/Cargo.toml

@@ -12,7 +12,6 @@ fn main() {
             ],
             // We use Docker to generate a reproducible ELF that will be identical across all platforms
             // (https://docs.succinct.xyz/docs/sp1/writing-programs/compiling#production-builds)
-            docker: true,
             ..Default::default()
         }
     });
@@ -21,10 +20,7 @@ fn main() {
     // regardless of the machine or local environment, will produce the same ImageID
     let docker_options = DockerOptionsBuilder::default().build().unwrap();
     // Reference: https://github.com/risc0/risc0/blob/main/risc0/build/src/config.rs#L73-L90
-    let guest_options = GuestOptionsBuilder::default()
-        .use_docker(docker_options)
-        .build()
-        .unwrap();
+    let guest_options = GuestOptionsBuilder::default().build().unwrap();
 
     risc0_build::embed_methods_with_options(HashMap::from([(
         "risc0_aggregation_program",

aggregation_mode/proof_aggregator/build.rs

@@ -1,3 +1,4 @@
+use db::types::Task;
 use sqlx::{
     postgres::PgPoolOptions,
     types::{BigDecimal, Uuid},
@@ -12,6 +13,7 @@ pub struct Db {
 #[derive(Debug, Clone)]
 pub enum DbError {
     ConnectError(String),
+    Query(String),
 }
 
 impl Db {
@@ -25,13 +27,22 @@ impl Db {
         Ok(Self { pool })
     }
 
-    pub async fn get_tasks_and_mark_them_as_processed() {}
+    pub async fn get_pending_tasks_and_mark_them_as_processed(
+        &self,
+        limit: i64,
+    ) -> Result<Vec<Task>, DbError> {
+        sqlx::query_as::<_, Task>("SELECT * FROM tasks WHERE status = 'pending' LIMIT $1")
+            .bind(limit)
+            .fetch_all(&self.pool)
+            .await
+            .map_err(|e| DbError::Query(e.to_string()))
+    }
 
-    pub async fn mark_tasks_as_pending() {}
+    pub async fn mark_tasks_as_pending(&self) {}
 
-    pub async fn mark_tasks_as_processing() {}
+    pub async fn mark_tasks_as_processing(&self) {}
 
-    pub async fn mark_tasks_as_verified() {}
+    pub async fn mark_tasks_as_verified(&self) {}
 
-    pub async fn mark_tasks_as_submitted() {}
+    pub async fn mark_tasks_as_submitted(&self) {}
 }

aggregation_mode/proof_aggregator/src/backend/db.rs

@@ -53,7 +53,7 @@ pub struct ProofAggregator {
 }
 
 impl ProofAggregator {
-    pub fn new(config: Config) -> Self {
+    pub async fn new(config: Config) -> Self {
         let rpc_url = config.eth_rpc_url.parse().expect("RPC URL should be valid");
         let signer = LocalSigner::decrypt_keystore(
             config.ecdsa.private_key_store_path.clone(),
@@ -84,7 +84,9 @@ impl ProofAggregator {
                 .try_into()
                 .expect("Risc0 chunk aggregator image id must be 32 bytes");
 
-        let db = Db::try_new(self.config.db_connection_url).expect("To connect to db");
+        let db = Db::try_new(&config.db_connection_url)
+            .await
+            .expect("To connect to db");
 
         Self {
             engine,

aggregation_mode/proof_aggregator/src/backend/mod.rs

@@ -28,6 +28,6 @@ async fn main() {
     let config = Config::from_file(&config_file_path).expect("Config is valid");
     tracing::info!("Config loaded");
 
-    let mut proof_aggregator = ProofAggregator::new(config);
+    let mut proof_aggregator = ProofAggregator::new(config).await;
     proof_aggregator.start().await;
 }