AskaEth
99a644d545
state ∈ 有效 / 被拒（需重扫）/ 未知；🔴 -14 是**被拒不是过期**，且**未知原因绝不写成「过期」** （面板文案与错误码都如实）。三态判定为纯函数 + 三态各有用例。 ⚠️ 未做（如实登记）：每账号逐条三态（accounts[] 数组上的三态）本片未做。
383 lines
16 KiB
Go
383 lines
16 KiB
Go
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)
|
||
}
|