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

.

Browse Source
This commit is contained in:
Ren Amamiya
2026-07-27 10:46:10 +07:00
parent 8bbb472103
commit b349656c56
3 changed files with 168 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2,7 +2,7 @@ use std::cmp::Reverse;
use std::collections::{BTreeSet, HashMap, HashSet};
use std::hash::{DefaultHasher, Hash, Hasher};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, RwLock};
use std::sync::{Arc, LazyLock, RwLock};
use std::time::Duration;
use anyhow::{Context as AnyhowContext, Error, anyhow};
@@ -23,6 +23,9 @@ mod room;
pub use message::*;
pub use room::*;
/// A static keypair used only for signing locally-cached rumor events.
static LOCAL_KEYS: LazyLock<Keys> = LazyLock::new(Keys::generate);
pub fn init(window: &mut Window, cx: &mut App) {
ChatRegistry::set_global(cx.new(|cx| ChatRegistry::new(window, cx)), cx);
}
@@ -329,43 +332,42 @@ impl ChatRegistry {
return;
};
self.tasks.push(cx.background_spawn(async move {
let opts = SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE);
let subscribe = cx.background_spawn({
let client = client.clone();
async move {
let opts =
SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE);
// Construct filter for msg relays
let msg_relays = Filter::new()
.kind(Kind::InboxRelays)
.author(public_key)
.limit(1);
// Construct filter for contact list
let contact_list = Filter::new()
.kind(Kind::ContactList)
.author(public_key)
.limit(1);
// Subscribe
client
.subscribe(vec![msg_relays, contact_list])
.close_on(opts)
.await?;
.await
}
});
Ok(())
}));
let client = nostr.read(cx).client();
// Spawn a task to verify user inbox relays after 5 seconds
self.tasks.push(cx.spawn(async move |this, cx| {
// Wait for subscription to complete (or fail silently)
_ = subscribe.await;
// Give relays time to respond
cx.background_executor().timer(Duration::from_secs(5)).await;
// Construct inbox relays filter
let filter = Filter::new()
.kind(Kind::InboxRelays)
.author(public_key)
.limit(1);
// Check the latest inbox relays event in database
let found = client
.database()
.query(filter)
@@ -390,16 +392,15 @@ impl ChatRegistry {
let client = nostr.read(cx).client();
let signer = nostr.read(cx).signer();
self.tasks.push(cx.spawn(async move |this, cx| {
let task: Task<Result<(), Error>> = cx.background_spawn(async move {
let public_key = signer.get_public_key_async().await?;
// Construct inbox relays filter
let filter = Filter::new()
.kind(Kind::InboxRelays)
.author(public_key)
.limit(1);
// Get the latest inbox relays event in database
let event = client
.database()
.query(filter)
@@ -407,19 +408,14 @@ impl ChatRegistry {
.first_owned()
.ok_or(anyhow::anyhow!("No inbox relays found"))?;
// Extract relay list from event
let relays: Vec<RelayUrl> = nip17::extract_relay_list(&event).collect();
// Ensure relay connections
for url in relays.iter() {
client.add_relay(url).and_connect().await?;
}
// Construct gift wrap event filter
let filter = Filter::new().kind(Kind::GiftWrap).pubkey(public_key);
let id = SubscriptionId::new(format!("{}-msg", public_key.to_hex()));
// Construct target for subscription
let target: HashMap<RelayUrl, Filter> = relays
.into_iter()
.map(|relay| (relay, filter.clone()))
@@ -430,12 +426,12 @@ impl ChatRegistry {
Ok(())
});
self.tasks.push(cx.spawn(async move |this, cx| {
if let Err(e) = task.await {
this.update(cx, |_this, cx| {
cx.emit(ChatEvent::Error(e.to_string()));
})?;
}
Ok(())
}));
}
@@ -519,25 +515,18 @@ impl ChatRegistry {
/// Add a new room to the start of list.
pub fn add_room<I>(&mut self, room: I, cx: &mut Context<Self>)
where
I: Into<Room> + 'static,
I: Into<Room>,
{
let nostr = NostrRegistry::global(cx);
let Some(public_key) = nostr.read(cx).current_user() else {
return;
};
self.tasks.push(cx.spawn(async move |this, cx| {
let room: Room = room.into().organize(&public_key);
self.rooms.insert(0, cx.new(|_| room));
this.update(cx, |this, cx| {
this.rooms.insert(0, cx.new(|_| room));
cx.emit(ChatEvent::Ping);
cx.notify();
})?;
Ok(())
}));
}
/// Emit an open room event.
@@ -791,7 +780,7 @@ async fn extract_rumor(
}
// Try to unwrap with the available signer
let unwrapped = try_unwrap(signer, gift_wrap).await?;
let unwrapped = try_unwrap_with(signer, gift_wrap).await?;
let mut rumor = unwrapped.rumor;
// Generate event id for the rumor if it doesn't have one
@@ -805,25 +794,6 @@ async fn extract_rumor(
Ok(rumor)
}
/// Helper method to try unwrapping with different signers
async fn try_unwrap(signer: &UniversalSigner, gift_wrap: &Event) -> Result<UnwrappedGift, Error> {
/*
* // Try with the device signer first
if let Some(signer) = signer.get_encryption_signer().await {
log::info!("trying with encryption key");
if let Ok(unwrapped) = try_unwrap_with(gift_wrap, &signer).await {
return Ok(unwrapped);
}
}
// Fallback to the user's signer
let user_signer = signer.get().await;
*/
let unwrapped = try_unwrap_with(signer, gift_wrap).await?;
Ok(unwrapped)
}
/// Attempts to unwrap a gift wrap event with a given signer.
async fn try_unwrap_with(
signer: &UniversalSigner,
@@ -882,7 +852,7 @@ async fn set_rumor(client: &Client, id: EventId, rumor: &UnsignedEvent) -> Resul
// Construct the event
let event = EventBuilder::new(Kind::ApplicationSpecificData, content)
.tags(tags)
.finalize_async(&Keys::generate())
.finalize_async(&*LOCAL_KEYS)
.await?;
// Save the event to the database

104
crates/chat_ui/src/lib.rs
View File

@@ -1,7 +1,7 @@
use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::sync::Arc;
#[cfg(target_arch = "wasm32")]
use std::sync::RwLock;
use std::sync::RwLock as AsyncRwLock;
use std::sync::{Arc, RwLock};
pub use actions::*;
use anyhow::{Context as AnyhowContext, Error};
@@ -21,7 +21,7 @@ use person::{Person, PersonRegistry};
use settings::{AppSettings, SignerKind};
use smallvec::{SmallVec, smallvec};
#[cfg(not(target_arch = "wasm32"))]
use smol::lock::RwLock;
use smol::lock::RwLock as AsyncRwLock;
use state::{NostrRegistry, upload};
use theme::ActiveTheme;
use ui::avatar::Avatar;
@@ -63,7 +63,7 @@ pub struct ChatPanel {
rendered_texts_by_id: BTreeMap<EventId, RenderedText>,
/// Mapping message (rumor event) ids to their reports
reports_by_id: Entity<BTreeMap<EventId, Vec<SendReport>>>,
reports_by_id: Arc<RwLock<BTreeMap<EventId, Vec<SendReport>>>>,
/// Chat input state
input: Entity<InputState>,
@@ -75,7 +75,7 @@ pub struct ChatPanel {
subject_bar: Entity<bool>,
/// Sent message ids
sent_ids: Arc<RwLock<Vec<EventId>>>,
sent_ids: Arc<AsyncRwLock<Vec<EventId>>>,
/// Replies to
replies_to: Entity<HashSet<EventId>>,
@@ -98,7 +98,7 @@ impl ChatPanel {
// Define attachments and replies_to entities
let attachments = cx.new(|_| vec![]);
let replies_to = cx.new(|_| HashSet::new());
let reports_by_id = cx.new(|_| BTreeMap::new());
let reports_by_id = Arc::new(RwLock::new(BTreeMap::new()));
// Define list of messages
let messages = BTreeSet::default();
@@ -172,7 +172,7 @@ impl ChatPanel {
attachments,
rendered_texts_by_id: BTreeMap::new(),
reports_by_id,
sent_ids: Arc::new(RwLock::new(Vec::new())),
sent_ids: Arc::new(AsyncRwLock::new(Vec::new())),
uploading: false,
subscriptions,
tasks: vec![],
@@ -192,7 +192,7 @@ impl ChatPanel {
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let sent_ids = self.sent_ids.clone();
let reports = self.reports_by_id.downgrade();
let reports = self.reports_by_id.clone();
let (tx, rx) = flume::bounded::<Arc<SendStatus>>(256);
@@ -223,10 +223,11 @@ impl ChatPanel {
Ok(())
}));
self.tasks.push(cx.spawn(async move |_this, cx| {
self.tasks.push(cx.spawn(async move |this, cx| {
while let Ok(status) = rx.recv_async().await {
reports.update(cx, |this, cx| {
for reports in this.values_mut() {
{
let mut map = reports.write().unwrap();
for reports in map.values_mut() {
for report in reports.iter_mut() {
let Some(output) = report.output.as_mut() else {
continue;
@@ -243,10 +244,10 @@ impl ChatPanel {
}
}
}
cx.notify();
}
}
})?;
}
this.update(cx, |_, cx| cx.notify()).ok();
}
Ok(())
}));
@@ -345,69 +346,46 @@ impl ChatPanel {
return;
}
// Get room entity
let room = self.room.clone();
// Get content and replies
let replies: Vec<EventId> = self.replies_to.read(cx).iter().copied().collect();
let content = value.to_string();
self.tasks.push(cx.spawn_in(window, async move |this, cx| {
let room = room.upgrade().context("Room is not available")?;
this.update_in(cx, |this, window, cx| {
match room.read(cx).rumor(content, replies, cx) {
Some(rumor) => {
this.insert_message(&rumor, true, cx);
this.send_and_wait(rumor, window, cx);
this.clear(window, cx);
}
None => {
window.push_notification("Failed to create message", cx);
}
}
})?;
Ok(())
}));
}
/// Send message in the background and wait for the response
fn send_and_wait(&mut self, rumor: UnsignedEvent, window: &mut Window, cx: &mut Context<Self>) {
let sent_ids = self.sent_ids.clone();
// This can't fail, because we already ensured that the ID is set
let id = rumor.id.unwrap();
// Add empty reports
self.insert_reports(id, vec![], cx);
// Upgrade room reference
let Some(room) = self.room.upgrade() else {
// Upgrade room and create the rumor synchronously
let Some(room_entity) = room.upgrade() else {
return;
};
let Some(rumor) = room_entity.read(cx).rumor(content.clone(), replies, cx) else {
window.push_notification("Failed to create message", cx);
return;
};
// Get the send message task
let Some(task) = room.read(cx).send(rumor, cx) else {
let id = rumor.id.expect("rumor must have an id");
// Insert optimistic message and clear input
self.insert_message(&rumor, true, cx);
self.insert_reports(id, vec![], cx);
self.clear(window, cx);
// Get the send task
let Some(send_task) = room_entity.read(cx).send(rumor, cx) else {
window.push_notification("Failed to send message", cx);
return;
};
// Spawn a single task to await the send and update reports
self.tasks.push(cx.spawn_in(window, async move |this, cx| {
// Send and get reports
let outputs = task.await;
let outputs = send_task.await;
// Add sent IDs to the list
let mut sent_ids = sent_ids.write().await;
sent_ids.extend(outputs.iter().filter_map(|output| output.gift_wrap_id));
// Update the state
this.update(cx, |this, cx| {
this.insert_reports(id, outputs, cx);
})?;
Ok(())
}))
}));
}
/// Clear the input field, attachments, and replies
@@ -429,10 +407,13 @@ impl ChatPanel {
/// Insert reports
fn insert_reports(&mut self, id: EventId, reports: Vec<SendReport>, cx: &mut Context<Self>) {
self.reports_by_id.update(cx, |this, cx| {
this.entry(id).or_default().extend(reports);
self.reports_by_id
.write()
.unwrap()
.entry(id)
.or_default()
.extend(reports);
cx.notify();
});
}
/// Insert a message into the chat panel
@@ -466,13 +447,12 @@ impl ChatPanel {
}
/// Check if a message has any reports
fn has_reports(&self, id: &EventId, cx: &App) -> bool {
self.reports_by_id.read(cx).get(id).is_some()
fn has_reports(&self, id: &EventId, _cx: &App) -> bool {
self.reports_by_id.read().unwrap().get(id).is_some()
}
/// Get all sent reports for a message by its ID
fn sent_reports(&self, id: &EventId, cx: &App) -> Option<Vec<SendReport>> {
self.reports_by_id.read(cx).get(id).cloned()
fn sent_reports(&self, id: &EventId, _cx: &App) -> Option<Vec<SendReport>> {
self.reports_by_id.read().unwrap().get(id).cloned()
}
/// Get a message by its ID

76
crates/person/src/lib.rs
View File

@@ -1,6 +1,5 @@
use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use std::rc::Rc;
use std::time::Duration;
use anyhow::{Error, anyhow};
@@ -24,9 +23,9 @@ impl Global for GlobalPersonRegistry {}
#[derive(Debug, Clone)]
enum Dispatch {
Person(Box<Person>),
Announcement(Box<Event>),
Relays(Box<Event>),
Person(Person),
Announcement(Event),
Relays(Event),
}
/// Person Registry
@@ -36,7 +35,7 @@ pub struct PersonRegistry {
persons: HashMap<PublicKey, Entity<Person>>,
/// Set of public keys that have been seen
seens: Rc<RefCell<HashSet<PublicKey>>>,
seens: RefCell<HashSet<PublicKey>>,
/// Sender for requesting metadata
sender: flume::Sender<PublicKey>,
@@ -67,36 +66,26 @@ impl PersonRegistry {
let mut tasks = smallvec![];
tasks.push(
// Handle nostr notifications
cx.background_spawn({
tasks.push(cx.background_spawn({
let client = client.clone();
async move {
Self::handle_notifications(&client, &tx).await;
}
}),
);
}));
tasks.push(
// Handle metadata requests
cx.background_spawn({
tasks.push(cx.background_spawn({
let client = client.clone();
async move {
Self::handle_requests(&client, &mta_rx).await;
}
}),
);
}));
tasks.push(
// Update GPUI state
cx.spawn(async move |this, cx| {
tasks.push(cx.spawn(async move |this, cx| {
while let Ok(event) = rx.recv_async().await {
this.update(cx, |this, cx| {
match event {
Dispatch::Person(person) => {
this.insert(*person, cx);
this.insert(person, cx);
}
Dispatch::Announcement(event) => {
this.set_announcement(&event, cx);
@@ -108,8 +97,7 @@ impl PersonRegistry {
})
.ok();
}
}),
);
}));
// Load all user profiles from the database
cx.defer_in(window, |this, _window, cx| {
@@ -118,7 +106,7 @@ impl PersonRegistry {
Self {
persons: HashMap::new(),
seens: Rc::new(RefCell::new(HashSet::new())),
seens: RefCell::new(HashSet::new()),
sender: mta_tx,
tasks,
}
@@ -145,24 +133,25 @@ impl PersonRegistry {
Kind::Metadata => {
let metadata = Metadata::from_json(&event.content).unwrap_or_default();
let person = Person::new(event.pubkey, metadata);
let val = Box::new(person);
// Send
tx.send_async(Dispatch::Person(val)).await.ok();
if tx.send_async(Dispatch::Person(person)).await.is_err() {
log::warn!("PersonRegistry channel closed, dropping metadata event");
}
}
Kind::ContactList => {
let public_keys = event.extract_public_keys();
// Get metadata for all public keys
get_metadata(client, public_keys).await.ok();
if let Err(e) = get_metadata(client, public_keys).await {
log::warn!("Failed to get metadata for contact list: {e}");
}
}
Kind::InboxRelays => {
let val = Box::new(event.into_owned());
// Send
tx.send_async(Dispatch::Relays(val)).await.ok();
tx.send_async(Dispatch::Relays(event.into_owned()))
.await
.ok();
}
Kind::Custom(10044) => {
let val = Box::new(event.into_owned());
// Send
tx.send_async(Dispatch::Announcement(val)).await.ok();
tx.send_async(Dispatch::Announcement(event.into_owned()))
.await
.ok();
}
_ => {}
}
@@ -182,13 +171,17 @@ impl PersonRegistry {
Ok(Some(public_key)) => {
batch.insert(public_key);
// Process the batch if it's full
if batch.len() >= 20 {
get_metadata(client, std::mem::take(&mut batch)).await.ok();
if batch.len() >= 20
&& let Err(e) = get_metadata(client, std::mem::take(&mut batch)).await
{
log::warn!("Failed to get metadata batch: {e}");
}
}
_ => {
if !batch.is_empty() {
get_metadata(client, std::mem::take(&mut batch)).await.ok();
if !batch.is_empty()
&& let Err(e) = get_metadata(client, std::mem::take(&mut batch)).await
{
log::warn!("Failed to get metadata batch: {e}");
}
}
}
@@ -290,15 +283,14 @@ impl PersonRegistry {
}
let public_key = *public_key;
let mut seen = self.seens.borrow_mut();
if seen.insert(public_key) {
if self.seens.borrow_mut().insert(public_key) {
let sender = self.sender.clone();
// Spawn background task to request metadata
cx.background_spawn(async move {
if let Err(e) = sender.send_async(public_key).await {
log::warn!("Failed to send public key for metadata request: {}", e);
log::warn!("Failed to send public key for metadata request: {e}");
}
})
.detach();