feat(channel): 微信 iLink 账号状态三态（诚实口径）
state ∈ 有效 / 被拒（需重扫）/ 未知；🔴 -14 是**被拒不是过期**，且**未知原因绝不写成「过期」** （面板文案与错误码都如实）。三态判定为纯函数 + 三态各有用例。 ⚠️ 未做（如实登记）：每账号逐条三态（accounts[] 数组上的三态）本片未做。
This commit is contained in:
AskaEth committed
1 parent
7505c46b70
commit
99a644d545
12 files changed
+2041
-57
No files matched your search
@@ -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 把内部键拼成 `"<platform>/<accountID>"`（日志与 States 的可读形态，
|
||||
// 与 channelbridge 的内存降级计数字典同形）。
|
||||
func statusKeyString(k statusKey) string {
|
||||
return k.platform + "/" + k.accountID
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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": <ChannelEventJSON>}`。
|
||||
//
|
||||
// 刻意不把事件字段摊平到 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)
|
||||
}
|
||||
}
|
||||
@@ -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),
|
||||
}
|
||||
}
|
||||
@@ -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：
|
||||
|
||||
@@ -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.* 主题保持
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user