From 54e58763ff16de24c6afc0186cd75df7907b9831 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Thu, 3 Sep 2026 10:37:42 +0800 Subject: [PATCH] Improve message forward compatibility --- docs/IM-前后端对接文档.md | 15 +-- server/im_http.go | 203 ++++++++++++++++++++++++++++++++++---- server/im_http_test.go | 169 +++++++++++++++++++++++++++++++ 3 files changed, 361 insertions(+), 26 deletions(-) diff --git a/docs/IM-前后端对接文档.md b/docs/IM-前后端对接文档.md index b0cf3e6..7bad68c 100644 --- a/docs/IM-前后端对接文档.md +++ b/docs/IM-前后端对接文档.md @@ -1194,7 +1194,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> **状态：新增。** - 目标列表：`GET /v1/im/forward-targets?query=&cursor=`，只返回 `can_receive_forward=true` 的会话和文件助手。 -- 选中后本地进入确认页，不立即发送；服务端在最终转发接口再次校验目标 Topic、原消息和附件权限。 +- 选中后本地进入确认页，不立即发送；服务端在最终转发接口再次校验目标 Topic、原消息和附件权限。P2P 目标可以传转发目标列表返回的 `topic`，也可以传对方 `usr...` 用户 ID，服务端会归一化为当前用户与对方的 P2P topic。 ### 4.18 转发-多选.png @@ -1203,7 +1203,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> **状态：新增。** - 目标查询同 4.17；多选最多取 `bootstrap.limits.max_forward_targets`，首版建议 9。 -- 确认前只保存目标 Topic 字符串；不得缓存或提交 `tenant_id`。跨租户 Topic 在服务端统一拒绝。 +- 确认前只保存目标 Topic 字符串或对方 `usr...` 用户 ID；不得缓存或提交 `tenant_id`。跨租户 Topic 在服务端统一拒绝。 ### 4.19 转发—逐条转发.png @@ -1211,7 +1211,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> **状态：新增。** -- `POST /v1/im/messages/forward`，Body `{"mode":"separate","sources":[{"topic":"p2p...","seq":"12"}],"target_topics":["grp..."],"comment":"..."}`。 +- `POST /v1/im/messages/forward`，Body `{"mode":"separate","sources":[{"topic":"p2p...","seq":12}],"target_topics":["grp..."],"comment":"..."}`。`seq` 兼容数字和字符串；P2P 源会话也可以传对方 `usr...` 用户 ID，服务端会归一化为 P2P topic。 - 返回每个目标和源消息的 `client_mid/seq/status`；服务端保持源顺序，重新生成目标消息，不能复用原 `seq`。 - 附件必须建立新的授权引用；原消息已撤回、过期或不可见时该项失败。留言作为每个目标的一条独立文本消息发送。 @@ -1221,7 +1221,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> **状态：新增。** -- 调用 4.19 接口，`mode:"bundle"`；服务端生成 `forward_bundle`，内容含 `bundle_id,title,item_count,preview[]`，完整内容通过 `GET /v1/im/forward-bundles/{bundle_id}?cursor=` 获取。 +- 调用 4.19 接口，`mode:"bundle"`；服务端生成一条 `forward_bundle` 目标消息，内容含 `bundle_id,title,item_count,preview[],items[],source_refs[]`。首版 Bundle 明细以内联快照随消息下发，后续如消息量增大再扩展 `GET /v1/im/forward-bundles/{bundle_id}?cursor=` 分页详情。 - 快照保留显示名、时间和内容摘要，对撤回/敏感字段按策略脱敏；不得允许接收端利用 Bundle 获取原会话额外权限。 - 多目标共享不可变 Bundle 内容但分别生成目标消息与授权记录。 @@ -1336,7 +1336,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> **状态：新增。** -- `GET /v1/im/groups/{topic}/members/{user_id}/profile` 返回群内昵称、角色和允许公开的账号资料。 +- `GET /v1/im/groups/{topic}/members/{user_id}/profile` 返回群内昵称、角色和允许公开的账号资料。`{topic}` 必须是群聊 topic（`grp...`）；单聊、名片或普通用户资料页使用 `GET /v1/im/users/{identifier}/profile`，不要把 `usr...` 用户 ID 放到 groups 路径。 - `can_add_friend` 同时受群 `allow_member_add_friend`、目标隐私和租户策略控制；为 false 时隐藏添加入口。 - 从该页申请时 5.6 传 `source:"group"` 和当前 Topic；服务端校验双方仍在群内。 @@ -1733,7 +1733,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `PUT /v1/im/groups/{topic}/members/me/profile` | `{"nickname":"我在群里的昵称","version":1}` | `{"nickname":"我在群里的昵称","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `POST /v1/im/groups/{topic}/leave` | `{}` | `{"topic":"grp...","status":"left"}` | | `GET /v1/im/groups/{topic}/members?query=&cursor=&mentionable=true` | Query only | `{"topic":"grp...","items":[{"user_id":"usr...","display_name":"张三","avatar_file_id":"","role":"member","mentionable":true,"acs":{"want":"JRWP","given":"JRWP","mode":"JRWP"}}],"next_cursor":"","permissions":{"can_mention_all":true}}` | -| `GET /v1/im/groups/{topic}/members/{user_id}/profile` | Path topic/user | `{"topic":"grp...","user_id":"usr...","display_name":"张三","group_nickname":"小张","role":"member","relationship":"friend","capabilities":{"can_send":true,"can_call":true}}` | +| `GET /v1/im/groups/{topic}/members/{user_id}/profile` | Path topic/user；`topic` 必须是 `grp...`，不能传 `usr...` | `{"topic":"grp...","user_id":"usr...","display_name":"张三","group_nickname":"小张","role":"member","relationship":"friend","capabilities":{"can_send":true,"can_call":true}}` | | `POST /v1/im/groups/{topic}/qrcodes` | `{"expires_in_seconds":86400}` | `{"qr_token":"token...","qr_payload":"im://qr?token=...&purpose=join_group","expires_at":"2026-08-21T09:00:00Z","group_snapshot":{"topic":"grp...","name":"项目群","member_count":2}}` | | `POST /v1/im/groups/{topic}/invitations` | `{"user_ids":["usr..."],"message":"进群一起讨论"}` | `{"topic":"grp...","items":[{"user_id":"usr...","status":"joined","invitation_id":"ginv_1"}]}` | | `POST /v1/im/group-invitations/{invitation_id}/accept` 待实现 | `{}` | `{"invitation_id":"ginv_1","status":"joined","topic":"grp..."}` | @@ -1938,7 +1938,8 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `POST /v1/im/favorites/batch-delete` | `{"favorite_ids":["fav_1","fav_2"]}` | `{"items":[{"favorite_id":"fav_1","deleted":true},{"favorite_id":"fav_2","deleted":true}],"failed":[]}` | | `POST /v1/im/favorites/forward` | `{"favorite_ids":["fav_1"],"target_topics":["grp..."],"comment":"转给你"}` | `{"items":[{"favorite_id":"fav_1","target_topic":"grp...","status":"sent","seq":20}],"failed":[]}` | | `GET /v1/im/forward-targets?query=&cursor=` | Query only | `{"items":[{"topic":"grp...","type":"group","display_name":"项目群","avatar_file_id":"","can_receive_forward":true},{"topic":"slf","type":"file_assistant","display_name":"文件助手","can_receive_forward":true}],"next_cursor":"","has_more":false}` | -| `POST /v1/im/messages/forward` 待实现 | `{"mode":"separate","sources":[{"topic":"grp...","seq":"12"}],"target_topics":["p2p..."],"comment":""}` | `{"items":[{"source":{"topic":"grp...","seq":"12"},"target_topic":"p2p...","client_mid":"mid_1","seq":30,"status":"sent"}],"failed":[]}` | +| `POST /v1/im/messages/forward` 逐条转发 | `{"mode":"separate","sources":[{"topic":"grp...","seq":12}],"target_topics":["usr..."],"comment":""}` | `{"items":[{"source":{"topic":"grp...","seq":12},"target_topic":"p2p...","client_mid":"mid_1","seq":30,"status":"sent"}],"failed":[]}` | +| `POST /v1/im/messages/forward` 合并转发 | `{"mode":"bundle","sources":[{"topic":"grp...","seq":12},{"topic":"grp...","seq":13}],"target_topics":["usr..."],"comment":""}` | `{"items":[{"sources":[{"topic":"grp...","seq":12},{"topic":"grp...","seq":13}],"target_topic":"p2p...","client_mid":"mid_1","seq":31,"status":"sent","mode":"bundle","bundle_id":"fbundle_1"}],"failed":[]}` | | `GET /v1/im/forward-bundles/{bundle_id}?cursor=` 待实现 | Query only | `{"bundle_id":"fbundle_1","title":"聊天记录","item_count":2,"items":[{"from_display":"张三","ts":"2026-08-20T09:00:00Z","summary":"文本摘要"}],"next_cursor":""}` | | `POST /v1/im/security/urls/check` | `{"url":"https://example.com","source":"favorite"}` | `{"url":"https://example.com","decision":"allow","risk_level":"low","reason":"ok","source":"favorite"}` | diff --git a/server/im_http.go b/server/im_http.go index 023997e..10d586f 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -425,13 +425,37 @@ type imFavoritesForwardRequest struct { type imMessagesForwardRequest struct { Mode string `json:"mode"` Sources []struct { - Topic string `json:"topic"` - Seq int64 `json:"seq,string"` + Topic string `json:"topic"` + Seq imForwardSeq `json:"seq"` } `json:"sources"` TargetTopics []string `json:"target_topics"` Comment string `json:"comment,omitempty"` } +type imForwardSeq int64 + +func (s *imForwardSeq) UnmarshalJSON(data []byte) error { + var number int64 + if err := json.Unmarshal(data, &number); err == nil { + *s = imForwardSeq(number) + return nil + } + var text string + if err := json.Unmarshal(data, &text); err != nil { + return err + } + text = strings.TrimSpace(text) + if text == "" { + return nil + } + parsed, err := strconv.ParseInt(text, 10, 64) + if err != nil { + return err + } + *s = imForwardSeq(parsed) + return nil +} + type imFriendRequestCreateRequest struct { TargetUserID string `json:"target_user_id"` Message string `json:"message,omitempty"` @@ -3529,8 +3553,8 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { if mode == "" { mode = "separate" } - if mode != "separate" { - writeIMError(wrt, req, http.StatusNotImplemented, 30004, "NOT_IMPLEMENTED", "forward bundle is not implemented", map[string]any{"field": "mode"}) + if mode != "separate" && mode != "bundle" { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "mode"}) return } if len(body.Sources) == 0 || len(body.Sources) > 100 || len(body.TargetTopics) == 0 || len(body.TargetTopics) > 20 { @@ -3542,10 +3566,16 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { items := make([]map[string]any, 0, len(body.TargetTopics)*len(body.Sources)) failed := make([]map[string]any, 0) for _, rawTarget := range body.TargetTopics { - target := strings.TrimSpace(rawTarget) + target, targetOK := normalizeIMForwardTopic(principal.Uid, rawTarget) if target == "" { for _, source := range body.Sources { - failed = append(failed, forwardFailurePayload(source.Topic, source.Seq, target, imReasonMalformedRequest)) + failed = append(failed, forwardFailurePayload(source.Topic, int64(source.Seq), strings.TrimSpace(rawTarget), imReasonMalformedRequest)) + } + continue + } + if !targetOK { + for _, source := range body.Sources { + failed = append(failed, forwardFailurePayload(source.Topic, int64(source.Seq), target, imReasonMalformedRequest)) } continue } @@ -3556,14 +3586,44 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { } if targetErr != nil { for _, source := range body.Sources { - failed = append(failed, forwardFailurePayload(source.Topic, source.Seq, target, relationErrorReason(targetErr))) + failed = append(failed, forwardFailurePayload(source.Topic, int64(source.Seq), target, relationErrorReason(targetErr))) } continue } + if mode == "bundle" { + bundle, sourceFailures := buildForwardBundle(principal, body.Sources, target, sourceReadable, body.Comment) + if len(sourceFailures) > 0 { + failed = append(failed, sourceFailures...) + continue + } + clientMid := "fwd_" + store.Store.GetUidString() + forwarded := &types.Message{ + TenantID: principal.TenantID, + Topic: target, + From: principal.Uid.String(), + Head: forwardBundleMessageHead(clientMid), + Content: bundle, + } + saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true) + if err != nil { + failed = append(failed, forwardFailurePayload("", 0, target, relationErrorReason(err))) + continue + } + items = append(items, map[string]any{ + "sources": bundle["source_refs"], + "target_topic": target, + "client_mid": clientMid, + "seq": saved.SeqId, + "status": "sent", + "mode": "bundle", + "bundle_id": bundle["bundle_id"], + }) + continue + } for _, source := range body.Sources { - sourceTopic := strings.TrimSpace(source.Topic) - if sourceTopic == "" || source.Seq <= 0 { - failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, imReasonMalformedRequest)) + sourceTopic, sourceOK := normalizeIMForwardTopic(principal.Uid, source.Topic) + if sourceTopic == "" || int64(source.Seq) <= 0 || !sourceOK { + failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, imReasonMalformedRequest)) continue } readErr, ok := sourceReadable[sourceTopic] @@ -3572,20 +3632,20 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { sourceReadable[sourceTopic] = readErr } if readErr != nil { - failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, relationErrorReason(readErr))) + failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(readErr))) continue } - msg, err := store.Messages.GetOne(principal.TenantID, sourceTopic, source.Seq) + msg, err := store.Messages.GetOne(principal.TenantID, sourceTopic, int64(source.Seq)) if err != nil { - failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, relationErrorReason(err))) + failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err))) continue } if msg == nil { - failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, "NOT_FOUND")) + failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, "NOT_FOUND")) continue } if messageAvailability(*msg) != "available" { - failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, "MESSAGE_UNAVAILABLE")) + failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, "MESSAGE_UNAVAILABLE")) continue } clientMid := "fwd_" + store.Store.GetUidString() @@ -3598,11 +3658,11 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { } saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true) if err != nil { - failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, relationErrorReason(err))) + failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err))) continue } items = append(items, map[string]any{ - "source": map[string]any{"topic": sourceTopic, "seq": source.Seq}, + "source": map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)}, "target_topic": target, "client_mid": clientMid, "seq": saved.SeqId, @@ -5455,8 +5515,16 @@ func serveIMGroupMemberProfile(wrt http.ResponseWriter, req *http.Request, topic if !ok { return } - if types.GetTopicCat(topic) != types.TopicCatGrp || memberID.IsZero() { - writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) + if types.GetTopicCat(topic) != types.TopicCatGrp { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{ + "field": "topic", + "expected": "group topic, e.g. grp...", + "actual": topic, + }) + return + } + if memberID.IsZero() { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "user_id"}) return } selfSub, err := store.Subs.Get(principal.TenantID, topic, principal.Uid, false) @@ -8791,6 +8859,94 @@ func forwardFailurePayload(topic string, seq int64, target, reason string) map[s } } +func normalizeIMForwardTopic(viewer types.Uid, topic string) (string, bool) { + topic = strings.TrimSpace(topic) + if topic == "" { + return "", false + } + if len(topic) < 3 { + return "", false + } + switch types.GetTopicCat(topic) { + case types.TopicCatMe: + peer := types.ParseUserId(topic) + if peer.IsZero() { + return "", false + } + p2p := viewer.P2PName(peer) + return p2p, p2p != "" + case types.TopicCatP2P, types.TopicCatGrp, types.TopicCatSlf: + return topic, true + default: + return "", false + } +} + +func buildForwardBundle(principal *imPrincipal, sources []struct { + Topic string `json:"topic"` + Seq imForwardSeq `json:"seq"` +}, target string, sourceReadable map[string]error, comment string) (map[string]any, []map[string]any) { + items := make([]map[string]any, 0, len(sources)) + preview := make([]map[string]any, 0, min(len(sources), 3)) + sourceRefs := make([]map[string]any, 0, len(sources)) + failures := make([]map[string]any, 0) + for _, source := range sources { + sourceTopic, sourceOK := normalizeIMForwardTopic(principal.Uid, source.Topic) + if sourceTopic == "" || int64(source.Seq) <= 0 || !sourceOK { + failures = append(failures, forwardFailurePayload(sourceTopic, int64(source.Seq), target, imReasonMalformedRequest)) + continue + } + readErr, ok := sourceReadable[sourceTopic] + if !ok { + readErr = requireTopicReader(principal, sourceTopic) + sourceReadable[sourceTopic] = readErr + } + if readErr != nil { + failures = append(failures, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(readErr))) + continue + } + msg, err := store.Messages.GetOne(principal.TenantID, sourceTopic, int64(source.Seq)) + if err != nil { + failures = append(failures, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err))) + continue + } + if msg == nil { + failures = append(failures, forwardFailurePayload(sourceTopic, int64(source.Seq), target, "NOT_FOUND")) + continue + } + if messageAvailability(*msg) != "available" { + failures = append(failures, forwardFailurePayload(sourceTopic, int64(source.Seq), target, "MESSAGE_UNAVAILABLE")) + continue + } + snapshot := favoriteSnapshot(msg) + items = append(items, snapshot) + ref := map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)} + sourceRefs = append(sourceRefs, ref) + if len(preview) < 3 { + preview = append(preview, map[string]any{ + "from": msg.From, + "type": imMessageType(msg), + "summary": textSnippet(messageText(*msg), ""), + "ts": msg.CreatedAt.UTC().Format(time.RFC3339Nano), + }) + } + } + if len(failures) > 0 { + return nil, failures + } + bundleID := "fbundle_" + store.Store.GetUidString() + return map[string]any{ + "bundle_id": bundleID, + "title": "聊天记录", + "item_count": len(items), + "preview": preview, + "items": items, + "source_refs": sourceRefs, + "comment": strings.TrimSpace(comment), + "forwarded": true, + }, nil +} + func forwardedMessageHead(msg *types.Message, clientMid string) types.KVMap { head := types.KVMap{} if msg != nil { @@ -8811,6 +8967,15 @@ func forwardedMessageHead(msg *types.Message, clientMid string) types.KVMap { return head } +func forwardBundleMessageHead(clientMid string) types.KVMap { + return types.KVMap{ + "type": "forward_bundle", + "x-im-type": "forward_bundle", + "x-im-forwarded": true, + "x-im-client-mid": clientMid, + } +} + func forwardedMessageContent(msg *types.Message, comment string) any { if strings.TrimSpace(comment) == "" { if msg == nil { diff --git a/server/im_http_test.go b/server/im_http_test.go index e6ee1f8..f4eafc4 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -3600,6 +3600,30 @@ func TestServeIMGroupMemberProfile(t *testing.T) { } } +func TestServeIMGroupMemberProfileRejectsUserTopic(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + expectIMBearer(t, ctrl, "token-1", 7, 42) + memberID := types.Uid(42).UserId() + req := httptest.NewRequest(http.MethodGet, "/v1/im/groups/"+memberID+"/members/"+memberID+"/profile", nil) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + + serveIMGroupResource(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusBadRequest, rec.Body.String()) + } + body := decodeIMHTTPResponse(t, rec) + if body.Reason != imReasonMalformedRequest { + t.Fatalf("reason: got %q, want %q", body.Reason, imReasonMalformedRequest) + } + details := body.Details.(map[string]any) + if details["field"] != "topic" || details["actual"] != memberID { + t.Fatalf("unexpected details: %#v", details) + } +} + func TestServeIMGroupList(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -4551,6 +4575,151 @@ func TestServeIMMessageForwardSeparateSavesMessages(t *testing.T) { } } +func TestServeIMMessageForwardAcceptsNumericSeqAndUserTarget(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockStore := expectIMBearer(t, ctrl, "token-1", 7, 42) + mockStore.EXPECT().GetUidString().Return("mid-2") + + oldSubs := store.Subs + oldMessages := store.Messages + subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + messages := mock_store.NewMockMessagesPersistenceInterface(ctrl) + store.Subs = subs + store.Messages = messages + t.Cleanup(func() { + store.Subs = oldSubs + store.Messages = oldMessages + }) + + targetUser := types.Uid(99) + targetTopic := types.Uid(42).P2PName(targetUser) + subs.EXPECT().Get(types.TenantID(7), targetTopic, types.Uid(42), false).Return(&types.Subscription{ + TenantID: 7, + User: types.Uid(42).String(), + Topic: targetTopic, + ModeWant: types.ModeCPublic, + ModeGiven: types.ModeCPublic, + }, nil) + subs.EXPECT().Get(types.TenantID(7), "grpSource", types.Uid(42), false).Return(&types.Subscription{ + TenantID: 7, + User: types.Uid(42).String(), + Topic: "grpSource", + ModeWant: types.ModeCPublic, + ModeGiven: types.ModeCPublic, + }, nil) + messages.EXPECT().GetOne(types.TenantID(7), "grpSource", int64(60)).Return(&types.Message{ + TenantID: 7, + Topic: "grpSource", + SeqId: 60, + From: targetUser.String(), + Head: types.KVMap{"x-im-type": "text"}, + Content: map[string]any{"text": "hello"}, + ObjHeader: types.ObjHeader{ + CreatedAt: time.Date(2026, 8, 13, 9, 0, 0, 0, time.UTC), + }, + }, nil) + messages.EXPECT().Save(types.TenantID(7), gomock.Any(), gomock.Nil(), true).DoAndReturn( + func(_ types.TenantID, msg *types.Message, _ []string, _ bool) (*types.MessageSaveResult, error) { + if msg.Topic != targetTopic { + t.Fatalf("target topic: got %q, want %q", msg.Topic, targetTopic) + } + return &types.MessageSaveResult{SeqId: 22, MarkedReadBySender: true}, nil + }) + + req := httptest.NewRequest(http.MethodPost, "/v1/im/messages/forward", bytes.NewBufferString(`{"mode":"separate","sources":[{"topic":"grpSource","seq":60}],"target_topics":["`+targetUser.UserId()+`"]}`)) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + serveIMMessageForward(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("forward status: got %d body=%s", rec.Code, rec.Body.String()) + } + data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) + items := data["items"].([]any) + if len(items) != 1 { + t.Fatalf("unexpected forward response: %#v", data) + } + got := items[0].(map[string]any) + if got["target_topic"] != targetTopic || got["seq"].(float64) != 22 { + t.Fatalf("unexpected forward item: %#v", got) + } +} + +func TestServeIMMessageForwardBundleSavesForwardBundle(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockStore := expectIMBearer(t, ctrl, "token-1", 7, 42) + gomock.InOrder( + mockStore.EXPECT().GetUidString().Return("bundle-1"), + mockStore.EXPECT().GetUidString().Return("mid-3"), + ) + + oldSubs := store.Subs + oldMessages := store.Messages + subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + messages := mock_store.NewMockMessagesPersistenceInterface(ctrl) + store.Subs = subs + store.Messages = messages + t.Cleanup(func() { + store.Subs = oldSubs + store.Messages = oldMessages + }) + + subs.EXPECT().Get(types.TenantID(7), "grpTarget", types.Uid(42), false).Return(&types.Subscription{ + TenantID: 7, + User: types.Uid(42).String(), + Topic: "grpTarget", + ModeWant: types.ModeCPublic, + ModeGiven: types.ModeCPublic, + }, nil) + subs.EXPECT().Get(types.TenantID(7), "grpSource", types.Uid(42), false).Return(&types.Subscription{ + TenantID: 7, + User: types.Uid(42).String(), + Topic: "grpSource", + ModeWant: types.ModeCPublic, + ModeGiven: types.ModeCPublic, + }, nil) + messages.EXPECT().GetOne(types.TenantID(7), "grpSource", int64(60)).Return(&types.Message{ + TenantID: 7, + Topic: "grpSource", + SeqId: 60, + From: types.Uid(99).String(), + Head: types.KVMap{"x-im-type": "text"}, + Content: map[string]any{"text": "hello"}, + ObjHeader: types.ObjHeader{ + CreatedAt: time.Date(2026, 8, 13, 9, 0, 0, 0, time.UTC), + }, + }, nil) + messages.EXPECT().Save(types.TenantID(7), gomock.Any(), gomock.Nil(), true).DoAndReturn( + func(_ types.TenantID, msg *types.Message, _ []string, _ bool) (*types.MessageSaveResult, error) { + if msg.Topic != "grpTarget" || msg.Head["x-im-type"] != "forward_bundle" { + t.Fatalf("unexpected bundle message: %#v", msg) + } + content := msg.Content.(map[string]any) + if content["bundle_id"] != "fbundle_bundle-1" || content["item_count"] != 1 { + t.Fatalf("unexpected bundle content: %#v", content) + } + return &types.MessageSaveResult{SeqId: 23, MarkedReadBySender: true}, nil + }) + + req := httptest.NewRequest(http.MethodPost, "/v1/im/messages/forward", bytes.NewBufferString(`{"mode":"bundle","sources":[{"topic":"grpSource","seq":60}],"target_topics":["grpTarget"]}`)) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + serveIMMessageForward(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("forward status: got %d body=%s", rec.Code, rec.Body.String()) + } + data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) + items := data["items"].([]any) + if len(items) != 1 { + t.Fatalf("unexpected forward response: %#v", data) + } + got := items[0].(map[string]any) + if got["mode"] != "bundle" || got["bundle_id"] != "fbundle_bundle-1" || got["seq"].(float64) != 23 { + t.Fatalf("unexpected forward item: %#v", got) + } +} + func TestServeIMSecurityURLCheck(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish()