local thread subscriptions

This adds local nostrdb thread subscriptions. When navigating to a
thread, we first check to see if we have any active nostrdb
subscriptions for that thread. If not, we create a new subscription. If
we do, we re-use that subscription.

This works by storing thread state in the Threads struct in the Damus
application state.

When we pop a route, we check to see if its a thread route. If it is,
then we try to unsubscribe, but only if that is the last remaining
subscriber for that thread, as there could be more than one.

Signed-off-by: William Casarin <jb55@jb55.com>

Cargo.lock

@@ -2296,7 +2296,7 @@ dependencies = [
 [[package]]
 name = "nostrdb"
 version = "0.3.4"
-source = "git+https://github.com/damus-io/nostrdb-rs?rev=8ef4b9c26145572ad7543d955778499e84723099#8ef4b9c26145572ad7543d955778499e84723099"
+source = "git+https://github.com/damus-io/nostrdb-rs?branch=threads#27e7c19c8941fe996490a82512fd2660e5da1900"
 dependencies = [
  "bindgen",
  "cc",

Cargo.toml

@@ -33,7 +33,8 @@ serde_json = "1.0.89"
 env_logger = "0.10.0"
 puffin_egui = { version = "0.27.0", optional = true }
 puffin = { version = "0.19.0", optional = true }
-nostrdb = { git = "https://github.com/damus-io/nostrdb-rs", rev = "8ef4b9c26145572ad7543d955778499e84723099" }
+nostrdb = { git = "https://github.com/damus-io/nostrdb-rs", branch = "threads" }
+#nostrdb = { path = "/Users/jb55/dev/github/damus-io/nostrdb-rs" }
 #nostrdb = "0.3.4"
 hex = "0.4.3"
 base32 = "0.4.0"

src/actionbar.rs

@@ -1,5 +1,7 @@
-use crate::{route::Route, Damus};
+use crate::{route::Route, thread::Thread, Damus};
 use enostr::NoteId;
+use nostrdb::Transaction;
+use tracing::{info, warn};
 
 #[derive(Debug, Eq, PartialEq, Copy, Clone)]
 pub enum BarAction {
@@ -8,7 +10,13 @@ pub enum BarAction {
 }
 
 impl BarAction {
-    pub fn execute(self, app: &mut Damus, timeline: usize, replying_to: &[u8; 32]) {
+    pub fn execute(
+        self,
+        app: &mut Damus,
+        timeline: usize,
+        replying_to: &[u8; 32],
+        txn: &Transaction,
+    ) {
         match self {
             BarAction::Reply => {
                 let timeline = &mut app.timelines[timeline];
@@ -19,11 +27,44 @@ impl BarAction {
             }
 
             BarAction::OpenThread => {
-                let timeline = &mut app.timelines[timeline];
-                timeline
-                    .routes
-                    .push(Route::Thread(NoteId::new(replying_to.to_owned())));
-                timeline.navigating = true;
+                {
+                    let timeline = &mut app.timelines[timeline];
+                    timeline
+                        .routes
+                        .push(Route::Thread(NoteId::new(replying_to.to_owned())));
+                    timeline.navigating = true;
+                }
+
+                let root_id = crate::note::root_note_id_from_selected_id(app, txn, replying_to);
+                let thread = app.threads.thread_mut(&app.ndb, txn, root_id);
+
+                // only start a subscription on nav and if we don't have
+                // an active subscription for this thread.
+                if thread.subscription().is_none() {
+                    *thread.subscription_mut() = app.ndb.subscribe(Thread::filters(root_id)).ok();
+
+                    match thread.subscription() {
+                        Some(_sub) => {
+                            thread.subscribers += 1;
+                            info!(
+                                "Locally subscribing to thread. {} total active subscriptions, {} on this thread",
+                                app.ndb.subscription_count(),
+                                thread.subscribers,
+                            );
+                        }
+                        None => warn!(
+                            "Error subscribing locally to selected note '{}''s thread",
+                            hex::encode(replying_to)
+                        ),
+                    }
+                } else {
+                    thread.subscribers += 1;
+                    info!(
+                        "Re-using existing thread subscription. {} total active subscriptions, {} on this thread",
+                        app.ndb.subscription_count(),
+                        thread.subscribers,
+                    )
+                }
             }
         }
     }

src/app.rs

@@ -2,7 +2,6 @@ use crate::account_manager::AccountManager;
 use crate::app_creation::setup_cc;
 use crate::app_style::user_requested_visuals_change;
 use crate::draft::Drafts;
-use crate::error::Error;
 use crate::frame_history::FrameHistory;
 use crate::imgcache::ImageCache;
 use crate::key_storage::KeyStorageType;
@@ -10,9 +9,9 @@ use crate::note::NoteRef;
 use crate::notecache::{CachedNote, NoteCache};
 use crate::relay_pool_manager::RelayPoolManager;
 use crate::route::Route;
-use crate::thread::Threads;
+use crate::thread::{DecrementResult, Threads};
 use crate::timeline;
-use crate::timeline::{MergeKind, Timeline, ViewFilter};
+use crate::timeline::{Timeline, TimelineSource, ViewFilter};
 use crate::ui::note::PostAction;
 use crate::ui::{self, AccountSelectionWidget, DesktopGlobalPopup};
 use crate::ui::{DesktopSidePanel, RelayView, View};
@@ -231,7 +230,8 @@ fn try_process_event(damus: &mut Damus, ctx: &egui::Context) -> Result<()> {
     let txn = Transaction::new(&damus.ndb)?;
     let mut unknown_ids: HashSet<UnknownId> = HashSet::new();
     for timeline in 0..damus.timelines.len() {
-        if let Err(err) = poll_notes_for_timeline(damus, &txn, timeline, &mut unknown_ids) {
+        let src = TimelineSource::column(timeline);
+        if let Err(err) = src.poll_notes_into_view(damus, &txn, &mut unknown_ids) {
             error!("{}", err);
         }
     }
@@ -250,7 +250,7 @@ fn try_process_event(damus: &mut Damus, ctx: &egui::Context) -> Result<()> {
 }
 
 #[derive(Hash, Clone, Copy, PartialEq, Eq)]
-enum UnknownId<'a> {
+pub enum UnknownId<'a> {
     Pubkey(&'a [u8; 32]),
     Id(&'a [u8; 32]),
 }
@@ -271,7 +271,7 @@ impl<'a> UnknownId<'a> {
     }
 }
 
-fn get_unknown_note_ids<'a>(
+pub fn get_unknown_note_ids<'a>(
     ndb: &Ndb,
     _cached_note: &CachedNote,
     txn: &'a Transaction,
@@ -354,103 +354,6 @@ fn get_unknown_note_ids<'a>(
     Ok(())
 }
 
-fn poll_notes_for_timeline<'a>(
-    damus: &mut Damus,
-    txn: &'a Transaction,
-    timeline_ind: usize,
-    ids: &mut HashSet<UnknownId<'a>>,
-) -> Result<()> {
-    let sub = if let Some(sub) = &damus.timelines[timeline_ind].subscription {
-        sub
-    } else {
-        return Err(Error::NoActiveSubscription);
-    };
-
-    let new_note_ids = damus.ndb.poll_for_notes(sub, 100);
-    if new_note_ids.is_empty() {
-        return Ok(());
-    } else {
-        debug!("{} new notes! {:?}", new_note_ids.len(), new_note_ids);
-    }
-
-    let new_refs: Vec<(Note, NoteRef)> = new_note_ids
-        .iter()
-        .map(|key| {
-            let note = damus.ndb.get_note_by_key(txn, *key).expect("no note??");
-            let cached_note = damus
-                .note_cache_mut()
-                .cached_note_or_insert(*key, &note)
-                .clone();
-            let _ = get_unknown_note_ids(&damus.ndb, &cached_note, txn, &note, *key, ids);
-
-            let created_at = note.created_at();
-            (
-                note,
-                NoteRef {
-                    key: *key,
-                    created_at,
-                },
-            )
-        })
-        .collect();
-
-    // ViewFilter::NotesAndReplies
-    {
-        let refs: Vec<NoteRef> = new_refs.iter().map(|(_note, nr)| *nr).collect();
-
-        insert_notes_into_timeline(damus, timeline_ind, ViewFilter::NotesAndReplies, &refs)
-    }
-
-    //
-    // handle the filtered case (ViewFilter::Notes, no replies)
-    //
-    // TODO(jb55): this is mostly just copied from above, let's just use a loop
-    //             I initially tried this but ran into borrow checker issues
-    {
-        let mut filtered_refs = Vec::with_capacity(new_refs.len());
-        for (note, nr) in &new_refs {
-            let cached_note = damus.note_cache_mut().cached_note_or_insert(nr.key, note);
-
-            if ViewFilter::filter_notes(cached_note, note) {
-                filtered_refs.push(*nr);
-            }
-        }
-
-        insert_notes_into_timeline(damus, timeline_ind, ViewFilter::Notes, &filtered_refs);
-    }
-
-    Ok(())
-}
-
-fn insert_notes_into_timeline(
-    app: &mut Damus,
-    timeline_ind: usize,
-    filter: ViewFilter,
-    new_refs: &[NoteRef],
-) {
-    let timeline = &mut app.timelines[timeline_ind];
-    let num_prev_items = timeline.notes(filter).len();
-    let (notes, merge_kind) = timeline::merge_sorted_vecs(timeline.notes(filter), new_refs);
-    debug!(
-        "got merge kind {:?} for {:?} on timeline {}",
-        merge_kind, filter, timeline_ind
-    );
-
-    timeline.view_mut(filter).notes = notes;
-    let new_items = timeline.notes(filter).len() - num_prev_items;
-
-    // TODO: technically items could have been added inbetween
-    if new_items > 0 {
-        let mut list = app.timelines[timeline_ind].view(filter).list.borrow_mut();
-
-        match merge_kind {
-            // TODO: update egui_virtual_list to support spliced inserts
-            MergeKind::Spliced => list.reset(),
-            MergeKind::FrontInsert => list.items_inserted_at_start(new_items),
-        }
-    }
-}
-
 #[cfg(feature = "profiling")]
 fn setup_profiling() {
     puffin::set_scopes_on(true); // tell puffin to collect data
@@ -787,7 +690,7 @@ impl Damus {
             // TODO: should pull this from settings
             None,
             // TODO: use correct KeyStorage mechanism for current OS arch
-            determine_key_storage_type(),
+            KeyStorageType::None,
         );
 
         for key in parsed_args.keys {
@@ -996,6 +899,50 @@ fn render_panel(ctx: &egui::Context, app: &mut Damus, timeline_ind: usize) {
     });
 }
 
+/// Local thread unsubscribe
+fn thread_unsubscribe(app: &mut Damus, id: &[u8; 32]) {
+    let unsubscribe = {
+        let txn = Transaction::new(&app.ndb).expect("txn");
+        let root_id = crate::note::root_note_id_from_selected_id(app, &txn, id);
+
+        debug!("thread unsubbing from root_id {}", hex::encode(root_id));
+
+        app.threads
+            .thread_mut(&app.ndb, &txn, root_id)
+            .decrement_sub()
+    };
+
+    match unsubscribe {
+        Ok(DecrementResult::LastSubscriber(sub_id)) => {
+            if let Err(e) = app.ndb.unsubscribe(sub_id) {
+                error!("failed to unsubscribe from thread: {e}, subid:{sub_id}, {} active subscriptions", app.ndb.subscription_count());
+            } else {
+                info!(
+                    "Unsubscribed from thread subid:{}. {} active subscriptions",
+                    sub_id,
+                    app.ndb.subscription_count()
+                );
+            }
+        }
+
+        Ok(DecrementResult::ActiveSubscribers) => {
+            info!(
+                "Keeping thread subscription. {} active subscriptions.",
+                app.ndb.subscription_count()
+            );
+            // do nothing
+        }
+
+        Err(e) => {
+            // something is wrong!
+            error!(
+                "Thread unsubscribe error: {e}. {} active subsciptions.",
+                app.ndb.subscription_count()
+            );
+        }
+    }
+}
+
 fn render_nav(routes: Vec<Route>, timeline_ind: usize, app: &mut Damus, ui: &mut egui::Ui) {
     let navigating = app.timelines[timeline_ind].navigating;
     let returning = app.timelines[timeline_ind].returning;
@@ -1026,12 +973,7 @@ fn render_nav(routes: Vec<Route>, timeline_ind: usize, app: &mut Damus, ui: &mut
 
             Route::Thread(id) => {
                 let app = &mut app_ctx.borrow_mut();
-                if let Ok(txn) = Transaction::new(&app.ndb) {
-                    if let Ok(note) = app.ndb.get_note_by_id(&txn, id.bytes()) {
-                        ui::ThreadView::new(app, timeline_ind, &note).ui(ui);
-                    }
-                }
-
+                ui::ThreadView::new(app, timeline_ind, id.bytes()).ui(ui);
                 None
             }
 
@@ -1063,18 +1005,21 @@ fn render_nav(routes: Vec<Route>, timeline_ind: usize, app: &mut Damus, ui: &mut
             }
         });
 
+    let mut app = app_ctx.borrow_mut();
     if let Some(reply_response) = nav_response.inner {
         if let Some(PostAction::Post(_np)) = reply_response.inner.action {
-            app_ctx.borrow_mut().timelines[timeline_ind].returning = true;
+            app.timelines[timeline_ind].returning = true;
         }
     }
 
     if let Some(NavAction::Returned) = nav_response.action {
-        let mut app = app_ctx.borrow_mut();
-        app.timelines[timeline_ind].routes.pop();
+        let popped = app.timelines[timeline_ind].routes.pop();
+        if let Some(Route::Thread(id)) = popped {
+            thread_unsubscribe(&mut app, id.bytes());
+        }
         app.timelines[timeline_ind].returning = false;
     } else if let Some(NavAction::Navigated) = nav_response.action {
-        app_ctx.borrow_mut().timelines[timeline_ind].navigating = false;
+        app.timelines[timeline_ind].navigating = false;
     }
 }
 

src/error.rs

@@ -1,8 +1,41 @@
 use std::{fmt, io};
 
+#[derive(Debug, Eq, PartialEq, Copy, Clone)]
+pub enum SubscriptionError {
+    //#[error("No active subscriptions")]
+    NoActive,
+
+    /// When a timeline has an unexpected number
+    /// of active subscriptions. Should only happen if there
+    /// is a bug in notedeck
+    //#[error("Unexpected subscription count")]
+    UnexpectedSubscriptionCount(i32),
+}
+
+impl Error {
+    pub fn unexpected_sub_count(c: i32) -> Self {
+        Error::SubscriptionError(SubscriptionError::UnexpectedSubscriptionCount(c))
+    }
+
+    pub fn no_active_sub() -> Self {
+        Error::SubscriptionError(SubscriptionError::NoActive)
+    }
+}
+
+impl fmt::Display for SubscriptionError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::NoActive => write!(f, "No active subscriptions"),
+            Self::UnexpectedSubscriptionCount(c) => {
+                write!(f, "Unexpected subscription count: {}", c)
+            }
+        }
+    }
+}
+
 #[derive(Debug)]
 pub enum Error {
-    NoActiveSubscription,
+    SubscriptionError(SubscriptionError),
     LoadFailed,
     Io(io::Error),
     Nostr(enostr::Error),
@@ -14,8 +47,8 @@ pub enum Error {
 impl fmt::Display for Error {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            Self::NoActiveSubscription => {
-                write!(f, "subscription not active in timeline")
+            Self::SubscriptionError(sub_err) => {
+                write!(f, "{sub_err}")
             }
             Self::LoadFailed => {
                 write!(f, "load failed")

src/note.rs

@@ -1,4 +1,5 @@
-use nostrdb::{NoteKey, QueryResult};
+use crate::Damus;
+use nostrdb::{NoteKey, QueryResult, Transaction};
 use std::cmp::Ordering;
 
 #[derive(Debug, Eq, PartialEq, Copy, Clone)]
@@ -35,3 +36,32 @@ impl PartialOrd for NoteRef {
         Some(self.cmp(other))
     }
 }
+
+pub fn root_note_id_from_selected_id<'a>(
+    app: &mut Damus,
+    txn: &'a Transaction,
+    selected_note_id: &'a [u8; 32],
+) -> &'a [u8; 32] {
+    let selected_note_key = if let Ok(key) = app
+        .ndb
+        .get_notekey_by_id(txn, selected_note_id)
+        .map(NoteKey::new)
+    {
+        key
+    } else {
+        return selected_note_id;
+    };
+
+    let note = if let Ok(note) = app.ndb.get_note_by_key(txn, selected_note_key) {
+        note
+    } else {
+        return selected_note_id;
+    };
+
+    app.note_cache_mut()
+        .cached_note_or_insert(selected_note_key, &note)
+        .reply
+        .borrow(note.tags())
+        .root()
+        .map_or_else(|| selected_note_id, |nr| nr.id)
+}

src/thread.rs

@@ -1,12 +1,21 @@
 use crate::note::NoteRef;
 use crate::timeline::{TimelineView, ViewFilter};
-use nostrdb::{Ndb, Transaction};
+use crate::Error;
+use nostrdb::{Filter, Ndb, Subscription, Transaction};
 use std::collections::HashMap;
 use tracing::debug;
 
 #[derive(Default)]
 pub struct Thread {
     pub view: TimelineView,
+    sub: Option<Subscription>,
+    pub subscribers: i32,
+}
+
+#[derive(Debug, Eq, PartialEq, Copy, Clone)]
+pub enum DecrementResult {
+    LastSubscriber(u64),
+    ActiveSubscribers,
 }
 
 impl Thread {
@@ -17,24 +26,79 @@ impl Thread {
         }
         let mut view = TimelineView::new_with_capacity(ViewFilter::NotesAndReplies, cap);
         view.notes = notes;
+        let sub: Option<Subscription> = None;
+        let subscribers: i32 = 0;
 
-        Thread { view }
+        Thread {
+            view,
+            sub,
+            subscribers,
+        }
+    }
+
+    pub fn decrement_sub(&mut self) -> Result<DecrementResult, Error> {
+        debug!("decrementing sub {:?}", self.subscription().map(|s| s.id));
+        self.subscribers -= 1;
+        if self.subscribers == 0 {
+            // unsub from thread
+            if let Some(sub) = self.subscription() {
+                Ok(DecrementResult::LastSubscriber(sub.id))
+            } else {
+                Err(Error::no_active_sub())
+            }
+        } else if self.subscribers < 0 {
+            Err(Error::unexpected_sub_count(self.subscribers))
+        } else {
+            Ok(DecrementResult::ActiveSubscribers)
+        }
+    }
+
+    pub fn subscription(&self) -> Option<&Subscription> {
+        self.sub.as_ref()
+    }
+
+    pub fn subscription_mut(&mut self) -> &mut Option<Subscription> {
+        &mut self.sub
+    }
+}
+
+impl Thread {
+    pub fn filters(root: &[u8; 32]) -> Vec<Filter> {
+        vec![
+            nostrdb::Filter::new().kinds(vec![1]).event(root).build(),
+            nostrdb::Filter::new()
+                .kinds(vec![1])
+                .ids(vec![*root])
+                .build(),
+        ]
     }
 }
 
 #[derive(Default)]
 pub struct Threads {
-    threads: HashMap<[u8; 32], Thread>,
+    /// root id to thread
+    pub root_id_to_thread: HashMap<[u8; 32], Thread>,
 }
 
 impl Threads {
-    pub fn thread_mut(&mut self, ndb: &Ndb, txn: &Transaction, root_id: &[u8; 32]) -> &mut Thread {
+    pub fn thread_expected_mut(&mut self, root_id: &[u8; 32]) -> &mut Thread {
+        self.root_id_to_thread
+            .get_mut(root_id)
+            .expect("thread_expected_mut used but there was no thread")
+    }
+
+    pub fn thread_mut<'a>(
+        &mut self,
+        ndb: &Ndb,
+        txn: &Transaction,
+        root_id: &[u8; 32],
+    ) -> &mut Thread {
         // we can't use the naive hashmap entry API here because lookups
         // require a copy, wait until we have a raw entry api. We could
         // also use hashbrown?
 
-        if self.threads.contains_key(root_id) {
-            return self.threads.get_mut(root_id).unwrap();
+        if self.root_id_to_thread.contains_key(root_id) {
+            return self.root_id_to_thread.get_mut(root_id).unwrap();
         }
 
         // looks like we don't have this thread yet, populate it
@@ -43,24 +107,16 @@ impl Threads {
             root
         } else {
             debug!("couldnt find root note for id {}", hex::encode(root_id));
-            self.threads.insert(root_id.to_owned(), Thread::new(vec![]));
-            return self.threads.get_mut(root_id).unwrap();
+            self.root_id_to_thread
+                .insert(root_id.to_owned(), Thread::new(vec![]));
+            return self.root_id_to_thread.get_mut(root_id).unwrap();
         };
 
         // we don't have the thread, query for it!
-        let filter = vec![
-            nostrdb::Filter::new()
-                .kinds(vec![1])
-                .event(root.id())
-                .build(),
-            nostrdb::Filter::new()
-                .kinds(vec![1])
-                .ids(vec![*root.id()])
-                .build(),
-        ];
+        let filters = Thread::filters(root_id);
 
         // TODO: what should be the max results ?
-        let notes = if let Ok(mut results) = ndb.query(txn, filter, 10000) {
+        let notes = if let Ok(mut results) = ndb.query(txn, filters, 10000) {
             results.reverse();
             results
                 .into_iter()
@@ -75,8 +131,9 @@ impl Threads {
         };
 
         debug!("found thread with {} notes", notes.len());
-        self.threads.insert(root_id.to_owned(), Thread::new(notes));
-        self.threads.get_mut(root_id).unwrap()
+        self.root_id_to_thread
+            .insert(root_id.to_owned(), Thread::new(notes));
+        self.root_id_to_thread.get_mut(root_id).unwrap()
     }
 
     //fn thread_by_id(&self, ndb: &Ndb, id: &[u8; 32]) -> &mut Thread {

src/timeline.rs

@@ -1,8 +1,10 @@
+use crate::app::{get_unknown_note_ids, UnknownId};
 use crate::draft::DraftSource;
+use crate::error::Error;
 use crate::note::NoteRef;
 use crate::notecache::CachedNote;
 use crate::ui::note::PostAction;
-use crate::{ui, Damus};
+use crate::{ui, Damus, Result};
 
 use crate::route::Route;
 use egui::containers::scroll_area::ScrollBarVisibility;
@@ -13,10 +15,137 @@ use egui_virtual_list::VirtualList;
 use enostr::Filter;
 use nostrdb::{Note, Subscription, Transaction};
 use std::cell::RefCell;
+use std::collections::HashSet;
 use std::rc::Rc;
 
 use tracing::{debug, info, warn};
 
+#[derive(Debug, Copy, Clone)]
+pub enum TimelineSource<'a> {
+    Column { ind: usize },
+    Thread(&'a [u8; 32]),
+}
+
+impl<'a> TimelineSource<'a> {
+    pub fn column(ind: usize) -> Self {
+        TimelineSource::Column { ind }
+    }
+
+    pub fn view<'b>(
+        self,
+        app: &'b mut Damus,
+        txn: &Transaction,
+        filter: ViewFilter,
+    ) -> &'b mut TimelineView {
+        match self {
+            TimelineSource::Column { ind, .. } => app.timelines[ind].view_mut(filter),
+            TimelineSource::Thread(root_id) => {
+                // TODO: replace all this with the raw entry api eventually
+
+                let thread = if app.threads.root_id_to_thread.contains_key(root_id) {
+                    app.threads.thread_expected_mut(root_id)
+                } else {
+                    app.threads.thread_mut(&app.ndb, txn, root_id)
+                };
+
+                &mut thread.view
+            }
+        }
+    }
+
+    pub fn sub<'b>(self, app: &'b mut Damus, txn: &Transaction) -> Option<&'b Subscription> {
+        match self {
+            TimelineSource::Column { ind, .. } => app.timelines[ind].subscription.as_ref(),
+            TimelineSource::Thread(root_id) => {
+                // TODO: replace all this with the raw entry api eventually
+
+                let thread = if app.threads.root_id_to_thread.contains_key(root_id) {
+                    app.threads.thread_expected_mut(root_id)
+                } else {
+                    app.threads.thread_mut(&app.ndb, txn, root_id)
+                };
+
+                thread.subscription()
+            }
+        }
+    }
+
+    pub fn poll_notes_into_view(
+        &self,
+        app: &mut Damus,
+        txn: &'a Transaction,
+        ids: &mut HashSet<UnknownId<'a>>,
+    ) -> Result<()> {
+        let sub_id = if let Some(sub_id) = self.sub(app, txn).map(|s| s.id) {
+            sub_id
+        } else {
+            return Err(Error::no_active_sub());
+        };
+
+        //
+        // TODO(BUG!): poll for these before the txn, otherwise we can hit
+        // a race condition where we hit the "no note??" expect below. This may
+        // require some refactoring due to the missing ids logic
+        //
+        let new_note_ids = app.ndb.poll_for_notes(sub_id, 100);
+        if new_note_ids.is_empty() {
+            return Ok(());
+        } else {
+            debug!("{} new notes! {:?}", new_note_ids.len(), new_note_ids);
+        }
+
+        let new_refs: Vec<(Note, NoteRef)> = new_note_ids
+            .iter()
+            .map(|key| {
+                let note = app.ndb.get_note_by_key(txn, *key).expect("no note??");
+                let cached_note = app
+                    .note_cache_mut()
+                    .cached_note_or_insert(*key, &note)
+                    .clone();
+                let _ = get_unknown_note_ids(&app.ndb, &cached_note, txn, &note, *key, ids);
+
+                let created_at = note.created_at();
+                (
+                    note,
+                    NoteRef {
+                        key: *key,
+                        created_at,
+                    },
+                )
+            })
+            .collect();
+
+        // ViewFilter::NotesAndReplies
+        {
+            let refs: Vec<NoteRef> = new_refs.iter().map(|(_note, nr)| *nr).collect();
+
+            self.view(app, txn, ViewFilter::NotesAndReplies)
+                .insert(&refs);
+        }
+
+        //
+        // handle the filtered case (ViewFilter::Notes, no replies)
+        //
+        // TODO(jb55): this is mostly just copied from above, let's just use a loop
+        //             I initially tried this but ran into borrow checker issues
+        {
+            let mut filtered_refs = Vec::with_capacity(new_refs.len());
+            for (note, nr) in &new_refs {
+                let cached_note = app.note_cache_mut().cached_note_or_insert(nr.key, note);
+
+                if ViewFilter::filter_notes(cached_note, note) {
+                    filtered_refs.push(*nr);
+                }
+            }
+
+            self.view(app, txn, ViewFilter::Notes)
+                .insert(&filtered_refs);
+        }
+
+        Ok(())
+    }
+}
+
 #[derive(Copy, Clone, Eq, PartialEq, Debug, Default)]
 pub enum ViewFilter {
     Notes,
@@ -88,6 +217,25 @@ impl TimelineView {
         }
     }
 
+    pub fn insert(&mut self, new_refs: &[NoteRef]) {
+        let num_prev_items = self.notes.len();
+        let (notes, merge_kind) = crate::timeline::merge_sorted_vecs(&self.notes, new_refs);
+
+        self.notes = notes;
+        let new_items = self.notes.len() - num_prev_items;
+
+        // TODO: technically items could have been added inbetween
+        if new_items > 0 {
+            let mut list = self.list.borrow_mut();
+
+            match merge_kind {
+                // TODO: update egui_virtual_list to support spliced inserts
+                MergeKind::Spliced => list.reset(),
+                MergeKind::FrontInsert => list.items_inserted_at_start(new_items),
+            }
+        }
+    }
+
     pub fn select_down(&mut self) {
         debug!("select_down {}", self.selection + 1);
         if self.selection + 1 > self.notes.len() as i32 {
@@ -335,7 +483,7 @@ pub fn timeline_view(ui: &mut egui::Ui, app: &mut Damus, timeline: usize) {
                             .show(ui);
 
                         if let Some(action) = resp.action {
-                            action.execute(app, timeline, note.id());
+                            action.execute(app, timeline, note.id(), &txn);
                         } else if resp.response.clicked() {
                             debug!("clicked note");
                         }

src/ui/thread.rs

@@ -1,47 +1,86 @@
-use crate::{ui, Damus};
-use nostrdb::{Note, NoteReply};
+use crate::{timeline::TimelineSource, ui, Damus};
+use nostrdb::{NoteKey, Transaction};
+use std::collections::HashSet;
 use tracing::warn;
 
 pub struct ThreadView<'a> {
     app: &'a mut Damus,
     timeline: usize,
-    selected_note: &'a Note<'a>,
+    selected_note_id: &'a [u8; 32],
 }
 
 impl<'a> ThreadView<'a> {
-    pub fn new(app: &'a mut Damus, timeline: usize, selected_note: &'a Note<'a>) -> Self {
+    pub fn new(app: &'a mut Damus, timeline: usize, selected_note_id: &'a [u8; 32]) -> Self {
         ThreadView {
             app,
             timeline,
-            selected_note,
+            selected_note_id,
         }
     }
 
     pub fn ui(&mut self, ui: &mut egui::Ui) {
-        let txn = self.selected_note.txn().unwrap();
-        let key = self.selected_note.key().unwrap();
+        let txn = Transaction::new(&self.app.ndb).expect("txn");
+
+        let selected_note_key = if let Ok(key) = self
+            .app
+            .ndb
+            .get_notekey_by_id(&txn, self.selected_note_id)
+            .map(NoteKey::new)
+        {
+            key
+        } else {
+            // TODO: render 404 ?
+            return;
+        };
+
         let scroll_id = egui::Id::new((
             "threadscroll",
             self.app.timelines[self.timeline].selected_view,
             self.timeline,
-            key,
+            selected_note_key,
         ));
+
         ui.label(
             egui::RichText::new("Threads ALPHA! It's not done. Things will be broken.")
                 .color(egui::Color32::RED),
         );
+
         egui::ScrollArea::vertical()
             .id_source(scroll_id)
             .animated(false)
             .auto_shrink([false, false])
             .scroll_bar_visibility(egui::scroll_area::ScrollBarVisibility::AlwaysVisible)
             .show(ui, |ui| {
-                let root_id = NoteReply::new(self.selected_note.tags())
-                    .root()
-                    .map_or_else(|| self.selected_note.id(), |nr| nr.id);
+                let note = if let Ok(note) = self.app.ndb.get_note_by_key(&txn, selected_note_key) {
+                    note
+                } else {
+                    return;
+                };
+
+                let root_id = {
+                    let cached_note = self
+                        .app
+                        .note_cache_mut()
+                        .cached_note_or_insert(selected_note_key, &note);
+
+                    cached_note
+                        .reply
+                        .borrow(note.tags())
+                        .root()
+                        .map_or_else(|| self.selected_note_id, |nr| nr.id)
+                };
+
+                // poll for new notes and insert them into our existing notes
+                {
+                    let mut ids = HashSet::new();
+                    let _ = TimelineSource::Thread(root_id)
+                        .poll_notes_into_view(self.app, &txn, &mut ids);
+                    // TODO: do something with unknown ids
+                }
 
                 let (len, list) = {
-                    let thread = self.app.threads.thread_mut(&self.app.ndb, txn, root_id);
+                    let thread = self.app.threads.thread_mut(&self.app.ndb, &txn, root_id);
+
                     let len = thread.view.notes.len();
                     (len, &mut thread.view.list)
                 };
@@ -53,11 +92,11 @@ impl<'a> ThreadView<'a> {
                         ui.spacing_mut().item_spacing.x = 4.0;
 
                         let note_key = {
-                            let thread = self.app.threads.thread_mut(&self.app.ndb, txn, root_id);
+                            let thread = self.app.threads.thread_mut(&self.app.ndb, &txn, root_id);
                             thread.view.notes[start_index].key
                         };
 
-                        let note = if let Ok(note) = self.app.ndb.get_note_by_key(txn, note_key) {
+                        let note = if let Ok(note) = self.app.ndb.get_note_by_key(&txn, note_key) {
                             note
                         } else {
                             warn!("failed to query note {:?}", note_key);
@@ -71,7 +110,7 @@ impl<'a> ThreadView<'a> {
                                 .show(ui);
 
                             if let Some(action) = resp.action {
-                                action.execute(self.app, self.timeline, note.id());
+                                action.execute(self.app, self.timeline, note.id(), &txn);
                             }
                         });