diff --git a/crates/community/src/community.rs b/crates/community/src/community.rs index e5cd9a24..4e138493 100644 --- a/crates/community/src/community.rs +++ b/crates/community/src/community.rs @@ -349,7 +349,9 @@ impl Community { } fn start_round(&mut self, channel: ChannelId, intent: Intent, cx: &mut Context) { - let client = NostrRegistry::global(cx).read(cx).client(); + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + let held = self.held_keys(&channel); let relays = self.state.relays.clone(); let pages = self.pages.clone(); @@ -498,7 +500,8 @@ impl Community { /// Write the local state document out. fn persist(&mut self, cx: &Context) { - let client = NostrRegistry::global(cx).read(cx).client(); + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); let state = self.state.clone(); let task = cx.background_spawn(async move { @@ -520,7 +523,9 @@ impl Community { limit: usize, cx: &App, ) -> Task> { - let client = NostrRegistry::global(cx).read(cx).client(); + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + let channel = *channel; let owner = self.state.owner; let community_id = self.state.id; diff --git a/crates/nip29/src/group.rs b/crates/nip29/src/group.rs index 76cb7e26..8b107da6 100644 --- a/crates/nip29/src/group.rs +++ b/crates/nip29/src/group.rs @@ -1,6 +1,6 @@ use std::collections::{HashMap, HashSet}; -use anyhow::Result; +use anyhow::{Result, anyhow, bail}; use chat::Message; use gpui::{App, AppContext, Context, EventEmitter, Task}; use nostr_sdk::prelude::*; @@ -25,6 +25,7 @@ enum Absorbed { #[derive(Debug, Clone, PartialEq, Eq)] pub enum GroupEvent { Incoming(Event), + Sent(Event), Updated, Eose, Refused(String), @@ -109,6 +110,97 @@ impl Group { self.window.previous_tag(me) } + pub fn send_message( + &mut self, + content: &str, + replies: Vec, + cx: &mut Context, + ) -> Task> { + let nostr = NostrRegistry::global(cx); + + let Some(me) = nostr.read(cx).current_user() else { + return Task::ready(Err(anyhow!("nip29: no signed-in account"))); + }; + + let previous = self.window.previous_tag(me); + let builder = self.key.id().message(content, &replies, previous); + + self.dispatch(builder, cx) + } + + pub fn send_reaction( + &mut self, + target: EventId, + emoji: String, + cx: &mut Context, + ) -> Task> { + let nostr = NostrRegistry::global(cx); + + let Some(me) = nostr.read(cx).current_user() else { + return Task::ready(Err(anyhow!("nip29: no signed-in account"))); + }; + + let previous = self.window.previous_tag(me); + let builder = self.key.id().reaction(target, &emoji, previous); + + self.dispatch(builder, cx) + } + + fn dispatch(&mut self, builder: EventBuilder, cx: &mut Context) -> Task> { + let published = self.publish(builder, cx); + + cx.spawn(async move |this, cx| { + let event = published.await?; + + this.update(cx, |_this, cx| { + cx.emit(GroupEvent::Sent(event)); + })?; + + Ok(()) + }) + } + + pub(crate) fn publish(&self, builder: EventBuilder, cx: &App) -> Task> { + let nostr = NostrRegistry::global(cx).read(cx); + let client = nostr.client(); + let signer = nostr.signer(); + let relay = self.key.relay().clone(); + + cx.background_spawn(async move { + if let Err(error) = client.add_relay(relay.clone()).and_connect().await { + log::warn!("nip29: connecting {relay} failed: {error}"); + } + + let event = builder.finalize_async(&signer).await?; + let output = client.send_event(&event).to([relay]).await?; + + if let Some(reason) = output.failed.into_values().next() { + bail!("{reason}"); + } + + Ok(event) + }) + } + + pub(crate) fn mark_pending(&mut self, since: Timestamp, cx: &mut Context) { + if self.membership != Membership::Member { + self.membership = Membership::Pending { since }; + cx.emit(GroupEvent::Updated); + cx.notify(); + } + } + + pub(crate) fn mark_refused(&mut self, reason: String, cx: &mut Context) { + if self.membership == Membership::Member { + return; + } + self.membership = Membership::Refused { + reason: Some(reason), + }; + cx.emit(GroupEvent::Updated); + cx.notify(); + } + pub(crate) fn connect(&mut self, cx: &mut Context) { let nostr = NostrRegistry::global(cx); let client = nostr.read(cx).client(); @@ -144,8 +236,8 @@ impl Group { pub(crate) fn save_envelope(&self, event: Event, cx: &App) -> Task> { let cache_tag = self.key.cache_tag(); - let nostr = NostrRegistry::global(cx).read(cx); - let client = nostr.client(); + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); cx.background_spawn(async move { let tags = vec![ @@ -167,8 +259,8 @@ impl Group { pub(crate) fn load_envelopes(&self, cx: &App) -> Task>> { let cache_tag = self.key.cache_tag(); - let nostr = NostrRegistry::global(cx).read(cx); - let client = nostr.client(); + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); cx.background_spawn(async move { let filter = Filter::new() diff --git a/crates/nip29/src/lib.rs b/crates/nip29/src/lib.rs index 4946343d..f10735fd 100644 --- a/crates/nip29/src/lib.rs +++ b/crates/nip29/src/lib.rs @@ -43,7 +43,7 @@ fn list_subscription_id() -> SubscriptionId { SubscriptionId::new(LIST_SUBSCRIPTION) } -pub(crate) fn subscription_id(key: &GroupKey) -> SubscriptionId { +fn subscription_id(key: &GroupKey) -> SubscriptionId { SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32)) } @@ -58,6 +58,7 @@ pub struct GroupsRegistry { groups: Vec>, list: GroupList, routes: HashMap, + observers: HashMap, signal_tx: flume::Sender, signal_rx: flume::Receiver, notification_listener: Option>, @@ -106,6 +107,7 @@ impl GroupsRegistry { groups: Vec::new(), list: GroupList::default(), routes: HashMap::new(), + observers: HashMap::new(), signal_tx: tx, signal_rx: rx, notification_listener: None, @@ -126,6 +128,67 @@ impl GroupsRegistry { .cloned() } + pub fn join(&mut self, reference: GroupReference, cx: &mut Context) { + let key = reference.key(); + let invite = reference.invite().map(str::to_owned); + + self.ensure(key.clone(), cx); + + let Some(group) = self.group(&key, cx) else { + return; + }; + + self.upsert_entry(&key, None); + self.publish_list(cx); + + cx.emit(GroupsEvent::Updated); + cx.notify(); + + let builder = key.id().join_request(invite.as_deref(), ""); + let sent = group.read(cx).publish(builder, cx); + + group.update(cx, |group, cx| { + group.mark_pending(Timestamp::now(), cx); + }); + + self.tasks.push(cx.spawn(async move |this, cx| { + if let Err(error) = sent.await { + let reason = error.to_string(); + + if let Err(error) = this.update(cx, |this, cx| { + if let Some(group) = this.group(&key, cx) { + group.update(cx, |group, cx| { + group.mark_refused(reason, cx); + }); + } + }) { + log::warn!("nip29: marking the refused join failed: {error}"); + } + } + })); + } + + pub fn leave(&mut self, key: &GroupKey, cx: &mut Context) { + let Some(group) = self.group(key, cx) else { + return; + }; + + let builder = key.id().leave_request(""); + let sent = group.read_with(cx, |group, cx| group.publish(builder, cx)); + let key = key.clone(); + + self.tasks.push(cx.spawn(async move |this, cx| { + if let Err(error) = sent.await { + log::warn!("nip29: leaving {} failed: {error}", key.id()); + return; + } + this.update(cx, |this, cx| { + this.forget(key, cx); + }) + .ok(); + })); + } + pub fn reset(&mut self, cx: &mut Context) { self.notification_listener = None; self.signal_consumer = None; @@ -134,9 +197,11 @@ impl GroupsRegistry { let routes = std::mem::take(&mut self.routes); self.groups.clear(); + self.observers.clear(); self.list = GroupList::default(); - let client = NostrRegistry::global(cx).read(cx).client(); + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); let mut ids: Vec = routes.into_keys().collect(); ids.push(list_subscription_id()); @@ -201,7 +266,21 @@ impl GroupsRegistry { } let group = cx.new(|_| Group::new(key.clone())); - self.groups.push(group); + self.groups.push(group.clone()); + + self.observers.insert( + key.clone(), + cx.subscribe(&group, |this, group, event, cx| match event { + GroupEvent::Sent(event) => { + let key = group.read(cx).key().clone(); + this.reload(key, Some(event.clone()), cx); + } + GroupEvent::Updated => { + this.refresh_entry(group, cx); + } + _ => {} + }), + ); self.subscribe(key.clone(), cx); self.reload(key, None, cx); @@ -219,6 +298,108 @@ impl GroupsRegistry { }); } + fn forget(&mut self, key: GroupKey, cx: &mut Context) { + let id = subscription_id(&key); + let cache_tag = key.cache_tag(); + + self.routes.remove(&id); + self.observers.remove(&key); + self.groups.retain(|group| group.read(cx).key() != &key); + self.list.remove(&key); + + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + + self.tasks.push(cx.background_spawn(async move { + if let Err(error) = client.unsubscribe(&id).await { + log::warn!("nip29: unsubscribing {id} failed: {error}"); + } + + let filter = Filter::new() + .kind(Kind::ApplicationSpecificData) + .custom_tag(SingleLetterTag::LOWERCASE_R, cache_tag); + + if let Err(error) = client.database().delete(filter).await { + log::warn!("nip29: clearing the group cache failed: {error}"); + } + })); + + self.publish_list(cx); + + cx.emit(GroupsEvent::Updated); + cx.notify(); + } + + fn entry_name(&self, key: &GroupKey) -> Option { + self.list + .entries() + .iter() + .find(|entry| entry.id == *key.id() && entry.relay == *key.relay()) + .and_then(|entry| entry.name.clone()) + } + + fn upsert_entry(&mut self, key: &GroupKey, name: Option) { + let name = name.or_else(|| self.entry_name(key)); + + self.list.upsert(GroupListEntry { + id: key.id().clone(), + relay: key.relay().clone(), + name, + }); + } + + fn refresh_entry(&mut self, group: Entity, cx: &mut Context) { + let (key, Some(name)) = ({ + let group = group.read(cx); + + ( + group.key().clone(), + group + .metadata() + .and_then(|metadata| metadata.name().map(str::to_owned)), + ) + }) else { + return; + }; + + if self.entry_name(&key).as_deref() == Some(name.as_str()) { + return; + } + + self.upsert_entry(&key, Some(name)); + self.publish_list(cx); + + cx.emit(GroupsEvent::Updated); + cx.notify(); + } + + fn publish_list(&mut self, cx: &mut Context) { + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + let signer = nostr.read(cx).signer(); + + let builder = self.list.to_builder(); + + self.tasks.push(cx.spawn(async move |_this, _cx| { + let event = match builder.finalize_async(&signer).await { + Ok(event) => event, + Err(error) => { + log::warn!("nip29: signing the group list failed: {error}"); + return; + } + }; + + match client.send_event(&event).to_nip65().await { + Ok(output) if output.failed.is_empty() => {} + Ok(output) => log::warn!( + "nip29: {} relay(s) rejected the group list", + output.failed.len() + ), + Err(error) => log::warn!("nip29: publishing the group list failed: {error}"), + } + })); + } + fn reload(&mut self, key: GroupKey, event: Option, cx: &mut Context) { let nostr = NostrRegistry::global(cx);