add groups registry

crates/nip29/Cargo.toml

@@ -7,3 +7,11 @@ publish.workspace = true
 [dependencies]
 nostr-sdk.workspace = true
 anyhow.workspace = true
+
+state = { path = "../state" }
+chat = { path = "../chat" }
+
+gpui.workspace = true
+flume.workspace = true
+log.workspace = true
+smallvec.workspace = true

crates/nip29/src/group.rs

@@ -0,0 +1,323 @@
+use std::collections::{HashMap, HashSet};
+
+use anyhow::Result;
+use chat::Message;
+use gpui::{App, AppContext, Context, EventEmitter, Task};
+use nostr_sdk::prelude::*;
+use smallvec::{SmallVec, smallvec};
+use state::NostrRegistry;
+
+use crate::LOCAL_KEYS;
+use crate::protocol::{
+    GroupAdmins, GroupKey, GroupMembers, GroupMetadata, GroupRoles, MEMBERSHIP_KINDS, Membership,
+    RENDER_KINDS, STATE_KINDS, TimelineWindow,
+};
+
+const REPLAY_LIMIT: usize = 100;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Absorbed {
+    Ignored,
+    Rendered { fresh: bool },
+    Changed,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum GroupEvent {
+    Incoming(Event),
+    Updated,
+    Eose,
+    Refused(String),
+}
+
+pub struct Group {
+    key: GroupKey,
+    metadata: Option<GroupMetadata>,
+    members: GroupMembers,
+    admins: GroupAdmins,
+    roles: GroupRoles,
+    membership: Membership,
+    membership_events: Vec<Event>,
+    state_ids: HashMap<Kind, EventId>,
+    messages: Vec<Message>,
+    index: HashSet<EventId>,
+    window: TimelineWindow,
+    loading: bool,
+    refused: Option<String>,
+    tasks: SmallVec<[Task<()>; 2]>,
+}
+
+impl EventEmitter<GroupEvent> for Group {}
+
+impl Group {
+    pub(crate) fn new(key: GroupKey) -> Self {
+        Self {
+            key,
+            metadata: None,
+            members: GroupMembers::default(),
+            admins: GroupAdmins::default(),
+            roles: GroupRoles::default(),
+            membership: Membership::Unknown,
+            membership_events: Vec::new(),
+            state_ids: HashMap::new(),
+            messages: Vec::new(),
+            index: HashSet::new(),
+            window: TimelineWindow::new(),
+            loading: true,
+            refused: None,
+            tasks: smallvec![],
+        }
+    }
+
+    pub fn key(&self) -> &GroupKey {
+        &self.key
+    }
+
+    pub fn metadata(&self) -> Option<&GroupMetadata> {
+        self.metadata.as_ref()
+    }
+
+    pub fn members(&self) -> &GroupMembers {
+        &self.members
+    }
+
+    pub fn admins(&self) -> &GroupAdmins {
+        &self.admins
+    }
+
+    pub fn roles(&self) -> &GroupRoles {
+        &self.roles
+    }
+
+    pub fn membership(&self) -> &Membership {
+        &self.membership
+    }
+
+    pub fn messages(&self) -> &[Message] {
+        &self.messages
+    }
+
+    pub fn loading(&self) -> bool {
+        self.loading
+    }
+
+    pub fn refused(&self) -> Option<&str> {
+        self.refused.as_deref()
+    }
+
+    pub fn previous_tag(&self, me: PublicKey) -> Option<Tag> {
+        self.window.previous_tag(me)
+    }
+
+    pub(crate) fn connect(&mut self, cx: &mut Context<Self>) {
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+
+        let Some(me) = nostr.read(cx).current_user() else {
+            return;
+        };
+
+        let key = self.key.clone();
+        let id = crate::subscription_id(&key);
+        let relay = key.relay().clone();
+
+        self.tasks.push(cx.background_spawn(async move {
+            if let Err(error) = client.add_relay(relay.clone()).and_connect().await {
+                log::warn!("nip29: connecting {relay} failed: {error}");
+            }
+
+            let filters = vec![
+                key.id().timeline_filter(REPLAY_LIMIT),
+                key.id().state_filter(),
+                key.id().membership_filter(me),
+            ];
+
+            if let Err(error) = client
+                .subscribe(ReqTarget::single(relay, filters))
+                .with_id(id)
+                .await
+            {
+                log::warn!("nip29: subscribing to {} failed: {error}", key.id());
+            }
+        }));
+    }
+
+    pub(crate) fn save_envelope(&self, event: Event, cx: &App) -> Task<Result<()>> {
+        let cache_tag = self.key.cache_tag();
+        let nostr = NostrRegistry::global(cx).read(cx);
+        let client = nostr.client();
+
+        cx.background_spawn(async move {
+            let tags = vec![
+                Tag::identifier(event.id),
+                Tag::custom("r", [cache_tag]),
+                Tag::custom("k", [event.kind.to_string()]),
+            ];
+
+            let envelope = EventBuilder::new(Kind::ApplicationSpecificData, event.as_json())
+                .tags(tags)
+                .finalize_async(&*LOCAL_KEYS)
+                .await?;
+
+            client.database().save_event(&envelope).await?;
+
+            Ok(())
+        })
+    }
+
+    pub(crate) fn load_envelopes(&self, cx: &App) -> Task<Result<Vec<Event>>> {
+        let cache_tag = self.key.cache_tag();
+        let nostr = NostrRegistry::global(cx).read(cx);
+        let client = nostr.client();
+
+        cx.background_spawn(async move {
+            let filter = Filter::new()
+                .kind(Kind::ApplicationSpecificData)
+                .custom_tag(SingleLetterTag::LOWERCASE_R, cache_tag);
+
+            let mut events: Vec<Event> = client
+                .database()
+                .query(filter)
+                .await?
+                .into_iter()
+                .filter_map(|envelope| Event::from_json(&envelope.content).ok())
+                .collect();
+
+            events.sort_by_key(|event| event.created_at);
+
+            Ok(events)
+        })
+    }
+
+    pub(crate) fn hydrate(&mut self, events: Vec<Event>, me: PublicKey, cx: &mut Context<Self>) {
+        if events.is_empty() {
+            return;
+        }
+
+        let mut rendered = Vec::new();
+        let mut changed = false;
+
+        for event in &events {
+            match self.absorb(event, me) {
+                Absorbed::Rendered { fresh: true } => rendered.push(event.clone()),
+                Absorbed::Changed => changed = true,
+                _ => {}
+            }
+        }
+
+        for event in rendered {
+            cx.emit(GroupEvent::Incoming(event));
+        }
+
+        if changed {
+            cx.emit(GroupEvent::Updated);
+        }
+
+        cx.notify();
+    }
+
+    pub(crate) fn eose(&mut self, cx: &mut Context<Self>) {
+        self.loading = false;
+        cx.emit(GroupEvent::Eose);
+        cx.notify();
+    }
+
+    pub(crate) fn refuse(&mut self, reason: String, cx: &mut Context<Self>) {
+        self.loading = false;
+        self.refused = Some(reason.clone());
+        cx.emit(GroupEvent::Refused(reason));
+        cx.notify();
+    }
+
+    fn absorb(&mut self, event: &Event, me: PublicKey) -> Absorbed {
+        if STATE_KINDS.contains(&event.kind) {
+            return if self.absorb_state(event) {
+                Absorbed::Changed
+            } else {
+                Absorbed::Ignored
+            };
+        }
+
+        if MEMBERSHIP_KINDS.contains(&event.kind) {
+            self.window.push(event.clone());
+
+            if !event.tags.public_keys().any(|key| key == me) {
+                return Absorbed::Ignored;
+            }
+
+            if !self
+                .membership_events
+                .iter()
+                .any(|existing| existing.id == event.id)
+            {
+                self.membership_events.push(event.clone());
+            }
+
+            return if self.refresh_membership(me) {
+                Absorbed::Changed
+            } else {
+                Absorbed::Ignored
+            };
+        }
+
+        self.window.push(event.clone());
+
+        if RENDER_KINDS.contains(&event.kind) {
+            Absorbed::Rendered {
+                fresh: self.push_message(event),
+            }
+        } else {
+            Absorbed::Ignored
+        }
+    }
+
+    fn absorb_state(&mut self, event: &Event) -> bool {
+        match event.kind {
+            Kind::GroupMetadata => match GroupMetadata::parse(event) {
+                Ok(metadata) => self.metadata = Some(metadata),
+                Err(_) => return false,
+            },
+            Kind::GroupAdmins => match GroupAdmins::parse(event) {
+                Ok(admins) => self.admins = admins,
+                Err(_) => return false,
+            },
+            Kind::GroupMembers => match GroupMembers::parse(event) {
+                Ok(members) => self.members = members,
+                Err(_) => return false,
+            },
+            Kind::GroupRoles => match GroupRoles::parse(event) {
+                Ok(roles) => self.roles = roles,
+                Err(_) => return false,
+            },
+            _ => return false,
+        }
+
+        self.state_ids.insert(event.kind, event.id) != Some(event.id)
+    }
+
+    fn push_message(&mut self, event: &Event) -> bool {
+        if !self.index.insert(event.id) {
+            return false;
+        }
+
+        let message = Message::from(event);
+        let position = self
+            .messages
+            .partition_point(|existing| existing.created_at <= message.created_at);
+
+        self.messages.insert(position, message);
+
+        true
+    }
+
+    fn refresh_membership(&mut self, me: PublicKey) -> bool {
+        let derived = Membership::from_events(&self.membership_events, me);
+
+        if derived != Membership::Unknown && derived != self.membership {
+            self.membership = derived;
+            return true;
+        }
+
+        false
+    }
+}

crates/nip29/src/lib.rs

@@ -1,3 +1,380 @@
+use std::collections::{HashMap, HashSet};
+use std::sync::LazyLock;
+
+use anyhow::{Error, Result};
+use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task};
+use nostr_sdk::prelude::*;
+use smallvec::{SmallVec, smallvec};
+use state::NostrRegistry;
+
+mod group;
 mod protocol;
 
+pub use group::*;
 pub use protocol::*;
+
+pub(crate) static LOCAL_KEYS: LazyLock<Keys> = LazyLock::new(Keys::generate);
+
+const LIST_SUBSCRIPTION: &str = "nip29-list";
+const GROUP_PREFIX: &str = "grp-";
+
+pub fn init(cx: &mut App) {
+    GroupsRegistry::set_global(cx.new(GroupsRegistry::new), cx);
+}
+
+struct GlobalGroupsRegistry(Entity<GroupsRegistry>);
+
+impl Global for GlobalGroupsRegistry {}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum GroupsEvent {
+    Updated,
+    Error(String),
+}
+
+#[derive(Debug)]
+enum Signal {
+    Event(SubscriptionId, Event),
+    Eose(SubscriptionId),
+    Closed(SubscriptionId, String),
+}
+
+fn list_subscription_id() -> SubscriptionId {
+    SubscriptionId::new(LIST_SUBSCRIPTION)
+}
+
+pub(crate) fn subscription_id(key: &GroupKey) -> SubscriptionId {
+    SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32))
+}
+
+fn is_group_subscription(id: &SubscriptionId) -> bool {
+    let id = id.as_str();
+    id == LIST_SUBSCRIPTION || id.starts_with(GROUP_PREFIX)
+}
+
+impl EventEmitter<GroupsEvent> for GroupsRegistry {}
+
+pub struct GroupsRegistry {
+    groups: Vec<Entity<Group>>,
+    list: GroupList,
+    routes: HashMap<SubscriptionId, GroupKey>,
+    signal_tx: flume::Sender<Signal>,
+    signal_rx: flume::Receiver<Signal>,
+    notification_listener: Option<Task<()>>,
+    signal_consumer: Option<Task<()>>,
+    tasks: SmallVec<[Task<()>; 2]>,
+    _subscriptions: SmallVec<[Subscription; 2]>,
+}
+
+impl GroupsRegistry {
+    pub fn global(cx: &App) -> Entity<Self> {
+        cx.global::<GlobalGroupsRegistry>().0.clone()
+    }
+
+    fn set_global(state: Entity<Self>, cx: &mut App) {
+        cx.set_global(GlobalGroupsRegistry(state));
+    }
+
+    fn new(cx: &mut Context<Self>) -> Self {
+        let nostr = NostrRegistry::global(cx);
+        let this = cx.entity().downgrade();
+
+        let (tx, rx) = flume::unbounded::<Signal>();
+        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(move |cx| {
+            if let Err(error) = this.update(cx, |this, cx| {
+                this.handle_notifications(cx);
+
+                if nostr.read(cx).current_user().is_some() {
+                    this.load(cx);
+                }
+            }) {
+                log::warn!("nip29: initializing the registry failed: {error}");
+            }
+        });
+
+        Self {
+            groups: Vec::new(),
+            list: GroupList::default(),
+            routes: HashMap::new(),
+            signal_tx: tx,
+            signal_rx: rx,
+            notification_listener: None,
+            signal_consumer: None,
+            tasks: smallvec![],
+            _subscriptions: subscriptions,
+        }
+    }
+
+    pub fn groups(&self) -> &[Entity<Group>] {
+        &self.groups
+    }
+
+    pub fn group(&self, key: &GroupKey, cx: &App) -> Option<Entity<Group>> {
+        self.groups
+            .iter()
+            .find(|group| group.read(cx).key() == key)
+            .cloned()
+    }
+
+    pub fn reset(&mut self, cx: &mut Context<Self>) {
+        self.notification_listener = None;
+        self.signal_consumer = None;
+        self.tasks.clear();
+
+        let routes = std::mem::take(&mut self.routes);
+
+        self.groups.clear();
+        self.list = GroupList::default();
+
+        let client = NostrRegistry::global(cx).read(cx).client();
+
+        let mut ids: Vec<SubscriptionId> = routes.into_keys().collect();
+        ids.push(list_subscription_id());
+
+        self.tasks.push(cx.background_spawn(async move {
+            for id in ids {
+                if let Err(error) = client.unsubscribe(&id).await {
+                    log::warn!("nip29: unsubscribing {id} failed: {error}");
+                }
+            }
+        }));
+
+        cx.notify();
+    }
+
+    fn load(&mut self, cx: &mut Context<Self>) {
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+        let signer = nostr.read(cx).signer();
+
+        let task: Task<Result<GroupList, Error>> = cx.background_spawn(async move {
+            let me = signer.get_public_key_async().await?;
+            let filter = Filter::new().author(me).kind(Kind::SimpleGroups).limit(1);
+
+            client
+                .subscribe(ReqTarget::auto([filter.clone()]))
+                .with_id(list_subscription_id())
+                .await?;
+
+            match client.database().query(filter).await?.into_iter().next() {
+                Some(event) => GroupList::parse(&event),
+                None => Ok(GroupList::default()),
+            }
+        });
+
+        self.tasks
+            .push(cx.spawn(async move |this, cx| match task.await {
+                Ok(list) => {
+                    if let Err(error) = this.update(cx, |this, cx| this.merge(list, cx)) {
+                        log::warn!("nip29: applying the group list failed: {error}");
+                    }
+                }
+                Err(error) => log::warn!("nip29: loading the group list failed: {error}"),
+            }));
+    }
+
+    fn merge(&mut self, list: GroupList, cx: &mut Context<Self>) {
+        let entries = list.entries().to_vec();
+
+        for entry in entries {
+            self.list.upsert(entry.clone());
+            self.ensure(GroupKey::new(entry.relay.clone(), entry.id.clone()), cx);
+        }
+
+        cx.emit(GroupsEvent::Updated);
+        cx.notify();
+    }
+
+    fn ensure(&mut self, key: GroupKey, cx: &mut Context<Self>) {
+        if self.group(&key, cx).is_some() {
+            return;
+        }
+
+        let group = cx.new(|_| Group::new(key.clone()));
+        self.groups.push(group);
+
+        self.subscribe(key.clone(), cx);
+        self.reload(key, None, cx);
+    }
+
+    fn subscribe(&mut self, key: GroupKey, cx: &mut Context<Self>) {
+        let Some(group) = self.group(&key, cx) else {
+            return;
+        };
+
+        self.routes.insert(subscription_id(&key), key);
+
+        group.update(cx, |group, cx| {
+            group.connect(cx);
+        });
+    }
+
+    fn reload(&mut self, key: GroupKey, event: Option<Event>, cx: &mut Context<Self>) {
+        let nostr = NostrRegistry::global(cx);
+
+        let Some(group) = self.group(&key, cx) else {
+            return;
+        };
+
+        let Some(me) = nostr.read(cx).current_user() else {
+            return;
+        };
+
+        let logged = key.clone();
+        let load = group.read(cx).load_envelopes(cx);
+
+        self.tasks.push(cx.spawn(async move |_this, cx| {
+            if let Some(event) = event
+                && let Err(error) = group
+                    .read_with(cx, |group, cx| group.save_envelope(event, cx))
+                    .await
+            {
+                log::warn!("nip29: caching failed: {error}");
+                return;
+            }
+
+            match load.await {
+                Ok(events) => {
+                    group.update(cx, |group, cx| {
+                        group.hydrate(events, me, cx);
+                    });
+                }
+                Err(error) => {
+                    log::warn!("nip29: reloading {} failed: {error}", logged.id());
+                }
+            }
+        }));
+    }
+
+    fn handle_event(&mut self, id: SubscriptionId, event: Event, cx: &mut Context<Self>) {
+        if id == list_subscription_id() {
+            match GroupList::parse(&event) {
+                Ok(list) => self.merge(list, cx),
+                Err(error) => log::warn!("nip29: ignoring a malformed group list: {error}"),
+            }
+            return;
+        }
+
+        let Some(key) = self.routes.get(&id).cloned() else {
+            return;
+        };
+
+        self.reload(key, Some(event), cx);
+    }
+
+    fn handle_eose(&mut self, id: SubscriptionId, cx: &mut Context<Self>) {
+        let Some(key) = self.routes.get(&id).cloned() else {
+            return;
+        };
+        let Some(group) = self.group(&key, cx) else {
+            return;
+        };
+
+        group.update(cx, |group, cx| {
+            group.eose(cx);
+        });
+        cx.notify();
+    }
+
+    fn handle_closed(&mut self, id: SubscriptionId, reason: String, cx: &mut Context<Self>) {
+        let Some(key) = self.routes.get(&id).cloned() else {
+            return;
+        };
+        let Some(group) = self.group(&key, cx) else {
+            return;
+        };
+
+        group.update(cx, |group, cx| {
+            group.refuse(reason, cx);
+        });
+        cx.notify();
+    }
+
+    fn handle_notifications(&mut self, cx: &mut Context<Self>) {
+        self.notification_listener = None;
+        self.signal_consumer = None;
+
+        let nostr = NostrRegistry::global(cx).read(cx);
+        let client = nostr.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();
+            let mut processed = HashSet::new();
+            const MAX_PROCESSED: usize = 10_000;
+
+            while let Some(notification) = notifications.next().await {
+                let ClientNotification::Message { message, .. } = notification else {
+                    continue;
+                };
+
+                let signal = match *message {
+                    RelayMessage::Event {
+                        subscription_id,
+                        event,
+                    } => {
+                        let id = subscription_id.into_owned();
+
+                        if !is_group_subscription(&id) {
+                            None
+                        } else {
+                            if processed.len() >= MAX_PROCESSED {
+                                processed.clear();
+                            }
+
+                            processed
+                                .insert(event.id)
+                                .then(|| Signal::Event(id, event.into_owned()))
+                        }
+                    }
+                    RelayMessage::EndOfStoredEvents(id) => {
+                        let id = id.into_owned();
+                        is_group_subscription(&id).then_some(Signal::Eose(id))
+                    }
+                    RelayMessage::Closed {
+                        subscription_id,
+                        message,
+                    } => {
+                        let id = subscription_id.into_owned();
+                        is_group_subscription(&id)
+                            .then_some(Signal::Closed(id, message.into_owned()))
+                    }
+                    _ => None,
+                };
+
+                if let Some(signal) = signal {
+                    tx.send_async(signal).await.ok();
+                }
+            }
+        }));
+
+        self.signal_consumer = Some(cx.spawn(async move |this, cx| {
+            while let Ok(signal) = rx.recv_async().await {
+                this.update(cx, |this, cx| match signal {
+                    Signal::Event(id, event) => {
+                        this.handle_event(id, event, cx);
+                    }
+                    Signal::Eose(id) => {
+                        this.handle_eose(id, cx);
+                    }
+                    Signal::Closed(id, reason) => {
+                        this.handle_closed(id, reason, cx);
+                    }
+                })
+                .ok();
+            }
+        }));
+    }
+}