package impl

import (
	"bytes"
	"context"
	"encoding/csv"
	"errors"
	"fmt"
	"io"
	"net/http"
	"slices"
	"time"

	servicecontrol "cloud.google.com/go/servicecontrol/apiv1"
	"cloud.google.com/go/storage"
	"github.com/google/uuid"
	"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/centralhandler"
	"github.com/pingcap/dbaas/billing2/handler/gcphandler/auth"
	"github.com/pingcap/dbaas/billing2/handler/paymenthandler"
	"github.com/pingcap/dbaas/billing2/model/payment"

	//"github.com/pingcap/dbaas/billing2/model/payment/paymentsub"
	"github.com/pingcap/dbaas/billing2/utils/httpclient"
	"github.com/pingcap/dbaas/billing2/utils/log"
	"github.com/pingcap/dbaas/billing2/utils/timeutils"
	pingcaperrors "github.com/pingcap/errors"
	"github.com/shopspring/decimal"
	"go.uber.org/zap"
	"golang.org/x/oauth2"
	"google.golang.org/api/option"
	servicecontrolpb "google.golang.org/genproto/googleapis/api/servicecontrol/v1"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const (
	gcpCustomerIDKey   = "external_account_id"
	gcpProductSKUIDKey = "sku_id"
	gcpRevenueKey      = "charges"
)

func getClientTS() (oauth2.TokenSource, error) {
	ts, err := auth.DefaultTokenSource()
	if err != nil {
		return nil, billing_errors.NewSystemError(err)
	}
	marketPlaceSA := config.Get().GetString(config.GCPMarketPlaceSA)

	if marketPlaceSA == "" {
		return nil, billing_errors.NewSystemError(errors.New("invalid marketplace sa"))
	}
	return auth.Impersonate(marketPlaceSA, ts)
}

func signGCPAuth(req *http.Request) error {
	ts, err := getClientTS()
	if err != nil {
		return err
	}

	token, err := ts.Token()
	if err != nil {
		return billing_errors.NewSystemError(err)
	}

	token.SetAuthHeader(req)

	return nil
}

func getGCPClient() httpclient.HTTPClient {
	baseUrl := config.Get().GetString(config.GCPProcurementBaseUrl)
	return httpclient.NewClient(signGCPAuth, baseUrl, "")
}

func getGcpMeteringClient() (*servicecontrol.ServiceControllerClient, error) {
	ts, err := getClientTS()
	if err != nil {
		return nil, err
	}

	ctx := context.Background()
	return servicecontrol.NewServiceControllerClient(ctx, option.WithTokenSource(ts))
}

const GCPActiveState = "active"

/*
EntitlementState at https://cloud.google.com/marketplace/docs/partners/commerce-procurement-api/reference/rest/v1/providers.entitlements#EntitlementState
*/
// ENTITLEMENT_ACTIVE Indicates that the entitlement is active. The procured item is now usable and any associated billing events will start occurring.
// ENTITLEMENT_PENDING_PLAN_CHANGE Indicates that the entitlement is currently active, but there is a pending plan change that is requested by the customer.
// ENTITLEMENT_PENDING_PLAN_CHANGE_APPROVAL Indicates that the entitlement is currently active, but there is a plan change request pending provider approval.

const ENTITLEMENT_ACTIVE = "ENTITLEMENT_ACTIVE"
const ENTITLEMENT_PENDING_PLAN_CHANGE = "ENTITLEMENT_PENDING_PLAN_CHANGE"
const ENTITLEMENT_PENDING_PLAN_CHANGE_APPROVAL = "ENTITLEMENT_PENDING_PLAN_CHANGE_APPROVAL"

type EntitlementInfo struct {
	Product          string `json:"productExternalName"`
	UsageReportingId string `json:"usageReportingId"`
	Account          string `json:"account"`
	State            string `json:"state"`
}

type ListResult struct {
	Entitlements []*EntitlementInfo `json:"entitlements"`
}

func getGCPReportID(accountCode string) (string, error) {
	partnerID := config.Get().GetString(config.GCPMarketPlaceProject)
	productCode := config.Get().GetString(config.GCPProductCode)
	_ = productCode
	url := fmt.Sprintf("/v1/providers/%s/entitlements?filter=account=%s", partnerID, accountCode)
	output := ListResult{}
	err := getGCPClient().Get(url, nil, &output)
	if err != nil {
		return "", err
	}
	log.Info("[getGCPReportID] request params", zap.String("accountCode", accountCode), zap.String("partnerID", partnerID), zap.String("productCode", productCode), zap.String("url", url))
	log.Info("[getGCPReportID] raw response", zap.Any("output", output))
	activeStateMap := map[string]bool{
		ENTITLEMENT_ACTIVE:                       true,
		ENTITLEMENT_PENDING_PLAN_CHANGE:          true,
		ENTITLEMENT_PENDING_PLAN_CHANGE_APPROVAL: true,
	}
	res := []*EntitlementInfo{}
	for _, outputResult := range output.Entitlements {
		log.Info("[getGCPReportID] entitlement detail", zap.String("state", outputResult.State), zap.String("product", outputResult.Product), zap.String("account", outputResult.Account), zap.String("usageReportingId", outputResult.UsageReportingId))
		if activeStateMap[outputResult.State] {
			if outputResult.Product == productCode && outputResult.Account == fmt.Sprintf("providers/%s/accounts/%s", partnerID, accountCode) {
				res = append(res, outputResult)
			}
		}
	}

	log.Info("[getGCPReportID] matched entitlements", zap.Int("count", len(res)))

	if len(res) == 0 {
		return "", billing_errors.NewSystemError(errors.New("no valid entitlements"))
	}

	if len(res) > 1 {
		return "", billing_errors.NewSystemError(errors.New("system error: multi valid entitlements"))
	}

	log.Info("[getGCPReportID] final result", zap.String("usageReportingId", res[0].UsageReportingId))
	return res[0].UsageReportingId, nil
}

type GcpPaymentHandlerImpl struct {
}

func (gcp *GcpPaymentHandlerImpl) GetThirdPartyName() string {
	return entity.PaymentType_GCP
}

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

func (gcp *GcpPaymentHandlerImpl) Setup(tenant *forms.Tenant) (string, error) {
	if !config.Get().GetBool(config.GCPPaymentEnable) {
		return fmt.Sprintf("mock_gcp_%v", tenant.TenantID), nil
	}
	if tenant.ThirdPartyAccountProvider != constant.GCPAccountProvider {
		return "", billing_errors.NewSystemError(errors.New("invalid third_party_account_handler"))
	}
	return tenant.ThirdPartyAccountCode, nil
}

const operationName = "TIDB Cloud Hourly Usage Report"

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

func (*GcpPaymentHandlerImpl) EnableReconciliation() bool {
	return true
}

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

func (gcp *GcpPaymentHandlerImpl) PayMultiInvoice(paymentInfo *forms.PaymentMethod, billingCompany *forms.BillingCompany, invoices []forms.Invoice) ([]forms.Invoice, error) {
	if !config.Get().GetBool(config.GCPPaymentEnable) {
		return invoices, nil
	}
	productCode := config.Get().GetString(config.GCPProductCode)
	unitCode := config.Get().GetString(config.GCPProductUnit)
	productPriceFactor := config.Get().GetInt64(config.GCPProudctUnitPriceFactor)

	lastInvoiceSentDate, err := (&payment.InvoiceModel{}).GetLastSendInvoiceDateByPaymentCode(entity.PaymentType_GCP, paymentInfo.PaymentCode)
	if err != nil {
		return nil, err
	}
	if lastInvoiceSentDate.Add(time.Hour).After(timeutils.GetNow()) {
		log.Info("gcp can not send invoice to gcp twice in less than one hour ", zap.Uint("tenantId", invoices[0].TenantID), zap.String("paymentCode", paymentInfo.PaymentCode))
		return nil, nil
	}

	client, err := getGcpMeteringClient()
	if err != nil {
		log.Error("get gcp metering client failed", zap.Error(err))
		return nil, err
	}

	reportID, err := getGCPReportID(paymentInfo.PaymentCode)
	if err != nil {
		log.Error("get gcp report id failed", zap.Error(err), zap.String("userCode", paymentInfo.PaymentCode))
		return nil, err
	}

	quantity := int64(0)
	for i := 0; i != len(invoices); i++ {
		price := invoices[i].CostPrice
		enlargedPrice := decimal.Zero
		if !paymentInfo.PaymentMeta.EnlargeMarketplaceMetering {
			enlargeFactor := decimal.NewFromInt(1).Div(paymentInfo.PaymentMeta.EnlargerDiscount).Sub(decimal.NewFromInt(1))
			enlargedPrice = price.Mul(enlargeFactor)
			log.Info("sending aws cppo metering, should enlarge the metering", zap.String("userCode", paymentInfo.PaymentCode), zap.String("AppliedCPPODiscount", paymentInfo.PaymentMeta.EnlargerDiscount.String()), zap.String("originPrice", price.String()), zap.String("enlargedPrice", enlargedPrice.String()))
			price = price.Add(enlargedPrice)
		}
		quantity += price.Mul(decimal.NewFromInt(int64(productPriceFactor))).IntPart()
	}

	start := timeutils.GetHourDate(timeutils.GetNow())

	req := &servicecontrolpb.ReportRequest{
		ServiceName: productCode,
		Operations: []*servicecontrolpb.Operation{
			{
				OperationId:   uuid.New().String(),
				OperationName: operationName,
				ConsumerId:    reportID,
				StartTime:     timestamppb.New(start.Add(-time.Hour)),
				EndTime:       timestamppb.New(start),
				MetricValueSets: []*servicecontrolpb.MetricValueSet{
					{
						MetricName: fmt.Sprintf("%s/%s", productCode, unitCode),
						MetricValues: []*servicecontrolpb.MetricValue{
							{
								Value: &servicecontrolpb.MetricValue_Int64Value{
									Int64Value: quantity,
								},
							},
						},
					},
				},
			},
		},
	}

	resp, err := client.Report(context.Background(), req)

	if err != nil {
		err = pingcaperrors.Trace(err)
		log.Error("failed call gcp metering", zap.String("input", req.String()), zap.Error(err))
		return nil, err
	}
	log.Info("success call gcp metering", zap.String("input", req.String()), zap.String("output", resp.String()))
	return invoices, nil
}

func (gcp *GcpPaymentHandlerImpl) ParseGCPReport(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
	tenantRevenueMap := make(map[string]*paymenthandler.TenantRevenue)
	for _, row := range rows {
		tmp := make(map[string]string)
		for i, colName := range header {
			tmp[colName] = row[i]
		}
		for _, colName := range []string{gcpRevenueKey, gcpProductSKUIDKey, gcpCustomerIDKey} {
			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[gcpCustomerIDKey]
		productSKUID := tmp[gcpProductSKUIDKey]
		if productSKUID != config.Get().GetString(config.GCPProductSKUID) {
			log.Info("no-target product detected", zap.String("productSKUID", productSKUID))
			continue
		}
		revenue, err := decimal.NewFromString(tmp[gcpRevenueKey])
		if err != nil {
			return nil, billing_errors.NewSystemError(errors.New(fmt.Sprintf("parse report failed, %s is not decimal", tmp[gcpRevenueKey])))
		}
		if _, ok := tenantRevenueMap[customerID]; !ok {
			tr := &paymenthandler.TenantRevenue{
				PaymentCode: customerID,
				Revenue:     decimal.Zero,
			}
			ret = append(ret, tr)
			tenantRevenueMap[customerID] = tr
		}
		tenantRevenueMap[customerID].Revenue = tenantRevenueMap[customerID].Revenue.Add(revenue)
	}
	return ret, nil
}

func (gcp *GcpPaymentHandlerImpl) getGCPReport(key string) ([]byte, error) {
	ctx := context.Background()
	ts, err := getClientTS()
	if err != nil {
		return nil, err
	}
	client, err := storage.NewClient(ctx, option.WithTokenSource(ts))
	if err != nil {
		return nil, err
	}
	bucketName := config.Get().GetString(config.ReconciliationGCSBucketName)
	bucket := client.Bucket(bucketName)

	obj := bucket.Object(key)
	reader, err := obj.NewReader(ctx)
	if err != nil {
		return nil, err
	}
	ret, err := io.ReadAll(reader)
	if err != nil {
		log.Error("failed to read resp", zap.Error(err))
	}
	return ret, nil
}

func (gcp *GcpPaymentHandlerImpl) GetDailyRevenue(date time.Time) ([]*paymenthandler.TenantRevenue, error) {
	if !config.Get().GetBool(config.GCPPaymentEnable) {
		return nil, nil
	}
	date = timeutils.GetDayDate(date)
	key := fmt.Sprintf("%s Daily Insights", date.Format("2006-01-02"))
	raw, err := gcp.getGCPReport(key)
	if err != nil {
		log.Error("failed to waiting for generate data set", zap.Error(err), zap.String("key", key))
		return nil, err
	}
	return gcp.ParseGCPReport(raw)
}

func (gcp *GcpPaymentHandlerImpl) GetMonthlyRevenue(date time.Time) ([]*paymenthandler.TenantRevenue, error) {
	if !config.Get().GetBool(config.GCPPaymentEnable) {
		return nil, nil
	}
	date = timeutils.GetMonthDate(date)
	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 := gcp.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 (gcp *GcpPaymentHandlerImpl) EnableTaxCalculation() bool {
	return false
}

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

func (gcp *GcpPaymentHandlerImpl) 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 (gcp *GcpPaymentHandlerImpl) NeedInMonthInvoice() bool {
	return true
}

func (gcp *GcpPaymentHandlerImpl) NeedSetInvoiceAutoConfirm() bool {
	return true
}