diff --git a/Cargo.lock b/Cargo.lock index f1e2cb2c..0f1b9e38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4602,9 +4602,12 @@ dependencies = [ "anyhow", "chat", "flume 0.11.1", + "futures", "gpui-pre", "log", + "nostr", "nostr-sdk", + "serde_json", "smallvec", "state", ] diff --git a/Cargo.toml b/Cargo.toml index 50839b5b..8ca3bc30 100644 --- a/Cargo.toml +++ b/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" diff --git a/crates/nip29/Cargo.toml b/crates/nip29/Cargo.toml index b69b7874..05fbb7c1 100644 --- a/crates/nip29/Cargo.toml +++ b/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" } diff --git a/crates/nip29/src/group.rs b/crates/nip29/src/group.rs index f1adc9af..2a41e138 100644 --- a/crates/nip29/src/group.rs +++ b/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, + elsewhere: Option, state_ids: HashMap, messages: Vec, index: HashSet, @@ -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, cx: &mut Context) { + if self.elsewhere == elsewhere { + return; + } + self.elsewhere = elsewhere; + cx.emit(GroupEvent::Updated); + cx.notify(); + } + pub(crate) fn connect(&mut self, cx: &mut Context) { 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> { + 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> { + 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> { let cache_tag = self.key.cache_tag(); let nostr = NostrRegistry::global(cx); diff --git a/crates/nip29/src/lib.rs b/crates/nip29/src/lib.rs index 17ab70ba..be42bdeb 100644 --- a/crates/nip29/src/lib.rs +++ b/crates/nip29/src/lib.rs @@ -19,11 +19,13 @@ pub use protocol::*; pub(crate) static LOCAL_KEYS: LazyLock = 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, routes: HashMap, observers: HashMap, + forks: HashMap, + fork_authors: HashSet, signal_tx: flume::Sender, signal_rx: flume::Receiver, notification_listener: Option>, @@ -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 = 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) { @@ -382,6 +399,101 @@ impl GroupsRegistry { cx.notify(); } + fn watch_forks(&mut self, cx: &mut Context) { + 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 { + 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) { + 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) { + 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) { + 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) { 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(); } diff --git a/crates/nip29/src/protocol.rs b/crates/nip29/src/protocol.rs index c6c3c41a..a1f8bf2a 100644 --- a/crates/nip29/src/protocol.rs +++ b/crates/nip29/src/protocol.rs @@ -251,6 +251,47 @@ impl GroupKey { self.hash(&mut hasher); hasher.finish() } + + pub fn livekit_token_url(&self) -> Result { + self.livekit_url(Some(self.id.as_str())) + } + + pub fn livekit_probe_url(&self) -> Result { + self.livekit_url(None) + } + + fn livekit_url(&self, group: Option<&str>) -> Result { + 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 { + 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, diff --git a/docs/nip29-backend-plan.md b/docs/nip29-backend-plan.md index 7275e067..77107bcc 100644 --- a/docs/nip29-backend-plan.md +++ b/docs/nip29-backend-plan.md @@ -490,18 +490,61 @@ group's relay only, and treat the relay's `OK` as the truth. "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/` token endpoint: a NIP-98 - `27235` event in `Authorization`, returning the LiveKit JWT and server URL, - plus the `204` probe at `/.well-known/nip29/livekit` that tells the UI - whether a relay supports AV at all. Not started. This is the only part that - needs an HTTP client in the crate, so it is where the dependency decision - (if any) lands; `livekit` on `39000` is already parsed but currently inert. - 3. Fork/migration detection: read the `10009` of the group's admins (pubkeys - from `39001`, cached locally) and compare the relay hint against the relay - we are talking to; a different hint means the group may have moved or been - forked, which is the client's cue to offer fetching from the new relay. - Not started; the periodic — and MUST-when-offline — trigger is what - `Monitor` would feed, so the two decisions go together. + 2. The `/.well-known/nip29/livekit/` 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`, + 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 ` — 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: , 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 @@ -514,6 +557,9 @@ group's relay only, and treat the relay's `OK` as the truth. | 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` | @@ -533,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 @@ -581,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.