Cyrene_v2/internal/services/channelbridge/reload.go

package channelbridge

// 本文件承载渠道桥接器的热重载机制（design/04
// §3.1）。面板把某平台的账号配置保存到 channels.json 并调用
// Bridge.Reload(platform)：工厂重建该平台的适配器，它们
// 先连接（新消息使用新配置），随后旧适配器
// 被优雅停止。已在途的消息仍在其旧适配器上完成，
// 因为消息处理是以桥接器生命周期 context 派生的，而非
// 重载所取消的按适配器读取 context。

import (
	"context"
	"errors"
	"fmt"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
	"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
)

// PlatformFactory 依据平台当前生效的账号配置
// （channels.json，回退到环境变量）重建其适配器。它以注入方式提供，
// 使桥接器能在面板保存新配置时热替换该平台的适配器
// 而无需重启。具体实现位于组合根
// （cmd/cyrene），平台适配器的构造函数由那里持有。
type PlatformFactory interface {
	// Build 依据生效的账号配置返回该平台的适配器。
	// 平台没有已配置账号时，返回的切片可能为空。
	Build(platform string) ([]adapter.PlatformAdapter, error)
}

// ChannelRegistry 是能力注册表的**窄端口**，只暴露热重载需要同步的两件事：
// 把某个名字的渠道条目换成新适配器、把已消失的账号条目删掉。桥接器用它让管理面板
// （渠道列表 / 在线状态 / open·close 解析）读到的永远是**当前**适配器实例，而不必
// 依赖 capability/registry 的具体类型 —— 注册表实现与桥接器因此保持解耦。
//
// 未接线（Options.Channels 为 nil）时桥接器照常热重载，但注册表条目会停在被保存前
// 的那个实例上：面板显示「未连接」、点「连接」驱动的是旧实例。此时只记一次 WARN。
type ChannelRegistry interface {
	// ReplaceChannel 用同名的新适配器覆盖（或新增）渠道条目。
	ReplaceChannel(ch sdk.Channel) error
	// RemoveChannel 删除该名字的渠道条目。
	RemoveChannel(name string) error
}

// ErrNilFactory 表示请求重载时未注入平台工厂
// （没有热重载配置来源）。
var ErrNilFactory = errors.New("channelbridge: nil platform factory")

// ErrBridgeClosed 表示在已停止的桥接器上请求重载。
var ErrBridgeClosed = errors.New("channelbridge: bridge is not running")

// connectAndRun 拨号单个适配器（受 connectTimeout 约束），成功时
// 装配其入站处理器，并在按适配器 context 下启动其读取
// 循环。热重载正是取消这个按适配器 context 来停止旧
// 读取循环；消息处理则以桥接器生命周期 context 派生，
// 因此 Reload 之前已在途的消息会在其旧适配器上完成。
func (b *Bridge) connectAndRun(a adapter.PlatformAdapter) {
	if a == nil {
		return
	}
	if !b.configured(a) {
		b.log.Warn("channel skipped: no credentials", "channel", a.Name())
		return
	}
	connCtx, cancel := context.WithTimeout(context.Background(), connectTimeout)
	err := a.Connect(connCtx)
	cancel()
	if err != nil {
		// 原始 dial 错误里带着整条拨号地址（`?access_token=…`），不能原样落 WARN；
		// connectFailureAttrs 只改文案，不动这里的判定与分支（见 connectlog.go）。
		b.log.Warn("channel connect failed, leaving disconnected",
			connectFailureAttrs(a.Name(), err)...)
		return
	}

	// 闭包捕获该适配器，使入站回调知道是哪个账号收到了
	// 消息——多账号回复路由，以及读取该账号的
	// 允许/拒绝名单都依赖它（design/16 §4.6 §5）。
	a.SetMessageHandler(func(_ context.Context, msg adapter.UnifiedMessage) {
		b.onMessage(b.ctx, msg, a)
	})
	b.mu.Lock()
	if b.closed {
		b.mu.Unlock()
		ctx, c := context.WithTimeout(context.Background(), connectTimeout)
		_ = a.Disconnect(ctx)
		c()
		return
	}
	readCtx, rcancel := context.WithCancel(b.ctx)
	b.readCancels[a.Name()] = rcancel
	b.connected[a.Name()] = true
	b.mu.Unlock()
	// 适配器已上线，此时缓存该账号的渠道级角色：
	// 入站路径读取缓存而不是配置，而重载（由面板保存
	// 触发）会重跑这里以拾取新的绑定。
	b.rememberAdapterPersona(a.Name())
	b.wg.Add(1)
	go b.readLoop(a, readCtx)
	b.log.Info("channel connected", "channel", a.Name())
}

// Reload 依据平台当前生效的配置（经由注入的工厂）重建其适配器，
// 并在不重启的前提下热替换正在运行的适配器
// （design/04 §3.1）。重建后的适配器先连接；到达旧适配器的
// 消息在旧适配器上完成，新消息则由新配置
// 接管。被移除的账号会停掉其旧适配器；名称未变的
// 账号在新适配器上线后停掉旧适配器。
func (b *Bridge) Reload(platform string) error {
	if b.factory == nil {
		return ErrNilFactory
	}
	b.mu.Lock()
	started := b.started
	closed := b.closed
	b.mu.Unlock()
	if !started || closed {
		return ErrBridgeClosed
	}

	newAdapters, err := b.factory.Build(platform)
	if err != nil {
		return fmt.Errorf("channelbridge: build %s: %w", platform, err)
	}
	old, oldCancels := b.adapterSnap(platform)
	newNames := make(map[string]bool, len(newAdapters))
	for _, a := range newAdapters {
		newNames[a.Name()] = true
	}
	for _, a := range newAdapters {
		b.connectAndRun(a)
		// 紧跟上线之后同步能力注册表：新实例先连上，面板才改为读它，
		// 避免「注册表已指向新实例、它却还没连」的窗口。
		b.replaceChannelEntry(a)
	}
	for _, o := range old {
		if newNames[o.Name()] {
			// 同名账号被重建后的适配器替换：只停旧实例的读取循环，
			// 注册表条目已由 replaceChannelEntry 指向新实例。
			b.cancelAndDisconnect(o, oldCancels[o.Name()])
			continue
		}
		// 账号被彻底移除（或更名）：停止它并删除其注册表
		// 条目，后续 snapshot 与面板便不再看到它。
		b.removeAdapter(o, oldCancels[o.Name()])
	}
	b.log.Info("channel reloaded", "platform", platform, "adapters", len(newAdapters))
	return nil
}

// adapterSnap 返回某平台当前已注册的适配器，以及
// 控制各自读取循环的 context 取消函数。Reload 用它
// 在重建的适配器上线后停止旧适配器。
func (b *Bridge) adapterSnap(platform string) ([]adapter.PlatformAdapter, map[string]context.CancelFunc) {
	b.mu.RLock()
	defer b.mu.RUnlock()
	var list []adapter.PlatformAdapter
	cancels := make(map[string]context.CancelFunc)
	for name, a := range b.adapters {
		if basePlatform(name) != platform {
			continue
		}
		list = append(list, a)
		if c, ok := b.readCancels[name]; ok {
			cancels[name] = c
		}
	}
	return list, cancels
}

// replaceChannelEntry 把刚上线的适配器写回能力注册表。注册表里的渠道条目只在启动
// 时写过一次，而热重载重建出来的是同名的新实例：不同步的话，管理面板（渠道列表、
// 账号在线状态、open/close 的目标解析）读到的仍是保存前的**旧实例** —— 卡片显示
// 「未连接」，点「连接」拨的是旧配置里的地址（design/04 §3.1）。
//
// 同步失败只记日志、绝不返回错误：面板保存不该被注册表同步拖垮。
func (b *Bridge) replaceChannelEntry(a adapter.PlatformAdapter) {
	if a == nil {
		return
	}
	name := a.Name()
	if !b.IsConnected(name) {
		// 没上线就不动注册表：换成连不上的实例只会让面板的在线状态更失真。
		b.log.Warn("channel registry left unchanged: adapter not connected", "channel", name)
		return
	}
	if b.channels == nil {
		b.warnChannelRegistryUnwired()
		return
	}
	if err := b.channels.ReplaceChannel(a); err != nil {
		b.log.Warn("channel registry replace failed", "channel", name, "error", err)
	}
}

// dropChannelEntry 在账号被移除（或更名）后删掉它的注册表条目，使管理面板不再列出
// 这条已经不存在的账号。同 replaceChannelEntry：失败只记日志。
func (b *Bridge) dropChannelEntry(name string) {
	if b.channels == nil {
		b.warnChannelRegistryUnwired()
		return
	}
	if err := b.channels.RemoveChannel(name); err != nil {
		b.log.Warn("channel registry remove failed", "channel", name, "error", err)
	}
}

// warnChannelRegistryUnwired 只在首次同步时提醒一次：未注入 Channels 端口时热重载
// 照常进行（适配器确实换了），只是面板读到的是保存前的旧实例。
func (b *Bridge) warnChannelRegistryUnwired() {
	b.channelsWarnOnce.Do(func() {
		b.log.Warn("channel registry not wired: admin panel keeps the pre-reload adapter instance")
	})
}

// cancelAndDisconnect 停止适配器的读取循环并拆除其会话。
// 它绝不改动桥接器自己的适配器表与能力注册表：替换同名适配器
// 的调用方希望（新）条目保留，而移除账号的调用方使用
// removeAdapter（它会连注册表条目一起清掉）。
func (b *Bridge) cancelAndDisconnect(o adapter.PlatformAdapter, c context.CancelFunc) {
	if c != nil {
		c()
	}
	ctx, cancel := context.WithTimeout(context.Background(), connectTimeout)
	_ = o.Disconnect(ctx)
	cancel()
	b.log.Info("channel (reload) stopped", "channel", o.Name())
}

// removeAdapter 停止账号的适配器并删除其注册表条目。它
// 只对未被 Reload 重建的名称调用，因此删除绝不会
// 误删如今占用该名称的替换适配器。
func (b *Bridge) removeAdapter(o adapter.PlatformAdapter, c context.CancelFunc) {
	b.cancelAndDisconnect(o, c)
	name := o.Name()
	b.mu.Lock()
	delete(b.adapters, name)
	delete(b.connected, name)
	delete(b.readCancels, name)
	b.mu.Unlock()
	b.forgetAdapterPersona(name)
	b.dropChannelEntry(name)
	b.log.Info("channel (reload) removed", "channel", name)
}