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

chore: Improve Chat Performance (#35)

Browse Source 
* refactor

* optimistically update message list

* fix

* update

* handle duplicate messages

* update ui

* refactor input

* update multi line input

* clean up
This commit is contained in:
reya
2025-05-18 15:35:33 +07:00
committed by GitHub
parent 4f066b7c00
commit 443dbc82a6
37 changed files with 3060 additions and 1979 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
crates/chats/src/lib.rs
View File

@@ -1,8 +1,9 @@
use std::{
cmp::Reverse,
collections::{BTreeMap, BTreeSet, HashMap},
collections::{BTreeMap, HashMap},
};
use account::Account;
use anyhow::Error;
use common::room_hash;
use fuzzy_matcher::{skim::SkimMatcherV2, FuzzyMatcher};
@@ -36,19 +37,21 @@ impl Global for GlobalChatRegistry {}
/// - Loading room data from the lmdb
/// - Handling messages and room creation
pub struct ChatRegistry {
/// Collection of all chat rooms
rooms: BTreeSet<Entity<Room>>,
/// Map of user public keys to their profile metadata
profiles: Entity<BTreeMap<PublicKey, Option<Metadata>>>,
/// Collection of all chat rooms
pub rooms: Vec<Entity<Room>>,
/// Indicates if rooms are currently being loaded
pub loading: bool,
///
/// Always equal to `true` when the app starts
pub wait_for_eose: bool,
/// Subscriptions for observing changes
#[allow(dead_code)]
subscriptions: SmallVec<[Subscription; 1]>,
subscriptions: SmallVec<[Subscription; 2]>,
}
impl ChatRegistry {
/// Retrieve the global ChatRegistry instance
/// Retrieve the Global ChatRegistry instance
pub fn global(cx: &App) -> Entity<Self> {
cx.global::<GlobalChatRegistry>().0.clone()
}
@@ -68,28 +71,37 @@ impl ChatRegistry {
let profiles = cx.new(|_| BTreeMap::new());
let mut subscriptions = smallvec![];
// Observe new Room creations to collect profile metadata
subscriptions.push(cx.observe_new::<Room>(|this, _, cx| {
let task = this.metadata(cx);
// When the ChatRegistry is created, load all rooms from the local database
subscriptions.push(cx.observe_new::<ChatRegistry>(|this, window, cx| {
if let Some(window) = window {
this.load_rooms(window, cx);
}
}));
cx.spawn(async move |_, cx| {
if let Ok(data) = task.await {
cx.update(|cx| {
for (public_key, metadata) in data.into_iter() {
Self::global(cx).update(cx, |this, cx| {
this.add_profile(public_key, metadata, cx);
})
}
})
.ok();
}
})
.detach();
// When any Room is created, load metadata for all members
subscriptions.push(cx.observe_new::<Room>(|this, window, cx| {
if let Some(window) = window {
let task = this.load_metadata(cx);
cx.spawn_in(window, async move |_, cx| {
if let Ok(data) = task.await {
cx.update(|_, cx| {
for (public_key, metadata) in data.into_iter() {
Self::global(cx).update(cx, |this, cx| {
this.add_profile(public_key, metadata, cx);
})
}
})
.ok();
}
})
.detach();
}
}));
Self {
rooms: BTreeSet::new(),
loading: true,
rooms: vec![],
wait_for_eose: true,
profiles,
subscriptions,
}
@@ -103,22 +115,13 @@ impl ChatRegistry {
.cloned()
}
/// Get all rooms grouped by their kind.
pub fn rooms(&self, cx: &App) -> BTreeMap<RoomKind, Vec<Entity<Room>>> {
let mut groups = BTreeMap::new();
groups.insert(RoomKind::Ongoing, Vec::new());
groups.insert(RoomKind::Trusted, Vec::new());
groups.insert(RoomKind::Unknown, Vec::new());
for room in self.rooms.iter() {
let kind = room.read(cx).kind;
groups
.entry(kind)
.or_insert_with(Vec::new)
.push(room.to_owned());
}
groups
/// Get rooms by its kind.
pub fn rooms_by_kind(&self, kind: RoomKind, cx: &App) -> Vec<Entity<Room>> {
self.rooms
.iter()
.filter(|room| room.read(cx).kind == kind)
.cloned()
.collect()
}
/// Get the IDs of all rooms.
@@ -149,13 +152,20 @@ impl ChatRegistry {
/// 3. Determines each room's type based on message frequency and trust status
/// 4. Creates Room entities for each unique room
pub fn load_rooms(&mut self, window: &mut Window, cx: &mut Context<Self>) {
// [event] is the Nostr Event
// [usize] is the total number of messages, used to determine an ongoing conversation
// [bool] is used to determine if the room is trusted
type Rooms = Vec<(Event, usize, bool)>;
let task: Task<Result<Rooms, Error>> = cx.background_spawn(async move {
let client = get_client();
let signer = client.signer().await?;
let public_key = signer.get_public_key().await?;
// If the user is not logged in, do nothing
let Some(user) = Account::get_global(cx).profile_ref() else {
return;
};
let client = get_client();
let public_key = user.public_key();
let task: Task<Result<Rooms, Error>> = cx.background_spawn(async move {
// Get messages sent by the user
let send = Filter::new()
.kind(Kind::PrivateDirectMessage)
@@ -179,7 +189,9 @@ impl ChatRegistry {
{
let hash = room_hash(&event);
// Check if room's author is seen in any contact list
let filter = Filter::new().kind(Kind::ContactList).pubkey(event.pubkey);
// If room's author is seen at least once, mark as trusted
let is_trust = client.database().count(filter).await? >= 1;
room_map
@@ -202,35 +214,32 @@ impl ChatRegistry {
cx.spawn_in(window, async move |this, cx| {
if let Ok(events) = task.await {
cx.update(|_, cx| {
this.update(cx, |this, cx| {
let ids = this.room_ids(cx);
let rooms: Vec<Entity<Room>> = events
.into_iter()
.filter_map(|(event, count, trusted)| {
let hash = room_hash(&event);
if !ids.iter().any(|this| this == &hash) {
let kind = if count > 2 {
// If frequency count is greater than 2, mark this room as ongoing
RoomKind::Ongoing
} else if trusted {
RoomKind::Trusted
} else {
RoomKind::Unknown
};
Some(cx.new(|_| Room::new(&event).kind(kind)))
this.update(cx, |this, cx| {
let ids = this.room_ids(cx);
let rooms: Vec<Entity<Room>> = events
.into_iter()
.filter_map(|(event, count, trusted)| {
let hash = room_hash(&event);
if !ids.iter().any(|this| this == &hash) {
let kind = if count > 2 {
// If frequency count is greater than 2, mark this room as ongoing
RoomKind::Ongoing
} else if trusted {
RoomKind::Trusted
} else {
None
}
})
.collect();
RoomKind::Unknown
};
Some(cx.new(|_| Room::new(&event).kind(kind)))
} else {
None
}
})
.collect();
this.rooms.extend(rooms);
this.loading = false;
this.rooms.extend(rooms);
this.wait_for_eose = false;
cx.notify();
})
.ok();
cx.notify();
})
.ok();
}
@@ -269,10 +278,24 @@ impl ChatRegistry {
Profile::new(*public_key, metadata)
}
/// Parse a Nostr event into a Room and push it to the registry
/// Push a Room Entity to the global registry
///
/// Returns the ID of the room
pub fn push_room(&mut self, room: Entity<Room>, cx: &mut Context<Self>) -> u64 {
let id = room.read(cx).id;
if !self.rooms.iter().any(|this| this.read(cx) == room.read(cx)) {
self.rooms.insert(0, room);
cx.notify();
}
id
}
/// Parse a Nostr event into a Coop Room and push it to the global registry
///
/// Returns the ID of the new room
pub fn push_event(
pub fn event_to_room(
&mut self,
event: &Event,
window: &mut Window,
@@ -282,7 +305,7 @@ impl ChatRegistry {
let id = room.id;
if !self.rooms.iter().any(|this| this.read(cx) == &room) {
self.rooms.insert(cx.new(|_| room));
self.rooms.insert(0, cx.new(|_| room));
cx.notify();
} else {
window.push_notification("Room already exists", cx);
@@ -291,38 +314,25 @@ impl ChatRegistry {
id
}
/// Parse a nostr event into Room and push to the registry
///
/// Returns the ID of the new room
pub fn push_room(&mut self, room: Entity<Room>, cx: &mut Context<Self>) -> u64 {
let id = room.read(cx).id;
if !self.rooms.iter().any(|this| this.read(cx) == room.read(cx)) {
self.rooms.insert(room);
cx.notify();
}
id
}
/// Push a new message to a room
/// Parse a Nostr event into a Coop Message and push it to the belonging room
///
/// If the room doesn't exist, it will be created.
/// Updates room ordering based on the most recent messages.
pub fn push_message(&mut self, event: Event, window: &mut Window, cx: &mut Context<Self>) {
pub fn event_to_message(&mut self, event: Event, window: &mut Window, cx: &mut Context<Self>) {
let id = room_hash(&event);
if let Some(room) = self.rooms.iter().find(|room| room.read(cx).id == id) {
room.update(cx, |this, cx| {
this.created_at(event.created_at, cx);
// Emit the new message to the room
cx.defer_in(window, |this, window, cx| {
this.emit_message(event, window, cx);
});
});
cx.notify();
} else {
// Push the new room to the front of the list
self.rooms.insert(cx.new(|_| Room::new(&event)));
self.rooms.insert(0, cx.new(|_| Room::new(&event)));
cx.notify();
}
}

48
crates/chats/src/message.rs
View File

@@ -2,6 +2,8 @@ use chrono::{Local, TimeZone};
use gpui::SharedString;
use nostr_sdk::prelude::*;
use crate::room::SendError;
/// # Message
///
/// Represents a message in the application.
@@ -18,8 +20,9 @@ pub struct Message {
pub id: EventId,
pub content: String,
pub author: Profile,
pub mentions: Vec<Profile>,
pub created_at: Timestamp,
pub mentions: Vec<Profile>,
pub errors: Option<Vec<SendError>>,
}
impl Message {
@@ -42,23 +45,10 @@ impl Message {
author,
created_at,
mentions: vec![],
errors: None,
}
}
/// Adds or replaces mentions in the message
///
/// # Arguments
///
/// * `mentions` - New list of mentioned profiles
///
/// # Returns
///
/// The same message with updated mentions
pub fn with_mentions(mut self, mentions: impl IntoIterator<Item = Profile>) -> Self {
self.mentions.extend(mentions);
self
}
/// Formats the message timestamp as a human-readable relative time
///
/// # Returns
@@ -85,6 +75,34 @@ impl Message {
}
.into()
}
/// Adds or replaces mentions in the message
///
/// # Arguments
///
/// * `mentions` - New list of mentioned profiles
///
/// # Returns
///
/// The same message with updated mentions
pub fn with_mentions(mut self, mentions: impl IntoIterator<Item = Profile>) -> Self {
self.mentions.extend(mentions);
self
}
/// Adds or replaces errors in the message
///
/// # Arguments
///
/// * `errors` - New list of errors
///
/// # Returns
///
/// The same message with updated errors
pub fn with_errors(mut self, errors: Vec<SendError>) -> Self {
self.errors = Some(errors);
self
}
}
/// # RoomMessage

310
crates/chats/src/room.rs
View File

@@ -1,14 +1,13 @@
use std::{cmp::Ordering, sync::Arc};
use account::Account;
use anyhow::Error;
use anyhow::{anyhow, Error};
use chrono::{Local, TimeZone};
use common::{compare, profile::SharedProfile, room_hash};
use global::get_client;
use gpui::{App, AppContext, Context, EventEmitter, SharedString, Task, Window};
use itertools::Itertools;
use nostr_sdk::prelude::*;
use smol::channel::Receiver;
use crate::{
constants::{DAYS_IN_MONTH, HOURS_IN_DAY, MINUTES_IN_HOUR, NOW, SECONDS_IN_MINUTE},
@@ -17,14 +16,12 @@ use crate::{
};
#[derive(Debug, Clone)]
pub struct IncomingEvent {
pub event: RoomMessage,
}
pub struct Incoming(pub Message);
#[derive(Debug)]
pub enum SendStatus {
Sent(EventId),
Failed(Error),
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SendError {
pub profile: Profile,
pub message: String,
}
#[derive(Clone, Copy, Hash, Debug, PartialEq, Eq, PartialOrd, Ord, Default)]
@@ -69,7 +66,7 @@ impl PartialEq for Room {
}
}
impl EventEmitter<IncomingEvent> for Room {}
impl EventEmitter<Incoming> for Room {}
impl Room {
/// Creates a new Room instance from a Nostr event
@@ -87,6 +84,7 @@ impl Room {
// Get all pubkeys from the event's tags
let mut pubkeys: Vec<PublicKey> = event.tags.public_keys().cloned().collect();
// The author is always put at the end of the vector
pubkeys.push(event.pubkey);
// Convert pubkeys into members
@@ -267,11 +265,13 @@ impl Room {
/// An Option<SharedString> containing the avatar:
/// - For a direct message: the other person's avatar
/// - For a group chat: None
pub fn display_image(&self, cx: &App) -> Option<SharedString> {
if !self.is_group() {
Some(self.first_member(cx).shared_avatar())
pub fn display_image(&self, cx: &App) -> SharedString {
if let Some(picture) = self.picture.as_ref() {
picture.clone()
} else if !self.is_group() {
self.first_member(cx).shared_avatar()
} else {
None
"brand/group.png".into()
}
}
@@ -327,12 +327,12 @@ impl Room {
///
/// A Task that resolves to Result<Vec<(PublicKey, Option<Metadata>)>, Error>
#[allow(clippy::type_complexity)]
pub fn metadata(
pub fn load_metadata(
&self,
cx: &mut Context<Self>,
) -> Task<Result<Vec<(PublicKey, Option<Metadata>)>, Error>> {
let client = get_client();
let public_keys = self.members.clone();
let public_keys = Arc::clone(&self.members);
cx.background_spawn(async move {
let mut output = vec![];
@@ -378,76 +378,6 @@ impl Room {
})
}
/// Sends a message to all members in the room
///
/// # Arguments
///
/// * `content` - The content of the message to send
/// * `cx` - The App context
///
/// # Returns
///
/// A Task that resolves to Result<Vec<String>, Error> where the
/// strings contain error messages for any failed sends
pub fn send_message(&self, content: String, cx: &App) -> Option<Receiver<SendStatus>> {
let profile = Account::global(cx).read(cx).profile.clone()?;
let public_key = profile.public_key();
let subject = self.subject.clone();
let picture = self.picture.clone();
let pubkeys = self.members.clone();
let (tx, rx) = smol::channel::bounded::<SendStatus>(pubkeys.len());
cx.background_spawn(async move {
let client = get_client();
let mut tags: Vec<Tag> = pubkeys
.iter()
.filter_map(|pubkey| {
if pubkey != &public_key {
Some(Tag::public_key(*pubkey))
} else {
None
}
})
.collect();
// Add subject tag if it's present
if let Some(subject) = subject {
tags.push(Tag::from_standardized(TagStandard::Subject(
subject.to_string(),
)));
}
// Add picture tag if it's present
if let Some(picture) = picture {
tags.push(Tag::custom(TagKind::custom("picture"), vec![picture]));
}
for pubkey in pubkeys.iter() {
match client
.send_private_msg(*pubkey, &content, tags.clone())
.await
{
Ok(output) => {
if let Err(e) = tx.send(SendStatus::Sent(output.val)).await {
log::error!("Failed to send message to {}: {}", pubkey, e);
}
}
Err(e) => {
if let Err(e) = tx.send(SendStatus::Failed(e.into())).await {
log::error!("Failed to send message to {}: {}", pubkey, e);
}
}
}
}
})
.detach();
Some(rx)
}
/// Loads all messages for this room from the database
///
/// # Arguments
@@ -461,7 +391,6 @@ impl Room {
pub fn load_messages(&self, cx: &App) -> Task<Result<Vec<RoomMessage>, Error>> {
let client = get_client();
let pubkeys = Arc::clone(&self.members);
let profiles: Vec<Profile> = pubkeys
.iter()
.map(|pubkey| ChatRegistry::global(cx).read(cx).profile(pubkey, cx))
@@ -492,11 +421,11 @@ impl Room {
.collect::<Vec<_>>();
for event in events.into_iter() {
let mut mentions = vec![];
let id = event.id;
let created_at = event.created_at;
let content = event.content.clone();
let tokens = parser.parse(&content);
let mut mentions = vec![];
let author = profiles
.iter()
@@ -545,66 +474,167 @@ impl Room {
///
/// # Effects
///
/// Processes the event and emits an IncomingEvent to the UI when complete
pub fn emit_message(&self, event: Event, window: &mut Window, cx: &mut Context<Self>) {
let pubkeys = self.members.clone();
let profiles: Vec<Profile> = pubkeys
.iter()
.map(|pubkey| ChatRegistry::global(cx).read(cx).profile(pubkey, cx))
.collect();
/// Processes the event and emits an Incoming to the UI when complete
pub fn emit_message(&self, event: Event, _window: &mut Window, cx: &mut Context<Self>) {
let author = ChatRegistry::get_global(cx).profile(&event.pubkey, cx);
let mentions = extract_mentions(&event.content, cx);
let message =
Message::new(event.id, event.content, author, event.created_at).with_mentions(mentions);
let task: Task<Result<RoomMessage, Error>> = cx.background_spawn(async move {
let parser = NostrParser::new();
let id = event.id;
let created_at = event.created_at;
let content = event.content.clone();
let tokens = parser.parse(&content);
let mut mentions = vec![];
cx.emit(Incoming(message));
}
let author = profiles
/// Creates a temporary message for optimistic updates
///
/// This constructs an unsigned message with the current user as the author,
/// extracts any mentions from the content, and packages it as a Message struct.
/// The message will have a generated ID but hasn't been published to relays.
///
/// # Arguments
///
/// * `content` - The message content text
/// * `cx` - The application context containing user profile information
///
/// # Returns
///
/// Returns `Some(Message)` containing the temporary message if the current user's profile is available,
/// or `None` if no account is found.
pub fn create_temp_message(&self, content: &str, cx: &App) -> Option<Message> {
let profile = Account::get_global(cx).profile.clone()?;
let public_key = profile.public_key();
let builder = EventBuilder::private_msg_rumor(public_key, content);
// Create a unsigned event to convert to Coop Message
let mut event = builder.build(public_key);
event.ensure_id();
// Extract all mentions from content
let mentions = extract_mentions(&event.content, cx);
Some(
Message::new(event.id.unwrap(), event.content, profile, event.created_at)
.with_mentions(mentions),
)
}
/// Sends a message to all members in the background task
///
/// # Arguments
///
/// * `content` - The content of the message to send
/// * `cx` - The App context
///
/// # Returns
///
/// A Task that resolves to Result<Vec<String>, Error> where the
/// strings contain error messages for any failed sends
pub fn send_in_background(&self, msg: &str, cx: &App) -> Task<Result<Vec<SendError>, Error>> {
let content = msg.to_owned();
let subject = self.subject.clone();
let picture = self.picture.clone();
let public_keys = Arc::clone(&self.members);
cx.background_spawn(async move {
let client = get_client();
let signer = client.signer().await?;
let public_key = signer.get_public_key().await?;
let mut reports = vec![];
let mut tags: Vec<Tag> = public_keys
.iter()
.find(|profile| profile.public_key() == event.pubkey)
.cloned()
.unwrap_or_else(|| Profile::new(event.pubkey, Metadata::default()));
let pubkey_tokens = tokens
.filter_map(|token| match token {
Token::Nostr(nip21) => match nip21 {
Nip21::Pubkey(pubkey) => Some(pubkey),
Nip21::Profile(profile) => Some(profile.public_key),
_ => None,
},
_ => None,
.filter_map(|pubkey| {
if pubkey != &public_key {
Some(Tag::public_key(*pubkey))
} else {
None
}
})
.collect::<Vec<_>>();
.collect();
for pubkey in pubkey_tokens {
mentions.push(
profiles
.iter()
.find(|profile| profile.public_key() == pubkey)
.cloned()
.unwrap_or_else(|| Profile::new(pubkey, Metadata::default())),
);
// Add subject tag if it's present
if let Some(subject) = subject {
tags.push(Tag::from_standardized(TagStandard::Subject(
subject.to_string(),
)));
}
let message = Message::new(id, content, author, created_at).with_mentions(mentions);
let room_message = RoomMessage::user(message);
Ok(room_message)
});
cx.spawn_in(window, async move |this, cx| {
if let Ok(message) = task.await {
cx.update(|_, cx| {
this.update(cx, |_, cx| {
cx.emit(IncomingEvent { event: message });
})
.ok();
})
.ok();
// Add picture tag if it's present
if let Some(picture) = picture {
tags.push(Tag::custom(TagKind::custom("picture"), vec![picture]));
}
let Some((current_user, receivers)) = public_keys.split_last() else {
return Err(anyhow!("Something is wrong. Cannot get receivers list."));
};
for receiver in receivers.iter() {
if let Err(e) = client
.send_private_msg(*receiver, &content, tags.clone())
.await
{
let metadata = client
.database()
.metadata(*receiver)
.await?
.unwrap_or_default();
let profile = Profile::new(*receiver, metadata);
let report = SendError {
profile,
message: e.to_string(),
};
reports.push(report);
}
}
// Only send a backup message to current user if there are no issues when sending to others
if reports.is_empty() {
if let Err(e) = client
.send_private_msg(*current_user, &content, tags.clone())
.await
{
let metadata = client
.database()
.metadata(*current_user)
.await?
.unwrap_or_default();
let profile = Profile::new(*current_user, metadata);
let report = SendError {
profile,
message: e.to_string(),
};
reports.push(report);
}
}
Ok(reports)
})
.detach();
}
}
pub fn extract_mentions(content: &str, cx: &App) -> Vec<Profile> {
let parser = NostrParser::new();
let tokens = parser.parse(content);
let mut mentions = vec![];
let profiles = ChatRegistry::get_global(cx).profiles.read(cx);
let pubkey_tokens = tokens
.filter_map(|token| match token {
Token::Nostr(nip21) => match nip21 {
Nip21::Pubkey(pubkey) => Some(pubkey),
Nip21::Profile(profile) => Some(profile.public_key),
_ => None,
},
_ => None,
})
.collect::<Vec<_>>();
for pubkey in pubkey_tokens.into_iter() {
if let Some(metadata) = profiles.get(&pubkey).cloned() {
mentions.push(Profile::new(pubkey, metadata.unwrap_or_default()));
}
}
mentions
}