package server

import (
	"context"
	"testing"

	"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"
	"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/pkg/dataplanemgr/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func TestGetSharedPoolScaleConstraints(t *testing.T) {
	t.Run("returns existing scale-relevant components and sorted candidates", func(t *testing.T) {
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
		ruConfigRepo := repomocks.NewRuConfigRepo(t)
		tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
		dataplaneClient := mocks.NewClient(t)
		dagExecutor := executormocks.NewRemoteExecutor(t)
		scaleManager := bractionmocks.NewScaleManager(t)

		sharedPool := &entity.SharedPool{
			SharedPool: apiv1.SharedPool{
				Id:            "10001",
				DataplaneName: "dp-1",
				Phase:         apiv1.SharedPool_AVAILABLE,
				ComponentGroups: []*apiv1.Component{
					{
						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},
							{Name: apiv1.Component_Storage_RAFTLOG, Type: "gp3", SizeGi: 50},
						},
					},
					{
						Id:            "default",
						ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
						Phase:         apiv1.Component_AVAILABLE,
						NodeQuota: &apiv1.NodeQuota{
							NodeCount: 1,
							NodeSpec:  &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8},
						},
					},
				},
			},
		}
		sharedPoolRepo.On("Load", mock.Anything, "10001", false).Return(sharedPool, nil).Once()
		dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
		scaleManager.On("AnyFullfillingDecision", mock.Anything, "10001").Return(false, nil)
		dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{CpuCore: 4, MemoryGi: 8, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600},
				{CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7600, AvailableMemoryMi: 15200},
				{CpuCore: 16, MemoryGi: 32, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 15000, AvailableMemoryMi: 30000},
				{CpuCore: 16, MemoryGi: 32, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 15000, AvailableMemoryMi: 30000, WithLocalDisk: true, LocalDiskSizeGi: 1500},
			},
		}, nil).Once()
		ruConfigRepo.On("GetRuConfig", mock.Anything, "10001", apiv1.RuConfig_SHARED_POOL).Return(testDefaultRUConfig(), nil).Once()
		tidbClusterRepo.On("List", mock.Anything, mock.MatchedBy(func(filter *repo.TiDBClusterListFilter) bool {
			return len(filter.SharedPoolIds) == 1 && filter.SharedPoolIds[0] == "10001" && !filter.Deleted
		})).Return([]*entity.TiDBCluster{}, nil).Once()

		server := &serverImpl{
			sharedPoolRepo:  sharedPoolRepo,
			ruConfigRepo:    ruConfigRepo,
			tidbClusterRepo: tidbClusterRepo,
			dataplaneClient: dataplaneClient,
			dagExecutor:     dagExecutor,
			scaleManager:    scaleManager,
		}

		resp, err := server.GetSharedPoolScaleConstraints(context.Background(), "10001")
		require.NoError(t, err)
		require.NotNil(t, resp)
		require.Len(t, resp.GetComponents(), 1)

		component := resp.GetComponents()[0]
		assert.Equal(t, "default", component.GetComponentId())
		assert.Equal(t, apiv1.ComponentType_TIKV, component.GetComponentType())
		assert.Equal(t, uint32(3), component.GetNodeCountConstraints().GetCurrent())
		assert.Equal(t, uint32(3), component.GetNodeCountConstraints().GetMin())
		assert.Equal(t, uint32(15), component.GetNodeCountConstraints().GetMax())
		assert.Equal(t, scaleConstraintRuleTypeMultipleOf, component.GetNodeCountConstraints().GetRule().GetType())
		assert.Equal(t, uint32(3), component.GetNodeCountConstraints().GetRule().GetValue())

		nodeSpecConstraints := component.GetNodeSpecConstraints()
		assert.Equal(t, "8C16G", nodeSpecConstraints.GetCurrent())
		require.Len(t, nodeSpecConstraints.GetCandidates(), 3)
		assert.Equal(t, "8C16G", nodeSpecConstraints.GetCandidates()[0].GetName())
		assert.True(t, nodeSpecConstraints.GetCandidates()[0].GetIsCurrent())
		assert.Equal(t, "4C8G", nodeSpecConstraints.GetCandidates()[1].GetName())
		assert.Equal(t, "16C32G", nodeSpecConstraints.GetCandidates()[2].GetName())
		assert.True(t, nodeSpecConstraints.GetCandidates()[2].GetAvailable())
		require.Len(t, nodeSpecConstraints.GetCandidates()[2].GetStorageConstraints(), 2)
		assert.Equal(t, uint32(500), nodeSpecConstraints.GetCandidates()[2].GetStorageConstraints()[0].GetCurrentSizeGi())
	})

	t.Run("marks unsupported storage entries as non-resizable", func(t *testing.T) {
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
		ruConfigRepo := repomocks.NewRuConfigRepo(t)
		tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
		dataplaneClient := mocks.NewClient(t)
		dagExecutor := executormocks.NewRemoteExecutor(t)
		scaleManager := bractionmocks.NewScaleManager(t)

		sharedPool := &entity.SharedPool{
			SharedPool: apiv1.SharedPool{
				Id:            "10005",
				DataplaneName: "dp-5",
				Phase:         apiv1.SharedPool_AVAILABLE,
				ComponentGroups: []*apiv1.Component{
					{
						Id:            "default",
						ComponentType: apiv1.ComponentType_TIKV,
						Phase:         apiv1.Component_AVAILABLE,
						NodeQuota: &apiv1.NodeQuota{
							NodeCount:    3,
							MinNodeCount: 3,
							MaxNodeCount: 15,
							NodeSpec: &apiv1.NodeSpec{
								Cpu:      4,
								MemoryGi: 8,
							},
						},
						Storages: []*apiv1.Component_Storage{
							{Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 500},
							{Name: apiv1.Component_Storage_SLOWLOG, Type: "gp3", SizeGi: 30},
						},
					},
				},
			},
		}
		sharedPoolRepo.On("Load", mock.Anything, "10005", false).Return(sharedPool, nil).Once()
		dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
		scaleManager.On("AnyFullfillingDecision", mock.Anything, "10005").Return(false, nil)
		dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{CpuCore: 4, MemoryGi: 8, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600},
			},
		}, nil).Once()
		ruConfigRepo.On("GetRuConfig", mock.Anything, "10005", apiv1.RuConfig_SHARED_POOL).Return(testDefaultRUConfig(), nil).Once()
		tidbClusterRepo.On("List", mock.Anything, mock.MatchedBy(func(filter *repo.TiDBClusterListFilter) bool {
			return len(filter.SharedPoolIds) == 1 && filter.SharedPoolIds[0] == "10005" && !filter.Deleted
		})).Return([]*entity.TiDBCluster{}, nil).Once()

		server := &serverImpl{
			sharedPoolRepo:  sharedPoolRepo,
			ruConfigRepo:    ruConfigRepo,
			tidbClusterRepo: tidbClusterRepo,
			dataplaneClient: dataplaneClient,
			dagExecutor:     dagExecutor,
			scaleManager:    scaleManager,
		}

		resp, err := server.GetSharedPoolScaleConstraints(context.Background(), "10005")
		require.NoError(t, err)
		require.Len(t, resp.GetComponents(), 1)

		candidates := resp.GetComponents()[0].GetNodeSpecConstraints().GetCandidates()
		require.Len(t, candidates, 1)
		require.Len(t, candidates[0].GetStorageConstraints(), 2)
		assert.Equal(t, apiv1.Component_Storage_SLOWLOG, candidates[0].GetStorageConstraints()[1].GetName())
		assert.False(t, candidates[0].GetStorageConstraints()[1].GetResizable())
		assert.Equal(t, uint32(30), candidates[0].GetStorageConstraints()[1].GetCurrentSizeGi())
		assert.Nil(t, candidates[0].GetStorageConstraints()[1].MaxSizeGi)
	})

	t.Run("keeps current node spec in candidates even when it is absent from candidate set", func(t *testing.T) {
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
		ruConfigRepo := repomocks.NewRuConfigRepo(t)
		tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
		dataplaneClient := mocks.NewClient(t)
		dagExecutor := executormocks.NewRemoteExecutor(t)
		scaleManager := bractionmocks.NewScaleManager(t)

		sharedPool := &entity.SharedPool{
			SharedPool: apiv1.SharedPool{
				Id:            "10003",
				DataplaneName: "dp-3",
				Phase:         apiv1.SharedPool_AVAILABLE,
				ComponentGroups: []*apiv1.Component{
					{
						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},
						},
					},
				},
			},
		}
		sharedPoolRepo.On("Load", mock.Anything, "10003", false).Return(sharedPool, nil).Once()
		dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
		scaleManager.On("AnyFullfillingDecision", mock.Anything, "10003").Return(false, nil)
		dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{CpuCore: 4, MemoryGi: 8, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600},
				{CpuCore: 16, MemoryGi: 32, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 15000, AvailableMemoryMi: 30000},
			},
		}, nil).Once()
		ruConfigRepo.On("GetRuConfig", mock.Anything, "10003", apiv1.RuConfig_SHARED_POOL).Return(testDefaultRUConfig(), nil).Once()
		tidbClusterRepo.On("List", mock.Anything, mock.MatchedBy(func(filter *repo.TiDBClusterListFilter) bool {
			return len(filter.SharedPoolIds) == 1 && filter.SharedPoolIds[0] == "10003" && !filter.Deleted
		})).Return([]*entity.TiDBCluster{}, nil).Once()

		server := &serverImpl{
			sharedPoolRepo:  sharedPoolRepo,
			ruConfigRepo:    ruConfigRepo,
			tidbClusterRepo: tidbClusterRepo,
			dataplaneClient: dataplaneClient,
			dagExecutor:     dagExecutor,
			scaleManager:    scaleManager,
		}

		resp, err := server.GetSharedPoolScaleConstraints(context.Background(), "10003")
		require.NoError(t, err)
		require.Len(t, resp.GetComponents(), 1)

		candidates := resp.GetComponents()[0].GetNodeSpecConstraints().GetCandidates()
		require.Len(t, candidates, 3)
		assert.Equal(t, "8C16G", candidates[0].GetName())
		assert.True(t, candidates[0].GetIsCurrent())
		assert.True(t, candidates[0].GetAvailable())
	})

	t.Run("uses stable reason code when candidate spec is not present in shared pool", func(t *testing.T) {
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
		ruConfigRepo := repomocks.NewRuConfigRepo(t)
		tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
		dataplaneClient := mocks.NewClient(t)
		dagExecutor := executormocks.NewRemoteExecutor(t)
		scaleManager := bractionmocks.NewScaleManager(t)

		sharedPool := &entity.SharedPool{
			SharedPool: apiv1.SharedPool{
				Id:            "10008",
				DataplaneName: "dp-8",
				Phase:         apiv1.SharedPool_AVAILABLE,
				ComponentGroups: []*apiv1.Component{
					{
						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},
						},
					},
				},
			},
		}
		sharedPoolRepo.On("Load", mock.Anything, "10008", false).Return(sharedPool, nil).Once()
		dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
		scaleManager.On("AnyFullfillingDecision", mock.Anything, "10008").Return(false, nil)
		dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{CpuCore: 4, MemoryGi: 8, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 0, AvailableMemoryMi: 7600},
				{CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7600, AvailableMemoryMi: 15200},
			},
		}, nil).Once()
		ruConfigRepo.On("GetRuConfig", mock.Anything, "10008", apiv1.RuConfig_SHARED_POOL).Return(testDefaultRUConfig(), nil).Once()
		tidbClusterRepo.On("List", mock.Anything, mock.MatchedBy(func(filter *repo.TiDBClusterListFilter) bool {
			return len(filter.SharedPoolIds) == 1 && filter.SharedPoolIds[0] == "10008" && !filter.Deleted
		})).Return([]*entity.TiDBCluster{}, nil).Once()

		server := &serverImpl{
			sharedPoolRepo:  sharedPoolRepo,
			ruConfigRepo:    ruConfigRepo,
			tidbClusterRepo: tidbClusterRepo,
			dataplaneClient: dataplaneClient,
			dagExecutor:     dagExecutor,
			scaleManager:    scaleManager,
		}

		resp, err := server.GetSharedPoolScaleConstraints(context.Background(), "10008")
		require.NoError(t, err)
		require.Len(t, resp.GetComponents(), 1)

		candidates := resp.GetComponents()[0].GetNodeSpecConstraints().GetCandidates()
		require.Len(t, candidates, 2)
		assert.Equal(t, "4C8G", candidates[1].GetName())
		assert.False(t, candidates[1].GetAvailable())
		require.Len(t, candidates[1].GetReasons(), 1)
		assert.Equal(t, scaleConstraintReasonSpecNotPresent, candidates[1].GetReasons()[0].GetCode())
	})

	t.Run("marks candidates unavailable for RU and local-disk limits", func(t *testing.T) {
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
		ruConfigRepo := repomocks.NewRuConfigRepo(t)
		tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
		dataplaneClient := mocks.NewClient(t)
		dagExecutor := executormocks.NewRemoteExecutor(t)
		scaleManager := bractionmocks.NewScaleManager(t)

		sharedPool := &entity.SharedPool{
			SharedPool: apiv1.SharedPool{
				Id:            "10002",
				DataplaneName: "dp-2",
				Phase:         apiv1.SharedPool_AVAILABLE,
				ComponentGroups: []*apiv1.Component{
					{
						Id:            "default",
						ComponentType: apiv1.ComponentType_TIKV,
						Phase:         apiv1.Component_UNAVAILABLE,
						NodeQuota: &apiv1.NodeQuota{
							NodeCount:    3,
							MinNodeCount: 3,
							MaxNodeCount: 15,
							NodeSpec: &apiv1.NodeSpec{
								Cpu:             8,
								MemoryGi:        16,
								WithLocalDisk:   true,
								LocalDiskSizeGi: 2000,
							},
						},
						Storages: []*apiv1.Component_Storage{
							{Name: apiv1.Component_Storage_DATA, Type: entity.LocalSSD, SizeGi: 2000},
						},
					},
				},
			},
		}
		sharedPoolRepo.On("Load", mock.Anything, "10002", false).Return(sharedPool, nil).Once()
		dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
		scaleManager.On("AnyFullfillingDecision", mock.Anything, "10002").Return(false, nil)
		dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{CpuCore: 4, MemoryGi: 8, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600, WithLocalDisk: true, LocalDiskSizeGi: 1500},
				{CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7600, AvailableMemoryMi: 15200, WithLocalDisk: true, LocalDiskSizeGi: 2000},
			},
		}, nil).Once()
		ruConfigRepo.On("GetRuConfig", mock.Anything, "10002", apiv1.RuConfig_SHARED_POOL).Return(&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: 0},
						},
					},
				},
			},
		}, nil).Once()
		tidbClusterRepo.On("List", mock.Anything, mock.MatchedBy(func(filter *repo.TiDBClusterListFilter) bool {
			return len(filter.SharedPoolIds) == 1 && filter.SharedPoolIds[0] == "10002" && !filter.Deleted
		})).Return([]*entity.TiDBCluster{
			{RequestUnit: &apiv1.Cluster_RequestUnit{Min: 10000}},
		}, nil).Once()

		server := &serverImpl{
			sharedPoolRepo:  sharedPoolRepo,
			ruConfigRepo:    ruConfigRepo,
			tidbClusterRepo: tidbClusterRepo,
			dataplaneClient: dataplaneClient,
			dagExecutor:     dagExecutor,
			scaleManager:    scaleManager,
		}

		resp, err := server.GetSharedPoolScaleConstraints(context.Background(), "10002")
		require.NoError(t, err)
		require.Len(t, resp.GetComponents(), 1)

		component := resp.GetComponents()[0]
		require.Len(t, component.GetBlockingReasons(), 1)
		assert.Equal(t, scaleConstraintReasonComponentPhaseUnavailable, component.GetBlockingReasons()[0].GetCode())

		candidates := component.GetNodeSpecConstraints().GetCandidates()
		require.Len(t, candidates, 2)
		assert.Equal(t, "8C16G", candidates[0].GetName())
		assert.True(t, candidates[0].GetAvailable())
		require.Len(t, candidates[0].GetStorageConstraints(), 1)
		assert.False(t, candidates[0].GetStorageConstraints()[0].GetResizable())
		assert.Equal(t, uint32(2000), candidates[0].GetStorageConstraints()[0].GetCurrentSizeGi())
		require.NotNil(t, candidates[0].GetStorageConstraints()[0].MaxSizeGi)

		assert.Equal(t, "4C8G", candidates[1].GetName())
		assert.False(t, candidates[1].GetAvailable())
		assert.Empty(t, candidates[1].GetStorageConstraints())
		require.Len(t, candidates[1].GetReasons(), 2)
		assert.Equal(t, scaleConstraintReasonBelowRURequirement, candidates[1].GetReasons()[0].GetCode())
		assert.Equal(t, scaleConstraintReasonInsufficientLocalDisk, candidates[1].GetReasons()[1].GetCode())
		require.NotNil(t, candidates[1].GetRequiredMinSpec())
		assert.Equal(t, "8C16G", candidates[1].GetRequiredMinSpec().GetName())
	})

	t.Run("sets allow cross zero for tiflash components", func(t *testing.T) {
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
		ruConfigRepo := repomocks.NewRuConfigRepo(t)
		tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
		dataplaneClient := mocks.NewClient(t)
		dagExecutor := executormocks.NewRemoteExecutor(t)
		scaleManager := bractionmocks.NewScaleManager(t)

		sharedPool := &entity.SharedPool{
			SharedPool: apiv1.SharedPool{
				Id:            "10004",
				DataplaneName: "dp-4",
				Phase:         apiv1.SharedPool_AVAILABLE,
				ComponentGroups: []*apiv1.Component{
					{
						Id:            "default",
						ComponentType: apiv1.ComponentType_TIFLASH_COMPUTE,
						Phase:         apiv1.Component_AVAILABLE,
						NodeQuota: &apiv1.NodeQuota{
							NodeCount:    1,
							MinNodeCount: 1,
							MaxNodeCount: 12,
							NodeSpec: &apiv1.NodeSpec{
								Cpu:      8,
								MemoryGi: 16,
							},
						},
						Storages: []*apiv1.Component_Storage{
							{Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 500},
						},
					},
				},
			},
		}
		sharedPoolRepo.On("Load", mock.Anything, "10004", false).Return(sharedPool, nil).Once()
		dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
		scaleManager.On("AnyFullfillingDecision", mock.Anything, "10004").Return(false, nil)
		dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7600, AvailableMemoryMi: 15200},
			},
		}, nil).Once()
		ruConfigRepo.On("GetRuConfig", mock.Anything, "10004", apiv1.RuConfig_SHARED_POOL).Return(testDefaultRUConfig(), nil).Once()
		tidbClusterRepo.On("List", mock.Anything, mock.MatchedBy(func(filter *repo.TiDBClusterListFilter) bool {
			return len(filter.SharedPoolIds) == 1 && filter.SharedPoolIds[0] == "10004" && !filter.Deleted
		})).Return([]*entity.TiDBCluster{}, nil).Once()

		server := &serverImpl{
			sharedPoolRepo:  sharedPoolRepo,
			ruConfigRepo:    ruConfigRepo,
			tidbClusterRepo: tidbClusterRepo,
			dataplaneClient: dataplaneClient,
			dagExecutor:     dagExecutor,
			scaleManager:    scaleManager,
		}

		resp, err := server.GetSharedPoolScaleConstraints(context.Background(), "10004")
		require.NoError(t, err)
		require.Len(t, resp.GetComponents(), 1)
		assert.Equal(t, uint32(0), resp.GetComponents()[0].GetNodeCountConstraints().GetMin())
		assert.True(t, resp.GetComponents()[0].GetNodeCountConstraints().GetAllowCrossZero())
	})

	t.Run("does not block topology-driven worker cross-zero recovery by component phase", func(t *testing.T) {
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
		ruConfigRepo := repomocks.NewRuConfigRepo(t)
		tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
		dataplaneClient := mocks.NewClient(t)
		dagExecutor := executormocks.NewRemoteExecutor(t)
		scaleManager := bractionmocks.NewScaleManager(t)

		sharedPool := &entity.SharedPool{
			SharedPool: apiv1.SharedPool{
				Id:            "10006",
				DataplaneName: "dp-6",
				Phase:         apiv1.SharedPool_AVAILABLE,
				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: 1,
							MaxNodeCount: 15,
							NodeSpec: &apiv1.NodeSpec{
								Cpu:      8,
								MemoryGi: 16,
							},
						},
						Storages: []*apiv1.Component_Storage{
							{Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 200},
						},
					},
				},
			},
		}
		sharedPoolRepo.On("Load", mock.Anything, "10006", false).Return(sharedPool, nil).Once()
		dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
		scaleManager.On("AnyFullfillingDecision", mock.Anything, "10006").Return(false, nil)
		dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7600, AvailableMemoryMi: 15200},
			},
		}, nil).Once()
		ruConfigRepo.On("GetRuConfig", mock.Anything, "10006", apiv1.RuConfig_SHARED_POOL).Return(testDefaultRUConfig(), nil).Once()
		tidbClusterRepo.On("List", mock.Anything, mock.MatchedBy(func(filter *repo.TiDBClusterListFilter) bool {
			return len(filter.SharedPoolIds) == 1 && filter.SharedPoolIds[0] == "10006" && !filter.Deleted
		})).Return([]*entity.TiDBCluster{}, nil).Once()

		server := &serverImpl{
			sharedPoolRepo:  sharedPoolRepo,
			ruConfigRepo:    ruConfigRepo,
			tidbClusterRepo: tidbClusterRepo,
			dataplaneClient: dataplaneClient,
			dagExecutor:     dagExecutor,
			scaleManager:    scaleManager,
		}

		resp, err := server.GetSharedPoolScaleConstraints(context.Background(), "10006")
		require.NoError(t, err)
		require.Len(t, resp.GetComponents(), 1)

		component := resp.GetComponents()[0]
		assert.Empty(t, component.GetBlockingReasons())
		assert.Equal(t, uint32(0), component.GetNodeCountConstraints().GetMin())
		assert.True(t, component.GetNodeCountConstraints().GetAllowCrossZero())
	})

	t.Run("keeps phase blocking for unavailable topology-driven worker with positive node count", func(t *testing.T) {
		sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
		ruConfigRepo := repomocks.NewRuConfigRepo(t)
		tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
		dataplaneClient := mocks.NewClient(t)
		dagExecutor := executormocks.NewRemoteExecutor(t)
		scaleManager := bractionmocks.NewScaleManager(t)

		sharedPool := &entity.SharedPool{
			SharedPool: apiv1.SharedPool{
				Id:            "10007",
				DataplaneName: "dp-7",
				Phase:         apiv1.SharedPool_AVAILABLE,
				CapacityMode:  apiv1.CapacityMode_TOPOLOGY_DRIVEN,
				ComponentGroups: []*apiv1.Component{
					{
						Id:            "default",
						ComponentType: apiv1.ComponentType_TIKV_WORKER,
						Phase:         apiv1.Component_UNAVAILABLE,
						NodeQuota: &apiv1.NodeQuota{
							NodeCount:    3,
							MinNodeCount: 1,
							MaxNodeCount: 15,
							NodeSpec: &apiv1.NodeSpec{
								Cpu:      8,
								MemoryGi: 16,
							},
						},
						Storages: []*apiv1.Component_Storage{
							{Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 200},
						},
					},
				},
			},
		}
		sharedPoolRepo.On("Load", mock.Anything, "10007", false).Return(sharedPool, nil).Once()
		dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
		scaleManager.On("AnyFullfillingDecision", mock.Anything, "10007").Return(false, nil)
		dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7600, AvailableMemoryMi: 15200},
			},
		}, nil).Once()
		ruConfigRepo.On("GetRuConfig", mock.Anything, "10007", apiv1.RuConfig_SHARED_POOL).Return(testDefaultRUConfig(), nil).Once()
		tidbClusterRepo.On("List", mock.Anything, mock.MatchedBy(func(filter *repo.TiDBClusterListFilter) bool {
			return len(filter.SharedPoolIds) == 1 && filter.SharedPoolIds[0] == "10007" && !filter.Deleted
		})).Return([]*entity.TiDBCluster{}, nil).Once()

		server := &serverImpl{
			sharedPoolRepo:  sharedPoolRepo,
			ruConfigRepo:    ruConfigRepo,
			tidbClusterRepo: tidbClusterRepo,
			dataplaneClient: dataplaneClient,
			dagExecutor:     dagExecutor,
			scaleManager:    scaleManager,
		}

		resp, err := server.GetSharedPoolScaleConstraints(context.Background(), "10007")
		require.NoError(t, err)
		require.Len(t, resp.GetComponents(), 1)

		component := resp.GetComponents()[0]
		require.Len(t, component.GetBlockingReasons(), 1)
		assert.Equal(t, scaleConstraintReasonComponentPhaseUnavailable, component.GetBlockingReasons()[0].GetCode())
		assert.Equal(t, uint32(0), component.GetNodeCountConstraints().GetMin())
		assert.True(t, component.GetNodeCountConstraints().GetAllowCrossZero())
	})
}

func TestGetSharedPoolScaleConstraintsTiDBGroupMetadata(t *testing.T) {
	tests := []struct {
		name                 string
		tidbGroup            *entity.TiDBGroup
		tidbGroupErr         error
		includeActiveTiKV    bool
		expectedComponentIDs []string
		expectedCode         codes.Code
	}{
		{
			name: "skips soft-deleted tidb group and keeps active components",
			tidbGroup: &entity.TiDBGroup{
				TiDBGroup: apiv1.TiDBGroup{Id: "deleted-tidb"},
				IsDeleted: true,
			},
			includeActiveTiKV:    true,
			expectedComponentIDs: []string{"default"},
			expectedCode:         codes.OK,
		},
		{
			name: "builds constraint for active tidb group",
			tidbGroup: &entity.TiDBGroup{
				TiDBGroup: apiv1.TiDBGroup{
					Id:        "active-tidb",
					ClusterId: "cluster-1",
				},
			},
			expectedComponentIDs: []string{"active-tidb"},
			expectedCode:         codes.OK,
		},
		{
			name:         "returns error when tidb group is absent including deleted records",
			tidbGroupErr: status.Error(codes.NotFound, "tidb group not found"),
			expectedCode: codes.Internal,
		},
		{
			name:         "returns error when loading tidb group metadata fails",
			tidbGroupErr: status.Error(codes.Internal, "database error"),
			expectedCode: codes.Internal,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
			ruConfigRepo := repomocks.NewRuConfigRepo(t)
			tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
			tidbGroupRepo := repomocks.NewTiDBGroupRepo(t)
			dataplaneClient := mocks.NewClient(t)
			dagExecutor := executormocks.NewRemoteExecutor(t)
			scaleManager := bractionmocks.NewScaleManager(t)

			tidbComponentID := "active-tidb"
			if tt.tidbGroup != nil {
				tidbComponentID = tt.tidbGroup.GetId()
			}
			if tt.tidbGroupErr != nil {
				tidbComponentID = "missing-tidb"
			}

			components := []*apiv1.Component{
				{
					Id:            tidbComponentID,
					ComponentType: apiv1.ComponentType_TIDB,
					Phase:         apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeCount:    1,
						MinNodeCount: 1,
						MaxNodeCount: 8,
						NodeSpec:     &apiv1.NodeSpec{Cpu: 2, MemoryGi: 4},
					},
				},
			}
			if tt.includeActiveTiKV {
				components = append(components, &apiv1.Component{
					Id:            "default",
					ComponentType: apiv1.ComponentType_TIKV,
					Phase:         apiv1.Component_AVAILABLE,
					NodeQuota: &apiv1.NodeQuota{
						NodeCount:    3,
						MinNodeCount: 3,
						MaxNodeCount: 15,
						NodeSpec:     &apiv1.NodeSpec{Cpu: 2, MemoryGi: 4},
					},
				})
			}

			sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
				Id:              "pool-1",
				DataplaneName:   "dp-1",
				Phase:           apiv1.SharedPool_AVAILABLE,
				ComponentGroups: components,
			}}
			sharedPoolRepo.On("Load", mock.Anything, "pool-1", false).Return(sharedPool, nil).Once()
			dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
			scaleManager.On("AnyFullfillingDecision", mock.Anything, "pool-1").Return(false, nil)
			dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
				NodePools: []*dplanemgrv1.NodePool{
					{CpuCore: 2, MemoryGi: 4, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 1900, AvailableMemoryMi: 3800},
				},
			}, nil).Once()
			ruConfigRepo.On("GetRuConfig", mock.Anything, "pool-1", apiv1.RuConfig_SHARED_POOL).Return(testDefaultRUConfig(), nil).Once()
			tidbGroupRepo.On("Load", mock.Anything, tidbComponentID, true).Return(tt.tidbGroup, tt.tidbGroupErr).Once()

			if tt.includeActiveTiKV {
				tidbClusterRepo.On("List", mock.Anything, mock.MatchedBy(func(filter *repo.TiDBClusterListFilter) bool {
					return len(filter.SharedPoolIds) == 1 && filter.SharedPoolIds[0] == "pool-1" && !filter.Deleted
				})).Return([]*entity.TiDBCluster{}, nil).Once()
			}
			if tt.tidbGroup != nil && !tt.tidbGroup.IsDeleted {
				tidbClusterRepo.On("Load", mock.Anything, "cluster-1", false).Return(&entity.TiDBCluster{
					RequestUnit: &apiv1.Cluster_RequestUnit{Min: 100},
				}, nil).Once()
			}

			server := &serverImpl{
				sharedPoolRepo:  sharedPoolRepo,
				ruConfigRepo:    ruConfigRepo,
				tidbClusterRepo: tidbClusterRepo,
				tidbGroupRepo:   tidbGroupRepo,
				dataplaneClient: dataplaneClient,
				dagExecutor:     dagExecutor,
				scaleManager:    scaleManager,
			}

			resp, err := server.GetSharedPoolScaleConstraints(context.Background(), "pool-1")
			assert.Equal(t, tt.expectedCode, status.Code(err))
			if tt.expectedCode != codes.OK {
				require.Error(t, err)
				assert.Contains(t, err.Error(), "including deleted records")
				return
			}

			require.NoError(t, err)
			require.NotNil(t, resp)
			componentIDs := make([]string, 0, len(resp.GetComponents()))
			for _, component := range resp.GetComponents() {
				componentIDs = append(componentIDs, component.GetComponentId())
			}
			assert.ElementsMatch(t, tt.expectedComponentIDs, componentIDs)
			if tt.tidbGroup != nil && !tt.tidbGroup.IsDeleted {
				require.Len(t, resp.GetComponents(), 1)
				assert.Equal(t, "2C4G", resp.GetComponents()[0].GetNodeSpecConstraints().GetCurrent())
			}
		})
	}
}

func TestGetSharedPoolScaleConstraintsErrors(t *testing.T) {
	server := &serverImpl{}

	_, err := server.GetSharedPoolScaleConstraints(context.Background(), "")
	require.Error(t, err)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))
}