diff --git a/crates/community/src/lib.rs b/crates/community/src/lib.rs index 483e194c..5ce3400c 100644 --- a/crates/community/src/lib.rs +++ b/crates/community/src/lib.rs @@ -27,6 +27,7 @@ impl Global for GlobalCommunityRegistry {} #[derive(Debug, Clone, PartialEq, Eq)] enum Signal { Event(CommunityId), + List, } impl EventEmitter for CommunityRegistry {} @@ -67,6 +68,7 @@ impl CommunityRegistry { if event.signer_changed() { this.reset(cx); this.handle_notifications(cx); + this.subscribe_list(cx); this.load(cx); } })); @@ -76,6 +78,7 @@ impl CommunityRegistry { .update(cx, |this, cx| { this.handle_notifications(cx); if nostr.read(cx).current_user().is_some() { + this.subscribe_list(cx); this.load(cx); } }) @@ -163,6 +166,25 @@ impl CommunityRegistry { cx.notify(); } + /// Subscribe to the account's community list. + fn subscribe_list(&mut self, cx: &mut Context) { + let nostr = NostrRegistry::global(cx); + let signer = nostr.read(cx).signer(); + let client = nostr.read(cx).client(); + + self.tasks.push(cx.spawn(async move |this, cx| { + let self_pk = signer.get_public_key_async().await?; + + if let Err(error) = sync::subscribe_list(&client, self_pk).await { + this.update(cx, |_this, cx| { + cx.emit(CommunityEvent::Error(error.to_string())); + })?; + } + + Ok(()) + })); + } + /// Discover the account's communities in the local database. fn load(&mut self, cx: &mut Context) { let nostr = NostrRegistry::global(cx); @@ -298,6 +320,11 @@ impl CommunityRegistry { continue; }; + if sync::is_list_subscription(&subscription_id) { + tx.send_async(Signal::List).await?; + continue; + } + if event.kind != Kind::from(KIND_WRAP) { continue; } @@ -313,10 +340,12 @@ impl CommunityRegistry { })); self.signal_consumer = Some(cx.spawn(async move |this, cx| { - while let Ok(Signal::Event(id)) = rx.recv_async().await { - this.update(cx, |this, cx| this.refresh(id, cx))?; + while let Ok(signal) = rx.recv_async().await { + match signal { + Signal::Event(id) => this.update(cx, |this, cx| this.refresh(id, cx))?, + Signal::List => this.update(cx, |this, cx| this.load(cx))?, + } } - Ok(()) })); } diff --git a/crates/community/src/sync.rs b/crates/community/src/sync.rs index e50303ad..0412d7d3 100644 --- a/crates/community/src/sync.rs +++ b/crates/community/src/sync.rs @@ -121,21 +121,135 @@ where Ok(state) } -/// Discovers the current account's communities from the local database. +/// The subscription id carrying the account's own Community List. +pub const LIST_SUBSCRIPTION: &str = "concord/list"; + +pub fn list_subscription_id() -> SubscriptionId { + SubscriptionId::new(LIST_SUBSCRIPTION) +} + +pub fn is_list_subscription(id: &SubscriptionId) -> bool { + id.as_str() == LIST_SUBSCRIPTION +} + +/// Subscribes to the account's community list. +pub async fn subscribe_list(client: &Client, self_pk: PublicKey) -> Result<()> { + let id = list_subscription_id(); + client.unsubscribe(&id).await?; + + let filter = Filter::new() + .kind(Kind::Custom(KIND_COMMUNITY_LIST)) + .author(self_pk); + + let output = client + .subscribe(ReqTarget::auto(vec![filter])) + .with_id(id) + .await?; + + if !output.failed.is_empty() { + log::warn!( + "community list: {} relay(s) rejected the subscription", + output.failed.len() + ); + } + + Ok(()) +} + +/// Discovers the current account's communities: every live membership the List +/// carries, plus any locally-held membership the List does not mention. +/// +/// A held membership is dropped only when the List carries a tombstone at least +/// as new as it, because absence from the List is never a fact (§8). pub async fn load( client: &Client, signer: &UniversalSigner, self_pk: PublicKey, ) -> Result> { - let mut states = store::load_states(client).await?; + let list = match load_list(client, signer, self_pk).await? { + Some(list) => list, + None => return store::load_states(client).await, + }; - if let Some(list) = load_list(client, signer, self_pk).await? { - states.retain(|state| list.is_live(&state.id)); + let mut held: BTreeMap = store::load_states(client) + .await? + .into_iter() + .map(|state| (state.id, state)) + .collect(); + + held.retain(|id, state| !retired(&list, id, state.added_at_ms)); + + for entry in &list.entries { + if !list.is_live(&entry.community_id) { + continue; + } + + let fresh = match CommunityState::from_join_material(&entry.current, entry.added_at) { + Ok(fresh) => fresh, + Err(error) => { + log::warn!( + "ignoring unreadable community {} from the list: {error}", + entry.community_id.to_hex() + ); + continue; + } + }; + + let state = match held.remove(&entry.community_id) { + Some(materialized) => refresh(materialized, fresh), + None => fresh, + }; + + store::save_state(client, &state).await?; + held.insert(entry.community_id, state); } - Ok(states) + Ok(held.into_values().collect()) } +fn retired(list: &CommunityList, id: &CommunityId, added_at_ms: u64) -> bool { + list.tombstones + .iter() + .find(|tombstone| tombstone.community_id == *id) + .is_some_and(|tombstone| tombstone.removed_at >= added_at_ms) +} + +fn refresh(mut held: CommunityState, fresh: CommunityState) -> CommunityState { + held.owner = fresh.owner; + held.owner_salt = fresh.owner_salt; + held.community_root = fresh.community_root; + held.root_epoch = fresh.root_epoch; + held.added_at_ms = fresh.added_at_ms; + + if fresh.control_root.is_some() { + held.control_root = fresh.control_root; + } + + for (epoch, address) in fresh.control_pks { + held.control_pks.insert(epoch, address); + } + + held.relays = fresh.relays; + + for channel in fresh.channels { + match held.channels.iter_mut().find(|held| held.id == channel.id) { + Some(held) => { + held.name = channel.name; + held.epoch = channel.epoch; + + if channel.private { + held.private = true; + held.key = channel.key; + } + } + None => held.channels.push(channel), + } + } + + held +} + +/// Every fragment of the account's list in the local database, merged. async fn load_list( client: &Client, signer: &UniversalSigner, @@ -143,14 +257,40 @@ async fn load_list( ) -> Result> { let filter = Filter::new() .kind(Kind::Custom(KIND_COMMUNITY_LIST)) - .author(self_pk) - .limit(1); + .author(self_pk); - let Some(event) = client.database().query(filter).await?.into_iter().next() else { - return Ok(None); - }; + let mut newest: BTreeMap = BTreeMap::new(); - Ok(Some(cord02::list::parse_list_event(signer, &event).await?)) + for event in client.database().query(filter).await? { + let Ok(index) = cord02::list::fragment_index(&event) else { + continue; + }; + + match newest.get(&index) { + Some(existing) if existing.created_at >= event.created_at => {} + _ => { + newest.insert(index, event); + } + } + } + + let mut merged: Option = None; + + for event in newest.into_values() { + match cord02::list::parse_list_event(signer, &event).await { + Ok(list) => { + merged = Some(match merged { + Some(held) => cord02::list::merge(held, list), + None => list, + }); + } + Err(error) => { + log::warn!("ignoring unreadable community list {}: {error}", event.id); + } + } + } + + Ok(merged) } /// Rebuilds a community from the wraps already in the local database. diff --git a/docs/concord-community-discovery-plan.md b/docs/concord-community-discovery-plan.md index 2dbc0251..1bc74d0f 100644 --- a/docs/concord-community-discovery-plan.md +++ b/docs/concord-community-discovery-plan.md @@ -76,7 +76,7 @@ Two consequences for coop: per-process `LOCAL_KEYS`, and never leave the machine. No equivalent exists anywhere in the spec. It is a local cache and must never be treated as the discovery source. -2. **Discovery is: fetch my `33302` from relays → materialize a community from +2. **Discovery is: subscribe to my `33302` → materialize a community from `current` join material → subscribe to its planes → fold.** The fold produces the authoritative state; the List only supplies the keys to start. @@ -88,15 +88,15 @@ Two consequences for coop: | 2 | one event per fragment, `d` = index, `frags` declared | no `frags`, single event, `d` unused, `load_list` `.limit(1)` | | 3 | 32-byte values unpadded base64url at any depth | hex: `JoinMaterial.owner`/`control_root` (`PublicKey`/`String`), `CommunityId` serde, `ChannelGrant.key` | | 4 | `seed` omitted when equal to `current`; embedded snapshot omits `community_id`; `seed`'s cosmetic fields rewritten from `current` | both snapshots always serialized verbatim; `community_id` always present | -| 5 | fetch from relays | local database only | +| 5 | fetch from relays | local database only — **fixed in Phase C** | | 6 | materialize `CommunityState` from join material | no such path; only `CommunityState::from_genesis` | | 7 | publish the List on create/join (read-modify-write) | `build_list_event` is referenced only by tests and docs | | 8 | private channel keys ride in join material | `ChannelKeyRef` has no key field | Divergences 1–4 meant that even if the fetch existed, coop could neither read -what accordion wrote nor write something accordion could read. **Phases A and B -are done**, so 1–4 and 6 are resolved; 5, 7 and 8 remain (8 only in that private -planes are still not subscribed). +what accordion wrote nor write something accordion could read. **Phases A, B and +C are done**, so 1–6 are resolved; 7 and 8 remain (8 only in that private planes +are still not subscribed). ## Plan @@ -181,35 +181,55 @@ a granted key surviving, a public grant staying keyless). In `community`, field's type is crate-private, so the materialization and the plane derivation are each proved where they live. -### Phase C — fetch the List from relays, then load +### Phase C — the List drives `load` — DONE -`crates/community/src/sync.rs` +`crates/community/src/sync.rs`, `crates/community/src/lib.rs` -1. `load` becomes: - - resolve where to ask: the account's NIP-65 write relays (kind `10002`) plus - the pool's connected relays. If only the app's bootstrap relays are queried, - a List published by another client (e.g. accordion on `relay.damus.io` / - `nos.lol`) will simply not be found. - - `client.fetch_events(Filter::new().kind(33302).author(self_pk))` — one - filter returns every fragment. Fetched events are persisted by the client - (`nostr-sdk/src/relay/inner.rs:1291`), so the database read stays valid. - - merge fragments → `CommunityList`. - - for each entry whose `is_live(&id)`: if a state document exists, keep its - `heads` (the fold's authority) and refresh relays/keys from `current`; - otherwise `from_join_material(..)`. - - `store::save_state` each result so the next `load` is warm. -2. `load_list` keeps reading `client.database()` — after the fetch it is - populated. It must stop using `.limit(1)`. -3. Drop the `states.retain(..)` shape: the List is now the *source* of states, - not just a filter over local ones. A local state whose membership is - tombstoned is still dropped, but a List entry with no local state now - produces one. +1. `subscribe_list(client, self_pk)` subscribes to `Kind::Custom(33302)` + `author(self_pk)` under a dedicated `concord/list` subscription id, using + `ReqTarget::auto`. With gossip enabled, `auto` breaks the filter down by + author, so it queries the account's NIP-65 write relays and adds/connects + them itself — bootstrap relays alone would miss a List published elsewhere. +2. `CommunityRegistry` calls `subscribe_list` once per signer (signer change and + the initial defer). It is deliberately **not** called from `load`: + re-subscribing on every List event would re-deliver the List and loop. `reset` + does not unsubscribe it either — `subscribe_list` replaces the subscription + itself, and a `reset`-issued unsubscribe could race the replacement and cancel + discovery. +3. The notification listener routes a `concord/list` event to a new `Signal::List`, + whose consumer re-runs `load`. Community planes keep using `Signal::Event(id)`. +4. `load_list` reads every `33302` event by `self_pk` from the database, keeps the + newest event per fragment index, decrypts and `merge`s them. `.limit(1)` is gone. + An incomplete List is read normally — a missing fragment is news not yet heard. +5. `load` unions two sources: every live List entry (materialized with + `from_join_material`, or refreshed if a state document already exists) and every + held local state the List does not mention. A held membership is dropped only + when a tombstone outranks its `added_at_ms`; absence from the List is never a + fact. Each list-derived state is `save_state`d, so the next `load` is warm. +6. `refresh(held, fresh)` keeps the fold's authority (`heads`, `banned`, + `dissolved`) and the control planes it learned, and takes the List's identity, + relays, and channel keys. Channels are merged by id rather than replaced, so a + public channel the fold discovered is not shed by a List snapshot that predates + it. -Tests (no network, `nostr-memory`): a `33302` fragment written by the account is -discovered with **no** state document present; a tombstoned id is dropped; a -missing fragment leaves the rest usable. A `nostr_sdk::local_relay::LocalRelay` -(in-process relay, public in this pinned revision) can drive the real -fetch/subscribe path end to end. +**As built, deviating from the sketch above.** The plan called for +`client.fetch_events(..)`; the SDK's own recommendation is to keep the request +path on a subscription and read the database. This is safer than it sounds: a +relay's event is persisted at `nostr-sdk/src/relay/inner.rs:1291` **before** the +notification is emitted, so a subscription plus a database read loses nothing and +needs no explicit save. The subscription is set up with `ReqTarget::auto` rather +than a hand-built NIP-65 relay map, because gossip already resolves the author's +write relays and connects them on demand. + +Tests (no network): a fragment in the database with **no** state document +materializes a community and writes one; a tombstone at `u64::MAX` drops a held +membership; a two-fragment List with only fragment 0 delivered still yields its +membership; a held membership the List never mentions is kept alongside the +discovered one; `refresh` keeps `heads`/`banned`/`dissolved` and both control +planes while taking the List's keys; and the `concord/list` id is not read as a +community subscription. Fragment events are built with `build_list_event` from a +§8 JSON payload, so the test exercises the real decrypt-and-merge path without a +relay. ### Phase D — publish @@ -234,7 +254,9 @@ rows in the sidebar. This is the first time the path can be exercised at all. - `cargo test -p concord` (A, B), `cargo test -p community` (B, C, D). - `cargo clippy --workspace --all-targets`, `cargo fmt --all -- --check`. - A is provable against the spec's worked example, so it needs no relay. -- C is provable with `nostr-memory` + `LocalRelay`, so it needs no network. +- C is provable with `nostr-memory`: fragments are built with `build_list_event` + and saved as the subscription would have, then `load` reads them. No relay, + no `LocalRelay`. - E is the only step that needs real relays. ## Risks and open decisions @@ -248,8 +270,11 @@ rows in the sidebar. This is the first time the path can be exercised at all. NIP-44 plaintext, which understates that by roughly a third, so the count cap is kept as a conservative stopgap until Phase D measures the built event and fragments on write. -- **Relay selection for the fetch is the difference between finding the account's - List and not.** NIP-65 write relays + pool, or a user-visible relay setting? +- **Relay selection is the difference between finding the account's List and + not.** Resolved in Phase C by `ReqTarget::auto`, whose gossip path resolves the + filter's author to their NIP-65 write relays and connects them. A List + published only to relays with no NIP-65 entry is still unreachable; that is a + user-visible relay setting if it ever bites. - **Private channels stay unsubscribed until `planes()` derives their address from the granted key** (Phase B gave `ChannelKeyRef` a home for it, but the discovery fix does not need it). Public discovery works regardless. diff --git a/docs/concord-usage.md b/docs/concord-usage.md index 71bf2ebf..a04cb89e 100644 --- a/docs/concord-usage.md +++ b/docs/concord-usage.md @@ -455,6 +455,14 @@ single-event design — §8 has **no membership limit**, its only bound is the 65,536-byte encoded event, and the real fix is to start a new fragment on write (see `docs/concord-community-discovery-plan.md`, Phase D). +Discovery is a **subscription, not a fetch**: subscribe with +`Filter::new().kind(Kind::Custom(KIND_COMMUNITY_LIST)).author(my_pk)` and read the +fragments back out of `client.database()`. The client persists a relay's event +before it notifies, so a subscription plus a database read loses nothing and +needs no explicit save. Parse each event with `parse_list_event`, keep the newest +per `fragment_index`, and `merge` them — reading an incomplete List is safe, since +a missing fragment is only news not yet heard. + ## GPUI integration `crates/concord` stays GPUI-free; the registry and sync engine live in @@ -575,10 +583,13 @@ client.subscribe(filter).with_id(sub_id).await?; `CommunityEvent::Error` through `log::error!`, and its "New community" row opens a name prompt that calls `CommunityRegistry::create`. `create` still persists the genesis locally without publishing it to the metadata's relays. Discovery - is local-only: `load` reads the state documents already in - `client.database()` and never fetches the account's CORD-02 Community List - (`33302`) from relays, so a fresh install — or one signing in as an account - that joined elsewhere — finds nothing and never subscribes. See + subscribes to the account's CORD-02 Community List (`33302`) under the + `concord/list` subscription id and reads the fragments back out of + `client.database()` — the SDK persists a relay's event before notifying, so the + read is always current. A `concord/list` notification re-runs `load`, which + materializes a community from each live List entry (`from_join_material`) and + keeps any state document the List does not mention, so a fresh install — or one + signing in as an account that joined elsewhere — finds its communities. See `docs/concord-community-discovery-plan.md`. - **Account-key writers take any signer, not `&Keys`.** `genesis`, `ControlWriter`, the guestbook and chat `seal_rumor`s, the `list` builders, and