refactor(qq_onebot): replace all http handler with a brand new websocket handler, seperate with msg recver

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"))
+	})
+}