Cyrene_v2/cmd/cyrene/wire_channel_events.go

package main

// 本文件把**渠道事件留存**接进组合根（design/26 §4.3/§4.4；切片 E3+E4）：按配置构造
// 事件存储、交给桥接层注入（wire_channels.go 的 channelbridge.Options.EventStore），
// 拉起常驻的「保留期清扫 + 内存计数落盘」循环，并在退出前最后一次落盘。
//
// 三条取舍逐字对齐参照实现 `wire_document_gc.go`（design/26 §4.4 明写「不另造」）：
//
//  1. **启动期清扫不阻塞启动**：它是常驻 goroutine 的第一件事（紧随子系统启动之后
//     开始，网关监听之前就绪），而不是在 newApp/run 里同步等它跑完。
//  2. **失败不退出进程**：清扫 / 落盘失败只记一条 WARN 并等下一轮；一轮失败只是
//     「这一轮没扫干净」，绝不让进程退出。
//  3. **关闭 = 显式配 ≤0**：`CHANNEL_EVENT_RETENTION=0`（或负值）或
//     `CHANNEL_EVENT_GC_INTERVAL=0` 都不开启清扫，NewStore 会把关闭原因记进日志
//     （静默的「配了却不生效」不存在）。
//
// ⚠️ 与参照实现的一处**已知差异**（如实登记，不假装对齐）：docGC 的 `stop()` 由
// `wire_services.go` 的 close 调用，而该文件由别的切片在改（本片不许动）。因此这里的
// 退出路径做成**循环自己的 ctx.Done 分支**：ctx 取消即用独立超时的 context 落最后一次盘。
// 语义与「退出前最后一次 Flush」相同，区别只在**不阻塞** close —— 计数本就是尽力而为的
// 近似值（§4.3），并与关库并发（真撞上只记 WARN，不丢留存面本身）。
//
// 三个键的读取走 `config.LoadDurationEnv`（进程环境 > `.env` > E1 导出的默认常量）：
// `.env` 不写回进程环境，直接 os.Getenv 会让只写在 `.env` 里的声明静默失效。

import (
	"context"
	"log/slog"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events"
)

// 三个 env 键名：必须与 `.env.example` **逐字一致**（AGENTS §10 的 E6 条目）。
const (
	channelEventRetentionEnv    = "CHANNEL_EVENT_RETENTION"
	channelEventGCIntervalEnv   = "CHANNEL_EVENT_GC_INTERVAL"
	channelEventCounterFlushEnv = "CHANNEL_EVENT_COUNTER_FLUSH"
)

// channelEventFlushTimeout 是**退出前最后一次 Flush** 的时长上界。它必须配一个独立的
// context：ctx 在这一刻已被取消，拿它去写库只会立刻拿到 context.Canceled —— 那正是
// 「退出时最后一次 flush」最容易悄悄失效的地方。
const channelEventFlushTimeout = 5 * time.Second

// wireChannelEvents 构造事件留存面并挂到 app 上（随后由 wireChannels 注入桥接层）。
// 它**不**启动任何 goroutine：常驻循环由 main 的 startChannelEvents 在子系统启动之后
// 拉起（与文档 GC 的分工逐字相同）。
func (a *app) wireChannelEvents() {
	a.channelEvents = a.newChannelEventStore(envPath)
}

// newChannelEventStore 按配置构造事件存储。库未接（`a.db == nil`，只可能出现在测试
// 夹具里）时返回 nil：桥接层随即走「未接线」的 fail-visible 降级路径（内存计数 +
// 首次一条 DEBUG），入站流水线照常，绝不因为没有存储而失败或丢帧。
//
// 默认值一律取 E1（`internal/runtime/events`）导出的常量 —— **不在这里另写一份字面量**：
// 两处各写一份 720h/1h/30s，改一处就会让「文档承诺的默认值」与「代码里的默认值」漂移。
// `MaxPendingKeys` 没有 env 键（design/26 §4.4 只给了三个旋钮），同样取 E1 的默认常量。
func (a *app) newChannelEventStore(envFilePath string) *events.SQLStore {
	if a.db == nil {
		a.log.Warn("channel event store skipped: no database connection")
		return nil
	}
	store, err := events.NewStore(events.Config{
		DB:             a.db,
		Log:            a.log,
		FlushInterval:  durationEnv(a.log, envFilePath, channelEventCounterFlushEnv, events.DefaultFlushInterval),
		MaxPendingKeys: events.DefaultMaxPendingKeys,
		GCInterval:     durationEnv(a.log, envFilePath, channelEventGCIntervalEnv, events.DefaultGCInterval),
		Retention:      durationEnv(a.log, envFilePath, channelEventRetentionEnv, events.DefaultRetention),
	})
	if err != nil {
		// NewStore 唯一的失败是 DB 为 nil（上面已挡掉）：仍如实处理，绝不吞错。
		a.log.Warn("channel event store unavailable", "err", err)
		return nil
	}
	return store
}

// durationEnv 读一个 duration 型 env 键（进程环境 > `.env` > def）。配了但读不懂时记
// 一条带键名与原始取值的 WARN 并回落 def —— 错的是那一个键，不该让整条留存面失效
// （AGENTS §10：非法配置不许静默失效，但也不许一票否决）。消息刻意保持中立（不带
// 具体功能名），使渠道事件与链路图两处共用同一份读取口径（切片 E4 / F2）。
func durationEnv(log *slog.Logger, envFilePath, key string, def time.Duration) time.Duration {
	value, err := config.LoadDurationEnv(envFilePath, key, def)
	if err != nil {
		log.Warn("duration env ignored, falling back to default",
			"env", key, "error", err, "fallback", def.String())
	}
	return value
}

// channelEventStore 返回注入桥接层的留存面：未接线时返回**真正的 nil 接口**，而不是
// 「装着 nil 指针的接口」—— 后者在 Go 里不等于 nil，会让桥接层的「未接线 ⇒ 内存计数」
// 判据失效（症状是每帧都撞一次 nil 指针 panic）。
func (a *app) channelEventStore() events.Store {
	if a.channelEvents == nil {
		return nil
	}
	return a.channelEvents
}

// startChannelEvents 拉起留存维护循环（未接线时是空操作）。
func (a *app) startChannelEvents(ctx context.Context) {
	if a.channelEvents == nil {
		a.log.Info("channel event upkeep disabled: store not wired",
			"effect", "events are counted in memory only (bridged to the api layer counters)")
		return
	}
	keeper := &channelEventKeeper{
		store: a.channelEvents,
		log:   logger.WithSource(a.log, "cmd/cyrene/wire_channel_events"),
		done:  make(chan struct{}),
	}
	a.channelEventKeeper = keeper
	go keeper.loop(ctx)
}

// channelEventKeeper 是渠道事件留存的常驻维护任务：保留期清扫 + 内存计数落盘 +
// 退出前最后一次落盘。它持有**存储句柄**（不是 app），因此生命周期与用法一目了然。
type channelEventKeeper struct {
	store *events.SQLStore
	log   *slog.Logger
	// done 在循环退出时关闭。它让「退出前最后一次 Flush」**可被观测**：测试据此确认
	// 内存里的计数真的落了盘，而不是随 goroutine 一起消失。
	done chan struct{}
}

// loop 是维护循环：启动期先清扫一次、落一次盘，此后按各自间隔定时跑，直到 ctx 被取消
// （取消时落最后一次盘再退出）。
func (k *channelEventKeeper) loop(ctx context.Context) {
	defer close(k.done)
	k.sweep(ctx)
	k.flush(ctx)

	gcTicks, stopGC := k.gcTicker()
	defer stopGC()
	flushTicks := time.NewTicker(k.store.FlushInterval())
	defer flushTicks.Stop()

	for {
		select {
		case <-ctx.Done():
			k.flushOnShutdown()
			return
		case <-gcTicks:
			k.sweep(ctx)
		case <-flushTicks.C:
			k.flush(ctx)
		}
	}
}

// gcTicker 给出清扫的定时通道；GC 关闭时返回 **nil 通道**（select 上永不触发）。
// `time.NewTicker(0)` 会 panic，所以「配了 ≤0 = 关闭」必须在**这里**就分岔，而不是
// 建一个零间隔 ticker 再指望它不跑。
func (k *channelEventKeeper) gcTicker() (<-chan time.Time, func()) {
	if !k.store.Enabled() {
		return nil, func() {}
	}
	ticker := time.NewTicker(k.store.Interval())
	return ticker.C, ticker.Stop
}

// sweep 跑一轮保留期清扫。每轮的**成功**汇总由 store 自己记（E1 的 Sweep 已记一条，
// 含删除行数与批次），这里只负责把「这一轮整个没跑成」记成 WARN 并让循环继续。
func (k *channelEventKeeper) sweep(ctx context.Context) {
	if _, err := k.store.Sweep(ctx); err != nil {
		k.log.Warn("channel event sweep failed", "err", err)
	}
}

// flush 把内存里待落盘的计数写进库；失败只 WARN（计数是尽力而为的近似值，§4.3）。
func (k *channelEventKeeper) flush(ctx context.Context) {
	if err := k.store.Flush(ctx); err != nil {
		k.log.Warn("channel event counter flush failed", "err", err)
	}
}

// flushOnShutdown 是退出前的最后一次落盘：用独立超时的 context（此刻调用方的 ctx
// 已取消），失败只 WARN —— 退出路径绝不因为一次写库失败而 panic 或阻塞。
func (k *channelEventKeeper) flushOnShutdown() {
	ctx, cancel := context.WithTimeout(context.Background(), channelEventFlushTimeout)
	defer cancel()
	k.flush(ctx)
}