/*
Copyright 2022.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package iam

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"strings"

	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/authorization/armauthorization/v3"
	mapset "github.com/deckarep/golang-set"
	"github.com/go-logr/logr"
	"github.com/google/go-cmp/cmp"
	iamgcpapi "google.golang.org/api/iam/v1"
	iamgcppb "google.golang.org/genproto/googleapis/iam/v1"
	"google.golang.org/genproto/googleapis/type/expr"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	observabilitypingcapcomv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/metrics"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/common/utils"
	"github.com/tidbcloud/observability/pkg/resourceworker/controllers/reconciler"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/azure"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/gcp"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/types"
)

const (
	StorageBlobDataContributorRoleDefIDTmpl = "/subscriptions/%s/providers/Microsoft.Authorization/roleDefinitions/ba92f5b4-2d11-453d-a403-e96b0029c9fe"
)

// IAMReconciler reconciles a IAM object
type IAMReconciler struct {
	reconciler *reconciler.Reconciler
	Log        logr.Logger
}

const (
	iamfinalizerName            = "observability.pingcap.com/iam"
	accountConditionPrefix      = "o11yr"
	serviceAccoutPrefix         = "serviceAccount:"
	errNotIAM                   = "not IAM resource"
	errNotFoundRoleName         = "not found role name"
	errNotSupportProvidedVendor = "not support provided vendor"
)

func NewIAMReconciler(mgr ctrl.Manager, log logr.Logger, serviceInfo structs.ServiceInfo) *IAMReconciler {
	return &IAMReconciler{
		Log: log,
		reconciler: &reconciler.Reconciler{
			Client: mgr.GetClient(),
			ExternalObj: iamConnector{
				cli:         mgr.GetClient(),
				serviceInfo: serviceInfo,
			},
			NewResource: func() client.Object {
				return &observabilitypingcapcomv1.IAM{}
			},
		},
	}
}

type iamConnector struct {
	cli         client.Client
	serviceInfo structs.ServiceInfo
}

func (iamc iamConnector) getAWSClient() (providers.IAMClient, error) {
	return providers.GetAWSIAMClient(), nil
}

func (iamc iamConnector) getAliCloudClient() (providers.IAMClientAliCloud, error) {
	return providers.GetAliCloudIAMClient(), nil
}

func (iamc iamConnector) getGCPClient() (providers.IAMClientGCP, error) {
	return providers.GetGCPIAMClient(), nil
}

func (iamc iamConnector) getAzureClient(iam *observabilitypingcapcomv1.IAM) (providers.IAMClientAzure, error) {
	cred, err := azidentity.NewDefaultAzureCredential(nil)
	if err != nil {
		return nil, fmt.Errorf("failed to create default Azure credentials: %w", err)
	}
	return azure.NewIAMClient(iam.Spec.Azure.SubscriptionID, iam.Spec.Azure.ResourceGroup, iamc.serviceInfo.Region, cred)
}

func (iamc iamConnector) Connect(ctx context.Context, obj client.Object) (_ reconciler.ExternalObjClient, errResult error) {
	awsclient, err := iamc.getAWSClient()
	if err != nil {
		return nil, err
	}
	iam, ok := obj.(*observabilitypingcapcomv1.IAM)
	if !ok {
		return nil, fmt.Errorf("type mismatch expect iam got %v", obj)
	}
	if iam.Spec.Aws.Account != "" {
		// shloud assume role and create role in this account
		bizType := iam.GetLabels()[constants.LabelBizType]
		if _, ok := iamc.serviceInfo.BizTypeConfigs[bizType]; !ok {
			return nil, fmt.Errorf("not supported biz type %s", bizType)
		}
		managerole := fmt.Sprintf("arn:aws:iam::%s:role/%s", iam.Spec.Aws.Account, iamc.serviceInfo.BizTypeConfigs[bizType].DataplaneRole)
		awsclient = awsclient.AssumeRole(ctx, managerole)
	}

	gcpclient, err := iamc.getGCPClient()
	if err != nil {
		return nil, err
	}

	var azureCli providers.IAMClientAzure
	if iam.Spec.Azure != nil {
		azureCli, err = iamc.getAzureClient(iam)
		if err != nil {
			return nil, err
		}
	}

	aliClient, err := iamc.getAliCloudClient()
	if err != nil {
		return nil, err
	}
	if iam.Spec.AliCloud.Account != "" {
		// should assume role and create role in this account
		bizType := iam.GetLabels()[constants.LabelBizType]
		if _, ok := iamc.serviceInfo.BizTypeConfigs[bizType]; !ok {
			return nil, fmt.Errorf("not supported biz type %s", bizType)
		}

		managerole := fmt.Sprintf("acs:ram::%s:role/%s", iam.Spec.AliCloud.Account, iamc.serviceInfo.BizTypeConfigs[bizType].DataplaneRole)
		aliClient = aliClient.AssumeRole(ctx, managerole)
	}

	return &IAM{
		iamcli:      awsclient,
		gcpcli:      gcpclient,
		azureCli:    azureCli,
		aliCli:      aliClient,
		cli:         iamc.cli,
		serviceInfo: iamc.serviceInfo,
	}, nil
}

type IAM struct {
	iamcli      providers.IAMClient
	gcpcli      providers.IAMClientGCP
	azureCli    providers.IAMClientAzure
	aliCli      providers.IAMClientAliCloud
	cli         client.Client
	serviceInfo structs.ServiceInfo
}

type roleWarp struct {
	AWSAssumeRolePolicyDocument      *string
	AliCloudAssumeRolePolicyDocument *string
	GCPAssumeRolePolicy              *iamgcpapi.Policy
	AzureWorkloadIdentity            *types.AzureWorkloadIdentity
}

func buildAzureRoleAssignmentProperties(subscriptionID, principalID, roleCondition string, principalType *armauthorization.PrincipalType) *armauthorization.RoleAssignmentProperties {
	props := &armauthorization.RoleAssignmentProperties{
		PrincipalID:      to.Ptr(principalID),
		RoleDefinitionID: to.Ptr(fmt.Sprintf(StorageBlobDataContributorRoleDefIDTmpl, subscriptionID)),
	}
	if principalType != nil {
		props.PrincipalType = principalType
	}
	if strings.TrimSpace(roleCondition) != "" {
		props.Condition = to.Ptr(roleCondition)
		props.ConditionVersion = to.Ptr("2.0")
	}
	return props
}

func getAzureAdditionalPrincipalIDs(iamv1 *observabilitypingcapcomv1.IAM) []string {
	if iamv1 == nil || iamv1.Spec.Azure == nil {
		return nil
	}
	ret := make([]string, 0, len(iamv1.Spec.Azure.AdditionalRoleMembers))
	for _, id := range iamv1.Spec.Azure.AdditionalRoleMembers {
		id = strings.TrimSpace(id)
		if id == "" {
			continue
		}
		ret = append(ret, id)
	}
	return ret
}

func getAzureLastAdditionalPrincipalIDs(iamv1 *observabilitypingcapcomv1.IAM) []string {
	if iamv1 == nil || iamv1.Spec.Azure == nil {
		return nil
	}
	ret := make([]string, 0, len(iamv1.Spec.Azure.LastAdditionalRoleMembers))
	for _, id := range iamv1.Spec.Azure.LastAdditionalRoleMembers {
		id = strings.TrimSpace(id)
		if id == "" {
			continue
		}
		ret = append(ret, id)
	}
	return ret
}

func unionStrings(left, right []string) []string {
	set := mapset.NewSet()
	for _, v := range left {
		set.Add(v)
	}
	for _, v := range right {
		set.Add(v)
	}
	ret := make([]string, 0, set.Cardinality())
	for v := range set.Iter() {
		ret = append(ret, v.(string))
	}
	return ret
}

func useGCPExternalPrincipal(iamv1 *observabilitypingcapcomv1.IAM) bool {
	if iamv1 == nil {
		return false
	}
	return iamv1.Spec.Gcp.UseExternalPrincipal
}

func (iam *IAM) getRoleGCPExternal(ctx context.Context, cli providers.IAMClientGCP, iamv1 *observabilitypingcapcomv1.IAM, name string) (*roleWarp, bool, error) {
	for k, policyList := range iamv1.Spec.Gcp.Policies {
		_, resource, err := observabilitypingcapcomv1.GetFromGCPPolicyKey(k)
		if err != nil {
			return nil, false, err
		}
		policy, exists, err := cli.GetBucketPolicy(ctx, resource)
		if err != nil {
			return nil, false, err
		}
		if !exists {
			continue
		}
		for _, p := range policyList {
			title := buildGCPConditionTitle(accountConditionPrefix, name, p.PolicyName)
			if idx, _ := peekGCPPolicy(policy.Bindings, title); idx >= 0 {
				return &roleWarp{}, true, nil
			}
		}
	}
	return nil, false, nil
}

func (iam *IAM) getRole(ctx context.Context, iamv1 *observabilitypingcapcomv1.IAM, name string) (*roleWarp, bool, error) {
	switch iam.serviceInfo.Vendor {
	case string(constants.AWS):
		{
			out, exists, err := iam.iamcli.GetRole(ctx, name)
			return func() *roleWarp {
				if out == nil || out.Role == nil {
					return nil
				}
				return &roleWarp{
					AWSAssumeRolePolicyDocument: out.Role.AssumeRolePolicyDocument,
				}
			}(), exists, err
		}
	case string(constants.GCP):
		{
			var cli providers.IAMClientGCP
			if iamv1.Spec.Gcp.ProjectID != "" {
				cli = iam.gcpcli.AssumeProject(ctx, iamv1.Spec.Gcp.ProjectID)
			} else {
				cli = iam.gcpcli
			}
			if useGCPExternalPrincipal(iamv1) {
				return iam.getRoleGCPExternal(ctx, cli, iamv1, name)
			}
			_, exists, err := cli.GetServiceAccount(ctx, name)
			if !exists || err != nil {
				return nil, exists, err
			}
			//query AssumeRolePolicy
			policy, err := cli.GetServiceAccountIAM(ctx, name)
			return func() *roleWarp {
				if policy == nil {
					return nil
				}
				return &roleWarp{
					GCPAssumeRolePolicy: policy,
				}
			}(), exists, err
		}
	case string(constants.Azure):
		if iamv1.Spec.Azure != nil && iamv1.Spec.Azure.UseExternalPrincipal {
			for _, principalID := range getAzureAdditionalPrincipalIDs(iamv1) {
				_, exists, err := iam.azureCli.GetRoleAssignmentByPrincipalID(ctx, iamv1.Spec.Azure.RoleScope, principalID)
				if err != nil {
					return nil, false, err
				}
				if exists {
					return &roleWarp{}, true, nil
				}
			}
			return nil, false, nil
		}

		id, exists, err := iam.azureCli.GetWorkloadIdentity(ctx, name)
		if !exists || err != nil {
			return nil, exists, err
		}
		return &roleWarp{
			AzureWorkloadIdentity: id,
		}, exists, err
	case string(constants.AliCloud):
		{
			out, exists, err := iam.aliCli.GetRole(ctx, name)
			return func() *roleWarp {
				if out == nil || out.Body == nil || out.Body.Role == nil {
					return nil
				}
				return &roleWarp{
					AliCloudAssumeRolePolicyDocument: out.Body.Role.AssumeRolePolicyDocument,
				}
			}(), exists, err
		}
	}
	return nil, false, errors.New(errNotSupportProvidedVendor)
}

func (iam *IAM) Observe(ctx context.Context, obj client.Object) (_ reconciler.ExternalObservation, errResult error) {
	log := log.FromContext(ctx).WithName("Observe")
	iamv1, ok := obj.(*observabilitypingcapcomv1.IAM)
	if !ok {
		iam.reconcilerRepoter(obj.GetName(), "", errNotIAM)
		return reconciler.ExternalObservation{Droped: true}, errors.New(errNotIAM)
	}
	log.V(1).Info("IAM crd info", "crd", fmt.Sprintf("%v", iamv1))
	if iamv1.Status.Phase == constants.CREATION_FAILED {
		// final state, nothing to do
		iam.reconcilerRepoter(iamv1.Name, "", "")
		return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil
	}
	roleName, ok := iamv1.Labels[constants.LabelAlias]
	if !ok {
		iam.reconcilerRepoter(iamv1.Name, "", errNotFoundRoleName)
		return reconciler.ExternalObservation{}, errors.New(errNotFoundRoleName)
	}
	role, exists, err := iam.getRole(ctx, iamv1, roleName)
	if err != nil {
		iam.reconcilerRepoter(iamv1.Name, "", err.Error())
		return reconciler.ExternalObservation{}, err
	}
	if !exists {
		//if not exists then we should create
		return reconciler.ExternalObservation{ResourceExists: false}, nil
	}

	if iamv1.Status.Phase == constants.ACTIVE {
		equaled, err := iam.equalPolicy(ctx, log, roleName, *role, iamv1)
		if err != nil {
			iam.reconcilerRepoter(iamv1.Name, "", err.Error())
			return reconciler.ExternalObservation{}, err
		}
		if equaled {
			iam.reconcilerRepoter(iamv1.Name, "", "")
			return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil
		}
	}
	// exists but Phase not get final state so we should update state
	return reconciler.ExternalObservation{ResourceExists: true}, nil
}
func (iam *IAM) CreateAWS(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	docbyetes, err := base64.StdEncoding.DecodeString(iamv1.Spec.Aws.AssumeRolePolicyDocument)
	if err != nil {
		return err
	}

	if err := iam.iamcli.CreateRole(ctx, roleName, string(docbyetes)); err != nil {
		return err
	}
	//create inline policies
	for _, policy := range iamv1.Spec.Aws.Policies {
		docbyetes, err := base64.StdEncoding.DecodeString(policy.PolicyDocument)
		if err != nil {
			return err
		}
		if err := iam.iamcli.PutRolePolicy(ctx, roleName, policy.PolicyName, string(docbyetes)); err != nil {
			return err
		}
	}
	return nil
}
func (iam *IAM) CreateAliCloud(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	docbyetes, err := base64.StdEncoding.DecodeString(iamv1.Spec.AliCloud.AssumeRolePolicyDocument)
	if err != nil {
		return err
	}

	if err := iam.aliCli.CreateRole(ctx, roleName, string(docbyetes)); err != nil {
		return err
	}
	//create inline policies
	for _, policy := range iamv1.Spec.AliCloud.Policies {
		docbyetes, err := base64.StdEncoding.DecodeString(policy.PolicyDocument)
		if err != nil {
			return err
		}
		if err := iam.aliCli.CreateRolePolicy(ctx, roleName, policy.PolicyName, string(docbyetes)); err != nil {
			return err
		}
	}
	return nil
}

func (iam *IAM) CreateGCP(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	var cli providers.IAMClientGCP
	if iamv1.Spec.Gcp.ProjectID != "" {
		cli = iam.gcpcli.AssumeProject(ctx, iamv1.Spec.Gcp.ProjectID)
	} else {
		cli = iam.gcpcli
	}
	if useGCPExternalPrincipal(iamv1) {
		return iam.updatePolicyGCPExternal(ctx, cli, roleName, iamv1)
	}
	// create serviceaccount
	accout, err := cli.CreateServiceAccount(ctx, roleName)
	if err != nil {
		return err
	}
	// set serviceaccount policy
	policy, err := cli.GetServiceAccountIAM(context.Background(), roleName)
	if err != nil {
		return err
	}
	//ensure binding is empty,beaceuse of new create
	_, binding := gcp.FindBindingPolicyorCreate(policy.Bindings)
	binding.Members = append(binding.Members, iamv1.Spec.Gcp.AssumeRoleAccounts...)
	policy.Bindings = append(policy.Bindings, &binding)
	err = cli.SetServiceAccountIAM(ctx, roleName, policy)
	if err != nil {
		return err
	}
	// create resourcepolicy
	for k, v := range iamv1.Spec.Gcp.Policies {
		_, resource, err := observabilitypingcapcomv1.GetFromGCPPolicyKey(k)
		if err != nil {
			return err
		}
		policy, _, err := cli.GetBucketPolicy(ctx, resource)
		if err != nil {
			return err
		}
		for _, iv := range v {
			var bindedMember []string
			//auth to service account
			bindedMember = append(bindedMember, serviceAccoutPrefix+accout.Email)
			//auth to user owner service account

			bindedMember = append(bindedMember, filterGCPWorkWorkloadIdentity(iamv1.Spec.Gcp.AssumeRoleAccounts)...)
			policy.Bindings = append(policy.Bindings, &iamgcppb.Binding{
				Role:    iv.PolicyRole,
				Members: bindedMember,
				Condition: &expr.Expr{
					Title:      buildGCPConditionTitle(accountConditionPrefix, roleName, iv.PolicyName),
					Expression: iv.PolicyCondition,
				},
			})
		}
		err = cli.SetBucketPolicy(ctx, resource, policy)
		if err != nil {
			return err
		}
	}
	return nil
}

func (iam *IAM) CreateAzure(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	if iamv1.Spec.Azure.UseExternalPrincipal {
		principalIDs := getAzureAdditionalPrincipalIDs(iamv1)
		if len(principalIDs) == 0 {
			return fmt.Errorf("additionalRoleMembers must not be empty when useExternalPrincipal is enabled")
		}
		for _, id := range principalIDs {
			if _, err := iam.azureCli.CreateOrUpdateRoleAssignmentForPrincipal(ctx, iamv1.Spec.Azure.RoleScope,
				buildAzureRoleAssignmentProperties(iam.serviceInfo.AccountInfo.SubscriptionID, id, iamv1.Spec.Azure.RoleCondition,
					to.Ptr(armauthorization.PrincipalTypeServicePrincipal))); err != nil {
				return err
			}
		}
		if !equalMembers(getAzureLastAdditionalPrincipalIDs(iamv1), principalIDs) {
			iamv1.Spec.Azure.LastAdditionalRoleMembers = principalIDs
			if err := iam.cli.Update(ctx, iamv1); err != nil && !apierrors.IsConflict(err) {
				return err
			}
		}
		return nil
	}

	// 1. create user-managed identity
	identity, err := iam.azureCli.CreateUserIdentity(ctx, roleName)
	if err != nil {
		return err
	}

	// 2. create federated identity attached to the managed identity
	err = iam.azureCli.CreateOrUpdateFederatedIdentity(ctx, roleName, iamv1.Spec.Azure.WorkloadOIDCIssuer, iamv1.Spec.Azure.WorkloadServiceAccounts)
	if err != nil {
		return err
	}

	if identity.Properties.PrincipalID == nil {
		return fmt.Errorf("identity principal id is nil")
	}

	// 3. create role assignment
	_, err = iam.azureCli.CreateOrUpdateRoleAssignmentForPrincipal(ctx, iamv1.Spec.Azure.RoleScope,
		buildAzureRoleAssignmentProperties(iam.serviceInfo.AccountInfo.SubscriptionID, *identity.Properties.PrincipalID, iamv1.Spec.Azure.RoleCondition, nil))
	if err != nil {
		return err
	}

	// 4. create role assignment for additional principals.
	for _, id := range getAzureAdditionalPrincipalIDs(iamv1) {
		if _, err = iam.azureCli.CreateOrUpdateRoleAssignmentForPrincipal(ctx, iamv1.Spec.Azure.RoleScope,
			buildAzureRoleAssignmentProperties(iam.serviceInfo.AccountInfo.SubscriptionID, id, iamv1.Spec.Azure.RoleCondition, nil)); err != nil {
			return err
		}
	}
	return nil
}

func (iam *IAM) updateIAMStatusID(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	if iamv1.Spec.Azure == nil || iam.serviceInfo.Vendor != string(constants.Azure) {
		return nil
	}
	if iamv1.Spec.Azure.UseExternalPrincipal {
		return nil
	}
	id, exist, err := iam.azureCli.GetWorkloadIdentity(ctx, roleName)
	if !exist || err != nil {
		return fmt.Errorf("failed to get workload identity when update status: %v", err)
	}
	// need to update client-id annotation of the sa under the same namespace first
	// it's used by prometheus in exclusive mode
	sa := &corev1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: iamv1.Namespace,
			Name:      iamv1.Namespace,
		},
	}
	err = iam.cli.Get(ctx, client.ObjectKeyFromObject(sa), sa)
	// if not found, do nothing
	if client.IgnoreNotFound(err) != nil {
		return err
	}

	identity := id.Identity
	if err == nil {
		if sa.Annotations != nil {
			sa.Annotations[constants.AnnotationServiceAccountRoleAzure] = *identity.Properties.ClientID
		} else {
			sa.Annotations = map[string]string{
				constants.AnnotationServiceAccountRoleAzure: *identity.Properties.ClientID,
			}
		}
		if err := iam.cli.Update(ctx, sa); err != nil {
			return err
		}
	}

	iamv1.Status.ClientId = *identity.Properties.ClientID
	iamv1.Status.PrincipalID = *identity.Properties.PrincipalID
	return nil
}

func (iam *IAM) Create(ctx context.Context, obj client.Object) (errResult error) {
	iamv1, ok := obj.(*observabilitypingcapcomv1.IAM)
	if !ok {
		iam.reconcilerRepoter(obj.GetName(), "", errNotIAM)
		return errors.New(errNotIAM)
	}
	//update state and then create iam
	iamv1.Status.Phase = constants.CREATING
	if err := iam.cli.Status().Update(ctx, iamv1); err != nil {
		iam.reconcilerRepoter(iamv1.Name, "", err.Error())
		return err
	}
	roleName, ok := iamv1.Labels[constants.LabelAlias]
	if !ok {
		iam.reconcilerRepoter(iamv1.Name, "", errNotFoundRoleName)
		return errors.New(errNotFoundRoleName)
	}
	var err error
	switch iam.serviceInfo.Vendor {
	case string(constants.AWS):
		err = iam.CreateAWS(ctx, roleName, iamv1)
	case string(constants.GCP):
		err = iam.CreateGCP(ctx, roleName, iamv1)
	case string(constants.Azure):
		err = iam.CreateAzure(ctx, roleName, iamv1)
	case string(constants.AliCloud):
		err = iam.CreateAliCloud(ctx, roleName, iamv1)
	}
	if err != nil {
		iam.reconcilerRepoter(iamv1.Name, "", err.Error())
		return err
	}
	iam.reconcilerRepoter(iamv1.Name, "", "")
	return nil
}

func (iam *IAM) Update(ctx context.Context, obj client.Object) (errResult error) {
	iamv1, ok := obj.(*observabilitypingcapcomv1.IAM)
	if !ok {
		iam.reconcilerRepoter(obj.GetName(), "", errNotIAM)
		return errors.New(errNotIAM)
	}
	roleName, ok := iamv1.Labels[constants.LabelAlias]
	if !ok {
		iam.reconcilerRepoter(iamv1.Name, "", errNotFoundRoleName)
		return errors.New(errNotFoundRoleName)
	}
	//update policies
	err := iam.updateRole(ctx, roleName, iamv1)
	if err != nil {
		iam.reconcilerRepoter(iamv1.Name, "", err.Error())
		return err
	}
	if iamv1.Status.Phase != constants.ACTIVE {
		if err := iam.updateIAMStatusID(ctx, roleName, iamv1); err != nil {
			return err
		}
		iamv1.Status.Phase = constants.ACTIVE
		if err := iam.cli.Status().Update(ctx, iamv1); err != nil {
			iam.reconcilerRepoter(iamv1.Name, "", err.Error())
			return err
		}
	}
	iam.reconcilerRepoter(iamv1.Name, "", "")
	return nil
}
func (iam *IAM) DeleteAWS(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	out, err := iam.iamcli.ListRolePolicies(ctx, roleName, nil)
	if err != nil {
		return err
	}
	for _, policyname := range out.PolicyNames {
		err := iam.iamcli.DeleteRolePolicy(ctx, roleName, policyname)
		if err != nil {
			return err
		}
	}
	err = iam.iamcli.DeleteRole(ctx, roleName)
	if err != nil {
		return err
	}
	return nil
}
func (iam *IAM) DeleteAliCloud(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	out, err := iam.aliCli.ListRolePolicies(ctx, roleName, nil)
	if err != nil {
		return err
	}
	if out != nil && out.Body != nil && out.Body.Policies != nil && out.Body.Policies.Policy != nil {
		for _, policy := range out.Body.Policies.Policy {
			err := iam.aliCli.DeleteRolePolicy(ctx, roleName, *policy.PolicyName)
			if err != nil {
				return err
			}
		}
	}

	err = iam.aliCli.DeleteRole(ctx, roleName)
	if err != nil {
		return err
	}
	return nil
}

func (iam *IAM) DeleteGCP(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	var cli providers.IAMClientGCP
	if iamv1.Spec.Gcp.ProjectID != "" {
		cli = iam.gcpcli.AssumeProject(ctx, iamv1.Spec.Gcp.ProjectID)
	} else {
		cli = iam.gcpcli
	}
	if !useGCPExternalPrincipal(iamv1) {
		//delete service account
		err := cli.DeleteServiceAccount(ctx, roleName)
		if err != nil {
			return err
		}
	}
	//delete policy on resource
	for k, v := range iamv1.Spec.Gcp.Policies {
		_, resource, err := observabilitypingcapcomv1.GetFromGCPPolicyKey(k)
		if err != nil {
			return err
		}
		policy, exists, err := cli.GetBucketPolicy(ctx, resource)
		if !exists {
			//ingore not exists policy
			continue
		}
		if err != nil {
			return err
		}
		//delete
		var saved []*iamgcppb.Binding
		for _, binding := range policy.Bindings {
			if isGCPPolicyBinding(binding, roleName) {
				//if gcp policy  not be included by crd define
				if !isGCPIncludePolicy(roleName, binding, v) {
					//save them
					saved = append(saved, binding)
				}
			} else {
				saved = append(saved, binding)
			}
		}
		policy.Bindings = saved
		err = cli.SetBucketPolicy(ctx, resource, policy)
		if err != nil {
			return err
		}
	}
	return nil
}

func (iam *IAM) DeleteAzure(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	if iamv1.Spec.Azure != nil && iamv1.Spec.Azure.UseExternalPrincipal {
		for _, id := range unionStrings(getAzureAdditionalPrincipalIDs(iamv1), getAzureLastAdditionalPrincipalIDs(iamv1)) {
			if err := iam.azureCli.DeleteRoleAssignmentByPrincipalID(ctx, iamv1.Spec.Azure.RoleScope, id); err != nil {
				return err
			}
		}
		return nil
	}

	identity, exists, err := iam.azureCli.GetUserIdentity(ctx, roleName)
	if err != nil {
		return err
	}
	if !exists {
		return nil
	}

	if err := iam.azureCli.DeleteRoleAssignmentByPrincipalID(ctx, iamv1.Spec.Azure.RoleScope, *identity.Properties.PrincipalID); err != nil {
		return err
	}

	for _, id := range getAzureAdditionalPrincipalIDs(iamv1) {
		if err := iam.azureCli.DeleteRoleAssignmentByPrincipalID(ctx, iamv1.Spec.Azure.RoleScope, id); err != nil {
			return err
		}
	}

	// federated identity will be deleted by azure when user identity is deleted.
	return iam.azureCli.DeleteUserIdentity(ctx, roleName)
}

func (iam *IAM) Delete(ctx context.Context, obj client.Object) (_ reconciler.ExternalDeletion, errResult error) {
	//iam delete path
	iamv1, ok := obj.(*observabilitypingcapcomv1.IAM)
	if !ok {
		iam.reconcilerRepoter(obj.GetName(), "", errNotIAM)
		return reconciler.ExternalDeletion{}, errors.New(errNotIAM)
	}
	iamv1.Status.Phase = constants.DELETING
	if err := iam.cli.Status().Update(ctx, iamv1); err != nil {
		return reconciler.ExternalDeletion{}, err
	}
	roleName, ok := iamv1.Labels[constants.LabelAlias]
	if !ok {
		iam.reconcilerRepoter(iamv1.Name, "", errNotFoundRoleName)
		return reconciler.ExternalDeletion{}, errors.New(errNotFoundRoleName)
	}
	var err error
	switch iam.serviceInfo.Vendor {
	case string(constants.AWS):
		err = iam.DeleteAWS(ctx, roleName, iamv1)
	case string(constants.GCP):
		err = iam.DeleteGCP(ctx, roleName, iamv1)
	case string(constants.Azure):
		err = iam.DeleteAzure(ctx, roleName, iamv1)
	case string(constants.AliCloud):
		err = iam.DeleteAliCloud(ctx, roleName, iamv1)

	}
	if err != nil {
		iam.reconcilerRepoter(iamv1.Name, "", err.Error())
		return reconciler.ExternalDeletion{}, err
	}
	iam.reconcilerRepoter(iamv1.Name, "", "")
	return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
}

func (iam *IAM) updateAWSRole(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	//policy
	err := iam.updatePolicyAWS(ctx, roleName, iamv1.Spec.Aws.Policies)
	if err != nil {
		return err
	}

	//AssumeRolePolicyDocument
	assumeDocument, err := iam.updateAssumeEntityAWS(ctx, roleName, iamv1)
	if err != nil {
		return err
	}

	// A nil document means the trust update was intentionally skipped (the merge
	// produced an empty policy). Leave the CRD's desired/last trust documents
	// untouched so a later reconcile can converge once AWS state changes.
	if assumeDocument == nil {
		return nil
	}

	//crd
	base64AssumeDocument := base64.StdEncoding.EncodeToString(assumeDocument)
	iamv1.Spec.Aws.LastAssumeRolePolicyDocument = base64AssumeDocument
	iamv1.Spec.Aws.AssumeRolePolicyDocument = base64AssumeDocument

	err = iam.cli.Update(ctx, iamv1)
	if err == nil {
		return nil
	}

	if apierrors.IsConflict(err) {
		return nil
	}

	return err
}

func (iam *IAM) updateAliCloudRole(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	//policy
	err := iam.updatePolicyAliCloud(ctx, roleName, iamv1.Spec.AliCloud.Policies)
	if err != nil {
		return err
	}

	//AssumeRolePolicyDocument
	assumeDocument, err := iam.updateAssumeEntityAliCloud(ctx, roleName, iamv1)
	if err != nil {
		return err
	}

	//crd
	base64AssumeDocument := base64.StdEncoding.EncodeToString(assumeDocument)
	iamv1.Spec.AliCloud.LastAssumeRolePolicyDocument = base64AssumeDocument
	iamv1.Spec.AliCloud.AssumeRolePolicyDocument = base64AssumeDocument

	err = iam.cli.Update(ctx, iamv1)
	if err == nil {
		return nil
	}

	if apierrors.IsConflict(err) {
		return nil
	}

	return err
}

func (iam *IAM) updatePolicyAWS(ctx context.Context, roleName string, policies []observabilitypingcapcomv1.AWSRolePolicy) error {

	specset := mapset.NewSet()
	for _, policy := range policies {
		specset.Add(policy.PolicyName)
	}

	out, err := iam.iamcli.ListRolePolicies(ctx, roleName, nil)
	if err != nil {
		return err
	}
	observset := mapset.NewSet()
	for _, policy := range out.PolicyNames {
		observset.Add(policy)
	}

	// Add missing policies and replace existing ones.
	for policy := range specset.Difference(observset).Iter() {
		newpolicy := peekPolicy(policies, policy.(string))
		docbyetes, err := base64.StdEncoding.DecodeString(newpolicy.PolicyDocument)
		if err != nil {
			return err
		}
		err = iam.iamcli.PutRolePolicy(ctx, roleName, newpolicy.PolicyName, string(docbyetes))
		if err != nil {
			return err
		}
	}
	//delete
	for policy := range observset.Difference(specset).Iter() {
		err := iam.iamcli.DeleteRolePolicy(ctx, roleName, policy.(string))
		if err != nil {
			return err
		}
	}

	return nil
}

func (iam *IAM) updatePolicyAliCloud(ctx context.Context, roleName string, policies []observabilitypingcapcomv1.AliCloudPolicy) error {

	specset := mapset.NewSet()
	for _, policy := range policies {
		specset.Add(policy.PolicyName)
	}

	out, err := iam.aliCli.ListRolePolicies(ctx, roleName, nil)
	if err != nil {
		return err
	}
	observset := mapset.NewSet()
	if out != nil && out.Body != nil && out.Body.Policies != nil && out.Body.Policies.Policy != nil {
		for _, policy := range out.Body.Policies.Policy {
			observset.Add(*policy.PolicyName)
		}
	}
	//add
	for policy := range specset.Difference(observset).Iter() {
		newPolicy := peekPolicyAli(policies, policy.(string))
		docBytes, err := base64.StdEncoding.DecodeString(newPolicy.PolicyDocument)
		if err != nil {
			return err
		}
		err = iam.aliCli.CreateRolePolicy(ctx, roleName, newPolicy.PolicyName, string(docBytes))
		if err != nil {
			return err
		}
	}
	//update
	for policy := range specset.Intersect(observset).Iter() {
		newPolicy := peekPolicyAli(policies, policy.(string))
		docBytes, err := base64.StdEncoding.DecodeString(newPolicy.PolicyDocument)
		if err != nil {
			return err
		}
		err = iam.aliCli.UpdateRolePolicy(ctx, roleName, newPolicy.PolicyName, string(docBytes))
		if err != nil {
			return err
		}
	}
	//delete
	for policy := range observset.Difference(specset).Iter() {
		err := iam.aliCli.DeleteRolePolicy(ctx, roleName, policy.(string))
		if err != nil {
			return err
		}
	}

	return nil
}

func (iam *IAM) getServiceAccountEmail(name string) string {
	return fmt.Sprintf("%s@%s.iam.gserviceaccount.com", name, iam.serviceInfo.AccountInfo.Project)
}
func equalMembers(older []string, newer []string) bool {
	olderset := mapset.NewSet()
	for _, v := range older {
		olderset.Add(v)
	}
	newerset := mapset.NewSet()
	for _, v := range newer {
		newerset.Add(v)
	}
	return !(olderset.Difference(newerset).Cardinality() > 0 || newerset.Difference(olderset).Cardinality() > 0)
}

func (iam *IAM) updateGCPRole(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	var cli providers.IAMClientGCP
	if iamv1.Spec.Gcp.ProjectID != "" {
		cli = iam.gcpcli.AssumeProject(ctx, iamv1.Spec.Gcp.ProjectID)
	} else {
		cli = iam.gcpcli
	}
	if useGCPExternalPrincipal(iamv1) {
		iamv1.Spec.Gcp.LastAssumeRoleAccounts = append([]string{}, iamv1.Spec.Gcp.AssumeRoleAccounts...)
		err := iam.updatePolicyGCPExternal(ctx, cli, roleName, iamv1)
		if err != nil {
			return err
		}
		err = iam.cli.Update(ctx, iamv1)
		if err == nil || apierrors.IsConflict(err) {
			return nil
		}
		return err
	}
	accounts, err := iam.calcAssumeEntityGCP(ctx, cli, roleName, iamv1)
	if err != nil {
		return err
	}

	iamv1.Spec.Gcp.LastAssumeRoleAccounts = accounts
	iamv1.Spec.Gcp.AssumeRoleAccounts = accounts
	err = iam.updatePolicyGCP(ctx, cli, roleName, iamv1)
	if err != nil {
		return err
	}

	err = iam.cli.Update(ctx, iamv1)
	if err == nil {
		return nil
	}

	if apierrors.IsConflict(err) {
		return nil
	}

	return err
}

func gcpExternalBindingMembers(accounts []string) []string {
	return filterGCPWorkWorkloadIdentity(accounts)
}

func (iam *IAM) updatePolicyGCPExternal(ctx context.Context, cli providers.IAMClientGCP, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	members := gcpExternalBindingMembers(iamv1.Spec.Gcp.AssumeRoleAccounts)
	for k, v := range iamv1.Spec.Gcp.Policies {
		_, resourcename, err := observabilitypingcapcomv1.GetFromGCPPolicyKey(k)
		if err != nil {
			return err
		}
		policy, exists, err := cli.GetBucketPolicy(ctx, resourcename)
		if err != nil {
			return err
		}
		if !exists {
			continue
		}

		changed := false
		expected := make(map[string]observabilitypingcapcomv1.GCPRolePolicy, len(v))
		for _, item := range v {
			title := buildGCPConditionTitle(accountConditionPrefix, roleName, item.PolicyName)
			expected[title] = item
		}

		saved := make([]*iamgcppb.Binding, 0, len(policy.Bindings))
		for _, b := range policy.Bindings {
			if !isGCPPolicyBinding(b, roleName) {
				saved = append(saved, b)
				continue
			}
			if b.Condition == nil {
				changed = true
				continue
			}
			item, ok := expected[b.Condition.Title]
			if !ok {
				changed = true
				continue
			}
			if b.Role != item.PolicyRole || !equalMembers(b.Members, members) || b.Condition.Expression != item.PolicyCondition {
				changed = true
				b.Role = item.PolicyRole
				b.Members = members
				b.Condition.Expression = item.PolicyCondition
			}
			saved = append(saved, b)
			delete(expected, b.Condition.Title)
		}

		for title, item := range expected {
			changed = true
			saved = append(saved, &iamgcppb.Binding{
				Role:    item.PolicyRole,
				Members: members,
				Condition: &expr.Expr{
					Title:      title,
					Expression: item.PolicyCondition,
				},
			})
		}

		if changed {
			policy.Bindings = saved
			if err := cli.SetBucketPolicy(ctx, resourcename, policy); err != nil {
				return err
			}
		}
	}
	return nil
}

func (iam *IAM) updatePolicyGCP(ctx context.Context, cli providers.IAMClientGCP, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	//cmp and update AssumeRolePolicy
	policy, err := cli.GetServiceAccountIAM(ctx, roleName)
	if err != nil {
		return err
	}
	//flash access member to resource policy flag
	needflash := false
	index, binding := gcp.FindBindingPolicyorCreate(policy.Bindings)
	//if binding not exists we should update policy
	if index < 0 {
		binding.Members = append(binding.Members, iamv1.Spec.Gcp.AssumeRoleAccounts...)
		policy.Bindings = append(policy.Bindings, &binding)
		err = cli.SetServiceAccountIAM(ctx, roleName, policy)
		if err != nil {
			return err
		}
	} else {
		//we use CRD defined member override iam binding in GCP
		if !equalMembers(binding.Members, iamv1.Spec.Gcp.AssumeRoleAccounts) {
			needflash = true
			binding.Members = iamv1.Spec.Gcp.AssumeRoleAccounts
			policy.Bindings[index] = &binding
			err = cli.SetServiceAccountIAM(ctx, roleName, policy)
			if err != nil {
				return err
			}
		}
	}
	//cmp and update resource policy
	for k, v := range iamv1.Spec.Gcp.Policies {
		//now we have only one resourcetype
		//get iam from cloud storage
		_, resourcename, err := observabilitypingcapcomv1.GetFromGCPPolicyKey(k)
		if err != nil {
			return err
		}
		// flag for which policy be modified
		var changed bool = false
		policy, exists, err := cli.GetBucketPolicy(ctx, resourcename)
		if !exists {
			//ingore not exists policy
			continue
		}
		if err != nil {
			return err
		}
		//delete
		var saved []*iamgcppb.Binding
		for _, binding := range policy.Bindings {
			if isGCPPolicyBinding(binding, roleName) {
				//if gcp policy be included by crd define
				if isGCPIncludePolicy(roleName, binding, v) {
					//save them
					saved = append(saved, binding)
				} else {
					//ingnore the other poloicy, then will be deleted
					//set flag is any changed
					changed = true
				}

			} else {
				saved = append(saved, binding)
			}
		}
		policy.Bindings = saved
		//add
		for _, iv := range v {
			title := buildGCPConditionTitle(accountConditionPrefix, roleName, iv.PolicyName)
			index, one := peekGCPPolicy(policy.Bindings, title)
			if index < 0 {
				//if not found we add this policy
				//auth to service account
				var bindedMember []string
				bindedMember = append(bindedMember, serviceAccoutPrefix+iam.getServiceAccountEmail(roleName))
				//auth to user owner service account
				bindedMember = append(bindedMember, filterGCPWorkWorkloadIdentity(iamv1.Spec.Gcp.AssumeRoleAccounts)...)
				policy.Bindings = append(policy.Bindings, &iamgcppb.Binding{
					Role:    iv.PolicyRole,
					Members: bindedMember,
					Condition: &expr.Expr{
						Title:      title,
						Expression: iv.PolicyCondition,
					},
				})
				changed = true
			} else {
				//if access member changed we should flash policy to resource members
				if needflash {
					changed = true
					var bindedMember []string
					bindedMember = append(bindedMember, serviceAccoutPrefix+iam.getServiceAccountEmail(roleName))
					//auth to user owner service account
					bindedMember = append(bindedMember, filterGCPWorkWorkloadIdentity(iamv1.Spec.Gcp.AssumeRoleAccounts)...)
					one.Members = bindedMember
					policy.Bindings[index] = one
				}
			}
		}
		if changed {
			//update policy
			err = cli.SetBucketPolicy(ctx, resourcename, policy)
			if err != nil {
				return err
			}
		}
	}
	return nil
}

func (iam *IAM) updateAzureRole(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	if iamv1.Spec.Azure != nil && iamv1.Spec.Azure.UseExternalPrincipal {
		desiredPrincipalIDs := getAzureAdditionalPrincipalIDs(iamv1)
		if len(desiredPrincipalIDs) == 0 {
			return errors.New("updating azure role with external principal but no principal ids")
		}
		lastAppliedPrincipalIDs := getAzureLastAdditionalPrincipalIDs(iamv1)

		desiredSet := mapset.NewSet()
		for _, id := range desiredPrincipalIDs {
			desiredSet.Add(id)
		}
		for _, id := range lastAppliedPrincipalIDs {
			if !desiredSet.Contains(id) {
				if err := iam.azureCli.DeleteRoleAssignmentByPrincipalID(ctx, iamv1.Spec.Azure.RoleScope, id); err != nil {
					return err
				}
			}
		}

		for _, id := range desiredPrincipalIDs {
			if _, err := iam.azureCli.CreateOrUpdateRoleAssignmentForPrincipal(ctx, iamv1.Spec.Azure.RoleScope,
				buildAzureRoleAssignmentProperties(iam.serviceInfo.AccountInfo.SubscriptionID, id, iamv1.Spec.Azure.RoleCondition,
					to.Ptr(armauthorization.PrincipalTypeServicePrincipal))); err != nil {
				return err
			}
		}

		if !equalMembers(lastAppliedPrincipalIDs, desiredPrincipalIDs) {
			iamv1.Spec.Azure.LastAdditionalRoleMembers = desiredPrincipalIDs
			if err := iam.cli.Update(ctx, iamv1); err != nil && !apierrors.IsConflict(err) {
				return err
			}
		}
		return nil
	}

	identity, exist, err := iam.azureCli.GetUserIdentity(ctx, roleName)
	if err != nil {
		return err
	}
	if !exist {
		return errors.New("updating a non-existent user identity")
	}

	// 1. make sure all federatedIdentity is created
	if err := iam.azureCli.CreateOrUpdateFederatedIdentity(ctx, roleName, iamv1.Spec.Azure.WorkloadOIDCIssuer, iamv1.Spec.Azure.WorkloadServiceAccounts); err != nil {
		return err
	}
	// 2. make sure role assignment and role Condition is set
	_, err = iam.azureCli.CreateOrUpdateRoleAssignmentForPrincipal(ctx, iamv1.Spec.Azure.RoleScope,
		buildAzureRoleAssignmentProperties(iam.serviceInfo.AccountInfo.SubscriptionID, *identity.Properties.PrincipalID, iamv1.Spec.Azure.RoleCondition, nil))

	for _, id := range getAzureAdditionalPrincipalIDs(iamv1) {
		if _, err = iam.azureCli.CreateOrUpdateRoleAssignmentForPrincipal(ctx, iamv1.Spec.Azure.RoleScope,
			buildAzureRoleAssignmentProperties(iam.serviceInfo.AccountInfo.SubscriptionID, id, iamv1.Spec.Azure.RoleCondition, nil)); err != nil {
			return err
		}
	}
	return err
}

func (iam *IAM) updateRole(ctx context.Context, roleName string, iamv1 *observabilitypingcapcomv1.IAM) error {
	switch iam.serviceInfo.Vendor {
	case string(constants.AWS):
		return iam.updateAWSRole(ctx, roleName, iamv1)
	case string(constants.GCP):
		return iam.updateGCPRole(ctx, roleName, iamv1)
	case string(constants.Azure):
		return iam.updateAzureRole(ctx, roleName, iamv1)
	case string(constants.AliCloud):
		return iam.updateAliCloudRole(ctx, roleName, iamv1)
	}
	return nil
}

func (iam *IAM) equalPolicyAWS(ctx context.Context, log logr.Logger, roleName string, AWSassumeRolePolicyDocument string, assumeDocument string, awspoliciesNames []string, crdpolicies []observabilitypingcapcomv1.AWSRolePolicy) (bool, error) {
	log = log.WithName("AWS")
	//AssumeRolePolicyDocument
	tmp, err := url.QueryUnescape(AWSassumeRolePolicyDocument)
	if err != nil {
		return false, err
	}
	doc := base64.StdEncoding.EncodeToString([]byte(tmp))
	awsassumedoc := AssumeRolePolicy{}
	err = json.Unmarshal([]byte(tmp), &awsassumedoc)
	if err != nil {
		return false, err
	}
	//CRD assume doc
	crdassumedoc, err := getAWSAssumeEntityFromCRD(assumeDocument)
	if err != nil {
		return false, err
	}
	//do policy
	specset := mapset.NewSet()
	for _, policy := range crdpolicies {
		specset.Add(policy.PolicyName)
	}
	observset := mapset.NewSet()
	for _, policy := range awspoliciesNames {
		observset.Add(policy)
	}

	if specset.Difference(observset).Cardinality() > 0 || observset.Difference(specset).Cardinality() > 0 || !cmp.Equal(awsassumedoc, crdassumedoc) {
		log.V(1).Info("crd policy diff IAM", "cap", specset.Difference(observset).Cardinality(), "eq", specset.Difference(observset).Cardinality() > 0)
		log.V(1).Info("IAM policy diff crd", "cap", observset.Difference(specset).Cardinality(), "eq", observset.Difference(specset).Cardinality() > 0)
		log.V(1).Info("assume policy diff", "iam", doc, "crddoc", assumeDocument, "diff", cmp.Diff(awsassumedoc, crdassumedoc))
		return false, nil
	}
	return true, nil
}

func (iam *IAM) equalPolicyAliCloud(ctx context.Context, log logr.Logger, roleName string, AliCloudAssumeRolePolicyDocument string, assumeDocument string, aliPoliciesNames []string, aliPolicyDocuments map[string]string, crdpolicies []observabilitypingcapcomv1.AliCloudPolicy) (bool, error) {
	log = log.WithName("AliCloud")
	//AssumeRolePolicyDocument
	tmp, err := url.QueryUnescape(AliCloudAssumeRolePolicyDocument)
	if err != nil {
		return false, err
	}
	doc := base64.StdEncoding.EncodeToString([]byte(tmp))
	policy := AssumeRolePolicy{}
	err = json.Unmarshal([]byte(tmp), &policy)
	if err != nil {
		return false, err
	}
	//CRD assume doc
	crdassumedoc, err := getAliCloudAssumeEntityFromCRD(assumeDocument)
	if err != nil {
		return false, err
	}
	//do policy
	specset := mapset.NewSet()
	for _, policy := range crdpolicies {
		specset.Add(policy.PolicyName)
	}
	observset := mapset.NewSet()
	for _, policy := range aliPoliciesNames {
		observset.Add(policy)
	}

	if specset.Difference(observset).Cardinality() > 0 || observset.Difference(specset).Cardinality() > 0 || !cmp.Equal(policy, crdassumedoc) {
		log.V(1).Info("crd policy diff IAM", "cap", specset.Difference(observset).Cardinality(), "eq", specset.Difference(observset).Cardinality() > 0)
		log.V(1).Info("IAM policy diff crd", "cap", observset.Difference(specset).Cardinality(), "eq", observset.Difference(specset).Cardinality() > 0)
		log.V(1).Info("assume policy diff", "iam", doc, "crddoc", assumeDocument, "diff", cmp.Diff(policy, crdassumedoc))
		return false, nil
	}
	for _, crdpolicy := range crdpolicies {
		observedDocument, ok := aliPolicyDocuments[crdpolicy.PolicyName]
		if !ok {
			log.V(1).Info("IAM policy document missing", "policy", crdpolicy.PolicyName)
			return false, nil
		}
		equaled, err := equalAliCloudPolicyDocument(observedDocument, crdpolicy.PolicyDocument)
		if err != nil {
			return false, err
		}
		if !equaled {
			log.V(1).Info("IAM policy document diff", "policy", crdpolicy.PolicyName)
			return false, nil
		}
	}
	return true, nil
}

func equalAliCloudPolicyDocument(observedDocument, crdPolicyDocument string) (bool, error) {
	decodedCRDPolicyDocument, err := base64.StdEncoding.DecodeString(crdPolicyDocument)
	if err != nil {
		return false, err
	}

	unescapedObservedDocument, err := url.QueryUnescape(observedDocument)
	if err != nil {
		unescapedObservedDocument = observedDocument
	}

	var observedPolicy interface{}
	if err := json.Unmarshal([]byte(unescapedObservedDocument), &observedPolicy); err != nil {
		return false, err
	}

	var crdPolicy interface{}
	if err := json.Unmarshal(decodedCRDPolicyDocument, &crdPolicy); err != nil {
		return false, err
	}

	return cmp.Equal(observedPolicy, crdPolicy), nil
}

func isGCPIncludePolicy(roleName string, binding *iamgcppb.Binding, policies []observabilitypingcapcomv1.GCPRolePolicy) bool {
	for _, v := range policies {
		if binding.Condition.Title == buildGCPConditionTitle(accountConditionPrefix, roleName, v.PolicyName) {
			return true
		}
	}
	return false
}

func peekGCPPolicy(binding []*iamgcppb.Binding, title string) (int, *iamgcppb.Binding) {
	for i, v := range binding {
		if v.Condition != nil && v.Condition.Title == title {
			return i, v
		}
	}
	return -1, nil
}

func getGCPPolicyLenth(binding []*iamgcppb.Binding, account string) int {
	var count int
	for _, v := range binding {
		if isGCPPolicyBinding(v, account) {
			count++
		}
	}
	return count
}

func isGCPPolicyBinding(binding *iamgcppb.Binding, account string) bool {
	return binding.Condition != nil && strings.HasPrefix(binding.Condition.Title, accountConditionPrefix+account)
}

func buildGCPConditionTitle(prefix, account, policyName string) string {
	return fmt.Sprintf("%s%s%s", prefix, account, policyName)
}

func filterGCPWorkWorkloadIdentity(source []string) []string {
	var saved []string
	for i, v := range source {
		if !strings.Contains(v, "goog[") {
			saved = append(saved, source[i])
		}
	}
	return saved
}

func (iam *IAM) equalPolicyGCP(ctx context.Context, log logr.Logger, roleName string, GCPassumeRolePolicy *iamgcpapi.Policy, iamv1 *observabilitypingcapcomv1.IAM) (bool, error) {
	log = log.WithName("GCP")
	//check AssumeRolePolicy
	index, binding := gcp.FindBindingPolicyorCreate(GCPassumeRolePolicy.Bindings)
	//if binding not exists we should update policy
	if index < 0 {
		log.V(1).Info("not found service account access policy")
		return false, nil
	}
	//if policy more than crd will touch this condition
	if len(binding.Members) != len(iamv1.Spec.Gcp.AssumeRoleAccounts) {
		log.V(1).Info("iam members not eq AssumeRoleAccounts lenth")
		return false, nil
	}
	//check crd account members not in policy
	for _, v := range iamv1.Spec.Gcp.AssumeRoleAccounts {
		if !utils.IsIncludeString(binding.Members, v) {
			log.V(1).Info("AssumeRoleAccounts not in gcp iam")
			return false, nil
		}
	}
	//check resource binding policy
	for k, v := range iamv1.Spec.Gcp.Policies {
		//now we have only one resourcetype
		//get iam from cloud storage
		_, resourcename, err := observabilitypingcapcomv1.GetFromGCPPolicyKey(k)
		if err != nil {
			return false, err
		}
		policy, exists, err := iam.gcpcli.GetBucketPolicy(ctx, resourcename)
		if !exists {
			//ingore not exists policy
			continue
		}
		if err != nil {
			return false, err
		}
		if getGCPPolicyLenth(policy.Bindings, roleName) != len(v) {
			log.V(1).Info("gcp policy length not eq crd lenth")
			return false, nil
		}
		for _, iv := range v {
			title := buildGCPConditionTitle(accountConditionPrefix, roleName, iv.PolicyName)
			index, _ := peekGCPPolicy(policy.Bindings, title)
			if index < 0 {
				log.V(1).Info("crd policy not in gcp iam")
				return false, nil
			}
		}
	}
	log.V(1).Info("iam in crd is equaled gcp iam")
	return true, nil
}

func (iam *IAM) equalPolicyGCPExternal(ctx context.Context, log logr.Logger, roleName string, iamv1 *observabilitypingcapcomv1.IAM) (bool, error) {
	log = log.WithName("GCPExternal")
	var cli providers.IAMClientGCP
	if iamv1.Spec.Gcp.ProjectID != "" {
		cli = iam.gcpcli.AssumeProject(ctx, iamv1.Spec.Gcp.ProjectID)
	} else {
		cli = iam.gcpcli
	}
	expectedMembers := gcpExternalBindingMembers(iamv1.Spec.Gcp.AssumeRoleAccounts)
	for k, v := range iamv1.Spec.Gcp.Policies {
		_, resourcename, err := observabilitypingcapcomv1.GetFromGCPPolicyKey(k)
		if err != nil {
			return false, err
		}
		policy, exists, err := cli.GetBucketPolicy(ctx, resourcename)
		if err != nil {
			return false, err
		}
		if !exists {
			continue
		}
		if getGCPPolicyLenth(policy.Bindings, roleName) != len(v) {
			return false, nil
		}
		for _, iv := range v {
			title := buildGCPConditionTitle(accountConditionPrefix, roleName, iv.PolicyName)
			index, one := peekGCPPolicy(policy.Bindings, title)
			if index < 0 || one == nil {
				return false, nil
			}
			if one.Role != iv.PolicyRole || one.Condition == nil || one.Condition.Expression != iv.PolicyCondition || !equalMembers(one.Members, expectedMembers) {
				return false, nil
			}
		}
	}
	return true, nil
}

func (iam *IAM) equalPolicyAzure(ctx context.Context, log logr.Logger, roleName string, workloadIdentity *types.AzureWorkloadIdentity, iamv1 *observabilitypingcapcomv1.IAM) (bool, error) {
	log = log.WithName("Azure")
	federatedIdentites := workloadIdentity.FederatedIdentities
	if len(federatedIdentites) < len(iamv1.Spec.Azure.WorkloadServiceAccounts) {
		return false, nil
	}

	subjectSet := map[string]struct{}{}
	for _, id := range federatedIdentites {
		if *id.Issuer != iamv1.Spec.Azure.WorkloadOIDCIssuer {
			return false, nil
		}
		subjectSet[*id.Subject] = struct{}{}
	}
	// 1. check if oidc and serviceAccounts match the federatedIdentities
	for _, v := range iamv1.Spec.Azure.WorkloadServiceAccounts {
		if _, ok := subjectSet[v]; !ok {
			log.Info("iam subject not in azure iam")
			return false, nil
		}
	}

	// 2. check if role assignment been created and the roleCondition is correct.
	ra, exists, err := iam.azureCli.GetRoleAssignmentByPrincipalID(ctx, iamv1.Spec.Azure.RoleScope, *workloadIdentity.Identity.Properties.PrincipalID)
	if !exists || err != nil {
		return false, err
	}
	conditionEqual := (ra.Properties.Condition == nil && strings.TrimSpace(iamv1.Spec.Azure.RoleCondition) == "") ||
		(ra.Properties.Condition != nil && *ra.Properties.Condition == iamv1.Spec.Azure.RoleCondition)
	for _, id := range getAzureAdditionalPrincipalIDs(iamv1) {
		r, exists, err := iam.azureCli.GetRoleAssignmentByPrincipalID(ctx, iamv1.Spec.Azure.RoleScope, id)
		if !exists || err != nil {
			return false, err
		}
		conditionEqual = conditionEqual && ((r.Properties.Condition == nil && strings.TrimSpace(iamv1.Spec.Azure.RoleCondition) == "") ||
			(r.Properties.Condition != nil && *r.Properties.Condition == iamv1.Spec.Azure.RoleCondition))
	}

	return conditionEqual, nil
}

func (iam *IAM) equalPolicyAzureExternal(ctx context.Context, iamv1 *observabilitypingcapcomv1.IAM) (bool, error) {
	principalIDs := getAzureAdditionalPrincipalIDs(iamv1)
	if len(principalIDs) == 0 {
		return false, nil
	}
	// Force update path when recorded last-applied set drifts from desired set.
	if !equalMembers(getAzureLastAdditionalPrincipalIDs(iamv1), principalIDs) {
		return false, nil
	}
	for _, id := range principalIDs {
		ra, exists, err := iam.azureCli.GetRoleAssignmentByPrincipalID(ctx, iamv1.Spec.Azure.RoleScope, id)
		if !exists || err != nil {
			return false, err
		}
		if (ra.Properties.Condition == nil && strings.TrimSpace(iamv1.Spec.Azure.RoleCondition) != "") ||
			(ra.Properties.Condition != nil && *ra.Properties.Condition != iamv1.Spec.Azure.RoleCondition) {
			return false, nil
		}
	}
	return true, nil
}

func (iam *IAM) equalPolicy(ctx context.Context, log logr.Logger, roleName string, role roleWarp, iamv1 *observabilitypingcapcomv1.IAM) (bool, error) {
	log = log.WithName("equalPolicy")
	if iamv1.Spec.Azure != nil && iamv1.Spec.Azure.UseExternalPrincipal {
		return iam.equalPolicyAzureExternal(ctx, iamv1)
	}
	if useGCPExternalPrincipal(iamv1) {
		return iam.equalPolicyGCPExternal(ctx, log, roleName, iamv1)
	}
	if role.AWSAssumeRolePolicyDocument != nil {
		//AWS
		out, err := iam.iamcli.ListRolePolicies(ctx, roleName, nil)
		if err != nil {
			return false, err
		}
		return iam.equalPolicyAWS(ctx, log, roleName, *role.AWSAssumeRolePolicyDocument, iamv1.Spec.Aws.AssumeRolePolicyDocument, out.PolicyNames, iamv1.Spec.Aws.Policies)
	} else if role.GCPAssumeRolePolicy != nil {
		//GCP
		return iam.equalPolicyGCP(ctx, log, roleName, role.GCPAssumeRolePolicy, iamv1)
	} else if role.AzureWorkloadIdentity != nil && role.AzureWorkloadIdentity.FederatedIdentities != nil {
		return iam.equalPolicyAzure(ctx, log, roleName, role.AzureWorkloadIdentity, iamv1)
	} else if role.AliCloudAssumeRolePolicyDocument != nil {
		out, err := iam.aliCli.ListRolePolicies(ctx, roleName, nil)
		if err != nil {
			return false, err
		}

		policyNames := []string{}
		policyDocuments := map[string]string{}

		if out != nil && out.Body != nil && out.Body.Policies != nil && out.Body.Policies.Policy != nil {
			for _, p := range out.Body.Policies.Policy {
				policyNames = append(policyNames, *p.PolicyName)
			}
		}

		if samePolicyNameSet(policyNames, iamv1.Spec.AliCloud.Policies) {
			for _, p := range iamv1.Spec.AliCloud.Policies {
				policyDocument, exists, err := iam.aliCli.GetCustomPolicyDocument(ctx, p.PolicyName)
				if err != nil {
					return false, err
				}
				if exists {
					policyDocuments[p.PolicyName] = policyDocument
				}
			}
		}
		return iam.equalPolicyAliCloud(ctx, log, roleName, *role.AliCloudAssumeRolePolicyDocument, iamv1.Spec.AliCloud.AssumeRolePolicyDocument, policyNames, policyDocuments, iamv1.Spec.AliCloud.Policies)
	}
	return true, errors.New(errNotSupportProvidedVendor)
}

func samePolicyNameSet(policyNames []string, crdpolicies []observabilitypingcapcomv1.AliCloudPolicy) bool {
	specset := mapset.NewSet()
	for _, policy := range crdpolicies {
		specset.Add(policy.PolicyName)
	}
	observset := mapset.NewSet()
	for _, policyName := range policyNames {
		observset.Add(policyName)
	}
	return specset.Difference(observset).Cardinality() == 0 && observset.Difference(specset).Cardinality() == 0
}

func peekPolicy(policies []observabilitypingcapcomv1.AWSRolePolicy, policyName string) *observabilitypingcapcomv1.AWSRolePolicy {
	for _, policy := range policies {
		if policy.PolicyName == policyName {
			return &policy
		}
	}
	return nil
}

func peekPolicyAli(policies []observabilitypingcapcomv1.AliCloudPolicy, policyName string) *observabilitypingcapcomv1.AliCloudPolicy {
	for _, policy := range policies {
		if policy.PolicyName == policyName {
			return &policy
		}
	}
	return nil
}

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// TODO(user): Modify the Reconcile function to compare the state specified by
// the IAM object against the actual cluster state, and then
// perform operations to make the cluster state reflect the state specified by
// the user.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.11.2/pkg/reconcile
func (r *IAMReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := r.Log.WithValues("IAM", req.NamespacedName)
	ctx = log.IntoContext(ctx, logger)
	return r.reconciler.Reconcile(ctx, req, iamfinalizerName)
}

// SetupWithManager sets up the controller with the Manager.
func (r *IAMReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&observabilitypingcapcomv1.IAM{}).
		Complete(r)
}

func (iam *IAM) reconcilerRepoter(name string, interval string, err string) {
	metrics.HandleRegionalControllerRunsMetrics(iam.serviceInfo.Vendor, iam.serviceInfo.Region, "IAMController", name, interval, err)
}