update sidebar

crates/signed_state/src/profile.rs

@@ -1,7 +1,9 @@
 use std::collections::{HashMap, HashSet};
-use std::time::Duration;
+use std::sync::RwLock;
+use std::time::{Duration, Instant};
 
 use anyhow::Error;
+use flume::{Receiver, RecvTimeoutError, Sender};
 use gpui::{App, AppContext, Context, Entity, Global, SharedString, Subscription, Task};
 use nostr_sdk::prelude::*;
 
@@ -63,15 +65,23 @@ pub fn shorten_pubkey(public_key: PublicKey, len: usize) -> String {
     format!("{}...{}", &npub[..(len + 5)], &npub[npub.len() - len..])
 }
 
+/// Message from the fetch task to the main thread.
+enum Dispatch {
+    /// A batched sync finished; re-read seen profiles from the database.
+    Synced,
+}
+
+/// How long to wait for more requests before firing a batched sync.
+const BATCH_TIMEOUT: Duration = Duration::from_millis(500);
+
 /// Global profile cache. Profiles are fetched in batches and kept as plain
 /// data; the whole store notifies on change.
 pub struct ProfileStore {
     profiles: HashMap<PublicKey, Profile>,
     /// Public keys we've already requested this session.
-    seen: HashSet<PublicKey>,
-    /// Public keys queued for the next batched fetch.
-    queued: HashSet<PublicKey>,
-    fetching: bool,
+    seen: RwLock<HashSet<PublicKey>>,
+    /// Sender for queuing fetch requests, batched by a background task.
+    sender: Sender<PublicKey>,
     tasks: Vec<Task<Result<(), Error>>>,
     _subscription: Subscription,
 }
@@ -106,12 +116,31 @@ impl ProfileStore {
             _ => {}
         });
 
+        // Fetch requests are queued on a channel and synced in batches by a
+        // background task.
+        let client = backend.read(cx).client();
+        let (sender, receiver) = flume::unbounded::<PublicKey>();
+        let (dispatch_tx, dispatch_rx) = flume::unbounded::<Dispatch>();
+
+        let mut tasks = Vec::new();
+
+        tasks.push(cx.background_spawn(async move {
+            Self::handle_requests(&client, &dispatch_tx, &receiver).await
+        }));
+
+        // Re-read seen profiles from the database after each batch sync.
+        tasks.push(cx.spawn(async move |this, cx| {
+            while let Ok(Dispatch::Synced) = dispatch_rx.recv_async().await {
+                this.update(cx, |this, cx| this.apply_seen(cx)).ok();
+            }
+            Ok(())
+        }));
+
         let mut store = Self {
             profiles: HashMap::new(),
-            seen: HashSet::new(),
-            queued: HashSet::new(),
-            fetching: false,
-            tasks: Vec::new(),
+            seen: RwLock::new(HashSet::new()),
+            sender,
+            tasks,
             _subscription: subscription,
         };
 
@@ -121,14 +150,17 @@ impl ProfileStore {
 
     /// Get a profile. Returns a placeholder (default metadata) and queues a
     /// fetch if the profile isn't cached yet.
-    pub fn get(&mut self, public_key: PublicKey, cx: &mut Context<Self>) -> Profile {
-        if let Some(profile) = self.profiles.get(&public_key) {
+    pub fn get(&self, public_key: &PublicKey) -> Profile {
+        if let Some(profile) = self.profiles.get(public_key) {
             return profile.clone();
         }
 
-        if self.seen.insert(public_key) {
-            self.queued.insert(public_key);
-            self.queue_fetch(cx);
+        let public_key = *public_key;
+
+        if self.seen.write().unwrap().insert(public_key)
+            && let Err(e) = self.sender.send(public_key)
+        {
+            log::warn!("failed to queue profile fetch: {e}");
         }
 
         Profile::new(public_key, Metadata::default())
@@ -205,12 +237,12 @@ impl ProfileStore {
     /// Re-read the latest metadata of every requested author from the local
     /// database (used after a sync, which produces no NostrUpdate events).
     fn apply_seen(&mut self, cx: &mut Context<Self>) {
-        if self.seen.is_empty() {
+        let authors: Vec<PublicKey> = self.seen.read().unwrap().iter().copied().collect();
+        if authors.is_empty() {
             return;
         }
 
         let client = Backend::global(cx).read(cx).client();
-        let authors: Vec<PublicKey> = self.seen.iter().copied().collect();
 
         let work = cx.background_spawn(async move {
             let filter = Filter::new().kind(Kind::Metadata).authors(authors);
@@ -255,51 +287,47 @@ impl ProfileStore {
         }));
     }
 
-    /// Drain the queue in a batched fetch, debounced to collect requests.
-    fn queue_fetch(&mut self, cx: &mut Context<Self>) {
-        if self.fetching {
-            return;
-        }
-        self.fetching = true;
+    /// Sync metadata for requested authors in batches, debounced to collect
+    /// requests. Runs on a background thread; results are dispatched to the
+    /// main thread, which re-reads the database.
+    async fn handle_requests(
+        client: &Client,
+        dispatch: &Sender<Dispatch>,
+        receiver: &Receiver<PublicKey>,
+    ) -> Result<(), Error> {
+        let mut batch: HashSet<PublicKey> = HashSet::new();
 
-        let client = Backend::global(cx).read(cx).client();
-
-        let task = cx.spawn(async move |this, cx| {
-            loop {
-                // Collect more requests before firing the batch.
-                cx.background_executor()
-                    .timer(Duration::from_millis(500))
-                    .await;
-
-                let batch = this.update(cx, |this, _cx| std::mem::take(&mut this.queued))?;
-
-                if batch.is_empty() {
-                    this.update(cx, |this, _cx| {
-                        this.fetching = false;
-                    })?;
-                    break;
+        loop {
+            // Wait for the first request of a batch.
+            match receiver.recv_timeout(BATCH_TIMEOUT) {
+                Ok(public_key) => {
+                    batch.insert(public_key);
                 }
+                Err(RecvTimeoutError::Disconnected) => return Ok(()),
+                Err(RecvTimeoutError::Timeout) => continue,
+            };
 
-                let filter = Filter::new()
-                    .kind(Kind::Metadata)
-                    .authors(batch.into_iter().collect::<Vec<PublicKey>>());
-
-                // Negentropy-sync with the bootstrap relays. Synced events
-                // are written to the database directly (no NostrUpdate), so
-                // re-apply from the database afterwards.
-                match sync_bootstrap_only(&client, filter, SyncOptions::default()).await {
-                    Ok(_) => {
-                        this.update(cx, |this, cx| this.apply_seen(cx))?;
-                    }
-                    Err(e) => {
-                        log::warn!("profile sync failed: {e}");
-                    }
-                }
+            // Collect everything that arrives within the debounce window.
+            let deadline = Instant::now() + BATCH_TIMEOUT;
+            while let Ok(public_key) = receiver.recv_deadline(deadline) {
+                batch.insert(public_key);
             }
 
-            Ok(())
-        });
+            let filter = Filter::new()
+                .kind(Kind::Metadata)
+                .authors(batch.drain().collect::<Vec<PublicKey>>());
 
-        self.tasks.push(task);
+            // Negentropy-sync with the bootstrap relays. Synced events are
+            // written to the database directly (no NostrUpdate), so re-apply
+            // from the database afterwards.
+            match sync_bootstrap_only(client, filter, SyncOptions::default()).await {
+                Ok(_) => {
+                    if dispatch.send(Dispatch::Synced).is_err() {
+                        log::warn!("profile dispatch channel closed, dropping sync result");
+                    }
+                }
+                Err(e) => log::warn!("profile sync failed: {e}"),
+            }
+        }
     }
 }