package operation

import (
	"context"
	"fmt"
	"maps"
	"strconv"
	"strings"
	"time"

	"github.com/Masterminds/semver"
	"github.com/crossplane-contrib/provider-aws/apis/eks/v1beta1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/timestamppb"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"

	internalutil "github.com/tidbcloud/cluster-controller/server/internal/util"
	"github.com/tidbcloud/cluster-controller/server/server"
	"github.com/tidbcloud/cluster-controller/server/server/cvt"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/infra-provider/api/tidbcloud/common"
	"github.com/tidbcloud/infra-provider/pkg/cmpver"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/gkemindomains"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	utiltime "github.com/tidbcloud/infra-provider/pkg/util/time"
)

const (
	scheduleDelay                       = 30 * time.Second
	minTiDBFallbackableSupportedVersion = "v6.5.1"
)

type Server struct {
	apiv1.UnimplementedClusterOperationServiceServer

	KubeClient    client.Client
	clusterReader internalutil.ClusterReader

	clock utiltime.Clock
}

func NewServer(kubeClient client.Client, clock utiltime.Clock) *Server {
	return &Server{
		KubeClient:    kubeClient,
		clusterReader: &internalutil.LabelSelectorReader{},
		clock:         clock,
	}
}

func (s *Server) CreateOperation(ctx context.Context, req *apiv1.CreateOperationRequest) (*apiv1.CreateOperationResponse, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))

	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	op := req.Operation
	wfType := cvt.OperationType2WFType(op.Type)
	wf := &cloudv1.Workflow{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: cluster.Namespace,
			Name:      genOperationWFName(op),
			Labels: map[string]string{
				cloudv1.LabelServiceType:   cloudv1.ServiceTypeDedicated,
				cloudv1.WorkflowTypeTagKey: string(wfType),
				cloudv1.IdempotencyKey:     op.IdempotencyKey,
				cloudv1.TenantTagKey:       cluster.GetTenantID(),
				cloudv1.ProjectTagKey:      cluster.GetProjectID(),
				cloudv1.ClusterTagKey:      cluster.GetClusterID(),
				cloudv1.ProviderTypeLabel:  cluster.GetProviderType(),
				cloudv1.RegionLabel:        cluster.GetRegion(),
			},
		},
		Spec: cloudv1.WorkflowSpec{
			Type: wfType,
		},
	}
	// use the same reconciler as the cluster
	if reconciler, ok := cluster.Labels[cloudv1.ReconcilerLabelKey]; ok {
		wf.Labels[cloudv1.ReconcilerLabelKey] = reconciler
	}

	switch op.Type {
	case apiv1.Operation_RECREATE_TC_MEMBER:
		err = s.buildRecreateTCMemberWF(ctx, wf, req.Operation.GetRecreateTcMember(), cluster)

	case apiv1.Operation_RETIRE_NODE:
		err = s.buildRetireNodeWF(ctx, wf, req.Operation.GetRetireNode(), cluster)

	case apiv1.Operation_SHUTDOWN_UPGRADE_NODES:
		args := req.Operation.GetShutdownUpgradeNodes()
		if yes, err := s.canUpgradeNodes(ctx, cluster, args); err != nil || !yes {
			return nil, status.Errorf(codes.InvalidArgument, "can not do shutdown upgrade nodes: %s", err)
		}
		s.buildShutDownUpgradeNodesWF(wf, args, cluster)

	case apiv1.Operation_RESTART_OTHER_PODS:
		err = s.buildRestartOtherPodsWF(wf, req.Operation.GetRestartOtherPods(), cluster)

	case apiv1.Operation_MIGRATE_TIKV_TO_DUAL_DISK:
		err = s.buildMigrateTiKVToDualDiskWF(wf, req.Operation.GetMigrateTikvToDualDisk(), cluster)

	case apiv1.Operation_FIX_TIKV_ILLEGAL_BOUNDARY:
		err = s.buildFixTiKVIllegalBoundaryWF(wf, req.Operation.GetFixTikvIllegalBoundary(), cluster)

	case apiv1.Operation_ENABLE_PROXY_PROTOCOL_FOR_CLUSTER:
		args := req.Operation.GetEnableProxyProtocolForCluster()
		if err := s.canEnablePPv2(ctx, cluster, args); err != nil {
			return nil, status.Errorf(codes.InvalidArgument, "failed enable PPv2: %s", err)
		}
		err = s.buildEnablePPv2ForClusterWF(wf, args, cluster)

	case apiv1.Operation_ENABLE_LOAD_BALANCE_LOGS:
		err = s.buildModifyLoadBalanceLogsWF(wf, req.Operation.GetEnableLoadBalanceLogs(), cluster)

	case apiv1.Operation_ENABLE_VPC_FLOW_LOGS:
		err = s.buildModifyVPCFlowLogsWF(wf, req.Operation.GetEnableVpcFlowLogs(), cluster)

	case apiv1.Operation_OFFLINE_AZ:
		err = s.buildOfflineAZWF(wf, req.Operation.GetOfflineAz(), cluster)

	case apiv1.Operation_RECOVER_AZ:
		err = s.buildRecoverAZWF(wf, req.Operation.GetRecoverAz(), cluster)

	case apiv1.Operation_START_ZONAL_SHIFT:
		err = s.buildStartZonalShiftWF(wf, req.Operation.GetStartZonalShift(), cluster)

	case apiv1.Operation_UPDATE_ZONAL_SHIFT:
		err = s.buildUpdateZonalShiftWF(wf, req.Operation.GetUpdateZonalShift(), cluster)

	case apiv1.Operation_STOP_ZONAL_SHIFT:
		err = s.buildStopZonalShiftWF(wf, req.Operation.GetStopZonalShift(), cluster)

	default:
		return nil, status.Errorf(codes.InvalidArgument, "invalid type: %s", op.Type)
	}
	if err != nil {
		return nil, err
	}
	logger.Debug("wf", zap.Any("wf", wf))
	if err := k8s.Apply(ctx, s.KubeClient, wf); err != nil {
		return nil, fmt.Errorf("apply workflow, err: %w", err)
	}

	logger.Info("workflow created", zap.String("wfName", wf.NamespacedName()))
	resp := &apiv1.CreateOperationResponse{
		Id: wf.Name,
	}
	return resp, nil
}

func (s *Server) buildRecreateTCMemberWF(ctx context.Context, wf *cloudv1.Workflow, args *apiv1.Operation_RecreateTCMember, cluster *cloudv1.Cluster) error {
	steps := cloudv1.WorkflowSteps{
		{
			Name: "Get pod info",
			Uses: consts.WfActionGetPodInfoAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"podName":     args.MemberName,
				}},
		},
		{
			Name: "Create standby node",
			Uses: consts.WfActionSetStandbyNodeAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":     cluster.Namespace,
					"clusterName":   cluster.Name,
					"componentType": "$steps.Get pod info.outputs.componentType",
					"zone":          "$steps.Get pod info.outputs.zone",
					"count":         int64(1),
					"waitNodeReady": true,
				}},
		},
		{
			Name: "Cordon old node",
			Uses: consts.WfActionCordonNodeAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"memberName":  args.MemberName,
				}},
		},
		{
			Name: "Delete TC Member",
			Uses: consts.WfActionDeleteTCMemberAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"memberName":  args.MemberName,
				}},
		},
		{
			Name: "Wait pod to be Running",
			Uses: consts.WfActionWaitPodPhaseAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"podName":     "$steps.Delete TC Member.outputs.newMemberName",
					"phase":       string(corev1.PodRunning),
				}},
		},
		{
			Name: "Remove standby node",
			Uses: consts.WfActionSetStandbyNodeAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":     cluster.Namespace,
					"clusterName":   cluster.Name,
					"componentType": "$steps.Get pod info.outputs.componentType",
					"zone":          "$steps.Get pod info.outputs.zone",
					"count":         int64(0),
				}},
		},
	}
	wf.Spec.Steps = steps

	return nil
}

func (s *Server) buildRetireNodeWF(ctx context.Context, wf *cloudv1.Workflow, args *apiv1.Operation_RetireNode, cluster *cloudv1.Cluster) error {
	steps := cloudv1.WorkflowSteps{
		{
			Name: "Get pod info",
			Uses: consts.WfActionGetPodInfoAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"podName":     args.PodName,
				}},
		},
		{
			Name: "Create standby node",
			Uses: consts.WfActionSetStandbyNodeAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":     cluster.Namespace,
					"clusterName":   cluster.Name,
					"componentType": "$steps.Get pod info.outputs.componentType",
					"zone":          "$steps.Get pod info.outputs.zone",
					"count":         int64(1),
					"waitNodeReady": true,
				}},
		},
		{
			Name: "Cordon old node",
			Uses: consts.WfActionCordonNodeAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"memberName":  args.PodName,
				}},
		},
		{
			Name: "Restart pod to use new node",
			Uses: consts.WfActionRestartPodAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"podName":     args.PodName,
				}},
		},
		{
			Name: "Wait pod to be Running",
			Uses: consts.WfActionWaitPodPhaseAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"podName":     args.PodName,
					"phase":       string(corev1.PodRunning),
					// ensure the pod is recreated by 'Restart pod to use new node' step
					"ensureCreatedAfter": "$steps.Restart pod to use new node.status.beginTime",
				}},
		},
		{
			Name: "Remove standby node",
			Uses: consts.WfActionSetStandbyNodeAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":     cluster.Namespace,
					"clusterName":   cluster.Name,
					"componentType": "$steps.Get pod info.outputs.componentType",
					"zone":          "$steps.Get pod info.outputs.zone",
					"count":         int64(0),
				}},
		},
	}
	wf.Spec.Steps = steps

	return nil
}

func (s *Server) buildShutDownUpgradeNodesWF(wf *cloudv1.Workflow, args *apiv1.Operation_ShutDownUpgradeNodes, cluster *cloudv1.Cluster) {
	steps := cloudv1.WorkflowSteps{
		{
			Name: "Pause cluster reconciliation",
			Uses: consts.WfActionPauseReconciliationAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace": cluster.Namespace,
					"name":      cluster.Name,
					"type":      "cluster",
				},
			},
		},
		{
			Name: "Pause all TCs reconciliation",
			Uses: consts.WfActionPauseReconciliationAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace": cluster.Namespace,
					"name":      cluster.Name,
					"type":      "tidbcluster",
				},
			},
		},
		{
			Name: "Cordon old version nodes",
			Uses: consts.WfActionCordonNodeAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"memberName":  "ALL",
				},
			},
		},
		{
			// Note: this action will not delete TiCDC's ASTS
			Name: "Delete all ASTS",
			Uses: consts.WfActionDeleteASTSAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace": cluster.Namespace,
					"cluster":   cluster.Name,
				},
			},
		},
		{
			// Note: this action will not delete TiCDC nodes,
			// because they use managed node groups.
			Name: "Delete all ASGs of a tidb cluster to delete nodes",
			Uses: consts.WfActionDeleteTiDBClusterNodesAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"cluster":   cluster.Name,
					"namespace": cluster.Namespace,
				},
			},
		},
		{
			Name: "Set the desired version",
			Uses: consts.WfActionUpdateClusterStatusAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"name":        cluster.Name,
					"namespace":   cluster.Namespace,
					"nodeVersion": args.DesiredVersion,
				},
			},
		},
		{
			Name: "Resume cluster reconciliation",
			Uses: consts.WfActionResumeReconciliationAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace": cluster.Namespace,
					"name":      cluster.Name,
					"type":      "cluster",
				},
			},
		},
		{
			Name: "Resume all TCs reconciliation",
			Uses: consts.WfActionResumeReconciliationAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace": cluster.Namespace,
					"name":      cluster.Name,
					"type":      "tidbcluster",
				},
			},
		},
	}
	wf.Spec.Steps = steps
}

func (s *Server) buildRestartOtherPodsWF(wf *cloudv1.Workflow, args *apiv1.Operation_RestartOtherPods, cluster *cloudv1.Cluster) error {
	steps := cloudv1.WorkflowSteps{}
	for i, podName := range args.PodNames {
		steps = append(steps, cloudv1.WorkflowStep{
			Name: fmt.Sprintf("Restart Other Pod %v", i),
			Uses: consts.WfActionRestartOtherPodAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterNamespace": cluster.Namespace,
					"clusterName":      cluster.Name,
					"namespace":        args.Namespace,
					"podName":          podName,
					"force":            args.Force,
				}},
		})
	}
	wf.Spec.Steps = steps

	return nil
}

func (s *Server) buildMigrateTiKVToDualDiskWF(wf *cloudv1.Workflow, args *apiv1.Operation_MigrateTiKVToDualDisk, cluster *cloudv1.Cluster) error {
	if err := validateMigrateTiKVToDualDiskWithActiveGKEVAC(args, cluster); err != nil {
		return err
	}
	storageClassName := cvt.InfraStorageClassName(
		common.ComponentType_TIKV, args.AdditionalStorage.Type, int32(args.AdditionalStorage.Iops), int32(args.AdditionalStorage.ThroughputMbs))
	// convert to string to avoid "panic triggered: cannot deep copy uint32"
	sizeGiStr := strconv.FormatUint(uint64(args.AdditionalStorage.SizeGi), 10)
	steps := cloudv1.WorkflowSteps{
		{
			Name: "Update Cluster CR",
			Uses: "RaftMigrationUpdateClusterAction",
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace": cluster.Namespace,
					"name":      cluster.Name,
					"additionalStorage": map[string]interface{}{
						"name":             args.AdditionalStorage.Name,
						"storageClassName": storageClassName,
						"sizeGi":           sizeGiStr,
						"mountPath":        args.AdditionalStorage.MountPath,
					},
					"config": args.ConfigurationFileContents,
				},
			},
		},
		{
			// if the ConfigMap of TiKV is updated (by TiDB Operator),
			// then we can also ensure the additional volumes also represented in the TidbCluster CR.
			// as the config and the volume of the TidbCluster should be updated in one request by the cluster-controller.
			Name: "Wait for TiKV ConfigMap to be updated",
			Uses: "EnsureDataplaneConfigMapContainAction",
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterNamespace": cluster.Namespace,
					"clusterName":      cluster.Name,
					"namespace":        cluster.PrefixedId(),
					"asts":             genAstsName(cloudv1.ComponentTiKV), // always "db-tikv"
					// the configmap key is always "config-file" for PD, TiKV and TiDB
					"cmKey": "config-file",
					// when migrating to dual disk, the config should contain "raft-engine.dir"
					"configKey": "raft-engine.dir", // the target key in configmap
				},
			},
		},
		{
			// re-create the ASTS to contain the new volume
			Name: "Delete TiKV asts",
			Uses: "DeleteOneASTSAction",
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterNamespace": cluster.Namespace,
					"clusterName":      cluster.Name,
					"namespace":        cluster.PrefixedId(),
					"name":             genAstsName(cloudv1.ComponentTiKV), // always "db-tikv"
				},
			},
		},
		{
			Name: "Wait for TiKV asts to be recreated with the new volume",
			Uses: "EnsureASTSContainVolumeAction",
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterNamespace": cluster.Namespace,
					"clusterName":      cluster.Name,
					"namespace":        cluster.PrefixedId(),
					"name":             genAstsName(cloudv1.ComponentTiKV), // always "db-tikv"
					// e.g. `tikv-raftlog`
					"volume": string(opv1alpha1.GetStorageVolumeName(args.AdditionalStorage.Name, opv1alpha1.TiKVMemberType)),
				},
			},
		},
		{
			// in the current ASTS controller, without this step the pods may be re-created accidentally
			Name: "Wait for new asts to adopt pods",
			Uses: "EnsureAstsAdoptPodsAction",
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterNamespace": cluster.Namespace,
					"clusterName":      cluster.Name,
					"namespace":        cluster.PrefixedId(),
					"componentType":    string(cloudv1.ComponentTiKV),
					"name":             genAstsName(cloudv1.ComponentTiKV), // always "db-tikv"
				},
			},
		},
		{
			// the ASTS controller will not create this new PVCs automatically
			// so we need to create them manually before copying the raft data
			Name: "Create new PVCs for TiKV",
			Uses: "CreatePVCsAction",
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterNamespace": cluster.Namespace,
					"clusterName":      cluster.Name,
					"namespace":        cluster.PrefixedId(),
					"componentType":    string(cloudv1.ComponentTiKV),
					// the base name of each PVC, e.g. "tikv-raftlog", the suffix will be the Pod name
					// then the full PVC name will be "tikv-raftlog-db-tikv-0", "tikv-raftlog-db-tikv-1", ...
					"baseName":         string(opv1alpha1.GetStorageVolumeName(args.AdditionalStorage.Name, opv1alpha1.TiKVMemberType)),
					"storageClassName": storageClassName,
					"sizeGi":           sizeGiStr,
				},
			},
		},
		{
			// as we need to add annotation to the ASTS to pause it
			// so we need to pause the TidbCluster first to avoid overwriting the ASTS
			Name: "Pause TidbCluster reconciliation",
			Uses: consts.WfActionPauseReconciliationAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace": cluster.Namespace,
					"name":      cluster.Name,
					"type":      "tidbcluster",
				},
			},
		},
	}

	// create a copy-raft action for each TiKV pod
	// the ordinal in the pod names may not be match the replicas index
	// as we may have "delete-slots" for the ASTS, so we need to get the current real TiKV pods name.
	// the real pod names are sorted & saved in the outputs of the "Create new PVCs for TiKV" step
	for i := range cluster.Spec.Cluster.TiKV.Replicas {
		steps = append(steps, cloudv1.WorkflowStep{
			Name: fmt.Sprintf("Copy raft data to new disk %d", i),
			Uses: "CopyRaftDataAction",
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterNamespace": cluster.Namespace,
					"clusterName":      cluster.Name,
					"namespace":        cluster.PrefixedId(),
					"asts":             genAstsName(cloudv1.ComponentTiKV), // always "db-tikv"
					// all of these names should have been joined with ","
					"podNames": "$steps.Create new PVCs for TiKV.outputs.podNames",
					// limited by the ASTS controller, we need to copy the raft from 0 to N
					// ref: https://github.com/pingcap/advanced-statefulset/blob/v0.8.0/pkg/controller/statefulset/stateful_set_control.go#L504
					"podIndex": strconv.Itoa(i), // "cannot deep copy int"
				},
			},
		})
	}

	// then resume the TidbCluster reconciliation
	steps = append(steps, cloudv1.WorkflowStep{
		Name: "Resume TidbCluster reconciliation",
		Uses: consts.WfActionResumeReconciliationAction,
		Inputs: cloudv1.InputsWrapper{
			Object: map[string]interface{}{
				"namespace": cluster.Namespace,
				"name":      cluster.Name,
				"type":      "tidbcluster",
			},
		},
	})

	wf.Spec.Steps = steps

	return nil
}

func validateMigrateTiKVToDualDiskWithActiveGKEVAC(args *apiv1.Operation_MigrateTiKVToDualDisk, cluster *cloudv1.Cluster) error {
	if cluster.GetProviderType() != cloudv1.ProviderTypeGKE {
		return nil
	}
	additionalStorage := args.GetAdditionalStorage()
	if additionalStorage.GetType() == consts.GCPVolumeTypeHyperdiskBalanced {
		return nil
	}
	if cluster.Status.DB == nil || cluster.Status.DB.TiKV == nil || cluster.Status.DB.TiKV.Volume == nil {
		if cluster.FeatureEnabled(consts.ClusterFeatureUseVolumeAttributesClass) {
			return status.Error(codes.Unavailable, "GKE TiKV volume status is not available")
		}
		return nil
	}
	volumeStatus := cluster.Status.DB.TiKV.Volume
	if volumeStatus.CurrentVolumeAttributesClass == "" && volumeStatus.ModifiedVolumeAttributesClass == "" {
		return nil
	}
	return status.Errorf(
		codes.FailedPrecondition,
		"cannot add TiKV storage type %q while the GKE TiKV VolumeAttributesClass is active; use %s",
		additionalStorage.GetType(),
		consts.GCPVolumeTypeHyperdiskBalanced,
	)
}

func (s *Server) buildFixTiKVIllegalBoundaryWF(wf *cloudv1.Workflow, args *apiv1.Operation_FixTiKVIllegalBoundary, cluster *cloudv1.Cluster) error {

	steps := cloudv1.WorkflowSteps{
		{
			Name: "Check if tiflash pods are running or stop fixing",
			Uses: consts.WfActionCheckPodStatusAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterName":   cluster.Name,
					"namespace":     cluster.Namespace,
					"dryRun":        args.DryRun,
					"componentType": string(cloudv1.ComponentTiFlash),
					"expectState":   string(corev1.PodRunning),
				},
			},
		},
		{
			Name: "Check regions invalid TiFlash key",
			Uses: "CheckTFRegionKeysAction",
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
				},
			},
		},
		{
			Name: "Merge regions invalid TiFlash key",
			Uses: "MergeTFRegionKeys",
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterName": cluster.Name,
					"namespace":   cluster.Namespace,
					"dryRun":      args.DryRun,
					"invalidKeys": "$steps.Check regions invalid TiFlash key.outputs.invalid_keys",
				},
			},
		},
		{
			Name: "Exec tiFlash learner check and remove",
			Uses: "RepairInvalidLearnerAction",
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterName": cluster.Name,
					"namespace":   cluster.Namespace,
					"dryRun":      args.DryRun,
					"invalidKeys": "$steps.Check regions invalid TiFlash key.outputs.invalid_keys",
				},
			},
		},
	}

	wf.Spec.Steps = steps

	return nil
}

func (s *Server) buildEnablePPv2ForClusterWF(wf *cloudv1.Workflow, args *apiv1.Operation_EnableProxyProtocolForCluster, cluster *cloudv1.Cluster) error {
	steps := cloudv1.WorkflowSteps{
		{
			Name: "Mutate PPv2 annotations to reconcile TiDB",
			Uses: consts.WorkflowActionMutateAnnotations,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"name":      cluster.Name,
					"namespace": cluster.Namespace,
					"type":      "cluster",
					"annotations": map[string]interface{}{
						cluster.FormatFeature(consts.ClusterFeatureUsePPv2TiDBComponents): "true",
					},
				},
			},
		},
		{
			Name: "Ensure TiDB annotations synced to TC",
			Uses: consts.WorkflowActionEnsureAnnotateToTC,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterName": cluster.Name,
					"namespace":   cluster.Namespace,
					"annotations": map[string]interface{}{
						cluster.FormatFeature(consts.ClusterFeatureUsePPv2TiDBComponents): "true",
					},
				},
			},
		},
		{
			Name: "Restart all tidb pods to reload configs",
			Uses: consts.WfActionRestartComponentAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"targetName":  string(cloudv1.ComponentTiDB),
				}},
		},
		{
			Name: "Wait pod to be Running",
			Uses: consts.WfActionWaitTargetReadyAction,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"namespace":   cluster.Namespace,
					"clusterName": cluster.Name,
					"targetType":  "component",
					"targetName":  string(cloudv1.ComponentTiDB),
					"phase":       string(corev1.PodRunning),
					// ensure the pod is recreated by RestartPodAction step
					"ensureCreatedAfter": "$steps.Restart all tidb pods to reload configs.status.beginTime",
				}},
		},
		{
			Name: "Mutate PPv2 attributes to reconcile tiproxy",
			Uses: consts.WorkflowActionMutateAnnotations,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"name":      cluster.Name,
					"namespace": cluster.Namespace,
					"type":      "cluster",
					"annotations": map[string]interface{}{
						cluster.FormatFeature(consts.ClusterFeatureUsePPv2TiProxy): "true",
					},
				},
			},
		},
		{
			Name: "Ensure tiproxy annotations synced to TC",
			Uses: consts.WorkflowActionEnsureAnnotateToTC,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterName": cluster.Name,
					"namespace":   cluster.Namespace,
					"annotations": map[string]interface{}{
						cluster.FormatFeature(consts.ClusterFeatureUsePPv2TiProxy): "true",
					},
				},
			},
		},
		{
			Name: "Mutate PPv2 attributes to reconcile private NLB",
			Uses: consts.WorkflowActionMutateAnnotations,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"name":      cluster.Name,
					"namespace": cluster.Namespace,
					"type":      "cluster",
					"annotations": map[string]interface{}{
						cluster.FormatFeature(consts.ClusterFeatureUsePPv2PrivateNLB): "true",
					},
				},
			},
		},
		{
			Name: "Ensure NLB annotations synced to TC",
			Uses: consts.WorkflowActionEnsureAnnotateToTC,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"clusterName": cluster.Name,
					"namespace":   cluster.Namespace,
					"annotations": map[string]interface{}{
						cluster.FormatFeature(consts.ClusterFeatureUsePPv2PrivateNLB): "true",
					},
				},
			},
		},
	}

	wf.Spec.Steps = steps

	return nil
}

func (s *Server) buildModifyLoadBalanceLogsWF(wf *cloudv1.Workflow, args *apiv1.Operation_EnableLoadBalanceLogs, cluster *cloudv1.Cluster) error {
	if err := s.canModifyLBLoggings(cluster, args); err != nil {
		return status.Errorf(codes.InvalidArgument, "failed modify LB logs: %s", err)
	}

	optionalFields := args.GetOptionalFields()
	optionalFieldsInterface := make([]interface{}, len(optionalFields))
	for i, field := range optionalFields {
		optionalFieldsInterface[i] = field
	}
	inputs := make(map[string]interface{})
	inputs["namespace"] = cluster.Namespace
	inputs["clusterName"] = cluster.Name
	inputs["enabled"] = args.GetEnabled()
	if args.GetEnabled() {
		inputs["metadataFields"] = optionalFieldsInterface
		inputs["metadataMode"] = args.GetOptionalMode()
		inputs["sampleRate"] = args.GetSampleRate()
	}

	steps := cloudv1.WorkflowSteps{
		{
			Name: "Modify Load Balance logs",
			Uses: consts.WfActionModifyLoadBalanceLogsAction,
			Inputs: cloudv1.InputsWrapper{
				Object: inputs,
			},
		},
	}

	wf.Spec.Steps = steps

	return nil
}

func (s *Server) buildModifyVPCFlowLogsWF(wf *cloudv1.Workflow, args *apiv1.Operation_EnableVPCFlowLogs, cluster *cloudv1.Cluster) error {
	if err := s.canModifyVPCFlowLogs(cluster, args); err != nil {
		return status.Errorf(codes.InvalidArgument, "failed modify VPC FlowLogs: %s", err)
	}

	optionalFields := args.GetMetadataFields()
	optionalFieldsInterface := make([]interface{}, len(optionalFields))
	for i, field := range optionalFields {
		optionalFieldsInterface[i] = field
	}

	inputs := make(map[string]interface{})
	inputs["namespace"] = cluster.Namespace
	inputs["clusterName"] = cluster.Name
	inputs["enabled"] = args.GetEnabled()

	if args.GetEnabled() {
		inputs["metadataFields"] = optionalFieldsInterface
		inputs["metadataMode"] = args.GetMetadataMode()
		inputs["aggregationInterval"] = args.GetAggregationInterval()
		inputs["sampleRate"] = args.GetSampleRate()
	}
	steps := cloudv1.WorkflowSteps{
		{
			Name: "Modify VPC Flow logs",
			Uses: consts.WfActionModifyVPCFlowLogsAction,
			Inputs: cloudv1.InputsWrapper{
				Object: inputs,
			},
		},
	}

	wf.Spec.Steps = steps

	return nil
}

func (s *Server) buildOfflineAZWF(wf *cloudv1.Workflow, args *apiv1.Operation_OfflineAZ, cluster *cloudv1.Cluster) error {
	if args == nil || args.GetAvailabilityZone() == "" {
		return status.Errorf(codes.InvalidArgument, "availability_zone is required")
	}
	if err := validateGroupIDs(cluster, args.GetGroupIds()); err != nil {
		return status.Errorf(codes.InvalidArgument, "%v", err)
	}

	inputs := map[string]interface{}{
		"namespace":        cluster.Namespace,
		"clusterName":      cluster.Name,
		"availabilityZone": args.GetAvailabilityZone(),
	}
	if args.GetGracefulTimeout() != nil {
		inputs["gracefulTimeout"] = args.GetGracefulTimeout().AsDuration().String()
	}
	if ids := args.GetGroupIds(); len(ids) > 0 {
		inputs["groupIds"] = stringSliceToInterfaceSlice(ids)
	}

	wf.Spec.Steps = cloudv1.WorkflowSteps{
		{
			Name: "Offline AZ",
			Uses: consts.WfActionOfflineAZAction,
			Inputs: cloudv1.InputsWrapper{
				Object: inputs,
			},
		},
	}
	return nil
}

func (s *Server) buildRecoverAZWF(wf *cloudv1.Workflow, args *apiv1.Operation_RecoverAZ, cluster *cloudv1.Cluster) error {
	if args == nil || args.GetAvailabilityZone() == "" {
		return status.Errorf(codes.InvalidArgument, "availability_zone is required")
	}
	if err := validateGroupIDs(cluster, args.GetGroupIds()); err != nil {
		return status.Errorf(codes.InvalidArgument, "%v", err)
	}

	inputs := map[string]interface{}{
		"namespace":        cluster.Namespace,
		"clusterName":      cluster.Name,
		"availabilityZone": args.GetAvailabilityZone(),
	}
	if ids := args.GetGroupIds(); len(ids) > 0 {
		inputs["groupIds"] = stringSliceToInterfaceSlice(ids)
	}

	wf.Spec.Steps = cloudv1.WorkflowSteps{
		{
			Name: "Recover AZ",
			Uses: consts.WfActionRecoverAZAction,
			Inputs: cloudv1.InputsWrapper{
				Object: inputs,
			},
		},
	}
	return nil
}

const defaultZonalShiftWaitDuration = "30m"

func (s *Server) canZonalShift(cluster *cloudv1.Cluster) error {
	if cluster.Spec.Kubernetes.Provider != cloudv1.ProviderTypeEKS {
		return fmt.Errorf("unsupported provider type: %s", cluster.Spec.Kubernetes.Provider)
	}
	return nil
}

func (s *Server) buildStartZonalShiftWF(wf *cloudv1.Workflow, args *apiv1.Operation_StartZonalShift, cluster *cloudv1.Cluster) error {
	if err := s.canZonalShift(cluster); err != nil {
		return status.Errorf(codes.InvalidArgument, "failed start zonal shift: %s", err)
	}
	if args.GetAvailabilityZone() == "" {
		return status.Errorf(codes.InvalidArgument, "availability_zone is required")
	}
	if args.GetExpiresIn() == "" {
		return status.Errorf(codes.InvalidArgument, "expires_in is required")
	}
	if _, err := time.ParseDuration(args.GetExpiresIn()); err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid expires_in %q: %v", args.GetExpiresIn(), err)
	}
	if err := validateGroupIDs(cluster, args.GetGroupIds()); err != nil {
		return status.Errorf(codes.InvalidArgument, "%v", err)
	}

	inputs := map[string]interface{}{
		"namespace":        cluster.Namespace,
		"clusterName":      cluster.Name,
		"availabilityZone": args.GetAvailabilityZone(),
		"expiresIn":        args.GetExpiresIn(),
	}
	if ids := args.GetGroupIds(); len(ids) > 0 {
		// []string is not JSON-deepcopyable for Workflow InputsWrapper.
		inputs["groupIds"] = stringSliceToInterfaceSlice(ids)
	}

	wf.Spec.Steps = cloudv1.WorkflowSteps{
		{
			Name: "Start zonal shift",
			Uses: consts.WfActionStartZonalShiftAction,
			Inputs: cloudv1.InputsWrapper{
				Object: inputs,
			},
		},
		{
			Name: "Mark cluster AZ fault active",
			Uses: consts.WorkflowActionMutateAnnotations,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"name":      cluster.Name,
					"namespace": cluster.Namespace,
					"type":      "cluster",
					"annotations": map[string]interface{}{
						consts.AnnoKeyAZFaultActive: args.GetAvailabilityZone(),
					},
				},
			},
		},
	}
	return nil
}

func (s *Server) buildUpdateZonalShiftWF(wf *cloudv1.Workflow, args *apiv1.Operation_UpdateZonalShift, cluster *cloudv1.Cluster) error {
	if err := s.canZonalShift(cluster); err != nil {
		return status.Errorf(codes.InvalidArgument, "failed update zonal shift: %s", err)
	}
	if args.GetExpiresIn() == "" {
		return status.Errorf(codes.InvalidArgument, "expires_in is required")
	}
	if _, err := time.ParseDuration(args.GetExpiresIn()); err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid expires_in %q: %v", args.GetExpiresIn(), err)
	}
	if err := validateGroupIDs(cluster, args.GetGroupIds()); err != nil {
		return status.Errorf(codes.InvalidArgument, "%v", err)
	}

	inputs := map[string]interface{}{
		"namespace":   cluster.Namespace,
		"clusterName": cluster.Name,
		"expiresIn":   args.GetExpiresIn(),
	}
	if ids := args.GetGroupIds(); len(ids) > 0 {
		inputs["groupIds"] = stringSliceToInterfaceSlice(ids)
	}

	wf.Spec.Steps = cloudv1.WorkflowSteps{
		{
			Name: "Update zonal shift",
			Uses: consts.WfActionUpdateZonalShiftAction,
			Inputs: cloudv1.InputsWrapper{
				Object: inputs,
			},
		},
	}
	return nil
}

func (s *Server) buildStopZonalShiftWF(wf *cloudv1.Workflow, args *apiv1.Operation_StopZonalShift, cluster *cloudv1.Cluster) error {
	if args == nil {
		return status.Errorf(codes.InvalidArgument, "stop_zonal_shift args are required for STOP_ZONAL_SHIFT operation")
	}
	if err := s.canZonalShift(cluster); err != nil {
		return status.Errorf(codes.InvalidArgument, "failed stop zonal shift: %s", err)
	}
	if err := validateGroupIDs(cluster, args.GetGroupIds()); err != nil {
		return status.Errorf(codes.InvalidArgument, "%v", err)
	}
	waitDuration := args.GetWaitDuration()
	if waitDuration == "" {
		waitDuration = defaultZonalShiftWaitDuration
	}
	if _, err := time.ParseDuration(waitDuration); err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid wait_duration %q: %v", waitDuration, err)
	}

	commonInputs := map[string]interface{}{
		"namespace":   cluster.Namespace,
		"clusterName": cluster.Name,
	}
	if ids := args.GetGroupIds(); len(ids) > 0 {
		commonInputs["groupIds"] = stringSliceToInterfaceSlice(ids)
	}
	wf.Spec.Steps = cloudv1.WorkflowSteps{
		{
			Name: "Cancel zonal shift",
			Uses: consts.WfActionCancelZonalShiftAction,
			Inputs: cloudv1.InputsWrapper{
				Object: commonInputs,
			},
		},
		{
			Name: "Wait for traffic recovery",
			Uses: consts.WorkflowActionSleep,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"duration": waitDuration,
				},
			},
		},
		{
			Name: "Cleanup zonal shift",
			Uses: consts.WfActionCleanupZonalShiftAction,
			Inputs: cloudv1.InputsWrapper{
				Object: commonInputs,
			},
		},
		{
			Name: "Clear cluster AZ fault active",
			Uses: consts.WorkflowActionMutateAnnotations,
			Inputs: cloudv1.InputsWrapper{
				Object: map[string]interface{}{
					"name":      cluster.Name,
					"namespace": cluster.Namespace,
					"type":      "cluster",
					"annotations": map[string]interface{}{
						consts.AnnoKeyAZFaultActive: "-",
					},
				},
			},
		},
	}
	return nil
}

// stringSliceToInterfaceSlice converts []string for Workflow InputsWrapper.
// runtime.DeepCopyJSON only accepts JSON types; []string panics with
// "cannot deep copy []string".
func stringSliceToInterfaceSlice(in []string) []interface{} {
	out := make([]interface{}, len(in))
	for i, v := range in {
		out[i] = v
	}
	return out
}

// validateGroupIDs ensures every requested group ID belongs to the cluster.
// An empty list is valid and means default group + all ExtraGroups.
func validateGroupIDs(cluster *cloudv1.Cluster, groupIDs []string) error {
	if len(groupIDs) == 0 {
		return nil
	}
	known := map[string]struct{}{}
	if defaultID := cluster.GetDefaultGroupID(); defaultID != "" {
		known[defaultID] = struct{}{}
	}
	for i := range cluster.Spec.ExtraGroups {
		if cluster.Spec.ExtraGroups[i].Deleted() {
			continue
		}
		known[cluster.Spec.ExtraGroups[i].Name] = struct{}{}
	}
	var unknown []string
	for _, id := range groupIDs {
		if id == "" {
			return fmt.Errorf("group_ids must not contain empty string")
		}
		if _, ok := known[id]; !ok {
			unknown = append(unknown, id)
		}
	}
	if len(unknown) > 0 {
		return fmt.Errorf("unknown group_ids: %s", strings.Join(unknown, ","))
	}
	return nil
}

// canUpgradeNodes checks the corresponding k8s's control plane's version.
func (s *Server) canUpgradeNodes(ctx context.Context, cluster *cloudv1.Cluster, args *apiv1.Operation_ShutDownUpgradeNodes) (bool, error) {
	var cpVersion *semver.Version
	desiredVersion, err := semver.NewVersion(args.DesiredVersion)
	if err != nil {
		return false, fmt.Errorf("parse desired version %s: %w", args.DesiredVersion, err)
	}

	switch cluster.Spec.Kubernetes.Provider {
	case cloudv1.ProviderTypeEKS:
		var eks v1beta1.Cluster
		if err := s.KubeClient.Get(ctx, client.ObjectKey{Name: cluster.Spec.Kubernetes.Name}, &eks); err != nil {
			return false, fmt.Errorf("get eks cluster: %w", err)
		}
		cpVersion, err = semver.NewVersion(eks.Status.AtProvider.Version)
		if err != nil {
			return false, fmt.Errorf("parse eks version %s: %w", eks.Status.AtProvider.Version, err)
		}

	default:
		return false, fmt.Errorf("unsupported provider type: %s", cluster.Spec.Kubernetes.Provider)
	}

	return cpVersion != nil && (cpVersion.Equal(desiredVersion) || cpVersion.GreaterThan(desiredVersion)), nil
}

func (s *Server) GetOperation(ctx context.Context, req *apiv1.GetOperationRequest) (*apiv1.Operation, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId), zap.String("id", req.OperationId))
	logger.Info("get operation")
	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	var wf cloudv1.Workflow
	if err := s.KubeClient.Get(ctx, client.ObjectKey{Name: req.OperationId, Namespace: cluster.Namespace}, &wf); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Workflow %s not found", req.OperationId)
		}
		return nil, fmt.Errorf("get workflow: %w", err)
	}

	return cvt.Workflow2PB(&wf), nil
}

// canEnablePPv2 checks the parameters.
func (s *Server) canEnablePPv2(ctx context.Context, cluster *cloudv1.Cluster, args *apiv1.Operation_EnableProxyProtocolForCluster) error {

	currentVer := cluster.Status.DB.Version

	// Must >= v6.5.1 since which the TiDB.fallbackable is first supported.
	// https://docs.pingcap.com/zh/tidb/stable/tidb-configuration-file/#fallbackable-%E4%BB%8E-v651-%E7%89%88%E6%9C%AC%E5%BC%80%E5%A7%8B%E5%BC%95%E5%85%A5
	passed, err := cmpver.Compare(currentVer, cmpver.GreaterOrEqual, minTiDBFallbackableSupportedVersion)
	if !passed || err != nil {
		if err != nil {
			return fmt.Errorf("unsupported version '%s'(must %s%s), got err: %w", currentVer, cmpver.GreaterOrEqual, minTiDBFallbackableSupportedVersion, err)
		}
		return fmt.Errorf("unsupported version '%s'(must %s%s)", currentVer, cmpver.GreaterOrEqual, minTiDBFallbackableSupportedVersion)
	}

	// So far PPv2 has been supported only under AWS.
	if cluster.Spec.Kubernetes.Provider != cloudv1.ProviderTypeEKS {
		return fmt.Errorf("unsupported provider type: %s", cluster.Spec.Kubernetes.Provider)
	}

	return nil
}

// canModifyLBLoggings checks the parameters.
func (s *Server) canModifyLBLoggings(cluster *cloudv1.Cluster, args *apiv1.Operation_EnableLoadBalanceLogs) error {
	// So far has been supported only under GCP.
	if cluster.Spec.Kubernetes.Provider != cloudv1.ProviderTypeGKE {
		return fmt.Errorf("unsupported provider type: %s", cluster.Spec.Kubernetes.Provider)
	}

	return nil
}

// canModifyVPCFlowLogs checks the parameters.
func (s *Server) canModifyVPCFlowLogs(cluster *cloudv1.Cluster, args *apiv1.Operation_EnableVPCFlowLogs) error {
	// So far has been supported only under GCP.
	if cluster.Spec.Kubernetes.Provider != cloudv1.ProviderTypeGKE {
		return fmt.Errorf("unsupported provider type: %s", cluster.Spec.Kubernetes.Provider)
	}

	return nil
}

func (s *Server) UpgradeNodes(ctx context.Context, req *apiv1.UpgradeNodesRequest) (*apiv1.UpgradeNodesResponse, error) {
	logger := tilog.WithContextL(ctx).With(zap.Any("req", req))

	cluster, err := server.GetClusterCR(ctx, s.clusterReader, s.KubeClient, req.ClusterId)
	if err != nil {
		// If cluster is not found, the step should be marked as completed
		if status.Code(err) == codes.NotFound {
			logger.Warn("skip not found cluster", zap.String("cluster", req.ClusterId), zap.Error(err))
			return &apiv1.UpgradeNodesResponse{EndTime: timestamppb.New(s.clock.Now())}, nil
		}
		logger.Error("failed to get cluster", zap.Error(err))
		return &apiv1.UpgradeNodesResponse{}, err
	}

	var uo cloudv1.UpgradeOperation
	// There may be multiple tidb clusters in a k8s cluster,
	// so add the cluster name to distinguish the UpgradeOperation CRs.
	uoName := fmt.Sprintf("%s-%s", req.IdempotencyKey, req.ClusterId)
	if err = s.KubeClient.Get(ctx, client.ObjectKey{Name: uoName, Namespace: cluster.Namespace}, &uo); err != nil && !apierrors.IsNotFound(err) {
		logger.Error("failed to get UpgradeOperation CR", zap.Error(err))
		return &apiv1.UpgradeNodesResponse{}, err
	}

	if apierrors.IsNotFound(err) {
		// Create a UpgradeOperation CR
		uo = cloudv1.UpgradeOperation{
			ObjectMeta: metav1.ObjectMeta{
				Name:      uoName,
				Namespace: cluster.Namespace,
				// predicate match label selector
				Labels: map[string]string{
					cloudv1.ProviderTypeLabel: cluster.GetProviderType(),
					cloudv1.RegionLabel:       cluster.GetRegion(),
				},
			},
			Spec: cloudv1.UpgradeOperationSpec{
				// If the schedule time is already passed, the validating webhook will reject the request.
				Schedule: s.clock.Now().Add(scheduleDelay).Format(cloudv1.UpgradeOperationScheduleLayout),
				Paused:   false,
				Steps:    cloudv1.Steps{makeUpgradeStep(req)},
			},
		}
		if err := s.KubeClient.Create(ctx, &uo); err != nil {
			logger.Error("failed to create UpgradeOperation CR", zap.Error(err))
			return &apiv1.UpgradeNodesResponse{}, err
		}
		logger.Info("created UpgradeOperation CR", zap.String("name", uo.Name), zap.String("namespace", uo.Namespace))
	} else {
		if uo.Spec.Paused != req.Paused {
			// Update the UpgradeOperation CR
			uo.Spec.Paused = req.Paused
			if err := s.KubeClient.Update(ctx, &uo); err != nil {
				logger.Error("failed to update UpgradeOperation CR", zap.Error(err))
				return &apiv1.UpgradeNodesResponse{}, err
			}
			logger.Info("updated UpgradeOperation CR", zap.String("name", uo.Name), zap.String("namespace", uo.Namespace), zap.Bool("paused", uo.Spec.Paused))
		}
	}

	return cvt.UpgradeOperation2PB(&uo), nil
}

func makeUpgradeStep(req *apiv1.UpgradeNodesRequest) cloudv1.Step {
	var target cloudv1.UpgradeOperationTarget
	switch req.NodeType {
	case apiv1.UpgradeNodesRequest_DM:
		target = cloudv1.DMNodes
	case apiv1.UpgradeNodesRequest_CDC:
		target = cloudv1.CDCNodes
	case apiv1.UpgradeNodesRequest_TIDB:
		target = cloudv1.TiDBNodes
	}

	options := maps.Clone(req.Options)
	// The controller owns this option and persists only a resolved scheduling
	// decision. Never accept it from an UpgradeNodes caller.
	delete(options, gkemindomains.UpgradeDecisionOption)
	if len(options) == 0 {
		options = nil
	}

	return cloudv1.Step{
		Target:  target,
		Version: req.DesiredVersion,
		Cluster: req.ClusterId,
		Options: options,
	}
}

func genOperationWFName(op *apiv1.Operation) string {
	tp := strings.ReplaceAll(string(cvt.OperationType2WFType(op.Type)), "_", "-")
	return fmt.Sprintf("%s-%s", tp, op.IdempotencyKey)
}

func genAstsName(componentType cloudv1.ComponentType) string {
	return fmt.Sprintf("%s-%s", consts.ClusterPrefix, componentType)
}