From b349656c56b243f986dd8469afb523f2ae1f96d4 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Mon, 27 Jul 2026 10:46:10 +0700 Subject: [PATCH] . --- crates/chat/src/lib.rs | 166 ++++++++++++++++---------------------- crates/chat_ui/src/lib.rs | 106 ++++++++++-------------- crates/person/src/lib.rs | 122 +++++++++++++--------------- 3 files changed, 168 insertions(+), 226 deletions(-) diff --git a/crates/chat/src/lib.rs b/crates/chat/src/lib.rs index 2b67a0e..c5f71ca 100644 --- a/crates/chat/src/lib.rs +++ b/crates/chat/src/lib.rs @@ -2,7 +2,7 @@ use std::cmp::Reverse; use std::collections::{BTreeSet, HashMap, HashSet}; use std::hash::{DefaultHasher, Hash, Hasher}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, RwLock}; +use std::sync::{Arc, LazyLock, RwLock}; use std::time::Duration; use anyhow::{Context as AnyhowContext, Error, anyhow}; @@ -23,6 +23,9 @@ mod room; pub use message::*; pub use room::*; +/// A static keypair used only for signing locally-cached rumor events. +static LOCAL_KEYS: LazyLock = LazyLock::new(Keys::generate); + pub fn init(window: &mut Window, cx: &mut App) { ChatRegistry::set_global(cx.new(|cx| ChatRegistry::new(window, cx)), cx); } @@ -329,43 +332,42 @@ impl ChatRegistry { return; }; - self.tasks.push(cx.background_spawn(async move { - let opts = SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE); + let subscribe = cx.background_spawn({ + let client = client.clone(); - // Construct filter for msg relays - let msg_relays = Filter::new() - .kind(Kind::InboxRelays) - .author(public_key) - .limit(1); + async move { + let opts = + SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE); - // Construct filter for contact list - let contact_list = Filter::new() - .kind(Kind::ContactList) - .author(public_key) - .limit(1); + let msg_relays = Filter::new() + .kind(Kind::InboxRelays) + .author(public_key) + .limit(1); - // Subscribe - client - .subscribe(vec![msg_relays, contact_list]) - .close_on(opts) - .await?; + let contact_list = Filter::new() + .kind(Kind::ContactList) + .author(public_key) + .limit(1); - Ok(()) - })); + client + .subscribe(vec![msg_relays, contact_list]) + .close_on(opts) + .await + } + }); - let client = nostr.read(cx).client(); - - // Spawn a task to verify user inbox relays after 5 seconds self.tasks.push(cx.spawn(async move |this, cx| { + // Wait for subscription to complete (or fail silently) + _ = subscribe.await; + + // Give relays time to respond cx.background_executor().timer(Duration::from_secs(5)).await; - // Construct inbox relays filter let filter = Filter::new() .kind(Kind::InboxRelays) .author(public_key) .limit(1); - // Check the latest inbox relays event in database let found = client .database() .query(filter) @@ -390,52 +392,46 @@ impl ChatRegistry { let client = nostr.read(cx).client(); let signer = nostr.read(cx).signer(); - let task: Task> = cx.background_spawn(async move { - let public_key = signer.get_public_key_async().await?; - - // Construct inbox relays filter - let filter = Filter::new() - .kind(Kind::InboxRelays) - .author(public_key) - .limit(1); - - // Get the latest inbox relays event in database - let event = client - .database() - .query(filter) - .await? - .first_owned() - .ok_or(anyhow::anyhow!("No inbox relays found"))?; - - // Extract relay list from event - let relays: Vec = nip17::extract_relay_list(&event).collect(); - - // Ensure relay connections - for url in relays.iter() { - client.add_relay(url).and_connect().await?; - } - - // Construct gift wrap event filter - let filter = Filter::new().kind(Kind::GiftWrap).pubkey(public_key); - let id = SubscriptionId::new(format!("{}-msg", public_key.to_hex())); - - // Construct target for subscription - let target: HashMap = relays - .into_iter() - .map(|relay| (relay, filter.clone())) - .collect(); - - client.subscribe(target).with_id(id).await?; - - Ok(()) - }); - self.tasks.push(cx.spawn(async move |this, cx| { + let task: Task> = cx.background_spawn(async move { + let public_key = signer.get_public_key_async().await?; + + let filter = Filter::new() + .kind(Kind::InboxRelays) + .author(public_key) + .limit(1); + + let event = client + .database() + .query(filter) + .await? + .first_owned() + .ok_or(anyhow::anyhow!("No inbox relays found"))?; + + let relays: Vec = nip17::extract_relay_list(&event).collect(); + for url in relays.iter() { + client.add_relay(url).and_connect().await?; + } + + let filter = Filter::new().kind(Kind::GiftWrap).pubkey(public_key); + let id = SubscriptionId::new(format!("{}-msg", public_key.to_hex())); + + let target: HashMap = relays + .into_iter() + .map(|relay| (relay, filter.clone())) + .collect(); + + client.subscribe(target).with_id(id).await?; + + Ok(()) + }); + if let Err(e) = task.await { this.update(cx, |_this, cx| { cx.emit(ChatEvent::Error(e.to_string())); })?; } + Ok(()) })); } @@ -519,25 +515,18 @@ impl ChatRegistry { /// Add a new room to the start of list. pub fn add_room(&mut self, room: I, cx: &mut Context) where - I: Into + 'static, + I: Into, { let nostr = NostrRegistry::global(cx); - let Some(public_key) = nostr.read(cx).current_user() else { return; }; - self.tasks.push(cx.spawn(async move |this, cx| { - let room: Room = room.into().organize(&public_key); + let room: Room = room.into().organize(&public_key); + self.rooms.insert(0, cx.new(|_| room)); - this.update(cx, |this, cx| { - this.rooms.insert(0, cx.new(|_| room)); - cx.emit(ChatEvent::Ping); - cx.notify(); - })?; - - Ok(()) - })); + cx.emit(ChatEvent::Ping); + cx.notify(); } /// Emit an open room event. @@ -791,7 +780,7 @@ async fn extract_rumor( } // Try to unwrap with the available signer - let unwrapped = try_unwrap(signer, gift_wrap).await?; + let unwrapped = try_unwrap_with(signer, gift_wrap).await?; let mut rumor = unwrapped.rumor; // Generate event id for the rumor if it doesn't have one @@ -805,25 +794,6 @@ async fn extract_rumor( Ok(rumor) } -/// Helper method to try unwrapping with different signers -async fn try_unwrap(signer: &UniversalSigner, gift_wrap: &Event) -> Result { - /* - * // Try with the device signer first - if let Some(signer) = signer.get_encryption_signer().await { - log::info!("trying with encryption key"); - if let Ok(unwrapped) = try_unwrap_with(gift_wrap, &signer).await { - return Ok(unwrapped); - } - } - - // Fallback to the user's signer - let user_signer = signer.get().await; - */ - let unwrapped = try_unwrap_with(signer, gift_wrap).await?; - - Ok(unwrapped) -} - /// Attempts to unwrap a gift wrap event with a given signer. async fn try_unwrap_with( signer: &UniversalSigner, @@ -882,7 +852,7 @@ async fn set_rumor(client: &Client, id: EventId, rumor: &UnsignedEvent) -> Resul // Construct the event let event = EventBuilder::new(Kind::ApplicationSpecificData, content) .tags(tags) - .finalize_async(&Keys::generate()) + .finalize_async(&*LOCAL_KEYS) .await?; // Save the event to the database diff --git a/crates/chat_ui/src/lib.rs b/crates/chat_ui/src/lib.rs index 9fc1164..a809145 100644 --- a/crates/chat_ui/src/lib.rs +++ b/crates/chat_ui/src/lib.rs @@ -1,7 +1,7 @@ use std::collections::{BTreeMap, BTreeSet, HashSet}; -use std::sync::Arc; #[cfg(target_arch = "wasm32")] -use std::sync::RwLock; +use std::sync::RwLock as AsyncRwLock; +use std::sync::{Arc, RwLock}; pub use actions::*; use anyhow::{Context as AnyhowContext, Error}; @@ -21,7 +21,7 @@ use person::{Person, PersonRegistry}; use settings::{AppSettings, SignerKind}; use smallvec::{SmallVec, smallvec}; #[cfg(not(target_arch = "wasm32"))] -use smol::lock::RwLock; +use smol::lock::RwLock as AsyncRwLock; use state::{NostrRegistry, upload}; use theme::ActiveTheme; use ui::avatar::Avatar; @@ -63,7 +63,7 @@ pub struct ChatPanel { rendered_texts_by_id: BTreeMap, /// Mapping message (rumor event) ids to their reports - reports_by_id: Entity>>, + reports_by_id: Arc>>>, /// Chat input state input: Entity, @@ -75,7 +75,7 @@ pub struct ChatPanel { subject_bar: Entity, /// Sent message ids - sent_ids: Arc>>, + sent_ids: Arc>>, /// Replies to replies_to: Entity>, @@ -98,7 +98,7 @@ impl ChatPanel { // Define attachments and replies_to entities let attachments = cx.new(|_| vec![]); let replies_to = cx.new(|_| HashSet::new()); - let reports_by_id = cx.new(|_| BTreeMap::new()); + let reports_by_id = Arc::new(RwLock::new(BTreeMap::new())); // Define list of messages let messages = BTreeSet::default(); @@ -172,7 +172,7 @@ impl ChatPanel { attachments, rendered_texts_by_id: BTreeMap::new(), reports_by_id, - sent_ids: Arc::new(RwLock::new(Vec::new())), + sent_ids: Arc::new(AsyncRwLock::new(Vec::new())), uploading: false, subscriptions, tasks: vec![], @@ -192,7 +192,7 @@ impl ChatPanel { let nostr = NostrRegistry::global(cx); let client = nostr.read(cx).client(); let sent_ids = self.sent_ids.clone(); - let reports = self.reports_by_id.downgrade(); + let reports = self.reports_by_id.clone(); let (tx, rx) = flume::bounded::>(256); @@ -223,10 +223,11 @@ impl ChatPanel { Ok(()) })); - self.tasks.push(cx.spawn(async move |_this, cx| { + self.tasks.push(cx.spawn(async move |this, cx| { while let Ok(status) = rx.recv_async().await { - reports.update(cx, |this, cx| { - for reports in this.values_mut() { + { + let mut map = reports.write().unwrap(); + for reports in map.values_mut() { for report in reports.iter_mut() { let Some(output) = report.output.as_mut() else { continue; @@ -243,10 +244,10 @@ impl ChatPanel { } } } - cx.notify(); } } - })?; + } + this.update(cx, |_, cx| cx.notify()).ok(); } Ok(()) })); @@ -345,69 +346,46 @@ impl ChatPanel { return; } - // Get room entity let room = self.room.clone(); - - // Get content and replies let replies: Vec = self.replies_to.read(cx).iter().copied().collect(); let content = value.to_string(); - - self.tasks.push(cx.spawn_in(window, async move |this, cx| { - let room = room.upgrade().context("Room is not available")?; - - this.update_in(cx, |this, window, cx| { - match room.read(cx).rumor(content, replies, cx) { - Some(rumor) => { - this.insert_message(&rumor, true, cx); - this.send_and_wait(rumor, window, cx); - this.clear(window, cx); - } - None => { - window.push_notification("Failed to create message", cx); - } - } - })?; - - Ok(()) - })); - } - - /// Send message in the background and wait for the response - fn send_and_wait(&mut self, rumor: UnsignedEvent, window: &mut Window, cx: &mut Context) { let sent_ids = self.sent_ids.clone(); - // This can't fail, because we already ensured that the ID is set - let id = rumor.id.unwrap(); - - // Add empty reports - self.insert_reports(id, vec![], cx); - - // Upgrade room reference - let Some(room) = self.room.upgrade() else { + // Upgrade room and create the rumor synchronously + 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; }; - // Get the send message task - let Some(task) = room.read(cx).send(rumor, cx) else { + let id = rumor.id.expect("rumor must have an id"); + + // Insert optimistic message and clear input + self.insert_message(&rumor, true, cx); + 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| { - // Send and get reports - let outputs = task.await; + let outputs = send_task.await; - // Add sent IDs to the list let mut sent_ids = sent_ids.write().await; sent_ids.extend(outputs.iter().filter_map(|output| output.gift_wrap_id)); - // Update the state this.update(cx, |this, cx| { this.insert_reports(id, outputs, cx); })?; Ok(()) - })) + })); } /// Clear the input field, attachments, and replies @@ -429,10 +407,13 @@ impl ChatPanel { /// Insert reports fn insert_reports(&mut self, id: EventId, reports: Vec, cx: &mut Context) { - self.reports_by_id.update(cx, |this, cx| { - this.entry(id).or_default().extend(reports); - cx.notify(); - }); + self.reports_by_id + .write() + .unwrap() + .entry(id) + .or_default() + .extend(reports); + cx.notify(); } /// Insert a message into the chat panel @@ -466,13 +447,12 @@ impl ChatPanel { } /// Check if a message has any reports - fn has_reports(&self, id: &EventId, cx: &App) -> bool { - self.reports_by_id.read(cx).get(id).is_some() + fn has_reports(&self, id: &EventId, _cx: &App) -> bool { + self.reports_by_id.read().unwrap().get(id).is_some() } - /// Get all sent reports for a message by its ID - fn sent_reports(&self, id: &EventId, cx: &App) -> Option> { - self.reports_by_id.read(cx).get(id).cloned() + fn sent_reports(&self, id: &EventId, _cx: &App) -> Option> { + self.reports_by_id.read().unwrap().get(id).cloned() } /// Get a message by its ID diff --git a/crates/person/src/lib.rs b/crates/person/src/lib.rs index b8423f1..0c9d994 100644 --- a/crates/person/src/lib.rs +++ b/crates/person/src/lib.rs @@ -1,6 +1,5 @@ use std::cell::RefCell; use std::collections::{HashMap, HashSet}; -use std::rc::Rc; use std::time::Duration; use anyhow::{Error, anyhow}; @@ -24,9 +23,9 @@ impl Global for GlobalPersonRegistry {} #[derive(Debug, Clone)] enum Dispatch { - Person(Box), - Announcement(Box), - Relays(Box), + Person(Person), + Announcement(Event), + Relays(Event), } /// Person Registry @@ -36,7 +35,7 @@ pub struct PersonRegistry { persons: HashMap>, /// Set of public keys that have been seen - seens: Rc>>, + seens: RefCell>, /// Sender for requesting metadata sender: flume::Sender, @@ -67,49 +66,38 @@ impl PersonRegistry { let mut tasks = smallvec![]; - tasks.push( - // Handle nostr notifications - cx.background_spawn({ - let client = client.clone(); + tasks.push(cx.background_spawn({ + let client = client.clone(); + async move { + Self::handle_notifications(&client, &tx).await; + } + })); - async move { - Self::handle_notifications(&client, &tx).await; - } - }), - ); + tasks.push(cx.background_spawn({ + let client = client.clone(); + async move { + Self::handle_requests(&client, &mta_rx).await; + } + })); - tasks.push( - // Handle metadata requests - cx.background_spawn({ - let client = client.clone(); - - async move { - Self::handle_requests(&client, &mta_rx).await; - } - }), - ); - - tasks.push( - // Update GPUI state - cx.spawn(async move |this, cx| { - while let Ok(event) = rx.recv_async().await { - this.update(cx, |this, cx| { - match event { - Dispatch::Person(person) => { - this.insert(*person, cx); - } - Dispatch::Announcement(event) => { - this.set_announcement(&event, cx); - } - Dispatch::Relays(event) => { - this.set_messaging_relays(&event, cx); - } - }; - }) - .ok(); - } - }), - ); + tasks.push(cx.spawn(async move |this, cx| { + while let Ok(event) = rx.recv_async().await { + this.update(cx, |this, cx| { + match event { + Dispatch::Person(person) => { + this.insert(person, cx); + } + Dispatch::Announcement(event) => { + this.set_announcement(&event, cx); + } + Dispatch::Relays(event) => { + this.set_messaging_relays(&event, cx); + } + }; + }) + .ok(); + } + })); // Load all user profiles from the database cx.defer_in(window, |this, _window, cx| { @@ -118,7 +106,7 @@ impl PersonRegistry { Self { persons: HashMap::new(), - seens: Rc::new(RefCell::new(HashSet::new())), + seens: RefCell::new(HashSet::new()), sender: mta_tx, tasks, } @@ -145,24 +133,25 @@ impl PersonRegistry { Kind::Metadata => { let metadata = Metadata::from_json(&event.content).unwrap_or_default(); let person = Person::new(event.pubkey, metadata); - let val = Box::new(person); - // Send - tx.send_async(Dispatch::Person(val)).await.ok(); + if tx.send_async(Dispatch::Person(person)).await.is_err() { + log::warn!("PersonRegistry channel closed, dropping metadata event"); + } } Kind::ContactList => { let public_keys = event.extract_public_keys(); - // Get metadata for all public keys - get_metadata(client, public_keys).await.ok(); + if let Err(e) = get_metadata(client, public_keys).await { + log::warn!("Failed to get metadata for contact list: {e}"); + } } Kind::InboxRelays => { - let val = Box::new(event.into_owned()); - // Send - tx.send_async(Dispatch::Relays(val)).await.ok(); + tx.send_async(Dispatch::Relays(event.into_owned())) + .await + .ok(); } Kind::Custom(10044) => { - let val = Box::new(event.into_owned()); - // Send - tx.send_async(Dispatch::Announcement(val)).await.ok(); + tx.send_async(Dispatch::Announcement(event.into_owned())) + .await + .ok(); } _ => {} } @@ -182,13 +171,17 @@ impl PersonRegistry { Ok(Some(public_key)) => { batch.insert(public_key); // Process the batch if it's full - if batch.len() >= 20 { - get_metadata(client, std::mem::take(&mut batch)).await.ok(); + if batch.len() >= 20 + && let Err(e) = get_metadata(client, std::mem::take(&mut batch)).await + { + log::warn!("Failed to get metadata batch: {e}"); } } _ => { - if !batch.is_empty() { - get_metadata(client, std::mem::take(&mut batch)).await.ok(); + if !batch.is_empty() + && let Err(e) = get_metadata(client, std::mem::take(&mut batch)).await + { + log::warn!("Failed to get metadata batch: {e}"); } } } @@ -290,15 +283,14 @@ impl PersonRegistry { } let public_key = *public_key; - let mut seen = self.seens.borrow_mut(); - if seen.insert(public_key) { + if self.seens.borrow_mut().insert(public_key) { let sender = self.sender.clone(); // Spawn background task to request metadata cx.background_spawn(async move { if let Err(e) = sender.send_async(public_key).await { - log::warn!("Failed to send public key for metadata request: {}", e); + log::warn!("Failed to send public key for metadata request: {e}"); } }) .detach();