breez_sdk_spark/sdk/

sync.rs

use platform_utils::time::Instant;
use platform_utils::tokio;
use std::collections::HashSet;
use std::sync::Arc;
use tracing::{debug, error, info, trace, warn};

use super::{
    BreezSdk, CLAIM_TX_SIZE_VBYTES, SYNC_PAGING_LIMIT, SyncType,
    deposits::{
        InstantClaimOutcome, claim_already_made, is_already_claimed_error, larger_ceiling,
        needs_own_ceiling_resolution,
    },
};
use crate::utils::time::now_secs;
use crate::{
    DepositInfo, Fee, InputType, InstantClaimStatus, MaxFee, PaymentDetails, PaymentType,
    error::SdkError,
    events::{InternalSyncedEvent, SdkEvent},
    lnurl::ListMetadataRequest,
    models::{Payment, SyncWalletRequest, SyncWalletResponse},
    persist::{ObjectCacheRepository, UpdateDepositPayload},
    sync::SparkSyncService,
    utils::{
        deposit_chain_syncer::{DepositChainSyncer, TxOutput},
        payments::update_balances,
        utxo_fetcher::DetailedUtxo,
    },
};

/// Whether the cascade should (re)attempt an early claim. A decline is retried
/// once the deposit gains a confirmation, or once the ceiling that declined it
/// rises.
fn instant_claim_worth_attempting(
    status: Option<&InstantClaimStatus>,
    confirmations: u32,
    ceiling_sats: u64,
) -> bool {
    match status {
        Some(InstantClaimStatus::Declined {
            max_fee_sats,
            confirmations: declined_at,
        }) => confirmations > *declined_at || max_fee_sats.is_some_and(|prev| ceiling_sats > prev),
        _ => true,
    }
}

impl BreezSdk {
    pub(in crate::sdk) async fn sync_single_lnurl_metadata(&self, payment: &mut Payment) {
        if payment.payment_type != PaymentType::Receive {
            return;
        }

        let Some(PaymentDetails::Lightning {
            invoice,
            lnurl_receive_metadata,
            ..
        }) = &mut payment.details
        else {
            return;
        };

        if lnurl_receive_metadata.is_some() {
            // Already have lnurl metadata
            return;
        }

        let Ok(input) = self.input_parser.parse(invoice).await.map(InputType::from) else {
            error!(
                "Failed to parse invoice for lnurl metadata sync of payment {}",
                payment.id
            );
            debug!("Unparseable invoice: {invoice}");
            return;
        };

        let InputType::Bolt11Invoice(details) = input else {
            error!(
                "Input is not a Bolt11 invoice for lnurl metadata sync of payment {}",
                payment.id
            );
            debug!("Non-Bolt11 input: {invoice}");
            return;
        };

        // If there is a description hash, we assume this is a lnurl payment.
        if details.description_hash.is_none() {
            return;
        }

        // Let's check whether the lnurl receive metadata was already synced, then return early.
        // Important: Only return early if metadata is actually present (Some), otherwise we need
        // to trigger a sync. This prevents a race condition where the payment is in storage but
        // metadata sync from TransferClaimStarting hasn't completed yet.
        if let Ok(db_payment) = self.storage.get_payment_by_id(payment.id.clone()).await
            && let Some(PaymentDetails::Lightning {
                lnurl_receive_metadata: db_lnurl_receive_metadata @ Some(_),
                ..
            }) = db_payment.details
        {
            *lnurl_receive_metadata = db_lnurl_receive_metadata;
            return;
        }

        // Sync lnurl metadata directly instead of going through the sync trigger,
        // because this function is called from the sync loop's event handler,
        // which would deadlock waiting for itself to process the trigger.
        if let Err(e) = self.sync_lnurl_metadata().await {
            error!(
                "Failed to sync lnurl metadata for payment {}: {e}",
                payment.id
            );
            return;
        }

        let db_payment = match self.storage.get_payment_by_id(payment.id.clone()).await {
            Ok(p) => p,
            Err(e) => {
                debug!("Payment not found in storage for invoice {}: {e}", invoice);
                return;
            }
        };

        let Some(PaymentDetails::Lightning {
            lnurl_receive_metadata: db_lnurl_receive_metadata,
            ..
        }) = db_payment.details
        else {
            debug!(
                "No lnurl receive metadata in storage for invoice {}",
                invoice
            );
            return;
        };
        *lnurl_receive_metadata = db_lnurl_receive_metadata;
    }

    #[allow(clippy::too_many_lines)]
    pub(super) async fn sync_wallet_internal(
        &self,
        sync_type: SyncType,
        force: bool,
    ) -> Result<(), SdkError> {
        let cache = ObjectCacheRepository::new(self.storage.clone());
        let sync_interval_secs = u64::from(self.config.sync_interval_secs);
        let now = now_secs();

        // Skip if we synced recently (unless forced).
        if !force
            && let Some(last) = cache.get_last_sync_time().await?
            && now.saturating_sub(last) < sync_interval_secs
        {
            debug!("sync_wallet_internal: Synced recently, skipping");
            // When another instance shares our storage and keeps winning the sync
            // race, we would otherwise never emit a Synced event. Emit it here so
            // consumers are still notified that storage is up to date.
            self.event_emitter.emit(&SdkEvent::Synced).await;
            return Ok(());
        }

        // Update last sync time if this is a full sync.
        if sync_type.contains(SyncType::Full)
            && let Err(e) = cache.set_last_sync_time(now).await
        {
            error!("sync_wallet_internal: Failed to update last sync time: {e:?}");
        }

        let start_time = Instant::now();

        let sync_wallet = async {
            let wallet_synced = if sync_type.contains(SyncType::Wallet) {
                debug!("sync_wallet_internal: Starting Wallet sync");
                let wallet_start = Instant::now();
                match self.spark_wallet.sync().await {
                    Ok(()) => {
                        debug!(
                            "sync_wallet_internal: Wallet sync completed in {:?}",
                            wallet_start.elapsed()
                        );
                        true
                    }
                    Err(e) => {
                        error!(
                            "sync_wallet_internal: Spark wallet sync failed in {:?}: {e:?}",
                            wallet_start.elapsed()
                        );
                        false
                    }
                }
            } else {
                trace!("sync_wallet_internal: Skipping Wallet sync");
                false
            };

            // Finish any Lightning send that committed leaves with the operators
            // but never reached the SSP. Runs with the wallet sync because it
            // needs the operators' view of the transfer.
            if sync_type.contains(SyncType::Wallet) {
                self.lightning_sender.resume_pending_sends().await;
            }

            let wallet_state_synced = if sync_type.contains(SyncType::WalletState) {
                debug!("sync_wallet_internal: Starting WalletState sync");
                let wallet_state_start = Instant::now();
                match self.sync_wallet_state_to_storage().await {
                    Ok(()) => {
                        debug!(
                            "sync_wallet_internal: WalletState sync completed in {:?}",
                            wallet_state_start.elapsed()
                        );
                        true
                    }
                    Err(e) => {
                        error!(
                            "sync_wallet_internal: Failed to sync wallet state to storage in {:?}: {e:?}",
                            wallet_state_start.elapsed()
                        );
                        false
                    }
                }
            } else {
                trace!("sync_wallet_internal: Skipping WalletState sync");
                false
            };

            (wallet_synced, wallet_state_synced)
        };

        let sync_lnurl = async {
            if sync_type.contains(SyncType::LnurlMetadata) {
                debug!("sync_wallet_internal: Starting LnurlMetadata sync");
                let lnurl_start = Instant::now();
                match self.sync_lnurl_metadata().await {
                    Ok(()) => {
                        debug!(
                            "sync_wallet_internal: LnurlMetadata sync completed in {:?}",
                            lnurl_start.elapsed()
                        );
                        true
                    }
                    Err(e) => {
                        error!(
                            "sync_wallet_internal: Failed to sync lnurl metadata in {:?}: {e:?}",
                            lnurl_start.elapsed()
                        );
                        false
                    }
                }
            } else {
                trace!("sync_wallet_internal: Skipping LnurlMetadata sync");
                false
            }
        };

        let sync_deposits = async {
            if sync_type.contains(SyncType::Deposits) {
                debug!("sync_wallet_internal: Starting Deposits sync");
                let deposits_start = Instant::now();
                match self.check_and_claim_static_deposits().await {
                    Ok(()) => {
                        debug!(
                            "sync_wallet_internal: Deposits sync completed in {:?}",
                            deposits_start.elapsed()
                        );
                        true
                    }
                    Err(e) => {
                        error!(
                            "sync_wallet_internal: Failed to check and claim static deposits in {:?}: {e:?}",
                            deposits_start.elapsed()
                        );
                        false
                    }
                }
            } else {
                trace!("sync_wallet_internal: Skipping Deposits sync");
                false
            }
        };

        let ((wallet, wallet_state), lnurl_metadata, deposits) =
            tokio::join!(sync_wallet, sync_lnurl, sync_deposits);

        let elapsed = start_time.elapsed();
        let event = InternalSyncedEvent {
            wallet,
            wallet_state,
            lnurl_metadata,
            deposits,
            storage_incoming: None,
        };
        info!("sync_wallet_internal: Wallet sync completed in {elapsed:?}: {event:?}");
        self.event_emitter.emit_synced(&event).await;
        Ok(())
    }

    /// Synchronizes wallet state to persistent storage, making sure we have the latest balances and payments.
    pub(super) async fn sync_wallet_state_to_storage(&self) -> Result<(), SdkError> {
        update_balances(self.spark_wallet.clone(), self.storage.clone()).await?;

        let initial_sync_complete = *self.initial_synced_watcher.borrow();
        let sync_service = SparkSyncService::new(
            self.spark_wallet.clone(),
            self.storage.clone(),
            self.event_emitter.clone(),
        );
        sync_service.sync_payments(initial_sync_complete).await?;

        Ok(())
    }

    #[allow(clippy::too_many_lines)]
    pub(super) async fn check_and_claim_static_deposits(&self) -> Result<(), SdkError> {
        self.maybe_ensure_spark_private_mode_initialized().await?;
        let existing_deposits = self.storage.list_deposits().await?;
        let existing_keys: HashSet<TxOutput> = existing_deposits
            .iter()
            .map(|d| TxOutput {
                txid: d.txid.clone(),
                vout: d.vout,
            })
            .collect();

        let all_utxos = DepositChainSyncer::new(
            self.chain_service.clone(),
            self.storage.clone(),
            self.spark_wallet.clone(),
        )
        .sync()
        .await?;

        // Emit NewDeposits for any deposits not previously known
        let new_deposits: Vec<DepositInfo> = all_utxos
            .iter()
            .filter(|(u, _)| {
                !existing_keys.contains(&TxOutput {
                    txid: u.txid.to_string(),
                    vout: u.vout,
                })
            })
            .map(|(u, is_mature)| u.clone().into_deposit_info(*is_mature))
            .collect();
        if !new_deposits.is_empty() {
            self.event_emitter
                .emit(&SdkEvent::NewDeposits { new_deposits })
                .await;
        }

        let has_immature = all_utxos.iter().any(|(_, is_mature)| !is_mature);
        // Resolved once per pass, and only when an immature deposit could use it.
        // A deposit whose own ceiling differs from it resolves that separately, below.
        let instant_ceiling = if has_immature {
            match self
                .resolve_max_claim_fee(self.config.max_deposit_claim_fee.clone())
                .await
            {
                Ok(resolved) => resolved,
                Err(e) => {
                    warn!("Could not resolve the max claim fee, skipping instant claims: {e}");
                    None
                }
            }
        } else {
            None
        };

        // The tip is the same for every deposit in the pass, so it is read once and
        // reused. Read on first need rather than up front: whether any deposit gets
        // as far as needing it depends on the ceiling each one resolves to, which is
        // only known inside the loop. A failed read falls back to a per-deposit one.
        let mut tip_height: Option<u32> = None;

        let mut claimed_deposits: Vec<DepositInfo> = Vec::new();
        let mut unclaimed_deposits: Vec<DepositInfo> = Vec::new();
        for (detailed_utxo, is_mature) in all_utxos {
            let key = TxOutput {
                txid: detailed_utxo.txid.to_string(),
                vout: detailed_utxo.vout,
            };
            // Skip a deposit an explicit claim_deposit call is already working on.
            let Some(_claim_guard) = self.claim_guards.try_acquire(key.clone()) else {
                continue;
            };
            let stored = match self.stored_deposit(&key.txid, key.vout).await {
                Ok(stored) => stored,
                Err(e) => {
                    warn!(
                        "Could not read the stored deposit {}:{}: {e}",
                        key.txid, key.vout
                    );
                    continue;
                }
            };
            let instant_status = stored.as_ref().and_then(|d| d.instant_claim_status.clone());
            if claim_already_made(instant_status.as_ref()) {
                continue;
            }
            let stored_max_fee = stored.as_ref().and_then(|d| d.max_claim_fee.as_ref());
            if let Some(max_fee) = stored_max_fee {
                debug!(
                    "Deposit {}:{} runs under its own max claim fee {max_fee:?} \
                     rather than the configured one",
                    detailed_utxo.txid, detailed_utxo.vout
                );
            }
            let res = if is_mature {
                // Mature deposit: claim via the normal path.
                self.claim_utxo_and_resolve_deposit(
                    &detailed_utxo,
                    self.mature_claim_ceiling(stored_max_fee).await,
                    stored_max_fee.cloned(),
                    &mut claimed_deposits,
                    &mut unclaimed_deposits,
                )
                .await
            } else {
                // Not yet mature: attempt one early claim, held to the same
                // ceiling. Without a ceiling there is nothing to claim within.
                let ceiling = if needs_own_ceiling_resolution(
                    stored_max_fee,
                    self.config.max_deposit_claim_fee.as_ref(),
                ) {
                    match self.resolve_max_claim_fee(stored_max_fee.cloned()).await {
                        Ok(resolved) => resolved,
                        Err(e) => {
                            warn!(
                                "Could not resolve the max claim fee for {}:{}: {e}",
                                detailed_utxo.txid, detailed_utxo.vout
                            );
                            continue;
                        }
                    }
                } else {
                    instant_ceiling.clone()
                };
                let Some(ceiling) = ceiling else {
                    continue;
                };
                if tip_height.is_none() {
                    tip_height = self.chain_service.tip_height().await.ok();
                }
                let Ok(confirmations) = self
                    .deposit_confirmations_at_tip(&detailed_utxo.txid.to_string(), tip_height)
                    .await
                else {
                    continue;
                };
                if !instant_claim_worth_attempting(
                    instant_status.as_ref(),
                    confirmations,
                    ceiling.1,
                ) {
                    continue;
                }
                self.instant_claim_utxo_and_resolve_deposit(
                    &detailed_utxo,
                    Some(ceiling),
                    confirmations,
                    stored_max_fee.cloned(),
                    &mut claimed_deposits,
                )
                .await
            };

            if let Err(e) = res {
                warn!(
                    "Failed to update deposit for utxo {}:{}: {e}",
                    detailed_utxo.txid, detailed_utxo.vout
                );
            }
        }

        info!(
            "background claim completed, unclaimed deposits: {}",
            unclaimed_deposits.len()
        );
        debug!("unclaimed deposits: {unclaimed_deposits:?}");

        if !unclaimed_deposits.is_empty() {
            self.event_emitter
                .emit(&SdkEvent::UnclaimedDeposits { unclaimed_deposits })
                .await;
        }
        if !claimed_deposits.is_empty() {
            self.event_emitter
                .emit(&SdkEvent::ClaimedDeposits { claimed_deposits })
                .await;
        }
        Ok(())
    }

    /// The ceiling a mature deposit's automatic claim runs under. Both sides are
    /// resolved here because comparing them needs sats, and a ceiling that will not
    /// resolve leaves the deposit on its own, which is what was asked for.
    async fn mature_claim_ceiling(&self, stored: Option<&MaxFee>) -> Option<MaxFee> {
        let config_default = self.config.max_deposit_claim_fee.as_ref();
        let (Some(stored), Some(config_default)) = (stored, config_default) else {
            return stored.or(config_default).cloned();
        };
        let stored_sats = self.ceiling_sats(stored).await;
        let config_sats = self.ceiling_sats(config_default).await;
        match (stored_sats, config_sats) {
            (Some(stored_sats), Some(config_sats)) => Some(larger_ceiling(
                stored,
                stored_sats,
                config_default,
                config_sats,
            )),
            _ => Some(stored.clone()),
        }
    }

    /// One ceiling in sats, or `None` when it cannot be resolved.
    async fn ceiling_sats(&self, max_fee: &MaxFee) -> Option<u64> {
        self.resolve_max_claim_fee(Some(max_fee.clone()))
            .await
            .inspect_err(|e| warn!("Could not resolve a max claim fee: {e}"))
            .ok()
            .flatten()
            .map(|(_, sats)| sats)
    }

    async fn claim_utxo_and_resolve_deposit(
        &self,
        detailed_utxo: &DetailedUtxo,
        max_claim_fee: Option<MaxFee>,
        stored_max_fee: Option<MaxFee>,
        claimed_deposits: &mut Vec<DepositInfo>,
        unclaimed_deposits: &mut Vec<DepositInfo>,
    ) -> Result<(), SdkError> {
        match self.claim_utxo(detailed_utxo, max_claim_fee).await {
            Ok(_) => {
                info!("Claimed utxo {}:{}", detailed_utxo.txid, detailed_utxo.vout);
                self.storage
                    .delete_deposit(detailed_utxo.txid.to_string(), detailed_utxo.vout)
                    .await?;
                let mut info = detailed_utxo.clone().into_deposit_info(true);
                info.max_claim_fee = stored_max_fee;
                claimed_deposits.push(info);
            }
            // The deposit is settled, not failed: an earlier claim took it, here
            // or elsewhere. Marking it stops every later pass re-claiming it, and
            // records no claim error, which would have asked the user to act on a
            // deposit nothing more can be done with.
            Err(e) if is_already_claimed_error(&e.to_string()) => {
                info!(
                    "Deposit {}:{} was already claimed, marking it",
                    detailed_utxo.txid, detailed_utxo.vout
                );
                self.storage
                    .update_deposit(
                        detailed_utxo.txid.to_string(),
                        detailed_utxo.vout,
                        UpdateDepositPayload::InstantClaim {
                            status: InstantClaimStatus::Claimed,
                        },
                    )
                    .await?;
            }
            Err(e) => {
                warn!(
                    "Failed to claim utxo {}:{}: {e}",
                    detailed_utxo.txid, detailed_utxo.vout
                );
                unclaimed_deposits.push(
                    self.record_unclaimed_deposit(detailed_utxo, e, stored_max_fee)
                        .await?,
                );
            }
        }
        Ok(())
    }

    async fn instant_claim_utxo_and_resolve_deposit(
        &self,
        detailed_utxo: &DetailedUtxo,
        resolved_max_fee: Option<(Fee, u64)>,
        confirmations: u32,
        stored_max_fee: Option<MaxFee>,
        claimed_deposits: &mut Vec<DepositInfo>,
    ) -> Result<(), SdkError> {
        let outcome = match self
            .instant_claim_utxo(detailed_utxo, resolved_max_fee, confirmations)
            .await
        {
            Ok(outcome) => outcome,
            Err(e) => {
                // Transient transport/indexing error (e.g. the SSP has not indexed
                // the mempool tx yet): do NOT mark, so the next sync retries.
                warn!(
                    "Instant claim transient error for utxo {}:{}, will retry: {e}",
                    detailed_utxo.txid, detailed_utxo.vout
                );
                return Ok(());
            }
        };

        // Mark, don't delete: reconcile_deposits removes the row once the UTXO
        // leaves the feed.
        let status = outcome.status(confirmations);
        self.storage
            .update_deposit(
                detailed_utxo.txid.to_string(),
                detailed_utxo.vout,
                UpdateDepositPayload::InstantClaim {
                    status: status.clone(),
                },
            )
            .await?;

        match outcome {
            InstantClaimOutcome::Submitted(claim_id) => {
                info!(
                    "Instant claimed utxo {}:{} with claim_id: {claim_id}",
                    detailed_utxo.txid, detailed_utxo.vout
                );
                let mut info = detailed_utxo.clone().into_deposit_info(false);
                info.instant_claim_status = Some(status);
                info.max_claim_fee = stored_max_fee;
                claimed_deposits.push(info);
            }
            InstantClaimOutcome::Declined { error, .. } => {
                // Declining an early claim leaves the deposit to claim at maturity,
                // which is the ordinary outcome rather than a failure. No claim_error
                // is persisted and no UnclaimedDeposits event fires, unlike a claim at
                // maturity that fails.
                info!(
                    "Instant claim declined for utxo {}:{}: {error}",
                    detailed_utxo.txid, detailed_utxo.vout
                );
            }
        }
        Ok(())
    }

    /// Persists a claim failure on the deposit and returns the matching
    /// `DepositInfo`, carrying the failure and the deposit's own ceiling, for the
    /// `UnclaimedDeposits` event.
    async fn record_unclaimed_deposit(
        &self,
        utxo: &DetailedUtxo,
        error: SdkError,
        max_claim_fee: Option<MaxFee>,
    ) -> Result<DepositInfo, SdkError> {
        self.storage
            .update_deposit(
                utxo.txid.to_string(),
                utxo.vout,
                UpdateDepositPayload::ClaimError {
                    error: error.clone().into(),
                },
            )
            .await?;
        let mut info = utxo.clone().into_deposit_info(true);
        info.claim_error = Some(error.into());
        info.max_claim_fee = max_claim_fee;
        Ok(info)
    }

    pub(super) async fn sync_lnurl_metadata(&self) -> Result<(), SdkError> {
        let Some(lnurl_server_client) = self.lnurl_server_client.clone() else {
            return Ok(());
        };

        let cache = ObjectCacheRepository::new(Arc::clone(&self.storage));
        let mut updated_after = cache.fetch_lnurl_metadata_updated_after().await?;

        loop {
            debug!("Syncing lnurl metadata from updated_after {updated_after}");
            let metadata = lnurl_server_client
                .list_metadata(&ListMetadataRequest {
                    offset: None,
                    limit: Some(SYNC_PAGING_LIMIT),
                    updated_after: Some(updated_after),
                })
                .await?;

            if metadata.metadata.is_empty() {
                debug!("No more lnurl metadata on offset {updated_after}");
                break;
            }

            let len = u32::try_from(metadata.metadata.len())?;
            let last_updated_at = metadata.metadata.last().map(|m| m.updated_at);
            self.storage
                .set_lnurl_metadata(metadata.metadata.into_iter().map(From::from).collect())
                .await?;

            debug!(
                "Synchronized {} lnurl metadata at updated_after {updated_after}",
                len
            );
            updated_after = last_updated_at.unwrap_or(updated_after);
            cache
                .save_lnurl_metadata_updated_after(updated_after)
                .await?;

            if len < SYNC_PAGING_LIMIT {
                // No more invoices to fetch
                break;
            }
        }

        Ok(())
    }

    /// Resolves an on-chain max claim fee to `(fee, ceiling_sats)`, where the sat
    /// ceiling is computed over the claim tx size. `None` means no ceiling is set,
    /// which the caller treats as rejecting the claim.
    pub(super) async fn resolve_max_claim_fee(
        &self,
        max_claim_fee: Option<MaxFee>,
    ) -> Result<Option<(Fee, u64)>, SdkError> {
        match max_claim_fee {
            None => Ok(None),
            Some(max_fee) => {
                let fee = max_fee.to_fee(self.chain_service.as_ref()).await?;
                let sats = fee.to_sats(CLAIM_TX_SIZE_VBYTES);
                Ok(Some((fee, sats)))
            }
        }
    }

    /// Submits a static deposit claim for `detailed_utxo` and returns the
    /// resulting transfer id.
    pub(super) async fn claim_utxo(
        &self,
        detailed_utxo: &DetailedUtxo,
        max_claim_fee: Option<MaxFee>,
    ) -> Result<String, SdkError> {
        info!(
            "Fetching static deposit claim quote for deposit tx {}:{} and amount: {}",
            detailed_utxo.txid, detailed_utxo.vout, detailed_utxo.value
        );
        let quote = self
            .spark_wallet
            .fetch_static_deposit_claim_quote(detailed_utxo.tx.clone(), Some(detailed_utxo.vout))
            .await?;

        // The quote is bound to this outpoint, so its credit is priced against the
        // value read from the funding output. A credit above that value is not a
        // fee at all, and must not read as a free claim.
        let spark_requested_fee_sats = detailed_utxo
            .value
            .checked_sub(quote.credit_amount_sats)
            .ok_or_else(|| {
                SdkError::Generic(format!(
                    "Static deposit quote credits {} sats for {}:{}, which is worth {} sats",
                    quote.credit_amount_sats,
                    detailed_utxo.txid,
                    detailed_utxo.vout,
                    detailed_utxo.value
                ))
            })?;

        let spark_requested_fee_rate = spark_requested_fee_sats.div_ceil(CLAIM_TX_SIZE_VBYTES);

        let resolved_max_fee = self.resolve_max_claim_fee(max_claim_fee).await?;
        if let Some((_, max_fee_sats)) = &resolved_max_fee {
            info!(
                "Claiming {}:{} under a {max_fee_sats} sat ceiling, \
                 provider asks {spark_requested_fee_sats} sats",
                detailed_utxo.txid, detailed_utxo.vout
            );
        }
        let within_limit = resolved_max_fee
            .as_ref()
            .is_some_and(|(_, max_fee_sats)| spark_requested_fee_sats <= *max_fee_sats);
        if !within_limit {
            return Err(SdkError::MaxDepositClaimFeeExceeded {
                tx: detailed_utxo.txid.to_string(),
                vout: detailed_utxo.vout,
                max_fee: resolved_max_fee.map(|(fee, _)| fee),
                required_fee_sats: spark_requested_fee_sats,
                required_fee_rate_sat_per_vbyte: spark_requested_fee_rate,
            });
        }

        info!(
            "Claiming static deposit for utxo {}:{}",
            detailed_utxo.txid, detailed_utxo.vout
        );
        let credit_amount_sats = quote.credit_amount_sats;
        let transfer_id = self
            .spark_wallet
            .claim_static_deposit(&detailed_utxo.tx, quote)
            .await?;
        info!(
            "Claimed static deposit for utxo {}:{} (deposit value {}, credit {}), transfer {transfer_id}",
            detailed_utxo.txid, detailed_utxo.vout, detailed_utxo.value, credit_amount_sats,
        );
        Ok(transfer_id)
    }
}

#[cfg_attr(feature = "uniffi", uniffi::export(async_runtime = "tokio"))]
#[allow(clippy::needless_pass_by_value)]
impl BreezSdk {
    /// Synchronizes the wallet with the Spark network.
    ///
    /// Also collects the data a unilateral exit needs for any leaf still missing
    /// it, and waits for that before returning. This happens regardless of
    /// [`exit_chain_auto_fetch_enabled`](crate::Config::exit_chain_auto_fetch_enabled),
    /// which governs only the automatic collection: syncing is how to run one at
    /// a moment of your choosing with that turned off.
    #[allow(unused_variables)]
    pub async fn sync_wallet(
        &self,
        request: SyncWalletRequest,
    ) -> Result<SyncWalletResponse, SdkError> {
        self.runtime
            .run_user_sync(self, super::SyncType::Full, true)
            .await?;
        // Whatever `exit_chain_auto_fetch_enabled` says: that flag governs the
        // automatic collection, and an explicit sync is the way to ask for one
        // with it turned off. Awaited rather than left to the background, so a
        // caller that syncs before going offline knows the collection has run by
        // the time this returns. After the sync, so the leaves it brought in are
        // collected for too.
        self.runtime.collect_exit_chains(self).await?;
        Ok(SyncWalletResponse {})
    }
}

#[cfg(test)]
mod tests {
    use super::{claim_already_made, instant_claim_worth_attempting};
    use crate::InstantClaimStatus;

    fn declined(max_fee_sats: Option<u64>, confirmations: u32) -> InstantClaimStatus {
        InstantClaimStatus::Declined {
            max_fee_sats,
            confirmations,
        }
    }

    #[test]
    fn instant_retry_declines_once_per_confirmation() {
        // Never attempted -> attempt.
        assert!(instant_claim_worth_attempting(None, 1, 500));
        // A confirmation deeper than the decline: the provider lowers both its
        // spread and its floor with depth, so retry.
        assert!(instant_claim_worth_attempting(
            Some(&declined(Some(500), 1)),
            2,
            500
        ));
        assert!(instant_claim_worth_attempting(
            Some(&declined(None, 1)),
            2,
            500
        ));
        // Same depth, but the ceiling that declined it has been raised.
        assert!(instant_claim_worth_attempting(
            Some(&declined(Some(500), 1)),
            1,
            600
        ));
        // Same depth and the same ceiling: nothing has moved, so do not re-quote.
        assert!(!instant_claim_worth_attempting(
            Some(&declined(Some(500), 1)),
            1,
            500
        ));
        // A decline no ceiling caused does not re-quote on a ceiling change.
        assert!(!instant_claim_worth_attempting(
            Some(&declined(None, 1)),
            1,
            600
        ));
        // A shallower depth than the decline, as a reorg would give.
        assert!(!instant_claim_worth_attempting(
            Some(&declined(Some(500), 2)),
            1,
            500
        ));
    }

    #[test]
    fn a_ceiling_set_on_one_deposit_drives_its_retry() {
        // Declined under the configured ceiling, then given a higher one of its
        // own at the same depth: the deposit is re-quoted without waiting for a
        // confirmation. This is what makes raising a ceiling claim a deposit
        // early unattended.
        assert!(instant_claim_worth_attempting(
            Some(&declined(Some(99), 0)),
            0,
            50_000
        ));
        // Lowered below the ceiling that declined it, at the same depth: nothing
        // has opened up, so the hold-to-maturity direction costs no provider call.
        assert!(!instant_claim_worth_attempting(
            Some(&declined(Some(50_000), 0)),
            0,
            99
        ));
    }

    #[test]
    fn a_claim_already_taken_is_recognised() {
        // The cascade skips these before either claim path runs, so this is the
        // guard rather than a second opinion alongside one.
        let submitted = InstantClaimStatus::Submitted {
            claim_id: "claim-1".to_string(),
        };
        assert!(claim_already_made(Some(&submitted)));
        // A credited deposit is still reported until the provider spends the
        // output, so it has to be skipped for as long as the record lives.
        assert!(claim_already_made(Some(&InstantClaimStatus::Claimed)));
        assert!(!claim_already_made(None));
        // A decline leaves the deposit to be claimed at maturity.
        assert!(!claim_already_made(Some(&declined(Some(500), 1))));
    }
}