package dm

import (
	"context"
	"fmt"
	"net"
	"strings"
	"time"

	"github.com/pingcap/errors"
	"github.com/tidbcloud/account/idl/pbgen/proto/account"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/internal/db"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

const (
	// provider:region
	nodeUnitPattern = "dm-rcu:%s:%s"
	// provider
	endpointUnitPattern = "pl-endpoint:%s"
	// input network (pl): provider
	// private link only charge for cross AZ
	privateLinkUnitPattern = "pl-network:%s"
	// input network (vpc): provider:CrossAZ/CrossRegion
	// vpc peering charge for cross AZ & cross region
	vpcPeeringUnitPattern = "vpcpeering:%s:%s"
	// dm network: provider
	dmNetworkUnitPattern = "dm-network:%s"
	// dm nat: dm-nat:aws:us-west2; for gcp it's fixed dm-nat:gcp:all
	natUnitPattern = "dm-nat:%s:%s"
	// dm public ip: dm-internet:aws:all
	publicIPUnitPattern = "dm-internet:%s:%s"
	crossRegion         = "CrossRegion"
	//nolint:deadcode,unused,varcheck
	crossAZ = "CrossAZ" // will be used in the future

	dmCategory    = "dm-metering"
	hourInMinutes = 60
)

func recordNewMeterings(dmMeterings []*entity.DMMetering) error {
	return db.Transaction(func(tx *gorm.DB) error {
		for _, rec := range dmMeterings {
			if err := rec.Create(tx); err != nil {
				return err // auto rollback
			}
		}
		return nil
	})
}

func getPrevMeteringRecord(jobID entity.ID, startTs time.Time, kind entity.MeteringKind) (entity.DMMetering, error) {
	var metering entity.DMMetering
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewDMMeteringQuerySet(tx).JobIDEq(jobID).KindEq(kind).StartTSLte(startTs).OrderDescByStartTS().One(&metering)
	})
	return metering, err
}

func getCurMeteringRecord(jobID entity.ID, startTs, endTs time.Time, kind entity.MeteringKind) (entity.DMMetering, error) {
	var metering entity.DMMetering
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewDMMeteringQuerySet(tx).JobIDEq(jobID).KindEq(kind).StartTSEq(startTs).EndTSEq(endTs).One(&metering)
	})
	return metering, err
}

func getRCUBySpecID(specID entity.ID) (float64, error) {
	var spec entity.DMJobSpec
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewDMJobSpecQuerySet(tx).IDEq(specID).One(&spec)
	})
	if err != nil {
		return 0, err
	}
	return spec.CapacityUnit, nil
}

func getUnsyncedDMMeterings() ([]entity.DMMetering, error) {
	var meterings []entity.DMMetering
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewDMMeteringQuerySet(tx).SyncedEq(false).OrderAscByStartTS().All(&meterings)
	})
	if err != nil {
		return nil, err
	}
	return meterings, nil
}

func (m Model) SyncDMMetering(ctx context.Context) error {
	logger := m.Log().With(zap.String("sync dm metering", time.Now().String()))
	meterings, err := getUnsyncedDMMeterings()
	if err != nil {
		logger.Error("fail to get unsynced dm meterings", zap.Error(err))
		return err
	}
	if len(meterings) < 1 {
		logger.Info("no unsynced dm meterings")
		return nil
	}
	success := 0
	formatSingleJob := func(metering entity.DMMetering) {
		logger := logger.With(zap.Uint64("metering-id", uint64(metering.ID)))
		defer func() {
			if err := recover(); err != nil {
				logger.Error("format metering panic occurred", zap.Any("err", err))
				return
			}
		}()
		var (
			fn            func(context.Context, entity.DMMetering) (*entity.BillingRecord, error)
			billingRecord *entity.BillingRecord
			err           error
		)
		switch metering.Kind {
		case entity.MeteringKindEndpoint:
			fn = formatEndpointMetering
		case entity.MeteringKindNode:
			fn = formatNodeMetering
		case entity.MeteringKindInput:
			fn = formatInputNetworkMetering
		case entity.MeteringKindOutput:
			fn = formatOutputNetworkMetering
		case entity.MeteringKindNat:
			fn = formatNatMetering
		default:
			// should not reach here
			logger.Warn("unknown metering kind", zap.String("kind", string(metering.Kind)))
			return
		}
		if billingRecord, err = fn(ctx, metering); err != nil {
			logger.Error("fail to format metering", zap.Error(err))
			return
		}
		err = db.Transaction(func(tx *gorm.DB) error {
			// a billingRecord is nil iff the metering should not be billed
			// e.g. the network usage from the same AZ.
			if billingRecord != nil {
				if err1 := billingRecord.Create(tx); err1 != nil {
					return err
				}
			}
			metering.Synced = true
			return metering.Update(tx, entity.DMMeteringDBSchema.Synced)
		})
		if err != nil {
			logger.Error("fail to sync metering", zap.Error(err))
			return
		}
	}
	// no need to retry for each kind (like in DMMetering)
	// because each kind of metering is ONE unique record in the table.
	for _, metering := range meterings {
		formatSingleJob(metering)
		success++
	}
	logger.Info("finish sync dm metering", zap.Int("success", success), zap.Int("total", len(meterings)))
	return nil
}

func getLabelClusterSourcesByMetering(ctx context.Context, metering *entity.DMMetering) (*core.Cluster, map[string]string, []entity.DMJobSource, error) {
	cluster, err := grpcli.GetClusterByIDAndExcludeStatuses(ctx, uint64(metering.ClusterID), nil)
	if err != nil {
		return nil, nil, nil, err
	}
	project, err := grpcli.GetProjectByID(ctx, &account.GetProjectByIDRequest{Id: cluster.ProjectId})
	if err != nil {
		return nil, nil, nil, err
	}
	labels := map[string]string{
		"cluster": cluster.Name,
		"project": project.Project.Name,
		"region":  cluster.RegionName,
	}
	sources, err1 := GetDMJobSources(metering.JobID)
	if err != nil {
		return nil, nil, nil, err1
	}
	return cluster, labels, sources, nil
}

func formatNatMetering(ctx context.Context, metering entity.DMMetering) (*entity.BillingRecord, error) {
	var (
		labels  map[string]string
		cluster *core.Cluster
		err     error
		unit    string
		cost    float64
	)
	if cluster, labels, _, err = getLabelClusterSourcesByMetering(ctx, &metering); err != nil {
		return nil, err
	}
	if cluster.CloudProviderName == infraapi.ProviderTypeAWS || cluster.CloudProviderName == infraapi.ProviderTypeAzure {
		unit = fmt.Sprintf(natUnitPattern, cluster.CloudProviderName, cluster.RegionName)
	} else {
		// gcp, nat price is same among all regions
		// see https://cloud.google.com/nat/pricing
		unit = fmt.Sprintf(natUnitPattern, cluster.CloudProviderName, "all")
	}
	startTs := metering.StartTS.Add(-1 * time.Hour)
	if prevMetering, err := getPrevMeteringRecord(metering.JobID, startTs, entity.MeteringKindNat); err == nil {
		cost = metering.Cost - prevMetering.Cost
	} else if err == gorm.ErrRecordNotFound {
		cost = metering.Cost
	} else {
		return nil, err
	}
	billingRecord := entity.BillingRecord{
		TenantID:    entity.ID(cluster.TenantId),
		ProjectID:   entity.ID(cluster.ProjectId),
		ClusterID:   entity.ID(cluster.Id),
		MeasuredTS:  metering.StartTS,
		SKU:         unit,
		Total:       int64(cost * hourInMinutes),
		ExtraLabels: labels,
		CostID:      metering.CostID,
		Category:    dmCategory,
	}
	return &billingRecord, nil
}

func formatNodeMetering(ctx context.Context, metering entity.DMMetering) (*entity.BillingRecord, error) {
	var (
		labels  map[string]string
		cluster *core.Cluster
		err     error
	)
	if cluster, labels, _, err = getLabelClusterSourcesByMetering(ctx, &metering); err != nil {
		return nil, err
	}
	unit := fmt.Sprintf(nodeUnitPattern, cluster.CloudProviderName, cluster.RegionName)
	billingRecord := entity.BillingRecord{
		TenantID:    entity.ID(cluster.TenantId),
		ProjectID:   entity.ID(cluster.ProjectId),
		ClusterID:   entity.ID(cluster.Id),
		MeasuredTS:  metering.StartTS,
		SKU:         unit,
		Total:       int64(metering.Cost * hourInMinutes),
		ExtraLabels: labels,
		CostID:      metering.CostID,
		Category:    dmCategory,
	}
	return &billingRecord, nil
}

func formatEndpointMetering(ctx context.Context, metering entity.DMMetering) (*entity.BillingRecord, error) {
	var (
		labels  map[string]string
		cluster *core.Cluster
		err     error
	)
	if cluster, labels, _, err = getLabelClusterSourcesByMetering(ctx, &metering); err != nil {
		return nil, err
	}
	unit := fmt.Sprintf(endpointUnitPattern, cluster.CloudProviderName)
	billingRecord := entity.BillingRecord{
		TenantID:    entity.ID(cluster.TenantId),
		ProjectID:   entity.ID(cluster.ProjectId),
		ClusterID:   entity.ID(cluster.Id),
		MeasuredTS:  metering.StartTS,
		SKU:         unit,
		Total:       int64(metering.Cost * hourInMinutes),
		ExtraLabels: labels,
		CostID:      metering.CostID,
		Category:    dmCategory,
	}
	return &billingRecord, nil
}

func formatInputNetworkMetering(ctx context.Context, metering entity.DMMetering) (*entity.BillingRecord, error) {
	var (
		labels  map[string]string
		cluster *core.Cluster
		err     error
		unit    string
		sources []entity.DMJobSource
		cost    float64
	)
	if cluster, labels, sources, err = getLabelClusterSourcesByMetering(ctx, &metering); err != nil {
		return nil, err
	}
	// TODO: only one sources is supported now
	switch sources[0].ConnProfile.ConnMethod {
	case entity.ConnMethodPrivateLink:
		// TODO: check cross AZ or not in the future
		// if not cross AZ, we should return nil
		unit = fmt.Sprintf(privateLinkUnitPattern, cluster.CloudProviderName)
	case entity.ConnMethodVPCPeering:
		if !sources[0].ConnProfile.SameRegion {
			// cross region
			unit = fmt.Sprintf(vpcPeeringUnitPattern, cluster.CloudProviderName, crossRegion)
		} else {
			// same region
			// TODO: check cross AZ or not in the future
			// if not cross AZ, we should return nil
			unit = fmt.Sprintf(vpcPeeringUnitPattern, cluster.CloudProviderName, crossAZ)
		}
	case entity.ConnMethodDirect:
		// temp: all region same price
		unit = fmt.Sprintf(publicIPUnitPattern, cluster.CloudProviderName, "all")
	}
	startTs := metering.StartTS.Add(-1 * time.Hour)
	if prevMetering, err := getPrevMeteringRecord(metering.JobID, startTs, entity.MeteringKindInput); err == nil {
		cost = metering.Cost - prevMetering.Cost
	} else if err == gorm.ErrRecordNotFound {
		cost = metering.Cost
	} else {
		return nil, err
	}
	if cost == 0 {
		// no need to meter
		return nil, nil
	}
	billingRecord := entity.BillingRecord{
		TenantID:    entity.ID(cluster.TenantId),
		ProjectID:   entity.ID(cluster.ProjectId),
		ClusterID:   entity.ID(cluster.Id),
		MeasuredTS:  metering.StartTS,
		SKU:         unit,
		Total:       int64(cost * hourInMinutes),
		ExtraLabels: labels,
		CostID:      metering.CostID,
		Category:    dmCategory,
	}
	return &billingRecord, nil
}

func formatOutputNetworkMetering(ctx context.Context, metering entity.DMMetering) (*entity.BillingRecord, error) {
	var (
		labels  map[string]string
		cluster *core.Cluster
		err     error
		cost    float64
	)
	if cluster, labels, _, err = getLabelClusterSourcesByMetering(ctx, &metering); err != nil {
		return nil, err
	}
	startTs := metering.StartTS.Add(-1 * time.Hour)
	if prevMetering, err := getPrevMeteringRecord(metering.JobID, startTs, entity.MeteringKindOutput); err == nil {
		cost = metering.Cost - prevMetering.Cost
	} else if err == gorm.ErrRecordNotFound {
		cost = metering.Cost
	} else {
		return nil, err
	}
	if cost == 0 {
		// no need to meter
		return nil, nil
	}
	unit := fmt.Sprintf(dmNetworkUnitPattern, cluster.CloudProviderName)
	billingRecord := entity.BillingRecord{
		TenantID:    entity.ID(cluster.TenantId),
		ProjectID:   entity.ID(cluster.ProjectId),
		ClusterID:   entity.ID(cluster.Id),
		MeasuredTS:  metering.StartTS,
		SKU:         unit,
		Total:       int64(cost * hourInMinutes),
		ExtraLabels: labels,
		CostID:      metering.CostID,
		Category:    dmCategory,
	}
	return &billingRecord, nil
}

type VpcPeeringInfo struct {
	Id           uint64
	Cidr         string
	TenantRegion string
}

func IsVpcPeeringNotFound(err error) bool {
	msg := err.Error()
	return strings.Contains(msg, "vpc peering not found") ||
		strings.Contains(msg, "no matched vpc peering")
}

func GetVpcPeering(ctx context.Context, cluster *core.Cluster, ips []net.IP) (*VpcPeeringInfo, error) {
	if len(ips) == 0 {
		return nil, errors.New("input ips param is empty")
	}
	networkPeers, err := grpcli.ListNetworkPeersByCluster(ctx, cluster)
	if err != nil {
		return nil, errors.Annotate(err, "list network peers by cluster using rGRPC failed")
	}
	if len(networkPeers) == 0 {
		return nil, errors.New("vpc peering not found")
	}
outer:
	for _, p := range networkPeers {
		vpcCIDR := p.GetPeerVpcCidr()
		_, subnet, err := net.ParseCIDR(vpcCIDR)
		if err != nil {
			// should not parse error
			return nil, errors.Annotatef(err, "parse VPC CIDR failed: %s", vpcCIDR)
		}
		for _, ip := range ips {
			if !subnet.Contains(ip) {
				continue outer
			}
		}
		// we've found a VPC, whose subnet contains all the IP ranges
		return &VpcPeeringInfo{
			Id:           p.GetId(),
			Cidr:         vpcCIDR,
			TenantRegion: p.GetRegion(),
		}, nil
	}
	return nil, errors.New("no matched vpc peering")
}