package cluster

import (
	"fmt"
	"strconv"
	"time"

	kmsapi "github.com/aws/aws-sdk-go-v2/service/kms"
	kmstypes "github.com/aws/aws-sdk-go-v2/service/kms/types"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	awskmsv1alpha1 "github.com/crossplane/provider-aws/apis/kms/v1alpha1"
	"github.com/pingcap/errors"
	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/utils/pointer"

	"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/pkg/consts"
	tidbclusterutil "github.com/tidbcloud/shared-provider/pkg/pingcap/util/tidbcluster"
	"github.com/tidbcloud/shared-provider/pkg/types"
	"github.com/tidbcloud/shared-provider/pkg/utils"
	"github.com/tidbcloud/shared-provider/pkg/utils/labels"
	overlayutil "github.com/tidbcloud/shared-provider/pkg/utils/overlay"
)

const (
	defaultUserID  = 1000
	defaultGroupID = 2000
)

const (
	MB = 1024 * 1024

	defaultWaitInterval = 10 * time.Second

	kmsPermissionFinishAnno = "kms.cloud.pingcap.com/finish"

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

	// nolint: lll // line is 238 characters
	initSQL = `SET GLOBAL tidb_analyze_version = 1;

SET GLOBAL tidb_redact_log = 1;

CREATE USER cloud_admin;
GRANT BACKUP_ADMIN, RESTORE_ADMIN ON *.* TO 'cloud_admin'@'%';
GRANT DASHBOARD_CLIENT on *.* TO 'cloud_admin'@'%';
GRANT SYSTEM_VARIABLES_ADMIN ON *.* TO 'cloud_admin'@'%';
GRANT CONNECTION_ADMIN ON *.* TO 'cloud_admin'@'%';

GRANT RESTRICTED_VARIABLES_ADMIN ON *.* TO 'cloud_admin'@'%';
GRANT RESTRICTED_STATUS_ADMIN ON *.* TO 'cloud_admin'@'%';
GRANT RESTRICTED_CONNECTION_ADMIN ON *.* TO 'cloud_admin'@'%';
GRANT RESTRICTED_USER_ADMIN ON *.* TO 'cloud_admin'@'%';
GRANT RESTRICTED_TABLES_ADMIN ON *.* TO 'cloud_admin'@'%';
GRANT RESTRICTED_REPLICA_WRITER_ADMIN ON *.* TO 'cloud_admin'@'%';

GRANT CREATE USER ON *.* TO 'cloud_admin'@'%';
GRANT RELOAD ON *.* TO 'cloud_admin'@'%';
GRANT PROCESS ON *.* TO 'cloud_admin'@'%';
GRANT SELECT, INSERT, UPDATE, DELETE ON mysql.* TO 'cloud_admin'@'%';
GRANT SELECT ON information_schema.* TO 'cloud_admin'@'%';
GRANT SELECT ON performance_schema.* TO 'cloud_admin'@'%';
GRANT SHOW DATABASES on *.* TO 'cloud_admin'@'%';

# Cross region requires these permissions, See https://docs.pingcap.com/tidb/stable/troubleshoot-ticdc/#when-the-sink-of-the-replication-downstream-is-tidb-or-mysql-what-permissions-do-users-of-the-downstream-database-need for more detail
GRANT SELECT ON *.* TO 'cloud_admin'@'%';
GRANT INDEX ON *.* TO 'cloud_admin'@'%';
GRANT INSERT ON *.* TO 'cloud_admin'@'%';
GRANT UPDATE ON *.* TO 'cloud_admin'@'%';
GRANT DELETE ON *.* TO 'cloud_admin'@'%';
GRANT CREATE ON *.* TO 'cloud_admin'@'%';
GRANT DROP ON *.* TO 'cloud_admin'@'%';
GRANT ALTER ON *.* TO 'cloud_admin'@'%';
GRANT CREATE VIEW ON *.* TO 'cloud_admin'@'%';

GRANT SHUTDOWN, CONFIG ON *.* TO 'cloud_admin'@'%';
REVOKE SHUTDOWN, CONFIG ON *.* FROM root;
`
)

var defaultPVReclaimPolicy = corev1.PersistentVolumeReclaimDelete

func (r *reconcileContext) createCluster() error {
	if err := cloudv1.EnsureClusterStorageClass(r.ctx, r.Client, r.provider, r.cluster); err != nil {
		return err
	}

	if !r.haveKms() {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "creating kms key"); err != nil {
			return err
		}
		if err := r.createKMS(); err != nil {
			return err
		}
	}

	if !r.haveTc() {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring tidb cluster"); err != nil {
			return err
		}
		if err := r.createTidbCluster(); err != nil {
			return err
		}
	}

	if !r.haveTi() {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring tidb initializer"); err != nil {
			return err
		}
		if err := r.createTidbInitializer(); err != nil {
			return err
		}
	}

	if !r.haveTm() {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring tidb monitor"); err != nil {
			return err
		}
		if err := r.createTidbMonitor(); err != nil {
			return err
		}
	}

	if r.needPublicEndpoint() {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring public endpoints"); err != nil {
			return err
		}
		if err := r.reconcilePublicEndpoint(); err != nil {
			return err
		}
	}

	if r.needPrivateLink() {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring private link"); err != nil {
			return err
		}
		if err := r.reconcilePrivateLink(); err != nil {
			return err
		}
	}

	if err := r.ensureSecurityEnhancedModeEnabled(); err != nil {
		return err
	}

	// NOTE: check TidbCluster is ready if any update is made to TidbCluster after `ensureSecurityEnhancedModeEnabled`.
	return r.observeCreation()
}

func (r *reconcileContext) createKMS() error {
	var err error

	// create key
	k := &awskmsv1alpha1.Key{
		ObjectMeta: metav1.ObjectMeta{
			Name: r.cluster.ClusterScopeName(),
		},
		Spec: awskmsv1alpha1.KeySpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: types.CrossplaneProviderConfigReferenceName,
				},
			},
			ForProvider: awskmsv1alpha1.KeyParameters{
				Region: r.awsRegion,
				Tags: []*awskmsv1alpha1.Tag{
					{
						TagKey: pointer.StringPtr("sa"),
						TagValue: pointer.StringPtr(fmt.Sprintf("system:serviceaccount:%s:%s",
							r.tcKey.Namespace, types.S3BackupServiceAccount)),
					},
				},
			},
		},
	}
	if err = r.createOrGetWithOwnerLabel(k); err != nil {
		return errors.Annotate(err, "creating kms key")
	}

	// wait KeyID
	if k.Status.AtProvider.KeyID == nil {
		r.V(4).Info("kms key not ready, requeuing")
		return &utils.RequeueAfterError{Cause: nil, RequeueAfter: defaultWaitInterval}
	}

	// disable key rotation
	if _, err = r.AwsClient.KMS().DisableKeyRotation(r.ctx, &kmsapi.DisableKeyRotationInput{
		KeyId: k.Status.AtProvider.KeyID,
	}); err != nil {
		return errors.Annotate(err, "disable key rotation")
	}

	// grant key permission to TiKV
	if _, err = r.AwsClient.KMS().CreateGrant(r.ctx, &kmsapi.CreateGrantInput{
		KeyId:            k.Status.AtProvider.KeyID,
		GranteePrincipal: &r.backupRole,
		Operations: []kmstypes.GrantOperation{
			kmstypes.GrantOperationEncrypt,
			kmstypes.GrantOperationDecrypt,
			kmstypes.GrantOperationGenerateDataKey,
			kmstypes.GrantOperationGenerateDataKeyWithoutPlaintext,
		},
	}); err != nil {
		return errors.Annotate(err, "grant key permission to TiKV")
	}

	if r.cluster.Annotations == nil {
		r.cluster.Annotations = map[string]string{}
	}
	r.cluster.Annotations[kmsPermissionFinishAnno] = "true"
	if err = r.InfraClient.Update(r.ctx, r.cluster); err != nil {
		return errors.Annotate(err, "update cluster kmsPermissionFinishAnno")
	}

	r.kmsKeyId = *k.Status.AtProvider.KeyID

	return nil
}

// nolint: funlen,gomnd // nolint
func (r *reconcileContext) createTidbCluster() error {
	dashboardEncryptionKey := cloudv1.RandomString(32)

	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-dashboard-session-secret", r.tcKey.Name),
			Namespace: r.tcKey.Namespace,
		},
		Data: map[string][]byte{
			"encryption_key": []byte(dashboardEncryptionKey),
		},
	}
	if err := r.createOrGetWithOwnerLabel(secret); err != nil {
		return errors.Annotate(err, "create dashboard session secret")
	}

	cc, err := r.getClusterConfig()
	if err != nil {
		r.InfraRecorder.Eventf(r.cluster, corev1.EventTypeWarning, "getClusterConfig", "%v", err)
		return errors.Annotate(err, "error get cluster config")
	}
	pdConfig, ok := getComponentConfig(cc, api.ComponentPD)
	if !ok {
		msg := "no PD configuration provided"
		r.InfraRecorder.Event(r.cluster, corev1.EventTypeWarning, "getComponentConfig", msg)
		return errors.New(msg)
	}
	tidbConfig, ok := getComponentConfig(cc, api.ComponentTiDB)
	if !ok {
		msg := "no TiDB configuration provided"
		r.InfraRecorder.Event(r.cluster, corev1.EventTypeWarning, "getComponentConfig", msg)
		return errors.New(msg)
	}

	topo := r.cluster.Spec.Cluster
	pdResource, err := r.ConfigProvider.GetResourceSpec(r.provider, &topo.PD)
	if err != nil {
		r.InfraRecorder.Eventf(r.cluster, corev1.EventTypeWarning, "GetPDResourceSpec", "%v", err)
		return errors.Annotate(err, "get pd resources spec")
	}
	tidbResource, err := r.ConfigProvider.GetResourceSpec(r.provider, &topo.TiDB)
	if err != nil {
		r.InfraRecorder.Eventf(r.cluster, corev1.EventTypeWarning, "GetTidbResourceSpec", "%v", err)
		return errors.Annotate(err, "get tidb resources spec")
	}
	size := r.cluster.Spec.Cluster.PD.StorageSize
	if !size.IsZero() {
		pdResource.Resources.Requests[corev1.ResourceStorage] = size
	}
	tc := &v1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      r.tcKey.Name,
			Namespace: r.tcKey.Namespace,
			Annotations: map[string]string{
				annSkipTLSWhenConnectTiDB: "",
				api.ConfigVersionKey:      strconv.FormatUint(cc.ConfigMeta.ConfigVersion, 10),
				api.ConfigIDKey:           cc.ConfigMeta.ID,
				api.ConfigCommitKey:       cc.ConfigMeta.GitCommitID,
			},
		},
		Spec: v1alpha1.TidbClusterSpec{
			PodSecurityContext: &corev1.PodSecurityContext{
				RunAsUser:  pointer.Int64Ptr(defaultUserID),
				RunAsGroup: pointer.Int64Ptr(defaultGroupID),
				FSGroup:    pointer.Int64Ptr(defaultGroupID),
			},
			Discovery: v1alpha1.DiscoverySpec{
				ResourceRequirements: corev1.ResourceRequirements{
					Requests: corev1.ResourceList{
						corev1.ResourceCPU:    *resource.NewMilliQuantity(20, resource.DecimalSI),
						corev1.ResourceMemory: *resource.NewQuantity(10*MB, resource.BinarySI),
					},
					Limits: corev1.ResourceList{
						corev1.ResourceCPU:    *resource.NewMilliQuantity(100, resource.DecimalSI),
						corev1.ResourceMemory: *resource.NewQuantity(50*MB, resource.BinarySI),
					},
				},
			},
			PD: &v1alpha1.PDSpec{
				ComponentSpec: v1alpha1.ComponentSpec{
					Version: pdConfig.Image.Tag,
					Env: []corev1.EnvVar{
						{
							Name: "DASHBOARD_SESSION_SECRET",
							ValueFrom: &corev1.EnvVarSource{
								SecretKeyRef: &corev1.SecretKeySelector{
									LocalObjectReference: corev1.LocalObjectReference{
										Name: fmt.Sprintf("%s-dashboard-session-secret", r.tcKey.Name),
									},
									Key: "encryption_key",
								},
							},
						},
					},
				},
				MaxFailoverCount:         pointer.Int32Ptr(0),
				ResourceRequirements:     pdResource.Resources,
				Replicas:                 int32(topo.PD.Replicas),
				BaseImage:                pdConfig.Image.BaseImage,
				StorageClassName:         firstNonEmpty(topo.PD.StorageClassName, pdResource.StorageClass),
				MountClusterClientSecret: pointer.BoolPtr(true),
				Config:                   r.computePDConfig(cc),
			},
			TiDB: &v1alpha1.TiDBSpec{
				ReadinessProbe: &v1alpha1.TiDBProbe{
					Type: pointer.String(v1alpha1.CommandProbeType),
				},
				ComponentSpec: v1alpha1.ComponentSpec{
					Annotations: map[string]string{
						consts.AnnSysctlInit: consts.AnnSysctlInitVal,
					},
					PodSecurityContext: &corev1.PodSecurityContext{
						RunAsUser:  pointer.Int64Ptr(defaultUserID),
						RunAsGroup: pointer.Int64Ptr(defaultGroupID),
						FSGroup:    pointer.Int64Ptr(defaultGroupID),
						Sysctls: []corev1.Sysctl{
							{
								Name:  "net.core.somaxconn",
								Value: "32768",
							},
							{
								Name:  "net.ipv4.tcp_keepalive_intvl",
								Value: "75",
							},
							{
								Name:  "net.ipv4.tcp_keepalive_time",
								Value: "300",
							},
						},
					},
					Version: tidbConfig.Image.Tag,
				},
				ResourceRequirements: tidbResource.Resources,
				Replicas:             int32(topo.TiDB.Replicas),
				BaseImage:            tidbConfig.Image.BaseImage,
				BinlogEnabled:        pointer.BoolPtr(false),
				MaxFailoverCount:     pointer.Int32Ptr(3),
				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),
						},
					},
				},
				Config: r.computeTiDBConfig(cc),
				Service: &v1alpha1.TiDBServiceSpec{
					ServiceSpec: v1alpha1.ServiceSpec{
						Type: corev1.ServiceTypeClusterIP,
					},
				},
			},
			Helper: &v1alpha1.HelperSpec{
				Image: pointer.StringPtr(utils.GetBusyboxImage()),
			},
			Version:         cc.Version,
			PVReclaimPolicy: &defaultPVReclaimPolicy,
			TLSCluster: &v1alpha1.TLSCluster{
				Enabled: true,
			},
			ConfigUpdateStrategy:       v1alpha1.ConfigUpdateStrategyRollingUpdate,
			EnableDynamicConfiguration: pointer.BoolPtr(true),
			Timezone:                   "UTC",
			TopologySpreadConstraints: []v1alpha1.TopologySpreadConstraint{
				{TopologyKey: "kubernetes.io/hostname"},
				{TopologyKey: "topology.kubernetes.io/zone"},
			},
		},
	}

	if err := r.overlayTiKV(tc); err != nil {
		return err
	}

	if err = r.overlayTiFlash(tc); err != nil {
		return err
	}

	if err = r.overlayTags(tc); err != nil {
		return err
	}

	if err = r.overlayReplicas(tc); err != nil {
		return err
	}

	if err = r.createTLSCertBetweenComponents(); err != nil {
		return err
	}

	if err = r.createTLSCertForMySQLClient(); err != nil {
		return err
	}

	if err = r.Client.Create(r.ctx, tc); err == nil {
		r.tc = tc
	} else if apierrors.IsAlreadyExists(err) {
		err = r.Cache.Get(r.ctx, r.tcKey, r.tc)
	}

	return err
}

func (r *reconcileContext) createTidbInitializer() error {
	initsql := initSQL

	ti := &v1alpha1.TidbInitializer{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: r.tiKey.Namespace,
			Name:      r.tiKey.Name,
		},
		Spec: v1alpha1.TidbInitializerSpec{
			Clusters: v1alpha1.TidbClusterRef{Name: r.tcKey.Name, Namespace: r.tcKey.Namespace},
			InitSql:  &initsql,
			Image:    utils.GetMysqlClientImage(),
		},
	}

	err := r.createOrGetWithOwnerLabel(ti)
	if err == nil {
		r.ti = ti
	}
	return err
}

// nolint: gomnd // nolint
func (r *reconcileContext) createTidbMonitor() error {
	tm := &v1alpha1.TidbMonitor{
		ObjectMeta: metav1.ObjectMeta{
			Name:      r.tmKey.Name,
			Namespace: r.tmKey.Namespace,
		},
		Spec: v1alpha1.TidbMonitorSpec{
			Clusters: []v1alpha1.TidbClusterRef{{Name: r.tcKey.Name, Namespace: r.tcKey.Namespace}},
			Prometheus: v1alpha1.PrometheusSpec{
				MonitorContainer: v1alpha1.MonitorContainer{
					BaseImage: utils.GetThirdPartyImage(utils.PrometheusImage),
					Version:   utils.PrometheusVersion,
					ResourceRequirements: corev1.ResourceRequirements{
						Requests: corev1.ResourceList{
							corev1.ResourceCPU:    *resource.NewMilliQuantity(80, resource.DecimalSI),
							corev1.ResourceMemory: *resource.NewQuantity(200*MB, resource.BinarySI),
						},
						Limits: corev1.ResourceList{
							corev1.ResourceCPU:    *resource.NewMilliQuantity(300, resource.DecimalSI),
							corev1.ResourceMemory: *resource.NewQuantity(800*MB, resource.BinarySI),
						},
					},
				},
				LogLevel:      "info",
				RetentionTime: pointer.StringPtr("2h"),
				RemoteWrite: []*v1alpha1.RemoteWriteSpec{
					{URL: r.RemoteWriteUrl},
				},
			},
			Reloader: v1alpha1.ReloaderSpec{
				MonitorContainer: v1alpha1.MonitorContainer{
					BaseImage: utils.GetBaseImage(utils.TiDBMonitorReloaderImage),
					Version:   utils.TiDBMonitorReloaderVersion,
					ResourceRequirements: corev1.ResourceRequirements{
						Requests: corev1.ResourceList{
							corev1.ResourceCPU:    *resource.NewMilliQuantity(20, resource.DecimalSI),
							corev1.ResourceMemory: *resource.NewQuantity(10*MB, resource.BinarySI),
						},
						Limits: corev1.ResourceList{
							corev1.ResourceCPU:    *resource.NewMilliQuantity(100, resource.DecimalSI),
							corev1.ResourceMemory: *resource.NewQuantity(50*MB, resource.BinarySI),
						},
					},
				},
			},
			Initializer: v1alpha1.InitializerSpec{
				MonitorContainer: v1alpha1.MonitorContainer{
					BaseImage: utils.GetBaseImage(utils.TiDBMonitorInitializerImage),
					Version:   r.tc.Spec.Version,
					ResourceRequirements: corev1.ResourceRequirements{
						Requests: corev1.ResourceList{
							corev1.ResourceCPU:    *resource.NewMilliQuantity(20, resource.DecimalSI),
							corev1.ResourceMemory: *resource.NewQuantity(10*MB, resource.BinarySI),
						},
						Limits: corev1.ResourceList{
							corev1.ResourceCPU:    *resource.NewMilliQuantity(100, resource.DecimalSI),
							corev1.ResourceMemory: *resource.NewQuantity(50*MB, resource.BinarySI),
						},
					},
				},
			},
			ExternalLabels: map[string]string{
				"cluster_id":    r.cluster.Labels[labels.TagLabelPrefix+"cluster"],
				"cluster":       fmt.Sprintf("sharedtier/eks/%s", r.KubeClusterName),
				"provider_type": r.providerType,
				"projectId":     r.cluster.Labels[labels.TagLabelPrefix+"billing-project-id"],
				"tenantId":      r.cluster.Labels[labels.TagLabelPrefix+"billing-tenant-id"],
				"region":        r.network.Spec.Region,
			},
			PVReclaimPolicy:  &defaultPVReclaimPolicy,
			Persistent:       true,
			StorageClassName: pointer.StringPtr(types.DefaultStorageClassName),
			Storage:          "2Gi",
			PodSecurityContext: &corev1.PodSecurityContext{
				RunAsUser:  pointer.Int64Ptr(1000),
				RunAsGroup: pointer.Int64Ptr(2000),
				FSGroup:    pointer.Int64Ptr(2000),
			},
		},
	}

	err := r.createOrGetWithOwnerLabel(tm)
	if err == nil {
		r.tm = tm
	}

	return err
}

func (r *reconcileContext) observeCreation() error {
	// no need to check TidbCluster is ready, because it has been checked in `ensureSecurityEnhancedModeEnabled`.
	// set `FromProvider` status for some usages like Backup
	if err := r.ensureStatus(func() error {
		return cloudv1.MutateSharedTierClusterStatus(r.cluster, func(status *cloudv1.SharedTierClusterStatus) error {
			if status.Kubernetes == nil {
				status.Kubernetes = &cloudv1.KubernetesStatus{}
			}
			// TODO(csuzhangxc): set GCP project ID for status
			status.Kubernetes.Namespace = r.tc.Namespace
			return nil
		})
	}); err != nil {
		return errors.Annotate(err, "failed to set cluster provider status")
	}

	return r.updateProgress(cloudv1.ClusterPhaseRunning, "cluster is running")
}

func (r *reconcileContext) getClusterConfig() (*api.DescribeClusterConfigOutput, error) {
	var comps []api.ComponentInput
	for _, comp := range r.cluster.GetComponents() {
		comps = append(comps, api.ComponentInput{
			Component:    api.ComponentName(comp.GetComponentType()),
			InstanceType: fmt.Sprintf("%s.%s", r.providerType, comp.GetInstanceType()),
		})
	}
	// TODO: gitops should abandon the concept of 'tenants'
	var legacyTenantId *uint64
	parsed, err := strconv.ParseUint(r.network.GetProjectId(), 10, 64)
	if err != nil {
		r.Info("error parsing the project name as a legacy 'tenantId' for gitops, use base config instead",
			"project", r.network.GetProjectId())
		dummy := uint64(0)
		legacyTenantId = &dummy
	} else {
		legacyTenantId = &parsed
	}

	return r.ConfigService.DescribeClusterConfiguration(&api.DescribeClusterConfigInput{
		TenantID:   legacyTenantId,
		Components: comps,
	})
}

func (r *reconcileContext) overlayAuthConfig(c *v1alpha1.ComponentSpec) {
	if r.providerType == types.ProviderTypeGCPSharedTier {
		// For pods in GKE to visit AWS API
		c.Env = append(c.Env, overlayutil.GetAWSEnv(r.backupRole, r.awsRegion)...)
		c.AdditionalVolumes = append(c.AdditionalVolumes, *overlayutil.GetAWSProjectedVolume())
		c.AdditionalVolumeMounts = append(c.AdditionalVolumeMounts, *overlayutil.GetAWSProjectedVolumeMount())
	}
}

func (r *reconcileContext) overlayTiKV(tc *v1alpha1.TidbCluster) error {
	topo := r.cluster.Spec.Cluster
	// TODO: reuse cluster config but not get again
	cc, err := r.getClusterConfig()
	if err != nil {
		r.InfraRecorder.Eventf(r.cluster, corev1.EventTypeWarning, "getClusterConfig", "%v", err)
		return errors.Annotate(err, "error get cluster config")
	}
	tikvConfig, ok := getComponentConfig(cc, api.ComponentTiKV)
	if !ok {
		msg := "no TiKV configuration provided"
		r.InfraRecorder.Event(r.cluster, corev1.EventTypeWarning, "getComponentConfig", msg)
		return errors.New(msg)
	}
	tikvResource, err := r.ConfigProvider.GetResourceSpec(r.provider, &topo.TiKV)
	if err != nil {
		r.InfraRecorder.Eventf(r.cluster, corev1.EventTypeWarning, "GetTikvResourceSpec", "%v", err)
		return errors.Annotate(err, "get tikv resources spec")
	}

	size := r.cluster.Spec.Cluster.TiKV.StorageSize
	if !size.IsZero() {
		tikvResource.Resources.Requests[corev1.ResourceStorage] = size
	}

	spec := &v1alpha1.TiKVSpec{
		ComponentSpec: v1alpha1.ComponentSpec{
			PodSecurityContext: &corev1.PodSecurityContext{
				RunAsUser:  pointer.Int64Ptr(defaultUserID),
				RunAsGroup: pointer.Int64Ptr(defaultGroupID),
				FSGroup:    pointer.Int64Ptr(defaultGroupID),
				Sysctls: []corev1.Sysctl{
					{
						Name:  "net.core.somaxconn",
						Value: "32768",
					},
				},
			},
			Annotations: map[string]string{
				consts.AnnSysctlInit: consts.AnnSysctlInitVal,
			},
			Version:                tikvConfig.Image.Tag,
			Env:                    []corev1.EnvVar{},
			AdditionalVolumes:      []corev1.Volume{},
			AdditionalVolumeMounts: []corev1.VolumeMount{},
		},
		EvictLeaderTimeout:       pointer.StringPtr("1m"),
		ResourceRequirements:     tikvResource.Resources,
		BaseImage:                tikvConfig.Image.BaseImage,
		Replicas:                 int32(topo.TiKV.Replicas),
		MaxFailoverCount:         pointer.Int32Ptr(0),
		StorageClassName:         firstNonEmpty(topo.TiKV.StorageClassName, tikvResource.StorageClass),
		Config:                   r.computeTiKVConfig(cc),
		MountClusterClientSecret: pointer.BoolPtr(true),
		ServiceAccount:           types.S3BackupServiceAccount,
	}

	r.overlayAuthConfig(&spec.ComponentSpec)
	tc.Spec.TiKV = spec

	return nil
}

// nolint: gomnd // nolint
func (r *reconcileContext) overlayTiFlash(tc *v1alpha1.TidbCluster) error {
	wantTiflash := r.cluster.Spec.Cluster.TiFlash.Replicas > 0
	haveTiflash := tc.Spec.TiFlash != nil && tc.Spec.TiFlash.Replicas > 0
	topo := r.cluster.Spec.Cluster
	if wantTiflash && haveTiflash {
		// skip creation of tiflash
		return nil
	}
	if !wantTiflash && !haveTiflash {
		// skip deletion of tiflash
		return nil
	}

	if haveTiflash {
		// delete tiflash
		tc.Spec.TiFlash.Replicas = 0
		return nil
	}

	// TODO: reuse cluster config but not get again
	cc, err := r.getClusterConfig()
	if err != nil {
		r.InfraRecorder.Eventf(r.cluster, corev1.EventTypeWarning, "getClusterConfig", "%v", err)
		return errors.Annotate(err, "error get cluster config")
	}

	tiflashConfig, ok := getComponentConfig(cc, api.ComponentTiFlash)
	if !ok {
		msg := "no TiFlash configuration provided"
		r.InfraRecorder.Event(r.cluster, corev1.EventTypeWarning, "getComponentConfig", msg)
		return errors.New(msg)
	}
	tiflashResource, err := r.ConfigProvider.GetResourceSpec(r.provider, &r.cluster.Spec.Cluster.TiFlash)
	if err != nil {
		r.InfraRecorder.Eventf(r.cluster, corev1.EventTypeWarning, "GetTiflashResourceSpec", "%v", err)
		return errors.Annotate(err, "get tiflash resources spec")
	}

	size := r.cluster.Spec.Cluster.TiFlash.StorageSize
	if !size.IsZero() {
		tiflashResource.Resources.Requests[corev1.ResourceStorage] = size
	}

	spec := &v1alpha1.TiFlashSpec{
		ComponentSpec: v1alpha1.ComponentSpec{
			Version: tiflashConfig.Image.Tag,
		},
		ResourceRequirements: tiflashResource.Resources,
		BaseImage:            tiflashConfig.Image.BaseImage,
		Replicas:             int32(r.cluster.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: r.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),
		}},
		ServiceAccount:   types.S3BackupServiceAccount,
		MaxFailoverCount: pointer.Int32Ptr(0),
	}

	r.overlayAuthConfig(&spec.ComponentSpec)
	tc.Spec.TiFlash = spec

	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, err := labels.TagsFromLabels(r.cluster.Labels)
	if err != nil {
		r.Error(err.ToAggregate(), "warning from TagsFromLabels")
	}
	for k, v := range tags {
		tc.Labels[k] = v
	}
	// add labels so that we can back reference to cluster object from tc
	labels.NewOwnerLabels(r.cluster, tc.Labels)

	// add all tc labels to pods
	tc.Spec.Labels = tc.Labels

	return nil
}

func (r *reconcileContext) overlayReplicas(tc *v1alpha1.TidbCluster) error {
	topo := r.cluster.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
}

// When creating a new cluster, security enhanced mode is disabled in the config.
// After the initializer succeeded, the SEM will be enabled and TiDB pods will restart.
// Before the SEM enabled and the TidbCluster becomes ready again, the Cluster should stay in Creating phase.
func (r *reconcileContext) ensureSecurityEnhancedModeEnabled() error {
	if r.ti.Status.JobStatus.Succeeded == 0 {
		if r.ti.Status.Phase == v1alpha1.InitializePhaseFailed {
			// log an error and return nil as we can't setup the cluster anymore.
			// maybe we can add a new phase for this case if needed later.
			r.Error(nil, "initializer failed, can't setup the cluster")
			return nil
		} else {
			if err := r.updateProgress(cloudv1.ClusterPhaseCreating,
				"waiting for initializer to complete"); err != nil {
				return err
			}
			return &utils.RequeueAfterError{RequeueAfter: defaultWaitInterval}
		}
	}

	// wait for TidbCluster ready and record the last ready time before enable SEM,
	// so that we can differentiate the cluster is ready again after SEM enabled.
	lastTimeStr, ok := r.cluster.Annotations[types.AnnoKeySEMTcLastReadyTime]
	if !ok {
		tcReady := tidbclusterutil.GetTidbClusterReadyCondition(r.tc)
		if tcReady == nil || tcReady.Status != corev1.ConditionTrue {
			if err := r.updateProgress(cloudv1.ClusterPhaseCreating,
				"waiting for tidbcluster ready before enable SEM"); err != nil {
				return err
			}
			return &utils.RequeueAfterError{RequeueAfter: defaultWaitInterval}
		}

		lastTimeStr = tcReady.LastTransitionTime.Format(time.RFC3339Nano)
		r.cluster.Annotations[types.AnnoKeySEMTcLastReadyTime] = lastTimeStr
		if err := r.InfraClient.Update(r.ctx, r.cluster); err != nil {
			return errors.Annotate(err, "update cluster annotations for SEM")
		}
	}

	// enable SEM by update the config of TiDB
	mutateFn := func(tc *v1alpha1.TidbCluster) error {
		if tc.Spec.TiDB.Config == nil {
			tc.Spec.TiDB.Config = v1alpha1.NewTiDBConfig()
		}
		tc.Spec.TiDB.Config.Set("security.enable-sem", true)
		return nil
	}
	if mutated, err := r.mutateTidbCluster(mutateFn); err != nil {
		return err
	} else if mutated {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "enabling security enhanced mode"); err != nil {
			return err
		}
		// re-queue reconcile to wait for TidbCluster.status to be updated as we mutated the config of TiDB.
		return &utils.RequeueAfterError{RequeueAfter: defaultWaitInterval}
	}

	// check TidbCluster is ready again after SEM enabled
	tcReady := tidbclusterutil.GetTidbClusterReadyCondition(r.tc)
	if tcReady == nil || tcReady.Status != corev1.ConditionTrue {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating,
			"waiting for tidbcluster ready after enable SEM"); err != nil {
			return err
		}
		return &utils.RequeueAfterError{RequeueAfter: defaultWaitInterval}
	}

	lastTime, err := time.Parse(time.RFC3339Nano, lastTimeStr)
	if err != nil {
		return errors.Annotate(err, "invalid last ready time for SEM in annotations")
	}

	if !tcReady.LastTransitionTime.After(lastTime) {
		// still need to wait for TidbCluster ready again after SEM enabled
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating,
			"waiting for tidbcluster ready after enable SEM"); err != nil {
			return err
		}
		return &utils.RequeueAfterError{RequeueAfter: defaultWaitInterval}
	}

	// SEM enabled and TidbCluster is ready again
	return nil
}

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

	return conf
}

func (r *reconcileContext) computeTiDBConfig(cc *api.DescribeClusterConfigOutput) *v1alpha1.TiDBConfigWraper {
	conf := v1alpha1.NewTiDBConfig()
	if dbConfig, ok := getComponentConfig(cc, api.ComponentTiDB); ok {
		conf.MP = dbConfig.Config.MP
	}
	conf.Set("security.cluster-verify-cn", []string{utils.FnvHash(r.tcKey.Name)})
	conf.Set("performance.tcp-keep-alive", true)

	return conf
}

func (r *reconcileContext) computeTiKVConfig(cc *api.DescribeClusterConfigOutput) *v1alpha1.TiKVConfigWraper {
	conf := v1alpha1.NewTiKVConfig()
	if kvConfig, ok := getComponentConfig(cc, api.ComponentTiKV); ok {
		conf.MP = kvConfig.Config.MP
	}
	conf.Set("security.cert-allowed-cn", []string{utils.FnvHash(r.tcKey.Name)})
	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", r.kmsKeyId)
	conf.Set("security.encryption.master-key.region", r.awsRegion)

	return conf
}

func (r *reconcileContext) computeTiFlashConfig(cc *api.DescribeClusterConfigOutput) *v1alpha1.TiFlashConfigWraper {
	conf := v1alpha1.NewTiFlashConfig()
	if tiflashConfig, ok := getComponentConfig(cc, api.ComponentTiFlash); ok {
		conf.Common.MP = tiflashConfig.Config.MP
	}
	conf.Common.Set("security.cert_allowed_cn", []string{utils.FnvHash(r.tcKey.Name)})
	conf.Proxy.Set("security.cert-allowed-cn", []string{utils.FnvHash(r.tcKey.Name)})
	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", r.kmsKeyId)
	conf.Proxy.Set("security.encryption.master-key.region", r.awsRegion)

	return conf
}

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
}

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

	return pointer.String(rhs)
}