lb_rs/service/
events.rs

1use db_rs::View;
2pub use tokio::sync::broadcast::{self, Receiver, Sender};
3use tracing::*;
4use uuid::Uuid;
5
6use crate::io::CoreV5;
7use crate::{Lb, LbErrKind};
8
9#[derive(Clone)]
10pub struct EventSubs {
11    tx: Sender<Event>,
12}
13
14#[derive(Clone, Debug)]
15pub enum Event {
16    /// A metadata for a given id or it's descendants changed. The id returned
17    /// may be deleted. Updates to document contents will not cause this
18    /// message to be sent (unless a document was deleted).
19    MetadataChanged(Actor),
20
21    /// The contents of this document have changed either by this lb
22    /// library or as a result of sync
23    DocumentWritten(Uuid, Actor),
24
25    /// Changes from another process have been applied to our local views.
26    /// Cached metadata and document contents may need to be reloaded.
27    IpcChangesApplied,
28
29    PendingSharesChanged,
30
31    Sync(SyncIncrement),
32
33    StatusUpdated,
34
35    UserSignedIn,
36}
37
38#[derive(Debug, Clone, PartialEq)]
39pub enum Actor {
40    /// A write initiated locally. The id identifies the writer (e.g. a
41    /// workspace instance) so subscribers can tell their own writes apart
42    /// from writes by other workspaces or tools sharing this lb.
43    User(Option<Uuid>),
44    Sync,
45}
46
47impl Default for EventSubs {
48    fn default() -> Self {
49        let (tx, _) = broadcast::channel::<Event>(10000);
50        Self { tx }
51    }
52}
53
54impl EventSubs {
55    pub(crate) fn pending_shares_changed(&self) {
56        self.queue(Event::PendingSharesChanged);
57    }
58
59    pub(crate) fn meta_changed(&self, actor: Actor) {
60        self.queue(Event::MetadataChanged(actor));
61    }
62
63    pub(crate) fn doc_written(&self, id: Uuid, actor: Actor) {
64        self.queue(Event::DocumentWritten(id, actor));
65    }
66
67    pub(crate) fn sync_update(&self, s: SyncIncrement) {
68        self.queue(Event::Sync(s));
69    }
70
71    pub(crate) fn status_updated(&self) {
72        self.queue(Event::StatusUpdated);
73    }
74
75    /// executed after root and account are created
76    pub(crate) fn signed_in(&self) {
77        self.queue(Event::UserSignedIn);
78    }
79
80    fn queue(&self, evt: Event) {
81        if let Err(e) = self.tx.send(evt.clone()) {
82            error!(?evt, ?e, "could not queue");
83        }
84    }
85}
86
87impl Lb {
88    pub(crate) async fn notify_catch_up(&self, db: &CoreV5, previous_seq: u64) {
89        if db.account.last_modified() > previous_seq && self.get_account().is_err() {
90            if let Some(account) = db.account.as_ref() {
91                self.keychain.cache_account(account.clone()).await.unwrap();
92                self.events.signed_in();
93            }
94        }
95
96        let changed = [
97            db.account.last_modified(),
98            db.root.last_modified(),
99            db.base_metadata.last_modified(),
100            db.local_metadata.last_modified(),
101            db.pub_key_lookup.last_modified(),
102            db.pinned_files.last_modified(),
103            db.last_synced.last_modified(),
104        ]
105        .into_iter()
106        .any(|seq| seq > previous_seq);
107
108        // Read activity is deliberately excluded: reloading documents records
109        // more reads, which must not trigger reloads in the other process.
110        if changed {
111            self.events.queue(Event::IpcChangesApplied);
112        }
113    }
114
115    pub fn subscribe(&self) -> Receiver<Event> {
116        self.events.tx.subscribe()
117    }
118}
119
120#[derive(Debug, Clone)]
121pub enum SyncIncrement {
122    SyncStarted,
123    PullingDocument(Uuid, bool),
124    PushingDocument(Uuid, bool),
125    SyncFinished(Option<LbErrKind>),
126}
lb_rs/service/events.rs

lb_rs/service/
events.rs
