feat(channels): enhance QQ channel with group/typing/media support and URL sanitization (#1208)

* feat(channels): enhance QQ channel with group support, typing, media, and URL sanitization Add group message routing alongside existing C2C (direct) support using chatType sync.Map to track whether a chatID is group or direct. Implement passive reply with msg_id/msg_seq tracking for multi-part responses. Add StartTyping (InputNotify msg_type=6 with periodic resend), SendMedia (RichMediaMessage for HTTP/HTTPS URLs), and configurable Markdown message support. Replace unbounded dedup map with TTL-based expiry and janitor goroutine. Sanitize URLs in group messages by replacing dots in domains with fullwidth period to avoid QQ's URL blacklist rejection (error 40054010). Add rate limit config (5 msg/s) and MaxMessageLength/SendMarkdown config fields. * fix(channels): address review feedback on QQ channel implementation - Fix goroutine leak: reinitialize done channel and sync.Once in Start() to prevent multiple janitor goroutines on restart - Fix double-close panic: guard close(done) with sync.Once in Stop() - Fix StartTyping context: use c.ctx (channel lifecycle) instead of caller's ctx (request lifecycle) for typing goroutine - Refactor: extract getChatKind() helper to deduplicate chatType lookup across Send(), StartTyping(), and SendMedia() - Fix: use new(atomic.Uint64) instead of taking address of local var - Fix: require explicit http(s):// scheme in URL regex to avoid false positives on version strings like "1.2.3" - Optimize: collect expired keys before deleting in dedupJanitor to reduce lock hold time - Fix: remove MaxMessageLength zero-value override in NewQQChannel since defaults.go already sets 2000 * fix(channels): address second round of review feedback on QQ channel - Fix SendMedia: bypass media store for direct http(s) URLs in part.Ref; only fall back to store.Resolve for media:// refs; log clear warning for local-only paths instead of silently skipping - Fix chatType routing: default unknown chatIDs to "group" (safer for QQ since outbound-only destinations like reasoning_channel_id are groups); pre-register reasoning_channel_id as group at Start() time; add debug log for untracked chatIDs - Add dedup hard cap (10000 entries): evict oldest entry when map exceeds capacity to prevent unbounded memory growth under high traffic
2026-06-12 18:08:54 +00:00 · 2026-03-10 12:07:02 +08:00
parent b89f6445d1
commit 2312553286
4 changed files with 370 additions and 33 deletions
@@ -63,6 +63,7 @@ var channelRateConfig = map[string]float64{
 	"slack":    1,
 	"matrix":   2,
 	"line":     10,
+	"qq":       5,
 	"irc":      2,
 }

@@ -3,7 +3,10 @@ package qq
 import (
 	"context"
 	"fmt"
+	"regexp"
+	"strings"
 	"sync"
+	"sync/atomic"
 	"time"

 	"github.com/tencent-connect/botgo"
@@ -20,6 +23,14 @@ import (
 	"github.com/sipeed/picoclaw/pkg/logger"
 )

+const (
+	dedupTTL      = 5 * time.Minute
+	dedupInterval = 60 * time.Second
+	dedupMaxSize  = 10000 // hard cap on dedup map entries
+	typingResend  = 8 * time.Second
+	typingSeconds = 10
+)
+
 type QQChannel struct {
 	*channels.BaseChannel
 	config         config.QQConfig
@@ -28,20 +39,37 @@ type QQChannel struct {
 	ctx            context.Context
 	cancel         context.CancelFunc
 	sessionManager botgo.SessionManager
-	processedIDs   map[string]bool
-	mu             sync.RWMutex
+
+	// Chat routing: track whether a chatID is group or direct.
+	chatType sync.Map // chatID → "group" | "direct"
+
+	// Passive reply: store last inbound message ID per chat.
+	lastMsgID sync.Map // chatID → string
+
+	// msg_seq: per-chat atomic counter for multi-part replies.
+	msgSeqCounters sync.Map // chatID → *atomic.Uint64
+
+	// Time-based dedup replacing the unbounded map.
+	dedup   map[string]time.Time
+	muDedup sync.Mutex
+
+	// done is closed on Stop to shut down the dedup janitor.
+	done     chan struct{}
+	stopOnce sync.Once
 }

 func NewQQChannel(cfg config.QQConfig, messageBus *bus.MessageBus) (*QQChannel, error) {
 	base := channels.NewBaseChannel("qq", cfg, messageBus, cfg.AllowFrom,
+		channels.WithMaxMessageLength(cfg.MaxMessageLength),
 		channels.WithGroupTrigger(cfg.GroupTrigger),
 		channels.WithReasoningChannelID(cfg.ReasoningChannelID),
 	)

 	return &QQChannel{
-		BaseChannel:  base,
-		config:       cfg,
-		processedIDs: make(map[string]bool),
+		BaseChannel: base,
+		config:      cfg,
+		dedup:       make(map[string]time.Time),
+		done:        make(chan struct{}),
 	}, nil
 }

@@ -52,6 +80,10 @@ func (c *QQChannel) Start(ctx context.Context) error {

 	logger.InfoC("qq", "Starting QQ bot (WebSocket mode)")

+	// Reinitialize shutdown signal for clean restart.
+	c.done = make(chan struct{})
+	c.stopOnce = sync.Once{}
+
 	// create token source
 	credentials := &token.QQBotCredentials{
 		AppID:     c.config.AppID,
@@ -99,6 +131,15 @@ func (c *QQChannel) Start(ctx context.Context) error {
 		}
 	}()

+	// start dedup janitor goroutine
+	go c.dedupJanitor()
+
+	// Pre-register reasoning_channel_id as group chat if configured,
+	// so outbound-only destinations are routed correctly.
+	if c.config.ReasoningChannelID != "" {
+		c.chatType.Store(c.config.ReasoningChannelID, "group")
+	}
+
 	c.SetRunning(true)
 	logger.InfoC("qq", "QQ bot started successfully")

@@ -109,6 +150,9 @@ func (c *QQChannel) Stop(ctx context.Context) error {
 	logger.InfoC("qq", "Stopping QQ bot")
 	c.SetRunning(false)

+	// Signal the dedup janitor to stop (idempotent).
+	c.stopOnce.Do(func() { close(c.done) })
+
 	if c.cancel != nil {
 		c.cancel()
 	}
@@ -116,21 +160,82 @@ func (c *QQChannel) Stop(ctx context.Context) error {
 	return nil
 }

+// getChatKind returns the chat type for a given chatID ("group" or "direct").
+// Unknown chatIDs default to "group" and log a warning, since QQ group IDs are
+// more common as outbound-only destinations (e.g. reasoning_channel_id).
+func (c *QQChannel) getChatKind(chatID string) string {
+	if v, ok := c.chatType.Load(chatID); ok {
+		if k, ok := v.(string); ok {
+			return k
+		}
+	}
+	logger.DebugCF("qq", "Unknown chat type for chatID, defaulting to group", map[string]any{
+		"chat_id": chatID,
+	})
+	return "group"
+}
+
 func (c *QQChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
 	if !c.IsRunning() {
 		return channels.ErrNotRunning
 	}

-	// construct message
+	chatKind := c.getChatKind(msg.ChatID)
+
+	// Build message with content.
 	msgToCreate := &dto.MessageToCreate{
 		Content: msg.Content,
+		MsgType: dto.TextMsg,
+	}
+
+	// Use Markdown message type if enabled in config.
+	if c.config.SendMarkdown {
+		msgToCreate.MsgType = dto.MarkdownMsg
+		msgToCreate.Markdown = &dto.Markdown{
+			Content: msg.Content,
+		}
+		// Clear plain content to avoid sending duplicate text.
+		msgToCreate.Content = ""
+	}
+
+	// Attach passive reply msg_id and msg_seq if available.
+	if v, ok := c.lastMsgID.Load(msg.ChatID); ok {
+		if msgID, ok := v.(string); ok && msgID != "" {
+			msgToCreate.MsgID = msgID
+
+			// Increment msg_seq atomically for multi-part replies.
+			if counterVal, ok := c.msgSeqCounters.Load(msg.ChatID); ok {
+				if counter, ok := counterVal.(*atomic.Uint64); ok {
+					seq := counter.Add(1)
+					msgToCreate.MsgSeq = uint32(seq)
+				}
+			}
+		}
+	}
+
+	// Sanitize URLs in group messages to avoid QQ's URL blacklist rejection.
+	if chatKind == "group" {
+		if msgToCreate.Content != "" {
+			msgToCreate.Content = sanitizeURLs(msgToCreate.Content)
+		}
+		if msgToCreate.Markdown != nil && msgToCreate.Markdown.Content != "" {
+			msgToCreate.Markdown.Content = sanitizeURLs(msgToCreate.Markdown.Content)
+		}
+	}
+
+	// Route to group or C2C.
+	var err error
+	if chatKind == "group" {
+		_, err = c.api.PostGroupMessage(ctx, msg.ChatID, msgToCreate)
+	} else {
+		_, err = c.api.PostC2CMessage(ctx, msg.ChatID, msgToCreate)
 	}

-	// send C2C message
-	_, err := c.api.PostC2CMessage(ctx, msg.ChatID, msgToCreate)
 	if err != nil {
-		logger.ErrorCF("qq", "Failed to send C2C message", map[string]any{
-			"error": err.Error(),
+		logger.ErrorCF("qq", "Failed to send message", map[string]any{
+			"chat_id":   msg.ChatID,
+			"chat_kind": chatKind,
+			"error":     err.Error(),
 		})
 		return fmt.Errorf("qq send: %w", channels.ErrTemporary)
 	}
@@ -138,7 +243,150 @@ func (c *QQChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
 	return nil
 }

-// handleC2CMessage handles QQ private messages
+// StartTyping implements channels.TypingCapable.
+// It sends an InputNotify (msg_type=6) immediately and re-sends every 8 seconds.
+// The returned stop function is idempotent and cancels the goroutine.
+func (c *QQChannel) StartTyping(ctx context.Context, chatID string) (func(), error) {
+	// We need a stored msg_id for passive InputNotify; skip if none available.
+	v, ok := c.lastMsgID.Load(chatID)
+	if !ok {
+		return func() {}, nil
+	}
+	msgID, ok := v.(string)
+	if !ok || msgID == "" {
+		return func() {}, nil
+	}
+
+	chatKind := c.getChatKind(chatID)
+
+	sendTyping := func(sendCtx context.Context) {
+		typingMsg := &dto.MessageToCreate{
+			MsgType: dto.InputNotifyMsg,
+			MsgID:   msgID,
+			InputNotify: &dto.InputNotify{
+				InputType:   1,
+				InputSecond: typingSeconds,
+			},
+		}
+
+		var err error
+		if chatKind == "group" {
+			_, err = c.api.PostGroupMessage(sendCtx, chatID, typingMsg)
+		} else {
+			_, err = c.api.PostC2CMessage(sendCtx, chatID, typingMsg)
+		}
+		if err != nil {
+			logger.DebugCF("qq", "Failed to send typing indicator", map[string]any{
+				"chat_id": chatID,
+				"error":   err.Error(),
+			})
+		}
+	}
+
+	// Send immediately.
+	sendTyping(c.ctx)
+
+	typingCtx, cancel := context.WithCancel(c.ctx)
+	go func() {
+		ticker := time.NewTicker(typingResend)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-typingCtx.Done():
+				return
+			case <-ticker.C:
+				sendTyping(typingCtx)
+			}
+		}
+	}()
+
+	return cancel, nil
+}
+
+// SendMedia implements the channels.MediaSender interface.
+// QQ RichMediaMessage requires an HTTP/HTTPS URL — local file paths are not supported.
+// If part.Ref is already an http(s) URL it is used directly; otherwise we try
+// the media store, and skip with a warning if the resolved path is not an HTTP URL.
+func (c *QQChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) error {
+	if !c.IsRunning() {
+		return channels.ErrNotRunning
+	}
+
+	chatKind := c.getChatKind(msg.ChatID)
+
+	for _, part := range msg.Parts {
+		// If the ref is already an HTTP(S) URL, use it directly.
+		mediaURL := part.Ref
+		if !isHTTPURL(mediaURL) {
+			// Try resolving through media store.
+			store := c.GetMediaStore()
+			if store == nil {
+				logger.WarnCF("qq", "QQ media requires HTTP/HTTPS URL, no media store available", map[string]any{
+					"ref": part.Ref,
+				})
+				continue
+			}
+
+			resolved, err := store.Resolve(part.Ref)
+			if err != nil {
+				logger.ErrorCF("qq", "Failed to resolve media ref", map[string]any{
+					"ref":   part.Ref,
+					"error": err.Error(),
+				})
+				continue
+			}
+
+			if !isHTTPURL(resolved) {
+				logger.WarnCF("qq", "QQ media requires HTTP/HTTPS URL, local files not supported", map[string]any{
+					"ref":      part.Ref,
+					"resolved": resolved,
+				})
+				continue
+			}
+
+			mediaURL = resolved
+		}
+
+		// Map part type to QQ file type: 1=image, 2=video, 3=audio, 4=file.
+		var fileType uint64
+		switch part.Type {
+		case "image":
+			fileType = 1
+		case "video":
+			fileType = 2
+		case "audio":
+			fileType = 3
+		default:
+			fileType = 4 // file
+		}
+
+		richMedia := &dto.RichMediaMessage{
+			FileType:   fileType,
+			URL:        mediaURL,
+			SrvSendMsg: true,
+		}
+
+		var sendErr error
+		if chatKind == "group" {
+			_, sendErr = c.api.PostGroupMessage(ctx, msg.ChatID, richMedia)
+		} else {
+			_, sendErr = c.api.PostC2CMessage(ctx, msg.ChatID, richMedia)
+		}
+
+		if sendErr != nil {
+			logger.ErrorCF("qq", "Failed to send media", map[string]any{
+				"type":    part.Type,
+				"chat_id": msg.ChatID,
+				"error":   sendErr.Error(),
+			})
+			return fmt.Errorf("qq send media: %w", channels.ErrTemporary)
+		}
+	}
+
+	return nil
+}
+
+// handleC2CMessage handles QQ private messages.
 func (c *QQChannel) handleC2CMessage() event.C2CMessageEventHandler {
 	return func(event *dto.WSPayload, data *dto.WSC2CMessageData) error {
 		// deduplication check
@@ -167,7 +415,13 @@ func (c *QQChannel) handleC2CMessage() event.C2CMessageEventHandler {
 			"length": len(content),
 		})

-		// 转发到消息总线
+		// Store chat routing context.
+		c.chatType.Store(senderID, "direct")
+		c.lastMsgID.Store(senderID, data.ID)
+
+		// Reset msg_seq counter for new inbound message.
+		c.msgSeqCounters.Store(senderID, new(atomic.Uint64))
+
 		metadata := map[string]string{}

 		sender := bus.SenderInfo{
@@ -195,7 +449,7 @@ func (c *QQChannel) handleC2CMessage() event.C2CMessageEventHandler {
 	}
 }

-// handleGroupATMessage handles QQ group @ messages
+// handleGroupATMessage handles QQ group @ messages.
 func (c *QQChannel) handleGroupATMessage() event.GroupATMessageEventHandler {
 	return func(event *dto.WSPayload, data *dto.WSGroupATMessageData) error {
 		// deduplication check
@@ -232,7 +486,13 @@ func (c *QQChannel) handleGroupATMessage() event.GroupATMessageEventHandler {
 			"length": len(content),
 		})

-		// 转发到消息总线（使用 GroupID 作为 ChatID）
+		// Store chat routing context using GroupID as chatID.
+		c.chatType.Store(data.GroupID, "group")
+		c.lastMsgID.Store(data.GroupID, data.ID)
+
+		// Reset msg_seq counter for new inbound message.
+		c.msgSeqCounters.Store(data.GroupID, new(atomic.Uint64))
+
 		metadata := map[string]string{
 			"group_id": data.GroupID,
 		}
@@ -262,29 +522,102 @@ func (c *QQChannel) handleGroupATMessage() event.GroupATMessageEventHandler {
 	}
 }

-// isDuplicate 检查消息是否重复
+// isDuplicate checks whether a message has been seen within the TTL window.
+// It also enforces a hard cap on map size by evicting oldest entries.
 func (c *QQChannel) isDuplicate(messageID string) bool {
-	c.mu.Lock()
-	defer c.mu.Unlock()
+	c.muDedup.Lock()
+	defer c.muDedup.Unlock()

-	if c.processedIDs[messageID] {
+	if ts, exists := c.dedup[messageID]; exists && time.Since(ts) < dedupTTL {
 		return true
 	}

-	c.processedIDs[messageID] = true
-
-	// 简单清理：限制 map 大小
-	if len(c.processedIDs) > 10000 {
-		// 清空一半
-		count := 0
-		for id := range c.processedIDs {
-			if count >= 5000 {
-				break
+	// Enforce hard cap: evict oldest entries when at capacity.
+	if len(c.dedup) >= dedupMaxSize {
+		var oldestID string
+		var oldestTS time.Time
+		for id, ts := range c.dedup {
+			if oldestID == "" || ts.Before(oldestTS) {
+				oldestID = id
+				oldestTS = ts
 			}
-			delete(c.processedIDs, id)
-			count++
+		}
+		if oldestID != "" {
+			delete(c.dedup, oldestID)
 		}
 	}

+	c.dedup[messageID] = time.Now()
 	return false
 }
+
+// dedupJanitor periodically evicts expired entries from the dedup map.
+func (c *QQChannel) dedupJanitor() {
+	ticker := time.NewTicker(dedupInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-c.done:
+			return
+		case <-ticker.C:
+			// Collect expired keys under read-like scan.
+			c.muDedup.Lock()
+			now := time.Now()
+			var expired []string
+			for id, ts := range c.dedup {
+				if now.Sub(ts) >= dedupTTL {
+					expired = append(expired, id)
+				}
+			}
+			for _, id := range expired {
+				delete(c.dedup, id)
+			}
+			c.muDedup.Unlock()
+		}
+	}
+}
+
+// isHTTPURL returns true if s starts with http:// or https://.
+func isHTTPURL(s string) bool {
+	return strings.HasPrefix(s, "http://") || strings.HasPrefix(s, "https://")
+}
+
+// urlPattern matches URLs with explicit http(s):// scheme.
+// Only scheme-prefixed URLs are matched to avoid false positives on bare text
+// like version numbers (e.g., "1.2.3") or domain-like fragments.
+var urlPattern = regexp.MustCompile(
+	`(?i)` +
+		`https?://` + // required scheme
+		`(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+` + // domain parts
+		`[a-zA-Z]{2,}` + // TLD
+		`(?:[/?#]\S*)?`, // optional path/query/fragment
+)
+
+// sanitizeURLs replaces dots in URL domains with "。" (fullwidth period)
+// to prevent QQ's URL blacklist from rejecting the message.
+func sanitizeURLs(text string) string {
+	return urlPattern.ReplaceAllStringFunc(text, func(match string) string {
+		// Split into scheme + rest (scheme is always present).
+		idx := strings.Index(match, "://")
+		scheme := match[:idx+3]
+		rest := match[idx+3:]
+
+		// Find where the domain ends (first / ? or #).
+		domainEnd := len(rest)
+		for i, ch := range rest {
+			if ch == '/' || ch == '?' || ch == '#' {
+				domainEnd = i
+				break
+			}
+		}
+
+		domain := rest[:domainEnd]
+		path := rest[domainEnd:]
+
+		// Replace dots in domain only.
+		domain = strings.ReplaceAll(domain, ".", "。")
+
+		return scheme + domain + path
+	})
+}
@@ -312,6 +312,8 @@ type QQConfig struct {
 	AppSecret          string              `json:"app_secret"              env:"PICOCLAW_CHANNELS_QQ_APP_SECRET"`
 	AllowFrom          FlexibleStringSlice `json:"allow_from"              env:"PICOCLAW_CHANNELS_QQ_ALLOW_FROM"`
 	GroupTrigger       GroupTriggerConfig  `json:"group_trigger,omitempty"`
+	MaxMessageLength   int                 `json:"max_message_length"      env:"PICOCLAW_CHANNELS_QQ_MAX_MESSAGE_LENGTH"`
+	SendMarkdown       bool                `json:"send_markdown"           env:"PICOCLAW_CHANNELS_QQ_SEND_MARKDOWN"`
 	ReasoningChannelID string              `json:"reasoning_channel_id"    env:"PICOCLAW_CHANNELS_QQ_REASONING_CHANNEL_ID"`
 }

@@ -80,10 +80,11 @@ func DefaultConfig() *Config {
 				AllowFrom: FlexibleStringSlice{},
 			},
 			QQ: QQConfig{
-				Enabled:   false,
-				AppID:     "",
-				AppSecret: "",
-				AllowFrom: FlexibleStringSlice{},
+				Enabled:          false,
+				AppID:            "",
+				AppSecret:        "",
+				AllowFrom:        FlexibleStringSlice{},
+				MaxMessageLength: 2000,
 			},
 			DingTalk: DingTalkConfig{
 				Enabled:      false,