reya/signed
1
0
Fork 0
Code Issues Pull Requests 1 Actions 4 Releases Activity

add inbox view

Browse Source
This commit is contained in:
reya
2026-09-11 09:35:38 +07:00
parent b7221ef814
commit e63e58c125
7 changed files with 1119 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/signed_state/src/backend.rs
+45 -89
View File
@@ -34,10 +34,6 @@ pub const BOOTSTRAP_RELAYS: [&str; 4] = [
pub const INDEXER_RELAYS: [&str; 2] = ["wss://indexer.coracle.social", "wss://user.kindpag.es"];
/// Delay the notification pump waits for more events before emitting a batch.
///
/// A negentropy sync can deliver hundreds of events in a burst; batching
/// them here means every subscriber debounces the burst once, not once per
/// subscriber.
const PUMP_DEBOUNCE: Duration = Duration::from_millis(200);
#[derive(Debug, Clone)]
@@ -79,9 +75,6 @@ impl BackendEvent {
}
}
/// The global backend entity.
///
/// Owns the nostr client, the signer, the notification pump and the inbox.
pub struct Backend {
client: Client,
signer: UniversalSigner,
@@ -162,9 +155,9 @@ impl Backend {
// Collect and emit the collected events.
let batch = std::mem::take(&mut pending);
if let Err(e) = this.update(cx, |this, cx| {
this.emit(BackendEvent::NostrUpdate(batch), cx)
}) {
if let Err(e) =
this.update(cx, |_this, cx| cx.emit(BackendEvent::NostrUpdate(batch)))
{
log::warn!("failed to emit nostr update: {e}");
}
}
@@ -221,9 +214,7 @@ impl Backend {
})?;
}
Err(e) => {
this.update(cx, |this, cx| {
this.emit(BackendEvent::error(e.to_string()), cx)
})?;
this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
}
}
Ok::<(), Error>(())
@@ -231,14 +222,13 @@ impl Backend {
notify_task.detach();
}
/// Restore the saved session from the keyring.
/// Restore the saved session from the Keyring.
///
/// Emits [`BackendEvent::SignerRequired`] when no credential is stored.
///
/// Emits [`BackendEvent::PassphraseRequired`] for a NIP-49 encrypted identity.
/// - Emits [`BackendEvent::SignerRequired`] when no credential is stored.
/// - Emits [`BackendEvent::PassphraseRequired`] for a NIP-49 encrypted identity.
pub fn restore_session(&mut self, cx: &mut Context<Self>) {
if cfg!(target_arch = "wasm32") {
self.emit(BackendEvent::SignerRequired, cx);
cx.emit(BackendEvent::SignerRequired);
return;
}
@@ -248,7 +238,7 @@ impl Backend {
let content = match user.await {
Ok(Some((_username, secret))) => String::from_utf8(secret)?,
_ => {
this.update(cx, |this, cx| this.emit(BackendEvent::SignerRequired, cx))?;
this.update(cx, |_this, cx| cx.emit(BackendEvent::SignerRequired))?;
return Ok(());
}
};
@@ -272,10 +262,10 @@ impl Backend {
// A passphrase is required to decrypt it before the session can resume.
this.update(cx, |this, cx| {
this.passphrase_required = true;
this.emit(BackendEvent::PassphraseRequired, cx);
cx.emit(BackendEvent::PassphraseRequired);
})?;
} else {
this.update(cx, |this, cx| this.emit(BackendEvent::SignerRequired, cx))?;
this.update(cx, |_this, cx| cx.emit(BackendEvent::SignerRequired))?;
}
Ok::<_, Error>(())
@@ -283,9 +273,9 @@ impl Backend {
.await;
if let Err(e) = result {
this.update(cx, |this, cx| {
this.emit(BackendEvent::error(e.to_string()), cx);
this.emit(BackendEvent::SignerRequired, cx);
this.update(cx, |_this, cx| {
cx.emit(BackendEvent::error(e.to_string()));
cx.emit(BackendEvent::SignerRequired);
})?;
}
@@ -369,8 +359,10 @@ impl Backend {
this.signer.swap_inner(keys);
this.current_user = Some(public_key);
this.bootstrap_user(public_key, cx);
this.emit(BackendEvent::SignerChanged, cx);
cx.emit(BackendEvent::SignerChanged);
this.sync_inbox(cx);
cx.notify();
let relays: Vec<(RelayUrl, Option<RelayMetadata>)> = [
@@ -978,7 +970,7 @@ impl Backend {
} else if credential.starts_with("bunker://") {
self.login_with_bunker(credential, cx);
} else {
self.emit(BackendEvent::error("Unsupported credential."), cx);
cx.emit(BackendEvent::error("Unsupported credential."));
}
}
@@ -996,7 +988,7 @@ impl Backend {
let keys = match SecretKey::parse(nsec) {
Ok(secret) => Keys::new(secret),
Err(e) => {
self.emit(BackendEvent::error(e.to_string()), cx);
cx.emit(BackendEvent::error(e.to_string()));
return;
}
};
@@ -1007,9 +999,7 @@ impl Backend {
let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
if let Err(e) = write.await {
this.update(cx, |this, cx| {
this.emit(BackendEvent::error(e.to_string()), cx)
})?;
this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
return Ok(());
}
this.update(cx, |this, cx| this.set_signer(keys, cx))?;
@@ -1025,7 +1015,7 @@ impl Backend {
let connect_uri = match NostrConnectUri::parse(&uri_string) {
Ok(uri) => uri,
Err(e) => {
self.emit(BackendEvent::error(e.to_string()), cx);
cx.emit(BackendEvent::error(e.to_string()));
return;
}
};
@@ -1055,9 +1045,7 @@ impl Backend {
.await;
if let Err(e) = result {
this.update(cx, |this, cx| {
this.emit(BackendEvent::error(e.to_string()), cx)
})?;
this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
}
Ok(())
@@ -1076,8 +1064,8 @@ impl Backend {
this.signer.swap_inner(Keys::generate());
this.current_user = None;
this.passphrase_required = false;
this.emit(BackendEvent::SignerChanged, cx);
this.emit(BackendEvent::SignerRequired, cx);
cx.emit(BackendEvent::SignerChanged);
cx.emit(BackendEvent::SignerRequired);
this.sync_inbox(cx);
cx.notify();
})?;
@@ -1109,9 +1097,7 @@ impl Backend {
.await;
if let Err(e) = result {
this.update(cx, |this, cx| {
this.emit(BackendEvent::error(e.to_string()), cx)
})?;
this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
}
Ok(())
@@ -1155,31 +1141,15 @@ impl Backend {
/// Surface an error message through [`BackendEvent::Error`].
pub fn emit_error(&mut self, message: impl Into<String>, cx: &mut Context<Self>) {
self.emit(BackendEvent::error(message), cx);
}
/// Update the inbox, then emit `event` to the other stores.
fn emit(&self, event: BackendEvent, cx: &mut Context<Self>) {
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);
cx.emit(BackendEvent::error(message));
}
/// 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<Self>) {
if let Some(me) = self.current_user {
let client = self.client.clone();
let me = self.current_user;
if let Some(me) = me {
self.subscribe_bootstrap(filters::notifications(me), cx);
self.subscribe_bootstrap(vec![filters::authored_activity(me)], cx);
@@ -1197,20 +1167,9 @@ impl Backend {
}
}
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}");
}
self.inbox.update(cx, |inbox, cx| match me {
Some(me) => inbox.activate(me, client, cx),
None => inbox.reset(cx),
});
}
@@ -1235,14 +1194,14 @@ impl Backend {
this.current_user = Some(public_key);
this.passphrase_required = false;
this.bootstrap_user(public_key, cx);
this.emit(BackendEvent::SignerChanged, cx);
cx.emit(BackendEvent::SignerChanged);
this.sync_inbox(cx);
cx.notify();
})?;
}
Err(e) => {
this.update(cx, |this, cx| {
this.emit(BackendEvent::error(e.to_string()), cx);
this.update(cx, |_this, cx| {
cx.emit(BackendEvent::error(e.to_string()));
})?;
}
}
@@ -1279,8 +1238,8 @@ impl Backend {
cx.spawn(async move |this, cx| {
if let Err(e) = fetch.await {
this.update(cx, |this, cx| {
this.emit(BackendEvent::error(e.to_string()), cx);
this.update(cx, |_this, cx| {
cx.emit(BackendEvent::error(e.to_string()));
})?;
}
Ok::<(), Error>(())
@@ -1308,13 +1267,10 @@ impl Backend {
let alive = this.update(cx, |this, cx| {
this.sync_progress = Some((progress.total, progress.current));
this.emit(
BackendEvent::SyncProgress {
total: progress.total,
current: progress.current,
},
cx,
);
cx.emit(BackendEvent::SyncProgress {
total: progress.total,
current: progress.current,
});
cx.notify();
});
@@ -1343,14 +1299,14 @@ impl Backend {
);
this.update(cx, |this, cx| {
this.sync_progress = None;
this.emit(BackendEvent::Synced, cx);
cx.emit(BackendEvent::Synced);
cx.notify();
})?;
}
Err(e) => {
this.update(cx, |this, cx| {
this.sync_progress = None;
this.emit(BackendEvent::error(e.to_string()), cx)
cx.emit(BackendEvent::error(e.to_string()))
})?;
}
}
@@ -1364,7 +1320,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<Self>) {
self.emit(BackendEvent::Published(Box::new(event)), cx);
cx.emit(BackendEvent::Published(Box::new(event)));
}
/// Publish a NIP-09 deletion for each of `events`, best-effort.
crates/signed_state/src/inbox.rs
+231 -126
View File
@@ -16,131 +16,87 @@ 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.
/// The user's persisted inbox read state.
#[derive(Default)]
pub struct Inbox {
/// Notifications grouped by thread root, newest activity first.
pub notifications: Arc<Vec<InboxItem>>,
/// The user's own recent git activity, newest first.
pub activity: Arc<Vec<Event>>,
/// 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,
/// Unread notification groups, published by [`InboxStore`] for the sidebar badge.
pub unread_count: usize,
}
impl Inbox {
/// Mark every event in the group rooted at `root` as read.
pub fn mark_read(&mut self, root: EventId, cx: &mut Context<Self>) {
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);
/// The current read/archive cutoffs.
pub fn state(&self) -> &InboxReadState {
&self.state
}
/// Archive the group rooted at `root`. Archived events are always read too.
pub fn mark_archived(&mut self, root: EventId, cx: &mut Context<Self>) {
let Some(me) = Backend::global(cx).read(cx).current_user() else {
return;
};
/// Whether the stored state has been read for the current user.
pub fn is_loaded(&self) -> bool {
self.state_loaded
}
let Some(events) = self.group_events(root) else {
/// Publish the unread count derived by [`InboxStore`].
pub fn set_unread_count(&mut self, count: usize, cx: &mut Context<Self>) {
if self.unread_count == count {
return;
};
}
self.unread_count = count;
cx.notify();
}
for event in &events {
/// Mark the events of one notification group read, then bound the id sets.
pub fn mark_read(
&mut self,
group: &[Event],
all: &[Event],
me: PublicKey,
cx: &mut Context<Self>,
) {
for event in group {
self.state.mark_read(event);
}
self.state.advance_read(all, me, Timestamp::now());
self.persist(cx);
cx.notify();
}
/// Archive one notification group. Archived events are always read too.
pub fn mark_archived(
&mut self,
group: &[Event],
all: &[Event],
me: PublicKey,
cx: &mut Context<Self>,
) {
for event in group {
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);
self.state.advance_archived(all, me, now);
self.state.advance_read(all, me, now);
self.persist(cx);
cx.notify();
}
/// Mark every known notification read.
pub fn mark_all_read(&mut self, cx: &mut Context<Self>) {
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<Self>) {
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<Self>) {
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>) {
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();
pub fn mark_all_read(&mut self, all: &[Event], me: PublicKey, cx: &mut Context<Self>) {
self.state.mark_all_read(all, me, Timestamp::now());
self.persist(cx);
cx.notify();
}
/// Read the stored state, then run the first refresh.
fn load_state(&mut self, me: PublicKey, cx: &mut Context<Self>) {
let backend = Backend::global(cx);
let client = backend.read(cx).client();
/// Load the stored state for current user.
pub(crate) fn activate(&mut self, me: PublicKey, client: Client, cx: &mut Context<Self>) {
self.state = InboxReadState::default();
self.state_loaded = false;
self.unread_count = 0;
cx.notify();
let backend = Backend::global(cx);
let work = cx.background_spawn(async move { load_state(&client, me).await });
cx.spawn(async move |this, cx| {
@@ -158,7 +114,7 @@ impl Inbox {
}
this.state_loaded = true;
this.refresh_initial(cx);
cx.notify();
})?;
Ok::<(), Error>(())
@@ -166,6 +122,161 @@ impl Inbox {
.detach();
}
/// Clear the state of the signed-out user.
pub(crate) fn reset(&mut self, cx: &mut Context<Self>) {
self.state = InboxReadState::default();
self.state_loaded = false;
self.unread_count = 0;
cx.notify();
}
/// Sign the state with a random key and store it locally.
fn persist(&mut self, cx: &mut Context<Self>) {
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<Result<(), Error>> = cx.background_spawn(async move {
if let Err(error) = save_state(&client, me, &state).await {
log::warn!("failed to save inbox state: {error}");
}
Ok(())
});
task.detach();
}
}
/// Derives the inbox home screen's notification and activity lists.
///
/// Created by the inbox panel, so the database work only happens while the
/// panel is open. The persisted read state stays in the global [`Inbox`].
#[derive(Default)]
pub struct InboxStore {
/// Notifications grouped by thread root, newest activity first.
pub notifications: Arc<Vec<InboxItem>>,
/// The user's own recent git activity, newest first.
pub activity: Arc<Vec<Event>>,
/// Number of non-archived groups with an unread event.
pub unread_count: usize,
/// Copy of the global read state the current lists were derived with.
state: InboxReadState,
/// Set once the global state has been read for the current user.
state_loaded: bool,
refresh: RefreshGate,
}
impl InboxStore {
/// Create the store and derive the lists from the current global state.
pub fn new(cx: &mut Context<Self>) -> Self {
let weak = cx.entity().downgrade();
cx.defer(move |cx| {
if let Err(error) = weak.update(cx, |this, cx| this.sync_state(cx)) {
log::warn!("inbox store dropped before bootstrap could run: {error}");
}
});
Self::default()
}
/// Mark every event in the group rooted at `root` read.
pub fn mark_read(&mut self, root: EventId, cx: &mut Context<Self>) {
let Some(me) = Backend::global(cx).read(cx).current_user() else {
return;
};
let Some(group) = self.group_events(root) else {
return;
};
let all = self.all_notification_events();
let inbox = Backend::global(cx).read(cx).inbox();
inbox.update(cx, |inbox, cx| inbox.mark_read(&group, &all, me, cx));
}
/// Archive the group rooted at `root`.
pub fn mark_archived(&mut self, root: EventId, cx: &mut Context<Self>) {
let Some(me) = Backend::global(cx).read(cx).current_user() else {
return;
};
let Some(group) = self.group_events(root) else {
return;
};
let all = self.all_notification_events();
let inbox = Backend::global(cx).read(cx).inbox();
inbox.update(cx, |inbox, cx| inbox.mark_archived(&group, &all, me, cx));
}
/// Mark every known notification read.
pub fn mark_all_read(&mut self, cx: &mut Context<Self>) {
let Some(me) = Backend::global(cx).read(cx).current_user() else {
return;
};
let all = self.all_notification_events();
let inbox = Backend::global(cx).read(cx).inbox();
inbox.update(cx, |inbox, cx| inbox.mark_all_read(&all, me, cx));
}
/// Re-derive from the global state when it is loaded or changes.
pub fn sync_state(&mut self, cx: &mut Context<Self>) {
let inbox = Backend::global(cx).read(cx).inbox();
let (loaded, state) = {
let inbox = inbox.read(cx);
(inbox.is_loaded(), inbox.state().clone())
};
if !loaded {
let was_present =
self.state_loaded || !self.notifications.is_empty() || !self.activity.is_empty();
self.clear();
if was_present {
cx.notify();
}
return;
}
if !self.state_loaded {
self.state_loaded = true;
self.state = state;
self.refresh_initial(cx);
return;
}
if self.state != state {
self.state = state;
self.regroup();
self.publish_unread_count(cx);
cx.notify();
}
}
/// Handle a backend event that can change the derived lists.
pub fn handle_backend_event(&mut self, event: &BackendEvent, cx: &mut Context<Self>) {
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);
}
}
_ => {}
}
}
/// One-shot initial load, no debounce.
fn refresh_initial(&mut self, cx: &mut Context<Self>) {
debug_assert!(!self.refresh.debouncing());
@@ -198,12 +309,13 @@ impl Inbox {
fn run_refresh(&mut self, cx: &mut Context<Self>) {
self.refresh.begin();
let Some(me) = Backend::global(cx).read(cx).current_user() else {
let backend = Backend::global(cx);
let Some(me) = backend.read(cx).current_user() else {
self.refresh.abort();
return;
};
let client = Backend::global(cx).read(cx).client();
let client = backend.read(cx).client();
let state = self.state.clone();
let work = cx.background_spawn(async move {
@@ -259,6 +371,7 @@ impl Inbox {
this.notifications = Arc::new(notifications);
this.activity = Arc::new(activity);
this.unread_count = unread_count;
this.publish_unread_count(cx);
cx.notify();
this.refresh.finish()
@@ -274,13 +387,6 @@ impl Inbox {
task.detach();
}
/// Advance the cutoffs, re-derive the groups and persist the state.
fn after_state_change(&mut self, cx: &mut Context<Self>) {
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();
@@ -293,23 +399,22 @@ impl Inbox {
self.notifications = Arc::new(items);
}
/// Sign the state with a random key and store it locally.
fn persist(&mut self, cx: &mut Context<Self>) {
let Some(me) = Backend::global(cx).read(cx).current_user() else {
return;
};
/// Publish the derived unread count for the sidebar badge.
fn publish_unread_count(&self, cx: &mut Context<Self>) {
let count = self.unread_count;
let inbox = Backend::global(cx).read(cx).inbox();
inbox.update(cx, |inbox, cx| inbox.set_unread_count(count, cx));
}
let client = Backend::global(cx).read(cx).client();
let state = self.state.clone();
let task: Task<Result<(), Error>> = 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();
/// Forget everything derived for the current user.
fn clear(&mut self) {
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();
}
/// Events of the group rooted at `root`.
crates/signed_state/src/lib.rs
+1 -1
View File
@@ -13,7 +13,7 @@ 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};
pub use inbox::Inbox;
pub use inbox::{Inbox, InboxStore};
pub use nostr_sdk::prelude::Timestamp;
pub use profile::{Profile, ProfileStore};
pub use repo::RepoStore;