diff --git a/crates/nip29/src/discovery.rs b/crates/nip29/src/discovery.rs index a1074338..18d834b1 100644 --- a/crates/nip29/src/discovery.rs +++ b/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)] +#[derive(Default, Clone, Debug)] pub struct Discovery { refs: HashMap>, discovered: Vec, members: HashMap, - loading: bool, } impl Discovery { + /// The discovered groups, sorted by name. pub fn discovered(&self) -> &[GroupCandidate] { &self.discovered } - pub fn loading(&self) -> bool { - self.loading - } - - /// The discovery is starting over: nothing has answered yet. - pub(crate) fn begin(&mut self, cx: &mut Context) { - self.loading = true; - cx.notify(); - } - - /// Every pending discovery step has finished. - pub(crate) fn settle(&mut self, cx: &mut Context) { - 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, - ) -> 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 member count of a group, when its member list is known. + pub fn members(&self, key: &GroupKey) -> Option { + self.members.get(key).map(|(_, members)| *members) } /// 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, + joined: &HashSet<(RelayUrl, GroupId)>, + ) { + let newest: HashMap = + 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) { + 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( - &mut self, - key: GroupKey, - created_at: Timestamp, - members: usize, - cx: &mut Context, - ) { + pub fn set_members(&mut self, key: GroupKey, created_at: Timestamp, members: usize) { if self .members .get(&key) @@ -108,21 +97,191 @@ impl Discovery { return; } - if self.members.insert(key, (created_at, members)) != Some((created_at, members)) { - cx.notify(); + self.members.insert(key, (created_at, members)); + } + + /// Seed the discoveries from the cached group metadata. + pub(crate) fn absorb_cached_previews(&mut self, previews: Vec) { + let newest: HashMap = + 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. - pub fn members(&self, key: &GroupKey) -> Option { - self.members.get(key).map(|(_, members)| *members) + /// Seed the member counts from the cached member lists. + pub(crate) fn absorb_cached_members(&mut self, members: Vec) { + let newest: HashMap = + 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.refs.clear(); - self.discovered.clear(); - self.members.clear(); - self.loading = false; - cx.notify(); + /// Load the discoverable groups from the local database only. + pub async fn cached(client: Client, joined: HashSet<(RelayUrl, GroupId)>) -> Self { + let mut discovery = Self::default(); + + // The group lists of every cached author. + let lists: Vec = 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 = 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 = 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 = 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 = 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 = 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 = 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; } } diff --git a/crates/nip29/src/lib.rs b/crates/nip29/src/lib.rs index 393706f0..ad019b6e 100644 --- a/crates/nip29/src/lib.rs +++ b/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 = 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), 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::>().len(); - - Some((id, event.created_at, members)) -} - -async fn reconcile( - client: &Client, - filter: Filter, - urls: Vec, -) -> HashMap> { - 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>, -) -> Result, Error> { - if received.is_empty() { - return Ok(Vec::new()); - } - - let events: HashMap = 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 for GroupsRegistry {} pub struct GroupsRegistry { @@ -160,8 +71,8 @@ pub struct GroupsRegistry { observers: HashMap, forks: HashMap, fork_authors: HashSet, - /// The groups discovered from the cache and the relays. - discovery: Entity, + /// The relays with a discovery reconciliation in flight. + syncing: HashSet, signal_tx: flume::Sender, signal_rx: flume::Receiver, notification_listener: Option>, @@ -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. - pub fn discovery(&self) -> Entity { - self.discovery.clone() - } - - pub fn discover(&mut self, cx: &mut Context) { - 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, Vec); - - let task: Task> = 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) -> Task, 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, cx: &mut Context) { - let newest: HashMap = - 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) -> Task, 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, cx: &mut Context) { - let newest: HashMap = - 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, - cx: &mut Context, - ) -> Task, 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 = 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) -> Task, 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 = 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) { - 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) -> Task, 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 = fetched - .iter() - .flat_map(|(_, received)| received.keys().cloned()) - .collect(); - - if ids.is_empty() { - return Ok(Vec::new()); - } - - let events: HashMap = 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::>() - }) - .collect()) - }) - } - - fn absorb_group_list(&mut self, event: &Event, cx: &mut Context) { - 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 + /// The (relay, group) pairs of every joined group. + fn joined_keys(&self, cx: &App) -> HashSet<(RelayUrl, GroupId)> { + self.groups .iter() .map(|group| { let key = group.read(cx).key(); (key.relay().clone(), key.id().clone()) }) - .collect(); - - self.discovery.update(cx, |discovery, cx| { - discovery.absorb_list(&list, &joined, cx); - }); + .collect() } - /// Absorb the newest group list of every author. - fn absorb_lists(&mut self, lists: Vec, cx: &mut Context) { - let newest: HashMap = - 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 - } - } - }); + /// Load the discoverable groups from the local database only. + pub fn cached_discovery(&self, cx: &App) -> Task { + let client = NostrRegistry::global(cx).read(cx).client(); + let joined = self.joined_keys(cx); - for event in newest.into_values() { - self.absorb_group_list(&event, cx); - } + cx.background_spawn(Discovery::cached(client, joined)) } - fn absorb_preview(&mut self, relay: RelayUrl, event: &Event, cx: &mut Context) { - if let Some((group, created_at, members)) = parse_members(event) { - self.discovery.update(cx, |discovery, cx| { - discovery.set_members(GroupKey::new(relay, group), created_at, members, cx); - }); - return; + /// Reconcile the discoverable groups with the relays. + pub fn discover(&self, cx: &App) -> Task<()> { + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + let signer = nostr.read(cx).signer(); + let joined = self.joined_keys(cx); + + if nostr.read(cx).current_user().is_none() { + return Task::ready(()); } - let metadata = match GroupMetadata::parse(event) { - Ok(metadata) => metadata, - Err(error) => { - log::warn!("ignoring malformed group metadata from {relay}: {error}"); + cx.background_spawn(async move { + let Ok(me) = signer.get_public_key_async().await else { 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; - }; - - 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, - ); - }); + }; + Discovery::default().sync(client, me, joined).await; + }) } pub fn load_more(&mut self, key: &GroupKey, cx: &mut Context) { @@ -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| { - discovery.set_members(GroupKey::new(relay, group), created_at, members, cx); + cx.emit(GroupsEvent::Members { + key: GroupKey::new(relay, group), + created_at, + members, }); return; } @@ -1554,44 +965,46 @@ 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, - } => { - let id = subscription_id.into_owned(); + } if is_own_subscription(&subscription_id) => { + if processed.len() >= MAX_PROCESSED { + processed.clear(); + } - if !is_own_subscription(&id) { - None - } else { - if processed.len() >= MAX_PROCESSED { - processed.clear(); - } - processed - .insert(event.id) - .then(|| Signal::Event(id, Box::new(event.into_owned()))) + if processed.insert(event.id) { + let signal = Signal::Event( + subscription_id.into_owned(), + Box::new(event.into_owned()), + ); + + tx.send_async(signal).await.ok(); } } - RelayMessage::EndOfStoredEvents(id) => { - let id = id.into_owned(); - is_own_subscription(&id).then_some(Signal::Eose(id)) + RelayMessage::EndOfStoredEvents(id) if is_own_subscription(&id) => { + tx.send_async(Signal::Eose(id.into_owned())).await.ok(); } RelayMessage::Closed { subscription_id, message, - } => { - let id = subscription_id.into_owned(); - is_own_subscription(&id).then_some(Signal::Closed(id, message.into_owned())) + } if is_own_subscription(&subscription_id) => { + let signal = + Signal::Closed(subscription_id.into_owned(), message.into_owned()); + tx.send_async(signal).await.ok(); } - _ => 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); } diff --git a/crates/nip29/src/utils.rs b/crates/nip29/src/utils.rs new file mode 100644 index 00000000..f66d4bc5 --- /dev/null +++ b/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::>().len(); + + Some((id, event.created_at, members)) +} diff --git a/crates/workspace/src/panels/browse.rs b/crates/workspace/src/panels/browse.rs index 9fd0b4f3..e43030ed 100644 --- a/crates/workspace/src/panels/browse.rs +++ b/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, /// The groups discovered from the cache and the relays - discovery: Entity, + 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, window: &mut Window, cx: &mut Context) -> 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(®istry, |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| { - GroupsRegistry::global(cx).update(cx, |registry, cx| { - registry.discover(cx); - }); + cx.defer_in(window, |this, _window, cx| { + this.load_cached(cx); + this.sync_discovery(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) { + 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.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.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) { + 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) { let registry = GroupsRegistry::global(cx); @@ -189,7 +259,6 @@ impl BrowsePanel { let cards: Vec = 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) -> 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 = 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) };