send: refuse replies no remote recipient host can accept (#36)

Per SPEC §10.3 a host rejects a reply whose parent it has not stored
(code 6). Found live: chain replies to a domain whose copy of the parent
never arrived bounce there with no feedback to the sender. On send of a
draft with pid, verify each remote recipient domain against the parent's
recorded delivery: refuse with 409 (naming the domains and the add-to /
new-thread remedy) when the parent was never addressed to the domain or
every delivery attempt there concluded in rejection. Parents still in
flight pass — sequencing in-flight chains is the host's outbound concern
(fmsgd, tracked separately) — as does the parent's originating domain,
which retains its own outgoing messages.

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

Author: markmnl

README.md

@@ -609,6 +609,16 @@ Deletes a draft message and all its attachments from the database and disk. Only
 
 Marks a draft message as sent by setting `time_sent` to the current timestamp. Only the owner may send.
 
+For a reply (a draft with `pid`), the route first verifies that every remote
+recipient domain can actually accept it: per the fmsg spec a host rejects a
+reply whose parent it has not stored (response code 6), so if the parent was
+never addressed to a recipient's domain — or every delivery attempt of the
+parent to that domain concluded in rejection — the send is refused with `409`
+naming the domain(s) and the remedy (add the recipients to the parent via
+add-to, or start a new thread). Domains where the parent's delivery is still
+in flight are allowed; the reply's parent's own originating domain always
+passes (it retains its outgoing messages). Local recipients are unaffected.
+
 **Response:** `200 OK` with `{"id": <int>, "time": <float64>}`.
 
 **Errors:**
@@ -618,6 +628,7 @@ Marks a draft message as sent by setting `time_sent` to the current timestamp. O
 | `403`  | Not the owner |
 | `404`  | Message not found |
 | `409`  | Message already sent |
+| `409`  | Reply can never be accepted by one or more remote recipient domains (parent never addressed there, or its delivery there failed) |
 
 ### POST `/fmsg/:id/read`
 

internal/handlers/messages.go

@@ -137,6 +137,109 @@ func (h *MessageHandler) resolveLocalDelivery(ctx context.Context, table string,
 	}
 }
 
+// parentDomainDelivery summarizes the parent message's recorded delivery for
+// one recipient domain.
+type parentDomainDelivery struct {
+	delivered bool  // at least one recipient there was delivered (parent is stored)
+	pending   bool  // at least one recipient there has no outcome recorded yet
+	codes     []int // failure response codes recorded for the domain
+}
+
+// undeliverableReplyDomains returns, for each remote recipient domain of a
+// reply, the reason the reply can never be accepted there: the parent message
+// was never addressed to the domain, or every delivery attempt there
+// concluded in rejection. Per the fmsg spec a host rejects a reply whose
+// parent it has not stored (code 6), so sending such a reply is a client
+// error worth immediate feedback — the remedy is add-to on the parent or a
+// resend, not a retry. Domains where the parent is merely still in flight are
+// allowed: sequencing in-flight chains is the host's outbound concern, not
+// the client's.
+func undeliverableReplyDomains(replyDomains []string, parentFromDomain string, byDomain map[string]parentDomainDelivery) []string {
+	var blocked []string
+	for _, d := range replyDomains {
+		if strings.EqualFold(d, parentFromDomain) {
+			continue // the originating host retains its own outgoing messages
+		}
+		s, ok := byDomain[strings.ToLower(d)]
+		switch {
+		case !ok:
+			blocked = append(blocked, fmt.Sprintf("%s: the message being replied to was never addressed to this domain", d))
+		case s.delivered || s.pending:
+			// stored there, or still in flight
+		default:
+			blocked = append(blocked, fmt.Sprintf("%s: delivery of the message being replied to failed there (response code(s) %v)", d, s.codes))
+		}
+	}
+	return blocked
+}
+
+// remoteRecipientDomains returns the reply's recipient domains excluding the
+// local domain, deduplicated case-insensitively.
+func remoteRecipientDomains(msg *models.Message, localDomain string) []string {
+	seen := map[string]bool{}
+	var out []string
+	add := func(addr string) {
+		_, domain := parseAddr(addr)
+		key := strings.ToLower(domain)
+		if domain == "" || strings.EqualFold(domain, localDomain) || seen[key] {
+			return
+		}
+		seen[key] = true
+		out = append(out, domain)
+	}
+	for _, a := range msg.To {
+		add(a)
+	}
+	for _, b := range msg.AddTo {
+		for _, a := range b.To {
+			add(a)
+		}
+	}
+	return out
+}
+
+// parentDeliveryByDomain loads the parent's sender domain and a per-domain
+// summary of its recorded recipient delivery outcomes.
+func (h *MessageHandler) parentDeliveryByDomain(ctx context.Context, parentID int64) (string, map[string]parentDomainDelivery, error) {
+	var fromAddr string
+	if err := h.DB.Pool.QueryRow(ctx, "SELECT from_addr FROM msg WHERE id = $1", parentID).Scan(&fromAddr); err != nil {
+		return "", nil, err
+	}
+	rows, err := h.DB.Pool.Query(ctx, `
+		SELECT addr, time_delivered IS NOT NULL, response_code FROM (
+			SELECT addr, time_delivered, response_code FROM msg_to WHERE msg_id = $1
+			UNION ALL
+			SELECT addr, time_delivered, response_code FROM msg_add_to WHERE msg_id = $1
+		) r`, parentID)
+	if err != nil {
+		return "", nil, err
+	}
+	defer rows.Close()
+	byDomain := map[string]parentDomainDelivery{}
+	for rows.Next() {
+		var addr string
+		var delivered bool
+		var code *int
+		if err := rows.Scan(&addr, &delivered, &code); err != nil {
+			return "", nil, err
+		}
+		_, domain := parseAddr(addr)
+		key := strings.ToLower(domain)
+		s := byDomain[key]
+		switch {
+		case delivered:
+			s.delivered = true
+		case code != nil:
+			s.codes = append(s.codes, *code)
+		default:
+			s.pending = true
+		}
+		byDomain[key] = s
+	}
+	_, fromDomain := parseAddr(fromAddr)
+	return fromDomain, byDomain, rows.Err()
+}
+
 // messageInput is used for JSON binding on Create/Update — includes Data for the message body.
 // The outer AddTo field shadows models.Message.AddTo (same JSON name, shallower
 // depth wins), capturing any add_to in the body into an ignored value of any
@@ -763,6 +866,28 @@ func (h *MessageHandler) Send(c *gin.Context) {
 		return
 	}
 
+	// A reply can only be accepted by hosts that store its parent (SPEC
+	// §10.3, reject code 6). Refuse now — with the reason — when a remote
+	// recipient domain can never accept it, rather than letting the reply
+	// bounce there later. Domains where the parent is still in flight pass:
+	// sequencing those is the host's outbound concern.
+	if existing.PID != nil {
+		if replyDomains := remoteRecipientDomains(existing, h.LocalDomain); len(replyDomains) > 0 {
+			parentFromDomain, byDomain, derr := h.parentDeliveryByDomain(ctx, *existing.PID)
+			if derr != nil {
+				log.Printf("send message %d: verify parent %d delivery: %v", msgID, *existing.PID, derr)
+				c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to verify parent delivery"})
+				return
+			}
+			if blocked := undeliverableReplyDomains(replyDomains, parentFromDomain, byDomain); len(blocked) > 0 {
+				c.JSON(http.StatusConflict, gin.H{"error": "reply cannot be accepted by recipient host(s): " +
+					strings.Join(blocked, "; ") +
+					" — add the recipients to the parent message (add-to) or start a new thread with them"})
+				return
+			}
+		}
+	}
+
 	now := float64(time.Now().UnixMicro()) / 1e6
 	if _, err = h.DB.Pool.Exec(ctx, "UPDATE msg SET time_sent = $1 WHERE id = $2", now, msgID); err != nil {
 		log.Printf("send message %d: %v", msgID, err)

internal/handlers/messages_test.go

@@ -1,6 +1,7 @@
 package handlers
 
 import (
+	"github.com/markmnl/fmsg-webapi/internal/models"
 	"os"
 	"path/filepath"
 	"strings"
@@ -282,3 +283,50 @@ func TestExtractShortText(t *testing.T) {
 		}
 	})
 }
+
+func TestUndeliverableReplyDomains(t *testing.T) {
+	byDomain := map[string]parentDomainDelivery{
+		"delivered.example": {delivered: true},
+		"pending.example":   {pending: true},
+		"failed.example":    {codes: []int{6}},
+		"mixed.example":     {delivered: true, codes: []int{100}},
+		"partfail.example":  {pending: true, codes: []int{6}},
+	}
+	cases := []struct {
+		name         string
+		replyDomains []string
+		fromDomain   string
+		wantBlocked  int
+	}{
+		{"delivered parent passes", []string{"delivered.example"}, "origin.example", 0},
+		{"in-flight parent passes", []string{"pending.example"}, "origin.example", 0},
+		{"partially failed but still pending passes", []string{"partfail.example"}, "origin.example", 0},
+		{"delivered outweighs a failed sibling", []string{"mixed.example"}, "origin.example", 0},
+		{"originating domain always passes", []string{"origin.example"}, "origin.example", 0},
+		{"originating domain passes case-insensitively", []string{"Origin.Example"}, "origin.example", 0},
+		{"never-addressed domain blocked", []string{"stranger.example"}, "origin.example", 1},
+		{"permanently failed domain blocked", []string{"failed.example"}, "origin.example", 1},
+		{"mixed reply blocks only the bad domains", []string{"delivered.example", "failed.example", "stranger.example"}, "origin.example", 2},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := undeliverableReplyDomains(tc.replyDomains, tc.fromDomain, byDomain)
+			if len(got) != tc.wantBlocked {
+				t.Fatalf("blocked = %v, want %d entries", got, tc.wantBlocked)
+			}
+		})
+	}
+}
+
+func TestRemoteRecipientDomains(t *testing.T) {
+	msg := &models.Message{
+		To: []string{"@a@remote.example", "@b@Remote.Example", "@c@local.example"},
+		AddTo: []models.AddToBatch{
+			{To: []string{"@d@other.example", "@e@local.example"}},
+		},
+	}
+	got := remoteRecipientDomains(msg, "local.example")
+	if len(got) != 2 || !strings.EqualFold(got[0], "remote.example") || !strings.EqualFold(got[1], "other.example") {
+		t.Fatalf("domains = %v", got)
+	}
+}