reya/coop
1
0
Fork 1
Code Issues Pull Requests 1 Releases 3 Activity

refactor send

Browse Source
This commit is contained in:
reya
2026-09-30 04:40:02 +07:00
parent be15873e0d
commit 7dbf55d48e
5 changed files with 121 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
-2
View File
@@ -1097,8 +1097,6 @@ dependencies = [
"anyhow",
"chat",
"common",
"flume 0.11.1",
"futures",
"gpui-pre",
"itertools 0.13.0",
"log",
crates/chat/src/lib.rs
+6 -18
View File
@@ -73,41 +73,29 @@ impl Signal {
pub struct ChatRegistry {
/// Chat rooms
rooms: Vec<Entity<Room>>,
/// O(1) room lookup by room ID
room_index: HashMap<u64, Entity<Room>>,
/// Events that failed to unwrap for any reason
trash: Entity<BTreeSet<FailedMessage>>,
/// Tracking events seen on which relays in the current session
seen: Arc<RwLock<HashMap<EventId, HashSet<RelayUrl>>>>,
/// Mapping of unwrapped event ids to their gift wrap event ids
event_map: Arc<RwLock<HashMap<EventId, EventId>>>,
/// True while the initial event backlog is still loading
tracking: Arc<AtomicBool>,
/// Channel for sending signals to the UI.
signal_tx: flume::Sender<Signal>,
/// Channel for receiving signals from the UI.
signal_rx: flume::Receiver<Signal>,
/// Async tasks
tasks: SmallVec<[Task<Result<(), Error>>; 2]>,
/// Notification listener task (cancelled on signer change)
notification_listener: Option<Task<Result<(), Error>>>,
/// Signal consumer task (cancelled on signer change)
signal_consumer: Option<Task<Result<(), Error>>>,
/// 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>) -> Self {
let entity = cx.entity().downgrade();
let nostr = NostrRegistry::global(cx);
let entity = cx.entity().downgrade();
let (tx, rx) = flume::unbounded::<Signal>();
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;
}
crates/chat/src/room.rs
+109 -153
View File
@@ -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<EventId>,
pub error: Option<SharedString>,
pub output: Option<Output<EventId, EventSendStatus>>,
/// The per-relay send output, or the error that prevented the send.
result: Result<Output<EventId, EventSendStatus>, 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<EventId, EventSendStatus>> {
self.result.as_ref().ok()
}
/// Set the output.
pub fn output(mut self, output: Output<EventId, EventSendStatus>) -> Self {
self.output = Some(output);
self
}
/// Set the error message.
pub fn error<T>(mut self, error: T) -> Self
where
T: Into<SharedString>,
{
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<Task<Vec<SendReport>>> {
let config = self.config.clone();
@@ -560,7 +514,7 @@ impl Room {
let sender = persons.read(cx).get(&current_user, cx);
// Get all members (excluding sender)
let members: Vec<Person> = self
let receivers: Vec<Person> = 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<SendReport> = 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<UniversalSigner>,
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<SendReport, Error> {
let k_tag = Tag::custom("k", [rumor.kind.to_string()]);
let mut extra_tags = vec![k_tag];
) -> Result<Output<EventId, EventSendStatus>, 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)
}
crates/chat_ui/Cargo.toml
-2
View File
@@ -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
crates/chat_ui/src/lib.rs
+6 -90
View File
@@ -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<bool>,
/// Sent message ids
sent_ids: Arc<Mutex<Vec<EventId>>>,
/// Replies to
replies_to: Entity<HashSet<EventId>>,
@@ -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<Self>) {
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::<Arc<SendStatus>>(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<Self>) {
let Some(room) = self.room.upgrade() else {
@@ -513,7 +435,6 @@ impl ChatPanel {
cx: &mut Context<Self>,
) {
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()