message: degrade quoted-parent to client fallback on transient history outage (#400)

Author: hmchangw

docs/client-api.md

@@ -4530,7 +4530,7 @@ The same subject and request body cover three send variants: plain message, thre
 | `attachments` | string[] | no | Optional. Each entry is base64-encoded bytes — the JSON of one [Attachment](#attachment) from the upload endpoint ([§2.3](#23-http--protected-image-uploaddownload)). Max 1 entry, ≤ 8 KiB total. Stored opaquely and returned **decoded** (as `Attachment[]`) in message payloads. |
 | `threadParentMessageId` | string | no | Set when posting a thread reply. Must be a valid 20-char base62 message ID. |
 | `tshow` | boolean | no | The "Also send to channel" option. Only meaningful on a thread reply (`threadParentMessageId` set): the reply is persisted into the parent room's channel timeline as well as the thread (dual-write into `messages_by_room` in addition to `thread_messages_by_thread` + `messages_by_id`), and is surfaced with `tshow: true` on the persisted message. On a non-thread send the flag is **ignored and normalized to `false`** — the request is not rejected. |
-| `quotedParentMessageId` | string | no | Set when posting a quoted message. The gatekeeper fetches the parent and embeds a snapshot in the persisted message; the client does not send the snapshot itself. |
+| `quotedParentMessageId` | string | no | Set when posting a quoted message. The gatekeeper fetches the authoritative parent snapshot from message history and embeds it in the persisted message. If that fetch fails *transiently* (history briefly unavailable), the message is not dropped: the gatekeeper inserts a placeholder snapshot for live delivery (body `"Content temporarily unavailable"`), and `message-worker` re-projects the authoritative snapshot (or drops the quote) from history before the durable write, so the placeholder never persists. A genuinely missing/forbidden parent is still rejected. |
 
 ##### Plain message
 
@@ -4564,6 +4564,8 @@ The same subject and request body cover three send variants: plain message, thre
 }
 ```
 
+The client sends `quotedParentMessageId`; the server fetches and embeds the authoritative quoted-parent snapshot. During a transient history outage the server fills the live copy with a `"Content temporarily unavailable"` placeholder and re-projects the authoritative quote (or drops it) before persisting, so the placeholder never persists.
+
 ##### Thread reply quoting the thread-starter
 
 A thread reply may quote the thread's own parent message (the message that started the thread) by setting both `threadParentMessageId` and `quotedParentMessageId` to the same ID. The quoted snapshot is embedded in the response like any other quote.

message-gatekeeper/debug_log_test.go

@@ -101,7 +101,7 @@ func TestHandler_processMessage_DebugBreadcrumbs(t *testing.T) {
 	store.EXPECT().GetSubscription(gomock.Any(), account, roomID).Return(sub, nil).AnyTimes()
 	store.EXPECT().GetRoomMeta(gomock.Any(), roomID).Return(roommetacache.Meta{ID: roomID, UserCount: 1}, nil).AnyTimes()
 
-	h := NewHandler(store, nil, makePublishFunc(nil, nil), func(context.Context, *nats.Msg) error { return nil }, siteID, nil, 500, 1, 8192)
+	h := NewHandler(store, nil, makePublishFunc(nil, nil), func(context.Context, *nats.Msg) error { return nil }, siteID, nil, 500, 1, 8192, "")
 
 	newReq := func() model.SendMessageRequest {
 		return model.SendMessageRequest{ID: idgen.GenerateMessageID(), Content: "hello world", RequestID: reqID}

message-gatekeeper/fetcher_history.go

@@ -98,7 +98,7 @@ func (f *historyParentFetcher) FetchQuotedParent(
 		Msg:                   parent.Msg,
 		Mentions:              parent.Mentions,
 		DecodedAttachments:    parent.DecodedAttachments,
-		MessageLink:           fmt.Sprintf("%s/%s/%s", f.chatBaseURL, parent.RoomID, messageID),
+		MessageLink:           messageLink(f.chatBaseURL, parent.RoomID, messageID),
 		ThreadParentID:        parent.ThreadParentID,
 		ThreadParentCreatedAt: parent.ThreadParentCreatedAt,
 	}, nil

message-gatekeeper/handler.go

@@ -27,6 +27,11 @@ import (
 
 const maxContentBytes = 20 * 1024 // 20 KB
 
+// quotedParentUnavailablePlaceholder is the degraded-mode quoted-parent body used
+// when the authoritative fetch fails transiently. It never persists —
+// message-worker re-projects the real snapshot before the durable write.
+const quotedParentUnavailablePlaceholder = "Content temporarily unavailable"
+
 // replyFunc is the function signature for publishing a reply to a NATS subject.
 type replyFunc func(ctx context.Context, msg *nats.Msg) error
 
@@ -51,12 +56,16 @@ type Handler struct {
 	largeRoomThreshold int
 	maxAttachments     int
 	maxAttachmentBytes int
+	// chatBaseURL builds the messageLink on the degraded-mode placeholder quoted
+	// snapshot, from trusted inputs (the send room + the validated quoted message
+	// ID) so the link is correct even on the outage path.
+	chatBaseURL string
 }
 
 // NewHandler constructs a new Handler with the given dependencies.
 // users may be nil; when nil, sender display-name resolution is skipped and
 // downstream consumers fall back to UserAccount.
-func NewHandler(store Store, users UserGetter, publish publishFunc, reply replyFunc, siteID string, parentFetcher ParentMessageFetcher, largeRoomThreshold, maxAttachments, maxAttachmentBytes int) *Handler {
+func NewHandler(store Store, users UserGetter, publish publishFunc, reply replyFunc, siteID string, parentFetcher ParentMessageFetcher, largeRoomThreshold, maxAttachments, maxAttachmentBytes int, chatBaseURL string) *Handler {
 	return &Handler{
 		store:              store,
 		users:              users,
@@ -67,6 +76,7 @@ func NewHandler(store Store, users UserGetter, publish publishFunc, reply replyF
 		largeRoomThreshold: largeRoomThreshold,
 		maxAttachments:     maxAttachments,
 		maxAttachmentBytes: maxAttachmentBytes,
+		chatBaseURL:        chatBaseURL,
 	}
 }
 
@@ -230,6 +240,13 @@ func (h *Handler) processMessage(ctx context.Context, account, roomID, siteID st
 		return nil, errcode.BadRequest(fmt.Sprintf("invalid thread parent message ID %q: must be a 20-char base62 string", req.ThreadParentMessageID))
 	}
 
+	// Validate the quoted parent ID at the boundary too: on the degrade path it is
+	// copied verbatim into the snapshot MessageID and messageLink, so a malformed
+	// value must fail fast rather than leak into the canonical event.
+	if req.QuotedParentMessageID != "" && !idgen.IsValidMessageID(req.QuotedParentMessageID) {
+		return nil, errcode.BadRequest(fmt.Sprintf("invalid quoted parent message ID %q: must be a 20-char base62 string", req.QuotedParentMessageID))
+	}
+
 	// A message with attachments may carry empty content.
 	if req.Content == "" && len(req.Attachments) == 0 {
 		return nil, errcode.BadRequest("content must not be empty")
@@ -307,20 +324,20 @@ func (h *Handler) processMessage(ctx context.Context, account, roomID, siteID st
 	// Build Message
 	now := time.Now().UTC()
 
-	quotedSnapshot, err := h.resolveQuoteSnapshot(ctx, account, roomID, siteID, req.QuotedParentMessageID, req.ThreadParentMessageID)
+	quotedSnapshot, quotedUnverified, err := h.resolveQuoteSnapshot(ctx, account, roomID, siteID, req.QuotedParentMessageID, req.ThreadParentMessageID, now)
 	if err != nil {
 		return nil, err
 	}
 	if req.QuotedParentMessageID != "" {
 		// debug: quote passed the same-conversation-context check.
-		slog.DebugContext(ctx, "gatekeeper quote resolved", "request_id", req.RequestID, "quoted_id", req.QuotedParentMessageID)
+		slog.DebugContext(ctx, "gatekeeper quote resolved", "request_id", req.RequestID, "quoted_id", req.QuotedParentMessageID, "unverified", quotedUnverified)
 	}
 
 	// #322: resolve the thread parent's createdAt server-side. The
 	// server-resolved value always wins over any client-sent value — a wrong
 	// client value must not corrupt downstream consumers. Done after the quote
 	// resolution so a quote-context failure short-circuits the extra fetch.
-	threadParentCreatedAt, err := h.resolveThreadParentCreatedAt(ctx, account, roomID, siteID, req.ThreadParentMessageID, req.QuotedParentMessageID, quotedSnapshot)
+	threadParentCreatedAt, err := h.resolveThreadParentCreatedAt(ctx, account, roomID, siteID, req.ThreadParentMessageID, req.QuotedParentMessageID, quotedSnapshot, quotedUnverified)
 	if err != nil {
 		return nil, err
 	}
@@ -363,8 +380,11 @@ func (h *Handler) processMessage(ctx context.Context, account, roomID, siteID st
 		Attachments:                  req.Attachments,
 	}
 
-	// Publish MessageEvent to MESSAGES_CANONICAL
-	evt := model.MessageEvent{Event: model.EventCreated, Message: msg, SiteID: siteID, Timestamp: now.UnixMilli()}
+	// Publish MessageEvent to MESSAGES_CANONICAL. QuotedParentUnverified rides the
+	// envelope (not the persisted Message) so message-worker knows to re-project
+	// the authoritative snapshot before the durable write when the gatekeeper had
+	// to fall back to the untrusted client snapshot.
+	evt := model.MessageEvent{Event: model.EventCreated, Message: msg, SiteID: siteID, Timestamp: now.UnixMilli(), QuotedParentUnverified: quotedUnverified}
 	evtData, err := sonic.Marshal(evt)
 	if err != nil {
 		return nil, fmt.Errorf("marshal message event: %w", err)
@@ -391,13 +411,17 @@ func (h *Handler) resolveThreadParentCreatedAt(
 	ctx context.Context,
 	account, roomID, siteID, threadParentMessageID, quotedParentMessageID string,
 	quotedSnapshot *cassandra.QuotedParentMessage,
+	quotedUnverified bool,
 ) (*time.Time, error) {
 	if threadParentMessageID == "" {
 		return nil, nil
 	}
 
-	// Reuse the quote snapshot when the quoted parent is the thread parent.
-	if quotedSnapshot != nil && quotedParentMessageID == threadParentMessageID {
+	// Reuse the quote snapshot's CreatedAt only when authoritative. The unverified
+	// placeholder carries a synthetic timestamp, and the thread parent's createdAt
+	// (#322) isn't re-resolved downstream, so fetch it authoritatively instead
+	// (NAKs if history is still down).
+	if quotedSnapshot != nil && !quotedUnverified && quotedParentMessageID == threadParentMessageID {
 		t := quotedSnapshot.CreatedAt.UTC()
 		return &t, nil
 	}
@@ -419,40 +443,101 @@ func (h *Handler) resolveThreadParentCreatedAt(
 	return &t, nil
 }
 
-// resolveQuoteSnapshot fetches the quoted parent and returns its snapshot.
-// The strict same-conversation-context rule rejects cross-thread quotes:
-// main-room messages may only quote main-room parents, and thread-T messages
-// may only quote other thread-T messages — including the thread's own parent.
-func (h *Handler) resolveQuoteSnapshot(ctx context.Context, account, roomID, siteID, quotedParentMessageID, newMessageThreadID string) (*cassandra.QuotedParentMessage, error) {
+// resolveQuoteSnapshot resolves the quoted parent into a snapshot, preferring the
+// authoritative history fetch and enforcing the same-conversation rule (a message
+// may only quote parents in its own thread or main room). The bool is the
+// "unverified" marker — true when the snapshot is the degraded placeholder.
+//
+// Fetch failures are tiered:
+//   - terminal (not_found, forbidden, bad_request): reject — a quote must never
+//     resurrect a missing parent or bypass access control.
+//   - transient (history unavailable/internal, NATS timeout): degrade to a
+//     placeholder (marked unverified) with the fixed "Content temporarily
+//     unavailable" body — which message-worker re-projects (or drops) before the
+//     durable write.
+func (h *Handler) resolveQuoteSnapshot(ctx context.Context, account, roomID, siteID, quotedParentMessageID, newMessageThreadID string, now time.Time) (*cassandra.QuotedParentMessage, bool, error) {
 	if quotedParentMessageID == "" {
-		return nil, nil
+		return nil, false, nil
 	}
 	snap, err := h.parentFetcher.FetchQuotedParent(ctx, account, roomID, siteID, quotedParentMessageID)
-	switch {
-	case err != nil:
-		// Preserve upstream errcode classification (transient → Unavailable,
-		// real 404 → NotFound). For non-errcode infra failures (NATS timeout,
-		// no-responders, unmarshal), classify as Unavailable — a transient
-		// quoted-parent fetch failure shouldn't surface to the client as 404.
+	if err == nil && snap == nil {
+		// A nil snapshot with no error is a fetcher contract violation, not a
+		// genuine missing parent. Synthesize a transient error so we degrade to the
+		// placeholder rather than treating it as authoritative-empty.
+		err = fmt.Errorf("fetch quoted parent %s: fetcher returned nil snapshot", quotedParentMessageID)
+	}
+	if err != nil {
 		var ee *errcode.Error
-		if errors.As(err, &ee) {
-			return nil, ee
+		if quoteFetchErrIsTerminal(err) && errors.As(err, &ee) {
+			// Terminal typed *errcode.Error → reply + Ack; preserves the upstream
+			// category (not_found, forbidden, …) for the client. The errors.As guard
+			// is belt-and-suspenders: quoteFetchErrIsTerminal only reports true for a
+			// typed errcode today, but guarding the bool means a future predicate
+			// change can't fall through here returning a nil *errcode.Error.
+			return nil, false, ee
 		}
-		return nil, errcode.Unavailable(fmt.Sprintf("fetch quoted parent %s", quotedParentMessageID), errcode.WithCause(err))
-	case snap == nil:
-		// A nil snapshot with no error is a fetcher contract violation, not a
-		// genuine missing parent. Return a bare error so the caller's branch
-		// classifies this as infra (Nak for redelivery + log) rather than
-		// permanently dropping the message via a 404 reply+Ack.
-		return nil, fmt.Errorf("fetch quoted parent %s: fetcher returned nil snapshot", quotedParentMessageID)
-	case snap.ThreadParentID != newMessageThreadID &&
+		// Transient failure: degrade to a placeholder snapshot rather than NAK, so
+		// the message still flows through the outage. message-worker re-projects the
+		// authoritative snapshot (or drops the quote) before the durable write.
+		ph := h.placeholderQuoteSnapshot(roomID, quotedParentMessageID, newMessageThreadID, now)
+		slog.WarnContext(ctx, "quoted-parent fetch failed; using placeholder snapshot",
+			"request_id", natsutil.RequestIDFromContext(ctx), "quoted_id", quotedParentMessageID, "error", err)
+		return ph, true, nil
+	}
+	if cerr := checkQuoteThreadContext(snap, quotedParentMessageID, newMessageThreadID); cerr != nil {
+		return nil, false, cerr
+	}
+	return snap, false, nil
+}
+
+// quoteFetchErrIsTerminal reports whether a quoted-parent fetch error is a
+// permanent reason not to quote (reject) vs a transient infra failure (degrade
+// to the placeholder). Only unavailable/internal errcodes and non-errcode infra
+// failures (NATS timeout, no-responders, unmarshal) are transient; every other
+// errcode category (not_found, forbidden, bad_request, …) is terminal.
+// history-service collapses a Cassandra read failure to code=internal, so
+// internal is treated as transient here.
+func quoteFetchErrIsTerminal(err error) bool {
+	var ee *errcode.Error
+	if errors.As(err, &ee) {
+		switch ee.Code {
+		case errcode.CodeUnavailable, errcode.CodeInternal:
+			return false
+		default:
+			return true
+		}
+	}
+	return false
+}
+
+// placeholderQuoteSnapshot builds the degraded-mode quoted-parent snapshot for a
+// transient fetch failure. Only identity/link fields are real; the body is always
+// the fixed "Content temporarily unavailable" placeholder (the client cannot
+// supply the quoted-parent content). ThreadParentID mirrors the quoting message's
+// thread to satisfy the same-conversation rule, and CreatedAt is cosmetic. The
+// caller marks it unverified (see resolveQuoteSnapshot); the body is re-projected
+// authoritatively by message-worker before the durable write.
+func (h *Handler) placeholderQuoteSnapshot(roomID, messageID, newMessageThreadID string, now time.Time) *cassandra.QuotedParentMessage {
+	return &cassandra.QuotedParentMessage{
+		MessageID:      messageID,
+		RoomID:         roomID,
+		CreatedAt:      now,
+		Msg:            quotedParentUnavailablePlaceholder,
+		MessageLink:    messageLink(h.chatBaseURL, roomID, messageID),
+		ThreadParentID: newMessageThreadID,
+	}
+}
+
+// checkQuoteThreadContext enforces the same-conversation rule between the quoted
+// parent and the quoting message.
+func checkQuoteThreadContext(snap *cassandra.QuotedParentMessage, quotedParentMessageID, newMessageThreadID string) error {
+	if snap.ThreadParentID != newMessageThreadID &&
 		// Thread-root quote: starter is a main-room msg (ThreadParentID=="") whose ID is the thread parent — allowed.
-		(snap.ThreadParentID != "" || quotedParentMessageID != newMessageThreadID):
-		return nil, errcode.BadRequest(fmt.Sprintf("quoted parent %s thread context mismatch: parent thread %q, new message thread %q",
+		(snap.ThreadParentID != "" || quotedParentMessageID != newMessageThreadID) {
+		return errcode.BadRequest(fmt.Sprintf("quoted parent %s thread context mismatch: parent thread %q, new message thread %q",
 			quotedParentMessageID, snap.ThreadParentID, newMessageThreadID))
-	default:
-		return snap, nil
 	}
+	return nil
 }
 
 // canBypassLargeRoomCap reports whether the subscriber is exempt from the