feat: add support for NIP-29 (#56)
Reviewed-on: #56
This commit was merged in pull request #56.
This commit is contained in:
@@ -0,0 +1,909 @@
|
||||
use std::collections::hash_map::DefaultHasher;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::hash::{Hash, Hasher};
|
||||
use std::sync::LazyLock;
|
||||
use std::time::Duration;
|
||||
|
||||
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 FORK_SUBSCRIPTION: &str = "nip29-forks";
|
||||
const GROUP_PREFIX: &str = "grp-";
|
||||
const BROWSE_PREFIX: &str = "brw-";
|
||||
const PAGE_PREFIX: &str = "pg-";
|
||||
const BROWSE_LIMIT: usize = 500;
|
||||
const BROWSE_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
const FORK_INTERVAL: Duration = Duration::from_secs(15 * 60);
|
||||
|
||||
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)]
|
||||
pub enum GroupsEvent {
|
||||
Updated,
|
||||
Error(String),
|
||||
Browsed(Box<GroupCandidate>),
|
||||
BrowseSettled(RelayUrl),
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
enum Signal {
|
||||
Event(SubscriptionId, Event),
|
||||
Eose(SubscriptionId),
|
||||
Closed(SubscriptionId, String),
|
||||
BrowseSettled(SubscriptionId),
|
||||
Forks,
|
||||
}
|
||||
|
||||
fn list_subscription_id() -> SubscriptionId {
|
||||
SubscriptionId::new(LIST_SUBSCRIPTION)
|
||||
}
|
||||
|
||||
fn fork_subscription_id() -> SubscriptionId {
|
||||
SubscriptionId::new(FORK_SUBSCRIPTION)
|
||||
}
|
||||
|
||||
fn subscription_id(key: &GroupKey) -> SubscriptionId {
|
||||
SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32))
|
||||
}
|
||||
|
||||
fn browse_subscription(relay: &RelayUrl) -> SubscriptionId {
|
||||
let mut hasher = DefaultHasher::new();
|
||||
relay.hash(&mut hasher);
|
||||
|
||||
SubscriptionId::new(format!("{BROWSE_PREFIX}{:08x}", hasher.finish() as u32))
|
||||
}
|
||||
|
||||
fn page_subscription(key: &GroupKey) -> SubscriptionId {
|
||||
SubscriptionId::new(format!("{PAGE_PREFIX}{:08x}", key.uniq_id() as u32))
|
||||
}
|
||||
|
||||
fn is_page(id: &SubscriptionId) -> bool {
|
||||
id.as_str().starts_with(PAGE_PREFIX)
|
||||
}
|
||||
|
||||
fn is_own_subscription(id: &SubscriptionId) -> bool {
|
||||
let id = id.as_str();
|
||||
id == LIST_SUBSCRIPTION
|
||||
|| id == FORK_SUBSCRIPTION
|
||||
|| id.starts_with(GROUP_PREFIX)
|
||||
|| id.starts_with(BROWSE_PREFIX)
|
||||
|| id.starts_with(PAGE_PREFIX)
|
||||
}
|
||||
|
||||
impl EventEmitter<GroupsEvent> for GroupsRegistry {}
|
||||
|
||||
pub struct GroupsRegistry {
|
||||
groups: Vec<Entity<Group>>,
|
||||
list: GroupList,
|
||||
browse: HashMap<SubscriptionId, RelayUrl>,
|
||||
routes: HashMap<SubscriptionId, GroupKey>,
|
||||
observers: HashMap<GroupKey, Subscription>,
|
||||
forks: HashMap<PublicKey, (Timestamp, GroupList)>,
|
||||
fork_authors: HashSet<PublicKey>,
|
||||
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(),
|
||||
browse: HashMap::new(),
|
||||
routes: HashMap::new(),
|
||||
observers: HashMap::new(),
|
||||
forks: HashMap::new(),
|
||||
fork_authors: HashSet::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 join(&mut self, reference: GroupReference, cx: &mut Context<Self>) {
|
||||
self.join_key(reference.key(), reference.invite().map(str::to_owned), cx);
|
||||
}
|
||||
|
||||
pub fn join_key(&mut self, key: GroupKey, invite: Option<String>, cx: &mut Context<Self>) {
|
||||
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 create(&mut self, key: GroupKey, metadata: GroupMetadata, cx: &mut Context<Self>) {
|
||||
self.ensure(key.clone(), cx);
|
||||
|
||||
let Some(group) = self.group(&key, cx) else {
|
||||
return;
|
||||
};
|
||||
|
||||
self.upsert_entry(&key, metadata.name().map(str::to_owned));
|
||||
self.publish_list(cx);
|
||||
|
||||
cx.emit(GroupsEvent::Updated);
|
||||
cx.notify();
|
||||
|
||||
let builder = key.id().create_group(&metadata, None);
|
||||
let published = group.read(cx).publish(builder, cx);
|
||||
let logged = key.clone();
|
||||
|
||||
self.tasks.push(cx.spawn(async move |this, cx| {
|
||||
if let Err(e) = published.await {
|
||||
this.update(cx, |this, cx| {
|
||||
this.forget(logged, cx);
|
||||
cx.emit(GroupsEvent::Error(e.to_string()));
|
||||
})
|
||||
.ok();
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
pub fn browse(&mut self, relay: RelayUrl, cx: &mut Context<Self>) {
|
||||
let nostr = NostrRegistry::global(cx);
|
||||
let client = nostr.read(cx).client();
|
||||
|
||||
let id = browse_subscription(&relay);
|
||||
let tx = self.signal_tx.clone();
|
||||
|
||||
self.browse.insert(id.clone(), relay.clone());
|
||||
|
||||
self.tasks.push(cx.background_spawn(async move {
|
||||
client.add_relay(relay.clone()).and_connect().await.ok();
|
||||
|
||||
let filter = Filter::new().kind(Kind::GroupMetadata).limit(BROWSE_LIMIT);
|
||||
let opts = SubscribeAutoCloseOptions::default()
|
||||
.exit_policy(ReqExitPolicy::ExitOnEOSE)
|
||||
.timeout(Some(BROWSE_TIMEOUT));
|
||||
|
||||
if let Err(error) = client
|
||||
.subscribe(ReqTarget::single(relay.clone(), [filter]))
|
||||
.with_id(id.clone())
|
||||
.close_on(opts)
|
||||
.await
|
||||
{
|
||||
log::warn!("nip29: browsing {relay} failed: {error}");
|
||||
}
|
||||
|
||||
if let Err(error) = tx.send_async(Signal::BrowseSettled(id)).await {
|
||||
log::warn!("nip29: the browse settle signal was dropped: {error}");
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
fn settle_browse(&mut self, id: SubscriptionId, cx: &mut Context<Self>) {
|
||||
let Some(relay) = self.browse.get(&id).cloned() else {
|
||||
return;
|
||||
};
|
||||
cx.emit(GroupsEvent::BrowseSettled(relay));
|
||||
}
|
||||
|
||||
pub fn load_more(&mut self, key: &GroupKey, cx: &mut Context<Self>) {
|
||||
let Some(group) = self.group(key, cx) else {
|
||||
return;
|
||||
};
|
||||
|
||||
if group.read(cx).loading_more() || group.read(cx).oldest().is_none() {
|
||||
return;
|
||||
}
|
||||
|
||||
self.routes.insert(page_subscription(key), key.clone());
|
||||
|
||||
group.update(cx, |group, cx| {
|
||||
group.load_page(cx);
|
||||
});
|
||||
}
|
||||
|
||||
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;
|
||||
self.tasks.clear();
|
||||
|
||||
let routes = std::mem::take(&mut self.routes);
|
||||
|
||||
self.groups.clear();
|
||||
self.observers.clear();
|
||||
self.forks.clear();
|
||||
self.fork_authors.clear();
|
||||
self.list = GroupList::default();
|
||||
|
||||
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());
|
||||
ids.push(fork_subscription_id());
|
||||
ids.extend(self.browse.keys().cloned());
|
||||
|
||||
self.browse.clear();
|
||||
|
||||
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}"),
|
||||
}));
|
||||
|
||||
self.watch_forks(cx);
|
||||
}
|
||||
|
||||
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 watch_forks(&mut self, cx: &mut Context<Self>) {
|
||||
let executor = cx.background_executor().clone();
|
||||
let tx = self.signal_tx.clone();
|
||||
|
||||
self.tasks.push(cx.background_spawn(async move {
|
||||
loop {
|
||||
executor.timer(FORK_INTERVAL).await;
|
||||
|
||||
if tx.send_async(Signal::Forks).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
fn admin_keys(&self, cx: &App) -> HashSet<PublicKey> {
|
||||
let mut keys = HashSet::new();
|
||||
|
||||
for group in &self.groups {
|
||||
for (admin, _) in group.read(cx).admins().entries() {
|
||||
keys.insert(*admin);
|
||||
}
|
||||
}
|
||||
|
||||
keys
|
||||
}
|
||||
|
||||
fn sync_forks(&mut self, force: bool, cx: &mut Context<Self>) {
|
||||
let authors = self.admin_keys(cx);
|
||||
|
||||
if authors.is_empty() || (!force && authors == self.fork_authors) {
|
||||
return;
|
||||
}
|
||||
|
||||
let nostr = NostrRegistry::global(cx);
|
||||
let client = nostr.read(cx).client();
|
||||
let filter = Filter::new()
|
||||
.authors(authors.iter().copied())
|
||||
.kind(Kind::SimpleGroups);
|
||||
|
||||
self.fork_authors = authors;
|
||||
|
||||
self.tasks.push(cx.background_spawn(async move {
|
||||
if let Err(error) = client
|
||||
.subscribe(ReqTarget::auto([filter]))
|
||||
.with_id(fork_subscription_id())
|
||||
.await
|
||||
{
|
||||
log::warn!("nip29: subscribing to the admins' group lists failed: {error}");
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
fn absorb_fork_list(&mut self, event: &Event, cx: &mut Context<Self>) {
|
||||
let Ok(list) = GroupList::parse(event) else {
|
||||
log::warn!("nip29: ignoring a malformed admin group list");
|
||||
return;
|
||||
};
|
||||
|
||||
let author = event.pubkey;
|
||||
|
||||
if let Some((seen, _)) = self.forks.get(&author)
|
||||
&& *seen >= event.created_at
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
self.forks.insert(author, (event.created_at, list));
|
||||
self.evaluate_forks(cx);
|
||||
}
|
||||
|
||||
fn evaluate_forks(&mut self, cx: &mut Context<Self>) {
|
||||
let groups = self.groups.clone();
|
||||
|
||||
for group in groups {
|
||||
let elsewhere = {
|
||||
let group = group.read(cx);
|
||||
let key = group.key();
|
||||
|
||||
group
|
||||
.admins()
|
||||
.entries()
|
||||
.iter()
|
||||
.filter_map(|(admin, _)| self.forks.get(admin))
|
||||
.flat_map(|(_, list)| list.entries())
|
||||
.find(|entry| entry.id == *key.id() && entry.relay != *key.relay())
|
||||
.map(|entry| GroupKey::new(entry.relay.clone(), entry.id.clone()))
|
||||
};
|
||||
|
||||
group.update(cx, |group, cx| {
|
||||
group.set_elsewhere(elsewhere, cx);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
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.clone());
|
||||
|
||||
self.observers.insert(
|
||||
key.clone(),
|
||||
cx.subscribe(&group, |this, group, event, cx| match event {
|
||||
GroupEvent::Sent(event) => {
|
||||
if event.kind == Kind::GroupDeleteEvent {
|
||||
this.forget_event(group, event.clone(), cx);
|
||||
} else {
|
||||
let key = group.read(cx).key().clone();
|
||||
this.reload(key, Some(event.clone()), cx);
|
||||
}
|
||||
}
|
||||
GroupEvent::Updated => {
|
||||
this.refresh_entry(group, cx);
|
||||
this.sync_forks(false, cx);
|
||||
|
||||
cx.emit(GroupsEvent::Updated);
|
||||
cx.notify();
|
||||
}
|
||||
_ => {}
|
||||
}),
|
||||
);
|
||||
|
||||
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 forget(&mut self, key: GroupKey, cx: &mut Context<Self>) {
|
||||
let id = subscription_id(&key);
|
||||
let page = page_subscription(&key);
|
||||
let cache_tag = key.cache_tag();
|
||||
|
||||
self.routes.remove(&id);
|
||||
self.routes.remove(&page);
|
||||
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 {
|
||||
for id in [id, page] {
|
||||
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);
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
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 forget_event(&mut self, group: Entity<Group>, event: Event, cx: &mut Context<Self>) {
|
||||
let Some(target) = event.tags.event_ids().next() else {
|
||||
return;
|
||||
};
|
||||
|
||||
let key = group.read(cx).key().clone();
|
||||
let cache_tag = key.cache_tag();
|
||||
|
||||
let nostr = NostrRegistry::global(cx);
|
||||
let client = nostr.read(cx).client();
|
||||
|
||||
let purge = cx.background_spawn(async move {
|
||||
let filter = Filter::new()
|
||||
.kind(Kind::ApplicationSpecificData)
|
||||
.identifier(target.to_hex())
|
||||
.custom_tag(SingleLetterTag::LOWERCASE_R, cache_tag);
|
||||
|
||||
client.database().delete(filter).await
|
||||
});
|
||||
|
||||
self.tasks.push(cx.spawn(async move |this, cx| {
|
||||
if let Err(error) = purge.await {
|
||||
log::warn!("nip29: dropping the deleted envelope failed: {error}");
|
||||
}
|
||||
this.update(cx, |this, cx| {
|
||||
group.update(cx, |group, cx| {
|
||||
group.forget(target, cx);
|
||||
});
|
||||
this.reload(key, Some(event), cx);
|
||||
})
|
||||
.ok();
|
||||
}));
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
if id == fork_subscription_id() {
|
||||
self.absorb_fork_list(&event, cx);
|
||||
return;
|
||||
}
|
||||
|
||||
if self.browse.contains_key(&id) {
|
||||
self.absorb_candidate(&id, &event, cx);
|
||||
return;
|
||||
}
|
||||
|
||||
let Some(key) = self.routes.get(&id).cloned() else {
|
||||
return;
|
||||
};
|
||||
|
||||
self.reload(key, Some(event), cx);
|
||||
}
|
||||
|
||||
fn absorb_candidate(&mut self, id: &SubscriptionId, event: &Event, cx: &mut Context<Self>) {
|
||||
let Some(relay) = self.browse.get(id).cloned() else {
|
||||
return;
|
||||
};
|
||||
|
||||
let Ok(metadata) = GroupMetadata::parse(event) else {
|
||||
log::warn!("nip29: ignoring malformed group metadata from {relay}");
|
||||
return;
|
||||
};
|
||||
|
||||
let Some(group) = metadata.id().cloned() else {
|
||||
return;
|
||||
};
|
||||
|
||||
cx.emit(GroupsEvent::Browsed(Box::new(GroupCandidate {
|
||||
key: GroupKey::new(relay, group),
|
||||
metadata,
|
||||
})));
|
||||
}
|
||||
|
||||
fn handle_eose(&mut self, id: SubscriptionId, cx: &mut Context<Self>) {
|
||||
let Some(key) = self.routes.get(&id).cloned() else {
|
||||
return;
|
||||
};
|
||||
|
||||
if is_page(&id) {
|
||||
self.routes.remove(&id);
|
||||
|
||||
if let Some(group) = self.group(&key, cx) {
|
||||
group.update(cx, |group, cx| {
|
||||
group.page_finished(cx);
|
||||
});
|
||||
}
|
||||
|
||||
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>) {
|
||||
if let Some(relay) = self.browse.get(&id) {
|
||||
log::warn!("nip29: browsing {relay} was refused: {reason}");
|
||||
return;
|
||||
}
|
||||
|
||||
if id == fork_subscription_id() {
|
||||
log::warn!("nip29: the admins' group lists were refused: {reason}");
|
||||
return;
|
||||
}
|
||||
|
||||
let Some(key) = self.routes.get(&id).cloned() else {
|
||||
return;
|
||||
};
|
||||
|
||||
if is_page(&id) {
|
||||
self.routes.remove(&id);
|
||||
log::warn!(
|
||||
"nip29: loading older messages in {} was refused: {reason}",
|
||||
key.id()
|
||||
);
|
||||
|
||||
if let Some(group) = self.group(&key, cx) {
|
||||
group.update(cx, |group, cx| {
|
||||
group.page_finished(cx);
|
||||
});
|
||||
}
|
||||
|
||||
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_own_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_own_subscription(&id).then_some(Signal::Eose(id))
|
||||
}
|
||||
RelayMessage::Closed {
|
||||
subscription_id,
|
||||
message,
|
||||
} => {
|
||||
let id = subscription_id.into_owned();
|
||||
is_own_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);
|
||||
}
|
||||
Signal::BrowseSettled(id) => {
|
||||
this.settle_browse(id, cx);
|
||||
}
|
||||
Signal::Forks => {
|
||||
this.sync_forks(true, cx);
|
||||
}
|
||||
})
|
||||
.ok();
|
||||
}
|
||||
}));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user