From 926b4bbd81c82ca047a53ab14261e3c3114cb2ee Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Fri, 11 Sep 2026 09:00:38 +0800 Subject: [PATCH] fix im forwarding and message list regressions --- docs/IM-前后端对接文档.md | 8 +- .../db/mysql/report_group_regression_test.go | 54 +++++++ server/db/mysql/tenant_adapter.go | 12 +- server/im_http.go | 104 +++++++++++-- server/im_http_test.go | 137 +++++++++++++++++- server/kafka_message_persist.go | 12 +- server/store/group_message_runtime.go | 17 +++ server/store/group_message_runtime_test.go | 6 + server/store/types/types.go | 1 + 9 files changed, 331 insertions(+), 20 deletions(-) diff --git a/docs/IM-前后端对接文档.md b/docs/IM-前后端对接文档.md index ae07a34..b1d60b9 100644 --- a/docs/IM-前后端对接文档.md +++ b/docs/IM-前后端对接文档.md @@ -1358,7 +1358,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> **状态：新增。** - `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`。 +- 返回每个目标和源消息的 `client_mid/seq/status`；`status` 取值为 `sent`、`pending`、`failed`。`pending` 表示服务端已分配 `seq/client_mid`，但 Kafka 投递结果暂不确定，客户端应按“发送中”展示，保留同一个 `Idempotency-Key` 重试或等待后续消息同步，不要立即生成第二条本地消息。服务端保持源顺序，重新生成目标消息，不能复用原 `seq`。 - 附件必须建立新的授权引用；原消息已撤回、过期或不可见时该项失败。留言作为每个目标的一条独立文本消息发送。 ### 4.20 转发—合并转发—多选状态.png @@ -2084,10 +2084,10 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `GET /v1/im/favorites/{favorite_id}` | Path favorite ID | `{"favorite_id":"fav_1","type":"file","snapshot":{"file_id":"file_1","name":"a.pdf"},"source":{"topic":"grp...","seq":12},"availability":"available"}` | | `DELETE /v1/im/favorites/{favorite_id}` | Path favorite ID | `{"favorite_id":"fav_1","deleted":true}` | | `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":[]}` | +| `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":[]}`；Kafka 投递结果不确定时返回 `status:"pending"`、`reason:"MESSAGE_PERSIST_PENDING"` | | `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":["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":[]}` | +| `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":[]}`；Kafka 投递结果不确定时返回 `status:"pending"`、`reason:"MESSAGE_PERSIST_PENDING"` | +| `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":[]}`；Kafka 投递结果不确定时返回 `status:"pending"`、`reason:"MESSAGE_PERSIST_PENDING"` | | `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/db/mysql/report_group_regression_test.go b/server/db/mysql/report_group_regression_test.go index 6d1b22b..9002107 100644 --- a/server/db/mysql/report_group_regression_test.go +++ b/server/db/mysql/report_group_regression_test.go @@ -55,6 +55,47 @@ func TestMessageSearchDoesNotClearScannedHead(t *testing.T) { } } +func TestSearchPaginationCursorUsesLastReturnedRow(t *testing.T) { + source, err := os.ReadFile("tenant_adapter.go") + if err != nil { + t.Fatal(err) + } + body := string(source) + for _, tc := range []struct { + name string + function string + forbidden string + required string + }{ + { + name: "p2p message search", + function: "func (a *adapter) TenantMessageSearch", + forbidden: "nextCursor = strconv.FormatInt(msg.SeqId, 10)", + required: "nextCursor = strconv.FormatInt(lastSeq, 10)", + }, + { + name: "group message search", + function: "func (a *adapter) tenantGroupMessageSearch", + forbidden: "nextCursor = strconv.FormatInt(msg.SeqId, 10)", + required: "nextCursor = strconv.FormatInt(lastSeq, 10)", + }, + { + name: "favorite list", + function: "func (a *adapter) TenantFavoriteList", + forbidden: "nextCursor = strconv.FormatInt(item.ID, 10)", + required: "nextCursor = strconv.FormatInt(lastID, 10)", + }, + } { + fn := functionSource(t, body, tc.function) + if strings.Contains(fn, tc.forbidden) { + t.Fatalf("%s must not use the extra fetched row as next cursor", tc.name) + } + if !strings.Contains(fn, tc.required) { + t.Fatalf("%s must use the last returned row as next cursor", tc.name) + } + } +} + func TestTenantFileAccessIncludesAvatarAudienceRules(t *testing.T) { source, err := os.ReadFile("tenant_adapter.go") if err != nil { @@ -72,3 +113,16 @@ func TestTenantFileAccessIncludesAvatarAudienceRules(t *testing.T) { } } } + +func functionSource(t *testing.T, source, marker string) string { + t.Helper() + start := strings.Index(source, marker) + if start < 0 { + t.Fatalf("function %q not found", marker) + } + next := strings.Index(source[start+len(marker):], "\nfunc ") + if next < 0 { + return source[start:] + } + return source[start : start+len(marker)+next] +} diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index ae9de50..48a57a1 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -2486,6 +2486,7 @@ func (a *adapter) TenantMessageSearch(id t.TenantID, topic string, uid t.Uid, qu defer rows.Close() msgs := make([]t.Message, 0, limit) nextCursor := "" + lastSeq := int64(0) for rows.Next() { row := struct { t.Message @@ -2496,10 +2497,11 @@ func (a *adapter) TenantMessageSearch(id t.TenantID, topic string, uid t.Uid, qu } msg := finalizeTenantMessage(id, row.Message, row.RecalledBy) if len(msgs) == limit { - nextCursor = strconv.FormatInt(msg.SeqId, 10) + nextCursor = strconv.FormatInt(lastSeq, 10) continue } msgs = append(msgs, msg) + lastSeq = msg.SeqId } if err = rows.Err(); err != nil { return nil, "", err @@ -2546,6 +2548,7 @@ func (a *adapter) tenantGroupMessageSearch(id t.TenantID, topic string, uid t.Ui defer rows.Close() msgs := make([]t.Message, 0, limit) nextCursor := "" + lastSeq := int64(0) for rows.Next() { row := struct { t.Message @@ -2556,10 +2559,11 @@ func (a *adapter) tenantGroupMessageSearch(id t.TenantID, topic string, uid t.Ui } msg := normalizeGroupMessage(id, row.Message, row.RecalledBy) if len(msgs) == limit { - nextCursor = strconv.FormatInt(msg.SeqId, 10) + nextCursor = strconv.FormatInt(lastSeq, 10) continue } msgs = append(msgs, msg) + lastSeq = msg.SeqId } if err = rows.Err(); err != nil { return nil, "", err @@ -7230,6 +7234,7 @@ func (a *adapter) TenantFavoriteList(id t.TenantID, uid t.Uid, itemType, queryTe defer rows.Close() items := make([]t.FavoriteItem, 0, limit) nextCursor := "" + lastID := int64(0) for rows.Next() { item, err := scanFavorite(rows) if err != nil { @@ -7237,10 +7242,11 @@ func (a *adapter) TenantFavoriteList(id t.TenantID, uid t.Uid, itemType, queryTe } item.TenantID = id if len(items) == limit { - nextCursor = strconv.FormatInt(item.ID, 10) + nextCursor = strconv.FormatInt(lastID, 10) continue } items = append(items, item) + lastID = item.ID } if err = rows.Err(); err != nil { return nil, "", err diff --git a/server/im_http.go b/server/im_http.go index eb37fbb..07dcd08 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -3745,9 +3745,17 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) { } 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"}) + 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"}) + continue + } results = append(results, map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "sent", "seq": saved.SeqId}) } } @@ -3814,7 +3822,7 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { failed = append(failed, sourceFailures...) continue } - clientMid := "fwd_" + store.Store.GetUidString() + clientMid := forwardClientMid(req.Header.Get("Idempotency-Key"), "bundle", target, len(body.Sources), body.Comment) forwarded := &types.Message{ TenantID: principal.TenantID, Topic: target, @@ -3828,9 +3836,36 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { } 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{ + "sources": bundle["source_refs"], + "target_topic": target, + "client_mid": clientMid, + "seq": saved.SeqId, + "status": "pending", + "reason": "MESSAGE_PERSIST_PENDING", + "mode": "bundle", + "bundle_id": bundle["bundle_id"], + }) + continue + } failed = append(failed, forwardFailurePayload("", 0, target, relationErrorReason(err))) continue } + if saved != nil && saved.Pending { + items = append(items, map[string]any{ + "sources": bundle["source_refs"], + "target_topic": target, + "client_mid": clientMid, + "seq": saved.SeqId, + "status": "pending", + "reason": "MESSAGE_PERSIST_PENDING", + "mode": "bundle", + "bundle_id": bundle["bundle_id"], + }) + continue + } forwarded.SeqId = saved.SeqId broadcastHTTPForwardMessage(principal.TenantID, forwarded) items = append(items, map[string]any{ @@ -3872,7 +3907,7 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, "MESSAGE_UNAVAILABLE")) continue } - clientMid := "fwd_" + store.Store.GetUidString() + clientMid := forwardClientMid(req.Header.Get("Idempotency-Key"), "separate", target, sourceTopic, int64(source.Seq)) forwarded := &types.Message{ TenantID: principal.TenantID, Topic: target, @@ -3886,9 +3921,32 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { } 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{ + "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", + }) + continue + } failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err))) continue } + if saved != nil && saved.Pending { + items = append(items, 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", + }) + continue + } forwarded.SeqId = saved.SeqId broadcastHTTPForwardMessage(principal.TenantID, forwarded) items = append(items, map[string]any{ @@ -8042,7 +8100,7 @@ func messageSearchPayload(msg types.Message, query string) map[string]any { snippet := textSnippet(text, query) return map[string]any{ "seq": msg.SeqId, - "from": msg.From, + "from": messageFromUserID(msg.From), "ts": msg.CreatedAt.UTC().Format(time.RFC3339Nano), "type": imMessageType(&msg), "version": msg.Version, @@ -8056,7 +8114,7 @@ func messageSearchPayload(msg types.Message, query string) map[string]any { func messageContextPayload(msg types.Message) map[string]any { return map[string]any{ "seq": msg.SeqId, - "from": msg.From, + "from": messageFromUserID(msg.From), "ts": msg.CreatedAt.UTC().Format(time.RFC3339Nano), "type": imMessageType(&msg), "version": msg.Version, @@ -8070,7 +8128,7 @@ func mediaMessagePayload(msg types.Message) map[string]any { snapshot := favoriteSnapshot(&msg) data := map[string]any{ "seq": msg.SeqId, - "from": msg.From, + "from": messageFromUserID(msg.From), "ts": msg.CreatedAt.UTC().Format(time.RFC3339Nano), "availability": messageAvailability(msg), } @@ -8082,6 +8140,23 @@ func mediaMessagePayload(msg types.Message) map[string]any { return data } +func messageFromUserID(from string) string { + from = strings.TrimSpace(from) + if from == "" { + return "" + } + if strings.HasPrefix(from, "usr") { + if uid := types.ParseUserId(from); !uid.IsZero() { + return uid.UserId() + } + return from + } + if uid := types.ParseUid(from); !uid.IsZero() { + return uid.UserId() + } + return from +} + func callRecordPayloads(records []types.CallRecord, viewer types.Uid) []map[string]any { items := make([]map[string]any, 0, len(records)) for _, record := range records { @@ -9379,7 +9454,7 @@ func userProfilePayload(user *types.User, ctx *types.RelationshipContext) map[st } func copyOptionalProfileFields(data map[string]any, public map[string]any) { - for _, key := range []string{"gender", "region", "public_account", "signature"} { + for _, key := range []string{"nickname", "gender", "region", "public_account", "signature"} { if value, ok := public[key]; ok { data[key] = value } @@ -9436,7 +9511,7 @@ func displayInitial(value any) string { func writeIMRelationError(wrt http.ResponseWriter, req *http.Request, err error) { switch { case errors.Is(err, types.ErrNotFound): - writeIMError(wrt, req, http.StatusNotFound, 20004, imReasonMalformedRequest, "resource not found", nil) + writeIMError(wrt, req, http.StatusNotFound, 20004, imReasonResourceNotFound, "resource not found", nil) case errors.Is(err, types.ErrPermissionDenied): writeIMError(wrt, req, http.StatusForbidden, 20005, imReasonPermissionDenied, "permission denied", nil) case errors.Is(err, types.ErrInvalidResponse): @@ -9625,7 +9700,7 @@ func favoriteSnapshot(msg *types.Message) map[string]any { return map[string]any{ "topic": msg.Topic, "seq": msg.SeqId, - "from": msg.From, + "from": messageFromUserID(msg.From), "ts": msg.CreatedAt.UTC().Format(time.RFC3339Nano), "type": imMessageType(msg), "version": msg.Version, @@ -9677,6 +9752,17 @@ func forwardFailurePayload(topic string, seq int64, target, reason string) map[s } } +func forwardClientMid(requestKey string, parts ...any) string { + requestKey = strings.TrimSpace(requestKey) + if requestKey == "" { + return "fwd_" + store.Store.GetUidString() + } + args := make([]any, 0, len(parts)+2) + args = append(args, "forward", requestKey) + args = append(args, parts...) + return "fwd_" + stableIMEventID(args...) +} + func normalizeIMForwardTopic(viewer types.Uid, topic string) (string, bool) { return normalizeIMUserVisibleTopic(viewer, topic) } @@ -10624,7 +10710,7 @@ func normalizeIMAbuseTarget(tenantID types.TenantID, reporter types.Uid, target } snapshot["topic"] = topic snapshot["seq"] = seq - snapshot["from"] = msg.From + snapshot["from"] = messageFromUserID(msg.From) snapshot["ts"] = msg.CreatedAt.UTC().Format(time.RFC3339Nano) return targetType, 0, topic, seq, snapshot, nil default: diff --git a/server/im_http_test.go b/server/im_http_test.go index 9913a69..d631816 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -1625,6 +1625,21 @@ func TestServeIMTenantResolveRequiresAPIKey(t *testing.T) { } } +func TestWriteIMRelationErrorUsesResourceNotFoundReason(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/v1/im/test", nil) + rec := httptest.NewRecorder() + + writeIMRelationError(rec, req, types.ErrNotFound) + + if rec.Code != http.StatusNotFound { + t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusNotFound, rec.Body.String()) + } + resp := decodeIMHTTPResponse(t, rec) + if resp.Reason != imReasonResourceNotFound { + t.Fatalf("reason: got %q, want %q", resp.Reason, imReasonResourceNotFound) + } +} + func TestServeIMTenantResolveRejectsMalformedTenantCode(t *testing.T) { req := httptest.NewRequest(http.MethodPost, "/v1/im/tenants/resolve", bytes.NewBufferString(`{"tenant_code":"`+strings.Repeat("a", 65)+`"}`)) @@ -5288,6 +5303,10 @@ func TestServeIMFavoriteBatchCreateNormalizesP2PUserTopic(t *testing.T) { if favorites.created == nil || favorites.created.SourceTopic != normalizedTopic { t.Fatalf("favorite source topic was not normalized: %#v", favorites.created) } + snapshot := favorites.created.Snapshot.(map[string]any) + if snapshot["from"] != peer.UserId() { + t.Fatalf("favorite snapshot from must use full user_id: %#v", favorites.created.Snapshot) + } } func TestServeIMFavoriteForwardSavesMessages(t *testing.T) { @@ -5679,6 +5698,100 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) { } } +func TestServeIMMessageForwardKafkaPublishUnknownReturnsPending(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + expectIMBearer(t, ctrl, "token-1", 7, 42) + + oldKafkaBus := kafkaBus + oldPublisher := publishKafkaMessagePersist + oldHub := globals.hub + oldSubs := store.Subs + oldTopics := store.Topics + oldMessages := store.Messages + subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + topics := mock_store.NewMockTopicsPersistenceInterface(ctrl) + messages := mock_store.NewMockMessagesPersistenceInterface(ctrl) + kafkaBus = &kafkaRuntime{} + globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 3)} + store.Subs = subs + store.Topics = topics + store.Messages = messages + t.Cleanup(func() { + kafkaBus = oldKafkaBus + publishKafkaMessagePersist = oldPublisher + globals.hub = oldHub + store.Subs = oldSubs + store.Topics = oldTopics + 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) + topics.EXPECT().Get(types.TenantID(7), targetTopic).Return(&types.Topic{ + TenantID: 7, + SeqId: 40, + }, nil) + messages.EXPECT().GetOne(types.TenantID(7), "grpSource", int64(14)).Return(&types.Message{ + TenantID: 7, + Topic: "grpSource", + SeqId: 14, + 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) + + publishKafkaMessagePersist = func(_ context.Context, _, _ string, _ kafkaEnvelope) error { + return context.DeadlineExceeded + } + + req := httptest.NewRequest(http.MethodPost, "/v1/im/messages/forward", bytes.NewBufferString(`{"mode":"separate","sources":[{"topic":"grpSource","seq":14}],"target_topics":["`+targetUser.UserId()+`"]}`)) + req.Header.Set("Authorization", "Bearer token-1") + req.Header.Set("Idempotency-Key", "idem-forward-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["status"] != "pending" || got["reason"] != "MESSAGE_PERSIST_PENDING" || got["seq"].(float64) != 41 { + t.Fatalf("unexpected pending item: %#v", got) + } + if got["client_mid"] != "fwd_"+stableIMEventID("forward", "idem-forward-1", "separate", targetTopic, "grpSource", int64(14)) { + t.Fatalf("unexpected stable client mid: %#v", got) + } + if failed := data["failed"].([]any); len(failed) != 0 { + t.Fatalf("unexpected failed items: %#v", failed) + } + select { + case delivered := <-globals.hub.routeSrv: + t.Fatalf("pending forward should not broadcast immediately: %#v", delivered) + default: + } +} + func TestServeIMMessageForwardBundleSavesForwardBundle(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -6024,6 +6137,15 @@ func TestServeIMMessageSearchContextAndMedia(t *testing.T) { if searchRec.Code != http.StatusOK { t.Fatalf("search status: got %d body=%s", searchRec.Code, searchRec.Body.String()) } + searchData := decodeIMHTTPResponse(t, searchRec).Data.(map[string]any) + searchItems := searchData["items"].([]any) + if len(searchItems) != 1 || searchItems[0].(map[string]any)["from"] != types.Uid(42).UserId() { + t.Fatalf("search from must use full user_id: %#v", searchData) + } + searchSnapshot := searchItems[0].(map[string]any)["snapshot"].(map[string]any) + if searchSnapshot["from"] != types.Uid(42).UserId() { + t.Fatalf("search snapshot from must use full user_id: %#v", searchSnapshot) + } contextReq := httptest.NewRequest(http.MethodGet, "/v1/im/topics/grpAlpha/messages/3/context", nil) contextReq.Header.Set("Authorization", "Bearer token-1") @@ -6032,6 +6154,11 @@ func TestServeIMMessageSearchContextAndMedia(t *testing.T) { if contextRec.Code != http.StatusOK { t.Fatalf("context status: got %d body=%s", contextRec.Code, contextRec.Body.String()) } + contextData := decodeIMHTTPResponse(t, contextRec).Data.(map[string]any) + contextItems := contextData["items"].([]any) + if len(contextItems) != 1 || contextItems[0].(map[string]any)["from"] != types.Uid(42).UserId() { + t.Fatalf("context from must use full user_id: %#v", contextData) + } mediaReq := httptest.NewRequest(http.MethodGet, "/v1/im/topics/grpAlpha/media?type=image", nil) mediaReq.Header.Set("Authorization", "Bearer token-1") @@ -6040,6 +6167,11 @@ func TestServeIMMessageSearchContextAndMedia(t *testing.T) { if mediaRec.Code != http.StatusOK { t.Fatalf("media status: got %d body=%s", mediaRec.Code, mediaRec.Body.String()) } + mediaData := decodeIMHTTPResponse(t, mediaRec).Data.(map[string]any) + mediaItems := mediaData["items"].([]any) + if len(mediaItems) != 1 || mediaItems[0].(map[string]any)["from"] != types.Uid(42).UserId() { + t.Fatalf("media from must use full user_id: %#v", mediaData) + } } func TestServeIMTopicReadActionsNormalizeP2PUserTopic(t *testing.T) { @@ -7318,7 +7450,7 @@ func TestServeIMUserProfile(t *testing.T) { users.EXPECT().Get(types.TenantID(7), peer).Return(&types.User{ ObjHeader: types.ObjHeader{Id: peer.String()}, TenantID: 7, - Public: map[string]any{"display_name": "Alice", "avatar_file_id": "file-1", "signature": "hello"}, + Public: map[string]any{"display_name": "Alice", "nickname": "Alice", "avatar_file_id": "file-1", "signature": "hello"}, }, nil) store.Users = users store.Relations = &testRelationsStore{context: &types.RelationshipContext{ @@ -7353,6 +7485,9 @@ func TestServeIMUserProfile(t *testing.T) { if data["display_name"] != "老同事" || data["remark"] != "老同事" { t.Fatalf("profile display name should use friend remark: %#v", data) } + if data["nickname"] != "Alice" { + t.Fatalf("profile should include original nickname: %#v", data) + } } func TestServeIMUserQRCodes(t *testing.T) { diff --git a/server/kafka_message_persist.go b/server/kafka_message_persist.go index ad60aff..4cf157e 100644 --- a/server/kafka_message_persist.go +++ b/server/kafka_message_persist.go @@ -48,7 +48,10 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty if runtimeReady && !claimed { if claim != nil && claim.Seq > 0 { msg.SeqId = claim.Seq - return &types.MessageSaveResult{SeqId: claim.Seq, Duplicate: true}, nil + return &types.MessageSaveResult{SeqId: claim.Seq, Duplicate: claim.Status == "done", Pending: claim.Status == "pending"}, nil + } + if claim != nil && claim.Status == "pending" { + return &types.MessageSaveResult{Pending: true}, types.ErrDuplicate } return nil, types.ErrDuplicate } @@ -64,9 +67,12 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty msg.InitTimes() if err = enqueueKafkaMessagePersist(ctx, tenantID, msg, attachmentURLs, readBySender); err != nil { if clientMsgID != "" { - _ = store.MarkMessageUniqueFailed(tenantID, msg.Topic, fromUid, clientMsgID) + if markErr := store.MarkMessageUniquePending(tenantID, msg.Topic, fromUid, clientMsgID, msg.SeqId); markErr != nil { + logs.Warn.Printf("kafka message persist: failed to mark message unique key pending topic=%s seq=%d: %v", msg.Topic, msg.SeqId, markErr) + } } - return nil, err + logs.Warn.Printf("kafka message persist: publish state unknown topic=%s seq=%d client_mid=%s err=%v", msg.Topic, msg.SeqId, clientMsgID, err) + return &types.MessageSaveResult{SeqId: msg.SeqId, Pending: true, MarkedReadBySender: readBySender && !fromUid.IsZero()}, err } if clientMsgID != "" { if err = store.MarkMessageUniqueDone(tenantID, msg.Topic, fromUid, clientMsgID, msg.SeqId, ""); err != nil { diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index a2fdf0d..0d4c017 100644 --- a/server/store/group_message_runtime.go +++ b/server/store/group_message_runtime.go @@ -55,6 +55,7 @@ type groupMessageRuntime interface { Close() error Claim(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) (*groupMessageUniqueRecord, bool, error) NextSeq(tenantID types.TenantID, topic string, baseline int64) (int64, error) + MarkPending(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error MarkDone(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error MarkFailed(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) error CacheHotMessage(tenantID types.TenantID, msg *types.Message) error @@ -111,6 +112,13 @@ func MarkMessageUniqueDone(tenantID types.TenantID, topic string, from types.Uid return groupMsgRuntime.MarkDone(tenantID, topic, from, clientMsgID, seq, msgID) } +func MarkMessageUniquePending(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error { + if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || from.IsZero() || clientMsgID == "" || seq <= 0 { + return nil + } + return groupMsgRuntime.MarkPending(tenantID, topic, from, clientMsgID, seq) +} + func MarkMessageUniqueFailed(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) error { if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || from.IsZero() || clientMsgID == "" { return nil @@ -464,6 +472,15 @@ func (r *redisGroupMessageRuntime) MarkDone(tenantID types.TenantID, topic strin return r.client.Set(context.Background(), r.uniqueKey(tenantID, topic, from, clientMsgID), raw, time.Duration(r.cfg.DoneTTL)*time.Second).Err() } +func (r *redisGroupMessageRuntime) MarkPending(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error { + if clientMsgID == "" || seq <= 0 { + return nil + } + rec := &groupMessageUniqueRecord{Status: "pending", Seq: seq, CreatedAt: time.Now().UTC()} + raw, _ := json.Marshal(rec) + return r.client.Set(context.Background(), r.uniqueKey(tenantID, topic, from, clientMsgID), raw, time.Duration(r.cfg.PendingTTL)*time.Second).Err() +} + func (r *redisGroupMessageRuntime) MarkFailed(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) error { if clientMsgID == "" { return nil diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index c32c297..dfa1df7 100644 --- a/server/store/group_message_runtime_test.go +++ b/server/store/group_message_runtime_test.go @@ -49,6 +49,12 @@ func (r *testGroupMessageRuntime) MarkDone(_ types.TenantID, topic string, from return nil } +func (r *testGroupMessageRuntime) MarkPending(_ types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error { + key := topic + ":" + from.UserId() + ":" + clientMsgID + r.claims[key] = &groupMessageUniqueRecord{Status: "pending", Seq: seq, CreatedAt: time.Now().UTC()} + return nil +} + func (r *testGroupMessageRuntime) MarkFailed(_ types.TenantID, topic string, from types.Uid, clientMsgID string) error { key := topic + ":" + from.UserId() + ":" + clientMsgID r.claims[key] = &groupMessageUniqueRecord{Status: "failed", CreatedAt: time.Now().UTC()} diff --git a/server/store/types/types.go b/server/store/types/types.go index 5b82afa..b5c3409 100644 --- a/server/store/types/types.go +++ b/server/store/types/types.go @@ -1269,6 +1269,7 @@ type MessageSaveResult struct { SeqId int64 MarkedReadBySender bool Duplicate bool + Pending bool } // Range is a range of message SeqIDs. Low end is inclusive (closed), high end is exclusive (open): [Low, Hi).