eleven / messenger Public

Unread status syncs across devices (server + web) (#424)

* server: per-user read watermark β€” read_marks table, POST /api/threads/{id}/read, per-user WS fan-out

Forward-only last_read_at per (user, thread), returned in /api/threads;
advancing it broadcasts a "read" frame to that user's own connections only.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* web: unread dots ride the synced read watermark

Seed dots from last_read_at on load (so they survive reloads), advance the
mark on open / live reading / refocus, and clear dots on the "read" frame
from the user's other devices.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* web: seeding a dot never clears one a silent app event follows

Recomputing purely from the last message let a silent event (todo.done)
clear a dot an earlier notifying event had lit; clear only when the
watermark covers last_at or the last message is our own.

---------
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 7e38b1273c3f9081b16d82f272914046a2a6c78b parent 5e4cf71
6 files changed, +312 −5
  • changelog/2026-07-08-read-sync.md +1 −0
  • db.go +32 −2
  • handlers_test.go +155 −0
  • hub.go +9 −0
  • main.go +42 −0
  • web/static/shell.js +73 −3
diff --git a/changelog/2026-07-08-read-sync.md b/changelog/2026-07-08-read-sync.md
new file mode 100644
index 0000000..db53415
--- /dev/null
+++ b/changelog/2026-07-08-read-sync.md
@@ -0,0 +1 @@
+πŸ€– Unread dots now sync across your devices: catch up on a chat on your phone and the dot clears on your laptop too, instead of every device keeping its own idea of what you've read. They also survive a page reload β€” a message that arrived while you were away stays marked unread until you actually open it.
diff --git a/db.go b/db.go
index 83628ae..7d69baf 100644
--- a/db.go
+++ b/db.go
@@ -301,6 +301,15 @@ CREATE TABLE IF NOT EXISTS pinned_threads (
thread_id TEXT NOT NULL,
PRIMARY KEY (user_id, thread_id)
);
+-- per-user read watermark: the time of the newest message that user has read in
+-- a thread, so "read here" clears the unread dot on their other devices too.
+-- A timestamp only β€” no content, and it never affects other members.
+CREATE TABLE IF NOT EXISTS read_marks (
+ user_id TEXT NOT NULL,
+ thread_id TEXT NOT NULL,
+ last_read_at INTEGER NOT NULL,
+ PRIMARY KEY (user_id, thread_id)
+);
-- Console-managed config for an agent (bot), keyed by its user id: the model,
-- system prompt (personality), and reference docs the sidecar applies. This is
-- operator config, not message content; the sidecar pulls it with its own token.
@@ -1478,6 +1487,10 @@ type Thread struct {
Avatar string `json:"avatar,omitempty"`
// Hidden is true when the requesting user has closed (hidden) this chat.
Hidden bool `json:"hidden,omitempty"`
+ // LastReadAt is the requesting user's read watermark β€” the time of the
+ // newest message they've read here, on any of their devices (0 = never).
+ // The client compares it to LastAt to decide the unread dot.
+ LastReadAt int64 `json:"last_read_at,omitempty"`
// Pinned is true when the requesting user keeps this chat at the top of
// their list.
Pinned bool `json:"pinned,omitempty"`
@@ -1610,7 +1623,8 @@ func (db *DB) threadsForUser(userID string) ([]Thread, error) {
COALESCE((SELECT created_at FROM messages WHERE thread_id = t.id ORDER BY id DESC LIMIT 1), 0),
COALESCE(t.avatar_blob, ''),
EXISTS(SELECT 1 FROM hidden_threads h WHERE h.thread_id = t.id AND h.user_id = m.user_id),
- EXISTS(SELECT 1 FROM pinned_threads p WHERE p.thread_id = t.id AND p.user_id = m.user_id) AS pinned
+ EXISTS(SELECT 1 FROM pinned_threads p WHERE p.thread_id = t.id AND p.user_id = m.user_id) AS pinned,
+ COALESCE((SELECT last_read_at FROM read_marks r WHERE r.thread_id = t.id AND r.user_id = m.user_id), 0)
FROM threads t
JOIN thread_members m ON m.thread_id = t.id
LEFT JOIN thread_keys k ON k.thread_id = t.id AND k.user_id = m.user_id
@@ -1628,7 +1642,7 @@ func (db *DB) threadsForUser(userID string) ([]Thread, error) {
out := []Thread{}
for rows.Next() {
var t Thread
- if err := rows.Scan(&t.ID, &t.Kind, &t.NameBlob, &t.CreatedAt, &t.Wrapped, &t.LastBlob, &t.LastAt, &t.Avatar, &t.Hidden, &t.Pinned); err != nil {
+ if err := rows.Scan(&t.ID, &t.Kind, &t.NameBlob, &t.CreatedAt, &t.Wrapped, &t.LastBlob, &t.LastAt, &t.Avatar, &t.Hidden, &t.Pinned, &t.LastReadAt); err != nil {
return nil, err
}
out = append(out, t)
@@ -1688,6 +1702,22 @@ func (db *DB) setThreadPinned(userID, threadID string, pinned bool) error {
return err
}
+// setThreadRead advances one user's read watermark for a thread. Forward-only:
+// a stale write from a lagging device can never wind the mark back and re-light
+// unread dots the user's other devices already cleared. Reports whether the
+// mark actually advanced (callers skip the fan-out when it didn't).
+func (db *DB) setThreadRead(userID, threadID string, at int64) (bool, error) {
+ res, err := db.sql.Exec(`
+ INSERT INTO read_marks(user_id, thread_id, last_read_at) VALUES(?, ?, ?)
+ ON CONFLICT(user_id, thread_id) DO UPDATE SET last_read_at = excluded.last_read_at
+ WHERE excluded.last_read_at > last_read_at`, userID, threadID, at)
+ if err != nil {
+ return false, err
+ }
+ n, err := res.RowsAffected()
+ return n > 0, err
+}
+
// clearThreadHidden reopens a thread for every user who had closed it β€” called on
// new activity so a closed chat returns to the list when it's messaged.
func (db *DB) clearThreadHidden(threadID string) error {
diff --git a/handlers_test.go b/handlers_test.go
index a99fafa..7e9982c 100644
--- a/handlers_test.go
+++ b/handlers_test.go
@@ -1016,6 +1016,161 @@ func TestThreadPinned(t *testing.T) {
}
}
+// The read watermark is per-user and forward-only: POSTing it advances
+// last_read_at in getThreads, a stale (older) write never regresses it, and
+// the "read" WS frame fans out to the reader's OTHER devices but never to
+// other members.
+func TestThreadRead(t *testing.T) {
+ ts, srv := newTestServer(t)
+
+ tokA, uidA, tid := joinedMember(t, srv, "Alice")
+ m, err := srv.db.insertMessage(tid, "blob", "author")
+ if err != nil {
+ t.Fatalf("insertMessage: %v", err)
+ }
+
+ post := func(path, body string) int {
+ resp, err := http.Post(ts.URL+path, "application/json", strings.NewReader(body))
+ if err != nil {
+ t.Fatalf("post %s: %v", path, err)
+ }
+ resp.Body.Close()
+ return resp.StatusCode
+ }
+ lastReadForAlice := func() int64 {
+ resp, err := http.Get(ts.URL + "/api/threads?token=" + tokA)
+ if err != nil {
+ t.Fatalf("getThreads: %v", err)
+ }
+ defer resp.Body.Close()
+ var out struct {
+ Threads []Thread `json:"threads"`
+ }
+ if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+ t.Fatalf("decode: %v", err)
+ }
+ for _, th := range out.Threads {
+ if th.ID == tid {
+ return th.LastReadAt
+ }
+ }
+ t.Fatalf("thread %s missing from list", tid)
+ return 0
+ }
+
+ if got := lastReadForAlice(); got != 0 {
+ t.Fatalf("before any read, last_read_at should be 0, got %d", got)
+ }
+
+ // A second device for Alice (same user, own token) and a second member, both
+ // on WebSocket β€” the read fan-out must reach the former and never the latter.
+ if err := srv.db.createKeyInvite("inv-Alice-2", uidA, "wrap", testClaimHash, 0); err != nil {
+ t.Fatalf("createKeyInvite: %v", err)
+ }
+ tokA2, ok, err := srv.db.claimKeyInvite("inv-Alice-2", "Alice", "", testClaimSecret)
+ if err != nil || !ok {
+ t.Fatalf("claimKeyInvite: ok=%v err=%v", ok, err)
+ }
+ tokB, uidB, _ := joinedMember(t, srv, "Bob")
+ if err := srv.db.addThreadMember(tid, uidB, "wrap-b"); err != nil {
+ t.Fatalf("addThreadMember: %v", err)
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ dial := func(tok string) *websocket.Conn {
+ conn, _, err := websocket.Dial(ctx, "ws"+strings.TrimPrefix(ts.URL, "http")+"/ws?token="+tok, nil)
+ if err != nil {
+ t.Fatalf("ws dial: %v", err)
+ }
+ return conn
+ }
+ readUntil := func(conn *websocket.Conn, stage string, want func(map[string]any) bool) map[string]any {
+ for {
+ _, data, err := conn.Read(ctx)
+ if err != nil {
+ t.Fatalf("%s read: %v", stage, err)
+ }
+ var msg map[string]any
+ if json.Unmarshal(data, &msg) != nil {
+ continue
+ }
+ if want(msg) {
+ return msg
+ }
+ }
+ }
+ connA2 := dial(tokA2)
+ defer connA2.CloseNow()
+ connB := dial(tokB)
+ defer connB.CloseNow()
+ // Prove both registrations landed before the POST below (Dial returns before
+ // the server's register does): a stored-message round-trip through each.
+ if err := connA2.Write(ctx, websocket.MessageText, []byte(`{"type":"msg","thread_id":"`+tid+`","blob":"warmup-a2"}`)); err != nil {
+ t.Fatalf("warmup write: %v", err)
+ }
+ readUntil(connA2, "warmup-a2", func(m map[string]any) bool { return m["blob"] == "warmup-a2" })
+ if err := connB.Write(ctx, websocket.MessageText, []byte(`{"type":"msg","thread_id":"`+tid+`","blob":"warmup-b"}`)); err != nil {
+ t.Fatalf("warmup write: %v", err)
+ }
+ readUntil(connB, "warmup-b", func(m map[string]any) bool { return m["blob"] == "warmup-b" })
+ readUntil(connA2, "warmup-b on A2", func(m map[string]any) bool { return m["blob"] == "warmup-b" })
+
+ at := m.CreatedAt + 10 // past the warmup messages' time
+ if code := post("/api/threads/"+tid+"/read", `{"token":"`+tokA+`","last_read_at":`+strconv.FormatInt(at, 10)+`}`); code != 200 {
+ t.Fatalf("read: got %d", code)
+ }
+ if got := lastReadForAlice(); got != at {
+ t.Fatalf("last_read_at should be %d, got %d", at, got)
+ }
+
+ // Alice's other device hears about it…
+ frame := readUntil(connA2, "read frame", func(m map[string]any) bool { return m["type"] == "read" })
+ if frame["thread_id"] != tid || int64(frame["last_read_at"].(float64)) != at {
+ t.Fatalf("read frame: got %v", frame)
+ }
+ // …Bob never does: the next thing on his socket is a later message, with no
+ // read frame before it (the hub is FIFO, so the read fan-out would have come
+ // first if it had been addressed to him).
+ if err := connA2.Write(ctx, websocket.MessageText, []byte(`{"type":"msg","thread_id":"`+tid+`","blob":"after"}`)); err != nil {
+ t.Fatalf("after write: %v", err)
+ }
+ _, data, err := connB.Read(ctx)
+ if err != nil {
+ t.Fatalf("bob read: %v", err)
+ }
+ var bobMsg map[string]any
+ json.Unmarshal(data, &bobMsg)
+ if bobMsg["type"] != "msg" || bobMsg["blob"] != "after" {
+ t.Fatalf("bob should only see the message, got %s", data)
+ }
+
+ // Forward-only: an older mark from a lagging device changes nothing.
+ if code := post("/api/threads/"+tid+"/read", `{"token":"`+tokA+`","last_read_at":`+strconv.FormatInt(at-5, 10)+`}`); code != 200 {
+ t.Fatalf("stale read: got %d", code)
+ }
+ if got := lastReadForAlice(); got != at {
+ t.Fatalf("stale write must not regress the mark: want %d, got %d", at, got)
+ }
+
+ // Bob has his own mark; Alice's is untouched by it.
+ if code := post("/api/threads/"+tid+"/read", `{"token":"`+tokB+`","last_read_at":`+strconv.FormatInt(at+5, 10)+`}`); code != 200 {
+ t.Fatalf("bob read: got %d", code)
+ }
+ if got := lastReadForAlice(); got != at {
+ t.Fatalf("another member's read must not move Alice's mark: want %d, got %d", at, got)
+ }
+
+ // Guardrails: a missing/zero timestamp is a 400; non-members get 403.
+ if code := post("/api/threads/"+tid+"/read", `{"token":"`+tokA+`"}`); code != 400 {
+ t.Fatalf("zero timestamp should be 400, got %d", code)
+ }
+ tokC, _, _ := joinedMember(t, srv, "Carol")
+ if code := post("/api/threads/"+tid+"/read", `{"token":"`+tokC+`","last_read_at":1}`); code != 403 {
+ t.Fatalf("non-member read should be 403, got %d", code)
+ }
+}
+
// joinedMember sets up a real "joined" member β€” a user with a device token and a
// registered public key β€” plus a thread they belong to, for exercising the
// thread-scoped APIs. Returns the device token, the user id, and the thread id.
diff --git a/hub.go b/hub.go
index ffa2f71..ee2b41a 100644
--- a/hub.go
+++ b/hub.go
@@ -191,6 +191,15 @@ type wsTypingOut struct {
Blob string `json:"blob"`
}
+// wsReadOut tells a user's other devices their read watermark for a thread
+// advanced, so their unread dots clear live. Routed only to that user's own
+// connections β€” never to other members; reading habits aren't shared.
+type wsReadOut struct {
+ Type string `json:"type"`
+ ThreadID string `json:"thread_id"`
+ LastReadAt int64 `json:"last_read_at"`
+}
+
// allowedReactions is the fixed emoji set the UI offers. Restricting it keeps
// the (plaintext) emoji column to known values β€” no arbitrary strings.
var allowedReactions = map[string]bool{
diff --git a/main.go b/main.go
index 6c0c77a..85339bc 100644
--- a/main.go
+++ b/main.go
@@ -362,6 +362,7 @@ func (s *Server) handler() http.Handler {
mux.HandleFunc("POST /api/threads/{id}/remove", s.postThreadRemove)
mux.HandleFunc("POST /api/threads/{id}/hidden", s.postThreadHidden)
mux.HandleFunc("POST /api/threads/{id}/pinned", s.postThreadPinned)
+ mux.HandleFunc("POST /api/threads/{id}/read", s.postThreadRead)
// Mini-apps (miniapps.go): scheduled "future messages" + per-thread state
// snapshots. Both carry only client-encrypted blobs; both are member-gated.
@@ -1388,6 +1389,47 @@ func (s *Server) postThreadPinned(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
+// postThreadRead advances the caller's read watermark for a thread β€” "read
+// here, read everywhere". last_read_at is the time of the newest message the
+// device has shown; when the mark advances, it fans out over WebSocket to the
+// caller's OTHER connections so their unread dots clear live. Forward-only (a
+// lagging device can't re-light dots), per-user only (never other members),
+// and just a timestamp β€” the server still can't read anything.
+func (s *Server) postThreadRead(w http.ResponseWriter, r *http.Request) {
+ var body struct {
+ Token string `json:"token"`
+ LastReadAt int64 `json:"last_read_at"`
+ }
+ if json.NewDecoder(r.Body).Decode(&body) != nil || body.LastReadAt <= 0 {
+ httpError(w, http.StatusBadRequest, "bad request")
+ return
+ }
+ if !s.tokenOK(body.Token) {
+ httpError(w, http.StatusForbidden, "not a member")
+ return
+ }
+ uid, ok := s.db.deviceTokenUser(body.Token)
+ if !ok {
+ httpError(w, http.StatusForbidden, "no user for token")
+ return
+ }
+ threadID := r.PathValue("id")
+ if !s.db.threadMember(threadID, uid) {
+ httpError(w, http.StatusForbidden, "not a thread member")
+ return
+ }
+ advanced, err := s.db.setThreadRead(uid, threadID, body.LastReadAt)
+ if err != nil {
+ httpError(w, http.StatusInternalServerError, "db error")
+ return
+ }
+ if advanced {
+ out, _ := json.Marshal(wsReadOut{Type: "read", ThreadID: threadID, LastReadAt: body.LastReadAt})
+ s.hub.broadcast <- outbound{members: map[string]bool{uid: true}, data: out}
+ }
+ writeJSON(w, http.StatusOK, map[string]any{"ok": true})
+}
+
// postThreadRemove removes another member from a group. Any member may remove
// any other (matching how adding/renaming work today β€” the group is flat). Like
// leaving, the key isn't rotated: the server just stops delivering to them.
diff --git a/web/static/shell.js b/web/static/shell.js
index f51d72f..aa3ab6e 100644
--- a/web/static/shell.js
+++ b/web/static/shell.js
@@ -476,6 +476,10 @@ const threadPreview = {}; // thread id -> last-message snippet
// recovered after a reload (the pending chat offers "re-send" to mint a fresh one).
const inviteLinks = {};
const unread = {}; // thread id -> bool
+// thread id -> read watermark (the newest message time we've read, in server
+// seconds). Synced from the server so "read on one device" clears the dot on
+// the others; advanced locally first so the UI never waits on the POST.
+const readMarks = {};
const hidden = {}; // thread id -> bool: closed (hidden from the list), synced from the server
const pinned = {}; // thread id -> bool: pinned to the top of the list, synced from the server
let chatFilter = ""; // lowercased substring filtering the thread list (search box)
@@ -1253,15 +1257,47 @@ async function loadThreads() {
threadName[t.id] = await displayName(t);
hidden[t.id] = !!t.hidden; // closed chats (server-synced) are dropped from the list
pinned[t.id] = !!t.pinned; // pinned chats (server-synced) float to the top
+ // The server's watermark can trail a POST we just made β€” never regress ours.
+ readMarks[t.id] = Math.max(readMarks[t.id] || 0, t.last_read_at || 0);
if (t.last_blob && threadKeys[t.id]) {
try {
- threadPreview[t.id] = previewText(t, await decrypt(threadKeys[t.id], t.last_blob));
+ const payload = await decrypt(threadKeys[t.id], t.last_blob);
+ threadPreview[t.id] = previewText(t, payload);
+ // Seed the unread dot from the synced watermark, applying the same
+ // gates as a live frame: your own last message never lights it (you
+ // sent it, or replying elsewhere means you read it), and app events
+ // only light it if the app says so. A silent app event neither lights
+ // nor clears β€” an earlier lit dot stays lit, matching the live path.
+ if (t.id !== activeId) {
+ if ((t.last_at || 0) <= readMarks[t.id] || ownMessage(payload, myUserId, session.name)) {
+ unread[t.id] = false;
+ } else if (!isAppType(payload.type) || appUnread(payload)) {
+ unread[t.id] = true;
+ }
+ }
} catch {}
}
}
syncThreadKeysToSW();
}
+// Advance this user's read watermark for a thread to its newest message β€” the
+// server relays it to their other devices so the dot clears everywhere, and
+// remembers it so a reload doesn't resurrect it. Local-first and deduped: the
+// mark moves before the POST, and an already-current mark makes this a no-op
+// (so calling it on every live message while a thread is open stays cheap).
+function markThreadRead(id) {
+ const t = threads.find((x) => x.id === id);
+ const at = (t && t.last_at) || 0;
+ if (!at || at <= (readMarks[id] || 0)) return;
+ readMarks[id] = at;
+ fetch(`/api/threads/${encodeURIComponent(id)}/read`, {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify({ token: session.token, last_read_at: at }),
+ }).catch(() => {});
+}
+
// Hand the service worker the per-thread keys (+ name/kind) it needs to decrypt
// push payloads. Stored in IndexedDB (the SW can't read localStorage).
function syncThreadKeysToSW() {
@@ -1329,7 +1365,7 @@ function startHeartbeat() {
document.addEventListener("visibilitychange", () => {
if (document.visibilityState === "visible") {
checkForUpdate();
- resyncActive();
+ resyncActive(); // its tail marks the active thread read once caught up
focusChatComposer();
clearNotifications();
}
@@ -1338,7 +1374,18 @@ function startHeartbeat() {
// another window/app, where the tab stayed "visible" so the handler above
// doesn't fire). Both routes drop the caret into the composer so you're ready
// to type β€” the Mac app does the same on didBecomeActive.
- window.addEventListener("focus", focusChatComposer);
+ window.addEventListener("focus", () => {
+ focusChatComposer();
+ onSawActiveThread();
+ });
+}
+
+// Coming back to the app with a thread open means you've now seen whatever
+// arrived in it while you were away β€” advance its synced read watermark
+// (messages landing in the ACTIVE thread never set the local dot, so without
+// this the user's other devices would keep showing one).
+function onSawActiveThread() {
+ if (activeId) markThreadRead(activeId);
}
// Being in the app means you've seen what the banners were shouting about: close
@@ -2241,6 +2288,7 @@ async function openThread(id) {
activeId = id;
localStorage.setItem(ACTIVE_KEY, id); // remember where we are, to restore on refresh
unread[id] = false;
+ markThreadRead(id); // …and tell the server, so your other devices clear too
// Opening a closed chat reopens it (locally + server, synced to your devices).
if (hidden[id]) {
hidden[id] = false;
@@ -4634,6 +4682,9 @@ async function resyncActive() {
await loadThreads();
renderThreadList();
await catchUpActiveMessages();
+ // Whatever we just caught up on in the open thread, the user is looking at
+ // β€” advance the synced watermark (only now that last_at is current).
+ if (document.hasFocus()) onSawActiveThread();
} catch (e) {
console.error("resync", e);
} finally {
@@ -4763,6 +4814,20 @@ function connectWS() {
refreshAvatars();
return;
}
+ if (msg.type === "read") {
+ // Our own read watermark advanced on another of this user's devices (the
+ // server routes this frame only to the owner's connections) β€” clear the
+ // dot if that mark covers the thread's newest message. Also echoes back
+ // to the device that POSTed it; the max() makes that a no-op.
+ const tid = msg.thread_id;
+ readMarks[tid] = Math.max(readMarks[tid] || 0, msg.last_read_at || 0);
+ const t = threads.find((x) => x.id === tid);
+ if (t && unread[tid] && (t.last_at || 0) <= readMarks[tid]) {
+ unread[tid] = false;
+ renderThreadList();
+ }
+ return;
+ }
if (msg.type === "apps_changed") {
// Something in the app set changed: a member app (bot build / App Studio),
// or an admin-installed custom app (an ES module, which needs a reload).
@@ -4900,6 +4965,8 @@ function connectWS() {
}
const at = threads.find((x) => x.id === tid);
if (at) at.last_at = msg.created_at || Math.floor(Date.now() / 1000);
+ // Watching the app live counts as reading β€” advance the synced watermark.
+ if (tid === activeId && document.hasFocus()) markThreadRead(tid);
renderThreadList();
return;
}
@@ -4926,6 +4993,9 @@ function connectWS() {
maybeNativeNotify(tid, payload, preview);
}
}
+ // Actually reading it live (thread open, window focused) advances the
+ // synced watermark β€” your other devices' dots clear without a tap here.
+ if (tid === activeId && document.hasFocus()) markThreadRead(tid);
renderThreadList();
// If the just-joined invitee's message cleared the pending state, drop the
// waiting card and reveal the normal feed/composer.