Files
AskaEth ec3f6e810a feat(events): 渠道事件存储地基(Event/Store + SQLite 实现 + 有界计数聚合 + 保留期 GC + 迁移 0021)
- 新包 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 红线)
2026-10-05 13:47:25 +08:00

265 lines
14 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
// 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)
}