package cluster

import (
	"context"
	"fmt"
	"sort"
	"time"

	"github.com/gogo/protobuf/types"
	"github.com/pingcap/dbaas/central/config"
	"github.com/pingcap/dbaas/central/db"
	"github.com/pingcap/dbaas/central/db/dbutil"
	"github.com/pingcap/dbaas/central/db/entity"
	"github.com/pingcap/dbaas/central/grpc/grpcli"
	"github.com/pingcap/dbaas/central/model/tenant/handler"
	"github.com/pingcap/dbaas/central/service/cvt"
	"github.com/tidbcloud/idl/pbgen/proto/central"
	"github.com/tidbcloud/idl/pbgen/proto/core"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"go.opentelemetry.io/otel/codes"
	"go.uber.org/zap"
)

func filterCloudProviderByMarketplace(provider *central.CloudProvider, tenant *entity.Tenant) bool {
	if tenant != nil && tenant.ThirdPartyAccountProvider != nil && *tenant.ThirdPartyAccountProvider == handler.AWSAccountProvider && provider.Name != entity.Aws.String() {
		return true
	}
	if tenant != nil && tenant.ThirdPartyAccountProvider != nil && *tenant.ThirdPartyAccountProvider == handler.GCPAccountProvider && provider.Name != entity.Gcp.String() {
		return true
	}
	return false
}

func convertCloudProviders(orgID uint64, resp *core.ListCloudProvidersResp, tenant *entity.Tenant) *central.CloudProviders {
	providers := []*central.CloudProvider{}

	for k, v := range resp.Name2CloudProvider {
		cloudProvider := &central.CloudProvider{
			Name:    k,
			Regions: []*central.Region{},
		}

		if filterCloudProviderByMarketplace(cloudProvider, tenant) {
			continue
		}

		for _, r := range v.Name2Region {
			region := &central.Region{
				Name:                     r.Name,
				DisplayName:              r.DisplayName,
				CrossAccountDeploymentId: r.XaccountDeployId,
			}
			cloudProvider.Regions = append(cloudProvider.Regions, region)
		}

		sort.Slice(cloudProvider.Regions, func(i, j int) bool {
			return cloudProvider.Regions[i].DisplayName < cloudProvider.Regions[j].DisplayName
		})
		providers = append(providers, cloudProvider)
	}
	sort.Slice(providers, func(i, j int) bool {
		return providers[i].Name < providers[j].Name
	})

	return &central.CloudProviders{
		CloudProviders: providers,
		BaseResp:       &central.BaseResp{},
	}
}

func (s *clusterService) ListCloudProviders(ctx context.Context, req *central.ListCloudProvidersReq) (*central.CloudProviders, error) {
	userContext := s.userCluster(ctx)
	orgID := userContext.TenantID()
	projectID := req.ProjectId

	if projectID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "project_id parameter is required")
	}
	err := userContext.CheckOrgIDAndProjectID(entity.ID(orgID), entity.ID(projectID))
	if err != nil {
		return nil, err
	}

	deployType := getClusterDeployType(req.IsDevTier)
	listCloudProvidersReq := &core.ListCloudProvidersReq{
		SpecificTenantId:            &orgID,
		FilterOutEmptyCloudProvider: true,
		FilterOutEmptyRegion:        true,
		DeployType:                  &deployType,
		FilterOutInactiveRegion:     true,
		FilterOutInactiveProvider:   true,
		SpecificProjectId:           &projectID,
		WithXaccountDeployId:        true,
	}

	resp, err := grpcli.ListCloudProviders(ctx, listCloudProvidersReq)
	if err != nil {
		return nil, err
	}

	tx := db.NewProvider().WriteDB(ctx)
	tenant, err := dbutil.GetTenantByID(tx, entity.ID(orgID))
	if err != nil {
		return nil, err
	}
	return convertCloudProviders(orgID, resp, &tenant), nil
}

func getClusterDeployType(isDevTier bool) core.DeployType {
	if isDevTier {
		return core.DeployType_DEPLOY_TYPE_SHARED
	}
	return core.DeployType_DEPLOY_TYPE_DEDICATED
}

func (s *clusterService) CalcNodeProfiles(ctx context.Context, req *central.CalcNodeProfilesReq) (*central.CalcNodeProfilesResp, error) {
	userContext := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)

	err := userContext.CheckOrgID(orgID)
	if err != nil {
		return nil, err
	}

	listNodeProfiles4TenantReq := &core.ListNodeProfiles4TenantReq{
		TenantId:          req.OrgId,
		CloudProviderName: req.Provider,
		RegionName:        req.Region,
		DeployType:        getClusterDeployType(req.IsDevTier),
	}
	resp, err := grpcli.ListNodeProfiles4Tenant(ctx, listNodeProfiles4TenantReq)
	if err != nil {
		return nil, err
	}

	nodeProfileMap := make(map[core.ComponentType][]*core.NodeProfile4Tenant)
	for _, nodeProfile := range resp.Profiles {
		nodeProfileMap[nodeProfile.ComponentType] = append(nodeProfileMap[nodeProfile.ComponentType], nodeProfile)
	}

	var nodeProfiles []*central.ComponentNodeProfiles
	for k, v := range nodeProfileMap {
		componentNodeProfiles := &central.ComponentNodeProfiles{
			ComponentType:    cvt.ComponentTypeCore2Central(k),
			Profiles:         nil, //fill later
			DefaultProfileId: 0,
		}
		profiles := make([]*central.ClusterNodeProfile, 0, len(v))
		for _, nodeProfile := range v {
			profiles = append(profiles, cvt.NodeProfileCore2Central(nodeProfile))
		}
		sort.Slice(profiles, func(i, j int) bool {
			return profiles[i].CpuNum < profiles[j].CpuNum
		})
		componentNodeProfiles.Profiles = profiles

		if len(profiles) >= 1 {
			// default policy is the first one
			componentNodeProfiles.DefaultProfileId = profiles[0].Id

			recommendedNodeCpuNum := config.RecommendedNodeCpuNum
			if k == core.ComponentType_COMPONENT_TYPE_TIFLASH {
				recommendedNodeCpuNum = config.RecommendedTiFlashNodeCpuNum
			}

			for _, profile := range profiles {
				if profile.CpuNum == recommendedNodeCpuNum {
					componentNodeProfiles.DefaultProfileId = profile.Id
					break
				}
			}
		}

		nodeProfiles = append(nodeProfiles, componentNodeProfiles)
	}

	calcNodeProfilesResp := &central.CalcNodeProfilesResp{
		NodeProfiles: nodeProfiles,
		BaseResp:     &central.BaseResp{},
	}
	return calcNodeProfilesResp, nil
}

func (s *clusterService) CreateClusterV2(ctx context.Context, req *central.CreateClusterReq) (_ *central.CreateClusterResp, err error) {
	ctx, span := Tracer.Start(ctx, "clusterService.CreateClusterV2")
	defer func() {
		if err != nil {
			span.RecordError(err)
			span.SetStatus(codes.Error, err.Error())
		}
		span.End()
	}()
	logger := log.WithContextS(ctx)

	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err = context.CheckOrgIDAndProjectID(orgID, projectID)
	span.AddEvent("checkOrgIDAndProjectID")
	if err != nil {
		return nil, err
	}
	userID := context.UserID()

	// if primary cluster ID is not zero, it means we should create a cross region replication cluster
	// set a trigger for create replication cluster
	if req.PrimaryClusterId != 0 {
		logger.Infof("try to create replication, primary cluster id: %d", req.PrimaryClusterId)

		_, err := grpcli.CreateReplication(ctx, &core.CreateReplicationReq{
			Type:                         core.Replication_TYPE_CDC,
			PrimaryClusterId:             req.PrimaryClusterId,
			SecondaryComponents:          cvt.ClusterComponentsCentral2Core(req.Components),
			SecondaryClusterRootPassword: req.RootPassword,
			SecondaryClusterRegion:       req.Region,
			SecondaryClusterName:         req.Name,
			SecondaryPort:                req.Port,
			CreateById:                   userID,
			IgnoreReplication:            req.IgnoreReplication,
			BaseReq:                      nil,
		})
		if err != nil {
			return nil, err
		}

		resp := &central.CreateClusterResp{
			Id:       req.PrimaryClusterId,
			BaseResp: &central.BaseResp{},
		}
		return resp, nil
	}

	createClusterReq := &core.CreateClusterReq{
		TenantId:         req.OrgId,
		ProjectId:        req.ProjectId,
		Name:             req.Name,
		RootPassword:     req.RootPassword,
		CloudProvider:    req.Provider,
		Region:           req.Region,
		DeployType:       getClusterDeployType(req.IsDevTier),
		IsDevTier:        req.IsDevTier,
		Components:       cvt.ClusterComponentsCentral2Core(req.Components),
		CreateById:       userID,
		PrimaryClusterId: 0,
		XaccountDeployId: req.CrossAccountDeploymentId,
		Port:             req.Port,
		BaseReq:          nil,
	}
	createClusterResp, err := grpcli.CreateCluster(ctx, createClusterReq)
	if err != nil {
		return nil, err
	}

	billingCtx := s.billingModel.User(s.CurrentUser(ctx))

	if req.IsDevTier {
		if err := billingCtx.EnableOrgDevTier(); err != nil {
			// skip error
			s.Log(ctx).Error("EnableOrgDevTier", zap.Uint64("tenantID", req.OrgId), zap.Error(err))
		}
	}

	resp := &central.CreateClusterResp{
		Id:       createClusterResp.Id,
		BaseResp: &central.BaseResp{},
	}
	return resp, nil
}

func (s *clusterService) ScaleClusterV2(ctx context.Context, req *central.ScaleClusterReq) (*central.DefaultResp, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	scaleClusterReq := &core.ScaleClusterReq{
		ClusterId:           req.ClusterId,
		NewComponents:       cvt.ClusterComponentsCentral2Core(req.Components),
		AllowScaleIn:        false,
		AllowScaleSecondary: false,
		BaseReq:             nil,
	}
	_, err = grpcli.ScaleCluster(ctx, scaleClusterReq)
	if err != nil {
		return nil, err
	}
	resp := &central.DefaultResp{
		BaseResp: &central.BaseResp{},
	}
	return resp, nil
}

func (s *clusterService) CalcClusterFee(ctx context.Context, req *central.CalcClusterFeeReq) (*central.CalcClusterFeeResp, error) {
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	calcClusterFeeReq := &core.CalcClusterFeeReq{
		TenantId:          req.OrgId,
		ProjectId:         req.ProjectId,
		CloudProviderName: req.Provider,
		RegionName:        req.Region,
		Duration:          int64(time.Hour),
		DeployType:        getClusterDeployType(req.IsDevTier),
		Components:        cvt.ClusterComponentsCentral2Core(req.Components),
	}
	calcClusterFeeResp, err := grpcli.CalcClusterFee(ctx, calcClusterFeeReq)
	if err != nil {
		return nil, err
	}

	costs := make([]*central.CalcClusterFeeResp_ComponentCost, 0, len(calcClusterFeeResp.Costs))
	for _, cost := range calcClusterFeeResp.Costs {
		c := &central.CalcClusterFeeResp_ComponentCost{
			ComponentType: cvt.ComponentTypeCore2Central(cost.ComponentType),
			Compute:       cost.Compute,
			Storage:       cost.Storage,
			RawCompute:    cost.RawCompute,
			RawStorage:    cost.RawStorage,
		}
		costs = append(costs, c)
	}
	resp := &central.CalcClusterFeeResp{
		Costs:    costs,
		BaseResp: &central.BaseResp{},
	}
	return resp, nil
}

func (s *clusterService) ValidateProjectCIDR(ctx context.Context, req *central.CreateProjectCIDRReq) (*types.Empty, error) {
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	userContext := s.userCluster(ctx)
	err := userContext.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	_, err = validateCreateProjectCIDRReq(ctx, req)
	if err != nil {
		return nil, err
	}

	return &types.Empty{}, nil
}

func buildSecondaryClusterProfiles(ctx context.Context, logger *zap.Logger, cluster *core.Cluster, deployType core.DeployType, cloudProvider *core.CloudProvider) ([]*central.SecondaryClusterProfile, bizerr.BizError) {
	SecondaryClusterProfiles := []*central.SecondaryClusterProfile{}
	cloudProviderName := cluster.CloudProviderName

	for regionName, region := range cloudProvider.Name2Region {
		listNodeProfiles4TenantReq := &core.ListNodeProfiles4TenantReq{
			TenantId:          cluster.TenantId,
			CloudProviderName: cloudProviderName,
			RegionName:        regionName,
			DeployType:        deployType,
		}

		ListNodeProfiles4TenantResp, err := grpcli.ListNodeProfiles4Tenant(ctx, listNodeProfiles4TenantReq)
		if err != nil {
			return nil, err
		}
		logger.Info("get ListNodeProfiles4TenantResp", zap.Any("ListNodeProfiles4TenantResp", ListNodeProfiles4TenantResp))

		nodeProfilesMap := make(map[string]*core.NodeProfile4Tenant)
		for _, nodeProfile := range ListNodeProfiles4TenantResp.Profiles {
			uniqKey := fmt.Sprintf("%s-%s", nodeProfile.ComponentType, nodeProfile.Name)
			nodeProfilesMap[uniqKey] = nodeProfile
		}
		logger.Info("get nodeProfilesMap", zap.Any("nodeProfilesMap", nodeProfilesMap))

		nodeProfiles := make([]*core.NodeProfile4Tenant, 0, len(cluster.Components))
		skipRegion := false
		for _, component := range cluster.Components {
			if !config.UserVisibleClusterComponentsContains(component.ComponentType) {
				continue
			}
			uniqKey := fmt.Sprintf("%s-%s", component.ComponentType, component.NodeProfileName)
			logger.Info("get uniqKey", zap.Any("uniqKey", uniqKey))
			nodeProfile, ok := nodeProfilesMap[uniqKey]
			if !ok {
				skipRegion = true
				break
			}
			nodeProfiles = append(nodeProfiles, nodeProfile)
		}

		if skipRegion {
			continue
		}

		logger.Info("get nodeProfiles", zap.Any("nodeProfiles", nodeProfiles))
		centralComponentNodeProfiles := make([]*central.ComponentNodeProfiles, 0, len(nodeProfiles))
		for _, nodeProfile := range nodeProfiles {
			centralComponentNodeProfiles = append(centralComponentNodeProfiles, &central.ComponentNodeProfiles{
				ComponentType: cvt.ComponentTypeCore2Central(nodeProfile.ComponentType),
				Profiles:      []*central.ClusterNodeProfile{cvt.NodeProfileCore2Central(nodeProfile)},
			})
		}

		secondaryClusterProfile := &central.SecondaryClusterProfile{
			Region: &central.Region{
				Name:        regionName,
				DisplayName: region.DisplayName,
			},
			NodeProfiles: centralComponentNodeProfiles,
		}
		SecondaryClusterProfiles = append(SecondaryClusterProfiles, secondaryClusterProfile)
	}
	return SecondaryClusterProfiles, nil
}

func (s *clusterService) GetSecondaryClusterProfiles(ctx context.Context, req *central.ClusterGetReq) (*central.GetSecondaryClusterProfilesResp, error) {
	logger := s.Log(ctx)
	context := s.userCluster(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	primaryClusterID := entity.ID(req.Id)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	// get primary cluster
	cluster, err := grpcli.GetCluster(ctx, orgID.Uint64(), projectID.Uint64(), primaryClusterID.Uint64())
	if err != nil {
		logger.Error("failed to get primary cluster", zap.Uint64("clusterID", req.Id), zap.Error(err))
		return nil, err
	}

	resp := &central.GetSecondaryClusterProfilesResp{
		BaseResp: &central.BaseResp{},
	}
	deployType := getClusterDeployType(cluster.IsDevTier)

	// disable CDC replication for dev tier by default
	if deployType == core.DeployType_DEPLOY_TYPE_SHARED {
		return resp, nil
	}

	// get regions
	tenantID := orgID.Uint64()
	cloudProviderName := cluster.CloudProviderName
	listCloudProvidersReq := &core.ListCloudProvidersReq{
		SpecificTenantId:            &tenantID,
		FilterOutEmptyCloudProvider: true,
		FilterOutEmptyRegion:        true,
		DeployType:                  &deployType,
		FilterOutInactiveRegion:     true,
		FilterOutInactiveProvider:   true,
	}
	ListCloudProvidersResp, err := grpcli.ListCloudProviders(ctx, listCloudProvidersReq)
	if err != nil {
		return nil, err
	}
	cloudProvider, ok := ListCloudProvidersResp.Name2CloudProvider[cloudProviderName]
	if !ok {
		return resp, nil
	}
	logger.Info("get cloud provider", zap.Any("cloudProvider", cloudProvider))

	// build SecondaryClusterProfiles
	secondaryClusterProfiles, err := buildSecondaryClusterProfiles(ctx, logger, cluster, deployType, cloudProvider)
	if err != nil {
		return nil, err
	}
	sort.Slice(secondaryClusterProfiles, func(i, j int) bool {
		return secondaryClusterProfiles[i].Region.DisplayName < secondaryClusterProfiles[j].Region.DisplayName
	})
	logger.Info("get secondaryClusterProfiles", zap.Any("secondaryClusterProfiles", secondaryClusterProfiles))

	resp.SecondaryClusterProfiles = secondaryClusterProfiles

	return resp, nil
}