clean up

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",

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
 

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;

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`
+impl Update {
-/// until the receiving end is dropped or the client shuts down.
+    /// Build an update from a received event.
-///
+    pub fn from_event(event: &Event) -> Self {
-/// Spawn this once, e.g. inside `cx.background_spawn`.
-pub async fn run(client: Client, tx: Sender<Update>) {
-    let mut notifications = client.notifications();
-
-    while let Some(notification) = notifications.next().await {
-        let ClientNotification::Event { event, .. } = notification else {
-            continue;
-        };
-
         let coordinate = event
             .tags
             .iter()
@@ -32,15 +22,11 @@ pub async fn run(client: Client, tx: Sender<Update>) {
             .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;
         }
     }
 }

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>) -> Self {
-        // Pump: relays -> LMDB (automatic) -> flume -> BackendEvent::NostrUpdate.
-        let (tx, rx) = flume::bounded::<Update>(4096);
         let client = inner.client();
 
-        let pump = cx.background_spawn(async move {
+        let pump = cx.spawn(async move |this, cx| {
-            signed_nostr::pump::run(client, tx).await;
+            let mut notifications = client.notifications();
-            Ok(())
-        });
 
-        let forward = cx.spawn(async move |this, cx| {
+            while let Some(notification) = notifications.next().await {
-            while let Ok(update) = rx.recv_async().await {
+                let ClientNotification::Event { event, .. } = notification else {
-                this.update(cx, |_, cx| cx.emit(BackendEvent::NostrUpdate(update)))?;
+                    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| {
+                    this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
-                        cx.emit(BackendEvent::error(e.to_string()))
-                    })?;
                 }
             }
             Ok(())
@@ -155,15 +160,11 @@ impl Backend {
     pub fn subscribe(&mut self, filter: Filter, cx: &mut Context<Self>) {
         let backend = self.inner.clone();
 
-        let task = cx.background_spawn(async move {
+        let task = cx.background_spawn(async move { backend.subscribe(filter).await.map(|_| ()) });
-            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| {
+                this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
-                    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<Self>) {
+    ///
-        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<Self>,
+    ) -> flume::Receiver<Result<Event, Error>> {
+        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| {
+                    this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
-                        cx.emit(BackendEvent::error(e.to_string()))
-                    })?;
                 }
             }
-            Ok(())
+
+            tx.send_async(result)
+                .await
+                .map_err(|_| anyhow!("action result receiver dropped"))
         }));
+
+        rx
     }
 }

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<Announcement>,
     /// `(refname, commit-id)` pairs from the latest state announcement.
     pub refs: Vec<(String, String)>,
@@ -20,9 +19,12 @@ pub struct RepoStore {
     pub patches: Vec<Event>,
     pub pull_requests: Vec<Event>,
     statuses: Vec<Event>,
-
+    /// Error of the last action initiated from this store, if any.
-    _subscription: Subscription,
+    pub last_error: Option<String>,
+    refreshing: bool,
+    refresh_dirty: bool,
     tasks: Vec<Task<Result<(), Error>>>,
+    _subscription: Subscription,
 }
 
 impl RepoStore {
@@ -37,8 +39,7 @@ impl RepoStore {
                             && update.author == this.addr.owner)
                 }
                 BackendEvent::Published(event) => {
-                    event.kind == Kind::GitRepoAnnouncement
+                    event.kind == Kind::GitRepoAnnouncement && event.pubkey == this.addr.owner
-                        && 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<Self>) {
+        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 announcements = db.query(filters::announcement(&addr)).await?;
-            let states = db.query(filters::state(&addr)).await?;
+                    let states = db.query(filters::state(&addr)).await?;
-            let activity = db.query(filters::activity(&addr)).await?;
+                    let activity = db.query(filters::activity(&addr)).await?;
 
-            this.update(cx, |this, cx| {
+                    Ok::<_, Error>((announcements, states, activity))
-                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;
                 }
+                .await;
 
-                this.issues.clear();
+                let (announcements, states, activity) = match queries {
-                this.patches.clear();
+                    Ok(results) => results,
-                this.pull_requests.clear();
+                    Err(e) => {
-                this.statuses.clear();
+                        return this.update(cx, |this, cx| {
-
+                            this.refreshing = false;
-                for event in activity {
+                            this.last_error = Some(e.to_string());
-                    match event.kind {
+                            cx.notify();
-                        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 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)
+        let builder = EventBuilder::new(Kind::GitPatch, patch).tags([
-            .tags([
+            Tag::coordinate(self.addr.coordinate(), None),
-                Tag::coordinate(self.addr.coordinate(), None),
+            Tag::public_key(self.addr.owner),
-                Tag::public_key(self.addr.owner),
+            root_marker,
-                root_marker,
+        ]);
-            ]);
 
         self.send(builder, cx);
     }
@@ -197,8 +243,23 @@ impl RepoStore {
         self.send(builder, cx);
     }
 
-    fn send(&self, builder: EventBuilder, cx: &mut Context<Self>) {
+    fn send(&mut self, builder: EventBuilder, cx: &mut Context<Self>) {
-        Backend::global(cx).update(cx, |backend, cx| backend.send(builder, cx));
+        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);
     }
 }
 

crates/signed_state/src/repo_list.rs

@@ -11,9 +11,10 @@ use crate::backend::{Backend, BackendEvent};
 pub struct RepoListStore {
     pub announcements: Vec<Announcement>,
     author: Option<PublicKey>,
-
+    refreshing: bool,
-    _subscription: Subscription,
+    refresh_dirty: bool,
     tasks: Vec<Task<Result<(), Error>>>,
+    _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<Self>) {
+        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 {
+            loop {
-                Some(a) => filters::announcements_by(a),
+                let filter = match author {
-                None => filters::all_announcements(500),
+                    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 again = this.update(cx, |this, cx| {
-                let mut by_repo: HashMap<(String, String), Announcement> = HashMap::new();
+                    let mut by_repo: HashMap<(String, String), Announcement> = HashMap::new();
 
-                for event in events {
+                    for event in events {
-                    let Some(announcement) = Announcement::from_event(&event) else {
+                        let Some(announcement) = Announcement::from_event(&event) else {
-                        continue;
+                            continue;
-                    };
+                        };
 
-                    let key = (announcement.owner.to_hex(), announcement.id.clone());
+                        let key = (announcement.owner.to_hex(), announcement.id.clone());
 
-                    match by_repo.get(&key) {
+                        match by_repo.get(&key) {
-                        Some(existing) if existing.created_at >= announcement.created_at => {}
+                            Some(existing) if existing.created_at >= announcement.created_at => {}
-                        _ => {
+                            _ => {
-                            by_repo.insert(key, announcement);
+                                by_repo.insert(key, announcement);
+                            }
                         }
                     }
+
+                    let mut announcements: Vec<Announcement> = 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<Announcement> = by_repo.into_values().collect();
-                announcements.sort_by_key(|a| std::cmp::Reverse(a.created_at));
-
-                this.announcements = announcements;
-                cx.notify();
-            })?;
 
             Ok(())
         });