Merge pull request #46 from synonymdev/feat/bulk-upserts

feat: add bulk upsert methods for all database entities

Author: ovitrif

src/lib.rs

@@ -402,12 +402,66 @@ pub fn get_all_unique_tags() -> Result<Vec<String>, ActivityError> {
 }
 
 #[uniffi::export]
-pub fn insert_closed_channel(channel: ClosedChannelDetails) -> Result<(), ActivityError> {
+pub fn upsert_closed_channel(channel: ClosedChannelDetails) -> Result<(), ActivityError> {
     let mut guard = get_activity_db()?;
     let db = guard.activity_db.as_mut().ok_or(ActivityError::ConnectionError {
         error_details: "Database not initialized. Call init_db first.".to_string()
     })?;
-    db.insert_closed_channel(&channel)
+        db.upsert_closed_channel(&channel)
+}
+
+#[uniffi::export]
+pub fn upsert_closed_channels(channels: Vec<ClosedChannelDetails>) -> Result<(), ActivityError> {
+    let mut guard = get_activity_db()?;
+    let db = guard.activity_db.as_mut().ok_or(ActivityError::ConnectionError {
+        error_details: "Database not initialized. Call init_db first.".to_string()
+    })?;
+    db.upsert_closed_channels(&channels)
+}
+
+#[uniffi::export]
+pub fn upsert_onchain_activities(activities: Vec<OnchainActivity>) -> Result<(), ActivityError> {
+    let mut guard = get_activity_db()?;
+    let db = guard.activity_db.as_mut().ok_or(ActivityError::ConnectionError {
+        error_details: "Database not initialized. Call init_db first.".to_string()
+    })?;
+    db.upsert_onchain_activities(&activities)
+}
+
+#[uniffi::export]
+pub fn upsert_lightning_activities(activities: Vec<LightningActivity>) -> Result<(), ActivityError> {
+    let mut guard = get_activity_db()?;
+    let db = guard.activity_db.as_mut().ok_or(ActivityError::ConnectionError {
+        error_details: "Database not initialized. Call init_db first.".to_string()
+    })?;
+    db.upsert_lightning_activities(&activities)
+}
+
+#[uniffi::export]
+pub fn upsert_activities(activities: Vec<Activity>) -> Result<(), ActivityError> {
+    let mut guard = get_activity_db()?;
+    let db = guard.activity_db.as_mut().ok_or(ActivityError::ConnectionError {
+        error_details: "Database not initialized. Call init_db first.".to_string()
+    })?;
+
+    let mut onchain_list: Vec<OnchainActivity> = Vec::new();
+    let mut lightning_list: Vec<LightningActivity> = Vec::new();
+
+    for activity in activities {
+        match activity {
+            Activity::Onchain(a) => onchain_list.push(a),
+            Activity::Lightning(a) => lightning_list.push(a),
+        }
+    }
+
+    if !onchain_list.is_empty() {
+        db.upsert_onchain_activities(&onchain_list)?;
+    }
+    if !lightning_list.is_empty() {
+        db.upsert_lightning_activities(&lightning_list)?;
+    }
+
+    Ok(())
 }
 
 #[uniffi::export]
@@ -1320,6 +1374,62 @@ pub async fn blocktank_wipe_all() -> Result<(), BlocktankError> {
     }).await.unwrap()
 }
 
+#[uniffi::export]
+pub async fn upsert_info(info: IBtInfo) -> Result<(), BlocktankError> {
+    let rt = ensure_runtime();
+    rt.spawn(async move {
+        let cell = ASYNC_DB.get().ok_or(BlocktankError::ConnectionError {
+            error_details: "Database not initialized. Call init_db first.".to_string()
+        })?;
+        let guard = cell.lock().await;
+        let db = guard.blocktank_db.as_ref().ok_or(BlocktankError::ConnectionError {
+            error_details: "Database not initialized. Call init_db first.".to_string()
+        })?;
+        let external_info: rust_blocktank_client::IBtInfo = info.into();
+        db.upsert_info(&external_info).await
+    }).await.unwrap_or_else(|e| Err(BlocktankError::ConnectionError {
+        error_details: format!("Runtime error: {}", e)
+    }))
+}
+
+#[uniffi::export]
+pub async fn upsert_orders(orders: Vec<IBtOrder>) -> Result<(), BlocktankError> {
+    let rt = ensure_runtime();
+    rt.spawn(async move {
+        let cell = ASYNC_DB.get().ok_or(BlocktankError::ConnectionError {
+            error_details: "Database not initialized. Call init_db first.".to_string()
+        })?;
+        let guard = cell.lock().await;
+        let db = guard.blocktank_db.as_ref().ok_or(BlocktankError::ConnectionError {
+            error_details: "Database not initialized. Call init_db first.".to_string()
+        })?;
+
+        let external_orders: Vec<rust_blocktank_client::IBtOrder> = orders.into_iter().map(|order| order.into()).collect();
+        db.upsert_orders(&external_orders).await
+    }).await.unwrap_or_else(|e| Err(BlocktankError::ConnectionError {
+        error_details: format!("Runtime error: {}", e)
+    }))
+}
+
+#[uniffi::export]
+pub async fn upsert_cjit_entries(entries: Vec<ICJitEntry>) -> Result<(), BlocktankError> {
+    let rt = ensure_runtime();
+    rt.spawn(async move {
+        let cell = ASYNC_DB.get().ok_or(BlocktankError::ConnectionError {
+            error_details: "Database not initialized. Call init_db first.".to_string()
+        })?;
+        let guard = cell.lock().await;
+        let db = guard.blocktank_db.as_ref().ok_or(BlocktankError::ConnectionError {
+            error_details: "Database not initialized. Call init_db first.".to_string()
+        })?;
+
+        let external_entries: Vec<rust_blocktank_client::ICJitEntry> = entries.into_iter().map(|e| e.into()).collect();
+        db.upsert_cjit_entries(&external_entries).await
+    }).await.unwrap_or_else(|e| Err(BlocktankError::ConnectionError {
+        error_details: format!("Runtime error: {}", e)
+    }))
+}
+
 #[uniffi::export]
 pub async fn wipe_all_databases() -> Result<String, DbError> {
     let rt = ensure_runtime();

src/modules/activity/implementation.rs

@@ -80,6 +80,17 @@ const CREATE_CLOSED_CHANNELS_TABLE: &str = "
         channel_closure_reason TEXT NOT NULL
     )";
 
+const UPSERT_CLOSED_CHANNEL_SQL: &str = "
+    INSERT OR REPLACE INTO closed_channels (
+        channel_id, counterparty_node_id, funding_txo_txid, funding_txo_index,
+        channel_value_sats, closed_at, outbound_capacity_msat, inbound_capacity_msat,
+        counterparty_unspendable_punishment_reserve, unspendable_punishment_reserve,
+        forwarding_fee_proportional_millionths, forwarding_fee_base_msat,
+        channel_name, channel_closure_reason
+    ) VALUES (
+        ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14
+    )";
+
 const INDEX_STATEMENTS: &[&str] = &[
     // Activity indexes
     "CREATE INDEX IF NOT EXISTS idx_activities_type_timestamp ON activities(activity_type, timestamp DESC)",
@@ -379,6 +390,138 @@ impl ActivityDB {
         Ok(())
     }
 
+    pub fn upsert_onchain_activities(&mut self, activities: &[OnchainActivity]) -> Result<(), ActivityError> {
+        if activities.is_empty() {
+            return Ok(());
+        }
+
+        let tx = self.conn.transaction().map_err(|e| ActivityError::DataError {
+            error_details: format!("Failed to start transaction: {}", e),
+        })?;
+
+        {
+            let mut stmt_act = tx.prepare(
+                "INSERT OR REPLACE INTO activities (id, activity_type, tx_type, timestamp) VALUES (?1, 'onchain', ?2, ?3)"
+            ).map_err(|e| ActivityError::DataError {
+                error_details: format!("Failed to prepare activities statement: {}", e),
+            })?;
+            let mut stmt_onchain = tx.prepare(
+                "INSERT OR REPLACE INTO onchain_activity (
+                    id, tx_id, address, confirmed, value, fee, fee_rate, is_boosted,
+                    boost_tx_ids, is_transfer, does_exist, confirm_timestamp,
+                    channel_id, transfer_tx_id
+                ) VALUES (
+                    ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14
+                )"
+            ).map_err(|e| ActivityError::DataError {
+                error_details: format!("Failed to prepare onchain statement: {}", e),
+            })?;
+
+            for activity in activities {
+                if activity.id.is_empty() {
+                    return Err(ActivityError::DataError {
+                        error_details: "Activity ID cannot be empty".to_string(),
+                    });
+                }
+
+                stmt_act.execute((
+                    &activity.id,
+                    Self::payment_type_to_string(&activity.tx_type),
+                    activity.timestamp,
+                )).map_err(|e| ActivityError::InsertError {
+                    error_details: format!("Failed to upsert activities: {}", e),
+                })?;
+
+                let boost_tx_ids_str = activity.boost_tx_ids.join(",");
+                stmt_onchain.execute((
+                    &activity.id,
+                    &activity.tx_id,
+                    &activity.address,
+                    activity.confirmed,
+                    activity.value,
+                    activity.fee,
+                    activity.fee_rate,
+                    activity.is_boosted,
+                    &boost_tx_ids_str,
+                    activity.is_transfer,
+                    activity.does_exist,
+                    activity.confirm_timestamp,
+                    &activity.channel_id,
+                    &activity.transfer_tx_id,
+                )).map_err(|e| ActivityError::InsertError {
+                    error_details: format!("Failed to upsert onchain_activity: {}", e),
+                })?;
+            }
+        }
+
+        tx.commit().map_err(|e| ActivityError::DataError {
+            error_details: format!("Failed to commit transaction: {}", e),
+        })?;
+
+        Ok(())
+    }
+
+    pub fn upsert_lightning_activities(&mut self, activities: &[LightningActivity]) -> Result<(), ActivityError> {
+        if activities.is_empty() {
+            return Ok(());
+        }
+
+        let tx = self.conn.transaction().map_err(|e| ActivityError::DataError {
+            error_details: format!("Failed to start transaction: {}", e),
+        })?;
+
+        {
+            let mut stmt_act = tx.prepare(
+                "INSERT OR REPLACE INTO activities (id, activity_type, tx_type, timestamp) VALUES (?1, 'lightning', ?2, ?3)"
+            ).map_err(|e| ActivityError::DataError {
+                error_details: format!("Failed to prepare activities statement: {}", e),
+            })?;
+            let mut stmt_ln = tx.prepare(
+                "INSERT OR REPLACE INTO lightning_activity (
+                    id, invoice, value, status, fee, message, preimage
+                ) VALUES (
+                    ?1, ?2, ?3, ?4, ?5, ?6, ?7
+                )"
+            ).map_err(|e| ActivityError::DataError {
+                error_details: format!("Failed to prepare lightning statement: {}", e),
+            })?;
+
+            for activity in activities {
+                if activity.id.is_empty() {
+                    return Err(ActivityError::DataError {
+                        error_details: "Activity ID cannot be empty".to_string(),
+                    });
+                }
+
+                stmt_act.execute((
+                    &activity.id,
+                    Self::payment_type_to_string(&activity.tx_type),
+                    activity.timestamp,
+                )).map_err(|e| ActivityError::InsertError {
+                    error_details: format!("Failed to upsert activities: {}", e),
+                })?;
+
+                stmt_ln.execute((
+                    &activity.id,
+                    &activity.invoice,
+                    activity.value,
+                    Self::payment_state_to_string(&activity.status),
+                    activity.fee,
+                    &activity.message,
+                    &activity.preimage,
+                )).map_err(|e| ActivityError::InsertError {
+                    error_details: format!("Failed to upsert lightning_activity: {}", e),
+                })?;
+            }
+        }
+
+        tx.commit().map_err(|e| ActivityError::DataError {
+            error_details: format!("Failed to commit transaction: {}", e),
+        })?;
+
+        Ok(())
+    }
+
     pub fn get_activities(
         &self,
         filter: Option<ActivityFilter>,
@@ -1019,26 +1162,15 @@ impl ActivityDB {
         Ok(tags)
     }
 
-    pub fn insert_closed_channel(&mut self, channel: &ClosedChannelDetails) -> Result<(), ActivityError> {
+    pub fn upsert_closed_channel(&mut self, channel: &ClosedChannelDetails) -> Result<(), ActivityError> {
         if channel.channel_id.is_empty() {
             return Err(ActivityError::DataError {
                 error_details: "Channel ID cannot be empty".to_string(),
             });
         }
 
-        let sql = "
-            INSERT INTO closed_channels (
-                channel_id, counterparty_node_id, funding_txo_txid, funding_txo_index,
-                channel_value_sats, closed_at, outbound_capacity_msat, inbound_capacity_msat,
-                counterparty_unspendable_punishment_reserve, unspendable_punishment_reserve,
-                forwarding_fee_proportional_millionths, forwarding_fee_base_msat,
-                channel_name, channel_closure_reason
-            ) VALUES (
-                ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14
-            )";
-
         self.conn.execute(
-            sql,
+            UPSERT_CLOSED_CHANNEL_SQL,
             rusqlite::params![
                 &channel.channel_id,
                 &channel.counterparty_node_id,
@@ -1062,6 +1194,55 @@ impl ActivityDB {
         Ok(())
     }
 
+    pub fn upsert_closed_channels(&mut self, channels: &[ClosedChannelDetails]) -> Result<(), ActivityError> {
+        if channels.is_empty() {
+            return Ok(());
+        }
+
+        let tx = self.conn.transaction().map_err(|e| ActivityError::DataError {
+            error_details: format!("Failed to start transaction: {}", e),
+        })?;
+
+        {
+            let mut stmt = tx.prepare(UPSERT_CLOSED_CHANNEL_SQL).map_err(|e| ActivityError::DataError {
+                error_details: format!("Failed to prepare statement: {}", e),
+            })?;
+
+            for channel in channels {
+                if channel.channel_id.is_empty() {
+                    return Err(ActivityError::DataError {
+                        error_details: "Channel ID cannot be empty".to_string(),
+                    });
+                }
+
+                stmt.execute(rusqlite::params![
+                    &channel.channel_id,
+                    &channel.counterparty_node_id,
+                    &channel.funding_txo_txid,
+                    channel.funding_txo_index as i64,
+                    channel.channel_value_sats as i64,
+                    channel.closed_at as i64,
+                    channel.outbound_capacity_msat as i64,
+                    channel.inbound_capacity_msat as i64,
+                    channel.counterparty_unspendable_punishment_reserve as i64,
+                    channel.unspendable_punishment_reserve as i64,
+                    channel.forwarding_fee_proportional_millionths as i64,
+                    channel.forwarding_fee_base_msat as i64,
+                    &channel.channel_name,
+                    &channel.channel_closure_reason,
+                ]).map_err(|e| ActivityError::InsertError {
+                    error_details: format!("Failed to insert closed channel {}: {}", channel.channel_id, e),
+                })?;
+            }
+        }
+
+        tx.commit().map_err(|e| ActivityError::DataError {
+            error_details: format!("Failed to commit transaction: {}", e),
+        })?;
+
+        Ok(())
+    }
+
     pub fn get_closed_channel_by_id(&self, channel_id: &str) -> Result<Option<ClosedChannelDetails>, ActivityError> {
         let sql = "
             SELECT