From 6e6360c488650b70f6c53b5adf8dc721738cc513 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Tue, 20 Oct 2026 20:42:55 +0700 Subject: [PATCH] update --- crates/chat/src/lib.rs | 27 +- crates/chat/src/room.rs | 15 +- crates/person/src/wot.rs | 655 +++++++++++---------------------------- 3 files changed, 218 insertions(+), 479 deletions(-) diff --git a/crates/chat/src/lib.rs b/crates/chat/src/lib.rs index c64db7fb..f306fd9b 100644 --- a/crates/chat/src/lib.rs +++ b/crates/chat/src/lib.rs @@ -14,6 +14,7 @@ use gpui::{ use instant::Duration; use nip4e::{DeviceEvent, DeviceRegistry}; use nostr_sdk::prelude::*; +use person::PersonRegistry; use smallvec::{SmallVec, smallvec}; use state::{DEVICE_GIFTWRAP, NostrRegistry, USER_GIFTWRAP}; @@ -139,13 +140,13 @@ impl ChatRegistry { /// Create a new chat registry instance fn new(cx: &mut Context) -> Self { let nostr = NostrRegistry::global(cx); + let device = DeviceRegistry::global(cx); + let wot = PersonRegistry::global(cx).read(cx).wot(); let entity = cx.entity().downgrade(); let (tx, rx) = flume::unbounded::(); let mut subscriptions = smallvec![]; - let device = DeviceRegistry::global(cx); - subscriptions.push(cx.subscribe(&nostr, |this, _nostr, event, cx| { if event.signer_changed() { this.reset(cx); @@ -163,6 +164,10 @@ impl ChatRegistry { } })); + subscriptions.push(cx.subscribe(&wot, |this, _wot, _event, cx| { + this.promote_trusted(cx); + })); + cx.defer(move |cx| { let _ = entity.update(cx, |this, cx| { this.get_rooms(cx); @@ -567,7 +572,23 @@ impl ChatRegistry { .unwrap_or_default() } - /// Add a new room to the start of list. + /// Promote request rooms whose senders are inside the web of trust + fn promote_trusted(&mut self, cx: &mut Context) { + let nostr = NostrRegistry::global(cx); + let Some(me) = nostr.read(cx).current_user() else { + return; + }; + + for room in &self.rooms { + room.update(cx, |room, cx| { + if room.kind == RoomKind::Request && room.auto_trusted(me, cx) { + room.set_ongoing(cx); + } + }); + } + } + + /// Promote a room to ongoing when its latest message is from the current user pub(crate) fn add_room(&mut self, room: I, cx: &mut Context) where I: Into, diff --git a/crates/chat/src/room.rs b/crates/chat/src/room.rs index ce649a04..42454a69 100644 --- a/crates/chat/src/room.rs +++ b/crates/chat/src/room.rs @@ -195,17 +195,24 @@ impl Room { Self::trusted_by(&self.members, me, |key| wot.is_in_network(key)) } - /// Whether the room looks like spam: everyone other than the current user - /// is outside the web of trust + /// Whether the room looks like spam: the web of trust is loaded and + /// everyone other than the current user is outside of it pub fn is_spam(&self, cx: &App) -> bool { + let person = PersonRegistry::global(cx); let nostr = NostrRegistry::global(cx); + let Some(me) = nostr.read(cx).current_user() else { return false; }; - let wot = PersonRegistry::global(cx).read(cx).wot().read(cx); + let wot = person.read(cx).wot().read(cx); - Self::trusted_by(&self.members, me, |key| !wot.is_outside(key)) + // An unloaded graph cannot classify anything as spam + if !wot.is_loaded() { + return false; + } + + Self::trusted_by(&self.members, me, |key| wot.is_outside(key)) } /// Whether every member other than `me` passes the check diff --git a/crates/person/src/wot.rs b/crates/person/src/wot.rs index fead0fcd..f0475edb 100644 --- a/crates/person/src/wot.rs +++ b/crates/person/src/wot.rs @@ -1,6 +1,6 @@ use std::collections::{HashMap, HashSet}; -use gpui::{AppContext, Context, EventEmitter, Task}; +use gpui::{Context, EventEmitter, Task}; use instant::Duration; use nostr_sdk::prelude::*; use smallvec::{SmallVec, smallvec}; @@ -11,38 +11,89 @@ const MAX_FOLLOWS: usize = 2000; /// Cap a followed author's contact list to bound hostile publishers const MAX_TAGS_PER_LIST: usize = 5000; /// Time between reconciliations with the relays -const INTERVAL: u64 = 15 * 60; +const SYNC_INTERVAL: u64 = 15 * 60; +/// Longest wait between retries while the graph has no data yet +const MAX_RETRY_WAIT: u64 = 60; /// The web of trust graph changed #[derive(Debug, Clone)] pub struct WoTEvent; +/// A built graph: the follow set and, per public key, who follows it +#[derive(Debug)] +struct Snapshot { + follows: HashSet, + scores: HashMap, + mutuals: HashMap>, + disabled: bool, +} + +impl Snapshot { + /// Build the snapshot from the newest stored contact list of every author + fn build(self_key: PublicKey, follows: HashSet, events: Vec) -> Self { + let disabled = follows.len() > MAX_FOLLOWS; + + let mut snapshot = Self { + follows, + scores: HashMap::new(), + mutuals: HashMap::new(), + disabled, + }; + + if snapshot.disabled { + return snapshot; + } + + // Keep the newest contact list per author + let mut latest: HashMap = HashMap::new(); + + for event in events { + if event.pubkey == self_key || !snapshot.follows.contains(&event.pubkey) { + continue; + } + + match latest.get(&event.pubkey) { + Some(current) if current.created_at > event.created_at => continue, + _ => { + latest.insert(event.pubkey, event); + } + } + } + + for (author, event) in latest { + for key in event.tags.public_keys().take(MAX_TAGS_PER_LIST) { + if key == author || key == self_key { + continue; + } + + *snapshot.scores.entry(key).or_default() += 1; + snapshot.mutuals.entry(key).or_default().push(author); + } + } + + for contacts in snapshot.mutuals.values_mut() { + contacts.sort(); + } + + snapshot + } +} + impl EventEmitter for WebOfTrust {} /// Web of trust /// -/// Scores public keys by how many of the current user's contacts follow them, -/// built from the contact lists (NIP-02) of the people the user follows. +/// Scores public keys by how many of the current user's contacts follow them. #[derive(Debug)] pub struct WebOfTrust { /// Current user's public key self_key: Option, - /// Latest known contact list of the current user - follows: HashSet, - /// Created at of the latest contact list of the current user - follows_at: Option, - /// Latest known contact list per followed author, with its created at - lists: HashMap)>, - /// Reverse index: target -> contacts who follow them - reverse: HashMap>, - /// Whether scoring is disabled due to an oversized follow set - disabled: bool, - /// Whether the initial load has finished - ready: bool, - /// Task loading the graph, replaced by newer loads - hydration: Option>, - /// Tasks for asynchronous operations - tasks: SmallVec<[Task<()>; 4]>, + /// Latest built snapshot, none until the first refresh completes + snapshot: Option, + /// In-flight refresh, replaced by newer ones + refresh_task: Option>, + /// Keeps the periodic reconciliation loop alive + _tasks: SmallVec<[Task<()>; 4]>, } impl WebOfTrust { @@ -52,14 +103,24 @@ impl WebOfTrust { let mut tasks = smallvec![]; tasks.push(cx.spawn(async move |this, cx| { + let mut delay = Duration::from_secs(5); + loop { - cx.background_executor() - .timer(Duration::from_secs(INTERVAL)) - .await; + cx.background_executor().timer(delay).await; + + let mut has_data = false; + // Ignore errors, the next refresh will retry let _ = this.update(cx, |this, cx| { - this.sync(cx); + this.refresh(cx); + has_data = this.has_data(); }); + + delay = if has_data { + Duration::from_secs(SYNC_INTERVAL) + } else { + (delay * 2).min(Duration::from_secs(MAX_RETRY_WAIT)) + }; } })); @@ -71,72 +132,35 @@ impl WebOfTrust { Self { self_key: None, - follows: HashSet::new(), - follows_at: None, - lists: HashMap::new(), - reverse: HashMap::new(), - disabled: false, - ready: false, - hydration: None, - tasks, + snapshot: None, + refresh_task: None, + _tasks: tasks, } } /// Load the graph for the current account pub(crate) fn hydrate(&mut self, cx: &mut Context) { let nostr = NostrRegistry::global(cx); - let client = nostr.read(cx).client(); - let Some(self_key) = nostr.read(cx).current_user() else { return; }; self.self_key = Some(self_key); - - // The locally stored contact list, applied before the relays respond - let local: Task> = cx.background_spawn(async move { - let filter = Filter::new() - .author(self_key) - .kind(Kind::ContactList) - .limit(1); - - client - .database() - .query(filter) - .await - .unwrap_or_default() - .into_iter() - .next() - }); - - self.hydration = Some(cx.spawn(async move |this, cx| { - let latest = local.await; - let at = latest.as_ref().map(|event| event.created_at); - - let snapshot: HashSet = latest - .as_ref() - .map(|event| event.tags.public_keys().take(MAX_FOLLOWS + 1).collect()) - .unwrap_or_default(); - - // Credit the local graph before the relays respond - this.update(cx, |this, cx| { - if this.self_key == Some(self_key) && this.store_follows(at, snapshot.clone()) { - cx.emit(WoTEvent); - cx.notify(); - } - }) - .ok(); - - // Reconcile with the relays; the refresh this triggers finalizes the initial load - this.update(cx, |this, cx| { - this.sync(cx); - }) - .ok(); - })); + self.refresh(cx); } - /// Reconcile the follow set's contact lists with the relays and apply whatever came back - fn sync(&mut self, cx: &mut Context) { + /// Clear the graph on sign out or account switch + pub(crate) fn reset(&mut self, cx: &mut Context) { + self.self_key = None; + self.snapshot = None; + self.refresh_task = None; + + cx.emit(WoTEvent); + cx.notify(); + } + + /// Reconcile the follow set's contact lists with the relays and rebuild. + fn refresh(&mut self, cx: &mut Context) { let nostr = NostrRegistry::global(cx); let client = nostr.read(cx).client(); @@ -144,444 +168,131 @@ impl WebOfTrust { return; }; - let covered: HashSet = self.follows.clone(); + self.refresh_task = Some(cx.spawn(async move |this, cx| { + // Scope the reconcile to the follow set known so far + let known: HashSet = stored_contact_list(&client, self_key) + .await + .map(|event| event.tags.public_keys().take(MAX_FOLLOWS + 1).collect()) + .unwrap_or_default(); - let mut authors: Vec = covered.iter().copied().collect(); - authors.push(self_key); + let mut authors: Vec = known.iter().copied().collect(); + authors.push(self_key); - self.tasks.push(cx.spawn(async move |this, cx| { + // Negentropy transfers only the lists the relays hold + // and the database is missing let filter = Filter::new().kind(Kind::ContactList).authors(authors); - - // Negentropy transfers only the lists the relays hold and the database is missing if let Err(error) = client.sync(filter).await { log::warn!("Failed to reconcile contact lists: {error}"); } - let mut authors: Vec = covered.iter().copied().collect(); + // The follow set after the reconcile + let Some(own) = stored_contact_list(&client, self_key).await else { + return; + }; + + let follows: HashSet = + own.tags.public_keys().take(MAX_FOLLOWS + 1).collect(); + + let mut authors: Vec = follows.iter().copied().collect(); authors.push(self_key); - let filter = Filter::new().kind(Kind::ContactList).authors(authors); - let events: Vec = client - .database() - .query(filter) - .await - .unwrap_or_default() - .into_iter() - .collect(); + let events: Vec = if follows.is_empty() { + Vec::new() + } else { + let filter = Filter::new().kind(Kind::ContactList).authors(authors); + client + .database() + .query(filter) + .await + .unwrap_or_default() + .into_iter() + .collect() + }; - let mut grew = false; + let snapshot = Snapshot::build(self_key, follows.clone(), events); + log::debug!( + "Web of trust rebuilt: {} follows, {} scored keys", + snapshot.follows.len(), + snapshot.scores.len() + ); + + // Newly followed authors were not part of the reconcile + let grew = !known.is_superset(&follows); this.update(cx, |this, cx| { - let was_ready = this.ready; - let changed = this.apply_latest(self_key, events, cx); - this.ready = true; - - if changed || !was_ready { - cx.emit(WoTEvent); - cx.notify(); - } - - grew = !covered.is_superset(&this.follows); + this.snapshot = Some(snapshot); + cx.emit(WoTEvent); + cx.notify(); }) .ok(); - // The follow set grew beyond the covered authors, - // reconcile once more so their contact lists are included if grew { this.update(cx, |this, cx| { - this.sync(cx); + this.refresh(cx); }) .ok(); } })); } - /// Apply the newest stored contact list per author, returning whether the graph changed - fn apply_latest( - &mut self, - self_key: PublicKey, - events: Vec, - cx: &mut Context, - ) -> bool { - if self.self_key != Some(self_key) { - return false; - } - - // Keep the newest contact list per author - let mut latest: HashMap = HashMap::new(); - - for event in events { - if event.pubkey != self_key && (self.disabled || !self.follows.contains(&event.pubkey)) - { - continue; - } - match latest.get(&event.pubkey) { - Some(current) if current.created_at > event.created_at => continue, - _ => { - latest.insert(event.pubkey, event); - } - } - } - - let mut changed = false; - - for event in latest.into_values() { - let was_disabled = self.disabled; - - changed |= if event.pubkey == self_key { - self.apply_own(&event) - } else { - self.apply_contact(&event) - }; - - if was_disabled && !self.disabled { - // The graph was cleared while disabled and has to be rebuilt - self.sync(cx); - } - } - - changed - } - - /// Apply the current user's contact list, returning whether the graph changed - fn apply_own(&mut self, event: &Event) -> bool { - if self.follows_at.is_some_and(|at| at > event.created_at) { - return false; - } - - let follows: HashSet = event.tags.public_keys().take(MAX_FOLLOWS + 1).collect(); - - self.store_follows(Some(event.created_at), follows) - } - - /// Apply a followed author's contact list, returning whether the graph changed - fn apply_contact(&mut self, event: &Event) -> bool { - if self.disabled || !self.follows.contains(&event.pubkey) { - return false; - } - - let contacts = self.parse_contacts(event); - - self.set_contact_set(event.pubkey, event.created_at, contacts) - } - - /// Parse a capped contact set from a contact list event - fn parse_contacts(&self, event: &Event) -> HashSet { - event.tags.public_keys().take(MAX_TAGS_PER_LIST).collect() - } - - /// Store a follow set, un-credit removed contacts, and report changes - fn store_follows(&mut self, at: Option, follows: HashSet) -> bool { - self.follows_at = at; - - if follows.len() > MAX_FOLLOWS { - if self.disabled { - return false; - } - - self.follows.clear(); - self.lists.clear(); - self.reverse.clear(); - self.disabled = true; - return true; - } - - let removed: Vec = self.follows.difference(&follows).copied().collect(); - let added: Vec = follows.difference(&self.follows).copied().collect(); - - let changed = !removed.is_empty() || !added.is_empty(); - - self.follows = follows; - self.disabled = false; - - for follower in removed { - self.uncredit(&follower); - } - - changed - } - - /// Store a followed author's contact set, returning whether the graph changed - fn set_contact_set( - &mut self, - follower: PublicKey, - at: Timestamp, - contacts: HashSet, - ) -> bool { - if let Some((current, _)) = self.lists.get(&follower) - && *current > at - { - return false; - } - - let contacts: HashSet = contacts - .into_iter() - .filter(|key| *key != follower && Some(*key) != self.self_key) - .collect(); - - let (added, removed) = match self.lists.remove(&follower) { - Some((_, previous)) => ( - contacts - .difference(&previous) - .copied() - .collect::>(), - previous - .difference(&contacts) - .copied() - .collect::>(), - ), - None => (contacts.iter().copied().collect(), Vec::new()), - }; - - self.lists.insert(follower, (at, contacts)); - - let changed = !added.is_empty() || !removed.is_empty(); - - for target in added { - self.reverse.entry(target).or_default().insert(follower); - } - - for target in removed { - self.remove_credit(&follower, &target); - } - - changed - } - - /// Remove a contact's credits from the reverse index - fn uncredit(&mut self, follower: &PublicKey) { - if let Some((_, contacts)) = self.lists.remove(follower) { - for target in contacts { - self.remove_credit(follower, &target); - } - } - } - - /// Remove a single credit from the reverse index - fn remove_credit(&mut self, follower: &PublicKey, target: &PublicKey) { - if let Some(creditors) = self.reverse.get_mut(target) { - creditors.remove(follower); - - if creditors.is_empty() { - self.reverse.remove(target); - } - } - } - - /// Clear the graph on sign out or account switch - pub(crate) fn reset(&mut self, cx: &mut Context) { - self.self_key = None; - self.follows.clear(); - self.follows_at = None; - self.lists.clear(); - self.reverse.clear(); - self.disabled = false; - self.ready = false; - self.hydration = None; - - cx.emit(WoTEvent); - cx.notify(); - } - /// How many of the current user's contacts follow the public key pub fn score(&self, public_key: &PublicKey) -> usize { - self.reverse - .get(public_key) - .map_or(0, |creditors| creditors.len()) + self.snapshot.as_ref().map_or(0, |snapshot| { + snapshot.scores.get(public_key).copied().unwrap_or(0) + }) } /// Whether the public key is followed by the user or by any of their contacts pub fn is_in_network(&self, public_key: &PublicKey) -> bool { - self.follows.contains(public_key) || self.score(public_key) > 0 + self.snapshot + .as_ref() + .is_some_and(|snapshot| snapshot.follows.contains(public_key)) + || self.score(public_key) > 0 } /// Whether the graph is loaded and the public key is outside the network pub fn is_outside(&self, public_key: &PublicKey) -> bool { - self.ready && !self.disabled && !self.is_in_network(public_key) + self.is_loaded() && !self.is_in_network(public_key) } /// The current user's contacts who follow the public key pub fn mutuals(&self, public_key: &PublicKey) -> Vec { - self.reverse - .get(public_key) - .map_or_else(Vec::new, |creditors| { - let mut contacts: Vec = creditors.iter().copied().collect(); - contacts.sort(); - contacts - }) + self.snapshot.as_ref().map_or_else(Vec::new, |snapshot| { + snapshot + .mutuals + .get(public_key) + .cloned() + .unwrap_or_default() + }) } /// Whether the graph is loaded and score queries are meaningful pub fn is_loaded(&self) -> bool { - self.ready && !self.disabled + self.snapshot + .as_ref() + .is_some_and(|snapshot| !snapshot.disabled) + } + + /// Whether the snapshot covers a follow set + fn has_data(&self) -> bool { + self.snapshot + .as_ref() + .is_some_and(|snapshot| !snapshot.follows.is_empty()) } } -#[cfg(test)] -mod tests { - use super::*; +/// The newest stored contact list of the public key +async fn stored_contact_list(client: &Client, self_key: PublicKey) -> Option { + let filter = Filter::new() + .author(self_key) + .kind(Kind::ContactList) + .limit(1); - fn key(seed: usize) -> PublicKey { - let mut bytes = [0u8; 32]; - bytes[30] = (seed >> 8) as u8; - bytes[31] = seed as u8; - - Keys::new(SecretKey::from_slice(&bytes).expect("valid secret key")).public_key() - } - - fn registry() -> WebOfTrust { - WebOfTrust { - self_key: None, - follows: HashSet::new(), - follows_at: None, - lists: HashMap::new(), - reverse: HashMap::new(), - disabled: false, - ready: false, - hydration: None, - tasks: smallvec![], - } - } - - #[test] - fn credits_contacts_of_followed_authors() { - let mut wot = registry(); - let (me, alice, bob, carol) = (key(1), key(2), key(3), key(4)); - - wot.self_key = Some(me); - - assert!(wot.store_follows(Some(Timestamp::from_secs(10)), HashSet::from([alice, bob]))); - assert!(!wot.store_follows(Some(Timestamp::from_secs(10)), HashSet::from([alice, bob]))); - - assert!(wot.set_contact_set(alice, Timestamp::from_secs(10), HashSet::from([carol]),)); - assert_eq!(wot.score(&carol), 1); - assert_eq!(wot.mutuals(&carol), vec![alice]); - assert!(wot.is_in_network(&carol)); - - assert!(wot.set_contact_set(bob, Timestamp::from_secs(10), HashSet::from([carol]),)); - assert_eq!(wot.score(&carol), 2); - assert_eq!(wot.mutuals(&carol), vec![alice, bob]); - } - - #[test] - fn ignores_older_contact_lists() { - let mut wot = registry(); - let (me, alice, carol, dave) = (key(1), key(2), key(3), key(4)); - - wot.self_key = Some(me); - wot.store_follows(Some(Timestamp::from_secs(10)), HashSet::from([alice])); - - assert!(wot.set_contact_set(alice, Timestamp::from_secs(20), HashSet::from([carol]),)); - assert!(!wot.set_contact_set(alice, Timestamp::from_secs(10), HashSet::from([dave]),)); - - assert_eq!(wot.mutuals(&carol), vec![alice]); - assert_eq!(wot.score(&dave), 0); - } - - #[test] - fn uncredits_removed_follows() { - let mut wot = registry(); - let (me, alice, bob, carol) = (key(1), key(2), key(3), key(4)); - - wot.self_key = Some(me); - wot.store_follows(Some(Timestamp::from_secs(10)), HashSet::from([alice, bob])); - wot.set_contact_set(alice, Timestamp::from_secs(10), HashSet::from([carol])); - wot.set_contact_set(bob, Timestamp::from_secs(10), HashSet::from([carol])); - assert_eq!(wot.score(&carol), 2); - - assert!(wot.store_follows(Some(Timestamp::from_secs(11)), HashSet::from([bob]))); - - assert_eq!(wot.score(&carol), 1); - assert_eq!(wot.mutuals(&carol), vec![bob]); - assert!(!wot.lists.contains_key(&alice)); - } - - #[test] - fn excludes_self_and_follower() { - let mut wot = registry(); - let (me, alice, carol) = (key(1), key(2), key(3)); - - wot.self_key = Some(me); - wot.store_follows(Some(Timestamp::from_secs(10)), HashSet::from([alice])); - - wot.set_contact_set( - alice, - Timestamp::from_secs(10), - HashSet::from([me, alice, carol]), - ); - - assert_eq!(wot.score(&me), 0); - assert_eq!(wot.score(&alice), 0); - assert_eq!(wot.mutuals(&carol), vec![alice]); - } - - #[test] - fn disables_on_oversized_follow_set() { - let mut wot = registry(); - let me = key(1); - let follows: HashSet = (2..(MAX_FOLLOWS + 3)).map(key).collect(); - - wot.self_key = Some(me); - wot.set_contact_set(key(2), Timestamp::from_secs(10), HashSet::from([key(3)])); - assert!(wot.store_follows(Some(Timestamp::from_secs(10)), follows)); - - assert!(wot.disabled); - assert!(wot.follows.is_empty()); - assert!(wot.lists.is_empty()); - assert!(wot.reverse.is_empty()); - } - - #[test] - fn outside_requires_a_loaded_graph() { - let mut wot = registry(); - let (me, alice, carol) = (key(1), key(2), key(3)); - - wot.self_key = Some(me); - wot.store_follows(Some(Timestamp::from_secs(10)), HashSet::from([alice])); - - // The graph is not loaded yet - assert!(!wot.is_outside(&carol)); - - wot.ready = true; - - assert!(wot.is_outside(&carol)); - assert!(!wot.is_outside(&alice)); - - // A disabled graph marks nothing as outside - wot.disabled = true; - - assert!(!wot.is_outside(&carol)); - } - - #[test] - fn caps_ingested_contact_lists() { - let mut wot = registry(); - let (me, alice) = (key(1), key(2)); - - wot.self_key = Some(me); - wot.store_follows(Some(Timestamp::from_secs(10)), HashSet::from([alice])); - - let tags: Vec = (3..(MAX_TAGS_PER_LIST + 4)) - .map(|seed| Tag::public_key(key(seed))) - .collect(); - - let keys = Keys::new( - SecretKey::from_slice(&{ - let mut bytes = [0u8; 32]; - bytes[31] = 2; - bytes - }) - .expect("valid secret key"), - ); - - let event = EventBuilder::new(Kind::ContactList, "") - .tags(tags) - .finalize(&keys) - .expect("valid event"); - - let contacts = wot.parse_contacts(&event); - - assert_eq!(contacts.len(), MAX_TAGS_PER_LIST); - assert!(!contacts.contains(&key(MAX_TAGS_PER_LIST + 3))); - } + client + .database() + .query(filter) + .await + .unwrap_or_default() + .into_iter() + .next() }