fix: acknowledge DingTalk callbacks before async processing

internal/channels/dingtalk/stream.go

@@ -5,7 +5,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log"
 	"strings"
+	"sync"
+	"time"
 
 	sdkcard "github.com/open-dingtalk/dingtalk-stream-sdk-go/card"
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot"
@@ -222,18 +225,64 @@ func extractDownloadCode(raw interface{}) string {
 	return strings.TrimSpace(nested.Content.DownloadCode)
 }
 
-// StreamClient wraps the official DingTalk Stream SDK client. It wires
-// robot callbacks into the existing Handler synchronously (no goroutines,
-// no queue). The SDK default auto-reconnect stays enabled while running;
-// Close disables it so the connection stays closed.
+const callbackDedupTTL = 5 * time.Minute
+
+type chatCallbackHandler interface {
+	Handle(context.Context, BotCallback) error
+}
+
+// newAsyncChatCallbackRouter acknowledges valid DingTalk frames immediately,
+// then processes them outside the SDK callback lifetime. DingTalk retries slow
+// callbacks, so retaining MsgIDs for the platform's five-minute retry window
+// prevents a retry from creating a second streaming card.
+func newAsyncChatCallbackRouter(handler chatCallbackHandler) func(context.Context, *chatbot.BotCallbackDataModel) ([]byte, error) {
+	var mu sync.Mutex
+	seen := make(map[string]time.Time)
+	return func(ctx context.Context, data *chatbot.BotCallbackDataModel) ([]byte, error) {
+		cb, err := BotCallbackFromSDK(data)
+		if err != nil {
+			return nil, err
+		}
+		messageID := strings.TrimSpace(cb.MsgID)
+		if messageID == "" {
+			return nil, errors.New("dingtalk stream: msgId is required")
+		}
+
+		now := time.Now()
+		mu.Lock()
+		for id, expiresAt := range seen {
+			if !expiresAt.After(now) {
+				delete(seen, id)
+			}
+		}
+		if expiresAt, exists := seen[messageID]; exists && expiresAt.After(now) {
+			mu.Unlock()
+			return nil, nil
+		}
+		seen[messageID] = now.Add(callbackDedupTTL)
+		mu.Unlock()
+
+		workCtx := context.WithoutCancel(ctx)
+		go func() {
+			if err := handler.Handle(workCtx, cb); err != nil {
+				log.Printf("DingTalk async callback failed: stage=handler")
+			}
+		}()
+		return nil, nil
+	}
+}
+
+// StreamClient wraps the official DingTalk Stream SDK client. Robot callbacks
+// are acknowledged before the potentially slow QA/card work begins. The SDK
+// default auto-reconnect stays enabled while running; Close disables it so the
+// connection stays closed.
 type StreamClient struct {
 	inner *dtsdk.StreamClient
 }
 
 // NewStreamClient validates the config, builds the SDK client with the app
 // credential and registers the chatbot callback router. The handler is
-// invoked synchronously from the SDK callback; its return value only
-// acknowledges the frame, replies always go through the session webhook.
+// acknowledged immediately; replies always go through the session webhook.
 func NewStreamClient(config StreamConfig, handler *Handler, feedbackHandlers ...*CardFeedbackHandler) (*StreamClient, error) {
 	if err := config.Validate(); err != nil {
 		return nil, err
@@ -256,16 +305,7 @@ func NewStreamClient(config StreamConfig, handler *Handler, feedbackHandlers ...
 	inner := dtsdk.NewStreamClient(
 		dtsdk.WithAppCredential(cred),
 	)
-	inner.RegisterChatBotCallbackRouter(func(ctx context.Context, data *chatbot.BotCallbackDataModel) ([]byte, error) {
-		cb, err := BotCallbackFromSDK(data)
-		if err != nil {
-			return nil, err
-		}
-		if err := handler.Handle(ctx, cb); err != nil {
-			return nil, err
-		}
-		return nil, nil
-	})
+	inner.RegisterChatBotCallbackRouter(newAsyncChatCallbackRouter(handler))
 	if len(feedbackHandlers) == 1 {
 		inner.RegisterCardCallbackRouter(func(ctx context.Context, request *sdkcard.CardRequest) (*sdkcard.CardResponse, error) {
 			return feedbackHandlers[0].HandleSDK(ctx, request)

internal/channels/dingtalk/stream_test.go

@@ -1,12 +1,79 @@
 package dingtalk
 
 import (
+	"context"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot"
 )
 
+type blockingCallbackHandler struct {
+	started chan struct{}
+	release chan struct{}
+	mu      sync.Mutex
+	calls   int
+}
+
+func (h *blockingCallbackHandler) Handle(_ context.Context, _ BotCallback) error {
+	h.mu.Lock()
+	h.calls++
+	if h.calls == 1 {
+		close(h.started)
+	}
+	h.mu.Unlock()
+	<-h.release
+	return nil
+}
+
+func (h *blockingCallbackHandler) callCount() int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.calls
+}
+
+func TestAsyncChatCallbackAcknowledgesBeforeWorkAndDeduplicatesInflightMessage(t *testing.T) {
+	handler := &blockingCallbackHandler{started: make(chan struct{}), release: make(chan struct{})}
+	router := newAsyncChatCallbackRouter(handler)
+	data := &chatbot.BotCallbackDataModel{
+		ConversationType: "1",
+		Msgtype:          "text",
+		MsgId:            "same-message",
+		SenderStaffId:    "staff-1",
+		SessionWebhook:   "https://oapi.dingtalk.com/robot/sendBySession?session=test",
+		Text:             chatbot.BotCallbackDataTextModel{Content: "hello"},
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := router(context.Background(), data)
+		done <- err
+	}()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("callback ACK waited for slow QA work")
+	}
+	select {
+	case <-handler.started:
+	case <-time.After(time.Second):
+		t.Fatal("async handler did not start")
+	}
+	if _, err := router(context.Background(), data); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(20 * time.Millisecond)
+	if got := handler.callCount(); got != 1 {
+		t.Fatalf("same in-flight MsgID handled %d times, want 1", got)
+	}
+	close(handler.release)
+}
+
 func TestStreamConfigValidate(t *testing.T) {
 	t.Parallel()
 	tests := []struct {