| 1 | // Package dingtalk 实现钉钉企业内部应用机器人（Stream 模式）适配器。 |
| 2 | // 参考 dsh-dingtalk-channel 的设计： |
| 3 | // - Stream 模式（WebSocket 长连接），无需公网回调地址 |
| 4 | // - @mention gating（群聊） |
| 5 | // - 消息去重（msgId） |
| 6 | // - 会话 webhook 回复（markdown/text） |
| 7 | package dingtalk |
| 8 | |
| 9 | import ( |
| 10 | "bytes" |
| 11 | "context" |
| 12 | "crypto/sha256" |
| 13 | "encoding/hex" |
| 14 | "encoding/json" |
| 15 | "fmt" |
| 16 | "io" |
| 17 | "log/slog" |
| 18 | "net/http" |
| 19 | "net/url" |
| 20 | "os" |
| 21 | "strings" |
| 22 | "sync" |
| 23 | "time" |
| 24 | |
| 25 | "reasonix/internal/bot" |
| 26 | "reasonix/internal/config" |
| 27 | |
| 28 | "github.com/gorilla/websocket" |
| 29 | ) |
| 30 | |
| 31 | const ( |
| 32 | // getTokenURL 获取应用 access_token（旧版接口，GET query 传参）。 |
| 33 | getTokenURL = "https://oapi.dingtalk.com/gettoken" |
| 34 | // gatewayURL 换取 WebSocket 连接端点。 |
| 35 | gatewayURL = "https://api.dingtalk.com/v1.0/gateway/connections/open" |
| 36 | // robotMessagesTopic 机器人消息推送 topic。 |
| 37 | robotMessagesTopic = "/v1.0/im/bot/messages/get" |
| 38 | |
| 39 | // tokenTTL 钉钉 access_token 有效期（秒），提前 60s 刷新。 |
| 40 | tokenTTL = 7200 * time.Second |
| 41 | // connTimeout 建连/HTTP 超时。 |
| 42 | connTimeout = 15 * time.Second |
| 43 | // sendTimeout 单条回复超时。 |
| 44 | sendTimeout = 15 * time.Second |
| 45 | // maxMessageBytes 入站消息上限。 |
| 46 | maxMessageBytes = 1 << 20 |
| 47 | // thinkingEmotion 收到消息时贴的表情（"已收到，处理中"信号，回复完成后撤回）。 |
| 48 | // 与 dsh-dingtalk-channel transport.ts 的 THINKING_EMOTION 一致。 |
| 49 | thinkingEmotion = "🤔思考中" |
| 50 | ) |
| 51 | |
| 52 | // emotionURL 贴/撤机器人消息表情的基址（reply=贴，recall=撤）。包级变量 |
| 53 | // 以便测试覆盖为 httptest 桩地址。 |
| 54 | var emotionURL = "https://api.dingtalk.com/v1.0/robot/emotion" |
| 55 | |
| 56 | // adapter 钉钉适配器实现。 |
| 57 | type adapter struct { |
| 58 | cfg config.DingtalkBotConfig |
| 59 | logger *slog.Logger |
| 60 | msgCh chan bot.InboundMessage |
| 61 | cancel context.CancelFunc |
| 62 | |
| 63 | // httpClient 复用连接，避免每消息重建。 |
| 64 | httpClient *http.Client |
| 65 | // webhookHosts 与 allowHTTPWebhook 属于 adapter，避免测试修改包级安全策略。 |
| 66 | webhookHosts []string |
| 67 | allowHTTPWebhook bool |
| 68 | // tokenMu 保护 token 缓存。 |
| 69 | tokenMu sync.Mutex |
| 70 | token string |
| 71 | tokenAt time.Time |
| 72 | |
| 73 | // seenMu 保护消息去重。 |
| 74 | seenMu sync.Mutex |
| 75 | seen map[string]bool |
| 76 | |
| 77 | // conn 当前 WebSocket 连接。 |
| 78 | connMu sync.Mutex |
| 79 | conn *websocket.Conn |
| 80 | |
| 81 | // webhookMu 保护 chatID→webhook 映射（钉钉无全局发送 API，回复必须 |
| 82 | // POST 到会话 webhook；映射从入站消息学习，DSH 的 sessionWebhooks 模式）。 |
| 83 | webhookMu sync.Mutex |
| 84 | webhooks map[string]string |
| 85 | // lastChatID 最近一次学到 webhook 的会话，供桌面端测试发送使用。 |
| 86 | lastChatID string |
| 87 | |
| 88 | // msgChats 记录 messageID→chatID，供 AddPendingReaction 贴/撤表情时 |
| 89 | // 还原会话（gateway 的 reaction 接口只传 messageID）。 |
| 90 | msgChatsMu sync.Mutex |
| 91 | msgChats map[string]string |
| 92 | } |
| 93 | |
| 94 | // robotMessage 钉钉 Stream 模式机器人消息载荷（/v1.0/im/bot/messages/get）。 |
| 95 | type robotMessage struct { |
| 96 | SenderStaffID string `json:"senderStaffId"` |
| 97 | SenderNick string `json:"senderNick"` |
| 98 | ConversationID string `json:"conversationId"` |
| 99 | ConversationType string `json:"conversationType"` // "1"=单聊 "2"=群聊 |
| 100 | MsgID string `json:"msgId"` |
| 101 | MsgType string `json:"msgtype"` |
| 102 | Text *robotTextContent `json:"text"` |
| 103 | SessionWebhook string `json:"sessionWebhook"` |
| 104 | // IsInAtList 官方回调的结构化 @ 标记：true 表示本消息 @ 了机器人。 |
| 105 | // 正文示例不保留前导 @token，须以该字段为准（见 dingtalk 开放平台文档）。 |
| 106 | IsInAtList bool `json:"isInAtList"` |
| 107 | } |
| 108 | |
| 109 | // robotTextContent 钉钉文本消息内容。 |
| 110 | type robotTextContent struct { |
| 111 | Content string `json:"content"` |
| 112 | } |
| 113 | |
| 114 | // New 创建钉钉适配器。 |
| 115 | func New(cfg config.DingtalkBotConfig, logger *slog.Logger) bot.Adapter { |
| 116 | return &adapter{ |
| 117 | cfg: cfg, |
| 118 | logger: logger.With("platform", "dingtalk"), |
| 119 | seen: make(map[string]bool), |
| 120 | webhooks: make(map[string]string), |
| 121 | msgChats: make(map[string]string), |
| 122 | httpClient: &http.Client{Timeout: connTimeout}, |
| 123 | webhookHosts: append([]string(nil), dingtalkWebhookHosts...), |
| 124 | } |
| 125 | } |
| 126 | |
| 127 | func (a *adapter) Platform() bot.Platform { return bot.PlatformDingtalk } |
| 128 | func (a *adapter) Name() string { return "dingtalk" } |
| 129 | |
| 130 | func (a *adapter) Start(ctx context.Context) error { |
| 131 | a.msgCh = make(chan bot.InboundMessage, 64) |
| 132 | ctx, cancel := context.WithCancel(ctx) |
| 133 | // 同步校验凭据与连接，失败直接报错而非标记 running 后由重连循环静默失败。 |
| 134 | checkCtx, checkCancel := context.WithTimeout(ctx, connTimeout) |
| 135 | defer checkCancel() |
| 136 | if id := a.clientID(); strings.TrimSpace(id) == "" { |
| 137 | cancel() |
| 138 | return fmt.Errorf("dingtalk client_id is not configured") |
| 139 | } |
| 140 | if secret := a.clientSecret(); strings.TrimSpace(secret) == "" { |
| 141 | cancel() |
| 142 | return fmt.Errorf("dingtalk client_secret is not configured") |
| 143 | } |
| 144 | if _, err := a.accessToken(checkCtx); err != nil { |
| 145 | cancel() |
| 146 | return fmt.Errorf("dingtalk credentials rejected: %w", err) |
| 147 | } |
| 148 | conn, err := a.dialConnection(checkCtx) |
| 149 | if err != nil { |
| 150 | cancel() |
| 151 | return fmt.Errorf("dingtalk connection failed: %w", err) |
| 152 | } |
| 153 | a.cancel = cancel |
| 154 | a.setConn(conn) |
| 155 | go a.runWithRetry(ctx, conn) |
| 156 | return nil |
| 157 | } |
| 158 | |
| 159 | func (a *adapter) Stop() error { |
| 160 | if a.cancel != nil { |
| 161 | a.cancel() |
| 162 | } |
| 163 | a.closeConn() |
| 164 | return nil |
| 165 | } |
| 166 | |
| 167 | func (a *adapter) Send(ctx context.Context, msg bot.OutboundMessage) (bot.SendResult, error) { |
| 168 | return a.sendMessage(ctx, msg) |
| 169 | } |
| 170 | |
| 171 | func (a *adapter) SendTyping(ctx context.Context, chatID string) error { |
| 172 | return nil |
| 173 | } |
| 174 | |
| 175 | func (a *adapter) Messages() <-chan bot.InboundMessage { |
| 176 | return a.msgCh |
| 177 | } |
| 178 | |
| 179 | func (a *adapter) closeConn() { |
| 180 | a.connMu.Lock() |
| 181 | defer a.connMu.Unlock() |
| 182 | if a.conn != nil { |
| 183 | _ = a.conn.Close() |
| 184 | a.conn = nil |
| 185 | } |
| 186 | } |
| 187 | |
| 188 | func (a *adapter) setConn(conn *websocket.Conn) { |
| 189 | a.connMu.Lock() |
| 190 | a.conn = conn |
| 191 | a.connMu.Unlock() |
| 192 | } |
| 193 | |
| 194 | func (a *adapter) releaseConn(conn *websocket.Conn) { |
| 195 | a.connMu.Lock() |
| 196 | if a.conn == conn { |
| 197 | a.conn = nil |
| 198 | } |
| 199 | a.connMu.Unlock() |
| 200 | _ = conn.Close() |
| 201 | } |
| 202 | |
| 203 | // clientID 返回配置或环境变量中的 AppKey。 |
| 204 | func (a *adapter) clientID() string { |
| 205 | if v := strings.TrimSpace(a.cfg.ClientID); v != "" { |
| 206 | return v |
| 207 | } |
| 208 | return os.Getenv(a.cfg.ClientIDEnv) |
| 209 | } |
| 210 | |
| 211 | // clientSecret 返回配置或环境变量中的 AppSecret。 |
| 212 | func (a *adapter) clientSecret() string { |
| 213 | if v := strings.TrimSpace(a.cfg.ClientSecret); v != "" { |
| 214 | return v |
| 215 | } |
| 216 | return os.Getenv(a.cfg.SecretEnv) |
| 217 | } |
| 218 | |
| 219 | // accessToken 获取并缓存钉钉应用 access_token。 |
| 220 | func (a *adapter) accessToken(ctx context.Context) (string, error) { |
| 221 | a.tokenMu.Lock() |
| 222 | defer a.tokenMu.Unlock() |
| 223 | if a.token != "" && time.Since(a.tokenAt) < tokenTTL-60*time.Second { |
| 224 | return a.token, nil |
| 225 | } |
| 226 | id := a.clientID() |
| 227 | secret := a.clientSecret() |
| 228 | if id == "" || secret == "" { |
| 229 | return "", fmt.Errorf("dingtalk client_id or client_secret is not configured") |
| 230 | } |
| 231 | req, err := http.NewRequestWithContext(ctx, http.MethodGet, |
| 232 | fmt.Sprintf("%s?appkey=%s&appsecret=%s", getTokenURL, url.QueryEscape(id), url.QueryEscape(secret)), nil) |
| 233 | if err != nil { |
| 234 | return "", err |
| 235 | } |
| 236 | resp, err := a.httpClient.Do(req) |
| 237 | if err != nil { |
| 238 | return "", err |
| 239 | } |
| 240 | defer resp.Body.Close() |
| 241 | var body struct { |
| 242 | AccessToken string `json:"access_token"` |
| 243 | ErrCode int `json:"errcode"` |
| 244 | ErrMsg string `json:"errmsg"` |
| 245 | } |
| 246 | if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&body); err != nil { |
| 247 | return "", err |
| 248 | } |
| 249 | if resp.StatusCode != http.StatusOK || body.AccessToken == "" { |
| 250 | return "", fmt.Errorf("dingtalk gettoken failed: code=%d msg=%s", body.ErrCode, body.ErrMsg) |
| 251 | } |
| 252 | a.token = body.AccessToken |
| 253 | a.tokenAt = time.Now() |
| 254 | return a.token, nil |
| 255 | } |
| 256 | |
| 257 | // gatewayEndpoint 换取 WebSocket 连接端点与 ticket。 |
| 258 | type gatewayEndpoint struct { |
| 259 | Endpoint string `json:"endpoint"` |
| 260 | Ticket string `json:"ticket"` |
| 261 | } |
| 262 | |
| 263 | // openConnection 向网关换取 WebSocket 地址。 |
| 264 | func (a *adapter) openConnection(ctx context.Context) (string, error) { |
| 265 | payload := map[string]any{ |
| 266 | "clientId": a.clientID(), |
| 267 | "clientSecret": a.clientSecret(), |
| 268 | "ua": "reasonix", |
| 269 | "subscriptions": []map[string]string{ |
| 270 | {"type": "CALLBACK", "topic": robotMessagesTopic}, |
| 271 | }, |
| 272 | } |
| 273 | raw, err := json.Marshal(payload) |
| 274 | if err != nil { |
| 275 | return "", err |
| 276 | } |
| 277 | req, err := http.NewRequestWithContext(ctx, http.MethodPost, gatewayURL, bytes.NewReader(raw)) |
| 278 | if err != nil { |
| 279 | return "", err |
| 280 | } |
| 281 | req.Header.Set("Content-Type", "application/json") |
| 282 | req.Header.Set("Accept", "application/json") |
| 283 | resp, err := a.httpClient.Do(req) |
| 284 | if err != nil { |
| 285 | return "", err |
| 286 | } |
| 287 | defer resp.Body.Close() |
| 288 | body, err := io.ReadAll(io.LimitReader(resp.Body, maxMessageBytes)) |
| 289 | if err != nil { |
| 290 | return "", err |
| 291 | } |
| 292 | var ep gatewayEndpoint |
| 293 | if err := json.Unmarshal(body, &ep); err != nil { |
| 294 | return "", fmt.Errorf("dingtalk open connection: bad response: %s", truncate(string(body), 200)) |
| 295 | } |
| 296 | if ep.Endpoint == "" || ep.Ticket == "" { |
| 297 | return "", fmt.Errorf("dingtalk open connection: endpoint/ticket empty: %s", truncate(string(body), 200)) |
| 298 | } |
| 299 | return ep.Endpoint + "?ticket=" + url.QueryEscape(ep.Ticket), nil |
| 300 | } |
| 301 | |
| 302 | // dialConnection 换取一次连接地址并完成 WebSocket 握手。 |
| 303 | func (a *adapter) dialConnection(ctx context.Context) (*websocket.Conn, error) { |
| 304 | wsURL, err := a.openConnection(ctx) |
| 305 | if err != nil { |
| 306 | return nil, err |
| 307 | } |
| 308 | dialer := websocket.Dialer{HandshakeTimeout: connTimeout} |
| 309 | conn, _, err := dialer.DialContext(ctx, wsURL, nil) |
| 310 | return conn, err |
| 311 | } |
| 312 | |
| 313 | // runWithRetry 先处理 Start 已握手的连接，断线后带退避重连。 |
| 314 | func (a *adapter) runWithRetry(ctx context.Context, conn *websocket.Conn) { |
| 315 | backoff := time.Second |
| 316 | for { |
| 317 | if conn == nil { |
| 318 | var err error |
| 319 | conn, err = a.dialConnection(ctx) |
| 320 | if err != nil { |
| 321 | if ctx.Err() != nil { |
| 322 | return |
| 323 | } |
| 324 | a.logger.Warn("dingtalk connection failed; reconnecting", "err", err, "backoff", backoff) |
| 325 | if !waitForRetry(ctx, backoff) { |
| 326 | return |
| 327 | } |
| 328 | if backoff < 30*time.Second { |
| 329 | backoff *= 2 |
| 330 | } |
| 331 | continue |
| 332 | } |
| 333 | if ctx.Err() != nil { |
| 334 | _ = conn.Close() |
| 335 | return |
| 336 | } |
| 337 | a.setConn(conn) |
| 338 | backoff = time.Second |
| 339 | } |
| 340 | if err := a.serveConnection(ctx, conn); err != nil && ctx.Err() == nil { |
| 341 | a.logger.Warn("dingtalk connection closed; reconnecting", "err", err, "backoff", backoff) |
| 342 | } |
| 343 | conn = nil |
| 344 | if ctx.Err() != nil || !waitForRetry(ctx, backoff) { |
| 345 | return |
| 346 | } |
| 347 | if backoff < 30*time.Second { |
| 348 | backoff *= 2 |
| 349 | } |
| 350 | } |
| 351 | } |
| 352 | |
| 353 | func waitForRetry(ctx context.Context, delay time.Duration) bool { |
| 354 | timer := time.NewTimer(delay) |
| 355 | defer timer.Stop() |
| 356 | select { |
| 357 | case <-ctx.Done(): |
| 358 | return false |
| 359 | case <-timer.C: |
| 360 | return true |
| 361 | } |
| 362 | } |
| 363 | |
| 364 | // serveConnection 处理一条已握手连接，直到断开或 ctx 取消。 |
| 365 | func (a *adapter) serveConnection(ctx context.Context, conn *websocket.Conn) error { |
| 366 | defer a.releaseConn(conn) |
| 367 | a.logger.Info("dingtalk stream connected") |
| 368 | |
| 369 | // 读消息循环（处理 SYSTEM/CALLBACK）。 |
| 370 | readErr := make(chan error, 1) |
| 371 | go func() { |
| 372 | for { |
| 373 | if _, data, err := conn.ReadMessage(); err != nil { |
| 374 | readErr <- err |
| 375 | return |
| 376 | } else { |
| 377 | a.handleDownstream(ctx, conn, data) |
| 378 | } |
| 379 | } |
| 380 | }() |
| 381 | select { |
| 382 | case <-ctx.Done(): |
| 383 | return nil |
| 384 | case err := <-readErr: |
| 385 | return err |
| 386 | } |
| 387 | } |
| 388 | |
| 389 | // downstreamFrame 服务端推送帧。 |
| 390 | type downstreamFrame struct { |
| 391 | Type string `json:"type"` // SYSTEM | CALLBACK | EVENT |
| 392 | Headers struct { |
| 393 | Topic string `json:"topic"` |
| 394 | MessageID string `json:"messageId"` |
| 395 | } `json:"headers"` |
| 396 | Data json.RawMessage `json:"data"` |
| 397 | } |
| 398 | |
| 399 | // handleDownstream 处理一帧服务端消息。 |
| 400 | func (a *adapter) handleDownstream(ctx context.Context, conn *websocket.Conn, data []byte) { |
| 401 | var frame downstreamFrame |
| 402 | if err := json.Unmarshal(data, &frame); err != nil { |
| 403 | a.logger.Warn("dingtalk bad frame", "err", err) |
| 404 | return |
| 405 | } |
| 406 | switch frame.Type { |
| 407 | case "SYSTEM": |
| 408 | a.handleSystem(conn, frame) |
| 409 | case "CALLBACK": |
| 410 | a.handleCallback(ctx, conn, frame) |
| 411 | } |
| 412 | } |
| 413 | |
| 414 | // handleSystem 处理 SYSTEM 帧（ping 心跳回执、REGISTERED 等）。 |
| 415 | func (a *adapter) handleSystem(conn *websocket.Conn, frame downstreamFrame) { |
| 416 | switch frame.Headers.Topic { |
| 417 | case "ping": |
| 418 | // 服务端 ping 需原样回包。 |
| 419 | reply, _ := json.Marshal(map[string]any{ |
| 420 | "code": 200, |
| 421 | "headers": frame.Headers, |
| 422 | "message": "OK", |
| 423 | "data": frame.Data, |
| 424 | }) |
| 425 | _ = conn.WriteMessage(websocket.TextMessage, reply) |
| 426 | case "disconnect": |
| 427 | a.logger.Warn("dingtalk server requested disconnect") |
| 428 | _ = conn.Close() |
| 429 | } |
| 430 | } |
| 431 | |
| 432 | // handleCallback 处理机器人消息回调：归一化、去重、入队。 |
| 433 | func (a *adapter) handleCallback(ctx context.Context, conn *websocket.Conn, frame downstreamFrame) { |
| 434 | // 立即回执，避免服务端 60s 后重试。 |
| 435 | ack, _ := json.Marshal(map[string]any{ |
| 436 | "code": 200, |
| 437 | "headers": map[string]string{ |
| 438 | "contentType": "application/json", |
| 439 | "messageId": frame.Headers.MessageID, |
| 440 | }, |
| 441 | "message": "OK", |
| 442 | "data": `{"response":{"status":"SUCCESS"}}`, |
| 443 | }) |
| 444 | _ = conn.WriteMessage(websocket.TextMessage, ack) |
| 445 | |
| 446 | if frame.Headers.Topic != robotMessagesTopic { |
| 447 | return |
| 448 | } |
| 449 | raw, ok := decodeRobotMessage(frame.Data) |
| 450 | if !ok { |
| 451 | a.logger.Warn("dingtalk bad robot message", "err", "cannot decode payload") |
| 452 | return |
| 453 | } |
| 454 | msg := a.normalizeMessage(raw) |
| 455 | if msg == nil { |
| 456 | return |
| 457 | } |
| 458 | select { |
| 459 | case a.msgCh <- *msg: |
| 460 | case <-ctx.Done(): |
| 461 | } |
| 462 | } |
| 463 | |
| 464 | // decodeRobotMessage 解码 Stream 回调的机器人消息载荷。钉钉回调的 data 是 |
| 465 | // JSON 编码的字符串：先解码出字符串再解析为消息对象（与 dsh-dingtalk-channel |
| 466 | // transport.ts 的 JSON.parse(res.data) 一致）；个别网关版本直接下发对象， |
| 467 | // 兼容之。 |
| 468 | func decodeRobotMessage(data json.RawMessage) (robotMessage, bool) { |
| 469 | var raw robotMessage |
| 470 | var rawStr string |
| 471 | if err := json.Unmarshal(data, &rawStr); err == nil { |
| 472 | if err := json.Unmarshal([]byte(rawStr), &raw); err != nil { |
| 473 | return robotMessage{}, false |
| 474 | } |
| 475 | return raw, true |
| 476 | } |
| 477 | if err := json.Unmarshal(data, &raw); err != nil { |
| 478 | return robotMessage{}, false |
| 479 | } |
| 480 | return raw, true |
| 481 | } |
| 482 | |
| 483 | // normalizeMessage 归一化钉钉机器人消息；不满足门控时返回 nil。 |
| 484 | func (a *adapter) normalizeMessage(raw robotMessage) *bot.InboundMessage { |
| 485 | messageID := strings.TrimSpace(raw.MsgID) |
| 486 | chatID := strings.TrimSpace(raw.ConversationID) |
| 487 | userID := strings.TrimSpace(raw.SenderStaffID) |
| 488 | userName := strings.TrimSpace(raw.SenderNick) |
| 489 | if userName == "" { |
| 490 | userName = userID |
| 491 | } |
| 492 | if messageID == "" || chatID == "" { |
| 493 | return nil |
| 494 | } |
| 495 | if a.markSeen(messageID) { |
| 496 | return nil |
| 497 | } |
| 498 | isGroup := strings.TrimSpace(raw.ConversationType) == "2" |
| 499 | text := "" |
| 500 | if raw.Text != nil { |
| 501 | text = strings.TrimSpace(raw.Text.Content) |
| 502 | } |
| 503 | // 群聊 @ 判断：以官方回调的结构化 isInAtList 为准（正文示例不保留 |
| 504 | // 前导 @token）；仅当回调未标记被 @ 时，才回退到文本前导 @ 解析。 |
| 505 | if isGroup { |
| 506 | if a.cfg.RequireMention { |
| 507 | mentioned := raw.IsInAtList |
| 508 | if !mentioned { |
| 509 | mentioned, text = a.splitMention(text) |
| 510 | } else { |
| 511 | // 结构化标记已确认被 @，仅剥离可能残留的前导 @token。 |
| 512 | text = a.stripGroupMention(text) |
| 513 | } |
| 514 | if !mentioned { |
| 515 | a.logger.Info("dingtalk message ignored", "reason", "missing_mention", "chat", logHash(chatID), "message", logHash(messageID)) |
| 516 | return nil |
| 517 | } |
| 518 | } else { |
| 519 | text = a.stripGroupMention(text) |
| 520 | } |
| 521 | } |
| 522 | chatType := bot.ChatDM |
| 523 | if isGroup { |
| 524 | chatType = bot.ChatGroup |
| 525 | } |
| 526 | webhook := strings.TrimSpace(raw.SessionWebhook) |
| 527 | if webhook != "" && a.validDingtalkWebhook(webhook) { |
| 528 | // 记录会话 webhook，供后续回复查表使用；同时记录最近会话供测试发送。 |
| 529 | // 仅记录钉钉官方域名的 webhook，恶意/伪造回调无法注入任意回复目标。 |
| 530 | a.webhookMu.Lock() |
| 531 | a.webhooks[chatID] = webhook |
| 532 | a.lastChatID = chatID |
| 533 | a.webhookMu.Unlock() |
| 534 | } else if webhook != "" { |
| 535 | // 非白名单 webhook 不记录、不透传，防止伪造回调注入任意回复目标。 |
| 536 | a.logger.Warn("dingtalk ignored non-dingtalk session webhook", "chat", logHash(chatID)) |
| 537 | webhook = "" |
| 538 | } |
| 539 | // 记录 messageID→chatID，供 AddPendingReaction 贴/撤表情时还原会话。 |
| 540 | a.msgChatsMu.Lock() |
| 541 | a.msgChats[messageID] = chatID |
| 542 | if len(a.msgChats) > 10000 { |
| 543 | a.msgChats = make(map[string]string) |
| 544 | a.msgChats[messageID] = chatID |
| 545 | } |
| 546 | a.msgChatsMu.Unlock() |
| 547 | return &bot.InboundMessage{ |
| 548 | Platform: bot.PlatformDingtalk, |
| 549 | ChatType: chatType, |
| 550 | ChatID: chatID, |
| 551 | UserID: userID, |
| 552 | UserName: userName, |
| 553 | Text: text, |
| 554 | MessageID: messageID, |
| 555 | SessionWebhook: webhook, |
| 556 | } |
| 557 | } |
| 558 | |
| 559 | // webhookFor 返回某会话最新的回复 webhook；无记录时返回空串。 |
| 560 | func (a *adapter) webhookFor(chatID string) string { |
| 561 | a.webhookMu.Lock() |
| 562 | defer a.webhookMu.Unlock() |
| 563 | return a.webhooks[strings.TrimSpace(chatID)] |
| 564 | } |
| 565 | |
| 566 | // chatForMessageID 返回消息所属会话；无记录时返回空串。 |
| 567 | func (a *adapter) chatForMessageID(messageID string) string { |
| 568 | a.msgChatsMu.Lock() |
| 569 | defer a.msgChatsMu.Unlock() |
| 570 | return a.msgChats[strings.TrimSpace(messageID)] |
| 571 | } |
| 572 | |
| 573 | // AddPendingReaction 在收到的消息上贴 🤔思考中 表情（"已收到，处理中"）， |
| 574 | // 返回的 cleanup 在回合结束后撤回表情。实现 gateway 的 pendingReactionAdapter |
| 575 | // 接口（DSH dsh-dingtalk-channel 的 addEmotion/recallEmotion）。 |
| 576 | func (a *adapter) AddPendingReaction(ctx context.Context, messageID string) (func(), error) { |
| 577 | messageID = strings.TrimSpace(messageID) |
| 578 | chatID := a.chatForMessageID(messageID) |
| 579 | if chatID == "" { |
| 580 | return nil, fmt.Errorf("dingtalk emotion: unknown chat for message %s", logHash(messageID)) |
| 581 | } |
| 582 | if err := a.setEmotion(ctx, chatID, messageID, "reply"); err != nil { |
| 583 | return nil, err |
| 584 | } |
| 585 | return func() { |
| 586 | // 撤回失败仅记录，不阻塞回合收尾。 |
| 587 | recallCtx, cancel := context.WithTimeout(context.Background(), sendTimeout) |
| 588 | defer cancel() |
| 589 | if err := a.setEmotion(recallCtx, chatID, messageID, "recall"); err != nil { |
| 590 | a.logger.Warn("dingtalk recall emotion failed", "chat", logHash(chatID), "message", logHash(messageID), "err", err) |
| 591 | } |
| 592 | }, nil |
| 593 | } |
| 594 | |
| 595 | // emotionBody 构造贴/撤表情的请求体。robotCode 即应用 AppKey（client id）： |
| 596 | // 钉钉将机器人并入应用，client id 就是机器人 code（与 dsh transport.ts 一致）。 |
| 597 | func emotionBody(robotCode, chatID, messageID, action string) map[string]any { |
| 598 | body := map[string]any{ |
| 599 | "robotCode": robotCode, |
| 600 | "openMsgId": messageID, |
| 601 | "openConversationId": chatID, |
| 602 | "emotionType": 2, |
| 603 | "emotionName": thinkingEmotion, |
| 604 | } |
| 605 | if action == "reply" { |
| 606 | body["textEmotion"] = map[string]any{ |
| 607 | "emotionId": "2659900", |
| 608 | "emotionName": thinkingEmotion, |
| 609 | "text": thinkingEmotion, |
| 610 | "backgroundId": "im_bg_1", |
| 611 | } |
| 612 | } |
| 613 | return body |
| 614 | } |
| 615 | |
| 616 | // setEmotion 调钉钉表情接口：action 为 reply（贴）或 recall（撤）。 |
| 617 | func (a *adapter) setEmotion(ctx context.Context, chatID, messageID, action string) error { |
| 618 | robotCode := a.clientID() |
| 619 | if robotCode == "" { |
| 620 | return fmt.Errorf("dingtalk emotion: client_id is not configured") |
| 621 | } |
| 622 | token, err := a.accessToken(ctx) |
| 623 | if err != nil { |
| 624 | return err |
| 625 | } |
| 626 | payload, err := json.Marshal(emotionBody(robotCode, chatID, messageID, action)) |
| 627 | if err != nil { |
| 628 | return err |
| 629 | } |
| 630 | ctx, cancel := context.WithTimeout(ctx, sendTimeout) |
| 631 | defer cancel() |
| 632 | req, err := http.NewRequestWithContext(ctx, http.MethodPost, |
| 633 | emotionURL+"/"+action, bytes.NewReader(payload)) |
| 634 | if err != nil { |
| 635 | return err |
| 636 | } |
| 637 | req.Header.Set("Content-Type", "application/json") |
| 638 | req.Header.Set("x-acs-dingtalk-access-token", token) |
| 639 | resp, err := a.httpClient.Do(req) |
| 640 | if err != nil { |
| 641 | return err |
| 642 | } |
| 643 | defer resp.Body.Close() |
| 644 | body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) |
| 645 | if resp.StatusCode != http.StatusOK { |
| 646 | return fmt.Errorf("dingtalk emotion %s failed (%d): %s", action, resp.StatusCode, truncate(string(body), 300)) |
| 647 | } |
| 648 | return nil |
| 649 | } |
| 650 | |
| 651 | // stripGroupMention 剥离开头的 @xxx 词元。 |
| 652 | func (a *adapter) stripGroupMention(text string) string { |
| 653 | mentionsBot, rest := a.splitMention(text) |
| 654 | if mentionsBot { |
| 655 | return rest |
| 656 | } |
| 657 | return text |
| 658 | } |
| 659 | |
| 660 | // splitMention 检查文本是否以 @机器人 开头；是则返回 (true, 剥离后的文本)。 |
| 661 | func (a *adapter) splitMention(text string) (bool, string) { |
| 662 | text = strings.TrimSpace(text) |
| 663 | if !strings.HasPrefix(text, "@") { |
| 664 | return false, text |
| 665 | } |
| 666 | rest := text[1:] |
| 667 | end := strings.IndexAny(rest, " \t\n") |
| 668 | name := rest |
| 669 | if end >= 0 { |
| 670 | name = rest[:end] |
| 671 | } |
| 672 | botName := strings.TrimSpace(a.cfg.BotName) |
| 673 | if botName == "" { |
| 674 | // 未配置昵称：任意 @ 开头都视为 @ 机器人（与飞书旧行为一致）。 |
| 675 | if end < 0 { |
| 676 | return true, "" |
| 677 | } |
| 678 | return true, strings.TrimSpace(rest[end:]) |
| 679 | } |
| 680 | if name != botName { |
| 681 | return false, text |
| 682 | } |
| 683 | if end < 0 { |
| 684 | return true, "" |
| 685 | } |
| 686 | return true, strings.TrimSpace(rest[end:]) |
| 687 | } |
| 688 | |
| 689 | // markSeen 消息去重；返回 true 表示已见过。 |
| 690 | func (a *adapter) markSeen(messageID string) bool { |
| 691 | a.seenMu.Lock() |
| 692 | defer a.seenMu.Unlock() |
| 693 | if a.seen[messageID] { |
| 694 | return true |
| 695 | } |
| 696 | if len(a.seen) > 10000 { |
| 697 | a.seen = make(map[string]bool) |
| 698 | } |
| 699 | a.seen[messageID] = true |
| 700 | return false |
| 701 | } |
| 702 | |
| 703 | // sendMessage 发送出站消息（文本或 markdown）。 |
| 704 | // 钉钉无全局发送 API：回复必须 POST 到会话 webhook。webhook 来源： |
| 705 | // 1) msg.SessionWebhook（入站消息透传，含持久化恢复场景） |
| 706 | // 2) chatID→webhook 映射表（入站消息学习而来） |
| 707 | // 两者皆无时拒绝并给出可读错误。 |
| 708 | // 注意：gateway 的 sendText 会把 ReplyToMsgID 填成入站消息 ID（非 URL）， |
| 709 | // 这里完全不把 ReplyToMsgID 当 URL 使用——回复目标只来自钉钉官方回调携带 |
| 710 | // 的 sessionWebhook。所有 webhook 必须通过 validDingtalkWebhook 校验 |
| 711 | // （仅钉钉域名），且回复 POST 不携带 access token（官方文档：会话 webhook |
| 712 | // 无需额外认证），避免向任意 URL 泄漏 token 或形成 SSRF。 |
| 713 | func (a *adapter) sendMessage(ctx context.Context, msg bot.OutboundMessage) (bot.SendResult, error) { |
| 714 | webhook := strings.TrimSpace(msg.SessionWebhook) |
| 715 | if webhook != "" && !a.validDingtalkWebhook(webhook) { |
| 716 | return bot.SendResult{}, fmt.Errorf("dingtalk send rejected: session webhook %q is not a dingtalk endpoint", truncate(webhook, 64)) |
| 717 | } |
| 718 | if webhook == "" { |
| 719 | webhook = a.webhookFor(msg.ChatID) |
| 720 | } |
| 721 | if webhook == "" { |
| 722 | return bot.SendResult{}, fmt.Errorf("dingtalk send requires a session webhook: no webhook known for chat %s (bot must receive a message in this chat first)", logHash(msg.ChatID)) |
| 723 | } |
| 724 | if !a.validDingtalkWebhook(webhook) { |
| 725 | return bot.SendResult{}, fmt.Errorf("dingtalk send rejected: webhook %q is not a dingtalk endpoint", truncate(webhook, 64)) |
| 726 | } |
| 727 | // 一律以 markdown 类型发送：Reasonix bot 回复是 markdown 文本，钉钉 |
| 728 | // text 类型按纯文本显示、不渲染语法（与飞书 buildMarkdownCard 一致）。 |
| 729 | title := "Reasonix" |
| 730 | if msg.Card != nil && strings.TrimSpace(msg.Card.Header) != "" { |
| 731 | title = msg.Card.Header |
| 732 | } |
| 733 | payload := map[string]any{"msgtype": "markdown", "markdown": map[string]string{ |
| 734 | "title": title, |
| 735 | "text": msg.Text, |
| 736 | }} |
| 737 | raw, err := json.Marshal(payload) |
| 738 | if err != nil { |
| 739 | return bot.SendResult{}, err |
| 740 | } |
| 741 | ctx, cancel := context.WithTimeout(ctx, sendTimeout) |
| 742 | defer cancel() |
| 743 | req, err := http.NewRequestWithContext(ctx, http.MethodPost, webhook, bytes.NewReader(raw)) |
| 744 | if err != nil { |
| 745 | return bot.SendResult{}, err |
| 746 | } |
| 747 | req.Header.Set("Content-Type", "application/json") |
| 748 | resp, err := a.webhookClient().Do(req) |
| 749 | if err != nil { |
| 750 | return bot.SendResult{}, err |
| 751 | } |
| 752 | defer resp.Body.Close() |
| 753 | body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) |
| 754 | if resp.StatusCode != http.StatusOK { |
| 755 | return bot.SendResult{}, fmt.Errorf("dingtalk send failed (%d): %s", resp.StatusCode, truncate(string(body), 300)) |
| 756 | } |
| 757 | return bot.SendResult{MessageID: ""}, nil |
| 758 | } |
| 759 | |
| 760 | // TestSend 向最近一个交互过的会话发送测试消息，验证凭据与发送链路。 |
| 761 | // 钉钉无全局发送 API，必须先收到过该会话的消息（学到 webhook）才能回复。 |
| 762 | func (a *adapter) TestSend(ctx context.Context, text string) (bot.SendResult, error) { |
| 763 | a.webhookMu.Lock() |
| 764 | chatID := a.lastChatID |
| 765 | a.webhookMu.Unlock() |
| 766 | if chatID == "" { |
| 767 | return bot.SendResult{}, fmt.Errorf("dingtalk test send requires a known chat: the bot must have received a message first") |
| 768 | } |
| 769 | return a.sendMessage(ctx, bot.OutboundMessage{ChatID: chatID, ChatType: bot.ChatDM, Text: text}) |
| 770 | } |
| 771 | |
| 772 | func truncate(s string, n int) string { |
| 773 | runes := []rune(s) |
| 774 | if len(runes) <= n { |
| 775 | return s |
| 776 | } |
| 777 | return string(runes[:n]) + "…" |
| 778 | } |
| 779 | |
| 780 | // logHash 对 chat/消息 ID 做脱敏哈希，用于日志（不泄漏完整 ID）。 |
| 781 | func logHash(id string) string { |
| 782 | if id == "" { |
| 783 | return "" |
| 784 | } |
| 785 | sum := sha256.Sum256([]byte(id)) |
| 786 | return hex.EncodeToString(sum[:])[:12] |
| 787 | } |
| 788 |