AskaEth
c8053b1f36
- 两个只读端点：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
372 lines
14 KiB
Go
372 lines
14 KiB
Go
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)
|
||
}
|
||
}
|