package cluster

import (
	"context"
	"encoding/json"
	"fmt"
	"math/rand/v2"
	"strconv"
	"testing"

	"github.com/samber/lo"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	clusterdedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"go.uber.org/mock/gomock"
	"google.golang.org/protobuf/proto"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/account"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/account/mock_account"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/billing"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/billing/mock_billing"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/clusterconfig/mock_clusterconfig"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/clustermetric/mock_clustermetric"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/regionserver/mock_regionserver"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/remoteencryptor/mock_remoteencryptor"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	skusvc "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/spu"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/dag"
	clusterrepoimpl "github.com/tidbcloud/tidb-management-service/dedicated-global-service/infra/repoimpl/cluster"
	spurepoimpl "github.com/tidbcloud/tidb-management-service/dedicated-global-service/infra/repoimpl/spu"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	featgatesvc "github.com/tidbcloud/tidb-management-service/utilities/featuregate"
	"github.com/tidbcloud/tidb-management-service/utilities/featuregate/mock_featuregate"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow/mock_workflow"
)

func TestScaleCluster(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID:             ProjectID,
		TenantID:       TenantID,
		Name:           "test",
		AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID:   TenantID,
		Name: "test",
		Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), &regionaldedicatedv1.ListVpcsRequest{
		TenantId:  strconv.FormatUint(TenantID, 10),
		ProjectId: strconv.FormatUint(ProjectID, 10),
		Statuses: []regionaldedicatedv1.Vpc_Status{
			regionaldedicatedv1.Vpc_ACTIVE,
			regionaldedicatedv1.Vpc_CREATING,
			regionaldedicatedv1.Vpc_FAILED,
		},
	}).Return(&regionaldedicatedv1.ListVpcsResponse{
		Vpcs: []*regionaldedicatedv1.Vpc{
			{
				VpcId:      "vpc-xxxxxxxxxxxxxxxxx",
				TenantIds:  []string{strconv.FormatUint(TenantID, 10)},
				ProjectIds: []string{strconv.FormatUint(ProjectID, 10)},
				CidrBlock:  "0.0.0.0/0",
			},
		},
	}, nil).AnyTimes()
	regionalCli.EXPECT().GetCluster(gomock.Any(), gomock.Any(), gomock.Any()).Return(&clusterdedicatedv1.Cluster{
		ProxyBaseUrl: "http://127.0.0.1:8080",
	}, nil).AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().
		BillingActionCheck(gomock.Any(), billing.Action_Create, uint64(TenantID), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).
		AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().
		ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).
		AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, true)

	t.Run("change pd/tidb/tikv/tiflash SPU at the same time", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:          555,
								Iops:            555,
								ThroughputMibps: 555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdTidbComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// scale the cluster.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIDB,
					SpuName:       "7c7g",
					TidbGroupId:   createdTidbComp.GetTidbGroupId(),
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "7c7g",
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
					SpuName:       "7c7g",
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// skus of tidb/tikv/tiflash should change to the expected ones.
		scaledTidbComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledTidbComp.GetSpuName())
		assert.Equal(t, "tidb-7c7g-arm", scaledTidbComp.GetSkuKey())
		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledTikvComp.GetSpuName())
		assert.Equal(t, "tikv-7c7g-arm-data:gp3-raft:none-fixed", scaledTikvComp.GetSkuKey())
		scaledTiflash, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIFLASH
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledTiflash.GetSpuName())
		assert.Equal(t, "tiflash-7c7g-arm-data:gp3-dynamic", scaledTiflash.GetSkuKey())

		// pd sku should override the condition-mapped sku.
		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledPd.GetSpuName())
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())

		// check multi-separated dag instances are created.
		dagIns, err_ := svc.workflow.ListRunningDagInstancesByNamePrefix(
			t.Context(),
			"dynamic_dag__global_scale_",
			strconv.FormatUint(createdCluster.GetClusterId(), 10),
		)
		require.NoError(t, err_)
		assert.Len(t, dagIns, 1)
	})

	t.Run("no current component and request 0 count should be ignored", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdTidbComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// scale the cluster.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIDB,
					SpuName:       "7c7g",
					TidbGroupId:   createdTidbComp.GetTidbGroupId(),
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "7c7g",
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// skus of tidb/tikv/tiflash should change to the expected ones.
		scaledTidbComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledTidbComp.GetSpuName())
		assert.Equal(t, "tidb-7c7g-arm", scaledTidbComp.GetSkuKey())
		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledTikvComp.GetSpuName())
		assert.Equal(t, "tikv-7c7g-arm-data:gp3-raft:none-fixed", scaledTikvComp.GetSkuKey())
		_, ok = lo.Find(scaledCluster.GetClusterComponents(), func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
			return i.GetComponentType() == globaldedicatedv1.ComponentType_TIFLASH
		})
		require.False(t, ok)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledPd.GetSpuName())
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())

		// check multi-separated dag instances are created.
		dagIns, err_ := svc.workflow.ListRunningDagInstancesByNamePrefix(
			t.Context(),
			"dynamic_dag__global_scale_",
			strconv.FormatUint(createdCluster.GetClusterId(), 10),
		)
		require.NoError(t, err_)
		assert.Len(t, dagIns, 1)
	})

	t.Run("scale up tikv and pd scale up automatically", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "2c2g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "2c2g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdTidbComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		createPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", createPdComp.GetSpuName())
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", createPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// scale the cluster.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIDB,
					SpuName:       "7c7g",
					TidbGroupId:   createdTidbComp.GetTidbGroupId(),
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// skus of tidb/tikv/tiflash should change to the expected ones.
		scaledTidbComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledTidbComp.GetSpuName())
		assert.Equal(t, "tidb-7c7g-arm", scaledTidbComp.GetSkuKey())
		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledTikvComp.GetSpuName())
		assert.Equal(t, "tikv-7c7g-arm-data:gp3-raft:none-fixed", scaledTikvComp.GetSkuKey())

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "8c8g", scaledPd.GetSpuName())
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())

		// check multi-separated dag instances are created.
		dagIns, err_ := svc.workflow.ListRunningDagInstancesByNamePrefix(
			t.Context(),
			"dynamic_dag__global_scale_",
			strconv.FormatUint(createdCluster.GetClusterId(), 10),
		)
		require.NoError(t, err_)
		assert.Len(t, dagIns, 1)
	})

	t.Run("scale down tikv and pd scale down automatically", func(t *testing.T) {
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "7c7g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "7c7g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdTidbComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		createPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "8c8g", createPdComp.GetSpuName())
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// scale the cluster.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIDB,
					SpuName:       "2c2g",
					TidbGroupId:   createdTidbComp.GetTidbGroupId(),
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "2c2g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// skus of tidb/tikv/tiflash should change to the expected ones.
		scaledTidbComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		assert.Equal(t, "2c2g", scaledTidbComp.GetSpuName())
		assert.Equal(t, "tidb-2c2g-arm", scaledTidbComp.GetSkuKey())
		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "2c2g", scaledTikvComp.GetSpuName())
		assert.Equal(t, "tikv-2c2g-arm-data:gp3-raft:none-fixed", scaledTikvComp.GetSkuKey())

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledPd.GetSpuName())
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())

		// check multi-separated dag instances are created.
		dagIns, err_ := svc.workflow.ListRunningDagInstancesByNamePrefix(
			t.Context(),
			"dynamic_dag__global_scale_",
			strconv.FormatUint(createdCluster.GetClusterId(), 10),
		)
		require.NoError(t, err_)
		assert.Len(t, dagIns, 1)
	})

	t.Run("scale up tikv storage size and scale in tikv at the same time", func(t *testing.T) {
		metricsBody := []byte(
			`{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1671163316.789,"30396913663"]}]}}`,
		)
		metricCli.EXPECT().
			QueryClusterMetrics(gomock.Any(), gomock.Any(), gomock.Any()).
			Return(metricsBody, nil).
			AnyTimes()

		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "7c7g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "7c7g",
						Count:         lo.ToPtr(uint32(12)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     1000,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// scale the cluster.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Count:         lo.ToPtr(uint32(9)),
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							SizeGi: 2000,
						},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, uint32(2000), scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetSizeGi())
	})
}

func TestChangeTikvSpuPlusToNonPlus(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID:             ProjectID,
		TenantID:       TenantID,
		Name:           "test",
		AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID:   TenantID,
		Name: "test",
		Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), &regionaldedicatedv1.ListVpcsRequest{
		TenantId:  strconv.FormatUint(TenantID, 10),
		ProjectId: strconv.FormatUint(ProjectID, 10),
		Statuses: []regionaldedicatedv1.Vpc_Status{
			regionaldedicatedv1.Vpc_ACTIVE,
			regionaldedicatedv1.Vpc_CREATING,
			regionaldedicatedv1.Vpc_FAILED,
		},
	}).Return(&regionaldedicatedv1.ListVpcsResponse{
		Vpcs: []*regionaldedicatedv1.Vpc{
			{
				VpcId:      "vpc-xxxxxxxxxxxxxxxxx",
				TenantIds:  []string{strconv.FormatUint(TenantID, 10)},
				ProjectIds: []string{strconv.FormatUint(ProjectID, 10)},
				CidrBlock:  "0.0.0.0/0",
			},
		},
	}, nil).AnyTimes()
	regionalCli.EXPECT().GetCluster(gomock.Any(), gomock.Any(), gomock.Any()).Return(&clusterdedicatedv1.Cluster{
		ProxyBaseUrl: "http://127.0.0.1:8080",
	}, nil).AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().
		BillingActionCheck(gomock.Any(), billing.Action_Create, uint64(TenantID), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).
		AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().
		ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).
		AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, true)

	t.Run("non-plus to plus, tikv sku and pd sku should not change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())
		createdTikvComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "tikv-8c8g-arm-data:io2-raft:none-fixed", createdTikvComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// scale to 8c8g-plus.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "8c8g-plus",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
		scaledTikv, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "tikv-8c8g-arm-data:io2-raft:none-fixed:plus", scaledTikv.GetSkuKey())
	})

	t.Run("plus to non-plus, tikv sku and pd sku should not change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g-plus",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())
		createdTikvComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "tikv-8c8g-arm-data:io2-raft:none-fixed:plus", createdTikvComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// scale to 8c8g-plus.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "8c8g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
		scaledTikv, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "tikv-8c8g-arm-data:io2-raft:none-fixed", scaledTikv.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		// switch storage type
		// scale to 8c8g-plus.
		scaleReq1 := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
						},
					},
				},
			},
		}
		scaledCluster1, err := svc.Scale(t.Context(), scaleReq1)
		require.NoError(t, err)

		scaledPd1, ok := lo.Find(
			scaledCluster1.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", scaledPd1.GetSkuKey())
		scaledTikv1, ok := lo.Find(
			scaledCluster1.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:none-fixed", scaledTikv1.GetSkuKey())
	})
}

func TestChangeTikvStorageTypeShouldNotChangePdSku(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID:             ProjectID,
		TenantID:       TenantID,
		Name:           "test",
		AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID:   TenantID,
		Name: "test",
		Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), &regionaldedicatedv1.ListVpcsRequest{
		TenantId:  strconv.FormatUint(TenantID, 10),
		ProjectId: strconv.FormatUint(ProjectID, 10),
		Statuses: []regionaldedicatedv1.Vpc_Status{
			regionaldedicatedv1.Vpc_ACTIVE,
			regionaldedicatedv1.Vpc_CREATING,
			regionaldedicatedv1.Vpc_FAILED,
		},
	}).Return(&regionaldedicatedv1.ListVpcsResponse{
		Vpcs: []*regionaldedicatedv1.Vpc{
			{
				VpcId:      "vpc-xxxxxxxxxxxxxxxxx",
				TenantIds:  []string{strconv.FormatUint(TenantID, 10)},
				ProjectIds: []string{strconv.FormatUint(ProjectID, 10)},
				CidrBlock:  "0.0.0.0/0",
			},
		},
	}, nil).AnyTimes()
	regionalCli.EXPECT().GetCluster(gomock.Any(), gomock.Any(), gomock.Any()).Return(&clusterdedicatedv1.Cluster{
		ProxyBaseUrl: "http://127.0.0.1:8080",
	}, nil).AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().
		BillingActionCheck(gomock.Any(), billing.Action_Create, uint64(TenantID), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).
		AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().
		ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).
		AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, true)

	t.Run("volume type change basic to plus", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
						},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("volume type change standard to performance", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
						},
						globaldedicatedv1.Sku_RAFT_LOG.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
						},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("volume type change basic to standard", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
						},
						globaldedicatedv1.Sku_RAFT_LOG.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
						},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("volume type change basic to performance", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
						},
						globaldedicatedv1.Sku_RAFT_LOG.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
						},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})
}

func TestChangeTikvShouldNotChangePdSku(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID:             ProjectID,
		TenantID:       TenantID,
		Name:           "test",
		AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID:   TenantID,
		Name: "test",
		Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), &regionaldedicatedv1.ListVpcsRequest{
		TenantId:  strconv.FormatUint(TenantID, 10),
		ProjectId: strconv.FormatUint(ProjectID, 10),
		Statuses: []regionaldedicatedv1.Vpc_Status{
			regionaldedicatedv1.Vpc_ACTIVE,
			regionaldedicatedv1.Vpc_CREATING,
			regionaldedicatedv1.Vpc_FAILED,
		},
	}).Return(&regionaldedicatedv1.ListVpcsResponse{
		Vpcs: []*regionaldedicatedv1.Vpc{
			{
				VpcId:      "vpc-xxxxxxxxxxxxxxxxx",
				TenantIds:  []string{strconv.FormatUint(TenantID, 10)},
				ProjectIds: []string{strconv.FormatUint(ProjectID, 10)},
				CidrBlock:  "0.0.0.0/0",
			},
		},
	}, nil).AnyTimes()
	regionalCli.EXPECT().GetCluster(gomock.Any(), gomock.Any(), gomock.Any()).Return(&clusterdedicatedv1.Cluster{
		ProxyBaseUrl: "http://127.0.0.1:8080",
	}, nil).AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().
		BillingActionCheck(gomock.Any(), billing.Action_Create, uint64(TenantID), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).
		AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().
		ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).
		AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, true)

	t.Run("data disk size change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// ops set pd to 7c7g.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		scaleReq = &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							SizeGi: 888,
						},
					},
				},
			},
		}
		scaledCluster, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("data disk iops change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// ops set pd to 7c7g.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		scaleReq = &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							Iops: 888,
						},
					},
				},
			},
		}
		scaledCluster, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("data disk additional iops change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// ops set pd to 7c7g.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		scaleReq = &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							AdditionalIops: lo.ToPtr(uint32(888)),
						},
					},
				},
			},
		}
		scaledCluster, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("data disk throughput change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// ops set pd to 7c7g.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		scaleReq = &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							ThroughputMibps: 888,
						},
					},
				},
			},
		}
		scaledCluster, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("raft disk size change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// ops set pd to 7c7g.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		scaleReq = &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_RAFT_LOG.String(): {
							SizeGi: 888,
						},
					},
				},
			},
		}
		scaledCluster, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("raft disk iops change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// ops set pd to 7c7g.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		scaleReq = &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_RAFT_LOG.String(): {
							Iops: 888,
						},
					},
				},
			},
		}
		scaledCluster, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("raft disk additional iops change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// ops set pd to 7c7g.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		scaleReq = &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_RAFT_LOG.String(): {
							AdditionalIops: lo.ToPtr(uint32(888)),
						},
					},
				},
			},
		}
		scaledCluster, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("raft disk throughput change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// ops set pd to 7c7g.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		scaleReq = &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_RAFT_LOG.String(): {
							ThroughputMibps: 888,
						},
					},
				},
			},
		}
		scaledCluster, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})

	t.Run("tikv count change", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-8c8g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// ops set pd to 7c7g.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		scaleReq = &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Count:         lo.ToPtr(uint32(6)),
				},
			},
		}
		scaledCluster, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledPd, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "pd-7c7g-arm-data:gp3-dynamic", scaledPd.GetSkuKey())
	})
}

func TestChangeStorageSpecs(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID:             ProjectID,
		TenantID:       TenantID,
		Name:           "test",
		AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID:   TenantID,
		Name: "test",
		Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), &regionaldedicatedv1.ListVpcsRequest{
		TenantId:  strconv.FormatUint(TenantID, 10),
		ProjectId: strconv.FormatUint(ProjectID, 10),
		Statuses: []regionaldedicatedv1.Vpc_Status{
			regionaldedicatedv1.Vpc_ACTIVE,
			regionaldedicatedv1.Vpc_CREATING,
			regionaldedicatedv1.Vpc_FAILED,
		},
	}).Return(&regionaldedicatedv1.ListVpcsResponse{
		Vpcs: []*regionaldedicatedv1.Vpc{
			{
				VpcId:      "vpc-xxxxxxxxxxxxxxxxx",
				TenantIds:  []string{strconv.FormatUint(TenantID, 10)},
				ProjectIds: []string{strconv.FormatUint(ProjectID, 10)},
				CidrBlock:  "0.0.0.0/0",
			},
		},
	}, nil).AnyTimes()
	regionalCli.EXPECT().
		GetCluster(gomock.Any(), gomock.Any(), gomock.Any()).
		Return(&clusterdedicatedv1.Cluster{ProxyBaseUrl: "http://127.0.0.1:8080"}, nil).
		AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().
		BillingActionCheck(gomock.Any(), billing.Action_Create, uint64(TenantID), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).
		AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().
		ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).
		AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, true)

	t.Log("currently there is no way to remove raft disk by scale api.")

	t.Run("change data:gp3/raft:io2 to data:gp3/raft:gp3", func(t *testing.T) {
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
								SizeGi:     55,
							},
						},
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:          555,
								Iops:            555,
								ThroughputMibps: 555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
						},
						globaldedicatedv1.Sku_RAFT_LOG.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
						},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "8c8g", scaledTikvComp.GetSpuName())
		assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:gp3-dynamic", scaledTikvComp.GetSkuKey())

		dataStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, dataStorage.GetVolumeType())
		assert.Equal(t, uint32(555), dataStorage.GetSizeGi())
		assert.Equal(t, uint32(999), dataStorage.GetIops())
		assert.Equal(t, uint32(278), dataStorage.GetThroughputMibps())

		raftStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, raftStorage.GetVolumeType())
		assert.Equal(t, uint32(55), raftStorage.GetSizeGi())
		assert.Equal(t, uint32(110), raftStorage.GetIops())
		assert.Equal(t, uint32(28), raftStorage.GetThroughputMibps())
	})

	t.Run("change data:gp3/raft:none to data:gp3/raft:io2", func(t *testing.T) {
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:          555,
								Iops:            555,
								ThroughputMibps: 555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3)},
						globaldedicatedv1.Sku_RAFT_LOG.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
							SizeGi:     55,
						},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "8c8g", scaledTikvComp.GetSpuName())
		assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:io2-dynamic", scaledTikvComp.GetSkuKey())

		dataStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, dataStorage.GetVolumeType())
		assert.Equal(t, uint32(555), dataStorage.GetSizeGi())
		assert.Equal(t, uint32(1110), dataStorage.GetIops())
		assert.Equal(t, uint32(278), dataStorage.GetThroughputMibps())

		raftStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_IO2, raftStorage.GetVolumeType())
		assert.Equal(t, uint32(55), raftStorage.GetSizeGi())
		assert.Equal(t, uint32(1100), raftStorage.GetIops())
		assert.Equal(t, uint32(0), raftStorage.GetThroughputMibps())
	})

	t.Run("change additional iops then scale out tidb", func(t *testing.T) {
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     200,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
								SizeGi:     55,
							},
						},
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:          555,
								Iops:            555,
								ThroughputMibps: 555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							VolumeType:     lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
							AdditionalIops: lo.ToPtr(uint32(50)),
						},
						globaldedicatedv1.Sku_RAFT_LOG.String(): {
							VolumeType:     lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
							AdditionalIops: lo.ToPtr(uint32(1)),
						},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "8c8g", scaledTikvComp.GetSpuName())
		assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:gp3-dynamic", scaledTikvComp.GetSkuKey())

		dataStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, dataStorage.GetVolumeType())
		assert.Equal(t, uint32(200), dataStorage.GetSizeGi())
		assert.Equal(t, uint32(450), dataStorage.GetIops())
		assert.Equal(t, uint32(100), dataStorage.GetThroughputMibps())

		raftStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, raftStorage.GetVolumeType())
		assert.Equal(t, uint32(55), raftStorage.GetSizeGi())
		assert.Equal(t, uint32(111), raftStorage.GetIops())
		assert.Equal(t, uint32(28), raftStorage.GetThroughputMibps())

		// mimic the cluster is in available state.
		scaledCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), scaledCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		// scale out tidb after change tikv data disk additional iops.
		// 1. no err when scale out tidb after change tikv data disk additional iops.
		// 2. tikv data disk iops and throughput are not affected after tidb scale out.
		scaleReq = &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: scaledCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIDB,
					Count:         lo.ToPtr(uint32(2)),
				},
			},
		}
		scaledCluster, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		tikvComponent, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)

		dataStorage = tikvComponent.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		assert.Equal(t, uint32(200), dataStorage.GetSizeGi())
		assert.Equal(t, uint32(450), dataStorage.GetIops())
		assert.Equal(t, uint32(100), dataStorage.GetThroughputMibps())

		raftStorage = tikvComponent.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, raftStorage.GetVolumeType())
		assert.Equal(t, uint32(55), raftStorage.GetSizeGi())
		assert.Equal(t, uint32(111), raftStorage.GetIops())
		assert.Equal(t, uint32(28), raftStorage.GetThroughputMibps())
	})

	t.Run("create cluster with additional iops then scale up tikv", func(t *testing.T) {
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType:     lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:         200,
								AdditionalIops: lo.ToPtr(uint32(50)),
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     55,
							},
						},
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:          555,
								Iops:            555,
								ThroughputMibps: 555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType:     lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								AdditionalIops: lo.ToPtr(uint32(1)),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		tikvComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "8c8g", tikvComp.GetSpuName())
		assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:gp3-dynamic", tikvComp.GetSkuKey())

		dataStorage := tikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, dataStorage.GetVolumeType())
		assert.Equal(t, uint32(200), dataStorage.GetSizeGi())
		assert.Equal(t, uint32(450), dataStorage.GetIops())
		assert.Equal(t, uint32(100), dataStorage.GetThroughputMibps())

		raftStorage := tikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, raftStorage.GetVolumeType())
		assert.Equal(t, uint32(55), raftStorage.GetSizeGi())
		assert.Equal(t, uint32(110), raftStorage.GetIops())
		assert.Equal(t, uint32(28), raftStorage.GetThroughputMibps())

		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// tikv data disk iops and throughput is not affected after tikv scale up.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "16c16g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		assert.Equal(t, "16c16g", scaledTikvComp.GetSpuName())
		assert.Equal(t, "tikv-16c16g-arm-data:gp3-raft:gp3-dynamic", scaledTikvComp.GetSkuKey())

		dataStorage = scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		require.True(t, ok)
		assert.Equal(t, uint32(200), dataStorage.GetSizeGi())
		assert.Equal(t, uint32(450), dataStorage.GetIops())
		assert.Equal(t, uint32(100), dataStorage.GetThroughputMibps())

		raftStorage = scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, raftStorage.GetVolumeType())
		assert.Equal(t, uint32(55), raftStorage.GetSizeGi())
		assert.Equal(t, uint32(110), raftStorage.GetIops())
		assert.Equal(t, uint32(28), raftStorage.GetThroughputMibps())
	})

	t.Run("explicitly set size, iops and throughput", func(t *testing.T) {
		t.Run("has raft disk", func(t *testing.T) {
			createReq := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
									SizeGi:     55,
								},
							},
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:          555,
									Iops:            555,
									ThroughputMibps: 555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			createdCluster, err := svc.Create(t.Context(), createReq)
			require.NoError(t, err)

			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)

			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:          556,
								Iops:            10,
								ThroughputMibps: 10,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								Iops:       10,
							},
						},
					},
				},
			}
			scaledCluster, err := svc.Scale(t.Context(), scaleReq)
			require.NoError(t, err)

			scaledTikvComp, ok := lo.Find(
				scaledCluster.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
				},
			)
			require.True(t, ok)
			assert.Equal(t, "8c8g", scaledTikvComp.GetSpuName())
			assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:gp3-dynamic", scaledTikvComp.GetSkuKey())

			dataStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
			assert.Equal(t, globaldedicatedv1.StorageMu_GP3, dataStorage.GetVolumeType())
			assert.Equal(t, uint32(556), dataStorage.GetSizeGi())
			assert.EqualValues(t, 999, dataStorage.GetIops())
			assert.Equal(t, uint32(278), dataStorage.GetThroughputMibps())

			raftStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]
			assert.Equal(t, globaldedicatedv1.StorageMu_GP3, raftStorage.GetVolumeType())
			assert.Equal(t, uint32(55), raftStorage.GetSizeGi())
			assert.Equal(t, uint32(110), raftStorage.GetIops())
			assert.Equal(t, uint32(28), raftStorage.GetThroughputMibps())
		})

		t.Run("no raft disk", func(t *testing.T) {
			createReq := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:          555,
									Iops:            555,
									ThroughputMibps: 555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			createdCluster, err := svc.Create(t.Context(), createReq)
			require.NoError(t, err)

			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)

			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType:      lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:          556,
								Iops:            10,
								ThroughputMibps: 10,
							},
						},
					},
				},
			}
			scaledCluster, err := svc.Scale(t.Context(), scaleReq)
			require.NoError(t, err)

			scaledTikvComp, ok := lo.Find(
				scaledCluster.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
				},
			)
			require.True(t, ok)
			assert.Equal(t, "8c8g", scaledTikvComp.GetSpuName())
			assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:none-fixed", scaledTikvComp.GetSkuKey())

			dataStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
			assert.Equal(t, globaldedicatedv1.StorageMu_GP3, dataStorage.GetVolumeType())
			assert.Equal(t, uint32(556), dataStorage.GetSizeGi())
			assert.Equal(t, uint32(10), dataStorage.GetIops())
			assert.Equal(t, uint32(10), dataStorage.GetThroughputMibps())
		})
	})

	t.Run("fixed iops change size, iops should not changed", func(t *testing.T) {
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
								Iops:       500,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {Iops: 600},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)

		dataStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, dataStorage.GetVolumeType())
		assert.Equal(t, uint32(555), dataStorage.GetSizeGi())
		assert.Equal(t, uint32(600), dataStorage.GetIops())
	})

	t.Run("dynamic iops set iops, iops should be dynamic, and plus additional iops", func(t *testing.T) {
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
								SizeGi:     55,
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {Iops: 600, AdditionalIops: lo.ToPtr(uint32(10))},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "8c8g", scaledTikvComp.GetSpuName())
		assert.Equal(t, "tikv-8c8g-arm-data:gp3-raft:io2-dynamic", scaledTikvComp.GetSkuKey())

		dataStorage := scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
		assert.Equal(t, globaldedicatedv1.StorageMu_GP3, dataStorage.GetVolumeType())
		assert.Equal(t, uint32(555), dataStorage.GetSizeGi())
		assert.Equal(t, uint32(1120), dataStorage.GetIops())
		assert.Equal(t, uint32(278), dataStorage.GetThroughputMibps())
	})

	t.Run("change sku, fixed iops should changed according to the new sku", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdTidbComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// scale the cluster.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIDB,
					SpuName:       "7c7g",
					TidbGroupId:   createdTidbComp.GetTidbGroupId(),
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "7c7g",
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_TIFLASH,
					SpuName:       "7c7g",
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "7c7g",
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// skus of tidb/tikv/tiflash should change to the expected ones.
		scaledTidbComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledTidbComp.GetSpuName())
		assert.Equal(t, "tidb-7c7g-arm", scaledTidbComp.GetSkuKey())
		scaledTikvComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledTikvComp.GetSpuName())
		assert.Equal(t, "tikv-7c7g-arm-data:gp3-raft:none-fixed", scaledTikvComp.GetSkuKey())
		assert.EqualValues(t, 2, scaledTikvComp.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetIops())
	})

	t.Run("multiple disk changes in a roll should fail", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// scale the cluster.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_IO2),
						},
					},
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// mimic the cluster is in available state.
		scaledCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		// scale the cluster.
		scaleReq1 := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {
							Iops: 888,
						},
					},
				},
			},
		}
		_, err = svc.Scale(t.Context(), scaleReq1)
		require.ErrorContains(t, err, "You've reached the maximum modification rate per storage of")
	})
}

func TestScaleTidbGroupComponent(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID:             ProjectID,
		TenantID:       TenantID,
		Name:           "test",
		AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID:   TenantID,
		Name: "test",
		Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), &regionaldedicatedv1.ListVpcsRequest{
		TenantId:  strconv.FormatUint(TenantID, 10),
		ProjectId: strconv.FormatUint(ProjectID, 10),
		Statuses: []regionaldedicatedv1.Vpc_Status{
			regionaldedicatedv1.Vpc_ACTIVE,
			regionaldedicatedv1.Vpc_CREATING,
			regionaldedicatedv1.Vpc_FAILED,
		},
	}).Return(&regionaldedicatedv1.ListVpcsResponse{
		Vpcs: []*regionaldedicatedv1.Vpc{
			{
				VpcId:      "vpc-xxxxxxxxxxxxxxxxx",
				TenantIds:  []string{strconv.FormatUint(TenantID, 10)},
				ProjectIds: []string{strconv.FormatUint(ProjectID, 10)},
				CidrBlock:  "0.0.0.0/0",
			},
		},
	}, nil).AnyTimes()
	regionalCli.EXPECT().
		GetCluster(gomock.Any(), gomock.Any(), gomock.Any()).
		Return(&clusterdedicatedv1.Cluster{ProxyBaseUrl: "http://127.0.0.1:8080"}, nil).
		AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().
		BillingActionCheck(gomock.Any(), billing.Action_Create, uint64(TenantID), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).
		AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().
		ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).
		AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, true)

	t.Run("request with empty tidb group id operates in the default group", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		_, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// scale the cluster.
		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIDB,
					SpuName:       "7c7g",
					Count:         lo.ToPtr(uint32(5)),
					TidbGroupId:   0,
				},
				{
					ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
					SpuName:       "8c8g",
					Count:         lo.ToPtr(uint32(2)),
					TidbGroupId:   0,
				},
			},
		}
		scaledCluster, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)

		// skus of tidb/tikv/tiflash should change to the expected ones.
		scaledTidbComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)
		assert.Equal(t, "7c7g", scaledTidbComp.GetSpuName())
		assert.Equal(t, "tidb-7c7g-arm", scaledTidbComp.GetSkuKey())
		assert.Equal(t, uint32(5), scaledTidbComp.GetCount())
		scaledTiproxyComp, ok := lo.Find(
			scaledCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY
			},
		)
		require.True(t, ok)
		assert.Equal(t, "8c8g", scaledTiproxyComp.GetSpuName())
		assert.Equal(t, "tiproxy-8c8g-arm", scaledTiproxyComp.GetSkuKey())
		assert.Equal(t, uint32(2), scaledTiproxyComp.GetCount())
	})

	t.Run("scale tiproxy", func(t *testing.T) {
		// create cluster first.
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(2)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		_, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
			},
		)
		require.True(t, ok)

		// mimic the cluster is in available state.
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		t.Run("scale tiproxy to 2c2g 2 instance should success", func(t *testing.T) {
			// scale the cluster.
			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
						SpuName:       "2c2g",
						Count:         lo.ToPtr(uint32(2)),
						TidbGroupId:   0,
					},
				},
			}
			scaledCluster, err := svc.Scale(t.Context(), scaleReq)
			require.NoError(t, err)

			// skus of tidb/tikv/tiflash should change to the expected ones.
			scaledTiproxyComp, ok := lo.Find(
				scaledCluster.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY
				},
			)
			require.True(t, ok)
			assert.Equal(t, "2c2g", scaledTiproxyComp.GetSpuName())
			assert.Equal(t, "tiproxy-2c2g-arm", scaledTiproxyComp.GetSkuKey())
			assert.Equal(t, uint32(2), scaledTiproxyComp.GetCount())
		})

		t.Run("scale tiproxy to 3 instance should success", func(t *testing.T) {
			// mimic the cluster is in available state.
			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)
			err_ := svc.workflow.DeleteDagInstances(
				t.Context(),
				strconv.FormatUint(createdCluster.GetClusterId(), 10),
				nil,
			)
			require.NoError(t, err_)
			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
						SpuName:       "2c2g",
						Count:         lo.ToPtr(uint32(3)),
						TidbGroupId:   0,
					},
				},
			}
			scaledCluster, serr := svc.Scale(t.Context(), scaleReq)
			require.NoError(t, serr)
			require.NotNil(t, scaledCluster)
			scaledTiproxyComp, ok := lo.Find(
				scaledCluster.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY
				},
			)
			require.True(t, ok)
			assert.Equal(t, "2c2g", scaledTiproxyComp.GetSpuName())
			assert.Equal(t, "tiproxy-2c2g-arm", scaledTiproxyComp.GetSkuKey())
			assert.Equal(t, uint32(3), scaledTiproxyComp.GetCount())
		})

		t.Run("scale tiproxy to 0 instance should success", func(t *testing.T) {
			// mimic the cluster is in available state.
			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)
			err_ := svc.workflow.DeleteDagInstances(
				t.Context(),
				strconv.FormatUint(createdCluster.GetClusterId(), 10),
				nil,
			)
			require.NoError(t, err_)
			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
						SpuName:       "2c2g",
						Count:         lo.ToPtr(uint32(0)),
						TidbGroupId:   0,
					},
				},
			}
			scaledCluster, serr := svc.Scale(t.Context(), scaleReq)
			require.NoError(t, serr)
			require.NotNil(t, scaledCluster)
			scaledTiproxyComp, ok := lo.Find(
				scaledCluster.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY
				},
			)
			require.True(t, ok)
			assert.Equal(t, "2c2g", scaledTiproxyComp.GetSpuName())
			assert.Equal(t, "tiproxy-2c2g-arm", scaledTiproxyComp.GetSkuKey())
			assert.Equal(t, uint32(0), scaledTiproxyComp.GetCount())
		})

		t.Run("scale tiproxy to 1 instance should failed", func(t *testing.T) {
			// mimic the cluster is in available state.
			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)
			err_ := svc.workflow.DeleteDagInstances(
				t.Context(),
				strconv.FormatUint(createdCluster.GetClusterId(), 10),
				nil,
			)
			require.NoError(t, err_)
			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
						TidbGroupId:   0,
					},
				},
			}
			_, serr := svc.Scale(t.Context(), scaleReq)
			require.ErrorContains(t, serr, "is not allowed")
		})

		t.Run("scale tiproxy to 4 instance should success", func(t *testing.T) {
			// mimic the cluster is in available state.
			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)
			err_ := svc.workflow.DeleteDagInstances(
				t.Context(),
				strconv.FormatUint(createdCluster.GetClusterId(), 10),
				nil,
			)
			require.NoError(t, err_)
			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						TidbGroupId:   0,
					},
				},
			}
			_, serr := svc.Scale(t.Context(), scaleReq)
			require.NoError(t, serr)
		})

		t.Run("scale tiproxy to 25 instance should failed", func(t *testing.T) {
			// mimic the cluster is in available state.
			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)
			err_ := svc.workflow.DeleteDagInstances(
				t.Context(),
				strconv.FormatUint(createdCluster.GetClusterId(), 10),
				nil,
			)
			require.NoError(t, err_)
			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(25)),
						TidbGroupId:   0,
					},
				},
			}
			_, serr := svc.Scale(t.Context(), scaleReq)
			require.ErrorContains(t, serr, "is not allowed")
		})

		t.Run("scale other tiproxy group to 2c2g should failed", func(t *testing.T) {
			// mimic the cluster is in available state.
			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)
			err_ := svc.workflow.DeleteDagInstances(
				t.Context(),
				strconv.FormatUint(createdCluster.GetClusterId(), 10),
				nil,
			)
			require.NoError(t, err_)
			require.NoError(t, svc.tidbGroupRepo.Create(t.Context(), &entity.TidbNodeGroup{
				Id:          1,
				ClusterId:   createdCluster.GetClusterId(),
				Default:     false,
				DisplayName: "ng-1",
			}))

			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(2)),
						TidbGroupId:   1,
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
						SpuName:       "2c2g",
						Count:         lo.ToPtr(uint32(2)),
						TidbGroupId:   1,
					},
				},
			}
			_, serr := svc.Scale(t.Context(), scaleReq)
			require.ErrorContains(t, serr, "components who belong to the same tidb group have different sku")
		})

		t.Run("scale other tiproxy group to 8c8g should success", func(t *testing.T) {
			// mimic the cluster is in available state.
			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)
			err_ := svc.workflow.DeleteDagInstances(
				t.Context(),
				strconv.FormatUint(createdCluster.GetClusterId(), 10),
				nil,
			)
			require.NoError(t, err_)
			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(2)),
						TidbGroupId:   1,
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(2)),
						TidbGroupId:   1,
					},
				},
			}
			scaleCluster, serr := svc.Scale(t.Context(), scaleReq)
			require.NoError(t, serr)
			require.NotNil(t, scaleCluster)
			scaledTiproxyComp, ok := lo.Find(
				scaleCluster.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY && i.GetTidbGroupId() == 1
				},
			)
			require.True(t, ok)
			assert.Equal(t, "8c8g", scaledTiproxyComp.GetSpuName())
			assert.Equal(t, "tiproxy-8c8g-arm", scaledTiproxyComp.GetSkuKey())
			assert.Equal(t, uint32(2), scaledTiproxyComp.GetCount())
		})

		t.Run("scale other tiproxy group to 0 instance should success", func(t *testing.T) {
			// mimic the cluster is in available state.
			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)
			err_ := svc.workflow.DeleteDagInstances(
				t.Context(),
				strconv.FormatUint(createdCluster.GetClusterId(), 10),
				nil,
			)
			require.NoError(t, err_)
			scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIPROXY,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(0)),
						TidbGroupId:   1,
					},
				},
			}
			scaleCluster, serr := svc.Scale(t.Context(), scaleReq)
			require.NoError(t, serr)
			require.NotNil(t, scaleCluster)
			scaledTiproxyComp, ok := lo.Find(
				scaleCluster.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY && i.GetTidbGroupId() == 1
				},
			)
			require.True(t, ok)
			assert.Equal(t, "8c8g", scaledTiproxyComp.GetSpuName())
			assert.Equal(t, "tiproxy-8c8g-arm", scaledTiproxyComp.GetSkuKey())
			assert.Equal(t, uint32(0), scaledTiproxyComp.GetCount())
		})
	})
}

func TestPreventPdScaleDown(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID:             ProjectID,
		TenantID:       TenantID,
		Name:           "test",
		AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID:   TenantID,
		Name: "test",
		Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), &regionaldedicatedv1.ListVpcsRequest{
		TenantId:  strconv.FormatUint(TenantID, 10),
		ProjectId: strconv.FormatUint(ProjectID, 10),
		Statuses: []regionaldedicatedv1.Vpc_Status{
			regionaldedicatedv1.Vpc_ACTIVE,
			regionaldedicatedv1.Vpc_CREATING,
			regionaldedicatedv1.Vpc_FAILED,
		},
	}).Return(&regionaldedicatedv1.ListVpcsResponse{
		Vpcs: []*regionaldedicatedv1.Vpc{
			{
				VpcId:      "vpc-xxxxxxxxxxxxxxxxx",
				TenantIds:  []string{strconv.FormatUint(TenantID, 10)},
				ProjectIds: []string{strconv.FormatUint(ProjectID, 10)},
				CidrBlock:  "0.0.0.0/0",
			},
		},
	}, nil).AnyTimes()
	regionalCli.EXPECT().
		GetCluster(gomock.Any(), gomock.Any(), gomock.Any()).
		Return(&clusterdedicatedv1.Cluster{ProxyBaseUrl: "http://127.0.0.1:8080"}, nil).
		AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().
		BillingActionCheck(gomock.Any(), billing.Action_Create, uint64(TenantID), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).
		AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().
		ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).
		AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, true)

	// Enable the BlockPdScaleDown feature gate for this test
	cfg := globalConf.Load()
	cfg.ServerConfig.FeatureGates = map[string]featgatesvc.FeatureGate{
		featgatesvc.PreferArm:        {All: true},
		featgatesvc.BlockPdScaleDown: {All: true},
	}
	featGateSvc := featgatesvc.NewFeatureGate(
		cfg.ServerConfig.FeatureGates,
		account.BuildOrgCreateTimeFetcher(accountCli),
	)
	svc.featGateCli = featGateSvc

	t.Run("pd scale down should be blocked, scale up should be allowed", func(t *testing.T) {
		// Create cluster with a larger PD component (8c8g)
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		// Get the PD component to confirm initial state
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "8c8g", createdPdComp.GetSpuName())

		// Mimic the cluster is in available state
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// Attempt to scale down tikv to 7c7g (smaller size)
		scaleDownReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "7c7g", // Try to scale down to a smaller size
				},
			},
		}
		scaledCluster1, err := svc.Scale(t.Context(), scaleDownReq)
		require.NoError(t, err)

		// Verify that PD remains at the original size (scale down blocked)
		scaledPdComp1, ok := lo.Find(
			scaledCluster1.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(
			t,
			"8c8g",
			scaledPdComp1.GetSpuName(),
			"PD should not be scaled down when BlockPdScaleDown feature is enabled",
		)

		// Reset cluster state for next test
		scaledCluster1.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), scaledCluster1)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster1.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		// Now attempt to scale up tikv to a larger size (16c16g)
		scaleUpReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: scaledCluster1.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "16c16g", // Try to scale up to a larger size
				},
			},
		}
		scaledUpCluster, err := svc.Scale(t.Context(), scaleUpReq)
		require.NoError(t, err)

		// Verify that PD was successfully scaled up
		scaledUpPdComp, ok := lo.Find(
			scaledUpCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(
			t,
			"16c16g",
			scaledUpPdComp.GetSpuName(),
			"PD should be allowed to scale up when BlockPdScaleDown feature is enabled",
		)
	})

	t.Run("override pd size first, pd scale down should be blocked, scale up should be allowed", func(t *testing.T) {
		// Create cluster with a larger PD component (8c8g)
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS,
				RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIDB,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(1)),
					},
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "8c8g",
						Count:         lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
								SizeGi:     555,
							},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {
								VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
							},
						},
					},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)

		// Get the PD component to confirm initial state
		createdPdComp, ok := lo.Find(
			createdCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(t, "8c8g", createdPdComp.GetSpuName())

		// Mimic the cluster is in available state
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), createdCluster)
		require.NoError(t, err)

		// Override PD size to 9c9g
		// scale the cluster.
		overrideReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_PD,
					SpuName:       "9c9g",
				},
			},
		}
		overrideCluster, err := svc.Scale(t.Context(), overrideReq)
		require.NoError(t, err)

		// Verify that PD was successfully overridden to 9c9g
		overriddenPdComp, ok := lo.Find(
			overrideCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			})
		require.True(t, ok)
		assert.Equal(t, "9c9g", overriddenPdComp.GetSpuName(), "PD should be overridden to 9c9g")
		assert.Equal(
			t,
			"pd-9c9g-arm-data:gp3-dynamic",
			overriddenPdComp.GetSkuKey(),
			"PD SKU should match the overridden size",
		)

		// Reset cluster state for next test
		overrideCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), overrideCluster)
		require.NoError(t, err)
		err_ := svc.workflow.DeleteDagInstances(
			t.Context(),
			strconv.FormatUint(overrideCluster.GetClusterId(), 10),
			nil,
		)
		require.NoError(t, err_)

		// Attempt to scale down tikv to 7c7g (smaller size)
		scaleDownReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "7c7g", // Try to scale down to a smaller size
				},
			},
		}
		scaledCluster1, err := svc.Scale(t.Context(), scaleDownReq)
		require.NoError(t, err)

		// Verify that PD remains at the original size (scale down blocked)
		scaledPdComp1, ok := lo.Find(
			scaledCluster1.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(
			t,
			"9c9g",
			scaledPdComp1.GetSpuName(),
			"PD should not be scaled down when BlockPdScaleDown feature is enabled",
		)
		assert.Equal(
			t,
			"pd-9c9g-arm-data:gp3-dynamic",
			scaledPdComp1.GetSkuKey(),
			"PD SKU should match the overridden size",
		)

		// Reset cluster state for next test
		scaledCluster1.State = globaldedicatedv1.Cluster_AVAILABLE
		err = svc.clusterRepo.Save(t.Context(), scaledCluster1)
		require.NoError(t, err)
		err_ = svc.workflow.DeleteDagInstances(t.Context(), strconv.FormatUint(scaledCluster1.GetClusterId(), 10), nil)
		require.NoError(t, err_)

		// Now attempt to scale up tikv to a larger size (16c16g)
		scaleUpReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: scaledCluster1.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{
					ComponentType: globaldedicatedv1.ComponentType_TIKV,
					SpuName:       "16c16g", // Try to scale up to a larger size
				},
			},
		}
		scaledUpCluster, err := svc.Scale(t.Context(), scaleUpReq)
		require.NoError(t, err)

		// Verify that PD was successfully scaled up
		scaledUpPdComp, ok := lo.Find(
			scaledUpCluster.GetClusterComponents(),
			func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
			},
		)
		require.True(t, ok)
		assert.Equal(
			t,
			"16c16g",
			scaledUpPdComp.GetSpuName(),
			"PD should be allowed to scale up when BlockPdScaleDown feature is enabled",
		)
		assert.Equal(
			t,
			"pd-16c16g-arm-data:gp3-dynamic",
			scaledUpPdComp.GetSkuKey(),
			"PD SKU should match the scaled up size",
		)
	})

	t.Run(
		"tenant pd-tikv mapping existed in history, pd scale down should be blocked, scale up should be allowed",
		func(t *testing.T) {
			// Adhoc pd-tikv mapping for the tenant
			// Get the existing tikv 8c8g spu
			spuSvc := skusvc.NewSpuService(
				spurepoimpl.NewSpuRepo(*dbProvider.Load()),
				spurepoimpl.NewSkuRepo(*dbProvider.Load()),
				svc.regionRepo,
				clusterrepoimpl.NewClusterRepo(*dbProvider.Load()),
				svc.skuSvc,
				featGateSvc,
				svc.transaction,
			)
			res, err := spuSvc.List(t.Context(), &globaldedicatedv1.ListSpusRequest{
				Request: &globaldedicatedv1.ListSpusRequest_WithoutSku{
					WithoutSku: &globaldedicatedv1.ListSpusRequest_FilterWithoutSku{
						ComponentTypes: []globaldedicatedv1.ComponentType{globaldedicatedv1.ComponentType_TIKV},
						Name:           "8c8g",
					},
				},
			})
			require.NoError(t, err)
			spuTikv8c8g := res[0]

			// Create a new tikv 8c8g SKU with 9c9g PD
			targetSkuKey := "tikv-8c8g-arm-data:gp3-raft:none-fixed"
			skuTikv8c8g, err := svc.skuSvc.GetSkuByKey(t.Context(), targetSkuKey)
			require.NoError(t, err)
			skuTikv8c8gWith9c9gPd := proto.CloneOf(skuTikv8c8g)
			skuTikv8c8gWith9c9gPd.Key = fmt.Sprintf("%s-9c9gpd", targetSkuKey)
			skuTikv8c8gWith9c9gPd.Id = 0
			_, err = svc.skuSvc.CreateSku(t.Context(), skuTikv8c8gWith9c9gPd)
			require.NoError(t, err)
			skuPd9c9g, err := svc.skuSvc.GetSkuByKey(t.Context(), "pd-9c9g-arm-data:gp3-dynamic")
			require.NoError(t, err)
			// Mapping tikv 8c8g with pd 9c9g
			_, err = svc.skuSvc.UpdateSku(t.Context(), &globaldedicatedv1.BatchUpsertSkusRequest_Update{
				Id: skuPd9c9g.GetId(),
				Conditions: []*globaldedicatedv1.Sku_Condition{
					{
						Key:      globaldedicatedv1.Sku_Condition_ATTACH_TIERUNIT,
						Operator: globaldedicatedv1.Sku_Condition_IN,
						Values: []string{
							skuTikv8c8gWith9c9gPd.GetKey(),
						},
					},
				},
			})
			require.NoError(t, err)

			// Update spu mapping to use the new tikv 8c8g with 9c9g pd
			_, err = spuSvc.Update(t.Context(), &globaldedicatedv1.BatchUpsertSpusRequest_Update{
				Id: spuTikv8c8g.GetId(),
				Mappings: append([]*globaldedicatedv1.Spu_Mapping{
					{
						CloudProvider:   globalcommonv1.CloudProvider_AWS,
						PhysicalRegions: []globalcommonv1.Region_Name{globalcommonv1.Region_AWS_US_WEST_2},
						OrganizationIds: []uint64{TenantID},
						SkuKey:          skuTikv8c8gWith9c9gPd.GetKey(),
					},
				}, spuTikv8c8g.GetMappings()...),
			})
			require.NoError(t, err)

			// Create cluster with a larger PD component (9c9g)
			createReq := &globaldedicatedv1.CreateClusterRequest{
				Cluster: &globaldedicatedv1.Cluster{
					ClusterId:       uint64(rand.Int64()),
					OrganizationId:  TenantID,
					ProjectId:       ProjectID,
					CloudProviderId: globalcommonv1.CloudProvider_AWS,
					RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					Port:            4000,
					ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
						{
							ComponentType: globaldedicatedv1.ComponentType_TIDB,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(1)),
						},
						{
							ComponentType: globaldedicatedv1.ComponentType_TIKV,
							SpuName:       "8c8g",
							Count:         lo.ToPtr(uint32(3)),
							Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
								globaldedicatedv1.Sku_DATA.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3),
									SizeGi:     555,
								},
								globaldedicatedv1.Sku_RAFT_LOG.String(): {
									VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
								},
							},
						},
					},
				},
			}
			createdCluster, err := svc.Create(t.Context(), createReq)
			require.NoError(t, err)

			// Get the PD component to confirm initial state
			createdPdComp, ok := lo.Find(
				createdCluster.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
				},
			)
			require.True(t, ok)
			assert.Equal(t, "9c9g", createdPdComp.GetSpuName())
			assert.Equal(t, "pd-9c9g-arm-data:gp3-dynamic", createdPdComp.GetSkuKey())

			// Mimic the cluster is in available state
			createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), createdCluster)
			require.NoError(t, err)

			// Recover spu mapping to use the 8c8g tikv with 8c8g pd
			_, err = spuSvc.Update(t.Context(), &globaldedicatedv1.BatchUpsertSpusRequest_Update{
				Id:       spuTikv8c8g.GetId(),
				Mappings: spuTikv8c8g.GetMappings(),
			})
			require.NoError(t, err)

			// Attempt to scale down tikv to 7c7g (smaller size)
			scaleDownReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: createdCluster.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "7c7g", // Try to scale down to a smaller size
					},
				},
			}
			scaledCluster1, err := svc.Scale(t.Context(), scaleDownReq)
			require.NoError(t, err)

			// Verify that PD remains at the original size (scale down blocked)
			scaledPdComp1, ok := lo.Find(
				scaledCluster1.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
				},
			)
			require.True(t, ok)
			assert.Equal(
				t,
				"9c9g",
				scaledPdComp1.GetSpuName(),
				"PD should not be scaled down when BlockPdScaleDown feature is enabled",
			)
			assert.Equal(
				t,
				"pd-9c9g-arm-data:gp3-dynamic",
				scaledPdComp1.GetSkuKey(),
				"PD SKU should match the overridden size",
			)

			// Reset cluster state for next test
			scaledCluster1.State = globaldedicatedv1.Cluster_AVAILABLE
			err = svc.clusterRepo.Save(t.Context(), scaledCluster1)
			require.NoError(t, err)
			err_ := svc.workflow.DeleteDagInstances(
				t.Context(),
				strconv.FormatUint(scaledCluster1.GetClusterId(), 10),
				nil,
			)
			require.NoError(t, err_)

			// Now attempt to scale up tikv to a larger size (16c16g)
			scaleUpReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
				ClusterId: scaledCluster1.GetClusterId(),
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{
						ComponentType: globaldedicatedv1.ComponentType_TIKV,
						SpuName:       "16c16g", // Try to scale up to a larger size
					},
				},
			}
			scaledUpCluster, err := svc.Scale(t.Context(), scaleUpReq)
			require.NoError(t, err)

			// Verify that PD was successfully scaled up
			scaledUpPdComp, ok := lo.Find(
				scaledUpCluster.GetClusterComponents(),
				func(i *globaldedicatedv1.Cluster_ClusterComponent) bool {
					return i.GetComponentType() == globaldedicatedv1.ComponentType_PD
				},
			)
			require.True(t, ok)
			assert.Equal(
				t,
				"16c16g",
				scaledUpPdComp.GetSpuName(),
				"PD should be allowed to scale up when BlockPdScaleDown feature is enabled",
			)
			assert.Equal(
				t,
				"pd-16c16g-arm-data:gp3-dynamic",
				scaledUpPdComp.GetSkuKey(),
				"PD SKU should match the scaled up size",
			)
		},
	)
}

func TestScaleDisableGKEHeterogeneousOptionDeployExists(t *testing.T) {
	ctrl := gomock.NewController(t, gomock.WithOverridableExpectations())

	var capturedOpt *workflow.CreateDynamicDagOption
	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(gomock.Any(), gomock.Any(), gomock.Any()).
		Return(nil, nil).AnyTimes()
	mockWf.EXPECT().RegisterDynamicDagThenRun(gomock.Any(), gomock.Any(), gomock.Any()).
		DoAndReturn(func(_ context.Context, _ string, opt *workflow.CreateDynamicDagOption) error {
			capturedOpt = opt
			return nil
		}).AnyTimes()
	mockWf.EXPECT().RunDagWithBizId(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
		Return(nil).AnyTimes()
	mockWf.EXPECT().DeleteDagInstances(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
	mockWf.EXPECT().DeleteDagInstancesWithDagId(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
	mockWf.EXPECT().RegisterDag(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()

	encryptor := mock_remoteencryptor.NewMockClient(ctrl)
	encryptor.EXPECT().EncryptBase64(gomock.Any(), gomock.Any()).Return("encrypted", nil).AnyTimes()
	encryptor.EXPECT().DecryptBase64(gomock.Any(), gomock.Any()).Return("password", nil).AnyTimes()

	accountCli := mock_account.NewMockClient(ctrl)
	accountCli.EXPECT().GetProject(gomock.Any(), uint64(ProjectID)).Return(&account.ProjectInfo{
		ID: ProjectID, TenantID: TenantID, Name: "test", AwsCmekEnabled: false,
	}, nil).AnyTimes()
	accountCli.EXPECT().GetTenant(gomock.Any(), uint64(TenantID)).Return(&account.TenantInfo{
		ID: TenantID, Name: "test", Plan: account.TenantPlanPOC,
	}, nil).AnyTimes()

	regionalCli := mock_regionserver.NewMockClient(ctrl)
	regionalCli.EXPECT().ListVpcs(gomock.Any(), gomock.Any(), gomock.Any()).Return(
		&regionaldedicatedv1.ListVpcsResponse{}, nil).AnyTimes()
	regionalCli.EXPECT().GetCluster(gomock.Any(), gomock.Any(), gomock.Any()).Return(
		&clusterdedicatedv1.Cluster{ProxyBaseUrl: "http://127.0.0.1:8080"}, nil).AnyTimes()

	metricCli := mock_clustermetric.NewMockClient(ctrl)

	billingCli := mock_billing.NewMockClient(ctrl)
	billingCli.EXPECT().BillingActionCheck(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
		Return(&billing.ActionCheckResult{Pass: true}, nil).AnyTimes()

	clusterConfigCli := mock_clusterconfig.NewMockClient(ctrl)
	clusterConfigCli.EXPECT().GetClusterVersion(gomock.Any(), gomock.Any()).Return(ClusterVersion, nil).AnyTimes()
	clusterConfigCli.EXPECT().ListComponentConfigurations(gomock.Any(), gomock.Any()).
		Return([]*regionaldedicatedv1.ComponentConfiguration{}, nil).AnyTimes()

	svc := newTestService(encryptor, accountCli, regionalCli, metricCli, billingCli, clusterConfigCli, true)
	svc.workflow = mockWf
	svc.featGateCli = featgatesvc.NewFeatureGate(
		map[string]featgatesvc.FeatureGate{
			featgatesvc.PreferArm: {
				All:                    true,
				DisabledCloudProviders: []string{"gcp", "azure"},
			},
			featgatesvc.AllowCpuArchOrGenerationChangeCompletely: {All: true},
			featgatesvc.AllowHeterogeneousCpuGeneration:          {All: true},
		},
		account.BuildOrgCreateTimeFetcher(accountCli),
	)

	hasDisableGKEHeterogeneousOption := func(t *testing.T, opt *workflow.CreateDynamicDagOption) bool {
		t.Helper()
		require.NotNil(t, opt)

		for _, task := range opt.DagTasks {
			body, ok := task.Params[common.ScaleTaskKeyRequestBody]
			if !ok {
				continue
			}
			bodyStr, ok := body.(string)
			require.True(t, ok)

			var scaleParam struct {
				Options []string `json:"options"`
			}
			require.NoError(t, json.Unmarshal([]byte(bodyStr), &scaleParam))
			if lo.Contains(scaleParam.Options, dag.OptionDisableGKEHeterogeneous) {
				return true
			}
		}
		return false
	}

	createGCPN2Cluster := func(t *testing.T) *entity.Cluster {
		t.Helper()

		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId:       uint64(rand.Int64()),
				OrganizationId:  TenantID,
				ProjectId:       ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_GCP,
				RegionId:        globalcommonv1.Region_GCP_US_WEST1,
				Port:            4000,
				ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{ComponentType: globaldedicatedv1.ComponentType_TIDB, SpuName: "TidbGcpX86N2", Count: lo.ToPtr(uint32(1))},
					{ComponentType: globaldedicatedv1.ComponentType_TIKV, SpuName: "TikvGcpX86N2", Count: lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String():     {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_PD_SSD), SizeGi: 555},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED)},
						}},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		require.NoError(t, svc.clusterRepo.Save(t.Context(), createdCluster))
		return createdCluster
	}

	scaleGCPN2ClusterToN4 := func(t *testing.T, clusterID uint64) {
		t.Helper()

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: clusterID,
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{ComponentType: globaldedicatedv1.ComponentType_TIKV, SpuName: "TikvGcpX86N4", Count: lo.ToPtr(uint32(3)),
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String():     {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_PD_SSD), SizeGi: 555},
						globaldedicatedv1.Sku_RAFT_LOG.String(): {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED)},
					}},
			},
		}
		_, err := svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)
	}

	expectGKEHeterogeneousDeployLookup := func(deploy *regionaldedicatedv1.GKEHeterogeneousDeploy, berr bizerr.BizError) {
		regionalCli.EXPECT().
			GetGKEHeterogeneousDeploy(gomock.Any(), gomock.Any(), gomock.Any()).
			Return(deploy, berr).
			AnyTimes()
	}

	t.Run("GCP enabled deploy + spec-change includes disable option", func(t *testing.T) {
		capturedOpt = nil
		createdCluster := createGCPN2Cluster(t)
		expectGKEHeterogeneousDeployLookup(&regionaldedicatedv1.GKEHeterogeneousDeploy{
			Enabled:       true,
			ComponentType: regionaldedicatedv1.ComponentType_TIKV,
			Status:        regionaldedicatedv1.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_READY_FOR_GATEKEEPER,
		}, nil)

		scaleGCPN2ClusterToN4(t, createdCluster.GetClusterId())

		require.True(t, hasDisableGKEHeterogeneousOption(t, capturedOpt),
			"GCP instance-type scale with active heterogeneous deploy should include %s",
			dag.OptionDisableGKEHeterogeneous)
	})

	t.Run("enabled deploy + spec-change excludes option on non-GCP", func(t *testing.T) {
		capturedOpt = nil
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId: uint64(rand.Int64()), OrganizationId: TenantID, ProjectId: ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS, RegionId: globalcommonv1.Region_AWS_US_WEST_2,
				Port: 4000, ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{ComponentType: globaldedicatedv1.ComponentType_TIDB, SpuName: "8c8g", Count: lo.ToPtr(uint32(1))},
					{ComponentType: globaldedicatedv1.ComponentType_TIKV, SpuName: "8c8g", Count: lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String():     {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3), SizeGi: 555},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED)},
						}},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		require.NoError(t, svc.clusterRepo.Save(t.Context(), createdCluster))

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{ComponentType: globaldedicatedv1.ComponentType_TIKV, SpuName: "7c7g", Count: lo.ToPtr(uint32(3)),
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String():     {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3), SizeGi: 555},
						globaldedicatedv1.Sku_RAFT_LOG.String(): {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED)},
					}},
			},
		}
		_, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)
		require.NotNil(t, capturedOpt)

		require.False(t, hasDisableGKEHeterogeneousOption(t, capturedOpt),
			"non-GCP scale should not include %s", dag.OptionDisableGKEHeterogeneous)
	})

	t.Run("PREPARED deploy + spec-change excludes option", func(t *testing.T) {
		capturedOpt = nil
		createdCluster := createGCPN2Cluster(t)
		expectGKEHeterogeneousDeployLookup(&regionaldedicatedv1.GKEHeterogeneousDeploy{
			Enabled:       false,
			ComponentType: regionaldedicatedv1.ComponentType_TIKV,
			Status:        regionaldedicatedv1.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_PREPARED,
		}, nil)

		scaleGCPN2ClusterToN4(t, createdCluster.GetClusterId())

		require.False(t, hasDisableGKEHeterogeneousOption(t, capturedOpt),
			"PREPARED scale should not include %s", dag.OptionDisableGKEHeterogeneous)
	})

	t.Run("count-only scale excludes option", func(t *testing.T) {
		capturedOpt = nil
		createReq := &globaldedicatedv1.CreateClusterRequest{
			Cluster: &globaldedicatedv1.Cluster{
				ClusterId: uint64(rand.Int64()), OrganizationId: TenantID, ProjectId: ProjectID,
				CloudProviderId: globalcommonv1.CloudProvider_AWS, RegionId: globalcommonv1.Region_AWS_US_WEST_2,
				Port: 4000, ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
					{ComponentType: globaldedicatedv1.ComponentType_TIDB, SpuName: "8c8g", Count: lo.ToPtr(uint32(1))},
					{ComponentType: globaldedicatedv1.ComponentType_TIKV, SpuName: "8c8g", Count: lo.ToPtr(uint32(3)),
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String():     {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3), SizeGi: 555},
							globaldedicatedv1.Sku_RAFT_LOG.String(): {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED)},
						}},
				},
			},
		}
		createdCluster, err := svc.Create(t.Context(), createReq)
		require.NoError(t, err)
		createdCluster.State = globaldedicatedv1.Cluster_AVAILABLE
		require.NoError(t, svc.clusterRepo.Save(t.Context(), createdCluster))

		scaleReq := &globaldedicatedv1.ScaleClusterWithMultiComponentRequest{
			ClusterId: createdCluster.GetClusterId(),
			ClusterComponents: []*globaldedicatedv1.Cluster_ClusterComponent{
				{ComponentType: globaldedicatedv1.ComponentType_TIKV, SpuName: "8c8g", Count: lo.ToPtr(uint32(6)),
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String():     {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_GP3), SizeGi: 555},
						globaldedicatedv1.Sku_RAFT_LOG.String(): {VolumeType: lo.ToPtr(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED)},
					}},
			},
		}
		_, err = svc.Scale(t.Context(), scaleReq)
		require.NoError(t, err)
		require.NotNil(t, capturedOpt)

		require.False(t, hasDisableGKEHeterogeneousOption(t, capturedOpt),
			"count-only scale should not include %s", dag.OptionDisableGKEHeterogeneous)
	})

	t.Run("RecordNotFound + spec-change excludes option", func(t *testing.T) {
		capturedOpt = nil
		createdCluster := createGCPN2Cluster(t)
		expectGKEHeterogeneousDeployLookup(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"))

		scaleGCPN2ClusterToN4(t, createdCluster.GetClusterId())

		require.False(t, hasDisableGKEHeterogeneousOption(t, capturedOpt),
			"RecordNotFound scale should not include %s", dag.OptionDisableGKEHeterogeneous)
	})
}

// TestStepIopsTikvPlusStorage verifies IOPS handling of the data disk of TiKV Plus storage type when scaling up|down tikv.
// TiKV Plus storage type is a DATA disk backed by an IO2 volume with fixed IOPS and no raft-log storage.
// When the PreserveTiKVProvisionedIOPS feature gate is enabled the desired IOPS is
// max(current IOPS, new SKU default IOPS); when it is disabled IOPS is reset to the SKU default.
func TestStepIopsTikvPlusStorage(t *testing.T) {
	newSku := func(componentType globaldedicatedv1.ComponentType, volumeType globaldedicatedv1.StorageMu_VolumeType, defaultIops int32, raftlog *globaldedicatedv1.Sku_Storage) *entity.Sku {
		return &entity.Sku{
			ComponentType: componentType,
			CloudProvider: globalcommonv1.CloudProvider_AWS,
			DataStorage: &globaldedicatedv1.Sku_Storage{
				DiskType:   globaldedicatedv1.Sku_DATA,
				VolumeType: volumeType,
				IopsMin:    1,
				IopsMax:    32000,
				Iops: &globaldedicatedv1.Sku_Storage_IopsFixed_{
					IopsFixed: &globaldedicatedv1.Sku_Storage_IopsFixed{Iops: defaultIops},
				},
			},
			RaftlogStorage: raftlog,
		}
	}
	newRaftlog := func(volumeType globaldedicatedv1.StorageMu_VolumeType) *globaldedicatedv1.Sku_Storage {
		return &globaldedicatedv1.Sku_Storage{
			DiskType:   globaldedicatedv1.Sku_RAFT_LOG,
			VolumeType: volumeType,
		}
	}

	tests := []struct {
		name        string
		gateEnabled bool
		curIops     uint32
		sku         *entity.Sku
		wantIops    uint32
		wantErr     bool
	}{
		{
			name:        "plus storage keeps current iops when it is higher than sku default",
			gateEnabled: true,
			curIops:     2000,
			sku:         newSku(globaldedicatedv1.ComponentType_TIKV, globaldedicatedv1.StorageMu_IO2, 1000, nil),
			wantIops:    2000,
		},
		{
			name:        "plus storage bumps to sku default when current iops is lower",
			gateEnabled: true,
			curIops:     500,
			sku:         newSku(globaldedicatedv1.ComponentType_TIKV, globaldedicatedv1.StorageMu_IO2, 1000, nil),
			wantIops:    1000,
		},
		{
			name:        "plus storage keeps current iops when raft-log volume type is unspecified",
			gateEnabled: true,
			curIops:     2000,
			sku: newSku(
				globaldedicatedv1.ComponentType_TIKV,
				globaldedicatedv1.StorageMu_IO2,
				1000,
				newRaftlog(globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED),
			),
			wantIops: 2000,
		},
		{
			name:        "plus storage resets to sku default when gate is disabled",
			gateEnabled: false,
			curIops:     2000,
			sku:         newSku(globaldedicatedv1.ComponentType_TIKV, globaldedicatedv1.StorageMu_IO2, 1000, nil),
			wantIops:    1000,
		},
		{
			name:        "non-io2 storage always resets to sku default",
			gateEnabled: true,
			curIops:     2000,
			sku:         newSku(globaldedicatedv1.ComponentType_TIKV, globaldedicatedv1.StorageMu_GP3, 1000, nil),
			wantIops:    1000,
		},
		{
			name:        "storage with io2 raft-log always resets to sku default",
			gateEnabled: true,
			curIops:     2000,
			sku: newSku(
				globaldedicatedv1.ComponentType_TIKV,
				globaldedicatedv1.StorageMu_IO2,
				1000,
				newRaftlog(globaldedicatedv1.StorageMu_IO2),
			),
			wantIops: 1000,
		},
		{
			name:        "non-tikv component always resets to sku default",
			gateEnabled: true,
			curIops:     2000,
			sku:         newSku(globaldedicatedv1.ComponentType_TIFLASH, globaldedicatedv1.StorageMu_IO2, 1000, nil),
			wantIops:    1000,
		},
		{
			// When the gate is enabled and the current provisioned IOPS exceeds the target SKU's
			// max IOPS, stepIops preserves the current IOPS, so the subsequent create/scale
			// validation (componentStorageIopsCheck) must reject it for exceeding IopsMax.
			name:        "plus storage keeps current iops above sku max and is rejected by validation when gate enabled",
			gateEnabled: true,
			curIops:     40000,
			sku:         newSku(globaldedicatedv1.ComponentType_TIKV, globaldedicatedv1.StorageMu_IO2, 1000, nil),
			wantIops:    40000,
			wantErr:     true,
		},
		{
			// When the gate is disabled, stepIops resets to the SKU default IOPS even if the current
			// IOPS exceeds the target SKU's max IOPS, so validation passes.
			name:        "plus storage resets to sku default when gate disabled even if current iops above sku max",
			gateEnabled: false,
			curIops:     40000,
			sku:         newSku(globaldedicatedv1.ComponentType_TIKV, globaldedicatedv1.StorageMu_IO2, 1000, nil),
			wantIops:    1000,
			wantErr:     false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctrl := gomock.NewController(t)
			gate := mock_featuregate.NewMockService(ctrl)
			gate.EXPECT().
				CheckFeatureEnabled(
					featgatesvc.PreserveTiKVProvisionedIOPS,
					uint64(TenantID),
					uint64(ProjectID),
					globalcommonv1.CloudProvider_AWS.String(),
					globalcommonv1.Region_AWS_US_WEST_2.String(),
				).
				Return(tt.gateEnabled).
				AnyTimes()

			mgr := &buildManager{
				ctx:        t.Context(),
				clusterSvc: &serviceImpl{featGateCli: gate},
				cluster: &entity.Cluster{
					Cluster: &globaldedicatedv1.Cluster{
						OrganizationId:  TenantID,
						ProjectId:       ProjectID,
						CloudProviderId: globalcommonv1.CloudProvider_AWS,
						RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
					},
				},
				currentComponentMayNil: &globaldedicatedv1.Cluster_ClusterComponent{
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
						globaldedicatedv1.Sku_DATA.String(): {Iops: tt.curIops},
					},
				},
				requestComponent: &globaldedicatedv1.Cluster_ClusterComponent{
					Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{},
				},
				desiredComponent: &entity.DesiredClusterComponent{
					ClusterComponent: &globaldedicatedv1.Cluster_ClusterComponent{
						Storages: map[string]*globaldedicatedv1.Cluster_ClusterComponent_Storage{
							globaldedicatedv1.Sku_DATA.String(): {},
						},
					},
					Sku: tt.sku,
				},
			}

			mgr.stepIops(globaldedicatedv1.Sku_DATA, true)

			got := mgr.desiredComponent.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetIops()
			assert.Equal(t, tt.wantIops, got)

			// The scale flow validates the desired IOPS via componentStorageIopsCheck, which rejects
			// any desired IOPS above the target SKU's IopsMax. Verify the gate's behavior end-to-end.
			err := componentStorageIopsCheck(got, tt.sku, globaldedicatedv1.Sku_DATA)
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}