FMSG-006 indicators: POST /fmsg/:id/indicate and the indicator event

README.md

@@ -359,6 +359,7 @@ the application.
 | `POST`   | `/fmsg/:id/restore`              | Take a message or thread out of the trash |
 | `POST`   | `/fmsg/:id/add-to`               | Add recipients           |
 | `POST`   | `/fmsg/:id/react`                | Set or clear an emoji reaction (FMSG-005) |
+| `POST`   | `/fmsg/:id/indicate`             | Show or clear an indicator, such as composing a reply (FMSG-006) |
 | `GET`    | `/fmsg/:id/data`                 | Download message data    |
 | `GET`    | `/fmsg/:id/thread`               | Render direct ancestry as plain text |
 | `GET`    | `/fmsg/:id/thread/messages`      | Load direct ancestry as structured JSON |
@@ -566,6 +567,7 @@ event types can be added without breaking clients:
 | `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. |
+| `indicator` | `{"id": <int>, "root_id": <int>, "from": "<address>", "emoji": "<emoji, or empty>", "ttl": <seconds>}`: `from`'s state on message `id` (in the thread starting at `root_id`), valid for `ttl` seconds unless a later one replaces it. An empty `emoji` clears it. | A participant of a message the user is on [shows or clears an indicator](#post-fmsgidindicate) on it, on this host or another. Never stored, and no Web Push. |
 | `trash` | The message object with its `trash` state, or `{"id": <int>, "trash": "purged"}` for a message deleted for good. | The user moved a message into or out of their [trash](#trash), or deleted it for good, on this or any other connection. |
 
 No event about a message the user deleted for good is ever pushed again,
@@ -993,6 +995,43 @@ the caller has no reaction returns `200 OK` with `null` values.
 | `413`, `422`, `429` | The reaction would exceed one of the reactor's send limits, as for send |
 | `503`  | fmsgid unavailable, so the send limits can't be checked |
 
+### POST `/fmsg/:id/indicate`
+
+Shows or clears the authenticated identity's indicator on a message, per
+[FMSG-006 Indicators](https://github.com/markmnl/fmsg/blob/main/standards/fmsg-006-indicators.md):
+a short-lived state such as ✍️ composing a reply or ⏳ working on one. The
+identity must be a participant of the message, and the message must be sent and
+not terminal.
+
+Nothing is stored. The indicator is published on the PostgreSQL `NOTIFY`
+channel `indicator` as JSON `{"id", "from", "emoji", "ttl"}`; every instance's
+WebSocket hub sends it to the message's other connected participants as an
+[`indicator` event](#get-fmsgws), and the host sends it on to participants on
+other hosts. A client should repeat it every couple of seconds while the state
+lasts, and clear it when the state ends (for example when the reply is sent);
+an uncleared state lapses after `ttl` seconds.
+
+**Request body (JSON):**
+
+| Field   | Type      | Required | Description |
+| ------- | --------- | -------- | ----------- |
+| `emoji` | `string`  | no       | A single emoji, as for reactions. `""` or `null` clears the indicator. |
+| `ttl`   | `integer` | no       | Seconds the state lasts unless repeated: 1 to 120 (larger values are clamped), default 6. `0` clears. |
+
+Indicators set more than once a second by one identity on one message are
+accepted but dropped; clears always go through.
+
+**Response:** `202 Accepted` with no body.
+
+**Errors:**
+
+| Status | Condition |
+| ------ | --------- |
+| `400`  | `emoji` is not a single emoji, or `ttl` is negative |
+| `403`  | Authenticated identity is not a participant |
+| `404`  | Message not found |
+| `409`  | Message is a draft or is terminal |
+
 ### GET `/fmsg/:id/data`
 
 Downloads the binary body of a message. The authenticated identity must be a participant — the sender (`from`) or a recipient (listed in `to` or `add_to`).

cmd/fmsg-webapi/main.go

@@ -296,6 +296,7 @@ func registerRoutes(router *gin.Engine, jwtMiddleware gin.HandlerFunc, msgHandle
 		fmsg.POST("/:id/restore", msgHandler.Atomic((*handlers.MessageHandler).Restore))
 		fmsg.POST("/:id/add-to", msgHandler.Atomic((*handlers.MessageHandler).AddRecipients))
 		fmsg.POST("/:id/react", msgHandler.Atomic((*handlers.MessageHandler).React))
+		fmsg.POST("/:id/indicate", msgHandler.Indicate)
 		fmsg.GET("/:id/data", msgHandler.DownloadData)
 		fmsg.GET("/:id/thread", msgHandler.ThreadText)
 		fmsg.GET("/:id/thread/messages", msgHandler.ThreadMessages)

internal/handlers/finalization_integration_test.go

@@ -102,6 +102,7 @@ func newFinalizationAPIWithID(t *testing.T, fmsgID http.Handler) *finalizationAP
 	r.POST("/fmsg/:id/restore", h.Atomic((*MessageHandler).Restore))
 	r.POST("/fmsg/:id/add-to", h.Atomic((*MessageHandler).AddRecipients))
 	r.POST("/fmsg/:id/react", h.Atomic((*MessageHandler).React))
+	r.POST("/fmsg/:id/indicate", h.Indicate)
 	r.GET("/fmsg/:id/thread", h.ThreadText)
 	r.GET("/fmsg/:id/thread/messages", h.ThreadMessages)
 	r.POST("/fmsg/:id/attach", a.Atomic((*AttachmentHandler).Upload))

internal/handlers/hub.go

@@ -21,6 +21,7 @@ const (
 	eventReaction        = "reaction"
 	eventRead            = "read"
 	eventTrash           = "trash"
+	eventIndicator       = "indicator"
 )
 
 // wsEnvelope is the JSON shape of every frame pushed over a WebSocket. The
@@ -51,6 +52,10 @@ type Hub struct {
 	// as new_msg.
 	reactionSubject func(ctx context.Context, msgID int64) (int64, bool)
 
+	// indicatorAudience reports a message's participants and its thread's
+	// first message, for FMSG-006 indicators.
+	indicatorAudience func(ctx context.Context, msgID int64) ([]string, int64, error)
+
 	mu sync.RWMutex
 	// registry maps a lower-cased user address to the set of that user's
 	// currently connected clients (a user may have several connections).
@@ -66,9 +71,10 @@ func (h *Hub) SetPushNotifier(fn func(ctx context.Context, msgID int64, addr str
 // NewHub creates a Hub that builds pushed message payloads via msgs.
 func NewHub(msgs *MessageHandler) *Hub {
 	return &Hub{
-		buildItem:       msgs.messageItemFor,
-		reactionSubject: msgs.reactionSubject,
-		registry:        make(map[string]map[*wsClient]struct{}),
+		buildItem:         msgs.messageItemFor,
+		reactionSubject:   msgs.reactionSubject,
+		indicatorAudience: msgs.indicatorAudience,
+		registry:          make(map[string]map[*wsClient]struct{}),
 	}
 }
 
@@ -148,14 +154,21 @@ func (h *Hub) listen(ctx context.Context, onConnected func()) error {
 	if _, err := conn.Exec(ctx, "LISTEN msg_trash"); err != nil {
 		return err
 	}
-	log.Println("ws hub: listening on new_msg, delivered, recipients_added, msg_read, msg_trash")
+	if _, err := conn.Exec(ctx, "LISTEN "+IndicatorChannel); err != nil {
+		return err
+	}
+	log.Println("ws hub: listening on new_msg, delivered, recipients_added, msg_read, msg_trash, indicator")
 	onConnected()
 
 	for {
 		n, err := conn.WaitForNotification(ctx)
 		if err != nil {
 			return err
 		}
+		if n.Channel == IndicatorChannel {
+			h.routeIndicator(ctx, n.Payload)
+			continue
+		}
 		msgID, addr, ok := parseNotifyPayload(n.Payload)
 		if !ok {
 			log.Printf("ws hub: ignoring malformed notification payload %q", n.Payload)
@@ -312,3 +325,38 @@ func (h *Hub) send(clients []*wsClient, msgID int64, env wsEnvelope) {
 		}
 	}
 }
+
+// routeIndicator delivers an FMSG-006 indicator (a JSON Indicator payload on
+// the indicator channel) to every connected participant of its message except
+// the indicating one. It is sent as is, with the thread's first message id
+// added; nothing is stored and no Web Push is sent.
+func (h *Hub) routeIndicator(ctx context.Context, payload string) {
+	var ind Indicator
+	if err := json.Unmarshal([]byte(payload), &ind); err != nil || ind.ID <= 0 || ind.From == "" {
+		log.Printf("ws hub: ignoring malformed indicator payload %q", payload)
+		return
+	}
+	if h.indicatorAudience == nil {
+		return
+	}
+	addrs, root, err := h.indicatorAudience(ctx, ind.ID)
+	if err != nil {
+		log.Printf("ws hub: indicator on %d: %v", ind.ID, err)
+		return
+	}
+	ind.RootID = root
+	var clients []*wsClient
+	seen := map[string]bool{strings.ToLower(ind.From): true}
+	for _, a := range addrs {
+		k := strings.ToLower(a)
+		if seen[k] {
+			continue
+		}
+		seen[k] = true
+		clients = append(clients, h.clientsOf(a)...)
+	}
+	if len(clients) == 0 {
+		return
+	}
+	h.send(clients, ind.ID, wsEnvelope{Type: eventIndicator, Data: ind})
+}

internal/handlers/indicators.go

@@ -0,0 +1,200 @@
+package handlers
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/jackc/pgx/v5"
+
+	"github.com/markmnl/fmsg-webapi/internal/emoji"
+	"github.com/markmnl/fmsg-webapi/internal/middleware"
+)
+
+// FMSG-006 indicators: a participant's short-lived, single-emoji state on a
+// message, such as composing a reply. Nothing is stored. An indicator is
+// published on the PostgreSQL NOTIFY channel "indicator"; the WebSocket hub
+// of every instance delivers it to the message's connected participants, and
+// the host (fmsgd) sends it on to the participants' other hosts.
+
+const (
+	// IndicatorChannel is the NOTIFY channel indicators travel on, between
+	// this service and the host, in both directions.
+	IndicatorChannel = "indicator"
+	// indicatorTTLMax is the longest ttl accepted (FMSG-006 INDICATOR_TTL_MAX).
+	indicatorTTLMax = 120
+	// indicatorTTLDefault is the ttl when a request leaves it out.
+	indicatorTTLDefault = 6
+	// indicatorMinInterval is the shortest time between indicators one
+	// address sets on one message; more frequent ones are dropped.
+	indicatorMinInterval = time.Second
+)
+
+// Indicator is the NOTIFY payload and, with RootID, the WebSocket event body.
+// ID is the subject message; From the indicating participant; an empty Emoji
+// or a zero TTL clears the state.
+type Indicator struct {
+	ID     int64  `json:"id"`
+	RootID int64  `json:"root_id,omitempty"`
+	From   string `json:"from"`
+	Emoji  string `json:"emoji"`
+	TTL    int    `json:"ttl"`
+}
+
+type indicateInput struct {
+	Emoji *string `json:"emoji"`
+	TTL   *int    `json:"ttl"`
+}
+
+// indicatorThrottle drops indicators set more often than indicatorMinInterval
+// for one address on one message. Clears are never throttled.
+type indicatorThrottle struct {
+	mu   sync.Mutex
+	last map[string]time.Time
+}
+
+var throttle = &indicatorThrottle{last: map[string]time.Time{}}
+
+func (t *indicatorThrottle) allow(key string, now time.Time) bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if prev, ok := t.last[key]; ok && now.Sub(prev) < indicatorMinInterval {
+		return false
+	}
+	if len(t.last) > 10000 {
+		for k, v := range t.last {
+			if now.Sub(v) >= indicatorMinInterval {
+				delete(t.last, k)
+			}
+		}
+	}
+	t.last[key] = now
+	return true
+}
+
+// Indicate handles POST /fmsg/:id/indicate {"emoji", "ttl"}: sets or clears
+// the caller's indicator on a message they are a participant of. It answers
+// 202 Accepted with no body; an indicator is best effort.
+func (h *MessageHandler) Indicate(c *gin.Context) {
+	identity := middleware.GetIdentity(c)
+	msgID, ok := resolveID(c, h.q())
+	if !ok {
+		return
+	}
+	var in indicateInput
+	if err := c.ShouldBindJSON(&in); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	ind := Indicator{ID: msgID, From: identity, TTL: indicatorTTLDefault}
+	if in.Emoji != nil {
+		ind.Emoji = *in.Emoji
+	}
+	if in.TTL != nil {
+		if *in.TTL < 0 {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "ttl must be 0 to clear, or 1 to 120 seconds"})
+			return
+		}
+		ind.TTL = min(*in.TTL, indicatorTTLMax)
+	}
+	if ind.Emoji != "" && !emoji.IsSingle(ind.Emoji) {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "emoji must be a single emoji, or empty to clear"})
+		return
+	}
+	if ind.Emoji == "" || ind.TTL == 0 {
+		ind.Emoji, ind.TTL = "", 0
+	}
+
+	ctx := c.Request.Context()
+	subject, _, err := h.fetchMessage(ctx, msgID)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			c.JSON(http.StatusNotFound, gin.H{"error": "message not found"})
+		} else {
+			log.Printf("indicate on %d: fetch: %v", msgID, err)
+			c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve message"})
+		}
+		return
+	}
+	if draftHidden([]string{identity}, subject.From, subject.Time) || purgedFor(ctx, h.q(), identity, msgID) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "message not found"})
+		return
+	}
+	if !strings.EqualFold(subject.From, identity) && !isRecipient(subject.To, identity) && !addToContains(subject.AddTo, identity) {
+		c.JSON(http.StatusForbidden, gin.H{"error": "only participants may show an indicator on a message"})
+		return
+	}
+	if subject.Time == nil {
+		c.JSON(http.StatusConflict, gin.H{"error": "cannot show an indicator on a draft"})
+		return
+	}
+	if subject.Terminal {
+		c.JSON(http.StatusConflict, gin.H{"error": "message is terminal; it cannot be replied to"})
+		return
+	}
+
+	if ind.Emoji != "" && !throttle.allow(strings.ToLower(identity)+"|"+strconv.FormatInt(msgID, 10), time.Now()) {
+		c.Status(http.StatusAccepted)
+		return
+	}
+	payload, err := json.Marshal(ind)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to publish indicator"})
+		return
+	}
+	if _, err := h.q().Exec(ctx, `SELECT pg_notify($1, $2)`, IndicatorChannel, string(payload)); err != nil {
+		log.Printf("indicate on %d: notify: %v", msgID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to publish indicator"})
+		return
+	}
+	c.Status(http.StatusAccepted)
+}
+
+// indicatorAudience returns every participant of message msgID (its sender,
+// recipients, and each add-to batch's adder and added recipients) and the id
+// of its thread's first message.
+func (h *MessageHandler) indicatorAudience(ctx context.Context, msgID int64) ([]string, int64, error) {
+	rows, err := h.q().Query(ctx, `
+		SELECT from_addr FROM msg WHERE id = $1
+		UNION SELECT addr FROM msg_to WHERE msg_id = $1
+		UNION SELECT add_to_from FROM msg_add_to_batch WHERE msg_id = $1
+		UNION SELECT addr FROM msg_add_to WHERE msg_id = $1`, msgID)
+	if err != nil {
+		return nil, 0, err
+	}
+	var addrs []string
+	for rows.Next() {
+		var a string
+		if err := rows.Scan(&a); err != nil {
+			rows.Close()
+			return nil, 0, err
+		}
+		addrs = append(addrs, a)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return nil, 0, err
+	}
+	var root int64
+	err = h.q().QueryRow(ctx, `
+		WITH RECURSIVE chain AS (
+		    SELECT id, pid, 0 AS depth FROM msg WHERE id = $1
+		    UNION ALL
+		    SELECT m.id, m.pid, c.depth + 1 FROM msg m JOIN chain c ON m.id = c.pid WHERE c.depth < 10000
+		)
+		SELECT id FROM chain ORDER BY depth DESC LIMIT 1`, msgID).Scan(&root)
+	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+		return nil, 0, err
+	}
+	if root == 0 {
+		root = msgID
+	}
+	return addrs, root, nil
+}