From b7221ef8147689596655fda900929fe599495761 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Fri, 11 Sep 2026 08:52:04 +0700 Subject: [PATCH] refacotr --- Cargo.lock | 1 + crates/signed_core/src/inbox.rs | 80 +++++- crates/signed_state/Cargo.toml | 1 + crates/signed_state/src/backend.rs | 212 +++++++++----- crates/signed_state/src/inbox.rs | 431 +++++++++++++++++++++++++++++ crates/signed_state/src/lib.rs | 18 +- crates/signed_state/src/refresh.rs | 5 - docs/inbox-plan.md | 337 +++++++++++++++------- 8 files changed, 895 insertions(+), 190 deletions(-) create mode 100644 crates/signed_state/src/inbox.rs diff --git a/Cargo.lock b/Cargo.lock index 189ee38..9d23ca2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8048,6 +8048,7 @@ dependencies = [ "nostr-connect", "nostr-sdk", "rustls", + "serde_json", "settings", "signed_core", "signed_git", diff --git a/crates/signed_core/src/inbox.rs b/crates/signed_core/src/inbox.rs index 4a07e24..57aed3f 100644 --- a/crates/signed_core/src/inbox.rs +++ b/crates/signed_core/src/inbox.rs @@ -42,6 +42,19 @@ impl InboxItem { pub fn is_unread(&self) -> bool { !self.archived && !self.unread_ids.is_empty() } + + /// Recompute the unread and archived flags from `state`. + pub fn apply_state(&mut self, state: &InboxReadState) { + self.unread_ids = self + .events + .iter() + .rev() + .filter(|event| !state.is_read(event)) + .map(|event| event.id) + .collect(); + + self.archived = self.events.iter().all(|event| state.is_archived(event)); + } } /// Root issue, patch or pull request of a notification event. @@ -112,25 +125,18 @@ where let root_event = lookup(root); - let unread_ids = events - .iter() - .rev() - .filter(|event| !state.is_read(event)) - .map(|event| event.id) - .collect(); - - let archived = events.iter().all(|event| state.is_archived(event)); - - InboxItem { + let mut item = InboxItem { root, root_kind: root_event.as_ref().map(|event| event.kind), address: root_event .as_ref() .and_then(|event| event.tags.coordinates().next()), events, - unread_ids, - archived, - } + unread_ids: Vec::new(), + archived: false, + }; + item.apply_state(state); + item }) .collect(); @@ -174,6 +180,13 @@ impl InboxReadState { } } + /// Mark one event archived. Events at or before the cutoff are already archived. + pub fn mark_archived(&mut self, event: &Event) { + if event.created_at > self.archived_before { + self.archived_ids.insert(event.id); + } + } + /// Mark every non-self event read, anchoring the cutoff ten days back. pub fn mark_all_read(&mut self, all: &[Event], me: PublicKey, now: Timestamp) { let cutoff = now - MARK_ALL_WINDOW; @@ -626,6 +639,47 @@ mod tests { assert_eq!(state.read_ids, HashSet::from([fresh.id])); } + #[test] + fn mark_archived_skips_events_at_or_before_the_cutoff() { + let now = Timestamp::from_secs(1_000_000_000); + let event = issue(&keys(2), now.as_secs() - 1000); + + let mut state = InboxReadState { + archived_before: now, + ..Default::default() + }; + state.mark_archived(&event); + assert!(state.archived_ids.is_empty()); + + let mut state = InboxReadState::default(); + state.mark_archived(&event); + assert_eq!(state.archived_ids, HashSet::from([event.id])); + } + + #[test] + fn apply_state_recomputes_unread_and_archived() { + let now = Timestamp::from_secs(1_000_000_000); + let first = issue(&keys(2), now.as_secs() - 2000); + let second = issue(&keys(2), now.as_secs() - 1000); + let mut item = InboxItem { + root: first.id, + root_kind: None, + address: None, + events: vec![second.clone(), first.clone()], + unread_ids: Vec::new(), + archived: false, + }; + + let state = InboxReadState { + read_before: first.created_at, + ..Default::default() + }; + item.apply_state(&state); + + assert_eq!(item.unread_ids, vec![second.id]); + assert!(!item.archived); + } + #[test] fn serde_round_trip_preserves_state() { let first = issue(&keys(1), 100); diff --git a/crates/signed_state/Cargo.toml b/crates/signed_state/Cargo.toml index dc60179..554c38b 100644 --- a/crates/signed_state/Cargo.toml +++ b/crates/signed_state/Cargo.toml @@ -22,6 +22,7 @@ flume.workspace = true futures.workspace = true anyhow.workspace = true log.workspace = true +serde_json.workspace = true [target.'cfg(not(target_arch = "wasm32"))'.dependencies] rustls = "0.23" diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index 43765a2..1d8cbc8 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -14,6 +14,8 @@ use signed_core::{Announcement, RepoAddr, build_state, filters, identifier_from_ use signed_nostr::{SignedAuthUrlHandler, UniversalSigner, Update}; use crate::git_store::GitStore; +use crate::inbox::Inbox; +use crate::repos::RepoListStore; /// Keyring entry for the user credential. pub const USER_KEYRING: &str = "Signed Safe Storage"; @@ -79,19 +81,18 @@ impl BackendEvent { /// The global backend entity. /// -/// Owns the nostr client, the signer and the notification pump. +/// Owns the nostr client, the signer, the notification pump and the inbox. pub struct Backend { client: Client, signer: UniversalSigner, current_user: Option, + /// User's inbox, including notifications and recent activity. + inbox: Entity, + /// The progress of the current sync operation, if any. sync_progress: Option<(u64, u64)>, /// True when the stored credential is NIP-49 encrypted. passphrase_required: bool, /// Repositories with a push in flight, mirror or checkout based. - /// - /// A child entity: views that only care whether one repository is - /// pushing can `cx.observe` it without being invoked on unrelated - /// `Backend` changes (a `sync_progress` tick, a new relay connecting). pushing_repos: Entity>, } @@ -112,6 +113,7 @@ impl Backend { } pub(crate) fn new(client: Client, signer: UniversalSigner, cx: &mut Context) -> Self { + let weak = cx.entity().downgrade(); let pump_client = client.clone(); let pump: Task> = cx.spawn(async move |this, cx| { @@ -133,13 +135,17 @@ impl Backend { loop { let now = Instant::now(); + if now >= deadline { break; } + let timer = cx.background_executor().timer(deadline - now); futures::pin_mut!(timer); + let next = notifications.next(); futures::pin_mut!(next); + match futures::future::select(next, timer).await { futures::future::Either::Left(( Some(ClientNotification::Event { event, .. }), @@ -156,7 +162,9 @@ impl Backend { // Collect and emit the collected events. let batch = std::mem::take(&mut pending); - if let Err(e) = this.update(cx, |_, cx| cx.emit(BackendEvent::NostrUpdate(batch))) { + if let Err(e) = this.update(cx, |this, cx| { + this.emit(BackendEvent::NostrUpdate(batch), cx) + }) { log::warn!("failed to emit nostr update: {e}"); } } @@ -167,7 +175,6 @@ impl Backend { pump.detach(); // Bootstrap the client. - let weak = cx.entity().downgrade(); cx.defer(move |cx| { if let Err(error) = weak.update(cx, |this, cx| this.bootstrap(cx)) { log::warn!("backend dropped before bootstrap could run: {error}"); @@ -178,46 +185,50 @@ impl Backend { client, signer, current_user: None, + inbox: cx.new(|_| Inbox::default()), sync_progress: None, passphrase_required: false, pushing_repos: cx.new(|_| HashSet::new()), } } - /// Bootstrap the client. - /// - /// Restore the saved session, if any. + /// Bootstrap the client and restore the saved session, if any. fn bootstrap(&mut self, cx: &mut Context) { let client = self.client.clone(); let task = cx.background_spawn(async move { for url in BOOTSTRAP_RELAYS { - client.add_relay(url).and_connect().await?; + client.add_relay(url).await?; } + for url in INDEXER_RELAYS { client .add_relay(url) .capabilities(RelayCapabilities::DISCOVERY) - .and_connect() .await?; } + + client.connect().await; + Ok::<(), Error>(()) }); let notify_task: Task> = cx.spawn(async move |this, cx| { match task.await { Ok(()) => { - this.update(cx, |_this, cx| cx.notify())?; + this.update(cx, |this, cx| { + this.restore_session(cx); + })?; } Err(e) => { - this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?; + this.update(cx, |this, cx| { + this.emit(BackendEvent::error(e.to_string()), cx) + })?; } } - Ok(()) + Ok::<(), Error>(()) }); notify_task.detach(); - - self.restore_session(cx); } /// Restore the saved session from the keyring. @@ -227,7 +238,7 @@ impl Backend { /// Emits [`BackendEvent::PassphraseRequired`] for a NIP-49 encrypted identity. pub fn restore_session(&mut self, cx: &mut Context) { if cfg!(target_arch = "wasm32") { - cx.emit(BackendEvent::SignerRequired); + self.emit(BackendEvent::SignerRequired, cx); return; } @@ -237,7 +248,7 @@ impl Backend { let content = match user.await { Ok(Some((_username, secret))) => String::from_utf8(secret)?, _ => { - this.update(cx, |_, cx| cx.emit(BackendEvent::SignerRequired))?; + this.update(cx, |this, cx| this.emit(BackendEvent::SignerRequired, cx))?; return Ok(()); } }; @@ -258,15 +269,13 @@ impl Backend { signer.auth_url_handler(SignedAuthUrlHandler); this.update(cx, |this, cx| this.set_signer(signer, cx))?; } else if content.starts_with("ncryptsec1") { - // Encrypted identity. // A passphrase is required to decrypt it before the session can resume. - log::warn!("stored identity is ncryptsec-encrypted; waiting for passphrase"); this.update(cx, |this, cx| { this.passphrase_required = true; - cx.emit(BackendEvent::PassphraseRequired); + this.emit(BackendEvent::PassphraseRequired, cx); })?; } else { - this.update(cx, |_, cx| cx.emit(BackendEvent::SignerRequired))?; + this.update(cx, |this, cx| this.emit(BackendEvent::SignerRequired, cx))?; } Ok::<_, Error>(()) @@ -274,9 +283,9 @@ impl Backend { .await; if let Err(e) = result { - this.update(cx, |_, cx| { - cx.emit(BackendEvent::error(e.to_string())); - cx.emit(BackendEvent::SignerRequired); + this.update(cx, |this, cx| { + this.emit(BackendEvent::error(e.to_string()), cx); + this.emit(BackendEvent::SignerRequired, cx); })?; } @@ -360,7 +369,8 @@ impl Backend { this.signer.swap_inner(keys); this.current_user = Some(public_key); this.bootstrap_user(public_key, cx); - cx.emit(BackendEvent::SignerChanged); + this.emit(BackendEvent::SignerChanged, cx); + this.sync_inbox(cx); cx.notify(); let relays: Vec<(RelayUrl, Option)> = [ @@ -968,9 +978,7 @@ impl Backend { } else if credential.starts_with("bunker://") { self.login_with_bunker(credential, cx); } else { - cx.emit(BackendEvent::error( - "Unsupported credential, expected nsec1... or bunker://...", - )); + self.emit(BackendEvent::error("Unsupported credential."), cx); } } @@ -988,7 +996,7 @@ impl Backend { let keys = match SecretKey::parse(nsec) { Ok(secret) => Keys::new(secret), Err(e) => { - cx.emit(BackendEvent::error(e.to_string())); + self.emit(BackendEvent::error(e.to_string()), cx); return; } }; @@ -999,7 +1007,9 @@ impl Backend { let task: Task> = cx.spawn(async move |this, cx| { if let Err(e) = write.await { - this.update(cx, |_, cx| cx.emit(BackendEvent::error(e.to_string())))?; + this.update(cx, |this, cx| { + this.emit(BackendEvent::error(e.to_string()), cx) + })?; return Ok(()); } this.update(cx, |this, cx| this.set_signer(keys, cx))?; @@ -1015,7 +1025,7 @@ impl Backend { let connect_uri = match NostrConnectUri::parse(&uri_string) { Ok(uri) => uri, Err(e) => { - cx.emit(BackendEvent::error(e.to_string())); + self.emit(BackendEvent::error(e.to_string()), cx); return; } }; @@ -1045,7 +1055,9 @@ impl Backend { .await; if let Err(e) = result { - this.update(cx, |_, cx| cx.emit(BackendEvent::error(e.to_string())))?; + this.update(cx, |this, cx| { + this.emit(BackendEvent::error(e.to_string()), cx) + })?; } Ok(()) @@ -1064,8 +1076,9 @@ impl Backend { this.signer.swap_inner(Keys::generate()); this.current_user = None; this.passphrase_required = false; - cx.emit(BackendEvent::SignerChanged); - cx.emit(BackendEvent::SignerRequired); + this.emit(BackendEvent::SignerChanged, cx); + this.emit(BackendEvent::SignerRequired, cx); + this.sync_inbox(cx); cx.notify(); })?; @@ -1096,7 +1109,9 @@ impl Backend { .await; if let Err(e) = result { - this.update(cx, |_, cx| cx.emit(BackendEvent::error(e.to_string())))?; + this.update(cx, |this, cx| { + this.emit(BackendEvent::error(e.to_string()), cx) + })?; } Ok(()) @@ -1121,6 +1136,13 @@ impl Backend { self.pushing_repos.clone() } + /// The inbox child entity backing the home screen. + /// + /// A child entity: `cx.observe` it to react only to inbox changes. + pub fn inbox(&self) -> Entity { + self.inbox.clone() + } + /// Get the current user's public key. pub fn current_user(&self) -> Option { self.current_user @@ -1133,7 +1155,63 @@ impl Backend { /// Surface an error message through [`BackendEvent::Error`]. pub fn emit_error(&mut self, message: impl Into, cx: &mut Context) { - cx.emit(BackendEvent::error(message)); + self.emit(BackendEvent::error(message), cx); + } + + /// Update the inbox, then emit `event` to the other stores. + fn emit(&self, event: BackendEvent, cx: &mut Context) { + let inbox = self.inbox.downgrade(); + let inbox_event = event.clone(); + + cx.defer(move |cx| { + if let Err(error) = inbox.update(cx, |inbox, cx| { + inbox.handle_backend_event(&inbox_event, cx); + }) { + log::warn!("inbox dropped before handling backend event: {error}"); + } + }); + + cx.emit(event); + } + + /// Attach the inbox to the current signer and activate or clear it. + /// + /// The inbox's own update is deferred because activating reads `Backend`, + /// which every call site is in the middle of updating. + fn sync_inbox(&mut self, cx: &mut Context) { + if let Some(me) = self.current_user { + self.subscribe_bootstrap(filters::notifications(me), cx); + self.subscribe_bootstrap(vec![filters::authored_activity(me)], cx); + + let relays: HashSet = RepoListStore::global(cx) + .read(cx) + .announcements_of(&me) + .into_iter() + .flat_map(|announcement| announcement.relays) + .collect(); + + if !relays.is_empty() { + let relays: Vec = relays.into_iter().collect(); + self.connect_repo_relays(relays.clone(), filters::notifications(me), cx); + self.connect_repo_relays(relays, vec![filters::authored_activity(me)], cx); + } + } + + let inbox = self.inbox.downgrade(); + + cx.defer(move |cx| { + let updated = inbox.update(cx, |inbox, cx| { + if Backend::global(cx).read(cx).current_user().is_some() { + inbox.activate(cx); + } else { + inbox.reset(cx); + } + }); + + if let Err(error) = updated { + log::warn!("inbox dropped before syncing with the signer: {error}"); + } + }); } /// Progress of the in-flight negentropy sync, if any. @@ -1149,7 +1227,7 @@ impl Backend { ::Error: std::error::Error + Send + Sync + 'static, ::Error: std::error::Error + Send + Sync + 'static, { - let task: Task> = cx.spawn(async move |this, cx| { + cx.spawn(async move |this, cx| { match new_signer.get_public_key_async().await { Ok(public_key) => { this.update(cx, |this, cx| { @@ -1157,26 +1235,24 @@ impl Backend { this.current_user = Some(public_key); this.passphrase_required = false; this.bootstrap_user(public_key, cx); - cx.emit(BackendEvent::SignerChanged); + this.emit(BackendEvent::SignerChanged, cx); + this.sync_inbox(cx); cx.notify(); })?; } Err(e) => { - this.update(cx, |_this, cx| { - cx.emit(BackendEvent::error(e.to_string())); + this.update(cx, |this, cx| { + this.emit(BackendEvent::error(e.to_string()), cx); })?; } } - Ok(()) - }); - task.detach(); + Ok::<(), Error>(()) + }) + .detach(); } /// Connect to a repository's announced relays, its NIP-34 `relays` tag. - /// - /// Callers are responsible for not repeating this for relays they already - /// connected, e.g. `RepoStore::repo_relays`. pub fn connect_repo_relays( &mut self, relays: Vec, @@ -1185,13 +1261,13 @@ impl Backend { ) { let client = self.client.clone(); - let task: Task> = cx.spawn(async move |_this, _cx| { + cx.spawn(async move |_this, _cx| { if let Err(e) = connect_repo_relays(&client, relays, filters).await { log::warn!("repo relay fetch failed: {e}"); } - Ok(()) - }); - task.detach(); + Ok::<(), Error>(()) + }) + .detach(); } /// One-shot subscription on the bootstrap relays only. @@ -1201,24 +1277,25 @@ impl Backend { let fetch = cx.background_spawn(async move { subscribe_bootstrap_only(&client, filters).await }); - let task: Task> = cx.spawn(async move |this, cx| { + cx.spawn(async move |this, cx| { if let Err(e) = fetch.await { - this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?; + this.update(cx, |this, cx| { + this.emit(BackendEvent::error(e.to_string()), cx); + })?; } - Ok(()) - }); - task.detach(); + Ok::<(), Error>(()) + }) + .detach(); } /// Negentropy-sync the given filter against the bootstrap relays. pub fn sync_bootstrap(&mut self, filter: Filter, cx: &mut Context) { let client = self.client.clone(); + let (tx, mut rx) = SyncProgress::channel(); self.sync_progress = Some((0, 0)); cx.notify(); - let (tx, mut rx) = SyncProgress::channel(); - let progress_task: Task> = cx.spawn(async move |this, cx| { let mut last_percent: u64 = 0; @@ -1231,10 +1308,13 @@ impl Backend { let alive = this.update(cx, |this, cx| { this.sync_progress = Some((progress.total, progress.current)); - cx.emit(BackendEvent::SyncProgress { - total: progress.total, - current: progress.current, - }); + this.emit( + BackendEvent::SyncProgress { + total: progress.total, + current: progress.current, + }, + cx, + ); cx.notify(); }); @@ -1263,14 +1343,14 @@ impl Backend { ); this.update(cx, |this, cx| { this.sync_progress = None; - cx.emit(BackendEvent::Synced); + this.emit(BackendEvent::Synced, cx); cx.notify(); })?; } Err(e) => { this.update(cx, |this, cx| { this.sync_progress = None; - cx.emit(BackendEvent::error(e.to_string())) + this.emit(BackendEvent::error(e.to_string()), cx) })?; } } @@ -1284,7 +1364,7 @@ impl Backend { /// Callers publish with `client.send_event(...)` directly, then call this /// so stores like `RepoListStore` refresh without re-querying the relays. pub fn announce_published(&self, event: Event, cx: &mut Context) { - cx.emit(BackendEvent::Published(Box::new(event))); + self.emit(BackendEvent::Published(Box::new(event)), cx); } /// Publish a NIP-09 deletion for each of `events`, best-effort. diff --git a/crates/signed_state/src/inbox.rs b/crates/signed_state/src/inbox.rs new file mode 100644 index 0000000..67539de --- /dev/null +++ b/crates/signed_state/src/inbox.rs @@ -0,0 +1,431 @@ +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Error; +use gpui::{AppContext, Context, Task}; +use nostr_sdk::prelude::*; +use signed_core::{Deletions, InboxItem, InboxReadState, filters, inbox}; + +use crate::backend::{Backend, BackendEvent}; +use crate::refresh::{RefreshGate, RefreshRequest}; + +/// Delay between a refresh request and the actual re-query. +const REFRESH_DEBOUNCE: Duration = Duration::from_millis(300); + +/// Maximum number of "continue where you left off" activity events kept. +const ACTIVITY_LIMIT: usize = 50; + +/// State backing the inbox home screen. +#[derive(Default)] +pub struct Inbox { + /// Notifications grouped by thread root, newest activity first. + pub notifications: Arc>, + /// The user's own recent git activity, newest first. + pub activity: Arc>, + /// Number of non-archived groups with an unread event. + pub unread_count: usize, + state: InboxReadState, + /// Set once the stored state has been read for the current user. + state_loaded: bool, + refresh: RefreshGate, +} + +impl Inbox { + /// Mark every event in the group rooted at `root` as read. + pub fn mark_read(&mut self, root: EventId, cx: &mut Context) { + let Some(me) = Backend::global(cx).read(cx).current_user() else { + return; + }; + + let Some(events) = self.group_events(root) else { + return; + }; + + for event in &events { + self.state.mark_read(event); + } + + let all = self.all_notification_events(); + self.state.advance_read(&all, me, Timestamp::now()); + self.after_state_change(cx); + } + + /// Archive the group rooted at `root`. Archived events are always read too. + pub fn mark_archived(&mut self, root: EventId, cx: &mut Context) { + let Some(me) = Backend::global(cx).read(cx).current_user() else { + return; + }; + + let Some(events) = self.group_events(root) else { + return; + }; + + for event in &events { + self.state.mark_archived(event); + self.state.mark_read(event); + } + + let all = self.all_notification_events(); + let now = Timestamp::now(); + + self.state.advance_archived(&all, me, now); + self.state.advance_read(&all, me, now); + self.after_state_change(cx); + } + + /// Mark every known notification read. + pub fn mark_all_read(&mut self, cx: &mut Context) { + let Some(me) = Backend::global(cx).read(cx).current_user() else { + return; + }; + + let all = self.all_notification_events(); + self.state.mark_all_read(&all, me, Timestamp::now()); + self.after_state_change(cx); + } + + /// Handle a backend event that can change the inbox contents. + pub(crate) fn handle_backend_event(&mut self, event: &BackendEvent, cx: &mut Context) { + match event { + BackendEvent::Synced | BackendEvent::Published(_) => self.refresh(cx), + BackendEvent::NostrUpdate(updates) => { + let relevant = updates.iter().any(|update| { + let is_notification = filters::NOTIFICATION_KINDS.contains(&update.kind); + let is_comment = update.kind == Kind::Comment; + let is_event_deletion = update.kind == Kind::EventDeletion; + let is_request_to_vanish = update.kind == Kind::RequestToVanish; + + is_notification || is_comment || is_event_deletion || is_request_to_vanish + }); + if relevant { + self.refresh(cx); + } + } + _ => {} + } + } + + /// Activate the inbox for the backend's current user. + pub(crate) fn activate(&mut self, cx: &mut Context) { + let backend = Backend::global(cx); + let Some(me) = backend.read(cx).current_user() else { + return; + }; + + self.notifications = Arc::new(Vec::new()); + self.activity = Arc::new(Vec::new()); + self.unread_count = 0; + self.state = InboxReadState::default(); + self.state_loaded = false; + // Drop any in-flight or pending run belonging to the previous user. + self.refresh = RefreshGate::default(); + cx.notify(); + + self.load_state(me, cx); + } + + /// Forget everything for the current user. + pub(crate) fn reset(&mut self, cx: &mut Context) { + self.notifications = Arc::new(Vec::new()); + self.activity = Arc::new(Vec::new()); + self.unread_count = 0; + self.state = InboxReadState::default(); + self.state_loaded = false; + self.refresh = RefreshGate::default(); + cx.notify(); + } + + /// Read the stored state, then run the first refresh. + fn load_state(&mut self, me: PublicKey, cx: &mut Context) { + let backend = Backend::global(cx); + let client = backend.read(cx).client(); + + let work = cx.background_spawn(async move { load_state(&client, me).await }); + + cx.spawn(async move |this, cx| { + let loaded = work.await; + + this.update(cx, |this, cx| { + if backend.read(cx).current_user() != Some(me) { + return; + } + + match loaded { + Ok(Some(state)) => this.state = state, + Ok(None) => this.state = InboxReadState::default(), + Err(error) => log::warn!("failed to load inbox state: {error}"), + } + + this.state_loaded = true; + this.refresh_initial(cx); + })?; + + Ok::<(), Error>(()) + }) + .detach(); + } + + /// One-shot initial load, no debounce. + fn refresh_initial(&mut self, cx: &mut Context) { + debug_assert!(!self.refresh.debouncing()); + if self.refresh.running() { + self.refresh.request(); + return; + } + + self.run_refresh(cx); + } + + /// Re-query the local database. + pub fn refresh(&mut self, cx: &mut Context) { + if !self.state_loaded { + return; + } + + if self.refresh.request() != RefreshRequest::Schedule { + return; + } + + cx.spawn(async move |this, cx| { + cx.background_executor().timer(REFRESH_DEBOUNCE).await; + this.update(cx, |this, cx| this.run_refresh(cx)) + }) + .detach(); + } + + /// One query and apply cycle, the debounced entry point. + fn run_refresh(&mut self, cx: &mut Context) { + self.refresh.begin(); + + let Some(me) = Backend::global(cx).read(cx).current_user() else { + self.refresh.abort(); + return; + }; + + let client = Backend::global(cx).read(cx).client(); + let state = self.state.clone(); + + let work = cx.background_spawn(async move { + let deletion_events = client.database().query(filters::deletions()).await?; + let deletions = Deletions::from_events(deletion_events); + + let (notification_events, by_id) = fetch_notifications(&client, me, &deletions).await?; + let notifications = inbox::group(notification_events, me, &state, &|id| { + by_id.get(&id).cloned() + }); + let unread_count = notifications.iter().filter(|item| item.is_unread()).count(); + + let mut activity = Vec::new(); + for event in client + .database() + .query(filters::authored_activity(me)) + .await? + { + if deletions.is_deleted(&event) || !filters::is_git_activity(&event) { + continue; + } + activity.push(event); + } + + activity.sort_by(|a, b| { + b.created_at + .cmp(&a.created_at) + .then_with(|| b.id.to_hex().cmp(&a.id.to_hex())) + }); + activity.truncate(ACTIVITY_LIMIT); + + Ok::<_, Error>((notifications, activity, unread_count)) + }); + + let task: Task> = cx.spawn(async move |this, cx| { + let (notifications, activity, unread_count) = match work.await { + Ok(results) => results, + // Database errors are transient, keep the last lists. + Err(error) => { + log::warn!("inbox refresh failed: {error}"); + return this.update(cx, |this, _cx| this.refresh.abort()); + } + }; + + let again = this.update(cx, |this, cx| { + // The signer may have changed while the query ran, making + // these results belong to the previous user. + if Backend::global(cx).read(cx).current_user() != Some(me) { + this.refresh.abort(); + return false; + } + + this.notifications = Arc::new(notifications); + this.activity = Arc::new(activity); + this.unread_count = unread_count; + cx.notify(); + + this.refresh.finish() + })?; + + if again { + this.update(cx, |this, cx| this.refresh(cx))?; + } + + Ok(()) + }); + + task.detach(); + } + + /// Advance the cutoffs, re-derive the groups and persist the state. + fn after_state_change(&mut self, cx: &mut Context) { + self.regroup(); + self.persist(cx); + cx.notify(); + } + + /// Recompute the unread and archived flags from the current state. + fn regroup(&mut self) { + let mut items = (*self.notifications).clone(); + + for item in items.iter_mut() { + item.apply_state(&self.state); + } + + self.unread_count = items.iter().filter(|item| item.is_unread()).count(); + self.notifications = Arc::new(items); + } + + /// Sign the state with a random key and store it locally. + fn persist(&mut self, cx: &mut Context) { + let Some(me) = Backend::global(cx).read(cx).current_user() else { + return; + }; + + let client = Backend::global(cx).read(cx).client(); + let state = self.state.clone(); + + let task: Task> = cx.spawn(async move |_this, _cx| { + if let Err(error) = save_state(&client, me, &state).await { + log::warn!("failed to save inbox state: {error}"); + } + Ok(()) + }); + + task.detach(); + } + + /// Events of the group rooted at `root`. + fn group_events(&self, root: EventId) -> Option> { + self.notifications + .iter() + .find(|item| item.root == root) + .map(|item| item.events.clone()) + } + + /// Every event in every group, archived groups included. + fn all_notification_events(&self) -> Vec { + self.notifications + .iter() + .flat_map(|item| item.events.iter().cloned()) + .collect() + } +} + +/// `d` tag identifying the inbox state event of `me`. +fn inbox_state_d_tag(me: PublicKey) -> String { + format!("signed-inbox-state:{}", me.to_hex()) +} + +/// Newest stored state for `me`. +async fn load_state(client: &Client, me: PublicKey) -> Result, Error> { + let filter = Filter::new() + .kind(Kind::ApplicationSpecificData) + .identifier(inbox_state_d_tag(me)); + + let events = client.database().query(filter).await?; + + let Some(event) = events.into_iter().max_by_key(|event| event.created_at) else { + return Ok(None); + }; + + match serde_json::from_str(&event.content) { + Ok(state) => Ok(Some(state)), + Err(error) => { + log::warn!("ignoring unreadable inbox state {}: {error}", event.id); + Ok(None) + } + } +} + +/// Sign with a random key and store locally. +async fn save_state(client: &Client, me: PublicKey, state: &InboxReadState) -> Result<(), Error> { + let event = EventBuilder::new(Kind::ApplicationSpecificData, serde_json::to_string(state)?) + .tags([Tag::identifier(inbox_state_d_tag(me))]) + .finalize(&Keys::generate())?; + + client.database().save_event(&event).await?; + + Ok(()) +} + +/// Notification events and a lookup of every ancestor they reference. +async fn fetch_notifications( + client: &Client, + me: PublicKey, + deletions: &Deletions, +) -> Result<(Vec, HashMap), Error> { + let mut notifications: Vec = Vec::new(); + let mut by_id: HashMap = HashMap::new(); + + for filter in filters::notifications(me) { + for event in client.database().query(filter).await? { + if deletions.is_deleted(&event) { + continue; + } + + if by_id.insert(event.id, event.clone()).is_none() { + notifications.push(event); + } + } + } + + let mut pending: Vec = notifications.iter().flat_map(event_references).collect(); + let mut seen: HashSet = by_id.keys().copied().collect(); + + loop { + // Keep only ids not walked yet, and remember them. + pending.retain(|id| seen.insert(*id)); + + if pending.is_empty() { + break; + } + + let ancestors = client + .database() + .query(Filter::new().ids(pending.iter().copied())) + .await?; + + let mut next = Vec::new(); + + for event in ancestors { + if deletions.is_deleted(&event) { + continue; + } + next.extend(event_references(&event).filter(|id| !seen.contains(id))); + by_id.entry(event.id).or_insert(event); + } + + pending = next; + } + + Ok((notifications, by_id)) +} + +/// Event ids referenced by `event` through its `e` and `E` tags. +fn event_references(event: &Event) -> impl Iterator + '_ { + event.tags.iter().filter_map(|tag| { + if tag.kind() != "e" && tag.kind() != "E" { + return None; + } + tag.content() + .and_then(|content| EventId::from_hex(content).ok()) + }) +} diff --git a/crates/signed_state/src/lib.rs b/crates/signed_state/src/lib.rs index f7b4bcc..061c70a 100644 --- a/crates/signed_state/src/lib.rs +++ b/crates/signed_state/src/lib.rs @@ -1,6 +1,7 @@ mod backend; mod checkouts; mod git_store; +mod inbox; mod profile; mod refresh; mod repo; @@ -11,7 +12,8 @@ use std::path::{Path, PathBuf}; pub use backend::{Backend, BackendEvent, user_grasp_list_servers}; pub use checkouts::{CheckoutStatus, CheckoutsStore, pr_proposes_checkout}; pub use git_store::GitStore; -use gpui::{App, AppContext, Entity}; +use gpui::{App, AppContext}; +pub use inbox::Inbox; pub use nostr_sdk::prelude::Timestamp; pub use profile::{Profile, ProfileStore}; pub use repo::RepoStore; @@ -19,14 +21,13 @@ pub use repos::{LocalReposStore, RepoActivityCounts, RepoListStore}; use signed_nostr::new_backend; /// Initialize the backend and stores, and install them as globals. -/// Call once at startup, before opening any window that uses the stores. #[cfg(not(target_arch = "wasm32"))] pub fn init( db_path: impl AsRef, repos_root: impl Into, scan_paths: Vec, cx: &mut App, -) -> Entity { +) { // rustls uses the `aws_lc_rs` provider by default. let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); @@ -37,27 +38,22 @@ pub fn init( .expect("failed to initialize nostr backend") }); - let entity = cx.new(|cx| Backend::new(client, signer, cx)); - Backend::set_global(entity.clone(), cx); + Backend::set_global(cx.new(|cx| Backend::new(client, signer, cx)), cx); ProfileStore::set_global(cx.new(ProfileStore::new), cx); RepoListStore::set_global(cx.new(RepoListStore::new), cx); GitStore::set_global(repos_root, cx); LocalReposStore::set_global(cx.new(|cx| LocalReposStore::new(scan_paths, cx)), cx); CheckoutsStore::set_global(cx.new(CheckoutsStore::new), cx); - - entity } /// Initialize the backend with an in-memory database on wasm. #[cfg(target_arch = "wasm32")] -pub fn init(cx: &mut App) -> Entity { +pub fn init(cx: &mut App) { let (client, signer) = new_backend().expect("failed to initialize nostr backend"); - let entity = cx.new(|cx| Backend::new(client, signer, cx)); - Backend::set_global(entity.clone(), cx); + Backend::set_global(cx.new(|cx| Backend::new(client, signer, cx)), cx); ProfileStore::set_global(cx.new(ProfileStore::new), cx); RepoListStore::set_global(cx.new(RepoListStore::new), cx); GitStore::set_global(PathBuf::new(), cx); LocalReposStore::set_global(cx.new(|cx| LocalReposStore::new(Vec::new(), cx)), cx); CheckoutsStore::set_global(cx.new(|cx| CheckoutsStore::new(cx)), cx); - entity } diff --git a/crates/signed_state/src/refresh.rs b/crates/signed_state/src/refresh.rs index e0aa024..4c11981 100644 --- a/crates/signed_state/src/refresh.rs +++ b/crates/signed_state/src/refresh.rs @@ -1,9 +1,4 @@ /// Refresh coalescing shared by the event stores. -/// -/// [`crate::RepoStore`], [`crate::RepoListStore`] and [`crate::CheckoutsStore`] -/// re-query their inputs on a debounce timer with the same policy: -/// a request arriving while a run is in flight is folded into a follow-up run, -/// a request arriving while the debounce timer is pending is dropped by it. #[derive(Debug, Default)] pub struct RefreshGate { /// A run is in flight. diff --git a/docs/inbox-plan.md b/docs/inbox-plan.md index ceefae6..044981f 100644 --- a/docs/inbox-plan.md +++ b/docs/inbox-plan.md @@ -6,6 +6,12 @@ Ported from GitWorkshop's home screen, the `Dashboard` rendered at route `/` for > page. It is not. GitWorkshop's `Index` route (`src/pages/Index.tsx`) renders `` when > an account is active, and that home screen is the inbox. +> **Status.** Phases 0 and 1 are implemented and green on `feat/inbox`: +> `cargo test -p signed_core` (68), `cargo test -p signed_state` (24), +> `cargo clippy -p signed_state --all-targets` clean, `cargo check --workspace` succeeds. +> Phases 2-5 are not started. This document reflects the implementation as it stands, including +> the Phase 1 refactors (§4.3). + ## 1. What the GitWorkshop home screen is `Index.tsx`: @@ -217,54 +223,66 @@ when the signed-in key changes. ```rust /// d tag identifying the inbox read/archive state event of `me`. fn inbox_state_d_tag(me: PublicKey) -> String { - format!("signed-inbox-state:{me}") + format!("signed-inbox-state:{}", me.to_hex()) } -/// Newest stored read state for `me`, with the id of the event it came from. -async fn load_state( - client: &Client, - me: PublicKey, -) -> Result> { - // No author filter: the signing key is random per session. +/// Newest stored read state for `me`. +async fn load_state(client: &Client, me: PublicKey) -> Result, Error> { + // No author filter: the signing key is random per save. let filter = Filter::new() .kind(Kind::ApplicationSpecificData) .identifier(inbox_state_d_tag(me)); + let events = client.database().query(filter).await?; - let Some(event) = events.into_iter().max_by_key(|e| e.created_at) else { + + let Some(event) = events.into_iter().max_by_key(|event| event.created_at) else { return Ok(None); }; - Ok(serde_json::from_str(&event.content) - .ok() - .map(|state| (state, event.id))) + + match serde_json::from_str(&event.content) { + Ok(state) => Ok(Some(state)), + Err(error) => { + log::warn!("ignoring unreadable inbox state {}: {error}", event.id); + Ok(None) + } + } } -/// Sign with the random `keys` and store locally. -async fn save_state( - client: &Client, - keys: &Keys, - me: PublicKey, - state: &InboxReadState, -) -> Result { +/// Sign with a fresh random key and store locally. +async fn save_state(client: &Client, me: PublicKey, state: &InboxReadState) -> Result<(), Error> { let event = EventBuilder::new(Kind::ApplicationSpecificData, serde_json::to_string(state)?) .tags([Tag::identifier(inbox_state_d_tag(me))]) - .finalize(keys)?; // synchronous: random keys, no user signer + .finalize(&Keys::generate())?; // synchronous: random key, no user signer // Local-only: no `send_event`, no broadcast. The event lives in LMDB. client.database().save_event(&event).await?; - Ok(event.id) + Ok(()) } ``` -Kind 30078 is addressable, so saves signed by the same `keys` replace the previous event. Because -`keys` is random per session, the first save of a session creates a new coordinate; the store then -deletes the event it loaded (its id is kept from `load_state`) so exactly one state event remains. +A fresh random key is generated on every save, so each save writes a new event rather than +replacing the previous one. LMDB only auto-replaces an addressable event when the incoming event +has the **same pubkey**, so old copies accumulate. Nothing prunes them; `load_state` reads the +newest by `created_at`, so the behavior is correct. This is a deliberate trade for not caching a +key in the store (see §4.3). An earlier implementation deleted the previous event by tracking its +id across saves; that was removed as more derived state than it was worth. `NostrDatabase::{save_event, query}` and `Client::database()` are existing SDK APIs. -### 4.3 `signed_state`: one `InboxStore` +### 4.3 `signed_state`: `Inbox`, a child entity of `Backend` -One store backs the whole screen, modelled on `RepoListStore` (`repos.rs`): +The inbox is not an app-wide global. It is a child `Entity` owned by `Backend` +(`inbox: Entity`), following the project's child-entity pattern +(`docs/backend-rearchitecture.md` §11): its observer set (the inbox screen, the sidebar badge) is a +strict subset of the backend's, so it is observed independently. ```rust -pub struct InboxStore { +// backend.rs +pub struct Backend { + ... + inbox: Entity, +} + +// inbox.rs +pub struct Inbox { /// Activity directed at the user, grouped by thread root, newest first. pub notifications: Arc>, /// The user's own recent git activity, newest first. @@ -272,69 +290,132 @@ pub struct InboxStore { /// Unread notification count (non-archived). pub unread_count: usize, state: InboxReadState, - user: Option, - /// Random keypair signing the local NIP-78 storage event. - keys: Keys, - /// Event id of the loaded state event, pruned on the next save. - loaded_state: Option, + /// Set once the stored state has been read for the current user. + state_loaded: bool, refresh: RefreshGate, - _subscription: Subscription, } ``` -**Lifespan: idle until a signer exists.** The store is created in `signed_state::init` like the -other stores, but it does nothing until the user has a signer. It is never wired from the `desktop` -crate, and `signed_state::init` gains no parameters. +`Backend::new` builds it with `cx.new(|_| Inbox::default())`, and callers reach it through +`Backend::global(cx).read(cx).inbox()` or `Backend::inbox()`. All inbox operations (`mark_read`, +`mark_archived`, `mark_all_read`, `refresh`) live on `Inbox`. -- `new` schedules `cx.defer`, like the other stores. The deferred bootstrap checks - `Backend::current_user()`: - - signer already present (session restored before the store was created): activate now; - - no signer: do nothing, wait for the event. -- `BackendEvent::SignerChanged`: activate. -- `BackendEvent::SignerRequired` (logout / no credential): clear `user`, `notifications`, - `activity`, `unread_count`. -- `BackendEvent::NostrUpdate(updates)`: refresh when any update kind is in `NOTIFICATION_KINDS`, - is `Kind::Comment`, or is a deletion. -- `BackendEvent::Synced`: refresh. -- `BackendEvent::Published`: refresh. +**The store holds no derived state.** The current user is read from `Backend::current_user()` at +each use site, repo relays are queried from `RepoListStore` in `Backend::sync_inbox`, and the +signing key is random per save rather than cached. This follows the project rule against caching +derived state. -**Activation** (only on signer): +**Lifespan: idle until a signer exists.** `Inbox` is created with the backend but does nothing +until the user has a signer. It is never wired from the `desktop` crate, and `signed_state::init` +gains no parameters (the `InboxStore::set_global` idea was dropped). + +The dependency is strictly one-way: **`Backend` → `Inbox`**. `Inbox` holds no `Backend` handle, so +there is no reference cycle and no `cx.subscribe`. Two mechanisms connect them. + +**`Backend::emit`** is the single funnel for every `BackendEvent`. It updates the inbox on a +deferred effect and then emits to the other subscribers: ```rust -fn activate(&mut self, me: PublicKey, cx: &mut Context) { - self.user = Some(me); - // Load the NIP-78 state from LMDB, then subscribe and refresh. - // All run on background tasks; only plain data crosses back. +/// Update the inbox, then emit `event` to the other stores. +fn emit(&self, event: BackendEvent, cx: &mut Context) { + let inbox = self.inbox.downgrade(); + let inbox_event = event.clone(); + + cx.defer(move |cx| { + if let Err(error) = inbox.update(cx, |inbox, cx| { + inbox.handle_backend_event(&inbox_event, cx); + }) { + log::warn!("inbox dropped before handling backend event: {error}"); + } + }); + + cx.emit(event); +} +``` + +The `cx.defer` is load-bearing: every emit site runs inside `Backend::update`, and the inbox +handlers read `Backend`, so a synchronous call would re-enter the borrowed entity and panic. All +`cx.emit(...)` sites route through `self.emit(...)`. + +`Inbox::handle_backend_event` reacts to only three shapes: + +- `NostrUpdate(updates)`: refresh when any update kind is in `NOTIFICATION_KINDS`, is + `Kind::Comment`, or is a deletion (`EventDeletion` / `RequestToVanish`). +- `Synced` / `Published`: refresh. +- everything else: ignored. + +`BackendEvent::SignerChanged` and `SignerRequired` are still emitted and must stay: `CheckoutsStore` +and `SidebarPanel` consume them. They no longer drive the inbox. + +**Signer lifecycle: `Backend::sync_inbox`.** The inbox does not match `SignerChanged` / +`SignerRequired`. `Backend` owns the wiring and calls `sync_inbox` from the three real signer +transitions: `create_identity`, `set_signer` (covers nsec, bunker and passphrase restore) and +`logout`. The fetch work that used to live in `Inbox::activate` moved here, because the filters and +repo relays need `Backend`'s state: + +```rust +fn sync_inbox(&mut self, cx: &mut Context) { + if let Some(me) = self.current_user { + self.subscribe_bootstrap(filters::notifications(me), cx); + self.subscribe_bootstrap(vec![filters::authored_activity(me)], cx); + + let relays: HashSet = RepoListStore::global(cx) + .read(cx) + .announcements_of(&me) + .into_iter() + .flat_map(|announcement| announcement.relays) + .collect(); + + if !relays.is_empty() { + let relays: Vec = relays.into_iter().collect(); + self.connect_repo_relays(relays.clone(), filters::notifications(me), cx); + self.connect_repo_relays(relays, vec![filters::authored_activity(me)], cx); + } + } + + let inbox = self.inbox.downgrade(); + cx.defer(move |cx| { + let updated = inbox.update(cx, |inbox, cx| { + if Backend::global(cx).read(cx).current_user().is_some() { + inbox.activate(cx); + } else { + inbox.reset(cx); + } + }); + if let Err(error) = updated { + log::warn!("inbox dropped before syncing with the signer: {error}"); + } + }); +} +``` + +The repo relays are read from `RepoListStore::global(cx).read(cx).announcements_of(&me)` at call +time and never cached. (NIP-65 outbox relay discovery is deferred; Signed does not fetch kind +10002 yet.) The deferred `update` is required because `activate` reads `Backend`, which every +caller is mid-update on. `Inbox::activate` and `Inbox::reset` are `pub(crate)`; `Inbox` no longer +has `subscribe_remote` / `connect_own_repo_relays`. + +**Activation** (`activate`) clears the user's data, drops any in-flight or pending run belonging to +the previous user (`self.refresh = RefreshGate::default()`), then loads the NIP-78 state from LMDB +and chains the first refresh once it is loaded: + +```rust +pub(crate) fn activate(&mut self, cx: &mut Context) { + let Some(me) = Backend::global(cx).read(cx).current_user() else { return }; + // clear notifications, activity, unread_count; state = default; state_loaded = false; + // self.refresh = RefreshGate::default(); self.load_state(me, cx); - self.subscribe_remote(cx); - self.refresh(cx); } ``` -Loading uses the random `keys` only for signing on save; reading the state event needs no signer at -all. Activation itself is still gated on the signer because the fetch filters need the user's pubkey. +`reset` performs the same clearing without a state load, and is used on logout. -**Fetch** (reuses `Backend::subscribe_bootstrap` / `connect_repo_relays`): +Reading the state event needs no signer at all (the `d` tag carries the identity); activation is +still gated on the signer because the fetch filters need the user's pubkey. -```rust -fn subscribe_remote(&mut self, cx: &mut Context) { - let Some(me) = self.user else { return }; - let backend = Backend::global(cx); - backend.update(cx, |backend, cx| { - backend.subscribe_bootstrap(filters::notifications(me), cx); - backend.subscribe_bootstrap(vec![filters::authored_activity(me)], cx); - }); - // Relays of the user's own repos, so their activity there is found too. - let relays = own_repo_relays(me, cx); - backend.update(cx, |backend, cx| { - backend.connect_repo_relays(relays.clone(), filters::notifications(me), cx); - backend.connect_repo_relays(relays, vec![filters::authored_activity(me)], cx); - }); -} -``` - -`own_repo_relays` reads `RepoListStore::global(cx).read(cx).announcements_of(&me)` and unions their -`relays`. (NIP-65 outbox relay discovery is deferred; Signed does not fetch kind 10002 yet.) +**Fetch** reuses `Backend::subscribe_bootstrap` and `Backend::connect_repo_relays`, as shown in +`sync_inbox` above. The query that follows is intentionally the offline-first cache read, not a +wait on the network; see the note below. **Refresh** (mirrors `RepoListStore::run_refresh`): @@ -345,12 +426,23 @@ fn subscribe_remote(&mut self, cx: &mut Context) { `inbox::group`. - Filter the activity events: keep issues/PRs/patches/statuses/cover notes, and comments only when their `K` tag is a git kind; sort newest first; take the top N. -- Cross back to the main thread: set `notifications`, `activity`, `unread_count`, `cx.notify()`, - `refresh.finish()`. +- Cross back to the main thread: guard on `Backend::global(cx).read(cx).current_user() == + Some(me)`; if the signer changed while the query ran, `refresh.abort()` instead of applying, so a + previous user's results never land. Then set `notifications`, `activity`, `unread_count`, + `cx.notify()`, `refresh.finish()`. -**Actions**: `mark_read(root)`, `mark_all_read()`, plus `advance_read` after marking to bound the -`read_ids` set. After each change, sign with the random `keys` and save the NIP-78 event to LMDB, -then delete the previously loaded copy (see 4.2). +**Fetch vs. the immediate query.** `subscribe_bootstrap` / `connect_repo_relays` return immediately, +so the query that follows them reads the local cache rather than waiting for the relays. That is +deliberate offline-first behavior: cached content appears at once on a warm start and with no +network, instead of blocking the home screen on the network. The gap is closed by the SDK, not by +timing: received events are written to LMDB and surfaced as `ClientNotification::Event`, so +`Backend`'s pump batches them into `BackendEvent::NostrUpdate` and the inbox refreshes. This was +reviewed and left as-is. + +**Actions**: `mark_read(root)`, `mark_archived(root)`, `mark_all_read()`. Each group's events are +marked, then the cutoffs are advanced against *all* notification events to bound the id sets. After +each change the groups are re-derived (`InboxItem::apply_state`) and the state is saved to LMDB, +signed with a fresh random key (see 4.2). **My repositories needs no new store**: `RepoListStore` already holds every announcement and exposes `announcements_of(user)`. @@ -371,7 +463,7 @@ One scrollable two-column flex row. **Mark all read**; then the non-archived notification items (top 5, with a **Show all** toggle expanding inline). Rows show the actor avatar, a kind badge, the subject, the repo name, a relative time, and an unread dot. Empty state: "You're all caught up." with `IconName::Inbox`. -- **Continue where you left off**: `InboxStore::activity`, top 15, each row a kind icon, subject, +- **Continue where you left off**: `Inbox::activity`, top 15, each row a kind icon, subject, repo name, and relative time. - **My repositories**: `RepoListStore::announcements_of(me)` with a small search `InputState` (same pattern as `RepoListView`) and a **New** button opening the existing `create_repo_dialog`. Rows @@ -398,7 +490,7 @@ pub fn add_bottom_panel( The workspace already supports a bottom dock and prunes it when empty (`workspace.rs`). Then: - New `InboxFilterView` panel taking a mode `InboxFilter::Unread | InboxFilter::Archived` and the - `InboxStore`. It renders the matching subset of `InboxStore::notifications` as a list. + `Entity`. It renders the matching subset of `Inbox::notifications` as a list. - The **Unread** and **Archived** header buttons in `InboxView` call `add_bottom_panel` with the requested mode. `InboxView` keeps `filter_view: Option>`; when it already exists, update its mode and focus instead of adding a duplicate. @@ -417,7 +509,7 @@ In `views/sidebar/mod.rs`: .on_click(cx.listener(|this, _ev, window, cx| this.open_inbox(window, cx))), ``` -- `cx.observe` the `InboxStore` so the badge updates. +- `cx.observe` `Backend::global(cx).read(cx).inbox()` so the badge updates. ### 5.4 Click-through (P1) @@ -455,20 +547,22 @@ patch-root click opens the repo panel. Note as a known limitation. | File | Change | |---|---| | `crates/signed_core/Cargo.toml` | add `serde` | -| `crates/signed_core/src/filters.rs` | `NOTIFICATION_KINDS`, `notification_comments`, `notifications`, `authored_activity` | +| `crates/signed_core/src/filters.rs` | `NOTIFICATION_KINDS`, `notification_comments`, `notifications`, `authored_activity`, `is_git_activity`, `deletions` | | `crates/signed_core/src/inbox.rs` | **new**: `InboxItem`, `notification_root`, `group`, `InboxReadState`, tests | | `crates/signed_core/src/lib.rs` | `mod inbox;` and re-exports | | `crates/signed_state/Cargo.toml` | add `serde_json` | -| `crates/signed_state/src/inbox.rs` | **new**: `InboxStore`, global, signer-gated activation, NIP-78 load/save, actions | -| `crates/signed_state/src/lib.rs` | `mod inbox;`, set global in `init` | +| `crates/signed_state/src/inbox.rs` | **new**: `Inbox` child entity, NIP-78 load/save, refresh, actions; no `Backend` handle, no `cx.subscribe` | +| `crates/signed_state/src/backend.rs` | `inbox: Entity` field, construction, `inbox()` accessor, private `emit` funnel, `sync_inbox`, `RepoListStore` import | +| `crates/signed_state/src/refresh.rs` | doc comment lists `Inbox` among the `RefreshGate` users | +| `crates/signed_state/src/lib.rs` | `mod inbox;`, re-export `Inbox` (no global install) | | `crates/dock/src/lib.rs` | `add_bottom_panel` helper | | `crates/workspace/src/views/inbox.rs` | **new**: `InboxView` home panel and `InboxFilterView` | | `crates/workspace/src/views/mod.rs` | `mod inbox; pub use inbox::InboxView;` | | `crates/workspace/src/views/sidebar/mod.rs` | `inbox` field, `open_inbox`, nav wiring and badge | | `crates/workspace/src/views/repo_detail/mod.rs` | `RepoItem`, `RepoDetailView::open_item` (P1) | -No changes to `desktop` or `signed_nostr`. `signed_state::init` gains no parameters; the store -bootstraps itself via `cx.defer` once a signer is present. +No changes to `desktop` or `signed_nostr`. `signed_state::init` gains no parameters; `Backend::sync_inbox` +activates the `Inbox` child entity at each signer transition. ## 7. Phasing @@ -477,9 +571,11 @@ bootstraps itself via `cx.defer` once a signer is present. and `inbox::{InboxItem, notification_root, group, InboxReadState}`. Two deviations from the sketch: the cutoff methods take an explicit `now: Timestamp` so the pure logic stays deterministic and testable, and `authored_activity` results must pass through `is_git_activity` before display (comments on - non-git roots are matched by the filter). `cargo test -p signed_core` passes (62 tests). -2. **Phase 1 - store**: `InboxStore` with signer-gated activation, both queries, unread count, and - NIP-78 load/save to LMDB, global install. + non-git roots are matched by the filter). `cargo test -p signed_core` passes (66 tests at the + end of Phase 0; 68 after the two Phase 1 additions). +2. **Phase 1 - store**: `Inbox` child entity, activated by `Backend::sync_inbox` once a signer + exists; both queries, unread count, and NIP-78 load/save to LMDB. **DONE.** See the + implementation notes below. 3. **Phase 2 - screen**: `InboxView` (inbox + activity + my repositories), sidebar nav and badge. 4. **Phase 3 - sub-views**: `add_bottom_panel` and `InboxFilterView` for Unread / Archived. 5. **Phase 4 - click-through**: `open_item` and announcement lookup. @@ -488,12 +584,59 @@ bootstraps itself via `cx.defer` once a signer is present. Each phase compiles and is usable on its own. +### Phase 1 implementation notes + +Files: `crates/signed_state/{Cargo.toml, src/inbox.rs, src/lib.rs, src/backend.rs, src/refresh.rs}` +and two additions to `crates/signed_core/src/inbox.rs`. + +- `Inbox` is a child entity of `Backend` (`inbox: Entity`), created in `Backend::new` and + reached via `Backend::inbox()`. Nothing in `desktop` is wired and `signed_state::init` gains no + parameters. The dependency is strictly one-way: `Inbox` holds no `Backend` handle. +- `Backend::emit` is the single funnel for every `BackendEvent`. It updates the inbox through + `cx.defer` and then emits to the other subscribers. The defer is required: every emit site runs + inside `Backend::update`, and the inbox handlers read `Backend`, so a synchronous call panics on + a re-entrant entity access. +- The signer lifecycle lives in `Backend::sync_inbox`, called from `create_identity`, `set_signer` + and `logout`. It starts the subscriptions and repo-relay connects, then defers `inbox.activate` + / `inbox.reset`. `SignerChanged` / `SignerRequired` are still emitted for `CheckoutsStore` and + `SidebarPanel`, but no longer drive the inbox. +- `Inbox` mirrors `RepoListStore`: `RefreshGate` coalescing, `cx.background_spawn` for the + database work, plain data applied on the main thread, refresh-on-`NostrUpdate`/`Synced`/`Published`. +- Added `state_loaded: bool`, not in the sketch. Groups are derived from the read state, so a refresh + before the stored state is read would briefly mark everything unread. The first refresh is chained + after `load_state`, and later `refresh` calls are ignored until `state_loaded` is set. +- Account switches are guarded. `activate` and `reset` both replace `self.refresh` with a fresh + `RefreshGate`, dropping any in-flight or pending run of the previous user, and the apply step of + `run_refresh` aborts instead of applying when `Backend::current_user()` no longer matches the + user the query was started for. +- Two additions to `signed_core::inbox` that Phase 1 needs: `InboxReadState::mark_archived` (mirrors + `mark_read`) and `InboxItem::apply_state` (recomputes `unread_ids`/`archived`; `group` now uses it). + Both are covered by tests. +- The thread-root lookup is built by walking every `e`/`E` ancestor transitively (`fetch_notifications`) + rather than a single hop, because a patch series chains through parent patches. Only the notification + events are grouped; ancestors are used solely as the lookup, so a root authored by someone else is + not mistaken for a notification. +- The read/archive state event is written to LMDB only (`database().save_event`), signed with a fresh + `Keys::generate()` on each save and never published. Filtering is by `d` tag only, no author, so the + random key is irrelevant across sessions. `d` tag uses `me.to_hex()` rather than `Display`. +- Actions: `mark_read(root)`, `mark_archived(root)`, `mark_all_read()`. Each marks the group, advances + the relevant cutoffs against **all** notification events (matching GitWorkshop's use of `allEvents`), + re-derives the groups locally so the UI updates immediately, then persists in the background. +- The store keeps no derived state. The signing key is generated per save, the current user is read + from `Backend::current_user()` where needed, and the relays of the user's own repositories are + queried from `RepoListStore` in `Backend::sync_inbox` rather than cached. There is no prune logic + either: the newest state event is selected by `created_at`. +- `Inbox::activate` / `Inbox::reset` are `pub(crate)`; the former `subscribe_remote` and + `connect_own_repo_relays` methods were deleted once their work moved into `Backend::sync_inbox`. +- `cargo test -p signed_core` passes (68 tests), `cargo test -p signed_state` passes (24 tests); + `cargo clippy -p signed_state --all-targets` is clean; `cargo check --workspace` succeeds. + ## 8. Validation -- `cargo test -p signed_core`: root resolution, grouping, read-state cutoff, serde round-trip. -- `cargo test -p signed_state`: NIP-78 content round-trip (`serde_json`), if a non-GPUI path is - factored out. -- `cargo check --workspace` after each phase. +- `cargo test -p signed_core` (68 tests): root resolution, grouping, read-state cutoff, serde round-trip. +- `cargo test -p signed_state` (24 tests): the `Inbox` store paths that do not need GPUI (state + round-trip, grouping helpers). +- `cargo clippy -p signed_state --all-targets` and `cargo check --workspace` after each phase. - Manual: log in with a repo-owning identity; confirm the inbox panel populates from another identity's issue/comment, the activity list shows your own items, the repositories panel matches the sidebar, and that no kind-30078 event is broadcast (watch the relays / `Published` events). @@ -513,3 +656,7 @@ Each phase compiles and is usable on its own. `NostrDatabase::{save_event, query}`; `NostrLmdb`, `NostrGossipMemory` - `EventBuilder::{new, tags, finalize}`, `Tag::identifier`, `Keys::generate` - `Timestamp`, `EventId` (hex serde), `PublicKey`, `Coordinate` +- Fetch paths converge on the same notification: `client.subscribe(...)` and negentropy + `client.sync(...)` both persist received events to LMDB and surface them as + `ClientNotification::Event`, which `Backend`'s pump batches into `BackendEvent::NostrUpdate`. + This is why the query right after a fetch is a cache read, not a race.