amadan
web+server: fix notification catch-up, deep-linking, and own-message suppression (#30)
* web: sync missed messages + deep-link notifications to their thread Two fixes to the notification → chat flow: - Catch up on missed messages. The WebSocket hub only fans out *live* messages — nothing is replayed on connect — so anything that arrived while the socket was down (phone asleep, tab hidden, a tapped notification re-focusing a now-stale tab) never reached the open feed until you left and re-entered the thread. Add resyncActive(): on reconnect (guarded so it skips the first connect) and on refocus, it refreshes the thread list and appends any messages the open feed is missing. Incremental/append-only so a plain refocus doesn't clear the feed or fight the scroll position. - Deep-link notifications. The push payload already carried the thread id but the click handler ignored it. Carry it on the notification and, on tap, open straight to that thread: postMessage to an already-open tab (SPA, no reload), or ride a #thread=<id> fragment on a cold start. The fragment keeps the thread id client-side (never in the server's access log), matching how invite secrets are handled. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * server: don't push your own messages to your other devices Own-message suppression only skipped the exact sending device (sub.Token == senderToken). Since the multi-device refactor each device carries its own token, so a message sent from one device still pushed to the sender's *other* (idle) devices — a notification for your own message. Exclude by owning user, not just exact token: resolve the sender's user from their device token and skip every subscription belonging to that user. Factor the selection out of sendToRecipients into a pure, testable selectRecipients, and add a regression test. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> ---------
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
4 files changed,
+191
−27
db_test.go+41 −0push.go+46 −22web/static/chat.js+87 −3web/static/sw.js+17 −2
diff --git a/db_test.go b/db_test.go| index cbd8f8e..c881f0e 100644 |
| --- a/db_test.go |
| +++ b/db_test.go |
| @@ -1,6 +1,7 @@ |
| package main |
| import ( |
| + "fmt" |
| "testing" |
| "time" |
| ) |
| @@ -188,6 +189,46 @@ func TestPushSubscriptions(t *testing.T) { |
| } |
| } |
| +// A message must never push to any of the sender's own devices — not just the |
| +// one it was sent from. Each device carries its own token now, so excluding by |
| +// the exact sending token alone let the sender's *other* devices buzz for their |
| +// own message (the reported regression). |
| +func TestSelectRecipientsExcludesAllSenderDevices(t *testing.T) { |
| + db := newTestDB(t) |
| + s := &Server{db: db} |
| + |
| + mkDevice := func(id, user string) string { |
| + if err := db.createKeyInvite(id, user, "w", 0); err != nil { |
| + t.Fatalf("createKeyInvite %s: %v", id, err) |
| + } |
| + tok, ok, err := db.claimKeyInvite(id, "dev") |
| + if err != nil || !ok { |
| + t.Fatalf("claimKeyInvite %s: ok=%v err=%v", id, ok, err) |
| + } |
| + return tok |
| + } |
| + aliceLaptop := mkDevice("a1", "alice") |
| + alicePhone := mkDevice("a2", "alice") |
| + bobPhone := mkDevice("b1", "bob") |
| + |
| + for i, tok := range []string{aliceLaptop, alicePhone, bobPhone} { |
| + db.addSubscription(PushSub{Endpoint: fmt.Sprintf("https://e/%d", i), Token: tok, P256dh: "p", Auth: "a"}) |
| + } |
| + subs, _ := db.listSubscriptions() |
| + |
| + // Alice sends from her laptop → only Bob is notified; neither Alice device is. |
| + got := s.selectRecipients(subs, aliceLaptop, true, nil) |
| + if len(got) != 1 || got[0].Token != bobPhone { |
| + t.Fatalf("expected only Bob's phone, got %d: %+v", len(got), got) |
| + } |
| + |
| + // And an active recipient is skipped too (they're already reading). |
| + s.markActive(bobPhone) |
| + if got := s.selectRecipients(subs, aliceLaptop, true, nil); len(got) != 0 { |
| + t.Fatalf("active recipient should be skipped, got %d: %+v", len(got), got) |
| + } |
| +} |
| + |
| func TestReactions(t *testing.T) { |
| db := newTestDB(t) |
| m, _ := db.insertMessage("t1", "blob", "author-a") |
diff --git a/push.go b/push.go| index bbba99c..739894b 100644 |
| --- a/push.go |
| +++ b/push.go |
| @@ -85,8 +85,9 @@ func (db *DB) ensureVAPID() (pub, priv string, err error) { |
| // the service worker decrypts it locally with the group key. Expired subscriptions |
| // (404/410) are pruned. Best-effort; runs in its own goroutine. |
| // pushNewMessage notifies everyone except the sender — you don't want a push on |
| -// your own devices for a message you just sent. The sender is identified by the |
| -// invite token from their WebSocket (their subscriptions carry the same token). |
| +// any of your own devices for a message you just sent. The sender's device token |
| +// (from their WebSocket) identifies both that device and, via its owning user, |
| +// all their other devices, which selectRecipients excludes. |
| func (s *Server) pushNewMessage(blob, threadID, senderToken string) { |
| members, err := s.db.memberUserIDs(threadID) |
| if err != nil { |
| @@ -115,36 +116,59 @@ func (s *Server) sendTestPush() int { |
| // so fanning out turns O(N) sequential latency into ~O(N/concurrency). |
| const maxPushConcurrency = 32 |
| -// sendToRecipients delivers a payload to subscriptions, skipping any whose token |
| -// equals exceptToken and (when skipActive) any token currently active on some |
| -// device. When recipients is non-nil, only subscriptions whose token belongs to |
| -// a user in that set are notified (thread membership). Prunes dead subs |
| -// (404/410). Sends run concurrently, bounded by maxPushConcurrency. Returns the |
| -// number of successful sends. |
| +// selectRecipients decides which subscriptions should receive a payload. It |
| +// excludes the sender — by exact device token AND by owning user, so none of the |
| +// sender's *other* devices buzz for a message they sent (each device carries its |
| +// own token now, so an exact-token match alone would miss the sibling devices). |
| +// When skipActive it also drops any device the recipient is currently reading on, |
| +// and when recipients is non-nil it keeps only devices whose owner is in that set |
| +// (thread membership). |
| +func (s *Server) selectRecipients(subs []PushSub, exceptToken string, skipActive bool, recipients map[string]bool) []PushSub { |
| + senderUser := "" |
| + if exceptToken != "" { |
| + senderUser, _ = s.db.deviceTokenUser(exceptToken) |
| + } |
| + out := make([]PushSub, 0, len(subs)) |
| + for _, sub := range subs { |
| + if exceptToken != "" && sub.Token == exceptToken { |
| + continue // the exact device that sent it |
| + } |
| + // Resolve the owner once — needed for sender-user exclusion and membership. |
| + var uid string |
| + var known bool |
| + if senderUser != "" || recipients != nil { |
| + uid, known = s.db.deviceTokenUser(sub.Token) |
| + } |
| + if senderUser != "" && known && uid == senderUser { |
| + continue // another of the sender's own devices — don't notify yourself |
| + } |
| + if skipActive && s.isActive(sub.Token) { |
| + continue // they're reading the chat on this device right now |
| + } |
| + if recipients != nil && (!known || !recipients[uid]) { |
| + continue // not a member of this thread |
| + } |
| + out = append(out, sub) |
| + } |
| + return out |
| +} |
| + |
| +// sendToRecipients delivers a payload to the subscriptions selectRecipients |
| +// picks. Prunes dead subs (404/410). Sends run concurrently, bounded by |
| +// maxPushConcurrency. Returns the number of successful sends. |
| func (s *Server) sendToRecipients(payload []byte, kind, exceptToken string, skipActive bool, recipients map[string]bool) int { |
| subs, err := s.db.listSubscriptions() |
| if err != nil { |
| log.Println("push: list subs:", err) |
| return 0 |
| } |
| - log.Printf("push[%s]: %d subscription(s)", kind, len(subs)) |
| + targets := s.selectRecipients(subs, exceptToken, skipActive, recipients) |
| + log.Printf("push[%s]: %d subscription(s), %d after exclusions", kind, len(subs), len(targets)) |
| sem := make(chan struct{}, maxPushConcurrency) |
| var wg sync.WaitGroup |
| var sent atomic.Int64 |
| - for _, sub := range subs { |
| - if exceptToken != "" && sub.Token == exceptToken { |
| - continue // don't notify the sender's own devices |
| - } |
| - if skipActive && s.isActive(sub.Token) { |
| - continue // they're reading the chat on another device right now |
| - } |
| - if recipients != nil { |
| - uid, ok := s.db.deviceTokenUser(sub.Token) |
| - if !ok || !recipients[uid] { |
| - continue // not a member of this thread |
| - } |
| - } |
| + for _, sub := range targets { |
| wg.Add(1) |
| sem <- struct{}{} |
| go func(sub PushSub) { |
diff --git a/web/static/chat.js b/web/static/chat.js| index 611a652..a9e682d 100644 |
| --- a/web/static/chat.js |
| +++ b/web/static/chat.js |
| @@ -259,6 +259,7 @@ const unread = {}; // thread id -> bool |
| let chatFilter = ""; // lowercased substring filtering the thread list (search box) |
| let activeId = null; |
| let ws = null; |
| +let everConnected = false; // has the socket ever opened? (distinguishes reconnect from first connect) |
| let pingTimer = null; // keepalive ping interval while the socket is open |
| let reconnectStatusTimer = null; // delays the "Reconnecting…" pill past a brief blip |
| @@ -281,6 +282,13 @@ async function boot() { |
| connectWS(); |
| registerSW(session); // best-effort; subscribes if notifications already on |
| startHeartbeat(); |
| + // Tapping a notification while a tab is already open posts us the thread to |
| + // open (the SW can't navigate a live SPA) — jump straight to it. |
| + if (navigator.serviceWorker) { |
| + navigator.serviceWorker.addEventListener("message", (e) => { |
| + if (e.data && e.data.type === "open-thread" && e.data.thread) openThreadById(e.data.thread); |
| + }); |
| + } |
| } catch (e) { |
| console.error(e); |
| app.innerHTML = |
| @@ -503,9 +511,16 @@ function startHeartbeat() { |
| beat(); |
| setInterval(beat, 30000); |
| document.addEventListener("visibilitychange", beat); |
| - // Returning to the app (tab/window visible again) → check for a newer build. |
| + // Returning to the app (tab/window visible again) → check for a newer build, |
| + // and catch up on any messages missed while hidden. This is what fixes a tapped |
| + // notification landing on a stale thread: the socket may still look OPEN (the OS |
| + // froze it without firing onclose), so a reconnect never fires — but refocus |
| + // always does, and re-fetches the missed messages. |
| document.addEventListener("visibilitychange", () => { |
| - if (document.visibilityState === "visible") checkForUpdate(); |
| + if (document.visibilityState === "visible") { |
| + checkForUpdate(); |
| + resyncActive(); |
| + } |
| }); |
| } |
| @@ -696,7 +711,33 @@ function renderApp() { |
| }); |
| renderThreadList(); |
| - if (!isCoarse && threads.length) openThread(threads[0].id); |
| + // A notification tapped while the app was closed opens with #thread=<id> — |
| + // land straight in it (on any platform). Otherwise desktop shows the first |
| + // chat; touch stays on the list. |
| + const deep = initialThreadId(); |
| + if (deep && threads.some((t) => t.id === deep)) { |
| + history.replaceState(null, "", location.pathname + location.search); // consume the deep link |
| + openThread(deep); |
| + } else if (!isCoarse && threads.length) { |
| + openThread(threads[0].id); |
| + } |
| +} |
| + |
| +// The thread id from a deep-link fragment (#thread=<id>), or null. Kept in the |
| +// fragment (not a query) so it never reaches the server — see sw.js. |
| +function initialThreadId() { |
| + const m = /(?:^|[#&])thread=([^&]+)/.exec(location.hash || ""); |
| + return m ? decodeURIComponent(m[1]) : null; |
| +} |
| + |
| +// Open a thread by id, tolerating one we haven't loaded yet (e.g. a notification |
| +// for a thread created while this tab was asleep). Used by the SW deep-link. |
| +async function openThreadById(id) { |
| + if (!threads.some((t) => t.id === id)) { |
| + await loadThreads(); |
| + renderThreadList(); |
| + } |
| + if (threads.some((t) => t.id === id)) openThread(id); |
| } |
| function renderThreadList() { |
| @@ -1856,6 +1897,44 @@ function showConnStatus(text) { |
| el.classList.toggle("hidden", !text); |
| } |
| +// Catch up on messages missed while we were offline. The WebSocket only relays |
| +// *live* messages — nothing is replayed on connect — so anything that arrived |
| +// while the socket was down (phone asleep, tab hidden, a tapped notification |
| +// re-focusing a now-stale tab) never reaches the open feed until you leave and |
| +// re-enter the thread. Called on reconnect and on refocus: refresh the thread |
| +// list (previews / last-activity / any new threads) and append any messages the |
| +// open feed is missing. Incremental (append-only) so it doesn't clear the feed |
| +// or fight the scroll position when there's nothing new. |
| +let resyncing = false; |
| +async function resyncActive() { |
| + if (resyncing || !session) return; |
| + resyncing = true; |
| + try { |
| + await loadThreads(); |
| + renderThreadList(); |
| + const id = activeId; |
| + if (!id || !threadKeys[id]) return; |
| + const r = await ( |
| + await fetch(`/api/messages?token=${encodeURIComponent(session.token)}&thread=${encodeURIComponent(id)}`) |
| + ).json(); |
| + if (id !== activeId) return; // switched threads while fetching — abandon |
| + const feed = document.getElementById("feed"); |
| + let added = 0; |
| + for (const m of r.messages || []) { |
| + if (msgEls[m.id]) continue; // already rendered (live or from the initial load) |
| + await appendMessage(id, m, false); |
| + added++; |
| + } |
| + // Only jump to the newest when we actually appended something, so a plain |
| + // refocus doesn't yank someone out of the history they were reading. |
| + if (added && id === activeId && feed) pinFeedToBottom(feed, openSeq); |
| + } catch (e) { |
| + console.error("resync", e); |
| + } finally { |
| + resyncing = false; |
| + } |
| +} |
| + |
| function connectWS() { |
| const proto = location.protocol === "https:" ? "wss://" : "ws://"; |
| ws = new WebSocket(`${proto}${location.host}/ws?token=${encodeURIComponent(session.token)}`); |
| @@ -1872,6 +1951,11 @@ function connectWS() { |
| pingTimer = setInterval(() => { |
| if (ws && ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: "ping" })); |
| }, 25000); |
| + // A reconnect (not the first connect) means we were offline — pull anything |
| + // the socket missed while it was down. The initial connect follows boot()'s |
| + // fresh load, so there's nothing to catch up on then. |
| + if (everConnected) resyncActive(); |
| + everConnected = true; |
| }; |
| ws.onmessage = async (ev) => { |
| let msg; |
diff --git a/web/static/sw.js b/web/static/sw.js| index 1e228ad..bbc6246 100644 |
| --- a/web/static/sw.js |
| +++ b/web/static/sw.js |
| @@ -106,11 +106,16 @@ async function handlePush(event) { |
| badge: "/static/icon-192.png", |
| tag: "lchat-message", |
| renotify: true, |
| + // Carry the thread id so a tap can open straight to it. Present even on the |
| + // generic fallback (payload had it; we just couldn't decrypt the body); |
| + // absent on the test notification, which has no thread. |
| + data: { thread: data.thread }, |
| }); |
| } |
| self.addEventListener("notificationclick", (event) => { |
| event.notification.close(); |
| + const thread = event.notification.data && event.notification.data.thread; |
| event.waitUntil( |
| (async () => { |
| // Clear any other lchat notifications too, so a single tap dismisses the |
| @@ -118,11 +123,21 @@ self.addEventListener("notificationclick", (event) => { |
| try { |
| for (const n of await self.registration.getNotifications()) n.close(); |
| } catch {} |
| + // Open straight to the notification's thread. An already-open tab gets a |
| + // postMessage (it's a single-page app — no reload); a cold start rides the |
| + // URL hash, which the page reads on boot. Fragments stay client-side, so |
| + // the thread id never hits the server's access log. |
| const wins = await self.clients.matchAll({ type: "window", includeUncontrolled: true }); |
| for (const c of wins) { |
| - if ("focus" in c) return c.focus(); |
| + if ("focus" in c) { |
| + await c.focus(); |
| + if (thread) c.postMessage({ type: "open-thread", thread }); |
| + return; |
| + } |
| + } |
| + if (self.clients.openWindow) { |
| + return self.clients.openWindow(thread ? `/#thread=${encodeURIComponent(thread)}` : "/"); |
| } |
| - if (self.clients.openWindow) return self.clients.openWindow("/"); |
| })() |
| ); |
| }); |