use negentropy sync

crates/signed_state/src/backend.rs

@@ -1,8 +1,10 @@
+use std::collections::HashMap;
 use std::time::Duration;
 
 use anyhow::{Error, anyhow};
 use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Task};
 use nostr_connect::prelude::*;
+use nostr_sdk::client::SyncSummary;
 use nostr_sdk::prelude::*;
 use signed_core::filters;
 use signed_nostr::{NostrBackend, SignedAuthUrlHandler, UniversalSigner, Update};
@@ -39,6 +41,10 @@ pub enum BackendEvent {
     Connected,
     /// A new event was received from a relay and stored in the database.
     NostrUpdate(Update),
+    /// A negentropy sync completed; the database was updated directly,
+    /// so stores should re-query (no [`BackendEvent::NostrUpdate`] is fired
+    /// for synced events).
+    Synced,
     /// An event built locally was signed, broadcast and stored.
     Published(Box<Event>),
     /// An error occurred.
@@ -479,6 +485,56 @@ impl Backend {
         }));
     }
 
+    /// Start a one-shot subscription targeted only at the bootstrap relays,
+    /// auto-closing after EOSE or a short timeout. Matching events are stored
+    /// in the database and surface as [`BackendEvent::NostrUpdate`] while the
+    /// subscription is open.
+    pub fn subscribe_bootstrap(&mut self, filters: Vec<Filter>, cx: &mut Context<Self>) {
+        let backend = self.inner.clone();
+
+        let task = cx.background_spawn(async move {
+            subscribe_bootstrap_only(&backend.client(), filters).await
+        });
+
+        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())))?;
+            }
+            Ok(())
+        }));
+    }
+
+    /// Negentropy-sync the given filter against the bootstrap relays:
+    /// reconciles the local database with the relays in both directions.
+    /// Emits [`BackendEvent::Synced`] on completion.
+    pub fn sync_bootstrap(&mut self, filter: Filter, cx: &mut Context<Self>) {
+        let backend = self.inner.clone();
+
+        let task = cx.background_spawn(async move {
+            sync_bootstrap_only(&backend.client(), filter).await
+        });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            match task.await {
+                Ok(summary) => {
+                    log::debug!(
+                        "sync done: {} received, {} sent",
+                        summary.received.len(),
+                        summary.sent.len()
+                    );
+                    this.update(cx, |_this, cx| {
+                        cx.emit(BackendEvent::Synced);
+                        cx.notify();
+                    })?;
+                }
+                Err(e) => {
+                    this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
+                }
+            }
+            Ok(())
+        }));
+    }
+
     /// Sign, broadcast and locally store an event. Emits
     /// [`BackendEvent::Published`] on success so stores can refresh.
     ///
@@ -517,3 +573,30 @@ impl Backend {
         rx
     }
 }
+
+/// Subscribe only on the bootstrap relays, auto-closing after EOSE or a
+/// short timeout. Use for one-shot data fetches (repo events, profiles)
+/// instead of persistent gossip-routed subscriptions.
+pub(crate) async fn subscribe_bootstrap_only(client: &Client, filters: Vec<Filter>) -> Result<(), Error> {
+    let opts = SubscribeAutoCloseOptions::default()
+        .exit_policy(ReqExitPolicy::ExitOnEOSE)
+        .timeout(Some(Duration::from_secs(10)));
+
+    let target: HashMap<&str, Vec<Filter>> = BOOTSTRAP_RELAYS
+        .iter()
+        .map(|relay| (*relay, filters.clone()))
+        .collect();
+
+    client.subscribe(target).close_on(opts).await?;
+
+    Ok(())
+}
+
+/// Negentropy-sync the filter against the bootstrap relays only.
+pub(crate) async fn sync_bootstrap_only(
+    client: &Client,
+    filter: Filter,
+) -> Result<SyncSummary, Error> {
+    let output = client.sync(filter).with(BOOTSTRAP_RELAYS).await?;
+    Ok(output.value)
+}

crates/signed_state/src/profile.rs

@@ -5,7 +5,7 @@ use anyhow::Error;
 use gpui::{App, Context, Entity, Global, SharedString, Subscription, Task};
 use nostr_sdk::prelude::*;
 
-use crate::backend::{Backend, BackendEvent};
+use crate::backend::{Backend, BackendEvent, sync_bootstrap_only};
 
 /// A user profile (kind `0` metadata), as plain data for the UI.
 #[derive(Debug, Clone)]
@@ -181,6 +181,50 @@ impl ProfileStore {
         self.tasks.push(task);
     }
 
+    /// Re-read the latest metadata of every requested author from the local
+    /// database (used after a sync, which produces no NostrUpdate events).
+    fn apply_seen(&mut self, cx: &mut Context<Self>) {
+        if self.seen.is_empty() {
+            return;
+        }
+
+        let client = Backend::global(cx).read(cx).client();
+        let authors: Vec<PublicKey> = self.seen.iter().copied().collect();
+
+        let task = cx.spawn(async move |this, cx| {
+            let filter = Filter::new().kind(Kind::Metadata).authors(authors);
+            let events = client.database().query(filter).await?;
+
+            let mut latest: HashMap<PublicKey, (Timestamp, Metadata)> = HashMap::new();
+            for event in events {
+                match latest.get(&event.pubkey) {
+                    Some((ts, _)) if *ts >= event.created_at => {}
+                    _ => {
+                        latest.insert(
+                            event.pubkey,
+                            (
+                                event.created_at,
+                                Metadata::from_json(&event.content).unwrap_or_default(),
+                            ),
+                        );
+                    }
+                }
+            }
+
+            this.update(cx, |this, cx| {
+                for (public_key, (_, metadata)) in latest {
+                    this.profiles
+                        .insert(public_key, Profile::new(public_key, metadata));
+                }
+                cx.notify();
+            })?;
+
+            Ok(())
+        });
+
+        self.tasks.push(task);
+    }
+
     /// Drain the queue in a batched fetch, debounced to collect requests.
     fn queue_fetch(&mut self, cx: &mut Context<Self>) {
         if self.fetching {
@@ -210,10 +254,16 @@ impl ProfileStore {
                     .kind(Kind::Metadata)
                     .authors(batch.into_iter().collect::<Vec<PublicKey>>());
 
-                // Gossip routes the fetch to each author's relays. Fetched
+                // Negentropy-sync with the bootstrap relays. Synced events
-                // events land in the database and surface via NostrUpdate.
+                // are written to the database directly (no NostrUpdate), so
-                if let Err(e) = client.fetch_events(filter).await {
+                // re-apply from the database afterwards.
-                    log::warn!("profile fetch failed: {e}");
+                match sync_bootstrap_only(&client, filter).await {
+                    Ok(_) => {
+                        this.update(cx, |this, cx| this.apply_seen(cx))?;
+                    }
+                    Err(e) => {
+                        log::warn!("profile sync failed: {e}");
+                    }
                 }
             }
 

crates/signed_state/src/repo.rs

@@ -78,14 +78,20 @@ impl RepoStore {
         &self.addr
     }
 
-    /// Subscribe the relay pool to this repository's activity.
+    /// Fetch this repository's events from the bootstrap relays (one-shot,
+    /// auto-closing subscription).
     fn subscribe_remote(&mut self, cx: &mut Context<Self>) {
         let addr = self.addr.clone();
 
         Backend::global(cx).update(cx, |backend, cx| {
-            backend.subscribe(filters::announcement(&addr), cx);
+            backend.subscribe_bootstrap(
-            backend.subscribe(filters::state(&addr), cx);
+                vec![
-            backend.subscribe(filters::activity(&addr), cx);
+                    filters::announcement(&addr),
+                    filters::state(&addr),
+                    filters::activity(&addr),
+                ],
+                cx,
+            );
         });
     }
 

crates/signed_state/src/repo_list.rs

@@ -31,6 +31,7 @@ impl RepoListStore {
                     event.kind == Kind::GitRepoAnnouncement
                         && this.author.is_none_or(|a| a == event.pubkey)
                 }
+                BackendEvent::Synced => true,
                 _ => false,
             };
 
@@ -60,6 +61,7 @@ impl RepoListStore {
         self.refresh(cx);
     }
 
+    /// Negentropy-sync announcements with the bootstrap relays.
     fn subscribe_remote(&mut self, cx: &mut Context<Self>) {
         let backend = Backend::global(cx);
         let author = self.author;
@@ -69,7 +71,7 @@ impl RepoListStore {
                 Some(a) => filters::announcements_by(a),
                 None => filters::all_announcements(500),
             };
-            backend.subscribe(filter, cx);
+            backend.sync_bootstrap(filter, cx);
         });
     }