reya/coop
1
0
Fork 1
Code Issues Pull Requests 1 Releases 3 Activity

extract group discovery into a discovery entity

Browse Source
This commit is contained in:
reya
2026-10-07 09:10:25 +07:00
parent c058270eff
commit 96ceacb5fb
3 changed files with 429 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/nip29/src/discovery.rs
+158
View File
@@ -0,0 +1,158 @@
use std::collections::{HashMap, HashSet};
use gpui::Context;
use nostr_sdk::prelude::*;
use crate::protocol::{GroupCandidate, GroupId, 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<RelayUrl, HashSet<GroupId>>,
/// The groups whose metadata has resolved so far.
discovered: Vec<GroupCandidate>,
/// The in-flight preview subscriptions, keyed by subscription id.
previews: HashMap<SubscriptionId, RelayUrl>,
/// Whether the contacts' group lists have been answered.
lists_settled: bool,
/// Whether the pool-wide metadata discovery has been answered.
sweep_settled: bool,
}
impl Discovery {
pub fn discovered(&self) -> &[GroupCandidate] {
&self.discovered
}
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
}
/// The discovery is starting over: nothing has answered yet.
pub(crate) fn begin(&mut self, cx: &mut Context<Self>) {
self.lists_settled = false;
self.sweep_settled = false;
cx.notify();
}
/// There is nothing on the way: there is nothing to discover.
pub(crate) fn settle_all(&mut self, cx: &mut Context<Self>) {
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>) {
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>) {
self.sweep_settled = true;
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.
pub(crate) fn refs(&self) -> &HashMap<RelayUrl, HashSet<GroupId>> {
&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<RelayUrl> {
self.previews.get(id).cloned()
}
/// The ids of the in-flight preview subscriptions.
pub(crate) fn preview_ids(&self) -> Vec<SubscriptionId> {
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<Self>) -> bool {
let removed = self.previews.remove(id).is_some();
if removed {
cx.notify();
}
removed
}
/// Add or replace a discovered group.
pub(crate) fn upsert(&mut self, candidate: GroupCandidate, cx: &mut Context<Self>) {
match self.discovered.iter_mut().find(|d| d.key == candidate.key) {
Some(existing) => *existing = candidate,
None => self.discovered.push(candidate),
}
self.discovered.sort_by(|a, b| {
a.display_name()
.to_lowercase()
.cmp(&b.display_name().to_lowercase())
});
cx.notify();
}
pub(crate) fn reset(&mut self, cx: &mut Context<Self>) {
self.refs.clear();
self.discovered.clear();
self.previews.clear();
self.lists_settled = false;
self.sweep_settled = false;
cx.notify();
}
}
crates/nip29/src/lib.rs
+227 -245
View File
@@ -10,9 +10,11 @@ use nostr_sdk::prelude::*;
use smallvec::{SmallVec, smallvec};
use state::NostrRegistry;
mod discovery;
mod group;
mod protocol;
pub use discovery::Discovery;
pub use group::*;
pub use protocol::*;
@@ -21,6 +23,7 @@ pub(crate) static LOCAL_KEYS: LazyLock<Keys> = 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-";
@@ -43,18 +46,17 @@ pub enum GroupsEvent {
Error(String),
Browsed(Box<GroupCandidate>),
BrowseSettled(RelayUrl),
Suggested(Box<GroupCandidate>),
SuggestSettled,
}
#[derive(Debug)]
enum Signal {
Event(SubscriptionId, Event),
Event(SubscriptionId, RelayUrl, Box<Event>),
Eose(SubscriptionId),
Closed(SubscriptionId, String),
BrowseSettled(SubscriptionId),
ContactListsSettled,
ContactPreviewSettled(SubscriptionId),
DiscoverySettled,
Forks,
}
@@ -70,6 +72,10 @@ 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))
}
@@ -101,6 +107,7 @@ fn is_own_subscription(id: &SubscriptionId) -> bool {
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)
@@ -117,16 +124,8 @@ pub struct GroupsRegistry {
observers: HashMap<GroupKey, Subscription>,
forks: HashMap<PublicKey, (Timestamp, GroupList)>,
fork_authors: HashSet<PublicKey>,
/// The follow-list authors the contacts subscription covers.
contact_authors: HashSet<PublicKey>,
/// The (relay, group) references of the contacts, minus the joined groups.
contact_refs: HashMap<RelayUrl, HashSet<GroupId>>,
/// The contact groups whose metadata has resolved so far.
suggestions: Vec<GroupCandidate>,
/// The in-flight preview subscriptions, keyed by subscription id.
previews: HashMap<SubscriptionId, RelayUrl>,
/// Whether the contacts' group lists have been answered.
contact_settled: bool,
/// The groups discovered from the cache and the relays.
discovery: Entity<Discovery>,
signal_tx: flume::Sender<Signal>,
signal_rx: flume::Receiver<Signal>,
notification_listener: Option<Task<()>>,
@@ -181,11 +180,7 @@ impl GroupsRegistry {
observers: HashMap::new(),
forks: HashMap::new(),
fork_authors: HashSet::new(),
contact_authors: HashSet::new(),
contact_refs: HashMap::new(),
suggestions: Vec::new(),
previews: HashMap::new(),
contact_settled: false,
discovery: cx.new(|_| Discovery::default()),
signal_tx: tx,
signal_rx: rx,
notification_listener: None,
@@ -314,16 +309,32 @@ impl GroupsRegistry {
cx.emit(GroupsEvent::BrowseSettled(relay));
}
pub fn load_contacts_groups(&mut self, cx: &mut Context<Self>) {
/// The groups discovered from the cache and the relays.
pub fn discovery(&self) -> Entity<Discovery> {
self.discovery.clone()
}
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_all(cx);
});
return;
}
let task: Task<Result<HashSet<PublicKey>, Error>> = cx.background_spawn(async move {
// 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);
@@ -336,48 +347,26 @@ impl GroupsRegistry {
.map(|event| event.tags.public_keys().collect())
.unwrap_or_default();
Ok(contacts)
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 = match task.await {
Ok(authors) => authors,
let (authors, lists) = match task.await {
Ok(ok) => ok,
Err(error) => {
log::warn!("nip29: loading the contact list failed: {error}");
return;
log::warn!("nip29: loading the cached groups failed: {error}");
(HashSet::new(), Vec::new())
}
};
if authors.is_empty() {
if let Err(error) = this.update(cx, |this, cx| {
// Nobody is followed, so there is nothing to discover.
this.contact_settled = true;
cx.emit(GroupsEvent::SuggestSettled);
cx.notify();
}) {
log::warn!("nip29: settling the contacts failed: {error}");
}
return;
}
// Show the cached lists before the relays answer.
let lists = match this.update(cx, |this, cx| this.start_contacts_hydration(authors, cx))
{
Ok(Some(task)) => match task.await {
Ok(lists) => lists,
Err(error) => {
log::warn!("nip29: loading the cached group lists failed: {error}");
Vec::new()
}
},
Ok(None) => return,
Err(error) => {
log::warn!("nip29: hydrating the contacts failed: {error}");
return;
}
};
let refs = match this.update(cx, |this, cx| {
if let Err(error) = this.update(cx, |this, cx| {
let newest: HashMap<PublicKey, Event> =
lists.into_iter().fold(HashMap::new(), |mut newest, event| {
match newest.get(&event.pubkey) {
@@ -390,20 +379,14 @@ impl GroupsRegistry {
});
for event in newest.into_values() {
this.absorb_contact_list(&event, cx);
this.absorb_group_list(&event, cx);
}
this.contact_refs.clone()
}) {
Ok(refs) => refs,
Err(error) => {
log::warn!("nip29: applying the cached group lists failed: {error}");
return;
}
};
log::warn!("nip29: applying the cached group lists failed: {error}");
return;
}
// Show the cached metadata before the relays answer.
let previews = match this.update(cx, |this, cx| this.load_cached_previews(refs, cx)) {
let previews = match this.update(cx, |this, cx| this.load_cached_previews(cx)) {
Ok(task) => match task.await {
Ok(previews) => previews,
Err(error) => {
@@ -412,67 +395,25 @@ impl GroupsRegistry {
}
},
Err(error) => {
log::warn!("nip29: hydrating the contacts failed: {error}");
log::warn!("nip29: discovering the groups failed: {error}");
return;
}
};
if let Err(error) = this.update(cx, |this, cx| {
this.absorb_cached_previews(previews, cx);
this.subscribe_contacts(cx);
this.subscribe_contacts(authors, cx);
this.subscribe_discovery(cx);
}) {
log::warn!("nip29: applying the cached group metadata failed: {error}");
log::warn!("nip29: discovering the groups failed: {error}");
}
}));
}
/// The contact groups whose metadata has resolved so far.
pub fn contacts_candidates(&self) -> &[GroupCandidate] {
&self.suggestions
}
/// Whether the contact lists or any of their previews are still on the way.
pub fn contacts_loading(&self) -> bool {
!self.contact_settled || !self.previews.is_empty()
}
/// Reset the contacts state and start loading the cached group lists.
fn start_contacts_hydration(
&mut self,
authors: HashSet<PublicKey>,
cx: &mut Context<Self>,
) -> Option<Task<Result<Vec<Event>, Error>>> {
if authors == self.contact_authors {
return None;
}
self.contact_authors = authors;
self.contact_refs.clear();
self.suggestions.clear();
self.previews.clear();
self.contact_settled = false;
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let filter = Filter::new()
.authors(self.contact_authors.iter().copied())
.kind(Kind::SimpleGroups);
let task: Task<Result<Vec<Event>, Error>> = cx.background_spawn(async move {
Ok(client.database().query(filter).await?.into_iter().collect())
});
cx.notify();
Some(task)
}
/// Load the cached metadata for the referenced groups from the database.
fn load_cached_previews(
&self,
refs: HashMap<RelayUrl, HashSet<GroupId>>,
cx: &mut Context<Self>,
) -> Task<Result<Vec<Event>, Error>> {
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()));
}
@@ -488,10 +429,8 @@ impl GroupsRegistry {
})
}
/// Seed the suggestions from the cached group metadata.
/// Seed the discoveries from the cached group metadata.
fn absorb_cached_previews(&mut self, previews: Vec<Event>, cx: &mut Context<Self>) {
// Keep the newest metadata per group, the database may still hold
// superseded versions.
let newest: HashMap<GroupId, (Timestamp, GroupMetadata)> =
previews
.into_iter()
@@ -517,7 +456,9 @@ impl GroupsRegistry {
}
});
for (relay, ids) in self.contact_refs.clone() {
let mut candidates = Vec::new();
for (relay, ids) in self.discovery.read(cx).refs().clone() {
for id in ids {
let Some((_, metadata)) = newest.get(&id) else {
continue;
@@ -528,25 +469,33 @@ impl GroupsRegistry {
continue;
}
self.upsert_suggestion(
GroupCandidate {
key: GroupKey::new(relay.clone(), id.clone()),
metadata: metadata.clone(),
},
cx,
);
candidates.push(GroupCandidate {
key: GroupKey::new(relay.clone(), id.clone()),
metadata: metadata.clone(),
});
}
}
self.discovery.update(cx, |discovery, cx| {
for candidate in candidates {
discovery.upsert(candidate, cx);
}
});
}
/// Subscribe to the contacts' public group lists.
fn subscribe_contacts(&mut self, cx: &mut Context<Self>) {
fn subscribe_contacts(&mut self, authors: HashSet<PublicKey>, cx: &mut Context<Self>) {
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;
}
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let filter = Filter::new()
.authors(self.contact_authors.iter().copied())
.kind(Kind::SimpleGroups);
let filter = Filter::new().authors(authors).kind(Kind::SimpleGroups);
let opts = SubscribeAutoCloseOptions::default()
.exit_policy(ReqExitPolicy::ExitOnEOSE)
@@ -577,19 +526,87 @@ impl GroupsRegistry {
cx.notify();
}
/// Ask every relay hosting referenced groups for the groups' metadata.
fn sync_contact_previews(&mut self, cx: &mut Context<Self>) {
/// Subscribe to the metadata of every group the pool relays host.
fn subscribe_discovery(&mut self, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
for (relay, ids) in self.contact_refs.clone() {
let filter = Filter::new().kind(Kind::GroupMetadata).limit(BROWSE_LIMIT);
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 {
let refused = match client
.subscribe(ReqTarget::auto([filter]))
.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();
}
/// Absorb a discovered group metadata event from the relay serving it.
fn absorb_discovered(&mut self, relay: RelayUrl, event: &Event, cx: &mut Context<Self>) {
let metadata = match GroupMetadata::parse(event) {
Ok(metadata) => metadata,
Err(error) => {
log::warn!("nip29: 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),
metadata,
},
cx,
);
});
}
/// Ask every relay hosting referenced groups for the groups' metadata.
fn sync_previews(&mut self, cx: &mut Context<Self>) {
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.previews.values().any(|in_flight| in_flight == &relay) {
if self.discovery.read(cx).preview_in_flight(&relay) {
continue;
}
let id = preview_subscription(&relay);
self.previews.insert(id.clone(), relay.clone());
self.discovery.update(cx, |discovery, _| {
discovery.track_preview(id.clone(), relay.clone());
});
let client = client.clone();
let filter = Filter::new()
@@ -626,11 +643,9 @@ impl GroupsRegistry {
}
}));
}
cx.notify();
}
fn absorb_contact_list(&mut self, event: &Event, cx: &mut Context<Self>) {
fn absorb_group_list(&mut self, event: &Event, cx: &mut Context<Self>) {
let Ok(list) = GroupList::parse(event) else {
log::warn!(
"nip29: ignoring a malformed group list from {}",
@@ -648,38 +663,17 @@ impl GroupsRegistry {
})
.collect();
let mut added = false;
let added = self.discovery.update(cx, |discovery, cx| {
discovery.absorb_list(&list, &joined, cx)
});
for entry in list.entries() {
if joined.contains(&(entry.relay.clone(), entry.id.clone())) {
continue;
}
if self
.contact_refs
.entry(entry.relay.clone())
.or_default()
.insert(entry.id.clone())
{
added = true;
}
if added && self.discovery.read(cx).lists_settled() {
self.sync_previews(cx);
}
if !added {
return;
}
// Lists may still arrive after the first EOSE,
// so previews sync as the references grow as well.
if self.contact_settled {
self.sync_contact_previews(cx);
}
cx.notify();
}
fn absorb_contact_group(&mut self, id: &SubscriptionId, event: &Event, cx: &mut Context<Self>) {
let Some(relay) = self.previews.get(id).cloned() else {
fn absorb_preview(&mut self, id: &SubscriptionId, event: &Event, cx: &mut Context<Self>) {
let Some(relay) = self.discovery.read(cx).preview_relay(id) else {
return;
};
@@ -702,75 +696,37 @@ impl GroupsRegistry {
};
if !self
.contact_refs
.discovery
.read(cx)
.refs()
.get(&relay)
.is_some_and(|ids| ids.contains(&group))
{
return;
}
self.upsert_suggestion(
GroupCandidate {
key: GroupKey::new(relay, group),
metadata,
},
cx,
);
}
fn upsert_suggestion(&mut self, candidate: GroupCandidate, cx: &mut Context<Self>) {
match self.suggestions.iter_mut().find(|s| s.key == candidate.key) {
Some(existing) => *existing = candidate.clone(),
None => self.suggestions.push(candidate.clone()),
}
self.suggestions.sort_by(|a, b| {
a.display_name()
.to_lowercase()
.cmp(&b.display_name().to_lowercase())
self.discovery.update(cx, |discovery, cx| {
discovery.upsert(
GroupCandidate {
key: GroupKey::new(relay, group),
metadata,
},
cx,
);
});
cx.emit(GroupsEvent::Suggested(Box::new(candidate)));
cx.notify();
}
/// Drop suggestions whose group is no longer referenced by any contact.
fn prune_suggestions(&mut self, cx: &mut Context<Self>) {
let before = self.suggestions.len();
self.suggestions.retain(|suggestion| {
self.contact_refs
.get(suggestion.key.relay())
.is_some_and(|ids| ids.contains(suggestion.key.id()))
});
if self.suggestions.len() != before {
cx.notify();
}
}
/// The contacts' group lists have been answered: resolve the metadata.
fn handle_contact_lists_settled(&mut self, cx: &mut Context<Self>) {
self.contact_settled = true;
self.sync_contact_previews(cx);
if self.previews.is_empty() {
self.prune_suggestions(cx);
cx.emit(GroupsEvent::SuggestSettled);
cx.notify();
}
fn handle_lists_settled(&mut self, cx: &mut Context<Self>) {
self.discovery
.update(cx, |discovery, cx| discovery.settle_lists(cx));
self.sync_previews(cx);
}
fn handle_contact_preview_settled(&mut self, id: &SubscriptionId, cx: &mut Context<Self>) {
if self.previews.remove(id).is_none() {
return;
}
if self.previews.is_empty() {
self.prune_suggestions(cx);
cx.emit(GroupsEvent::SuggestSettled);
cx.notify();
}
fn handle_preview_settled(&mut self, id: &SubscriptionId, cx: &mut Context<Self>) {
self.discovery.update(cx, |discovery, cx| {
discovery.settle_preview(id, cx);
});
}
pub fn load_more(&mut self, key: &GroupKey, cx: &mut Context<Self>) {
@@ -829,15 +785,13 @@ impl GroupsRegistry {
let mut ids: Vec<SubscriptionId> = 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.previews.keys().cloned());
ids.extend(self.discovery.read(cx).preview_ids());
self.browse.clear();
self.previews.clear();
self.contact_authors.clear();
self.contact_refs.clear();
self.suggestions.clear();
self.contact_settled = false;
self.discovery
.update(cx, |discovery, cx| discovery.reset(cx));
self.tasks.push(cx.background_spawn(async move {
for id in ids {
@@ -1302,7 +1256,13 @@ impl GroupsRegistry {
}));
}
fn handle_event(&mut self, id: SubscriptionId, event: Event, cx: &mut Context<Self>) {
fn handle_event(
&mut self,
id: SubscriptionId,
relay: RelayUrl,
event: Event,
cx: &mut Context<Self>,
) {
if id == list_subscription_id() {
match GroupList::parse(&event) {
Ok(list) => self.merge(list, cx),
@@ -1317,7 +1277,12 @@ impl GroupsRegistry {
}
if id == contacts_subscription_id() {
self.absorb_contact_list(&event, cx);
self.absorb_group_list(&event, cx);
return;
}
if id == discovery_subscription_id() {
self.absorb_discovered(relay, &event, cx);
return;
}
@@ -1326,8 +1291,8 @@ impl GroupsRegistry {
return;
}
if self.previews.contains_key(&id) {
self.absorb_contact_group(&id, &event, cx);
if self.discovery.read(cx).preview_relay(&id).is_some() {
self.absorb_preview(&id, &event, cx);
return;
}
@@ -1368,12 +1333,18 @@ impl GroupsRegistry {
fn handle_eose(&mut self, id: SubscriptionId, cx: &mut Context<Self>) {
if id == contacts_subscription_id() {
self.handle_contact_lists_settled(cx);
self.handle_lists_settled(cx);
return;
}
if self.previews.contains_key(&id) {
self.handle_contact_preview_settled(&id, cx);
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;
}
@@ -1407,13 +1378,20 @@ impl GroupsRegistry {
fn handle_closed(&mut self, id: SubscriptionId, reason: String, cx: &mut Context<Self>) {
if id == contacts_subscription_id() {
log::warn!("nip29: the contacts' group lists were refused: {reason}");
self.handle_contact_lists_settled(cx);
self.handle_lists_settled(cx);
return;
}
if let Some(relay) = self.previews.get(&id).cloned() {
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_contact_preview_settled(&id, cx);
self.handle_preview_settled(&id, cx);
return;
}
@@ -1474,7 +1452,7 @@ 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;
};
@@ -1493,7 +1471,7 @@ impl GroupsRegistry {
}
processed
.insert(event.id)
.then(|| Signal::Event(id, event.into_owned()))
.then(|| Signal::Event(id, relay_url, Box::new(event.into_owned())))
}
}
RelayMessage::EndOfStoredEvents(id) => {
@@ -1519,8 +1497,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, event) => {
this.handle_event(id, event, cx);
Signal::Event(id, relay, event) => {
this.handle_event(id, relay, *event, cx);
}
Signal::Eose(id) => {
this.handle_eose(id, cx);
@@ -1532,10 +1510,14 @@ impl GroupsRegistry {
this.settle_browse(id, cx);
}
Signal::ContactListsSettled => {
this.handle_contact_lists_settled(cx);
this.handle_lists_settled(cx);
}
Signal::ContactPreviewSettled(id) => {
this.handle_contact_preview_settled(&id, cx);
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);
crates/workspace/src/panels/browse.rs
+44 -78
View File
@@ -6,7 +6,7 @@ use gpui::{
IntoElement, ParentElement, Render, SharedString, Styled, Subscription, WeakEntity, Window,
div, uniform_list,
};
use nip29::{GroupCandidate, GroupKey, GroupsEvent, GroupsRegistry};
use nip29::{Discovery, GroupCandidate, GroupKey, GroupsEvent, GroupsRegistry};
use nostr_sdk::prelude::RelayUrl;
use settings::AppSettings;
use smallvec::{SmallVec, smallvec};
@@ -35,19 +35,18 @@ pub struct BrowsePanel {
relay: Option<RelayUrl>,
/// The groups the relay has announced so far
candidates: Vec<GroupCandidate>,
/// The communities of the user's contacts
contacts: Vec<GroupCandidate>,
/// Whether the contacts' communities are still resolving
contacts_loading: bool,
/// The groups discovered from the cache and the relays
discovery: Entity<Discovery>,
/// Whether the relay is still answering
loading: bool,
_subscriptions: SmallVec<[Subscription; 2]>,
_subscriptions: SmallVec<[Subscription; 3]>,
}
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![];
@@ -59,6 +58,10 @@ impl BrowsePanel {
}),
);
subscriptions.push(cx.observe(&discovery, |_, _, cx| {
cx.notify();
}));
subscriptions.push(
cx.subscribe(&registry, |this, _registry, event, cx| match event {
GroupsEvent::Browsed(candidate) => {
@@ -67,19 +70,15 @@ impl BrowsePanel {
GroupsEvent::BrowseSettled(relay) => {
this.settle(relay.clone(), cx);
}
GroupsEvent::Suggested(candidate) => {
this.push_contact(candidate.as_ref().clone(), cx);
}
GroupsEvent::SuggestSettled => {
this.settle_contacts(cx);
}
_ => {}
}),
);
// Wait for the panel to leave the stack before kicking off the sync.
cx.defer_in(window, |this, _window, cx| {
this.load_contacts(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);
});
});
Self {
@@ -88,50 +87,12 @@ impl BrowsePanel {
input,
relay: None,
candidates: Vec::new(),
contacts: Vec::new(),
contacts_loading: true,
discovery,
loading: false,
_subscriptions: subscriptions,
}
}
/// Pull the resolved contacts' communities and start the contacts sync.
fn load_contacts(&mut self, cx: &mut Context<Self>) {
let registry = GroupsRegistry::global(cx);
let (candidates, loading) = registry.update(cx, |registry, cx| {
registry.load_contacts_groups(cx);
(
registry.contacts_candidates().to_vec(),
registry.contacts_loading(),
)
});
self.contacts = candidates;
self.contacts_loading = loading;
cx.notify();
}
fn push_contact(&mut self, candidate: GroupCandidate, cx: &mut Context<Self>) {
match self.contacts.iter_mut().find(|c| c.key == candidate.key) {
Some(existing) => *existing = candidate,
None => self.contacts.push(candidate),
}
self.contacts.sort_by(|a, b| {
a.display_name()
.to_lowercase()
.cmp(&b.display_name().to_lowercase())
});
cx.notify();
}
fn settle_contacts(&mut self, cx: &mut Context<Self>) {
self.contacts_loading = false;
cx.notify();
}
/// Browse the relay in the input.
fn browse(&mut self, window: &mut Window, cx: &mut Context<Self>) {
let registry = GroupsRegistry::global(cx);
@@ -211,10 +172,10 @@ impl BrowsePanel {
range: Range<usize>,
cx: &mut Context<Self>,
) -> Vec<AnyElement> {
let registry = GroupsRegistry::global(cx);
let hide_avatar = AppSettings::get_hide_avatar(cx);
let panel = cx.entity().downgrade();
let registry = GroupsRegistry::global(cx);
let joined: Vec<GroupKey> = registry
.read(cx)
.groups()
@@ -229,16 +190,19 @@ impl BrowsePanel {
.map(|candidate| {
let key = candidate.key.clone();
let name = SharedString::from(candidate.display_name().to_owned());
let seed = key.cache_tag();
let is_joined = joined.contains(&key);
let picture = candidate
.display_image()
.map(|picture| SharedString::from(picture.to_owned()));
let seed = key.cache_tag();
let detail = if relay_detail {
key.relay().domain().unwrap_or_else(|| key.relay().as_str())
} else {
key.id().as_str()
};
let is_joined = joined.contains(&key);
let action = if is_joined {
Button::new(format!("browse-joined-{seed}"))
@@ -335,9 +299,9 @@ impl Focusable for BrowsePanel {
impl Render for BrowsePanel {
fn render(&mut self, _window: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
let count = self.candidates.len();
let from_contacts = self.relay.is_none();
let discover = self.relay.is_none();
let (contacts, contact_count) = if from_contacts {
let (discovered, discovered_count, discovering) = if discover {
let registry = GroupsRegistry::global(cx);
let joined: Vec<GroupKey> = registry
.read(cx)
@@ -346,34 +310,37 @@ impl Render for BrowsePanel {
.map(|group| group.read(cx).key().clone())
.collect();
let contacts: Vec<GroupCandidate> = self
.contacts
let discovery = self.discovery.read(cx);
let discovering = discovery.loading();
let discovered: Vec<GroupCandidate> = discovery
.discovered()
.iter()
.filter(|candidate| !joined.contains(&candidate.key))
.cloned()
.collect();
let count = contacts.len();
(contacts, count)
let count = discovered.len();
(discovered, count, discovering)
} else {
(Vec::new(), 0)
(Vec::new(), 0, false)
};
let body = if from_contacts {
if contact_count > 0 {
let body = if discover {
if discovered_count > 0 {
uniform_list(
"browse-contacts",
contact_count,
"browse-discovered",
discovered_count,
cx.processor(move |this, range, _window, cx| {
this.render_rows(&contacts, true, range, cx)
this.render_rows(&discovered, true, range, cx)
}),
)
.h_full()
.into_any_element()
} else if self.contacts_loading {
notice("Loading communities from your contacts…", cx)
} else if discovering {
notice("Discovering groups…", cx)
} else {
notice("No communities from your contacts.", cx)
notice("No groups found.", cx)
}
} else if count > 0 {
uniform_list(
@@ -407,19 +374,18 @@ impl Render for BrowsePanel {
this.browse(window, cx);
})),
)
.when(
self.loading || (from_contacts && self.contacts_loading),
|this| this.child(Indicator::new().small().color(cx.theme().icon_accent)),
),
.when(self.loading || (discover && discovering), |this| {
this.child(Indicator::new().small().color(cx.theme().icon_accent))
}),
)
.when(from_contacts && contact_count > 0, |this| {
.when(discover && discovered_count > 0, |this| {
this.child(
div()
.px_3()
.pb_1()
.text_xs()
.text_color(cx.theme().text_muted)
.child("From your contacts"),
.child("Discovered groups"),
)
})
.child(div().flex_1().min_h_0().child(body))