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

.

Browse Source
This commit is contained in:
Ren Amamiya
2026-07-27 15:02:50 +07:00
parent b349656c56
commit 2ff83079b8
9 changed files with 167 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -82,15 +82,15 @@ pub struct ChatRegistry {
rooms: Vec<Entity<Room>>,
/// Events that failed to unwrap for any reason
trashes: Entity<BTreeSet<FailedMessage>>,
trash: Entity<BTreeSet<FailedMessage>>,
/// Tracking events seen on which relays in the current session
seens: Arc<RwLock<HashMap<EventId, HashSet<RelayUrl>>>>,
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>>>,
/// Tracking the status of unwrapping gift wrap events.
/// True while the initial event backlog is still loading
tracking: Arc<AtomicBool>,
/// Channel for sending signals to the UI.
@@ -108,10 +108,31 @@ pub struct ChatRegistry {
/// 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]>,
}
/// Wrapper to provide Debug for SkimMatcherV2
struct CachedMatcher(SkimMatcherV2);
impl std::fmt::Debug for CachedMatcher {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("CachedMatcher { .. }")
}
}
impl std::ops::Deref for CachedMatcher {
type Target = SkimMatcherV2;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl EventEmitter<ChatEvent> for ChatRegistry {}
impl ChatRegistry {
@@ -145,16 +166,16 @@ impl ChatRegistry {
// Run at the end of the current cycle
cx.defer_in(window, |this, _window, cx| {
this.tracking(cx);
this.get_rooms(cx);
});
Self {
rooms: vec![],
trashes: cx.new(|_| BTreeSet::default()),
seens: Arc::new(RwLock::new(HashMap::default())),
trash: cx.new(|_| BTreeSet::default()),
seen: Arc::new(RwLock::new(HashMap::default())),
event_map: Arc::new(RwLock::new(HashMap::default())),
tracking: Arc::new(AtomicBool::new(false)),
tracking: Arc::new(AtomicBool::new(true)),
matcher: CachedMatcher(SkimMatcherV2::default()),
signal_rx: rx,
signal_tx: tx,
tasks: smallvec![],
@@ -174,12 +195,10 @@ impl ChatRegistry {
let client = nostr.read(cx).client();
let signer = nostr.read(cx).signer();
let tracking = self.tracking.clone();
let seens = self.seens.clone();
let seen = self.seen.clone();
let event_map = self.event_map.clone();
let trashes = self.trashes.downgrade();
let trash = self.trash.downgrade();
let initialized_at = Timestamp::now();
let sub_id1 = SubscriptionId::new(DEVICE_GIFTWRAP);
let sub_id2 = SubscriptionId::new(USER_GIFTWRAP);
@@ -190,16 +209,19 @@ impl ChatRegistry {
self.notification_listener = Some(cx.background_spawn(async move {
let mut notifications = client.notifications();
let mut processed_events = HashSet::new();
const MAX_PROCESSED: usize = 10_000;
while let Some(notification) = notifications.next().await {
let ClientNotification::Message { message, relay_url } = notification else {
// Skip non-message notifications
continue;
};
match *message {
RelayMessage::Event { event, .. } => {
// De-duplicate events by their ID
// Prune the dedup set before it grows unbounded
if processed_events.len() >= MAX_PROCESSED {
processed_events.clear();
}
if !processed_events.insert(event.id) {
continue;
}
@@ -207,7 +229,6 @@ impl ChatRegistry {
// Handle msg relays event to determine when the app is ready to subscribe
if event.kind == Kind::InboxRelays {
let current_user = signer.get_public_key_async().await?;
// Emit the inbox ready signal
if event.pubkey == current_user {
tx.send_async(Signal::InboxReady).await?;
}
@@ -220,14 +241,13 @@ impl ChatRegistry {
// Keep track of which relays have seen this event
{
let mut seens = seens.write().unwrap();
seens.entry(event.id).or_default().insert(relay_url);
let mut seen = seen.write().unwrap();
seen.entry(event.id).or_default().insert(relay_url);
}
// Extract the rumor from the gift wrap event
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;
@@ -237,20 +257,14 @@ impl ChatRegistry {
event_map.insert(rumor_id, event.id);
}
// Check if the rumor has a recipient
if rumor.tags.is_empty() {
let signal = Signal::error(&event, "Recipient is missing");
tx.send_async(signal).await?;
}
// Check if the rumor was created after the chat was initialized (for detecting new messages)
if rumor.created_at >= initialized_at {
let signal = Signal::message(event.id, rumor);
tx.send_async(signal).await?;
} else {
// Mark the chat still processing new messages
tracking.store(true, Ordering::Release);
}
// Emit message for both new and backlog events
let signal = Signal::message(event.id, rumor);
tx.send_async(signal).await?;
}
Err(e) => {
let reason = format!("Failed to extract rumor: {e}");
@@ -286,12 +300,13 @@ impl ChatRegistry {
}
Signal::Eose => {
this.update(cx, |this, cx| {
this.get_rooms(cx);
this.tracking.store(false, Ordering::Release);
this.sort(cx);
})?;
}
Signal::Error(trash) => {
trashes.update(cx, |this, cx| {
this.insert(trash);
Signal::Error(failed) => {
trash.update(cx, |this, cx| {
this.insert(failed);
cx.notify();
})?;
}
@@ -302,27 +317,6 @@ impl ChatRegistry {
}));
}
/// 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| {
loop {
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;
}
}
}));
}
/// Get all necessary metadata from relays for current user
pub fn get_metadata(&mut self, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
@@ -332,37 +326,29 @@ impl ChatRegistry {
return;
};
let subscribe = cx.background_spawn({
let client = client.clone();
async move {
let opts =
SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE);
let msg_relays = Filter::new()
.kind(Kind::InboxRelays)
.author(public_key)
.limit(1);
let contact_list = Filter::new()
.kind(Kind::ContactList)
.author(public_key)
.limit(1);
client
.subscribe(vec![msg_relays, contact_list])
.close_on(opts)
.await
}
});
self.tasks.push(cx.spawn(async move |this, cx| {
// Wait for subscription to complete (or fail silently)
_ = subscribe.await;
// Subscribe to metadata from relays
let opts = SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE);
let msg_relays = Filter::new()
.kind(Kind::InboxRelays)
.author(public_key)
.limit(1);
let contact_list = Filter::new()
.kind(Kind::ContactList)
.author(public_key)
.limit(1);
_ = client
.subscribe(vec![msg_relays, contact_list])
.close_on(opts)
.await;
// Give relays time to respond
cx.background_executor().timer(Duration::from_secs(5)).await;
// Verify inbox relays were received
let filter = Filter::new()
.kind(Kind::InboxRelays)
.author(public_key)
@@ -436,8 +422,8 @@ impl ChatRegistry {
}));
}
/// Refresh the chat registry, fetching messages and contact list from relays.
pub fn refresh(&mut self, cx: &mut Context<Self>) {
/// Reload the chat registry, fetching messages and contact list from relays.
pub fn reload(&mut self, cx: &mut Context<Self>) {
self.reset(cx);
self.get_metadata(cx);
self.get_rooms(cx);
@@ -475,22 +461,22 @@ impl ChatRegistry {
/// Count the number of messages seen by a given relay.
pub fn count_messages(&self, relay_url: &RelayUrl) -> usize {
self.seens
self.seen
.read()
.unwrap()
.values()
.filter(|seen| seen.contains(relay_url))
.filter(|s| s.contains(relay_url))
.count()
}
/// Count the number of trash messages.
pub fn count_trash_messages(&self, cx: &App) -> usize {
self.trashes.read(cx).len()
self.trash.read(cx).len()
}
/// Get the trash messages entity.
pub fn trashes(&self) -> Entity<BTreeSet<FailedMessage>> {
self.trashes.clone()
pub fn trash(&self) -> Entity<BTreeSet<FailedMessage>> {
self.trash.clone()
}
/// Get the relays that have seen a given rumor id.
@@ -504,7 +490,7 @@ impl ChatRegistry {
/// Get the relays that have seen a given gift wrap id.
pub fn seen_on(&self, id: &EventId) -> HashSet<RelayUrl> {
self.seens
self.seen
.read()
.unwrap()
.get(id)
@@ -560,8 +546,6 @@ impl ChatRegistry {
/// Finding rooms based on a query.
pub fn find(&self, query: &str, cx: &App) -> Vec<Entity<Room>> {
let matcher = SkimMatcherV2::default();
if let Ok(public_key) = PublicKey::parse(query) {
self.rooms
.iter()
@@ -572,7 +556,7 @@ impl ChatRegistry {
self.rooms
.iter()
.filter(|room| {
matcher
self.matcher
.fuzzy_match(room.read(cx).display_name(cx).as_ref(), query)
.is_some()
})
@@ -584,7 +568,7 @@ impl ChatRegistry {
/// Reset the registry.
pub fn reset(&mut self, cx: &mut Context<Self>) {
self.rooms.clear();
self.trashes.update(cx, |this, cx| {
self.trash.update(cx, |this, cx| {
this.clear();
cx.notify();
});