Push a read event when a message is marked read

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": <int>, "time_read": <float64>}`.
 

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)
 	}
 }
 

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})
 }
 

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)
+	}
+}