Cyrene_v2/backend/services/platform-bridge/qq/adapter.go

// Package qq provides an adapter for QQ (OBv11 protocol) integration.
package qq

import (
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"github.com/gorilla/websocket"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/services/platform-bridge"
)

// Config holds QQ adapter configuration.
type Config struct {
	Name        string // Adapter name (e.g. "qq-main")
	WSURL       string // OBv11 WebSocket URL ("ws://localhost:3001")
	AccessToken string // OBv11 access token
	BotQQ       string // Bot's own QQ number
	MinGapMs    int    // Min interval between sends in ms (rate limit, default 0 = no limit)
}

// Adapter implements bridge.Adapter for QQ OBv11.
// It supervises the WebSocket connection and reconnects with backoff on failure.
type Adapter struct {
	cfg      Config
	conn     *websocket.Conn
	incoming chan *bridge.UnifiedMessage
	mu       sync.Mutex
	done     chan struct{}
	stopOnce sync.Once
	lastSend time.Time
}

// New creates a QQ adapter.
func New(cfg Config) *Adapter {
	return &Adapter{
		cfg:      cfg,
		incoming: make(chan *bridge.UnifiedMessage, 64),
		done:     make(chan struct{}),
	}
}

func (a *Adapter) Name() string     { return a.cfg.Name }
func (a *Adapter) Platform() string { return "qq" }

// Start launches the connection supervisor and returns immediately.
// The supervisor keeps the WebSocket connected, reconnecting with
// exponential backoff (1s → 30s) when the connection drops.
func (a *Adapter) Start() error {
	if a.cfg.WSURL == "" {
		return fmt.Errorf("qq ws_url is empty")
	}
	go a.supervise()
	return nil
}

func (a *Adapter) supervise() {
	backoff := time.Second
	for {
		select {
		case <-a.done:
			return
		default:
		}

		err := a.runOnce()
		select {
		case <-a.done:
			return
		default:
		}

		logger.Infof("[qq:%s] connection lost: %v, reconnecting in %v", a.cfg.Name, err, backoff)
		timer := time.NewTimer(backoff)
		select {
		case <-a.done:
			timer.Stop()
			return
		case <-timer.C:
		}
		if backoff < 30*time.Second {
			backoff *= 2
		}
	}
}

// runOnce dials and reads until the connection fails.
func (a *Adapter) runOnce() error {
	conn, _, err := websocket.DefaultDialer.Dial(a.cfg.WSURL, nil)
	if err != nil {
		return fmt.Errorf("qq ws connect: %w", err)
	}
	a.mu.Lock()
	a.conn = conn
	a.mu.Unlock()
	defer func() {
		a.mu.Lock()
		a.conn = nil
		a.mu.Unlock()
		conn.Close()
	}()

	logger.Infof("[qq:%s] connected to %s", a.cfg.Name, a.cfg.WSURL)

	// Send auth
	authMsg := map[string]any{
		"type":         "meta::connect",
		"access_token": a.cfg.AccessToken,
	}
	conn.WriteJSON(authMsg)

	for {
		_, raw, err := conn.ReadMessage()
		if err != nil {
			return err
		}
		var event map[string]any
		if err := json.Unmarshal(raw, &event); err != nil {
			continue
		}
		msg := a.parseMessage(event)
		if msg != nil {
			select {
			case a.incoming <- msg:
			case <-a.done:
				return fmt.Errorf("stopped")
			}
		}
	}
}

func (a *Adapter) Stop() error {
	a.stopOnce.Do(func() {
		close(a.done)
		a.mu.Lock()
		if a.conn != nil {
			a.conn.Close()
		}
		a.mu.Unlock()
	})
	return nil
}

func (a *Adapter) Incoming() <-chan *bridge.UnifiedMessage { return a.incoming }

func (a *Adapter) IsConnected() bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.conn != nil
}

func (a *Adapter) Send(target string, content string, opts bridge.SendOptions) error {
	// Rate limit: enforce the min gap between sends
	gap := opts.MinGap
	if gap == 0 && a.cfg.MinGapMs > 0 {
		gap = time.Duration(a.cfg.MinGapMs) * time.Millisecond
	}

	a.mu.Lock()
	if gap > 0 && !a.lastSend.IsZero() {
		if wait := gap - time.Since(a.lastSend); wait > 0 {
			a.mu.Unlock()
			time.Sleep(wait)
			a.mu.Lock()
		}
	}
	a.lastSend = time.Now()
	conn := a.conn
	a.mu.Unlock()

	if conn == nil {
		return fmt.Errorf("not connected")
	}

	// Apply length limit
	if opts.MaxLength > 0 && len([]rune(content)) > opts.MaxLength {
		runes := []rune(content)
		content = string(runes[:opts.MaxLength]) + "..."
	}

	msg := map[string]any{
		"action": "send_msg",
		"params": map[string]any{
			"user_id":  target,
			"message":  content,
			"auto_escape": false,
		},
	}
	return conn.WriteJSON(msg)
}

func (a *Adapter) parseMessage(event map[string]any) *bridge.UnifiedMessage {
	postType, _ := event["post_type"].(string)
	if postType != "message" {
		return nil
	}

	msgType, _ := event["message_type"].(string)
	sender, _ := event["sender"].(map[string]any)
	if sender == nil {
		return nil
	}
	rawMsg, _ := event["raw_message"].(string)
	senderID := fmt.Sprintf("%v", sender["user_id"])
	senderName, _ := sender["nickname"].(string)

	isDM := msgType == "private"
	groupID := ""
	if !isDM {
		groupID = fmt.Sprintf("%v", event["group_id"])
	}

	// Check if bot was @mentioned
	atBot := false
	if atList, ok := event["at_list"].([]any); ok {
		for _, at := range atList {
			if fmt.Sprintf("%v", at) == a.cfg.BotQQ {
				atBot = true
				break
			}
		}
	}

	ts := time.Now()
	if t, ok := event["time"].(float64); ok {
		ts = time.Unix(int64(t), 0)
	}

	return &bridge.UnifiedMessage{
		ID:         fmt.Sprintf("%v", event["message_id"]),
		Platform:   "qq",
		Source:     groupID,
		SenderID:   senderID,
		SenderName: senderName,
		Content:    rawMsg,
		IsAtBot:    atBot,
		IsDM:       isDM,
		RawData:    event,
		Timestamp:  ts,
	}
}