package cprelationbroadcast

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

	"chatapp3-golang/internal/common"

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

// StartMessageConsumer 启动 CP 关系成功 MQ 消费链路。
func (s *Service) StartMessageConsumer(ctx context.Context) error {
	if !s.cfg.CPRelationBroadcast.MQ.Enabled {
		log.Printf("cp relation broadcast rocketmq consumer disabled")
		return nil
	}
	if strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.Endpoint) == "" ||
		strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.AccessKey) == "" ||
		strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.AccessSecret) == "" ||
		strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.Topic) == "" ||
		strings.TrimSpace(s.cfg.CPRelationBroadcast.MQ.ConsumerGroup) == "" {
		log.Printf("cp relation broadcast rocketmq consumer skipped: endpoint, credentials, topic or consumer group missing")
		return nil
	}

	go s.startRocketMQConsumerWithRetry(ctx)
	return nil
}

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 cp relation broadcast 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.CPRelationBroadcast.MQ.Tag); tag != "" && tag != "*" {
		filter = rmq.NewFilterExpression(tag)
	}

	consumer, err := rmq.NewPushConsumer(&rmq.Config{
		Endpoint:      s.cfg.CPRelationBroadcast.MQ.Endpoint,
		NameSpace:     s.cfg.CPRelationBroadcast.MQ.Namespace,
		ConsumerGroup: s.cfg.CPRelationBroadcast.MQ.ConsumerGroup,
		Credentials: &credentials.SessionCredentials{
			AccessKey:     s.cfg.CPRelationBroadcast.MQ.AccessKey,
			AccessSecret:  s.cfg.CPRelationBroadcast.MQ.AccessSecret,
			SecurityToken: s.cfg.CPRelationBroadcast.MQ.SecurityToken,
		},
	},
		rmq.WithPushSubscriptionExpressions(map[string]*rmq.FilterExpression{
			s.cfg.CPRelationBroadcast.MQ.Topic: filter,
		}),
		rmq.WithPushConsumptionThreadCount(4),
		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("cp relation broadcast 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("cp relation broadcast rocketmq consumer started. topic=%s group=%s tag=%s",
		s.cfg.CPRelationBroadcast.MQ.Topic,
		s.cfg.CPRelationBroadcast.MQ.ConsumerGroup,
		s.cfg.CPRelationBroadcast.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.ProcessPayload(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 cp relation broadcast message. messageId=%s err=%v", messageView.GetMessageId(), err)
			return nil
		}
		var appErr *common.AppError
		if errors.As(err, &appErr) && appErr.Status >= http.StatusBadRequest && appErr.Status < http.StatusInternalServerError {
			log.Printf("drop invalid cp relation broadcast message. messageId=%s code=%s err=%v", messageView.GetMessageId(), appErr.Code, err)
			return nil
		}
		return err
	}
	return nil
}