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

refacotr

Browse Source
This commit is contained in:
reya
2026-09-11 08:52:04 +07:00
parent b7a020767e
commit b7221ef814
8 changed files with 895 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/signed_core/src/inbox.rs
+67 -13
View File
@@ -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);
crates/signed_state/Cargo.toml
+1
View File
@@ -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"
crates/signed_state/src/backend.rs
+146 -66
View File
@@ -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<PublicKey>,
/// User's inbox, including notifications and recent activity.
inbox: Entity<Inbox>,
/// 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<HashSet<RepoAddr>>,
}
@@ -112,6 +113,7 @@ impl Backend {
}
pub(crate) fn new(client: Client, signer: UniversalSigner, cx: &mut Context<Self>) -> Self {
let weak = cx.entity().downgrade();
let pump_client = client.clone();
let pump: Task<Result<(), Error>> = 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<Self>) {
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<Result<(), Error>> = 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<Self>) {
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<RelayMetadata>)> = [
@@ -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<Result<(), Error>> = 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<Inbox> {
self.inbox.clone()
}
/// Get the current user's public key.
pub fn current_user(&self) -> Option<PublicKey> {
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<String>, cx: &mut Context<Self>) {
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<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);
}
/// 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 {
self.subscribe_bootstrap(filters::notifications(me), cx);
self.subscribe_bootstrap(vec![filters::authored_activity(me)], cx);
let relays: HashSet<RelayUrl> = RepoListStore::global(cx)
.read(cx)
.announcements_of(&me)
.into_iter()
.flat_map(|announcement| announcement.relays)
.collect();
if !relays.is_empty() {
let relays: Vec<RelayUrl> = 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 {
<T as AsyncSignEvent>::Error: std::error::Error + Send + Sync + 'static,
<T as AsyncNip44>::Error: std::error::Error + Send + Sync + 'static,
{
let task: Task<Result<(), Error>> = 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<RelayUrl>,
@@ -1185,13 +1261,13 @@ impl Backend {
) {
let client = self.client.clone();
let task: Task<Result<(), Error>> = 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<Result<(), Error>> = 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<Self>) {
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<Result<(), Error>> = 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<Self>) {
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.
crates/signed_state/src/inbox.rs
+431
View File
@@ -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<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,
}
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);
}
/// 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;
};
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<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();
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();
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<Self>) {
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<Self>) {
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>) {
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<Result<(), Error>> = 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>) {
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<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.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<Vec<Event>> {
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<Event> {
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<Option<InboxReadState>, 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<Event>, HashMap<EventId, Event>), Error> {
let mut notifications: Vec<Event> = Vec::new();
let mut by_id: HashMap<EventId, Event> = 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<EventId> = notifications.iter().flat_map(event_references).collect();
let mut seen: HashSet<EventId> = 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<Item = EventId> + '_ {
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())
})
}
crates/signed_state/src/lib.rs
+7 -11
View File
@@ -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<Path>,
repos_root: impl Into<PathBuf>,
scan_paths: Vec<PathBuf>,
cx: &mut App,
) -> Entity<Backend> {
) {
// 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<Backend> {
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
}
crates/signed_state/src/refresh.rs
-5
View File
@@ -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.