Real-time Sync: Add push logic and run method

lib/core/src/model.rs

@@ -1615,3 +1615,19 @@ macro_rules! get_invoice_description {
         }
     };
 }
+
+#[macro_export]
+macro_rules! get_updated_fields {
+    ($($var:ident),* $(,)?) => {{
+        let mut options = Vec::new();
+        $(
+            if $var.is_some() {
+                options.push(stringify!($var).to_string());
+            }
+        )*
+        match options.len() > 0 {
+            true => Some(options),
+            false => None,
+        }
+    }};
+}

lib/core/src/persist/chain.rs

@@ -2,22 +2,25 @@ use std::collections::HashMap;
 
 use anyhow::Result;
 use boltz_client::swaps::boltz::{ChainSwapDetails, CreateChainResponse};
-use rusqlite::{named_params, params, Connection, Row};
+use rusqlite::{named_params, params, Connection, Row, TransactionBehavior};
 use sdk_common::bitcoin::hashes::{hex::ToHex, sha256, Hash};
 use serde::{Deserialize, Serialize};
 
-use crate::ensure_sdk;
 use crate::error::PaymentError;
 use crate::model::*;
 use crate::persist::{get_where_clause_state_in, Persister};
+use crate::sync::model::RecordType;
+use crate::{ensure_sdk, get_updated_fields};
 
 impl Persister {
     pub(crate) fn insert_chain_swap(&self, chain_swap: &ChainSwap) -> Result<()> {
-        let con = self.get_connection()?;
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(TransactionBehavior::Immediate)?;
 
         // There is a limit of 16 param elements in a single tuple in rusqlite,
         // so we split up the insert into two statements.
-        let mut stmt = con.prepare(
+        let id_hash = sha256::Hash::hash(chain_swap.id.as_bytes()).to_hex();
+        tx.execute(
             "
             INSERT INTO chain_swaps (
                 id,
@@ -38,28 +41,27 @@ impl Persister {
                 state
             )
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                &chain_swap.id,
+                &id_hash,
+                &chain_swap.direction,
+                &chain_swap.claim_address,
+                &chain_swap.lockup_address,
+                &chain_swap.timeout_block_height,
+                &chain_swap.preimage,
+                &chain_swap.payer_amount_sat,
+                &chain_swap.receiver_amount_sat,
+                &chain_swap.accept_zero_conf,
+                &chain_swap.create_response_json,
+                &chain_swap.claim_private_key,
+                &chain_swap.refund_private_key,
+                &chain_swap.claim_fees_sat,
+                &chain_swap.created_at,
+                &chain_swap.state,
+            ),
         )?;
-        let id_hash = sha256::Hash::hash(chain_swap.id.as_bytes()).to_hex();
-        _ = stmt.execute((
-            &chain_swap.id,
-            &id_hash,
-            &chain_swap.direction,
-            &chain_swap.claim_address,
-            &chain_swap.lockup_address,
-            &chain_swap.timeout_block_height,
-            &chain_swap.preimage,
-            &chain_swap.payer_amount_sat,
-            &chain_swap.receiver_amount_sat,
-            &chain_swap.accept_zero_conf,
-            &chain_swap.create_response_json,
-            &chain_swap.claim_private_key,
-            &chain_swap.refund_private_key,
-            &chain_swap.claim_fees_sat,
-            &chain_swap.created_at,
-            &chain_swap.state,
-        ))?;
 
-        con.execute(
+        tx.execute(
             "UPDATE chain_swaps
             SET
                 description = :description,
@@ -79,6 +81,10 @@ impl Persister {
             },
         )?;
 
+        Self::commit_outgoing(&tx, &chain_swap.id, RecordType::Chain, None)?;
+
+        tx.commit()?;
+
         Ok(())
     }
 
@@ -319,8 +325,10 @@ impl Persister {
         refund_tx_id: Option<&str>,
     ) -> Result<(), PaymentError> {
         // Do not overwrite server_lockup_tx_id, user_lockup_tx_id, claim_tx_id, refund_tx_id
-        let con: Connection = self.get_connection()?;
-        con.execute(
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(TransactionBehavior::Immediate)?;
+
+        tx.execute(
             "UPDATE chain_swaps
             SET
                 server_lockup_tx_id =
@@ -360,6 +368,11 @@ impl Persister {
             },
         )?;
 
+        let updated_fields = get_updated_fields!(server_lockup_tx_id);
+        Self::commit_outgoing(&tx, swap_id, RecordType::Chain, updated_fields)?;
+
+        tx.commit()?;
+
         Ok(())
     }
 }

lib/core/src/persist/receive.rs

@@ -2,20 +2,23 @@ use std::collections::HashMap;
 
 use anyhow::Result;
 use boltz_client::swaps::boltz::CreateReverseResponse;
-use rusqlite::{named_params, params, Connection, Row};
+use rusqlite::{named_params, params, Connection, Row, TransactionBehavior};
 use sdk_common::bitcoin::hashes::{hex::ToHex, sha256, Hash};
 use serde::{Deserialize, Serialize};
 
-use crate::ensure_sdk;
 use crate::error::PaymentError;
 use crate::model::*;
 use crate::persist::{get_where_clause_state_in, Persister};
+use crate::sync::model::RecordType;
+use crate::{ensure_sdk, get_updated_fields};
 
 impl Persister {
     pub(crate) fn insert_receive_swap(&self, receive_swap: &ReceiveSwap) -> Result<()> {
-        let con = self.get_connection()?;
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(TransactionBehavior::Immediate)?;
 
-        let mut stmt = con.prepare(
+        let id_hash = sha256::Hash::hash(receive_swap.id.as_bytes()).to_hex();
+        tx.execute(
             "
             INSERT INTO receive_swaps (
                 id,
@@ -34,26 +37,25 @@ impl Persister {
                 state
             )
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                &receive_swap.id,
+                id_hash,
+                &receive_swap.preimage,
+                &receive_swap.create_response_json,
+                &receive_swap.claim_private_key,
+                &receive_swap.invoice,
+                &receive_swap.payment_hash,
+                &receive_swap.payer_amount_sat,
+                &receive_swap.receiver_amount_sat,
+                &receive_swap.created_at,
+                &receive_swap.claim_fees_sat,
+                &receive_swap.mrh_address,
+                &receive_swap.mrh_script_pubkey,
+                &receive_swap.state,
+            ),
         )?;
-        let id_hash = sha256::Hash::hash(receive_swap.id.as_bytes()).to_hex();
-        _ = stmt.execute((
-            &receive_swap.id,
-            id_hash,
-            &receive_swap.preimage,
-            &receive_swap.create_response_json,
-            &receive_swap.claim_private_key,
-            &receive_swap.invoice,
-            &receive_swap.payment_hash,
-            &receive_swap.payer_amount_sat,
-            &receive_swap.receiver_amount_sat,
-            &receive_swap.created_at,
-            &receive_swap.claim_fees_sat,
-            &receive_swap.mrh_address,
-            &receive_swap.mrh_script_pubkey,
-            &receive_swap.state,
-        ))?;
 
-        con.execute(
+        tx.execute(
             "UPDATE receive_swaps
             SET
                 description = :description,
@@ -69,6 +71,10 @@ impl Persister {
             },
         )?;
 
+        Self::commit_outgoing(&tx, &receive_swap.id, RecordType::Receive, None)?;
+
+        tx.commit()?;
+
         Ok(())
     }
 
@@ -282,8 +288,10 @@ impl Persister {
         mrh_amount_sat: Option<u64>,
     ) -> Result<(), PaymentError> {
         // Do not overwrite claim_tx_id or lockup_tx_id
-        let con: Connection = self.get_connection()?;
-        con.execute(
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+
+        tx.execute(
             "UPDATE receive_swaps
             SET
                 claim_tx_id =
@@ -316,6 +324,12 @@ impl Persister {
             },
         )?;
 
+        // NOTE: Receive currently does not update any fields, bypassing the commit logic for now
+        // let updated_fields = None;
+        // Self::commit_outgoing(&tx, swap_id, RecordType::Receive, updated_fields)?;
+
+        tx.commit()?;
+
         Ok(())
     }
 }

lib/core/src/persist/send.rs

@@ -6,16 +6,19 @@ use rusqlite::{named_params, params, Connection, Row};
 use sdk_common::bitcoin::hashes::{hex::ToHex, sha256, Hash};
 use serde::{Deserialize, Serialize};
 
-use crate::ensure_sdk;
 use crate::error::PaymentError;
 use crate::model::*;
 use crate::persist::{get_where_clause_state_in, Persister};
+use crate::sync::model::RecordType;
+use crate::{ensure_sdk, get_updated_fields};
 
 impl Persister {
     pub(crate) fn insert_send_swap(&self, send_swap: &SendSwap) -> Result<()> {
-        let con = self.get_connection()?;
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
 
-        let mut stmt = con.prepare(
+        let id_hash = sha256::Hash::hash(send_swap.id.as_bytes()).to_hex();
+        tx.execute(
             "
             INSERT INTO send_swaps (
                 id,
@@ -34,24 +37,27 @@ impl Persister {
                 state
             )
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                &send_swap.id,
+                &id_hash,
+                &send_swap.invoice,
+                &send_swap.bolt12_offer,
+                &send_swap.payment_hash,
+                &send_swap.description,
+                &send_swap.payer_amount_sat,
+                &send_swap.receiver_amount_sat,
+                &send_swap.create_response_json,
+                &send_swap.refund_private_key,
+                &send_swap.lockup_tx_id,
+                &send_swap.refund_tx_id,
+                &send_swap.created_at,
+                &send_swap.state,
+            ),
         )?;
-        let id_hash = sha256::Hash::hash(send_swap.id.as_bytes()).to_hex();
-        _ = stmt.execute((
-            &send_swap.id,
-            &id_hash,
-            &send_swap.invoice,
-            &send_swap.bolt12_offer,
-            &send_swap.payment_hash,
-            &send_swap.description,
-            &send_swap.payer_amount_sat,
-            &send_swap.receiver_amount_sat,
-            &send_swap.create_response_json,
-            &send_swap.refund_private_key,
-            &send_swap.lockup_tx_id,
-            &send_swap.refund_tx_id,
-            &send_swap.created_at,
-            &send_swap.state,
-        ))?;
+
+        Self::commit_outgoing(&tx, &send_swap.id, RecordType::Send, None)?;
+
+        tx.commit()?;
 
         Ok(())
     }
@@ -207,8 +213,10 @@ impl Persister {
         refund_tx_id: Option<&str>,
     ) -> Result<(), PaymentError> {
         // Do not overwrite preimage, lockup_tx_id, refund_tx_id
-        let con: Connection = self.get_connection()?;
-        con.execute(
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+
+        tx.execute(
             "UPDATE send_swaps
             SET
                 preimage =
@@ -241,6 +249,11 @@ impl Persister {
             },
         )?;
 
+        let updated_fields = get_updated_fields!(preimage);
+        Self::commit_outgoing(&tx, swap_id, RecordType::Send, updated_fields)?;
+
+        tx.commit()?;
+
         Ok(())
     }