feat: orchestrator for payment poller

Author: MarcosNicolau

aggregation_mode/db/src/orchestrator.rs

@@ -74,7 +74,7 @@ impl DbOrchestartor {
         }
 
         let nodes = connection_urls
-            .into_iter()
+            .iter()
             .map(|url| {
                 let pool = PgPoolOptions::new().max_connections(5).connect_lazy(url)?;
 

aggregation_mode/payments_poller/Cargo.toml

@@ -8,6 +8,7 @@ serde =  { workspace = true }
 serde_json = { workspace = true }
 serde_yaml = { workspace = true }
 aligned-sdk = { workspace = true }
+db = { workspace = true }
 tracing = { version = "0.1", features = ["log"] }
 tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
 actix-web = "4"

aggregation_mode/payments_poller/src/db.rs

@@ -1,8 +1,9 @@
-use sqlx::{postgres::PgPoolOptions, types::BigDecimal, Pool, Postgres};
+use db::{orchestrator::DbOrchestartor, retry::RetryConfig};
+use sqlx::types::BigDecimal;
 
 #[derive(Clone, Debug)]
 pub struct Db {
-    pool: Pool<Postgres>,
+    orchestartor: DbOrchestartor,
 }
 
 #[derive(Debug, Clone)]
@@ -11,14 +12,19 @@ pub enum DbError {
 }
 
 impl Db {
-    pub async fn try_new(connection_url: &str) -> Result<Self, DbError> {
-        let pool = PgPoolOptions::new()
-            .max_connections(5)
-            .connect(connection_url)
-            .await
-            .map_err(|e| DbError::ConnectError(e.to_string()))?;
+    pub async fn try_new(connection_urls: &[&str]) -> Result<Self, DbError> {
+        let orchestartor = DbOrchestartor::try_new(
+            connection_urls,
+            RetryConfig {
+                factor: 0.0,
+                max_delay_seconds: 0,
+                max_times: 0,
+                min_delay_millis: 0,
+            },
+        )
+        .map_err(|e| DbError::ConnectError(e.to_string()))?;
 
-        Ok(Self { pool })
+        Ok(Self { orchestartor })
     }
 
     pub async fn insert_payment_event(
@@ -29,18 +35,23 @@ impl Db {
         valid_until: &BigDecimal,
         tx_hash: &str,
     ) -> Result<(), sqlx::Error> {
-        sqlx::query(
-            "INSERT INTO payment_events (address, started_at, amount, valid_until, tx_hash)
-            VALUES ($1, $2, $3, $4, $5)
-            ON CONFLICT (tx_hash) DO NOTHING",
-        )
-        .bind(address.to_lowercase())
-        .bind(started_at)
-        .bind(amount)
-        .bind(valid_until)
-        .bind(tx_hash)
-        .execute(&self.pool)
-        .await
-        .map(|_| ())
+        self.orchestartor
+            .write(async |pool| {
+                sqlx::query(
+                    "INSERT INTO payment_events (address, started_at, amount, valid_until, tx_hash)
+                VALUES ($1, $2, $3, $4, $5)
+                ON CONFLICT (tx_hash) DO NOTHING",
+                )
+                .bind(address.to_lowercase())
+                .bind(started_at)
+                .bind(amount)
+                .bind(valid_until)
+                .bind(tx_hash)
+                .execute(&pool)
+                .await?;
+
+                Ok(())
+            })
+            .await
     }
 }

aggregation_mode/payments_poller/src/main.rs

@@ -26,7 +26,7 @@ async fn main() {
     let config = Config::from_file(&config_file_path).expect("Config is valid");
     tracing::info!("Config loaded");
 
-    let db = Db::try_new(&config.db_connection_url)
+    let db = Db::try_new(&[config.db_connection_url.as_str()])
         .await
         .expect("db to start");