feat(gateway): 渠道事件查询面 + 面板「渠道事件」页（design/26 E6）
- 两个只读端点：GET /admin/channels/events（列表，8 个过滤条件 + 分页钳制）与 .../events/counters（逐类计数，恒带 approx:true 如实标「近似」） - 诚实边界：>90 天窗口 ⇒ 400（不静默裁剪）；未知 platform/档位 ⇒ 400；**未接线 ⇒ 500 channel_events_unavailable**（绝不回空 200 假装「没有事件」）；微信平台的空如实说明「该渠道不提供非消息事件」而不借别平台顶替 - 面板：新 section「渠道事件」（计数卡 + 事件行 + 过滤条 + 30s 静默刷新），空/错误态分开且失败显示错误不显示 0 - 组合根接线（主会话补）：wire_frontend.go 的 Deps 注入 EventStore: a.channelEventStore()（未接线时为真 nil ⇒ 保留诚实 500） - 契约不升版（查询面 v0.67 已登记）：就地改「未实现 → 已实现 + 端点路径」，并补登记 5 处原先留白的落地值（camelCase / 通用 bad_request / RBAC owner|admin / data.approx / total 语义） - 顺带把 §5 三个 CHANNEL_EVENT_* env 键由「尚未实现」改「已接线」（E4） - 变异实测：去掉 90 天上限 ⇒ status 200 want 400；未接线回空 200 ⇒ status 200 want 500 - 债务登记：events.Filter 无 AccountID 列 ⇒ 服务端暂不能按账号筛（列表仍如实下发 accountId），要加需先改 internal/runtime/events
This commit is contained in:
AskaEth committed
1 parent
a0abf0199a
commit
c8053b1f36
21 files changed
+1805
-30
No files matched your search
@@ -0,0 +1,369 @@
|
||||
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` 行的线上投影（键名与同族的
|
||||
// `counters[]` 一致，取 camelCase）。
|
||||
//
|
||||
// 列与 `events.Event` **一一对应且不多不少**：它没有、也不会有承载平台原文、
|
||||
// 媒体 URL、凭据或 OneBot `flag` 的字段（§4.2 红线）。`summary` 是适配器生成的
|
||||
// 固定文案（如「群成员加入」），**已净化**，查不出原始文本。
|
||||
type channelEventDTO struct {
|
||||
ID int64 `json:"id"`
|
||||
Platform string `json:"platform"`
|
||||
AccountID string `json:"accountId"`
|
||||
Tier string `json:"tier"`
|
||||
Category string `json:"category"`
|
||||
PlatformType string `json:"platformType"`
|
||||
ChannelID string `json:"channelId"`
|
||||
ChannelType string `json:"channelType"`
|
||||
GroupID string `json:"groupId"`
|
||||
ActorID string `json:"actorId"`
|
||||
ActorRole string `json:"actorRole"`
|
||||
TargetID string `json:"targetId"`
|
||||
SelfRelated bool `json:"selfRelated"`
|
||||
MessageID string `json:"messageId"`
|
||||
Summary string `json:"summary"`
|
||||
CreatedAt string `json:"createdAt"`
|
||||
UpdatedAt string `json:"updatedAt"`
|
||||
}
|
||||
|
||||
// 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 channelEventDTO{
|
||||
ID: ev.ID,
|
||||
Platform: ev.Platform,
|
||||
AccountID: ev.AccountID,
|
||||
Tier: string(ev.Tier),
|
||||
Category: string(ev.Category),
|
||||
PlatformType: ev.PlatformType,
|
||||
ChannelID: ev.ChannelID,
|
||||
ChannelType: ev.ChannelType,
|
||||
GroupID: ev.GroupID,
|
||||
ActorID: ev.ActorID,
|
||||
ActorRole: ev.ActorRole,
|
||||
TargetID: ev.TargetID,
|
||||
SelfRelated: ev.SelfRelated,
|
||||
MessageID: ev.MessageID,
|
||||
Summary: ev.Summary,
|
||||
CreatedAt: ev.CreatedAt.UTC().Format(time.RFC3339),
|
||||
UpdatedAt: ev.UpdatedAt.UTC().Format(time.RFC3339),
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
@@ -0,0 +1,177 @@
|
||||
package gateway
|
||||
|
||||
// 本文件是渠道事件查询面的**跨层**闭环（design/26 §4.5；切片 E6）：用真的写入
|
||||
// 路径（`events.SQLStore.Record` + `Flush`）把事件落进真 SQLite（跑迁移 0021），
|
||||
// 再经**真的** HTTP handler 读回来。它钉住三件只有端到端才成立的事：
|
||||
//
|
||||
// - 计数面读的是**已落盘**的计数表且恒带 `approx:true`（近似值语义）；
|
||||
// - T1 遥测（心跳）**只计数不落行**（这正是「近似值」的来源）；
|
||||
// - `weixin-ilink` 没有非消息事件时如实为空，**绝不借别的平台的事件顶替**（§3.4/§8.5）。
|
||||
//
|
||||
// 参数翻译与诚实边界在 `admin_channel_events_test.go`。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events"
|
||||
)
|
||||
|
||||
type channelEventCounterRowDTO 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"`
|
||||
}
|
||||
|
||||
type channelEventCountersDTO struct {
|
||||
Approx bool `json:"approx"`
|
||||
Counters []channelEventCounterRowDTO `json:"counters"`
|
||||
}
|
||||
|
||||
// decodeChannelEventCounters 解出计数端点的 data 本体。
|
||||
func decodeChannelEventCounters(t *testing.T, body []byte) channelEventCountersDTO {
|
||||
t.Helper()
|
||||
var out channelEventCountersDTO
|
||||
if err := json.Unmarshal(decodeBody(t, body)["data"], &out); err != nil {
|
||||
t.Fatalf("decode counters data: %v (body=%s)", err, body)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// channelEventTestStore 打开一个真实的 SQLite 文件库、跑迁移，并构造 E1 的
|
||||
// 事件存储 —— 端点与写入面用的是**同一台** store（与生产同一条路）。
|
||||
func channelEventTestStore(t *testing.T, dir string) (*sql.DB, *events.SQLStore) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
conn, err := db.Open(ctx, filepath.Join(dir, "cyrene.db"), gwTestLogger())
|
||||
if err != nil {
|
||||
t.Fatalf("open db: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = conn.Close() })
|
||||
if err := db.MigrateUp(ctx, conn, gwTestLogger()); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
store, err := events.NewStore(events.Config{DB: conn, Log: gwTestLogger()})
|
||||
if err != nil {
|
||||
t.Fatalf("new event store: %v", err)
|
||||
}
|
||||
return conn, store
|
||||
}
|
||||
|
||||
// TestAdminChannelEventsCountersAreApproxAndOnlyWhatIsOnDisk 断言已落盘计数的
|
||||
// 真值（心跳 2 帧 = count 2）与 T1「只计数不落行」，并顺带钉住微信平台的空。
|
||||
func TestAdminChannelEventsCountersAreApproxAndOnlyWhatIsOnDisk(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
_, store := channelEventTestStore(t, t.TempDir())
|
||||
heartbeat := events.Event{
|
||||
Platform: events.PlatformQQ, AccountID: "acct-1", Tier: events.TierT1,
|
||||
Category: events.CategoryHeartbeat, PlatformType: "heartbeat",
|
||||
}
|
||||
for i := 0; i < 2; i++ {
|
||||
if err := store.Record(ctx, heartbeat); err != nil {
|
||||
t.Fatalf("record heartbeat: %v", err)
|
||||
}
|
||||
}
|
||||
join := events.Event{
|
||||
Platform: events.PlatformQQ, AccountID: "acct-1", Tier: events.TierT2,
|
||||
Category: events.CategoryJoin, PlatformType: "group_increase",
|
||||
ChannelID: "g1", ChannelType: events.ChannelGroup, ActorID: "10001", Summary: "群成员加入",
|
||||
}
|
||||
if err := store.Record(ctx, join); err != nil {
|
||||
t.Fatalf("record join: %v", err)
|
||||
}
|
||||
if err := store.Flush(ctx); err != nil {
|
||||
t.Fatalf("flush counters: %v", err)
|
||||
}
|
||||
|
||||
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{EventStore: store})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
status, body := getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events/counters", "", tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("counters status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
counters := decodeChannelEventCounters(t, body)
|
||||
if !counters.Approx {
|
||||
t.Error("counters response must carry approx:true（计数是内存聚合 + 周期 flush 的近似值）")
|
||||
}
|
||||
byType := make(map[string]int64, len(counters.Counters))
|
||||
for _, c := range counters.Counters {
|
||||
byType[c.Platform+"/"+c.PlatformType] = c.Count
|
||||
}
|
||||
if byType["qq-obv11/heartbeat"] != 2 {
|
||||
t.Errorf("heartbeat count = %d, want 2 (%+v)", byType["qq-obv11/heartbeat"], counters.Counters)
|
||||
}
|
||||
if byType["qq-obv11/group_increase"] != 1 {
|
||||
t.Errorf("group_increase count = %d, want 1 (%+v)", byType["qq-obv11/group_increase"], counters.Counters)
|
||||
}
|
||||
|
||||
status, body = getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events", "platform=qq-obv11", tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("list status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
list := decodeChannelEvents(t, body)
|
||||
if list.Total != 1 || len(list.Events) != 1 {
|
||||
t.Fatalf("qq list = %+v, want exactly the T2 row (T1 心跳只计数不落行)", list)
|
||||
}
|
||||
if list.Events[0].PlatformType != "group_increase" {
|
||||
t.Errorf("qq row = %+v, want the join event", list.Events[0])
|
||||
}
|
||||
|
||||
// 微信渠道**没有**非消息事件（§3.4）：如实为空，绝不借别的平台顶替。
|
||||
status, body = getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events",
|
||||
"platform="+events.PlatformWeixinILink, tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("weixin list status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
weixin := decodeChannelEvents(t, body)
|
||||
if weixin.Total != 0 || len(weixin.Events) != 0 {
|
||||
t.Errorf("weixin list = %+v, want empty（该渠道不提供非消息事件）", weixin)
|
||||
}
|
||||
if strings.Contains(string(body), events.PlatformQQ) {
|
||||
t.Errorf("weixin response borrowed another platform's events: %s", body)
|
||||
}
|
||||
|
||||
status, body = getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events/counters",
|
||||
"platform="+events.PlatformWeixinILink, tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("weixin counters status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
weixinCounters := decodeChannelEventCounters(t, body)
|
||||
if len(weixinCounters.Counters) != 0 {
|
||||
t.Errorf("weixin counters = %+v, want empty", weixinCounters.Counters)
|
||||
}
|
||||
if !strings.Contains(string(body), `"counters":[]`) {
|
||||
t.Errorf("empty counters must serialise as [] not null: %s", body)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminChannelEventsEmptyListSerialisesAsArray 钉住线上的空形状：没有事件时
|
||||
// `events` 是 `[]`（客户端不必在 null 与空数组之间分叉）。
|
||||
func TestAdminChannelEventsEmptyListSerialisesAsArray(t *testing.T) {
|
||||
t.Parallel()
|
||||
stub := &channelEventStoreStub{}
|
||||
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{EventStore: stub})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
status, body := getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events", "platform=kook", tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
if !strings.Contains(string(body), `"events":[]`) {
|
||||
t.Errorf("empty list must serialise as [] not null: %s", body)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,371 @@
|
||||
package gateway
|
||||
|
||||
// 本文件钉住渠道事件只读查询面的**参数翻译与诚实边界**（design/26 §4.5；切片 E6，
|
||||
// 契约 v0.67 登记的两个 admin 端点）：
|
||||
//
|
||||
// - 过滤（平台/档位/类别/会话/时间窗）逐条落到存储层条件上；
|
||||
// - **跨度 > 90 天 ⇒ 400**（不静默裁剪）；没给时间窗就**不注入**时间窗；
|
||||
// - 分页钳制（超上限钳到 MaxListLimit）；
|
||||
// - 未接线 / 读库失败 ⇒ **500 channel_events_unavailable**（绝不回全 0 假汇总）；
|
||||
// - 空响应是 `[]` 而不是 `null`。
|
||||
//
|
||||
// 真实存储的端到端行为（同一台 SQLStore、真 SQLite、真迁移）在
|
||||
// `admin_channel_events_store_test.go`。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events"
|
||||
)
|
||||
|
||||
// channelEventStoreStub 是 ChannelEventStore 的测试替身：记录**端点传下去的**
|
||||
// 查询条件，回放预置结果。它让「端点怎么翻译查询参数」与「端点怎么翻线上形状」
|
||||
// 可以分别断言。
|
||||
type channelEventStoreStub struct {
|
||||
rows []events.Event
|
||||
counters []events.Counter
|
||||
|
||||
listErr error
|
||||
countErr error
|
||||
countersErr error
|
||||
|
||||
gotList []events.Filter
|
||||
gotCount []events.Filter
|
||||
gotPlatforms []string
|
||||
}
|
||||
|
||||
func (s *channelEventStoreStub) List(_ context.Context, f events.Filter) ([]events.Event, error) {
|
||||
s.gotList = append(s.gotList, f)
|
||||
if s.listErr != nil {
|
||||
return nil, s.listErr
|
||||
}
|
||||
return s.rows, nil
|
||||
}
|
||||
|
||||
func (s *channelEventStoreStub) Count(_ context.Context, f events.Filter) (int64, error) {
|
||||
s.gotCount = append(s.gotCount, f)
|
||||
if s.countErr != nil {
|
||||
return 0, s.countErr
|
||||
}
|
||||
return int64(len(s.rows)), nil
|
||||
}
|
||||
|
||||
func (s *channelEventStoreStub) Counters(_ context.Context, platform string) ([]events.Counter, error) {
|
||||
s.gotPlatforms = append(s.gotPlatforms, platform)
|
||||
if s.countersErr != nil {
|
||||
return nil, s.countersErr
|
||||
}
|
||||
return s.counters, nil
|
||||
}
|
||||
|
||||
// 线上形状的本地复刻（键名与 admin_channel_events.go 逐字对齐；写错就解不出来）。
|
||||
type channelEventRowDTO struct {
|
||||
ID int64 `json:"id"`
|
||||
Platform string `json:"platform"`
|
||||
AccountID string `json:"accountId"`
|
||||
Tier string `json:"tier"`
|
||||
Category string `json:"category"`
|
||||
PlatformType string `json:"platformType"`
|
||||
ChannelID string `json:"channelId"`
|
||||
ChannelType string `json:"channelType"`
|
||||
GroupID string `json:"groupId"`
|
||||
ActorID string `json:"actorId"`
|
||||
ActorRole string `json:"actorRole"`
|
||||
TargetID string `json:"targetId"`
|
||||
SelfRelated bool `json:"selfRelated"`
|
||||
MessageID string `json:"messageId"`
|
||||
Summary string `json:"summary"`
|
||||
CreatedAt string `json:"createdAt"`
|
||||
UpdatedAt string `json:"updatedAt"`
|
||||
}
|
||||
|
||||
type channelEventListDTO struct {
|
||||
Events []channelEventRowDTO `json:"events"`
|
||||
Total int64 `json:"total"`
|
||||
}
|
||||
|
||||
// getChannelEvents 以 owner token 请求一个渠道事件端点。
|
||||
func getChannelEvents(t *testing.T, url, path, query, token string) (int, []byte) {
|
||||
t.Helper()
|
||||
if query != "" {
|
||||
path += "?" + query
|
||||
}
|
||||
resp, body := doReq(t, http.MethodGet, url+path, nil, token)
|
||||
return resp.StatusCode, body
|
||||
}
|
||||
|
||||
// decodeChannelEvents 解出列表端点的 data 本体。
|
||||
func decodeChannelEvents(t *testing.T, body []byte) channelEventListDTO {
|
||||
t.Helper()
|
||||
var out channelEventListDTO
|
||||
if err := json.Unmarshal(decodeBody(t, body)["data"], &out); err != nil {
|
||||
t.Fatalf("decode events data: %v (body=%s)", err, body)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// channelEventErrorCode 取出错误信封里的 code（形状不对时 fatal）。
|
||||
func channelEventErrorCode(t *testing.T, body []byte) string {
|
||||
t.Helper()
|
||||
wantErrorShape(t, body)
|
||||
var eobj struct {
|
||||
Code string `json:"code"`
|
||||
}
|
||||
if err := json.Unmarshal(decodeBody(t, body)["error"], &eobj); err != nil {
|
||||
t.Fatalf("decode error envelope: %v (body=%s)", err, body)
|
||||
}
|
||||
return eobj.Code
|
||||
}
|
||||
|
||||
// TestAdminChannelEventsListPassesEveryFilterDown 断言 8 个查询参数**逐条**落到
|
||||
// 存储层条件上（少传一个就是静默放宽筛选），且回上来的行按线上形状投影。
|
||||
func TestAdminChannelEventsListPassesEveryFilterDown(t *testing.T) {
|
||||
t.Parallel()
|
||||
created := time.Date(2026, 10, 5, 3, 4, 5, 600_000_000, time.UTC)
|
||||
stub := &channelEventStoreStub{rows: []events.Event{{
|
||||
ID: 7, Platform: events.PlatformQQ, AccountID: "acct-1", Tier: events.TierT2,
|
||||
Category: events.CategoryJoin, PlatformType: "group_increase",
|
||||
ChannelID: "g1", ChannelType: events.ChannelGroup, GroupID: "g1",
|
||||
ActorID: "10001", ActorRole: events.RoleMember, TargetID: "0", SelfRelated: true,
|
||||
MessageID: "m-9", Summary: "群成员加入", CreatedAt: created, UpdatedAt: created,
|
||||
}}}
|
||||
since := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
|
||||
until := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
|
||||
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{EventStore: stub})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
query := url.Values{
|
||||
"platform": {events.PlatformQQ},
|
||||
"tier": {string(events.TierT2)},
|
||||
"category": {string(events.CategoryJoin)},
|
||||
"channelId": {"g1"},
|
||||
"since": {since.Format(time.RFC3339)},
|
||||
"until": {until.Format(time.RFC3339)},
|
||||
"limit": {"25"},
|
||||
"offset": {"5"},
|
||||
}
|
||||
status, body := getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events", query.Encode(), tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
if len(stub.gotList) != 1 {
|
||||
t.Fatalf("store.List calls = %d, want exactly 1", len(stub.gotList))
|
||||
}
|
||||
got := stub.gotList[0]
|
||||
want := events.Filter{
|
||||
Platform: events.PlatformQQ, Tier: events.TierT2, Category: events.CategoryJoin,
|
||||
ChannelID: "g1", Since: since, Until: until, Limit: 25, Offset: 5,
|
||||
}
|
||||
if got != want {
|
||||
t.Errorf("store filter = %+v, want %+v", got, want)
|
||||
}
|
||||
// Count 用的是**同一个**条件（否则 total 与 events 说的不是同一件事）。
|
||||
if len(stub.gotCount) != 1 || stub.gotCount[0] != want {
|
||||
t.Errorf("store count filters = %+v, want one %+v", stub.gotCount, want)
|
||||
}
|
||||
|
||||
list := decodeChannelEvents(t, body)
|
||||
if list.Total != 1 || len(list.Events) != 1 {
|
||||
t.Fatalf("list = %+v, want one row and total=1", list)
|
||||
}
|
||||
row := list.Events[0]
|
||||
if row.ID != 7 || row.Platform != events.PlatformQQ || row.AccountID != "acct-1" {
|
||||
t.Errorf("row identity = %+v", row)
|
||||
}
|
||||
if row.Tier != string(events.TierT2) || row.Category != string(events.CategoryJoin) {
|
||||
t.Errorf("row tier/category = %q/%q", row.Tier, row.Category)
|
||||
}
|
||||
if row.ChannelType != events.ChannelGroup || row.GroupID != "g1" || row.ActorID != "10001" {
|
||||
t.Errorf("row channel/actor = %+v", row)
|
||||
}
|
||||
if !row.SelfRelated || row.MessageID != "m-9" || row.Summary != "群成员加入" {
|
||||
t.Errorf("row self/message/summary = %+v", row)
|
||||
}
|
||||
if row.CreatedAt != created.Format(time.RFC3339) || row.UpdatedAt != created.Format(time.RFC3339) {
|
||||
t.Errorf("row times = %q/%q, want UTC RFC3339 %q", row.CreatedAt, row.UpdatedAt, created.Format(time.RFC3339))
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminChannelEventsListRejectsWindowOver90Days 钉住 §4.5 的「跨度上限 90 天，
|
||||
// 超 → 400」：**绝不静默裁剪**成 90 天（那会让客户端把「我没查那么久」误读成
|
||||
// 「就那么少」）。只给 since 时同样按「现在」算跨度。
|
||||
func TestAdminChannelEventsListRejectsWindowOver90Days(t *testing.T) {
|
||||
t.Parallel()
|
||||
stub := &channelEventStoreStub{}
|
||||
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{EventStore: stub})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
now := time.Now().UTC()
|
||||
old := now.Add(-91 * 24 * time.Hour)
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
query url.Values
|
||||
}{
|
||||
{"both bounds", url.Values{"since": {old.Format(time.RFC3339)}, "until": {now.Format(time.RFC3339)}}},
|
||||
{"since only", url.Values{"since": {old.Format(time.RFC3339)}}},
|
||||
} {
|
||||
tc := tc
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
status, body := getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events", tc.query.Encode(), tok)
|
||||
if status != http.StatusBadRequest {
|
||||
t.Fatalf("status = %d, want 400 (body=%s)", status, body)
|
||||
}
|
||||
if code := channelEventErrorCode(t, body); code != "bad_request" {
|
||||
t.Errorf("error code = %q, want bad_request", code)
|
||||
}
|
||||
if !strings.Contains(string(body), "90d") && !strings.Contains(string(body), "2160h") {
|
||||
t.Errorf("400 message must name the cap: %s", body)
|
||||
}
|
||||
// 被拒的请求**不得**摸到存储层（否则「静默裁剪」会以另一种形式回来）。
|
||||
if len(stub.gotList) != 0 {
|
||||
t.Errorf("store was queried %d time(s) for a rejected window", len(stub.gotList))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminChannelEventsListDoesNotInventAWindow 断言「没给窗口就不注入窗口」：
|
||||
// 端点的缺省不是「近 24h」，而是「不加时间条件」—— 替客户端造下界就是把
|
||||
// 「我没指定」偷偷换成「我只看了最近一段」。
|
||||
func TestAdminChannelEventsListDoesNotInventAWindow(t *testing.T) {
|
||||
t.Parallel()
|
||||
stub := &channelEventStoreStub{}
|
||||
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{EventStore: stub})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
status, body := getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events", "", tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
if len(stub.gotList) != 1 {
|
||||
t.Fatalf("store.List calls = %d, want 1", len(stub.gotList))
|
||||
}
|
||||
if f := stub.gotList[0]; !f.Since.IsZero() || !f.Until.IsZero() {
|
||||
t.Errorf("filter window = %v..%v, want both zero (no invented window)", f.Since, f.Until)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminChannelEventsListClampsPagination 钉住分页钳制：超上限**钳到**
|
||||
// events.MaxListLimit（500），未给/负值保持「缺省」，非法整数 400（形状先于语义）。
|
||||
func TestAdminChannelEventsListClampsPagination(t *testing.T) {
|
||||
t.Parallel()
|
||||
stub := &channelEventStoreStub{}
|
||||
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{EventStore: stub})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
for _, tc := range []struct {
|
||||
query string
|
||||
wantLimit int
|
||||
wantOff int
|
||||
}{
|
||||
{"limit=100000", events.MaxListLimit, 0},
|
||||
{"limit=0", 0, 0},
|
||||
{"limit=-3&offset=-9", 0, 0},
|
||||
{"limit=17&offset=3", 17, 3},
|
||||
} {
|
||||
tc := tc
|
||||
t.Run(tc.query, func(t *testing.T) {
|
||||
before := len(stub.gotList)
|
||||
status, body := getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events", tc.query, tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
if len(stub.gotList) != before+1 {
|
||||
t.Fatalf("store.List calls = %d, want %d", len(stub.gotList), before+1)
|
||||
}
|
||||
f := stub.gotList[len(stub.gotList)-1]
|
||||
if f.Limit != tc.wantLimit || f.Offset != tc.wantOff {
|
||||
t.Errorf("filter limit/offset = %d/%d, want %d/%d", f.Limit, f.Offset, tc.wantLimit, tc.wantOff)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// 非法整数是**形状错误**，不是「悄悄用缺省」。
|
||||
status, body := getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events", "limit=abc", tok)
|
||||
if status != http.StatusBadRequest {
|
||||
t.Fatalf("limit=abc: status = %d, want 400 (body=%s)", status, body)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminChannelEventsListRejectsUnknownPlatformAndTier 钉住 §4.5 的
|
||||
// 「`platform` 未知 → 400」以及档位白名单：未知平台**绝不**被当成「不过滤」
|
||||
// 静默放行（那会让一次拼错的筛选返回全部渠道的事件）。类别则相反 —— 类别轴的
|
||||
// 权威在适配器层且会持续增加，因此未知类别是一次**精确匹配**（返回空），不是 400。
|
||||
func TestAdminChannelEventsListRejectsUnknownPlatformAndTier(t *testing.T) {
|
||||
t.Parallel()
|
||||
stub := &channelEventStoreStub{}
|
||||
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{EventStore: stub})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
for _, query := range []string{"platform=tiktok", "tier=t9"} {
|
||||
status, body := getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events", query, tok)
|
||||
if status != http.StatusBadRequest {
|
||||
t.Fatalf("%s: status = %d, want 400 (body=%s)", query, status, body)
|
||||
}
|
||||
}
|
||||
if len(stub.gotList) != 0 {
|
||||
t.Errorf("store was queried for rejected parameters: %+v", stub.gotList)
|
||||
}
|
||||
|
||||
status, body := getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events", "category=not_a_category", tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("unknown category: status = %d, want 200 exact-match (body=%s)", status, body)
|
||||
}
|
||||
if got := stub.gotList[len(stub.gotList)-1].Category; got != "not_a_category" {
|
||||
t.Errorf("category filter = %q, want the exact value passed through", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminChannelEventsUnwiredReturns500 钉住 §6.5 的诚实边界：留存面未接线时
|
||||
// 两个端点都以 **500 channel_events_unavailable** 作答，绝不回一份全 0 / 空的
|
||||
// 假汇总（面板据此显示错误，而不是一个看着正常的 0）。
|
||||
func TestAdminChannelEventsUnwiredReturns500(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
for _, path := range []string{
|
||||
"/api/v1/admin/channels/events",
|
||||
"/api/v1/admin/channels/events/counters",
|
||||
} {
|
||||
status, body := getChannelEvents(t, ts.URL, path, "", tok)
|
||||
if status != http.StatusInternalServerError {
|
||||
t.Fatalf("%s: status = %d, want 500 (body=%s)", path, status, body)
|
||||
}
|
||||
if code := channelEventErrorCode(t, body); code != "channel_events_unavailable" {
|
||||
t.Errorf("%s: error code = %q, want channel_events_unavailable", path, code)
|
||||
}
|
||||
if strings.Contains(string(body), `"data"`) {
|
||||
t.Errorf("%s: unwired response must not carry a data envelope: %s", path, body)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminChannelEventsReadFailureIsUnavailable 断言读库失败与「未接线」同一种
|
||||
// 诚实：500 + 稳定 code，绝不降级成「200 + 空列表」（那正是把故障画成事实）。
|
||||
func TestAdminChannelEventsReadFailureIsUnavailable(t *testing.T) {
|
||||
t.Parallel()
|
||||
stub := &channelEventStoreStub{listErr: fmt.Errorf("disk on fire")}
|
||||
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{EventStore: stub})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
status, body := getChannelEvents(t, ts.URL, "/api/v1/admin/channels/events", "", tok)
|
||||
if status != http.StatusInternalServerError {
|
||||
t.Fatalf("status = %d, want 500 (body=%s)", status, body)
|
||||
}
|
||||
if code := channelEventErrorCode(t, body); code != "channel_events_unavailable" {
|
||||
t.Errorf("error code = %q, want channel_events_unavailable", code)
|
||||
}
|
||||
// 原始错误（可能带 SQL / 路径）只进日志，绝不出现在响应里。
|
||||
if strings.Contains(string(body), "disk on fire") {
|
||||
t.Errorf("internal error leaked into the response: %s", body)
|
||||
}
|
||||
}
|
||||
@@ -90,17 +90,17 @@ func (s *Server) adminLlmUsage(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, llmUsageResponseOf(summary))
|
||||
}
|
||||
|
||||
// llmUsageQuery 解析并**校验**查询参数：`since`/`until` 见 parseLlmUsageInstant，
|
||||
// llmUsageQuery 解析并**校验**查询参数：`since`/`until` 见 parseInstantParam，
|
||||
// 缺省窗口为近 24h，跨度上限见 model.UsageMaxWindow。
|
||||
//
|
||||
// 校验顺序是「形状先于语义」：时间点解析不了就直接 400，不会拿一个零值窗口去查库。
|
||||
func llmUsageQuery(r *http.Request) (model.UsageQuery, error) {
|
||||
params := r.URL.Query()
|
||||
since, err := parseLlmUsageInstant(params.Get("since"))
|
||||
since, err := parseInstantParam(params.Get("since"))
|
||||
if err != nil {
|
||||
return model.UsageQuery{}, fmt.Errorf("since 非法：%w", err)
|
||||
}
|
||||
until, err := parseLlmUsageInstant(params.Get("until"))
|
||||
until, err := parseInstantParam(params.Get("until"))
|
||||
if err != nil {
|
||||
return model.UsageQuery{}, fmt.Errorf("until 非法：%w", err)
|
||||
}
|
||||
@@ -123,14 +123,18 @@ func llmUsageQuery(r *http.Request) (model.UsageQuery, error) {
|
||||
}, nil
|
||||
}
|
||||
|
||||
// parseLlmUsageInstant 解析一个时间点参数。接受两种形态：
|
||||
// parseInstantParam 解析一个时间点参数。接受两种形态：
|
||||
//
|
||||
// - **RFC3339**（如 `2026-10-05T00:00:00Z`，JS 的 `toISOString()` 就是这个）；
|
||||
// - **Unix 毫秒**（纯数字，如 `1759622400000`）—— 与 `/sessions/{id}/messages`
|
||||
// 的 `before`/`after` 游标同一种口径，便于客户端复用游标。
|
||||
//
|
||||
// 空串 = 未给（返回零值）。其它形态报错（绝不静默当成「现在」）。
|
||||
func parseLlmUsageInstant(raw string) (time.Time, error) {
|
||||
//
|
||||
// 它是**跨端点共用**的解析器：`/admin/llm/usage` 与 `/admin/channels/events`
|
||||
// （切片 E6）的 `since`/`until` 口径相同（同一形态 ⇒ 同一解析 ⇒ 同一比较语义），
|
||||
// 因此只此一份，不各写一遍。
|
||||
func parseInstantParam(raw string) (time.Time, error) {
|
||||
raw = strings.TrimSpace(raw)
|
||||
if raw == "" {
|
||||
return time.Time{}, nil
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/persona/emotion"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader"
|
||||
personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
@@ -261,3 +262,25 @@ type LlmUsageStore interface {
|
||||
// handler 校验并补齐（缺省近 24h、上限 90 天）。
|
||||
Summary(ctx context.Context, q model.UsageQuery) (model.UsageSummary, error)
|
||||
}
|
||||
|
||||
// ChannelEventStore 是渠道事件留存的**只读查询出口**（design/26 §4.5 / §6.5；
|
||||
// 切片 E6）。生产实现是 `internal/runtime/events` 的 `*SQLStore`，组合根把
|
||||
// 同一个句柄既交给桥接层（写入面，E3）也交给网关（本查询面）。
|
||||
//
|
||||
// 它刻意只声明三个**读**方法：写入、计数落盘与保留期清扫是桥接层与组合根的
|
||||
// 职责，管理查询面拿不到、也不该拿到它们（最小权限）。
|
||||
//
|
||||
// 查询结果里**没有任何平台原文 / 媒体 URL / 凭据 / OneBot `flag`** —— 事件表
|
||||
// 本身就没有这些列（§4.2 红线），因此这里不存在「回显敏感字段」的代码路径。
|
||||
//
|
||||
// 为 nil 时两个端点诚实作答 500 `channel_events_unavailable`，绝不回一份全 0
|
||||
// 的假汇总（与 `usage_unavailable` / `storage_unavailable` 同一口径）。
|
||||
type ChannelEventStore interface {
|
||||
// List 按时间倒序返回符合条件的事件行（Limit/Offset 见 events.Filter）。
|
||||
List(ctx context.Context, f events.Filter) ([]events.Event, error)
|
||||
// Count 统计符合条件的事件行数（List 的分页配套）。
|
||||
Count(ctx context.Context, f events.Filter) (int64, error)
|
||||
// Counters 返回逐类计数汇总；platform 为空表示全部平台。它读的是**已落盘**
|
||||
// 的表，不含内存里待 flush 的部分（即 §4.3 的「近似值」）。
|
||||
Counters(ctx context.Context, platform string) ([]events.Counter, error)
|
||||
}
|
||||
@@ -142,6 +142,12 @@ type Deps struct {
|
||||
// GET /api/v1/admin/llm/usage 诚实作答 500 usage_unavailable，
|
||||
// 绝不回一份全 0 的假汇总。
|
||||
Usage LlmUsageStore
|
||||
// EventStore 是渠道事件留存的只读查询出口（design/26 §4.5；切片 E6）。
|
||||
// 组合根注入的正是交给桥接层写入的**同一个** `*events.SQLStore` 句柄
|
||||
// （同一个库、同一份口径）。可选；为 nil 时
|
||||
// GET /api/v1/admin/channels/events 与 `.../counters` 诚实作答
|
||||
// 500 channel_events_unavailable，绝不回一份全 0 的假汇总。
|
||||
EventStore ChannelEventStore
|
||||
// MaxParallelTurns 是单会话在途并行路数的上限（design/15 §2.5.7）。
|
||||
// 零值/负数 ⇒ 编排器用默认 2；1 ⇒ 关闭真并行（恒 1 路，完全回落到
|
||||
// queue + 回合内 steer 的现状）；超过硬上限 3 ⇒ 钳到 3。
|
||||
|
||||
@@ -44,6 +44,8 @@ var ownerAdminReadRoutes = []rbacRoute{
|
||||
{"memories", http.MethodGet, "/api/v1/admin/memories"},
|
||||
{"channels/messages", http.MethodGet, "/api/v1/admin/channels/messages"},
|
||||
{"channels/{platform}/messages", http.MethodGet, "/api/v1/admin/channels/kook/messages"},
|
||||
{"channels/events", http.MethodGet, "/api/v1/admin/channels/events"},
|
||||
{"channels/events/counters", http.MethodGet, "/api/v1/admin/channels/events/counters"},
|
||||
{"llm", http.MethodGet, "/api/v1/admin/llm"},
|
||||
{"users", http.MethodGet, "/api/v1/admin/users"},
|
||||
{"backthink", http.MethodGet, "/api/v1/admin/backthink"},
|
||||
|
||||
@@ -167,6 +167,12 @@ func (s *Server) registerAdminRoutes(
|
||||
mux.Handle("GET /api/v1/admin/memories", writeAuth(http.HandlerFunc(s.adminListMemories)))
|
||||
mux.Handle("GET /api/v1/admin/channels/messages", writeAuth(http.HandlerFunc(s.handleChannelMessages)))
|
||||
mux.Handle("GET /api/v1/admin/channels/{platform}/messages", writeAuth(http.HandlerFunc(s.handleChannelPlatformLog)))
|
||||
// 渠道事件查询面（design/26 §4.5；切片 E6，契约 **v0.67 已登记**）：非消息事件的
|
||||
// 只读投影 + 逐类计数（近似值）。与同族的 `/admin/channels/messages` 同门
|
||||
// （owner|admin）：事件行带群 id / 成员 id / 关联消息 id，属私有内容面，
|
||||
// 不是给 viewer 的监控投影（契约 §7.4-12 登记「落地前不要假设 viewer 能读」）。
|
||||
mux.Handle("GET /api/v1/admin/channels/events", writeAuth(http.HandlerFunc(s.adminChannelEventsList)))
|
||||
mux.Handle("GET /api/v1/admin/channels/events/counters", writeAuth(http.HandlerFunc(s.adminChannelEventsCounters)))
|
||||
mux.Handle("GET /api/v1/admin/llm", writeAuth(http.HandlerFunc(s.adminGetLlm)))
|
||||
// LLM 用量汇总（切片 U1，**v0.64**）：只读、不审计，与同页的 /admin/llm
|
||||
// 同门（owner|admin）—— 它含成本与模型/用途分解，不是给 viewer 的监控投影。
|
||||
|
||||
@@ -86,6 +86,10 @@ type Server struct {
|
||||
// usage 是 LLM 用量汇总的只读出口（切片 U1，见 Deps.Usage）。为 nil 时
|
||||
// GET /api/v1/admin/llm/usage 回 500 usage_unavailable。
|
||||
usage LlmUsageStore
|
||||
// channelEvents 是渠道事件留存的只读查询出口（切片 E6，见 Deps.EventStore）。
|
||||
// 为 nil 时 GET /api/v1/admin/channels/events（及其 /counters）回
|
||||
// 500 channel_events_unavailable —— 未接线就说未接线，不回全 0 假汇总。
|
||||
channelEvents ChannelEventStore
|
||||
// db 是组合根注入的活 SQLite 连接池（见 Deps.Database）。管理侧
|
||||
// 「存储状态」只读端点用它读页级统计与 documents 表；为 nil 时该
|
||||
// 端点回 500 storage_unavailable（不另开池、不假装有数据）。
|
||||
@@ -141,6 +145,7 @@ func New(cfg *config.Config, deps Deps) (*Server, error) {
|
||||
ownerRegistrar: deps.OwnerRegistrar,
|
||||
webhooks: deps.Webhooks,
|
||||
usage: deps.Usage,
|
||||
channelEvents: deps.EventStore,
|
||||
dbPath: deps.DatabasePath,
|
||||
dataDir: deps.DataDir,
|
||||
db: deps.Database,
|
||||
|
||||
Reference in new issue
Block a user