From 79825d18257f04705abebc52c5af9af5998c0e50 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Mon, 12 Oct 2026 16:13:53 +0700 Subject: [PATCH] add person change events and batched metadata fetching --- crates/person/src/directory.rs | 167 +++++++++++++++++++++++++++++++++ crates/person/src/lib.rs | 85 +++++++++++++++-- 2 files changed, 242 insertions(+), 10 deletions(-) create mode 100644 crates/person/src/directory.rs diff --git a/crates/person/src/directory.rs b/crates/person/src/directory.rs new file mode 100644 index 00000000..fbe7314d --- /dev/null +++ b/crates/person/src/directory.rs @@ -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, + /// Keys awaiting their first update, guarded for render-time access + pending: RwLock>, + 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 { + let registry = PersonRegistry::global(cx); + + cx.new(|cx| { + let subscription = cx.subscribe(®istry, |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, cx: &mut Context) { + let missing: Vec = 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, Error>> = cx.background_spawn(async move { + let filter = Filter::new().kind(Kind::Metadata).authors(authors); + let mut events: Vec = + 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 = + 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) { + 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, cx: &mut Context) { + 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(&self, then: impl FnOnce(&mut HashSet) -> R) -> R { + let mut pending = match self.pending.write() { + Ok(pending) => pending, + Err(poisoned) => poisoned.into_inner(), + }; + + then(&mut pending) + } +} diff --git a/crates/person/src/lib.rs b/crates/person/src/lib.rs index 91e035e2..1a82e9cf 100644 --- a/crates/person/src/lib.rs +++ b/crates/person/src/lib.rs @@ -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); impl Global for GlobalPersonRegistry {} +/// A change to a person's data +#[derive(Debug, Clone)] +pub enum PersonEvent { + Updated(Person), +} + +impl EventEmitter for PersonRegistry {} + #[derive(Debug, Clone)] enum Dispatch { Person(Person), @@ -37,6 +47,8 @@ pub struct PersonRegistry { persons: HashMap>, /// Set of public keys that have been seen seen: RwLock>, + /// Public keys with a metadata request in flight + in_flight: RwLock>, sender: flume::Sender, 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) { 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) { let urls: Vec = 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.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, + ) { 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, cx: &mut Context) { - 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) { 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(&self, keys: I) + where + I: IntoIterator, + { + 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);