use std::cmp::Reverse; use std::collections::{HashMap, HashSet}; use anyhow::Error; use common::event::EventUtils; use fuzzy_matcher::skim::SkimMatcherV2; use fuzzy_matcher::FuzzyMatcher; use gpui::{ App, AppContext, AsyncApp, Context, Entity, EventEmitter, Global, Task, WeakEntity, Window, }; use nostr_sdk::prelude::*; use room::RoomKind; use settings::AppSettings; use smallvec::{smallvec, SmallVec}; use states::app_state; use crate::room::Room; pub mod message; pub mod room; pub fn init(cx: &mut App) { Registry::set_global(cx.new(Registry::new), cx); } struct GlobalRegistry(Entity); impl Global for GlobalRegistry {} #[derive(Debug)] pub enum RegistryEvent { Open(WeakEntity), Close(u64), NewRequest(RoomKind), } pub struct Registry { /// Collection of all chat rooms pub rooms: Vec>, /// Collection of all persons (user profiles) pub persons: HashMap>, /// Loading status of the registry pub loading: bool, /// Public Key of the currently activated signer signer_pubkey: Option, /// Tasks for asynchronous operations _tasks: SmallVec<[Task<()>; 2]>, } impl EventEmitter for Registry {} impl Registry { /// Retrieve the global registry state pub fn global(cx: &App) -> Entity { cx.global::().0.clone() } /// Retrieve the registry instance pub fn read_global(cx: &App) -> &Self { cx.global::().0.read(cx) } /// Set the global registry instance pub(crate) fn set_global(state: Entity, cx: &mut App) { cx.set_global(GlobalRegistry(state)); } /// Create a new registry instance pub(crate) fn new(cx: &mut Context) -> Self { let mut tasks = smallvec![]; tasks.push( // Load all user profiles from the database cx.spawn(async move |this, cx| { match Self::load_persons(cx).await { Ok(profiles) => { this.update(cx, |this, cx| { this.set_persons(profiles, cx); }) .ok(); } Err(e) => { log::error!("Failed to load persons: {e}"); } }; }), ); Self { rooms: vec![], persons: HashMap::new(), signer_pubkey: None, loading: true, _tasks: tasks, } } /// Create a task to load all user profiles from the database fn load_persons(cx: &AsyncApp) -> Task, Error>> { cx.background_spawn(async move { let client = app_state().client(); let filter = Filter::new().kind(Kind::Metadata).limit(200); let events = client.database().query(filter).await?; let mut profiles = vec![]; for event in events.into_iter() { let metadata = Metadata::from_json(event.content).unwrap_or_default(); let profile = Profile::new(event.pubkey, metadata); profiles.push(profile); } Ok(profiles) }) } /// Returns the public key of the currently activated signer. pub fn signer_pubkey(&self) -> Option { self.signer_pubkey } /// Update the public key of the currently activated signer. pub fn set_signer_pubkey(&mut self, public_key: PublicKey, cx: &mut Context) { self.signer_pubkey = Some(public_key); cx.notify(); } /// Insert batch of persons pub fn set_persons(&mut self, profiles: Vec, cx: &mut Context) { for profile in profiles.into_iter() { self.persons .insert(profile.public_key(), cx.new(|_| profile)); } cx.notify(); } /// Get single person pub fn get_person(&self, public_key: &PublicKey, cx: &App) -> Profile { self.persons .get(public_key) .map(|e| e.read(cx)) .cloned() .unwrap_or(Profile::new(public_key.to_owned(), Metadata::default())) } /// Get group of persons pub fn get_group_person(&self, public_keys: &[PublicKey], cx: &App) -> Vec { let mut profiles = vec![]; for public_key in public_keys.iter() { let profile = self.get_person(public_key, cx); profiles.push(profile); } profiles } /// Insert or update a person pub fn insert_or_update_person(&mut self, profile: Profile, cx: &mut App) { let public_key = profile.public_key(); match self.persons.get(&public_key) { Some(person) => { person.update(cx, |this, cx| { *this = profile; cx.notify(); }); } None => { self.persons.insert(public_key, cx.new(|_| profile)); } } } pub fn set_loading(&mut self, loading: bool, cx: &mut Context) { self.loading = loading; cx.notify(); } /// Get a room by its ID. pub fn room(&self, id: &u64, cx: &App) -> Option> { self.rooms .iter() .find(|model| model.read(cx).id == *id) .cloned() } /// Get all ongoing rooms. pub fn ongoing_rooms(&self, cx: &App) -> Vec> { self.rooms .iter() .filter(|room| room.read(cx).kind == RoomKind::Ongoing) .cloned() .collect() } /// Get all request rooms. pub fn request_rooms(&self, cx: &App) -> Vec> { self.rooms .iter() .filter(|room| room.read(cx).kind != RoomKind::Ongoing) .cloned() .collect() } /// Add a new room to the start of list. pub fn add_room(&mut self, room: Entity, cx: &mut Context) { self.rooms.insert(0, room); cx.notify(); } /// Close a room. pub fn close_room(&mut self, id: u64, cx: &mut Context) { if self.rooms.iter().any(|r| r.read(cx).id == id) { cx.emit(RegistryEvent::Close(id)); } } /// Sort rooms by their created at. pub fn sort(&mut self, cx: &mut Context) { self.rooms.sort_by_key(|ev| Reverse(ev.read(cx).created_at)); cx.notify(); } /// Search rooms by their name. pub fn search(&self, query: &str, cx: &App) -> Vec> { let matcher = SkimMatcherV2::default(); self.rooms .iter() .filter(|room| { matcher .fuzzy_match(room.read(cx).display_name(cx).as_ref(), query) .is_some() }) .cloned() .collect() } /// Search rooms by public keys. pub fn search_by_public_key(&self, public_key: PublicKey, cx: &App) -> Vec> { self.rooms .iter() .filter(|room| room.read(cx).members.contains(&public_key)) .cloned() .collect() } /// Reset the registry. pub fn reset(&mut self, cx: &mut Context) { // Clear the current identity self.signer_pubkey = None; // Clear all current rooms self.rooms.clear(); cx.notify(); } /// Load all rooms from the database. pub fn load_rooms(&mut self, window: &mut Window, cx: &mut Context) { log::info!("Starting to load chat rooms..."); // Get the contact bypass setting let bypass_setting = AppSettings::get_contact_bypass(cx); let task: Task, Error>> = cx.background_spawn(async move { let client = app_state().client(); let signer = client.signer().await?; let public_key = signer.get_public_key().await?; let contacts = client.database().contacts_public_keys(public_key).await?; let authored_filter = Filter::new() .kind(Kind::ApplicationSpecificData) .custom_tag(SingleLetterTag::lowercase(Alphabet::A), public_key); let addressed_filter = Filter::new() .kind(Kind::ApplicationSpecificData) .custom_tag(SingleLetterTag::lowercase(Alphabet::P), public_key); let authored = client.database().query(authored_filter).await?; let addressed = client.database().query(addressed_filter).await?; let events = authored.merge(addressed); let mut rooms: HashSet = HashSet::new(); let mut grouped: HashMap> = HashMap::new(); // Process each event and group by room hash for raw in events.into_iter() { match UnsignedEvent::from_json(&raw.content) { Ok(rumor) => { if rumor.tags.public_keys().peekable().peek().is_some() { grouped.entry(rumor.uniq_id()).or_default().push(rumor); } } Err(e) => log::warn!("Failed to parse stored rumor: {e}"), } } for (_room_id, mut messages) in grouped.into_iter() { messages.sort_by_key(|m| Reverse(m.created_at)); let Some(latest) = messages.first() else { continue; }; let mut room = Room::from(latest); if rooms.iter().any(|r| r.id == room.id) { continue; } let mut public_keys: Vec = room.members().to_vec(); public_keys.retain(|pk| pk != &public_key); let user_sent = messages.iter().any(|m| m.pubkey == public_key); let mut bypassed = false; if bypass_setting { bypassed = public_keys.iter().any(|k| contacts.contains(k)); } if user_sent || bypassed { room = room.kind(RoomKind::Ongoing); } rooms.insert(room); } Ok(rooms) }); cx.spawn_in(window, async move |this, cx| { match task.await { Ok(rooms) => { this.update_in(cx, move |this, _window, cx| { this.extend_rooms(rooms, cx); this.sort(cx); }) .ok(); } Err(e) => { log::error!("Failed to load rooms: {e}") } }; }) .detach(); } pub(crate) fn extend_rooms(&mut self, rooms: HashSet, cx: &mut Context) { let mut room_map: HashMap = self .rooms .iter() .enumerate() .map(|(idx, room)| (room.read(cx).id, idx)) .collect(); for new_room in rooms.into_iter() { // Check if we already have a room with this ID if let Some(&index) = room_map.get(&new_room.id) { self.rooms[index].update(cx, |this, cx| { if new_room.created_at > this.created_at { *this = new_room; cx.notify(); } }); } else { let new_room_id = new_room.id; self.rooms.push(cx.new(|_| new_room)); let new_index = self.rooms.len(); room_map.insert(new_room_id, new_index); } } } /// Push a new Room to the global registry pub fn push_room(&mut self, room: Entity, cx: &mut Context) { let other_id = room.read(cx).id; let find_room = self.rooms.iter().find(|this| this.read(cx).id == other_id); let weak_room = if let Some(room) = find_room { room.downgrade() } else { let weak_room = room.downgrade(); // Add this room to the registry self.add_room(room, cx); weak_room }; cx.emit(RegistryEvent::Open(weak_room)); } /// Refresh messages for a room in the global registry pub fn refresh_rooms(&mut self, ids: Option>, cx: &mut Context) { if let Some(ids) = ids { for room in self.rooms.iter() { if ids.contains(&room.read(cx).id) { room.update(cx, |this, cx| { this.emit_refresh(cx); }); } } } } /// 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 event_to_message( &mut self, gift_wrap: EventId, event: UnsignedEvent, window: &mut Window, cx: &mut Context, ) { let id = event.uniq_id(); let author = event.pubkey; let Some(public_key) = self.signer_pubkey else { return; }; if let Some(room) = self.rooms.iter().find(|room| room.read(cx).id == id) { let is_new_event = event.created_at > room.read(cx).created_at; let created_at = event.created_at; let event_for_emit = event.clone(); // Update room room.update(cx, |this, cx| { if is_new_event { this.set_created_at(created_at, cx); } // Set this room is ongoing if the new message is from current user if author == public_key { this.set_ongoing(cx); } // Emit the new message to the room let event_to_emit = event_for_emit.clone(); cx.defer_in(window, move |this, _window, cx| { this.emit_message(gift_wrap, event_to_emit, cx); }); }); // Resort all rooms in the registry by their created at (after updated) if is_new_event { cx.defer_in(window, |this, _window, cx| { this.sort(cx); }); } } else { // Push the new room to the front of the list self.add_room(cx.new(|_| Room::from(&event)), cx); // Notify the UI about the new room cx.defer_in(window, move |_this, _window, cx| { cx.emit(RegistryEvent::NewRequest(RoomKind::default())); }); } } }