add NIP-29 LiveKit token and fork detection

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

@@ -3,7 +3,10 @@ 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;
@@ -11,13 +14,14 @@ use state::NostrRegistry;
 use crate::LOCAL_KEYS;
 use crate::protocol::{
     GroupAdmins, GroupId, GroupKey, GroupMembers, GroupMetadata, GroupParticipants, GroupPins,
-    GroupRoles, MEMBERSHIP_KINDS, Membership, PIN_LIST, Pin, RENDER_KINDS, STATE_KINDS,
-    TimelineWindow,
+    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 {
@@ -45,6 +49,7 @@ pub struct Group {
     pins: GroupPins,
     membership: Membership,
     membership_events: Vec<Event>,
+    elsewhere: Option<GroupKey>,
     state_ids: HashMap<Kind, EventId>,
     messages: Vec<Message>,
     index: HashSet<EventId>,
@@ -70,6 +75,7 @@ impl Group {
             pins: GroupPins::default(),
             membership: Membership::Unknown,
             membership_events: Vec::new(),
+            elsewhere: None,
             state_ids: HashMap::new(),
             messages: Vec::new(),
             index: HashSet::new(),
@@ -114,6 +120,10 @@ impl Group {
         &self.membership
     }
 
+    pub fn elsewhere(&self) -> Option<&GroupKey> {
+        self.elsewhere.as_ref()
+    }
+
     pub fn messages(&self) -> &[Message] {
         &self.messages
     }
@@ -258,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();
@@ -330,6 +349,68 @@ impl Group {
         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);

crates/nip29/src/lib.rs

@@ -19,11 +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);
@@ -51,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))
 }
@@ -79,6 +86,7 @@ fn is_page(id: &SubscriptionId) -> bool {
 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)
@@ -92,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<()>>,
@@ -142,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,
@@ -317,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);
@@ -324,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();
@@ -368,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>) {
@@ -382,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;
@@ -399,6 +511,7 @@ impl GroupsRegistry {
                 }
                 GroupEvent::Updated => {
                     this.refresh_entry(group, cx);
+                    this.sync_forks(false, cx);
                 }
                 _ => {}
             }),
@@ -572,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;
@@ -660,6 +778,11 @@ 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;
         };
@@ -761,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

@@ -251,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)]
@@ -605,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,