AskaEth
ba17820040
- 适配器不再整类丢弃：notice/request/meta_event/message_sent/未知 post_type/JSON 解析失败( decode_error ) 一律归一化投递并留痕
- notice 映射补 group_ban/sub_type 全枚举，未知类型走 NoticeUnknown 兜底（fail-safe 只登记、绝不猜、绝不放行）
- 不投影 flag/comment/file/duration；归一化 DEBUG 只记结构（脱敏，变异实测可打红）
- 传输层端到端已覆盖（真实 http+server 上报路径：加好友请求/心跳/未知类型）
- ⚠️ 主人 2026-10-05 决定「先不连 QQ」（风控）⇒ 本片为通用红线（绝不静默丢弃）+ 现成证据链，QQ 侧后续增量工作暂停
400 lines
12 KiB
Go
400 lines
12 KiB
Go
package qq_obv11
|
||
|
||
// This file holds the message-flow half of the Adapter: the inbound read loop
|
||
// (server/client mode sessions, reconnect, dispatch) and the outbound send path
|
||
// (rate limit, chunk to the connection). It is split out of adapter.go to keep
|
||
// each file under the project's 500-line limit.
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||
)
|
||
|
||
// ReadMessages registers emit as the pull-based sink and blocks until ctx is
|
||
// canceled. Inbound events arrive via the read loop and are dispatched to
|
||
// SetMessageHandler (push) when set, otherwise to emit.
|
||
func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMessage)) {
|
||
a.emitMu.Lock()
|
||
a.emit = emit
|
||
a.emitMu.Unlock()
|
||
defer func() {
|
||
a.emitMu.Lock()
|
||
a.emit = nil
|
||
a.emitMu.Unlock()
|
||
}()
|
||
<-ctx.Done()
|
||
}
|
||
|
||
// SendMessage delivers a reply (possibly several messages) to the channel,
|
||
// subject to the send rate limit.
|
||
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)
|
||
}
|
||
// HTTP 传输没有 ws 连接：只要出站 HTTP API 客户端已装配（http+client 或
|
||
// http+server 配了 api_base_url）就照样能回消息。
|
||
conn := a.currentConn()
|
||
if conn == nil && a.currentAPI() == nil {
|
||
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 {
|
||
return fmt.Errorf("qq-obv11: send message #%d: %w", i, err)
|
||
}
|
||
}
|
||
return 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).
|
||
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)
|
||
}
|
||
conn := a.currentConn()
|
||
if conn == nil && a.currentAPI() == nil {
|
||
return fmt.Errorf("qq-obv11: send proactive: %w", adapter.ErrNotConnected)
|
||
}
|
||
return a.sendMessage(ctx, conn, msg)
|
||
}
|
||
|
||
// SetMessageHandler installs the push-style inbound callback. A nil fn clears it.
|
||
func (a *Adapter) SetMessageHandler(fn adapter.MessageHandler) {
|
||
a.handlerMu.Lock()
|
||
a.handler = fn
|
||
a.handlerMu.Unlock()
|
||
}
|
||
|
||
// Capabilities declares what the QQ OneBot v11 adapter supports.
|
||
func (a *Adapter) Capabilities() adapter.Capabilities {
|
||
return adapter.Capabilities{
|
||
MaxMessageLength: defaultMaxMessageLength,
|
||
Markdown: true,
|
||
Image: true,
|
||
Voice: true,
|
||
Reaction: false,
|
||
TypingHint: false,
|
||
RecommendBurstMax: defaultRecommendBurstMax,
|
||
Proactive: true,
|
||
}.Normalize()
|
||
}
|
||
|
||
// ListChannels returns the groups the adapter currently knows about (learned from
|
||
// inbound events). 全量好友/群列表要靠 OneBot 的 get_group_list / get_friend_list
|
||
// 动作，HTTP 出站客户端（api_base_url）现已具备，但把它接进本方法属后续工作，
|
||
// 本次四端改造不扩大范围。
|
||
func (a *Adapter) ListChannels(ctx context.Context) ([]adapter.ChannelInfo, error) {
|
||
a.groupMu.RLock()
|
||
defer a.groupMu.RUnlock()
|
||
out := make([]adapter.ChannelInfo, 0, len(a.groups))
|
||
for id, name := range a.groups {
|
||
out = append(out, adapter.ChannelInfo{
|
||
ChannelID: id,
|
||
ChannelType: adapter.ChannelGroup,
|
||
ChannelName: name,
|
||
GroupID: id,
|
||
GroupName: name,
|
||
})
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// Addr returns the bound listen address in server mode ("" when not serving).
|
||
func (a *Adapter) Addr() string {
|
||
a.mu.RLock()
|
||
defer a.mu.RUnlock()
|
||
if a.ln == nil {
|
||
return ""
|
||
}
|
||
return a.ln.Addr().String()
|
||
}
|
||
|
||
// readLoop is the single inbound-reader goroutine. It reads frames from the
|
||
// active connection, dispatching message events; in client mode it connects and
|
||
// reconnects after drops (exponential backoff, see reconnect.go), in server mode
|
||
// it waits for a new OneBot client.
|
||
func (a *Adapter) readLoop(ctx context.Context) {
|
||
for {
|
||
conn := a.currentConn()
|
||
if conn == nil {
|
||
if a.cfg.Mode == modeClient {
|
||
// ws+client：首连失败与断线重连**共用这一条路径**——Connect 首连失败
|
||
// 只是记一条 WARN 就返回（不致命），真正的重试在这里按退避进行。
|
||
// 先等待、再拨号：日志里承诺的 retryIn 就是这次等待。等待结束后再查一次
|
||
// ctx：与 Disconnect 赛跑时宁可少拨一次，也不发起注定失败的拨号。
|
||
if !sleepCtx(ctx, a.reconnect.wait(a.policy)) || ctx.Err() != nil {
|
||
return
|
||
}
|
||
if err := a.connectClient(ctx); err != nil {
|
||
a.noteConnectFailure(err)
|
||
continue
|
||
}
|
||
a.reconnect.reset()
|
||
} else if !a.waitConn(ctx) {
|
||
return
|
||
}
|
||
conn = a.currentConn()
|
||
}
|
||
a.readConn(ctx, conn)
|
||
if ctx.Err() != nil {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// readConn reads frames from conn until it closes, the context is canceled, or
|
||
// a close frame is received, marking the adapter disconnected on exit.
|
||
func (a *Adapter) readConn(ctx context.Context, conn *wsConn) {
|
||
watchDone := make(chan struct{})
|
||
go func() {
|
||
select {
|
||
case <-ctx.Done():
|
||
_ = conn.Close()
|
||
case <-watchDone:
|
||
}
|
||
}()
|
||
defer close(watchDone)
|
||
|
||
for {
|
||
opcode, payload, err := conn.ReadMessage()
|
||
if err != nil {
|
||
a.setDisconnected()
|
||
return
|
||
}
|
||
switch opcode {
|
||
case opPing:
|
||
_ = conn.WriteMessage(opPong, payload)
|
||
case opText, opBinary:
|
||
a.handlePayload(ctx, payload)
|
||
case opClose:
|
||
_ = conn.WriteMessage(opClose, nil)
|
||
a.setDisconnected()
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// handlePayload parses a text frame, converting message events into normalized
|
||
// UnifiedMessage and dispatching them. **消息之外的一切上报同样被归一化后投递**
|
||
// （design/26 §1.1 / §3.1：notice / request / meta_event / message_sent 与未知
|
||
// post_type 一律不许静默丢弃）——它们带 ContentType=ContentAction 的结构化标记，
|
||
// 由 channelbridge 的「传入前审查」决定是否 / 如何进入她的上下文，并由留存层落库计数
|
||
// （design/26 §5）。丢弃前一律留一条**脱敏结构 DEBUG**（logInboundFrame），成功投递的
|
||
// 另加一条归一化结果 DEBUG（logNormalizedEvent）：上游到底有没有把这一帧推过来，
|
||
// 是排障时最先要回答的问题，而「被丢弃」与「从未到达」在日志里必须能分开。
|
||
//
|
||
// 唯一的例外是**我们自己主动 action 的回执**（空 post_type + echo）：它交给等待中的
|
||
// 主动 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。
|
||
if a.routeActionResponse(payload) {
|
||
return
|
||
}
|
||
ev, err := parseEvent(payload)
|
||
if err != nil {
|
||
a.log.Debug("undecodable inbound frame",
|
||
"err", err, "bytes", len(payload))
|
||
a.dispatch(ctx, decodeErrorMessage(payload))
|
||
return
|
||
}
|
||
a.logInboundFrame(ev)
|
||
if isActionReceipt(ev) {
|
||
return
|
||
}
|
||
if ev.PostType == postTypeMessage {
|
||
msg := a.toUnified(ev, payload)
|
||
if err := msg.Validate(); err != nil {
|
||
a.log.Warn("invalid unified message dropped", "err", err)
|
||
return
|
||
}
|
||
a.dispatch(ctx, msg)
|
||
return
|
||
}
|
||
a.logNormalizedEvent(ctx, ev)
|
||
msg := a.toEventMessage(ev, payload)
|
||
if err := msg.Validate(); err != nil {
|
||
a.log.Warn("invalid non-message event dropped", "post_type", ev.PostType, "err", err)
|
||
return
|
||
}
|
||
a.dispatch(ctx, msg)
|
||
}
|
||
|
||
// dispatch routes an inbound message to the registered push handler, falling
|
||
// back to the ReadMessages sink. 附件字节在**路由之前**内联（切片 IM-1）：WS 两端
|
||
// 与 HTTP 上报三条入站路径都经这里，下游因此永远拿到「已带字节 / 明确没有字节」的
|
||
// 消息；失败只留日志，绝不挡住投递。
|
||
func (a *Adapter) dispatch(ctx context.Context, msg adapter.UnifiedMessage) {
|
||
a.inlineInboundMedia(ctx, &msg)
|
||
a.handlerMu.RLock()
|
||
fn := a.handler
|
||
a.handlerMu.RUnlock()
|
||
if fn != nil {
|
||
fn(ctx, msg)
|
||
return
|
||
}
|
||
a.emitMu.Lock()
|
||
emit := a.emit
|
||
a.emitMu.Unlock()
|
||
if emit != nil {
|
||
emit(msg)
|
||
}
|
||
}
|
||
|
||
// 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 {
|
||
segs, err := outboundSegments(msg)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
messageType, userID, groupID, err := resolveTarget(msg)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
echo := fmt.Sprintf("echo_%d", time.Now().UnixNano())
|
||
payload, err := buildSendReq(messageType, userID, groupID, segs, echo)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if a.cfg.SendInterval > 0 {
|
||
if err := a.waitSendInterval(ctx); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if err := a.tryAcquireSend(msg); err != nil {
|
||
return 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)
|
||
}
|
||
}
|
||
a.mu.Lock()
|
||
a.lastSend = time.Now()
|
||
a.mu.Unlock()
|
||
return nil
|
||
}
|
||
|
||
// waitSendInterval sleeps so consecutive sends are at least SendInterval apart.
|
||
func (a *Adapter) waitSendInterval(ctx context.Context) error {
|
||
for {
|
||
a.mu.RLock()
|
||
last := a.lastSend
|
||
a.mu.RUnlock()
|
||
if last.IsZero() {
|
||
return nil
|
||
}
|
||
wait := a.cfg.SendInterval - time.Since(last)
|
||
if wait <= 0 {
|
||
return nil
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
return fmt.Errorf("qq-obv11: send interval: %w", ctx.Err())
|
||
case <-time.After(wait):
|
||
}
|
||
}
|
||
}
|
||
|
||
// tryAcquireSend 检查本条出站消息所属会话（群聊/私聊）的发送滑窗，额度用尽时返回
|
||
// ErrRateLimited。额度按会话独立计数（裁决 06.4.3）：给 A 群发消息不占用 B 群或私聊
|
||
// 的额度，键口径与入站滑窗一致（adapter.ConversationKey）。适配器仍保留账号级的
|
||
// SendInterval 发送间隔（更保守，只收紧不放宽）。
|
||
func (a *Adapter) tryAcquireSend(msg adapter.UnifiedMessage) error {
|
||
if !a.limiter.allow(adapter.ConversationKey(msg), time.Now()) {
|
||
return fmt.Errorf("qq-obv11: %w", ErrRateLimited)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// setConn installs a new active connection, closing any previous one and waking
|
||
// the read loop (server mode).
|
||
func (a *Adapter) setConn(conn *wsConn) {
|
||
a.mu.Lock()
|
||
old := a.conn
|
||
a.conn = conn
|
||
a.connected = true
|
||
a.mu.Unlock()
|
||
if old != nil && old != conn {
|
||
_ = old.Close()
|
||
}
|
||
select {
|
||
case a.newConn <- struct{}{}:
|
||
default:
|
||
}
|
||
}
|
||
|
||
// setDisconnected marks the session down, dropping the current connection.
|
||
func (a *Adapter) setDisconnected() {
|
||
a.mu.Lock()
|
||
if a.conn != nil {
|
||
_ = a.conn.Close()
|
||
a.conn = nil
|
||
}
|
||
a.connected = false
|
||
a.mu.Unlock()
|
||
}
|
||
|
||
// currentConn returns the active wsConn, or nil.
|
||
func (a *Adapter) currentConn() *wsConn {
|
||
a.mu.RLock()
|
||
defer a.mu.RUnlock()
|
||
return a.conn
|
||
}
|
||
|
||
// waitConn blocks in server mode until a OneBot client connects or ctx ends.
|
||
func (a *Adapter) waitConn(ctx context.Context) bool {
|
||
for a.currentConn() == nil {
|
||
select {
|
||
case <-a.newConn:
|
||
case <-ctx.Done():
|
||
return false
|
||
case <-time.After(time.Second):
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
func (a *Adapter) rememberGroup(id, name string) {
|
||
if id == "" {
|
||
return
|
||
}
|
||
a.groupMu.Lock()
|
||
a.groups[id] = name
|
||
a.groupMu.Unlock()
|
||
}
|
||
|
||
func (a *Adapter) groupName(id string) string {
|
||
if id == "" {
|
||
return ""
|
||
}
|
||
a.groupMu.RLock()
|
||
defer a.groupMu.RUnlock()
|
||
return a.groups[id]
|
||
}
|
||
|
||
func sleepCtx(ctx context.Context, d time.Duration) bool {
|
||
select {
|
||
case <-ctx.Done():
|
||
return false
|
||
case <-time.After(d):
|
||
return true
|
||
}
|
||
}
|