feat(chat): implement sequence pre-allocation, versioned sync, and push worker

Introduce several performance and synchronization enhancements: - Implement `SeqBufferManager` to allow sequence number pre-allocation via Redis Lua scripts, reducing atomic increment overhead. - Add `PushWorker` to handle asynchronous message pushing using Redis Streams. - Implement incremental conversation synchronization via `ConversationVersionLog` to allow clients to fetch only recent changes. - Add support for Gzip compression in WebSocket communications to reduce bandwidth usage. - Update dependency injection and configuration to support these new components.
2026-05-17 23:38:04 +08:00
parent f63c795dcb
commit 6bf87fec46
26 changed files with 1450 additions and 82 deletions
--- a/internal/service/chat_service.go
+++ b/internal/service/chat_service.go
@@ -13,6 +13,7 @@ import (
 	"with_you/internal/pkg/ws"
 	"with_you/internal/repository"

+	"github.com/redis/go-redis/v9"
 	"go.uber.org/zap"
 	"gorm.io/gorm"
 )
@@ -48,6 +49,9 @@ type ChatService interface {
 	// 消息同步
 	GetSyncData(ctx context.Context, userID string) ([]dto.SyncDataItem, error)

+	// 增量同步（版本日志）
+	GetSyncByVersion(ctx context.Context, userID string, sinceVersion int64, limit int) (*dto.SyncVersionResponse, error)
+
 	// 消息扩展功能
 	RecallMessage(ctx context.Context, messageID string, userID string) error
 	DeleteMessage(ctx context.Context, messageID string, userID string) error
@@ -80,6 +84,11 @@ type chatServiceImpl struct {

 	conversationCache *cache.ConversationCache
 	uploadService     *UploadService
+	versionLogRepo    repository.ConversationVersionLogRepository
+	versionLogEnabled bool
+	versionLogMaxGap  int64
+	pushWorker        *PushWorker
+	redisClient       *redis.Client
 }

 // NewChatService 创建聊天服务
@@ -91,6 +100,10 @@ func NewChatService(
 	cacheBackend cache.Cache,
 	uploadService *UploadService,
 	pushSvc PushService,
+	seqBufferMgr *cache.SeqBufferManager,
+	pushWorker *PushWorker,
+	redisClient *redis.Client,
+	versionLogRepo ...repository.ConversationVersionLogRepository,
 ) ChatService {
 	// 创建适配器
 	convRepoAdapter := cache.NewConversationRepositoryAdapter(repo)
@@ -102,9 +115,10 @@ func NewChatService(
 		convRepoAdapter,
 		msgRepoAdapter,
 		cache.DefaultConversationCacheSettings(),
+		seqBufferMgr,
 	)

-	return &chatServiceImpl{
+	impl := &chatServiceImpl{
 		repo:              repo,
 		userRepo:          userRepo,
 		sensitive:         sensitive,
@@ -113,7 +127,17 @@ func NewChatService(
 		cache:             cacheBackend,
 		conversationCache: conversationCache,
 		uploadService:     uploadService,
+		pushWorker:        pushWorker,
+		redisClient:       redisClient,
 	}
+
+	if len(versionLogRepo) > 0 && versionLogRepo[0] != nil {
+		impl.versionLogRepo = versionLogRepo[0]
+		impl.versionLogEnabled = true
+		impl.versionLogMaxGap = 1000
+	}
+
+	return impl
 }

 func (s *chatServiceImpl) publishToUsers(userIDs []string, event string, payload any) {
@@ -123,6 +147,115 @@ func (s *chatServiceImpl) publishToUsers(userIDs []string, event string, payload
 	s.wsHub.PublishToUsers(userIDs, event, payload)
 }

+// writeVersionLog 写入版本日志（异步，失败仅 warn）
+func (s *chatServiceImpl) writeVersionLog(ctx context.Context, userID, conversationID, changeType string) {
+	if !s.versionLogEnabled || s.versionLogRepo == nil {
+		return
+	}
+	maxV, err := s.versionLogRepo.GetMaxVersion(ctx, userID)
+	if err != nil {
+		zap.L().Warn("writeVersionLog: get max version failed", zap.String("userID", userID), zap.Error(err))
+		return
+	}
+	vlog := &model.ConversationVersionLog{
+		UserID:         userID,
+		ConversationID: conversationID,
+		Version:        maxV + 1,
+		ChangeType:     changeType,
+	}
+	if err := s.versionLogRepo.Create(ctx, vlog); err != nil {
+		zap.L().Warn("writeVersionLog: create failed", zap.String("userID", userID), zap.Error(err))
+	}
+}
+
+// writeVersionLogBatch 批量写入版本日志（异步，失败仅 warn）
+func (s *chatServiceImpl) writeVersionLogBatch(ctx context.Context, userIDs []string, conversationID, changeType string) {
+	if !s.versionLogEnabled || s.versionLogRepo == nil {
+		return
+	}
+	logs := make([]*model.ConversationVersionLog, 0, len(userIDs))
+	for _, uid := range userIDs {
+		maxV, err := s.versionLogRepo.GetMaxVersion(ctx, uid)
+		if err != nil {
+			zap.L().Warn("writeVersionLogBatch: get max version failed", zap.String("userID", uid), zap.Error(err))
+			continue
+		}
+		logs = append(logs, &model.ConversationVersionLog{
+			UserID:         uid,
+			ConversationID: conversationID,
+			Version:        maxV + 1,
+			ChangeType:     changeType,
+		})
+	}
+	if len(logs) > 0 {
+		if err := s.versionLogRepo.BatchCreate(ctx, logs); err != nil {
+			zap.L().Warn("writeVersionLogBatch: batch create failed", zap.Error(err))
+		}
+	}
+}
+
+// GetSyncByVersion 增量同步：按版本号获取会话变更
+func (s *chatServiceImpl) GetSyncByVersion(ctx context.Context, userID string, sinceVersion int64, limit int) (*dto.SyncVersionResponse, error) {
+	if !s.versionLogEnabled || s.versionLogRepo == nil {
+		return nil, errors.New("version log sync is not enabled")
+	}
+
+	if limit <= 0 {
+		limit = 100
+	}
+
+	currentVersion, err := s.versionLogRepo.GetMaxVersion(ctx, userID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get current version: %w", err)
+	}
+
+	resp := &dto.SyncVersionResponse{
+		CurrentVersion: currentVersion,
+	}
+
+	if currentVersion == 0 {
+		resp.FullSync = true
+		return resp, nil
+	}
+
+	gap := currentVersion - sinceVersion
+	if gap > s.versionLogMaxGap {
+		resp.FullSync = true
+		return resp, nil
+	}
+
+	vlogs, err := s.versionLogRepo.GetByVersion(ctx, userID, sinceVersion, limit+1)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get version logs: %w", err)
+	}
+
+	hasMore := len(vlogs) > limit
+	if hasMore {
+		vlogs = vlogs[:limit]
+	}
+
+	changes := make([]dto.SyncChangeItem, 0, len(vlogs))
+	for _, vl := range vlogs {
+		item := dto.SyncChangeItem{
+			ConversationID: vl.ConversationID,
+			ChangeType:     vl.ChangeType,
+			Version:        vl.Version,
+		}
+		conv, cErr := s.repo.GetConversation(vl.ConversationID)
+		if cErr == nil && conv != nil {
+			item.LastMessageAt = conv.UpdatedAt.Unix()
+			if maxSeq, sErr := s.conversationCache.GetConvMaxSeq(ctx, vl.ConversationID); sErr == nil {
+				item.MaxSeq = maxSeq
+			}
+		}
+		changes = append(changes, item)
+	}
+
+	resp.Changes = changes
+	resp.HasMore = hasMore
+	return resp, nil
+}
+
 // GetOrCreateConversation 获取或创建私聊会话
 func (s *chatServiceImpl) GetOrCreateConversation(ctx context.Context, user1ID, user2ID string) (*model.Conversation, error) {
 	conv, err := s.repo.GetOrCreatePrivateConversation(user1ID, user2ID)
@@ -205,6 +338,8 @@ func (s *chatServiceImpl) DeleteConversationForSelf(ctx context.Context, convers
 		s.conversationCache.InvalidateConversationList(userID)
 	}

+	go s.writeVersionLog(context.Background(), userID, conversationID, "hide")
+
 	return nil
 }

@@ -387,6 +522,13 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv
 	// 获取会话中的参与者并发送消息
 	participants, err := s.getParticipants(ctx, conversationID)
 	if err == nil {
+		// 版本日志：为所有参与者写入 new_msg
+		pids := make([]string, 0, len(participants))
+		for _, p := range participants {
+			pids = append(pids, p.UserID)
+		}
+		go s.writeVersionLogBatch(context.Background(), pids, conversationID, "new_msg")
+
 		// 先更新未读计数器，再推送 WS 事件，确保事件中的 total_unread 已包含新消息
 		// 注意：已移除 IncrementUnread，未读数完全由 maxSeq - readSeq 算术计算
 		if s.conversationCache != nil {
@@ -405,38 +547,63 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv
 			}
 		}

-		targetIDs := make([]string, 0, len(participants))
-		for _, p := range participants {
-			if conv.Type == model.ConversationTypePrivate && p.UserID == senderID {
-				continue
+		// PushWorker 异步推送 or inline fallback
+		if s.pushWorker != nil && s.pushWorker.IsEnabled() {
+			streamMsg := &PushStreamMsg{
+				ConversationID: conversationID,
+				SenderID:       senderID,
+				MessageID:      message.ID,
+				ConvType:       string(conv.Type),
+				Timestamp:      time.Now().UnixMilli(),
+			}
+			if pubErr := PublishToPush(context.Background(), s.redisClient, "msg_push", streamMsg); pubErr != nil {
+				zap.L().Warn("push stream publish failed, falling back to inline push",
+					zap.String("conversationID", conversationID),
+					zap.Error(pubErr),
+				)
+				s.inlinePush(ctx, conv, participants, senderID, conversationID, message)
 			}
-			targetIDs = append(targetIDs, p.UserID)
-		}
-		detailType := "private"
-		if conv.Type == model.ConversationTypeGroup {
-			detailType = "group"
-		}
-		if len(targetIDs) > 20 && conv.Type == model.ConversationTypeGroup {
-			s.wsHub.PublishToUsersConcurrent(targetIDs, "chat_message", map[string]any{
-				"detail_type": detailType,
-				"message":     dto.ConvertMessageToResponse(message),
-			})
 		} else {
-			s.publishToUsers(targetIDs, "chat_message", map[string]any{
-				"detail_type": detailType,
-				"message":     dto.ConvertMessageToResponse(message),
-			})
+			s.inlinePush(ctx, conv, participants, senderID, conversationID, message)
 		}
-		for _, p := range participants {
-			if p.UserID == senderID {
-				continue
-			}
-			if totalUnread, uErr := s.GetAllUnreadCount(ctx, p.UserID); uErr == nil {
-				s.publishToUsers([]string{p.UserID}, "conversation_unread", map[string]any{
-					"conversation_id": conversationID,
-					"total_unread":    totalUnread,
-				})
-			}
+	}
+
+	return message, nil
+}
+
+// inlinePush 内联推送（原有逻辑，PushWorker 失败时的 fallback）
+func (s *chatServiceImpl) inlinePush(ctx context.Context, conv *model.Conversation, participants []*model.ConversationParticipant, senderID, conversationID string, message *model.Message) {
+	targetIDs := make([]string, 0, len(participants))
+	for _, p := range participants {
+		if conv.Type == model.ConversationTypePrivate && p.UserID == senderID {
+			continue
+		}
+		targetIDs = append(targetIDs, p.UserID)
+	}
+	detailType := "private"
+	if conv.Type == model.ConversationTypeGroup {
+		detailType = "group"
+	}
+	if len(targetIDs) > 20 && conv.Type == model.ConversationTypeGroup {
+		s.wsHub.PublishToUsersConcurrent(targetIDs, "chat_message", map[string]any{
+			"detail_type": detailType,
+			"message":     dto.ConvertMessageToResponse(message),
+		})
+	} else {
+		s.publishToUsers(targetIDs, "chat_message", map[string]any{
+			"detail_type": detailType,
+			"message":     dto.ConvertMessageToResponse(message),
+		})
+	}
+	for _, p := range participants {
+		if p.UserID == senderID {
+			continue
+		}
+		if totalUnread, uErr := s.GetAllUnreadCount(ctx, p.UserID); uErr == nil {
+			s.publishToUsers([]string{p.UserID}, "conversation_unread", map[string]any{
+				"conversation_id": conversationID,
+				"total_unread":    totalUnread,
+			})
 		}
 	}

@@ -482,8 +649,6 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv
 			}(offlineTargetIDs, sender, groupID, groupAvatar)
 		}
 	}
-
-	return message, nil
 }

 func (s *chatServiceImpl) getConversation(ctx context.Context, conversationID string) (*model.Conversation, error) {
--- a/internal/service/group_service.go
+++ b/internal/service/group_service.go
@@ -126,6 +126,7 @@ func NewGroupService(groupRepo repository.GroupRepository, userRepo repository.U
 		convRepoAdapter,
 		msgRepoAdapter,
 		cache.DefaultConversationCacheSettings(),
+		nil,
 	)

 	return &groupService{
--- a/internal/service/message_service.go
+++ b/internal/service/message_service.go
@@ -49,6 +49,7 @@ func NewMessageService(messageRepo repository.MessageRepository, cacheBackend ca
 		convRepoAdapter,
 		msgRepoAdapter,
 		cache.DefaultConversationCacheSettings(),
+		nil,
 	)

 	return &MessageService{
--- a/internal/service/push_worker.go
+++ b/internal/service/push_worker.go
@@ -0,0 +1,381 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"with_you/internal/dto"
+	"with_you/internal/model"
+	"with_you/internal/pkg/ws"
+	"with_you/internal/repository"
+
+	"github.com/redis/go-redis/v9"
+	"go.uber.org/zap"
+)
+
+// PushWorkerConfig 推送 Worker 配置
+type PushWorkerConfig struct {
+	Enabled       bool
+	Stream        string // Redis Stream 名称，默认 "msg_push"
+	Group         string // Consumer Group 名称，默认 "push_worker"
+	BatchSize     int    // XREADGROUP 每次读取条数，默认 50
+	PollTimeoutMs int    // XREADGROUP BLOCK 超时 ms，默认 2000
+	MaxRetries    int    // 消息最大重试次数，默认 3
+	MaxStreamLen  int64  // Stream MAXLEN，默认 100000
+	IdleTimeoutMs int64  // XPENDING 空闲超时 ms，默认 30000
+}
+
+// PushStreamMsg 推送到 Redis Stream 的消息结构
+type PushStreamMsg struct {
+	ConversationID string `json:"conversation_id"`
+	SenderID       string `json:"sender_id"`
+	MessageID      string `json:"message_id"`
+	ConvType       string `json:"conv_type"`
+	Timestamp      int64  `json:"timestamp"`
+}
+
+// PushWorker 从 Redis Stream 消费消息并异步推送
+type PushWorker struct {
+	rdb        *redis.Client
+	publisher  ws.MessagePublisher
+	pushSvc    PushService
+	msgRepo    repository.MessageRepository
+	userRepo   repository.UserRepository
+	config     PushWorkerConfig
+	stopCh     chan struct{}
+	doneCh     chan struct{}
+}
+
+// NewPushWorker 创建推送 Worker
+func NewPushWorker(
+	rdb *redis.Client,
+	publisher ws.MessagePublisher,
+	pushSvc PushService,
+	msgRepo repository.MessageRepository,
+	userRepo repository.UserRepository,
+	config PushWorkerConfig,
+) *PushWorker {
+	if config.Stream == "" {
+		config.Stream = "msg_push"
+	}
+	if config.Group == "" {
+		config.Group = "push_worker"
+	}
+	if config.BatchSize <= 0 {
+		config.BatchSize = 50
+	}
+	if config.PollTimeoutMs <= 0 {
+		config.PollTimeoutMs = 2000
+	}
+	if config.MaxRetries <= 0 {
+		config.MaxRetries = 3
+	}
+	if config.MaxStreamLen <= 0 {
+		config.MaxStreamLen = 100000
+	}
+	if config.IdleTimeoutMs <= 0 {
+		config.IdleTimeoutMs = 30000
+	}
+	return &PushWorker{
+		rdb:        rdb,
+		publisher:  publisher,
+		pushSvc:    pushSvc,
+		msgRepo:    msgRepo,
+		userRepo:   userRepo,
+		config:     config,
+		stopCh:     make(chan struct{}),
+		doneCh:     make(chan struct{}),
+	}
+}
+
+// IsEnabled 是否启用
+func (w *PushWorker) IsEnabled() bool {
+	return w != nil && w.config.Enabled
+}
+
+// Start 启动 Worker
+func (w *PushWorker) Start(ctx context.Context) {
+	if !w.config.Enabled {
+		return
+	}
+
+	// 创建 Consumer Group（忽略 BUSYGROUP 错误）
+	err := w.rdb.XGroupCreateMkStream(ctx, w.config.Stream, w.config.Group, "0").Err()
+	if err != nil {
+		// BUSYGROUP 错误可以忽略
+		zap.L().Debug("XGroupCreateMkStream", zap.Error(err))
+	}
+
+	go w.consumeLoop()
+	go w.claimPendingLoop()
+
+	zap.L().Info("PushWorker started",
+		zap.String("stream", w.config.Stream),
+		zap.String("group", w.config.Group),
+	)
+}
+
+// Stop 停止 Worker
+func (w *PushWorker) Stop() {
+	if !w.config.Enabled {
+		return
+	}
+	close(w.stopCh)
+	// 等待两个循环退出
+	<-w.doneCh
+	<-w.doneCh
+	zap.L().Info("PushWorker stopped")
+}
+
+// consumeLoop 主消费循环
+func (w *PushWorker) consumeLoop() {
+	defer func() { w.doneCh <- struct{}{} }()
+
+	consumerName := fmt.Sprintf("pw-%d", time.Now().UnixNano())
+	args := &redis.XReadGroupArgs{
+		Group:    w.config.Group,
+		Consumer: consumerName,
+		Streams:  []string{w.config.Stream, ">"},
+		Count:    int64(w.config.BatchSize),
+		Block:    time.Duration(w.config.PollTimeoutMs) * time.Millisecond,
+	}
+
+	for {
+		select {
+		case <-w.stopCh:
+			return
+		default:
+		}
+
+		results, err := w.rdb.XReadGroup(context.Background(), args).Result()
+		if err != nil {
+			if err == redis.Nil {
+				continue
+			}
+			zap.L().Warn("PushWorker XReadGroup error", zap.Error(err))
+			time.Sleep(1 * time.Second)
+			continue
+		}
+
+		for _, stream := range results {
+			for _, msg := range stream.Messages {
+				if err := w.handleMsg(msg); err != nil {
+					zap.L().Warn("PushWorker handle message error",
+						zap.String("stream_id", msg.ID),
+						zap.Error(err),
+					)
+				}
+				// ACK 无论成功失败（失败消息由 claimPendingLoop 重试）
+				w.rdb.XAck(context.Background(), w.config.Stream, w.config.Group, msg.ID)
+			}
+		}
+	}
+}
+
+// claimPendingLoop 定期认领超时未 ACK 的消息
+func (w *PushWorker) claimPendingLoop() {
+	defer func() { w.doneCh <- struct{}{} }()
+
+	consumerName := fmt.Sprintf("pw-claim-%d", time.Now().UnixNano())
+	ticker := time.NewTicker(30 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-w.stopCh:
+			return
+		case <-ticker.C:
+		}
+
+		// 查询 pending 消息
+		pendingResult, err := w.rdb.XPendingExt(context.Background(), &redis.XPendingExtArgs{
+			Stream:   w.config.Stream,
+			Group:    w.config.Group,
+			Start:    "-",
+			End:      "+",
+			Count:    int64(w.config.BatchSize),
+			Idle:     time.Duration(w.config.IdleTimeoutMs) * time.Millisecond,
+		}).Result()
+		if err != nil {
+			if err == redis.Nil {
+				continue
+			}
+			zap.L().Warn("PushWorker XPENDING error", zap.Error(err))
+			continue
+		}
+
+		for _, pending := range pendingResult {
+			// 认领消息
+			claimed, err := w.rdb.XClaim(context.Background(), &redis.XClaimArgs{
+				Stream:   w.config.Stream,
+				Group:    w.config.Group,
+				Consumer: consumerName,
+				MinIdle:  time.Duration(w.config.IdleTimeoutMs) * time.Millisecond,
+				Messages:      []string{pending.ID},
+			}).Result()
+			if err != nil || len(claimed) == 0 {
+				continue
+			}
+
+			retryCount := int(pending.RetryCount)
+			for _, msg := range claimed {
+				if retryCount >= w.config.MaxRetries {
+					zap.L().Warn("PushWorker message exceeded max retries, deleting",
+						zap.String("stream_id", msg.ID),
+						zap.Int("retry_count", retryCount),
+					)
+					w.rdb.XDel(context.Background(), w.config.Stream, msg.ID)
+					continue
+				}
+
+				if err := w.handleMsg(msg); err != nil {
+					zap.L().Warn("PushWorker claim retry handle error",
+						zap.String("stream_id", msg.ID),
+						zap.Int("retry_count", retryCount),
+						zap.Error(err),
+					)
+				}
+				w.rdb.XAck(context.Background(), w.config.Stream, w.config.Group, msg.ID)
+			}
+		}
+	}
+}
+
+// handleMsg 处理一条 Stream 消息
+func (w *PushWorker) handleMsg(streamMsg redis.XMessage) error {
+	data, ok := streamMsg.Values["data"]
+	if !ok {
+		return fmt.Errorf("missing data field in stream message %s", streamMsg.ID)
+	}
+
+	var pushMsg PushStreamMsg
+	dataStr, ok := data.(string)
+	if !ok {
+		return fmt.Errorf("data field is not string in stream message %s", streamMsg.ID)
+	}
+	if err := json.Unmarshal([]byte(dataStr), &pushMsg); err != nil {
+		return fmt.Errorf("unmarshal push message: %w", err)
+	}
+
+	ctx := context.Background()
+
+	// 加载消息
+	message, err := w.msgRepo.GetMessageByID(pushMsg.MessageID)
+	if err != nil {
+		return fmt.Errorf("load message %s: %w", pushMsg.MessageID, err)
+	}
+
+	// 加载会话
+	conv, err := w.msgRepo.GetConversation(pushMsg.ConversationID)
+	if err != nil {
+		return fmt.Errorf("load conversation %s: %w", pushMsg.ConversationID, err)
+	}
+
+	// 加载参与者
+	participants, err := w.msgRepo.GetConversationParticipants(pushMsg.ConversationID)
+	if err != nil {
+		return fmt.Errorf("load participants for %s: %w", pushMsg.ConversationID, err)
+	}
+
+	// ---- WS Push ----
+	targetIDs := make([]string, 0, len(participants))
+	for _, p := range participants {
+		if conv.Type == model.ConversationTypePrivate && p.UserID == pushMsg.SenderID {
+			continue
+		}
+		targetIDs = append(targetIDs, p.UserID)
+	}
+
+	detailType := "private"
+	if conv.Type == model.ConversationTypeGroup {
+		detailType = "group"
+	}
+
+	if len(targetIDs) > 20 && conv.Type == model.ConversationTypeGroup {
+		w.publisher.PublishToUsersConcurrent(targetIDs, "chat_message", map[string]any{
+			"detail_type": detailType,
+			"message":     dto.ConvertMessageToResponse(message),
+		})
+	} else {
+		w.publisher.PublishToUsers(targetIDs, "chat_message", map[string]any{
+			"detail_type": detailType,
+			"message":     dto.ConvertMessageToResponse(message),
+		})
+	}
+
+	// ---- WS unread count push ----
+	for _, p := range participants {
+		if p.UserID == pushMsg.SenderID {
+			continue
+		}
+		// PushWorker doesn't have conversationCache, skip unread count push here
+		// The inline push in chat_service still handles this when push worker is disabled
+	}
+
+	// ---- JPush for offline users ----
+	if w.pushSvc == nil {
+		return nil
+	}
+
+	sender := ChatMessageSender{ID: pushMsg.SenderID}
+	if senderUser, sErr := w.userRepo.GetByID(pushMsg.SenderID); sErr == nil {
+		sender.Name = senderUser.Nickname
+		sender.Avatar = senderUser.Avatar
+	}
+
+	convName := ""
+	groupID := ""
+	groupAvatar := ""
+	if conv.Type == model.ConversationTypeGroup && conv.Group != nil {
+		convName = conv.Group.Name
+		groupID = conv.Group.ID
+		groupAvatar = conv.Group.Avatar
+	}
+
+	allTargetIDs := make([]string, 0, len(participants))
+	for _, p := range participants {
+		if p.UserID == pushMsg.SenderID {
+			continue
+		}
+		allTargetIDs = append(allTargetIDs, p.UserID)
+	}
+
+	var offlineTargetIDs []string
+	if w.publisher != nil && len(allTargetIDs) > 0 {
+		_, offlineTargetIDs = w.publisher.FilterOnline(allTargetIDs)
+	} else {
+		offlineTargetIDs = allTargetIDs
+	}
+
+	for _, uid := range offlineTargetIDs {
+		if pushErr := w.pushSvc.PushChatMessage(ctx, uid, pushMsg.ConversationID, &sender, conv.Type, convName, message, groupID, groupAvatar); pushErr != nil {
+			zap.L().Debug("push worker: JPush chat message failed",
+				zap.String("userID", uid),
+				zap.String("conversationID", pushMsg.ConversationID),
+				zap.Error(pushErr),
+			)
+		}
+	}
+
+	return nil
+}
+
+// PublishToPush 将消息发布到推送 Stream
+func PublishToPush(ctx context.Context, rdb *redis.Client, stream string, msg *PushStreamMsg) error {
+	data, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("marshal push message: %w", err)
+	}
+
+	return rdb.XAdd(ctx, &redis.XAddArgs{
+		Stream: stream,
+		MaxLen: 100000,
+		Approx: true,
+		ID:     "*",
+		Values: map[string]any{
+			"data": string(data),
+		},
+	}).Err()
+}