feat(channel): 出站消息 id 采集(可选 seam + qq_obv11 两分支取回)

「接她的话」缺的 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 同样失败)。
This commit is contained in:
AskaEth committed 2026-10-13 20:38:12 +08:00
1 parent a6f3817061
commit b504df01e0
8 files changed
+1053 -63

No files matched your search

+98 -28
View File
@@ -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.