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()
}