package tioperator

import (
	"context"
	"fmt"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	meta "github.com/pingcap/tidb-operator/api/v2/meta/v1alpha1"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
)

const (
	dmMasterDataVolume  = "dm-master"
	dmWorkerRelayVolume = "dm-worker-relay"
	dmMasterContainer   = "dm-master"
	dmWorkerContainer   = "dm-worker"
	dmMasterStorageGi   = 50
	dmWorkerRelaySizeGi = 10
)

func GetDMClusterV2(ctx context.Context, remoteCli client.Client, dmCluster *entity.DMCluster) (*entity.DMCluster, error) {
	if dmCluster == nil {
		return nil, fmt.Errorf("dm cluster is nil")
	}

	ns, err := resolveDMClusterV2ReadNamespace(ctx, remoteCli, dmCluster)
	if err != nil {
		return nil, err
	}

	masterGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return nil, err
	}
	workerGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return nil, err
	}

	masterGroup, err := getDMGroupV2(ctx, remoteCli, ns, masterGroupName)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "dm group not found: %v", err)
		}
		return nil, fmt.Errorf("failed to get dm group: %w", err)
	}
	workerGroup, err := getDMWorkerGroupV2(ctx, remoteCli, ns, workerGroupName)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "dm worker group not found: %v", err)
		}
		return nil, fmt.Errorf("failed to get dm worker group: %w", err)
	}

	return &entity.DMCluster{
		Id:              dmCluster.GetId(),
		SharedPoolId:    dmCluster.GetSharedPoolId(),
		DataplaneName:   dmCluster.GetDataplaneName(),
		MasterNodeQuota: cvtDMGroupToNodeQuota(masterGroup),
		WorkerNodeQuota: cvtDMWorkerGroupToNodeQuota(workerGroup),
		Phase:           determineDMClusterPhaseV2(masterGroup, workerGroup),
	}, nil
}

func ApplyDMClusterV2(ctx context.Context, remoteCli client.Client, dmCluster *entity.DMCluster, dmVersion, dmImage string, updateImage bool) error {
	if dmCluster == nil {
		return fmt.Errorf("dm cluster is nil")
	}
	if updateImage && dmVersion == "" {
		return fmt.Errorf("dm version is empty")
	}
	if updateImage && dmImage == "" {
		return fmt.Errorf("dm image is empty")
	}

	namespace, clusterRefName, features, err := resolveDMClusterV2Target(ctx, remoteCli, dmCluster)
	if err != nil {
		return err
	}

	if err := applyDMGroupV2(ctx, remoteCli, dmCluster, namespace, clusterRefName, features, dmVersion, dmImage, updateImage); err != nil {
		return err
	}
	if err := applyDMWorkerGroupV2(ctx, remoteCli, dmCluster, namespace, clusterRefName, features, dmVersion, dmImage, updateImage); err != nil {
		return err
	}
	return nil
}

func DeleteDMClusterV2(ctx context.Context, remoteCli client.Client, dmCluster *entity.DMCluster) error {
	if dmCluster == nil {
		return fmt.Errorf("dm cluster is nil")
	}

	ns := dmClusterV2Namespace(dmCluster)
	if ns == "" {
		return nil
	}

	workerGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return err
	}
	masterGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return err
	}

	for _, obj := range []client.Object{
		&v1alpha1.DMWorkerGroup{ObjectMeta: objectMeta(ns, workerGroupName)},
		&v1alpha1.DMGroup{ObjectMeta: objectMeta(ns, masterGroupName)},
	} {
		if err := remoteCli.Delete(ctx, obj); err != nil && !apierrors.IsNotFound(err) {
			return err
		}
	}
	return nil
}

func getDMGroupV2(ctx context.Context, remoteCli client.Client, namespace, name string) (*v1alpha1.DMGroup, error) {
	var group v1alpha1.DMGroup
	if err := remoteCli.Get(ctx, types.NamespacedName{Name: name, Namespace: namespace}, &group); err != nil {
		return nil, err
	}
	return &group, nil
}

func getDMWorkerGroupV2(ctx context.Context, remoteCli client.Client, namespace, name string) (*v1alpha1.DMWorkerGroup, error) {
	var group v1alpha1.DMWorkerGroup
	if err := remoteCli.Get(ctx, types.NamespacedName{Name: name, Namespace: namespace}, &group); err != nil {
		return nil, err
	}
	return &group, nil
}

func applyDMGroupV2(ctx context.Context, remoteCli client.Client, dmCluster *entity.DMCluster, namespace, clusterRefName string, features []meta.Feature, dmVersion, dmImage string, updateImage bool) error {
	var err error
	dmVersion, dmImage, err = resolveDMGroupV2ImageAndVersion(ctx, remoteCli, dmCluster, namespace, dmVersion, dmImage, updateImage)
	if err != nil {
		return err
	}

	group, err := buildDMGroupV2(dmCluster, namespace, clusterRefName, features, dmVersion, dmImage)
	if err != nil {
		return err
	}

	current := &v1alpha1.DMGroup{ObjectMeta: objectMeta(group.Namespace, group.Name)}
	_, err = controllerutil.CreateOrUpdate(ctx, remoteCli, current, func() error {
		preserveDMGroupImageAndVersion(&group.Spec.Template.Spec, &current.Spec.Template.Spec, updateImage)
		current.Spec = group.Spec
		current.Labels = mergeStringMap(current.Labels, group.Labels)
		return nil
	})
	return err
}

func applyDMWorkerGroupV2(ctx context.Context, remoteCli client.Client, dmCluster *entity.DMCluster, namespace, clusterRefName string, features []meta.Feature, dmVersion, dmImage string, updateImage bool) error {
	var err error
	dmVersion, dmImage, err = resolveDMWorkerGroupV2ImageAndVersion(ctx, remoteCli, dmCluster, namespace, dmVersion, dmImage, updateImage)
	if err != nil {
		return err
	}

	group, err := buildDMWorkerGroupV2(dmCluster, namespace, clusterRefName, features, dmVersion, dmImage)
	if err != nil {
		return err
	}

	current := &v1alpha1.DMWorkerGroup{ObjectMeta: objectMeta(group.Namespace, group.Name)}
	_, err = controllerutil.CreateOrUpdate(ctx, remoteCli, current, func() error {
		preserveDMWorkerGroupImageAndVersion(&group.Spec.Template.Spec, &current.Spec.Template.Spec, updateImage)
		current.Spec = group.Spec
		current.Labels = mergeStringMap(current.Labels, group.Labels)
		return nil
	})
	return err
}

func resolveDMGroupV2ImageAndVersion(ctx context.Context, remoteCli client.Client, dmCluster *entity.DMCluster, namespace, dmVersion, dmImage string, updateImage bool) (string, string, error) {
	if updateImage {
		return dmVersion, dmImage, nil
	}

	groupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return "", "", err
	}
	current, err := getDMGroupV2(ctx, remoteCli, namespace, groupName)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return candidateDMImageAndVersion(dmVersion, dmImage, namespace, groupName)
		}
		return "", "", fmt.Errorf("failed to get existing dm group %s/%s: %w", namespace, groupName, err)
	}
	return currentDMImageAndVersion(current.Spec.Template.Spec.Version, current.Spec.Template.Spec.Image, dmVersion, dmImage, namespace, groupName)
}

func resolveDMWorkerGroupV2ImageAndVersion(ctx context.Context, remoteCli client.Client, dmCluster *entity.DMCluster, namespace, dmVersion, dmImage string, updateImage bool) (string, string, error) {
	if updateImage {
		return dmVersion, dmImage, nil
	}

	groupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return "", "", err
	}
	current, err := getDMWorkerGroupV2(ctx, remoteCli, namespace, groupName)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return candidateDMImageAndVersion(dmVersion, dmImage, namespace, groupName)
		}
		return "", "", fmt.Errorf("failed to get existing dm worker group %s/%s: %w", namespace, groupName, err)
	}
	return currentDMImageAndVersion(current.Spec.Template.Spec.Version, current.Spec.Template.Spec.Image, dmVersion, dmImage, namespace, groupName)
}

func currentDMImageAndVersion(version string, image *string, fallbackVersion, fallbackImage, namespace, name string) (string, string, error) {
	if version == "" {
		version = fallbackVersion
	}
	if image == nil || *image == "" {
		return candidateDMImageAndVersion(version, fallbackImage, namespace, name)
	}
	return candidateDMImageAndVersion(version, *image, namespace, name)
}

func candidateDMImageAndVersion(version, image, namespace, name string) (string, string, error) {
	if version == "" {
		return "", "", fmt.Errorf("dm component %s/%s version is empty", namespace, name)
	}
	if image == "" {
		return "", "", fmt.Errorf("dm component %s/%s image is empty", namespace, name)
	}
	return version, image, nil
}

func preserveDMGroupImageAndVersion(target, current *v1alpha1.DMTemplateSpec, updateImage bool) {
	if updateImage || target == nil || current == nil {
		return
	}
	if current.Image != nil {
		target.Image = current.Image
	}
	if current.Version != "" {
		target.Version = current.Version
	}
}

func preserveDMWorkerGroupImageAndVersion(target, current *v1alpha1.DMWorkerTemplateSpec, updateImage bool) {
	if updateImage || target == nil || current == nil {
		return
	}
	if current.Image != nil {
		target.Image = current.Image
	}
	if current.Version != "" {
		target.Version = current.Version
	}
}

func buildDMGroupV2(dmCluster *entity.DMCluster, namespace, clusterRefName string, features []meta.Feature, dmVersion, dmImage string) (*v1alpha1.DMGroup, error) {
	templateSpec, labels, err := buildDMTemplateSpec(dmCluster.GetMasterNodeQuota(), dmImage, dmVersion, true)
	if err != nil {
		return nil, err
	}

	replicas := int32(dmCluster.GetMasterNodeQuota().GetNodeCount())
	masterGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return nil, err
	}
	objMeta := objectMeta(
		namespace,
		masterGroupName,
	)
	objMeta.Labels = labels
	return &v1alpha1.DMGroup{
		TypeMeta:   typeMeta("DMGroup"),
		ObjectMeta: objMeta,
		Spec: v1alpha1.DMGroupSpec{
			Cluster:  v1alpha1.ClusterReference{Name: clusterRefName},
			Features: features,
			Replicas: &replicas,
			Template: v1alpha1.DMTemplate{
				Spec: *templateSpec,
			},
		},
	}, nil
}

func buildDMWorkerGroupV2(dmCluster *entity.DMCluster, namespace, clusterRefName string, features []meta.Feature, dmVersion, dmImage string) (*v1alpha1.DMWorkerGroup, error) {
	templateSpec, labels, err := buildDMWorkerTemplateSpec(dmCluster.GetWorkerNodeQuota(), dmImage, dmVersion)
	if err != nil {
		return nil, err
	}

	replicas := int32(dmCluster.GetWorkerNodeQuota().GetNodeCount())
	workerGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return nil, err
	}
	masterGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return nil, err
	}
	objMeta := objectMeta(
		namespace,
		workerGroupName,
	)
	objMeta.Labels = labels
	return &v1alpha1.DMWorkerGroup{
		TypeMeta:   typeMeta("DMWorkerGroup"),
		ObjectMeta: objMeta,
		Spec: v1alpha1.DMWorkerGroupSpec{
			Cluster:    v1alpha1.ClusterReference{Name: clusterRefName},
			DMGroupRef: v1alpha1.DMGroupReference{Name: masterGroupName},
			Features:   features,
			Replicas:   &replicas,
			Template: v1alpha1.DMWorkerTemplate{
				Spec: *templateSpec,
			},
		},
	}, nil
}

func buildDMTemplateSpec(nodeQuota *entity.NodeQuota, image, version string, includeDataVolume bool) (*v1alpha1.DMTemplateSpec, map[string]string, error) {
	resources, labels, nodeSelector, err := buildDMResourcesAndNodeSelector(nodeQuota)
	if err != nil {
		return nil, nil, err
	}

	spec := &v1alpha1.DMTemplateSpec{
		Version: version,
		Image:   &image,
		// tidb-operator overlays managed dm-master config fields, including
		// openapi=true. Keep Config empty so those fields stay operator-owned.
		// DM v2 uses the overlay container resources as the effective source of
		// requests and limits so the read path can round-trip the applied values.
		Overlay: buildDMOverlay(nodeSelector, dmMasterContainer, resources),
	}
	if includeDataVolume {
		spec.DataVolume = buildDMVolume(dmMasterDataVolume, v1alpha1.VolumeMountTypeDMData, dmMasterStorageGi)
	}
	return spec, labels, nil
}

func buildDMWorkerTemplateSpec(nodeQuota *entity.NodeQuota, image, version string) (*v1alpha1.DMWorkerTemplateSpec, map[string]string, error) {
	resources, labels, nodeSelector, err := buildDMResourcesAndNodeSelector(nodeQuota)
	if err != nil {
		return nil, nil, err
	}

	return &v1alpha1.DMWorkerTemplateSpec{
		Version:     version,
		Image:       &image,
		RelayVolume: buildDMVolumePtr(dmWorkerRelayVolume, v1alpha1.VolumeMountTypeDMWorkerRelay, dmWorkerRelaySizeGi),
		Overlay:     buildDMOverlay(nodeSelector, dmWorkerContainer, resources),
	}, labels, nil
}

func buildDMResourcesAndNodeSelector(nodeQuota *entity.NodeQuota) (corev1.ResourceRequirements, map[string]string, map[string]string, error) {
	labels := map[string]string{}
	nodeSelector := map[string]string{}

	resourceRequirements, err := setNodeSpec(nodeQuota.GetNodeSpec(), labels, nodeSelector, apiv1.ComponentType_DATA_MIGRATION)
	if err != nil {
		return corev1.ResourceRequirements{}, nil, nil, err
	}
	if nodeQuota.GetNodeSpec().GetNodeVersion() != "" {
		nodeSelector[schemav1.K8sVersionLabelKey] = nodeQuota.GetNodeSpec().GetNodeVersion()
	}

	return resourceRequirements, labels, nodeSelector, nil
}

func buildDMOverlay(nodeSelector map[string]string, containerName string, resources corev1.ResourceRequirements) *v1alpha1.Overlay {
	return &v1alpha1.Overlay{
		Pod: &v1alpha1.PodOverlay{
			Spec: &corev1.PodSpec{
				NodeSelector: nodeSelector,
				Tolerations: []corev1.Toleration{
					{
						Key:      "dedicated",
						Operator: corev1.TolerationOpEqual,
						Value:    "worker",
						Effect:   corev1.TaintEffectNoSchedule,
					},
					{
						Key:               "node.kubernetes.io/not-ready",
						Operator:          corev1.TolerationOpExists,
						Effect:            corev1.TaintEffectNoExecute,
						TolerationSeconds: int64Ptr(300),
					},
					{
						Key:               "node.kubernetes.io/unreachable",
						Operator:          corev1.TolerationOpExists,
						Effect:            corev1.TaintEffectNoExecute,
						TolerationSeconds: int64Ptr(300),
					},
				},
				Containers: []corev1.Container{
					{
						Name:      containerName,
						Resources: resources,
					},
				},
			},
		},
	}
}

func resolveDMClusterV2Target(ctx context.Context, remoteCli client.Client, dmCluster *entity.DMCluster) (string, string, []meta.Feature, error) {
	if dmCluster.GetSharedPoolId() == "" {
		return "", "", nil, fmt.Errorf("dm cluster shared pool id is empty")
	}

	namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())
	var cluster v1alpha1.Cluster
	key := types.NamespacedName{Name: clusterCRName, Namespace: namespace}
	if err := remoteCli.Get(ctx, key, &cluster); err != nil {
		if apierrors.IsNotFound(err) {
			return "", "", nil, status.Errorf(codes.NotFound, "shared pool cluster %s not found in namespace %s", clusterCRName, namespace)
		}
		return "", "", nil, fmt.Errorf("failed to get shared pool cluster %s in namespace %s: %w", clusterCRName, namespace, err)
	}

	return namespace, cluster.Name, featureGatesToFeatures(cluster.Spec.FeatureGates), nil
}

func featureGatesToFeatures(gates []meta.FeatureGate) []meta.Feature {
	if len(gates) == 0 {
		return nil
	}
	features := make([]meta.Feature, 0, len(gates))
	for _, gate := range gates {
		features = append(features, gate.Name)
	}
	return features
}

func resolveDMClusterV2ReadNamespace(ctx context.Context, remoteCli client.Client, dmCluster *entity.DMCluster) (string, error) {
	namespace := dmClusterV2Namespace(dmCluster)
	if namespace == "" {
		return "", fmt.Errorf("failed to resolve dm cluster namespace")
	}

	masterGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return "", err
	}
	workerGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return "", err
	}

	masterErr := groupExistsV2(ctx, remoteCli, namespace, masterGroupName, &v1alpha1.DMGroup{})
	if masterErr != nil && !apierrors.IsNotFound(masterErr) {
		return "", fmt.Errorf("failed to get dm group: %w", masterErr)
	}

	workerErr := groupExistsV2(ctx, remoteCli, namespace, workerGroupName, &v1alpha1.DMWorkerGroup{})
	if workerErr != nil && !apierrors.IsNotFound(workerErr) {
		return "", fmt.Errorf("failed to get dm worker group: %w", workerErr)
	}

	return namespace, nil
}

func groupExistsV2(ctx context.Context, remoteCli client.Client, namespace, name string, obj client.Object) error {
	return remoteCli.Get(ctx, types.NamespacedName{Name: name, Namespace: namespace}, obj)
}

func dmClusterV2Namespace(dmCluster *entity.DMCluster) string {
	if dmCluster == nil || dmCluster.GetSharedPoolId() == "" {
		return ""
	}
	return entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())
}

func buildDMVolume(name string, mountType v1alpha1.VolumeMountType, sizeGi int64) v1alpha1.Volume {
	size := resource.MustParse(fmt.Sprintf("%dGi", sizeGi))
	storageClassName := entity.CloudSSD
	return v1alpha1.Volume{
		Name:             name,
		Mounts:           []v1alpha1.VolumeMount{{Type: mountType}},
		Storage:          size,
		StorageClassName: &storageClassName,
	}
}

func buildDMVolumePtr(name string, mountType v1alpha1.VolumeMountType, sizeGi int64) *v1alpha1.Volume {
	volume := buildDMVolume(name, mountType, sizeGi)
	return &volume
}

func cvtDMGroupToNodeQuota(group *v1alpha1.DMGroup) *entity.NodeQuota {
	if group == nil {
		return nil
	}
	return cvtDMResourcesToNodeQuota(
		group.Labels,
		group.Spec.Replicas,
		// DM v2 reads back the effective resources from the named overlay
		// container because that is where this implementation writes them.
		dmContainerLimits(group.Spec.Template.Spec.Overlay, dmMasterContainer),
		group.Spec.Template.Spec.Overlay,
	)
}

func cvtDMWorkerGroupToNodeQuota(group *v1alpha1.DMWorkerGroup) *entity.NodeQuota {
	if group == nil {
		return nil
	}
	return cvtDMResourcesToNodeQuota(
		group.Labels,
		group.Spec.Replicas,
		// DM v2 reads back the effective resources from the named overlay
		// container because that is where this implementation writes them.
		dmContainerLimits(group.Spec.Template.Spec.Overlay, dmWorkerContainer),
		group.Spec.Template.Spec.Overlay,
	)
}

func dmContainerLimits(overlay *v1alpha1.Overlay, containerName string) corev1.ResourceList {
	if overlay == nil || overlay.Pod == nil || overlay.Pod.Spec == nil {
		return nil
	}

	for i := range overlay.Pod.Spec.Containers {
		if overlay.Pod.Spec.Containers[i].Name == containerName {
			return overlay.Pod.Spec.Containers[i].Resources.Limits
		}
	}

	return nil
}

func cvtDMResourcesToNodeQuota(labels map[string]string, replicas *int32, resourceList corev1.ResourceList, overlay *v1alpha1.Overlay) *entity.NodeQuota {
	var nodeSelector map[string]string
	if overlay != nil && overlay.Pod != nil && overlay.Pod.Spec != nil {
		nodeSelector = overlay.Pod.Spec.NodeSelector
	}

	nodeSpec := &entity.NodeSpec{}
	if spec := labels[schemav1.LabelNodeSpecTypeKey]; spec != "" {
		if cpu, memory, err := parseNodeSpec(spec); err == nil {
			nodeSpec.Cpu = uint32(cpu)
			nodeSpec.MemoryGi = uint32(memory)
		}
	}
	if nodeSelector != nil {
		nodeSpec.NodeVersion = nodeSelector[schemav1.K8sVersionLabelKey]
	}

	if cpu, ok := resourceList[corev1.ResourceCPU]; ok {
		nodeSpec.AvailableCpu = uint32(cpu.AsApproximateFloat64())
	}
	if memory, ok := resourceList[corev1.ResourceMemory]; ok {
		nodeSpec.AvailableMemoryMi = uint32(memory.AsApproximateFloat64() / 1024 / 1024)
	}

	return &entity.NodeQuota{
		// DM v1/v2 read paths only have the current desired replicas. We mirror the
		// legacy API shape by treating the current size as a fixed-size cluster,
		// rather than claiming we know independent autoscaling bounds here.
		NodeCount:    derefInt32(replicas),
		MinNodeCount: derefInt32(replicas),
		MaxNodeCount: derefInt32(replicas),
		NodeSpec:     nodeSpec,
	}
}

func determineDMClusterPhaseV2(master *v1alpha1.DMGroup, worker *v1alpha1.DMWorkerGroup) apiv1.DMCluster_Phase {
	masterDesired := derefInt32(master.Spec.Replicas)
	workerDesired := derefInt32(worker.Spec.Replicas)

	masterReady := isReady(master.Status.Conditions) &&
		isUpToDate(master.Generation, master.Status.CommonStatus) &&
		master.Status.ReadyReplicas == int32(masterDesired) &&
		master.Status.UpdatedReplicas == int32(masterDesired)
	workerReady := isReady(worker.Status.Conditions) &&
		isUpToDate(worker.Generation, worker.Status.CommonStatus) &&
		worker.Status.ReadyReplicas == int32(workerDesired) &&
		worker.Status.UpdatedReplicas == int32(workerDesired)

	if masterReady && workerReady {
		return apiv1.DMCluster_AVAILABLE
	}

	if master.Status.ObservedGeneration == 0 || worker.Status.ObservedGeneration == 0 {
		return apiv1.DMCluster_CREATING
	}

	if !isUpToDate(master.Generation, master.Status.CommonStatus) || !isUpToDate(worker.Generation, worker.Status.CommonStatus) {
		return apiv1.DMCluster_MODIFYING
	}

	if master.Status.ReadyReplicas != int32(masterDesired) || worker.Status.ReadyReplicas != int32(workerDesired) {
		return apiv1.DMCluster_MODIFYING
	}

	return apiv1.DMCluster_UNAVAILABLE
}

func typeMeta(kind string) metav1.TypeMeta {
	return metav1.TypeMeta{
		APIVersion: "core.pingcap.com/v1alpha1",
		Kind:       kind,
	}
}

func objectMeta(namespace, name string) metav1.ObjectMeta {
	return metav1.ObjectMeta{
		Namespace: namespace,
		Name:      name,
	}
}

func mergeStringMap(current, desired map[string]string) map[string]string {
	if len(current) == 0 && len(desired) == 0 {
		return nil
	}

	merged := make(map[string]string, len(current)+len(desired))
	for k, v := range current {
		merged[k] = v
	}
	for k, v := range desired {
		merged[k] = v
	}
	return merged
}

func int64Ptr(v int64) *int64 {
	return &v
}

func derefInt32(v *int32) uint32 {
	if v == nil {
		return 0
	}
	return uint32(*v)
}

func parseNodeSpec(spec string) (uint64, uint64, error) {
	cpu, memory, err := utils.ParseNodeSpec(spec)
	return uint64(cpu), uint64(memory), err
}