feat(bridge): QQ adapter reconnect and send rate limiting

- supervisor loop keeps the OBv11 WebSocket alive, reconnecting with
  exponential backoff (1s→30s) when the connection drops; Start no
  longer fails permanently when NapCat is temporarily unreachable
- Send enforces MinGap between messages (from SendOptions or the
  adapter's send_interval_ms setting, now wired from the admin panel)
- parseMessage no longer panics on events missing sender/time fields
- tests: reconnect after drop, rate limit enforcement, malformed events,
  send while disconnected
This commit is contained in:
Aska committed 2026-08-21 12:48:04 +08:00
1 parent 8603839711
commit 9f30cdd086
3 files changed
+237 -33

No files matched your search

+114 -33
View File
@@ -9,6 +9,7 @@ import (
"github.com/gorilla/websocket"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
"git.yeij.top/AskaEth/Cyrene_v2/services/platform-bridge"
)
@@ -18,15 +19,19 @@ type Config struct {
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
cfg Config
conn *websocket.Conn
incoming chan *bridge.UnifiedMessage
mu sync.Mutex
done chan struct{}
mu sync.Mutex
done chan struct{}
stopOnce sync.Once
lastSend time.Time
}
// New creates a QQ adapter.
@@ -41,12 +46,64 @@ func New(cfg Config) *Adapter {
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{
@@ -55,35 +112,35 @@ func (a *Adapter) Start() error {
}
conn.WriteJSON(authMsg)
// Read loop
go func() {
defer close(a.incoming)
for {
_, raw, err := conn.ReadMessage()
if err != nil {
return
}
var event map[string]any
if err := json.Unmarshal(raw, &event); err != nil {
continue
}
msg := a.parseMessage(event)
if msg != nil {
a.incoming <- msg
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")
}
}
}()
return nil
}
}
func (a *Adapter) Stop() error {
a.mu.Lock()
defer a.mu.Unlock()
if a.conn != nil {
a.conn.Close()
}
close(a.done)
a.stopOnce.Do(func() {
close(a.done)
a.mu.Lock()
if a.conn != nil {
a.conn.Close()
}
a.mu.Unlock()
})
return nil
}
@@ -96,9 +153,25 @@ func (a *Adapter) IsConnected() bool {
}
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()
defer a.mu.Unlock()
if a.conn == nil {
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")
}
@@ -116,7 +189,7 @@ func (a *Adapter) Send(target string, content string, opts bridge.SendOptions) e
"auto_escape": false,
},
}
return a.conn.WriteJSON(msg)
return conn.WriteJSON(msg)
}
func (a *Adapter) parseMessage(event map[string]any) *bridge.UnifiedMessage {
@@ -126,7 +199,10 @@ func (a *Adapter) parseMessage(event map[string]any) *bridge.UnifiedMessage {
}
msgType, _ := event["message_type"].(string)
sender := event["sender"].(map[string]any)
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)
@@ -148,6 +224,11 @@ func (a *Adapter) parseMessage(event map[string]any) *bridge.UnifiedMessage {
}
}
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",
@@ -158,6 +239,6 @@ func (a *Adapter) parseMessage(event map[string]any) *bridge.UnifiedMessage {
IsAtBot: atBot,
IsDM: isDM,
RawData: event,
Timestamp: time.Unix(int64(event["time"].(float64)), 0),
Timestamp: ts,
}
}