backend/internal/service/push_worker.go

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

	// 直接对所有参与者推送（不按 WS 在线过滤）：
	// 多设备场景下某端在线不代表所有端在线，由 PushChatMessage 推送到用户所有移动设备
	for _, uid := range allTargetIDs {
		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()
}