Cyrene_v2/internal/services/channelbridge/connect.go

package channelbridge

import (
	"context"
	"net/http"

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

// 本文件承载渠道桥接器的适配器注册表与连接生命周期
// （design/08 §0）：适配器经 Add 注册、由 Start 拨号
// （各自受 connectTimeout 约束）、由 Stop 拆除，并在不带
// 可用配置时被凭据闸门跳过。

// Add 按适配器的 Name 注册它。在 Start 之前调用是安全的；
// Start 之后再添加不影响已在运行的适配器。
func (b *Bridge) Add(a adapter.PlatformAdapter) {
	if a == nil {
		return
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.adapters[a.Name()] = a
}

// WebhookHandlers 返回每个暴露 HTTP 回调的已注册适配器
// （如 *webhook.Adapter）的入站 HTTP 处理器，键为适配器配置中的
// 端点路径。gateway 挂载它们，把外部 webhook 回调
// 接入运行时。当没有任何适配器提供入站处理器时
// 返回 nil map。
func (b *Bridge) WebhookHandlers() map[string]http.Handler {
	b.mu.RLock()
	defer b.mu.RUnlock()
	out := make(map[string]http.Handler)
	for _, a := range b.adapters {
		wh, ok := a.(webhookAdapter)
		if !ok {
			continue
		}
		path := wh.WebhookPath()
		if path == "" {
			// 适配器当前未暴露 webhook（例如 websocket 模式的
			// KOOK）；跳过它可避免把处理器挂在过期或冲突的路由上。
			continue
		}
		h := wh.InboundHandler()
		if h == nil {
			continue
		}
		out[path] = h
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

// Start 连接每个已配置的适配器、装配入站处理器并
// 启动读取循环。没有凭据（或 Connect 失败）的适配器
// 只记日志并保持未连接；它们绝不会导致启动失败。Start 只
// 阻塞到连接完成（每个适配器受 connectTimeout 约束）便返回。它
// 是幂等的：第二次调用无副作用。
func (b *Bridge) Start() {
	b.mu.Lock()
	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() {
		b.connectAndRun(a)
	}
	b.log.Info("channel bridge started", "connected", b.ConnectedCount(),
		"registered", len(b.snapshot()))
}

// Stop 取消读取循环、断开所有适配器，并等待
// 处理中的消息完成。它是幂等的。
func (b *Bridge) Stop(ctx context.Context) {
	b.mu.Lock()
	if b.closed {
		b.mu.Unlock()
		return
	}
	b.closed = true
	cancel := b.cancel
	b.mu.Unlock()

	cancel()
	for name, a := range b.snapshot() {
		if err := a.Disconnect(ctx); err != nil {
			b.log.Warn("channel disconnect", "channel", name, "error", err)
		}
	}
	b.wg.Wait()
	b.log.Info("channel bridge stopped")
}

// IsConnected 报告指定名称的适配器当前是否已连接。
func (b *Bridge) IsConnected(name string) bool {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return b.connected[name]
}

// ConnectedCount 报告当前已连接的适配器数量。
func (b *Bridge) ConnectedCount() int {
	b.mu.RLock()
	defer b.mu.RUnlock()
	n := 0
	for _, ok := range b.connected {
		if ok {
			n++
		}
	}
	return n
}

// snapshot 返回已注册适配器的副本，以便无锁遍历。
func (b *Bridge) snapshot() []adapter.PlatformAdapter {
	b.mu.RLock()
	defer b.mu.RUnlock()
	out := make([]adapter.PlatformAdapter, 0, len(b.adapters))
	for _, a := range b.adapters {
		out = append(out, a)
	}
	return out
}

// configured 报告适配器是否带有足够配置可供尝试
// Connect。它检查适配器自身的 Config() 快照来看各平台所需的
// 凭据，因此空壳（无配置）适配器会被跳过而不是拨号。
// 正是这一点避免了 weixin-ilink 在缺少 bot token 时
// 落入会阻塞的扫码登录流程。
func (b *Bridge) configured(a adapter.PlatformAdapter) bool {
	cfg := a.Config()
	switch basePlatform(a.Name()) {
	case "telegram", "weixin-ilink", "kook":
		return cfgStr(cfg, "bot_token") != ""
	case "qq-obv11":
		if cfgStr(cfg, "mode") == "client" {
			return cfgStr(cfg, "remoteURL") != ""
		}
		return cfgStr(cfg, "listenAddr") != ""
	case "webhook":
		// 仅入站：维持会话不需要外部凭据。
		return true
	default:
		return true
	}
}

// cfgStr 从适配器配置 map 中读取字符串字段，缺失时
// 返回 ""。
func cfgStr(m map[string]any, key string) string {
	if m == nil {
		return ""
	}
	if s, ok := m[key].(string); ok {
		return s
	}
	return ""
}

// adapter 返回平台名对应的已注册适配器，没有则返回 nil。
func (b *Bridge) adapter(platform string) adapter.PlatformAdapter {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return b.adapters[platform]
}