Split into two functions

Author: dangeross

crates/breez-sdk/core/src/sdk.rs

@@ -18,9 +18,9 @@ use breez_sdk_common::{
 };
 use spark_wallet::{
     ExitSpeed, InvoiceDescription, ListTokenTransactionsRequest, SparkAddress, SparkWallet,
-    TokenInputs, TransferTokenOutput, WalletEvent, WalletTransfer,
+    SspUserRequest, TokenInputs, TransferTokenOutput, WalletEvent, WalletTransfer,
 };
-use std::{str::FromStr, sync::Arc};
+use std::{collections::HashMap, str::FromStr, sync::Arc};
 use tracing::{error, info, trace};
 use web_time::{Duration, SystemTime};
 
@@ -68,6 +68,11 @@ enum SyncType {
     PaymentsOnly,
 }
 
+struct AddressTransactionsWithSspUserRequests {
+    address_transactions: Vec<sparkscan::types::AddressTransaction>,
+    ssp_user_requests: HashMap<String, SspUserRequest>,
+}
+
 /// `BreezSDK` is a wrapper around `SparkSDK` that provides a more structured API
 /// with request/response objects and comprehensive error handling.
 #[derive(Clone)]
@@ -337,7 +342,7 @@ impl BreezSdk {
 
         self.sync_balances_to_storage(&object_repository).await?;
         self.sync_pending_payments().await?;
-        self.sync_payments_to_storage(&object_repository, true, None)
+        self.sync_payments_head_to_storage(&object_repository)
             .await?;
 
         Ok(())
@@ -364,10 +369,40 @@ impl BreezSdk {
         Ok(())
     }
 
-    async fn sync_payments_tail_to_storage(&self) -> Result<(), SdkError> {
-        let object_repository = ObjectCacheRepository::new(self.storage.clone());
-        self.sync_payments_to_storage(&object_repository, false, Some(PAYMENT_SYNC_TAIL_MAX_PAGES))
-            .await
+    async fn fetch_address_transactions_with_ssp_user_requests(
+        &self,
+        legacy_spark_address: &str,
+        offset: u64,
+    ) -> Result<AddressTransactionsWithSspUserRequests, SdkError> {
+        let response = sparkscan::Client::new(&self.config.sparkscan_api_url)
+            .get_address_transactions_v1_address_address_transactions_get()
+            .network(sparkscan::types::Network::from(self.config.network))
+            .address(legacy_spark_address)
+            .offset(offset)
+            .limit(PAYMENT_SYNC_BATCH_SIZE)
+            .send()
+            .await?;
+        let address_transactions = response.data.clone();
+        let ssp_transfer_types = [
+            sparkscan::types::AddressTransactionType::BitcoinDeposit,
+            sparkscan::types::AddressTransactionType::BitcoinWithdrawal,
+            sparkscan::types::AddressTransactionType::LightningPayment,
+        ];
+        let ssp_user_requests = self
+            .spark_wallet
+            .query_ssp_user_requests(
+                address_transactions
+                    .iter()
+                    .filter(|tx| ssp_transfer_types.contains(&tx.type_))
+                    .map(|tx| tx.id.clone())
+                    .collect(),
+            )
+            .await?;
+
+        Ok(AddressTransactionsWithSspUserRequests {
+            address_transactions,
+            ssp_user_requests,
+        })
     }
 
     /// Synchronizes payments from the Spark network to persistent storage using the Sparkscan API.
@@ -376,100 +411,57 @@ impl BreezSdk {
     /// `last_synced_payment_id`. If `max_pages` is reached or we get an error response from the Sparkscan
     /// API, it will update the `next_head_offset` for the next sync. This allows us to gradually sync the
     /// head up to the `last_synced_payment_id` in multiple sync cycles.
-    ///
-    /// When syncing from tail, it will start from the count of completed payments in storage and page for
-    /// a maximum of `PAYMENT_SYNC_TAIL_MAX_PAGES` for one sync cycle. Once it reaches the end page,
-    /// it will update `tail_synced` in the sync info and the tail will no longer be synced in future cycles.
-    #[allow(clippy::too_many_lines)]
-    async fn sync_payments_to_storage(
+    async fn sync_payments_head_to_storage(
         &self,
         object_repository: &ObjectCacheRepository,
-        from_head: bool,
-        max_pages: Option<u64>,
     ) -> Result<(), SdkError> {
-        info!("Syncing payments to storage, from_head = {from_head}, max_pages = {max_pages:?}");
+        info!("Syncing payments head to storage");
         let cached_sync_info = object_repository
             .fetch_sync_info()
             .await?
             .unwrap_or_default();
-        if cached_sync_info.tail_synced && !from_head {
-            info!("Payments tail already synced, skipping");
-            return Ok(());
-        }
-        let last_synced_id = cached_sync_info.last_synced_payment_id;
-        let mut max_pages = max_pages.unwrap_or(u64::MAX);
-        let mut found_last_synced = false;
-        let mut more_to_sync = true;
-
-        if last_synced_id.is_none() && from_head {
-            // There is no cached last synced payment id, limit the number of pages when syncing
-            // from head. Then let the rest of the payments be synced by the tail sync.
-            // Set `found_last_synced` to true to store the first payment as the last synced payment id.
-            info!("No last synced payment id found syncing from head, setting max_pages to 1");
-            max_pages = 1;
-            found_last_synced = true;
-        }
-
         // TODO: use new spark address format once sparkscan supports it
         let legacy_spark_address = self
             .spark_wallet
             .get_spark_address()?
             .to_string_with_hrp_legacy();
-
-        let mut next_offset = if from_head {
-            // Sync from the next head offset in case we didn't finish syncing the head last time
-            cached_sync_info.next_head_offset
+        let last_synced_id = cached_sync_info.last_synced_payment_id;
+        let (max_pages, mut head_synced) = if last_synced_id.is_some() {
+            (u64::MAX, false)
         } else {
-            // Count the completed payments and start the sync from there
-            self.storage
-                .list_payments(None, None, Some(PaymentStatus::Completed))
-                .await?
-                .len() as u64
+            info!("No last synced payment id found syncing from head, setting max_pages to 1");
+            // There is no cached last synced payment id, limit the number of pages when syncing
+            // from head. Then let the rest of the payments be synced by the tail sync.
+            // Set `head_synced` to true to store the first payment as the last synced payment id.
+            (1, true)
         };
-
+        // Sync from the next head offset in case we didn't finish syncing the head last time
+        let mut next_offset = cached_sync_info.next_head_offset;
         let mut payments_to_sync = Vec::new();
         'page_loop: for page in 1..=max_pages {
             info!(
                 "Fetching address transactions, offset = {next_offset}, page = {page}/{max_pages}"
             );
-            let response = sparkscan::Client::new(&self.config.sparkscan_api_url)
-                .get_address_transactions_v1_address_address_transactions_get()
-                .network(sparkscan::types::Network::from(self.config.network))
-                .address(legacy_spark_address.to_string())
-                .offset(next_offset)
-                .limit(PAYMENT_SYNC_BATCH_SIZE)
-                .send()
-                .await;
-            let address_transactions = match &response {
-                Ok(response) => &response.data,
-                Err(e) => {
-                    error!("Failed to fetch address transactions: {e}");
-                    break 'page_loop;
-                }
-            };
-
-            let ssp_transfer_types = [
-                sparkscan::types::AddressTransactionType::BitcoinDeposit,
-                sparkscan::types::AddressTransactionType::BitcoinWithdrawal,
-                sparkscan::types::AddressTransactionType::LightningPayment,
-            ];
-            let ssp_user_requests = self
-                .spark_wallet
-                .query_ssp_user_requests(
-                    address_transactions
-                        .iter()
-                        .filter(|tx| ssp_transfer_types.contains(&tx.type_))
-                        .map(|tx| tx.id.clone())
-                        .collect(),
+            let Ok(AddressTransactionsWithSspUserRequests {
+                address_transactions,
+                ssp_user_requests,
+            }) = self
+                .fetch_address_transactions_with_ssp_user_requests(
+                    &legacy_spark_address,
+                    next_offset,
                 )
-                .await?;
+                .await
+            else {
+                error!("Failed to fetch address transactions, stopping sync");
+                break 'page_loop;
+            };
 
             info!(
                 "Processing address transactions, offset = {next_offset}, transactions = {}",
                 address_transactions.len()
             );
             // Process transactions in this batch
-            for transaction in address_transactions {
+            for transaction in &address_transactions {
                 // Create payment records
                 let payments = payments_from_address_transaction_and_ssp_request(
                     transaction,
@@ -483,7 +475,7 @@ impl BreezSdk {
                             "Last synced payment id found ({last_synced_id:?}), stopping sync and proceeding to insert {} payments",
                             payments_to_sync.len()
                         );
-                        found_last_synced = true;
+                        head_synced = true;
                         break 'page_loop;
                     }
                     payments_to_sync.push(payment);
@@ -493,39 +485,103 @@ impl BreezSdk {
             // Check if we have more transfers to fetch
             next_offset = next_offset.saturating_add(u64::try_from(address_transactions.len())?);
             if (address_transactions.len() as u64) < PAYMENT_SYNC_BATCH_SIZE {
-                more_to_sync = false;
+                head_synced = true;
                 break 'page_loop;
             }
         }
 
-        let (next_head_offset, tail_synced) = match (from_head, found_last_synced) {
-            // If syncing from head and found the last synced payment, reset next head offset
-            (true, true) => (Some(0), None),
-            // If syncing from head and did not find the last synced payment, set the next head offset
-            (true, false) => (Some(next_offset), None),
-            // If syncing from tail and there is no more to sync, mark tail as synced
-            _ => (None, Some(!more_to_sync)),
-        };
         // Insert what synced payments we have into storage from oldest to newest
         for payment in payments_to_sync.iter().rev() {
             self.storage.insert_payment(payment.clone()).await?;
             info!("Inserted payment: {payment:?}");
-            let last_synced_payment_id = if from_head && found_last_synced {
-                Some(payment.id.clone())
+            let (last_synced_payment_id, next_head_offset) = if head_synced {
+                (Some(payment.id.clone()), 0)
             } else {
-                None
+                (None, next_offset)
             };
             object_repository
-                .merge_sync_info(last_synced_payment_id, next_head_offset, tail_synced)
+                .merge_sync_info(last_synced_payment_id, Some(next_head_offset), None)
                 .await?;
         }
-        if !from_head && payments_to_sync.is_empty() {
-            // If syncing from tail and no new payments were found, set tail synced
-            object_repository
-                .merge_sync_info(None, None, tail_synced)
-                .await?;
+
+        Ok(())
+    }
+
+    /// Synchronizes payments from the Spark network to persistent storage using the Sparkscan API.
+    ///
+    /// When syncing from tail, it will start from the count of completed payments in storage and page for
+    /// a maximum of `PAYMENT_SYNC_TAIL_MAX_PAGES` for one sync cycle. Once it reaches the end page,
+    /// it will update `tail_synced` in the sync info and the tail will no longer be synced in future cycles.
+    async fn sync_payments_tail_to_storage(&self) -> Result<(), SdkError> {
+        info!("Syncing payments tail to storage");
+        let object_repository = ObjectCacheRepository::new(self.storage.clone());
+        let cached_sync_info = object_repository
+            .fetch_sync_info()
+            .await?
+            .unwrap_or_default();
+        if cached_sync_info.tail_synced {
+            info!("Payments tail already synced, skipping");
+            return Ok(());
+        }
+        // TODO: use new spark address format once sparkscan supports it
+        let legacy_spark_address = self
+            .spark_wallet
+            .get_spark_address()?
+            .to_string_with_hrp_legacy();
+        let mut next_offset = self
+            .storage
+            .list_payments(None, None, Some(PaymentStatus::Completed))
+            .await?
+            .len() as u64;
+        let mut tail_synced = false;
+
+        for page in 1..=PAYMENT_SYNC_TAIL_MAX_PAGES {
+            info!(
+                "Fetching address transactions, offset = {next_offset}, page = {page}/{PAYMENT_SYNC_TAIL_MAX_PAGES}"
+            );
+            let Ok(AddressTransactionsWithSspUserRequests {
+                address_transactions,
+                ssp_user_requests,
+            }) = self
+                .fetch_address_transactions_with_ssp_user_requests(
+                    &legacy_spark_address,
+                    next_offset,
+                )
+                .await
+            else {
+                error!("Failed to fetch address transactions, stopping sync");
+                break;
+            };
+
+            info!(
+                "Processing address transactions, offset = {next_offset}, transactions = {}",
+                address_transactions.len()
+            );
+            // Process transactions in this batch
+            for transaction in &address_transactions {
+                let payments = payments_from_address_transaction_and_ssp_request(
+                    transaction,
+                    ssp_user_requests.get(&transaction.id),
+                    &legacy_spark_address,
+                )?;
+                // Insert payments
+                for payment in payments {
+                    self.storage.insert_payment(payment).await?;
+                }
+            }
+
+            // Check if we have more transfers to fetch
+            next_offset = next_offset.saturating_add(u64::try_from(address_transactions.len())?);
+            if (address_transactions.len() as u64) < PAYMENT_SYNC_BATCH_SIZE {
+                tail_synced = true;
+                break;
+            }
         }
 
+        object_repository
+            .merge_sync_info(None, None, Some(tail_synced))
+            .await?;
+
         Ok(())
     }