use std::collections::{HashMap, HashSet}; use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task}; use instant::Duration; use nostr_sdk::prelude::*; use smallvec::{SmallVec, smallvec}; use state::NostrRegistry; /// Disable scoring when the follow set exceeds this size 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 SYNC_INTERVAL: u64 = 15 * 60; struct GlobalWebOfTrust(Entity); impl Global for GlobalWebOfTrust {} /// A change to the web of trust #[derive(Debug, Clone)] pub enum WoTEvent { /// The graph changed Updated, /// The initial load finished Ready, } 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. #[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, /// Whether the next refresh completes the initial load initial: bool, /// Task loading the graph, replaced by newer loads hydration: Option>, /// Tasks for asynchronous operations tasks: SmallVec<[Task<()>; 4]>, /// Event subscriptions _subscriptions: SmallVec<[Subscription; 1]>, } impl WebOfTrust { /// Retrieve the global web of trust pub fn global(cx: &App) -> Entity { cx.global::().0.clone() } /// Set the global web of trust instance pub(crate) fn set_global(state: Entity, cx: &mut App) { cx.set_global(GlobalWebOfTrust(state)); } /// Create a new web of trust instance pub(crate) fn new(cx: &mut Context) -> Self { let nostr = NostrRegistry::global(cx); let entity = cx.entity().downgrade(); let mut tasks = smallvec![]; tasks.push(cx.spawn(async move |this, cx| { loop { cx.background_executor() .timer(Duration::from_secs(SYNC_INTERVAL)) .await; // Ignore errors, the next refresh will retry let _ = this.update(cx, |this, cx| { this.sync(cx); }); } })); let subscription = cx.subscribe(&nostr, |this, _nostr, event, cx| { if event.signer_changed() { this.reset(cx); this.hydrate(cx); } }); cx.defer(move |cx| { let _ = entity.update(cx, |this, cx| { this.hydrate(cx); }); }); Self { self_key: None, follows: HashSet::new(), follows_at: None, lists: HashMap::new(), reverse: HashMap::new(), disabled: false, ready: false, initial: false, hydration: None, tasks, _subscriptions: smallvec![subscription], } } /// Load the graph for the current account 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); self.initial = true; // 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::Updated); cx.notify(); } }) .ok(); // Reconcile with the relays; the refresh this triggers finalizes the initial load this.update(cx, |this, cx| { this.sync(cx); }) .ok(); })); } /// Reconcile the follow set's contact lists with the relays and apply whatever came back fn sync(&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; }; let covered: HashSet = self.follows.clone(); let mut authors: Vec = covered.iter().copied().collect(); authors.push(self_key); self.tasks.push(cx.spawn(async move |this, cx| { 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(); 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 mut grew = false; this.update(cx, |this, cx| { let changed = this.apply_latest(self_key, events, cx); if this.initial { this.initial = false; this.ready = true; cx.emit(WoTEvent::Ready); } if changed { cx.emit(WoTEvent::Updated); cx.notify(); } grew = !covered.is_superset(&this.follows); }) .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); }) .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 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.initial = false; self.hydration = None; cx.emit(WoTEvent::Updated); 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()) } /// 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 } /// 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 }) } /// Whether the initial load has finished pub fn is_ready(&self) -> bool { self.ready } /// Whether scoring is disabled due to an oversized follow set pub fn is_disabled(&self) -> bool { self.disabled } } #[cfg(test)] mod tests { use super::*; 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, initial: false, hydration: None, tasks: smallvec![], _subscriptions: 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 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))); } }