add person change events and batched metadata fetching

2026-10-12 16:13:53 +07:00
parent b711a99192
commit 79825d1825
2 changed files with 242 additions and 10 deletions
@@ -0,0 +1,167 @@
+use std::cmp::Reverse;
+use std::collections::{HashMap, HashSet};
+use std::sync::RwLock;
+
+use anyhow::Error;
+use gpui::{App, AppContext, Context, Entity, Subscription, Task};
+use nostr_sdk::prelude::*;
+use smallvec::{SmallVec, smallvec};
+use state::NostrRegistry;
+
+use crate::{Person, PersonEvent, PersonRegistry};
+
+/// Directory of persons tracked by a view
+#[derive(Debug)]
+pub struct Directory {
+    persons: HashMap<PublicKey, Person>,
+    /// Keys awaiting their first update, guarded for render-time access
+    pending: RwLock<HashSet<PublicKey>>,
+    tasks: SmallVec<[Task<()>; 2]>,
+    _subscriptions: SmallVec<[Subscription; 1]>,
+}
+
+impl Directory {
+    /// Create a directory following the registry's events
+    pub fn new(cx: &mut App) -> Entity<Self> {
+        let registry = PersonRegistry::global(cx);
+
+        cx.new(|cx| {
+            let subscription = cx.subscribe(&registry, |this: &mut Self, _registry, event, cx| {
+                this.handle_event(event.clone(), cx);
+            });
+
+            Self {
+                persons: HashMap::new(),
+                pending: RwLock::new(HashSet::new()),
+                tasks: SmallVec::new(),
+                _subscriptions: smallvec![subscription],
+            }
+        })
+    }
+
+    /// Track keys: load known profiles from the database, request the rest
+    pub fn track(&mut self, keys: impl IntoIterator<Item = PublicKey>, cx: &mut Context<Self>) {
+        let missing: Vec<PublicKey> = keys
+            .into_iter()
+            .filter(|key| {
+                !self.persons.contains_key(key)
+                    && !self.with_pending(|pending| pending.contains(key))
+            })
+            .collect();
+
+        if missing.is_empty() {
+            return;
+        }
+
+        // Accept events that arrive while the database query is running
+        self.with_pending(|pending| pending.extend(missing.iter().copied()));
+
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+        let authors = missing.clone();
+
+        let task: Task<Result<Vec<Person>, Error>> = cx.background_spawn(async move {
+            let filter = Filter::new().kind(Kind::Metadata).authors(authors);
+            let mut events: Vec<Event> =
+                client.database().query(filter).await?.into_iter().collect();
+
+            // Keep only the newest event per author
+            events.sort_by_key(|event| Reverse(event.created_at));
+            let mut seen = HashSet::new();
+
+            Ok(events
+                .into_iter()
+                .filter(|event| seen.insert(event.pubkey))
+                .map(|event| {
+                    let metadata = Metadata::from_json(event.content).unwrap_or_default();
+                    Person::new(event.pubkey, metadata)
+                })
+                .collect())
+        });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let loaded = task.await;
+
+            this.update(cx, |this, cx| {
+                let remaining = match loaded {
+                    Ok(persons) => {
+                        let found: HashSet<PublicKey> =
+                            persons.iter().map(Person::public_key).collect();
+                        this.bulk_insert(persons, cx);
+                        missing
+                            .into_iter()
+                            .filter(|key| !found.contains(key))
+                            .collect()
+                    }
+                    Err(error) => {
+                        log::warn!("Failed to load persons from the local database: {error}");
+                        missing
+                    }
+                };
+
+                if !remaining.is_empty() {
+                    let registry = PersonRegistry::global(cx);
+                    registry.read(cx).fetch(remaining);
+                }
+            })
+            .ok();
+        }));
+    }
+
+    /// The person for a key, or a placeholder that queues a request
+    pub fn person(&self, public_key: &PublicKey, cx: &App) -> Person {
+        if let Some(person) = self.persons.get(public_key) {
+            return person.clone();
+        }
+
+        if self.with_pending(|pending| pending.insert(*public_key)) {
+            let registry = PersonRegistry::global(cx);
+            registry.read(cx).fetch([*public_key]);
+        }
+
+        Person::new(*public_key, Metadata::default())
+    }
+
+    /// Accept events for tracked keys and drop the rest
+    fn handle_event(&mut self, event: PersonEvent, cx: &mut Context<Self>) {
+        let PersonEvent::Updated(person) = event;
+        let public_key = person.public_key();
+
+        if self.persons.contains_key(&public_key)
+            || self.with_pending(|pending| pending.remove(&public_key))
+        {
+            self.persons.insert(public_key, person);
+            cx.notify();
+        }
+    }
+
+    /// Insert loaded persons, keeping data already received from relays
+    fn bulk_insert(&mut self, persons: Vec<Person>, cx: &mut Context<Self>) {
+        let mut inserted = false;
+
+        for person in persons {
+            let public_key = person.public_key();
+
+            if self.persons.contains_key(&public_key) {
+                continue;
+            }
+
+            self.persons.insert(public_key, person);
+            inserted = true;
+        }
+
+        if inserted {
+            cx.notify();
+        }
+    }
+
+    /// Run `then` with the pending set, tolerating poisoning: it is only a dedupe cache
+    fn with_pending<R>(&self, then: impl FnOnce(&mut HashSet<PublicKey>) -> R) -> R {
+        let mut pending = match self.pending.write() {
+            Ok(pending) => pending,
+            Err(poisoned) => poisoned.into_inner(),
+        };
+
+        then(&mut pending)
+    }
+}
@@ -4,14 +4,16 @@ use std::sync::RwLock;

 use anyhow::{Error, anyhow};
 use common::{EventExt, emojis_from_tags};
-use gpui::{App, AppContext, Context, Entity, Global, Subscription, Task};
+use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task};
 use instant::Duration;
 use nostr_sdk::prelude::*;
 use smallvec::{SmallVec, smallvec};
 use state::{Announcement, BOOTSTRAP_RELAYS, NostrRegistry, TIMEOUT};

+mod directory;
 mod person;

+pub use directory::*;
 pub use person::*;

 pub fn init(cx: &mut App) {
@@ -22,6 +24,14 @@ struct GlobalPersonRegistry(Entity<PersonRegistry>);

 impl Global for GlobalPersonRegistry {}

+/// A change to a person's data
+#[derive(Debug, Clone)]
+pub enum PersonEvent {
+    Updated(Person),
+}
+
+impl EventEmitter<PersonEvent> for PersonRegistry {}
+
 #[derive(Debug, Clone)]
 enum Dispatch {
    Person(Person),
@@ -37,6 +47,8 @@ pub struct PersonRegistry {
    persons: HashMap<PublicKey, Entity<Person>>,
    /// Set of public keys that have been seen
    seen: RwLock<HashSet<PublicKey>>,
+    /// Public keys with a metadata request in flight
+    in_flight: RwLock<HashSet<PublicKey>>,
    sender: flume::Sender<PublicKey>,
    tasks: SmallVec<[Task<()>; 4]>,
    _subscriptions: SmallVec<[Subscription; 1]>,
@@ -114,6 +126,7 @@ impl PersonRegistry {
        Self {
            persons: HashMap::new(),
            seen: RwLock::new(HashSet::new()),
+            in_flight: RwLock::new(HashSet::new()),
            sender: metadata_tx,
            tasks,
            _subscriptions: smallvec![subscription],
@@ -283,7 +296,7 @@ impl PersonRegistry {
    }

    /// Set profile encryption keys announcement
-    fn set_announcement(&mut self, event: &Event, cx: &mut App) {
+    fn set_announcement(&mut self, event: &Event, cx: &mut Context<Self>) {
        let Some(announcement) = Announcement::from_event(event) else {
            log::warn!("Ignoring announcement with missing or invalid key tag");
            return;
@@ -294,6 +307,7 @@ impl PersonRegistry {
                person.set_announcement(announcement);
                cx.notify();
            });
+            cx.emit(PersonEvent::Updated(person.read(cx).clone()));
        } else {
            let person =
                Person::new(event.pubkey, Metadata::default()).with_announcement(announcement);
@@ -302,7 +316,7 @@ impl PersonRegistry {
    }

    /// Set messaging relays for a person
-    fn set_messaging_relays(&mut self, event: &Event, cx: &mut App) {
+    fn set_messaging_relays(&mut self, event: &Event, cx: &mut Context<Self>) {
        let urls: Vec<RelayUrl> = nip17::extract_relay_list(event).collect();

        if let Some(person) = self.persons.get(&event.pubkey) {
@@ -310,6 +324,7 @@ impl PersonRegistry {
                person.set_messaging_relays(urls);
                cx.notify();
            });
+            cx.emit(PersonEvent::Updated(person.read(cx).clone()));
        } else {
            let person = Person::new(event.pubkey, Metadata::default()).with_messaging_relays(urls);
            self.insert(person, cx);
@@ -317,17 +332,23 @@ impl PersonRegistry {
    }

    /// Set custom emoji for a person from their emoji list event
-    fn set_emojis(&mut self, event: &Event, cx: &mut App) {
+    fn set_emojis(&mut self, event: &Event, cx: &mut Context<Self>) {
        self.set_emojis_for(event.pubkey, emojis_from_tags(&event.tags), cx);
    }

    /// Set custom emoji for a person
-    fn set_emojis_for(&mut self, public_key: PublicKey, emojis: Vec<(String, Url)>, cx: &mut App) {
+    fn set_emojis_for(
+        &mut self,
+        public_key: PublicKey,
+        emojis: Vec<(String, Url)>,
+        cx: &mut Context<Self>,
+    ) {
        if let Some(person) = self.persons.get(&public_key) {
            person.update(cx, |person, cx| {
                person.set_emojis(emojis);
                cx.notify();
            });
+            cx.emit(PersonEvent::Updated(person.read(cx).clone()));
        } else {
            let person = Person::new(public_key, Metadata::default()).with_emojis(emojis);
            self.insert(person, cx);
@@ -336,18 +357,23 @@ impl PersonRegistry {

    /// Insert batch of persons
    fn bulk_insert(&mut self, persons: Vec<Person>, cx: &mut Context<Self>) {
-        for person in persons.into_iter() {
+        for person in persons {
            let public_key = person.public_key();
-            self.persons
-                .entry(public_key)
-                .or_insert_with(|| cx.new(|_| person));
+
+            if self.persons.contains_key(&public_key) {
+                continue;
+            }
+
+            cx.emit(PersonEvent::Updated(person.clone()));
+            self.persons.insert(public_key, cx.new(|_| person));
        }
        cx.notify();
    }

    /// Insert or update a person
-    pub fn insert(&mut self, person: Person, cx: &mut App) {
+    pub fn insert(&mut self, person: Person, cx: &mut Context<Self>) {
        let public_key = person.public_key();
+        self.release_in_flight(&public_key);

        match self.persons.get(&public_key) {
            Some(this) => {
@@ -355,9 +381,13 @@ impl PersonRegistry {
                    this.set_metadata(person.metadata());
                    cx.notify();
                });
+                cx.emit(PersonEvent::Updated(this.read(cx).clone()));
            }
            None => {
+                cx.emit(PersonEvent::Updated(person.clone()));
                self.persons.insert(public_key, cx.new(|_| person));
+                // Views rendering a placeholder only observed this registry
+                cx.notify();
            }
        }
    }
@@ -386,6 +416,41 @@ impl PersonRegistry {
        Person::new(public_key, Metadata::default())
    }

+    /// Queue a batched metadata request for keys without one in flight
+    pub(crate) fn fetch<I>(&self, keys: I)
+    where
+        I: IntoIterator<Item = PublicKey>,
+    {
+        let mut queue = Vec::new();
+        {
+            let mut in_flight = match self.in_flight.write() {
+                Ok(in_flight) => in_flight,
+                Err(poisoned) => poisoned.into_inner(),
+            };
+
+            for key in keys {
+                if in_flight.insert(key) {
+                    queue.push(key);
+                }
+            }
+        }
+
+        for public_key in queue {
+            if let Err(error) = self.sender.send(public_key) {
+                log::warn!("Failed to queue metadata request: {error}");
+            }
+        }
+    }
+
+    /// Allow future fetches for a key once its metadata has arrived
+    fn release_in_flight(&self, public_key: &PublicKey) {
+        let mut in_flight = match self.in_flight.write() {
+            Ok(in_flight) => in_flight,
+            Err(poisoned) => poisoned.into_inner(),
+        };
+        in_flight.remove(public_key);
+    }
+
    /// The signed-in user's custom emoji
    pub fn current_emojis(cx: &App) -> Vec<(String, Url)> {
        let registry = Self::global(cx);