Cyrene_v2/internal/services/gateway/admin_channel_events.go

package gateway

// 渠道事件**只读查询面**（design/26 §4.5 / §6.5；切片 E6）。
//
// 它回答的正是 design/26 §0.2 那次无法定案的事故：「上游到底推过没有？是不是
// 到了适配器却被我们过滤掉了？」—— E1/E2/E3+E4 把每一帧的脱敏结构痕迹落进了
// `channel_events`（T2/T3/unknown/link 逐条）与 `channel_event_counters`（所有帧
// 的计数），本文件把那份证据变成**查得到**的内容，供管理面板「渠道事件」页使用。
//
// 四条口径（每一条都有测试钉住，绝不靠注释自觉）：
//
//  1. **只读**：本文件只有 List / Count / Counters 三个查询调用，没有写入路径，
//     也不碰任何凭据 —— 事件行里本就没有正文 / URL / `flag`（§4.2）。
//  2. **不静默裁剪窗口**：`since`/`until` 跨度超过 90 天回 400 并点名上限
//     （与 `adminLlmUsage` 同口径）；**没给窗口就不注入窗口** —— 替客户端造一个
//     下界，等于把「我没查那么久」伪装成「就那么少」。
//  3. **计数是近似值**：`Counters()` 读的是**已落盘**的计数表，不含内存里待 flush
//     的那部分（§4.3），因此响应恒带 `approx:true`，绝不谎称精确账。
//  4. **平台差异如实**：`weixin-ilink` 平台事实上没有非消息事件（§3.4），对它查询
//     就是空 —— 绝不拿别的渠道的事件顶替、绝不伪造（§8.5）。
//
// 未接线（组合根没注入留存面）时两个端点诚实作答 **500 `channel_events_unavailable`**，
// 绝不回一份全 0 的假汇总（与 `usage_unavailable` / `storage_unavailable` 同一口径）。

import (
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events"
)

// channelEventsLogSource 是本文件全部日志的 source 标签（AGENTS §10：禁无来源裸日志）。
const channelEventsLogSource = "gateway/admin_channel_events"

// channelEventsMaxWindow 是一次查询允许的最大时间窗（90 天）。它**刻意**与
// `model.UsageMaxWindow` 取同一个值：design/26 §4.5 明写「跨度上限 90 天」且
// 「与 `llm/usage` 同口径」，两处各对着同一条设计口径写死，谁也别想只改一处。
const channelEventsMaxWindow = 90 * 24 * time.Hour

// channelEventDTO 是一条 `channel_events` 行的线上投影。它现在是
// `events.ChannelEventJSON` 的**类型别名**（切片 E7）：同一条事件有两条线上出口 ——
// 这里的 REST 列表与实时 `channel_event` 帧的 `data.event`。两个出口共用一份结构，
// 是「不许两处各写一份形状」这条纪律的落地（否则迟早出现「帧里有这个键、列表里
// 没有」这类无从察觉的差异）。
//
// 列与 `events.Event` 一一对应且不多不少：它没有、也不会有承载平台原文、
// 媒体 URL、凭据或 OneBot `flag` 的字段（§4.2 红线）。`summary` 是适配器生成的
// 固定文案（如「群成员加入」）或本包的固定迁移文案，**已净化**，查不出原始文本。
// `id` 可缺席（实时帧里那行的 id 尚未回读），列表路径上它恒有值。
type channelEventDTO = events.ChannelEventJSON

// channelEventListResponse 是列表端点的 data 本体（§4.5：`{events, total}`）。
// `events` 恒为非 nil 切片（无数据时是 `[]` 而非 `null`，客户端不必判空）。
type channelEventListResponse struct {
	Events []channelEventDTO `json:"events"`
	Total  int64             `json:"total"`
}

// channelEventCounterDTO 是计数表的一行（§4.5 逐字给出字段名）。
type channelEventCounterDTO struct {
	Platform     string `json:"platform"`
	AccountID    string `json:"accountId"`
	Tier         string `json:"tier"`
	Category     string `json:"category"`
	PlatformType string `json:"platformType"`
	Count        int64  `json:"count"`
	FirstSeenAt  string `json:"firstSeenAt"`
	LastSeenAt   string `json:"lastSeenAt"`
}

// channelEventCountersResponse 是计数端点的 data 本体。
//
// `approx` 不是装饰：计数在内存里聚合、每 30s 落一次盘（§4.3），面板读到的表
// **不含**待 flush 的部分，因此它只少不多 —— 是个下界，不是精确账。这个字段就是
// 把这句话传到线上，免得面板把近似值画成事实。
type channelEventCountersResponse struct {
	Approx   bool                     `json:"approx"`
	Counters []channelEventCounterDTO `json:"counters"`
}

// adminChannelEventsList 处理 GET /api/v1/admin/channels/events（列表，倒序）。
// RBAC 沿用既有管理读中间件（与同族的 `/admin/channels/messages` 同门）。
func (s *Server) adminChannelEventsList(w http.ResponseWriter, r *http.Request) {
	if s.channelEvents == nil {
		writeError(w, apperr.Wrap(apperr.ErrInternal, "channel_events_unavailable", "渠道事件留存未接线"))
		return
	}
	filter, err := channelEventsListFilter(r)
	if err != nil {
		writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "%s", err.Error()))
		return
	}
	rows, err := s.channelEvents.List(r.Context(), filter)
	if err != nil {
		// 原始错误（可能带 SQL/路径）只进日志；响应只给稳定 code + 中文说明。
		logger.WithSource(s.log, channelEventsLogSource).Error("list channel events", "err", err)
		writeError(w, apperr.Wrap(apperr.ErrInternal, "channel_events_unavailable", "读取渠道事件列表失败"))
		return
	}
	total, err := s.channelEvents.Count(r.Context(), filter)
	if err != nil {
		logger.WithSource(s.log, channelEventsLogSource).Error("count channel events", "err", err)
		writeError(w, apperr.Wrap(apperr.ErrInternal, "channel_events_unavailable", "统计渠道事件条数失败"))
		return
	}
	writeJSON(w, http.StatusOK, channelEventsListResponseOf(rows, total))
}

// adminChannelEventsCounters 处理 GET /api/v1/admin/channels/events/counters
// （逐类计数汇总，近似值）。`platform` 空 = 全部平台。
func (s *Server) adminChannelEventsCounters(w http.ResponseWriter, r *http.Request) {
	if s.channelEvents == nil {
		writeError(w, apperr.Wrap(apperr.ErrInternal, "channel_events_unavailable", "渠道事件留存未接线"))
		return
	}
	platform, err := parseChannelEventPlatform(r.URL.Query().Get("platform"))
	if err != nil {
		writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "%s", err.Error()))
		return
	}
	counters, err := s.channelEvents.Counters(r.Context(), platform)
	if err != nil {
		logger.WithSource(s.log, channelEventsLogSource).Error("read channel event counters", "err", err)
		writeError(w, apperr.Wrap(apperr.ErrInternal, "channel_events_unavailable", "读取渠道事件计数失败"))
		return
	}
	writeJSON(w, http.StatusOK, channelEventCountersResponseOf(counters))
}

// channelEventsListFilter 解析并**校验**列表端点的查询参数。校验顺序是「形状先于
// 语义」：一个拼错的时间点 / 平台名直接 400，绝不拿零值条件去查库。
func channelEventsListFilter(r *http.Request) (events.Filter, error) {
	params := r.URL.Query()
	platform, err := parseChannelEventPlatform(params.Get("platform"))
	if err != nil {
		return events.Filter{}, err
	}
	tier, err := parseChannelEventTier(params.Get("tier"))
	if err != nil {
		return events.Filter{}, err
	}
	since, until, err := parseChannelEventWindow(params.Get("since"), params.Get("until"))
	if err != nil {
		return events.Filter{}, err
	}
	limit, err := parseChannelEventCount("limit", params.Get("limit"))
	if err != nil {
		return events.Filter{}, err
	}
	offset, err := parseChannelEventCount("offset", params.Get("offset"))
	if err != nil {
		return events.Filter{}, err
	}
	return events.Filter{
		Platform:  platform,
		Tier:      tier,
		Category:  channelEventCategory(params.Get("category")),
		ChannelID: strings.TrimSpace(params.Get("channelId")),
		Since:     since,
		Until:     until,
		Limit:     clampChannelEventLimit(limit),
		Offset:    offset,
	}, nil
}

// channelEventPlatforms 返回**已知平台**（design/26 §4.1 的 platform 列取值，
// 与 `internal/runtime/events` 的平台常量逐字一致）。它按调用返回而不是包级变量：
// AGENTS §10 禁包级状态，且白名单只有这一处定义，不存在两份漂移。
//
// 已知集合**不含** `weixin-ilink` 以外的微信别名 —— 平台名就是适配器写进
// `Platform` 字段的那个字符串（`plugins/channels/*` 的 platformName）。
func channelEventPlatforms() []string {
	return []string{
		events.PlatformQQ,
		events.PlatformKook,
		events.PlatformTelegram,
		events.PlatformWeixinILink,
		events.PlatformWebhook,
	}
}

// parseChannelEventPlatform 校验 `platform` 过滤参数：空串 = 不过滤；不在白名单里
// ⇒ 报错（§4.5：`platform` 未知 → 400）。
//
// 绝不把未知平台当成「不过滤」静默放行 —— 那会让一次拼错的筛选返回**全部**渠道的
// 事件，比返回空更误导（这正是「不静默」要防的那一类）。
func parseChannelEventPlatform(raw string) (string, error) {
	platform := strings.TrimSpace(raw)
	if platform == "" {
		return "", nil
	}
	for _, known := range channelEventPlatforms() {
		if platform == known {
			return platform, nil
		}
	}
	return "", fmt.Errorf("platform 未知：%q（取值 %s）", raw, strings.Join(channelEventPlatforms(), "/"))
}

// parseChannelEventTier 校验 `tier` 过滤参数。档位只有三个取值（§2.1，迁移 0021 的
// 列取值即它们），非法值直接 400：一个写错的档位若被当成「不过滤」，面板会显示
// 全部档位却让人以为只看了一档。
func parseChannelEventTier(raw string) (events.Tier, error) {
	switch tier := events.Tier(strings.TrimSpace(raw)); tier {
	case "":
		return "", nil
	case events.TierT1, events.TierT2, events.TierT3:
		return tier, nil
	default:
		return "", fmt.Errorf("tier 非法：%q（取值 t1/t2/t3）", raw)
	}
}

// channelEventCategory 归一 `category` 过滤参数（精确匹配，空 = 不过滤）。
//
// 它**刻意不做白名单校验**：类别轴的权威在适配器层（`adapter.NoticeCategory*`），
// 会随新平台持续增加（KOOK / Telegram 的 `reaction`/`button`/`item`/`vote` 等），
// 在网关侧钉一份封闭集合只会让将来新登记的类别被 400 挡住。未知类别是一次
// **精确匹配**（返回空），不会把筛选放宽成「全部」—— 与未知平台的处理相反，理由
// 也正在这里：返回空是诚实的，返回全部才不是。
func channelEventCategory(raw string) events.Category {
	return events.Category(strings.TrimSpace(raw))
}

// parseChannelEventCount 解析一个非负整数参数（limit / offset）。空串 = 未给（0）；
// 非法整数报错；负值归一成 0（= 缺省语义），绝不把一个负页大小悄悄变成「全表」。
func parseChannelEventCount(name, raw string) (int, error) {
	trimmed := strings.TrimSpace(raw)
	if trimmed == "" {
		return 0, nil
	}
	n, err := strconv.Atoi(trimmed)
	if err != nil {
		return 0, fmt.Errorf("%s 必须是整数：%q", name, raw)
	}
	if n < 0 {
		return 0, nil
	}
	return n, nil
}

// clampChannelEventLimit 把页大小钳到上限（§4.5 的分页钳制）。超上限即钳住而不是
// 报错 —— 上限由 `events.MaxListLimit` 单点给出，这里不复写一份数字；0（未给 /
// 负值）保持 0，由存储层用 `events.DefaultListLimit` 补缺省。
func clampChannelEventLimit(n int) int {
	if n > events.MaxListLimit {
		return events.MaxListLimit
	}
	return n
}

// parseChannelEventWindow 解析 `since`/`until` 成一个闭区间，并守住 90 天上限。
//
// 归一规则（三条，都是「不静默裁剪」的直接推论）：
//
//   - **两个都没给**：不注入任何时间条件（客户端要的就是不加时间过滤）；
//   - **只给 since**：上界按「现在」算来判断跨度（`since` 早于 90 天前 ⇒ 400），
//     但**不**把「现在」写进 until —— 判跨度与改条件是两回事；
//   - **只给 until**：只给上界，不替客户端造下界。造了就是把「我没指定下界」
//     偷偷换成「我只看最近 N 天」。
//
// 时间点接受 RFC3339 与 Unix 毫秒两种形态（复用同包的 `parseInstantParam`，与
// `llm/usage` 同一口径）；落进 Filter 前是 `time.Time`，由存储层格式化成
// UTC 毫秒 + 字面量 `Z` 的定宽字符串，从而吃得上 `idx_channel_events_created_at`。
func parseChannelEventWindow(sinceRaw, untilRaw string) (time.Time, time.Time, error) {
	since, err := parseInstantParam(sinceRaw)
	if err != nil {
		return time.Time{}, time.Time{}, fmt.Errorf("since 非法：%w", err)
	}
	until, err := parseInstantParam(untilRaw)
	if err != nil {
		return time.Time{}, time.Time{}, fmt.Errorf("until 非法：%w", err)
	}
	switch {
	case since.IsZero() && until.IsZero():
		return time.Time{}, time.Time{}, nil
	case since.IsZero():
		return time.Time{}, until, nil
	case until.IsZero():
		if span := time.Since(since); span > channelEventsMaxWindow {
			return time.Time{}, time.Time{}, fmt.Errorf("时间窗超过上限 %s", channelEventsMaxWindow)
		}
		return since, time.Time{}, nil
	}
	if !since.Before(until) {
		return time.Time{}, time.Time{}, fmt.Errorf("since 必须早于 until")
	}
	if until.Sub(since) > channelEventsMaxWindow {
		return time.Time{}, time.Time{}, fmt.Errorf("时间窗超过上限 %s", channelEventsMaxWindow)
	}
	return since, until, nil
}

// channelEventsListResponseOf 把存储层结果翻成线上结构。时间一律以 **UTC RFC3339**
// 回显（与 `llm/usage` 同口径）。
func channelEventsListResponseOf(rows []events.Event, total int64) channelEventListResponse {
	out := channelEventListResponse{
		Events: make([]channelEventDTO, 0, len(rows)),
		Total:  total,
	}
	for _, ev := range rows {
		out.Events = append(out.Events, channelEventDTOOf(ev))
	}
	return out
}

// channelEventDTOOf 是单行投影（一处换算，避免 List 与实时帧各写一遍）。
func channelEventDTOOf(ev events.Event) channelEventDTO {
	return ev.JSON()
}

// channelEventCountersResponseOf 把计数表结果翻成线上结构。`approx` 恒为 true：
// 计数面的近似语义来自数据源本身（内存聚合 + 周期 flush），不是某次查询的偶然。
func channelEventCountersResponseOf(counters []events.Counter) channelEventCountersResponse {
	out := channelEventCountersResponse{
		Approx:   true,
		Counters: make([]channelEventCounterDTO, 0, len(counters)),
	}
	for _, c := range counters {
		out.Counters = append(out.Counters, channelEventCounterDTO{
			Platform:     c.Platform,
			AccountID:    c.AccountID,
			Tier:         string(c.Tier),
			Category:     string(c.Category),
			PlatformType: c.PlatformType,
			Count:        c.Count,
			FirstSeenAt:  c.FirstSeenAt.UTC().Format(time.RFC3339),
			LastSeenAt:   c.LastSeenAt.UTC().Format(time.RFC3339),
		})
	}
	return out
}