From 4f6a996922ad2e7fdab1e47e5c2a6d7eeb2159e0 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Tue, 15 Sep 2026 10:46:07 +0800 Subject: [PATCH] feat: support batch message recall --- docs/IM-前后端对接文档.md | 5 +- docs/对接文档最新/IM-前后端对接文档.md | 5 +- server/im_http.go | 118 +++++++++++++++++++++++++ server/im_http_test.go | 77 ++++++++++++++++ 4 files changed, 201 insertions(+), 4 deletions(-) diff --git a/docs/IM-前后端对接文档.md b/docs/IM-前后端对接文档.md index 19c4d3d..af666bd 100644 --- a/docs/IM-前后端对接文档.md +++ b/docs/IM-前后端对接文档.md @@ -1418,7 +1418,8 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `PATCH /v1/im/conversations/{topic}/settings` 待实现 | `{"pinned":true,"mute_notifications":false,"version":1}` | `{"pinned":true,"mute_notifications":false,"message_retention":"default","blocked":false,"remark":"","version":2}` | | `GET /v1/im/conversations/{topic}/messages` 待实现 | Query `?since=12&limit=20` | `{"topic":"p2p...","items":[{"seq":13,"from":"usr...","ts":"2026-08-20T09:00:00Z","head":{"x-im-type":"text"},"content":{"text":"你好"}}],"next_cursor":""}` | | `PUT /v1/im/topics/{topic}/draft` 待实现 | `{"content":"未发送草稿","version":1}` | `{"topic":"grp...","content":"未发送草稿","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | -| `POST /v1/im/topics/{topic}/messages/{seq}/recall` | `{}`；可选 `{"version":1,"reason":"mistake"}` | `{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | +| `POST /v1/im/topics/{topic}/messages/{seq}/recall` | 单条撤回；`topic` 可传 `p2p...` 或 `grp...`；Body `{}`，可选 `{"version":1,"reason":"mistake"}` | `{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | +| `POST /v1/im/topics/{topic}/messages/recall` | 多选批量撤回；`{"items":[{"seq":12,"version":1},{"seq":13}],"reason":"mistake"}`；`items` 1～50 条，`version` 可选 | `{"topic":"grp...","items":[{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"},{"seq":13,"status":"failed","http":409,"code":30006,"reason":"VERSION_CONFLICT","message":"message version conflict"}]}` | | `POST /v1/im/topics/{topic}/messages/{seq}/revisions` | `{"content":{"text":"修改后的文本"},"client_mid":"mid_1","version":1}` | `{"topic":"grp...","seq":12,"status":"edited","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `GET /v1/im/topics/{topic}/messages/{seq}/receipts?cursor=` | Query only | `{"topic":"grp...","seq":12,"items":[{"user_id":"usr...","read_at":"2026-08-20T09:00:00Z","recv_at":"2026-08-20T08:59:00Z"}],"next_cursor":""}` | | `GET /v1/im/topics/{topic}/messages/search?q=&type=text&cursor=` | Query only | `{"topic":"grp...","items":[{"seq":12,"from":"usr...","ts":"2026-08-20T09:00:00Z","snippet":"命中的文本","matches":[{"start":0,"end":2}]}],"next_cursor":""}` | @@ -1426,7 +1427,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `GET /v1/im/topics/{topic}/messages/{seq}/context?before=20&after=20` | Query only | `{"topic":"grp...","anchor_seq":12,"items":[{"seq":11,"content":{"text":"上一条"}},{"seq":12,"content":{"text":"命中消息"}}]}` | | `GET /v1/im/topics/{topic}/media?type=image&cursor=` | Query only | `{"topic":"grp...","items":[{"seq":12,"thumbnail_file_id":"file_thumb","file_id":"file_img","width":1080,"height":720,"ts":"2026-08-20T09:00:00Z","availability":"available"}],"next_cursor":""}` | -消息撤回和编辑成功后，Topic 在线成员会收到 0.5.4.6 定义的 `info.what="im"` 业务事件：`message.recalled` 或 `message.edited`。前端收到后按 `topic + seq + version` 刷新本地消息，不要等待用户重新进入会话才更新。 +消息撤回和编辑成功后，Topic 在线成员会收到 0.5.4.6 定义的 `info.what="im"` 业务事件：`message.recalled` 或 `message.edited`。批量撤回按成功项逐条产生 `message.recalled` 事件；失败项不会产生事件。前端收到后按 `topic + seq + version` 刷新本地消息，不要等待用户重新进入会话才更新。 ## 5. 通讯录、好友关系与黑名单 diff --git a/docs/对接文档最新/IM-前后端对接文档.md b/docs/对接文档最新/IM-前后端对接文档.md index 9a11672..80abb3c 100644 --- a/docs/对接文档最新/IM-前后端对接文档.md +++ b/docs/对接文档最新/IM-前后端对接文档.md @@ -1377,7 +1377,8 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `PATCH /v1/im/conversations/{topic}/settings` 待实现 | `{"pinned":true,"mute_notifications":false,"version":1}` | `{"pinned":true,"mute_notifications":false,"message_retention":"default","blocked":false,"remark":"","version":2}` | | `GET /v1/im/conversations/{topic}/messages` 待实现 | Query `?since=12&limit=20` | `{"topic":"p2p...","items":[{"seq":13,"from":"usr...","ts":"2026-08-20T09:00:00Z","head":{"x-im-type":"text"},"content":{"text":"你好"}}],"next_cursor":""}` | | `PUT /v1/im/topics/{topic}/draft` 待实现 | `{"content":"未发送草稿","version":1}` | `{"topic":"grp...","content":"未发送草稿","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | -| `POST /v1/im/topics/{topic}/messages/{seq}/recall` | `{}`；可选 `{"version":1,"reason":"mistake"}` | `{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | +| `POST /v1/im/topics/{topic}/messages/{seq}/recall` | 单条撤回；`topic` 可传 `p2p...` 或 `grp...`；Body `{}`，可选 `{"version":1,"reason":"mistake"}` | `{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | +| `POST /v1/im/topics/{topic}/messages/recall` | 多选批量撤回；`{"items":[{"seq":12,"version":1},{"seq":13}],"reason":"mistake"}`；`items` 1～50 条，`version` 可选 | `{"topic":"grp...","items":[{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"},{"seq":13,"status":"failed","http":409,"code":30006,"reason":"VERSION_CONFLICT","message":"message version conflict"}]}` | | `POST /v1/im/topics/{topic}/messages/{seq}/revisions` | `{"content":{"text":"修改后的文本"},"client_mid":"mid_1","version":1}` | `{"topic":"grp...","seq":12,"status":"edited","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `GET /v1/im/topics/{topic}/messages/{seq}/receipts?cursor=` | Query only | `{"topic":"grp...","seq":12,"items":[{"user_id":"usr...","read_at":"2026-08-20T09:00:00Z","recv_at":"2026-08-20T08:59:00Z"}],"next_cursor":""}` | | `GET /v1/im/topics/{topic}/messages/search?q=&type=text&cursor=` | Query only | `{"topic":"grp...","items":[{"seq":12,"from":"usr...","ts":"2026-08-20T09:00:00Z","snippet":"命中的文本","matches":[{"start":0,"end":2}]}],"next_cursor":""}` | @@ -1385,7 +1386,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `GET /v1/im/topics/{topic}/messages/{seq}/context?before=20&after=20` | Query only | `{"topic":"grp...","anchor_seq":12,"items":[{"seq":11,"content":{"text":"上一条"}},{"seq":12,"content":{"text":"命中消息"}}]}` | | `GET /v1/im/topics/{topic}/media?type=image&cursor=` | Query only | `{"topic":"grp...","items":[{"seq":12,"thumbnail_file_id":"file_thumb","file_id":"file_img","width":1080,"height":720,"ts":"2026-08-20T09:00:00Z","availability":"available"}],"next_cursor":""}` | -消息撤回和编辑成功后，Topic 在线成员会收到 0.5.4.6 定义的 `info.what="im"` 业务事件：`message.recalled` 或 `message.edited`。前端收到后按 `topic + seq + version` 刷新本地消息，不要等待用户重新进入会话才更新。 +消息撤回和编辑成功后，Topic 在线成员会收到 0.5.4.6 定义的 `info.what="im"` 业务事件：`message.recalled` 或 `message.edited`。批量撤回按成功项逐条产生 `message.recalled` 事件；失败项不会产生事件。前端收到后按 `topic + seq + version` 刷新本地消息，不要等待用户重新进入会话才更新。 ## 5. 通讯录、好友关系与黑名单 diff --git a/server/im_http.go b/server/im_http.go index 84c3746..3b7e6b4 100644 --- a/server/im_http.go +++ b/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" diff --git a/server/im_http_test.go b/server/im_http_test.go index 8610feb..fdaed49 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -7647,6 +7647,83 @@ func TestServeIMMessageRecall(t *testing.T) { } } +func TestServeIMMessageBatchRecallPartialSuccess(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + actor := types.Uid(42) + expectIMBearer(t, ctrl, "token-1", 7, actor) + oldSubs := store.Subs + oldMessages := store.Messages + subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + messages := mock_store.NewMockMessagesPersistenceInterface(ctrl) + store.Subs = subs + store.Messages = messages + defer func() { + store.Subs = oldSubs + store.Messages = oldMessages + }() + + sub := &types.Subscription{ + TenantID: 7, + User: actor.String(), + Topic: "grpRecall", + ModeWant: types.ModeCFull, + ModeGiven: types.ModeCFull, + } + subs.EXPECT().Get(types.TenantID(7), "grpRecall", actor, false).Return(sub, nil).Times(2) + messages.EXPECT().GetOne(types.TenantID(7), "grpRecall", int64(9)).Return(&types.Message{ + TenantID: 7, + SeqId: 9, + Topic: "grpRecall", + From: actor.String(), + Version: 1, + ObjHeader: types.ObjHeader{CreatedAt: time.Now().Add(-time.Second)}, + }, nil) + messages.EXPECT().Recall(types.TenantID(7), "grpRecall", int64(9), actor, 1, gomock.Any()).Return(&types.Message{ + TenantID: 7, + SeqId: 9, + Topic: "grpRecall", + From: actor.String(), + Version: 2, + RecalledAt: ptrTime(time.Now().UTC()), + RecalledBy: actor.String(), + ObjHeader: types.ObjHeader{UpdatedAt: time.Now().UTC()}, + }, nil) + messages.EXPECT().GetOne(types.TenantID(7), "grpRecall", int64(10)).Return(&types.Message{ + TenantID: 7, + SeqId: 10, + Topic: "grpRecall", + From: actor.String(), + Version: 2, + ObjHeader: types.ObjHeader{CreatedAt: time.Now().Add(-time.Second)}, + }, nil) + messages.EXPECT().Recall(types.TenantID(7), "grpRecall", int64(10), actor, 1, gomock.Any()).Return(nil, types.ErrDuplicate) + + req := httptest.NewRequest(http.MethodPost, "/v1/im/topics/grpRecall/messages/recall", + bytes.NewBufferString(`{"items":[{"seq":9},{"seq":10,"version":1}],"reason":"mistake"}`)) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + + serveIMMessageBatchRecall(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) + items := data["items"].([]any) + if len(items) != 2 { + t.Fatalf("items length: got %d, want 2", len(items)) + } + first := items[0].(map[string]any) + if first["seq"].(float64) != 9 || first["status"] != "recalled" || first["version"].(float64) != 2 { + t.Fatalf("unexpected first item: %#v", first) + } + second := items[1].(map[string]any) + if second["seq"].(float64) != 10 || second["status"] != "failed" || second["reason"] != "VERSION_CONFLICT" { + t.Fatalf("unexpected second item: %#v", second) + } +} + func TestServeIMMessageRecallNormalizesP2PUserTopic(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish()