0.5.5

2026-04-24 12:01:35 +02:00
parent 54778373b8
commit 0ad6145091
23 changed files with 1653 additions and 5 deletions
--- a/kb_lib/src/db/queries/liquidity_event.rs
+++ b/kb_lib/src/db/queries/liquidity_event.rs
@@ -0,0 +1,167 @@
+// file: kb_lib/src/db/queries/liquidity_event.rs
+
+//! Queries for `kb_liquidity_events`.
+
+/// Inserts or updates one normalized liquidity event row.
+pub async fn upsert_liquidity_event(
+    database: &crate::KbDatabase,
+    dto: &crate::KbLiquidityEventDto,
+) -> Result<i64, crate::KbError> {
+    let slot_i64 = match dto.slot {
+        Some(slot) => {
+            let slot_result = i64::try_from(slot);
+            match slot_result {
+                Ok(slot) => Some(slot),
+                Err(error) => {
+                    return Err(crate::KbError::Db(format!(
+                        "cannot convert liquidity event slot '{}' to i64: {}",
+                        slot, error
+                    )));
+                }
+            }
+        }
+        None => None,
+    };
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query(
+                r#"
+INSERT INTO kb_liquidity_events (
+    dex_id,
+    pool_id,
+    pair_id,
+    signature,
+    instruction_index,
+    slot,
+    event_kind,
+    actor_wallet,
+    base_token_id,
+    quote_token_id,
+    lp_token_id,
+    base_amount,
+    quote_amount,
+    lp_amount,
+    executed_at
+)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ON CONFLICT(signature, instruction_index) DO UPDATE SET
+    dex_id = excluded.dex_id,
+    pool_id = excluded.pool_id,
+    pair_id = excluded.pair_id,
+    slot = excluded.slot,
+    event_kind = excluded.event_kind,
+    actor_wallet = excluded.actor_wallet,
+    base_token_id = excluded.base_token_id,
+    quote_token_id = excluded.quote_token_id,
+    lp_token_id = excluded.lp_token_id,
+    base_amount = excluded.base_amount,
+    quote_amount = excluded.quote_amount,
+    lp_amount = excluded.lp_amount,
+    executed_at = excluded.executed_at
+                "#,
+            )
+            .bind(dto.dex_id)
+            .bind(dto.pool_id)
+            .bind(dto.pair_id)
+            .bind(dto.signature.clone())
+            .bind(dto.instruction_index)
+            .bind(slot_i64)
+            .bind(dto.event_kind.to_i16())
+            .bind(dto.actor_wallet.clone())
+            .bind(dto.base_token_id)
+            .bind(dto.quote_token_id)
+            .bind(dto.lp_token_id)
+            .bind(dto.base_amount.clone())
+            .bind(dto.quote_amount.clone())
+            .bind(dto.lp_amount.clone())
+            .bind(dto.executed_at.to_rfc3339())
+            .execute(pool)
+            .await;
+            if let Err(error) = query_result {
+                return Err(crate::KbError::Db(format!(
+                    "cannot upsert kb_liquidity_events on sqlite: {}",
+                    error
+                )));
+            }
+            let id_result = sqlx::query_scalar::<sqlx::Sqlite, i64>(
+                r#"
+SELECT id
+FROM kb_liquidity_events
+WHERE signature = ? AND instruction_index = ?
+LIMIT 1
+                "#,
+            )
+            .bind(dto.signature.clone())
+            .bind(dto.instruction_index)
+            .fetch_one(pool)
+            .await;
+            match id_result {
+                Ok(id) => Ok(id),
+                Err(error) => Err(crate::KbError::Db(format!(
+                    "cannot fetch kb_liquidity_events id for signature '{}' and instruction_index '{}' on sqlite: {}",
+                    dto.signature, dto.instruction_index, error
+                ))),
+            }
+        }
+    }
+}
+
+/// Lists recent liquidity events ordered from newest to oldest.
+pub async fn list_recent_liquidity_events(
+    database: &crate::KbDatabase,
+    limit: u32,
+) -> Result<std::vec::Vec<crate::KbLiquidityEventDto>, crate::KbError> {
+    if limit == 0 {
+        return Ok(std::vec::Vec::new());
+    }
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query_as::<sqlx::Sqlite, crate::KbLiquidityEventEntity>(
+                r#"
+SELECT
+    id,
+    dex_id,
+    pool_id,
+    pair_id,
+    signature,
+    instruction_index,
+    slot,
+    event_kind,
+    actor_wallet,
+    base_token_id,
+    quote_token_id,
+    lp_token_id,
+    base_amount,
+    quote_amount,
+    lp_amount,
+    executed_at
+FROM kb_liquidity_events
+ORDER BY id DESC
+LIMIT ?
+                "#,
+            )
+            .bind(i64::from(limit))
+            .fetch_all(pool)
+            .await;
+            let entities = match query_result {
+                Ok(entities) => entities,
+                Err(error) => {
+                    return Err(crate::KbError::Db(format!(
+                        "cannot list kb_liquidity_events on sqlite: {}",
+                        error
+                    )));
+                }
+            };
+            let mut dtos = std::vec::Vec::new();
+            for entity in entities {
+                let dto_result = crate::KbLiquidityEventDto::try_from(entity);
+                let dto = match dto_result {
+                    Ok(dto) => dto,
+                    Err(error) => return Err(error),
+                };
+                dtos.push(dto);
+            }
+            Ok(dtos)
+        }
+    }
+}