feat: revamp the chat panel ui (#7)

Reviewed-on: #7

crates/chat/src/lib.rs

@@ -7,16 +7,14 @@ use std::time::Duration;
 
 use anyhow::{anyhow, Context as AnyhowContext, Error};
 use common::EventUtils;
-use device::DeviceRegistry;
-use flume::Sender;
 use fuzzy_matcher::skim::SkimMatcherV2;
 use fuzzy_matcher::FuzzyMatcher;
 use gpui::{
-    App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task, WeakEntity,
+    App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task, WeakEntity, Window,
 };
 use nostr_sdk::prelude::*;
 use smallvec::{smallvec, SmallVec};
-use state::{tracker, NostrRegistry, RelayState, DEVICE_GIFTWRAP, USER_GIFTWRAP};
+use state::{NostrRegistry, DEVICE_GIFTWRAP, USER_GIFTWRAP};
 
 mod message;
 mod room;
@@ -24,8 +22,8 @@ mod room;
 pub use message::*;
 pub use room::*;
 
-pub fn init(cx: &mut App) {
-    ChatRegistry::set_global(cx.new(ChatRegistry::new), cx);
+pub fn init(window: &mut Window, cx: &mut App) {
+    ChatRegistry::set_global(cx.new(|cx| ChatRegistry::new(window, cx)), cx);
 }
 
 struct GlobalChatRegistry(Entity<ChatRegistry>);
@@ -45,11 +43,9 @@ pub enum ChatEvent {
 
 /// Channel signal.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
-enum NostrEvent {
+enum Signal {
     /// Message received from relay pool
     Message(NewMessage),
-    /// Unwrapping status
-    Unwrapping(bool),
     /// Eose received from relay pool
     Eose,
 }
@@ -60,23 +56,11 @@ pub struct ChatRegistry {
     /// Collection of all chat rooms
     rooms: Vec<Entity<Room>>,
 
-    /// Loading status of the registry
-    loading: bool,
-
-    /// Channel's sender for communication between nostr and gpui
-    sender: Sender<NostrEvent>,
-
     /// Tracking the status of unwrapping gift wrap events.
     tracking_flag: Arc<AtomicBool>,
 
-    /// Handle tracking asynchronous task
-    tracking: Option<Task<Result<(), Error>>>,
-
-    /// Handle notifications asynchronous task
-    notifications: Option<Task<()>>,
-
-    /// Tasks for asynchronous operations
-    tasks: Vec<Task<()>>,
+    /// Async tasks
+    tasks: SmallVec<[Task<Result<(), Error>>; 2]>,
 
     /// Subscriptions
     _subscriptions: SmallVec<[Subscription; 1]>,
@@ -96,82 +80,38 @@ impl ChatRegistry {
     }
 
     /// Create a new chat registry instance
-    fn new(cx: &mut Context<Self>) -> Self {
+    fn new(window: &mut Window, cx: &mut Context<Self>) -> Self {
         let nostr = NostrRegistry::global(cx);
-        let nip17_state = nostr.read(cx).nip17_state();
+        let nip65 = nostr.read(cx).nip65_state();
+        let nip17 = nostr.read(cx).nip17_state();
 
-        let device = DeviceRegistry::global(cx);
-        let device_signer = device.read(cx).device_signer.clone();
-
-        // A flag to indicate if the registry is loading
-        let tracking_flag = Arc::new(AtomicBool::new(false));
-
-        // Channel for communication between nostr and gpui
-        let (tx, rx) = flume::bounded::<NostrEvent>(2048);
-
-        let mut tasks = vec![];
         let mut subscriptions = smallvec![];
 
         subscriptions.push(
-            // Observe the identity
-            cx.observe(&nip17_state, |this, state, cx| {
-                if state.read(cx) == &RelayState::Configured {
-                    // Handle nostr notifications
-                    this.handle_notifications(cx);
-                    // Track unwrapping progress
-                    this.tracking(cx);
+            // Observe the nip65 state and load chat rooms on every state change
+            cx.observe(&nip65, |this, state, cx| {
+                if state.read(cx).idle() {
+                    this.reset(cx);
                 }
-                // Get chat rooms from the database on every identity change
+            }),
+        );
+
+        subscriptions.push(
+            // Observe the nip17 state and load chat rooms on every state change
+            cx.observe(&nip17, |this, _state, cx| {
                 this.get_rooms(cx);
             }),
         );
 
-        subscriptions.push(
-            // Observe the device signer state
-            cx.observe(&device_signer, |this, state, cx| {
-                if state.read(cx).is_some() {
-                    this.handle_notifications(cx);
-                }
-            }),
-        );
-
-        tasks.push(
-            // Update GPUI states
-            cx.spawn(async move |this, cx| {
-                while let Ok(message) = rx.recv_async().await {
-                    match message {
-                        NostrEvent::Message(message) => {
-                            this.update(cx, |this, cx| {
-                                this.new_message(message, cx);
-                            })
-                            .ok();
-                        }
-                        NostrEvent::Unwrapping(status) => {
-                            this.update(cx, |this, cx| {
-                                this.set_loading(status, cx);
-                                this.get_rooms(cx);
-                            })
-                            .ok();
-                        }
-                        NostrEvent::Eose => {
-                            this.update(cx, |this, cx| {
-                                this.get_rooms(cx);
-                            })
-                            .ok();
-                        }
-                    };
-                }
-            }),
-        );
+        cx.defer_in(window, |this, _window, cx| {
+            this.handle_notifications(cx);
+            this.tracking(cx);
+        });
 
         Self {
             rooms: vec![],
-            loading: false,
-            sender: tx.clone(),
-            tracking_flag,
-            tracking: None,
-            notifications: None,
-            tasks,
+            tracking_flag: Arc::new(AtomicBool::new(false)),
+            tasks: smallvec![],
             _subscriptions: subscriptions,
         }
     }
@@ -180,18 +120,18 @@ impl ChatRegistry {
     fn handle_notifications(&mut self, cx: &mut Context<Self>) {
         let nostr = NostrRegistry::global(cx);
         let client = nostr.read(cx).client();
-
-        let device = DeviceRegistry::global(cx);
-        let device_signer = device.read(cx).signer(cx);
-
+        let signer = nostr.read(cx).signer();
         let status = self.tracking_flag.clone();
-        let tx = self.sender.clone();
 
-        self.notifications = Some(cx.background_spawn(async move {
-            let initialized_at = Timestamp::now();
-            let sub_id1 = SubscriptionId::new(DEVICE_GIFTWRAP);
-            let sub_id2 = SubscriptionId::new(USER_GIFTWRAP);
+        let initialized_at = Timestamp::now();
+        let sub_id1 = SubscriptionId::new(DEVICE_GIFTWRAP);
+        let sub_id2 = SubscriptionId::new(USER_GIFTWRAP);
 
+        // Channel for communication between nostr and gpui
+        let (tx, rx) = flume::bounded::<Signal>(1024);
+
+        self.tasks.push(cx.background_spawn(async move {
+            let device_signer = signer.get_encryption_signer().await;
             let mut notifications = client.notifications();
             let mut processed_events = HashSet::new();
 
@@ -213,23 +153,16 @@ impl ChatRegistry {
                             continue;
                         }
 
+                        log::info!("Received gift wrap event: {:?}", event);
+
                         // Extract the rumor from the gift wrap event
                         match Self::extract_rumor(&client, &device_signer, event.as_ref()).await {
                             Ok(rumor) => match rumor.created_at >= initialized_at {
                                 true => {
-                                    // Check if the event is sent by coop
-                                    let sent_by_coop = {
-                                        let tracker = tracker().read().await;
-                                        tracker.is_sent_by_coop(&event.id)
-                                    };
-                                    // No need to emit if sent by coop
-                                    // the event is already emitted
-                                    if !sent_by_coop {
-                                        let new_message = NewMessage::new(event.id, rumor);
-                                        let signal = NostrEvent::Message(new_message);
+                                    let new_message = NewMessage::new(event.id, rumor);
+                                    let signal = Signal::Message(new_message);
 
-                                        tx.send_async(signal).await.ok();
-                                    }
+                                    tx.send_async(signal).await?;
                                 }
                                 false => {
                                     status.store(true, Ordering::Release);
@@ -242,29 +175,46 @@ impl ChatRegistry {
                     }
                     RelayMessage::EndOfStoredEvents(id) => {
                         if id.as_ref() == &sub_id1 || id.as_ref() == &sub_id2 {
-                            tx.send_async(NostrEvent::Eose).await.ok();
+                            tx.send_async(Signal::Eose).await?;
                         }
                     }
                     _ => {}
                 }
             }
+
+            Ok(())
+        }));
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            while let Ok(message) = rx.recv_async().await {
+                match message {
+                    Signal::Message(message) => {
+                        this.update(cx, |this, cx| {
+                            this.new_message(message, cx);
+                        })?;
+                    }
+                    Signal::Eose => {
+                        this.update(cx, |this, cx| {
+                            this.get_rooms(cx);
+                        })?;
+                    }
+                };
+            }
+
+            Ok(())
         }));
     }
 
     /// Tracking the status of unwrapping gift wrap events.
     fn tracking(&mut self, cx: &mut Context<Self>) {
         let status = self.tracking_flag.clone();
-        let tx = self.sender.clone();
 
-        self.tracking = Some(cx.background_spawn(async move {
-            let loop_duration = Duration::from_secs(12);
+        self.tasks.push(cx.background_spawn(async move {
+            let loop_duration = Duration::from_secs(10);
 
             loop {
                 if status.load(Ordering::Acquire) {
                     _ = status.compare_exchange(true, false, Ordering::Release, Ordering::Relaxed);
-                    tx.send_async(NostrEvent::Unwrapping(true)).await.ok();
-                } else {
-                    tx.send_async(NostrEvent::Unwrapping(false)).await.ok();
                 }
                 smol::Timer::after(loop_duration).await;
             }
@@ -273,13 +223,7 @@ impl ChatRegistry {
 
     /// Get the loading status of the chat registry
     pub fn loading(&self) -> bool {
-        self.loading
-    }
-
-    /// Set the loading status of the chat registry
-    pub fn set_loading(&mut self, loading: bool, cx: &mut Context<Self>) {
-        self.loading = loading;
-        cx.notify();
+        self.tracking_flag.load(Ordering::Acquire)
     }
 
     /// Get a weak reference to a room by its ID.
@@ -315,19 +259,19 @@ impl ChatRegistry {
         let nostr = NostrRegistry::global(cx);
         let client = nostr.read(cx).client();
 
-        self.tasks.push(cx.spawn(async move |this, cx| {
-            if let Some(signer) = client.signer() {
-                if let Ok(public_key) = signer.get_public_key().await {
-                    this.update(cx, |this, cx| {
-                        this.rooms
-                            .insert(0, cx.new(|_| room.into().organize(&public_key)));
-                        cx.emit(ChatEvent::Ping);
-                        cx.notify();
-                    })
-                    .ok();
-                }
-            }
-        }));
+        cx.spawn(async move |this, cx| {
+            let signer = client.signer()?;
+            let public_key = signer.get_public_key().await.ok()?;
+            let room: Room = room.into().organize(&public_key);
+
+            this.update(cx, |this, cx| {
+                this.rooms.insert(0, cx.new(|_| room));
+                cx.emit(ChatEvent::Ping);
+                cx.notify();
+            })
+            .ok()
+        })
+        .detach();
     }
 
     /// Emit an open room event.
@@ -420,20 +364,16 @@ impl ChatRegistry {
     pub fn get_rooms(&mut self, cx: &mut Context<Self>) {
         let task = self.get_rooms_from_database(cx);
 
-        self.tasks.push(cx.spawn(async move |this, cx| {
-            match task.await {
-                Ok(rooms) => {
-                    this.update(cx, move |this, cx| {
-                        this.extend_rooms(rooms, cx);
-                        this.sort(cx);
-                    })
-                    .ok();
-                }
-                Err(e) => {
-                    log::error!("Failed to load rooms: {e}")
-                }
-            };
-        }));
+        cx.spawn(async move |this, cx| {
+            let rooms = task.await.ok()?;
+
+            this.update(cx, move |this, cx| {
+                this.extend_rooms(rooms, cx);
+                this.sort(cx);
+            })
+            .ok()
+        })
+        .detach();
     }
 
     /// Create a task to load rooms from the database

crates/chat/src/message.rs

@@ -6,8 +6,8 @@ use nostr_sdk::prelude::*;
 /// New message.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
 pub struct NewMessage {
-    pub gift_wrap: EventId,
     pub room: u64,
+    pub gift_wrap: EventId,
     pub rumor: UnsignedEvent,
 }
 
@@ -16,8 +16,8 @@ impl NewMessage {
         let room = rumor.uniq_id();
 
         Self {
-            gift_wrap,
             room,
+            gift_wrap,
             rumor,
         }
     }

crates/chat/src/room.rs

@@ -1,5 +1,4 @@
 use std::cmp::Ordering;
-use std::collections::{HashMap, HashSet};
 use std::hash::{Hash, Hasher};
 use std::time::Duration;
 
@@ -9,73 +8,59 @@ use gpui::{App, AppContext, Context, EventEmitter, SharedString, Task};
 use itertools::Itertools;
 use nostr_sdk::prelude::*;
 use person::{Person, PersonRegistry};
-use state::{tracker, NostrRegistry};
+use settings::{RoomConfig, SignerKind};
+use state::{NostrRegistry, TIMEOUT};
 
 use crate::{ChatRegistry, NewMessage};
 
-const SEND_RETRY: usize = 10;
-
 #[derive(Debug, Clone)]
 pub struct SendReport {
     pub receiver: PublicKey,
-    pub status: Option<Output<EventId>>,
+    pub gift_wrap_id: Option<EventId>,
     pub error: Option<SharedString>,
-    pub on_hold: Option<Event>,
-    pub encryption: bool,
-    pub relays_not_found: bool,
-    pub device_not_found: bool,
+    pub output: Option<Output<EventId>>,
 }
 
 impl SendReport {
     pub fn new(receiver: PublicKey) -> Self {
         Self {
             receiver,
-            status: None,
+            gift_wrap_id: None,
             error: None,
-            on_hold: None,
-            encryption: false,
-            relays_not_found: false,
-            device_not_found: false,
+            output: None,
         }
     }
 
-    pub fn status(mut self, output: Output<EventId>) -> Self {
-        self.status = Some(output);
+    /// Set the gift wrap ID.
+    pub fn gift_wrap_id(mut self, gift_wrap_id: EventId) -> Self {
+        self.gift_wrap_id = Some(gift_wrap_id);
         self
     }
 
-    pub fn error(mut self, error: impl Into<SharedString>) -> Self {
+    /// Set the output.
+    pub fn output(mut self, output: Output<EventId>) -> Self {
+        self.output = Some(output);
+        self
+    }
+
+    /// Set the error message.
+    pub fn error<T>(mut self, error: T) -> Self
+    where
+        T: Into<SharedString>,
+    {
         self.error = Some(error.into());
         self
     }
 
-    pub fn on_hold(mut self, event: Event) -> Self {
-        self.on_hold = Some(event);
-        self
+    /// Returns true if the send is pending.
+    pub fn pending(&self) -> bool {
+        self.output.is_none() && self.error.is_none()
     }
 
-    pub fn encryption(mut self) -> Self {
-        self.encryption = true;
-        self
-    }
-
-    pub fn relays_not_found(mut self) -> Self {
-        self.relays_not_found = true;
-        self
-    }
-
-    pub fn device_not_found(mut self) -> Self {
-        self.device_not_found = true;
-        self
-    }
-
-    pub fn is_relay_error(&self) -> bool {
-        self.error.is_some() || self.relays_not_found
-    }
-
-    pub fn is_sent_success(&self) -> bool {
-        if let Some(output) = self.status.as_ref() {
-            !output.success.is_empty()
+    /// Returns true if the send was successful.
+    pub fn success(&self) -> bool {
+        if let Some(output) = self.output.as_ref() {
+            !output.failed.is_empty()
         } else {
             false
         }
@@ -115,6 +100,9 @@ pub struct Room {
 
     /// Kind
     pub kind: RoomKind,
+
+    /// Configuration
+    config: RoomConfig,
 }
 
 impl Ord for Room {
@@ -161,6 +149,7 @@ impl From<&UnsignedEvent> for Room {
             subject,
             members,
             kind: RoomKind::default(),
+            config: RoomConfig::default(),
         }
     }
 }
@@ -320,34 +309,43 @@ impl Room {
     }
 
     /// Get gossip relays for each member
-    pub fn connect(&self, cx: &App) -> Task<Result<(), Error>> {
+    pub fn early_connect(&self, cx: &App) -> Task<Result<(), Error>> {
         let nostr = NostrRegistry::global(cx);
         let client = nostr.read(cx).client();
+
         let members = self.members();
-        let id = SubscriptionId::new(format!("room-{}", self.id));
+        let subscription_id = SubscriptionId::new(format!("room-{}", self.id));
 
         cx.background_spawn(async move {
             let signer = client.signer().context("Signer not found")?;
             let public_key = signer.get_public_key().await?;
 
-            // Subscription options
-            let opts = SubscribeAutoCloseOptions::default()
-                .timeout(Some(Duration::from_secs(2)))
-                .exit_policy(ReqExitPolicy::ExitOnEOSE);
-
             for member in members.into_iter() {
                 if member == public_key {
                     continue;
                 };
 
-                // Construct a filter for gossip relays
-                let filter = Filter::new().kind(Kind::RelayList).author(member).limit(1);
+                // Construct a filter for messaging relays
+                let inbox = Filter::new()
+                    .kind(Kind::InboxRelays)
+                    .author(member)
+                    .limit(1);
+
+                // Construct a filter for announcement
+                let announcement = Filter::new()
+                    .kind(Kind::Custom(10044))
+                    .author(member)
+                    .limit(1);
 
                 // Subscribe to get member's gossip relays
                 client
-                    .subscribe(filter)
-                    .close_on(opts)
-                    .with_id(id.clone())
+                    .subscribe(vec![inbox, announcement])
+                    .with_id(subscription_id.clone())
+                    .close_on(
+                        SubscribeAutoCloseOptions::default()
+                            .timeout(Some(Duration::from_secs(TIMEOUT)))
+                            .exit_policy(ReqExitPolicy::ExitOnEOSE),
+                    )
                     .await?;
             }
 
@@ -379,7 +377,194 @@ impl Room {
         })
     }
 
-    /// Create a new unsigned message event
+    // Construct a rumor event for direct message
+    pub fn rumor<S, I>(&self, content: S, replies: I, cx: &App) -> Option<UnsignedEvent>
+    where
+        S: Into<String>,
+        I: IntoIterator<Item = EventId>,
+    {
+        let kind = Kind::PrivateDirectMessage;
+        let content: String = content.into();
+        let replies: Vec<EventId> = replies.into_iter().collect();
+
+        let persons = PersonRegistry::global(cx);
+        let nostr = NostrRegistry::global(cx);
+
+        // Get current user's public key
+        let sender = nostr.read(cx).signer().public_key()?;
+
+        // Get all members
+        let members: Vec<Person> = self
+            .members
+            .iter()
+            .filter(|public_key| public_key != &&sender)
+            .map(|member| persons.read(cx).get(member, cx))
+            .collect();
+
+        // Construct event's tags
+        let mut tags = vec![];
+
+        // Add subject tag if present
+        if let Some(value) = self.subject.as_ref() {
+            tags.push(Tag::from_standardized_without_cell(TagStandard::Subject(
+                value.to_string(),
+            )));
+        }
+
+        // Add all reply tags
+        for id in replies.into_iter() {
+            tags.push(Tag::event(id))
+        }
+
+        // Add all receiver tags
+        for member in members.into_iter() {
+            // Skip current user
+            if member.public_key() == sender {
+                continue;
+            }
+
+            tags.push(Tag::from_standardized_without_cell(
+                TagStandard::PublicKey {
+                    public_key: member.public_key(),
+                    relay_url: member.messaging_relay_hint(),
+                    alias: None,
+                    uppercase: false,
+                },
+            ));
+        }
+
+        // Construct a direct message rumor event
+        // WARNING: never sign and send this event to relays
+        let mut event = EventBuilder::new(kind, content).tags(tags).build(sender);
+
+        // Ensure that the ID is set
+        event.ensure_id();
+
+        Some(event)
+    }
+
+    /// Send rumor event to all members's messaging relays
+    pub fn send(&self, rumor: UnsignedEvent, cx: &App) -> Option<Task<Vec<SendReport>>> {
+        let persons = PersonRegistry::global(cx);
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+        let signer = nostr.read(cx).signer();
+
+        // Get room's config
+        let config = self.config.clone();
+
+        // Get current user's public key
+        let sender = nostr.read(cx).signer().public_key()?;
+
+        // Get all members (excluding sender)
+        let members: Vec<Person> = self
+            .members
+            .iter()
+            .filter(|public_key| public_key != &&sender)
+            .map(|member| persons.read(cx).get(member, cx))
+            .collect();
+
+        Some(cx.background_spawn(async move {
+            let signer_kind = config.signer_kind();
+            let user_signer = signer.get().await;
+            let encryption_signer = signer.get_encryption_signer().await;
+
+            let mut reports = Vec::new();
+
+            for member in members {
+                let relays = member.messaging_relays();
+                let announcement = member.announcement();
+
+                // Skip if member has no messaging relays
+                if relays.is_empty() {
+                    reports.push(SendReport::new(member.public_key()).error("No messaging relays"));
+                    continue;
+                }
+
+                // Ensure relay connections
+                for url in relays.iter() {
+                    client
+                        .add_relay(url)
+                        .and_connect()
+                        .capabilities(RelayCapabilities::GOSSIP)
+                        .await
+                        .ok();
+                }
+
+                // When forced to use encryption signer, skip if receiver has no announcement
+                if signer_kind.encryption() && announcement.is_none() {
+                    reports
+                        .push(SendReport::new(member.public_key()).error("Encryption not found"));
+                    continue;
+                }
+
+                // Determine receiver and signer based on signer kind
+                let (receiver, signer_to_use) = match signer_kind {
+                    SignerKind::Auto => {
+                        if let Some(announcement) = announcement {
+                            if let Some(enc_signer) = encryption_signer.as_ref() {
+                                (announcement.public_key(), enc_signer.clone())
+                            } else {
+                                (member.public_key(), user_signer.clone())
+                            }
+                        } else {
+                            (member.public_key(), user_signer.clone())
+                        }
+                    }
+                    SignerKind::Encryption => {
+                        let Some(encryption_signer) = encryption_signer.as_ref() else {
+                            reports.push(
+                                SendReport::new(member.public_key()).error("Encryption not found"),
+                            );
+                            continue;
+                        };
+                        let Some(announcement) = announcement else {
+                            reports.push(
+                                SendReport::new(member.public_key())
+                                    .error("Announcement not found"),
+                            );
+                            continue;
+                        };
+                        (announcement.public_key(), encryption_signer.clone())
+                    }
+                    SignerKind::User => (member.public_key(), user_signer.clone()),
+                };
+
+                // Create and send gift-wrapped event
+                match EventBuilder::gift_wrap(&signer_to_use, &receiver, rumor.clone(), []).await {
+                    Ok(event) => {
+                        match client
+                            .send_event(&event)
+                            .to(relays)
+                            .ack_policy(AckPolicy::none())
+                            .await
+                        {
+                            Ok(output) => {
+                                reports.push(
+                                    SendReport::new(member.public_key())
+                                        .gift_wrap_id(event.id)
+                                        .output(output),
+                                );
+                            }
+                            Err(e) => {
+                                reports.push(
+                                    SendReport::new(member.public_key()).error(e.to_string()),
+                                );
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        reports.push(SendReport::new(member.public_key()).error(e.to_string()));
+                    }
+                }
+            }
+
+            reports
+        }))
+    }
+
+    /*
+    * /// Create a new unsigned message event
     pub fn create_message(
         &self,
         content: &str,
@@ -444,7 +629,7 @@ impl Room {
             // WARNING: never sign and send this event to relays
             let mut event = EventBuilder::new(Kind::PrivateDirectMessage, content)
                 .tags(tags)
-                .build(Keys::generate().public_key());
+                .build(public_key);
 
             // Ensure the event ID has been generated
             event.ensure_id();
@@ -594,4 +779,5 @@ impl Room {
             Ok(resend_reports)
         })
     }
+    */
 }