From 8bbb4721032ecc651d1cd0c90b69c16651052297 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Mon, 27 Jul 2026 10:14:19 +0700 Subject: [PATCH] optimize --- crates/chat/src/lib.rs | 102 ++++++++++++++----------- crates/device/src/lib.rs | 80 +++++++++---------- crates/state/src/lib.rs | 8 +- crates/workspace/src/dialogs/import.rs | 7 +- crates/workspace/src/lib.rs | 14 ++-- crates/workspace/src/panels/greeter.rs | 12 ++- crates/workspace/src/panels/profile.rs | 7 +- 7 files changed, 123 insertions(+), 107 deletions(-) diff --git a/crates/chat/src/lib.rs b/crates/chat/src/lib.rs index acd5c7f..2b67a0e 100644 --- a/crates/chat/src/lib.rs +++ b/crates/chat/src/lib.rs @@ -1,10 +1,8 @@ use std::cmp::Reverse; use std::collections::{BTreeSet, HashMap, HashSet}; use std::hash::{DefaultHasher, Hash, Hasher}; -use std::sync::Arc; -#[cfg(target_arch = "wasm32")] -use std::sync::RwLock; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, RwLock}; use std::time::Duration; use anyhow::{Context as AnyhowContext, Error, anyhow}; @@ -17,8 +15,6 @@ use gpui::{ }; use nostr_sdk::prelude::*; use smallvec::{SmallVec, smallvec}; -#[cfg(not(target_arch = "wasm32"))] -use smol::lock::RwLock; use state::{DEVICE_GIFTWRAP, NostrRegistry, USER_GIFTWRAP, UniversalSigner}; mod message; @@ -103,6 +99,12 @@ pub struct ChatRegistry { /// 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>>, + /// Subscriptions _subscriptions: SmallVec<[Subscription; 2]>, } @@ -153,12 +155,18 @@ impl ChatRegistry { signal_rx: rx, signal_tx: tx, tasks: smallvec![], + notification_listener: None, + signal_consumer: None, _subscriptions: subscriptions, } } /// Handle nostr notifications fn handle_notifications(&mut self, cx: &mut Context) { + // Cancel previous notification tasks before spawning new ones + self.notification_listener = None; + self.signal_consumer = None; + let nostr = NostrRegistry::global(cx); let client = nostr.read(cx).client(); let signer = nostr.read(cx).signer(); @@ -176,7 +184,7 @@ impl ChatRegistry { let tx = self.signal_tx.clone(); let rx = self.signal_rx.clone(); - self.tasks.push(cx.background_spawn(async move { + self.notification_listener = Some(cx.background_spawn(async move { let mut notifications = client.notifications(); let mut processed_events = HashSet::new(); @@ -209,7 +217,7 @@ impl ChatRegistry { // Keep track of which relays have seen this event { - let mut seens = seens.write().await; + let mut seens = seens.write().unwrap(); seens.entry(event.id).or_default().insert(relay_url); } @@ -217,9 +225,13 @@ impl ChatRegistry { 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; + }; { - let mut event_map = event_map.write().await; - event_map.insert(rumor.id.unwrap(), event.id); + let mut event_map = event_map.write().unwrap(); + event_map.insert(rumor_id, event.id); } // Check if the rumor has a recipient @@ -256,7 +268,7 @@ impl ChatRegistry { Ok(()) })); - self.tasks.push(cx.spawn(async move |this, cx| { + self.signal_consumer = Some(cx.spawn(async move |this, cx| { while let Ok(message) = rx.recv_async().await { match message { Signal::Message(message) => { @@ -287,21 +299,23 @@ impl ChatRegistry { })); } - /// Tracking the status of unwrapping gift wrap events. + /// 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| { - let loop_duration = Duration::from_secs(15); loop { - if status.load(Ordering::Acquire) { - _ = status.compare_exchange(true, false, Ordering::Release, Ordering::Relaxed); - _ = tx.send_async(Signal::Eose).await; - } else { + 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; } - cx.background_executor().timer(loop_duration).await; } })); } @@ -345,25 +359,22 @@ impl ChatRegistry { self.tasks.push(cx.spawn(async move |this, cx| { cx.background_executor().timer(Duration::from_secs(5)).await; - if !cx - .background_spawn(async move { - // Construct inbox relays filter - let filter = Filter::new() - .kind(Kind::InboxRelays) - .author(public_key) - .limit(1); + // Construct inbox relays filter + let filter = Filter::new() + .kind(Kind::InboxRelays) + .author(public_key) + .limit(1); - // Check the latest inbox relays event in database - client - .database() - .query(filter) - .await - .unwrap_or_default() - .first_owned() - .is_some() - }) + // Check the latest inbox relays event in database + let found = client + .database() + .query(filter) .await - { + .unwrap_or_default() + .first_owned() + .is_some(); + + if !found { this.update(cx, |_this, cx| { cx.emit(ChatEvent::InboxRelayNotFound); })?; @@ -469,7 +480,8 @@ impl ChatRegistry { /// Count the number of messages seen by a given relay. pub fn count_messages(&self, relay_url: &RelayUrl) -> usize { self.seens - .read_blocking() + .read() + .unwrap() .values() .filter(|seen| seen.contains(relay_url)) .count() @@ -488,7 +500,8 @@ impl ChatRegistry { /// Get the relays that have seen a given rumor id. pub fn rumor_seen_on(&self, id: &EventId) -> Option> { self.event_map - .read_blocking() + .read() + .unwrap() .get(id) .map(|id| self.seen_on(id)) } @@ -496,7 +509,8 @@ impl ChatRegistry { /// Get the relays that have seen a given gift wrap id. pub fn seen_on(&self, id: &EventId) -> HashSet { self.seens - .read_blocking() + .read() + .unwrap() .get(id) .cloned() .unwrap_or_default() @@ -513,17 +527,17 @@ impl ChatRegistry { return; }; - cx.spawn(async move |this, cx| { + self.tasks.push(cx.spawn(async move |this, cx| { let room: Room = room.into().organize(&public_key); this.update(cx, |this, cx| { this.rooms.insert(0, cx.new(|_| room)); cx.emit(ChatEvent::Ping); cx.notify(); - }) - .ok() - }) - .detach(); + })?; + + Ok(()) + })); } /// Emit an open room event. @@ -635,7 +649,9 @@ impl ChatRegistry { })?; } Err(e) => { - log::error!("Failed to load rooms: {}", e); + this.update(cx, |_, cx| { + cx.emit(ChatEvent::Error(e.to_string())); + })?; } }; diff --git a/crates/device/src/lib.rs b/crates/device/src/lib.rs index 5c06f1c..5694104 100644 --- a/crates/device/src/lib.rs +++ b/crates/device/src/lib.rs @@ -258,28 +258,20 @@ impl DeviceRegistry { })); let announcement_existed = self.announcement_existed.clone(); - let executor = cx.background_executor().clone(); self.tasks.push(cx.spawn(async move |this, cx| { - if !cx - .background_spawn(async move { - // Wait for 5 seconds - executor.timer(Duration::from_secs(5)).await; + // Wait for 5 seconds + cx.background_executor().timer(Duration::from_secs(5)).await; - // Then check if the msg relays have been found - if !announcement_existed.load(Ordering::Acquire) { - return true; - } - - false - }) - .await - { - this.update(cx, |_this, cx| { - cx.emit(DeviceEvent::NotSet); - })?; + // Then check if the msg relays have been found + if announcement_existed.load(Ordering::Acquire) { + return Ok(()); } + this.update(cx, |_this, cx| { + cx.emit(DeviceEvent::NotSet); + })?; + Ok(()) })); } @@ -404,11 +396,10 @@ impl DeviceRegistry { let client = nostr.read(cx).client(); let signer = nostr.read(cx).signer(); - let Ok(app_keys) = get_or_init_app_keys(cx) else { - return; - }; + let app_keys_task = get_or_init_app_keys(cx); let task: Task, Error>> = cx.background_spawn(async move { + let app_keys = app_keys_task.await?; let app_pubkey = app_keys.public_key(); let public_key = signer.get_public_key_async().await?; @@ -489,11 +480,10 @@ impl DeviceRegistry { /// Parse the approval event to get encryption key then set it fn extract_encryption(&mut self, event: Event, cx: &mut Context) { - let Ok(app_keys) = get_or_init_app_keys(cx) else { - return; - }; + let app_keys_task = get_or_init_app_keys(cx); let task: Task> = cx.background_spawn(async move { + let app_keys = app_keys_task.await?; let master = event .tags .iter() @@ -573,7 +563,7 @@ impl DeviceRegistry { Ok(()) }); - cx.spawn_in(window, async move |_this, cx| { + self.tasks.push(cx.spawn_in(window, async move |_this, cx| { match task.await { Ok(_) => { cx.update(|window, cx| { @@ -591,8 +581,9 @@ impl DeviceRegistry { .ok(); } }; - }) - .detach(); + + Ok(()) + })); } /// Handle encryption request @@ -715,33 +706,34 @@ impl DeviceRegistry { struct DeviceNotification; -/// Get or create new app keys -fn get_or_init_app_keys(cx: &App) -> Result { +/// Get or create new app keys (async, returns a task) +fn get_or_init_app_keys(cx: &App) -> Task> { let read = cx.read_credentials(CLIENT_NAME); - let stored_keys: Option = cx.foreground_executor().block_on(async move { - if let Ok(Some((_, secret))) = read.await { - SecretKey::from_slice(&secret).map(Keys::new).ok() - } else { - None - } - }); - if let Some(keys) = stored_keys { - Ok(keys) - } else { + cx.spawn(async move |cx| { + if let Ok(Some((_, secret))) = read.await + && let Ok(keys) = SecretKey::from_slice(&secret).map(Keys::new) + { + return Ok(keys); + } + + // No stored keys found or invalid — generate new ones let keys = Keys::generate(); let user = keys.public_key().to_hex(); let secret = keys.secret_key().to_secret_bytes(); - let write = cx.write_credentials(CLIENT_NAME, &user, &secret); - cx.foreground_executor().block_on(async move { - if let Err(e) = write.await { - log::error!("Keyring not available or panic: {e}") - } + cx.update(|cx| { + let write = cx.write_credentials(CLIENT_NAME, &user, &secret); + cx.background_spawn(async move { + if let Err(e) = write.await { + log::error!("Keyring not available or panic: {e}") + } + }) + .detach(); }); Ok(keys) - } + }) } /// Encrypt and store device keys in the local database. diff --git a/crates/state/src/lib.rs b/crates/state/src/lib.rs index 811d3f1..929aa0d 100644 --- a/crates/state/src/lib.rs +++ b/crates/state/src/lib.rs @@ -164,7 +164,7 @@ impl NostrRegistry { ::Error: std::error::Error + Send + Sync + 'static, ::Error: std::error::Error + Send + Sync + 'static, { - cx.spawn(async move |this, cx| { + let task = cx.spawn(async move |this, cx| { match new_signer.get_public_key_async().await { Ok(public_key) => { this.update(cx, |this, cx| { @@ -181,9 +181,9 @@ impl NostrRegistry { } }; - Ok::<(), anyhow::Error>(()) - }) - .detach(); + Ok(()) + }); + self.tasks.push(task); } /// Connect to the bootstrapping relays diff --git a/crates/workspace/src/dialogs/import.rs b/crates/workspace/src/dialogs/import.rs index fedc53e..37a6763 100644 --- a/crates/workspace/src/dialogs/import.rs +++ b/crates/workspace/src/dialogs/import.rs @@ -146,7 +146,7 @@ impl ImportIdentity { let password = uri.to_string(); let save = cx.write_credentials(USER_KEYRING, "bunker", password.as_bytes()); - cx.spawn_in(window, async move |_this, cx| { + self.tasks.push(cx.spawn_in(window, async move |_this, cx| { let keys = master_keys.await; let timeout = Duration::from_secs(30); @@ -162,9 +162,8 @@ impl ImportIdentity { cx.notify(); }); - Ok::<(), anyhow::Error>(()) - }) - .detach(); + Ok(()) + })); } fn set_loading(&mut self, status: bool, cx: &mut Context) { diff --git a/crates/workspace/src/lib.rs b/crates/workspace/src/lib.rs index 6d80c60..6ccb5e5 100644 --- a/crates/workspace/src/lib.rs +++ b/crates/workspace/src/lib.rs @@ -1,13 +1,14 @@ use std::sync::Arc; use ::settings::AppSettings; +use anyhow::Error; use chat::{ChatEvent, ChatRegistry}; use common::{CoopImageCache, download_dir}; use device::{DeviceEvent, DeviceRegistry}; use gpui::prelude::FluentBuilder; use gpui::{ Action, App, AppContext, Axis, Context, Entity, InteractiveElement, IntoElement, ParentElement, - Render, SharedString, StatefulInteractiveElement, Styled, Subscription, Window, div, + Render, SharedString, StatefulInteractiveElement, Styled, Subscription, Task, Window, div, image_cache, px, relative, }; use nostr_sdk::prelude::*; @@ -65,6 +66,9 @@ pub struct Workspace { /// App's Image Cache image_cache: Entity, + /// Async tasks + tasks: Vec>>, + /// Event subscriptions _subscriptions: SmallVec<[Subscription; 6]>, } @@ -245,6 +249,7 @@ impl Workspace { Self { dock, image_cache, + tasks: vec![], _subscriptions: subscriptions, } } @@ -390,7 +395,7 @@ impl Workspace { let device = DeviceRegistry::global(cx).downgrade(); let save_dialog = cx.prompt_for_new_path(download_dir(), Some("encryption.txt")); - cx.spawn_in(window, async move |_this, cx| { + self.tasks.push(cx.spawn_in(window, async move |_this, cx| { // Get the output path from the save dialog let output_path = match save_dialog.await { Ok(Ok(Some(path))) => path, @@ -417,9 +422,8 @@ impl Workspace { cx.open_with_system(output_path.as_path()); })?; - Ok::<_, anyhow::Error>(()) - }) - .detach(); + Ok(()) + })); } Command::ImportEncryption => { self.import_encryption(window, cx); diff --git a/crates/workspace/src/panels/greeter.rs b/crates/workspace/src/panels/greeter.rs index 69ddad0..3be32d2 100644 --- a/crates/workspace/src/panels/greeter.rs +++ b/crates/workspace/src/panels/greeter.rs @@ -1,6 +1,7 @@ +use anyhow::Error; use gpui::{ AnyElement, App, AppContext, Context, Entity, EventEmitter, FocusHandle, Focusable, - IntoElement, ParentElement, Render, SharedString, Styled, Window, div, svg, + IntoElement, ParentElement, Render, SharedString, Styled, Task, Window, div, svg, }; use state::NostrRegistry; use theme::ActiveTheme; @@ -18,6 +19,7 @@ pub fn init(window: &mut Window, cx: &mut App) -> Entity { pub struct GreeterPanel { name: SharedString, focus_handle: FocusHandle, + tasks: Vec>>, } impl GreeterPanel { @@ -25,6 +27,7 @@ impl GreeterPanel { Self { name: "Onboarding".into(), focus_handle: cx.focus_handle(), + tasks: vec![], } } @@ -32,7 +35,7 @@ impl GreeterPanel { let nostr = NostrRegistry::global(cx); if let Some(public_key) = nostr.read(cx).current_user() { - cx.spawn_in(window, async move |_this, cx| { + self.tasks.push(cx.spawn_in(window, async move |_this, cx| { cx.update(|window, cx| { Workspace::add_panel( profile::init(public_key, window, cx), @@ -42,8 +45,9 @@ impl GreeterPanel { ); }) .ok(); - }) - .detach(); + + Ok(()) + })); } } } diff --git a/crates/workspace/src/panels/profile.rs b/crates/workspace/src/panels/profile.rs index d8da358..0ef620e 100644 --- a/crates/workspace/src/panels/profile.rs +++ b/crates/workspace/src/panels/profile.rs @@ -132,7 +132,7 @@ impl ProfilePanel { cx.notify(); if status { - cx.spawn_in(window, async move |this, cx| { + self.tasks.push(cx.spawn_in(window, async move |this, cx| { cx.background_executor().timer(Duration::from_secs(2)).await; // Reset the copied state after a delay @@ -143,8 +143,9 @@ impl ProfilePanel { .ok(); }) .ok(); - }) - .detach(); + + Ok(()) + })); } }