feat(channels/qq_obv11): OneBot v11 渠道插件（正/反 WS 双模式 + CQ 媒体 + 限频，stdlib 自研 WS）

2026-09-06 06:46:43 +08:00
1 parent f9f5a78a02
commit 7831b0ea1b
@@ -0,0 +1,346 @@
+package qq_obv11
+
+// This file holds the message-flow half of the Adapter: the inbound read loop
+// (server/client mode sessions, reconnect, dispatch) and the outbound send path
+// (rate limit, chunk to the connection). It is split out of adapter.go to keep
+// each file under the project's 500-line limit.
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+)
+
+// ReadMessages registers emit as the pull-based sink and blocks until ctx is
+// canceled. Inbound events arrive via the read loop and are dispatched to
+// SetMessageHandler (push) when set, otherwise to emit.
+func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMessage)) {
+	a.emitMu.Lock()
+	a.emit = emit
+	a.emitMu.Unlock()
+	defer func() {
+		a.emitMu.Lock()
+		a.emit = nil
+		a.emitMu.Unlock()
+	}()
+	<-ctx.Done()
+}
+
+// SendMessage delivers a reply (possibly several messages) to the channel,
+// subject to the send rate limit.
+func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) error {
+	if err := ctx.Err(); err != nil {
+		return fmt.Errorf("qq-obv11: send message: %w", err)
+	}
+	conn := a.currentConn()
+	if conn == nil {
+		return fmt.Errorf("qq-obv11: send message: %w", adapter.ErrNotConnected)
+	}
+	for i, msg := range resp.Messages {
+		if err := a.sendMessage(ctx, conn, msg); err != nil {
+			return fmt.Errorf("qq-obv11: send message #%d: %w", i, err)
+		}
+	}
+	return nil
+}
+
+// SendProactive pushes a message not triggered by a recent inbound. It is
+// rate limited (see design/08 section 3: OneBot needs a rate limit to avoid
+// unusual burst patterns).
+func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error {
+	if err := ctx.Err(); err != nil {
+		return fmt.Errorf("qq-obv11: send proactive: %w", err)
+	}
+	conn := a.currentConn()
+	if conn == nil {
+		return fmt.Errorf("qq-obv11: send proactive: %w", adapter.ErrNotConnected)
+	}
+	return a.sendMessage(ctx, conn, msg)
+}
+
+// SetMessageHandler installs the push-style inbound callback. A nil fn clears it.
+func (a *Adapter) SetMessageHandler(fn adapter.MessageHandler) {
+	a.handlerMu.Lock()
+	a.handler = fn
+	a.handlerMu.Unlock()
+}
+
+// Capabilities declares what the QQ OneBot v11 adapter supports.
+func (a *Adapter) Capabilities() adapter.Capabilities {
+	return adapter.Capabilities{
+		MaxMessageLength:  defaultMaxMessageLength,
+		Markdown:          true,
+		Image:             true,
+		Voice:             true,
+		Reaction:          false,
+		TypingHint:        false,
+		RecommendBurstMax: defaultRecommendBurstMax,
+		Proactive:         true,
+	}.Normalize()
+}
+
+// ListChannels returns the groups the adapter currently knows about. Enumerating
+// friends / full group list requires the OneBot HTTP API, which is out of scope
+// for the WebSocket-only M6 adapter; it is reported as a follow-up.
+func (a *Adapter) ListChannels(ctx context.Context) ([]adapter.ChannelInfo, error) {
+	a.groupMu.RLock()
+	defer a.groupMu.RUnlock()
+	out := make([]adapter.ChannelInfo, 0, len(a.groups))
+	for id, name := range a.groups {
+		out = append(out, adapter.ChannelInfo{
+			ChannelID:   id,
+			ChannelType: adapter.ChannelGroup,
+			ChannelName: name,
+			GroupID:     id,
+			GroupName:   name,
+		})
+	}
+	return out, nil
+}
+
+// Addr returns the bound listen address in server mode ("" when not serving).
+func (a *Adapter) Addr() string {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	if a.ln == nil {
+		return ""
+	}
+	return a.ln.Addr().String()
+}
+
+// readLoop is the single inbound-reader goroutine. It reads frames from the
+// active connection, dispatching message events; in client mode it reconnects
+// after drops, in server mode it waits for a new OneBot client.
+func (a *Adapter) readLoop(ctx context.Context) {
+	for {
+		conn := a.currentConn()
+		if conn == nil {
+			if a.cfg.Mode == "client" {
+				if err := a.connectClient(ctx); err != nil {
+					a.log.Warn("onebot reconnect failed", "err", err)
+					if !sleepCtx(ctx, 3*time.Second) {
+						return
+					}
+					continue
+				}
+			} else if !a.waitConn(ctx) {
+				return
+			}
+			conn = a.currentConn()
+		}
+		a.readConn(ctx, conn)
+		if ctx.Err() != nil {
+			return
+		}
+	}
+}
+
+// readConn reads frames from conn until it closes, the context is canceled, or
+// a close frame is received, marking the adapter disconnected on exit.
+func (a *Adapter) readConn(ctx context.Context, conn *wsConn) {
+	watchDone := make(chan struct{})
+	go func() {
+		select {
+		case <-ctx.Done():
+			_ = conn.Close()
+		case <-watchDone:
+		}
+	}()
+	defer close(watchDone)
+
+	for {
+		opcode, payload, err := conn.ReadMessage()
+		if err != nil {
+			a.setDisconnected()
+			return
+		}
+		switch opcode {
+		case opPing:
+			_ = conn.WriteMessage(opPong, payload)
+		case opText, opBinary:
+			a.handlePayload(ctx, payload)
+		case opClose:
+			_ = conn.WriteMessage(opClose, nil)
+			a.setDisconnected()
+			return
+		}
+	}
+}
+
+// handlePayload parses a text frame, converting message events into normalized
+// UnifiedMessage and dispatching them.
+func (a *Adapter) handlePayload(ctx context.Context, payload []byte) {
+	ev, err := parseEvent(payload)
+	if err != nil {
+		a.log.Debug("non-event payload ignored", "err", err)
+		return
+	}
+	if ev.PostType != "message" {
+		return
+	}
+	msg := a.toUnified(ev, payload)
+	if err := msg.Validate(); err != nil {
+		a.log.Warn("invalid unified message dropped", "err", err)
+		return
+	}
+	a.dispatch(ctx, msg)
+}
+
+// dispatch routes an inbound message to the registered push handler, falling
+// back to the ReadMessages sink.
+func (a *Adapter) dispatch(ctx context.Context, msg adapter.UnifiedMessage) {
+	a.handlerMu.RLock()
+	fn := a.handler
+	a.handlerMu.RUnlock()
+	if fn != nil {
+		fn(ctx, msg)
+		return
+	}
+	a.emitMu.Lock()
+	emit := a.emit
+	a.emitMu.Unlock()
+	if emit != nil {
+		emit(msg)
+	}
+}
+
+// sendMessage builds and writes a single send_msg request, enforcing the rate
+// limit and writing the payload to the connection.
+func (a *Adapter) sendMessage(ctx context.Context, conn *wsConn, msg adapter.UnifiedMessage) error {
+	segs, err := outboundSegments(msg)
+	if err != nil {
+		return err
+	}
+	messageType, userID, groupID, err := resolveTarget(msg)
+	if err != nil {
+		return err
+	}
+	echo := fmt.Sprintf("echo_%d", time.Now().UnixNano())
+	payload, err := buildSendReq(messageType, userID, groupID, segs, echo)
+	if err != nil {
+		return err
+	}
+	if a.cfg.SendInterval > 0 {
+		if err := a.waitSendInterval(ctx); err != nil {
+			return err
+		}
+	}
+	if err := a.tryAcquireSend(); err != nil {
+		return err
+	}
+	if err := conn.WriteMessage(opText, payload); err != nil {
+		a.setDisconnected()
+		return fmt.Errorf("qq-obv11: write send_msg: %w", err)
+	}
+	a.mu.Lock()
+	a.lastSend = time.Now()
+	a.mu.Unlock()
+	return nil
+}
+
+// waitSendInterval sleeps so consecutive sends are at least SendInterval apart.
+func (a *Adapter) waitSendInterval(ctx context.Context) error {
+	for {
+		a.mu.RLock()
+		last := a.lastSend
+		a.mu.RUnlock()
+		if last.IsZero() {
+			return nil
+		}
+		wait := a.cfg.SendInterval - time.Since(last)
+		if wait <= 0 {
+			return nil
+		}
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("qq-obv11: send interval: %w", ctx.Err())
+		case <-time.After(wait):
+		}
+	}
+}
+
+// tryAcquireSend checks the send rate limiter, returning ErrRateLimited when the
+// configured window quota is exhausted.
+func (a *Adapter) tryAcquireSend() error {
+	if !a.limiter.allow(time.Now()) {
+		return fmt.Errorf("qq-obv11: %w", ErrRateLimited)
+	}
+	return nil
+}
+
+// setConn installs a new active connection, closing any previous one and waking
+// the read loop (server mode).
+func (a *Adapter) setConn(conn *wsConn) {
+	a.mu.Lock()
+	old := a.conn
+	a.conn = conn
+	a.connected = true
+	a.mu.Unlock()
+	if old != nil && old != conn {
+		_ = old.Close()
+	}
+	select {
+	case a.newConn <- struct{}{}:
+	default:
+	}
+}
+
+// setDisconnected marks the session down, dropping the current connection.
+func (a *Adapter) setDisconnected() {
+	a.mu.Lock()
+	if a.conn != nil {
+		_ = a.conn.Close()
+		a.conn = nil
+	}
+	a.connected = false
+	a.mu.Unlock()
+}
+
+// currentConn returns the active wsConn, or nil.
+func (a *Adapter) currentConn() *wsConn {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	return a.conn
+}
+
+// waitConn blocks in server mode until a OneBot client connects or ctx ends.
+func (a *Adapter) waitConn(ctx context.Context) bool {
+	for a.currentConn() == nil {
+		select {
+		case <-a.newConn:
+		case <-ctx.Done():
+			return false
+		case <-time.After(time.Second):
+		}
+	}
+	return true
+}
+
+func (a *Adapter) rememberGroup(id, name string) {
+	if id == "" {
+		return
+	}
+	a.groupMu.Lock()
+	a.groups[id] = name
+	a.groupMu.Unlock()
+}
+
+func (a *Adapter) groupName(id string) string {
+	if id == "" {
+		return ""
+	}
+	a.groupMu.RLock()
+	defer a.groupMu.RUnlock()
+	return a.groups[id]
+}
+
+func sleepCtx(ctx context.Context, d time.Duration) bool {
+	select {
+	case <-ctx.Done():
+		return false
+	case <-time.After(d):
+		return true
+	}
+}