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 红线）
405 lines
16 KiB
Go
405 lines
16 KiB
Go
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
|
||
}
|