feat(dingtalk): add proactive messaging support with OpenAPI fallback
- Add access token management with auto-refresh (every 5 minutes) - Implement proactive messaging via DingTalk OpenAPI: - Single chat: /v1.0/robot/oToMessages/batchSend - Group chat: /v1.0/robot/groupMessages/send - Add fallback from session_webhook to proactive API - Enable proactive messaging without requiring prior user interaction - Add unit tests for token refresh and proactive send - Update documentation with API references and usage examples This enables heartbeat notifications, device alerts, and other proactive message scenarios even when session_webhook expires. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
fishtrees
parent
cff85cfe5c
commit
81e9f0b77a
3 changed files with 747 additions and 28 deletions
|
|
@ -11,7 +11,11 @@
|
|||
"enabled": true,
|
||||
"client_id": "YOUR_CLIENT_ID",
|
||||
"client_secret": "YOUR_CLIENT_SECRET",
|
||||
"allow_from": []
|
||||
"allow_from": [],
|
||||
"group_trigger": {
|
||||
"mention_only": false,
|
||||
"prefixes": []
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -20,14 +24,105 @@
|
|||
| 字段 | 类型 | 必填 | 描述 |
|
||||
| ------------- | ------ | ---- | -------------------------------- |
|
||||
| enabled | bool | 是 | 是否启用钉钉频道 |
|
||||
| client_id | string | 是 | 钉钉应用的 Client ID |
|
||||
| client_secret | string | 是 | 钉钉应用的 Client Secret |
|
||||
| client_id | string | 是 | 钉钉应用的 AppKey（也作为 robotCode 使用） |
|
||||
| client_secret | string | 是 | 钉钉应用的 AppSecret |
|
||||
| allow_from | array | 否 | 用户ID白名单，空表示允许所有用户 |
|
||||
| group_trigger | object | 否 | 群聊触发配置 |
|
||||
|
||||
## 设置流程
|
||||
|
||||
1. 前往 [钉钉开放平台](https://open.dingtalk.com/)
|
||||
2. 创建一个企业内部应用
|
||||
3. 从应用设置中获取 Client ID 和 Client Secret
|
||||
4. 配置OAuth和事件订阅(如需要)
|
||||
5. 将 Client ID 和 Client Secret 填入配置文件中
|
||||
3. 从应用设置中获取 AppKey（Client ID）和 AppSecret（Client Secret）
|
||||
4. 配置机器人的回调模式和事件订阅
|
||||
5. 将 AppKey 和 AppSecret 填入配置文件中
|
||||
|
||||
## 消息发送机制
|
||||
|
||||
钉钉频道支持两种消息发送方式：
|
||||
|
||||
### 1. Session Webhook 回复（优先）
|
||||
|
||||
当用户发送消息给机器人时，钉钉会提供一个临时的 `session_webhook`，有效期约 2 小时。系统优先使用此方式回复消息，因为它更简单且不需要额外的 API 调用。
|
||||
|
||||
### 2. 主动消息（Proactive Messaging）
|
||||
|
||||
当 `session_webhook` 不可用或已过期时，系统会自动切换到钉钉 OpenAPI 发送主动消息：
|
||||
|
||||
- **单聊消息**: 使用 `/v1.0/robot/oToMessages/batchSend` API
|
||||
- **群聊消息**: 使用 `/v1.0/robot/groupMessages/send` API
|
||||
|
||||
主动消息功能使得以下场景成为可能：
|
||||
- 心跳通知（Heartbeat）
|
||||
- 设备告警
|
||||
- 定时任务提醒
|
||||
- 其他无需用户先发起对话的消息推送
|
||||
|
||||
### 主动消息的工作原理
|
||||
|
||||
1. **首次交互**: 用户发送消息后，系统会存储用户的 `staffId`（单聊）或 `openConversationId`（群聊）
|
||||
2. **后续推送**: 即使 `session_webhook` 过期，系统仍可通过 OpenAPI 主动发送消息
|
||||
3. **无历史交互**: 如果用户从未发送过消息，只要知道用户的 `staffId`，系统也可以主动发送单聊消息
|
||||
|
||||
## API 参考
|
||||
|
||||
| API | 用途 |
|
||||
|-----|------|
|
||||
| `POST /v1.0/oauth2/accessToken` | 获取访问令牌 |
|
||||
| `POST /v1.0/robot/oToMessages/batchSend` | 发送单聊消息 |
|
||||
| `POST /v1.0/robot/groupMessages/send` | 发送群聊消息 |
|
||||
|
||||
### 获取访问令牌
|
||||
|
||||
```http
|
||||
POST https://api.dingtalk.com/v1.0/oauth2/accessToken
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"appKey": "YOUR_APP_KEY",
|
||||
"appSecret": "YOUR_APP_SECRET"
|
||||
}
|
||||
```
|
||||
|
||||
响应：
|
||||
```json
|
||||
{
|
||||
"accessToken": "xxx",
|
||||
"expireIn": 7200
|
||||
}
|
||||
```
|
||||
|
||||
### 发送单聊消息
|
||||
|
||||
```http
|
||||
POST https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend
|
||||
Content-Type: application/json
|
||||
X-Acs-Dingtalk-Access-Token: ACCESS_TOKEN
|
||||
|
||||
{
|
||||
"robotCode": "YOUR_APP_KEY",
|
||||
"userIds": ["STAFF_ID"],
|
||||
"msgKey": "sampleMarkdown",
|
||||
"msgParam": "{\"title\":\"标题\",\"text\":\"内容\"}"
|
||||
}
|
||||
```
|
||||
|
||||
### 发送群聊消息
|
||||
|
||||
```http
|
||||
POST https://api.dingtalk.com/v1.0/robot/groupMessages/send
|
||||
Content-Type: application/json
|
||||
X-Acs-Dingtalk-Access-Token: ACCESS_TOKEN
|
||||
|
||||
{
|
||||
"robotCode": "YOUR_APP_KEY",
|
||||
"openConversationId": "CONVERSATION_ID",
|
||||
"msgKey": "sampleMarkdown",
|
||||
"msgParam": "{\"title\":\"标题\",\"text\":\"内容\"}"
|
||||
}
|
||||
```
|
||||
|
||||
## 官方文档
|
||||
|
||||
- [钉钉机器人开发文档](https://open.dingtalk.com/document/orgapp/the-robot-sends-a-group-message)
|
||||
- [获取访问令牌](https://open.dingtalk.com/document/development/obtain-the-access-token-of-an-internal-app)
|
||||
|
|
|
|||
|
|
@ -1,12 +1,17 @@
|
|||
// PicoClaw - Ultra-lightweight personal AI agent
|
||||
// DingTalk channel implementation using Stream Mode
|
||||
// DingTalk channel implementation using Stream Mode with proactive messaging support
|
||||
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot"
|
||||
"github.com/open-dingtalk/dingtalk-stream-sdk-go/client"
|
||||
|
|
@ -20,18 +25,43 @@ import (
|
|||
"github.com/sipeed/picoclaw/pkg/utils"
|
||||
)
|
||||
|
||||
const dingtalkAPIBase = "https://api.dingtalk.com"
|
||||
|
||||
// chatInfo stores information needed for proactive messaging
|
||||
type chatInfo struct {
|
||||
sessionWebhook string
|
||||
sessionWebhookExp time.Time // From sessionWebhookExpiredTime
|
||||
senderStaffId string // For single chat proactive send
|
||||
openConversationId string // For group chat proactive send (ConversationId)
|
||||
conversationType string // "1" = single, "2" = group
|
||||
}
|
||||
|
||||
// BatchSendResponse represents the batch send API response
|
||||
type BatchSendResponse struct {
|
||||
Code string `json:"code"`
|
||||
Message string `json:"message"`
|
||||
ProcessQueryKeys map[string]string `json:"processQueryKeys"`
|
||||
InvalidStaffIdList []string `json:"invalidStaffIdList"`
|
||||
}
|
||||
|
||||
// DingTalkChannel implements the Channel interface for DingTalk (钉钉)
|
||||
// It uses WebSocket for receiving messages via stream mode and API for sending
|
||||
type DingTalkChannel struct {
|
||||
*channels.BaseChannel
|
||||
config config.DingTalkConfig
|
||||
clientID string
|
||||
clientSecret string
|
||||
clientID string // AppKey (also used as robotCode for proactive messaging)
|
||||
clientSecret string // AppSecret
|
||||
streamClient *client.StreamClient
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
// Map to store session webhooks for each chat
|
||||
sessionWebhooks sync.Map // chatID -> sessionWebhook
|
||||
// Map to store chat info for each chat (includes session webhook and proactive send info)
|
||||
chatInfos sync.Map // chatID -> *chatInfo
|
||||
|
||||
// HTTP client for proactive API calls
|
||||
httpClient *http.Client
|
||||
accessToken string
|
||||
tokenExpiry time.Time
|
||||
tokenMu sync.RWMutex
|
||||
}
|
||||
|
||||
// NewDingTalkChannel creates a new DingTalk channel instance
|
||||
|
|
@ -52,8 +82,9 @@ func NewDingTalkChannel(cfg config.DingTalkConfig, messageBus *bus.MessageBus) (
|
|||
return &DingTalkChannel{
|
||||
BaseChannel: base,
|
||||
config: cfg,
|
||||
clientID: cfg.ClientID,
|
||||
clientID: cfg.ClientID, // Also used as robotCode for proactive messaging
|
||||
clientSecret: cfg.ClientSecret,
|
||||
httpClient: &http.Client{Timeout: 30 * time.Second},
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
|
@ -80,6 +111,16 @@ func (c *DingTalkChannel) Start(ctx context.Context) error {
|
|||
return fmt.Errorf("failed to start stream client: %w", err)
|
||||
}
|
||||
|
||||
// Get initial access token for proactive messaging
|
||||
if err := c.refreshAccessToken(); err != nil {
|
||||
logger.WarnCF("dingtalk", "Failed to get initial access token", map[string]any{
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
|
||||
// Start token refresh goroutine
|
||||
go c.tokenRefreshLoop()
|
||||
|
||||
c.SetRunning(true)
|
||||
logger.InfoC("dingtalk", "DingTalk channel started (Stream Mode)")
|
||||
return nil
|
||||
|
|
@ -102,30 +143,171 @@ func (c *DingTalkChannel) Stop(ctx context.Context) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// Send sends a message to DingTalk via the chatbot reply API
|
||||
// Send sends a message to DingTalk with fallback from session_webhook to proactive API
|
||||
func (c *DingTalkChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
|
||||
if !c.IsRunning() {
|
||||
return channels.ErrNotRunning
|
||||
}
|
||||
|
||||
// Get session webhook from storage
|
||||
sessionWebhookRaw, ok := c.sessionWebhooks.Load(msg.ChatID)
|
||||
if !ok {
|
||||
return fmt.Errorf("no session_webhook found for chat %s, cannot send message", msg.ChatID)
|
||||
}
|
||||
|
||||
sessionWebhook, ok := sessionWebhookRaw.(string)
|
||||
if !ok {
|
||||
return fmt.Errorf("invalid session_webhook type for chat %s", msg.ChatID)
|
||||
}
|
||||
|
||||
logger.DebugCF("dingtalk", "Sending message", map[string]any{
|
||||
"chat_id": msg.ChatID,
|
||||
"preview": utils.Truncate(msg.Content, 100),
|
||||
})
|
||||
|
||||
// Use the session webhook to send the reply
|
||||
return c.SendDirectReply(ctx, sessionWebhook, msg.Content)
|
||||
// 1. Try session_webhook first (if available and not expired)
|
||||
if info, ok := c.getChatInfo(msg.ChatID); ok {
|
||||
if info.sessionWebhook != "" && time.Now().Before(info.sessionWebhookExp) {
|
||||
err := c.SendDirectReply(ctx, info.sessionWebhook, msg.Content)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
// Log error and fall through to proactive API
|
||||
logger.DebugCF("dingtalk", "session_webhook failed, trying proactive API", map[string]any{
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Fall back to proactive API
|
||||
return c.sendProactive(ctx, msg.ChatID, msg.Content)
|
||||
}
|
||||
|
||||
// getChatInfo safely retrieves chat info
|
||||
func (c *DingTalkChannel) getChatInfo(chatID string) (*chatInfo, bool) {
|
||||
raw, ok := c.chatInfos.Load(chatID)
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
info, ok := raw.(*chatInfo)
|
||||
return info, ok
|
||||
}
|
||||
|
||||
// sendProactive sends a message using the proactive API
|
||||
// If chatInfo exists (from prior user message), use stored info for conversation type.
|
||||
// If not, assume single chat and use chatID directly as staffId - this allows
|
||||
// proactive messaging to users whose staffId is known (e.g., from state/config).
|
||||
func (c *DingTalkChannel) sendProactive(ctx context.Context, chatID, content string) error {
|
||||
accessToken := c.getAccessToken()
|
||||
if accessToken == "" {
|
||||
return fmt.Errorf("no valid access token available: %w", channels.ErrTemporary)
|
||||
}
|
||||
|
||||
info, ok := c.getChatInfo(chatID)
|
||||
if ok {
|
||||
// Use stored info (preferred - we know conversation type)
|
||||
if info.conversationType == "1" {
|
||||
// Single chat - use batch send API
|
||||
return c.sendProactiveSingleChat(ctx, accessToken, info.senderStaffId, content)
|
||||
}
|
||||
// Group chat - use group messages API
|
||||
return c.sendProactiveGroupChat(ctx, accessToken, info.openConversationId, content)
|
||||
}
|
||||
|
||||
// No stored chatInfo - assume single chat and use chatID directly as staffId
|
||||
// This enables proactive messaging without requiring prior user interaction
|
||||
logger.DebugCF("dingtalk", "No stored chatInfo, assuming single chat", map[string]any{
|
||||
"chat_id": chatID,
|
||||
})
|
||||
return c.sendProactiveSingleChat(ctx, accessToken, chatID, content)
|
||||
}
|
||||
|
||||
// sendProactiveSingleChat sends message via oToMessages/batchSend API for single chats
|
||||
// robotCode = clientID (AppKey)
|
||||
func (c *DingTalkChannel) sendProactiveSingleChat(ctx context.Context, accessToken, staffId, content string) error {
|
||||
msgParam := buildMarkdownMsgParam("PicoClaw", content)
|
||||
reqBody := map[string]any{
|
||||
"robotCode": c.clientID, // robotCode = AppKey
|
||||
"userIds": []string{staffId},
|
||||
"msgKey": "sampleMarkdown",
|
||||
"msgParam": msgParam,
|
||||
}
|
||||
|
||||
bodyBytes, err := json.Marshal(reqBody)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to marshal request: %w", err)
|
||||
}
|
||||
|
||||
apiURL := fmt.Sprintf("%s/v1.0/robot/oToMessages/batchSend", dingtalkAPIBase)
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", apiURL, bytes.NewReader(bodyBytes))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create request: %w", err)
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("X-Acs-Dingtalk-Access-Token", accessToken)
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to send request: %w", channels.ErrTemporary)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
respBody, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to read response: %w", err)
|
||||
}
|
||||
|
||||
var result BatchSendResponse
|
||||
if err := json.Unmarshal(respBody, &result); err != nil {
|
||||
return fmt.Errorf("failed to parse response: %w", err)
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK || (result.Code != "" && result.Code != "0" && result.Code != "success") {
|
||||
return fmt.Errorf("dingtalk API error: %s (code: %s, status: %d)", result.Message, result.Code, resp.StatusCode)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// sendProactiveGroupChat sends message via groupMessages API for group chats
|
||||
// robotCode = clientID (AppKey)
|
||||
func (c *DingTalkChannel) sendProactiveGroupChat(
|
||||
ctx context.Context,
|
||||
accessToken, openConversationId, content string,
|
||||
) error {
|
||||
msgParam := buildMarkdownMsgParam("PicoClaw", content)
|
||||
reqBody := map[string]any{
|
||||
"openConversationId": openConversationId,
|
||||
"robotCode": c.clientID, // robotCode = AppKey
|
||||
"msgKey": "sampleMarkdown",
|
||||
"msgParam": msgParam,
|
||||
}
|
||||
|
||||
bodyBytes, err := json.Marshal(reqBody)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to marshal request: %w", err)
|
||||
}
|
||||
|
||||
apiURL := fmt.Sprintf("%s/v1.0/robot/groupMessages/send", dingtalkAPIBase)
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", apiURL, bytes.NewReader(bodyBytes))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create request: %w", err)
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("X-Acs-Dingtalk-Access-Token", accessToken)
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to send request: %w", channels.ErrTemporary)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
respBody, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to read response: %w", err)
|
||||
}
|
||||
|
||||
var result BatchSendResponse
|
||||
if err := json.Unmarshal(respBody, &result); err != nil {
|
||||
return fmt.Errorf("failed to parse response: %w", err)
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK || (result.Code != "" && result.Code != "0" && result.Code != "success") {
|
||||
return fmt.Errorf("dingtalk API error: %s (code: %s, status: %d)", result.Message, result.Code, resp.StatusCode)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// onChatBotMessageReceived implements the IChatBotMessageHandler function signature
|
||||
|
|
@ -158,8 +340,21 @@ func (c *DingTalkChannel) onChatBotMessageReceived(
|
|||
chatID = data.ConversationId
|
||||
}
|
||||
|
||||
// Store the session webhook for this chat so we can reply later
|
||||
c.sessionWebhooks.Store(chatID, data.SessionWebhook)
|
||||
// Parse expiry time from sessionWebhookExpiredTime
|
||||
var webhookExpiry time.Time
|
||||
if data.SessionWebhookExpiredTime > 0 {
|
||||
webhookExpiry = time.Unix(data.SessionWebhookExpiredTime/1000, 0)
|
||||
}
|
||||
|
||||
// Store extended chat info for proactive messaging
|
||||
info := &chatInfo{
|
||||
sessionWebhook: data.SessionWebhook,
|
||||
sessionWebhookExp: webhookExpiry,
|
||||
senderStaffId: data.SenderStaffId,
|
||||
openConversationId: data.ConversationId,
|
||||
conversationType: data.ConversationType,
|
||||
}
|
||||
c.chatInfos.Store(chatID, info)
|
||||
|
||||
metadata := map[string]string{
|
||||
"sender_name": senderNick,
|
||||
|
|
@ -229,3 +424,103 @@ func (c *DingTalkChannel) SendDirectReply(ctx context.Context, sessionWebhook, c
|
|||
|
||||
return nil
|
||||
}
|
||||
|
||||
// refreshAccessToken fetches a new access token from DingTalk API
|
||||
// API: POST /v1.0/oauth2/accessToken
|
||||
// Body: {"appKey": "...", "appSecret": "..."}
|
||||
// Response: {"accessToken": "...", "expireIn": 7200}
|
||||
func (c *DingTalkChannel) refreshAccessToken() error {
|
||||
reqBody := map[string]string{
|
||||
"appKey": c.clientID,
|
||||
"appSecret": c.clientSecret,
|
||||
}
|
||||
|
||||
bodyBytes, err := json.Marshal(reqBody)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to marshal request: %w", err)
|
||||
}
|
||||
|
||||
apiURL := fmt.Sprintf("%s/v1.0/oauth2/accessToken", dingtalkAPIBase)
|
||||
|
||||
req, err := http.NewRequestWithContext(c.ctx, "POST", apiURL, bytes.NewReader(bodyBytes))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create request: %w", err)
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to request access token: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to read response: %w", err)
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("access token request failed with status %d: %s", resp.StatusCode, string(body))
|
||||
}
|
||||
|
||||
var tokenResp struct {
|
||||
AccessToken string `json:"accessToken"`
|
||||
ExpireIn int `json:"expireIn"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &tokenResp); err != nil {
|
||||
return fmt.Errorf("failed to parse response: %w", err)
|
||||
}
|
||||
|
||||
c.tokenMu.Lock()
|
||||
c.accessToken = tokenResp.AccessToken
|
||||
// Refresh 5 minutes before expiry
|
||||
c.tokenExpiry = time.Now().Add(time.Duration(tokenResp.ExpireIn-300) * time.Second)
|
||||
c.tokenMu.Unlock()
|
||||
|
||||
logger.DebugCF("dingtalk", "Access token refreshed successfully", map[string]any{
|
||||
"expire_in": tokenResp.ExpireIn,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// tokenRefreshLoop periodically refreshes the access token
|
||||
func (c *DingTalkChannel) tokenRefreshLoop() {
|
||||
ticker := time.NewTicker(5 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if err := c.refreshAccessToken(); err != nil {
|
||||
logger.ErrorCF("dingtalk", "Failed to refresh access token", map[string]any{
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// getAccessToken returns the current valid access token
|
||||
func (c *DingTalkChannel) getAccessToken() string {
|
||||
c.tokenMu.RLock()
|
||||
defer c.tokenMu.RUnlock()
|
||||
|
||||
if time.Now().After(c.tokenExpiry) {
|
||||
return ""
|
||||
}
|
||||
|
||||
return c.accessToken
|
||||
}
|
||||
|
||||
// buildMarkdownMsgParam builds the msgParam for markdown messages
|
||||
func buildMarkdownMsgParam(title, content string) string {
|
||||
param := map[string]string{
|
||||
"title": title,
|
||||
"text": content,
|
||||
}
|
||||
data, _ := json.Marshal(param)
|
||||
return string(data)
|
||||
}
|
||||
|
|
|
|||
329
pkg/channels/dingtalk/dingtalk_test.go
Normal file
329
pkg/channels/dingtalk/dingtalk_test.go
Normal file
|
|
@ -0,0 +1,329 @@
|
|||
package dingtalk
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/config"
|
||||
)
|
||||
|
||||
// mockTransport is a custom http.RoundTripper for testing
|
||||
type mockTransport struct {
|
||||
response *http.Response
|
||||
responseBody any
|
||||
requestErr error
|
||||
requests []*http.Request
|
||||
}
|
||||
|
||||
func (m *mockTransport) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
m.requests = append(m.requests, req)
|
||||
|
||||
if m.requestErr != nil {
|
||||
return nil, m.requestErr
|
||||
}
|
||||
|
||||
if m.response != nil {
|
||||
return m.response, nil
|
||||
}
|
||||
|
||||
// Generate response from responseBody
|
||||
bodyBytes, _ := json.Marshal(m.responseBody)
|
||||
return &http.Response{
|
||||
StatusCode: http.StatusOK,
|
||||
Header: http.Header{"Content-Type": []string{"application/json"}},
|
||||
Body: io.NopCloser(bytes.NewReader(bodyBytes)),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// TestRefreshAccessToken tests the access token refresh functionality
|
||||
func TestRefreshAccessToken(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
clientID string
|
||||
clientSecret string
|
||||
serverResponse any
|
||||
serverStatus int
|
||||
expectError bool
|
||||
expectToken string
|
||||
}{
|
||||
{
|
||||
name: "successful token refresh",
|
||||
clientID: "test_app_key",
|
||||
clientSecret: "test_app_secret",
|
||||
serverResponse: map[string]any{
|
||||
"accessToken": "test_access_token_123",
|
||||
"expireIn": 7200,
|
||||
},
|
||||
serverStatus: http.StatusOK,
|
||||
expectError: false,
|
||||
expectToken: "test_access_token_123",
|
||||
},
|
||||
{
|
||||
name: "invalid credentials",
|
||||
clientID: "invalid_client",
|
||||
clientSecret: "invalid_secret",
|
||||
serverResponse: map[string]any{
|
||||
"code": "invalid.client",
|
||||
"message": "Invalid client credentials",
|
||||
},
|
||||
serverStatus: http.StatusBadRequest,
|
||||
expectError: true,
|
||||
expectToken: "",
|
||||
},
|
||||
{
|
||||
name: "server error",
|
||||
clientID: "test_app_key",
|
||||
clientSecret: "test_app_secret",
|
||||
serverResponse: map[string]any{
|
||||
"code": "server.error",
|
||||
"message": "Internal server error",
|
||||
},
|
||||
serverStatus: http.StatusInternalServerError,
|
||||
expectError: true,
|
||||
expectToken: "",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// Create mock transport
|
||||
mock := &mockTransport{}
|
||||
|
||||
if tt.serverStatus > 0 {
|
||||
bodyBytes, _ := json.Marshal(tt.serverResponse)
|
||||
mock.response = &http.Response{
|
||||
StatusCode: tt.serverStatus,
|
||||
Header: http.Header{"Content-Type": []string{"application/json"}},
|
||||
Body: io.NopCloser(bytes.NewReader(bodyBytes)),
|
||||
}
|
||||
}
|
||||
|
||||
// Create HTTP client with mock transport
|
||||
httpClient := &http.Client{
|
||||
Transport: mock,
|
||||
}
|
||||
|
||||
// Create channel
|
||||
cfg := config.DingTalkConfig{
|
||||
Enabled: true,
|
||||
ClientID: tt.clientID,
|
||||
ClientSecret: tt.clientSecret,
|
||||
}
|
||||
|
||||
channel := &DingTalkChannel{
|
||||
config: cfg,
|
||||
clientID: tt.clientID,
|
||||
clientSecret: tt.clientSecret,
|
||||
httpClient: httpClient,
|
||||
ctx: context.Background(),
|
||||
}
|
||||
|
||||
// Test the token refresh
|
||||
err := channel.refreshAccessToken()
|
||||
|
||||
if tt.expectError {
|
||||
if err == nil {
|
||||
t.Errorf("Expected error but got none")
|
||||
}
|
||||
} else {
|
||||
if err != nil {
|
||||
t.Errorf("Unexpected error: %v", err)
|
||||
}
|
||||
if channel.accessToken != tt.expectToken {
|
||||
t.Errorf("Expected token %s, got %s", tt.expectToken, channel.accessToken)
|
||||
}
|
||||
}
|
||||
|
||||
// Verify request format if a request was made
|
||||
if len(mock.requests) > 0 {
|
||||
req := mock.requests[0]
|
||||
|
||||
// Verify method
|
||||
if req.Method != "POST" {
|
||||
t.Errorf("Expected POST request, got %s", req.Method)
|
||||
}
|
||||
|
||||
// Verify URL path is correct
|
||||
expectedPath := "/v1.0/oauth2/accessToken"
|
||||
if req.URL.Path != expectedPath {
|
||||
t.Errorf("Expected path %s, got %s", expectedPath, req.URL.Path)
|
||||
}
|
||||
|
||||
// Verify Content-Type
|
||||
if req.Header.Get("Content-Type") != "application/json" {
|
||||
t.Errorf("Expected Content-Type: application/json, got %s", req.Header.Get("Content-Type"))
|
||||
}
|
||||
|
||||
// Parse and verify request body
|
||||
bodyBytes, _ := io.ReadAll(req.Body)
|
||||
var reqBody map[string]string
|
||||
json.Unmarshal(bodyBytes, &reqBody)
|
||||
|
||||
// Verify new API format: appKey and appSecret
|
||||
if reqBody["appKey"] != tt.clientID {
|
||||
t.Errorf("Expected appKey %s, got %s", tt.clientID, reqBody["appKey"])
|
||||
}
|
||||
if reqBody["appSecret"] != tt.clientSecret {
|
||||
t.Errorf("Expected appSecret %s, got %s", tt.clientSecret, reqBody["appSecret"])
|
||||
}
|
||||
// Verify old fields are NOT present
|
||||
if _, exists := reqBody["client_id"]; exists {
|
||||
t.Error("client_id should not be present in request body")
|
||||
}
|
||||
if _, exists := reqBody["client_secret"]; exists {
|
||||
t.Error("client_secret should not be present in request body")
|
||||
}
|
||||
if _, exists := reqBody["grant_type"]; exists {
|
||||
t.Error("grant_type should not be present in request body")
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestChatInfoStorage tests that chat info is properly stored
|
||||
func TestChatInfoStorage(t *testing.T) {
|
||||
channel := &DingTalkChannel{
|
||||
chatInfos: sync.Map{},
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
info := &chatInfo{
|
||||
sessionWebhook: "https://webhook.example.com/test",
|
||||
sessionWebhookExp: now.Add(2 * time.Hour),
|
||||
senderStaffId: "staff123",
|
||||
openConversationId: "conv456",
|
||||
conversationType: "1",
|
||||
}
|
||||
|
||||
// Store the info
|
||||
channel.chatInfos.Store("test_chat_id", info)
|
||||
|
||||
// Retrieve and verify
|
||||
retrieved, ok := channel.getChatInfo("test_chat_id")
|
||||
if !ok {
|
||||
t.Fatal("Failed to retrieve chat info")
|
||||
}
|
||||
|
||||
if retrieved.sessionWebhook != info.sessionWebhook {
|
||||
t.Errorf("Expected sessionWebhook %s, got %s", info.sessionWebhook, retrieved.sessionWebhook)
|
||||
}
|
||||
if retrieved.senderStaffId != info.senderStaffId {
|
||||
t.Errorf("Expected senderStaffId %s, got %s", info.senderStaffId, retrieved.senderStaffId)
|
||||
}
|
||||
if retrieved.conversationType != info.conversationType {
|
||||
t.Errorf("Expected conversationType %s, got %s", info.conversationType, retrieved.conversationType)
|
||||
}
|
||||
|
||||
// Test non-existent chat
|
||||
_, ok = channel.getChatInfo("non_existent")
|
||||
if ok {
|
||||
t.Error("Expected false for non-existent chat")
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildMarkdownMsgParam tests the markdown message parameter builder
|
||||
func TestBuildMarkdownMsgParam(t *testing.T) {
|
||||
result := buildMarkdownMsgParam("Test Title", "Test Content")
|
||||
|
||||
var parsed map[string]string
|
||||
if err := json.Unmarshal([]byte(result), &parsed); err != nil {
|
||||
t.Fatalf("Failed to parse result: %v", err)
|
||||
}
|
||||
|
||||
if parsed["title"] != "Test Title" {
|
||||
t.Errorf("Expected title 'Test Title', got %s", parsed["title"])
|
||||
}
|
||||
if parsed["text"] != "Test Content" {
|
||||
t.Errorf("Expected text 'Test Content', got %s", parsed["text"])
|
||||
}
|
||||
}
|
||||
|
||||
// TestTokenExpiry tests the token expiry logic
|
||||
func TestTokenExpiry(t *testing.T) {
|
||||
channel := &DingTalkChannel{
|
||||
accessToken: "test_token",
|
||||
tokenExpiry: time.Now().Add(1 * time.Hour),
|
||||
}
|
||||
|
||||
// Token should be valid
|
||||
token := channel.getAccessToken()
|
||||
if token != "test_token" {
|
||||
t.Errorf("Expected token 'test_token', got %s", token)
|
||||
}
|
||||
|
||||
// Set token as expired
|
||||
channel.tokenExpiry = time.Now().Add(-1 * time.Hour)
|
||||
|
||||
// Token should be empty (expired)
|
||||
token = channel.getAccessToken()
|
||||
if token != "" {
|
||||
t.Errorf("Expected empty token for expired, got %s", token)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSendProactiveWithoutToken tests that proactive send fails gracefully without token
|
||||
func TestSendProactiveWithoutToken(t *testing.T) {
|
||||
channel := &DingTalkChannel{
|
||||
chatInfos: sync.Map{},
|
||||
}
|
||||
|
||||
// Store some chat info
|
||||
channel.chatInfos.Store("test_chat", &chatInfo{
|
||||
conversationType: "1",
|
||||
senderStaffId: "staff123",
|
||||
})
|
||||
|
||||
// No token set, should fail
|
||||
err := channel.sendProactive(context.Background(), "test_chat", "test message")
|
||||
if err == nil {
|
||||
t.Error("Expected error when no access token available")
|
||||
}
|
||||
}
|
||||
|
||||
// TestSendProactiveWithoutChatInfo tests that proactive send works without stored chat info
|
||||
// When no chatInfo is stored, it assumes single chat and uses chatID as staffId
|
||||
func TestSendProactiveWithoutChatInfo(t *testing.T) {
|
||||
mock := &mockTransport{
|
||||
responseBody: map[string]any{},
|
||||
}
|
||||
httpClient := &http.Client{
|
||||
Transport: mock,
|
||||
}
|
||||
|
||||
channel := &DingTalkChannel{
|
||||
chatInfos: sync.Map{},
|
||||
accessToken: "test_token",
|
||||
tokenExpiry: time.Now().Add(1 * time.Hour),
|
||||
httpClient: httpClient,
|
||||
clientID: "test_app_key", // robotCode = clientID
|
||||
}
|
||||
|
||||
// No chat info stored - should try to send as single chat using clientID as robotCode
|
||||
err := channel.sendProactive(context.Background(), "staff123", "test message")
|
||||
if err != nil {
|
||||
t.Errorf("Expected no error, got: %v", err)
|
||||
}
|
||||
|
||||
// Verify the request was made to the correct endpoint
|
||||
if len(mock.requests) != 1 {
|
||||
t.Fatalf("Expected 1 request, got %d", len(mock.requests))
|
||||
}
|
||||
|
||||
req := mock.requests[0]
|
||||
expectedPath := "/v1.0/robot/oToMessages/batchSend"
|
||||
if req.URL.Path != expectedPath {
|
||||
t.Errorf("Expected path %s, got %s", expectedPath, req.URL.Path)
|
||||
}
|
||||
|
||||
// Verify the header
|
||||
if req.Header.Get("X-Acs-Dingtalk-Access-Token") != "test_token" {
|
||||
t.Errorf("Expected X-Acs-Dingtalk-Access-Token header, got %s", req.Header.Get("X-Acs-Dingtalk-Access-Token"))
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue