diff --git a/Cargo.lock b/Cargo.lock index d6c2684..81ac321 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1056,6 +1056,7 @@ dependencies = [ "common", "device", "flume 0.11.1", + "futures", "fuzzy-matcher", "gpui", "itertools 0.13.0", diff --git a/crates/chat/Cargo.toml b/crates/chat/Cargo.toml index 3117b91..01f1184 100644 --- a/crates/chat/Cargo.toml +++ b/crates/chat/Cargo.toml @@ -20,6 +20,7 @@ smallvec.workspace = true log.workspace = true flume.workspace = true +futures.workspace = true fuzzy-matcher = "0.3.7" [target.'cfg(not(target_arch = "wasm32"))'.dependencies] diff --git a/crates/chat/src/lib.rs b/crates/chat/src/lib.rs index c5f71ca..574f7f8 100644 --- a/crates/chat/src/lib.rs +++ b/crates/chat/src/lib.rs @@ -82,15 +82,15 @@ pub struct ChatRegistry { rooms: Vec>, /// Events that failed to unwrap for any reason - trashes: Entity>, + trash: Entity>, /// Tracking events seen on which relays in the current session - seens: Arc>>>, + seen: Arc>>>, /// Mapping of unwrapped event ids to their gift wrap event ids event_map: Arc>>, - /// Tracking the status of unwrapping gift wrap events. + /// True while the initial event backlog is still loading tracking: Arc, /// Channel for sending signals to the UI. @@ -108,10 +108,31 @@ pub struct ChatRegistry { /// Signal consumer task (cancelled on signer change) signal_consumer: Option>>, + /// Fuzzy matcher for room search (cached; intentionally excluded from Debug) + #[allow(dead_code)] + matcher: CachedMatcher, + /// Subscriptions _subscriptions: SmallVec<[Subscription; 2]>, } +/// Wrapper to provide Debug for SkimMatcherV2 +struct CachedMatcher(SkimMatcherV2); + +impl std::fmt::Debug for CachedMatcher { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("CachedMatcher { .. }") + } +} + +impl std::ops::Deref for CachedMatcher { + type Target = SkimMatcherV2; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + impl EventEmitter for ChatRegistry {} impl ChatRegistry { @@ -145,16 +166,16 @@ impl ChatRegistry { // Run at the end of the current cycle cx.defer_in(window, |this, _window, cx| { - this.tracking(cx); this.get_rooms(cx); }); Self { rooms: vec![], - trashes: cx.new(|_| BTreeSet::default()), - seens: Arc::new(RwLock::new(HashMap::default())), + trash: cx.new(|_| BTreeSet::default()), + seen: Arc::new(RwLock::new(HashMap::default())), event_map: Arc::new(RwLock::new(HashMap::default())), - tracking: Arc::new(AtomicBool::new(false)), + tracking: Arc::new(AtomicBool::new(true)), + matcher: CachedMatcher(SkimMatcherV2::default()), signal_rx: rx, signal_tx: tx, tasks: smallvec![], @@ -174,12 +195,10 @@ impl ChatRegistry { let client = nostr.read(cx).client(); let signer = nostr.read(cx).signer(); - let tracking = self.tracking.clone(); - let seens = self.seens.clone(); + let seen = self.seen.clone(); let event_map = self.event_map.clone(); - let trashes = self.trashes.downgrade(); + let trash = self.trash.downgrade(); - let initialized_at = Timestamp::now(); let sub_id1 = SubscriptionId::new(DEVICE_GIFTWRAP); let sub_id2 = SubscriptionId::new(USER_GIFTWRAP); @@ -190,16 +209,19 @@ impl ChatRegistry { self.notification_listener = Some(cx.background_spawn(async move { let mut notifications = client.notifications(); let mut processed_events = HashSet::new(); + const MAX_PROCESSED: usize = 10_000; while let Some(notification) = notifications.next().await { let ClientNotification::Message { message, relay_url } = notification else { - // Skip non-message notifications continue; }; match *message { RelayMessage::Event { event, .. } => { - // De-duplicate events by their ID + // Prune the dedup set before it grows unbounded + if processed_events.len() >= MAX_PROCESSED { + processed_events.clear(); + } if !processed_events.insert(event.id) { continue; } @@ -207,7 +229,6 @@ impl ChatRegistry { // Handle msg relays event to determine when the app is ready to subscribe if event.kind == Kind::InboxRelays { let current_user = signer.get_public_key_async().await?; - // Emit the inbox ready signal if event.pubkey == current_user { tx.send_async(Signal::InboxReady).await?; } @@ -220,14 +241,13 @@ impl ChatRegistry { // Keep track of which relays have seen this event { - let mut seens = seens.write().unwrap(); - seens.entry(event.id).or_default().insert(relay_url); + let mut seen = seen.write().unwrap(); + seen.entry(event.id).or_default().insert(relay_url); } // Extract the rumor from the gift wrap event match extract_rumor(&client, &signer, event.as_ref()).await { Ok(rumor) => { - // Map the rumor id to the gift wrap event id for later lookup let Some(rumor_id) = rumor.id else { log::error!("Rumor missing id after ensure_id"); continue; @@ -237,20 +257,14 @@ impl ChatRegistry { event_map.insert(rumor_id, event.id); } - // Check if the rumor has a recipient if rumor.tags.is_empty() { let signal = Signal::error(&event, "Recipient is missing"); tx.send_async(signal).await?; } - // Check if the rumor was created after the chat was initialized (for detecting new messages) - if rumor.created_at >= initialized_at { - let signal = Signal::message(event.id, rumor); - tx.send_async(signal).await?; - } else { - // Mark the chat still processing new messages - tracking.store(true, Ordering::Release); - } + // Emit message for both new and backlog events + let signal = Signal::message(event.id, rumor); + tx.send_async(signal).await?; } Err(e) => { let reason = format!("Failed to extract rumor: {e}"); @@ -286,12 +300,13 @@ impl ChatRegistry { } Signal::Eose => { this.update(cx, |this, cx| { - this.get_rooms(cx); + this.tracking.store(false, Ordering::Release); + this.sort(cx); })?; } - Signal::Error(trash) => { - trashes.update(cx, |this, cx| { - this.insert(trash); + Signal::Error(failed) => { + trash.update(cx, |this, cx| { + this.insert(failed); cx.notify(); })?; } @@ -302,27 +317,6 @@ impl ChatRegistry { })); } - /// Check periodically whether old gift-wrap events have finished processing, - /// and refresh rooms once the backlog is caught up. - fn tracking(&mut self, cx: &mut Context) { - let status = self.tracking.clone(); - let tx = self.signal_tx.clone(); - let check_interval = Duration::from_secs(15); - - self.tasks.push(cx.spawn(async move |_, cx| { - loop { - cx.background_executor().timer(check_interval).await; - - // Only trigger a room refresh if old events were being tracked - // (i.e. the notification handler set the flag while catching up). - // `swap` atomically reads and clears the flag. - if status.swap(false, Ordering::AcqRel) { - _ = tx.send_async(Signal::Eose).await; - } - } - })); - } - /// Get all necessary metadata from relays for current user pub fn get_metadata(&mut self, cx: &mut Context) { let nostr = NostrRegistry::global(cx); @@ -332,37 +326,29 @@ impl ChatRegistry { return; }; - let subscribe = cx.background_spawn({ - let client = client.clone(); - - async move { - let opts = - SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE); - - let msg_relays = Filter::new() - .kind(Kind::InboxRelays) - .author(public_key) - .limit(1); - - let contact_list = Filter::new() - .kind(Kind::ContactList) - .author(public_key) - .limit(1); - - client - .subscribe(vec![msg_relays, contact_list]) - .close_on(opts) - .await - } - }); - self.tasks.push(cx.spawn(async move |this, cx| { - // Wait for subscription to complete (or fail silently) - _ = subscribe.await; + // Subscribe to metadata from relays + let opts = SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE); + + let msg_relays = Filter::new() + .kind(Kind::InboxRelays) + .author(public_key) + .limit(1); + + let contact_list = Filter::new() + .kind(Kind::ContactList) + .author(public_key) + .limit(1); + + _ = client + .subscribe(vec![msg_relays, contact_list]) + .close_on(opts) + .await; // Give relays time to respond cx.background_executor().timer(Duration::from_secs(5)).await; + // Verify inbox relays were received let filter = Filter::new() .kind(Kind::InboxRelays) .author(public_key) @@ -436,8 +422,8 @@ impl ChatRegistry { })); } - /// Refresh the chat registry, fetching messages and contact list from relays. - pub fn refresh(&mut self, cx: &mut Context) { + /// Reload the chat registry, fetching messages and contact list from relays. + pub fn reload(&mut self, cx: &mut Context) { self.reset(cx); self.get_metadata(cx); self.get_rooms(cx); @@ -475,22 +461,22 @@ impl ChatRegistry { /// Count the number of messages seen by a given relay. pub fn count_messages(&self, relay_url: &RelayUrl) -> usize { - self.seens + self.seen .read() .unwrap() .values() - .filter(|seen| seen.contains(relay_url)) + .filter(|s| s.contains(relay_url)) .count() } /// Count the number of trash messages. pub fn count_trash_messages(&self, cx: &App) -> usize { - self.trashes.read(cx).len() + self.trash.read(cx).len() } /// Get the trash messages entity. - pub fn trashes(&self) -> Entity> { - self.trashes.clone() + pub fn trash(&self) -> Entity> { + self.trash.clone() } /// Get the relays that have seen a given rumor id. @@ -504,7 +490,7 @@ impl ChatRegistry { /// Get the relays that have seen a given gift wrap id. pub fn seen_on(&self, id: &EventId) -> HashSet { - self.seens + self.seen .read() .unwrap() .get(id) @@ -560,8 +546,6 @@ impl ChatRegistry { /// Finding rooms based on a query. pub fn find(&self, query: &str, cx: &App) -> Vec> { - let matcher = SkimMatcherV2::default(); - if let Ok(public_key) = PublicKey::parse(query) { self.rooms .iter() @@ -572,7 +556,7 @@ impl ChatRegistry { self.rooms .iter() .filter(|room| { - matcher + self.matcher .fuzzy_match(room.read(cx).display_name(cx).as_ref(), query) .is_some() }) @@ -584,7 +568,7 @@ impl ChatRegistry { /// Reset the registry. pub fn reset(&mut self, cx: &mut Context) { self.rooms.clear(); - self.trashes.update(cx, |this, cx| { + self.trash.update(cx, |this, cx| { this.clear(); cx.notify(); }); diff --git a/crates/chat/src/room.rs b/crates/chat/src/room.rs index 9971a85..885dbe5 100644 --- a/crates/chat/src/room.rs +++ b/crates/chat/src/room.rs @@ -363,22 +363,31 @@ impl Room { .exit_policy(ReqExitPolicy::ExitOnEOSE) .timeout(Some(Duration::from_secs(TIMEOUT))); - for public_key in members.into_iter() { - let inbox = Filter::new() - .author(public_key) - .kind(Kind::InboxRelays) - .limit(1); + let tasks: Vec<_> = members + .into_iter() + .map(|public_key| { + let client = client.clone(); + async move { + let inbox = Filter::new() + .author(public_key) + .kind(Kind::InboxRelays) + .limit(1); - let announcement = Filter::new() - .author(public_key) - .kind(Kind::Custom(10044)) - .limit(1); + let announcement = Filter::new() + .author(public_key) + .kind(Kind::Custom(10044)) + .limit(1); - // Subscribe to the target - client - .subscribe(vec![inbox, announcement]) - .close_on(opts) - .await?; + client + .subscribe(vec![inbox, announcement]) + .close_on(opts) + .await + } + }) + .collect(); + + for result in futures::future::join_all(tasks).await { + result?; } Ok(()) @@ -429,14 +438,6 @@ impl Room { // Get current user's public key let sender = nostr.read(cx).current_user()?; - // Get all members, excluding the sender - let members: Vec = self - .members - .iter() - .filter(|public_key| public_key != &&sender) - .map(|member| persons.read(cx).get(member, cx)) - .collect(); - // Construct event's tags let mut tags = vec![]; @@ -450,8 +451,9 @@ impl Room { tags.push(Tag::event(id)) } - // Add all receiver tags - for member in members.into_iter() { + // Add all receiver tags (no intermediate allocation) + for public_key in self.members.iter().filter(|pk| *pk != &sender) { + let member = persons.read(cx).get(public_key, cx); tags.push( Nip01Tag::PublicKey { public_key: member.public_key(), @@ -473,6 +475,30 @@ impl Room { Some(event) } + /// Select the appropriate signer based on signer kind and available keys. + fn select_signer( + signer_kind: &SignerKind, + has_announcement: bool, + encryption_signer: &Option, + user_signer: &UniversalSigner, + ) -> UniversalSigner { + match signer_kind { + SignerKind::Auto => { + if has_announcement { + encryption_signer + .clone() + .unwrap_or_else(|| user_signer.clone()) + } else { + user_signer.clone() + } + } + SignerKind::Encryption => encryption_signer + .clone() + .expect("encryption signer must be set"), + SignerKind::User => user_signer.clone(), + } + } + /// Send rumor event to all members's messaging relays pub fn send(&self, rumor: UnsignedEvent, cx: &App) -> Option>> { let config = self.config.clone(); @@ -525,23 +551,12 @@ impl Room { } // Determine the signer to use - let signer = match signer_kind { - SignerKind::Auto => { - if announcement.is_some() - && let Some(encryption_signer) = encryption_signer.clone() - { - // Safe to unwrap due to earlier checks - encryption_signer - } else { - user_signer.clone() - } - } - SignerKind::Encryption => { - // Safe to unwrap due to earlier checks - encryption_signer.as_ref().unwrap().clone() - } - SignerKind::User => user_signer.clone(), - }; + let signer = Self::select_signer( + signer_kind, + announcement.is_some(), + &encryption_signer, + &user_signer, + ); // Send the gift wrap event and collect the report match send_gift_wrap(&client, &signer, &member, &rumor, signer_kind).await { @@ -561,23 +576,12 @@ impl Room { let public_key = sender.public_key(); // Determine the signer to use - let signer = match signer_kind { - SignerKind::Auto => { - if sender.announcement().is_some() - && let Some(encryption_signer) = encryption_signer.clone() - { - // Safe to unwrap due to earlier checks - encryption_signer - } else { - user_signer.clone() - } - } - SignerKind::Encryption => { - // Safe to unwrap due to earlier checks - encryption_signer.as_ref().unwrap().clone() - } - SignerKind::User => user_signer.clone(), - }; + let signer = Self::select_signer( + signer_kind, + sender.announcement().is_some(), + &encryption_signer, + &user_signer, + ); match send_gift_wrap(&client, &signer, &sender, &rumor, signer_kind).await { Ok(report) => reports.push(report), diff --git a/crates/chat_ui/src/lib.rs b/crates/chat_ui/src/lib.rs index a809145..a00434f 100644 --- a/crates/chat_ui/src/lib.rs +++ b/crates/chat_ui/src/lib.rs @@ -351,13 +351,20 @@ impl ChatPanel { let content = value.to_string(); let sent_ids = self.sent_ids.clone(); - // Upgrade room and create the rumor synchronously + // Upgrade room and create rumor + send task in a single read lock let Some(room_entity) = room.upgrade() else { return; }; - let Some(rumor) = room_entity.read(cx).rumor(content.clone(), replies, cx) else { - window.push_notification("Failed to create message", cx); - return; + let (rumor, send_task) = match room_entity.read_with(cx, |room, cx| { + let rumor = room.rumor(content.clone(), replies, cx)?; + let send_task = room.send(rumor.clone(), cx)?; + Some((rumor, send_task)) + }) { + Some(pair) => pair, + None => { + window.push_notification("Failed to create message", cx); + return; + } }; let id = rumor.id.expect("rumor must have an id"); @@ -367,12 +374,6 @@ impl ChatPanel { self.insert_reports(id, vec![], cx); self.clear(window, cx); - // Get the send task - let Some(send_task) = room_entity.read(cx).send(rumor, cx) else { - window.push_notification("Failed to send message", cx); - return; - }; - // Spawn a single task to await the send and update reports self.tasks.push(cx.spawn_in(window, async move |this, cx| { let outputs = send_task.await; diff --git a/crates/device/src/lib.rs b/crates/device/src/lib.rs index 5694104..d3b5da3 100644 --- a/crates/device/src/lib.rs +++ b/crates/device/src/lib.rs @@ -92,7 +92,6 @@ impl DeviceRegistry { fn new(window: &mut Window, cx: &mut Context) -> Self { let nostr = NostrRegistry::global(cx); let settings = AppSettings::global(cx); - let nip4e_enabled = settings.read(cx).is_nip4e_enabled(cx); let signer = cx.new(|_| None); let mut subscriptions = smallvec![]; @@ -109,7 +108,7 @@ impl DeviceRegistry { subscriptions.push( // Observe the user signer cx.subscribe(&nostr, move |this, _nostr, event, cx| { - if event.signer_changed() && nip4e_enabled { + if event.signer_changed() && settings.read(cx).is_nip4e_enabled(cx) { this.get_announcement(cx); } }), diff --git a/crates/person/src/lib.rs b/crates/person/src/lib.rs index 0c9d994..94d7c63 100644 --- a/crates/person/src/lib.rs +++ b/crates/person/src/lib.rs @@ -35,7 +35,7 @@ pub struct PersonRegistry { persons: HashMap>, /// Set of public keys that have been seen - seens: RefCell>, + seen: RefCell>, /// Sender for requesting metadata sender: flume::Sender, @@ -62,22 +62,18 @@ impl PersonRegistry { // Channel for communication between nostr and gpui let (tx, rx) = flume::bounded::(100); - let (mta_tx, mta_rx) = flume::unbounded::(); + let (metadata_tx, metadata_rx) = flume::unbounded::(); let mut tasks = smallvec![]; - tasks.push(cx.background_spawn({ - let client = client.clone(); - async move { - Self::handle_notifications(&client, &tx).await; - } + let client2 = client.clone(); + tasks.push(cx.background_spawn(async move { + Self::handle_notifications(&client2, &tx).await; })); - tasks.push(cx.background_spawn({ - let client = client.clone(); - async move { - Self::handle_requests(&client, &mta_rx).await; - } + let client3 = client.clone(); + tasks.push(cx.background_spawn(async move { + Self::handle_requests(&client3, &metadata_rx).await; })); tasks.push(cx.spawn(async move |this, cx| { @@ -106,8 +102,8 @@ impl PersonRegistry { Self { persons: HashMap::new(), - seens: RefCell::new(HashSet::new()), - sender: mta_tx, + seen: RefCell::new(HashSet::new()), + sender: metadata_tx, tasks, } } @@ -210,7 +206,7 @@ impl PersonRegistry { self.tasks.push(cx.spawn(async move |this, cx| { if let Ok(persons) = task.await { this.update(cx, |this, cx| { - this.bulk_inserts(persons, cx); + this.bulk_insert(persons, cx); }) .ok(); } @@ -249,7 +245,7 @@ impl PersonRegistry { } /// Insert batch of persons - fn bulk_inserts(&mut self, persons: Vec, cx: &mut Context) { + fn bulk_insert(&mut self, persons: Vec, cx: &mut Context) { for person in persons.into_iter() { let public_key = person.public_key(); self.persons @@ -284,7 +280,7 @@ impl PersonRegistry { let public_key = *public_key; - if self.seens.borrow_mut().insert(public_key) { + if self.seen.borrow_mut().insert(public_key) { let sender = self.sender.clone(); // Spawn background task to request metadata diff --git a/crates/workspace/src/lib.rs b/crates/workspace/src/lib.rs index 6ccb5e5..ef5b812 100644 --- a/crates/workspace/src/lib.rs +++ b/crates/workspace/src/lib.rs @@ -366,7 +366,7 @@ impl Workspace { let chat = ChatRegistry::global(cx); // Trigger a refresh of the chat registry chat.update(cx, |this, cx| { - this.refresh(cx); + this.reload(cx); }); } Command::ShowRelayList => { diff --git a/crates/workspace/src/panels/trash.rs b/crates/workspace/src/panels/trash.rs index 2a2220c..7da5b10 100644 --- a/crates/workspace/src/panels/trash.rs +++ b/crates/workspace/src/panels/trash.rs @@ -37,9 +37,9 @@ impl TrashPanel { fn copy(&self, ix: usize, cx: &App) { let chat = ChatRegistry::global(cx); - let trashes = chat.read(cx).trashes(); + let trash_events = chat.read(cx).trash(); - if let Some(message) = trashes.read(cx).iter().nth(ix) { + if let Some(message) = trash_events.read(cx).iter().nth(ix) { let item = ClipboardItem::new_string(message.raw_event.to_string()); cx.write_to_clipboard(item); } @@ -52,9 +52,9 @@ impl TrashPanel { cx: &mut Context, ) -> AnyElement { let chat = ChatRegistry::global(cx); - let trashes = chat.read(cx).trashes(); + let trash_events = chat.read(cx).trash(); - if let Some(message) = trashes.read(cx).iter().nth(ix) { + if let Some(message) = trash_events.read(cx).iter().nth(ix) { v_flex() .id(ix) .p_2()