From 7dbf55d48e48be9c187ef4e7135017dd3003a869 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Wed, 30 Sep 2026 04:40:02 +0700 Subject: [PATCH] refactor send --- Cargo.lock | 2 - crates/chat/src/lib.rs | 24 +--- crates/chat/src/room.rs | 262 ++++++++++++++++---------------------- crates/chat_ui/Cargo.toml | 2 - crates/chat_ui/src/lib.rs | 96 +------------- 5 files changed, 121 insertions(+), 265 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7683cbd6..64364dd7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1097,8 +1097,6 @@ dependencies = [ "anyhow", "chat", "common", - "flume 0.11.1", - "futures", "gpui-pre", "itertools 0.13.0", "log", diff --git a/crates/chat/src/lib.rs b/crates/chat/src/lib.rs index 1da7beaf..c35b7143 100644 --- a/crates/chat/src/lib.rs +++ b/crates/chat/src/lib.rs @@ -73,41 +73,29 @@ impl Signal { pub struct ChatRegistry { /// Chat rooms rooms: Vec>, - /// O(1) room lookup by room ID room_index: HashMap>, - /// Events that failed to unwrap for any reason trash: Entity>, - /// Tracking events seen on which relays in the current session seen: Arc>>>, - /// Mapping of unwrapped event ids to their gift wrap event ids event_map: Arc>>, - /// True while the initial event backlog is still loading tracking: Arc, - /// Channel for sending signals to the UI. signal_tx: flume::Sender, - /// Channel for receiving signals from the UI. signal_rx: flume::Receiver, - /// Async tasks tasks: SmallVec<[Task>; 2]>, - /// Notification listener task (cancelled on signer change) notification_listener: Option>>, - /// 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]>, } @@ -144,8 +132,9 @@ impl ChatRegistry { /// Create a new chat registry instance fn new(cx: &mut Context) -> Self { - let entity = cx.entity().downgrade(); let nostr = NostrRegistry::global(cx); + let entity = cx.entity().downgrade(); + let (tx, rx) = flume::unbounded::(); let mut subscriptions = smallvec![]; @@ -162,11 +151,9 @@ impl ChatRegistry { ); cx.defer(move |cx| { - entity - .update(cx, |this, cx| { - this.get_rooms(cx); - }) - .ok(); + let _ = entity.update(cx, |this, cx| { + this.get_rooms(cx); + }); }); Self { @@ -237,6 +224,7 @@ impl ChatRegistry { if processed_events.len() >= MAX_PROCESSED { processed_events.clear(); } + if !processed_events.insert(event.id) { continue; } diff --git a/crates/chat/src/room.rs b/crates/chat/src/room.rs index cea715e7..74ba9a66 100644 --- a/crates/chat/src/room.rs +++ b/crates/chat/src/room.rs @@ -17,85 +17,38 @@ use crate::{FileAttachment, KIND_FILE_MESSAGE, NewMessage}; const NO_DEKEY: &str = "User hasn't set up a decoupled encryption key yet."; const USER_NO_DEKEY: &str = "You haven't set up a decoupled encryption key or it's not available."; +/// The outcome of delivering an encrypted rumor to a single receiver. #[derive(Debug, Clone)] pub struct SendReport { + /// The receiver the rumor was addressed to. pub receiver: PublicKey, - pub gift_wrap_id: Option, - pub error: Option, - pub output: Option>, + /// The per-relay send output, or the error that prevented the send. + result: Result, SharedString>, } impl SendReport { - pub fn new(receiver: PublicKey) -> Self { - Self { - receiver, - gift_wrap_id: None, - error: None, - output: None, - } + /// The error that prevented the send, if any. + pub fn error(&self) -> Option<&SharedString> { + self.result.as_ref().err() } - /// Set the gift wrap ID. - pub fn gift_wrap_id(mut self, gift_wrap_id: EventId) -> Self { - self.gift_wrap_id = Some(gift_wrap_id); - self + /// The per-relay send output, if the gift wrap was dispatched. + pub fn output(&self) -> Option<&Output> { + self.result.as_ref().ok() } - /// Set the output. - pub fn output(mut self, output: Output) -> Self { - self.output = Some(output); - self - } - - /// Set the error message. - pub fn error(mut self, error: T) -> Self - where - T: Into, - { - self.error = Some(error.into()); - self - } - - /// Returns true if the send is pending. - pub fn pending(&self) -> bool { - self.error.is_none() - && self - .output - .as_ref() - .is_some_and(|o| o.success.is_empty() && o.failed.is_empty()) - } - - /// Returns true if the send was successful. + /// Whether at least one relay accepted the event. pub fn success(&self) -> bool { - self.error.is_none() && self.output.as_ref().is_some_and(|o| !o.success.is_empty()) + self.output() + .is_some_and(|output| !output.success.is_empty()) } - /// Returns true if the send failed. + /// Whether no relay accepted the event. pub fn failed(&self) -> bool { - self.error.is_some() && self.output.as_ref().is_some_and(|o| !o.failed.is_empty()) - } -} - -#[derive(Debug, Clone)] -pub enum SendStatus { - Ok { - id: EventId, - relay: RelayUrl, - }, - Failed { - id: EventId, - relay: RelayUrl, - message: String, - }, -} - -impl SendStatus { - pub fn ok(id: EventId, relay: RelayUrl) -> Self { - Self::Ok { id, relay } - } - - pub fn failed(id: EventId, relay: RelayUrl, message: String) -> Self { - Self::Failed { id, relay, message } + match &self.result { + Err(_) => true, + Ok(output) => output.success.is_empty(), + } } } @@ -543,7 +496,8 @@ impl Room { } } - /// Send rumor event to all members's messaging relays + /// Send the rumor to every member's messaging relays + /// and to the sender as a backup, when enabled. pub fn send(&self, rumor: UnsignedEvent, cx: &App) -> Option>> { let config = self.config.clone(); @@ -560,7 +514,7 @@ impl Room { let sender = persons.read(cx).get(¤t_user, cx); // Get all members (excluding sender) - let members: Vec = self + let receivers: Vec = self .members .iter() .filter(|public_key| public_key != &&sender.public_key()) @@ -569,71 +523,33 @@ impl Room { Some(cx.background_spawn(async move { let signer_kind = config.signer_kind(); - let backup = config.backup(); - let mut sents = 0; - let mut reports = Vec::new(); - - // Process each member - for member in members { - let announcement = member.announcement(); - let public_key = member.public_key(); - - // Handle encryption signer requirements - if signer_kind.encryption() { - // Receiver didn't set up a decoupled encryption key - if announcement.is_none() { - reports.push(SendReport::new(public_key).error(NO_DEKEY)); - continue; - } - - // Sender didn't set up a decoupled encryption key - if encryption_signer.is_none() { - reports.push(SendReport::new(sender.public_key()).error(USER_NO_DEKEY)); - continue; - } - } - - // Determine the signer to use - let signer = Self::select_signer( + // Deliver to every receiver concurrently + let sends = receivers.into_iter().map(|receiver| { + deliver( + &client, signer_kind, - announcement.is_some(), + receiver, + &rumor, &encryption_signer, &user_signer, + ) + }); + let mut reports: Vec = futures::future::join_all(sends).await; + + // Back up the message to ourselves once at least one receiver got it + if config.backup() && reports.iter().any(SendReport::success) { + reports.push( + deliver( + &client, + signer_kind, + sender, + &rumor, + &encryption_signer, + &user_signer, + ) + .await, ); - - // Send the gift wrap event and collect the report - match send_gift_wrap(&client, &signer, &member, &rumor, signer_kind).await { - Ok(report) => { - reports.push(report); - sents += 1; - } - Err(error) => { - let report = SendReport::new(public_key).error(error.to_string()); - reports.push(report); - } - } - } - - // Send backup to current user if needed - if backup && sents >= 1 { - let public_key = sender.public_key(); - - // Determine the signer to use - 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), - Err(error) => { - let report = SendReport::new(public_key).error(error.to_string()); - reports.push(report); - } - } } reports @@ -641,55 +557,95 @@ impl Room { } } -// Helper function to send a gift-wrapped event +/// Deliver a rumor to a single receiver. +async fn deliver( + client: &Client, + signer_kind: &SignerKind, + receiver: Person, + rumor: &UnsignedEvent, + encryption_signer: &Option, + user_signer: &UniversalSigner, +) -> SendReport { + let public_key = receiver.public_key(); + let announcement = receiver.announcement(); + + if signer_kind.encryption() { + // Receiver didn't set up a decoupled encryption key + if announcement.is_none() { + return SendReport { + receiver: public_key, + result: Err(NO_DEKEY.into()), + }; + } + + // Sender didn't set up a decoupled encryption key + if encryption_signer.is_none() { + return SendReport { + receiver: public_key, + result: Err(USER_NO_DEKEY.into()), + }; + } + } + + // Determine the signer to use + let signer = Room::select_signer( + signer_kind, + announcement.is_some(), + encryption_signer, + user_signer, + ); + + match send_gift_wrap(client, &signer, &receiver, rumor, signer_kind).await { + Ok(output) => SendReport { + receiver: public_key, + result: Ok(output), + }, + Err(error) => SendReport { + receiver: public_key, + result: Err(error.to_string().into()), + }, + } +} + +/// Build the gift-wrapped event for a rumor and send it. async fn send_gift_wrap( client: &Client, signer: &UniversalSigner, receiver: &Person, rumor: &UnsignedEvent, config: &SignerKind, -) -> Result { - let k_tag = Tag::custom("k", [rumor.kind.to_string()]); - let mut extra_tags = vec![k_tag]; +) -> Result, Error> { + let mut extra_tags = vec![Tag::custom("k", [rumor.kind.to_string()])]; // Determine the receiver public key based on the config - let receiver = match config { - SignerKind::Auto => { - if let Some(announcement) = receiver.announcement().as_ref() { + let receiver_key = match config { + SignerKind::Auto => match receiver.announcement().as_ref() { + Some(announcement) => { extra_tags.push(Tag::public_key(receiver.public_key())); announcement.public_key() - } else { - receiver.public_key() } - } - SignerKind::Encryption => { - if let Some(announcement) = receiver.announcement().as_ref() { + None => receiver.public_key(), + }, + SignerKind::Encryption => match receiver.announcement().as_ref() { + Some(announcement) => { extra_tags.push(Tag::public_key(receiver.public_key())); announcement.public_key() - } else { - return Err(anyhow!("User has no encryption announcement")); } - } + None => return Err(anyhow!("User has no encryption announcement")), + }, SignerKind::User => receiver.public_key(), }; // Construct the gift wrap event - let event = nip59::GiftWrapBuilder::new(receiver, rumor.clone()) + let event = nip59::GiftWrapBuilder::new(receiver_key, rumor.clone()) .extra_tags(extra_tags) .finalize_async(signer) .await?; - // Send the gift wrap event and collect the report - let report = client + // Send to the receiver's NIP-17 relays. + client .send_event(&event) .to_nip17() - .ack_policy(AckPolicy::none()) .await - .map(|output| { - SendReport::new(receiver) - .gift_wrap_id(event.id) - .output(output) - })?; - - Ok(report) + .map_err(Into::into) } diff --git a/crates/chat_ui/Cargo.toml b/crates/chat_ui/Cargo.toml index 2eb07f81..fa820e42 100644 --- a/crates/chat_ui/Cargo.toml +++ b/crates/chat_ui/Cargo.toml @@ -18,8 +18,6 @@ nostr-sdk.workspace = true anyhow.workspace = true itertools.workspace = true smallvec.workspace = true -flume.workspace = true -futures.workspace = true log.workspace = true serde.workspace = true diff --git a/crates/chat_ui/src/lib.rs b/crates/chat_ui/src/lib.rs index e054c119..7cdd7670 100644 --- a/crates/chat_ui/src/lib.rs +++ b/crates/chat_ui/src/lib.rs @@ -4,9 +4,8 @@ use std::sync::{Arc, LazyLock, RwLock}; pub use actions::*; use anyhow::Error; -use chat::{ChatRegistry, Message, Room, RoomEvent, SendReport, SendStatus}; +use chat::{ChatRegistry, Message, Room, RoomEvent, SendReport}; use common::TimestampExt; -use futures::lock::Mutex; use gpui::prelude::FluentBuilder; use gpui::{ AnyElement, App, AppContext, ClipboardItem, Context, Entity, EventEmitter, FocusHandle, @@ -95,9 +94,6 @@ pub struct ChatPanel { /// Subject bar visibility subject_bar: Entity, - /// Sent message ids - sent_ids: Arc>>, - /// Replies to replies_to: Entity>, @@ -181,7 +177,6 @@ impl ChatPanel { // Define all functions that will run after the current cycle cx.defer_in(window, |this, window, cx| { this.connect(cx); - this.handle_notifications(cx); this.subscribe_room_events(window, cx); this.get_messages(window, cx); }); @@ -203,7 +198,6 @@ impl ChatPanel { decrypted_files: HashMap::new(), rendered_texts_by_id: BTreeMap::new(), reports_by_id, - sent_ids: Arc::new(Mutex::new(Vec::new())), uploading: false, subscriptions, tasks: vec![], @@ -218,78 +212,6 @@ impl ChatPanel { } } - /// Handle nostr notifications - fn handle_notifications(&mut self, cx: &mut Context) { - let nostr = NostrRegistry::global(cx); - let client = nostr.read(cx).client(); - let sent_ids = self.sent_ids.clone(); - let reports = self.reports_by_id.clone(); - - let (tx, rx) = flume::bounded::>(256); - - self.tasks.push(cx.background_spawn(async move { - let mut notifications = client.notifications(); - - while let Some(notification) = notifications.next().await { - if let ClientNotification::Message { message, relay_url } = notification - && let RelayMessage::Ok { - event_id, - status, - message, - } = *message - { - let sent_ids = sent_ids.lock().await; - - if sent_ids.contains(&event_id) { - let status = if status { - SendStatus::ok(event_id, relay_url) - } else { - SendStatus::failed(event_id, relay_url, message.into()) - }; - tx.send_async(Arc::new(status)).await.ok(); - } - } - } - - Ok(()) - })); - - self.tasks.push(cx.spawn(async move |this, cx| { - while let Ok(status) = rx.recv_async().await { - { - let mut map = reports.write().unwrap(); - let status_id = match &*status { - SendStatus::Ok { id, .. } => *id, - SendStatus::Failed { id, .. } => *id, - }; - - // Find the matching report and update it (exit early on first match) - 'outer: for reports_list in map.values_mut() { - for report in reports_list.iter_mut() { - let Some(output) = report.output.as_mut() else { - continue; - }; - if *output.id() != status_id { - continue; - } - match &*status { - SendStatus::Ok { relay, .. } => { - output.success.insert(relay.clone(), EventSendStatus::Sent); - } - SendStatus::Failed { relay, message, .. } => { - output.failed.insert(relay.clone(), message.clone()); - } - } - break 'outer; - } - } - } - this.update(cx, |_, cx| cx.notify()).ok(); - } - Ok(()) - })); - } - /// Subscribe to room events fn subscribe_room_events(&mut self, window: &mut Window, cx: &mut Context) { let Some(room) = self.room.upgrade() else { @@ -513,7 +435,6 @@ impl ChatPanel { cx: &mut Context, ) { let id = rumor.id.expect("rumor must have an id"); - let sent_ids = self.sent_ids.clone(); // Insert optimistic message and clear input if rumor.kind != Kind::Reaction { @@ -528,13 +449,10 @@ impl ChatPanel { // 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; - - let mut sent_ids = sent_ids.lock().await; - sent_ids.extend(outputs.iter().filter_map(|output| output.gift_wrap_id)); + let reports = send_task.await; this.update(cx, |this, cx| { - this.insert_reports(id, outputs, cx); + this.insert_reports(id, reports, cx); })?; Ok(()) @@ -1391,9 +1309,7 @@ impl ChatPanel { fn render_sent_reports(&self, id: &EventId, cx: &App) -> impl IntoElement { let reports = self.sent_reports(id); - let pending = reports - .as_ref() - .is_some_and(|reports| reports.is_empty() || reports.iter().any(|r| r.pending())); + let pending = reports.as_ref().is_some_and(|reports| reports.is_empty()); let success = reports .as_ref() @@ -1462,7 +1378,7 @@ impl ChatPanel { .child(name.clone()), ), ) - .when_some(report.error.clone(), |this, error| { + .when_some(report.error().cloned(), |this, error| { this.child( h_flex() .flex_wrap() @@ -1477,7 +1393,7 @@ impl ChatPanel { .child(div().flex_1().w_full().text_center().child(error)), ) }) - .when_some(report.output.clone(), |this, output| { + .when_some(report.output().cloned(), |this, output| { this.child( v_flex() .gap_2()