Cyrene_v2/internal/services/channelbridge/events.go

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 是按 `"<platform>/<platform_type>"` 聚合的帧数快照（与计数表的键同形）。
	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"
	}
}