opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(channels): improve dingtalk client robustness

Browse source 
- 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
This commit is contained in:
zhaoyunxing 2026-03-12 21:25:47 +08:00
parent b7961434ae
commit fba80d0596
2 changed files with 36 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
pkg/channels/dingtalk/client.go
View file

@ -8,6 +8,7 @@ import (
"fmt" "fmt"
"io" "io"
"net/http" "net/http"
"sync"
"time" "time"
"github.com/google/uuid" "github.com/google/uuid"
@ -50,6 +51,7 @@ type Client struct {
expires time.Time expires time.Time
client *http.Client 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. // 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. // 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) { if time.Now().Before(c.expires) {
return c.token, nil return c.token, nil
} }
@ -81,7 +86,7 @@ func (c *Client) GetToken() (string, error) {
Expires int64 `json:"expireIn"` Expires int64 `json:"expireIn"`
Token string `json:"accessToken"` 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 { if err != nil {
return "", err return "", err
} }
@ -91,24 +96,24 @@ func (c *Client) GetToken() (string, error) {
} }
// BatchSendMessages sends a message to multiple users in a batch. // 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{ body := map[string]any{
"robotCode": c.robotCode, "robotCode": c.robotCode,
"msgKey": msgType, "msgKey": msgType,
"userIds": userIds, "userIds": userIDs,
"msgParam": c.buildSendMessages(msgType, content), "msgParam": c.buildSendMessages(msgType, content),
} }
return c.httpRequest(ctx, http.MethodPost, batchSendMessages, body, nil) return c.httpRequest(ctx, http.MethodPost, batchSendMessages, body, nil)
} }
// CardStreaming updates the content of a card instance identified by cardInstanceId. // CardStreaming updates the content of a card instance identified by cardInstanceID.
func (c *Client) CardStreaming(ctx context.Context, cardInstanceId, content string) error { func (c *Client) CardStreaming(ctx context.Context, cardInstanceID, content string) error {
id, err := uuid.NewUUID() id, err := uuid.NewUUID()
if err != nil { if err != nil {
return err return err
} }
body := map[string]any{ body := map[string]any{
"outTrackId": cardInstanceId, "outTrackId": cardInstanceID,
"guid": id.String(), "guid": id.String(),
"key": c.cardTemplateContentKey, "key": c.cardTemplateContentKey,
"content": content, "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. // 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. // If the chatbot is in a group conversation, the card will be delivered to the group; otherwise, it will be delivered to the user.
// <a href="https://open.dingtalk.com/document/development/create-and-deliver-cards">Card Delivery API Documentation</a> // <a href="https://open.dingtalk.com/document/development/create-and-deliver-cards">Card Delivery API Documentation</a>
func (c *Client) CardCreateAndDeliver(ctx context.Context, chatbot *chatbot.BotCallbackDataModel) (string, error) { 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 ( var (
group = chatbot.ConversationType == "2" group = chatbot.ConversationType == "2"
openSpaceId = "dtv1.card//IM_ROBOT." + chatbot.SenderStaffId openSpaceID = "dtv1.card//IM_ROBOT." + chatbot.SenderStaffId
imRobotOpenDeliverModel = map[string]any{ imRobotOpenDeliverModel = map[string]any{
"spaceType": "IM_ROBOT", "spaceType": "IM_ROBOT",
} }
@ -138,7 +143,7 @@ func (c *Client) CardCreateAndDeliver(ctx context.Context, chatbot *chatbot.BotC
"supportForward": true, "supportForward": true,
} }
imGroupOpenDeliverModel = map[string]any{ 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 return "", err
} }
if group { if group {
openSpaceId = "dtv1.card//IM_GROUP." + chatbot.ConversationId openSpaceID = "dtv1.card//IM_GROUP." + chatbot.ConversationId
imRobotOpenDeliverModel = map[string]any{} imRobotOpenDeliverModel = map[string]any{}
imGroupOpenDeliverModel["robotCode"] = c.clientID imGroupOpenDeliverModel["robotCode"] = c.robotCode
} }
body := map[string]any{ body := map[string]any{
"cardTemplateId": c.cardTemplateID, "cardTemplateId": c.cardTemplateID,
"outTrackId": id.String(), "outTrackId": id.String(),
"cardData": map[string]any{}, "cardData": map[string]any{},
"openSpaceId": openSpaceId, "openSpaceId": openSpaceID,
"userIdType": 1, "userIdType": 1,
"imGroupOpenDeliverModel": imGroupOpenDeliverModel, "imGroupOpenDeliverModel": imGroupOpenDeliverModel,
"imGroupOpenSpaceModel": imGroupOpenSpaceModel, "imGroupOpenSpaceModel": imGroupOpenSpaceModel,
@ -181,13 +186,13 @@ func (c *Client) CardCreateAndDeliver(ctx context.Context, chatbot *chatbot.BotC
resp := struct { resp := struct {
Result struct { Result struct {
DeliverResults []struct { DeliverResults []struct {
SpaceId string `json:"spaceId"` SpaceID string `json:"spaceId"`
SpaceType string `json:"spaceType"` SpaceType string `json:"spaceType"`
Success bool `json:"success"` Success bool `json:"success"`
CarrierId string `json:"carrierId"` CarrierID string `json:"carrierId"`
ErrorMsg string `json:"errorMsg"` ErrorMsg string `json:"errorMsg"`
} `json:"deliverResults"` } `json:"deliverResults"`
OutTrackId string `json:"outTrackId"` OutTrackID string `json:"outTrackId"`
} `json:"result"` } `json:"result"`
Success bool `json:"success"` Success bool `json:"success"`
}{} }{}
@ -195,17 +200,17 @@ func (c *Client) CardCreateAndDeliver(ctx context.Context, chatbot *chatbot.BotC
return "", err return "", err
} }
if resp.Success { if resp.Success {
return resp.Result.OutTrackId, nil return resp.Result.OutTrackID, nil
} }
return "", errors.New("failed to create and deliver card instance") return "", errors.New("failed to create and deliver card instance")
} }
// PrivateChatMessages sends a message to a user in a private chat. // 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{ body := map[string]any{
"msgKey": msgType, "msgKey": msgType,
"msgParam": c.buildSendMessages(msgType, content), "msgParam": c.buildSendMessages(msgType, content),
"openConversationId": openConversationId, "openConversationId": openConversationID,
"robotCode": c.robotCode, "robotCode": c.robotCode,
} }
return c.httpRequest(ctx, http.MethodPost, privateChatMessages, body, nil) 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 { if path == accessToken {
token = "" token = ""
} else { } else {
token, err = c.GetToken() token, err = c.GetToken(ctx)
if err != nil { if err != nil {
return err return err
} }

24
pkg/channels/dingtalk/dingtalk.go
View file

@ -117,15 +117,15 @@ func (c *DingTalkChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
return channels.ErrNotRunning return channels.ErrNotRunning
} }
// Check if we have a card instance ID for this chat (indicating we can send a card reply) // 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 { if !ok {
return c.SendDirectReply(ctx, msg) return c.SendDirectReply(ctx, msg)
} }
cardInstanceId, ok := cardInstanceIdRaw.(string) cardInstanceID, ok := cardInstanceIDRaw.(string)
if !ok { if !ok {
return c.SendDirectReply(ctx, msg) 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 // onChatBotMessageReceived implements the IChatBotMessageHandler function signature
@ -197,13 +197,14 @@ func (c *DingTalkChannel) onChatBotMessageReceived(
return nil, nil 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 { 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(), "error": err.Error(),
"chat_id": chatID, "chat_id": chatID,
"sender_id": senderID, "sender_id": senderID,
}) })
return nil, nil
} }
// Store the session webhook for this chat so we can reply later // Store the session webhook for this chat so we can reply later
c.sessionWebhooks.Store(chatID, data.SessionWebhook) c.sessionWebhooks.Store(chatID, data.SessionWebhook)
@ -218,7 +219,7 @@ func (c *DingTalkChannel) onChatBotMessageReceived(
// SendDirectReply sends a direct reply using the session webhook // SendDirectReply sends a direct reply using the session webhook
func (c *DingTalkChannel) SendDirectReply(ctx context.Context, msg bus.OutboundMessage) error { func (c *DingTalkChannel) SendDirectReply(ctx context.Context, msg bus.OutboundMessage) error {
// Get session webhook from storage // Get session webhook from storage
sessionWebhookRaw, ok := c.sessionWebhooks.LoadAndDelete(msg.ChatID) sessionWebhookRaw, ok := c.sessionWebhooks.Load(msg.ChatID)
if !ok { if !ok {
return fmt.Errorf("no session_webhook found for chat %s, cannot send message", msg.ChatID) 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 return nil
} }
func (c *DingTalkChannel) SendCardReply(ctx context.Context, cardInstanceId, content string) error { func (c *DingTalkChannel) SendCardReply(ctx context.Context, cardInstanceID, content string) error {
if err := c.client.CardStreaming(ctx, cardInstanceId, content); err != nil { return c.client.CardStreaming(ctx, cardInstanceID, content)
return err
}
return nil
} }
func (c *DingTalkChannel) tryCardCreateAndDeliver(ctx context.Context, chatID string, data *chatbot.BotCallbackDataModel) error { func (c *DingTalkChannel) tryCardCreateAndDeliver(ctx context.Context, chatID string, data *chatbot.BotCallbackDataModel) error {
if c.config.CardTemplateID == "" { if c.config.CardTemplateID == "" {
return nil return nil
} }
cardInstanceId, err := c.client.CardCreateAndDeliver(ctx, data) cardInstanceID, err := c.client.CardCreateAndDeliver(ctx, data)
if err != nil { if err != nil {
return err return err
} }
c.cardInstanceIDs.Store(chatID, cardInstanceId) c.cardInstanceIDs.Store(chatID, cardInstanceID)
return nil return nil
} }