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

.

Browse Source
This commit is contained in:
reya
2026-01-07 14:05:24 +07:00
parent 2d9563b888
commit 967355cd4a
12 changed files with 511 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -6,7 +6,7 @@ use std::sync::Arc;
use std::time::Duration;
use anyhow::{anyhow, Context as AnyhowContext, Error};
use common::{EventUtils, BOOTSTRAP_RELAYS, METADATA_BATCH_LIMIT};
use common::EventUtils;
use flume::Sender;
use fuzzy_matcher::skim::SkimMatcherV2;
use fuzzy_matcher::FuzzyMatcher;
@@ -154,7 +154,6 @@ impl ChatRegistry {
let subscription_id = SubscriptionId::new(GIFTWRAP_SUBSCRIPTION);
let mut notifications = client.notifications();
let mut public_keys = HashSet::new();
let mut processed_events = HashSet::new();
while let Ok(notification) = notifications.recv().await {
@@ -177,41 +176,26 @@ impl ChatRegistry {
// Extract the rumor from the gift wrap event
match Self::extract_rumor(client, event.as_ref()).await {
Ok(rumor) => {
// Get all public keys
public_keys.extend(rumor.all_pubkeys());
Ok(rumor) => match rumor.created_at >= initialized_at {
true => {
let sent_by_coop = {
let tracker = tracker().read().await;
tracker.is_sent_by_coop(&event.id)
};
let limit_reached = public_keys.len() >= METADATA_BATCH_LIMIT;
let done = !loading.load(Ordering::Acquire) && !public_keys.is_empty();
if !sent_by_coop {
let new_message = NewMessage::new(event.id, rumor);
let signal = NostrEvent::Message(new_message);
// Get metadata for all public keys if the limit is reached
if limit_reached || done {
let public_keys = std::mem::take(&mut public_keys);
// Get metadata for the public keys
Self::get_metadata(client, public_keys).await.ok();
}
match rumor.created_at >= initialized_at {
true => {
let sent_by_coop = {
let tracker = tracker().read().await;
tracker.is_sent_by_coop(&event.id)
};
if !sent_by_coop {
let new_message = NewMessage::new(event.id, rumor);
let signal = NostrEvent::Message(new_message);
if let Err(e) = tx.send_async(signal).await {
log::error!("Failed to send signal: {}", e);
}
if let Err(e) = tx.send_async(signal).await {
log::error!("Failed to send signal: {}", e);
}
}
false => {
loading.store(true, Ordering::Release);
}
}
}
false => {
loading.store(true, Ordering::Release);
}
},
Err(e) => {
log::warn!("Failed to unwrap gift wrap event: {}", e);
}
@@ -230,7 +214,7 @@ impl ChatRegistry {
}
async fn unwrapping_status(client: &Client, status: &Arc<AtomicBool>, tx: &Sender<NostrEvent>) {
let loop_duration = Duration::from_secs(20);
let loop_duration = Duration::from_secs(12);
let mut is_start_processing = false;
let mut total_loops = 0;
@@ -240,22 +224,16 @@ impl ChatRegistry {
if status.load(Ordering::Acquire) {
is_start_processing = true;
// Reset gift wrap processing flag
_ = status.compare_exchange(true, false, Ordering::Release, Ordering::Relaxed);
// Send loading signal
if let Err(e) = tx.send_async(NostrEvent::Unwrapping(true)).await {
log::error!("Failed to send signal: {}", e);
}
tx.send_async(NostrEvent::Unwrapping(true)).await.ok();
} else {
// Only run further if we are already processing
// Wait until after 2 loops to prevent exiting early while events are still being processed
if is_start_processing && total_loops >= 2 {
// Send loading signal
if let Err(e) = tx.send_async(NostrEvent::Unwrapping(false)).await {
log::error!("Failed to send signal: {}", e);
}
tx.send_async(NostrEvent::Unwrapping(false)).await.ok();
// Reset the counter
is_start_processing = false;
total_loops = 0;
@@ -654,33 +632,6 @@ impl ChatRegistry {
}
}
/// Get metadata for a list of public keys
async fn get_metadata<I>(client: &Client, public_keys: I) -> Result<(), Error>
where
I: IntoIterator<Item = PublicKey>,
{
let authors: Vec<PublicKey> = public_keys.into_iter().collect();
let opts = SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE);
let kinds = vec![Kind::Metadata, Kind::ContactList];
// Return if the list is empty
if authors.is_empty() {
return Err(anyhow!("You need at least one public key".to_string(),));
}
let filter = Filter::new()
.limit(authors.len() * kinds.len())
.authors(authors)
.kinds(kinds);
// Subscribe to filters to the bootstrap relays
client
.subscribe_to(BOOTSTRAP_RELAYS, filter, Some(opts))
.await?;
Ok(())
}
/// Get the conversation ID for a given rumor (message).
fn conversation_id(rumor: &UnsignedEvent) -> u64 {
let mut hasher = DefaultHasher::new();

25
crates/chat/src/room.rs
View File

@@ -147,7 +147,7 @@ impl From<&UnsignedEvent> for Room {
let created_at = val.created_at;
// Get the members from the event's tags and event's pubkey
let members = val.all_pubkeys();
let members = val.extract_public_keys();
// Get subject from tags
let subject = val
@@ -228,16 +228,25 @@ impl Room {
}
/// Returns the members of the room with their messaging relays
pub fn members_with_relays(&self, cx: &App) -> Vec<(PublicKey, Vec<RelayUrl>)> {
pub fn members_with_relays(&self, cx: &App) -> Task<Vec<(PublicKey, Vec<RelayUrl>)>> {
let nostr = NostrRegistry::global(cx);
let mut result = vec![];
let mut tasks = vec![];
for member in self.members.iter() {
let messaging_relays = nostr.read(cx).messaging_relays(member, cx);
result.push((member.to_owned(), messaging_relays));
let task = nostr.read(cx).messaging_relays(member, cx);
tasks.push((*member, task));
}
result
cx.background_spawn(async move {
let mut results = vec![];
for (public_key, task) in tasks.into_iter() {
let urls = task.await;
results.push((public_key, urls));
}
results
})
}
/// Checks if the room has more than two members (group)
@@ -457,10 +466,12 @@ impl Room {
let rumor = rumor.to_owned();
// Get all members and their messaging relays
let mut members = self.members_with_relays(cx);
let task = self.members_with_relays(cx);
cx.background_spawn(async move {
let signer = client.signer().await?;
let current_user_relays = current_user_relays.await;
let mut members = task.await;
// Remove the current user's public key from the list of receivers
// the current user will be handled separately