feat(wecom): enhance WeCom AIBot WebSocket handling with message deduplication and support for file and video messages

pkg/channels/wecom/aibot_ws.go

@@ -33,6 +33,7 @@ const (
 	wsSubscribeTimeout  = 10 * time.Second
 	wsSendMsgTimeout    = 10 * time.Second
 	wsRespondMsgTimeout = 10 * time.Second
+	wsWelcomeMsgTimeout = 5 * time.Second // WeCom requires welcome reply within 5 seconds
 	wsMaxReconnectWait  = 60 * time.Second
 	wsInitialReconnect  = time.Second
 
@@ -68,6 +69,9 @@ type WeComAIBotWSChannel struct {
 	conn   *websocket.Conn
 	connMu sync.Mutex
 
+	// dedupe prevents duplicate message processing (WeCom may re-deliver).
+	dedupe *MessageDeduplicator
+
 	// reqStates holds per-req_id runtime state.
 	// It unifies active task state and late-reply fallback routing.
 	reqStates   map[string]*wsReqState
@@ -81,14 +85,13 @@ type WeComAIBotWSChannel struct {
 
 // wsTask tracks one in-progress agent reply for a single chat turn.
 type wsTask struct {
-	ReqID       string // req_id echoed in all replies for this turn
+	ReqID    string // req_id echoed in all replies for this turn
-	ChatID      string
+	ChatID   string
-	ChatType    uint32
+	ChatType uint32
-	StreamID    string // our generated stream.id
+	StreamID string      // our generated stream.id
-	CreatedTime time.Time
+	answerCh chan string // agent delivers its reply here via Send()
-	answerCh    chan string // agent delivers its reply here via Send()
+	ctx      context.Context
-	ctx         context.Context
+	cancel   context.CancelFunc
-	cancel      context.CancelFunc
 }
 
 type wsReqState struct {
@@ -183,7 +186,7 @@ type WeComAIBotWSMessage struct {
 		AESKey string `json:"aeskey,omitempty"` // long-connection: per-resource decrypt key
 	} `json:"image,omitempty"`
 	Voice *struct {
-		Text string `json:"text"` // WeCom transcribes voice to text in callbacks
+		Content string `json:"content"` // WeCom transcribes voice to text in callbacks
 	} `json:"voice,omitempty"`
 	Mixed *struct {
 		MsgItem []struct {
@@ -200,6 +203,14 @@ type WeComAIBotWSMessage struct {
 	Event *struct {
 		EventType string `json:"eventtype"`
 	} `json:"event,omitempty"`
+	File *struct {
+		URL    string `json:"url"`
+		AESKey string `json:"aeskey,omitempty"`
+	} `json:"file,omitempty"`
+	Video *struct {
+		URL    string `json:"url"`
+		AESKey string `json:"aeskey,omitempty"`
+	} `json:"video,omitempty"`
 }
 
 // ---- Constructor ----
@@ -221,6 +232,7 @@ func newWeComAIBotWSChannel(
 	return &WeComAIBotWSChannel{
 		BaseChannel: base,
 		config:      cfg,
+		dedupe:      NewMessageDeduplicator(wecomMaxProcessedMessages),
 		reqStates:   make(map[string]*wsReqState),
 		reqPending:  make(map[string]chan wsEnvelope),
 	}, nil
@@ -494,6 +506,8 @@ func (c *WeComAIBotWSChannel) sendAndWait(
 }
 
 // heartbeatLoop sends a ping every wsHeartbeatInterval until conn is closed.
+// It validates the server's pong response via sendAndWait; a failed pong
+// triggers a reconnection by closing the connection.
 func (c *WeComAIBotWSChannel) heartbeatLoop(conn *websocket.Conn) {
 	ticker := time.NewTicker(wsHeartbeatInterval)
 	defer ticker.Stop()
@@ -501,18 +515,23 @@ func (c *WeComAIBotWSChannel) heartbeatLoop(conn *websocket.Conn) {
 		select {
 		case <-ticker.C:
 			reqID := wsGenerateID()
-			data, _ := json.Marshal(wsCommand{
+			resp, err := c.sendAndWait(conn, reqID, wsCommand{
 				Cmd:     "ping",
 				Headers: wsHeaders{ReqID: reqID},
-			})
+			}, wsHeartbeatInterval)
-			c.connMu.Lock()
-			err := conn.WriteMessage(websocket.TextMessage, data)
-			c.connMu.Unlock()
 			if err != nil {
-				logger.WarnCF("wecom_aibot", "Heartbeat write failed", map[string]any{"error": err.Error()})
+				logger.WarnCF("wecom_aibot", "Heartbeat failed, closing connection",
+					map[string]any{"error": err.Error()})
+				conn.Close()
 				return
 			}
-			logger.DebugCF("wecom_aibot", "Heartbeat sent", map[string]any{"req_id": reqID})
+			if resp.ErrCode != 0 {
+				logger.WarnCF("wecom_aibot", "Heartbeat rejected",
+					map[string]any{"errcode": resp.ErrCode, "errmsg": resp.ErrMsg})
+				conn.Close()
+				return
+			}
+			logger.DebugCF("wecom_aibot", "Heartbeat pong received", map[string]any{"req_id": reqID})
 		case <-c.ctx.Done():
 			return
 		}
@@ -540,8 +559,9 @@ func (c *WeComAIBotWSChannel) readLoop(conn *websocket.Conn) error {
 			continue
 		}
 
-		// If there is a waiting sendAndWait() call for this req_id, forward
+		// Command responses have an empty Cmd field; forward to any waiting
-		// the envelope to it.  Command responses have an empty Cmd field.
+		// sendAndWait() call, or silently drop if no one is waiting (e.g.
+		// late responses after timeout).
 		if env.Cmd == "" && env.Headers.ReqID != "" {
 			c.reqPendingMu.Lock()
 			ch, ok := c.reqPending[env.Headers.ReqID]
@@ -551,8 +571,8 @@ func (c *WeComAIBotWSChannel) readLoop(conn *websocket.Conn) error {
 			c.reqPendingMu.Unlock()
 			if ok {
 				ch <- env
-				continue
 			}
+			continue
 		}
 
 		// Dispatch to appropriate handler in a separate goroutine so the
@@ -585,6 +605,13 @@ func (c *WeComAIBotWSChannel) handleMsgCallback(env wsEnvelope) {
 		return
 	}
 
+	// Deduplicate by msgid (WeCom may re-deliver on network issues).
+	if msg.MsgID != "" && !c.dedupe.MarkMessageProcessed(msg.MsgID) {
+		logger.DebugCF("wecom_aibot", "Duplicate message ignored",
+			map[string]any{"msgid": msg.MsgID})
+		return
+	}
+
 	reqID := env.Headers.ReqID
 	switch msg.MsgType {
 	case "text":
@@ -595,6 +622,10 @@ func (c *WeComAIBotWSChannel) handleMsgCallback(env wsEnvelope) {
 		c.handleWSVoiceMessage(reqID, msg)
 	case "mixed":
 		c.handleWSMixedMessage(reqID, msg)
+	case "file":
+		c.handleWSFileMessage(reqID, msg)
+	case "video":
+		c.handleWSVideoMessage(reqID, msg)
 	default:
 		logger.WarnCF("wecom_aibot", "Unsupported message type",
 			map[string]any{"msgtype": msg.MsgType})
@@ -612,6 +643,13 @@ func (c *WeComAIBotWSChannel) handleEventCallback(env wsEnvelope) {
 		return
 	}
 
+	// Deduplicate by msgid.
+	if msg.MsgID != "" && !c.dedupe.MarkMessageProcessed(msg.MsgID) {
+		logger.DebugCF("wecom_aibot", "Duplicate event ignored",
+			map[string]any{"msgid": msg.MsgID})
+		return
+	}
+
 	var eventType string
 	if msg.Event != nil {
 		eventType = msg.Event.EventType
@@ -653,26 +691,30 @@ func (c *WeComAIBotWSChannel) handleWSImageMessage(reqID string, msg WeComAIBotW
 		c.wsSendStreamFinish(reqID, wsGenerateID(), "Image message could not be processed.")
 		return
 	}
+	c.wsHandleMediaMessage(reqID, msg, msg.Image.URL, msg.Image.AESKey, "image")
+}
 
-	chatID := msg.ChatID
+// wsHandleMediaMessage is a shared helper for image, file and video messages.
-	if chatID == "" {
+// It downloads the resource, stores it in MediaStore, and dispatches to the agent.
-		chatID = msg.From.UserID
+func (c *WeComAIBotWSChannel) wsHandleMediaMessage(
-	}
+	reqID string, msg WeComAIBotWSMessage,
+	resourceURL, aesKey, label string,
+) {
+	chatID := wsChatID(msg)
 
 	ctx, cancel := context.WithTimeout(c.ctx, wsImageDownloadTimeout)
 	defer cancel()
 
-	mediaRefs := make([]string, 0, 1)
+	ref, err := c.storeWSImage(ctx, chatID, msg.MsgID, resourceURL, aesKey)
-	ref, err := c.storeWSImage(ctx, chatID, msg.MsgID, msg.Image.URL, msg.Image.AESKey)
 	if err != nil {
-		logger.WarnCF("wecom_aibot", "Failed to download/store WS image",
+		logger.WarnCF("wecom_aibot", "Failed to download/store WS "+label,
-			map[string]any{"error": err.Error(), "url": msg.Image.URL})
+			map[string]any{"error": err.Error(), "url": resourceURL})
-		c.wsSendStreamFinish(reqID, wsGenerateID(), "Image message could not be processed.")
+		c.wsSendStreamFinish(reqID, wsGenerateID(),
+			strings.ToUpper(label[:1])+label[1:]+" message could not be processed.")
 		return
 	}
-	mediaRefs = append(mediaRefs, ref)
 
-	c.dispatchWSAgentTask(reqID, msg, "[image]", mediaRefs)
+	c.dispatchWSAgentTask(reqID, msg, "["+label+"]", []string{ref})
 }
 
 // handleWSMixedMessage handles mixed text+image messages.
@@ -685,10 +727,7 @@ func (c *WeComAIBotWSChannel) handleWSMixedMessage(reqID string, msg WeComAIBotW
 		return
 	}
 
-	chatID := msg.ChatID
+	chatID := wsChatID(msg)
-	if chatID == "" {
-		chatID = msg.From.UserID
-	}
 
 	ctx, cancel := context.WithTimeout(c.ctx, wsImageDownloadTimeout)
 	defer cancel()
@@ -744,24 +783,20 @@ func (c *WeComAIBotWSChannel) dispatchWSAgentTask(
 	}
 	// actualChatID is the real WeCom chat/user ID used for peer identification.
 	// reqID is used as the routing chatID so each turn is independently addressable.
-	actualChatID := msg.ChatID
+	actualChatID := wsChatID(msg)
-	if actualChatID == "" {
-		actualChatID = userID
-	}
 
 	streamID := wsGenerateID()
 	chatType := wsChatTypeValue(msg.ChatType)
 	taskCtx, taskCancel := context.WithCancel(c.ctx)
 
 	task := &wsTask{
-		ReqID:       reqID,
+		ReqID:    reqID,
-		ChatID:      actualChatID,
+		ChatID:   actualChatID,
-		ChatType:    chatType,
+		ChatType: chatType,
-		StreamID:    streamID,
+		StreamID: streamID,
-		CreatedTime: time.Now(),
+		answerCh: make(chan string, 1),
-		answerCh:    make(chan string, 1),
+		ctx:      taskCtx,
-		ctx:         taskCtx,
+		cancel:   taskCancel,
-		cancel:      taskCancel,
 	}
 	// Each req_id is unique per WeCom turn; tasks run concurrently, no cancellation.
 	c.setReqState(reqID, &wsReqState{
@@ -873,25 +908,35 @@ func (c *WeComAIBotWSChannel) dispatchWSAgentTask(
 
 // handleWSVoiceMessage handles voice messages.
 // WeCom transcribes voice to text in the callback; if the transcription is
-// present it is forwarded as a text message.
+// present it is dispatched as plain text to the agent.
 func (c *WeComAIBotWSChannel) handleWSVoiceMessage(reqID string, msg WeComAIBotWSMessage) {
-	if msg.Voice != nil && msg.Voice.Text != "" {
+	if msg.Voice != nil && msg.Voice.Content != "" {
-		c.handleWSTextMessage(reqID, WeComAIBotWSMessage{
+		c.dispatchWSAgentTask(reqID, msg, msg.Voice.Content, nil)
-			MsgID:    msg.MsgID,
-			AIBotID:  msg.AIBotID,
-			ChatID:   msg.ChatID,
-			ChatType: msg.ChatType,
-			From:     msg.From,
-			MsgType:  "text",
-			Text: &struct {
-				Content string `json:"content"`
-			}{Content: msg.Voice.Text},
-		})
 		return
 	}
 	c.wsSendStreamFinish(reqID, wsGenerateID(), "Voice messages are not yet supported.")
 }
 
+// handleWSFileMessage handles file messages.
+func (c *WeComAIBotWSChannel) handleWSFileMessage(reqID string, msg WeComAIBotWSMessage) {
+	if msg.File == nil {
+		logger.WarnC("wecom_aibot", "File message missing file field")
+		c.wsSendStreamFinish(reqID, wsGenerateID(), "File message could not be processed.")
+		return
+	}
+	c.wsHandleMediaMessage(reqID, msg, msg.File.URL, msg.File.AESKey, "file")
+}
+
+// handleWSVideoMessage handles video messages.
+func (c *WeComAIBotWSChannel) handleWSVideoMessage(reqID string, msg WeComAIBotWSMessage) {
+	if msg.Video == nil {
+		logger.WarnC("wecom_aibot", "Video message missing video field")
+		c.wsSendStreamFinish(reqID, wsGenerateID(), "Video message could not be processed.")
+		return
+	}
+	c.wsHandleMediaMessage(reqID, msg, msg.Video.URL, msg.Video.AESKey, "video")
+}
+
 // ---- WebSocket write helpers ----
 
 // wsSendStreamChunk sends an aibot_respond_msg stream frame.
@@ -939,7 +984,7 @@ func (c *WeComAIBotWSChannel) wsSendWelcomeMsg(reqID, content string) {
 			Text:    &wsTextContent{Content: content},
 		},
 	}
-	if err := c.writeWSAndWait(cmd, wsRespondMsgTimeout); err != nil {
+	if err := c.writeWSAndWait(cmd, wsWelcomeMsgTimeout); err != nil {
 		logger.WarnCF("wecom_aibot", "Welcome message ack failed",
 			map[string]any{"req_id": reqID, "error": err.Error()})
 	}
@@ -952,15 +997,7 @@ func (c *WeComAIBotWSChannel) wsSendActivePush(chatID string, chatType uint32, c
 		return fmt.Errorf("chatid is empty")
 	}
 	reqID := wsGenerateID()
-
+	return c.writeWSAndWait(wsCommand{
-	c.connMu.Lock()
-	conn := c.conn
-	c.connMu.Unlock()
-	if conn == nil {
-		return fmt.Errorf("websocket not connected")
-	}
-
-	resp, err := c.sendAndWait(conn, reqID, wsCommand{
 		Cmd:     "aibot_send_msg",
 		Headers: wsHeaders{ReqID: reqID},
 		Body: wsSendMsgBody{
@@ -970,13 +1007,6 @@ func (c *WeComAIBotWSChannel) wsSendActivePush(chatID string, chatType uint32, c
 			Markdown: &wsMarkdownContent{Content: content},
 		},
 	}, wsSendMsgTimeout)
-	if err != nil {
-		return err
-	}
-	if resp.ErrCode != 0 {
-		return fmt.Errorf("aibot_send_msg rejected (errcode=%d): %s", resp.ErrCode, resp.ErrMsg)
-	}
-	return nil
 }
 
 // writeWSAndWait writes cmd to the active connection and validates the command response.
@@ -1071,6 +1101,15 @@ func wsChatTypeValue(chatType string) uint32 {
 	return 1
 }
 
+// wsChatID returns the effective chat ID from a WS message.
+// For group messages it is msg.ChatID; for single chats it falls back to the sender's UserID.
+func wsChatID(msg WeComAIBotWSMessage) string {
+	if msg.ChatID != "" {
+		return msg.ChatID
+	}
+	return msg.From.UserID
+}
+
 // wsGenerateID generates a random 10-character alphanumeric ID.
 // It is package-level (not a method) so it can be shared by both channel modes.
 func wsGenerateID() string {