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

breez_sdk_spark/
events.rs
