amadan
server: suppress push per-user, not per-device (cross-device active) (#57)
When you are reading a chat on any device, a new message should not push to your OTHER devices either. The active check in selectRecipients was per-device (isActive(sub.Token)) even though the sender exclusion had already moved to per-user — so your phone still buzzed while you read on your laptop (contradicting the code comment). Roll the "active" signal up to the owner (activeUsers) in markActive, and skip a recipient device when its owner is active on ANY device. This also means the upcoming native (APNs) push inherits the correct behaviour, since it flows through the same selectRecipients. Extends TestSelectRecipientsExcludesAllSenderDevices to assert both of a users devices are skipped when one is active.
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
4 files changed,
+56
−28
db_test.go+21 −13hub.go+20 −4main.go+6 −4push.go+9 −7
diff --git a/db_test.go b/db_test.go| index 6de222c..ffd6c55 100644 |
| --- a/db_test.go |
| +++ b/db_test.go |
| @@ -69,11 +69,11 @@ func TestMessages(t *testing.T) { |
| } |
| db.insertMessage("t1", "blob2", "author-a") |
| - all, err := db.messagesAfter("t1",0, 100) |
| + all, err := db.messagesAfter("t1", 0, 100) |
| if err != nil || len(all) != 2 { |
| t.Fatalf("messagesAfter(0): %v len=%d", err, len(all)) |
| } |
| - after, _ := db.messagesAfter("t1",m1.ID, 100) |
| + after, _ := db.messagesAfter("t1", m1.ID, 100) |
| if len(after) != 1 || after[0].Blob != "blob2" { |
| t.Fatalf("messagesAfter(id): %+v", after) |
| } |
| @@ -110,19 +110,19 @@ func TestMessagesPagination(t *testing.T) { |
| } |
| // after==0 returns the most recent `limit`, ascending. |
| - last2, _ := db.messagesAfter("t1",0, 2) |
| + last2, _ := db.messagesAfter("t1", 0, 2) |
| if len(last2) != 2 || last2[0].ID != ids[3] || last2[1].ID != ids[4] { |
| t.Fatalf("messagesAfter(0,2): %+v", last2) |
| } |
| // before returns the `limit` messages just older than the cursor, ascending. |
| - before, _ := db.messagesBefore("t1",ids[3], 2) |
| + before, _ := db.messagesBefore("t1", ids[3], 2) |
| if len(before) != 2 || before[0].ID != ids[1] || before[1].ID != ids[2] { |
| t.Fatalf("messagesBefore(ids[3],2): %+v", before) |
| } |
| // Fewer than `limit` near the start signals no more older messages. |
| - head, _ := db.messagesBefore("t1",ids[1], 5) |
| + head, _ := db.messagesBefore("t1", ids[1], 5) |
| if len(head) != 1 || head[0].ID != ids[0] { |
| t.Fatalf("messagesBefore(ids[1],5): %+v", head) |
| } |
| @@ -145,7 +145,7 @@ func TestEditMessage(t *testing.T) { |
| if err != nil || !ok || edited.Blob != "fixed" || edited.EditedAt == 0 { |
| t.Fatalf("edit by author should succeed: ok=%v err=%v m=%+v", ok, err, edited) |
| } |
| - got, _ := db.messagesAfter("t1",0, 100) |
| + got, _ := db.messagesAfter("t1", 0, 100) |
| if len(got) != 1 || got[0].Blob != "fixed" || got[0].EditedAt == 0 { |
| t.Fatalf("edit not persisted: %+v", got) |
| } |
| @@ -210,22 +210,30 @@ func TestSelectRecipientsExcludesAllSenderDevices(t *testing.T) { |
| aliceLaptop := mkDevice("a1", "alice") |
| alicePhone := mkDevice("a2", "alice") |
| bobPhone := mkDevice("b1", "bob") |
| + bobLaptop := mkDevice("b2", "bob") |
| - for i, tok := range []string{aliceLaptop, alicePhone, bobPhone} { |
| + for i, tok := range []string{aliceLaptop, alicePhone, bobPhone, bobLaptop} { |
| 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. |
| + // Alice sends from her laptop → both of Bob's devices are notified; neither of |
| + // Alice's own devices is (per-user sender exclusion). |
| 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) |
| + if len(got) != 2 { |
| + t.Fatalf("expected Bob's two devices, got %d: %+v", len(got), got) |
| + } |
| + for _, g := range got { |
| + if g.Token == aliceLaptop || g.Token == alicePhone { |
| + t.Fatalf("a sender device was included: %+v", got) |
| + } |
| } |
| - // And an active recipient is skipped too (they're already reading). |
| - s.markActive(bobPhone) |
| + // Bob starts reading on his laptop → NONE of Bob's devices get pushed, not just |
| + // the laptop (per-user active suppression — the reported requirement). |
| + s.markActive(bobLaptop) |
| if got := s.selectRecipients(subs, aliceLaptop, true, nil); len(got) != 0 { |
| - t.Fatalf("active recipient should be skipped, got %d: %+v", len(got), got) |
| + t.Fatalf("all of an active user's devices should be skipped, got %d: %+v", len(got), got) |
| } |
| } |
diff --git a/hub.go b/hub.go| index 72a37f2..e5182cc 100644 |
| --- a/hub.go |
| +++ b/hub.go |
| @@ -111,13 +111,18 @@ 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. |
| +// activeWindow: a device counts as "active" (open + focused) if it signalled within |
| +// this window. New messages skip push to a person active on ANY of their devices, 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()) |
| + now := time.Now().Unix() |
| + s.active.Store(token, now) |
| + // Roll the signal up to the owner so suppression is per-person, not per-device. |
| + if uid, ok := s.db.deviceTokenUser(token); ok && uid != "" { |
| + s.activeUsers.Store(uid, now) |
| + } |
| } |
| func (s *Server) isActive(token string) bool { |
| @@ -127,6 +132,17 @@ func (s *Server) isActive(token string) bool { |
| return false |
| } |
| +// isUserActive reports whether the user is reading on any of their devices right now. |
| +func (s *Server) isUserActive(uid string) bool { |
| + if uid == "" { |
| + return false |
| + } |
| + if v, ok := s.activeUsers.Load(uid); ok { |
| + return time.Since(time.Unix(v.(int64), 0)) < activeWindow |
| + } |
| + return false |
| +} |
| + |
| type wsReactionOut struct { |
| Type string `json:"type"` |
| ThreadID string `json:"thread_id,omitempty"` |
diff --git a/main.go b/main.go| index d3e3bed..1818fe7 100644 |
| --- a/main.go |
| +++ b/main.go |
| @@ -47,10 +47,12 @@ 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 |
| + // active tracks the last time each device token reported being open + focused |
| + // (token -> unix seconds); activeUsers rolls that up per owner (user id -> unix |
| + // seconds) so a new message skips push to ALL of a person's devices when they're |
| + // reading on any one of them — not just the exact device they're on. |
| + active sync.Map |
| + activeUsers sync.Map |
| } |
| func main() { |
diff --git a/push.go b/push.go| index 31ca476..2bfa357 100644 |
| --- a/push.go |
| +++ b/push.go |
| @@ -125,9 +125,10 @@ const maxPushConcurrency = 32 |
| // 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). |
| +// When skipActive it also drops ALL of a recipient's devices if they're reading on |
| +// any one of them (per-user, mirroring the sender exclusion — a per-device check |
| +// would still buzz your phone while you read on your laptop), 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 != "" { |
| @@ -138,17 +139,18 @@ func (s *Server) selectRecipients(subs []PushSub, exceptToken string, skipActive |
| if exceptToken != "" && sub.Token == exceptToken { |
| continue // the exact device that sent it |
| } |
| - // Resolve the owner once — needed for sender-user exclusion and membership. |
| + // Resolve the owner once — needed for sender-user exclusion, the per-user |
| + // active check, and membership. |
| var uid string |
| var known bool |
| - if senderUser != "" || recipients != nil { |
| + if senderUser != "" || recipients != nil || skipActive { |
| 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 skipActive && s.isUserActive(uid) { |
| + continue // this person is reading on one of their devices right now |
| } |
| if recipients != nil && (!known || !recipients[uid]) { |
| continue // not a member of this thread |