diff --git a/README.md b/README.md index 108133a..27866a7 100644 --- a/README.md +++ b/README.md @@ -77,9 +77,9 @@ nukumizu-backend/ │ ├── email.go # Email notification pipe │ ├── ntfy.go # ntfy notification pipe │ ├── webhook.go # Outgoing webhook notification pipe -│ ├── qq_napcat/ +│ ├── qq_onebot/ │ │ ├── qq.go # QQ (NapCat / OneBot 11) bot controller -│ │ └── napcat.go # NapCat WebSocket + HTTP API client +│ │ └── onebot.go # OneBot 11 WebSocket client (events + API) │ └── telegram/ │ ├── telegram.go # Telegram bot controller (go-telegram/bot, long polling) │ └── send.go # Message sending / splitting (Telegram Markdown) @@ -101,7 +101,7 @@ nukumizu-backend/ - Go **1.25** or newer - [Node.js](https://nodejs.org) **22** or newer — only needed to build the web console; a prebuilt binary does not require it - A running [Komari](https://www.komari.wiki) Dashboard instance reachable from this host -- For QQ: a [NapCat](https://napneko.github.io/) instance exposing an OneBot 11 WebSocket + HTTP endpoint +- For QQ: a [NapCat](https://napneko.github.io/) instance exposing an OneBot 11 forward WebSocket endpoint (`/api` and `/event`) - For Telegram: a bot token from [@BotFather](https://t.me/BotFather) Key dependencies: `github.com/go-telegram/bot`, `github.com/gorilla/websocket`, `gopkg.in/mail.v2`, `modernc.org/sqlite`. @@ -218,7 +218,7 @@ There are two configuration files, both read from the working directory unless o Field notes: -- `system.networkProxy` is a **system-wide** proxy URL. A controller only uses it when its own `networkUseProxy` is `true`. Applied to Telegram HTTP polling, NapCat HTTP/WebSocket, ntfy and webhook requests, and Email SMTP (tunneled via HTTP CONNECT). +- `system.networkProxy` is a **system-wide** proxy URL. A controller only uses it when its own `networkUseProxy` is `true`. Applied to Telegram HTTP polling, QQ OneBot WebSocket, ntfy and webhook requests, and Email SMTP (tunneled via HTTP CONNECT). - `webhook` configures the **incoming** webhook API (see [Incoming webhook API](#incoming-webhook-api)); `controllerMethod.webhook` configures the outgoing webhook notification channel. They are independent. - `markdown` is a per-channel switch on all five channels. With it `false` (the default) every rendered value is inserted as plain text; with it `true` the values meant to be read verbatim (UUIDs, event messages, commands, command results, alert source and alert content) are wrapped in Markdown code spans / fenced blocks. Nothing is inferred from the channel name, so a channel only ever gets the formatting you asked for — turn it off for a channel whose platform does not render Markdown. On Telegram it also picks the `parse_mode`: with `markdown` off, messages are sent without one, so text containing `*` or `_` is delivered as-is rather than rejected by the API as malformed Markdown. - `controllerMethod.qq(napcat).listenMethod` / `telegram.listenMethod` — see [Bot recognition modes](#bot-recognition-modes). @@ -387,7 +387,7 @@ Status codes: `200` delivered, `400` malformed body or empty `subject`/`content` ## Bots -QQ (NapCat) and Telegram bots share one command engine and authorization pipeline, implemented in `internal/controller/`. NapCat speaks OneBot 11 (WebSocket event stream + HTTP actions); Telegram uses `go-telegram/bot` long polling. +QQ (NapCat) and Telegram bots share one command engine and authorization pipeline, implemented in `internal/controller/`. NapCat speaks OneBot 11 over WebSocket (event stream + actions); Telegram uses `go-telegram/bot` long polling. ### Permission model diff --git a/config/variables.go b/config/variables.go index db0410f..29277d9 100644 --- a/config/variables.go +++ b/config/variables.go @@ -232,7 +232,7 @@ type BotUser_TelegramConfig struct { // and trusted groups (chat targets) for each bot channel separately from the // main config.json. type BotUserConfig struct { - QQ BotUser_QQConfig `json:"qq(onebot)"` + QQ BotUser_QQConfig `json:"qq_onebot"` Telegram BotUser_TelegramConfig `json:"telegram"` } diff --git a/internal/controller/pipes/qq_onebot/onebot.go b/internal/controller/pipes/qq_onebot/onebot.go index 561e7d2..8b1affd 100644 --- a/internal/controller/pipes/qq_onebot/onebot.go +++ b/internal/controller/pipes/qq_onebot/onebot.go @@ -1,13 +1,13 @@ package qq_onebot import ( - "bytes" "encoding/json" "fmt" - "io" "net/http" "net/url" + "strconv" "sync" + "sync/atomic" "time" "github.com/gorilla/websocket" @@ -17,113 +17,136 @@ import ( "nukumizu-backend/postLog" ) -// actionLogEnabled reports whether the OneBot HTTP API methods echo each action -// they send, per the showOneBotAction toggle. Unlike the WebSocket message path -// in qq.go it deliberately does not also require debugMode: these methods have +const ( + // onebotReconnectDelay is the pause between reconnect attempts, shared by the + // event socket and the API socket. + onebotReconnectDelay = 5 * time.Second + // onebotAPITimeout bounds one callAPI response wait. It matches the timeout the + // removed HTTP client (netproxy.HTTPClient(useProxy, 30s)) imposed, so callers + // see the same worst-case latency they always did. + onebotAPITimeout = 30 * time.Second + // onebotConnectWait bounds how long callAPI waits for the API socket to come up + // when it is not connected yet. It is deliberately short: the startup bot + // initialization message is sent on the main goroutine, and a down OneBot + // server must not stall HTTP-server startup for the full API timeout. + onebotConnectWait = 5 * time.Second + // onebotWriteTimeout bounds a single socket write so a wedged peer cannot block + // the caller (and, through writeMu, every other writer) indefinitely. + onebotWriteTimeout = 10 * time.Second +) + +// actionLogEnabled reports whether the OneBot API methods echo each action they +// send, per the showOneBotAction toggle. Unlike the WebSocket message path in +// qq.go it deliberately does not also require debugMode: these methods have // always logged on this toggle alone, and preserving that is intentional. func actionLogEnabled() bool { cfg := config.Current() return cfg != nil && cfg.Debug.ShowOneBotAction } -// APIResponse mirrors OneBot's HTTP API response envelope. +// APIResponse mirrors OneBot's API response envelope. Echo is the correlation id +// the client sent, echoed back by the server; callAPI uses it internally to match +// a response to its request. Data stays raw so callers decode get_group_list / +// get_friend_list arrays themselves. type APIResponse struct { Status string `json:"status"` RetCode int `json:"retcode"` Message string `json:"message"` Data json.RawMessage `json:"data"` + Echo json.RawMessage `json:"echo"` } -// Client is the HTTP + WebSocket client for a single OneBot instance. -// It both listens for incoming OneBot events over WebSocket and issues -// outbound OneBot API calls over HTTP. -type Client struct { - addr string - port string - token string - useProxy bool - httpClient *http.Client +// apiRequest is one OneBot API call written to the API socket. A struct rather +// than a map guarantees Echo is encoded as a JSON string, which is what the +// server sends back verbatim. +type apiRequest struct { + Action string `json:"action"` + Params map[string]interface{} `json:"params"` + Echo string `json:"echo"` +} + +// apiResult carries one decoded API response (or a transport error) from the API +// read loop back to the callAPI invocation waiting on its echo. +type apiResult struct { + resp *APIResponse + err error +} + +// Client is the WebSocket client for a single OneBot instance. It uses two +// forward-WebSocket connections, per the OneBot 11 split between events and API +// calls: +// +// - the event socket ("/") only reads: one goroutine feeds every frame to +// onEvent. Nothing is ever written to it. +// - the API socket ("/api") only issues actions: callAPI writes a +// {"action","params","echo"} frame there, and its dedicated read loop matches +// the reply by echo. +// +// They must be separate: qq.go's event handler replies with SendMsg from inside +// the event read loop, so reading API responses on that same socket would +// deadlock (the loop would be blocked in the handler, waiting on a response only +// it could deliver). +type Client struct { + addr string + port string + token string + useProxy bool + + // Event socket ("/"): one reader (listenOnce), closed only by Stop or its own + // cleanup, never written to. + connMu sync.Mutex + conn *websocket.Conn + + // API socket ("/api"). apiMu guards the pointer, the epoch channel and the + // pending table; writeMu serializes the actual socket writes, so apiMu is + // never held across I/O (which would stall response dispatch behind a slow + // write). + apiMu sync.Mutex + apiConn *websocket.Conn + apiChanged chan struct{} // closed and replaced on every connect/disconnect + writeMu sync.Mutex + pending map[string]chan apiResult + echoSeq atomic.Uint64 + stopped bool - connMu sync.Mutex - conn *websocket.Conn stopCh chan struct{} stopOnce sync.Once } -// NewClient creates a OneBot client for the given host/port/token. When -// useProxy is set, HTTP API calls and the WebSocket connection are routed -// through the system-wide network proxy. +// NewClient creates a OneBot client for the given host/port/token. When useProxy +// is set, both WebSocket connections are routed through the system-wide network +// proxy. It performs no I/O; Listen starts the sockets. func NewClient(addr, port, token string, useProxy bool) *Client { return &Client{ addr: addr, port: port, token: token, useProxy: useProxy, - httpClient: netproxy.HTTPClient(useProxy, 30*time.Second), + apiChanged: make(chan struct{}), + pending: make(map[string]chan apiResult), stopCh: make(chan struct{}), } } -func (c *Client) baseURL() string { - return fmt.Sprintf("http://%s:%s", c.addr, c.port) -} - -// sendRequest performs an HTTP call to OneBot, injecting access_token into POST -// bodies AND setting Authorization: Bearer (compatible with HTTP Server adapters -// created in the OneBot Web UI). -func (c *Client) sendRequest(method, endpoint string, body []byte) ([]byte, error) { - url := c.baseURL() + endpoint - - // For POST requests, inject access_token into the JSON body before sending. - var finalBody []byte - if body != nil && c.token != "" { - var bodyMap map[string]interface{} - if err := json.Unmarshal(body, &bodyMap); err == nil { - bodyMap["access_token"] = c.token - finalBody, _ = json.Marshal(bodyMap) - } - } - if finalBody == nil { - finalBody = body - } - - var req *http.Request - var err error - if finalBody != nil { - req, err = http.NewRequest(method, url, bytes.NewReader(finalBody)) - } else { - req, err = http.NewRequest(method, url, nil) - } - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } - - req.Header.Set("Content-Type", "application/json") +// dialWS opens one forward-WebSocket connection to the given path. The token is +// passed both as an access_token query parameter and as an Authorization: Bearer +// header (forward adapters check one or the other), and the dial goes through the +// system proxy when useProxy is set. +func (c *Client) dialWS(path string) (*websocket.Conn, error) { + wsURL := fmt.Sprintf("ws://%s:%s%s", c.addr, c.port, path) + header := http.Header{} if c.token != "" { - req.Header.Set("Authorization", "Bearer "+c.token) + wsURL += "?access_token=" + url.QueryEscape(c.token) + header.Set("Authorization", "Bearer "+c.token) } - resp, err := c.httpClient.Do(req) + dialer := *websocket.DefaultDialer + dialer.Proxy = netproxy.ProxyFunc(c.useProxy) + conn, _, err := dialer.Dial(wsURL, header) if err != nil { - return nil, fmt.Errorf("failed to send request to onebot: %w", err) + return nil, fmt.Errorf("failed to connect to onebot websocket %s: %w", path, err) } - defer resp.Body.Close() - - respBody, err := io.ReadAll(resp.Body) - if err != nil { - return nil, fmt.Errorf("failed to read onebot response: %w", err) - } - - return respBody, nil -} - -// parseResponse unmarshals OneBot's raw response body into an APIResponse. -func (c *Client) parseResponse(respBody []byte, caller string) (*APIResponse, error) { - var onebotResp APIResponse - if err := json.Unmarshal(respBody, &onebotResp); err != nil { - return nil, fmt.Errorf("failed to unmarshal onebot response (%s): %w\n%s", caller, err, string(respBody)) - } - return &onebotResp, nil + return conn, nil } // SendMsg sends a message via OneBot. @@ -139,19 +162,19 @@ func (c *Client) SendMsg(targetType string, targetID int64, msg string, hasAt bo message = fmt.Sprintf("[CQ:at,qq=%d] %s", atTargetID, msg) } - var endpoint string - var onebotReq map[string]interface{} + var action string + var params map[string]interface{} switch targetType { case "group": - endpoint = "/send_group_msg" - onebotReq = map[string]interface{}{ + action = "send_group_msg" + params = map[string]interface{}{ "group_id": targetID, "message": message, } case "private": - endpoint = "/send_private_msg" - onebotReq = map[string]interface{}{ + action = "send_private_msg" + params = map[string]interface{}{ "user_id": targetID, "message": message, } @@ -159,97 +182,164 @@ func (c *Client) SendMsg(targetType string, targetID int64, msg string, hasAt bo return nil, fmt.Errorf("invalid targetType: %s, must be 'group' or 'private'", targetType) } - reqBody, err := json.Marshal(onebotReq) - if err != nil { - return nil, fmt.Errorf("failed to marshal request: %w", err) - } - if actionLogEnabled() { - postLog.Debug(fmt.Sprintf("[qq_onebot] SendMsg -> %s (%s): %s", endpoint, targetType, message)) + postLog.Debug(fmt.Sprintf("[qq_onebot] SendMsg -> %s (%s): %s", action, targetType, message)) } - respBody, err := c.sendRequest(http.MethodPost, endpoint, reqBody) - if err != nil { - return nil, err - } - - return c.parseResponse(respBody, "SendMsg") + return c.callAPI(action, params) } // RecallMsg recalls (deletes) a sent message via OneBot. func (c *Client) RecallMsg(msgID int64) (*APIResponse, error) { - reqBody, err := json.Marshal(map[string]interface{}{ - "message_id": msgID, - }) - if err != nil { - return nil, fmt.Errorf("failed to marshal request: %w", err) - } - if actionLogEnabled() { - postLog.Debug(fmt.Sprintf("[qq_onebot] RecallMsg -> /delete_msg: %d", msgID)) + postLog.Debug(fmt.Sprintf("[qq_onebot] RecallMsg -> delete_msg: %d", msgID)) } - respBody, err := c.sendRequest(http.MethodPost, "/delete_msg", reqBody) - if err != nil { - return nil, err - } - - return c.parseResponse(respBody, "RecallMsg") + return c.callAPI("delete_msg", map[string]interface{}{"message_id": msgID}) } // GetGroupList retrieves the list of joined groups from OneBot. func (c *Client) GetGroupList() (*APIResponse, error) { if actionLogEnabled() { - postLog.Debug("[qq_onebot] GetGroupList -> /get_group_list") + postLog.Debug("[qq_onebot] GetGroupList -> get_group_list") } - respBody, err := c.sendRequest(http.MethodGet, "/get_group_list", nil) - if err != nil { - return nil, err - } - - return c.parseResponse(respBody, "GetGroupList") + return c.callAPI("get_group_list", nil) } // GetGroupInfo retrieves detailed information about a specific group from OneBot. func (c *Client) GetGroupInfo(groupID int64) (*APIResponse, error) { - reqBody, err := json.Marshal(map[string]interface{}{ - "group_id": groupID, - }) - if err != nil { - return nil, fmt.Errorf("failed to marshal request: %w", err) - } - if actionLogEnabled() { - postLog.Debug(fmt.Sprintf("[qq_onebot] GetGroupInfo -> /get_group_info: %d", groupID)) + postLog.Debug(fmt.Sprintf("[qq_onebot] GetGroupInfo -> get_group_info: %d", groupID)) } - respBody, err := c.sendRequest(http.MethodPost, "/get_group_info", reqBody) - if err != nil { - return nil, err - } - - return c.parseResponse(respBody, "GetGroupInfo") + return c.callAPI("get_group_info", map[string]interface{}{"group_id": groupID}) } // GetFriendsList retrieves the friends list from OneBot. func (c *Client) GetFriendsList() (*APIResponse, error) { if actionLogEnabled() { - postLog.Debug("[qq_onebot] GetFriendsList -> /get_friend_list") + postLog.Debug("[qq_onebot] GetFriendsList -> get_friend_list") } - respBody, err := c.sendRequest(http.MethodGet, "/get_friend_list", nil) + return c.callAPI("get_friend_list", nil) +} + +// callAPI issues one OneBot action over the API socket and waits for its reply. +// +// It returns a Go error only for transport failures (not connected, write error, +// disconnect, timeout, Stop). An OneBot-level failure (status "failed", retcode +// 1xxx) is returned as an ordinary (*APIResponse, nil), matching the behavior of +// the HTTP client this replaces. +func (c *Client) callAPI(action string, params map[string]interface{}) (*APIResponse, error) { + // The spec allows omitting params entirely, but some adapters reject a null + // value, so a bodyless action still sends an empty object. + if params == nil { + params = map[string]interface{}{} + } + echo := strconv.FormatUint(c.echoSeq.Add(1), 10) + + payload, err := json.Marshal(apiRequest{Action: action, Params: params, Echo: echo}) + if err != nil { + return nil, fmt.Errorf("failed to marshal onebot request: %w", err) + } + + // Buffered cap 1: the API read loop and failPending can each deliver at most + // once, and delivering must never block the read loop. + ch := make(chan apiResult, 1) + + c.apiMu.Lock() + if c.stopped { + c.apiMu.Unlock() + return nil, fmt.Errorf("onebot client stopped") + } + c.pending[echo] = ch + c.apiMu.Unlock() + + // Deregister on every exit path. dispatchAPI deletes the entry too, so this is + // a no-op when the response already arrived. + defer func() { + c.apiMu.Lock() + delete(c.pending, echo) + c.apiMu.Unlock() + }() + + conn, err := c.waitForAPIConn() if err != nil { return nil, err } - return c.parseResponse(respBody, "GetFriendsList") + if err := c.writeAPI(conn, payload); err != nil { + return nil, fmt.Errorf("failed to send onebot request: %w", err) + } + + timer := time.NewTimer(onebotAPITimeout) + defer timer.Stop() + + select { + case res := <-ch: + return res.resp, res.err + case <-timer.C: + return nil, fmt.Errorf("onebot API %s timed out after %s", action, onebotAPITimeout) + case <-c.stopCh: + return nil, fmt.Errorf("onebot client stopped") + } } -// Listen connects to the OneBot WebSocket server and calls onEvent for every raw -// message. It blocks forever, reconnecting every 5s after a disconnect, until -// Stop is called. Run it in a goroutine. +// waitForAPIConn returns the live API connection, waiting (bounded by +// onebotConnectWait, or until Stop) for apiLoop to establish one when it is not +// connected yet. It re-checks the state on every apiChanged transition, so a +// closed channel left over from a previous connection cannot fool it. +func (c *Client) waitForAPIConn() (*websocket.Conn, error) { + timer := time.NewTimer(onebotConnectWait) + defer timer.Stop() + + for { + c.apiMu.Lock() + conn := c.apiConn + changed := c.apiChanged + stopped := c.stopped + c.apiMu.Unlock() + + if conn != nil { + return conn, nil + } + if stopped { + return nil, fmt.Errorf("onebot client stopped") + } + + select { + case <-changed: + // Connection state changed; re-check. + case <-timer.C: + return nil, fmt.Errorf("onebot API WebSocket not connected") + case <-c.stopCh: + return nil, fmt.Errorf("onebot client stopped") + } + } +} + +// writeAPI serializes one request frame onto conn under writeMu, with a write +// deadline. gorilla/websocket panics on concurrent writes, and both the event +// handler and every SendMessage path call callAPI, so concurrency is the norm. +func (c *Client) writeAPI(conn *websocket.Conn, payload []byte) error { + c.writeMu.Lock() + defer c.writeMu.Unlock() + + if err := conn.SetWriteDeadline(time.Now().Add(onebotWriteTimeout)); err != nil { + return err + } + return conn.WriteMessage(websocket.TextMessage, payload) +} + +// Listen connects the event WebSocket and calls onEvent for every raw message. It +// starts the API socket loop as well, since NewClient is side-effect-free and +// Listen is the single entry point the controller drives. It blocks forever, +// reconnecting every onebotReconnectDelay after a disconnect, until Stop is +// called. Run it in a goroutine. func (c *Client) Listen(onEvent func(raw []byte)) { + go c.apiLoop() + for { select { case <-c.stopCh: @@ -259,30 +349,21 @@ func (c *Client) Listen(onEvent func(raw []byte)) { c.listenOnce(onEvent) - postLog.Warning("[qq_onebot] WebSocket disconnected, reconnecting in 5s...") + postLog.Warning("[qq_onebot] Event WebSocket disconnected, reconnecting in 5s...") select { case <-c.stopCh: return - case <-time.After(5 * time.Second): + case <-time.After(onebotReconnectDelay): } } } -// listenOnce connects to the OneBot WebSocket and reads events until the +// listenOnce connects to the OneBot event WebSocket and reads events until the // connection drops or Stop is called. func (c *Client) listenOnce(onEvent func(raw []byte)) { - wsURL := fmt.Sprintf("ws://%s:%s/", c.addr, c.port) - header := http.Header{} - if c.token != "" { - wsURL += "?access_token=" + url.QueryEscape(c.token) - header.Set("Authorization", "Bearer "+c.token) - } - - dialer := *websocket.DefaultDialer - dialer.Proxy = netproxy.ProxyFunc(c.useProxy) - conn, _, err := dialer.Dial(wsURL, header) + conn, err := c.dialWS("/") if err != nil { - postLog.Error(fmt.Sprintf("[qq_onebot] Failed to connect to OneBot WebSocket: %v", err)) + postLog.Error(fmt.Sprintf("[qq_onebot] Failed to connect to OneBot event WebSocket: %v", err)) return } @@ -298,7 +379,7 @@ func (c *Client) listenOnce(onEvent func(raw []byte)) { conn.Close() }() - postLog.Info(fmt.Sprintf("[qq_onebot] Connected to OneBot WebSocket at %s", wsURL)) + postLog.Info("[qq_onebot] Connected to OneBot event WebSocket") for { select { @@ -309,7 +390,7 @@ func (c *Client) listenOnce(onEvent func(raw []byte)) { _, msgBytes, err := conn.ReadMessage() if err != nil { - postLog.Error(fmt.Sprintf("[qq_onebot] WebSocket read error: %v", err)) + postLog.Error(fmt.Sprintf("[qq_onebot] Event WebSocket read error: %v", err)) return } @@ -317,14 +398,165 @@ func (c *Client) listenOnce(onEvent func(raw []byte)) { } } -// Stop closes any open WebSocket connection and unblocks the Listen reconnect loop. -func (c *Client) Stop() { - c.stopOnce.Do(func() { close(c.stopCh) }) +// apiLoop owns the API socket: it keeps a connection to "/api" up, dispatching +// any reply its read loop receives and reconnecting every onebotReconnectDelay. +// It exits on Stop. +func (c *Client) apiLoop() { + for { + select { + case <-c.stopCh: + return + default: + } - c.connMu.Lock() - defer c.connMu.Unlock() - if c.conn != nil { - c.conn.Close() - c.conn = nil + c.runAPIConnection() + + select { + case <-c.stopCh: + return + case <-time.After(onebotReconnectDelay): + } } } + +// runAPIConnection dials "/api", publishes the connection to waiters, and reads +// response frames until the connection drops or Stop is called. On the way out it +// marks the socket down (waking waiters and failing in-flight calls) so a caller +// does not wait the full timeout for a reply that can no longer arrive. +func (c *Client) runAPIConnection() { + conn, err := c.dialWS("/api") + if err != nil { + postLog.Error(fmt.Sprintf("[qq_onebot] Failed to connect to OneBot API WebSocket: %v", err)) + return + } + + c.apiMu.Lock() + c.apiConn = conn + close(c.apiChanged) + c.apiChanged = make(chan struct{}) + c.apiMu.Unlock() + + postLog.Info("[qq_onebot] Connected to OneBot API WebSocket") + + c.readAPILoop(conn) + + c.apiMu.Lock() + if c.apiConn == conn { + c.apiConn = nil + } + close(c.apiChanged) + c.apiChanged = make(chan struct{}) + c.apiMu.Unlock() + + conn.Close() + c.failPending(fmt.Errorf("onebot API WebSocket disconnected")) +} + +// readAPILoop reads response frames from the API socket until the connection +// drops or Stop is called. It is the only reader of that connection. +func (c *Client) readAPILoop(conn *websocket.Conn) { + for { + select { + case <-c.stopCh: + return + default: + } + + _, data, err := conn.ReadMessage() + if err != nil { + select { + case <-c.stopCh: + // Stopping: the read error is just the closed socket. + default: + postLog.Error(fmt.Sprintf("[qq_onebot] API WebSocket read error: %v", err)) + } + return + } + + c.dispatchAPI(data) + } +} + +// dispatchAPI decodes one inbound frame on the API socket. A frame with no usable +// echo is not an API response and is ignored (the event socket is the only place +// events are expected); otherwise the pending channel registered for that echo +// receives the envelope and is removed, both under apiMu. +func (c *Client) dispatchAPI(raw []byte) { + var resp APIResponse + if err := json.Unmarshal(raw, &resp); err != nil { + postLog.Debug("[qq_onebot] Failed to parse OneBot API response: " + err.Error()) + return + } + + key, ok := echoKey(resp.Echo) + if !ok { + postLog.Debug("[qq_onebot] Ignoring non-response frame on OneBot API WebSocket: " + string(raw)) + return + } + + c.apiMu.Lock() + ch := c.pending[key] + delete(c.pending, key) + c.apiMu.Unlock() + + if ch != nil { + ch <- apiResult{resp: &resp} + } +} + +// echoKey normalizes an echo value to its pending-map key. An absent or null echo +// yields ok=false, i.e. the frame is not an API response. A JSON string is +// unquoted; any other type falls back to its verbatim bytes so an adapter that +// returns the id unquoted still matches. +func echoKey(raw json.RawMessage) (string, bool) { + if len(raw) == 0 || string(raw) == "null" { + return "", false + } + var s string + if err := json.Unmarshal(raw, &s); err == nil { + return s, true + } + return string(raw), true +} + +// failPending delivers err to every call waiting on a response and clears the +// table, so a dropped API socket unblocks its callers immediately. Called on +// disconnect and by Stop; safe to call repeatedly. +func (c *Client) failPending(err error) { + c.apiMu.Lock() + pending := c.pending + c.pending = make(map[string]chan apiResult) + c.apiMu.Unlock() + + for _, ch := range pending { + // Buffered cap 1, and each entry is owned exactly once, so this never blocks. + ch <- apiResult{err: err} + } +} + +// Stop closes both WebSocket connections and unblocks the event reconnect loop, +// the API loop and every in-flight callAPI. It is idempotent and safe to call +// even when Listen was never started. +func (c *Client) Stop() { + c.stopOnce.Do(func() { + close(c.stopCh) + + c.apiMu.Lock() + c.stopped = true + apiConn := c.apiConn + c.apiMu.Unlock() + + c.connMu.Lock() + if c.conn != nil { + c.conn.Close() + c.conn = nil + } + c.connMu.Unlock() + + if apiConn != nil { + apiConn.Close() + } + + c.failPending(fmt.Errorf("onebot client stopped")) + }) +} diff --git a/internal/controller/pipes/qq_onebot/onebot_test.go b/internal/controller/pipes/qq_onebot/onebot_test.go new file mode 100644 index 0000000..9811d4a --- /dev/null +++ b/internal/controller/pipes/qq_onebot/onebot_test.go @@ -0,0 +1,235 @@ +package qq_onebot + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/gorilla/websocket" +) + +// fakeOneBot is a minimal OneBot 11 forward-WebSocket server. "/api" answers +// every action call by echoing its echo with a fixed ok response; "/" accepts the +// event socket and hands it to the test so events can be pushed on demand. +type fakeOneBot struct { + upgrader websocket.Upgrader + + server *httptest.Server + apiRequests chan fakeAPIRequest + eventConns chan *websocket.Conn +} + +type fakeAPIRequest struct { + Action string `json:"action"` + Params json.RawMessage `json:"params"` + Echo json.RawMessage `json:"echo"` +} + +func newFakeOneBot(t *testing.T) *fakeOneBot { + t.Helper() + + f := &fakeOneBot{ + apiRequests: make(chan fakeAPIRequest, 16), + eventConns: make(chan *websocket.Conn, 16), + } + + mux := http.NewServeMux() + mux.HandleFunc("/api", func(w http.ResponseWriter, r *http.Request) { + conn, err := f.upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + + for { + _, data, err := conn.ReadMessage() + if err != nil { + return + } + var req fakeAPIRequest + if err := json.Unmarshal(data, &req); err != nil { + continue + } + f.apiRequests <- req + + resp, _ := json.Marshal(map[string]json.RawMessage{ + "status": json.RawMessage(`"ok"`), + "retcode": json.RawMessage(`0`), + "data": json.RawMessage(`{"message_id":123}`), + "echo": req.Echo, + }) + if err := conn.WriteMessage(websocket.TextMessage, resp); err != nil { + return + } + } + }) + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + conn, err := f.upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + f.eventConns <- conn + + // Keep the event socket open until the client goes away. + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + }) + + f.server = httptest.NewServer(mux) + t.Cleanup(f.server.Close) + return f +} + +// client returns a Client pointed at this fake server with the proxy disabled. +func (f *fakeOneBot) client(t *testing.T) *Client { + t.Helper() + u, err := url.Parse(f.server.URL) + if err != nil { + t.Fatalf("parse test server URL: %v", err) + } + return NewClient(u.Hostname(), u.Port(), "", false) +} + +// eventConn waits for the event socket to be established and returns it. +func (f *fakeOneBot) eventConn(t *testing.T) *websocket.Conn { + t.Helper() + select { + case conn := <-f.eventConns: + return conn + case <-time.After(3 * time.Second): + t.Fatal("event WebSocket never connected") + return nil + } +} + +// TestSendMsgOverWebSocket checks an API call is issued over the "/api" socket +// and its response is decoded. +func TestSendMsgOverWebSocket(t *testing.T) { + f := newFakeOneBot(t) + + c := f.client(t) + go c.Listen(func([]byte) {}) + defer c.Stop() + + resp, err := c.SendMsg("group", 12345, "hello", false, 0) + if err != nil { + t.Fatalf("SendMsg: %v", err) + } + if resp.Status != "ok" || resp.RetCode != 0 { + t.Fatalf("unexpected response: %+v", resp) + } + + var data struct { + MessageID int64 `json:"message_id"` + } + if err := json.Unmarshal(resp.Data, &data); err != nil { + t.Fatalf("decode data: %v", err) + } + if data.MessageID != 123 { + t.Fatalf("message_id = %d, want 123", data.MessageID) + } + + req := <-f.apiRequests + if req.Action != "send_group_msg" { + t.Fatalf("action = %q, want send_group_msg", req.Action) + } + var params map[string]any + if err := json.Unmarshal(req.Params, ¶ms); err != nil { + t.Fatalf("decode params: %v", err) + } + if params["group_id"].(float64) != 12345 || params["message"].(string) != "hello" { + t.Fatalf("params = %v", params) + } + var echo string + if err := json.Unmarshal(req.Echo, &echo); err != nil { + t.Fatalf("echo is not a JSON string: %s", req.Echo) + } +} + +// TestBodylessActionSendsEmptyParams checks actions without arguments still send +// a params object (never null), which some adapters reject. +func TestBodylessActionSendsEmptyParams(t *testing.T) { + f := newFakeOneBot(t) + + c := f.client(t) + go c.Listen(func([]byte) {}) + defer c.Stop() + + if _, err := c.GetGroupList(); err != nil { + t.Fatalf("GetGroupList: %v", err) + } + + req := <-f.apiRequests + if req.Action != "get_group_list" { + t.Fatalf("action = %q, want get_group_list", req.Action) + } + if string(req.Params) != "{}" { + t.Fatalf("params = %s, want {}", req.Params) + } +} + +// TestSendMsgFromEventCallbackDoesNotDeadlock is the regression test for the +// reason the client uses two sockets: the event handler replies with SendMsg from +// inside the event read loop, so a response read on that same connection would +// never arrive. +func TestSendMsgFromEventCallbackDoesNotDeadlock(t *testing.T) { + f := newFakeOneBot(t) + + c := f.client(t) + + type result struct { + resp *APIResponse + err error + } + done := make(chan result, 1) + go c.Listen(func(raw []byte) { + resp, err := c.SendMsg("group", 1, "pong", false, 0) + done <- result{resp, err} + }) + defer c.Stop() + + evConn := f.eventConn(t) + event := []byte(`{"post_type":"message","message_type":"group","group_id":1,"user_id":2,"raw_message":"/ping","self_id":9}`) + if err := evConn.WriteMessage(websocket.TextMessage, event); err != nil { + t.Fatalf("write event: %v", err) + } + + select { + case r := <-done: + if r.err != nil { + t.Fatalf("SendMsg from event callback: %v", r.err) + } + if r.resp.Status != "ok" { + t.Fatalf("unexpected response: %+v", r.resp) + } + case <-time.After(5 * time.Second): + t.Fatal("deadlock: SendMsg did not return from within the event callback") + } +} + +// TestEchoKey covers the response-matcher: an absent or null echo is not a +// response, and both string and non-string echoes normalize to a key. +func TestEchoKey(t *testing.T) { + cases := []struct { + in string + key string + ok bool + }{ + {`"42"`, "42", true}, + {`42`, "42", true}, + {`null`, "", false}, + {``, "", false}, + } + for _, tc := range cases { + key, ok := echoKey(json.RawMessage(tc.in)) + if key != tc.key || ok != tc.ok { + t.Errorf("echoKey(%q) = (%q, %v), want (%q, %v)", tc.in, key, ok, tc.key, tc.ok) + } + } +} diff --git a/internal/controller/pipes/qq_onebot/qq.go b/internal/controller/pipes/qq_onebot/qq.go index 697643d..faa7b0b 100644 --- a/internal/controller/pipes/qq_onebot/qq.go +++ b/internal/controller/pipes/qq_onebot/qq.go @@ -49,7 +49,8 @@ func (q *QQController) Name() string { return "qq_onebot" } -// Start initializes the QQ controller and starts the NapCat WebSocket listener. +// Start initializes the QQ controller and starts the OneBot WebSocket listeners +// (the event socket and the API socket the client manages internally). func (q *QQController) Start() error { if !q.cfg.Enabled { // postLog.Info("QQ (Napcat) controller is disabled") @@ -208,6 +209,7 @@ func (q *QQController) processCommand(cmd controller.Command) string { postLog.Error("[qq_onebot] OneBot command processing failed: " + err.Error()) return "" } + // postLog.Debug("[qq_onebot] processCommand.resp: " + response) return response } diff --git a/internal/controller/trigger.go b/internal/controller/trigger.go index 83fc6a1..f9c15ad 100644 --- a/internal/controller/trigger.go +++ b/internal/controller/trigger.go @@ -3,6 +3,8 @@ package controller import ( "fmt" "strings" + + "nukumizu-backend/postLog" ) // Trigger validates an incoming command against the pipe's authorization @@ -16,9 +18,11 @@ func (m *Manager) Trigger(cmd Command, trustedGroups, admins []string, listenMet // Check whether the command arrived in a group or a private chat. if cmd.ChatType == "group" { // Group commands are only honored from trusted groups, otherwise discard. + // postLog.Debug(fmt.Sprintf("trustedGroups: %v", trustedGroups)) if !IsTrustedGroup(cmd.ChatID, trustedGroups) { return "", nil } + postLog.Debug("isTrustedGroup") } // Commands that require special permission must be sent by an admin. @@ -37,6 +41,7 @@ func (m *Manager) Trigger(cmd Command, trustedGroups, admins []string, listenMet if response == "" && listenMethod == "at" { return "Unknown command: /" + cmd.Command, nil } + // postLog.Debug("[controller] Trigger.resp: " + response) return response, nil }