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

optimize

Browse Source
This commit is contained in:
Ren Amamiya
2026-07-27 10:14:19 +07:00
parent dfda7ff157
commit 8bbb472103
7 changed files with 123 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
crates/chat/src/lib.rs
View File

@@ -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<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>>>,
/// 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<Self>) {
// 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<Self>) {
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<HashSet<RelayUrl>> {
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<RelayUrl> {
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()));
})?;
}
};