From f9e538257a53ea38d7d9bf9cb37f903be34ec72d Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Fri, 18 Sep 2026 20:00:12 +0700 Subject: [PATCH] update --- crates/community/src/community.rs | 122 +--------- crates/community/src/lib.rs | 318 ++++++++++++++++++++++++- crates/community/src/sync.rs | 186 +-------------- crates/concord/src/cords/cord02/mod.rs | 65 +---- crates/concord/src/store.rs | 179 +------------- 5 files changed, 352 insertions(+), 518 deletions(-) diff --git a/crates/community/src/community.rs b/crates/community/src/community.rs index 01d0611c..4dc6628d 100644 --- a/crates/community/src/community.rs +++ b/crates/community/src/community.rs @@ -1,5 +1,4 @@ use std::collections::{BTreeMap, BTreeSet}; -use std::sync::Arc; use anyhow::Result; use concord::cord02::ControlFold; @@ -7,11 +6,10 @@ use concord::store::{ChannelKeyRef, CommunityState}; use concord::{ChannelId, CommunityId, Epoch}; use gpui::{AppContext, Context, EventEmitter, Task}; use nostr_sdk::prelude::*; +use state::NostrRegistry; use crate::sync::{self, Snapshot}; -/// Everything that decides which planes a community is subscribed to. The -/// registry re-subscribes only when this changes. #[derive(Debug, Clone, PartialEq, Eq)] pub struct SubscriptionKey { control_pks: BTreeMap, @@ -31,6 +29,10 @@ impl SubscriptionKey { relays: state.relays.clone(), } } + + pub(crate) fn relays(&self) -> &[RelayUrl] { + &self.relays + } } #[derive(Debug, Clone)] @@ -43,7 +45,6 @@ pub struct Community { state: CommunityState, control: ControlFold, members: BTreeSet, - database: Arc, dirty: bool, refresh_task: Option>>, } @@ -51,12 +52,11 @@ pub struct Community { impl EventEmitter for Community {} impl Community { - pub fn new(state: CommunityState, database: Arc) -> Self { + pub fn new(state: CommunityState) -> Self { Self { state, control: ControlFold::default(), members: BTreeSet::new(), - database, dirty: false, refresh_task: None, } @@ -86,18 +86,18 @@ impl Community { SubscriptionKey::of(&self.state) } - /// Rebuilds the community from the wraps in the local database. A burst of - /// signals produces at most two folds: one running, one owed. + /// Rebuilds the community from the wraps in the local database. pub fn refresh(&mut self, cx: &mut Context) { if self.refresh_task.is_some() { self.dirty = true; return; } - let database = self.database.clone(); + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + let state = self.state.clone(); - let folded = - cx.background_spawn(async move { sync::fold(database.as_ref(), &state).await }); + let folded = cx.background_spawn(async move { sync::fold(&client, &state).await }); self.refresh_task = Some(cx.spawn(async move |this, cx| { let result = folded.await; @@ -127,103 +127,3 @@ impl Community { } } } - -#[cfg(test)] -mod tests { - use concord::cord02::GENERAL_CHANNEL; - use concord::cord02::guestbook::{build_join, build_leave, seal_rumor}; - use concord::derive::guestbook_group_key; - use concord::store::{load_state, save_state}; - use nostr_memory::MemoryDatabase; - - use super::*; - use crate::sync::fixtures::{AT_MS, community}; - - #[test] - fn genesis_folds_into_metadata_channels_and_the_owner() { - smol::block_on(async { - let owner = Keys::generate(); - let (genesis, state) = community(&owner); - let database = MemoryDatabase::unbounded(); - - for wrap in &genesis.wraps { - database.save_event(wrap).await.expect("saves wrap"); - } - save_state(&database, &state).await.expect("saves state"); - - let snapshot = sync::fold(&database, &state) - .await - .expect("folds") - .expect("genesis is a control edition"); - - let metadata = snapshot.control.community.as_ref().expect("metadata"); - assert_eq!(metadata.name, "Room"); - - let channel = snapshot - .control - .channels - .get(&genesis.channel_id) - .expect("general channel"); - assert_eq!(channel.name, GENERAL_CHANNEL); - assert!(!channel.private); - - assert_eq!(snapshot.members, BTreeSet::from([owner.public_key()])); - - let persisted = load_state(&database, &state.id) - .await - .expect("loads") - .expect("persisted"); - assert_eq!(persisted.id, state.id); - }); - } - - #[test] - fn a_join_adds_a_member_and_a_later_leave_removes_them() { - smol::block_on(async { - let owner = Keys::generate(); - let member = Keys::generate(); - let (genesis, state) = community(&owner); - let database = MemoryDatabase::unbounded(); - - for wrap in &genesis.wraps { - database.save_event(wrap).await.expect("saves wrap"); - } - save_state(&database, &state).await.expect("saves state"); - - let guestbook = guestbook_group_key(&state.community_root, &state.id, state.root_epoch) - .expect("guestbook key"); - - let join = seal_rumor( - &build_join(member.public_key(), None, AT_MS + 1_000), - &guestbook, - &member, - ) - .expect("seals join") - .0; - database.save_event(&join).await.expect("saves join"); - - let joined = sync::fold(&database, &state) - .await - .expect("folds") - .expect("control is still held"); - assert!(joined.members.contains(&member.public_key())); - assert_eq!(joined.members.len(), 2); - - let leave = seal_rumor( - &build_leave(member.public_key(), AT_MS + 2_000), - &guestbook, - &member, - ) - .expect("seals leave") - .0; - database.save_event(&leave).await.expect("saves leave"); - - let left = sync::fold(&database, &joined.state) - .await - .expect("folds") - .expect("control is still held"); - assert!(!left.members.contains(&member.public_key())); - assert_eq!(left.members, BTreeSet::from([owner.public_key()])); - }); - } -} diff --git a/crates/community/src/lib.rs b/crates/community/src/lib.rs index bb08fe8c..6d387ca8 100644 --- a/crates/community/src/lib.rs +++ b/crates/community/src/lib.rs @@ -1,4 +1,13 @@ -use gpui::{App, Window}; +use std::collections::HashMap; + +use anyhow::Result; +use concord::CommunityId; +use concord::cord01::KIND_WRAP; +use concord::store::CommunityState; +use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task, Window}; +use nostr_sdk::prelude::*; +use smallvec::{SmallVec, smallvec}; +use state::NostrRegistry; mod community; mod sync; @@ -6,4 +15,309 @@ mod sync; pub use community::*; pub use sync::*; -pub fn init(_window: &mut Window, _cx: &mut App) {} +pub fn init(window: &mut Window, cx: &mut App) { + CommunityRegistry::set_global(cx.new(|cx| CommunityRegistry::new(window, cx)), cx); +} + +struct GlobalCommunityRegistry(Entity); + +impl Global for GlobalCommunityRegistry {} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum Signal { + Event(CommunityId), +} + +impl EventEmitter for CommunityRegistry {} + +pub struct CommunityRegistry { + communities: Vec>, + index: HashMap>, + /// The plane set each community was last subscribed with + synced: HashMap, + /// One observer per tracked community, dropped on reset + observers: Vec, + signal_tx: flume::Sender, + signal_rx: flume::Receiver, + tasks: SmallVec<[Task>; 2]>, + /// Notification listener task (cancelled on signer change) + notification_listener: Option>>, + /// Signal consumer task (cancelled on signer change) + signal_consumer: Option>>, + _subscriptions: SmallVec<[Subscription; 2]>, +} + +impl CommunityRegistry { + pub fn global(cx: &App) -> Entity { + cx.global::().0.clone() + } + + fn set_global(state: Entity, cx: &mut App) { + cx.set_global(GlobalCommunityRegistry(state)); + } + + fn new(window: &mut Window, cx: &mut Context) -> Self { + let nostr = NostrRegistry::global(cx); + let (tx, rx) = flume::bounded::(256); + let mut subscriptions = smallvec![]; + + subscriptions.push(cx.subscribe(&nostr, |this, _nostr, event, cx| { + if event.signer_changed() { + this.reset(cx); + this.handle_notifications(cx); + this.load(cx); + } + })); + + cx.defer_in(window, move |this, _window, cx| { + this.handle_notifications(cx); + + if nostr.read(cx).current_user().is_some() { + this.load(cx); + } + }); + + Self { + communities: Vec::new(), + index: HashMap::new(), + synced: HashMap::new(), + observers: Vec::new(), + signal_tx: tx, + signal_rx: rx, + tasks: smallvec![], + notification_listener: None, + signal_consumer: None, + _subscriptions: subscriptions, + } + } + + pub fn communities(&self) -> &[Entity] { + &self.communities + } + + pub fn community(&self, id: &CommunityId) -> Option> { + self.index.get(id).cloned() + } + + /// Forget the current account and cancel everything in flight. + pub fn reset(&mut self, cx: &mut Context) { + self.notification_listener = None; + self.signal_consumer = None; + self.tasks.clear(); + self.observers.clear(); + + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + let ids: Vec = self.index.keys().copied().collect(); + + for id in ids { + let client = client.clone(); + let subscription = sync::subscription_id(&id); + + self.tasks.push(cx.background_spawn(async move { + client.unsubscribe(&subscription).await?; + Ok(()) + })); + } + + self.communities.clear(); + self.index.clear(); + self.synced.clear(); + cx.notify(); + } + + /// Discover the account's communities in the local database. + fn load(&mut self, cx: &mut Context) { + let nostr = NostrRegistry::global(cx); + let signer = nostr.read(cx).signer(); + let client = nostr.read(cx).client(); + + let task = cx.background_spawn(async move { + let self_pk = signer.get_public_key_async().await?; + sync::load(&client, &signer, self_pk).await + }); + + self.tasks.push(cx.spawn(async move |this, cx| { + match task.await { + Ok(states) => { + this.update(cx, |this, cx| this.track(states, cx))?; + } + Err(error) => { + this.update(cx, |_this, cx| { + cx.emit(CommunityEvent::Error(error.to_string())); + })?; + } + } + + Ok(()) + })); + } + + /// Replace the tracked communities with a freshly loaded set. + fn track(&mut self, states: Vec, cx: &mut Context) { + self.observers.clear(); + self.communities.clear(); + self.index.clear(); + self.synced.clear(); + + for state in states { + let id = state.id; + let community = cx.new(|_| Community::new(state)); + + self.observers + .push(cx.observe(&community, |this, _community, cx| { + this.sync_subscriptions(cx); + })); + self.index.insert(id, community.clone()); + self.communities.push(community); + } + + self.sync_subscriptions(cx); + + // A backlog already in the database produces no notification, so fold it once. + for community in self.communities.clone() { + community.update(cx, |community, cx| community.refresh(cx)); + } + + cx.notify(); + } + + fn refresh(&mut self, id: CommunityId, cx: &mut Context) { + let Some(community) = self.index.get(&id).cloned() else { + return; + }; + + community.update(cx, |community, cx| community.refresh(cx)); + } + + /// Re-subscribe every community whose held planes moved. + fn sync_subscriptions(&mut self, cx: &mut Context) { + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + + for community in self.communities.clone() { + let (id, key, state) = { + let community = community.read(cx); + ( + community.id(), + community.subscription_key(), + community.state().clone(), + ) + }; + + if self.synced.get(&id) == Some(&key) { + continue; + } + + let planes = match sync::planes(&state) { + Ok(planes) => planes, + Err(error) => { + cx.emit(CommunityEvent::Error(error.to_string())); + continue; + } + }; + + let subscription = sync::subscription_id(&id); + let filter = sync::subscription_filter(&planes); + let relays = key.relays().to_vec(); + self.synced.insert(id, key); + + let client = client.clone(); + self.tasks.push(cx.spawn(async move |this, cx| { + if let Err(error) = subscribe(&client, &subscription, &relays, filter).await { + this.update(cx, |_this, cx| { + cx.emit(CommunityEvent::Error(error.to_string())); + })?; + } + + Ok(()) + })); + } + } + + fn handle_notifications(&mut self, cx: &mut Context) { + self.notification_listener = None; + self.signal_consumer = None; + + let nostr = NostrRegistry::global(cx); + let client = nostr.read(cx).client(); + + let tx = self.signal_tx.clone(); + let rx = self.signal_rx.clone(); + + self.notification_listener = Some(cx.background_spawn(async move { + let mut notifications = client.notifications(); + + while let Some(notification) = notifications.next().await { + let ClientNotification::Event { + subscription_id, + event, + .. + } = notification + else { + continue; + }; + + if event.kind != Kind::from(KIND_WRAP) { + continue; + } + + let Some(id) = sync::community_of(&subscription_id) else { + continue; + }; + + tx.send_async(Signal::Event(id)).await?; + } + + Ok(()) + })); + + 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))?; + } + + Ok(()) + })); + } +} + +async fn subscribe( + client: &Client, + id: &SubscriptionId, + relays: &[RelayUrl], + filter: Filter, +) -> Result<()> { + client.unsubscribe(id).await?; + + for url in relays { + if let Err(error) = client.add_relay(url).and_connect().await { + log::warn!("community {id}: failed to add relay {url}: {error}"); + } + } + + // Concord wraps share kind 1059 with NIP-59 gift wraps, so an automatic + // target sends gossip after the plane authors as if they were DM peers. + // The community's own relays are the routing relays, so target them. + let target = if relays.is_empty() { + ReqTarget::auto(vec![filter]) + } else { + ReqTarget::manual( + relays + .iter() + .map(|url| (url.clone(), vec![filter.clone()])) + .collect::>(), + ) + }; + + let output = client.subscribe(target).with_id(id.clone()).await?; + + if !output.failed.is_empty() { + log::warn!( + "community {id}: {} relay(s) rejected the subscription", + output.failed.len() + ); + } + + Ok(()) +} diff --git a/crates/community/src/sync.rs b/crates/community/src/sync.rs index 78caf61f..bae60867 100644 --- a/crates/community/src/sync.rs +++ b/crates/community/src/sync.rs @@ -95,14 +95,14 @@ pub struct Snapshot { /// Discovers the current account's communities from the local database. pub async fn load( - database: &dyn NostrDatabase, + client: &Client, signer: &UniversalSigner, self_pk: PublicKey, ) -> Result> { let filter = Filter::new().kind(Kind::ApplicationSpecificData); let mut newest: BTreeMap = BTreeMap::new(); - for event in database.query(filter).await? { + for event in client.database().query(filter).await? { let Some(id) = state_document_of(&event) else { continue; }; @@ -124,7 +124,7 @@ pub async fn load( } } - if let Some(list) = load_list(database, signer, self_pk).await? { + if let Some(list) = load_list(client, signer, self_pk).await? { states.retain(|state| list.is_live(&state.id)); } @@ -138,7 +138,7 @@ fn state_document_of(event: &Event) -> Option { } async fn load_list( - database: &dyn NostrDatabase, + client: &Client, signer: &UniversalSigner, self_pk: PublicKey, ) -> Result> { @@ -147,7 +147,7 @@ async fn load_list( .author(self_pk) .limit(1); - let Some(event) = database.query(filter).await?.into_iter().next() else { + let Some(event) = client.database().query(filter).await?.into_iter().next() else { return Ok(None); }; @@ -157,17 +157,18 @@ async fn load_list( } /// Rebuilds a community from the wraps already in the local database. -pub async fn fold( - database: &dyn NostrDatabase, - state: &CommunityState, -) -> Result> { +pub async fn fold(client: &Client, state: &CommunityState) -> Result> { let planes = planes(state)?; if planes.is_empty() { return Ok(None); } - let wraps = database.query(subscription_filter(&planes)).await?; + let wraps = client + .database() + .query(subscription_filter(&planes)) + .await?; + let mut editions = Vec::new(); let mut observed: BTreeMap = BTreeMap::new(); let mut guestbook_rumors = Vec::new(); @@ -194,7 +195,7 @@ pub async fn fold( if let Ok((opened, rumor)) = concord::cord03::open(wrap, &plane.group, &channel, epoch) { - store::cache_rumor(database, &channel, &opened).await?; + store::cache_rumor(client, &channel, &opened).await?; observe(&mut observed, rumor.author, rumor.at_ms); } } @@ -243,7 +244,7 @@ pub async fn fold( let mut state = state.clone(); state.apply_fold(&control); - store::save_state(database, &state).await?; + store::save_state(client, &state).await?; Ok(Some(Snapshot { state, @@ -258,164 +259,3 @@ fn observe(observed: &mut BTreeMap, author: PublicKey, at_ms: u6 .and_modify(|seen| *seen = (*seen).max(at_ms)) .or_insert(at_ms); } - -#[cfg(test)] -pub(crate) mod fixtures { - use concord::cord02::{CommunityGenesis, CommunityMetadata, ROOT_EPOCH}; - use concord::cord04::ParsedEdition; - use concord::derive::control_signer_group_key; - - use super::*; - - pub const AT_MS: u64 = 1_719_800_000_000; - - /// A genesis and the state it folds into, ready for a test database. - pub fn community(owner: &Keys) -> (CommunityGenesis, CommunityState) { - let metadata = CommunityMetadata { - name: "Room".to_owned(), - ..Default::default() - }; - let genesis = cord02::genesis(owner, &metadata, AT_MS / 1000).expect("genesis"); - let read = control_group_key( - &genesis.community_root, - &genesis.identity.community_id, - ROOT_EPOCH, - ) - .expect("read key"); - let address = control_signer_group_key( - &genesis.control_root, - &genesis.identity.community_id, - ROOT_EPOCH, - ) - .expect("signer key") - .pk(); - - let editions: Vec = genesis - .wraps - .iter() - .map(|wrap| cord02::open_edition(wrap, &read, &address, true).expect("opens")) - .collect(); - - let state = CommunityState::from_genesis(&genesis, &editions, AT_MS).expect("state"); - - (genesis, state) - } -} - -#[cfg(test)] -mod tests { - use concord::cord02::ROOT_EPOCH; - use concord::cord02::list::{CommunityListEntry, JoinMaterial, Tombstone, build_list_event}; - use concord::store::save_state; - use nostr_memory::MemoryDatabase; - - use super::fixtures::{AT_MS, community}; - use super::*; - - fn material(state: &CommunityState) -> JoinMaterial { - JoinMaterial { - community_id: state.id, - owner: state.owner, - owner_salt: "00".repeat(32), - community_root: "11".repeat(32), - root_epoch: ROOT_EPOCH, - control_pk: None, - control_root: None, - channels: Vec::new(), - relays: Vec::new(), - name: "Room".to_owned(), - extra: Default::default(), - } - } - - fn entry(state: &CommunityState, added_at: u64) -> CommunityListEntry { - let material = material(state); - - CommunityListEntry { - community_id: state.id, - seed: material.clone(), - current: material, - added_at, - extra: Default::default(), - } - } - - #[test] - fn every_held_plane_routes_by_its_wrap_author() { - let owner = Keys::generate(); - let state = community(&owner).1; - let planes = planes(&state).expect("planes"); - - assert_eq!( - planes.len(), - 3, - "the control epoch, the guestbook and #general" - ); - - let filter = subscription_filter(&planes); - let expected: BTreeSet = planes.iter().map(|plane| plane.address).collect(); - assert_eq!(filter.authors, Some(expected)); - assert_eq!(filter.kinds, Some(BTreeSet::from([Kind::from(KIND_WRAP)]))); - - assert_eq!(community_of(&subscription_id(&state.id)), Some(state.id)); - assert_eq!(community_of(&SubscriptionId::new("device-giftwrap")), None); - } - - #[test] - fn loading_scans_state_documents_and_honours_the_list() { - smol::block_on(async { - let keys = Keys::generate(); - let signer = UniversalSigner::new(keys.clone()); - let owner = Keys::generate(); - let state = community(&owner).1; - - // With no list event, every state document is a community. - let no_list = MemoryDatabase::unbounded(); - save_state(&no_list, &state).await.expect("saves"); - let loaded = load(&no_list, &signer, keys.public_key()) - .await - .expect("loads"); - assert_eq!(loaded.len(), 1); - assert_eq!(loaded[0].id, state.id); - - // A live entry keeps it. - let event = build_list_event( - &keys, - &CommunityList { - entries: vec![entry(&state, AT_MS)], - ..Default::default() - }, - ) - .expect("builds"); - let live = MemoryDatabase::unbounded(); - save_state(&live, &state).await.expect("saves"); - live.save_event(&event).await.expect("saves list"); - let loaded = load(&live, &signer, keys.public_key()) - .await - .expect("loads"); - assert_eq!(loaded.len(), 1); - - // A newer tombstone than the entry retires it. - let event = build_list_event( - &keys, - &CommunityList { - entries: vec![entry(&state, AT_MS)], - tombstones: vec![Tombstone { - community_id: state.id, - removed_at: AT_MS + 1, - extra: Default::default(), - }], - ..Default::default() - }, - ) - .expect("builds"); - let retired = MemoryDatabase::unbounded(); - save_state(&retired, &state).await.expect("saves"); - retired.save_event(&event).await.expect("saves list"); - let loaded = load(&retired, &signer, keys.public_key()) - .await - .expect("loads"); - assert!(loaded.is_empty()); - }); - } -} diff --git a/crates/concord/src/cords/cord02/mod.rs b/crates/concord/src/cords/cord02/mod.rs index 36ea58e4..9c8fb876 100644 --- a/crates/concord/src/cords/cord02/mod.rs +++ b/crates/concord/src/cords/cord02/mod.rs @@ -731,15 +731,12 @@ fn seal_edition( #[cfg(test)] mod tests { - use nostr_memory::MemoryDatabase; - use super::*; use crate::cord03::{self, build_message, seal_rumor}; - use crate::cord04::fold; use crate::cord04::pins; use crate::cord04::roles::{Grant, MAX_BANLIST, MAX_ROLES_PER_MEMBER, Role, RoleScope}; use crate::derive::{channel_group_key, grant_locator}; - use crate::store::{CommunityState, load_state, save_state}; + use crate::store::CommunityState; use crate::{Extra, RoleId}; const AT: u64 = 1_700_000_000; @@ -768,66 +765,6 @@ mod tests { } } - #[test] - fn genesis_reopens_for_a_second_holder() { - let owner = Keys::generate(); - let community_metadata = CommunityMetadata { - name: "coop".to_owned(), - relays: vec!["wss://relay.example".to_owned()], - ..CommunityMetadata::default() - }; - - let minted = genesis(&owner, &community_metadata, AT).expect("mints"); - assert!(minted.identity.verify(), "identity is self-certifying"); - - // Only what an invite hands over: the roots, the community id and the owner salt. - let (read, signer) = holder(&minted); - let editions = open_all(&minted.wraps, &read, &signer.pk()); - - assert_eq!(editions.len(), 2); - - let community = &editions[0]; - assert_eq!(community.subkind, vsk::COMMUNITY_METADATA); - assert_eq!(community.entity, *minted.identity.community_id.as_bytes()); - assert_eq!(community.author, owner.public_key()); - assert_eq!((community.version, community.prev), (1, None)); - assert_eq!( - serde_json::from_str::(&community.content) - .expect("parses") - .name, - "coop" - ); - - let channel = &editions[1]; - assert_eq!(channel.subkind, vsk::CHANNEL_METADATA); - assert_eq!(channel.entity, *minted.channel_id.as_bytes()); - - for edition in &editions { - let folded = fold(&[EditionMeta::from(edition)], 0, None); - assert_eq!(folded.head, Some(0)); - assert!( - folded.anchored && !folded.gap, - "genesis anchors at its floor" - ); - } - - let state = CommunityState::from_genesis(&minted, &editions, AT * 1_000).expect("projects"); - - smol::block_on(async { - let database = MemoryDatabase::unbounded(); - save_state(&database, &state).await.expect("saves"); - let loaded = load_state(&database, &minted.identity.community_id) - .await - .expect("loads") - .expect("present"); - - assert_eq!(loaded.community_root, minted.community_root); - assert_eq!(loaded.control_root, Some(minted.control_root)); - assert_eq!(loaded.channels.len(), 1); - assert_eq!(loaded.heads.len(), 2); - }); - } - #[test] fn metadata_and_channel_edits_reach_a_second_client() { let owner = Keys::generate(); diff --git a/crates/concord/src/store.rs b/crates/concord/src/store.rs index 38accb13..abf0b613 100644 --- a/crates/concord/src/store.rs +++ b/crates/concord/src/store.rs @@ -27,10 +27,12 @@ const STATE_PREFIX: &str = "concord/"; /// An already-expired rumor is refused at ingest. Returns whether it was kept. pub async fn cache_rumor( - database: &dyn NostrDatabase, + client: &Client, channel: &ChannelId, opened: &OpenedStream, ) -> Result { + let at = Timestamp::from_secs(opened.at_ms / 1000); + if cord03::expiration_of(&opened.rumor)? .is_some_and(|expiration| expiration <= Timestamp::now()) { @@ -45,23 +47,19 @@ pub async fn cache_rumor( Tag::custom(CHANNEL_TAG.as_str(), [channel.to_hex()]), Tag::public_key(opened.author), ]; - let at = Timestamp::from_secs(opened.at_ms / 1000); + let event = EventBuilder::new(Kind::ApplicationSpecificData, opened.rumor.as_json()) .tags(tags) .custom_created_at(at) .finalize_async(&*LOCAL_KEYS) .await?; - database.save_event(&event).await?; + client.database().save_event(&event).await?; Ok(true) } -pub async fn purge_expired( - database: &dyn NostrDatabase, - channel: &ChannelId, - now: Timestamp, -) -> Result { +pub async fn purge_expired(client: &Client, channel: &ChannelId, now: Timestamp) -> Result { let filter = Filter::new() .kind(Kind::ApplicationSpecificData) .custom_tag(MARK_TAG, MARK_VALUE) @@ -69,7 +67,7 @@ pub async fn purge_expired( let mut expired = Vec::new(); - for event in database.query(filter).await? { + for event in client.database().query(filter).await? { let Ok(rumor) = UnsignedEvent::from_json(&event.content) else { continue; }; @@ -86,7 +84,7 @@ pub async fn purge_expired( let purged = expired.len(); if purged > 0 { - database.delete(Filter::new().ids(expired)).await?; + client.database().delete(Filter::new().ids(expired)).await?; } Ok(purged) @@ -288,16 +286,13 @@ fn state_identifier(id: &CommunityId) -> String { format!("{STATE_PREFIX}{}", id.to_hex()) } -pub async fn save_state(database: &D, state: &CommunityState) -> Result<()> -where - D: NostrDatabase + ?Sized, -{ +pub async fn save_state(client: &Client, state: &CommunityState) -> Result<()> { let event = EventBuilder::new(Kind::ApplicationSpecificData, serde_json::to_string(state)?) .tags([Tag::identifier(state.identifier())]) .finalize_async(&*LOCAL_KEYS) .await?; - database.save_event(&event).await?; + client.database().save_event(&event).await?; Ok(()) } @@ -319,7 +314,6 @@ where pub async fn backfill( client: &Client, - database: &dyn NostrDatabase, channel: &ChannelId, held: &[(Epoch, [u8; 32])], until: Option, @@ -342,7 +336,7 @@ pub async fn backfill( let (fresh, next) = advance(&page, &planes, channel, cursor, limit, &mut seen); for (opened, rumor) in fresh { - if cache_rumor(database, channel, &opened).await? { + if cache_rumor(client, channel, &opened).await? { found.push(rumor); } } @@ -416,13 +410,8 @@ async fn fetch_page( #[cfg(test)] mod tests { - use nostr_memory::MemoryDatabase; - use super::*; use crate::Epoch; - use crate::cord01::{ - KIND_WRAP, SealForm, build_rumor_ms, build_seal, channel_binding_tags, open_wrap, wrap_seal, - }; use crate::cord03::{build_message, seal_rumor}; use crate::derive::channel_group_key; @@ -499,150 +488,4 @@ mod tests { ["after the rekey", "still before", "before the rekey"] ); } - - #[test] - fn rumors_read_back_after_a_restart() { - let database = MemoryDatabase::unbounded(); - let channel = ChannelId::from_bytes([0xabu8; 32]); - let author = Keys::generate(); - - smol::block_on(async { - let group = channel_group_key(&SECRET, &channel, Epoch(0)).expect("derives"); - - for (content, at_ms) in [("first", 1_000_000u64), ("second", 2_000_000)] { - let rumor = build_rumor_ms( - 9, - author.public_key(), - content, - channel_binding_tags(&channel, Epoch(0)), - at_ms, - ); - let seal = build_seal(&rumor, SealForm::Encrypted, &group, &author).expect("seals"); - let (wrap, _) = wrap_seal( - &seal, - &group, - KIND_WRAP, - Timestamp::from_secs(at_ms / 1000), - &[], - ) - .expect("wraps"); - - let opened = open_wrap(&wrap, &group).expect("opens"); - cache_rumor(&database, &channel, &opened) - .await - .expect("caches"); - } - - // The group key is gone; only the local cache stands in for it. - let rumors = query_rumors(&database, &channel, None, 10) - .await - .expect("queries"); - assert_eq!(rumors.len(), 2, "both messages come back"); - assert_eq!(rumors[0].content, "second", "newest first"); - assert_eq!(rumors[1].content, "first"); - - // A page boundary in message time, not in cache time. - let until = Timestamp::from_secs(1_500); - let page = query_rumors(&database, &channel, Some(until), 10) - .await - .expect("queries"); - assert_eq!(page.len(), 1); - assert_eq!(page[0].content, "first"); - - let capped = query_rumors(&database, &channel, None, 1) - .await - .expect("queries"); - assert_eq!(capped.len(), 1); - assert_eq!(capped[0].content, "second"); - }); - } - - #[test] - fn an_expired_rumor_is_refused_at_ingest_and_purged_by_the_sweep() { - let database = MemoryDatabase::unbounded(); - let channel = ChannelId::from_bytes([0x77u8; 32]); - let author = Keys::generate(); - let group = channel_group_key(&SECRET, &channel, Epoch(0)).expect("derives"); - let now = Timestamp::now().as_secs(); - - smol::block_on(async { - // A live timer is stored; one that already elapsed is refused at ingest. - assert!( - cache( - &database, - &group, - &channel, - &author, - "live", - Some(3_600), - now - ) - .await - ); - assert!( - !cache( - &database, - &group, - &channel, - &author, - "gone", - Some(1), - now - 120 - ) - .await - ); - - let stored = query_rumors(&database, &channel, None, 10) - .await - .expect("queries"); - assert_eq!(stored.len(), 1); - assert_eq!(stored[0].content, "live"); - - // Hiding is not disappearing: the sweep removes the row itself, - // judged on the rumor's own signed tag. - let purged = purge_expired(&database, &channel, Timestamp::from_secs(now + 7_200)) - .await - .expect("sweeps"); - assert_eq!(purged, 1); - assert!( - query_rumors(&database, &channel, None, 10) - .await - .expect("queries") - .is_empty() - ); - - // An untimed rumor is never swept, whatever the clock says. - assert!(cache(&database, &group, &channel, &author, "timeless", None, now).await); - let purged = purge_expired(&database, &channel, Timestamp::from_secs(now + 86_400)) - .await - .expect("sweeps"); - assert_eq!(purged, 0); - }); - } - - async fn cache( - database: &MemoryDatabase, - group: &GroupKey, - channel: &ChannelId, - author: &Keys, - content: &str, - timer: Option, - at_secs: u64, - ) -> bool { - let rumor = build_message( - author.public_key(), - channel, - Epoch(0), - content, - None, - at_secs * 1_000, - timer, - ); - let (wrap, _) = seal_rumor(&rumor, group, author, false).expect("seals"); - let opened = open_wrap(&wrap, group).expect("opens"); - - cache_rumor(database, channel, &opened) - .await - .expect("caches") - } }