package configchange

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/go-logr/logr"
	"github.com/google/uuid"
	opv2 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/samber/lo"
	v1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	config2 "github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/controllers/common"
	reconcilerr "github.com/tidbcloud/infra-provider/pkg/controllers/common/error"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	sdk "github.com/tidbcloud/infra-provider/pkg/gitops-sdk"
	"github.com/tidbcloud/infra-provider/pkg/gitops-sdk/revision"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	metricsctrl "github.com/tidbcloud/infra-provider/pkg/metrics/controller"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/util/config"
)

const (
	Finalizer = "configchange.cloud.pingcap.com/finalizer"
	tcName    = "db"
)

var (
	// componentsManagedByConfigChange is component type list whose config managed by ConfigChange
	//componentsManagedByConfigChange = []cloudv1.ComponentType{
	//	//cloudv1.ComponentPD, // currently, the config of PD is always managed by cluster-controller
	//	cloudv1.ComponentTiProxy, cloudv1.ComponentTiFlash,
	//	cloudv1.ComponentTiKV, cloudv1.ComponentTiDB, cloudv1.ComponentTiCDC,
	//}
	// rollingRestartComponents
	// NOTE: config of Tiflash component is composited of TWO config fields(Tiflash and TiflashLearner) in ConfigRevision.Spec,
	// but there is only one component in TidbCluster.Spec.
	rollingRestartComponents = []cloudv1.ComponentType{
		// currently, the config of PD is always managed by cluster-controller, and config change of PD is always made by
		// HTTP API of pd, regardless of ConfigChange.Spec.ChangeStrategy.
		cloudv1.ComponentPD,
		// currently, config of TiProxy is always updated in InPlace strategy. So we don't and shouldn't to rolling restart it to apply config.
		cloudv1.ComponentTiProxy,
		cloudv1.ComponentTiFlash, cloudv1.ComponentTiFlashLearner, // tiflash and tiflashLearner have independent configurations, but belong to the same deployment
		cloudv1.ComponentTiKV, cloudv1.ComponentTiDB, cloudv1.ComponentTiCDC,
	}
	hotReloadComponents = []cloudv1.ComponentType{
		cloudv1.ComponentPD,
		cloudv1.ComponentTiProxy, // config of tiproxy is always updated in InPlace strategy
		cloudv1.ComponentTiFlashLearner,
		// tiflash supports auto reload config every two seconds by default
		cloudv1.ComponentTiFlash,
		cloudv1.ComponentTiKV,
		// confirmed with PM, tidb don't support changing *config* dynamically, ONLY variable can be changed dynamically
		//cloudv1.ComponentTiDB,
		// we don't support hotReload for ticdc for now.
		//cloudv1.ComponentTiCDC,
		// tiproxy is not supported to HotReload by ConfigChange for now
		//cloudv1.ComponentTiProxy,
	}

	// RequeueAfterWhenCCNotDone is the requeueAfter time when config change is not done
	// 30 seconds is the default
	RequeueAfterWhenCCNotDone = time.Second * 30

	RequeueAfterWhenConditionNotMet = time.Second * 30

	// DefaultTTL is the default TTL for Completed ConfigChanges, and can be overridden by annotation AnnoCCOverwrittenTTL
	DefaultTTL = time.Hour * 24 * 30
)

// ConfigApplier is used for applying ConfigRevision to cluster.
type ConfigApplier interface {
	InitSteps(ctx context.Context, cc *cloudv1.ConfigChange)
	RegisterStepAction(re *ConfigChangeReconciler)
}

type StepAction func(ctx context.Context, rx *ReconcileContext) error

// ConfigChangeReconciler reconciles a ClusterConfig object
type ConfigChangeReconciler struct {
	Client         client.Client
	cachedProvider dataplane.EnvProviderManager
	EventRecorder  record.EventRecorder

	revisioner revision.Interface
	g          sdk.CCInterface
	Log        logr.Logger
	RootConfig config2.RootConfig

	ActionRegistry  map[cloudv1.ConfigChangeStepName]StepAction
	ApplierRegistry map[cloudv1.ConfigChangeStrategy]ConfigApplier
}

type Option func(r *ConfigChangeReconciler)

func NewConfigChangeReconciler(mgr ctrl.Manager, cachedProvider dataplane.EnvProviderManager, log logr.Logger, rootConfig config2.RootConfig, opts ...Option) *ConfigChangeReconciler {
	r := &ConfigChangeReconciler{
		Log:           log,
		Client:        mgr.GetClient(),
		EventRecorder: mgr.GetEventRecorderFor("configchange-controller"),

		cachedProvider: cachedProvider,
		revisioner:     revision.NewRevisioner(mgr.GetClient()),
		g:              &sdk.Gitops{Cli: mgr.GetClient()},
		RootConfig:     rootConfig,

		ActionRegistry:  map[cloudv1.ConfigChangeStepName]StepAction{},
		ApplierRegistry: map[cloudv1.ConfigChangeStrategy]ConfigApplier{},
	}
	for _, opt := range opts {
		opt(r)
	}

	// register applier
	r.ApplierRegistry[cloudv1.ConfigChangeStrategyHotReload] = newHotReloadApplier(r.RootConfig)
	r.ApplierRegistry[cloudv1.ConfigChangeStrategyRollingRestart] = newRollingRestartApplier()
	r.ApplierRegistry[cloudv1.ConfigChangeStrategyChangeCMOnly] = newChangeCMOnlyApplier()
	r.ApplierRegistry[cloudv1.ConfigChangeStrategyEnsureUpdateToTC] = newEnsureUpdateToTCApplier()

	// register action
	r.RegisterAction(cloudv1.CCStepUpdateCR2Cluster, r.startConfigChangeIfNeed)
	r.RegisterAction(cloudv1.CCStepUpdateCR2TC, r.ensureConfigUpdatedToTC)
	r.RegisterAction(cloudv1.CCStepUpdateConfig2CM, r.ensureConfigUpdatedToCM)
	r.RegisterAction(cloudv1.CCStepEnsureTCReady, ensureTCsReady)
	for _, applier := range r.ApplierRegistry {
		applier.RegisterStepAction(r)
	}
	log.Info("registered actions", "actions", lo.Keys(r.ActionRegistry))
	return r
}

type ReconcileContext struct {
	Ctx context.Context

	RemoteClient client.Client
	LocalClient  client.Client

	ConfigChange         *cloudv1.ConfigChange
	Cluster              *cloudv1.Cluster // Maybe nil, if cluster is deleted.
	LastConfigRevision   *cloudv1.ConfigRevision
	TargetConfigRevision *cloudv1.ConfigRevision
	Applier              ConfigApplier

	Logger logr.Logger
}

func (r *ConfigChangeReconciler) RegisterAction(name cloudv1.ConfigChangeStepName, action StepAction) {
	if _, ok := r.ActionRegistry[name]; ok {
		panic(fmt.Errorf("action(%s) is already registered", name))
	}
	r.ActionRegistry[name] = action
}

// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=configchanges,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=configchanges/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=configchanges/finalizers,verbs=update
func (r *ConfigChangeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ctrl.Result, err error) {
	start := time.Now()
	logger := log.FromContext(ctx).
		WithValues(
			"reconcile-id", uuid.New().String(),
			"name", req.NamespacedName,
		)
	ctx = log.IntoContext(ctx, logger)
	defer func() {
		metricsctrl.ReconcileDuration.WithLabelValues("configchange").Observe(time.Since(start).Seconds())

		params := []interface{}{"take", time.Since(start)}
		if err == nil {
			params = append(params, "result", result)
		}
		logger.Info("finish reconcile", params...)
	}()

	logger.Info("start reconcile ConfigChange")
	cc := &cloudv1.ConfigChange{}
	if err = r.Client.Get(ctx, req.NamespacedName, cc); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}
	defer func() {
		if err == nil {
			r.EventRecorder.Event(cc, corev1.EventTypeNormal, "ReconcileSuccess", "")
		} else {
			r.EventRecorder.Event(cc, corev1.EventTypeWarning, "ReconcileFail", err.Error())
		}
	}()
	logger = logger.WithValues("cluster_name", cc.Spec.ClusterName)

	if cc.Spec.Paused != nil && *cc.Spec.Paused {
		logger.Info("skip, ConfigChange is paused")
		return ctrl.Result{}, nil
	}
	if cc.Done() {
		if err := r.unsetConfigChangeAnnoFromCluster(ctx, cc); err != nil {
			return ctrl.Result{}, err
		}
		if controllerutil.RemoveFinalizer(cc, Finalizer) {
			logger.Info("cc is done, remove finalizer")
			if err := r.Client.Update(ctx, cc); err != nil {
				return ctrl.Result{}, fmt.Errorf("update configChange to remove finalizer: %w", err)
			}
		}
		if err := r.deleteCCIfExpired(ctx, cc); err != nil {
			return ctrl.Result{}, fmt.Errorf("try to delete expired CC: %w", err)
		}
		logger.Info("skip, ConfigChange is done")
		return ctrl.Result{}, nil
	}

	// ensure finalizer if the object is not yet deleted
	if cc.DeletionTimestamp.IsZero() {
		if err := common.EnsureFinalizers(ctx, r.Client, cc, Finalizer); err != nil {
			return ctrl.Result{}, err
		}
	}

	// deal with cluster //
	cluster := &cloudv1.Cluster{}
	if err = r.Client.Get(ctx, types.NamespacedName{Name: cc.Spec.ClusterName, Namespace: cc.Namespace}, cluster); err != nil {
		if errors.IsNotFound(err) {
			if cc.Done() { // ignore if cc is done
				return ctrl.Result{}, nil
			}

			msg := fmt.Sprintf("Canceled: cluster not found, name=%s", cc.Spec.ClusterName)
			logger.Info(msg)
			return ctrl.Result{}, setConfigChangeAsCompleted(ctx, r.Client, cc, msg)
		}
		return ctrl.Result{}, fmt.Errorf("get cluster: %w", err)
	}
	if runningCCName := cluster.CurrentConfigChangeName(); runningCCName != "" && runningCCName != cc.Name {
		msg := fmt.Sprintf("another ConfigChange is running, name=%s", runningCCName)
		logger.Info(msg)
		return ctrl.Result{}, setConfigChangeAsFailed(ctx, r.Client, cc, msg)
	}
	// deal CC with cluster creation, pause/resume, scaling, deletion
	{
		// mark CC failed if the cluster is being deleted
		if cluster.DeletionTimestamp != nil && !cluster.DeletionTimestamp.IsZero() {
			logger.Info("Canceled: cluster is being deleted, set CC to Completed")
			return ctrl.Result{}, setConfigChangeAsCompleted(ctx, r.Client, cc, "cluster is being deleted")
		}
		// requeue if the cluster is creating
		if cluster.Status.Phase == "" || cluster.Status.Phase == cloudv1.ClusterPhasePending || cluster.Status.Phase == cloudv1.ClusterPhaseCreating {
			logger.Info("cluster is creating, wait next retry")
			return ctrl.Result{RequeueAfter: RequeueAfterWhenConditionNotMet}, nil
		}
		// pausing/paused: changes on TC will still be applied even if the cluster is pausing or paused, so it doesn't matter
		// resume: the behavior depends on when the changes of TC are applied.
		//      Changes applied before the resume operation, all components will be resumed normally and not be affected by ConfigChange.
		//      Changes applied after the resume operation, some pods may be re-created after it has been created, because
		//      ConfigChange will annotate them with RestartTiDBAnnotation.
		// scale: the scale operation doesn't disturb ConfigChange
	}

	remoteCli, err := r.cachedProvider.GetClient(ctx, cluster)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("get dataplane cli: %w", err)
	}

	// retrieve ConfigRevisions //
	desiredCr := &cloudv1.ConfigRevision{}
	if err := r.Client.Get(ctx, types.NamespacedName{Namespace: cc.Namespace, Name: cc.Spec.ConfigRevisionName}, desiredCr); err != nil {
		return ctrl.Result{}, fmt.Errorf("get desired ConfigRevision: %w", err)
	}
	lastCrKey := types.NamespacedName{Namespace: cc.Namespace, Name: cluster.Spec.ConfigRevisionRef.Name}
	if cc.Status.LastConfigRevisionName != "" {
		lastCrKey = types.NamespacedName{Namespace: cc.Namespace, Name: cc.Status.LastConfigRevisionName}
	}
	lastCr := &cloudv1.ConfigRevision{}
	if err := r.Client.Get(ctx, lastCrKey, lastCr); err != nil {
		return ctrl.Result{}, fmt.Errorf("get current ConfigRevision: %w", err)
	}

	// build the applier //
	applier, ok := r.ApplierRegistry[cc.Spec.ChangeStrategy]
	if !ok {
		logger.Error(fmt.Errorf("invalid applier"), "", "strategy", cc.Spec.ChangeStrategy)
		return ctrl.Result{}, setConfigChangeAsFailed(ctx, r.Client, cc, fmt.Sprintf("invalid strategy: %s", cc.Spec.ChangeStrategy))
	}

	// init the config change //
	if cc.Uninitialized() {
		logger.Info("init ConfigChange")
		if err := cc.Init(ctx, lastCr, desiredCr, applier.InitSteps); err != nil {
			return ctrl.Result{}, fmt.Errorf("init ConfigChange: %w", err)
		}

		if !cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
			// mark ConfigChange as failed if it tries to hot-reload tiflash which don't has annotation AnnTiflashMountCMInTiflashContainer in TC spec
			// TODO(chenfei): remove this code block after all tcs have annotation AnnTiflashMountCMInTiflashContainer
			tc := &opv1alpha1.TidbCluster{}
			if err := remoteCli.Get(ctx, types.NamespacedName{Namespace: cluster.PrefixedId(), Name: tcName}, tc); err != nil {
				return ctrl.Result{}, fmt.Errorf("get tc: %w", err)
			}
			if cc.Spec.ChangeStrategy == cloudv1.ConfigChangeStrategyHotReload &&
				lo.Contains(cc.Status.ChangedComponents, cloudv1.ComponentTiFlash) &&
				(tc.Spec.TiFlash != nil && !tc.Spec.TiFlash.DoesMountCMInTiflashContainer()) {
				return ctrl.Result{}, setConfigChangeAsFailed(ctx, r.Client, cc, "hot-reload tiflash of TC which don't has annotation AnnTiflashMountCMInTiflashContainer")
			}
		}

		// todo(chenfei): send notification and wait a configurable moment to continue
		if err := sendConfigChangeStartEv(ctx, cc, cluster); err != nil {
			logger.Error(err, "send event to clinic failed, ignore")
		}
		if err := r.Client.Status().Update(ctx, cc); err != nil {
			return ctrl.Result{}, fmt.Errorf("update configChange status: %w", err)
		}
	}

	// build the reconcile context
	rx := &ReconcileContext{
		Ctx:          ctx,
		RemoteClient: remoteCli,
		LocalClient:  r.Client,

		ConfigChange:         cc,
		LastConfigRevision:   lastCr,
		TargetConfigRevision: desiredCr,
		Cluster:              cluster,
		Applier:              applier,

		Logger: logger,
	}

	return r.reconcileLoop(ctx, rx)
}

// SetupWithManager sets up the controller with the Manager.
func (r *ConfigChangeReconciler) SetupWithManager(mgr ctrl.Manager, pds ...predicate.Predicate) error {
	// add index to ConfigChange object, used for list ConfigChange in Cluster
	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cloudv1.ConfigChange{}, "spec.clusterName", func(obj client.Object) []string {
		ccObj := obj.(*cloudv1.ConfigChange)
		return []string{ccObj.Spec.ClusterName}
	}); err != nil {
		return err
	}

	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.ConfigChange{}, builder.WithPredicates(pds...)).
		WithOptions(controller.Options{RateLimiter: k8s.RetryLimiter}).
		Complete(r)
}

func (r *ConfigChangeReconciler) startConfigChangeIfNeed(ctx context.Context, rx *ReconcileContext) error {
	logger := rx.Logger
	cc := rx.ConfigChange
	if cc.Status.Phase == cloudv1.ConfigChangePhaseRunning {
		logger.Info("cc is already running")
		return nil
	}

	if err := cc.Run(ctx); err != nil {
		return fmt.Errorf("run ConfigChange: %w", err)
	}
	cc.UpdateStep(ctx, cloudv1.CCStepUpdateCR2Cluster, cloudv1.CCStepStatusCompleted,
		fmt.Sprintf("updated %s to cluster(%s)", cc.Spec.ConfigRevisionName, cc.Spec.ClusterName))
	// update ConfigUpdateStrategy of TC to InPlace
	if err := r.setConfigUpdateStrategy(ctx, rx); err != nil {
		return fmt.Errorf("set configUpdateStrategy: %w", err)
	}
	// annotating cluster to prevent run multiple ConfigChange on same cluster in parallel
	if err := r.updateConfigRevision2Cluster(ctx, rx); err != nil {
		return fmt.Errorf("annotate cluster: %w", err)
	}
	// updating ConfigChange after updating cluster to make startConfigChangeIfNeed idempotent
	if err := r.Client.Status().Update(ctx, cc); err != nil {
		return fmt.Errorf("update ConfigChange: %w", err)
	}

	return nil
}

func (r *ConfigChangeReconciler) reconcileLoop(ctx context.Context, rx *ReconcileContext) (ctrl.Result, error) {
	oldCC := rx.ConfigChange.DeepCopy()

	for _, each := range rx.ConfigChange.Status.Steps {
		if each.Status == cloudv1.CCStepStatusCompleted {
			continue
		}

		action, ok := r.ActionRegistry[each.Name]
		if !ok {
			return ctrl.Result{}, setConfigChangeAsFailed(ctx, rx.LocalClient, rx.ConfigChange, "can't find action")
		}

		if err := action(ctx, rx); err != nil {
			return ctrl.Result{}, fmt.Errorf("execute action(%s): %w", each.Name, err)
		}
		if each.Status != cloudv1.CCStepStatusCompleted {
			rx.Logger.Info("step is NOT Completed, break", "step", each.Name)
			break
		}
	}

	// refresh phase
	rx.ConfigChange.RefreshPhase(ctx)
	if err := sendConfigChangeEv(ctx, oldCC, rx.ConfigChange, rx.Cluster); err != nil {
		rx.Logger.Error(err, "send event to clinic failed, ignore")
	}
	if err := rx.LocalClient.Status().Update(ctx, rx.ConfigChange); err != nil {
		return ctrl.Result{}, fmt.Errorf("update configChange status: %w", err)
	}

	if rx.ConfigChange.Done() {
		rx.Logger.Info("cc is done")
		return ctrl.Result{}, nil
	} else {
		rx.Logger.Info("cc is NOT done, requeue")
		return ctrl.Result{Requeue: true, RequeueAfter: RequeueAfterWhenCCNotDone}, nil
	}
}

func (r *ConfigChangeReconciler) setConfigUpdateStrategy(ctx context.Context, rx *ReconcileContext) error {
	remoteCli := rx.RemoteClient

	if rx.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		// Because it's already set to HotReload.
		rx.Logger.Info("no need to update configUpdateStrategy for tidb operator v2")
		return nil
	}

	tc := &opv1alpha1.TidbCluster{}
	if err := remoteCli.Get(ctx, types.NamespacedName{Namespace: rx.Cluster.PrefixedId(), Name: tcName}, tc); err != nil {
		return fmt.Errorf("get tc: %w", err)
	}

	if err := r.patchConfigUpdateStrategy(ctx, remoteCli, tc); err != nil {
		return fmt.Errorf("patch configUpdateStrategy for default tc: %w", err)
	}

	for _, extraGroup := range rx.Cluster.GetExtraGroups() {
		if extraGroup.Spec.TiDB.Replicas == 0 {
			rx.Logger.Info("filter out zero replicas extra group", "groupName", extraGroup.Name)
			continue
		}

		extraGroupNamer := namer.NewNamer().
			WithTiDBGroupName(extraGroup.Name).
			WithComponent(string(cloudv1.ComponentTiDB))
		extraGroupTCName := extraGroupNamer.GetTiDBName()
		extraGroupTC := &opv1alpha1.TidbCluster{}
		if err := remoteCli.Get(ctx, types.NamespacedName{Namespace: rx.Cluster.PrefixedId(), Name: extraGroupTCName}, extraGroupTC); err != nil {
			return fmt.Errorf("get extragroup tc, group=%s, tc=%s: %w", extraGroup.Name, extraGroupTCName, err)
		}
		if err := r.patchConfigUpdateStrategy(ctx, remoteCli, extraGroupTC); err != nil {
			return fmt.Errorf("patch configUpdateStrategy for extraGroupTC(%s): %w", extraGroupTCName, err)
		}
	}

	return nil
}

func (r *ConfigChangeReconciler) patchConfigUpdateStrategy(ctx context.Context, remoteCli client.Client, tc *opv1alpha1.TidbCluster) error {
	logger := log.FromContext(ctx)

	logger.Info("try to update tc.spec.configUpdateStrategy to InPlace", "tcName", tc.Name)
	// enable MergeFromWithOptimisticLock to prevent update conflict
	patch := client.MergeFromWithOptions(tc.DeepCopy(), &client.MergeFromWithOptimisticLock{})
	tc.Spec.ConfigUpdateStrategy = opv1alpha1.ConfigUpdateStrategyInPlace
	return remoteCli.Patch(ctx, tc, patch)
}

// updateConfigRevision2Cluster update Cluster.Spec.ConfigRevisionRef.Name, and that may cause some components restarts
// if their image or version have changed.
func (r *ConfigChangeReconciler) updateConfigRevision2Cluster(ctx context.Context, rx *ReconcileContext) error {
	// enable MergeFromWithOptimisticLock to prevent update conflict
	patch := client.MergeFromWithOptions(rx.Cluster.DeepCopy(), &client.MergeFromWithOptimisticLock{})

	// If the cluster version being changed, we must set a version Guard to ensure the version change is allowed when
	// cluster-controller try to sync ConfigRevision to TC. For more about version Guard, follow this link:
	// https://pingcap.feishu.cn/docx/PE0rdt5SFo9u0ux0mMKcJ5ZYnjf#NxHUdHQOyoMUNKxe54NcSY1gn6d
	needVersionGuard := rx.TargetConfigRevision.Spec.Version != rx.LastConfigRevision.Spec.Version
	rx.Cluster.UpdateConfigRevision(rx.TargetConfigRevision, needVersionGuard)
	rx.Cluster.SetConfigChange(rx.ConfigChange.Name)

	rx.Logger.Info("try to update cluster.configRevision",
		"old", rx.Cluster.Spec.ConfigRevisionRef.Name, "new", rx.ConfigChange.Spec.ConfigRevisionName)
	if err := r.Client.Patch(ctx, rx.Cluster, patch); err != nil {
		rx.Logger.Error(err, "patch cluster for config change failed")
		return err
	}
	return nil
}

func (r *ConfigChangeReconciler) ensureConfigUpdatedToTC(ctx context.Context, rx *ReconcileContext) error {
	cc := rx.ConfigChange
	remoteCli := rx.RemoteClient
	key := types.NamespacedName{Namespace: rx.Cluster.PrefixedId(), Name: tcName}

	var annotations map[string]string
	if rx.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		var tc opv2.Cluster
		if err := remoteCli.Get(ctx, key, &tc); err != nil {
			return fmt.Errorf("get tc: %w", err)
		}
		annotations = tc.Annotations
	} else {
		tc := &opv1alpha1.TidbCluster{}
		if err := remoteCli.Get(ctx, key, tc); err != nil {
			return fmt.Errorf("get tc: %w", err)
		}
		annotations = tc.Annotations
	}

	// wait cluster-controller to update config to tc
	// this annotation be set in /pkg/gitops-sdk/manager.go:patchTiDBCluster
	if annotations[cloudv1.ConfigRevisionIdKey] != rx.TargetConfigRevision.Name {
		msg := fmt.Sprintf("wait CR to be updated to TC, cur: %s, target: %s", annotations[cloudv1.ConfigRevisionIdKey], rx.TargetConfigRevision.Name)
		rx.Logger.Info(msg)
		cc.UpdateStep(ctx, cloudv1.CCStepUpdateCR2TC, cloudv1.CCStepStatusRunning, msg)
		if err := r.Client.Status().Update(ctx, cc); err != nil {
			return err
		}
		return reconcilerr.New(RequeueAfterWhenCCNotDone, nil)
	}

	cc.UpdateStep(ctx, cloudv1.CCStepUpdateCR2TC, cloudv1.CCStepStatusCompleted, "ConfigRevision has been updated to TC")
	return r.Client.Status().Update(ctx, cc)
}

func (r *ConfigChangeReconciler) ensureConfigUpdatedToCM(ctx context.Context, rx *ReconcileContext) error {
	cc := rx.ConfigChange
	remoteCli := rx.RemoteClient
	key := types.NamespacedName{Namespace: rx.Cluster.PrefixedId(), Name: tcName}
	useOperatorV2 := rx.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2)

	var (
		annotations map[string]string
		tcV1        opv1alpha1.TidbCluster
		tcV2        opv2.Cluster
	)
	if useOperatorV2 {
		if err := remoteCli.Get(ctx, key, &tcV2); err != nil {
			return fmt.Errorf("get tc: %w", err)
		}
		annotations = tcV2.Annotations
	} else {
		if err := remoteCli.Get(ctx, key, &tcV1); err != nil {
			return fmt.Errorf("get tc: %w", err)
		}
		annotations = tcV1.Annotations
	}

	// wait cluster-controller to update config to tc
	// this annotation be set in /pkg/gitops-sdk/manager.go:patchTiDBCluster
	if annotations[cloudv1.ConfigRevisionIdKey] != rx.TargetConfigRevision.Name {
		msg := fmt.Sprintf("wait CR to be updated to TC, cur: %s, target: %s", annotations[cloudv1.ConfigRevisionIdKey], rx.TargetConfigRevision.Name)
		rx.Logger.Info(msg)
		cc.UpdateStep(ctx, cloudv1.CCStepUpdateConfig2CM, cloudv1.CCStepStatusRunning, msg)
		if err := r.Client.Status().Update(ctx, cc); err != nil {
			return err
		}
		return reconcilerr.New(RequeueAfterWhenCCNotDone, nil)
	}

	// TC is paused means reconciliation of this TC is paused, the corresponding ConfigMaps will not be updated during this time.
	if tcV1.Spec.Paused || tcV2.Spec.Paused {
		msg := fmt.Sprintf("tc(%s) is paused, retry next time", tcName)
		rx.Logger.Info(msg)
		cc.UpdateStep(ctx, cloudv1.CCStepUpdateConfig2CM, cloudv1.CCStepStatusRunning, msg)
		if err := r.Client.Status().Update(ctx, cc); err != nil {
			return err
		}
		return reconcilerr.New(RequeueAfterWhenCCNotDone, nil)
	}

	var configNotWritten2CM []cloudv1.ComponentType
	// ensures the config has been updated to ConfigMap
	for _, comp := range cc.Status.ChangedComponents {
		comp = getDeploymentComponent(comp)
		if comp == cloudv1.ComponentTiProxy && rx.Cluster.GetComponent(comp).GetReplicas() == 0 {
			r.Log.Info("tiproxy is not deployed, skip")
			continue
		}
		if (comp == cloudv1.ComponentTiFlash || comp == cloudv1.ComponentTiFlashLearner) && rx.Cluster.GetComponent(comp).GetReplicas() == 0 {
			r.Log.Info("tiflash is not deployed, skip")
			continue
		}
		// since ticdc is NOT managed by Cluster, we must check it whether exists in TC
		if comp == cloudv1.ComponentTiCDC && (!useOperatorV2 && (tcV1.Spec.TiCDC == nil || tcV1.Spec.TiCDC.Replicas == 0)) {
			r.Log.Info("ticdc is not deployed, skip")
			continue
		}

		if useOperatorV2 {
			// In v2, we only need to check if the component group is synced.
			synced, err := areComponentGroupsSynced(ctx, remoteCli, key, comp, rx.TargetConfigRevision.Name)
			if err != nil {
				return fmt.Errorf("check if component(%s) groups are synced: %w", comp, err)
			}
			if !synced {
				configNotWritten2CM = append(configNotWritten2CM, comp)
			}
			continue
		}

		sts := &v1.StatefulSet{}
		if err := getAdvanceStatefulSet(ctx, remoteCli, types.NamespacedName{Namespace: key.Namespace, Name: memberName(tcName, comp)}, sts); err != nil {
			return fmt.Errorf("get sts(%s): %w", comp, err)
		}

		configMapName := FindConfigMapVolume(&sts.Spec.Template.Spec, func(name string) bool {
			return strings.HasPrefix(name, memberName(tcName, comp))
		})
		r.Log.Info("get config map", "component", comp, "configMapName", configMapName)
		if configMapName == "" {
			return fmt.Errorf("can't find configmap for component: %s", comp)
		}

		cm := &corev1.ConfigMap{}
		if err := remoteCli.Get(ctx, types.NamespacedName{Namespace: key.Namespace, Name: configMapName}, cm); err != nil {
			return fmt.Errorf("get cm(%s): %w", configMapName, err)
		}
		writtenCompleted := false
		if comp == cloudv1.ComponentTiFlash {
			tiflashConfigKey := "config_templ.toml"
			tiflashLearnerKey := "proxy_templ.toml" // tiflashLearner is tiflashProxy
			writtenTiflashCompleted, err := compareConfBetweenCMAndCR(rx.TargetConfigRevision, cloudv1.ComponentTiFlash, cm, tiflashConfigKey)
			if err != nil {
				return err
			}
			writtenTiflashLearnerCompleted, err := compareConfBetweenCMAndCR(rx.TargetConfigRevision, cloudv1.ComponentTiFlashLearner, cm, tiflashLearnerKey)
			if err != nil {
				return err
			}
			writtenCompleted = writtenTiflashCompleted && writtenTiflashLearnerCompleted
		} else {
			configKeyInCM := "config-file"
			var err error
			writtenCompleted, err = compareConfBetweenCMAndCR(rx.TargetConfigRevision, comp, cm, configKeyInCM)
			if err != nil {
				return err
			}
		}
		if !writtenCompleted {
			configNotWritten2CM = append(configNotWritten2CM, comp)
		}
	}
	if len(configNotWritten2CM) > 0 {
		msg := fmt.Sprintf("configNotWritten2CM, components: %v", configNotWritten2CM)
		rx.Logger.Info(msg)
		cc.UpdateStep(ctx, cloudv1.CCStepUpdateConfig2CM, cloudv1.CCStepStatusRunning, msg)
		if err := r.Client.Status().Update(ctx, cc); err != nil {
			return err
		}
		return reconcilerr.New(RequeueAfterWhenCCNotDone, nil)
	}
	r.Log.Info("all config have been written to cm")

	cc.UpdateStep(ctx, cloudv1.CCStepUpdateConfig2CM, cloudv1.CCStepStatusCompleted, "config of components have been written to cm")
	return r.Client.Status().Update(ctx, cc)
}

// unsetConfigChangeAnnoFromCluster unsets ConfigChange annotation, indicating the cluster is config changing, from cluster
func (r *ConfigChangeReconciler) unsetConfigChangeAnnoFromCluster(ctx context.Context, cc *cloudv1.ConfigChange) error {
	logger := log.FromContext(ctx)

	cluster := &cloudv1.Cluster{}
	if err := r.Client.Get(ctx, types.NamespacedName{Name: cc.Spec.ClusterName, Namespace: cc.Namespace}, cluster); err != nil {
		if errors.IsNotFound(err) {
			logger.Info("cluster not found, skip", "cluster_name", cc.Spec.ClusterName)
			return nil
		}
		return fmt.Errorf("get cluster: %w", err)
	}

	// enable MergeFromWithOptimisticLock to prevent update conflict
	patch := client.MergeFromWithOptions(cluster.DeepCopy(), &client.MergeFromWithOptimisticLock{})
	if !cluster.UnsetConfigChange(cc.Name) {
		return nil
	}

	logger.Info("try to unset ConfigChange from Cluster")
	if err := r.Client.Patch(context.Background(), cluster, patch); err != nil {
		logger.Error(err, "patch cluster for unset ConfigChange")
		return err
	}
	return nil
}

func (r *ConfigChangeReconciler) deleteCCIfExpired(ctx context.Context, cc *cloudv1.ConfigChange) error {
	logger := log.FromContext(ctx)
	if !cc.Done() {
		return nil
	}

	if cc.Status.EndTime == nil || cc.Status.EndTime.IsZero() {
		logger.Info("endTime is empty, use now as endTime")
		cc.Status.EndTime = &metav1.Time{Time: time.Now()}
		return r.Client.Status().Update(ctx, cc)
	}

	completedAt := cc.Status.EndTime.Time
	ttlInAnno, _ := time.ParseDuration(cc.Annotations[cloudv1.AnnoCCOverwrittenTTL])
	ttl, _ := lo.Coalesce(ttlInAnno, DefaultTTL)
	if time.Since(completedAt) > ttl {
		logger.Info("try to delete expired ConfigChange", "name", cc.Name, "completedAt", completedAt.String(), "ttl", ttl)
		return r.Client.Delete(ctx, cc)
	}
	return nil
}

func setConfigChangeAsFailed(ctx context.Context, localClient client.Client, cc *cloudv1.ConfigChange, reason string) error {
	if err := cc.Fail(ctx, reason); err != nil {
		return err
	}
	return localClient.Status().Update(ctx, cc)
}

func setConfigChangeAsCompleted(ctx context.Context, localClient client.Client, cc *cloudv1.ConfigChange, reason string) error {
	if err := cc.Complete(ctx, reason); err != nil {
		return err
	}
	return localClient.Status().Update(ctx, cc)
}

func getConfFromCM(cm *corev1.ConfigMap, configKey string) (*config.GenericConfig, error) {
	confData := []byte(cm.Data[configKey])
	conf := config.New(map[string]interface{}{})
	if err := conf.UnmarshalTOML(confData); err != nil {
		return nil, fmt.Errorf("unmarshal conf from cm, confKey=%s: %w", configKey, err)
	}
	return conf, nil
}

// getConfFromCR gets conf from ConfigRevision
func getConfFromCR(cr *cloudv1.ConfigRevision, comp cloudv1.ComponentType) (*config.GenericConfig, error) {
	configuration := cr.CompConfByType(comp)
	if configuration == nil {
		return nil, fmt.Errorf("can't find configuration for component: %s", comp)
	}
	desiredConf := config.New(map[string]interface{}{})
	if configuration.Config != nil {
		desiredConf = config.New(configuration.Config.MP)
	}

	return desiredConf, nil
}

// compareConfBetweenCMAndCR compare configs between conf in ConfigMap and conf in ConfigRevision
// Since cluster-controller or other services may overlay or add some configuration items on ConfigRevision,
// we should compare them based on the desired configuration, that is ConfigRevision. To ensure the function works correctly
// even if some configuration items are added into ConfigMap by cluster-controller.
func compareConfBetweenCMAndCR(cr *cloudv1.ConfigRevision, comp cloudv1.ComponentType, cm *corev1.ConfigMap, configKeyInCM string) (equal bool, _ error) {
	currConf, err := getConfFromCM(cm, configKeyInCM)
	if err != nil {
		return false, fmt.Errorf("get conf from configmap, comp=%s: %w", comp, err)
	}
	desiredConf, err := getConfFromCR(cr, comp)
	if err != nil {
		return false, fmt.Errorf("get conf from configRevision, comp=%s: %w", comp, err)
	}

	changedKeys := compareConfBaseDesired(desiredConf, currConf)
	// Since PD [only sync instance config](/pkg/gitops-sdk/manager.go:222) to TC & ConfigMap, so we should filter them out from changedKeys.
	// Why not sync all config items to TC?
	//   Because cluster-level config changes is expected to DON'T restart PD, and if we sync them we break the expectation.
	// Why not add a feature-gate logic to sync cluster-level config only when the IndependentConfigChange is enabled?
	//   Because the logic syncing config to TC is implemented in gitops-sdk, and that is an independent go module, we can't do that.
	if comp == cloudv1.ComponentPD {
		// filter out cluster config keys, since them are not synced to TC.
		changedKeys = lo.Filter(changedKeys, func(k string, _ int) bool {
			ks := strings.SplitN(k, ".", 2)
			return !cloudv1.IsPDClusterConfig(ks[0])
		})
	}
	return len(changedKeys) == 0, nil
}