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

breez_sdk_spark/
events.rs
