From e4720779413144ff957f0057abc13c4b53eff6f3 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Wed, 7 Oct 2026 11:54:33 +0700 Subject: [PATCH] simplify nip29 discovery and sync metadata via negentropy --- Cargo.lock | 2 + crates/nip29/Cargo.toml | 2 + crates/nip29/src/discovery.rs | 107 ++---- crates/nip29/src/lib.rs | 528 +++++++++++++------------- crates/nip29/src/protocol.rs | 21 +- crates/workspace/src/panels/browse.rs | 126 +++--- 6 files changed, 377 insertions(+), 409 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8dcc381a..7a399e2a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4517,9 +4517,11 @@ dependencies = [ "anyhow", "chat", "flume 0.11.1", + "futures", "gpui-pre", "log", "nostr-sdk", + "serde_json", "smallvec", "state", ] diff --git a/crates/nip29/Cargo.toml b/crates/nip29/Cargo.toml index ba605cc0..48332da4 100644 --- a/crates/nip29/Cargo.toml +++ b/crates/nip29/Cargo.toml @@ -7,6 +7,8 @@ publish.workspace = true [dependencies] nostr-sdk.workspace = true anyhow.workspace = true +futures.workspace = true +serde_json.workspace = true state = { path = "../state" } chat = { path = "../chat" } diff --git a/crates/nip29/src/discovery.rs b/crates/nip29/src/discovery.rs index f446ffe5..a1074338 100644 --- a/crates/nip29/src/discovery.rs +++ b/crates/nip29/src/discovery.rs @@ -8,18 +8,10 @@ use crate::protocol::{GroupCandidate, GroupId, GroupKey, GroupList}; /// The groups discovered from the cache and the relays. #[derive(Default)] pub struct Discovery { - /// The (relay, group) references of every cached group list. refs: HashMap>, - /// The groups whose metadata has resolved so far. discovered: Vec, - /// The member counts of the known groups. - members: HashMap, - /// The in-flight preview subscriptions, keyed by subscription id. - previews: HashMap, - /// Whether the contacts' group lists have been answered. - lists_settled: bool, - /// Whether the pool-wide metadata discovery has been answered. - sweep_settled: bool, + members: HashMap, + loading: bool, } impl Discovery { @@ -28,37 +20,18 @@ impl Discovery { } pub fn loading(&self) -> bool { - !self.lists_settled || !self.sweep_settled || !self.previews.is_empty() - } - - /// Whether the contacts' group lists have been answered. - pub(crate) fn lists_settled(&self) -> bool { - self.lists_settled + self.loading } /// The discovery is starting over: nothing has answered yet. pub(crate) fn begin(&mut self, cx: &mut Context) { - self.lists_settled = false; - self.sweep_settled = false; + self.loading = true; cx.notify(); } - /// There is nothing on the way: there is nothing to discover. - pub(crate) fn settle_all(&mut self, cx: &mut Context) { - self.lists_settled = true; - self.sweep_settled = true; - cx.notify(); - } - - /// The contacts' group lists have been answered. - pub(crate) fn settle_lists(&mut self, cx: &mut Context) { - self.lists_settled = true; - cx.notify(); - } - - /// The pool-wide metadata discovery has been answered. - pub(crate) fn settle_sweep(&mut self, cx: &mut Context) { - self.sweep_settled = true; + /// Every pending discovery step has finished. + pub(crate) fn settle(&mut self, cx: &mut Context) { + self.loading = false; cx.notify(); } @@ -100,43 +73,13 @@ impl Discovery { &self.refs } - /// Whether a preview subscription is already asking this relay. - pub(crate) fn preview_in_flight(&self, relay: &RelayUrl) -> bool { - self.previews.values().any(|in_flight| in_flight == relay) - } - - /// The relay a preview subscription is asking, if it is in flight. - pub(crate) fn preview_relay(&self, id: &SubscriptionId) -> Option { - self.previews.get(id).cloned() - } - - /// The ids of the in-flight preview subscriptions. - pub(crate) fn preview_ids(&self) -> Vec { - self.previews.keys().cloned().collect() - } - - /// A preview subscription starts asking a relay for metadata. - pub(crate) fn track_preview(&mut self, id: SubscriptionId, relay: RelayUrl) { - self.previews.insert(id, relay); - } - - /// A preview subscription has been answered. - /// - /// Returns whether it was in flight. - pub(crate) fn settle_preview(&mut self, id: &SubscriptionId, cx: &mut Context) -> bool { - let removed = self.previews.remove(id).is_some(); - - if removed { - cx.notify(); - } - - removed - } - - /// Add or replace a discovered group. + /// Add a discovered group, keeping the newest metadata. pub(crate) fn upsert(&mut self, candidate: GroupCandidate, cx: &mut Context) { match self.discovered.iter_mut().find(|d| d.key == candidate.key) { - Some(existing) => *existing = candidate, + Some(existing) if existing.created_at <= candidate.created_at => { + *existing = candidate; + } + Some(_) => return, None => self.discovered.push(candidate), } @@ -149,25 +92,37 @@ impl Discovery { cx.notify(); } - /// Apply a member count for a group. - pub(crate) fn set_members(&mut self, key: GroupKey, members: usize, cx: &mut Context) { - if self.members.insert(key, members) != Some(members) { + /// 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, + ) { + if self + .members + .get(&key) + .is_some_and(|(current, _)| *current > created_at) + { + return; + } + + if self.members.insert(key, (created_at, members)) != Some((created_at, members)) { cx.notify(); } } /// The member count of a group, when its member list is known. pub fn members(&self, key: &GroupKey) -> Option { - self.members.get(key).copied() + self.members.get(key).map(|(_, members)| *members) } pub(crate) fn reset(&mut self, cx: &mut Context) { self.refs.clear(); self.discovered.clear(); self.members.clear(); - self.previews.clear(); - self.lists_settled = false; - self.sweep_settled = false; + self.loading = false; cx.notify(); } } diff --git a/crates/nip29/src/lib.rs b/crates/nip29/src/lib.rs index 905f80b4..80dbae97 100644 --- a/crates/nip29/src/lib.rs +++ b/crates/nip29/src/lib.rs @@ -22,11 +22,8 @@ pub(crate) static LOCAL_KEYS: LazyLock = LazyLock::new(Keys::generate); const LIST_SUBSCRIPTION: &str = "nip29-list"; const FORK_SUBSCRIPTION: &str = "nip29-forks"; -const CONTACTS_SUBSCRIPTION: &str = "nip29-contacts"; -const DISCOVERY_SUBSCRIPTION: &str = "nip29-discovery"; const GROUP_PREFIX: &str = "grp-"; const BROWSE_PREFIX: &str = "brw-"; -const PREVIEW_PREFIX: &str = "prv-"; const PAGE_PREFIX: &str = "pg-"; const BROWSE_LIMIT: usize = 500; const BROWSE_TIMEOUT: Duration = Duration::from_secs(10); @@ -50,13 +47,10 @@ pub enum GroupsEvent { #[derive(Debug)] enum Signal { - Event(SubscriptionId, RelayUrl, Box), + Event(SubscriptionId, Box), Eose(SubscriptionId), Closed(SubscriptionId, String), BrowseSettled(SubscriptionId), - ContactListsSettled, - ContactPreviewSettled(SubscriptionId), - DiscoverySettled, Forks, } @@ -68,14 +62,6 @@ fn fork_subscription_id() -> SubscriptionId { SubscriptionId::new(FORK_SUBSCRIPTION) } -fn contacts_subscription_id() -> SubscriptionId { - SubscriptionId::new(CONTACTS_SUBSCRIPTION) -} - -fn discovery_subscription_id() -> SubscriptionId { - SubscriptionId::new(DISCOVERY_SUBSCRIPTION) -} - fn subscription_id(key: &GroupKey) -> SubscriptionId { SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32)) } @@ -87,13 +73,6 @@ fn browse_subscription(relay: &RelayUrl) -> SubscriptionId { SubscriptionId::new(format!("{BROWSE_PREFIX}{:08x}", hasher.finish() as u32)) } -fn preview_subscription(relay: &RelayUrl) -> SubscriptionId { - let mut hasher = DefaultHasher::new(); - relay.hash(&mut hasher); - - SubscriptionId::new(format!("{PREVIEW_PREFIX}{:08x}", hasher.finish() as u32)) -} - fn page_subscription(key: &GroupKey) -> SubscriptionId { SubscriptionId::new(format!("{PAGE_PREFIX}{:08x}", key.uniq_id() as u32)) } @@ -106,16 +85,13 @@ fn is_own_subscription(id: &SubscriptionId) -> bool { let id = id.as_str(); id == LIST_SUBSCRIPTION || id == FORK_SUBSCRIPTION - || id == CONTACTS_SUBSCRIPTION - || id == DISCOVERY_SUBSCRIPTION || id.starts_with(GROUP_PREFIX) || id.starts_with(BROWSE_PREFIX) - || id.starts_with(PREVIEW_PREFIX) || id.starts_with(PAGE_PREFIX) } -/// The group id and member count of a group members event. -fn parse_members(event: &Event) -> Option<(GroupId, usize)> { +/// The group id, member count and timestamp of a group members event. +fn parse_members(event: &Event) -> Option<(GroupId, Timestamp, usize)> { if event.kind != Kind::GroupMembers { return None; } @@ -123,7 +99,61 @@ fn parse_members(event: &Event) -> Option<(GroupId, usize)> { let id = GroupId::new(event.tags.identifier()?).ok()?; let members = event.tags.public_keys().collect::>().len(); - Some((id, members)) + Some((id, event.created_at, members)) +} + +/// Reconcile a filter with the relays via negentropy, returning the events +/// that were downloaded to the local database. +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!("nip29: the reconciliation failed: {error}"); + return HashMap::new(); + } + }; + + for (relay, error) in &output.failed { + log::debug!("nip29: syncing with {relay} failed: {error}"); + } + + output.value.received +} + +/// Pair the downloaded events with the relay that served them. +/// +/// The database does not keep relay associations, so the pairing has to +/// happen while the reconciliation still knows 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 {} @@ -336,7 +366,7 @@ impl GroupsRegistry { if nostr.read(cx).current_user().is_none() { // There is nothing to discover without a user. self.discovery.update(cx, |discovery, cx| { - discovery.settle_all(cx); + discovery.settle(cx); }); return; } @@ -381,20 +411,7 @@ impl GroupsRegistry { }; if let Err(error) = this.update(cx, |this, cx| { - 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() { - this.absorb_group_list(&event, cx); - } + this.absorb_lists(lists, cx); }) { log::warn!("nip29: applying the cached group lists failed: {error}"); return; @@ -433,11 +450,82 @@ impl GroupsRegistry { if let Err(error) = this.update(cx, |this, cx| { this.absorb_cached_previews(previews, cx); this.absorb_cached_members(members, cx); + }) { + log::warn!("nip29: discovering the groups failed: {error}"); + return; + } - // The contacts' lists resolve groups on relays outside the - // pool, the sweep covers every relay in it. - this.subscribe_contacts(authors, cx); - this.subscribe_discovery(cx); + // Reconcile the contacts' group lists with the pool relays: only + // the events missing locally are downloaded and stored. + let lists = match this.update(cx, |this, cx| this.sync_lists(authors, cx)) { + Ok(task) => task, + Err(error) => { + log::warn!("nip29: 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!("nip29: discovering the groups failed: {error}"); + return; + } + }; + + let lists = match lists.await { + Ok(lists) => lists, + Err(error) => { + log::warn!("nip29: 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!("nip29: applying the contacts group lists failed: {error}"); + } + + let sweep = match sweep.await { + Ok(events) => events, + Err(error) => { + log::warn!("nip29: 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!("nip29: 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!("nip29: syncing the group previews failed: {error}"); + Vec::new() + } + }, + Err(error) => { + log::warn!("nip29: 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!("nip29: discovering the groups failed: {error}"); } @@ -494,7 +582,7 @@ impl GroupsRegistry { for (relay, ids) in self.discovery.read(cx).refs().clone() { for id in ids { - let Some((_, metadata)) = newest.get(&id) else { + let Some((created_at, metadata)) = newest.get(&id) else { continue; }; @@ -505,6 +593,7 @@ impl GroupsRegistry { candidates.push(GroupCandidate { key: GroupKey::new(relay.clone(), id.clone()), + created_at: *created_at, metadata: metadata.clone(), }); } @@ -542,14 +631,14 @@ impl GroupsRegistry { members .into_iter() .fold(HashMap::new(), |mut newest, event| { - let Some((id, count)) = parse_members(&event) else { + let Some((id, created_at, count)) = parse_members(&event) else { return newest; }; match newest.get(&id) { - Some((created_at, _)) if *created_at >= event.created_at => newest, + Some((current, _)) if *current >= created_at => newest, _ => { - newest.insert(id, (event.created_at, count)); + newest.insert(id, (created_at, count)); newest } } @@ -557,26 +646,34 @@ impl GroupsRegistry { for (relay, ids) in self.discovery.read(cx).refs().clone() { for id in ids { - let Some((_, count)) = newest.get(&id) else { + 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()), count, cx); + discovery.set_members( + GroupKey::new(relay.clone(), id.clone()), + created_at, + count, + cx, + ); }); } } } - /// Subscribe to the contacts' public group lists. - fn subscribe_contacts(&mut self, authors: HashSet, cx: &mut Context) { + /// 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() { - // Nobody is followed, so there is nothing to discover from them. - self.discovery - .update(cx, |discovery, cx| discovery.settle_lists(cx)); - return; + return Task::ready(Ok(Vec::new())); } let nostr = NostrRegistry::global(cx); @@ -584,77 +681,35 @@ impl GroupsRegistry { let filter = Filter::new().authors(authors).kind(Kind::SimpleGroups); - let opts = SubscribeAutoCloseOptions::default() - .exit_policy(ReqExitPolicy::ExitOnEOSE) - .timeout(Some(BROWSE_TIMEOUT)); + cx.background_spawn(async move { + let urls: Vec = client.relays().await.into_keys().collect(); + reconcile(&client, filter.clone(), urls).await; - let tx = self.signal_tx.clone(); - - self.tasks.push(cx.background_spawn(async move { - let refused = match client - .subscribe(ReqTarget::auto([filter])) - .with_id(contacts_subscription_id()) - .close_on(opts) - .await - { - Ok(output) => output.success.is_empty(), - Err(error) => { - log::warn!("subscribing to the contacts group lists failed: {error}"); - true - } - }; - - // When every relay refused the request no EOSE will arrive. - if refused && let Err(error) = tx.send_async(Signal::ContactListsSettled).await { - log::warn!("the contacts settle signal was dropped: {error}"); - } - })); - - cx.notify(); + Ok(client.database().query(filter).await?.into_iter().collect()) + }) } - /// Subscribe to the metadata of every group the pool relays host. - fn subscribe_discovery(&mut self, cx: &mut Context) { + /// 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 metadata = Filter::new().kind(Kind::GroupMetadata).limit(BROWSE_LIMIT); - let members = Filter::new().kind(Kind::GroupMembers).limit(BROWSE_LIMIT); + let filter = Filter::new().kinds([Kind::GroupMetadata, Kind::GroupMembers]); - let opts = SubscribeAutoCloseOptions::default() - .exit_policy(ReqExitPolicy::ExitOnEOSE) - .timeout(Some(BROWSE_TIMEOUT)); + cx.background_spawn(async move { + let urls: Vec = client.relays().await.into_keys().collect(); + let received = reconcile(&client, filter, urls).await; - let tx = self.signal_tx.clone(); - - self.tasks.push(cx.background_spawn(async move { - let refused = match client - .subscribe(ReqTarget::auto([metadata, members])) - .with_id(discovery_subscription_id()) - .close_on(opts) - .await - { - Ok(output) => output.success.is_empty(), - Err(error) => { - log::warn!("nip29: the group discovery failed: {error}"); - true - } - }; - - // When every relay refused the request no EOSE will arrive. - if refused && let Err(error) = tx.send_async(Signal::DiscoverySettled).await { - log::warn!("the discovery settle signal was dropped: {error}"); - } - })); - - cx.notify(); + 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, members)) = parse_members(event) { + if let Some((id, created_at, members)) = parse_members(event) { self.discovery.update(cx, |discovery, cx| { - discovery.set_members(GroupKey::new(relay, id), members, cx); + discovery.set_members(GroupKey::new(relay, id), created_at, members, cx); }); return; } @@ -680,6 +735,7 @@ impl GroupsRegistry { discovery.upsert( GroupCandidate { key: GroupKey::new(relay, id), + created_at: event.created_at, metadata, }, cx, @@ -687,61 +743,70 @@ impl GroupsRegistry { }); } - /// Ask every relay hosting referenced groups for the groups' metadata. - fn sync_previews(&mut self, cx: &mut Context) { + /// 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(); - for (relay, ids) in self.discovery.read(cx).refs().clone() { - // A subscription is already asking this relay for previews. - if self.discovery.read(cx).preview_in_flight(&relay) { - continue; + 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!("nip29: 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 id = preview_subscription(&relay); - self.discovery.update(cx, |discovery, _| { - discovery.track_preview(id.clone(), relay.clone()); - }); - let client = client.clone(); + let events: HashMap = client + .database() + .query(Filter::new().ids(ids)) + .await? + .into_iter() + .map(|event| (event.id, event)) + .collect(); - let metadata = Filter::new() - .kind(Kind::GroupMetadata) - .identifiers(ids.iter().map(GroupId::as_str)); - - let members = Filter::new() - .kind(Kind::GroupMembers) - .identifiers(ids.iter().map(GroupId::as_str)); - - let opts = SubscribeAutoCloseOptions::default() - .exit_policy(ReqExitPolicy::ExitOnEOSE) - .timeout(Some(BROWSE_TIMEOUT)); - - let tx = self.signal_tx.clone(); - - self.tasks.push(cx.background_spawn(async move { - client.add_relay(relay.clone()).and_connect().await.ok(); - - let refused = match client - .subscribe(ReqTarget::single(relay.clone(), [metadata, members])) - .with_id(id.clone()) - .close_on(opts) - .await - { - Ok(output) => output.success.is_empty(), - Err(error) => { - log::warn!("fetching group previews from {relay} failed: {error}"); - true - } - }; - - // When the relay refused the request no EOSE will arrive. - if refused - && let Err(error) = tx.send_async(Signal::ContactPreviewSettled(id)).await - { - log::warn!("the preview settle signal was dropped: {error}"); - } - })); - } + 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) { @@ -762,23 +827,33 @@ impl GroupsRegistry { }) .collect(); - let added = self.discovery.update(cx, |discovery, cx| { - discovery.absorb_list(&list, &joined, cx) + self.discovery.update(cx, |discovery, cx| { + discovery.absorb_list(&list, &joined, cx); }); + } - if added && self.discovery.read(cx).lists_settled() { - self.sync_previews(cx); + /// 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 + } + } + }); + + for event in newest.into_values() { + self.absorb_group_list(&event, cx); } } - fn absorb_preview(&mut self, id: &SubscriptionId, event: &Event, cx: &mut Context) { - let Some(relay) = self.discovery.read(cx).preview_relay(id) else { - return; - }; - - if let Some((group, members)) = parse_members(event) { + 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), members, cx); + discovery.set_members(GroupKey::new(relay, group), created_at, members, cx); }); return; } @@ -796,7 +871,7 @@ impl GroupsRegistry { return; } - // A relay may answer with more than the filter asked for. + // A relay may answer with more than the reconciliation asked for. let Some(group) = metadata.id().cloned() else { return; }; @@ -815,6 +890,7 @@ impl GroupsRegistry { discovery.upsert( GroupCandidate { key: GroupKey::new(relay, group), + created_at: event.created_at, metadata, }, cx, @@ -822,19 +898,6 @@ impl GroupsRegistry { }); } - /// The contacts' group lists have been answered: resolve the metadata. - fn handle_lists_settled(&mut self, cx: &mut Context) { - self.discovery - .update(cx, |discovery, cx| discovery.settle_lists(cx)); - self.sync_previews(cx); - } - - fn handle_preview_settled(&mut self, id: &SubscriptionId, cx: &mut Context) { - self.discovery.update(cx, |discovery, cx| { - discovery.settle_preview(id, cx); - }); - } - pub fn load_more(&mut self, key: &GroupKey, cx: &mut Context) { let Some(group) = self.group(key, cx) else { return; @@ -891,9 +954,7 @@ impl GroupsRegistry { let mut ids: Vec = routes.into_keys().collect(); ids.push(list_subscription_id()); ids.push(fork_subscription_id()); - ids.push(discovery_subscription_id()); ids.extend(self.browse.keys().cloned()); - ids.extend(self.discovery.read(cx).preview_ids()); self.browse.clear(); self.discovery @@ -1362,13 +1423,7 @@ impl GroupsRegistry { })); } - fn handle_event( - &mut self, - id: SubscriptionId, - relay: RelayUrl, - event: Event, - cx: &mut Context, - ) { + fn handle_event(&mut self, id: SubscriptionId, event: Event, cx: &mut Context) { if id == list_subscription_id() { match GroupList::parse(&event) { Ok(list) => self.merge(list, cx), @@ -1382,26 +1437,11 @@ impl GroupsRegistry { return; } - if id == contacts_subscription_id() { - self.absorb_group_list(&event, cx); - return; - } - - if id == discovery_subscription_id() { - self.absorb_discovered(relay, &event, cx); - return; - } - if self.browse.contains_key(&id) { self.absorb_candidate(&id, &event, cx); return; } - if self.discovery.read(cx).preview_relay(&id).is_some() { - self.absorb_preview(&id, &event, cx); - return; - } - let Some(key) = self.routes.get(&id).cloned() else { return; }; @@ -1422,9 +1462,9 @@ impl GroupsRegistry { return; }; - if let Some((group, members)) = parse_members(event) { + if let Some((group, created_at, members)) = parse_members(event) { self.discovery.update(cx, |discovery, cx| { - discovery.set_members(GroupKey::new(relay, group), members, cx); + discovery.set_members(GroupKey::new(relay, group), created_at, members, cx); }); return; } @@ -1440,27 +1480,12 @@ impl GroupsRegistry { cx.emit(GroupsEvent::Browsed(Box::new(GroupCandidate { key: GroupKey::new(relay, group), + created_at: event.created_at, metadata, }))); } fn handle_eose(&mut self, id: SubscriptionId, cx: &mut Context) { - if id == contacts_subscription_id() { - self.handle_lists_settled(cx); - return; - } - - if id == discovery_subscription_id() { - self.discovery - .update(cx, |discovery, cx| discovery.settle_sweep(cx)); - return; - } - - if self.discovery.read(cx).preview_relay(&id).is_some() { - self.handle_preview_settled(&id, cx); - return; - } - let Some(key) = self.routes.get(&id).cloned() else { return; }; @@ -1489,25 +1514,6 @@ impl GroupsRegistry { } fn handle_closed(&mut self, id: SubscriptionId, reason: String, cx: &mut Context) { - if id == contacts_subscription_id() { - log::warn!("nip29: the contacts' group lists were refused: {reason}"); - self.handle_lists_settled(cx); - return; - } - - if id == discovery_subscription_id() { - log::warn!("nip29: the group discovery was closed: {reason}"); - self.discovery - .update(cx, |discovery, cx| discovery.settle_sweep(cx)); - return; - } - - if let Some(relay) = self.discovery.read(cx).preview_relay(&id) { - log::warn!("nip29: fetching previews from {relay} was refused: {reason}"); - self.handle_preview_settled(&id, cx); - return; - } - if let Some(relay) = self.browse.get(&id) { log::warn!("nip29: browsing {relay} was refused: {reason}"); return; @@ -1565,7 +1571,7 @@ impl GroupsRegistry { const MAX_PROCESSED: usize = 10_000; while let Some(notification) = notifications.next().await { - let ClientNotification::Message { relay_url, message } = notification else { + let ClientNotification::Message { message, .. } = notification else { continue; }; @@ -1584,7 +1590,7 @@ impl GroupsRegistry { } processed .insert(event.id) - .then(|| Signal::Event(id, relay_url, Box::new(event.into_owned()))) + .then(|| Signal::Event(id, Box::new(event.into_owned()))) } } RelayMessage::EndOfStoredEvents(id) => { @@ -1610,8 +1616,8 @@ impl GroupsRegistry { self.signal_consumer = Some(cx.spawn(async move |this, cx| { while let Ok(signal) = rx.recv_async().await { this.update(cx, |this, cx| match signal { - Signal::Event(id, relay, event) => { - this.handle_event(id, relay, *event, cx); + Signal::Event(id, event) => { + this.handle_event(id, *event, cx); } Signal::Eose(id) => { this.handle_eose(id, cx); @@ -1622,16 +1628,6 @@ impl GroupsRegistry { Signal::BrowseSettled(id) => { this.settle_browse(id, cx); } - Signal::ContactListsSettled => { - this.handle_lists_settled(cx); - } - Signal::ContactPreviewSettled(id) => { - this.handle_preview_settled(&id, cx); - } - Signal::DiscoverySettled => { - this.discovery - .update(cx, |discovery, cx| discovery.settle_sweep(cx)); - } Signal::Forks => { this.sync_forks(true, cx); } diff --git a/crates/nip29/src/protocol.rs b/crates/nip29/src/protocol.rs index 1607b2b0..6fc3ab4d 100644 --- a/crates/nip29/src/protocol.rs +++ b/crates/nip29/src/protocol.rs @@ -335,6 +335,17 @@ impl GroupMetadata { .identifier() .ok_or_else(|| anyhow!("group metadata has no d tag"))?; + let content: serde_json::Value = + serde_json::from_str(event.content.as_str()).unwrap_or(serde_json::Value::Null); + + let field = |key: &str| -> Option { + content + .get(key) + .and_then(serde_json::Value::as_str) + .map(str::to_owned) + .or_else(|| tag_value(&event.tags, key).map(str::to_owned)) + }; + let mut supported_kinds: Option> = None; for tag in event @@ -363,10 +374,10 @@ impl GroupMetadata { } Ok(Self { - name: tag_value(&event.tags, "name").map(str::to_owned), - picture: tag_value(&event.tags, "picture").map(str::to_owned), - banner: tag_value(&event.tags, "banner").map(str::to_owned), - about: tag_value(&event.tags, "about").map(str::to_owned), + name: field("name"), + picture: field("picture"), + banner: field("banner"), + about: field("about"), private: has_tag(&event.tags, "private"), restricted: has_tag(&event.tags, "restricted"), hidden: has_tag(&event.tags, "hidden"), @@ -412,6 +423,8 @@ impl GroupMetadata { #[derive(Debug, Clone)] pub struct GroupCandidate { pub key: GroupKey, + /// The timestamp of the metadata event the candidate was built from. + pub(crate) created_at: Timestamp, pub(crate) metadata: GroupMetadata, } diff --git a/crates/workspace/src/panels/browse.rs b/crates/workspace/src/panels/browse.rs index 7fcef5dc..772889e7 100644 --- a/crates/workspace/src/panels/browse.rs +++ b/crates/workspace/src/panels/browse.rs @@ -5,7 +5,7 @@ use gpui::prelude::FluentBuilder; use gpui::{ AnyElement, App, AppContext, Context, Entity, EventEmitter, FocusHandle, Focusable, IntoElement, ParentElement, Render, ScrollStrategy, SharedString, Styled, Subscription, - WeakEntity, Window, div, px, size, + WeakEntity, Window, div, px, relative, size, }; use gpui_base::{VirtualListScrollHandle, v_virtual_list}; use nip29::{Discovery, GroupCandidate, GroupKey, GroupsEvent, GroupsRegistry}; @@ -24,9 +24,7 @@ use ui::{Disableable, Icon, IconName, Sizable, StyledExt, WindowExtension, h_fle const TITLE: &str = "Browse"; const INPUT_PLACEHOLDER: &str = "wss://relay.example.com"; - -const CARD_HEIGHT: f32 = 150.0; -const DESCRIPTION_HEIGHT: f32 = 32.0; +const CARD_HEIGHT: f32 = 164.0; const NO_DESCRIPTION: &str = "No description"; pub fn init(dock: WeakEntity, window: &mut Window, cx: &mut App) -> Entity { @@ -201,7 +199,7 @@ impl BrowsePanel { .cloned() .collect(); - self.render_card_rows(&cards, range, cx) + self.render_card_rows(&cards, cx) } fn render_browse_rows( @@ -213,20 +211,23 @@ impl BrowsePanel { let end = (range.end * 2).min(self.candidates.len()); let cards = self.candidates.get(start..end).unwrap_or(&[]); - self.render_card_rows(cards, range, cx) + self.render_card_rows(cards, cx) } /// A row holds two group cards side by side. + /// + /// The cards are the relative slice of the visible rows, so they pair up + /// in order, two per row. fn render_card_rows( &self, cards: &[GroupCandidate], - range: Range, cx: &mut Context, ) -> Vec { - range - .map(|row| { - let first = cards.get(row * 2); - let second = cards.get(row * 2 + 1); + cards + .chunks(2) + .map(|pair| { + let first = pair.first(); + let second = pair.get(1); h_flex() .w_full() @@ -234,7 +235,12 @@ impl BrowsePanel { .p_2() .gap_2() .children(first.map(|card| self.render_card(card, cx))) - .children(second.map(|card| self.render_card(card, cx))) + // Keep an odd last card in one column: an empty spacer + // takes the second slot so the card cannot stretch. + .child(match second { + Some(card) => self.render_card(card, cx), + None => div().flex_1().into_any_element(), + }) .into_any_element() }) .collect() @@ -261,7 +267,7 @@ impl BrowsePanel { let joined = registry.read(cx).group(&key, cx).is_some(); let action = if joined { - Button::new(format!("browse-joined-{seed}")) + Button::new(format!("joined-{seed}")) .label("Joined") .small() .ghost() @@ -271,7 +277,7 @@ impl BrowsePanel { let panel = cx.entity().downgrade(); let clicked = key; - Button::new(format!("browse-join-{seed}")) + Button::new(format!("join-{seed}")) .label("Join") .small() .ghost() @@ -290,37 +296,38 @@ impl BrowsePanel { .overflow_hidden() .border_1() .border_color(cx.theme().border) - .rounded_lg() - .bg(cx.theme().surface_background) + .rounded(cx.theme().radius) .child( v_flex() .flex_1() .min_h_0() .p_2() - .gap_1p5() + .gap_2() .child( h_flex() .w_full() .items_center() - .gap_1p5() + .gap_2() .when(!hide_avatar, |this| { - this.child( - Avatar::from_source(picture) - .seed(seed) - .small() - .flex_shrink_0(), - ) + this.child(Avatar::from_source(picture).seed(seed).flex_shrink_0()) }) .child( v_flex() .flex_1() .min_w_0() - .child(div().text_sm().font_medium().truncate().child(name)) .child( div() + .text_sm() + .truncate() + .font_semibold() + .line_height(relative(1.2)) + .child(name), + ) + .child( + div() + .truncate() .text_xs() .text_color(cx.theme().text_placeholder) - .truncate() .child(relay), ), ) @@ -328,44 +335,37 @@ impl BrowsePanel { ) .child( div() - .h(px(DESCRIPTION_HEIGHT)) - .text_xs() + .min_w_0() + .text_sm() .text_color(cx.theme().text_muted) - .when_none(&about, |this| { - this.child( - div() - .text_color(cx.theme().text_placeholder) - .child(NO_DESCRIPTION), - ) - }) - .when_some(about, |this, about| this.child(about)), + .text_ellipsis() + .line_clamp(3) + .map(|this| { + if let Some(about) = about { + this.child(about) + } else { + this.child(NO_DESCRIPTION) + } + }), ) .child(div().flex_1()) .child( h_flex() .w_full() - .items_center() - .gap_1() - .text_xs() - .text_color(cx.theme().text_placeholder) - .child(Icon::new(IconName::Group).text_color(cx.theme().icon_muted)) - .child(format!("{members} members")), + .gap_1p5() + .child(Icon::new(IconName::Group)) + .child( + div() + .text_sm() + .text_color(cx.theme().text_muted) + .child(SharedString::from(format!("{members} members"))), + ), ), ) .into_any_element() } } -fn notice(text: &'static str, cx: &App) -> AnyElement { - h_flex() - .size_full() - .justify_center() - .text_sm() - .text_color(cx.theme().text_muted) - .child(text) - .into_any_element() -} - impl Panel for BrowsePanel { fn panel_id(&self) -> SharedString { TITLE.into() @@ -460,7 +460,7 @@ impl Render for BrowsePanel { .child( h_flex() .gap_2() - .p_3() + .px_2() .flex_shrink_0() .child(Input::new(&self.input).cleanable(true).flex_1()) .child( @@ -475,16 +475,6 @@ impl Render for BrowsePanel { this.child(Indicator::new().small().color(cx.theme().icon_accent)) }), ) - .when(discover && discovered_count > 0, |this| { - this.child( - div() - .px_3() - .pb_1() - .text_xs() - .text_color(cx.theme().text_muted) - .child("Discovered groups"), - ) - }) .child( div() .flex_1() @@ -495,3 +485,13 @@ impl Render for BrowsePanel { ) } } + +fn notice(text: &'static str, cx: &App) -> AnyElement { + h_flex() + .size_full() + .justify_center() + .text_sm() + .text_color(cx.theme().text_muted) + .child(text) + .into_any_element() +}