package weekstar

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

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

// StartMessageConsumer 启动送礼消息消费链路。
func (s *WeekStarService) StartMessageConsumer(ctx context.Context) error {
	return s.startRocketMQConsumer(ctx)
}

// startRocketMQConsumer 启动 RocketMQ 推送消费者，订阅送礼事件。
func (s *WeekStarService) startRocketMQConsumer(ctx context.Context) error {
	if !s.cfg.WeekStar.RocketMQ.Enabled {
		log.Printf("week star rocketmq consumer disabled")
		return nil
	}
	if strings.TrimSpace(s.cfg.WeekStar.RocketMQ.Endpoint) == "" ||
		strings.TrimSpace(s.cfg.WeekStar.RocketMQ.AccessKey) == "" ||
		strings.TrimSpace(s.cfg.WeekStar.RocketMQ.AccessSecret) == "" {
		log.Printf("week star rocketmq consumer skipped: endpoint or credentials missing")
		return nil
	}

	filter := rmq.SUB_ALL
	if tag := strings.TrimSpace(s.cfg.WeekStar.RocketMQ.Tag); tag != "" {
		filter = rmq.NewFilterExpression(tag)
	}

	consumer, err := rmq.NewPushConsumer(&rmq.Config{
		Endpoint:      s.cfg.WeekStar.RocketMQ.Endpoint,
		NameSpace:     s.cfg.WeekStar.RocketMQ.Namespace,
		ConsumerGroup: s.cfg.WeekStar.RocketMQ.ConsumerGroup,
		Credentials: &credentials.SessionCredentials{
			AccessKey:     s.cfg.WeekStar.RocketMQ.AccessKey,
			AccessSecret:  s.cfg.WeekStar.RocketMQ.AccessSecret,
			SecurityToken: s.cfg.WeekStar.RocketMQ.SecurityToken,
		},
	},
		rmq.WithPushSubscriptionExpressions(map[string]*rmq.FilterExpression{
			s.cfg.WeekStar.RocketMQ.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
				}
				// 单条消息处理失败时返回 FAILURE，让 RocketMQ 触发重试。
				if err := s.processRocketMQMessage(context.Background(), messageView); err != nil {
					log.Printf("week star 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("week star rocketmq consumer started. topic=%s group=%s tag=%s",
		s.cfg.WeekStar.RocketMQ.Topic,
		s.cfg.WeekStar.RocketMQ.ConsumerGroup,
		s.cfg.WeekStar.RocketMQ.Tag,
	)
	go func() {
		<-ctx.Done()
		if s.rocketConsumer != nil {
			_ = s.rocketConsumer.GracefulStop()
		}
	}()
	return nil
}

// processRocketMQMessage 解析 RocketMQ 消息体并交给业务逻辑处理。
func (s *WeekStarService) 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 week star message. messageId=%s err=%v", messageView.GetMessageId(), err)
			return nil
		}
		return err
	}
	return nil
}