diff --git a/crates/signed_state/src/lib.rs b/crates/signed_state/src/lib.rs index fc47a83..3feebf2 100644 --- a/crates/signed_state/src/lib.rs +++ b/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; diff --git a/crates/signed_state/src/profile.rs b/crates/signed_state/src/profile.rs index 0f59430..c609877 100644 --- a/crates/signed_state/src/profile.rs +++ b/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::{Receiver, Sender}; +use flume::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) -> Vec { + let mut latest: HashMap = 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), +} + +/// Batches metadata requests into relay syncs feeding the local database. pub struct ProfileStore { - profiles: HashMap, - seen: RefCell>, + in_flight: RefCell>, sender: Sender, + _task: Task>, _subscription: Subscription, } @@ -76,6 +101,8 @@ struct GlobalProfileStore(Entity); impl Global for GlobalProfileStore {} +impl EventEmitter for ProfileStore {} + impl ProfileStore { pub fn global(cx: &App) -> Entity { cx.global::().0.clone() @@ -87,201 +114,256 @@ impl ProfileStore { pub(crate) fn new(cx: &mut Context) -> Self { let backend = Backend::global(cx); - let entity = cx.entity().downgrade(); let client = backend.read(cx).client(); let (sender, receiver) = flume::unbounded::(); 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| { - if let Err(e) = Self::handle_requests(this, &client, &receiver, cx).await { - log::error!("Failed to handle requests: {e}"); - } - }) - .detach(); + let task = cx.spawn(async move |this, cx| { + let mut batch: HashSet = HashSet::new(); - cx.defer(move |cx| { - entity.update(cx, |this, cx| this.load(cx)).ok(); + 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 = 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(), - seen: RefCell::new(HashSet::new()), + in_flight: RefCell::new(HashSet::new()), sender, + _task: task, _subscription: subscription, } } - // Returns a placeholder until fetched; queues a fetch when uncached. - pub fn get(&self, public_key: &PublicKey) -> Profile { - if let Some(profile) = self.profiles.get(public_key) { - return profile.clone(); - } - - let public_key = *public_key; - - if self.seen.borrow_mut().insert(public_key) - && let Err(e) = self.sender.send(public_key) + /// Queue a batched relay sync for keys without one in flight. + pub fn fetch(&self, keys: impl IntoIterator) { + let mut queue = Vec::new(); { - log::warn!("failed to queue profile fetch: {e}"); - } + let mut in_flight = self.in_flight.borrow_mut(); - Profile::new(public_key, Metadata::default()) - } - - fn load(&mut self, cx: &mut Context) { - 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 = 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); + for key in keys { + if in_flight.insert(key) { + queue.push(key); } - cx.notify(); - })?; - - Ok::<_, Error>(()) - }) - .detach(); + } + } + for public_key in queue { + if let Err(e) = self.sender.send(public_key) { + log::warn!("failed to queue profile fetch: {e}"); + } + } } - fn apply_authors(&mut self, authors: Vec, cx: &mut Context) { + // Releases the in-flight keys so future fetches can be queued again. + fn emit_updated(&mut self, authors: Vec, cx: &mut Context) { if authors.is_empty() { return; } - 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).authors(authors); - let events = client.database().query(filter).await?; - // Pick the latest metadata per author off the main thread. - let mut latest: HashMap = 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 = 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) { - let authors: Vec = self.seen.borrow().iter().copied().collect(); - self.apply_authors(authors, cx); - } - - async fn handle_requests( - this: WeakEntity, - client: &Client, - receiver: &Receiver, - cx: &mut AsyncApp, - ) -> Result<(), Error> { - let mut batch: HashSet = 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 = 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(()); + { + let mut in_flight = self.in_flight.borrow_mut(); + for author in &authors { + in_flight.remove(author); } } + + cx.emit(ProfileEvent::Updated(authors)); + } +} + +/// Profile cache bound to one panel or screen. +pub struct ProfileDirectory { + profiles: HashMap, + pending: RefCell>, + tasks: Vec>, + _subscriptions: Vec, +} + +impl ProfileDirectory { + pub fn new(cx: &mut App) -> Entity { + 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, cx: &mut Context) { + let missing: Vec = 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) { + let tracked: Vec = 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, queue_sync: bool, cx: &mut Context) { + // 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 = + 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, cx: &mut Context) { + 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(&self, then: impl FnOnce(&mut HashSet) -> R) -> R { + let mut pending = self.pending.borrow_mut(); + then(&mut pending) } }