use std::cmp::Reverse; use std::collections::{BTreeMap, BTreeSet, HashMap}; use anyhow::Error; use common::room_hash; use fuzzy_matcher::skim::SkimMatcherV2; use fuzzy_matcher::FuzzyMatcher; use global::nostr_client; use gpui::{ App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task, WeakEntity, Window, }; use identity::Identity; use itertools::Itertools; use nostr_sdk::prelude::*; use room::RoomKind; use smallvec::{smallvec, SmallVec}; use crate::room::Room; pub mod message; pub mod room; i18n::init!(); 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 RoomEmitter { Open(WeakEntity), Request(RoomKind), } /// Main registry for managing chat rooms and user profiles pub struct Registry { /// Collection of all chat rooms pub rooms: Vec>, /// Collection of all persons (user profiles) pub persons: BTreeMap>, /// Indicates if rooms are currently being loaded /// /// Always equal to `true` when the app starts pub loading: bool, /// Subscriptions for observing changes #[allow(dead_code)] subscriptions: SmallVec<[Subscription; 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 subscriptions = smallvec![]; // Load all user profiles from the database when the Registry is created subscriptions.push(cx.observe_new::(|this, _window, cx| { let task = this.load_local_person(cx); this.set_persons_from_task(task, cx); })); // When any Room is created, load members metadata subscriptions.push(cx.observe_new::(|this, _window, cx| { let task = this.load_metadata(cx); Self::global(cx).update(cx, |this, cx| { this.set_persons_from_task(task, cx); }); })); Self { rooms: vec![], persons: BTreeMap::new(), loading: true, subscriptions, } } pub(crate) fn set_persons_from_task( &mut self, task: Task, Error>>, cx: &mut Context, ) { cx.spawn(async move |this, cx| { if let Ok(profiles) = task.await { this.update(cx, |this, cx| { for profile in profiles { this.persons .insert(profile.public_key(), cx.new(|_| profile)); } cx.notify(); }) .ok(); } }) .detach(); } pub(crate) fn load_local_person(&self, cx: &App) -> Task, Error>> { cx.background_spawn(async move { let filter = Filter::new().kind(Kind::Metadata).limit(100); let events = nostr_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) }) } 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())) } 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.persons.get(public_key).map(|e| e.read(cx)).cloned(); profiles.push(profile); } profiles } pub fn insert_or_update_person(&mut self, event: Event, cx: &mut App) { let public_key = event.pubkey; let Ok(metadata) = Metadata::from_json(event.content) else { // Invalid metadata, no need to process further. return; }; if let Some(person) = self.persons.get(&public_key) { person.update(cx, |this, cx| { *this = Profile::new(public_key, metadata); cx.notify(); }); } else { self.persons .insert(public_key, cx.new(|_| Profile::new(public_key, metadata))); } } /// 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, trusted_only: bool, cx: &App) -> Vec> { self.rooms .iter() .filter(|room| { if trusted_only { room.read(cx).kind == RoomKind::Trusted } else { 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(); } /// 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() } /// Set the loading status of the registry. pub fn set_loading(&mut self, status: bool, cx: &mut Context) { self.loading = status; cx.notify(); } /// Load all rooms from the lmdb. /// /// This method: /// 1. Fetches all private direct messages from the lmdb /// 2. Groups them by ID /// 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) { log::info!("Starting to load rooms from database..."); let task: Task, Error>> = cx.background_spawn(async move { let client = nostr_client(); let signer = client.signer().await?; let public_key = signer.get_public_key().await?; // Get messages sent by the user let send = Filter::new() .kind(Kind::PrivateDirectMessage) .author(public_key); // Get messages received by the user let recv = Filter::new() .kind(Kind::PrivateDirectMessage) .pubkey(public_key); let send_events = client.database().query(send).await?; let recv_events = client.database().query(recv).await?; let events = send_events.merge(recv_events); let mut rooms: BTreeSet = BTreeSet::new(); let mut trusted_keys: BTreeSet = BTreeSet::new(); // Process each event and group by room hash for event in events .into_iter() .sorted_by_key(|event| Reverse(event.created_at)) .filter(|ev| ev.tags.public_keys().peekable().peek().is_some()) { let hash = room_hash(&event); if rooms.iter().any(|room| room.id == hash) { continue; } let mut public_keys = event.tags.public_keys().copied().collect_vec(); public_keys.push(event.pubkey); let mut is_trust = trusted_keys.contains(&event.pubkey); if !is_trust { // 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 is_trust = client.database().count(filter).await.unwrap_or(0) >= 1; if is_trust { trusted_keys.insert(event.pubkey); } } // Check if current_user has sent a message to this room at least once let filter = Filter::new() .kind(Kind::PrivateDirectMessage) .author(public_key) .pubkeys(public_keys); // If current user has sent a message at least once, mark as ongoing let is_ongoing = client.database().count(filter).await.unwrap_or(1) >= 1; if is_ongoing { rooms.insert(Room::new(&event).kind(RoomKind::Ongoing)); } else if is_trust { rooms.insert(Room::new(&event).kind(RoomKind::Trusted)); } else { rooms.insert(Room::new(&event)); } } Ok(rooms) }); cx.spawn_in(window, async move |this, cx| { match task.await { Ok(rooms) => { this.update(cx, |this, cx| { this.extend_rooms(rooms, cx); this.sort(cx); }) .ok(); } Err(e) => { // TODO: push notification log::error!("Failed to load rooms: {e}") } }; }) .detach(); } pub(crate) fn extend_rooms(&mut self, rooms: BTreeSet, cx: &mut Context) { let mut room_map: HashMap = HashMap::with_capacity(self.rooms.len()); for (index, room) in self.rooms.iter().enumerate() { room_map.insert(room.read(cx).id, index); } 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| { *this = new_room; cx.notify(); }); } else { let new_index = self.rooms.len(); room_map.insert(new_room.id, new_index); self.rooms.push(cx.new(|_| new_room)); } } } /// 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(RoomEmitter::Open(weak_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 event_to_message(&mut self, event: Event, window: &mut Window, cx: &mut Context) { let id = room_hash(&event); let author = event.pubkey; let Some(identity) = Identity::read_global(cx).public_key() else { return; }; if let Some(room) = self.rooms.iter().find(|room| room.read(cx).id == id) { // Update room room.update(cx, |this, cx| { this.created_at(event.created_at, cx); // Set this room is ongoing if the new message is from current user if author == identity { this.set_ongoing(cx); } // Emit the new message to the room cx.defer_in(window, |this, window, cx| { this.emit_message(event, window, cx); }); }); // Re-sort the rooms registry by their created at self.sort(cx); } else { let room = Room::new(&event).kind(RoomKind::Unknown); let kind = room.kind; // Push the new room to the front of the list self.add_room(cx.new(|_| room), cx); // Notify the UI about the new room cx.defer_in(window, move |_this, _window, cx| { cx.emit(RoomEmitter::Request(kind)); }); } } }