convert account relay list support to subman

Author: ksedgwic

crates/notedeck/src/accounts.rs

@@ -5,12 +5,11 @@ use crate::{
     KeyStorageResponse, KeyStorageType, MuteFun, Muted, RelaySpec, SingleUnkIdAction, SubError,
     SubMan, UnknownIds, UserAccount,
 };
-use enostr::{ClientMessage, FilledKeypair, Keypair, RelayPool};
-use nostrdb::{Filter, Ndb, Note, NoteBuilder, NoteKey, Subscription, Transaction};
+use enostr::{FilledKeypair, Keypair, RelayPool};
+use nostrdb::{Filter, Ndb, Note, NoteBuilder, NoteKey, Transaction};
 use std::cmp::Ordering;
 use std::collections::{BTreeMap, BTreeSet};
 use url::Url;
-use uuid::Uuid;
 
 // TODO: remove this
 use std::sync::{Arc, Mutex};
@@ -38,10 +37,9 @@ pub enum AccountsAction {
 
 pub struct AccountRelayData {
     filter: Filter,
-    subid: Option<String>,
-    sub: Option<Subscription>,
-    local: BTreeSet<RelaySpec>,      // used locally but not advertised
-    advertised: BTreeSet<RelaySpec>, // advertised via NIP-65
+    subrmtid: Option<RmtId>,
+    local: BTreeSet<RelaySpec>, // used locally but not advertised
+    advertised: Arc<Mutex<BTreeSet<RelaySpec>>>, // advertised via NIP-65
 }
 
 #[derive(Default)]
@@ -83,47 +81,64 @@ impl AccountRelayData {
 
         AccountRelayData {
             filter,
-            subid: None,
-            sub: None,
+            subrmtid: None,
             local: BTreeSet::new(),
-            advertised: relays.into_iter().collect(),
+            advertised: Arc::new(Mutex::new(relays.into_iter().collect())),
         }
     }
 
     // make this account the current selected account
-    pub fn activate(&mut self, ndb: &Ndb, pool: &mut RelayPool) {
+    pub fn activate(&mut self, subman: &mut SubMan, default_relays: &[RelaySpec]) {
         debug!("activating relay sub {}", self.filter.json().unwrap());
-        assert_eq!(self.subid, None, "subid already exists");
-        assert_eq!(self.sub, None, "sub already exists");
-
-        // local subscription
-        let sub = ndb
-            .subscribe(&[self.filter.clone()])
-            .expect("ndb relay list subscription");
-
-        // remote subscription
-        let subid = Uuid::new_v4().to_string();
-        pool.subscribe(subid.clone(), vec![self.filter.clone()]);
-
-        self.sub = Some(sub);
-        self.subid = Some(subid);
+        assert!(self.subrmtid.is_none(), "subscription already exists");
+        let ndb = subman.ndb();
+        let subspec = SubSpecBuilder::new()
+            .filters(vec![self.filter.clone()])
+            .build();
+        debug!(
+            "activating account relay sub {}: {}",
+            subspec.rmtid,
+            self.filter.json().unwrap()
+        );
+        if let Ok(mut rcvr) = subman.subscribe(subspec, default_relays) {
+            let idstr = rcvr.idstr();
+            self.subrmtid = rcvr.rmtid();
+            let advertisedref = self.advertised.clone();
+            tokio::spawn(async move {
+                loop {
+                    match rcvr.next().await {
+                        Err(SubError::StreamEnded) => {
+                            debug!("account relays: sub {} complete", idstr);
+                            break;
+                        }
+                        Err(err) => {
+                            error!("account relays: sub {}: error: {:?}", idstr, err);
+                            break;
+                        }
+                        Ok(nks) => {
+                            debug!("account relays: sub {}: note keys: {:?}", idstr, nks);
+                            let txn = Transaction::new(&ndb).expect("txn");
+                            let relays = Self::harvest_nip65_relays(&ndb, &txn, &nks);
+                            debug!("updated relays {:?}", relays);
+                            *advertisedref.lock().unwrap() = relays.into_iter().collect();
+                        }
+                    }
+                }
+            });
+        }
     }
 
     // this account is no longer the selected account
-    pub fn deactivate(&mut self, ndb: &mut Ndb, pool: &mut RelayPool) {
-        debug!("deactivating relay sub {}", self.filter.json().unwrap());
-        assert_ne!(self.subid, None, "subid doesn't exist");
-        assert_ne!(self.sub, None, "sub doesn't exist");
-
-        // remote subscription
-        pool.unsubscribe(self.subid.as_ref().unwrap().clone());
-
-        // local subscription
-        ndb.unsubscribe(self.sub.unwrap())
-            .expect("ndb relay list unsubscribe");
-
-        self.sub = None;
-        self.subid = None;
+    pub fn deactivate(&mut self, subman: &mut SubMan) {
+        assert!(self.subrmtid.is_some(), "subscription doesn't exist");
+        let rmtid = self.subrmtid.as_ref().unwrap();
+        debug!(
+            "deactivating account relays sub {}: {}",
+            rmtid,
+            self.filter.json().unwrap()
+        );
+        subman.unsubscribe_rmtid(rmtid).ok();
+        self.subrmtid = None;
     }
 
     // standardize the format (ie, trailing slashes) to avoid dups
@@ -173,7 +188,7 @@ impl AccountRelayData {
 
     pub fn publish_nip65_relays(&self, seckey: &[u8; 32], pool: &mut RelayPool) {
         let mut builder = NoteBuilder::new().kind(10002).content("");
-        for rs in &self.advertised {
+        for rs in self.advertised.lock().unwrap().iter() {
             builder = builder.start_tag().tag_str("r").tag_str(&rs.url);
             if rs.has_read_marker {
                 builder = builder.tag_str("read");
@@ -542,26 +557,6 @@ impl Accounts {
         Box::new(|_: &Note, _: &[u8; 32]| false)
     }
 
-    pub fn send_initial_filters(&mut self, pool: &mut RelayPool, relay_url: &str) {
-        for data in self.account_data.values() {
-            // send the active account's relay list subscription
-            if let Some(relay_subid) = &data.relay.subid {
-                let filters = vec![data.relay.filter.clone()];
-                debug!(
-                    "Account send_initial_filters: sending sub {} {}: {:?}",
-                    relay_subid,
-                    relay_url,
-                    filters
-                        .iter()
-                        .map(|f| f.json().unwrap_or_default())
-                        .collect::<Vec<_>>(),
-                );
-
-                pool.send_to(&ClientMessage::req(relay_subid.clone(), filters), relay_url);
-            }
-        }
-    }
-
     // Return accounts which have no account_data yet (added) and accounts
     // which have still data but are no longer in our account list (removed).
     fn delta_accounts(&self) -> (Vec<[u8; 32]>, Vec<[u8; 32]>) {
@@ -597,27 +592,6 @@ impl Accounts {
         self.account_data.remove(pubkey);
     }
 
-    fn poll_for_updates(&mut self, ndb: &Ndb) -> bool {
-        let mut changed = false;
-        for (pubkey, data) in &mut self.account_data {
-            if let Some(sub) = data.relay.sub {
-                let nks = ndb.poll_for_notes(sub, 1);
-                if !nks.is_empty() {
-                    let txn = Transaction::new(ndb).expect("txn");
-                    let relays = AccountRelayData::harvest_nip65_relays(ndb, &txn, &nks);
-                    debug!(
-                        "pubkey {}: updated relays {:?}",
-                        hex::encode(pubkey),
-                        relays
-                    );
-                    data.relay.advertised = relays.into_iter().collect();
-                    changed = true;
-                }
-            }
-        }
-        changed
-    }
-
     fn update_relay_configuration(
         &mut self,
         pool: &mut RelayPool,
@@ -636,7 +610,7 @@ impl Accounts {
         if desired_relays.is_empty() {
             if let Some(data) = self.get_selected_account_data() {
                 desired_relays.extend(data.relay.local.iter().cloned());
-                desired_relays.extend(data.relay.advertised.iter().cloned());
+                desired_relays.extend(data.relay.advertised.lock().unwrap().iter().cloned());
             }
         }
 
@@ -681,6 +655,8 @@ impl Accounts {
         let mut relays = if let Some(data) = data_option {
             data.relay
                 .advertised
+                .lock()
+                .unwrap()
                 .iter()
                 .filter(|&x| filter(x))
                 .cloned()
@@ -729,11 +705,9 @@ impl Accounts {
             if Some(ndx) != self.currently_selected_account {
                 // this account is not currently selected
                 if let Some(data) = self.account_data.get_mut(account.pubkey.bytes()) {
-                    if data.relay.sub.is_some() {
+                    if data.relay.subrmtid.is_some() {
                         // this account has relay subs, deactivate them
-                        let mut ndb = subman.ndb().clone();
-                        let pool = subman.pool();
-                        data.relay.deactivate(&mut ndb, pool);
+                        data.relay.deactivate(subman);
                     }
                     if data.muted.subrmtid.is_some() {
                         // this account has muted subs, deactivate them
@@ -757,9 +731,6 @@ impl Accounts {
             need_reconfig = true;
         }
 
-        // Did any accounts receive updates (ie NIP-65 relay lists)
-        need_reconfig = self.poll_for_updates(&ndb) || need_reconfig;
-
         // If needed, update the relay configuration
         if need_reconfig {
             self.update_relay_configuration(pool, wakeup);
@@ -769,9 +740,9 @@ impl Accounts {
         // Do we need to activate account subs?
         let default_relays = self.get_all_selected_account_relays();
         if let Some(data) = self.get_selected_account_data_mut() {
-            if data.relay.sub.is_none() {
+            if data.relay.subrmtid.is_none() {
                 // the currently selected account doesn't have relay subs, activate them
-                data.relay.activate(&ndb, pool);
+                data.relay.activate(subman, &default_relays);
             }
             if data.muted.subrmtid.is_none() {
                 // the currently selected account doesn't have muted subs, activate them
@@ -812,7 +783,7 @@ impl Accounts {
                     match self.account_data.get_mut(&key_bytes) {
                         None => error!("no account data found for the provided key."),
                         Some(account_data) => {
-                            let advertised = &mut account_data.relay.advertised;
+                            let advertised = &mut account_data.relay.advertised.lock().unwrap();
                             if advertised.is_empty() {
                                 // If the selected account has no advertised relays,
                                 // initialize with the bootstrapping set.

crates/notedeck_columns/src/app.rs

@@ -87,7 +87,6 @@ struct RelayHandler<'a> {
     // From AppContext
     unknown_ids: &'a mut UnknownIds,
     note_cache: &'a mut NoteCache,
-    accounts: &'a mut Accounts,
 
     // From Damus
     subscriptions: &'a mut Subscriptions,
@@ -100,14 +99,12 @@ impl<'a> RelayHandler<'a> {
     fn new(
         unknown_ids: &'a mut UnknownIds,
         note_cache: &'a mut NoteCache,
-        accounts: &'a mut Accounts,
         subscriptions: &'a mut Subscriptions,
         timeline_cache: &'a mut TimelineCache,
         since_optimize: bool,
     ) -> Self {
         RelayHandler {
             unknown_ids,
-            accounts,
             note_cache,
             subscriptions,
             timeline_cache,
@@ -119,7 +116,6 @@ impl<'a> RelayHandler<'a> {
 impl LegacyRelayHandler for RelayHandler<'_> {
     /// Handle relay opened
     fn handle_opened(&mut self, ndb: &mut Ndb, pool: &mut RelayPool, relay: &str) {
-        self.accounts.send_initial_filters(pool, relay);
         timeline::send_initial_timeline_filters(
             ndb,
             self.since_optimize,
@@ -159,7 +155,6 @@ fn try_process_event<'a>(
         let mut relay_handler = RelayHandler::new(
             app_ctx.unknown_ids,
             app_ctx.note_cache,
-            app_ctx.accounts,
             &mut damus.subscriptions,
             &mut damus.timeline_cache,
             damus.since_optimize,