package channelbridge // 本文件承载**渠道事件留存**在桥接层的唯一接线点(design/26 §4/§5;切片 E3): // 把适配器归一化出来的结构化事件**无条件**写进 `Options.EventStore`(T2/T3 与 unknown // 逐条落行、T1 只累加计数),两轴各归各位 —— 留存档(本文件按归一化类别 + SelfRelated // 映射,§2.3)与上下文档(notice.go 的 preEntryReview,既有)**互不决定**。 // // 三条不变量写在这里: // // 1. **留存无条件**:留存与「进不进上下文」是两个独立动作(§4.3)。因此本层在任何 // 早退分支(仅登记 / fail-safe 静默 / 升级真会话)**之前**记账,且**绝不**决定 loop // 走不走、也绝不改写消息内容。 // 2. **绝不静默,也绝不 fail-crash**(§1.1/§4.3):未接线(store 为 nil)或落库失败 // 都只降级为「内存计数 + 日志」——**绝不**让入站流水线失败、丢帧或 panic。 // 这正是 §0.2 事故要的:看不见可以,但必须留得下痕、答得出「上游推过没有」。 // 3. **脱敏唯一出处**(§4.2):摘要一律过 `adapter.SanitizeNoticeText`。`events.Event` // 里**没有**承载平台正文 / 媒体 URL / OneBot `flag` 的字段,因此凭据不可能随这条 // 链路流出去(§9.2 第 8 条:连结构都没有那些字段)。 // // 类别命名归 `adapter`(`adapter.NoticeCategory*`),本层**逐字搬运**、绝不改名 // (§9.2 第 7 条:两轴常量分居两包,防两表漂移)。 import ( "context" "log/slog" "strings" "sync" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" ) // 摘要的固定文案(design/26 §4.2 的「能落」清单:一句话常量,**不含任何平台原文**)。 const ( // heartbeatSummary / lifecycleSummary / selfEchoSummary 覆盖 T1 三类刻意留空 // SafeText 的类别(§9.2 第 9 条:适配器留空,摘要由**留存侧**按平台原生类型生成)。 heartbeatSummary = "心跳(链路遥测)" lifecycleSummary = "生命周期(链路遥测)" selfEchoSummary = "她自己发出消息的回声" // decodeErrorSummary 对应 §3.7 的「JSON 坏」:把「解析失败」这件事本身说清楚。 decodeErrorSummary = "事件载荷解析失败,只留结构痕迹" // unknownSummaryPrefix 后面拼**平台原生类型码**(notice_type / extra.type / // Telegram 的 update 键名):那是「上游到底报了什么类型」的答案(§2.4), // 而且它是平台枚举值,不是用户文本。 unknownSummaryPrefix = "未识别的事件类型:" unknownSummary = "未识别的事件类型" // platformEventSummary 是类别与种类都归不了类的兜底文案(既非未知类型、也无安全文本)。 platformEventSummary = "平台系统事件" ) // recordEvent 是渠道事件留存的**唯一插入点**(design/26 §5.1/§6.2):在入站归一化 // 之后、任何早退分支之前调用一次,把一条结构化事件交给 `Options.EventStore`。 // // 它只影响**留存面**:不改写 msg、不决定 loop 走不走、不返回值 —— 调用方照旧走既有 // 流水线(仅登记 / 限定注入 / 升级真会话)。普通聊天消息(非 ContentAction)在这里 // 直接返回:主消息路径**零副作用**。 func (b *Bridge) recordEvent(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter, out noticeOutcome) { if !out.Applicable { return } ev := b.buildEvent(msg, a, out) if b.eventStore == nil { // 未接线(§4.3 的 fail-visible 降级):内存计数 + **首次**一条 DEBUG, // 绝不中断入站、绝不丢帧。 b.noteUnrecorded(ev, "no event store wired") return } if err := b.eventStore.Record(ctx, ev); err != nil { // 逐条留存的失败**必须可见**(§4.3):WARN + 内存计数。但既**不**中断入站, // 也**不**关掉留存面 —— 下一帧照旧尝试。 b.eventLog.Warn("channel event not persisted", "platform", ev.Platform, "platform_type", ev.PlatformType, "tier", string(ev.Tier), "category", string(ev.Category), "err", err) b.noteUnrecorded(ev, "record failed") return } b.logEventRecorded(ctx, ev, out) } // buildEvent 把归一化结果投影成一条待留存事件(design/26 §4.1 的列,逐列对应)。 // // 投影**只取结构化字段**:平台 / 账号 / 会话 id、归一化类别与平台原生类型、当事人 id、 // 是否命中她自己、关联消息 id,以及一句由安全文本(T2/T3)或固定文案(T1)得来的摘要。 // CreatedAt 刻意留零值:由存储层填**我们的接收时刻**(适配器给的平台时刻可能是 0 ⇒ // 1970,一条 1970 的行会被保留期立刻扫掉,反而丢掉刚发生的证据)。 func (b *Bridge) buildEvent(msg adapter.UnifiedMessage, a adapter.PlatformAdapter, out noticeOutcome) events.Event { notice := out.Notice return events.Event{ Platform: msg.Platform, AccountID: accountIDFromAdapter(a), Tier: eventTier(notice), Category: eventCategory(notice), PlatformType: notice.PlatformType, ChannelID: msg.ChannelID, ChannelType: string(msg.ChannelType), GroupID: msg.GroupID, ActorID: notice.Actor.UserID, ActorRole: b.eventActorRole(msg.Platform, notice.Actor.UserID), TargetID: notice.Target.UserID, SelfRelated: notice.SelfRelated, MessageID: firstNonEmptyStr(notice.RefMessage, msg.MessageID), Summary: eventSummary(notice), } } // logEventRecorded 是这一帧成功进入留存面后的**结构级 DEBUG**(不留存则走 noteUnrecorded)。 // // 它记的是 §9.2 第 2 条要求的那条日志:**NoticeClass 的裁决结果只进日志、不入事件行** // (§4.1 的 DDL 本就没有该列),连留存档一起记,方便回答「这条为什么没进她上下文」。 // 只在真的开了 DEBUG 时才组装字段(与前缀一致,默认 INFO 下零开销), // 且**绝不**记 SafeText 之外的平台文本,更绝不记 flag / 正文 / URL。 func (b *Bridge) logEventRecorded(ctx context.Context, ev events.Event, out noticeOutcome) { if !b.eventLog.Enabled(ctx, slog.LevelDebug) { return } b.eventLog.Debug("channel event recorded", "platform", ev.Platform, "platform_type", ev.PlatformType, "category", string(ev.Category), "tier", string(ev.Tier), "self_related", ev.SelfRelated, "notice_class", noticeClassName(out.Class), "notice_reason", out.Reason) } // noteUnrecorded 是「这一帧没能进入留存面」的唯一收口(未接线 / 落库失败): // 内存计数**永远**累加(fail-visible:有数可查),日志只在首帧记一条 DEBUG —— // 心跳每 ~15s 一帧,逐帧刷日志只会把日志淹没(AGENTS §10 的压测纪律:有界、不刷屏)。 func (b *Bridge) noteUnrecorded(ev events.Event, reason string) { stats := b.eventFallback.append(ev.Platform, ev.PlatformType) if stats.Frames != 1 { return } b.eventLog.Debug("channel event not recorded, counting in memory instead", "reason", reason, "platform", ev.Platform, "platform_type", ev.PlatformType, "tier", string(ev.Tier), "category", string(ev.Category)) } // EventFallbackStats 是「没能进入留存面」的计数快照(design/26 §4.3 的降级面)。 type EventFallbackStats struct { // Keys 是按 `"/"` 聚合的帧数快照(与计数表的键同形)。 Keys map[string]int64 // Frames 是**未进入留存面**的总帧数(含下面溢出键的那部分)。 Frames int64 // OverflowKeys 是键上界用尽后未能按类记账的帧数:它们仍算进 Frames, // 只是不再有各自的分类(键数**有界**,绝不无界增长)。 OverflowKeys int64 } // EventFallback 返回降级计数的快照。未接线(Options.EventStore 为 nil)或落库失败时 // 才有内容 —— 它是「有多少帧没留下可查痕迹」的**唯一**答案:面板与排障据此知道留存面 // 当时是活的还是降级的(design/26 §1.1 的诚实边界)。 func (b *Bridge) EventFallback() EventFallbackStats { return b.eventFallback.snapshot() } // eventFallback 是未接线 / 落库失败时的内存计数面(§4.3 的降级)。 // // 键数上界复用 `events.DefaultMaxPendingKeys`(与存储层的内存计数缓冲同界:绝不因为 // 一直失败而无界增长),键口径与计数表的唯一键同形(平台 + 平台原生类型;不含账号 // —— 降级面连账号归属都未必取得准,键越简单越不会误导)。 type eventFallback struct { mu sync.Mutex keys map[string]int64 frames int64 overflow int64 } // newEventFallback 构造空的内存计数面。 func newEventFallback() *eventFallback { return &eventFallback{keys: make(map[string]int64)} } // append 累加一帧并返回**当前**快照(日志据此判断「是不是首帧」)。 func (f *eventFallback) append(platform, platformType string) EventFallbackStats { key := platform + "/" + platformType f.mu.Lock() defer f.mu.Unlock() f.frames++ if _, ok := f.keys[key]; ok { f.keys[key]++ } else if len(f.keys) < events.DefaultMaxPendingKeys { f.keys[key] = 1 } else { f.overflow++ } return f.snapshotLocked() } // snapshot 返回快照(拷贝键值,调用方拿到的是私有副本)。 func (f *eventFallback) snapshot() EventFallbackStats { f.mu.Lock() defer f.mu.Unlock() return f.snapshotLocked() } // snapshotLocked 是 snapshot 的加锁内核(append 已在锁内,直接复用)。 func (f *eventFallback) snapshotLocked() EventFallbackStats { keys := make(map[string]int64, len(f.keys)) for k, v := range f.keys { keys[k] = v } return EventFallbackStats{Keys: keys, Frames: f.frames, OverflowKeys: f.overflow} } // eventTier 是 design/26 §2 的两轴换算:**归一化类别 → 留存档**(§2.3 的换算表)。 // // 判据顺序刻意固定为「先按类别定档、再按 self 命中升档」: // // 1. 类别(缺类别时按 Kind)给出基础档; // 2. `SelfRelated`(结构化 id 命中她自己 / 她的账号)对**平台事件类**做 T1 → T2 升档 // (§2.2:T1 → T2、T2 → 高价值;T3 不因 self 命中改变档位); // 3. **帧级链路遥测不升档**(见 isFrameTelemetry):见那里的取舍说明。 func eventTier(notice adapter.SystemNotice) events.Tier { tier := baseEventTier(notice) if tier != events.TierT1 || !notice.SelfRelated { return tier } if isFrameTelemetry(notice.Category) { return events.TierT1 } return events.TierT2 } // baseEventTier 给出归一化类别的基础留存档(design/26 §3 逐平台映射表收敛后的共同口径)。 // // 类别为零值时按 Kind 兜底 —— `adapter.SystemNotice.Category` 的注释明写「零值表示 // 适配器未给类别,留存侧按 Kind 兜底」。 // // default 分两支刻意都取**更全的留存**(T2 = 逐条落行):新增类别漏登记时会多落行, // 绝不静默丢帧(design/26 §1.1)。E5 新增平台类别时**必须**在这里补档位。 func baseEventTier(notice adapter.SystemNotice) events.Tier { if notice.Category == "" { return kindTier(notice.Kind) } switch notice.Category { case adapter.NoticeCategoryHeartbeat, adapter.NoticeCategoryLifecycle, adapter.NoticeCategoryMessageSent, adapter.NoticeCategoryClientStatus, adapter.NoticeCategoryUnknown, adapter.NoticeCategoryDecodeError: return events.TierT1 case adapter.NoticeCategoryPoke, adapter.NoticeCategoryLuckyKing: return events.TierT3 default: // join / friend_add / leave / recall / admin / ban / ban_all / upload / essence / // card / offline_file / honor / title / friend_request / group_request:一律 T2。 return events.TierT2 } } // kindTier 是类别缺失时按 Kind 兜底(与 §3 的逐条映射同口径):成员 / 权限 / 消息状态的 // 变动与请求一律 T2,纯遥测 T1,定向交互 T3。default 取 T1 —— 此时类别会兜底成 // `unknown`,而 `unknown` 在存储层本就是**逐条落行**的(§3.7),留痕一条不少。 func kindTier(kind adapter.NoticeKind) events.Tier { switch kind { case adapter.NoticeInteraction: return events.TierT3 case adapter.NoticeJoin, adapter.NoticeLeave, adapter.NoticeRecall, adapter.NoticeBotAdd, adapter.NoticeRequest: return events.TierT2 default: return events.TierT1 } } // isFrameTelemetry 报告该类别是否属于**帧级链路遥测**:心跳(~15s 一帧)与生命周期。 // 这两类的 self 命中是**恒定**的(帧帧都命中 self_id),所以按 §2.2 升档等于「每帧写一行 // channel_events」—— 而 §4.3 明令「心跳 / 在线状态逐帧写库是纯浪费」(计数表存在的理由), // §3.1 给这两类的档位也正是 T1。 // // 🔴 **`message_sent`(她自己发出消息的回声)不在此列**(切 M13;原 M12 登记的缺口): // 它**不是**帧级遥测 —— 一帧对应她真发出的一条消息,回声带的是那条出站消息的 id // (`RefMessage`)。此前把它一并豁免,会让 self 命中**不升档** ⇒ 整条回声只进计数, // `message_id` 被丢,「她自己最近发了哪些消息」在库里查不到(M12 的 A 层出站 id 因此 // 断在这里)。摘出后它按 §2.2 正常 T1 → T2 逐条落行。 // // ⚠️ **由此登记的债**(design/26 §2.6「她自己账号的在线 / 离线」= T2 高价值): // 那是**状态迁移**(在线→离线 / 离线→在线),不是每一帧。判定它需要迁移信号 // (heartbeat 的 `status.online` / lifecycle 的 `sub_type`),而 `adapter.SystemNotice` // 目前没有投影这两个字段(E2 定稿,本片不许改 plugins/**),`qq_obv11/probe.go` 的 // 主动探活也还没有留存入点。本片只能保证**每一帧都进了计数**(「上游推过没有」答得上), // 迁移事件(T2)留待后续切片;已在切片报告「未做 / 存疑」中列明。 func isFrameTelemetry(cat adapter.NoticeCategory) bool { switch cat { case adapter.NoticeCategoryHeartbeat, adapter.NoticeCategoryLifecycle: return true default: return false } } // eventCategory 把适配器的归一化类别投影到 events 的类别轴。 // // 两轴常量分居两包(§9.2 第 7 条):**类别的命名归 adapter**,events 只承载类型与存储。 // 这里逐字搬运、绝不改名 —— adapter 的类别比 events 的常量集**更细一档** // (ban_all / honor / title / lucky_king …),改名只能靠丢信息来对齐,而丢掉的正是 // 「这条为什么没进她上下文」的答案。适配器未给类别(零值)时兜底成 `unknown`: // 「读不出具体类别」本身就该如实记为未知(存储层对 `unknown` **逐条落行**,§3.7)。 func eventCategory(notice adapter.SystemNotice) events.Category { cat := strings.TrimSpace(string(notice.Category)) if cat == "" { return events.CategoryUnknown } return events.Category(cat) } // eventSummary 给出一条事件的**脱敏结构级摘要**(§4.1 的 summary 列)。 // // 适配器给了安全文本就用它(T2/T3 的固定文案,如「群文件上传」);T1 四类刻意留空 // (§9.2 第 9 条)⇒ 由本函数按类别 / 平台原生类型**生成固定文案**。 // // 净化唯一出处是 `adapter.SanitizeNoticeText`(§4.2)。即便 SafeText 已在 // preEntryReview 里净化过一遍,这里再过一次:保证「Event.Summary 恒已净化」与调用 // 顺序无关(该函数幂等)。 func eventSummary(notice adapter.SystemNotice) string { text := strings.TrimSpace(notice.SafeText) if text == "" { text = fallbackSummary(notice) } return adapter.SanitizeNoticeText(text) } // fallbackSummary 按归一化类别生成固定文案(T1 那几类 SafeText 恒为空)。 // 只描述「这是什么类型的事件」,绝不夹带平台原文 / URL / 凭据;未知类型拼的是 // **平台原生类型码**(平台枚举值,回答「上游报了哪种类型」)。 func fallbackSummary(notice adapter.SystemNotice) string { switch notice.Category { case adapter.NoticeCategoryHeartbeat: return heartbeatSummary case adapter.NoticeCategoryLifecycle: return lifecycleSummary case adapter.NoticeCategoryMessageSent: return selfEchoSummary case adapter.NoticeCategoryDecodeError: return decodeErrorSummary case adapter.NoticeCategoryUnknown: if notice.PlatformType != "" { return unknownSummaryPrefix + notice.PlatformType } return unknownSummary default: return kindSummary(notice.Kind) } } // kindSummary 是类别与安全文本都缺时的兜底文案(按 Kind 说清是哪一类系统事件)。 func kindSummary(kind adapter.NoticeKind) string { switch kind { case adapter.NoticeJoin: return "有成员加入了会话" case adapter.NoticeLeave: return "有成员离开了会话" case adapter.NoticeRecall: return "有人撤回了一条消息" case adapter.NoticeTyping: return "有会话正在输入" case adapter.NoticeBotAdd: return "她的账号成员身份发生变更" case adapter.NoticeRequest: return "有一条好友 / 入群请求" case adapter.NoticeInteraction: return "有一个定向互动事件" case adapter.NoticeUnknown: return unknownSummary default: return platformEventSummary } } // eventActorRole 只按**白名单**判定事件当事人的角色(owner.yaml 的账号白名单 + // 每渠道 adminIds 层,与 session.ResolveActor 同源),**绝不看正文、也绝不采信平台 // 上报的 role 字段**(design/26 §1.2 / AGENTS §9 红线:身份只由结构化元数据 + 白名单 // + 策略引擎决定)。 // // 三种取值:没有当事人的事件(心跳 / 链路遥测)来自平台系统本身 ⇒ system;白名单命中 // ⇒ owner;其余 ⇒ **unknown**。刻意不填 member:「不是主人」不等于「是某个群的成员」 // (加好友请求的申请人本就不是任何群的成员),判不出来时如实说判不出来。 func (b *Bridge) eventActorRole(platform, actorID string) string { if strings.TrimSpace(actorID) == "" { return events.RoleSystem } if b.owner.IsOwner(platform, actorID) { return events.RoleOwner } return events.RoleUnknown } // accountIDFromAdapter 从适配器注册名里取配置账号 id("telegram:conf0" → "conf0", // 多账号命名口径见 cmd/cyrene/channels.go)。单账号(名字里没有 ':')时返回空串: // `events.Event.AccountID` 把空串定义为「单账号 / 未知」(§4.1),绝不编一个 id 出来。 func accountIDFromAdapter(a adapter.PlatformAdapter) string { _, accountID, ok := strings.Cut(adapterName(a), ":") if !ok { return "" } return accountID } // noticeClassName 把 `adapter.NoticeClass`(上下文档的裁决结果)渲染成日志里的可读名。 // 它**只进日志、不入事件行**(§9.2 第 2 条):留存档记 tier + category 两列, // 这一轴是另一个正交轴的结论,回答「这条为什么没进她上下文」。 func noticeClassName(class adapter.NoticeClass) string { switch class { case adapter.NoticeRegisterOnly: return "register_only" case adapter.NoticeLimitedInject: return "limited_inject" case adapter.NoticeEscalate: return "escalate" default: return "unknown" } }