fix: broadcast forwarded messages after kafka publish

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"

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")
 	}
 }
 

server/kafka_message_persist.go

@@ -19,7 +19,6 @@ type kafkaMessagePersistPayload struct {
 	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"`
 }
 
 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,7 +99,7 @@ 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"
@@ -115,7 +110,6 @@ func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, ms
 		ReadBySender:   readBySender,
 		MessageKind:    kind,
 		ShardMonth:     msg.ShardMonth,
-		BroadcastAfterPersist: broadcastAfterPersist,
 	})
 	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 = store.PersistPreparedMessage(tenantID, &payload.Message, nil, payload.ReadBySender); err != nil && !errors.Is(err, types.ErrDuplicate) {
-	if 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))
 }