From a3df9d269c21c02a8cf2fa9a2f097e0f60bd03bf Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Thu, 24 Sep 2026 00:00:35 +0800 Subject: [PATCH] fix: broadcast forwarded messages after kafka publish --- server/im_http.go | 26 ++++++++++++++++---- server/im_http_test.go | 9 ++++--- server/kafka_message_persist.go | 43 ++++++++++----------------------- 3 files changed, 39 insertions(+), 39 deletions(-) diff --git a/server/im_http.go b/server/im_http.go index f797218..2e3e0b1 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -2734,7 +2734,7 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) { continue } commentResult = ensureComment() - saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(req.Context(), principal.TenantID, msg, nil, true, true) + saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, msg, nil, true) if err != nil { if saved != nil && saved.Pending { result := forwardResultUnknownPayload(target, saved.SeqId, relationErrorReason(err)) @@ -2756,6 +2756,10 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) { continue } if saved != nil && saved.Pending { + msg.SeqId = saved.SeqId + if !saved.Duplicate { + broadcastHTTPForwardMessage(principal.TenantID, msg, saved.MarkedReadBySender) + } result := map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "pending", "seq": saved.SeqId, "reason": "MESSAGE_PERSIST_PENDING"} result["client_mid"] = clientMid mergeForwardCommentResult(result, commentResult) @@ -2858,7 +2862,7 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { continue } commentResult = ensureComment() - saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(req.Context(), principal.TenantID, forwarded, nil, true, true) + saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true) if err != nil { if saved != nil && saved.Pending { forwarded.SeqId = saved.SeqId @@ -2885,6 +2889,10 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { continue } if saved != nil && saved.Pending { + forwarded.SeqId = saved.SeqId + if !saved.Duplicate { + broadcastHTTPForwardMessage(principal.TenantID, forwarded, saved.MarkedReadBySender) + } item := map[string]any{ "sources": bundle["source_refs"], "target_topic": target, @@ -2957,7 +2965,7 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { continue } commentResult = ensureComment() - saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(req.Context(), principal.TenantID, forwarded, nil, true, true) + saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true) if err != nil { if saved != nil && saved.Pending { forwarded.SeqId = saved.SeqId @@ -2980,6 +2988,10 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { continue } if saved != nil && saved.Pending { + forwarded.SeqId = saved.SeqId + if !saved.Duplicate { + broadcastHTTPForwardMessage(principal.TenantID, forwarded, saved.MarkedReadBySender) + } item := map[string]any{ "source": map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)}, "target_topic": target, @@ -5376,7 +5388,7 @@ func publishGroupAnnouncementTextMessage(ctx context.Context, tenantID types.Ten }, Content: map[string]any{"text": "群公告\n" + content}, } - saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(ctx, tenantID, msg, nil, true, true) + saved, err := saveMessageMaybeKafka(ctx, tenantID, msg, nil, true) if err != nil { return err } @@ -9511,7 +9523,7 @@ func saveForwardCommentMessage(ctx context.Context, principal *imPrincipal, targ result["comment_reason"] = relationErrorReason(err) return result } - saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(ctx, principal.TenantID, msg, nil, true, true) + saved, err := saveMessageMaybeKafka(ctx, principal.TenantID, msg, nil, true) if err != nil { if saved != nil && saved.Pending { result["comment_status"] = "pending" @@ -9524,6 +9536,10 @@ func saveForwardCommentMessage(ctx context.Context, principal *imPrincipal, targ return result } if saved != nil && saved.Pending { + msg.SeqId = saved.SeqId + if !saved.Duplicate { + broadcastHTTPForwardMessage(principal.TenantID, msg, saved.MarkedReadBySender) + } result["comment_status"] = "pending" result["comment_seq"] = saved.SeqId result["comment_reason"] = "MESSAGE_PERSIST_PENDING" diff --git a/server/im_http_test.go b/server/im_http_test.go index fabbfb5..400f28a 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -5828,6 +5828,7 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) { TenantID: 7, SeqId: 34, }, nil) + topics.EXPECT().GetSubs(types.TenantID(7), targetTopic, nil).Return(nil, nil) messages.EXPECT().GetOne(types.TenantID(7), "grpSource", int64(14)).Return(&types.Message{ TenantID: 7, Topic: "grpSource", @@ -5877,13 +5878,13 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) { if publishedPayload.Message.Topic != targetTopic || publishedPayload.Message.SeqId != 35 { t.Fatalf("unexpected kafka message: %#v", publishedPayload.Message) } - if !publishedPayload.BroadcastAfterPersist { - t.Fatalf("forwarded HTTP message must broadcast after kafka persistence: %#v", publishedPayload) - } select { case delivered := <-globals.hub.routeSrv: - t.Fatalf("pending forward should not broadcast immediately: %#v", delivered) + if delivered.Data == nil || delivered.Data.Topic != targetTopic || delivered.Data.SeqId != 35 { + t.Fatalf("unexpected immediate broadcast: %#v", delivered) + } default: + t.Fatalf("pending forward should broadcast immediately after kafka publish") } } diff --git a/server/kafka_message_persist.go b/server/kafka_message_persist.go index d6c1b96..fc11970 100644 --- a/server/kafka_message_persist.go +++ b/server/kafka_message_persist.go @@ -14,12 +14,11 @@ import ( ) type kafkaMessagePersistPayload struct { - Message types.Message `json:"message"` - AttachmentURLs []string `json:"attachment_urls,omitempty"` - ReadBySender bool `json:"read_by_sender,omitempty"` - MessageKind string `json:"message_kind"` - ShardMonth string `json:"shard_month,omitempty"` - BroadcastAfterPersist bool `json:"broadcast_after_persist,omitempty"` + Message types.Message `json:"message"` + AttachmentURLs []string `json:"attachment_urls,omitempty"` + ReadBySender bool `json:"read_by_sender,omitempty"` + MessageKind string `json:"message_kind"` + ShardMonth string `json:"shard_month,omitempty"` } var publishKafkaMessagePersist = publishKafkaEnvelope @@ -30,10 +29,6 @@ func startKafkaMessagePersistConsumer() { } func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) (*types.MessageSaveResult, error) { - return saveMessageMaybeKafkaWithPostPersistBroadcast(ctx, tenantID, msg, attachmentURLs, readBySender, false) -} - -func saveMessageMaybeKafkaWithPostPersistBroadcast(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender, broadcastAfterPersist bool) (*types.MessageSaveResult, error) { if !kafkaEnabled() { return store.Messages.Save(tenantID, msg, attachmentURLs, readBySender) } @@ -82,7 +77,7 @@ func saveMessageMaybeKafkaWithPostPersistBroadcast(ctx context.Context, tenantID if msg.ShardMonth == "" && isGroupMessage { msg.ShardMonth = candidateMonth } - if err = enqueueKafkaMessagePersist(ctx, tenantID, msg, attachmentURLs, readBySender, broadcastAfterPersist); err != nil { + if err = enqueueKafkaMessagePersist(ctx, tenantID, msg, attachmentURLs, readBySender); err != nil { if 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) @@ -104,18 +99,17 @@ func saveMessageMaybeKafkaWithPostPersistBroadcast(ctx context.Context, tenantID return &types.MessageSaveResult{SeqId: msg.SeqId, Pending: true, MarkedReadBySender: readBySender && !fromUid.IsZero()}, nil } -func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender, broadcastAfterPersist bool) error { +func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) error { kind := "p2p" if types.GetTopicCat(msg.Topic) == types.TopicCatGrp { kind = "group" } payload, err := json.Marshal(kafkaMessagePersistPayload{ - Message: *msg, - AttachmentURLs: attachmentURLs, - ReadBySender: readBySender, - MessageKind: kind, - ShardMonth: msg.ShardMonth, - BroadcastAfterPersist: broadcastAfterPersist, + Message: *msg, + AttachmentURLs: attachmentURLs, + ReadBySender: readBySender, + MessageKind: kind, + ShardMonth: msg.ShardMonth, }) if err != nil { return err @@ -175,8 +169,7 @@ func handleKafkaMessagePersistFromTopic(ctx context.Context, envelope kafkaEnvel if err = validateKafkaMessageKind(payload); err != nil { return markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error()) } - saveResult, err := store.PersistPreparedMessage(tenantID, &payload.Message, nil, payload.ReadBySender) - if err != nil && !errors.Is(err, types.ErrDuplicate) { + if _, err = store.PersistPreparedMessage(tenantID, &payload.Message, nil, payload.ReadBySender); err != nil && !errors.Is(err, types.ErrDuplicate) { failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, kafkaCommandErrorReason(err), err.Error(), time.Now().UTC().Round(time.Millisecond)) if markErr != nil { @@ -202,16 +195,6 @@ func handleKafkaMessagePersistFromTopic(ctx context.Context, envelope kafkaEnvel return err } } - if payload.BroadcastAfterPersist && (saveResult == nil || !saveResult.Duplicate) { - markedReadBySender := payload.ReadBySender - if saveResult != nil { - if saveResult.SeqId > 0 { - payload.Message.SeqId = saveResult.SeqId - } - markedReadBySender = saveResult.MarkedReadBySender - } - broadcastHTTPForwardMessage(tenantID, &payload.Message, markedReadBySender) - } return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond)) }