WIP: use subman API to resolve unknown ids

Author: ksedgwic

crates/notedeck/src/lib.rs

@@ -47,7 +47,7 @@ pub use storage::{
     DataPath, DataPathType, Directory, FileKeyStorage, KeyStorageResponse, KeyStorageType,
 };
 pub use style::NotedeckTextStyle;
-pub use subman::SubMan;
+pub use subman::{SubError, SubMan};
 pub use theme::ColorTheme;
 pub use time::time_ago_since;
 pub use timecache::TimeCached;

crates/notedeck/src/unknowns.rs

@@ -1,12 +1,13 @@
 use crate::{
     note::NoteRef,
     notecache::{CachedNote, NoteCache},
+    subman::{SubConstraint, SubSpec, SubSpecBuilder},
     Result,
 };
 
 use enostr::{Filter, NoteId, Pubkey};
 use nostrdb::{BlockType, Mention, Ndb, Note, NoteKey, Transaction};
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::time::{Duration, Instant};
 use tracing::error;
 
@@ -118,13 +119,72 @@ impl UnknownIds {
         &mut self.ids
     }
 
+    pub fn numids(&self) -> usize {
+        self.ids.len()
+    }
+
     pub fn clear(&mut self) {
         self.ids = HashMap::default();
     }
 
-    pub fn filter(&self) -> Option<Vec<Filter>> {
-        let ids: Vec<&UnknownId> = self.ids.keys().collect();
-        get_unknown_ids_filter(&ids)
+    pub fn generate_resolution_requests(&self) -> Vec<SubSpec> {
+        // 1. resolve as many ids per request as possible
+        // 2. each request only has one filter (https://github.com/nostr-protocol/nips/pull/1645)
+        // 3. each request is limited to MAX_CHUNK_IDS
+        // 4. use relay hints when available
+
+        // Collect the unknown ids by relay
+        let mut ids_by_relay: BTreeMap<RelayUrl, (Vec<Pubkey>, Vec<NoteId>)> = BTreeMap::new();
+        for (unknown_id, relay_hints) in self.ids.iter() {
+            // 1. use default relays (empty RelayUrl) if no hints are available
+            // 2. query the default relays even when hints are available
+            for relay in std::iter::once("".to_string()).chain(relay_hints.iter().cloned()) {
+                match unknown_id {
+                    UnknownId::Pubkey(pk) => {
+                        ids_by_relay
+                            .entry(relay)
+                            .or_insert_with(|| (Vec::new(), Vec::new()))
+                            .0
+                            .push(*pk);
+                    }
+                    UnknownId::Id(nid) => {
+                        ids_by_relay
+                            .entry(relay)
+                            .or_insert_with(|| (Vec::new(), Vec::new()))
+                            .1
+                            .push(*nid);
+                    }
+                }
+            }
+        }
+
+        const MAX_CHUNK_IDS: usize = 500;
+
+        let mut subspecs = vec![];
+        for (relay, (pubkeys, noteids)) in ids_by_relay {
+            // make a template SubSpecBuilder w/ the common parts
+            let mut ssb = SubSpecBuilder::new().constraint(SubConstraint::OneShot);
+            if !relay.is_empty() {
+                ssb = ssb.constraint(SubConstraint::AllowedRelays(vec![relay]));
+            }
+            for chunk in pubkeys.chunks(MAX_CHUNK_IDS) {
+                let pks: Vec<&[u8; 32]> = chunk.iter().map(|pk| pk.bytes()).collect();
+                subspecs.push(
+                    ssb.clone()
+                        .filters(vec![Filter::new().authors(pks).kinds([0]).build()])
+                        .build(),
+                );
+            }
+            for chunk in noteids.chunks(MAX_CHUNK_IDS) {
+                let nids: Vec<&[u8; 32]> = chunk.iter().map(|nid| nid.bytes()).collect();
+                subspecs.push(
+                    ssb.clone()
+                        .filters(vec![Filter::new().ids(nids).build()])
+                        .build(),
+                );
+            }
+        }
+        subspecs
     }
 
     /// We've updated some unknown ids, update the last_updated time to now
@@ -350,31 +410,3 @@ pub fn get_unknown_note_ids<'a>(
 
     Ok(())
 }
-
-fn get_unknown_ids_filter(ids: &[&UnknownId]) -> Option<Vec<Filter>> {
-    if ids.is_empty() {
-        return None;
-    }
-
-    let ids = &ids[0..500.min(ids.len())];
-    let mut filters: Vec<Filter> = vec![];
-
-    let pks: Vec<&[u8; 32]> = ids
-        .iter()
-        .flat_map(|id| id.is_pubkey().map(|pk| pk.bytes()))
-        .collect();
-    if !pks.is_empty() {
-        let pk_filter = Filter::new().authors(pks).kinds([0]).build();
-        filters.push(pk_filter);
-    }
-
-    let note_ids: Vec<&[u8; 32]> = ids
-        .iter()
-        .flat_map(|id| id.is_id().map(|id| id.bytes()))
-        .collect();
-    if !note_ids.is_empty() {
-        filters.push(Filter::new().ids(note_ids).build());
-    }
-
-    Some(filters)
-}

crates/notedeck_columns/src/app.rs

@@ -13,9 +13,12 @@ use crate::{
     Result,
 };
 
-use notedeck::{Accounts, AppContext, DataPath, DataPathType, FilterState, ImageCache, UnknownIds};
+use notedeck::{
+    Accounts, AppContext, DataPath, DataPathType, FilterState, ImageCache, SubError, SubMan,
+    UnknownIds,
+};
 
-use enostr::{ClientMessage, Keypair, PoolRelay, Pubkey, RelayEvent, RelayMessage, RelayPool};
+use enostr::{ClientMessage, Keypair, PoolRelay, Pubkey, RelayEvent, RelayMessage};
 use uuid::Uuid;
 
 use egui_extras::{Size, StripBuilder};
@@ -157,22 +160,41 @@ fn try_process_event(
     }
 
     if app_ctx.unknown_ids.ready_to_send() {
-        unknown_id_send(app_ctx.unknown_ids, app_ctx.subman.pool());
+        unknown_id_send(app_ctx.unknown_ids, app_ctx.subman);
     }
 
     Ok(())
 }
 
-fn unknown_id_send(unknown_ids: &mut UnknownIds, pool: &mut RelayPool) {
-    debug!("unknown_id_send called on: {:?}", &unknown_ids);
-    let filter = unknown_ids.filter().expect("filter");
-    info!(
-        "Getting {} unknown ids from relays",
-        unknown_ids.ids_iter().len()
-    );
-    let msg = ClientMessage::req("unknownids".to_string(), filter);
+fn unknown_id_send(unknown_ids: &mut UnknownIds, subman: &mut SubMan) {
+    info!("Getting {} unknown ids from relays", &unknown_ids.numids());
+    for subspec in unknown_ids.generate_resolution_requests() {
+        debug!("unknown_ids subscribe: {:?}", subspec);
+        match subman.subscribe(subspec) {
+            Err(err) => error!("unknown_id_send subscribe failed: {:?}", err),
+            Ok(mut rcvr) => {
+                tokio::spawn(async move {
+                    loop {
+                        match rcvr.next().await {
+                            Err(SubError::StreamEnded) => {
+                                debug!("unknown_id_send: {} complete", rcvr.idstr());
+                                break;
+                            }
+                            Err(err) => {
+                                error!("unknown_id_send: {}: error: {:?}", rcvr.idstr(), err);
+                                break;
+                            }
+                            Ok(note_keys) => {
+                                debug!("received note keys: {:?}", note_keys);
+                                // only need the prefetch into ndb, all done
+                            }
+                        }
+                    }
+                });
+            }
+        }
+    }
     unknown_ids.clear();
-    pool.send(&msg);
 }
 
 fn update_damus(damus: &mut Damus, app_ctx: &mut AppContext<'_>, ctx: &egui::Context) {
@@ -233,7 +255,7 @@ fn handle_eose(
             );
             // this is possible if this is the first time
             if ctx.unknown_ids.ready_to_send() {
-                unknown_id_send(ctx.unknown_ids, ctx.subman.pool());
+                unknown_id_send(ctx.unknown_ids, ctx.subman);
             }
         }