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

fix: broadcast async forwarded messages after persistence

Browse Source
This commit is contained in:
richjohnrambo
2026-09-23 18:14:34 +08:00
parent 3e92021254
commit da02eae6a4
3 changed files with 38 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/im_http.go
+5 -5
View File
@@ -2734,7 +2734,7 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) {
continue continue
} }
commentResult = ensureComment() commentResult = ensureComment()
saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, msg, nil, true) saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(req.Context(), principal.TenantID, msg, nil, true, true)
if err != nil { if err != nil {
if saved != nil && saved.Pending { if saved != nil && saved.Pending {
result := forwardResultUnknownPayload(target, saved.SeqId, relationErrorReason(err)) result := forwardResultUnknownPayload(target, saved.SeqId, relationErrorReason(err))
@@ -2858,7 +2858,7 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
continue continue
} }
commentResult = ensureComment() commentResult = ensureComment()
saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true) saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(req.Context(), principal.TenantID, forwarded, nil, true, true)
if err != nil { if err != nil {
if saved != nil && saved.Pending { if saved != nil && saved.Pending {
forwarded.SeqId = saved.SeqId forwarded.SeqId = saved.SeqId
@@ -2957,7 +2957,7 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
continue continue
} }
commentResult = ensureComment() commentResult = ensureComment()
saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true) saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(req.Context(), principal.TenantID, forwarded, nil, true, true)
if err != nil { if err != nil {
if saved != nil && saved.Pending { if saved != nil && saved.Pending {
forwarded.SeqId = saved.SeqId forwarded.SeqId = saved.SeqId
@@ -5376,7 +5376,7 @@ func publishGroupAnnouncementTextMessage(ctx context.Context, tenantID types.Ten
}, },
Content: map[string]any{"text": "群公告\n" + content}, Content: map[string]any{"text": "群公告\n" + content},
} }
saved, err := saveMessageMaybeKafka(ctx, tenantID, msg, nil, true) saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(ctx, tenantID, msg, nil, true, true)
if err != nil { if err != nil {
return err return err
} }
@@ -9511,7 +9511,7 @@ func saveForwardCommentMessage(ctx context.Context, principal *imPrincipal, targ
result["comment_reason"] = relationErrorReason(err) result["comment_reason"] = relationErrorReason(err)
return result return result
} }
saved, err := saveMessageMaybeKafka(ctx, principal.TenantID, msg, nil, true) saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(ctx, principal.TenantID, msg, nil, true, true)
if err != nil { if err != nil {
if saved != nil && saved.Pending { if saved != nil && saved.Pending {
result["comment_status"] = "pending" result["comment_status"] = "pending"
server/im_http_test.go
+3
View File
@@ -5877,6 +5877,9 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) {
if publishedPayload.Message.Topic != targetTopic || publishedPayload.Message.SeqId != 35 { if publishedPayload.Message.Topic != targetTopic || publishedPayload.Message.SeqId != 35 {
t.Fatalf("unexpected kafka message: %#v", publishedPayload.Message) t.Fatalf("unexpected kafka message: %#v", publishedPayload.Message)
} }
if !publishedPayload.BroadcastAfterPersist {
t.Fatalf("forwarded HTTP message must broadcast after kafka persistence: %#v", publishedPayload)
}
select { select {
case delivered := <-globals.hub.routeSrv: case delivered := <-globals.hub.routeSrv:
t.Fatalf("pending forward should not broadcast immediately: %#v", delivered) t.Fatalf("pending forward should not broadcast immediately: %#v", delivered)
server/kafka_message_persist.go
+30 -13
View File
@@ -14,11 +14,12 @@ import (
) )
type kafkaMessagePersistPayload struct { type kafkaMessagePersistPayload struct {
Message types.Message `json:"message"` Message types.Message `json:"message"`
AttachmentURLs []string `json:"attachment_urls,omitempty"` AttachmentURLs []string `json:"attachment_urls,omitempty"`
ReadBySender bool `json:"read_by_sender,omitempty"` ReadBySender bool `json:"read_by_sender,omitempty"`
MessageKind string `json:"message_kind"` MessageKind string `json:"message_kind"`
ShardMonth string `json:"shard_month,omitempty"` ShardMonth string `json:"shard_month,omitempty"`
BroadcastAfterPersist bool `json:"broadcast_after_persist,omitempty"`
} }
var publishKafkaMessagePersist = publishKafkaEnvelope var publishKafkaMessagePersist = publishKafkaEnvelope
@@ -29,6 +30,10 @@ func startKafkaMessagePersistConsumer() {
} }
func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) (*types.MessageSaveResult, error) { 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() { if !kafkaEnabled() {
return store.Messages.Save(tenantID, msg, attachmentURLs, readBySender) return store.Messages.Save(tenantID, msg, attachmentURLs, readBySender)
} }
@@ -77,7 +82,7 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty
if msg.ShardMonth == "" && isGroupMessage { if msg.ShardMonth == "" && isGroupMessage {
msg.ShardMonth = candidateMonth msg.ShardMonth = candidateMonth
} }
if err = enqueueKafkaMessagePersist(ctx, tenantID, msg, attachmentURLs, readBySender); err != nil { if err = enqueueKafkaMessagePersist(ctx, tenantID, msg, attachmentURLs, readBySender, broadcastAfterPersist); err != nil {
if clientMsgID != "" { if clientMsgID != "" {
if markErr := store.MarkMessageUniquePending(tenantID, msg.Topic, fromUid, clientMsgID, msg.SeqId); markErr != nil { 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) logs.Warn.Printf("kafka message persist: failed to mark message unique key pending topic=%s seq=%d: %v", msg.Topic, msg.SeqId, markErr)
@@ -99,17 +104,18 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty
return &types.MessageSaveResult{SeqId: msg.SeqId, Pending: true, MarkedReadBySender: readBySender && !fromUid.IsZero()}, nil 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 bool) error { func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender, broadcastAfterPersist bool) error {
kind := "p2p" kind := "p2p"
if types.GetTopicCat(msg.Topic) == types.TopicCatGrp { if types.GetTopicCat(msg.Topic) == types.TopicCatGrp {
kind = "group" kind = "group"
} }
payload, err := json.Marshal(kafkaMessagePersistPayload{ payload, err := json.Marshal(kafkaMessagePersistPayload{
Message: *msg, Message: *msg,
AttachmentURLs: attachmentURLs, AttachmentURLs: attachmentURLs,
ReadBySender: readBySender, ReadBySender: readBySender,
MessageKind: kind, MessageKind: kind,
ShardMonth: msg.ShardMonth, ShardMonth: msg.ShardMonth,
BroadcastAfterPersist: broadcastAfterPersist,
}) })
if err != nil { if err != nil {
return err return err
@@ -169,7 +175,8 @@ func handleKafkaMessagePersistFromTopic(ctx context.Context, envelope kafkaEnvel
if err = validateKafkaMessageKind(payload); err != nil { if err = validateKafkaMessageKind(payload); err != nil {
return markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error()) return markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error())
} }
if _, err = store.PersistPreparedMessage(tenantID, &payload.Message, nil, payload.ReadBySender); err != nil && !errors.Is(err, types.ErrDuplicate) { saveResult, err := store.PersistPreparedMessage(tenantID, &payload.Message, nil, payload.ReadBySender)
if err != nil && !errors.Is(err, types.ErrDuplicate) {
failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup,
kafkaCommandErrorReason(err), err.Error(), time.Now().UTC().Round(time.Millisecond)) kafkaCommandErrorReason(err), err.Error(), time.Now().UTC().Round(time.Millisecond))
if markErr != nil { if markErr != nil {
@@ -195,6 +202,16 @@ func handleKafkaMessagePersistFromTopic(ctx context.Context, envelope kafkaEnvel
return err 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)) return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond))
} }