feat: support batch message recall

server/im_http.go

@@ -397,6 +397,17 @@ type imMessageRecallRequest struct {
 	Reason  string `json:"reason,omitempty"`
 }
 
+type imMessageBatchRecallRequest struct {
+	Items  []imMessageRecallItem `json:"items"`
+	Reason string                `json:"reason,omitempty"`
+}
+
+type imMessageRecallItem struct {
+	Seq     int64  `json:"seq"`
+	Version int    `json:"version,omitempty"`
+	Reason  string `json:"reason,omitempty"`
+}
+
 type imMessageRevisionRequest struct {
 	Version   int         `json:"version"`
 	Content   any         `json:"content"`
@@ -4071,6 +4082,8 @@ func serveIMTopicMessageAction(wrt http.ResponseWriter, req *http.Request) {
 		serveIMTopicMedia(wrt, req)
 	case strings.HasSuffix(req.URL.Path, "/receipts"):
 		serveIMMessageReceipts(wrt, req)
+	case strings.HasSuffix(req.URL.Path, "/messages/recall"):
+		serveIMMessageBatchRecall(wrt, req)
 	case strings.HasSuffix(req.URL.Path, "/recall"):
 		serveIMMessageRecall(wrt, req)
 	case strings.HasSuffix(req.URL.Path, "/revisions"):
@@ -4366,6 +4379,71 @@ func serveIMMessageRecall(wrt http.ResponseWriter, req *http.Request) {
 	writeIMSuccess(wrt, req, messageMutationPayload(updated, "recalled"))
 }
 
+func serveIMMessageBatchRecall(wrt http.ResponseWriter, req *http.Request) {
+	if req.Method != http.MethodPost {
+		writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
+		return
+	}
+	principal, ok := authenticateIMHTTPBearer(wrt, req)
+	if !ok {
+		return
+	}
+	topic, ok := topicMessagesActionPath(req.URL.Path, "recall")
+	if !ok {
+		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
+		return
+	}
+	topic = normalizeConversationSettingsTopic(topic, principal.Uid)
+	var body imMessageBatchRecallRequest
+	if !decodeIMJSONBody(wrt, req, &body) {
+		return
+	}
+	if len(body.Items) == 0 || len(body.Items) > 50 {
+		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "items must contain 1-50 messages", nil)
+		return
+	}
+	items := make([]map[string]any, 0, len(body.Items))
+	recalledAt := time.Now().UTC().Round(time.Millisecond)
+	for _, item := range body.Items {
+		if item.Seq <= 0 {
+			items = append(items, messageBatchRecallErrorItem(item.Seq, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request"))
+			continue
+		}
+		msg, authorized, err := authorizeIMMessageMutation(principal, topic, item.Seq, true)
+		if err != nil {
+			items = append(items, messageMutationErrorItem(item.Seq, err))
+			continue
+		}
+		if !authorized {
+			items = append(items, messageBatchRecallErrorItem(item.Seq, http.StatusForbidden, 30005, imReasonPermissionDenied, "message recall permission required"))
+			continue
+		}
+		if time.Since(msg.CreatedAt) > 120*time.Second {
+			items = append(items, messageBatchRecallErrorItem(item.Seq, http.StatusForbidden, 30005, imReasonPermissionDenied, "message recall window expired"))
+			continue
+		}
+		version := item.Version
+		if version <= 0 {
+			version = msg.Version
+		}
+		updated, err := store.Messages.Recall(principal.TenantID, topic, item.Seq, principal.Uid, version, recalledAt)
+		if err != nil {
+			items = append(items, messageMutationErrorItem(item.Seq, err))
+			continue
+		}
+		result := messageMutationPayload(updated, "recalled")
+		if imReliableHTTPEventsEnabled(req) {
+			eventID := stableIMEventID("message", topic, item.Seq, updated.Version, "recalled")
+			if err = enqueueIMBusinessEvent(principal.TenantID, eventID, "topic", topic, principal.Uid,
+				"message.recalled", map[string]any{"topic": topic, "seq": item.Seq, "version": updated.Version}); err != nil {
+				result["event_status"] = "enqueue_failed"
+			}
+		}
+		items = append(items, result)
+	}
+	writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": items})
+}
+
 func serveIMMessageRevision(wrt http.ResponseWriter, req *http.Request) {
 	if req.Method != http.MethodPost {
 		writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
@@ -8500,6 +8578,32 @@ func writeIMMessageMutationError(wrt http.ResponseWriter, req *http.Request, err
 	}
 }
 
+func messageMutationErrorItem(seq int64, err error) map[string]any {
+	switch {
+	case errors.Is(err, types.ErrNotFound):
+		return messageBatchRecallErrorItem(seq, http.StatusNotFound, 30004, imReasonMalformedRequest, "message not found")
+	case errors.Is(err, types.ErrDuplicate):
+		return messageBatchRecallErrorItem(seq, http.StatusConflict, 30006, "VERSION_CONFLICT", "message version conflict")
+	case errors.Is(err, types.ErrPermissionDenied):
+		return messageBatchRecallErrorItem(seq, http.StatusForbidden, 30005, imReasonPermissionDenied, "topic read permission required")
+	case errors.Is(err, types.ErrMalformed):
+		return messageBatchRecallErrorItem(seq, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request")
+	default:
+		return messageBatchRecallErrorItem(seq, http.StatusInternalServerError, 90000, imReasonInternal, "internal error")
+	}
+}
+
+func messageBatchRecallErrorItem(seq int64, status, code int, reason, message string) map[string]any {
+	return map[string]any{
+		"seq":     seq,
+		"status":  "failed",
+		"http":    status,
+		"code":    code,
+		"reason":  reason,
+		"message": message,
+	}
+}
+
 func messageMutationPayload(msg *types.Message, status string) map[string]any {
 	data := map[string]any{
 		"status":     status,
@@ -11664,6 +11768,20 @@ func messageActionPath(path, action string) (string, int64, bool) {
 	return topic, seq, true
 }
 
+func topicMessagesActionPath(path, action string) (string, bool) {
+	const marker = "/v1/im/topics/"
+	idx := strings.Index(path, marker)
+	if idx < 0 {
+		return "", false
+	}
+	rest := strings.TrimPrefix(path[idx:], marker)
+	topic, suffix, ok := strings.Cut(rest, "/messages/")
+	if !ok || topic == "" || suffix != action {
+		return "", false
+	}
+	return topic, true
+}
+
 func conversationContextTopicFromPath(path string) string {
 	const prefix = "/v1/im/conversations/"
 	const suffix = "/context"