[IM] WeCom WebSocket connecting (bot_id=xxx)...

{
  "scopes": {
    "tenant": [
      "aily:file:read",
      "aily:file:write",
      "application:application.app_message_stats.overview:readonly",
      "application:application:self_manage",
      "application:bot.menu:write",
      "cardkit:card:write",
      "contact:user.employee_id:readonly",
      "corehr:file:download",
      "docs:document.content:read",
      "event:ip_list",
      "im:chat",
      "im:chat.access_event.bot_p2p_chat:read",
      "im:chat.members:bot_access",
      "im:message",
      "im:message.group_at_msg:readonly",
      "im:message.group_msg",
      "im:message.p2p_msg:readonly",
      "im:message:readonly",
      "im:message:send_as_bot",
      "im:resource",
      "sheets:spreadsheet",
      "wiki:wiki:readonly"
    ],
    "user": [
      "aily:file:read",
      "aily:file:write",
      "im:chat.access_event.bot_p2p_chat:read"
    ]
  }
}

[IM] Feishu WebSocket connecting (app_id=xxx)...

[IM] Slack WebSocket connecting...

[IM] Telegram long polling connecting...

curl -X POST "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/setWebhook" \
  -H "Content-Type: application/json" \
  -d '{"url": "<YOUR_CALLBACK_URL>", "secret_token": "<YOUR_SECRET_TOKEN>"}'

[IM] DingTalk Stream connecting...

┌──────────────────────────────────────────────────────────────────────────────┐
│                              IM 集成架构                                     │
│                                                                              │
│   ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│   │ 企业微信  │  │   飞书    │  │  Slack   │  │ Telegram │  │   钉钉   │  IM层│
│   └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘      │
│        │ WH/WS       │ WH/WS      │ WH/WS       │ WH/LP       │ WH/Stream  │
│   ─────┼─────────────┼────────────┼─────────────┼─────────────┼──────────   │
│        ▼             ▼            ▼             ▼             ▼              │
│   ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│   │  WeCom   │  │  Feishu  │  │  Slack   │  │ Telegram │  │ DingTalk │ 适配 │
│   │ Adapter  │  │ Adapter  │  │ Adapter  │  │ Adapter  │  │ Adapter  │ 器层 │
│   └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘      │
│        │             │             │              │             │             │
│   ─────┼─────────────┼─────────────┼──────────────┼─────────────┼────────   │
│        └─────────────┼─────────────┼──────────────┘─────────────┘            │
│                      │                                                       │
│                 ┌────┴────────────┐                                            │
│                 │ Mattermost      │ Webhook-only                               │
│                 └────────┬────────┘                                            │
│                          ▼                                                    │
│                 ┌────────────────┐                                             │
│                 │ Mattermost     │                                             │
│                 │ Adapter        │                                             │
│                 └────────┬────────┘                                             │
│                          │                                                     │
│        ──────────────────┴────────────────────────────────────────────────────│
│                        ▼                                                     │
│   ┌──────────────────────────────────┐                                       │
│   │         im.Service               │     服务编排层                        │
│   │                                  │     · IM 渠道管理 (CRUD)              │
│   │  ┌────────────────────────────┐  │     · Adapter Factory (动态创建)      │
│   │  │ CommandRegistry            │  │     · 斜杠指令分发                    │
│   │  │ qaQueue (Worker Pool)      │  │     · QA 队列调度 (有界, 异步)        │
│   │  │ rateLimiter (滑动窗口)      │  │     · 滑动窗口限流                    │
│   │  │ processedMsgs (去重)       │  │     · 消息去重 (MessageID + TTL)      │
│   │  │ inflight (取消跟踪)         │  │     · 会话映射 (ChannelSession)       │
│   │  └────────────────────────────┘  │     · 流式/全量路由                    │
│   └──────────────┬───────────────────┘                                       │
│                  │                                                           │
│   ───────────────┼───────────────────────────────────────────────────────    │
│                  ▼                                                           │
│   ┌──────────────────────────────────────┐                                   │
│   │     WeKnora Core (QA Pipeline)       │     核心层                       │
│   │   SessionService · MessageService    │                                   │
│   │   TenantService  · AgentService      │                                   │
│   │   KnowledgeService (文件保存)         │                                   │
│   └──────────────────────────────────────┘                                   │
└──────────────────────────────────────────────────────────────────────────────┘

CREATE TABLE im_channels (
    id                VARCHAR(36) PRIMARY KEY,
    tenant_id         BIGINT NOT NULL,
    agent_id          VARCHAR(36) NOT NULL,       -- 绑定的 Agent ID
    platform          VARCHAR(20) NOT NULL,       -- 'wecom' | 'feishu' | 'slack' | 'telegram' | 'dingtalk' | 'mattermost'
    name              VARCHAR(255) NOT NULL DEFAULT '',
    enabled           BOOLEAN NOT NULL DEFAULT true,
    mode              VARCHAR(20) NOT NULL DEFAULT 'websocket',  -- 'webhook' | 'websocket'
    output_mode       VARCHAR(20) NOT NULL DEFAULT 'stream',     -- 'stream' | 'full'
    knowledge_base_id VARCHAR(36),                -- 可选，绑定知识库以接收文件消息
    bot_identity      VARCHAR(255),               -- 计算字段，防止重复机器人
    credentials       JSONB NOT NULL DEFAULT '{}',               -- 平台凭证
    created_at        TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at        TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    deleted_at        TIMESTAMPTZ
);

(im_channel_id, Platform, UserID, ChatID, TenantID)  →  SessionID

type IncomingMessage struct {
    Platform    Platform          // "wecom" | "feishu" | "slack" | "telegram" | "dingtalk" | "mattermost"
    MessageType MessageType       // "text" | "file" | "image"
    UserID      string            // 平台用户标识
    UserName    string            // 显示名 (可选)
    ChatID      string            // 群聊 ID (私聊为空)
    ChatType    ChatType          // "direct" | "group"
    Content     string            // 纯文本内容
    MessageID   string            // 平台消息 ID (用于去重)
    FileKey     string            // 文件标识 (文件/图片消息)
    FileName    string            // 文件名 (文件/图片消息)
    FileSize    int64             // 文件大小 (字节)
    Extra       map[string]string // 平台特有字段 (如 req_id、aes_key)
}

type ReplyMessage struct {
    Content    string            // Markdown 文本
    IsStreaming bool             // 是否为流式块
    IsFinal    bool             // 是否为最后一块
    Extra      map[string]string // 平台特有字段
}

用户在 IM 中发送消息
        │
        ▼
┌─ HTTP Handler / WebSocket 回调 ─────────────────┐
│  1. 根据 channel_id 查找渠道配置                   │
│  2. 获取对应 Adapter                              │
│  3. 签名验证 (VerifyCallback)                     │
│  4. URL 验证处理 (HandleURLVerification)           │
│  5. 解密 + 解析 → IncomingMessage (ParseCallback)  │
│  6. 立即返回 HTTP 200 (异步处理)                    │
└──────────────────────────┬──────────────────────-┘
                           │ goroutine
                           ▼
┌─ im.Service.HandleMessage ──────────────────────┐
│  1. 去重检查 (MessageID, 5 分钟 TTL)             │
│  2. 内容长度校验 (≤ 4096 rune，超出截断)           │
│  3. 斜杠指令检测 → 命中则分发到 CommandRegistry   │
│  4. 限流检查 (滑动窗口, 10次/60s)                 │
│  5. 从渠道配置获取 agent_id、tenant_id            │
│  6. 解析/创建 ChannelSession                     │
│  7. 获取 WeKnora Session                         │
│  8. 加载 Agent 配置（获取知识库、模型等信息）       │
│  9. 文件消息？→ 下载并保存到知识库                  │
│ 10. 提交到 qaQueue (有界队列, 异步执行)            │
└───────────┬─────────────────────────────────────┘
            │
            ▼
┌─ qaQueue Worker ────────────────────────────────┐
│  从队列取出请求，记录 inflight，判断流式/全量模式   │
└───────────┬─────────────────────┬───────────────┘
            │                     │
     流式模式 ▼              全量模式 ▼
┌────────────────────┐  ┌─────────────────────┐
│ handleMessageStream│  │ runQA (阻塞收集完整  │
│                    │  │ 回答后一次性发送)     │
│ · StartStream      │  └─────────────────────┘
│ · EventBus 订阅    │
│ · 300ms 批量刷新   │
│ · 工具事件展示     │
│ · SendStreamChunk  │
│ · EndStream        │
└────────────────────┘
            │
            ▼
    消息持久化 (user + assistant)

渠道创建/更新 (前端 UI)
        │
        ▼
┌─ im.Service ──────────────────────────┐
│  1. 保存渠道配置到数据库                │
│  2. 如果渠道已启用：                    │
│     a. AdapterFactory 创建 Adapter     │
│     b. WebSocket 模式：建立长连接       │
│     c. Webhook 模式：注册回调处理       │
│  3. 维护 channels map (channel_id →    │
│     channelState{Channel, Adapter})    │
└────────────────────────────────────────┘

服务启动时：
  LoadAndStartChannels() → 从 DB 加载所有 enabled 的渠道 → 逐个 StartChannel()

渠道停用/删除时：
  StopChannel() → 取消 Adapter 上下文 → 从 map 移除

type Adapter interface {
    Platform() Platform
    VerifyCallback(c *gin.Context) error
    ParseCallback(c *gin.Context) (*IncomingMessage, error)
    SendReply(ctx context.Context, incoming *IncomingMessage, reply *ReplyMessage) error
    HandleURLVerification(c *gin.Context) bool
}

type StreamSender interface {
    StartStream(ctx context.Context, incoming *IncomingMessage) (streamID string, err error)
    SendStreamChunk(ctx context.Context, incoming *IncomingMessage, streamID string, content string) error
    EndStream(ctx context.Context, incoming *IncomingMessage, streamID string) error
}

type FileDownloader interface {
    DownloadFile(ctx context.Context, msg *IncomingMessage) (io.ReadCloser, string, error)
}

type AdapterFactory func(ctx context.Context, channel *IMChannel, msgHandler func(*IncomingMessage)) (Adapter, CancelFunc, error)

企业微信服务器 ──HTTP POST──▶ /api/v1/im/callback/{channel_id}
                                      │
                              解密 (AES-256-CBC)
                              解析 XML → IncomingMessage
                                      │
                              处理完成后调用 WeCom REST API 回复

LongConnClient ══WebSocket══▶ wss://openws.work.weixin.qq.com
       │
  1. 发送 aibot_subscribe (bot_id + secret)
  2. 接收 aibot_msg_callback 消息帧
  3. 通过 aibot_respond_msg 回复
  4. 每 30s 心跳保活 (ping/pong)
  5. 断连自动重连 (指数退避 1s → 30s)

飞书服务器 ──HTTP POST──▶ /api/v1/im/callback/{channel_id}
                                   │
                           解密 (AES-256-CBC，可选)
                           解析 JSON → IncomingMessage
                                   │
                           通过飞书 Open API 回复

StartStream:
  1. POST /cardkit/v1/cards              → 创建卡片实体 (streaming_mode: true)
  2. POST /im/v1/messages                → 发送卡片消息到聊天

SendStreamChunk:
  3. PUT /cardkit/v1/cards/{id}/elements/{eid}/content  → 更新元素内容 (累积全文)

EndStream:
  4. PATCH /cardkit/v1/cards/{id}/settings  → 设置 streaming_mode: false

> 💭 **思考过程**
> [thinking content line 1]
> [thinking content line 2]

Slack 服务器 ──HTTP POST──▶ /api/v1/im/callback/{channel_id}
                                   │
                           签名验证 (HMAC-SHA256)
                           解析 JSON → IncomingMessage
                                   │
                           通过 Slack Web API 回复

LongConnClient ══WebSocket══▶ wss://wss-primary.slack.com
       │
  1. 使用 App Token 建立连接
  2. 接收 Events API 消息帧
  3. 确认消息 (Ack)
  4. 通过 Slack Web API 回复

StartStream:
  1. POST /chat.postMessage              → 发送初始消息，获取 ts (timestamp)

SendStreamChunk:
  2. POST /chat.update                   → 根据 ts 更新消息内容 (累积全文)

EndStream:
  3. 无需特殊操作

Telegram 服务器 ──HTTP POST──▶ /api/v1/im/callback/{channel_id}
                                   │
                           Secret Token 验证（可选）
                           解析 JSON → IncomingMessage
                                   │
                           通过 Telegram Bot API 回复

LongConnClient ──HTTP POST──▶ https://api.telegram.org/bot<token>/getUpdates
       │
  1. 发送 getUpdates (timeout=30s，长轮询)
  2. 解析返回的 Update 列表
  3. 更新 offset 以确认已处理
  4. 循环继续
  5. 错误时退避 3 秒重试

StartStream:
  1. POST sendMessage               → 发送初始 "正在思考..." 消息，获取 message_id

SendStreamChunk:
  2. POST editMessageText            → 根据 message_id 更新消息内容（累积全文）

EndStream:
  3. POST editMessageText            → 最终更新，启用 Markdown 解析

> 💭 *思考过程*
> [thinking content line 1]
> [thinking content line 2]

钉钉服务器 ──HTTP POST──▶ /api/v1/im/callback/{channel_id}
                                   │
                           HmacSHA256 签名验证
                           解析 JSON → IncomingMessage
                                   │
                           通过 sessionWebhook 或 OpenAPI 回复

LongConnClient ══Stream══▶ 钉钉 Stream 服务
       │
  1. 使用 ClientID + ClientSecret 建立连接
  2. 注册 ChatBot 回调处理
  3. 接收消息事件
  4. SDK 内置重连和心跳

StartStream:
  1. POST /v1.0/card/instances/createAndDeliver  → 创建并投递 AI 卡片

SendStreamChunk:
  2. PUT /v1.0/card/streaming                     → 流式更新卡片内容（累积全文）

EndStream:
  3. PUT /v1.0/card/streaming (isFinalize=true)   → 标记流式结束

> 💭 **思考过程**
> [thinking content line 1]
> [thinking content line 2]

Mattermost 服务器 ──HTTP POST──▶ /api/v1/im/callback/{channel_id}
                                        │
                                校验 body 中 token = outgoing_token
                                解析 JSON 或 x-www-form-urlencoded
                                        │
                                通过 REST API v4 回复（Bearer Bot Token）

StartStream:
  1. POST /api/v4/posts                    → 创建占位帖（如「正在思考...」），得到 post id

SendStreamChunk:
  2. PUT /api/v4/posts/{post_id}/patch     → Patch message 字段（累积全文）

EndStream:
  3. 最后一次 Patch，与 SendStreamChunk 相同逻辑

用户消息 ──▶ HandleMessage
               │
               ├─ 以 "/" 开头？
               │      │
               │      ├─ 已注册指令 → CommandRegistry.Parse → Command.Execute → 回复结果
               │      │                                             │
               │      │                                     ActionClear → 软删除 ChannelSession
               │      │                                     ActionStop  → 取消排队/执行中的 QA
               │      │
               │      └─ LooksLikeCommand() = true 但未注册
               │             → 回复 "未知指令，发送 /help 查看"
               │         LooksLikeCommand() = false (如 "/api/v2/users")
               │             → 当作普通消息，进入 QA 管道
               │
               └─ 普通消息 → 限流检查 → qaQueue → QA 管道

type Command interface {
    Name() string        // 指令名 (不含 "/")
    Description() string // 一行描述，用于 /help 输出
    Execute(ctx context.Context, cmdCtx *CommandContext, args []string) (*CommandResult, error)
}

消息 ──▶ Enqueue ──▶ [ 等待队列 (≤50) ] ──▶ Worker Pool (5 workers) ──▶ QA 管道
              │                                      │
              ├─ 队列已满 → 拒绝并回复提示               │
              ├─ 用户排队超限 (≤3) → 拒绝               ├─ 等待超时 (>60s) → 丢弃并通知
              └─ /stop → Remove(userKey) 取消           └─ 正常执行 QA

QA 管道 ──chunk──chunk──chunk──▶ EventBus
                                    │
                              每 300ms 刷新
                                    │
                        ┌───────────▼───────────┐
                        │ 累积内容 → 完整替换推送 │
                        │ (非增量，每次发送全文)   │
                        └───────────────────────┘

用户发送文件/图片消息
        │
        ▼
  消息类型 = file/image？
  渠道配置了 knowledge_base_id？
  Adapter 实现了 FileDownloader？
        │ 全部满足
        ▼
  1. adapter.DownloadFile(msg) → io.ReadCloser + fileName
  2. 通知用户 "正在处理文件..."
  3. knowledgeService.Save(file, knowledgeBaseID)
  4. 通知用户 "文件已保存到知识库"

package myplatform

type Adapter struct { /* 平台配置 */ }

func (a *Adapter) Platform() im.Platform     { return "myplatform" }
func (a *Adapter) VerifyCallback(c *gin.Context) error { /* 签名验证 */ }
func (a *Adapter) ParseCallback(c *gin.Context) (*im.IncomingMessage, error) { /* 解析消息 */ }
func (a *Adapter) SendReply(ctx context.Context, incoming *im.IncomingMessage, reply *im.ReplyMessage) error { /* 发送回复 */ }
func (a *Adapter) HandleURLVerification(c *gin.Context) bool { /* URL 验证 */ }

imService.RegisterAdapterFactory("myplatform", func(ctx context.Context, channel *im.IMChannel, msgHandler func(*im.IncomingMessage)) (im.Adapter, im.CancelFunc, error) {
    creds := parseCredentials(channel.Credentials)
    appKey := getString(creds, "app_key")
    appSecret := getString(creds, "app_secret")

    adapter := myplatform.NewAdapter(appKey, appSecret)

    // WebSocket 模式需要启动长连接
    if channel.Mode == "websocket" {
        cancelCtx, cancel := context.WithCancel(ctx)
        go adapter.StartLongConn(cancelCtx, msgHandler)
        return adapter, func() { cancel() }, nil
    }

    return adapter, func() {}, nil
})