fix im workbench and profile regressions

2026-09-11 17:25:24 +08:00
parent 1c2132eebc
commit 367bed2940
6 changed files with 702 additions and 71 deletions
@@ -54,6 +54,7 @@ const (
 	imReasonFileUnavailable     = "FILE_UNAVAILABLE"
 	imReasonRateLimited         = "RATE_LIMITED"
 	imReasonResourceNotFound    = "RESOURCE_NOT_FOUND"
+	imReasonValidationFailed    = "VALIDATION_FAILED"
 )

 const (
@@ -1787,6 +1788,11 @@ func serveIMWorkbench(wrt http.ResponseWriter, req *http.Request) {
 		writeIMRelationError(wrt, req, err)
 		return
 	}
+	modulePayloads := workbenchModulePayloads(modules)
+	if err := populateWorkbenchModuleUnreadCounts(principal.TenantID, principal.Uid, modulePayloads); err != nil {
+		writeIMRelationError(wrt, req, err)
+		return
+	}
 	tenant, err := store.Tenants.GetByID(principal.TenantID)
 	if err != nil || tenant == nil || strings.TrimSpace(tenant.Code) == "" {
 		writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
@@ -1794,7 +1800,7 @@ func serveIMWorkbench(wrt http.ResponseWriter, req *http.Request) {
 	}
 	writeIMSuccess(wrt, req, map[string]any{
 		"layout_version": layoutVersion,
-		"modules":        workbenchModulePayloads(modules),
+		"modules":        modulePayloads,
 		"enterprise": map[string]any{
 			"tenant_code": tenant.Code,
 			"feature_flags": map[string]any{
@@ -2193,8 +2199,8 @@ func serveIMMeetings(wrt http.ResponseWriter, req *http.Request) {
 			return
 		}
 	}
-	publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "meeting", "meeting.created", meeting.MeetingID,
-		meetingParticipantUIDs(meeting), "会议通知", meeting.Title, map[string]any{"meeting_id": meeting.MeetingID, "version": meeting.Version, "state": meeting.State})
+	publishIMAuditOnly(req.Context(), principal.TenantID, principal.Uid, "meeting", "meeting.created", meeting.MeetingID,
+		map[string]any{"meeting_id": meeting.MeetingID, "version": meeting.Version, "state": meeting.State})
 	writeIMSuccess(wrt, req, meetingPayloadForTenant(principal.TenantID, meeting, principal.Uid))
 }

@@ -2466,8 +2472,8 @@ func serveIMMeetingUpdate(wrt http.ResponseWriter, req *http.Request, meetingID
 			return
 		}
 	}
-	publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "meeting", "meeting.updated", meeting.MeetingID,
-		meetingParticipantUIDs(meeting), "会议通知", meeting.Title, map[string]any{"meeting_id": meeting.MeetingID, "version": meeting.Version, "state": meeting.State})
+	publishIMAuditOnly(req.Context(), principal.TenantID, principal.Uid, "meeting", "meeting.updated", meeting.MeetingID,
+		map[string]any{"meeting_id": meeting.MeetingID, "version": meeting.Version, "state": meeting.State})
 	writeIMSuccess(wrt, req, meetingPayloadForTenant(principal.TenantID, meeting, principal.Uid))
 }

@@ -2495,8 +2501,8 @@ func serveIMMeetingCancel(wrt http.ResponseWriter, req *http.Request, meetingID
 			return
 		}
 	}
-	publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "meeting", "meeting.canceled", meeting.MeetingID,
-		meetingParticipantUIDs(meeting), "会议通知", meeting.Title, map[string]any{"meeting_id": meeting.MeetingID, "version": meeting.Version, "state": meeting.State})
+	publishIMAuditOnly(req.Context(), principal.TenantID, principal.Uid, "meeting", "meeting.canceled", meeting.MeetingID,
+		map[string]any{"meeting_id": meeting.MeetingID, "version": meeting.Version, "state": meeting.State})
 	writeIMSuccess(wrt, req, meetingPayloadForTenant(principal.TenantID, meeting, principal.Uid))
 }

@@ -2530,8 +2536,8 @@ func serveIMMeetingReminders(wrt http.ResponseWriter, req *http.Request, meeting
 			return
 		}
 	}
-	publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "meeting", "meeting.reminded", meetingID,
-		meetingParticipantUserIDs(participants), "会议提醒", meetingID, map[string]any{"meeting_id": meetingID, "reminded_count": len(participants)})
+	publishIMAuditOnly(req.Context(), principal.TenantID, principal.Uid, "meeting", "meeting.reminded", meetingID,
+		map[string]any{"meeting_id": meetingID, "reminded_count": len(participants)})
 	writeIMSuccess(wrt, req, map[string]any{
 		"meeting_id": meetingID,
 		"reminded": meetingParticipantPayloadsWithSummaries(participants,
@@ -2566,8 +2572,8 @@ func serveIMMeetingMyResponse(wrt http.ResponseWriter, req *http.Request, meetin
 			return
 		}
 	}
-	publishIMWorkbenchAndAudit(req.Context(), principal.TenantID, principal.Uid, "meeting", "meeting.response.updated", meeting.MeetingID,
-		meetingParticipantUIDs(meeting), "会议通知", meeting.Title, map[string]any{"meeting_id": meeting.MeetingID, "version": meeting.Version, "state": meeting.State})
+	publishIMAuditOnly(req.Context(), principal.TenantID, principal.Uid, "meeting", "meeting.response.updated", meeting.MeetingID,
+		map[string]any{"meeting_id": meeting.MeetingID, "version": meeting.Version, "state": meeting.State})
 	writeIMSuccess(wrt, req, meetingPayloadForTenant(principal.TenantID, meeting, principal.Uid))
 }

@@ -3031,6 +3037,16 @@ func serveIMMailboxSubmissions(wrt http.ResponseWriter, req *http.Request) {
 	if !decodeIMJSONBody(wrt, req, &body) {
 		return
 	}
+	subject := strings.TrimSpace(body.Subject)
+	content := strings.TrimSpace(body.Content)
+	if len([]rune(subject)) > 200 {
+		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", map[string]any{"field": "subject", "max_length": 200})
+		return
+	}
+	if len([]rune(content)) > 5000 {
+		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", map[string]any{"field": "content", "max_length": 5000})
+		return
+	}
 	token, err := randomIMToken(12)
 	if err != nil {
 		writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
@@ -3039,8 +3055,8 @@ func serveIMMailboxSubmissions(wrt http.ResponseWriter, req *http.Request) {
 	submission, err := store.Mailbox.Submit(principal.TenantID, principal.Uid, &types.MailboxSubmission{
 		SubmissionID:      "mb_" + token,
 		SenderID:          principal.Uid,
-		Subject:           strings.TrimSpace(body.Subject),
-		Content:           strings.TrimSpace(body.Content),
+		Subject:           subject,
+		Content:           content,
 		Anonymous:         body.Anonymous,
 		AttachmentFileIDs: body.AttachmentFileIDs,
 		NoticeVersion:     strings.TrimSpace(body.NoticeVersion),
@@ -3050,6 +3066,10 @@ func serveIMMailboxSubmissions(wrt http.ResponseWriter, req *http.Request) {
 			writeIMError(wrt, req, http.StatusServiceUnavailable, 30050, "MAILBOX_UNAVAILABLE", "mailbox unavailable", nil)
 			return
 		}
+		if errors.Is(err, types.ErrPolicy) {
+			writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", nil)
+			return
+		}
 		writeIMRelationError(wrt, req, err)
 		return
 	}
@@ -3721,6 +3741,16 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) {
 			results = append(results, map[string]any{"target_topic": target, "status": "failed", "reason": relationErrorReason(err)})
 			continue
 		}
+		var commentResult map[string]any
+		commentSaved := false
+		ensureComment := func() map[string]any {
+			if commentSaved || comment == "" {
+				return commentResult
+			}
+			commentSaved = true
+			commentResult = saveForwardCommentMessage(req.Context(), principal, target, comment, forwardClientMid(req.Header.Get("Idempotency-Key"), "favorite-comment", target))
+			return commentResult
+		}
 		for _, favoriteID := range body.FavoriteIDs {
 			favoriteID = strings.TrimSpace(favoriteID)
 			item, err := store.Favorites.Get(principal.TenantID, principal.Uid, favoriteID)
@@ -3728,35 +3758,39 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) {
 				results = append(results, map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "failed", "reason": relationErrorReason(err)})
 				continue
 			}
+			head, content := favoriteForwardMessageHeadContent(item)
 			msg := &types.Message{
 				TenantID: principal.TenantID,
 				Topic:    target,
 				From:     principal.Uid.String(),
-				Head: types.KVMap{
-					"type":           item.ItemType,
-					"x-im-type":      item.ItemType,
-					"x-im-forwarded": true,
-				},
-				Content: favoriteForwardContent(item, comment),
+				Head:     head,
+				Content:  content,
 			}
 			if err := prepareHTTPForwardMessage(principal.TenantID, msg); err != nil {
 				results = append(results, map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "failed", "reason": relationErrorReason(err)})
 				continue
 			}
+			commentResult = ensureComment()
 			saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, msg, nil, true)
 			if err != nil {
 				if saved != nil && saved.Pending {
-					results = append(results, map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "pending", "seq": saved.SeqId, "reason": "MESSAGE_PERSIST_PENDING"})
+					result := map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "pending", "seq": saved.SeqId, "reason": "MESSAGE_PERSIST_PENDING"}
+					mergeForwardCommentResult(result, commentResult)
+					results = append(results, result)
 					continue
 				}
 				results = append(results, map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "failed", "reason": relationErrorReason(err)})
 				continue
 			}
 			if saved != nil && saved.Pending {
-				results = append(results, map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "pending", "seq": saved.SeqId, "reason": "MESSAGE_PERSIST_PENDING"})
+				result := map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "pending", "seq": saved.SeqId, "reason": "MESSAGE_PERSIST_PENDING"}
+				mergeForwardCommentResult(result, commentResult)
+				results = append(results, result)
 				continue
 			}
-			results = append(results, map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "sent", "seq": saved.SeqId})
+			result := map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "sent", "seq": saved.SeqId}
+			mergeForwardCommentResult(result, commentResult)
+			results = append(results, result)
 		}
 	}
 	writeIMSuccess(wrt, req, map[string]any{"items": results})
@@ -3816,8 +3850,19 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
 			}
 			continue
 		}
+		comment := strings.TrimSpace(body.Comment)
+		var commentResult map[string]any
+		commentSaved := false
+		ensureComment := func() map[string]any {
+			if commentSaved || comment == "" {
+				return commentResult
+			}
+			commentSaved = true
+			commentResult = saveForwardCommentMessage(req.Context(), principal, target, comment, forwardClientMid(req.Header.Get("Idempotency-Key"), "message-comment", target))
+			return commentResult
+		}
 		if mode == "bundle" {
-			bundle, sourceFailures := buildForwardBundle(principal, body.Sources, target, sourceReadable, body.Comment)
+			bundle, sourceFailures := buildForwardBundle(principal, body.Sources, target, sourceReadable, "")
 			if len(sourceFailures) > 0 {
 				failed = append(failed, sourceFailures...)
 				continue
@@ -3834,11 +3879,12 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
 				failed = append(failed, forwardFailurePayload("", 0, target, relationErrorReason(err)))
 				continue
 			}
+			commentResult = ensureComment()
 			saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true)
 			if err != nil {
 				if saved != nil && saved.Pending {
 					forwarded.SeqId = saved.SeqId
-					items = append(items, map[string]any{
+					item := map[string]any{
 						"sources":      bundle["source_refs"],
 						"target_topic": target,
 						"client_mid":   clientMid,
@@ -3847,14 +3893,16 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
 						"reason":       "MESSAGE_PERSIST_PENDING",
 						"mode":         "bundle",
 						"bundle_id":    bundle["bundle_id"],
-					})
+					}
+					mergeForwardCommentResult(item, commentResult)
+					items = append(items, item)
 					continue
 				}
 				failed = append(failed, forwardFailurePayload("", 0, target, relationErrorReason(err)))
 				continue
 			}
 			if saved != nil && saved.Pending {
-				items = append(items, map[string]any{
+				item := map[string]any{
 					"sources":      bundle["source_refs"],
 					"target_topic": target,
 					"client_mid":   clientMid,
@@ -3863,12 +3911,14 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
 					"reason":       "MESSAGE_PERSIST_PENDING",
 					"mode":         "bundle",
 					"bundle_id":    bundle["bundle_id"],
-				})
+				}
+				mergeForwardCommentResult(item, commentResult)
+				items = append(items, item)
 				continue
 			}
 			forwarded.SeqId = saved.SeqId
 			broadcastHTTPForwardMessage(principal.TenantID, forwarded)
-			items = append(items, map[string]any{
+			item := map[string]any{
 				"sources":      bundle["source_refs"],
 				"target_topic": target,
 				"client_mid":   clientMid,
@@ -3876,7 +3926,9 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
 				"status":       "sent",
 				"mode":         "bundle",
 				"bundle_id":    bundle["bundle_id"],
-			})
+			}
+			mergeForwardCommentResult(item, commentResult)
+			items = append(items, item)
 			continue
 		}
 		for _, source := range body.Sources {
@@ -3913,49 +3965,56 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
 				Topic:    target,
 				From:     principal.Uid.String(),
 				Head:     forwardedMessageHead(msg, clientMid),
-				Content:  forwardedMessageContent(msg, body.Comment),
+				Content:  forwardedMessageContent(msg, ""),
 			}
 			if err := prepareHTTPForwardMessage(principal.TenantID, forwarded); err != nil {
 				failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err)))
 				continue
 			}
+			commentResult = ensureComment()
 			saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true)
 			if err != nil {
 				if saved != nil && saved.Pending {
 					forwarded.SeqId = saved.SeqId
-					items = append(items, map[string]any{
+					item := map[string]any{
 						"source":       map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)},
 						"target_topic": target,
 						"client_mid":   clientMid,
 						"seq":          saved.SeqId,
 						"status":       "pending",
 						"reason":       "MESSAGE_PERSIST_PENDING",
-					})
+					}
+					mergeForwardCommentResult(item, commentResult)
+					items = append(items, item)
 					continue
 				}
 				failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err)))
 				continue
 			}
 			if saved != nil && saved.Pending {
-				items = append(items, map[string]any{
+				item := map[string]any{
 					"source":       map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)},
 					"target_topic": target,
 					"client_mid":   clientMid,
 					"seq":          saved.SeqId,
 					"status":       "pending",
 					"reason":       "MESSAGE_PERSIST_PENDING",
-				})
+				}
+				mergeForwardCommentResult(item, commentResult)
+				items = append(items, item)
 				continue
 			}
 			forwarded.SeqId = saved.SeqId
 			broadcastHTTPForwardMessage(principal.TenantID, forwarded)
-			items = append(items, map[string]any{
+			item := map[string]any{
 				"source":       map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)},
 				"target_topic": target,
 				"client_mid":   clientMid,
 				"seq":          saved.SeqId,
 				"status":       "sent",
-			})
+			}
+			mergeForwardCommentResult(item, commentResult)
+			items = append(items, item)
 		}
 	}
 	writeIMSuccess(wrt, req, map[string]any{"items": items, "failed": failed})
@@ -5376,16 +5435,55 @@ func serveIMGroupFiles(wrt http.ResponseWriter, req *http.Request, topic, tail s
 			writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "file_id"})
 			return
 		}
+		fileName := strings.TrimSpace(body.FileName)
+		mimeType := strings.TrimSpace(body.MIMEType)
+		fileType := normalizeIMFileType(body.FileType)
+		if fileName == "" || len([]rune(fileName)) > 255 {
+			writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", map[string]any{"field": "file_name", "max_length": 255})
+			return
+		}
+		if len([]rune(mimeType)) > 128 {
+			writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", map[string]any{"field": "mime_type", "max_length": 128})
+			return
+		}
+		if body.FileSize < 0 {
+			writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", map[string]any{"field": "file_size"})
+			return
+		}
+		uploaded, err := store.Files.Get(principal.TenantID, fileID)
+		if err != nil {
+			writeIMRelationError(wrt, req, err)
+			return
+		}
+		if uploaded == nil || uploaded.Status != types.UploadCompleted {
+			writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonFileUnavailable, "file unavailable", map[string]any{"field": "file_id"})
+			return
+		}
+		if body.FileSize > 0 && uploaded.Size > 0 && body.FileSize != uploaded.Size {
+			writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonValidationFailed, "validation failed", map[string]any{"field": "file_size"})
+			return
+		}
+		if globals.maxFileUploadSize > 0 && uploaded.Size > globals.maxFileUploadSize {
+			writeIMError(wrt, req, http.StatusBadRequest, 30046, "FILE_TOO_LARGE", "file too large", map[string]any{"field": "file_size", "max_size": globals.maxFileUploadSize})
+			return
+		}
+		if mimeType == "" {
+			mimeType = uploaded.MimeType
+		}
+		fileSize := body.FileSize
+		if fileSize == 0 {
+			fileSize = uploaded.Size
+		}
 		file, err := store.GroupBusiness.CreateFile(principal.TenantID, &types.GroupFile{
 			Topic:          topic,
 			FileID:         fileID,
 			SourceTopic:    strings.TrimSpace(body.SourceTopic),
 			SourceSeq:      body.SourceSeq,
 			UploaderID:     principal.Uid,
-			FileName:       strings.TrimSpace(body.FileName),
-			MIMEType:       strings.TrimSpace(body.MIMEType),
-			FileType:       normalizeIMFileType(body.FileType),
-			FileSize:       body.FileSize,
+			FileName:       fileName,
+			MIMEType:       mimeType,
+			FileType:       fileType,
+			FileSize:       fileSize,
 			Status:         "active",
 			IdempotencyKey: idempotencyKey,
 		})
@@ -5667,10 +5765,15 @@ func serveIMGroupAnnouncements(wrt http.ResponseWriter, req *http.Request, topic
 		}
 		writeIMSuccess(wrt, req, map[string]any{"announcement_id": announcementID, "read_at": now.Format(time.RFC3339Nano)})
 	case strings.HasPrefix(tail, "announcements/") && req.Method == http.MethodGet:
-		ann, err := store.GroupBusiness.LatestAnnouncement(principal.TenantID, topic)
-		if err != nil && !errors.Is(err, types.ErrNotFound) {
-			logs.Warn.Println("im group announcement get: returning empty after store error", principal.TenantID, topic, err)
-			ann = nil
+		announcementID := strings.TrimSpace(strings.TrimPrefix(tail, "announcements/"))
+		if announcementID == "" || strings.Contains(announcementID, "/") {
+			writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "announcement_id"})
+			return
+		}
+		ann, err := store.GroupBusiness.GetAnnouncement(principal.TenantID, topic, announcementID)
+		if err != nil {
+			writeIMRelationError(wrt, req, err)
+			return
 		}
 		writeIMSuccess(wrt, req, groupAnnouncementPayloadWithPermissions(topic, ann, selfSub))
 	default:
@@ -8811,6 +8914,10 @@ func publishIMWorkbenchAndAudit(ctx context.Context, tenantID types.TenantID, ac
 	if err := enqueueKafkaWorkbenchEvent(ctx, tenantID, actor, module, action, resourceID, targets, title, summary, data); err != nil {
 		logs.Warn.Printf("im http: failed to enqueue workbench event action=%s resource=%s: %v", action, resourceID, err)
 	}
+	publishIMAuditOnly(ctx, tenantID, actor, module, action, resourceID, data)
+}
+
+func publishIMAuditOnly(ctx context.Context, tenantID types.TenantID, actor types.Uid, module, action, resourceID string, data map[string]any) {
 	if err := enqueueKafkaAuditEvent(ctx, tenantID, "user", actor, action, module, resourceID, "succeeded", "", data); err != nil {
 		logs.Warn.Printf("im http: failed to enqueue audit event action=%s resource=%s: %v", action, resourceID, err)
 	}
@@ -9743,6 +9850,35 @@ func favoriteForwardContent(item *types.FavoriteItem, comment string) map[string
 	return payload
 }

+func favoriteForwardMessageHeadContent(item *types.FavoriteItem) (types.KVMap, any) {
+	itemType := "text"
+	if item != nil && strings.TrimSpace(item.ItemType) != "" {
+		itemType = strings.TrimSpace(item.ItemType)
+	}
+	head := types.KVMap{"type": itemType, "x-im-type": itemType, "x-im-forwarded": true}
+	if item != nil {
+		if snapshot, ok := item.Snapshot.(map[string]any); ok {
+			if rawHead, ok := snapshot["head"].(map[string]any); ok && len(rawHead) > 0 {
+				head = types.KVMap{}
+				for key, value := range rawHead {
+					head[key] = value
+				}
+			}
+			head["x-im-forwarded"] = true
+			if _, ok := head["x-im-type"]; !ok {
+				head["x-im-type"] = itemType
+			}
+			if _, ok := head["type"]; !ok {
+				head["type"] = itemType
+			}
+			if content, ok := snapshot["content"]; ok {
+				return head, content
+			}
+		}
+	}
+	return forwardBundleMessageHead(""), favoriteForwardContent(item, "")
+}
+
 func forwardFailurePayload(topic string, seq int64, target, reason string) map[string]any {
 	return map[string]any{
 		"source":       map[string]any{"topic": strings.TrimSpace(topic), "seq": seq},
@@ -9752,6 +9888,68 @@ func forwardFailurePayload(topic string, seq int64, target, reason string) map[s
 	}
 }

+func saveForwardCommentMessage(ctx context.Context, principal *imPrincipal, target, comment, clientMid string) map[string]any {
+	comment = strings.TrimSpace(comment)
+	if principal == nil || comment == "" {
+		return nil
+	}
+	msg := &types.Message{
+		TenantID: principal.TenantID,
+		Topic:    target,
+		From:     principal.Uid.String(),
+		Head: types.KVMap{
+			"type":            "text",
+			"x-im-type":       "text",
+			"x-im-forwarded":  true,
+			"x-im-client-mid": clientMid,
+		},
+		Content: map[string]any{"text": comment},
+	}
+	result := map[string]any{"comment_client_mid": clientMid}
+	if err := prepareHTTPForwardMessage(principal.TenantID, msg); err != nil {
+		result["comment_status"] = "failed"
+		result["comment_reason"] = relationErrorReason(err)
+		return result
+	}
+	saved, err := saveMessageMaybeKafka(ctx, principal.TenantID, msg, nil, true)
+	if err != nil {
+		if saved != nil && saved.Pending {
+			result["comment_status"] = "pending"
+			result["comment_seq"] = saved.SeqId
+			result["comment_reason"] = "MESSAGE_PERSIST_PENDING"
+			return result
+		}
+		result["comment_status"] = "failed"
+		result["comment_reason"] = relationErrorReason(err)
+		return result
+	}
+	if saved != nil && saved.Pending {
+		result["comment_status"] = "pending"
+		result["comment_seq"] = saved.SeqId
+		result["comment_reason"] = "MESSAGE_PERSIST_PENDING"
+		return result
+	}
+	if saved != nil {
+		msg.SeqId = saved.SeqId
+		broadcastHTTPForwardMessage(principal.TenantID, msg)
+		result["comment_status"] = "sent"
+		result["comment_seq"] = saved.SeqId
+		return result
+	}
+	result["comment_status"] = "failed"
+	result["comment_reason"] = imReasonInternal
+	return result
+}
+
+func mergeForwardCommentResult(item, comment map[string]any) {
+	if item == nil || comment == nil {
+		return
+	}
+	for key, value := range comment {
+		item[key] = value
+	}
+}
+
 func forwardClientMid(requestKey string, parts ...any) string {
 	requestKey = strings.TrimSpace(requestKey)
 	if requestKey == "" {