package handler

import (
	"context"
	"encoding/json"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/marketplacemetering"
	"github.com/aws/aws-sdk-go/service/sqs"
	"github.com/gin-gonic/gin"
	"github.com/pingcap/dbaas/central/config"
	"github.com/pingcap/dbaas/central/handler/billing2"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

type awsAccountProviderImpl struct {
}

func GetAWSDefaultSession() (*session.Session, error) {
	sess, err := session.NewSession()
	return sess, errors.Trace(err)
}

func getAWSMeteringClient() (*marketplacemetering.MarketplaceMetering, error) {
	sess, err := GetAWSDefaultSession()
	if err != nil {
		return nil, err
	}
	role := config.Global().MarketPlaceConfig.AWSRole
	creds := stscreds.NewCredentials(sess, role)
	sess = sess.Copy(&aws.Config{Credentials: creds})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return marketplacemetering.New(sess), nil
}

func (provider *awsAccountProviderImpl) ResolveToken(token *string) (*string, error) {
	marketClient, err := getAWSMeteringClient()

	if err != nil {
		return nil, err
	}
	output, err := marketClient.ResolveCustomer(&marketplacemetering.ResolveCustomerInput{RegistrationToken: token})
	if err != nil {
		return nil, err
	}

	productCode := config.Global().MarketPlaceConfig.AWSProductCode

	if (output.ProductCode == nil) || (*output.ProductCode != productCode) {
		return nil, errors.Errorf("Wrong product code %v", output.ProductCode)
	}

	return output.CustomerIdentifier, nil
}

const AWSTokenKey = "x-amzn-marketplace-token"

func (provider *awsAccountProviderImpl) GetTokenFromRedirect(c *gin.Context) (*string, error) {
	token := c.Request.PostFormValue(AWSTokenKey)
	if token == "" {
		return nil, errors.New("Invalid AWS Token")
	}
	return &token, nil
}

func getAWSSQSClient() (*sqs.SQS, error) {
	sess, err := GetAWSDefaultSession()
	if err != nil {
		return nil, err
	}

	return sqs.New(sess), nil
}

const QueueTimeout int64 = 30

func processOneAWSSubscribeChange(ctx context.Context, msgBody *string, processor ProcessTenantSubscribeChangeFunc) (bool, error) {
	m := make(map[string]interface{})
	err := json.Unmarshal([]byte(*msgBody), &m)
	if err != nil {
		return false, errors.Errorf("Invalid marketplace sqs msg body. %v", *msgBody)
	}
	action := m["action"].(string)
	change := SubscribeChange{}
	switch action {
	case "subscribe-success":
		change.SubscribeStatus = SubscribeStatusSubscribed
	case "unsubscribe-pendihng":
		change.SubscribeStatus = SubscribeStatusUnsubscribed
	default:
		return true, nil
	}
	change.AccountProvider = AWSAccountProvider
	change.AccountCode = m["customer-identifier"].(string)
	return processor(ctx, change)
}

func (provider *awsAccountProviderImpl) ProcessTenantSubscribeChange(ctx context.Context, processor ProcessTenantSubscribeChangeFunc) error {
	if !config.Global().MarketPlaceConfig.EnableAwsBackground {
		log.Info("Skip process aws tenant subscribe by config")
		return nil
	}
	sqsClient, err := getAWSSQSClient()
	if err != nil {
		return err
	}
	queueURL := config.Global().MarketPlaceConfig.AWSSQS
	timeout := QueueTimeout
	allErrors := make([]error, 0)
	for {
		msgResult, err := sqsClient.ReceiveMessage(&sqs.ReceiveMessageInput{
			QueueUrl:            &queueURL,
			MaxNumberOfMessages: aws.Int64(1),
			VisibilityTimeout:   &timeout,
		})
		if err != nil {
			return errors.Trace(err)
		}
		if len(msgResult.Messages) == 0 {
			log.Info("All aws subscription message processed.")
			break
		}
		log.Info("aws subscription message received.", zap.Int("count", len(msgResult.Messages)))

		for _, msg := range msgResult.Messages {
			processed, err := processOneAWSSubscribeChange(ctx, msg.Body, processor)
			if err != nil {
				log.Error("aws subscribe message processed failed.", zap.Error(err))
				allErrors = append(allErrors, err)
				continue
			}
			if !processed {
				log.Info("aws subscribe message processed ignored.")
				continue
			}
			_, err = sqsClient.DeleteMessage(&sqs.DeleteMessageInput{
				QueueUrl:      &queueURL,
				ReceiptHandle: msg.ReceiptHandle,
			})
			if err != nil {
				log.Error("aws subscribe delete message failed.", zap.Error(err))
				allErrors = append(allErrors, err)
				continue
			}
		}
	}
	if len(allErrors) != 0 {
		return errors.Errorf("Process tenant subscribe change error. %v", len(allErrors))
	}
	return nil
}

func (provider *awsAccountProviderImpl) GetPaymentType() string {
	return billing2.PaymentType_AWS
}