From ec3f6e810a3326fe89f09c5e7ab69578b81c3de2 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 5 Oct 2026 13:47:25 +0800 Subject: [PATCH] =?UTF-8?q?feat(events):=20=E6=B8=A0=E9=81=93=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=AD=98=E5=82=A8=E5=9C=B0=E5=9F=BA=EF=BC=88Event/Sto?= =?UTF-8?q?re=20+=20SQLite=20=E5=AE=9E=E7=8E=B0=20+=20=E6=9C=89=E7=95=8C?= =?UTF-8?q?=E8=AE=A1=E6=95=B0=E8=81=9A=E5=90=88=20+=20=E4=BF=9D=E7=95=99?= =?UTF-8?q?=E6=9C=9F=20GC=20+=20=E8=BF=81=E7=A7=BB=200021=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新包 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 红线） --- internal/db/migrate_channel_events_test.go | 175 +++++++ .../migrations/0021_channel_events.down.sql | 10 + .../db/migrations/0021_channel_events.up.sql | 71 +++ internal/runtime/events/counters.go | 261 ++++++++++ internal/runtime/events/counters_test.go | 210 ++++++++ internal/runtime/events/events.go | 264 +++++++++++ internal/runtime/events/gc.go | 121 +++++ internal/runtime/events/gc_test.go | 209 ++++++++ internal/runtime/events/store.go | 404 ++++++++++++++++ internal/runtime/events/store_test.go | 447 ++++++++++++++++++ 10 files changed, 2172 insertions(+) create mode 100644 internal/db/migrate_channel_events_test.go create mode 100644 internal/db/migrations/0021_channel_events.down.sql create mode 100644 internal/db/migrations/0021_channel_events.up.sql create mode 100644 internal/runtime/events/counters.go create mode 100644 internal/runtime/events/counters_test.go create mode 100644 internal/runtime/events/events.go create mode 100644 internal/runtime/events/gc.go create mode 100644 internal/runtime/events/gc_test.go create mode 100644 internal/runtime/events/store.go create mode 100644 internal/runtime/events/store_test.go diff --git a/internal/db/migrate_channel_events_test.go b/internal/db/migrate_channel_events_test.go new file mode 100644 index 0000000..2e2064e --- /dev/null +++ b/internal/db/migrate_channel_events_test.go @@ -0,0 +1,175 @@ +package db + +import ( + "context" + "database/sql" + "testing" +) + +// channelEventsMigrationID 是渠道事件留存两张表的迁移 id（design/26 §4.1，切片 E1）。 +const channelEventsMigrationID = "0021_channel_events" + +// schemaObjectExists 报告 sqlite_master 里是否存在某类对象（table / index）。 +// 与 columnExists 同族，都只读元数据、不碰 data/cyrene.db。 +func schemaObjectExists(t *testing.T, ctx context.Context, conn *sql.DB, kind, name string) bool { + t.Helper() + var n int + if err := conn.QueryRowContext(ctx, + "SELECT count(*) FROM sqlite_master WHERE type = ? AND name = ?", kind, name).Scan(&n); err != nil { + t.Fatalf("sqlite_master(%s %s): %v", kind, name, err) + } + return n == 1 +} + +// TestMigrateChannelEventsTablesReversible 钉住 0021（channel_events + +// channel_event_counters，design/26 §4.1）：up 建两张表与全部索引、重复 up 是空操作、 +// down 整表消失、再 up 恢复端到端可用。它用自己的内存库，绝不碰 data/cyrene.db。 +// +// 断言的是**库里的结构事实**（表/列/索引在不在、默认值读出来是什么），不按日志文本 +// 断言 —— 与 migrate_turn_id_test.go / migrate_document_test.go 同一口径。 +func TestMigrateChannelEventsTablesReversible(t *testing.T) { + t.Parallel() + ctx := context.Background() + conn := openMemory(t) + + if err := MigrateUp(ctx, conn, testLogger()); err != nil { + t.Fatalf("MigrateUp: %v", err) + } + + // 两张表与它们的列必须逐个到位（列名错一个，写入面就会静默写错地方）。 + columns := map[string][]string{ + "channel_events": { + "id", "platform", "account_id", "tier", "category", "platform_type", + "channel_id", "channel_type", "group_id", "actor_id", "actor_role", + "target_id", "self_related", "message_id", "summary", "created_at", "updated_at", + }, + "channel_event_counters": { + "id", "platform", "account_id", "tier", "platform_type", "category", + "count", "first_seen_at", "last_seen_at", + }, + } + for table, cols := range columns { + if !schemaObjectExists(t, ctx, conn, "table", table) { + t.Fatalf("%s should exist after MigrateUp", table) + } + for _, col := range cols { + if !columnExists(t, ctx, conn, table, col) { + t.Fatalf("%s.%s should exist after MigrateUp", table, col) + } + } + } + + // 索引：查询面与 GC 依赖 idx_channel_events_created_at，upsert 依赖计数表的唯一键。 + for _, idx := range []string{ + "idx_channel_events_created_at", + "idx_channel_events_platform_category", + "idx_channel_events_channel_id", + "idx_channel_events_tier", + "idx_channel_event_counters_key", + "idx_channel_event_counters_category", + } { + if !schemaObjectExists(t, ctx, conn, "index", idx) { + t.Fatalf("%s should exist after MigrateUp", idx) + } + } + + // 默认值：只写必填列也能落行，可选列读回来是 ''/0，时间戳由 SQLite 默认填。 + if _, err := conn.ExecContext(ctx, + `INSERT INTO channel_events (platform) VALUES (?)`, "qq-obv11"); err != nil { + t.Fatalf("insert minimal channel event: %v", err) + } + var ( + tier, category, summary, createdAt string + selfRelated int + ) + if err := conn.QueryRowContext(ctx, + `SELECT tier, category, summary, self_related, created_at FROM channel_events WHERE platform = ?`, + "qq-obv11").Scan(&tier, &category, &summary, &selfRelated, &createdAt); err != nil { + t.Fatalf("read minimal channel event: %v", err) + } + if tier != "" || category != "" || summary != "" || selfRelated != 0 { + t.Fatalf("defaults = (%q, %q, %q, %d), want ('', '', '', 0)", tier, category, summary, selfRelated) + } + if createdAt == "" { + t.Fatal("created_at should be filled by the migration default") + } + + // 重复 up：已记录的迁移被跳过，结构仍在且不报错（幂等）。 + if err := MigrateUp(ctx, conn, testLogger()); err != nil { + t.Fatalf("MigrateUp second call: %v", err) + } + if !schemaObjectExists(t, ctx, conn, "table", "channel_events") { + t.Fatal("channel_events should still exist after idempotent re-run") + } + + // 恰好回滚到 0021（含它之后新加的迁移），日后再加迁移也不会让本测试失效。 + ids := upMigrationIDs(t) + idx := -1 + for i, id := range ids { + if id == channelEventsMigrationID { + idx = i + } + } + if idx < 0 { + t.Fatalf("%s not found in %v", channelEventsMigrationID, ids) + } + if err := MigrateDown(ctx, conn, len(ids)-idx, testLogger()); err != nil { + t.Fatalf("MigrateDown(%d): %v", len(ids)-idx, err) + } + for _, table := range []string{"channel_events", "channel_event_counters"} { + if schemaObjectExists(t, ctx, conn, "table", table) { + t.Fatalf("%s should be dropped by the 0021 down", table) + } + } + + // 重新 up 恢复端到端可用。 + if err := MigrateUp(ctx, conn, testLogger()); err != nil { + t.Fatalf("re-MigrateUp: %v", err) + } + for _, table := range []string{"channel_events", "channel_event_counters"} { + if !schemaObjectExists(t, ctx, conn, "table", table) { + t.Fatalf("%s should exist again after re-MigrateUp", table) + } + } +} + +// TestMigrate0021CountersUniqueKey_正常 钉住计数表唯一键的**行为**（不只是「索引 +// 存在」）：同一 `(platform, account_id, platform_type)` 二次 upsert 必须累加成 +// **一行**，否则「上游推过几次」的答案会分裂成多行、计数上界也随之失控 +// （design/26 §4.1 表二的全部理由）。 +func TestMigrate0021CountersUniqueKey_正常(t *testing.T) { + t.Parallel() + ctx := context.Background() + conn := openMemory(t) + + if err := MigrateUp(ctx, conn, testLogger()); err != nil { + t.Fatalf("MigrateUp: %v", err) + } + + const upsert = `INSERT INTO channel_event_counters + (platform, account_id, tier, platform_type, category, count, first_seen_at, last_seen_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(platform, account_id, platform_type) DO UPDATE SET + count = count + excluded.count, + last_seen_at = excluded.last_seen_at` + for _, count := range []int{1, 2} { + if _, err := conn.ExecContext(ctx, upsert, + "qq-obv11", "bot1", "t1", "heartbeat", "heartbeat", count, + "2026-10-05T00:00:00.000Z", "2026-10-05T00:00:00.000Z"); err != nil { + t.Fatalf("upsert counter (%d): %v", count, err) + } + } + + var rows, total int + if err := conn.QueryRowContext(ctx, + `SELECT count(*), COALESCE(sum(count), 0) FROM channel_event_counters WHERE platform = ?`, + "qq-obv11").Scan(&rows, &total); err != nil { + t.Fatalf("read counters: %v", err) + } + if rows != 1 { + t.Fatalf("counter rows = %d, want 1 (the unique key must merge, not duplicate)", rows) + } + if total != 3 { + t.Fatalf("counter total = %d, want 3 (1 + 2)", total) + } +} diff --git a/internal/db/migrations/0021_channel_events.down.sql b/internal/db/migrations/0021_channel_events.down.sql new file mode 100644 index 0000000..d8d08c4 --- /dev/null +++ b/internal/db/migrations/0021_channel_events.down.sql @@ -0,0 +1,10 @@ +-- 0021 的回滚：先删索引、再删表（与 0019 的回滚同一顺序口径）。 +-- 两张表都是本迁移新建的，回滚不触碰任何既有表。 +DROP INDEX IF EXISTS idx_channel_event_counters_category; +DROP INDEX IF EXISTS idx_channel_event_counters_key; +DROP TABLE IF EXISTS channel_event_counters; +DROP INDEX IF EXISTS idx_channel_events_tier; +DROP INDEX IF EXISTS idx_channel_events_channel_id; +DROP INDEX IF EXISTS idx_channel_events_platform_category; +DROP INDEX IF EXISTS idx_channel_events_created_at; +DROP TABLE IF EXISTS channel_events; diff --git a/internal/db/migrations/0021_channel_events.up.sql b/internal/db/migrations/0021_channel_events.up.sql new file mode 100644 index 0000000..43e4a24 --- /dev/null +++ b/internal/db/migrations/0021_channel_events.up.sql @@ -0,0 +1,71 @@ +-- 0021_channel_events: 渠道非消息事件的**留存地基**（design/26 §4.1；切片 E1）。 +-- +-- 动因（design/26 §0.2，已确证事故）：qq_obv11 的入站总闸曾把 post_type != message +-- 的帧零日志、零计数地丢掉，于是「上游推过没有 / 是不是被我们过滤了」在数据里 +-- 再也答不上来。本表就是那块地基：让到达适配器的每一帧至少留下可查的结构级痕迹。 +-- +-- 两张表**刻意不合并**（design/26 §4.1 已论证）： +-- ① 写放大 —— 心跳 / client_status / 在线状态是高频遥测，逐帧一行会把库写胖； +-- 计数表按「平台+账号+平台原生类型」upsert，行数上界是「平台数 × 事件类型数」。 +-- ② 保留期不同 —— channel_events 有审计价值、按保留期清扫； +-- 计数表本身就是「历史已浓缩成一行」的形态，**删了就再也答不上「推过没有」**， +-- 因此它保留 first_seen_at / last_seen_at，**不参与 GC**。 +-- ③ 两张表不同构（一张逐条、一张聚合），不是「同一份东西抄了两遍」。 +-- T3 与 T2 的列**完全相同**（只差 tier 取值），故合在同一张表里：拆开会让「单一 +-- 查询面 / 单一保留期 / 单一 GC」变成三份实现（design/26 §4.1）。 +-- +-- 命名口径：表名复数、主键 id、时间戳 created_at/updated_at、索引 idx_{table}_{col} +-- （援引 AGENTS §10 已登记例外，与既有迁移一致）。时间戳沿用 0001 起的 +-- `strftime('%Y-%m-%dT%H:%M:%fZ','now')`（UTC ISO-8601 毫秒），定宽 ⇒ 字典序比较 +-- 等价于时间序，时间窗筛选因此能直接吃上 idx_channel_events_created_at。 +-- +-- 脱敏（design/26 §4.2，**表结构层面也留了痕**）：本表**没有**承载平台原文的列 —— +-- summary 只存适配器按事件类型生成的**固定文案**，OneBot request.flag（可执行凭据） +-- 与平台正文**绝不入库**（故本迁移里连一个 flag/comment/raw/content 列都不存在）。 + +CREATE TABLE IF NOT EXISTS channel_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + platform TEXT NOT NULL, -- qq-obv11 / kook / telegram / weixin-ilink / webhook + account_id TEXT NOT NULL DEFAULT '', -- 多账号（CHANNELS_JSON 的账号 id）；空 = 单账号/未知 + tier TEXT NOT NULL DEFAULT '', -- t1 / t2 / t3（design/26 §2 的留存档） + category TEXT NOT NULL DEFAULT '', -- 归一化类别（join/leave/recall/ban/admin/friend_request/ + -- group_request/poke/reaction/button/item/vote/link/unknown） + platform_type TEXT NOT NULL DEFAULT '', -- 平台原生类型码（notice_type / meta_event_type / request_type / + -- KOOK extra.type / Telegram update 键名），原样保留便于排障 + channel_id TEXT NOT NULL DEFAULT '', + channel_type TEXT NOT NULL DEFAULT '', -- direct / group + group_id TEXT NOT NULL DEFAULT '', + actor_id TEXT NOT NULL DEFAULT '', -- 结构化字段，**绝不来自文本** + actor_role TEXT NOT NULL DEFAULT '', -- owner/member/system/unknown（结构化解析所得） + target_id TEXT NOT NULL DEFAULT '', -- 被作用者（poke / 禁言 / 运气王…） + self_related INTEGER NOT NULL DEFAULT 0, -- 是否命中 self_id / 她自己 + message_id TEXT NOT NULL DEFAULT '', -- 关联消息 id（撤回 / 回应 / 按钮 / 置顶） + summary TEXT NOT NULL DEFAULT '', -- 脱敏结构级摘要（固定文案，不含平台原文） + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) +); + +-- 查询面与 GC 的**唯一依赖**：列表倒序、时间窗筛选、过期行清扫都靠它。 +CREATE INDEX IF NOT EXISTS idx_channel_events_created_at ON channel_events (created_at); +-- 面板/端点的「平台 + 类别」筛选（design/26 §4.5 的查询参数）。 +CREATE INDEX IF NOT EXISTS idx_channel_events_platform_category ON channel_events (platform, category); +CREATE INDEX IF NOT EXISTS idx_channel_events_channel_id ON channel_events (channel_id); +CREATE INDEX IF NOT EXISTS idx_channel_events_tier ON channel_events (tier); + +CREATE TABLE IF NOT EXISTS channel_event_counters ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + platform TEXT NOT NULL, + account_id TEXT NOT NULL DEFAULT '', + tier TEXT NOT NULL DEFAULT '', + platform_type TEXT NOT NULL DEFAULT '', + category TEXT NOT NULL DEFAULT '', + count INTEGER NOT NULL DEFAULT 0, + first_seen_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + last_seen_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) +); + +-- 唯一约束承载 upsert（`ON CONFLICT(platform,account_id,platform_type) DO UPDATE`）： +-- 一次 flush 因此是同键累加，而不是每次心跳插一行。 +CREATE UNIQUE INDEX IF NOT EXISTS idx_channel_event_counters_key + ON channel_event_counters (platform, account_id, platform_type); +CREATE INDEX IF NOT EXISTS idx_channel_event_counters_category ON channel_event_counters (category); diff --git a/internal/runtime/events/counters.go b/internal/runtime/events/counters.go new file mode 100644 index 0000000..bb174a6 --- /dev/null +++ b/internal/runtime/events/counters.go @@ -0,0 +1,261 @@ +package events + +// 本文件归属**计数面**（design/26 §4.1 表二 + §4.3；切片 E1）：内存聚合 + 有界 +// 缓冲 + 定期 flush + 逐类汇总读取。 +// +// 为什么计数不逐帧落行：心跳 / 在线状态是**高频遥测**，逐帧插一行会把库写胖，而 +// 问题「上游到底推过没有」只要求「某个平台某类事件推过几次」。按 +// `(platform, account_id, platform_type)` upsert 之后，行数上界是「平台数 × 事件 +// 类型数」，与帧数无关（§4.1 表二）。 +// +// 三条不变量（每条都有用例钉住）： +// +// 1. **有界**：待落盘的键数绝不超过 `Config.MaxPendingKeys`。缓冲满了先整批落盘， +// 绝不无界增长（§4.3 的背压；对齐 AGENTS §10 的压测纪律：有界，不靠堆进程）。 +// 2. **尽力而为**：落盘失败只记 WARN / 丢键记 DEBUG，**绝不**让 `Record` 失败 —— +// 入站流水线不该因为计数写不进去而失败或丢帧（§4.3）。 +// 3. **首次权威**：同一个键的 tier/category 只认第一次（upsert 不刷写这两列）， +// 与 `documents` 的 `ON CONFLICT(name) DO NOTHING` 同口径。 + +import ( + "context" + "fmt" + "sync" + "time" +) + +// counterKey 是计数聚合的键，**逐列对齐**迁移 0021 的唯一索引 +// `idx_channel_event_counters_key (platform, account_id, platform_type)`。 +// 键里刻意不含 tier/category：它们由「平台原生类型」决定，不该让同一类帧分裂成 +// 两行（那会让计数上界失控，也让「推过几次」的答案分叉）。 +type counterKey struct { + platform string + accountID string + platformType string +} + +// counterEntry 是一个键的累计值与归类。 +type counterEntry struct { + tier Tier + category Category + count int64 + firstSeen time.Time + lastSeen time.Time +} + +// counterBuffer 是内存计数聚合。它由互斥量保护（Record 可能来自多条渠道 goroutine）。 +// +// 锁的纪律（AGENTS §六）：**锁内绝不做 IO** —— drain 在锁内只做「搬运 + 清空」， +// 真正的落盘在锁外进行；失败时的回填也走一次短暂的加锁。 +type counterBuffer struct { + mu sync.Mutex + max int + pending map[counterKey]counterEntry +} + +// newCounterBuffer 构造聚合缓冲；最大键数 ≤0 时取 DefaultMaxPendingKeys。 +func newCounterBuffer(max int) *counterBuffer { + if max <= 0 { + max = DefaultMaxPendingKeys + } + return &counterBuffer{max: max, pending: make(map[counterKey]counterEntry, max)} +} + +// add 把一帧的计数累加进缓冲。返回值报告**这次计数是否被吞掉**：缓冲已达最大键数 +// 且这是一个新键时为 false（该键不落内存）。它是「有界」在代码里的落点 —— +// 无论落盘成功与否，`len()` 都绝不超过 `max`。 +func (b *counterBuffer) add(key counterKey, tier Tier, category Category, at time.Time) bool { + b.mu.Lock() + defer b.mu.Unlock() + if entry, ok := b.pending[key]; ok { + entry.count++ + entry.lastSeen = at + b.pending[key] = entry + return true + } + if len(b.pending) >= b.max { + return false + } + b.pending[key] = counterEntry{tier: tier, category: category, count: 1, firstSeen: at, lastSeen: at} + return true +} + +// len 返回待落盘的键数（≤ max）。 +func (b *counterBuffer) len() int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.pending) +} + +// drain 取走全部待落盘项并清空缓冲（锁内只搬运，不 IO）。 +func (b *counterBuffer) drain() map[counterKey]counterEntry { + b.mu.Lock() + defer b.mu.Unlock() + if len(b.pending) == 0 { + return nil + } + out := b.pending + b.pending = make(map[counterKey]counterEntry, b.max) + return out +} + +// restore 把一次**落盘失败**的项回填进缓冲，使计数不在瞬时 DB 故障里丢掉。 +// 回填同样受 `max` 约束：装不下的项被丢弃并返回丢弃数（调用方记 WARN）—— +// 「尽力而为」与「有界」冲突时，**有界**优先（§4.3 的背压是硬约束）。 +func (b *counterBuffer) restore(entries map[counterKey]counterEntry) int { + b.mu.Lock() + defer b.mu.Unlock() + dropped := 0 + for key, entry := range entries { + if existing, ok := b.pending[key]; ok { + // 落盘失败期间同一键可能又累加了新计数：合并而不是覆盖。 + existing.count += entry.count + if entry.lastSeen.After(existing.lastSeen) { + existing.lastSeen = entry.lastSeen + } + if entry.firstSeen.Before(existing.firstSeen) { + existing.firstSeen = entry.firstSeen + } + b.pending[key] = existing + continue + } + if len(b.pending) >= b.max { + dropped++ + continue + } + b.pending[key] = entry + } + return dropped +} + +// recordCounter 把一帧计入内存聚合（design/26 §4.3：T1 与**全部帧**都计数）。 +// +// 它**不返回错误**：计数是尽力而为，绝不让入站流水线因计数而失败。缓冲满时先尝试 +// 整批落盘；落盘失败记 WARN（DB 出问题时每次 Record 都会试一次，这也让「DB 恢复后 +// 自动收敛」不需要额外的重试器）。被吞掉的新键记 DEBUG —— 单个键的丢弃是上述 WARN +// 的后果，不在这里重复刷 WARN（避免 DB 长时间不可用时日志洪泛）。 +func (s *SQLStore) recordCounter(ctx context.Context, ev Event) { + if s.counters.len() >= s.counters.max { + if err := s.flushCounters(ctx); err != nil { + s.log.Warn("channel event counter flush failed", + "err", err, "pending", s.counters.len(), "max_pending", s.counters.max) + } + } + key := counterKey{platform: ev.Platform, accountID: ev.AccountID, platformType: ev.PlatformType} + if !s.counters.add(key, ev.Tier, ev.Category, ev.CreatedAt) { + s.log.Debug("channel event counter key dropped: buffer full", + "platform", ev.Platform, "platform_type", ev.PlatformType) + } +} + +// Flush 把内存里待落盘的计数写进库（组合根的常驻 ticker 与退出前的最后一次调用 +// 都走它）。没有待落盘项时是空操作。 +func (s *SQLStore) Flush(ctx context.Context) error { + return s.flushCounters(ctx) +} + +// flushCounters 是 Flush 的实现：取走 → 落盘 → 失败回填。 +func (s *SQLStore) flushCounters(ctx context.Context) error { + pending := s.counters.drain() + if len(pending) == 0 { + return nil + } + if err := s.writeCounters(ctx, pending); err != nil { + if dropped := s.counters.restore(pending); dropped > 0 { + s.log.Warn("channel event counter entries dropped after failed flush", + "dropped", dropped, "max_pending", s.counters.max) + } + return fmt.Errorf("events: flush 计数: %w", err) + } + s.log.Debug("channel event counters flushed", "keys", len(pending)) + return nil +} + +// upsertCounterSQL 是批量落盘的唯一语句：单事务 + 同一条语句反复执行，避免 N 次 +// 往返以外的组织开销（design/26 §4.3）。`count = count + excluded.count` 使同键 +// 累加；tier/category 只在**插入**时写入，冲突时保持首次的归类（首次权威）。 +const upsertCounterSQL = `INSERT INTO channel_event_counters + (platform, account_id, tier, platform_type, category, count, first_seen_at, last_seen_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(platform, account_id, platform_type) DO UPDATE SET + count = count + excluded.count, + last_seen_at = excluded.last_seen_at` + +// writeCounters 在一个事务里 upsert 整批计数：要么全落，要么全不落（失败时由 +// flushCounters 回填缓冲，交给下一次 flush 重试）。 +func (s *SQLStore) writeCounters(ctx context.Context, pending map[counterKey]counterEntry) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("events: 计数落盘开事务: %w", err) + } + defer func() { _ = tx.Rollback() }() + + for key, entry := range pending { + if _, err := tx.ExecContext(ctx, upsertCounterSQL, + key.platform, key.accountID, string(entry.tier), key.platformType, string(entry.category), + entry.count, sqliteTimeStamp(entry.firstSeen), sqliteTimeStamp(entry.lastSeen)); err != nil { + return fmt.Errorf("events: 计数落盘(platform=%s type=%s): %w", key.platform, key.platformType, err) + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("events: 计数落盘提交: %w", err) + } + return nil +} + +// listCountersSQL 的列显式列出（禁 `SELECT *`），排序固定，便于客户端做稳定展示。 +const listCountersSQL = `SELECT platform, account_id, tier, platform_type, category, count, + first_seen_at, last_seen_at +FROM channel_event_counters` + +// Counters 返回逐类计数汇总，按 `(platform, account_id, platform_type)` 升序。 +// platform 为空表示不过滤（返回全部平台）。 +// +// 它读的是**已落盘**的表，不含内存里尚未 flush 的部分 —— 这正是「计数为近似值」 +// 的来源，查询面必须如实标注（design/26 §4.3/§4.5）。需要新鲜数据时先 Flush。 +func (s *SQLStore) Counters(ctx context.Context, platform string) ([]Counter, error) { + query := listCountersSQL + args := make([]any, 0, 1) + if platform != "" { + query += " WHERE platform = ?" + args = append(args, platform) + } + query += " ORDER BY platform, account_id, platform_type" + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("events: 查计数行: %w", err) + } + defer func() { _ = rows.Close() }() + + out := make([]Counter, 0, 16) + for rows.Next() { + var ( + c Counter + tier, category string + first, last string + ) + if err := rows.Scan(&c.Platform, &c.AccountID, &tier, &c.PlatformType, &category, + &c.Count, &first, &last); err != nil { + return nil, fmt.Errorf("events: 读计数行: %w", err) + } + c.Tier = Tier(tier) + c.Category = Category(category) + at, ok := parseSQLiteTime(first) + if !ok { + return nil, fmt.Errorf("events: 计数行 %s/%s 的 first_seen_at 不可解析(%q): %w", + c.Platform, c.PlatformType, first, ErrBadTimestamp) + } + c.FirstSeenAt = at + if ts, ok := parseSQLiteTime(last); ok { + c.LastSeenAt = ts + } else { + c.LastSeenAt = at + } + out = append(out, c) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("events: 遍历计数行: %w", err) + } + return out, nil +} diff --git a/internal/runtime/events/counters_test.go b/internal/runtime/events/counters_test.go new file mode 100644 index 0000000..a0a9a5b --- /dev/null +++ b/internal/runtime/events/counters_test.go @@ -0,0 +1,210 @@ +package events + +// 本文件是**计数面**的验收（design/26 §4.1 表二 + §4.3；切片 E1）。三条不变量在这里 +// 各自有对应用例：**有界**（缓冲满即落盘、绝不无界增长）、**尽力而为**（落盘失败 +// 不上抛、不丢帧）、**近似值语义**（读的是已落盘的表，不是内存里的实时值）。 + +import ( + "context" + "fmt" + "testing" + "time" +) + +// TestCounters_BoundedBuffer_边界：缓冲设上界后，**必须**在满时整批落盘，且待落盘 +// 键数在任何时刻都不超过上界（design/26 §4.3 的背压；对齐 AGENTS §10 压测纪律的 +// 「有界」）。同时验证落盘不是「丢计数」：最终总数与录制条数一致。 +func TestCounters_BoundedBuffer_边界(t *testing.T) { + t.Parallel() + ctx := context.Background() + const maxKeys = 4 + st, _ := newTestStore(t, func(cfg *Config) { cfg.MaxPendingKeys = maxKeys }) + + const frames = 20 + for i := 0; i < frames; i++ { + if err := st.Record(ctx, Event{ + Platform: PlatformQQ, + Tier: TierT1, + Category: CategoryHeartbeat, + PlatformType: fmt.Sprintf("type-%02d", i), + }); err != nil { + t.Fatalf("Record(#%d): %v", i, err) + } + if got := st.PendingKeys(); got > maxKeys { + t.Fatalf("PendingKeys = %d after frame #%d, want ≤ %d（缓冲必须有界）", got, i, maxKeys) + } + } + + // 「有界」的直接推论：这一路必须**已经落过盘**（否则内存里会攒着 20 个键）。 + flushed, err := st.Counters(ctx, PlatformQQ) + if err != nil { + t.Fatalf("Counters: %v", err) + } + if len(flushed) == 0 { + t.Fatal("no counters flushed to the database: a bounded buffer must flush when full") + } + + // 落盘 = 计数搬家而不是丢掉：最后一次 flush 后总数必须恰为 frames。 + if err := st.Flush(ctx); err != nil { + t.Fatalf("Flush: %v", err) + } + var total int64 + for _, c := range mustCounters(t, st, PlatformQQ) { + total += c.Count + } + if total != frames { + t.Fatalf("flushed count total = %d, want %d", total, frames) + } + if got := st.PendingKeys(); got != 0 { + t.Fatalf("PendingKeys = %d after Flush, want 0", got) + } +} + +// TestCounters_FlushFailureStaysBounded_异常：数据库不可用时，计数落盘失败只记 WARN、 +// **不上抛**（Record 仍返回 nil —— 计数是尽力而为，绝不让入站流水线失败，§4.3）， +// 并且缓冲**仍然有界**（绝不因为一直失败而无界增长）。 +func TestCounters_FlushFailureStaysBounded_异常(t *testing.T) { + t.Parallel() + ctx := context.Background() + const maxKeys = 2 + st, conn := newTestStore(t, func(cfg *Config) { cfg.MaxPendingKeys = maxKeys }) + if err := conn.Close(); err != nil { + t.Fatalf("关闭内存库: %v", err) + } + + for i := 0; i < 5; i++ { + if err := st.Record(ctx, Event{ + Platform: PlatformQQ, + Tier: TierT1, + Category: CategoryHeartbeat, + PlatformType: fmt.Sprintf("type-%02d", i), + }); err != nil { + t.Fatalf("Record(#%d) err = %v, want nil（计数失败绝不能上抛）", i, err) + } + if got := st.PendingKeys(); got > maxKeys { + t.Fatalf("PendingKeys = %d after failing flush #%d, want ≤ %d", got, i, maxKeys) + } + } + if got := st.PendingKeys(); got != maxKeys { + t.Fatalf("PendingKeys = %d, want %d (失败项回填后仍受上界约束)", got, maxKeys) + } +} + +// TestCounters_SameKeyAccumulates_正常：同一个键（平台+账号+原生类型）反复计数是 +// **累加同一行**，且 tier/category 首次权威、first/last_seen_at 如实推进。 +func TestCounters_SameKeyAccumulates_正常(t *testing.T) { + t.Parallel() + ctx := context.Background() + st, _ := newTestStore(t, nil) + + ev := Event{ + Platform: PlatformQQ, + AccountID: "bot-1", + Tier: TierT1, + Category: CategoryHeartbeat, + PlatformType: "heartbeat", + } + for i := 0; i < 3; i++ { + if err := st.Record(ctx, ev); err != nil { + t.Fatalf("Record(#%d): %v", i, err) + } + } + if err := st.Flush(ctx); err != nil { + t.Fatalf("Flush: %v", err) + } + + later := ev + later.CreatedAt = testNow().Add(time.Minute) + for i := 0; i < 2; i++ { + if err := st.Record(ctx, later); err != nil { + t.Fatalf("Record(later #%d): %v", i, err) + } + } + if err := st.Flush(ctx); err != nil { + t.Fatalf("Flush: %v", err) + } + + rows := mustCounters(t, st, PlatformQQ) + if len(rows) != 1 { + t.Fatalf("counter rows = %d, want 1 (同键必须合并)", len(rows)) + } + got := rows[0] + if got.Count != 5 { + t.Fatalf("count = %d, want 5 (3 + 2)", got.Count) + } + if !got.FirstSeenAt.Equal(testNow()) { + t.Fatalf("first_seen_at = %v, want %v (首次权威)", got.FirstSeenAt, testNow()) + } + if !got.LastSeenAt.Equal(later.CreatedAt) { + t.Fatalf("last_seen_at = %v, want %v", got.LastSeenAt, later.CreatedAt) + } + if got.AccountID != "bot-1" || got.PlatformType != "heartbeat" { + t.Fatalf("key columns = (%q, %q), want (bot-1, heartbeat)", got.AccountID, got.PlatformType) + } +} + +// TestCounters_PlatformFilter_正常：Counters 的 platform 参数为空表示全部平台， +// 非空即精确过滤（design/26 §4.5 的查询参数语义）。 +func TestCounters_PlatformFilter_正常(t *testing.T) { + t.Parallel() + ctx := context.Background() + st, _ := newTestStore(t, nil) + + for _, platform := range []string{PlatformQQ, PlatformKook} { + if err := st.Record(ctx, Event{ + Platform: platform, Tier: TierT1, Category: CategoryHeartbeat, PlatformType: "heartbeat", + }); err != nil { + t.Fatalf("Record(%s): %v", platform, err) + } + } + if err := st.Flush(ctx); err != nil { + t.Fatalf("Flush: %v", err) + } + + if rows := mustCounters(t, st, ""); len(rows) != 2 { + t.Fatalf("all platforms = %d rows, want 2", len(rows)) + } + rows := mustCounters(t, st, PlatformKook) + if len(rows) != 1 || rows[0].Platform != PlatformKook { + t.Fatalf("filtered rows = %+v, want exactly the kook counter", rows) + } + if rows := mustCounters(t, st, PlatformTelegram); len(rows) != 0 { + t.Fatalf("unknown platform = %d rows, want 0", len(rows)) + } +} + +// TestCounters_PendingNotVisibleBeforeFlush_边界：Counters 读的是**已落盘**的表 —— +// 这正是「计数为近似值」的来源（design/26 §4.3/§4.5），查询面必须如实标注。 +func TestCounters_PendingNotVisibleBeforeFlush_边界(t *testing.T) { + t.Parallel() + ctx := context.Background() + st, _ := newTestStore(t, nil) + + if err := st.Record(ctx, Event{ + Platform: PlatformQQ, Tier: TierT1, Category: CategoryHeartbeat, PlatformType: "heartbeat", + }); err != nil { + t.Fatalf("Record: %v", err) + } + if rows := mustCounters(t, st, PlatformQQ); len(rows) != 0 { + t.Fatalf("rows before Flush = %d, want 0（内存计数不算已落盘）", len(rows)) + } + if st.PendingKeys() != 1 { + t.Fatalf("PendingKeys = %d, want 1", st.PendingKeys()) + } + if err := st.Flush(ctx); err != nil { + t.Fatalf("Flush: %v", err) + } + if rows := mustCounters(t, st, PlatformQQ); len(rows) != 1 { + t.Fatalf("rows after Flush = %d, want 1", len(rows)) + } +} + +// mustCounters 是 Counters 的测试包装（失败即 Fatal）。 +func mustCounters(t *testing.T, st *SQLStore, platform string) []Counter { + t.Helper() + rows, err := st.Counters(context.Background(), platform) + if err != nil { + t.Fatalf("Counters(%q): %v", platform, err) + } + return rows +} diff --git a/internal/runtime/events/events.go b/internal/runtime/events/events.go new file mode 100644 index 0000000..d10d275 --- /dev/null +++ b/internal/runtime/events/events.go @@ -0,0 +1,264 @@ +// Package events 是**渠道事件留存**的存储地基（design/26 §4；切片 E1）。 +// +// 它补的是 design/26 §0.2 那次无法定案的事故：适配器把非消息帧静默丢掉之后， +// 「上游推过没有 / 是不是被我们过滤了」在数据里再也答不上来。本包只做一件事 —— +// 让到达适配器的每一帧至少留下**可查**（逐条落行）且**可计数**（按平台+账号+ +// 平台原生类型聚合）的脱敏结构痕迹。 +// +// 两张表的分工（design/26 §4.1，刻意不合并，理由见迁移 0021）： +// +// - `channel_events`：T2/T3 逐条、全部 unknown 与链路事件；低频、有审计价值、 +// 参与保留期 GC（Sweep）； +// - `channel_event_counters`：所有帧的计数，行数上界 = 平台数 × 事件类型数， +// **不参与 GC**（删了就答不上「上游推过没有」）。 +// +// 职责边界（别在这里越界）： +// +// - **不判身份、不判权限**：actor/role 由 channel 层的结构化元数据 + 白名单 + +// 策略引擎决定，本包只如实落列（AGENTS §9 红线）； +// - **不决定进不进上下文**：那是 `adapter.NoticeClass`（design/26 §2.3 的另一个 +// 正交轴）与 channelbridge 的事； +// - **适配器不依赖本包**：design/26 §6.1 要求适配器对存储零依赖，写入统一由 +// 桥接层（E3）经 `channelbridge.Options.EventStore` 调用。 +// +// 脱敏是**调用方的责任**（design/26 §4.2）：`Event.Summary` 必须是适配器按事件类型 +// 生成的固定文案，净化走既有的 `adapter.SanitizeNoticeText`（唯一出处，本包不另造）。 +// 因此 `Event` 里**没有任何承载平台原文的字段** —— 尤其 OneBot `request.flag` +// （可执行凭据，等同 token）**不在本结构里，也绝不得入表**（design/26 §4.2 特例治理）。 +package events + +import ( + "context" + "errors" + "time" +) + +// 存储侧的哨兵错误。调用方一律用 `errors.Is` 分类，**绝不按错误文本匹配**。 +var ( + // ErrNilDB 报告构造 Store 时没有可用的数据库连接。宁可明确失败，也不给出 + // 一个写不进去的替身（与 `history.NewStore` 同一口径）。 + ErrNilDB = errors.New("events: 缺少数据库连接") + + // ErrNoPlatform 报告一条事件没有来源平台。平台是「哪条链路」的唯一标识， + // 缺了它这一行既查不到也归不了类 —— 这是调用方的编程错误，必须显式失败， + // 而不是写一行 platform='' 的行进库。 + ErrNoPlatform = errors.New("events: 事件缺少来源平台") + + // ErrBadTimestamp 报告读回的事件行携带一个无法解析的时间戳。宁可如实报错， + // 也不静默给一个零值时间（那会让 GC 与查询面基于假时刻做判断）。 + ErrBadTimestamp = errors.New("events: 事件行时间戳不可解析") +) + +// Tier 是 design/26 §2 的**留存档**：回答「要不要留在本地、要不要告知、怎么归档」。 +// 它与 `adapter.NoticeClass`（回答「进不进她的上下文」）是**两个正交轴**（§2.3）， +// 两轴的结果都落库：前者是本包的行/计数归属，后者记进日志与事件行之外。 +type Tier string + +// 三档留存（design/26 §2.1）。取值即数据库列 `tier` 的字面值。 +const ( + // TierT1 只留存：纯遥测 / 存在性 / 资源（心跳、生命周期、他人在线状态、 + // 资料与频道/角色/表情增删改…）。**绝不进模型上下文，绝不推送**。 + TierT1 Tier = "t1" + // TierT2 留存 + 上行：成员 / 权限 / 消息状态的变动（成员进出、撤回、禁言、 + // 管理员变更、加好友/加群请求…）。落行 + 以独立 system 行告知她。 + TierT2 Tier = "t2" + // TierT3 交互事件：人对人的定向动作（戳一戳、表情回应、按钮、道具、投票…）， + // 统一形态是「谁对谁做了什么」。默认只登记、不自动回应（§5.5）。 + TierT3 Tier = "t3" +) + +// Category 是**归一化类别**：把各平台的原生类型码收敛成一套跨平台可查的词汇。 +// +// 每个常量的取值都取自 design/26 §4.1 的列注释与 §2.4 的归一化产出（不自行发明）； +// 平台原生码原样留在 `Event.PlatformType` 里，排障时两者并用（design/26 §2.4）。 +type Category string + +// 归一化类别。前 15 项逐个对应 design/26 §4.1 的 category 注释， +// 后 2 项取自 §3.7（帧解析失败）与 §3.6（适配器链路事件）。 +const ( + // CategoryHeartbeat 心跳（OneBot meta_event.heartbeat）。 + CategoryHeartbeat Category = "heartbeat" + // CategoryLifecycle 生命周期（meta_event.lifecycle 的 connect/enable/disable）。 + CategoryLifecycle Category = "lifecycle" + // CategoryJoin 成员加入（进群 / 进服务器）。 + CategoryJoin Category = "join" + // CategoryLeave 成员离开（退群 / 被踢 / 退服）。 + CategoryLeave Category = "leave" + // CategoryRecall 消息被撤回。 + CategoryRecall Category = "recall" + // CategoryBan 禁言 / 解禁（含全员禁言归一成的独立类别）。 + CategoryBan Category = "ban" + // CategoryAdmin 管理员身份变更。 + CategoryAdmin Category = "admin" + // CategoryFriendRequest 加好友请求（OneBot request.friend；**默认不自动同意**）。 + CategoryFriendRequest Category = "friend_request" + // CategoryGroupRequest 加群请求 / 邀请入群（OneBot request.group）。 + CategoryGroupRequest Category = "group_request" + // CategoryPoke 戳一戳（定向交互）。 + CategoryPoke Category = "poke" + // CategoryReaction 表情回应（Telegram message_reaction / KOOK added_reaction…）。 + CategoryReaction Category = "reaction" + // CategoryButton 按钮点击（callback_query / message_btn_click）。 + CategoryButton Category = "button" + // CategoryItem 道具 / 礼物 / 支付（KOOK type=12 道具消息、Telegram 礼物流程…）。 + CategoryItem Category = "item" + // CategoryVote 投票状态与投票作答（poll / poll_answer）。 + CategoryVote Category = "vote" + // CategoryLink 适配器自身链路事件（连接 / 断开 / 重连 / 会话被拒），§3.6。 + CategoryLink Category = "link" + // CategoryUnknown 未知类型（未知 post_type / notice_type / extra.type / + // Update 字段）。**兜底不等于放行**：它照旧落行计数，但绝不进上下文（§3.7）。 + CategoryUnknown Category = "unknown" + // CategoryDecodeError 帧解析失败（JSON 坏）。按 §3.7 留痕 + 计数， + // **只记错误类型与字节数，绝不记原文**。 + CategoryDecodeError Category = "decode_error" +) + +// 平台标识（取值即 design/26 §4.1 的 platform 列注释）。它们同时也是 +// `channel_event_counters` 唯一键的第一列。 +const ( + // PlatformQQ 是 OneBot v11（go-cqhttp 家族）渠道。 + PlatformQQ = "qq-obv11" + // PlatformKook 是 KOOK 渠道。 + PlatformKook = "kook" + // PlatformTelegram 是 Telegram 渠道。 + PlatformTelegram = "telegram" + // PlatformWeixinILink 是微信 iLink 渠道（平台事实：该渠道**没有** T2/T3 事件， + // design/26 §3.4 要求查询面如实标注，**绝不伪造、绝不用别的渠道顶替**）。 + PlatformWeixinILink = "weixin-ilink" + // PlatformWebhook 是通用 webhook 渠道。 + PlatformWebhook = "webhook" +) + +// 会话形态与行为者角色（取值即迁移 0021 的列注释）。 +const ( + // ChannelDirect 是私聊会话。 + ChannelDirect = "direct" + // ChannelGroup 是群聊 / 频道会话。 + ChannelGroup = "group" + + // RoleOwner 是被白名单认定为主人的行为者。 + RoleOwner = "owner" + // RoleMember 是普通成员。 + RoleMember = "member" + // RoleSystem 是平台系统自身（合成 id "system"，**任何平台字段都不得冒充主人**）。 + RoleSystem = "system" + // RoleUnknown 是未能结构化归类的行为者。它**不是**「陌生人」的对等物： + // 判主与授权永远由结构化字段 + 白名单 + 策略引擎决定（design/26 §1.2）。 + RoleUnknown = "unknown" +) + +// Event 是写入 `channel_events` 的一行（迁移 0021）。列与 design/26 §4.1 一一对应。 +// +// 它**不是**帧的容器：原始帧留在适配器侧的脱敏 DEBUG 日志与 `RawData` 里，本结构 +// 只承载「可查的结构级事实」（§4.2 的「能落」清单）。两个字段刻意缺席： +// +// - OneBot `request.flag` —— 可执行凭据，等同 token，**绝不入库**（§4.2）； +// - 任何平台正文 / 验证信息 / 媒体 URL —— 属会话内容或临时凭据，事件表不是它们的家。 +type Event struct { + // ID 是数据库自增主键。**写侧忽略它**，只在 List 的结果里回填，便于排障与分页定位。 + ID int64 + // Platform 是来源平台（platform 常量之一）。**必填**，否则 Record 返回 ErrNoPlatform。 + Platform string + // AccountID 是多账号标识（CHANNELS_JSON 的账号 id）；空 = 单账号 / 未知。 + AccountID string + // Tier 是留存档。**取值非法时按 TierT1 处理并记 WARN**（fail-safe：宁可只计数， + // 也绝不因为一个字段写错而把整帧丢出留存面）。 + Tier Tier + // Category 是归一化类别。 + Category Category + // PlatformType 是平台原生类型码，原样保留便于排障（notice_type / meta_event_type / + // request_type / KOOK extra.type / Telegram update 键名）。 + PlatformType string + // ChannelID 是会话 / 频道标识。 + ChannelID string + // ChannelType 是会话形态（direct / group）。 + ChannelType string + // GroupID 是群标识（会话之外单独留一列，便于按群查询）。 + GroupID string + // ActorID 是行为者的**结构化 id**（绝不来自昵称或正文）。 + ActorID string + // ActorRole 是行为者角色（结构化解析所得，取值见 Role* 常量）。 + ActorRole string + // TargetID 是被作用者（poke 的 target、被禁言者、运气王…）；无则空。 + TargetID string + // SelfRelated 报告该事件是否命中 self_id / 她自己（design/26 §2.2 的升档依据）。 + SelfRelated bool + // MessageID 是关联消息 id（撤回 / 回应 / 按钮 / 置顶）。 + MessageID string + // Summary 是**脱敏结构级摘要**：适配器按事件类型生成的固定文案，一行常量， + // 不含任何平台原文。净化唯一出处是 `adapter.SanitizeNoticeText`（§4.2）。 + Summary string + // CreatedAt 是事件发生时刻。零值 ⇒ 由 Store 填当前时刻（测试可注入时钟）。 + CreatedAt time.Time + // UpdatedAt 是行被读回时的更新时间戳。事件行不可变，写侧忽略它（写时与 + // CreatedAt 同值），只在 List 的结果里回填。 + UpdatedAt time.Time +} + +// Filter 是事件查询条件（List / Count 共用）。零值字段一律表示「不过滤」。 +type Filter struct { + // Platform / Tier / Category / ChannelID 精确匹配；空串（或零值档）不过滤。 + Platform string + Tier Tier + Category Category + ChannelID string + // Since / Until 是闭区间时间窗；零值不过滤。比较按 UTC 定宽字符串的字典序 + // 进行（与 `llm/usage` 同口径，定宽 ⇒ 字典序等价于时间序，且吃得上 + // idx_channel_events_created_at）。 + Since time.Time + Until time.Time + // Limit / Offset 只作用于 List。Limit ≤0 取 DefaultListLimit，超过 + // MaxListLimit 被**钳到上限**（绝不静默返回全表）。 + Limit int + Offset int +} + +// Counter 是 `channel_event_counters` 的一行：某个平台某类事件的累计计数。 +// +// 它是**近似值**：计数在内存里聚合、定期 flush（design/26 §4.3），crash 会丢 +// ≤一个 flush 间隔的计数。查询面（E6）必须对此如实标注 `approx:true`，**绝不假装 +// 它是精确账**。 +type Counter struct { + // Platform / AccountID / PlatformType 是唯一键的三列。 + Platform string + AccountID string + PlatformType string + // Tier / Category 是**首次**见到该键时记下的归类（同键后续帧不刷写它们， + // 与 `documents` 的 `ON CONFLICT DO NOTHING` 同一「首次权威」口径）。 + Tier Tier + Category Category + // Count 是累计帧数。 + Count int64 + // FirstSeenAt / LastSeenAt 是该键首次 / 最近一次被计数的时刻。它们让「上游 + // 到底推没推」在**没有逐条行**（T1）时依然答得出来（§4.1 不删计数表的理由）。 + FirstSeenAt time.Time + LastSeenAt time.Time +} + +// Store 是事件留存的**唯一对外面**（design/26 §6.3）。桥接层（E3）与组合根（E4） +// 只依赖它，不依赖 `SQLStore` 这个具体类型（便于测试替身与「未接线」的 nil 判定）。 +// +// 实现必须同时满足两条不变量（design/26 §4.3）： +// +// - **逐条留存的失败必须上抛**（`Record`）：T2/T3/unknown 有审计价值，丢不起； +// - **计数是尽力而为**：内存聚合 + 有界缓冲，绝不因为计数落盘失败而让调用方的 +// 入站流水线失败或丢帧。 +type Store interface { + // Record 记录一帧事件：按档决定逐条落行（T2/T3/unknown/链路）还是只累加计数 + // （T1），两者都同时进计数。逐条落行失败时返回错误；计数侧失败只记日志。 + Record(ctx context.Context, ev Event) error + // Count 统计符合条件的事件行数（List 的分页配套）。 + Count(ctx context.Context, f Filter) (int64, error) + // List 按时间倒序返回符合条件的事件行（Limit/Offset 见 Filter）。 + List(ctx context.Context, f Filter) ([]Event, error) + // Counters 返回逐类计数汇总；platform 为空表示全部平台。它读的是**已 flush + // 的表**，不含内存里尚未落盘的部分（那是近似值语义的一部分，见 Counter）。 + Counters(ctx context.Context, platform string) ([]Counter, error) + // Flush 把内存里待落盘的计数写进库。组合根的常驻 ticker 与退出前的最后一次 + // 调用都走它。 + Flush(ctx context.Context) error + // Sweep 按保留期清扫 `channel_events` 的过期行。保留期或间隔 ≤0（GC 关闭） + // 时是空操作；计数表**不参与 GC**。 + Sweep(ctx context.Context) (SweepStats, error) +} diff --git a/internal/runtime/events/gc.go b/internal/runtime/events/gc.go new file mode 100644 index 0000000..478ea2c --- /dev/null +++ b/internal/runtime/events/gc.go @@ -0,0 +1,121 @@ +package events + +// 本文件归属**保留期清扫（Sweep）**（design/26 §4.4；切片 E1）。它逐字沿用 +// `internal/runtime/documents/gc.go` + `cmd/cyrene/wire_document_gc.go` 的既有 GC +// 范式（design/26 §4.4 明写「不另造」），三条取舍也照抄： +// +// 1. **启动期扫一次、此后按间隔定时扫**（间隔默认 DefaultGCInterval）。启动期那次 +// 与定时循环都**不**阻塞启动、不绑在磁盘上 —— 常驻循环住在组合根（E4 的 +// `wire_channel_events.go` 逐字对齐 `wire_document_gc.go`），本包只提供 +// `Sweep` 与它需要的三个口径（Enabled / Interval / Retention）。 +// 2. **失败只记 WARN、绝不让进程退出**：`Sweep` 把查询/删除失败**上抛**给调用方 +// （组合根记 WARN 后继续等下一轮）；已经删掉的行不会因为后续批次失败而「回滚」， +// 未删的行也不会被清空 —— 一轮失败就只是「这一轮没扫干净」。 +// 3. **≤0 = 关闭**（显式），关闭原因在 NewStore 里记进日志 —— 「配了却不生效」 +// 不存在（与文档 GC 同口径）。 +// +// 清理条件只有一条：`created_at < now − 保留期` 的 `channel_events` 行逐批删除。 +// **计数表不参与 GC**：它本身就是「历史已浓缩成一行」的形态，删了就再也答不上 +// 「上游推过没有」——那正是本切片存在的理由（design/26 §4.4）。 + +import ( + "context" + "fmt" + "time" +) + +const ( + // DefaultGCInterval 是 `CHANNEL_EVENT_GC_INTERVAL` 的默认值：每小时清扫一次。 + DefaultGCInterval = time.Hour + + // DefaultRetention 是 `CHANNEL_EVENT_RETENTION` 的默认值：30 天。它比文档保留期 + // （168h）长得多，因为事件表的核心用途是**事后定案**（design/26 §0.2）—— + // 一周太短，事故复盘常常发生在一周之后（§4.4）。 + DefaultRetention = 720 * time.Hour + + // sweepBatchSize 是单批删除的行数上限；maxSweepBatches 是一轮清扫的批数上限。 + // 两者一起把「一轮清扫」的时长与单次内存占用**有界**（AGENTS §10 的压测纪律： + // 有界，不靠堆进程）：积压百万行时不是一次 DELETE 全表，而是每轮最多 + // maxSweepBatches × sweepBatchSize 行，剩下的留给下一轮。 + sweepBatchSize = 500 + maxSweepBatches = 100 +) + +// SweepStats 是一轮清扫的汇总（成功件只记总数，绝不记任何行内容）。 +type SweepStats struct { + // Batches 是实际执行的删除批数。 + Batches int + // Deleted 是删除的过期行数。 + Deleted int64 +} + +// Enabled 报告清扫是否真的会删行（保留期与间隔都 > 0）。 +func (s *SQLStore) Enabled() bool { return s != nil && s.gcEnabled } + +// Interval 返回定时清扫间隔（关闭时为 0）。 +func (s *SQLStore) Interval() time.Duration { + if s == nil { + return 0 + } + return s.gcInterval +} + +// Retention 返回事件行保留期（关闭时为 0）。 +func (s *SQLStore) Retention() time.Duration { + if s == nil { + return 0 + } + return s.retention +} + +// sweepSQL 按时间戳删一批过期行。子查询带 `ORDER BY created_at LIMIT ?`，因此它 +// 吃得上 `idx_channel_events_created_at`，且每批的扫描面有界（design/26 §4.4）。 +// 时间戳用定宽 UTC 字符串比较：字典序等价于时间序（见 store.go 的口径注释）。 +const sweepSQL = `DELETE FROM channel_events WHERE id IN ( + SELECT id FROM channel_events WHERE created_at < ? ORDER BY created_at LIMIT ?)` + +// Sweep 按保留期清扫 `channel_events` 的过期行，返回一轮汇总。 +// +// 关闭时（Enabled() 为 false）是**空操作**：返回零值与 nil 错误、不碰任何一行。 +// 这与「配了却不生效」是两回事 —— 关闭原因在 NewStore 里已记 WARN。 +// +// 失败语义：查询/删除失败返回错误并**带上已完成的统计**（调用方据此知道这一轮 +// 走到了哪一步，记 WARN 后等下一轮）。ctx 取消时立刻停下：**已删的保持删除、 +// 未删的绝不因此被清空** —— 这是「失败不清空」在代码里的落点。 +func (s *SQLStore) Sweep(ctx context.Context) (SweepStats, error) { + if !s.Enabled() { + return SweepStats{}, nil + } + cutoff := sqliteTimeStamp(s.now().Add(-s.retention)) + var stats SweepStats + for i := 0; i < maxSweepBatches; i++ { + if err := ctx.Err(); err != nil { + return stats, fmt.Errorf("events: sweep: %w", err) + } + deleted, err := s.sweepBatch(ctx, cutoff) + if err != nil { + return stats, err + } + stats.Batches++ + stats.Deleted += deleted + if deleted < int64(sweepBatchSize) { + break + } + } + s.log.Info("channel event sweep", + "deleted", stats.Deleted, "batches", stats.Batches, "retention", s.retention.String()) + return stats, nil +} + +// sweepBatch 删一批过期行并返回本批删掉的行数。 +func (s *SQLStore) sweepBatch(ctx context.Context, cutoff string) (int64, error) { + res, err := s.db.ExecContext(ctx, sweepSQL, cutoff, sweepBatchSize) + if err != nil { + return 0, fmt.Errorf("events: sweep 删过期行: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return 0, fmt.Errorf("events: sweep 读取删除行数: %w", err) + } + return n, nil +} diff --git a/internal/runtime/events/gc_test.go b/internal/runtime/events/gc_test.go new file mode 100644 index 0000000..8791e4f --- /dev/null +++ b/internal/runtime/events/gc_test.go @@ -0,0 +1,209 @@ +package events + +// 本文件是**保留期清扫**的验收（design/26 §4.4；切片 E1）。三类必测： +// +// 1. **≤0 = 关闭**（保留期 ≤0、间隔 ≤0 各一条）：`Enabled()` 为 false 且 `Sweep` +// 不动任何一行 —— 「配了却不生效」不存在（关闭原因由 NewStore 记 WARN）； +// 2. **正数清扫**：超期行按 `created_at < now − 保留期` 逐批删除，未超期的行与原样 +// 保留，计数表**不参与 GC**； +// 3. **失败不退出、不清空**：库不可用 / ctx 取消时 `Sweep` 返回错误（组合根记 WARN +// 后等下一轮，绝不因此退出进程），**已经删的保持删除、未删的绝不被清空**。 + +import ( + "context" + "testing" + "time" +) + +// seedExpired 写入一条指定时刻的 T2 事件（走真实写入路径，不直接拼 SQL）。 +func seedExpired(t *testing.T, st *SQLStore, at time.Time, platformType string) { + t.Helper() + ev := sampleT2() + ev.PlatformType = platformType + ev.CreatedAt = at + if err := st.Record(context.Background(), ev); err != nil { + t.Fatalf("Record(%s): %v", platformType, err) + } +} + +// TestSweep_RetentionDisabled_边界：保留期 ≤0 ⇒ GC 关闭（显式），Sweep 是空操作。 +func TestSweep_RetentionDisabled_边界(t *testing.T) { + t.Parallel() + ctx := context.Background() + st, _ := newTestStore(t, func(cfg *Config) { + cfg.GCInterval = time.Hour + cfg.Retention = 0 + }) + + if st.Enabled() { + t.Fatal("Enabled() = true, want false（保留期 ≤0 = 关闭 GC）") + } + if st.Interval() != time.Hour || st.Retention() != 0 { + t.Fatalf("Interval/Retention = %v/%v, want 1h/0", st.Interval(), st.Retention()) + } + seedExpired(t, st, testNow().Add(-1000*time.Hour), "group_recall") + + stats, err := st.Sweep(ctx) + if err != nil { + t.Fatalf("Sweep(disabled) err = %v, want nil", err) + } + if stats.Batches != 0 || stats.Deleted != 0 { + t.Fatalf("Sweep(disabled) = %+v, want zero stats", stats) + } + if n, err := st.Count(ctx, Filter{}); err != nil { + t.Fatalf("Count: %v", err) + } else if n != 1 { + t.Fatalf("rows = %d, want 1（GC 关闭时一行都不许动）", n) + } +} + +// TestSweep_IntervalDisabled_边界：间隔 ≤0 ⇒ 同样是关闭（与文档 GC 逐字同口径）。 +func TestSweep_IntervalDisabled_边界(t *testing.T) { + t.Parallel() + ctx := context.Background() + st, _ := newTestStore(t, func(cfg *Config) { + cfg.GCInterval = 0 + cfg.Retention = DefaultRetention + }) + + if st.Enabled() { + t.Fatal("Enabled() = true, want false（间隔 ≤0 = 关闭 GC）") + } + seedExpired(t, st, testNow().Add(-1000*time.Hour), "group_recall") + + stats, err := st.Sweep(ctx) + if err != nil { + t.Fatalf("Sweep(disabled) err = %v, want nil", err) + } + if stats.Deleted != 0 { + t.Fatalf("Sweep(disabled) deleted = %d, want 0", stats.Deleted) + } + if n, err := st.Count(ctx, Filter{}); err != nil { + t.Fatalf("Count: %v", err) + } else if n != 1 { + t.Fatalf("rows = %d, want 1", n) + } +} + +// TestSweep_DeletesExpiredOnly_正常：正数保留期 ⇒ 只删超期行；未超期的行保留， +// **计数表不参与 GC**（design/26 §4.4：删了就答不上「上游推过没有」）。 +func TestSweep_DeletesExpiredOnly_正常(t *testing.T) { + t.Parallel() + ctx := context.Background() + const retention = 24 * time.Hour + st, _ := newTestStore(t, func(cfg *Config) { + cfg.GCInterval = time.Hour + cfg.Retention = retention + }) + if !st.Enabled() { + t.Fatal("Enabled() = false, want true") + } + + seedExpired(t, st, testNow().Add(-48*time.Hour), "expired") + seedExpired(t, st, testNow().Add(-time.Hour), "fresh") + if err := st.Record(ctx, Event{ + Platform: PlatformQQ, Tier: TierT1, Category: CategoryHeartbeat, PlatformType: "heartbeat", + }); err != nil { + t.Fatalf("Record(heartbeat): %v", err) + } + if err := st.Flush(ctx); err != nil { + t.Fatalf("Flush: %v", err) + } + + stats, err := st.Sweep(ctx) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + if stats.Deleted != 1 { + t.Fatalf("Deleted = %d, want 1", stats.Deleted) + } + + rows, err := st.List(ctx, Filter{}) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(rows) != 1 || rows[0].PlatformType != "fresh" { + t.Fatalf("rows after sweep = %+v, want exactly the fresh one", rows) + } + // 计数表不参与 GC：即便某个键的**事件行**被清扫掉，它的计数行照旧在（design/26 + // §4.4 —— 删了就再也答不上「上游推过没有」）。 + counters := mustCounters(t, st, PlatformQQ) + seen := make(map[string]int64, len(counters)) + for _, c := range counters { + seen[c.PlatformType] = c.Count + } + for _, key := range []string{"expired", "fresh", "heartbeat"} { + if seen[key] != 1 { + t.Fatalf("counter %q = %d, want 1（全部计数行都必须留着）: %+v", key, seen[key], counters) + } + } +} + +// TestSweep_MultipleBatches_边界：积压超过单批上限时按批删除（一轮的时长与内存占用 +// 有界），且一轮内能把超过一批的量扫完。 +func TestSweep_MultipleBatches_边界(t *testing.T) { + t.Parallel() + ctx := context.Background() + st, _ := newTestStore(t, func(cfg *Config) { + cfg.GCInterval = time.Hour + cfg.Retention = time.Hour + }) + + const rows = sweepBatchSize + 1 + old := testNow().Add(-100 * time.Hour) + for i := 0; i < rows; i++ { + seedExpired(t, st, old, "expired") + } + + stats, err := st.Sweep(ctx) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + if stats.Deleted != int64(rows) { + t.Fatalf("Deleted = %d, want %d", stats.Deleted, rows) + } + if stats.Batches != 2 { + t.Fatalf("Batches = %d, want 2 (单批上限 %d)", stats.Batches, sweepBatchSize) + } + 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) + } +} + +// TestSweep_FailureDoesNotClear_异常：库不可用时 Sweep 返回错误（组合根记 WARN 后 +// 继续），ctx 取消时立刻停下且**不清空**未删的行。 +func TestSweep_FailureDoesNotClear_异常(t *testing.T) { + t.Parallel() + ctx := context.Background() + st, conn := newTestStore(t, func(cfg *Config) { + cfg.GCInterval = time.Hour + cfg.Retention = time.Hour + }) + seedExpired(t, st, testNow().Add(-100*time.Hour), "expired") + + // ① ctx 取消：立即停下、返回错误、一行都不删（未删的绝不因此被清空）。 + canceled, cancel := context.WithCancel(ctx) + cancel() + stats, err := st.Sweep(canceled) + if err == nil { + t.Fatal("Sweep(canceled ctx) err = nil, want an error") + } + if stats.Deleted != 0 { + t.Fatalf("Sweep(canceled ctx) deleted = %d, want 0", stats.Deleted) + } + if n, cerr := st.Count(ctx, Filter{}); cerr != nil { + t.Fatalf("Count: %v", cerr) + } else if n != 1 { + t.Fatalf("rows = %d, want 1 (取消不得清空)", n) + } + + // ② 库不可用：同样返回错误（调用方等下一轮），且不 panic。 + if err := conn.Close(); err != nil { + t.Fatalf("关闭内存库: %v", err) + } + if _, err := st.Sweep(ctx); err == nil { + t.Fatal("Sweep(closed db) err = nil, want an error") + } +} diff --git a/internal/runtime/events/store.go b/internal/runtime/events/store.go new file mode 100644 index 0000000..bfb782d --- /dev/null +++ b/internal/runtime/events/store.go @@ -0,0 +1,404 @@ +package events + +// 本文件归属**事件存储核心**（design/26 §4.1/§4.3；切片 E1）：构造、写入路径 +// （`Record` 的逐条落行 + 计数归属）与两个只读查询（`Count` / `List`）。 +// +// 三条取舍写在这里： +// +// 1. **T2/T3/unknown 同步逐条落行，失败上抛**（§4.3）：这些行有审计价值、丢不起， +// 它们的失败必须让调用方看见（E3 据此记 WARN，但**不**因此丢帧 —— 留存与 +// 「进不进上下文」是两个独立动作）。 +// 2. **T1 只累加计数**（§4.3）：心跳/在线状态是高频遥测，逐帧写库是纯浪费。 +// 两个**例外**逐条落行（见 persistsAsRow）：`unknown`/`decode_error`（§3.7 明令 +// 「留痕 + 计数」）与 §3.6 的链路事件（`link`，它回答的正是「链路当时是不是 +// 活的」——design/26 §0.2 事故定案所依赖的那份证据）。 +// 3. **时间戳口径与既有迁移逐字同形**（`strftime('%Y-%m-%dT%H:%M:%fZ','now')`， +// UTC 毫秒 + 字面量 Z）：定宽 ⇒ 字典序比较等价于时间序，时间窗筛选因此能直接 +// 吃上 `idx_channel_events_created_at`（与 `runtime/model` 的 `llm/usage` 同口径）。 + +import ( + "context" + "database/sql" + "fmt" + "io" + "log/slog" + "strings" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +const ( + // DefaultFlushInterval 是 `CHANNEL_EVENT_COUNTER_FLUSH` 的默认值（design/26 §4.3）： + // 内存计数每 30s 落盘一次。30s 是「心跳不把库写胖」与「crash 后还能答上近期 + // 有没有推」之间的取舍值。 + DefaultFlushInterval = 30 * time.Second + + // DefaultMaxPendingKeys 是计数缓冲的**键数上界**（design/26 §4.3 的背压）： + // 不同 `(platform, account_id, platform_type)` 键超过它就整批落盘。有了它， + // 内存占用与「DB 长时间不可用」时长脱钩 —— 绝不无界增长。 + DefaultMaxPendingKeys = 512 + + // DefaultListLimit 是 List 的缺省页大小；MaxListLimit 是它的硬上界 + // （超过即钳住，绝不静默返回全表）。 + DefaultListLimit = 100 + MaxListLimit = 500 + + // sqliteTimeLayout 与迁移 0021 的默认值 `strftime('%Y-%m-%dT%H:%M:%fZ','now')` + // 逐字同形（UTC、毫秒、字面量 Z）。写入与解析共用它，使「我们写进去的」与 + // 「SQLite 默认生成的」可以直接比较（与 `documents` 同一手法）。 + sqliteTimeLayout = "2006-01-02T15:04:05.000" + + // eventsLogSource 是全部日志的 source 标签（AGENTS §10：禁无来源裸日志）。 + eventsLogSource = "runtime/events" +) + +// Config 是事件存储的接线参数。时间与开关口径由调用方（组合根）给出：本包**不读 +// 环境变量**、不猜数据目录布局（与 `documents.StoreConfig` / `documents.GCConfig` +// 同一理由）。 +type Config struct { + // DB 是活连接（组合根注入既有 SQLite 连接池；测试注入临时库）。为 nil 时 + // NewStore 返回 ErrNilDB —— 计数的 fail-visible 降级（§4.3「store 为 nil ⇒ + // 只记 DEBUG + 内存计数」）由桥接层（E3）负责，不在本包伪造替身。 + DB *sql.DB + // Log 是基础 logger；nil 时替换为标记来源 runtime/events 的丢弃 logger。 + Log *slog.Logger + // Now 是可注入时钟（测试用）；nil 时用 time.Now。 + Now func() time.Time + // FlushInterval 是内存计数的定期落盘间隔（`CHANNEL_EVENT_COUNTER_FLUSH`）。 + // **≤0 取 DefaultFlushInterval**（与 GC 不同：计数 flush 不是破坏性动作， + // 「配了 0 就永远不落盘」只会悄悄丢掉计数证据，因此这里不做「显式关闭」）。 + FlushInterval time.Duration + // MaxPendingKeys 是计数缓冲的键数上界；≤0 取 DefaultMaxPendingKeys。 + MaxPendingKeys int + // GCInterval 是保留期清扫的间隔（`CHANNEL_EVENT_GC_INTERVAL`）。 + // **≤0 = 关闭 GC**（含启动期清扫），默认 DefaultGCInterval。 + GCInterval time.Duration + // Retention 是 `channel_events` 的保留期（`CHANNEL_EVENT_RETENTION`）。 + // **≤0 = 关闭 GC**，默认 DefaultRetention。0 刻意**不**解释成「立即清扫」： + // 那会让一次笔误把整张事件表清空 —— 而它正是「上游推过没有」的唯一证据。 + Retention time.Duration +} + +// SQLStore 是基于 SQLite 的事件存储实现。它可安全并发使用：连接池本身串行化写 +// （`db.Open` 的 SetMaxOpenConns(1)），内存计数缓冲由自身互斥量保护。 +type SQLStore struct { + db *sql.DB + log *slog.Logger + now func() time.Time + counters *counterBuffer + flushInterval time.Duration + gcInterval time.Duration + retention time.Duration + gcEnabled bool +} + +// NewStore 构造事件存储。唯一的失败是 DB 为 nil（ErrNilDB）；GC 的关闭**不是** +// 构造失败，它是可见的正常状态（Enabled() 为 false + 一条记明原因的 WARN）。 +func NewStore(cfg Config) (*SQLStore, error) { + if cfg.DB == nil { + return nil, fmt.Errorf("events: new store: %w", ErrNilDB) + } + log := sourcedLog(cfg.Log) + now := cfg.Now + if now == nil { + now = time.Now + } + flush := cfg.FlushInterval + if flush <= 0 { + flush = DefaultFlushInterval + } + s := &SQLStore{ + db: cfg.DB, + log: log, + now: now, + counters: newCounterBuffer(cfg.MaxPendingKeys), + flushInterval: flush, + gcInterval: cfg.GCInterval, + retention: cfg.Retention, + } + switch { + case cfg.GCInterval <= 0: + log.Warn("channel event gc disabled: interval <= 0", "interval", cfg.GCInterval.String()) + case cfg.Retention <= 0: + log.Warn("channel event gc disabled: retention <= 0", "retention", cfg.Retention.String()) + default: + s.gcEnabled = true + } + log.Info("channel event store ready", + "flush_interval", s.flushInterval.String(), + "max_pending_keys", s.counters.max, + "gc_enabled", s.gcEnabled, + "gc_interval", s.gcInterval.String(), + "retention", s.retention.String()) + return s, nil +} + +// sourcedLog 把基础 logger 标上 source=runtime/events；nil 时给出丢弃 logger， +// 使调用方永远可以安全记日志（与 `documents` / `history` 同一手法）。 +func sourcedLog(l *slog.Logger) *slog.Logger { + if l == nil { + l = slog.New(slog.NewTextHandler(io.Discard, nil)) + } + return logger.WithSource(l, eventsLogSource) +} + +// FlushInterval 返回生效的计数落盘间隔（组合根的常驻 ticker 用它）。 +func (s *SQLStore) FlushInterval() time.Duration { + if s == nil { + return 0 + } + return s.flushInterval +} + +// PendingKeys 返回内存里待落盘的计数的键数。它有上界（MaxPendingKeys），因此可以 +// 直接作为健康指标暴露给面板：只涨不落 = DB 出了问题。 +func (s *SQLStore) PendingKeys() int { + if s == nil { + return 0 + } + return s.counters.len() +} + +// Record 记录一帧事件（design/26 §4.3 的唯一写入口）。 +// +// 判定顺序刻意固定：**先校验 → 再归一化 → 再按档落行 → 最后计数**。这样即便逐条 +// 落行失败（返回错误），调用方也能明确知道「这一帧没能留下可查痕迹」，而不会拿到 +// 一个「成功了但其实什么都没落」的假象。 +// +// 计数侧的错误**不上抛**：计数是尽力而为（§4.3），绝不因为计数落盘失败而让入站 +// 流水线失败或丢帧。缓冲满了先整批落盘，失败则记 WARN。 +func (s *SQLStore) Record(ctx context.Context, ev Event) error { + if strings.TrimSpace(ev.Platform) == "" { + return fmt.Errorf("events: record: %w", ErrNoPlatform) + } + ev = s.normalize(ev) + if persistsAsRow(ev.Tier, ev.Category) { + if err := s.insertEvent(ctx, ev); err != nil { + return fmt.Errorf("events: record 落行: %w", err) + } + } + s.recordCounter(ctx, ev) + return nil +} + +// normalize 补齐一条待写事件：时钟零值填当前时刻；tier 取值非法时归一为 TierT1 +// 并记 WARN —— 这是 fail-safe（§1.5）：看不懂的档**只**降级成「只计数」，绝不让 +// 一个字段写错演变成「这一帧什么都没有留下」。 +func (s *SQLStore) normalize(ev Event) Event { + switch ev.Tier { + case TierT1, TierT2, TierT3: + default: + s.log.Warn("channel event tier invalid, falling back to t1", + "platform", ev.Platform, "platform_type", ev.PlatformType, "tier", string(ev.Tier)) + ev.Tier = TierT1 + } + if ev.CreatedAt.IsZero() { + ev.CreatedAt = s.now() + } + return ev +} + +// persistsAsRow 判定一帧是否要逐条落 `channel_events`（design/26 §4.1/§4.3/§3.6/§3.7）： +// +// - T2 / T3：逐条（有审计价值，丢不起）； +// - unknown / decode_error：逐条（§3.7「落库 + 计数 + 脱敏结构留痕」，且 +// **绝不进上下文**——兜底不等于放行）； +// - link：逐条（§3.6 明写「落 channel_events（category=link）」；链路事件是 +// §0.2 事故里「当时链路活没活」的唯一答案，即便它是 T1 也不能只留一个计数）。 +// +// 它是纯函数（按 AGENTS §10 不引入包级可变状态）。 +func persistsAsRow(tier Tier, category Category) bool { + switch tier { + case TierT2, TierT3: + return true + } + switch category { + case CategoryUnknown, CategoryDecodeError, CategoryLink: + return true + default: + return false + } +} + +// insertEventSQL 是逐条落行的唯一语句：列名显式列出（禁 `SELECT *` 的同一条纪律 +// 在写侧的表现），值全部参数化。updated_at 与 created_at 同值：事件行不可变， +// 两列同形使「读回时的 updated_at」不会比事实更晚。 +const insertEventSQL = `INSERT INTO channel_events + (platform, account_id, tier, category, platform_type, channel_id, channel_type, + group_id, actor_id, actor_role, target_id, self_related, message_id, summary, + created_at, updated_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + +// insertEvent 写一行 `channel_events`。 +func (s *SQLStore) insertEvent(ctx context.Context, ev Event) error { + stamp := sqliteTimeStamp(ev.CreatedAt) + if _, err := s.db.ExecContext(ctx, insertEventSQL, + ev.Platform, ev.AccountID, string(ev.Tier), string(ev.Category), ev.PlatformType, + ev.ChannelID, ev.ChannelType, ev.GroupID, ev.ActorID, ev.ActorRole, ev.TargetID, + boolToInt(ev.SelfRelated), ev.MessageID, ev.Summary, stamp, stamp); err != nil { + return fmt.Errorf("events: 落事件行(platform=%s type=%s): %w", ev.Platform, ev.PlatformType, err) + } + return nil +} + +// Count 统计符合条件的事件行数。 +func (s *SQLStore) Count(ctx context.Context, f Filter) (int64, error) { + where, args := f.where() + var n int64 + if err := s.db.QueryRowContext(ctx, "SELECT count(*) FROM channel_events"+where, args...).Scan(&n); err != nil { + return 0, fmt.Errorf("events: 统计事件行: %w", err) + } + return n, nil +} + +// listEventsSQL 的列显式列出（与迁移 0021 同序），便于与 scanEvent 对齐核对。 +const listEventsSQL = `SELECT id, platform, account_id, tier, category, platform_type, + channel_id, channel_type, group_id, actor_id, actor_role, target_id, self_related, + message_id, summary, created_at, updated_at +FROM channel_events` + +// List 按 `created_at DESC, id DESC`（倒序，design/26 §4.5）返回符合条件的事件行。 +// id 参与排序是为了让同一毫秒内的多行有确定顺序，分页因此不会重复或漏行。 +func (s *SQLStore) List(ctx context.Context, f Filter) ([]Event, error) { + where, args := f.where() + limit, offset := f.pageLimit(), f.pageOffset() + query := listEventsSQL + where + " ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?" + args = append(args, limit, offset) + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("events: 查事件行: %w", err) + } + defer func() { _ = rows.Close() }() + + out := make([]Event, 0, limit) + for rows.Next() { + ev, err := scanEvent(rows) + if err != nil { + return nil, err + } + out = append(out, ev) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("events: 遍历事件行: %w", err) + } + return out, nil +} + +// scanEvent 读出一行 `channel_events`。时间戳读不出来时返回错误而不是零值时间： +// 一个读不懂的时间戳意味着这一行被谁改坏了，「如实报错」比「静默给一个 1970 年」 +// 更接近 design/26 的诚实口径（与 `documents` GC 的「读不懂的时间绝不当成已超期」 +// 同一个保守方向）。 +func scanEvent(rows *sql.Rows) (Event, error) { + var ( + ev Event + tier, category string + selfRelated int64 + created, updated string + ) + if err := rows.Scan(&ev.ID, &ev.Platform, &ev.AccountID, &tier, &category, &ev.PlatformType, + &ev.ChannelID, &ev.ChannelType, &ev.GroupID, &ev.ActorID, &ev.ActorRole, &ev.TargetID, + &selfRelated, &ev.MessageID, &ev.Summary, &created, &updated); err != nil { + return Event{}, fmt.Errorf("events: 读事件行: %w", err) + } + ev.Tier = Tier(tier) + ev.Category = Category(category) + ev.SelfRelated = selfRelated != 0 + at, ok := parseSQLiteTime(created) + if !ok { + return Event{}, fmt.Errorf("events: 事件行 %d 的 created_at 不可解析(%q): %w", ev.ID, created, ErrBadTimestamp) + } + ev.CreatedAt = at + if ts, ok := parseSQLiteTime(updated); ok { + ev.UpdatedAt = ts + } else { + ev.UpdatedAt = at + } + return ev, nil +} + +// where 把 Filter 编译成参数化的 WHERE 子句（列名是常量，值全部走占位符）。 +// 有意的取舍：**只拼列名与运算符，绝不拼值**（SQL 全参数化，AGENTS §10）。 +func (f Filter) where() (string, []any) { + clauses := make([]string, 0, 6) + args := make([]any, 0, 6) + if platform := strings.TrimSpace(f.Platform); platform != "" { + clauses = append(clauses, "platform = ?") + args = append(args, platform) + } + if f.Tier != "" { + clauses = append(clauses, "tier = ?") + args = append(args, string(f.Tier)) + } + if f.Category != "" { + clauses = append(clauses, "category = ?") + args = append(args, string(f.Category)) + } + if channelID := strings.TrimSpace(f.ChannelID); channelID != "" { + clauses = append(clauses, "channel_id = ?") + args = append(args, channelID) + } + if !f.Since.IsZero() { + clauses = append(clauses, "created_at >= ?") + args = append(args, sqliteTimeStamp(f.Since)) + } + if !f.Until.IsZero() { + clauses = append(clauses, "created_at <= ?") + args = append(args, sqliteTimeStamp(f.Until)) + } + if len(clauses) == 0 { + return "", nil + } + return " WHERE " + strings.Join(clauses, " AND "), args +} + +// pageLimit 把 Limit 归一：≤0 取缺省、超过硬上界即钳住。 +func (f Filter) pageLimit() int { + switch { + case f.Limit <= 0: + return DefaultListLimit + case f.Limit > MaxListLimit: + return MaxListLimit + default: + return f.Limit + } +} + +// pageOffset 把 Offset 归一到 ≥0。 +func (f Filter) pageOffset() int { + if f.Offset < 0 { + return 0 + } + return f.Offset +} + +// sqliteTimeStamp 把时刻格式化成库里的时间戳口径（UTC 毫秒 + 字面量 Z）。 +func sqliteTimeStamp(t time.Time) string { + return t.UTC().Format(sqliteTimeLayout) + "Z" +} + +// parseSQLiteTime 解析库里的时间戳列。它先试本包写出的口径，再退回 RFC3339 +// （覆盖带偏移或非毫秒精度的历史值）；两者都不成立时返回 false，调用方据此保守 +// 处理（绝不把读不懂的时间当成一个真实时刻）。 +func parseSQLiteTime(raw string) (time.Time, bool) { + trimmed := strings.TrimSpace(raw) + if trimmed == "" { + return time.Time{}, false + } + if t, err := time.Parse(sqliteTimeLayout+"Z", trimmed); err == nil { + return t.UTC(), true + } + if t, err := time.Parse(time.RFC3339Nano, trimmed); err == nil { + return t.UTC(), true + } + return time.Time{}, false +} + +// boolToInt 把布尔列写成 SQLite 的 0/1（列本身是 INTEGER NOT NULL DEFAULT 0）。 +func boolToInt(v bool) int { + if v { + return 1 + } + return 0 +} diff --git a/internal/runtime/events/store_test.go b/internal/runtime/events/store_test.go new file mode 100644 index 0000000..b461fe9 --- /dev/null +++ b/internal/runtime/events/store_test.go @@ -0,0 +1,447 @@ +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)) + } +}