From 5a5a98f8147a0a9a85cdb796cf45c62b8a642966 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Sun, 18 Oct 2026 06:09:45 +0800 Subject: [PATCH] Push a read event when a message is marked read POST /fmsg/:id/read now notifies msg_read the first time a message becomes read, and the hub pushes a `read` event, carrying the message with its read state, to all of the reader's WebSocket connections. A client open in another tab or on another device can then stop showing the message as unread instead of waiting for a reload. The hub's channel switch moves into route so routing is unit tested. Co-Authored-By: Claude Opus 5.5 --- README.md | 3 ++ internal/handlers/hub.go | 81 ++++++++++++++++------------ internal/handlers/messages.go | 7 +++ internal/handlers/read_event_test.go | 75 ++++++++++++++++++++++++++ 4 files changed, 133 insertions(+), 33 deletions(-) create mode 100644 internal/handlers/read_event_test.go diff --git a/README.md b/README.md index efaf7cc..8a96314 100644 --- a/README.md +++ b/README.md @@ -552,6 +552,7 @@ event types can be added without breaking clients: | `new_msg` | A message object, same shape as an item in the `GET /fmsg` list response (includes `id`). | A new message arrives for the authenticated user. Not sent for reactions. | | `delivered` | The refreshed message object. | A recipient's delivery state changed on a message the user sent. | | `recipients_added` | The refreshed message object. | An add-to batch was recorded on a message the user participates in. | +| `read` | The message object, with `read` and `time_read` set. | The user marked a message read (`POST /fmsg/:id/read`), on this or any other connection, so every tab and device can stop showing it as unread. Sent once, when the message first becomes read. | | `reaction` | The refreshed **subject** message object, with its `reactions` up to date. | A reaction (FMSG-005) on a message the user participates in arrives. The reaction message itself is not pushed as `new_msg` and triggers no Web Push. | A client only ever receives events for messages it is a participant on. The @@ -848,6 +849,8 @@ a recipient over one is not delivered and its `response_code` is `101`. Marks a message as read by the authenticated recipient by setting `time_read` on the caller's `msg_to` or `msg_add_to` row. Idempotent: re-reading an already-read message returns the original `time_read` without updating it. +The first read pushes a `read` event to all of the caller's WebSocket +connections (see [`GET /fmsg/ws`](#get-fmsgws)). **Response:** `200 OK` with `{"id": , "time_read": }`. diff --git a/internal/handlers/hub.go b/internal/handlers/hub.go index ff5b07f..6d4f52e 100644 --- a/internal/handlers/hub.go +++ b/internal/handlers/hub.go @@ -19,6 +19,7 @@ const ( eventDelivered = "delivered" eventRecipientsAdded = "recipients_added" eventReaction = "reaction" + eventRead = "read" ) // wsEnvelope is the JSON shape of every frame pushed over a WebSocket. The @@ -118,9 +119,9 @@ func (h *Hub) Run(ctx context.Context) { } } -// listen opens a dedicated connection, LISTENs on new_msg and delivered, and -// dispatches every notification until the connection fails or ctx is -// cancelled. onConnected is invoked once both LISTENs have succeeded so the +// listen opens a dedicated connection, LISTENs on every notification channel, +// and routes each notification until the connection fails or ctx is +// cancelled. onConnected is invoked once every LISTEN has succeeded so the // caller can reset its backoff. func (h *Hub) listen(ctx context.Context, onConnected func()) error { // An empty connection string makes pgx read the standard PG* environment @@ -140,7 +141,10 @@ func (h *Hub) listen(ctx context.Context, onConnected func()) error { if _, err := conn.Exec(ctx, "LISTEN recipients_added"); err != nil { return err } - log.Println("ws hub: listening on new_msg, delivered, recipients_added") + if _, err := conn.Exec(ctx, "LISTEN msg_read"); err != nil { + return err + } + log.Println("ws hub: listening on new_msg, delivered, recipients_added, msg_read") onConnected() for { @@ -153,37 +157,48 @@ func (h *Hub) listen(ctx context.Context, onConnected func()) error { log.Printf("ws hub: ignoring malformed notification payload %q", n.Payload) continue } - switch n.Channel { - case "new_msg": - if h.reactionSubject != nil { - if subject, ok := h.reactionSubject(ctx, msgID); ok { - // FMSG-005: a reaction is a low-priority event on its - // subject, not a new message. No Web Push is sent. - h.dispatch(ctx, subject, addr, eventReaction) - continue - } - } - // Web Push runs in its own goroutine: it is network I/O to many - // endpoints and must not stall the single listener. It is - // dispatched here, alongside (not inside) dispatch, so a push is - // sent even when the recipient has no live WebSocket. - if h.notifyPush != nil { - go h.notifyPush(ctx, msgID, addr) + h.route(ctx, n.Channel, msgID, addr) + } +} + +// route handles one notification on channel for message msgID and address +// addr. +func (h *Hub) route(ctx context.Context, channel string, msgID int64, addr string) { + switch channel { + case "new_msg": + if h.reactionSubject != nil { + if subject, ok := h.reactionSubject(ctx, msgID); ok { + // FMSG-005: a reaction is a low-priority event on its + // subject, not a new message. No Web Push is sent. + h.dispatch(ctx, subject, addr, eventReaction) + return } - h.dispatch(ctx, msgID, addr, eventNewMsg) - case "delivered": - // addr here is the message's sender (see notify_delivered in - // dd.sql), not a recipient -- no Web Push for this event yet. - h.dispatch(ctx, msgID, addr, eventDelivered) - case "recipients_added": - // An add-to batch was recorded against the message; addr is one - // of its participants (see notify_recipients_added in fmsgd's - // dd.sql). Pushes the refreshed message so clients can show the - // updated recipient list. - h.dispatch(ctx, msgID, addr, eventRecipientsAdded) - default: - log.Printf("ws hub: ignoring notification on unknown channel %q", n.Channel) } + // Web Push runs in its own goroutine: it is network I/O to many + // endpoints and must not stall the single listener. It is + // dispatched here, alongside (not inside) dispatch, so a push is + // sent even when the recipient has no live WebSocket. + if h.notifyPush != nil { + go h.notifyPush(ctx, msgID, addr) + } + h.dispatch(ctx, msgID, addr, eventNewMsg) + case "delivered": + // addr here is the message's sender (see notify_delivered in + // dd.sql), not a recipient -- no Web Push for this event yet. + h.dispatch(ctx, msgID, addr, eventDelivered) + case "msg_read": + // A recipient marked the message read (see MarkRead); addr is that + // recipient. Pushes the message with its read state so the reader's + // other tabs and devices stop showing it as unread. + h.dispatch(ctx, msgID, addr, eventRead) + case "recipients_added": + // An add-to batch was recorded against the message; addr is one + // of its participants (see notify_recipients_added in fmsgd's + // dd.sql). Pushes the refreshed message so clients can show the + // updated recipient list. + h.dispatch(ctx, msgID, addr, eventRecipientsAdded) + default: + log.Printf("ws hub: ignoring notification on unknown channel %q", channel) } } diff --git a/internal/handlers/messages.go b/internal/handlers/messages.go index 63cdaba..db44f6e 100644 --- a/internal/handlers/messages.go +++ b/internal/handlers/messages.go @@ -1096,6 +1096,13 @@ func (h *MessageHandler) MarkRead(c *gin.Context) { return } + // Tell the reader's other connections, so another tab or device stops + // showing the message as unread. A failure is logged: the message is read. + if _, err = h.q().Exec(ctx, `SELECT pg_notify('msg_read', $1)`, + fmt.Sprintf("%d,%s", msgID, identity)); err != nil { + log.Printf("mark read %d: notify: %v", msgID, err) + } + c.JSON(http.StatusOK, gin.H{"id": msgID, "time_read": now}) } diff --git a/internal/handlers/read_event_test.go b/internal/handlers/read_event_test.go new file mode 100644 index 0000000..ae6ced3 --- /dev/null +++ b/internal/handlers/read_event_test.go @@ -0,0 +1,75 @@ +package handlers + +import ( + "context" + "encoding/json" + "testing" + "time" +) + +// A msg_read notification is pushed to the reader as a read event carrying +// the message. +func TestHubRoute_MsgReadPushesReadEvent(t *testing.T) { + hub := &Hub{ + buildItem: func(_ context.Context, msgID int64, _ string) (*messageListItem, error) { + return &messageListItem{ID: msgID, Read: true}, nil + }, + registry: make(map[string]map[*wsClient]struct{}), + } + c := &wsClient{addr: "@bob@example.com", send: make(chan []byte, 4)} + hub.Register(c) + + hub.route(context.Background(), "msg_read", 9, "@Bob@example.com") + select { + case frame := <-c.send: + var env struct { + Type string `json:"type"` + Data messageListItem `json:"data"` + } + if err := json.Unmarshal(frame, &env); err != nil { + t.Fatal(err) + } + if env.Type != eventRead || env.Data.ID != 9 || !env.Data.Read { + t.Errorf("got %s for %d (read %v), want %s for 9 (read)", env.Type, env.Data.ID, env.Data.Read, eventRead) + } + default: + t.Fatal("no frame pushed") + } +} + +// Marking a message read notifies msg_read with the message and its reader, +// once: reading it again changes nothing and notifies nothing. +func TestMarkReadNotifies(t *testing.T) { + a := newFinalizationAPI(t) + const alice, bob = "@alice@example.com", "@bob@example.com" + ctx := context.Background() + conn, err := a.pool.Acquire(ctx) + if err != nil { + t.Fatal(err) + } + defer conn.Release() + if _, err := conn.Exec(ctx, "LISTEN msg_read"); err != nil { + t.Fatal(err) + } + + id := a.draft(t, alice, "hello", nil) + jsonResponse(t, a.request(alice, "POST", "/fmsg/"+id+"/send", map[string]any{}), 200) + jsonResponse(t, a.request(bob, "POST", "/fmsg/"+id+"/read", map[string]any{}), 200) + + wait, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + n, err := conn.Conn().WaitForNotification(wait) + if err != nil { + t.Fatalf("no msg_read notification: %v", err) + } + if want := id + "," + bob; n.Payload != want { + t.Errorf("payload %q, want %q", n.Payload, want) + } + + jsonResponse(t, a.request(bob, "POST", "/fmsg/"+id+"/read", map[string]any{}), 200) + again, cancel2 := context.WithTimeout(ctx, 300*time.Millisecond) + defer cancel2() + if n, err := conn.Conn().WaitForNotification(again); err == nil { + t.Errorf("second read notified %q", n.Payload) + } +}