package configmanager

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

	"github.com/go-logr/logr"
	"github.com/google/go-cmp/cmp"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	operatorconfig "github.com/pingcap/tidb-operator/pkg/apis/util/config"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/applier"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/gitops-sdk/utils"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

var (
	ConfigUpdateAllComponents = []cloudv1.ComponentType{cloudv1.ComponentTiDB, cloudv1.ComponentTiKV, cloudv1.ComponentPD,
		cloudv1.ComponentTiFlash, cloudv1.ComponentTiCDC, cloudv1.ComponentTiProxy}
)

const (
	// recommendedEvictLeaderTimeout is set according to
	// https://docs.pingcap.com/tidb-in-kubernetes/dev/configure-a-tidb-cluster#configure-graceful-upgrade-for-tikv-cluster
	recommendedEvictLeaderTimeout = "10000m"
	defaultTiDBImage              = "gcr.io/pingcap-public/dbaas/tidb"
	defaultTiKVImage              = "gcr.io/pingcap-public/dbaas/tikv"
	defaultPDImage                = "gcr.io/pingcap-public/dbaas/pd"
	defaultTiCDCImage             = "gcr.io/pingcap-public/dbaas/ticdc"
	defaultTiFlashImage           = "gcr.io/pingcap-public/dbaas/tiflash"
	defaultBRImage                = "gcr.io/pingcap-public/dbaas/br"
	defaultTiProxyImage           = "gcr.io/pingcap-public/dbaas/tiproxy"

	// annoKeyPDClusterConfHash is the key of the annotation that stores the hash of the PD cluster config.
	// This is used to avoid unnecessary updates on PD where the `isPDClusterConfigChanged` can't distinguish false positive.
	// For example, the order of configuration item `schedule.schedulers` is different between the current and the desired config.
	annoKeyPDClusterConfHash = "tidbcloud.com/pd-cluster-conf-hash"

	// pdAPIErrSchedulersNotFound matches PD HTTP 400 bodies such as:
	// `config item schedulers not found` when the server no longer accepts `schedule.schedulers` via the config API.
	pdAPIErrSchedulersNotFound = "schedulers not found"
)

type configManager struct {
	client.Client
	// preConfigKeys defines the keys that are not managed by this configManager,
	// and they will not be modified during configuration updates.
	preConfigKeys map[cloudv1.ComponentType][]string

	rootConfig *config.RootConfig
}

func NewConfigManager(infraCli client.Client, preConfig map[cloudv1.ComponentType][]string, conf *config.RootConfig) Interface {
	return &configManager{
		Client:        infraCli,
		preConfigKeys: preConfig,
		rootConfig:    conf,
	}
}

func (m *configManager) InitConfig(cluster *cloudv1.Cluster, revFinder RevisionFinder, tc *v1alpha1.TidbCluster) error {
	revision, err := revFinder(cluster)
	if err != nil {
		return err
	}
	// set the initConfig and image version for components via desired config
	m.patchTiDBCluster(context.TODO(), cluster, tc, revision, ConfigUpdateAllComponents, false /* update */, false)
	return nil
}

func (m *configManager) UpdateConfig(ctx context.Context, cluster *cloudv1.Cluster, regionalManaged bool, revFinder RevisionFinder, tc *v1alpha1.TidbCluster,
	pdCli DynamicPDClient, comps []cloudv1.ComponentType) error {
	logger := m.logger(ctx).WithValues("cluster", cluster.Name, "tc", tc.Name)

	oldRevisionName := tc.Annotations[cloudv1.ConfigRevisionIdKey]
	if oldRevisionName == "" {
		logger.Error(fmt.Errorf("can't fid old ConfigRevision from tc, use cluster.Spec.ConfigRevisionRef as old one"), "", "cluster", cluster.Name)
		oldRevisionName = cluster.Spec.ConfigRevisionRef.Name
	}
	oldRevision, err := m.GetConfigRevisions(ctx, types.NamespacedName{Name: oldRevisionName})
	if err != nil {
		return err
	}

	revision, err := revFinder(cluster)
	if err != nil {
		return err
	}

	// revision is nil means cluster.ConfigRevision is nil and cc.IsDryRun is true
	// This is a wrong case, we can't go update the config
	// Most of the cases are caused by the failure of the configmanager webhock
	if revision == nil {
		return err
	}

	logger.Info("get configRevision", "curr", oldRevision.Name, "new", revision.Name)
	// check guard if the cluster version is changed
	if revision.Spec.Version != oldRevision.Spec.Version {
		logger.Info("cluster version change detected", "curr_version", oldRevision.Spec.Version, "new_version", revision.Spec.Version)
		if err := cluster.CheckVersionGuard(revision); err != nil {
			GuardCheckFail.WithLabelValues(cluster.Name, cloudv1.ClusterGuardVersionUpgrade).Set(1)
			return err
		} else {
			logger.Info("guard check pass", "target_revision", revision.Name, "old_revision", oldRevision.Name)
			GuardCheckFail.WithLabelValues(cluster.Name, cloudv1.ClusterGuardVersionUpgrade).Set(0)
		}
	}

	pdNeeded := false
	for _, comp := range comps {
		if comp == cloudv1.ComponentPD {
			pdNeeded = true
			break
		}
	}
	if tc.Spec.PD != nil && pdNeeded {
		if pdCli == nil {
			return errors.New("pdClient should not be nil when PD is present")
		}
		// get pd cluster config that cloud be online tuned from desired config
		pcc := cloudv1.GetPDClusterConfig(revision.Spec.PD.Config)
		// refer to https://docs.pingcap.com/tidb-in-kubernetes/dev/modify-tidb-configuration#modify-pd-configuration
		// pd config can't be updated in tidb-operator, update it via pdClient.
		err := m.updatePDClusterConfig(ctx, logger, pdCli, pcc, cluster)
		if err != nil {
			return err
		}
	}

	updateCDC := true
	if _, ok := cluster.Annotations[cloudv1.AnnoTiCDCIgnoreConfigUpdate]; ok || regionalManaged {
		// if the annotation exists, it means ticdc controller has taken over config update of the ticdc
		// cluster controller should ignore ticdc config.
		// skip if it's regional managed, whatever the cluster is migrated from cloud-native arch or initially created in 3-layers arch.
		updateCDC = false
	}

	// after mutate all the component configs, update tc,
	// if conflicted, let provider do the retry thing.
	err = m.updateTidbClusterConfig(ctx, tc, cluster, func() error {
		cloned := tc.DeepCopy()
		m.patchTiDBCluster(ctx, cluster, tc, revision, comps, true /* update */, updateCDC)

		return m.ensureNoUnexpectedConfigChange(ctx, cluster, cloned, tc)
	})
	if err != nil {
		return err
	}

	return err
}

// NOTE: do not forget to update `ConfigUpdateAllComponents` if any new component is added.
func (m *configManager) patchTiDBCluster(ctx context.Context, cluster *cloudv1.Cluster, tc *v1alpha1.TidbCluster, revision *cloudv1.ConfigRevision,
	comps []cloudv1.ComponentType, update bool, updateCDC bool) {
	logger := m.logger(ctx).WithValues("cluster", cluster.Name)
	if tc.Annotations == nil {
		tc.Annotations = map[string]string{}
	}
	isTiDBVersionChanged, curTiDBVersion, newTiDBVersion := CheckIfIsTiDBVersionChanged(tc, revision)
	logger.Info("patching tidbcluster", "update", update, "comps", comps, "isTiDBVersionChanged", isTiDBVersionChanged, "curTiDBVersion", curTiDBVersion, "newTiDBVersion", newTiDBVersion)
	// cluster level fileds will always be updated
	tc.Annotations[cloudv1.ConfigRevisionIdKey] = revision.Name
	tc.Spec.Version = revision.Spec.Version

	for _, comp := range comps {
		switch comp {
		case cloudv1.ComponentTiDB:
			if tc.Spec.TiDB != nil {
				confMap := make(map[string]interface{})
				if tc.Spec.TiDB.Config != nil {
					confMap = tc.Spec.TiDB.Config.MP
					if update {
						confMap = m.getPreConfig(tc.Spec.TiDB.Config.GenericConfig, cloudv1.ComponentTiDB)
					}
				}
				desired := revision.Spec.TiDB
				conf := operatorconfig.New(utils.MergeMaps(desired.Config.MP, confMap))

				hackForGracefulWaitBeforeShutdownForTiDB(cluster, tc, conf.MP, tc.Spec.TiDB.Config.MP)

				tc.Spec.TiDB.Config = &v1alpha1.TiDBConfigWraper{GenericConfig: conf}
				tc.Spec.TiDB.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultTiDBImage)
				if isTiDBVersionChanged {
					m.UpdateTiDBAuditLogSidecarAnno(ctx, cluster)
				}
				tc.Spec.TiDB.Version = desired.Tag
			}
		case cloudv1.ComponentTiProxy:
			desired := revision.Spec.TiProxy
			if desired != nil && tc.Spec.TiProxy != nil {
				confMap := make(map[string]interface{})
				if tc.Spec.TiProxy.Config != nil {
					confMap = tc.Spec.TiProxy.Config.MP
					if update {
						confMap = m.getPreConfig(tc.Spec.TiProxy.Config.GenericConfig, cloudv1.ComponentTiProxy)
					}
				}
				if desired.Config != nil {
					conf := operatorconfig.New(utils.MergeMaps(desired.Config.MP, confMap))
					tc.Spec.TiProxy.Config = &v1alpha1.TiProxyConfigWraper{GenericConfig: conf}
				} else {
					tc.Spec.TiProxy.Config = v1alpha1.NewTiProxyConfig()
				}
				tc.Spec.TiProxy.Version = desired.Tag
				tc.Spec.TiProxy.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultTiProxyImage)
			}
		case cloudv1.ComponentTiKV:
			if tc.Spec.TiKV != nil {
				confMap := make(map[string]interface{})
				if tc.Spec.TiKV.Config != nil {
					confMap = tc.Spec.TiKV.Config.MP
					if update {
						confMap = m.getPreConfig(tc.Spec.TiKV.Config.GenericConfig, cloudv1.ComponentTiKV)
					}
				}
				desired := revision.Spec.TiKV
				conf := operatorconfig.New(utils.MergeMaps(desired.Config.MP, confMap))
				tc.Spec.TiKV.Config = &v1alpha1.TiKVConfigWraper{GenericConfig: conf}
				tc.Spec.TiKV.Version = desired.Tag
				tc.Spec.TiKV.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultTiKVImage)

				// Turn up the `evictLeaderTimeout`.
				// The default is 10 minutes, which may cause an error.
				// See more details in https://jira.tidbcloud.com/browse/DM-1051
				// Note: this modification will not make TiKV restart.
				evictLeaderTimeout := recommendedEvictLeaderTimeout
				tc.Spec.TiKV.EvictLeaderTimeout = &evictLeaderTimeout
			}
		case cloudv1.ComponentPD:
			if tc.Spec.PD != nil {
				confMap := make(map[string]interface{})
				if tc.Spec.PD.Config != nil {
					confMap = tc.Spec.PD.Config.MP
					if update {
						currentPCC := cloudv1.ExtractPDClusterConfig(tc.Spec.PD.Config.MP)
						// combine pd cluster-level config and pre defined config
						confMap = utils.MergeMaps(m.getPreConfig(tc.Spec.PD.Config.GenericConfig, cloudv1.ComponentPD), currentPCC.MP)
						// drop the cluster-level config
					}
				}
				desired := revision.Spec.PD
				// to avoid unnecessary rolling-restart, remove the cluster-level config from desired config,
				// then overlay the desired instance-level config by combined config,
				// thus to keep the cluster config untouched at instance level.
				desiredInstanceConf := cloudv1.GetPDInstanceConfig(desired.Config)
				conf := operatorconfig.New(utils.MergeMaps(desiredInstanceConf.MP, confMap))
				tc.Spec.PD.Config = &v1alpha1.PDConfigWraper{GenericConfig: conf}
				tc.Spec.PD.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultPDImage)
				tc.Spec.PD.Version = desired.Tag
			}
		case cloudv1.ComponentTiFlash:
			if tc.Spec.TiFlash != nil {
				commonConfMap, proxyConfMap := make(map[string]interface{}), make(map[string]interface{})
				if tc.Spec.TiFlash.Config != nil {
					if tc.Spec.TiFlash.Config.Common != nil {
						commonConfMap = tc.Spec.TiFlash.Config.Common.MP
						if update {
							commonConfMap = m.getPreConfig(tc.Spec.TiFlash.Config.Common.GenericConfig, cloudv1.ComponentTiFlash)
						}
					}
					if tc.Spec.TiFlash.Config.Proxy != nil {
						proxyConfMap = tc.Spec.TiFlash.Config.Proxy.MP
						if update {
							proxyConfMap = m.getPreConfig(tc.Spec.TiFlash.Config.Proxy.GenericConfig, cloudv1.ComponentTiFlashLearner)
						}
					}
				}

				desired := revision.Spec.TiFlash
				conf := v1alpha1.NewTiFlashConfig()
				conf.Common = &v1alpha1.TiFlashCommonConfigWraper{GenericConfig: operatorconfig.New(utils.MergeMaps(desired.Config.MP, commonConfMap))}
				if revision.Spec.TiFlashLearner != nil {
					proxyConfMap = utils.MergeMaps(revision.Spec.TiFlashLearner.Config.MP, proxyConfMap)
				}
				conf.Proxy = &v1alpha1.TiFlashProxyConfigWraper{GenericConfig: operatorconfig.New(proxyConfMap)}
				tc.Spec.TiFlash.Config = conf
				tc.Spec.TiFlash.Version = desired.Tag
				tc.Spec.TiFlash.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultTiFlashImage)
			}
		case cloudv1.ComponentTiCDC:
			// update config for ticdc only when meet the following conditions
			// 1. config of ticdc is managed by ConfigRevision: revision.Spec.TiCDC != nil
			// 2. cluster is NOT annotated with cloudv1.AnnoTiCDCIgnoreConfigUpdate: updateCDC == true
			// 3. TC has ticdc deployed: tc.spec.TiCDC != nil
			if revision.Spec.TiCDC != nil && updateCDC && tc.Spec.TiCDC != nil {
				confMap := make(map[string]interface{})
				if tc.Spec.TiCDC.Config != nil {
					confMap = tc.Spec.TiCDC.Config.GenericConfig.MP
					if update {
						confMap = m.getPreConfig(tc.Spec.TiCDC.Config.GenericConfig, cloudv1.ComponentTiCDC)
					}
				}
				desired := revision.Spec.TiCDC
				conf := operatorconfig.New(utils.MergeMaps(desired.Config.MP, confMap))
				tc.Spec.TiCDC.Config = &v1alpha1.CDCConfigWraper{GenericConfig: conf}
				tc.Spec.TiCDC.Version = desired.Tag
				// don't update base image of ticdc, because the ticdc base image in revision is not overridden,
				// and the ticdc pods will be restarted due to base image changed
				// tc.Spec.TiCDC.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultTiCDCImage)
			}
		}
	}
}

// TODO(ideascf): remove this hack
// hackForGracefulWaitBeforeShutdownForTiDB is used for managing 'graceful-wait-before-shutdown' config for tidb.
//
//  1. if the target config has graceful-wait-before-shutdown, then use the target config's value.
//  2. if the target config has no graceful-wait-before-shutdown, and the current config has, then use the current config's value.
//  3. if the target config has no graceful-wait-before-shutdown, and the current config has no graceful-wait-before-shutdown,
//     then automatically set the graceful-wait-before-shutdown to the default value when the tiproxy is enabled.
func hackForGracefulWaitBeforeShutdownForTiDB(cluster *cloudv1.Cluster, tc *v1alpha1.TidbCluster, target, current map[string]interface{}) {
	_, targetExists := target[consts.TiDBGracefulShutdownWaitSecondsKey]
	if targetExists {
		return
	}

	gracefulWaitBeforeShutdownInCurrent, currentExists := current[consts.TiDBGracefulShutdownWaitSecondsKey]
	// if the target config has no graceful-wait-before-shutdown, and the current config has, then use the current config's value.
	// In other words, once the graceful-wait-before-shutdown is set, it will not be removed.
	if currentExists {
		target[consts.TiDBGracefulShutdownWaitSecondsKey] = gracefulWaitBeforeShutdownInCurrent
		return
	}

	// Note: we have already changed the tc.spec.configUpdateStrategy to InPlace, so config changes of tidb will NOT directly
	// trigger restarts of tidb. Instead, the config change will be applied by changing the .spec.tidb.terminationGracePeriodSeconds or
	// .spec.tidb.tlsClient.skipInternalClientCA or other spec fields.
	//
	// Automatically set the graceful-wait-before-shutdown when tiproxy is enabled and it's not set.
	if _, ok := target[consts.TiDBGracefulShutdownWaitSecondsKey]; !ok && (tc.Spec.TiProxy != nil && tc.Spec.TiProxy.Replicas > 0) {
		target[consts.TiDBGracefulShutdownWaitSecondsKey] = GetDefaultTidbGracefulShutdownWaitSeconds(cluster)
	}
}

const (
	AnnoSkipEnsureNoUnexpectedConfigChange = "cluster.tidbcloud.com/skip-ensure-no-unexpected-config-change"

	// the following two annotations are copied from pkg/consts/consts.go

	// ClusterUpdateAnnotationKeyInstanceType indicates the cluster need to be updated as instance types changed.
	// the value should be a list of component names, separated by comma.
	ClusterUpdateAnnotationKeyInstanceType = "eks.infra.pingcap.com/cluster-update-instance-type"
	AnnoInstanceTypeSeparator              = ","
)

func (m *configManager) ensureNoUnexpectedConfigChange(ctx context.Context, cluster *cloudv1.Cluster, currentTC, newTC *v1alpha1.TidbCluster) error {
	logger := m.logger(ctx).WithValues("clusterName", cluster.ClusterScopeName())
	changedComponent := map[cloudv1.ComponentType]bool{}

	if cluster.Annotations[AnnoSkipEnsureNoUnexpectedConfigChange] == "true" {
		return nil
	}
	if cluster.Status.Phase == cloudv1.ClusterPhaseCreating {
		return nil
	}

	// get changed components from cluster annotation if the cluster is scaling
	compAnno := cluster.Annotations[ClusterUpdateAnnotationKeyInstanceType]
	comps := strings.Split(compAnno, AnnoInstanceTypeSeparator)
	for _, each := range comps {
		changedComponent[cloudv1.ComponentType(each)] = true
	}
	// Pay Attention: if you scale tiflash out from zero, tiflash doesn't appear in annotation `ClusterUpdateAnnotationKeyInstanceType`.
	// But you can't add tiflash into `changedComponent` by this like below, because full spec of tiflash will be
	// set into currentTC before call `t.ConfigManager.UpdateConfig` in tidbcluster.go:Create.
	// In other words, currentTc.spec.tiflash always is equal to newTc.spec.tiflash in here.
	//if (currentTC.Spec.TiFlash == nil || currentTC.Spec.TiFlash.Replicas == 0) &&
	//	(newTC.Spec.TiFlash != nil && newTC.Spec.TiFlash.Replicas != 0) {
	//	changedComponent[cloudv1.ComponentTiFlash] = true
	//}

	// get changed components from ConfigChange status if the cluster is config changing
	if cluster.ConfigChanging() {
		var cc cloudv1.ConfigChange
		if err := m.Client.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.CurrentConfigChangeName()}, &cc); err != nil {
			return fmt.Errorf("get ConfigChange(%s): %w", cluster.CurrentConfigChangeName(), err)
		}
		for _, each := range cc.Status.ChangedComponents {
			changedComponent[(each)] = true
		}
	}

	tiproxyCntInSpec := cluster.Spec.Cluster.TiProxy.Replicas
	tiproxyCntInStatus := len(currentTC.Status.TiProxy.Members)
	// enable/disable tiproxy will implicitly change configuration for tidb, so hack this logic here.
	// changed configuration items:
	//   - graceful-wait-before-shutdown
	if (tiproxyCntInSpec != 0 && tiproxyCntInStatus == 0) || (tiproxyCntInSpec == 0 && tiproxyCntInStatus != 0) {
		logger.Info("mark tidb as changed as disabling/enabling tiproxy", "tiproxyCntInSpec", tiproxyCntInSpec, "tiproxyCntInStatus", tiproxyCntInStatus)
		changedComponent[cloudv1.ComponentTiDB] = true
	}

	tikvAdditionalVolumesInSpec := cluster.Spec.Cluster.TiKV.StorageVolumes
	tikvVolumesInStatus := currentTC.Status.TiKV.Volumes
	// add/remove tikv additional volumes will implicitly change configuration for tikv, so hack this logic here.
	if len(tikvAdditionalVolumesInSpec) != len(tikvVolumesInStatus)-1 { // the default data volume should be excluded
		logger.Info("mark tikv as changed as adding/removing tikv additional volumes")
		changedComponent[cloudv1.ComponentTiKV] = true
	}

	logger.Info("changed components", "components", changedComponent)

	for _, compType := range ConfigUpdateAllComponents {
		currentComp := currentTC.ComponentSpec(v1alpha1.MemberType(compType))
		if currentComp == nil {
			logger.Info("component not found in current tc", "component", compType)
			continue
		}
		newComp := newTC.ComponentSpec(v1alpha1.MemberType(compType))
		if newComp == nil {
			logger.Info("component not found in new tc", "component", compType)
			continue
		}

		var (
			currentConfig, newConfig                                   interface{}
			currentBaseImage, currentVersion, newBaseImage, newVersion string
		)
		switch compType {
		case cloudv1.ComponentTiProxy:
			currentConfig = currentTC.Spec.TiProxy.Config
			currentBaseImage = currentTC.Spec.TiProxy.BaseImage
			currentVersion = pointer.StringDeref(currentTC.Spec.TiProxy.Version, "")
			newConfig = newTC.Spec.TiProxy.Config
			newBaseImage = newTC.Spec.TiProxy.BaseImage
			newVersion = pointer.StringDeref(newTC.Spec.TiProxy.Version, "")
		case cloudv1.ComponentTiDB:
			currentConfig = currentTC.Spec.TiDB.Config
			currentBaseImage = currentTC.Spec.TiDB.BaseImage
			currentVersion = pointer.StringDeref(currentTC.Spec.TiDB.Version, "")
			newConfig = newTC.Spec.TiDB.Config
			newBaseImage = newTC.Spec.TiDB.BaseImage
			newVersion = pointer.StringDeref(newTC.Spec.TiDB.Version, "")
		case cloudv1.ComponentTiKV:
			currentConfig = currentTC.Spec.TiKV.Config
			currentBaseImage = currentTC.Spec.TiKV.BaseImage
			currentVersion = pointer.StringDeref(currentTC.Spec.TiKV.Version, "")
			newConfig = newTC.Spec.TiKV.Config
			newBaseImage = newTC.Spec.TiKV.BaseImage
			newVersion = pointer.StringDeref(newTC.Spec.TiKV.Version, "")
		case cloudv1.ComponentPD:
			currentConfig = currentTC.Spec.PD.Config
			currentBaseImage = currentTC.Spec.PD.BaseImage
			currentVersion = pointer.StringDeref(currentTC.Spec.PD.Version, "")
			newConfig = newTC.Spec.PD.Config
			newBaseImage = newTC.Spec.PD.BaseImage
			newVersion = pointer.StringDeref(newTC.Spec.PD.Version, "")
		case cloudv1.ComponentTiFlash:
			currentConfig = currentTC.Spec.TiFlash.Config
			currentBaseImage = currentTC.Spec.TiFlash.BaseImage
			currentVersion = pointer.StringDeref(currentTC.Spec.TiFlash.Version, "")
			newConfig = newTC.Spec.TiFlash.Config
			newBaseImage = newTC.Spec.TiFlash.BaseImage
			newVersion = pointer.StringDeref(newTC.Spec.TiFlash.Version, "")
		case cloudv1.ComponentTiCDC:
			currentConfig = currentTC.Spec.TiCDC.Config
			currentBaseImage = currentTC.Spec.TiCDC.BaseImage
			currentVersion = pointer.StringDeref(currentTC.Spec.TiCDC.Version, "")
			newConfig = newTC.Spec.TiCDC.Config
			newBaseImage = newTC.Spec.TiCDC.BaseImage
			newVersion = pointer.StringDeref(newTC.Spec.TiCDC.Version, "")
		default:
			continue
		}

		if apiequality.Semantic.DeepEqual(currentConfig, newConfig) &&
			currentBaseImage == newBaseImage &&
			currentVersion == newVersion {
			continue
		}

		if !changedComponent[compType] {
			logger.Info("expect component unchanged but changed",
				"component", compType,
				"currentConfig", currentConfig,
				"newConfig", newConfig,
				"currentBaseImage", currentBaseImage,
				"newBaseImage", newBaseImage,
				"currentVersion", currentVersion,
				"newVersion", newVersion,
			)
			UnexpectedConfigChange.WithLabelValues(cluster.Name, string(compType), string(cluster.Status.Phase)).Set(1)
		} else {
			logger.Info("ensureNoUnexpectedConfigChange pass")
			UnexpectedConfigChange.WithLabelValues(cluster.Name, string(compType), string(cluster.Status.Phase)).Set(0)
		}
	}
	return nil
}

func (m *configManager) getPreConfig(config *operatorconfig.GenericConfig, component cloudv1.ComponentType) map[string]interface{} {
	preConfig := operatorconfig.New(make(map[string]interface{}))
	for _, key := range m.preConfigKeys[component] {
		v := config.Get(key)
		if v != nil {
			preConfig.Set(key, v.Interface())
		}
	}
	return preConfig.MP
}

func (m *configManager) updatePDClusterConfig(ctx context.Context, logger logr.Logger, pdCli DynamicPDClient, desired *cloudv1.PDClusterConfig, c *cloudv1.Cluster) error {
	// TODO(ideascf): we can use only hash to check if pd cluster config changed.
	conf, err := pdCli.GetConfig(ctx)
	if err != nil {
		return fmt.Errorf("get pd config: %w", err)
	}
	pcc := cloudv1.ExtractPDClusterConfig(conf)
	changed, err := isPDClusterConfigChanged(logger, pcc.MP, desired.MP)
	if err != nil {
		return err
	}
	if !changed {
		return nil
	}

	// As the `isPDClusterConfigChanged` can be false-positive, we double check it by hash.
	// As we don't flush pd cluster config to TC, so we add hash annotation to cluster CR instead of tc.
	newHash, err := isPDClusterConfHashChanged(logger, c, desired.MP)
	if err != nil {
		return err
	}
	if newHash == "" {
		return nil
	}

	if err := pdCli.PutConfig(ctx, desired.MP); err != nil {
		if !isPDPutConfigSchedulersNotFoundErr(err) {
			return fmt.Errorf("put pd cluster config: %w", err)
		}
		fallbackMP, cloneErr := clonePDClusterConfigMP(desired.MP)
		if cloneErr != nil {
			return fmt.Errorf("put pd cluster config: %w; clone config for schedulers retry: %v", err, cloneErr)
		}
		deleteScheduleSchedulers(fallbackMP)
		if err2 := pdCli.PutConfig(ctx, fallbackMP); err2 != nil {
			return fmt.Errorf("put pd cluster config (without schedule.schedulers): %w", err2)
		}
		logger.Info("put pd cluster config succeeded after retry without schedule.schedulers")
	}
	// update hash annotation after calling PutConfig successfully for idempotency.
	c.UpdateAnnotation(annoKeyPDClusterConfHash, newHash)
	return nil
}

func isPDPutConfigSchedulersNotFoundErr(err error) bool {
	if err == nil {
		return false
	}
	return strings.Contains(strings.ToLower(err.Error()), strings.ToLower(pdAPIErrSchedulersNotFound))
}

func clonePDClusterConfigMP(mp map[string]interface{}) (map[string]interface{}, error) {
	data, err := json.Marshal(mp)
	if err != nil {
		return nil, fmt.Errorf("marshal pd cluster config: %w", err)
	}
	out := make(map[string]interface{})
	if err := json.Unmarshal(data, &out); err != nil {
		return nil, fmt.Errorf("unmarshal pd cluster config: %w", err)
	}
	return out, nil
}

func deleteScheduleSchedulers(mp map[string]interface{}) {
	scheduleI, ok := mp["schedule"]
	if !ok {
		return
	}
	schedule, ok := scheduleI.(map[string]interface{})
	if !ok {
		return
	}
	delete(schedule, "schedulers")
}

func isPDClusterConfigChanged(logger logr.Logger, current, desired map[string]interface{}) (bool, error) {
	// Note:
	// 1. the `schedulers-v2` is a deprecated field, see https://github.com/tikv/pd/issues/8474.
	// 2. PD always convert `schedulers` to `schedulers-v2` and discard `schedulers` when get config from pd api.
	// 3. To avoid meaningless diff and updates to PD, we need to copy `schedulers-v2` back to `schedulers` for current before we compare it.
	if scheduleI, ok := current["schedule"]; ok {
		if schedule, ok := scheduleI.(map[string]interface{}); ok {
			schedulersV2, schedulersV2Existed := schedule["schedulers-v2"]
			_, schedulersExisted := schedule["schedulers"]
			if schedulersV2Existed && !schedulersExisted {
				schedule["schedulers"] = schedulersV2
			}
		}
	}
	updated := utils.MergeMaps(current, desired)

	// `current` is json.Unmarshal-ed from pd api, some fields will be float64 type.
	// However, the `updated` is got from ConfigRevision, some fields would be setin integer type.
	// So we use json marshal to make it with same type before compare it.
	data, err := json.Marshal(updated)
	if err != nil {
		return false, fmt.Errorf("marshal merged pd cluster config: %w", err)
	}
	updated = make(map[string]interface{})
	err = json.Unmarshal(data, &updated)
	if err != nil {
		return false, fmt.Errorf("unmarshal merged pd cluster config: %w", err)
	}

	if apiequality.Semantic.DeepEqual(current, updated) {
		return false, nil
	}
	// As configuration-based comparison is not accurate and cause false-positive, we don't log the diff here.
	// d := cmp.Diff(current, updated)
	// logger.Info("isPDClusterConfigChange: pd cluster config diff", "diff", d)
	return true, nil
}

// isPDClusterConfHashChanged return new hash if pd cluster config changed, otherwise return empty string.
func isPDClusterConfHashChanged(logger logr.Logger, c *cloudv1.Cluster, desired map[string]interface{}) (newHash string, _ error) {
	desiredhash, err := applier.Hash(desired)
	if err != nil {
		return "", fmt.Errorf("hash pd cluster config: %w", err)
	}
	curHash := c.Annotations[annoKeyPDClusterConfHash]
	if curHash == desiredhash {
		return "", nil
	}

	logger.Info("pd cluster config hash mismatch", "current", curHash, "desired", desiredhash)
	return desiredhash, nil
}

func (m *configManager) updateTidbClusterConfig(ctx context.Context, tc *v1alpha1.TidbCluster, c *cloudv1.Cluster, mutateFn func() error) error {
	logger := m.logger(ctx)

	old := tc.DeepCopy()
	err := mutateFn()
	if err != nil {
		return err
	}
	// no update
	if apiequality.Semantic.DeepEqual(old, tc) {
		return nil
	}
	logPlan(logger, old, tc, fmt.Sprintf("TidbCluster %s/%s resource of cluster %s/%s", tc.Namespace, tc.Name, c.Namespace, c.Name))
	return nil
}

func logPlan(logger logr.Logger, before, after interface{}, target string) {
	mode := "execute"
	d := cmp.Diff(before, after)
	msg := fmt.Sprintf("[%s] update %s, diff:\n%s", mode, target, d)
	logger.Info(msg)
}

func (m *configManager) InitBackupConfig(cluster *cloudv1.Cluster, revFinder RevisionFinder, backup *v1alpha1.Backup) error {
	revision, err := revFinder(cluster)
	if err != nil {
		return err
	}
	if backup.Spec.BR == nil {
		backup.Spec.BR = &v1alpha1.BRConfig{}
	}
	configBytes, err := json.Marshal(revision.Spec.Backup.Config.MP)
	if err != nil {
		return err
	}
	err = MergeBRConfig(backup.Spec.BR, configBytes)
	if err != nil {
		return err
	}

	tag := utils.DefaultIfEmpty(revision.Spec.Backup.Tag, revision.Spec.Version)
	baseImage := utils.DefaultIfEmpty(revision.Spec.Backup.BaseImage, defaultBRImage)
	backup.Spec.ToolImage = fmt.Sprintf("%s:%s", baseImage, tag)
	return nil
}

func (m *configManager) InitRestoreConfig(cluster *cloudv1.Cluster, revFinder RevisionFinder, restore *v1alpha1.Restore) error {
	revision, err := revFinder(cluster)
	if err != nil {
		return err
	}
	if restore.Spec.BR == nil {
		restore.Spec.BR = &v1alpha1.BRConfig{}
	}
	configBytes, err := json.Marshal(revision.Spec.Restore.Config.MP)
	if err != nil {
		return err
	}
	err = MergeBRConfig(restore.Spec.BR, configBytes)
	if err != nil {
		return err
	}

	tag := utils.DefaultIfEmpty(revision.Spec.Restore.Tag, revision.Spec.Version)
	baseImage := utils.DefaultIfEmpty(revision.Spec.Restore.BaseImage, defaultBRImage)
	restore.Spec.ToolImage = fmt.Sprintf("%s:%s", baseImage, tag)
	return nil
}

func (m *configManager) GetConfigRevisions(ctx context.Context, name types.NamespacedName) (*cloudv1.ConfigRevision, error) {
	cr := &cloudv1.ConfigRevision{}
	err := m.Client.Get(ctx, name, cr)
	if err != nil {
		return nil, errors.Annotatef(err, "get configRevisions with name %s", name.Name)
	}
	return cr, nil
}

func (m *configManager) logger(ctx context.Context) logr.Logger {
	return log.FromContext(ctx).WithValues("client", "gitops-sdk")
}

func MergeBRConfig(BR *v1alpha1.BRConfig, brConfBytes []byte) error {
	var brConfig v1alpha1.BRConfig
	if err := json.Unmarshal(brConfBytes, &brConfig); err != nil {
		return err
	}

	BR.LogLevel = brConfig.LogLevel
	BR.StatusAddr = brConfig.StatusAddr
	BR.TimeAgo = brConfig.TimeAgo
	BR.Options = brConfig.Options
	if brConfig.Concurrency != nil {
		BR.Concurrency = brConfig.Concurrency
	}
	if brConfig.RateLimit != nil {
		BR.RateLimit = brConfig.RateLimit
	}
	if brConfig.Checksum != nil {
		BR.Checksum = brConfig.Checksum
	}
	if brConfig.OnLine != nil {
		BR.OnLine = brConfig.OnLine
	}
	return nil
}

func GetDefaultTidbGracefulShutdownWaitSeconds(cluster *cloudv1.Cluster) int64 {
	if cluster.FeatureEnabled(consts.ClusterFeatureLegacyTiProxySettings) {
		return consts.DefaultTidbGracefulShutdownWaitSeconds
	}
	return consts.DefaultTidbGracefulShutdownWaitSecondsV2
}

// UpdateTiDBAuditLogSidecarAnno updates the audit log sidecar image tag annotation for the TiDB component in the cluster.
// The cluster must at least meet the following conditions
//  1. The cluster has enabled the auditlog sidecar
//  2. The default sidecar image tag has been configured
//
// In addition, at least one of the conditions must be met
//  1. the feature flag is enabled
//  2. the auto upgrade annotation is set to true
//
// Note: once the audit log sidecar image annotation is set, it will not be deleted.
// This is to avoid the rollback of the sidecar version, which will cause TIDB to restart
//
// Note: The user can manually remove the auto upgrade annotation and feature flag if they want to lock auditlog sidecar tag.
func (m *configManager) UpdateTiDBAuditLogSidecarAnno(ctx context.Context, cluster *cloudv1.Cluster) {

	logger := m.logger(ctx).WithValues("cluster", cluster.Name)

	if cluster.Spec.Cluster.TiDB.Audit == nil || !cluster.Spec.Cluster.TiDB.Audit.Enable ||
		m.rootConfig == nil || m.rootConfig.ClusterController.Audit.DefaultTag == "" {
		logger.Info("skip update audit log sidecar image tag annotation for cluster, because audit log sidecar is not enabled or default tag is not configured")
		return
	}

	if !config.IsFeatureEnabled(m.rootConfig.ClusterController.Audit.DefaultTagFeature, cluster.GetTenantID(), cluster.GetProjectID()) &&
		util.GetAnnotation(cluster, consts.AuditLogSidecarAutoUpgrade) != "true" {
		logger.Info("skip update audit log sidecar image tag annotation for cluster, because the feature flag is disabled and auto upgrade annotation is not set to true")
		return
	}

	util.SetAnnotation(cluster, consts.AuditLogSidecarImageTag, m.rootConfig.ClusterController.Audit.DefaultTag)
	logger.Info("update audit log sidecar image tag annotation for cluster", "tag", m.rootConfig.ClusterController.Audit.DefaultTag)
}

func CheckIfIsTiDBVersionChanged(tc *v1alpha1.TidbCluster, revision *cloudv1.ConfigRevision) (bool, string, string) {

	if tc == nil || tc.Spec.TiDB == nil || revision == nil || revision.Spec.TiDB == nil {
		return false, "", ""
	}

	var tcVersion, desiredVersion string

	if version := pointer.StringDeref(tc.Spec.TiDB.Version, ""); version != "" {
		tcVersion = version
	} else {
		tcVersion = tc.Spec.Version
	}

	if version := pointer.StringDeref(revision.Spec.TiDB.Tag, ""); version != "" {
		desiredVersion = version
	} else {
		desiredVersion = revision.Spec.Version
	}

	return tcVersion != desiredVersion, tcVersion, desiredVersion
}