add community browse panel

crates/community/src/discover.rs

@@ -1,7 +1,7 @@
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::LazyLock;
 
-use concord::cord05;
+use concord::cord05::{self, CommunityInvite};
 use nostr_sdk::prelude::*;
 use regex::Regex;
 
@@ -15,21 +15,23 @@ static INVITE_URL: LazyLock<Regex> = LazyLock::new(|| {
 });
 
 #[derive(Debug, Clone)]
-pub struct Invite {
+pub struct Listing {
     pub invite_url: String,
     pub link_signer: PublicKey,
-    pub source: Event,
+    pub community: Option<CommunityInvite>,
+    pub active_at: Option<Timestamp>,
 }
 
-impl Invite {
-    pub fn from_event(event: &Event) -> Option<Invite> {
+impl Listing {
+    pub fn from_event(event: &Event) -> Option<Listing> {
         let invite_url = extract_invite_urls(&event.content).into_iter().next()?;
         let link_signer = cord05::parse_link(&invite_url).ok()?.link_signer;
 
-        Some(Invite {
+        Some(Listing {
             invite_url,
             link_signer,
-            source: event.clone(),
+            community: None,
+            active_at: None,
         })
     }
 }
@@ -71,7 +73,7 @@ pub fn build_announcement_deletion(ids: &[EventId]) -> EventBuilder {
     EventBuilder::new(Kind::EventDeletion, "").tags(tags)
 }
 
-pub fn fold_announcements(announcements: &[Event], deletes: &[Event]) -> Vec<Invite> {
+pub fn fold_announcements(announcements: &[Event], deletes: &[Event]) -> Vec<Listing> {
     let authors: BTreeMap<EventId, PublicKey> = announcements
         .iter()
         .filter(|event| event.kind == Kind::Custom(KIND_COMMUNITY_ANNOUNCEMENT))
@@ -91,12 +93,12 @@ pub fn fold_announcements(announcements: &[Event], deletes: &[Event]) -> Vec<Inv
             continue;
         }
 
-        let Some(invite) = Invite::from_event(event) else {
+        let Some(listing) = Listing::from_event(event) else {
             continue;
         };
 
-        if seen.insert(invite.link_signer) {
-            folded.push(invite);
+        if seen.insert(listing.link_signer) {
+            folded.push(listing);
         }
     }
 

crates/community/src/history.rs

@@ -353,7 +353,7 @@ fn history_subscription() -> SubscriptionId {
     SubscriptionId::new(format!("history-{}", NEXT.fetch_add(1, Ordering::Relaxed)))
 }
 
-async fn within<F>(limit: Duration, future: F) -> Option<F::Output>
+pub(crate) async fn within<F>(limit: Duration, future: F) -> Option<F::Output>
 where
     F: Future,
 {

crates/community/src/lib.rs

@@ -23,12 +23,10 @@ mod rekey;
 mod sync;
 
 pub use community::*;
+pub use discover::Listing;
 pub use sync::*;
 
-/// How long a burst of relay notifications is collected before it is folded.
 const PUMP_WINDOW: Duration = Duration::from_millis(200);
-/// How long a community's relays may deliver nothing before
-/// its standing subscription is torn down and re-issued.
 const LIVE_ROTATE: Duration = Duration::from_secs(90);
 
 pub fn init(cx: &mut App) {
@@ -143,24 +141,23 @@ pub struct CommunityRegistry {
     index: HashMap<CommunityId, Entity<Community>>,
     /// The plane set each community was last subscribed with
     synced: HashMap<CommunityId, SubscriptionKey>,
-    /// When a relay last delivered something for a community,
-    /// which is the only evidence the standing subscription is alive.
+    /// When a relay last delivered something for a community.
     last_event: HashMap<CommunityId, Instant>,
-    /// One observer per tracked community, dropped on reset
+    /// One observer per tracked community, dropped on reset.
     observers: HashMap<CommunityId, Subscription>,
-    signal_tx: flume::Sender<Signal>,
-    signal_rx: flume::Receiver<Signal>,
     /// The page subscriptions in flight, shared with the notification pump.
     pages: PageRegistry,
     /// The rekey watch subscriptions, resolved to their community.
     watches: WatchRegistry,
-    tasks: SmallVec<[Task<Result<()>>; 2]>,
+    signal_tx: flume::Sender<Signal>,
+    signal_rx: flume::Receiver<Signal>,
     /// Notification listener task (cancelled on signer change)
     notification_listener: Option<Task<Result<()>>>,
     /// Signal consumer task (cancelled on signer change)
     signal_consumer: Option<Task<Result<()>>>,
     /// The round scheduler (cancelled on signer change)
     scheduler: Option<Task<Result<()>>>,
+    tasks: SmallVec<[Task<Result<()>>; 2]>,
     _subscriptions: SmallVec<[Subscription; 2]>,
 }
 
@@ -174,8 +171,9 @@ impl CommunityRegistry {
     }
 
     fn new(cx: &mut Context<Self>) -> Self {
-        let entity = cx.entity().downgrade();
         let nostr = NostrRegistry::global(cx);
+        let entity = cx.entity().downgrade();
+
         let (tx, rx) = flume::bounded::<Signal>(256);
         let mut subscriptions = smallvec![];
 
@@ -189,15 +187,14 @@ impl CommunityRegistry {
         }));
 
         cx.defer(move |cx| {
-            entity
-                .update(cx, |this, cx| {
-                    this.handle_notifications(cx);
-                    if nostr.read(cx).current_user().is_some() {
-                        this.subscribe_list(cx);
-                        this.load(cx);
-                    }
-                })
-                .ok();
+            let _ = entity.update(cx, |this, cx| {
+                this.handle_notifications(cx);
+
+                if nostr.read(cx).current_user().is_some() {
+                    this.subscribe_list(cx);
+                    this.load(cx);
+                }
+            });
         });
 
         Self {
@@ -637,8 +634,6 @@ impl CommunityRegistry {
             while let Ok(signal) = rx.recv_async().await {
                 match signal {
                     Signal::Event(id) => this.update(cx, |this, cx| {
-                        // The only proof a community's subscription is still
-                        // delivering anything.
                         this.last_event.insert(id, Instant::now());
                         this.refresh(id, cx);
                     })?,

crates/community/src/sync.rs

@@ -1,6 +1,7 @@
 use std::collections::{BTreeMap, BTreeSet};
 use std::path::PathBuf;
-use std::time::Duration;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
 
 use anyhow::{Context, Result};
 use concord::cord01::KIND_WRAP_EPHEMERAL;
@@ -20,10 +21,11 @@ use concord::{ChannelId, CommunityId, Epoch, GroupKey, decode_hex_32};
 use data_encoding::HEXLOWER;
 use gpui::AsyncApp;
 use nostr_sdk::prelude::*;
-use state::UniversalSigner;
+use state::{BOOTSTRAP_RELAYS, UniversalSigner};
 
 use crate::cache::{self, Observed};
-use crate::history::{CURSOR_OVERLAP, Window};
+use crate::discover::{self, Listing};
+use crate::history::{CURSOR_OVERLAP, Window, within};
 
 /// How much of what a relay stores a cold subscription replays per relay.
 const LIVE_REPLAY: usize = 500;
@@ -274,7 +276,7 @@ where
     S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + ?Sized,
 {
     let link = cord05::parse_link(url)?;
-    let invite = resolve_invite(client, &link).await?;
+    let invite = resolve_bundle(client, &link, &[]).await?;
 
     let material = cord02::list::join_material(&invite, None);
     let self_pk = signer.get_public_key_async().await?;
@@ -299,36 +301,34 @@ where
     Ok(state)
 }
 
-async fn resolve_invite(client: &Client, link: &ParsedInviteLink) -> Result<CommunityInvite> {
-    let relays: Vec<RelayUrl> = link
+pub(crate) async fn resolve_bundle(
+    client: &Client,
+    link: &ParsedInviteLink,
+    fallback: &[RelayUrl],
+) -> Result<CommunityInvite> {
+    let mut relays: Vec<RelayUrl> = link
         .bootstrap_relays
         .iter()
         .filter_map(|relay| RelayUrl::parse(relay).ok())
         .collect();
 
     if relays.is_empty() {
-        anyhow::bail!("invite link names no relays");
+        relays = fallback.to_vec();
     }
 
-    connect_relays(client, &relays).await;
+    if relays.is_empty() {
+        anyhow::bail!("invite link names no relays");
+    }
 
     let filter = Filter::new()
         .kind(Kind::Custom(cord05::KIND_BUNDLE))
         .author(link.link_signer)
         .identifier("");
 
-    let targets: Vec<(RelayUrl, Vec<Filter>)> = relays
-        .into_iter()
-        .map(|relay| (relay, vec![filter.clone()]))
-        .collect();
-
-    let events = client
-        .fetch_events(targets)
-        .timeout(Duration::from_secs(10))
-        .await?;
+    let events = read_events(client, &relays, std::slice::from_ref(&filter)).await?;
 
     let Some(event) = events.into_iter().max_by_key(|event| event.created_at) else {
-        anyhow::bail!("invite link not found on its relays");
+        anyhow::bail!("invite bundle not found on its relays");
     };
 
     let bundle_key = invite_bundle_key(&link.token);
@@ -392,6 +392,325 @@ where
     Ok(merged)
 }
 
+const DISCOVER_LIMIT: usize = 100;
+const ACTIVITY_FILTERS_PER_REQ: usize = 20;
+
+pub(crate) struct ActivityTarget {
+    pub link_signer: PublicKey,
+    pub authors: Vec<PublicKey>,
+    pub relays: Vec<RelayUrl>,
+}
+
+pub(crate) async fn discover_relays(client: &Client, self_pk: Option<PublicKey>) -> Vec<RelayUrl> {
+    let mut relays: BTreeSet<RelayUrl> = BOOTSTRAP_RELAYS
+        .into_iter()
+        .filter_map(|url| RelayUrl::parse(url).ok())
+        .collect();
+
+    if let Some(self_pk) = self_pk {
+        let filter = Filter::new().kind(Kind::RelayList).author(self_pk).limit(1);
+
+        if let Ok(events) = client.database().query(filter).await
+            && let Some(event) = events.into_iter().max_by_key(|event| event.created_at)
+        {
+            for (url, metadata) in nip65::extract_relay_list(&event) {
+                if metadata != Some(RelayMetadata::Write) {
+                    relays.insert(url);
+                }
+            }
+        }
+    }
+
+    relays.into_iter().collect()
+}
+
+pub async fn discover_page(client: &Client, self_pk: Option<PublicKey>) -> Result<Vec<Listing>> {
+    let relays = discover_relays(client, self_pk).await;
+
+    if relays.is_empty() {
+        return Ok(Vec::new());
+    }
+
+    let (announcements, deletes) = read_announcements(client, &relays).await?;
+    let invites = discover::fold_announcements(&announcements, &deletes);
+
+    let mut listings = futures::future::join_all(invites.into_iter().map(|mut listing| {
+        let fallback = relays.clone();
+
+        async move {
+            listing.community = match cord05::parse_link(&listing.invite_url) {
+                Ok(link) => resolve_bundle(client, &link, &fallback).await.ok(),
+                Err(_) => None,
+            };
+
+            listing
+        }
+    }))
+    .await;
+
+    let targets: Vec<ActivityTarget> = listings.iter().filter_map(activity_target).collect();
+
+    if !targets.is_empty() {
+        for (link_signer, at) in probe_activity(client, &targets).await {
+            if let Some(listing) = listings
+                .iter_mut()
+                .find(|listing| listing.link_signer == link_signer)
+            {
+                listing.active_at = Some(at);
+            }
+        }
+    }
+
+    Ok(listings)
+}
+
+fn activity_target(listing: &Listing) -> Option<ActivityTarget> {
+    let community = listing.community.as_ref()?;
+    let authors = activity_authors(community);
+
+    if authors.is_empty() {
+        return None;
+    }
+
+    Some(ActivityTarget {
+        link_signer: listing.link_signer,
+        authors,
+        relays: community
+            .relays
+            .iter()
+            .filter_map(|relay| RelayUrl::parse(relay).ok())
+            .collect(),
+    })
+}
+
+fn activity_authors(invite: &CommunityInvite) -> Vec<PublicKey> {
+    let Ok(root) = decode_hex_32(&invite.community_root) else {
+        return Vec::new();
+    };
+
+    let mut authors = Vec::new();
+
+    if let Ok(group) = guestbook_group_key(&root, &invite.community_id, invite.root_epoch) {
+        authors.push(group.pk());
+    }
+
+    match invite.control_pk {
+        Some(control_pk) => authors.push(control_pk),
+        None => {
+            if let Ok(group) = control_group_key(&root, &invite.community_id, invite.root_epoch) {
+                authors.push(group.pk());
+            }
+        }
+    }
+
+    for channel in &invite.channels {
+        if let Some(key) = &channel.key
+            && let Ok(key) = decode_hex_32(key)
+            && let Ok(group) = channel_group_key(&key, &channel.id, channel.epoch)
+        {
+            authors.push(group.pk());
+        }
+    }
+
+    authors.sort();
+    authors.dedup();
+    authors
+}
+
+async fn read_announcements(
+    client: &Client,
+    relays: &[RelayUrl],
+) -> Result<(Vec<Event>, Vec<Event>)> {
+    let filters = vec![
+        Filter::new()
+            .kind(Kind::Custom(discover::KIND_COMMUNITY_ANNOUNCEMENT))
+            .limit(DISCOVER_LIMIT),
+        Filter::new()
+            .kind(Kind::EventDeletion)
+            .custom_tag(
+                SingleLetterTag::LOWERCASE_K,
+                discover::KIND_COMMUNITY_ANNOUNCEMENT.to_string(),
+            )
+            .limit(DISCOVER_LIMIT),
+    ];
+
+    let events = read_events(client, relays, &filters).await?;
+
+    let mut found = Vec::new();
+    let mut deleted = Vec::new();
+
+    for event in events {
+        if event.kind == Kind::Custom(discover::KIND_COMMUNITY_ANNOUNCEMENT) {
+            found.push(event);
+        } else if event.kind == Kind::EventDeletion {
+            deleted.push(event);
+        }
+    }
+
+    Ok((found, deleted))
+}
+
+async fn probe_activity(
+    client: &Client,
+    targets: &[ActivityTarget],
+) -> Vec<(PublicKey, Timestamp)> {
+    let now = Timestamp::now();
+
+    let mut by_author: BTreeMap<PublicKey, BTreeSet<PublicKey>> = BTreeMap::new();
+    let mut groups: BTreeMap<Vec<RelayUrl>, Vec<&ActivityTarget>> = BTreeMap::new();
+
+    for target in targets {
+        if target.relays.is_empty() {
+            continue;
+        }
+
+        for author in &target.authors {
+            by_author
+                .entry(*author)
+                .or_default()
+                .insert(target.link_signer);
+        }
+
+        groups
+            .entry(target.relays.clone())
+            .or_default()
+            .push(target);
+    }
+
+    let mut newest: BTreeMap<PublicKey, Timestamp> = BTreeMap::new();
+
+    for (relays, group) in groups {
+        for chunk in group.chunks(ACTIVITY_FILTERS_PER_REQ) {
+            let filters: Vec<Filter> = chunk
+                .iter()
+                .map(|target| {
+                    Filter::new()
+                        .kind(Kind::GiftWrap)
+                        .authors(target.authors.iter().copied())
+                        .until(now)
+                        .limit(1)
+                })
+                .collect();
+
+            let events = match read_events(client, &relays, &filters).await {
+                Ok(events) => events,
+                Err(error) => {
+                    log::warn!("discover: activity probe failed: {error}");
+                    continue;
+                }
+            };
+
+            for event in events {
+                if event.created_at > now {
+                    continue;
+                }
+
+                if let Some(signers) = by_author.get(&event.pubkey) {
+                    for signer in signers {
+                        let entry = newest.entry(*signer).or_insert(event.created_at);
+
+                        if event.created_at > *entry {
+                            *entry = event.created_at;
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    newest.into_iter().collect()
+}
+
+async fn pool_targets(
+    client: &Client,
+    relays: &[RelayUrl],
+    filters: &[Filter],
+) -> Vec<(RelayUrl, Vec<Filter>)> {
+    let mut targets = Vec::with_capacity(relays.len());
+
+    for url in relays {
+        if let Ok(Some(_)) = client.relay(url).await {
+            targets.push((url.clone(), filters.to_vec()));
+        }
+    }
+
+    targets
+}
+
+const READ_TIMEOUT: Duration = Duration::from_secs(10);
+
+fn read_subscription() -> SubscriptionId {
+    static NEXT: AtomicU64 = AtomicU64::new(0);
+    SubscriptionId::new(format!("read-{}", NEXT.fetch_add(1, Ordering::Relaxed)))
+}
+
+fn settled_subscription<'m>(message: &'m RelayMessage<'_>) -> Option<&'m SubscriptionId> {
+    match message {
+        RelayMessage::EndOfStoredEvents(id) => Some(id),
+        RelayMessage::Closed {
+            subscription_id, ..
+        } => Some(subscription_id),
+        _ => None,
+    }
+}
+
+async fn read_events(
+    client: &Client,
+    relays: &[RelayUrl],
+    filters: &[Filter],
+) -> Result<Vec<Event>> {
+    connect_relays(client, relays).await;
+
+    let targets = pool_targets(client, relays, filters).await;
+
+    if targets.is_empty() {
+        anyhow::bail!("no relay is reachable");
+    }
+
+    let expected: BTreeSet<RelayUrl> = targets.iter().map(|(url, _)| url.clone()).collect();
+    let id = read_subscription();
+    let mut notifications = client.notifications();
+
+    let options = SubscribeAutoCloseOptions::default()
+        .exit_policy(ReqExitPolicy::ExitOnEOSE)
+        .timeout(Some(READ_TIMEOUT));
+
+    let output = client
+        .subscribe(ReqTarget::manual(targets))
+        .with_id(id.clone())
+        .close_on(options)
+        .await?;
+
+    let deadline = Instant::now() + READ_TIMEOUT;
+    let mut settled: BTreeSet<RelayUrl> = output.failed.into_keys().collect();
+
+    while settled.len() < expected.len() {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+
+        let Some(notification) = within(remaining, notifications.next()).await.flatten() else {
+            break;
+        };
+
+        match notification {
+            ClientNotification::Message { relay_url, message }
+                if settled_subscription(&message) == Some(&id) =>
+            {
+                settled.insert(relay_url);
+            }
+            ClientNotification::Shutdown => break,
+            _ => {}
+        }
+    }
+
+    let mut events = BTreeSet::new();
+
+    for filter in filters {
+        events.extend(client.database().query(filter.clone()).await?);
+    }
+
+    Ok(events.into_iter().collect())
+}
+
 /// Best-effort publication of the genesis wraps to the community's relays.
 pub(crate) async fn publish_wraps(client: &Client, wraps: &[Event], relays: &[RelayUrl]) {
     for wrap in wraps {