Cyrene_v2/internal/runtime/events/watcher.go

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