add NIP-29 moderation, pins, browse, and create-group support

crates/community/src/community.rs

@@ -349,7 +349,9 @@ impl Community {
     }
 
     fn start_round(&mut self, channel: ChannelId, intent: Intent, cx: &mut Context<Self>) {
-        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<Self>) {
-        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<Result<Timeline>> {
-        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;

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::*;
@@ -9,8 +9,8 @@ use state::NostrRegistry;
 
 use crate::LOCAL_KEYS;
 use crate::protocol::{
-    GroupAdmins, GroupKey, GroupMembers, GroupMetadata, GroupRoles, MEMBERSHIP_KINDS, Membership,
+    GroupAdmins, GroupId, GroupKey, GroupMembers, GroupMetadata, GroupPins, GroupRoles,
-    RENDER_KINDS, STATE_KINDS, TimelineWindow,
+    MEMBERSHIP_KINDS, Membership, PIN_LIST, Pin, RENDER_KINDS, STATE_KINDS, TimelineWindow,
 };
 
 const REPLAY_LIMIT: usize = 100;
@@ -25,6 +25,7 @@ enum Absorbed {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum GroupEvent {
     Incoming(Event),
+    Sent(Event),
     Updated,
     Eose,
     Refused(String),
@@ -36,6 +37,7 @@ pub struct Group {
     members: GroupMembers,
     admins: GroupAdmins,
     roles: GroupRoles,
+    pins: GroupPins,
     membership: Membership,
     membership_events: Vec<Event>,
     state_ids: HashMap<Kind, EventId>,
@@ -57,6 +59,7 @@ impl Group {
             members: GroupMembers::default(),
             admins: GroupAdmins::default(),
             roles: GroupRoles::default(),
+            pins: GroupPins::default(),
             membership: Membership::Unknown,
             membership_events: Vec::new(),
             state_ids: HashMap::new(),
@@ -89,6 +92,10 @@ impl Group {
         &self.roles
     }
 
+    pub fn pins(&self) -> &GroupPins {
+        &self.pins
+    }
+
     pub fn membership(&self) -> &Membership {
         &self.membership
     }
@@ -109,6 +116,126 @@ impl Group {
         self.window.previous_tag(me)
     }
 
+    pub fn send_message(
+        &mut self,
+        content: &str,
+        replies: Vec<EventId>,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<()>> {
+        self.dispatch(move |id, prev| id.message(content, &replies, prev), cx)
+    }
+
+    pub fn send_reaction(
+        &mut self,
+        target: EventId,
+        emoji: String,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<()>> {
+        self.dispatch(move |id, prev| id.reaction(target, &emoji, prev), cx)
+    }
+
+    pub fn put_user(
+        &mut self,
+        public_key: PublicKey,
+        roles: Vec<String>,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<()>> {
+        self.dispatch(move |id, prev| id.put_user(public_key, &roles, prev), cx)
+    }
+
+    pub fn remove_user(
+        &mut self,
+        public_key: PublicKey,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<()>> {
+        self.dispatch(move |id, prev| id.remove_user(public_key, prev), cx)
+    }
+
+    pub fn edit_metadata(
+        &mut self,
+        metadata: &GroupMetadata,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<()>> {
+        self.dispatch(move |id, prev| id.edit_metadata(metadata, prev), cx)
+    }
+
+    pub fn delete_event(&mut self, id: EventId, cx: &mut Context<Self>) -> Task<Result<()>> {
+        self.dispatch(move |gid, prev| gid.delete_event(id, prev), cx)
+    }
+
+    pub fn create_invite(&mut self, code: &str, cx: &mut Context<Self>) -> Task<Result<()>> {
+        self.dispatch(move |id, prev| id.create_invite(code, prev), cx)
+    }
+
+    pub fn update_pins(&mut self, pins: Vec<Pin>, cx: &mut Context<Self>) -> Task<Result<()>> {
+        self.dispatch(move |id, prev| id.update_pin_list(&pins, prev), cx)
+    }
+
+    fn dispatch<F>(&mut self, build: F, cx: &mut Context<Self>) -> Task<Result<()>>
+    where
+        F: FnOnce(&GroupId, Option<Tag>) -> EventBuilder,
+    {
+        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 builder = build(self.key.id(), self.window.previous_tag(me));
+        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<Result<Event>> {
+        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<Self>) {
+        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<Self>) {
+        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<Self>) {
         let nostr = NostrRegistry::global(cx);
         let client = nostr.read(cx).client();
@@ -144,8 +271,8 @@ impl Group {
 
     pub(crate) fn save_envelope(&self, event: Event, cx: &App) -> Task<Result<()>> {
         let cache_tag = self.key.cache_tag();
-        let nostr = NostrRegistry::global(cx).read(cx);
+        let nostr = NostrRegistry::global(cx);
-        let client = nostr.client();
+        let client = nostr.read(cx).client();
 
         cx.background_spawn(async move {
             let tags = vec![
@@ -167,8 +294,8 @@ impl Group {
 
     pub(crate) fn load_envelopes(&self, cx: &App) -> Task<Result<Vec<Event>>> {
         let cache_tag = self.key.cache_tag();
-        let nostr = NostrRegistry::global(cx).read(cx);
+        let nostr = NostrRegistry::global(cx);
-        let client = nostr.client();
+        let client = nostr.read(cx).client();
 
         cx.background_spawn(async move {
             let filter = Filter::new()
@@ -289,6 +416,10 @@ impl Group {
                 Ok(roles) => self.roles = roles,
                 Err(_) => return false,
             },
+            Kind::Custom(PIN_LIST) => match GroupPins::parse(event) {
+                Ok(pins) => self.pins = pins,
+                Err(_) => return false,
+            },
             _ => return false,
         }
 

crates/nip29/src/lib.rs

@@ -1,5 +1,8 @@
+use std::collections::hash_map::DefaultHasher;
 use std::collections::{HashMap, HashSet};
+use std::hash::{Hash, Hasher};
 use std::sync::LazyLock;
+use std::time::Duration;
 
 use anyhow::{Error, Result};
 use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task};
@@ -17,6 +20,9 @@ pub(crate) static LOCAL_KEYS: LazyLock<Keys> = LazyLock::new(Keys::generate);
 
 const LIST_SUBSCRIPTION: &str = "nip29-list";
 const GROUP_PREFIX: &str = "grp-";
+const BROWSE_PREFIX: &str = "brw-";
+const BROWSE_LIMIT: usize = 500;
+const BROWSE_TIMEOUT: Duration = Duration::from_secs(10);
 
 pub fn init(cx: &mut App) {
     GroupsRegistry::set_global(cx.new(GroupsRegistry::new), cx);
@@ -32,6 +38,13 @@ pub enum GroupsEvent {
     Error(String),
 }
 
+#[derive(Debug, Clone)]
+pub struct Browse {
+    pub relay: RelayUrl,
+    pub loading: bool,
+    pub groups: Vec<GroupCandidate>,
+}
+
 #[derive(Debug)]
 enum Signal {
     Event(SubscriptionId, Event),
@@ -43,13 +56,20 @@ 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))
 }
 
-fn is_group_subscription(id: &SubscriptionId) -> bool {
+fn browse_subscription(relay: &RelayUrl) -> SubscriptionId {
+    let mut hasher = DefaultHasher::new();
+    relay.hash(&mut hasher);
+
+    SubscriptionId::new(format!("{BROWSE_PREFIX}{:08x}", hasher.finish() as u32))
+}
+
+fn is_own_subscription(id: &SubscriptionId) -> bool {
     let id = id.as_str();
-    id == LIST_SUBSCRIPTION || id.starts_with(GROUP_PREFIX)
+    id == LIST_SUBSCRIPTION || id.starts_with(GROUP_PREFIX) || id.starts_with(BROWSE_PREFIX)
 }
 
 impl EventEmitter<GroupsEvent> for GroupsRegistry {}
@@ -57,7 +77,9 @@ impl EventEmitter<GroupsEvent> for GroupsRegistry {}
 pub struct GroupsRegistry {
     groups: Vec<Entity<Group>>,
     list: GroupList,
+    browse: HashMap<SubscriptionId, Browse>,
     routes: HashMap<SubscriptionId, GroupKey>,
+    observers: HashMap<GroupKey, Subscription>,
     signal_tx: flume::Sender<Signal>,
     signal_rx: flume::Receiver<Signal>,
     notification_listener: Option<Task<()>>,
@@ -105,7 +127,9 @@ impl GroupsRegistry {
         Self {
             groups: Vec::new(),
             list: GroupList::default(),
+            browse: HashMap::new(),
             routes: HashMap::new(),
+            observers: HashMap::new(),
             signal_tx: tx,
             signal_rx: rx,
             notification_listener: None,
@@ -126,6 +150,136 @@ impl GroupsRegistry {
             .cloned()
     }
 
+    pub fn join(&mut self, reference: GroupReference, cx: &mut Context<Self>) {
+        self.join_key(reference.key(), reference.invite().map(str::to_owned), cx);
+    }
+
+    pub fn join_key(&mut self, key: GroupKey, invite: Option<String>, cx: &mut Context<Self>) {
+        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 create(&mut self, key: GroupKey, metadata: GroupMetadata, cx: &mut Context<Self>) {
+        self.ensure(key.clone(), cx);
+
+        let Some(group) = self.group(&key, cx) else {
+            return;
+        };
+
+        self.upsert_entry(&key, metadata.name().map(str::to_owned));
+        self.publish_list(cx);
+
+        cx.emit(GroupsEvent::Updated);
+        cx.notify();
+
+        let builder = key.id().create_group(&metadata, None);
+        let published = group.read(cx).publish(builder, cx);
+        let logged = key.clone();
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            if let Err(e) = published.await {
+                this.update(cx, |this, cx| {
+                    this.forget(logged, cx);
+                    cx.emit(GroupsEvent::Error(e.to_string()));
+                })
+                .ok();
+            }
+        }));
+    }
+
+    pub fn browse(&mut self, relay: RelayUrl, cx: &mut Context<Self>) {
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+        let id = browse_subscription(&relay);
+
+        self.browse.insert(
+            id.clone(),
+            Browse {
+                relay: relay.clone(),
+                loading: true,
+                groups: Vec::new(),
+            },
+        );
+
+        cx.emit(GroupsEvent::Updated);
+        cx.notify();
+
+        self.tasks.push(cx.background_spawn(async move {
+            client.add_relay(relay.clone()).and_connect().await.ok();
+
+            let filter = Filter::new().kind(Kind::GroupMetadata).limit(BROWSE_LIMIT);
+            let opts = SubscribeAutoCloseOptions::default()
+                .exit_policy(ReqExitPolicy::ExitOnEOSE)
+                .timeout(Some(BROWSE_TIMEOUT));
+
+            if let Err(error) = client
+                .subscribe(ReqTarget::single(relay, [filter]))
+                .with_id(id)
+                .close_on(opts)
+                .await
+            {
+                log::warn!("nip29: browsing failed: {error}");
+            }
+        }));
+    }
+
+    pub fn browsed(&self, relay: &RelayUrl) -> Option<&Browse> {
+        self.browse.get(&browse_subscription(relay))
+    }
+
+    pub fn leave(&mut self, key: &GroupKey, cx: &mut Context<Self>) {
+        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>) {
         self.notification_listener = None;
         self.signal_consumer = None;
@@ -134,12 +288,17 @@ 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<SubscriptionId> = routes.into_keys().collect();
         ids.push(list_subscription_id());
+        ids.extend(self.browse.keys().cloned());
+
+        self.browse.clear();
 
         self.tasks.push(cx.background_spawn(async move {
             for id in ids {
@@ -201,7 +360,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 +392,108 @@ impl GroupsRegistry {
         });
     }
 
+    fn forget(&mut self, key: GroupKey, cx: &mut Context<Self>) {
+        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<String> {
+        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<String>) {
+        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<Group>, cx: &mut Context<Self>) {
+        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<Self>) {
+        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<Event>, cx: &mut Context<Self>) {
         let nostr = NostrRegistry::global(cx);
 
@@ -265,6 +540,11 @@ impl GroupsRegistry {
             return;
         }
 
+        if self.browse.contains_key(&id) {
+            self.absorb_candidate(&id, &event, cx);
+            return;
+        }
+
         let Some(key) = self.routes.get(&id).cloned() else {
             return;
         };
@@ -272,10 +552,49 @@ impl GroupsRegistry {
         self.reload(key, Some(event), cx);
     }
 
+    fn absorb_candidate(&mut self, id: &SubscriptionId, event: &Event, cx: &mut Context<Self>) {
+        let Some(browse) = self.browse.get_mut(id) else {
+            return;
+        };
+
+        let Ok(metadata) = GroupMetadata::parse(event) else {
+            log::warn!(
+                "nip29: ignoring malformed group metadata from {}",
+                browse.relay
+            );
+            return;
+        };
+
+        let Some(group) = metadata.id().cloned() else {
+            return;
+        };
+
+        let candidate = GroupCandidate {
+            key: GroupKey::new(browse.relay.clone(), group),
+            metadata,
+        };
+
+        match browse.groups.iter_mut().find(|e| e.key == candidate.key) {
+            Some(existing) => *existing = candidate,
+            None => browse.groups.push(candidate),
+        }
+
+        cx.emit(GroupsEvent::Updated);
+        cx.notify();
+    }
+
     fn handle_eose(&mut self, id: SubscriptionId, cx: &mut Context<Self>) {
+        if let Some(browse) = self.browse.get_mut(&id) {
+            browse.loading = false;
+            cx.emit(GroupsEvent::Updated);
+            cx.notify();
+            return;
+        }
+
         let Some(key) = self.routes.get(&id).cloned() else {
             return;
         };
+
         let Some(group) = self.group(&key, cx) else {
             return;
         };
@@ -283,13 +602,24 @@ impl GroupsRegistry {
         group.update(cx, |group, cx| {
             group.eose(cx);
         });
+
         cx.notify();
     }
 
     fn handle_closed(&mut self, id: SubscriptionId, reason: String, cx: &mut Context<Self>) {
+        if let Some(browse) = self.browse.get_mut(&id) {
+            log::warn!("nip29: browsing {} was refused: {reason}", browse.relay);
+            browse.loading = false;
+
+            cx.emit(GroupsEvent::Updated);
+            cx.notify();
+            return;
+        }
+
         let Some(key) = self.routes.get(&id).cloned() else {
             return;
         };
+
         let Some(group) = self.group(&key, cx) else {
             return;
         };
@@ -297,6 +627,7 @@ impl GroupsRegistry {
         group.update(cx, |group, cx| {
             group.refuse(reason, cx);
         });
+
         cx.notify();
     }
 
@@ -327,13 +658,12 @@ impl GroupsRegistry {
                     } => {
                         let id = subscription_id.into_owned();
 
-                        if !is_group_subscription(&id) {
+                        if !is_own_subscription(&id) {
                             None
                         } else {
                             if processed.len() >= MAX_PROCESSED {
                                 processed.clear();
                             }
-
                             processed
                                 .insert(event.id)
                                 .then(|| Signal::Event(id, event.into_owned()))
@@ -341,15 +671,14 @@ impl GroupsRegistry {
                     }
                     RelayMessage::EndOfStoredEvents(id) => {
                         let id = id.into_owned();
-                        is_group_subscription(&id).then_some(Signal::Eose(id))
+                        is_own_subscription(&id).then_some(Signal::Eose(id))
                     }
                     RelayMessage::Closed {
                         subscription_id,
                         message,
                     } => {
                         let id = subscription_id.into_owned();
-                        is_group_subscription(&id)
+                        is_own_subscription(&id).then_some(Signal::Closed(id, message.into_owned()))
-                            .then_some(Signal::Closed(id, message.into_owned()))
                     }
                     _ => None,
                 };

crates/nip29/src/protocol.rs

@@ -19,11 +19,15 @@ pub const RENDER_KINDS: [Kind; 5] = [
     Kind::Reaction,
 ];
 
-pub const STATE_KINDS: [Kind; 4] = [
+pub const PIN_LIST: u16 = 39_005;
+pub const UPDATE_PIN_LIST: u16 = 9_010;
+
+pub const STATE_KINDS: [Kind; 5] = [
     Kind::GroupMetadata,
     Kind::GroupAdmins,
     Kind::GroupMembers,
     Kind::GroupRoles,
+    Kind::Custom(PIN_LIST),
 ];
 
 pub const MEMBERSHIP_KINDS: [Kind; 2] = [Kind::GroupPutUser, Kind::GroupRemoveUser];
@@ -106,6 +110,106 @@ impl GroupId {
     pub fn leave_request(&self, reason: &str) -> EventBuilder {
         EventBuilder::new(Kind::GroupLeaveRequest, reason).tag(self.h_tag())
     }
+
+    pub fn put_user(
+        &self,
+        public_key: PublicKey,
+        roles: &[String],
+        previous: Option<Tag>,
+    ) -> EventBuilder {
+        let mut values = Vec::with_capacity(1 + roles.len());
+        values.push(public_key.to_hex());
+        values.extend(roles.iter().cloned());
+
+        EventBuilder::new(Kind::GroupPutUser, "")
+            .tag(self.h_tag())
+            .tag(Tag::custom("p", values))
+            .tag_maybe(previous)
+    }
+
+    pub fn remove_user(&self, public_key: PublicKey, previous: Option<Tag>) -> EventBuilder {
+        EventBuilder::new(Kind::GroupRemoveUser, "")
+            .tag(self.h_tag())
+            .tag(Tag::public_key(public_key))
+            .tag_maybe(previous)
+    }
+
+    pub fn edit_metadata(&self, metadata: &GroupMetadata, previous: Option<Tag>) -> EventBuilder {
+        EventBuilder::new(Kind::GroupEditMetadata, "")
+            .tags(self.metadata_tags(metadata))
+            .tag_maybe(previous)
+    }
+
+    pub fn create_group(&self, metadata: &GroupMetadata, previous: Option<Tag>) -> EventBuilder {
+        EventBuilder::new(Kind::GroupCreateGroup, "")
+            .tags(self.metadata_tags(metadata))
+            .tag_maybe(previous)
+    }
+
+    fn metadata_tags(&self, metadata: &GroupMetadata) -> Vec<Tag> {
+        let mut tags = vec![self.h_tag()];
+
+        for (name, value) in [
+            ("name", &metadata.name),
+            ("picture", &metadata.picture),
+            ("banner", &metadata.banner),
+            ("about", &metadata.about),
+        ] {
+            if let Some(value) = value {
+                tags.push(Tag::custom(name, [value.clone()]));
+            }
+        }
+
+        for (name, set) in [
+            ("private", metadata.private),
+            ("restricted", metadata.restricted),
+            ("hidden", metadata.hidden),
+            ("closed", metadata.closed),
+            ("livekit", metadata.livekit),
+        ] {
+            if set {
+                tags.push(Tag::custom(name, Vec::<String>::new()));
+            }
+        }
+
+        if let Some(parent) = &metadata.parent {
+            tags.push(Tag::custom("parent", [parent.as_str()]));
+        }
+
+        for child in &metadata.children {
+            tags.push(Tag::custom("child", [child.as_str()]));
+        }
+
+        if let Some(kinds) = &metadata.supported_kinds {
+            tags.push(Tag::custom(
+                "supported_kinds",
+                kinds.iter().map(ToString::to_string),
+            ));
+        }
+
+        tags
+    }
+
+    pub fn delete_event(&self, id: EventId, previous: Option<Tag>) -> EventBuilder {
+        EventBuilder::new(Kind::GroupDeleteEvent, "")
+            .tag(self.h_tag())
+            .tag(Tag::event(id))
+            .tag_maybe(previous)
+    }
+
+    pub fn create_invite(&self, code: &str, previous: Option<Tag>) -> EventBuilder {
+        EventBuilder::new(Kind::GroupCreateInvite, "")
+            .tag(self.h_tag())
+            .tag(Tag::custom("code", [code]))
+            .tag_maybe(previous)
+    }
+
+    pub fn update_pin_list(&self, pins: &[Pin], previous: Option<Tag>) -> EventBuilder {
+        EventBuilder::new(Kind::Custom(UPDATE_PIN_LIST), "")
+            .tag(self.h_tag())
+            .tags(pins.iter().map(Pin::tag))
+            .tag_maybe(previous)
+    }
 }
 
 impl fmt::Display for GroupId {
@@ -210,7 +314,7 @@ impl FromStr for GroupReference {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
 pub struct GroupMetadata {
     pub name: Option<String>,
     pub picture: Option<String>,
@@ -222,9 +326,10 @@ pub struct GroupMetadata {
     pub closed: bool,
     pub livekit: bool,
     pub parent: Option<GroupId>,
+    pub children: Vec<GroupId>,
     pub supported_kinds: Option<Vec<Kind>>,
-    id: GroupId,
+    id: Option<GroupId>,
-    author: PublicKey,
+    author: Option<PublicKey>,
 }
 
 impl GroupMetadata {
@@ -257,6 +362,14 @@ impl GroupMetadata {
             None => None,
         };
 
+        let mut children = Vec::new();
+
+        for tag in event.tags.iter().filter(|tag| tag.kind() == "child") {
+            if let Some(child) = tag.content() {
+                children.push(GroupId::new(child)?);
+            }
+        }
+
         Ok(Self {
             name: tag_value(&event.tags, "name").map(str::to_owned),
             picture: tag_value(&event.tags, "picture").map(str::to_owned),
@@ -268,12 +381,17 @@ impl GroupMetadata {
             closed: has_tag(&event.tags, "closed"),
             livekit: has_tag(&event.tags, "livekit"),
             parent,
+            children,
             supported_kinds,
-            id: GroupId::new(identifier)?,
+            id: Some(GroupId::new(identifier)?),
-            author: event.pubkey,
+            author: Some(event.pubkey),
         })
     }
 
+    pub(crate) fn id(&self) -> Option<&GroupId> {
+        self.id.as_ref()
+    }
+
     pub fn name(&self) -> Option<&str> {
         self.name.as_deref()
     }
@@ -286,8 +404,10 @@ impl GroupMetadata {
     }
 
     pub fn naddr(&self, relay: &RelayUrl, invite: Option<&str>) -> Result<String> {
-        let coordinate =
+        let id = self.id.as_ref().ok_or_else(|| anyhow!("no id"))?;
-            Coordinate::new(Kind::GroupMetadata, self.author).identifier(self.id.as_str());
+        let author = self.author.ok_or_else(|| anyhow!("no author"))?;
+
+        let coordinate = Coordinate::new(Kind::GroupMetadata, author).identifier(id.as_str());
         let address = Nip19Coordinate::new(coordinate, [relay.clone()]).to_bech32()?;
 
         Ok(match invite {
@@ -297,6 +417,12 @@ impl GroupMetadata {
     }
 }
 
+#[derive(Debug, Clone)]
+pub struct GroupCandidate {
+    pub key: GroupKey,
+    pub metadata: GroupMetadata,
+}
+
 #[derive(Debug, Clone, Default)]
 pub struct GroupAdmins(Vec<(PublicKey, Vec<String>)>);
 
@@ -389,6 +515,60 @@ impl GroupRoles {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Pin {
+    Event(EventId),
+    Address(Coordinate),
+}
+
+impl Pin {
+    fn tag(&self) -> Tag {
+        match self {
+            Pin::Event(id) => Tag::event(*id),
+            Pin::Address(coordinate) => Tag::coordinate(coordinate.clone(), None),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct GroupPins(Vec<Pin>);
+
+impl GroupPins {
+    pub fn parse(event: &Event) -> Result<Self> {
+        if event.kind != Kind::Custom(PIN_LIST) {
+            bail!("not a pinned list event: kind {}", event.kind);
+        }
+
+        let mut pins = Vec::new();
+
+        for tag in event.tags.iter() {
+            match tag.kind() {
+                "e" => {
+                    let Some(id) = tag.content() else {
+                        continue;
+                    };
+
+                    pins.push(Pin::Event(EventId::from_hex(id)?));
+                }
+                "a" => {
+                    let Some(address) = tag.content() else {
+                        continue;
+                    };
+
+                    pins.push(Pin::Address(Coordinate::from_str(address)?));
+                }
+                _ => {}
+            }
+        }
+
+        Ok(Self(pins))
+    }
+
+    pub fn pins(&self) -> &[Pin] {
+        &self.0
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct GroupListEntry {
     pub id: GroupId,

docs/nip29-backend-plan.md

@@ -393,11 +393,50 @@ group's relay only, and treat the relay's `OK` as the truth.
 
 1. Moderation writes: `9000` put-user (roles), `9001` remove-user, `9002`
    edit-metadata, `9005` delete-event, `9009` create-invite; gate the UI on the
-   caller appearing in `39001`, let the relay be the enforcer.
+   caller appearing in `39001`, let the relay be the enforcer. **Done** as pure
-2. Group creation (`9007`) where the relay supports it, and relay browsing
+   writes: builders on `GroupId`, senders on `Group` sharing `dispatch`.
-   (`fetch_events(ReqTarget::single(relay, [Filter::new().kind(Kind::GroupMetadata).limit(..)]))`).
+   `edit_metadata` takes a `GroupMetadata`, which is now `Default`-able (its
-3. Pins: `9010` write, `39005` read (`Kind::Custom`); subgroups: parse
+   private `id`/`author` origin are `Option`) so the UI can start an edit from
-   `parent`/`child` and lay groups out as a tree.
+   scratch or from `Group::metadata()`. **`9005` is write-only for now**: the
+   relay stops serving the event but nothing is removed locally, so the deleted
+   message lingers in the cache and timeline. The UI phase must drop the target
+   envelope (filter on its `d` = target id and the group's `r` cache tag) and
+   evict it from `messages`/`index`, otherwise only a reload clears it.
+2. Group creation (`9007`) where the relay supports it, and relay browsing.
+   **Done.** `GroupId::create_group` reuses the `9002` metadata tags, so the
+   proposed id (`h`) and the initial name/flags go in one event; the relay
+   republishes `39000`, which the standing subscription picks up.
+   `GroupsRegistry::create` mirrors `join`: `ensure` → optimistic `10009` entry
+   + `publish_list` → publish, and on `OK false` it `forget`s the group and emits
+   `GroupsEvent::Error(reason)` (that variant was previously unused). There is
+   no NIP-11 support check available, so "where the relay supports it" is
+   enforced by the relay refusing the `9007`.
+   `GroupsRegistry::browse(relay, cx)` follows the same pipeline as every
+   other read: it registers a `brw-<hash>` subscription (an opaque id, like
+   `grp-<hash>`; the relay is keyed in state, not encoded in the id, because the
+   pump needs a prefix to recognise its own ids and a bare
+   `SubscriptionId::generate()` would be indistinguishable from the other
+   crates' ids on the shared client). `add_relay` +
+   `subscribe` (auto-close on EOSE, 10s cap), and returns immediately. The
+   **global pump** observes the events, and `handle_event`/`handle_eose`/
+   `handle_closed` fold each relay-signed `39000` into `self.browse`, keyed by
+   the subscription id, so no DB scoping is needed. `GroupsRegistry::browsed(relay)`
+   returns `Browse { relay, loading, groups: Vec<GroupCandidate> }` for the UI, and
+   `GroupsEvent::Updated` says when to re-read it. No `fetch_events` anywhere:
+   the DB cannot be scoped to a relay (a stored event carries no origin), so the
+   subscription's own events are the source. `join` now delegates to a new
+   `join_key(key, invite)`, which is what a browse result needs (a candidate has
+   a key, not a `GroupReference`).
+3. Pins and subgroups. **Pins done:** `Pin::{Event, Address}` and
+   `GroupPins` in `protocol.rs`, with `PIN_LIST` (39005) and `UPDATE_PIN_LIST`
+   (9010) as `u16` consts used through `Kind::Custom`. `39005` joins
+   `STATE_KINDS`, so the standing subscription fetches it, `absorb_state` folds
+   it into `Group::pins()`, and `Group::update_pins(pins)` writes `9010` through
+   the shared `dispatch`. **Subgroups parsed:** `GroupMetadata` gained
+   `pub children: Vec<GroupId>` (order preserved, parsed from `child` tags) and
+   `metadata_tags` re-emits them — `9002` requires it, an edit that drops a
+   child is rejected. `parent`/`children` are per-relay, so assembling the tree
+   from `Group::metadata()` is UI work; there is no backend tree helper.
 4. Range backfill: page the timeline with `until` for "load more".
 5. LiveKit AV (`livekit` tag, `39004`, the `/.well-known/nip29/livekit` token
    endpoint with a NIP-98 `27235` authorization event) and fork/migration
@@ -415,6 +454,11 @@ group's relay only, and treat the relay's `OK` as the truth.
 | Membership badge (member, pending, refused, removed) | `Group::membership()` |
 | Copy an invite link | `GroupMetadata::naddr(relay, invite)` |
 | Leave | `GroupsRegistry::leave` |
+| Create a group | `GroupMetadata { ..Default::default() }` → `GroupsRegistry::create(key, metadata, cx)`; failure surfaces as `GroupsEvent::Error` |
+| Browse a relay's groups | `GroupsRegistry::browse(relay, cx)` → re-read `GroupsRegistry::browsed(relay)` on `GroupsEvent::Updated`; join a result with `GroupsRegistry::join_key(candidate.key, None, cx)` |
+| Moderate (put/remove user, edit metadata, delete event, invite) | `Group::put_user/remove_user/edit_metadata/delete_event/create_invite` |
+| Pins | `Group::pins()` to read, `Group::update_pins(pins)` to replace the list |
+| Subgroup tree | `Group::metadata()` → `parent`/`children`, assembled per relay |
 
 Where it lands: the sidebar gains a section reading `GroupsRegistry::groups()`
 (the same shape as its Communities section) with pending and refused groups