package impl

import (
	"bytes"
	"encoding/csv"
	"errors"
	"fmt"
	"io"
	"math"
	"slices"
	"time"

	"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/s3"
	"github.com/pingcap/dbaas/billing2/billing_errors"
	"github.com/pingcap/dbaas/billing2/config"
	"github.com/pingcap/dbaas/billing2/constant"
	"github.com/pingcap/dbaas/billing2/db/entity"
	"github.com/pingcap/dbaas/billing2/forms"
	"github.com/pingcap/dbaas/billing2/handler/awshandler"
	"github.com/pingcap/dbaas/billing2/handler/centralhandler"
	"github.com/pingcap/dbaas/billing2/handler/paymenthandler"
	"github.com/pingcap/dbaas/billing2/model/payment"
	"github.com/pingcap/dbaas/billing2/utils/log"
	"github.com/pingcap/dbaas/billing2/utils/timeutils"
	"github.com/shopspring/decimal"
	"go.uber.org/zap"
)

func getAwsDefaultSession() (*session.Session, error) {
	region := config.Get().GetString(config.AwsMarketplaceRegion)
	sess, err := session.NewSession()
	if err != nil {
		return nil, err
	}
	creds := stscreds.NewCredentials(sess, config.Get().GetString(config.AWSRole))
	sess = sess.Copy(&aws.Config{Credentials: creds, Region: &region})
	return sess, nil
}

func getAwsMeteringClient() (*marketplacemetering.MarketplaceMetering, error) {
	sess, err := getAwsDefaultSession()
	if err != nil {
		return nil, err
	}

	return marketplacemetering.New(sess), nil
}

type AWSInvoice struct {
	TenantID              uint
	PaymentCode           string
	ProductCode           string
	ProductUnit           string
	UnitPriceFactor       int64
	EnlargeDiscountFactor decimal.Decimal
	Invoices              []forms.Invoice
}

func SendInvoice(inv *AWSInvoice) ([]forms.Invoice, error) {
	var err error
	now := timeutils.GetNow()
	total := int64(0)
	for i := 0; i != len(inv.Invoices); i++ {
		price := inv.Invoices[i].CostPrice
		enlargedPrice := decimal.Zero
		if !inv.EnlargeDiscountFactor.Equal(decimal.NewFromInt(1)) {
			enlargeFactor := decimal.NewFromInt(1).Div(inv.EnlargeDiscountFactor).Sub(decimal.NewFromInt(1))
			enlargedPrice = price.Mul(enlargeFactor)
			log.Info("sending aws cppo metering, should enlarge the metering", zap.String("userCode", inv.PaymentCode), zap.String("AppliedCPPODiscount", inv.EnlargeDiscountFactor.String()), zap.String("originPrice", price.String()), zap.String("enlargedPrice", enlargedPrice.String()))
			price = price.Add(enlargedPrice)
		}
		quantity := price.Mul(decimal.NewFromInt(inv.UnitPriceFactor)).IntPart()
		accountRevenue := decimal.NewFromInt(quantity).Mul(inv.EnlargeDiscountFactor).Div(decimal.NewFromInt(inv.UnitPriceFactor))
		inv.Invoices[i].ChannelLoss = inv.Invoices[i].CostPrice.Sub(accountRevenue)
		total += quantity
	}
	if total == 0 {
		log.Info("skip send", zap.String("userCode", inv.PaymentCode), zap.Int64("quantity", total), zap.String("unit", inv.ProductUnit))
		return inv.Invoices, nil
	}
	if total > math.MaxInt32 {
		err = errors.New("quantity out of limit")
		log.Error(err.Error(), zap.String("userCode", inv.PaymentCode), zap.Int64("quantity", total), zap.String("unit", inv.ProductUnit))
		return nil, err
	}
	if !config.Get().GetBool(config.AWSPaymentEnable) {
		return inv.Invoices, nil
	}

	client, err := getAwsMeteringClient()
	if err != nil {
		return nil, err
	}

	awsRecords := []*marketplacemetering.UsageRecord{
		{
			CustomerIdentifier: &inv.PaymentCode,
			Dimension:          &inv.ProductUnit,
			Quantity:           &total,
			Timestamp:          &now,
		},
	}
	input := marketplacemetering.BatchMeterUsageInput{}
	input.SetUsageRecords(awsRecords)
	input.SetProductCode(inv.ProductCode)

	output, err := client.BatchMeterUsage(&input)
	if err != nil {
		return nil, err
	}
	if len(output.UnprocessedRecords) > 0 {
		unMatchedKeys := make([]string, len(output.UnprocessedRecords))
		for index, record := range output.UnprocessedRecords {
			unMatchedKeys[index] = *record.Dimension
		}
		log.Error("aws marketplace metering usage records unprocessed", zap.Uint("tenantId", inv.TenantID), zap.Strings("UnprocessedRecords", unMatchedKeys), zap.String("input", input.String()), zap.String("output", output.String()))
	}

	for _, result := range output.Results {
		if *result.Status != marketplacemetering.UsageRecordResultStatusSuccess {
			log.Error("aws marketplace metering usage records status not success", zap.Uint("tenantId", inv.TenantID), zap.String("input", input.String()), zap.String("status", *result.Status), zap.String("output", output.String()))
		}
	}

	log.Info("success call aws  metering", zap.Uint("tenantId", inv.TenantID), zap.String("input", input.String()), zap.String("output", output.String()))
	return inv.Invoices, nil
}

type AwsPaymentHandlerImpl struct {
}

const awsDailyUsageDataType = "daily_business_usage_by_instance_type"
const (
	awsCustomerIDKey = "Customer Reference ID"
	awsProductKey    = "Product Code"
	awsRevenueKey    = "Estimated Revenue"
)

func (aws *AwsPaymentHandlerImpl) Setup(tenant *forms.Tenant) (string, error) {
	if !config.Get().GetBool(config.AWSPaymentEnable) {
		return fmt.Sprintf("mock_aws_%v", tenant.TenantID), nil
	}
	if tenant.ThirdPartyAccountProvider != constant.AwsAccountProvider && !tenant.IsCPPOReseller {
		return "", billing_errors.NewSystemError(errors.New("invalid third_party_account_handler"))
	}
	return tenant.ThirdPartyAccountCode, nil
}

func (aws *AwsPaymentHandlerImpl) Payable(tenantID uint, userCode string) (bool, error) {
	tenant, err := centralhandler.GetCentralHandler().GetTenant(tenantID)
	return !tenant.ThirdPartyFreeTrial, err
}

func (*AwsPaymentHandlerImpl) NeedMultiInvoiceSendOnce() bool {
	return true
}

func (aws *AwsPaymentHandlerImpl) Pay(paymentInfo *forms.PaymentMethod, billingCompany *forms.BillingCompany, invoice *forms.Invoice) (*forms.Invoice, error) {
	return nil, billing_errors.NewNotImplementedError(errors.New("can not send aws marketplace once per invoice"))
}

func (aws *AwsPaymentHandlerImpl) PayMultiInvoice(paymentInfo *forms.PaymentMethod, billingCompany *forms.BillingCompany, invoices []forms.Invoice) ([]forms.Invoice, error) {
	lastInvoiceSentDate, err := (&payment.InvoiceModel{}).GetLastSendInvoiceDateByPaymentCode(entity.PaymentType_AWS, paymentInfo.PaymentCode)
	if err != nil {
		return nil, err
	}
	if lastInvoiceSentDate.Add(time.Hour).After(timeutils.GetNow()) {
		log.Info("can not send invoice to aws twice in less than one hour ", zap.Uint("tenantId", paymentInfo.TenantID), zap.String("paymentCode", paymentInfo.PaymentCode))
		return nil, nil
	}
	productCode := config.Get().GetString(config.AWSProductCode)
	productUnit := config.Get().GetString(config.AWSProductUnit)
	productPriceFactor := config.Get().GetInt(config.AWSProductUnitPriceFactor)
	if config.Get().GetBool(config.EnableAWSProduct1DollarUnit) && !slices.Contains(config.Get().GetStringSlice(config.AWSProduct1DollarUnitBlacklist), paymentInfo.PaymentCode) {
		productUnit = config.Get().GetString(config.AWSProduct1DollarUnit)
		productPriceFactor = config.Get().GetInt(config.AWSProduct1DollarUnitPriceFactor)
	}
	if productPriceFactor <= 0 {
		err := errors.New("invalid productPriceFactor")
		log.Error("invalid productPriceFactor", zap.Int("productPriceFactor", productPriceFactor))
		return nil, err
	}
	enlargeDiscountFactor := decimal.NewFromInt(1)
	if paymentInfo.PaymentMeta.EnlargeMarketplaceMetering {
		enlargeDiscountFactor = paymentInfo.PaymentMeta.EnlargerDiscount
	}
	for i := 0; i != len(invoices); i++ {
		invoices[i].PaymentCode = &paymentInfo.PaymentCode
		invoices[i].PaymentMethodID = &paymentInfo.PaymentMethodID
		invoices[i].Meta.PaymentMeta = paymentInfo.PaymentMeta
	}
	return SendInvoice(&AWSInvoice{
		TenantID:              paymentInfo.TenantID,
		ProductCode:           productCode,
		PaymentCode:           paymentInfo.PaymentCode,
		ProductUnit:           productUnit,
		UnitPriceFactor:       int64(productPriceFactor),
		EnlargeDiscountFactor: enlargeDiscountFactor,
		Invoices:              invoices,
	})
}

func (aws *AwsPaymentHandlerImpl) GetThirdPartyName() string {
	return entity.PaymentType_AWS
}

func (aws *AwsPaymentHandlerImpl) EnableReconciliation() bool {
	return true
}

func (aws *AwsPaymentHandlerImpl) ParseAWSReport(body []byte) ([]*paymenthandler.TenantRevenue, error) {
	csvReader := csv.NewReader(bytes.NewBuffer(body))
	header, err := csvReader.Read()
	if err != nil {
		return nil, err
	}
	rows, err := csvReader.ReadAll()
	if err != nil {
		return nil, err
	}
	var ret []*paymenthandler.TenantRevenue
	for _, row := range rows {
		tmp := make(map[string]string)
		for i, colName := range header {
			tmp[colName] = row[i]
		}
		for _, colName := range []string{awsCustomerIDKey, awsProductKey, awsRevenueKey} {
			val, ok := tmp[colName]
			if !ok || val == "" {
				return nil, billing_errors.NewSystemError(errors.New(fmt.Sprintf("parse report failed, col: %s not existed", colName)))
			}
		}
		customerID := tmp[awsCustomerIDKey]
		productCode := tmp[awsProductKey]
		if productCode != config.Get().GetString(config.AWSProductCode) {
			log.Info("no-target product detected", zap.String("productCode", productCode))
			continue
		}
		revenue, err := decimal.NewFromString(tmp[awsRevenueKey])
		if err != nil {
			return nil, billing_errors.NewSystemError(errors.New(fmt.Sprintf("parse report failed, %s is not decimal", tmp[awsRevenueKey])))
		}
		tr := &paymenthandler.TenantRevenue{
			PaymentCode: customerID,
			Revenue:     revenue,
		}
		ret = append(ret, tr)
	}
	return ret, nil
}

func (aws *AwsPaymentHandlerImpl) getAWSReport(key string) ([]byte, error) {
	s3BucketName := config.Get().GetString(config.ReconciliationAWSS3BucketName)
	s3Cli, err := awshandler.NewS3Client()
	if err != nil {
		return nil, err
	}
	resp, err := s3Cli.GetObject(&s3.GetObjectInput{
		Bucket: &s3BucketName,
		Key:    &key,
	})
	if err != nil {
		return nil, err
	}
	ret, err := io.ReadAll(resp.Body)
	if err != nil {
		log.Error("failed to read resp", zap.Error(err))
		return nil, err
	}
	return ret, nil
}

func (aws *AwsPaymentHandlerImpl) GetDailyRevenue(date time.Time) ([]*paymenthandler.TenantRevenue, error) {
	if !config.Get().GetBool(config.AWSPaymentEnable) {
		return nil, nil
	}
	date = timeutils.GetDayDate(date).AddDate(0, 0, 1)
	key := fmt.Sprintf("%s/%s/%s_%s.csv", config.Get().GetString(config.ENV), awsDailyUsageDataType, awsDailyUsageDataType, date.Format("2006-01-02"))
	raw, err := aws.getAWSReport(key)
	if err != nil {
		log.Error("failed to waiting for generate data set", zap.Error(err), zap.String("key", key))
		return nil, err
	}
	return aws.ParseAWSReport(raw)
}

func (aws *AwsPaymentHandlerImpl) GetMonthlyRevenue(date time.Time) ([]*paymenthandler.TenantRevenue, error) {
	if !config.Get().GetBool(config.AWSPaymentEnable) {
		return nil, nil
	}
	s := timeutils.GetMonthDate(date)
	e := s.AddDate(0, 1, 0)
	trMap := make(map[string]*paymenthandler.TenantRevenue)
	for d := s; d.Before(e); d.AddDate(0, 0, 1) {
		tenantRevenueList, err := aws.GetDailyRevenue(d)
		if err != nil {
			log.Error("failed get daily revenue on monthly reconciliation", zap.Error(err))
			return nil, err
		}
		for _, tr := range tenantRevenueList {
			if _, ok := trMap[tr.PaymentCode]; !ok {
				trMap[tr.PaymentCode] = &paymenthandler.TenantRevenue{
					Revenue:     decimal.Zero,
					PaymentCode: tr.PaymentCode,
				}
			}
			trMap[tr.PaymentCode].Revenue = trMap[tr.PaymentCode].Revenue.Add(tr.Revenue)
		}
	}
	ret := make([]*paymenthandler.TenantRevenue, len(trMap))
	index := 0
	for _, tr := range trMap {
		ret[index] = tr
	}
	return ret, nil
}

func (aws *AwsPaymentHandlerImpl) EnableTaxCalculation() bool {
	return false
}

func (aws *AwsPaymentHandlerImpl) EnableHourlySending(tenantID, billingConfigID uint) bool {
	return slices.Contains(config.Get().GetStringSlice(config.HourlyInvoiceTenantBillingConfigIDList), fmt.Sprintf("%d-%d", tenantID, billingConfigID))
}

func (aws *AwsPaymentHandlerImpl) EnableDailySending(tenantID, billingConfigID uint) (bool, int) {
	return slices.Contains(config.Get().GetStringSlice(config.DailyInvoiceTenantBillingConfigIDList),
		fmt.Sprintf("%d-%d", tenantID, billingConfigID)), config.Get().GetInt(config.DailyInvoiceIssueDelayDays)
}

func (aws *AwsPaymentHandlerImpl) NeedInMonthInvoice() bool {
	return true
}

func (aws *AwsPaymentHandlerImpl) NeedSetInvoiceAutoConfirm() bool {
	return true
}