# 消息处理流程

> 本文档描述 awada-server 的消息处理流程，awada-server 是对 wechaty 项目的重写，采用 qiweapi 作为底层通信协议。

---

## 一、架构对比

### 1.1 旧架构（wechaty）

```
┌─────────────────────────────────────────────────────────┐
│                    wechaty SDK                          │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐          │
│  │  scan    │    │  login   │    │ message  │          │
│  └──────────┘    └──────────┘    └──────────┘          │
│       ↓              ↓              ↓                   │
│  事件监听回调 → onMessage → 业务处理 → msg.say()        │
└─────────────────────────────────────────────────────────┘
```

**特点**：
- SDK 方式，事件驱动
- 通过 `bot.on('message')` 监听消息
- 直接调用 `msg.say()` 发送消息

### 1.2 新架构（awada-server / qiweapi）

```
┌─────────────────────────────────────────────────────────────────┐
│                    qiweapi HTTP API                              │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐        │
│  │ 设置回调地址  │    │ 消息回调推送  │    │ 发送消息API  │        │
│  └──────────────┘    └──────────────┘    └──────────────┘        │
│         ↓                  ↓                    ↓                 │
│  Webhook接口 → 消息处理服务 → Redis Stream → Bot处理 → 发送API   │
└─────────────────────────────────────────────────────────────────┘
```

**特点**：
- HTTP API + Webhook 回调模式
- 消息通过回调地址推送接收
- 使用 Redis Streams 进行消息队列管理
- 采用 Inbound/Outbound 事件驱动架构

---

## 二、消息接收流程

### 2.1 回调接收（Webhook）

**入口**：`POST /webhook`

```typescript
qiweapi 平台
  ↓ HTTP POST
Webhook 路由 (src/routes/webhook.ts)
  ↓ 解析回调数据
handleRawMessage()
  ↓ 根据 cmd 类型分发
```

**回调类型（cmd）**：
- `11016`: 账号状态变化消息
- `20000`: API异步消息
- `15500`: VX系统消息（好友申请、群成员变动等）
  - `msgType=2357/2132`: 好友申请通知 → 调用 `onFriendApply()`
  - `msgType=1002/1003/1005`: 群成员变动 → TODO
- `15000`: VX普通消息（文本、图片、文件、语音等）

### 2.2 消息解析

**普通消息解析**（cmd=15000）：

```typescript
parseMessage(rawMsg)
  ↓
提取字段：
  - content: 文本内容
  - atList: @列表
  - fromRoomId: 群ID（群消息时）
  - msgType: 消息类型
  - senderId: 发送者ID
  ↓
CallbackMessage 标准格式
```

---

## 三、消息处理流程

### 3.1 处理入口

**文件**：`src/services/message/index.ts`

**函数**：`handleMessage(message: CallbackMessage)`

### 3.2 完整处理流程图

```
┌─────────────────────────────────────────────────────────────┐
│                   消息到达 handleMessage()                    │
└─────────────────────────────────────────────────────────────┘
                            ↓
        ┌───────────────────┴───────────────────┐
        │                                       │
    cmd === 15000?                         其他 cmd
        │                                       │
       是                                      否
        │                                  返回未处理
        ↓
┌───────────────────────────────────────────────────────────────┐
│           检查立即响应的导演指令                                │
│  isImmediateDirectorCommand()                                 │
│    - /ding: 私聊/群聊均可                                     │
│    - /start: 群聊 + @机器人                                   │
│    - /stop: 群聊 + @机器人                                    │
└───────────────────────────────────────────────────────────────┘
        ↓
   是否立即响应?
        │
    ┌───┴───┐
   是      否
    │       │
    │       ↓
    │   ┌─────────────────────────────────────────────┐
    │   │        检查群消息权限                        │
    │   │  - 是否群消息?                              │
    │   │  - 是否@了机器人?                          │
    │   │  - 群是否已开启权限?                        │
    │   └─────────────────────────────────────────────┘
    │           ↓
    │       权限检查通过?
    │           │
    │       ┌───┴───┐
    │      是      否
    │       │       │
    │       │       └─→ 发送权限提示消息，返回
    │       │
    │       ↓
    │   ┌─────────────────────────────────────────────┐
    │   │        转换消息为 Payload                    │
    │   │  - 文本消息: [{type: "text", text: "..."}] │
    │   │  - 多媒体: [{type: "image", ...}, ...]    │
    │   └─────────────────────────────────────────────┘
    │           ↓
    │   ┌─────────────────────────────────────────────┐
    │   │        发布到 Redis Inbound Stream           │
    │   │  - 构建 InboundEvent                        │
    │   │  - 写入 awada:events:inbound:{lane}         │
    │   └─────────────────────────────────────────────┘
    │
    └─→ 处理指令并返回
```

### 3.3 详细处理步骤

#### 步骤 1: 消息类型检查

```typescript
if (message.cmd !== 15000) {
  return { handled: false }; // 只处理普通消息
}
```

#### 步骤 2: 立即响应指令检查

**支持的指令**：
- `/ding`: 测试指令，私聊/群聊均可
- `/start`: 开启群权限（群聊 + @机器人）
- `/stop`: 关闭群权限（群聊 + @机器人）

**处理逻辑**：
```typescript
// /start 指令
if (content === '/start' && message.fromRoomId) {
  1. 调用群详情接口获取群信息
  2. 保存群信息到 room_users.json
  3. 发送响应消息
  4. 返回 handled: true
}

// /stop 指令
if (content === '/stop' && message.fromRoomId) {
  1. 从 room_users.json 移除群
  2. 发送响应消息
  3. 返回 handled: true
}
```

#### 步骤 3: 群权限检查

**检查条件**：
- 必须是群消息（`fromRoomId` 存在且不为 0）
- 必须@了机器人（`atList` 中包含机器人 userId）

**权限判断**：
```typescript
if (isGroupMessage && isMentioned) {
  if (!isRoomEnabled(roomId)) {
    // 未开启权限
    - 发送权限提示消息
    - 返回 handled: true, immediateResponse: 'no_permission'
  }
}
```

**权限数据来源**：
- `room_users.json` 文件中存在的群 = 已开启权限
- 不在文件中的群 = 未开启权限

#### 步骤 4: 消息转换

**文本消息**：
```typescript
payload = [{ type: 'text', text: message.content }]
```

**多媒体消息**（图片、文件、语音）：
```typescript
payload = [
  { type: 'text', text: '...' },      // 可选文本
  { type: 'image', file_url: '...' }, // 图片
  { type: 'file', file_id: '...' },   // 文件
  { type: 'audio', file_id: '...' }   // 语音
]
```

#### 步骤 5: 发布到 Redis

**构建 InboundEvent**：
```typescript
{
  schema_version: 1,
  event_id: "evt_xxx",
  type: "MESSAGE_NEW",
  timestamp: 1234567890,
  meta: {
    platform: "wechat",
    tenant_id: "...",
    channel_id: "...", // 群ID 或 "0"（私聊）
    lane: "linfen",
    user_id_external: "...",
    session_id: "...",
    source_message_id: "..."
  },
  payload: [...] // ContentObject[] 数组
}
```

**发布到 Stream**：
- Stream Key: `awada:events:inbound:{lane}`
- 自动管理 `session_seq`（保证顺序）

---

## 四、Bot 处理流程（下游）

### 4.1 Bot 消费 Inbound Stream

```
Bot 实例
  ↓
订阅 Redis Stream: awada:events:inbound:{lane}
  ↓
XREADGROUP 消费消息
  ↓
幂等检查（event_id）
  ↓
Session 锁 + 序号检查
  ↓
业务处理（AI问答、工具调用等）
  ↓
生成回复消息
  ↓
发布 OutboundEvent 到 Redis
```

### 4.2 OutboundEvent 格式

```typescript
{
  schema_version: 1,
  event_id: "evt_resp_xxx",
  reply_to_event_id: "evt_xxx", // 关联的 Inbound 事件
  type: "REPLY_MESSAGE",
  target: {
    platform: "wechat",
    user_id_external: "...",
    channel_id: "...", // 群ID 或 "0"（私聊）
    conversation_id: "..."
  },
  payload: [
    { type: 'text', text: '...' },
    { type: 'image', file_url: '...' }
  ]
}
```

---

## 五、消息发送流程

### 5.1 Outbound 消费

**文件**：`src/services/outbound/index.ts`

**流程**：
```
Server 订阅: awada:events:outbound:{lane}
  ↓
XREADGROUP 消费 OutboundEvent
  ↓
幂等检查
  ↓
根据 platform 分发
  ↓
按 payload 数组顺序发送消息
```

### 5.2 消息发送顺序

**重要**：`payload` 数组中的消息**必须按顺序发送**

```typescript
for (let i = 0; i < payload.length; i++) {
  const obj = payload[i];
  
  switch (obj.type) {
    case 'text':
      await sendMessage(toId, obj.text, ...);
      break;
    case 'image':
      await sendImageMsg(toId, obj.file_url, ...);
      break;
    case 'file':
      await sendFileMsg(toId, {...}, ...);
      break;
    case 'audio':
      // TODO: 音频发送
      break;
  }
}
```

### 5.3 发送接口映射

| Payload 类型 | qiweapi 接口 | 说明 |
|-------------|-------------|------|
| `text` | `/msg/sendText` | 发送纯文本消息 |
| `image` | `/msg/sendImage` | 发送图片消息（JPG格式） |
| `file` | `/msg/sendFile` | 发送文件消息 |
| `audio` | `/msg/sendVoice` | 发送语音消息（AMR格式） |

---

## 六、权限管理机制

### 6.1 群权限管理

**开启权限**：
- 导演在群中@机器人并发送 `/start`
- 系统调用 `/room/batchGetRoomDetail` 获取群详情
- 保存到 `database/wechatyui/room_users.json`

**关闭权限**：
- 导演在群中@机器人并发送 `/stop`
- 从 `room_users.json` 中移除群信息

**权限检查**：
- 群消息且@了机器人 → 检查群是否在 `room_users.json` 中
- 未开启权限 → 拒绝处理，发送提示消息
- 已开启权限 → 正常处理

### 6.2 私聊权限

- **不受群权限限制**：私聊消息直接处理，无需权限检查
- 用户添加机器人好友后即可私聊问答

### 6.3 导演权限

**导演定义**：
- 配置在 `config.json` 的 `directors` 数组中
- 导演可以发送指令（`/ding`, `/start`, `/stop`）

**指令权限**：
- `/ding`: 私聊/群聊均可
- `/start`, `/stop`: 必须在群聊中且@机器人

### 6.4 好友申请处理

**处理入口**：`src/services/friendship/index.ts`

**触发条件**：
- 系统消息（cmd=15500）
- 消息类型：`SystemMsgType.FRIEND_APPLY` (2357) 或 `SystemMsgType.FRIEND_APPLY_2` (2132)

**处理流程**：

```
┌─────────────────────────────────────────────────────────────┐
│           收到好友申请回调 (FriendApplyCallback)               │
└─────────────────────────────────────────────────────────────┘
                        ↓
        ┌───────────────────────────────┐
        │    检查用户权限                │
        │  - 是否在导演列表中?          │
        │  - 是否在权限群组成员列表中?  │
        └───────────────────────────────┘
                        ↓
            ┌───────────┴───────────┐
           是                      否
            │                       │
            ↓                       ↓
    ┌───────────────┐       ┌───────────────┐
    │ 自动同意申请   │       │ 不自动同意     │
    │ 1. 获取corpId  │       │ 记录日志       │
    │ 2. 调用同意API │       └───────────────┘
    │ 3. 保存打招呼  │
    │    消息(可选)  │
    └───────────────┘
            ↓
    ┌───────────────┐
    │ 发送欢迎语     │
    │ (person_speech│
    │  .welcome)     │
    └───────────────┘
```

**权限检查逻辑**：
```typescript
// 检查用户是否在权限列表中
function hasPermission(userId: string): boolean {
  // 1. 检查是否是导演
  if (directors.includes(userId)) return true;
  
  // 2. 检查是否在权限群组的成员列表中
  const allMemberIds = roomUsers.reduce((acc, entry) => {
    return [...acc, ...entry.room.memberIdList];
  }, []);
  
  return allMemberIds.includes(userId);
}
```

**自动同意流程**：
1. 调用 `checkLogin(guid)` 获取当前登录用户的 `corpId`
2. 调用 `agreeContact(userId, corpId, guid)` 同意好友申请
3. 保存打招呼消息到 `HelloMap`（如果存在）
4. 发送欢迎语：`person_speech.welcome`（从 `config.json` 读取）

**打招呼消息存储**：
- 使用 `HelloMap` 对象存储：`userId -> helloMessage`
- 可通过 `Hello.get(userId)` 获取
- 可通过 `Hello.add(userId, message)` 添加
- 可通过 `Hello.remove(userId)` 移除

**参考实现**：
- wechaty 项目：`service/bot/friendship.ts`
- awada-server：`src/services/friendship/index.ts`

---

## 七、消息类型处理

### 7.1 支持的消息类型

| msgType | 说明 | 处理方式 |
|---------|------|---------|
| 0, 2 | 文本消息 | 直接提取 `content` |
| 7, 14 | 企微图片 | 提取 `fileId` 或 `fileHttpUrl` |
| 101 | 个微图片 | 提取 `fileBigHttpUrl` / `fileMiddleHttpUrl` |
| 15, 20 | 企微文件 | 提取 `fileId` 或 `fileHttpUrl` |
| 102 | 个微文件 | 下载转换为 `cloudUrl` |
| 16 | 语音消息 | 提取 `fileId` 或 `fileHttpUrl` |

### 7.2 消息转换规则

**文本消息**：
```typescript
payload = [{ type: 'text', text: message.content }]
```

**多媒体消息**：
```typescript
payload = [
  { type: 'text', text: '...' },        // 可选
  { type: 'image', file_url: '...' },   // 图片
  { type: 'file', file_id: '...' },     // 文件
  { type: 'audio', file_id: '...' }     // 语音
]
```

**约束**：
- 一个 payload 数组中最多包含 1 条 `text` 类型消息
- 可以包含多个 `file`、`image`、`audio` 类型的消息
- 当存在 `text` 时，必须同时存在至少 1 条 `file` 或 `image` 消息

---

## 八、关键差异对比

### 8.1 wechaty vs awada-server

| 功能点 | wechaty | awada-server |
|--------|---------|--------------|
| **消息接收** | SDK 事件监听 | Webhook HTTP 回调 |
| **消息发送** | `msg.say()` | HTTP API 调用 |
| **群权限** | `WechatyUi.getPermissionRoom()` | `room_users.json` 文件 |
| **@检测** | `msg.mentionSelf()` | 解析 `atList` 数组 |
| **消息队列** | 无（直接处理） | Redis Streams |
| **并发控制** | 无 | Session 锁 + 序号 |
| **消息格式** | wechaty Message 对象 | 标准化的 Payload 数组 |

### 8.2 处理流程差异

**wechaty**：
```
消息到达 → onMessage → 过滤 → 业务处理 → msg.say()
```

**awada-server**：
```
消息到达 → Webhook → 解析 → 权限检查 → 转换 → Redis → Bot处理 → Redis → 发送API
```

---

## 九、数据流图

```
┌─────────────────────────────────────────────────────────────────┐
│                        消息接收层                                 │
│  qiweapi 平台 → Webhook (POST /webhook) → 消息解析              │
└─────────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────────┐
│                        消息处理层                                 │
│  指令检查 → 权限检查 → 消息转换 → 发布到 Redis Inbound          │
└─────────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────────┐
│                      Redis Streams                              │
│  awada:events:inbound:{lane}  ←── Server 写入                   │
│  awada:events:outbound:{lane}  ──→ Server 读取                   │
└─────────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────────┐
│                        Bot 处理层                                │
│  消费 Inbound → 业务处理 → 生成回复 → 发布到 Outbound           │
└─────────────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────────────┐
│                        消息发送层                                 │
│  消费 Outbound → 按顺序发送 → qiweapi 发送消息API                │
└─────────────────────────────────────────────────────────────────┘
```

---

## 十、关键配置

### 10.1 配置文件

**`config/config.json`**：
```json
{
  "directors": ["7881301697926769", "7881302994934588"],
  "room_order": {
    "start": "start",
    "stop": "stop"
  },
  "room_speech": {
    "start": "欢迎使用...",
    "stop": "服务已关闭",
    "no_permission": "请管理员先开启本群服务权限"
  }
}
```

### 10.2 数据文件

**`database/wechatyui/room_users.json`**：
```json
[
  {
    "room": {
      "id": "群ID",
      "memberIdList": ["成员ID1", "成员ID2"]
    },
    "users": [
      {
        "id": "成员ID",
        "name": "成员昵称",
        "roomAlias": "群内备注"
      }
    ]
  }
]
```

---

## 十一、错误处理

### 11.1 消息处理失败

- 记录错误日志
- 不抛出异常，避免影响其他消息处理
- 返回 `handled: false`

### 11.2 发送失败

- 记录错误日志
- 继续发送后续消息（不中断）
- 支持重试机制（通过 Redis Streams 的 Pending 机制）

### 11.3 权限检查失败

- 发送提示消息给用户
- 返回 `handled: true, immediateResponse: 'no_permission'`

---

## 十二、性能优化

### 12.1 懒加载

- EventProducer、ConversationManager 等实例采用懒加载
- 避免模块加载时初始化 Redis（此时 Redis 可能还未初始化）

### 12.2 缓存

- 机器人 userId 缓存（避免重复调用 API）
- Conversation ID 缓存（Redis 存储）

### 12.3 批量处理

- Redis Streams 支持批量消费
- 支持 Pipeline 批量操作

---

## 十三、监控与日志

### 13.1 关键日志点

- 消息接收：`[Webhook] 收到回调`
- 消息处理：`[MessageService] 已发布消息到 Redis`
- 权限检查：`[Message] ⚠️ 群未开启权限`
- 指令处理：`[Message] 处理 /start 指令`
- 消息发送：`[Outbound] ✅ 已完成 N 条消息的发送`

### 13.2 监控指标

- Inbound/Outbound lag（消息延迟）
- Pending 数量（待处理消息）
- 成功/失败率
- 处理耗时 P95/P99

---

**文档版本**：v1.0  
**创建日期**：2025-12-20  
**最后更新**：2025-12-20

