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