diff --git a/internal/runtime/events/account_status.go b/internal/runtime/events/account_status.go new file mode 100644 index 0000000..a601bc3 --- /dev/null +++ b/internal/runtime/events/account_status.go @@ -0,0 +1,260 @@ +package events + +// 本文件是「**她自己账号的在线 / 离线**」这条高价值事件的**迁移检测与归一化** +// （design/26 §2.6；切片 E7）。 +// +// 它补的是 §2.6 里那条**债**：`channelbridge/events.go` 的 `isFrameTelemetry` 明写 +// 「心跳 / 生命周期逐帧写库是纯浪费（T1 + 只计数）」，而 §2.6 要求的是**状态迁移** +// （在线→离线 / 离线→在线）落一行 T2 —— 当时没有迁移信号可接（`adapter.SystemNotice` +// 没投影 `status.online`），`qq_obv11/probe.go` 的主动探活也只有内存状态 + 一条日志。 +// 本文件把那句话变成可实现的东西：**给它一个状态样本，它回答「这是不是一次迁移」**， +// 迁移就落一条 T2 行（逐条、参与查询面与 GC），没迁移就什么都不做。 +// +// 三条口径写在这里（都有测试钉住，不靠注释自觉）： +// +// 1. **只在迁移时落行**：每 60s 采一次样，只有状态**变了**才落 —— 这既满足 §2.6 +// （迁移是事实），又不违背 §4.3（绝不逐帧写库）。 +// 2. **未知 / 异常一律归一化留痕，不整类丢弃**：样本的另一半取值是 `known=false` +// （尚未探到 / 一直失败），它归一化成 `AccountStateUnknown` 并**照常落一行**； +// 检测通道码为空时归一成固定码，**绝不**因为「字段看着怪」把整类事件丢掉 +// （§1.1「绝不静默」）。这与 `gateway` 的 `liveAccountOnline`（未知即缺席）是 +// **两件不同的事**：那里是「面板要如实显示未知」，这里是「未知也是要留痕的事实」。 +// 3. **身份不在这里判定**：本文件不读白名单、不看正文、不决定权限。行的 +// `actor_role` 恒为 `system`（状态变更由平台侧上报，不是人做出的动作 —— 与 +// `channelbridge` 的 `eventActorRole` 对「无当事人事件」的口径逐字一致）， +// `self_related` 恒为 true（§2.2 的 self 命中正是这次升档的依据）。 + +import ( + "context" + "fmt" + "strings" + "sync" + "time" +) + +// AccountState 是账号在线状态的**归一化三态**（§2.6 的「known 区分确知离线与尚未探到」）。 +type AccountState string + +// 三态取值。它们同时也是写进 `summary`（固定文案）的那个词，绝不落平台原文。 +const ( + // AccountStateUnknown 是**尚未探到**：探活刚启动 / 一直失败 / 未接线。它既不是 + // 「在线」也不是「离线」——把未知写成离线，正是 2026-10-05 事故（面板一片绿） + // 的镜像错误，因此它有自己的取值、自己的文案，且**照常落行**。 + AccountStateUnknown AccountState = "unknown" + // AccountStateOnline 是平台侧回复「账号在线」（OneBot `status.online=true`）。 + AccountStateOnline AccountState = "online" + // AccountStateOffline 是**确知离线**（探活成功且 `online=false`）。它与 + // 「我们连不上 OneBot」是两件事（§2.6 第 4 条：连接 ≠ 账号在线）。 + AccountStateOffline AccountState = "offline" +) + +// 检测通道码（写进 `platform_type` 列，§4.1「平台原生类型码原样保留，排障时两者并用」）。 +// +// 两条通道并列（§2.6）：主动探活（本片接线）与 `meta_event` 心跳 / 生命周期 +// （取决于对端是否配置推送；其**逐帧**留存仍由 E2 的 T1 计数承担，迁移事件可复用 +// 本文件 —— 把样本的 Source 换成下面两个常量即可，不需要第二个实现）。 +const ( + // StatusSourceProbe 是 OneBot 标准 action `get_status` 的主动探活（§2.6 的兜底通道）。 + StatusSourceProbe = "get_status" + // StatusSourceHeartbeat 是上游主动推送的 `meta_event.heartbeat`（§2.6 的一手通道）。 + StatusSourceHeartbeat = "meta_event.heartbeat" + // StatusSourceLifecycle 是上游主动推送的 `meta_event.lifecycle`（connect/enable/disable）。 + StatusSourceLifecycle = "meta_event.lifecycle" + // statusSourceFallback 是**通道码缺失**时的归一化取值。它是我们自己的码（不是任何 + // 平台枚举值），因此不会被误读成「上游报了这种类型」；有它，「字段没给」这件事 + // 不会演变成「整条事件被丢掉」。 + statusSourceFallback = "account_status" +) + +// 账号状态迁移事件的固定文案（§4.2：一句话常量，**不含任何平台原文**）。 +// +// 九种 (前态, 后态) 组合**逐条给出**，没有 default 兜底 —— 兜底会让「漏了一种组合」 +// 悄悄退化成一句含糊的话；而这里的每一种组合都对应一个真实可查的事实。前态为空表示 +// 「本进程首次观测到该账号」（也不是「未知」：未知有自己的取值）。 +// +// 表按**调用返回**而不是包级变量（AGENTS §10 禁包级状态；与 `admin_channel_events.go` +// 的 `channelEventPlatforms()` 同一手法）。它每次迁移才构造一次，迁移本就是低频事实， +// 这点分配可以忽略。 +func statusSummaries() map[string]string { + return map[string]string{ + statusSummaryKey("", AccountStateOnline): "账号在线（首次确知）", + statusSummaryKey("", AccountStateOffline): "账号离线（首次确知）", + statusSummaryKey("", AccountStateUnknown): "账号状态未知（尚未探到）", + statusSummaryKey(AccountStateOnline, AccountStateOffline): "账号离线（在线→离线）", + statusSummaryKey(AccountStateOnline, AccountStateUnknown): "账号状态未知（在线→未知）", + statusSummaryKey(AccountStateOffline, AccountStateOnline): "账号上线（离线→在线）", + statusSummaryKey(AccountStateOffline, AccountStateUnknown): "账号状态未知（离线→未知）", + statusSummaryKey(AccountStateUnknown, AccountStateOnline): "账号上线（未知→在线）", + statusSummaryKey(AccountStateUnknown, AccountStateOffline): "账号离线（未知→离线）", + } +} + +// statusSummaryKey 拼出一个迁移在文案表里的键（前后态之间用一个固定分隔符）。 +func statusSummaryKey(prev, next AccountState) string { + return string(prev) + "→" + string(next) +} + +// StatusSample 是一次**账号状态观测**：谁（平台 + 账号）在哪个通道上被看到处于什么状态。 +// +// 它是本文件唯一的输入形状，刻意与适配器的既有能力面同构 +// （`AccountStatus() (online, known bool)`，见 `admin_channel_status.go` 的 +// `accountStatusAware`），因此采样侧（网关）不需要发明第二套语义。 +type StatusSample struct { + // Platform 是来源平台（platform 常量之一）。**必填**，否则 Observe 返回 ErrNoPlatform。 + Platform string + // AccountID 是多账号标识；空 = 单账号 / 未知（§4.1 的列口径，绝不编一个 id 出来）。 + AccountID string + // Online 是平台侧报告的在线位。**只有 Known 为真时才有意义**（OneBot 的 + // `status.online`）：Known 为假时它被忽略，绝不拿来当「离线」。 + Online bool + // Known 区分「确知」与「尚未探到」——它是三态而不是二态的**唯一**来源。 + Known bool + // Source 是检测通道码（StatusSource* 常量）。空时归一成 statusSourceFallback， + // **不报错**：一个字段缺失不该让整类事件消失（§1.1）。非白名单取值原样保留： + // 它是排障用的平台原生码（§4.1），在这里做封闭校验只会让将来的新通道被拒。 + Source string +} + +// state 把一次观测归一化成三态（归一化规则见 StatusSample 的字段注释）。 +// 唯一会失败的情形是**没有平台**：缺了它这一行既查不到也归不了类（与 Store.Record +// 的 ErrNoPlatform 同一判据，本函数只是更早地把错误交给调用方）。 +func (s StatusSample) state() (AccountState, error) { + if strings.TrimSpace(s.Platform) == "" { + return "", fmt.Errorf("events: 账号状态样本缺少来源平台: %w", ErrNoPlatform) + } + switch { + case !s.Known: + return AccountStateUnknown, nil + case s.Online: + return AccountStateOnline, nil + default: + return AccountStateOffline, nil + } +} + +// statusKey 是迁移表的键：平台 + 账号（与计数表的唯一键前两列同形）。 +type statusKey struct { + platform string + accountID string +} + +// StatusResult 是 Observe 的结果。 +type StatusResult struct { + // State 是本次观测归一化后的状态（**无论是否迁移都给出**）。 + State AccountState + // Previous 是上一次观测到的状态；空串表示「本进程首次见到该账号」。 + Previous AccountState + // Changed 报告这**是不是一次迁移**（首次观测也算：它是「第一份事实」）。 + Changed bool + // Event 是归一化后的事件行，仅在 Changed 时有意义 —— 它是落进 `channel_events` + // 的那一行，也是上 bus 推成 WS 帧的载荷（见 ChannelEventFrame）。 + Event Event + // Recorded 报告这一行是否**真的落库成功**。Store 未接线或落库失败时它为 false， + // 此时事件**不得**被当成「已留存的事实」去推给面板（落不下去就不声称）。 + Recorded bool +} + +// StatusTracker 是账号状态迁移的判定与落盘。它**只做状态机**：不采样、不定时、 +// 不推送（那是 `Watcher` 与网关的事），因此可以用一张表把九种迁移逐个钉住。 +// +// 它可安全并发使用；`seen` 只在临界区内读写，落库调用**在锁外**执行（绝不在持锁时 +// 做 I/O —— 一次慢写不该让别的账号的观测排队）。 +type StatusTracker struct { + store Store + now func() time.Time + mu sync.Mutex + seen map[statusKey]AccountState +} + +// NewStatusTracker 构造迁移判定器。store 为 nil 时**照常判定**，只是不落库 +// （§4.3 的 fail-visible 降级由调用方决定怎么呈现；这里绝不 panic、也绝不假装写过）。 +// now 为 nil 时用 time.Now（测试注入固定时钟）。 +func NewStatusTracker(store Store, now func() time.Time) *StatusTracker { + if now == nil { + now = time.Now + } + return &StatusTracker{store: store, now: now, seen: make(map[statusKey]AccountState)} +} + +// Observe 收一次观测：归一化 → 判迁移 → 迁移则落一行 T2（并在结果里带回那行）。 +// +// 返回错误的两种情形都是**调用方该看见的**：样本缺平台（ErrNoPlatform 的包装）、 +// 或者迁移行**落库失败**（此时 Recorded 为 false，调用方据此不推帧、只记 WARN）。 +// 非迁移观测返回零错误 —— 「状态没变」是正常结果，不是异常。 +func (t *StatusTracker) Observe(ctx context.Context, s StatusSample) (StatusResult, error) { + state, err := s.state() + if err != nil { + return StatusResult{}, err + } + key := statusKey{platform: strings.TrimSpace(s.Platform), accountID: s.AccountID} + t.mu.Lock() + prev, seen := t.seen[key] + if prev != state { + t.seen[key] = state + } + t.mu.Unlock() + + res := StatusResult{State: state, Previous: prev, Changed: !seen || prev != state} + if !res.Changed { + return res, nil + } + res.Event = statusEvent(key, s, state, prev, t.now()) + if t.store == nil { + return res, nil + } + if err := t.store.Record(ctx, res.Event); err != nil { + return res, fmt.Errorf("events: 落账号状态迁移行: %w", err) + } + res.Recorded = true + return res, nil +} + +// States 返回当前已观测到的账号状态快照（键为「平台 + 账号」）。它给测试与排障用： +// 迁移判定的内部状态可见，但**不会被任何生产路径依赖**。 +func (t *StatusTracker) States() map[string]AccountState { + t.mu.Lock() + defer t.mu.Unlock() + out := make(map[string]AccountState, len(t.seen)) + for k, v := range t.seen { + out[statusKeyString(k)] = v + } + return out +} + +// statusEvent 把一次迁移投影成一条待留存事件（design/26 §4.1 的列，逐列对应）。 +// +// 取值口径（每一条都是「如实」而非「猜」）： +// - `tier = t2`：§2.6 明令「她自己账号的在线 / 离线」是 T2 高价值（留存 + 上行）； +// 也正因它 T2，存储层才**逐条落行**（persistsAsRow），这一行才查得到； +// - `category = lifecycle`：它由生命周期 / 探活派生（§3.1 的 meta_event 行注明了这条 +// 派生关系），不另造新类别； +// - `platform_type` = 检测通道码（get_status / meta_event.*），排障时据此分辨 +// 「这条是探活探到的」还是「上游心跳推来的」； +// - `actor_id` 空 + `actor_role = system`：状态变更来自平台侧，没有人的当事人 —— +// 与 channelbridge 的 eventActorRole 对「无当事人事件」的口径逐字一致； +// - `self_related = true`：命中的正是她自己的账号（§2.2 升档的那条判据）； +// - `channel_id` / `group_id` / `message_id` 留空：账号状态不属任何会话或消息。 +func statusEvent(key statusKey, s StatusSample, next, prev AccountState, now time.Time) Event { + source := strings.TrimSpace(s.Source) + if source == "" { + source = statusSourceFallback + } + summary := statusSummaries()[statusSummaryKey(prev, next)] + return Event{ + Platform: key.platform, + AccountID: key.accountID, + Tier: TierT2, + Category: CategoryLifecycle, + PlatformType: source, + ActorRole: RoleSystem, + SelfRelated: true, + Summary: summary, + CreatedAt: now, + } +} + +// statusKeyString 把内部键拼成 `"/"`（日志与 States 的可读形态， +// 与 channelbridge 的内存降级计数字典同形）。 +func statusKeyString(k statusKey) string { + return k.platform + "/" + k.accountID +} diff --git a/internal/runtime/events/account_status_test.go b/internal/runtime/events/account_status_test.go new file mode 100644 index 0000000..7666979 --- /dev/null +++ b/internal/runtime/events/account_status_test.go @@ -0,0 +1,395 @@ +package events + +// 本文件钉住**账号在线状态迁移**的判定、归一化与落行（design/26 §2.6；切片 E7）。 +// +// 断言的是**库里的事实**（多了一行没有、那一行的 tier/category/platform_type/summary +// 是不是预期的），不按日志文本断言 —— 与 store_test.go 同一口径。 +// +// 覆盖的三类口径，每一类都对应一条设计约束： +// - 九种 (前态 → 后态) 组合**逐个落行**，包括涉及「未知」的那四种（§2.6 的 +// `known` 三态；§1.1「绝不静默」）； +// - **没变就不写**（§4.3：账号状态不是逐帧写库的高频遥测，只有迁移是事实）； +// - **缺平台是调用方的编程错误**（ErrNoPlatform），而**缺通道码只是归一化** +// （不报错、不丢类）。 + +import ( + "context" + "errors" + "sync" + "testing" + "time" +) + +// accountStatusStoreStub 是只记「落了什么」的测试替身：它让「判定」与「落库」两面 +// 可以分别断言（真库的端到端断言在下面用 newTestStore）。 +type accountStatusStoreStub struct { + mu sync.Mutex + rows []Event + err error + calls int +} + +func (s *accountStatusStoreStub) Record(_ context.Context, ev Event) error { + s.mu.Lock() + defer s.mu.Unlock() + s.calls++ + if s.err != nil { + return s.err + } + s.rows = append(s.rows, ev) + return nil +} + +func (s *accountStatusStoreStub) Count(context.Context, Filter) (int64, error) { return 0, nil } +func (s *accountStatusStoreStub) List(context.Context, Filter) ([]Event, error) { + return nil, nil +} +func (s *accountStatusStoreStub) Counters(context.Context, string) ([]Counter, error) { + return nil, nil +} +func (s *accountStatusStoreStub) Flush(context.Context) error { return nil } +func (s *accountStatusStoreStub) Sweep(context.Context) (SweepStats, error) { + return SweepStats{}, nil +} + +// recorded 返回已记录的行的副本（避免调用方在锁外读被并发改写）。 +func (s *accountStatusStoreStub) recorded() []Event { + s.mu.Lock() + defer s.mu.Unlock() + return append([]Event(nil), s.rows...) +} + +// sample 构造一条观测样本（平台固定 qq-obv11，账号 id 由用例决定）。 +func sample(account string, online, known bool) StatusSample { + return StatusSample{ + Platform: PlatformQQ, + AccountID: account, + Online: online, + Known: known, + Source: StatusSourceProbe, + } +} + +// TestStatusTracker_九种迁移逐个落行：把 (前态 → 后态) 的**全部**组合跑一遍。 +// +// 为什么逐个列而不是抽样：§2.6 的价值全在「她自己的账号状态**变了**」这一件事上， +// 而实现里最容易漏的恰恰是「未知 ↔ 确知」那四条边 —— 一个 `if !known { continue }` +// 就会把它们整类丢掉（这正是 gateway 的 liveAccountOnline 必须缺席、而留存面 +// 必须留痕的那条分界线）。 +func TestStatusTracker_九种迁移逐个落行(t *testing.T) { + t.Parallel() + const acct = "bot-1" + type step struct { + online bool + known bool + } + for _, tc := range []struct { + name string + seq []step + prev AccountState + next AccountState + summary string + }{ + {"首次→在线", []step{{true, true}}, "", AccountStateOnline, "账号在线（首次确知）"}, + {"首次→离线", []step{{false, true}}, "", AccountStateOffline, "账号离线（首次确知）"}, + {"首次→未知", []step{{false, false}}, "", AccountStateUnknown, "账号状态未知（尚未探到）"}, + {"在线→离线", []step{{true, true}, {false, true}}, AccountStateOnline, AccountStateOffline, "账号离线（在线→离线）"}, + {"离线→在线", []step{{false, true}, {true, true}}, AccountStateOffline, AccountStateOnline, "账号上线（离线→在线）"}, + {"未知→在线", []step{{false, false}, {true, true}}, AccountStateUnknown, AccountStateOnline, "账号上线（未知→在线）"}, + {"未知→离线", []step{{false, false}, {false, true}}, AccountStateUnknown, AccountStateOffline, "账号离线（未知→离线）"}, + {"在线→未知", []step{{true, true}, {false, false}}, AccountStateOnline, AccountStateUnknown, "账号状态未知（在线→未知）"}, + {"离线→未知", []step{{false, true}, {false, false}}, AccountStateOffline, AccountStateUnknown, "账号状态未知（离线→未知）"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + tr := NewStatusTracker(store, testNow) + ctx := context.Background() + + // 前几步只负责把状态推到 tc.prev（它们自己也是迁移，照常落行）。 + for i := 0; i < len(tc.seq)-1; i++ { + if _, err := tr.Observe(ctx, sample(acct, tc.seq[i].online, tc.seq[i].known)); err != nil { + t.Fatalf("铺垫第 %d 步: %v", i, err) + } + } + last := tc.seq[len(tc.seq)-1] + res, err := tr.Observe(ctx, sample(acct, last.online, last.known)) + if err != nil { + t.Fatalf("Observe: %v", err) + } + if !res.Changed || res.State != tc.next || res.Previous != tc.prev { + t.Fatalf("res = {changed:%v state:%s prev:%s}, want {true %s %s}", + res.Changed, res.State, res.Previous, tc.next, tc.prev) + } + if !res.Recorded { + t.Fatalf("最后一步未落库（Recorded=false）") + } + if res.Event.Summary != tc.summary { + t.Errorf("summary = %q, want %q", res.Event.Summary, tc.summary) + } + // 落行进库的字段逐列核对：T2（§2.6 高价值）+ lifecycle（§3.1 的派生关系） + // + 探活通道码 + 合成 id system 的角色（无当事人的事件）。 + ev := res.Event + if ev.Tier != TierT2 || ev.Category != CategoryLifecycle { + t.Errorf("tier/category = %s/%s, want t2/lifecycle", ev.Tier, ev.Category) + } + if ev.PlatformType != StatusSourceProbe { + t.Errorf("platform_type = %q, want %q", ev.PlatformType, StatusSourceProbe) + } + if !ev.SelfRelated { + t.Error("self_related = false, want true（这是她自己的账号）") + } + if ev.ActorID != "" || ev.ActorRole != RoleSystem { + t.Errorf("actor = %q/%q, want \"\"/system（状态由平台侧上报，没有人的当事人）", + ev.ActorID, ev.ActorRole) + } + if ev.ChannelID != "" || ev.MessageID != "" { + t.Errorf("channel/message = %q/%q, want 空（账号状态不属任何会话或消息）", + ev.ChannelID, ev.MessageID) + } + + // 读回库：最后一步的那一行必须在，且 summary 逐字一致。 + rows, err := store.List(ctx, Filter{Platform: PlatformQQ}) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(rows) != len(tc.seq) { + t.Fatalf("库里有 %d 行, want %d（每一步迁移一行）", len(rows), len(tc.seq)) + } + if rows[0].Summary != tc.summary { + t.Errorf("库里最新一行 summary = %q, want %q", rows[0].Summary, tc.summary) + } + if rows[0].Tier != TierT2 || !rows[0].SelfRelated { + t.Errorf("读回的行 = %+v, want t2 + self_related", rows[0]) + } + }) + } +} + +// TestStatusTracker_未变化不落行：同一状态连采多次只留**首行**。 +// 这是 §4.3 与 §2.6 的分界线：逐帧写库是纯浪费，而迁移是事实。 +func TestStatusTracker_未变化不落行(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + tr := NewStatusTracker(store, testNow) + ctx := context.Background() + for i := 0; i < 5; i++ { + res, err := tr.Observe(ctx, sample("bot-1", true, true)) + if err != nil { + t.Fatalf("第 %d 次 Observe: %v", i, err) + } + if want := i == 0; res.Changed != want { + t.Fatalf("第 %d 次 Changed = %v, want %v", i, res.Changed, want) + } + } + rows, err := store.List(ctx, Filter{}) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(rows) != 1 { + t.Fatalf("库里有 %d 行, want 1（只有首次观测是迁移）", len(rows)) + } +} + +// TestStatusTracker_缺平台是编程错误：样本没有平台 ⇒ 包装的 ErrNoPlatform， +// 且**什么都不写**（既落不进行，也不在迁移表里留一个空键的假状态）。 +func TestStatusTracker_缺平台是编程错误(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + tr := NewStatusTracker(store, testNow) + res, err := tr.Observe(context.Background(), StatusSample{Online: true, Known: true}) + if !errors.Is(err, ErrNoPlatform) { + t.Fatalf("err = %v, want wrap of ErrNoPlatform", err) + } + if res.Changed || res.Recorded { + t.Errorf("res = %+v, want 未判定也未落库", res) + } + if rows, _ := store.List(context.Background(), Filter{}); len(rows) != 0 { + t.Fatalf("库里 %d 行, want 0", len(rows)) + } + if states := tr.States(); len(states) != 0 { + t.Errorf("States() = %v, want 空（坏样本不该污染迁移表）", states) + } +} + +// TestStatusTracker_通道码缺失只是归一化：空通道码不报错、不丢类， +// 归一成固定兜底码写进 platform_type（§4.1：排障要能分辨「这条是谁记的」）。 +func TestStatusTracker_通道码缺失只是归一化(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + tr := NewStatusTracker(store, testNow) + res, err := tr.Observe(context.Background(), StatusSample{ + Platform: PlatformQQ, AccountID: "bot-1", Online: true, Known: true, + }) + if err != nil { + t.Fatalf("Observe: %v", err) + } + if !res.Recorded { + t.Fatal("Recorded = false, want true（缺通道码不该阻止落行）") + } + if res.Event.PlatformType != statusSourceFallback { + t.Errorf("platform_type = %q, want %q", res.Event.PlatformType, statusSourceFallback) + } +} + +// TestStatusTracker_未接线只判定不落库：store 为 nil 时照常判迁移（调用方仍能据此 +// 知道「状态变了」），但 Recorded=false —— 绝不假装写过（§4.3 的 fail-visible）。 +func TestStatusTracker_未接线只判定不落库(t *testing.T) { + t.Parallel() + tr := NewStatusTracker(nil, testNow) + res, err := tr.Observe(context.Background(), sample("bot-1", false, true)) + if err != nil { + t.Fatalf("Observe: %v", err) + } + if !res.Changed || res.State != AccountStateOffline { + t.Errorf("res = %+v, want changed + offline", res) + } + if res.Recorded { + t.Error("Recorded = true, want false（未接线就没有可查的行）") + } +} + +// TestStatusTracker_落库失败_错误上抛且不声称：§4.3 要求逐条留存的失败**必须**让 +// 调用方看见（E3 据此记 WARN）。此时 Recorded 必须为 false —— 否则面板会收到一条 +// 查询面永远查不到的幽灵帧。 +func TestStatusTracker_落库失败_错误上抛且不声称(t *testing.T) { + t.Parallel() + boom := errors.New("db 挂了") + stub := &accountStatusStoreStub{err: boom} + tr := NewStatusTracker(stub, testNow) + res, err := tr.Observe(context.Background(), sample("bot-1", true, true)) + if err == nil || !errors.Is(err, boom) { + t.Fatalf("err = %v, want wrap of %v", err, boom) + } + if res.Recorded { + t.Error("Recorded = true, want false（落库失败了）") + } + if !res.Changed { + t.Error("Changed = false, want true（判定与落库是两件事）") + } + if n := len(stub.recorded()); n != 0 { + t.Errorf("替身收到 %d 行, want 0", n) + } +} + +// TestStatusTracker_并发观测同一迁移只落一行：8 路并发观测同一个账号的同一次迁移。 +// 判定表在临界区内读写，因此只能有一个赢家 —— 多落几行就等于「一次掉线写了 N 行」。 +// （刻意只起 8 个 goroutine：AGENTS §10 的压测纪律要求有界，不堆进程。） +func TestStatusTracker_并发观测同一迁移只落一行(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + tr := NewStatusTracker(store, testNow) + const n = 8 + var wg sync.WaitGroup + changed := make([]bool, n) + wg.Add(n) + for i := 0; i < n; i++ { + go func(i int) { + defer wg.Done() + res, err := tr.Observe(context.Background(), sample("bot-1", true, true)) + if err != nil { + t.Errorf("Observe(%d): %v", i, err) + return + } + changed[i] = res.Changed + }(i) + } + wg.Wait() + winners := 0 + for _, c := range changed { + if c { + winners++ + } + } + if winners != 1 { + t.Errorf("Changed 为真的路数 = %d, want 1", winners) + } + rows, err := store.List(context.Background(), Filter{}) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(rows) != 1 { + t.Fatalf("库里有 %d 行, want 1（同一迁移只该落一行）", len(rows)) + } +} + +// TestStatusTracker_多账号互不影响：两个账号各自迁移，行数与账号对应， +// 绝不因为「同平台」把它们糊成一行。 +func TestStatusTracker_多账号互不影响(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + tr := NewStatusTracker(store, testNow) + ctx := context.Background() + mustObserve(t, tr, sample("bot-1", true, true)) + mustObserve(t, tr, sample("bot-2", false, true)) + // 第二轮的三个观测都不改变状态 ⇒ 不落行。 + mustObserve(t, tr, sample("bot-1", true, true)) + mustObserve(t, tr, sample("bot-2", false, true)) + rows, err := store.List(ctx, Filter{}) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(rows) != 2 { + t.Fatalf("库里有 %d 行, want 2", len(rows)) + } + ids := map[string]bool{rows[0].AccountID: true, rows[1].AccountID: true} + if !ids["bot-1"] || !ids["bot-2"] { + t.Errorf("account_id = %v, want bot-1 与 bot-2 各一行", ids) + } + if states := tr.States(); states[PlatformQQ+"/bot-1"] != AccountStateOnline || + states[PlatformQQ+"/bot-2"] != AccountStateOffline { + t.Errorf("States() = %v, want bot-1 在线 / bot-2 离线", states) + } +} + +// TestStatusSummaries_九种组合逐条有文案：文案表必须覆盖全部 (前态, 后态) 组合且每一条 +// 非空 —— 「漏了一种」在这里变红，而不是悄悄落一行空 summary 进库（空摘要等于没留痕）。 +func TestStatusSummaries_九种组合逐条有文案(t *testing.T) { + t.Parallel() + table := statusSummaries() + if n := len(table); n != 9 { + t.Errorf("文案表 %d 条, want 9（3 种首次观测 + 3×2 种迁移）", n) + } + prevStates := []AccountState{"", AccountStateOnline, AccountStateOffline, AccountStateUnknown} + nextStates := []AccountState{AccountStateOnline, AccountStateOffline, AccountStateUnknown} + for _, prev := range prevStates { + for _, next := range nextStates { + if prev == next { + continue // 同态不构成迁移 + } + if s := table[statusSummaryKey(prev, next)]; s == "" { + t.Errorf("(%q → %q) 没有固定文案，want 一句话常量", prev, next) + } + } + } +} + +// mustObserve 是「这一步必须成功」的简写（失败直接结束用例）。 +func mustObserve(t *testing.T, tr *StatusTracker, s StatusSample) StatusResult { + t.Helper() + res, err := tr.Observe(context.Background(), s) + if err != nil { + t.Fatalf("Observe(%s/%s): %v", s.Platform, s.AccountID, err) + } + return res +} + +// TestStatusTracker_时钟零值时由存储层补当前时刻：CreatedAt 留零值（任何真实写入方 +// 都不该自己造时刻），落库后读回来必须是一个**非零**时刻，而不是 1970（1970 会被 +// 保留期立刻扫掉 —— 那正是「刚发生的证据反被丢掉」）。 +func TestStatusTracker_时钟零值时由存储层补当前时刻(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + tr := NewStatusTracker(store, func() time.Time { return time.Time{} }) + res := mustObserve(t, tr, sample("bot-1", true, true)) + if !res.Event.CreatedAt.IsZero() { + t.Fatalf("CreatedAt = %v, want 零值交给存储层填", res.Event.CreatedAt) + } + rows, err := store.List(context.Background(), Filter{}) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(rows) != 1 || rows[0].CreatedAt.IsZero() { + t.Fatalf("读回的行 = %+v, want 一个非零 created_at", rows) + } +} diff --git a/internal/runtime/events/publish_store.go b/internal/runtime/events/publish_store.go new file mode 100644 index 0000000..7e43ff7 --- /dev/null +++ b/internal/runtime/events/publish_store.go @@ -0,0 +1,113 @@ +package events + +// 本文件是渠道事件**实时面的通用出口**（design/26 §4.5；切片 E7）：一个 Store 装饰器， +// 把每一次**落库成功**的事件顺带推上 bus 主题 `SubjectChannelEvent`。 +// +// 为什么需要它：非消息事件的写入方是桥接层（E3，`channelbridge.Options.EventStore`）， +// 迁移事件（本片的 `StatusTracker`）是另一个写入方。面板要实时，就必须让**所有**新行 +// 都经过同一个出口 —— 而 `Store.Record` 正是那个唯一入口（§4.3）。在这里装饰，等于 +// 「谁写的都在内」，不必让两个写入方各推一遍（那样迟早只推一半，且两处口径会漂移）。 +// +// 三条取舍写在这里： +// +// 1. **只在落库成功之后推**：帧是给面板看的「又发生了一条」，而权威列表是 REST 轮询。 +// 先推帧、后落库失败 ⇒ 面板出现一条查询面永远查不到的幽灵行，正是「画出来的与 +// 事实无关」（§0.2 要根除的那类错误）。落不下去就不声称。 +// 2. **推不出去绝不影响写入**：`Publish` 的失败只记 WARN 后咽下（bus 的订阅缓冲满时 +// 本来就会自行丢帧 —— 它是 best-effort 通道）。`Record` 的返回值只反映**落库** +// 的成败，与推送无关。 +// 3. **绝不装饰 nil**：`inner` 为 nil 时返回**真正的 nil**。桥接层用 `eventStore == nil` +// 判定「未接线 ⇒ 降级为内存计数」（§4.3），把一个 nil 包成非 nil 接口会让那条 +// 降级路径失效，进而变成对 nil 的解引用 —— 这正是本构造函数宁可返回 nil 的理由。 + +import ( + "context" + "log/slog" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +// PublishingStore 是包装一个 Store 并顺带发布实时帧的装饰器。它实现与内层**完全 +// 相同**的语义（含各方法的时间参数、错误包装），差别只有「Record 成功后多发一条 +// bus 事件」。 +// +// ⚠️ **只装饰一处分发（接线须知）**：本片有两条推帧路径 —— 本装饰器（覆盖桥接层写入的 +// 全部渠道事件）与 `Watcher`（账号状态迁移，它自己推）。两者都发在 `SubjectChannelEvent` +// 上。因此**只应装饰桥接层用的那个句柄**（组合根的 `wire_channels.go` → `Options.EventStore`）， +// 不要同时把网关侧（`wire_frontend.go` → `Deps.EventStore`）也换成装饰器：那会让采样侧 +// 写的每一行被推两次（重复帧不会丢数据，但会让面板收到重复的增量）。 +type PublishingStore struct { + inner Store + pub Publisher + log *slog.Logger +} + +// NewPublishingStore 构造装饰器。 +// +// - `inner == nil` ⇒ 返回 nil（见文件头第 3 条；调用方据此保持既有的未接线降级）。 +// - `pub == nil` ⇒ 仍然返回装饰器：此时它退化成**透明转发**（与直接使用 inner +// 逐字等价），这样调用方不必分两种写法，也不会因为「没接 bus」而丢掉 inner。 +// - `log` 为 nil ⇒ 丢弃型 logger（`sourcedLog`），调用方永远可以安全记日志。 +func NewPublishingStore(inner Store, pub Publisher, log *slog.Logger) *PublishingStore { + if inner == nil { + return nil + } + return &PublishingStore{ + inner: inner, + pub: pub, + log: logger.Scoped(sourcedLog(log), eventsLogSource, "publish_store"), + } +} + +// Record 先走内层落库，**成功之后**才推一帧。失败原样上抛（§4.3：逐条留存的失败 +// 必须让调用方看见 —— E3 会据此记 WARN 并把这一帧计入内存降级计数），且**不推帧**。 +func (p *PublishingStore) Record(ctx context.Context, ev Event) error { + if err := p.inner.Record(ctx, ev); err != nil { + return err + } + p.publish(ctx, ev) + return nil +} + +// publish 推一帧实时事件。`pub` 缺席或推送失败都只记 WARN：事件已经落库， +// 面板还有轮询兜底，这里没有任何理由让调用方看见失败。 +func (p *PublishingStore) publish(ctx context.Context, ev Event) { + if p.pub == nil { + return + } + payload := ChannelEventFrame{Event: ev.JSON()} + if err := p.pub.Publish(ctx, SubjectChannelEvent, payload); err != nil { + p.log.Warn("publish channel event frame", + "subject", SubjectChannelEvent, "platform", ev.Platform, + "platform_type", ev.PlatformType, "err", err) + } +} + +// Count 转发到内层（只读查询，与实时面无关）。 +func (p *PublishingStore) Count(ctx context.Context, f Filter) (int64, error) { + return p.inner.Count(ctx, f) +} + +// List 转发到内层。 +func (p *PublishingStore) List(ctx context.Context, f Filter) ([]Event, error) { + return p.inner.List(ctx, f) +} + +// Counters 转发到内层。 +func (p *PublishingStore) Counters(ctx context.Context, platform string) ([]Counter, error) { + return p.inner.Counters(ctx, platform) +} + +// Flush 转发到内层（组合根的常驻 ticker 与退出前的一次调用）。 +func (p *PublishingStore) Flush(ctx context.Context) error { + return p.inner.Flush(ctx) +} + +// Sweep 转发到内层（保留期清扫）。 +func (p *PublishingStore) Sweep(ctx context.Context) (SweepStats, error) { + return p.inner.Sweep(ctx) +} + +// 装饰器必须仍是**一个 Store**：否则组合根把它交给桥接层时会编译不过 —— +// 这条断言比注释更能防住「转发少了一个方法」。 +var _ Store = (*PublishingStore)(nil) diff --git a/internal/runtime/events/publish_test.go b/internal/runtime/events/publish_test.go new file mode 100644 index 0000000..ff98964 --- /dev/null +++ b/internal/runtime/events/publish_test.go @@ -0,0 +1,403 @@ +package events + +// 本文件钉住渠道事件**实时面**的两条出口（design/26 §4.5；切片 E7）： +// 采样循环（`Watcher`）与 Store 装饰器（`PublishingStore`）。 +// +// 两条口径是这里反复断言的： +// - **只在落库成功之后推帧**：落库失败 ⇒ 一帧都不发（面板不会出现查询面查不到的 +// 幽灵行）； +// - **推不出去绝不影响写入**：bus 缺席、Publish 报错都只记 WARN，事件本身照常落库。 + +import ( + "context" + "encoding/json" + "errors" + "sync" + "sync/atomic" + "testing" + "time" +) + +// fakePublisher 记录被推上 bus 的主题与载荷。 +type fakePublisher struct { + mu sync.Mutex + subjects []string + payloads []any + err error +} + +func (p *fakePublisher) Publish(_ context.Context, subject string, payload any) error { + p.mu.Lock() + defer p.mu.Unlock() + p.subjects = append(p.subjects, subject) + p.payloads = append(p.payloads, payload) + return p.err +} + +func (p *fakePublisher) snapshot() ([]string, []any) { + p.mu.Lock() + defer p.mu.Unlock() + return append([]string(nil), p.subjects...), append([]any(nil), p.payloads...) +} + +// staticSource 是固定的状态快照来源（每次调用返回同一份）。 +type staticSource struct { + samples []StatusSample + calls atomic.Int64 +} + +func (s *staticSource) AccountStatusSamples() []StatusSample { + s.calls.Add(1) + return s.samples +} + +// forwardingStoreStub 记录每个方法被调用了几次，并回一个哨兵值 —— 用来断言装饰器 +// 把它们**逐个转发**到内层（少转发一个方法就编译不过，但转发到错的地方仍要测）。 +type forwardingStoreStub struct { + mu sync.Mutex + recordCalls int + countCalls int + listCalls int + counterCalls int + flushCalls int + sweepCalls int + recordErr error +} + +func (s *forwardingStoreStub) Record(context.Context, Event) error { + s.mu.Lock() + defer s.mu.Unlock() + s.recordCalls++ + return s.recordErr +} + +func (s *forwardingStoreStub) Count(context.Context, Filter) (int64, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.countCalls++ + return 42, nil +} + +func (s *forwardingStoreStub) List(context.Context, Filter) ([]Event, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.listCalls++ + return []Event{{ID: 7}}, nil +} + +func (s *forwardingStoreStub) Counters(context.Context, string) ([]Counter, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.counterCalls++ + return []Counter{{Platform: PlatformQQ}}, nil +} + +func (s *forwardingStoreStub) Flush(context.Context) error { + s.mu.Lock() + defer s.mu.Unlock() + s.flushCalls++ + return nil +} + +func (s *forwardingStoreStub) Sweep(context.Context) (SweepStats, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.sweepCalls++ + return SweepStats{Deleted: 3}, nil +} + +// TestNewPublishingStore_未接线返回真nil：桥接层用 `eventStore == nil` 判定「未接线 ⇒ +// 降级为内存计数」（§4.3）。把一个 nil 包成非 nil 接口会让那条路径失效、直接解引用崩掉。 +func TestNewPublishingStore_未接线返回真nil(t *testing.T) { + t.Parallel() + if got := NewPublishingStore(nil, &fakePublisher{}, nil); got != nil { + t.Fatalf("NewPublishingStore(nil, …) = %#v, want 真 nil", got) + } +} + +// TestPublishingStore_落库成功才推帧：成功推一帧、失败一帧不推（且错误原样上抛）。 +func TestPublishingStore_落库成功才推帧(t *testing.T) { + t.Parallel() + pub := &fakePublisher{} + inner := &forwardingStoreStub{} + st := NewPublishingStore(inner, pub, nil) + ev := sampleT2() + if err := st.Record(context.Background(), ev); err != nil { + t.Fatalf("Record: %v", err) + } + subjects, payloads := pub.snapshot() + if len(subjects) != 1 || subjects[0] != SubjectChannelEvent { + t.Fatalf("subjects = %v, want [%s]", subjects, SubjectChannelEvent) + } + frame, ok := payloads[0].(ChannelEventFrame) + if !ok { + t.Fatalf("payload 类型 = %T, want ChannelEventFrame", payloads[0]) + } + if frame.Event.Platform != ev.Platform || frame.Event.Category != string(ev.Category) { + t.Errorf("frame.Event = %+v, want 与事件行一致", frame.Event) + } + // 载荷必须是可 JSON 化的线上形状（面板按 `data.event` 解）。 + raw, err := json.Marshal(frame) + if err != nil { + t.Fatalf("marshal frame: %v", err) + } + var decoded struct { + Event ChannelEventJSON `json:"event"` + } + if err := json.Unmarshal(raw, &decoded); err != nil { + t.Fatalf("decode frame: %v (raw=%s)", err, raw) + } + if decoded.Event.Summary != ev.Summary || decoded.Event.PlatformType != ev.PlatformType { + t.Errorf("解出的 event = %+v, want 与事件行一致", decoded.Event) + } + + // 失败：错误上抛、**一帧都不推**。 + boom := errors.New("落库失败") + inner.recordErr = boom + if err := st.Record(context.Background(), ev); !errors.Is(err, boom) { + t.Fatalf("err = %v, want %v", err, boom) + } + if subjects, _ := pub.snapshot(); len(subjects) != 1 { + t.Errorf("推送次数 = %d, want 1（失败的那次不许推）", len(subjects)) + } +} + +// TestPublishingStore_无Publisher时透明转发：bus 缺席（测试构造 / 未接 bus）不是错误， +// 事件照常落库，且**不 panic**。 +func TestPublishingStore_无Publisher时透明转发(t *testing.T) { + t.Parallel() + inner := &forwardingStoreStub{} + st := NewPublishingStore(inner, nil, nil) + if st == nil { + t.Fatal("NewPublishingStore(inner, nil, nil) = nil, want 装饰器（内层非 nil）") + } + if err := st.Record(context.Background(), sampleT2()); err != nil { + t.Fatalf("Record: %v", err) + } + if inner.recordCalls != 1 { + t.Errorf("内层 Record 调用 = %d, want 1", inner.recordCalls) + } +} + +// TestPublishingStore_推送失败不影响写入：bus 报错只记 WARN，Record 照常回 nil —— +// 帧是 best-effort，落库那一行才是权威（§4.5 的轮询兜底正是为此存在）。 +func TestPublishingStore_推送失败不影响写入(t *testing.T) { + t.Parallel() + pub := &fakePublisher{err: errors.New("bus 满了")} + st := NewPublishingStore(&forwardingStoreStub{}, pub, nil) + if err := st.Record(context.Background(), sampleT2()); err != nil { + t.Fatalf("Record = %v, want nil（推送失败不该冒泡）", err) + } + if subjects, _ := pub.snapshot(); len(subjects) != 1 { + t.Errorf("推送尝试 = %d, want 1（尝试过即记 WARN）", len(subjects)) + } +} + +// TestPublishingStore_逐个转发其余方法：装饰器必须仍是**同一个 Store** —— +// 查询/计数/flush/清扫一律原样转发（哨兵值原样回来即证明到底了）。 +func TestPublishingStore_逐个转发其余方法(t *testing.T) { + t.Parallel() + inner := &forwardingStoreStub{} + st := NewPublishingStore(inner, nil, nil) + ctx := context.Background() + if n, err := st.Count(ctx, Filter{}); err != nil || n != 42 { + t.Errorf("Count = %d/%v, want 42/nil", n, err) + } + if rows, err := st.List(ctx, Filter{}); err != nil || len(rows) != 1 || rows[0].ID != 7 { + t.Errorf("List = %+v/%v, want [{ID:7}]/nil", rows, err) + } + if cs, err := st.Counters(ctx, PlatformQQ); err != nil || len(cs) != 1 || cs[0].Platform != PlatformQQ { + t.Errorf("Counters = %+v/%v, want 一行 qq-obv11", cs, err) + } + if err := st.Flush(ctx); err != nil { + t.Errorf("Flush = %v, want nil", err) + } + if stats, err := st.Sweep(ctx); err != nil || stats.Deleted != 3 { + t.Errorf("Sweep = %+v/%v, want {Deleted:3}", stats, err) + } + if inner.countCalls != 1 || inner.listCalls != 1 || inner.counterCalls != 1 || + inner.flushCalls != 1 || inner.sweepCalls != 1 { + t.Errorf("内层调用次数 = %+v, want 每个方法各 1 次", inner) + } +} + +// TestNewWatcher_未接线明确失败：没有留存面就没有「可查的历史」这件事可做， +// 宁可明确失败，也不给一个只推帧不落库的替身（与 NewStore 的 ErrNilDB 同一口径）。 +func TestNewWatcher_未接线明确失败(t *testing.T) { + t.Parallel() + w, err := NewWatcher(WatcherConfig{Source: &staticSource{}}) + if !errors.Is(err, ErrNilDB) { + t.Fatalf("err = %v, want wrap of ErrNilDB", err) + } + if w != nil { + t.Fatalf("NewWatcher = %#v, want nil", w) + } +} + +// TestWatcher_推帧只推已落库的_载荷形状：两个账号各一次迁移 ⇒ 两帧， +// 主题与载荷形状逐项核对；随后再采一轮（状态没变）⇒ 一帧都不多。 +func TestWatcher_推帧只推已落库的_载荷形状(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + pub := &fakePublisher{} + src := &staticSource{samples: []StatusSample{ + sample("bot-1", true, true), + sample("bot-2", false, true), + {Platform: PlatformQQ, AccountID: "bot-3", Known: false, Source: StatusSourceProbe}, + }} + w, err := NewWatcher(WatcherConfig{ + Store: store, Source: src, Publisher: pub, Interval: time.Hour, Now: testNow, + }) + if err != nil { + t.Fatalf("NewWatcher: %v", err) + } + ctx := context.Background() + w.ObserveOnce(ctx) + subjects, payloads := pub.snapshot() + if len(subjects) != 3 { + t.Fatalf("帧数 = %d, want 3（三个账号各一次首次观测）", len(subjects)) + } + for i, s := range subjects { + if s != SubjectChannelEvent { + t.Errorf("第 %d 帧主题 = %q, want %q", i, s, SubjectChannelEvent) + } + if _, ok := payloads[i].(ChannelEventFrame); !ok { + t.Errorf("第 %d 帧载荷类型 = %T, want ChannelEventFrame", i, payloads[i]) + } + } + // 「未知」也是一个迁移、也要落行、也要推帧（§1.1 的绝不静默）。 + frame, _ := payloads[2].(ChannelEventFrame) + if frame.Event.Platform != PlatformQQ || frame.Event.AccountID != "bot-3" { + t.Errorf("第三帧 = %+v, want bot-3", frame.Event) + } + if frame.Event.Summary == "" { + t.Error("未知态的 summary 为空，want 一句话常量") + } + // 「未知」的 ID 尚未回读 ⇒ id 键整个缺席，绝不写 0（0 会让所有实时帧看起来是同一行）。 + raw, err := json.Marshal(frame) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var keys map[string]any + if err := json.Unmarshal(raw, &keys); err != nil { + t.Fatalf("unmarshal: %v", err) + } + evKeys, ok := keys["event"].(map[string]any) + if !ok { + t.Fatalf("event 载荷 = %v, want 一个对象", keys["event"]) + } + if _, present := evKeys["id"]; present { + t.Errorf("实时帧里出现了 id 键: %s（写侧不知道刚插入的行的 id，如实缺席）", raw) + } + + // 第二轮：状态未变 ⇒ 不落行、不推帧。 + w.ObserveOnce(ctx) + if subjects, _ := pub.snapshot(); len(subjects) != 3 { + t.Errorf("第二轮后又推了帧（共 %d），want 仍是 3", len(subjects)) + } + rows, err := store.List(ctx, Filter{}) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(rows) != 3 { + t.Errorf("库里有 %d 行, want 3", len(rows)) + } +} + +// TestWatcher_落库失败既不推帧也不崩：逐条留存的失败只影响它自己那个账号， +// 其余账号照常走完这一轮（绝不整轮丢弃）。 +func TestWatcher_落库失败既不推帧也不崩(t *testing.T) { + t.Parallel() + pub := &fakePublisher{} + stub := &accountStatusStoreStub{err: errors.New("写不进去")} + src := &staticSource{samples: []StatusSample{sample("bot-1", true, true)}} + w, err := NewWatcher(WatcherConfig{Store: stub, Source: src, Publisher: pub, Interval: time.Hour}) + if err != nil { + t.Fatalf("NewWatcher: %v", err) + } + w.ObserveOnce(context.Background()) + if subjects, _ := pub.snapshot(); len(subjects) != 0 { + t.Errorf("推送次数 = %d, want 0（落不下去就不声称）", len(subjects)) + } + if stub.calls != 1 { + t.Errorf("落库调用 = %d, want 1（判定照常发生，只是没成功）", stub.calls) + } +} + +// TestWatcher_未接Publisher只落库：未接 bus（真机未接 / 单元测试构造）时 +// 事件照常留痕，只是没有实时帧。 +func TestWatcher_未接Publisher只落库(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + src := &staticSource{samples: []StatusSample{sample("bot-1", true, true)}} + w, err := NewWatcher(WatcherConfig{Store: store, Source: src, Interval: time.Hour}) + if err != nil { + t.Fatalf("NewWatcher: %v", err) + } + w.ObserveOnce(context.Background()) + rows, err := store.List(context.Background(), Filter{}) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(rows) != 1 { + t.Fatalf("库里有 %d 行, want 1（只落库不推帧）", len(rows)) + } +} + +// TestWatcher_源缺席不崩：Source 为 nil 时每拍都是空快照（等于不观测）， +// 绝不因为未接线而 panic 或空转刷日志。 +func TestWatcher_源缺席不崩(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + w, err := NewWatcher(WatcherConfig{Store: store, Interval: time.Hour}) + if err != nil { + t.Fatalf("NewWatcher: %v", err) + } + w.ObserveOnce(context.Background()) + if rows, _ := store.List(context.Background(), Filter{}); len(rows) != 0 { + t.Fatalf("库里有 %d 行, want 0", len(rows)) + } +} + +// TestWatcher_立即观测且可干净停止：Start 后**不等一个间隔**就先采一次（进程刚起来 +// 时那次观测不该被推迟），ctx 取消后循环退出、Done 关闭。 +func TestWatcher_立即观测且可干净停止(t *testing.T) { + t.Parallel() + store, _ := newTestStore(t, nil) + src := &staticSource{samples: []StatusSample{sample("bot-1", true, true)}} + // 节拍取得很短：这条用例要的是「循环真的在跑、真的会停」，而不是等 60s。 + w, err := NewWatcher(WatcherConfig{Store: store, Source: src, Interval: 5 * time.Millisecond}) + if err != nil { + t.Fatalf("NewWatcher: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + w.Start(ctx) + // 重复 Start 必须幂等：否则两个循环同时 `close(w.done)` 会 panic。 + w.Start(ctx) + if !waitFor(func() bool { return src.calls.Load() >= 3 }, 2*time.Second) { + t.Fatalf("采样调用 = %d, want ≥3（立即一次 + 至少两次节拍）", src.calls.Load()) + } + cancel() + select { + case <-w.Done(): + case <-time.After(2 * time.Second): + t.Fatal("ctx 取消后 Done 未关闭（采样循环没干净退出）") + } + settled := src.calls.Load() + time.Sleep(30 * time.Millisecond) + if got := src.calls.Load(); got != settled { + t.Errorf("循环退出后仍在采样：%d → %d", settled, got) + } +} + +// waitFor 轮询直到 cond 为真或超时（避免给循环用例引入固定 sleep 的脆性）。 +func waitFor(cond func() bool, timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return true + } + time.Sleep(time.Millisecond) + } + return cond() +} diff --git a/internal/runtime/events/watcher.go b/internal/runtime/events/watcher.go new file mode 100644 index 0000000..1da146b --- /dev/null +++ b/internal/runtime/events/watcher.go @@ -0,0 +1,205 @@ +package events + +// 本文件把「账号状态迁移」接到**实时面**（design/26 §2.6 / §4.5；切片 E7）。 +// +// 三段分工，谁都不越界： +// +// - **判定与落盘**在 `StatusTracker`（account_status.go）——它不看时钟、不推任何东西； +// - **采样节拍**在本文件的 `Watcher`：立即采一次，此后按间隔采（默认 60s，与适配器 +// `get_status` 探活的节奏一致：采得比探活快只会拿到同一个值，纯浪费）； +// - **送达面板**走上行 bus 主题 `channel.event`：网关的 hub 把它规范成 `channel_event` +// WS 帧并过角色门（fail-closed ⇒ 只发 owner + admin）。**不新开监听端口、不做 HTTP +// 互调**（AGENTS §9 红线）：整条路是「进程内 bus → hub relay → 连接」。 +// +// 为什么推送**只在落库成功之后**发生（见 observe）：帧是给面板看的「又发生了一条」， +// 而面板的权威列表是 REST 轮询（帧是 best-effort，缓冲满即丢）。若先推帧、落库却失败， +// 面板会出现一条**查询面永远查不到**的幽灵行 —— 那正是「画出来的与事实无关」。 +// 落不下去就不声称，与 §0.2 要根除的那类错误同向。 + +import ( + "context" + "fmt" + "log/slog" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +const ( + // SubjectChannelEvent 是渠道事件实时推送的 bus 主题。 + // + // 它是**领域主题**（生产者定义、网关做 WS 归一），与 `orchestration.SubagentResultSubject` + // / `observability.log` 同一个套路：生产者不必知道 WS 帧叫什么名字，`ws_bridge.go` + // 的 eventBridge 负责把主题映射成帧类型。 + SubjectChannelEvent = "channel.event" + + // DefaultWatchInterval 是状态采样间隔（§2.6 的探活节奏约 60s）。 + // 它**不是**可配项：采样快于探活拿不到新信息，慢于探活只会推迟发现掉线。 + DefaultWatchInterval = 60 * time.Second +) + +// Publisher 是**进程内 bus** 的最小面（`bus.Bus` 直接满足它）。 +// +// 刻意不在本包 import bus：events 只依赖「能发一个主题」这件事，谁来发（进程内 bus、 +// 将来的 NATS、测试替身）由调用方决定（与 Store 接口同一取舍，§6.3）。 +type Publisher interface { + Publish(ctx context.Context, subject string, payload any) error +} + +// ChannelEventFrame 是实时帧的 data 本体：`{"event": }`。 +// +// 刻意不把事件字段摊平到 data 顶层：将来要给帧加**帧级**元信息（如「这是重放的」） +// 时，不必与事件字段抢名字；面板的增量合并逻辑也只认 `data.event` 一处。 +type ChannelEventFrame struct { + Event ChannelEventJSON `json:"event"` +} + +// StatusSource 报告**当前所有可观测账号**的状态快照。网关实现它（读能力注册表）， +// 本包只消费 —— 于是采样与领域逻辑分开，两边各自可测。 +type StatusSource interface { + AccountStatusSamples() []StatusSample +} + +// WatcherConfig 是实时面的接线参数。 +type WatcherConfig struct { + // Store 是事件留存面。为 nil 时只判定不落库**也不推帧**（见 observe 的取舍）， + // 调用方应据此不启动 Watcher（网关的 channelEventWriter 就返回 nil）。 + Store Store + // Source 是状态快照来源。为 nil 时每拍都拿到空快照（等于不观测），不 panic。 + Source StatusSource + // Publisher 是 bus 出口；nil 时只落库不推帧（例如未接 bus 的测试构造）。 + Publisher Publisher + // Log 是来源已标注的 logger；nil 时用丢弃 logger（与 store.go 的 sourcedLog 同形）。 + Log *slog.Logger + // Interval 是采样间隔；≤0 取 DefaultWatchInterval。 + Interval time.Duration + // Now 是可注入时钟（测试用）；nil 时用 time.Now。 + Now func() time.Time +} + +// Watcher 按节拍采样账号状态、判定迁移、落行并把新行推上 bus。 +// +// 它只被自己的 `Run` goroutine 使用（tracker 自带互斥量），因此没有额外的并发约定； +// `Run` 返回即代表已停止（ctx 取消或 store 未接线）。 +type Watcher struct { + tracker *StatusTracker + source StatusSource + publisher Publisher + log *slog.Logger + interval time.Duration + // mu 只保护 running/done 这对一次性标志，让 Start 幂等（重复 Start 不会起第二个 + // 采样循环 —— 两个循环会把「首次观测」判成两次迁移，落两行）。 + mu sync.Mutex + running bool + done chan struct{} +} + +// NewWatcher 构造 Watcher。唯一的失败是**没有留存面**（Store 为 nil）：那就没有 +// 「可查的历史」这件事可做，宁可明确失败，也不给一个只推帧不落库的替身（与 +// `NewStore` 的 ErrNilDB 同一口径）。Publisher 为 nil 是合法的：只落库不推帧。 +func NewWatcher(cfg WatcherConfig) (*Watcher, error) { + if cfg.Store == nil { + return nil, fmt.Errorf("events: 账号状态实时面: %w", ErrNilDB) + } + interval := cfg.Interval + if interval <= 0 { + interval = DefaultWatchInterval + } + log := logger.Scoped(sourcedLog(cfg.Log), eventsLogSource, "account_status") + return &Watcher{ + tracker: NewStatusTracker(cfg.Store, cfg.Now), + source: cfg.Source, + publisher: cfg.Publisher, + log: log, + interval: interval, + done: make(chan struct{}), + }, nil +} + +// Interval 返回生效的采样间隔。调用方（网关的启动日志）用它，而不是再写一份 +// `DefaultWatchInterval` 字面量 —— 两份字面量迟早会漂移成「日志说的」与「实际跑的」。 +func (w *Watcher) Interval() time.Duration { return w.interval } + +// Start 起一个采样 goroutine 并立即返回。重复调用只生效一次（见 Watcher.mu 的注释）。 +// ctx 取消后循环干净退出（不泄漏 goroutine）。 +func (w *Watcher) Start(ctx context.Context) { + w.mu.Lock() + start := !w.running + if start { + w.running = true + } + w.mu.Unlock() + if !start { + return + } + go w.run(ctx) +} + +// Done 在采样循环退出后关闭。给测试与组合根一个「它真的停了」的判据， +// 而不是靠 sleep 猜。 +func (w *Watcher) Done() <-chan struct{} { return w.done } + +// run 是采样循环：**先立即采一次**（进程刚起来时的那次观测不该等一个间隔）， +// 此后按节拍采，直到 ctx 取消。 +func (w *Watcher) run(ctx context.Context) { + defer close(w.done) + w.ObserveOnce(ctx) + ticker := time.NewTicker(w.interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + w.ObserveOnce(ctx) + } + } +} + +// ObserveOnce 采一轮：遍历来源快照，逐个判定迁移，迁移且落库成功的推一帧。 +// +// 它**不阻塞**在 I/O 上更久于一次落库：源为 nil 或返回空快照时什么都不做。 +// 每一类失败都只影响它自己那一个账号（缺平台的样本记 WARN 后继续）， +// 绝不让一个坏样本拖掉整轮的其余账号 —— 「整类丢弃」是本项目反复点名的病。 +// +// **推帧的门就是那条错误分支**：`StatusTracker.Observe` 只在行确实落进库时才回 +// nil（落库失败、样本缺平台都回错误），因此「推了帧却没落库」在这条路上不可能发生。 +// `StatusResult.Recorded` 因此不在这里再查一遍 —— 两个等价的判据并排写只会让 +// 后来的人以为其中一个是可以删的。 +func (w *Watcher) ObserveOnce(ctx context.Context) { + if w.source == nil { + return + } + for _, sample := range w.source.AccountStatusSamples() { + res, err := w.tracker.Observe(ctx, sample) + if err != nil { + w.log.Warn("channel account status not recorded", + "platform", sample.Platform, "account", sample.AccountID, "err", err) + continue + } + if !res.Changed { + continue + } + w.log.Info("channel account status changed", + "platform", res.Event.Platform, "account", res.Event.AccountID, + "state", string(res.State), "previous", string(res.Previous), + "source", res.Event.PlatformType, "recorded", res.Recorded) + w.publish(ctx, res.Event) + } +} + +// publish 把一条**已落库**的事件推上 bus。失败只记 WARN：帧是 best-effort +// （Bus 的订阅缓冲满时自行丢弃），落库那一行才是权威证据；绝不因为推不出去而 +// 让采样循环失败或回滚任何东西。 +func (w *Watcher) publish(ctx context.Context, ev Event) { + if w.publisher == nil { + return + } + payload := ChannelEventFrame{Event: ev.JSON()} + if err := w.publisher.Publish(ctx, SubjectChannelEvent, payload); err != nil { + w.log.Warn("publish channel event frame", + "subject", SubjectChannelEvent, "platform", ev.Platform, + "platform_type", ev.PlatformType, "err", err) + } +} diff --git a/internal/runtime/events/wire_event.go b/internal/runtime/events/wire_event.go new file mode 100644 index 0000000..9bd22be --- /dev/null +++ b/internal/runtime/events/wire_event.go @@ -0,0 +1,66 @@ +package events + +// 本文件是**一条事件行到线上的投影**（design/26 §4.5；切片 E7）。 +// +// 为什么投影住在 events 包而不是网关：同一条事件有**两个**线上出口 —— +// 管理 REST 的列表端点（`admin_channel_events.go`）与实时 WS 帧（`channel.event` +// 主题，见 watcher.go / publish_store.go）。两处若各写一份结构，迟早会漂移成 +// 「帧里有这个键、列表里没有」这类无从察觉的差异（正是 §9.2 反复强调的那类债）。 +// 因此投影只写一次：**REST 的 DTO 是它的类型别名，实时帧的载荷也是它**。 +// +// 列与 `Event` 一一对应且不多不少：**没有**承载平台原文、媒体 URL、凭据或 OneBot +// `flag` 的字段（§4.2 红线）。`summary` 是适配器或本包生成的固定文案，已净化， +// 查不出原始文本。 + +import "time" + +// ChannelEventJSON 是一条 `channel_events` 行的线上投影（键名 camelCase，与契约 +// v0.67 登记的 `events[]` 逐字一致）。 +type ChannelEventJSON struct { + // ID 是数据库自增主键。**可缺席**：实时帧里的事件行是**刚落库、id 还没回读**的 + // 那一行，此时它被整个省略（而不是写一个 0）—— 0 会让客户端以为「所有实时帧 + // 是同一行」。列表端点读的是已落盘的行，id 恒有值。 + ID int64 `json:"id,omitempty"` + Platform string `json:"platform"` + AccountID string `json:"accountId"` + Tier string `json:"tier"` + Category string `json:"category"` + PlatformType string `json:"platformType"` + ChannelID string `json:"channelId"` + ChannelType string `json:"channelType"` + GroupID string `json:"groupId"` + ActorID string `json:"actorId"` + ActorRole string `json:"actorRole"` + TargetID string `json:"targetId"` + SelfRelated bool `json:"selfRelated"` + MessageID string `json:"messageId"` + Summary string `json:"summary"` + CreatedAt string `json:"createdAt"` + UpdatedAt string `json:"updatedAt"` +} + +// JSON 把一条事件行投影成线上结构。时间一律以 **UTC RFC3339** 回显（与 `llm/usage` +// 同口径）：本地时区偏移会让同一行的两个出口显示出不同的时刻。 +// +// id 为 0 时留空（见 ID 的注释）：写侧不知道刚插入的行的 id，如实缺席比编一个强。 +func (ev Event) JSON() ChannelEventJSON { + return ChannelEventJSON{ + ID: ev.ID, + Platform: ev.Platform, + AccountID: ev.AccountID, + Tier: string(ev.Tier), + Category: string(ev.Category), + PlatformType: ev.PlatformType, + ChannelID: ev.ChannelID, + ChannelType: ev.ChannelType, + GroupID: ev.GroupID, + ActorID: ev.ActorID, + ActorRole: ev.ActorRole, + TargetID: ev.TargetID, + SelfRelated: ev.SelfRelated, + MessageID: ev.MessageID, + Summary: ev.Summary, + CreatedAt: ev.CreatedAt.UTC().Format(time.RFC3339), + UpdatedAt: ev.UpdatedAt.UTC().Format(time.RFC3339), + } +} diff --git a/internal/services/gateway/admin_channel_events.go b/internal/services/gateway/admin_channel_events.go index ff3dc37..f5d0de2 100644 --- a/internal/services/gateway/admin_channel_events.go +++ b/internal/services/gateway/admin_channel_events.go @@ -42,31 +42,17 @@ const channelEventsLogSource = "gateway/admin_channel_events" // 「与 `llm/usage` 同口径」，两处各对着同一条设计口径写死，谁也别想只改一处。 const channelEventsMaxWindow = 90 * 24 * time.Hour -// channelEventDTO 是一条 `channel_events` 行的线上投影（键名与同族的 -// `counters[]` 一致，取 camelCase）。 +// channelEventDTO 是一条 `channel_events` 行的线上投影。它现在是 +// `events.ChannelEventJSON` 的**类型别名**（切片 E7）：同一条事件有两条线上出口 —— +// 这里的 REST 列表与实时 `channel_event` 帧的 `data.event`。两个出口共用一份结构， +// 是「不许两处各写一份形状」这条纪律的落地（否则迟早出现「帧里有这个键、列表里 +// 没有」这类无从察觉的差异）。 // -// 列与 `events.Event` **一一对应且不多不少**：它没有、也不会有承载平台原文、 +// 列与 `events.Event` 一一对应且不多不少：它没有、也不会有承载平台原文、 // 媒体 URL、凭据或 OneBot `flag` 的字段（§4.2 红线）。`summary` 是适配器生成的 -// 固定文案（如「群成员加入」），**已净化**，查不出原始文本。 -type channelEventDTO struct { - ID int64 `json:"id"` - Platform string `json:"platform"` - AccountID string `json:"accountId"` - Tier string `json:"tier"` - Category string `json:"category"` - PlatformType string `json:"platformType"` - ChannelID string `json:"channelId"` - ChannelType string `json:"channelType"` - GroupID string `json:"groupId"` - ActorID string `json:"actorId"` - ActorRole string `json:"actorRole"` - TargetID string `json:"targetId"` - SelfRelated bool `json:"selfRelated"` - MessageID string `json:"messageId"` - Summary string `json:"summary"` - CreatedAt string `json:"createdAt"` - UpdatedAt string `json:"updatedAt"` -} +// 固定文案（如「群成员加入」）或本包的固定迁移文案，**已净化**，查不出原始文本。 +// `id` 可缺席（实时帧里那行的 id 尚未回读），列表路径上它恒有值。 +type channelEventDTO = events.ChannelEventJSON // channelEventListResponse 是列表端点的 data 本体（§4.5：`{events, total}`）。 // `events` 恒为非 nil 切片（无数据时是 `[]` 而非 `null`，客户端不必判空）。 @@ -323,27 +309,9 @@ func channelEventsListResponseOf(rows []events.Event, total int64) channelEventL return out } -// channelEventDTOOf 是单行投影（一处换算，避免 List 与将来可能的详情端点各写一遍）。 +// channelEventDTOOf 是单行投影（一处换算，避免 List 与实时帧各写一遍）。 func channelEventDTOOf(ev events.Event) channelEventDTO { - return channelEventDTO{ - ID: ev.ID, - Platform: ev.Platform, - AccountID: ev.AccountID, - Tier: string(ev.Tier), - Category: string(ev.Category), - PlatformType: ev.PlatformType, - ChannelID: ev.ChannelID, - ChannelType: ev.ChannelType, - GroupID: ev.GroupID, - ActorID: ev.ActorID, - ActorRole: ev.ActorRole, - TargetID: ev.TargetID, - SelfRelated: ev.SelfRelated, - MessageID: ev.MessageID, - Summary: ev.Summary, - CreatedAt: ev.CreatedAt.UTC().Format(time.RFC3339), - UpdatedAt: ev.UpdatedAt.UTC().Format(time.RFC3339), - } + return ev.JSON() } // channelEventCountersResponseOf 把计数表结果翻成线上结构。`approx` 恒为 true： diff --git a/internal/services/gateway/ws_bridge.go b/internal/services/gateway/ws_bridge.go index 12cd82f..e28338e 100644 --- a/internal/services/gateway/ws_bridge.go +++ b/internal/services/gateway/ws_bridge.go @@ -4,6 +4,7 @@ import ( "strings" "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/comm" @@ -51,6 +52,11 @@ var eventBridge = map[string]string{ obsLogSubject: wsObservabilityLog, obsTraceSubject: wsObservabilityTrace, obsMetricsSubject: wsObservabilityMetrics, + // 渠道事件实时帧（design/26 §2.6 / §4.5；切片 E7）：账号状态迁移与（经 + // PublishingStore 装饰的）其余渠道事件都发在同一个领域主题上，在这里统一落成 + // 帧类型。载荷即 `events.ChannelEventFrame`（`{"event":{…}}`），与 REST 列表的 + // `events[]` 同形 —— 面板不需要第二套解码。 + events.SubjectChannelEvent: wsChannelEvent, } // bridgePatterns 是 hub 在 ws.> 之外额外订阅的 bus 模式，用于 @@ -67,6 +73,9 @@ var bridgePatterns = []string{ // 后台子代理批次结果：它既不匹配 ws.>，也不是 cyrene.agent 的 4 段广播形态， // 所以必须在这里登记订阅，否则后台结果永远上不了前台（只能看到进度帧）。 orchestration.SubagentResultSubject, + // 渠道事件实时面（切片 E7）：精确主题，不用 channel.> —— 不替将来可能的 + // 其他 channel.* 生产者顺手开口子（多订阅一个主题就多一次解码尝试）。 + events.SubjectChannelEvent, } // normalizeEvent 把 bus 事件映射为下行 wsEvent。ws.* 主题保持 diff --git a/internal/services/gateway/ws_channel_events.go b/internal/services/gateway/ws_channel_events.go new file mode 100644 index 0000000..2726d25 --- /dev/null +++ b/internal/services/gateway/ws_channel_events.go @@ -0,0 +1,159 @@ +package gateway + +// 本文件是渠道事件的**实时面**（design/26 §2.6 / §4.5；切片 E7）。 +// +// 与同族的 `admin_channel_events.go` 分工明确、互不重叠： +// +// - 那边是**只读查询面**（REST，面板轮询的权威）：`/admin/channels/events` 与 +// `/counters`； +// - 这边是**增量面**：周期采样账号状态 → 判定迁移 → 落一行 T2（`events.StatusTracker`） +// → 推上进程内 bus（`channel.event`）→ hub 的 relay 规范成 `channel_event` 帧 → +// 角色门 → 连接。**不新开监听端口、不做 HTTP 互调**（AGENTS §9 红线）。 +// +// 为什么采样要**常驻**（而不是像可观测推送那样「有订阅才跑」）：§0.2 的事故正是 +// 「没人看面板的那几分钟里，账号掉线没有任何痕迹」。留存面必须**不看面板也在跑**， +// 否则「可查的历史」只在有人盯着时才存在。推帧只是顺带的产物。 +// +// 采样节拍 60s 与 `qq_obv11` 的 `get_status` 探活同拍：采得比探活快只会反复读到 +// 同一个值（纯浪费），慢于探活只会推迟发现掉线。它**不是可配项** —— 调优类旋钮 +// 一律走 `$DATA_DIR/config/**`（AGENTS §10 的 `.env` 冻结红线），而这里没有任何 +// 值得调的东西。 + +import ( + "context" + "sort" + + capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" + "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" +) + +// wsChannelEvent 是渠道事件实时帧的类型名。它**不在** `ws_frame_scope.go` 的 +// `frameReachesAll` 白名单里，因此按 fail-closed 默认档只发 `owner` / `admin` +// 连接；`viewer` 继续走既有的 REST 只读档（design/11 §7.1）。帧壳： +// +// {"type":"channel_event","data":{"event":{…ChannelEventJSON…}}} +// +// 帧里**没有**任何平台原文 / 媒体 URL / 凭据：它承载的就是 `channel_events` 的一行， +// 那行的列在迁移 0021 里就没有那些字段（§4.2）。 +const wsChannelEvent = "channel_event" + +// channelAccountStatusSamples 采集注册表里**每一个可观测账号**的在线状态快照。 +// +// 与 `liveAccountOnline`（管理列表用）的三点差别，每一点都是刻意的： +// +// 1. **未知也收**：那边「只收确知的结果」，known=false 的账号整类不进 map（面板上 +// 表现为字段缺席）；这边 **known=false 正是要留痕的状态之一**（§2.6 的「尚未探到」 +// ≠ 离线），归一化成 `unknown` 照常落行。这正是本片要补的洞：未知不是「不报」， +// 而是「如实报未知」。 +// 2. **带上账号 id**：管理列表按平台聚合，留存行按 `account_id` 落列（§4.1）。 +// 3. **排序确定**：注册表的 List 走 map，顺序随机；这里按（平台, 账号）排序， +// 使日志与测试顺序可复现（同一拍里多个账号的先后不该是随机的）。 +// +// 没有那条状态面的适配器（telegram / webhook / weixin-ilink…）**不进快照**：本能力 +// 观测的是「问得到的账号状态」，凭空为它们造一个 `unknown` 等于把「这个平台根本没有 +// 这条信号」写成「我们探不到」—— 那是两件事（与 §3.4 对微信「不提供非消息事件」的 +// 如实标注同一个方向：没有就说没有，不伪造）。 +func (s *Server) channelAccountStatusSamples() []events.StatusSample { + if s == nil || s.caps == nil { + return nil + } + out := make([]events.StatusSample, 0, 4) + for _, cap := range s.caps.List(capreg.KindChannel) { + ch, ok := cap.Value.(sdk.Channel) + if !ok { + continue + } + as, ok := asAccountStatusAware(cap.Value) + if !ok { + continue + } + online, known := as.AccountStatus() + platform, account := splitAdapterName(ch.Name()) + out = append(out, events.StatusSample{ + Platform: platform, AccountID: account, + Online: online, Known: known, + // 这条状态面当前**唯一**的写入点是 `qq_obv11/probe.go` 的 + // recordAccountStatus（get_status 主动探活）。将来若把 meta_event 心跳 + // 也接进同一状态面，通道码应改由适配器一并报告（那时这个常量就要让位）。 + Source: events.StatusSourceProbe, + }) + } + sort.Slice(out, func(i, j int) bool { + if out[i].Platform != out[j].Platform { + return out[i].Platform < out[j].Platform + } + return out[i].AccountID < out[j].AccountID + }) + return out +} + +// channelEventWriter 返回渠道事件的**写入面**。 +// +// `Deps.EventStore` 声明的类型是只读查询面（E6 的两个端点只需要 List/Count/Counters）， +// 而组合根实际注入的是实现了完整 `events.Store` 的具体存储。这里按**能力断言**取 +// 写入面（与 `asAccountStatusAware` / `asRescanAware` 同一套路）： +// +// - 拿到 ⇒ 实时面可用（落行 + 推帧）； +// - 拿不到（真 nil 接口，或测试里的只读替身）⇒ 返回 nil，实时面**整个不启动**， +// 绝不伪造一个写不进去的替身、也绝不留一条「看着在跑其实什么都没记」的循环。 +// +// ⚠️ 这条断言是**零改动的接线**（不动 deps.go / server.go / contracts.go 三个热点文件）。 +// 若将来把写入面正式纳入 `Deps`（另一个字段），这里的断言应换成那个字段。 +func (s *Server) channelEventWriter() events.Store { + if s == nil || s.channelEvents == nil { + return nil + } + store, ok := s.channelEvents.(events.Store) + if !ok { + return nil + } + return store +} + +// statusSourceFunc 把「返回快照的函数」适配成 events.StatusSource（网关不实现实体类型， +// 只交出这个函数，避免为一个方法再引入一个结构体）。 +type statusSourceFunc func() []events.StatusSample + +// AccountStatusSamples 实现 events.StatusSource；函数为 nil 时返回空快照。 +func (f statusSourceFunc) AccountStatusSamples() []events.StatusSample { + if f == nil { + return nil + } + return f() +} + +// startChannelEventWatch 启动渠道事件的实时面（常驻，直到 ctx 取消）。 +// +// 启停**都留一行日志**（对齐 `qq_obv11/probe.go:logProbeStart` 的同一条纪律）： +// 「实时面根本没跑」与「跑了但一直没迁移」在日志里否则长得一模一样（都是没有）， +// 而排障时要能一眼分开这两件事。落点是**启动**（带生效间隔、有无 bus、有无状态源） +// 与**未启动**（带原因）两种；停止不单独记（进程退出本身有日志，再记一行只是噪音）。 +// +// 未接线（没有写入面）时记一行 INFO 说明原因后安静返回：这既不是错误也不是降级， +// 只是「这份构建没有留存面」。真出错（构造失败）记 WARN —— 绝不静默。 +func (h *hub) startChannelEventWatch(ctx context.Context) { + if h == nil { + return + } + if h.eventStore == nil { + h.log.Info("channel event watch disabled", "reason", "no writable event store wired") + return + } + w, err := events.NewWatcher(events.WatcherConfig{ + Store: h.eventStore, + Source: statusSourceFunc(h.accountStatus), + Publisher: h.bus, + Log: h.log, + }) + if err != nil { + h.log.Warn("channel event watch disabled", "err", err) + return + } + h.channelWatch = w + w.Start(ctx) + h.log.Info("channel event watch started", + "interval", w.Interval().String(), + "publisher", h.bus != nil, + "source", h.accountStatus != nil) +} diff --git a/internal/services/gateway/ws_channel_events_test.go b/internal/services/gateway/ws_channel_events_test.go new file mode 100644 index 0000000..6124a61 --- /dev/null +++ b/internal/services/gateway/ws_channel_events_test.go @@ -0,0 +1,382 @@ +package gateway + +// 本文件钉住渠道事件**实时面**的网关侧（design/26 §2.6 / §4.5 / §6.5；切片 E7）： +// +// - 账号状态**快照**的三态口径：确知在线 / 确知离线 / **尚未探到**都要进快照 +// （`liveAccountOnline` 会把未知整类丢掉，那是面板口径；留存面必须收）， +// 而没有那条状态面的平台不进快照（「没有这条信号」≠「我们探不到」）； +// - 写入面按**能力断言**取：未接线的只读替身 ⇒ 实时面不启动（不伪造替身）； +// - **真走法**端到端：能力注册表里的账号状态迁移 → 落一行 T2 → 上进程内 bus → +// hub 的 relay → `channel_event` 帧；owner/admin 收得到，**viewer 收不到** +// （frameReachesAll 的 fail-closed 默认档，design/11 §7.1）。 + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" +) + +// watchStubAdapter 是带**可变、并发安全**账号状态面的渠道替身：它模拟 +// `*qq_obv11.Adapter`（`AccountStatus()` + 探活首帧之前 known=false）。 +// 加锁是必要的 —— 测试从另一条 goroutine 改状态，采样循环在自己的 goroutine 上读。 +type watchStubAdapter struct { + *rescanStubAdapter + mu sync.Mutex + online bool + known bool +} + +func (a *watchStubAdapter) AccountStatus() (bool, bool) { + a.mu.Lock() + defer a.mu.Unlock() + return a.online, a.known +} + +// set 改一次底层状态（模拟探活返回了新的 online/known）。 +func (a *watchStubAdapter) set(online, known bool) { + a.mu.Lock() + a.online, a.known = online, known + a.mu.Unlock() +} + +var _ adapter.PlatformAdapter = (*watchStubAdapter)(nil) +var _ accountStatusAware = (*watchStubAdapter)(nil) + +// newWatchStub 构造一个已注册形态的渠道替身（与组合根一致：包在 namedAdapter 里， +// 因此网关必须拆开内嵌字段才能抵达状态面）。 +func newWatchStub(name string, online, known bool) *watchStubAdapter { + return &watchStubAdapter{ + rescanStubAdapter: &rescanStubAdapter{name: name, connected: true}, + online: online, + known: known, + } +} + +// registerChannelStub 把一个渠道替身按组合根的形态注册进注册表。 +func registerChannelStub(t *testing.T, caps *capreg.Registry, stub adapter.PlatformAdapter, name string) { + t.Helper() + if err := caps.Register(capreg.Capability{ + Name: name, + Kind: capreg.KindChannel, + Value: namedAccountAdapter{PlatformAdapter: stub}, + }); err != nil { + t.Fatalf("register %s: %v", name, err) + } +} + +// TestChannelAccountStatusSamples_三态齐全且不丢未知 是这条能力面的口径测试： +// 三个 qq-obv11 账号（在线 / 离线 / 尚未探到）都必须进快照，且未知的样本 +// **必须带着 Known=false** 出来（它会被归一化成 `unknown` 并落行）； +// 没有状态面的平台（telegram）不进快照。 +func TestChannelAccountStatusSamples_三态齐全且不丢未知(t *testing.T) { + t.Parallel() + caps := capreg.New() + registerChannelStub(t, caps, newWatchStub("qq-obv11:aa", true, true), "qq-obv11:aa") + registerChannelStub(t, caps, newWatchStub("qq-obv11:bb", false, true), "qq-obv11:bb") + registerChannelStub(t, caps, newWatchStub("qq-obv11:cc", false, false), "qq-obv11:cc") + // telegram 没有独立账号状态面：它不进快照（不伪造一个 unknown）。 + registerChannelStub(t, caps, &rescanStubAdapter{name: "telegram:dd", connected: true}, "telegram:dd") + + srv := buildServerOnly(t, cfgDefault(), Deps{Auth: mustJWT(t), Registry: caps}) + got := srv.channelAccountStatusSamples() + if len(got) != 3 { + t.Fatalf("快照 %d 条, want 3（未知也要收，没有状态面的平台不收）: %+v", len(got), got) + } + // 顺序确定（按平台 + 账号排序），因此可以按下标断言。 + want := []struct { + account string + online bool + known bool + }{{"aa", true, true}, {"bb", false, true}, {"cc", false, false}} + for i, w := range want { + if got[i].Platform != events.PlatformQQ || got[i].AccountID != w.account { + t.Fatalf("快照[%d] = %s/%s, want qq-obv11/%s", i, got[i].Platform, got[i].AccountID, w.account) + } + if got[i].Online != w.online || got[i].Known != w.known { + t.Errorf("快照[%d] = online:%v known:%v, want online:%v known:%v", + i, got[i].Online, got[i].Known, w.online, w.known) + } + if got[i].Source != events.StatusSourceProbe { + t.Errorf("快照[%d].Source = %q, want %q", i, got[i].Source, events.StatusSourceProbe) + } + } +} + +// TestChannelAccountStatusSamples_注册表缺席时为空：未接注册表（只可能出现在 +// 精简的测试构造里）⇒ 空快照，而不是 panic 或一条编出来的样本。 +func TestChannelAccountStatusSamples_注册表缺席时为空(t *testing.T) { + t.Parallel() + srv := buildServerOnly(t, cfgDefault(), Deps{Auth: mustJWT(t)}) + if got := srv.channelAccountStatusSamples(); len(got) != 0 { + t.Fatalf("快照 = %+v, want 空", got) + } +} + +// TestChannelEventWriter_只读替身不给写入面：`Deps.EventStore` 声明的是只读查询面， +// 组合根实际注入的是完整存储 ⇒ 能力断言才拿得到写入面。拿不到时返回**真 nil**， +// 实时面据此不启动（绝不包一个写不进去的替身）。 +func TestChannelEventWriter_只读替身不给写入面(t *testing.T) { + t.Parallel() + // ① 只读替身（E6 的查询面测试就是用它）⇒ 没有写入面。 + readOnly := buildServerOnly(t, cfgDefault(), Deps{Auth: mustJWT(t), EventStore: &channelEventStoreStub{}}) + if got := readOnly.channelEventWriter(); got != nil { + t.Fatalf("channelEventWriter(只读替身) = %#v, want nil", got) + } + if readOnly.hub.channelWatch != nil { + t.Error("只读替身下实时面仍被启动（应当整个不启动）") + } + // ② 未接线（真 nil 接口）⇒ 同样 nil。 + unwired := buildServerOnly(t, cfgDefault(), Deps{Auth: mustJWT(t)}) + if got := unwired.channelEventWriter(); got != nil { + t.Fatalf("channelEventWriter(未接线) = %#v, want nil", got) + } + // ③ 真存储 ⇒ 写入面可用，且实时面真的起来了。 + _, store := channelEventTestStore(t, t.TempDir()) + wired := buildServerOnly(t, cfgDefault(), Deps{Auth: mustJWT(t), EventStore: store}) + if wired.channelEventWriter() == nil { + t.Fatal("channelEventWriter(真存储) = nil, want 写入面") + } + if wired.hub.channelWatch == nil { + t.Fatal("真存储下实时面未启动（账号状态迁移将不会留痕）") + } + wired.hub.close() +} + +// TestChannelEventFrame_真实主题只投给owner与admin：在真实的生产主题 +// `events.SubjectChannelEvent` 上发布一帧，经 bus → relay → 帧门 → 连接。 +// viewer 只收到哨兵（状态机信号），绝不收到渠道事件帧 —— 这是 design/11 §7.1 +// 的 fail-closed 默认档，`channel_event` 不在白名单里（也不该在）。 +func TestChannelEventFrame_真实主题只投给owner与admin(t *testing.T) { + t.Parallel() + b := bus.New() + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Bus: b}) + ownerConn := dialWS(t, ts.URL, signAny(t, jwt, "owner", auth.RoleOwner)) + defer ownerConn.Close() + adminConn := dialWS(t, ts.URL, signAny(t, jwt, "admin", auth.RoleAdmin)) + defer adminConn.Close() + viewerConn := dialWS(t, ts.URL, signAny(t, jwt, "viewer", auth.RoleViewer)) + defer viewerConn.Close() + // 连接注册发生在服务端的升级处理里，dial 返回不等于已注册；不等它，发布就会 + // 打在「零连接」上（帧投递按当时的连接集合快照进行）。 + waitConnCount(t, srv, 3) + + payload := events.ChannelEventFrame{Event: events.Event{ + Platform: events.PlatformQQ, AccountID: "bot-1", Tier: events.TierT2, + Category: events.CategoryLifecycle, PlatformType: events.StatusSourceProbe, + ActorRole: events.RoleSystem, SelfRelated: true, Summary: "账号离线（在线→离线）", + }.JSON()} + if err := b.Publish(context.Background(), events.SubjectChannelEvent, payload); err != nil { + t.Fatalf("publish: %v", err) + } + // 哨兵：bus 的每个订阅按序处理，因此看到它即代表它之前的帧已投递完（把 + // 「viewer 没收到别的」从猜测的空闲窗口变成精确判定）。 + publishSentinel(t, b) + + // owner/admin 各收 2 帧（渠道事件 + 哨兵），viewer 只收哨兵 —— 条数本身就是断言。 + owner := readTypedFrames(t, ownerConn, 2) + admin := readTypedFrames(t, adminConn, 2) + viewer := readTypedFrames(t, viewerConn, 1) + for _, tc := range []struct { + role string + typed []typedFrame + }{{"owner", owner}, {"admin", admin}} { + frame, ok := findTypedFrame(tc.typed, wsChannelEvent) + if !ok { + t.Fatalf("%s 未收到 %s 帧（收到 %v）", tc.role, wsChannelEvent, typedFrameTypes(tc.typed)) + } + var data struct { + Event events.ChannelEventJSON `json:"event"` + } + if err := json.Unmarshal(frame.Data, &data); err != nil { + t.Fatalf("%s 的 %s 载荷解不出: %v (raw=%s)", tc.role, wsChannelEvent, err, frame.Data) + } + if data.Event.Summary != "账号离线（在线→离线）" || data.Event.Platform != events.PlatformQQ { + t.Errorf("%s 帧载荷 = %+v, want 与发布的事件一致", tc.role, data.Event) + } + } + if _, ok := findTypedFrame(viewer, wsChannelEvent); ok { + t.Fatalf("viewer 收到了 %s 帧（收到 %v）：它含渠道元数据，属 fail-closed 默认档", + wsChannelEvent, typedFrameTypes(viewer)) + } + if _, ok := findTypedFrame(viewer, EventSessionIdle); !ok { + t.Fatalf("viewer 没收到哨兵 %s（收到 %v）", EventSessionIdle, typedFrameTypes(viewer)) + } +} + +// TestChannelEventWatch_迁移落行且只推frames给owner与admin 是这条腿的**真走法**： +// 组合根形态的注册表 + 真 SQLite 事件存储 + 真 bus + 真 WS 连接。 +// +// 它一次钉住三件事： +// - 服务构造时实时面**常驻启动**，第一拍（首次观测）就把账号状态落了行 +// —— 「没人看面板时也留痕」这条要求正是靠它成立； +// - 状态**迁移**跟着落第二行，并把那一行推成帧（owner 收得到，viewer 收不到）； +// - 每拍都落行会把库写胖（§4.3），所以第三拍（状态没变）**不多落行、不多推帧**。 +func TestChannelEventWatch_迁移落行且只推frames给owner与admin(t *testing.T) { + t.Parallel() + _, store := channelEventTestStore(t, t.TempDir()) + caps := capreg.New() + stub := newWatchStub("qq-obv11:bot-1", true, false) // 尚未探到 ⇒ 首拍落「未知」 + registerChannelStub(t, caps, stub, "qq-obv11:bot-1") + + b := bus.New() + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Registry: caps, EventStore: store, Bus: b}) + if srv.hub.channelWatch == nil { + t.Fatal("实时面未启动（真存储 + 注册表都在）") + } + ctx := context.Background() + // 首拍（构造期已立即采过一次）：unknown 也是一次迁移、也要留痕。 + waitRowCount(t, store, 1) + + adminConn := dialWS(t, ts.URL, signAny(t, jwt, "admin", auth.RoleAdmin)) + defer adminConn.Close() + viewerConn := dialWS(t, ts.URL, signAny(t, jwt, "viewer", auth.RoleViewer)) + defer viewerConn.Close() + waitConnCount(t, srv, 2) + + // 探活第一次成功、账号在线 ⇒ 一次真实迁移（未知 → 在线）。 + stub.set(true, true) + srv.hub.channelWatch.ObserveOnce(ctx) + waitRowCount(t, store, 2) + + rows, err := store.List(ctx, events.Filter{Platform: events.PlatformQQ}) + if err != nil { + t.Fatalf("List: %v", err) + } + latest := rows[0] + if latest.AccountID != "bot-1" || latest.Tier != events.TierT2 || + latest.Category != events.CategoryLifecycle || latest.PlatformType != events.StatusSourceProbe { + t.Fatalf("最新一行 = %+v, want bot-1 / t2 / lifecycle / get_status", latest) + } + if latest.Summary != "账号上线（未知→在线）" { + t.Errorf("summary = %q, want %q", latest.Summary, "账号上线（未知→在线）") + } + if !latest.SelfRelated || latest.ActorRole != events.RoleSystem { + t.Errorf("行 = %+v, want self_related + actor_role=system", latest) + } + + publishSentinel(t, b) + admin := readTypedFrames(t, adminConn, 2) + viewer := readTypedFrames(t, viewerConn, 1) + frame, ok := findTypedFrame(admin, wsChannelEvent) + if !ok { + t.Fatalf("admin 未收到 %s 帧（收到 %v）", wsChannelEvent, typedFrameTypes(admin)) + } + var data struct { + Event events.ChannelEventJSON `json:"event"` + } + if err := json.Unmarshal(frame.Data, &data); err != nil { + t.Fatalf("解载荷: %v (raw=%s)", err, frame.Data) + } + if data.Event.Summary != latest.Summary || data.Event.AccountID != "bot-1" { + t.Errorf("帧载荷 = %+v, want 与刚落的那一行一致（summary=%q）", data.Event, latest.Summary) + } + if data.Event.ID != 0 { + t.Errorf("实时帧带上了 id=%d，want 0/缺席（写侧还没回读那一行的 id）", data.Event.ID) + } + if _, ok := findTypedFrame(viewer, wsChannelEvent); ok { + t.Fatalf("viewer 收到了渠道事件帧（收到 %v）", typedFrameTypes(viewer)) + } + + // 第三拍：状态没变 ⇒ 不落行、不推帧（§4.3 的「绝不逐帧写库」）。 + srv.hub.channelWatch.ObserveOnce(ctx) + time.Sleep(30 * time.Millisecond) + rows2, err := store.List(ctx, events.Filter{}) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(rows2) != 2 { + t.Errorf("库里 %d 行, want 2（状态未变不该多落）", len(rows2)) + } + srv.hub.close() +} + +// typedFrame 是一帧的原始类型与载荷（读连接用）。 +type typedFrame struct { + Type string + Data json.RawMessage +} + +// readTypedFrames 读满 want 帧（每帧各自带读超时）。 +// +// 为什么按**条数**读、而不按「读到哨兵为止」：哨兵发在 `ws.>` 主题上，而被测 +// 帧发在它自己的领域主题上 —— 两个订阅各有独立的 worker，bus 只保证**同一订阅内** +// 的顺序。按哨兵截断会与另一条订阅赛跑（实测：哨兵先到，帧被截在窗口外）。 +// 条数则是确定的：本用例在每条连接上发布的帧数可以精确数出来。 +func readTypedFrames(t *testing.T, conn *websocket.Conn, want int) []typedFrame { + t.Helper() + out := make([]typedFrame, 0, want) + for len(out) < want { + _ = conn.SetReadDeadline(time.Now().Add(3 * time.Second)) + _, msg, err := conn.ReadMessage() + if err != nil { + t.Fatalf("读第 %d/%d 帧失败（已收 %v）: %v", len(out)+1, want, typedFrameTypes(out), err) + } + var raw struct { + Type string `json:"type"` + Data json.RawMessage `json:"data"` + } + if err := json.Unmarshal(msg, &raw); err != nil { + t.Fatalf("解帧失败 %s: %v", msg, err) + } + out = append(out, typedFrame{Type: raw.Type, Data: raw.Data}) + } + return out +} + +// publishSentinel 发布批次哨兵（`session_idle` 是白名单状态信号，人人都收）。 +func publishSentinel(t *testing.T, b bus.Bus) { + t.Helper() + if err := b.Publish(context.Background(), wsPrefix+EventSessionIdle, map[string]any{"sessionId": "web"}); err != nil { + t.Fatalf("publish sentinel: %v", err) + } +} + +// findFrame 在收到的帧里找指定类型。 +func findTypedFrame(frames []typedFrame, typ string) (typedFrame, bool) { + for _, f := range frames { + if f.Type == typ { + return f, true + } + } + return typedFrame{}, false +} + +// frameTypesOf 把帧类型拼成可读串（失败信息用）。 +func typedFrameTypes(frames []typedFrame) string { + types := make([]string, 0, len(frames)) + for _, f := range frames { + types = append(types, f.Type) + } + return "[" + strings.Join(types, " ") + "]" +} + +// waitRowCount 轮询直到库里的事件行数达到 want（首拍的落行发生在另一条 goroutine +// 上，因此不能假设它已经完成 —— 但也绝不靠固定 sleep 猜）。 +func waitRowCount(t *testing.T, store events.Store, want int) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + var last int + for time.Now().Before(deadline) { + n, err := store.Count(context.Background(), events.Filter{}) + if err != nil { + t.Fatalf("Count: %v", err) + } + last = int(n) + if last >= want { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("事件行数 = %d, want ≥%d", last, want) +} diff --git a/internal/services/gateway/ws_frame_scope_test.go b/internal/services/gateway/ws_frame_scope_test.go index 9290781..5164ba8 100644 --- a/internal/services/gateway/ws_frame_scope_test.go +++ b/internal/services/gateway/ws_frame_scope_test.go @@ -119,6 +119,12 @@ func frameScopeCases() []frameScopeCase { relayed("mood", false), relayed("device_update", false), relayed("reminder", false), + // 渠道事件实时帧（design/26 §2.6 / §4.5；切片 E7）：载荷是 `channel_events` + // 的一行（平台 / 账号 / 类别 / 摘要…），属**管理面元数据** —— 不进白名单， + // 按 fail-closed 默认档只发 owner + admin。它真实的生产主题是 + // `events.SubjectChannelEvent`（"channel.event"，带点），端到端路径由 + // ws_channel_events_test.go 单独驱动；这一行钉的是**帧名**的分类。 + relayed(wsChannelEvent, false), } } diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index f02416f..042be74 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -12,6 +12,7 @@ import ( "github.com/gorilla/websocket" "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" ) // WebSocket 调参常量。ping/pong 节奏用于防止中间代理把空闲连接断开。 @@ -66,12 +67,22 @@ type hub struct { // 解析器同源，使 WS 帧与 REST 的文档校验口径一致。为 nil 时带附件的回合 // 被明确拒绝（见 resolveDocuments），绝不静默丢弃。 documents documentResolver - log *slog.Logger - upgrader websocket.Upgrader - origins map[string]bool - allowAll bool - cancel context.CancelFunc - streamSeq atomic.Uint64 + // eventStore 是渠道事件的**写入面**（design/26 §2.6；切片 E7）。为 nil 表示 + // 这份构建没有留存面 ⇒ 渠道事件实时面整个不启动（见 startChannelEventWatch）。 + eventStore events.Store + // accountStatus 采集各账号的在线状态快照（含未知）。它是纯函数式的取数口， + // 由 server 提供（见 channelAccountStatusSamples），hub 只负责按节拍调用。 + accountStatus func() []events.StatusSample + // channelWatch 是渠道事件采样循环的句柄（未接线时为 nil）。持有它有两个理由： + // close() 之外还想让「循环到底起没起」有一个可断言的对象；测试可经 + // ObserveOnce 在**确定的一拍**上驱动迁移，而不必等 60s。 + channelWatch *events.Watcher + log *slog.Logger + upgrader websocket.Upgrader + origins map[string]bool + allowAll bool + cancel context.CancelFunc + streamSeq atomic.Uint64 } // newHub 构建一个绑定到 server 依赖的 hub。 @@ -88,9 +99,13 @@ func newHub(s *Server) *hub { history: s.history, envResolver: s.envResolver, documents: s.resolveChatDocuments, - log: s.log, - origins: origins, - allowAll: allowAll, + // 渠道事件实时面的两个来源都在构造期取定：写入面按能力断言取（未接线时为 + // 真 nil ⇒ 循环不启动），状态快照直接引用 server 的方法。 + eventStore: s.channelEventWriter(), + accountStatus: s.channelAccountStatusSamples, + log: s.log, + origins: origins, + allowAll: allowAll, upgrader: websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, @@ -101,13 +116,16 @@ func newHub(s *Server) *hub { // start 让 hub 订阅 bus 事件流。发往 ws.> 的事件到达此处后广播给客户端； // 生产者桥接模式（iot/task/automation/reminder/subagent-progress）也一并订阅， -// 并归一为 ws.* 事件类型。未接入 bus 时什么都不做。 +// 并归一为 ws.* 事件类型。它同时启动**渠道事件实时面**（切片 E7）：那一条腿 +// 只依赖写入面与状态源，没有 bus 也照常采（落行不推帧），因此不能挂在 bus 的 +// 判定之后。ctx 与订阅共用：close() 取消它，两条腿一起干净退出。 func (h *hub) start() { + ctx, cancel := context.WithCancel(context.Background()) + h.cancel = cancel + h.startChannelEventWatch(ctx) if h.bus == nil { return } - ctx, cancel := context.WithCancel(context.Background()) - h.cancel = cancel patterns := append([]string{wsEventsPattern}, bridgePatterns...) for _, p := range patterns { if _, err := h.bus.Subscribe(ctx, p, h.relay); err != nil { @@ -116,8 +134,8 @@ func (h *hub) start() { } } -// close 取消 bus 订阅、中断所有 orchestrator 会话（使后台回合不会比 hub 活得 -// 更久），并关闭每条在线连接。 +// close 取消 bus 订阅与渠道事件采样循环、中断所有 orchestrator 会话（使后台回合 +// 不会比 hub 活得更久），并关闭每条在线连接。 func (h *hub) close() { if h.cancel != nil { h.cancel()