f3220a2381
- Upgrade Go version from 1.24 to 1.26.1 with updated dependencies
- Replace interface{} with any type throughout codebase
- Add message deduplication in OneBot11 adapter to prevent duplicate event processing
- URL-encode access token in WebSocket connections to handle special characters
- Remove duplicate bot startup hooks in DI providers (now handled by botManager.StartAll)
- Use slices.Clone and errgroup.Go patterns for cleaner concurrent code
- Apply omitzero JSON tags for optional fields (Go 1.24+ feature)
563 lines
15 KiB
Go
563 lines
15 KiB
Go
package onebot11
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"net/url"
|
||
"sync"
|
||
"time"
|
||
|
||
"cellbot/internal/engine"
|
||
"cellbot/internal/protocol"
|
||
"cellbot/pkg/net"
|
||
|
||
"github.com/bytedance/sonic"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
// Adapter OneBot11协议适配器
|
||
type Adapter struct {
|
||
config *Config
|
||
logger *zap.Logger
|
||
wsManager *net.WebSocketManager
|
||
httpClient *HTTPClient
|
||
wsWaiter *WSResponseWaiter
|
||
eventBus *engine.EventBus
|
||
selfID string
|
||
connected bool
|
||
mu sync.RWMutex
|
||
wsConnection *net.WebSocketConnection
|
||
ctx context.Context
|
||
cancel context.CancelFunc
|
||
|
||
// 消息去重
|
||
processedMessages map[int64]time.Time
|
||
messageMu sync.RWMutex
|
||
}
|
||
|
||
// Config OneBot11配置
|
||
type Config struct {
|
||
// 连接配置
|
||
ConnectionType string `json:"connection_type"` // ws, ws-reverse, http, http-post
|
||
Host string `json:"host"`
|
||
Port int `json:"port"`
|
||
AccessToken string `json:"access_token"`
|
||
|
||
// WebSocket配置
|
||
WSUrl string `json:"ws_url"` // 正向WS地址
|
||
WSReverseUrl string `json:"ws_reverse_url"` // 反向WS监听地址
|
||
Heartbeat int `json:"heartbeat"` // 心跳间隔(秒)
|
||
ReconnectInterval int `json:"reconnect_interval"` // 重连间隔(秒)
|
||
|
||
// HTTP配置
|
||
HTTPUrl string `json:"http_url"` // 正向HTTP地址
|
||
HTTPPostUrl string `json:"http_post_url"` // HTTP POST上报地址
|
||
Secret string `json:"secret"` // 签名密钥
|
||
Timeout int `json:"timeout"` // 超时时间(秒)
|
||
|
||
// 其他配置
|
||
SelfID string `json:"self_id"` // 机器人QQ号
|
||
Nickname string `json:"nickname"` // 机器人昵称
|
||
}
|
||
|
||
// NewAdapter 创建OneBot11适配器
|
||
func NewAdapter(config *Config, logger *zap.Logger, wsManager *net.WebSocketManager, eventBus *engine.EventBus) *Adapter {
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
|
||
timeout := time.Duration(config.Timeout) * time.Second
|
||
if timeout == 0 {
|
||
timeout = 30 * time.Second
|
||
}
|
||
|
||
adapter := &Adapter{
|
||
config: config,
|
||
logger: logger.Named("onebot11"),
|
||
wsManager: wsManager,
|
||
wsWaiter: NewWSResponseWaiter(timeout, logger),
|
||
eventBus: eventBus,
|
||
selfID: config.SelfID,
|
||
ctx: ctx,
|
||
cancel: cancel,
|
||
processedMessages: make(map[int64]time.Time),
|
||
}
|
||
|
||
// 启动定期清理过期消息的 goroutine
|
||
go adapter.cleanupProcessedMessages()
|
||
|
||
// 如果使用HTTP连接，初始化HTTP客户端
|
||
if config.ConnectionType == "http" && config.HTTPUrl != "" {
|
||
adapter.httpClient = NewHTTPClient(config.HTTPUrl, config.AccessToken, timeout, logger)
|
||
}
|
||
|
||
return adapter
|
||
}
|
||
|
||
// cleanupProcessedMessages 定期清理过期的已处理消息记录
|
||
func (a *Adapter) cleanupProcessedMessages() {
|
||
ticker := time.NewTicker(5 * time.Minute)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-a.ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
a.messageMu.Lock()
|
||
now := time.Now()
|
||
for msgID, processedAt := range a.processedMessages {
|
||
if now.Sub(processedAt) > 10*time.Minute {
|
||
delete(a.processedMessages, msgID)
|
||
}
|
||
}
|
||
a.messageMu.Unlock()
|
||
}
|
||
}
|
||
}
|
||
|
||
// isMessageProcessed 检查消息是否已处理
|
||
func (a *Adapter) isMessageProcessed(messageID int64) bool {
|
||
a.messageMu.RLock()
|
||
_, exists := a.processedMessages[messageID]
|
||
a.messageMu.RUnlock()
|
||
return exists
|
||
}
|
||
|
||
// markMessageProcessed 标记消息为已处理
|
||
func (a *Adapter) markMessageProcessed(messageID int64) {
|
||
a.messageMu.Lock()
|
||
a.processedMessages[messageID] = time.Now()
|
||
a.messageMu.Unlock()
|
||
}
|
||
|
||
// Name 获取协议名称
|
||
func (a *Adapter) Name() string {
|
||
return "OneBot"
|
||
}
|
||
|
||
// Version 获取协议版本
|
||
func (a *Adapter) Version() string {
|
||
return "11"
|
||
}
|
||
|
||
// Connect 建立连接
|
||
func (a *Adapter) Connect(ctx context.Context) error {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
|
||
if a.connected {
|
||
return fmt.Errorf("already connected")
|
||
}
|
||
|
||
a.logger.Info("Starting OneBot11 connection",
|
||
zap.String("connection_type", a.config.ConnectionType),
|
||
zap.String("self_id", a.selfID))
|
||
|
||
switch a.config.ConnectionType {
|
||
case "ws":
|
||
return a.connectWebSocket(ctx)
|
||
case "ws-reverse":
|
||
return a.connectWebSocketReverse(ctx)
|
||
case "http":
|
||
return a.connectHTTP(ctx)
|
||
case "http-post":
|
||
return a.connectHTTPPost(ctx)
|
||
default:
|
||
return fmt.Errorf("unsupported connection type: %s", a.config.ConnectionType)
|
||
}
|
||
}
|
||
|
||
// Disconnect 断开连接
|
||
func (a *Adapter) Disconnect(ctx context.Context) error {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
|
||
if !a.connected {
|
||
a.logger.Debug("Already disconnected, skipping")
|
||
return nil
|
||
}
|
||
|
||
a.logger.Info("Disconnecting OneBot11 adapter",
|
||
zap.String("connection_type", a.config.ConnectionType))
|
||
|
||
// 取消上下文
|
||
if a.cancel != nil {
|
||
a.cancel()
|
||
a.logger.Debug("Context cancelled")
|
||
}
|
||
|
||
// 关闭WebSocket连接
|
||
if a.wsConnection != nil {
|
||
a.logger.Info("Closing WebSocket connection",
|
||
zap.String("connection_id", a.wsConnection.ID))
|
||
a.wsManager.RemoveConnection(a.wsConnection.ID)
|
||
a.wsConnection = nil
|
||
}
|
||
|
||
// 关闭HTTP客户端
|
||
if a.httpClient != nil {
|
||
if err := a.httpClient.Close(); err != nil {
|
||
a.logger.Error("Failed to close HTTP client", zap.Error(err))
|
||
} else {
|
||
a.logger.Debug("HTTP client closed")
|
||
}
|
||
}
|
||
|
||
a.connected = false
|
||
a.logger.Info("OneBot11 adapter disconnected successfully")
|
||
return nil
|
||
}
|
||
|
||
// IsConnected 检查连接状态
|
||
func (a *Adapter) IsConnected() bool {
|
||
a.mu.RLock()
|
||
defer a.mu.RUnlock()
|
||
return a.connected
|
||
}
|
||
|
||
// GetSelfID 获取机器人自身ID
|
||
func (a *Adapter) GetSelfID() string {
|
||
return a.selfID
|
||
}
|
||
|
||
// SendAction 发送动作
|
||
func (a *Adapter) SendAction(ctx context.Context, action protocol.Action) (map[string]any, error) {
|
||
// 序列化为OneBot11格式
|
||
data, err := a.SerializeAction(action)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
switch a.config.ConnectionType {
|
||
case "ws", "ws-reverse":
|
||
return a.sendActionWebSocket(data)
|
||
case "http":
|
||
return a.sendActionHTTP(data)
|
||
default:
|
||
return nil, fmt.Errorf("unsupported connection type for sending action: %s", a.config.ConnectionType)
|
||
}
|
||
}
|
||
|
||
// HandleEvent 处理事件
|
||
func (a *Adapter) HandleEvent(ctx context.Context, event protocol.Event) error {
|
||
a.logger.Debug("Handling event",
|
||
zap.String("type", string(event.GetType())),
|
||
zap.String("detail_type", event.GetDetailType()))
|
||
return nil
|
||
}
|
||
|
||
// ParseMessage 解析原始消息为Event
|
||
func (a *Adapter) ParseMessage(raw []byte) (protocol.Event, error) {
|
||
var rawEvent RawEvent
|
||
if err := sonic.Unmarshal(raw, &rawEvent); err != nil {
|
||
return nil, fmt.Errorf("failed to unmarshal raw event: %w", err)
|
||
}
|
||
|
||
// 忽略机器人自己发送的消息
|
||
if rawEvent.PostType == "message_sent" {
|
||
return nil, fmt.Errorf("ignoring message_sent event")
|
||
}
|
||
|
||
return a.convertToEvent(&rawEvent)
|
||
}
|
||
|
||
// SerializeAction 序列化Action为协议格式
|
||
func (a *Adapter) SerializeAction(action protocol.Action) ([]byte, error) {
|
||
// 转换为OneBot11格式
|
||
ob11ActionName := ConvertAction(action)
|
||
|
||
// 检查是否有未转换的动作类型（如果转换后的名称与原始类型相同，说明没有匹配到）
|
||
originalType := string(action.GetType())
|
||
if ob11ActionName == originalType {
|
||
a.logger.Warn("Action type not converted, using original type",
|
||
zap.String("action_type", originalType),
|
||
zap.String("hint", "This action type may not be supported by OneBot11"))
|
||
}
|
||
|
||
// 复制参数并转换消息链
|
||
params := make(map[string]any)
|
||
for k, v := range action.GetParams() {
|
||
// 检查是否是消息链
|
||
if k == "message" {
|
||
if chain, ok := v.(protocol.MessageChain); ok {
|
||
// 转换为 OneBot11 格式
|
||
params[k] = ConvertMessageChainToOB11(chain)
|
||
continue
|
||
}
|
||
}
|
||
params[k] = v
|
||
}
|
||
|
||
ob11Action := &OB11Action{
|
||
Action: ob11ActionName,
|
||
Params: params,
|
||
}
|
||
|
||
return sonic.Marshal(ob11Action)
|
||
}
|
||
|
||
// connectWebSocket 正向WebSocket连接
|
||
func (a *Adapter) connectWebSocket(ctx context.Context) error {
|
||
if a.config.WSUrl == "" {
|
||
return fmt.Errorf("ws_url is required for ws connection")
|
||
}
|
||
|
||
a.logger.Info("Connecting to OneBot WebSocket server",
|
||
zap.String("url", a.config.WSUrl),
|
||
zap.Bool("has_token", a.config.AccessToken != ""))
|
||
|
||
// 添加访问令牌到URL（需要对token进行URL编码以处理特殊字符）
|
||
wsURL := a.config.WSUrl
|
||
if a.config.AccessToken != "" {
|
||
wsURL += "?access_token=" + url.QueryEscape(a.config.AccessToken)
|
||
a.logger.Debug("Added access token to WebSocket URL")
|
||
}
|
||
|
||
a.logger.Info("Dialing WebSocket...",
|
||
zap.String("full_url", wsURL))
|
||
|
||
wsConn, err := a.wsManager.Dial(wsURL, a.selfID)
|
||
if err != nil {
|
||
a.logger.Error("Failed to connect WebSocket",
|
||
zap.String("url", a.config.WSUrl),
|
||
zap.Error(err))
|
||
return fmt.Errorf("failed to connect websocket: %w", err)
|
||
}
|
||
|
||
a.wsConnection = wsConn
|
||
a.connected = true
|
||
|
||
a.logger.Info("WebSocket connected successfully",
|
||
zap.String("url", a.config.WSUrl),
|
||
zap.String("remote_addr", wsConn.RemoteAddr),
|
||
zap.String("connection_id", wsConn.ID))
|
||
|
||
// 启动消息接收处理
|
||
go a.handleWebSocketMessages()
|
||
|
||
a.logger.Info("WebSocket message handler started")
|
||
|
||
return nil
|
||
}
|
||
|
||
// connectWebSocketReverse 反向WebSocket连接
|
||
func (a *Adapter) connectWebSocketReverse(ctx context.Context) error {
|
||
// 反向WebSocket由客户端主动连接到服务器
|
||
// WebSocket服务器会在主Server中启动
|
||
// 这里只需要标记为已连接状态，等待客户端通过HTTP服务器连接
|
||
a.connected = true
|
||
|
||
a.logger.Info("OneBot11 adapter ready for reverse WebSocket connections",
|
||
zap.String("bot_id", a.selfID),
|
||
zap.String("listen_addr", a.config.WSReverseUrl))
|
||
|
||
// 注意：实际的WebSocket服务器由pkg/net/server.go提供
|
||
// OneBot客户端需要连接到 ws://host:port/ws?bot_id=<selfID>
|
||
|
||
return nil
|
||
}
|
||
|
||
// connectHTTP 正向HTTP连接
|
||
func (a *Adapter) connectHTTP(ctx context.Context) error {
|
||
if a.config.HTTPUrl == "" {
|
||
return fmt.Errorf("http_url is required for http connection")
|
||
}
|
||
|
||
// 创建HTTP客户端
|
||
// TODO: 实现HTTP轮询
|
||
a.connected = true
|
||
|
||
a.logger.Info("HTTP connected",
|
||
zap.String("url", a.config.HTTPUrl))
|
||
|
||
return nil
|
||
}
|
||
|
||
// connectHTTPPost HTTP POST上报
|
||
func (a *Adapter) connectHTTPPost(ctx context.Context) error {
|
||
if a.config.HTTPPostUrl == "" {
|
||
return fmt.Errorf("http_post_url is required for http-post connection")
|
||
}
|
||
|
||
// HTTP POST由客户端主动推送事件
|
||
a.connected = true
|
||
|
||
a.logger.Info("HTTP POST ready",
|
||
zap.String("url", a.config.HTTPPostUrl))
|
||
|
||
return nil
|
||
}
|
||
|
||
// sendActionWebSocket 通过WebSocket发送动作
|
||
func (a *Adapter) sendActionWebSocket(data []byte) (map[string]any, error) {
|
||
if a.wsConnection == nil {
|
||
return nil, fmt.Errorf("websocket connection not established")
|
||
}
|
||
|
||
// 解析请求以获取或添加echo
|
||
var req OB11Action
|
||
if err := sonic.Unmarshal(data, &req); err != nil {
|
||
return nil, fmt.Errorf("failed to unmarshal action: %w", err)
|
||
}
|
||
|
||
// 如果没有echo，生成一个
|
||
if req.Echo == "" {
|
||
req.Echo = GenerateEcho()
|
||
var err error
|
||
data, err = sonic.Marshal(req)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("failed to marshal action with echo: %w", err)
|
||
}
|
||
}
|
||
|
||
// 发送消息
|
||
if err := a.wsConnection.SendMessage(data); err != nil {
|
||
return nil, fmt.Errorf("failed to send action: %w", err)
|
||
}
|
||
|
||
// 等待响应
|
||
resp, err := a.wsWaiter.Wait(req.Echo)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 检查响应状态
|
||
if resp.Status != "ok" && resp.Status != "async" {
|
||
return resp.Data, fmt.Errorf("action failed (retcode=%d)", resp.RetCode)
|
||
}
|
||
|
||
return resp.Data, nil
|
||
}
|
||
|
||
// sendActionHTTP 通过HTTP发送动作
|
||
func (a *Adapter) sendActionHTTP(data []byte) (map[string]any, error) {
|
||
if a.httpClient == nil {
|
||
return nil, fmt.Errorf("http client not initialized")
|
||
}
|
||
|
||
// 解析请求
|
||
var req OB11Action
|
||
if err := sonic.Unmarshal(data, &req); err != nil {
|
||
return nil, fmt.Errorf("failed to unmarshal action: %w", err)
|
||
}
|
||
|
||
// 调用HTTP API
|
||
resp, err := a.httpClient.Call(a.ctx, req.Action, req.Params)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 检查响应状态
|
||
if resp.Status != "ok" && resp.Status != "async" {
|
||
return resp.Data, fmt.Errorf("action failed (retcode=%d)", resp.RetCode)
|
||
}
|
||
|
||
return resp.Data, nil
|
||
}
|
||
|
||
// handleWebSocketMessages 处理WebSocket消息
|
||
func (a *Adapter) handleWebSocketMessages() {
|
||
a.logger.Info("WebSocket message handler started, waiting for messages...")
|
||
|
||
for {
|
||
select {
|
||
case <-a.ctx.Done():
|
||
a.logger.Info("Context cancelled, stopping WebSocket message handler")
|
||
return
|
||
default:
|
||
}
|
||
|
||
if a.wsConnection == nil || a.wsConnection.Conn == nil {
|
||
a.logger.Warn("WebSocket connection is nil, stopping message handler")
|
||
return
|
||
}
|
||
|
||
// 读取消息
|
||
_, message, err := a.wsConnection.Conn.ReadMessage()
|
||
if err != nil {
|
||
a.logger.Error("Failed to read WebSocket message",
|
||
zap.Error(err),
|
||
zap.String("connection_id", a.wsConnection.ID))
|
||
return
|
||
}
|
||
|
||
a.logger.Debug("Received WebSocket message",
|
||
zap.Int("size", len(message)),
|
||
zap.String("preview", string(message[:min(len(message), 200)])))
|
||
|
||
// 尝试解析为响应（先检查是否有 echo 字段）
|
||
var tempMap map[string]any
|
||
if err := sonic.Unmarshal(message, &tempMap); err == nil {
|
||
if echo, ok := tempMap["echo"].(string); ok && echo != "" {
|
||
// 有 echo 字段，说明是API响应
|
||
var resp OB11Response
|
||
if err := sonic.Unmarshal(message, &resp); err == nil {
|
||
a.logger.Debug("Received API response",
|
||
zap.String("echo", resp.Echo),
|
||
zap.String("status", resp.Status),
|
||
zap.Int("retcode", resp.RetCode))
|
||
a.wsWaiter.Notify(&resp)
|
||
continue
|
||
} else {
|
||
a.logger.Warn("Failed to parse API response",
|
||
zap.Error(err),
|
||
zap.String("echo", echo))
|
||
}
|
||
}
|
||
}
|
||
|
||
// 否则当作事件处理
|
||
a.logger.Info("Received OneBot event",
|
||
zap.ByteString("raw_event", message))
|
||
|
||
// 解析事件
|
||
a.logger.Info("Parsing OneBot event...")
|
||
event, err := a.ParseMessage(message)
|
||
if err != nil {
|
||
// 如果是忽略的事件（如 message_sent），只记录 debug 日志
|
||
if err.Error() == "ignoring message_sent event" {
|
||
a.logger.Debug("Ignoring message_sent event")
|
||
} else {
|
||
a.logger.Error("Failed to parse event",
|
||
zap.Error(err),
|
||
zap.ByteString("raw_message", message))
|
||
}
|
||
continue
|
||
}
|
||
|
||
// 消息去重检查（基于 message_id）
|
||
data := event.GetData()
|
||
// message_id 可能是 int32 或 int64
|
||
var messageID int64
|
||
if mid, ok := data["message_id"].(int32); ok && mid > 0 {
|
||
messageID = int64(mid)
|
||
} else if mid, ok := data["message_id"].(int64); ok && mid > 0 {
|
||
messageID = mid
|
||
}
|
||
if messageID > 0 {
|
||
if a.isMessageProcessed(messageID) {
|
||
a.logger.Warn("Duplicate message detected, skipping",
|
||
zap.Int64("message_id", messageID))
|
||
continue
|
||
}
|
||
a.markMessageProcessed(messageID)
|
||
}
|
||
|
||
// 发布事件到事件总线
|
||
a.logger.Info("Publishing event to event bus",
|
||
zap.String("event_type", string(event.GetType())),
|
||
zap.String("detail_type", event.GetDetailType()),
|
||
zap.String("self_id", event.GetSelfID()))
|
||
|
||
a.eventBus.Publish(event)
|
||
|
||
a.logger.Info("Event published successfully")
|
||
}
|
||
}
|
||
|
||
func min(a, b int) int {
|
||
if a < b {
|
||
return a
|
||
}
|
||
return b
|
||
}
|