add NIP-29 LiveKit token and fork detection

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",
@@ -4593,6 +4595,23 @@ version = "1.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086"
 
+[[package]]
+name = "nip29"
+version = "1.0.2"
+dependencies = [
+ "anyhow",
+ "chat",
+ "flume 0.11.1",
+ "futures",
+ "gpui-pre",
+ "log",
+ "nostr",
+ "nostr-sdk",
+ "serde_json",
+ "smallvec",
+ "state",
+]
+
 [[package]]
 name = "no_std_io2"
 version = "0.9.4"

Cargo.toml

@@ -32,7 +32,7 @@ nostr-blossom = { git = "https://github.com/rust-nostr/nostr" }
 nostr-gossip-memory = { git = "https://github.com/rust-nostr/nostr" }
 nostr-connect = { git = "https://github.com/rust-nostr/nostr" }
 nostr-sdk = { git = "https://github.com/rust-nostr/nostr" }
-nostr = { git = "https://github.com/rust-nostr/nostr", features = [ "nip59", "nip49", "nip44" ] }
+nostr = { git = "https://github.com/rust-nostr/nostr", features = [ "nip59", "nip49", "nip44", "nip98" ] }
 
 # Crypto (NIP-17 encrypted file messages)
 aes-gcm = "0.10"

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/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "nip29"
+version.workspace = true
+edition.workspace = true
+publish.workspace = true
+
+[dependencies]
+nostr.workspace = true
+nostr-sdk.workspace = true
+anyhow.workspace = true
+futures.workspace = true
+serde_json.workspace = true
+
+state = { path = "../state" }
+chat = { path = "../chat" }
+
+gpui.workspace = true
+flume.workspace = true
+log.workspace = true
+smallvec.workspace = true

crates/nip29/src/group.rs

@@ -0,0 +1,608 @@
+use std::collections::{HashMap, HashSet};
+use std::time::Duration;
+
+use anyhow::{Result, anyhow, bail};
+use chat::Message;
+use futures::io::AsyncReadExt;
+use gpui::http_client::{AsyncBody, Builder, HttpRequestExt, Method, RedirectPolicy, StatusCode};
+use gpui::{App, AppContext, Context, EventEmitter, Task};
+use nostr::nips::nip98::{HttpData, HttpMethod};
+use nostr_sdk::prelude::*;
+use smallvec::{SmallVec, smallvec};
+use state::NostrRegistry;
+
+use crate::LOCAL_KEYS;
+use crate::protocol::{
+    GroupAdmins, GroupId, GroupKey, GroupMembers, GroupMetadata, GroupParticipants, GroupPins,
+    GroupRoles, LivekitSession, MEMBERSHIP_KINDS, Membership, PIN_LIST, Pin, RENDER_KINDS,
+    STATE_KINDS, TimelineWindow,
+};
+
+const REPLAY_LIMIT: usize = 100;
+const PAGE_LIMIT: usize = 50;
+const PAGE_TIMEOUT: Duration = Duration::from_secs(10);
+const LIVEKIT_TIMEOUT: Duration = Duration::from_secs(10);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Absorbed {
+    Ignored,
+    Rendered { fresh: bool },
+    Changed,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum GroupEvent {
+    Incoming(Event),
+    Sent(Event),
+    Updated,
+    Eose,
+    Refused(String),
+}
+
+pub struct Group {
+    key: GroupKey,
+    metadata: Option<GroupMetadata>,
+    members: GroupMembers,
+    admins: GroupAdmins,
+    roles: GroupRoles,
+    participants: GroupParticipants,
+    pins: GroupPins,
+    membership: Membership,
+    membership_events: Vec<Event>,
+    elsewhere: Option<GroupKey>,
+    state_ids: HashMap<Kind, EventId>,
+    messages: Vec<Message>,
+    index: HashSet<EventId>,
+    window: TimelineWindow,
+    oldest: Option<Timestamp>,
+    loading: bool,
+    loading_more: bool,
+    refused: Option<String>,
+    tasks: SmallVec<[Task<()>; 2]>,
+}
+
+impl EventEmitter<GroupEvent> for Group {}
+
+impl Group {
+    pub(crate) fn new(key: GroupKey) -> Self {
+        Self {
+            key,
+            metadata: None,
+            members: GroupMembers::default(),
+            admins: GroupAdmins::default(),
+            roles: GroupRoles::default(),
+            participants: GroupParticipants::default(),
+            pins: GroupPins::default(),
+            membership: Membership::Unknown,
+            membership_events: Vec::new(),
+            elsewhere: None,
+            state_ids: HashMap::new(),
+            messages: Vec::new(),
+            index: HashSet::new(),
+            window: TimelineWindow::new(),
+            oldest: None,
+            loading: true,
+            loading_more: false,
+            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 participants(&self) -> &GroupParticipants {
+        &self.participants
+    }
+
+    pub fn pins(&self) -> &GroupPins {
+        &self.pins
+    }
+
+    pub fn membership(&self) -> &Membership {
+        &self.membership
+    }
+
+    pub fn elsewhere(&self) -> Option<&GroupKey> {
+        self.elsewhere.as_ref()
+    }
+
+    pub fn messages(&self) -> &[Message] {
+        &self.messages
+    }
+
+    pub fn loading(&self) -> bool {
+        self.loading
+    }
+
+    pub fn loading_more(&self) -> bool {
+        self.loading_more
+    }
+
+    pub fn oldest(&self) -> Option<Timestamp> {
+        self.oldest
+    }
+
+    pub fn refused(&self) -> Option<&str> {
+        self.refused.as_deref()
+    }
+
+    pub fn previous_tag(&self, me: PublicKey) -> Option<Tag> {
+        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 set_elsewhere(&mut self, elsewhere: Option<GroupKey>, cx: &mut Context<Self>) {
+        if self.elsewhere == elsewhere {
+            return;
+        }
+        self.elsewhere = elsewhere;
+        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();
+
+        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 load_page(&mut self, cx: &mut Context<Self>) {
+        let Some(until) = self.oldest else {
+            return;
+        };
+
+        self.loading_more = true;
+        cx.notify();
+
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+
+        let id = crate::page_subscription(&self.key);
+        let relay = self.key.relay().clone();
+        let filter = self.key.id().timeline_page(PAGE_LIMIT, until);
+
+        self.tasks.push(cx.background_spawn(async move {
+            client.add_relay(relay.clone()).and_connect().await.ok();
+
+            let opts = SubscribeAutoCloseOptions::default()
+                .exit_policy(ReqExitPolicy::ExitOnEOSE)
+                .timeout(Some(PAGE_TIMEOUT));
+
+            if let Err(error) = client
+                .subscribe(ReqTarget::single(relay, [filter]))
+                .with_id(id)
+                .close_on(opts)
+                .await
+            {
+                log::warn!("nip29: loading older messages failed: {error}");
+            }
+        }));
+    }
+
+    pub(crate) fn page_finished(&mut self, cx: &mut Context<Self>) {
+        self.loading_more = false;
+        cx.emit(GroupEvent::Updated);
+        cx.notify();
+    }
+
+    pub fn livekit_session(&self, cx: &App) -> Task<Result<LivekitSession>> {
+        let url = match self.key.livekit_token_url() {
+            Ok(url) => url,
+            Err(error) => return Task::ready(Err(error)),
+        };
+
+        let nostr = NostrRegistry::global(cx);
+        let signer = nostr.read(cx).signer();
+
+        let client = cx.http_client();
+
+        cx.background_spawn(async move {
+            let authorization = HttpData::new(url.clone(), HttpMethod::GET)
+                .to_authorization(&signer)
+                .await?;
+
+            let request = Builder::new()
+                .uri(url.as_str())
+                .method(Method::GET)
+                .header("Authorization", authorization)
+                .follow_redirects(RedirectPolicy::NoFollow)
+                .timeout(LIVEKIT_TIMEOUT)
+                .body(AsyncBody::default())?;
+
+            let response = client.send(request).await?;
+
+            if !response.status().is_success() {
+                bail!(
+                    "nip29: the AV token endpoint answered {}",
+                    response.status()
+                );
+            }
+
+            let mut body = Vec::new();
+            response.into_body().read_to_end(&mut body).await?;
+
+            LivekitSession::parse(&body)
+        })
+    }
+
+    pub fn livekit_supported(&self, cx: &App) -> Task<Result<bool>> {
+        let url = match self.key.livekit_probe_url() {
+            Ok(url) => url,
+            Err(error) => return Task::ready(Err(error)),
+        };
+
+        let client = cx.http_client();
+
+        cx.background_spawn(async move {
+            let request = Builder::new()
+                .uri(url.as_str())
+                .method(Method::GET)
+                .follow_redirects(RedirectPolicy::NoFollow)
+                .timeout(LIVEKIT_TIMEOUT)
+                .body(AsyncBody::default())?;
+
+            let response = client.send(request).await?;
+
+            Ok(response.status() == StatusCode::NO_CONTENT)
+        })
+    }
+
+    pub(crate) fn save_envelope(&self, event: Event, cx: &App) -> Task<Result<()>> {
+        let cache_tag = self.key.cache_tag();
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).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<Result<Vec<Event>>> {
+        let cache_tag = self.key.cache_tag();
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+
+        cx.background_spawn(async move {
+            let filter = Filter::new()
+                .kind(Kind::ApplicationSpecificData)
+                .custom_tag(SingleLetterTag::LOWERCASE_R, cache_tag);
+
+            let mut events: Vec<Event> = 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<Event>, me: PublicKey, cx: &mut Context<Self>) {
+        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>) {
+        self.loading = false;
+        cx.emit(GroupEvent::Eose);
+        cx.notify();
+    }
+
+    pub(crate) fn refuse(&mut self, reason: String, cx: &mut Context<Self>) {
+        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.track(event);
+
+            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.track(event);
+
+        if RENDER_KINDS.contains(&event.kind) {
+            Absorbed::Rendered {
+                fresh: self.push_message(event),
+            }
+        } else {
+            Absorbed::Ignored
+        }
+    }
+
+    fn track(&mut self, event: &Event) {
+        self.window.push(event.clone());
+
+        if self.oldest.is_none_or(|oldest| event.created_at < oldest) {
+            self.oldest = Some(event.created_at);
+        }
+    }
+
+    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,
+            },
+            Kind::GroupLivekitParticipants => match GroupParticipants::parse(event) {
+                Ok(participants) => self.participants = participants,
+                Err(_) => return false,
+            },
+            Kind::Custom(PIN_LIST) => match GroupPins::parse(event) {
+                Ok(pins) => self.pins = pins,
+                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
+    }
+}

crates/nip29/src/lib.rs

@@ -0,0 +1,895 @@
+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};
+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<Keys> = LazyLock::new(Keys::generate);
+
+const LIST_SUBSCRIPTION: &str = "nip29-list";
+const FORK_SUBSCRIPTION: &str = "nip29-forks";
+const GROUP_PREFIX: &str = "grp-";
+const BROWSE_PREFIX: &str = "brw-";
+const PAGE_PREFIX: &str = "pg-";
+const BROWSE_LIMIT: usize = 500;
+const BROWSE_TIMEOUT: Duration = Duration::from_secs(10);
+const FORK_INTERVAL: Duration = Duration::from_secs(15 * 60);
+
+pub fn init(cx: &mut App) {
+    GroupsRegistry::set_global(cx.new(GroupsRegistry::new), cx);
+}
+
+struct GlobalGroupsRegistry(Entity<GroupsRegistry>);
+
+impl Global for GlobalGroupsRegistry {}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum GroupsEvent {
+    Updated,
+    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),
+    Eose(SubscriptionId),
+    Closed(SubscriptionId, String),
+    Forks,
+}
+
+fn list_subscription_id() -> SubscriptionId {
+    SubscriptionId::new(LIST_SUBSCRIPTION)
+}
+
+fn fork_subscription_id() -> SubscriptionId {
+    SubscriptionId::new(FORK_SUBSCRIPTION)
+}
+
+fn subscription_id(key: &GroupKey) -> SubscriptionId {
+    SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32))
+}
+
+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 page_subscription(key: &GroupKey) -> SubscriptionId {
+    SubscriptionId::new(format!("{PAGE_PREFIX}{:08x}", key.uniq_id() as u32))
+}
+
+fn is_page(id: &SubscriptionId) -> bool {
+    id.as_str().starts_with(PAGE_PREFIX)
+}
+
+fn is_own_subscription(id: &SubscriptionId) -> bool {
+    let id = id.as_str();
+    id == LIST_SUBSCRIPTION
+        || id == FORK_SUBSCRIPTION
+        || id.starts_with(GROUP_PREFIX)
+        || id.starts_with(BROWSE_PREFIX)
+        || id.starts_with(PAGE_PREFIX)
+}
+
+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>,
+    forks: HashMap<PublicKey, (Timestamp, GroupList)>,
+    fork_authors: HashSet<PublicKey>,
+    signal_tx: flume::Sender<Signal>,
+    signal_rx: flume::Receiver<Signal>,
+    notification_listener: Option<Task<()>>,
+    signal_consumer: Option<Task<()>>,
+    tasks: SmallVec<[Task<()>; 2]>,
+    _subscriptions: SmallVec<[Subscription; 2]>,
+}
+
+impl GroupsRegistry {
+    pub fn global(cx: &App) -> Entity<Self> {
+        cx.global::<GlobalGroupsRegistry>().0.clone()
+    }
+
+    fn set_global(state: Entity<Self>, cx: &mut App) {
+        cx.set_global(GlobalGroupsRegistry(state));
+    }
+
+    fn new(cx: &mut Context<Self>) -> Self {
+        let nostr = NostrRegistry::global(cx);
+        let this = cx.entity().downgrade();
+
+        let (tx, rx) = flume::unbounded::<Signal>();
+        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(),
+            browse: HashMap::new(),
+            routes: HashMap::new(),
+            observers: HashMap::new(),
+            forks: HashMap::new(),
+            fork_authors: HashSet::new(),
+            signal_tx: tx,
+            signal_rx: rx,
+            notification_listener: None,
+            signal_consumer: None,
+            tasks: smallvec![],
+            _subscriptions: subscriptions,
+        }
+    }
+
+    pub fn groups(&self) -> &[Entity<Group>] {
+        &self.groups
+    }
+
+    pub fn group(&self, key: &GroupKey, cx: &App) -> Option<Entity<Group>> {
+        self.groups
+            .iter()
+            .find(|group| group.read(cx).key() == key)
+            .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 load_more(&mut self, key: &GroupKey, cx: &mut Context<Self>) {
+        let Some(group) = self.group(key, cx) else {
+            return;
+        };
+
+        if group.read(cx).loading_more() || group.read(cx).oldest().is_none() {
+            return;
+        }
+
+        self.routes.insert(page_subscription(key), key.clone());
+
+        group.update(cx, |group, cx| {
+            group.load_page(cx);
+        });
+    }
+
+    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;
+        self.tasks.clear();
+
+        let routes = std::mem::take(&mut self.routes);
+
+        self.groups.clear();
+        self.observers.clear();
+        self.forks.clear();
+        self.fork_authors.clear();
+        self.list = GroupList::default();
+
+        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.push(fork_subscription_id());
+        ids.extend(self.browse.keys().cloned());
+
+        self.browse.clear();
+
+        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<Self>) {
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+        let signer = nostr.read(cx).signer();
+
+        let task: Task<Result<GroupList, Error>> = 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}"),
+            }));
+
+        self.watch_forks(cx);
+    }
+
+    fn merge(&mut self, list: GroupList, cx: &mut Context<Self>) {
+        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 watch_forks(&mut self, cx: &mut Context<Self>) {
+        let executor = cx.background_executor().clone();
+        let tx = self.signal_tx.clone();
+
+        self.tasks.push(cx.background_spawn(async move {
+            loop {
+                executor.timer(FORK_INTERVAL).await;
+
+                if tx.send_async(Signal::Forks).await.is_err() {
+                    break;
+                }
+            }
+        }));
+    }
+
+    fn admin_keys(&self, cx: &App) -> HashSet<PublicKey> {
+        let mut keys = HashSet::new();
+
+        for group in &self.groups {
+            for (admin, _) in group.read(cx).admins().entries() {
+                keys.insert(*admin);
+            }
+        }
+
+        keys
+    }
+
+    fn sync_forks(&mut self, force: bool, cx: &mut Context<Self>) {
+        let authors = self.admin_keys(cx);
+
+        if authors.is_empty() || (!force && authors == self.fork_authors) {
+            return;
+        }
+
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+        let filter = Filter::new()
+            .authors(authors.iter().copied())
+            .kind(Kind::SimpleGroups);
+
+        self.fork_authors = authors;
+
+        self.tasks.push(cx.background_spawn(async move {
+            if let Err(error) = client
+                .subscribe(ReqTarget::auto([filter]))
+                .with_id(fork_subscription_id())
+                .await
+            {
+                log::warn!("nip29: subscribing to the admins' group lists failed: {error}");
+            }
+        }));
+    }
+
+    fn absorb_fork_list(&mut self, event: &Event, cx: &mut Context<Self>) {
+        let Ok(list) = GroupList::parse(event) else {
+            log::warn!("nip29: ignoring a malformed admin group list");
+            return;
+        };
+
+        let author = event.pubkey;
+
+        if let Some((seen, _)) = self.forks.get(&author)
+            && *seen >= event.created_at
+        {
+            return;
+        }
+
+        self.forks.insert(author, (event.created_at, list));
+        self.evaluate_forks(cx);
+    }
+
+    fn evaluate_forks(&mut self, cx: &mut Context<Self>) {
+        let groups = self.groups.clone();
+
+        for group in groups {
+            let elsewhere = {
+                let group = group.read(cx);
+                let key = group.key();
+
+                group
+                    .admins()
+                    .entries()
+                    .iter()
+                    .filter_map(|(admin, _)| self.forks.get(admin))
+                    .flat_map(|(_, list)| list.entries())
+                    .find(|entry| entry.id == *key.id() && entry.relay != *key.relay())
+                    .map(|entry| GroupKey::new(entry.relay.clone(), entry.id.clone()))
+            };
+
+            group.update(cx, |group, cx| {
+                group.set_elsewhere(elsewhere, cx);
+            });
+        }
+    }
+
+    fn ensure(&mut self, key: GroupKey, cx: &mut Context<Self>) {
+        if self.group(&key, cx).is_some() {
+            return;
+        }
+
+        let group = cx.new(|_| Group::new(key.clone()));
+        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);
+                    this.sync_forks(false, cx);
+                }
+                _ => {}
+            }),
+        );
+
+        self.subscribe(key.clone(), cx);
+        self.reload(key, None, cx);
+    }
+
+    fn subscribe(&mut self, key: GroupKey, cx: &mut Context<Self>) {
+        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 forget(&mut self, key: GroupKey, cx: &mut Context<Self>) {
+        let id = subscription_id(&key);
+        let page = page_subscription(&key);
+        let cache_tag = key.cache_tag();
+
+        self.routes.remove(&id);
+        self.routes.remove(&page);
+        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 {
+            for id in [id, page] {
+                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);
+
+        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<Self>) {
+        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;
+        }
+
+        if id == fork_subscription_id() {
+            self.absorb_fork_list(&event, cx);
+            return;
+        }
+
+        if self.browse.contains_key(&id) {
+            self.absorb_candidate(&id, &event, cx);
+            return;
+        }
+
+        let Some(key) = self.routes.get(&id).cloned() else {
+            return;
+        };
+
+        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;
+        };
+
+        if is_page(&id) {
+            self.routes.remove(&id);
+
+            if let Some(group) = self.group(&key, cx) {
+                group.update(cx, |group, cx| {
+                    group.page_finished(cx);
+                });
+            }
+
+            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<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;
+        }
+
+        if id == fork_subscription_id() {
+            log::warn!("nip29: the admins' group lists were refused: {reason}");
+            return;
+        }
+
+        let Some(key) = self.routes.get(&id).cloned() else {
+            return;
+        };
+
+        if is_page(&id) {
+            self.routes.remove(&id);
+            log::warn!(
+                "nip29: loading older messages in {} was refused: {reason}",
+                key.id()
+            );
+
+            if let Some(group) = self.group(&key, cx) {
+                group.update(cx, |group, cx| {
+                    group.page_finished(cx);
+                });
+            }
+
+            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>) {
+        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_own_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_own_subscription(&id).then_some(Signal::Eose(id))
+                    }
+                    RelayMessage::Closed {
+                        subscription_id,
+                        message,
+                    } => {
+                        let id = subscription_id.into_owned();
+                        is_own_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);
+                    }
+                    Signal::Forks => {
+                        this.sync_forks(true, cx);
+                    }
+                })
+                .ok();
+            }
+        }));
+    }
+}

crates/nip29/src/protocol.rs

@@ -0,0 +1,837 @@
+use std::collections::hash_map::DefaultHasher;
+use std::fmt;
+use std::hash::{Hash, Hasher};
+use std::str::FromStr;
+
+use anyhow::{Result, anyhow, bail};
+use nostr_sdk::prelude::*;
+
+pub const TIMELINE_WINDOW: usize = 50;
+
+const PREVIOUS_REFS: usize = 3;
+const PREVIOUS_REF_LEN: usize = 8;
+
+pub const RENDER_KINDS: [Kind; 5] = [
+    Kind::TextNote,
+    Kind::ChatMessage,
+    Kind::Thread,
+    Kind::Comment,
+    Kind::Reaction,
+];
+
+pub const PIN_LIST: u16 = 39_005;
+pub const UPDATE_PIN_LIST: u16 = 9_010;
+
+pub const STATE_KINDS: [Kind; 6] = [
+    Kind::GroupMetadata,
+    Kind::GroupAdmins,
+    Kind::GroupMembers,
+    Kind::GroupRoles,
+    Kind::GroupLivekitParticipants,
+    Kind::Custom(PIN_LIST),
+];
+
+pub const MEMBERSHIP_KINDS: [Kind; 2] = [Kind::GroupPutUser, Kind::GroupRemoveUser];
+
+fn has_tag(tags: &Tags, name: &str) -> bool {
+    tags.iter().any(|tag| tag.kind() == name)
+}
+
+fn tag_value<'a>(tags: &'a Tags, name: &str) -> Option<&'a str> {
+    tags.iter()
+        .find(|tag| tag.kind() == name)
+        .and_then(|tag| tag.content())
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct GroupId(String);
+
+impl GroupId {
+    pub fn new(id: impl Into<String>) -> Result<Self> {
+        let id = id.into();
+        let id = id.trim();
+
+        if id.is_empty() {
+            bail!("group id is empty");
+        }
+
+        Ok(Self(id.to_owned()))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+
+    pub fn h_tag(&self) -> Tag {
+        Tag::custom("h", [self.0.clone()])
+    }
+
+    pub fn timeline_filter(&self, limit: usize) -> Filter {
+        Filter::new()
+            .custom_tag(SingleLetterTag::LOWERCASE_H, self.0.clone())
+            .limit(limit)
+    }
+
+    pub fn timeline_page(&self, limit: usize, until: Timestamp) -> Filter {
+        self.timeline_filter(limit).until(until)
+    }
+
+    pub fn state_filter(&self) -> Filter {
+        Filter::new().identifier(self.0.clone()).kinds(STATE_KINDS)
+    }
+
+    pub fn membership_filter(&self, me: PublicKey) -> Filter {
+        Filter::new()
+            .custom_tag(SingleLetterTag::LOWERCASE_H, self.0.clone())
+            .custom_tag(SingleLetterTag::LOWERCASE_P, me.to_hex())
+            .kinds(MEMBERSHIP_KINDS)
+    }
+
+    pub fn message(
+        &self,
+        content: &str,
+        replies: &[EventId],
+        previous: Option<Tag>,
+    ) -> EventBuilder {
+        EventBuilder::new(Kind::ChatMessage, content)
+            .tag(self.h_tag())
+            .tags(replies.iter().copied().map(Tag::event))
+            .tag_maybe(previous)
+    }
+
+    pub fn reaction(&self, target: EventId, emoji: &str, previous: Option<Tag>) -> EventBuilder {
+        EventBuilder::new(Kind::Reaction, emoji)
+            .tag(self.h_tag())
+            .tag(Tag::event(target))
+            .tag_maybe(previous)
+    }
+
+    pub fn join_request(&self, code: Option<&str>, reason: &str) -> EventBuilder {
+        EventBuilder::new(Kind::GroupJoinRequest, reason)
+            .tag(self.h_tag())
+            .tag_maybe(code.map(|code| Tag::custom("code", [code])))
+    }
+
+    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 {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(&self.0)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct GroupKey {
+    relay: RelayUrl,
+    id: GroupId,
+}
+
+impl GroupKey {
+    pub fn new(relay: RelayUrl, id: GroupId) -> Self {
+        Self { relay, id }
+    }
+
+    pub fn relay(&self) -> &RelayUrl {
+        &self.relay
+    }
+
+    pub fn id(&self) -> &GroupId {
+        &self.id
+    }
+
+    pub fn cache_tag(&self) -> String {
+        format!("{}|{}", self.relay.as_str(), self.id)
+    }
+
+    pub fn uniq_id(&self) -> u64 {
+        let mut hasher = DefaultHasher::new();
+        self.hash(&mut hasher);
+        hasher.finish()
+    }
+
+    pub fn livekit_token_url(&self) -> Result<Url> {
+        self.livekit_url(Some(self.id.as_str()))
+    }
+
+    pub fn livekit_probe_url(&self) -> Result<Url> {
+        self.livekit_url(None)
+    }
+
+    fn livekit_url(&self, group: Option<&str>) -> Result<Url> {
+        let mut url: Url = self.relay.clone().into();
+        let scheme = if self.relay.scheme().is_secure() {
+            "https"
+        } else {
+            "http"
+        };
+
+        url.set_path("/");
+        url.set_query(None);
+        url.set_fragment(None);
+        url.set_scheme(scheme)
+            .map_err(|_| anyhow!("{scheme} is not available for {}", self.relay))?;
+
+        {
+            let mut segments = url
+                .path_segments_mut()
+                .map_err(|_| anyhow!("{} cannot carry a path", self.relay))?;
+
+            segments
+                .pop_if_empty()
+                .push(".well-known")
+                .push("nip29")
+                .push("livekit");
+
+            if let Some(group) = group {
+                segments.push(group);
+            }
+        }
+
+        Ok(url)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct GroupReference {
+    id: GroupId,
+    relay: RelayUrl,
+    invite: Option<String>,
+}
+
+impl GroupReference {
+    pub fn parse(input: &str) -> Result<Self> {
+        let input = input.trim();
+
+        let (address, query) = match input.split_once('?') {
+            Some((address, query)) => (address, Some(query)),
+            None => (input, None),
+        };
+
+        let coordinate = Nip19Coordinate::from_bech32(address)?;
+
+        if coordinate.kind != Kind::GroupMetadata {
+            bail!("not a group reference: kind {}", coordinate.kind);
+        }
+
+        let relay = coordinate
+            .relays
+            .first()
+            .cloned()
+            .ok_or_else(|| anyhow!("group reference has no relay hint"))?;
+
+        let id = GroupId::new(coordinate.identifier.as_str())?;
+
+        let invite = query
+            .and_then(|query| {
+                query
+                    .split('&')
+                    .find_map(|pair| pair.strip_prefix("invite="))
+            })
+            .map(str::to_owned);
+
+        Ok(Self { id, relay, invite })
+    }
+
+    pub fn id(&self) -> &GroupId {
+        &self.id
+    }
+
+    pub fn relay(&self) -> &RelayUrl {
+        &self.relay
+    }
+
+    pub fn invite(&self) -> Option<&str> {
+        self.invite.as_deref()
+    }
+
+    pub fn key(&self) -> GroupKey {
+        GroupKey::new(self.relay.clone(), self.id.clone())
+    }
+}
+
+impl FromStr for GroupReference {
+    type Err = anyhow::Error;
+
+    fn from_str(input: &str) -> Result<Self> {
+        Self::parse(input)
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct GroupMetadata {
+    pub name: Option<String>,
+    pub picture: Option<String>,
+    pub banner: Option<String>,
+    pub about: Option<String>,
+    pub private: bool,
+    pub restricted: bool,
+    pub hidden: bool,
+    pub closed: bool,
+    pub livekit: bool,
+    pub parent: Option<GroupId>,
+    pub children: Vec<GroupId>,
+    pub supported_kinds: Option<Vec<Kind>>,
+    id: Option<GroupId>,
+    author: Option<PublicKey>,
+}
+
+impl GroupMetadata {
+    pub fn parse(event: &Event) -> Result<Self> {
+        if event.kind != Kind::GroupMetadata {
+            bail!("not a group metadata event: kind {}", event.kind);
+        }
+
+        let identifier = event
+            .tags
+            .identifier()
+            .ok_or_else(|| anyhow!("group metadata has no d tag"))?;
+
+        let mut supported_kinds: Option<Vec<Kind>> = None;
+
+        for tag in event
+            .tags
+            .iter()
+            .filter(|tag| tag.kind() == "supported_kinds")
+        {
+            let kinds = supported_kinds.get_or_insert_with(Vec::new);
+
+            for value in tag.as_slice().iter().skip(1) {
+                kinds.push(Kind::from_str(value)?);
+            }
+        }
+
+        let parent = match tag_value(&event.tags, "parent") {
+            Some(parent) => Some(GroupId::new(parent)?),
+            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),
+            banner: tag_value(&event.tags, "banner").map(str::to_owned),
+            about: tag_value(&event.tags, "about").map(str::to_owned),
+            private: has_tag(&event.tags, "private"),
+            restricted: has_tag(&event.tags, "restricted"),
+            hidden: has_tag(&event.tags, "hidden"),
+            closed: has_tag(&event.tags, "closed"),
+            livekit: has_tag(&event.tags, "livekit"),
+            parent,
+            children,
+            supported_kinds,
+            id: Some(GroupId::new(identifier)?),
+            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()
+    }
+
+    pub fn supports(&self, kind: Kind) -> bool {
+        match &self.supported_kinds {
+            None => true,
+            Some(kinds) => kinds.contains(&kind),
+        }
+    }
+
+    pub fn naddr(&self, relay: &RelayUrl, invite: Option<&str>) -> Result<String> {
+        let id = self.id.as_ref().ok_or_else(|| anyhow!("no id"))?;
+        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 {
+            Some(invite) => format!("{address}?invite={invite}"),
+            None => address,
+        })
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct GroupCandidate {
+    pub key: GroupKey,
+    pub metadata: GroupMetadata,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct GroupAdmins(Vec<(PublicKey, Vec<String>)>);
+
+impl GroupAdmins {
+    pub fn parse(event: &Event) -> Result<Self> {
+        if event.kind != Kind::GroupAdmins {
+            bail!("not a group admins event: kind {}", event.kind);
+        }
+
+        let mut admins = Vec::new();
+
+        for tag in event.tags.iter().filter(|tag| tag.kind() == "p") {
+            let mut fields = tag.as_slice().iter().skip(1);
+
+            let public_key = fields
+                .next()
+                .ok_or_else(|| anyhow!("admin tag has no public key"))?;
+            let public_key = PublicKey::from_hex(public_key)?;
+
+            admins.push((public_key, fields.cloned().collect()));
+        }
+
+        Ok(Self(admins))
+    }
+
+    pub fn entries(&self) -> &[(PublicKey, Vec<String>)] {
+        &self.0
+    }
+
+    pub fn contains(&self, public_key: &PublicKey) -> bool {
+        self.0.iter().any(|(key, _)| key == public_key)
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct GroupMembers(Vec<PublicKey>);
+
+impl GroupMembers {
+    pub fn parse(event: &Event) -> Result<Self> {
+        if event.kind != Kind::GroupMembers {
+            bail!("not a group members event: kind {}", event.kind);
+        }
+
+        Ok(Self(event.tags.public_keys().collect()))
+    }
+
+    pub fn contains(&self, public_key: &PublicKey) -> bool {
+        self.0.contains(public_key)
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = &PublicKey> {
+        self.0.iter()
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Role {
+    pub name: String,
+    pub description: Option<String>,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct GroupRoles(Vec<Role>);
+
+impl GroupRoles {
+    pub fn parse(event: &Event) -> Result<Self> {
+        if event.kind != Kind::GroupRoles {
+            bail!("not a group roles event: kind {}", event.kind);
+        }
+
+        let mut roles = Vec::new();
+
+        for tag in event.tags.iter().filter(|tag| tag.kind() == "role") {
+            let mut fields = tag.as_slice().iter().skip(1);
+            let name = fields
+                .next()
+                .ok_or_else(|| anyhow!("role tag has no name"))?;
+
+            roles.push(Role {
+                name: name.clone(),
+                description: fields.next().cloned(),
+            });
+        }
+
+        Ok(Self(roles))
+    }
+
+    pub fn roles(&self) -> &[Role] {
+        &self.0
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct GroupParticipants(Vec<PublicKey>);
+
+impl GroupParticipants {
+    pub fn parse(event: &Event) -> Result<Self> {
+        if event.kind != Kind::GroupLivekitParticipants {
+            bail!("not a group participants event: kind {}", event.kind);
+        }
+
+        let mut participants = Vec::new();
+
+        for tag in event.tags.iter().filter(|tag| tag.kind() == "participant") {
+            let Some(public_key) = tag.content() else {
+                continue;
+            };
+
+            participants.push(PublicKey::parse(public_key)?);
+        }
+
+        Ok(Self(participants))
+    }
+
+    pub fn contains(&self, public_key: &PublicKey) -> bool {
+        self.0.contains(public_key)
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = &PublicKey> {
+        self.0.iter()
+    }
+}
+
+#[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 LivekitSession {
+    pub token: String,
+    pub server_url: String,
+}
+
+impl LivekitSession {
+    pub fn parse(body: &[u8]) -> Result<Self> {
+        let value: serde_json::Value = serde_json::from_slice(body)?;
+
+        let token = value
+            .get("token")
+            .or_else(|| value.get("jwt"))
+            .and_then(serde_json::Value::as_str)
+            .ok_or_else(|| anyhow!("the AV response carries no token"))?;
+
+        let server_url = value
+            .get("url")
+            .or_else(|| value.get("server_url"))
+            .and_then(serde_json::Value::as_str)
+            .ok_or_else(|| anyhow!("the AV response carries no server url"))?;
+
+        Ok(Self {
+            token: token.to_owned(),
+            server_url: server_url.to_owned(),
+        })
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct GroupListEntry {
+    pub id: GroupId,
+    pub relay: RelayUrl,
+    pub name: Option<String>,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct GroupList(Vec<GroupListEntry>);
+
+impl GroupList {
+    pub fn parse(event: &Event) -> Result<Self> {
+        if event.kind != Kind::SimpleGroups {
+            bail!("not a group list event: kind {}", event.kind);
+        }
+
+        let mut entries = Vec::new();
+
+        for tag in event.tags.iter().filter(|tag| tag.kind() == "group") {
+            let fields = tag.as_slice();
+
+            let (Some(id), Some(relay)) = (fields.get(1), fields.get(2)) else {
+                continue;
+            };
+
+            let Ok(id) = GroupId::new(id.as_str()) else {
+                continue;
+            };
+            let Ok(relay) = RelayUrl::parse(relay.as_str()) else {
+                continue;
+            };
+
+            entries.push(GroupListEntry {
+                id,
+                relay,
+                name: fields.get(3).cloned(),
+            });
+        }
+
+        Ok(Self(entries))
+    }
+
+    pub fn entries(&self) -> &[GroupListEntry] {
+        &self.0
+    }
+
+    pub fn upsert(&mut self, entry: GroupListEntry) {
+        match self
+            .0
+            .iter_mut()
+            .find(|existing| existing.id == entry.id && existing.relay == entry.relay)
+        {
+            Some(existing) => *existing = entry,
+            None => self.0.push(entry),
+        }
+    }
+
+    pub fn remove(&mut self, key: &GroupKey) {
+        self.0
+            .retain(|entry| entry.id != *key.id() || entry.relay != *key.relay());
+    }
+
+    pub fn to_builder(&self) -> EventBuilder {
+        let mut tags: Vec<Tag> = Vec::with_capacity(self.0.len() * 2);
+
+        for entry in &self.0 {
+            let mut values = vec![
+                entry.id.as_str().to_owned(),
+                entry.relay.as_str().to_owned(),
+            ];
+
+            if let Some(name) = &entry.name {
+                values.push(name.clone());
+            }
+
+            tags.push(Tag::custom("group", values));
+            tags.push(Tag::custom("r", [entry.relay.as_str()]));
+        }
+
+        EventBuilder::new(Kind::SimpleGroups, "").tags(tags)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Membership {
+    Unknown,
+    Pending { since: Timestamp },
+    Member,
+    Refused { reason: Option<String> },
+    Removed,
+}
+
+impl Membership {
+    pub fn from_events(events: &[Event], me: PublicKey) -> Self {
+        let mut latest: Option<&Event> = None;
+
+        for event in events {
+            if !MEMBERSHIP_KINDS.contains(&event.kind) {
+                continue;
+            }
+
+            if !event.tags.public_keys().any(|key| key == me) {
+                continue;
+            }
+
+            let newer = match latest {
+                None => true,
+                Some(current) => {
+                    event.created_at > current.created_at
+                        || (event.created_at == current.created_at
+                            && current.kind == Kind::GroupPutUser
+                            && event.kind == Kind::GroupRemoveUser)
+                }
+            };
+
+            if newer {
+                latest = Some(event);
+            }
+        }
+
+        match latest {
+            Some(event) if event.kind == Kind::GroupRemoveUser => Membership::Removed,
+            Some(_) => Membership::Member,
+            None => Membership::Unknown,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct TimelineWindow(Vec<Event>);
+
+impl TimelineWindow {
+    pub fn new() -> Self {
+        Self(Vec::new())
+    }
+
+    pub fn push(&mut self, event: Event) {
+        self.0.retain(|existing| existing.id != event.id);
+        let position = self.0.partition_point(|e| e.created_at > event.created_at);
+
+        self.0.insert(position, event);
+        self.0.truncate(TIMELINE_WINDOW);
+    }
+
+    pub fn previous_tag(&self, me: PublicKey) -> Option<Tag> {
+        let references: Vec<String> = self
+            .0
+            .iter()
+            .filter(|event| event.pubkey != me)
+            .take(PREVIOUS_REFS)
+            .map(|event| event.id.to_hex().chars().take(PREVIOUS_REF_LEN).collect())
+            .collect();
+
+        if references.is_empty() {
+            None
+        } else {
+            Some(Tag::custom("previous", references))
+        }
+    }
+}

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" }

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);
 

docs/nip29-backend-plan.md

@@ -0,0 +1,650 @@
+# NIP-29 groups backend
+
+Coop has two relay-backed backends today: `crates/chat` (NIP-17 gift-wrap DMs,
+rendered by `crates/chat_ui`) and `crates/community` (Concord). This document is a
+plan, not a description of shipped code; nothing below is wired up yet. It covers
+the backend only — reading, writing, joining and leaving relay-based groups as
+defined by [NIP-29](https://github.com/nostr-protocol/nips/blob/master/29.md) —
+and ends with the hooks a UI will use later.
+
+## The spec in backend terms
+
+A group is a `(relay, id)` pair. The relay enforces membership and signs the
+group's state; there is no cross-relay identity, no encryption, and no portable
+"create". The same `id` on two relays is two groups (a fork), which the plan
+models as two entities.
+
+| Concern | Wire | Backend rule |
+| --- | --- | --- |
+| Group id | random string, relay-chosen | `h` tag on everything a user sends; `d` tag on relay state events |
+| Timeline | any event kind + `h` | v1 renders kinds `1`, `9`, `11`, `1111` and `7`; every other kind is cached for the `previous` window but not rendered, except the membership kinds below |
+| Timeline refs | `previous` tag, first 8 hex chars of event ids | built from the last 50 events seen on that relay (any kind — NIP-29 timeline references), excluding our own; send up to 3 |
+| Relay state | `39000` metadata, `39001` admins, `39002` members, `39003` roles, `39004` LiveKit participants | relay-signed, addressable by `d`; read, never written by the client |
+| Membership | `9021` join, `9022` leave (user-signed); relay answers with `9000` put-user / `9001` remove-user | membership = latest `9000`/`9001` naming our pubkey; no event means not a member, and `Pending`/`Refused` are tracked locally for private groups |
+| Own group list | `10009` (NIP-51 simple groups) | `["group", id, relay, name?]` plus `["r", relay]`; replaceable, published to NIP-65 relays |
+| Reference | `naddr1…` of kind `39000` with the relay self pubkey and a relay hint, optionally `?invite=<code>` | the self pubkey is the author of the fetched `39000`; no `39000` means no canonical naddr |
+
+Two behaviours that shape the code:
+
+- **Late publication is rejected.** Relays drop messages with old timestamps, so
+  there is no offline send queue in v1: a send either goes now or fails visibly.
+- **`previous` must reference events the relay already has.** Relays may enforce
+  at least 3 references; a group we have little history for may have its sends
+  rejected until the backlog loads. The relay's `OK false` message is surfaced
+  as-is.
+
+### Not in v1
+
+Moderation writes (`9000`–`9010`), group creation (`9007`, relay-dependent),
+pinned events (`39005`/`9010`), subgroups (`parent`/`child` — parsed but inert),
+LiveKit audio/video (`livekit` tag, `39004`, the token endpoint), fork/migration
+detection through admins' `10009`, and relay browsing (`{"kinds":[39000]}` to
+list a relay's public groups). All are additive; none change the model above.
+
+## What the local rust-nostr checkout gives us
+
+Cross-checked against the pinned checkout
+(`~/.cargo/git/checkouts/nostr-619b808bb247a9ed/01f1950`):
+
+- **Kinds exist** for every NIP-29 event in v1: `Kind::GroupPutUser` (9000),
+  `GroupRemoveUser` (9001), `GroupEditMetadata` (9002), `GroupDeleteEvent`
+  (9005), `GroupCreateGroup` (9007), `GroupDeleteGroup` (9008),
+  `GroupCreateInvite` (9009), `GroupJoinRequest` (9021), `GroupLeaveRequest`
+  (9022), `GroupMetadata` (39000), `GroupAdmins` (39001), `GroupMembers`
+  (39002), `GroupRoles` (39003), `GroupLivekitParticipants` (39004), and
+  `Kind::SimpleGroups` (10009). `Kind::ChatMessage` is kind 9.
+- **`9010` and `39005` are missing** from this revision (`Kind::Custom(9010)`,
+  `Kind::Custom(39005)` if they are ever needed — they are not, in v1).
+- **There is no NIP-29 module.** `nostr/src/nips/` has no `nip29.rs`, and no
+  `GroupId`, metadata type, parser or builder exists. Everything in `protocol.rs`
+  below is ours; it is thin — tags and filters, no crypto.
+- **The APIs to lean on** are all present and already used in the app:
+  - `Tag::custom`, `Tag::event`, `Tag::identifier`, `Tag::public_key`,
+    `Tag::content`, `Tag::as_slice`.
+  - `Filter::custom_tag(SingleLetterTag::LOWERCASE_H, id)`,
+    `.identifier(id)`, `.author`, `.kinds`, `.limit`, `.since`.
+  - `ReqTarget::single(url, filters)` for a group-scoped REQ,
+    `client.subscribe(..).with_id(..)`, `client.fetch_events(..)`,
+    `client.send_event(&event).to([url])`.
+  - `EventBuilder::new(kind, content).tags(tags).finalize_async(&signer)` — the
+    app's `UniversalSigner` works here (NIP-46 included), as it does in
+    `crates/chat/src/room.rs`.
+  - `client.database().query/save_event/delete`.
+  - `Nip19Coordinate { coordinate, relays }` with `ToBech32`/`FromBech32` for
+    `naddr`.
+- **NIP-11 is not exposed.** `Relay` in this SDK has no document accessor, so the
+  relay self pubkey is taken from the author of the group's `39000` event; a
+  group whose relay publishes no `39000` therefore has no canonical naddr and
+  cannot be shared or joined in v1.
+
+## What the app already does (and we copy)
+
+The NIP-17 backend is the template. `crates/chat` has a global `ChatRegistry`
+entity, one `Room` entity per conversation, a background task over
+`client.notifications()` that sends signals through a `flume` channel, and a
+foreground `cx.spawn` consumer that is the only place entities change. Local
+messages are cached as `ApplicationSpecificData` (30078) events with a `d` tag
+for the target id and an `r` tag for the room key (`set_rumor`/`get_rumor`,
+`crates/chat/src/lib.rs`). `crates/chat_ui` renders `chat::Message` into a
+`ListState`, resolving names through `PersonRegistry`.
+
+`crates/community` shows the same pattern for relay-scoped state: routing
+notifications by subscription id (`route_of`), bringing relays into the pool
+before use (`connect_relays`), and treating a `CLOSED` message as a refusal.
+Replacing a list is done in `crates/workspace/src/panels/contact_list.rs`:
+sign the builder, `client.send_event(&event).to_nip65().await` — the client saves
+the event locally even when the publish fails.
+
+NIP-29 reuses all of this and drops what only NIP-17 needs: no encryption, no
+per-member fan-out, no gift wraps, no `SendReport` per receiver — one relay, one
+event, one `OK`.
+
+## Where this lands
+
+A new crate `crates/nip29`, because NIP-29 is a different transport (plain relay
+events), a different state authority (relay-signed `39000`s) and a different UI
+later. Making `chat::Room` polymorphic over gift-wrapped DMs and relay groups
+would put two protocols in one entity. The alternative — modules inside
+`crates/chat` — saves a `Cargo.toml` but was set aside (see Decisions).
+
+| File | Owns |
+| --- | --- |
+| `crates/nip29/src/protocol.rs` | Pure NIP-29 types and their `impl` blocks: `GroupId`, `GroupKey`, `GroupReference`, `GroupMetadata`, admins/members/roles, `GroupList`, `Membership`, `TimelineWindow`. No I/O. |
+| `crates/nip29/src/group.rs` | `Group` entity: state, timeline, membership, `connect`, `ingest`, `send_message`, `send_reaction`, `join`, `leave`, `GroupEvent`. |
+| `crates/nip29/src/lib.rs` | `GroupsRegistry` global, `init`, standing subscriptions and the signal pump, the envelope cache, `load`/`publish` of the account's `10009`. |
+| `desktop/src/main.rs`, `web/src/lib.rs` | `nip29::init(cx)` after `chat::init(cx)`. |
+| UI later | `crates/chat_ui` (reuse `chat::Message` rendering) or a new `crates/nip29_ui`; sidebar section; a join dialog. |
+
+`protocol` is re-exported at the crate root (`pub use protocol::*`) so callers
+write `nip29::GroupId`, `nip29::GroupMetadata`. The crate's dependencies are the
+ones `chat` already carries: `state` (client and signer), `chat` (`Message`),
+`gpui`, `nostr-sdk`, `anyhow`, `flume`, `futures`, `smallvec`, `log`;
+dev-dependencies `nostr-memory` and `nostr-sdk` with the `local-relay` feature
+for the tests below.
+
+## Data flow
+
+```
+own 10009 (NIP-65 relays)              group relay (one per group)
+      │ load / subscribe                     │  REQ #h=<id>  (timeline + 9000/9001)
+      ▼                                      │  REQ #d=<id>  (39000..39003)
+GroupsRegistry ──ensure relay──► Group entity ◄──ingest── background listener
+      │ index: GroupKey                       │              (saves envelope first)
+      │                                       │
+      └── flume Signal ◄──────────────────────┘
+                 ▲
+     UI later: GroupPanel (Vec<Message>), JoinDialog (naddr)
+```
+
+A group's subscriptions are the standing `grp-<hash>` REQ (timeline + state +
+membership, the timeline replayed up to 100 events) and, on demand, a
+`pg-<hash>` REQ that re-asks the same timeline filter with `until` set to the
+oldest event hydrated so far. Both are recognised by the same pump and land in
+the same envelope cache, so paging needs no separate read path.
+
+## Phase 0 — `crates/nip29/src/protocol.rs` (pure, no I/O)
+
+The protocol is expressed as types with `impl` blocks: a parsed thing is a struct
+and every operation on it is a method, so call sites read `id.message(...)`,
+`metadata.naddr(...)`, `admins.contains(&pk)` instead of handing the pieces to
+free builder functions.
+
+```rust
+/// The relay-chosen group id.
+pub struct GroupId(String);
+
+impl GroupId {
+    pub fn new(id: impl Into<String>) -> Result<Self, Error>;   // trimmed, non-empty
+    pub fn as_str(&self) -> &str;
+    pub fn h_tag(&self) -> Tag;                                 // ["h", id]
+
+    pub fn timeline_filter(&self, limit: usize) -> Filter;      // #h = id
+    pub fn state_filter(&self) -> Filter;                       // #d = id, kinds 39000..39003
+    pub fn membership_filter(&self, me: PublicKey) -> Filter;   // #h = id, #p = me, kinds 9000/9001
+
+    pub fn message(&self, content: String, replies: &[EventId], previous: Option<Tag>) -> EventBuilder; // kind 9
+    pub fn reaction(&self, target: EventId, emoji: String, previous: Option<Tag>) -> EventBuilder;      // kind 7 + e
+    pub fn join_request(&self, code: Option<&str>, reason: &str) -> EventBuilder;                       // 9021
+    pub fn leave_request(&self, reason: &str) -> EventBuilder;                                          // 9022
+}
+
+/// Entity identity: the same id on another relay is another group (a fork).
+pub struct GroupKey { relay: RelayUrl, id: GroupId }
+
+impl GroupKey {
+    pub fn new(relay: RelayUrl, id: GroupId) -> Self;
+    pub fn relay(&self) -> &RelayUrl;
+    pub fn id(&self) -> &GroupId;
+    pub fn cache_tag(&self) -> String;   // "<relay>|<id>"
+    pub fn uniq_id(&self) -> u64;        // hashed, for panel and element ids
+}
+
+/// What a user pastes to join: naddr1…(+"?invite=<code>").
+pub struct GroupReference { id: GroupId, relay: RelayUrl, invite: Option<String> }
+
+impl GroupReference {
+    pub fn parse(input: &str) -> Result<Self, Error>;
+    pub fn id(&self) -> &GroupId;
+    pub fn relay(&self) -> &RelayUrl;
+    pub fn invite(&self) -> Option<&str>;
+    pub fn key(&self) -> GroupKey;
+}
+
+impl FromStr for GroupReference { .. }   // so a pasted naddr can use `.parse()` too
+
+/// kind:39000, relay-signed.
+pub struct GroupMetadata {
+    pub name: Option<String>, pub picture: Option<String>, pub banner: Option<String>,
+    pub about: Option<String>, pub private: bool, pub restricted: bool, pub hidden: bool,
+    pub closed: bool, pub livekit: bool, pub parent: Option<GroupId>,
+    pub supported_kinds: Option<Vec<Kind>>,   // None = unspecified (all); Some(vec![]) = none
+}
+
+impl GroupMetadata {
+    pub fn parse(event: &Event) -> Result<Self, Error>;
+    pub fn name(&self) -> Option<&str>;
+    pub fn supports(&self, kind: Kind) -> bool;
+    pub fn naddr(&self, relay: &RelayUrl, invite: Option<&str>) -> Result<String, Error>;
+    // Coordinate { kind: 39000, public_key: <the event author>, identifier: <the d tag> }
+}
+
+/// kind:39001 — each pubkey with its role labels.
+pub struct GroupAdmins(Vec<(PublicKey, Vec<String>)>);
+
+impl GroupAdmins {
+    pub fn parse(event: &Event) -> Result<Self, Error>;
+    pub fn entries(&self) -> &[(PublicKey, Vec<String>)];
+    pub fn contains(&self, public_key: &PublicKey) -> bool;
+}
+
+/// kind:39002 — may be absent or a subset; absence means "not published".
+pub struct GroupMembers(Vec<PublicKey>);
+
+impl GroupMembers {
+    pub fn parse(event: &Event) -> Result<Self, Error>;
+    pub fn contains(&self, public_key: &PublicKey) -> bool;
+    pub fn iter(&self) -> impl Iterator<Item = &PublicKey>;
+}
+
+/// kind:39003 — role names are relay-defined.
+pub struct GroupRoles(Vec<Role>);
+pub struct Role { pub name: String, pub description: Option<String> }
+
+impl GroupRoles {
+    pub fn parse(event: &Event) -> Result<Self, Error>;
+    pub fn roles(&self) -> &[Role];
+}
+
+/// kind:39004 — who is live in the group's AV room; `participant` tags, not `p`.
+/// Absence means nothing is known, which is not the same as an empty room.
+pub struct GroupParticipants(Vec<PublicKey>);
+
+impl GroupParticipants {
+    pub fn parse(event: &Event) -> Result<Self, Error>;
+    pub fn contains(&self, public_key: &PublicKey) -> bool;
+    pub fn iter(&self) -> impl Iterator<Item = &PublicKey>;
+}
+
+/// kind:10009 — the account's own list.
+pub struct GroupList(Vec<GroupListEntry>);
+pub struct GroupListEntry { pub id: GroupId, pub relay: RelayUrl, pub name: Option<String> }
+
+impl GroupList {
+    pub fn parse(event: &Event) -> Result<Self, Error>;
+    pub fn entries(&self) -> &[GroupListEntry];
+    pub fn upsert(&mut self, entry: GroupListEntry);
+    pub fn remove(&mut self, key: &GroupKey);
+    pub fn to_builder(&self) -> EventBuilder;   // kind 10009, group + r tags
+}
+
+pub enum Membership {
+    Unknown,
+    Pending { since: Timestamp },          // our 9021 went out; no answer yet
+    Member,
+    Refused { reason: Option<String> },    // the relay's OK false / CLOSED message
+    Removed,
+}
+
+impl Membership {
+    /// Folds 9000/9001 only; `Pending` and `Refused` are set by the join path.
+    pub fn from_events(events: &[Event], me: PublicKey) -> Self;
+}
+
+/// The newest-first window (≤50) that `previous` references are drawn from.
+pub struct TimelineWindow(Vec<Event>);
+
+impl TimelineWindow {
+    pub fn new() -> Self;
+    pub fn push(&mut self, event: Event);                       // dedups, caps at 50
+    pub fn previous_tag(&self, me: PublicKey) -> Option<Tag>;   // excludes `me`, ≤3 distinct 8-hex
+}
+```
+
+Details that must be right:
+
+- Every user event carries `["h", id]` (from `GroupId::h_tag`); `previous` is
+  added only when we have at least one eligible event, and its values are the
+  first 8 characters of the hex ids.
+- `GroupReference::parse` splits `?invite=` before bech32 decoding; a missing
+  relay hint is an error (there is nowhere to send the join request).
+- `GroupList::parse` accepts `group` tags with 1–3 values (id, relay, name).
+- `Membership::from_events` takes the newest `9000`/`9001` whose `p` tags name
+  `me`; on an exact tie it prefers `9001` (removal). `Pending` and `Refused` are
+  not things the relay folds into an event: the join path sets them from the
+  `9021` send and from the relay's refusal message.
+- `GroupMetadata::supports` distinguishes "tag absent" from "tag present with no
+  values": the first means all kinds, the second means no text messages.
+- The builders return `EventBuilder`s so the caller decides when to sign; nothing
+  in this file does I/O.
+
+## Phase 1 — read path (`group.rs` + `lib.rs`)
+
+`GroupsRegistry` (global entity, mirroring `CommunityRegistry`):
+
+```rust
+pub struct GroupsRegistry {
+    groups: Vec<Entity<Group>>,
+    index: HashMap<GroupKey, Entity<Group>>,
+    /// The account's 10009, loaded at start and mutated by join/leave
+    list: GroupList,
+    /// subscription id → group, rebuilt whenever a group resubscribes
+    routes: Arc<RwLock<HashMap<SubscriptionId, GroupKey>>>,
+    signal_tx: flume::Sender<Signal>, signal_rx: flume::Receiver<Signal>,
+    notification_listener: Option<Task<Result<()>>>,  // cancelled on signer change
+    signal_consumer: Option<Task<Result<()>>>,
+    tasks: SmallVec<[Task<Result<()>>; 2]>,
+    _subscriptions: SmallVec<[Subscription; 2]>,
+}
+
+impl GroupsRegistry {
+    pub fn global(cx: &App) -> Entity<Self>;
+    pub fn groups(&self) -> &[Entity<Group>];
+    pub fn group(&self, key: &GroupKey) -> Option<Entity<Group>>;
+    pub fn join(&mut self, reference: GroupReference, cx: &mut Context<Self>);
+    pub fn leave(&mut self, key: &GroupKey, cx: &mut Context<Self>);
+    fn load(&mut self, cx);           // own 10009 from the database into `self.list`
+    fn publish_list(&mut self, cx);   // sign `self.list.to_builder()`, to_nip65, best-effort
+}
+```
+
+- **Subscriptions.** One standing REQ per group, two filters in one id:
+  `key.id().timeline_filter(100)` and `key.id().state_filter()`. The id is
+  `grp-<8 hex of the group key hash>` (NIP-01 caps ids at 64 chars; the real key
+  lives in `routes`). The relay is added and connected first
+  (`add_relay(url).and_connect()`), like `community::sync::connect_relays`. On
+  open, also issue a one-shot `key.id().membership_filter(me)` fetch so a restart
+  learns `9000`/`9001` immediately; the state filter is fetched once (limit 1 per
+  kind) so the sidebar has names before the live REQ settles. A dead relay costs
+  only its own group.
+- **Pump.** The listener matches `ClientNotification::Message` and routes
+  `RelayMessage::Event`/`EndOfStoredEvents`/`Closed` by subscription id through
+  `routes`. For `h`-tagged events it saves the envelope first, then sends
+  `Signal::Event(key, event)`; state events send `Signal::State(key, event)`;
+  `Closed` sends `Signal::Refused(key, message)`. The consumer calls
+  `Group::ingest`, which is the only place entity state changes.
+- **Envelope cache.** Same shape as `set_rumor`/`get_rumor` in `crates/chat`:
+  `ApplicationSpecificData` (30078, addressable) with `d` = event id (dedup on
+  replay), `r` = `"<relay>|<id>"` (per-fork history), `k` = kind, content = the
+  raw event JSON, signed by a crate-local `LOCAL_KEYS`. Every `h`-tagged event
+  the standing subscription delivers is enveloped — any kind, because the
+  `previous` window is literally "the last 50 events seen" (NIP-29 timeline
+  references) — while the `d`-tagged state events (`39000`–`39003`) are not.
+  Reading is `Filter::new().custom_tag(LOWERCASE_R, key)`, parsed and sorted by
+  `created_at`, then partitioned: the render set (`1`, `7`, `9`, `11`, `1111`)
+  becomes messages, `9000`/`9001` fold membership, the rest only feed the
+  window. This is what makes the timeline fork-safe and restart-safe without
+  teaching the database about relays.
+- **`Group` state.** `GroupKey`, `metadata: Option<GroupMetadata>`,
+  `metadata_event: Option<Event>` (what `GroupMetadata::naddr` is built from),
+  `members`, `admins`, `roles`, `membership: Membership` (including the local
+  `Pending`/`Refused` access states), `messages: Vec<Message>` for the render set
+  (reusing `chat::Message`, which already parses any event into display fields),
+  an id index, a `TimelineWindow` for `previous`, `loading`, `refused`. It emits
+  `GroupEvent::{Incoming(Event), Updated, Refused(String), Eose}` — the panel
+  subscribes to these the way `ChatPanel` subscribes to `RoomEvent`.
+- **`load()`.** Subscribe to our own `10009` (`ReqTarget::auto`, limit 1), then
+  read the newest one from the database into `self.list`, and add one group per
+  entry. A newer `10009` arriving from another device is merged
+  (`GroupList::upsert`) and the new groups are subscribed; removals by another
+  client cannot be detected in a replaceable list and are not guessed at.
+
+## Phase 2 — write path
+
+All writes sign with the app signer (`finalize_async(&signer)`), send to the
+group's relay only, and treat the relay's `OK` as the truth.
+
+- **Send a message.** Build with
+  `key.id().message(content, &replies, window.previous_tag(me))`, sign, then
+  `client.send_event(&event).to([relay])`. Kind 9 is the default message kind;
+  when `metadata.supports(Kind::ChatMessage)` is false the composer is disabled
+  rather than sending a kind the relay does not take. On acceptance: save the
+  envelope, insert into `messages`, emit `GroupEvent::Incoming`. On `OK false`:
+  return the relay's message (it already carries `invalid:` / `restricted:` /
+  `duplicate:` prefixes) and let the panel show it. Reactions go through the same
+  path with `key.id().reaction(..)`.
+- **Join.** `GroupReference::parse` → error if no relay hint → add/connect the
+  relay → fetch `39000` (`limit 1`) to confirm the group and learn the self
+  pubkey; `GroupMetadata::parse` it; a hidden group may refuse this — proceed
+  without metadata rather than failing → create the entity, fetch state +
+  membership, subscribe → send `key.id().join_request(code, ..)` — the `h` tag
+  plus the `code` tag when an invite code was parsed → mark
+  `Membership::Pending { since }` and `self.list.upsert(..)`, so the group stays
+  listed while it waits. The relay either emits `9000` (member), refuses the
+  `9021` with `OK false` (set `Refused { reason }` from its message — the NIP
+  says the message explains whether the rejection is final, pending review or
+  payment-required), or refuses the timeline REQ with `restricted:` /
+  `auth-required:` in a `CLOSED` (keep the state, do not raise an error toast).
+  All three are observed on the standing subscription or the send result.
+- **Leave.** Send `key.id().leave_request(..)`; on `OK`, forget the group's
+  envelopes (`client.database().delete(filter)` by `r` tag), drop the entity and
+  `self.list.remove(key)`, then publish. The relay also emits `9001`, which the
+  (now gone) subscription would have seen.
+- **Publish `10009`.** Sign `self.list.to_builder()` and publish with
+  `to_nip65()`. The registry's `GroupList` is the one loaded at start, join
+  `upsert`s an entry and leave `remove`s one, so a rewrite can never drop a group
+  we loaded. The client saves the event locally even when the publish fails, so
+  the list survives offline.
+
+## Phase 3 — later, in rough order
+
+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. **Done** as pure
+   writes: builders on `GroupId`, senders on `Group` sharing `dispatch`.
+   `edit_metadata` takes a `GroupMetadata`, which is now `Default`-able (its
+   private `id`/`author` origin are `Option`) so the UI can start an edit from
+   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". **Done.**
+   `GroupId::timeline_page(limit, until)` is `timeline_filter` plus `.until(..)`,
+   and `Group::load_page` sends it as a `pg-<hash>` subscription (auto-close on
+   EOSE, 10s cap) alongside the standing `grp-<hash>` one, which keeps running so
+   paging cannot replace the live tail. The read path is the usual one: `pg-`
+   joins `is_own_subscription`, the id is registered in `routes` against the same
+   key, and `handle_event` therefore caches each page event and `reload`s the
+   group, so it arrives through the envelope cache and the timeline like any
+   other event. `GroupsRegistry::load_more(key, cx)` is the entry point: it does
+   nothing while a page is in flight (`Group::loading_more`) or before anything is
+   cached, and registers the route before `Group::load_page` sets the flag.
+   `handle_eose` and a page `CLOSED` remove the route and call
+   `Group::page_finished`, which clears the flag and emits `GroupEvent::Updated`
+   so a panel's spinner can settle. The cursor is `Group::oldest()`, the oldest
+   `h`-tagged event ever hydrated, updated by `track` next to `window.push` —
+   *not* read from `TimelineWindow`, which keeps only the newest 50 events, so a
+   cursor taken from it would never move and would page the same window forever.
+   `until` is inclusive, so a page repeats the event it starts from; `hydrate`
+   dedups by id, so that costs one redundant fetch, never a gap. `forget` also
+   drops and unsubscribes the `pg-` route (`reset` already sweeps `routes`). Two
+   gaps left for the UI phase: a relay that accepts the REQ but never EOSEs leaves
+   `loading_more` true (the auto-close timeout closes the subscription locally and
+   emits no notification — the same stickiness as `Browse::loading`), and a group
+   whose next page shares the `until` second cannot advance the cursor, so it can
+   re-fetch the same page. Neither loops on its own; both need a "load older
+   messages" button bound to `Group::loading_more()` to be observable at all.
+5. LiveKit AV and fork/migration detection, split in three:
+   1. **`39004` participants. Done.** `GroupParticipants` joins `protocol.rs`
+      with the same shape as `GroupMembers` but reads `participant` tags (the NIP
+      does not use `p` here); one malformed pubkey fails the whole event, as in
+      `GroupAdmins`. The kind joins `STATE_KINDS`, so the standing subscription
+      already carries it, and `absorb_state` folds it into `Group::participants()`.
+      Nothing is written by us — the relay publishes it and updates it as people
+      come and go, so it rides the live tail like any other state event.
+      "Absent" and "empty" are deliberately the same value here: unlike
+      `supported_kinds`, the NIP gives the empty list no special meaning, so the
+      UI reads "no one is live" either way.
+   2. The `/.well-known/nip29/livekit/<group-id>` token endpoint and the `204`
+      probe at `/.well-known/nip29/livekit`. **Done.** The HTTP client is the one
+      GPUI already owns: `cx.http_client()` returning `Arc<dyn HttpClient>`,
+      already wired to `reqwest_client::ReqwestClient` on desktop and to `fetch`
+      on wasm — the same path `state::nip05` uses, which is why no HTTP
+      dependency lands in this crate. Its `get()` convenience cannot set headers,
+      so the request goes through `HttpClient::send` with a `Builder`, an
+      explicit `RedirectPolicy::NoFollow` (the `Authorization` header is signed
+      for exactly one URL, so a redirect could only ever fail verification) and a
+      10s `timeout` — both are opt-in extensions, and a request without them has
+      none. The header itself is NIP-98 from the SDK: `nostr/nip98` is now
+      enabled on the workspace `nostr` entry and nip29 declares `nostr` itself,
+      so `HttpData::new(url, HttpMethod::GET).to_authorization(&signer)` builds
+      `Nostr <base64>` — the `u` + `method` tags that the crate's own
+      `verify_auth_header` demands, though the NIP's prose mentions only `u`.
+      `GroupKey::livekit_token_url` / `livekit_probe_url` derive the endpoint
+      purely: the relay URL is rewritten to http(s) (keeping host and port),
+      reset to the origin, and the path built with `pop_if_empty().push(..)` so
+      each segment is escaped — a relay with its own path prefix loses it, since
+      the NIP puts the endpoint at the absolute path. `Group::livekit_session`
+      and `Group::livekit_supported` are the two calls, both `&App` +
+      `background_spawn` like `publish`. The session is returned, not stored:
+      tokens are per-join artifacts, not group state, so the caller owns it.
+      Only `livekit_supported`'s unauthenticated GET ignores status beyond
+      `204`; the token endpoint's response shape is **not specified by the NIP**
+      (see Decisions) and no LiveKit client library is involved — the backend
+      stops at the JWT and server URL.
+   3. Fork/migration detection. **Done.** One standing `nip29-forks`
+      subscription over `{authors: <every admin of every group>, kinds: [10009]}`
+      on `ReqTarget::auto` — the admins' own group lists live on their outbox
+      relays, so this works with the group's relay down, and the admin pubkeys
+      come from `39001`, which is hydrated from the envelope cache, so it works
+      after a restart too. Each event is folded into `forks` (author → newest
+      `GroupList`, guarded on `created_at` because `10009` is replaceable), then
+      `evaluate_forks` recomputes every group. `Group::elsewhere()` is `Some`
+      when an admin of that group lists the same group id at a different relay;
+      the data cannot distinguish a move from a fork, hence the neutral name.
+      Evaluation is incremental, on each event, rather than batched on EOSE:
+      `ReqTarget::auto` fans out to several relays, so the first EOSE would
+      otherwise evaluate — and clear — against one relay's partial view.
+      Nothing is auto-joined, since the NIP asks the client to *offer*; the
+      offer is `GroupsRegistry::join_key(elsewhere, None, cx)`, which is exactly
+      "fetch from the new relay and update our own `10009`". Two triggers,
+      because the NIP's "periodically" and its "MUST when the relay is offline"
+      are different things. Reactively, the group observer calls
+      `sync_forks(false)` on `GroupEvent::Updated`, subscribing as soon as the
+      first `39001` lands and re-subscribing only when the admin set actually
+      changes — the guard matters, since `Updated` also fires for metadata,
+      pins, membership and a finished page. Periodically, a
+      `background_executor().timer` loop started in `load` pushes
+      `Signal::Forks` every 15 minutes and `sync_forks(true)` re-queries
+      regardless: that is the recovery path for a subscription that died or a
+      client that was offline, not the detection path. `reset` drops the fork
+      state and unsubscribes the REQ; a `CLOSED` on it is only logged, because
+      the tick retries and there is nothing to repair.
+
+## UI later — the hooks this backend leaves
+
+| UI action | Call |
+| --- | --- |
+| Paste an naddr and join (or retry a refused join) | `GroupReference::parse` → `GroupsRegistry::join` |
+| Open a group | `GroupsRegistry::group(&key, cx)` → panel subscribes to `GroupEvent` |
+| Timeline | `Group::messages()` (already `chat::Message`s) |
+| Load older messages | `GroupsRegistry::load_more(&key, cx)`; spinner from `Group::loading_more()` |
+| Send / react | `Group::send_message` / `send_reaction` |
+| Members, admins, roles, flags | `Group::members/admins/roles/metadata` |
+| Who is in the AV room | `Group::participants()` |
+| Join AV (get a LiveKit session) | `Group::livekit_session(cx)` → `await` → `LivekitSession { token, server_url }` |
+| Is AV available here | `Group::livekit_supported(cx)` after seeing `metadata().livekit` |
+| The group may have moved or been forked | `Group::elsewhere()` → offer `GroupsRegistry::join_key(elsewhere, None, cx)` |
+| 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
+listed alongside members; a new panel renders `Vec<Message>` with the existing
+`ListState` + `ui::message::MessageRow` and `PersonRegistry` for authors. The
+message *data* is already shared (`chat::Message`), so only `ChatPanel`'s
+row-rendering methods need lifting into `ui` (or copying) rather than the type. A
+join dialog accepts a pasted naddr (with or without `?invite=`). System
+notifications can hang off `GroupEvent::Incoming` when the window is inactive.
+
+Nothing new is needed from GPUI to support this backend. The local checkouts
+are `gpui-pre` 0.3.7 (aliased as `gpui`) and `gpui-base` 0.7.0, which `crates/ui`
+styles; the GPUI APIs the backend uses are the ones `chat` and `community`
+already use — `Entity`/`WeakEntity`, `Context::spawn`, `background_spawn`,
+`EventEmitter`, `Subscription` — plus two that `state` already uses:
+`App::http_client()` for the AV token fetch and
+`App::background_executor().timer(..)` for the fork tick. The panel's components
+(`ListState`, `MessageRow`, `Avatar`, `Button`, `Scrollbar`) already exist in the
+app.
+
+## Tests
+
+- `protocol.rs` unit tests: `GroupReference::parse` (naddr, `?invite=`, garbage,
+  missing relay hint, an naddr whose kind is not `39000`); `GroupMetadata::parse`
+  flags and `supports` with `supported_kinds` absent vs empty; admins with roles;
+  members; roles; `GroupList` round trip (`parse` → `upsert` → `to_builder`);
+  `TimelineWindow::previous_tag` (excludes self, caps at 3, 8-char prefixes,
+  any-kind window, `None` when empty); `Membership::from_events` (`9000` vs
+  `9001`, newest wins, tie prefers removal; `Pending`/`Refused` are exercised
+  through the join path); `GroupId::message` carries `h` and the reply `e` tags;
+  `GroupMetadata::naddr` round trip with and without an invite code.
+- Cache round trip with `nostr-memory` as a dev-dependency (the pattern
+  `crates/community` already uses): save two envelopes (a kind 9 and a kind
+  `9000`), read them back sorted, assert the partition — the kind 9 renders, the
+  `9000` only folds membership — re-save one and confirm no duplicate, `delete`
+  clears the group.
+- Routing: subscription id ↔ `GroupKey` mapping survives a resubscribe.
+- One integration test against the SDK's `LocalRelay`
+  (`nostr-sdk` dev-dependency with the `local-relay` feature): seed a `39000`
+  and a kind 9 with an `h` tag, run `Group::connect`, assert metadata and the
+  timeline land. Timeouts in GPUI-driven tests use
+  `cx.background_executor().timer(..)`, never `smol::Timer`, per `AGENTS.md`.
+
+## Decisions
+
+1. **Crate.** `crates/nip29` (the body above is written for it).
+2. **Timeline kinds.** Render `1`, `9`, `11`, `1111` and `7`; every other kind
+   is still cached and feeds the `previous` window, per the NIP's timeline
+   references rule.
+3. **Default message kind.** Kind 9; the composer is disabled when
+   `supported_kinds` is present and lacks it.
+4. **Pending joins.** Pending (and refused) groups are listed in the sidebar,
+   with an access state on the row.
+5. **`10009` merge.** Union-only; removals made by another client are not
+   detected and no diffing is done.
+6. **Leave and the cache.** Leaving wipes the group's envelopes.
+7. **Restricted/private groups.** A visible access state is required:
+   `Membership::{Unknown, Pending, Member, Refused, Removed}` with the relay's
+   refusal reason shown and a retry, while the client's NIP-42 authenticator
+   handles the reads.
+8. **References.** `naddr` only (`?invite=` accepted); `id@relay` is not. A group
+   whose relay publishes no `39000` cannot be referenced or joined in v1.
+9. **LiveKit response shape.** The NIP says the token endpoint returns "the
+   LiveKit JWT and the URL of the LiveKit server" but names no fields, so
+   `LivekitSession::parse` reads `token` (or `jwt`) and `url` (or `server_url`)
+   and fails loudly on anything else. This is a guess, not a spec, and it is the
+   one part of the AV path that a real relay can contradict; it is worth
+   re-checking against whichever relay is targeted before the UI lands.
+   Likewise `livekit_supported` treats "not `204`" as "no AV" rather than an
+   error, because it only feeds a button.
+10. **`Monitor`.** Not adopted. It is only reachable through
+    `ClientBuilder::monitor(Monitor::new(n))`, so nothing can read per-relay
+    status until `crates/state` opts in, and its `StatusChanged` stream does not
+    distinguish a permanent outage from a reconnect — so it would need filtering
+    before it could mean anything. The only thing it buys here is running the
+    admins' `10009` check the instant a group's relay drops, rather than at the
+    next 15-minute tick. The NIP frames that as MUST, but the substance is
+    already covered: the admins come from the cached `39001`, so the check runs
+    with the relay gone, just later. If that latency ever matters, the hook is
+    `sync_forks(true, cx)` driven by a `StatusChanged` consumer — one place.

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" }
 

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))
         })