breez_sdk_spark/
events.rs

1use core::fmt;
2use std::{
3    collections::BTreeMap,
4    sync::atomic::{AtomicBool, AtomicU64, Ordering},
5};
6
7use platform_utils::time::Instant;
8use serde::Serialize;
9use tokio::sync::{Mutex, RwLock};
10use tracing::info;
11use uuid::Uuid;
12
13use crate::{DepositInfo, LightningAddressInfo, Payment, sdk::RuntimeEvent};
14
15/// Events emitted by the SDK
16#[allow(clippy::large_enum_variant)]
17#[derive(Debug, Clone, Serialize)]
18#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
19pub enum SdkEvent {
20    /// Emitted when the wallet has been synchronized with the network
21    Synced,
22    /// Emitted when the SDK was unable to claim deposits. Each deposit carries a
23    /// `claim_error` with the reason.
24    UnclaimedDeposits {
25        unclaimed_deposits: Vec<DepositInfo>,
26    },
27    /// Emitted when deposits were claimed into the wallet. The resulting payment
28    /// is emitted separately as `PaymentSucceeded`.
29    ClaimedDeposits { claimed_deposits: Vec<DepositInfo> },
30    /// Emitted when a payment completed. The cached balance is refreshed before
31    /// this event, so `get_info` returns the new value.
32    PaymentSucceeded { payment: Payment },
33    /// Emitted when a payment is in flight. The same payment is emitted again as
34    /// succeeded or failed once it settles.
35    PaymentPending { payment: Payment },
36    /// Emitted when a payment failed.
37    PaymentFailed { payment: Payment },
38    /// Emitted when details of an already stored payment changed, such as the
39    /// conversion info of a cross-chain receive arriving after the payment
40    /// settled. Carries the updated payment.
41    PaymentMetadataUpdated { payment: Payment },
42    /// Emitted while the background auto-optimizer is running.
43    ///
44    /// Only fired from the auto path (enabled via
45    /// `LeafOptimizationConfig::auto_enabled`). Manually-triggered runs
46    /// via `BreezSdk::optimize_leaves` do not emit events: they return an
47    /// `OptimizationOutcome` instead.
48    AutoOptimization {
49        // Named with `optimization` prefix to avoid collision with `event` keyword in C#
50        optimization_event: AutoOptimizationEvent,
51    },
52    /// Emitted when the Lightning address changed on another device. The address
53    /// is unset when it was deleted.
54    LightningAddressChanged {
55        lightning_address: Option<LightningAddressInfo>,
56    },
57    /// Emitted when on-chain deposits are detected. Only deposits whose
58    /// `is_mature` is set have enough confirmations to be claimed.
59    NewDeposits { new_deposits: Vec<DepositInfo> },
60    /// Emitted when the data a unilateral exit is built from has changed, so an
61    /// exit state exported earlier is out of date and should be exported again.
62    UnilateralExitStateChanged,
63    /// Emitted when a Stable Balance conversion failed: sweeping received
64    /// bitcoin into the stable token, or converting the token back to bitcoin
65    /// on deactivation. The SDK tries again after a growing delay, and
66    /// `retry_in_secs` is the soonest it will. It is unset for a received
67    /// payment's own conversion, which is not retried: its sats go to the next
68    /// batch conversion, which reports its own failures.
69    StableBalanceConversionFailed {
70        conversion: StableBalanceConversionKind,
71        error: String,
72        retry_in_secs: Option<u64>,
73    },
74    /// Emitted when sync finds new recoverable funds. Carries the total
75    /// `get_info` reports.
76    RecoverableFunds { recoverable_funds_sats: u64 },
77}
78
79/// Which Stable Balance conversion an [`SdkEvent::StableBalanceConversionFailed`]
80/// refers to.
81#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
82#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
83pub enum StableBalanceConversionKind {
84    /// A single received payment being converted to the stable token.
85    PerReceive,
86    /// Bitcoin above the threshold being swept into the stable token.
87    AutoConvert,
88    /// The stable token being converted back to bitcoin after deactivation.
89    Deactivation,
90}
91
92impl SdkEvent {
93    pub(crate) fn from_payment(payment: Payment) -> Self {
94        match payment.status {
95            crate::PaymentStatus::Completed => SdkEvent::PaymentSucceeded { payment },
96            crate::PaymentStatus::Pending => SdkEvent::PaymentPending { payment },
97            crate::PaymentStatus::Failed => SdkEvent::PaymentFailed { payment },
98        }
99    }
100}
101
102impl fmt::Display for SdkEvent {
103    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
104        match self {
105            SdkEvent::Synced => write!(f, "Synced"),
106            SdkEvent::UnclaimedDeposits { unclaimed_deposits } => {
107                write!(f, "UnclaimedDeposits[{}]", unclaimed_deposits.len())
108            }
109            SdkEvent::ClaimedDeposits { claimed_deposits } => {
110                write!(f, "ClaimedDeposits[{}]", claimed_deposits.len())
111            }
112            SdkEvent::PaymentSucceeded { payment } => {
113                write!(f, "PaymentSucceeded({})", payment.id)
114            }
115            SdkEvent::PaymentPending { payment } => {
116                write!(f, "PaymentPending({})", payment.id)
117            }
118            SdkEvent::PaymentFailed { payment } => {
119                write!(f, "PaymentFailed({})", payment.id)
120            }
121            SdkEvent::PaymentMetadataUpdated { payment } => {
122                write!(f, "PaymentMetadataUpdated({})", payment.id)
123            }
124            SdkEvent::AutoOptimization {
125                optimization_event: event,
126            } => {
127                write!(f, "AutoOptimization: {event:?}")
128            }
129            SdkEvent::LightningAddressChanged { .. } => {
130                write!(f, "LightningAddressChanged")
131            }
132            SdkEvent::NewDeposits { new_deposits } => {
133                write!(f, "NewDeposits[{}]", new_deposits.len())
134            }
135            SdkEvent::UnilateralExitStateChanged => write!(f, "UnilateralExitStateChanged"),
136            SdkEvent::StableBalanceConversionFailed {
137                conversion,
138                retry_in_secs: Some(retry_in_secs),
139                ..
140            } => write!(
141                f,
142                "StableBalanceConversionFailed({conversion:?}, retry in {retry_in_secs}s)"
143            ),
144            SdkEvent::StableBalanceConversionFailed { conversion, .. } => {
145                write!(f, "StableBalanceConversionFailed({conversion:?})")
146            }
147            SdkEvent::RecoverableFunds {
148                recoverable_funds_sats,
149            } => write!(f, "RecoverableFunds[{recoverable_funds_sats}]"),
150        }
151    }
152}
153
154#[derive(Debug, Clone, Serialize)]
155#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
156pub enum AutoOptimizationEvent {
157    /// Optimization has started with the given number of rounds.
158    Started { total_rounds: u32 },
159    /// A round has completed.
160    RoundCompleted {
161        current_round: u32,
162        total_rounds: u32,
163    },
164    /// Optimization completed successfully.
165    Completed,
166    /// Optimization was cancelled.
167    Cancelled,
168    /// Optimization failed with an error.
169    Failed { error: String },
170    /// Optimization was skipped because leaves are already optimal.
171    Skipped,
172}
173
174#[allow(clippy::struct_excessive_bools)]
175#[derive(Debug, Default)]
176pub struct InternalSyncedEvent {
177    pub wallet: bool,
178    pub wallet_state: bool,
179    pub deposits: bool,
180    pub lnurl_metadata: bool,
181    pub storage_incoming: Option<u32>,
182}
183
184impl InternalSyncedEvent {
185    pub fn any(&self) -> bool {
186        self.wallet
187            || self.wallet_state
188            || self.deposits
189            || self.lnurl_metadata
190            || self.storage_incoming.is_some()
191    }
192
193    pub fn any_non_zero(&self) -> bool {
194        self.wallet
195            || self.wallet_state
196            || self.deposits
197            || self.lnurl_metadata
198            || self.storage_incoming.is_some_and(|v| v > 0)
199    }
200
201    pub fn merge(&self, other: &InternalSyncedEvent) -> Self {
202        Self {
203            wallet: self.wallet || other.wallet,
204            wallet_state: self.wallet_state || other.wallet_state,
205            deposits: self.deposits || other.deposits,
206            lnurl_metadata: self.lnurl_metadata || other.lnurl_metadata,
207            storage_incoming: self
208                .storage_incoming
209                .zip(other.storage_incoming)
210                .map(|(a, b)| a.saturating_add(b))
211                .or(self.storage_incoming)
212                .or(other.storage_incoming),
213        }
214    }
215}
216
217/// Trait for event listeners
218#[cfg_attr(feature = "uniffi", uniffi::export(callback_interface))]
219#[macros::async_trait]
220pub trait EventListener: Send + Sync {
221    /// Called when an event occurs
222    async fn on_event(&self, event: SdkEvent);
223}
224
225/// Middleware that can intercept and transform events before they reach external listeners.
226///
227/// Middleware processes events in a chain. Each middleware receives the event from the
228/// previous one and can:
229/// - Pass it through unchanged: `Some(event)`
230/// - Transform it: `Some(modified_event)`
231/// - Suppress it: `None`
232#[macros::async_trait]
233pub trait EventMiddleware: Send + Sync {
234    /// Process an event. Return `Some` to forward (possibly modified), `None` to suppress.
235    async fn process(&self, event: SdkEvent) -> Option<SdkEvent>;
236}
237
238/// Event publisher that manages event listeners and middleware.
239///
240/// Events flow through three phases:
241/// 1. Internal listeners see raw events (SDK components like `ClientSyncListener`)
242/// 2. Middleware chain can transform or suppress events
243/// 3. External listeners see processed events (client event handlers)
244pub struct EventEmitter {
245    has_real_time_sync: bool,
246    rtsync_failed: AtomicBool,
247    listener_index: AtomicU64,
248    runtime_event_handlers: RwLock<Vec<Box<dyn RuntimeEventHandler>>>,
249    /// Internal listeners see ALL events before middleware processing
250    internal_listeners: RwLock<BTreeMap<String, Box<dyn EventListener>>>,
251    /// Middleware chain that can transform/suppress events
252    middleware: RwLock<Vec<Box<dyn EventMiddleware>>>,
253    /// External listeners see events after middleware processing
254    external_listeners: RwLock<BTreeMap<String, Box<dyn EventListener>>>,
255    synced_event_buffer: Mutex<Option<InternalSyncedEvent>>,
256}
257
258/// Handlers registered here are owned by the `EventEmitter` for its whole
259/// lifetime, and the emitter is itself owned by the SDK object graph. To keep
260/// that graph droppable, a handler must not hold a strong reference back to
261/// the `EventEmitter` (directly or via a `BreezSdk` clone); the emitter is
262/// passed into `handle` instead.
263#[macros::async_trait]
264pub(crate) trait RuntimeEventHandler: Send + Sync {
265    async fn handle(&self, emitter: &EventEmitter, event: RuntimeEvent);
266}
267
268impl EventEmitter {
269    /// Create a new event emitter
270    pub fn new(has_real_time_sync: bool) -> Self {
271        Self {
272            has_real_time_sync,
273            rtsync_failed: AtomicBool::new(false),
274            listener_index: AtomicU64::new(0),
275            runtime_event_handlers: RwLock::new(Vec::new()),
276            internal_listeners: RwLock::new(BTreeMap::new()),
277            middleware: RwLock::new(Vec::new()),
278            external_listeners: RwLock::new(BTreeMap::new()),
279            synced_event_buffer: Mutex::new(Some(InternalSyncedEvent::default())),
280        }
281    }
282
283    /// Add an external listener to receive events
284    ///
285    /// # Arguments
286    ///
287    /// * `listener` - The listener to add
288    ///
289    /// # Returns
290    ///
291    /// A unique identifier for the listener, which can be used to remove it later
292    pub async fn add_external_listener(&self, listener: Box<dyn EventListener>) -> String {
293        let index = self.listener_index.fetch_add(1, Ordering::Relaxed);
294        let id = format!("listener_{}-{}", index, Uuid::new_v4());
295        let mut listeners = self.external_listeners.write().await;
296        listeners.insert(id.clone(), listener);
297        id
298    }
299
300    /// Remove an external listener by its ID
301    ///
302    /// # Arguments
303    ///
304    /// * `id` - The ID returned from `add_listener`
305    ///
306    /// # Returns
307    ///
308    /// `true` if the listener was found and removed, `false` otherwise
309    pub async fn remove_external_listener(&self, id: &str) -> bool {
310        let mut listeners = self.external_listeners.write().await;
311        listeners.remove(id).is_some()
312    }
313
314    /// Remove all external listeners.
315    ///
316    /// Listeners are owned by the emitter, so a listener that references the
317    /// SDK pins the whole instance; dropping them here on disconnect makes
318    /// the instance releasable regardless of what listeners capture.
319    pub async fn clear_external_listeners(&self) {
320        let mut listeners = self.external_listeners.write().await;
321        listeners.clear();
322    }
323
324    /// Add an internal listener that sees all raw events before middleware processing.
325    ///
326    /// Used by SDK components (e.g., `ClientSyncListener`) that need to observe events
327    /// that middleware may suppress.
328    pub async fn add_internal_listener(&self, listener: Box<dyn EventListener>) -> String {
329        let index = self.listener_index.fetch_add(1, Ordering::Relaxed);
330        let id = format!("internal_{}-{}", index, Uuid::new_v4());
331        let mut listeners = self.internal_listeners.write().await;
332        listeners.insert(id.clone(), listener);
333        id
334    }
335
336    /// Remove an internal listener by its ID
337    pub async fn remove_internal_listener(&self, id: &str) -> bool {
338        let mut listeners = self.internal_listeners.write().await;
339        listeners.remove(id).is_some()
340    }
341
342    /// Add middleware to the event processing chain.
343    ///
344    /// Middleware can transform or suppress events before they reach external listeners.
345    ///
346    /// Middleware is owned by the emitter for its whole lifetime, so it must
347    /// not hold a reference back to the emitter (directly or transitively),
348    /// or the SDK object graph can never be dropped. Code that needs to emit
349    /// must receive the emitter from its caller instead (see
350    /// `RuntimeEventHandler` and `TokenConverter::convert`).
351    pub async fn add_middleware(&self, middleware: Box<dyn EventMiddleware>) {
352        let mut mw = self.middleware.write().await;
353        mw.push(middleware);
354    }
355
356    pub(crate) async fn add_runtime_event_handler(&self, handler: Box<dyn RuntimeEventHandler>) {
357        let mut handlers = self.runtime_event_handlers.write().await;
358        handlers.push(handler);
359    }
360
361    pub(crate) async fn emit_runtime_event(&self, event: RuntimeEvent) {
362        let handlers = self.runtime_event_handlers.read().await;
363        for handler in handlers.iter() {
364            handler.handle(self, event.clone()).await;
365        }
366    }
367
368    /// Emit an event through the three-phase pipeline:
369    /// 1. Internal listeners see the raw event
370    /// 2. Middleware chain can transform or suppress
371    /// 3. External listeners see the processed event
372    pub async fn emit(&self, event: &SdkEvent) {
373        let start = Instant::now();
374        let event_label = format!("{event}");
375        let mut internal_total = std::time::Duration::ZERO;
376        let mut middleware_total = std::time::Duration::ZERO;
377        let mut external_total = std::time::Duration::ZERO;
378
379        // Phase 1: Internal listeners see raw event
380        let internal = self.internal_listeners.read().await;
381        let internal_count = internal.len();
382        for (id, listener) in internal.iter() {
383            let t = Instant::now();
384            listener.on_event(event.clone()).await;
385            let dt = t.elapsed();
386            internal_total = internal_total.saturating_add(dt);
387            info!("emit({event_label}) internal listener {id}: {dt:?}");
388        }
389        drop(internal);
390
391        // Phase 2: Middleware chain
392        let mut event = Some(event.clone());
393        let middleware = self.middleware.read().await;
394        let middleware_count = middleware.len();
395        for (i, mw) in middleware.iter().enumerate() {
396            if let Some(e) = event {
397                let t = Instant::now();
398                event = mw.process(e).await;
399                let dt = t.elapsed();
400                middleware_total = middleware_total.saturating_add(dt);
401                info!("emit({event_label}) middleware #{i}: {dt:?}");
402            } else {
403                break;
404            }
405        }
406        drop(middleware);
407
408        // Phase 3: External listeners see processed event
409        let mut external_count = 0;
410        if let Some(ref event) = event {
411            let listeners = self.external_listeners.read().await;
412            external_count = listeners.len();
413            for (id, listener) in listeners.iter() {
414                let t = Instant::now();
415                listener.on_event(event.clone()).await;
416                let dt = t.elapsed();
417                external_total = external_total.saturating_add(dt);
418                info!("emit({event_label}) external listener {id}: {dt:?}");
419            }
420        }
421
422        info!(
423            "emit({event_label}) completed in {:?} (internal[{}]={:?}, middleware[{}]={:?}, external[{}]={:?})",
424            start.elapsed(),
425            internal_count,
426            internal_total,
427            middleware_count,
428            middleware_total,
429            external_count,
430            external_total
431        );
432    }
433
434    pub async fn emit_synced(&self, synced: &InternalSyncedEvent) {
435        if !synced.any() {
436            // Nothing to emit
437            return;
438        }
439
440        let mut mtx = self.synced_event_buffer.lock().await;
441
442        let is_first_event = if let Some(buffered) = &*mtx {
443            let merged = buffered.merge(synced);
444
445            // The first synced event emitted should at least have the wallet synced.
446            // Subsequent events might have only partial syncs.
447            if merged.wallet
448                && (!self.has_real_time_sync
449                    || merged.storage_incoming.is_some()
450                    || self.rtsync_failed.load(Ordering::Relaxed))
451            {
452                *mtx = None;
453            } else {
454                *mtx = Some(merged);
455                return;
456            }
457
458            true
459        } else {
460            false
461        };
462
463        drop(mtx);
464
465        // Only emit zero real-time syncs on the first event.
466        if !is_first_event && !synced.any_non_zero() {
467            return;
468        }
469
470        // Emit the merged event
471        self.emit(&SdkEvent::Synced).await;
472    }
473
474    /// Notify that real-time sync has failed. If the first synced event is still
475    /// buffered and the wallet has already synced, release it immediately instead
476    /// of waiting for a remote pull that may never arrive.
477    pub async fn notify_rtsync_failed(&self) {
478        self.rtsync_failed.store(true, Ordering::Relaxed);
479
480        let mut mtx = self.synced_event_buffer.lock().await;
481        if let Some(buffered) = &*mtx
482            && buffered.wallet
483        {
484            *mtx = None;
485            drop(mtx);
486            self.emit(&SdkEvent::Synced).await;
487        }
488    }
489}
490
491impl Default for EventEmitter {
492    fn default() -> Self {
493        Self::new(false)
494    }
495}
496
497#[cfg(test)]
498mod tests {
499    use super::*;
500    use std::sync::Arc;
501    use std::sync::atomic::{AtomicBool, Ordering};
502
503    use macros::async_test_all;
504
505    #[cfg(feature = "browser-tests")]
506    wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
507
508    struct TestListener {
509        received: Arc<AtomicBool>,
510    }
511
512    #[macros::async_trait]
513    impl EventListener for TestListener {
514        async fn on_event(&self, _event: SdkEvent) {
515            self.received.store(true, Ordering::Relaxed);
516        }
517    }
518
519    #[async_test_all]
520    async fn test_event_emission() {
521        let emitter = EventEmitter::new(false);
522        let received = Arc::new(AtomicBool::new(false));
523
524        // Create the listener with a shared reference to the atomic boolean
525        let listener = Box::new(TestListener {
526            received: received.clone(),
527        });
528
529        let _ = emitter.add_external_listener(listener).await;
530
531        let event = SdkEvent::Synced {};
532
533        emitter.emit(&event).await;
534
535        // Check if event was received using the shared reference
536        assert!(received.load(Ordering::Relaxed));
537    }
538
539    #[async_test_all]
540    async fn test_remove_listener() {
541        let emitter = EventEmitter::new(false);
542
543        // Create shared atomic booleans to track event reception
544        let received1 = Arc::new(AtomicBool::new(false));
545        let received2 = Arc::new(AtomicBool::new(false));
546
547        // Create listeners with their own shared references
548        let listener1 = Box::new(TestListener {
549            received: received1.clone(),
550        });
551
552        let listener2 = Box::new(TestListener {
553            received: received2.clone(),
554        });
555
556        let id1 = emitter.add_external_listener(listener1).await;
557        let id2 = emitter.add_external_listener(listener2).await;
558
559        // Remove the first listener
560        assert!(emitter.remove_external_listener(&id1).await);
561
562        // Emit an event
563        let event = SdkEvent::Synced {};
564        emitter.emit(&event).await;
565
566        // The first listener should not receive the event
567        assert!(!received1.load(Ordering::Relaxed));
568
569        // The second listener should receive the event
570        assert!(received2.load(Ordering::Relaxed));
571
572        // Remove the second listener
573        assert!(emitter.remove_external_listener(&id2).await);
574
575        // Try to remove a non-existent listener
576        assert!(!emitter.remove_external_listener("non-existent-id").await);
577    }
578
579    #[async_test_all]
580    async fn test_clear_external_listeners() {
581        let emitter = EventEmitter::new(false);
582
583        let received1 = Arc::new(AtomicBool::new(false));
584        let received2 = Arc::new(AtomicBool::new(false));
585
586        let id1 = emitter
587            .add_external_listener(Box::new(TestListener {
588                received: received1.clone(),
589            }))
590            .await;
591        let id2 = emitter
592            .add_external_listener(Box::new(TestListener {
593                received: received2.clone(),
594            }))
595            .await;
596
597        emitter.clear_external_listeners().await;
598
599        emitter.emit(&SdkEvent::Synced).await;
600
601        assert!(!received1.load(Ordering::Relaxed));
602        assert!(!received2.load(Ordering::Relaxed));
603
604        // Already cleared, so individual removal finds nothing
605        assert!(!emitter.remove_external_listener(&id1).await);
606        assert!(!emitter.remove_external_listener(&id2).await);
607    }
608
609    #[async_test_all]
610    async fn test_synced_event_only_emitted_with_wallet_sync() {
611        let emitter = EventEmitter::new(false);
612        let received = Arc::new(AtomicBool::new(false));
613
614        let listener = Box::new(TestListener {
615            received: received.clone(),
616        });
617
618        emitter.add_external_listener(listener).await;
619
620        // Emit synced event without wallet sync - should NOT emit Synced
621        emitter
622            .emit_synced(&InternalSyncedEvent {
623                wallet: false,
624                wallet_state: true,
625                deposits: true,
626                lnurl_metadata: true,
627                storage_incoming: None,
628            })
629            .await;
630
631        assert!(!received.load(Ordering::Relaxed));
632
633        // Emit synced event with wallet sync - should emit Synced
634        emitter
635            .emit_synced(&InternalSyncedEvent {
636                wallet: true,
637                wallet_state: false,
638                deposits: false,
639                lnurl_metadata: false,
640                storage_incoming: Some(1),
641            })
642            .await;
643
644        assert!(received.load(Ordering::Relaxed));
645    }
646
647    #[async_test_all]
648    async fn test_has_real_time_sync_synced_event_only_emitted_with_wallet_and_storage_sync() {
649        let emitter = EventEmitter::new(true);
650        let received = Arc::new(AtomicBool::new(false));
651
652        let listener = Box::new(TestListener {
653            received: received.clone(),
654        });
655
656        emitter.add_external_listener(listener).await;
657
658        // Emit synced event with storage
659        emitter
660            .emit_synced(&InternalSyncedEvent {
661                wallet: false,
662                wallet_state: false,
663                deposits: false,
664                lnurl_metadata: false,
665                storage_incoming: Some(0),
666            })
667            .await;
668
669        assert!(!received.load(Ordering::Relaxed));
670
671        // Emit synced event with wallet sync - should emit Synced
672        emitter
673            .emit_synced(&InternalSyncedEvent {
674                wallet: true,
675                wallet_state: false,
676                deposits: false,
677                lnurl_metadata: false,
678                storage_incoming: None,
679            })
680            .await;
681
682        assert!(received.load(Ordering::Relaxed));
683    }
684
685    #[async_test_all]
686    async fn test_has_real_time_sync_synced_event_only_emitted_with_wallet_and_storage_sync_reverse()
687     {
688        let emitter = EventEmitter::new(true);
689        let received = Arc::new(AtomicBool::new(false));
690
691        let listener = Box::new(TestListener {
692            received: received.clone(),
693        });
694
695        emitter.add_external_listener(listener).await;
696
697        // Emit synced event with wallet sync
698        emitter
699            .emit_synced(&InternalSyncedEvent {
700                wallet: true,
701                wallet_state: false,
702                deposits: false,
703                lnurl_metadata: false,
704                storage_incoming: None,
705            })
706            .await;
707
708        assert!(!received.load(Ordering::Relaxed));
709
710        // Emit synced event with storage - should emit Synced
711        emitter
712            .emit_synced(&InternalSyncedEvent {
713                wallet: false,
714                wallet_state: false,
715                deposits: false,
716                lnurl_metadata: false,
717                storage_incoming: Some(0),
718            })
719            .await;
720
721        assert!(received.load(Ordering::Relaxed));
722    }
723
724    #[async_test_all]
725    async fn test_rtsync_failed_emits_synced_on_wallet_alone() {
726        let emitter = EventEmitter::new(true);
727        let received = Arc::new(AtomicBool::new(false));
728
729        let listener = Box::new(TestListener {
730            received: received.clone(),
731        });
732
733        emitter.add_external_listener(listener).await;
734
735        // Wallet synced but rtsync hasn't failed yet — should NOT emit
736        emitter
737            .emit_synced(&InternalSyncedEvent {
738                wallet: true,
739                wallet_state: false,
740                deposits: false,
741                lnurl_metadata: false,
742                storage_incoming: None,
743            })
744            .await;
745
746        assert!(!received.load(Ordering::Relaxed));
747
748        // rtsync fails — should immediately release the buffered event
749        emitter.notify_rtsync_failed().await;
750
751        assert!(received.load(Ordering::Relaxed));
752    }
753
754    #[async_test_all]
755    async fn test_rtsync_failed_before_wallet_sync_emits_on_wallet() {
756        let emitter = EventEmitter::new(true);
757        let received = Arc::new(AtomicBool::new(false));
758
759        let listener = Box::new(TestListener {
760            received: received.clone(),
761        });
762
763        emitter.add_external_listener(listener).await;
764
765        // rtsync fails before wallet syncs — nothing to release yet
766        emitter.notify_rtsync_failed().await;
767
768        assert!(!received.load(Ordering::Relaxed));
769
770        // Wallet syncs — should emit immediately (rtsync already marked failed)
771        emitter
772            .emit_synced(&InternalSyncedEvent {
773                wallet: true,
774                wallet_state: false,
775                deposits: false,
776                lnurl_metadata: false,
777                storage_incoming: None,
778            })
779            .await;
780
781        assert!(received.load(Ordering::Relaxed));
782    }
783
784    #[async_test_all]
785    async fn test_synced_event_buffers_until_wallet_sync() {
786        let emitter = EventEmitter::new(false);
787        let received = Arc::new(AtomicBool::new(false));
788
789        let listener = Box::new(TestListener {
790            received: received.clone(),
791        });
792
793        emitter.add_external_listener(listener).await;
794
795        // Emit multiple partial syncs without wallet sync
796        emitter
797            .emit_synced(&InternalSyncedEvent {
798                wallet: false,
799                wallet_state: true,
800                deposits: false,
801                lnurl_metadata: false,
802                storage_incoming: None,
803            })
804            .await;
805
806        assert!(!received.load(Ordering::Relaxed));
807
808        emitter
809            .emit_synced(&InternalSyncedEvent {
810                wallet: false,
811                wallet_state: false,
812                deposits: true,
813                lnurl_metadata: false,
814                storage_incoming: None,
815            })
816            .await;
817
818        assert!(!received.load(Ordering::Relaxed));
819
820        emitter
821            .emit_synced(&InternalSyncedEvent {
822                wallet: false,
823                wallet_state: false,
824                deposits: false,
825                lnurl_metadata: true,
826                storage_incoming: None,
827            })
828            .await;
829
830        assert!(!received.load(Ordering::Relaxed));
831
832        emitter
833            .emit_synced(&InternalSyncedEvent {
834                wallet: false,
835                wallet_state: false,
836                deposits: false,
837                lnurl_metadata: false,
838                storage_incoming: None,
839            })
840            .await;
841
842        assert!(!received.load(Ordering::Relaxed));
843
844        // Finally emit wallet sync - should emit Synced
845        emitter
846            .emit_synced(&InternalSyncedEvent {
847                wallet: true,
848                wallet_state: false,
849                deposits: false,
850                lnurl_metadata: false,
851                storage_incoming: None,
852            })
853            .await;
854
855        assert!(received.load(Ordering::Relaxed));
856    }
857
858    #[async_test_all]
859    async fn test_synced_event_all_true() {
860        let emitter = EventEmitter::new(false);
861        let received = Arc::new(AtomicBool::new(false));
862
863        let listener = Box::new(TestListener {
864            received: received.clone(),
865        });
866
867        emitter.add_external_listener(listener).await;
868
869        // Emit synced event with wallet and other components - should emit Synced
870        emitter
871            .emit_synced(&InternalSyncedEvent {
872                wallet: true,
873                wallet_state: true,
874                deposits: true,
875                lnurl_metadata: true,
876                storage_incoming: Some(1),
877            })
878            .await;
879
880        assert!(received.load(Ordering::Relaxed));
881    }
882
883    #[async_test_all]
884    async fn test_synced_event_empty_does_not_emit() {
885        let emitter = EventEmitter::new(false);
886        let received = Arc::new(AtomicBool::new(false));
887
888        let listener = Box::new(TestListener {
889            received: received.clone(),
890        });
891
892        emitter.add_external_listener(listener).await;
893
894        // Emit empty synced event - should NOT emit Synced
895        emitter
896            .emit_synced(&InternalSyncedEvent {
897                wallet: false,
898                wallet_state: false,
899                deposits: false,
900                lnurl_metadata: false,
901                storage_incoming: None,
902            })
903            .await;
904
905        assert!(!received.load(Ordering::Relaxed));
906    }
907
908    #[async_test_all]
909    async fn test_subsequent_syncs_after_wallet_emit_immediately() {
910        use std::sync::atomic::AtomicUsize;
911
912        struct CountingListener {
913            count: Arc<AtomicUsize>,
914        }
915
916        #[macros::async_trait]
917        impl EventListener for CountingListener {
918            async fn on_event(&self, event: SdkEvent) {
919                if matches!(event, SdkEvent::Synced) {
920                    self.count.fetch_add(1, Ordering::Relaxed);
921                }
922            }
923        }
924
925        let emitter = EventEmitter::new(true);
926        let count = Arc::new(AtomicUsize::new(0));
927
928        let listener = Box::new(CountingListener {
929            count: count.clone(),
930        });
931
932        emitter.add_external_listener(listener).await;
933
934        // First sync with wallet - should emit
935        emitter
936            .emit_synced(&InternalSyncedEvent {
937                wallet: true,
938                wallet_state: false,
939                deposits: false,
940                lnurl_metadata: false,
941                storage_incoming: Some(0),
942            })
943            .await;
944
945        assert_eq!(count.load(Ordering::Relaxed), 1);
946
947        // Subsequent partial sync without wallet - should emit (buffer cleared after first wallet sync)
948        emitter
949            .emit_synced(&InternalSyncedEvent {
950                wallet: false,
951                wallet_state: true,
952                deposits: false,
953                lnurl_metadata: false,
954                storage_incoming: None,
955            })
956            .await;
957
958        assert_eq!(count.load(Ordering::Relaxed), 2);
959
960        // Another partial sync - should emit
961        emitter
962            .emit_synced(&InternalSyncedEvent {
963                wallet: false,
964                wallet_state: false,
965                deposits: true,
966                lnurl_metadata: false,
967                storage_incoming: None,
968            })
969            .await;
970
971        assert_eq!(count.load(Ordering::Relaxed), 3);
972
973        emitter
974            .emit_synced(&InternalSyncedEvent {
975                wallet: false,
976                wallet_state: false,
977                deposits: false,
978                lnurl_metadata: true,
979                storage_incoming: None,
980            })
981            .await;
982
983        assert_eq!(count.load(Ordering::Relaxed), 4);
984
985        emitter
986            .emit_synced(&InternalSyncedEvent {
987                wallet: false,
988                wallet_state: false,
989                deposits: false,
990                lnurl_metadata: false,
991                storage_incoming: Some(1),
992            })
993            .await;
994
995        assert_eq!(count.load(Ordering::Relaxed), 5);
996
997        // storage_incoming with Some(0) - should NOT emit after first sync
998        emitter
999            .emit_synced(&InternalSyncedEvent {
1000                wallet: false,
1001                wallet_state: false,
1002                deposits: false,
1003                lnurl_metadata: false,
1004                storage_incoming: Some(0),
1005            })
1006            .await;
1007
1008        assert_eq!(count.load(Ordering::Relaxed), 5);
1009
1010        emitter
1011            .emit_synced(&InternalSyncedEvent {
1012                wallet: true,
1013                wallet_state: false,
1014                deposits: false,
1015                lnurl_metadata: false,
1016                storage_incoming: None,
1017            })
1018            .await;
1019
1020        assert_eq!(count.load(Ordering::Relaxed), 6);
1021    }
1022
1023    // ── Helpers for middleware / internal listener tests ──
1024
1025    /// Listener that records all received events
1026    struct RecordingListener {
1027        events: Arc<Mutex<Vec<String>>>,
1028    }
1029
1030    impl RecordingListener {
1031        fn new() -> (Self, Arc<Mutex<Vec<String>>>) {
1032            let events = Arc::new(Mutex::new(Vec::new()));
1033            (
1034                Self {
1035                    events: events.clone(),
1036                },
1037                events,
1038            )
1039        }
1040    }
1041
1042    #[macros::async_trait]
1043    impl EventListener for RecordingListener {
1044        async fn on_event(&self, event: SdkEvent) {
1045            self.events.lock().await.push(format!("{event}"));
1046        }
1047    }
1048
1049    /// Middleware that suppresses all Synced events
1050    struct SuppressSyncedMiddleware;
1051
1052    #[macros::async_trait]
1053    impl EventMiddleware for SuppressSyncedMiddleware {
1054        async fn process(&self, event: SdkEvent) -> Option<SdkEvent> {
1055            match event {
1056                SdkEvent::Synced => None,
1057                other => Some(other),
1058            }
1059        }
1060    }
1061
1062    /// Middleware that replaces `PaymentSucceeded` with `PaymentPending`
1063    struct DowngradePaymentMiddleware;
1064
1065    #[macros::async_trait]
1066    impl EventMiddleware for DowngradePaymentMiddleware {
1067        async fn process(&self, event: SdkEvent) -> Option<SdkEvent> {
1068            match event {
1069                SdkEvent::PaymentSucceeded { payment } => {
1070                    Some(SdkEvent::PaymentPending { payment })
1071                }
1072                other => Some(other),
1073            }
1074        }
1075    }
1076
1077    /// Middleware that suppresses all events
1078    struct SuppressAllMiddleware;
1079
1080    #[macros::async_trait]
1081    impl EventMiddleware for SuppressAllMiddleware {
1082        async fn process(&self, _event: SdkEvent) -> Option<SdkEvent> {
1083            None
1084        }
1085    }
1086
1087    fn test_payment() -> Payment {
1088        Payment {
1089            id: "test-id".to_string(),
1090            payment_type: crate::PaymentType::Receive,
1091            status: crate::PaymentStatus::Completed,
1092            amount: 1000,
1093            fees: 10,
1094            timestamp: 123_456,
1095            method: crate::PaymentMethod::Spark,
1096            details: None,
1097            conversion_details: None,
1098        }
1099    }
1100
1101    // ── Internal listener tests ──
1102
1103    #[async_test_all]
1104    async fn test_internal_listener_receives_events() {
1105        let emitter = EventEmitter::new(false);
1106        let (listener, events) = RecordingListener::new();
1107
1108        emitter.add_internal_listener(Box::new(listener)).await;
1109
1110        emitter.emit(&SdkEvent::Synced).await;
1111
1112        let log = events.lock().await;
1113        assert_eq!(log.len(), 1);
1114        assert!(log[0].contains("Synced"));
1115    }
1116
1117    #[async_test_all]
1118    async fn test_remove_internal_listener() {
1119        let emitter = EventEmitter::new(false);
1120        let (listener, events) = RecordingListener::new();
1121
1122        let id = emitter.add_internal_listener(Box::new(listener)).await;
1123
1124        assert!(emitter.remove_internal_listener(&id).await);
1125        assert!(!emitter.remove_internal_listener(&id).await);
1126
1127        emitter.emit(&SdkEvent::Synced).await;
1128
1129        assert!(events.lock().await.is_empty());
1130    }
1131
1132    // ── Middleware tests ──
1133
1134    #[async_test_all]
1135    async fn test_middleware_suppresses_event_for_external() {
1136        let emitter = EventEmitter::new(false);
1137        let (ext, ext_events) = RecordingListener::new();
1138
1139        emitter.add_external_listener(Box::new(ext)).await;
1140        emitter
1141            .add_middleware(Box::new(SuppressSyncedMiddleware))
1142            .await;
1143
1144        emitter.emit(&SdkEvent::Synced).await;
1145
1146        // External should NOT see the suppressed event
1147        assert!(ext_events.lock().await.is_empty());
1148    }
1149
1150    #[async_test_all]
1151    async fn test_middleware_transforms_event() {
1152        let emitter = EventEmitter::new(false);
1153        let (ext, ext_events) = RecordingListener::new();
1154
1155        emitter.add_external_listener(Box::new(ext)).await;
1156        emitter
1157            .add_middleware(Box::new(DowngradePaymentMiddleware))
1158            .await;
1159
1160        let event = SdkEvent::PaymentSucceeded {
1161            payment: test_payment(),
1162        };
1163        emitter.emit(&event).await;
1164
1165        let log = ext_events.lock().await;
1166        assert_eq!(log.len(), 1);
1167        assert!(log[0].contains("PaymentPending"));
1168    }
1169
1170    #[async_test_all]
1171    async fn test_middleware_passthrough_unmatched_events() {
1172        let emitter = EventEmitter::new(false);
1173        let (ext, ext_events) = RecordingListener::new();
1174
1175        emitter.add_external_listener(Box::new(ext)).await;
1176        emitter
1177            .add_middleware(Box::new(SuppressSyncedMiddleware))
1178            .await;
1179
1180        // Synced is suppressed, PaymentSucceeded passes through
1181        emitter.emit(&SdkEvent::Synced).await;
1182        let event = SdkEvent::PaymentSucceeded {
1183            payment: test_payment(),
1184        };
1185        emitter.emit(&event).await;
1186
1187        let log = ext_events.lock().await;
1188        assert_eq!(log.len(), 1);
1189        assert!(log[0].contains("PaymentSucceeded"));
1190    }
1191
1192    #[async_test_all]
1193    async fn test_middleware_chain_ordering() {
1194        let emitter = EventEmitter::new(false);
1195        let (ext, ext_events) = RecordingListener::new();
1196
1197        emitter.add_external_listener(Box::new(ext)).await;
1198
1199        // First middleware transforms PaymentSucceeded → PaymentPending
1200        emitter
1201            .add_middleware(Box::new(DowngradePaymentMiddleware))
1202            .await;
1203        // Second middleware suppresses Synced (doesn't affect PaymentPending)
1204        emitter
1205            .add_middleware(Box::new(SuppressSyncedMiddleware))
1206            .await;
1207
1208        let event = SdkEvent::PaymentSucceeded {
1209            payment: test_payment(),
1210        };
1211        emitter.emit(&event).await;
1212
1213        let log = ext_events.lock().await;
1214        assert_eq!(log.len(), 1);
1215        assert!(log[0].contains("PaymentPending"));
1216    }
1217
1218    #[async_test_all]
1219    async fn test_suppress_all_middleware_stops_chain() {
1220        let emitter = EventEmitter::new(false);
1221        let (ext, ext_events) = RecordingListener::new();
1222
1223        emitter.add_external_listener(Box::new(ext)).await;
1224
1225        // SuppressAll first — nothing should reach the next middleware or external
1226        emitter
1227            .add_middleware(Box::new(SuppressAllMiddleware))
1228            .await;
1229        emitter
1230            .add_middleware(Box::new(DowngradePaymentMiddleware))
1231            .await;
1232
1233        emitter.emit(&SdkEvent::Synced).await;
1234        let event = SdkEvent::PaymentSucceeded {
1235            payment: test_payment(),
1236        };
1237        emitter.emit(&event).await;
1238
1239        assert!(ext_events.lock().await.is_empty());
1240    }
1241
1242    // ── Three-phase flow tests ──
1243
1244    #[async_test_all]
1245    async fn test_three_phase_emit_flow() {
1246        let emitter = EventEmitter::new(false);
1247        let (int, int_events) = RecordingListener::new();
1248        let (ext, ext_events) = RecordingListener::new();
1249
1250        emitter.add_internal_listener(Box::new(int)).await;
1251        emitter.add_external_listener(Box::new(ext)).await;
1252        emitter
1253            .add_middleware(Box::new(SuppressSyncedMiddleware))
1254            .await;
1255
1256        // Synced: internal sees it, middleware suppresses it, external doesn't
1257        emitter.emit(&SdkEvent::Synced).await;
1258
1259        assert_eq!(int_events.lock().await.len(), 1);
1260        assert!(ext_events.lock().await.is_empty());
1261
1262        // PaymentSucceeded: both see it (middleware passes it through)
1263        let event = SdkEvent::PaymentSucceeded {
1264            payment: test_payment(),
1265        };
1266        emitter.emit(&event).await;
1267
1268        assert_eq!(int_events.lock().await.len(), 2);
1269        assert_eq!(ext_events.lock().await.len(), 1);
1270    }
1271
1272    #[async_test_all]
1273    async fn test_internal_sees_raw_event_external_sees_transformed() {
1274        let emitter = EventEmitter::new(false);
1275        let (int, int_events) = RecordingListener::new();
1276        let (ext, ext_events) = RecordingListener::new();
1277
1278        emitter.add_internal_listener(Box::new(int)).await;
1279        emitter.add_external_listener(Box::new(ext)).await;
1280        emitter
1281            .add_middleware(Box::new(DowngradePaymentMiddleware))
1282            .await;
1283
1284        let event = SdkEvent::PaymentSucceeded {
1285            payment: test_payment(),
1286        };
1287        emitter.emit(&event).await;
1288
1289        let int_log = int_events.lock().await;
1290        let ext_log = ext_events.lock().await;
1291
1292        // Internal sees the original PaymentSucceeded
1293        assert_eq!(int_log.len(), 1);
1294        assert!(int_log[0].contains("PaymentSucceeded"));
1295
1296        // External sees the transformed PaymentPending
1297        assert_eq!(ext_log.len(), 1);
1298        assert!(ext_log[0].contains("PaymentPending"));
1299    }
1300
1301    #[async_test_all]
1302    async fn test_no_listeners_no_middleware_does_not_panic() {
1303        let emitter = EventEmitter::new(false);
1304        emitter.emit(&SdkEvent::Synced).await;
1305        // Should not panic
1306    }
1307
1308    #[async_test_all]
1309    async fn test_empty_event_does_not_emit_after_wallet_sync() {
1310        use std::sync::atomic::AtomicUsize;
1311
1312        struct CountingListener {
1313            count: Arc<AtomicUsize>,
1314        }
1315
1316        #[macros::async_trait]
1317        impl EventListener for CountingListener {
1318            async fn on_event(&self, event: SdkEvent) {
1319                if matches!(event, SdkEvent::Synced) {
1320                    self.count.fetch_add(1, Ordering::Relaxed);
1321                }
1322            }
1323        }
1324
1325        let emitter = EventEmitter::new(false);
1326        let count = Arc::new(AtomicUsize::new(0));
1327
1328        let listener = Box::new(CountingListener {
1329            count: count.clone(),
1330        });
1331
1332        emitter.add_external_listener(listener).await;
1333
1334        // First sync with wallet - should emit
1335        emitter
1336            .emit_synced(&InternalSyncedEvent {
1337                wallet: true,
1338                wallet_state: false,
1339                deposits: false,
1340                lnurl_metadata: false,
1341                storage_incoming: None,
1342            })
1343            .await;
1344
1345        assert_eq!(count.load(Ordering::Relaxed), 1);
1346
1347        // Empty sync after wallet sync - should NOT emit (all fields false)
1348        emitter
1349            .emit_synced(&InternalSyncedEvent {
1350                wallet: false,
1351                wallet_state: false,
1352                deposits: false,
1353                lnurl_metadata: false,
1354                storage_incoming: None,
1355            })
1356            .await;
1357
1358        assert_eq!(count.load(Ordering::Relaxed), 1); // Count should remain 1
1359
1360        // Another non-empty sync - should emit
1361        emitter
1362            .emit_synced(&InternalSyncedEvent {
1363                wallet: false,
1364                wallet_state: true,
1365                deposits: false,
1366                lnurl_metadata: false,
1367                storage_incoming: None,
1368            })
1369            .await;
1370
1371        assert_eq!(count.load(Ordering::Relaxed), 2); // Now count should be 2
1372    }
1373}
breez_sdk_spark/events.rs

breez_sdk_spark/
events.rs
