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