package cluster

import (
	"context"
	"math"

	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/proto"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cluster"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/tidbgroup"
	svccommon "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/common"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/spu"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/featuregate"
	spuutil "github.com/tidbcloud/tidb-management-service/utilities/spu"
)

func (s *serviceImpl) BuildDesiredComponents(
	ctx context.Context,
	req *globaldedicatedv1.BuildDesiredComponentsRequest,
) (*globaldedicatedv1.BuildDesiredComponentsResponse, bizerr.BizError) {
	var resClusters []*entity.Cluster

	switch target := req.GetTarget().(type) {
	case *globaldedicatedv1.BuildDesiredComponentsRequest_CreateCluster:
		dummyCluster := &entity.Cluster{Cluster: &globaldedicatedv1.Cluster{
			OrganizationId:    target.CreateCluster.GetTenantId(),
			ProjectId:         target.CreateCluster.GetProjectId(),
			CloudProviderId:   target.CreateCluster.GetCloudProvider(),
			RegionId:          target.CreateCluster.GetRegion(),
			ClusterComponents: target.CreateCluster.GetClusterComponents(),
		}}
		desiredComps, err := s.buildDesiredComponents(
			ctx,
			[]*entity.TidbNodeGroup{{}},
			dummyCluster,
			target.CreateCluster.GetClusterComponents(),
			true,
		)
		if err != nil {
			return nil, err
		}
		dummyCluster.DesiredComponents = desiredComps
		resClusters = append(resClusters, dummyCluster)

	case *globaldedicatedv1.BuildDesiredComponentsRequest_ModifyCluster:
		cluster, err := s.clusterRepo.GetAndLoadComponents(ctx, target.ModifyCluster.GetClusterId())
		if err != nil {
			return nil, err
		}
		tidbGroups, err := s.tidbGroupRepo.Query(ctx, &tidbgroup.Filter{
			ClusterIds: []uint64{cluster.ClusterId},
		})
		if err != nil {
			return nil, err
		}
		if len(tidbGroups) == 0 {
			return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "empty tidb group")
		}
		cluster.DesiredComponents, err = s.buildDesiredComponents(
			ctx,
			tidbGroups,
			cluster,
			target.ModifyCluster.GetRequestedClusterComponents(),
			false,
		)
		if err != nil {
			return nil, err
		}
		resClusters = append(resClusters, cluster)

	case *globaldedicatedv1.BuildDesiredComponentsRequest_Scan_:
		clusters, err := s.clusterRepo.QueryAndLoadComponents(ctx, &clusterrepo.Filter{
			ClusterIds:         target.Scan.GetClusterIds(),
			ProjectIds:         target.Scan.GetProjectIds(),
			OrganizationIds:    target.Scan.GetTenantIds(),
			RegionNames:        target.Scan.GetRegions(),
			CloudProviderNames: target.Scan.GetCloudProviders(),
		})
		if err != nil {
			return nil, err
		}
		resClusters = append(resClusters, clusters...)
	}

	diffs := lo.Map(
		resClusters,
		func(c *entity.Cluster, _ int) *globaldedicatedv1.BuildDesiredComponentsResponse_ClusterDiff {
			diff := &globaldedicatedv1.BuildDesiredComponentsResponse_ClusterDiff{
				ClusterId: c.ClusterId,
			}

			buildDefaultComponents := func(
				effectiveComp, effectiveTikvComp *globaldedicatedv1.Cluster_ClusterComponent,
				sku, tikvSku *entity.Sku,
			) *globaldedicatedv1.Cluster_ClusterComponent {
				// The default PD component is defined by the effective TiKV component.
				if effectiveComp.GetComponentType() == globaldedicatedv1.ComponentType_PD && effectiveTikvComp != nil {
					dummyDesiredComps := []*entity.DesiredClusterComponent{
						{
							ClusterComponent: effectiveTikvComp,
							Sku:              tikvSku,
						},
					}
					const newCluster = true
					const preventPdScaleDown = false
					defaultPd, changed, err := s.buildDesiredPd(
						ctx,
						c,
						dummyDesiredComps,
						newCluster,
						preventPdScaleDown,
					)
					if err != nil {
						tilog.Error(
							ctx,
							"failed to build default PD component",
							zap.Error(err),
							zap.Uint64("cluster", c.ClusterId),
						)
					} else if changed {
						sku = defaultPd.Sku
					}
				}
				defaultComp := &globaldedicatedv1.Cluster_ClusterComponent{
					ComponentType:      sku.GetComponentType(),
					SpuName:            sku.GetSpuName(),
					SpuKey:             sku.GetSpuKey(),
					SpuDisplayName:     sku.GetSpuDisplayName(),
					SpuVersion:         sku.GetSpuVersion(),
					SkuKey:             sku.GetKey(),
					Count:              lo.ToPtr(uint32(sku.GetDefaultNodeCount())),
					Storages:           make(map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage),
					TidbGroupId:        effectiveComp.GetTidbGroupId(),
					IsDefaultTidbGroup: effectiveComp.GetIsDefaultTidbGroup(),
					InstanceType:       sku.GetInstanceType(),
					CpuNum:             uint32(sku.GetCpuCore()),
					MemorySizeGi:       uint32(sku.GetMemoryGi()),
					InstanceCpuArch:    sku.GetCpuArch(),
				}
				if storage := sku.GetDataStorage(); storage.GetStorageMuId() > 0 {
					defaultComp.Storages[globaldedicatedv1.Sku_DATA.String()] = &globaldedicatedv1.Cluster_ClusterComponent_Storage{
						DiskType:        storage.GetDiskType(),
						SizeGi:          uint32(storage.GetSizeGiDefault()),
						Iops:            uint32(storage.GetIopsFixed().GetIops()),
						ThroughputMibps: uint32(storage.GetThroughputFixed().GetThroughputMibps()),
						VolumeType:      lo.ToPtr(storage.GetVolumeType()),
					}
				}
				if storage := sku.GetRaftlogStorage(); storage.GetStorageMuId() > 0 {
					defaultComp.Storages[globaldedicatedv1.Sku_DATA.String()] = &globaldedicatedv1.Cluster_ClusterComponent_Storage{
						DiskType:        storage.GetDiskType(),
						SizeGi:          uint32(storage.GetSizeGiDefault()),
						Iops:            uint32(storage.GetIopsFixed().GetIops()),
						ThroughputMibps: uint32(storage.GetThroughputFixed().GetThroughputMibps()),
						VolumeType:      lo.ToPtr(storage.GetVolumeType()),
					}
				}
				return defaultComp
			}

			// Build current component results.
			diff.CurrentEffectiveComponents = c.GetClusterComponents()
			currentEffectiveTikvComp, ok := lo.Find(
				diff.CurrentEffectiveComponents,
				func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
				},
			)
			if !ok {
				tilog.Warn(ctx, "current effective tikv component not found", zap.Uint64("cluster", c.ClusterId))
			}
			diff.CurrentDefaultComponents = lo.Map(
				c.GetClusterComponents(),
				func(comp *globaldedicatedv1.Cluster_ClusterComponent, _ int) *globaldedicatedv1.Cluster_ClusterComponent {
					sku, ok := c.CurrentComponentSkus[comp.GetSkuKey()]
					if !ok {
						tilog.Warn(
							ctx,
							"current component sku not found",
							zap.Uint64("cluster", c.ClusterId),
							zap.String("component", comp.ComponentType.String()),
						)
						return &globaldedicatedv1.Cluster_ClusterComponent{}
					}
					return buildDefaultComponents(
						comp,
						currentEffectiveTikvComp,
						sku,
						c.CurrentComponentSkus[currentEffectiveTikvComp.GetSkuKey()],
					)
				},
			)

			// Build desired component results.
			diff.DesiredEffectiveComponents = lo.Map(
				c.DesiredComponents,
				func(comp *entity.DesiredClusterComponent, _ int) *globaldedicatedv1.Cluster_ClusterComponent {
					return comp.ClusterComponent
				},
			)
			desiredEffectiveTikvComp, ok := lo.Find(
				c.DesiredComponents,
				func(comp *entity.DesiredClusterComponent) bool {
					return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
				},
			)
			if !ok && len(c.DesiredComponents) > 0 {
				tilog.Warn(ctx, "desired effective tikv component not found", zap.Uint64("cluster", c.ClusterId))
			}
			diff.DesiredDefaultComponents = lo.Map(
				c.DesiredComponents,
				func(comp *entity.DesiredClusterComponent, _ int) *globaldedicatedv1.Cluster_ClusterComponent {
					return buildDefaultComponents(
						comp.ClusterComponent,
						lo.FromPtr(desiredEffectiveTikvComp).ClusterComponent,
						comp.Sku,
						lo.FromPtr(desiredEffectiveTikvComp).Sku,
					)
				},
			)

			return diff
		},
	)

	return &globaldedicatedv1.BuildDesiredComponentsResponse{ClusterDiffs: diffs}, nil
}

func (s *serviceImpl) buildDesiredComponents(
	ctx context.Context,
	tidbGroups []*entity.TidbNodeGroup,
	cluster *entity.Cluster,
	requestComponents []*globaldedicatedv1.Cluster_ClusterComponent,
	newCluster bool,
) ([]*entity.DesiredClusterComponent, bizerr.BizError) {
	desiredComps := make([]*entity.DesiredClusterComponent, 0, len(requestComponents))

	// Build desired components.
	for _, requestComp := range requestComponents {
		if newCluster && requestComp.GetCount() == 0 {
			tilog.Warn(
				ctx,
				"no current component and req count is 0",
				zap.Uint64("cluster", cluster.ClusterId),
				zap.String("component", requestComp.ComponentType.String()),
			)
			continue
		}

		if _, ok := entity.ComponentsHaveExtraGroup[requestComp.ComponentType]; !ok && requestComp.TidbGroupId != 0 {
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"%s component does not support tidb node group parameter",
				requestComp.ComponentType,
			)
		}

		requestTidbGroup, ok := lo.Find(tidbGroups, func(item *entity.TidbNodeGroup) bool {
			// create cluster, the tidb group is not specified, and the cluster has only one default tidb group.
			if len(tidbGroups) == 1 {
				if requestComp.TidbGroupId == 0 {
					return true
				}
			}
			// non-extra-group component, return the default tidb group.
			if _, ok := entity.ComponentsHaveExtraGroup[requestComp.ComponentType]; !ok {
				return item.Default
			}
			// when scale tidb or tiproxy, if tidb group id is not specified, return the default tidb group.
			if requestComp.TidbGroupId == 0 {
				return item.Default
			}
			// scale cluster, and specify tidb or tiproxy component in the request.
			return requestComp.TidbGroupId == item.Id
		})
		if !ok {
			return nil, bizerr.New(
				bizerr.ErrCodeRecordNotFound,
				"not found tidb group: reqGroupId=%d, groups=%+v",
				requestComp.TidbGroupId,
				tidbGroups,
			)
		}
		// when scale tidb or tiproxy, if tidb group id is not specified, override with the default tidb group id.
		requestComp.TidbGroupId = requestTidbGroup.Id

		var currentComp *globaldedicatedv1.Cluster_ClusterComponent
		if !newCluster {
			// If newCluster is true, then the cluster.GetClusterComponents() is comps from the request.
			currentComp, _ = lo.Find(
				cluster.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.ComponentType == requestComp.ComponentType && i.TidbGroupId == requestTidbGroup.Id
				},
			)
		}

		if currentComp == nil && requestComp.GetCount() == 0 {
			tilog.Warn(
				ctx,
				"no current component and req count is 0",
				zap.Uint64("cluster", cluster.ClusterId),
				zap.String("component", requestComp.ComponentType.String()),
			)
			continue
		}

		// Initialize desiredComponent.
		var desiredComp *globaldedicatedv1.Cluster_ClusterComponent
		if currentComp != nil {
			desiredComp = proto.CloneOf(currentComp)
		} else {
			requestComp.ClusterId = cluster.ClusterId
			desiredComp = proto.CloneOf(requestComp)
		}

		mgr := &buildManager{
			ctx:                    ctx,
			clusterSvc:             s,
			cluster:                cluster,
			requestComponent:       requestComp,
			currentComponentMayNil: currentComp,
			desiredComponent:       &entity.DesiredClusterComponent{ClusterComponent: desiredComp},
			expectCpuArch:          svccommon.GetCpuArch(s.featGateCli, cluster, requestComp, newCluster),
			expectCpuGeneration:    svccommon.GetCpuGenerationFilter(s.featGateCli, cluster, requestComp, newCluster),
		}

		var (
			dc      *entity.DesiredClusterComponent
			changed bool
			err     bizerr.BizError
		)
		switch requestComp.GetComponentType() {
		case globaldedicatedv1.ComponentType_TIDB, globaldedicatedv1.ComponentType_TIPROXY:
			dc, changed, err = buildTidbTiproxy(mgr)
		case globaldedicatedv1.ComponentType_TIKV:
			dc, changed, err = buildTiKV(mgr)
		case globaldedicatedv1.ComponentType_TIFLASH, globaldedicatedv1.ComponentType_PD:
			dc, changed, err = buildPdTiflash(mgr)
		}
		if err != nil {
			return nil, err
		}
		if changed {
			desiredComps = append(desiredComps, dc)
		}
	}

	dc, changed, err := s.buildDesiredPd(
		ctx,
		cluster,
		desiredComps,
		newCluster,
		s.featGateCli.CheckFeatureEnabled(
			featuregate.BlockPdScaleDown,
			cluster.GetOrganizationId(),
			cluster.GetProjectId(),
			cluster.GetCloudProviderId().String(),
			cluster.GetRegionId().String(),
		),
	)
	if err != nil {
		return nil, err
	}
	if changed {
		desiredComps = append(desiredComps, dc)
	}

	return desiredComps, nil
}

func buildTidbTiproxy(mgr *buildManager) (*entity.DesiredClusterComponent, bool, bizerr.BizError) {
	changedSku, err := mgr.stepSku()
	if err != nil {
		return nil, false, err
	}
	changedNodeCount := mgr.stepNodeCount()

	return mgr.desiredComponent, changedSku || changedNodeCount, nil
}

func buildTiKV(mgr *buildManager) (*entity.DesiredClusterComponent, bool, bizerr.BizError) {
	changedSku, err := mgr.stepSku()
	if err != nil {
		return nil, false, err
	}
	changedNodeCount := mgr.stepNodeCount()
	changedSizeGi := mgr.stepSizeGi(globaldedicatedv1.Sku_DATA)
	changedIops := mgr.stepIops(globaldedicatedv1.Sku_DATA, changedSku)
	changedThroughput := mgr.stepThroughput(globaldedicatedv1.Sku_DATA, changedSku)
	changedRaftSizeGi := mgr.stepSizeGi(globaldedicatedv1.Sku_RAFT_LOG)
	changedRaftIops := mgr.stepIops(globaldedicatedv1.Sku_RAFT_LOG, changedSku)
	changedRaftThroughput := mgr.stepThroughput(globaldedicatedv1.Sku_RAFT_LOG, changedSku)

	return mgr.desiredComponent, changedSku || changedNodeCount || changedSizeGi || changedIops || changedThroughput ||
		changedRaftSizeGi ||
		changedRaftIops ||
		changedRaftThroughput, nil
}

func buildPdTiflash(mgr *buildManager) (*entity.DesiredClusterComponent, bool, bizerr.BizError) {
	changedSku, err := mgr.stepSku()
	if err != nil {
		return nil, false, err
	}
	changedNodeCount := mgr.stepNodeCount()
	changedStorageSizeGi := mgr.stepSizeGi(globaldedicatedv1.Sku_DATA)
	changedIops := mgr.stepIops(globaldedicatedv1.Sku_DATA, changedSku)
	changedThroughput := mgr.stepThroughput(globaldedicatedv1.Sku_DATA, changedSku)

	return mgr.desiredComponent, changedSku || changedNodeCount || changedStorageSizeGi || changedIops ||
		changedThroughput, nil
}

func (s *serviceImpl) buildDesiredPd(
	ctx context.Context,
	cluster *entity.Cluster,
	desiredComps []*entity.DesiredClusterComponent,
	newCluster, preventPdScaleDown bool,
) (*entity.DesiredClusterComponent, bool, bizerr.BizError) {
	// PD needs to be selected in these cases:
	//   For new clusters:
	//   - If a PD component is explicitly passed, then use it.
	//   - Otherwise, select a default PD component based on the TiKV component.
	//   For existing clusters when all of these conditions are met:
	//   - If a PD component is explicitly passed, then use it.
	// 	 - If not, select a PD component based on the TiKV component only when:
	//     - There is a TiKV component being modified in the desiredComps list
	//     - The TiKV component's CPU or memory specifications are changing compared to the current configuration
	needToSelectPd := func() bool {
		_, ok := lo.Find(desiredComps, func(i *entity.DesiredClusterComponent) bool {
			return i.ComponentType == globaldedicatedv1.ComponentType_PD
		})
		if ok {
			return false
		}

		if newCluster {
			return true
		}

		desiredTiKV, ok := lo.Find(desiredComps, func(i *entity.DesiredClusterComponent) bool {
			return i.ComponentType == globaldedicatedv1.ComponentType_TIKV
		})
		if !ok {
			return false
		}

		currentTiKV, _ := lo.Find(
			cluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.ComponentType == globaldedicatedv1.ComponentType_TIKV &&
					i.TidbGroupId == desiredTiKV.GetTidbGroupId()
			},
		)

		return desiredTiKV.GetCpuNum() != currentTiKV.GetCpuNum() ||
			desiredTiKV.GetMemorySizeGi() != currentTiKV.GetMemorySizeGi()
	}

	if !needToSelectPd() {
		return nil, false, nil
	}

	// Select PD SKU based on the PD-TiKV mapping.
	desiredTiKV, _ := lo.Find(desiredComps, func(i *entity.DesiredClusterComponent) bool {
		return i.ComponentType == globaldedicatedv1.ComponentType_TIKV
	})
	pdSkus, err := s.skuSvc.SelectSkusByComponentType(ctx, globaldedicatedv1.ComponentType_PD, &spu.SelectSkusRequest{
		TenantId:      cluster.OrganizationId,
		ProjectId:     cluster.ProjectId,
		Provider:      cluster.CloudProviderId,
		Region:        cluster.RegionId,
		TiKVComponent: desiredTiKV,
	})
	if err != nil {
		return nil, false, err
	}
	pdSku := pdSkus[0]

	currentPd, _ := lo.Find(cluster.GetClusterComponents(), func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
		return i.ComponentType == globaldedicatedv1.ComponentType_PD && i.TidbGroupId == desiredTiKV.GetTidbGroupId()
	})

	// Select PD SKU **again** based on the buildManager steps.
	// mimic requested component based on default values of pdSku.
	requestPd := &globaldedicatedv1.Cluster_ClusterComponent{
		ClusterId:     cluster.GetClusterId(),
		ComponentType: globaldedicatedv1.ComponentType_PD,
		SpuName:       pdSku.GetSpuName(),
		SkuKey:        pdSku.GetKey(),
		Count:         lo.ToPtr(uint32(pdSku.GetDefaultNodeCount())),
		Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
			globaldedicatedv1.Sku_DATA.String(): {
				SizeGi:     uint32(pdSku.GetDataStorage().GetSizeGiDefault()),
				VolumeType: lo.ToPtr(pdSku.GetDataStorage().VolumeType),
			},
		},
	}
	desiredPd := &entity.DesiredClusterComponent{
		Sku: pdSku,
	}
	if currentPd != nil {
		desiredPd.ClusterComponent = proto.CloneOf(currentPd)
	} else {
		desiredPd.ClusterComponent = proto.CloneOf(requestPd)
	}

	pdMgr := &buildManager{
		ctx:                    ctx,
		clusterSvc:             s,
		cluster:                cluster,
		requestComponent:       requestPd,
		currentComponentMayNil: currentPd,
		desiredComponent:       desiredPd,
		expectCpuArch:          lo.ToPtr(desiredTiKV.GetInstanceCpuArch()),
		expectCpuGeneration:    lo.ToPtr(desiredTiKV.GetInstanceCpuGeneration()),
	}

	changedSku, err := pdMgr.stepSku()
	if err != nil {
		return nil, false, err
	}

	// Only PD scale up is allowed. Refer to spec https://pingcap.feishu.cn/wiki/JZpKwtmuFi0k4YkqqQgcugZ8nWe.
	if preventPdScaleDown {
		if spuutil.CompareComponentSize(
			&spuutil.ComponentSize{
				CPUNum: pdMgr.currentComponentMayNil.GetCpuNum(),
				MemGi:  pdMgr.currentComponentMayNil.GetMemorySizeGi(),
			},
			&spuutil.ComponentSize{
				CPUNum: pdMgr.desiredComponent.GetCpuNum(),
				MemGi:  pdMgr.desiredComponent.GetMemorySizeGi(),
			},
		) > 0 {
			pdMgr.desiredComponent = &entity.DesiredClusterComponent{
				ClusterComponent: pdMgr.currentComponentMayNil,
				Sku:              cluster.CurrentComponentSkus[pdMgr.currentComponentMayNil.GetSkuKey()],
			}
		}
	}

	changedNodeCount := pdMgr.stepNodeCount()
	changedStorageSizeGi := pdMgr.stepSizeGi(globaldedicatedv1.Sku_DATA)
	changedIops := pdMgr.stepIops(globaldedicatedv1.Sku_DATA, changedSku)
	changedThroughput := pdMgr.stepThroughput(globaldedicatedv1.Sku_DATA, changedSku)

	return pdMgr.desiredComponent, changedSku || changedNodeCount || changedStorageSizeGi || changedIops ||
		changedThroughput, nil
}

type buildManager struct {
	ctx                    context.Context
	clusterSvc             *serviceImpl
	cluster                *entity.Cluster
	requestComponent       *globaldedicatedv1.Cluster_ClusterComponent
	currentComponentMayNil *entity.ClusterComponent
	desiredComponent       *entity.DesiredClusterComponent
	expectCpuArch          *globaldedicatedv1.InstanceMu_CpuArch
	expectCpuGeneration    *globaldedicatedv1.InstanceMu_CpuGeneration
}

func (m *buildManager) stepSku() (bool, bizerr.BizError) {
	// if not set tier name, use the current tier key
	if m.requestComponent.SpuName == "" {
		m.requestComponent.SpuName = m.currentComponentMayNil.GetSpuName()
	}

	// build sku request.
	skuRequest := &spu.SelectSkusRequest{
		TenantId:       m.cluster.OrganizationId,
		ProjectId:      m.cluster.ProjectId,
		Provider:       m.cluster.CloudProviderId,
		Region:         m.cluster.RegionId,
		VolumeType:     nil, // fill later
		RaftVolumeType: nil, // fill later
		CpuArch:        m.expectCpuArch,
		CpuGeneration:  m.expectCpuGeneration,
		SpuName:        lo.ToPtr(m.requestComponent.SpuName),
	}
	if _, ok := entity.ComponentsHaveStorage[m.requestComponent.ComponentType]; ok {
		if m.currentComponentMayNil == nil {
			// create cluster.
			if storage := m.requestComponent.Storages[globaldedicatedv1.Sku_DATA.String()]; storage != nil {
				skuRequest.VolumeType = storage.VolumeType
			} else {
				skuRequest.VolumeType = nil
			}
		} else {
			// update cluster.
			if storage := m.requestComponent.Storages[globaldedicatedv1.Sku_DATA.String()]; storage != nil && storage.VolumeType != nil {
				skuRequest.VolumeType = storage.VolumeType
			} else {
				skuRequest.VolumeType = lo.ToPtr(m.currentComponentMayNil.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetVolumeType())
			}
		}
		if m.currentComponentMayNil == nil {
			// create cluster.
			if storage := m.requestComponent.Storages[globaldedicatedv1.Sku_RAFT_LOG.String()]; storage != nil {
				skuRequest.RaftVolumeType = storage.VolumeType
			} else {
				skuRequest.RaftVolumeType = nil
			}
		} else {
			// update cluster.
			if storage := m.requestComponent.Storages[globaldedicatedv1.Sku_RAFT_LOG.String()]; storage != nil && storage.VolumeType != nil {
				skuRequest.RaftVolumeType = storage.VolumeType
			} else {
				skuRequest.RaftVolumeType = lo.ToPtr(m.currentComponentMayNil.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()].GetVolumeType())
			}
		}
	}

	skus, err := m.clusterSvc.skuSvc.SelectSkusByComponentType(m.ctx, m.requestComponent.ComponentType, skuRequest)
	if err != nil {
		return false, err
	}
	sku := skus[0]

	// Merge sku fields to desired component.
	m.desiredComponent.Sku = sku
	m.desiredComponent.SpuName = sku.GetSpuName()
	m.desiredComponent.SpuKey = sku.GetSpuKey()
	m.desiredComponent.SpuDisplayName = sku.GetSpuDisplayName()
	m.desiredComponent.SkuKey = sku.GetKey()
	m.desiredComponent.InstanceType = sku.GetInstanceType()
	m.desiredComponent.CpuNum = uint32(sku.GetCpuCore())
	m.desiredComponent.MemorySizeGi = uint32(sku.GetMemoryGi())
	m.desiredComponent.InstanceCpuArch = sku.GetCpuArch()
	m.desiredComponent.InstanceCpuGeneration = sku.GetCpuGeneration()
	if len(m.desiredComponent.GetStorages()) == 0 {
		m.desiredComponent.Storages = make(map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage)
	}
	if m.desiredComponent.GetStorages()[globaldedicatedv1.Sku_DATA.String()] == nil {
		m.desiredComponent.Storages[globaldedicatedv1.Sku_DATA.String()] = &globaldedicatedv1.Cluster_ClusterComponent_Storage{}
	}
	m.desiredComponent.GetStorages()[globaldedicatedv1.Sku_DATA.String()].VolumeType = lo.EmptyableToPtr(
		sku.GetDataStorage().GetVolumeType(),
	)
	if m.desiredComponent.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()] == nil {
		m.desiredComponent.Storages[globaldedicatedv1.Sku_RAFT_LOG.String()] = &globaldedicatedv1.Cluster_ClusterComponent_Storage{}
	}
	m.desiredComponent.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()].VolumeType = lo.EmptyableToPtr(
		sku.GetRaftlogStorage().GetVolumeType(),
	)

	// check whether sku changed.
	if m.currentComponentMayNil != nil {
		if sku.Key == m.currentComponentMayNil.GetSkuKey() {
			tilog.Info(
				m.ctx,
				"select the same tier unit",
				zap.Uint64("cluster", m.cluster.ClusterId),
				zap.String("component", m.requestComponent.ComponentType.String()),
			)
			return false, nil
		}
	}

	return true, nil
}

func (m *buildManager) stepNodeCount() bool {
	if m.requestComponent.Count == nil {
		return false
	}
	if m.currentComponentMayNil != nil {
		if *m.requestComponent.Count == m.currentComponentMayNil.GetCount() {
			return false
		}
	}

	m.desiredComponent.Count = m.requestComponent.Count

	return true
}

func (m *buildManager) stepSizeGi(diskType globaldedicatedv1.Sku_DiskType) bool {
	reqSize := m.requestComponent.GetStorages()[diskType.String()].GetSizeGi()
	curSize := m.currentComponentMayNil.GetStorages()[diskType.String()].GetSizeGi()

	var skuStorage *globaldedicatedv1.Sku_Storage
	switch diskType {
	case globaldedicatedv1.Sku_DATA:
		skuStorage = m.desiredComponent.Sku.GetDataStorage()
	case globaldedicatedv1.Sku_RAFT_LOG:
		skuStorage = m.desiredComponent.Sku.GetRaftlogStorage()
	}

	if _, ok := m.desiredComponent.GetStorages()[diskType.String()]; !ok {
		m.desiredComponent.Storages[diskType.String()] = &globaldedicatedv1.Cluster_ClusterComponent_Storage{}
	}

	switch {
	case reqSize == 0 && curSize == 0:
		// create component and not specify size, use sku default size.
		m.desiredComponent.GetStorages()[diskType.String()].SizeGi = uint32(skuStorage.GetSizeGiDefault())
	case reqSize == 0:
		// update component and not specify size, use current size.
		m.desiredComponent.GetStorages()[diskType.String()].SizeGi = curSize
	default:
		m.desiredComponent.GetStorages()[diskType.String()].SizeGi = reqSize
	}

	return m.desiredComponent.GetStorages()[diskType.String()].SizeGi != curSize
}

// if the iops is not specified in req, and selected sku changed, then the iops could be changed at the same time.
func (m *buildManager) stepIops(diskType globaldedicatedv1.Sku_DiskType, skuChanged bool) bool {
	curIops := m.currentComponentMayNil.GetStorages()[diskType.String()].GetIops()
	curAdditionIops := m.currentComponentMayNil.GetStorages()[diskType.String()].GetAdditionalIops()
	reqIops := m.requestComponent.GetStorages()[diskType.String()].GetIops()
	var reqAdditionIops *uint32
	if reqStorage := m.requestComponent.GetStorages()[diskType.String()]; reqStorage != nil {
		reqAdditionIops = reqStorage.AdditionalIops
	}
	desiredStorage := m.desiredComponent.GetStorages()[diskType.String()]
	var skuStorage *globaldedicatedv1.Sku_Storage
	switch diskType {
	case globaldedicatedv1.Sku_DATA:
		skuStorage = m.desiredComponent.Sku.GetDataStorage()
	case globaldedicatedv1.Sku_RAFT_LOG:
		skuStorage = m.desiredComponent.Sku.GetRaftlogStorage()
	}

	calcSkuIops := func(algo *globaldedicatedv1.Sku_Storage_IopsDynamic) {
		desiredStorage.Iops = uint32(
			math.Ceil(
				float64(
					algo.GetIopsPerSizeGi(),
				) * float64(
					m.desiredComponent.GetStorages()[diskType.String()].GetSizeGi(),
				),
			),
		)
		desiredStorage.Iops = clamp(
			desiredStorage.Iops,
			uint32(skuStorage.GetIopsMin()),
			uint32(skuStorage.GetIopsMax()),
		)

		// handle additional iops
		// if reqAdditionIops is specified, use it.
		// else if current additional iops > 0, keep it unchanged.
		// else no additional iops.
		if reqAdditionIops != nil {
			desiredStorage.Iops += *reqAdditionIops
			desiredStorage.AdditionalIops = reqAdditionIops
		} else if curAdditionIops > 0 {
			desiredStorage.Iops += curAdditionIops
			desiredStorage.AdditionalIops = lo.ToPtr(curAdditionIops)
		}

		if desiredStorage.AdditionalIops != nil {
			desiredStorage.Iops = clamp(
				desiredStorage.Iops,
				uint32(skuStorage.GetIopsMin()),
				uint32(skuStorage.GetIopsMax()),
			)
		}
	}

	if skuStorage.GetIopsFixed() != nil {
		// Fixed IOPS algorithm
		switch {
		case reqIops != 0:
			// If requested IOPS is specified, use it
			desiredStorage.Iops = reqIops
		case skuChanged:
			skuDefaultIops := uint32(skuStorage.GetIopsFixed().GetIops())
			if m.isTikvPlusDataDisk(diskType) && m.preserveTiKVProvisionedIOPS() {
				// For the data disk of TiKV Plus storage type, keep the larger of the current IOPS and the new SKU
				// default IOPS, so scaling up|down TiKV does not lower a previously provisioned IOPS.
				desiredStorage.Iops = max(curIops, skuDefaultIops)
			} else {
				// Use the fixed iops from SKU
				desiredStorage.Iops = skuDefaultIops
			}
		default:
			// Otherwise, keep the current iops unchanged
			desiredStorage.Iops = curIops
		}
	} else if skuStorage.GetIopsDynamic() != nil {
		// Dynamic IOPS algorithm - always calculate based on size
		calcSkuIops(skuStorage.GetIopsDynamic())
	}

	return curIops != desiredStorage.GetIops()
}

// isTikvPlusDataDisk reports whether the desired component uses TiKV Plus storage type, and the given
// disk type is DATA. TiKV Plus storage type is a DATA disk backed by an IO2 volume with a fixed IOPS SKU and
// no separate raft-log storage.
func (m *buildManager) isTikvPlusDataDisk(diskType globaldedicatedv1.Sku_DiskType) bool {
	if diskType != globaldedicatedv1.Sku_DATA {
		return false
	}

	if m.desiredComponent.Sku.GetComponentType() != globaldedicatedv1.ComponentType_TIKV {
		return false
	}

	dataStorage := m.desiredComponent.Sku.GetDataStorage()
	raftStorage := m.desiredComponent.Sku.GetRaftlogStorage()
	return dataStorage.GetVolumeType() == globaldedicatedv1.StorageMu_IO2 &&
		dataStorage.GetIopsFixed() != nil &&
		raftStorage.GetVolumeType() == globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED
}

func (m *buildManager) preserveTiKVProvisionedIOPS() bool {
	return m.clusterSvc.featGateCli.CheckFeatureEnabled(
		featuregate.PreserveTiKVProvisionedIOPS,
		m.cluster.GetOrganizationId(),
		m.cluster.GetProjectId(),
		m.cluster.GetCloudProviderId().String(),
		m.cluster.GetRegionId().String(),
	)
}

// if the throughput is not specified in req, and selected sku changed, then the throughput could be changed at the same time.
func (m *buildManager) stepThroughput(diskType globaldedicatedv1.Sku_DiskType, skuChanged bool) bool {
	curThroughput := m.currentComponentMayNil.GetStorages()[diskType.String()].GetThroughputMibps()
	reqThroughput := m.requestComponent.GetStorages()[diskType.String()].GetThroughputMibps()
	desiredStorage := m.desiredComponent.GetStorages()[diskType.String()]
	var skuStorage *globaldedicatedv1.Sku_Storage
	switch diskType {
	case globaldedicatedv1.Sku_DATA:
		skuStorage = m.desiredComponent.Sku.GetDataStorage()
	case globaldedicatedv1.Sku_RAFT_LOG:
		skuStorage = m.desiredComponent.Sku.GetRaftlogStorage()
	}

	calcSkuThroughput := func(algo *globaldedicatedv1.Sku_Storage_ThroughputDynamic) {
		t1 := math.Ceil(
			float64(
				algo.GetThroughputMibpsPerSizeGi(),
			) * float64(
				m.desiredComponent.GetStorages()[diskType.String()].GetSizeGi(),
			),
		)
		t2 := math.Ceil(
			float64(
				algo.GetThroughputMibpsPerIops(),
			) * float64(
				m.desiredComponent.GetStorages()[diskType.String()].GetIops(),
			),
		)
		desiredStorage.ThroughputMibps = uint32(max(t1, t2))
		desiredStorage.ThroughputMibps = clamp(
			desiredStorage.ThroughputMibps,
			uint32(skuStorage.GetThroughputMibpsMin()),
			uint32(skuStorage.GetThroughputMibpsMax()),
		)
	}

	if skuStorage.GetThroughputFixed() != nil {
		switch {
		case reqThroughput != 0:
			// If requested throughput is specified, use it
			desiredStorage.ThroughputMibps = reqThroughput
		case skuChanged:
			// If sku changed, use the fixed throughput from SKU
			desiredStorage.ThroughputMibps = uint32(skuStorage.GetThroughputFixed().GetThroughputMibps())
		default:
			// Otherwise, keep the current throughput unchanged
			desiredStorage.ThroughputMibps = curThroughput
		}
	} else if skuStorage.GetThroughputDynamic() != nil {
		// Dynamic throughput algorithm - always calculate based on size and current IOPS
		calcSkuThroughput(skuStorage.GetThroughputDynamic())
	}

	return curThroughput != desiredStorage.GetThroughputMibps()
}

func clamp(v, minv, maxv uint32) uint32 {
	if minv != 0 && v < minv {
		return minv
	}
	if maxv != 0 && v > maxv {
		return maxv
	}
	return v
}