diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index d288578..f47c6e3 100644 --- a/crates/signed_state/src/backend.rs +++ b/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), /// 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, cx: &mut Context) { + 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) { + 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) -> Result<(), Error> { + let opts = SubscribeAutoCloseOptions::default() + .exit_policy(ReqExitPolicy::ExitOnEOSE) + .timeout(Some(Duration::from_secs(10))); + + let target: HashMap<&str, Vec> = 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 { + let output = client.sync(filter).with(BOOTSTRAP_RELAYS).await?; + Ok(output.value) +} diff --git a/crates/signed_state/src/profile.rs b/crates/signed_state/src/profile.rs index b72f62c..64742e3 100644 --- a/crates/signed_state/src/profile.rs +++ b/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) { + if self.seen.is_empty() { + return; + } + + let client = Backend::global(cx).read(cx).client(); + let authors: Vec = 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 = 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) { if self.fetching { @@ -210,10 +254,16 @@ impl ProfileStore { .kind(Kind::Metadata) .authors(batch.into_iter().collect::>()); - // Gossip routes the fetch to each author's relays. Fetched - // events land in the database and surface via NostrUpdate. - if let Err(e) = client.fetch_events(filter).await { - log::warn!("profile fetch failed: {e}"); + // Negentropy-sync with the bootstrap relays. Synced events + // are written to the database directly (no NostrUpdate), so + // re-apply from the database afterwards. + 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}"); + } } } diff --git a/crates/signed_state/src/repo.rs b/crates/signed_state/src/repo.rs index e288145..43de9ef 100644 --- a/crates/signed_state/src/repo.rs +++ b/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) { let addr = self.addr.clone(); Backend::global(cx).update(cx, |backend, cx| { - backend.subscribe(filters::announcement(&addr), cx); - backend.subscribe(filters::state(&addr), cx); - backend.subscribe(filters::activity(&addr), cx); + backend.subscribe_bootstrap( + vec![ + filters::announcement(&addr), + filters::state(&addr), + filters::activity(&addr), + ], + cx, + ); }); } diff --git a/crates/signed_state/src/repo_list.rs b/crates/signed_state/src/repo_list.rs index 379e678..ed83a7b 100644 --- a/crates/signed_state/src/repo_list.rs +++ b/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) { 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); }); }