Add varialized interval for intermediate saving during updates

src/api.rs

@@ -6,6 +6,8 @@ use std::ops::Index;
 use std::str::FromStr;
 use std::string::FromUtf8Error;
 use std::sync::{Mutex, MutexGuard, OnceLock, PoisonError};
+use futures_util::StreamExt;
+use sdk_common::updates::{init_update_sink, StateUpdate, WasmUpdateSink};
 use web_time::{Duration, Instant};
 use std::u32;
 
@@ -296,13 +298,90 @@ pub fn create_new_device(birthday: u32, network_str: String) -> ApiResult<String
 
 #[wasm_bindgen]
 pub async fn scan_blocks(tip_height: u32, blindbit_url: String) -> ApiResult<()> {
+    let (sink, mut scan_rx, mut state_rx) = WasmUpdateSink::new();
+        
+    // Initialize the global sink
+    init_update_sink(sink);
+
+    wasm_bindgen_futures::spawn_local(async move {
+        while let Some(progress) = scan_rx.next().await {
+            log::info!(
+                "Scan progress: {}/{} ({}%)", 
+                progress.current, 
+                progress.end,
+                (progress.current * 100) / progress.end
+            );
+        }
+    });
+
+    wasm_bindgen_futures::spawn_local(async move {
+        while let Some(update) = state_rx.next().await {
+            match update {
+                StateUpdate::Update {
+                    blkheight,
+                    blkhash,
+                    found_outputs,
+                    found_inputs,
+                } => {
+                    log::info!(
+                        "Processing update at height {}: {} outputs, {} inputs", 
+                        blkheight, found_outputs.len(), found_inputs.len()
+                    );
+                    
+                    // Actually update the device's wallet state in WASM context
+                    if let Ok(mut local_device) = lock_local_device() {
+                        // Get direct access to the device's outputs to modify them
+                        let device_outputs = local_device.get_mut_outputs();
+                        
+                        // Mark found inputs as spent
+                        for outpoint in found_inputs {
+                            if let Some(output) = device_outputs.get_mut(&outpoint) {
+                                output.spend_status = OutputSpendStatus::Spent(blkhash.to_byte_array());
+                            } else {
+                                // This should never happen, but we'll log it just in case
+                                log::error!("Found a spent input that we don't know about: {}", outpoint.to_string());
+                            }
+                        }
+                        
+                        // Add new found outputs to the device
+                        device_outputs.extend(found_outputs);
+
+                        // update last_scan
+                        local_device.get_mut_sp_wallet().set_last_scan(blkheight.to_consensus_u32());
+                        
+                        log::debug!("Updated device outputs and marked inputs as spent");
+                    } else {
+                        log::error!("Failed to lock local device for state update");
+                    }
+                    
+                    log::debug!("State update processed at height: {}", blkheight);
+                }
+                StateUpdate::NoUpdate { blkheight } => {
+                    log::debug!("No update at height: {}", blkheight);
+                    // We still need to update last_scan
+                    if let Ok(mut local_device) = lock_local_device() {
+                        let sp_wallet = local_device.get_mut_sp_wallet();
+                        sp_wallet.set_last_scan(blkheight.to_consensus_u32());
+                    } else {
+                        log::error!("Failed to lock local device for last_scan update");
+                    }
+                }
+            }
+        }
+    });
+
+    let (sp_client, owned_outpoints, last_scan) = {
     let local_device = lock_local_device()?;
+        let owned_outpoints: HashSet<OutPoint> = local_device.get_sp_wallet().get_unspent_outputs().keys().map(|o| *o).collect();
+        let sp_client = local_device.get_sp_client().clone();
+        let last_scan = local_device.get_sp_wallet().get_last_scan();
 
-    let sp_wallet = local_device.get_sp_wallet();
+        (sp_client, owned_outpoints, last_scan)
+    };
 
-    let last_scan = sp_wallet.get_last_scan();
     let n_blocks_to_scan = tip_height - last_scan;
-    crate::wallet::scan_blocks(n_blocks_to_scan, &blindbit_url, sp_wallet, tip_height, last_scan).await?;
+
+    crate::wallet::scan_blocks(n_blocks_to_scan, &blindbit_url, sp_client, owned_outpoints, tip_height, last_scan, 10).await?;
 
     Ok(())
 }
@@ -470,7 +549,7 @@ pub fn get_opreturn(transaction: String) -> ApiResult<String> {
 #[wasm_bindgen]
 pub fn process_commit_new_state(mut process: Process, state_id: String, new_tip: String) -> ApiResult<Process> {
     let state_id_array: [u8; 32] = Vec::from_hex(&state_id)?.try_into().unwrap();
-    let new_tip = OutPoint::from_str(&new_tip)?;
+    let new_tip = OutPoint::new(Txid::from_str(&new_tip)?, 0);
     let new_state: ProcessState;
     if let Ok(commited_state) = process.get_state_for_id(&state_id_array) {
         new_state = commited_state.clone();

src/scanner.rs

@@ -21,7 +21,6 @@ impl<'a> WasmSpScanner<'a> {
         owned_outpoints: HashSet<OutPoint>,
         keep_scanning: &'a AtomicBool,
     ) -> Self {
-        log::info!("Creating WasmSpScanner");
         Self {
             client,
             updater,
@@ -42,6 +41,9 @@ impl<'a> WasmSpScanner<'a> {
         let mut update_time = Instant::now();
         let mut stream = block_data_stream;
 
+        let save_interval = 10;
+        let mut blocks_scanned = 1;
+
         while let Some(blockdata) = stream.next().await {
             let blockdata = blockdata?;
             let blkheight = blockdata.blkheight;
@@ -55,8 +57,8 @@ impl<'a> WasmSpScanner<'a> {
 
             let mut save_to_storage = false;
 
-            // always save on last block or after 30 seconds since last save
-            if blkheight == end || update_time.elapsed() > Duration::from_secs(30) {
+            // always save on last block or after scanning some number of blocks
+            if blkheight == end || blocks_scanned % save_interval == 0 {
                 save_to_storage = true;
             }
 
@@ -79,6 +81,8 @@ impl<'a> WasmSpScanner<'a> {
                 self.save_state()?;
                 update_time = Instant::now();
             }
+
+            blocks_scanned += 1;
         }
 
         Ok(())

src/wallet.rs

@@ -31,9 +31,15 @@ pub fn generate_sp_wallet(network: Network) -> anyhow::Result<SpClient> {
     )
 }
 
-pub async fn scan_blocks(mut n_blocks_to_scan: u32, blindbit_url: &str, sp_wallet: &SpWallet, tip_height: u32, scan_height: u32) -> anyhow::Result<()> {
-    log::info!("Starting a rescan");
-
+pub async fn scan_blocks(
+    mut n_blocks_to_scan: u32, 
+    blindbit_url: &str, 
+    sp_client: SpClient, 
+    owned_outpoints: HashSet<OutPoint>,
+    tip_height: u32, 
+    scan_height: u32,
+    save_interval: u32,
+) -> anyhow::Result<()> {
     // 0 means scan to tip
     if n_blocks_to_scan == 0 {
         n_blocks_to_scan = tip_height - scan_height;
@@ -53,15 +59,11 @@ pub async fn scan_blocks(mut n_blocks_to_scan: u32, blindbit_url: &str, sp_walle
     let updater = StateUpdater::new();
     let backend = BlindbitBackend::new(blindbit_url.to_string())?;
 
-    let owned_outpoints = sp_wallet.get_unspent_outputs().keys().map(|o| *o).collect();
-
     let keep_scanning = Arc::new(AtomicBool::new(true));
 
-    log::info!("start: {} end: {}", start, end);
     let start_time = Instant::now();
-    log::info!("{:?}", start_time);
     let mut scanner = WasmSpScanner::new(
-        sp_wallet.get_sp_client().clone(),
+        sp_client,
         Box::new(updater),
         Box::new(backend),
         owned_outpoints,
@@ -78,11 +80,5 @@ pub async fn scan_blocks(mut n_blocks_to_scan: u32, blindbit_url: &str, sp_walle
         )
         .await?;
 
-    // time elapsed for the scan
-    log::info!(
-        "Scan complete in {} seconds",
-        start_time.elapsed().as_secs()
-    );
-
     Ok(())
 }