Notify participant domains and push recipients_added on add-to (#34)

AddRecipients now records one msg_add_to_notify row per remote
participant domain of the message -- the domains of from and every to
address, excluding domains hosting one of the batch's new recipients
(they learn through normal delivery) and the local domain (this
database is its record) -- so fmsgd delivers the add-to message to
every participant domain per SPEC §10.2, not only the new batch's.

Addresses already added to the message are now rejected up front:
msg_add_to is unique per (msg, addr), so re-adding silently no-opped
and could leave a batch with no recipients, which fmsgd would deliver
as an invalid add-to message. Re-adding an original to recipient stays
allowed (SPEC §10.3 NOTE II).

The websocket hub listens on the new recipients_added channel (fired by
fmsgd's dd.sql when a batch is recorded, whether added locally or
received from a remote host) and pushes the refreshed message to every
connected participant, closing the gap where existing participants got
no realtime signal that recipients were added.

Requires fmsgd's updated dd.sql (msg_add_to_notify table and
recipients_added trigger).

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: markmnl

internal/handlers/hub.go

@@ -15,8 +15,9 @@ import (
 // Event type discriminators for the WebSocket envelope. Adding a new event
 // type means adding a constant here and a producer that dispatches it.
 const (
-	eventNewMsg    = "new_msg"
-	eventDelivered = "delivered"
+	eventNewMsg          = "new_msg"
+	eventDelivered       = "delivered"
+	eventRecipientsAdded = "recipients_added"
 )
 
 // wsEnvelope is the JSON shape of every frame pushed over a WebSocket. The
@@ -128,7 +129,10 @@ func (h *Hub) listen(ctx context.Context, onConnected func()) error {
 	if _, err := conn.Exec(ctx, "LISTEN delivered"); err != nil {
 		return err
 	}
-	log.Println("ws hub: listening on new_msg, delivered")
+	if _, err := conn.Exec(ctx, "LISTEN recipients_added"); err != nil {
+		return err
+	}
+	log.Println("ws hub: listening on new_msg, delivered, recipients_added")
 	onConnected()
 
 	for {
@@ -155,6 +159,12 @@ func (h *Hub) listen(ctx context.Context, onConnected func()) error {
 			// 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)
 		}

internal/handlers/messages.go

@@ -917,6 +917,30 @@ func (h *MessageHandler) AddRecipients(c *gin.Context) {
 		return
 	}
 
+	// Reject addresses already added to this message: msg_add_to is unique per
+	// (msg, addr), so re-adding would silently no-op and could leave a batch
+	// with no recipients — which fmsgd would then deliver as an invalid add-to
+	// message. Re-adding an original to recipient stays allowed (SPEC §10.3
+	// NOTE II — it re-sends the message to a recipient who may no longer have
+	// it).
+	loweredAddTo := make([]string, len(input.AddTo))
+	for i, addr := range input.AddTo {
+		loweredAddTo[i] = strings.ToLower(addr)
+	}
+	var alreadyAdded int
+	if err = h.DB.Pool.QueryRow(ctx,
+		"SELECT COUNT(*) FROM msg_add_to WHERE msg_id = $1 AND lower(addr) = ANY($2)",
+		msgID, loweredAddTo,
+	).Scan(&alreadyAdded); err != nil {
+		log.Printf("add recipients: check existing for msg %d: %v", msgID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to add recipients"})
+		return
+	}
+	if alreadyAdded > 0 {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "address(es) already added to this message"})
+		return
+	}
+
 	// Insert the new add_to recipients and record who added them. Both run in a
 	// single transaction so a partial failure leaves the message unchanged.
 	tx, err := h.DB.Pool.Begin(ctx)
@@ -952,6 +976,33 @@ func (h *MessageHandler) AddRecipients(c *gin.Context) {
 		}
 	}
 
+	// SPEC §10.2: an add-to message is delivered to every participant domain
+	// of the message being added to — the domains of from and every to address
+	// — not only the domains hosting the new recipients, so all existing
+	// participants learn recipients were added. Domains hosting one of this
+	// batch's new recipients learn through normal delivery, and the local
+	// domain's record is this database itself, so neither needs a notify row.
+	newDomains := make([]string, 0, len(input.AddTo))
+	for _, addr := range input.AddTo {
+		_, domain := parseAddr(addr)
+		newDomains = append(newDomains, strings.ToLower(domain))
+	}
+	if _, err = tx.Exec(ctx, `
+		INSERT INTO msg_add_to_notify (batch_id, domain)
+		SELECT DISTINCT $1::bigint, lower(split_part(p.addr, '@', 3))
+		FROM (
+			SELECT from_addr AS addr FROM msg WHERE id = $2
+			UNION
+			SELECT addr FROM msg_to WHERE msg_id = $2
+		) p
+		WHERE lower(split_part(p.addr, '@', 3)) <> lower($3)
+		  AND NOT (lower(split_part(p.addr, '@', 3)) = ANY($4))
+	`, batchID, msgID, h.LocalDomain, newDomains); err != nil {
+		log.Printf("add recipients: insert notify rows for msg %d: %v", msgID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to add recipients"})
+		return
+	}
+
 	if err = tx.Commit(ctx); err != nil {
 		log.Printf("add recipients: commit tx for msg %d: %v", msgID, err)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to add recipients"})