package sqsmeasurehandler

import (
	"encoding/json"
	"fmt"
	"github.com/pingcap/dbaas/cost-metering-service/config"
	"github.com/pingcap/dbaas/cost-metering-service/forms"
	"github.com/pingcap/dbaas/cost-metering-service/log"
	"github.com/pingcap/dbaas/cost-metering-service/model/billing"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/sqs"
	"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
}

func (s *SqsHandlerImpl) RunSqsMeasure() (err error) {
	// create AWS SQS Session
	sess, err := session.NewSession(aws.NewConfig().WithRegion(config.Get().GetString(config.CostQueueRegin)))
	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.CostQueueUrl)
	if isDl {
		queueURL = config.Get().GetString(config.CostDlQueueUrl)
	}
	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
		}
		processMessageBatch(sqsClient, queueURL, result.Messages, queueName)
	}
}

func processMessageBatch(sqsClient *sqs.SQS, queueURL string, messages []*sqs.Message, queueName string) {
	var deleteData []*sqs.DeleteMessageBatchRequestEntry
	measure := &billing.MeasureModel{}
	for _, msg := range messages {

		//fmt.Printf("Received message: %s\n", *msg.Body)
		measureCost := &forms.UnitCostsInput{}
		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))
			deleteData = append(deleteData, &sqs.DeleteMessageBatchRequestEntry{
				Id:            msg.MessageId,
				ReceiptHandle: msg.ReceiptHandle,
			})
			continue

		}
		_, err = measure.ProcessMeasure(&forms.UnitCostsInput{
			CostID:   measureCost.CostID,
			CostTime: measureCost.CostTime,
			Unit:     measureCost.Unit,
			Cost:     measureCost.Cost,
			Labels:   measureCost.Labels,
		})
		if err != nil {
			log.Error("Failed to process message: ", zap.Error(err), zap.String("MessageId", *msg.MessageId), zap.String("cost_id", measureCost.CostID))
			continue
		}
		deleteData = append(deleteData, &sqs.DeleteMessageBatchRequestEntry{
			Id:            msg.MessageId,
			ReceiptHandle: msg.ReceiptHandle,
		})
	}
	if len(deleteData) > 0 {
		deleteMessageBatch(sqsClient, queueURL, deleteData, queueName)
	}
}

func deleteMessageBatch(sqsClient *sqs.SQS, queueURL string, deleteEntries []*sqs.DeleteMessageBatchRequestEntry, queueName string) {
	_, err := sqsClient.DeleteMessageBatch(&sqs.DeleteMessageBatchInput{
		QueueUrl: aws.String(queueURL),
		Entries:  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.CostDlQueueName)
	} else {
		queueName += config.Get().GetString(config.CostQueueName)
	}
	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
	}
}