add NIP-29 LiveKit token and fork detection

Cargo.lock

@@ -4602,9 +4602,12 @@ dependencies = [
  "anyhow",
  "chat",
  "flume 0.11.1",
+ "futures",
  "gpui-pre",
  "log",
+ "nostr",
  "nostr-sdk",
+ "serde_json",
  "smallvec",
  "state",
 ]

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

@@ -5,8 +5,11 @@ 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" }

crates/nip29/src/group.rs

@@ -1,19 +1,27 @@
 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, GroupPins, GroupRoles,
-    MEMBERSHIP_KINDS, Membership, PIN_LIST, Pin, RENDER_KINDS, STATE_KINDS, TimelineWindow,
+    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 {
@@ -37,14 +45,18 @@ pub struct Group {
     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]>,
 }
@@ -59,14 +71,18 @@ impl Group {
             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![],
         }
@@ -92,6 +108,10 @@ impl Group {
         &self.roles
     }
 
+    pub fn participants(&self) -> &GroupParticipants {
+        &self.participants
+    }
+
     pub fn pins(&self) -> &GroupPins {
         &self.pins
     }
@@ -100,6 +120,10 @@ impl Group {
         &self.membership
     }
 
+    pub fn elsewhere(&self) -> Option<&GroupKey> {
+        self.elsewhere.as_ref()
+    }
+
     pub fn messages(&self) -> &[Message] {
         &self.messages
     }
@@ -108,6 +132,14 @@ impl Group {
         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()
     }
@@ -236,6 +268,15 @@ impl Group {
         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();
@@ -269,6 +310,107 @@ impl Group {
         }));
     }
 
+    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);
@@ -366,7 +508,7 @@ impl Group {
         }
 
         if MEMBERSHIP_KINDS.contains(&event.kind) {
-            self.window.push(event.clone());
+            self.track(event);
 
             if !event.tags.public_keys().any(|key| key == me) {
                 return Absorbed::Ignored;
@@ -387,7 +529,7 @@ impl Group {
             };
         }
 
-        self.window.push(event.clone());
+        self.track(event);
 
         if RENDER_KINDS.contains(&event.kind) {
             Absorbed::Rendered {
@@ -398,6 +540,14 @@ impl Group {
         }
     }
 
+    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) {
@@ -416,6 +566,10 @@ impl Group {
                 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,

crates/nip29/src/lib.rs

@@ -19,10 +19,13 @@ 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);
@@ -50,12 +53,17 @@ 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))
 }
@@ -67,9 +75,21 @@ fn browse_subscription(relay: &RelayUrl) -> SubscriptionId {
     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.starts_with(GROUP_PREFIX) || id.starts_with(BROWSE_PREFIX)
+    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 {}
@@ -80,6 +100,8 @@ pub struct GroupsRegistry {
     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<()>>,
@@ -130,6 +152,8 @@ impl GroupsRegistry {
             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,
@@ -259,6 +283,22 @@ impl GroupsRegistry {
         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;
@@ -289,6 +329,8 @@ impl GroupsRegistry {
 
         self.groups.clear();
         self.observers.clear();
+        self.forks.clear();
+        self.fork_authors.clear();
         self.list = GroupList::default();
 
         let nostr = NostrRegistry::global(cx);
@@ -296,6 +338,7 @@ impl GroupsRegistry {
 
         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();
@@ -340,6 +383,8 @@ impl GroupsRegistry {
                 }
                 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>) {
@@ -354,6 +399,101 @@ impl GroupsRegistry {
         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;
@@ -371,6 +511,7 @@ impl GroupsRegistry {
                 }
                 GroupEvent::Updated => {
                     this.refresh_entry(group, cx);
+                    this.sync_forks(false, cx);
                 }
                 _ => {}
             }),
@@ -394,9 +535,11 @@ impl GroupsRegistry {
 
     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);
@@ -405,8 +548,10 @@ impl GroupsRegistry {
         let client = nostr.read(cx).client();
 
         self.tasks.push(cx.background_spawn(async move {
-            if let Err(error) = client.unsubscribe(&id).await {
-                log::warn!("nip29: unsubscribing {id} failed: {error}");
+            for id in [id, page] {
+                if let Err(error) = client.unsubscribe(&id).await {
+                    log::warn!("nip29: unsubscribing {id} failed: {error}");
+                }
             }
 
             let filter = Filter::new()
@@ -540,6 +685,11 @@ impl GroupsRegistry {
             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;
@@ -595,6 +745,18 @@ impl GroupsRegistry {
             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;
         };
@@ -616,10 +778,31 @@ impl GroupsRegistry {
             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;
         };
@@ -701,6 +884,9 @@ impl GroupsRegistry {
                     Signal::Closed(id, reason) => {
                         this.handle_closed(id, reason, cx);
                     }
+                    Signal::Forks => {
+                        this.sync_forks(true, cx);
+                    }
                 })
                 .ok();
             }

crates/nip29/src/protocol.rs

@@ -22,11 +22,12 @@ pub const RENDER_KINDS: [Kind; 5] = [
 pub const PIN_LIST: u16 = 39_005;
 pub const UPDATE_PIN_LIST: u16 = 9_010;
 
-pub const STATE_KINDS: [Kind; 5] = [
+pub const STATE_KINDS: [Kind; 6] = [
     Kind::GroupMetadata,
     Kind::GroupAdmins,
     Kind::GroupMembers,
     Kind::GroupRoles,
+    Kind::GroupLivekitParticipants,
     Kind::Custom(PIN_LIST),
 ];
 
@@ -71,6 +72,10 @@ impl GroupId {
             .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)
     }
@@ -246,6 +251,47 @@ impl GroupKey {
         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)]
@@ -515,6 +561,37 @@ impl GroupRoles {
     }
 }
 
+#[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),
@@ -569,6 +646,35 @@ impl GroupPins {
     }
 }
 
+#[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,

docs/nip29-backend-plan.md

@@ -19,7 +19,7 @@ models as two entities.
 | 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 | relay-signed, addressable by `d`; read, never written by the client |
+| 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 |
@@ -136,6 +136,12 @@ GroupsRegistry ──ensure relay──► Group entity ◄──ingest── ba
      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
@@ -229,6 +235,16 @@ impl GroupRoles {
     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> }
@@ -437,10 +453,98 @@ group's relay only, and treat the relay's `OK` as the truth.
    `metadata_tags` re-emits them — `9002` requires it, an edit that drops a
    child is rejected. `parent`/`children` are per-relay, so assembling the tree
    from `Group::metadata()` is UI work; there is no backend tree helper.
-4. Range backfill: page the timeline with `until` for "load more".
-5. LiveKit AV (`livekit` tag, `39004`, the `/.well-known/nip29/livekit` token
-   endpoint with a NIP-98 `27235` authorization event) and fork/migration
-   detection through the admins' `10009`.
+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
 
@@ -449,8 +553,13 @@ group's relay only, and treat the relay's `OK` as the truth.
 | 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` |
@@ -470,11 +579,14 @@ 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.5 (aliased as `gpui`) and `gpui-base` 0.7.0, which `crates/ui`
-styles; the only GPUI APIs the backend uses are the ones `chat` and `community`
+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` — and the panel's components (`ListState`,
-`MessageRow`, `Avatar`, `Button`, `Scrollbar`) already exist in the app.
+`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
 
@@ -518,3 +630,21 @@ already use — `Entity`/`WeakEntity`, `Context::spawn`, `background_spawn`,
    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.