diff --git a/server/im_http.go b/server/im_http.go
index e90d52b..f797218 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 := saveMessageMaybeKafka(req.Context(), principal.TenantID, msg, nil, true)
+			saved, err := saveMessageMaybeKafkaWithPostPersistBroadcast(req.Context(), principal.TenantID, msg, nil, true, true)
 			if err != nil {
 				if saved != nil && saved.Pending {
 					result := forwardResultUnknownPayload(target, saved.SeqId, relationErrorReason(err))
@@ -2858,7 +2858,7 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
 				continue
 			}
 			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 saved != nil && saved.Pending {
 					forwarded.SeqId = saved.SeqId
@@ -2957,7 +2957,7 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
 				continue
 			}
 			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 saved != nil && saved.Pending {
 					forwarded.SeqId = saved.SeqId
@@ -5376,7 +5376,7 @@ func publishGroupAnnouncementTextMessage(ctx context.Context, tenantID types.Ten
 		},
 		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 {
 		return err
 	}
@@ -9511,7 +9511,7 @@ func saveForwardCommentMessage(ctx context.Context, principal *imPrincipal, targ
 		result["comment_reason"] = relationErrorReason(err)
 		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 saved != nil && saved.Pending {
 			result["comment_status"] = "pending"
diff --git a/server/im_http_test.go b/server/im_http_test.go
index 0923a52..fabbfb5 100644
--- a/server/im_http_test.go
+++ b/server/im_http_test.go
@@ -5877,6 +5877,9 @@ 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)
diff --git a/server/kafka_message_persist.go b/server/kafka_message_persist.go
index fc11970..d6c1b96 100644
--- a/server/kafka_message_persist.go
+++ b/server/kafka_message_persist.go
@@ -14,11 +14,12 @@ 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"`
+	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"`
 }
 
 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) {
+	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)
 	}
@@ -77,7 +82,7 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty
 	if msg.ShardMonth == "" && isGroupMessage {
 		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 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)
@@ -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
 }
 
-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"
 	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,
+		Message:               *msg,
+		AttachmentURLs:        attachmentURLs,
+		ReadBySender:          readBySender,
+		MessageKind:           kind,
+		ShardMonth:            msg.ShardMonth,
+		BroadcastAfterPersist: broadcastAfterPersist,
 	})
 	if err != nil {
 		return err
@@ -169,7 +175,8 @@ func handleKafkaMessagePersistFromTopic(ctx context.Context, envelope kafkaEnvel
 	if err = validateKafkaMessageKind(payload); err != nil {
 		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,
 			kafkaCommandErrorReason(err), err.Error(), time.Now().UTC().Round(time.Millisecond))
 		if markErr != nil {
@@ -195,6 +202,16 @@ 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))
 }