yumi-golang/internal/service/voiceroomrocket/lifecycle.go

package voiceroomrocket

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"strings"
	"time"

	rmq "github.com/apache/rocketmq-clients/golang/v5"
	"github.com/apache/rocketmq-clients/golang/v5/credentials"
)

// Start 启动语音房火箭完整后台任务：MQ 消费、在房奖励延迟扫描、奖励自动发放。
func (s *Service) Start(ctx context.Context) error {
	if err := s.StartMessageConsumer(ctx); err != nil {
		return err
	}
	s.StartLaunchWorker(ctx)
	s.StartInRoomRewardWorker(ctx)
	s.StartRewardGrantWorker(ctx)
	return nil
}

// StartMessageConsumer 启动语音房火箭送礼消息消费。
func (s *Service) StartMessageConsumer(ctx context.Context) error {
	if !s.cfg.VoiceRoomRocket.MQ.Enabled {
		log.Printf("voice room rocket rocketmq consumer disabled")
		return nil
	}
	if strings.TrimSpace(s.cfg.VoiceRoomRocket.MQ.Endpoint) == "" ||
		strings.TrimSpace(s.cfg.VoiceRoomRocket.MQ.AccessKey) == "" ||
		strings.TrimSpace(s.cfg.VoiceRoomRocket.MQ.AccessSecret) == "" {
		log.Printf("voice room rocket rocketmq consumer skipped: endpoint or credentials missing")
		return nil
	}
	go s.startRocketMQConsumerWithRetry(ctx)
	return nil
}

// StartLaunchWorker 启动满进度后延迟发射扫描。
func (s *Service) StartLaunchWorker(ctx context.Context) {
	interval := time.Duration(s.cfg.VoiceRoomRocket.InRoomRewardScanIntervalSeconds) * time.Second
	if interval <= 0 {
		interval = defaultWorkerInterval
	}
	batchSize := int64(s.cfg.VoiceRoomRocket.InRoomRewardBatchSize)
	if batchSize <= 0 {
		batchSize = defaultWorkerBatchSize
	}
	log.Printf("voice room rocket launch worker started. interval=%s batch=%d delay=%ds", interval, batchSize, defaultLaunchDelaySeconds)
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			if _, err := s.ProcessDueLaunches(ctx, time.Now(), batchSize); err != nil && ctx.Err() == nil {
				log.Printf("voice room rocket launch scan failed: %v", err)
			}
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
			}
		}
	}()
}

// StartInRoomRewardWorker 启动在房奖励到期扫描。
func (s *Service) StartInRoomRewardWorker(ctx context.Context) {
	if s.repo.Redis == nil {
		log.Printf("voice room rocket in-room reward worker skipped: redis is nil")
		return
	}
	interval := time.Duration(s.cfg.VoiceRoomRocket.InRoomRewardScanIntervalSeconds) * time.Second
	if interval <= 0 {
		interval = defaultWorkerInterval
	}
	batchSize := int64(s.cfg.VoiceRoomRocket.InRoomRewardBatchSize)
	if batchSize <= 0 {
		batchSize = defaultWorkerBatchSize
	}
	log.Printf("voice room rocket in-room reward worker started. interval=%s batch=%d", interval, batchSize)
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			if _, err := s.ProcessDueInRoomRewards(ctx, time.Now(), batchSize); err != nil && ctx.Err() == nil {
				log.Printf("voice room rocket in-room reward scan failed: %v", err)
			}
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
			}
		}
	}()
}

// StartRewardGrantWorker 启动奖励自动发放扫描。
func (s *Service) StartRewardGrantWorker(ctx context.Context) {
	interval := time.Duration(s.cfg.VoiceRoomRocket.RewardGrantScanIntervalSeconds) * time.Second
	if interval <= 0 {
		interval = defaultWorkerInterval
	}
	batchSize := s.cfg.VoiceRoomRocket.RewardGrantBatchSize
	if batchSize <= 0 {
		batchSize = defaultWorkerBatchSize
	}
	log.Printf("voice room rocket reward grant worker started. interval=%s batch=%d", interval, batchSize)
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			if _, err := s.ProcessPendingRewards(ctx, batchSize); err != nil && ctx.Err() == nil {
				log.Printf("voice room rocket reward grant scan failed: %v", err)
			}
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
			}
		}
	}()
}

func (s *Service) startRocketMQConsumerWithRetry(ctx context.Context) {
	retryDelay := 5 * time.Second
	for {
		if ctx.Err() != nil {
			return
		}
		if err := s.startRocketMQConsumer(ctx); err != nil {
			log.Printf("start voice room rocket rocketmq consumer failed: %v; retry in %s", err, retryDelay)
			timer := time.NewTimer(retryDelay)
			select {
			case <-ctx.Done():
				timer.Stop()
				return
			case <-timer.C:
			}
			if retryDelay < time.Minute {
				retryDelay *= 2
				if retryDelay > time.Minute {
					retryDelay = time.Minute
				}
			}
			continue
		}
		return
	}
}

func (s *Service) startRocketMQConsumer(ctx context.Context) error {
	filter := rmq.SUB_ALL
	if tag := strings.TrimSpace(s.cfg.VoiceRoomRocket.MQ.Tag); tag != "" {
		filter = rmq.NewFilterExpression(tag)
	}
	consumer, err := rmq.NewPushConsumer(&rmq.Config{
		Endpoint:      s.cfg.VoiceRoomRocket.MQ.Endpoint,
		NameSpace:     s.cfg.VoiceRoomRocket.MQ.Namespace,
		ConsumerGroup: s.cfg.VoiceRoomRocket.MQ.ConsumerGroup,
		Credentials: &credentials.SessionCredentials{
			AccessKey:     s.cfg.VoiceRoomRocket.MQ.AccessKey,
			AccessSecret:  s.cfg.VoiceRoomRocket.MQ.AccessSecret,
			SecurityToken: s.cfg.VoiceRoomRocket.MQ.SecurityToken,
		},
	},
		rmq.WithPushSubscriptionExpressions(map[string]*rmq.FilterExpression{
			s.cfg.VoiceRoomRocket.MQ.Topic: filter,
		}),
		rmq.WithPushConsumptionThreadCount(8),
		rmq.WithPushMaxCacheMessageCount(256),
		rmq.WithPushMessageListener(&rmq.FuncMessageListener{
			Consume: func(messageView *rmq.MessageView) rmq.ConsumerResult {
				if messageView == nil {
					return rmq.SUCCESS
				}
				if err := s.processRocketMQMessage(context.Background(), messageView); err != nil {
					log.Printf("voice room rocket mq process failed. messageId=%s err=%v", messageView.GetMessageId(), err)
					return rmq.FAILURE
				}
				return rmq.SUCCESS
			},
		}),
	)
	if err != nil {
		return err
	}
	if err := consumer.Start(); err != nil {
		return err
	}
	s.rocketConsumer = consumer
	log.Printf("voice room rocket rocketmq consumer started. topic=%s group=%s tag=%s",
		s.cfg.VoiceRoomRocket.MQ.Topic,
		s.cfg.VoiceRoomRocket.MQ.ConsumerGroup,
		s.cfg.VoiceRoomRocket.MQ.Tag,
	)
	go func() {
		<-ctx.Done()
		if s.rocketConsumer != nil {
			_ = s.rocketConsumer.GracefulStop()
		}
	}()
	return nil
}

func (s *Service) processRocketMQMessage(ctx context.Context, messageView *rmq.MessageView) error {
	payload := strings.TrimSpace(string(messageView.GetBody()))
	if payload == "" {
		return nil
	}
	if _, err := s.ProcessGiftPayload(ctx, payload); err != nil {
		var syntaxErr *json.SyntaxError
		var typeErr *json.UnmarshalTypeError
		if errors.As(err, &syntaxErr) || errors.As(err, &typeErr) {
			log.Printf("drop malformed voice room rocket message. messageId=%s err=%v", messageView.GetMessageId(), err)
			return nil
		}
		return err
	}
	return nil
}