From fba80d059632b9edd8f9bf415f1c4d573598b595 Mon Sep 17 00:00:00 2001 From: zhaoyunxing Date: Thu, 12 Mar 2026 21:25:47 +0800 Subject: [PATCH] fix(channels): improve dingtalk client robustness - add mutex to Client to protect token/expires from concurrent refresh - pass context through GetToken so cancellation/deadlines propagate - rename Id/Ids identifiers to ID/IDs per Go conventions - fall back to direct reply instead of dropping message on card delivery failure - use Load instead of LoadAndDelete for sessionWebhooks and cardInstanceIDs to support multiple outbound messages per turn --- pkg/channels/dingtalk/client.go | 45 +++++++++++++++++-------------- pkg/channels/dingtalk/dingtalk.go | 24 ++++++++--------- 2 files changed, 36 insertions(+), 33 deletions(-) diff --git a/pkg/channels/dingtalk/client.go b/pkg/channels/dingtalk/client.go index d73e72595..e864db467 100644 --- a/pkg/channels/dingtalk/client.go +++ b/pkg/channels/dingtalk/client.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "net/http" + "sync" "time" "github.com/google/uuid" @@ -50,6 +51,7 @@ type Client struct { expires time.Time client *http.Client + mu sync.Mutex // protects token and expires } // NewClient creates a new Client instance with the provided client ID, client secret, and optional configurations. @@ -69,7 +71,10 @@ func NewClient(clientID, clientSecret string, opts ...ClientOption) *Client { } // GetToken retrieves the access token, refreshing it if it has expired. -func (c *Client) GetToken() (string, error) { +func (c *Client) GetToken(ctx context.Context) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + if time.Now().Before(c.expires) { return c.token, nil } @@ -81,7 +86,7 @@ func (c *Client) GetToken() (string, error) { Expires int64 `json:"expireIn"` Token string `json:"accessToken"` }{} - err := c.httpRequest(context.Background(), http.MethodPost, accessToken, data, &resp) + err := c.httpRequest(ctx, http.MethodPost, accessToken, data, &resp) if err != nil { return "", err } @@ -91,24 +96,24 @@ func (c *Client) GetToken() (string, error) { } // BatchSendMessages sends a message to multiple users in a batch. -func (c *Client) BatchSendMessages(ctx context.Context, msgType MessageType, userIds []string, content string) error { +func (c *Client) BatchSendMessages(ctx context.Context, msgType MessageType, userIDs []string, content string) error { body := map[string]any{ "robotCode": c.robotCode, "msgKey": msgType, - "userIds": userIds, + "userIds": userIDs, "msgParam": c.buildSendMessages(msgType, content), } return c.httpRequest(ctx, http.MethodPost, batchSendMessages, body, nil) } -// CardStreaming updates the content of a card instance identified by cardInstanceId. -func (c *Client) CardStreaming(ctx context.Context, cardInstanceId, content string) error { +// CardStreaming updates the content of a card instance identified by cardInstanceID. +func (c *Client) CardStreaming(ctx context.Context, cardInstanceID, content string) error { id, err := uuid.NewUUID() if err != nil { return err } body := map[string]any{ - "outTrackId": cardInstanceId, + "outTrackId": cardInstanceID, "guid": id.String(), "key": c.cardTemplateContentKey, "content": content, @@ -118,7 +123,7 @@ func (c *Client) CardStreaming(ctx context.Context, cardInstanceId, content stri } // CardCreateAndDeliver creates a card instance and delivers it to the user or group. -// It returns the outTrackId of the created card instance, which can be used for subsequent updates via CardStreaming. +// It returns the outTrackID of the created card instance, which can be used for subsequent updates via CardStreaming. // If the chatbot is in a group conversation, the card will be delivered to the group; otherwise, it will be delivered to the user. // Card Delivery API Documentation func (c *Client) CardCreateAndDeliver(ctx context.Context, chatbot *chatbot.BotCallbackDataModel) (string, error) { @@ -127,7 +132,7 @@ func (c *Client) CardCreateAndDeliver(ctx context.Context, chatbot *chatbot.BotC } var ( group = chatbot.ConversationType == "2" - openSpaceId = "dtv1.card//IM_ROBOT." + chatbot.SenderStaffId + openSpaceID = "dtv1.card//IM_ROBOT." + chatbot.SenderStaffId imRobotOpenDeliverModel = map[string]any{ "spaceType": "IM_ROBOT", } @@ -138,7 +143,7 @@ func (c *Client) CardCreateAndDeliver(ctx context.Context, chatbot *chatbot.BotC "supportForward": true, } imGroupOpenDeliverModel = map[string]any{ - "robotCode": c.clientID, + "robotCode": c.robotCode, } ) @@ -147,16 +152,16 @@ func (c *Client) CardCreateAndDeliver(ctx context.Context, chatbot *chatbot.BotC return "", err } if group { - openSpaceId = "dtv1.card//IM_GROUP." + chatbot.ConversationId + openSpaceID = "dtv1.card//IM_GROUP." + chatbot.ConversationId imRobotOpenDeliverModel = map[string]any{} - imGroupOpenDeliverModel["robotCode"] = c.clientID + imGroupOpenDeliverModel["robotCode"] = c.robotCode } body := map[string]any{ "cardTemplateId": c.cardTemplateID, "outTrackId": id.String(), "cardData": map[string]any{}, - "openSpaceId": openSpaceId, + "openSpaceId": openSpaceID, "userIdType": 1, "imGroupOpenDeliverModel": imGroupOpenDeliverModel, "imGroupOpenSpaceModel": imGroupOpenSpaceModel, @@ -181,13 +186,13 @@ func (c *Client) CardCreateAndDeliver(ctx context.Context, chatbot *chatbot.BotC resp := struct { Result struct { DeliverResults []struct { - SpaceId string `json:"spaceId"` + SpaceID string `json:"spaceId"` SpaceType string `json:"spaceType"` Success bool `json:"success"` - CarrierId string `json:"carrierId"` + CarrierID string `json:"carrierId"` ErrorMsg string `json:"errorMsg"` } `json:"deliverResults"` - OutTrackId string `json:"outTrackId"` + OutTrackID string `json:"outTrackId"` } `json:"result"` Success bool `json:"success"` }{} @@ -195,17 +200,17 @@ func (c *Client) CardCreateAndDeliver(ctx context.Context, chatbot *chatbot.BotC return "", err } if resp.Success { - return resp.Result.OutTrackId, nil + return resp.Result.OutTrackID, nil } return "", errors.New("failed to create and deliver card instance") } // PrivateChatMessages sends a message to a user in a private chat. -func (c *Client) PrivateChatMessages(ctx context.Context, msgType MessageType, openConversationId, content string) error { +func (c *Client) PrivateChatMessages(ctx context.Context, msgType MessageType, openConversationID, content string) error { body := map[string]any{ "msgKey": msgType, "msgParam": c.buildSendMessages(msgType, content), - "openConversationId": openConversationId, + "openConversationId": openConversationID, "robotCode": c.robotCode, } return c.httpRequest(ctx, http.MethodPost, privateChatMessages, body, nil) @@ -239,7 +244,7 @@ func (c *Client) httpRequest(ctx context.Context, method, path string, body inte if path == accessToken { token = "" } else { - token, err = c.GetToken() + token, err = c.GetToken(ctx) if err != nil { return err } diff --git a/pkg/channels/dingtalk/dingtalk.go b/pkg/channels/dingtalk/dingtalk.go index 2af5c55cd..093f10b7d 100644 --- a/pkg/channels/dingtalk/dingtalk.go +++ b/pkg/channels/dingtalk/dingtalk.go @@ -117,15 +117,15 @@ func (c *DingTalkChannel) Send(ctx context.Context, msg bus.OutboundMessage) err return channels.ErrNotRunning } // Check if we have a card instance ID for this chat (indicating we can send a card reply) - cardInstanceIdRaw, ok := c.cardInstanceIDs.LoadAndDelete(msg.ChatID) + cardInstanceIDRaw, ok := c.cardInstanceIDs.Load(msg.ChatID) if !ok { return c.SendDirectReply(ctx, msg) } - cardInstanceId, ok := cardInstanceIdRaw.(string) + cardInstanceID, ok := cardInstanceIDRaw.(string) if !ok { return c.SendDirectReply(ctx, msg) } - return c.SendCardReply(ctx, cardInstanceId, msg.Content) + return c.SendCardReply(ctx, cardInstanceID, msg.Content) } // onChatBotMessageReceived implements the IChatBotMessageHandler function signature @@ -197,13 +197,14 @@ func (c *DingTalkChannel) onChatBotMessageReceived( return nil, nil } + // Try to create and deliver card (optional feature) + // If it fails, log the error but continue with normal message handling if err := c.tryCardCreateAndDeliver(ctx, chatID, data); err != nil { - logger.ErrorCF("dingtalk", "Failed to create or deliver card", map[string]any{ + logger.WarnCF("dingtalk", "Failed to create or deliver card, falling back to direct reply", map[string]any{ "error": err.Error(), "chat_id": chatID, "sender_id": senderID, }) - return nil, nil } // Store the session webhook for this chat so we can reply later c.sessionWebhooks.Store(chatID, data.SessionWebhook) @@ -218,7 +219,7 @@ func (c *DingTalkChannel) onChatBotMessageReceived( // SendDirectReply sends a direct reply using the session webhook func (c *DingTalkChannel) SendDirectReply(ctx context.Context, msg bus.OutboundMessage) error { // Get session webhook from storage - sessionWebhookRaw, ok := c.sessionWebhooks.LoadAndDelete(msg.ChatID) + sessionWebhookRaw, ok := c.sessionWebhooks.Load(msg.ChatID) if !ok { return fmt.Errorf("no session_webhook found for chat %s, cannot send message", msg.ChatID) } @@ -250,21 +251,18 @@ func (c *DingTalkChannel) SendDirectReply(ctx context.Context, msg bus.OutboundM return nil } -func (c *DingTalkChannel) SendCardReply(ctx context.Context, cardInstanceId, content string) error { - if err := c.client.CardStreaming(ctx, cardInstanceId, content); err != nil { - return err - } - return nil +func (c *DingTalkChannel) SendCardReply(ctx context.Context, cardInstanceID, content string) error { + return c.client.CardStreaming(ctx, cardInstanceID, content) } func (c *DingTalkChannel) tryCardCreateAndDeliver(ctx context.Context, chatID string, data *chatbot.BotCallbackDataModel) error { if c.config.CardTemplateID == "" { return nil } - cardInstanceId, err := c.client.CardCreateAndDeliver(ctx, data) + cardInstanceID, err := c.client.CardCreateAndDeliver(ctx, data) if err != nil { return err } - c.cardInstanceIDs.Store(chatID, cardInstanceId) + c.cardInstanceIDs.Store(chatID, cardInstanceID) return nil }