AskaEth
ec3f6e810a
- 新包 internal/runtime/events：Tier/Category/Platform/ChannelType/ActorRole 具名常量 + Event/Filter/Counter + Store 六方法 - 留存口径：T2/T3 逐条落行；T1 只计数；unknown / decode_error / link 三例外逐条落行（链路是否活着是事故定案唯一证据） - 计数：内存聚合 + 上限 512 的有界缓冲 + 满即整批 upsert，落盘失败回填而不丢；GC：保留期 720h/间隔 1h，≤0 = 关闭 - 迁移 0021_channel_events（两表 + 6 索引，含计数 upsert 唯一键）；Event 无 flag/正文/URL 字段（design/26 §4.2 红线）
448 lines
14 KiB
Go
448 lines
14 KiB
Go
package events
|
||
|
||
// 本文件是**事件存储写入与查询**的验收（design/26 §4.1/§4.3/§4.5；切片 E1）。
|
||
// 断言的是**库里的事实**（行在不在、字段读回来是不是原值、计数是不是累计值），
|
||
// 不按日志文本断言 —— 与 `internal/runtime/documents` 的测试同一口径。
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
|
||
)
|
||
|
||
// testNow 是夹具的固定时钟（可复现：所有时间断言都基于它）。
|
||
func testNow() time.Time {
|
||
return time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
|
||
}
|
||
|
||
// newTestStore 建一套隔离夹具：一个跑过迁移的内存库 + 一个把时钟钉在 testNow 的
|
||
// 存储。默认 **关闭 GC**（事件存储的用例与保留期无关，GC 由 gc_test.go 专门覆盖），
|
||
// tune 可覆盖任何配置项。第二个返回值是底层连接，供用例做结构级断言。
|
||
func newTestStore(t *testing.T, tune func(*Config)) (*SQLStore, *sql.DB) {
|
||
t.Helper()
|
||
ctx := context.Background()
|
||
conn, err := db.Open(ctx, ":memory:", nil)
|
||
if err != nil {
|
||
t.Fatalf("打开内存库: %v", err)
|
||
}
|
||
if err := db.MigrateUp(ctx, conn, nil); err != nil {
|
||
_ = conn.Close()
|
||
t.Fatalf("内存库跑迁移: %v", err)
|
||
}
|
||
t.Cleanup(func() { _ = conn.Close() })
|
||
|
||
cfg := Config{DB: conn, Now: testNow, GCInterval: 0, Retention: 0}
|
||
if tune != nil {
|
||
tune(&cfg)
|
||
}
|
||
st, err := NewStore(cfg)
|
||
if err != nil {
|
||
t.Fatalf("NewStore: %v", err)
|
||
}
|
||
return st, conn
|
||
}
|
||
|
||
// sampleT2 是一条带全部字段的 T2 事件（用于往返断言）。
|
||
func sampleT2() Event {
|
||
return Event{
|
||
Platform: PlatformQQ,
|
||
AccountID: "bot-1",
|
||
Tier: TierT2,
|
||
Category: CategoryRecall,
|
||
PlatformType: "group_recall",
|
||
ChannelID: "group-1070898847",
|
||
ChannelType: ChannelGroup,
|
||
GroupID: "1070898847",
|
||
ActorID: "10001",
|
||
ActorRole: RoleMember,
|
||
TargetID: "10002",
|
||
SelfRelated: true,
|
||
MessageID: "msg-9",
|
||
Summary: "群消息撤回",
|
||
CreatedAt: testNow().Add(-time.Minute),
|
||
}
|
||
}
|
||
|
||
// TestNewStore_NilDB_异常：没有数据库就没有存储 —— 宁可明确失败，也不给一个
|
||
// 写不进去的替身（§4.3 的「store 为 nil 的降级」由桥接层负责，不在存储层伪造）。
|
||
func TestNewStore_NilDB_异常(t *testing.T) {
|
||
t.Parallel()
|
||
st, err := NewStore(Config{})
|
||
if !errors.Is(err, ErrNilDB) {
|
||
t.Fatalf("NewStore(nil db) err = %v, want wrap of ErrNilDB", err)
|
||
}
|
||
if st != nil {
|
||
t.Fatalf("NewStore(nil db) returned %#v, want nil", st)
|
||
}
|
||
}
|
||
|
||
// TestStore_InterfaceUsable_正常：E3/E4 只依赖 `Store` 接口，因此这里**经接口**
|
||
// 走一遍全部六个方法（否则「实现满足接口」这件事只在编译期被偶然发现）。
|
||
func TestStore_InterfaceUsable_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, func(cfg *Config) {
|
||
cfg.GCInterval = DefaultGCInterval
|
||
cfg.Retention = DefaultRetention
|
||
})
|
||
var store Store = st
|
||
|
||
if err := store.Record(ctx, sampleT2()); err != nil {
|
||
t.Fatalf("Record: %v", err)
|
||
}
|
||
if err := store.Flush(ctx); err != nil {
|
||
t.Fatalf("Flush: %v", err)
|
||
}
|
||
n, err := store.Count(ctx, Filter{Platform: PlatformQQ})
|
||
if err != nil {
|
||
t.Fatalf("Count: %v", err)
|
||
}
|
||
if n != 1 {
|
||
t.Fatalf("Count = %d, want 1", n)
|
||
}
|
||
events, err := store.List(ctx, Filter{})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(events) != 1 {
|
||
t.Fatalf("List = %d rows, want 1", len(events))
|
||
}
|
||
counters, err := store.Counters(ctx, PlatformQQ)
|
||
if err != nil {
|
||
t.Fatalf("Counters: %v", err)
|
||
}
|
||
if len(counters) != 1 {
|
||
t.Fatalf("Counters = %d rows, want 1", len(counters))
|
||
}
|
||
if _, err := store.Sweep(ctx); err != nil {
|
||
t.Fatalf("Sweep: %v", err)
|
||
}
|
||
}
|
||
|
||
// TestRecordT2_WriteAndList_RoundTrip_正常：T2 逐条落行，且**每个字段原样读回**
|
||
// （字段错位是这类写入面最危险的静默 bug）。
|
||
func TestRecordT2_WriteAndList_RoundTrip_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, nil)
|
||
|
||
want := sampleT2()
|
||
if err := st.Record(ctx, want); err != nil {
|
||
t.Fatalf("Record: %v", err)
|
||
}
|
||
|
||
got, err := st.List(ctx, Filter{})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(got) != 1 {
|
||
t.Fatalf("List = %d rows, want 1", len(got))
|
||
}
|
||
ev := got[0]
|
||
if ev.ID == 0 {
|
||
t.Fatal("ID should be filled by the database")
|
||
}
|
||
checks := []struct {
|
||
name string
|
||
got, want string
|
||
}{
|
||
{"platform", ev.Platform, want.Platform},
|
||
{"account_id", ev.AccountID, want.AccountID},
|
||
{"tier", string(ev.Tier), string(want.Tier)},
|
||
{"category", string(ev.Category), string(want.Category)},
|
||
{"platform_type", ev.PlatformType, want.PlatformType},
|
||
{"channel_id", ev.ChannelID, want.ChannelID},
|
||
{"channel_type", ev.ChannelType, want.ChannelType},
|
||
{"group_id", ev.GroupID, want.GroupID},
|
||
{"actor_id", ev.ActorID, want.ActorID},
|
||
{"actor_role", ev.ActorRole, want.ActorRole},
|
||
{"target_id", ev.TargetID, want.TargetID},
|
||
{"message_id", ev.MessageID, want.MessageID},
|
||
{"summary", ev.Summary, want.Summary},
|
||
}
|
||
for _, c := range checks {
|
||
if c.got != c.want {
|
||
t.Errorf("%s = %q, want %q", c.name, c.got, c.want)
|
||
}
|
||
}
|
||
if !ev.SelfRelated {
|
||
t.Error("self_related = false, want true")
|
||
}
|
||
if !ev.CreatedAt.Equal(want.CreatedAt) {
|
||
t.Errorf("created_at = %v, want %v", ev.CreatedAt, want.CreatedAt)
|
||
}
|
||
if !ev.UpdatedAt.Equal(want.CreatedAt) {
|
||
t.Errorf("updated_at = %v, want %v (events are immutable)", ev.UpdatedAt, want.CreatedAt)
|
||
}
|
||
}
|
||
|
||
// TestRecord_ZeroClock_FillsNow_正常：CreatedAt 零值时由存储填当前时刻（调用方
|
||
// 不必自己取时钟），且同一帧仍进计数。
|
||
func TestRecord_ZeroClock_FillsNow_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, nil)
|
||
|
||
ev := sampleT2()
|
||
ev.CreatedAt = time.Time{}
|
||
if err := st.Record(ctx, ev); err != nil {
|
||
t.Fatalf("Record: %v", err)
|
||
}
|
||
got, err := st.List(ctx, Filter{})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(got) != 1 || !got[0].CreatedAt.Equal(testNow()) {
|
||
t.Fatalf("created_at = %v, want the injected clock %v", got[0].CreatedAt, testNow())
|
||
}
|
||
}
|
||
|
||
// TestRecordT1_CountsWithoutRow_正常：T1 是**只计数**（design/26 §4.3）—— 行表里
|
||
// 没有它，计数表里有它。这条用例钉住「心跳不把库写胖」这个设计取舍。
|
||
func TestRecordT1_CountsWithoutRow_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, nil)
|
||
|
||
for i := 0; i < 3; i++ {
|
||
if err := st.Record(ctx, Event{
|
||
Platform: PlatformQQ,
|
||
Tier: TierT1,
|
||
Category: CategoryHeartbeat,
|
||
PlatformType: "heartbeat",
|
||
Summary: "链路心跳",
|
||
}); err != nil {
|
||
t.Fatalf("Record T1 (#%d): %v", i, err)
|
||
}
|
||
}
|
||
if n, err := st.Count(ctx, Filter{}); err != nil {
|
||
t.Fatalf("Count: %v", err)
|
||
} else if n != 0 {
|
||
t.Fatalf("T1 wrote %d rows into channel_events, want 0", n)
|
||
}
|
||
if err := st.Flush(ctx); err != nil {
|
||
t.Fatalf("Flush: %v", err)
|
||
}
|
||
counters, err := st.Counters(ctx, PlatformQQ)
|
||
if err != nil {
|
||
t.Fatalf("Counters: %v", err)
|
||
}
|
||
if len(counters) != 1 {
|
||
t.Fatalf("Counters = %d rows, want 1", len(counters))
|
||
}
|
||
got := counters[0]
|
||
if got.Count != 3 {
|
||
t.Fatalf("count = %d, want 3", got.Count)
|
||
}
|
||
if got.Tier != TierT1 || got.Category != CategoryHeartbeat {
|
||
t.Fatalf("tier/category = (%q, %q), want (t1, heartbeat)", got.Tier, got.Category)
|
||
}
|
||
if !got.FirstSeenAt.Equal(testNow()) || !got.LastSeenAt.Equal(testNow()) {
|
||
t.Fatalf("first/last seen = (%v, %v), want %v", got.FirstSeenAt, got.LastSeenAt, testNow())
|
||
}
|
||
}
|
||
|
||
// TestRecordT1_FallbackCategoriesPersist_边界：T1 里三个**例外**类别仍要逐条落行
|
||
// （design/26 §3.7 的 unknown/decode_error「留痕 + 计数」，§3.6 的链路事件
|
||
// 「落 channel_events(category=link)」）。这条用例是那三处例外在代码里的证据。
|
||
func TestRecordT1_FallbackCategoriesPersist_边界(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, nil)
|
||
|
||
for _, category := range []Category{CategoryUnknown, CategoryDecodeError, CategoryLink} {
|
||
if err := st.Record(ctx, Event{
|
||
Platform: PlatformQQ,
|
||
Tier: TierT1,
|
||
Category: category,
|
||
PlatformType: string(category),
|
||
}); err != nil {
|
||
t.Fatalf("Record (%s): %v", category, err)
|
||
}
|
||
}
|
||
n, err := st.Count(ctx, Filter{})
|
||
if err != nil {
|
||
t.Fatalf("Count: %v", err)
|
||
}
|
||
if n != 3 {
|
||
t.Fatalf("rows = %d, want 3 (unknown / decode_error / link 必须留痕)", n)
|
||
}
|
||
|
||
// 对照：同样 T1、但类别不在例外里的帧**不**落行。
|
||
if err := st.Record(ctx, Event{
|
||
Platform: PlatformQQ, Tier: TierT1, Category: CategoryHeartbeat, PlatformType: "heartbeat",
|
||
}); err != nil {
|
||
t.Fatalf("Record (heartbeat): %v", err)
|
||
}
|
||
if n, err := st.Count(ctx, Filter{}); err != nil {
|
||
t.Fatalf("Count: %v", err)
|
||
} else if n != 3 {
|
||
t.Fatalf("rows = %d, want 3 (heartbeat 仍应只计数)", n)
|
||
}
|
||
}
|
||
|
||
// TestRecord_NoPlatform_异常：缺平台是调用方的编程错误，必须显式失败且**不写任何行**
|
||
// （写一行 platform=” 的行只会让查询面多出一个查不到的垃圾桶）。
|
||
func TestRecord_NoPlatform_异常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, nil)
|
||
|
||
ev := sampleT2()
|
||
ev.Platform = " "
|
||
if err := st.Record(ctx, ev); !errors.Is(err, ErrNoPlatform) {
|
||
t.Fatalf("Record(no platform) err = %v, want wrap of ErrNoPlatform", err)
|
||
}
|
||
if n, err := st.Count(ctx, Filter{}); err != nil {
|
||
t.Fatalf("Count: %v", err)
|
||
} else if n != 0 {
|
||
t.Fatalf("rows = %d, want 0", n)
|
||
}
|
||
}
|
||
|
||
// TestRecord_InvalidTier_FallsBackToT1_边界：非法档**降级成 T1**（fail-safe）：宁可
|
||
// 只计数，也不让一个字段写错把整帧丢出留存面。T2 类别 + 非法档 ⇒ 不落行、但计数在。
|
||
func TestRecord_InvalidTier_FallsBackToT1_边界(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, nil)
|
||
|
||
ev := sampleT2()
|
||
ev.Tier = Tier("t9")
|
||
if err := st.Record(ctx, ev); err != nil {
|
||
t.Fatalf("Record: %v", err)
|
||
}
|
||
if n, err := st.Count(ctx, Filter{}); err != nil {
|
||
t.Fatalf("Count: %v", err)
|
||
} else if n != 0 {
|
||
t.Fatalf("rows = %d, want 0 (非法档按 T1 处理 ⇒ 只计数)", n)
|
||
}
|
||
if st.PendingKeys() != 1 {
|
||
t.Fatalf("PendingKeys = %d, want 1 (降级后的帧仍必须计数)", st.PendingKeys())
|
||
}
|
||
}
|
||
|
||
// TestList_FilterAndOrder_正常：筛选（平台 / 档 / 类别 / 会话 / 时间窗）与倒序排序。
|
||
// 它们是查询面（E6）与「事后定案」直接依赖的行为。
|
||
func TestList_FilterAndOrder_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, nil)
|
||
|
||
base := testNow()
|
||
seeds := []Event{
|
||
{Platform: PlatformQQ, Tier: TierT2, Category: CategoryRecall, PlatformType: "group_recall",
|
||
ChannelID: "group-a", CreatedAt: base.Add(-3 * time.Minute), Summary: "群消息撤回"},
|
||
{Platform: PlatformQQ, Tier: TierT3, Category: CategoryPoke, PlatformType: "notify.poke",
|
||
ChannelID: "group-a", CreatedAt: base.Add(-2 * time.Minute), Summary: "戳一戳"},
|
||
{Platform: PlatformKook, Tier: TierT2, Category: CategoryUnknown, PlatformType: "weird",
|
||
ChannelID: "channel-b", CreatedAt: base.Add(-time.Minute), Summary: "未知事件"},
|
||
}
|
||
for i, ev := range seeds {
|
||
if err := st.Record(ctx, ev); err != nil {
|
||
t.Fatalf("Record(#%d): %v", i, err)
|
||
}
|
||
}
|
||
|
||
cases := []struct {
|
||
name string
|
||
filter Filter
|
||
want int
|
||
}{
|
||
{"无筛选", Filter{}, 3},
|
||
{"按平台", Filter{Platform: PlatformQQ}, 2},
|
||
{"按平台(无匹配)", Filter{Platform: PlatformTelegram}, 0},
|
||
{"按档", Filter{Tier: TierT2}, 2},
|
||
{"按类别", Filter{Category: CategoryPoke}, 1},
|
||
{"按会话", Filter{ChannelID: "group-a"}, 2},
|
||
{"时间窗(闭区间单点)", Filter{Since: base.Add(-2 * time.Minute), Until: base.Add(-2 * time.Minute)}, 1},
|
||
{"时间窗(早于下界的不返回)", Filter{Since: base.Add(-2*time.Minute - time.Second)}, 2},
|
||
{"多条件合取", Filter{Platform: PlatformQQ, Tier: TierT3}, 1},
|
||
}
|
||
for _, c := range cases {
|
||
t.Run(c.name, func(t *testing.T) {
|
||
t.Parallel()
|
||
rows, err := st.List(ctx, c.filter)
|
||
if err != nil {
|
||
t.Fatalf("List(%+v): %v", c.filter, err)
|
||
}
|
||
if len(rows) != c.want {
|
||
t.Fatalf("List(%+v) = %d rows, want %d", c.filter, len(rows), c.want)
|
||
}
|
||
n, err := st.Count(ctx, c.filter)
|
||
if err != nil {
|
||
t.Fatalf("Count(%+v): %v", c.filter, err)
|
||
}
|
||
if int(n) != c.want {
|
||
t.Fatalf("Count(%+v) = %d, want %d", c.filter, n, c.want)
|
||
}
|
||
})
|
||
}
|
||
|
||
// 倒序：最新的在前（design/26 §4.5 的「事件列表（倒序）」）。
|
||
rows, err := st.List(ctx, Filter{})
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
for i := 1; i < len(rows); i++ {
|
||
if rows[i-1].CreatedAt.Before(rows[i].CreatedAt) {
|
||
t.Fatalf("rows are not in descending time order: %v then %v", rows[i-1].CreatedAt, rows[i].CreatedAt)
|
||
}
|
||
}
|
||
if rows[0].Platform != PlatformKook {
|
||
t.Fatalf("first row platform = %q, want the newest (%q)", rows[0].Platform, PlatformKook)
|
||
}
|
||
}
|
||
|
||
// TestList_Paging_边界：Limit/Offset 生效，且 Limit 被钳在硬上界内、超过范围的
|
||
// Offset 返回空而不是报错（分页尾页是正常请求）。
|
||
func TestList_Paging_边界(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, nil)
|
||
|
||
base := testNow()
|
||
for i := 0; i < 5; i++ {
|
||
if err := st.Record(ctx, Event{
|
||
Platform: PlatformQQ, Tier: TierT2, Category: CategoryJoin, PlatformType: "group_increase",
|
||
CreatedAt: base.Add(time.Duration(i) * time.Minute),
|
||
}); err != nil {
|
||
t.Fatalf("Record(#%d): %v", i, err)
|
||
}
|
||
}
|
||
|
||
first, err := st.List(ctx, Filter{Limit: 2})
|
||
if err != nil {
|
||
t.Fatalf("List(limit 2): %v", err)
|
||
}
|
||
second, err := st.List(ctx, Filter{Limit: 2, Offset: 2})
|
||
if err != nil {
|
||
t.Fatalf("List(limit 2 offset 2): %v", err)
|
||
}
|
||
if len(first) != 2 || len(second) != 2 {
|
||
t.Fatalf("page sizes = (%d, %d), want (2, 2)", len(first), len(second))
|
||
}
|
||
if !first[0].CreatedAt.After(second[0].CreatedAt) {
|
||
t.Fatal("the second page should contain older rows than the first")
|
||
}
|
||
if got, err := st.List(ctx, Filter{Limit: -1}); err != nil {
|
||
t.Fatalf("List(limit -1): %v", err)
|
||
} else if len(got) != 5 {
|
||
t.Fatalf("List(limit -1) = %d rows, want all 5 (≤0 取缺省)", len(got))
|
||
}
|
||
if got, err := st.List(ctx, Filter{Limit: MaxListLimit * 4}); err != nil {
|
||
t.Fatalf("List(huge limit): %v", err)
|
||
} else if len(got) != 5 {
|
||
t.Fatalf("List(huge limit) = %d rows, want 5 (超上界即钳住，但不是报错)", len(got))
|
||
}
|
||
if got, err := st.List(ctx, Filter{Offset: 99}); err != nil {
|
||
t.Fatalf("List(offset 99): %v", err)
|
||
} else if len(got) != 0 {
|
||
t.Fatalf("List(offset 99) = %d rows, want 0", len(got))
|
||
}
|
||
}
|