diff --git a/crates/nip29/src/group.rs b/crates/nip29/src/group.rs index c0f64d07..c15a8f40 100644 --- a/crates/nip29/src/group.rs +++ b/crates/nip29/src/group.rs @@ -1,4 +1,5 @@ use std::collections::{HashMap, HashSet}; +use std::time::Duration; use anyhow::{Result, anyhow, bail}; use chat::Message; @@ -14,6 +15,8 @@ use crate::protocol::{ }; const REPLAY_LIMIT: usize = 100; +const PAGE_LIMIT: usize = 50; +const PAGE_TIMEOUT: Duration = Duration::from_secs(10); #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum Absorbed { @@ -44,7 +47,9 @@ pub struct Group { messages: Vec, index: HashSet, window: TimelineWindow, + oldest: Option, loading: bool, + loading_more: bool, refused: Option, tasks: SmallVec<[Task<()>; 2]>, } @@ -66,7 +71,9 @@ impl Group { messages: Vec::new(), index: HashSet::new(), window: TimelineWindow::new(), + oldest: None, loading: true, + loading_more: false, refused: None, tasks: smallvec![], } @@ -108,6 +115,14 @@ impl Group { self.loading } + pub fn loading_more(&self) -> bool { + self.loading_more + } + + pub fn oldest(&self) -> Option { + self.oldest + } + pub fn refused(&self) -> Option<&str> { self.refused.as_deref() } @@ -269,6 +284,45 @@ impl Group { })); } + pub(crate) fn load_page(&mut self, cx: &mut Context) { + 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.loading_more = false; + cx.emit(GroupEvent::Updated); + cx.notify(); + } + pub(crate) fn save_envelope(&self, event: Event, cx: &App) -> Task> { let cache_tag = self.key.cache_tag(); let nostr = NostrRegistry::global(cx); @@ -366,7 +420,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 +441,7 @@ impl Group { }; } - self.window.push(event.clone()); + self.track(event); if RENDER_KINDS.contains(&event.kind) { Absorbed::Rendered { @@ -398,6 +452,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) { diff --git a/crates/nip29/src/lib.rs b/crates/nip29/src/lib.rs index 2585aebc..17ab70ba 100644 --- a/crates/nip29/src/lib.rs +++ b/crates/nip29/src/lib.rs @@ -21,6 +21,7 @@ pub(crate) static LOCAL_KEYS: LazyLock = LazyLock::new(Keys::generate); const LIST_SUBSCRIPTION: &str = "nip29-list"; 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); @@ -67,9 +68,20 @@ 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.starts_with(GROUP_PREFIX) + || id.starts_with(BROWSE_PREFIX) + || id.starts_with(PAGE_PREFIX) } impl EventEmitter for GroupsRegistry {} @@ -259,6 +271,22 @@ impl GroupsRegistry { self.browse.get(&browse_subscription(relay)) } + pub fn load_more(&mut self, key: &GroupKey, cx: &mut Context) { + 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) { let Some(group) = self.group(key, cx) else { return; @@ -394,9 +422,11 @@ impl GroupsRegistry { fn forget(&mut self, key: GroupKey, cx: &mut Context) { 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 +435,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() @@ -595,6 +627,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; }; @@ -620,6 +664,22 @@ impl GroupsRegistry { 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; }; diff --git a/crates/nip29/src/protocol.rs b/crates/nip29/src/protocol.rs index 2e1bbcc4..fbf82431 100644 --- a/crates/nip29/src/protocol.rs +++ b/crates/nip29/src/protocol.rs @@ -71,6 +71,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) } diff --git a/docs/nip29-backend-plan.md b/docs/nip29-backend-plan.md index be8bbee8..e592e28a 100644 --- a/docs/nip29-backend-plan.md +++ b/docs/nip29-backend-plan.md @@ -136,6 +136,12 @@ GroupsRegistry ──ensure relay──► Group entity ◄──ingest── ba UI later: GroupPanel (Vec), JoinDialog (naddr) ``` +A group's subscriptions are the standing `grp-` REQ (timeline + state + +membership, the timeline replayed up to 100 events) and, on demand, a +`pg-` 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 @@ -437,7 +443,32 @@ 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". +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-` subscription (auto-close on + EOSE, 10s cap) alongside the standing `grp-` 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 (`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`. @@ -449,6 +480,7 @@ 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` | | Membership badge (member, pending, refused, removed) | `Group::membership()` |