update

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<u64, PublicKey>,
@@ -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<PublicKey>,
-    database: Arc<dyn NostrDatabase>,
     dirty: bool,
     refresh_task: Option<Task<Result<()>>>,
 }
@@ -51,12 +52,11 @@ pub struct Community {
 impl EventEmitter<CommunityEvent> for Community {}
 
 impl Community {
-    pub fn new(state: CommunityState, database: Arc<dyn NostrDatabase>) -> 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<Self>) {
         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()]));
-        });
-    }
-}

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<CommunityRegistry>);
+
+impl Global for GlobalCommunityRegistry {}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum Signal {
+    Event(CommunityId),
+}
+
+impl EventEmitter<CommunityEvent> for CommunityRegistry {}
+
+pub struct CommunityRegistry {
+    communities: Vec<Entity<Community>>,
+    index: HashMap<CommunityId, Entity<Community>>,
+    /// The plane set each community was last subscribed with
+    synced: HashMap<CommunityId, SubscriptionKey>,
+    /// One observer per tracked community, dropped on reset
+    observers: Vec<Subscription>,
+    signal_tx: flume::Sender<Signal>,
+    signal_rx: flume::Receiver<Signal>,
+    tasks: SmallVec<[Task<Result<()>>; 2]>,
+    /// Notification listener task (cancelled on signer change)
+    notification_listener: Option<Task<Result<()>>>,
+    /// Signal consumer task (cancelled on signer change)
+    signal_consumer: Option<Task<Result<()>>>,
+    _subscriptions: SmallVec<[Subscription; 2]>,
+}
+
+impl CommunityRegistry {
+    pub fn global(cx: &App) -> Entity<Self> {
+        cx.global::<GlobalCommunityRegistry>().0.clone()
+    }
+
+    fn set_global(state: Entity<Self>, cx: &mut App) {
+        cx.set_global(GlobalCommunityRegistry(state));
+    }
+
+    fn new(window: &mut Window, cx: &mut Context<Self>) -> Self {
+        let nostr = NostrRegistry::global(cx);
+        let (tx, rx) = flume::bounded::<Signal>(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<Community>] {
+        &self.communities
+    }
+
+    pub fn community(&self, id: &CommunityId) -> Option<Entity<Community>> {
+        self.index.get(id).cloned()
+    }
+
+    /// Forget the current account and cancel everything in flight.
+    pub fn reset(&mut self, cx: &mut Context<Self>) {
+        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<CommunityId> = 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<Self>) {
+        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<CommunityState>, cx: &mut Context<Self>) {
+        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<Self>) {
+        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<Self>) {
+        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>) {
+        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::<Vec<_>>(),
+        )
+    };
+
+    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(())
+}

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<Vec<CommunityState>> {
     let filter = Filter::new().kind(Kind::ApplicationSpecificData);
     let mut newest: BTreeMap<CommunityId, Event> = 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<CommunityId> {
 }
 
 async fn load_list(
-    database: &dyn NostrDatabase,
+    client: &Client,
     signer: &UniversalSigner,
     self_pk: PublicKey,
 ) -> Result<Option<CommunityList>> {
@@ -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<Option<Snapshot>> {
+pub async fn fold(client: &Client, state: &CommunityState) -> Result<Option<Snapshot>> {
     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<PublicKey, u64> = 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<PublicKey, u64>, 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<ParsedEdition> = 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<PublicKey> = 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());
-        });
-    }
-}