add contact-based community suggestions

crates/nip29/src/lib.rs

@@ -20,8 +20,10 @@ 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 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);
@@ -41,6 +43,8 @@ pub enum GroupsEvent {
     Error(String),
     Browsed(Box<GroupCandidate>),
     BrowseSettled(RelayUrl),
+    Suggested(Box<GroupCandidate>),
+    SuggestSettled,
 }
 
 #[derive(Debug)]
@@ -49,6 +53,8 @@ enum Signal {
     Eose(SubscriptionId),
     Closed(SubscriptionId, String),
     BrowseSettled(SubscriptionId),
+    ContactListsSettled,
+    ContactPreviewSettled(SubscriptionId),
     Forks,
 }
 
@@ -60,6 +66,10 @@ fn fork_subscription_id() -> SubscriptionId {
     SubscriptionId::new(FORK_SUBSCRIPTION)
 }
 
+fn contacts_subscription_id() -> SubscriptionId {
+    SubscriptionId::new(CONTACTS_SUBSCRIPTION)
+}
+
 fn subscription_id(key: &GroupKey) -> SubscriptionId {
     SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32))
 }
@@ -71,6 +81,13 @@ 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))
 }
@@ -83,8 +100,10 @@ fn is_own_subscription(id: &SubscriptionId) -> bool {
     let id = id.as_str();
     id == LIST_SUBSCRIPTION
         || id == FORK_SUBSCRIPTION
+        || id == CONTACTS_SUBSCRIPTION
         || id.starts_with(GROUP_PREFIX)
         || id.starts_with(BROWSE_PREFIX)
+        || id.starts_with(PREVIEW_PREFIX)
         || id.starts_with(PAGE_PREFIX)
 }
 
@@ -98,6 +117,16 @@ 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,
     signal_tx: flume::Sender<Signal>,
     signal_rx: flume::Receiver<Signal>,
     notification_listener: Option<Task<()>>,
@@ -152,6 +181,11 @@ 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,
             signal_tx: tx,
             signal_rx: rx,
             notification_listener: None,
@@ -280,6 +314,465 @@ impl GroupsRegistry {
         cx.emit(GroupsEvent::BrowseSettled(relay));
     }
 
+    pub fn load_contacts_groups(&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() {
+            return;
+        }
+
+        let task: Task<Result<HashSet<PublicKey>, 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);
+
+            let contacts = client
+                .database()
+                .query(filter)
+                .await?
+                .into_iter()
+                .next()
+                .map(|event| event.tags.public_keys().collect())
+                .unwrap_or_default();
+
+            Ok(contacts)
+        });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let authors = match task.await {
+                Ok(authors) => authors,
+                Err(error) => {
+                    log::warn!("nip29: loading the contact list failed: {error}");
+                    return;
+                }
+            };
+
+            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| {
+                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_contact_list(&event, cx);
+                }
+
+                this.contact_refs.clone()
+            }) {
+                Ok(refs) => refs,
+                Err(error) => {
+                    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)) {
+                Ok(task) => match task.await {
+                    Ok(previews) => previews,
+                    Err(error) => {
+                        log::warn!("nip29: loading the cached group metadata failed: {error}");
+                        Vec::new()
+                    }
+                },
+                Err(error) => {
+                    log::warn!("nip29: hydrating the contacts failed: {error}");
+                    return;
+                }
+            };
+
+            if let Err(error) = this.update(cx, |this, cx| {
+                this.absorb_cached_previews(previews, cx);
+                this.subscribe_contacts(cx);
+            }) {
+                log::warn!("nip29: applying the cached group metadata 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>> {
+        if refs.is_empty() {
+            return Task::ready(Ok(Vec::new()));
+        }
+
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+
+        let ids: HashSet<&str> = refs.values().flatten().map(GroupId::as_str).collect();
+        let filter = Filter::new().kind(Kind::GroupMetadata).identifiers(ids);
+
+        cx.background_spawn(async move {
+            Ok(client.database().query(filter).await?.into_iter().collect())
+        })
+    }
+
+    /// Seed the suggestions 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()
+                .fold(HashMap::new(), |mut newest, event| {
+                    let metadata = match GroupMetadata::parse(&event) {
+                        Ok(metadata) => metadata,
+                        Err(error) => {
+                            log::warn!("nip29: ignoring malformed cached group metadata: {error}");
+                            return newest;
+                        }
+                    };
+
+                    let Some(id) = metadata.id().cloned() else {
+                        return newest;
+                    };
+
+                    match newest.get(&id) {
+                        Some((created_at, _)) if *created_at >= event.created_at => newest,
+                        _ => {
+                            newest.insert(id, (event.created_at, metadata));
+                            newest
+                        }
+                    }
+                });
+
+        for (relay, ids) in self.contact_refs.clone() {
+            for id in ids {
+                let Some((_, metadata)) = newest.get(&id) else {
+                    continue;
+                };
+
+                // Hidden groups are not discoverable (NIP-29).
+                if metadata.hidden {
+                    continue;
+                }
+
+                self.upsert_suggestion(
+                    GroupCandidate {
+                        key: GroupKey::new(relay.clone(), id.clone()),
+                        metadata: metadata.clone(),
+                    },
+                    cx,
+                );
+            }
+        }
+    }
+
+    /// Subscribe to the contacts' public group lists.
+    fn subscribe_contacts(&mut self, cx: &mut Context<Self>) {
+        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 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(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();
+    }
+
+    /// Ask every relay hosting referenced groups for the groups' metadata.
+    fn sync_contact_previews(&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() {
+            // A subscription is already asking this relay for previews.
+            if self.previews.values().any(|in_flight| in_flight == &relay) {
+                continue;
+            }
+
+            let id = preview_subscription(&relay);
+            self.previews.insert(id.clone(), relay.clone());
+            let client = client.clone();
+
+            let filter = Filter::new()
+                .kind(Kind::GroupMetadata)
+                .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(), [filter]))
+                    .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}");
+                }
+            }));
+        }
+
+        cx.notify();
+    }
+
+    fn absorb_contact_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 {}",
+                event.pubkey
+            );
+            return;
+        };
+
+        let joined: HashSet<(RelayUrl, GroupId)> = self
+            .groups
+            .iter()
+            .map(|group| {
+                let key = group.read(cx).key();
+                (key.relay().clone(), key.id().clone())
+            })
+            .collect();
+
+        let mut added = false;
+
+        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 {
+            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 {
+            return;
+        };
+
+        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;
+        }
+
+        // A relay may answer with more than the filter asked for.
+        let Some(group) = metadata.id().cloned() else {
+            return;
+        };
+
+        if !self
+            .contact_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())
+        });
+
+        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_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();
+        }
+    }
+
     pub fn load_more(&mut self, key: &GroupKey, cx: &mut Context<Self>) {
         let Some(group) = self.group(key, cx) else {
             return;
@@ -337,8 +830,14 @@ impl GroupsRegistry {
         ids.push(list_subscription_id());
         ids.push(fork_subscription_id());
         ids.extend(self.browse.keys().cloned());
+        ids.extend(self.previews.keys().cloned());
 
         self.browse.clear();
+        self.previews.clear();
+        self.contact_authors.clear();
+        self.contact_refs.clear();
+        self.suggestions.clear();
+        self.contact_settled = false;
 
         self.tasks.push(cx.background_spawn(async move {
             for id in ids {
@@ -817,11 +1316,21 @@ impl GroupsRegistry {
             return;
         }
 
+        if id == contacts_subscription_id() {
+            self.absorb_contact_list(&event, cx);
+            return;
+        }
+
         if self.browse.contains_key(&id) {
             self.absorb_candidate(&id, &event, cx);
             return;
         }
 
+        if self.previews.contains_key(&id) {
+            self.absorb_contact_group(&id, &event, cx);
+            return;
+        }
+
         let Some(key) = self.routes.get(&id).cloned() else {
             return;
         };
@@ -858,6 +1367,16 @@ impl GroupsRegistry {
     }
 
     fn handle_eose(&mut self, id: SubscriptionId, cx: &mut Context<Self>) {
+        if id == contacts_subscription_id() {
+            self.handle_contact_lists_settled(cx);
+            return;
+        }
+
+        if self.previews.contains_key(&id) {
+            self.handle_contact_preview_settled(&id, cx);
+            return;
+        }
+
         let Some(key) = self.routes.get(&id).cloned() else {
             return;
         };
@@ -886,6 +1405,18 @@ 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);
+            return;
+        }
+
+        if let Some(relay) = self.previews.get(&id).cloned() {
+            log::warn!("nip29: fetching previews from {relay} was refused: {reason}");
+            self.handle_contact_preview_settled(&id, cx);
+            return;
+        }
+
         if let Some(relay) = self.browse.get(&id) {
             log::warn!("nip29: browsing {relay} was refused: {reason}");
             return;
@@ -1000,6 +1531,12 @@ impl GroupsRegistry {
                     Signal::BrowseSettled(id) => {
                         this.settle_browse(id, cx);
                     }
+                    Signal::ContactListsSettled => {
+                        this.handle_contact_lists_settled(cx);
+                    }
+                    Signal::ContactPreviewSettled(id) => {
+                        this.handle_contact_preview_settled(&id, cx);
+                    }
                     Signal::Forks => {
                         this.sync_forks(true, cx);
                     }