NixMsg/internal/admin/messages.go

package admin

import (
	"context"
	"database/sql"
	"encoding/json"
	"net/http"
	"strconv"
	"strings"

	"git.asio.asia/nixevol/NixMsg/internal/httpx"
)

// 消息查询只读 messages / deliveries，不读 message_bodies，响应不含 body/正文。

func (h *Handler) handleMessageList(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	limit, offset, ok := parsePage(w, q.Get("limit"), q.Get("cursor"))
	if !ok {
		return
	}

	where, args, ok := buildMessageFilter(w, q)
	if !ok {
		return
	}

	var total int
	if err := h.db.Read.QueryRowContext(r.Context(),
		`SELECT COUNT(*) FROM messages m WHERE `+where, args...).Scan(&total); err != nil {
		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
		return
	}

	listArgs := append(append([]any{}, args...), limit, offset)
	rows, err := h.db.Read.QueryContext(r.Context(), `
SELECT m.seq, m.id, m.sender_id, m.dest_kind, m.dest_id, m.state, m.reason,
  m.send_at, m.created_at, m.keep, m.receipt, m.content_type
FROM messages m
WHERE `+where+`
ORDER BY m.seq DESC
LIMIT ? OFFSET ?`, listArgs...)
	if err != nil {
		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
		return
	}
	defer func() { _ = rows.Close() }()

	items := make([]map[string]any, 0)
	seqs := make([]int64, 0)
	for rows.Next() {
		var seq int64
		var id, sender, destKind, destID, state, reason, contentType string
		var sendAt, created int64
		var keep, receipt int
		if scanErr := rows.Scan(&seq, &id, &sender, &destKind, &destID, &state, &reason,
			&sendAt, &created, &keep, &receipt, &contentType); scanErr != nil {
			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
			return
		}
		seqs = append(seqs, seq)
		items = append(items, map[string]any{
			"seq":           seq,
			"id":            id,
			"sender_id":     sender,
			"dest_kind":     destKind,
			"dest_id":       destID,
			"state":         state,
			"reason":        reason,
			"send_at_ms":    sendAt,
			"created_at_ms": created,
			"keep":          keep != 0,
			"receipt":       receipt != 0,
			"content_type":  contentType,
		})
	}
	if rowsErr := rows.Err(); rowsErr != nil {
		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
		return
	}

	counts, err := h.deliveryCounts(r.Context(), seqs)
	if err != nil {
		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
		return
	}
	for i, seq := range seqs {
		c := counts[seq]
		if c == nil {
			c = emptyDeliveryCounts()
		}
		items[i]["delivery_counts"] = c
	}

	next := ""
	if offset+len(items) < total {
		next = strconv.Itoa(offset + len(items))
	}
	httpx.WriteOK(w, map[string]any{"items": items, "next_cursor": next, "total": total})
}

func (h *Handler) handleMessageGet(w http.ResponseWriter, r *http.Request) {
	seqStr := r.PathValue("seq")
	seq, err := strconv.ParseInt(seqStr, 10, 64)
	if err != nil || seq < 1 {
		httpx.WriteError(w, http.StatusBadRequest, "bad_request", "seq 无效")
		return
	}
	q := r.URL.Query()
	limit, offset, ok := parsePage(w, q.Get("limit"), q.Get("cursor"))
	if !ok {
		return
	}

	var id, sender, destKind, destID, state, reason, contentType, meta string
	var sendAt, created int64
	err = h.db.Read.QueryRowContext(r.Context(), `
SELECT id, sender_id, dest_kind, dest_id, state, reason, send_at, created_at,
  content_type, meta
FROM messages WHERE seq = ?`, seq).Scan(
		&id, &sender, &destKind, &destID, &state, &reason, &sendAt, &created,
		&contentType, &meta,
	)
	if isNoRows(err) {
		httpx.WriteError(w, http.StatusNotFound, "not_found", "消息不存在")
		return
	}
	if err != nil {
		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
		return
	}

	var total int
	_ = h.db.Read.QueryRowContext(r.Context(),
		`SELECT COUNT(*) FROM deliveries WHERE seq = ?`, seq).Scan(&total)

	rows, err := h.db.Read.QueryContext(r.Context(), `
SELECT endpoint_id, state, reason, attempts, pushed_at, updated_at
FROM deliveries WHERE seq = ?
ORDER BY endpoint_id ASC
LIMIT ? OFFSET ?`, seq, limit, offset)
	if err != nil {
		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
		return
	}
	defer func() { _ = rows.Close() }()

	deliveries := make([]map[string]any, 0)
	for rows.Next() {
		var endpointID, dState, dReason string
		var attempts int
		var pushedAt, updatedAt sql.NullInt64
		if scanErr := rows.Scan(&endpointID, &dState, &dReason, &attempts, &pushedAt, &updatedAt); scanErr != nil {
			httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
			return
		}
		deliveries = append(deliveries, map[string]any{
			"endpoint_id":   endpointID,
			"state":         dState,
			"reason":        dReason,
			"attempts":      attempts,
			"pushed_at_ms":  nullInt64API(pushedAt),
			"updated_at_ms": updatedAt.Int64,
		})
	}
	if err := rows.Err(); err != nil {
		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
		return
	}

	next := ""
	if offset+len(deliveries) < total {
		next = strconv.Itoa(offset + len(deliveries))
	}

	metaObj := any(map[string]any{})
	if strings.TrimSpace(meta) != "" && meta != "{}" {
		metaObj = jsonRawOrObject(meta)
	}

	httpx.WriteOK(w, map[string]any{
		"seq":           seq,
		"id":            id,
		"sender_id":     sender,
		"dest_kind":     destKind,
		"dest_id":       destID,
		"state":         state,
		"reason":        reason,
		"send_at_ms":    sendAt,
		"created_at_ms": created,
		"meta":          metaObj,
		"content_type":  contentType,
		"deliveries":    deliveries,
		"next_cursor":   next,
	})
}

func buildMessageFilter(w http.ResponseWriter, q interface{ Get(string) string }) (where string, args []any, ok bool) {
	conds := []string{"1=1"}
	args = make([]any, 0, 8)

	if v := strings.TrimSpace(q.Get("sender_id")); v != "" {
		conds = append(conds, "m.sender_id = ?")
		args = append(args, v)
	}
	if v := strings.TrimSpace(q.Get("group_id")); v != "" {
		conds = append(conds, "m.dest_kind = 'group' AND m.dest_id = ?")
		args = append(args, v)
	}
	if v := strings.TrimSpace(q.Get("state")); v != "" {
		switch v {
		case "scheduled", "dispatched", "completed":
			conds = append(conds, "m.state = ?")
			args = append(args, v)
		default:
			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "state 无效")
			return "", nil, false
		}
	}
	if v := strings.TrimSpace(q.Get("from_ms")); v != "" {
		n, okParse := parseInt64Query(v)
		if !okParse {
			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "from_ms 无效")
			return "", nil, false
		}
		conds = append(conds, "m.created_at >= ?")
		args = append(args, n)
	}
	if v := strings.TrimSpace(q.Get("to_ms")); v != "" {
		n, okParse := parseInt64Query(v)
		if !okParse {
			httpx.WriteError(w, http.StatusBadRequest, "bad_request", "to_ms 无效")
			return "", nil, false
		}
		conds = append(conds, "m.created_at <= ?")
		args = append(args, n)
	}
	if v := strings.TrimSpace(q.Get("endpoint_id")); v != "" {
		conds = append(conds, `EXISTS (
SELECT 1 FROM deliveries d WHERE d.seq = m.seq AND d.endpoint_id = ?)`)
		args = append(args, v)
	}

	return strings.Join(conds, " AND "), args, true
}

func (h *Handler) deliveryCounts(ctx context.Context, seqs []int64) (map[int64]map[string]int, error) {
	out := make(map[int64]map[string]int, len(seqs))
	if len(seqs) == 0 {
		return out, nil
	}
	placeholders := make([]string, len(seqs))
	args := make([]any, len(seqs))
	for i, s := range seqs {
		placeholders[i] = "?"
		args[i] = s
		out[s] = emptyDeliveryCounts()
	}
	rows, err := h.db.Read.QueryContext(ctx, `
SELECT seq, state, COUNT(*) FROM deliveries
WHERE seq IN (`+strings.Join(placeholders, ",")+`)
GROUP BY seq, state`, args...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var seq int64
		var state string
		var n int
		if scanErr := rows.Scan(&seq, &state, &n); scanErr != nil {
			return nil, scanErr
		}
		c := out[seq]
		switch state {
		case "pending", "accepted", "recalled", "expired", "dropped", "rejected":
			c[state] = n
		}
	}
	return out, rows.Err()
}

func emptyDeliveryCounts() map[string]int {
	return map[string]int{
		"pending":  0,
		"accepted": 0,
		"recalled": 0,
		"expired":  0,
		"dropped":  0,
		"rejected": 0,
	}
}

func jsonRawOrObject(s string) any {
	// meta 存的是 JSON 文本；解析失败时返回空对象，避免把原始串当正文泄漏路径
	var v any
	if err := json.Unmarshal([]byte(s), &v); err != nil || v == nil {
		return map[string]any{}
	}
	return v
}