From c6f7956dffe5a83d4698bcf966e187ea0a58e964 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Wed, 30 Sep 2026 09:23:57 +0700 Subject: [PATCH] add groups registry --- Cargo.lock | 8 + crates/nip29/Cargo.toml | 8 + crates/nip29/src/group.rs | 323 ++++++++++++++++++++++++++++++++ crates/nip29/src/lib.rs | 377 ++++++++++++++++++++++++++++++++++++++ desktop/Cargo.toml | 1 + desktop/src/main.rs | 3 + web/Cargo.toml | 1 + web/src/lib.rs | 3 + 8 files changed, 724 insertions(+) create mode 100644 crates/nip29/src/group.rs diff --git a/Cargo.lock b/Cargo.lock index 8e255ccc..f1e2cb2c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1458,6 +1458,7 @@ dependencies = [ "gpui-pre-reqwest-client", "gpui-pre-windows", "log", + "nip29", "nostr-sdk", "person", "settings", @@ -1487,6 +1488,7 @@ dependencies = [ "gpui-pre-web", "instant", "log", + "nip29", "person", "settings", "state", @@ -4598,7 +4600,13 @@ name = "nip29" version = "1.0.2" dependencies = [ "anyhow", + "chat", + "flume 0.11.1", + "gpui-pre", + "log", "nostr-sdk", + "smallvec", + "state", ] [[package]] diff --git a/crates/nip29/Cargo.toml b/crates/nip29/Cargo.toml index 8d299d56..b69b7874 100644 --- a/crates/nip29/Cargo.toml +++ b/crates/nip29/Cargo.toml @@ -7,3 +7,11 @@ publish.workspace = true [dependencies] nostr-sdk.workspace = true anyhow.workspace = true + +state = { path = "../state" } +chat = { path = "../chat" } + +gpui.workspace = true +flume.workspace = true +log.workspace = true +smallvec.workspace = true diff --git a/crates/nip29/src/group.rs b/crates/nip29/src/group.rs new file mode 100644 index 00000000..76cb7e26 --- /dev/null +++ b/crates/nip29/src/group.rs @@ -0,0 +1,323 @@ +use std::collections::{HashMap, HashSet}; + +use anyhow::Result; +use chat::Message; +use gpui::{App, AppContext, Context, EventEmitter, Task}; +use nostr_sdk::prelude::*; +use smallvec::{SmallVec, smallvec}; +use state::NostrRegistry; + +use crate::LOCAL_KEYS; +use crate::protocol::{ + GroupAdmins, GroupKey, GroupMembers, GroupMetadata, GroupRoles, MEMBERSHIP_KINDS, Membership, + RENDER_KINDS, STATE_KINDS, TimelineWindow, +}; + +const REPLAY_LIMIT: usize = 100; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Absorbed { + Ignored, + Rendered { fresh: bool }, + Changed, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum GroupEvent { + Incoming(Event), + Updated, + Eose, + Refused(String), +} + +pub struct Group { + key: GroupKey, + metadata: Option, + members: GroupMembers, + admins: GroupAdmins, + roles: GroupRoles, + membership: Membership, + membership_events: Vec, + state_ids: HashMap, + messages: Vec, + index: HashSet, + window: TimelineWindow, + loading: bool, + refused: Option, + tasks: SmallVec<[Task<()>; 2]>, +} + +impl EventEmitter for Group {} + +impl Group { + pub(crate) fn new(key: GroupKey) -> Self { + Self { + key, + metadata: None, + members: GroupMembers::default(), + admins: GroupAdmins::default(), + roles: GroupRoles::default(), + membership: Membership::Unknown, + membership_events: Vec::new(), + state_ids: HashMap::new(), + messages: Vec::new(), + index: HashSet::new(), + window: TimelineWindow::new(), + loading: true, + refused: None, + tasks: smallvec![], + } + } + + pub fn key(&self) -> &GroupKey { + &self.key + } + + pub fn metadata(&self) -> Option<&GroupMetadata> { + self.metadata.as_ref() + } + + pub fn members(&self) -> &GroupMembers { + &self.members + } + + pub fn admins(&self) -> &GroupAdmins { + &self.admins + } + + pub fn roles(&self) -> &GroupRoles { + &self.roles + } + + pub fn membership(&self) -> &Membership { + &self.membership + } + + pub fn messages(&self) -> &[Message] { + &self.messages + } + + pub fn loading(&self) -> bool { + self.loading + } + + pub fn refused(&self) -> Option<&str> { + self.refused.as_deref() + } + + pub fn previous_tag(&self, me: PublicKey) -> Option { + self.window.previous_tag(me) + } + + pub(crate) fn connect(&mut self, cx: &mut Context) { + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + + let Some(me) = nostr.read(cx).current_user() else { + return; + }; + + let key = self.key.clone(); + let id = crate::subscription_id(&key); + let relay = key.relay().clone(); + + self.tasks.push(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 filters = vec![ + key.id().timeline_filter(REPLAY_LIMIT), + key.id().state_filter(), + key.id().membership_filter(me), + ]; + + if let Err(error) = client + .subscribe(ReqTarget::single(relay, filters)) + .with_id(id) + .await + { + log::warn!("nip29: subscribing to {} failed: {error}", key.id()); + } + })); + } + + 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(); + + cx.background_spawn(async move { + let tags = vec![ + Tag::identifier(event.id), + Tag::custom("r", [cache_tag]), + Tag::custom("k", [event.kind.to_string()]), + ]; + + let envelope = EventBuilder::new(Kind::ApplicationSpecificData, event.as_json()) + .tags(tags) + .finalize_async(&*LOCAL_KEYS) + .await?; + + client.database().save_event(&envelope).await?; + + Ok(()) + }) + } + + 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(); + + cx.background_spawn(async move { + let filter = Filter::new() + .kind(Kind::ApplicationSpecificData) + .custom_tag(SingleLetterTag::LOWERCASE_R, cache_tag); + + let mut events: Vec = client + .database() + .query(filter) + .await? + .into_iter() + .filter_map(|envelope| Event::from_json(&envelope.content).ok()) + .collect(); + + events.sort_by_key(|event| event.created_at); + + Ok(events) + }) + } + + pub(crate) fn hydrate(&mut self, events: Vec, me: PublicKey, cx: &mut Context) { + if events.is_empty() { + return; + } + + let mut rendered = Vec::new(); + let mut changed = false; + + for event in &events { + match self.absorb(event, me) { + Absorbed::Rendered { fresh: true } => rendered.push(event.clone()), + Absorbed::Changed => changed = true, + _ => {} + } + } + + for event in rendered { + cx.emit(GroupEvent::Incoming(event)); + } + + if changed { + cx.emit(GroupEvent::Updated); + } + + cx.notify(); + } + + pub(crate) fn eose(&mut self, cx: &mut Context) { + self.loading = false; + cx.emit(GroupEvent::Eose); + cx.notify(); + } + + pub(crate) fn refuse(&mut self, reason: String, cx: &mut Context) { + self.loading = false; + self.refused = Some(reason.clone()); + cx.emit(GroupEvent::Refused(reason)); + cx.notify(); + } + + fn absorb(&mut self, event: &Event, me: PublicKey) -> Absorbed { + if STATE_KINDS.contains(&event.kind) { + return if self.absorb_state(event) { + Absorbed::Changed + } else { + Absorbed::Ignored + }; + } + + if MEMBERSHIP_KINDS.contains(&event.kind) { + self.window.push(event.clone()); + + if !event.tags.public_keys().any(|key| key == me) { + return Absorbed::Ignored; + } + + if !self + .membership_events + .iter() + .any(|existing| existing.id == event.id) + { + self.membership_events.push(event.clone()); + } + + return if self.refresh_membership(me) { + Absorbed::Changed + } else { + Absorbed::Ignored + }; + } + + self.window.push(event.clone()); + + if RENDER_KINDS.contains(&event.kind) { + Absorbed::Rendered { + fresh: self.push_message(event), + } + } else { + Absorbed::Ignored + } + } + + fn absorb_state(&mut self, event: &Event) -> bool { + match event.kind { + Kind::GroupMetadata => match GroupMetadata::parse(event) { + Ok(metadata) => self.metadata = Some(metadata), + Err(_) => return false, + }, + Kind::GroupAdmins => match GroupAdmins::parse(event) { + Ok(admins) => self.admins = admins, + Err(_) => return false, + }, + Kind::GroupMembers => match GroupMembers::parse(event) { + Ok(members) => self.members = members, + Err(_) => return false, + }, + Kind::GroupRoles => match GroupRoles::parse(event) { + Ok(roles) => self.roles = roles, + Err(_) => return false, + }, + _ => return false, + } + + self.state_ids.insert(event.kind, event.id) != Some(event.id) + } + + fn push_message(&mut self, event: &Event) -> bool { + if !self.index.insert(event.id) { + return false; + } + + let message = Message::from(event); + let position = self + .messages + .partition_point(|existing| existing.created_at <= message.created_at); + + self.messages.insert(position, message); + + true + } + + fn refresh_membership(&mut self, me: PublicKey) -> bool { + let derived = Membership::from_events(&self.membership_events, me); + + if derived != Membership::Unknown && derived != self.membership { + self.membership = derived; + return true; + } + + false + } +} diff --git a/crates/nip29/src/lib.rs b/crates/nip29/src/lib.rs index 87f025a7..4946343d 100644 --- a/crates/nip29/src/lib.rs +++ b/crates/nip29/src/lib.rs @@ -1,3 +1,380 @@ +use std::collections::{HashMap, HashSet}; +use std::sync::LazyLock; + +use anyhow::{Error, Result}; +use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task}; +use nostr_sdk::prelude::*; +use smallvec::{SmallVec, smallvec}; +use state::NostrRegistry; + +mod group; mod protocol; +pub use group::*; pub use protocol::*; + +pub(crate) static LOCAL_KEYS: LazyLock = LazyLock::new(Keys::generate); + +const LIST_SUBSCRIPTION: &str = "nip29-list"; +const GROUP_PREFIX: &str = "grp-"; + +pub fn init(cx: &mut App) { + GroupsRegistry::set_global(cx.new(GroupsRegistry::new), cx); +} + +struct GlobalGroupsRegistry(Entity); + +impl Global for GlobalGroupsRegistry {} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum GroupsEvent { + Updated, + Error(String), +} + +#[derive(Debug)] +enum Signal { + Event(SubscriptionId, Event), + Eose(SubscriptionId), + Closed(SubscriptionId, String), +} + +fn list_subscription_id() -> SubscriptionId { + SubscriptionId::new(LIST_SUBSCRIPTION) +} + +pub(crate) fn subscription_id(key: &GroupKey) -> SubscriptionId { + SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32)) +} + +fn is_group_subscription(id: &SubscriptionId) -> bool { + let id = id.as_str(); + id == LIST_SUBSCRIPTION || id.starts_with(GROUP_PREFIX) +} + +impl EventEmitter for GroupsRegistry {} + +pub struct GroupsRegistry { + groups: Vec>, + list: GroupList, + routes: HashMap, + signal_tx: flume::Sender, + signal_rx: flume::Receiver, + notification_listener: Option>, + signal_consumer: Option>, + tasks: SmallVec<[Task<()>; 2]>, + _subscriptions: SmallVec<[Subscription; 2]>, +} + +impl GroupsRegistry { + pub fn global(cx: &App) -> Entity { + cx.global::().0.clone() + } + + fn set_global(state: Entity, cx: &mut App) { + cx.set_global(GlobalGroupsRegistry(state)); + } + + fn new(cx: &mut Context) -> Self { + let nostr = NostrRegistry::global(cx); + let this = cx.entity().downgrade(); + + let (tx, rx) = flume::unbounded::(); + let mut subscriptions = smallvec![]; + + subscriptions.push(cx.subscribe(&nostr, |this, _nostr, event, cx| { + if event.signer_changed() { + this.reset(cx); + this.handle_notifications(cx); + this.load(cx); + } + })); + + cx.defer(move |cx| { + if let Err(error) = this.update(cx, |this, cx| { + this.handle_notifications(cx); + + if nostr.read(cx).current_user().is_some() { + this.load(cx); + } + }) { + log::warn!("nip29: initializing the registry failed: {error}"); + } + }); + + Self { + groups: Vec::new(), + list: GroupList::default(), + routes: HashMap::new(), + signal_tx: tx, + signal_rx: rx, + notification_listener: None, + signal_consumer: None, + tasks: smallvec![], + _subscriptions: subscriptions, + } + } + + pub fn groups(&self) -> &[Entity] { + &self.groups + } + + pub fn group(&self, key: &GroupKey, cx: &App) -> Option> { + self.groups + .iter() + .find(|group| group.read(cx).key() == key) + .cloned() + } + + pub fn reset(&mut self, cx: &mut Context) { + self.notification_listener = None; + self.signal_consumer = None; + self.tasks.clear(); + + let routes = std::mem::take(&mut self.routes); + + self.groups.clear(); + self.list = GroupList::default(); + + let client = NostrRegistry::global(cx).read(cx).client(); + + let mut ids: Vec = routes.into_keys().collect(); + ids.push(list_subscription_id()); + + self.tasks.push(cx.background_spawn(async move { + for id in ids { + if let Err(error) = client.unsubscribe(&id).await { + log::warn!("nip29: unsubscribing {id} failed: {error}"); + } + } + })); + + cx.notify(); + } + + fn load(&mut self, cx: &mut Context) { + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + let signer = nostr.read(cx).signer(); + + let task: Task> = cx.background_spawn(async move { + let me = signer.get_public_key_async().await?; + let filter = Filter::new().author(me).kind(Kind::SimpleGroups).limit(1); + + client + .subscribe(ReqTarget::auto([filter.clone()])) + .with_id(list_subscription_id()) + .await?; + + match client.database().query(filter).await?.into_iter().next() { + Some(event) => GroupList::parse(&event), + None => Ok(GroupList::default()), + } + }); + + self.tasks + .push(cx.spawn(async move |this, cx| match task.await { + Ok(list) => { + if let Err(error) = this.update(cx, |this, cx| this.merge(list, cx)) { + log::warn!("nip29: applying the group list failed: {error}"); + } + } + Err(error) => log::warn!("nip29: loading the group list failed: {error}"), + })); + } + + fn merge(&mut self, list: GroupList, cx: &mut Context) { + let entries = list.entries().to_vec(); + + for entry in entries { + self.list.upsert(entry.clone()); + self.ensure(GroupKey::new(entry.relay.clone(), entry.id.clone()), cx); + } + + cx.emit(GroupsEvent::Updated); + cx.notify(); + } + + fn ensure(&mut self, key: GroupKey, cx: &mut Context) { + if self.group(&key, cx).is_some() { + return; + } + + let group = cx.new(|_| Group::new(key.clone())); + self.groups.push(group); + + self.subscribe(key.clone(), cx); + self.reload(key, None, cx); + } + + fn subscribe(&mut self, key: GroupKey, cx: &mut Context) { + let Some(group) = self.group(&key, cx) else { + return; + }; + + self.routes.insert(subscription_id(&key), key); + + group.update(cx, |group, cx| { + group.connect(cx); + }); + } + + fn reload(&mut self, key: GroupKey, event: Option, cx: &mut Context) { + let nostr = NostrRegistry::global(cx); + + let Some(group) = self.group(&key, cx) else { + return; + }; + + let Some(me) = nostr.read(cx).current_user() else { + return; + }; + + let logged = key.clone(); + let load = group.read(cx).load_envelopes(cx); + + self.tasks.push(cx.spawn(async move |_this, cx| { + if let Some(event) = event + && let Err(error) = group + .read_with(cx, |group, cx| group.save_envelope(event, cx)) + .await + { + log::warn!("nip29: caching failed: {error}"); + return; + } + + match load.await { + Ok(events) => { + group.update(cx, |group, cx| { + group.hydrate(events, me, cx); + }); + } + Err(error) => { + log::warn!("nip29: reloading {} failed: {error}", logged.id()); + } + } + })); + } + + fn handle_event(&mut self, id: SubscriptionId, event: Event, cx: &mut Context) { + if id == list_subscription_id() { + match GroupList::parse(&event) { + Ok(list) => self.merge(list, cx), + Err(error) => log::warn!("nip29: ignoring a malformed group list: {error}"), + } + return; + } + + let Some(key) = self.routes.get(&id).cloned() else { + return; + }; + + self.reload(key, Some(event), cx); + } + + fn handle_eose(&mut self, id: SubscriptionId, cx: &mut Context) { + let Some(key) = self.routes.get(&id).cloned() else { + return; + }; + let Some(group) = self.group(&key, cx) else { + return; + }; + + group.update(cx, |group, cx| { + group.eose(cx); + }); + cx.notify(); + } + + fn handle_closed(&mut self, id: SubscriptionId, reason: String, cx: &mut Context) { + let Some(key) = self.routes.get(&id).cloned() else { + return; + }; + let Some(group) = self.group(&key, cx) else { + return; + }; + + group.update(cx, |group, cx| { + group.refuse(reason, cx); + }); + cx.notify(); + } + + fn handle_notifications(&mut self, cx: &mut Context) { + self.notification_listener = None; + self.signal_consumer = None; + + let nostr = NostrRegistry::global(cx).read(cx); + let client = nostr.client(); + + let tx = self.signal_tx.clone(); + let rx = self.signal_rx.clone(); + + self.notification_listener = Some(cx.background_spawn(async move { + let mut notifications = client.notifications(); + let mut processed = HashSet::new(); + const MAX_PROCESSED: usize = 10_000; + + while let Some(notification) = notifications.next().await { + let ClientNotification::Message { message, .. } = notification else { + continue; + }; + + let signal = match *message { + RelayMessage::Event { + subscription_id, + event, + } => { + let id = subscription_id.into_owned(); + + if !is_group_subscription(&id) { + None + } else { + if processed.len() >= MAX_PROCESSED { + processed.clear(); + } + + processed + .insert(event.id) + .then(|| Signal::Event(id, event.into_owned())) + } + } + RelayMessage::EndOfStoredEvents(id) => { + let id = id.into_owned(); + is_group_subscription(&id).then_some(Signal::Eose(id)) + } + RelayMessage::Closed { + subscription_id, + message, + } => { + let id = subscription_id.into_owned(); + is_group_subscription(&id) + .then_some(Signal::Closed(id, message.into_owned())) + } + _ => None, + }; + + if let Some(signal) = signal { + tx.send_async(signal).await.ok(); + } + } + })); + + self.signal_consumer = Some(cx.spawn(async move |this, cx| { + while let Ok(signal) = rx.recv_async().await { + this.update(cx, |this, cx| match signal { + Signal::Event(id, event) => { + this.handle_event(id, event, cx); + } + Signal::Eose(id) => { + this.handle_eose(id, cx); + } + Signal::Closed(id, reason) => { + this.handle_closed(id, reason, cx); + } + }) + .ok(); + } + })); + } +} diff --git a/desktop/Cargo.toml b/desktop/Cargo.toml index 82568676..79e3baed 100644 --- a/desktop/Cargo.toml +++ b/desktop/Cargo.toml @@ -36,6 +36,7 @@ state = { path = "../crates/state" } device = { path = "../crates/device" } chat = { path = "../crates/chat" } community = { path = "../crates/community" } +nip29 = { path = "../crates/nip29" } settings = { path = "../crates/settings" } auto_update = { path = "../crates/auto_update" } person = { path = "../crates/person" } diff --git a/desktop/src/main.rs b/desktop/src/main.rs index a275dc70..43f6d829 100644 --- a/desktop/src/main.rs +++ b/desktop/src/main.rs @@ -81,6 +81,9 @@ fn main() { // Initialize community registry community::init(cx); + // Initialize groups registry + nip29::init(cx); + // Initialize auto update auto_update::init(cx); diff --git a/web/Cargo.toml b/web/Cargo.toml index 0f172056..10055154 100644 --- a/web/Cargo.toml +++ b/web/Cargo.toml @@ -17,6 +17,7 @@ state = { path = "../crates/state" } device = { path = "../crates/device" } chat = { path = "../crates/chat" } community = { path = "../crates/community" } +nip29 = { path = "../crates/nip29" } settings = { path = "../crates/settings" } person = { path = "../crates/person" } diff --git a/web/src/lib.rs b/web/src/lib.rs index d8130655..52774800 100644 --- a/web/src/lib.rs +++ b/web/src/lib.rs @@ -73,6 +73,9 @@ pub fn run() -> Result<(), JsValue> { // Initialize community registry community::init(cx); + // Initialize groups registry + nip29::init(cx); + cx.open_window(WindowOptions::default(), |window, cx| { cx.new(|cx| Root::new(workspace::init(window, cx).into(), window, cx)) })