Add per-panel ProfileDirectory over batched ProfileStore

crates/signed_state/src/lib.rs

@@ -23,7 +23,7 @@ use gpui::{App, AppContext};
 pub use inbox::Inbox;
 pub use local_repos::{LocalReposStore, ResolvedLocalRepo};
 pub use nostr_sdk::prelude::Timestamp;
-pub use profile::{Profile, ProfileStore};
+pub use profile::{Profile, ProfileDirectory, ProfileEvent, ProfileStore};
 pub use push::{GraspServer, PushOutcome};
 pub use refresh::{RefreshGate, RefreshRequest};
 pub use repo::RepoStore;

crates/signed_state/src/profile.rs

@@ -2,10 +2,9 @@ use std::cell::RefCell;
 use std::collections::{HashMap, HashSet};
 use std::time::{Duration, Instant};
 
-use anyhow::Error;
+use flume::Sender;
-use flume::{Receiver, Sender};
 use gpui::{
-    App, AppContext, AsyncApp, Context, Entity, Global, SharedString, Subscription, WeakEntity,
+    App, AppContext, Context, Entity, EventEmitter, Global, SharedString, Subscription, Task,
 };
 use nostr_sdk::prelude::*;
 use utils::shorten_pubkey;
@@ -15,7 +14,6 @@ use crate::bootstrap::sync_bootstrap_only;
 
 const BATCH_TIMEOUT: Duration = Duration::from_millis(500);
 const REQUEST_CHUNK: usize = 100;
-const WARM_LIMIT: usize = 500;
 
 #[derive(Debug, Clone)]
 pub struct Profile {
@@ -63,12 +61,39 @@ impl Profile {
             .filter(|p| !p.is_empty())
             .map(|p| SharedString::from(p.clone()))
     }
+
+    /// The newest profile per author from metadata events.
+    fn latest_profiles(events: impl IntoIterator<Item = Event>) -> Vec<Self> {
+        let mut latest: HashMap<PublicKey, (Timestamp, Metadata)> = HashMap::new();
+
+        for event in events {
+            match latest.get(&event.pubkey) {
+                Some((created_at, _)) if *created_at >= event.created_at => {}
+                _ => {
+                    let metadata = Metadata::from_json(&event.content).unwrap_or_default();
+                    latest.insert(event.pubkey, (event.created_at, metadata));
+                }
+            }
+        }
+
+        latest
+            .into_iter()
+            .map(|(public_key, (_, metadata))| Profile::new(public_key, metadata))
+            .collect()
+    }
 }
 
+/// Metadata for these authors may have changed in the local database.
+#[derive(Debug, Clone)]
+pub enum ProfileEvent {
+    Updated(Vec<PublicKey>),
+}
+
+/// Batches metadata requests into relay syncs feeding the local database.
 pub struct ProfileStore {
-    profiles: HashMap<PublicKey, Profile>,
+    in_flight: RefCell<HashSet<PublicKey>>,
-    seen: RefCell<HashSet<PublicKey>>,
     sender: Sender<PublicKey>,
+    _task: Task<Result<(), anyhow::Error>>,
     _subscription: Subscription,
 }
 
@@ -76,6 +101,8 @@ struct GlobalProfileStore(Entity<ProfileStore>);
 
 impl Global for GlobalProfileStore {}
 
+impl EventEmitter<ProfileEvent> for ProfileStore {}
+
 impl ProfileStore {
     pub fn global(cx: &App) -> Entity<Self> {
         cx.global::<GlobalProfileStore>().0.clone()
@@ -87,201 +114,256 @@ impl ProfileStore {
 
     pub(crate) fn new(cx: &mut Context<Self>) -> Self {
         let backend = Backend::global(cx);
-        let entity = cx.entity().downgrade();
         let client = backend.read(cx).client();
 
         let (sender, receiver) = flume::unbounded::<PublicKey>();
 
         let subscription = cx.subscribe(&backend, |this, _backend, event, cx| {
             if let BackendEvent::ProfileUpdates(authors) = event {
-                this.apply_authors(authors.clone(), cx);
+                this.emit_updated(authors.clone(), cx);
             }
         });
 
-        cx.spawn(async move |this, cx| {
+        let task = cx.spawn(async move |this, cx| {
-            if let Err(e) = Self::handle_requests(this, &client, &receiver, cx).await {
+            let mut batch: HashSet<PublicKey> = HashSet::new();
-                log::error!("Failed to handle requests: {e}");
-            }
-        })
-        .detach();
 
-        cx.defer(move |cx| {
+            loop {
-            entity.update(cx, |this, cx| this.load(cx)).ok();
+                match receiver.recv_async().await {
+                    Ok(public_key) => {
+                        batch.insert(public_key);
+                    }
+                    Err(_) => return Ok(()),
+                }
+
+                // The channel has no async timeout, race the receive against a timer.
+                let deadline = Instant::now() + BATCH_TIMEOUT;
+
+                loop {
+                    let now = Instant::now();
+
+                    if now >= deadline {
+                        break;
+                    }
+
+                    let timer = cx.background_executor().timer(deadline - now);
+                    futures::pin_mut!(timer);
+
+                    let recv = receiver.recv_async();
+                    futures::pin_mut!(recv);
+
+                    match futures::future::select(recv, timer).await {
+                        futures::future::Either::Left((Ok(public_key), _)) => {
+                            batch.insert(public_key);
+                        }
+                        futures::future::Either::Left((Err(_), _)) => return Ok(()),
+                        futures::future::Either::Right(_) => break,
+                    }
+                }
+
+                let authors: Vec<PublicKey> = batch.drain().collect();
+
+                for chunk in authors.chunks(REQUEST_CHUNK) {
+                    let opts = SyncOptions::default();
+                    let filter = Filter::new()
+                        .kind(Kind::Metadata)
+                        .authors(chunk.iter().copied());
+
+                    sync_bootstrap_only(&client, filter, opts).await.ok();
+                }
+
+                // Emit even when the sync failed, the database may hold newer data.
+                this.update(cx, |this, cx| {
+                    this.emit_updated(authors, cx);
+                })?;
+            }
         });
 
         Self {
-            profiles: HashMap::new(),
+            in_flight: RefCell::new(HashSet::new()),
-            seen: RefCell::new(HashSet::new()),
             sender,
+            _task: task,
             _subscription: subscription,
         }
     }
 
-    // Returns a placeholder until fetched; queues a fetch when uncached.
+    /// Queue a batched relay sync for keys without one in flight.
-    pub fn get(&self, public_key: &PublicKey) -> Profile {
+    pub fn fetch(&self, keys: impl IntoIterator<Item = PublicKey>) {
-        if let Some(profile) = self.profiles.get(public_key) {
+        let mut queue = Vec::new();
-            return profile.clone();
-        }
-
-        let public_key = *public_key;
-
-        if self.seen.borrow_mut().insert(public_key)
-            && let Err(e) = self.sender.send(public_key)
         {
-            log::warn!("failed to queue profile fetch: {e}");
+            let mut in_flight = self.in_flight.borrow_mut();
-        }
 
-        Profile::new(public_key, Metadata::default())
+            for key in keys {
-    }
+                if in_flight.insert(key) {
-
+                    queue.push(key);
-    fn load(&mut self, cx: &mut Context<Self>) {
-        let backend = Backend::global(cx);
-        let client = backend.read(cx).client();
-
-        let work = cx.background_spawn(async move {
-            let filter = Filter::new().kind(Kind::Metadata).limit(WARM_LIMIT);
-            let events = client.database().query(filter).await?;
-
-            let profiles: Vec<Profile> = events
-                .into_iter()
-                .map(|event| {
-                    let metadata = Metadata::from_json(&event.content).unwrap_or_default();
-                    Profile::new(event.pubkey, metadata)
-                })
-                .collect();
-
-            Ok::<_, Error>(profiles)
-        });
-
-        cx.spawn(async move |this, cx| {
-            let profiles = work.await?;
-
-            this.update(cx, |this, cx| {
-                for profile in profiles {
-                    this.profiles.insert(profile.public_key(), profile);
                 }
-                cx.notify();
+            }
-            })?;
+        }
-
+        for public_key in queue {
-            Ok::<_, Error>(())
+            if let Err(e) = self.sender.send(public_key) {
-        })
+                log::warn!("failed to queue profile fetch: {e}");
-        .detach();
+            }
+        }
     }
 
-    fn apply_authors(&mut self, authors: Vec<PublicKey>, cx: &mut Context<Self>) {
+    // Releases the in-flight keys so future fetches can be queued again.
+    fn emit_updated(&mut self, authors: Vec<PublicKey>, cx: &mut Context<Self>) {
         if authors.is_empty() {
             return;
         }
 
-        let backend = Backend::global(cx);
+        {
-        let client = backend.read(cx).client();
+            let mut in_flight = self.in_flight.borrow_mut();
-
+            for author in &authors {
-        let work = cx.background_spawn(async move {
+                in_flight.remove(author);
-            let filter = Filter::new().kind(Kind::Metadata).authors(authors);
-            let events = client.database().query(filter).await?;
-            // Pick the latest metadata per author off the main thread.
-            let mut latest: HashMap<PublicKey, (Timestamp, Metadata)> = HashMap::new();
-
-            for event in events {
-                match latest.get(&event.pubkey) {
-                    Some((ts, _)) if *ts >= event.created_at => {}
-                    _ => {
-                        latest.insert(
-                            event.pubkey,
-                            (
-                                event.created_at,
-                                Metadata::from_json(&event.content).unwrap_or_default(),
-                            ),
-                        );
-                    }
-                }
-            }
-
-            let profiles: Vec<Profile> = latest
-                .into_iter()
-                .map(|(public_key, (_, metadata))| Profile::new(public_key, metadata))
-                .collect();
-
-            Ok::<_, Error>(profiles)
-        });
-
-        cx.spawn(async move |this, cx| {
-            let profiles = work.await?;
-
-            this.update(cx, |this, cx| {
-                for profile in profiles {
-                    this.profiles.insert(profile.public_key(), profile);
-                }
-                cx.notify();
-            })?;
-
-            Ok::<_, Error>(())
-        })
-        .detach();
-    }
-
-    fn apply_seen(&mut self, cx: &mut Context<Self>) {
-        let authors: Vec<PublicKey> = self.seen.borrow().iter().copied().collect();
-        self.apply_authors(authors, cx);
-    }
-
-    async fn handle_requests(
-        this: WeakEntity<ProfileStore>,
-        client: &Client,
-        receiver: &Receiver<PublicKey>,
-        cx: &mut AsyncApp,
-    ) -> Result<(), Error> {
-        let mut batch: HashSet<PublicKey> = HashSet::new();
-
-        loop {
-            match receiver.recv_async().await {
-                Ok(public_key) => {
-                    batch.insert(public_key);
-                }
-                Err(_) => return Ok(()),
-            }
-
-            // The channel has no async timeout, race the receive against a timer.
-            let deadline = Instant::now() + BATCH_TIMEOUT;
-
-            loop {
-                let now = Instant::now();
-
-                if now >= deadline {
-                    break;
-                }
-
-                let timer = cx.background_executor().timer(deadline - now);
-                futures::pin_mut!(timer);
-
-                let recv = receiver.recv_async();
-                futures::pin_mut!(recv);
-
-                match futures::future::select(recv, timer).await {
-                    futures::future::Either::Left((Ok(public_key), _)) => {
-                        batch.insert(public_key);
-                    }
-                    futures::future::Either::Left((Err(_), _)) => return Ok(()),
-                    futures::future::Either::Right(_) => break,
-                }
-            }
-
-            let authors: Vec<PublicKey> = batch.drain().collect();
-
-            for chunk in authors.chunks(REQUEST_CHUNK) {
-                let opts = SyncOptions::default();
-                let filter = Filter::new()
-                    .kind(Kind::Metadata)
-                    .authors(chunk.iter().copied());
-
-                if let Err(e) = sync_bootstrap_only(client, filter, opts).await {
-                    log::warn!("profile fetch failed: {e}");
-                }
-            }
-
-            if this.update(cx, |this, cx| this.apply_seen(cx)).is_err() {
-                return Ok(());
             }
         }
+
+        cx.emit(ProfileEvent::Updated(authors));
+    }
+}
+
+/// Profile cache bound to one panel or screen.
+pub struct ProfileDirectory {
+    profiles: HashMap<PublicKey, Profile>,
+    pending: RefCell<HashSet<PublicKey>>,
+    tasks: Vec<Task<()>>,
+    _subscriptions: Vec<Subscription>,
+}
+
+impl ProfileDirectory {
+    pub fn new(cx: &mut App) -> Entity<Self> {
+        let store = ProfileStore::global(cx);
+
+        cx.new(|cx| {
+            let subscription =
+                cx.subscribe(&store, |this: &mut Self, _store, event, cx| match event {
+                    ProfileEvent::Updated(authors) => this.refresh(authors, cx),
+                });
+
+            Self {
+                profiles: HashMap::new(),
+                pending: RefCell::new(HashSet::new()),
+                tasks: Vec::new(),
+                _subscriptions: vec![subscription],
+            }
+        })
+    }
+
+    /// Resolve keys from the local database, queue the rest for a relay sync.
+    pub fn track(&mut self, keys: impl IntoIterator<Item = PublicKey>, cx: &mut Context<Self>) {
+        let missing: Vec<PublicKey> = keys
+            .into_iter()
+            .filter(|key| {
+                !self.profiles.contains_key(key)
+                    && !self.with_pending(|pending| pending.contains(key))
+            })
+            .collect();
+
+        if missing.is_empty() {
+            return;
+        }
+
+        self.load(missing, true, cx);
+    }
+
+    /// The profile for a key, or a placeholder that queues a relay sync.
+    pub fn profile(&self, public_key: &PublicKey, cx: &App) -> Profile {
+        if let Some(profile) = self.profiles.get(public_key) {
+            return profile.clone();
+        }
+
+        if self.with_pending(|pending| pending.insert(*public_key)) {
+            let store = ProfileStore::global(cx);
+            store.read(cx).fetch([*public_key]);
+        }
+
+        Profile::new(*public_key, Metadata::default())
+    }
+
+    /// Refreshes profiles for keys that are already tracked.
+    fn refresh(&mut self, authors: &[PublicKey], cx: &mut Context<Self>) {
+        let tracked: Vec<PublicKey> = authors
+            .iter()
+            .copied()
+            .filter(|key| {
+                self.profiles.contains_key(key)
+                    || self.with_pending(|pending| pending.contains(key))
+            })
+            .collect();
+
+        if tracked.is_empty() {
+            return;
+        }
+
+        self.load(tracked, false, cx);
+    }
+
+    /// Loads profiles from the local database, queueing a relay sync for any missing keys.
+    fn load(&mut self, keys: Vec<PublicKey>, queue_sync: bool, cx: &mut Context<Self>) {
+        // Accept events that arrive while the database query is running.
+        self.with_pending(|pending| {
+            pending.extend(keys.iter().copied());
+        });
+
+        let backend = Backend::global(cx);
+        let client = backend.read(cx).client();
+        let query_keys = keys.clone();
+
+        let work = cx.background_spawn(async move {
+            let authors = query_keys.iter().copied();
+            let filter = Filter::new().kind(Kind::Metadata).authors(authors);
+            let events = client.database().query(filter).await?;
+
+            Ok::<_, anyhow::Error>(Profile::latest_profiles(events))
+        });
+
+        self.tasks.retain(|task| !task.is_ready());
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let loaded = work.await;
+
+            this.update(cx, |this, cx| {
+                let remaining = match loaded {
+                    Ok(profiles) => {
+                        let found: HashSet<PublicKey> =
+                            profiles.iter().map(Profile::public_key).collect();
+                        this.insert_loaded(profiles, cx);
+                        keys.into_iter()
+                            .filter(|key| !found.contains(key))
+                            .collect()
+                    }
+                    Err(e) => {
+                        log::warn!("failed to load profiles from the database: {e}");
+                        keys
+                    }
+                };
+
+                if queue_sync && !remaining.is_empty() {
+                    let store = ProfileStore::global(cx);
+                    store.read(cx).fetch(remaining);
+                }
+            })
+            .ok();
+        }));
+    }
+
+    // Inserts database results, replacing anything loaded earlier.
+    fn insert_loaded(&mut self, profiles: Vec<Profile>, cx: &mut Context<Self>) {
+        let mut inserted = false;
+
+        for profile in profiles {
+            let public_key = profile.public_key();
+            self.profiles.insert(public_key, profile);
+            self.with_pending(|pending| pending.remove(&public_key));
+            inserted = true;
+        }
+
+        if inserted {
+            cx.notify();
+        }
+    }
+
+    fn with_pending<R>(&self, then: impl FnOnce(&mut HashSet<PublicKey>) -> R) -> R {
+        let mut pending = self.pending.borrow_mut();
+        then(&mut pending)
     }
 }