From 9f30cdd086230188ea01ca554f225d0f090e620a Mon Sep 17 00:00:00 2001 From: Aska Date: Fri, 21 Aug 2026 12:48:04 +0800 Subject: [PATCH] feat(bridge): QQ adapter reconnect and send rate limiting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- backend/cmd/cyrene/main.go | 3 + .../services/platform-bridge/qq/adapter.go | 147 ++++++++++++++---- .../platform-bridge/qq/adapter_test.go | 120 ++++++++++++++ 3 files changed, 237 insertions(+), 33 deletions(-) create mode 100644 backend/services/platform-bridge/qq/adapter_test.go diff --git a/backend/cmd/cyrene/main.go b/backend/cmd/cyrene/main.go index dea3312..bdf40f2 100644 --- a/backend/cmd/cyrene/main.go +++ b/backend/cmd/cyrene/main.go @@ -462,9 +462,12 @@ func loadAdaptersFromConfig(router *bridge.Router, cfg *bridge.BridgeConfig, orc if !ac.Enabled { continue } switch ac.Type { case "qq-obv11": + minGapMs := 0 + fmt.Sscanf(ac.Settings["send_interval_ms"], "%d", &minGapMs) router.Register(bridgeqq.New(bridgeqq.Config{ Name: ac.Name, WSURL: ac.Settings["ws_url"], AccessToken: ac.Settings["access_token"], BotQQ: ac.Settings["bot_qq"], + MinGapMs: minGapMs, })) case "telegram": router.Register(bridgetg.New(bridgetg.Config{Name: ac.Name, Token: ac.Settings["token"]})) diff --git a/backend/services/platform-bridge/qq/adapter.go b/backend/services/platform-bridge/qq/adapter.go index d6c3010..1932112 100644 --- a/backend/services/platform-bridge/qq/adapter.go +++ b/backend/services/platform-bridge/qq/adapter.go @@ -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, } } diff --git a/backend/services/platform-bridge/qq/adapter_test.go b/backend/services/platform-bridge/qq/adapter_test.go new file mode 100644 index 0000000..232df96 --- /dev/null +++ b/backend/services/platform-bridge/qq/adapter_test.go @@ -0,0 +1,120 @@ +package qq + +import ( + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/services/platform-bridge" +) + +func wsServer(t *testing.T, onConn func(*websocket.Conn, int)) *httptest.Server { + t.Helper() + var upgrader = websocket.Upgrader{} + var count atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + n := int(count.Add(1)) + onConn(conn, n) + })) + t.Cleanup(srv.Close) + return srv +} + +func wsURL(s *httptest.Server) string { + return "ws" + strings.TrimPrefix(s.URL, "http") +} + +func TestParseMessageMissingSender(t *testing.T) { + a := New(Config{Name: "test", BotQQ: "123"}) + // Should not panic on malformed events + if msg := a.parseMessage(map[string]any{"post_type": "message", "message_type": "private"}); msg != nil { + t.Error("expected nil for event without sender") + } + if msg := a.parseMessage(map[string]any{"post_type": "notice"}); msg != nil { + t.Error("expected nil for non-message post_type") + } +} + +func TestReconnectAfterDrop(t *testing.T) { + srv := wsServer(t, func(conn *websocket.Conn, n int) { + if n == 1 { + conn.Close() // drop immediately to trigger reconnect + return + } + time.Sleep(2 * time.Second) // hold the second connection + conn.Close() + }) + + a := New(Config{Name: "test", WSURL: wsURL(srv)}) + if err := a.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + defer a.Stop() + + // Wait for the second connection (first is dropped immediately) + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if a.IsConnected() { + return // reconnected successfully + } + time.Sleep(50 * time.Millisecond) + } + t.Fatal("adapter did not reconnect after connection drop") +} + +func TestSendRateLimit(t *testing.T) { + received := make(chan struct{}, 4) + srv := wsServer(t, func(conn *websocket.Conn, n int) { + go func() { + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + received <- struct{}{} + } + }() + time.Sleep(3 * time.Second) + conn.Close() + }) + + a := New(Config{Name: "test", WSURL: wsURL(srv)}) + a.Start() + defer a.Stop() + + deadline := time.Now().Add(3 * time.Second) + for !a.IsConnected() && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if !a.IsConnected() { + t.Fatal("not connected") + } + + <-received // auth message + + start := time.Now() + if err := a.Send("10001", "第一条", bridge.SendOptions{MinGap: 300 * time.Millisecond}); err != nil { + t.Fatalf("send 1: %v", err) + } + if err := a.Send("10001", "第二条", bridge.SendOptions{MinGap: 300 * time.Millisecond}); err != nil { + t.Fatalf("send 2: %v", err) + } + if elapsed := time.Since(start); elapsed < 300*time.Millisecond { + t.Errorf("rate limit not enforced: two sends took %v", elapsed) + } +} + +func TestSendWhenDisconnected(t *testing.T) { + a := New(Config{Name: "test", WSURL: "ws://127.0.0.1:1/unreachable"}) + if err := a.Send("10001", "hi", bridge.SendOptions{}); err == nil { + t.Error("expected error when not connected") + } +}