fix(qq): use openid routing and allowlist compatibility

pkg/channels/qq/qq.go

@@ -41,6 +41,7 @@ const (
 	typingResend  = 8 * time.Second
 	typingSeconds = 10
 	bytesPerMiB   = 1024 * 1024
+	qqStartupProbe = 15 * time.Second
 )
 
 type qqAPI interface {
@@ -82,6 +83,60 @@ type QQChannel struct {
 	stopOnce sync.Once
 }
 
+type qqRawAuthor struct {
+	UserOpenID   string `json:"user_openid"`
+	MemberOpenID string `json:"member_openid"`
+}
+
+type qqRawEnvelope struct {
+	D struct {
+		Author      qqRawAuthor `json:"author"`
+		GroupOpenID string      `json:"group_openid"`
+	} `json:"d"`
+}
+
+func parseQQOpenIDs(raw []byte) (userOpenID, memberOpenID, groupOpenID string) {
+	if len(raw) == 0 {
+		return "", "", ""
+	}
+
+	var env qqRawEnvelope
+	if err := json.Unmarshal(raw, &env); err != nil {
+		return "", "", ""
+	}
+
+	return strings.TrimSpace(env.D.Author.UserOpenID),
+		strings.TrimSpace(env.D.Author.MemberOpenID),
+		strings.TrimSpace(env.D.GroupOpenID)
+}
+
+// resolveAllowedSender validates sender against allow_from using openid first,
+// then falls back to legacy author/member id for backward compatibility.
+func (c *QQChannel) resolveAllowedSender(primaryOpenID, legacyID string) (bus.SenderInfo, bool, bool) {
+	primary := bus.SenderInfo{
+		Platform:    "qq",
+		PlatformID:  primaryOpenID,
+		CanonicalID: identity.BuildCanonicalID("qq", primaryOpenID),
+	}
+
+	if c.IsAllowedSender(primary) {
+		return primary, true, false
+	}
+
+	legacyID = strings.TrimSpace(legacyID)
+	if legacyID != "" && legacyID != primaryOpenID {
+		legacy := bus.SenderInfo{
+			Platform:   "qq",
+			PlatformID: legacyID,
+		}
+		if c.IsAllowedSender(legacy) {
+			return legacy, true, true
+		}
+	}
+
+	return primary, false, false
+}
+
 func NewQQChannel(cfg config.QQConfig, messageBus *bus.MessageBus) (*QQChannel, error) {
 	base := channels.NewBaseChannel("qq", cfg, messageBus, cfg.AllowFrom,
 		channels.WithMaxMessageLength(cfg.MaxMessageLength),
@@ -127,8 +182,24 @@ func (c *QQChannel) Start(ctx context.Context) error {
 	// initialize OpenAPI client
 	c.api = botgo.NewOpenAPI(c.config.AppID, c.tokenSource).WithTimeout(5 * time.Second)
 
+	readyCh := make(chan struct{}, 1)
+	sessionErrCh := make(chan error, 1)
+
 	// register event handlers
 	intent := event.RegisterHandlers(
+		event.ReadyHandler(func(_ *dto.WSPayload, _ *dto.WSReadyData) {
+			select {
+			case readyCh <- struct{}{}:
+			default:
+			}
+		}),
+		event.ErrorNotifyHandler(func(err error) {
+			fmt.Printf("QQ gateway error: %v\n", err)
+			select {
+			case sessionErrCh <- err:
+			default:
+			}
+		}),
 		c.handleC2CMessage(),
 		c.handleGroupATMessage(),
 	)
@@ -145,17 +216,50 @@ func (c *QQChannel) Start(ctx context.Context) error {
 
 	// create and save sessionManager
 	c.sessionManager = botgo.NewSessionManager()
+	startupErr := make(chan error, 1)
 
 	// start WebSocket connection in goroutine to avoid blocking
 	go func() {
-		if err := c.sessionManager.Start(wsInfo, c.tokenSource, &intent); err != nil {
+		err := c.sessionManager.Start(wsInfo, c.tokenSource, &intent)
+		if err != nil {
 			logger.ErrorCF("qq", "WebSocket session error", map[string]any{
 				"error": err.Error(),
 			})
+			fmt.Printf("QQ WebSocket session error: %v\n", err)
 			c.SetRunning(false)
 		}
+		select {
+		case startupErr <- err:
+		default:
+		}
 	}()
 
+	select {
+	case <-readyCh:
+		fmt.Println("QQ WebSocket ready")
+	case err := <-sessionErrCh:
+		if c.cancel != nil {
+			c.cancel()
+		}
+		if err != nil {
+			return fmt.Errorf("QQ websocket failed before ready: %w", err)
+		}
+		return fmt.Errorf("QQ websocket failed before ready")
+	case err := <-startupErr:
+		if c.cancel != nil {
+			c.cancel()
+		}
+		if err != nil {
+			return fmt.Errorf("failed to establish QQ websocket session: %w", err)
+		}
+		return fmt.Errorf("QQ websocket session exited unexpectedly during startup")
+	case <-time.After(qqStartupProbe):
+		if c.cancel != nil {
+			c.cancel()
+		}
+		return fmt.Errorf("timeout waiting for QQ websocket READY event")
+	}
+
 	// start dedup janitor goroutine
 	go c.dedupJanitor()
 
@@ -597,29 +701,53 @@ func (c *QQChannel) maxBase64FileSizeBytes() int64 {
 // handleC2CMessage handles QQ private messages.
 func (c *QQChannel) handleC2CMessage() event.C2CMessageEventHandler {
 	return func(event *dto.WSPayload, data *dto.WSC2CMessageData) error {
+		if data == nil {
+			return nil
+		}
+
 		// deduplication check
 		if c.isDuplicate(data.ID) {
 			return nil
 		}
 
-		// extract user info
+		var raw []byte
-		var senderID string
+		if event != nil {
-		if data.Author != nil && data.Author.ID != "" {
+			raw = event.RawMessage
-			senderID = data.Author.ID
+		}
-		} else {
+		userOpenID, _, _ := parseQQOpenIDs(raw)
+		legacyAuthorID := ""
+		if data.Author != nil {
+			legacyAuthorID = strings.TrimSpace(data.Author.ID)
+		}
+
+		// QQ C2C endpoint requires user_openid; fallback to author.id for compatibility.
+		senderID := userOpenID
+		if senderID == "" {
+			senderID = legacyAuthorID
+		}
+		if senderID == "" {
 			logger.WarnC("qq", "Received message with no sender ID")
 			return nil
 		}
 
-		sender := bus.SenderInfo{
+		sender, allowed, usedLegacyFallback := c.resolveAllowedSender(senderID, legacyAuthorID)
-			Platform:    "qq",
+		if !allowed {
-			PlatformID:  data.Author.ID,
+			logger.WarnCF("qq", "Dropped C2C message by allow_from", map[string]any{
-			CanonicalID: identity.BuildCanonicalID("qq", data.Author.ID),
+				"sender_openid": senderID,
-		}
+				"legacy_id":     legacyAuthorID,
-
+				"message_id":    data.ID,
-		if !c.IsAllowedSender(sender) {
+			})
+			fmt.Println("QQ inbound C2C dropped by allow_from")
 			return nil
 		}
+		if usedLegacyFallback {
+			logger.WarnCF("qq", "allow_from matched legacy QQ sender id", map[string]any{
+				"sender_openid": senderID,
+				"legacy_id":     legacyAuthorID,
+				"message_id":    data.ID,
+			})
+			fmt.Println("QQ inbound C2C allow_from matched legacy id")
+		}
 
 		content := strings.TrimSpace(data.Content)
 		mediaPaths, attachmentNotes := c.extractInboundAttachments(senderID, data.ID, data.Attachments)
@@ -638,21 +766,25 @@ func (c *QQChannel) handleC2CMessage() event.C2CMessageEventHandler {
 		})
 
 		// Store chat routing context.
-		c.chatType.Store(senderID, "direct")
+		chatID := senderID
-		c.lastMsgID.Store(senderID, data.ID)
+		c.chatType.Store(chatID, "direct")
+		c.lastMsgID.Store(chatID, data.ID)
 
 		// Reset msg_seq counter for new inbound message.
-		c.msgSeqCounters.Store(senderID, new(atomic.Uint64))
+		c.msgSeqCounters.Store(chatID, new(atomic.Uint64))
 
 		metadata := map[string]string{
 			"account_id": senderID,
 		}
+		if legacyAuthorID != "" && legacyAuthorID != senderID {
+			metadata["legacy_account_id"] = legacyAuthorID
+		}
 
 		c.HandleMessage(c.ctx,
-			bus.Peer{Kind: "direct", ID: senderID},
+			bus.Peer{Kind: "direct", ID: chatID},
 			data.ID,
 			senderID,
-			senderID,
+			chatID,
 			content,
 			mediaPaths,
 			metadata,
@@ -666,32 +798,67 @@ func (c *QQChannel) handleC2CMessage() event.C2CMessageEventHandler {
 // handleGroupATMessage handles QQ group @ messages.
 func (c *QQChannel) handleGroupATMessage() event.GroupATMessageEventHandler {
 	return func(event *dto.WSPayload, data *dto.WSGroupATMessageData) error {
+		if data == nil {
+			return nil
+		}
+
 		// deduplication check
 		if c.isDuplicate(data.ID) {
 			return nil
 		}
 
-		// extract user info
+		var raw []byte
-		var senderID string
+		if event != nil {
-		if data.Author != nil && data.Author.ID != "" {
+			raw = event.RawMessage
-			senderID = data.Author.ID
+		}
-		} else {
+		_, memberOpenID, groupOpenID := parseQQOpenIDs(raw)
+		legacyMemberID := ""
+		if data.Author != nil {
+			legacyMemberID = strings.TrimSpace(data.Author.ID)
+		}
+
+		// For QQ group callbacks, member_openid/group_openid are preferred identifiers.
+		senderID := memberOpenID
+		if senderID == "" {
+			senderID = legacyMemberID
+		}
+		if senderID == "" {
 			logger.WarnC("qq", "Received group message with no sender ID")
 			return nil
 		}
 
-		sender := bus.SenderInfo{
+		chatID := groupOpenID
-			Platform:    "qq",
+		if chatID == "" {
-			PlatformID:  data.Author.ID,
+			chatID = strings.TrimSpace(data.GroupID)
-			CanonicalID: identity.BuildCanonicalID("qq", data.Author.ID),
 		}
-
+		if chatID == "" {
-		if !c.IsAllowedSender(sender) {
+			logger.WarnC("qq", "Received group message with no group ID")
 			return nil
 		}
 
+		sender, allowed, usedLegacyFallback := c.resolveAllowedSender(senderID, legacyMemberID)
+		if !allowed {
+			logger.WarnCF("qq", "Dropped group message by allow_from", map[string]any{
+				"sender_openid": senderID,
+				"legacy_id":     legacyMemberID,
+				"group_id":      chatID,
+				"message_id":    data.ID,
+			})
+			fmt.Println("QQ inbound group dropped by allow_from")
+			return nil
+		}
+		if usedLegacyFallback {
+			logger.WarnCF("qq", "allow_from matched legacy QQ member id", map[string]any{
+				"sender_openid": senderID,
+				"legacy_id":     legacyMemberID,
+				"group_id":      chatID,
+				"message_id":    data.ID,
+			})
+			fmt.Println("QQ inbound group allow_from matched legacy id")
+		}
+
 		content := strings.TrimSpace(data.Content)
-		mediaPaths, attachmentNotes := c.extractInboundAttachments(data.GroupID, data.ID, data.Attachments)
+		mediaPaths, attachmentNotes := c.extractInboundAttachments(chatID, data.ID, data.Attachments)
 		for _, note := range attachmentNotes {
 			content = appendContent(content, note)
 		}
@@ -709,28 +876,31 @@ func (c *QQChannel) handleGroupATMessage() event.GroupATMessageEventHandler {
 
 		logger.InfoCF("qq", "Received group AT message", map[string]any{
 			"sender":      senderID,
-			"group":       data.GroupID,
+			"group":       chatID,
 			"length":      len(content),
 			"media_count": len(mediaPaths),
 		})
 
 		// Store chat routing context using GroupID as chatID.
-		c.chatType.Store(data.GroupID, "group")
+		c.chatType.Store(chatID, "group")
-		c.lastMsgID.Store(data.GroupID, data.ID)
+		c.lastMsgID.Store(chatID, data.ID)
 
 		// Reset msg_seq counter for new inbound message.
-		c.msgSeqCounters.Store(data.GroupID, new(atomic.Uint64))
+		c.msgSeqCounters.Store(chatID, new(atomic.Uint64))
 
 		metadata := map[string]string{
 			"account_id": senderID,
-			"group_id":   data.GroupID,
+			"group_id":   chatID,
+		}
+		if legacyMemberID != "" && legacyMemberID != senderID {
+			metadata["legacy_account_id"] = legacyMemberID
 		}
 
 		c.HandleMessage(c.ctx,
-			bus.Peer{Kind: "group", ID: data.GroupID},
+			bus.Peer{Kind: "group", ID: chatID},
 			data.ID,
 			senderID,
-			data.GroupID,
+			chatID,
 			content,
 			mediaPaths,
 			metadata,

pkg/channels/qq/qq_test.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"errors"
 	"os"
+	"path/filepath"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -62,6 +63,187 @@ func TestHandleC2CMessage_IncludesAccountIDMetadata(t *testing.T) {
 	}
 }
 
+func TestHandleC2CMessage_PrefersUserOpenIDFromRawPayload(t *testing.T) {
+	messageBus := bus.NewMessageBus()
+	ch := &QQChannel{
+		BaseChannel: channels.NewBaseChannel("qq", nil, messageBus, nil),
+		dedup:       make(map[string]time.Time),
+		done:        make(chan struct{}),
+		ctx:         context.Background(),
+	}
+
+	raw := []byte(`{"d":{"author":{"id":"legacy-id","user_openid":"user-openid-123"}}}`)
+	err := ch.handleC2CMessage()(&dto.WSPayload{RawMessage: raw}, &dto.WSC2CMessageData{
+		ID:      "msg-openid",
+		Content: "hello",
+		Author: &dto.User{
+			ID: "legacy-id",
+		},
+	})
+	if err != nil {
+		t.Fatalf("handleC2CMessage() error = %v", err)
+	}
+
+	inbound := waitInboundMessage(t, messageBus)
+	if inbound.Metadata["account_id"] != "user-openid-123" {
+		t.Fatalf("account_id metadata = %q, want %q", inbound.Metadata["account_id"], "user-openid-123")
+	}
+	if inbound.ChatID != "user-openid-123" {
+		t.Fatalf("inbound.ChatID = %q, want %q", inbound.ChatID, "user-openid-123")
+	}
+}
+
+func TestHandleC2CMessage_AllowListFallsBackToLegacyID(t *testing.T) {
+	messageBus := bus.NewMessageBus()
+	ch := &QQChannel{
+		BaseChannel: channels.NewBaseChannel("qq", nil, messageBus, []string{"legacy-id"}),
+		dedup:       make(map[string]time.Time),
+		done:        make(chan struct{}),
+		ctx:         context.Background(),
+	}
+
+	raw := []byte(`{"d":{"author":{"id":"legacy-id","user_openid":"user-openid-123"}}}`)
+	err := ch.handleC2CMessage()(&dto.WSPayload{RawMessage: raw}, &dto.WSC2CMessageData{
+		ID:      "msg-allow-fallback",
+		Content: "hello",
+		Author: &dto.User{
+			ID: "legacy-id",
+		},
+	})
+	if err != nil {
+		t.Fatalf("handleC2CMessage() error = %v", err)
+	}
+
+	inbound := waitInboundMessage(t, messageBus)
+	if inbound.SenderID != "user-openid-123" {
+		t.Fatalf("inbound.SenderID = %q, want %q", inbound.SenderID, "user-openid-123")
+	}
+	if inbound.Metadata["account_id"] != "user-openid-123" {
+		t.Fatalf("account_id metadata = %q, want %q", inbound.Metadata["account_id"], "user-openid-123")
+	}
+	if inbound.Metadata["legacy_account_id"] != "legacy-id" {
+		t.Fatalf("legacy_account_id metadata = %q, want %q", inbound.Metadata["legacy_account_id"], "legacy-id")
+	}
+}
+
+func TestHandleGroupATMessage_PrefersGroupOpenIDFromRawPayload(t *testing.T) {
+	messageBus := bus.NewMessageBus()
+	ch := &QQChannel{
+		BaseChannel: channels.NewBaseChannel("qq", nil, messageBus, nil),
+		dedup:       make(map[string]time.Time),
+		done:        make(chan struct{}),
+		ctx:         context.Background(),
+	}
+
+	raw := []byte(`{"d":{"group_openid":"group-openid-abc","author":{"id":"legacy-member-id","member_openid":"member-openid-xyz"}}}`)
+	err := ch.handleGroupATMessage()(&dto.WSPayload{RawMessage: raw}, &dto.WSGroupATMessageData{
+		ID:      "group-openid-msg",
+		GroupID: "legacy-group-id",
+		Content: "@bot hello",
+		Author: &dto.User{
+			ID: "legacy-member-id",
+		},
+	})
+	if err != nil {
+		t.Fatalf("handleGroupATMessage() error = %v", err)
+	}
+
+	inbound := waitInboundMessage(t, messageBus)
+	if inbound.Peer.ID != "group-openid-abc" {
+		t.Fatalf("inbound.Peer.ID = %q, want %q", inbound.Peer.ID, "group-openid-abc")
+	}
+	if inbound.Metadata["group_id"] != "group-openid-abc" {
+		t.Fatalf("group_id metadata = %q, want %q", inbound.Metadata["group_id"], "group-openid-abc")
+	}
+	if inbound.Metadata["account_id"] != "member-openid-xyz" {
+		t.Fatalf("account_id metadata = %q, want %q", inbound.Metadata["account_id"], "member-openid-xyz")
+	}
+}
+
+func TestHandleGroupATMessage_AllowListFallsBackToLegacyMemberID(t *testing.T) {
+	messageBus := bus.NewMessageBus()
+	ch := &QQChannel{
+		BaseChannel: channels.NewBaseChannel("qq", nil, messageBus, []string{"legacy-member-id"}),
+		dedup:       make(map[string]time.Time),
+		done:        make(chan struct{}),
+		ctx:         context.Background(),
+	}
+
+	raw := []byte(`{"d":{"group_openid":"group-openid-abc","author":{"id":"legacy-member-id","member_openid":"member-openid-xyz"}}}`)
+	err := ch.handleGroupATMessage()(&dto.WSPayload{RawMessage: raw}, &dto.WSGroupATMessageData{
+		ID:      "group-allow-fallback",
+		GroupID: "legacy-group-id",
+		Content: "@bot hello",
+		Author: &dto.User{
+			ID: "legacy-member-id",
+		},
+	})
+	if err != nil {
+		t.Fatalf("handleGroupATMessage() error = %v", err)
+	}
+
+	inbound := waitInboundMessage(t, messageBus)
+	if inbound.SenderID != "member-openid-xyz" {
+		t.Fatalf("inbound.SenderID = %q, want %q", inbound.SenderID, "member-openid-xyz")
+	}
+	if inbound.Metadata["account_id"] != "member-openid-xyz" {
+		t.Fatalf("account_id metadata = %q, want %q", inbound.Metadata["account_id"], "member-openid-xyz")
+	}
+	if inbound.Metadata["legacy_account_id"] != "legacy-member-id" {
+		t.Fatalf("legacy_account_id metadata = %q, want %q", inbound.Metadata["legacy_account_id"], "legacy-member-id")
+	}
+}
+
+func TestParseQQOpenIDs(t *testing.T) {
+	tests := []struct {
+		name         string
+		raw          []byte
+		wantUser     string
+		wantMember   string
+		wantGroup    string
+	}{
+		{
+			name:       "empty",
+			raw:        nil,
+			wantUser:   "",
+			wantMember: "",
+			wantGroup:  "",
+		},
+		{
+			name:       "invalid json",
+			raw:        []byte("{not-json}"),
+			wantUser:   "",
+			wantMember: "",
+			wantGroup:  "",
+		},
+		{
+			name:       "all openids",
+			raw:        []byte(`{"d":{"group_openid":"group-1","author":{"user_openid":"user-1","member_openid":"member-1"}}}`),
+			wantUser:   "user-1",
+			wantMember: "member-1",
+			wantGroup:  "group-1",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			userOpenID, memberOpenID, groupOpenID := parseQQOpenIDs(tc.raw)
+			if userOpenID != tc.wantUser || memberOpenID != tc.wantMember || groupOpenID != tc.wantGroup {
+				t.Fatalf(
+					"parseQQOpenIDs(%s) = (%q, %q, %q), want (%q, %q, %q)",
+					tc.name,
+					userOpenID,
+					memberOpenID,
+					groupOpenID,
+					tc.wantUser,
+					tc.wantMember,
+					tc.wantGroup,
+				)
+			}
+		})
+	}
+}
+
 func TestHandleC2CMessage_AttachmentOnlyPublishesMedia(t *testing.T) {
 	messageBus := bus.NewMessageBus()
 	store := media.NewFileMediaStore()
@@ -682,7 +864,7 @@ func waitInboundMessage(t *testing.T, messageBus *bus.MessageBus) bus.InboundMes
 func writeTempFile(t *testing.T, dir, name string, content []byte) string {
 	t.Helper()
 
-	path := dir + "/" + name
+	path := filepath.Join(dir, name)
 	if err := os.WriteFile(path, content, 0o600); err != nil {
 		t.Fatalf("WriteFile() error = %v", err)
 	}