feat(config): 运行时配置体系②——渠道配置落 channels.json + 面板落盘 + 热更新
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- runtimeconfig/channels.go: channels.json 承载 map[platform][]ChannelAccount(token/baseURL/黑白名单/群聊限流), LoadChannels 优先文件/回退env, SaveChannels(0600+tmp原子写), Upsert/Delete/Masked脱敏
- channelbridge/reload.go: adapterManager 热替换——工厂重建新adapter先连接, 再原子断旧adapter(独立readCtx/进行中不中断); Bridge.Reload(platform) 逐账号增删
- 面板 PUT/DELETE /admin/channels/{platform}/config(owner+审计) 落盘channels.json+Reload热更新; GET /admin/channels 附账号掩码不泄裸token
- 前端 Channels.tsx 配置表单(id/token/baseURL/黑白名单/限流+config JSON)+保存热更新提示+掩码展示/删除
- .gitignore 加 channels.json; .env 退兜底(.env.example注明); token不打印/Config()脱敏
- channelbridge.go 拆 logic→reload.go(742→731); 全仓 go build/vet/test绿+前端build绿
- 未做真token端到端(需渠道凭据,不读.env)——如实标注待凭据
This commit is contained in:
Aska committed 2026-09-07 13:29:41 +08:00
1 parent 2d7151f979
commit aedae2190b
18 files changed
+1743 -126

No files matched your search

@@ -114,6 +114,10 @@ type Options struct {
// Now is the wall-clock source for quiet-hour judgement. Nil uses time.Now,
// which makes time-dependent behaviour injectable in tests.
Now func() time.Time
// Factory rebuilds a platform's adapters from its effective account config
// (channels.json → env fallback). When non-nil the bridge supports hot reload;
// nil keeps the bridge boot-only (Reload returns ErrNilFactory).
Factory PlatformFactory
}
// Bridge owns the channel adapter lifecycle and the inbound message pipeline. It
@@ -133,10 +137,13 @@ type Bridge struct {
quietEndHour int
now func() time.Time
mu sync.RWMutex
adapters map[string]adapter.PlatformAdapter
connected map[string]bool
closed bool
mu sync.RWMutex
adapters map[string]adapter.PlatformAdapter
connected map[string]bool
readCancels map[string]context.CancelFunc
started bool
closed bool
factory PlatformFactory
// gateMu guards the per-adapter flow gates (design/12 §6 策略三/四). They are
// built lazily on the first inbound message per account and are independent of
@@ -189,7 +196,9 @@ func New(opts Options) (*Bridge, error) {
now: now,
adapters: make(map[string]adapter.PlatformAdapter),
connected: make(map[string]bool),
readCancels: make(map[string]context.CancelFunc),
gates: make(map[string]*adapter.FlowGate),
factory: opts.Factory,
cancel: func() {},
sem: make(chan struct{}, maxConcurrent),
}, nil
@@ -236,42 +245,20 @@ func (b *Bridge) WebhookHandlers() map[string]http.Handler {
// Start connects every configured adapter, installs the inbound handler, and
// launches the read loops. Adapters without credentials (or whose Connect fails)
// are logged and left disconnected; they never fail startup. Start blocks only
// long enough to connect (bounded by connectTimeout per adapter) and returns.
// long enough to connect (bounded by connectTimeout per adapter) and returns. It
// is idempotent: a second call is a no-op.
func (b *Bridge) Start() {
b.mu.Lock()
if b.closed {
if b.closed || b.started {
b.mu.Unlock()
return
}
b.ctx, b.cancel = context.WithCancel(context.Background())
b.started = true
b.mu.Unlock()
for _, a := range b.snapshot() {
if !b.configured(a) {
b.log.Warn("channel skipped: no credentials", "channel", a.Name())
continue
}
connCtx, cancel := context.WithTimeout(context.Background(), connectTimeout)
err := a.Connect(connCtx)
cancel()
if err != nil {
b.log.Warn("channel connect failed, leaving disconnected",
"channel", a.Name(), "error", err)
continue
}
b.mu.Lock()
b.connected[a.Name()] = true
b.mu.Unlock()
// Close over the adapter so the inbound callback knows which account
// received the message — required for multi-account reply routing and for
// reading that account's allow/deny lists (design/16 §4.6 §5).
a.SetMessageHandler(func(ctx context.Context, msg adapter.UnifiedMessage) {
b.onMessage(ctx, msg, a)
})
b.wg.Add(1)
go b.readLoop(a)
b.log.Info("channel connected", "channel", a.Name())
b.connectAndRun(a)
}
b.log.Info("channel bridge started", "connected", b.ConnectedCount(),
"registered", len(b.snapshot()))
@@ -363,14 +350,16 @@ func basePlatform(name string) string {
return name
}
// readLoop runs an adapter's blocking read path until the bridge stops. The emit
// sink is wired to onMessage so poll-style adapters (Telegram getUpdates, weixin
// long poll) deliver through the same pipeline as push-style adapters.
func (b *Bridge) readLoop(a adapter.PlatformAdapter) {
// readLoop runs an adapter's blocking read path until the bridge stops or the
// adapter's own context is cancelled (a hot reload). The emit sink is wired to
// onMessage so poll-style adapters (Telegram getUpdates, weixin long poll)
// deliver through the same pipeline as push-style adapters. Message processing
// uses the bridge-lifetime context (not readCtx) so a reload leaves in-flight
// messages alone.
func (b *Bridge) readLoop(a adapter.PlatformAdapter, readCtx context.Context) {
defer b.wg.Done()
ctx := b.ctx
a.ReadMessages(ctx, func(msg adapter.UnifiedMessage) {
b.onMessage(ctx, msg, a)
a.ReadMessages(readCtx, func(msg adapter.UnifiedMessage) {
b.onMessage(b.ctx, msg, a)
})
}
+173
View File
@@ -0,0 +1,173 @@
package channelbridge
// This file holds the hot-reload machinery for the channel bridge (design/04
// §3.1). The panel saves a platform's account config to channels.json and calls
// Bridge.Reload(platform): the factory rebuilds that platform's adapters, they
// connect first (new messages use the new config), then the old adapters are
// gracefully stopped. A message already in flight finishes on its old adapter
// because processing is spawned with the bridge-lifetime context, not the
// per-adapter read context that reload cancels.
import (
"context"
"errors"
"fmt"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
)
// PlatformFactory rebuilds the adapters for a platform from its current effective
// account config (channels.json, falling back to env). It is injected so the
// bridge can hot-replace a platform's adapters when the panel saves a new config
// without a restart. The concrete implementation lives in the composition root
// (cmd/cyrene), which owns the platform adapter constructors.
type PlatformFactory interface {
// Build returns the adapters for platform from its effective account config.
// The returned slice may be empty when the platform has no configured account.
Build(platform string) ([]adapter.PlatformAdapter, error)
}
// ErrNilFactory is returned when a reload is requested without a platform factory
// injected (no hot-reload configuration source).
var ErrNilFactory = errors.New("channelbridge: nil platform factory")
// ErrBridgeClosed is returned when a reload is requested on a stopped bridge.
var ErrBridgeClosed = errors.New("channelbridge: bridge is not running")
// connectAndRun dials a single adapter (bound by connectTimeout) and, on success,
// installs its inbound handler and launches its read loop under a per-adapter
// context. The per-adapter context is what a hot reload cancels to stop the old
// read loop; message processing is spawned with the bridge-lifetime context so an
// in-flight message started before a Reload completes on its old adapter.
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 {
b.log.Warn("channel connect failed, leaving disconnected",
"channel", a.Name(), "error", err)
return
}
// Close over the adapter so the inbound callback knows which account received
// the message — required for multi-account reply routing and for reading that
// account's allow/deny lists (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.wg.Add(1)
go b.readLoop(a, readCtx)
b.log.Info("channel connected", "channel", a.Name())
}
// Reload rebuilds the adapters for a platform from its current effective config
// (via the injected factory) and hot-replaces the running ones without a restart
// (design/04 §3.1). The rebuilt adapters connect first; messages that arrived on
// the old adapters finish on them, and new messages are picked up by the new
// configuration. An account that was removed stops its old adapter, and an
// unchanged-name account stops the old adapter once the new one is live.
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)
}
for _, o := range old {
if newNames[o.Name()] {
// Same-name account replaced by the rebuilt adapter: stop the old read
// loop but leave the registry entry — it now points at the new adapter.
b.cancelAndDisconnect(o, oldCancels[o.Name()])
continue
}
// Account removed (or renamed) entirely: stop it and drop its registry
// entries, so a subsequent snapshot no longer sees it.
b.removeAdapter(o, oldCancels[o.Name()])
}
b.log.Info("channel reloaded", "platform", platform, "adapters", len(newAdapters))
return nil
}
// adapterSnap returns the adapters currently registered for a platform, together
// with the context-cancel func that controls each one's read loop. It is used by
// Reload to stop the old adapters after the rebuilt ones are live.
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
}
// cancelAndDisconnect stops an adapter's read loop and tears down its session.
// It never touches the registry map: callers that replace a same-name adapter
// want the (new) entry to stay, and callers that remove an account use
// 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 stops an account's adapter and drops its registry entries. It is
// only called for names that were NOT rebuilt by Reload, so the delete can never
// remove a replacement adapter that now owns the name.
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.log.Info("channel (reload) removed", "channel", name)
}
@@ -0,0 +1,152 @@
package channelbridge
import (
"context"
"errors"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
)
var errReload = errors.New("build failed")
// stubFactory returns a fixed adapter set for a platform, so a reload test can
// drive the hot-replace without real network I/O.
type stubFactory struct {
adpts []adapter.PlatformAdapter
err error
}
func (f *stubFactory) Build(_ string) ([]adapter.PlatformAdapter, error) { return f.adpts, f.err }
func newReloadBridge(t *testing.T, factory PlatformFactory) *Bridge {
t.Helper()
b, err := New(Options{
Loop: &stubChatter{res: &loop.Result{Text: "hi"}},
Sessions: &stubSessions{sess: session.Session{SessionID: "s", Owner: true, MemWritable: true}},
Owner: &user.OwnerProfile{},
Factory: factory,
})
if err != nil {
t.Fatalf("New: %v", err)
}
t.Cleanup(func() { b.Stop(context.Background()) })
return b
}
func adapterIn(b *Bridge, name string) bool {
for _, a := range b.snapshot() {
if a.Name() == name {
return true
}
}
return false
}
func TestReloadReplacesAdapterWithNewConfig(t *testing.T) {
t.Parallel()
old := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "old"}}
factory := &stubFactory{}
b := newReloadBridge(t, factory)
b.Add(old)
b.Start()
if !b.IsConnected("kook:conf0") {
t.Fatal("expected old adapter connected after Start")
}
newConf0 := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "new"}}
newConf1 := &fakeAdapter{name: "kook:conf1", cfg: map[string]any{"bot_token": "next"}}
factory.adpts = []adapter.PlatformAdapter{newConf0, newConf1}
if err := b.Reload("kook"); err != nil {
t.Fatalf("Reload: %v", err)
}
// The new adapters are connected and registered.
if !newConf0.IsConnected() || !newConf1.IsConnected() {
t.Fatalf("new adapters not connected: conf0=%v conf1=%v", newConf0.IsConnected(), newConf1.IsConnected())
}
if !b.IsConnected("kook:conf0") || !b.IsConnected("kook:conf1") {
t.Fatalf("bridge does not report new adapters connected")
}
// The old adapter instance was disconnected (its read loop stopped).
if old.IsConnected() {
t.Fatal("old adapter should be disconnected after reload")
}
}
func TestReloadRemovesRemovedAccount(t *testing.T) {
t.Parallel()
conf0 := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "t0"}}
conf1 := &fakeAdapter{name: "kook:conf1", cfg: map[string]any{"bot_token": "t1"}}
factory := &stubFactory{}
b := newReloadBridge(t, factory)
b.Add(conf0)
b.Add(conf1)
b.Start()
// New config only keeps conf1 (conf0 is deleted).
factory.adpts = []adapter.PlatformAdapter{
&fakeAdapter{name: "kook:conf1", cfg: map[string]any{"bot_token": "t1b"}},
}
if err := b.Reload("kook"); err != nil {
t.Fatalf("Reload: %v", err)
}
if adapterIn(b, "kook:conf0") {
t.Fatal("removed account's adapter should be dropped from the registry")
}
if !adapterIn(b, "kook:conf1") {
t.Fatal("kept account's adapter should still be registered")
}
if !b.IsConnected("kook:conf1") {
t.Fatal("kept account should be connected")
}
}
func TestReloadOtherPlatformUntouched(t *testing.T) {
t.Parallel()
tg := &fakeAdapter{name: "telegram", cfg: map[string]any{"bot_token": "tg"}}
kook := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "k"}}
factory := &stubFactory{}
b := newReloadBridge(t, factory)
b.Add(tg)
b.Add(kook)
b.Start()
factory.adpts = []adapter.PlatformAdapter{
&fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "k2"}},
}
if err := b.Reload("kook"); err != nil {
t.Fatalf("Reload: %v", err)
}
if !adapterIn(b, "telegram") || !tg.IsConnected() {
t.Fatal("reloading kook should leave the telegram adapter untouched")
}
}
func TestReloadNilFactoryErrors(t *testing.T) {
t.Parallel()
b := newReloadBridge(t, nil)
if err := b.Reload("kook"); err != ErrNilFactory {
t.Fatalf("err = %v, want ErrNilFactory", err)
}
}
func TestReloadBeforeStartErrors(t *testing.T) {
t.Parallel()
b := newReloadBridge(t, &stubFactory{})
if err := b.Reload("kook"); err != ErrBridgeClosed {
t.Fatalf("err = %v, want ErrBridgeClosed (not started)", err)
}
}
func TestReloadBuildError(t *testing.T) {
t.Parallel()
b := newReloadBridge(t, &stubFactory{err: errReload})
if err := b.Reload("kook"); err == nil {
t.Fatal("expected a build error to surface")
}
}