Cyrene_v2/cmd/cyrene/wire_flow_graphs.go

package main

// 本文件把**链路图落盘**接进组合根（design/19 §4 P3；切片 F2）：按配置构造链路图
// 存储，交给发布器当「每轮收尾落一条」的出口，并拉起常驻的保留期清扫循环。网关的
// 只读出口（契约 v0.73 的 `?turnId=` 与 `turns[]`）在 wire_frontend.go 里注入**同一个**
// 存储句柄，因此「写进去的」与「查得到的」永远是一份状态。
//
// 三条取舍逐字对齐参照实现 `wire_channel_events.go` / `wire_document_gc.go`
// （design 明写「不另造」）：
//
//  1. **启动期清扫不阻塞启动**：它是常驻 goroutine 的第一件事，而不是在 newApp/run 里
//     同步等它跑完。
//  2. **失败不退出进程**：清扫失败只记一条 WARN 并等下一轮；一轮失败只是「这一轮没
//     扫干净」，绝不让进程退出。
//  3. **关闭 = 显式配 ≤0**：`FLOW_GRAPH_RETENTION=0`（或负值）或
//     `FLOW_GRAPH_GC_INTERVAL=0` 都不开启清扫，NewStore 会把关闭原因记进日志（静默的
//     「配了却不生效」不存在）。
//
// ⚠️ 与 channelEventKeeper 的一处刻意留白：本 keeper 只在 ctx 取消时退出循环（不额外
// 挂退出钩子）—— 与 `wire_channel_events.go` 的已知差异一致，因为 `wire_services.go`
// 由别的切片在改（本片不许动）。清扫本就是幂等且尽力而为的。
//
// 两个键的读取走 `config.LoadDurationEnv`（进程环境 > `.env` > 本包导出的默认常量）：
// `.env` 不写回进程环境，直接 os.Getenv 会让只写在 `.env` 里的声明静默失效。

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

	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/flowgraph"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway"
)

// 两个 env 键名：必须与 `.env.example` **逐字一致**（AGENTS §10 的 E6 条目）。
const (
	flowGraphRetentionEnv  = "FLOW_GRAPH_RETENTION"
	flowGraphGCIntervalEnv = "FLOW_GRAPH_GC_INTERVAL"
)

// wireFlowGraphs 构造链路图存储并挂到 app 上（随后 wireRuntime 把它交给发布器作落盘
// 出口，wireFrontend 把它交给网关作只读查询出口）。它**不**启动任何 goroutine：常驻
// 循环由 main 的 startFlowGraphs 在子系统启动之后拉起（与渠道事件留存的分工逐字相同）。
func (a *app) wireFlowGraphs() {
	a.flowStore = a.newFlowGraphStore(envPath)
}

// newFlowGraphStore 按配置构造链路图存储。库未接（`a.db == nil`，只可能出现在测试
// 夹具里）时返回 nil：发布器随即走「不落盘」的纯内存路径（F1 之前的行为），会话收尾
// 照常，绝不因为没有存储而失败或改变实时帧。
//
// 默认值一律取 `internal/runtime/flowgraph` 导出的常量 —— **不在这里另写一份字面量**：
// 两处各写一份 168h/1h，改一处就会让「文档承诺的默认值」与「代码里的默认值」漂移。
func (a *app) newFlowGraphStore(envFilePath string) *flowgraph.SQLStore {
	if a.db == nil {
		a.log.Warn("flow graph store skipped: no database connection")
		return nil
	}
	store, err := flowgraph.NewStore(flowgraph.Config{
		DB:         a.db,
		Log:        a.log,
		GCInterval: durationEnv(a.log, envFilePath, flowGraphGCIntervalEnv, flowgraph.DefaultGCInterval),
		Retention:  durationEnv(a.log, envFilePath, flowGraphRetentionEnv, flowgraph.DefaultRetention),
	})
	if err != nil {
		// NewStore 唯一的失败是 DB 为 nil（上面已挡掉）：仍如实处理，绝不吞错。
		a.log.Warn("flow graph store unavailable", "err", err)
		return nil
	}
	return store
}

// flowRoundRecorder 返回注入发布器的落盘出口：未接线时返回**真正的 nil 接口**，而不是
// 「装着 nil 指针的接口」—— 后者在 Go 里不等于 nil，会让发布器的 `p.record != nil`
// 判据失效（症状是每一轮收尾都撞一次 nil 接收者并记一条 WARN）。
func (a *app) flowRoundRecorder() loop.RoundRecorder {
	if a.flowStore == nil {
		return nil
	}
	return a.flowStore
}

// flowGraphStore 返回注入网关的只读出口：未接线时同样交出**真 nil 接口**（与
// channelEventStore 同一取舍）。发布器与网关拿到的是**同一个**存储句柄。
func (a *app) flowGraphStore() gateway.FlowGraphStore {
	if a.flowStore == nil {
		return nil
	}
	return a.flowStore
}

// startFlowGraphs 拉起链路图留存的维护循环（未接线时是空操作）。
func (a *app) startFlowGraphs(ctx context.Context) {
	if a.flowStore == nil {
		a.log.Info("flow graph upkeep disabled: store not wired",
			"effect", "rounds are kept in memory only (no local history)")
		return
	}
	keeper := &flowGraphKeeper{
		store: a.flowStore,
		log:   logger.WithSource(a.log, "cmd/cyrene/wire_flow_graphs"),
		done:  make(chan struct{}),
	}
	a.flowGraphKeeper = keeper
	go keeper.loop(ctx)
}

// flowGraphKeeper 是链路图留存的常驻维护任务（保留期清扫）。它持有**存储句柄**
// （不是 app），因此生命周期与用法一目了然。
type flowGraphKeeper struct {
	store *flowgraph.SQLStore
	log   *slog.Logger
	// done 在循环退出时关闭（让退出路径可观测）。
	done chan struct{}
}

// loop 是维护循环：启动期先清扫一次，此后按间隔定时清扫，直到 ctx 被取消。
// GC 关闭时启动期那次是空操作，循环随即返回（不留空转 goroutine）。
func (k *flowGraphKeeper) loop(ctx context.Context) {
	defer close(k.done)
	k.sweep(ctx)
	ticks, stop := k.gcTicker()
	defer stop()
	if ticks == nil {
		return
	}
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticks:
			k.sweep(ctx)
		}
	}
}

// gcTicker 给出清扫的定时通道；GC 关闭时返回 **nil 通道**（select 上永不触发）。
// `time.NewTicker(0)` 会 panic，所以「配了 ≤0 = 关闭」必须**在这里**就分岔，而不是
// 建一个零间隔 ticker 再指望它不跑。
func (k *flowGraphKeeper) 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 自己记（含删除行数与批次），
// 这里只负责把「这一轮整个没跑成」记成 WARN 并让循环继续。
func (k *flowGraphKeeper) sweep(ctx context.Context) {
	if _, err := k.store.Sweep(ctx); err != nil {
		k.log.Warn("flow graph sweep failed", "err", err)
	}
}