yumi-golang/internal/service/registerreward/consumer.go

package registerreward

import (
	"context"
	"errors"
	"log"
	"strings"
	"time"

	"github.com/redis/go-redis/v9"
)

// Start 启动注册奖励消费任务；若未配置 Stream，则直接跳过。
func (s *RegisterRewardService) Start(ctx context.Context) error {
	if strings.TrimSpace(s.cfg.Worker.RegisterReward.StreamKey) == "" {
		log.Printf("register reward consumer disabled: stream key is blank")
		return nil
	}
	if err := s.ensureConsumerGroup(ctx); err != nil {
		return err
	}
	log.Printf(
		"register reward consumer started. stream=%s group=%s consumer=%s",
		s.cfg.Worker.RegisterReward.StreamKey,
		s.cfg.Worker.RegisterReward.ConsumerGroup,
		s.cfg.Worker.RegisterReward.ConsumerName,
	)
	go s.consumeLoop(ctx)
	return nil
}

// consumeLoop 先认领超时消息，再持续读取新消息。
func (s *RegisterRewardService) consumeLoop(ctx context.Context) {
	for {
		if ctx.Err() != nil {
			return
		}

		reclaimed, err := s.reclaimPending(ctx)
		if err != nil {
			s.handleLoopError(ctx, "reclaim register reward messages", err)
			continue
		}
		if reclaimed > 0 {
			continue
		}

		err = s.readNewMessages(ctx)
		if err == nil || errors.Is(err, redis.Nil) {
			continue
		}
		s.handleLoopError(ctx, "read register reward messages", err)
	}
}

// handleLoopError 统一处理消费循环中的错误，并在必要时重建消费者组。
func (s *RegisterRewardService) handleLoopError(ctx context.Context, action string, err error) {
	if ctx.Err() != nil {
		return
	}
	if isRegisterRewardNoGroupErr(err) {
		if ensureErr := s.ensureConsumerGroup(ctx); ensureErr != nil {
			log.Printf("%s failed and recreate consumer group failed: %v", action, ensureErr)
			sleepWithContext(ctx, time.Second)
			return
		}
		return
	}
	log.Printf("%s failed: %v", action, err)
	sleepWithContext(ctx, time.Second)
}

// ensureConsumerGroup 确保 Redis Stream 消费组存在。
func (s *RegisterRewardService) ensureConsumerGroup(ctx context.Context) error {
	err := s.repo.Redis.XGroupCreateMkStream(
		ctx,
		s.cfg.Worker.RegisterReward.StreamKey,
		s.cfg.Worker.RegisterReward.ConsumerGroup,
		"0",
	).Err()
	if err != nil && !strings.Contains(strings.ToUpper(err.Error()), "BUSYGROUP") {
		return err
	}
	return nil
}

// reclaimPending 认领长时间未确认的 pending 消息，避免消息堆积。
func (s *RegisterRewardService) reclaimPending(ctx context.Context) (int, error) {
	messages, _, err := s.repo.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{
		Stream:   s.cfg.Worker.RegisterReward.StreamKey,
		Group:    s.cfg.Worker.RegisterReward.ConsumerGroup,
		Consumer: s.cfg.Worker.RegisterReward.ConsumerName,
		MinIdle:  s.cfg.Worker.RegisterReward.MinIdle,
		Start:    "0-0",
		Count:    s.cfg.Worker.RegisterReward.BatchSize,
	}).Result()
	if err != nil {
		return 0, err
	}
	if len(messages) == 0 {
		return 0, nil
	}
	return len(messages), s.processMessages(ctx, messages)
}

// readNewMessages 按消费者组协议读取新消息。
func (s *RegisterRewardService) readNewMessages(ctx context.Context) error {
	streams, err := s.repo.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group:    s.cfg.Worker.RegisterReward.ConsumerGroup,
		Consumer: s.cfg.Worker.RegisterReward.ConsumerName,
		Streams:  []string{s.cfg.Worker.RegisterReward.StreamKey, ">"},
		Count:    s.cfg.Worker.RegisterReward.BatchSize,
		Block:    s.cfg.Worker.RegisterReward.Block,
	}).Result()
	if err != nil {
		return err
	}
	for _, stream := range streams {
		if err := s.processMessages(ctx, stream.Messages); err != nil {
			return err
		}
	}
	return nil
}

// processMessages 顺序处理消息，并在可确认时回 ACK。
func (s *RegisterRewardService) processMessages(ctx context.Context, messages []redis.XMessage) error {
	for _, message := range messages {
		ack, err := s.processMessage(ctx, message)
		if ack {
			if ackErr := s.repo.Redis.XAck(
				context.Background(),
				s.cfg.Worker.RegisterReward.StreamKey,
				s.cfg.Worker.RegisterReward.ConsumerGroup,
				message.ID,
			).Err(); ackErr != nil {
				log.Printf("ack register reward message failed. streamId=%s err=%v", message.ID, ackErr)
			}
		}
		if err != nil {
			return err
		}
	}
	return nil
}

// processMessage 处理单条注册奖励消息，流程包含解析、幂等、解析配置、发奖与状态更新。
func (s *RegisterRewardService) processMessage(ctx context.Context, message redis.XMessage) (bool, error) {
	event, err := parseRegisterRewardEvent(message)
	if err != nil {
		log.Printf("drop malformed register reward message. streamId=%s err=%v values=%v", message.ID, err, message.Values)
		return true, nil
	}

	// 先创建或读取日志，保证失败重试时仍能延续同一条处理记录。
	logRecord, terminal, err := s.getOrCreateLog(ctx, event, message.ID)
	if err != nil {
		return false, err
	}
	if terminal {
		return true, nil
	}

	resolvedConfig, ack, err := s.resolveRewardConfig(ctx, logRecord, event)
	if err != nil || ack {
		return ack, err
	}

	grantPlan, err := s.resolveGrantPlan(ctx, logRecord, event, resolvedConfig)
	if err != nil {
		return false, err
	}
	if grantPlan.SkipDispatch {
		if err := s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusSuccess, "", resolvedConfig); err != nil {
			return false, err
		}
		return true, nil
	}

	// 金币和道具奖励分开发放，任何一步失败都保留失败状态供后续重试。
	if err := s.dispatchGold(ctx, event, resolvedConfig); err != nil {
		_ = s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusFailed, err.Error(), resolvedConfig)
		return false, err
	}
	if err := s.dispatchProps(ctx, logRecord.ID, event, resolvedConfig); err != nil {
		_ = s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusFailed, err.Error(), resolvedConfig)
		return false, err
	}
	if err := s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusSuccess, "", resolvedConfig); err != nil {
		return false, err
	}
	if err := s.pushRewardArrivalNotice(ctx, event, logRecord, resolvedConfig); err != nil {
		log.Printf(
			"push register reward notice failed. logId=%d eventId=%s userId=%d err=%v",
			logRecord.ID,
			event.EventID,
			event.UserID,
			err,
		)
	}
	return true, nil
}