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

feat: rewrite the nip-4e implementation (#1)
Some checks failed
Rust / build (macos-latest, stable) (push) Has been cancelled
Rust / build (ubuntu-latest, stable) (push) Has been cancelled
Rust / build (windows-latest, stable) (push) Has been cancelled

Browse Source 
Make NIP-4e a core feature, not an optional feature.

Note:
- The UI is broken and needs to be updated in a separate PR.

Reviewed-on: #1
This commit was merged in pull request #1.
This commit is contained in:
reya
2026-01-13 16:00:08 +08:00
parent bb455871e5
commit 75c3783522
50 changed files with 2818 additions and 3458 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
crates/chat/Cargo.toml
View File

@@ -7,8 +7,7 @@ publish.workspace = true
[dependencies]
common = { path = "../common" }
state = { path = "../state" }
account = { path = "../account" }
encryption = { path = "../encryption" }
device = { path = "../device" }
person = { path = "../person" }
settings = { path = "../settings" }

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

@@ -5,24 +5,26 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use account::Account;
use anyhow::{anyhow, Context as AnyhowContext, Error};
use common::{EventUtils, BOOTSTRAP_RELAYS, METADATA_BATCH_LIMIT};
use encryption::Encryption;
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};
pub use message::*;
use gpui::{
App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task, WeakEntity,
};
use nostr_sdk::prelude::*;
pub use room::*;
use settings::AppSettings;
use smallvec::{smallvec, SmallVec};
use state::{initialized_at, NostrRegistry, GIFTWRAP_SUBSCRIPTION};
use state::{tracker, NostrRegistry, GIFTWRAP_SUBSCRIPTION};
mod message;
mod room;
pub use message::*;
pub use room::*;
pub fn init(cx: &mut App) {
ChatRegistry::set_global(cx.new(ChatRegistry::new), cx);
}
@@ -31,37 +33,51 @@ struct GlobalChatRegistry(Entity<ChatRegistry>);
impl Global for GlobalChatRegistry {}
/// Chat event.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum ChatEvent {
/// An event to open a room by its ID
OpenRoom(u64),
/// An event to close a room by its ID
CloseRoom(u64),
/// An event to notify UI about a new chat request
Ping,
}
/// Channel signal.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
enum NostrEvent {
/// Message received from relay pool
Message(NewMessage),
/// Unwrapping status
Unwrapping(bool),
/// Eose received from relay pool
Eose,
}
/// Chat Registry
#[derive(Debug)]
pub struct ChatRegistry {
/// Collection of all chat rooms
pub rooms: Vec<Entity<Room>>,
rooms: Vec<Entity<Room>>,
/// Loading status of the registry
pub loading: bool,
loading: bool,
/// Async task for handling notifications
handle_notifications: Task<()>,
/// Tracking the status of unwrapping gift wrap events.
tracking_flag: Arc<AtomicBool>,
/// Event subscriptions
_subscriptions: SmallVec<[Subscription; 1]>,
/// Channel's sender for communication between nostr and gpui
sender: Sender<NostrEvent>,
/// Handle notifications asynchronous task
notifications: Option<Task<Result<(), Error>>>,
/// Tasks for asynchronous operations
_tasks: SmallVec<[Task<()>; 4]>,
}
tasks: Vec<Task<()>>,
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum ChatEvent {
OpenRoom(u64),
CloseRoom(u64),
NewChatRequest(RoomKind),
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum Signal {
Loading(bool),
Message(NewMessage),
Eose,
/// Subscriptions
_subscriptions: SmallVec<[Subscription; 1]>,
}
impl EventEmitter<ChatEvent> for ChatRegistry {}
@@ -79,81 +95,65 @@ impl ChatRegistry {
/// Create a new chat registry instance
fn new(cx: &mut Context<Self>) -> Self {
let encryption = Encryption::global(cx);
let encryption_key = encryption.read(cx).encryption.clone();
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let identity = nostr.read(cx).identity();
let status = Arc::new(AtomicBool::new(true));
let (tx, rx) = flume::bounded::<Signal>(2048);
let device = DeviceRegistry::global(cx);
let device_signer = device.read(cx).device_signer.clone();
let handle_notifications = cx.background_spawn({
let client = nostr.read(cx).client();
let status = Arc::clone(&status);
let tx = tx.clone();
let signer: Option<Arc<dyn NostrSigner>> = None;
// A flag to indicate if the registry is loading
let tracking_flag = Arc::new(AtomicBool::new(true));
async move { Self::handle_notifications(&client, &signer, &tx, &status).await }
});
// Channel for communication between nostr and gpui
let (tx, rx) = flume::bounded::<NostrEvent>(2048);
let mut tasks = vec![];
let mut subscriptions = smallvec![];
let mut tasks = smallvec![];
subscriptions.push(
// Observe the encryption global state
cx.observe(&encryption_key, {
let status = Arc::clone(&status);
let tx = tx.clone();
// Observe the identity
cx.observe(&identity, |this, state, cx| {
if state.read(cx).has_public_key() {
// Handle nostr notifications
this.handle_notifications(cx);
// Track unwrapping progress
this.tracking(cx);
}
}),
);
move |this, state, cx| {
if let Some(signer) = state.read(cx).clone() {
this.handle_notifications = cx.background_spawn({
let client = nostr.read(cx).client();
let status = Arc::clone(&status);
let tx = tx.clone();
let signer = Some(signer);
async move {
Self::handle_notifications(&client, &signer, &tx, &status).await
}
});
cx.notify();
}
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(
// Handle unwrapping status
cx.background_spawn(
async move { Self::handle_unwrapping(&client, &status, &tx).await },
),
);
tasks.push(
// Handle new messages
// Update GPUI states
cx.spawn(async move |this, cx| {
while let Ok(message) = rx.recv_async().await {
match message {
Signal::Message(message) => {
NostrEvent::Message(message) => {
this.update(cx, |this, cx| {
this.new_message(message, cx);
})
.expect("Entity has been released");
.ok();
}
Signal::Eose => {
NostrEvent::Eose => {
this.update(cx, |this, cx| {
this.get_rooms(cx);
})
.expect("Entity has been released");
.ok();
}
Signal::Loading(status) => {
NostrEvent::Unwrapping(status) => {
this.update(cx, |this, cx| {
this.set_loading(status, cx);
this.get_rooms(cx);
})
.expect("Entity has been released");
.ok();
}
};
}
@@ -163,127 +163,137 @@ impl ChatRegistry {
Self {
rooms: vec![],
loading: true,
handle_notifications,
tracking_flag,
sender: tx.clone(),
notifications: None,
tasks,
_subscriptions: subscriptions,
_tasks: tasks,
}
}
async fn handle_notifications<T>(
client: &Client,
signer: &Option<T>,
tx: &Sender<Signal>,
status: &Arc<AtomicBool>,
) where
T: NostrSigner,
{
let initialized_at = initialized_at();
let subscription_id = SubscriptionId::new(GIFTWRAP_SUBSCRIPTION);
/// Handle nostr notifications
fn handle_notifications(&mut self, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let mut notifications = client.notifications();
let mut public_keys = HashSet::new();
let mut processed_events = HashSet::new();
let device = DeviceRegistry::global(cx);
let device_signer = device.read(cx).signer(cx);
while let Ok(notification) = notifications.recv().await {
let RelayPoolNotification::Message { message, .. } = notification else {
// Skip non-message notifications
continue;
};
let status = self.tracking_flag.clone();
let tx = self.sender.clone();
match message {
RelayMessage::Event { event, .. } => {
if !processed_events.insert(event.id) {
// Skip if the event has already been processed
continue;
}
self.tasks.push(cx.background_spawn(async move {
let initialized_at = Timestamp::now();
let subscription_id = SubscriptionId::new(GIFTWRAP_SUBSCRIPTION);
if event.kind != Kind::GiftWrap {
// Skip non-gift wrap events
continue;
}
let mut notifications = client.notifications();
let mut processed_events = HashSet::new();
// Extract the rumor from the gift wrap event
match Self::extract_rumor(client, signer, event.as_ref()).await {
Ok(rumor) => {
// Get all public keys
public_keys.extend(rumor.all_pubkeys());
while let Ok(notification) = notifications.recv().await {
let RelayPoolNotification::Message { message, .. } = notification else {
// Skip non-message notifications
continue;
};
let limit_reached = public_keys.len() >= METADATA_BATCH_LIMIT;
let done = !status.load(Ordering::Acquire) && !public_keys.is_empty();
match message {
RelayMessage::Event { event, .. } => {
if !processed_events.insert(event.id) {
// Skip if the event has already been processed
continue;
}
// Get metadata for all public keys if the limit is reached
if limit_reached || done {
let public_keys = std::mem::take(&mut public_keys);
// Get metadata for the public keys
Self::get_metadata(client, public_keys).await.ok();
}
if event.kind != Kind::GiftWrap {
// Skip non-gift wrap events
continue;
}
match &rumor.created_at >= initialized_at {
// 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 => {
let new_message = NewMessage::new(event.id, rumor);
let signal = Signal::Message(new_message);
// 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);
if let Err(e) = tx.send_async(signal).await {
log::error!("Failed to send signal: {}", e);
tx.send_async(signal).await.ok();
}
}
false => {
status.store(true, Ordering::Release);
}
},
Err(e) => {
log::warn!("Failed to unwrap: {e}");
}
}
Err(e) => {
log::warn!("Failed to unwrap gift wrap event: {}", e);
}
RelayMessage::EndOfStoredEvents(id) => {
if id.as_ref() == &subscription_id {
tx.send_async(NostrEvent::Eose).await.ok();
}
}
_ => {}
}
RelayMessage::EndOfStoredEvents(id) => {
if id.as_ref() == &subscription_id {
if let Err(e) = tx.send_async(Signal::Eose).await {
log::error!("Failed to send signal: {}", e);
}
}
}
_ => {}
}
}
}));
}
async fn handle_unwrapping(client: &Client, status: &Arc<AtomicBool>, tx: &Sender<Signal>) {
let loop_duration = Duration::from_secs(20);
let mut is_start_processing = false;
let mut total_loops = 0;
/// Tracking the status of unwrapping gift wrap events.
fn tracking(&mut self, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
loop {
if client.has_signer().await {
total_loops += 1;
let status = self.tracking_flag.clone();
let tx = self.sender.clone();
if status.load(Ordering::Acquire) {
is_start_processing = true;
self.notifications = Some(cx.background_spawn(async move {
let loop_duration = Duration::from_secs(12);
// Reset gift wrap processing flag
_ = status.compare_exchange(true, false, Ordering::Release, Ordering::Relaxed);
let mut is_start_processing = false;
let mut total_loops = 0;
// Send loading signal
if let Err(e) = tx.send_async(Signal::Loading(true)).await {
log::error!("Failed to send signal: {}", e);
}
} else {
// Only run further if we are already processing
// Wait until after 2 loops to prevent exiting early while events are still being processed
if is_start_processing && total_loops >= 2 {
// Send loading signal
if let Err(e) = tx.send_async(Signal::Loading(false)).await {
log::error!("Failed to send signal: {}", e);
loop {
if client.has_signer().await {
total_loops += 1;
if status.load(Ordering::Acquire) {
is_start_processing = true;
// Reset gift wrap processing flag
_ = status.compare_exchange(
true,
false,
Ordering::Release,
Ordering::Relaxed,
);
tx.send_async(NostrEvent::Unwrapping(true)).await.ok();
} else {
// Only run further if we are already processing
// Wait until after 2 loops to prevent exiting early while events are still being processed
if is_start_processing && total_loops >= 2 {
tx.send_async(NostrEvent::Unwrapping(false)).await.ok();
// Reset the counter
is_start_processing = false;
total_loops = 0;
}
// Reset the counter
is_start_processing = false;
total_loops = 0;
}
}
smol::Timer::after(loop_duration).await;
}
smol::Timer::after(loop_duration).await;
}
}));
}
/// Get the loading status of the chat registry
pub fn loading(&self) -> bool {
self.loading
}
/// Set the loading status of the chat registry
@@ -292,12 +302,12 @@ impl ChatRegistry {
cx.notify();
}
/// Get a room by its ID.
pub fn room(&self, id: &u64, cx: &App) -> Option<Entity<Room>> {
/// Get a weak reference to a room by its ID.
pub fn room(&self, id: &u64, cx: &App) -> Option<WeakEntity<Room>> {
self.rooms
.iter()
.find(|model| model.read(cx).id == *id)
.cloned()
.find(|this| &this.read(cx).id == id)
.map(|this| this.downgrade())
}
/// Get all ongoing rooms.
@@ -319,11 +329,30 @@ impl ChatRegistry {
}
/// Add a new room to the start of list.
pub fn add_room(&mut self, room: Entity<Room>, cx: &mut Context<Self>) {
self.rooms.insert(0, room);
pub fn add_room<I>(&mut self, room: I, cx: &mut Context<Self>)
where
I: Into<Room>,
{
self.rooms.insert(0, cx.new(|_| room.into()));
cx.notify();
}
/// Emit an open room event.
/// If the room is new, add it to the registry.
pub fn emit_room(&mut self, room: WeakEntity<Room>, cx: &mut Context<Self>) {
if let Some(room) = room.upgrade() {
let id = room.read(cx).id;
// If the room is new, add it to the registry.
if !self.rooms.iter().any(|r| r.read(cx).id == id) {
self.rooms.insert(0, room);
}
// Emit the open room event.
cx.emit(ChatEvent::OpenRoom(id));
}
}
/// Close a room.
pub fn close_room(&mut self, id: u64, cx: &mut Context<Self>) {
if self.rooms.iter().any(|r| r.read(cx).id == id) {
@@ -367,17 +396,6 @@ impl ChatRegistry {
cx.notify();
}
/// Push a new room to the chat registry
pub fn push_room(&mut self, room: Entity<Room>, cx: &mut Context<Self>) {
let id = room.read(cx).id;
if !self.rooms.iter().any(|r| r.read(cx).id == id) {
self.add_room(room, cx);
}
cx.emit(ChatEvent::OpenRoom(id));
}
/// Extend the registry with new rooms.
fn extend_rooms(&mut self, rooms: HashSet<Room>, cx: &mut Context<Self>) {
let mut room_map: HashMap<u64, usize> = self
@@ -410,7 +428,7 @@ impl ChatRegistry {
pub fn get_rooms(&mut self, cx: &mut Context<Self>) {
let task = self.create_get_rooms_task(cx);
self._tasks.push(
self.tasks.push(
// Run and finished in the background
cx.spawn(async move |this, cx| {
match task.await {
@@ -528,59 +546,61 @@ impl ChatRegistry {
/// If the room doesn't exist, it will be created.
/// Updates room ordering based on the most recent messages.
pub fn new_message(&mut self, message: NewMessage, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
// Get the unique id
let id = message.rumor.uniq_id();
// Get the author
let author = message.rumor.pubkey;
let account = Account::global(cx);
if let Some(room) = self.rooms.iter().find(|room| room.read(cx).id == id) {
let is_new_event = message.rumor.created_at > room.read(cx).created_at;
let created_at = message.rumor.created_at;
let event_for_emit = message.rumor.clone();
match self.rooms.iter().find(|room| room.read(cx).id == id) {
Some(room) => {
let new_message = message.rumor.created_at > room.read(cx).created_at;
let created_at = message.rumor.created_at;
// Update room
room.update(cx, |this, cx| {
if is_new_event {
this.set_created_at(created_at, cx);
// Update room
room.update(cx, |this, cx| {
// Update the last timestamp if the new message is newer
if new_message {
this.set_created_at(created_at, cx);
}
// Set this room is ongoing if the new message is from current user
if author == nostr.read(cx).identity().read(cx).public_key() {
this.set_ongoing(cx);
}
// Emit the new message to the room
this.emit_message(message, cx);
});
// Resort all rooms in the registry by their created at (after updated)
if new_message {
self.sort(cx);
}
// Set this room is ongoing if the new message is from current user
if author == account.read(cx).public_key() {
this.set_ongoing(cx);
}
// Emit the new message to the room
this.emit_message(message.gift_wrap, event_for_emit.clone(), cx);
});
// Resort all rooms in the registry by their created at (after updated)
if is_new_event {
self.sort(cx);
}
} else {
// Push the new room to the front of the list
self.add_room(cx.new(|_| Room::from(&message.rumor)), cx);
None => {
// Push the new room to the front of the list
self.add_room(&message.rumor, cx);
// Notify the UI about the new room
cx.emit(ChatEvent::NewChatRequest(RoomKind::default()));
// Notify the UI about the new room
cx.emit(ChatEvent::Ping);
}
}
}
// Unwraps a gift-wrapped event and processes its contents.
async fn extract_rumor<T>(
async fn extract_rumor(
client: &Client,
signer: &Option<T>,
device_signer: &Option<Arc<dyn NostrSigner>>,
gift_wrap: &Event,
) -> Result<UnsignedEvent, Error>
where
T: NostrSigner,
{
) -> Result<UnsignedEvent, Error> {
// Try to get cached rumor first
if let Ok(event) = Self::get_rumor(client, gift_wrap.id).await {
return Ok(event);
}
// Try to unwrap with the available signer
let unwrapped = Self::try_unwrap(client, signer, gift_wrap).await?;
let unwrapped = Self::try_unwrap(client, device_signer, gift_wrap).await?;
let mut rumor_unsigned = unwrapped.rumor;
// Generate event id for the rumor if it doesn't have one
@@ -593,37 +613,26 @@ impl ChatRegistry {
}
// Helper method to try unwrapping with different signers
async fn try_unwrap<T>(
async fn try_unwrap(
client: &Client,
signer: &Option<T>,
device_signer: &Option<Arc<dyn NostrSigner>>,
gift_wrap: &Event,
) -> Result<UnwrappedGift, Error>
where
T: NostrSigner,
{
if let Some(custom_signer) = signer.as_ref() {
if let Ok(seal) = custom_signer
) -> Result<UnwrappedGift, Error> {
if let Some(signer) = device_signer.as_ref() {
let seal = signer
.nip44_decrypt(&gift_wrap.pubkey, &gift_wrap.content)
.await
{
let seal: Event = Event::from_json(seal)?;
seal.verify_with_ctx(&SECP256K1)?;
.await?;
// Decrypt the rumor
// TODO: verify the sender
let rumor = custom_signer
.nip44_decrypt(&seal.pubkey, &seal.content)
.await?;
let seal: Event = Event::from_json(seal)?;
seal.verify_with_ctx(&SECP256K1)?;
// Construct the unsigned event
let rumor = UnsignedEvent::from_json(rumor)?;
let rumor = signer.nip44_decrypt(&seal.pubkey, &seal.content).await?;
let rumor = UnsignedEvent::from_json(rumor)?;
// Return the unwrapped gift
return Ok(UnwrappedGift {
sender: rumor.pubkey,
rumor,
});
}
return Ok(UnwrappedGift {
sender: seal.pubkey,
rumor,
});
}
let signer = client.signer().await?;
@@ -695,33 +704,6 @@ impl ChatRegistry {
}
}
/// Get metadata for a list of public keys
async fn get_metadata<I>(client: &Client, public_keys: I) -> Result<(), Error>
where
I: IntoIterator<Item = PublicKey>,
{
let authors: Vec<PublicKey> = public_keys.into_iter().collect();
let opts = SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE);
let kinds = vec![Kind::Metadata, Kind::ContactList];
// Return if the list is empty
if authors.is_empty() {
return Err(anyhow!("You need at least one public key".to_string(),));
}
let filter = Filter::new()
.limit(authors.len() * kinds.len())
.authors(authors)
.kinds(kinds);
// Subscribe to filters to the bootstrap relays
client
.subscribe_to(BOOTSTRAP_RELAYS, filter, Some(opts))
.await?;
Ok(())
}
/// Get the conversation ID for a given rumor (message).
fn conversation_id(rumor: &UnsignedEvent) -> u64 {
let mut hasher = DefaultHasher::new();

80
crates/chat/src/message.rs
View File

@@ -2,6 +2,7 @@ use std::hash::Hash;
use nostr_sdk::prelude::*;
/// New message.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct NewMessage {
pub gift_wrap: EventId,
@@ -14,6 +15,7 @@ impl NewMessage {
}
}
/// Message.
#[derive(Debug, Clone, Hash, PartialEq, Eq)]
pub enum Message {
User(RenderedMessage),
@@ -22,11 +24,17 @@ pub enum Message {
}
impl Message {
pub fn user(user: impl Into<RenderedMessage>) -> Self {
pub fn user<I>(user: I) -> Self
where
I: Into<RenderedMessage>,
{
Self::User(user.into())
}
pub fn warning(content: impl Into<String>) -> Self {
pub fn warning<I>(content: I) -> Self
where
I: Into<String>,
{
Self::Warning(content.into(), Timestamp::now())
}
@@ -43,6 +51,18 @@ impl Message {
}
}
impl From<&NewMessage> for Message {
fn from(val: &NewMessage) -> Self {
Self::User(val.into())
}
}
impl From<&UnsignedEvent> for Message {
fn from(val: &UnsignedEvent) -> Self {
Self::User(val.into())
}
}
impl Ord for Message {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
match (self, other) {
@@ -63,6 +83,7 @@ impl PartialOrd for Message {
}
}
/// Rendered message.
#[derive(Debug, Clone)]
pub struct RenderedMessage {
pub id: EventId,
@@ -78,48 +99,53 @@ pub struct RenderedMessage {
pub replies_to: Vec<EventId>,
}
impl From<Event> for RenderedMessage {
fn from(inner: Event) -> Self {
let mentions = extract_mentions(&inner.content);
let replies_to = extract_reply_ids(&inner.tags);
impl From<&Event> for RenderedMessage {
fn from(val: &Event) -> Self {
let mentions = extract_mentions(&val.content);
let replies_to = extract_reply_ids(&val.tags);
Self {
id: inner.id,
author: inner.pubkey,
content: inner.content,
created_at: inner.created_at,
id: val.id,
author: val.pubkey,
content: val.content.clone(),
created_at: val.created_at,
mentions,
replies_to,
}
}
}
impl From<UnsignedEvent> for RenderedMessage {
fn from(inner: UnsignedEvent) -> Self {
let mentions = extract_mentions(&inner.content);
let replies_to = extract_reply_ids(&inner.tags);
impl From<&UnsignedEvent> for RenderedMessage {
fn from(val: &UnsignedEvent) -> Self {
let mentions = extract_mentions(&val.content);
let replies_to = extract_reply_ids(&val.tags);
Self {
// Event ID must be known
id: inner.id.unwrap(),
author: inner.pubkey,
content: inner.content,
created_at: inner.created_at,
id: val.id.unwrap(),
author: val.pubkey,
content: val.content.clone(),
created_at: val.created_at,
mentions,
replies_to,
}
}
}
impl From<Box<Event>> for RenderedMessage {
fn from(inner: Box<Event>) -> Self {
(*inner).into()
}
}
impl From<&NewMessage> for RenderedMessage {
fn from(val: &NewMessage) -> Self {
let mentions = extract_mentions(&val.rumor.content);
let replies_to = extract_reply_ids(&val.rumor.tags);
impl From<&Box<Event>> for RenderedMessage {
fn from(inner: &Box<Event>) -> Self {
inner.to_owned().into()
Self {
// Event ID must be known
id: val.rumor.id.unwrap(),
author: val.rumor.pubkey,
content: val.rumor.content.clone(),
created_at: val.rumor.created_at,
mentions,
replies_to,
}
}
}
@@ -149,6 +175,7 @@ impl Hash for RenderedMessage {
}
}
/// Extracts all mentions (public keys) from a content string.
fn extract_mentions(content: &str) -> Vec<PublicKey> {
let parser = NostrParser::new();
let tokens = parser.parse(content);
@@ -165,6 +192,7 @@ fn extract_mentions(content: &str) -> Vec<PublicKey> {
.collect::<Vec<_>>()
}
/// Extracts all reply (ids) from the event tags.
fn extract_reply_ids(inner: &Tags) -> Vec<EventId> {
let mut replies_to = vec![];

303
crates/chat/src/room.rs
View File

@@ -3,43 +3,18 @@ use std::collections::{HashMap, HashSet};
use std::hash::{Hash, Hasher};
use std::time::Duration;
use account::Account;
use anyhow::{anyhow, Error};
use common::{EventUtils, RenderedProfile};
use encryption::{Encryption, SignerKind};
use anyhow::Error;
use common::EventUtils;
use gpui::{App, AppContext, Context, EventEmitter, SharedString, Task};
use itertools::Itertools;
use nostr_sdk::prelude::*;
use person::PersonRegistry;
use state::NostrRegistry;
use person::{Person, PersonRegistry};
use state::{tracker, NostrRegistry};
use crate::NewMessage;
const SEND_RETRY: usize = 10;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct SendOptions {
pub backup: bool,
pub signer_kind: SignerKind,
}
impl SendOptions {
pub fn new() -> Self {
Self {
backup: true,
signer_kind: SignerKind::default(),
}
}
pub fn backup(&self) -> bool {
self.backup
}
}
impl Default for SendOptions {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug, Clone)]
pub struct SendReport {
pub receiver: PublicKey,
@@ -107,17 +82,21 @@ impl SendReport {
}
}
#[derive(Debug, Clone)]
pub enum RoomSignal {
NewMessage((EventId, UnsignedEvent)),
Refresh,
/// Room event.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum RoomEvent {
/// Incoming message.
Incoming(NewMessage),
/// Reloads the current room's messages.
Reload,
}
/// Room kind.
#[derive(Clone, Copy, Hash, Debug, PartialEq, Eq, PartialOrd, Ord, Default)]
pub enum RoomKind {
Ongoing,
#[default]
Request,
Ongoing,
}
#[derive(Debug)]
@@ -160,7 +139,7 @@ impl Hash for Room {
impl Eq for Room {}
impl EventEmitter<RoomSignal> for Room {}
impl EventEmitter<RoomEvent> for Room {}
impl From<&UnsignedEvent> for Room {
fn from(val: &UnsignedEvent) -> Self {
@@ -168,7 +147,7 @@ impl From<&UnsignedEvent> for Room {
let created_at = val.created_at;
// Get the members from the event's tags and event's pubkey
let members = val.all_pubkeys();
let members = val.extract_public_keys();
// Get subject from tags
let subject = val
@@ -248,6 +227,28 @@ impl Room {
self.members.clone()
}
/// Returns the members of the room with their messaging relays
pub fn members_with_relays(&self, cx: &App) -> Task<Vec<(PublicKey, Vec<RelayUrl>)>> {
let nostr = NostrRegistry::global(cx);
let mut tasks = vec![];
for member in self.members.iter() {
let task = nostr.read(cx).messaging_relays(member, cx);
tasks.push((*member, task));
}
cx.background_spawn(async move {
let mut results = vec![];
for (public_key, task) in tasks.into_iter() {
let urls = task.await;
results.push((public_key, urls));
}
results
})
}
/// Checks if the room has more than two members (group)
pub fn is_group(&self) -> bool {
self.members.len() > 2
@@ -263,9 +264,9 @@ impl Room {
}
/// Gets the display image for the room
pub fn display_image(&self, proxy: bool, cx: &App) -> SharedString {
pub fn display_image(&self, cx: &App) -> SharedString {
if !self.is_group() {
self.display_member(cx).avatar(proxy)
self.display_member(cx).avatar()
} else {
SharedString::from("brand/group.png")
}
@@ -274,10 +275,10 @@ impl Room {
/// Get a member to represent the room
///
/// Display member is always different from the current user.
pub fn display_member(&self, cx: &App) -> Profile {
pub fn display_member(&self, cx: &App) -> Person {
let persons = PersonRegistry::global(cx);
let account = Account::global(cx);
let public_key = account.read(cx).public_key();
let nostr = NostrRegistry::global(cx);
let public_key = nostr.read(cx).identity().read(cx).public_key();
let target_member = self
.members
@@ -286,7 +287,7 @@ impl Room {
.or_else(|| self.members.first())
.expect("Room should have at least one member");
persons.read(cx).get_person(target_member, cx)
persons.read(cx).get(target_member, cx)
}
/// Merge the names of the first two members of the room.
@@ -294,10 +295,10 @@ impl Room {
let persons = PersonRegistry::global(cx);
if self.is_group() {
let profiles: Vec<Profile> = self
let profiles: Vec<Person> = self
.members
.iter()
.map(|public_key| persons.read(cx).get_person(public_key, cx))
.map(|public_key| persons.read(cx).get(public_key, cx))
.collect();
let mut name = profiles
@@ -313,18 +314,18 @@ impl Room {
SharedString::from(name)
} else {
self.display_member(cx).display_name()
self.display_member(cx).name()
}
}
/// Emits a new message signal to the current room
pub fn emit_message(&self, id: EventId, event: UnsignedEvent, cx: &mut Context<Self>) {
cx.emit(RoomSignal::NewMessage((id, event)));
pub fn emit_message(&self, message: NewMessage, cx: &mut Context<Self>) {
cx.emit(RoomEvent::Incoming(message));
}
/// Emits a signal to refresh the current room's messages.
/// Emits a signal to reload the current room's messages.
pub fn emit_refresh(&mut self, cx: &mut Context<Self>) {
cx.emit(RoomSignal::Refresh);
cx.emit(RoomEvent::Reload);
}
/// Get gossip relays for each member
@@ -332,11 +333,16 @@ impl Room {
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let members = self.members();
let id = SubscriptionId::new(format!("room-{}", self.id));
cx.background_spawn(async move {
let signer = client.signer().await?;
let public_key = signer.get_public_key().await?;
let opts = SubscribeAutoCloseOptions::default().exit_policy(ReqExitPolicy::ExitOnEOSE);
// 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 {
@@ -347,7 +353,9 @@ impl Room {
let filter = Filter::new().kind(Kind::RelayList).author(member).limit(1);
// Subscribe to get member's gossip relays
client.subscribe(filter, Some(opts)).await?;
client
.subscribe_with_id(id.clone(), filter, Some(opts))
.await?;
}
Ok(())
@@ -381,12 +389,9 @@ impl Room {
/// Create a new message event (unsigned)
pub fn create_message(&self, content: &str, replies: &[EventId], cx: &App) -> UnsignedEvent {
let nostr = NostrRegistry::global(cx);
let gossip = nostr.read(cx).gossip();
let read_gossip = gossip.read_blocking();
// Get current user
let account = Account::global(cx);
let public_key = account.read(cx).public_key();
let public_key = nostr.read(cx).identity().read(cx).public_key();
// Get room's subject
let subject = self.subject.clone();
@@ -398,7 +403,7 @@ impl Room {
// NOTE: current user will be removed from the list of receivers
for member in self.members.iter() {
// Get relay hint if available
let relay_url = read_gossip.messaging_relays(member).first().cloned();
let relay_url = nostr.read(cx).relay_hint(member, cx);
// Construct a public key tag with relay hint
let tag = TagStandard::PublicKey {
@@ -449,98 +454,65 @@ impl Room {
pub fn send_message(
&self,
rumor: &UnsignedEvent,
opts: &SendOptions,
cx: &App,
) -> Task<Result<Vec<SendReport>, Error>> {
let encryption = Encryption::global(cx);
let encryption_key = encryption.read(cx).encryption_key(cx);
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let gossip = nostr.read(cx).gossip();
let tracker = nostr.read(cx).tracker();
// Get current user's public key and relays
let current_user = nostr.read(cx).identity().read(cx).public_key();
let current_user_relays = nostr.read(cx).messaging_relays(&current_user, cx);
let rumor = rumor.to_owned();
let opts = opts.to_owned();
// Get all members
let mut members = self.members();
// Get all members and their messaging relays
let task = self.members_with_relays(cx);
cx.background_spawn(async move {
let signer_kind = opts.signer_kind;
let gossip = gossip.read().await;
// Get current user's signer and public key
let user_signer = client.signer().await?;
let user_pubkey = user_signer.get_public_key().await?;
// Get the encryption public key
let encryption_pubkey = if let Some(signer) = encryption_key.as_ref() {
signer.get_public_key().await.ok()
} else {
None
};
let signer = client.signer().await?;
let current_user_relays = current_user_relays.await;
let mut members = task.await;
// Remove the current user's public key from the list of receivers
// the current user will be handled separately
members.retain(|&pk| pk != user_pubkey);
// Determine the signer will be used based on the provided options
let signer = Self::select_signer(&opts.signer_kind, user_signer, encryption_key)?;
members.retain(|(this, _)| this != &current_user);
// Collect the send reports
let mut reports: Vec<SendReport> = vec![];
for member in members.into_iter() {
// Get user's messaging relays
let urls = gossip.messaging_relays(&member);
// Get user's encryption public key if available
let encryption = gossip.announcement(&member).map(|a| a.public_key());
for (receiver, relays) in members.into_iter() {
// Check if there are any relays to send the message to
if urls.is_empty() {
reports.push(SendReport::new(member).relays_not_found());
if relays.is_empty() {
reports.push(SendReport::new(receiver).relays_not_found());
continue;
}
// Skip sending if using encryption signer but receiver's encryption keys not found
if encryption.is_none() && matches!(signer_kind, SignerKind::Encryption) {
reports.push(SendReport::new(member).device_not_found());
continue;
// Ensure relay connection
for url in relays.iter() {
client.add_relay(url).await?;
client.connect_relay(url).await?;
}
// Ensure connections to the relays
gossip.ensure_connections(&client, &urls).await;
// Determine the receiver based on the signer kind
let receiver = Self::select_receiver(&signer_kind, member, encryption)?;
// Construct the gift wrap event
let event = EventBuilder::gift_wrap(
&signer,
&receiver,
rumor.clone(),
vec![Tag::public_key(member)],
)
.await?;
let event =
EventBuilder::gift_wrap(&signer, &receiver, rumor.clone(), vec![]).await?;
// Send the gift wrap event to the messaging relays
match client.send_event_to(urls, &event).await {
match client.send_event_to(relays, &event).await {
Ok(output) => {
let id = output.id().to_owned();
let auth = output.failed.iter().any(|(_, s)| s.starts_with("auth-"));
let report = SendReport::new(receiver).status(output);
let tracker = tracker().read().await;
if auth {
// Wait for authenticated and resent event successfully
for attempt in 0..=SEND_RETRY {
let tracker = tracker.read().await;
let ids = tracker.resent_ids();
// Check if event was successfully resent
if let Some(output) = ids.iter().find(|e| e.id() == &id).cloned() {
let output = SendReport::new(receiver).status(output);
reports.push(output);
if tracker.is_sent_by_coop(&id) {
let output = Output::new(id);
let report = SendReport::new(receiver).status(output);
reports.push(report);
break;
}
@@ -562,55 +534,35 @@ impl Room {
}
}
// Return early if the user disabled backup.
//
// Coop will not send a gift wrap event to the current user.
if !opts.backup() {
return Ok(reports);
}
// Skip sending if using encryption signer but receiver's encryption keys not found
if encryption_pubkey.is_none() && matches!(signer_kind, SignerKind::Encryption) {
reports.push(SendReport::new(user_pubkey).device_not_found());
return Ok(reports);
}
// Determine the receiver based on the signer kind
let receiver = Self::select_receiver(&signer_kind, user_pubkey, encryption_pubkey)?;
// Construct the gift-wrapped event
let event = EventBuilder::gift_wrap(
&signer,
&receiver,
rumor.clone(),
vec![Tag::public_key(user_pubkey)],
)
.await?;
let event =
EventBuilder::gift_wrap(&signer, &current_user, rumor.clone(), vec![]).await?;
// Only send a backup message to current user if sent successfully to others
if reports.iter().all(|r| r.is_sent_success()) {
let urls = gossip.messaging_relays(&user_pubkey);
// Check if there are any relays to send the event to
if urls.is_empty() {
reports.push(SendReport::new(user_pubkey).relays_not_found());
if current_user_relays.is_empty() {
reports.push(SendReport::new(current_user).relays_not_found());
return Ok(reports);
}
// Ensure connections to the relays
gossip.ensure_connections(&client, &urls).await;
// Ensure relay connection
for url in current_user_relays.iter() {
client.add_relay(url).await?;
client.connect_relay(url).await?;
}
// Send the event to the messaging relays
match client.send_event_to(urls, &event).await {
match client.send_event_to(current_user_relays, &event).await {
Ok(output) => {
reports.push(SendReport::new(user_pubkey).status(output));
reports.push(SendReport::new(current_user).status(output));
}
Err(e) => {
reports.push(SendReport::new(user_pubkey).error(e.to_string()));
reports.push(SendReport::new(current_user).error(e.to_string()));
}
}
} else {
reports.push(SendReport::new(user_pubkey).on_hold(event));
reports.push(SendReport::new(current_user).on_hold(event));
}
Ok(reports)
@@ -625,10 +577,8 @@ impl Room {
) -> Task<Result<Vec<SendReport>, Error>> {
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let gossip = nostr.read(cx).gossip();
cx.background_spawn(async move {
let gossip = gossip.read().await;
let mut resend_reports = vec![];
for report in reports.into_iter() {
@@ -657,23 +607,13 @@ impl Room {
// Process the on hold event if it exists
if let Some(event) = report.on_hold {
let urls = gossip.messaging_relays(&receiver);
// Check if there are any relays to send the event to
if urls.is_empty() {
resend_reports.push(SendReport::new(receiver).relays_not_found());
} else {
// Ensure connections to the relays
gossip.ensure_connections(&client, &urls).await;
// Send the event to the messaging relays
match client.send_event_to(urls, &event).await {
Ok(output) => {
resend_reports.push(SendReport::new(receiver).status(output));
}
Err(e) => {
resend_reports.push(SendReport::new(receiver).error(e.to_string()));
}
// Send the event to the messaging relays
match client.send_event(&event).await {
Ok(output) => {
resend_reports.push(SendReport::new(receiver).status(output));
}
Err(e) => {
resend_reports.push(SendReport::new(receiver).error(e.to_string()));
}
}
}
@@ -682,31 +622,4 @@ impl Room {
Ok(resend_reports)
})
}
fn select_signer<T>(kind: &SignerKind, user: T, encryption: Option<T>) -> Result<T, Error>
where
T: NostrSigner,
{
match kind {
SignerKind::Encryption => {
Ok(encryption.ok_or_else(|| anyhow!("No encryption key found"))?)
}
SignerKind::User => Ok(user),
SignerKind::Auto => Ok(encryption.unwrap_or(user)),
}
}
fn select_receiver(
kind: &SignerKind,
member: PublicKey,
encryption: Option<PublicKey>,
) -> Result<PublicKey, Error> {
match kind {
SignerKind::Encryption => {
Ok(encryption.ok_or_else(|| anyhow!("Receiver's encryption key not found"))?)
}
SignerKind::User => Ok(member),
SignerKind::Auto => Ok(encryption.unwrap_or(member)),
}
}
}