Commilitia-Drop/internal/httpapi/message.go

package httpapi

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"log/slog"
	"net/http"
	"sort"
	"time"

	"commilitia.net/cdrop/internal/db"
	"commilitia.net/cdrop/internal/hub"
	"commilitia.net/cdrop/internal/jwtauth"
	"commilitia.net/cdrop/internal/push"
)

// pendingMessageTTL bounds how long an offline message waits in the queue before the
// reaper drops it. A day matches the push TTL — past that the message is stale anyway.
const pendingMessageTTL = 24 * 60 * 60

// 4 KB caps DoS-via-paste; longer payloads should use file transfer instead.
const maxMessageBytes = 4 * 1024

// maxMessageBodyBytes bounds the raw request body BEFORE decode (R5): the 4 KB
// limit above checks the decoded Go string, so on its own the whole body is
// still read into memory first. The wire form can be larger than the decoded
// text — JSON escaping inflates quotes/control chars (worst case ~6×) — so the
// body cap is the content cap with headroom, not equal to it, to avoid falsely
// rejecting a legitimate 4 KB message while still bounding memory per request.
const maxMessageBodyBytes = maxMessageBytes * 8

type messageReq struct {
	To   string `json:"to"`
	Text string `json:"text"`
}

// handleMessage forwards a one-shot text payload to a peer device. Reuses the
// SSE Hub the same way /api/hub/signal does — no DB row, no transfer state
// machine. Receiver sees a `message` event; the frontend keeps it in memory
// only until the tab closes (per user spec).
//
// 204 if delivered live, 202 if the peer is offline but a Web Push was sent
// (the notification carries the text — the message itself is not persisted),
// 410 if offline with no push subscription, 413 if oversized.
func (s *Server) handleMessage(w http.ResponseWriter, r *http.Request) {
	claims, _ := jwtauth.ClaimsFromContext(r.Context())
	from, _ := jwtauth.DeviceNameFromContext(r.Context())

	r.Body = http.MaxBytesReader(w, r.Body, maxMessageBodyBytes)
	var req messageReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		var mbe *http.MaxBytesError
		if errors.As(err, &mbe) {
			writeJSON(w, http.StatusRequestEntityTooLarge,
				map[string]string{"error": "message exceeds 4 KB"})
			return
		}
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid json"})
		return
	}
	if req.To == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "missing 'to'"})
		return
	}
	if req.Text == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "empty text"})
		return
	}
	if len(req.Text) > maxMessageBytes {
		writeJSON(w, http.StatusRequestEntityTooLarge,
			map[string]string{"error": "message exceeds 4 KB"})
		return
	}
	if req.To == from {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "cannot send to self"})
		return
	}

	delivered := s.hub.SendTo(claims.UserID, req.To, hub.Event{
		Type: "message",
		Data: map[string]any{
			"from":    from,
			"text":    req.Text,
			"sent_at": time.Now().Unix(),
		},
	})
	if !delivered {
		// 收件设备页面 / app 关闭（无活 SSE）：把消息入离线队列，待其唤醒 / 回前台经
		// GET /api/messages/pending 取走入库并累积未读——无论是否配推送都入队，故消息不再「只随
		// 推送横幅一闪而过、不入收件列表」。同时发推送唤醒（横幅即时可见）。返回 202（已受理、离线投递）。
		sentAt := time.Now().Unix()
		if err := s.queries.InsertPendingMessage(r.Context(), db.InsertPendingMessageParams{
			ID:         newMessageID(),
			UserID:     claims.UserID,
			ToDevice:   req.To,
			FromDevice: from,
			Text:       req.Text,
			SentAt:     sentAt,
			CreatedAt:  sentAt,
			ExpiresAt:  sentAt + pendingMessageTTL,
		}); err != nil {
			slog.Error("queue offline message failed", "err", err, "user", claims.UserID, "to", req.To)
		}
		n := push.Notification{
			Type:   push.KindMessage,
			Params: map[string]string{"sender": from, "text": req.Text},
		}
		if s.push.Enabled() {
			go s.push.Notify(context.Background(), claims.UserID, req.To, n)
		}
		if s.apns.Enabled() {
			go s.apns.Notify(context.Background(), claims.UserID, req.To, n)
		}
		w.WriteHeader(http.StatusAccepted)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

// newMessageID returns a random opaque id for a queued message so the client can dedup it
// against the live SSE path (which assigns its own ids).
func newMessageID() string {
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		// crypto/rand only fails if the OS RNG is broken; panicking is correct.
		panic("crypto/rand: " + err.Error())
	}
	return hex.EncodeToString(b[:])
}

type pendingMessageWire struct {
	ID     string `json:"id"`
	From   string `json:"from"`
	Text   string `json:"text"`
	SentAt int64  `json:"sent_at"`
}

// handlePendingMessages delivers (once) the messages queued for this device while it was
// offline, then deletes them (DELETE...RETURNING — at-most-once). The device polls this on wake
// / foreground, saves them locally, and accrues unread. Returns [] when none. Guests included
// (messaging is allowed for guest sessions). A device with no managed name gets [].
func (s *Server) handlePendingMessages(w http.ResponseWriter, r *http.Request) {
	claims, _ := jwtauth.ClaimsFromContext(r.Context())
	device, _ := jwtauth.DeviceNameFromContext(r.Context())
	if device == "" {
		writeJSON(w, http.StatusOK, []pendingMessageWire{})
		return
	}
	rows, err := s.queries.PopPendingMessages(r.Context(), db.PopPendingMessagesParams{
		UserID:   claims.UserID,
		ToDevice: device,
	})
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "db"})
		return
	}
	// DELETE...RETURNING order is unspecified; present oldest-first for natural chat order.
	sort.Slice(rows, func(i, j int) bool { return rows[i].SentAt < rows[j].SentAt })
	out := make([]pendingMessageWire, 0, len(rows))
	for _, m := range rows {
		out = append(out, pendingMessageWire{ID: m.ID, From: m.FromDevice, Text: m.Text, SentAt: m.SentAt})
	}
	writeJSON(w, http.StatusOK, out)
}