diff --git a/Cargo.lock b/Cargo.lock index 4428f5e..763b2d9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6815,7 +6815,6 @@ name = "signed_nostr" version = "1.0.0" dependencies = [ "anyhow", - "flume 0.11.1", "nostr", "nostr-connect", "nostr-gossip-memory", diff --git a/crates/signed_nostr/Cargo.toml b/crates/signed_nostr/Cargo.toml index 9d321c5..4ee5f32 100644 --- a/crates/signed_nostr/Cargo.toml +++ b/crates/signed_nostr/Cargo.toml @@ -12,7 +12,6 @@ nostr-sdk.workspace = true nostr-connect.workspace = true nostr-gossip-memory.workspace = true -flume.workspace = true anyhow.workspace = true webbrowser.workspace = true diff --git a/crates/signed_nostr/src/lib.rs b/crates/signed_nostr/src/lib.rs index 27e6cee..1328bf4 100644 --- a/crates/signed_nostr/src/lib.rs +++ b/crates/signed_nostr/src/lib.rs @@ -1,7 +1,7 @@ mod backend; -pub mod pump; mod signer; +mod update; pub use backend::NostrBackend; -pub use pump::Update; pub use signer::{SignedAuthUrlHandler, UniversalSigner}; +pub use update::Update; diff --git a/crates/signed_nostr/src/pump.rs b/crates/signed_nostr/src/update.rs similarity index 53% rename from crates/signed_nostr/src/pump.rs rename to crates/signed_nostr/src/update.rs index 3e5cf3b..1f2222f 100644 --- a/crates/signed_nostr/src/pump.rs +++ b/crates/signed_nostr/src/update.rs @@ -1,4 +1,3 @@ -use flume::Sender; use nostr_sdk::prelude::*; /// A lightweight "something changed" signal for the UI. @@ -13,18 +12,9 @@ pub struct Update { pub event_id: EventId, } -/// Consume the client notification stream and forward [`Update`]s into `tx` -/// until the receiving end is dropped or the client shuts down. -/// -/// Spawn this once, e.g. inside `cx.background_spawn`. -pub async fn run(client: Client, tx: Sender) { - let mut notifications = client.notifications(); - - while let Some(notification) = notifications.next().await { - let ClientNotification::Event { event, .. } = notification else { - continue; - }; - +impl Update { + /// Build an update from a received event. + pub fn from_event(event: &Event) -> Self { let coordinate = event .tags .iter() @@ -32,15 +22,11 @@ pub async fn run(client: Client, tx: Sender) { .and_then(|t| t.content()) .map(str::to_owned); - let update = Update { + Self { kind: event.kind, coordinate, author: event.pubkey, event_id: event.id, - }; - - if tx.send_async(update).await.is_err() { - break; } } } diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index cdb4c15..e9f2a79 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -1,7 +1,7 @@ -use anyhow::Error; +use anyhow::{Error, anyhow}; use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Task}; use nostr_sdk::prelude::*; -use signed_nostr::{NostrBackend, UniversalSigner, pump::Update}; +use signed_nostr::{NostrBackend, UniversalSigner, Update}; #[derive(Debug, Clone)] pub enum BackendEvent { @@ -54,26 +54,33 @@ impl Backend { } pub(crate) fn new(inner: NostrBackend, cx: &mut Context) -> Self { - // Pump: relays -> LMDB (automatic) -> flume -> BackendEvent::NostrUpdate. - let (tx, rx) = flume::bounded::(4096); let client = inner.client(); - let pump = cx.background_spawn(async move { - signed_nostr::pump::run(client, tx).await; - Ok(()) - }); + let pump = cx.spawn(async move |this, cx| { + let mut notifications = client.notifications(); - let forward = cx.spawn(async move |this, cx| { - while let Ok(update) = rx.recv_async().await { - this.update(cx, |_, cx| cx.emit(BackendEvent::NostrUpdate(update)))?; + while let Some(notification) = notifications.next().await { + let ClientNotification::Event { event, .. } = notification else { + continue; + }; + + let update = Update::from_event(&event); + + if this + .update(cx, |_, cx| cx.emit(BackendEvent::NostrUpdate(update))) + .is_err() + { + break; + } } + Ok(()) }); Self { inner, current_user: None, - tasks: vec![pump, forward], + tasks: vec![pump], } } @@ -141,9 +148,7 @@ impl Backend { this.update(cx, |_this, cx| cx.emit(BackendEvent::Connected))?; } Err(e) => { - this.update(cx, |_this, cx| { - cx.emit(BackendEvent::error(e.to_string())) - })?; + this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?; } } Ok(()) @@ -155,15 +160,11 @@ impl Backend { pub fn subscribe(&mut self, filter: Filter, cx: &mut Context) { let backend = self.inner.clone(); - let task = cx.background_spawn(async move { - backend.subscribe(filter).await.map(|_| ()) - }); + let task = cx.background_spawn(async move { backend.subscribe(filter).await.map(|_| ()) }); self.tasks.push(cx.spawn(async move |this, cx| { if let Err(e) = task.await { - this.update(cx, |_this, cx| { - cx.emit(BackendEvent::error(e.to_string())) - })?; + this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?; } Ok(()) })); @@ -171,25 +172,39 @@ impl Backend { /// Sign, broadcast and locally store an event. Emits /// [`BackendEvent::Published`] on success so stores can refresh. - pub fn send(&mut self, builder: EventBuilder, cx: &mut Context) { - let backend = self.inner.clone(); + /// + /// The returned receiver yields the outcome of this specific action, + /// so callers can show inline progress/errors instead of relying on + /// the global [`BackendEvent::Error`]. + pub fn send( + &mut self, + builder: EventBuilder, + cx: &mut Context, + ) -> flume::Receiver> { + let (tx, rx) = flume::bounded(1); + let backend = self.inner.clone(); let task = cx.background_spawn(async move { backend.send(builder).await }); self.tasks.push(cx.spawn(async move |this, cx| { - match task.await { + let result = task.await; + + match &result { Ok(event) => { this.update(cx, |_this, cx| { - cx.emit(BackendEvent::Published(Box::new(event))); + cx.emit(BackendEvent::Published(Box::new(event.clone()))); })?; } Err(e) => { - this.update(cx, |_this, cx| { - cx.emit(BackendEvent::error(e.to_string())) - })?; + this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?; } } - Ok(()) + + tx.send_async(result) + .await + .map_err(|_| anyhow!("action result receiver dropped")) })); + + rx } } diff --git a/crates/signed_state/src/repo.rs b/crates/signed_state/src/repo.rs index ed9efb8..e288145 100644 --- a/crates/signed_state/src/repo.rs +++ b/crates/signed_state/src/repo.rs @@ -10,7 +10,6 @@ use crate::backend::{Backend, BackendEvent}; pub struct RepoStore { addr: RepoAddr, addr_string: String, - pub announcement: Option, /// `(refname, commit-id)` pairs from the latest state announcement. pub refs: Vec<(String, String)>, @@ -20,9 +19,12 @@ pub struct RepoStore { pub patches: Vec, pub pull_requests: Vec, statuses: Vec, - - _subscription: Subscription, + /// Error of the last action initiated from this store, if any. + pub last_error: Option, + refreshing: bool, + refresh_dirty: bool, tasks: Vec>>, + _subscription: Subscription, } impl RepoStore { @@ -37,8 +39,7 @@ impl RepoStore { && update.author == this.addr.owner) } BackendEvent::Published(event) => { - event.kind == Kind::GitRepoAnnouncement - && event.pubkey == this.addr.owner + event.kind == Kind::GitRepoAnnouncement && event.pubkey == this.addr.owner || event.tags.iter().any(|t| { t.kind() == "a" && t.content() == Some(this.addr_string.as_str()) }) @@ -61,6 +62,9 @@ impl RepoStore { patches: Vec::new(), pull_requests: Vec::new(), statuses: Vec::new(), + last_error: None, + refreshing: false, + refresh_dirty: false, _subscription: subscription, tasks: Vec::new(), }; @@ -86,49 +90,92 @@ impl RepoStore { } /// Re-query the local database and update all fields. + /// + /// Debounced: concurrent requests are coalesced into a single re-query + /// after the running one finishes. pub fn refresh(&mut self, cx: &mut Context) { + if self.refreshing { + self.refresh_dirty = true; + return; + } + self.refreshing = true; + let client = Backend::global(cx).read(cx).client(); let addr = self.addr.clone(); let task = cx.spawn(async move |this, cx| { - let db = client.database(); + loop { + let queries = async { + let db = client.database(); - let announcements = db.query(filters::announcement(&addr)).await?; - let states = db.query(filters::state(&addr)).await?; - let activity = db.query(filters::activity(&addr)).await?; + let announcements = db.query(filters::announcement(&addr)).await?; + let states = db.query(filters::state(&addr)).await?; + let activity = db.query(filters::activity(&addr)).await?; - this.update(cx, |this, cx| { - this.announcement = latest(announcements).as_ref().and_then(Announcement::from_event); - - if let Some(state) = latest(states) { - let (refs, head) = parse_state(&state); - this.refs = refs; - this.head = head; + Ok::<_, Error>((announcements, states, activity)) } + .await; - this.issues.clear(); - this.patches.clear(); - this.pull_requests.clear(); - this.statuses.clear(); - - for event in activity { - match event.kind { - Kind::GitIssue => this.issues.push(event), - Kind::GitPatch => this.patches.push(event), - Kind::GitPullRequest | Kind::GitPullRequestUpdate => { - this.pull_requests.push(event) - } - kind if RepoStatus::from_kind(kind).is_some() => this.statuses.push(event), - _ => {} + let (announcements, states, activity) = match queries { + Ok(results) => results, + Err(e) => { + return this.update(cx, |this, cx| { + this.refreshing = false; + this.last_error = Some(e.to_string()); + cx.notify(); + }); } + }; + + let again = this.update(cx, |this, cx| { + this.announcement = latest(announcements) + .as_ref() + .and_then(Announcement::from_event); + + if let Some(state) = latest(states) { + let (refs, head) = parse_state(&state); + this.refs = refs; + this.head = head; + } + + this.issues.clear(); + this.patches.clear(); + this.pull_requests.clear(); + this.statuses.clear(); + + for event in activity { + match event.kind { + Kind::GitIssue => this.issues.push(event), + Kind::GitPatch => this.patches.push(event), + Kind::GitPullRequest | Kind::GitPullRequestUpdate => { + this.pull_requests.push(event) + } + kind if RepoStatus::from_kind(kind).is_some() => { + this.statuses.push(event) + } + _ => {} + } + } + + sort_newest_first(&mut this.issues); + sort_newest_first(&mut this.patches); + sort_newest_first(&mut this.pull_requests); + + cx.notify(); + + if this.refresh_dirty { + this.refresh_dirty = false; + true + } else { + this.refreshing = false; + false + } + })?; + + if !again { + break; } - - sort_newest_first(&mut this.issues); - sort_newest_first(&mut this.patches); - sort_newest_first(&mut this.pull_requests); - - cx.notify(); - })?; + } Ok(()) }); @@ -171,12 +218,11 @@ impl RepoStore { return; }; - let builder = EventBuilder::new(Kind::GitPatch, patch) - .tags([ - Tag::coordinate(self.addr.coordinate(), None), - Tag::public_key(self.addr.owner), - root_marker, - ]); + let builder = EventBuilder::new(Kind::GitPatch, patch).tags([ + Tag::coordinate(self.addr.coordinate(), None), + Tag::public_key(self.addr.owner), + root_marker, + ]); self.send(builder, cx); } @@ -197,8 +243,23 @@ impl RepoStore { self.send(builder, cx); } - fn send(&self, builder: EventBuilder, cx: &mut Context) { - Backend::global(cx).update(cx, |backend, cx| backend.send(builder, cx)); + fn send(&mut self, builder: EventBuilder, cx: &mut Context) { + self.last_error = None; + + let rx = Backend::global(cx).update(cx, |backend, cx| backend.send(builder, cx)); + + let task = cx.spawn(async move |this, cx| { + if let Ok(Err(e)) = rx.recv_async().await { + this.update(cx, |this, cx| { + this.last_error = Some(e.to_string()); + cx.notify(); + })?; + } + + Ok(()) + }); + + self.tasks.push(task); } } diff --git a/crates/signed_state/src/repo_list.rs b/crates/signed_state/src/repo_list.rs index fe8ef9b..7d88dfd 100644 --- a/crates/signed_state/src/repo_list.rs +++ b/crates/signed_state/src/repo_list.rs @@ -11,9 +11,10 @@ use crate::backend::{Backend, BackendEvent}; pub struct RepoListStore { pub announcements: Vec, author: Option, - - _subscription: Subscription, + refreshing: bool, + refresh_dirty: bool, tasks: Vec>>, + _subscription: Subscription, } impl RepoListStore { @@ -40,6 +41,8 @@ impl RepoListStore { let mut store = Self { announcements: Vec::new(), author, + refreshing: false, + refresh_dirty: false, _subscription: subscription, tasks: Vec::new(), }; @@ -69,42 +72,72 @@ impl RepoListStore { } /// Re-query the local database. Latest announcement per repository wins. + /// + /// Debounced: concurrent requests are coalesced into a single re-query + /// after the running one finishes. pub fn refresh(&mut self, cx: &mut Context) { + if self.refreshing { + self.refresh_dirty = true; + return; + } + self.refreshing = true; + let client = Backend::global(cx).read(cx).client(); let author = self.author; let task = cx.spawn(async move |this, cx| { - let filter = match author { - Some(a) => filters::announcements_by(a), - None => filters::all_announcements(500), - }; + loop { + let filter = match author { + Some(a) => filters::announcements_by(a), + None => filters::all_announcements(500), + }; - let events = client.database().query(filter).await?; + let events = match client.database().query(filter).await { + Ok(events) => events, + Err(_) => { + return this.update(cx, |this, _cx| { + this.refreshing = false; + }); + } + }; - this.update(cx, |this, cx| { - let mut by_repo: HashMap<(String, String), Announcement> = HashMap::new(); + let again = this.update(cx, |this, cx| { + let mut by_repo: HashMap<(String, String), Announcement> = HashMap::new(); - for event in events { - let Some(announcement) = Announcement::from_event(&event) else { - continue; - }; + for event in events { + let Some(announcement) = Announcement::from_event(&event) else { + continue; + }; - let key = (announcement.owner.to_hex(), announcement.id.clone()); + let key = (announcement.owner.to_hex(), announcement.id.clone()); - match by_repo.get(&key) { - Some(existing) if existing.created_at >= announcement.created_at => {} - _ => { - by_repo.insert(key, announcement); + match by_repo.get(&key) { + Some(existing) if existing.created_at >= announcement.created_at => {} + _ => { + by_repo.insert(key, announcement); + } } } + + let mut announcements: Vec = by_repo.into_values().collect(); + announcements.sort_by_key(|a| std::cmp::Reverse(a.created_at)); + + this.announcements = announcements; + cx.notify(); + + if this.refresh_dirty { + this.refresh_dirty = false; + true + } else { + this.refreshing = false; + false + } + })?; + + if !again { + break; } - - let mut announcements: Vec = by_repo.into_values().collect(); - announcements.sort_by_key(|a| std::cmp::Reverse(a.created_at)); - - this.announcements = announcements; - cx.notify(); - })?; + } Ok(()) });