napcat: NapCatQQ (OneBot 11) Go SDK, initial release

2026-10-18 20:20:29 +08:00
commit 7f1738a987
14 changed files with 2879 additions and 0 deletions
--- a/client.go
+++ b/client.go
@ -0,0 +1,477 @@
+package napcat
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// HandlerFunc 是事件处理函数。
+type HandlerFunc func(c *Context)
+
+// CommandFunc 是命令处理函数，args 为按空白切分的参数。
+type CommandFunc func(c *Context, args []string)
+
+// Connector 是一种通信方式（正向 WS / 反向 WS / HTTP 上报 / HTTP API）。
+// 常规连接直接用 Client 的 Set* 方法配置，本接口留给自定义传输扩展。
+type Connector interface {
+	Name() string
+	Serve(ctx context.Context, bot *Client) error
+}
+
+type handlerSet struct {
+	onEvent     []HandlerFunc
+	onMessage   []HandlerFunc
+	onPrivate   []HandlerFunc
+	onGroup     []HandlerFunc
+	onNotice    []HandlerFunc
+	onRequest   []HandlerFunc
+	onMeta      []HandlerFunc
+	noticeTypes map[string][]HandlerFunc
+	commands    map[string]CommandFunc
+}
+
+// Client 是与 NapCat 交互的客户端对象：Set* 配置连接，On* 注册事件处理，
+// Run 启动后直接调用其内嵌的 API 方法即可操作 NapCat。
+type Client struct {
+	// API 内嵌了全部类型化 API 方法，走 Client 的默认调用通道。
+	API
+
+	log        logger
+	token      string // OneBot access_token
+	secret     string // HTTP 上报 HMAC-SHA1 密钥
+	connectors []Connector
+	httpAPI    *HTTPAPI
+	current    atomic.Pointer[Session]
+	lastBeat   atomic.Int64
+
+	sessMu   sync.Mutex
+	sessions map[*Session]struct{}
+
+	mu sync.Mutex
+	hs handlerSet
+}
+
+// New 创建一个与 NapCat 交互的客户端对象。
+// 用 Set* 方法配置连接后调用 Run 启动：
+//
+//	bot := napcat.New().SetToken("xxx").SetForwardWS("ws://127.0.0.1:3001")
+//	bot.OnMessage(func(c *napcat.Context) { c.ReplyText("hi") })
+//	bot.Run(ctx)
+func New() *Client {
+	c := &Client{log: slog.Default(), sessions: map[*Session]struct{}{}}
+	c.API = API{call: c}
+	c.hs.noticeTypes = map[string][]HandlerFunc{}
+	c.hs.commands = map[string]CommandFunc{}
+	return c
+}
+
+// SetToken 设置 OneBot access_token（WS 握手与 HTTP 调用自动携带，可留空）。
+func (c *Client) SetToken(token string) *Client {
+	c.mu.Lock()
+	c.token = token
+	if c.httpAPI != nil {
+		c.httpAPI.token.Store(token)
+	}
+	c.mu.Unlock()
+	return c
+}
+
+// SetSecret 设置 HTTP 上报的 HMAC-SHA1 校验密钥，留空表示不校验。
+func (c *Client) SetSecret(secret string) *Client {
+	c.mu.Lock()
+	c.secret = secret
+	c.mu.Unlock()
+	return c
+}
+
+// SetLogger 设置自定义日志器，默认 slog.Default()。
+func (c *Client) SetLogger(l *slog.Logger) *Client {
+	if l == nil {
+		l = slog.Default()
+	}
+	c.log = l
+	return c
+}
+
+// SetForwardWS 配置正向 WebSocket：本框架作为客户端连接 NapCat 的 WS 服务。
+// 可传多个地址同时连接多个 NapCat 实例；重复调用会替换之前的正向 WS 配置。
+func (c *Client) SetForwardWS(urls ...string) *Client {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.connectors = dropConnectors[*ForwardWS](c.connectors)
+	for _, u := range urls {
+		c.connectors = append(c.connectors, &ForwardWS{url: u})
+	}
+	return c
+}
+
+// SetReverseWS 配置反向 WebSocket：本框架监听 addr，由 NapCat 主动连入。
+// 每个监听地址可承接任意多个 NapCat 实例；重复调用会替换之前的反向 WS 配置。
+func (c *Client) SetReverseWS(addrs ...string) *Client {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.connectors = dropConnectors[*ReverseWS](c.connectors)
+	for _, a := range addrs {
+		c.connectors = append(c.connectors, &ReverseWS{addr: a})
+	}
+	return c
+}
+
+// SetHTTPAPI 配置 HTTP API：执行 action 时 POST 到 NapCat 的 HTTP 服务。
+// 没有任何 WS 会话时 Client.Call 走该通道；重复调用替换之前的配置。
+func (c *Client) SetHTTPAPI(baseURL string) *Client {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.connectors = dropConnectors[*HTTPAPI](c.connectors)
+	api := NewHTTPAPI(baseURL)
+	api.token.Store(c.token)
+	c.httpAPI = api
+	c.connectors = append(c.connectors, api)
+	return c
+}
+
+// SetHTTPReport 配置 HTTP POST 上报：本框架监听 addr，NapCat 把事件 POST 过来，
+// handler 里的 Reply/Approve 以 OneBot 快速操作形式随响应返回。
+// 可配置多个监听地址；重复调用会替换之前的上报配置。
+func (c *Client) SetHTTPReport(addrs ...string) *Client {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.connectors = dropConnectors[*HTTPReport](c.connectors)
+	for _, a := range addrs {
+		c.connectors = append(c.connectors, &HTTPReport{addr: a})
+	}
+	return c
+}
+
+// Use 注册自定义 Connector（高级用法；常规连接请用 Set* 方法）。
+func (c *Client) Use(conns ...Connector) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for _, conn := range conns {
+		if api, ok := conn.(*HTTPAPI); ok {
+			c.httpAPI = api
+			api.token.Store(c.token)
+		}
+	}
+	c.connectors = append(c.connectors, conns...)
+}
+
+// Token 返回当前 access_token。
+func (c *Client) Token() string { return c.token }
+
+// dropConnectors 过滤掉类型为 T 的连接器。
+func dropConnectors[T Connector](conns []Connector) []Connector {
+	out := make([]Connector, 0, len(conns))
+	for _, conn := range conns {
+		if _, ok := any(conn).(T); ok {
+			continue
+		}
+		out = append(out, conn)
+	}
+	return out
+}
+
+// SelfID 返回当前会话的机器人 QQ 号（连接后有效，否则 0）。
+func (c *Client) SelfID() int64 {
+	if s := c.current.Load(); s != nil {
+		return s.SelfID()
+	}
+	return 0
+}
+
+// LastHeartbeat 返回最近一次心跳时间；从未收到过心跳则返回零值。
+func (c *Client) LastHeartbeat() time.Time {
+	unix := c.lastBeat.Load()
+	if unix == 0 {
+		return time.Time{}
+	}
+	return time.Unix(0, unix)
+}
+
+// Run 启动所有已配置的连接并阻塞，直到 ctx 取消（正常返回 nil）或某个连接致命出错。
+// 取消后会主动关闭所有会话与监听，Run 返回即代表端口与连接已释放。
+// 请在 Run 之前完成 Set* 配置与 On* 注册。
+func (c *Client) Run(ctx context.Context) error {
+	c.mu.Lock()
+	conns := append([]Connector(nil), c.connectors...)
+	c.mu.Unlock()
+	if len(conns) == 0 {
+		return errors.New("napcat: no connection configured, call SetForwardWS/SetReverseWS/SetHTTPReport first")
+	}
+	runCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	var wg sync.WaitGroup
+	errCh := make(chan error, len(conns))
+	for _, conn := range conns {
+		conn := conn
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			err := conn.Serve(runCtx, c)
+			if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, http.ErrServerClosed) {
+				return
+			}
+			select {
+			case errCh <- fmt.Errorf("connector %s: %w", conn.Name(), err):
+			default:
+			}
+		}()
+	}
+
+	var runErr error
+	select {
+	case runErr = <-errCh:
+		c.log.Error("napcat: connector fatal, shutting down", "err", runErr)
+	case <-ctx.Done():
+	}
+	cancel()
+	c.closeSessions()
+	shCtx, shCancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer shCancel()
+	for _, conn := range conns {
+		if sh, ok := conn.(interface{ Shutdown(context.Context) error }); ok {
+			_ = sh.Shutdown(shCtx)
+		}
+	}
+	wg.Wait()
+	if runErr != nil {
+		return runErr
+	}
+	return nil
+}
+
+func (c *Client) closeSessions() {
+	c.sessMu.Lock()
+	defer c.sessMu.Unlock()
+	for s := range c.sessions {
+		_ = s.Close()
+	}
+}
+
+// Call 实现 Caller：优先走当前 WS 会话，其次走 HTTP API，都没有则报错。
+func (c *Client) Call(ctx context.Context, action string, params any, out any) error {
+	if s := c.current.Load(); s != nil {
+		return s.Call(ctx, action, params, out)
+	}
+	if c.httpAPI != nil {
+		return c.httpAPI.Call(ctx, action, params, out)
+	}
+	return ErrNotConnected
+}
+
+func (c *Client) bindSession(s *Session) {
+	c.sessMu.Lock()
+	c.sessions[s] = struct{}{}
+	c.sessMu.Unlock()
+	c.current.Store(s)
+}
+
+func (c *Client) unbindSession(s *Session) {
+	c.sessMu.Lock()
+	delete(c.sessions, s)
+	c.sessMu.Unlock()
+	c.current.CompareAndSwap(s, nil)
+}
+
+// handleRaw 处理 WS 连接上的一条原始消息：事件则入队分发，带 echo 的则为 API 响应。
+func (c *Client) handleRaw(s *Session, data []byte) {
+	var probe struct {
+		PostType string `json:"post_type"`
+	}
+	if err := json.Unmarshal(data, &probe); err != nil {
+		c.log.Warn("napcat: invalid payload", "err", err)
+		return
+	}
+	if probe.PostType != "" {
+		ev, err := decodeEvent(data)
+		if err != nil {
+			c.log.Warn("napcat: decode event failed", "err", err)
+			return
+		}
+		if ev.SelfID != 0 {
+			s.selfID.Store(ev.SelfID)
+		}
+		if ev.Meta != nil {
+			if ev.Meta.MetaType == "heartbeat" {
+				c.lastBeat.Store(time.Now().UnixNano())
+			}
+			if ev.Meta.MetaType == "lifecycle" && ev.Meta.SubType == "connect" {
+				c.log.Info("napcat: bot connected", "self_id", ev.SelfID)
+			}
+		}
+		s.enqueue(ev, nil)
+		return
+	}
+	s.caller.routeResponse(data, c.log)
+}
+
+// ---- 事件处理注册 ----
+
+// OnEvent 注册处理所有事件的函数。
+func (c *Client) OnEvent(fn HandlerFunc) { c.add(&c.hs.onEvent, fn) }
+
+// OnMessage 注册处理私聊+群聊消息事件（含 message_sent）的函数。
+func (c *Client) OnMessage(fn HandlerFunc) { c.add(&c.hs.onMessage, fn) }
+
+// OnPrivateMessage 注册处理私聊消息事件的函数。
+func (c *Client) OnPrivateMessage(fn HandlerFunc) { c.add(&c.hs.onPrivate, fn) }
+
+// OnGroupMessage 注册处理群聊消息事件的函数。
+func (c *Client) OnGroupMessage(fn HandlerFunc) { c.add(&c.hs.onGroup, fn) }
+
+// OnNotice 注册处理所有通知事件的函数。
+func (c *Client) OnNotice(fn HandlerFunc) { c.add(&c.hs.onNotice, fn) }
+
+// OnNoticeType 注册处理指定 notice_type 通知事件的函数（如 "notify"、"group_recall"）。
+func (c *Client) OnNoticeType(noticeType string, fn HandlerFunc) {
+	c.mu.Lock()
+	c.hs.noticeTypes[noticeType] = append(c.hs.noticeTypes[noticeType], fn)
+	c.mu.Unlock()
+}
+
+// OnRequest 注册处理所有请求事件（好友/群）的函数。
+func (c *Client) OnRequest(fn HandlerFunc) { c.add(&c.hs.onRequest, fn) }
+
+// OnMeta 注册处理元事件（心跳/生命周期）的函数。
+func (c *Client) OnMeta(fn HandlerFunc) { c.add(&c.hs.onMeta, fn) }
+
+// OnCommand 注册文本命令：消息文本等于 name 或以 name+空白 开头时触发。
+// 命令名自带头尾风格（如 "/ping" 或 "ping"）。
+func (c *Client) OnCommand(name string, fn CommandFunc) {
+	c.mu.Lock()
+	c.hs.commands[name] = fn
+	c.mu.Unlock()
+}
+
+func (c *Client) add(dst *[]HandlerFunc, fn HandlerFunc) {
+	c.mu.Lock()
+	*dst = append(*dst, fn)
+	c.mu.Unlock()
+}
+
+func (c *Client) snapshot() handlerSet {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	hs := handlerSet{
+		onEvent:     slicesClone(c.hs.onEvent),
+		onMessage:   slicesClone(c.hs.onMessage),
+		onPrivate:   slicesClone(c.hs.onPrivate),
+		onGroup:     slicesClone(c.hs.onGroup),
+		onNotice:    slicesClone(c.hs.onNotice),
+		onRequest:   slicesClone(c.hs.onRequest),
+		onMeta:      slicesClone(c.hs.onMeta),
+		noticeTypes: mapsCloneOf(c.hs.noticeTypes),
+		commands:    mapsCloneCmd(c.hs.commands),
+	}
+	return hs
+}
+
+// dispatch 按事件类型把事件交给已注册的处理函数；HTTP 上报模式下 quick 非空。
+func (c *Client) dispatch(ev *Event, caller Caller, quick *quickBox) {
+	if caller == nil {
+		caller = c
+	}
+	ctx := &Context{Bot: c, Event: ev, API: API{call: caller}, quick: quick}
+	hs := c.snapshot()
+
+	run := func(fns []HandlerFunc) {
+		for _, fn := range fns {
+			func() {
+				defer func() {
+					if r := recover(); r != nil {
+						c.log.Error("napcat: handler panic", "post_type", ev.PostType, "panic", fmt.Sprint(r))
+					}
+				}()
+				fn(ctx)
+			}()
+		}
+	}
+
+	run(hs.onEvent)
+	switch ev.PostType {
+	case "message", "message_sent":
+		run(hs.onMessage)
+		if ev.Message != nil {
+			if ev.Message.MessageType == "group" {
+				run(hs.onGroup)
+			} else {
+				run(hs.onPrivate)
+			}
+			c.matchCommand(ctx, hs, run)
+		}
+	case "notice":
+		run(hs.onNotice)
+		if ev.Notice != nil {
+			run(hs.noticeTypes[ev.Notice.NoticeType])
+		}
+	case "request":
+		run(hs.onRequest)
+	case "meta_event":
+		run(hs.onMeta)
+	}
+}
+
+// matchCommand 命中已注册命令则调用对应处理函数。
+func (c *Client) matchCommand(ctx *Context, hs handlerSet, run func([]HandlerFunc)) {
+	if len(hs.commands) == 0 || ctx.Event.Message == nil {
+		return
+	}
+	text := strings.TrimSpace(SegmentsText(ctx.Segments()))
+	if text == "" {
+		return
+	}
+	for name, fn := range hs.commands {
+		if text != name && !strings.HasPrefix(text, name+" ") {
+			continue
+		}
+		rest := strings.TrimSpace(strings.TrimPrefix(text, name))
+		args := []string{}
+		if rest != "" {
+			args = strings.Fields(rest)
+		}
+		func() {
+			defer func() {
+				if r := recover(); r != nil {
+					c.log.Error("napcat: command handler panic", "cmd", name, "panic", fmt.Sprint(r))
+				}
+			}()
+			fn(ctx, args)
+		}()
+		return
+	}
+}
+
+// 小工具：避免每个文件都引 slices/maps 包。
+func slicesClone(s []HandlerFunc) []HandlerFunc {
+	if s == nil {
+		return nil
+	}
+	out := make([]HandlerFunc, len(s))
+	copy(out, s)
+	return out
+}
+
+func mapsCloneOf(m map[string][]HandlerFunc) map[string][]HandlerFunc {
+	out := make(map[string][]HandlerFunc, len(m))
+	for k, v := range m {
+		out[k] = slicesClone(v)
+	}
+	return out
+}
+
+func mapsCloneCmd(m map[string]CommandFunc) map[string]CommandFunc {
+	out := make(map[string]CommandFunc, len(m))
+	for k, v := range m {
+		out[k] = v
+	}
+	return out
+}