package common

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	opv2 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	opv2meta "github.com/pingcap/tidb-operator/api/v2/meta/v1alpha1"
	"github.com/tidbcloud/config-service/sdk/api"
	corev1 "k8s.io/api/core/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	"github.com/tidbcloud/eks-provider/controllers/cluster/task/common/builder"
	"github.com/tidbcloud/eks-provider/controllers/cluster/task/common/feature"
	"github.com/tidbcloud/eks-provider/controllers/utils"
	"github.com/tidbcloud/eks-provider/utils/tidb-operator/tidbcluster"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	sdk "github.com/tidbcloud/infra-provider/pkg/gitops-sdk"
	gitopsConsts "github.com/tidbcloud/infra-provider/pkg/gitops-sdk/consts"
	infrak8s "github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/metrics"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const (
	backupManagerServiceAccountName = "tidb-backup-manager"
)

type TiDBClusterV2 struct {
	*task.BaseTask

	pdg    *opv2.PDGroup
	dbg    *opv2.TiDBGroup
	kvg    *opv2.TiKVGroup
	flashg *opv2.TiFlashGroup

	k8sVersion         string
	helperImage        string
	configRevisionName string
	ref                opv2.ClusterReference
	featureGates       feature.Gates
	componentContexts  map[cloudv1.ComponentType]*ComponentContext
	schedulePolicies   []opv2.SchedulePolicy
}

var _ deploy.Task = &TiDBClusterV2{}

func NewTiDBClusterV2(ctx *task.Context) *TiDBClusterV2 {
	helperImage := config.GetBusyboxImage()
	if img := util.GetAnnotation(ctx.Cluster, gitopsConsts.OverrideHelperImageKey); img != "" {
		helperImage = img
	}

	var crName string
	if ctx.Cluster.Spec.ConfigRevisionRef != nil {
		crName = ctx.Cluster.Spec.ConfigRevisionRef.Name
	}
	return &TiDBClusterV2{
		BaseTask:           &task.BaseTask{Context: ctx},
		k8sVersion:         ctx.GetNodeVersion(true),
		helperImage:        helperImage,
		schedulePolicies:   buildSchedulePolicies(ctx.Zones()),
		configRevisionName: crName,
	}
}

func (t *TiDBClusterV2) Name() string {
	return "TiDBClusterV2"
}

func (t *TiDBClusterV2) buildComponentContext(clusterConfig *api.DescribeClusterConfigOutput) error {
	components := t.Cluster.GetComponents()
	t.componentContexts = make(map[cloudv1.ComponentType]*ComponentContext, len(components))
	allowedCN := util.FnvHash(t.Cluster.PrefixedId())
	for _, comp := range components {
		compType := comp.GetComponentType()
		// Now tidbgroup is built by builder pkg
		// TODO(liubo02): refactor all components to use builder
		if compType == cloudv1.ComponentTiDB {
			continue
		}
		cfg, ok := getComponentConfig(clusterConfig, api.ComponentName(compType))
		if !ok {
			return fmt.Errorf("no %s configuration provided", compType)
		}
		resourceSpec, err := t.ConfigProvider.GetResourceSpec(t.Provider, comp)
		if err != nil {
			return fmt.Errorf("failed to get %s resources spec: %w", compType, err)
		}

		sc, ok := comp.GetStorageClassName()
		if !ok {
			sc = resourceSpec.StorageClass
		}
		sc = t.CMEKStorageClassMapper.Name(sc)
		var vac *string
		if sc != "" && t.featureGates.Enabled(feature.UseVolumeAttributesClass) {
			vac = ptr.To(sc)
		}

		version := clusterConfig.Version
		if cfg.Image.Tag != nil {
			version = *cfg.Image.Tag
		}

		spec := t.Cluster.GetComponent(compType)
		var storageSize resource.Quantity
		ss, ok := spec.GetStorageSize()
		if ok {
			storageSize = *ss
		}

		cc := &ComponentContext{
			Config:  cfg,
			Version: version,
			Resources: ComponentResources{
				Requests: opv2.ResourceRequirements{
					CPU:    resourceSpec.Resources.Requests.Cpu(),
					Memory: resourceSpec.Resources.Requests.Memory(),
				},
				Limits: opv2.ResourceRequirements{
					// no CPU in limits now
					Memory: resourceSpec.Resources.Limits.Memory(),
				},
			},
			allowedCN:                 allowedCN,
			StorageClassName:          sc,
			VolumeAttributesClassName: vac,
			Replicas:                  ptr.To(int32(spec.GetReplicas())),
			InstanceType:              spec.GetInstanceType(),
			StorageSize:               storageSize,
		}
		cc.NodeSelector, cc.Tolerations = t.buildNodeSelectorAndTolerations(compType, cc.InstanceType)

		if compType == cloudv1.ComponentTiFlash {
			secondCfg, ok2 := getComponentConfig(clusterConfig, api.ComponentTiFlashLearner)
			if !ok2 {
				return fmt.Errorf("no %s configuration provided", compType)
			}
			cc.SecondConfig = secondCfg
		}
		t.componentContexts[comp.GetComponentType()] = cc
	}
	return nil
}

func (t *TiDBClusterV2) Create(ctx context.Context) (res deploy.Result) {
	t.BeginAt = ptr.To(time.Now())

	conf, err := sdk.GetCurrentClusterConfiguration(t.Local, t.Cluster.Spec.ConfigRevisionRef)
	if err != nil {
		return deploy.Error(fmt.Errorf("failed to get current cluster configuration: %w", err))
	}
	clusterConfig, err := sdk.GenerateConfigOutput(t.Cluster, conf)
	if err != nil {
		return deploy.Error(fmt.Errorf("failed to generate cluster configuration: %w", err))
	}

	t.featureGates = feature.Init(t.Cluster, t.Config.Provider, t.k8sVersion, clusterConfig.Version)

	if err := t.buildComponentContext(clusterConfig); err != nil {
		return deploy.Error(fmt.Errorf("failed to build component context: %w", err))
	}

	cli := t.EnvProvider.GetClient()
	tc := t.expectedCluster()
	if err := infrak8s.Apply(ctx, cli, tc); err != nil {
		return deploy.Error(fmt.Errorf("failed to apply tidb cluster CR: %w", err))
	}
	t.TidbClusterV2 = tc
	getter := &tidbcluster.V2Getter{Cli: cli, Cluster: tc}
	t.TCGetter = getter
	t.ref = opv2.ClusterReference{Name: tc.Name}

	updatableComps := t.getUpdatableComponents()

	if err = t.createOrUpdateSubResourcesForBR(ctx, cli); err != nil {
		return deploy.Error(fmt.Errorf("failed to create/update sub resources for BR: %w", err))
	}
	if err = t.createOrUpdateSubResourcesForPD(ctx, cli); err != nil {
		return deploy.Error(fmt.Errorf("failed to create/update sub resources for PD: %w", err))
	}
	pdg, err := t.expectedPDGroup()
	if err != nil {
		return deploy.Error(fmt.Errorf("failed to get expected pd group: %w", err))
	}
	if updatableComps[cloudv1.ComponentPD] {
		if err = infrak8s.Apply(ctx, cli, pdg); err != nil {
			return deploy.Error(fmt.Errorf("failed to apply pd group CR: %w", err))
		}
	} else {
		t.Logger.Info("delay pd group apply as it is waiting for new instance type nodes to be ready")
		// We need to get the current PDGroup CR for status check (e.g. whether all instances are ready)
		if err := cli.Get(ctx, client.ObjectKey{
			Name:      consts.ClusterPrefix,
			Namespace: t.Cluster.PrefixedId(),
		}, pdg); err != nil {
			return deploy.Error(fmt.Errorf("failed to get pd group CR: %w", err))
		}
	}
	t.pdg = pdg
	getter.PDGroup = t.pdg

	dbg, err := t.expectedTiDBGroup(ctx, clusterConfig)
	if err != nil {
		return deploy.Error(fmt.Errorf("failed to get expected tidb group: %w", err))
	}
	if updatableComps[cloudv1.ComponentTiDB] {
		if err = infrak8s.Apply(ctx, cli, dbg); err != nil {
			return deploy.Error(fmt.Errorf("failed to apply tidb group CR: %w", err))
		}
	} else {
		t.Logger.Info("delay tidb group apply as it is waiting for new instance type nodes to be ready")
		if err := cli.Get(ctx, client.ObjectKey{
			Name:      consts.ClusterPrefix,
			Namespace: t.Cluster.PrefixedId(),
		}, dbg); err != nil {
			return deploy.Error(fmt.Errorf("failed to get tidb group CR: %w", err))
		}
	}
	t.dbg = dbg
	getter.TiDBGroup = t.dbg

	tikvGroup, err := t.expectedTiKVGroup()
	if err != nil {
		return deploy.Error(fmt.Errorf("failed to get expected tikv group: %w", err))
	}
	if updatableComps[cloudv1.ComponentTiKV] {
		if err = infrak8s.Apply(ctx, cli, tikvGroup); err != nil {
			return deploy.Error(fmt.Errorf("failed to apply tikv group CR: %w", err))
		}
	} else {
		t.Logger.Info("delay tikv group apply as it is waiting for new instance type nodes to be ready")
		if err := cli.Get(ctx, client.ObjectKey{
			Name:      consts.ClusterPrefix,
			Namespace: t.Cluster.PrefixedId(),
		}, tikvGroup); err != nil {
			return deploy.Error(fmt.Errorf("failed to get tikv group CR: %w", err))
		}
	}
	t.kvg = tikvGroup
	getter.TiKVGroup = t.kvg

	if t.Cluster.Spec.Cluster.TiFlash.Replicas > 0 {
		tiflashGroup, err := t.expectedTiFlashGroup()
		if err != nil {
			return deploy.Error(fmt.Errorf("failed to get expected tiflash group: %w", err))
		}
		if updatableComps[cloudv1.ComponentTiFlash] {
			if err = infrak8s.Apply(ctx, cli, tiflashGroup); err != nil {
				return deploy.Error(fmt.Errorf("failed to apply tiflash group CR: %w", err))
			}
		} else {
			t.Logger.Info("delay tiflash group apply as it is waiting for new instance type nodes to be ready")
			if err := cli.Get(ctx, client.ObjectKey{
				Name:      consts.ClusterPrefix,
				Namespace: t.Cluster.PrefixedId(),
			}, tiflashGroup); err != nil {
				return deploy.Error(fmt.Errorf("failed to get tiflash group CR: %w", err))
			}
		}
		t.flashg = tiflashGroup
		getter.TiFlashGroup = t.flashg
	} else {
		if err = infrak8s.DeleteObjectCompleted(ctx, cli, &opv2.TiFlashGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name:      consts.ClusterPrefix,
				Namespace: t.Cluster.PrefixedId(),
			},
		}); err != nil {
			return deploy.Error(fmt.Errorf("failed to delete tiflash group CR: %w", err))
		}
	}

	defer func() {
		observeRes := t.observeTidbCluster(ctx, cli, clusterConfig.Version)
		if res.Error() != nil {
			if observeRes.Error() != nil {
				t.Logger.Info("observe tidb cluster when creating/updating tc failed", "observeResult", observeRes)
			}
			return
		}
		t.Context.TidbClusterLastUpdateTime = ptr.To(time.Now())
		res = observeRes // return the observe result
	}()

	return deploy.Done()
}

// getUpdatableComponents returns the components that can be updated.
// If a component is scaling up/down, it should delay the update until the nodes are ready.
func (t *TiDBClusterV2) getUpdatableComponents() map[cloudv1.ComponentType]bool {
	updatable := make(map[cloudv1.ComponentType]bool)
	for _, comp := range NonInstanceTypeScalingComponents(t.Cluster) {
		updatable[comp] = true
	}

	compAnno := t.Cluster.Annotations[consts.ClusterUpdateAnnotationKeyInstanceType]
	if compAnno != "" {
		// new nodes are ready
		comps := strings.Split(compAnno, consts.AnnoInstanceTypeSeparator)
		for _, c := range comps {
			updatable[cloudv1.ComponentType(c)] = true
		}
	}
	return updatable
}

func (t *TiDBClusterV2) observeTidbCluster(ctx context.Context, remoteCli client.Client, version string) deploy.Result {
	if t.CommonK8sClusterStatus != nil {
		t.CommonK8sClusterStatus.Kubernetes = &cloudv1.KubernetesStatus{Namespace: t.Namer.GetTiDBNamespace()}
	}

	if t.Cluster.Status.Phase != cloudv1.ClusterPhasePending &&
		t.Cluster.Status.Phase != cloudv1.ClusterPhaseCreating {
		r := &TiDBClusterStatusResolverV2{
			Client:      remoteCli,
			Provider:    t.Provider,
			Target:      t.Cluster,
			TiDBCluster: t.TidbClusterV2,
			Mapper:      t.CMEKStorageClassMapper,

			Gates:   t.featureGates,
			Version: version,
			Log:     t.Logger,
			Namer:   t.Namer,
		}
		if err := r.Resolve(ctx); err != nil {
			return deploy.Error(err)
		}
	}
	return deploy.Done()
}

func (t *TiDBClusterV2) PostCreate(ctx context.Context) deploy.Result {
	if t.TidbClusterV2 == nil {
		return deploy.Error(fmt.Errorf("tidb cluster is not ready yet"))
	}

	if t.tikvTiflashScaleIn(ctx, t.EnvProvider.GetClient()) {
		// record PD low space ratio for reset after scaled in
		if pdLowSpaceRatio := t.getPDLowSpaceRatio(ctx); pdLowSpaceRatio != nil {
			if t.TidbClusterV2.Annotations == nil {
				t.TidbClusterV2.Annotations = make(map[string]string)
			}
			t.TidbClusterV2.Annotations[annoKeyPDLowSpaceRatio] = fmt.Sprintf("%f", *pdLowSpaceRatio)
			t.Logger.Info("record PD low-space-ratio before scale in TiKV/TiFlash", "ratio", *pdLowSpaceRatio)
		}
	}

	u, err := t.GetClusterProxyBaseURL(ctx)
	if err != nil {
		return deploy.Error(err)
	}
	t.Cluster.Status.Proxy = &cloudv1.ProxyStatus{Endpoint: u}

	// set cluster gateway base url
	gatewayEp, err := t.GetClusterGatewayBaseURL(ctx)
	if err != nil {
		return deploy.Error(err)
	}
	t.Cluster.Status.Gateway = &cloudv1.GatewayStatus{Endpoint: gatewayEp}

	// set/reset PD low-space-ratio
	if err = t.setPDLowSpaceRatio(ctx); err != nil {
		return deploy.Error(err)
	}

	// always requeue to sync if cluster is scaling or upgrading,
	// or PD is not Normal as PD's status is not considered in `Phase`
	phase := t.Cluster.Status.Phase
	if phase == cloudv1.ClusterPhaseScaling ||
		phase == cloudv1.ClusterPhaseUpgrading ||
		phase == cloudv1.ClusterPhaseSuspend ||
		phase == cloudv1.ClusterPhaseUnavailable ||
		(phase == cloudv1.ClusterPhaseRunning && (t.Cluster.Status.DB == nil ||
			t.Cluster.Status.DB.SecondaryStatus == cloudv1.DBSecondaryStatusUnhealthy ||
			t.Cluster.Status.DB.PD == nil || t.Cluster.Status.DB.PD.Status != cloudv1.ComponentNormal)) {
		t.logModifyProgress()
		return deploy.Error(fmt.Errorf("cluster is %s", phase)).WithRequeuAfter(clusterRequeueAfter)
	}

	ready, err := t.isClusterReady(ctx)
	if !ready {
		return deploy.Error(fmt.Errorf("tidb cluster is not ready yet: %v", err)).WithRequeuAfter(clusterRequeueAfter)
	}

	return deploy.Done()
}

func buildTiDBClusterFeatureGates(gates feature.Gates) []opv2meta.FeatureGate {
	fg := make([]opv2meta.FeatureGate, 0)
	if gates.Enabled(feature.UseVolumeAttributesClass) {
		fg = append(fg, opv2meta.FeatureGate{Name: opv2meta.VolumeAttributesClass})
	}
	if gates.Enabled(feature.DisableV2PDReadinessProbe) {
		fg = append(fg, opv2meta.FeatureGate{Name: opv2meta.DisablePDDefaultReadinessProbe})
	}
	if gates.Enabled(feature.UsePDReadyAPI) {
		fg = append(fg, opv2meta.FeatureGate{Name: opv2meta.UsePDReadyAPI})
	}
	fg = append(fg, opv2meta.FeatureGate{Name: opv2meta.FeatureModification})
	return fg
}

func (t *TiDBClusterV2) expectedCluster() *opv2.Cluster {
	tc := &opv2.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: t.Cluster.PrefixedId(),
			Name:      consts.ClusterPrefix,
			Labels: map[string]string{
				cloudv1.LabelUsedFor: cloudv1.UsedForCluster,
			},
			Annotations: map[string]string{
				cloudv1.ConfigRevisionIdKey: t.configRevisionName,
			},
		},
		Spec: opv2.ClusterSpec{
			TLSCluster:    &opv2.TLSCluster{Enabled: true},
			UpgradePolicy: opv2.UpgradePolicyDefault,
			SuspendAction: &opv2.SuspendAction{SuspendCompute: t.suspendCompute()},
			BootstrapSQL: &corev1.LocalObjectReference{
				Name: t.Namer.BootstrapSQLConfigMapName(),
			},
			FeatureGates: buildTiDBClusterFeatureGates(t.featureGates),
		},
	}
	reconcileNormalizeLastAppliedConfigAnnotation(tc, config.FeatureEnabled(t.Config.FeatureGates, config.NormalizeLastAppliedConfiguration, t.Cluster.GetTenantID(), t.Cluster.GetProjectID()))
	return tc
}

func (t *TiDBClusterV2) expectedPDGroup() (*opv2.PDGroup, error) {
	ctx, ok := t.componentContexts[cloudv1.ComponentPD]
	if !ok {
		return nil, fmt.Errorf("pd component context not found")
	}
	pdCfg, err := ctx.computePDConfig(t.Cluster)
	if err != nil {
		return nil, fmt.Errorf("failed to compute pd config: %w", err)
	}

	pdGroup := &opv2.PDGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      consts.ClusterPrefix,
			Namespace: t.Cluster.PrefixedId(),
			Annotations: map[string]string{
				cloudv1.ConfigRevisionIdKey: t.configRevisionName,
			},
		},
		Spec: opv2.PDGroupSpec{
			Cluster:          t.ref,
			Replicas:         ctx.Replicas,
			SchedulePolicies: t.schedulePolicies,
			Template: opv2.PDTemplate{
				ObjectMeta: opv2.ObjectMeta{
					Annotations: map[string]string{
						// This annotation is still needed to determine whether the Pod
						// is running on a specific instance type during the scaling process.
						consts.TcInstanceTypeAnnotationKey: ctx.InstanceType,
					},
				},
				Spec: opv2.PDTemplateSpec{
					Version: ctx.Version,
					Image:   ptr.To(ctx.Config.Image.BaseImage),
					UpdateStrategy: opv2.UpdateStrategy{
						Config: opv2.ConfigUpdateStrategyHotReload,
					},
					Config: pdCfg,
					Volumes: []opv2.Volume{
						{
							Name:                      "data",
							Mounts:                    []opv2.VolumeMount{{Type: opv2.VolumeMountTypePDData}},
							Storage:                   ctx.StorageSize,
							StorageClassName:          ptr.To(ctx.StorageClassName),
							VolumeAttributesClassName: ctx.VolumeAttributesClassName,
						},
					},
					Overlay: &opv2.Overlay{
						PersistentVolumeClaims: []opv2.NamedPersistentVolumeClaimOverlay{
							{
								Name: "data",
								PersistentVolumeClaim: opv2.PersistentVolumeClaimOverlay{
									ObjectMeta: opv2.ObjectMeta{
										// All app.kubernetes.io labels are for compatibility with v1
										// It may be used in prometheus rules and svc selectors in v1
										Labels: map[string]string{
											"app.kubernetes.io/component":  "pd",
											"app.kubernetes.io/instance":   consts.ClusterPrefix,
											"app.kubernetes.io/managed-by": "tidb-operator",
											"app.kubernetes.io/name":       "tidb-cluster",
										},
									},
								},
							},
						},
						Pod: &opv2.PodOverlay{
							ObjectMeta: opv2.ObjectMeta{
								Annotations: map[string]string{
									consts.TcInstanceTypeAnnotationKey: ctx.InstanceType,
								},
								// All app.kubernetes.io labels are for compatibility with v1
								// It may be used in prometheus rules and svc selectors in v1
								Labels: map[string]string{
									"app.kubernetes.io/component":  "pd",
									"app.kubernetes.io/instance":   consts.ClusterPrefix,
									"app.kubernetes.io/managed-by": "tidb-operator",
									"app.kubernetes.io/name":       "tidb-cluster",
								},
							},
							Spec: &corev1.PodSpec{
								Containers: []corev1.Container{
									{
										Name: opv2.ContainerNamePD,
										Resources: corev1.ResourceRequirements{
											// different requests and limits in Dedicated
											Requests: corev1.ResourceList{
												corev1.ResourceCPU:    *ctx.Resources.Requests.CPU,
												corev1.ResourceMemory: *ctx.Resources.Requests.Memory,
											},
											Limits: corev1.ResourceList{
												corev1.ResourceMemory: *ctx.Resources.Limits.Memory,
											},
										},
										Env: []corev1.EnvVar{
											{
												Name:  "TZ",
												Value: "UTC",
											},
											{
												Name: "DASHBOARD_SESSION_SECRET",
												ValueFrom: &corev1.EnvVarSource{
													SecretKeyRef: &corev1.SecretKeySelector{
														LocalObjectReference: corev1.LocalObjectReference{
															Name: fmt.Sprintf("%s-dashboard-session-secret", consts.ClusterPrefix),
														},
														Key: "encryption_key",
													},
												},
											},
										},
									},
								},
								HostNetwork:       true,
								DNSPolicy:         corev1.DNSClusterFirstWithHostNet,
								PriorityClassName: consts.PriorityClasName,
								Affinity:          utils.Affinity(cloudv1.ComponentPD),
								Tolerations:       ctx.Tolerations,
								NodeSelector:      ctx.NodeSelector,
								// No need to overlay TopologySpreadConstraints for PD,
								// because it only has 3 replicas across 3 AZs.
							},
						},
					},
				},
			},
		},
	}
	return pdGroup, nil
}

// getAuditSidecarConfig returns the audit sidecar image and envs.
// It prefers SidecarImageV2 if set, otherwise falls back to SidecarImage.
// If the cluster has an AuditLogSidecarImageTag annotation, that tag is applied on top,
// and if the sidecar version supports dynamic log path, envs are returned.
func (t *TiDBClusterV2) getAuditSidecarConfig(ctx context.Context, groupName string) (image string, sidecarEnvs []corev1.EnvVar) {
	return resolveAuditSidecarConfig(ctx, t.Cluster, t.Config, groupName)
}

func (t *TiDBClusterV2) NewTiDBGroupConfig(
	ctx context.Context,
	gates feature.Gates,
	comp cloudv1.ClusterComponent,
	cfg *api.DescribeClusterConfigOutput,
) (*builder.TiDBGroupConfig, error) {
	n := t.Namer.WithComponent(string(cloudv1.ComponentTiDB))

	sidecarImage, sidecarEnvs := t.getAuditSidecarConfig(ctx, t.Cluster.GetDefaultGroupID())
	dbgcfg := &builder.TiDBGroupConfig{
		Name:                n.GetTiDBName(),
		Namespace:           n.GetTiDBNamespace(),
		ClusterRef:          n.GetDefaultTiDBName(),
		ClusterName:         n.GetClusterName(),
		AuthTokenSecretName: n.TidbAuthTokenJwksSecret(),
		ConfigRevisionName:  t.configRevisionName,

		// directly from cluster component
		InstanceType: comp.GetInstanceType(),
		Replicas:     int32(comp.GetReplicas()),
		Arch:         getInstanceArch(t.Provider, comp.GetInstanceType()),

		HelperImage: config.GetOverrideBusyboxImage(t.Cluster.Annotations),

		// from context
		Zones:      t.Zones(),
		K8sVersion: t.GetNodeVersion(true),

		// from cluster
		TenantID:  t.Cluster.GetTenantID(),
		ProjectID: t.Cluster.GetProjectID(),
		ClusterID: t.Cluster.GetClusterID(),
		Provider:  t.Config.Provider,

		AuditLogSidecarImage:   sidecarImage,
		AuditLogSidecarCommand: t.Config.ClusterController.Audit.SidecarCommand,
		AuditLogSidecarEnvs:    sidecarEnvs,
		PreStopSleepSeconds:    int32(t.Config.ClusterController.GetTiDBPreStopHookSleepSeconds()),
	}

	var compConfig *api.ComponentOutput
	for i, c := range cfg.Components {
		if c.Component == api.ComponentTiDB {
			compConfig = &cfg.Components[i]
			break
		}
	}

	// OriginalConfig
	dbgcfg.OriginalConfig = compConfig.Config.MP

	// Resources
	resourceSpec, err := t.ConfigProvider.GetResourceSpec(t.Provider, comp)
	if err != nil {
		return nil, fmt.Errorf("failed to get resources spec: %w", err)
	}
	dbgcfg.Resources = resourceSpec.Resources

	// StorageClassName
	sc, ok := comp.GetStorageClassName()
	if !ok {
		sc = resourceSpec.StorageClass
	}
	sc = t.CMEKStorageClassMapper.Name(sc)
	dbgcfg.StorageClassName = sc

	// StorageSize
	size, ok := comp.GetStorageSize()
	if ok {
		dbgcfg.StorageSize = size
	}

	// BaseImage and ImageTag
	dbgcfg.BaseImage = compConfig.Image.BaseImage
	dbgcfg.ImageTag = cfg.Version
	if compConfig.Image.Tag != nil {
		dbgcfg.ImageTag = *compConfig.Image.Tag
	}

	return dbgcfg, nil
}

func (t *TiDBClusterV2) expectedTiDBGroup(ctx context.Context, cfg *api.DescribeClusterConfigOutput) (*opv2.TiDBGroup, error) {
	dbgcfg, err := t.NewTiDBGroupConfig(ctx, t.featureGates, t.Cluster.GetComponent(cloudv1.ComponentTiDB), cfg)
	if err != nil {
		return nil, err
	}

	return builder.NewTiDBGroupBuilder().Build(t.featureGates, dbgcfg), nil
}

func (t *TiDBClusterV2) expectedTiKVGroup() (*opv2.TiKVGroup, error) {
	ctx, ok := t.componentContexts[cloudv1.ComponentTiKV]
	if !ok {
		return nil, fmt.Errorf("tikv component context not found")
	}
	cfg, err := ctx.computeTiKVConfig()
	if err != nil {
		return nil, fmt.Errorf("failed to compute tikv config: %w", err)
	}

	tikvGroup := &opv2.TiKVGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      consts.ClusterPrefix,
			Namespace: t.Cluster.PrefixedId(),
			Annotations: map[string]string{
				cloudv1.ConfigRevisionIdKey: t.configRevisionName,
			},
		},
		Spec: opv2.TiKVGroupSpec{
			Cluster:          t.ref,
			Replicas:         ctx.Replicas,
			SchedulePolicies: t.schedulePolicies,
			Template: opv2.TiKVTemplate{
				ObjectMeta: opv2.ObjectMeta{
					Annotations: map[string]string{
						consts.TcInstanceTypeAnnotationKey: ctx.InstanceType,
					},
				},
				Spec: opv2.TiKVTemplateSpec{
					Version: ctx.Version,
					UpdateStrategy: opv2.UpdateStrategy{
						Config: opv2.ConfigUpdateStrategyHotReload,
					},
					Image:  ptr.To(ctx.Config.Image.BaseImage),
					Config: cfg,
					Volumes: []opv2.Volume{
						{
							Name: "data",
							Mounts: []opv2.VolumeMount{
								{Type: opv2.VolumeMountTypeTiKVData},
							},
							Storage:                   ctx.StorageSize,
							StorageClassName:          ptr.To(ctx.StorageClassName),
							VolumeAttributesClassName: ctx.VolumeAttributesClassName,
						},
					},
					PreStop: &opv2.TiKVPreStop{
						Image: ptr.To(t.Config.ClusterController.TiKVPreStopImage),
					},
					Overlay: &opv2.Overlay{
						PersistentVolumeClaims: []opv2.NamedPersistentVolumeClaimOverlay{
							{
								Name: "data",
								PersistentVolumeClaim: opv2.PersistentVolumeClaimOverlay{
									ObjectMeta: opv2.ObjectMeta{
										// All app.kubernetes.io labels are for compatibility with v1
										// It may be used in prometheus rules and svc selectors in v1
										Labels: map[string]string{
											"app.kubernetes.io/component":  "tikv",
											"app.kubernetes.io/instance":   consts.ClusterPrefix,
											"app.kubernetes.io/managed-by": "tidb-operator",
											"app.kubernetes.io/name":       "tidb-cluster",
										},
									},
								},
							},
						},
						Pod: &opv2.PodOverlay{
							ObjectMeta: opv2.ObjectMeta{
								Annotations: map[string]string{
									consts.TcInstanceTypeAnnotationKey: ctx.InstanceType,
								},
								// All app.kubernetes.io labels are for compatibility with v1
								// It may be used in prometheus rules and svc selectors in v1
								Labels: map[string]string{
									"app.kubernetes.io/component":  "tikv",
									"app.kubernetes.io/instance":   consts.ClusterPrefix,
									"app.kubernetes.io/managed-by": "tidb-operator",
									"app.kubernetes.io/name":       "tidb-cluster",
								},
							},
							Spec: &corev1.PodSpec{
								Tolerations:               ctx.Tolerations,
								NodeSelector:              ctx.NodeSelector,
								HostNetwork:               true,
								DNSPolicy:                 corev1.DNSClusterFirstWithHostNet,
								PriorityClassName:         consts.PriorityClasName,
								Affinity:                  utils.Affinity(cloudv1.ComponentTiKV),
								TopologySpreadConstraints: t.buildTopologySpreadConstraints(),
								InitContainers: []corev1.Container{
									{
										Name:  "init",
										Image: t.helperImage,
										// Because this sysctl is unsafe, we have to set it through a privileged init container, rather than PodSecurityContext.
										Command:         []string{"sh", "-c", "sysctl", "-w", "net.core.somaxconn=32768"},
										SecurityContext: &corev1.SecurityContext{Privileged: ptr.To(true)},
										Resources: corev1.ResourceRequirements{
											// same resources as the main container
											Requests: corev1.ResourceList{
												corev1.ResourceCPU:    *ctx.Resources.Requests.CPU,
												corev1.ResourceMemory: *ctx.Resources.Requests.Memory,
											},
											Limits: corev1.ResourceList{
												corev1.ResourceMemory: *ctx.Resources.Limits.Memory,
											},
										},
									},
								},
								Containers: []corev1.Container{
									t.buildStoreContainer(opv2.ContainerNameTiKV, ctx.Resources),
								},
							},
						},
					},
				},
			},
		},
	}

	for _, vol := range t.Cluster.Spec.Cluster.TiKV.StorageVolumes {
		sc := t.CMEKStorageClassMapper.Name(vol.StorageClassName)
		var vacName *string
		if t.featureGates.Enabled(feature.UseVolumeAttributesClass) {
			vacName = ptr.To(sc)
		}
		tikvGroup.Spec.Template.Spec.Volumes = append(tikvGroup.Spec.Template.Spec.Volumes, opv2.Volume{
			Name:                      vol.Name,
			StorageClassName:          ptr.To(sc),
			VolumeAttributesClassName: vacName,
			Storage:                   vol.StorageSize,
			Mounts: []opv2.VolumeMount{
				{
					MountPath: vol.MountPath,
					Type:      opv2.VolumeMountType(vol.Name),
				},
			},
		})
	}

	return tikvGroup, nil
}

func (t *TiDBClusterV2) expectedTiFlashGroup() (*opv2.TiFlashGroup, error) {
	compCtx, ok := t.componentContexts[cloudv1.ComponentTiFlash]
	if !ok {
		return nil, fmt.Errorf("tiflash component context not found")
	}
	cfg, proxyCfg, err := compCtx.computeTiFlashConfig()
	if err != nil {
		return nil, fmt.Errorf("failed to compute tiflash config: %w", err)
	}

	tiflashGroup := &opv2.TiFlashGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      consts.ClusterPrefix,
			Namespace: t.Cluster.PrefixedId(),
			Annotations: map[string]string{
				cloudv1.ConfigRevisionIdKey: t.configRevisionName,
			},
		},
		Spec: opv2.TiFlashGroupSpec{
			Cluster:          t.ref,
			Replicas:         compCtx.Replicas,
			SchedulePolicies: t.schedulePolicies,
			Template: opv2.TiFlashTemplate{
				ObjectMeta: opv2.ObjectMeta{
					Annotations: map[string]string{
						consts.TcInstanceTypeAnnotationKey: compCtx.InstanceType,
					},
				},
				Spec: opv2.TiFlashTemplateSpec{
					Version:     compCtx.Version,
					Image:       ptr.To(compCtx.Config.Image.BaseImage),
					Config:      cfg,
					ProxyConfig: proxyCfg,
					UpdateStrategy: opv2.UpdateStrategy{
						Config: opv2.ConfigUpdateStrategyHotReload,
					},
					Volumes: []opv2.Volume{
						{
							Name:                      "data",
							Storage:                   compCtx.StorageSize,
							StorageClassName:          ptr.To(compCtx.StorageClassName),
							VolumeAttributesClassName: compCtx.VolumeAttributesClassName,
							Mounts: []opv2.VolumeMount{
								{
									MountPath: "/data0",
									Type:      opv2.VolumeMountTypeTiFlashData,
								},
							},
						},
					},
					LogTailer: &opv2.TiFlashLogTailer{
						Image: ptr.To(t.helperImage),
						Resources: opv2.ResourceRequirements{
							CPU:    resource.NewMilliQuantity(100, resource.DecimalSI),
							Memory: resource.NewQuantity(50*MB, resource.BinarySI),
						},
					},
					Overlay: &opv2.Overlay{
						PersistentVolumeClaims: []opv2.NamedPersistentVolumeClaimOverlay{
							{
								Name: "data",
								PersistentVolumeClaim: opv2.PersistentVolumeClaimOverlay{
									ObjectMeta: opv2.ObjectMeta{
										// All app.kubernetes.io labels are for compatibility with v1
										// It may be used in prometheus rules and svc selectors in v1
										Labels: map[string]string{
											"app.kubernetes.io/component":  "tiflash",
											"app.kubernetes.io/instance":   consts.ClusterPrefix,
											"app.kubernetes.io/managed-by": "tidb-operator",
											"app.kubernetes.io/name":       "tidb-cluster",
										},
									},
								},
							},
						},
						Pod: &opv2.PodOverlay{
							ObjectMeta: opv2.ObjectMeta{
								Annotations: map[string]string{
									consts.TcInstanceTypeAnnotationKey: compCtx.InstanceType,
								},
								// All app.kubernetes.io labels are for compatibility with v1
								// It may be used in prometheus rules and svc selectors in v1
								Labels: map[string]string{
									"app.kubernetes.io/component":  "tiflash",
									"app.kubernetes.io/instance":   consts.ClusterPrefix,
									"app.kubernetes.io/managed-by": "tidb-operator",
									"app.kubernetes.io/name":       "tidb-cluster",
								},
							},
							Spec: &corev1.PodSpec{
								Tolerations:               compCtx.Tolerations,
								NodeSelector:              compCtx.NodeSelector,
								HostNetwork:               true,
								DNSPolicy:                 corev1.DNSClusterFirstWithHostNet,
								PriorityClassName:         consts.PriorityClasName,
								Affinity:                  utils.Affinity(cloudv1.ComponentTiFlash),
								TopologySpreadConstraints: t.buildTopologySpreadConstraints(),
								InitContainers: []corev1.Container{
									{
										Name: opv2.ContainerNameTiFlashServerLog,
										Resources: corev1.ResourceRequirements{
											Requests: corev1.ResourceList{
												corev1.ResourceCPU:    *resource.NewMilliQuantity(20, resource.DecimalSI),
												corev1.ResourceMemory: *resource.NewQuantity(5*MB, resource.BinarySI),
											},
											Limits: corev1.ResourceList{
												corev1.ResourceCPU:    *resource.NewMilliQuantity(100, resource.DecimalSI),
												corev1.ResourceMemory: *resource.NewQuantity(50*MB, resource.BinarySI),
											},
										},
									},
									{
										Name: opv2.ContainerNameTiFlashErrorLog,
										Resources: corev1.ResourceRequirements{
											Requests: corev1.ResourceList{
												corev1.ResourceCPU:    *resource.NewMilliQuantity(20, resource.DecimalSI),
												corev1.ResourceMemory: *resource.NewQuantity(5*MB, resource.BinarySI),
											},
											Limits: corev1.ResourceList{
												corev1.ResourceCPU:    *resource.NewMilliQuantity(100, resource.DecimalSI),
												corev1.ResourceMemory: *resource.NewQuantity(50*MB, resource.BinarySI),
											},
										},
									},
								},
								Containers: []corev1.Container{
									t.buildStoreContainer(opv2.ContainerNameTiFlash, compCtx.Resources),
								},
							},
						},
					},
				},
			},
		},
	}
	return tiflashGroup, nil
}

func (t *TiDBClusterV2) createOrUpdateSubResourcesForPD(ctx context.Context, cli client.Client) error {
	objs := []client.Object{
		// Required for PD dashboard
		&corev1.Secret{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "db-dashboard-session-secret",
				Namespace: t.Namer.GetTiDBNamespace(),
			},
			Data: map[string][]byte{
				"encryption_key": []byte(cloudv1.RandomString(32)),
			},
		},
	}

	for _, obj := range objs {
		if _, err := controllerutil.CreateOrUpdate(ctx, cli, obj, func() error { return nil }); err != nil {
			return err
		}
	}
	return nil
}

// TODO(ideascf): move to backup/restore-controller
func (t *TiDBClusterV2) createOrUpdateSubResourcesForBR(ctx context.Context, cli client.Client) error {
	objs := []client.Object{
		&corev1.ServiceAccount{
			ObjectMeta: metav1.ObjectMeta{
				Name:      backupManagerServiceAccountName,
				Namespace: t.Namer.GetTiDBNamespace(),
			},
		},
		&rbacv1.Role{
			ObjectMeta: metav1.ObjectMeta{
				Name:      backupManagerServiceAccountName,
				Namespace: t.Namer.GetTiDBNamespace(),
			},
			Rules: []rbacv1.PolicyRule{
				{
					APIGroups: []string{""},
					Resources: []string{"events"},
					Verbs:     []string{"*"},
				},
				{
					APIGroups: []string{"br.pingcap.com"},
					Resources: []string{"backups", "restores", "backups/status", "restores/status"},
					Verbs:     []string{"get", "watch", "list", "update"},
				},
			},
		},
		&rbacv1.RoleBinding{
			ObjectMeta: metav1.ObjectMeta{
				Name:      backupManagerServiceAccountName,
				Namespace: t.Namer.GetTiDBNamespace(),
			},
			Subjects: []rbacv1.Subject{
				{
					Kind: rbacv1.ServiceAccountKind,
					Name: backupManagerServiceAccountName,
				},
			},
			RoleRef: rbacv1.RoleRef{
				APIGroup: "rbac.authorization.k8s.io",
				Kind:     "Role",
				Name:     backupManagerServiceAccountName,
			},
		},
	}

	for _, obj := range objs {
		if _, err := controllerutil.CreateOrUpdate(ctx, cli, obj, func() error { return nil }); err != nil {
			return err
		}
	}
	return nil
}

func buildSchedulePolicies(zones []string) []opv2.SchedulePolicy {
	topologies := make([]opv2.ScheduleTopology, len(zones))
	for i, zone := range zones {
		topologies[i] = opv2.ScheduleTopology{Topology: map[string]string{consts.ZoneLabelKey: zone}}
	}
	return []opv2.SchedulePolicy{
		{
			Type: opv2.SchedulePolicyTypeEvenlySpread,
			EvenlySpread: &opv2.SchedulePolicyEvenlySpread{
				Topologies: topologies,
			},
		},
	}
}

// isClusterReady checks whether the tidb cluster is ready
// 1. check whether the tc is ready(If ready condition is true, it means all components are ready)
// 2. check whether the tidb pods are ready(If ready condition is true, it means readiness gates are ready and tidb service is ready to accept traffic)
func (t *TiDBClusterV2) isClusterReady(ctx context.Context) (bool, error) {
	var ready bool
	for _, c := range t.TidbClusterV2.Status.Conditions {
		if c.Type == opv2.ClusterCondAvailable && c.Status == metav1.ConditionTrue {
			ready = true
			break
		}
	}
	if !ready {
		return false, fmt.Errorf("tc is not ready yet")
	}

	remoteCli := t.EnvProvider.GetClient()
	selectors := getComponentPodLabels(cloudv1.ComponentTiDB, t.TidbClusterV2.Name, consts.ClusterPrefix)
	pods, err := infrak8s.SelectPods(ctx, remoteCli, t.TidbClusterV2.Namespace, selectors)
	if err != nil {
		return false, fmt.Errorf("failed to select tidb pods: %w", err)
	}

	tidb2Pods := make(map[string]corev1.Pod, len(pods))
	for _, pod := range pods {
		for _, owner := range pod.GetOwnerReferences() {
			if owner.Kind == "TiDB" && owner.APIVersion == opv2.SchemeGroupVersion.String() {
				tidb2Pods[owner.Name] = pod
			}
		}
	}

	var tidbList opv2.TiDBList
	if err = remoteCli.List(ctx, &tidbList, client.InNamespace(t.TidbClusterV2.Namespace), client.MatchingLabels(selectors)); err != nil {
		return false, fmt.Errorf("failed to list TiDB instances: %w", err)
	}

	for _, tidb := range tidbList.Items {
		pod, ok := tidb2Pods[tidb.Name]
		if !ok {
			return false, fmt.Errorf("the pod of tidb(%s) is not found", tidb.Name)
		}
		if !infrak8s.IsPodReady2(pod) {
			return false, fmt.Errorf("the pod of tidb(%s) is not ready", tidb.Name)
		}
	}

	return true, nil
}

func (t *TiDBClusterV2) logModifyProgress() {
	t.Logger.Info("up to date replicas",
		"pd", t.Cluster.GetUpToDateReplicas(cloudv1.ComponentPD),
		"tikv", t.Cluster.GetUpToDateReplicas(cloudv1.ComponentTiKV),
		"tidb", t.Cluster.GetUpToDateReplicas(cloudv1.ComponentTiDB),
		"tiproxy", t.Cluster.GetUpToDateReplicas(cloudv1.ComponentTiProxy),
		"tiflash", t.Cluster.GetUpToDateReplicas(cloudv1.ComponentTiFlash))
	t.Logger.Info("remaining to delete replicas",
		"pd", t.Cluster.GetRemainingDeleteReplicas(cloudv1.ComponentPD),
		"tikv", t.Cluster.GetRemainingDeleteReplicas(cloudv1.ComponentTiKV),
		"tidb", t.Cluster.GetRemainingDeleteReplicas(cloudv1.ComponentTiDB),
		"tiproxy", t.Cluster.GetRemainingDeleteReplicas(cloudv1.ComponentTiProxy),
		"tiflash", t.Cluster.GetRemainingDeleteReplicas(cloudv1.ComponentTiFlash))
}

func (t *TiDBClusterV2) Delete(ctx context.Context) deploy.Result {
	cli := t.EnvProvider.GetClient()
	return deploy.ErrorOrDone(infrak8s.DeleteObject(ctx, cli, &opv2.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.Namer.GetTiDBClusterName(),
			Namespace: t.Namer.GetTiDBNamespace(),
		},
	}))
}

func (t *TiDBClusterV2) PostDelete(ctx context.Context) deploy.Result {
	cli := t.EnvProvider.GetClient()
	key := types.NamespacedName{Name: t.Namer.GetTiDBClusterName(), Namespace: t.Namer.GetTiDBNamespace()}
	return deploy.ErrorOrDone(infrak8s.GetDeletedObject(ctx, cli, key, &opv2.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.Namer.GetTiDBClusterName(),
			Namespace: t.Namer.GetTiDBNamespace(),
		},
	}))
}

func (t *TiDBClusterV2) tikvTiflashScaleIn(ctx context.Context, cli client.Client) bool {
	var currentTiKVGroup opv2.TiKVGroup
	if err := cli.Get(ctx, client.ObjectKey{Name: consts.ClusterPrefix, Namespace: t.Cluster.PrefixedId()}, &currentTiKVGroup); err != nil {
		t.Logger.Error(err, "failed to get current TiKVGroup")
	} else if currentTiKVGroup.Spec.Replicas != nil && int(*currentTiKVGroup.Spec.Replicas) > t.Cluster.Spec.Cluster.TiKV.Replicas {
		return true
	}

	if desiredReplicas := t.Cluster.Spec.Cluster.TiFlash.Replicas; desiredReplicas > 0 {
		var currentTiFlashGroup opv2.TiFlashGroup
		if err := cli.Get(ctx, client.ObjectKey{Name: consts.ClusterPrefix, Namespace: t.Cluster.PrefixedId()}, &currentTiFlashGroup); client.IgnoreNotFound(err) != nil {
			t.Logger.Error(err, "failed to get current TiFlashGroup")
		} else if currentTiFlashGroup.Spec.Replicas != nil && int(*currentTiFlashGroup.Spec.Replicas) > desiredReplicas {
			return true
		}
	}

	return false
}

// when scaling in TiKV or TiFlash, PD will choose schedule targets for regions based on `low-space-ratio` and `high-space-ratio`.
// we have a pre-check for the remaining storage based on `low-space-ratio` in core-svc,
// but the scaling operation may still be blocked if more new data is written, data compression ratio is changed, the data is not evenly distributed, etc.
// so we set a greater value for `low-space-ratio` before scaling in and reset it after scaling in.
// ref:
// - https://github.com/tikv/pd/blob/c55e632b164b5c4d339f0d6247173b8a8200a92f/pkg/core/store.go#L421
// - https://github.com/tidbcloud/core-svc/pull/737
// NOTE: GitOps may also modify the `low-space-ratio` value.
//
// In each reconcile loop, we will:
// 1. query the current `low-space-ratio` value from PD
//   - this value may be set by GitOps or by this cluster controller or just the default value
// 2. record the current `low-space-ratio` value in the annotation of the TidbCluster if and ONLY if a scale in operation will be triggered for TiKV/TiFlash
//   - if apply TidbCluster to K8s failed, retry should work as no new value if updated to PD
// 3. set `low-space-ratio` to a greater value if TiKV/TiFlash is scaling in and the current value (may be set by GitOps) is less than the target value
//   - if the current value is greater than or equal to the target, do nothing, as it may be set by GitOps
//   - scaling in check: len(members) > replicas-in-spec now
// 4. reset `low-space-ratio` to the value recorded in the annotation of the TidbCluster if TiKV/TiFlash is normal and the current value is greater than the recorded value
//   - if the current value is less than or equal to the recorded value, do nothing, as it may be set by GitOps
//
// When scaling in, the max(target value, GitOps value) will be used.
// Otherwise, the min(current value, recorded value) will be used.

func (t *TiDBClusterV2) getPDLowSpaceRatio(ctx context.Context) *float64 {
	if t.PDClient == nil {
		return nil
	}

	// NOTE: `t.PDClient.GetConfig` may get a stale config, so we use HTTP API directly
	body, err := t.PDClient.HTTPGet(ctx, pdScheduleConfigPath)
	if err != nil {
		metrics.ConnectToClusterProxyFailed.WithLabelValues(t.Cluster.GetProjectID(), t.EnvProvider.GetRegion(), err.Error()).Inc()
		t.Logger.Error(err, "failed to get PD schedule config")
		return nil
	}
	cfg := map[string]any{}
	if err = json.Unmarshal(body, &cfg); err != nil {
		t.Logger.Error(err, "failed to unmarshal PD config", "config", string(body))
		return nil
	}

	if schedule, ok := cfg["low-space-ratio"]; ok {
		if v, ok := schedule.(float64); ok {
			return &v
		}
	}
	return nil
}

func (t *TiDBClusterV2) setPDLowSpaceRatio(ctx context.Context) error {
	if t.PDClient == nil || t.TidbClusterV2 == nil || t.Cluster.Status.DB == nil ||
		(t.Cluster.Status.Phase != cloudv1.ClusterPhaseScaling && t.Cluster.Status.Phase != cloudv1.ClusterPhaseRunning) ||
		(t.Cluster.Status.DB.TiKV == nil && t.Cluster.Status.DB.TiFlash == nil) {
		return nil
	}

	// only set/reset PD low space ratio if the previous value is recorded in the annotation
	var (
		previous float64
		err      error
	)
	if v, ok := t.TidbClusterV2.Annotations[annoKeyPDLowSpaceRatio]; !ok {
		return nil
	} else if previous, err = strconv.ParseFloat(v, 64); err != nil {
		t.Logger.Error(err, "failed to parse previous low space ratio", "previous", v)
		return nil
	}

	current := t.getPDLowSpaceRatio(ctx)
	if current == nil {
		return nil
	}

	putPDLowSpaceRatio := func(lowSpaceRatio float64) error {
		if t.Config.DryRun.Enabled {
			return fmt.Errorf("[DryRun] Cannot put low space ratio to %v, current is %v, previous is %v", lowSpaceRatio, current, previous)
		}
		body := map[string]any{
			"schedule.low-space-ratio": lowSpaceRatio,
		}
		data, err := json.Marshal(body)
		if err != nil {
			return err
		}
		// NOTE: `t.PDClient.SetConfig` may NOT work for `low-space-ratio` (from 0.9 to 0.8), so we use HTTP API directly
		return metrics.PDClientErrorHandler(t.PDClient.HTTPPost(ctx, pdConfigPrefix, data), t.Cluster.GetProjectID(), t.EnvProvider.GetRegion())
	}

	switch t.Cluster.Status.Phase {
	case cloudv1.ClusterPhaseScaling:
		if *current >= pdLowSpaceRatioInScalingIn {
			return nil
		}

		tikvStatus := t.Cluster.Status.DB.TiKV
		tiflashStatus := t.Cluster.Status.DB.TiFlash
		var scalingIn bool
		if tikvStatus != nil && tikvStatus.Status == cloudv1.ComponentScaling &&
			len(tikvStatus.Members) > int(ptr.Deref(t.kvg.Spec.Replicas, 0)) {
			scalingIn = true
		}
		if tiflashStatus != nil && tiflashStatus.Status == cloudv1.ComponentScaling && t.flashg != nil &&
			len(tiflashStatus.Members) > int(ptr.Deref(t.flashg.Spec.Replicas, 0)) && t.flashg.Spec.Replicas != nil && *t.flashg.Spec.Replicas > 0 {
			// special case: not set `low-space-ratio` if scaling to 0
			scalingIn = true
		}

		if scalingIn {
			if err := putPDLowSpaceRatio(pdLowSpaceRatioInScalingIn); err != nil {
				return err
			}
			t.Logger.Info("set PD low-space-ratio when scaling in TiKV/TiFlash", "from", *current, "to", pdLowSpaceRatioInScalingIn)
		}
	case cloudv1.ClusterPhaseRunning:
		if *current <= previous || *current != pdLowSpaceRatioInScalingIn {
			// if a different value is set by GitOps, do nothing
			return nil
		}
		if err := putPDLowSpaceRatio(previous); err != nil {
			return err
		}
		t.Logger.Info("reset PD low-space-ratio after scaled in TiKV/TiFlash", "from", *current, "to", previous)
	}

	return nil
}