Cyrene_v2/plugins/channels/qq_obv11/adapter.go

// Package qq_obv11 实现内置的 QQ OneBot v11 渠道适配器。它按 `transport`(ws|http)
// × `mode`(client|server) 的**四个端**（裁决 1.4）与 OneBot 实现通信：
//
//   - ws + client：Cyrene 主动拨号 OneBot 的 WS 端点，收发共用一个长连接；
//   - ws + server：OneBot 反连 Cyrene 的 `listen_addr`，收发共用一个长连接；
//   - http + client：Cyrene 作为 HTTP 客户端调用 OneBot 的 HTTP API（出站）；
//   - http + server：Cyrene 监听 `listen_addr`，接收 OneBot POST 上来的事件上报（入站）。
//
// 原生 CQ 码 / 数组段消息统一归一成 adapter.UnifiedMessage，回复走同一个端并受限频
// 约束。设计依据是 DevDocs/design/08-渠道接入.md §1/§3/§4；本包只依赖
// internal/channel/adapter、internal/logger 与标准库，WS 帧编解码自带（模块不引入
// 第三方 websocket 依赖）。
package qq_obv11

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

const (
	// adapterLogSource 是本包产出的日志的来源标签（AGENTS §10：日志必须标来源）。
	adapterLogSource = "plugins/channels/qq_obv11"

	// defaultMaxMessageLength bounds a single outgoing QQ message length.
	defaultMaxMessageLength = 4000
	// defaultRecommendBurstMax is the safe burst size for one reply.
	defaultRecommendBurstMax = 4
	// defaultMaxProactivePerWindow caps proactive sends within the window.
	defaultMaxProactivePerWindow = 5
	// defaultProactiveWindow is the sliding window for the send rate limit.
	defaultProactiveWindow = 10 * time.Second

	// transportWS / transportHTTP 是 OneBot v11 的两种通信介质（裁决 1.4）。
	transportWS   = "ws"
	transportHTTP = "http"
	// modeServer / modeClient 是每种介质下的两个角色：谁向谁发数据（裁决 1.4）。
	modeServer = "server"
	modeClient = "client"
	// defaultReportPath 是 http+server 接收上报的默认路径。
	defaultReportPath = "/"
)

// Config configures a qq_obv11 Adapter.
type Config struct {
	// Transport 选择通信介质："ws"（默认，保持既有部署行为）或 "http"（裁决 1.4）。
	// 空值与未知值一律回落 "ws"，绝不静默改走 HTTP。
	Transport string `json:"transport"`
	// Mode 选择该介质下的角色（谁向谁发数据），"server"（默认）或 "client"：
	// ws —— server 等 OneBot 反连本适配器，client 由本适配器拨号 OneBot；
	// http —— server 收 OneBot 的 HTTP 上报，client 调用 OneBot 的 HTTP API。
	Mode string `json:"mode"`
	// ListenAddr is the TCP address to bind in server mode (host:port).
	// **必须由配置显式给出**：OneBot 反连端口是 AGENTS §9 登记的唯一豁免，
	// 空值让 Connect 直接失败，绝不默认开监听。
	ListenAddr string `json:"listen_addr"`
	// RemoteURL 是 ws+client 拨号的 ws://（或 wss://）端点。
	RemoteURL string `json:"remote_url"`
	// APIBaseURL 是 OneBot HTTP API 基址（如 http://127.0.0.1:3000）：http+client
	// 必填；http+server 可选，给了才有出站（回消息）能力。必须显式配置——不从
	// RemoteURL 推推导（v1「剥掉最后一段路径」的推导是有损的，见 design/08 对照）。
	APIBaseURL string `json:"api_base_url"`
	// ReportPath 是 http+server 接收上报的路径。空值默认 "/"。
	ReportPath string `json:"report_path"`
	// AccessToken is the optional OneBot access token for all four ends.
	AccessToken string `json:"access_token"`
	// SelfID is the bot's own QQ number when known.
	SelfID string `json:"self_id"`
	// MaxProactivePerWindow caps sends within ProactiveWindow. Zero uses the
	// default.
	MaxProactivePerWindow int
	// ProactiveWindow is the rate-limit sliding window. Zero uses the default.
	ProactiveWindow time.Duration
	// SendInterval is a minimum delay between consecutive sends, when set.
	SendInterval time.Duration
	// DialContext dials the WS endpoint in client mode. Nil uses net.Dialer.
	DialContext DialContextFunc
	// Logger is the base logger. Nil builds one from internal/logger.
	Logger *slog.Logger
}

// Adapter is the OneBot v11 implementation of adapter.PlatformAdapter.
//
// It is concurrency-safe: one goroutine runs the read loop while multiple
// goroutines may send (writes are serialized on the wsConn and sends are rate
// limited). HTTP 传输没有长连接：入站由上报 handler 直接派发，出站走 httpAPI 的
// 请求-响应路径。
type Adapter struct {
	cfg Config
	log *slog.Logger

	mu        sync.RWMutex
	conn      *wsConn
	ln        net.Listener
	srv       *http.Server
	httpAPI   *obHTTPClient
	started   bool
	connected bool
	// lastConnectErr 是最近一次连接失败的**脱敏**原因，供 ConnectOutcome 如实回答
	// 「为什么没连上」（见 connect_outcome.go）。连上或断开即清空：一条已经过期的
	// 原因不该继续挂在当前状态旁边。由 mu 保护。
	lastConnectErr string
	cancel         context.CancelFunc
	lifeCtx        context.Context
	newConn        chan struct{}

	handlerMu sync.RWMutex
	handler   adapter.MessageHandler
	emitMu    sync.Mutex
	emit      func(adapter.UnifiedMessage)

	limiter *rateLimiter

	groupMu sync.RWMutex
	groups  map[string]string

	lastSend time.Time

	// reconnect / policy 只服务 ws+client 的首连与重连退避（见 reconnect.go）。
	reconnect reconnectState
	policy    reconnectPolicy

	// account 是**账号级**在线状态（get_status 探活所得，见 probe.go）。它与
	// connected（Cyrene 与 OneBot 实现的 WS/会话是否通）是两件事——2026-10-05
	// 事故正是把两者混为一谈：WS 通 ≠ QQ 账号在线。由 mu 保护。
	account accountStatus

	// actionMu / actionWait 是 WS 上「出站 action 请求 → echo 回执」的等待表：探活
	// （get_status，probe.go）与出站回执采集（send_msg，outbound_id.go）共用同一张表、
	// 同一条 routeActionResponse —— echo 前缀不同（status_ / send_），永不撞键。
	// probeEvery 是两次 get_status 探活之间的正常间隔（失败按 policy 退避）。见 probe.go。
	actionMu     sync.Mutex
	actionWait   map[string]chan *obAPIResponse
	probeEvery   time.Duration
	probeTimeout time.Duration
	// sendReceiptTimeout 是 SendMessageWithReceipt 在 WS 上等待 send_msg 回执的上限
	// （见 outbound_id.go；测试可收紧）。HTTP 端不涉及：响应本来就要等。
	sendReceiptTimeout time.Duration
	// probeSeq 是 WS 探活 echo 的单调序号（见 probe.go 的 nextProbeEcho）：唯一性不依赖
	// 时钟精度，两次探活绝不会撞在同一个 echo 上。
	probeSeq atomic.Uint64
	// sendSeq 是 WS 出站 send_msg echo 的单调序号（见 outbound_id.go 的 nextSendEcho），
	// 与 probeSeq 同理：同一毫秒内的多次发送绝不能撞在同一个 echo 上，否则先登记的那次
	// 发送会拿不到自己的回执（旧实现用 UnixNano，Windows 时钟粒度下真会撞）。
	sendSeq atomic.Uint64
}

// Compile-time conformance with the channel contract.
var _ adapter.PlatformAdapter = (*Adapter)(nil)

// Option customizes an Adapter at construction time.
type Option func(*Adapter)

// WithLogger injects the logger the adapter logs through — the wiring path the composition
// root uses (cmd/cyrene/channels.go), mirroring kook/telegram/webhook. A Config decoded from
// channels.json never carries a logger, so without this option an adapter would fall back to
// resolveLogger's default and every DEBUG record (the inbound frame forensics) would vanish.
// It takes precedence over Config.Logger when both are given.
func WithLogger(l *slog.Logger) Option {
	return func(a *Adapter) {
		if l != nil {
			a.log = logger.WithSource(l, adapterLogSource)
		}
	}
}

// NewAdapter builds a qq_obv11 Adapter from cfg, applying safe defaults. 默认值口径
// （裁决 1.4）：transport=ws、mode=server，即**不给出任何新配置时行为与改造前一致**。
func NewAdapter(cfg Config, opts ...Option) *Adapter {
	cfg.Transport = normalizeTransport(cfg.Transport)
	cfg.Mode = normalizeMode(cfg.Mode)
	cfg.ReportPath = normalizeReportPath(cfg.ReportPath)
	if cfg.MaxProactivePerWindow <= 0 {
		cfg.MaxProactivePerWindow = defaultMaxProactivePerWindow
	}
	if cfg.ProactiveWindow <= 0 {
		cfg.ProactiveWindow = defaultProactiveWindow
	}
	a := &Adapter{
		cfg:                cfg,
		groups:             map[string]string{},
		newConn:            make(chan struct{}, 1),
		limiter:            newRateLimiter(cfg.MaxProactivePerWindow, cfg.ProactiveWindow),
		policy:             defaultReconnectPolicy(),
		actionWait:         map[string]chan *obAPIResponse{},
		probeEvery:         defaultProbeInterval,
		probeTimeout:       defaultProbeTimeout,
		sendReceiptTimeout: defaultSendReceiptTimeout,
	}
	for _, opt := range opts {
		opt(a)
	}
	a.log = resolveLogger(a.log, cfg.Logger)
	return a
}

// resolveLogger 选定适配器的日志器（并统一加来源标注）。优先级：WithLogger 选项 →
// Config.Logger → 兜底。
//
// 兜底**不静默**：它用默认 INFO logger 并当场记一条 WARN。理由是入站帧取证
// （inbound_debug.go，级固定 DEBUG）是排查「群友 @ 没回」的地基——静默兜底会让「没人
// 注入 logger」在日志里毫无痕迹，与「上游压根没上报」无法区分（2026-10-05 实测：组合根
// 漏传 logger，全部归档里 `msg="inbound frame"` 一条都没有）。因此宁可留一条噪音，也要
// 让「Debug 会被丢弃」显式可见。兜底只服务裸适配器（测试/库内直构），生产路径由组合根
// 经 WithLogger 正式接线。
func resolveLogger(option, cfg *slog.Logger) *slog.Logger {
	if option != nil {
		return option
	}
	if cfg != nil {
		return logger.WithSource(cfg, adapterLogSource)
	}
	fallback := logger.WithSource(logger.New(logger.Options{}), adapterLogSource)
	fallback.Warn("qq-obv11: no logger injected, Debug forensics (inbound frame) will be dropped",
		"hint", "inject one via qq_obv11.WithLogger from the composition root")
	return fallback
}

// Name returns the platform identifier "qq-obv11".
func (a *Adapter) Name() string { return platformName }

// Config returns a snapshot of the adapter configuration as a plain map.
func (a *Adapter) Config() map[string]any {
	return map[string]any{
		"transport":              a.cfg.Transport,
		"mode":                   a.cfg.Mode,
		"listenAddr":             a.cfg.ListenAddr,
		"remoteURL":              configRemoteURL(a.cfg),
		"apiBaseURL":             a.cfg.APIBaseURL,
		"reportPath":             a.cfg.ReportPath,
		"accessToken":            a.cfg.AccessToken,
		"selfID":                 a.cfg.SelfID,
		"maxProactivePerWindow":  a.cfg.MaxProactivePerWindow,
		"proactiveWindowSeconds": a.cfg.ProactiveWindow.Seconds(),
		"sendIntervalSeconds":    a.cfg.SendInterval.Seconds(),
	}
}

// Connect establishes the session for the configured transport × mode. It is
// idempotent. ws+client 的首连失败**不致命**（对面可能比 Cyrene 后启动）：
// Connect 照常返回 nil，重试交给 readLoop 的同一条重连路径。
func (a *Adapter) Connect(ctx context.Context) error {
	a.mu.Lock()
	if a.started {
		a.mu.Unlock()
		return nil
	}
	a.mu.Unlock()

	var err error
	switch {
	case a.cfg.Transport == transportHTTP:
		err = a.connectHTTP(ctx)
	case a.cfg.Mode == modeClient:
		err = a.connectFirstClient(ctx)
	default:
		err = a.connectServer(ctx)
	}
	if err != nil {
		return fmt.Errorf("qq-obv11: connect: %w", err)
	}

	httpTransport := a.cfg.Transport == transportHTTP
	a.mu.Lock()
	a.started = true
	if httpTransport {
		// HTTP 没有持久连接可观察：监听已绑定 / 出站客户端已装配即视为可服务，
		// 真实可达性交给 HealthCheck 的 get_login_info 探活。
		a.connected = true
	}
	a.lifeCtx, a.cancel = context.WithCancel(context.Background())
	a.mu.Unlock()

	// WS 两端共用一个长连接读循环；HTTP 的入站由上报 handler 直接进 handlePayload。
	if !httpTransport {
		go a.readLoop(a.lifeCtx)
	}
	// 账号在线探活（get_status）与连接生命周期同生共死：lifeCtx 取消即干净退出。
	// 先报一行状态再起循环：日志必须能一眼分辨「探活没跑（含原因）」与「跑了」。
	a.logProbeStart()
	if a.probeEnabled() {
		go a.runProbeLoop(a.lifeCtx)
	}
	return nil
}

// Disconnect tears down the session. It is idempotent.
func (a *Adapter) Disconnect(ctx context.Context) error {
	a.mu.Lock()
	started := a.started
	a.started = false
	cancel := a.cancel
	a.cancel = nil
	a.lifeCtx = nil
	srv := a.srv
	ln := a.ln
	conn := a.conn
	api := a.httpAPI
	a.conn = nil
	a.connected = false
	a.ln = nil
	a.srv = nil
	a.httpAPI = nil
	// 断开即清空上一次失败原因：会话已被拆掉，「为什么没连上」不再是当前事实。
	a.lastConnectErr = ""
	// 断开即丢弃账号在线状态：新会话的 known=false 由下一次探活重新确立，绝不沿用
	// 上一轮的陈旧结论（AccountStatus 的诚实口径）。
	a.account = accountStatus{}
	a.mu.Unlock()

	if !started && cancel == nil {
		return nil
	}
	if cancel != nil {
		cancel()
	}
	if conn != nil {
		_ = conn.Close()
	}
	// 出站 HTTP 客户端没有长连接，只需释放连接池里的 keep-alive 连接。
	if api != nil {
		api.Close()
	}
	if srv != nil {
		shutdownCtx, c := context.WithTimeout(context.Background(), 3*time.Second)
		defer c()
		if serr := srv.Shutdown(shutdownCtx); serr != nil && !errors.Is(serr, context.DeadlineExceeded) {
			return fmt.Errorf("qq-obv11: server shutdown: %w", serr)
		}
	}
	if ln != nil {
		_ = ln.Close()
	}
	return nil
}

// IsConnected reports whether a live OneBot session exists. HTTP 传输没有长连接，
// 它反映的是「适配器已就绪」（监听已绑定 / 出站客户端已装配）。
//
// ⚠️ 它**只**回答「Cyrene 与 OneBot 实现（适配器服务器）通不通」——WS 两端是长连接
// 是否建立。它**不**回答「QQ 账号是否在线」：NapCat 等实现即使掉了账号，正向 WS 仍
// 可能保持连接（2026-10-05 事故现场）。账号在线状态见 AccountStatus（get_status 探活）。
func (a *Adapter) IsConnected() bool {
	a.mu.RLock()
	defer a.mu.RUnlock()
	return a.connected
}

// HealthCheck verifies the session is usable. HTTP 传输额外做一次 get_login_info
// 探活：没有长连接可观察，只有真实请求才能说明 OneBot 实现可达。
func (a *Adapter) HealthCheck(ctx context.Context) error {
	if !a.IsConnected() {
		return fmt.Errorf("qq-obv11: health check: %w", adapter.ErrNotConnected)
	}
	if api := a.currentAPI(); api != nil {
		return api.probe(ctx)
	}
	return nil
}

// currentAPI 返回已装配的 OneBot HTTP API 出站客户端；未装配时（ws 两端，或
// http+server 只收上报、没配 api_base_url）为 nil。
func (a *Adapter) currentAPI() *obHTTPClient {
	a.mu.RLock()
	defer a.mu.RUnlock()
	return a.httpAPI
}

// connectHTTP 装配 HTTP 传输（裁决 1.4）。OneBot v11 的 HTTP 通信天然分两半：动作
// （Cyrene→OneBot，client 端，走 OneBot 的 HTTP API）与上报（OneBot→Cyrene，server
// 端，走本适配器的监听端口）。`mode` 决定哪一半是**必需**的——server 必须有显式
// listen_addr（AGENTS §9 豁免口径：绝不默认开端口），client 必须有 api_base_url；
// 另一半只要对应字段也给齐就一并启用，于是单个适配器即可完成「收上报 + 回消息」闭环。
func (a *Adapter) connectHTTP(ctx context.Context) error {
	// 先做纯配置校验并构造出站客户端（无副作用），再绑监听端口：任何一端配置有误都
	// 不能留下半开的端口——桥接器对 Connect 失败不会回头调用 Disconnect。
	var api *obHTTPClient
	if a.cfg.APIBaseURL != "" {
		client, err := newOBHTTPClient(a.cfg.APIBaseURL, a.cfg.AccessToken)
		if err != nil {
			return err
		}
		api = client
	} else if a.cfg.Mode == modeClient {
		return errors.New("qq-obv11: http client mode requires apiBaseUrl")
	}

	if a.cfg.ListenAddr != "" {
		if err := a.connectHTTPServer(ctx); err != nil {
			return err
		}
	} else if a.cfg.Mode == modeServer {
		return errors.New("qq-obv11: http server mode requires listenAddr")
	}

	if api != nil {
		a.mu.Lock()
		a.httpAPI = api
		a.mu.Unlock()
	}
	return nil
}

// configRemoteURL 是 Config() 上报的「远端地址」：channelbridge 的凭据门按
// remoteURL（client 模式）/ listenAddr（server 模式）判断该渠道是否值得连接
// （internal/services/channelbridge/connect.go:143-147）。HTTP client 端的远端就是
// OneBot 的 HTTP API 基址，因此 RemoteURL 为空时回退到 APIBaseURL——否则
// transport=http + mode=client 会被误判成「无凭据」而永不连接。
func configRemoteURL(cfg Config) string {
	if cfg.RemoteURL != "" {
		return cfg.RemoteURL
	}
	return cfg.APIBaseURL
}

// normalizeTransport 归一化通信介质：空值保持现状 ws，未知值同样回落 ws，
// 不静默按 http 处理。
func normalizeTransport(raw string) string {
	if strings.EqualFold(strings.TrimSpace(raw), transportHTTP) {
		return transportHTTP
	}
	return transportWS
}

// normalizeMode 归一化角色：空值与未知值回落 server（与既有默认一致）。
func normalizeMode(raw string) string {
	if strings.EqualFold(strings.TrimSpace(raw), modeClient) {
		return modeClient
	}
	return modeServer
}

// normalizeReportPath 归一化上报路径：空值默认 "/"，并保证以 "/" 开头。
func normalizeReportPath(raw string) string {
	p := strings.TrimSpace(raw)
	switch {
	case p == "":
		return defaultReportPath
	case !strings.HasPrefix(p, "/"):
		return "/" + p
	default:
		return p
	}
}