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