package regionalservice

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"reflect"
	"regexp"
	"sort"
	"strconv"
	"strings"

	"github.com/google/martian/log"

	"github.com/tidbcloud/observability/pkg/resourcemanager"

	prometheusv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
	"k8s.io/apimachinery/pkg/util/intstr"

	v1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/configs"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"
	"github.com/tidbcloud/observability/pkg/common/message"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/common/utils"

	k8sv1 "k8s.io/api/core/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const (
	AzureRoleConditionTmpl = `(
 (
  !(ActionMatches{'Microsoft.Storage/storageAccounts/blobServices/containers/blobs/read'})
  AND
  !(ActionMatches{'Microsoft.Storage/storageAccounts/blobServices/containers/blobs/write'})
  AND
  !(ActionMatches{'Microsoft.Storage/storageAccounts/blobServices/containers/blobs/add/action'})
  AND
  !(ActionMatches{'Microsoft.Storage/storageAccounts/blobServices/containers/blobs/delete'})
  AND
  !(ActionMatches{'Microsoft.Storage/storageAccounts/blobServices/containers/blobs/move/action'})
 )
 OR
 (
  @Resource[Microsoft.Storage/storageAccounts/blobServices/containers:name] StringEquals '%s'
  AND
  @Resource[Microsoft.Storage/storageAccounts/blobServices/containers/blobs:path] StringLike '?/%s/*'
 )
)`
	// Scope at which the role assignment or definition applies to
	// in our case, it's the blob storage container under o11y's subscription.
	AzureRoleScopeTmpl = "/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Storage/storageAccounts/%s/blobServices/default/containers/%s"
)

func NamespaceName(bizType, tenantID string) string {
	return fmt.Sprintf(constants.NamespaceTemplate, bizType, tenantID)
}
func DataPlaneSANamespaceName(vendor constants.ProvidedVendor, applicationID string) string {
	switch {
	case vendor == constants.GCP:
		if applicationID == applicationIDK8sInfra {
			return constants.GCPK8sTiDBClusterVectorSAKey
		}
		return fmt.Sprintf(constants.GCPTiDBClusterVectorSAKeyTemplate, applicationID)
	case vendor == constants.Azure:
		if applicationID == applicationIDK8sInfra {
			return constants.AzureK8sTiDBClusterVectorSAKey
		}
		return fmt.Sprintf(constants.AzureTiDBClusterVectorSAKeyTemplate, applicationID)
	// AWS is using wildcard matching, so it does not need the namespaced name service account.
	default:
		return ""
	}
}

func ObjectStorageResourceName(baseInfo structs.ClusterBaseInfo) string {
	svcInfo := configs.GetServiceInfo()
	if bucket := svcInfo.BizTypeConfigs[baseInfo.BizType].SpecificBucket; len(bucket) > 0 {
		return bucket
	} else {
		return fmt.Sprintf(constants.StorageBucketNameTemplate, svcInfo.Env, constants.DefaultBucketGroup, svcInfo.LogicalRegionName)
	}
}

func ObjectStorageDataPrefix(baseInfo structs.ClusterBaseInfo, category constants.ObservabilityCategory) string {
	return fmt.Sprintf(constants.StorageDataPrefixTemplate,
		baseInfo.DataPropertyRights,
		baseInfo.TenantID,
		baseInfo.ProjectID,
		baseInfo.ApplicationID,
		category)
}

func ObjectStorageEndpoint(bucketName string) string {
	return fmt.Sprintf("%s.s3.%s.amazonaws.com", bucketName, configs.GetServiceInfo().Region)
}

func BuildNamespaceResourceRequirements(ctx context.Context, bizType, tenantID string, shareMode constants.ShareMode, nextgen bool) (*k8sv1.Namespace, string, error) {
	if len(bizType) == 0 || len(tenantID) == 0 {
		return nil, "", errors.NewErrorf(errors.ParameterInvalid, "bizType = %s, tenantID = %s", bizType, tenantID)
	}
	var vmAccountID string
	ns := &k8sv1.Namespace{
		ObjectMeta: metav1.ObjectMeta{
			Name: NamespaceName(bizType, tenantID),
			Labels: map[string]string{
				constants.LabelBizType:  bizType,
				constants.LabelTenantID: tenantID,
			},
		},
	}

	if isSharedVMRule(shareMode) &&
		configs.GetServiceInfo().BizTypeConfigs[bizType].MultiTenancyRule {
		ns.Labels[constants.LabelMultiTenancyRule] = "true"
	}
	if nextgen {
		accountID, err := resourcemanager.GetDataAccountReaderWriter().UniqueInt32ForCluster(ctx)
		if err != nil {
			return nil, "", err
		}
		vmAccountID = strconv.Itoa(int(accountID))
		ns.Labels[constants.LabelVMAccountID] = vmAccountID
	}
	if configs.GetServiceInfo().Vendor == string(constants.AliCloud) {
		ns.Labels[constants.LabelAliyunInjection] = "on"
	}
	return ns, vmAccountID, nil
}

func BuildObjectStorageResourceRequirements(bucketName string) (*v1.S3, error) {
	if len(bucketName) == 0 {
		return nil, errors.New(errors.ParameterInvalid, "bucket name is empty")
	}
	return &v1.S3{
		ObjectMeta: metav1.ObjectMeta{
			Name:        bucketName,
			Namespace:   constants.ObservabilityNamespace,
			Labels:      map[string]string{},
			Annotations: map[string]string{},
		},
		Spec: v1.S3Spec{
			Name: bucketName,
		},
	}, nil
}

var azIndex = -1

// pick default az
func pickDefaultAZ() (string, structs.AvailabilityZoneConfig, error) {
	azs := make(map[string]structs.AvailabilityZoneConfig, 0)
	var key string
	azConfigs := configs.GetServiceInfo().AZConfig
	for k, v := range azConfigs {
		if v.DefaultDataAZ {
			azs[k] = v
			key = k
			break
		}
	}
	if len(azs) == 0 {
		return "", structs.AvailabilityZoneConfig{}, errors.New(errors.AbsenceOfConfig, "no available zone")
	}
	return key, azs[key], nil
}

// pick az
func pickAZ() (string, structs.AvailabilityZoneConfig, error) {
	azConfigs := configs.GetServiceInfo().AZConfig

	azs := make(map[string]structs.AvailabilityZoneConfig, 0)
	for k, v := range azConfigs {
		if !v.SellOut {
			azs[k] = v
		}
	}

	if len(azs) == 0 {
		return "", structs.AvailabilityZoneConfig{}, errors.New(errors.AbsenceOfConfig, "no available zone")
	}

	azIndex = azIndex + 1
	if azIndex >= len(azs) {
		azIndex = 0
	}

	mapKeys := make([]string, 0)
	for key := range azs {
		mapKeys = append(mapKeys, key)
	}

	sort.Strings(mapKeys)

	key := mapKeys[azIndex]
	return key, azs[key], nil
}

func BuildPrometheusRuleResourceRequirements(req message.CreateClusterReq, clusterResource *v1.Cluster) (*prometheusv1.PrometheusRule, error) {
	spec := prometheusv1.PrometheusRuleSpec{}
	if req.MonitorPromRules.Groups != nil {
		spec = req.MonitorPromRules
	}
	promRule := &prometheusv1.PrometheusRule{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.ID,
			Namespace: NamespaceName(req.BizType, req.TenantID),
			Labels: map[string]string{
				constants.LabelClusterID:     req.ID,
				constants.LabelVendor:        string(req.Vendor),
				constants.LabelRegion:        req.Region,
				constants.LabelTenantID:      req.TenantID,
				constants.LabelProjectID:     req.ProjectID,
				constants.LabelApplicationID: req.ApplicationID,
				constants.LabelBizType:       req.BizType,
			},
			Annotations: map[string]string{},
		},
		Spec: spec,
	}

	if isSharedVMRule(req.ShareMode) &&
		configs.MultiTenancyRuleEnabled(req.BizType, req.ApplicationID) &&
		!req.MuteAlert {
		promRule.ObjectMeta.Labels[constants.LabelMultiTenancyRule] = "true"
		accountID := clusterResource.Spec.VictoriaAccount.AccountID
		projectID := clusterResource.Spec.VictoriaAccount.ProjectID
		promRule.ObjectMeta.Labels[constants.LabelVMAccountID] = fmt.Sprintf("%v", accountID)
		promRule.ObjectMeta.Labels[constants.LabelVMProjectID] = fmt.Sprintf("%v", projectID)
	}
	return promRule, nil
}

func replaceByPlaceHold(src string, placeholderVariables map[string]string) string {
	replaceConfig := make([]string, 0)
	for _, hold := range constants.Placeholder {
		if place, ok := placeholderVariables[hold]; ok {
			replaceConfig = append(replaceConfig,
				constants.PlaceholderPrefix+hold+constants.PlaceholderSufix,
				place)
		}
	}
	replaceMent := strings.NewReplacer(replaceConfig...)
	return replaceMent.Replace(src)
}

func BuildPrometheusRuleTemplateResourceRequirements(req message.ModifyAlertRulesTemplateReq) (*prometheusv1.PrometheusRule, error) {
	spec := prometheusv1.PrometheusRuleSpec{}
	if req.MonitorPromRules.Groups != nil {
		spec = req.MonitorPromRules
	}
	return &prometheusv1.PrometheusRule{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.TemplateID,
			Namespace: constants.AlertRuleTemplateNamespace,
			Labels: map[string]string{
				constants.LabelClusterID:       req.TemplateID,
				constants.ExternalLabelOverlay: "true",
			},
		},
		Spec: spec,
	}, nil
}

func getAlertingConfig() *prometheusv1.AlertingSpec {
	//get alerting config
	alertmanagerEndpoints := make([]prometheusv1.AlertmanagerEndpoints, 0)
	alertmanagerEndpoints = append(alertmanagerEndpoints, prometheusv1.AlertmanagerEndpoints{
		Namespace: configs.GetAlertManagerNameSpace(),
		Name:      configs.GetAlertManagerName(),
		Port:      intstr.FromString(configs.GetAlertManagerPort()),
	})

	return &prometheusv1.AlertingSpec{
		Alertmanagers: alertmanagerEndpoints,
	}
}

//nolint:gocyclo // function orchestrates cluster resource initialization; refactor would risk regressions
func BuildClusterResourceRequirements(ctx context.Context, req message.CreateClusterReq, vmaccountid string, parentCluster *v1.Cluster) (*v1.Cluster, error) {
	var (
		az  string
		err error
	)

	if req.ShareMode == constants.SHARED || req.ShareMode == constants.ENHANCED {
		az, _, err = pickDefaultAZ()
	} else {
		az, _, err = pickAZ()
	}

	if err != nil {
		return &v1.Cluster{}, err
	}

	// If agent manager arn is not passed as parameter then default config is used
	agentManagerARN := req.AgentManagerARN
	// if agent account not nil, then use the account from request
	if len(req.AgentAccessAccount) > 0 {
		switch configs.GetServiceInfo().Vendor {
		case string(constants.AWS):
			{
				agentManagerARN = fmt.Sprintf("arn:aws:iam::%s:root", req.AgentAccessAccount)
			}
		case string(constants.AliCloud):
			{
				agentManagerARN = fmt.Sprintf("acs:ram:*:%s:*", req.AgentAccessAccount)
			}
		case string(constants.Azure):
			{
				agentManagerARN = req.AgentAccessAccount
			}
		default:
			return &v1.Cluster{}, errors.NewErrorf(errors.ParameterInvalid, "agent access account is only supported for aws, alicloud and azure")
		}

	}
	if !req.AgentNextGen && len(agentManagerARN) == 0 {
		if req.AgentEnvironment != nil {
			switch configs.GetServiceInfo().Vendor {
			case string(constants.AWS):
				agentManagerARN = fmt.Sprintf("arn:aws:iam::%s:root", req.AgentEnvironment.AWS.Account)
			case string(constants.GCP):
				agentManagerARN = req.AgentEnvironment.GCP.GCPProject
			case string(constants.Azure):
				agentManagerARN = req.AgentEnvironment.Azure.SubscriptionID
			}
		} else {
			agentManagerARN, err = configs.GetAgentManagerARNByBizType(req.BizType)
			if err != nil {
				return &v1.Cluster{}, err
			}
		}
	}

	ruleTemplateID := req.RuleTemplateID
	if !req.MuteAlert && req.RuleTemplateID == "" {
		ruleTemplateID = configs.GetServiceInfo().BizTypeConfigs[req.BizType].MonitorPromRuleTemplate
	} else if req.MuteAlert {
		// ignore rule template if alert is muted during cluster creation.
		// this also avoids too much unnecessary reconciliation.
		ruleTemplateID = ""
	}

	cluster := &v1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.ID,
			Namespace: NamespaceName(req.BizType, req.TenantID),
			Labels: map[string]string{
				constants.LabelClusterID:       req.ID,
				constants.LabelVendor:          string(req.Vendor),
				constants.LabelRegion:          req.Region,
				constants.LabelTenantID:        req.TenantID,
				constants.LabelApplicationID:   req.ApplicationID,
				constants.LabelBizType:         req.BizType,
				constants.LabelAZ:              az,
				constants.LabelTemplateID:      ruleTemplateID,
				constants.LabelShareMode:       string(req.ShareMode),
				constants.LabelParentClusterID: req.ParentClusterID,
				constants.LabelNextGen:         strconv.FormatBool(req.AgentNextGen),
			},
			Annotations: map[string]string{
				constants.AnnotationNote:                       req.Note,
				constants.AnnotationVersion:                    string(req.Version),
				constants.AnnotationTags:                       utils.MarshalList(req.Tags),
				constants.AnnotationShareMode:                  string(req.ShareMode),
				constants.AnnotationMonitorSpec:                req.MonitorSpecID,
				constants.AnnotationObjectStorageBucket:        ObjectStorageResourceName(req.ClusterBaseInfo),
				constants.AnnotationBlobStorageAccount:         configs.GetAzureStorageAccount(),
				constants.AnnotationObjectStorageMetricsPrefix: ObjectStorageDataPrefix(req.ClusterBaseInfo, constants.METRICS),
				constants.AnnotationObjectStorageEndpoint:      ObjectStorageEndpoint(ObjectStorageResourceName(req.ClusterBaseInfo)),
				constants.AnnotationDataPropertyRights:         strconv.Itoa(int(req.DataPropertyRights)),
				constants.AnnotationProxyClusterID:             req.SeedClusterID,
			},
		},
		Spec: v1.ClusterSpec{
			MuteAlert: req.MuteAlert,
			Endpoint: v1.Endpoint{
				Name:    req.ID,
				UserARN: agentManagerARN, //if agentManagerARN is empty, then it will not create endpoint service CR
				Shared:  true,
			},
		},
		Status: v1.ClusterStatus{
			Phase: constants.ACTIVE,
		},
	}

	if parentCluster != nil && req.KeySpaceID != "" && req.ParentClusterID != "" && checkuseRDS(ctx) {
		// for serverless child cluster on rds mode, we don't need prepare namespace, reuse parentCluster namespace here.
		cluster.Namespace = parentCluster.Namespace
	}

	// if o11y agent role is needed, then set service account
	if !AgentRoleNotNeeded(&req.ClusterBaseInfo) && !req.AgentNextGen {
		cluster.Annotations[constants.AnnotationServiceAccount] = NamespaceName(req.BizType, req.TenantID)
	}
	if req.BYOCVPCPeering != nil {
		cluster.Annotations[constants.AnnotationBYOCVPCPeeringID] = req.BYOCVPCPeering.PeeringConnectionID
		cluster.Annotations[constants.AnnotationBYOCDataplaneVPCCIDR] = req.BYOCVPCPeering.DataplaneVPCCIDR
	}
	// projectid in next gen is optional, so we do not set it in labels if it is empty.
	if len(req.ProjectID) > 0 {
		cluster.ObjectMeta.Labels[constants.LabelProjectID] = req.ProjectID
	}
	mode := req.ShareMode

	err = appendMonitor(ctx, cluster, req, az, mode, vmaccountid)
	if err != nil {
		return nil, err
	}
	//append agent
	err = appendAgent(ctx, cluster, req)
	if err != nil {
		return nil, err
	}

	return cluster, err
}

func appendMonitor(ctx context.Context, cluster *v1.Cluster, req message.CreateClusterReq, az string, mode constants.ShareMode, vmaccountid string) error {
	if constants.SHARED == mode ||
		constants.ENHANCED == mode {
		err := appendVM(ctx, cluster, req, az, vmaccountid)
		if err != nil {
			return err
		}
	}

	if constants.EXCLUSIVE == mode ||
		constants.ENHANCED == mode {
		err := appendPrometheus(ctx, cluster, req, az)
		if err != nil {
			return err
		}
	}
	return nil
}

func appendPrometheus(ctx context.Context, cluster *v1.Cluster, req message.CreateClusterReq, az string) error {
	monitorResourceRequirements, err := MonitorSpecRequirements(req.MonitorSpecID)
	if err != nil {
		return err
	}

	cluster.Spec.Prometheus = v1.Prometheus{
		Name:     req.ID,
		Replicas: monitorReplicas(req),
		Alerting: getAlertingConfig(),
		RuleNamespaceSelector: &metav1.LabelSelector{
			MatchLabels: map[string]string{
				constants.LabelTenantID: req.TenantID,
			},
		},
		RuleSelector: &metav1.LabelSelector{
			MatchLabels: map[string]string{
				constants.LabelClusterID: req.ID,
			},
		},

		Version:                monitorVersion(req),
		Resources:              monitorResourceRequirements,
		Retention:              monitorDuration(req),
		AvailabilityZone:       az,
		MetricsToObjectStorage: true,
	}

	if s, ok := configs.GetServiceInfo().MonitorSpec[req.MonitorSpecID]; ok {
		cluster.Spec.Prometheus.Query = &prometheusv1.QuerySpec{
			MaxConcurrency: s.MaxQueryConcurrency,
			MaxSamples:     s.MaxQuerySamples,
		}
	}

	return nil
}

func getVMSpace(ctx context.Context, req message.CreateClusterReq) (accountID int32, projectID int32, err error) {
	parentClusterID, spaceID := req.ParentClusterID, req.KeySpaceID

	if len(parentClusterID) > 0 {
		parent, getErr := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, parentClusterID)
		if getErr != nil {
			log.Errorf("get parent cluster failed, parent cluster id = %s", parentClusterID)
			err = getErr
			return
		}

		accountID = parent.Spec.VictoriaAccount.AccountID
	} else {
		if strings.Contains(req.BizType, BizeTypeTidbCloud) && req.ApplicationID == constants.ApplicationIDK8sInfra {
			// re-use vmaccount id project-wide for dedicated k8s-infra cluster
			c, acErr := resourcemanager.GetArchivedClusterReaderWriter().Get(ctx, ObjectStorageResourceName(req.ClusterBaseInfo), structs.ClusterUniqueKey{
				BizType:       req.BizType,
				TenantID:      req.TenantID,
				ProjectID:     req.ProjectID,
				ApplicationID: req.ApplicationID,
			})
			if acErr != nil && !errors.Is(acErr, errors.ObjectNotFoundError) {
				err = acErr
				return
			}

			if len(c.VMAccountID) > 0 {
				id, parseErr := parseVMAccountID(c.VMAccountID)
				if parseErr != nil {
					err = parseErr
					return
				}
				accountID = id
				return
			}
		}
		accountID, err = resourcemanager.GetDataAccountReaderWriter().UniqueInt32ForCluster(ctx)
	}

	if len(spaceID) > 0 {
		pid, atoiErr := strconv.ParseInt(spaceID, 10, 32)
		if atoiErr != nil {
			err = atoiErr
		} else {
			projectID = int32(pid)
		}
	}
	return accountID, projectID, err
}

func appendVM(ctx context.Context, cluster *v1.Cluster, req message.CreateClusterReq, az string, accountid string) error {
	var vmAccountID, vmProjectID int32
	var err error
	//for next-gen, we reuse the vm account id from namespace label.
	if req.AgentNextGen {
		id, err := strconv.ParseInt(accountid, 10, 32)
		if err != nil {
			return errors.NewErrorf(errors.ParameterInvalid, "vm account id %s is not valid", accountid)
		}
		vmAccountID = int32(id)
		// we use project id from namespaced generation
		vmProjectID, err = resourcemanager.GetDataAccountReaderWriterFactory().GetReaderWriter(NamespaceName(req.BizType, req.TenantID)).UniqueInt32ForCluster(ctx)
		if err != nil {
			return err
		}
	} else {
		vmAccountID, vmProjectID, err = getVMSpace(ctx, req)
		if err != nil {
			return err
		}
	}
	vmClusters := configs.GetServiceInfo().VMClusterInfo
	if len(vmClusters) == 0 {
		return errors.NewErrorf(errors.AbsenceOfConfig, "no victoria metrics config")
	}

	cluster.Spec.VictoriaAccount = v1.VictoriaAccount{
		AccountID:        vmAccountID,
		ProjectID:        vmProjectID,
		AvailabilityZone: az,
		VMClusterName:    vmClusters[0].VMClusterName,
		Alerting:         getAlertingConfig(),
		RuleSelector: &metav1.LabelSelector{
			MatchLabels: map[string]string{
				constants.LabelClusterID: req.ID,
			},
		},
	}
	// deal rds mode serverless parent cluster
	if checkuseRDS(ctx) && req.KeySpaceID == "" && req.ParentClusterID == "" {
		cluster.Spec.VictoriaAccount.SupportServerless = true
	}
	return nil
}

func MonitorSpecRequirements(specID string) (k8sv1.ResourceRequirements, error) {
	if v, ok := configs.GetSpecContainer()[specID]; ok {
		return v, nil
	} else {
		return k8sv1.ResourceRequirements{}, errors.NewErrorf(errors.UnsupportedMonitorSpec, "spec %s is not supported", specID)
	}
}

func monitorReplicas(req message.CreateClusterReq) int32 {
	return int32(configs.GetServiceInfo().BizTypeConfigs[req.BizType].MonitorReplicas)
}

func monitorVersion(req message.CreateClusterReq) string {
	return configs.GetServiceInfo().BizTypeConfigs[req.BizType].MonitorVersion
}

func monitorDuration(req message.CreateClusterReq) v1.Duration {
	return v1.Duration(configs.GetServiceInfo().BizTypeConfigs[req.BizType].MonitorRetention)
}

func AgentRoleName(baseInfo structs.ClusterBaseInfo) string {
	return fmt.Sprintf(constants.AgentRoleNameTemplate, configs.GetServiceInfo().Env,
		baseInfo.BizType,
		baseInfo.TenantID,
		configs.GetServiceInfo().Region)
}

func AgentRoleAlias(vendor constants.ProvidedVendor) string {
	return fmt.Sprintf(constants.AgentRoleNameAliasTemplate, utils.ShortID())
}

type AgentBaseInfo struct {
	structs.ClusterBaseInfo
	Account string
}

func buildAgentBaseInfoFromReq(req message.CreateClusterReq) AgentBaseInfo {
	agentInfo := AgentBaseInfo{
		ClusterBaseInfo: req.ClusterBaseInfo,
	}
	if req.AgentEnvironment != nil && req.AgentEnvironment.AWS != nil {
		agentInfo.Account = req.AgentEnvironment.AWS.Account
	}
	return agentInfo
}

// AgentDataPlaneRoleNameWrap is a temporory function for finding AgentDataPlaneRoleName for aws agent.
// Since that agent deployed in aws might distribute in multi aws account under a single tenant,
// we are changing the name by adding an account suffix for aws agent, but the legacy role name does not contain the account info.
func AgentDataPlaneRoleNameWrap(ctx context.Context, namespace string, baseInfo AgentBaseInfo) (string, error) {
	if configs.GetServiceInfo().Vendor == string(constants.AWS) {
		legacyRoleName := fmt.Sprintf(constants.AgentDataplaneRoleNameTemplate, configs.GetServiceInfo().Env,
			baseInfo.BizType,
			baseInfo.TenantID,
			configs.GetServiceInfo().Region)
		dataRoleExisted, err := resourcemanager.GetIAMReaderWriter(ctx).IsRoleExists(ctx, legacyRoleName, namespace)
		if err != nil {
			log.Errorf("get legacy role failed, %s", err.Error())
			return "", err
		}
		if dataRoleExisted {
			iam, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, legacyRoleName, namespace)
			if err != nil {
				log.Errorf("get legacy role failed, %s", err.Error())
				return "", err
			}
			if iam.Spec.Aws.Account == baseInfo.Account {
				return legacyRoleName, nil
			}
		}
	}
	return AgentDataPlaneRoleName(baseInfo), nil
}

// Note: prefer using AgentDataPlaneRoleNameWrap instead of this function unless you know what you are doing.
func AgentDataPlaneRoleName(baseInfo AgentBaseInfo) string {
	switch configs.GetServiceInfo().Vendor {
	case string(constants.Azure):
		// for azure, we use an additional projectID as part of the uniq key,
		// since maximum of 20 federated identity credentials can be added to an application or user-assigned managed identity.
		// https://learn.microsoft.com/en-us/entra/workload-id/workload-identity-federation-considerations#unsupported-regions-user-assigned-managed-identities:~:text=A%20maximum%20of%2020%20federated%20identity%20credentials%20can%20be%20added%20to%20an%20application%20or%20user%2Dassigned%20managed%20identity.
		// this means a limit of "1 infra cluster + 19 tidb clusters" under a single aks(project).
		return fmt.Sprintf(constants.AzureAgentDataplaneRoleNameTemplate, configs.GetServiceInfo().Env,
			baseInfo.BizType,
			baseInfo.TenantID,
			baseInfo.ProjectID,
			configs.GetServiceInfo().Region)
	case string(constants.AWS):
		return fmt.Sprintf(constants.AWSAgentDataplaneRoleNameTemplate, configs.GetServiceInfo().Env,
			baseInfo.BizType,
			baseInfo.TenantID,
			configs.GetServiceInfo().Region,
			baseInfo.Account)
	default:
		return fmt.Sprintf(constants.AgentDataplaneRoleNameTemplate, configs.GetServiceInfo().Env,
			baseInfo.BizType,
			baseInfo.TenantID,
			configs.GetServiceInfo().Region)
	}
}

func AgentDataPlaneRoleAlias(vendor constants.ProvidedVendor) string {
	return fmt.Sprintf(constants.AgentDataplaneRoleNameAliasTemplate, utils.ShortID())
}

func BuildClusterRoleBindingResourceRequirements(req message.CreateClusterReq) (*rbacv1.ClusterRoleBinding, error) {
	return &rbacv1.ClusterRoleBinding{
		ObjectMeta: metav1.ObjectMeta{
			Name:      NamespaceName(req.BizType, req.TenantID),
			Namespace: NamespaceName(req.BizType, req.TenantID),
			Labels: map[string]string{
				constants.LabelClusterID:     req.ID,
				constants.LabelVendor:        string(req.Vendor),
				constants.LabelRegion:        req.Region,
				constants.LabelTenantID:      req.TenantID,
				constants.LabelProjectID:     req.ProjectID,
				constants.LabelApplicationID: req.ApplicationID,
				constants.LabelBizType:       req.BizType,
			},
			Annotations: map[string]string{},
		},
		Subjects: []rbacv1.Subject{
			{
				Kind:      "ServiceAccount",
				Name:      NamespaceName(req.BizType, req.TenantID),
				Namespace: NamespaceName(req.BizType, req.TenantID),
			},
		},
		RoleRef: rbacv1.RoleRef{
			Kind: "ClusterRole",
			Name: "prometheus",
		},
	}, nil
}

func BuildAgentRoleRequirements(req message.CreateClusterReq) (*v1.IAM, error) {
	vendor := configs.GetServiceInfo().Vendor
	switch vendor {
	case string(constants.AWS):
		return BuildAgentRoleRequirementsAWS(req)
	case string(constants.GCP):
		return BuildAgentRoleRequirementsGCP(req)
	case string(constants.Azure):
		return BuildAgentRoleRequirementsAzure(req)
	case string(constants.AliCloud):
		return BuildAgentRoleRequirementsAliCloud(req)
	default:
		return nil, errors.New(errors.ParameterInvalid, "vendor name is invalid")
	}
}

func BuildAgentRoleRequirementsAWS(req message.CreateClusterReq) (*v1.IAM, error) {
	policy := createAssumeRolePolicy(req.AgentAccessARN, NamespaceName(req.BizType, req.TenantID))
	policyJson, err := json.Marshal(policy)
	if err != nil {
		return nil, err
	}

	bucketName := ObjectStorageResourceName(req.ClusterBaseInfo)

	return &v1.IAM{
		ObjectMeta: buildIAMObjectMeta(req),
		Spec: v1.IAMSpec{
			Aws: v1.AWSIAM{
				AssumeRolePolicyDocument: base64.StdEncoding.EncodeToString(policyJson),
				Policies: []v1.AWSRolePolicy{
					{
						PolicyDocument: base64.StdEncoding.EncodeToString([]byte(buildPolicyDocument(bucketName, req))),
						PolicyName:     "AccessClusterData",
					},
				},
			},
		},
	}, nil
}

func BuildAgentRoleRequirementsAliCloud(req message.CreateClusterReq) (*v1.IAM, error) {
	policy := createAssumeRolePolicyAliCloud(req.AgentAccessARN, NamespaceName(req.BizType, req.TenantID))
	policyJson, err := json.Marshal(policy)
	if err != nil {
		return nil, err
	}

	bucketName := ObjectStorageResourceName(req.ClusterBaseInfo)

	return &v1.IAM{
		ObjectMeta: buildIAMObjectMeta(req),
		Spec: v1.IAMSpec{
			AliCloud: v1.AliCloudIAM{
				AssumeRolePolicyDocument: base64.StdEncoding.EncodeToString(policyJson),
				Policies: []v1.AliCloudPolicy{
					{
						PolicyDocument: base64.StdEncoding.EncodeToString([]byte(buildPolicyDocumentAliCloud(bucketName, req))),
						PolicyName:     fmt.Sprintf("AccessClusterData-%s", req.ID),
					},
				},
			},
		},
	}, nil
}

func buildPolicyDocument(bucketName string, req message.CreateClusterReq) string {
	if req.AgentSharedRole || len(req.AgentAccessAccount) > 0 {
		return fmt.Sprintf(accessS3SharedPolicyTemplate, bucketName, bucketName)
	} else {
		return fmt.Sprintf(accessS3PolicyTemplate,
			bucketName, req.TenantID,
			bucketName, req.TenantID,
			bucketName, bucketName, req.TenantID,
			bucketName, bucketName, req.TenantID,
			bucketName,
		)
	}
}

func buildPolicyDocumentAliCloud(bucketName string, req message.CreateClusterReq) string {
	if req.AgentSharedRole {
		return fmt.Sprintf(accessOSSSharedPolicyTemplate, bucketName, bucketName)
	} else {
		return fmt.Sprintf(accessOSSPolicyTemplate,
			bucketName, req.TenantID,
			bucketName, req.TenantID,
			bucketName, bucketName, req.TenantID,
			bucketName, bucketName, req.TenantID,
			bucketName,
		)
	}
}

const (
	GServicaAccount = "gserviceaccount.com"
	GCPSAPrefix     = "serviceAccount"
	AzureSAPrefix   = "system:serviceaccount"
)

func BuildAgentRoleRequirementsGCP(req message.CreateClusterReq) (*v1.IAM, error) {
	bucketName := ObjectStorageResourceName(req.ClusterBaseInfo)
	account, err := checkGCPServiceAccount(req.AgentAccessARN)
	if err != nil {
		return nil, err
	}
	useExternalPrincipal := req.AgentEnvironment == nil && strings.TrimSpace(req.AgentAccessARN) != ""
	assumeRoleAccounts := []string{account}
	if !useExternalPrincipal {
		assumeRoleAccounts = append([]string{
			fmt.Sprintf("%s:%s[%s/%s]", GCPSAPrefix, configs.GetOidc(), NamespaceName(req.BizType, req.TenantID), NamespaceName(req.BizType, req.TenantID)),
		}, assumeRoleAccounts...)
	}
	return &v1.IAM{
		ObjectMeta: buildIAMObjectMeta(req),
		Spec: v1.IAMSpec{
			Gcp: v1.GCPIAM{
				UseExternalPrincipal: useExternalPrincipal,
				AssumeRoleAccounts:   assumeRoleAccounts,
				Policies:             buildGcpPolicies(bucketName, req),
			},
		},
	}, nil
}

func BuildAgentRoleRequirementsAzure(req message.CreateClusterReq) (*v1.IAM, error) {
	bucketName := ObjectStorageResourceName(req.ClusterBaseInfo)
	useExternalPrincipal := req.AgentEnvironment == nil && strings.TrimSpace(req.AgentAccessARN) != ""

	azureIAM := &v1.AzureIAM{
		SubscriptionID: configs.GetSusbcriptionID(),
		ResourceGroup:  configs.GetResourceGroup(),
		RoleScope:      fmt.Sprintf(AzureRoleScopeTmpl, configs.GetSusbcriptionID(), configs.GetStorageResourceGroup(), configs.GetAzureStorageAccount(), bucketName),
	}

	if useExternalPrincipal {
		azureIAM.UseExternalPrincipal = true
		azureIAM.AdditionalRoleMembers = []string{req.AgentAccessARN}
		// External principal mode grants access at bucket scope only.
		azureIAM.RoleCondition = ""
	} else {
		azureIAM.WorkloadOIDCIssuer = configs.GetOidc()
		azureIAM.WorkloadServiceAccounts = []string{
			// the pattern of "prefix:{sa_namespace}:{sa_name}"
			fmt.Sprintf("%s:%s:%s", AzureSAPrefix, NamespaceName(req.BizType, req.TenantID), NamespaceName(req.BizType, req.TenantID)),
		}
		azureIAM.AdditionalRoleMembers = []string{req.AgentAccessARN}
		azureIAM.RoleCondition = fmt.Sprintf(AzureRoleConditionTmpl, bucketName, req.TenantID)
	}

	return &v1.IAM{
		ObjectMeta: buildIAMObjectMeta(req),
		Spec: v1.IAMSpec{
			Azure: azureIAM,
		},
	}, nil
}

func buildGcpPolicies(bucketName string, req message.CreateClusterReq) map[string][]v1.GCPRolePolicy {
	return map[string][]v1.GCPRolePolicy{
		v1.BuildGCPPolicyKey(v1.GCPResourceTypeCloudStorage, bucketName): {
			{
				PolicyName:      req.ID,
				PolicyRole:      "roles/storage.admin",
				PolicyCondition: "",
			},
		},
	}
}

func AppendAssumeEntity(ctx context.Context, role *v1.IAM, arn string) (*v1.IAM, bool, error) {
	switch configs.GetServiceInfo().Vendor {
	case string(constants.AWS):
		return AppendAssumeEntityAWS(ctx, role, arn)
	case string(constants.GCP):
		return AppendAssumeRoleAccountGCP(ctx, role, arn)
	case string(constants.Azure):
		// in azure, there is no need for assume operation.
		return nil, false, nil
	case string(constants.AliCloud):
		return AppendAssumeEntityAliCloud(ctx, role, arn)
	}
	return nil, false, errors.New(errors.ParameterInvalid, "vendor name is invalid")
}

func AppendAssumeEntityAWS(ctx context.Context, role *v1.IAM, arn string) (*v1.IAM, bool, error) {
	policy, err := GetIAMAssumePolicyAWS(ctx, role)
	if err != nil {
		return nil, false, err
	}
	newPolicy, updated := appendAssumeRolePolicy(*policy, arn)
	if !updated {
		return role, false, nil
	}
	newContentJson, err := json.Marshal(newPolicy)
	if err != nil {
		return nil, false, err
	}

	role.Spec.Aws.AssumeRolePolicyDocument = base64.StdEncoding.EncodeToString(newContentJson)
	return role, true, nil
}

func AppendAssumeEntityAliCloud(ctx context.Context, role *v1.IAM, arn string) (*v1.IAM, bool, error) {
	policy, err := GetIAMAssumePolicyAliCloud(ctx, role)
	if err != nil {
		return nil, false, err
	}
	newPolicy, updated := appendAssumeRolePolicyAliCloud(*policy, arn)
	if !updated {
		return role, false, nil
	}
	newContentJson, err := json.Marshal(newPolicy)
	if err != nil {
		return nil, false, err
	}

	role.Spec.AliCloud.AssumeRolePolicyDocument = base64.StdEncoding.EncodeToString(newContentJson)
	return role, true, nil
}

func GetIAMAssumePolicyAWS(ctx context.Context, role *v1.IAM) (*AssumeRolePolicy, error) {
	content, err := base64.StdEncoding.DecodeString(role.Spec.Aws.AssumeRolePolicyDocument)
	if err != nil {
		return nil, err
	}

	policy := &AssumeRolePolicy{}
	err = json.Unmarshal(content, policy)
	if err != nil {
		return nil, err
	}
	return policy, nil
}

func GetIAMAssumePolicyAliCloud(ctx context.Context, role *v1.IAM) (*AssumeRolePolicy, error) {
	content, err := base64.StdEncoding.DecodeString(role.Spec.AliCloud.AssumeRolePolicyDocument)
	if err != nil {
		return nil, err
	}

	policy := &AssumeRolePolicy{}
	err = json.Unmarshal(content, policy)
	if err != nil {
		return nil, err
	}
	return policy, nil
}

func AppendAssumeRoleAccountGCP(ctx context.Context, role *v1.IAM, arn string) (*v1.IAM, bool, error) {
	account := role.Spec.Gcp.AssumeRoleAccounts
	arn, err := checkGCPServiceAccount(arn)
	if err != nil {
		return nil, false, err
	}

	newAccount, updated := appendAssumeRoleAccountGCP(account, arn)
	if !updated {
		return role, false, nil
	}

	role.Spec.Gcp.AssumeRoleAccounts = newAccount
	return role, true, nil
}

type AssumeRolePolicy struct {
	Version   string      `json:"Version"`
	Statement []Statement `json:"Statement"`
}

type Statement struct {
	Effect    string                 `json:"Effect"`
	Principal map[string]interface{} `json:"Principal"`
	Action    string                 `json:"Action"`
	Condition map[string]interface{} `json:"Condition"`
}

func agentAssumeStatement(arn string) Statement {
	return Statement{
		Effect: "Allow",
		Principal: map[string]interface{}{
			"AWS": arn,
		},
		Action:    "sts:AssumeRole",
		Condition: map[string]interface{}{},
	}
}

func agentAssumeStatementAliCloud(arn string) Statement {
	return Statement{
		Effect: "Allow",
		Principal: map[string]interface{}{
			"RAM": arn,
		},
		Action:    "sts:AssumeRole",
		Condition: map[string]interface{}{},
	}
}

func serviceAccountAssumeStatement(namespace string) Statement {
	return Statement{
		Effect: "Allow",
		Principal: map[string]interface{}{
			"Federated": configs.GetOidcArn(),
		},
		Action: "sts:AssumeRoleWithWebIdentity",
		Condition: map[string]interface{}{
			"StringEquals": map[string]string{
				configs.GetOidc() + ":aud": "sts.amazonaws.com",
				configs.GetOidc() + ":sub": fmt.Sprintf("system:serviceaccount:%s:%s", namespace, namespace),
			},
		},
	}
}

func serviceAccountAssumeStatementAliCloud(namespace string) Statement {
	return Statement{
		Effect: "Allow",
		Principal: map[string]interface{}{
			"Federated": configs.GetOidcArn(),
		},
		Action: "sts:AssumeRole",
		Condition: map[string]interface{}{
			"StringEquals": map[string]string{
				"oidc:aud": "sts.aliyuncs.com",
				"oidc:sub": fmt.Sprintf("system:serviceaccount:%s:%s", namespace, namespace),
			},
		},
	}
}

func appendAssumeRolePolicy(policy AssumeRolePolicy, arn string) (AssumeRolePolicy, bool) {
	for _, s := range policy.Statement {
		if original, ok := s.Principal["AWS"]; ok && original == arn {
			return policy, false
		}
	}

	policy.Statement = append(policy.Statement, agentAssumeStatement(arn))
	return policy, true
}

func appendAssumeRolePolicyAliCloud(policy AssumeRolePolicy, arn string) (AssumeRolePolicy, bool) {
	for _, s := range policy.Statement {
		if original, ok := s.Principal["RAM"]; ok && original == arn {
			return policy, false
		}
	}

	policy.Statement = append(policy.Statement, agentAssumeStatementAliCloud(arn))
	return policy, true
}

func appendAssumeRoleAccountGCP(account []string, arn string) ([]string, bool) {
	for _, a := range account {
		if a == arn {
			return account, false
		}
	}

	account = append(account, arn)
	return account, true
}

func buildIAMObjectMeta(req message.CreateClusterReq) metav1.ObjectMeta {
	return metav1.ObjectMeta{
		Name:      AgentRoleName(req.ClusterBaseInfo),
		Namespace: NamespaceName(req.BizType, req.TenantID),
		Labels: map[string]string{
			constants.LabelVendor:    string(req.Vendor),
			constants.LabelRegion:    req.Region,
			constants.LabelBizType:   req.BizType,
			constants.LabelTenantID:  req.TenantID,
			constants.LabelProjectID: req.ProjectID,
			constants.LabelAlias:     AgentRoleAlias(req.Vendor),
		},
	}
}

func createAssumeRolePolicy(arn string, namespace string) AssumeRolePolicy {
	return AssumeRolePolicy{
		Version: "2012-10-17",
		Statement: []Statement{
			serviceAccountAssumeStatement(namespace),
			agentAssumeStatement(arn),
		},
	}
}

func createAssumeRolePolicyAliCloud(arn string, namespace string) AssumeRolePolicy {
	return AssumeRolePolicy{
		Version: "1",
		Statement: []Statement{
			serviceAccountAssumeStatementAliCloud(namespace),
			agentAssumeStatementAliCloud(arn),
		},
	}
}

func checkGCPServiceAccount(name string) (string, error) {
	matched, err := regexp.MatchString("[^\\s]{1,}@[^\\s]{1,}\\.iam\\.gserviceaccount\\.com", name)
	if err != nil {
		return name, err
	}
	if !matched {
		return name, errors.New(errors.ParameterInvalid, fmt.Sprintf("service account %s is invalid", name))
	} else if matched && !strings.HasPrefix(name, GCPSAPrefix) {
		name = fmt.Sprintf("%s:%s", GCPSAPrefix, name)
		return name, nil
	} else {
		return name, nil
	}
}

var accessS3SharedPolicyTemplate = `{
    "Version": "2012-10-17",
    "Statement": [
		{
			"Effect": "Allow",
			"Action": [
				"s3:ListBucket"
			],
			"Resource": "arn:aws:s3:::%s"
		},
        {
            "Effect": "Allow",
            "Action": [
                "s3:*"
			],
            "Resource": "arn:aws:s3:::%s/*"
        }
    ]
}`

var accessS3PolicyTemplate = `{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:*"
			],
            "Resource": "arn:aws:s3:::%s/0/%s/*"
        },
		{
            "Effect": "Allow",
            "Action": [
                "s3:*"
			],
            "Resource": "arn:aws:s3:::%s/1/%s/*"
        },
		{
            "Effect": "Allow",
            "Action": [
                "s3:*"
			],
            "Resource": "arn:aws:s3:::%s/%s/0/%s/*"
        },
		{
            "Effect": "Allow",
            "Action": [
                "s3:*"
			],
            "Resource": "arn:aws:s3:::%s/%s/1/%s/*"
        },
		{
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket"
            ],
            "Resource": "arn:aws:s3:::%s"
        }
    ]
}`

var assumeRolePolicyTemplate = `{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
			"Action": "sts:AssumeRole",
            "Resource": "*"
        }
    ]
}`

var accessOSSSharedPolicyTemplate = `{
    "Version": "1",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "oss:*",
            "Resource": [
                "acs:oss:*:*:%s/*",
                "acs:oss:*:*:%s"
            ]
        }
    ]
}`

var accessOSSPolicyTemplate = `{
    "Version": "1",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "oss:*",
            "Resource": [
                "acs:oss:*:*:%s/0/%s/*",
                "acs:oss:*:*:%s/1/%s/*",
                "acs:oss:*:*:%s/%s/0/%s/*",
                "acs:oss:*:*:%s/%s/1/%s/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": "oss:ListObjects",
            "Resource": "acs:oss:*:*:%s"
        }
    ]
}`
var assumeRolePolicyTemplateAliCloud = `{
    "Version": "1",
    "Statement": [
        {
            "Effect": "Allow",
			"Action": "sts:AssumeRole",
            "Resource": "*"
        }
    ]
}`

func buildDataPlaneIAMObjectMeta(req message.CreateClusterReq) metav1.ObjectMeta {
	return metav1.ObjectMeta{
		Name:      AgentDataPlaneRoleName(buildAgentBaseInfoFromReq(req)),
		Namespace: NamespaceName(req.BizType, req.TenantID),
		Labels: map[string]string{
			constants.LabelVendor:    string(req.Vendor),
			constants.LabelRegion:    req.Region,
			constants.LabelBizType:   req.BizType,
			constants.LabelTenantID:  req.TenantID,
			constants.LabelProjectID: req.ProjectID,
			constants.LabelAlias:     AgentDataPlaneRoleAlias(req.Vendor),
		},
	}
}

func BuildAgentDataPlaneRoleRequirements(req message.CreateClusterReq) (*v1.IAM, error) {
	vendor := configs.GetServiceInfo().Vendor
	switch constants.ProvidedVendor(vendor) {

	case constants.AWS:
		return BuildAgentDataPlaneRoleRequirementsAWS(req)
	case constants.AliCloud:
		return BuildAgentDataPlaneRoleRequirementsAliCloud(req)
	case constants.GCP:
		return BuildAgentDataPlaneRoleRequirementsGCP(req)
	case constants.Azure:
		return BuildAgentDataRoleRequirementsAzure(req), nil
	default:
		return nil, errors.New(errors.ParameterInvalid, "vendor name is invalid")
	}
}

func BuildAgentDataPlaneRoleRequirementsAWS(req message.CreateClusterReq) (*v1.IAM, error) {
	policy := createDataPlaneAssumeRolePolicy(req.AgentEnvironment.AWS.Account, req.AgentEnvironment.AWS.OIDC)
	policyJson, err := json.Marshal(policy)
	if err != nil {
		return nil, err
	}
	return &v1.IAM{
		ObjectMeta: buildDataPlaneIAMObjectMeta(req),
		Spec: v1.IAMSpec{
			Aws: v1.AWSIAM{
				Account:                  req.AgentEnvironment.AWS.Account,
				AssumeRolePolicyDocument: base64.StdEncoding.EncodeToString(policyJson),
				Policies: []v1.AWSRolePolicy{
					{
						PolicyDocument: base64.StdEncoding.EncodeToString([]byte(buildDataPlanePolicyDocument())),
						PolicyName:     "AssumePlolicy",
					},
				},
			},
		},
	}, nil
}

func BuildAgentDataPlaneRoleRequirementsAliCloud(req message.CreateClusterReq) (*v1.IAM, error) {
	policy := createDataPlaneAssumeRolePolicyAliCloud(req.AgentEnvironment.AliCloud.Account, req.AgentEnvironment.AliCloud.OIDC)
	policyJson, err := json.Marshal(policy)
	if err != nil {
		return nil, err
	}
	return &v1.IAM{
		ObjectMeta: buildDataPlaneIAMObjectMeta(req),
		Spec: v1.IAMSpec{
			AliCloud: v1.AliCloudIAM{
				Account:                  req.AgentEnvironment.AliCloud.Account,
				AssumeRolePolicyDocument: base64.StdEncoding.EncodeToString(policyJson),
				Policies: []v1.AliCloudPolicy{
					{
						PolicyDocument: base64.StdEncoding.EncodeToString([]byte(buildDataPlanePolicyDocumentAliCloud())),
						PolicyName:     "AssumePlolicy",
					},
				},
			},
		},
	}, nil
}

func createDataPlaneAssumeRolePolicy(account, oidc string) AssumeRolePolicy {
	return AssumeRolePolicy{
		Version: "2012-10-17",
		Statement: []Statement{
			serviceDataPlaneAccountAssumeStatement(account, oidc),
		},
	}
}

func createDataPlaneAssumeRolePolicyAliCloud(account, oidc string) AssumeRolePolicy {
	return AssumeRolePolicy{
		Version: "1",
		Statement: []Statement{
			serviceDataPlaneAccountAssumeStatementAliCloud(account, oidc),
		},
	}
}

func serviceDataPlaneAccountAssumeStatement(account, oidc string) Statement {
	return Statement{
		Effect: "Allow",
		Principal: map[string]interface{}{
			"Federated": fmt.Sprintf("arn:aws:iam::%s:oidc-provider/%s", account, oidc),
		},
		Action: "sts:AssumeRoleWithWebIdentity",
		Condition: map[string]interface{}{
			"StringEquals": map[string]string{
				oidc + ":aud": "sts.amazonaws.com",
			},
		},
	}
}

func serviceDataPlaneAccountAssumeStatementAliCloud(account, oidc string) Statement {
	return Statement{
		Effect: "Allow",
		Principal: map[string]interface{}{
			"Federated": []string{
				fmt.Sprintf("acs:ram::%s:oidc-provider/%s", account, oidc),
			},
		},
		Action: "sts:AssumeRoleWithOIDC",
		Condition: map[string]interface{}{
			"StringEquals": map[string]string{
				fmt.Sprintf("%s:aud", oidc): "sts.aliyuncs.com",
			},
		},
	}
}

func buildDataPlanePolicyDocument() string {
	return assumeRolePolicyTemplate
}

func buildDataPlanePolicyDocumentAliCloud() string {
	return assumeRolePolicyTemplateAliCloud
}

func BuildAgentDataPlaneRoleRequirementsGCP(req message.CreateClusterReq) (*v1.IAM, error) {
	return &v1.IAM{
		ObjectMeta: buildDataPlaneIAMObjectMeta(req),
		Spec: v1.IAMSpec{
			Gcp: v1.GCPIAM{
				ProjectID: req.AgentEnvironment.GCP.GCPProject,
				AssumeRoleAccounts: []string{
					//add monitor sa
					fmt.Sprintf("%s:%s[%s]", GCPSAPrefix, getWorkspace(req.AgentEnvironment.GCP.GCPProject), DataPlaneSANamespaceName(req.Vendor, req.ApplicationID)),
				},
			},
		},
	}, nil
}

func getWorkspace(project string) string {
	return fmt.Sprintf("%s.svc.id.goog", project)
}

func AppendDataPlaneAssumeEntity(ctx context.Context, role *v1.IAM, req message.CreateClusterReq) (*v1.IAM, bool, error) {
	switch configs.GetServiceInfo().Vendor {
	// todo iam
	case string(constants.AWS):
		// in aws we don't need to update the assume role policy
		return AppendTrustPolicyStatement(ctx, role.DeepCopy(), req)
	case string(constants.GCP):
		return AppendDataPlaneAssumeRoleAccountGCP(ctx, role.DeepCopy(), req)
	case string(constants.Azure):
		return AppendDataplaneWorkloadSA(ctx, role.DeepCopy(), req)
	case string(constants.AliCloud):
		return AppendTrustPolicyStatementAliCloud(ctx, role.DeepCopy(), req)
	}

	return nil, false, errors.New(errors.ParameterInvalid, "vendor name is invalid")
}

func AppendTrustPolicyStatement(ctx context.Context, role *v1.IAM, req message.CreateClusterReq) (*v1.IAM, bool, error) {
	trustPolicy, err := base64.StdEncoding.DecodeString(role.Spec.Aws.AssumeRolePolicyDocument)
	if err != nil {
		return nil, false, err
	}
	currentTrustPolicy := &AssumeRolePolicy{}
	if err := json.Unmarshal(trustPolicy, currentTrustPolicy); err != nil {
		return nil, false, err
	}

	policyStmt := serviceDataPlaneAccountAssumeStatement(req.AgentEnvironment.AWS.Account, req.AgentEnvironment.AWS.OIDC)
	exists := false
	for _, stmt := range currentTrustPolicy.Statement {
		if reflect.DeepEqual(stmt.Principal, policyStmt.Principal) {
			exists = true
			break
		}
	}
	if exists {
		return role, false, nil
	}
	currentTrustPolicy.Statement = append(currentTrustPolicy.Statement, policyStmt)
	policyJson, err := json.Marshal(currentTrustPolicy)
	if err != nil {
		return nil, false, err
	}
	role.Spec.Aws.AssumeRolePolicyDocument = base64.StdEncoding.EncodeToString(policyJson)
	return role, true, nil
}

func AppendTrustPolicyStatementAliCloud(ctx context.Context, role *v1.IAM, req message.CreateClusterReq) (*v1.IAM, bool, error) {
	trustPolicy, err := base64.StdEncoding.DecodeString(role.Spec.AliCloud.AssumeRolePolicyDocument)
	if err != nil {
		return nil, false, err
	}
	currentTrustPolicy := &AssumeRolePolicy{}
	if err := json.Unmarshal(trustPolicy, currentTrustPolicy); err != nil {
		return nil, false, err
	}

	policyStmt := serviceDataPlaneAccountAssumeStatementAliCloud(req.AgentEnvironment.AliCloud.Account, req.AgentEnvironment.AliCloud.OIDC)
	exists := false
	for _, stmt := range currentTrustPolicy.Statement {
		if reflect.DeepEqual(stmt.Principal, policyStmt.Principal) {
			exists = true
			break
		}
	}
	if exists {
		return role, false, nil
	}
	currentTrustPolicy.Statement = append(currentTrustPolicy.Statement, policyStmt)
	policyJson, err := json.Marshal(currentTrustPolicy)
	if err != nil {
		return nil, false, err
	}
	role.Spec.AliCloud.AssumeRolePolicyDocument = base64.StdEncoding.EncodeToString(policyJson)
	return role, true, nil
}

func AppendDataPlaneAssumeRoleAccountGCP(ctx context.Context, role *v1.IAM, req message.CreateClusterReq) (*v1.IAM, bool, error) {
	account := role.Spec.Gcp.AssumeRoleAccounts
	arn := getGCPDataPlaneServiceAccount(req.Vendor, req.AgentEnvironment.GCP.GCPProject, req.ApplicationID)
	newAccount, updated := appendAssumeRoleAccountGCP(account, arn)
	if !updated {
		return role, false, nil
	}

	role.Spec.Gcp.AssumeRoleAccounts = newAccount
	return role, true, nil
}

func AppendDataplaneWorkloadSA(ctx context.Context, role *v1.IAM, req message.CreateClusterReq) (*v1.IAM, bool, error) {
	newSA := fmt.Sprintf("%s:%s", AzureSAPrefix, DataPlaneSANamespaceName(req.Vendor, req.ApplicationID))
	if utils.IsIncludeString(role.Spec.Azure.WorkloadServiceAccounts, newSA) {
		return role, false, nil
	}
	role.Spec.Azure.WorkloadServiceAccounts = append(role.Spec.Azure.WorkloadServiceAccounts, newSA)
	return role, true, nil
}

func getGCPDataPlaneServiceAccount(vendor constants.ProvidedVendor, projectID, applicationID string) string {
	//get the service account name from the application id
	return fmt.Sprintf("%s:%s[%s]", GCPSAPrefix, getWorkspace(projectID), DataPlaneSANamespaceName(vendor, applicationID))
}

func appendAgent(ctx context.Context, cluster *v1.Cluster, req message.CreateClusterReq) error {
	if req.AgentEnvironment == nil {
		return nil
	}
	cluster.Spec.Agent = &v1.Agent{
		EndpointNamePerfix: getEndpointNamePerfix(req.BizType, req.ApplicationID),
		ReleaseNameSpace:   req.AgentEnvironment.ReleaseNameSpace,
		Dataplane: v1.Dataplane{
			ClusterName:       req.AgentEnvironment.Dataplane.ClusterName,
			ClusterEndpoint:   req.AgentEnvironment.Dataplane.ClusterEndpoint,
			ClusterCAData:     req.AgentEnvironment.Dataplane.ClusterCAData,
			ClusterKubeconfig: req.AgentEnvironment.Dataplane.ClusterKubeConfig,
		},
	}
	//add iam information to annotation
	//get sa role
	ns := NamespaceName(req.BizType, req.TenantID)
	if !strings.Contains(req.BizType, BizeTypeDP) && !strings.Contains(req.BizType, BizeTypeServerless) {
		dataroleName, err := AgentDataPlaneRoleNameWrap(ctx, ns, buildAgentBaseInfoFromReq(req))
		if err != nil {
			return err
		}
		datarole, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, dataroleName, ns)
		if err != nil {
			log.Errorf("get role failed, %s", err.Error())
			return err
		}
		cluster.Annotations[constants.AnnotationAgentServiceAccountRole] = BuildDataPlaneServiceAccountRoleARN(datarole)
	}

	if !AgentRoleNotNeeded(&req.ClusterBaseInfo) {
		// get assume role
		roleName := AgentRoleName(req.ClusterBaseInfo)
		role, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, roleName, ns)
		if err != nil {
			log.Errorf("get role failed, %s", err.Error())
			return err
		}
		cluster.Annotations[constants.AnnotationAgentAssumeRole] = buildServiceAccountRoleARN(role)
	}
	//append cloud dependant information
	switch configs.GetServiceInfo().Vendor {
	case string(constants.AWS):
		{
			cluster.Spec.Agent.AWS = &v1.AgentAWS{
				VpcID:           req.AgentEnvironment.AWS.VpcID,
				NodeSubnetIDs:   req.AgentEnvironment.AWS.NodeSubnetIDs,
				SecurityGroupID: req.AgentEnvironment.AWS.SecurityGroupID,
				Account:         req.AgentEnvironment.AWS.Account,
				OIDC:            req.AgentEnvironment.AWS.OIDC,
			}
		}
	case string(constants.GCP):
		{
			cluster.Spec.Agent.GCP = &v1.AgentGCP{
				GCPProject:  req.AgentEnvironment.GCP.GCPProject,
				NetworkName: req.AgentEnvironment.GCP.NetworkName,
				Subnets:     req.AgentEnvironment.GCP.Subnets,
			}
		}
	case string(constants.Azure):
		{
			cluster.Spec.Agent.Azure = &v1.AgentAzure{
				NetworkName:    req.AgentEnvironment.Azure.NetworkName,
				SubscriptionID: req.AgentEnvironment.Azure.SubscriptionID,
				ResourceGroup:  req.AgentEnvironment.Azure.ResourceGroup,
				Subnet:         req.AgentEnvironment.Azure.Subnet,
				IAMRoleRef: v1.Reference{
					Namespace: ns,
					Name:      AgentDataPlaneRoleName(buildAgentBaseInfoFromReq(req)),
				},
			}
		}
	}
	//append observed information
	if req.Observed.TiDBClusterTopology != nil {
		cluster.Spec.Agent.Observed = &v1.Observed{
			TiDBVersion: req.Observed.TiDBVersion,
			TiDBClusterTopology: &v1.ClusterTopologyType{
				TiDB: v1.TiDB{
					Replicas: req.Observed.TiDBClusterTopology.TiDB.Replicas,
				},
				PD: v1.PD{
					Replicas: req.Observed.TiDBClusterTopology.PD.Replicas,
				},
				TiKV: v1.TiKV{
					Replicas: req.Observed.TiDBClusterTopology.TiKV.Replicas,
				},
			},
		}
		if req.Observed.TiDBClusterTopology.TiFlash != nil {
			cluster.Spec.Agent.Observed.TiDBClusterTopology.TiFlash = &v1.TiFlash{
				Replicas: req.Observed.TiDBClusterTopology.TiFlash.Replicas,
			}
		}
	}
	return nil
}

func getEndpointNamePerfix(bizeType, applicationID string) string {

	if applicationID == applicationIDK8sInfra {
		//dataplane
		return O11yEndpointNamePrefixDataplane
	} else if bizeType == BizeTypeDP {
		//dp
		return O11yEndpointNamePrefixDataMigration
	}
	//tidb cluster
	return O11yEndpointNamePrefixTiDB
}

const gcpDataPlaneRoleArnTemplate = "%s@%s.iam.gserviceaccount.com"

func BuildDataPlaneServiceAccountRoleARN(iam *v1.IAM) string {
	switch configs.GetServiceInfo().Vendor {
	case string(constants.AWS):
		return fmt.Sprintf(roleArnTemplate, iam.Spec.Aws.Account, iam.Labels[constants.LabelAlias])
	case string(constants.GCP):
		return fmt.Sprintf(gcpDataPlaneRoleArnTemplate, iam.Labels[constants.LabelAlias], iam.Spec.Gcp.ProjectID)
	case string(constants.Azure):
		// in azure, identity's clientId is used as the value of serviceAccount annotation,
		// but the clientId is generated after identity creation, so it might be empty.
		// If it's empty, it will be finally set to the correct clientId during the agent cr creation
		// in the cluster_controller reconciling process.
		return iam.Status.ClientId
	case string(constants.AliCloud):
		return fmt.Sprintf(aliRoleArnTemplate, iam.Spec.AliCloud.Account, iam.Labels[constants.LabelAlias])
	}
	return ""
}

func BuildAgentDataRoleRequirementsAzure(req message.CreateClusterReq) *v1.IAM {
	bucketName := ObjectStorageResourceName(req.ClusterBaseInfo)
	return &v1.IAM{
		ObjectMeta: buildDataPlaneIAMObjectMeta(req),
		Spec: v1.IAMSpec{
			Azure: &v1.AzureIAM{
				WorkloadOIDCIssuer: req.AgentEnvironment.Azure.OIDC,
				WorkloadServiceAccounts: []string{
					fmt.Sprintf("%s:%s", AzureSAPrefix, DataPlaneSANamespaceName(req.Vendor, req.ApplicationID)),
				},
				SubscriptionID: req.AgentEnvironment.Azure.SubscriptionID,
				ResourceGroup:  req.AgentEnvironment.Azure.ResourceGroup,
				RoleCondition:  fmt.Sprintf(AzureRoleConditionTmpl, bucketName, req.TenantID),
				RoleScope:      fmt.Sprintf(AzureRoleScopeTmpl, configs.GetSusbcriptionID(), configs.GetStorageResourceGroup(), configs.GetAzureStorageAccount(), bucketName),
			},
		},
	}
}