package eks

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/json"
	"fmt"
	"net/url"
	"path"
	"sort"
	"text/template"
	"time"

	"github.com/crossplane-contrib/provider-aws/apis/iam/v1alpha1"
	"github.com/crossplane-contrib/provider-aws/apis/iam/v1beta1"
	apiv1beta1 "github.com/crossplane-contrib/provider-aws/apis/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/util/retry"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/yaml"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const (
	nodeRolePolicyName        = "full-manager-node-role-permission"
	awsLBControllerPolicyName = "aws-load-balancer-controller-permission"
	RoleRefLabelKey           = "role-reference"

	awsAuthConfigMapName         = "aws-auth"
	awsAuthConfigMapNamespace    = "kube-system"
	awsAuthConfigMapKeyRoles     = "mapRoles"
	awsAuthConfigMapNodeUsername = "system:node:{{EC2PrivateDNSName}}"

	sharedBucketReadonlyRolePolicyNameSuffix         = "bucket-readonly"
	sharedBucketDBAuditLogWriterRolePolicyNameSuffix = "db-audit-log-writer"
	sharedBucketReadwriteRolePolicyNameSuffix        = "sb-readwrite"
)

var (
	// tidb worker node role trust policy
	// admin node role and tidb cluster pod can assume to this role
	nodeTrustPolicyTmpl = `{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "ec2.amazonaws.com"
      },
      "Action": [
        "sts:AssumeRole"
      ]
    },
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "{{.RoleARN}}"
      },
      "Action": [
        "sts:AssumeRole"
      ]
    }
  ]
}`

	// sharedBucketReadonlyRoleTrustPolicyTmpl allows the cluster-server to assume shared bucket readonly role
	sharedBucketReadonlyRoleTrustPolicyTmpl = template.Must(template.New("sharedBucketReadonlyRoleTrustPolicy").Parse(`{
	    "Version": "2012-10-17",
	    "Statement": [
	        {
	            "Effect": "Allow",
	            "Principal": {
	                "AWS": [
	                    "{{.RoleARN}}"
	                ]
	            },
	            "Action": "sts:AssumeRole"
	        }
	    ]
	}`))

	sharedBucketReadonlyRolePolicyTmpl = template.Must(template.New("sharedBucketReadonlyRolePolicy").Parse(`{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:GetObjectVersionTagging",
                "s3:GetObjectTagging"
            ],
            "Resource": [
                "arn:aws:s3:::{{.BucketName}}{{.Prefix}}/*"
            ]
        },
		{
	        "Effect": "Allow",
	        "Action": [
	            "s3:ListBucket"
	        ],
	        "Resource": [
	            "arn:aws:s3:::{{.BucketName}}"
	        ]
		}
    ]
    }`))

	sharedBucketReadwriteRolePolicyStatementSid = "ReadWriteWithinPrefix"
	// sharedBucketReadwriteRolePolicyTmpl is appended to the node role to read/write from/to the corresponding shared bucket.
	sharedBucketReadwriteRolePolicyTmpl = template.Must(template.New("sharedBucketReadwriteRolePolicy").Parse(`{
    "Version": "2012-10-17",
    "Statement": [
        {
			"Sid": "{{.Sid}}",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:GetObject",
                "s3:DeleteObject",
                "s3:GetObjectTagging",
                "s3:PutObjectTagging",
                "s3:DeleteObjectTagging"
            ],
            "Resource": [
                "{{.ResourceARN}}"
            ]
        },
		{
	        "Effect": "Allow",
	        "Action": [
	            "s3:ListBucket"
	        ],
	        "Resource": [
	            "arn:aws:s3:::{{.BucketName}}"
	        ]
		}
    ]
    }`))

	// sharedBucketDBAuditLogWriterRoleTrustPolicyTmpl allows the corresponding node role to assume it to write db audit logs
	sharedBucketDBAuditLogWriterRoleTrustPolicyTmpl = template.Must(template.New("sharedBucketDBAuditLogWriterRoleTrustPolicy").Parse(`{
	    "Version": "2012-10-17",
	    "Statement": [
	        {
	            "Effect": "Allow",
	            "Principal": {
	                "AWS": [
	                    "{{.NodeRoleARN}}"
	                ]
	            },
	            "Action": "sts:AssumeRole"
	        }
	    ]
	}`))

	sharedBucketDBAuditLogWriterRolePolicyTmpl = template.Must(template.New("sharedBucketDBAuditLogWriterRolePolicy").Parse(`{
        "Version": "2012-10-17",
	    "Statement": [
	        {
	            "Effect": "Allow",
	            "Action": [
	                "s3:PutObject",
	                "s3:PutObjectTagging"
	            ],
	            "Resource": [
	                "arn:aws:s3:::{{.BucketName}}{{.Prefix}}/*"
	            ]
	        }
	    ]
	}`))

	// See https://github.com/kubernetes-sigs/aws-load-balancer-controller/blob/v2.4.7/docs/install/iam_policy.json
	awsLBControllerPolicyDocument = `{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "iam:CreateServiceLinkedRole"
            ],
            "Resource": "*",
            "Condition": {
                "StringEquals": {
                    "iam:AWSServiceName": "elasticloadbalancing.amazonaws.com"
                }
            }
        },
        {
            "Effect": "Allow",
            "Action": [
                "ec2:DescribeAccountAttributes",
                "ec2:DescribeAddresses",
                "ec2:DescribeAvailabilityZones",
                "ec2:DescribeInternetGateways",
                "ec2:DescribeVpcs",
                "ec2:DescribeVpcPeeringConnections",
                "ec2:DescribeSubnets",
                "ec2:DescribeSecurityGroups",
                "ec2:DescribeInstances",
                "ec2:DescribeNetworkInterfaces",
                "ec2:DescribeTags",
                "ec2:GetCoipPoolUsage",
                "ec2:DescribeCoipPools",
                "elasticloadbalancing:DescribeLoadBalancers",
                "elasticloadbalancing:DescribeLoadBalancerAttributes",
                "elasticloadbalancing:DescribeListeners",
                "elasticloadbalancing:DescribeListenerCertificates",
                "elasticloadbalancing:DescribeSSLPolicies",
                "elasticloadbalancing:DescribeRules",
                "elasticloadbalancing:DescribeTargetGroups",
                "elasticloadbalancing:DescribeTargetGroupAttributes",
                "elasticloadbalancing:DescribeTargetHealth",
                "elasticloadbalancing:DescribeTags"
            ],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "cognito-idp:DescribeUserPoolClient",
                "acm:ListCertificates",
                "acm:DescribeCertificate",
                "iam:ListServerCertificates",
                "iam:GetServerCertificate",
                "waf-regional:GetWebACL",
                "waf-regional:GetWebACLForResource",
                "waf-regional:AssociateWebACL",
                "waf-regional:DisassociateWebACL",
                "wafv2:GetWebACL",
                "wafv2:GetWebACLForResource",
                "wafv2:AssociateWebACL",
                "wafv2:DisassociateWebACL",
                "shield:GetSubscriptionState",
                "shield:DescribeProtection",
                "shield:CreateProtection",
                "shield:DeleteProtection"
            ],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "ec2:AuthorizeSecurityGroupIngress",
                "ec2:RevokeSecurityGroupIngress"
            ],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "ec2:CreateSecurityGroup"
            ],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "ec2:CreateTags"
            ],
            "Resource": "arn:aws:ec2:*:*:security-group/*",
            "Condition": {
                "StringEquals": {
                    "ec2:CreateAction": "CreateSecurityGroup"
                },
                "Null": {
                    "aws:RequestTag/elbv2.k8s.aws/cluster": "false"
                }
            }
        },
        {
            "Effect": "Allow",
            "Action": [
                "ec2:CreateTags",
                "ec2:DeleteTags"
            ],
            "Resource": "arn:aws:ec2:*:*:security-group/*",
            "Condition": {
                "Null": {
                    "aws:RequestTag/elbv2.k8s.aws/cluster": "true",
                    "aws:ResourceTag/elbv2.k8s.aws/cluster": "false"
                }
            }
        },
        {
            "Effect": "Allow",
            "Action": [
                "ec2:AuthorizeSecurityGroupIngress",
                "ec2:RevokeSecurityGroupIngress",
                "ec2:DeleteSecurityGroup"
            ],
            "Resource": "*",
            "Condition": {
                "Null": {
                    "aws:ResourceTag/elbv2.k8s.aws/cluster": "false"
                }
            }
        },
        {
            "Effect": "Allow",
            "Action": [
                "elasticloadbalancing:CreateLoadBalancer",
                "elasticloadbalancing:CreateTargetGroup"
            ],
            "Resource": "*",
            "Condition": {
                "Null": {
                    "aws:RequestTag/elbv2.k8s.aws/cluster": "false"
                }
            }
        },
        {
            "Effect": "Allow",
            "Action": [
                "elasticloadbalancing:CreateListener",
                "elasticloadbalancing:DeleteListener",
                "elasticloadbalancing:CreateRule",
                "elasticloadbalancing:DeleteRule"
            ],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "elasticloadbalancing:AddTags",
                "elasticloadbalancing:RemoveTags"
            ],
            "Resource": [
                "arn:aws:elasticloadbalancing:*:*:targetgroup/*/*",
                "arn:aws:elasticloadbalancing:*:*:loadbalancer/net/*/*",
                "arn:aws:elasticloadbalancing:*:*:loadbalancer/app/*/*"
            ],
            "Condition": {
                "Null": {
                    "aws:RequestTag/elbv2.k8s.aws/cluster": "true",
                    "aws:ResourceTag/elbv2.k8s.aws/cluster": "false"
                }
            }
        },
        {
            "Effect": "Allow",
            "Action": [
                "elasticloadbalancing:AddTags",
                "elasticloadbalancing:RemoveTags"
            ],
            "Resource": [
                "arn:aws:elasticloadbalancing:*:*:listener/net/*/*/*",
                "arn:aws:elasticloadbalancing:*:*:listener/app/*/*/*",
                "arn:aws:elasticloadbalancing:*:*:listener-rule/net/*/*/*",
                "arn:aws:elasticloadbalancing:*:*:listener-rule/app/*/*/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "elasticloadbalancing:ModifyLoadBalancerAttributes",
                "elasticloadbalancing:SetIpAddressType",
                "elasticloadbalancing:SetSecurityGroups",
                "elasticloadbalancing:SetSubnets",
                "elasticloadbalancing:DeleteLoadBalancer",
                "elasticloadbalancing:ModifyTargetGroup",
                "elasticloadbalancing:ModifyTargetGroupAttributes",
                "elasticloadbalancing:DeleteTargetGroup"
            ],
            "Resource": "*",
            "Condition": {
                "Null": {
                    "aws:ResourceTag/elbv2.k8s.aws/cluster": "false"
                }
            }
        },
        {
            "Effect": "Allow",
            "Action": [
                "elasticloadbalancing:AddTags"
            ],
            "Resource": [
                "arn:aws:elasticloadbalancing:*:*:targetgroup/*/*",
                "arn:aws:elasticloadbalancing:*:*:loadbalancer/net/*/*",
                "arn:aws:elasticloadbalancing:*:*:loadbalancer/app/*/*"
            ],
            "Condition": {
                "StringEquals": {
                    "elasticloadbalancing:CreateAction": [
                        "CreateTargetGroup",
                        "CreateLoadBalancer"
                    ]
                },
                "Null": {
                    "aws:RequestTag/elbv2.k8s.aws/cluster": "false"
                }
            }
        },
        {
            "Effect": "Allow",
            "Action": [
                "elasticloadbalancing:RegisterTargets",
                "elasticloadbalancing:DeregisterTargets"
            ],
            "Resource": "arn:aws:elasticloadbalancing:*:*:targetgroup/*/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "elasticloadbalancing:SetWebAcl",
                "elasticloadbalancing:ModifyListener",
                "elasticloadbalancing:AddListenerCertificates",
                "elasticloadbalancing:RemoveListenerCertificates",
                "elasticloadbalancing:ModifyRule"
            ],
            "Resource": "*"
        }
    ]
}`
)

const (
	iamReadyPropagationWait = time.Minute
	iamReadyRequeueInterval = 20 * time.Second
)

func waitNodeRoleIAMReady() deploy.Result {
	return deploy.Panic(fmt.Errorf("node role IAM resources are not ready")).
		WithRequeuAfter(iamReadyRequeueInterval)
}

// NodeRole task will create a role for all TiDB cluster nodes and attach the policy to the role.
type NodeRole struct {
	*task.BaseTask
	// used for clock injection in unit test
	Now func() time.Time
}

func NewNodeRole(ctx *task.Context) deploy.Task {
	return &NodeRole{BaseTask: &task.BaseTask{Context: ctx}, Now: time.Now}
}

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

func (t *NodeRole) Create(ctx context.Context) deploy.Result {
	if err := t.createNodeRole(ctx); err != nil {
		return deploy.Panic(err)
	}

	nodeRole := &v1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.GetAWSNodeRoleName(),
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(nodeRole), nodeRole); err != nil {
		return deploy.Panic(err)
	}

	if t.Cluster.SharedBucketEnabled() {
		// append the read/write permissions on shared bucket to node role
		if err := t.grantReadwritePolicyOnSharedBucketToNodeRole(ctx, nodeRole); err != nil {
			return deploy.Error(err)
		}

		// the following roles for shared bucket don't block latter tasks.
		if t.Cluster.Status.SharedBucket == nil {
			t.Cluster.Status.SharedBucket = &cloudv1.SharedBucketStatus{}
		}
		if t.Cluster.Status.SharedBucket.AWS == nil {
			t.Cluster.Status.SharedBucket.AWS = &cloudv1.AWSSharedBucketStatus{}
		}
		if err := t.createSharedBucketDBAuditLogWriterRole(ctx); err != nil {
			return deploy.Error(err)
		}
		if err := t.createSharedBucketReadonlyRole(ctx); err != nil {
			return deploy.Error(err)
		}
	}

	return deploy.Done()
}

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

	roleName := t.GetAWSNodeRoleName()

	// check the role is ready (Ready condition, not just ARN presence)
	role := &v1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{Name: roleName},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(role), role); err != nil {
		return deploy.Panic(err)
	}
	if role.Status.AtProvider.ARN == "" || !isReadyConditionTrue(role.Status.Conditions) {
		t.Logger.Info("node role not ready, waiting", "roleName", roleName, "arn", role.Status.AtProvider.ARN)
		return waitNodeRoleIAMReady()
	}

	// check all attachments are ready and match expected count
	attachments := &v1beta1.RolePolicyAttachmentList{}
	if err := t.Local.List(ctx, attachments, client.MatchingLabels{RoleRefLabelKey: roleName}); err != nil {
		return deploy.Panic(err)
	}
	expectedCount := t.expectedNodeRoleAttachmentCount()
	if len(attachments.Items) < expectedCount {
		t.Logger.Info("not all attachments created yet, waiting", "roleName", roleName, "found", len(attachments.Items), "expected", expectedCount)
		return waitNodeRoleIAMReady()
	}
	for _, a := range attachments.Items {
		if !isReadyConditionTrue(a.Status.Conditions) {
			t.Logger.Info("attachment not ready", "name", a.Name)
			return waitNodeRoleIAMReady()
		}
	}

	// check instance profile is ready
	instanceProfile := &v1alpha1.InstanceProfile{
		ObjectMeta: metav1.ObjectMeta{Name: roleName},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(instanceProfile), instanceProfile); err != nil {
		if apierrors.IsNotFound(err) {
			t.Logger.Info("instance profile not found yet, waiting", "roleName", roleName)
			return waitNodeRoleIAMReady()
		}
		return deploy.Panic(err)
	}
	if !isReadyConditionTrue(instanceProfile.Status.Conditions) {
		t.Logger.Info("instance profile not ready", "roleName", roleName)
		return waitNodeRoleIAMReady()
	}

	// wait for propagation cooldown after the latest resource became ready
	// include role, all attachments, and instance profile in the calculation
	latestRole := getReadyTransition(role.Status.Conditions)
	latestAttachment := latestReadyTransitionFromAttachments(attachments.Items)
	latestIP := getReadyTransition(instanceProfile.Status.Conditions)
	latest := latestRole
	if latestAttachment.After(latest) {
		latest = latestAttachment
	}
	if latestIP.After(latest) {
		latest = latestIP
	}
	if latest.IsZero() {
		t.Logger.Info("ready transition timestamps not populated yet, waiting", "roleName", roleName)
		return waitNodeRoleIAMReady()
	}
	if t.Now().Sub(latest) < iamReadyPropagationWait {
		t.Logger.Info("waiting IAM propagation cooldown", "since", t.Now().Sub(latest), "required", iamReadyPropagationWait)
		return waitNodeRoleIAMReady()
	}

	t.Logger.Info("node role IAM resources are fully ready", "roleName", roleName)
	return deploy.Done()
}

// expectedNodeRoleAttachmentCount returns the minimum number of RolePolicyAttachments
// that must exist before the node role is considered fully set up.
// This is: 2 custom policies (nodeRolePolicy + awsLBControllerPolicy) + managed policies from config
// + 1 shared bucket readwrite policy if SharedBucket is enabled.
func (t *NodeRole) expectedNodeRoleAttachmentCount() int {
	const customPolicyCount = 2 // nodeRolePolicyName + awsLBControllerPolicyName
	count := customPolicyCount + len(t.Config.ClusterController.AWS.NodeRoleManagedPolicies)
	if t.Cluster.SharedBucketEnabled() {
		count++ // shared bucket readwrite policy attachment
	}
	return count
}

func (t *NodeRole) createNodeRole(ctx context.Context) error {
	if t.IsLegacy() {
		// legacy network is available, use the node role created by Network Controller.
		// Cluster.Status.NodeIdentity is updated in k8s_info_saver task
		return nil
	}

	// TODO(cluster-controller): use namer to generate labels?
	nodeRoleARN, err := t.prepareNodeRole(ctx, t.Local, t.GetAWSNodeRoleName(), t.GetAWSProviderConfigName(), t.Cluster.Labels)
	if err != nil {
		return fmt.Errorf("prepareNodeRole: %w", err)
	}

	externalID := t.GetAWSAuditLogRoleExternalID()
	t.Cluster.Status.NodeIdentity = &cloudv1.NodeIdentity{
		AWS: &cloudv1.AWSIdentity{
			RoleARN:    nodeRoleARN,
			ExternalID: externalID,
		},
	}

	if err := t.addNodeRoleARNIntoAWSAuth(ctx, nodeRoleARN); err != nil {
		return fmt.Errorf("addNodeRoleARNIntoAWSAuth: %w", err)
	}

	return nil
}

// Delete will delete all role policy attachments and the role itself.
// Must delete the role attachments before deleting the role.
func (t *NodeRole) Delete(ctx context.Context) deploy.Result {
	if err := t.deleteNodeRole(ctx); err != nil {
		return deploy.Panic(err)
	}

	if t.Cluster.SharedBucketEnabled() {
		if err := t.deleteSharedBucketDBAuditLogWriterRole(ctx); err != nil {
			return deploy.Error(err)
		}
		if err := t.deleteSharedBucketReadonlyRole(ctx); err != nil {
			return deploy.Error(err)
		}
	}
	return deploy.Done()
}

func (t *NodeRole) deleteNodeRole(ctx context.Context) error {
	if t.IsLegacy() {
		return nil
	}

	roleName := t.GetAWSNodeRoleName()
	if err := client.IgnoreNotFound(t.Local.DeleteAllOf(context.Background(),
		&v1beta1.RolePolicyAttachment{}, client.MatchingLabels{RoleRefLabelKey: roleName})); err != nil {
		return fmt.Errorf("delete rolePolicyAttachment: %w", err)
	}

	if err := client.IgnoreNotFound(t.Local.DeleteAllOf(context.Background(),
		&v1beta1.Policy{}, client.MatchingLabels{RoleRefLabelKey: roleName})); err != nil {
		return fmt.Errorf("delete policy: %w", err)
	}

	instanceProfile := &v1alpha1.InstanceProfile{
		ObjectMeta: metav1.ObjectMeta{
			Name: roleName, // same as role name
		},
	}
	if err := client.IgnoreNotFound(t.Local.Delete(context.Background(), instanceProfile)); err != nil {
		return fmt.Errorf("delete instanceProfile: %w", err)
	}

	role := &v1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name: roleName,
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(role), role); err != nil {
		if errors.IsNotFound(err) {
			return nil
		}

		return fmt.Errorf("get IAM role: %w", err)
	}

	if err := t.removeRoleARNFromAWSAuth(ctx, role.Status.AtProvider.ARN); err != nil {
		return fmt.Errorf("removeRoleARNFromAWSAuth: %w", err)
	}

	if err := client.IgnoreNotFound(t.Local.Delete(context.Background(), role)); err != nil {
		return fmt.Errorf("delete IAM role: %w", err)
	}

	return nil
}

func (t *NodeRole) getTrustPolicy() (string, error) {
	var trustPolicyBuffer bytes.Buffer
	adminNodeRole := struct{ RoleARN string }{
		RoleARN: t.Cluster.Spec.Kubernetes.Cloud.AWS.AdminNodeRole,
	}
	tmpl := template.Must(template.New("trustPolicy").Parse(nodeTrustPolicyTmpl))
	if err := tmpl.Execute(&trustPolicyBuffer, adminNodeRole); err != nil {
		return "", err
	}
	return trustPolicyBuffer.String(), nil
}

func (t *NodeRole) prepareNodeRole(ctx context.Context, cli client.Client, roleName, providerConfig string, labels map[string]string) (string, error) {
	config := &apiv1beta1.ProviderConfig{
		ObjectMeta: metav1.ObjectMeta{
			Name: providerConfig,
		},
	}
	err := cli.Get(ctx, client.ObjectKey{Name: config.Name}, config)
	if err != nil {
		return "", err
	}

	if config.Spec.AssumeRoleARN == nil {
		return "", fmt.Errorf("assume role arn is not set")
	}
	trustPolicy, err := t.getTrustPolicy()
	if err != nil {
		return "", err
	}
	role, roleReady, err := createNodeRole(ctx, cli, roleName, trustPolicy, providerConfig, labels)
	if err != nil {
		return "", err
	}

	policy, err := util.GetAWSPolicyFromFile(t.Config.ClusterController.AWS.NodeRolePolicyFile)
	if err != nil {
		return "", err
	}
	policyARN, policyReady, err := createNodeRolePolicy(ctx, cli, roleName, policy, providerConfig)
	if err != nil {
		return "", err
	}

	awsLBCPolicyARN, awsLBCPolicyReady, err := createPolicyForALBController(ctx, cli, roleName, providerConfig)
	if err != nil {
		return "", fmt.Errorf("failed to create policy for AWS LB Controller: %v", err)
	}

	var nodeRoleARN string
	if roleReady && policyReady && awsLBCPolicyReady {
		err := t.createRoleAttachment(ctx, cli, []string{policyARN, awsLBCPolicyARN}, role, providerConfig)
		if err != nil {
			return "", err
		}

		// Create an instance profile per cluster for AutoScalingGroups.
		// Must wait for the role to be ready before creating the instance profile,
		// since https://github.com/crossplane-contrib/provider-aws/issues/1493
		if err = createInstanceProfile(ctx, cli, roleName, providerConfig); err != nil {
			return "", err
		}

		nodeRoleARN = role.Status.AtProvider.ARN
	} else {
		return "", fmt.Errorf("node role or policy is not ready")
	}

	return nodeRoleARN, nil
}

func createNodeRole(ctx context.Context, cli client.Client, roleName, trustPolicy, providerConfig string, labels map[string]string) (role *v1beta1.Role, ready bool, err error) {
	tags := cloudv1.TagsFromLabels(labels)
	tags = util.AddResourceName(tags, roleName)

	role = &v1beta1.Role{
		TypeMeta: metav1.TypeMeta{
			APIVersion: "iam.aws.crossplane.io/v1beta1",
			Kind:       "Role",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name: roleName,
		},
		Spec: v1beta1.RoleSpec{
			ForProvider: v1beta1.RoleParameters{
				Description:              pointer.String("Node role for TiDB clusters"),
				AssumeRolePolicyDocument: trustPolicy,
				Tags:                     iamTagFromLabels(tags),
			},
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}

	err = k8s.Apply(ctx, cli, role)
	if err != nil {
		return role, false, err
	}
	if role.Status.AtProvider.ARN != "" {
		ready = true
	}

	return role, ready, nil
}

func createInstanceProfile(ctx context.Context, cli client.Client, roleName, providerConfig string) error {
	instanceProfile := &v1alpha1.InstanceProfile{
		ObjectMeta: metav1.ObjectMeta{
			Name: roleName, // same as role name
		},
		Spec: v1alpha1.InstanceProfileSpec{
			ForProvider: v1alpha1.InstanceProfileParameters{
				CustomInstanceProfileParameters: v1alpha1.CustomInstanceProfileParameters{
					RoleRef: &xpv1.Reference{
						Name: roleName,
					},
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}

	return k8s.Apply(ctx, cli, instanceProfile)
}

func buildPolicyName(roleName, policyNameSuffix string) string {
	return fmt.Sprintf("%s-%s", roleName, policyNameSuffix)
}

func policyAttachmentName(name, policyARN string) string {
	hash := fmt.Sprintf("%x", md5.Sum([]byte(policyARN)))[:8]
	return fmt.Sprintf("%s-%s", name, hash)
}

// helpers for IAM readiness checks
func isReadyConditionTrue(conds []xpv1.Condition) bool {
	for _, c := range conds {
		if c.Type == xpv1.TypeReady && c.Status == corev1.ConditionTrue {
			return true
		}
	}
	return false
}

func getReadyTransition(conds []xpv1.Condition) time.Time {
	for _, c := range conds {
		if c.Type == xpv1.TypeReady && c.Status == corev1.ConditionTrue {
			if !c.LastTransitionTime.IsZero() {
				return c.LastTransitionTime.Time
			}
		}
	}
	return time.Time{}
}

func latestReadyTransitionFromAttachments(attachments []v1beta1.RolePolicyAttachment) time.Time {
	var latest time.Time
	for _, a := range attachments {
		t := getReadyTransition(a.Status.Conditions)
		if t.After(latest) {
			latest = t
		}
	}
	return latest
}

func (t *NodeRole) createRoleAttachment(ctx context.Context, cli client.Client, selfManagedPolicyARN []string, role *v1beta1.Role, providerConfig string) (err error) {
	policyARNs := append(selfManagedPolicyARN, t.Config.ClusterController.AWS.NodeRoleManagedPolicies...)
	for _, policyARN := range policyARNs {
		attachment := &v1beta1.RolePolicyAttachment{
			ObjectMeta: metav1.ObjectMeta{
				Name:   policyAttachmentName(role.Name, policyARN),
				Labels: map[string]string{RoleRefLabelKey: role.Name},
			},
			Spec: v1beta1.RolePolicyAttachmentSpec{
				ForProvider: v1beta1.RolePolicyAttachmentParameters{
					RoleNameRef: &xpv1.Reference{
						Name: role.Name,
					},
					PolicyARN: policyARN,
				},
				ResourceSpec: xpv1.ResourceSpec{
					DeletionPolicy:          xpv1.DeletionDelete,
					ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
				},
			},
		}

		err = controllerutil.SetOwnerReference(role, attachment, cli.Scheme())
		if err != nil {
			return err
		}

		_, err = controllerutil.CreateOrUpdate(ctx, cli, attachment, func() error {
			return nil
		})
		if err != nil {
			return err
		}
	}

	return nil
}

func createNodeRolePolicy(ctx context.Context, cli client.Client, roleName, policyDocument, providerConfig string) (policyARN string, ready bool, err error) {
	policy := v1beta1.Policy{
		ObjectMeta: metav1.ObjectMeta{
			Name:   buildPolicyName(roleName, nodeRolePolicyName),
			Labels: map[string]string{RoleRefLabelKey: roleName},
		},
		Spec: v1beta1.PolicySpec{
			ForProvider: v1beta1.PolicyParameters{
				Description: pointer.String("Node role permissions"),
				Name:        buildPolicyName(roleName, nodeRolePolicyName),
			},
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}
	_, err = controllerutil.CreateOrUpdate(ctx, cli, &policy, func() error {
		policy.Spec.ForProvider.Document = *pointer.String(policyDocument)
		return nil
	})
	if err != nil {
		return
	}
	if policy.Status.AtProvider.ARN == "" {
		ready = false
		return
	}

	ready = true
	return policy.Status.AtProvider.ARN, ready, nil
}

func createPolicyForALBController(ctx context.Context, cli client.Client, roleName, providerConfig string) (policyARN string, ready bool, err error) {
	policy := v1beta1.Policy{
		ObjectMeta: metav1.ObjectMeta{
			Name:   buildPolicyName(roleName, awsLBControllerPolicyName),
			Labels: map[string]string{RoleRefLabelKey: roleName},
		},
		Spec: v1beta1.PolicySpec{
			ForProvider: v1beta1.PolicyParameters{
				Description: pointer.String("AWS LB Controller permissions"),
				Name:        buildPolicyName(roleName, awsLBControllerPolicyName),
			},
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}

	_, err = controllerutil.CreateOrUpdate(ctx, cli, &policy, func() error {
		policy.Spec.ForProvider.Document = *pointer.String(awsLBControllerPolicyDocument)
		return nil
	})
	if err != nil {
		return
	}
	if policy.Status.AtProvider.ARN == "" {
		ready = false
		return
	}

	ready = true
	return policy.Status.AtProvider.ARN, ready, nil
}

func iamTagFromLabels(kv map[string]string) []v1beta1.Tag {
	var tags []v1beta1.Tag
	for k, v := range kv {
		tags = append(tags, v1beta1.Tag{Key: k, Value: v})
	}
	sort.Slice(tags, func(i, j int) bool {
		return tags[i].Key < tags[j].Key
	})
	return tags
}

type RoleIdentity struct {
	RoleARN  string   `json:"rolearn,omitempty"`
	Username string   `json:"username,omitempty"`
	Groups   []string `json:"groups,omitempty"`

	reserved map[string]interface{}
}

func (a *RoleIdentity) UnmarshalJSON(body []byte) error {
	d := map[string]interface{}{}

	if err := json.Unmarshal(body, &d); err != nil {
		return err
	}

	for k, v := range d {
		switch k {
		case "rolearn":
			roleARN, ok := v.(string)
			if !ok {
				return fmt.Errorf("rolearn is not string")
			}
			a.RoleARN = roleARN
		case "username":
			username, ok := v.(string)
			if !ok {
				return fmt.Errorf("username is not string")
			}
			a.Username = username
		case "groups":
			gs, ok := v.([]interface{})
			if !ok {
				return fmt.Errorf("groups is not []interface{}")
			}

			groups := []string{}
			for _, g := range gs {
				group, ok := g.(string)
				if !ok {
					return fmt.Errorf("groups is not []string{}")
				}
				groups = append(groups, group)
			}

			a.Groups = groups

		default:
			if a.reserved == nil {
				a.reserved = map[string]interface{}{}
			}
			a.reserved[k] = v
		}
	}

	return nil
}

func (a *RoleIdentity) MarshalJSON() ([]byte, error) {
	d := map[string]interface{}{}
	for k, v := range a.reserved {
		d[k] = v
	}

	if a.RoleARN != "" {
		d["rolearn"] = a.RoleARN
	}
	if a.Username != "" {
		d["username"] = a.Username
	}
	if a.Groups != nil {
		d["groups"] = a.Groups
	}

	return json.Marshal(&d)
}

func (t *NodeRole) addNodeRoleARNIntoAWSAuth(ctx context.Context, roleARN string) error {
	c := t.EnvProvider.GetClient()
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      awsAuthConfigMapName,
			Namespace: awsAuthConfigMapNamespace,
		},
	}
	if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
		if err := c.Get(ctx, client.ObjectKeyFromObject(cm), cm); err != nil {
			return err
		}

		rs, err := t.getRoleIdentities(cm)
		if err != nil {
			return err
		}

		for _, r := range rs {
			// found, just return
			if r.RoleARN == roleARN {
				return nil
			}
		}

		rs = append(rs, RoleIdentity{
			RoleARN:  roleARN,
			Username: awsAuthConfigMapNodeUsername,
			Groups: []string{
				"system:bootstrappers",
				"system:nodes",
			},
		})

		if err := t.setRoleIdentities(cm, rs); err != nil {
			return err
		}

		if err := c.Update(ctx, cm); err != nil {
			return err
		}

		return nil
	}); err != nil {
		return err
	}

	return nil

}

func (t *NodeRole) removeRoleARNFromAWSAuth(ctx context.Context, roleARN string) error {
	c := t.EnvProvider.GetClient()
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      awsAuthConfigMapName,
			Namespace: awsAuthConfigMapNamespace,
		},
	}

	if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
		if err := c.Get(ctx, client.ObjectKeyFromObject(cm), cm); err != nil {
			return err
		}

		rs, err := t.getRoleIdentities(cm)
		if err != nil {
			return err
		}

		found := false
		for _, r := range rs {
			if r.RoleARN == roleARN {
				found = true
			}
		}

		if !found {
			return nil
		}

		nrs := []RoleIdentity{}

		for _, r := range rs {
			if r.RoleARN != roleARN {
				nrs = append(nrs, r)
			}
		}

		if err := t.setRoleIdentities(cm, nrs); err != nil {
			return err
		}

		if err := c.Update(ctx, cm); err != nil {
			return err
		}

		return nil
	}); err != nil {
		return err
	}

	return nil
}

func (t *NodeRole) getRoleIdentities(cm *corev1.ConfigMap) ([]RoleIdentity, error) {
	data, ok := cm.Data[awsAuthConfigMapKeyRoles]
	if !ok {
		return nil, nil
	}

	rs := []RoleIdentity{}

	if err := yaml.Unmarshal([]byte(data), &rs); err != nil {
		return nil, err
	}

	return rs, nil
}

func (t *NodeRole) setRoleIdentities(cm *corev1.ConfigMap, rs []RoleIdentity) error {
	body, err := yaml.Marshal(&rs)
	if err != nil {
		return err
	}

	cm.Data[awsAuthConfigMapKeyRoles] = string(body)

	return nil
}

type sharedBucketReadonlyRoleTrustPolicyArgs struct {
	RoleARN string
}

type sharedBucketReadonlyRolePolicyArgs struct {
	BucketName string // for example: "bucket1"
	Prefix     string // for example: "/123/456/789/"
}

func (t *NodeRole) createSharedBucketReadonlyRole(ctx context.Context) error {
	// build role trust policy
	roleName := namer.SharedBucketReadonlyRoleName(t.Cluster)
	roleDescription := "Readonly role for accessing shared bucket, usually used by control-plane."
	var trustPolicyBuffer bytes.Buffer
	err := sharedBucketReadonlyRoleTrustPolicyTmpl.Execute(&trustPolicyBuffer, &sharedBucketReadonlyRoleTrustPolicyArgs{
		RoleARN: t.Config.FullManage.AWSClusterServerRoleARN,
	})
	if err != nil {
		return fmt.Errorf("build sharedBucketReadonlyRoleTrustPolicy: %w", err)
	}

	// build Policy document
	policyDescription := "Readonly permission on shared bucket"
	u, err := url.Parse(t.Cluster.Spec.SharedBucket.BucketURL)
	if err != nil {
		return fmt.Errorf("parse shared bucket URL: %w", err)
	}
	var policyBuffer bytes.Buffer
	err = sharedBucketReadonlyRolePolicyTmpl.Execute(&policyBuffer, &sharedBucketReadonlyRolePolicyArgs{
		BucketName: u.Host,
		Prefix:     path.Clean(u.Path),
	})
	if err != nil {
		return fmt.Errorf("build sharedBucketReadonlyRolePolicy: %w", err)
	}

	role, err := t.createRoleAndBindPolicy(ctx,
		roleName, roleDescription, trustPolicyBuffer.String(),
		sharedBucketReadonlyRolePolicyNameSuffix, policyBuffer.String(), policyDescription)
	if err != nil {
		return err
	}
	t.Cluster.Status.SharedBucket.AWS.ReadonlyRoleARN = role.Status.AtProvider.ARN
	return nil
}

type sharedBucketDBAuditLogWriterRoleTrustPolicyArgs struct {
	NodeRoleARN string
}

type sharedBucketDBAuditLogWriterRolePolicyArgs struct {
	BucketName string // for example: "bucket1"
	Prefix     string // for example: "/123/456/"
}

func (t *NodeRole) createSharedBucketDBAuditLogWriterRole(ctx context.Context) error {
	// build role trust policy
	roleName := namer.SharedBucketAuditLogWriterRoleName(t.Cluster)
	roleDescription := "DB audit log writer role which is used by node role to assume it to write db audit logs."
	var trustPolicyBuffer bytes.Buffer
	err := sharedBucketDBAuditLogWriterRoleTrustPolicyTmpl.Execute(&trustPolicyBuffer, &sharedBucketDBAuditLogWriterRoleTrustPolicyArgs{
		NodeRoleARN: t.GetAWSNodeRoleARN(),
	})
	if err != nil {
		return fmt.Errorf("build sharedBucketDBAuditLogWriterRoleTrustPolicy: %w", err)
	}

	// build policy document
	policyDescription := "Write permissions on shared bucket, used for writing db audit logs."
	u, err := url.Parse(t.Cluster.Spec.SharedBucket.BucketURL)
	if err != nil {
		return fmt.Errorf("parse shared bucket URL: %w", err)
	}
	var policyBuffer bytes.Buffer
	err = sharedBucketDBAuditLogWriterRolePolicyTmpl.Execute(&policyBuffer, &sharedBucketDBAuditLogWriterRolePolicyArgs{
		BucketName: u.Host,
		// grant permissions within audit_log directory only
		Prefix: path.Join(u.Path, consts.SharedBucketPrefixDBAuditLog),
	})
	if err != nil {
		return fmt.Errorf("build sharedBucketDBAuditLogWriterRolePolicy: %w", err)
	}

	role, err := t.createRoleAndBindPolicy(ctx,
		roleName, roleDescription, trustPolicyBuffer.String(),
		sharedBucketDBAuditLogWriterRolePolicyNameSuffix, policyBuffer.String(), policyDescription)
	if err != nil {
		return err
	}
	t.Cluster.Status.SharedBucket.AWS.DBAuditLogWriterRoleARN = role.Status.AtProvider.ARN
	return nil
}

func (t *NodeRole) deleteSharedBucketReadonlyRole(ctx context.Context) error {
	roleName := namer.SharedBucketReadonlyRoleName(t.Cluster)
	return t.cascadeDeleteRole(ctx, roleName)
}

func (t *NodeRole) deleteSharedBucketDBAuditLogWriterRole(ctx context.Context) error {
	roleName := namer.SharedBucketAuditLogWriterRoleName(t.Cluster)
	return t.cascadeDeleteRole(ctx, roleName)
}

// cascadeDeleteRole will cascade delete the Role, RolePolicy, RolePolicyAttachment
func (t *NodeRole) cascadeDeleteRole(ctx context.Context, roleName string) error {
	if err := client.IgnoreNotFound(t.Local.DeleteAllOf(ctx,
		&v1beta1.RolePolicyAttachment{}, client.MatchingLabels{RoleRefLabelKey: roleName})); err != nil {
		return fmt.Errorf("delete rolePolicyAttachment: %w", err)
	}
	if err := client.IgnoreNotFound(t.Local.DeleteAllOf(ctx,
		&v1beta1.Policy{}, client.MatchingLabels{RoleRefLabelKey: roleName})); err != nil {
		return fmt.Errorf("delete policy: %w", err)
	}
	role := &v1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name: roleName,
		},
	}
	if err := client.IgnoreNotFound(t.Local.Delete(ctx, role)); err != nil {
		return fmt.Errorf("delete IAM role: %w", err)
	}
	return nil
}

type sharedBucketReadwriteRolePolicyArgs struct {
	BucketName  string // for example: "bucket1"
	ResourceARN string // for example: "arn:aws:s3:::bucket1/123/456/789/*"
	Sid         string // for example: "ReadWriteWithinPrefix
}

func (t *NodeRole) grantReadwritePolicyOnSharedBucketToNodeRole(ctx context.Context, nodeRole *v1beta1.Role) error {
	// build Policy document
	policyDescription := "Readwrite permissions on shared bucket."
	u, err := url.Parse(t.Cluster.Spec.SharedBucket.BucketURL)
	if err != nil {
		return fmt.Errorf("parse shared bucket URL: %w", err)
	}
	resourceArn := fmt.Sprintf("arn:aws:s3:::%s%s/*", u.Host, path.Clean(u.Path))
	var policyBuffer bytes.Buffer
	err = sharedBucketReadwriteRolePolicyTmpl.Execute(&policyBuffer, &sharedBucketReadwriteRolePolicyArgs{
		BucketName:  u.Host,
		ResourceARN: resourceArn,
		Sid:         sharedBucketReadwriteRolePolicyStatementSid,
	})
	if err != nil {
		return fmt.Errorf("build sharedBucketReadWriteRolePolicy: %w", err)
	}

	policyName := buildPolicyName(nodeRole.Name, sharedBucketReadwriteRolePolicyNameSuffix)
	attachment := &v1beta1.RolePolicyAttachment{}
	err = t.Local.Get(ctx, client.ObjectKey{Name: policyAttachmentName(nodeRole.Name, policyName)}, attachment)
	if err != nil && !apierrors.IsNotFound(err) {
		return err
	}
	// If the policyAttachment is not found, it means this is the first time we grant permissions to the node role.
	if apierrors.IsNotFound(err) {
		return t.bindPolicyToRole(ctx, nodeRole, sharedBucketReadwriteRolePolicyNameSuffix, policyBuffer.String(), policyDescription)
	}

	// The maxium count of managed policies per role on AWS is 10 (hard limit is 20).
	// In most cases, we already bind 8 policies to a node role. So that we can't immoderately bind policies to the node role.
	// Instead, we will update the policy document to append the bucket resource of this cluster to the policy.
	// Refer to https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_iam-quotas.html .
	//
	// If got here, it means the policy is already created, so we just append the bucket resource of this cluster to the policy
	// Currently, only cloud-native clusters will use the shared node role per data-plane.
	policy := &v1beta1.Policy{}
	err = t.Local.Get(ctx, client.ObjectKey{Name: policyName}, policy)
	if err != nil {
		return err
	}
	document := &util.PolicyDocument{}
	if err := json.Unmarshal([]byte(policy.Spec.ForProvider.Document), document); err != nil {
		return fmt.Errorf("unmarshal policy document: %w", err)
	}
	readwriteStatement, ok := document.FindStatementBySid(sharedBucketReadwriteRolePolicyStatementSid)
	if !ok {
		return fmt.Errorf("readwrite statement not found")
	}
	readwriteStatement.AppendResource(resourceArn)
	newDoc, err := document.ToString()
	if err != nil {
		return fmt.Errorf("marshal policy document: %w", err)
	}
	if policy.Spec.ForProvider.Document == newDoc {
		return nil
	}

	policy.Spec.ForProvider.Document = newDoc
	return t.Local.Update(ctx, policy)
}

// createRoleAndBindPolicy creates an IAM role with trust policy and then bind a policy to grant permissions
func (t *NodeRole) createRoleAndBindPolicy(ctx context.Context, roleName, roleDescription, trustPolicyDocument, policyNameSuffix, policyDocument, policyDescription string) (*v1beta1.Role, error) {
	cli := t.Local
	providerConfig := t.GetAWSProviderConfigName()

	// build and create role
	tags := cloudv1.TagsFromLabels(t.Cluster.Labels)
	tags = util.AddResourceName(tags, roleName)
	role := &v1beta1.Role{
		TypeMeta: metav1.TypeMeta{
			APIVersion: "iam.aws.crossplane.io/v1beta1",
			Kind:       "Role",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name:   roleName,
			Labels: t.Cluster.Labels,
		},
		Spec: v1beta1.RoleSpec{
			ForProvider: v1beta1.RoleParameters{
				Description:              pointer.String(roleDescription),
				AssumeRolePolicyDocument: trustPolicyDocument,
				Tags:                     iamTagFromLabels(tags),
			},
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}
	if err := k8s.Apply(ctx, cli, role); err != nil {
		return nil, fmt.Errorf("apply role %s: %w", role.Name, err)
	}

	if err := t.bindPolicyToRole(ctx, role, policyNameSuffix, policyDocument, policyDescription); err != nil {
		return nil, err
	}

	return role, nil
}

func (t *NodeRole) bindPolicyToRole(ctx context.Context, role *v1beta1.Role, policyNameSuffix, policyDocument, policyDescription string) error {
	roleName := role.Name
	cli := t.Local
	providerConfig := t.GetAWSProviderConfigName()
	policyName := buildPolicyName(roleName, policyNameSuffix)

	// build and create Policy
	policy := v1beta1.Policy{
		ObjectMeta: metav1.ObjectMeta{
			Name:   policyName,
			Labels: map[string]string{RoleRefLabelKey: roleName},
		},
		Spec: v1beta1.PolicySpec{
			ForProvider: v1beta1.PolicyParameters{
				Description: pointer.String(policyDescription),
				Name:        policyName,
			},
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}
	_, err := controllerutil.CreateOrUpdate(ctx, cli, &policy, func() error {
		policy.Spec.ForProvider.Document = policyDocument
		return nil
	})
	if err != nil {
		return fmt.Errorf("apply policy %s: %w", policy.Name, err)
	}

	roleReady := role.Status.AtProvider.ARN != ""
	policyReady := policy.Status.AtProvider.ARN != ""
	if !roleReady || !policyReady {
		return fmt.Errorf("role(%s, %t) or policy(%s, %t) is not ready", role.Name, roleReady, policy.Name, policyReady)
	}

	// build and create RolePolicyAttachment
	// Note: attachment must be created after the policy is ready
	attachment := &v1beta1.RolePolicyAttachment{
		ObjectMeta: metav1.ObjectMeta{
			Name:   policyAttachmentName(role.Name, policy.Name),
			Labels: map[string]string{RoleRefLabelKey: role.Name},
		},
		Spec: v1beta1.RolePolicyAttachmentSpec{
			ForProvider: v1beta1.RolePolicyAttachmentParameters{
				RoleNameRef: &xpv1.Reference{
					Name: role.Name,
				},
				PolicyARN: policy.Status.AtProvider.ARN,
			},
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}
	err = controllerutil.SetOwnerReference(role, attachment, cli.Scheme())
	if err != nil {
		return err
	}
	_, err = controllerutil.CreateOrUpdate(ctx, cli, attachment, func() error {
		return nil
	})
	if err != nil {
		return err
	}
	attachmentReady := attachment.Status.AtProvider.AttachedPolicyARN != ""
	if !attachmentReady {
		return fmt.Errorf("attachment(%s) is not ready", attachment.Name)
	}

	return nil
}