package workflows

import (
	"context"
	"testing"

	"github.com/samber/lo"
	"github.com/stretchr/testify/assert"
	"k8s.io/apimachinery/pkg/api/resource"

	"github.com/tidbcloud/cluster-controller/server/pkg/cloudprovider"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

func TestBuildScaleScopeGroups(t *testing.T) {
	cases := map[string]struct {
		orig, changed      *cloudv1.TiKV
		newConfigRev       string
		wantScopes         [][]string
		wantStepNames      []string
		wantSchedulerSteps []int // indices of scheduler steps (Del=0, Add=1)
		wantErrContains    string
	}{
		"storage size changed - tikv with scheduler steps": {
			orig:               &cloudv1.TiKV{StorageSize: *util.StorageSizeGi2Quantity(100)},
			changed:            &cloudv1.TiKV{StorageSize: *util.StorageSizeGi2Quantity(200)},
			wantStepNames:      []string{"Remove scheduler balance-region-scheduler", "Scale tikv storage", "Add scheduler balance-region-scheduler"},
			wantScopes:         [][]string{nil, {consts.WfScaleClusterScopeStorage}, nil},
			wantSchedulerSteps: []int{0, 2}, // Del at index 0, Add at index 2
		},
		"storage class changed - no scheduler steps": {
			orig:          &cloudv1.TiKV{StorageClassName: "sc1"},
			changed:       &cloudv1.TiKV{StorageClassName: "sc2"},
			wantStepNames: []string{"Scale tikv storage"},
			wantScopes:    [][]string{{consts.WfScaleClusterScopeStorage}},
		},
		"useLocalStorage changed - no scheduler steps": {
			orig:          &cloudv1.TiKV{UseLocalStorage: lo.ToPtr(false)},
			changed:       &cloudv1.TiKV{UseLocalStorage: lo.ToPtr(true)},
			wantStepNames: []string{"Scale tikv storage"},
			wantScopes:    [][]string{{consts.WfScaleClusterScopeStorage}},
		},
		"instance type scale up": {
			orig:          &cloudv1.TiKV{InstanceType: "t2.medium"},
			changed:       &cloudv1.TiKV{InstanceType: "t2.large"},
			wantStepNames: []string{"Scale up tikv instance type"},
			wantScopes:    [][]string{{consts.WfScaleClusterScopeInstanceType}},
		},
		"instance type scale down": {
			orig:          &cloudv1.TiKV{InstanceType: "t2.large"},
			changed:       &cloudv1.TiKV{InstanceType: "t2.medium"},
			wantStepNames: []string{"Scale down tikv instance type"},
			wantScopes:    [][]string{{consts.WfScaleClusterScopeInstanceType}},
		},
		"replicas scale out": {
			orig:          &cloudv1.TiKV{Replicas: 3},
			changed:       &cloudv1.TiKV{Replicas: 5},
			wantStepNames: []string{"Scale out tikv replicas"},
			wantScopes:    [][]string{{consts.WfScaleClusterScopeReplicas}},
		},
		"replicas scale in": {
			orig:          &cloudv1.TiKV{Replicas: 5},
			changed:       &cloudv1.TiKV{Replicas: 3},
			wantStepNames: []string{"Scale in tikv replicas"},
			wantScopes:    [][]string{{consts.WfScaleClusterScopeReplicas}},
		},
		"config revision only": {
			orig:            &cloudv1.TiKV{},
			changed:         &cloudv1.TiKV{},
			newConfigRev:    "rev2",
			wantErrContains: "but instance type did not change",
		},
		"instance type and config revision": {
			orig:          &cloudv1.TiKV{InstanceType: "t2.medium"},
			changed:       &cloudv1.TiKV{InstanceType: "t2.large"},
			newConfigRev:  "rev2",
			wantStepNames: []string{"Scale up tikv instance type"},
			wantScopes:    [][]string{{consts.WfScaleClusterScopeInstanceType, consts.WfScaleClusterScopeConfigRevision}},
		},
		"up + in": {
			orig:               &cloudv1.TiKV{InstanceType: "t2.medium", Replicas: 5, StorageSize: *util.StorageSizeGi2Quantity(100)},
			changed:            &cloudv1.TiKV{InstanceType: "t2.large", Replicas: 3, StorageSize: *util.StorageSizeGi2Quantity(200)},
			newConfigRev:       "rev2",
			wantStepNames:      []string{"Remove scheduler balance-region-scheduler", "Scale tikv storage", "Add scheduler balance-region-scheduler", "Scale up tikv instance type", "Scale in tikv replicas"},
			wantScopes:         [][]string{nil, {consts.WfScaleClusterScopeStorage}, nil, {consts.WfScaleClusterScopeInstanceType, consts.WfScaleClusterScopeConfigRevision}, {consts.WfScaleClusterScopeReplicas}},
			wantSchedulerSteps: []int{0, 2},
		},
		"down + out": {
			orig:               &cloudv1.TiKV{InstanceType: "t2.large", Replicas: 3, StorageSize: *util.StorageSizeGi2Quantity(100)},
			changed:            &cloudv1.TiKV{InstanceType: "t2.medium", Replicas: 5, StorageSize: *util.StorageSizeGi2Quantity(200)},
			newConfigRev:       "rev2",
			wantStepNames:      []string{"Remove scheduler balance-region-scheduler", "Scale tikv storage", "Add scheduler balance-region-scheduler", "Scale out tikv replicas", "Scale down tikv instance type"},
			wantScopes:         [][]string{nil, {consts.WfScaleClusterScopeStorage}, nil, {consts.WfScaleClusterScopeReplicas}, {consts.WfScaleClusterScopeInstanceType, consts.WfScaleClusterScopeConfigRevision}},
			wantSchedulerSteps: []int{0, 2},
		},
		"storage + replicas without instance type change": {
			orig:               &cloudv1.TiKV{Replicas: 3, StorageSize: *util.StorageSizeGi2Quantity(100)},
			changed:            &cloudv1.TiKV{Replicas: 5, StorageSize: *util.StorageSizeGi2Quantity(200)},
			wantStepNames:      []string{"Remove scheduler balance-region-scheduler", "Scale tikv storage", "Add scheduler balance-region-scheduler", "Scale out tikv replicas"},
			wantScopes:         [][]string{nil, {consts.WfScaleClusterScopeStorage}, nil, {consts.WfScaleClusterScopeReplicas}},
			wantSchedulerSteps: []int{0, 2},
		},
		"instance type + storage without config": {
			orig:               &cloudv1.TiKV{InstanceType: "t2.medium", StorageSize: *util.StorageSizeGi2Quantity(100)},
			changed:            &cloudv1.TiKV{InstanceType: "t2.large", StorageSize: *util.StorageSizeGi2Quantity(200)},
			wantStepNames:      []string{"Remove scheduler balance-region-scheduler", "Scale tikv storage", "Add scheduler balance-region-scheduler", "Scale up tikv instance type"},
			wantScopes:         [][]string{nil, {consts.WfScaleClusterScopeStorage}, nil, {consts.WfScaleClusterScopeInstanceType}},
			wantSchedulerSteps: []int{0, 2},
		},
	}

	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			ctx := context.Background()
			mockProvider := cloudprovider.NewInstanceTypeProviderFromList(ctx, []cloudv1.InstanceType{
				{
					Name:   "",
					CPU:    resource.MustParse("0"),
					Memory: resource.MustParse("0"),
				},
				{
					Name:   "t2.medium",
					CPU:    resource.MustParse("2"),
					Memory: resource.MustParse("4Gi"),
				}, {
					Name:   "t2.large",
					CPU:    resource.MustParse("2"),
					Memory: resource.MustParse("8Gi"),
				},
			})

			// current cluster with an existing config revision
			cluster := &cloudv1.Cluster{
				Spec: cloudv1.ClusterSpec{
					ConfigRevisionRef: &cloudv1.Reference{Name: "rev1"},
				},
			}

			orig := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.orig}
			changed := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.changed}

			// if test doesn't specify a new config revision, keep it unchanged
			newConfigRev := tc.newConfigRev
			if newConfigRev == "" {
				newConfigRev = cluster.Spec.ConfigRevisionRef.Name
			}

			groups, err := buildScaleSteps(ctx, cluster, mockProvider, orig, changed, newConfigRev)
			if tc.wantErrContains != "" {
				assert.ErrorContains(t, err, tc.wantErrContains)
				return
			}
			assert.NoError(t, err)
			assert.Equal(t, len(tc.wantStepNames), len(groups), "number of groups mismatch")

			for i, wantStepName := range tc.wantStepNames {
				assert.Equal(t, wantStepName, groups[i].Name, "step name mismatch at index %d", i)

				// Check if this is a scheduler step
				isSchedulerStep := false
				for _, idx := range tc.wantSchedulerSteps {
					if idx == i {
						isSchedulerStep = true
						break
					}
				}

				if isSchedulerStep {
					// Verify scheduler step properties
					assert.Equal(t, consts.WfActionTogglePDScheduler, groups[i].Uses, "uses mismatch at scheduler step index %d", i)
					if op, ok := groups[i].Inputs.Object["op"].(string); ok {
						if wantStepName == "Remove scheduler balance-region-scheduler" {
							assert.Equal(t, "remove", op, "scheduler op mismatch at index %d", i)
						} else if wantStepName == "Add scheduler balance-region-scheduler" {
							assert.Equal(t, "add", op, "scheduler op mismatch at index %d", i)
						}
					} else {
						t.Errorf("failed to extract scheduler op from inputs at index %d", i)
					}
					if scheduler, ok := groups[i].Inputs.Object["scheduler"].(string); ok {
						assert.Equal(t, consts.PDSchedulerBalanceRegion, scheduler, "scheduler name mismatch at index %d", i)
					} else {
						t.Errorf("failed to extract scheduler name from inputs at index %d", i)
					}
				} else {
					// Extract scopes from the inputs map for non-scheduler steps
					if tc.wantScopes[i] != nil {
						if scopes, ok := groups[i].Inputs.Object["scopes"].([]string); ok {
							assert.Equal(t, tc.wantScopes[i], scopes, "scopes mismatch at index %d", i)
						} else {
							t.Errorf("failed to extract scopes from inputs at index %d", i)
						}
					}
				}
			}
		})
	}
}

func TestBuildScaleScopeGroupsForStorageChange(t *testing.T) {
	cases := map[string]struct {
		orig           cloudv1.ClusterComponent
		changed        cloudv1.ClusterComponent
		wantStepNames  []string
		wantPriorities []int
		wantUses       []cloudv1.WorkflowActionName
		wantOps        []string // for scheduler steps: "remove", "add", "" for non-scheduler
		wantScopes     [][]string
	}{
		"tikv storage size changed triggers scheduler steps": {
			orig:          &cloudv1.TiKV{StorageSize: *util.StorageSizeGi2Quantity(100)},
			changed:       &cloudv1.TiKV{StorageSize: *util.StorageSizeGi2Quantity(200)},
			wantStepNames: []string{"Remove scheduler balance-region-scheduler", "Add scheduler balance-region-scheduler", "Scale tikv storage"},
			wantPriorities: []int{
				scaleDelBalanceRegionSchedulerPriority,
				scaleAddBalanceRegionSchedulerPriority,
				scaleStoragePriority,
			},
			wantUses: []cloudv1.WorkflowActionName{
				consts.WfActionTogglePDScheduler,
				consts.WfActionTogglePDScheduler,
				consts.WorkflowActionScaleCluster,
			},
			wantOps:    []string{"remove", "add", ""},
			wantScopes: [][]string{nil, nil, {consts.WfScaleClusterScopeStorage}},
		},
		"tiflash storage size changed triggers scheduler steps": {
			orig:          &cloudv1.TiFlash{StorageSize: *util.StorageSizeGi2Quantity(100)},
			changed:       &cloudv1.TiFlash{StorageSize: *util.StorageSizeGi2Quantity(200)},
			wantStepNames: []string{"Remove scheduler balance-region-scheduler", "Add scheduler balance-region-scheduler", "Scale tiflash storage"},
			wantPriorities: []int{
				scaleDelBalanceRegionSchedulerPriority,
				scaleAddBalanceRegionSchedulerPriority,
				scaleStoragePriority,
			},
			wantUses: []cloudv1.WorkflowActionName{
				consts.WfActionTogglePDScheduler,
				consts.WfActionTogglePDScheduler,
				consts.WorkflowActionScaleCluster,
			},
			wantOps:    []string{"remove", "add", ""},
			wantScopes: [][]string{nil, nil, {consts.WfScaleClusterScopeStorage}},
		},
		"PD storage size changed does not trigger scheduler steps": {
			orig:          &cloudv1.PD{Replicas: 3, StorageSize: *util.StorageSizeGi2Quantity(100)},
			changed:       &cloudv1.PD{Replicas: 3, StorageSize: *util.StorageSizeGi2Quantity(200)},
			wantStepNames: []string{"Scale pd storage"},
			wantPriorities: []int{
				scaleStoragePriority,
			},
			wantUses:   []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantOps:    []string{""},
			wantScopes: [][]string{{consts.WfScaleClusterScopeStorage}},
		},
		"tikv storage class changed": {
			orig:          &cloudv1.TiKV{StorageClassName: "sc1"},
			changed:       &cloudv1.TiKV{StorageClassName: "sc2"},
			wantStepNames: []string{"Scale tikv storage"},
			wantPriorities: []int{
				scaleStoragePriority,
			},
			wantUses:   []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantOps:    []string{""},
			wantScopes: [][]string{{consts.WfScaleClusterScopeStorage}},
		},
		"tikv useLocalStorage changed": {
			orig:          &cloudv1.TiKV{UseLocalStorage: lo.ToPtr(false)},
			changed:       &cloudv1.TiKV{UseLocalStorage: lo.ToPtr(true)},
			wantStepNames: []string{"Scale tikv storage"},
			wantPriorities: []int{
				scaleStoragePriority,
			},
			wantUses:   []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantOps:    []string{""},
			wantScopes: [][]string{{consts.WfScaleClusterScopeStorage}},
		},
	}

	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			orig := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.orig}
			changed := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.changed}

			groups := buildScaleStepsForStorageChange(orig, changed)
			assert.Equal(t, len(tc.wantStepNames), len(groups), "number of groups mismatch")

			for i, g := range groups {
				assert.Equal(t, tc.wantStepNames[i], g.step.Name, "step name mismatch at %d", i)
				assert.Equal(t, tc.wantPriorities[i], g.priority, "priority mismatch at %d", i)
				assert.Equal(t, tc.wantUses[i], g.step.Uses, "uses mismatch at %d", i)
				if tc.wantOps[i] != "" {
					assert.Equal(t, tc.wantOps[i], g.step.Inputs.Object["op"], "op mismatch at %d", i)
					assert.Equal(t, consts.PDSchedulerBalanceRegion, g.step.Inputs.Object["scheduler"], "scheduler mismatch at %d", i)
				}
				if tc.wantScopes[i] != nil {
					if scopes, ok := g.step.Inputs.Object["scopes"].([]string); ok {
						assert.Equal(t, tc.wantScopes[i], scopes, "scopes mismatch at %d", i)
					} else {
						t.Errorf("failed to extract scopes from inputs at %d", i)
					}
				}
			}
		})
	}
}

func TestBuildScaleStepsForInstanceTypeChanged(t *testing.T) {
	cases := map[string]struct {
		orig             cloudv1.ClusterComponent
		changed          cloudv1.ClusterComponent
		newConfigRev     string
		currentConfigRev string
		wantStepNames    []string
		wantPriorities   []int
		wantUses         []cloudv1.WorkflowActionName
		wantScopes       [][]string
		wantErrContains  string
	}{
		"no instance type change, no config change": {
			orig:             &cloudv1.TiKV{InstanceType: "t2.medium"},
			changed:          &cloudv1.TiKV{InstanceType: "t2.medium"},
			currentConfigRev: "rev1",
			newConfigRev:     "rev1",
			wantStepNames:    []string{},
		},
		"no instance type change, config changed": {
			orig:             &cloudv1.TiKV{InstanceType: "t2.medium"},
			changed:          &cloudv1.TiKV{InstanceType: "t2.medium"},
			currentConfigRev: "rev1",
			newConfigRev:     "rev2",
			wantErrContains:  "but instance type did not change",
		},
		"instance type scale up without config change": {
			orig:             &cloudv1.TiKV{InstanceType: "t2.medium"},
			changed:          &cloudv1.TiKV{InstanceType: "t2.large"},
			currentConfigRev: "rev1",
			newConfigRev:     "rev1",
			wantStepNames:    []string{"Scale up tikv instance type"},
			wantPriorities:   []int{scaleUpPriority},
			wantUses:         []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:       [][]string{{consts.WfScaleClusterScopeInstanceType}},
		},
		"instance type scale up with config change": {
			orig:             &cloudv1.TiKV{InstanceType: "t2.medium"},
			changed:          &cloudv1.TiKV{InstanceType: "t2.large"},
			currentConfigRev: "rev1",
			newConfigRev:     "rev2",
			wantStepNames:    []string{"Scale up tikv instance type"},
			wantPriorities:   []int{scaleUpPriority},
			wantUses:         []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:       [][]string{{consts.WfScaleClusterScopeInstanceType, consts.WfScaleClusterScopeConfigRevision}},
		},
		"instance type scale down without config change": {
			orig:             &cloudv1.TiKV{InstanceType: "t2.large"},
			changed:          &cloudv1.TiKV{InstanceType: "t2.medium"},
			currentConfigRev: "rev1",
			newConfigRev:     "rev1",
			wantStepNames:    []string{"Scale down tikv instance type"},
			wantPriorities:   []int{scaleDownPriority},
			wantUses:         []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:       [][]string{{consts.WfScaleClusterScopeInstanceType}},
		},
		"instance type scale down with config change": {
			orig:             &cloudv1.TiKV{InstanceType: "t2.large"},
			changed:          &cloudv1.TiKV{InstanceType: "t2.medium"},
			currentConfigRev: "rev1",
			newConfigRev:     "rev2",
			wantStepNames:    []string{"Scale down tikv instance type"},
			wantPriorities:   []int{scaleDownPriority},
			wantUses:         []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:       [][]string{{consts.WfScaleClusterScopeInstanceType, consts.WfScaleClusterScopeConfigRevision}},
		},
		"tiflash instance type scale up": {
			orig:             &cloudv1.TiFlash{InstanceType: "t2.medium"},
			changed:          &cloudv1.TiFlash{InstanceType: "t2.large"},
			currentConfigRev: "rev1",
			newConfigRev:     "rev1",
			wantStepNames:    []string{"Scale up tiflash instance type"},
			wantPriorities:   []int{scaleUpPriority},
			wantUses:         []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:       [][]string{{consts.WfScaleClusterScopeInstanceType}},
		},
		"pd instance type scale up": {
			orig:             &cloudv1.PD{InstanceType: "t2.medium", Replicas: 3},
			changed:          &cloudv1.PD{InstanceType: "t2.large", Replicas: 3},
			currentConfigRev: "rev1",
			newConfigRev:     "rev1",
			wantStepNames:    []string{"Scale up pd instance type"},
			wantPriorities:   []int{scaleUpPriority},
			wantUses:         []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:       [][]string{{consts.WfScaleClusterScopeInstanceType}},
		},
	}

	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			ctx := context.Background()
			mockProvider := cloudprovider.NewInstanceTypeProviderFromList(ctx, []cloudv1.InstanceType{
				{
					Name:   "t2.medium",
					CPU:    resource.MustParse("2"),
					Memory: resource.MustParse("4Gi"),
				},
				{
					Name:   "t2.large",
					CPU:    resource.MustParse("2"),
					Memory: resource.MustParse("8Gi"),
				},
			})

			cluster := &cloudv1.Cluster{
				Spec: cloudv1.ClusterSpec{
					ConfigRevisionRef: &cloudv1.Reference{Name: tc.currentConfigRev},
				},
			}

			orig := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.orig}
			changed := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.changed}

			groups, err := buildScaleStepsForInstanceTypeChanged(ctx, mockProvider, cluster, orig, changed, tc.newConfigRev)

			if tc.wantErrContains != "" {
				assert.ErrorContains(t, err, tc.wantErrContains)
				return
			}

			assert.NoError(t, err)
			assert.Equal(t, len(tc.wantStepNames), len(groups), "number of groups mismatch")

			for i, g := range groups {
				assert.Equal(t, tc.wantStepNames[i], g.step.Name, "step name mismatch at %d", i)
				assert.Equal(t, tc.wantPriorities[i], g.priority, "priority mismatch at %d", i)
				assert.Equal(t, tc.wantUses[i], g.step.Uses, "uses mismatch at %d", i)
				if tc.wantScopes[i] != nil {
					if scopes, ok := g.step.Inputs.Object["scopes"].([]string); ok {
						assert.Equal(t, tc.wantScopes[i], scopes, "scopes mismatch at %d", i)
					} else {
						t.Errorf("failed to extract scopes from inputs at %d", i)
					}
				}
			}
		})
	}
}

func TestBuildScaleStepsForReplicasChanged(t *testing.T) {
	cases := map[string]struct {
		orig           cloudv1.ClusterComponent
		changed        cloudv1.ClusterComponent
		wantStepNames  []string
		wantPriorities []int
		wantUses       []cloudv1.WorkflowActionName
		wantScopes     [][]string
	}{
		"no replicas change": {
			orig:          &cloudv1.TiKV{Replicas: 3},
			changed:       &cloudv1.TiKV{Replicas: 3},
			wantStepNames: []string{},
		},
		"tikv scale out": {
			orig:           &cloudv1.TiKV{Replicas: 3},
			changed:        &cloudv1.TiKV{Replicas: 5},
			wantStepNames:  []string{"Scale out tikv replicas"},
			wantPriorities: []int{scaleOutPriority},
			wantUses:       []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:     [][]string{{consts.WfScaleClusterScopeReplicas}},
		},
		"tikv scale in": {
			orig:           &cloudv1.TiKV{Replicas: 5},
			changed:        &cloudv1.TiKV{Replicas: 3},
			wantStepNames:  []string{"Scale in tikv replicas"},
			wantPriorities: []int{scaleInPriority},
			wantUses:       []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:     [][]string{{consts.WfScaleClusterScopeReplicas}},
		},
		"tiflash scale out": {
			orig:           &cloudv1.TiFlash{Replicas: 2},
			changed:        &cloudv1.TiFlash{Replicas: 4},
			wantStepNames:  []string{"Scale out tiflash replicas"},
			wantPriorities: []int{scaleOutPriority},
			wantUses:       []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:     [][]string{{consts.WfScaleClusterScopeReplicas}},
		},
		"tiflash scale in": {
			orig:           &cloudv1.TiFlash{Replicas: 4},
			changed:        &cloudv1.TiFlash{Replicas: 2},
			wantStepNames:  []string{"Scale in tiflash replicas"},
			wantPriorities: []int{scaleInPriority},
			wantUses:       []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:     [][]string{{consts.WfScaleClusterScopeReplicas}},
		},
		"pd scale out": {
			orig:           &cloudv1.PD{Replicas: 3},
			changed:        &cloudv1.PD{Replicas: 5},
			wantStepNames:  []string{"Scale out pd replicas"},
			wantPriorities: []int{scaleOutPriority},
			wantUses:       []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:     [][]string{{consts.WfScaleClusterScopeReplicas}},
		},
		"pd scale in": {
			orig:           &cloudv1.PD{Replicas: 5},
			changed:        &cloudv1.PD{Replicas: 3},
			wantStepNames:  []string{"Scale in pd replicas"},
			wantPriorities: []int{scaleInPriority},
			wantUses:       []cloudv1.WorkflowActionName{consts.WorkflowActionScaleCluster},
			wantScopes:     [][]string{{consts.WfScaleClusterScopeReplicas}},
		},
	}

	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			orig := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.orig}
			changed := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.changed}

			groups := buildScaleStepsForReplicasChanged(orig, changed)
			assert.Equal(t, len(tc.wantStepNames), len(groups), "number of groups mismatch")

			for i, g := range groups {
				assert.Equal(t, tc.wantStepNames[i], g.step.Name, "step name mismatch at %d", i)
				assert.Equal(t, tc.wantPriorities[i], g.priority, "priority mismatch at %d", i)
				assert.Equal(t, tc.wantUses[i], g.step.Uses, "uses mismatch at %d", i)
				if tc.wantScopes[i] != nil {
					if scopes, ok := g.step.Inputs.Object["scopes"].([]string); ok {
						assert.Equal(t, tc.wantScopes[i], scopes, "scopes mismatch at %d", i)
					} else {
						t.Errorf("failed to extract scopes from inputs at %d", i)
					}
				}
			}
		})
	}
}

func TestIsAdditionalStorageChanged(t *testing.T) {
	cases := map[string]struct {
		orig    *cloudv1.TiKV
		changed *cloudv1.TiKV
		want    bool
	}{
		"no additional storage": {
			orig:    &cloudv1.TiKV{},
			changed: &cloudv1.TiKV{},
			want:    false,
		},
		"no change in additional storage": {
			orig: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
					{Name: "vol2", StorageSize: *util.StorageSizeGi2Quantity(200), StorageClassName: "sc2"},
				},
			},
			changed: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
					{Name: "vol2", StorageSize: *util.StorageSizeGi2Quantity(200), StorageClassName: "sc2"},
				},
			},
			want: false,
		},
		"additional storage added": {
			orig: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
				},
			},
			changed: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
					{Name: "vol2", StorageSize: *util.StorageSizeGi2Quantity(200), StorageClassName: "sc2"},
				},
			},
			want: true,
		},
		"additional storage removed": {
			orig: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
					{Name: "vol2", StorageSize: *util.StorageSizeGi2Quantity(200), StorageClassName: "sc2"},
				},
			},
			changed: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
				},
			},
			want: true,
		},
		"storage size changed": {
			orig: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
				},
			},
			changed: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(200), StorageClassName: "sc1"},
				},
			},
			want: true,
		},
		"storage class changed": {
			orig: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
				},
			},
			changed: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc2"},
				},
			},
			want: true,
		},
		"volume name changed": {
			orig: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
				},
			},
			changed: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol2", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
				},
			},
			want: true,
		},
		"both size and class changed": {
			orig: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
				},
			},
			changed: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(200), StorageClassName: "sc2"},
				},
			},
			want: true,
		},
		"multiple volumes with one changed": {
			orig: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
					{Name: "vol2", StorageSize: *util.StorageSizeGi2Quantity(200), StorageClassName: "sc2"},
					{Name: "vol3", StorageSize: *util.StorageSizeGi2Quantity(200), StorageClassName: "sc3"},
				},
			},
			changed: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
					{Name: "vol2", StorageSize: *util.StorageSizeGi2Quantity(250), StorageClassName: "sc2"},
					{Name: "vol3", StorageSize: *util.StorageSizeGi2Quantity(300), StorageClassName: "sc3"},
				},
			},
			want: true,
		},
		"order changed but content same": {
			orig: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
					{Name: "vol2", StorageSize: *util.StorageSizeGi2Quantity(200), StorageClassName: "sc2"},
				},
			},
			changed: &cloudv1.TiKV{
				StorageVolumes: []cloudv1.ComponentStorageVolume{
					{Name: "vol2", StorageSize: *util.StorageSizeGi2Quantity(200), StorageClassName: "sc2"},
					{Name: "vol1", StorageSize: *util.StorageSizeGi2Quantity(100), StorageClassName: "sc1"},
				},
			},
			want: false,
		},
		"from nil to empty": {
			orig:    &cloudv1.TiKV{StorageVolumes: nil},
			changed: &cloudv1.TiKV{StorageVolumes: []cloudv1.ComponentStorageVolume{}},
			want:    false,
		},
		"from empty to nil": {
			orig:    &cloudv1.TiKV{StorageVolumes: []cloudv1.ComponentStorageVolume{}},
			changed: &cloudv1.TiKV{StorageVolumes: nil},
			want:    false,
		},
	}

	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			orig := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.orig}
			changed := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.changed}

			got := isAdditionalStorageChanged(orig, changed)
			assert.Equal(t, tc.want, got)
		})
	}

	// tidb should return always false
	orig := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: &cloudv1.TiDB{}}
	changed := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: &cloudv1.TiDB{}}
	assert.Equalf(t, false, isAdditionalStorageChanged(orig, changed), "TiDB should not consider additional storage changes")
}

func TestHasStorageChanged(t *testing.T) {
	cases := map[string]struct {
		orig    *cloudv1.TiKV
		changed *cloudv1.TiKV
		want    bool
	}{
		"no change": {
			orig: &cloudv1.TiKV{
				StorageSize:      *util.StorageSizeGi2Quantity(100),
				StorageClassName: "sc1",
				UseLocalStorage:  lo.ToPtr(false),
			},
			changed: &cloudv1.TiKV{
				StorageSize:      *util.StorageSizeGi2Quantity(100),
				StorageClassName: "sc1",
				UseLocalStorage:  lo.ToPtr(false),
			},
			want: false,
		},
		"storage size changed": {
			orig:    &cloudv1.TiKV{StorageSize: *util.StorageSizeGi2Quantity(100)},
			changed: &cloudv1.TiKV{StorageSize: *util.StorageSizeGi2Quantity(200)},
			want:    true,
		},
		"storage class changed": {
			orig:    &cloudv1.TiKV{StorageClassName: "sc1"},
			changed: &cloudv1.TiKV{StorageClassName: "sc2"},
			want:    true,
		},
		"useLocalStorage nil to true": {
			orig:    &cloudv1.TiKV{UseLocalStorage: nil},
			changed: &cloudv1.TiKV{UseLocalStorage: lo.ToPtr(true)},
			want:    true,
		},
		"useLocalStorage both nil": {
			orig:    &cloudv1.TiKV{UseLocalStorage: nil},
			changed: &cloudv1.TiKV{UseLocalStorage: nil},
			want:    false,
		},
		"multiple storage changes": {
			orig: &cloudv1.TiKV{
				StorageSize:      *util.StorageSizeGi2Quantity(100),
				StorageClassName: "sc1",
				UseLocalStorage:  lo.ToPtr(false),
			},
			changed: &cloudv1.TiKV{
				StorageSize:      *util.StorageSizeGi2Quantity(200),
				StorageClassName: "sc2",
				UseLocalStorage:  lo.ToPtr(true),
			},
			want: true,
		},
		"only non-storage fields changed": {
			orig: &cloudv1.TiKV{
				StorageSize: *util.StorageSizeGi2Quantity(100),
				Replicas:    3,
			},
			changed: &cloudv1.TiKV{
				StorageSize: *util.StorageSizeGi2Quantity(100),
				Replicas:    5,
			},
			want: false,
		},
	}

	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			orig := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.orig}
			changed := &cloudv1.ClusterComponentSpecWrapper{ClusterComponent: tc.changed}

			got := hasStorageChanged(orig, changed)
			assert.Equal(t, tc.want, got)
		})
	}
}