package networkmetering

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/constants"
	"github.com/tidbcloud/metering/pkg/common/metrics"
	"github.com/tidbcloud/metering/pkg/common/models"
	"github.com/tidbcloud/metering/pkg/common/utils"
	"github.com/tidbcloud/metering/pkg/common/utils/apigateway"
	"go.uber.org/zap"
)

type NetworkMetering struct {
	loader  NetworkBillingLoader
	store   NetworkBillingStorer
	conf    *config.ServiceConfig
	gateway apigateway.ApiGateway
	log     *zap.Logger
}

func NewNetworkMetering(conf *config.ServiceConfig, log *zap.Logger) (*NetworkMetering, error) {
	loader, err := NewNetworkBillingLoad(conf, log).Init()
	if err != nil {
		return nil, err
	}
	store, err := NewNetworkBillingStore(conf).Init()
	if err != nil {
		return nil, err
	}
	gateway := apigateway.NewApiGatewayClient(log, conf.Vendor, conf.Env, conf.Secret.ApiGatewayKey, constants.ApiGatewayTimeout)
	return &NetworkMetering{
		loader:  loader,
		store:   store,
		conf:    conf,
		log:     log,
		gateway: gateway,
	}, nil
}
func (n *NetworkMetering) SetGateway(gateway apigateway.ApiGateway) {
	n.gateway = gateway
}
func (n *NetworkMetering) Metering(ctx context.Context, timestamp int64) (string, bool, error) {
	switch n.conf.Vendor {
	case constants.VendorAzure:
		{
			return n.MeteringAzure(ctx, timestamp)
		}
	case constants.VendorAWS:
		{
			return n.MeteringAWS(ctx, timestamp)
		}
	case constants.VendorGCP:
		{
			return n.MeteringGCP(ctx, timestamp)
		}
	}

	return "", false, fmt.Errorf("vendor %s not support", n.conf.Vendor)
}

func (n *NetworkMetering) MeteringAzure(ctx context.Context, timestamp int64) (string, bool, error) {
	billings, err := n.loader.LoadAzure(ctx, timestamp)
	if err != nil {
		//if no metering meta data, then we should retry metering later
		if errors.Is(err, errNometaFound) {
			return "", false, nil
		}
		return "", false, err
	}
	if len(billings) == 0 {
		//no billing data, we should retry metering later
		return "", false, nil
	}
	destbilling, destreporter, err := n.store.Load(ctx, timestamp)
	if err != nil {
		return "", false, err
	}
	// close slqite db after use
	defer n.store.CloseDB(ctx, timestamp)
	res, err := n.DealBillingAzure(billings, destbilling, timestamp)
	if err != nil {
		return "", false, err
	}
	repoterres, err := n.DealBillingRepoter(ctx, res, destreporter, timestamp)
	if err != nil {
		return "", false, err
	}
	err = n.store.Store(ctx, timestamp, res, repoterres)
	if err != nil {
		return "", false, err
	}

	// upload file to  storage
	blobpath, err := n.store.Upload(ctx, timestamp)
	if err != nil {
		return "", false, err
	}
	return blobpath, true, nil
}

func (n *NetworkMetering) DealBillingAzure(billings []*models.AzureBilling, meterings []*models.BillingMetering, timestamp int64) ([]*models.BillingMetering, error) {
	//split the metering to map
	res := make(map[string]*models.BillingMetering)
	oldres := make(map[string]*models.BillingMetering)
	//if metering not empty, then use metering as the base
	for _, metering := range meterings {
		key := metering.Unit + metering.TenantID + metering.ProjectID + metering.ClusterID + metering.ResourceID
		//convert 10^8 to 10^10 for high precision
		metering.Cost = metering.Cost * 100
		oldres[key] = metering
	}
	for _, billing := range billings {
		unit, skutype, ok := n.buildAzureUnit(billing)
		//skip all the unknown unit
		if !ok {
			continue
		}
		// unit + TenantID  + projectid + clusterid  + resourceid = key
		tags := n.PickAzureTagValue(billing.Tags)
		tenant := tags[constants.TagTenant]
		project := tags[constants.TagProject]
		cluster := tags[constants.TagCluster]
		usedby := tags[constants.TagUsedBy]
		environment := tags[constants.TagEnvironment]
		resourceid := billing.ResourceId
		//filter the environment
		if !n.filterEnvironmentAzure(resourceid, environment) {
			n.log.Debug("metering skipped env", zap.String("id", resourceid), zap.String("env", environment))
			continue
		}
		//fix ipv4 usedby
		if skutype == constants.TrafficIpv4 {
			serviceType := tags[constants.TagK8sService]
			if strings.Contains(serviceType, constants.CustomerIPName) {
				usedby = constants.UsedByCustomer
			}
		}
		//fix internal lb usedby
		if skutype == constants.TrafficLB {
			// for azure , all internal lb and public lb are billing to customer
			usedby = constants.UsedByCustomer
		}
		if usedby != constants.UsedByCustomer {
			n.log.Debug("metering skipped with no customer tag", zap.String("billing", fmt.Sprintf("%v", billing)))
			continue
		}
		if !utils.IsNumeric(cluster) {
			cluster = ""
		}
		key := unit + tenant + project + cluster + resourceid
		if _, ok := res[key]; !ok {
			res[key] = &models.BillingMetering{
				TenantID:    tenant,
				ProjectID:   project,
				ClusterID:   cluster,
				UsedBy:      usedby,
				ResourceID:  resourceid,
				Unit:        unit,
				BillingDate: timestamp,
				Cost:        billingFloat2Int64High(billing.CostInBillingCurrency),
			}
		} else {
			res[key].Cost += billingFloat2Int64High(billing.CostInBillingCurrency)
		}
	}
	// compare the old metering and new metering,if diff less than 5 then ignore the new metering
	for key, metering := range res {
		if oldmetering, ok := oldres[key]; ok {
			if metering.Cost-oldmetering.Cost < 500 {
				n.log.Debug(fmt.Sprintf("metering %v cost %d < oldmetering %v cost %d, less than 500 in 10 ^ 10", metering, metering.Cost, oldmetering, oldmetering.Cost))
				res[key] = oldmetering
			} else {
				n.log.Debug(fmt.Sprintf("metering %v cost %d > oldmetering %v cost %d, more than 500 in 10 ^ 10", metering, metering.Cost, oldmetering, oldmetering.Cost))
				//if the cost is more than 500, then we should use the new metering instead of the old metering
				res[key].ID = oldmetering.ID
				res[key].CreatedAt = oldmetering.CreatedAt
			}
		}
	}
	//return the result
	ret := make([]*models.BillingMetering, 0, len(res))
	for _, v := range res {
		//convert 10^10 to 10^8
		v.Cost = v.Cost / 100
		ret = append(ret, v)
	}
	return ret, nil
}

func (n *NetworkMetering) DealBillingRepoter(ctx context.Context, meterings []*models.BillingMetering, repoters []*models.BillingReport, timestamp int64) ([]*models.BillingReport, error) {
	repotermap := make(map[string][]*models.BillingReport)
	meteringmap := make(map[string]*models.BillingMetering)
	//if repoter not empty, then use repoter as the base
	for _, repoter := range repoters {
		// repoter key = Unit + TenantID + ProjectID + ClusterID
		lables, err := getjson(repoter.Labels)
		if err != nil {
			return nil, err
		}
		projectid := lables[constants.TagBillingProjectID]
		clusterid := lables[constants.TagBillingClusterID]
		key := repoter.Unit + strconv.FormatUint(repoter.TenantID, 10) + projectid + clusterid
		repotermap[key] = append(repotermap[key], repoter)
	}
	cachePorjectBilliable := make(map[string]bool)
	// merge resource metering to cluster metering
	for _, metering := range meterings {
		// skip the metering if tenantid or projectid or clusterid is empty
		if metering.TenantID == "" || metering.ProjectID == "" || !utils.IsNumeric(metering.TenantID) || !utils.IsNumeric(metering.ProjectID) {
			n.log.Info("metering skipped with wrong tenantid or projectid", zap.String("metering", fmt.Sprintf("%v", metering)))
			continue
		}
		//query global api to get the count of billable clusters, if the count is 0 then skip the metering
		if billable, ok := cachePorjectBilliable[metering.ProjectID]; !ok {
			startTime := time.Unix(timestamp, 0).UTC()
			endTime := time.Unix(timestamp, 0).UTC().Add(24 * time.Hour)
			count, err := n.gateway.CountBillableClusters(ctx, metering.TenantID, metering.ProjectID, startTime, endTime)
			if err != nil {
				return nil, err
			}
			if count == 0 {
				cachePorjectBilliable[metering.ProjectID] = false
				n.log.Info("project not billable", zap.String("projectid", metering.ProjectID))
				continue
			}
			cachePorjectBilliable[metering.ProjectID] = true
		} else {
			if !billable {
				n.log.Info("project not billable", zap.String("projectid", metering.ProjectID))
				continue
			}
		}

		// metering key = Unit + TenantID + ProjectID + ClusterID
		key := metering.Unit + metering.TenantID + metering.ProjectID + metering.ClusterID
		if _, ok := meteringmap[key]; !ok {
			meteringmap[key] = &models.BillingMetering{
				TenantID:  metering.TenantID,
				ProjectID: metering.ProjectID,
				ClusterID: metering.ClusterID,
				Unit:      metering.Unit,
				Cost:      metering.Cost,
			}
		} else {
			meteringmap[key].Cost += metering.Cost
		}
	}
	// merge metering to repoter
	for key, metering := range meteringmap {
		if repoters, ok := repotermap[key]; !ok {
			// this check we did in previous step, so we can skip it
			tenantid, err := strconv.ParseUint(metering.TenantID, 10, 64)
			if err != nil {
				return nil, err
			}
			//skip the metering if cost <=0
			if metering.Cost <= 0 {
				continue
			}
			labels := map[string]string{
				constants.TagBillingProjectID: metering.ProjectID,
			}
			if utils.IsNumeric(metering.ClusterID) {
				labels[constants.TagBillingClusterID] = metering.ClusterID
			} else {
				if metering.ClusterID != "" {
					n.log.Error("build repoter with wrong clusterid", zap.String("metering", fmt.Sprintf("%v", metering)), zap.String("cluster id", metering.ClusterID))
				}
			}
			labelsjson, err := tojson(labels)
			if err != nil {
				return nil, err
			}
			repotermap[key] = append(repotermap[key], &models.BillingReport{
				TenantID:      tenantid,
				CostID:        utils.GenerateID(),
				Cost:          metering.Cost,
				CostTimestamp: timestamp,
				Labels:        labelsjson,
				Unit:          metering.Unit,
			})
		} else {
			var cost int64
			for _, repoter := range repoters {
				cost += repoter.Cost
			}
			if cost < metering.Cost {
				labels := map[string]string{
					constants.TagBillingProjectID: metering.ProjectID,
				}
				if utils.IsNumeric(metering.ClusterID) {
					labels[constants.TagBillingClusterID] = metering.ClusterID
				} else {
					if metering.ClusterID != "" {
						n.log.Error("build repoter with wrong clusterid", zap.String("metering", fmt.Sprintf("%v", metering)), zap.String("cluster id", metering.ClusterID))
					}
				}
				labelsjson, err := tojson(labels)
				if err != nil {
					return nil, err
				}
				repotermap[key] = append(repotermap[key], &models.BillingReport{
					TenantID:      repoters[0].TenantID,
					CostID:        utils.GenerateID(),
					Cost:          metering.Cost - cost, // build the diff
					CostTimestamp: timestamp,
					Labels:        labelsjson,
					Unit:          metering.Unit,
				})
			}
			if cost > metering.Cost {
				n.log.Error("metering cost  > repoter cost", zap.Any("metering", metering), zap.Int64("cost", metering.Cost), zap.Int64("repoter cost", cost))
			}
		}
	}
	var isrerun bool
	var rerundiff, countrerun int64
	if v := ctx.Value(constants.CtxKeyRerunTsDiff); v != nil {
		rerundiff = v.(int64)
		if rerundiff > 0 {
			isrerun = true
		}
	}
	//return the result and metricing the rerun count
	ret := make([]*models.BillingReport, 0, len(repotermap))
	for _, v := range repotermap {
		if isrerun {
			for _, r := range v {
				if r.CreatedAt == 0 {
					// if createdat is 0, then it is the first time to report
					countrerun++
				}
			}
		}
		ret = append(ret, v...)
	}
	if isrerun {
		labels := map[string]string{
			constants.MetricsRerunDiffTs:     strconv.FormatInt(rerundiff, 10),
			constants.MetricsLblProviderType: n.conf.Vendor,
		}
		metrics.MetricsNetworkRerunCount.With(labels).Set(float64(countrerun))
	}
	return ret, nil
}

func (n *NetworkMetering) buildAzureUnit(billing *models.AzureBilling) (string, string, bool) {
	// check the billing schema version
	version := n.conf.TrafficMeteringConfig.Provider.Azure.BillingSchemaVersion
	if version == constants.AzureSchemaVersionV2024 {
		return buildAzureUnit2024(billing)
	} else if version == constants.AzureSchemaVersionV2021 {
		return buildAzureUnit2021(billing)
	}
	// default to 2024 version
	return buildAzureUnit2024(billing)
}

// sku like "TrafficSKU:provider:region:{corss_az|internat|lb|cross_region}"
func buildAzureUnit2024(billing *models.AzureBilling) (string, string, bool) {
	switch billing.MeterCategory {
	case "Bandwidth":
		{
			if billing.MeterSubCategory == "Inter-Region" {
				return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficCrossRegion), constants.TrafficCrossRegion, true
			}
			return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficInternet), constants.TrafficInternet, true
		}
	case "Load Balancer":
		{
			return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficLB), constants.TrafficLB, true
		}
	case "NAT Gateway":
		{
			return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficNAT), constants.TrafficNAT, true
		}
	case "Virtual Network":
		{
			if billing.MeterSubCategory == "IP Addresses" {
				return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficIpv4), constants.TrafficIpv4, true
			}
			//we remove the private link traffic now,because the private link endpoint traffic is  billable in owner account
			// if billing.MeterSubCategory == "Private Link" {
			// 	if strings.HasPrefix(billing.MeterName, "Standard Data Processed") {
			// 		return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficLB), true
			// 	}
			// }
		}
	}
	return "", "", false

}

func buildAzureUnit2021(billing *models.AzureBilling) (string, string, bool) {
	switch billing.MeterCategory {
	case "Bandwidth":
		{
			if strings.Contains(billing.MeterSubCategory, "Inter-Region") {
				return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficCrossRegion), constants.TrafficCrossRegion, true
			}
			return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficInternet), constants.TrafficInternet, true
		}
	case "Load Balancer":
		{
			return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficLB), constants.TrafficLB, true
		}
	case "NAT Gateway":
		{
			return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficNAT), constants.TrafficNAT, true
		}
	case "Virtual Network":
		{
			if billing.MeterSubCategory == "IP Addresses" {
				return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficIpv4), constants.TrafficIpv4, true
			}
			//we remove the private link traffic now,because the private link endpoint traffic is  billable in owner account
			// if billing.MeterSubCategory == "Private Link" {
			// 	if strings.HasPrefix(billing.MeterName, "Standard Data Processed") {
			// 		return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorAzure, strings.ToLower(billing.ResourceLocation), constants.TrafficLB), true
			// 	}
			// }
		}
	}
	return "", "", false

}
func pickAzureTagValue2024(input string) map[string]string {
	pairs := strings.Split(input, "\",")
	tagMap := make(map[string]string)
	for _, pair := range pairs {
		// Trim any extraneous characters and split by colon
		trimmedPair := strings.Trim(pair, "\"")
		parts := strings.SplitN(trimmedPair, "\": \"", 2)
		if len(parts) == 2 {
			tag := parts[0]
			value := parts[1]
			tagMap[tag] = value
		}
	}
	return tagMap
}

func pickAzureTagValue2021(input string) map[string]string {
	//{"aks-managed-bootstrap-profile-acr-name":"","aks-managed-consolidated-additional-properties":"852e83a3-67a6-11f0-adf2-c2e1fcecf750","aks-managed-coordination":"true","aks-managed-createOperationID":"2decd873-7072-432c-bd58-27b9a7b3f936","aks-managed-creationSource":"vmssclient-aks-n2295da69929-19159213-vmss","aks-managed-enable-imds-restriction":"false","aks-managed-kubeletIdentityClientID":"f5457795-1972-4a39-b773-f1cb51a1f8d0","aks-managed-orchestrator":"Kubernetes:1.32.5","aks-managed-poolName":"n2295da69929","aks-managed-resourceNameSuffix":"20660695","aks-managed-ssh-access":"LocalUser","cluster":"shared","project":"1369847559694040704","servicetype":"dedicated","tenant":"1369847559692509562","usedby":"customer-tidb"}
	//drop {}
	input = strings.TrimPrefix(input, "{")
	input = strings.TrimSuffix(input, "}")
	pairs := strings.Split(input, "\",")
	tagMap := make(map[string]string)
	for _, pair := range pairs {
		// Trim any extraneous characters and split by colon
		trimmedPair := strings.Trim(pair, "\"")
		parts := strings.SplitN(trimmedPair, "\":\"", 2)
		if len(parts) == 2 {
			tag := parts[0]
			value := parts[1]
			tagMap[tag] = value
		}
	}
	return tagMap
}

func (n *NetworkMetering) PickAzureTagValue(input string) map[string]string {
	version := n.conf.TrafficMeteringConfig.Provider.Azure.BillingSchemaVersion
	if version == constants.AzureSchemaVersionV2024 {
		return pickAzureTagValue2024(input)
	} else if version == constants.AzureSchemaVersionV2021 {
		return pickAzureTagValue2021(input)
	}
	// default to 2024
	return pickAzureTagValue2024(input)
}

// billingFloat2Int64High converts a float64 to an int64 by multiplying by 10^10
func billingFloat2Int64High(cost float64) int64 {
	return int64(cost * 10000000000)
}

func buildAWSSKUPrefix(billing *models.AWSBilling) string {
	if strings.EqualFold(strings.TrimSpace(billing.MeteringServicePlan), constants.ServicePlanPremiumReserved) {
		return fmt.Sprintf("%s:%s:%s:reserved", constants.TrafficSKU, constants.VendorAWS, strings.ToLower(billing.ProductRegion))
	}
	return fmt.Sprintf("%s:%s:%s", constants.TrafficSKU, constants.VendorAWS, strings.ToLower(billing.ProductRegion))
}

func getjson(s string) (map[string]string, error) {
	var result map[string]string
	err := json.Unmarshal([]byte(s), &result)
	if err != nil {

		return nil, err
	}
	return result, nil
}

func tojson(m map[string]string) (string, error) {
	var result string
	b, err := json.Marshal(m)
	if err != nil {
		return "", err
	}
	result = string(b)
	return result, nil
}

func (n *NetworkMetering) MeteringAWS(ctx context.Context, timestamp int64) (string, bool, error) {
	billings, err := n.loader.LoadAWS(ctx, timestamp)
	if err != nil {
		//if no metering meta data, then we should retry metering later
		if errors.Is(err, errNometaFound) {
			return "", false, nil
		}
		return "", false, err
	}
	if len(billings) == 0 {
		//no billing data, we should retry metering later
		return "", false, nil
	}
	destbilling, destreporter, err := n.store.Load(ctx, timestamp)
	if err != nil {
		return "", false, err
	}
	// close slqite db after use
	defer n.store.CloseDB(ctx, timestamp)
	res, err := n.DealBillingAWS(billings, destbilling, timestamp)
	if err != nil {
		return "", false, err
	}
	repoterres, err := n.DealBillingRepoter(ctx, res, destreporter, timestamp)
	if err != nil {
		return "", false, err
	}
	err = n.store.Store(ctx, timestamp, res, repoterres)
	if err != nil {
		return "", false, err
	}

	// upload file to  storage
	blobpath, err := n.store.Upload(ctx, timestamp)
	if err != nil {
		return "", false, err
	}
	return blobpath, true, nil
}

func (n *NetworkMetering) DealBillingAWS(billings []*models.AWSBilling, meterings []*models.BillingMetering, timestamp int64) ([]*models.BillingMetering, error) {
	//split the metering to map
	res := make(map[string]*models.BillingMetering)
	oldres := make(map[string]*models.BillingMetering)
	//if metering not empty, then use metering as the base
	for _, metering := range meterings {
		key := metering.Unit + metering.TenantID + metering.ProjectID + metering.ClusterID + metering.ResourceID
		//convert 10^8 to 10^10 for high precision
		metering.Cost = metering.Cost * 100
		oldres[key] = metering
	}
	for _, billing := range billings {
		servicePlan := strings.ToLower(strings.TrimSpace(billing.MeteringServicePlan))
		if servicePlan != "" {
			if servicePlan != constants.ServicePlanPremiumReserved {
				n.log.Debug("metering skipped with unsupported service plan",
					zap.String("service_plan", servicePlan),
					zap.String("resource_id", billing.LineItemResourceId),
				)
				continue
			}
			if !billing.MeteringPremiumResMatchedOnly {
				continue
			}
		}
		unit, ok := buildAWSUnit(billing)
		//skip all the unknown unit
		if !ok {
			continue
		}
		// unit + TenantID  + projectid + clusterid  + resourceid = key
		tenant := billing.ResourceTagsTenant
		project := billing.ResourceTagsProject
		cluster := billing.ResourceTagsCluster
		usedby := billing.ResourceTagsUsedby
		resourceid := billing.LineItemResourceId
		// skip the metering if not customer used in legacy dedicated mode.
		if servicePlan == "" && usedby != constants.UsedByCustomer {
			n.log.Debug("metering skipped with no customer tag", zap.String("billing", fmt.Sprintf("%v", billing)))
			continue
		}
		if servicePlan == constants.ServicePlanPremiumReserved && usedby == "" {
			usedby = constants.UsedByCustomer
		}
		// if cluster is not numeric, then set it to empty
		// because in dev , it may be set to useless value
		if !utils.IsNumeric(cluster) {
			cluster = ""
		}
		key := unit + tenant + project + cluster + resourceid
		if _, ok := res[key]; !ok {
			res[key] = &models.BillingMetering{
				TenantID:    tenant,
				ProjectID:   project,
				ClusterID:   cluster,
				UsedBy:      usedby,
				ResourceID:  resourceid,
				Unit:        unit,
				BillingDate: timestamp,
				Cost:        billingFloat2Int64High(billing.LineItemUnblendedCost),
			}
		} else {
			res[key].Cost += billingFloat2Int64High(billing.LineItemUnblendedCost)
		}
		// metering ipv4 with sku lb
		if billing.LineItemProductCode == "AWSELB" &&
			strings.HasPrefix(billing.LineItemOperation, "LoadBalancing:") &&
			billing.PricingUnit == "Hrs" &&
			billing.ResourceTagsComponent == "nlb" &&
			billing.LineItemUsageAmount > 0 {
			ipv4unit := fmt.Sprintf("%s:%s", buildAWSSKUPrefix(billing), constants.TrafficIpv4)
			key := ipv4unit + tenant + project + cluster + resourceid
			if _, ok := res[key]; !ok {
				res[key] = &models.BillingMetering{
					TenantID:    tenant,
					ProjectID:   project,
					ClusterID:   cluster,
					UsedBy:      usedby,
					ResourceID:  resourceid,
					Unit:        ipv4unit,
					BillingDate: timestamp,
					Cost:        billingFloat2Int64High(constants.AwsPublicIPv4AddressPricing * 3),
				}
			} else {
				res[key].Cost += billingFloat2Int64High(constants.AwsPublicIPv4AddressPricing * 3)
			}
		}
	}

	// compare the old metering and new metering,if diff less than 5 then ignore the new metering
	for key, metering := range res {
		if oldmetering, ok := oldres[key]; ok {
			if metering.Cost-oldmetering.Cost < 500 {
				n.log.Debug(fmt.Sprintf("metering %v cost %d < oldmetering %v cost %d, less than 500 in 10 ^ 10", metering, metering.Cost, oldmetering, oldmetering.Cost))
				res[key] = oldmetering
			} else {
				n.log.Debug(fmt.Sprintf("metering %v cost %d > oldmetering %v cost %d, more than 500 in 10 ^ 10", metering, metering.Cost, oldmetering, oldmetering.Cost))
				//if the cost is more than 500, then we should use the new metering instead of the old metering
				res[key].ID = oldmetering.ID
				res[key].CreatedAt = oldmetering.CreatedAt
			}
		}
	}
	//return the result
	ret := make([]*models.BillingMetering, 0, len(res))
	for _, v := range res {
		if v.ClusterID == "" {
			n.log.Info("metering found empty clusterid", zap.String("metering", fmt.Sprintf("%v", v)))
		}
		//convert 10^10 to 10^8
		v.Cost = v.Cost / 100
		ret = append(ret, v)
	}
	return ret, nil
}

/*
`SELECT "resource_tags_user_tenant","resource_tags_user_project","resource_tags_user_usedby",
	sum("line_item_blended_cost") as "line_item_blended_cost",sum("line_item_unblended_cost") as "line_item_unblended_cost",sum("line_item_usage_amount") as "line_item_usage_amount","pricing_unit",
    "line_item_resource_id",CONCAT("line_item_product_code",':',"product_servicename") AS "service_name","line_item_operation","product_region_code",
	"resource_tags_user_cluster","resource_tags_user_component","resource_tags_user_environment","resource_tags_user_servicetype"
FROM "{{ .Database }}"."{{ .Table }}"
WHERE "line_item_usage_start_date" >= timestamp '{{ .StartTime }}' AND "line_item_usage_start_date" < timestamp '{{ .EndTime }}'
    AND (
		"line_item_product_code" IN ('AWSELB','AmazonS3') OR
	    ("line_item_product_code"='AmazonEC2' AND "product_servicename" = 'AWS Data Transfer')
	)
	AND "line_item_unblended_cost" > 0
	AND "resource_tags_user_usedby" LIKE 'customer%'
	AND "resource_tags_user_component" != 'extra-tidb'
    AND "resource_tags_user_tenant" = '{{ .TenantID }}'
	{{ if .ProjectID }}AND "resource_tags_user_project" = '{{ .ProjectID }}'{{ end }}
GROUP BY "resource_tags_user_tenant","resource_tags_user_project","resource_tags_user_usedby",
"pricing_unit",
"line_item_resource_id","line_item_product_code","product_servicename","line_item_operation","product_region_code",
"resource_tags_user_cluster","resource_tags_user_component","resource_tags_user_environment","resource_tags_user_servicetype"
;
*/
// sku like "TrafficSKU:provider:region:{corss_az|internat|lb|cross_region}" for AWS
func buildAWSUnit(billing *models.AWSBilling) (string, bool) {
	prefix := buildAWSSKUPrefix(billing)
	switch billing.LineItemProductCode {
	case "AWSELB":
		{
			if strings.HasPrefix(billing.LineItemOperation, "LoadBalancing-PublicIP-") || strings.HasPrefix(billing.LineItemOperation, "LoadBalancing-NLB-PublicIP-") {
				return fmt.Sprintf("%s:%s", prefix, constants.TrafficInternet), true
			} else if billing.LineItemOperation == "LoadBalancing" || strings.HasPrefix(billing.LineItemOperation, "LoadBalancing:") {
				return fmt.Sprintf("%s:%s", prefix, constants.TrafficLB), true
			} else if strings.HasPrefix(billing.LineItemOperation, "InterZone-") || strings.HasPrefix(billing.LineItemOperation, "VPCPeering-") {
				return fmt.Sprintf("%s:%s", prefix, constants.TrafficCrossAZ), true
			}
		}
	case "AmazonEC2":
		{
			if billing.ProductServicename == "AWS Data Transfer" {
				if strings.HasPrefix(billing.LineItemOperation, "InterZone-") || strings.HasPrefix(billing.LineItemOperation, "VPCPeering-") {
					return fmt.Sprintf("%s:%s", prefix, constants.TrafficCrossAZ), true
				}
			}
		}
	}
	return "", false

}

func (n *NetworkMetering) filterEnvironmentAzure(resourceid string, environment string) bool {
	if environment == "" {
		//if the environment is empty, use the resourceid to filter
		perfix := fmt.Sprintf("dedicated-%s-", n.conf.Env)
		if strings.Contains(resourceid, perfix) {
			return true
		}
	}
	if environment == n.conf.Env {
		return true
	}
	return false
}

func (n *NetworkMetering) MeteringGCP(ctx context.Context, timestamp int64) (string, bool, error) {
	billings, err := n.loader.LoadGCP(ctx, timestamp)
	if err != nil {
		return "", false, err
	}
	if len(billings) == 0 {
		//no billing data, we should retry metering later
		return "", false, nil
	}
	destbilling, destreporter, err := n.store.Load(ctx, timestamp)
	if err != nil {
		return "", false, err
	}
	// close slqite db after use
	defer n.store.CloseDB(ctx, timestamp)
	res, err := n.DealBillingGCP(billings, destbilling, timestamp)
	if err != nil {
		return "", false, err
	}
	repoterres, err := n.DealBillingRepoter(ctx, res, destreporter, timestamp)
	if err != nil {
		return "", false, err
	}
	err = n.store.Store(ctx, timestamp, res, repoterres)
	if err != nil {
		return "", false, err
	}

	// upload file to  storage
	blobpath, err := n.store.Upload(ctx, timestamp)
	if err != nil {
		return "", false, err
	}
	return blobpath, true, nil
}

func ensureUsedBy(billing *models.GCPBilling) string {
	if billing.UserTagUsedBy == constants.UsedByCustomer {
		return billing.UserTagUsedBy
	}
	return billing.UserLabelUsedBy
}

// for gcp cluster id mybe shared ,so we should ingnore the cluster id which is not numeric
func (n *NetworkMetering) DealBillingGCP(billings []*models.GCPBilling, meterings []*models.BillingMetering, timestamp int64) ([]*models.BillingMetering, error) {
	//split the metering to map
	res := make(map[string]*models.BillingMetering)
	oldres := make(map[string]*models.BillingMetering)
	//if metering not empty, then use metering as the base
	for _, metering := range meterings {
		key := metering.Unit + metering.TenantID + metering.ProjectID + metering.ClusterID + metering.ResourceID
		//convert 10^8 to 10^10 for high precision
		metering.Cost = metering.Cost * 100
		oldres[key] = metering
	}
	for _, billing := range billings {
		unit, ok := buildGCPUnit(billing)
		//skip all the unknown unit
		if !ok {
			continue
		}
		// unit + TenantID  + projectid + clusterid  + resourceid = key
		tenant := billing.UserLabelTenantID
		project := billing.UserLabelProjectID
		cluster := billing.UserLabelClusterID
		usedby := ensureUsedBy(billing)
		resourceid := billing.ResourceName
		if usedby != constants.UsedByCustomer {
			n.log.Debug("metering skipped with no customer tag", zap.String("billing", fmt.Sprintf("%v", billing)))
			continue
		}
		if !utils.IsNumeric(cluster) {
			cluster = ""
		}
		n.log.Debug("gcp billing cost detail",
			zap.String("tenant", tenant),
			zap.String("project", project),
			zap.String("cluster", cluster),
			zap.String("resource", resourceid),
			zap.Float64("cost", billing.Cost),
			zap.Float64("cost_at_list", billing.CostAtList),
			zap.Float64("cost_diff", billing.CostAtList-billing.Cost),
		)
		key := unit + tenant + project + cluster + resourceid
		if _, ok := res[key]; !ok {
			res[key] = &models.BillingMetering{
				TenantID:    tenant,
				ProjectID:   project,
				ClusterID:   cluster,
				UsedBy:      usedby,
				ResourceID:  resourceid,
				Unit:        unit,
				BillingDate: timestamp,
				Cost:        billingFloat2Int64High(billing.CostAtList),
			}
		} else {
			res[key].Cost += billingFloat2Int64High(billing.CostAtList)
		}
	}
	// compare the old metering and new metering,if diff less than 5 then ignore the new metering
	for key, metering := range res {
		if oldmetering, ok := oldres[key]; ok {
			if metering.Cost-oldmetering.Cost < 500 {
				n.log.Debug(fmt.Sprintf("metering %v cost %d < oldmetering %v cost %d, less than 500 in 10 ^ 10", metering, metering.Cost, oldmetering, oldmetering.Cost))
				res[key] = oldmetering
			} else {
				n.log.Debug(fmt.Sprintf("metering %v cost %d > oldmetering %v cost %d, more than 500 in 10 ^ 10", metering, metering.Cost, oldmetering, oldmetering.Cost))
				//if the cost is more than 500, then we should use the new metering instead of the old metering
				res[key].ID = oldmetering.ID
				res[key].CreatedAt = oldmetering.CreatedAt
			}
		}
	}
	//return the result
	ret := make([]*models.BillingMetering, 0, len(res))
	for _, v := range res {
		//convert 10^10 to 10^8
		v.Cost = v.Cost / 100
		ret = append(ret, v)
	}
	return ret, nil
}

// sku like "TrafficSKU:provider:region:{corss_az|internat|lb|cross_region}" for AWS
func buildGCPUnit(billing *models.GCPBilling) (string, bool) {
	switch billing.ServiceName {
	case "Compute Engine":
		{
			if strings.HasPrefix(billing.Description, "Network Inter Zone") {
				return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorGCP, strings.ToLower(billing.Region), constants.TrafficCrossAZ), true
			} else if strings.HasPrefix(billing.Description, "Network Inter Region") {
				return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorGCP, strings.ToLower(billing.Region), constants.TrafficCrossRegion), true
			} else if strings.HasPrefix(billing.Description, "Network Internet") {
				return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorGCP, strings.ToLower(billing.Region), constants.TrafficInternet), true
			}
		}
	case "Networking":
		{
			if strings.Contains(billing.Description, "Network Load Balancer") {
				return fmt.Sprintf("%s:%s:%s:%s", constants.TrafficSKU, constants.VendorGCP, strings.ToLower(billing.Region), constants.TrafficLB), true
			}
		}
	}
	return "", false

}