eleven / messenger Public

chat: cross-device push suppression + tighter gap for grouped messages

- Don't notify a person already reading the chat on another device. The client sends a focus-gated 'active' signal over the WebSocket; the server tracks token→last-active and the message fan-out skips active tokens (within 45s), alongside the existing sender skip. Test pushes ignore it so they always land. (+ isActive test)

- Tighter vertical gap between consecutive same-sender, same-minute bubbles: a continuation gets a .cont class (2px top margin), reusing the same grouping that collapses repeated timestamps (refreshTime → refreshGrouping).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Paul Campbell 4 months ago c44bd06f1b52f25a232ad403faf3307df152cceb parent d46bf03

6 files changed, +103 −35

handlers_test.go +18 −0
hub.go +20 −0
main.go +4 −0
push.go +12 −7
web/static/chat.js +47 −28
web/static/style.css +2 −0

diff --git a/handlers_test.go b/handlers_test.go

index c6ae1bf..34da310 100644

--- a/handlers_test.go

+++ b/handlers_test.go

@@ -8,6 +8,7 @@ import (

"strconv"

"strings"

"testing"

+ "time"

)

func newTestServer(t *testing.T) (*httptest.Server, *Server) {

@@ -142,6 +143,23 @@ func TestMessagesIncludeReactions(t *testing.T) {

}

+func TestActiveTokenTracking(t *testing.T) {

+ _, srv := newTestServer(t)

+ if srv.isActive("tok") {

+ t.Fatal("unknown token should be inactive")

+ }

+ srv.markActive("tok")

+ if !srv.isActive("tok") {

+ t.Fatal("just-marked token should be active")

+ }

+ // A stale timestamp (well past the window) reads as inactive.

+ srv.active.Store("old", time.Now().Add(-2*activeWindow).Unix())

+ if srv.isActive("old") {

+ t.Fatal("stale token should be inactive")

+ }

func TestMessagesPaginationAPI(t *testing.T) {

ts, srv := newTestServer(t)

resp := adminPost(t, ts.URL+"/api/admin/invites", `{"label":"x"}`)

diff --git a/hub.go b/hub.go

index 1383e39..fbffab5 100644

--- a/hub.go

+++ b/hub.go

@@ -81,6 +81,22 @@ type wsOut struct {

// Enforced server-side against the message's created_at (see editMessage).

const editWindow = 15 * time.Minute

+// activeWindow: a token counts as "active" (open + focused on some device) if it

+// signalled within this window. New messages skip push to active tokens, so a

+// chat you're reading on one device doesn't also notify your others.

+const activeWindow = 45 * time.Second

+func (s *Server) markActive(token string) {

+ s.active.Store(token, time.Now().Unix())

+func (s *Server) isActive(token string) bool {

+ if v, ok := s.active.Load(token); ok {

+ return time.Since(time.Unix(v.(int64), 0)) < activeWindow

+ }

+ return false

type wsReactionOut struct {

Type string `json:"type"`

MessageID int64 `json:"message_id"`

@@ -139,6 +155,10 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {

if json.Unmarshal(data, &in) != nil {

continue

}

+ if in.Type == "active" {

+ s.markActive(token) // this device is open + focused; mute its siblings' push

+ continue

+ }

if in.Type != "msg" && in.Type != "edit" && in.Type != "react" && in.Type != "unreact" {

continue

}

diff --git a/main.go b/main.go

index 922f122..8bf5449 100644

--- a/main.go

+++ b/main.go

@@ -31,6 +31,10 @@ type Server struct {

vapidPriv string

// inflight de-duplicates concurrent unfurls of the same URL.

inflight sync.Map

+ // active tracks the last time each invite token reported being open + focused

+ // (token -> unix seconds), so a new message can skip push to a person already

+ // reading the chat on another device.

+ active sync.Map

}

func main() {

diff --git a/push.go b/push.go

index d429075..ea8147d 100644

--- a/push.go

+++ b/push.go

@@ -89,14 +89,15 @@ func (db *DB) ensureVAPID() (pub, priv string, err error) {

// invite token from their WebSocket (their subscriptions carry the same token).

func (s *Server) pushNewMessage(blob, senderToken string) {

payload, _ := json.Marshal(map[string]string{"blob": blob})

- s.sendToAll(payload, "msg", senderToken)

+ // skipActive=true: don't notify people currently reading on another device.

+ s.sendToAll(payload, "msg", senderToken, true)

}

// sendTestPush delivers a generic "it works" notification to every subscriber,

-// bypassing the message flow. Returns the number of successful sends.

+// bypassing the message flow (and ignoring active state, so it always lands).

func (s *Server) sendTestPush() int {

payload, _ := json.Marshal(map[string]bool{"test": true})

- return s.sendToAll(payload, "test", "")

+ return s.sendToAll(payload, "test", "", false)

}

// maxPushConcurrency bounds simultaneous sends so a large group doesn't open

@@ -104,10 +105,11 @@ func (s *Server) sendTestPush() int {

// so fanning out turns O(N) sequential latency into ~O(N/concurrency).

const maxPushConcurrency = 32

-// sendToAll delivers a payload to every subscription (skipping any whose token

-// equals exceptToken), pruning dead ones (404/410). Sends run concurrently,

-// bounded by maxPushConcurrency. Returns the number of successful sends.

-func (s *Server) sendToAll(payload []byte, kind, exceptToken string) int {

+// sendToAll delivers a payload to every subscription, skipping any whose token

+// equals exceptToken and (when skipActive) any token currently active on some

+// device. Prunes dead subs (404/410). Sends run concurrently, bounded by

+// maxPushConcurrency. Returns the number of successful sends.

+func (s *Server) sendToAll(payload []byte, kind, exceptToken string, skipActive bool) int {

subs, err := s.db.listSubscriptions()

if err != nil {

log.Println("push: list subs:", err)

@@ -122,6 +124,9 @@ func (s *Server) sendToAll(payload []byte, kind, exceptToken string) int {

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

+ }

wg.Add(1)

sem <- struct{}{}

go func(sub PushSub) {

diff --git a/web/static/chat.js b/web/static/chat.js

index a3f1750..abeb0a6 100644

--- a/web/static/chat.js

+++ b/web/static/chat.js

@@ -386,13 +386,21 @@ async function start(session) {

const isCoarse =

window.matchMedia && window.matchMedia("(pointer: coarse)").matches;

- // Heartbeat: stamp "active" while the app is visible so the service worker can

- // suppress push notifications for messages you're already watching live.

- // Stamps 0 when hidden so notifications resume the moment you leave.

- const markActive = () =>

- idbSet("lastActiveAt", document.visibilityState === "visible" ? Date.now() : 0).catch(

- () => {}

- );

+ // Declared up here so the heartbeat can reach it once connect() assigns it.

+ let ws = null;

+ // Heartbeat: stamp "active" while the app is visible. The service worker reads

+ // this (IndexedDB) to suppress notifications on THIS device; the WS "active"

+ // signal lets the server suppress them on your OTHER devices too (so a chat

+ // you're reading on desktop doesn't also buzz your phone). Stamp 0 when hidden

+ // so notifications resume the moment you leave.

+ const markActive = () => {

+ const visible = document.visibilityState === "visible";

+ idbSet("lastActiveAt", visible ? Date.now() : 0).catch(() => {});

+ if (visible && ws && ws.readyState === WebSocket.OPEN) {

+ ws.send(JSON.stringify({ type: "active" }));

+ }

+ };

markActive();

setInterval(markActive, 3000);

document.addEventListener("visibilitychange", markActive);

@@ -510,21 +518,32 @@ async function start(session) {

t.textContent = (edited ? "Edited · " : "") + formatTime(payload.ts || createdAt * 1000);

}

- // Collapse repeated times: a message shows its time unless the NEXT message is

- // from the same sender within the same minute (the run's time shows on its last

- // message). Edited messages always show their line so "Edited ·" stays visible.

- // Reads el.dataset.{sender,minute,edited}, set in render().

- function refreshTime(el) {

+ // Two messages belong to the same run if same sender within the same minute.

+ function sameGroup(a, b) {

+ return (

+ !!a &&

+ !!b &&

+ a.classList &&

+ a.classList.contains("msg") &&

+ b.classList &&

+ b.classList.contains("msg") &&

+ a.dataset.sender === b.dataset.sender &&

+ a.dataset.minute === b.dataset.minute

+ );

+ }

+ // Group adjacent messages: a continuation (same run as the message above) gets

+ // a tighter top gap (.cont), and a message hides its time unless it's the last

+ // of its run (the run's time shows once). Edited messages always show their

+ // line so "Edited ·" stays visible. Reads el.dataset.{sender,minute,edited}.

+ function refreshGrouping(el) {

if (!el || !el.classList || !el.classList.contains("msg")) return;

+ el.classList.toggle("cont", sameGroup(el.previousElementSibling, el));

const t = el.querySelector(":scope > .time");

- if (!t) return;

- const next = el.nextElementSibling;

- const groupedWithNext =

- !!next &&

- next.classList.contains("msg") &&

- next.dataset.sender === el.dataset.sender &&

- next.dataset.minute === el.dataset.minute;

- t.style.display = groupedWithNext && !el.dataset.edited ? "none" : "";

+ if (t) {

+ t.style.display =

+ sameGroup(el, el.nextElementSibling) && !el.dataset.edited ? "none" : "";

+ }

}

// Desktop-only: a reply icon that reveals beside the bubble on hover, so reply

@@ -585,7 +604,7 @@ async function start(session) {

el.className = "msg " + (mine ? "me" : "them");

el.dataset.id = msg.id;

if (payload.replyTo && payload.replyTo.id) el.dataset.replyTo = payload.replyTo.id;

- // Grouping keys for collapsing repeated times (see refreshTime).

+ // Grouping keys for tighter gaps + collapsing repeated times (see refreshGrouping).

el.dataset.sender = mine ? "me" : "them:" + payload.name;

el.dataset.minute = Math.floor((payload.ts || msg.created_at * 1000) / 60000);

if (msg.edited_at) el.dataset.edited = "1";

@@ -652,10 +671,11 @@ async function start(session) {

if (opts.before) feed.insertBefore(el, opts.before);

else feed.appendChild(el);

renderReactionsFor(msg.id);

- // Collapse repeated times: this message (vs its next), and its previous

- // neighbour (now that this message follows it).

- refreshTime(el);

- refreshTime(el.previousElementSibling);

+ // Re-group around the insertion point: this message and both neighbours

+ // (prev's time may now hide; next's continuation may now apply).

+ refreshGrouping(el);

+ refreshGrouping(el.previousElementSibling);

+ refreshGrouping(el.nextElementSibling);

if (!opts.noScroll && (atBottom || mine)) feed.scrollTop = feed.scrollHeight;

}

@@ -677,7 +697,7 @@ async function start(session) {

if (content) fillContent(content, payload);

stampTime(el, payload, msgCreatedAt.get(msg.id) || 0, msg.edited_at || 1);

el.dataset.edited = "1"; // edited messages always show their time line

- refreshTime(el);

+ refreshGrouping(el);

}

// Repaint the reaction chips for one message from current state.

@@ -1057,8 +1077,7 @@ async function start(session) {

if (feed.scrollTop < 200) loadOlder();

});

- // 2) Live updates over WebSocket, with reconnect + gap fill.

- let ws;

+ // 2) Live updates over WebSocket, with reconnect + gap fill. (ws declared above.)

let pingTimer = null;

// Only surface "Reconnecting…" once an outage has actually persisted — quick

// blips reconnect within ~1.5s and shouldn't flash anything at all.

diff --git a/web/static/style.css b/web/static/style.css

index c597318..ee1fa50 100644

--- a/web/static/style.css

+++ b/web/static/style.css

@@ -185,6 +185,8 @@ input:focus, textarea:focus { outline: 2px solid var(--accent); border-color: tr

margin-top: auto;

}

.msg { display: flex; flex-direction: column; max-width: 78%; margin-top: 10px; }

+/* continuation of a same-sender, same-minute run — tighter gap (see refreshGrouping) */

+.msg.cont { margin-top: 2px; }

.msg .name { font-size: 12px; color: var(--muted); margin: 0 6px 3px; }

.msg .bubble {

padding: 9px 13px;