From 6747a47c8a724a04c1641b258c9432e0ca4dba87 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Mon, 3 Aug 2026 17:31:00 +1000 Subject: [PATCH] Notify participant domains and push recipients_added on add-to MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- internal/handlers/hub.go | 16 ++++++++--- internal/handlers/messages.go | 51 +++++++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 3 deletions(-) diff --git a/internal/handlers/hub.go b/internal/handlers/hub.go index f8ef75b..5fa4eb6 100644 --- a/internal/handlers/hub.go +++ b/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) } diff --git a/internal/handlers/messages.go b/internal/handlers/messages.go index ff3984a..4dbcfa9 100644 --- a/internal/handlers/messages.go +++ b/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"})