feat: sync notification read state across sessions

CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to Allthing are recorded here. The format follows [Keep a Ch
 
 ### Added
 
+- Sync notification read progress across signed-in sessions with a private, attested SpacetimeDB watermark. The existing local cache works offline; a delayed background connection uses separate credentials without creating chat presence or DM keys.
 - Videos up to 250 MB upload in 8 MB chunks. Retry continues from the last acknowledged chunk; Remove cancels the pending upload. Preparation and finishing have visible states, and the cast stays disabled until its attachments are ready.
 - Compatible MP4 and MOV videos are prepared for HLS from bounded metadata and file slices, preserving their frames and sound without copying the entire video into WebAssembly. Existing image limits and storage safeguards remain.
 

crates/allthing-core/src/lib.rs

@@ -32,6 +32,7 @@ pub mod message;
 pub mod metadata;
 pub mod model;
 pub mod normalize;
+pub mod notification_state;
 pub mod notifications;
 pub mod params;
 pub mod people;

crates/allthing-core/src/notification_state.rs

@@ -0,0 +1,112 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//! Monotonic notification read state. Local progress works offline; only an
+//! acknowledged write or the caller's private subscription advances `remote`.
+
+/// A corrupt cache or provider clock cannot mark arbitrary future events read.
+pub fn valid_seen(at: i64, now: i64) -> i64 {
+    if at < 0 || at > now.saturating_add(60) { 0 } else { at.min(now.max(0)) }
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct ReadState {
+    seen: i64,
+    remote: i64,
+    baseline: Option<i64>,
+}
+
+impl ReadState {
+    pub fn new(local: i64, now: i64) -> Self {
+        Self { seen: valid_seen(local, now), ..Self::default() }
+    }
+
+    pub fn seen(&self) -> i64 {
+        self.seen
+    }
+
+    pub fn baseline(&self) -> Option<i64> {
+        self.baseline
+    }
+
+    pub fn observe_local(&mut self, at: i64, now: i64) {
+        self.seen = self.seen.max(valid_seen(at, now));
+    }
+
+    pub fn observe_remote(&mut self, at: i64, now: i64) {
+        self.remote = self.remote.max(valid_seen(at, now));
+        self.seen = self.seen.max(self.remote);
+    }
+
+    /// Capture the prior server state before uploading this visit's pending
+    /// reads. It lets a cold notification page correct its initial row tint.
+    pub fn hydrated(&mut self) {
+        self.baseline.get_or_insert(self.remote);
+    }
+
+    pub fn pending(&self) -> Option<i64> {
+        (self.baseline.is_some() && self.seen > self.remote).then_some(self.seen)
+    }
+
+    pub fn acknowledged(&mut self, at: i64) {
+        self.remote = self.remote.max(at.min(self.seen));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn merges_offline_progress_without_overwriting_a_newer_session() {
+        let mut state = ReadState::new(100, 1000);
+        assert_eq!(state.pending(), None);
+        state.observe_local(300, 1000);
+        state.observe_remote(200, 1000);
+        state.hydrated();
+        assert_eq!(state.baseline(), Some(200));
+        assert_eq!(state.pending(), Some(300));
+        state.observe_remote(400, 1000);
+        state.acknowledged(300);
+        state.observe_remote(50, 1000);
+        assert_eq!(state.seen(), 400);
+        assert_eq!(state.pending(), None);
+        assert_eq!(state.baseline(), Some(200));
+    }
+
+    #[test]
+    fn failed_writes_keep_progress_and_an_old_ack_cannot_clear_a_new_read() {
+        let mut state = ReadState::new(100, 1000);
+        state.hydrated();
+        let sent = state.pending().unwrap();
+        assert_eq!(state.pending(), Some(sent)); // no acknowledgement on failure
+        state.observe_local(500, 1000);
+        state.acknowledged(sent);
+        assert_eq!(state.pending(), Some(500));
+        state.acknowledged(500);
+        assert_eq!(state.pending(), None);
+    }
+
+    #[test]
+    fn delayed_initial_snapshot_does_not_include_this_visits_pending_mark() {
+        let mut state = ReadState::new(0, 1000);
+        state.observe_local(900, 1000);
+        state.observe_remote(700, 1000);
+        state.hydrated();
+        assert_eq!(state.baseline(), Some(700));
+        assert_eq!(state.pending(), Some(900));
+        state.observe_remote(950, 1000);
+        state.hydrated();
+        assert_eq!(state.baseline(), Some(700));
+    }
+
+    #[test]
+    fn untrusted_timestamps_cannot_poison_a_valid_marker() {
+        for at in [-1, i64::MIN, i64::MAX, 1061] {
+            assert_eq!(valid_seen(at, 1000), 0);
+        }
+        assert_eq!(valid_seen(1060, 1000), 1000);
+        let mut state = ReadState::new(i64::MAX, 1000);
+        state.observe_remote(900, 1000);
+        state.observe_local(i64::MAX, 1000);
+        assert_eq!(state.seen(), 900);
+    }
+}

crates/allthing-web/src/components/notifications.rs

@@ -4,8 +4,8 @@
 //! app starts, every three minutes while the page is in view, and when it
 //! is back in view or its window regains focus (a read within a minute is
 //! reused: `api`), and the newest time the reader has seen, which this
-//! browser keeps by account (`allthing-notifications-seen` in localStorage;
-//! the server keeps none). The header's bell counts the groups newer than
+//! browser caches by account (`allthing-notifications-seen` in localStorage)
+//! and SpacetimeDB shares with the account's other verified sessions. The header's bell counts the groups newer than
 //! that, and the notifications page (`pages::notifications`), which reads
 //! the whole payload from the same answer, marks them seen. Not in the
 //! static copy, which has no sign-in.
@@ -14,6 +14,7 @@ use std::collections::BTreeMap;
 use std::sync::Arc;
 use std::time::Duration;
 
+use allthing_core::notification_state::valid_seen;
 use allthing_core::notifications::{NOTIFICATIONS_PATH, Times, badge, bell_label};
 use leptos::prelude::*;
 use leptos::task::spawn_local;
@@ -47,6 +48,10 @@ struct State {
     loading: bool,
     /// The newest time `whose` has seen (Unix seconds), 0 for never.
     seen: i64,
+    /// Fences in-flight reads even if A signs out and signs back in as A.
+    generation: u64,
+    /// Prior remote state, captured before this visit's pending read is sent.
+    remote_baseline: Option<i64>,
 }
 
 fn storage() -> Option<web_sys::Storage> {
@@ -60,7 +65,21 @@ fn kept() -> BTreeMap<String, i64> {
 }
 
 fn seen_of(fid: u64) -> i64 {
-    kept().get(&fid.to_string()).copied().unwrap_or(0)
+    valid_seen(kept().get(&fid.to_string()).copied().unwrap_or(0), now())
+}
+
+fn now() -> i64 {
+    (js_sys::Date::now() / 1000.0) as i64
+}
+
+fn keep_seen(fid: u64, at: i64) -> i64 {
+    let at = valid_seen(at, now()).max(seen_of(fid));
+    let mut all = kept();
+    all.insert(fid.to_string(), at);
+    if let (Some(storage), Ok(text)) = (storage(), serde_json::to_string(&all)) {
+        let _ = storage.set_item(SEEN_KEY, &text);
+    }
+    at
 }
 
 impl Notifications {
@@ -76,8 +95,10 @@ impl Notifications {
             if notifications.whose() != fid {
                 api::forget(NOTIFICATIONS_PATH);
                 api::forget(allthing_core::following::FOLLOWING_PATH);
-                notifications.0.set(State { whose: fid, seen: fid.map_or(0, seen_of), ..State::default() });
+                let generation = notifications.generation().wrapping_add(1);
+                notifications.0.set(State { whose: fid, seen: fid.map_or(0, seen_of), generation, ..State::default() });
             }
+            crate::notification_sync::account(fid, notifications);
             if fid.is_some() {
                 notifications.refresh();
             }
@@ -94,16 +115,22 @@ impl Notifications {
         let woken = move |_| {
             if !document().hidden() {
                 notifications.refresh();
+                crate::notification_sync::wake();
             }
         };
         let _ = window_event_listener_untyped("visibilitychange", woken);
         let _ = window_event_listener_untyped("focus", woken);
+        let _ = window_event_listener_untyped("online", move |_| crate::notification_sync::wake());
         // Seen in another of this browser's tabs.
         let _ = window_event_listener(leptos::ev::storage, move |ev| {
             if ev.key().as_deref() == Some(SEEN_KEY)
                 && let Some(fid) = notifications.whose()
             {
-                notifications.0.update(|state| state.seen = seen_of(fid));
+                let seen = seen_of(fid).max(notifications.seen_by(fid));
+                if seen > notifications.seen_by(fid) {
+                    notifications.0.update(|state| state.seen = seen);
+                }
+                crate::notification_sync::seen(fid, seen);
             }
         });
         notifications
@@ -119,6 +146,34 @@ impl Notifications {
         self.whose() == Some(fid)
     }
 
+    pub fn generation(self) -> u64 {
+        self.0.with_untracked(|state| state.generation)
+    }
+
+    pub fn same_session(self, fid: u64, generation: u64) -> bool {
+        self.are_for(fid) && self.generation() == generation
+    }
+
+    pub fn sync_baseline(self, fid: u64) -> Option<i64> {
+        self.0.with(|state| if state.whose == Some(fid) { state.remote_baseline } else { None })
+    }
+
+    /// A private subscription has advanced. Do not echo it as another write.
+    pub fn synced(self, fid: u64, seen: i64, baseline: Option<i64>) {
+        if !self.are_for(fid) {
+            return;
+        }
+        let seen = keep_seen(fid, seen).max(self.seen_by(fid));
+        if self.0.with_untracked(|s| seen > s.seen || (s.remote_baseline.is_none() && baseline.is_some())) {
+            self.0.update(|state| {
+                state.seen = seen;
+                if state.remote_baseline.is_none() {
+                    state.remote_baseline = baseline;
+                }
+            });
+        }
+    }
+
     /// The first page's times once its latest refresh succeeded (tracked).
     /// Old times must not start a page read behind a refresh that can fail.
     pub fn times(self) -> Option<Arc<Times>> {
@@ -152,14 +207,15 @@ impl Notifications {
     /// within a minute can be reused or is on its way (`api::get`).
     pub fn refresh(self) {
         let Some(fid) = self.0.with_untracked(|state| state.whose.filter(|_| !state.loading)) else { return };
+        let generation = self.generation();
         // Coalesce before spawning: a second caller otherwise waits in
         // api::get for the first read. A throttled tab can resume that
         // waiter after the cache has expired and start a redundant read.
         self.0.update(|state| state.loading = true);
         spawn_local(async move {
             let read = api::get::<Times>(NOTIFICATIONS_PATH).await;
             // Signed out, or someone else, meanwhile.
-            if self.whose() != Some(fid) {
+            if !self.same_session(fid, generation) {
                 return;
             }
             self.0.update(|state| {
@@ -186,15 +242,13 @@ impl Notifications {
     /// `fid` has seen everything up to `newest` (Unix seconds): the bell
     /// clears, here and in the browser's other tabs.
     pub fn mark_seen(self, fid: u64, newest: i64) {
+        let newest = valid_seen(newest, now());
         if !self.are_for(fid) || newest <= self.seen_by(fid) {
             return;
         }
+        let newest = keep_seen(fid, newest);
         self.0.update(|state| state.seen = newest);
-        let mut all = kept();
-        all.insert(fid.to_string(), newest);
-        if let (Some(storage), Ok(text)) = (storage(), serde_json::to_string(&all)) {
-            let _ = storage.set_item(SEEN_KEY, &text);
-        }
+        crate::notification_sync::seen(fid, newest);
     }
 
     /// How many groups of the first page the reader hasn't seen (tracked).

crates/allthing-web/src/keys.rs

@@ -33,6 +33,8 @@ struct Database {
 const APP_KEYS: Database = Database { name: "allthing-keys", store: "keys" };
 /// SpacetimeDB identities and message keys.
 const REALTIME: Database = Database { name: "allthing-stdb", store: "identities" };
+/// Notification sync tokens have no message keys and cannot overwrite them.
+const NOTIFICATION_SYNC: Database = Database { name: "allthing-notification-sync", store: "identities" };
 /// The account a key made to sign in with is kept for, until the account
 /// that approves it is known.
 pub const SIGNING_IN: u64 = 0;
@@ -175,6 +177,34 @@ pub struct Realtime {
     pub key: Option<(CryptoKey, [u8; 32])>,
 }
 
+#[derive(Clone, serde::Deserialize, serde::Serialize)]
+pub struct NotificationIdentity {
+    pub fid: u64,
+    pub identity: String,
+    pub token: String,
+}
+
+pub async fn notification_identity(fid: u64) -> Result<Option<NotificationIdentity>, KeyError> {
+    let found =
+        request(NOTIFICATION_SYNC, IdbTransactionMode::Readonly, |store| store.get(&JsValue::from_f64(fid as f64)))
+            .await?;
+    let Some(identity) = field(&found, "identity").as_string() else { return Ok(None) };
+    let Some(token) = field(&found, "token").as_string() else { return Ok(None) };
+    Ok(Some(NotificationIdentity { fid, identity, token }))
+}
+
+pub async fn keep_notification_identity(kept: &NotificationIdentity) -> Result<(), KeyError> {
+    let record = Object::new();
+    for (name, value) in [
+        ("fid", JsValue::from_f64(kept.fid as f64)),
+        ("identity", JsValue::from_str(&kept.identity)),
+        ("token", JsValue::from_str(&kept.token)),
+    ] {
+        Reflect::set(&record, &name.into(), &value).map_err(failed)?;
+    }
+    request(NOTIFICATION_SYNC, IdbTransactionMode::Readwrite, |store| store.put(&record)).await.map(|_| ())
+}
+
 /// What this browser keeps for `fid`'s Messages, if anything.
 pub async fn realtime(fid: u64) -> Result<Option<Realtime>, KeyError> {
     let found =

crates/allthing-web/src/lib.rs

@@ -15,11 +15,15 @@ mod keys;
 mod leanings;
 mod miniapps;
 mod nav;
+#[cfg(not(feature = "static-host"))]
+mod notification_sync;
 mod pages;
 mod preload;
 mod press;
 mod seen;
 mod split;
+#[cfg(not(feature = "static-host"))]
+mod stdb_bridge;
 mod util;
 mod video;