package server

import (
	"context"
	"errors"
	"math"
	"net/netip"
	"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/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	dataplanemgrmocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr/mocks"
	repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/genproto/googleapis/rpc/errdetails"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
)

func TestNormalizeNodePoolConfigs(t *testing.T) {
	cloud := nodePoolConfig(4, 16, false, 0)
	local := nodePoolConfig(4, 16, true, 300)
	configs := []*dplanemgrv1.NodePoolConfig{
		nil,
		{NodeType: dplanemgrv1.NodeType_Shared, CpuCore: 2, MemoryGi: 8},
		{NodeType: dplanemgrv1.NodeType_Dedicated, MemoryGi: 8, AvailableCpuMilliCore: 1000, AvailableMemoryMi: 1024},
		{NodeType: dplanemgrv1.NodeType_Dedicated, CpuCore: 2, MemoryGi: 8, AvailableMemoryMi: 1024},
		{NodeType: dplanemgrv1.NodeType_Dedicated, CpuCore: 2, MemoryGi: 8, AvailableCpuMilliCore: 1000, AvailableMemoryMi: 1024, WithLocalDisk: true},
		local,
		cloud,
		proto.Clone(cloud).(*dplanemgrv1.NodePoolConfig),
		nodePoolConfig(2, 8, false, 0),
	}

	got, dropped, err := normalizeNodePoolConfigs(configs)
	require.NoError(t, err)
	require.Len(t, got, 3)
	assert.Equal(t, uint32(2), got[0].GetCpu())
	assert.Equal(t, uint32(2000), got[0].GetAvailableCpuMilli())
	assert.Equal(t, uint32(2000), got[0].GetMaxCpuMilli())
	assert.Equal(t, uint32(8192), got[0].GetMaxMemoryMi())
	assert.Equal(t, uint32(4), got[1].GetCpu())
	assert.False(t, got[1].GetWithLocalDisk(), "non-local variant sorts before local variant")
	assert.True(t, got[2].GetWithLocalDisk())
	assert.Equal(t, uint32(300), got[2].GetLocalDiskSizeGi())
	assert.Equal(t, map[string]int{
		nodeSpecDropUnsupportedNodeType: 2,
		nodeSpecDropInvalidCapacity:     1,
		nodeSpecDropInvalidAvailable:    1,
		nodeSpecDropInvalidLocalDisk:    1,
		nodeSpecDropDuplicate:           1,
	}, dropped)
}

func TestListNodeSpecs(t *testing.T) {
	ctx := context.Background()
	client := dataplanemgrmocks.NewClient(t)
	client.On("ListNodePoolConfigs", ctx, mock.Anything).Return(&dplanemgrv1.ListNodePoolConfigsResponse{
		NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{
			nodePoolConfig(4, 16, false, 0),
			nodePoolConfig(8, 32, false, 0),
			nodePoolConfig(8, 32, true, 300),
			nodePoolConfig(8, 32, true, 600),
			nodePoolConfig(16, 64, false, 0),
			nodePoolConfig(32, 128, false, 0),
		},
	}, nil).Twice()
	ruConfig := resourcePoolTestRuConfig()
	ruConfigRepo := repomnocks.NewRuConfigRepo(t)
	ruConfigRepo.On("GetRuConfig", ctx, "tenant-1", apiv1.RuConfig_TENANT).Return(ruConfig, nil).Twice()
	s := &serverImpl{
		dataplaneClient:   client,
		ruConfigRepo:      ruConfigRepo,
		storageSizePolicy: config.DefaultStorageSizePolicy(),
	}

	tikv, err := s.ListNodeSpecs(ctx, "tenant-1", apiv1.ComponentType_TIKV)
	require.NoError(t, err)
	require.Len(t, tikv.GetNodeSpecOptions(), 2)
	assert.EqualValues(t, 8, tikv.GetNodeSpecOptions()[0].GetNodeSpec().GetCpu())
	assert.Zero(t, tikv.GetNodeSpecOptions()[0].GetNodeSpec().GetAvailableCpu())
	assert.False(t, tikv.GetNodeSpecOptions()[0].GetNodeSpec().GetWithLocalDisk())
	assert.Equal(t, []*apiv1.StorageSizeOptions{
		{
			Name:          apiv1.Component_Storage_DATA,
			Medium:        apiv1.StorageMedium_STORAGE_MEDIUM_CLOUD,
			DefaultSizeGi: 500,
			MinSizeGi:     100,
			MaxSizeGi:     4000,
		},
		{
			Name:          apiv1.Component_Storage_DATA,
			Medium:        apiv1.StorageMedium_STORAGE_MEDIUM_LOCAL,
			DefaultSizeGi: 300,
			MinSizeGi:     300,
			MaxSizeGi:     300,
		},
		{
			Name:          apiv1.Component_Storage_DATA,
			Medium:        apiv1.StorageMedium_STORAGE_MEDIUM_LOCAL,
			DefaultSizeGi: 600,
			MinSizeGi:     600,
			MaxSizeGi:     600,
		},
		{
			Name:          apiv1.Component_Storage_RAFTLOG,
			Medium:        apiv1.StorageMedium_STORAGE_MEDIUM_CLOUD,
			DefaultSizeGi: 50,
			MinSizeGi:     20,
			MaxSizeGi:     500,
		},
	}, tikv.GetNodeSpecOptions()[0].GetStorageOptions())
	assert.EqualValues(t, 16, tikv.GetNodeSpecOptions()[1].GetNodeSpec().GetCpu())
	assert.EqualValues(t, 3, tikv.GetNodeCountOptions().GetDefaultNodeCount())
	assert.EqualValues(t, 3, tikv.GetNodeCountOptions().GetMinNodeCount())
	assert.EqualValues(t, 12, tikv.GetNodeCountOptions().GetMaxNodeCount())
	assert.Equal(t, apiv1.NodeCountRuleType_NODE_COUNT_RULE_TYPE_MULTIPLE_OF, tikv.GetNodeCountOptions().GetRule().GetType())
	assert.EqualValues(t, 3, tikv.GetNodeCountOptions().GetRule().GetValue())

	compute, err := s.ListNodeSpecs(ctx, "tenant-1", apiv1.ComponentType_TIFLASH_COMPUTE)
	require.NoError(t, err)
	require.Len(t, compute.GetNodeSpecOptions(), 1)
	assert.EqualValues(t, 16, compute.GetNodeSpecOptions()[0].GetNodeSpec().GetCpu())
	assert.Equal(t, []*apiv1.StorageSizeOptions{{
		Name:          apiv1.Component_Storage_DATA,
		Medium:        apiv1.StorageMedium_STORAGE_MEDIUM_CLOUD,
		DefaultSizeGi: 500,
		MinSizeGi:     100,
		MaxSizeGi:     4000,
	}}, compute.GetNodeSpecOptions()[0].GetStorageOptions())
	assert.EqualValues(t, 1, compute.GetNodeCountOptions().GetDefaultNodeCount())
	assert.EqualValues(t, 1, compute.GetNodeCountOptions().GetMinNodeCount())
	assert.EqualValues(t, 12, compute.GetNodeCountOptions().GetMaxNodeCount())
	assert.Equal(t, apiv1.NodeCountRuleType_NODE_COUNT_RULE_TYPE_NONE, compute.GetNodeCountOptions().GetRule().GetType())

	_, err = s.ListNodeSpecs(ctx, "tenant-1", apiv1.ComponentType_TIDB)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))
	assert.Equal(t, reasonInvalidBootstrapCapacity, errorInfoReason(err))
}

func TestBuildNodeSpecOptions(t *testing.T) {
	policy := config.DefaultStorageSizePolicy()
	nodeSpecs := []*entity.NodeSpec{
		{Cpu: 8, MemoryGi: 32, AvailableCpu: 7, AvailableMemoryMi: 27_000},
		{Cpu: 8, MemoryGi: 32, AvailableCpu: 7, AvailableMemoryMi: 27_000, WithLocalDisk: true, LocalDiskSizeGi: 1900},
		{Cpu: 8, MemoryGi: 32, AvailableCpu: 6, AvailableMemoryMi: 26_000, WithLocalDisk: true, LocalDiskSizeGi: 1900},
		{Cpu: 8, MemoryGi: 32, AvailableCpu: 7, AvailableMemoryMi: 27_000, WithLocalDisk: true, LocalDiskSizeGi: 3800},
		{Cpu: 16, MemoryGi: 64, AvailableCpu: 15, AvailableMemoryMi: 58_000, WithLocalDisk: true, LocalDiskSizeGi: 900},
	}

	tests := []struct {
		name          string
		componentType entity.ComponentType
		want          []*entity.NodeSpecOptions
	}{
		{
			name:          "TiKV unions cloud and multiple local variants",
			componentType: apiv1.ComponentType_TIKV,
			want: []*entity.NodeSpecOptions{
				{
					NodeSpec: &entity.NodeSpec{Cpu: 8, MemoryGi: 32},
					StorageOptions: []*entity.StorageSizeOptions{
						{Name: apiv1.Component_Storage_DATA, Medium: apiv1.StorageMedium_STORAGE_MEDIUM_CLOUD, DefaultSizeGi: 500, MinSizeGi: 100, MaxSizeGi: 4000},
						{Name: apiv1.Component_Storage_DATA, Medium: apiv1.StorageMedium_STORAGE_MEDIUM_LOCAL, DefaultSizeGi: 1900, MinSizeGi: 1900, MaxSizeGi: 1900},
						{Name: apiv1.Component_Storage_DATA, Medium: apiv1.StorageMedium_STORAGE_MEDIUM_LOCAL, DefaultSizeGi: 3800, MinSizeGi: 3800, MaxSizeGi: 3800},
						{Name: apiv1.Component_Storage_RAFTLOG, Medium: apiv1.StorageMedium_STORAGE_MEDIUM_CLOUD, DefaultSizeGi: 50, MinSizeGi: 20, MaxSizeGi: 500},
					},
				},
				{
					NodeSpec: &entity.NodeSpec{Cpu: 16, MemoryGi: 64},
					StorageOptions: []*entity.StorageSizeOptions{
						{Name: apiv1.Component_Storage_DATA, Medium: apiv1.StorageMedium_STORAGE_MEDIUM_LOCAL, DefaultSizeGi: 900, MinSizeGi: 900, MaxSizeGi: 900},
						{Name: apiv1.Component_Storage_RAFTLOG, Medium: apiv1.StorageMedium_STORAGE_MEDIUM_CLOUD, DefaultSizeGi: 50, MinSizeGi: 20, MaxSizeGi: 500},
					},
				},
			},
		},
		{
			name:          "TiFlash Compute uses only data storage",
			componentType: apiv1.ComponentType_TIFLASH_COMPUTE,
			want: []*entity.NodeSpecOptions{
				{
					NodeSpec: &entity.NodeSpec{Cpu: 8, MemoryGi: 32},
					StorageOptions: []*entity.StorageSizeOptions{
						{Name: apiv1.Component_Storage_DATA, Medium: apiv1.StorageMedium_STORAGE_MEDIUM_CLOUD, DefaultSizeGi: 500, MinSizeGi: 100, MaxSizeGi: 4000},
						{Name: apiv1.Component_Storage_DATA, Medium: apiv1.StorageMedium_STORAGE_MEDIUM_LOCAL, DefaultSizeGi: 1900, MinSizeGi: 1900, MaxSizeGi: 1900},
						{Name: apiv1.Component_Storage_DATA, Medium: apiv1.StorageMedium_STORAGE_MEDIUM_LOCAL, DefaultSizeGi: 3800, MinSizeGi: 3800, MaxSizeGi: 3800},
					},
				},
				{
					NodeSpec: &entity.NodeSpec{Cpu: 16, MemoryGi: 64},
					StorageOptions: []*entity.StorageSizeOptions{
						{Name: apiv1.Component_Storage_DATA, Medium: apiv1.StorageMedium_STORAGE_MEDIUM_LOCAL, DefaultSizeGi: 900, MinSizeGi: 900, MaxSizeGi: 900},
					},
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := buildNodeSpecOptions(nodeSpecs, tt.componentType, policy)
			require.Equal(t, tt.want, got)

			reversed := append([]*entity.NodeSpec(nil), nodeSpecs...)
			for left, right := 0, len(reversed)-1; left < right; left, right = left+1, right-1 {
				reversed[left], reversed[right] = reversed[right], reversed[left]
			}
			require.Equal(t, tt.want, buildNodeSpecOptions(reversed, tt.componentType, policy))
		})
	}
}

func TestStorageSizePolicyIsListOnly(t *testing.T) {
	storage := func(sizeGi uint32) *apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage {
		return &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type:           entity.GP3,
			SizeGi:         sizeGi,
			Iops:           3000,
			ThroughputMbps: 125,
		}
	}
	initConfig := &entity.InitSharedPoolConfig{
		NetworkCidr:        "10.0.0.0/16",
		TikvDataStorage:    storage(entity.MinTiKVDataStorageSize),
		TikvRaftlogStorage: storage(config.DefaultStorageSizePolicy().TiKV.Raftlog.MaxSizeGi + 1),
		TiflashCnStorage:   storage(entity.MinTiFlashCNStorageSize),
		TiflashWnStorage:   storage(entity.MinTiFlashWNStorageSize),
	}

	require.NoError(t, checkInitSharedPoolConfig(
		apiv1.ProviderType_AWS.String(),
		apiv1.CapacityMode_RCU_DRIVEN,
		nil,
		initConfig,
	), "create validation must keep accepting sizes outside the ListNodeSpecs soft range")
}

func TestListNodeSpecsFailures(t *testing.T) {
	ctx := context.Background()
	t.Run("DPM failure", func(t *testing.T) {
		client := dataplanemgrmocks.NewClient(t)
		client.On("ListNodePoolConfigs", ctx, mock.Anything).Return(nil, errors.New("DPM unavailable")).Once()
		ruConfigRepo := repomnocks.NewRuConfigRepo(t)
		ruConfigRepo.On("GetRuConfig", ctx, "tenant-1", apiv1.RuConfig_TENANT).Return(resourcePoolTestRuConfig(), nil).Once()
		_, err := (&serverImpl{dataplaneClient: client, ruConfigRepo: ruConfigRepo}).ListNodeSpecs(ctx, "tenant-1", apiv1.ComponentType_TIKV)
		assert.Equal(t, codes.Unavailable, status.Code(err))
		assert.Equal(t, reasonNodeSpecsUnavailable, errorInfoReason(err))
	})
	t.Run("empty DPM and RuConfig intersection", func(t *testing.T) {
		client := dataplanemgrmocks.NewClient(t)
		client.On("ListNodePoolConfigs", ctx, mock.Anything).Return(&dplanemgrv1.ListNodePoolConfigsResponse{
			NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{nodePoolConfig(4, 16, false, 0)},
		}, nil).Once()
		ruConfigRepo := repomnocks.NewRuConfigRepo(t)
		ruConfigRepo.On("GetRuConfig", ctx, "tenant-1", apiv1.RuConfig_TENANT).Return(resourcePoolTestRuConfig(), nil).Once()
		_, err := (&serverImpl{dataplaneClient: client, ruConfigRepo: ruConfigRepo}).ListNodeSpecs(ctx, "tenant-1", apiv1.ComponentType_TIKV)
		assert.Equal(t, codes.FailedPrecondition, status.Code(err))
		assert.Equal(t, reasonNodeSpecsUnavailable, errorInfoReason(err))
	})
}

func TestProvisionConfigurationAndO11yEnforcementUseSamePrefix(t *testing.T) {
	ctx := context.Background()
	prefix := netip.MustParsePrefix("10.20.0.0/16")
	client := dataplanemgrmocks.NewClient(t)
	client.On("ListNodePoolConfigs", ctx, mock.Anything).Return(&dplanemgrv1.ListNodePoolConfigsResponse{
		AvailableNodeZones: []string{"us-west-2a", "us-west-2b", "us-west-2c"},
	}, nil).Once()
	s := &serverImpl{
		dataplaneClient:           client,
		o11yClusterPrefix:         prefix,
		o11yClusterCIDRConfigured: true,
	}

	configuration, err := s.GetSharedPoolProvisionConfiguration(ctx)
	require.NoError(t, err)
	assert.Equal(t, prefix.String(), configuration.GetO11YClusterCidr())
	assert.Equal(t, []string{"us-west-2a", "us-west-2b", "us-west-2c"}, configuration.GetAvailableNodeZones())
	assert.Equal(t, codes.OK, status.Code(s.validateO11yNetworkNonOverlap(ctx, "10.21.0.0/16", true)))

	err = s.validateO11yNetworkNonOverlap(ctx, "10.20.1.0/24", true)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Equal(t, reasonO11yCIDRConflict, errorInfoReason(err))
	assert.Equal(t, codes.OK, status.Code(s.validateO11yNetworkNonOverlap(ctx, "10.19.0.0/16", true)), "adjacent ranges do not overlap")
}

func TestGetSharedPoolProvisionConfigurationHandlesDPMFailures(t *testing.T) {
	t.Parallel()
	prefix := netip.MustParsePrefix("10.20.0.0/16")

	tests := []struct {
		name     string
		response *dplanemgrv1.ListNodePoolConfigsResponse
		err      error
		code     codes.Code
	}{
		{name: "DPM unavailable", err: errors.New("DPM unavailable"), code: codes.Unavailable},
		{name: "empty DPM response", code: codes.Internal},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()
			ctx := context.Background()
			client := dataplanemgrmocks.NewClient(t)
			client.On("ListNodePoolConfigs", ctx, mock.Anything).Return(tt.response, tt.err).Once()
			s := &serverImpl{
				dataplaneClient:           client,
				o11yClusterPrefix:         prefix,
				o11yClusterCIDRConfigured: true,
			}

			_, err := s.GetSharedPoolProvisionConfiguration(ctx)
			assert.Equal(t, tt.code, status.Code(err))
			assert.Equal(t, reasonProvisionConfigurationUnavailable, errorInfoReason(err))
		})
	}
}

func TestProvisionConfigurationAndO11yEnforcementFailWhenUnconfigured(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}

	_, err := s.GetSharedPoolProvisionConfiguration(ctx)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Equal(t, reasonO11yCIDRConfigurationMissing, errorInfoReason(err))
	err = s.validateO11yNetworkNonOverlap(ctx, "10.20.0.0/16", true)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Equal(t, reasonO11yCIDRConfigurationMissing, errorInfoReason(err))
	assert.NoError(t, s.validateO11yNetworkNonOverlap(ctx, "not-a-cidr", false))
}

func TestValidateAndResolveBootstrapCapacity(t *testing.T) {
	cloudSpec := normalizedNodeSpec(nodePoolConfig(8, 32, false, 0))
	localSpec := normalizedNodeSpec(nodePoolConfig(8, 32, true, 300))
	computeSpec := normalizedNodeSpec(nodePoolConfig(16, 64, false, 0))
	nodeSpecs := []*entity.NodeSpec{cloudSpec, localSpec, computeSpec}
	cloudStorage := &entity.InitSharedPoolConfig{
		TikvDataStorage:  &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.CloudSSD},
		TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.CloudSSD},
	}
	ruConfig := resourcePoolTestRuConfig()

	tests := []struct {
		name      string
		bootstrap *SharedPoolBootstrapCapacity
		storage   *entity.InitSharedPoolConfig
	}{
		{name: "missing TiKV", bootstrap: &SharedPoolBootstrapCapacity{}, storage: cloudStorage},
		{name: "invalid TiKV count", bootstrap: &SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: cloudSpec, NodeCount: 4}}, storage: cloudStorage},
		{name: "TiKV count exceeds quota", bootstrap: &SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: cloudSpec, NodeCount: 15}}, storage: cloudStorage},
		{name: "unknown TiKV spec", bootstrap: &SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 1}, NodeCount: 3}}, storage: cloudStorage},
		{name: "local disk mismatch", bootstrap: &SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: localSpec, NodeCount: 3}}, storage: cloudStorage},
		{name: "missing compute spec", bootstrap: &SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: cloudSpec, NodeCount: 3}, TiFlashCompute: &BootstrapComponentCapacity{NodeCount: 1}}, storage: cloudStorage},
		{name: "invalid compute count", bootstrap: &SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: cloudSpec, NodeCount: 3}, TiFlashCompute: &BootstrapComponentCapacity{NodeSpec: cloudSpec, NodeCount: 2}}, storage: cloudStorage},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			_, err := validateAndResolveBootstrapCapacity(tt.bootstrap, tt.storage, nodeSpecs, ruConfig)
			assert.Equal(t, codes.InvalidArgument, status.Code(err))
			assert.Equal(t, reasonInvalidBootstrapCapacity, errorInfoReason(err))
		})
	}

	resolved, err := validateAndResolveBootstrapCapacity(&SharedPoolBootstrapCapacity{
		TiKV:           &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 8, MemoryGi: 32}, NodeCount: 6},
		TiFlashCompute: &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 16, MemoryGi: 64}, NodeCount: 1},
	}, cloudStorage, nodeSpecs, ruConfig)
	require.NoError(t, err)
	assert.EqualValues(t, 6, resolved.TiKV.NodeCount)
	assert.EqualValues(t, 1, resolved.TiFlashCompute.NodeCount)
	assert.True(t, proto.Equal(cloudSpec, resolved.TiKV.NodeSpec))
	assert.True(t, proto.Equal(computeSpec, resolved.TiFlashCompute.NodeSpec))
	assert.NotSame(t, cloudSpec, resolved.TiKV.NodeSpec)
}

func TestResolveBootstrapNodeSpec(t *testing.T) {
	cloud := &entity.NodeSpec{Cpu: 8, MemoryGi: 32, AvailableCpu: 7, AvailableMemoryMi: 27_000}
	local1900 := &entity.NodeSpec{Cpu: 8, MemoryGi: 32, AvailableCpu: 7, AvailableMemoryMi: 27_000, WithLocalDisk: true, LocalDiskSizeGi: 1900}
	local3800 := &entity.NodeSpec{Cpu: 8, MemoryGi: 32, AvailableCpu: 7, AvailableMemoryMi: 27_000, WithLocalDisk: true, LocalDiskSizeGi: 3800}

	tests := []struct {
		name          string
		candidate     *entity.NodeSpec
		nodeSpecs     []*entity.NodeSpec
		withLocalDisk bool
		want          *entity.NodeSpec
		wantOK        bool
	}{
		{
			name:      "logical cloud selection resolves physical shape",
			candidate: &entity.NodeSpec{Cpu: 8, MemoryGi: 32},
			nodeSpecs: []*entity.NodeSpec{cloud},
			want:      cloud,
			wantOK:    true,
		},
		{
			name:          "logical local selection resolves one local size",
			candidate:     &entity.NodeSpec{Cpu: 8, MemoryGi: 32},
			nodeSpecs:     []*entity.NodeSpec{local1900},
			withLocalDisk: true,
			want:          local1900,
			wantOK:        true,
		},
		{
			name:          "logical local selection rejects ambiguous sizes",
			candidate:     &entity.NodeSpec{Cpu: 8, MemoryGi: 32},
			nodeSpecs:     []*entity.NodeSpec{local1900, local3800},
			withLocalDisk: true,
		},
		{
			name:          "legacy physical local selection stays exact",
			candidate:     proto.Clone(local3800).(*entity.NodeSpec),
			nodeSpecs:     []*entity.NodeSpec{local1900, local3800},
			withLocalDisk: true,
			want:          local3800,
			wantOK:        true,
		},
		{
			name:      "storage medium mismatch is rejected",
			candidate: proto.Clone(local1900).(*entity.NodeSpec),
			nodeSpecs: []*entity.NodeSpec{cloud, local1900},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, ok := resolveBootstrapNodeSpec(tt.candidate, tt.nodeSpecs, tt.withLocalDisk)
			assert.Equal(t, tt.wantOK, ok)
			if !tt.wantOK {
				assert.Nil(t, got)
				return
			}
			require.True(t, proto.Equal(tt.want, got))
			assert.NotSame(t, tt.want, got)
		})
	}
}

func TestDeriveBootstrapRURangeUsesShapeFloorAndCapacityPromotion(t *testing.T) {
	ruConfig := &entity.RuConfig{RuConfig: apiv1.RuConfig{Components: map[string]*apiv1.RuConfig_ComponentsMap{
		"tikv": {
			NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
				"8c32g":  {Min: 0, Max: 199_999},
				"16c64g": {Min: 200_000, Max: 0},
			},
		},
	}}}

	tests := []struct {
		name      string
		spec      *entity.NodeSpec
		count     uint32
		wantMin   uint32
		wantMax   uint32
		unbounded bool
	}{
		{name: "small shape stays in first tier", spec: &entity.NodeSpec{Cpu: 8, MemoryGi: 32}, count: 3, wantMin: 60_000, wantMax: 199_999},
		{name: "small shape near first tier maximum", spec: &entity.NodeSpec{Cpu: 8, MemoryGi: 32}, count: 9, wantMin: 180_000, wantMax: 199_999},
		{name: "small shape count promotes tier", spec: &entity.NodeSpec{Cpu: 8, MemoryGi: 32}, count: 12, wantMin: 240_000, wantMax: math.MaxUint32, unbounded: true},
		{name: "large shape cannot demote", spec: &entity.NodeSpec{Cpu: 16, MemoryGi: 64}, count: 3, wantMin: 200_000, wantMax: math.MaxUint32, unbounded: true},
		{name: "compute does not affect inference", spec: &entity.NodeSpec{Cpu: 16, MemoryGi: 64}, count: 6, wantMin: 240_000, wantMax: math.MaxUint32, unbounded: true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			rcuRange, err := deriveBootstrapRURange(&SharedPoolBootstrapCapacity{
				TiKV:           &BootstrapComponentCapacity{NodeSpec: tt.spec, NodeCount: tt.count},
				TiFlashCompute: &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 64, MemoryGi: 128}, NodeCount: 1},
			}, ruConfig)
			require.NoError(t, err)
			assert.Equal(t, tt.wantMin, rcuRange.Min)
			assert.Equal(t, tt.wantMax, rcuRange.Max)
			assert.Equal(t, tt.unbounded, rcuRange.Unbounded)
		})
	}
}

func TestDeriveBootstrapRURangeUnbounded(t *testing.T) {
	ruConfig := &entity.RuConfig{RuConfig: apiv1.RuConfig{Components: map[string]*apiv1.RuConfig_ComponentsMap{
		"tikv": {
			NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
				"8c16g": {Min: 10_000, Max: 0},
			},
		},
	}}}

	rcuRange, err := deriveBootstrapRURange(&SharedPoolBootstrapCapacity{
		TiKV: &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 8, MemoryGi: 16}, NodeCount: 3},
	}, ruConfig)
	require.NoError(t, err)
	assert.EqualValues(t, 60_000, rcuRange.Min)
	assert.Equal(t, uint32(math.MaxUint32), rcuRange.Max)
	assert.True(t, rcuRange.Unbounded)
}

func TestDeriveBootstrapRURangeFailures(t *testing.T) {
	tests := []struct {
		name      string
		bootstrap *SharedPoolBootstrapCapacity
		ruConfig  *entity.RuConfig
	}{
		{
			name:      "shape is not configured",
			bootstrap: &SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 8, MemoryGi: 16}, NodeCount: 3}},
			ruConfig:  &entity.RuConfig{RuConfig: apiv1.RuConfig{Components: map[string]*apiv1.RuConfig_ComponentsMap{"tikv": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"16c32g": {Min: 0, Max: 0}}}}}},
		},
		{
			name:      "capacity falls in a tier gap",
			bootstrap: &SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 8, MemoryGi: 32}, NodeCount: 12}},
			ruConfig: &entity.RuConfig{RuConfig: apiv1.RuConfig{Components: map[string]*apiv1.RuConfig_ComponentsMap{"tikv": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
				"8c32g": {Min: 0, Max: 199_999}, "16c64g": {Min: 300_000, Max: 0},
			}}}}},
		},
		{
			name:      "non-identical tiers overlap",
			bootstrap: &SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 8, MemoryGi: 32}, NodeCount: 3}},
			ruConfig: &entity.RuConfig{RuConfig: apiv1.RuConfig{Components: map[string]*apiv1.RuConfig_ComponentsMap{"tikv": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
				"8c32g": {Min: 0, Max: 200_000}, "16c64g": {Min: 200_000, Max: 0},
			}}}}},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			_, err := deriveBootstrapRURange(tt.bootstrap, tt.ruConfig)
			assert.Equal(t, codes.FailedPrecondition, status.Code(err))
			assert.Equal(t, reasonBootstrapRURangeUnavailable, errorInfoReason(err))
		})
	}
}

func TestDeriveBootstrapRURangeDeduplicatesIdenticalTiers(t *testing.T) {
	ruConfig := &entity.RuConfig{RuConfig: apiv1.RuConfig{Components: map[string]*apiv1.RuConfig_ComponentsMap{
		"tikv": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
			"8c32g": {Min: 0, Max: 199_999},
			"8c64g": {Min: 0, Max: 199_999},
		}},
	}}}

	rcuRange, err := deriveBootstrapRURange(&SharedPoolBootstrapCapacity{
		TiKV: &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 8, MemoryGi: 32}, NodeCount: 3},
	}, ruConfig)
	require.NoError(t, err)
	assert.EqualValues(t, 60_000, rcuRange.Min)
	assert.EqualValues(t, 199_999, rcuRange.Max)
}

func TestApplyBootstrapComponentCapacityPreservesRuBounds(t *testing.T) {
	component := &entity.NodeQuota{NodeSpec: &entity.NodeSpec{Cpu: 4, MemoryGi: 16}, NodeCount: 3, MinNodeCount: 2, MaxNodeCount: 9}
	override := &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 8, MemoryGi: 32}, NodeCount: 6}

	applyBootstrapComponentCapacity(component, override)

	assert.EqualValues(t, 8, component.GetNodeSpec().GetCpu())
	assert.EqualValues(t, 6, component.GetNodeCount())
	assert.EqualValues(t, 2, component.GetMinNodeCount())
	assert.EqualValues(t, 9, component.GetMaxNodeCount())
	assert.NotSame(t, override.NodeSpec, component.GetNodeSpec())
}

func nodePoolConfig(cpu, memory uint32, local bool, disk uint32) *dplanemgrv1.NodePoolConfig {
	return &dplanemgrv1.NodePoolConfig{
		NodeType:              dplanemgrv1.NodeType_Dedicated,
		CpuCore:               cpu,
		MemoryGi:              memory,
		AvailableCpuMilliCore: cpu * 1000,
		AvailableMemoryMi:     memory * 1024,
		WithLocalDisk:         local,
		LocalDiskSizeGi:       disk,
	}
}

func normalizedNodeSpec(nodePoolConfig *dplanemgrv1.NodePoolConfig) *entity.NodeSpec {
	specs, _, err := normalizeNodePoolConfigs([]*dplanemgrv1.NodePoolConfig{nodePoolConfig})
	if err != nil || len(specs) != 1 {
		panic("invalid test node-pool config")
	}
	return specs[0]
}

func errorInfoReason(err error) string {
	st, ok := status.FromError(err)
	if !ok {
		return ""
	}
	for _, detail := range st.Details() {
		if info, ok := detail.(*errdetails.ErrorInfo); ok {
			return info.GetReason()
		}
	}
	return ""
}

func resourcePoolTestRuConfig() *entity.RuConfig {
	return &entity.RuConfig{RuConfig: apiv1.RuConfig{Components: map[string]*apiv1.RuConfig_ComponentsMap{
		"tikv": {
			MaxNodeCount: 13,
			NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
				"8c32g":  {Min: 0, Max: 199_999},
				"16c64g": {Min: 200_000, Max: 0},
			},
		},
		"tiflash_compute": {
			NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
				"16c64g": {Min: 0, Max: 0},
			},
			MaxNodeCount: 12,
		},
	}}}
}