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

simplify nip29 discovery and sync metadata via negentropy

Browse Source
This commit is contained in:
reya
2026-10-07 11:54:33 +07:00
parent d0f02dfb4b
commit e472077941
6 changed files with 377 additions and 409 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/nip29/Cargo.toml
+2
View File
@@ -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" }
crates/nip29/src/discovery.rs
+31 -76
View File
@@ -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<RelayUrl, HashSet<GroupId>>,
/// The groups whose metadata has resolved so far.
discovered: Vec<GroupCandidate>,
/// The member counts of the known groups.
members: HashMap<GroupKey, usize>,
/// 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,
members: HashMap<GroupKey, (Timestamp, usize)>,
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>) {
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>) {
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;
/// Every pending discovery step has finished.
pub(crate) fn settle(&mut self, cx: &mut Context<Self>) {
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<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.
/// Add a discovered group, keeping the newest metadata.
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,
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<Self>) {
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<Self>,
) {
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<usize> {
self.members.get(key).copied()
self.members.get(key).map(|(_, members)| *members)
}
pub(crate) fn reset(&mut self, cx: &mut Context<Self>) {
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();
}
}
crates/nip29/src/lib.rs
+262 -266
View File
@@ -22,11 +22,8 @@ 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-";
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>),
Event(SubscriptionId, Box<Event>),
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::<HashSet<_>>().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<RelayUrl>,
) -> HashMap<EventId, HashSet<RelayUrl>> {
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<EventId, HashSet<RelayUrl>>,
) -> Result<Vec<(RelayUrl, Event)>, Error> {
if received.is_empty() {
return Ok(Vec::new());
}
let events: HashMap<EventId, Event> = client
.database()
.query(Filter::new().ids(received.keys().cloned()))
.await?
.into_iter()
.map(|event| (event.id, event))
.collect();
Ok(received
.into_iter()
.flat_map(|(id, relays)| match events.get(&id) {
Some(event) => relays
.into_iter()
.map(|relay| (relay, event.clone()))
.collect(),
None => Vec::new(),
})
.collect())
}
impl EventEmitter<GroupsEvent> for GroupsRegistry {}
@@ -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<PublicKey, Event> =
lists.into_iter().fold(HashMap::new(), |mut newest, event| {
match newest.get(&event.pubkey) {
Some(current) if current.created_at >= event.created_at => newest,
_ => {
newest.insert(event.pubkey, event);
newest
}
}
});
for event in newest.into_values() {
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<PublicKey>, cx: &mut Context<Self>) {
/// Reconcile the contacts' group lists with the pool relays, returning
/// the newest lists now stored in the database.
fn sync_lists(
&self,
authors: HashSet<PublicKey>,
cx: &mut Context<Self>,
) -> Task<Result<Vec<Event>, Error>> {
if authors.is_empty() {
// 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<RelayUrl> = 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<Self>) {
/// Reconcile the groups of every pool relay, returning the newly
/// downloaded events with the relay that served them.
fn sync_sweep(&self, cx: &mut Context<Self>) -> Task<Result<Vec<(RelayUrl, Event)>, Error>> {
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let 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<RelayUrl> = 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<Self>) {
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<Self>) {
/// Reconcile the referenced groups with the relays hosting them,
/// returning the newly downloaded events with the relay that served
/// them.
fn sync_previews(&self, cx: &mut Context<Self>) -> Task<Result<Vec<(RelayUrl, Event)>, Error>> {
let refs = self.discovery.read(cx).refs().clone();
if refs.is_empty() {
return Task::ready(Ok(Vec::new()));
}
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
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<EventId> = 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<EventId, Event> = 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::<Vec<_>>()
})
.collect())
})
}
fn absorb_group_list(&mut self, event: &Event, cx: &mut Context<Self>) {
@@ -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<Event>, cx: &mut Context<Self>) {
let newest: HashMap<PublicKey, Event> =
lists.into_iter().fold(HashMap::new(), |mut newest, event| {
match newest.get(&event.pubkey) {
Some(current) if current.created_at >= event.created_at => newest,
_ => {
newest.insert(event.pubkey, event);
newest
}
}
});
for event in newest.into_values() {
self.absorb_group_list(&event, cx);
}
}
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;
};
if let Some((group, members)) = parse_members(event) {
fn absorb_preview(&mut self, relay: RelayUrl, event: &Event, cx: &mut Context<Self>) {
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>) {
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>) {
self.discovery.update(cx, |discovery, cx| {
discovery.settle_preview(id, cx);
});
}
pub fn load_more(&mut self, key: &GroupKey, cx: &mut Context<Self>) {
let Some(group) = self.group(key, cx) else {
return;
@@ -891,9 +954,7 @@ 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.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<Self>,
) {
fn handle_event(&mut self, id: SubscriptionId, event: Event, cx: &mut Context<Self>) {
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<Self>) {
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<Self>) {
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);
}
crates/nip29/src/protocol.rs
+17 -4
View File
@@ -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<String> {
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<Vec<Kind>> = 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,
}
crates/workspace/src/panels/browse.rs
+63 -63
View File
@@ -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<DockArea>, window: &mut Window, cx: &mut App) -> Entity<BrowsePanel> {
@@ -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<usize>,
cx: &mut Context<Self>,
) -> Vec<AnyElement> {
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()
}