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, SYNC_PAGING_LIMIT, SyncType,
    deposits::{
        InstantClaimOutcome, ResolvedMaxFee, claim_already_made, is_already_claimed_error,
        larger_ceiling, mature_quote_error, needs_own_ceiling_resolution,
    },
};
use crate::utils::time::now_secs;
use crate::{
    DepositInfo, InputType, InstantClaimStatus, MaxFee, PaymentDetails, PaymentType, SyncError,
    chain::FeeMarket,
    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::{CachedUtxoFetcher, 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_result = 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()
                        );
                        Some(Ok(()))
                    }
                    Err(e) => {
                        error!(
                            "sync_wallet_internal: Spark wallet sync failed in {:?}: {e:?}",
                            wallet_start.elapsed()
                        );
                        Some(Err(e))
                    }
                }
            } else {
                trace!("sync_wallet_internal: Skipping Wallet sync");
                None
            };

            // 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
            };

            // Runs after the wallet sync, in which the wallet stores the leaves it
            // reads.
            if sync_type.contains(SyncType::ExitChainState) {
                self.sync_recoverable_funds().await;
            }

            (wallet_result, 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_result, wallet_state), lnurl_metadata, deposits) =
            tokio::join!(sync_wallet, sync_lnurl, sync_deposits);
        let wallet = matches!(wallet_result, Some(Ok(())));
        // Set after the join so the balances are already written when these change.
        if wallet && let Err(e) = cache.set_last_synced_at(now).await {
            error!("sync_wallet_internal: Failed to update last synced at: {e:?}");
        }
        if let Some(result) = wallet_result {
            *self.last_sync_error.lock().await = result.err().map(SyncError::from);
        }

        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> {
        let account_info = update_balances(self.spark_wallet.clone(), self.storage.clone()).await?;
        if let Some(rewards_delegator) = &self.rewards_delegator {
            rewards_delegator.set_token_balances(&account_info.token_balances);
        }

        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(),
        )
        .with_sync_interval_secs(self.config.sync_interval_secs);
        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_deposits = 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_deposits
            .iter()
            .filter(|deposit| !existing_keys.contains(&deposit.key()))
            .map(|deposit| deposit.clone().into_deposit_info())
            .collect();
        if !new_deposits.is_empty() {
            self.event_emitter
                .emit(&SdkEvent::NewDeposits { new_deposits })
                .await;
        }

        let has_immature = all_deposits.iter().any(|deposit| !deposit.is_mature);
        // Every ceiling in the pass is priced at the same fee market.
        let fee_market = FeeMarket::new(self.chain_service.as_ref());
        // 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(), &fee_market)
                .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 utxo_fetcher = CachedUtxoFetcher::new(self.chain_service.clone(), self.storage.clone());
        let mut claimed_deposits: Vec<DepositInfo> = Vec::new();
        let mut unclaimed_deposits: Vec<DepositInfo> = Vec::new();
        for deposit in all_deposits {
            let key = deposit.key();
            let is_mature = deposit.is_mature;
            // 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;
            }
            // Claiming needs the deposit transaction, which only the chain service
            // has. Without it the deposit stays listed and is retried next pass.
            let detailed_utxo = match utxo_fetcher.fetch_detailed_utxo(&key.txid, key.vout).await {
                Ok(detailed_utxo) => detailed_utxo,
                Err(e) => {
                    warn!(
                        "Skipping the claim of {}:{} this pass, its transaction is unavailable: {e}",
                        key.txid, key.vout
                    );
                    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, detailed_utxo.value, &fee_market)
                        .await,
                    stored_max_fee.cloned(),
                    &fee_market,
                    &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(), &fee_market)
                        .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.ceiling_sats(detailed_utxo.value),
                ) {
                    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.
    pub(super) async fn mature_claim_ceiling(
        &self,
        stored: Option<&MaxFee>,
        deposit_sats: u64,
        fee_market: &FeeMarket<'_>,
    ) -> 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, deposit_sats, fee_market).await;
        let config_sats = self
            .ceiling_sats(config_default, deposit_sats, fee_market)
            .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 for a deposit worth `deposit_sats`, or `None` when it
    /// cannot be resolved.
    async fn ceiling_sats(
        &self,
        max_fee: &MaxFee,
        deposit_sats: u64,
        fee_market: &FeeMarket<'_>,
    ) -> Option<u64> {
        self.resolve_max_claim_fee(Some(max_fee.clone()), fee_market)
            .await
            .inspect_err(|e| warn!("Could not resolve a max claim fee: {e}"))
            .ok()
            .flatten()
            .map(|max_fee| max_fee.ceiling_sats(deposit_sats))
    }

    async fn claim_utxo_and_resolve_deposit(
        &self,
        detailed_utxo: &DetailedUtxo,
        max_claim_fee: Option<MaxFee>,
        stored_max_fee: Option<MaxFee>,
        fee_market: &FeeMarket<'_>,
        claimed_deposits: &mut Vec<DepositInfo>,
        unclaimed_deposits: &mut Vec<DepositInfo>,
    ) -> Result<(), SdkError> {
        match self
            .claim_utxo(detailed_utxo, max_claim_fee, fee_market)
            .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<ResolvedMaxFee>,
        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(())
    }

    /// Prices a max claim fee's on-chain part at the current fee market. `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>,
        fee_market: &FeeMarket<'_>,
    ) -> Result<Option<ResolvedMaxFee>, SdkError> {
        match max_claim_fee {
            None => Ok(None),
            Some(max_fee) => {
                let onchain = max_fee.to_fee(fee_market).await?;
                Ok(Some(ResolvedMaxFee::new(
                    onchain,
                    max_fee.proportional_ppm(),
                )))
            }
        }
    }

    /// 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>,
        fee_market: &FeeMarket<'_>,
    ) -> 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
            .map_err(|e| mature_quote_error(e, detailed_utxo))?;

        // 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 resolved_max_fee = self
            .resolve_max_claim_fee(max_claim_fee, fee_market)
            .await?;
        let max_fee_sats = resolved_max_fee
            .as_ref()
            .map(|max_fee| max_fee.ceiling_sats(detailed_utxo.value));
        if let Some(max_fee_sats) = max_fee_sats {
            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 =
            max_fee_sats.is_some_and(|max_fee_sats| spark_requested_fee_sats <= max_fee_sats);
        if !within_limit {
            return Err(SdkError::max_deposit_claim_fee_exceeded(
                detailed_utxo.txid.to_string(),
                detailed_utxo.vout,
                max_fee_sats,
                spark_requested_fee_sats,
            ));
        }

        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)
    }
}

#[macros::instance_span]
#[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?;
        let last_sync_error = self.last_sync_error.lock().await.clone();
        let last_synced_at = ObjectCacheRepository::new(self.storage.clone())
            .get_last_synced_at()
            .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 {
            last_sync_error,
            last_synced_at,
        })
    }
}

#[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))));
    }
}