From b504df01e0c3c2f3546ee8e0e26b6642e7034464 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Tue, 13 Oct 2026 20:38:12 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=20=E5=87=BA=E7=AB=99=E6=B6=88?= =?UTF-8?q?=E6=81=AF=20id=20=E9=87=87=E9=9B=86=EF=BC=88=E5=8F=AF=E9=80=89?= =?UTF-8?q?=20seam=20+=20qq=5Fobv11=20=E4=B8=A4=E5=88=86=E6=94=AF=E5=8F=96?= =?UTF-8?q?=E5=9B=9E=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 「接她的话」缺的 A 层：平台给她自己那条出站消息分配的 id，此前在解析的那一刻 就被丢掉（HTTP 只校 retcode、WS 写完不等回声、契约层没有 id 出口）。 - internal/channel/adapter/outbound.go（新）：SendReceipt/SendOutcome/SendIDStatus + 可选接口 OutboundReceiptAdapter + 调用方入口 SendWithReceipt。**不加进 PlatformAdapter** —— 四个未接线适配器与全部桩实现一行不改（向后兼容硬指标）； 也不用回调：id 是某一次投递的属性，必须随那次投递返回。 - qq_obv11：HTTP 从响应信封 data.message_id 取；WS 登记 send_msg 的 echo 等待者 （复用 probe.go 的 action 等待表与 routeActionResponse，不另开通道）。取不到 一律空 id + 三态可分辨（reported/absent/unreported），绝不编造。 - send_msg echo 键改用单调计数器（旧 UnixNano 在 Windows 时钟粒度下会撞键， 撞键即拿不到自己的回执）。 - 既有 SendMessage/SendProactive 路径行为不变（wantReceipt=false 不等待）。 验收：go build ./... + go vet ./... + go test ./... 全绿（cmd/cyrene、internal/webui 的 setup 失败是 worktree 缺 dist 的既有环境限制，base 同样失败）。 --- internal/channel/adapter/outbound.go | 137 ++++++ internal/channel/adapter/outbound_test.go | 150 ++++++ plugins/channels/qq_obv11/adapter.go | 34 +- plugins/channels/qq_obv11/http_client.go | 11 +- plugins/channels/qq_obv11/outbound_id.go | 168 +++++++ plugins/channels/qq_obv11/outbound_id_test.go | 458 ++++++++++++++++++ plugins/channels/qq_obv11/probe.go | 32 +- plugins/channels/qq_obv11/session.go | 126 +++-- 8 files changed, 1053 insertions(+), 63 deletions(-) create mode 100644 internal/channel/adapter/outbound.go create mode 100644 internal/channel/adapter/outbound_test.go create mode 100644 plugins/channels/qq_obv11/outbound_id.go create mode 100644 plugins/channels/qq_obv11/outbound_id_test.go diff --git a/internal/channel/adapter/outbound.go b/internal/channel/adapter/outbound.go new file mode 100644 index 0000000..b78fe3f --- /dev/null +++ b/internal/channel/adapter/outbound.go @@ -0,0 +1,137 @@ +package adapter + +// 本文件承载**出站消息 id 的出口**（切片 OUTBOUNDID / M12 的 A 层）。 +// +// 背景（M11 查明的缺口）：`PlatformAdapter.SendMessage(ctx, resp) error` 只回成败， +// **没有任何位置**放下「平台给这条消息分配的 id」。于是她自己刚发出的那条消息在系统里 +// 没有身份 —— 平台侧明明把 id 回给了我们，却在解析的那一刻就被丢掉。任何「接她的话」 +// （对方回复的是**她自己**那条消息 ⇒ `UnifiedMessage.ReplyToID` 落在她自己的出站 id 上） +// 的判定都无从谈起。 +// +// ## 形态选择：**可选接口 + 调用方探测**（不采用返回 id 的直接改签名，也不用回调） +// +// 范式与 `SystemNoticeNormalizer` / `DeliveryFailure` 逐字同族 —— 都在本包，都是 +// 「适配器按需实现、调用方探测」，理由也一样：契约层是唯一不必让任何一方反向依赖 +// 具体插件的落点。 +// +// - **不把它加进 `PlatformAdapter`**：加方法等于契约破坏 —— 四个未接线适配器 +// （kook / telegram / weixin-ilink / webhook）与全部桩实现会**立刻编译失败**， +// 而它们拿不到 id 的原因是**平台不给**，不是「没接线」。向后兼容是硬指标。 +// - **也不采用回调 / sink 注册**：那会把 id 变成异步、无关联的推送 —— 调用方拿到一条 +// id 时已经无法确定它属于哪一次发送、哪一个会话；重连与重试下还会把上一轮的旧 id +// 混进环形表。出站 id 是**某一次投递的属性**，就该随那次投递一起返回。 +// +// ## 三条铁律（任何实现都必须遵守） +// +// 1. **取不到就空**：`SendReceipt.MessageID` 为空一律表示「不知道」，绝不用占位值 / +// 序号 / 时间戳兜底。一个编出来的 id 会让「接她的话」凭空命中（把别人的话认成 +// 回给她的）。宁可判不出来，不许判错。 +// 2. **三态可分辨**：`SendIDStatus` 把「回执带回了 id」「回执到了但没带 id」 +// 「压根没等到回执」分开 —— 排障时才分得清「平台没上报」与「我们没等到」。 +// 3. **不改变既有投递语义**：这是**增量** seam。`SendMessage` 的成败判据、限流、 +// 目标解析一律照旧；没实现本接口的适配器行为一个字节都不变。 + +import "context" + +// SendOutcome 是一次投递的完整结果：**逐条**的投递凭据，顺序与 +// `UnifiedResponse.Messages` 一一对应。 +// +// 用结构体而不是裸切片：日后要给投递结果补别的属性（逐条耗时可观测、失败分类）时， +// 签名不必再动一次。 +type SendOutcome struct { + // Receipts 是每条已投递消息的回执。调用方必须按**下标**与 Messages 对齐 —— + // 长度不一致说明适配器没遵守契约（实现应保证等长）。 + Receipts []SendReceipt +} + +// SendReceipt 是一条出站消息的投递凭据。`MessageID` 是出站 id 的**唯一**出口。 +type SendReceipt struct { + // MessageID 是平台回执给出的消息 id（OneBot 的 `data.message_id`）。 + // + // **空串一律表示「不知道」**：没等到回执、回执里没有这个字段、或平台根本不给。 + // 绝不编造 —— 下游判「对方是不是在接她的话」时，空 id 必须落成「判不出来」， + // 而不是拿一个假值去撞环形表。 + MessageID string + // IDStatus 说明上面那个空 / 非空是怎么来的（见 SendIDStatus 的三态）。 + IDStatus SendIDStatus +} + +// SendIDStatus 是出站 id 的**到达状态**（三态，绝不合并）。 +// +// 零值即 `SendIDUnreported`（不知道）：把「不知道」设为零值是刻意的 fail-closed +// 设计 —— 一个只写了 `SendReceipt{}` 的实现天然表示「我取不到 id」，绝不会因为 +// 忘了填某个字段而被读成「取到了」。 +type SendIDStatus string + +const ( + // SendIDUnreported 表示**没等到回执**：等待超时 / 连接在投递中途断了 / 适配器 + // 压根没有取 id 的能力（此时它不该实现 OutboundReceiptAdapter，但调用方探测失败 + // 时也走这一态）。帧可能已经写出去了 ——「发没发出去」由 error 回答，这一态只回 + // 答「id 拿没拿到」。 + SendIDUnreported SendIDStatus = "" + // SendIDReported 表示**回执把 id 带回来了**（此时 MessageID 必非空）。 + SendIDReported SendIDStatus = "reported" + // SendIDAbsent 表示**回执到了，但它没带 id**（字段缺失 / 为 null / 形态不可用）。 + // 与 SendIDUnreported 分开：前者是「平台明确给了个空」，后者是「我们没等到」。 + SendIDAbsent SendIDStatus = "absent" +) + +// NewSendReceipt 由平台回执里的 id 构造凭据：非空 ⇒ Reported，空 ⇒ Absent。 +// +// 它是**唯一的构造出处**，因此 `MessageID` 与 `IDStatus` 不可能自相矛盾 +// （不会出现「状态说取到了、id 却是空的」）。只用于「回执已经到了」的场景 —— +// 没等到回执时用零值 `SendReceipt{}`（SendIDUnreported）。 +func NewSendReceipt(platformID string) SendReceipt { + if platformID == "" { + return SendReceipt{IDStatus: SendIDAbsent} + } + return SendReceipt{MessageID: platformID, IDStatus: SendIDReported} +} + +// Reported 报告这条凭据里是否真的有一个可用的平台 id。 +func (r SendReceipt) Reported() bool { + return r.IDStatus == SendIDReported && r.MessageID != "" +} + +// OutboundReceiptAdapter 是**可选**的出站凭据 seam：能在投递时给出平台消息 id 的适配器 +// 实现它。调用方用 `SendWithReceipt` 探测，绝不直接类型断言。 +type OutboundReceiptAdapter interface { + // SendMessageWithReceipt 的投递语义与 `SendMessage` **完全一致**（同一套目标解析、 + // 限流、错误与重试口径），额外逐条回一条投递凭据。 + // + // 返回的 error 只描述**投递**：拿到了回执但回执报错、或回执里没有 id，都**不是** + // 这里的 error（那是 `SendIDStatus` 的答案）。error 非 nil 时 `SendOutcome` 可能 + // 只带了前几条的凭据 —— 调用方按 `len(Receipts)` 取**已投递**的那几条。 + SendMessageWithReceipt(ctx context.Context, resp UnifiedResponse) (SendOutcome, error) +} + +// SendWithReceipt 投递一次回复并**尽力**取回平台消息 id —— 调用方（B 层的环形表） +// 的唯一入口。 +// +// - 适配器实现了 `OutboundReceiptAdapter`：走它的凭据路径（HTTP / WS 各按自己的 +// 平台协议取）。 +// - 没实现（kook / telegram / weixin-ilink / webhook 等）：退回既有的 `SendMessage`， +// 并**如实**为每条消息报一条 `SendIDUnreported` 的空凭据 —— 不是失败，是「这个 +// 渠道给不出 id」。调用方据此就知道该渠道的环形表永远是空的，而不是以为取到了。 +// +// 两种情况的 Receipts 长度都等于 len(resp.Messages)，调用方不必分叉处理。 +func SendWithReceipt(ctx context.Context, a PlatformAdapter, resp UnifiedResponse) (SendOutcome, error) { + if a == nil { + return SendOutcome{}, ErrNotConnected + } + if ra, ok := a.(OutboundReceiptAdapter); ok { + return ra.SendMessageWithReceipt(ctx, resp) + } + if err := a.SendMessage(ctx, resp); err != nil { + return SendOutcome{}, err + } + return SendOutcome{Receipts: makeUnknownReceipts(len(resp.Messages))}, nil +} + +// makeUnknownReceipts 造 n 条「不知道」的空凭据（零值即不知道，见 SendIDStatus）。 +func makeUnknownReceipts(n int) []SendReceipt { + if n <= 0 { + return nil + } + return make([]SendReceipt, n) +} diff --git a/internal/channel/adapter/outbound_test.go b/internal/channel/adapter/outbound_test.go new file mode 100644 index 0000000..1760fb0 --- /dev/null +++ b/internal/channel/adapter/outbound_test.go @@ -0,0 +1,150 @@ +package adapter + +// 本文件证明出站 id 出口（outbound.go）的三条硬性质：**向后兼容**（老适配器一行不改、 +// 探测失败即如实报「不知道」）、**三态可分辨**、**取不到就是空 / 绝不编造**。 + +import ( + "context" + "errors" + "testing" +) + +// receiptAdapter 是同时实现了 PlatformAdapter 与 OutboundReceiptAdapter 的桩： +// 它记录两条路径各被调用了几次，用来证明「实现了 seam 的适配器**只**走凭据路径」 +// （绝不出现一条消息发两遍）。 +type receiptAdapter struct { + mockAdapter + receiptCalls int + receipts []SendReceipt +} + +func (m *receiptAdapter) SendMessageWithReceipt(_ context.Context, resp UnifiedResponse) (SendOutcome, error) { + m.receiptCalls++ + m.sent = resp.Messages + return SendOutcome{Receipts: m.receipts}, nil +} + +var ( + _ PlatformAdapter = (*receiptAdapter)(nil) + _ OutboundReceiptAdapter = (*receiptAdapter)(nil) +) + +// failAdapter 的投递恒失败，用来验证错误原样穿过 SendWithReceipt。 +type failAdapter struct { + mockAdapter + err error +} + +func (f *failAdapter) SendMessage(context.Context, UnifiedResponse) error { return f.err } + +func TestNewSendReceiptNeverReportsEmptyID(t *testing.T) { + t.Parallel() + + absent := NewSendReceipt("") + if absent.MessageID != "" || absent.IDStatus != SendIDAbsent || absent.Reported() { + t.Errorf("NewSendReceipt(\"\") = %+v, want absent with empty id", absent) + } + + reported := NewSendReceipt("1700000012345") + if reported.MessageID != "1700000012345" || !reported.Reported() { + t.Errorf("NewSendReceipt(id) = %+v, want reported id", reported) + } +} + +// TestSendReceiptZeroValueIsUnknown 钉住 fail-closed 的零值口径：一个只写了 +// `SendReceipt{}` 的实现表示「不知道」，绝不能被读成「平台给了 id」。 +func TestSendReceiptZeroValueIsUnknown(t *testing.T) { + t.Parallel() + + var zero SendReceipt + if zero.IDStatus != SendIDUnreported { + t.Errorf("zero IDStatus = %q, want %q", zero.IDStatus, SendIDUnreported) + } + if zero.MessageID != "" || zero.Reported() { + t.Errorf("zero receipt = %+v, want empty and not reported", zero) + } +} + +// TestPlainAdapterIsNotAReceiptAdapter 是**向后兼容**的运行时证据：一个只实现 +// PlatformAdapter 的适配器（等价于 kook / telegram / weixin-ilink / webhook）**不是** +// OutboundReceiptAdapter —— 契约没被破坏，它们一行都不用改。 +func TestPlainAdapterIsNotAReceiptAdapter(t *testing.T) { + t.Parallel() + + var a PlatformAdapter = &mockAdapter{name: "telegram"} + if _, ok := a.(OutboundReceiptAdapter); ok { + t.Fatal("plain adapter must not satisfy OutboundReceiptAdapter") + } +} + +// TestSendWithReceiptFallsBackToSendMessage 覆盖未接线渠道：投递照旧走 SendMessage， +// 凭据如实报「未取到」——**逐条**与 Messages 等长，且 id 全空（绝不编造）。 +func TestSendWithReceiptFallsBackToSendMessage(t *testing.T) { + t.Parallel() + + ad := &mockAdapter{name: "telegram"} + resp := UnifiedResponse{Messages: []UnifiedMessage{ + {ChannelType: ChannelDirect, ContentType: ContentText, Content: "一"}, + {ChannelType: ChannelDirect, ContentType: ContentText, Content: "二"}, + }} + + out, err := SendWithReceipt(context.Background(), ad, resp) + if err != nil { + t.Fatalf("SendWithReceipt() error = %v", err) + } + if len(ad.sent) != 2 { + t.Fatalf("fallback did not deliver through SendMessage: sent = %+v", ad.sent) + } + if len(out.Receipts) != 2 { + t.Fatalf("Receipts len = %d, want 2 (one per message)", len(out.Receipts)) + } + for i, r := range out.Receipts { + if r.MessageID != "" || r.IDStatus != SendIDUnreported || r.Reported() { + t.Errorf("receipt[%d] = %+v, want empty id with unreported status", i, r) + } + } +} + +// TestSendWithReceiptUsesSeamWhenImplemented 证明实现了 seam 的适配器只走凭据路径 +// （不会两条路径各发一遍）。 +func TestSendWithReceiptUsesSeamWhenImplemented(t *testing.T) { + t.Parallel() + + ad := &receiptAdapter{receipts: []SendReceipt{NewSendReceipt("9001")}} + ad.name = "qq-obv11" + + out, err := SendWithReceipt(context.Background(), ad, UnifiedResponse{Messages: []UnifiedMessage{ + {ChannelType: ChannelGroup, GroupID: "123", ContentType: ContentText, Content: "pong"}, + }}) + if err != nil { + t.Fatalf("SendWithReceipt() error = %v", err) + } + if ad.receiptCalls != 1 { + t.Errorf("receipt path calls = %d, want 1", ad.receiptCalls) + } + if len(out.Receipts) != 1 || out.Receipts[0].MessageID != "9001" { + t.Errorf("Receipts = %+v, want one reported 9001", out.Receipts) + } +} + +func TestSendWithReceiptPropagatesError(t *testing.T) { + t.Parallel() + + want := errors.New("boom") + ad := &failAdapter{err: want} + + _, err := SendWithReceipt(context.Background(), ad, UnifiedResponse{Messages: []UnifiedMessage{ + {ChannelType: ChannelDirect, ContentType: ContentText, Content: "x"}, + }}) + if !errors.Is(err, want) { + t.Fatalf("SendWithReceipt() error = %v, want %v", err, want) + } +} + +func TestSendWithReceiptNilAdapter(t *testing.T) { + t.Parallel() + + if _, err := SendWithReceipt(context.Background(), nil, UnifiedResponse{}); !errors.Is(err, ErrNotConnected) { + t.Fatalf("SendWithReceipt(nil) error = %v, want ErrNotConnected", err) + } +} diff --git a/plugins/channels/qq_obv11/adapter.go b/plugins/channels/qq_obv11/adapter.go index 7a28b90..0cf1762 100644 --- a/plugins/channels/qq_obv11/adapter.go +++ b/plugins/channels/qq_obv11/adapter.go @@ -135,15 +135,24 @@ type Adapter struct { // 事故正是把两者混为一谈：WS 通 ≠ QQ 账号在线。由 mu 保护。 account accountStatus - // probeMu / probeWait 是 WS 上「action 请求 → echo 回执」的等待表；probeEvery - // 是两次 get_status 探活之间的正常间隔（失败按 policy 退避）。见 probe.go。 - probeMu sync.Mutex - probeWait map[string]chan *obAPIResponse + // actionMu / actionWait 是 WS 上「出站 action 请求 → echo 回执」的等待表：探活 + // （get_status，probe.go）与出站回执采集（send_msg，outbound_id.go）共用同一张表、 + // 同一条 routeActionResponse —— echo 前缀不同（status_ / send_），永不撞键。 + // probeEvery 是两次 get_status 探活之间的正常间隔（失败按 policy 退避）。见 probe.go。 + actionMu sync.Mutex + actionWait map[string]chan *obAPIResponse probeEvery time.Duration probeTimeout time.Duration + // sendReceiptTimeout 是 SendMessageWithReceipt 在 WS 上等待 send_msg 回执的上限 + // （见 outbound_id.go；测试可收紧）。HTTP 端不涉及：响应本来就要等。 + sendReceiptTimeout time.Duration // probeSeq 是 WS 探活 echo 的单调序号（见 probe.go 的 nextProbeEcho）：唯一性不依赖 // 时钟精度，两次探活绝不会撞在同一个 echo 上。 probeSeq atomic.Uint64 + // sendSeq 是 WS 出站 send_msg echo 的单调序号（见 outbound_id.go 的 nextSendEcho）， + // 与 probeSeq 同理：同一毫秒内的多次发送绝不能撞在同一个 echo 上，否则先登记的那次 + // 发送会拿不到自己的回执（旧实现用 UnixNano，Windows 时钟粒度下真会撞）。 + sendSeq atomic.Uint64 } // Compile-time conformance with the channel contract. @@ -178,14 +187,15 @@ func NewAdapter(cfg Config, opts ...Option) *Adapter { cfg.ProactiveWindow = defaultProactiveWindow } a := &Adapter{ - cfg: cfg, - groups: map[string]string{}, - newConn: make(chan struct{}, 1), - limiter: newRateLimiter(cfg.MaxProactivePerWindow, cfg.ProactiveWindow), - policy: defaultReconnectPolicy(), - probeWait: map[string]chan *obAPIResponse{}, - probeEvery: defaultProbeInterval, - probeTimeout: defaultProbeTimeout, + cfg: cfg, + groups: map[string]string{}, + newConn: make(chan struct{}, 1), + limiter: newRateLimiter(cfg.MaxProactivePerWindow, cfg.ProactiveWindow), + policy: defaultReconnectPolicy(), + actionWait: map[string]chan *obAPIResponse{}, + probeEvery: defaultProbeInterval, + probeTimeout: defaultProbeTimeout, + sendReceiptTimeout: defaultSendReceiptTimeout, } for _, opt := range opts { opt(a) diff --git a/plugins/channels/qq_obv11/http_client.go b/plugins/channels/qq_obv11/http_client.go index 630ff4e..1e35a7a 100644 --- a/plugins/channels/qq_obv11/http_client.go +++ b/plugins/channels/qq_obv11/http_client.go @@ -93,13 +93,6 @@ func (c *obHTTPClient) Close() { c.http.CloseIdleConnections() } -// send 把一条已构造好的 action 请求（payload 已含 action/params/echo）POST 给 OneBot -// 并等待结果。它替代 WS 端的「写一帧就返回」：HTTP 有明确响应，成败以 retcode 为准。 -func (c *obHTTPClient) send(ctx context.Context, payload []byte) error { - _, err := c.post(ctx, payload) - return err -} - // probe 用最轻量的标准 action `get_login_info` 探活 OneBot 实现，供 HealthCheck 使用。 // 它不打印 token，也不记录返回的机器人账号信息。 func (c *obHTTPClient) probe(ctx context.Context) error { @@ -124,6 +117,10 @@ func (c *obHTTPClient) call(ctx context.Context, action string, params any) (*ob // post 发送一条 action 请求并解析响应。任何失败都返回带上下文（action、retcode）的 // wrapped error；错误信息与日志都不含 token。 +// +// 它把**响应信封整个交回**调用方：send_msg 的出站消息 id 就在 `data.message_id` 里 +// （见 outbound_id.go 的 receiptFromResponse），不再像从前那样只校完 retcode 就把信封 +// 丢掉 —— 平台明明回了 id，我们却在解析的那一刻扔掉它，正是本切片要修的缺口。 func (c *obHTTPClient) post(ctx context.Context, payload []byte) (*obAPIResponse, error) { if c == nil { return nil, errors.New("qq-obv11: http api client not configured") diff --git a/plugins/channels/qq_obv11/outbound_id.go b/plugins/channels/qq_obv11/outbound_id.go new file mode 100644 index 0000000..76d2cb9 --- /dev/null +++ b/plugins/channels/qq_obv11/outbound_id.go @@ -0,0 +1,168 @@ +package qq_obv11 + +// 本文件承载 OneBot v11 **出站消息 id 的采集**（切片 OUTBOUNDID / M12 的 A 层）： +// 把「她刚发出的那条消息」在平台侧的 id 取出来，经 adapter.OutboundReceiptAdapter 交出去。 +// +// 两条出站通道各有各的取法（同一个 OneBot 协议的两半，不是两套实现）： +// +// - **HTTP**（Cyrene→OneBot 的 HTTP API）：响应信封本来就等回来了，id 就在 +// `data.message_id` 里。此前 http_client.post 只校 retcode/status 就把它扔掉 —— +// 本文件把它读出来。 +// - **WS**（两端共用的长连接）：`send_msg` 是请求-响应，回执按 `echo` 关联。此前只 +// `WriteMessage` 写帧、不等回声，id 根本无从取得；现在**复用 probe.go 的 action +// 等待表**登记等待者（同一张表、同一条 routeActionResponse，不另开请求通道， +// 也不新增读连接）。 +// +// 诚实口径（本文件存在的理由）：**取不到就是「不知道」**，绝不留一个像 id 的假值。 +// 三态（带回 / 没带 / 没等到）在 `adapter.SendReceipt.IDStatus` 里可分辨，逐条与出站 +// 消息对齐；一个编出来的 id 会让下一片（M13）的「接她的话」判定凭空命中。 + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "strconv" + "strings" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +const ( + // sendEchoPrefix 让 send_msg 的回执关联键与探活（status_）区分开：两者共用同一张 + // 等待表，前缀不同就永远撞不了键。 + sendEchoPrefix = "send_" + // defaultSendReceiptTimeout 是等待 send_msg 回执的上限（可由 Adapter.sendReceiptTimeout + // 收紧，测试用）。本地 OneBot 的回执是**毫秒级**的 —— 探活 get_status 早就在同一条 + // 路径上按 echo 关联，生产一直通。5s 只覆盖真实异常：对端假死、连接刚断、实现不回声。 + // 等不到就如实报「不知道」，绝不挂死发送路径。 + defaultSendReceiptTimeout = 5 * time.Second + // maxPlatformIDLen 是平台 id 的长度上限。id 是**对端给的**字符串：既不能照收一段 + // 任意长度的文本当 id，也不能截断它（截出来的半截 id 是编造）。超长一律判「没给」。 + maxPlatformIDLen = 64 +) + +// obSendData 是 send_msg 响应 `data` 载荷里本能力**唯一**需要的字段。 +// +// 用 json.RawMessage 收而不是 int64：实现之间有出入（go-cqhttp 家族回数字，有的回 +// 字符串，有的干脆不回），形态校验统一交给 platformMessageID —— 收成 int64 会把 +// 「回字符串」误判成 0（= 没给），那是拿实现差异当事实。 +type obSendData struct { + MessageID json.RawMessage `json:"message_id"` +} + +// receiptFromResponse 从一封装载 send_msg 回执的信封里取凭据。 +// +// 它只回答「这一封装载里有没有一个可用的 id」。信封自带的 retcode / status **不在这里 +// 判**：HTTP 侧由 http_client.post 按既有判据校验（非 0 直接是 error，根本走不到这里）， +// WS 侧由 logEchoFailure 如实记一条 WARN（判据与 post 相同）。两份判据同源，不重复。 +func receiptFromResponse(resp *obAPIResponse) adapter.SendReceipt { + if resp == nil { + // 没等到回执 —— 不知道。帧可能已经写出去了，但 id 与我们无关了。 + return adapter.SendReceipt{} + } + var data obSendData + if err := json.Unmarshal(resp.Data, &data); err != nil { + // 回执到了，但 data 读不出来（坏形态）。这仍是「回执没带 id」而**不是** + // 「没等到回执」：两态分开，排障时才分得清是谁的问题。 + return adapter.NewSendReceipt("") + } + return adapter.NewSendReceipt(platformMessageID(data.MessageID)) +} + +// platformMessageID 把回执里 message_id 的**原始 JSON 片段**归一成字符串 id。 +// +// 只接受两种可辨认的形态：裸数字片段（`12345`）与纯数字字符串（`"12345"`）—— 后者是 +// 实现差异，前者是规范形态。其余一切（null、字段缺失、对象 / 数组、浮点、负数、夹带 +// 别的字符、超长）一律返回空串（= 平台没给可用 id），绝不把原始片段当 id 用。 +func platformMessageID(raw json.RawMessage) string { + trimmed := strings.TrimSpace(string(raw)) + if trimmed == "" || trimmed == "null" { + return "" + } + if unquoted, err := strconv.Unquote(trimmed); err == nil { + return digitsOnly(unquoted) + } + return digitsOnly(trimmed) +} + +// digitsOnly 校验候选 id 是**纯十进制数字串**（OneBot 的 message_id 就是这样一个非负 +// 整数）。任何其它字符、空串或超长都说明这段文本不是 id —— 返回空串，绝不修正它。 +func digitsOnly(s string) string { + if s == "" || len(s) > maxPlatformIDLen { + return "" + } + for i := 0; i < len(s); i++ { + if s[i] < '0' || s[i] > '9' { + return "" + } + } + return s +} + +// nextSendEcho 生成 send_msg 的回执关联键，**必须进程内唯一**：等待表按 echo 索引， +// 撞键会让先登记的那次发送拿不到自己的回执（旧实现用 time.Now().UnixNano()，Windows +// 的时钟粒度下一毫秒内的多次发送会落在同一个 tick 上，两次等待互相覆盖）。单调计数器 +// 不依赖时钟精度，与 probe.go 的 nextProbeEcho 同款。 +func (a *Adapter) nextSendEcho() string { + return sendEchoPrefix + strconv.FormatUint(a.sendSeq.Add(1), 10) +} + +// sendOverWSAndAwait 写一条 send_msg 帧并等它的回执。 +// +// 返回 nil 信封**不是错误**：帧已经写出去（投递成败沿用既有判据），只是没等到回执 +// ⇒ 调用方如实报「不知道」。error 只留给**写帧失败**（既有语义：标记断开 + 返回 error）。 +func (a *Adapter) sendOverWSAndAwait(ctx context.Context, conn *wsConn, payload []byte, echo string) (*obAPIResponse, error) { + // 先登记等待者再写帧：回执可能比登记跑得更快，反序会让这一帧变成无人认领的散落 + // 回执（handlePayload 对「空 post_type + echo」的帧只交给等待表，命中不了就丢掉）。 + ch := a.registerActionWait(echo) + defer a.clearActionWait(echo, ch) + if err := conn.WriteMessage(opText, payload); err != nil { + a.setDisconnected() + return nil, fmt.Errorf("qq-obv11: write send_msg: %w", err) + } + timer := time.NewTimer(a.sendReceiptTimeout) + defer timer.Stop() + select { + case resp := <-ch: + return resp, nil + case <-timer.C: + return nil, nil + case <-ctx.Done(): + // 帧已经写出去了：ctx 取消不该把一次**已投递**的消息报成失败（调用方可能重发， + // 群里就收到两条）。只如实标「没等到回执」。 + return nil, nil + } +} + +// logSendReceipt 记一条**结构级 DEBUG**：这一条出站消息到底有没有取到平台 id。 +// +// 「生产里平台到底带不带 id」是下一片（M13 的环形表）能否成立的前提，必须能取证 —— +// 口径与 inbound_debug.go 的入站帧取证相同：只记**状态与 id**，不记正文、不记 token、 +// 不记完整 URL。取不到 id 时照样记（id_status=absent / 空），因为「平台没给」与 +// 「我们没等到」在排障时必须分得开。 +func (a *Adapter) logSendReceipt(ctx context.Context, receipt adapter.SendReceipt) { + if !a.log.Enabled(ctx, slog.LevelDebug) { + return + } + a.log.Debug("send_msg receipt captured", + "id_status", string(receipt.IDStatus), "message_id", receipt.MessageID) +} + +// logEchoFailure 报告一次 send_msg 回执自带的成败（判据与 http_client.post 的 +// retcode/status 判据相同）。 +// +// 回执报错**不**改成返回 error：帧已经写出去了，把「已投递」报成失败会让调用方重发。 +// 但**绝不静默** —— 观察到了一个「发送失败」却一声不吭，正是「数据看到了又丢掉」的 +// 老毛病（本切片要修的就是它）。只记 retcode / status / msg，不记正文。 +func (a *Adapter) logEchoFailure(resp *obAPIResponse) { + if resp == nil { + return + } + if resp.RetCode == 0 && (resp.Status == "" || strings.EqualFold(resp.Status, "ok")) { + return + } + a.log.Warn("send_msg echo reports failure", + "retcode", resp.RetCode, "status", resp.Status, "msg", resp.Msg) +} diff --git a/plugins/channels/qq_obv11/outbound_id_test.go b/plugins/channels/qq_obv11/outbound_id_test.go new file mode 100644 index 0000000..0e0240d --- /dev/null +++ b/plugins/channels/qq_obv11/outbound_id_test.go @@ -0,0 +1,458 @@ +package qq_obv11 + +// 本文件验收 M12 的 A 层（出站消息 id 采集）三态：**带回 / 没带 / 没等到**。 +// 逐条钉死两件事：① 取到就报出来（HTTP 从响应信封的 data.message_id，WS 从 echo 回执）； +// ② **取不到就是空**，绝不编造。另加一条「既有发送路径行为不变」的回归。 + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// fakeSendAPI 是只服务出站 action 的假 OneBot HTTP API：`data` 由用例给定（原样回）， +// 用来驱动「回带 id / 不回带 id」两态。 +type fakeSendAPI struct { + data string +} + +func (f *fakeSendAPI) handler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + data := f.data + if data == "" { + data = "{}" + } + _, _ = w.Write([]byte(`{"status":"ok","retcode":0,"msg":"fake","data":` + data + `}`)) + }) +} + +// fakeOneBotWS 是本地假的 OneBot WS 服务端：记录收到的每一帧，并按 reply 决定是否回执。 +// reply 返回 false 表示**不回**（「没等到」那一态就是这么造出来的）。 +type fakeOneBotWS struct { + mu sync.Mutex + received [][]byte + reply func(action, echo string, params json.RawMessage) ([]byte, bool) +} + +func (f *fakeOneBotWS) count() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.received) +} + +func (f *fakeOneBotWS) handler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgradeWS(w, r) + if err != nil { + return + } + defer func() { _ = conn.Close() }() + for { + _, payload, err := conn.ReadMessage() + if err != nil { + return + } + f.mu.Lock() + f.received = append(f.received, payload) + replyFn := f.reply + f.mu.Unlock() + if replyFn == nil { + continue + } + var req struct { + Action string `json:"action"` + Echo string `json:"echo"` + Params json.RawMessage `json:"params"` + } + if err := json.Unmarshal(payload, &req); err != nil { + continue + } + frame, ok := replyFn(req.Action, req.Echo, req.Params) + if !ok { + continue + } + if err := conn.WriteMessage(opText, frame); err != nil { + return + } + } + }) +} + +// wsEchoFrame 构造一条 OneBot 动作回执帧：**空 post_type + echo**（就是 routeActionResponse +// 认的那种形态），data 原样拼进去。 +func wsEchoFrame(echo, data string) []byte { + return []byte(`{"status":"ok","retcode":0,"data":` + data + `,"echo":"` + echo + `"}`) +} + +// echoOf 从请求体里取 echo，用来把回执发回给**发起它的那一次**发送。 +func echoOf(t *testing.T, payload []byte) string { + t.Helper() + var req struct { + Echo string `json:"echo"` + } + if err := json.Unmarshal(payload, &req); err != nil { + t.Fatalf("decode outbound frame: %v", err) + } + if req.Echo == "" { + t.Fatal("outbound send_msg must carry an echo for receipt correlation") + } + return req.Echo +} + +// newWSReceiptAdapter 起一个 client 模式适配器，指向本地假 OneBot，并把等待回执的上限 +// 收紧到 150ms（真实默认 5s 只覆盖异常，测试里不必真等）。 +func newWSReceiptAdapter(t *testing.T, srv *httptest.Server) *Adapter { + t.Helper() + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + a := NewAdapter(Config{Mode: modeClient, RemoteURL: url}) + a.sendReceiptTimeout = 150 * time.Millisecond + t.Cleanup(func() { _ = a.Disconnect(context.Background()) }) + return a +} + +func groupReply(text string) adapter.UnifiedResponse { + return adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{ + {ChannelType: adapter.ChannelGroup, GroupID: "123", ContentType: adapter.ContentText, Content: text}, + }} +} + +// TestHTTPOutboundIDReported：HTTP 分支从响应信封的 data.message_id 取到 id。 +func TestHTTPOutboundIDReported(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer((&fakeSendAPI{data: `{"message_id":12345}`}).handler()) + defer srv.Close() + + a := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient, APIBaseURL: srv.URL}) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong")) + if err != nil { + t.Fatalf("SendMessageWithReceipt error: %v", err) + } + if len(out.Receipts) != 1 { + t.Fatalf("Receipts len = %d, want 1", len(out.Receipts)) + } + got := out.Receipts[0] + if got.MessageID != "12345" || got.IDStatus != adapter.SendIDReported || !got.Reported() { + t.Errorf("receipt = %+v, want reported 12345", got) + } +} + +// TestHTTPOutboundIDAbsent：回执到了但**没带** id ⇒ 空 id + Absent（可分辨，不编造）。 +func TestHTTPOutboundIDAbsent(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer((&fakeSendAPI{data: `{}`}).handler()) + defer srv.Close() + + a := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient, APIBaseURL: srv.URL}) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong")) + if err != nil { + t.Fatalf("SendMessageWithReceipt error: %v", err) + } + got := out.Receipts[0] + if got.MessageID != "" { + t.Errorf("MessageID = %q, want empty (must never invent one)", got.MessageID) + } + if got.IDStatus != adapter.SendIDAbsent { + t.Errorf("IDStatus = %q, want %q", got.IDStatus, adapter.SendIDAbsent) + } +} + +// TestHTTPOutboundIDAsString：实现对 message_id 回**字符串**时同样要取到（实现之间有出入）。 +func TestHTTPOutboundIDAsString(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer((&fakeSendAPI{data: `{"message_id":"777888"}`}).handler()) + defer srv.Close() + + a := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient, APIBaseURL: srv.URL}) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong")) + if err != nil { + t.Fatalf("SendMessageWithReceipt error: %v", err) + } + if got := out.Receipts[0]; got.MessageID != "777888" || !got.Reported() { + t.Errorf("receipt = %+v, want reported 777888", got) + } +} + +// TestWSOutboundIDReported：WS 分支登记 send_msg 的 echo 等待者，从回执里取出 id。 +func TestWSOutboundIDReported(t *testing.T) { + t.Parallel() + + fake := &fakeOneBotWS{reply: func(action, echo string, _ json.RawMessage) ([]byte, bool) { + if action != "send_msg" { + return nil, false + } + return wsEchoFrame(echo, `{"message_id":9001}`), true + }} + srv := httptest.NewServer(fake.handler()) + defer srv.Close() + + a := newWSReceiptAdapter(t, srv) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + + out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong")) + if err != nil { + t.Fatalf("SendMessageWithReceipt error: %v", err) + } + if got := out.Receipts[0]; got.MessageID != "9001" || !got.Reported() { + t.Errorf("receipt = %+v, want reported 9001", got) + } +} + +// TestWSOutboundIDAbsent：回执到了但没带 id ⇒ 空 id + Absent。 +func TestWSOutboundIDAbsent(t *testing.T) { + t.Parallel() + + fake := &fakeOneBotWS{reply: func(action, echo string, _ json.RawMessage) ([]byte, bool) { + if action != "send_msg" { + return nil, false + } + return wsEchoFrame(echo, `{}`), true + }} + srv := httptest.NewServer(fake.handler()) + defer srv.Close() + + a := newWSReceiptAdapter(t, srv) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + + out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong")) + if err != nil { + t.Fatalf("SendMessageWithReceipt error: %v", err) + } + got := out.Receipts[0] + if got.MessageID != "" { + t.Errorf("MessageID = %q, want empty (must never invent one)", got.MessageID) + } + if got.IDStatus != adapter.SendIDAbsent { + t.Errorf("IDStatus = %q, want %q", got.IDStatus, adapter.SendIDAbsent) + } +} + +// TestWSOutboundIDTimeoutUnknown：对端**根本不回**回执 ⇒ 超时收敛成「不知道」， +// 而且帧确实写出去了（发送成功与「拿不到 id」是两件事）。 +func TestWSOutboundIDTimeoutUnknown(t *testing.T) { + t.Parallel() + + fake := &fakeOneBotWS{} // reply 为 nil：永远不回 + srv := httptest.NewServer(fake.handler()) + defer srv.Close() + + a := newWSReceiptAdapter(t, srv) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + + start := time.Now() + out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong")) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("SendMessageWithReceipt error: %v (写完帧不等于投递失败)", err) + } + if got := out.Receipts[0]; got.MessageID != "" || got.IDStatus != adapter.SendIDUnreported { + t.Errorf("receipt = %+v, want empty id with unreported status", got) + } + if elapsed < a.sendReceiptTimeout { + t.Errorf("returned after %v, want at least the %v receipt timeout", elapsed, a.sendReceiptTimeout) + } + if fake.count() != 1 { + t.Errorf("fake OneBot received %d frames, want 1 (frame must be written regardless)", fake.count()) + } +} + +// TestSendMessageDoesNotAwaitEcho 是**既有发送路径行为不变**的回归：SendMessage 不等 +// 回执（对端不回也立刻返回），帧照旧写出去。等待只发生在显式的凭据路径上。 +func TestSendMessageDoesNotAwaitEcho(t *testing.T) { + t.Parallel() + + fake := &fakeOneBotWS{} + srv := httptest.NewServer(fake.handler()) + defer srv.Close() + + a := newWSReceiptAdapter(t, srv) + // 故意把等待上限拉到 3s：若 SendMessage 误走等待路径，这条用例会明显变慢。 + a.sendReceiptTimeout = 3 * time.Second + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + + start := time.Now() + if err := a.SendMessage(context.Background(), groupReply("pong")); err != nil { + t.Fatalf("SendMessage error: %v", err) + } + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("SendMessage took %v, want no receipt wait on the legacy path", elapsed) + } + if !waitUntil(2*time.Second, func() bool { return fake.count() == 1 }) { + t.Errorf("fake OneBot received %d frames, want the send_msg frame", fake.count()) + } +} + +// TestWSSendReceiptsCorrelateByEcho：并发的多次发送必须各拿各的 id（echo 关联键唯一， +// **不靠时钟**）。假 OneBot 故意**乱序**回执：后发的先回。 +func TestWSSendReceiptsCorrelateByEcho(t *testing.T) { + t.Parallel() + + contents := []string{"m1", "m2", "m3"} + delay := map[string]time.Duration{"m1": 120 * time.Millisecond, "m2": 60 * time.Millisecond, "m3": 0} + fake := &fakeOneBotWS{reply: func(action, echo string, params json.RawMessage) ([]byte, bool) { + if action != "send_msg" { + return nil, false + } + var p struct { + Message []struct { + Data struct { + Text string `json:"text"` + } `json:"data"` + } `json:"message"` + } + if err := json.Unmarshal(params, &p); err != nil || len(p.Message) == 0 { + return nil, false + } + text := p.Message[0].Data.Text + time.Sleep(delay[text]) + return wsEchoFrame(echo, `{"message_id":`+strings.TrimPrefix(text, "m")+`}`), true + }} + srv := httptest.NewServer(fake.handler()) + defer srv.Close() + + a := newWSReceiptAdapter(t, srv) + a.sendReceiptTimeout = 2 * time.Second + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + + var wg sync.WaitGroup + got := make(map[string]string, len(contents)) + var mu sync.Mutex + for _, text := range contents { + wg.Add(1) + go func(text string) { + defer wg.Done() + out, err := a.SendMessageWithReceipt(context.Background(), groupReply(text)) + if err != nil || len(out.Receipts) != 1 { + t.Errorf("send %q: out=%+v err=%v", text, out, err) + return + } + mu.Lock() + got[text] = out.Receipts[0].MessageID + mu.Unlock() + }(text) + } + wg.Wait() + + for _, text := range contents { + want := strings.TrimPrefix(text, "m") + if got[text] != want { + t.Errorf("send %q got id %q, want %q (echo correlation broken)", text, got[text], want) + } + } +} + +// TestPlatformMessageIDForms 钉住 id 的**形态判据**：只有纯十进制数字串能当 id，其余 +// 一律空（= 平台没给可用 id），绝不把原始 JSON 片段当 id 用。 +func TestPlatformMessageIDForms(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + raw string + want string + }{ + {"整数", `12345`, "12345"}, + {"数字字符串", `"12345"`, "12345"}, + {"带前导零的数字串", `"0012"`, "0012"}, + {"null", `null`, ""}, + {"缺失字段", ``, ""}, + {"空字符串", `""`, ""}, + {"浮点", `12.5`, ""}, + {"负数", `-1`, ""}, + {"对象", `{"a":1}`, ""}, + {"数组", `[1]`, ""}, + {"夹带字母", `"12ab"`, ""}, + {"夹带空格", `" 12"`, ""}, + {"超长", `"` + strings.Repeat("9", maxPlatformIDLen+1) + `"`, ""}, + } + for _, tc := range cases { + if got := platformMessageID(json.RawMessage(tc.raw)); got != tc.want { + t.Errorf("platformMessageID(%s) = %q, want %q", tc.name, got, tc.want) + } + } +} + +// TestSendReceiptForensicsLog：出站 id 的采集必须在 **DEBUG 里可取证** —— 「生产里平台 +// 到底带不带 id」是下一片（M13 的环形表）能否成立的前提。日志只许记状态与 id，绝不记正文。 +func TestSendReceiptForensicsLog(t *testing.T) { + t.Parallel() + + buf, log := logCapture() + srv := httptest.NewServer((&fakeSendAPI{data: `{"message_id":4242}`}).handler()) + defer srv.Close() + + a := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient, APIBaseURL: srv.URL, Logger: log}) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + if _, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong")); err != nil { + t.Fatalf("SendMessageWithReceipt error: %v", err) + } + + line := buf.String() + if !strings.Contains(line, `msg="send_msg receipt captured"`) { + t.Errorf("missing send receipt forensics line; log = %s", line) + } + if !strings.Contains(line, "message_id=4242") { + t.Errorf("forensics line must carry the captured id; log = %s", line) + } + if strings.Contains(line, "pong") { + t.Errorf("outbound text must never reach the log; log = %s", line) + } +} + +// TestMessageSentEchoCarriesPlatformID 证明**适配器侧**已经把 message_sent 回声的平台 +// message_id 投影进了结构（SystemNotice.RefMessage）—— 也就是「她自己那条消息的 id」 +// 在归一化之后并未丢失，丢它的是下游的档位判定（见交付说明）。 +func TestMessageSentEchoCarriesPlatformID(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{}) + notice, ok := a.NormalizeSystemNotice(adapter.UnifiedMessage{RawData: []byte(messageSentPrivate)}) + if !ok { + t.Fatal("message_sent must normalize into a system notice (nil ok=false)") + } + if notice.Category != adapter.NoticeCategoryMessageSent { + t.Fatalf("category = %q, want %q", notice.Category, adapter.NoticeCategoryMessageSent) + } + if notice.RefMessage != "77" { + t.Errorf("RefMessage = %q, want 77 (platform message_id of her own echo)", notice.RefMessage) + } +} diff --git a/plugins/channels/qq_obv11/probe.go b/plugins/channels/qq_obv11/probe.go index 70c710a..2f0efba 100644 --- a/plugins/channels/qq_obv11/probe.go +++ b/plugins/channels/qq_obv11/probe.go @@ -242,8 +242,8 @@ func (a *Adapter) wsCall(ctx context.Context, action string, params any) (*obAPI if err != nil { return nil, fmt.Errorf("qq-obv11: encode %s call: %w", action, err) } - ch := a.registerProbeWait(echo) - defer a.clearProbeWait(echo, ch) + ch := a.registerActionWait(echo) + defer a.clearActionWait(echo, ch) if err := conn.WriteMessage(opText, payload); err != nil { return nil, fmt.Errorf("qq-obv11: write %s: %w", action, err) } @@ -262,22 +262,22 @@ func (a *Adapter) nextProbeEcho() string { return probeEchoPrefix + strconv.FormatUint(a.probeSeq.Add(1), 10) } -// registerProbeWait 登记一个待回执的 echo，返回容量 1 的投递通道（写方绝不阻塞）。 -func (a *Adapter) registerProbeWait(echo string) chan *obAPIResponse { +// registerActionWait 登记一个待回执的 echo，返回容量 1 的投递通道（写方绝不阻塞）。 +func (a *Adapter) registerActionWait(echo string) chan *obAPIResponse { ch := make(chan *obAPIResponse, 1) - a.probeMu.Lock() - a.probeWait[echo] = ch - a.probeMu.Unlock() + a.actionMu.Lock() + a.actionWait[echo] = ch + a.actionMu.Unlock() return ch } -// clearProbeWait 注销 echo（等待超时 / 已收到回执）。 -func (a *Adapter) clearProbeWait(echo string, ch chan *obAPIResponse) { - a.probeMu.Lock() - if a.probeWait[echo] == ch { - delete(a.probeWait, echo) +// clearActionWait 注销 echo（等待超时 / 已收到回执）。 +func (a *Adapter) clearActionWait(echo string, ch chan *obAPIResponse) { + a.actionMu.Lock() + if a.actionWait[echo] == ch { + delete(a.actionWait, echo) } - a.probeMu.Unlock() + a.actionMu.Unlock() } // routeActionResponse 尝试把一帧当作 action 回执投递给等待中的调用方。命中返回 true； @@ -290,9 +290,9 @@ func (a *Adapter) routeActionResponse(payload []byte) bool { if err := json.Unmarshal(payload, &head); err != nil || head.Echo == "" || head.PostType != "" { return false } - a.probeMu.Lock() - ch := a.probeWait[head.Echo] - a.probeMu.Unlock() + a.actionMu.Lock() + ch := a.actionWait[head.Echo] + a.actionMu.Unlock() if ch == nil { return false } diff --git a/plugins/channels/qq_obv11/session.go b/plugins/channels/qq_obv11/session.go index e68c546..aa20300 100644 --- a/plugins/channels/qq_obv11/session.go +++ b/plugins/channels/qq_obv11/session.go @@ -30,6 +30,9 @@ func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMes // SendMessage delivers a reply (possibly several messages) to the channel, // subject to the send rate limit. +// +// 它**不等 WS 回执**：既有语义（写完即成功）一个字节都不改 —— 换回平台消息 id 的 +// 调用方走 SendMessageWithReceipt（可选 seam，见 outbound_id.go）。 func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) error { if err := ctx.Err(); err != nil { return fmt.Errorf("qq-obv11: send message: %w", err) @@ -41,16 +44,48 @@ func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) return fmt.Errorf("qq-obv11: send message: %w", adapter.ErrNotConnected) } for i, msg := range resp.Messages { - if err := a.sendMessage(ctx, conn, msg); err != nil { + if _, err := a.sendOne(ctx, conn, msg, false); err != nil { return fmt.Errorf("qq-obv11: send message #%d: %w", i, err) } } return nil } +// SendMessageWithReceipt 实现 adapter.OutboundReceiptAdapter（可选 seam）：投递语义与 +// SendMessage **完全一致**（同一套目标解析 / 限流 / 错误），额外逐条回一条投递凭据 —— +// 平台给这条消息分配的 id，或明确「不知道」（见 adapter.SendIDStatus）。 +// +// 逐条对齐：返回值 Receipts[i] 对应 resp.Messages[i]；中途失败时返回**已投递**的那几条 +// 凭据 + error，调用方按 len(Receipts) 取。 +func (a *Adapter) SendMessageWithReceipt(ctx context.Context, resp adapter.UnifiedResponse) (adapter.SendOutcome, error) { + if err := ctx.Err(); err != nil { + return adapter.SendOutcome{}, fmt.Errorf("qq-obv11: send message: %w", err) + } + conn := a.currentConn() + if conn == nil && a.currentAPI() == nil { + return adapter.SendOutcome{}, fmt.Errorf("qq-obv11: send message: %w", adapter.ErrNotConnected) + } + out := adapter.SendOutcome{Receipts: make([]adapter.SendReceipt, 0, len(resp.Messages))} + for i, msg := range resp.Messages { + receipt, err := a.sendOne(ctx, conn, msg, true) + if err != nil { + return out, fmt.Errorf("qq-obv11: send message #%d: %w", i, err) + } + out.Receipts = append(out.Receipts, receipt) + } + return out, nil +} + +// Compile-time conformance with the optional outbound-receipt seam. +var _ adapter.OutboundReceiptAdapter = (*Adapter)(nil) + // SendProactive pushes a message not triggered by a recent inbound. It is // rate limited (see design/08 section 3: OneBot needs a rate limit to avoid // unusual burst patterns). +// +// 它走**既有**的出站路径，不带凭据：主动推送的平台消息 id 采集不在本切片范围。 +// 下一片（M13）的「她自己最近若干条出站 id」环形表若要覆盖主动推送，需要在这里 +// 也开一条凭据路径（同样的 sendOne + wantReceipt=true）。 func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error { if err := ctx.Err(); err != nil { return fmt.Errorf("qq-obv11: send proactive: %w", err) @@ -59,7 +94,8 @@ func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) if conn == nil && a.currentAPI() == nil { return fmt.Errorf("qq-obv11: send proactive: %w", adapter.ErrNotConnected) } - return a.sendMessage(ctx, conn, msg) + _, err := a.sendOne(ctx, conn, msg, false) + return err } // SetMessageHandler installs the push-style inbound callback. A nil fn clears it. @@ -191,9 +227,9 @@ func (a *Adapter) readConn(ctx context.Context, conn *wsConn) { // 主动 action（见 probe.go），不是上游推来的事件，因此不进事件流水线（但它照样落 // 脱敏结构日志）。载荷解析失败的帧**同样是事件**（decode_error 兜底），照样投递。 func (a *Adapter) handlePayload(ctx context.Context, payload []byte) { - // 先给 action 回执（空 post_type + echo）一个归宿：WS 上的 get_status 等主动 - // action 靠 echo 关联等待表（见 probe.go）。命中即由调用方消费；没有等待者的散落 - // 回执仍落回下面的脱敏结构 DEBUG。 + // 先给 action 回执（空 post_type + echo）一个归宿：WS 上的 get_status 探活 + // （probe.go）与 send_msg 出站回执（outbound_id.go）都靠 echo 关联同一张等待表。 + // 命中即由调用方消费；没有等待者的散落回执仍落回下面的脱敏结构 DEBUG。 if a.routeActionResponse(payload) { return } @@ -247,48 +283,82 @@ func (a *Adapter) dispatch(ctx context.Context, msg adapter.UnifiedMessage) { } } -// sendMessage builds and dispatches a single send_msg request, enforcing the -// rate limit first. 出站通道按 transport 分流：WS 两端把 JSON 帧写到长连接上； -// HTTP 端把同一条 action 请求 POST 给 OneBot 的 HTTP API 并等响应。 -func (a *Adapter) sendMessage(ctx context.Context, conn *wsConn, msg adapter.UnifiedMessage) error { +// sendOne 构造并派发一条 send_msg 请求（先过限流与发送间隔）。出站通道按 transport +// 分流：WS 两端把 JSON 帧写到长连接上；HTTP 端把同一条 action 请求 POST 给 OneBot 的 +// HTTP API 并等响应（见 dispatchSend）。 +// +// wantReceipt **只**决定「要不要把平台消息 id 取回来」： +// - false（既有路径）：WS 写完即返回，一个字节的行为都不改； +// - true（M12 的出站 id seam）：HTTP 从响应信封读 `data.message_id`，WS 登记 echo +// 等待者并等回执。 +// +// 两条路径共用这一个函数，因此目标解析、限流、错误包装不可能各写一份而互相漂移。 +func (a *Adapter) sendOne(ctx context.Context, conn *wsConn, msg adapter.UnifiedMessage, wantReceipt bool) (adapter.SendReceipt, error) { segs, err := outboundSegments(msg) if err != nil { - return err + return adapter.SendReceipt{}, err } messageType, userID, groupID, err := resolveTarget(msg) if err != nil { - return err + return adapter.SendReceipt{}, err } - echo := fmt.Sprintf("echo_%d", time.Now().UnixNano()) + echo := a.nextSendEcho() payload, err := buildSendReq(messageType, userID, groupID, segs, echo) if err != nil { - return err + return adapter.SendReceipt{}, err } if a.cfg.SendInterval > 0 { if err := a.waitSendInterval(ctx); err != nil { - return err + return adapter.SendReceipt{}, err } } if err := a.tryAcquireSend(msg); err != nil { - return err + return adapter.SendReceipt{}, err } - switch api := a.currentAPI(); { - case api != nil: - // HTTP 端：Cyrene→OneBot 走 OneBot 的 HTTP API，等响应判成败（裁决 1.4）。 - // 失败不把适配器标记为断开——HTTP 没有会话可重连，下次发送会重试。 - if err := api.send(ctx, payload); err != nil { - return err - } - default: - if err := conn.WriteMessage(opText, payload); err != nil { - a.setDisconnected() - return fmt.Errorf("qq-obv11: write send_msg: %w", err) - } + receipt, err := a.dispatchSend(ctx, conn, payload, echo, wantReceipt) + if err != nil { + return adapter.SendReceipt{}, err } a.mu.Lock() a.lastSend = time.Now() a.mu.Unlock() - return nil + return receipt, nil +} + +// dispatchSend 把一条已构造好的 send_msg 派发到当前可用的出站通道，并在要的时候 +// 把平台消息 id 取回来。 +func (a *Adapter) dispatchSend(ctx context.Context, conn *wsConn, payload []byte, echo string, wantReceipt bool) (adapter.SendReceipt, error) { + if api := a.currentAPI(); api != nil { + // HTTP 端：Cyrene→OneBot 走 OneBot 的 HTTP API，等响应判成败（裁决 1.4）。 + // 失败不把适配器标记为断开——HTTP 没有会话可重连，下次发送会重试。 + // + // 响应信封本来就等回来了，id 只是**顺手读出来**：不额外多一次请求、不多一次等待。 + resp, err := api.post(ctx, payload) + if err != nil { + return adapter.SendReceipt{}, err + } + if !wantReceipt { + return adapter.SendReceipt{}, nil + } + receipt := receiptFromResponse(resp) + a.logSendReceipt(ctx, receipt) + return receipt, nil + } + if !wantReceipt { + if err := conn.WriteMessage(opText, payload); err != nil { + a.setDisconnected() + return adapter.SendReceipt{}, fmt.Errorf("qq-obv11: write send_msg: %w", err) + } + return adapter.SendReceipt{}, nil + } + resp, err := a.sendOverWSAndAwait(ctx, conn, payload, echo) + if err != nil { + return adapter.SendReceipt{}, err + } + a.logEchoFailure(resp) + receipt := receiptFromResponse(resp) + a.logSendReceipt(ctx, receipt) + return receipt, nil } // waitSendInterval sleeps so consecutive sends are at least SendInterval apart.