Replace relay subscriptions with cache-backed discovery sync

crates/nip29/src/discovery.rs

@@ -1,71 +1,27 @@
 use std::collections::{HashMap, HashSet};
 
-use gpui::Context;
+use futures::future::join_all;
 use nostr_sdk::prelude::*;
 
-use crate::protocol::{GroupCandidate, GroupId, GroupKey, GroupList};
+use crate::protocol::{GroupCandidate, GroupId, GroupKey, GroupList, GroupMetadata};
+use crate::utils::parse_members;
 
-/// The groups discovered from the cache and the relays.
+#[derive(Default, Clone, Debug)]
-#[derive(Default)]
 pub struct Discovery {
     refs: HashMap<RelayUrl, HashSet<GroupId>>,
     discovered: Vec<GroupCandidate>,
     members: HashMap<GroupKey, (Timestamp, usize)>,
-    loading: bool,
 }
 
 impl Discovery {
+    /// The discovered groups, sorted by name.
     pub fn discovered(&self) -> &[GroupCandidate] {
         &self.discovered
     }
 
-    pub fn loading(&self) -> bool {
+    /// The member count of a group, when its member list is known.
-        self.loading
+    pub fn members(&self, key: &GroupKey) -> Option<usize> {
-    }
+        self.members.get(key).map(|(_, members)| *members)
-
-    /// The discovery is starting over: nothing has answered yet.
-    pub(crate) fn begin(&mut self, cx: &mut Context<Self>) {
-        self.loading = true;
-        cx.notify();
-    }
-
-    /// Every pending discovery step has finished.
-    pub(crate) fn settle(&mut self, cx: &mut Context<Self>) {
-        self.loading = false;
-        cx.notify();
-    }
-
-    /// Absorb the references of a group list, skipping the joined groups.
-    ///
-    /// Returns whether new references were added.
-    pub(crate) fn absorb_list(
-        &mut self,
-        list: &GroupList,
-        joined: &HashSet<(RelayUrl, GroupId)>,
-        cx: &mut Context<Self>,
-    ) -> bool {
-        let mut added = false;
-
-        for entry in list.entries() {
-            if joined.contains(&(entry.relay.clone(), entry.id.clone())) {
-                continue;
-            }
-
-            if self
-                .refs
-                .entry(entry.relay.clone())
-                .or_default()
-                .insert(entry.id.clone())
-            {
-                added = true;
-            }
-        }
-
-        if added {
-            cx.notify();
-        }
-
-        added
     }
 
     /// The (relay, group) references, for asking the relays for metadata.
@@ -73,8 +29,49 @@ impl Discovery {
         &self.refs
     }
 
+    /// Absorb the references of a group list, skipping the joined groups.
+    pub(crate) fn absorb_list(&mut self, list: &GroupList, joined: &HashSet<(RelayUrl, GroupId)>) {
+        for entry in list.entries() {
+            if joined.contains(&(entry.relay.clone(), entry.id.clone())) {
+                continue;
+            }
+
+            self.refs
+                .entry(entry.relay.clone())
+                .or_default()
+                .insert(entry.id.clone());
+        }
+    }
+
+    /// Absorb the newest group list of every author.
+    pub(crate) fn absorb_lists(
+        &mut self,
+        lists: Vec<Event>,
+        joined: &HashSet<(RelayUrl, GroupId)>,
+    ) {
+        let newest: HashMap<PublicKey, Event> =
+            lists.into_iter().fold(HashMap::new(), |mut newest, event| {
+                match newest.get(&event.pubkey) {
+                    Some(current) if current.created_at >= event.created_at => newest,
+                    _ => {
+                        newest.insert(event.pubkey, event);
+                        newest
+                    }
+                }
+            });
+
+        for event in newest.into_values() {
+            let Ok(list) = GroupList::parse(&event) else {
+                log::warn!("ignoring a malformed group list from {}", event.pubkey);
+                continue;
+            };
+
+            self.absorb_list(&list, joined);
+        }
+    }
+
     /// Add a discovered group, keeping the newest metadata.
-    pub(crate) fn upsert(&mut self, candidate: GroupCandidate, cx: &mut Context<Self>) {
+    pub(crate) fn upsert(&mut self, candidate: GroupCandidate) {
         match self.discovered.iter_mut().find(|d| d.key == candidate.key) {
             Some(existing) if existing.created_at <= candidate.created_at => {
                 *existing = candidate;
@@ -88,18 +85,10 @@ impl Discovery {
                 .to_lowercase()
                 .cmp(&b.display_name().to_lowercase())
         });
-
-        cx.notify();
     }
 
     /// Apply a member count, keeping the newest list.
-    pub(crate) fn set_members(
+    pub fn set_members(&mut self, key: GroupKey, created_at: Timestamp, members: usize) {
-        &mut self,
-        key: GroupKey,
-        created_at: Timestamp,
-        members: usize,
-        cx: &mut Context<Self>,
-    ) {
         if self
             .members
             .get(&key)
@@ -108,21 +97,191 @@ impl Discovery {
             return;
         }
 
-        if self.members.insert(key, (created_at, members)) != Some((created_at, members)) {
+        self.members.insert(key, (created_at, members));
-            cx.notify();
+    }
+
+    /// Seed the discoveries from the cached group metadata.
+    pub(crate) fn absorb_cached_previews(&mut self, previews: Vec<Event>) {
+        let newest: HashMap<GroupId, (Timestamp, GroupMetadata)> =
+            previews
+                .into_iter()
+                .fold(HashMap::new(), |mut newest, event| {
+                    let metadata = match GroupMetadata::parse(&event) {
+                        Ok(metadata) => metadata,
+                        Err(error) => {
+                            log::warn!("ignoring malformed cached group metadata: {error}");
+                            return newest;
+                        }
+                    };
+
+                    let Some(id) = metadata.id().cloned() else {
+                        return newest;
+                    };
+
+                    match newest.get(&id) {
+                        Some((created_at, _)) if *created_at >= event.created_at => newest,
+                        _ => {
+                            newest.insert(id, (event.created_at, metadata));
+                            newest
+                        }
+                    }
+                });
+
+        for (relay, ids) in self.refs.clone() {
+            for id in ids {
+                let Some((created_at, metadata)) = newest.get(&id) else {
+                    continue;
+                };
+
+                // Hidden groups are not discoverable (NIP-29).
+                if metadata.hidden {
+                    continue;
+                }
+
+                self.upsert(GroupCandidate {
+                    key: GroupKey::new(relay.clone(), id.clone()),
+                    created_at: *created_at,
+                    metadata: metadata.clone(),
+                });
+            }
         }
     }
 
-    /// The member count of a group, when its member list is known.
+    /// Seed the member counts from the cached member lists.
-    pub fn members(&self, key: &GroupKey) -> Option<usize> {
+    pub(crate) fn absorb_cached_members(&mut self, members: Vec<Event>) {
-        self.members.get(key).map(|(_, members)| *members)
+        let newest: HashMap<GroupId, (Timestamp, usize)> =
+            members
+                .into_iter()
+                .fold(HashMap::new(), |mut newest, event| {
+                    let Some((id, created_at, count)) = parse_members(&event) else {
+                        return newest;
+                    };
+
+                    match newest.get(&id) {
+                        Some((current, _)) if *current >= created_at => newest,
+                        _ => {
+                            newest.insert(id, (created_at, count));
+                            newest
+                        }
+                    }
+                });
+
+        for (relay, ids) in self.refs.clone() {
+            for id in ids {
+                let Some((created_at, count)) = newest.get(&id) else {
+                    continue;
+                };
+
+                self.set_members(
+                    GroupKey::new(relay.clone(), id.clone()),
+                    *created_at,
+                    *count,
+                );
+            }
+        }
     }
 
-    pub(crate) fn reset(&mut self, cx: &mut Context<Self>) {
+    /// Load the discoverable groups from the local database only.
-        self.refs.clear();
+    pub async fn cached(client: Client, joined: HashSet<(RelayUrl, GroupId)>) -> Self {
-        self.discovered.clear();
+        let mut discovery = Self::default();
-        self.members.clear();
+
-        self.loading = false;
+        // The group lists of every cached author.
-        cx.notify();
+        let lists: Vec<Event> = client
+            .database()
+            .query(Filter::new().kind(Kind::SimpleGroups))
+            .await
+            .map(|events| events.into_iter().collect())
+            .unwrap_or_default();
+
+        discovery.absorb_lists(lists, &joined);
+
+        let refs = discovery.refs().clone();
+
+        if refs.is_empty() {
+            return discovery;
+        }
+
+        let ids: HashSet<&str> = refs.values().flatten().map(GroupId::as_str).collect();
+
+        // The cached metadata of the referenced groups.
+        let previews: Vec<Event> = client
+            .database()
+            .query(
+                Filter::new()
+                    .kind(Kind::GroupMetadata)
+                    .identifiers(ids.clone()),
+            )
+            .await
+            .map(|events| events.into_iter().collect())
+            .unwrap_or_default();
+
+        discovery.absorb_cached_previews(previews);
+
+        // The cached member lists of the referenced groups.
+        let members: Vec<Event> = client
+            .database()
+            .query(Filter::new().kind(Kind::GroupMembers).identifiers(ids))
+            .await
+            .map(|events| events.into_iter().collect())
+            .unwrap_or_default();
+
+        discovery.absorb_cached_members(members);
+
+        discovery
+    }
+
+    /// Reconcile the discoverable groups with the relays.
+    pub async fn sync(
+        &mut self,
+        client: Client,
+        me: PublicKey,
+        joined: HashSet<(RelayUrl, GroupId)>,
+    ) {
+        let filter = Filter::new().author(me).kind(Kind::ContactList).limit(1);
+
+        let authors: HashSet<PublicKey> = client
+            .database()
+            .query(filter)
+            .await
+            .ok()
+            .and_then(|events| events.into_iter().next())
+            .map(|event| event.tags.public_keys().collect())
+            .unwrap_or_default();
+
+        let urls: Vec<RelayUrl> = client.relays().await.into_keys().collect();
+        let filter = Filter::new().authors(authors).kind(Kind::SimpleGroups);
+        client.sync(filter.clone()).with(urls).await.ok();
+
+        let lists: Vec<Event> = client
+            .database()
+            .query(filter)
+            .await
+            .map(|events| events.into_iter().collect())
+            .unwrap_or_default();
+
+        self.absorb_lists(lists, &joined);
+
+        // The pool-wide sweep of the group metadata and members.
+        let urls: Vec<RelayUrl> = client.relays().await.into_keys().collect();
+        let filter = Filter::new().kinds([Kind::GroupMetadata, Kind::GroupMembers]);
+        client.sync(filter).with(urls).await.ok();
+
+        // The previews cover the relays the lists just referenced.
+        let fetches = self.refs.clone().into_iter().map(|(relay, ids)| {
+            let client = client.clone();
+
+            async move {
+                client.add_relay(relay.clone()).and_connect().await.ok();
+
+                let filter = Filter::new()
+                    .kinds([Kind::GroupMetadata, Kind::GroupMembers])
+                    .identifiers(ids.iter().map(GroupId::as_str));
+
+                let urls = vec![relay];
+                client.sync(filter).with(urls).await.ok();
+            }
+        });
+
+        join_all(fetches).await;
     }
 }

crates/nip29/src/lib.rs

@@ -1,6 +1,4 @@
-use std::collections::hash_map::DefaultHasher;
 use std::collections::{HashMap, HashSet};
-use std::hash::{Hash, Hasher};
 use std::sync::LazyLock;
 use std::time::Duration;
 
@@ -13,18 +11,15 @@ use state::NostrRegistry;
 mod discovery;
 mod group;
 mod protocol;
+mod utils;
 
 pub use discovery::Discovery;
 pub use group::*;
 pub use protocol::*;
+pub(crate) use utils::*;
 
 pub(crate) static LOCAL_KEYS: LazyLock<Keys> = LazyLock::new(Keys::generate);
 
-const LIST_SUBSCRIPTION: &str = "nip29-list";
-const FORK_SUBSCRIPTION: &str = "nip29-forks";
-const GROUP_PREFIX: &str = "grp-";
-const BROWSE_PREFIX: &str = "brw-";
-const PAGE_PREFIX: &str = "pg-";
 const BROWSE_LIMIT: usize = 500;
 const BROWSE_TIMEOUT: Duration = Duration::from_secs(10);
 const FORK_INTERVAL: Duration = Duration::from_secs(15 * 60);
@@ -43,6 +38,14 @@ pub enum GroupsEvent {
     Error(String),
     Browsed(Box<GroupCandidate>),
     BrowseSettled(RelayUrl),
+    /// A browsed group announced its member list.
+    Members {
+        key: GroupKey,
+        created_at: Timestamp,
+        members: usize,
+    },
+    /// The discovery synced new groups into the local database.
+    Synced,
 }
 
 #[derive(Debug)]
@@ -51,105 +54,13 @@ enum Signal {
     Eose(SubscriptionId),
     Closed(SubscriptionId, String),
     BrowseSettled(SubscriptionId),
+    /// A reconciliation started on a relay.
+    Syncing(RelayUrl),
+    /// A relay finished answering a reconciliation.
+    Synced(RelayUrl),
     Forks,
 }
 
-fn list_subscription_id() -> SubscriptionId {
-    SubscriptionId::new(LIST_SUBSCRIPTION)
-}
-
-fn fork_subscription_id() -> SubscriptionId {
-    SubscriptionId::new(FORK_SUBSCRIPTION)
-}
-
-fn subscription_id(key: &GroupKey) -> SubscriptionId {
-    SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32))
-}
-
-fn browse_subscription(relay: &RelayUrl) -> SubscriptionId {
-    let mut hasher = DefaultHasher::new();
-    relay.hash(&mut hasher);
-
-    SubscriptionId::new(format!("{BROWSE_PREFIX}{:08x}", hasher.finish() as u32))
-}
-
-fn page_subscription(key: &GroupKey) -> SubscriptionId {
-    SubscriptionId::new(format!("{PAGE_PREFIX}{:08x}", key.uniq_id() as u32))
-}
-
-fn is_page(id: &SubscriptionId) -> bool {
-    id.as_str().starts_with(PAGE_PREFIX)
-}
-
-fn is_own_subscription(id: &SubscriptionId) -> bool {
-    let id = id.as_str();
-    id == LIST_SUBSCRIPTION
-        || id == FORK_SUBSCRIPTION
-        || id.starts_with(GROUP_PREFIX)
-        || id.starts_with(BROWSE_PREFIX)
-        || id.starts_with(PAGE_PREFIX)
-}
-
-fn parse_members(event: &Event) -> Option<(GroupId, Timestamp, usize)> {
-    if event.kind != Kind::GroupMembers {
-        return None;
-    }
-
-    let id = GroupId::new(event.tags.identifier()?).ok()?;
-    let members = event.tags.public_keys().collect::<HashSet<_>>().len();
-
-    Some((id, event.created_at, members))
-}
-
-async fn reconcile(
-    client: &Client,
-    filter: Filter,
-    urls: Vec<RelayUrl>,
-) -> HashMap<EventId, HashSet<RelayUrl>> {
-    let output = match client.sync(filter).with(urls).await {
-        Ok(output) => output,
-        Err(error) => {
-            log::warn!("the reconciliation failed: {error}");
-            return HashMap::new();
-        }
-    };
-
-    for (relay, error) in &output.failed {
-        log::debug!("syncing with {relay} failed: {error}");
-    }
-
-    output.value.received
-}
-
-/// Pair the downloaded events with the relay that served them.
-async fn pull_received(
-    client: &Client,
-    received: HashMap<EventId, HashSet<RelayUrl>>,
-) -> Result<Vec<(RelayUrl, Event)>, Error> {
-    if received.is_empty() {
-        return Ok(Vec::new());
-    }
-
-    let events: HashMap<EventId, Event> = client
-        .database()
-        .query(Filter::new().ids(received.keys().cloned()))
-        .await?
-        .into_iter()
-        .map(|event| (event.id, event))
-        .collect();
-
-    Ok(received
-        .into_iter()
-        .flat_map(|(id, relays)| match events.get(&id) {
-            Some(event) => relays
-                .into_iter()
-                .map(|relay| (relay, event.clone()))
-                .collect(),
-            None => Vec::new(),
-        })
-        .collect())
-}
-
 impl EventEmitter<GroupsEvent> for GroupsRegistry {}
 
 pub struct GroupsRegistry {
@@ -160,8 +71,8 @@ pub struct GroupsRegistry {
     observers: HashMap<GroupKey, Subscription>,
     forks: HashMap<PublicKey, (Timestamp, GroupList)>,
     fork_authors: HashSet<PublicKey>,
-    /// The groups discovered from the cache and the relays.
+    /// The relays with a discovery reconciliation in flight.
-    discovery: Entity<Discovery>,
+    syncing: HashSet<RelayUrl>,
     signal_tx: flume::Sender<Signal>,
     signal_rx: flume::Receiver<Signal>,
     notification_listener: Option<Task<()>>,
@@ -215,7 +126,7 @@ impl GroupsRegistry {
             observers: HashMap::new(),
             forks: HashMap::new(),
             fork_authors: HashSet::new(),
-            discovery: cx.new(|_| Discovery::default()),
+            syncing: HashSet::new(),
             signal_tx: tx,
             signal_rx: rx,
             notification_listener: None,
@@ -346,543 +257,42 @@ impl GroupsRegistry {
         cx.emit(GroupsEvent::BrowseSettled(relay));
     }
 
-    /// The groups discovered from the cache and the relays.
+    /// The (relay, group) pairs of every joined group.
-    pub fn discovery(&self) -> Entity<Discovery> {
+    fn joined_keys(&self, cx: &App) -> HashSet<(RelayUrl, GroupId)> {
-        self.discovery.clone()
+        self.groups
-    }
-
-    pub fn discover(&mut self, cx: &mut Context<Self>) {
-        let nostr = NostrRegistry::global(cx);
-        let client = nostr.read(cx).client();
-        let signer = nostr.read(cx).signer();
-
-        if nostr.read(cx).current_user().is_none() {
-            // There is nothing to discover without a user.
-            self.discovery.update(cx, |discovery, cx| {
-                discovery.settle(cx);
-            });
-            return;
-        }
-
-        // Every open sweeps the relays again, starting from the cache.
-        self.discovery.update(cx, |discovery, cx| {
-            discovery.begin(cx);
-        });
-
-        type Cached = (HashSet<PublicKey>, Vec<Event>);
-
-        let task: Task<Result<Cached, Error>> = cx.background_spawn(async move {
-            let me = signer.get_public_key_async().await?;
-            let filter = Filter::new().author(me).kind(Kind::ContactList).limit(1);
-
-            let contacts = client
-                .database()
-                .query(filter)
-                .await?
-                .into_iter()
-                .next()
-                .map(|event| event.tags.public_keys().collect())
-                .unwrap_or_default();
-
-            let lists = client
-                .database()
-                .query(Filter::new().kind(Kind::SimpleGroups))
-                .await?
-                .into_iter()
-                .collect();
-
-            Ok((contacts, lists))
-        });
-
-        self.tasks.push(cx.spawn(async move |this, cx| {
-            let (authors, lists) = match task.await {
-                Ok(ok) => ok,
-                Err(error) => {
-                    log::warn!("loading the cached groups failed: {error}");
-                    (HashSet::new(), Vec::new())
-                }
-            };
-
-            if let Err(error) = this.update(cx, |this, cx| {
-                this.absorb_lists(lists, cx);
-            }) {
-                log::warn!("applying the cached group lists failed: {error}");
-                return;
-            }
-
-            let previews = match this.update(cx, |this, cx| this.load_cached_previews(cx)) {
-                Ok(task) => match task.await {
-                    Ok(previews) => previews,
-                    Err(error) => {
-                        log::warn!("loading the cached group metadata failed: {error}");
-                        Vec::new()
-                    }
-                },
-                Err(error) => {
-                    log::warn!("discovering the groups failed: {error}");
-                    return;
-                }
-            };
-
-            // Cached member lists fill the cards' member counts.
-            let members = match this.update(cx, |this, cx| this.load_cached_members(cx)) {
-                Ok(task) => match task.await {
-                    Ok(members) => members,
-                    Err(error) => {
-                        log::warn!("loading the cached group members failed: {error}");
-                        Vec::new()
-                    }
-                },
-                Err(error) => {
-                    log::warn!("discovering the groups failed: {error}");
-                    return;
-                }
-            };
-
-            if let Err(error) = this.update(cx, |this, cx| {
-                this.absorb_cached_previews(previews, cx);
-                this.absorb_cached_members(members, cx);
-            }) {
-                log::warn!("discovering the groups failed: {error}");
-                return;
-            }
-
-            let lists = match this.update(cx, |this, cx| this.sync_lists(authors, cx)) {
-                Ok(task) => task,
-                Err(error) => {
-                    log::warn!("discovering the groups failed: {error}");
-                    return;
-                }
-            };
-
-            // The pool-wide sweep runs while the lists are reconciling.
-            let sweep = match this.update(cx, |this, cx| this.sync_sweep(cx)) {
-                Ok(task) => task,
-                Err(error) => {
-                    log::warn!("discovering the groups failed: {error}");
-                    return;
-                }
-            };
-
-            let lists = match lists.await {
-                Ok(lists) => lists,
-                Err(error) => {
-                    log::warn!("syncing the contacts group lists failed: {error}");
-                    Vec::new()
-                }
-            };
-
-            if let Err(error) = this.update(cx, |this, cx| {
-                this.absorb_lists(lists, cx);
-            }) {
-                log::warn!("applying the contacts group lists failed: {error}");
-            }
-
-            let sweep = match sweep.await {
-                Ok(events) => events,
-                Err(error) => {
-                    log::warn!("syncing the discovered groups failed: {error}");
-                    Vec::new()
-                }
-            };
-
-            if let Err(error) = this.update(cx, |this, cx| {
-                for (relay, event) in sweep {
-                    this.absorb_discovered(relay, &event, cx);
-                }
-            }) {
-                log::warn!("applying the discovered groups failed: {error}");
-            }
-
-            // The previews cover the relays the lists just referenced.
-            let previews = match this.update(cx, |this, cx| this.sync_previews(cx)) {
-                Ok(task) => match task.await {
-                    Ok(previews) => previews,
-                    Err(error) => {
-                        log::warn!("syncing the group previews failed: {error}");
-                        Vec::new()
-                    }
-                },
-                Err(error) => {
-                    log::warn!("discovering the groups failed: {error}");
-                    return;
-                }
-            };
-
-            if let Err(error) = this.update(cx, |this, cx| {
-                for (relay, event) in previews {
-                    this.absorb_preview(relay, &event, cx);
-                }
-
-                this.discovery
-                    .update(cx, |discovery, cx| discovery.settle(cx));
-            }) {
-                log::warn!("discovering the groups failed: {error}");
-            }
-        }));
-    }
-
-    /// Load the cached metadata for the referenced groups from the database.
-    fn load_cached_previews(&self, cx: &mut Context<Self>) -> Task<Result<Vec<Event>, Error>> {
-        let refs = self.discovery.read(cx).refs().clone();
-
-        if refs.is_empty() {
-            return Task::ready(Ok(Vec::new()));
-        }
-
-        let nostr = NostrRegistry::global(cx);
-        let client = nostr.read(cx).client();
-
-        let ids: HashSet<&str> = refs.values().flatten().map(GroupId::as_str).collect();
-        let filter = Filter::new().kind(Kind::GroupMetadata).identifiers(ids);
-
-        cx.background_spawn(async move {
-            Ok(client.database().query(filter).await?.into_iter().collect())
-        })
-    }
-
-    /// Seed the discoveries from the cached group metadata.
-    fn absorb_cached_previews(&mut self, previews: Vec<Event>, cx: &mut Context<Self>) {
-        let newest: HashMap<GroupId, (Timestamp, GroupMetadata)> =
-            previews
-                .into_iter()
-                .fold(HashMap::new(), |mut newest, event| {
-                    let metadata = match GroupMetadata::parse(&event) {
-                        Ok(metadata) => metadata,
-                        Err(error) => {
-                            log::warn!("ignoring malformed cached group metadata: {error}");
-                            return newest;
-                        }
-                    };
-
-                    let Some(id) = metadata.id().cloned() else {
-                        return newest;
-                    };
-
-                    match newest.get(&id) {
-                        Some((created_at, _)) if *created_at >= event.created_at => newest,
-                        _ => {
-                            newest.insert(id, (event.created_at, metadata));
-                            newest
-                        }
-                    }
-                });
-
-        let mut candidates = Vec::new();
-
-        for (relay, ids) in self.discovery.read(cx).refs().clone() {
-            for id in ids {
-                let Some((created_at, metadata)) = newest.get(&id) else {
-                    continue;
-                };
-
-                // Hidden groups are not discoverable (NIP-29).
-                if metadata.hidden {
-                    continue;
-                }
-
-                candidates.push(GroupCandidate {
-                    key: GroupKey::new(relay.clone(), id.clone()),
-                    created_at: *created_at,
-                    metadata: metadata.clone(),
-                });
-            }
-        }
-
-        self.discovery.update(cx, |discovery, cx| {
-            for candidate in candidates {
-                discovery.upsert(candidate, cx);
-            }
-        });
-    }
-
-    /// Load the cached member lists for the referenced groups from the database.
-    fn load_cached_members(&self, cx: &mut Context<Self>) -> Task<Result<Vec<Event>, Error>> {
-        let refs = self.discovery.read(cx).refs().clone();
-
-        if refs.is_empty() {
-            return Task::ready(Ok(Vec::new()));
-        }
-
-        let nostr = NostrRegistry::global(cx);
-        let client = nostr.read(cx).client();
-
-        let ids: HashSet<&str> = refs.values().flatten().map(GroupId::as_str).collect();
-        let filter = Filter::new().kind(Kind::GroupMembers).identifiers(ids);
-
-        cx.background_spawn(async move {
-            Ok(client.database().query(filter).await?.into_iter().collect())
-        })
-    }
-
-    /// Seed the member counts from the cached member lists.
-    fn absorb_cached_members(&mut self, members: Vec<Event>, cx: &mut Context<Self>) {
-        let newest: HashMap<GroupId, (Timestamp, usize)> =
-            members
-                .into_iter()
-                .fold(HashMap::new(), |mut newest, event| {
-                    let Some((id, created_at, count)) = parse_members(&event) else {
-                        return newest;
-                    };
-
-                    match newest.get(&id) {
-                        Some((current, _)) if *current >= created_at => newest,
-                        _ => {
-                            newest.insert(id, (created_at, count));
-                            newest
-                        }
-                    }
-                });
-
-        for (relay, ids) in self.discovery.read(cx).refs().clone() {
-            for id in ids {
-                let Some((created_at, count)) = newest.get(&id) else {
-                    continue;
-                };
-
-                let created_at = *created_at;
-                let count = *count;
-
-                self.discovery.update(cx, |discovery, cx| {
-                    discovery.set_members(
-                        GroupKey::new(relay.clone(), id.clone()),
-                        created_at,
-                        count,
-                        cx,
-                    );
-                });
-            }
-        }
-    }
-
-    /// Reconcile the contacts' group lists with the pool relays, returning
-    /// the newest lists now stored in the database.
-    fn sync_lists(
-        &self,
-        authors: HashSet<PublicKey>,
-        cx: &mut Context<Self>,
-    ) -> Task<Result<Vec<Event>, Error>> {
-        if authors.is_empty() {
-            return Task::ready(Ok(Vec::new()));
-        }
-
-        let nostr = NostrRegistry::global(cx);
-        let client = nostr.read(cx).client();
-
-        let filter = Filter::new().authors(authors).kind(Kind::SimpleGroups);
-
-        cx.background_spawn(async move {
-            let urls: Vec<RelayUrl> = client.relays().await.into_keys().collect();
-            reconcile(&client, filter.clone(), urls).await;
-
-            Ok(client.database().query(filter).await?.into_iter().collect())
-        })
-    }
-
-    /// Reconcile the groups of every pool relay, returning the newly
-    /// downloaded events with the relay that served them.
-    fn sync_sweep(&self, cx: &mut Context<Self>) -> Task<Result<Vec<(RelayUrl, Event)>, Error>> {
-        let nostr = NostrRegistry::global(cx);
-        let client = nostr.read(cx).client();
-
-        let filter = Filter::new().kinds([Kind::GroupMetadata, Kind::GroupMembers]);
-
-        cx.background_spawn(async move {
-            let urls: Vec<RelayUrl> = client.relays().await.into_keys().collect();
-            let received = reconcile(&client, filter, urls).await;
-
-            pull_received(&client, received).await
-        })
-    }
-
-    /// Absorb a discovered group metadata event from the relay serving it.
-    fn absorb_discovered(&mut self, relay: RelayUrl, event: &Event, cx: &mut Context<Self>) {
-        if let Some((id, created_at, members)) = parse_members(event) {
-            self.discovery.update(cx, |discovery, cx| {
-                discovery.set_members(GroupKey::new(relay, id), created_at, members, cx);
-            });
-            return;
-        }
-
-        let metadata = match GroupMetadata::parse(event) {
-            Ok(metadata) => metadata,
-            Err(error) => {
-                log::warn!("ignoring malformed group metadata from {relay}: {error}");
-                return;
-            }
-        };
-
-        // Hidden groups are not discoverable (NIP-29).
-        if metadata.hidden {
-            return;
-        }
-
-        let Some(id) = metadata.id().cloned() else {
-            return;
-        };
-
-        self.discovery.update(cx, |discovery, cx| {
-            discovery.upsert(
-                GroupCandidate {
-                    key: GroupKey::new(relay, id),
-                    created_at: event.created_at,
-                    metadata,
-                },
-                cx,
-            );
-        });
-    }
-
-    /// Reconcile the referenced groups with the relays hosting them,
-    /// returning the newly downloaded events with the relay that served
-    /// them.
-    fn sync_previews(&self, cx: &mut Context<Self>) -> Task<Result<Vec<(RelayUrl, Event)>, Error>> {
-        let refs = self.discovery.read(cx).refs().clone();
-
-        if refs.is_empty() {
-            return Task::ready(Ok(Vec::new()));
-        }
-
-        let nostr = NostrRegistry::global(cx);
-        let client = nostr.read(cx).client();
-
-        cx.background_spawn(async move {
-            let fetches = refs.into_iter().map(|(relay, ids)| {
-                let client = client.clone();
-
-                async move {
-                    if let Err(error) = client.add_relay(relay.clone()).and_connect().await {
-                        log::warn!("connecting to {relay} failed: {error}");
-                    }
-
-                    let filter = Filter::new()
-                        .kinds([Kind::GroupMetadata, Kind::GroupMembers])
-                        .identifiers(ids.iter().map(GroupId::as_str));
-
-                    let urls = vec![relay.clone()];
-                    let received = reconcile(&client, filter, urls).await;
-
-                    (relay, received)
-                }
-            });
-
-            let fetched = futures::future::join_all(fetches).await;
-            let ids: HashSet<EventId> = fetched
-                .iter()
-                .flat_map(|(_, received)| received.keys().cloned())
-                .collect();
-
-            if ids.is_empty() {
-                return Ok(Vec::new());
-            }
-
-            let events: HashMap<EventId, Event> = client
-                .database()
-                .query(Filter::new().ids(ids))
-                .await?
-                .into_iter()
-                .map(|event| (event.id, event))
-                .collect();
-
-            Ok(fetched
-                .into_iter()
-                .flat_map(|(relay, received)| {
-                    received
-                        .into_keys()
-                        .filter_map(|id| {
-                            let event = events.get(&id)?;
-                            Some((relay.clone(), event.clone()))
-                        })
-                        .collect::<Vec<_>>()
-                })
-                .collect())
-        })
-    }
-
-    fn absorb_group_list(&mut self, event: &Event, cx: &mut Context<Self>) {
-        let Ok(list) = GroupList::parse(event) else {
-            log::warn!("ignoring a malformed group list from {}", event.pubkey);
-            return;
-        };
-
-        let joined: HashSet<(RelayUrl, GroupId)> = self
-            .groups
             .iter()
             .map(|group| {
                 let key = group.read(cx).key();
                 (key.relay().clone(), key.id().clone())
             })
-            .collect();
+            .collect()
-
-        self.discovery.update(cx, |discovery, cx| {
-            discovery.absorb_list(&list, &joined, cx);
-        });
     }
 
-    /// Absorb the newest group list of every author.
+    /// Load the discoverable groups from the local database only.
-    fn absorb_lists(&mut self, lists: Vec<Event>, cx: &mut Context<Self>) {
+    pub fn cached_discovery(&self, cx: &App) -> Task<Discovery> {
-        let newest: HashMap<PublicKey, Event> =
+        let client = NostrRegistry::global(cx).read(cx).client();
-            lists.into_iter().fold(HashMap::new(), |mut newest, event| {
+        let joined = self.joined_keys(cx);
-                match newest.get(&event.pubkey) {
-                    Some(current) if current.created_at >= event.created_at => newest,
-                    _ => {
-                        newest.insert(event.pubkey, event);
-                        newest
-                    }
-                }
-            });
 
-        for event in newest.into_values() {
+        cx.background_spawn(Discovery::cached(client, joined))
-            self.absorb_group_list(&event, cx);
-        }
     }
 
-    fn absorb_preview(&mut self, relay: RelayUrl, event: &Event, cx: &mut Context<Self>) {
+    /// Reconcile the discoverable groups with the relays.
-        if let Some((group, created_at, members)) = parse_members(event) {
+    pub fn discover(&self, cx: &App) -> Task<()> {
-            self.discovery.update(cx, |discovery, cx| {
+        let nostr = NostrRegistry::global(cx);
-                discovery.set_members(GroupKey::new(relay, group), created_at, members, cx);
+        let client = nostr.read(cx).client();
-            });
+        let signer = nostr.read(cx).signer();
-            return;
+        let joined = self.joined_keys(cx);
+
+        if nostr.read(cx).current_user().is_none() {
+            return Task::ready(());
         }
 
-        let metadata = match GroupMetadata::parse(event) {
+        cx.background_spawn(async move {
-            Ok(metadata) => metadata,
+            let Ok(me) = signer.get_public_key_async().await else {
-            Err(error) => {
-                log::warn!("ignoring malformed group metadata from {relay}: {error}");
-                return;
-            }
-        };
-
-        // Hidden groups are not discoverable (NIP-29).
-        if metadata.hidden {
-            return;
-        }
-
-        // A relay may answer with more than the reconciliation asked for.
-        let Some(group) = metadata.id().cloned() else {
                 return;
             };
-
+            Discovery::default().sync(client, me, joined).await;
-        if !self
+        })
-            .discovery
-            .read(cx)
-            .refs()
-            .get(&relay)
-            .is_some_and(|ids| ids.contains(&group))
-        {
-            return;
-        }
-
-        self.discovery.update(cx, |discovery, cx| {
-            discovery.upsert(
-                GroupCandidate {
-                    key: GroupKey::new(relay, group),
-                    created_at: event.created_at,
-                    metadata,
-                },
-                cx,
-            );
-        });
     }
 
     pub fn load_more(&mut self, key: &GroupKey, cx: &mut Context<Self>) {
@@ -933,6 +343,7 @@ impl GroupsRegistry {
         self.observers.clear();
         self.forks.clear();
         self.fork_authors.clear();
+        self.syncing.clear();
         self.list = GroupList::default();
 
         let nostr = NostrRegistry::global(cx);
@@ -944,8 +355,6 @@ impl GroupsRegistry {
         ids.extend(self.browse.keys().cloned());
 
         self.browse.clear();
-        self.discovery
-            .update(cx, |discovery, cx| discovery.reset(cx));
 
         self.tasks.push(cx.background_spawn(async move {
             for id in ids {
@@ -1446,8 +855,10 @@ impl GroupsRegistry {
         };
 
         if let Some((group, created_at, members)) = parse_members(event) {
-            self.discovery.update(cx, |discovery, cx| {
+            cx.emit(GroupsEvent::Members {
-                discovery.set_members(GroupKey::new(relay, group), created_at, members, cx);
+                key: GroupKey::new(relay, group),
+                created_at,
+                members,
             });
             return;
         }
@@ -1554,45 +965,47 @@ impl GroupsRegistry {
             const MAX_PROCESSED: usize = 10_000;
 
             while let Some(notification) = notifications.next().await {
-                let ClientNotification::Message { message, .. } = notification else {
+                let ClientNotification::Message { relay_url, message } = notification else {
                     continue;
                 };
 
-                let signal = match *message {
+                match *message {
                     RelayMessage::Event {
                         subscription_id,
                         event,
-                    } => {
+                    } if is_own_subscription(&subscription_id) => {
-                        let id = subscription_id.into_owned();
-
-                        if !is_own_subscription(&id) {
-                            None
-                        } else {
                         if processed.len() >= MAX_PROCESSED {
                             processed.clear();
                         }
-                            processed
+
-                                .insert(event.id)
+                        if processed.insert(event.id) {
-                                .then(|| Signal::Event(id, Box::new(event.into_owned())))
+                            let signal = Signal::Event(
+                                subscription_id.into_owned(),
+                                Box::new(event.into_owned()),
+                            );
+
+                            tx.send_async(signal).await.ok();
                         }
                     }
-                    RelayMessage::EndOfStoredEvents(id) => {
+                    RelayMessage::EndOfStoredEvents(id) if is_own_subscription(&id) => {
-                        let id = id.into_owned();
+                        tx.send_async(Signal::Eose(id.into_owned())).await.ok();
-                        is_own_subscription(&id).then_some(Signal::Eose(id))
                     }
                     RelayMessage::Closed {
                         subscription_id,
                         message,
-                    } => {
+                    } if is_own_subscription(&subscription_id) => {
-                        let id = subscription_id.into_owned();
+                        let signal =
-                        is_own_subscription(&id).then_some(Signal::Closed(id, message.into_owned()))
+                            Signal::Closed(subscription_id.into_owned(), message.into_owned());
-                    }
-                    _ => None,
-                };
-
-                if let Some(signal) = signal {
                         tx.send_async(signal).await.ok();
                     }
+                    RelayMessage::NegMsg { .. } => {
+                        tx.send_async(Signal::Syncing(relay_url)).await.ok();
+                    }
+                    RelayMessage::NegErr { .. } | RelayMessage::EndOfStoredEvents(_) => {
+                        tx.send_async(Signal::Synced(relay_url)).await.ok();
+                    }
+                    _ => {}
+                }
             }
         }));
 
@@ -1611,6 +1024,14 @@ impl GroupsRegistry {
                     Signal::BrowseSettled(id) => {
                         this.settle_browse(id, cx);
                     }
+                    Signal::Syncing(relay) => {
+                        this.syncing.insert(relay);
+                    }
+                    Signal::Synced(relay) => {
+                        if this.syncing.remove(&relay) {
+                            cx.emit(GroupsEvent::Synced);
+                        }
+                    }
                     Signal::Forks => {
                         this.sync_forks(true, cx);
                     }

crates/nip29/src/utils.rs

@@ -0,0 +1,60 @@
+use std::collections::HashSet;
+use std::collections::hash_map::DefaultHasher;
+use std::hash::{Hash, Hasher};
+
+use nostr_sdk::prelude::*;
+
+use crate::{GroupId, GroupKey};
+
+const LIST_SUBSCRIPTION: &str = "nip29-list";
+const FORK_SUBSCRIPTION: &str = "nip29-forks";
+const GROUP_PREFIX: &str = "grp-";
+const BROWSE_PREFIX: &str = "brw-";
+const PAGE_PREFIX: &str = "pg-";
+
+pub(crate) fn list_subscription_id() -> SubscriptionId {
+    SubscriptionId::new(LIST_SUBSCRIPTION)
+}
+
+pub(crate) fn fork_subscription_id() -> SubscriptionId {
+    SubscriptionId::new(FORK_SUBSCRIPTION)
+}
+
+pub(crate) fn subscription_id(key: &GroupKey) -> SubscriptionId {
+    SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32))
+}
+
+pub(crate) fn browse_subscription(relay: &RelayUrl) -> SubscriptionId {
+    let mut hasher = DefaultHasher::new();
+    relay.hash(&mut hasher);
+
+    SubscriptionId::new(format!("{BROWSE_PREFIX}{:08x}", hasher.finish() as u32))
+}
+
+pub(crate) fn page_subscription(key: &GroupKey) -> SubscriptionId {
+    SubscriptionId::new(format!("{PAGE_PREFIX}{:08x}", key.uniq_id() as u32))
+}
+
+pub(crate) fn is_page(id: &SubscriptionId) -> bool {
+    id.as_str().starts_with(PAGE_PREFIX)
+}
+
+pub(crate) fn is_own_subscription(id: &SubscriptionId) -> bool {
+    let id = id.as_str();
+    id == LIST_SUBSCRIPTION
+        || id == FORK_SUBSCRIPTION
+        || id.starts_with(GROUP_PREFIX)
+        || id.starts_with(BROWSE_PREFIX)
+        || id.starts_with(PAGE_PREFIX)
+}
+
+pub(crate) fn parse_members(event: &Event) -> Option<(GroupId, Timestamp, usize)> {
+    if event.kind != Kind::GroupMembers {
+        return None;
+    }
+
+    let id = GroupId::new(event.tags.identifier()?).ok()?;
+    let members = event.tags.public_keys().collect::<HashSet<_>>().len();
+
+    Some((id, event.created_at, members))
+}

crates/workspace/src/panels/browse.rs

@@ -4,12 +4,12 @@ use std::rc::Rc;
 use gpui::prelude::FluentBuilder;
 use gpui::{
     AnyElement, App, AppContext, Context, Entity, EventEmitter, FocusHandle, Focusable,
-    IntoElement, ParentElement, Render, ScrollStrategy, SharedString, Styled, Subscription,
+    IntoElement, ParentElement, Render, ScrollStrategy, SharedString, Styled, Subscription, Task,
     WeakEntity, Window, div, px, relative, size,
 };
 use gpui_base::{VirtualListScrollHandle, v_virtual_list};
 use nip29::{Discovery, GroupCandidate, GroupKey, GroupsEvent, GroupsRegistry};
-use nostr_sdk::prelude::RelayUrl;
+use nostr_sdk::prelude::{RelayUrl, Timestamp};
 use settings::AppSettings;
 use smallvec::{SmallVec, smallvec};
 use theme::ActiveTheme;
@@ -40,19 +40,21 @@ pub struct BrowsePanel {
     /// The groups the relay has announced so far
     candidates: Vec<GroupCandidate>,
     /// The groups discovered from the cache and the relays
-    discovery: Entity<Discovery>,
+    discovery: Discovery,
+    /// Whether the discovery sync is still running
+    discovering: bool,
     /// The scroll position of the group list
     scroll_handle: VirtualListScrollHandle,
     /// Whether the relay is still answering
     loading: bool,
-    _subscriptions: SmallVec<[Subscription; 3]>,
+    tasks: SmallVec<[Task<()>; 2]>,
+    _subscriptions: SmallVec<[Subscription; 2]>,
 }
 
 impl BrowsePanel {
     fn new(dock: WeakEntity<DockArea>, window: &mut Window, cx: &mut Context<Self>) -> Self {
         let input = cx.new(|cx| InputState::new(window, cx).placeholder(INPUT_PLACEHOLDER));
         let registry = GroupsRegistry::global(cx);
-        let discovery = registry.read(cx).discovery();
 
         let mut subscriptions = smallvec![];
 
@@ -64,10 +66,6 @@ impl BrowsePanel {
             }),
         );
 
-        subscriptions.push(cx.observe(&discovery, |_, _, cx| {
-            cx.notify();
-        }));
-
         subscriptions.push(
             cx.subscribe(&registry, |this, _registry, event, cx| match event {
                 GroupsEvent::Browsed(candidate) => {
@@ -76,15 +74,23 @@ impl BrowsePanel {
                 GroupsEvent::BrowseSettled(relay) => {
                     this.settle(relay.clone(), cx);
                 }
+                GroupsEvent::Synced => {
+                    this.refresh_discovery(cx);
+                }
+                GroupsEvent::Members {
+                    key,
+                    created_at,
+                    members,
+                } => {
+                    this.push_members(key.clone(), *created_at, *members, cx);
+                }
                 _ => {}
             }),
         );
 
-        // Wait for the panel to leave the stack before kicking off the sweep.
+        cx.defer_in(window, |this, _window, cx| {
-        cx.defer_in(window, |_this, _window, cx| {
+            this.load_cached(cx);
-            GroupsRegistry::global(cx).update(cx, |registry, cx| {
+            this.sync_discovery(cx);
-                registry.discover(cx);
-            });
         });
 
         Self {
@@ -93,13 +99,77 @@ impl BrowsePanel {
             input,
             relay: None,
             candidates: Vec::new(),
-            discovery,
+            discovery: Discovery::default(),
+            discovering: true,
             scroll_handle: VirtualListScrollHandle::new(),
             loading: false,
+            tasks: smallvec![],
             _subscriptions: subscriptions,
         }
     }
 
+    /// Load the groups from the local database.
+    fn load_cached(&mut self, cx: &mut Context<Self>) {
+        let registry = GroupsRegistry::global(cx);
+        let discovery = registry.read(cx).cached_discovery(cx);
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let discovery = discovery.await;
+
+            this.update(cx, |this, cx| {
+                this.discovery = discovery;
+                cx.notify();
+            })
+            .ok();
+        }));
+    }
+
+    /// Sync the discoverable groups with the relays.
+    fn sync_discovery(&mut self, cx: &mut Context<Self>) {
+        self.discovering = true;
+
+        let registry = GroupsRegistry::global(cx);
+        let sync = registry.read(cx).discover(cx);
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            sync.await;
+
+            this.update(cx, |this, cx| {
+                this.discovering = false;
+                cx.notify();
+            })
+            .ok();
+        }));
+    }
+
+    /// A browsed group announced its member list.
+    fn push_members(
+        &mut self,
+        key: GroupKey,
+        created_at: Timestamp,
+        members: usize,
+        cx: &mut Context<Self>,
+    ) {
+        self.discovery.set_members(key, created_at, members);
+        cx.notify();
+    }
+
+    /// The discovery synced new groups: re-query the local database.
+    fn refresh_discovery(&mut self, cx: &mut Context<Self>) {
+        let registry = GroupsRegistry::global(cx);
+        let discovery = registry.read(cx).cached_discovery(cx);
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let discovery = discovery.await;
+
+            this.update(cx, |this, cx| {
+                this.discovery = discovery;
+                cx.notify();
+            })
+            .ok();
+        }));
+    }
+
     /// Browse the relay in the input.
     fn browse(&mut self, window: &mut Window, cx: &mut Context<Self>) {
         let registry = GroupsRegistry::global(cx);
@@ -189,7 +259,6 @@ impl BrowsePanel {
 
         let cards: Vec<GroupCandidate> = self
             .discovery
-            .read(cx)
             .discovered()
             .iter()
             .filter(|candidate| !joined.contains(&candidate.key))
@@ -242,7 +311,7 @@ impl BrowsePanel {
     fn render_card(&self, candidate: &GroupCandidate, cx: &mut Context<Self>) -> AnyElement {
         let key = candidate.key.clone();
         let seed = key.cache_tag();
-        let members = self.discovery.read(cx).members(&key).unwrap_or(0);
+        let members = self.discovery.members(&key).unwrap_or(0);
 
         let name = SharedString::from(candidate.display_name().to_owned());
         let relay = SharedString::from(key.relay().as_str().to_owned());
@@ -346,11 +415,11 @@ impl BrowsePanel {
                         h_flex()
                             .w_full()
                             .gap_1p5()
-                            .child(Icon::new(IconName::Group))
+                            .text_color(cx.theme().text_muted)
+                            .child(Icon::new(IconName::Book))
                             .child(
                                 div()
                                     .text_sm()
-                                    .text_color(cx.theme().text_muted)
                                     .child(SharedString::from(format!("{members} members"))),
                             ),
                     ),
@@ -392,8 +461,6 @@ impl Render for BrowsePanel {
 
         let (discovered_count, discovering) = if discover {
             let registry = GroupsRegistry::global(cx);
-            let discovery = self.discovery.read(cx);
-            let discovering = discovery.loading();
 
             let joined: Vec<GroupKey> = registry
                 .read(cx)
@@ -402,13 +469,14 @@ impl Render for BrowsePanel {
                 .map(|group| group.read(cx).key().clone())
                 .collect();
 
-            let discovered_count = discovery
+            let discovered_count = self
+                .discovery
                 .discovered()
                 .iter()
                 .filter(|candidate| !joined.contains(&candidate.key))
                 .count();
 
-            (discovered_count, discovering)
+            (discovered_count, self.discovering)
         } else {
             (0, false)
         };