package task

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

	"github.com/crossplane-contrib/provider-aws/apis/eks/v1beta1"
	"github.com/go-logr/logr"
	"github.com/google/go-cmp/cmp"
	"github.com/google/go-cmp/cmp/cmpopts"
	opv2 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/errors"
	"k8s.io/client-go/tools/record"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers/utils/configmanager"
	"github.com/tidbcloud/eks-provider/utils/tidb-operator/tidbcluster"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/account"
	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/gitops"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/nodegroup"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const Finalizer = "cluster.cloud.pingcap.com/finalizer"

type Context struct {
	EnvProvider dataplane.EnvProvider
	// Transient during a deployment
	LegacyRemote k8s.ClientGetter

	TCGetter                  tidbcluster.Getter
	ExtraGroupTCGetter        map[string]tidbcluster.Getter
	Tidbcluster               *opv1alpha1.TidbCluster
	TidbClusterV2             *opv2.Cluster
	TidbClusterLastUpdateTime *time.Time
	Namer                     *namer.Namer
	PDClient                  configmanager.DynamicPDClient

	// Injected from the caller
	Local         client.Client
	EventRecorder record.EventRecorder

	// original is the cluster copy before reconciling
	// it's used for avoiding unnecessary updation
	original *cloudv1.Cluster
	// DON'T update cluster in tasks actually
	// DON'T get or set cluster spec in tasks
	// Both will be done after all tasks are finished or aborted.
	// See update.go for more details.
	Cluster  *cloudv1.Cluster
	Provider config.CloudConfigProvider

	EKSClusterSpec *cloudv1.EKSClusterSpec
	GKEClusterSpec *cloudv1.GKEClusterSpec
	AKSClusterSpec *cloudv1.AKSClusterSpec

	// TODO: it should be migrated to cloudv1.ClusterStatus
	CommonK8sClusterStatus *cloudv1.CommonK8sClusterStatus

	LegacyNetwork        *cloudv1.Network
	LegacyEKSNetworkSpec *v1alpha1.EKSNetworkSpec
	LegacyGKENetworkSpec *v1alpha1.GKENetworkSpec

	Config         config.RootConfig
	ConfigProvider gitops.ConfigProvider
	ConfigManager  configmanager.Interface

	// Account should not be directly used in any task
	// Deprecated: this field should be removed after all prod configs have been double checked
	LegacyAccount *cloudv1.Account

	AcctInfoGetter account.Getter

	PCAGetter              PCAGetter
	KMSGetter              KMSGetter
	CMEKStorageClassMapper cloudv1.CMEKStorageClassMapperV2

	Logger logr.Logger

	// BeginAt denotes the starting time of the task.
	BeginAt *time.Time

	zones []string
}

func (c *Context) Default(sctx context.Context) (*Context, error) {
	logger := log.FromContext(sctx)

	if c.ExtraGroupTCGetter == nil {
		c.ExtraGroupTCGetter = make(map[string]tidbcluster.Getter)
	}
	c.original = c.Cluster.DeepCopy()
	c.Namer = namer.NewNamer().
		WithClusterName(c.Cluster.Name).
		WithTiDBNamespace(c.Cluster.PrefixedId()).
		WithClusterServerDNSNames(c.Cluster.Spec.SqlEndpoints.DNS.Identity, c.Config.FullManage.PublicHostedZoneDomain)

	if c.IsLegacy() {
		// NOTE: network name and network namespace are used in nodeSelector of tidb in tc
		// So it cannot be removed simply for existing cluster
		c.Namer = c.Namer.LegacyWithNetworkName(c.Config.Provider + "-" + c.Config.Network.Region).
			LegacyWithNetworkNamespace(c.Cluster.Namespace)
		zones, err := c.Config.FindRegionZones(c.Config.Network.Region)
		// 3-layer arch cluster-controller won't get zones from config file
		// but clusters migrated from cloud-native arch are also managed by 3-layer arch cluster-controller
		// So we ignore error here, and check len(c.zones) later
		if err == nil {
			c.zones = zones
		}
	}
	// Use a nop mapper as the default
	c.CMEKStorageClassMapper = cloudv1.NewNopMapper()

	providerName := c.Config.Provider
	switch providerName {
	case consts.ProviderTypeEKS:
		if c.LegacyNetwork != nil {
			networkSpec, err := v1alpha1.EKSNetworkSpecForNetwork(c.LegacyNetwork)
			if err != nil {
				return nil, fmt.Errorf("unable to get network spec: %v", err)
			}
			c.LegacyEKSNetworkSpec = networkSpec
		}
		// 3-layer architecture must get zones from cluster spec
		// ignoring the zone info in config file
		if cloudaws := c.Cluster.Spec.Kubernetes.Cloud.AWS; cloudaws != nil {
			zones := []string{}
			for _, group := range cloudaws.SubnetGroups {
				if group.Name == cloudv1.AWSSubnetGroupNodes {
					for _, subnet := range group.Subnets {
						zones = append(zones, subnet.Zone)
					}
				}
			}
			c.zones = zones
		}

		clusterSpec, err := cloudv1.EKSClusterSpecForCluster(c.Cluster)
		if err != nil {
			return nil, fmt.Errorf("unable to get cluster eks spec: %v", err)
		}
		c.EKSClusterSpec = clusterSpec

		clusterStatus, err := cloudv1.EKSClusterStatusFromCluster(c.Cluster)
		if err != nil {
			return nil, fmt.Errorf("unable to get cluster eks status: %v", err)
		}
		c.CommonK8sClusterStatus = &clusterStatus.CommonK8sClusterStatus

		c.PCAGetter = newPCAGetterFromEKSClusterSpec(clusterSpec)
		c.KMSGetter = newKMSGetterFromEKSClusterSpec(c.LegacyEKSNetworkSpec, clusterSpec)
		// NOTE: there is a production tidb cluster that has KMS ARN annotation but not CMEK enabled
		// so that we have to check if CMEK is enabled in the cluster rather than check kmsArn.
		// Refer to https://pingcap.feishu.cn/wiki/VyYswDSTgiTe3rkBkAdcNtlWnte
		if c.Cluster.FeatureEnabled(consts.ClusterFeatureUseCMEK) {
			kmsArn := c.KMSGetter.Get()
			c.CMEKStorageClassMapper, err = cloudv1.NewAWSStorageClassMapper(kmsArn)
			if err != nil {
				return nil, fmt.Errorf("unable to create aws storage class mapper: %w", err)
			}
			logger.Info("using CMEK storage class mapper", "kmsArn", kmsArn)
		}

	case consts.ProviderTypeGKE:
		if c.LegacyNetwork != nil {
			networkSpec, err := v1alpha1.GKENetworkSpecForNetwork(c.LegacyNetwork)
			if err != nil {
				return nil, fmt.Errorf("unable to get network spec: %v", err)
			}
			c.LegacyGKENetworkSpec = networkSpec
		}
		if cloudgcp := c.Cluster.Spec.Kubernetes.Cloud.GCP; cloudgcp != nil {
			c.zones = cloudgcp.NodeZones
		}

		clusterSpec, err := cloudv1.GKEClusterSpecForCluster(c.Cluster)
		if err != nil {
			return nil, fmt.Errorf("unable to get cluster gke spec: %v", err)
		}
		c.GKEClusterSpec = clusterSpec

		clusterStatus, err := cloudv1.GKEClusterStatusFromCluster(c.Cluster)
		if err != nil {
			return nil, fmt.Errorf("unable to get cluster gke status: %v", err)
		}
		c.CommonK8sClusterStatus = &clusterStatus.CommonK8sClusterStatus

		c.PCAGetter = newPCAGetterFromGKEClusterSpec(clusterSpec)
		c.KMSGetter = newKMSGetterFromGKEClusterSpec(clusterSpec)
	case consts.ProviderTypeAKS:
		clusterSpec, err := cloudv1.AKSClusterSpecForCluster(c.Cluster)
		if err != nil {
			return nil, fmt.Errorf("unable to get cluster aks spec: %v", err)
		}
		c.AKSClusterSpec = clusterSpec

		clusterStatus, err := cloudv1.AKSClusterStatusFromCluster(c.Cluster)
		if err != nil {
			return nil, fmt.Errorf("unable to get cluster aks status: %v", err)
		}
		c.CommonK8sClusterStatus = &clusterStatus.CommonK8sClusterStatus

		c.PCAGetter = newPCAGetterFromAKSClusterSpec(clusterSpec)
		if c.Cluster.FeatureEnabled(consts.ClusterFeatureUseCMEK) {
			c.CMEKStorageClassMapper, err = cloudv1.NewAzureStorageClassMapper(c.Cluster)
			if err != nil {
				return nil, fmt.Errorf("unable to create azure storage class mapper: %w", err)
			}
			logger.Info("using CMEK storage class mapper", "azureCMEK", clusterSpec.AzureCMEK)
		}
		if cloudaks := c.Cluster.Spec.Kubernetes.Cloud.Azure; cloudaks != nil {
			c.zones = cloudaks.NodeZones
		}
	default:
		return nil, fmt.Errorf("unknown provider type: %s", providerName)
	}

	// skip zone check for azure tidb cluster
	if providerName != consts.ProviderTypeAKS && len(c.zones) != 3 {
		return nil, fmt.Errorf("available zones %v, the length of zones must be 3", c.zones)
	}

	// add legacy provider
	if c.EnvProvider == nil {
		ctx := context.Background()
		cli, err := c.LegacyRemote.GetClient(ctx)
		if err != nil {
			return nil, err
		}
		helmCfg, err := c.LegacyRemote.GetHelmConfig(ctx)
		if err != nil {
			return nil, err
		}
		var eks *v1beta1.Cluster
		if c.Config.Provider == consts.ProviderTypeEKS {
			obj := v1beta1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name: util.GetAnnotation(c.LegacyNetwork, consts.AnnotationKeyDataplaneName),
				},
			}
			if err := c.Local.Get(ctx, client.ObjectKeyFromObject(&obj), &obj); err != nil {
				return nil, err
			}

			eks = &obj
		}
		c.EnvProvider = dataplane.NewLegacy(
			ctx,
			c.Config.Provider,
			c.Local,
			cli,
			helmCfg,
			c.LegacyAccount,
			c.LegacyNetwork,
			c.LegacyEKSNetworkSpec,
			c.LegacyGKENetworkSpec,
			eks,
		)
	}

	c.CheckLegacyEKSNetworkSpec()

	return c, nil
}

func (c *Context) IsClusterChanged() bool {
	diff := cmp.Diff(c.original, c.Cluster, cmpopts.EquateEmpty())
	if diff == "" {
		c.Logger.Info("cluster is synced")
		return false
	}
	c.Logger.Info("cluster is not synced", "diff", diff)
	return true
}

// Legacy means the cluster is not created by new arch(3-layers)
func (c *Context) IsLegacy() bool {
	return c.Cluster.Spec.NetworkName != ""
}

// Migrated means the cluster is legacy but has been migrated to regional infra api
func (c *Context) IsMigrated() bool {
	return c.IsLegacy() && c.LegacyNetwork == nil
}

// It means the cluster is managed in regional infra api
func (c *Context) IsRegionalManaged() bool {
	return c.LegacyNetwork == nil
}

// For cluster with network, we generate node group name with the leagcy way
// Example: tidbcloud-8d0c22f0-87c5de9b129-{xxx}-f0b8c4b8
// It's hard to get some useful messages from name.
// For new cluster, we generate node group name with the new way
// Example: tidb-t4g-medium-{yyy}-75a4588d-us-west-2a-1-27
// It's easy to get component, instanceType, zone and current node version
func (c *Context) GetNodeGroupName(ngName, machineType, componentType string, zones []string, extraGroupName string) string {
	if c.IsLegacy() {
		prefix := nodegroup.TiDBClusterNodeGroupNamePrefix(ngName, c.GetNodeVersion(true))
		return c.uniqueIdentifier(namer.WithPrefix(prefix))
	}

	region := c.EnvProvider.GetRegion()
	suffix := region
	for _, zone := range zones {
		suffix += strings.TrimPrefix(zone, region)
	}

	suffix += "-" + strings.ReplaceAll(c.GetNodeVersion(false), ".", "-")
	prefix := componentType + "-" + strings.ReplaceAll(machineType, ".", "-")

	identity := c.Cluster.ClusterScopeName()
	if extraGroupName != "" {
		identity += "-" + extraGroupName
		prefix = "ex-" + prefix
	}

	// {component}-{machineType}-{ns-cluster}-{hash}-{zones}-{version}
	// ex-{component}-{machineType}-{ns-cluster-extraGroupName}-{hash}-{zones}-{version}
	return namer.NewName(identity, prefix, suffix, -1)
}

func (c *Context) uniqueIdentifier(opts ...namer.IdentifierOptions) string {
	if c.IsLegacy() {
		return namer.UniqueIdentifier(metav1.ObjectMeta{
			Name:      c.Config.Provider + "-" + c.Config.Network.Region,
			Namespace: c.Cluster.Namespace,
		}, opts...)
	}

	return namer.UniqueIdentifier(c.Cluster.ObjectMeta, opts...)
}

// GetNodeVersion should be called after Guardian task
func (c *Context) GetNodeVersion(allowEmpty bool) string {
	if allowEmpty {
		return c.CommonK8sClusterStatus.NodeVersion
	}

	if c.CommonK8sClusterStatus.NodeVersion != "" {
		return c.CommonK8sClusterStatus.NodeVersion
	}

	return c.Cluster.Annotations[consts.AnnoKeyControlPlaneVersion]
}

func assertEqual[T any](name string, expected, actual T) error {
	if reflect.DeepEqual(expected, actual) {
		return nil
	}

	return fmt.Errorf("%s is unexpected: expected: %v, actual: %v", name, expected, actual)
}

func (c *Context) CheckLegacyEKSNetworkSpec() {
	if c.LegacyEKSNetworkSpec == nil {
		return
	}

	spec := c.LegacyEKSNetworkSpec
	c.LegacyEKSNetworkSpec = nil
	errList := []error{
		assertEqual("node role arn", c.GetAWSNodeRoleARN(), spec.NodeRoleArn),
		assertEqual("assume role arn", c.GetAWSAssumeRoleARNForXP(), spec.AssumeRoleArn),
		assertEqual("assume role external id", c.GetAWSAssumeRoleExternalIDForXP(), spec.AssumeExternalId),
		assertEqual("kms key", "", spec.KMSARN),
	}

	if err := errors.NewAggregate(errList); err != nil {
		c.EventRecorder.Event(
			c.Cluster,
			corev1.EventTypeWarning,
			"InvalidEKSNetworkSpec",
			err.Error(),
		)
	}

	// always reset spec
	c.LegacyEKSNetworkSpec = spec
}

// TODO: it's no need to recalculate assumed role every times.
func (c *Context) GetAWSAssumedRoleForSDK() *aws.AssumedRole {
	externalID := c.GetAWSAssumeRoleExternalIDForXP()
	ar := &aws.AssumedRole{
		RoleARN: c.getAWSAssumeRoleARNForSDK(),
	}
	if externalID != "" {
		ar.ExternalID = &externalID
	}

	return ar
}

// GetAWSProviderConfigName returns the name of the AWS Crossplane ProviderConfig object.
func (c *Context) GetAWSProviderConfigName() string {
	if c.IsLegacy() {
		// the provider config name is the same as the EKS cluster name
		return c.EnvProvider.GetClusterName()
	}
	// pc-{ns-cluster}-{hash}
	return namer.NewName(c.Cluster.ClusterScopeName(), "pc", "", -1)
}

func (c *Context) GetAWSNodeRoleARN() string {
	if c.LegacyEKSNetworkSpec != nil {
		// the NodeRole itself is created by the Network Controller
		return c.LegacyEKSNetworkSpec.NodeRoleArn
	}
	// TODO(cluster-controller): compatible with cluster migrated from global infra API server
	return util.GetRoleARN(c.EnvProvider.GetAccount(), c.GetAWSNodeRoleName())
}

func (c *Context) GetAWSNodeRoleName() string {
	if c.IsLegacy() {
		// the node role name is the same as the EKS cluster name
		return c.EnvProvider.GetClusterName()
	}
	// ng-role-{ns-cluster}-{hash}
	return namer.NewName(c.Cluster.ClusterScopeName(), "ng-role", "", -1)
}

// getAWSAssumeRoleARNForSDK gets the AWS IAM role ARN used for calling AWS SDK. So returning a readonly in dryrun mode is
// required. If you want to get the assumed AWS IAM role which is saved into crossplane ProviderConfig, call GetAWSAssumeRoleARNForXP to instead.
func (c *Context) getAWSAssumeRoleARNForSDK() string {
	if c.Config.DryRun.Enabled {
		return util.GetRoleARN(c.EnvProvider.GetAccount(), c.Config.FullManage.ReadonlyAssumeRoleName)
	}

	if c.LegacyEKSNetworkSpec != nil {
		return c.LegacyEKSNetworkSpec.AssumeRoleArn
	}

	return util.GetRoleARN(c.EnvProvider.GetAccount(), c.Config.FullManage.AssumeRoleName)
}

// GetAWSAssumeRoleARNForXP gets the actual assumed AWS IAM role ARN which is used for saving assume information into the
// // corresponding CrossPlane ProviderConfig CR.
// This method is used for saving info so that we ignore dryrun logic in it.
// In general cases, just call getAWSAssumeRoleARNForSDK to instead.
func (c *Context) GetAWSAssumeRoleARNForXP() string {
	if c.LegacyEKSNetworkSpec != nil {
		return c.LegacyEKSNetworkSpec.AssumeRoleArn
	}

	return util.GetRoleARN(c.EnvProvider.GetAccount(), c.Config.FullManage.AssumeRoleName)
}

func (c *Context) GetAWSAssumeRoleExternalIDForXP() string {
	if c.LegacyEKSNetworkSpec != nil {
		return c.LegacyEKSNetworkSpec.AssumeExternalId
	}

	return c.Config.FullManage.AssumeRoleExternalID
}

func (c *Context) GetAWSAuditLogRoleExternalID() string {
	if c.IsLegacy() {
		networkNamespace := c.Cluster.Namespace
		networkName := c.Config.Provider + "-" + c.Config.Network.Region
		return namer.GenExternalID(networkNamespace, networkName)
	}

	return c.Cluster.GetProjectID()
}

func (c *Context) GetGCPNodePoolServiceAccountName() string {
	if c.IsLegacy() {
		return c.uniqueIdentifier(namer.GCPNodePoolServiceAccountNameOptions()...)
	}

	// np-sa-{k8s-name}-{hash}
	// TODO: make sure k8s name is unique for all gcp projects
	return namer.NewName(c.EnvProvider.GetClusterName(), "np-sa", "", namer.GKEServiceAccountNameLimit)
}

func (c *Context) GetGCPNodePoolServiceAccountEmail() string {
	return fmt.Sprintf(
		"%s@%s.iam.gserviceaccount.com",
		c.GetGCPNodePoolServiceAccountName(),
		c.EnvProvider.GetAccount())
}

func (c *Context) GetGCPNodePoolServiceAccountKeySecretName() string {
	if c.IsLegacy() {
		return c.uniqueIdentifier(namer.GCPNodePoolServiceAccountKeySecretNameOptions()...)
	}

	// use same name as service account
	return c.GetGCPNodePoolServiceAccountName()
}

// GetGCPProviderConfigName returns the name of the GCP Crossplane ProviderConfig object.
func (c *Context) GetGCPProviderConfigName() string {
	if c.IsLegacy() {
		return c.uniqueIdentifier()
	}

	// pc-{k8s-name}-{hash}
	// all resources in the same GKE share the same provider config
	// TODO: make sure k8s name is unique for all gcp projects
	return namer.NewName(c.EnvProvider.GetClusterName(), "pc", "", -1)
}

// Deprecated: use GetAKSDataplaneProviderConfigName instead. The previous ProviderConfigs created with this name
// are mistakenly using xxx-controplane subscriptions. However, Azure resources created for tidb clusters should be
// created in xxx-dataplane subscriptions.
//
// GetAKSProviderConfigName returns the name of the GCP Crossplane ProviderConfig object which
// is associated with xxx-controplane subscriptions.
func (c *Context) GetAKSProviderConfigName() string {
	// pc-{k8s-name}-{hash}
	// all resources in the same AKS share the same provider config
	return namer.NewName(c.EnvProvider.GetClusterName(), "pc", "", -1)
}

// GetAKSDataplaneProviderConfigName returns the name of the Azure Crossplane ProviderConfig object which
// is associated with xxx-dataplane subscriptions.
func (c *Context) GetAKSDataplaneProviderConfigName() string {
	// pc-dp-{k8s-name}-{hash}
	// all resources in the same AKS share the same provider config
	return namer.NewName(c.EnvProvider.GetClusterName(), "pc-dp", "", -1)
}

func (c *Context) Zones() []string {
	if c.zones == nil {
		panic("unreachable")
	}
	return c.zones
}

var _ deploy.Task = &BaseTask{}

type BaseTask struct {
	*Context
}

func (t *BaseTask) Name() string {
	return ""
}

func (t *BaseTask) Category() deploy.TaskCategory {
	return deploy.ClusterTask
}

func (t *BaseTask) Create(ctx context.Context) deploy.Result {
	return deploy.Done()
}

func (t *BaseTask) PostCreate(ctx context.Context) deploy.Result {
	return deploy.Done()
}

func (t *BaseTask) Delete(ctx context.Context) deploy.Result {
	return deploy.Done()
}

func (t *BaseTask) PostDelete(ctx context.Context) deploy.Result {
	return deploy.Done()
}

// MetricsStartTime will return the `BeginAt` (nil by default). For
// Task whose metrics need to be exported, the `BeginAt` should be
// initialized when the `Create` is being called.
func (t *BaseTask) MetricsStartTime() *time.Time {
	return t.BeginAt
}

func (t *BaseTask) TidbClusterMetaObject() *opv1alpha1.TidbCluster {
	return &opv1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.Namer.GetTiDBClusterName(),
			Namespace: t.Namer.GetTiDBNamespace(),
		},
	}
}

func (t *Context) GetClusterProxyBaseURL(ctx context.Context) (string, error) {
	if config.FeatureEnabled(t.Config.FeatureGates, config.UseProxyGatewayIngress, t.Cluster.GetTenantID(), t.Cluster.GetProjectID()) {
		return t.GetClusterProxyBaseURLFromClusterGateway(ctx)
	}
	cli := t.EnvProvider.GetClient()
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "proxy-cluster-proxy",
			Namespace: "tidb-admin",
		},
	}
	if err := cli.Get(ctx, client.ObjectKeyFromObject(svc), svc); err != nil {
		return "", err
	}
	ings := svc.Status.LoadBalancer.Ingress
	if len(ings) < 1 {
		return "", fmt.Errorf("proxy endpoint is not ready")
	}
	// aws uses `.Status.LoadBalancer.Ingress.HostName` to record the lb address.
	proxyEp := ings[0].Hostname
	if proxyEp == "" {
		// gcp uses `.Status.LoadBalancer.Ingress.IP`
		proxyEp = ings[0].IP
	}

	return fmt.Sprintf("http://%s/proxy/cluster/%s", proxyEp, t.Cluster.Name), nil
}

func (t *Context) GetClusterProxyBaseURLFromClusterGateway(ctx context.Context) (string, error) {
	gatewayEp, err := t.GetClusterGatewayBaseURL(ctx)
	if err != nil {
		return "", err
	}
	return fmt.Sprintf("%s/proxy/cluster/%s", gatewayEp, t.Cluster.Name), nil
}

func (t *Context) GetClusterGatewayBaseURL(ctx context.Context) (string, error) {
	cli := t.EnvProvider.GetClient()
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "clustergateway-kong-proxy",
			Namespace: "tidb-admin",
		},
	}
	if err := cli.Get(ctx, client.ObjectKeyFromObject(svc), svc); err != nil {
		return "", fmt.Errorf("get cluster gateway service err: %w", err)
	}
	ings := svc.Status.LoadBalancer.Ingress
	if len(ings) < 1 {
		return "", fmt.Errorf("clustergateway endpoint is not ready")
	}
	// aws uses `.Status.LoadBalancer.Ingress.HostName` to record the lb address.
	gatewayEp := ings[0].Hostname
	if gatewayEp == "" {
		// gcp uses `.Status.LoadBalancer.Ingress.IP`
		gatewayEp = ings[0].IP
	}

	return fmt.Sprintf("http://%s", gatewayEp), nil
}

func (t *Context) GetKMSConfig() (enabled bool, keyRegion string, keyID string) {
	var err error
	isEnable, ok := t.Cluster.Annotations[consts.KMSEnabledKey]
	if ok && isEnable == consts.AnnoValTrue {
		keyARN := t.KMSGetter.Get()
		if keyARN == "" {
			err = fmt.Errorf("kms is enabled for %s/%s, but the kms key is not found", t.Cluster.Namespace, t.Cluster.Name)
			panic(err)
		}
		enabled = true

		keyRegion, keyID, err = util.ParseKeyARN(keyARN)
		if err != nil {
			panic(err)
		}
	}
	return enabled, keyRegion, keyID
}

func (t *Context) TidbGroupNamer(groupSpec *cloudv1.ExtraGroupSpec) *namer.Namer {
	return t.Namer.ForTiDBGroup(t.Cluster, groupSpec, t.Config.FullManage.PublicHostedZoneDomain)
}

func (t *Context) GetNamer(ctx context.Context, comp cloudv1.ClusterComponentSpecWrapped) *namer.Namer {
	logger := log.FromContext(ctx)
	if comp.GetGroupId() == t.Cluster.GetDefaultGroupID() {
		return t.Namer
	} else {
		group, ok := t.Cluster.GetExtraGroup(comp.GetGroupId())
		if !ok { // shouldn't happen
			logger.Info("extra group not found", "groupId", comp.GetGroupId())
			return nil
		}

		return t.TidbGroupNamer(group.Spec)
	}
}

func (t *Context) TidbGroupTLSEnabled(ctx context.Context, groupSpec *cloudv1.ExtraGroupSpec) (bool, error) {
	if groupSpec.Deleted() {
		// Safe to return false because it will not be used.
		return false, nil
	}

	if t.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		if t.TCGetter == nil {
			return false, fmt.Errorf("TCGetter is nil")
		}
		return t.TCGetter.IsTLSEnabled(), nil
	}

	remote := t.EnvProvider.GetClient()
	extraGroupNamer := t.TidbGroupNamer(groupSpec)
	key := client.ObjectKey{Namespace: extraGroupNamer.GetTiDBNamespace(), Name: extraGroupNamer.GetTiDBName()}
	tc := &opv1alpha1.TidbCluster{}
	if err := remote.Get(ctx, key, tc); err != nil {
		return false, fmt.Errorf("get tidbcluster %s/%s err: %w", extraGroupNamer.GetTiDBNamespace(), extraGroupNamer.GetTiDBName(), err)
	}
	return tidbcluster.NewV1Getter(tc).IsTLSEnabled(), nil
}

func (t *Context) GetTCGetter(ctx context.Context, groupId string) (tidbcluster.Getter, error) {
	var getter tidbcluster.Getter
	if t.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		// Currently, no clusters with extra groups created are using TiDB Operator V2.
		getter = t.TCGetter
	} else {
		if groupId == t.Cluster.GetDefaultGroupID() {
			getter = t.TCGetter
		} else {
			getter = t.ExtraGroupTCGetter[groupId]
		}
	}

	if getter == nil {
		return nil, fmt.Errorf("tidbcluster getter for group %s is nil", groupId)
	}
	return getter, nil
}

type KMSGetter interface {
	Get() string
}

type PCAGetter interface {
	Get() string
}

type GetFunc func() string

func (f GetFunc) Get() string {
	return f()
}

func newKMSGetterFromEKSClusterSpec(networkSpec *v1alpha1.EKSNetworkSpec, spec *cloudv1.EKSClusterSpec) KMSGetter {
	return GetFunc(func() string {
		arn := ""
		if networkSpec != nil && networkSpec.KMSARN != "" {
			arn = networkSpec.KMSARN
		}

		if spec == nil {
			return arn
		}
		if spec.Security == nil {
			return arn
		}

		if spec.Security.MasterKeyARN == nil {
			return arn
		}

		if *spec.Security.MasterKeyARN == "" {
			return arn
		}

		return *spec.Security.MasterKeyARN
	})
}

func newKMSGetterFromGKEClusterSpec(spec *cloudv1.GKEClusterSpec) KMSGetter {
	return GetFunc(func() string {
		if spec == nil {
			return ""
		}
		if spec.Security == nil {
			return ""
		}

		if spec.Security.MasterKeyARN == nil {
			return ""
		}

		return *spec.Security.MasterKeyARN
	})
}

func newPCAGetterFromEKSClusterSpec(spec *cloudv1.EKSClusterSpec) PCAGetter {
	return GetFunc(func() string {
		if spec == nil {
			return ""
		}
		if spec.ComponentPCAARN == nil {
			return ""
		}

		return *spec.ComponentPCAARN
	})
}

func newPCAGetterFromGKEClusterSpec(spec *cloudv1.GKEClusterSpec) PCAGetter {
	return GetFunc(func() string {
		if spec == nil {
			return ""
		}
		if spec.ComponentPCAARN == nil {
			return ""
		}

		return *spec.ComponentPCAARN
	})
}

func newPCAGetterFromAKSClusterSpec(spec *cloudv1.AKSClusterSpec) PCAGetter {
	return GetFunc(func() string {
		if spec == nil {
			return ""
		}
		if spec.ComponentPCAARN == nil {
			return ""
		}

		return *spec.ComponentPCAARN
	})
}