pro/im_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix im forwarding and message list regressions

Browse Source
This commit is contained in:
richjohnrambo
2026-09-11 09:00:38 +08:00
parent 75d1aabcb0
commit 926b4bbd81
9 changed files with 331 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/db/mysql/report_group_regression_test.go
+54
View File
@@ -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]
}
server/db/mysql/tenant_adapter.go
+9 -3
View File
@@ -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
server/im_http.go
+95 -9
View File
@@ -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:
server/im_http_test.go
+136 -1
View File
@@ -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) {
server/kafka_message_persist.go
+9 -3
View File
@@ -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 {
server/store/group_message_runtime.go
+17
View File
@@ -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
server/store/group_message_runtime_test.go
+6
View File
@@ -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()}
server/store/types/types.go
+1
View File
@@ -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).