1use platform_utils::time::Instant;
2use platform_utils::tokio;
3use std::collections::{HashMap, HashSet};
4use std::sync::Arc;
5use tracing::{debug, error, info, trace, warn};
6
7use super::{
8 BreezSdk, CLAIM_TX_SIZE_VBYTES, SYNC_PAGING_LIMIT, SyncType, deposits::InstantClaimOutcome,
9};
10use crate::utils::time::now_secs;
11use crate::{
12 DepositInfo, Fee, InputType, InstantClaimStatus, MaxFee, PaymentDetails, PaymentType,
13 error::SdkError,
14 events::{InternalSyncedEvent, SdkEvent},
15 lnurl::ListMetadataRequest,
16 models::{Payment, SyncWalletRequest, SyncWalletResponse},
17 persist::{ObjectCacheRepository, UpdateDepositPayload},
18 sync::SparkSyncService,
19 utils::{
20 deposit_chain_syncer::{DepositChainSyncer, TxOutput},
21 payments::update_balances,
22 utxo_fetcher::DetailedUtxo,
23 },
24};
25
26fn instant_claim_worth_attempting(
30 status: Option<&InstantClaimStatus>,
31 confirmations: u32,
32 ceiling_sats: u64,
33) -> bool {
34 match status {
35 Some(InstantClaimStatus::Declined {
36 max_fee_sats,
37 confirmations: declined_at,
38 }) => confirmations > *declined_at || max_fee_sats.is_some_and(|prev| ceiling_sats > prev),
39 _ => true,
40 }
41}
42
43fn instant_claim_status_map(deposits: &[DepositInfo]) -> HashMap<TxOutput, InstantClaimStatus> {
45 deposits
46 .iter()
47 .filter_map(|d| {
48 d.instant_claim_status.clone().map(|status| {
49 (
50 TxOutput {
51 txid: d.txid.clone(),
52 vout: d.vout,
53 },
54 status,
55 )
56 })
57 })
58 .collect()
59}
60
61impl BreezSdk {
62 pub(in crate::sdk) async fn sync_single_lnurl_metadata(&self, payment: &mut Payment) {
63 if payment.payment_type != PaymentType::Receive {
64 return;
65 }
66
67 let Some(PaymentDetails::Lightning {
68 invoice,
69 lnurl_receive_metadata,
70 ..
71 }) = &mut payment.details
72 else {
73 return;
74 };
75
76 if lnurl_receive_metadata.is_some() {
77 return;
79 }
80
81 let Ok(input) = self.input_parser.parse(invoice).await.map(InputType::from) else {
82 error!(
83 "Failed to parse invoice for lnurl metadata sync of payment {}",
84 payment.id
85 );
86 debug!("Unparseable invoice: {invoice}");
87 return;
88 };
89
90 let InputType::Bolt11Invoice(details) = input else {
91 error!(
92 "Input is not a Bolt11 invoice for lnurl metadata sync of payment {}",
93 payment.id
94 );
95 debug!("Non-Bolt11 input: {invoice}");
96 return;
97 };
98
99 if details.description_hash.is_none() {
101 return;
102 }
103
104 if let Ok(db_payment) = self.storage.get_payment_by_id(payment.id.clone()).await
109 && let Some(PaymentDetails::Lightning {
110 lnurl_receive_metadata: db_lnurl_receive_metadata @ Some(_),
111 ..
112 }) = db_payment.details
113 {
114 *lnurl_receive_metadata = db_lnurl_receive_metadata;
115 return;
116 }
117
118 if let Err(e) = self.sync_lnurl_metadata().await {
122 error!(
123 "Failed to sync lnurl metadata for payment {}: {e}",
124 payment.id
125 );
126 return;
127 }
128
129 let db_payment = match self.storage.get_payment_by_id(payment.id.clone()).await {
130 Ok(p) => p,
131 Err(e) => {
132 debug!("Payment not found in storage for invoice {}: {e}", invoice);
133 return;
134 }
135 };
136
137 let Some(PaymentDetails::Lightning {
138 lnurl_receive_metadata: db_lnurl_receive_metadata,
139 ..
140 }) = db_payment.details
141 else {
142 debug!(
143 "No lnurl receive metadata in storage for invoice {}",
144 invoice
145 );
146 return;
147 };
148 *lnurl_receive_metadata = db_lnurl_receive_metadata;
149 }
150
151 #[allow(clippy::too_many_lines)]
152 pub(super) async fn sync_wallet_internal(
153 &self,
154 sync_type: SyncType,
155 force: bool,
156 ) -> Result<(), SdkError> {
157 let cache = ObjectCacheRepository::new(self.storage.clone());
158 let sync_interval_secs = u64::from(self.config.sync_interval_secs);
159 let now = now_secs();
160
161 if !force
163 && let Some(last) = cache.get_last_sync_time().await?
164 && now.saturating_sub(last) < sync_interval_secs
165 {
166 debug!("sync_wallet_internal: Synced recently, skipping");
167 self.event_emitter.emit(&SdkEvent::Synced).await;
171 return Ok(());
172 }
173
174 if sync_type.contains(SyncType::Full)
176 && let Err(e) = cache.set_last_sync_time(now).await
177 {
178 error!("sync_wallet_internal: Failed to update last sync time: {e:?}");
179 }
180
181 let start_time = Instant::now();
182
183 let sync_wallet = async {
184 let wallet_synced = if sync_type.contains(SyncType::Wallet) {
185 debug!("sync_wallet_internal: Starting Wallet sync");
186 let wallet_start = Instant::now();
187 match self.spark_wallet.sync().await {
188 Ok(()) => {
189 debug!(
190 "sync_wallet_internal: Wallet sync completed in {:?}",
191 wallet_start.elapsed()
192 );
193 true
194 }
195 Err(e) => {
196 error!(
197 "sync_wallet_internal: Spark wallet sync failed in {:?}: {e:?}",
198 wallet_start.elapsed()
199 );
200 false
201 }
202 }
203 } else {
204 trace!("sync_wallet_internal: Skipping Wallet sync");
205 false
206 };
207
208 if sync_type.contains(SyncType::Wallet) {
212 self.lightning_sender.resume_pending_sends().await;
213 }
214
215 let wallet_state_synced = if sync_type.contains(SyncType::WalletState) {
216 debug!("sync_wallet_internal: Starting WalletState sync");
217 let wallet_state_start = Instant::now();
218 match self.sync_wallet_state_to_storage().await {
219 Ok(()) => {
220 debug!(
221 "sync_wallet_internal: WalletState sync completed in {:?}",
222 wallet_state_start.elapsed()
223 );
224 true
225 }
226 Err(e) => {
227 error!(
228 "sync_wallet_internal: Failed to sync wallet state to storage in {:?}: {e:?}",
229 wallet_state_start.elapsed()
230 );
231 false
232 }
233 }
234 } else {
235 trace!("sync_wallet_internal: Skipping WalletState sync");
236 false
237 };
238
239 (wallet_synced, wallet_state_synced)
240 };
241
242 let sync_lnurl = async {
243 if sync_type.contains(SyncType::LnurlMetadata) {
244 debug!("sync_wallet_internal: Starting LnurlMetadata sync");
245 let lnurl_start = Instant::now();
246 match self.sync_lnurl_metadata().await {
247 Ok(()) => {
248 debug!(
249 "sync_wallet_internal: LnurlMetadata sync completed in {:?}",
250 lnurl_start.elapsed()
251 );
252 true
253 }
254 Err(e) => {
255 error!(
256 "sync_wallet_internal: Failed to sync lnurl metadata in {:?}: {e:?}",
257 lnurl_start.elapsed()
258 );
259 false
260 }
261 }
262 } else {
263 trace!("sync_wallet_internal: Skipping LnurlMetadata sync");
264 false
265 }
266 };
267
268 let sync_deposits = async {
269 if sync_type.contains(SyncType::Deposits) {
270 debug!("sync_wallet_internal: Starting Deposits sync");
271 let deposits_start = Instant::now();
272 match self.check_and_claim_static_deposits().await {
273 Ok(()) => {
274 debug!(
275 "sync_wallet_internal: Deposits sync completed in {:?}",
276 deposits_start.elapsed()
277 );
278 true
279 }
280 Err(e) => {
281 error!(
282 "sync_wallet_internal: Failed to check and claim static deposits in {:?}: {e:?}",
283 deposits_start.elapsed()
284 );
285 false
286 }
287 }
288 } else {
289 trace!("sync_wallet_internal: Skipping Deposits sync");
290 false
291 }
292 };
293
294 let ((wallet, wallet_state), lnurl_metadata, deposits) =
295 tokio::join!(sync_wallet, sync_lnurl, sync_deposits);
296
297 let elapsed = start_time.elapsed();
298 let event = InternalSyncedEvent {
299 wallet,
300 wallet_state,
301 lnurl_metadata,
302 deposits,
303 storage_incoming: None,
304 };
305 info!("sync_wallet_internal: Wallet sync completed in {elapsed:?}: {event:?}");
306 self.event_emitter.emit_synced(&event).await;
307 Ok(())
308 }
309
310 pub(super) async fn sync_wallet_state_to_storage(&self) -> Result<(), SdkError> {
312 update_balances(self.spark_wallet.clone(), self.storage.clone()).await?;
313
314 let initial_sync_complete = *self.initial_synced_watcher.borrow();
315 let sync_service = SparkSyncService::new(
316 self.spark_wallet.clone(),
317 self.storage.clone(),
318 self.event_emitter.clone(),
319 );
320 sync_service.sync_payments(initial_sync_complete).await?;
321
322 Ok(())
323 }
324
325 #[allow(clippy::too_many_lines)]
326 pub(super) async fn check_and_claim_static_deposits(&self) -> Result<(), SdkError> {
327 self.maybe_ensure_spark_private_mode_initialized().await?;
328 let existing_deposits = self.storage.list_deposits().await?;
329 let existing_keys: HashSet<TxOutput> = existing_deposits
330 .iter()
331 .map(|d| TxOutput {
332 txid: d.txid.clone(),
333 vout: d.vout,
334 })
335 .collect();
336
337 let all_utxos = DepositChainSyncer::new(
338 self.chain_service.clone(),
339 self.storage.clone(),
340 self.spark_wallet.clone(),
341 )
342 .sync()
343 .await?;
344
345 let new_deposits: Vec<DepositInfo> = all_utxos
347 .iter()
348 .filter(|(u, _)| {
349 !existing_keys.contains(&TxOutput {
350 txid: u.txid.to_string(),
351 vout: u.vout,
352 })
353 })
354 .map(|(u, is_mature)| u.clone().into_deposit_info(*is_mature))
355 .collect();
356 if !new_deposits.is_empty() {
357 self.event_emitter
358 .emit(&SdkEvent::NewDeposits { new_deposits })
359 .await;
360 }
361
362 let instant_status = instant_claim_status_map(&self.storage.list_deposits().await?);
366
367 let instant_ceiling = if all_utxos.iter().any(|(_, is_mature)| !is_mature) {
369 match self
370 .resolve_max_claim_fee(self.config.max_deposit_claim_fee.clone())
371 .await
372 {
373 Ok(resolved) => resolved,
374 Err(e) => {
375 warn!("Could not resolve the max claim fee, skipping instant claims: {e}");
376 None
377 }
378 }
379 } else {
380 None
381 };
382
383 let tip_height = if instant_ceiling.is_some() {
386 self.chain_service.tip_height().await.ok()
387 } else {
388 None
389 };
390
391 let mut claimed_deposits: Vec<DepositInfo> = Vec::new();
392 let mut unclaimed_deposits: Vec<DepositInfo> = Vec::new();
393 for (detailed_utxo, is_mature) in all_utxos {
394 let key = TxOutput {
395 txid: detailed_utxo.txid.to_string(),
396 vout: detailed_utxo.vout,
397 };
398 let Some(_claim_guard) = self.claim_guards.try_acquire(key.clone()) else {
400 continue;
401 };
402 let res = if is_mature {
403 self.claim_utxo_and_resolve_deposit(
405 &detailed_utxo,
406 self.config.max_deposit_claim_fee.clone(),
407 &mut claimed_deposits,
408 &mut unclaimed_deposits,
409 )
410 .await
411 } else {
412 let Some(ceiling) = instant_ceiling.clone() else {
415 continue;
416 };
417 let Ok(confirmations) = self
418 .deposit_confirmations_at_tip(&detailed_utxo.txid.to_string(), tip_height)
419 .await
420 else {
421 continue;
422 };
423 if !instant_claim_worth_attempting(
424 instant_status.get(&key),
425 confirmations,
426 ceiling.1,
427 ) {
428 continue;
429 }
430 self.instant_claim_utxo_and_resolve_deposit(
431 &detailed_utxo,
432 Some(ceiling),
433 confirmations,
434 &mut claimed_deposits,
435 )
436 .await
437 };
438
439 if let Err(e) = res {
440 warn!(
441 "Failed to update deposit for utxo {}:{}: {e}",
442 detailed_utxo.txid, detailed_utxo.vout
443 );
444 }
445 }
446
447 info!(
448 "background claim completed, unclaimed deposits: {}",
449 unclaimed_deposits.len()
450 );
451 debug!("unclaimed deposits: {unclaimed_deposits:?}");
452
453 if !unclaimed_deposits.is_empty() {
454 self.event_emitter
455 .emit(&SdkEvent::UnclaimedDeposits { unclaimed_deposits })
456 .await;
457 }
458 if !claimed_deposits.is_empty() {
459 self.event_emitter
460 .emit(&SdkEvent::ClaimedDeposits { claimed_deposits })
461 .await;
462 }
463 Ok(())
464 }
465
466 async fn claim_utxo_and_resolve_deposit(
467 &self,
468 detailed_utxo: &DetailedUtxo,
469 max_claim_fee: Option<MaxFee>,
470 claimed_deposits: &mut Vec<DepositInfo>,
471 unclaimed_deposits: &mut Vec<DepositInfo>,
472 ) -> Result<(), SdkError> {
473 match self.claim_utxo(detailed_utxo, max_claim_fee).await {
474 Ok(_) => {
475 info!("Claimed utxo {}:{}", detailed_utxo.txid, detailed_utxo.vout);
476 self.storage
477 .delete_deposit(detailed_utxo.txid.to_string(), detailed_utxo.vout)
478 .await?;
479 claimed_deposits.push(detailed_utxo.clone().into_deposit_info(true));
480 }
481 Err(e) => {
482 warn!(
483 "Failed to claim utxo {}:{}: {e}",
484 detailed_utxo.txid, detailed_utxo.vout
485 );
486 unclaimed_deposits.push(self.record_unclaimed_deposit(detailed_utxo, e).await?);
487 }
488 }
489 Ok(())
490 }
491
492 async fn instant_claim_utxo_and_resolve_deposit(
493 &self,
494 detailed_utxo: &DetailedUtxo,
495 resolved_max_fee: Option<(Fee, u64)>,
496 confirmations: u32,
497 claimed_deposits: &mut Vec<DepositInfo>,
498 ) -> Result<(), SdkError> {
499 let outcome = match self
500 .instant_claim_utxo(detailed_utxo, resolved_max_fee, confirmations)
501 .await
502 {
503 Ok(outcome) => outcome,
504 Err(e) => {
505 warn!(
508 "Instant claim transient error for utxo {}:{}, will retry: {e}",
509 detailed_utxo.txid, detailed_utxo.vout
510 );
511 return Ok(());
512 }
513 };
514
515 let status = outcome.status(confirmations);
518 self.storage
519 .update_deposit(
520 detailed_utxo.txid.to_string(),
521 detailed_utxo.vout,
522 UpdateDepositPayload::InstantClaim {
523 status: status.clone(),
524 },
525 )
526 .await?;
527
528 match outcome {
529 InstantClaimOutcome::Submitted(claim_id) => {
530 info!(
531 "Instant claimed utxo {}:{} with claim_id: {claim_id}",
532 detailed_utxo.txid, detailed_utxo.vout
533 );
534 let mut info = detailed_utxo.clone().into_deposit_info(false);
535 info.instant_claim_status = Some(status);
536 claimed_deposits.push(info);
537 }
538 InstantClaimOutcome::Declined { error, .. } => {
539 info!(
544 "Instant claim declined for utxo {}:{}: {error}",
545 detailed_utxo.txid, detailed_utxo.vout
546 );
547 }
548 }
549 Ok(())
550 }
551
552 async fn record_unclaimed_deposit(
555 &self,
556 utxo: &DetailedUtxo,
557 error: SdkError,
558 ) -> Result<DepositInfo, SdkError> {
559 self.storage
560 .update_deposit(
561 utxo.txid.to_string(),
562 utxo.vout,
563 UpdateDepositPayload::ClaimError {
564 error: error.clone().into(),
565 },
566 )
567 .await?;
568 let mut info = utxo.clone().into_deposit_info(true);
569 info.claim_error = Some(error.into());
570 Ok(info)
571 }
572
573 pub(super) async fn sync_lnurl_metadata(&self) -> Result<(), SdkError> {
574 let Some(lnurl_server_client) = self.lnurl_server_client.clone() else {
575 return Ok(());
576 };
577
578 let cache = ObjectCacheRepository::new(Arc::clone(&self.storage));
579 let mut updated_after = cache.fetch_lnurl_metadata_updated_after().await?;
580
581 loop {
582 debug!("Syncing lnurl metadata from updated_after {updated_after}");
583 let metadata = lnurl_server_client
584 .list_metadata(&ListMetadataRequest {
585 offset: None,
586 limit: Some(SYNC_PAGING_LIMIT),
587 updated_after: Some(updated_after),
588 })
589 .await?;
590
591 if metadata.metadata.is_empty() {
592 debug!("No more lnurl metadata on offset {updated_after}");
593 break;
594 }
595
596 let len = u32::try_from(metadata.metadata.len())?;
597 let last_updated_at = metadata.metadata.last().map(|m| m.updated_at);
598 self.storage
599 .set_lnurl_metadata(metadata.metadata.into_iter().map(From::from).collect())
600 .await?;
601
602 debug!(
603 "Synchronized {} lnurl metadata at updated_after {updated_after}",
604 len
605 );
606 updated_after = last_updated_at.unwrap_or(updated_after);
607 cache
608 .save_lnurl_metadata_updated_after(updated_after)
609 .await?;
610
611 if len < SYNC_PAGING_LIMIT {
612 break;
614 }
615 }
616
617 Ok(())
618 }
619
620 pub(super) async fn resolve_max_claim_fee(
624 &self,
625 max_claim_fee: Option<MaxFee>,
626 ) -> Result<Option<(Fee, u64)>, SdkError> {
627 match max_claim_fee {
628 None => Ok(None),
629 Some(max_fee) => {
630 let fee = max_fee.to_fee(self.chain_service.as_ref()).await?;
631 let sats = fee.to_sats(CLAIM_TX_SIZE_VBYTES);
632 Ok(Some((fee, sats)))
633 }
634 }
635 }
636
637 pub(super) async fn claim_utxo(
640 &self,
641 detailed_utxo: &DetailedUtxo,
642 max_claim_fee: Option<MaxFee>,
643 ) -> Result<String, SdkError> {
644 info!(
645 "Fetching static deposit claim quote for deposit tx {}:{} and amount: {}",
646 detailed_utxo.txid, detailed_utxo.vout, detailed_utxo.value
647 );
648 let quote = self
649 .spark_wallet
650 .fetch_static_deposit_claim_quote(detailed_utxo.tx.clone(), Some(detailed_utxo.vout))
651 .await?;
652
653 let spark_requested_fee_sats = detailed_utxo
657 .value
658 .checked_sub(quote.credit_amount_sats)
659 .ok_or_else(|| {
660 SdkError::Generic(format!(
661 "Static deposit quote credits {} sats for {}:{}, which is worth {} sats",
662 quote.credit_amount_sats,
663 detailed_utxo.txid,
664 detailed_utxo.vout,
665 detailed_utxo.value
666 ))
667 })?;
668
669 let spark_requested_fee_rate = spark_requested_fee_sats.div_ceil(CLAIM_TX_SIZE_VBYTES);
670
671 let resolved_max_fee = self.resolve_max_claim_fee(max_claim_fee).await?;
672 if let Some((_, max_fee_sats)) = &resolved_max_fee {
673 info!("User max fee: {max_fee_sats} spark requested fee: {spark_requested_fee_sats}");
674 }
675 let within_limit = resolved_max_fee
676 .as_ref()
677 .is_some_and(|(_, max_fee_sats)| spark_requested_fee_sats <= *max_fee_sats);
678 if !within_limit {
679 return Err(SdkError::MaxDepositClaimFeeExceeded {
680 tx: detailed_utxo.txid.to_string(),
681 vout: detailed_utxo.vout,
682 max_fee: resolved_max_fee.map(|(fee, _)| fee),
683 required_fee_sats: spark_requested_fee_sats,
684 required_fee_rate_sat_per_vbyte: spark_requested_fee_rate,
685 });
686 }
687
688 info!(
689 "Claiming static deposit for utxo {}:{}",
690 detailed_utxo.txid, detailed_utxo.vout
691 );
692 let credit_amount_sats = quote.credit_amount_sats;
693 let transfer_id = self
694 .spark_wallet
695 .claim_static_deposit(&detailed_utxo.tx, quote)
696 .await?;
697 info!(
698 "Claimed static deposit for utxo {}:{} (deposit value {}, credit {}), transfer {transfer_id}",
699 detailed_utxo.txid, detailed_utxo.vout, detailed_utxo.value, credit_amount_sats,
700 );
701 Ok(transfer_id)
702 }
703}
704
705#[cfg_attr(feature = "uniffi", uniffi::export(async_runtime = "tokio"))]
706#[allow(clippy::needless_pass_by_value)]
707impl BreezSdk {
708 #[allow(unused_variables)]
716 pub async fn sync_wallet(
717 &self,
718 request: SyncWalletRequest,
719 ) -> Result<SyncWalletResponse, SdkError> {
720 self.runtime
721 .run_user_sync(self, super::SyncType::Full, true)
722 .await?;
723 self.runtime.collect_exit_chains(self).await?;
730 Ok(SyncWalletResponse {})
731 }
732}
733
734#[cfg(test)]
735mod tests {
736 use super::instant_claim_worth_attempting;
737 use crate::InstantClaimStatus;
738
739 fn declined(max_fee_sats: Option<u64>, confirmations: u32) -> InstantClaimStatus {
740 InstantClaimStatus::Declined {
741 max_fee_sats,
742 confirmations,
743 }
744 }
745
746 #[test]
747 fn instant_retry_declines_once_per_confirmation() {
748 assert!(instant_claim_worth_attempting(None, 1, 500));
750 assert!(instant_claim_worth_attempting(
753 Some(&declined(Some(500), 1)),
754 2,
755 500
756 ));
757 assert!(instant_claim_worth_attempting(
758 Some(&declined(None, 1)),
759 2,
760 500
761 ));
762 assert!(instant_claim_worth_attempting(
764 Some(&declined(Some(500), 1)),
765 1,
766 600
767 ));
768 assert!(!instant_claim_worth_attempting(
770 Some(&declined(Some(500), 1)),
771 1,
772 500
773 ));
774 assert!(!instant_claim_worth_attempting(
776 Some(&declined(None, 1)),
777 1,
778 600
779 ));
780 assert!(!instant_claim_worth_attempting(
782 Some(&declined(Some(500), 2)),
783 1,
784 500
785 ));
786 }
788}