package common

import (
	"context"
	"strconv"
	"strings"

	"github.com/pingcap/errors"
	oplabel "github.com/pingcap/tidb-operator/pkg/apis/label"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/tidbcloud/config-service/sdk/api"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/log"

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

const (
	MB = 1024 * 1024

	// the value of `GOMEMLIMIT` env is set to 80% of the memory limit of tidb container. @XuHuaiyu
	GOMEMLIMIT_RATE = 0.8

	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.
	// But when creating the cluster (restarting for SEM, audit log, etc.), we don't want to wait.
	annTiDBMinReadySeconds    = "tidb.pingcap.com/tidb-min-ready-seconds"
	annTiProxyMinReadySeconds = "tiproxy.pingcap.com/tiproxy-min-ready-seconds"
	defaultMinReadySeconds    = "180"

	gcpKMSSecretName string = "gcp-kms-credentials" // #nosec G101
)

func (c *ClusterContext) computeTiDBConfig(cc *api.DescribeClusterConfigOutput) *v1alpha1.TiDBConfigWraper {
	conf := v1alpha1.NewTiDBConfig()
	if dbConfig, ok := getComponentConfig(cc, api.ComponentTiDB); ok {
		conf.GenericConfig.MP = dbConfig.Config.MP
	}
	conf.Set("security.cluster-verify-cn", []string{util.FnvHash(c.Target.PrefixedId())})
	conf.Set("performance.tcp-keep-alive", true)
	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())})
	if c.KmsKeyId != nil {
		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
}

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

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

func buildBasicTidbCluster(cluster *cloudv1.Cluster, cc *api.DescribeClusterConfigOutput) (*v1alpha1.TidbCluster, error) {
	topologyConstraint := v1alpha1.TopologySpreadConstraint{
		TopologyKey: "topology.kubernetes.io/zone",
		// the following values should only take effect when the cluster is newly created
		MaxSkew: 1, // set the default value
	}

	// set minDomains to zones count for azure cluster
	if cluster.GetProviderType() == consts.ProviderTypeAKS {
		topologyConstraint.MinDomains = ptr.To(int32(len(cluster.Spec.Kubernetes.Cloud.Azure.NodeZones)))
	}

	helperImage := config.GetBusyboxImage()
	if image := util.GetAnnotation(cluster, gitopsConsts.OverrideHelperImageKey); image != "" {
		helperImage = image
	}

	tc := &v1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: cluster.PrefixedId(),
			Name:      consts.ClusterPrefix,
			Annotations: map[string]string{
				annSkipTLSWhenConnectTiDB: "",
			},
			Labels: map[string]string{
				// there are multiple TCs in one namespace in one dataplane, for example:
				// TCs for tidbgroups, TCs for ticdc. Adding this label to distinguish.
				cloudv1.LabelUsedFor: cloudv1.UsedForCluster,
			},
		},
		Spec: v1alpha1.TidbClusterSpec{
			Helper: &v1alpha1.HelperSpec{
				Image: ptr.To(helperImage),
			},
			Version: cc.Version,
			// in TiDB Cloud, there is no need to retain PV after scaled in.
			// this only affect the newly created cluster.
			PVReclaimPolicy:      ptr.To(corev1.PersistentVolumeReclaimDelete),
			ImagePullPolicy:      corev1.PullIfNotPresent,
			ConfigUpdateStrategy: v1alpha1.ConfigUpdateStrategyInPlace,
			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:                ptr.To(false),
			EnableDynamicConfiguration: ptr.To(true),
			TopologySpreadConstraints:  []v1alpha1.TopologySpreadConstraint{topologyConstraint},
			RecoveryMode:               cluster.Spec.RecoveryMode,
		},
	}
	return tc, nil
}

func (c *ClusterContext) buildTiDBConfig(tc *v1alpha1.TidbCluster, cc *api.DescribeClusterConfigOutput) error {
	tidbConfig, ok := getComponentConfig(cc, api.ComponentTiDB)
	if !ok {
		return errors.New("no TiDB configuration provided")
	}
	topo := c.Target.Spec.Cluster
	tidbResource, err := c.ConfigProvider.GetResourceSpec(c.Provider, &topo.TiDB)
	if err != nil {
		return errors.Annotate(err, "get tidb resources spec")
	}
	tidbSpec := &v1alpha1.TiDBSpec{
		ComponentSpec: v1alpha1.ComponentSpec{
			Affinity:     utils.Affinity(cloudv1.ComponentTiDB),
			NodeSelector: c.GetComponentNodeLabels(cloudv1.ComponentTiDB),
			Tolerations:  utils.Toleration(cloudv1.ComponentTiDB),
			HostNetwork:  ptr.To(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: tidbConfig.Image.Tag,
		},
		ResourceRequirements: tidbResource.Resources,
		Replicas:             int32(topo.TiDB.Replicas),
		BaseImage:            tidbConfig.Image.BaseImage,
		BinlogEnabled:        ptr.To(false),
		MaxFailoverCount:     ptr.To(int32(0)),
		SeparateSlowLog:      ptr.To(true),
		Service: &v1alpha1.TiDBServiceSpec{
			ExposeStatus: ptr.To(true),
			ServiceSpec: v1alpha1.ServiceSpec{
				Port: ptr.To(int32(consts.SQLEndpointTargetPort)),
				Type: corev1.ServiceTypeClusterIP,
			},
			ExternalTrafficPolicy: nil,
		},
		TLSClient: &v1alpha1.TiDBTLSClient{
			Enabled: true,
		},
		// Note: We shouldn't set TokenBasedAuthEnabled in here due to the cluster version in `cc` may be the DRYRUN
		// version. In other words, we may get an unexpected cluster version(dryrun version may be higher) and mistakenly
		// enable TokenBasedAuthEnabled for this cluster. refer to: https://pingcap.feishu.cn/docx/PwgZdyBvPou8I8xp7Iqcxy9XnKh
		//TokenBasedAuthEnabled: pointer.Bool(utils.TokenBasedAuthEnabled(c.Target, cc.Version)),
		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),
	}

	// for 6.1.3 <= TiDB-version < 6.5, we need to set `GOMEMLIMIT` env to avoid OOM.
	// as this env has no side effect for version >= 6.5 (TiDB kernel can handle it correctly, @XuHuaiyu),
	// and all newly created clusters should >= 6.1.3 now,
	// we can set it for all new clusters.
	// NOTE: do not set this env for existing clusters, otherwise the rolling update will be triggered.
	if tidbResource.Resources.Limits.Memory() != nil && !tidbResource.Resources.Limits.Memory().IsZero() {
		tidbGoLimit := int64(GOMEMLIMIT_RATE * float64(tidbResource.Resources.Limits.Memory().Value()))
		tidbSpec.Env = append(tidbSpec.Env, corev1.EnvVar{Name: "GOMEMLIMIT", Value: strconv.FormatInt(tidbGoLimit, 10)})
	}

	tc.Spec.TiDB = tidbSpec
	return nil
}

func (c *ClusterContext) buildTiFlashConfig(tc *v1alpha1.TidbCluster, cc *api.DescribeClusterConfigOutput) error {
	tiflashConfig, ok := getComponentConfig(cc, api.ComponentTiFlash)
	if !ok {
		return errors.New("no TiFlash configuration provided")
	}
	topo := c.Target.Spec.Cluster.TiFlash
	tiflashResource, err := c.ConfigProvider.GetResourceSpec(c.Provider, &topo)
	if err != nil {
		return errors.Annotate(err, "get tiflash resources spec")
	}
	tiflashSpec := &v1alpha1.TiFlashSpec{
		ComponentSpec: v1alpha1.ComponentSpec{
			Affinity:           utils.Affinity(cloudv1.ComponentTiFlash),
			NodeSelector:       c.GetComponentNodeLabels(cloudv1.ComponentTiFlash),
			Tolerations:        utils.Toleration(cloudv1.ComponentTiFlash),
			HostNetwork:        ptr.To(true),
			PodSecurityContext: &corev1.PodSecurityContext{},
			Version:            tiflashConfig.Image.Tag,
		},
		MaxFailoverCount:     ptr.To(int32(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.StorageClassName, tiflashResource.StorageClass),
		}},
	}
	tc.Spec.TiFlash = tiflashSpec
	return nil
}

// TODO(chenfei): remove it after we can remove CertService from cluster-agent
// syncFeatureAnnotations is used for syncing feature annotations from cluster CR to the main TC
func syncFeatureAnnotations(cluster *cloudv1.Cluster, tc *v1alpha1.TidbCluster) {
	if tc.Annotations == nil {
		tc.Annotations = map[string]string{}
	}

	for k, v := range cluster.Annotations {
		if !strings.HasPrefix(k, cloudv1.AnnoKeyFeatureNamePrefix) {
			continue
		}

		tc.Annotations[k] = v
	}
}

func (r *ClusterContext) OverlayMinReadyAnnotations(ctx context.Context, groupId string, tc *v1alpha1.TidbCluster) error {
	if tc.Annotations == nil {
		tc.Annotations = map[string]string{}
	}

	tc.Annotations[annTiDBMinReadySeconds] = r.Target.GetMinReadySeconds(ctx, groupId, cloudv1.ComponentTiDB)
	tc.Annotations[annTiProxyMinReadySeconds] = r.Target.GetMinReadySeconds(ctx, groupId, cloudv1.ComponentTiProxy)
	return nil
}

func (c *ClusterContext) OverlayTiFlash(tc *v1alpha1.TidbCluster, clusterConfig *api.DescribeClusterConfigOutput) error {
	wantTiflash := c.Target.Spec.Cluster.TiFlash.Replicas > 0
	haveTiflash := tc.Spec.TiFlash != nil

	// NOTE: `enableTiflash == true` is not equivalent to `wantTiflash && !haveTiflash`, because some TCs may
	// have tiflash spec but its replicas is 0, such as scale tiflash in to 0 and then scale out.
	enableTiflash := wantTiflash &&
		(tc.Spec.TiFlash == nil || tc.Spec.TiFlash.Replicas == 0)

	switch {
	case wantTiflash && !haveTiflash:
		// tiflash is needed but we do not have tiflash in tc yet, enable it
		if err := c.buildTiFlashConfig(tc, clusterConfig); err != nil {
			return err
		}
		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 {
			diff := tc.Spec.TiFlash.Replicas
			if diff > 0 {
				tc.Spec.TiFlash.ScalePolicy.ScaleInParallelism = &diff
			}
			tc.Spec.TiFlash.Replicas = 0
		}
	}
	if enableTiflash {
		log.FromContext(context.TODO()).Info("enable tiflash", "cluster", c.Target.ClusterScopeName())
		if tc.Spec.TiFlash.Annotations == nil {
			tc.Spec.TiFlash.Annotations = map[string]string{}
		}
		tc.Spec.TiFlash.Annotations[oplabel.AnnTiflashMountCMInTiflashContainer] = "true"

		if c.Target.FeatureEnabled(consts.ClusterFeatureEnableTiFlashGracefulShutdown) {
			// Similar to enableTiDBGracefulShutdown, but with a different timeout.
			// Set useSidecar to true refer to https://github.com/pingcap/tidb-operator/pull/6325.
			timeout := int64(consts.DefaultTiFlashGracefulShutdownTimeoutSeconds)
			tc.Spec.TiFlash.TerminationGracePeriodSeconds = &timeout
			tc.Spec.TiFlash.LogTailer.UseSidecar = true

			// Set sleepTimeSeconds refer to https://github.com/pingcap/tidb-operator/pull/6498.
			if c.Target.FeatureEnabled(consts.ClusterFeatureEnableTiFlashGracefulShutdownSleep) {
				val := c.Target.GetTiFlashGracefulShutdownSleepSeconds(context.TODO(), consts.DefaultTiFlashGracefulShutdownSleepSeconds)
				tc.Spec.TiFlash.LogTailer.SleepTimeSeconds = &val
			}
		}
	}
	return nil
}

func (c *ClusterContext) overlayTiFlashKmsCredentials(tc *v1alpha1.TidbCluster) {
	if c.Provider.GetProviderType() == 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 (c *ClusterContext) OverlayTiKVKmsCredentials(tc *v1alpha1.TidbCluster) error {
	if c.MountKmsSecret {
		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 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 ptr.To(lhs)
	}

	return ptr.To(rhs)
}