Notify participant domains and push recipients_added on add-to

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