package server

import (
	"context"
	"fmt"
	"testing"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/samber/lo"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	clusteragentv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/cluster-agent/v1"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	bractionmocks "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br/mocks"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	clustergatewaymocks "github.com/tidbcloud/cluster-service-ng/pkg/clustergateway/mocks"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	dataplanemgrmocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr/mocks"
	kubemocks "github.com/tidbcloud/cluster-service-ng/pkg/externalmocks/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks"
	mysqlschema "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
)

func TestScaleSharedPool(t *testing.T) {
	tests := []struct {
		name           string
		sharedPoolId   string
		scaleComponent *entity.Component
		force          bool
		setupMocks     func(
			*repomnocks.SharedPoolRepo,
			*repomnocks.LockRepo,
			*dataplanemgrmocks.Client,
			*executormocks.RemoteExecutor,
			*bractionmocks.ScaleManager,
			*dataplanemocks.Manager,
			*repomnocks.RuConfigRepo,
			*repomnocks.TiDBClusterRepo,
		)
		setupClusterGateway func(*clustergatewaymocks.Client)
		expectedError       string
		expectedCode        codes.Code
		validate            func(*testing.T, *entity.SharedPool, error)
	}{
		{
			name:         "successful scale - increase node count",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3, // scale from 2 to 3
				},
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneMgr *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				// Create existing shared pool with available status
				existingSharedPool := createTestSharedPool()
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)

				// Mock lock operations
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)

				// Mock dataplane node specs
				nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{
							CpuCore:               4,
							MemoryGi:              8,
							AvailableCpuMilliCore: 3800,
							AvailableMemoryMi:     7600,
							NodeType:              dplanemgrv1.NodeType_Dedicated,
							Version:               "v1.0.0",
						},
					},
				}
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(nodePoolResp, nil)

				// Mock DAG executor
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateComponent, mock.Anything, mock.Anything).Return(nil)

				// Mock HPA check - scale out requires scaleDown disabled check
				// Return empty HPA list (no HPA exists) which bypasses the check
				kubeclient := kubemocks.NewClient(t)
				kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
					Return(nil)
				dataplaneMgr.On("GetClient", mock.AnythingOfType("string")).Return(kubeclient, nil)
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testsharedpoolid", result.GetId())
				assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
			},
		},
		{
			name:         "force skips HPA gate on scale out",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3, // scale from 2 to 3
				},
			},
			force: true,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				_ *dataplanemocks.Manager,
				_ *repomnocks.RuConfigRepo,
				_ *repomnocks.TiDBClusterRepo,
			) {
				existingSharedPool := createTestSharedPool()
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)

				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)

				nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{
							CpuCore:               4,
							MemoryGi:              8,
							AvailableCpuMilliCore: 3800,
							AvailableMemoryMi:     7600,
							NodeType:              dplanemgrv1.NodeType_Dedicated,
							Version:               "v1.0.0",
						},
					},
				}
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(nodePoolResp, nil)

				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateComponent, mock.Anything, mock.Anything).Return(nil)
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testsharedpoolid", result.GetId())
				assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
			},
		},
		{
			name:         "force cannot bypass paused lifecycle guard",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota:     &apiv1.NodeQuota{NodeCount: 3},
			},
			force: true,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				_ *dataplanemgrmocks.Client,
				_ *executormocks.RemoteExecutor,
				_ *bractionmocks.ScaleManager,
				_ *dataplanemocks.Manager,
				_ *repomnocks.RuConfigRepo,
				_ *repomnocks.TiDBClusterRepo,
			) {
				sharedPool := createTestSharedPool()
				sharedPool.PauseResumeState = entity.PauseResumeStatePaused
				sharedPool.SuspendComputeDesired = true
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(sharedPool, nil).Once()
				lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()
			},
			expectedError: "does not accept this mutation while paused",
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:         "scale in ignores HPA gate",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 1, // scale from 2 to 1
				},
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				_ *dataplanemocks.Manager,
				_ *repomnocks.RuConfigRepo,
				_ *repomnocks.TiDBClusterRepo,
			) {
				existingSharedPool := createTestSharedPool()
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)

				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)

				nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{
							CpuCore:               4,
							MemoryGi:              8,
							AvailableCpuMilliCore: 3800,
							AvailableMemoryMi:     7600,
							NodeType:              dplanemgrv1.NodeType_Dedicated,
							Version:               "v1.0.0",
						},
					},
				}
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(nodePoolResp, nil)

				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateComponent, mock.Anything, mock.Anything).Return(nil)
			},
			setupClusterGateway: func(clusterGatewayMock *clustergatewaymocks.Client) {
				clusterGatewayMock.On("GetComponentResourceMetrics", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
					Return(&clusteragentv1.GetComponentResourceMetricsResp{
						MetricsServerAvailable: true,
						Components: []*clusteragentv1.ComponentResourceMetrics{
							{CpuCoresUsed: 1.0, MemoryBytesUsed: 1 * 1024 * 1024 * 1024},
							{CpuCoresUsed: 1.0, MemoryBytesUsed: 1 * 1024 * 1024 * 1024},
						},
					}, nil)
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testsharedpoolid", result.GetId())
				assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
			},
		},
		{
			name:         "missing sharedPool id",
			sharedPoolId: "",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIDB,
			},
			force: false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "sharedPool id is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:           "nil scale component",
			sharedPoolId:   "testsharedpoolid",
			scaleComponent: nil,
			force:          false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "scale components is empty",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "unspecified component type",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED,
			},
			force: false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "scale component type is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "unsupported storage type",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type: "unsupported-storage",
					},
				},
			},
			force: false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "invalid storage type unsupported-storage",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "gp3 storage missing required parameters",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name:   apiv1.Component_Storage_DATA,
						Type:   entity.GP3,
						SizeGi: 100,
						Iops:   0, // Missing IOPS
					},
				},
			},
			force: false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "gp3 storage iops must be greater than zero",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "io2 storage with throughput (should be zero)",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name:          apiv1.Component_Storage_DATA,
						Type:          entity.IO2,
						SizeGi:        100,
						Iops:          5000,
						ThroughputMbs: 125, // Should be zero for IO2
					},
				},
			},
			force: false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "io2 storage does not support setting throughput",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "tikv component with invalid node count (not multiple of 3)",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 4, // Should be multiple of 3
				},
			},
			force: false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "scale component tikv node count must be multiple of 3",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "pd component with invalid node count (not multiple of 3)",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_PD,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 2, // Should be multiple of 3
				},
			},
			force: false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "scale component pd node count must be multiple of 3",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "pd_tso component with invalid node count (must be 2)",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_PD_TSO,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3, // PD_TSO must be exactly 2
				},
			},
			force: false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "component pd-tso node count must be 2",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "pd_tso component scale to 1 (invalid)",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_PD_TSO,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 1, // PD_TSO must be exactly 2
				},
			},
			force: false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "component pd-tso node count must be 2",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "non-tiflash component with zero node count",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 0, // Only TiFlash can have zero nodes
				},
			},
			force: false,
			setupMocks: func(
				*repomnocks.SharedPoolRepo,
				*repomnocks.LockRepo,
				*dataplanemgrmocks.Client,
				*executormocks.RemoteExecutor,
				*bractionmocks.ScaleManager,
				*dataplanemocks.Manager,
				*repomnocks.RuConfigRepo,
				*repomnocks.TiDBClusterRepo,
			) {
			},
			expectedError: "scale component node count is required, only TiFlash node count can be zero",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "topology driven tidb worker scale from zero to positive when component unavailable",
			sharedPoolId: "topology-pool",
			scaleComponent: &entity.Component{
				Id:            "fc2d2fa2",
				ComponentType: apiv1.ComponentType_TIDB_WORKER,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 1,
				},
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
					Id:            "topology-pool",
					DataplaneName: "test-dataplane",
					Phase:         apiv1.SharedPool_AVAILABLE,
					CapacityMode:  apiv1.CapacityMode_TOPOLOGY_DRIVEN,
					ComponentGroups: []*apiv1.Component{
						{
							Id:            "fc2d2fa2",
							ComponentType: apiv1.ComponentType_TIDB_WORKER,
							Phase:         apiv1.Component_UNAVAILABLE,
							NodeQuota: &apiv1.NodeQuota{
								NodeCount:    0,
								MinNodeCount: 0,
								MaxNodeCount: 3,
								NodeSpec:     &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8},
							},
						},
					},
				}}
				sharedPoolRepo.On("Load", mock.Anything, "topology-pool", false).Return(sharedPool, nil)
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "topology-pool"
				})).Return(func() error { return nil }, nil)
				nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{NodePools: []*dplanemgrv1.NodePool{{
					CpuCore:               4,
					MemoryGi:              8,
					AvailableCpuMilliCore: 3800,
					AvailableMemoryMi:     7600,
					NodeType:              dplanemgrv1.NodeType_Dedicated,
					Version:               "v1.0.0",
				}}}
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(nodePoolResp, nil)
				kubeclient := kubemocks.NewClient(t)
				hpaList := &autoscalingv2.HorizontalPodAutoscalerList{}
				kubeclient.On("List", mock.Anything, mock.IsType(hpaList), mock.Anything).Return(nil)
				dataplaneManager.On("GetClient", mock.AnythingOfType("string")).Return(kubeclient, nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateComponent, mock.Anything, mock.Anything).Return(nil)
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
			},
		},
		{
			name:         "topology driven tikv worker scale from positive to zero",
			sharedPoolId: "topology-pool-2",
			scaleComponent: &entity.Component{
				Id:            "fc2d2fa2",
				ComponentType: apiv1.ComponentType_TIKV_WORKER,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 0,
				},
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
					Id:            "topology-pool-2",
					DataplaneName: "test-dataplane",
					Phase:         apiv1.SharedPool_AVAILABLE,
					CapacityMode:  apiv1.CapacityMode_TOPOLOGY_DRIVEN,
					ComponentGroups: []*apiv1.Component{
						{
							Id:            "fc2d2fa2",
							ComponentType: apiv1.ComponentType_TIKV_WORKER,
							Phase:         apiv1.Component_AVAILABLE,
							NodeQuota: &apiv1.NodeQuota{
								NodeCount:    1,
								MinNodeCount: 0,
								MaxNodeCount: 3,
								NodeSpec:     &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16},
							},
						},
					},
				}}
				sharedPoolRepo.On("Load", mock.Anything, "topology-pool-2", false).Return(sharedPool, nil)
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "topology-pool-2"
				})).Return(func() error { return nil }, nil)
				nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{NodePools: []*dplanemgrv1.NodePool{{
					CpuCore:               8,
					MemoryGi:              16,
					AvailableCpuMilliCore: 7600,
					AvailableMemoryMi:     15000,
					NodeType:              dplanemgrv1.NodeType_Dedicated,
					Version:               "v1.0.0",
				}}}
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(nodePoolResp, nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateComponent, mock.Anything, mock.Anything).Return(nil)
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
			},
		},
		{
			name:         "shared pool not found",
			sharedPoolId: "non-existent-pool",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIDB,
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)
				sharedPoolRepo.On("Load", mock.Anything, "non-existent-pool", false).Return(nil, status.Error(codes.NotFound, "not found"))
			},
			expectedError: "sharedPool non-existent-pool not found",
			expectedCode:  codes.NotFound,
		},
		{
			name:         "lock failure",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIDB,
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				// Mock lock failure - since lock fails, other methods won't be called
				lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, fmt.Errorf("lock failed"))
			},
			expectedError: "lock failed",
		},
		{
			name:         "component not available - force false",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
				},
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				existingSharedPool := createTestSharedPool()
				existingSharedPool.Phase = apiv1.SharedPool_MODIFYING
				tidbGroup, ok := lo.Find(existingSharedPool.ComponentGroups, func(item *apiv1.Component) bool {
					return item.ComponentType == apiv1.ComponentType_TIDB
				})
				assert.True(t, ok)
				tidbGroup.Phase = apiv1.Component_MODIFYING
				// Set component to MODIFYING phase
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)
			},
			expectedError: "target component tidb-default is not available, current phase: MODIFYING",
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:         "component not available - force true",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
				},
			},
			force: true,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				existingSharedPool := createTestSharedPool()
				existingSharedPool.ComponentGroups[0].Phase = apiv1.Component_MODIFYING
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)
				nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{
							CpuCore:               4,
							MemoryGi:              8,
							AvailableCpuMilliCore: 3800,
							AvailableMemoryMi:     7600,
							NodeType:              dplanemgrv1.NodeType_Dedicated,
							Version:               "v1.0.0",
						},
					},
				}
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(nodePoolResp, nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateComponent, mock.Anything, mock.Anything).Return(nil)
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testsharedpoolid", result.GetId())
				assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
			},
		},
		{
			name:         "scale storage - increase IOPS",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
				},
				Storages: []*apiv1.Component_Storage{
					{
						Name:   apiv1.Component_Storage_DATA,
						Type:   "io2",
						SizeGi: 500,  // Include size for io2 storage
						Iops:   5000, // Increase IOPS
					},
				},
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				existingSharedPool := createTestSharedPool()
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)
				nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{
							CpuCore:               8,
							MemoryGi:              16,
							AvailableCpuMilliCore: 7600,
							AvailableMemoryMi:     15000,
							NodeType:              dplanemgrv1.NodeType_Dedicated,
							Version:               "v1.0.0",
						},
					},
				}
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(nodePoolResp, nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateComponent, mock.Anything, mock.Anything).Return(nil)

				kubeclient := kubemocks.NewClient(t)
				kubeclient.On("Get", mock.Anything, mock.Anything, mock.IsType(new(v1alpha1.TiKVGroup))).
					Maybe().
					Run(func(args mock.Arguments) {
						tikvgroup := args.Get(2).(*v1alpha1.TiKVGroup)
						tikvgroup.Annotations = make(map[string]string)
					}).
					Return(nil)
				dataplaneManager.On("GetClient", mock.AnythingOfType("string")).Maybe().Return(kubeclient, nil)
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				// Check that the storage IOPS was updated
				tikvComp := result.ComponentGroups[1]
				assert.Equal(t, "io2", tikvComp.Storages[0].Type)
				assert.Equal(t, uint32(5000), tikvComp.Storages[0].Iops)
			},
		},
		{
			name:         "RU validation reject - node spec below minimum",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
					NodeSpec: &apiv1.NodeSpec{
						// Scale from 8c16g (existing) to 4c8g (new)
						// 4c8g supports max 5000 RU but shared pool has 10000 RU -> should fail
						Cpu:      4,
						MemoryGi: 8,
					},
				},
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				existingSharedPool := createTestSharedPool()
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)
				// Mock dataplane node specs - provides 4c8g
				nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{
							CpuCore:               4,
							MemoryGi:              8,
							AvailableCpuMilliCore: 3800,
							AvailableMemoryMi:     7600,
							NodeType:              dplanemgrv1.NodeType_Dedicated,
							Version:               "v1.0.0",
						},
					},
				}
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(nodePoolResp, nil)
				// Set strict RU config that requires 8c16g for 10000 RU
				strictRUConfig := &entity.RuConfig{
					RuConfig: apiv1.RuConfig{
						Components: map[string]*apiv1.RuConfig_ComponentsMap{
							"tikv": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
								"4c8g":  {Min: 0, Max: 5000},
								"8c16g": {Min: 5001, Max: 20000},
							}},
						},
					},
				}
				ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(strictRUConfig, nil)
				// Mock tidb cluster repo for shared pool RU calculation (returns 10000 min RU)
				tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{
					{Id: "test-cluster", RequestUnit: &apiv1.Cluster_RequestUnit{Min: 10000, Max: 20000}},
				}, nil).Once()
			},
			expectedError: "cannot support max RU",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:         "RU validation bypass with force - node spec below minimum but force true",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
					NodeSpec: &apiv1.NodeSpec{
						Cpu:      4, // 4c8g is below the minimum required for 10000 RU
						MemoryGi: 8,
					},
				},
			},
			force: true,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				existingSharedPool := createTestSharedPool()
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)
				// Mock dataplane node specs - provides 4c8g
				nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{
							CpuCore:               4,
							MemoryGi:              8,
							AvailableCpuMilliCore: 3800,
							AvailableMemoryMi:     7600,
							NodeType:              dplanemgrv1.NodeType_Dedicated,
							Version:               "v1.0.0",
						},
					},
				}
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(nodePoolResp, nil)
				dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateComponent, mock.Anything, mock.Anything).Return(nil)
				// Set strict RU config that requires 8c16g for 10000 RU
				strictRUConfig := &entity.RuConfig{
					RuConfig: apiv1.RuConfig{
						Components: map[string]*apiv1.RuConfig_ComponentsMap{
							"tikv": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
								"4c8g":  {Min: 0, Max: 5000},
								"8c16g": {Min: 5001, Max: 20000},
							}},
						},
					},
				}
				ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(strictRUConfig, nil)
				// Mock tidb cluster repo for shared pool RU calculation (returns 10000 min RU)
				tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{
					{Id: "test-cluster", RequestUnit: &apiv1.Cluster_RequestUnit{Min: 10000, Max: 20000}},
				}, nil)
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, "testsharedpoolid", result.GetId())
				assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
			},
		},
		{
			name:         "component not found",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "non-existent-component",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 1,
				},
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				existingSharedPool := createTestSharedPool()
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)
				// Note: ListNodePools is not called when component is not found, so no need to mock it
			},
			expectedError: "component id non-existent-component not found",
			expectedCode:  codes.NotFound,
		},
		{
			name:         "count-only scale-in blocked by high load - verifies clustergatewayCli call",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3, // scale from 6 to 3 (count-only scale-in)
				},
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				// Create shared pool with 6 TiKV nodes to allow scale-in to 3
				existingSharedPool := createTestSharedPoolWithTikvNodes(6)
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)
				// Need ListNodePools for genNodeSpecMap (called even for count-only scale)
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{
							CpuCore:               8,
							MemoryGi:              16,
							AvailableCpuMilliCore: 7600,
							AvailableMemoryMi:     15200,
							NodeType:              dplanemgrv1.NodeType_Dedicated,
							Version:               "v1.0.0",
						},
					},
				}, nil)
				// Note: tidbClusterRepo.List is not called for count-only scale (no spec change, so no RU validation)
			},
			setupClusterGateway: func(clusterGatewayMock *clustergatewaymocks.Client) {
				clusterGatewayMock.On("GetComponentResourceMetrics", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
					Return(&clusteragentv1.GetComponentResourceMetricsResp{
						MetricsServerAvailable: true,
						Components: []*clusteragentv1.ComponentResourceMetrics{
							{CpuCoresUsed: 6.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
							{CpuCoresUsed: 6.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
							{CpuCoresUsed: 6.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
							{CpuCoresUsed: 6.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
							{CpuCoresUsed: 6.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
							{CpuCoresUsed: 6.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
						},
					}, nil).Once()
			},
			expectedError: "scale blocked",
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:         "count-down with spec-up succeeds - R-1 fix validation",
			sharedPoolId: "testsharedpoolid",
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3, // scale from 6 to 3 (count down)
					NodeSpec: &entity.NodeSpec{ // spec up from 4c8g to 8c16g
						Cpu:      8,
						MemoryGi: 16,
					},
				},
			},
			force: false,
			setupMocks: func(
				sharedPoolRepo *repomnocks.SharedPoolRepo,
				lockRepo *repomnocks.LockRepo,
				dataplaneClient *dataplanemgrmocks.Client,
				dagExecutor *executormocks.RemoteExecutor,
				scaleManager *bractionmocks.ScaleManager,
				dataplaneManager *dataplanemocks.Manager,
				ruConfigRepo *repomnocks.RuConfigRepo,
				tidbClusterRepo *repomnocks.TiDBClusterRepo,
			) {
				// Create shared pool with 6 TiKV nodes (4c8g) to allow scale-in to 3 with spec up
				existingSharedPool := createTestSharedPoolWithTikvSpec(6, 4, 8)
				sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(existingSharedPool, nil)
				dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
				scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
				lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
					return lock.Key == "testsharedpoolid"
				})).Return(func() error { return nil }, nil)
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{
							CpuCore:               8,
							MemoryGi:              16,
							AvailableCpuMilliCore: 7600,
							AvailableMemoryMi:     15200,
							NodeType:              dplanemgrv1.NodeType_Dedicated,
							Version:               "v1.0.0",
						},
					},
				}, nil)
				dagExecutor.On("Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
				tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{
					{Id: "test-cluster", RequestUnit: &apiv1.Cluster_RequestUnit{Min: 1000, Max: 2000}},
				}, nil)
				// clustergatewayCli.GetComponentResourceMetrics will be mocked in the test runner
				// This test verifies that count-down + spec-up is allowed without metrics call
			},
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			// Create mocks
			sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
			lockRepo := repomnocks.NewLockRepo(t)
			dataplaneClient := dataplanemgrmocks.NewClient(t)
			dagExecutor := executormocks.NewRemoteExecutor(t)
			scaleManager := bractionmocks.NewScaleManager(t)
			dataplaneManager := dataplanemocks.NewManager(t)
			ruConfigRepo := repomnocks.NewRuConfigRepo(t)
			tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)

			// Setup mocks first (allows test-specific mocks to override defaults)
			tt.setupMocks(sharedPoolRepo, lockRepo, dataplaneClient, dagExecutor, scaleManager, dataplaneManager, ruConfigRepo, tidbClusterRepo)

			// Default: Use Maybe() because some tests fail early before reaching GetRuConfig call
			// This is set after setupMocks so test-specific settings take precedence
			ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Maybe().Return(testDefaultRUConfig(), nil)

			// Create clustergateway mock for scale protection tests
			clusterGatewayMock := clustergatewaymocks.NewClient(t)
			if tt.setupClusterGateway != nil {
				tt.setupClusterGateway(clusterGatewayMock)
			}
			// Create server implementation
			server := &serverImpl{
				cloudProvider:    "AWS",
				region:           "us-west-2",
				sharedPoolRepo:   sharedPoolRepo,
				lockRepo:         lockRepo,
				dataplaneManager: dataplaneManager,
				dataplaneClient:  dataplaneClient,
				dagExecutor:      dagExecutor,
				transactioner:    &mockTransactioner{},
				bizCfg: &config.BizConfig{
					DefaultDataplaneAccount: "test-account",
				},
				scaleManager:      scaleManager,
				ruConfigRepo:      ruConfigRepo,
				tidbClusterRepo:   tidbClusterRepo,
				clustergatewayCli: clusterGatewayMock,
			}

			// Execute test
			result, err := server.ScaleSharedPool(context.Background(), tt.sharedPoolId, tt.scaleComponent, tt.force)

			// Validate results
			switch {
			case tt.expectedError != "":
				require.Error(t, err)
				if tt.expectedCode != codes.OK {
					assert.Equal(t, tt.expectedCode, status.Code(err))
				}
				assert.Nil(t, result)
			case tt.validate != nil:
				tt.validate(t, result, err)
			default:
				require.NoError(t, err)
				require.NotNil(t, result)
			}
		})
	}
}

// TestMergeScaleComponent tests the mergeScaleComponent helper function
func TestMergeScaleComponent(t *testing.T) {
	newColumnarPool := func() *entity.SharedPool {
		return &entity.SharedPool{
			SharedPool: apiv1.SharedPool{
				Id: "sp-columnar",
				ComponentGroups: []*entity.Component{
					{
						Id:            "default",
						ComponentType: apiv1.ComponentType_TIFLASH_COMPUTE,
						Phase:         apiv1.Component_AVAILABLE,
						NodeQuota: &entity.NodeQuota{
							NodeCount:    0,
							MinNodeCount: 0,
							MaxNodeCount: 10,
							NodeSpec:     &entity.NodeSpec{Cpu: 4, MemoryGi: 8},
						},
					},
					{
						Id:            "default",
						ComponentType: apiv1.ComponentType_SCHEMA_MANAGER,
						Phase:         apiv1.Component_AVAILABLE,
						NodeQuota: &entity.NodeQuota{
							NodeCount:    0,
							MinNodeCount: 0,
							MaxNodeCount: 1,
							NodeSpec:     &entity.NodeSpec{Cpu: 2, MemoryGi: 8},
						},
					},
				},
			},
			TiflashMode: mysqlschema.SharedPoolTiflashModeColumnar,
		}
	}

	tests := []struct {
		name           string
		sharedPool     *entity.SharedPool
		scaleComponent *entity.Component
		nodeSpecMap    map[string][]*entity.NodeSpec
		expectedError  string
		validate       func(*testing.T, *entity.SharedPool, error)
	}{
		{
			name:       "successful node count scaling",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 4, // Scale from 2 to 4
				},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
				// Find the TiDB component and check node count
				for _, comp := range result.ComponentGroups {
					if comp.ComponentType == apiv1.ComponentType_TIDB {
						assert.Equal(t, uint32(4), comp.GetNodeQuota().GetNodeCount())
						assert.Equal(t, apiv1.Component_MODIFYING, comp.Phase)
						break
					}
				}
			},
		},
		{
			name:       "schema manager scales independently of tiflash compute",
			sharedPool: newColumnarPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_SCHEMA_MANAGER,
				NodeQuota:     &entity.NodeQuota{NodeCount: 1},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				schemaManager := findComponentByTypeAndId(result, apiv1.ComponentType_SCHEMA_MANAGER, "default")
				tiflashCompute := findComponentByTypeAndId(result, apiv1.ComponentType_TIFLASH_COMPUTE, "default")
				require.Equal(t, uint32(1), schemaManager.GetNodeQuota().GetNodeCount())
				require.Equal(t, uint32(0), tiflashCompute.GetNodeQuota().GetNodeCount())
			},
		},
		{
			name:       "tiflash compute scales independently of schema manager",
			sharedPool: newColumnarPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIFLASH_COMPUTE,
				NodeQuota:     &entity.NodeQuota{NodeCount: 1},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				tiflashCompute := findComponentByTypeAndId(result, apiv1.ComponentType_TIFLASH_COMPUTE, "default")
				schemaManager := findComponentByTypeAndId(result, apiv1.ComponentType_SCHEMA_MANAGER, "default")
				require.Equal(t, uint32(1), tiflashCompute.GetNodeQuota().GetNodeCount())
				require.Equal(t, uint32(0), schemaManager.GetNodeQuota().GetNodeCount())
			},
		},
		{
			name:       "node count exceeds maximum",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 15, // Exceeds max of 10
				},
			},
			nodeSpecMap:   createTestNodeSpecMap(),
			expectedError: "scale component node count 15 is greater than max node count 10",
		},
		{
			name:       "node count below minimum",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 1, // Below min of 3 for TiKV
				},
			},
			nodeSpecMap:   createTestNodeSpecMap(),
			expectedError: "scale component node count 1 is less than min node count 3",
		},
		{
			name: "topology worker can update min max and node count together",
			sharedPool: &entity.SharedPool{SharedPool: apiv1.SharedPool{
				Id:           "sp-topology",
				CapacityMode: apiv1.CapacityMode_TOPOLOGY_DRIVEN,
				ComponentGroups: []*apiv1.Component{
					{
						Id:            "default",
						ComponentType: apiv1.ComponentType_TIKV_WORKER,
						Phase:         apiv1.Component_UNAVAILABLE,
						NodeQuota: &apiv1.NodeQuota{
							NodeCount:    0,
							MinNodeCount: 0,
							MaxNodeCount: 0,
							NodeSpec: &apiv1.NodeSpec{
								Cpu:      8,
								MemoryGi: 16,
							},
						},
					},
				},
			}},
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV_WORKER,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount:    1,
					MinNodeCount: 1,
					MaxNodeCount: 1,
				},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				worker, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool {
					return item.ComponentType == apiv1.ComponentType_TIKV_WORKER
				})
				assert.True(t, ok)
				assert.Equal(t, uint32(1), worker.GetNodeQuota().GetNodeCount())
				assert.Equal(t, uint32(1), worker.GetNodeQuota().GetMinNodeCount())
				assert.Equal(t, uint32(1), worker.GetNodeQuota().GetMaxNodeCount())
			},
		},
		{
			name: "topology worker can scale down to zero with min max zero",
			sharedPool: &entity.SharedPool{SharedPool: apiv1.SharedPool{
				Id:           "sp-topology",
				CapacityMode: apiv1.CapacityMode_TOPOLOGY_DRIVEN,
				ComponentGroups: []*apiv1.Component{
					{
						Id:            "default",
						ComponentType: apiv1.ComponentType_TIKV_WORKER,
						Phase:         apiv1.Component_AVAILABLE,
						NodeQuota: &apiv1.NodeQuota{
							NodeCount:    1,
							MinNodeCount: 1,
							MaxNodeCount: 1,
							NodeSpec: &apiv1.NodeSpec{
								Cpu:      8,
								MemoryGi: 16,
							},
						},
					},
				},
			}},
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV_WORKER,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount:    0,
					MinNodeCount: 0,
					MaxNodeCount: 0,
				},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				worker, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool {
					return item.ComponentType == apiv1.ComponentType_TIKV_WORKER
				})
				assert.True(t, ok)
				assert.Equal(t, uint32(0), worker.GetNodeQuota().GetNodeCount())
				assert.Equal(t, uint32(0), worker.GetNodeQuota().GetMinNodeCount())
				assert.Equal(t, uint32(0), worker.GetNodeQuota().GetMaxNodeCount())
			},
		},
		{
			name:       "scale local disk storage should fail",
			sharedPool: createTestSharedPoolWithLocalDisk(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name:   apiv1.Component_Storage_DATA,
						SizeGi: 2000,
					},
				},
			},
			nodeSpecMap:   createTestNodeSpecMap(),
			expectedError: "data tikv-default do not support to modify local storage",
		},
		{
			name:       "storage not found in component",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				Storages: []*apiv1.Component_Storage{
					{
						Name:   apiv1.Component_Storage_RAFTLOG, // TiDB doesn't have raftlog
						SizeGi: 100,
					},
				},
			},
			nodeSpecMap:   createTestNodeSpecMap(),
			expectedError: "storage raftlog not found in current component tidb-default",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			// Create mocks
			ruConfigRepo := repomnocks.NewRuConfigRepo(t)
			ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Maybe().Return(testDefaultRUConfig(), nil)

			// Create server implementation with minimal setup
			server := &serverImpl{
				transactioner: &mockTransactioner{},
				ruConfigRepo:  ruConfigRepo,
			}

			// Execute test
			_, err := server.mergeScaleComponent(context.Background(), tt.sharedPool, tt.scaleComponent, tt.nodeSpecMap, false)

			// Validate results
			switch {
			case tt.expectedError != "":
				require.Error(t, err)
			case tt.validate != nil:
				tt.validate(t, tt.sharedPool, err)
			default:
				require.NoError(t, err)
			}
		})
	}
}

func TestGetMinRUForComponent_TiProxy(t *testing.T) {
	tests := []struct {
		name          string
		sharedPool    *entity.SharedPool
		component     *entity.Component
		setupMocks    func(*repomnocks.TiDBClusterRepo)
		expectedMinRU uint32
	}{
		{
			name: "shared TiProxy uses shared pool RU total",
			sharedPool: &entity.SharedPool{
				SharedPool: apiv1.SharedPool{
					Id: "shared-pool-id",
				},
				DeployType: mysqlschema.SharedPoolDeployTypeSharedTiProxy,
			},
			component: &entity.Component{
				Id:            "shared-tiproxy-component-id",
				ComponentType: apiv1.ComponentType_TIPROXY,
			},
			setupMocks: func(tidbClusterRepo *repomnocks.TiDBClusterRepo) {
				tidbClusterRepo.On("List", mock.Anything, mock.MatchedBy(func(filter *repo.TiDBClusterListFilter) bool {
					return assert.ObjectsAreEqual([]string{"shared-pool-id"}, filter.SharedPoolIds) && !filter.Deleted
				})).Return([]*entity.TiDBCluster{
					{Id: "tenant-cluster-1", RequestUnit: &apiv1.Cluster_RequestUnit{Min: 1000}},
					{Id: "tenant-cluster-2", RequestUnit: &apiv1.Cluster_RequestUnit{Min: 2000}},
				}, nil).Once()
			},
			expectedMinRU: 3000,
		},
		{
			name: "tenant-scoped TiProxy keeps loading tenant cluster by component id",
			sharedPool: &entity.SharedPool{
				SharedPool: apiv1.SharedPool{
					Id: "shared-pool-id",
				},
			},
			component: &entity.Component{
				Id:            "tenant-cluster-id",
				ComponentType: apiv1.ComponentType_TIPROXY,
			},
			setupMocks: func(tidbClusterRepo *repomnocks.TiDBClusterRepo) {
				tidbClusterRepo.On("Load", mock.Anything, "tenant-cluster-id", false).Return(
					&entity.TiDBCluster{
						Id:          "tenant-cluster-id",
						RequestUnit: &apiv1.Cluster_RequestUnit{Min: 4000},
					},
					nil,
				).Once()
			},
			expectedMinRU: 4000,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
			tt.setupMocks(tidbClusterRepo)
			server := &serverImpl{
				tidbClusterRepo: tidbClusterRepo,
			}

			minRU, err := server.getMinRUForComponent(context.Background(), tt.component, tt.sharedPool)

			require.NoError(t, err)
			assert.Equal(t, tt.expectedMinRU, minRU)
		})
	}
}

func createTestNodeSpecMap() map[string][]*entity.NodeSpec {
	return map[string][]*entity.NodeSpec{
		"4C8G": {
			{
				Cpu:               4,
				MemoryGi:          8,
				AvailableCpu:      3,
				AvailableMemoryMi: 7600,
				WithLocalDisk:     false,
			},
		},
		"8C16G": {
			{
				Cpu:               8,
				MemoryGi:          16,
				AvailableCpu:      7,
				AvailableMemoryMi: 15000,
				WithLocalDisk:     false,
			},
		},
	}
}

func createTestSharedPoolWithLocalDisk() *entity.SharedPool {
	sharedPool := createTestSharedPool()
	// Set TiKV data storage to local
	sharedPool.ComponentGroups[1].Storages[0].Type = entity.LocalSSD
	return sharedPool
}

func createTestSharedPool() *entity.SharedPool {
	return &entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            "testsharedpoolid",
			DataplaneName: "test-dataplane",
			Phase:         apiv1.SharedPool_AVAILABLE,
			ComponentGroups: []*apiv1.Component{
				{
					Id:            "default",
					ComponentType: apiv1.ComponentType_TIDB,
					Phase:         apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeCount:    2,
						MinNodeCount: 1,
						MaxNodeCount: 10,
						NodeSpec: &apiv1.NodeSpec{
							Cpu:      4,
							MemoryGi: 8,
						},
					},
					Storages: []*apiv1.Component_Storage{
						{
							Name:   apiv1.Component_Storage_DATA,
							Type:   "gp3",
							SizeGi: 100,
						},
					},
				},
				{
					Id:            "default",
					ComponentType: apiv1.ComponentType_TIKV,
					Phase:         apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeCount:    3,
						MinNodeCount: 3,
						MaxNodeCount: 15,
						NodeSpec: &apiv1.NodeSpec{
							Cpu:      8,
							MemoryGi: 16,
						},
					},
					Storages: []*apiv1.Component_Storage{
						{
							Name:   apiv1.Component_Storage_DATA,
							Type:   "gp3",
							SizeGi: 500,
							Iops:   3000,
						},
						{
							Name:   apiv1.Component_Storage_RAFTLOG,
							Type:   "gp3",
							SizeGi: 50,
							Iops:   3000,
						},
					},
				},
			},
		},
	}
}

// createTestSharedPoolWithTikvNodes creates a shared pool with specified TiKV node count (for scale-in tests)
func createTestSharedPoolWithTikvNodes(tikvNodes uint32) *entity.SharedPool {
	sharedPool := createTestSharedPool()
	// Find TiKV component and set node count
	for _, comp := range sharedPool.ComponentGroups {
		if comp.GetComponentType() == apiv1.ComponentType_TIKV {
			comp.NodeQuota.NodeCount = tikvNodes
			break
		}
	}
	return sharedPool
}

// createTestSharedPoolWithTikvSpec creates a shared pool with specified TiKV node count and spec (for R-1 fix tests)
func createTestSharedPoolWithTikvSpec(tikvNodes, cpu, memoryGi uint32) *entity.SharedPool {
	sharedPool := createTestSharedPool()
	// Find TiKV component and set node count and spec
	for _, comp := range sharedPool.ComponentGroups {
		if comp.GetComponentType() == apiv1.ComponentType_TIKV {
			comp.NodeQuota.NodeCount = tikvNodes
			comp.NodeQuota.NodeSpec.Cpu = cpu
			comp.NodeQuota.NodeSpec.MemoryGi = memoryGi
			break
		}
	}
	return sharedPool
}

func TestMergeScaleComponent_NodeSpecScaling(t *testing.T) {
	// Create mocks for repos
	tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
	tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{
		{Id: "test-cluster", RequestUnit: &apiv1.Cluster_RequestUnit{Min: 1000, Max: 2000}},
	}, nil)

	ruConfigRepo := repomnocks.NewRuConfigRepo(t)
	ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Maybe().Return(testDefaultRUConfig(), nil)

	server := &serverImpl{
		bizCfg:          &config.BizConfig{},
		tidbClusterRepo: tidbClusterRepo,
		ruConfigRepo:    ruConfigRepo,
	}

	tests := []struct {
		name           string
		sharedPool     *entity.SharedPool
		scaleComponent *entity.Component
		nodeSpecMap    map[string][]*entity.NodeSpec
		expectedError  string
		validate       func(*testing.T, *entity.SharedPool, error)
	}{
		{
			name:       "scale with node spec - cpu and memory both set",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV, // Use TiKV to avoid needing tidbGroupRepo
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 6, // Must be multiple of 3 for TiKV
					NodeSpec: &apiv1.NodeSpec{
						Cpu:      8,
						MemoryGi: 16,
					},
				},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				tikvComp, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool {
					return item.ComponentType == apiv1.ComponentType_TIKV
				})
				assert.True(t, ok)
				assert.Equal(t, uint32(8), tikvComp.GetNodeQuota().GetNodeSpec().GetCpu())
				assert.Equal(t, uint32(16), tikvComp.GetNodeQuota().GetNodeSpec().GetMemoryGi())
			},
		},
		{
			name:       "scale node count only - no node spec",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
					// NodeSpec is nil - should only scale node count
				},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				tidbComp, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool {
					return item.ComponentType == apiv1.ComponentType_TIDB
				})
				assert.True(t, ok)
				assert.Equal(t, uint32(3), tidbComp.GetNodeQuota().GetNodeCount())
				// Node spec should remain unchanged (4C8G)
				assert.Equal(t, uint32(4), tidbComp.GetNodeQuota().GetNodeSpec().GetCpu())
				assert.Equal(t, uint32(8), tidbComp.GetNodeQuota().GetNodeSpec().GetMemoryGi())
			},
		},
		{
			name:       "scale with node spec - only cpu set (zero memory) should return error",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
					NodeSpec: &apiv1.NodeSpec{
						Cpu:      8,
						MemoryGi: 0, // Zero memory - should return error
					},
				},
			},
			nodeSpecMap:   createTestNodeSpecMap(),
			expectedError: "node spec must specify both cpu and memory",
		},
		{
			name:       "scale with node spec - only memory set (zero cpu) should return error",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
					NodeSpec: &apiv1.NodeSpec{
						Cpu:      0, // Zero cpu - should return error
						MemoryGi: 16,
					},
				},
			},
			nodeSpecMap:   createTestNodeSpecMap(),
			expectedError: "node spec must specify both cpu and memory",
		},
		{
			name:       "scale with node spec - both cpu and memory zero should skip node spec update",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
					NodeSpec: &apiv1.NodeSpec{
						Cpu:      0,
						MemoryGi: 0,
					},
				},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, result *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				tidbComp, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool {
					return item.ComponentType == apiv1.ComponentType_TIDB
				})
				assert.True(t, ok)
				// Node spec should remain unchanged
				assert.Equal(t, uint32(4), tidbComp.GetNodeQuota().GetNodeSpec().GetCpu())
				assert.Equal(t, uint32(8), tidbComp.GetNodeQuota().GetNodeSpec().GetMemoryGi())
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			_, err := server.mergeScaleComponent(context.Background(), tt.sharedPool, tt.scaleComponent, tt.nodeSpecMap, false)

			if tt.expectedError != "" {
				require.Error(t, err)
				assert.Contains(t, err.Error(), tt.expectedError)
			} else {
				if tt.validate != nil {
					tt.validate(t, tt.sharedPool, err)
				} else {
					require.NoError(t, err)
				}
			}
		})
	}
}

func TestMergeScaleComponent_StorageConversionConstraints(t *testing.T) {
	server := &serverImpl{
		bizCfg: &config.BizConfig{},
	}

	tests := []struct {
		name           string
		sharedPool     *entity.SharedPool
		scaleComponent *entity.Component
		nodeSpecMap    map[string][]*entity.NodeSpec
		expectedError  string
		validate       func(*testing.T, *entity.SharedPool, error)
	}{
		{
			name:       "prevent local disk conversion to gp3",
			sharedPool: createTestSharedPoolWithLocalDisk(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name: apiv1.Component_Storage_DATA,
						Type: entity.GP3,
						Iops: 8000,
					},
				},
			},
			nodeSpecMap:   createTestNodeSpecMap(),
			expectedError: "DATA tidb-default do not support to modify local storage",
		},
		{
			name:       "prevent local disk conversion to io2",
			sharedPool: createTestSharedPoolWithLocalDisk(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name: apiv1.Component_Storage_DATA,
						Type: entity.IO2,
						Iops: 5000,
					},
				},
			},
			nodeSpecMap:   createTestNodeSpecMap(),
			expectedError: "DATA tidb-default do not support to modify local storage",
		},
		{
			name:       "allow gp3 to io2 conversion with custom iops",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name: apiv1.Component_Storage_DATA,
						Type: entity.IO2,
						Iops: 8000,
					},
				},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, sharedPool *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				// Verify storage type was updated
				tikvGroup, ok := lo.Find(sharedPool.ComponentGroups, func(item *apiv1.Component) bool {
					return item.ComponentType == apiv1.ComponentType_TIKV
				})
				assert.True(t, ok)
				dataStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool {
					return item.Name == apiv1.Component_Storage_DATA
				})
				assert.True(t, ok)
				assert.Equal(t, entity.IO2, dataStorage.Type)
				assert.Equal(t, uint32(8000), dataStorage.Iops)
			},
		},
		{
			name:       "allow io2 to gp3 conversion",
			sharedPool: createTestSharedPoolWithIO2Storage(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name: apiv1.Component_Storage_DATA,
						Type: entity.GP3,
						Iops: 12000,
					},
				},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, sharedPool *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				// Verify storage type was updated
				tikvGroup, ok := lo.Find(sharedPool.ComponentGroups, func(item *apiv1.Component) bool {
					return item.ComponentType == apiv1.ComponentType_TIKV
				})
				assert.True(t, ok)
				dataStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool {
					return item.Name == apiv1.Component_Storage_DATA
				})
				assert.True(t, ok)
				assert.Equal(t, entity.GP3, dataStorage.Type)
				assert.Equal(t, uint32(12000), dataStorage.Iops)
			},
		},
		{
			name:       "tikv io2 storage with non-zero iops should preserve iops",
			sharedPool: createTestSharedPool(),
			scaleComponent: &entity.Component{
				Id:            "default",
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name: apiv1.Component_Storage_DATA,
						Type: entity.IO2,
						// Iops is non-zero, should preserve the value
						Iops: 8000,
					},
				},
			},
			nodeSpecMap: createTestNodeSpecMap(),
			validate: func(t *testing.T, sharedPool *entity.SharedPool, err error) {
				t.Helper()
				require.NoError(t, err)
				// Verify that custom IOPS was preserved
				tikvGroup, ok := lo.Find(sharedPool.ComponentGroups, func(item *apiv1.Component) bool {
					return item.ComponentType == apiv1.ComponentType_TIKV
				})
				assert.True(t, ok)
				dataStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool {
					return item.Name == apiv1.Component_Storage_DATA
				})
				assert.True(t, ok)
				assert.Equal(t, entity.IO2, dataStorage.Type)
				assert.Equal(t, uint32(8000), dataStorage.Iops) // Should preserve custom IOPS
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			_, err := server.mergeScaleComponent(context.Background(), tt.sharedPool, tt.scaleComponent, tt.nodeSpecMap, false)

			if tt.expectedError != "" {
				require.Error(t, err)
			} else {
				if tt.validate != nil {
					tt.validate(t, tt.sharedPool, err)
				} else {
					require.NoError(t, err)
				}
			}
		})
	}
}

// Helper function to create test shared pool with IO2 storage
func createTestSharedPoolWithIO2Storage() *entity.SharedPool {
	sharedPool := createTestSharedPool()
	// Set TiKV data storage to IO2
	sharedPool.ComponentGroups[1].Storages[0].Type = entity.IO2
	sharedPool.ComponentGroups[1].Storages[0].Iops = 8000
	return sharedPool
}

func TestCheckScaleComponentValid(t *testing.T) {
	tests := []struct {
		name           string
		scaleComponent *entity.Component
		expectedError  string
		expectedCode   codes.Code
	}{
		{
			name:           "nil component",
			scaleComponent: nil,
			expectedError:  "scale components is empty",
			expectedCode:   codes.InvalidArgument,
		},
		{
			name: "unspecified component type",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED,
			},
			expectedError: "scale component type is required",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "valid tidb component",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 2,
				},
			},
		},
		{
			name: "worker zero count defer check in pre-validation",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIDB_WORKER,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 0,
				},
			},
		},
		{
			name: "invalid non-tiflash component with zero nodes",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIDB,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 0,
				},
			},
			expectedError: "scale component node count is required, only TiFlash node count can be zero",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "tikv component with invalid node count (not multiple of 3)",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 2, // Should be multiple of 3
				},
			},
			expectedError: "scale component tikv node count must be multiple of 3",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "pd component with invalid node count (not multiple of 3)",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_PD,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 4, // Should be multiple of 3
				},
			},
			expectedError: "scale component pd node count must be multiple of 3",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "valid tikv component with multiple of 3 nodes",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 6, // Multiple of 3
				},
			},
		},
		{
			name: "gp3 storage with missing iops",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type:   entity.GP3,
						SizeGi: 100,
						Iops:   0, // Missing IOPS
					},
				},
			},
			expectedError: "gp3 storage iops must be greater than zero",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "gp3 storage with missing throughput",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type:          entity.GP3,
						SizeGi:        100,
						Iops:          3000,
						ThroughputMbs: 0, // Missing throughput
					},
				},
			},
			expectedError: "gp3 storage throughput must be greater than zero",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "gp3 storage with missing size",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type:          entity.GP3,
						SizeGi:        0, // Missing size
						Iops:          3000,
						ThroughputMbs: 125,
					},
				},
			},
			expectedError: "gp3 storage size must be greater than zero",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "valid gp3 storage",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type:          entity.GP3,
						SizeGi:        100,
						Iops:          3000,
						ThroughputMbs: 125,
					},
				},
			},
		},
		{
			name: "io2 storage with missing iops",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type:   entity.IO2,
						SizeGi: 100,
						Iops:   0, // Missing IOPS
					},
				},
			},
			expectedError: "io2 storage iops must be greater than zero",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "io2 storage with missing size",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type:   entity.IO2,
						SizeGi: 0, // Missing size
						Iops:   5000,
					},
				},
			},
			expectedError: "io2 storage size must be greater than zero",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "io2 storage with non-zero throughput",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type:          entity.IO2,
						SizeGi:        100,
						Iops:          5000,
						ThroughputMbs: 125, // Should be zero for IO2
					},
				},
			},
			expectedError: "io2 storage does not support setting throughput",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "valid io2 storage",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type:   entity.IO2,
						SizeGi: 100,
						Iops:   5000,
					},
				},
			},
		},
		{
			name: "valid local ssd storage",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type: entity.LocalSSD,
						// Local SSD doesn't require size/iops validation
					},
				},
			},
		},
		{
			name: "invalid storage type",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type: "unsupported-storage-type",
					},
				},
			},
			expectedError: "invalid storage type 'unsupported-storage-type' for AWS",
			expectedCode:  codes.InvalidArgument,
		},
		{
			name: "tiflash write component with zero nodes",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIFLASH_WRITE,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 0, // TiFlash Write can have zero nodes
				},
			},
		},
		{
			name: "tiflash compute component with zero nodes",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIFLASH_COMPUTE,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 0, // TiFlash Compute can have zero nodes
				},
			},
		},
		{
			name: "storage type empty string - should skip validation",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Type: "", // Empty storage type should skip validation
					},
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			err := checkScaleComponentValid(context.Background(), tt.scaleComponent, "AWS")

			if tt.expectedError != "" {
				require.Error(t, err)
				assert.Contains(t, err.Error(), tt.expectedError)
				assert.Equal(t, tt.expectedCode, status.Code(err))
			} else {
				require.NoError(t, err)
			}
		})
	}
}

func TestNeedSendMeteringMetadata(t *testing.T) {
	tests := []struct {
		name           string
		scaleComponent *entity.Component
		expected       bool
	}{
		{
			name:           "nil component",
			scaleComponent: nil,
			expected:       false,
		},
		{
			name: "non-tikv component",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIDB,
				Storages: []*apiv1.Component_Storage{
					{
						Name: apiv1.Component_Storage_DATA,
						Type: "io2",
						Iops: 5000,
					},
				},
			},
			expected: false,
		},
		{
			name: "tikv with data storage type change",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name: apiv1.Component_Storage_DATA,
						Type: "io2", // Type change triggers metering
					},
				},
			},
			expected: true,
		},
		{
			name: "tikv with data storage iops change",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name: apiv1.Component_Storage_DATA,
						Iops: 5000, // IOPS change triggers metering
					},
				},
			},
			expected: true,
		},
		{
			name: "tikv with data storage throughput change",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name:          apiv1.Component_Storage_DATA,
						ThroughputMbs: 125, // Throughput change triggers metering
					},
				},
			},
			expected: true,
		},
		{
			name: "tikv with non-data storage change",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name: apiv1.Component_Storage_RAFTLOG, // Non-data storage
						Type: "io2",
						Iops: 5000,
					},
				},
			},
			expected: false,
		},
		{
			name: "tikv without storage changes",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 6, // Only node count change, no storage
				},
			},
			expected: false,
		},
		{
			name: "tikv with empty data storage changes",
			scaleComponent: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				Storages: []*apiv1.Component_Storage{
					{
						Name: apiv1.Component_Storage_DATA,
						// All fields empty/zero - no changes
					},
				},
			},
			expected: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := needSendMeteringMetadata(tt.scaleComponent)
			assert.Equal(t, tt.expected, result)
		})
	}
}

// testDefaultRUConfig returns a default RU config for testing that allows any spec
func testDefaultRUConfig() *entity.RuConfig {
	return &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"tidb":               {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
				"tikv":               {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
				"pd":                 {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
				"tiflash_compute":    {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
				"tiflash_write":      {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
				"tiproxy":            {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
				"pd_tso":             {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
				"pd_scheduling":      {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
				"tikv_worker":        {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
				"coprocessor_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
				"tidb_worker":        {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"2c4g": {Min: 0, Max: 100000}}},
			},
		},
	}
}

func TestCheckScaleInSafety(t *testing.T) {
	tests := []struct {
		name              string
		force             bool
		currentCount      uint32
		currentSpec       *entity.NodeSpec
		targetCount       uint32
		targetSpec        *entity.NodeSpec
		componentType     apiv1.ComponentType // defaults to TIKV if not set
		metricsResponse   *clusteragentv1.GetComponentResourceMetricsResp
		metricsError      error
		rawMetrics        map[string][]byte
		rawMetricsError   map[string]error
		expectedError     bool
		expectedCode      codes.Code
		expectedMessage   string
		unexpectedMessage string
	}{
		{
			name:          "force flag bypasses check",
			force:         true,
			currentCount:  3,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   2,
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			expectedError: false,
		},
		{
			name:          "capacity not decreased - scale up allowed",
			force:         false,
			currentCount:  2,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   3,
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			expectedError: false,
		},
		{
			name:          "capacity not decreased - spec upgrade allowed",
			force:         false,
			currentCount:  2,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   2,
			targetSpec:    &entity.NodeSpec{Cpu: 16, MemoryGi: 32},
			expectedError: false,
		},
		{
			name:          "current count zero skips metrics probe",
			force:         false,
			currentCount:  0,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   0,
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			expectedError: false,
		},
		{
			name:         "safe scale in - low usage",
			force:        false,
			currentCount: 3,
			currentSpec:  &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:  2,
			targetSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{CpuCoresUsed: 1.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
					{CpuCoresUsed: 1.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
					{CpuCoresUsed: 1.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
				},
			},
			expectedError: false,
		},
		{
			name:          "unsafe TiKV horizontal scale in - block cache metrics unavailable",
			force:         false,
			currentCount:  3,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   2,
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			componentType: apiv1.ComponentType_TIKV,
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{PodName: "db-tikv-0", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-1", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-2", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
				},
			},
			rawMetricsError: map[string]error{
				"db-tikv-0": fmt.Errorf("metrics proxy unavailable"),
				"db-tikv-1": fmt.Errorf("metrics proxy unavailable"),
				"db-tikv-2": fmt.Errorf("metrics proxy unavailable"),
			},
			expectedError: true,
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:          "unsafe TiKV horizontal scale in - metrics pod count mismatch",
			force:         false,
			currentCount:  3,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   2,
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			componentType: apiv1.ComponentType_TIKV,
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{PodName: "db-tikv-0", CpuCoresUsed: 1.0, MemoryBytesUsed: 14 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-1", CpuCoresUsed: 1.0, MemoryBytesUsed: 14 * 1024 * 1024 * 1024},
				},
			},
			rawMetrics: map[string][]byte{
				"db-tikv-0": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 15032385536\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 10737418240\n"),
				"db-tikv-1": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 15032385536\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 10737418240\n"),
			},
			expectedError: true,
			expectedCode:  codes.FailedPrecondition,
			// Adjustment was skipped, so the message must not claim the
			// reported memory excludes reclaimable block cache.
			unexpectedMessage: "non-reclaimable",
		},
		{
			name:          "unsafe TiKV horizontal scale in - partial raw metrics failure",
			force:         false,
			currentCount:  3,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   2,
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			componentType: apiv1.ComponentType_TIKV,
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{PodName: "db-tikv-0", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-1", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-2", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
				},
			},
			rawMetrics: map[string][]byte{
				"db-tikv-0": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 12884901888\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 11811160064\n"),
				"db-tikv-1": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 12884901888\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 11811160064\n"),
			},
			rawMetricsError: map[string]error{
				"db-tikv-2": fmt.Errorf("metrics proxy unavailable"),
			},
			expectedError: true,
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:          "unsafe TiKV horizontal scale in - non-cache memory remains too high",
			force:         false,
			currentCount:  3,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   2,
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			componentType: apiv1.ComponentType_TIKV,
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{PodName: "db-tikv-0", CpuCoresUsed: 1.0, MemoryBytesUsed: 14 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-1", CpuCoresUsed: 1.0, MemoryBytesUsed: 14 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-2", CpuCoresUsed: 1.0, MemoryBytesUsed: 14 * 1024 * 1024 * 1024},
				},
			},
			rawMetrics: map[string][]byte{
				"db-tikv-0": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 15032385536\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"default\"} 1073741824\n"),
				"db-tikv-1": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 15032385536\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"default\"} 1073741824\n"),
				"db-tikv-2": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 15032385536\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"default\"} 1073741824\n"),
			},
			expectedError:   true,
			expectedCode:    codes.FailedPrecondition,
			expectedMessage: "non-reclaimable memory demand",
		},
		{
			name:          "unsafe TiKV horizontal scale in - newer resident memory covers cache growth",
			force:         false,
			currentCount:  3,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   2,
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			componentType: apiv1.ComponentType_TIKV,
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{PodName: "db-tikv-0", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-1", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-2", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
				},
			},
			rawMetrics: map[string][]byte{
				"db-tikv-0": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 21474836480\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 16106127360\n"),
				"db-tikv-1": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 21474836480\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 16106127360\n"),
				"db-tikv-2": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 21474836480\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 16106127360\n"),
			},
			expectedError: true,
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:         "unsafe scale in - high cpu usage",
			force:        false,
			currentCount: 3,
			currentSpec:  &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:  2,
			targetSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{CpuCoresUsed: 6.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
					{CpuCoresUsed: 6.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
					{CpuCoresUsed: 6.0, MemoryBytesUsed: 2 * 1024 * 1024 * 1024},
				},
			},
			expectedError: true,
			expectedCode:  codes.FailedPrecondition,
			// CPU is the blocker, so the adjustment never runs and the reported
			// memory is the raw working set.
			unexpectedMessage: "non-reclaimable",
		},
		{
			name:         "unsafe scale in - high memory usage",
			force:        false,
			currentCount: 3,
			currentSpec:  &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:  2,
			targetSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
					{CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
					{CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
				},
			},
			expectedError: true,
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:          "safe TiKV horizontal scale in - block cache is reclaimable",
			force:         false,
			currentCount:  3,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   2,
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			componentType: apiv1.ComponentType_TIKV,
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{PodName: "db-tikv-0", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-1", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
					{PodName: "db-tikv-2", CpuCoresUsed: 1.0, MemoryBytesUsed: 12 * 1024 * 1024 * 1024},
				},
			},
			rawMetrics: map[string][]byte{
				"db-tikv-0": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 12884901888\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 11811160064\n"),
				"db-tikv-1": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 12884901888\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 11811160064\n"),
				"db-tikv-2": []byte("# TYPE process_resident_memory_bytes gauge\n" +
					"process_resident_memory_bytes 12884901888\n" +
					"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
					"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 11811160064\n"),
			},
			expectedError: false,
		},
		{
			name:         "metrics server unavailable",
			force:        false,
			currentCount: 3,
			currentSpec:  &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:  2,
			targetSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: false,
			},
			expectedError: true,
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:          "metrics fetch error",
			force:         false,
			currentCount:  3,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   2,
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			metricsError:  fmt.Errorf("connection refused"),
			expectedError: true,
		},
		{
			name:          "TiFlash scale to 0 without force - should be blocked",
			force:         false,
			currentCount:  2,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   0, // Scale to 0
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			componentType: apiv1.ComponentType_TIFLASH_WRITE,
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{CpuCoresUsed: 0.5, MemoryBytesUsed: 1 * 1024 * 1024 * 1024},
				},
			},
			expectedError: true, // Should be blocked (any component scale to 0 must use force)
			expectedCode:  codes.FailedPrecondition,
		},
		{
			name:          "TiFlash scale to 0 with force - should be allowed",
			force:         true,
			currentCount:  2,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   0, // Scale to 0
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			componentType: apiv1.ComponentType_TIFLASH_WRITE,
			// No metricsResponse needed: force=true skips metrics check
			expectedError: false, // Should be allowed with force
		},
		{
			name:          "TiKV scale to 0 with residual load - should be blocked",
			force:         false,
			currentCount:  2,
			currentSpec:   &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			targetCount:   0, // Scale to 0
			targetSpec:    &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
			componentType: apiv1.ComponentType_TIKV, // TiKV component - should NOT be exempted
			metricsResponse: &clusteragentv1.GetComponentResourceMetricsResp{
				MetricsServerAvailable: true,
				Components: []*clusteragentv1.ComponentResourceMetrics{
					{CpuCoresUsed: 2.0, MemoryBytesUsed: 4 * 1024 * 1024 * 1024},
					{CpuCoresUsed: 2.0, MemoryBytesUsed: 4 * 1024 * 1024 * 1024},
				},
			},
			expectedError: true, // Should be blocked (TiKV cannot scale to 0)
			expectedCode:  codes.FailedPrecondition,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			// Setup mocks
			clusterGatewayMock := clustergatewaymocks.NewClient(t)
			if tt.metricsResponse != nil || tt.metricsError != nil {
				clusterGatewayMock.On("GetComponentResourceMetrics", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
					Return(tt.metricsResponse, tt.metricsError)
			}
			for podName, rawMetrics := range tt.rawMetrics {
				clusterGatewayMock.On("GetComponentRawMetrics", mock.Anything, mock.Anything, "tikv", podName).
					Return(rawMetrics, nil)
			}
			for podName, rawMetricsErr := range tt.rawMetricsError {
				clusterGatewayMock.On("GetComponentRawMetrics", mock.Anything, mock.Anything, "tikv", podName).
					Return(nil, rawMetricsErr)
			}

			server := &serverImpl{
				clustergatewayCli: clusterGatewayMock,
			}

			sharedPool := &entity.SharedPool{
				SharedPool: apiv1.SharedPool{
					Id:            "test-pool",
					DataplaneName: "test-dp",
				},
			}
			compType := tt.componentType
			if compType == apiv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED {
				compType = apiv1.ComponentType_TIKV
			}
			component := &entity.Component{
				Id:            "test-comp",
				ComponentType: compType,
			}

			err := server.checkScaleInSafety(
				context.Background(),
				sharedPool,
				component,
				tt.currentCount,
				tt.currentSpec,
				tt.targetCount,
				tt.targetSpec,
				tt.force,
			)

			if tt.expectedError {
				require.Error(t, err)
				if tt.expectedCode != 0 {
					assert.Equal(t, tt.expectedCode, status.Code(err))
				}
				if tt.expectedMessage != "" {
					assert.Contains(t, err.Error(), tt.expectedMessage)
				}
				if tt.unexpectedMessage != "" {
					assert.NotContains(t, err.Error(), tt.unexpectedMessage)
				}
			} else {
				require.NoError(t, err)
			}

			clusterGatewayMock.AssertExpectations(t)
		})
	}
}

func TestParseTiKVMemoryMetrics(t *testing.T) {
	tests := []struct {
		name             string
		raw              string
		expectedResident uint64
		expectedCache    uint64
		expectErr        bool
	}{
		{
			name: "modern shared cache metric ignores duplicate raft sample",
			raw: "# TYPE process_resident_memory_bytes gauge\n" +
				"process_resident_memory_bytes 6442450944\n" +
				"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
				"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 4294967296\n" +
				"tikv_engine_block_cache_size_bytes{db=\"raft\",cf=\"all\"} 4294967296\n",
			expectedResident: 6442450944,
			expectedCache:    4294967296,
		},
		{
			name: "shared cache takes precedence over per-cf samples",
			raw: "# TYPE process_resident_memory_bytes gauge\n" +
				"process_resident_memory_bytes 6442450944\n" +
				"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
				"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 4294967296\n" +
				"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"default\"} 2147483648\n",
			expectedResident: 6442450944,
			expectedCache:    4294967296,
		},
		{
			name: "separate kv column family caches are summed",
			raw: "# TYPE process_resident_memory_bytes gauge\n" +
				"process_resident_memory_bytes 6442450944\n" +
				"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
				"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"default\"} 2147483648\n" +
				"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"write\"} 1073741824\n",
			expectedResident: 6442450944,
			expectedCache:    3221225472,
		},
		{
			name:      "missing resident memory metric returns error",
			raw:       "tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 1073741824\n",
			expectErr: true,
		},
		{
			name: "missing kv cache metric returns error",
			raw: "# TYPE process_resident_memory_bytes gauge\n" +
				"process_resident_memory_bytes 2147483648\n" +
				"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
				"tikv_engine_block_cache_size_bytes{db=\"raft\",cf=\"default\"} 1073741824\n",
			expectErr: true,
		},
		{
			name: "per-cf cache sum overflow returns error",
			raw: "# TYPE process_resident_memory_bytes gauge\n" +
				"process_resident_memory_bytes 18446744073709549568\n" +
				"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
				"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"default\"} 18446744073709549568\n" +
				"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"write\"} 4096\n",
			expectErr: true,
		},
		{
			name:      "malformed metrics return error",
			raw:       "not prometheus metrics",
			expectErr: true,
		},
		{
			name: "partial metrics body returns error",
			raw: "# TYPE process_resident_memory_bytes gauge\n" +
				"process_resident_memory_bytes 6442450944\n" +
				"# TYPE tikv_engine_block_cache_size_bytes gauge\n" +
				"tikv_engine_block_cache_size_bytes{db=\"kv\",cf=\"all\"} 4294967296\n" +
				"malformed trailing line",
			expectErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			actual, err := parseTiKVMemoryMetrics([]byte(tt.raw))
			if tt.expectErr {
				require.Error(t, err)
				return
			}
			require.NoError(t, err)
			require.Equal(t, tt.expectedResident, actual.residentBytes)
			require.Equal(t, tt.expectedCache, actual.blockCacheBytes)
		})
	}
}

func TestReclaimableTiKVBlockCacheBytes(t *testing.T) {
	blockCacheBytes := []float64{8, 2, 6, 0}

	require.InDelta(t, float64(0), reclaimableTiKVBlockCacheBytes(blockCacheBytes, 4, 1), 1e-9)
	require.InDelta(t, float64(2), reclaimableTiKVBlockCacheBytes(blockCacheBytes, 4, 2), 1e-9)
	require.InDelta(t, float64(8), reclaimableTiKVBlockCacheBytes(blockCacheBytes, 4, 3), 1e-9)
	require.InDelta(t, float64(16), reclaimableTiKVBlockCacheBytes(blockCacheBytes, 4, 4), 1e-9)
	require.InDelta(t, float64(16), reclaimableTiKVBlockCacheBytes(blockCacheBytes, 4, 5), 1e-9)
	require.InDelta(t, float64(0), reclaimableTiKVBlockCacheBytes(blockCacheBytes[:3], 4, 1), 1e-9)
	require.Equal(t, []float64{8, 2, 6, 0}, blockCacheBytes, "input must not be mutated")
}

// TestValidateScaleRequest tests the scale request validation
func TestValidateScaleRequest(t *testing.T) {
	tests := []struct {
		name          string
		sharedPoolId  string
		scaleComp     *entity.Component
		expectedError bool
		expectedCode  codes.Code
	}{
		{
			name:         "valid request",
			sharedPoolId: "test-pool",
			scaleComp: &entity.Component{
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: 3,
				},
			},
			expectedError: false,
		},
		{
			name:          "empty shared pool id",
			sharedPoolId:  "",
			scaleComp:     &entity.Component{ComponentType: apiv1.ComponentType_TIKV},
			expectedError: true,
			expectedCode:  codes.InvalidArgument,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			server := &serverImpl{cloudProvider: "aws"}
			err := server.validateScaleRequest(context.Background(), tt.sharedPoolId, tt.scaleComp)

			if tt.expectedError {
				require.Error(t, err)
				assert.Equal(t, tt.expectedCode, status.Code(err))
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

// TestFindComponentInSharedPool tests finding a component in shared pool
func TestFindComponentInSharedPool(t *testing.T) {
	tests := []struct {
		name          string
		components    []*entity.Component
		scaleComp     *entity.Component
		expectedFound bool
	}{
		{
			name: "component found",
			components: []*entity.Component{
				{Id: "comp-1", ComponentType: apiv1.ComponentType_TIKV},
				{Id: "comp-2", ComponentType: apiv1.ComponentType_TIDB},
			},
			scaleComp:     &entity.Component{Id: "comp-1", ComponentType: apiv1.ComponentType_TIKV},
			expectedFound: true,
		},
		{
			name: "component not found - wrong id",
			components: []*entity.Component{
				{Id: "comp-1", ComponentType: apiv1.ComponentType_TIKV},
			},
			scaleComp:     &entity.Component{Id: "comp-2", ComponentType: apiv1.ComponentType_TIKV},
			expectedFound: false,
		},
		{
			name: "component not found - wrong type",
			components: []*entity.Component{
				{Id: "comp-1", ComponentType: apiv1.ComponentType_TIKV},
			},
			scaleComp:     &entity.Component{Id: "comp-1", ComponentType: apiv1.ComponentType_TIDB},
			expectedFound: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			server := &serverImpl{}
			sharedPool := &entity.SharedPool{
				SharedPool: apiv1.SharedPool{
					ComponentGroups: tt.components,
				},
			}

			comp, err := server.findComponentInSharedPool(sharedPool, tt.scaleComp)

			if tt.expectedFound {
				require.NoError(t, err)
				assert.NotNil(t, comp)
				assert.Equal(t, tt.scaleComp.GetId(), comp.GetId())
				assert.Equal(t, tt.scaleComp.GetComponentType(), comp.GetComponentType())
			} else {
				require.Error(t, err)
				assert.Equal(t, codes.NotFound, status.Code(err))
			}
		})
	}
}

// TestValidateNodeCountForComponent tests node count validation for different components
func TestValidateNodeCountForComponent(t *testing.T) {
	tests := []struct {
		name          string
		componentType apiv1.ComponentType
		nodeCount     uint32
		expectedError bool
		expectedCode  codes.Code
	}{
		{
			name:          "tikv valid count",
			componentType: apiv1.ComponentType_TIKV,
			nodeCount:     6,
			expectedError: false,
		},
		{
			name:          "tikv invalid count",
			componentType: apiv1.ComponentType_TIKV,
			nodeCount:     4,
			expectedError: true,
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:          "pd valid count",
			componentType: apiv1.ComponentType_PD,
			nodeCount:     3,
			expectedError: false,
		},
		{
			name:          "pd invalid count",
			componentType: apiv1.ComponentType_PD,
			nodeCount:     2,
			expectedError: true,
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:          "pd_tso valid count",
			componentType: apiv1.ComponentType_PD_TSO,
			nodeCount:     2,
			expectedError: false,
		},
		{
			name:          "pd_tso invalid count",
			componentType: apiv1.ComponentType_PD_TSO,
			nodeCount:     1,
			expectedError: true,
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:          "tiflash can be zero",
			componentType: apiv1.ComponentType_TIFLASH_WRITE,
			nodeCount:     0,
			expectedError: false,
		},
		{
			name:          "tidb cannot be zero",
			componentType: apiv1.ComponentType_TIDB,
			nodeCount:     0,
			expectedError: true,
			expectedCode:  codes.InvalidArgument,
		},
		{
			name:          "topology driven tidb worker can be zero",
			componentType: apiv1.ComponentType_TIDB_WORKER,
			nodeCount:     0,
			expectedError: false,
		},
		{
			name:          "topology driven tikv worker can be zero",
			componentType: apiv1.ComponentType_TIKV_WORKER,
			nodeCount:     0,
			expectedError: false,
		},
		{
			name:          "schema manager can be zero",
			componentType: apiv1.ComponentType_SCHEMA_MANAGER,
			nodeCount:     0,
			expectedError: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			scaleComp := &entity.Component{
				ComponentType: tt.componentType,
				NodeQuota: &apiv1.NodeQuota{
					NodeCount: tt.nodeCount,
				},
			}

			sharedPool := &entity.SharedPool{
				SharedPool: apiv1.SharedPool{CapacityMode: apiv1.CapacityMode_RCU_DRIVEN},
			}
			if tt.componentType == apiv1.ComponentType_TIDB_WORKER || tt.componentType == apiv1.ComponentType_TIKV_WORKER {
				sharedPool.CapacityMode = apiv1.CapacityMode_TOPOLOGY_DRIVEN
			}

			err := validateNodeCountForComponent(context.Background(), scaleComp, sharedPool)

			if tt.expectedError {
				require.Error(t, err)
				assert.Equal(t, tt.expectedCode, status.Code(err))
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

// TestValidateNodeSpecAgainstRUConfig_ComponentNotInRuConfig ensures that
// components intentionally absent from the RU config (e.g. SCHEMA_MANAGER) skip
// RU node-spec validation. Previously scaling such components (notably
// SCHEMA_MANAGER to zero with a node spec in the request) failed with
// "component SCHEMA_MANAGER not found".
func TestValidateNodeSpecAgainstRUConfig_ComponentNotInRuConfig(t *testing.T) {
	// ru config only contains tidb; schema_manager is intentionally absent.
	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"tidb": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
					"8c16g": {Min: 0, Max: 20000},
				}},
			},
		},
	}

	tests := []struct {
		name          string
		componentType apiv1.ComponentType
		minRU         uint32
	}{
		{
			name:          "schema manager not in ru config - skip validation",
			componentType: apiv1.ComponentType_SCHEMA_MANAGER,
			minRU:         10000,
		},
		{
			name:          "schema manager with non-zero minRU still skips",
			componentType: apiv1.ComponentType_SCHEMA_MANAGER,
			minRU:         50000,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			err := validateNodeSpecAgainstRUConfig(
				context.Background(),
				tt.componentType,
				&entity.NodeSpec{Cpu: 2, MemoryGi: 4},
				ruConfig,
				tt.minRU,
				"test-component",
				false,
			)
			assert.NoError(t, err, "components not in ru config should skip RU validation")
		})
	}
}

// TestValidateStorageNotDecreasing tests storage size validation
func TestValidateStorageNotDecreasing(t *testing.T) {
	tests := []struct {
		name           string
		currStorages   []*entity.Storage
		targetStorages []*entity.Storage
		expectedError  bool
	}{
		{
			name: "valid increase",
			currStorages: []*entity.Storage{
				{Name: apiv1.Component_Storage_DATA, SizeGi: 100},
			},
			targetStorages: []*entity.Storage{
				{Name: apiv1.Component_Storage_DATA, SizeGi: 200},
			},
			expectedError: false,
		},
		{
			name: "same size",
			currStorages: []*entity.Storage{
				{Name: apiv1.Component_Storage_DATA, SizeGi: 100},
			},
			targetStorages: []*entity.Storage{
				{Name: apiv1.Component_Storage_DATA, SizeGi: 100},
			},
			expectedError: false,
		},
		{
			name: "decrease not allowed",
			currStorages: []*entity.Storage{
				{Name: apiv1.Component_Storage_DATA, SizeGi: 200},
			},
			targetStorages: []*entity.Storage{
				{Name: apiv1.Component_Storage_DATA, SizeGi: 100},
			},
			expectedError: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			currComp := &entity.Component{Storages: tt.currStorages}
			targetComp := &entity.Component{Storages: tt.targetStorages}

			err := validateStorageNotDecreasing(currComp, targetComp)

			if tt.expectedError {
				require.Error(t, err)
				assert.Equal(t, codes.InvalidArgument, status.Code(err))
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func setupTKRWScaleSuccessMocks(
	sharedPool *entity.SharedPool,
	dataplaneClient *dataplanemgrmocks.Client,
	dagExecutor *executormocks.RemoteExecutor,
) {
	sharedPool.DataplaneName = "test-dataplane"
	dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
		NodePools: []*dplanemgrv1.NodePool{{
			CpuCore:               8,
			MemoryGi:              16,
			AvailableCpuMilliCore: 7800,
			AvailableMemoryMi:     15000,
			NodeType:              dplanemgrv1.NodeType_Dedicated,
		}},
	}, nil).Once()
	dagExecutor.On("Submit", mock.Anything, dag.PlanUpdateComponent, mock.Anything, mock.Anything).Return(nil).Once()
}

func TestScaleSharedPoolForceBypassesTiKVRemoteWorkerGuard(t *testing.T) {
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	lockRepo := repomnocks.NewLockRepo(t)
	dataplaneClient := dataplanemgrmocks.NewClient(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	scaleManager := bractionmocks.NewScaleManager(t)
	dataplaneManager := newTKRWDataplaneManager(t)
	ruConfigRepo := repomnocks.NewRuConfigRepo(t)
	tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
	clusterGatewayMock := clustergatewaymocks.NewClient(t)

	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	setupTKRWScaleSuccessMocks(sharedPool, dataplaneClient, dagExecutor)

	sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(sharedPool, nil)
	lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == testTKRWSharedPoolID
	})).Return(func() error { return nil }, nil)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
	setupAllDAGMocksWithConflict(dagExecutor, dag.PlanEnableCoprocessorWorker, runningDagStatus(tkrwKey(t, tkrwEnabledBoth(t))))

	server := &serverImpl{
		cloudProvider:    "aws",
		region:           "us-west-2",
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dataplaneManager: dataplaneManager,
		dataplaneClient:  dataplaneClient,
		dagExecutor:      dagExecutor,
		transactioner:    &mockTransactioner{},
		bizCfg: &config.BizConfig{
			DefaultDataplaneAccount: "test-account",
		},
		scaleManager:      scaleManager,
		ruConfigRepo:      ruConfigRepo,
		tidbClusterRepo:   tidbClusterRepo,
		clustergatewayCli: clusterGatewayMock,
	}

	scaleComponent := &entity.Component{
		Id:            testTKRWComponentID,
		ComponentType: apiv1.ComponentType_TIKV,
		NodeQuota: &entity.NodeQuota{
			NodeCount: 6,
		},
	}

	result, err := server.ScaleSharedPool(context.Background(), testTKRWSharedPoolID, scaleComponent, true)
	require.NoError(t, err)
	require.NotNil(t, result)
	assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
}

func TestScaleSharedPoolTiKVRemoteWorkerTerminalDriftGuard(t *testing.T) {
	tests := []struct {
		name        string
		force       bool
		wantSuccess bool
	}{
		{name: "non-force request is blocked", force: false, wantSuccess: false},
		{name: "force request bypasses guard", force: true, wantSuccess: true},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
			lockRepo := repomnocks.NewLockRepo(t)
			dataplaneClient := dataplanemgrmocks.NewClient(t)
			dagExecutor := executormocks.NewRemoteExecutor(t)
			scaleManager := bractionmocks.NewScaleManager(t)
			dataplaneManager := newTKRWDataplaneManager(t)
			ruConfigRepo := repomnocks.NewRuConfigRepo(t)
			tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t)
			clusterGatewayMock := clustergatewaymocks.NewClient(t)

			sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
			if tt.wantSuccess {
				setupTKRWScaleSuccessMocks(sharedPool, dataplaneClient, dagExecutor)
			}

			sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(sharedPool, nil)
			lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
				return lock.Key == testTKRWSharedPoolID
			})).Return(func() error { return nil }, nil)
			scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil)
			setupAllDAGMocks(dagExecutor)

			server := &serverImpl{
				cloudProvider:    "aws",
				region:           "us-west-2",
				sharedPoolRepo:   sharedPoolRepo,
				lockRepo:         lockRepo,
				dataplaneManager: dataplaneManager,
				dataplaneClient:  dataplaneClient,
				dagExecutor:      dagExecutor,
				transactioner:    &mockTransactioner{},
				bizCfg: &config.BizConfig{
					DefaultDataplaneAccount: "test-account",
				},
				scaleManager:      scaleManager,
				ruConfigRepo:      ruConfigRepo,
				tidbClusterRepo:   tidbClusterRepo,
				clustergatewayCli: clusterGatewayMock,
			}

			scaleComponent := &entity.Component{
				Id:            testTKRWComponentID,
				ComponentType: apiv1.ComponentType_TIKV,
				NodeQuota: &entity.NodeQuota{
					NodeCount: 6,
				},
			}

			result, err := server.ScaleSharedPool(context.Background(), testTKRWSharedPoolID, scaleComponent, tt.force)
			if tt.wantSuccess {
				require.NoError(t, err)
				require.NotNil(t, result)
				assert.Equal(t, apiv1.SharedPool_MODIFYING, result.GetPhase())
				return
			}

			require.Error(t, err)
			assert.Equal(t, codes.FailedPrecondition, status.Code(err))
			require.ErrorContains(t, err, "tikv remote worker topology has not converged")
			assert.Nil(t, result)
			dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything)
		})
	}
}