mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-08-01 13:33:48 +08:00
feat(go-channels): register qqbot channel (#17598)
## Summary - Register QQBot in the Go chat-channel runtime. - Ensure QQBot accounts can be built and started by the reconciler.
This commit is contained in:
Hz_
@@ -279,6 +279,8 @@ func buildChannel(accountID string, wanted desiredChannel) (core.Channel, error)
|
||||
switch wanted.channel {
|
||||
case "discord":
|
||||
return newDiscordChannelFromConfig(accountID, wanted.credential)
|
||||
case "qqbot":
|
||||
return newQQBotChannelFromConfig(accountID, wanted.credential)
|
||||
case "whatsapp":
|
||||
return newWhatsAppChannelFromConfig(accountID, wanted.credential)
|
||||
case "telegram":
|
||||
|
||||
@@ -15,3 +15,576 @@
|
||||
//
|
||||
|
||||
package channels
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"runtime/debug"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
|
||||
"ragflow/internal/channels/core"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultQQBotBaseURL = "https://api.sgroup.qq.com"
|
||||
defaultQQBotTokenURL = "https://bots.qq.com/app/getAppAccessToken"
|
||||
qqBotRequestTimeout = 60 * time.Second
|
||||
qqBotReconnectDelay = 3 * time.Second
|
||||
qqBotIntents = (1 << 30) | (1 << 12) | (1 << 25) | (1 << 26)
|
||||
)
|
||||
|
||||
type qqBotAccount struct {
|
||||
AccountID string
|
||||
AppID string
|
||||
ClientSecret string
|
||||
BaseURL string
|
||||
TokenURL string
|
||||
}
|
||||
|
||||
type qqBotChannel struct {
|
||||
account qqBotAccount
|
||||
client *http.Client
|
||||
dialer *websocket.Dialer
|
||||
|
||||
mu sync.Mutex
|
||||
writeMu sync.Mutex
|
||||
handler core.MessageHandler
|
||||
cancel context.CancelFunc
|
||||
done chan struct{}
|
||||
sessionID string
|
||||
sequence *int64
|
||||
}
|
||||
|
||||
type qqBotGatewayPayload struct {
|
||||
Op int `json:"op"`
|
||||
Data json.RawMessage `json:"d"`
|
||||
Sequence *int64 `json:"s"`
|
||||
Type string `json:"t"`
|
||||
}
|
||||
|
||||
type qqBotHello struct {
|
||||
HeartbeatInterval int `json:"heartbeat_interval"`
|
||||
}
|
||||
|
||||
func newQQBotChannel(account qqBotAccount) *qqBotChannel {
|
||||
if strings.TrimSpace(account.BaseURL) == "" {
|
||||
account.BaseURL = defaultQQBotBaseURL
|
||||
}
|
||||
if strings.TrimSpace(account.TokenURL) == "" {
|
||||
account.TokenURL = defaultQQBotTokenURL
|
||||
}
|
||||
return &qqBotChannel{
|
||||
account: account,
|
||||
client: &http.Client{Timeout: qqBotRequestTimeout},
|
||||
dialer: websocket.DefaultDialer,
|
||||
}
|
||||
}
|
||||
|
||||
func newQQBotChannelFromConfig(accountID string, cfg map[string]any) (*qqBotChannel, error) {
|
||||
appID := firstString(cfg, "app_id")
|
||||
clientSecret := firstString(cfg, "client_secret")
|
||||
if appID == "" || clientSecret == "" {
|
||||
return nil, fmt.Errorf("qqbot account %q is missing app_id or client_secret", accountID)
|
||||
}
|
||||
return newQQBotChannel(qqBotAccount{
|
||||
AccountID: accountID,
|
||||
AppID: appID,
|
||||
ClientSecret: clientSecret,
|
||||
BaseURL: firstString(cfg, "base_url"),
|
||||
}), nil
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) ChannelID() string {
|
||||
return "qqbot"
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) AccountID() string {
|
||||
return c.account.AccountID
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) SetMessageHandler(handler core.MessageHandler) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.handler = handler
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) Start(ctx context.Context) error {
|
||||
c.mu.Lock()
|
||||
if c.cancel != nil {
|
||||
c.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
runCtx, cancel := context.WithCancel(ctx)
|
||||
done := make(chan struct{})
|
||||
c.cancel = cancel
|
||||
c.done = done
|
||||
c.mu.Unlock()
|
||||
|
||||
go func() {
|
||||
c.run(runCtx)
|
||||
c.mu.Lock()
|
||||
if c.done == done {
|
||||
c.cancel = nil
|
||||
c.done = nil
|
||||
}
|
||||
c.mu.Unlock()
|
||||
close(done)
|
||||
}()
|
||||
log.Printf("[qqbot:%s] starting gateway client", c.account.AccountID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) Stop(ctx context.Context) error {
|
||||
c.mu.Lock()
|
||||
cancel := c.cancel
|
||||
done := c.done
|
||||
c.mu.Unlock()
|
||||
|
||||
if cancel == nil {
|
||||
return nil
|
||||
}
|
||||
cancel()
|
||||
select {
|
||||
case <-done:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) Send(ctx context.Context, msg core.OutgoingMessage) error {
|
||||
chatType, peerID := normalizeQQBotTarget(msg.ChatID)
|
||||
if peerID == "" {
|
||||
return errors.New("chat_id is required")
|
||||
}
|
||||
|
||||
token, err := c.getAccessToken(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var path string
|
||||
switch chatType {
|
||||
case "c2c":
|
||||
path = "/v2/users/" + url.PathEscape(peerID) + "/messages"
|
||||
case "group":
|
||||
path = "/v2/groups/" + url.PathEscape(peerID) + "/messages"
|
||||
case "dm":
|
||||
path = "/dms/" + url.PathEscape(peerID) + "/messages"
|
||||
case "channel":
|
||||
path = "/channels/" + url.PathEscape(peerID) + "/messages"
|
||||
default:
|
||||
return fmt.Errorf("unsupported outbound chat type %q", chatType)
|
||||
}
|
||||
body := map[string]any{
|
||||
"content": msg.Text,
|
||||
"msg_type": 0,
|
||||
"msg_seq": qqBotMessageSequence(),
|
||||
}
|
||||
if msg.ReplyToMessageID != "" {
|
||||
body["msg_id"] = msg.ReplyToMessageID
|
||||
}
|
||||
return c.requestJSON(ctx, token, http.MethodPost, path, body, true, false, nil)
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) run(ctx context.Context) {
|
||||
for ctx.Err() == nil {
|
||||
token, err := c.getAccessToken(ctx)
|
||||
if err == nil {
|
||||
var gatewayURL string
|
||||
gatewayURL, err = c.getGatewayURL(ctx, token)
|
||||
if err == nil {
|
||||
err = c.runGatewaySession(ctx, gatewayURL, token)
|
||||
}
|
||||
}
|
||||
if err != nil && ctx.Err() == nil {
|
||||
log.Printf("[qqbot:%s] gateway loop error: %v", c.account.AccountID, err)
|
||||
}
|
||||
if !waitForContext(ctx, qqBotReconnectDelay) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) runGatewaySession(ctx context.Context, gatewayURL string, token string) error {
|
||||
conn, _, err := c.dialer.DialContext(ctx, gatewayURL, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Close()
|
||||
log.Printf("[qqbot:%s] connected to gateway", c.account.AccountID)
|
||||
|
||||
sessionCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
go func() {
|
||||
<-sessionCtx.Done()
|
||||
_ = conn.Close()
|
||||
}()
|
||||
|
||||
var heartbeatCancel context.CancelFunc
|
||||
defer func() {
|
||||
if heartbeatCancel != nil {
|
||||
heartbeatCancel()
|
||||
}
|
||||
}()
|
||||
|
||||
for ctx.Err() == nil {
|
||||
_, data, err := conn.ReadMessage()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var payload qqBotGatewayPayload
|
||||
if err := json.Unmarshal(data, &payload); err != nil {
|
||||
log.Printf("[qqbot:%s] invalid gateway payload: %.200s", c.account.AccountID, data)
|
||||
continue
|
||||
}
|
||||
if payload.Sequence != nil {
|
||||
c.setSequence(*payload.Sequence)
|
||||
}
|
||||
|
||||
switch payload.Op {
|
||||
case 10:
|
||||
var hello qqBotHello
|
||||
if err := json.Unmarshal(payload.Data, &hello); err != nil {
|
||||
return fmt.Errorf("decode gateway hello: %w", err)
|
||||
}
|
||||
if err := c.sendIdentifyOrResume(conn, token); err != nil {
|
||||
return err
|
||||
}
|
||||
if heartbeatCancel != nil {
|
||||
heartbeatCancel()
|
||||
}
|
||||
heartbeatCtx, stopHeartbeat := context.WithCancel(sessionCtx)
|
||||
heartbeatCancel = stopHeartbeat
|
||||
go c.runHeartbeat(heartbeatCtx, conn, hello.HeartbeatInterval)
|
||||
case 11:
|
||||
continue
|
||||
case 7:
|
||||
return errors.New("gateway requested reconnect")
|
||||
case 9:
|
||||
var canResume bool
|
||||
_ = json.Unmarshal(payload.Data, &canResume)
|
||||
if !canResume {
|
||||
c.clearSession()
|
||||
if !waitForContext(ctx, qqBotReconnectDelay) {
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("invalid gateway session (can_resume=%t)", canResume)
|
||||
case 0:
|
||||
if err := c.handleDispatch(ctx, payload.Type, payload.Data); err != nil {
|
||||
log.Printf("[qqbot:%s] ignored invalid %s event: %v", c.account.AccountID, payload.Type, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) runHeartbeat(ctx context.Context, conn *websocket.Conn, intervalMS int) {
|
||||
delay := time.Duration(intervalMS) * time.Millisecond
|
||||
if delay < time.Second {
|
||||
delay = time.Second
|
||||
}
|
||||
ticker := time.NewTicker(delay)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
payload := map[string]any{"op": 1, "d": c.currentSequence()}
|
||||
if err := c.writeJSON(conn, payload); err != nil {
|
||||
log.Printf("[qqbot:%s] heartbeat failed: %v", c.account.AccountID, err)
|
||||
_ = conn.Close()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) sendIdentifyOrResume(conn *websocket.Conn, token string) error {
|
||||
c.mu.Lock()
|
||||
sessionID := c.sessionID
|
||||
var sequence *int64
|
||||
if c.sequence != nil {
|
||||
value := *c.sequence
|
||||
sequence = &value
|
||||
}
|
||||
c.mu.Unlock()
|
||||
|
||||
if sessionID != "" && sequence != nil {
|
||||
return c.writeJSON(conn, map[string]any{
|
||||
"op": 6,
|
||||
"d": map[string]any{
|
||||
"token": "QQBot " + token,
|
||||
"session_id": sessionID,
|
||||
"seq": *sequence,
|
||||
},
|
||||
})
|
||||
}
|
||||
return c.writeJSON(conn, map[string]any{
|
||||
"op": 2,
|
||||
"d": map[string]any{
|
||||
"token": "QQBot " + token,
|
||||
"intents": qqBotIntents,
|
||||
"shard": []int{0, 1},
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) handleDispatch(ctx context.Context, eventType string, data json.RawMessage) error {
|
||||
var raw map[string]any
|
||||
if err := json.Unmarshal(data, &raw); err != nil {
|
||||
return err
|
||||
}
|
||||
if eventType == "READY" {
|
||||
c.mu.Lock()
|
||||
c.sessionID = stringValue(raw["session_id"])
|
||||
c.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
if eventType == "RESUMED" {
|
||||
return nil
|
||||
}
|
||||
incoming := c.normalizeIncomingEvent(eventType, raw)
|
||||
if incoming == nil {
|
||||
return nil
|
||||
}
|
||||
c.mu.Lock()
|
||||
handler := c.handler
|
||||
c.mu.Unlock()
|
||||
if handler != nil {
|
||||
go func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("[qqbot:%s] handler panic: %v\n%s", c.account.AccountID, r, debug.Stack())
|
||||
}
|
||||
}()
|
||||
if err := handler(ctx, *incoming); err != nil {
|
||||
log.Printf("[qqbot:%s] handler error: %v", c.account.AccountID, err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) normalizeIncomingEvent(eventType string, data map[string]any) *core.IncomingMessage {
|
||||
author, _ := data["author"].(map[string]any)
|
||||
message := &core.IncomingMessage{
|
||||
Channel: c.ChannelID(),
|
||||
AccountID: c.AccountID(),
|
||||
MessageID: stringValue(data["id"]),
|
||||
Text: stringValue(data["content"]),
|
||||
Raw: data,
|
||||
}
|
||||
switch eventType {
|
||||
case "C2C_MESSAGE_CREATE":
|
||||
message.SenderID = stringValue(author["user_openid"])
|
||||
if message.SenderID == "" {
|
||||
return nil
|
||||
}
|
||||
message.ChatID = incomingQQBotChatID("c2c", message.SenderID)
|
||||
message.ChatType = "p2p"
|
||||
case "GROUP_AT_MESSAGE_CREATE", "GROUP_MESSAGE_CREATE":
|
||||
groupID := stringValue(data["group_openid"])
|
||||
message.SenderID = stringValue(author["member_openid"])
|
||||
if groupID == "" || message.SenderID == "" {
|
||||
return nil
|
||||
}
|
||||
message.ChatID = incomingQQBotChatID("group", groupID)
|
||||
message.ChatType = "group"
|
||||
case "DIRECT_MESSAGE_CREATE":
|
||||
message.SenderID = stringValue(author["id"])
|
||||
if message.SenderID == "" {
|
||||
return nil
|
||||
}
|
||||
chatID := stringValue(data["guild_id"])
|
||||
if chatID == "" {
|
||||
chatID = message.SenderID
|
||||
}
|
||||
message.ChatID = incomingQQBotChatID("dm", chatID)
|
||||
message.ChatType = "dm"
|
||||
case "AT_MESSAGE_CREATE":
|
||||
message.SenderID = stringValue(author["id"])
|
||||
channelID := stringValue(data["channel_id"])
|
||||
if message.SenderID == "" || channelID == "" {
|
||||
return nil
|
||||
}
|
||||
message.ChatID = incomingQQBotChatID("channel", channelID)
|
||||
message.ChatType = "channel"
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
return message
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) getAccessToken(ctx context.Context) (string, error) {
|
||||
var response map[string]any
|
||||
err := c.requestJSON(ctx, "", http.MethodPost, c.account.TokenURL, map[string]any{
|
||||
"appId": c.account.AppID,
|
||||
"clientSecret": c.account.ClientSecret,
|
||||
}, false, true, &response)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
token := stringValue(response["access_token"])
|
||||
if token == "" {
|
||||
token = stringValue(response["accessToken"])
|
||||
}
|
||||
if token == "" {
|
||||
return "", fmt.Errorf("qqbot token response missing access_token: %v", response)
|
||||
}
|
||||
return token, nil
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) getGatewayURL(ctx context.Context, token string) (string, error) {
|
||||
var response map[string]any
|
||||
if err := c.requestJSON(ctx, token, http.MethodGet, "/gateway", nil, true, false, &response); err != nil {
|
||||
return "", err
|
||||
}
|
||||
gatewayURL := stringValue(response["url"])
|
||||
if gatewayURL == "" {
|
||||
return "", fmt.Errorf("qqbot gateway response missing url: %v", response)
|
||||
}
|
||||
return gatewayURL, nil
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) requestJSON(
|
||||
ctx context.Context,
|
||||
accessToken string,
|
||||
method string,
|
||||
path string,
|
||||
body map[string]any,
|
||||
auth bool,
|
||||
absoluteURL bool,
|
||||
out any,
|
||||
) error {
|
||||
requestURL := path
|
||||
if !absoluteURL {
|
||||
requestURL = strings.TrimRight(c.account.BaseURL, "/") + path
|
||||
}
|
||||
var reader io.Reader
|
||||
if body != nil {
|
||||
payload, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
reader = bytes.NewReader(payload)
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, method, requestURL, reader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if auth && accessToken != "" {
|
||||
req.Header.Set("Authorization", "QQBot "+accessToken)
|
||||
}
|
||||
resp, err := c.client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
payload, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if resp.StatusCode >= http.StatusBadRequest {
|
||||
return fmt.Errorf("status: %d, response: %s", resp.StatusCode, payload)
|
||||
}
|
||||
if len(bytes.TrimSpace(payload)) == 0 {
|
||||
return nil
|
||||
}
|
||||
if out == nil {
|
||||
out = &map[string]any{}
|
||||
}
|
||||
if err := json.Unmarshal(payload, out); err != nil {
|
||||
return fmt.Errorf("invalid json response from %s: %.200s: %w", path, payload, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) writeJSON(conn *websocket.Conn, value any) error {
|
||||
c.writeMu.Lock()
|
||||
defer c.writeMu.Unlock()
|
||||
return conn.WriteJSON(value)
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) setSequence(sequence int64) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.sequence = &sequence
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) currentSequence() any {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if c.sequence == nil {
|
||||
return nil
|
||||
}
|
||||
return *c.sequence
|
||||
}
|
||||
|
||||
func (c *qqBotChannel) clearSession() {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.sessionID = ""
|
||||
c.sequence = nil
|
||||
}
|
||||
|
||||
func normalizeQQBotTarget(chatID string) (string, string) {
|
||||
raw := strings.TrimSpace(chatID)
|
||||
raw = strings.TrimPrefix(raw, "qqbot:")
|
||||
for _, chatType := range []string{"group", "channel", "dm", "c2c"} {
|
||||
prefix := chatType + ":"
|
||||
if strings.HasPrefix(raw, prefix) {
|
||||
return chatType, strings.TrimPrefix(raw, prefix)
|
||||
}
|
||||
}
|
||||
return "c2c", raw
|
||||
}
|
||||
|
||||
func incomingQQBotChatID(chatType string, peerID string) string {
|
||||
return "qqbot:" + chatType + ":" + peerID
|
||||
}
|
||||
|
||||
func qqBotMessageSequence() int {
|
||||
sequence := int(time.Now().UnixMilli() % 65535)
|
||||
if sequence != 0 {
|
||||
return sequence
|
||||
}
|
||||
var value uint16
|
||||
if err := binary.Read(rand.Reader, binary.BigEndian, &value); err != nil || value == 0 {
|
||||
return 1
|
||||
}
|
||||
return int(value)
|
||||
}
|
||||
|
||||
func stringValue(value any) string {
|
||||
if value == nil {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprint(value)
|
||||
}
|
||||
|
||||
func waitForContext(ctx context.Context, delay time.Duration) bool {
|
||||
timer := time.NewTimer(delay)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-timer.C:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
340
internal/channels/qqbot_test.go
Normal file
340
internal/channels/qqbot_test.go
Normal file
@@ -0,0 +1,340 @@
|
||||
//
|
||||
// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
|
||||
package channels
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
|
||||
"ragflow/internal/channels/core"
|
||||
)
|
||||
|
||||
func TestNewQQBotChannelFromConfig(t *testing.T) {
|
||||
t.Run("valid", func(t *testing.T) {
|
||||
channel, err := newQQBotChannelFromConfig("account-1", map[string]any{
|
||||
"app_id": "app-1",
|
||||
"client_secret": "secret-1",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("newQQBotChannelFromConfig() error = %v", err)
|
||||
}
|
||||
if channel.account.BaseURL != defaultQQBotBaseURL {
|
||||
t.Fatalf("BaseURL = %q, want %q", channel.account.BaseURL, defaultQQBotBaseURL)
|
||||
}
|
||||
if channel.ChannelID() != "qqbot" || channel.AccountID() != "account-1" {
|
||||
t.Fatalf("channel identity = %s:%s, want qqbot:account-1", channel.ChannelID(), channel.AccountID())
|
||||
}
|
||||
})
|
||||
|
||||
for _, cfg := range []map[string]any{
|
||||
{"client_secret": "secret-1"},
|
||||
{"app_id": "app-1"},
|
||||
} {
|
||||
if _, err := newQQBotChannelFromConfig("account-1", cfg); err == nil {
|
||||
t.Fatalf("newQQBotChannelFromConfig(%v) succeeded, want credential error", cfg)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildChannelSupportsQQBot(t *testing.T) {
|
||||
channel, err := buildChannel("account-1", desiredChannel{
|
||||
channel: "qqbot",
|
||||
credential: map[string]any{
|
||||
"app_id": "app-1",
|
||||
"client_secret": "secret-1",
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("buildChannel() error = %v", err)
|
||||
}
|
||||
if channel.ChannelID() != "qqbot" {
|
||||
t.Fatalf("ChannelID() = %q, want qqbot", channel.ChannelID())
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeQQBotTarget(t *testing.T) {
|
||||
tests := []struct {
|
||||
chatID string
|
||||
wantType string
|
||||
wantID string
|
||||
}{
|
||||
{chatID: "user-1", wantType: "c2c", wantID: "user-1"},
|
||||
{chatID: "c2c:user-1", wantType: "c2c", wantID: "user-1"},
|
||||
{chatID: "qqbot:c2c:user-1", wantType: "c2c", wantID: "user-1"},
|
||||
{chatID: "qqbot:group:group-1", wantType: "group", wantID: "group-1"},
|
||||
{chatID: "qqbot:dm:guild-1", wantType: "dm", wantID: "guild-1"},
|
||||
{chatID: "qqbot:channel:channel-1", wantType: "channel", wantID: "channel-1"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.chatID, func(t *testing.T) {
|
||||
gotType, gotID := normalizeQQBotTarget(tt.chatID)
|
||||
if gotType != tt.wantType || gotID != tt.wantID {
|
||||
t.Fatalf("normalizeQQBotTarget(%q) = (%q, %q), want (%q, %q)", tt.chatID, gotType, gotID, tt.wantType, tt.wantID)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestQQBotSend(t *testing.T) {
|
||||
var requests []struct {
|
||||
path string
|
||||
authorization string
|
||||
body map[string]any
|
||||
}
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path == "/token" {
|
||||
var body map[string]any
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||||
t.Errorf("decode token request: %v", err)
|
||||
}
|
||||
if body["appId"] != "app-1" || body["clientSecret"] != "secret-1" {
|
||||
t.Errorf("token request body = %v", body)
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{"access_token": "token-1"})
|
||||
return
|
||||
}
|
||||
var body map[string]any
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||||
t.Errorf("decode send request: %v", err)
|
||||
}
|
||||
requests = append(requests, struct {
|
||||
path string
|
||||
authorization string
|
||||
body map[string]any
|
||||
}{r.URL.Path, r.Header.Get("Authorization"), body})
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{})
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
channel := newQQBotChannel(qqBotAccount{
|
||||
AccountID: "account-1",
|
||||
AppID: "app-1",
|
||||
ClientSecret: "secret-1",
|
||||
BaseURL: server.URL,
|
||||
TokenURL: server.URL + "/token",
|
||||
})
|
||||
tests := []struct {
|
||||
chatID string
|
||||
wantPath string
|
||||
}{
|
||||
{chatID: "qqbot:c2c:user-1", wantPath: "/v2/users/user-1/messages"},
|
||||
{chatID: "qqbot:group:group-1", wantPath: "/v2/groups/group-1/messages"},
|
||||
{chatID: "qqbot:dm:guild-1", wantPath: "/dms/guild-1/messages"},
|
||||
{chatID: "qqbot:channel:channel-1", wantPath: "/channels/channel-1/messages"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
if err := channel.Send(context.Background(), core.OutgoingMessage{
|
||||
ChatID: tt.chatID,
|
||||
Text: "answer",
|
||||
ReplyToMessageID: "message-1",
|
||||
}); err != nil {
|
||||
t.Fatalf("Send(%q) error = %v", tt.chatID, err)
|
||||
}
|
||||
request := requests[len(requests)-1]
|
||||
if request.path != tt.wantPath {
|
||||
t.Errorf("Send(%q) path = %q, want %q", tt.chatID, request.path, tt.wantPath)
|
||||
}
|
||||
if request.authorization != "QQBot token-1" {
|
||||
t.Errorf("Authorization = %q, want QQBot token-1", request.authorization)
|
||||
}
|
||||
if request.body["content"] != "answer" || request.body["msg_id"] != "message-1" || request.body["msg_type"] != float64(0) {
|
||||
t.Errorf("send body = %v", request.body)
|
||||
}
|
||||
if sequence, ok := request.body["msg_seq"].(float64); !ok || sequence < 1 || sequence > 65535 {
|
||||
t.Errorf("msg_seq = %v, want integer in [1, 65535]", request.body["msg_seq"])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestQQBotNormalizeIncomingEvent(t *testing.T) {
|
||||
channel := newQQBotChannel(qqBotAccount{AccountID: "account-1"})
|
||||
tests := []struct {
|
||||
name string
|
||||
eventType string
|
||||
data map[string]any
|
||||
wantChatID string
|
||||
wantType string
|
||||
wantSender string
|
||||
}{
|
||||
{
|
||||
name: "c2c",
|
||||
eventType: "C2C_MESSAGE_CREATE",
|
||||
data: map[string]any{"id": "m1", "content": "hello", "author": map[string]any{"user_openid": "user-1"}},
|
||||
wantChatID: "qqbot:c2c:user-1",
|
||||
wantType: "p2p",
|
||||
wantSender: "user-1",
|
||||
},
|
||||
{
|
||||
name: "group",
|
||||
eventType: "GROUP_AT_MESSAGE_CREATE",
|
||||
data: map[string]any{"id": "m2", "group_openid": "group-1", "author": map[string]any{"member_openid": "member-1"}},
|
||||
wantChatID: "qqbot:group:group-1",
|
||||
wantType: "group",
|
||||
wantSender: "member-1",
|
||||
},
|
||||
{
|
||||
name: "direct message uses guild",
|
||||
eventType: "DIRECT_MESSAGE_CREATE",
|
||||
data: map[string]any{"id": "m3", "guild_id": "guild-1", "author": map[string]any{"id": "user-2"}},
|
||||
wantChatID: "qqbot:dm:guild-1",
|
||||
wantType: "dm",
|
||||
wantSender: "user-2",
|
||||
},
|
||||
{
|
||||
name: "channel",
|
||||
eventType: "AT_MESSAGE_CREATE",
|
||||
data: map[string]any{"id": "m4", "channel_id": "channel-1", "author": map[string]any{"id": "user-3"}},
|
||||
wantChatID: "qqbot:channel:channel-1",
|
||||
wantType: "channel",
|
||||
wantSender: "user-3",
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
message := channel.normalizeIncomingEvent(tt.eventType, tt.data)
|
||||
if message == nil {
|
||||
t.Fatal("normalizeIncomingEvent() = nil")
|
||||
}
|
||||
if message.ChatID != tt.wantChatID || message.ChatType != tt.wantType || message.SenderID != tt.wantSender {
|
||||
t.Fatalf("message = %+v, want chat_id=%q chat_type=%q sender_id=%q", message, tt.wantChatID, tt.wantType, tt.wantSender)
|
||||
}
|
||||
if message.Channel != "qqbot" || message.AccountID != "account-1" || message.Raw == nil {
|
||||
t.Fatalf("message envelope = %+v", message)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestQQBotGatewayIdentifiesAndDispatches(t *testing.T) {
|
||||
upgrader := websocket.Upgrader{}
|
||||
identify := make(chan map[string]any, 1)
|
||||
var server *httptest.Server
|
||||
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
conn, err := upgrader.Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
t.Errorf("upgrade websocket: %v", err)
|
||||
return
|
||||
}
|
||||
defer conn.Close()
|
||||
if err := conn.WriteJSON(map[string]any{"op": 10, "d": map[string]any{"heartbeat_interval": 60000}}); err != nil {
|
||||
t.Errorf("write hello: %v", err)
|
||||
return
|
||||
}
|
||||
var auth map[string]any
|
||||
if err := conn.ReadJSON(&auth); err != nil {
|
||||
t.Errorf("read identify: %v", err)
|
||||
return
|
||||
}
|
||||
identify <- auth
|
||||
_ = conn.WriteJSON(map[string]any{"op": 0, "s": 1, "t": "READY", "d": map[string]any{"session_id": "session-1"}})
|
||||
_ = conn.WriteJSON(map[string]any{
|
||||
"op": 0, "s": 2, "t": "C2C_MESSAGE_CREATE",
|
||||
"d": map[string]any{"id": "message-1", "content": "hello", "author": map[string]any{"user_openid": "user-1"}},
|
||||
})
|
||||
_ = conn.WriteJSON(map[string]any{"op": 7})
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
channel := newQQBotChannel(qqBotAccount{AccountID: "account-1"})
|
||||
incoming := make(chan core.IncomingMessage, 1)
|
||||
channel.SetMessageHandler(func(_ context.Context, message core.IncomingMessage) error {
|
||||
incoming <- message
|
||||
return nil
|
||||
})
|
||||
gatewayURL := "ws" + strings.TrimPrefix(server.URL, "http")
|
||||
if err := channel.runGatewaySession(context.Background(), gatewayURL, "token-1"); err == nil || !strings.Contains(err.Error(), "reconnect") {
|
||||
t.Fatalf("runGatewaySession() error = %v, want reconnect request", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case auth := <-identify:
|
||||
if auth["op"] != float64(2) {
|
||||
t.Fatalf("identify op = %v, want 2", auth["op"])
|
||||
}
|
||||
data, _ := auth["d"].(map[string]any)
|
||||
if data["token"] != "QQBot token-1" || data["intents"] != float64(qqBotIntents) {
|
||||
t.Fatalf("identify data = %v", data)
|
||||
}
|
||||
default:
|
||||
t.Fatal("gateway did not receive identify payload")
|
||||
}
|
||||
select {
|
||||
case message := <-incoming:
|
||||
if message.ChatID != "qqbot:c2c:user-1" || message.MessageID != "message-1" || message.Text != "hello" {
|
||||
t.Fatalf("incoming message = %+v", message)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("timed out waiting for inbound message")
|
||||
}
|
||||
channel.mu.Lock()
|
||||
sessionID := channel.sessionID
|
||||
sequence := channel.sequence
|
||||
channel.mu.Unlock()
|
||||
if sessionID != "session-1" || sequence == nil || *sequence != 2 {
|
||||
t.Fatalf("gateway state = session %q sequence %v", sessionID, sequence)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQQBotGatewayResumesExistingSession(t *testing.T) {
|
||||
upgrader := websocket.Upgrader{}
|
||||
resume := make(chan map[string]any, 1)
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
conn, err := upgrader.Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
t.Errorf("upgrade websocket: %v", err)
|
||||
return
|
||||
}
|
||||
defer conn.Close()
|
||||
_ = conn.WriteJSON(map[string]any{"op": 10, "d": map[string]any{"heartbeat_interval": 60000}})
|
||||
var auth map[string]any
|
||||
if err := conn.ReadJSON(&auth); err != nil {
|
||||
t.Errorf("read resume: %v", err)
|
||||
return
|
||||
}
|
||||
resume <- auth
|
||||
_ = conn.WriteJSON(map[string]any{"op": 7})
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
channel := newQQBotChannel(qqBotAccount{AccountID: "account-1"})
|
||||
channel.sessionID = "session-1"
|
||||
channel.setSequence(42)
|
||||
gatewayURL := "ws" + strings.TrimPrefix(server.URL, "http")
|
||||
if err := channel.runGatewaySession(context.Background(), gatewayURL, "token-1"); err == nil {
|
||||
t.Fatal("runGatewaySession() succeeded, want reconnect request")
|
||||
}
|
||||
|
||||
select {
|
||||
case auth := <-resume:
|
||||
if auth["op"] != float64(6) {
|
||||
t.Fatalf("resume op = %v, want 6", auth["op"])
|
||||
}
|
||||
data, _ := auth["d"].(map[string]any)
|
||||
if data["token"] != "QQBot token-1" || data["session_id"] != "session-1" || data["seq"] != float64(42) {
|
||||
t.Fatalf("resume data = %v", data)
|
||||
}
|
||||
default:
|
||||
t.Fatal("gateway did not receive resume payload")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user