diff --git a/pkg/channels/dingtalk/api.go b/pkg/channels/dingtalk/api.go new file mode 100644 index 000000000..d83471990 --- /dev/null +++ b/pkg/channels/dingtalk/api.go @@ -0,0 +1,151 @@ +// PicoClaw - Ultra-lightweight personal AI agent +// DingTalk robot API for proactive messaging + +package dingtalk + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + + "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/logger" +) + +// sendGroupMessage sends a message to a group chat via robot API +func (c *DingTalkChannel) sendGroupMessage(ctx context.Context, accessToken, openConversationId, content string) error { + url := fmt.Sprintf("%s/v1.0/robot/groupMessages/send", dingtalkAPIBase) + + // Build msgParam for markdown message + msgParam, _ := json.Marshal(map[string]string{ + "title": "PicoClaw", + "text": content, + }) + + body := map[string]any{ + "openConversationId": openConversationId, + "robotCode": c.clientID, + "msgKey": "sampleMarkdown", + "msgParam": string(msgParam), + } + + return c.sendRobotAPIRequest(ctx, accessToken, url, body, "group") +} + +// sendOToMessage sends a message to a user via robot API (one-to-one) +func (c *DingTalkChannel) sendOToMessage(ctx context.Context, accessToken, userID, content string) error { + url := fmt.Sprintf("%s/v1.0/robot/oToMessages/batchSend", dingtalkAPIBase) + + // Build msgParam for markdown message + msgParam, _ := json.Marshal(map[string]string{ + "title": "PicoClaw", + "text": content, + }) + + body := map[string]any{ + "robotCode": c.clientID, + "userIds": []string{userID}, + "msgKey": "sampleMarkdown", + "msgParam": string(msgParam), + } + + return c.sendRobotAPIRequest(ctx, accessToken, url, body, "direct") +} + +// sendRobotAPIRequest sends a request to DingTalk robot API +func (c *DingTalkChannel) sendRobotAPIRequest( + ctx context.Context, + accessToken, url string, + body any, + msgType string, +) error { + jsonData, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("marshal request body failed: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData)) + if err != nil { + return fmt.Errorf("create request failed: %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 channels.ClassifyNetError(err) + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("read response failed: %w", err) + } + + if resp.StatusCode != http.StatusOK { + // Parse error response for better error message + var errResp struct { + Code string `json:"code"` + Msg string `json:"message"` + } + if json.Unmarshal(respBody, &errResp) == nil && errResp.Msg != "" { + return fmt.Errorf("dingtalk robot API error: %s (%s)", errResp.Msg, errResp.Code) + } + return channels.ClassifySendError(resp.StatusCode, + fmt.Errorf("dingtalk robot API error (status %d): %s", resp.StatusCode, string(respBody))) + } + + logger.InfoCF("dingtalk", "Robot API message sent successfully", map[string]any{ + "type": msgType, + "status": resp.StatusCode, + }) + + return nil +} + +// sendViaRobotAPI sends a message via the robot API +// For proactive messaging: chatID is used directly as staffId (direct) or openConversationId (group) +// Group IDs start with "cid", everything else is treated as direct message staffId +func (c *DingTalkChannel) sendViaRobotAPI(ctx context.Context, chatID, content string) error { + accessToken, err := c.ensureValidToken(ctx) + if err != nil { + return fmt.Errorf("failed to get access token: %w", err) + } + + // First, check if we have session info (from prior conversation) + session := c.getSession(chatID) + if session != nil { + if session.ConversationType == "1" { + // Direct message with session info + if session.SenderStaffId != "" { + return c.sendOToMessage(ctx, accessToken, session.SenderStaffId, content) + } + } else { + // Group message with session info + if session.OpenConversationId != "" { + return c.sendGroupMessage(ctx, accessToken, session.OpenConversationId, content) + } + } + } + + // No session info - detect type based on chatID format + // Group IDs start with "cid" (openConversationId format) + isGroup := len(chatID) > 3 && chatID[:3] == "cid" + + if isGroup { + logger.DebugCF("dingtalk", "Sending proactive group message", map[string]any{ + "open_conversation_id": chatID, + }) + return c.sendGroupMessage(ctx, accessToken, chatID, content) + } + + // Direct message - chatID is the staffId + logger.DebugCF("dingtalk", "Sending proactive direct message", map[string]any{ + "staff_id": chatID, + }) + return c.sendOToMessage(ctx, accessToken, chatID, content) +} diff --git a/pkg/channels/dingtalk/dingtalk.go b/pkg/channels/dingtalk/dingtalk.go index 2f8fe7767..26ac4af15 100644 --- a/pkg/channels/dingtalk/dingtalk.go +++ b/pkg/channels/dingtalk/dingtalk.go @@ -6,6 +6,7 @@ package dingtalk import ( "context" "fmt" + "net/http" "sync" "time" @@ -28,6 +29,16 @@ const ( recoveryRetryDelay = 30 * time.Second // Delay before retrying failed recovery ) +// DingTalkSession stores complete session information for proactive messaging +type DingTalkSession struct { + SessionWebhook string `json:"session_webhook"` + SessionWebhookExpiredTime int64 `json:"session_webhook_expired_time"` // milliseconds timestamp + OpenConversationId string `json:"open_conversation_id"` // Group chat ID + SenderStaffId string `json:"sender_staff_id"` // User ID for direct chat + ConversationType string `json:"conversation_type"` // "1" = direct, else = group + LastUpdated time.Time `json:"last_updated"` +} + // DingTalkChannel implements the Channel interface for DingTalk (钉钉) // It uses WebSocket for receiving messages via stream mode and API for sending type DingTalkChannel struct { @@ -38,8 +49,13 @@ type DingTalkChannel struct { 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 session info for each chat + sessions sync.Map // chatID -> *DingTalkSession + // Token management for proactive messaging via robot API + accessToken string + tokenExpiry time.Time + tokenMu sync.RWMutex + httpClient *http.Client // Health monitoring lastMessageTime time.Time mu sync.RWMutex @@ -72,6 +88,19 @@ func (c *DingTalkChannel) Start(ctx context.Context) error { c.ctx, c.cancel = context.WithCancel(ctx) c.lastMessageTime = time.Now() // Initialize on start + // Initialize HTTP client for robot API + c.httpClient = &http.Client{Timeout: 30 * time.Second} + + // If proactive send is enabled, get initial token and start refresh loop + if c.config.ProactiveSend { + if err := c.refreshAccessToken(); err != nil { + logger.WarnCF("dingtalk", "Failed to get initial access token, proactive send may not work", map[string]any{ + "error": err.Error(), + }) + } + go c.tokenRefreshLoop() + } + // Start the stream client if err := c.startStreamClient(); err != nil { return err @@ -81,7 +110,9 @@ func (c *DingTalkChannel) Start(ctx context.Context) error { go c.healthMonitor() c.SetRunning(true) - logger.InfoC("dingtalk", "DingTalk channel started (Stream Mode)") + logger.InfoCF("dingtalk", "DingTalk channel started", map[string]any{ + "proactive_send": c.config.ProactiveSend, + }) return nil } @@ -199,30 +230,60 @@ 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 +// Priority: 1) session_webhook (if valid), 2) robot API (if proactive_send enabled) 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 + session := c.getSession(msg.ChatID) + if session != nil && c.isSessionWebhookValid(session) { + err := c.SendDirectReply(ctx, session.SessionWebhook, msg.Content) + if err == nil { + return nil + } + // Session webhook failed, log and try fallback + logger.WarnCF("dingtalk", "Session webhook send failed, trying fallback", map[string]any{ + "error": err.Error(), + "chat_id": msg.ChatID, + }) + } + + // 2. Fallback to robot API if proactive send is enabled + if !c.config.ProactiveSend { + return fmt.Errorf("no valid session_webhook for chat %s and proactive_send is disabled", msg.ChatID) + } + + return c.sendViaRobotAPI(ctx, msg.ChatID, msg.Content) +} + +// getSession retrieves session info for a chat +func (c *DingTalkChannel) getSession(chatID string) *DingTalkSession { + if v, ok := c.sessions.Load(chatID); ok { + if session, ok := v.(*DingTalkSession); ok { + return session + } + } + return nil +} + +// isSessionWebhookValid checks if the session webhook is still valid +func (c *DingTalkChannel) isSessionWebhookValid(session *DingTalkSession) bool { + if session.SessionWebhook == "" { + return false + } + // Check expiry if available (SessionWebhookExpiredTime is in milliseconds) + if session.SessionWebhookExpiredTime > 0 { + return time.Now().UnixMilli() < session.SessionWebhookExpiredTime + } + return true // No expiry info, assume valid } // onChatBotMessageReceived implements the IChatBotMessageHandler function signature @@ -258,8 +319,16 @@ 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) + // Store complete session info for proactive messaging + session := &DingTalkSession{ + SessionWebhook: data.SessionWebhook, + SessionWebhookExpiredTime: data.SessionWebhookExpiredTime, + OpenConversationId: data.ConversationId, + SenderStaffId: data.SenderStaffId, + ConversationType: data.ConversationType, + LastUpdated: time.Now(), + } + c.sessions.Store(chatID, session) metadata := map[string]string{ "sender_name": senderNick, diff --git a/pkg/channels/dingtalk/token.go b/pkg/channels/dingtalk/token.go new file mode 100644 index 000000000..4ce2be481 --- /dev/null +++ b/pkg/channels/dingtalk/token.go @@ -0,0 +1,139 @@ +// PicoClaw - Ultra-lightweight personal AI agent +// DingTalk access token management + +package dingtalk + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" + + "github.com/sipeed/picoclaw/pkg/logger" +) + +const ( + dingtalkAPIBase = "https://api.dingtalk.com" + tokenRefreshInterval = 5 * time.Minute + tokenExpireBuffer = 5 * time.Minute // Refresh token 5 minutes before expiry +) + +// accessTokenResponse 钉钉 access_token 响应 +type accessTokenResponse struct { + AccessToken string `json:"accessToken"` + ExpireIn int64 `json:"expireIn"` // seconds +} + +// refreshAccessToken refreshes the access token from DingTalk API +func (c *DingTalkChannel) refreshAccessToken() error { + url := fmt.Sprintf("%s/v1.0/oauth2/accessToken", dingtalkAPIBase) + + body := map[string]string{ + "appKey": c.clientID, + "appSecret": c.clientSecret, + } + jsonData, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("marshal request body failed: %w", err) + } + + req, err := http.NewRequestWithContext(c.ctx, http.MethodPost, url, bytes.NewBuffer(jsonData)) + if err != nil { + return fmt.Errorf("create request failed: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("read response failed: %w", err) + } + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("dingtalk API error (status %d): %s", resp.StatusCode, string(respBody)) + } + + var tokenResp accessTokenResponse + if err := json.Unmarshal(respBody, &tokenResp); err != nil { + return fmt.Errorf("parse response failed: %w", err) + } + + if tokenResp.AccessToken == "" { + return fmt.Errorf("empty access token in response") + } + + c.tokenMu.Lock() + c.accessToken = tokenResp.AccessToken + c.tokenExpiry = time.Now().Add(time.Duration(tokenResp.ExpireIn) * time.Second) + c.tokenMu.Unlock() + + logger.DebugCF("dingtalk", "Access token refreshed successfully", map[string]any{ + "expires_in": tokenResp.ExpireIn, + }) + return nil +} + +// getAccessToken returns the current valid access token +// Returns empty string if token is expired or about to expire +func (c *DingTalkChannel) getAccessToken() string { + c.tokenMu.RLock() + defer c.tokenMu.RUnlock() + + if c.accessToken == "" { + return "" + } + + // Check if token is about to expire (within buffer time) + if time.Now().After(c.tokenExpiry.Add(-tokenExpireBuffer)) { + return "" // Token expired or about to expire + } + + return c.accessToken +} + +// ensureValidToken ensures we have a valid access token, refreshing if necessary +func (c *DingTalkChannel) ensureValidToken(ctx context.Context) (string, error) { + token := c.getAccessToken() + if token != "" { + return token, nil + } + + // Need to refresh + if err := c.refreshAccessToken(); err != nil { + return "", fmt.Errorf("failed to refresh access token: %w", err) + } + + return c.getAccessToken(), nil +} + +// tokenRefreshLoop runs periodically to refresh the access token +func (c *DingTalkChannel) tokenRefreshLoop() { + ticker := time.NewTicker(tokenRefreshInterval) + defer ticker.Stop() + + for { + select { + case <-c.ctx.Done(): + return + case <-ticker.C: + if c.config.ProactiveSend { + // Check if token needs refresh + if c.getAccessToken() == "" { + if err := c.refreshAccessToken(); err != nil { + logger.ErrorCF("dingtalk", "Failed to refresh access token", map[string]any{ + "error": err.Error(), + }) + } + } + } + } + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index b3ad050b7..1ce9bf6f8 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -322,6 +322,7 @@ type DingTalkConfig struct { AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_DINGTALK_ALLOW_FROM"` GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty"` ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_DINGTALK_REASONING_CHANNEL_ID"` + ProactiveSend bool `json:"proactive_send" env:"PICOCLAW_CHANNELS_DINGTALK_PROACTIVE_SEND"` } type SlackConfig struct {