0.5.4

kb_lib/src/db/queries/pool.rs

@@ -0,0 +1,64 @@
+// file: kb_lib/src/db/queries/pool.rs
+
+//! Queries for `kb_pools`.
+
+/// Inserts or updates one normalized pool row by address.
+pub async fn upsert_pool(
+    database: &crate::KbDatabase,
+    dto: &crate::KbPoolDto,
+) -> Result<i64, crate::KbError> {
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query(
+                r#"
+INSERT INTO kb_pools (
+    dex_id,
+    address,
+    pool_kind,
+    status,
+    first_seen_at,
+    updated_at
+)
+VALUES (?, ?, ?, ?, ?, ?)
+ON CONFLICT(address) DO UPDATE SET
+    dex_id = excluded.dex_id,
+    pool_kind = excluded.pool_kind,
+    status = excluded.status,
+    updated_at = excluded.updated_at
+                "#,
+            )
+            .bind(dto.dex_id)
+            .bind(dto.address.clone())
+            .bind(dto.pool_kind.to_i16())
+            .bind(dto.status.to_i16())
+            .bind(dto.first_seen_at.to_rfc3339())
+            .bind(dto.updated_at.to_rfc3339())
+            .execute(pool)
+            .await;
+            if let Err(error) = query_result {
+                return Err(crate::KbError::Db(format!(
+                    "cannot upsert kb_pools on sqlite: {}",
+                    error
+                )));
+            }
+            let id_result = sqlx::query_scalar::<sqlx::Sqlite, i64>(
+                r#"
+SELECT id
+FROM kb_pools
+WHERE address = ?
+LIMIT 1
+                "#,
+            )
+            .bind(dto.address.clone())
+            .fetch_one(pool)
+            .await;
+            match id_result {
+                Ok(id) => Ok(id),
+                Err(error) => Err(crate::KbError::Db(format!(
+                    "cannot fetch kb_pools id for address '{}' on sqlite: {}",
+                    dto.address, error
+                ))),
+            }
+        }
+    }
+}