add send and publish

crates/nip29/src/lib.rs

@@ -43,7 +43,7 @@ fn list_subscription_id() -> SubscriptionId {
     SubscriptionId::new(LIST_SUBSCRIPTION)
 }
 
-pub(crate) fn subscription_id(key: &GroupKey) -> SubscriptionId {
+fn subscription_id(key: &GroupKey) -> SubscriptionId {
     SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32))
 }
 
@@ -58,6 +58,7 @@ pub struct GroupsRegistry {
     groups: Vec<Entity<Group>>,
     list: GroupList,
     routes: HashMap<SubscriptionId, GroupKey>,
+    observers: HashMap<GroupKey, Subscription>,
     signal_tx: flume::Sender<Signal>,
     signal_rx: flume::Receiver<Signal>,
     notification_listener: Option<Task<()>>,
@@ -106,6 +107,7 @@ impl GroupsRegistry {
             groups: Vec::new(),
             list: GroupList::default(),
             routes: HashMap::new(),
+            observers: HashMap::new(),
             signal_tx: tx,
             signal_rx: rx,
             notification_listener: None,
@@ -126,6 +128,67 @@ impl GroupsRegistry {
             .cloned()
     }
 
+    pub fn join(&mut self, reference: GroupReference, cx: &mut Context<Self>) {
+        let key = reference.key();
+        let invite = reference.invite().map(str::to_owned);
+
+        self.ensure(key.clone(), cx);
+
+        let Some(group) = self.group(&key, cx) else {
+            return;
+        };
+
+        self.upsert_entry(&key, None);
+        self.publish_list(cx);
+
+        cx.emit(GroupsEvent::Updated);
+        cx.notify();
+
+        let builder = key.id().join_request(invite.as_deref(), "");
+        let sent = group.read(cx).publish(builder, cx);
+
+        group.update(cx, |group, cx| {
+            group.mark_pending(Timestamp::now(), cx);
+        });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            if let Err(error) = sent.await {
+                let reason = error.to_string();
+
+                if let Err(error) = this.update(cx, |this, cx| {
+                    if let Some(group) = this.group(&key, cx) {
+                        group.update(cx, |group, cx| {
+                            group.mark_refused(reason, cx);
+                        });
+                    }
+                }) {
+                    log::warn!("nip29: marking the refused join failed: {error}");
+                }
+            }
+        }));
+    }
+
+    pub fn leave(&mut self, key: &GroupKey, cx: &mut Context<Self>) {
+        let Some(group) = self.group(key, cx) else {
+            return;
+        };
+
+        let builder = key.id().leave_request("");
+        let sent = group.read_with(cx, |group, cx| group.publish(builder, cx));
+        let key = key.clone();
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            if let Err(error) = sent.await {
+                log::warn!("nip29: leaving {} failed: {error}", key.id());
+                return;
+            }
+            this.update(cx, |this, cx| {
+                this.forget(key, cx);
+            })
+            .ok();
+        }));
+    }
+
     pub fn reset(&mut self, cx: &mut Context<Self>) {
         self.notification_listener = None;
         self.signal_consumer = None;
@@ -134,9 +197,11 @@ impl GroupsRegistry {
         let routes = std::mem::take(&mut self.routes);
 
         self.groups.clear();
+        self.observers.clear();
         self.list = GroupList::default();
 
-        let client = NostrRegistry::global(cx).read(cx).client();
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
 
         let mut ids: Vec<SubscriptionId> = routes.into_keys().collect();
         ids.push(list_subscription_id());
@@ -201,7 +266,21 @@ impl GroupsRegistry {
         }
 
         let group = cx.new(|_| Group::new(key.clone()));
-        self.groups.push(group);
+        self.groups.push(group.clone());
+
+        self.observers.insert(
+            key.clone(),
+            cx.subscribe(&group, |this, group, event, cx| match event {
+                GroupEvent::Sent(event) => {
+                    let key = group.read(cx).key().clone();
+                    this.reload(key, Some(event.clone()), cx);
+                }
+                GroupEvent::Updated => {
+                    this.refresh_entry(group, cx);
+                }
+                _ => {}
+            }),
+        );
 
         self.subscribe(key.clone(), cx);
         self.reload(key, None, cx);
@@ -219,6 +298,108 @@ impl GroupsRegistry {
         });
     }
 
+    fn forget(&mut self, key: GroupKey, cx: &mut Context<Self>) {
+        let id = subscription_id(&key);
+        let cache_tag = key.cache_tag();
+
+        self.routes.remove(&id);
+        self.observers.remove(&key);
+        self.groups.retain(|group| group.read(cx).key() != &key);
+        self.list.remove(&key);
+
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+
+        self.tasks.push(cx.background_spawn(async move {
+            if let Err(error) = client.unsubscribe(&id).await {
+                log::warn!("nip29: unsubscribing {id} failed: {error}");
+            }
+
+            let filter = Filter::new()
+                .kind(Kind::ApplicationSpecificData)
+                .custom_tag(SingleLetterTag::LOWERCASE_R, cache_tag);
+
+            if let Err(error) = client.database().delete(filter).await {
+                log::warn!("nip29: clearing the group cache failed: {error}");
+            }
+        }));
+
+        self.publish_list(cx);
+
+        cx.emit(GroupsEvent::Updated);
+        cx.notify();
+    }
+
+    fn entry_name(&self, key: &GroupKey) -> Option<String> {
+        self.list
+            .entries()
+            .iter()
+            .find(|entry| entry.id == *key.id() && entry.relay == *key.relay())
+            .and_then(|entry| entry.name.clone())
+    }
+
+    fn upsert_entry(&mut self, key: &GroupKey, name: Option<String>) {
+        let name = name.or_else(|| self.entry_name(key));
+
+        self.list.upsert(GroupListEntry {
+            id: key.id().clone(),
+            relay: key.relay().clone(),
+            name,
+        });
+    }
+
+    fn refresh_entry(&mut self, group: Entity<Group>, cx: &mut Context<Self>) {
+        let (key, Some(name)) = ({
+            let group = group.read(cx);
+
+            (
+                group.key().clone(),
+                group
+                    .metadata()
+                    .and_then(|metadata| metadata.name().map(str::to_owned)),
+            )
+        }) else {
+            return;
+        };
+
+        if self.entry_name(&key).as_deref() == Some(name.as_str()) {
+            return;
+        }
+
+        self.upsert_entry(&key, Some(name));
+        self.publish_list(cx);
+
+        cx.emit(GroupsEvent::Updated);
+        cx.notify();
+    }
+
+    fn publish_list(&mut self, cx: &mut Context<Self>) {
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+        let signer = nostr.read(cx).signer();
+
+        let builder = self.list.to_builder();
+
+        self.tasks.push(cx.spawn(async move |_this, _cx| {
+            let event = match builder.finalize_async(&signer).await {
+                Ok(event) => event,
+                Err(error) => {
+                    log::warn!("nip29: signing the group list failed: {error}");
+                    return;
+                }
+            };
+
+            match client.send_event(&event).to_nip65().await {
+                Ok(output) if output.failed.is_empty() => {}
+                Ok(output) => log::warn!(
+                    "nip29: {} relay(s) rejected the group list",
+                    output.failed.len()
+                ),
+                Err(error) => log::warn!("nip29: publishing the group list failed: {error}"),
+            }
+        }));
+    }
+
     fn reload(&mut self, key: GroupKey, event: Option<Event>, cx: &mut Context<Self>) {
         let nostr = NostrRegistry::global(cx);