package report

import (
	"crypto/sha256"
	"fmt"

	"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/log"
	"go.uber.org/zap"

	"github.com/tidbcloud/billing-reporter/conf"
	"github.com/tidbcloud/billing-reporter/metrics"
)

type SqsMessage struct {
	Message    string            `json:"message"`
	Attributes map[string]string `json:"attributes"`
	Region     string            `json:"region"`
	Url        string            `json:"url"`
	Partition  string            `json:"partition"`
}

type SqsManager struct {
	sqsInstance *sqs.SQS
}

func NewSqsManager() *SqsManager {
	sess := session.Must(session.NewSessionWithOptions(session.Options{
		Config: aws.Config{
			Region: &conf.AwsRegion,
		},
		SharedConfigState: session.SharedConfigEnable,
	}))
	return &SqsManager{
		sqsInstance: sqs.New(sess),
	}
}

// TODO: send the message to aws SQS
func (sm *SqsManager) SendSqsMessage(sqsMsg SqsMessage) error {
	metrics.SQSMessage.WithLabelValues(metrics.SQSTotal).Inc()
	log.Info("start to send", zap.String("partition", sqsMsg.Partition))
	setMsg(&sqsMsg)

	// resending messages won't block the sending current message process.
	if err := sm.resendingFailureMessages(); err != nil {
		metrics.SQSMessage.WithLabelValues(metrics.QSSResendErr).Inc()
		log.Error("resending failure messages failed", zap.Error(err))
	}

	if err := sm.sendSqsMessage(sqsMsg); err != nil {
		metrics.SQSMessage.WithLabelValues(metrics.SQSFailed).Inc()
		// currently, If there existed error during sending messages, we persist the message into local file
		return writeSqsMessageToFile(sqsMsg)
	}
	metrics.SQSMessage.WithLabelValues(metrics.SQSSuccess).Inc()
	log.Info("send message done", zap.String("partition", sqsMsg.Partition))
	return nil
}

func (sm *SqsManager) sendSqsMessage(sqsMsg SqsMessage) error {
	_, err := sm.sqsInstance.SendMessage(&sqs.SendMessageInput{
		MessageBody:            &sqsMsg.Message,
		QueueUrl:               &conf.AwsSqsUrl,
		MessageGroupId:         aws.String(sqsMsg.Partition),
		MessageDeduplicationId: aws.String(fmt.Sprintf("%x", sha256.Sum256([]byte(sqsMsg.Message)))),
		MessageAttributes:      buildSqsMsgAttributes(sqsMsg.Attributes),
	})
	if err != nil {
		log.Error("failed to send sqs", zap.Error(err))
	}
	return err
}

// check whether there exist failure message remaining to be resent.
// If existed, resending the failure message. Not matter resending is success or not,
// it wouldn't block the the current sending message process.
func (sm *SqsManager) resendingFailureMessages() error {
	existed, err := isFileExisted(conf.FailureMessageFile)
	if err != nil {
		return err
	}
	if !existed {
		return nil
	}
	sqsMsgs, err := readFailureMessages(conf.FailureMessageFile)
	if err != nil {
		return err
	}
	if sqsMsgs == nil || len(sqsMsgs) < 1 {
		return nil
	}
	var rewriteMsgs []SqsMessage
	if len(sqsMsgs) > failureMessagesBatchLimit {
		rewriteMsgs = sqsMsgs[failureMessagesBatchLimit:]
		sqsMsgs = sqsMsgs[:failureMessagesBatchLimit]
	}
	bi := buildSqsBatchRequest(sqsMsgs)
	_, err = sm.sqsInstance.SendMessageBatch(bi)
	if err != nil {
		return err
	}
	metrics.SQSMessage.WithLabelValues(metrics.SQSResendSuccess).Add(float64(len(sqsMsgs)))
	// when resending success, we rewrite the remaining failure messages to file
	// if there is no remaining messages, the file would be removed.
	return rewriteMessagesToFile(rewriteMsgs)
}