package cluster

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

	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/tidbcloud/config-service/sdk/api"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/common"
	"github.com/tidbcloud/infra-api/controllers/mixeddeploypd"
	"github.com/tidbcloud/infra-api/controllers/utils"
	"github.com/tidbcloud/infra-api/pkg/config"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/gitops"
	"github.com/tidbcloud/infra-api/pkg/k8s"
	"github.com/tidbcloud/infra-api/pkg/provider"
	"github.com/tidbcloud/infra-api/pkg/util"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/utils/pointer"
)

const (
	MB = 1024 * 1024

	// The minimum TiDB version that supports SEM
	minSemSupportedVersion = "v5.3.0"

	minAnalyzeVersion            = "v5.1.0"
	minAddIndexAccelerateVersion = "v6.3.0"

	defaultScheduler = "default-scheduler"

	annSkipTLSWhenConnectTiDB = "tidb.tidb.pingcap.com/skip-tls-when-connect-tidb"

	// When rolling update tidb, the operator will wait min-ready-seconds to restart the next pod
	// after the current one is ready
	// It is because the operator should ensure that LB has added the restarted pod before restarting
	// the next one.
	// 180 is an arbitrary duration for AWS LB.
	annTiDBMinReadySeconds = "tidb.pingcap.com/tidb-min-ready-seconds"
	defaultMinReadySeconds = "180"
)

func (c *ClusterContext) computePDConfig(cc *api.DescribeClusterConfigOutput) *v1alpha1.PDConfigWraper {
	conf := v1alpha1.NewPDConfig()
	if pdConfig, ok := getComponentConfig(cc, api.ComponentPD); ok {
		conf.GenericConfig.MP = pdConfig.Config.MP
	}
	conf.Set("dashboard.public-path-prefix", fmt.Sprintf("/proxy/cluster/%s/pd/dashboard", c.Target.Name))
	conf.Set("security.cert-allowed-cn", []string{util.FnvHash(c.Target.PrefixedId())})
	if c.Target.Spec.Cluster.TiFlash.Replicas > 0 {
		conf.Set("replication.enable-placement-rules", true)
	}
	return conf
}

func (c *ClusterContext) computeTiDBConfig(cc *api.DescribeClusterConfigOutput, provider *cloudv1.Provider, tidbResource corev1.ResourceRequirements) *v1alpha1.TiDBConfigWraper {
	conf := v1alpha1.NewTiDBConfig()
	if dbConfig, ok := getComponentConfig(cc, api.ComponentTiDB); ok {
		conf.GenericConfig.MP = dbConfig.Config.MP
	}
	if provider.Spec.Type == consts.ProviderTypeAWS && addIndexAccelerateSupported(cc.Version, tidbResource.Requests.Cpu()) {
		conf.Set("temp-dir", k8s.TiDBAddIndexAccelerateGp3StorageDefaultDirPath)
	}
	conf.Set("security.cluster-verify-cn", []string{util.FnvHash(c.Target.PrefixedId())})
	conf.Set("performance.tcp-keep-alive", true)
	return conf
}

func (c *ClusterContext) computeTiKVConfig(cc *api.DescribeClusterConfigOutput) *v1alpha1.TiKVConfigWraper {
	conf := v1alpha1.NewTiKVConfig()
	if kvConfig, ok := getComponentConfig(cc, api.ComponentTiKV); ok {
		conf.GenericConfig.MP = kvConfig.Config.MP
	}
	conf.Set("security.cert-allowed-cn", []string{util.FnvHash(c.Target.PrefixedId())})
	conf.Set("security.encryption.data-encryption-method", "aes256-ctr")
	conf.Set("security.encryption.data-key-rotation-period", "7d")
	conf.Set("security.encryption.master-key.type", "kms")
	conf.Set("security.encryption.master-key.key-id", c.KmsKeyId)
	conf.Set("security.encryption.master-key.region", c.KmsRegion)
	return conf
}

func (c *ClusterContext) computeTiFlashConfig(cc *api.DescribeClusterConfigOutput) *v1alpha1.TiFlashConfigWraper {
	conf := v1alpha1.NewTiFlashConfig()
	if tiflashConfig, ok := getComponentConfig(cc, api.ComponentTiFlash); ok {
		conf.Common.MP = tiflashConfig.Config.MP
	}
	if learnerConfig, ok := getComponentConfig(cc, api.ComponentTiFlashLearner); ok {
		conf.Proxy.MP = learnerConfig.Config.MP
	}

	conf.Common.Set("security.cert_allowed_cn", []string{util.FnvHash(c.Target.PrefixedId())})
	conf.Proxy.Set("security.cert-allowed-cn", []string{util.FnvHash(c.Target.PrefixedId())})
	conf.Proxy.Set("security.encryption.data-encryption-method", "aes256-ctr")
	conf.Proxy.Set("security.encryption.data-key-rotation-period", "7d")
	conf.Proxy.Set("security.encryption.master-key.type", "kms")
	conf.Proxy.Set("security.encryption.master-key.key-id", c.KmsKeyId)
	conf.Proxy.Set("security.encryption.master-key.region", c.KmsRegion)
	return conf
}

func (c *ClusterContext) addExtraServiceEnvVars(version string, envs []corev1.EnvVar) []corev1.EnvVar {
	if semSupported(version) {
		return envs
	}

	envs = append(envs, []corev1.EnvVar{
		{
			Name:  "TIDB_OVERRIDE_ENDPOINT",
			Value: fmt.Sprintf("%s:%d", getPrivilegedTiDBAddress(consts.ClusterPrefix, c.Target.PrefixedId()), 4000),
		},
		{
			Name:  "TIDB_OVERRIDE_STATUS_ENDPOINT",
			Value: fmt.Sprintf("%s:%d", getPrivilegedTiDBAddress(consts.ClusterPrefix, c.Target.PrefixedId()), 10080),
		},
	}...)

	return envs
}

type ClusterContext struct {
	// Provider-agnostic context of a reconciliation.
	// Subject to change during the reconciliation.
	Target         *cloudv1.Cluster
	Network        *cloudv1.Network
	Provider       *cloudv1.Provider
	ConfigProvider gitops.ConfigProvider
	KmsRegion      *string
	KmsKeyId       *string

	// Provider-specific external dependencies.
	provider.Cluster
	provider.GitOps
}

func GetIndexAccelerate(provider *cloudv1.Provider, tc *v1alpha1.TidbCluster) []v1alpha1.StorageVolume {
	if provider.Spec.Type == consts.ProviderTypeAWS && addIndexAccelerateSupported(tc.Spec.Version, tc.Spec.TiDB.ResourceRequirements.Requests.Cpu()) {
		return []v1alpha1.StorageVolume{
			{
				Name:             k8s.TiDBAddIndexAccelerateGp3StorageDefaultDirName,
				StorageClassName: pointer.String(k8s.TiDBAddIndexAccelerateGp3StorageClassName),
				StorageSize:      k8s.TiDBAddIndexAccelerateGp3StorageDefaultSize,
				MountPath:        k8s.TiDBAddIndexAccelerateGp3StorageDefaultDirPath,
			},
		}
	}
	return nil
}

func (c *ClusterContext) AssembleTidbClusterValues(pdMixedDeployConfig *cloudv1.MixedDeployPDWithTiKV) (*v1alpha1.TidbCluster, error) {
	cc, err := c.GetClusterConfig(context.TODO(), c.Target)
	if err != nil {
		return nil, errors.Annotate(err, "error get cluster config")
	}
	defaultPVReclaimPolicy := corev1.PersistentVolumeReclaimRetain
	pdConfig, ok := getComponentConfig(cc, api.ComponentPD)
	if !ok {
		return nil, errors.New("no PD configuration provided")
	}
	tidbConfig, ok := getComponentConfig(cc, api.ComponentTiDB)
	if !ok {
		return nil, errors.New("no TiDB configuration provided")
	}
	tikvConfig, ok := getComponentConfig(cc, api.ComponentTiKV)
	if !ok {
		return nil, errors.New("no TiKV configuration provided")
	}

	topo := c.Target.Spec.Cluster
	pdResource, err := c.ConfigProvider.GetResourceSpec(c.Provider, &topo.PD)
	if err != nil {
		return nil, errors.Annotate(err, "get pd resources spec")
	}
	tidbResource, err := c.ConfigProvider.GetResourceSpec(c.Provider, &topo.TiDB)
	if err != nil {
		return nil, errors.Annotate(err, "get tidb resources spec")
	}

	tikvResource, err := c.ConfigProvider.GetResourceSpec(c.Provider, &topo.TiKV)
	if err != nil {
		return nil, errors.Annotate(err, "get tikv resources spec")
	}

	// a simple utility to override the version during development
	setVersion := func(tag *string) *string {
		return tag
		// v := "v5.3.0"; return &v
	}

	// TODO: most of the configurations of TC should come from gitops
	tc := &v1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: c.Target.PrefixedId(),
			Name:      consts.ClusterPrefix,
			Annotations: map[string]string{
				annSkipTLSWhenConnectTiDB: "",
				api.ConfigVersionKey:      strconv.FormatUint(cc.ConfigMeta.ConfigVersion, 10),
				api.ConfigIDKey:           cc.ConfigMeta.ID,
				api.ConfigCommitKey:       cc.ConfigMeta.GitCommitID,
				annTiDBMinReadySeconds:    defaultMinReadySeconds,
			},
		},
		Spec: v1alpha1.TidbClusterSpec{
			PD: &v1alpha1.PDSpec{
				ComponentSpec: v1alpha1.ComponentSpec{
					Affinity:     utils.Affinity(cloudv1.ComponentPD),
					NodeSelector: c.GetComponentNodeLabels(cloudv1.ComponentPD),
					Tolerations:  utils.Toleration(cloudv1.ComponentPD),
					HostNetwork:  pointer.BoolPtr(true),
					Version:      setVersion(pdConfig.Image.Tag),
					Env: c.addExtraServiceEnvVars(cc.Version, []corev1.EnvVar{
						{
							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",
								},
							},
						},
					}),
				},
				ResourceRequirements:     pdResource.Resources,
				Replicas:                 int32(topo.PD.Replicas),
				BaseImage:                pdConfig.Image.BaseImage,
				MaxFailoverCount:         pointer.Int32Ptr(0),
				Service:                  nil,
				StorageClassName:         firstNonEmpty(topo.PD.StorageClassName, pdResource.StorageClass),
				Config:                   c.computePDConfig(cc),
				MountClusterClientSecret: pointer.BoolPtr(true),
			},
			TiDB: &v1alpha1.TiDBSpec{
				ComponentSpec: v1alpha1.ComponentSpec{
					Affinity:     utils.Affinity(cloudv1.ComponentTiDB),
					NodeSelector: c.GetComponentNodeLabels(cloudv1.ComponentTiDB),
					Tolerations:  utils.Toleration(cloudv1.ComponentTiDB),
					HostNetwork:  pointer.BoolPtr(true),
					Annotations: map[string]string{
						consts.AnnSysctlInit: consts.AnnSysctlInitVal,
					},
					PodSecurityContext: &corev1.PodSecurityContext{
						Sysctls: []corev1.Sysctl{
							{
								Name:  "net.ipv4.tcp_keepalive_intvl",
								Value: "75",
							},
							{
								Name:  "net.ipv4.tcp_keepalive_time",
								Value: "300",
							},
						},
					},
					Version: setVersion(tidbConfig.Image.Tag),
				},
				ResourceRequirements: tidbResource.Resources,
				Replicas:             int32(topo.TiDB.Replicas),
				BaseImage:            tidbConfig.Image.BaseImage,
				BinlogEnabled:        pointer.BoolPtr(false),
				MaxFailoverCount:     pointer.Int32Ptr(0),
				SeparateSlowLog:      pointer.BoolPtr(true),
				TLSClient: &v1alpha1.TiDBTLSClient{
					Enabled: true,
				},
				SlowLogTailer: &v1alpha1.TiDBSlowLogTailerSpec{
					ResourceRequirements: 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),
						},
					},
				},
				Plugins: []string{},
				Config:  c.computeTiDBConfig(cc, c.Provider, tidbResource.Resources),
			},
			TiKV: &v1alpha1.TiKVSpec{
				ComponentSpec: v1alpha1.ComponentSpec{
					Affinity:     utils.Affinity(cloudv1.ComponentTiKV),
					NodeSelector: c.GetComponentNodeLabels(cloudv1.ComponentTiKV),
					Tolerations:  utils.Toleration(cloudv1.ComponentTiKV),
					// cross region require tikv use hostnetwork, cdc will use pod ip to access tikv
					HostNetwork: pointer.BoolPtr(true),
					PodSecurityContext: &corev1.PodSecurityContext{
						Sysctls: []corev1.Sysctl{
							{
								Name:  "net.core.somaxconn",
								Value: "32768",
							},
						},
					},
					Annotations: map[string]string{
						consts.AnnSysctlInit: consts.AnnSysctlInitVal,
					},
					Version: setVersion(tikvConfig.Image.Tag),
				},
				ResourceRequirements: tikvResource.Resources,
				BaseImage:            tikvConfig.Image.BaseImage,
				Replicas:             int32(topo.TiKV.Replicas),
				Privileged:           pointer.BoolPtr(false),
				MaxFailoverCount:     pointer.Int32Ptr(0),
				StorageClassName:     firstNonEmpty(topo.TiKV.StorageClassName, tikvResource.StorageClass),
				Config:               c.computeTiKVConfig(cc),
			},
			Helper: &v1alpha1.HelperSpec{
				Image: pointer.StringPtr(config.GetBusyboxImage()),
			},
			Version:              cc.Version,
			PVReclaimPolicy:      &defaultPVReclaimPolicy,
			ImagePullPolicy:      corev1.PullIfNotPresent,
			ConfigUpdateStrategy: v1alpha1.ConfigUpdateStrategyRollingUpdate,
			TLSCluster: &v1alpha1.TLSCluster{
				Enabled: true,
			},
			SchedulerName: defaultScheduler,
			Timezone:      "UTC",
			// Disable global hostNetwork configuration and enable it explicitly when necessary
			// Currently pd, tidb and lightning have host network enabled
			// while tiflash, extra-tidb, discovery has disabled host network
			HostNetwork:                pointer.BoolPtr(false),
			EnableDynamicConfiguration: pointer.BoolPtr(true),
		},
	}
	tidbStorageVolume := GetIndexAccelerate(c.Provider, tc)
	tc.Spec.TiDB.StorageVolumes = append(tc.Spec.TiDB.StorageVolumes, tidbStorageVolume...)
	tcName := types.NamespacedName{
		Namespace: tc.Namespace,
		Name:      tc.Name,
	}.String()
	mixedDeployPDWithTiKV, err := mixeddeploypd.ShouldMixedDeployPDWithTiKV(&topo, "creating", c.Target.Namespace, tcName, c.Provider, time.Now(), pdMixedDeployConfig)
	if err != nil {
		return nil, err
	}
	if mixedDeployPDWithTiKV {
		err = mixeddeploypd.ChangeTCForMixedDeploy(tc, c.Target, c.ConfigProvider, c.Provider, c.GetComponentNodeLabels)
		if err != nil {
			return nil, err
		}
	}

	if err := c.Overlay(tc); err != nil {
		return nil, err
	}

	return tc, nil
}

func (c *ClusterContext) OverlayTiFlash(tc *v1alpha1.TidbCluster) error {
	wantTiflash := c.Target.Spec.Cluster.TiFlash.Replicas > 0
	haveTiflash := tc.Spec.TiFlash != nil
	topo := c.Target.Spec.Cluster
	switch {
	case wantTiflash && !haveTiflash:
		// tiflash is needed but we do not have tiflash in tc yet, enable it
		cc, err := c.GetClusterConfig(context.TODO(), c.Target)
		if err != nil {
			return errors.Annotate(err, "error get cluster config")
		}
		tiflashConfig, ok := getComponentConfig(cc, api.ComponentTiFlash)
		if !ok {
			return errors.New("no TiFlash configuration provided")
		}
		tiflashResource, err := c.ConfigProvider.GetResourceSpec(c.Provider, &c.Target.Spec.Cluster.TiFlash)
		if err != nil {
			return errors.Annotate(err, "get tiflash resources spec")
		}
		tc.Spec.TiFlash = &v1alpha1.TiFlashSpec{
			ComponentSpec: v1alpha1.ComponentSpec{
				Affinity:           utils.Affinity(cloudv1.ComponentTiFlash),
				NodeSelector:       c.GetComponentNodeLabels(cloudv1.ComponentTiFlash),
				Tolerations:        utils.Toleration(cloudv1.ComponentTiFlash),
				HostNetwork:        pointer.BoolPtr(true),
				PodSecurityContext: &corev1.PodSecurityContext{},
				Version:            tiflashConfig.Image.Tag,
			},
			MaxFailoverCount:     pointer.Int32Ptr(0),
			ResourceRequirements: tiflashResource.Resources,
			BaseImage:            tiflashConfig.Image.BaseImage,
			Replicas:             int32(c.Target.Spec.Cluster.TiFlash.Replicas),
			LogTailer: &v1alpha1.LogTailerSpec{
				ResourceRequirements: 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),
					},
				},
			},
			Config: c.computeTiFlashConfig(cc),
			StorageClaims: []v1alpha1.StorageClaim{{
				Resources: corev1.ResourceRequirements{
					Requests: map[corev1.ResourceName]resource.Quantity{
						corev1.ResourceStorage: *tiflashResource.Resources.Requests.Storage(),
					},
				},
				StorageClassName: firstNonEmpty(topo.TiFlash.StorageClassName, tiflashResource.StorageClass),
			}},
		}
		c.overlayTiFlashKmsCredentials(tc)
	case !wantTiflash && haveTiflash:
		// tiflash is not expected but we have tiflash in tc spec, disable it
		// Note setting it as nil, operator WILL NOT handle it.
		// we scale replicas as 0, and this WILL NOT delete the STS.
		// So once the tiflash is enable, spec relate to STS(storage class can not be changed).
		if tc.Spec.TiFlash != nil {
			tc.Spec.TiFlash.Replicas = 0
		}
	}
	return nil
}

func (r *reconcileContext) getInitializer(tcName types.NamespacedName) (v1alpha1.TidbInitializer, error) {
	init := v1alpha1.TidbInitializer{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: tcName.Namespace,
			Name:      tcName.Name,
		},
	}
	return init, r.shootClient.Get(r.ctx, tcName, &init)
}

func (r *reconcileContext) initSucceeded(tc *v1alpha1.TidbCluster) (bool, *v1alpha1.TidbInitializer, error) {
	initName := getInitializerName(tc.Name)
	init, getErr := r.getInitializer(types.NamespacedName{Namespace: tc.Namespace, Name: initName})
	if getErr != nil {
		// skip no initializer cluster
		if apierrors.IsNotFound(getErr) {
			return false, nil, nil
		}
		return false, nil, getErr
	}
	return init.Status.JobStatus.Succeeded != 0, &init, nil
}

// overlaySuspendActions applies the suspend actions to the tidbcluster resource
func (r *reconcileContext) overlaySuspendActions(tc *v1alpha1.TidbCluster) error {
	if r.Target == nil || tc == nil {
		return nil
	}
	topo := r.Target.Spec.Cluster
	// The flow for the suspending the cluster is as below:
	// - If `suspendAction` is set in the spec, mark the cluster as
	//   suspending first by adding the `ComponentSuspended` condition
	// - Update TC with the `suspendAction` in the next reconcile,
	//   in case that the user may resume the cluster immediately
	//   after pausing the cluster
	// - Mark the cluster as suspended by updating the `ComponentSuspended`
	//   condition to `True` after the cluster is suspended
	// - Remove the `ComponentSuspended` condition when the user resumes the cluster
	if tc.Spec.TiDB != nil && !r.Target.TiDBStatusNil() &&
		(r.Target.Status.DB.TiDB.Suspending() || r.Target.Status.DB.TiDB.Suspended()) {
		tc.Spec.TiDB.SuspendAction = convertSuspendActions(topo.TiDB.GetSuspendActions(topo.SuspendAction))
	}
	if tc.Spec.PD != nil && !r.Target.PDStatusNil() &&
		(r.Target.Status.DB.PD.Suspending() || r.Target.Status.DB.PD.Suspended()) {
		tc.Spec.PD.SuspendAction = convertSuspendActions(topo.PD.GetSuspendActions(topo.SuspendAction))
	}
	if tc.Spec.TiKV != nil && !r.Target.TiKVStatusNil() &&
		(r.Target.Status.DB.TiKV.Suspending() || r.Target.Status.DB.TiKV.Suspended()) {
		tc.Spec.TiKV.SuspendAction = convertSuspendActions(topo.TiKV.GetSuspendActions(topo.SuspendAction))
	}
	if topo.TiFlash.Replicas > 0 && tc.Spec.TiFlash != nil && !r.Target.TiFlashStatusNil() &&
		(r.Target.Status.DB.TiFlash.Suspending() || r.Target.Status.DB.TiFlash.Suspended()) {
		tc.Spec.TiFlash.SuspendAction = convertSuspendActions(topo.TiFlash.GetSuspendActions(topo.SuspendAction))
	}
	return nil
}

func (r *reconcileContext) overlayAnnotations(tc *v1alpha1.TidbCluster) error {
	if tc.Annotations == nil {
		tc.Annotations = map[string]string{}
	}

	tc.Annotations[annTiDBMinReadySeconds] = defaultMinReadySeconds

	return nil
}

// overlayTags apply tag labels to the tidbcluster resource
func (r *reconcileContext) overlayTags(tc *v1alpha1.TidbCluster) error {
	if tc.Labels == nil {
		tc.Labels = map[string]string{}
	}
	tags := cloudv1.TagsFromLabels(r.Target.Labels)
	for k, v := range tags {
		tc.Labels[k] = v
	}
	// add sub-resource labels so that we can back reference to cluster object from tc
	for k, v := range r.SubResourceLabels() {
		tc.Labels[k] = v
	}
	return nil
}

func (r *reconcileContext) overlayReplicas(tc *v1alpha1.TidbCluster) error {
	topo := r.Target.Spec.Cluster
	tc.Spec.TiDB.Replicas = int32(topo.TiDB.Replicas)
	tc.Spec.PD.Replicas = int32(topo.PD.Replicas)
	tc.Spec.TiKV.Replicas = int32(topo.TiKV.Replicas)
	if topo.TiFlash.Replicas > 0 && tc.Spec.TiFlash != nil {
		tc.Spec.TiFlash.Replicas = int32(topo.TiFlash.Replicas)
	}
	return nil
}

func (r *reconcileContext) overlayTiDBEndpoint(tc *v1alpha1.TidbCluster) error {
	serviceExternalTrafficPolicyTypeLocal := corev1.ServiceExternalTrafficPolicyTypeLocal
	tc.Spec.TiDB.Service = &v1alpha1.TiDBServiceSpec{
		ExposeStatus: pointer.BoolPtr(true),
	}
	if r.Target.Spec.SqlEndpoints.Public.IsEnabled() {
		tc.Spec.TiDB.Service.ServiceSpec = v1alpha1.ServiceSpec{
			Type:                     corev1.ServiceTypeLoadBalancer,
			LoadBalancerSourceRanges: r.Target.Spec.SqlEndpoints.Public.LoadBalancerSourceRanges,
			Port:                     r.Target.Spec.SqlEndpoints.Public.Port,
		}
		tc.Spec.TiDB.Service.ExternalTrafficPolicy = &serviceExternalTrafficPolicyTypeLocal
		if r.ProviderType == consts.ProviderTypeAWS {
			serviceAnnotation := map[string]string{
				"service.beta.kubernetes.io/aws-load-balancer-cross-zone-load-balancing-enabled": "true",
				"service.beta.kubernetes.io/aws-load-balancer-internet-facing":                   "0.0.0.0/0",
				"service.beta.kubernetes.io/aws-load-balancer-type":                              "nlb",
			}

			tags := common.ConcatenateTags(cloudv1.TagsFromLabels(r.Target.Labels),
				map[string]string{
					common.Cluster:   r.Target.Name,
					common.Component: common.PrivateLoadbalance,
					common.UsedBy:    common.CustomerTiDB,
				})
			serviceTag := cloudv1.TagStringForAWS(tags)

			serviceAnnotation[K8SLabelKeyForAWSLoadBalanceResourceTag] = serviceTag
			tc.Spec.TiDB.Service.Annotations = serviceAnnotation
		} else if r.ProviderType == consts.ProviderTypeGCP {
			serviceAnnotation := map[string]string{"cloud.google.com/load-balancer-type": "External"}
			tc.Spec.TiDB.Service.Annotations = serviceAnnotation
		}
	} else {
		tc.Spec.TiDB.Service.ServiceSpec = v1alpha1.ServiceSpec{
			Type: corev1.ServiceTypeClusterIP,
			Port: r.Target.Spec.SqlEndpoints.Public.Port,
		}
	}
	if r.providerGardenerSpec.GetEnableGracefulShutdownWebhook() {
		if tc.Spec.TiDB.ComponentSpec.AdditionalContainers == nil {
			tc.Spec.TiDB.ComponentSpec.AdditionalContainers = []corev1.Container{
				{
					Name:            consts.HealthCheckerContainerName,
					Image:           config.GetHealthCheckerImage(),
					ImagePullPolicy: corev1.PullAlways,
					Command:         []string{"/health-checker", fmt.Sprintf("-allow-cidr=%s", r.Network.Spec.CIDR)},
					Ports: []corev1.ContainerPort{
						{
							Name:          "health-checker",
							ContainerPort: int32(8080),
							Protocol:      corev1.ProtocolTCP,
						},
					},
				},
			}
		}

		tidbAnnotations := tc.Spec.TiDB.ComponentSpec.Annotations
		if tidbAnnotations == nil {
			tidbAnnotations = map[string]string{}
		}
		tidbAnnotations[consts.AnnGracefulShutdownTiDB] = "true"
		switch r.ProviderType {
		case consts.ProviderTypeGCP:
			tidbAnnotations[consts.AnnDeletePodWaitTimeout] = "40s"
		case consts.ProviderTypeAWS:
			tidbAnnotations[consts.AnnDeletePodWaitTimeout] = "3m"
		default:
			tidbAnnotations[consts.AnnDeletePodWaitTimeout] = "40s"
		}
		tc.Spec.TiDB.ComponentSpec.Annotations = tidbAnnotations

		tc.Spec.TiDB.Service.AdditionalPorts = []corev1.ServicePort{
			{
				Name:       "health-checker",
				Protocol:   corev1.ProtocolTCP,
				Port:       8080,
				TargetPort: intstr.FromInt(8080),
			},
		}
	}
	return nil
}

func (c *ClusterContext) overlayTiFlashKmsCredentials(tc *v1alpha1.TidbCluster) {
	if c.Network.Spec.Provider == consts.ProviderTypeGCP {
		if c.Target.Spec.Cluster.TiFlash.Replicas > 0 && tc.Spec.TiFlash != nil {
			tc.Spec.TiFlash.Env = append(tc.Spec.TiFlash.Env,
				corev1.EnvVar{
					Name: consts.AWS_ACCESS_KEY_ENV_VAR,
					ValueFrom: &corev1.EnvVarSource{
						SecretKeyRef: &corev1.SecretKeySelector{
							LocalObjectReference: corev1.LocalObjectReference{Name: gcpKMSSecretName},
							Key:                  consts.AWS_ACCESS_KEY_ENV_VAR,
						},
					},
				},
				corev1.EnvVar{
					Name: consts.AWS_SECRET_KEY_ENV_VAR,
					ValueFrom: &corev1.EnvVarSource{
						SecretKeyRef: &corev1.SecretKeySelector{
							LocalObjectReference: corev1.LocalObjectReference{Name: gcpKMSSecretName},
							Key:                  consts.AWS_SECRET_KEY_ENV_VAR,
						},
					},
				},
			)
		}
	}
}

func (r *reconcileContext) overlayTiKVKmsCredentials(tc *v1alpha1.TidbCluster) error {
	// FIXME: GCP TiKV still need to access AWS KMS and static credentials are required
	if r.ProviderType == consts.ProviderTypeGCP {
		tc.Spec.TiKV.Env = append(tc.Spec.TiKV.Env,
			corev1.EnvVar{
				Name: consts.AWS_ACCESS_KEY_ENV_VAR,
				ValueFrom: &corev1.EnvVarSource{
					SecretKeyRef: &corev1.SecretKeySelector{
						LocalObjectReference: corev1.LocalObjectReference{Name: gcpKMSSecretName},
						Key:                  consts.AWS_ACCESS_KEY_ENV_VAR,
					},
				},
			},
			corev1.EnvVar{
				Name: consts.AWS_SECRET_KEY_ENV_VAR,
				ValueFrom: &corev1.EnvVarSource{
					SecretKeyRef: &corev1.SecretKeySelector{
						LocalObjectReference: corev1.LocalObjectReference{Name: gcpKMSSecretName},
						Key:                  consts.AWS_SECRET_KEY_ENV_VAR,
					},
				},
			},
		)
	}

	return nil
}

func (r *reconcileContext) overlayBillingLabels(tcv *v1alpha1.TidbCluster) error {
	if tcv.Labels == nil {
		tcv.Labels = make(map[string]string)
	}

	conf := HomogenousClusterConfig{}
	for _, cp := range r.Target.GetComponents() {
		scp := HomogenousComponent{
			Provider:     r.ProviderType,
			Replicas:     cp.GetReplicas(),
			Region:       r.Network.Spec.Region,
			InstanceType: cp.GetInstanceType(),
		}
		if size, ok := cp.GetStorageSize(); ok {
			scp.Disks = []Disk{
				{
					DiskType: string("ssd_network"),
					Capacity: size.ScaledValue(resource.Giga),
				},
			}
		}
		switch cp.GetComponentType() {
		case cloudv1.ComponentTiKV:
			conf.TiKV = scp
		case cloudv1.ComponentTiFlash:
			conf.TiFlash = scp
		case cloudv1.ComponentTiDB:
			conf.TiDB = scp
		}
	}
	data, err := json.Marshal(conf)
	if err != nil {
		return errors.Annotate(err, "failed to marshal config")
	}
	if tcv.Annotations == nil {
		tcv.Annotations = make(map[string]string)
	}
	tcv.Annotations[consts.BillingTiDBConfig] = string(data)
	return nil
}

// convertSuspendActions converts the cloudv1.SuspendAction to v1alpha1.SuspendAction
func convertSuspendActions(action *cloudv1.SuspendAction) *v1alpha1.SuspendAction {
	if action == nil {
		return nil
	}
	return &v1alpha1.SuspendAction{SuspendStatefulSet: action.SuspendCompute}

}

func generateConfigOutput(cluster *cloudv1.Cluster, cc *cloudv1.ClusterConfig) (*api.DescribeClusterConfigOutput, error) {
	componentConfig, err := generateComponentOutput(cluster, cc)
	if err != nil {
		return nil, err
	}
	output := api.DescribeClusterConfigOutput{
		Version: cc.Spec.Version,
		ConfigMeta: &api.ConfigMeta{
			ID:            cc.ObjectMeta.Name,
			GitCommitID:   cc.Spec.ConfigMeta.GitCommitID,
			ConfigVersion: cc.Spec.ConfigMeta.ConfigVersion,
		},
		Components: componentConfig,
	}

	return &output, nil
}

func getComponentConf(cc *cloudv1.ClusterConfig, comp cloudv1.ComponentType) *cloudv1.Configuration {
	switch comp {
	case cloudv1.ComponentTiKV:
		return cc.Spec.TiKV
	case cloudv1.ComponentPD:
		return cc.Spec.PD
	case cloudv1.ComponentTiDB:
		return cc.Spec.TiDB
	case cloudv1.ComponentTiFlash:
		return cc.Spec.TiFlash
	default:
		return nil
	}
}

// a convenience function for overriding the image during development
func getBaseImage(baseImage string, _ cloudv1.ComponentType) string {
	return baseImage
	// return "pingcap/" + string(ct)
}

func generateComponentOutput(cluster *cloudv1.Cluster, cc *cloudv1.ClusterConfig) ([]api.ComponentOutput, error) {
	var componentOutputs []api.ComponentOutput
	for _, component := range cluster.GetComponents() {
		componentConf := getComponentConf(cc, component.GetComponentType())
		if componentConf == nil {
			continue
		}

		co := api.ComponentOutput{
			Image: api.Image{
				BaseImage: getBaseImage(*componentConf.BaseImage, component.GetComponentType()),
				Tag:       componentConf.Tag,
			},
			Component: api.ComponentName(component.GetComponentType()),
			Config:    api.NewTomlConfig(componentConf.Config.MP),
		}

		componentOutputs = append(componentOutputs, co)
	}
	return componentOutputs, nil
}

func (r *reconcileContext) getClusterConfig() (*api.DescribeClusterConfigOutput, error) {
	if r.EnableGitOpsV2 {
		cc, err := r.Gitops.DescribeClusterConfiguration(r.Target)
		if err != nil {
			return nil, err
		}

		output, err := generateConfigOutput(r.Target, cc)
		return output, err
	} else {
		var comps []api.ComponentInput
		for _, comp := range r.Target.GetComponents() {
			comps = append(comps, api.ComponentInput{
				Component:    api.ComponentName(comp.GetComponentType()),
				InstanceType: fmt.Sprintf("%s.%s", r.ProviderType, comp.GetInstanceType()),
			})
			if comp.GetComponentType() == cloudv1.ComponentTiFlash {
				comps = append(comps, api.ComponentInput{
					Component:    api.ComponentTiFlashLearner,
					InstanceType: fmt.Sprintf("%s.%s", r.ProviderType, comp.GetInstanceType()),
				})
			}
		}
		tenantId := utils.ParsingLegacyId(r.Target, "tags.tidbcloud.com/tenant")
		if tenantId == nil {
			dummyId := uint64(0)
			tenantId = &dummyId
		}
		clusterId := utils.ParsingLegacyId(r.Target, "tags.tidbcloud.com/cluster")
		return r.ConfigService.DescribeClusterConfiguration(&api.DescribeClusterConfigInput{
			TenantID:   tenantId,
			ClusterID:  clusterId,
			Components: comps,
		})
	}
}

func getComponentConfig(cc *api.DescribeClusterConfigOutput, name api.ComponentName) (*api.ComponentOutput, bool) {
	// since each component has only one kind now, ignore instanceType and simply get by name
	// TODO: instanceType should be considered if for heterogeneous cluster

	for _, comp := range cc.Components {
		if comp.Component == name {
			return &comp, true
		}
	}
	return nil, false
}

type HomogenousClusterConfig struct {
	TiKV    HomogenousComponent `json:"tikv"`
	TiDB    HomogenousComponent `json:"tidb"`
	TiFlash HomogenousComponent `json:"tiflash"`
}

type HomogenousComponent struct {
	Provider     string `json:"provider"`
	Replicas     int    `json:"replicas"`
	Region       string `json:"region"`
	InstanceType string `json:"instanceType"`
	Disks        []Disk `json:"disks"`
}

type Disk struct {
	DiskType string `json:"diskType"`
	Capacity int64  `json:"capactity"`
}

func firstNonEmpty(lhs string, rhs string) *string {
	if lhs != "" {
		return pointer.String(lhs)
	}

	return pointer.String(rhs)
}