package handler

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"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/errors"
	"github.com/tidbcloud/account/commons/config"
	"github.com/tidbcloud/account/thirdparty/billing2"
	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/multierr"
	"go.uber.org/zap"
)

type awsAccountProviderImpl struct {
}

func GetAWSDefaultSession() (*session.Session, error) {
	sess, err := session.NewSession()
	if err != nil {
		return nil, errs.NewInternalServerError("get aws session", err)
	}
	return sess, nil
}

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 {
		typedErr, ok := err.(awserr.Error)
		if ok {
			if typedErr.Code() == "InvalidTokenException" {
				return nil, errs.NewBadRequest("Invalid AWS account token", nil)
			}

			if typedErr.Code() == "ExpiredTokenException" {
				return nil, errs.NewBadRequest("The AWS account token expired", nil)
			}
		}
		return nil, errs.NewInternalServerError("Failed to resolve AWS account token", err)
	}

	productCode := config.Global().MarketPlaceConfig.AWSProductCode
	CPPODiscountableProductCode := config.Global().MarketPlaceConfig.AWSCPPODiscountableProductCode
	CPPONonDiscountableProductCode := config.Global().MarketPlaceConfig.AWSCPPONonDiscountableProductCode

	var identifier string

	if output.ProductCode == nil {
		tilog.Error(context.Background(), "empty product code", zap.String("configuredProductCode", productCode))
		return nil, errs.NewBadRequest("Wrong product code", nil)
	}
	switch *output.ProductCode {
	case productCode:
		identifier = *output.CustomerIdentifier
	case CPPODiscountableProductCode, CPPONonDiscountableProductCode:
		identifier = fmt.Sprintf("%s-%s", *output.ProductCode, *output.CustomerIdentifier)
		tilog.Error(context.Background(), "AWS CPPO Product subscribed", zap.String("productCode", *output.ProductCode), zap.String("identifier", *output.CustomerIdentifier))
	default:
		tilog.Error(context.Background(), "Wrong product code", zap.String("productCode", *output.ProductCode), zap.String("configuredProductCode", productCode))
		return nil, errs.NewBadRequest("Wrong product code", nil)
	}

	return &identifier, nil
}

func (provider *awsAccountProviderImpl) ApproveCode(code *string) error {
	return nil
}

const AWSTokenKey = "x-amzn-marketplace-token"
const AWSOfferTypeKey = "x-amzn-marketplace-offer-type"

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 (provider *awsAccountProviderImpl) GetFreeTrialFromRedirect(c *gin.Context) (bool, error) {
	offerType := c.Request.PostFormValue(AWSOfferTypeKey)
	return offerType == "free-trial", nil
}

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

	return sqs.New(sess), nil
}

const QueueTimeout int64 = 30

type AWSMarketplaceMessage struct {
	Action                 string  `json:"action"`
	CustomerIdentifier     string  `json:"customer-identifier"`
	ProductCode            string  `json:"product-code"`
	OfferIdentifier        *string `json:"offer-identifier"`
	IsFreeTrialTermPresent string  `json:"isFreeTrialTermPresent"`
}

type AWSSNSNotification struct {
	Type             string `json:"Type"`
	MessageId        string `json:"MessageId"`
	Message          string `json:"Message"`
	TopicArn         string `json:"TopicArn"`
	Timestamp        string `json:"Timestamp"`
	SignatureVersion string `json:"SignatureVersion"`
	Signature        string `json:"Signature"`
	SigningCertURL   string `json:"SigningCertURL"`
	UnsubscribeURL   string `json:"UnsubscribeURL"`
}

func awsMarketplaceMessageDecoder(s string) (*AWSMarketplaceMessage, error) {
	var msg AWSMarketplaceMessage
	err := json.Unmarshal([]byte(s), &msg)
	if err != nil {
		return nil, err
	}
	return &msg, nil
}

func processOneAWSSubscribeChange(ctx context.Context, msgBody *AWSMarketplaceMessage, processor ProcessTenantSubscribeChangeFunc) (bool, error) {
	action := msgBody.Action
	change := SubscribeChange{}
	switch action {
	case "subscribe-success":
		change.SubscribeStatus = SubscribeStatusSubscribed
		change.FreeTrial = msgBody.IsFreeTrialTermPresent == "true"
	case "unsubscribe-pending":
		change.SubscribeStatus = SubscribeStatusUnsubscribed
	case "subscribe-fail":
		change.SubscribeStatus = SubscribeStatusSubscribeFailed
	default:
		return true, nil
	}
	AWSProductCode := config.Global().MarketPlaceConfig.AWSProductCode
	CPPODiscountableProductCode := config.Global().MarketPlaceConfig.AWSCPPODiscountableProductCode
	CPPONonDiscountableProductCode := config.Global().MarketPlaceConfig.AWSCPPONonDiscountableProductCode
	change.AccountProvider = AWSAccountProvider
	productCode := msgBody.ProductCode
	identifier := msgBody.CustomerIdentifier
	switch productCode {
	case AWSProductCode:
		change.AccountCode = identifier
		tilog.Info(ctx, "Customer subscribing status changed", zap.Bool("freeTrial", change.FreeTrial), zap.String("action", action), zap.String("productCode", productCode), zap.String("identifier", identifier))
	case CPPODiscountableProductCode, CPPONonDiscountableProductCode:
		change.AccountCode = fmt.Sprintf("%s-%s", productCode, identifier)
		tilog.Error(ctx, "CPPO Customer subscribing status changed", zap.String("action", action), zap.String("productCode", productCode), zap.String("identifier", identifier))
	default:
		tilog.Error(ctx, "Wrong product code", zap.String("productCode", productCode))
		return true, nil
	}

	// store or update the latest subscription
	if action == "subscribe-success" {
		tilog.Info(ctx, "start to create or active subscription record", zap.String("productCode", productCode), zap.String("identifier", identifier))
		_, err := billing2.GetHandler().CreateOrActiveAWSSubscriptionRecord(ctx, productCode, change.AccountCode, change.FreeTrial)
		if err != nil {
			tilog.Error(ctx, "error to create or active aws subscription record", zap.String("productCode", productCode), zap.Error(err))
			return false, err
		}
	}
	if action == "unsubscribe-pending" || action == "subscribe-fail" {
		tilog.Info(ctx, "start to inactive subscription record", zap.String("productCode", productCode), zap.String("identifier", identifier))
		_, err := billing2.GetHandler().InactiveAWSSubscriptionRecord(ctx, productCode, change.AccountCode)
		if err != nil {
			tilog.Error(ctx, "error to inactive aws subscription record", zap.String("productCode", productCode), zap.Error(err))
			return false, err
		}
	}
	return processor(ctx, change)
}

func (provider *awsAccountProviderImpl) ProcessTenantSubscribeChange(ctx context.Context, processor ProcessTenantSubscribeChangeFunc) error {
	if !config.Global().MarketPlaceConfig.EnableAwsBackground {
		tilog.Info(ctx, "Skip process aws tenant subscribe by config")
		return nil
	}
	sqsClient, err := getAWSSQSClient()
	if err != nil {
		return err
	}
	queueURL := config.Global().MarketPlaceConfig.AWSSQS
	cppoQueueURL := config.Global().MarketPlaceConfig.AWSCPPOSQS
	timeout := QueueTimeout
	allErrors := make([]error, 0)
	for q, decoder := range map[string]func(string) (*AWSMarketplaceMessage, error){
		queueURL: awsMarketplaceMessageDecoder,
		cppoQueueURL: func(s string) (*AWSMarketplaceMessage, error) {
			var msg AWSSNSNotification
			err = json.Unmarshal([]byte(s), &msg)
			if err != nil {
				return nil, err
			}
			return awsMarketplaceMessageDecoder(msg.Message)
		},
	} {
		for {
			tilog.Info(ctx, "start to get msg from sqs queue", zap.String("queue", q))
			msgResult, err := sqsClient.ReceiveMessage(&sqs.ReceiveMessageInput{
				QueueUrl:            &q,
				MaxNumberOfMessages: aws.Int64(1),
				VisibilityTimeout:   &timeout,
			})
			if err != nil {
				tilog.Error(ctx, "failed to get msg from sqs queue", zap.Error(err), zap.String("queue", q))
				break
			}
			if len(msgResult.Messages) == 0 {
				tilog.Info(ctx, "All aws subscription message processed.", zap.String("queue", q))
				break
			}
			tilog.Info(ctx, "aws subscription messages received.", zap.Int("count", len(msgResult.Messages)), zap.String("queue", q))

			for _, sqsMsg := range msgResult.Messages {
				tilog.Info(ctx, "aws subscription message received.", zap.Stringp("sqsMsg", sqsMsg.Body), zap.String("queue", q))
				msg, err := decoder(*sqsMsg.Body)
				if err != nil {
					tilog.Error(ctx, "aws subscribe message decode failed.", zap.Error(err), zap.String("queue", q))
					allErrors = append(allErrors, err)
					continue
				}
				processed, err := processOneAWSSubscribeChange(ctx, msg, processor)
				if err != nil {
					tilog.Error(ctx, "aws subscribe message processed failed.", zap.Error(err), zap.String("queue", q))
					allErrors = append(allErrors, err)
					continue
				}
				if !processed {
					tilog.Info(ctx, "aws subscribe message processed ignored.", zap.String("queue", q))
					continue
				}
				// processOneAWSSubscribeChange returned nil: DB state and eventbus
				// publish have already committed (see MarketplaceWorker.processSubscribed
				// / processUnsubscribed). Best-effort email/IM notifications are NOT
				// part of this contract — a transient notify failure must not block
				// the SQS delete, otherwise the message re-delivers and storms the
				// customer with repeated unsubscribe emails.
				_, err = sqsClient.DeleteMessage(&sqs.DeleteMessageInput{
					QueueUrl:      &q,
					ReceiptHandle: sqsMsg.ReceiptHandle,
				})
				if err != nil {
					tilog.Error(ctx, "aws subscribe delete message failed.", zap.Error(err), zap.String("queue", q))
					allErrors = append(allErrors, err)
					continue
				}
			}
		}
	}
	if len(allErrors) != 0 {
		// The previous message only carried the count, dropping every underlying
		// error. Root-cause analysis required manually grepping upstream log lines
		// by logid. Surface every per-message failure via multierr so the worker
		// log line preserves the real reason (ReceiveMessage / decode / billing2 /
		// processor / DeleteMessage) instead of a bare "1".
		return multierr.Combine(allErrors...)
	}
	return nil
}