package sqsmeasurehandler

import (
	"encoding/json"
	"fmt"
	"time"

	"github.com/pingcap/dbaas/billing2/handler/cachehandler"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/sqs"
	"github.com/pingcap/dbaas/billing2/config"
	"github.com/pingcap/dbaas/billing2/forms"
	"github.com/pingcap/dbaas/billing2/model/billing"
	"github.com/pingcap/dbaas/billing2/utils/log"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"go.uber.org/zap"
)

type SqsMeasureHandler interface {
	// start run handler
	RunSqsMeasure() error
}
type MeasureModel interface {
	CostUnits(tenantID uint, costs *forms.UnitCosts) (*forms.UnitCosts, error)
}
type SqsHandlerImpl struct {
	sqsClient sqs.SQS
	measure   MeasureModel
}

const (
	SqsDlReceiveSleepSecondsDefault = 1200
	SqsReceiveSleepSecondsDefault   = 600
)

var global SqsMeasureHandler = &SqsHandlerImpl{}

func Get() SqsMeasureHandler {
	return global
}

var (
	SqsMeasureCount = promauto.NewCounterVec(prometheus.CounterOpts{
		Namespace: "dbaas",
		Subsystem: "billing",
		Name:      "sqs_measure_count",
	}, []string{"action"})
	SqsMeasureDLCount = promauto.NewCounterVec(prometheus.CounterOpts{
		Namespace: "dbaas",
		Subsystem: "billing",
		Name:      "sqs_measure_dl_messages",
	}, []string{"action"})
)

func (s *SqsHandlerImpl) RunSqsMeasure() (err error) {
	// create AWS SQS Session
	sess, err := session.NewSession(aws.NewConfig().WithRegion(config.Get().GetString(config.MeasureQueueRegin)))
	if err != nil {
		return fmt.Errorf("failed to create AWS session: %v", err)
	}
	// create SQS measure Client
	sqsClient := sqs.New(sess)

	//  run SQS measure process
	go processMessages(sqsClient, false)
	go processMessages(sqsClient, true)

	return nil
}
func processMessages(sqsClient *sqs.SQS, isDl bool) {
	queueURL := config.Get().GetString(config.MeasureQueueUrl)
	if isDl {
		queueURL = config.Get().GetString(config.MeasureDlQueueUrl)
	}
	queueName := getQueueName(sqsClient.ServiceName, isDl)

	for {
		if !config.Get().GetBool(config.EnableQueueMeasure) {
			log.Info("SQSMeasure is disabled, waiting to be enabled")
			time.Sleep(time.Second * 60)
			continue
		}
		result, err := sqsClient.ReceiveMessage(&sqs.ReceiveMessageInput{
			QueueUrl:            aws.String(queueURL),
			MaxNumberOfMessages: aws.Int64(10),
			WaitTimeSeconds:     aws.Int64(20),
		})
		if err != nil {
			log.Error("Failed to receive SQS messages: ", zap.Error(err), zap.String("queueName", queueName), zap.String("queueURL", queueURL))
			time.Sleep(time.Second * 5)
			continue
		}
		if len(result.Messages) < 1 {
			log.Info("Received 0 messages from SQS", zap.String("queueName", queueName))
			sleepDuration := getSleepDuration(isDl)
			time.Sleep(time.Second * sleepDuration)
			continue
		}
		if isDl {
			SqsMeasureDLCount.WithLabelValues("Receive").Add(float64(len(result.Messages)))

		} else {
			SqsMeasureCount.WithLabelValues("Receive").Add(float64(len(result.Messages)))
		}
		processMessageBatch(sqsClient, queueURL, result.Messages, isDl)
	}
}

func processMessageBatch(sqsClient *sqs.SQS, queueURL string, messages []*sqs.Message, isDl bool) {
	var deleteData []*sqs.DeleteMessageBatchRequestEntry
	measure := &billing.MeasureModel{}
	cache := cachehandler.Get()
	funcName := "processMessageBatchLogError"

	for _, msg := range messages {

		//fmt.Printf("Received message: %s\n", *msg.Body)
		measureCost := &forms.UnitCostsSqs{}
		err := json.Unmarshal([]byte(*msg.Body), measureCost)
		if err != nil {
			log.Error("Failed to unmarshal message body: ", zap.Error(err), zap.String("queueURL", queueURL), zap.String("MessageId", *msg.MessageId), zap.String("Msg.Body", *msg.Body))
			deleteData = append(deleteData, &sqs.DeleteMessageBatchRequestEntry{
				Id:            msg.MessageId,
				ReceiptHandle: msg.ReceiptHandle,
			})
			continue

		}
		_, err = measure.CostUnits(uint(measureCost.TenantID), &forms.UnitCosts{
			CostID:        measureCost.CostID,
			CostTimeStamp: measureCost.CostTimeStamp,
			Unit:          measureCost.Unit,
			Cost:          measureCost.Cost,
			Labels:        measureCost.Labels,
		})
		if err != nil {
			var result bool
			found := cache.Get(funcName, &result, measureCost.Unit)
			if found {
				log.Info("Failed to process message: ", zap.Error(err), zap.String("MessageId", *msg.MessageId), zap.String("cost_id", measureCost.CostID))

			} else {
				log.Error("Failed to process message: ", zap.Error(err), zap.String("MessageId", *msg.MessageId), zap.String("cost_id", measureCost.CostID))
				cache.SetWithTimeOut(time.Hour, funcName, true, measureCost.Unit)
			}
			continue
		}
		deleteData = append(deleteData, &sqs.DeleteMessageBatchRequestEntry{
			Id:            msg.MessageId,
			ReceiptHandle: msg.ReceiptHandle,
		})
	}

	if len(deleteData) > 0 {
		deleteMessageBatch(sqsClient, queueURL, deleteData, isDl)
	}
}

func deleteMessageBatch(sqsClient *sqs.SQS, queueURL string, deleteEntries []*sqs.DeleteMessageBatchRequestEntry, isDl bool) {
	_, err := sqsClient.DeleteMessageBatch(&sqs.DeleteMessageBatchInput{
		QueueUrl: aws.String(queueURL),
		Entries:  deleteEntries,
	})
	if isDl {
		SqsMeasureDLCount.WithLabelValues("Delete").Add(float64(len(deleteEntries)))

	} else {
		SqsMeasureCount.WithLabelValues("Delete").Add(float64(len(deleteEntries)))

	}
	if err != nil {
		log.Error("Failed to delete SQS messages: ", zap.Error(err), zap.String("queueURL", queueURL), zap.String("deleteEntries", *deleteEntries[0].Id))

		// retry
		for _, entry := range deleteEntries {
			err := retryDeleteMessage(sqsClient, queueURL, entry)
			if err != nil {
				log.Error("Failed to delete SQS message after retry: ", zap.Error(err), zap.String("queueURL", queueURL), zap.String("messageId", *entry.Id))
			}
		}
	}
}

func retryDeleteMessage(sqsClient *sqs.SQS, queueURL string, entry *sqs.DeleteMessageBatchRequestEntry) error {
	maxRetries := 3
	for retry := 1; retry <= maxRetries; retry++ {
		_, err := sqsClient.DeleteMessageBatch(&sqs.DeleteMessageBatchInput{
			QueueUrl: aws.String(queueURL),
			Entries:  []*sqs.DeleteMessageBatchRequestEntry{entry},
		})
		if err == nil {
			return nil
		}
		log.Warn("Failed to delete SQS message, retrying...", zap.Error(err), zap.String("queueURL", queueURL), zap.String("messageId", *entry.Id), zap.Int("retry", retry), zap.Int("maxRetries", maxRetries))
		backoffDuration := time.Duration(retry) * time.Second
		time.Sleep(backoffDuration)
	}
	return fmt.Errorf("failed to delete SQS message after %d retries", maxRetries)
}

func getQueueName(serviceName string, isDl bool) string {
	queueName := serviceName + "-"
	if isDl {
		queueName += config.Get().GetString(config.MeasureDlQueueName)
	} else {
		queueName += config.Get().GetString(config.MeasureQueueName)
	}
	return queueName
}

func getSleepDuration(isDl bool) time.Duration {
	if isDl {
		sleepDuration := config.Get().GetDuration(config.SqsDlReceiveSleepSeconds)
		if sleepDuration == 0 {
			return SqsDlReceiveSleepSecondsDefault
		}
		return sleepDuration
	} else {
		sleepDuration := config.Get().GetDuration(config.SqsReceiveSleepSeconds)
		if sleepDuration == 0 {
			return SqsReceiveSleepSecondsDefault
		}
		return sleepDuration
	}
}