data "aws_iam_policy_document" "workload_assume_role" {
  count = local.iam_enabled ? 1 : 0

  statement {
    sid     = "TrustedCp9Principals"
    effect  = "Allow"
    actions = ["sts:AssumeRole"]

    principals {
      type        = "AWS"
      identifiers = local.trusted_principal_arns
    }

    condition {
      test     = "StringEquals"
      variable = "aws:PrincipalTag/cp9-owner"
      values   = [local.owner]
    }

    condition {
      test     = "StringEquals"
      variable = "aws:PrincipalTag/cp9-session"
      values   = [local.session_name]
    }
  }

  statement {
    sid     = "TrustedCp9TagSessionPrincipals"
    effect  = "Allow"
    actions = ["sts:TagSession"]

    principals {
      type        = "AWS"
      identifiers = local.trusted_principal_arns
    }
  }

  dynamic "statement" {
    for_each = local.remote_ec2_k3d_enabled ? [aws_iam_role.remote_runtime[0].arn] : []

    content {
      sid     = "TrustedRemoteRuntimePrincipal"
      effect  = "Allow"
      actions = ["sts:AssumeRole", "sts:TagSession"]

      principals {
        type        = "AWS"
        identifiers = [statement.value]
      }
    }
  }
}

data "aws_iam_policy_document" "remote_runtime_assume_role" {
  count = local.remote_ec2_k3d_enabled ? 1 : 0

  statement {
    sid     = "TrustedEC2Service"
    effect  = "Allow"
    actions = ["sts:AssumeRole"]

    principals {
      type        = "Service"
      identifiers = ["ec2.amazonaws.com"]
    }
  }
}

resource "aws_iam_role" "remote_runtime" {
  count = local.remote_ec2_k3d_enabled ? 1 : 0

  name                 = local.remote_runtime_role_name
  description          = "cp9 ${local.session_name} remote EC2 k3d runtime role"
  assume_role_policy   = data.aws_iam_policy_document.remote_runtime_assume_role[0].json
  permissions_boundary = local.developer_boundary_policy_arn

  tags = merge(local.common_tags, {
    Name = local.remote_runtime_role_name
  })
}

resource "aws_iam_role_policy_attachment" "remote_runtime_power" {
  count = local.remote_ec2_k3d_enabled ? 1 : 0

  role       = aws_iam_role.remote_runtime[0].name
  policy_arn = local.developer_power_policy_arn
}

resource "aws_iam_instance_profile" "remote_runtime" {
  count = local.remote_ec2_k3d_enabled ? 1 : 0

  name = local.remote_runtime_profile_name
  role = aws_iam_role.remote_runtime[0].name

  tags = merge(local.common_tags, {
    Name = local.remote_runtime_profile_name
  })
}

data "aws_iam_policy_document" "workload" {
  count = local.iam_enabled ? 1 : 0

  dynamic "statement" {
    for_each = length(local.dataplane_assume_role_arns) > 0 ? [1] : []

    content {
      sid       = "AssumeDataplaneRoles"
      effect    = "Allow"
      actions   = ["sts:AssumeRole", "sts:TagSession"]
      resources = local.dataplane_assume_role_arns
    }
  }

  dynamic "statement" {
    for_each = local.billing_sqs_arn == "" ? [] : [local.billing_sqs_arn]

    content {
      sid       = "SendBillingMessages"
      effect    = "Allow"
      actions   = ["sqs:SendMessage"]
      resources = [statement.value]
    }
  }

  dynamic "statement" {
    for_each = local.route53_enabled ? [1] : []

    content {
      sid    = "PremiumRoute53Operations"
      effect = "Allow"
      actions = [
        "route53:AssociateVPCWithHostedZone",
        "route53:ChangeResourceRecordSets",
        "route53:GetChange",
        "route53:GetHostedZone",
        "route53:ListResourceRecordSets",
      ]
      resources = ["*"]
    }
  }
}

data "aws_iam_policy_document" "workload_dedicated_core_a" {
  count = local.iam_enabled && local.dedicated_crossplane_enabled ? 1 : 0

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid    = "DedicatedCrossplaneEC2NetworkLifecycle"
      effect = "Allow"
      actions = [
        "ec2:AssociateRouteTable",
        "ec2:AllocateAddress",
        "ec2:AttachInternetGateway",
        "ec2:AuthorizeSecurityGroupIngress",
        "ec2:CreateInternetGateway",
        "ec2:CreateNatGateway",
        "ec2:CreateRoute",
        "ec2:CreateRouteTable",
        "ec2:CreateSecurityGroup",
        "ec2:CreateSubnet",
        "ec2:CreateTags",
        "ec2:CreateVpc",
        "ec2:DeleteInternetGateway",
        "ec2:DeleteNatGateway",
        "ec2:DeleteRoute",
        "ec2:DeleteRouteTable",
        "ec2:DeleteSecurityGroup",
        "ec2:DeleteSubnet",
        "ec2:DeleteVpc",
        "ec2:DescribeAvailabilityZones",
        "ec2:DescribeAddresses",
        "ec2:DescribeInternetGateways",
        "ec2:DescribeLaunchTemplates",
        "ec2:DescribeLaunchTemplateVersions",
        "ec2:DescribeNatGateways",
        "ec2:DescribeRouteTables",
        "ec2:DescribeSecurityGroups",
        "ec2:DescribeSubnets",
        "ec2:DescribeVpcEndpoints",
        "ec2:DescribeVpcAttribute",
        "ec2:DescribeVpcs",
        "ec2:DetachInternetGateway",
        "ec2:DisassociateRouteTable",
        "ec2:ModifySubnetAttribute",
        "ec2:ModifyVpcAttribute",
        "ec2:ReleaseAddress",
        "ec2:RevokeSecurityGroupIngress",
      ]
      resources = ["*"]
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid    = "DedicatedCrossplaneEKSLifecycle"
      effect = "Allow"
      actions = [
        "eks:CreateAddon",
        "eks:CreateCluster",
        "eks:CreateNodegroup",
        "eks:DeleteAddon",
        "eks:DeleteCluster",
        "eks:DeleteNodegroup",
        "eks:DescribeAddon",
        "eks:DescribeCluster",
        "eks:DescribeNodegroup",
        "eks:ListAddons",
        "eks:ListClusters",
        "eks:ListNodegroups",
        "eks:TagResource",
        "eks:UntagResource",
        "eks:UpdateAddon",
      ]
      resources = ["*"]
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedCrossplaneKMSCreate"
      effect    = "Allow"
      actions   = ["kms:CreateKey"]
      resources = ["*"]

      # The community crossplane-contrib KMS provider does not stamp
      # cp9-owner/cp9-session on the key at CreateKey time (the regional
      # resourceTagger applies them asynchronously), so gate on the servicetype
      # tag it does set. Session isolation is already enforced by the
      # per-session workload role, and reaper visibility is restored once the
      # resourceTagger stamps cp9-owner/cp9-session.
      condition {
        test     = "StringEquals"
        variable = "aws:RequestTag/servicetype"
        values   = ["dedicated"]
      }

      condition {
        test     = "ForAllValues:StringEquals"
        variable = "aws:TagKeys"
        values   = local.dedicated_crossplane_request_tag_keys
      }

      condition {
        test     = "StringEquals"
        variable = "aws:RequestedRegion"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedCrossplaneKMSCreateTag"
      effect    = "Allow"
      actions   = ["kms:TagResource"]
      resources = local.dedicated_crossplane_kms_key_arns

      # CreateKey passes tags in the same call, so kms:TagResource is evaluated
      # against the community provider's request that carries servicetype (not
      # cp9-owner/cp9-session). Gate on servicetype for the same reason as
      # DedicatedCrossplaneKMSCreate above.
      condition {
        test     = "StringEquals"
        variable = "aws:RequestTag/servicetype"
        values   = ["dedicated"]
      }

      condition {
        test     = "ForAllValues:StringEquals"
        variable = "aws:TagKeys"
        values   = local.dedicated_crossplane_request_tag_keys
      }

      condition {
        test     = "StringEquals"
        variable = "aws:RequestedRegion"
        values   = [local.target_region]
      }
    }
  }

  # Dedicated public endpoints (#703): the regional stack writes per-cluster
  # records under <session>.<runtime-account zone> so the cluster API returns a
  # resolvable SQL endpoint. Writes are scoped to session-prefixed names; the
  # delegated zone is account-bootstrap infrastructure and never session-owned.
  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedPublicEndpointRoute53Change"
      effect    = "Allow"
      actions   = ["route53:ChangeResourceRecordSets"]
      resources = ["arn:aws:route53:::hostedzone/${var.cp9_dns_zone_id}"]

      condition {
        test     = "ForAllValues:StringLike"
        variable = "route53:ChangeResourceRecordSetsNormalizedRecordNames"
        values   = ["*.${local.session_name}.${var.cp9_dns_zone_name}"]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedPublicEndpointRoute53Read"
      effect    = "Allow"
      actions   = ["route53:ListResourceRecordSets"]
      resources = ["arn:aws:route53:::hostedzone/${var.cp9_dns_zone_id}"]
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedPublicEndpointRoute53ChangeStatus"
      effect    = "Allow"
      actions   = ["route53:GetChange"]
      resources = ["arn:aws:route53:::change/*"]
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid    = "DedicatedCrossplaneKMSLifecycle"
      effect = "Allow"
      actions = [
        "kms:DescribeKey",
        "kms:GetKeyPolicy",
        "kms:GetKeyRotationStatus",
        "kms:ListResourceTags",
        "kms:ScheduleKeyDeletion",
        "kms:TagResource",
        "kms:UntagResource",
      ]
      resources = local.dedicated_crossplane_kms_key_arns

      # Right after CreateKey the key only carries the community provider's tags
      # (servicetype/cluster/project/tenant); cp9-owner/cp9-session land later via
      # the resourceTagger. Gate observe/lifecycle on servicetype so the provider
      # can DescribeKey the key it just created instead of stalling until the
      # async tagger runs.
      condition {
        test     = "StringEquals"
        variable = "aws:ResourceTag/servicetype"
        values   = ["dedicated"]
      }

      condition {
        test     = "StringEquals"
        variable = "aws:RequestedRegion"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedCrossplaneAutoscalingCreate"
      effect    = "Allow"
      actions   = ["autoscaling:CreateAutoScalingGroup", "autoscaling:CreateOrUpdateTags"]
      resources = local.dedicated_crossplane_autoscaling_group_arns

      # The node-group ASGs are stamped with dynamic EKS / cluster-autoscaler
      # tags (eks:cluster-name, k8s.io/cluster-autoscaler/<eks-cluster>, ...) and
      # carry neither cp9-owner/cp9-session nor an enumerable tag-key set, so a
      # RequestTag / ForAllValues aws:TagKeys guard cannot be satisfied. Scope by
      # region; session isolation comes from the per-session workload role.
      condition {
        test     = "StringEquals"
        variable = "aws:RequestedRegion"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid    = "DedicatedCrossplaneAutoscalingLifecycle"
      effect = "Allow"
      actions = [
        "autoscaling:CompleteLifecycleAction",
        "autoscaling:DeleteAutoScalingGroup",
        "autoscaling:DeleteLifecycleHook",
        "autoscaling:DeleteTags",
        "autoscaling:PutLifecycleHook",
        "autoscaling:SetDesiredCapacity",
        "autoscaling:SetInstanceProtection",
        "autoscaling:TerminateInstanceInAutoScalingGroup",
        "autoscaling:UpdateAutoScalingGroup",
      ]
      resources = local.dedicated_crossplane_autoscaling_group_arns

      # The node ASGs carry dynamic EKS/cluster-autoscaler tags, not
      # cp9-owner/cp9-session, so lifecycle ops (SetInstanceProtection,
      # UpdateAutoScalingGroup, ...) cannot use an autoscaling:ResourceTag guard.
      # Scope by region + ASG ARN; the per-session workload role isolates.
      condition {
        test     = "StringEquals"
        variable = "aws:RequestedRegion"
        values   = [local.target_region]
      }
    }
  }
}

data "aws_iam_policy_document" "workload_dedicated_core_b" {
  count = local.iam_enabled && local.dedicated_crossplane_enabled ? 1 : 0

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid    = "DedicatedCrossplaneLaunchConfigurationLifecycle"
      effect = "Allow"
      actions = [
        "autoscaling:CreateLaunchConfiguration",
        "autoscaling:DeleteLaunchConfiguration",
      ]
      resources = local.dedicated_crossplane_launch_configuration_arns

      condition {
        test     = "StringEquals"
        variable = "aws:RequestedRegion"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedCrossplaneAutoscalingReadOnly"
      effect    = "Allow"
      actions   = ["autoscaling:Describe*"]
      resources = ["*"]

      condition {
        test     = "StringEquals"
        variable = "aws:RequestedRegion"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedCrossplaneEC2Runtime"
      effect    = "Allow"
      actions   = ["ec2:*"]
      resources = ["*"]
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid    = "DedicatedCrossplaneELBV2Lifecycle"
      effect = "Allow"
      actions = [
        "elasticloadbalancing:AddTags",
        "elasticloadbalancing:CreateListener",
        "elasticloadbalancing:CreateLoadBalancer",
        "elasticloadbalancing:CreateTargetGroup",
        "elasticloadbalancing:DeleteListener",
        "elasticloadbalancing:DeleteLoadBalancer",
        "elasticloadbalancing:DeleteTargetGroup",
        "elasticloadbalancing:DescribeListeners",
        "elasticloadbalancing:DescribeLoadBalancerAttributes",
        "elasticloadbalancing:DescribeLoadBalancers",
        "elasticloadbalancing:DescribeTags",
        "elasticloadbalancing:DescribeTargetGroups",
        "elasticloadbalancing:ModifyLoadBalancerAttributes",
        "elasticloadbalancing:ModifyTargetGroupAttributes",
        "elasticloadbalancing:RemoveTags",
      ]
      resources = ["*"]
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid    = "DedicatedCrossplaneIAMLifecycle"
      effect = "Allow"
      actions = [
        "iam:AttachRolePolicy",
        "iam:AddRoleToInstanceProfile",
        "iam:DeleteInstanceProfile",
        "iam:CreatePolicy",
        "iam:CreateRole",
        "iam:DeletePolicy",
        "iam:DeletePolicyVersion",
        "iam:DeleteRole",
        "iam:DeleteRolePolicy",
        "iam:DetachRolePolicy",
        "iam:GetPolicy",
        "iam:GetPolicyVersion",
        "iam:GetInstanceProfile",
        "iam:GetRole",
        "iam:GetRolePolicy",
        "iam:ListAccountAliases",
        "iam:ListAttachedRolePolicies",
        "iam:ListEntitiesForPolicy",
        "iam:ListInstanceProfilesForRole",
        "iam:ListPolicyTags",
        "iam:ListPolicyVersions",
        "iam:ListRolePolicies",
        "iam:ListRoleTags",
        "iam:PassRole",
        "iam:PutRolePermissionsBoundary",
        "iam:PutRolePolicy",
        "iam:RemoveRoleFromInstanceProfile",
        "iam:TagPolicy",
        "iam:TagRole",
        "iam:UntagPolicy",
        "iam:UntagRole",
        "iam:UpdateAssumeRolePolicy",
      ]
      resources = concat(
        local.dedicated_crossplane_iam_instance_profile_arns,
        local.dedicated_crossplane_ticdc_instance_profile_arns,
        local.dedicated_crossplane_iam_role_arns,
        local.dedicated_crossplane_ticdc_role_arns,
        local.dedicated_crossplane_iam_policy_arns,
        local.dedicated_crossplane_ticdc_policy_arns,
      )
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedCrossplaneIAMRuntime"
      effect    = "Allow"
      actions   = ["iam:*"]
      resources = ["*"]
    }
  }
}


data "aws_iam_policy_document" "workload_dedicated_scoped" {
  count = local.iam_enabled && local.dedicated_crossplane_enabled ? 1 : 0

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedCrossplaneLaunchTemplateCreate"
      effect    = "Allow"
      actions   = ["ec2:CreateLaunchTemplate"]
      resources = local.dedicated_crossplane_launch_template_arns

      # The community provider creates the node launch template without any tags
      # (no cp9-owner/cp9-session), so the RequestTag / aws:TagKeys guard cannot
      # be satisfied. Scope by region + the launch-template resource ARN; session
      # isolation comes from the per-session workload role.
      condition {
        test     = "StringEquals"
        variable = "ec2:Region"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid    = "DedicatedCrossplaneLaunchTemplateLifecycle"
      effect = "Allow"
      actions = [
        "ec2:DeleteLaunchTemplate",
        "ec2:ModifyLaunchTemplate",
      ]
      resources = local.dedicated_crossplane_launch_template_arns

      # The node launch templates are created untagged, so lifecycle ops cannot
      # use an ec2:ResourceTag guard. Scope by region + launch-template ARN; the
      # per-session workload role isolates.
      condition {
        test     = "StringEquals"
        variable = "ec2:Region"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedCrossplaneS3VPCEndpointCreateEndpoint"
      effect    = "Allow"
      actions   = ["ec2:CreateVpcEndpoint"]
      resources = local.dedicated_crossplane_vpc_endpoint_arns

      condition {
        test     = "StringEquals"
        variable = "aws:RequestTag/cp9-owner"
        values   = [local.owner]
      }

      condition {
        test     = "StringEquals"
        variable = "aws:RequestTag/cp9-session"
        values   = [local.session_name]
      }

      condition {
        test     = "ForAllValues:StringEquals"
        variable = "aws:TagKeys"
        values   = local.dedicated_crossplane_request_tag_keys
      }

      condition {
        test     = "StringEquals"
        variable = "ec2:VpceServiceName"
        values   = [local.dedicated_crossplane_s3_vpc_endpoint_service_name]
      }

      condition {
        test     = "StringEquals"
        variable = "ec2:Region"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid     = "DedicatedCrossplaneS3VPCEndpointCreateNetwork"
      effect  = "Allow"
      actions = ["ec2:CreateVpcEndpoint"]
      resources = concat(
        local.dedicated_crossplane_vpc_arns,
        local.dedicated_crossplane_route_table_arns,
      )

      condition {
        test     = "StringEquals"
        variable = "ec2:ResourceTag/cp9-owner"
        values   = [local.owner]
      }

      condition {
        test     = "StringEquals"
        variable = "ec2:ResourceTag/cp9-session"
        values   = [local.session_name]
      }

      condition {
        test     = "StringEqualsIfExists"
        variable = "ec2:VpceServiceName"
        values   = [local.dedicated_crossplane_s3_vpc_endpoint_service_name]
      }

      condition {
        test     = "StringEquals"
        variable = "ec2:Region"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid       = "DedicatedCrossplaneS3VPCEndpointDelete"
      effect    = "Allow"
      actions   = ["ec2:DeleteVpcEndpoints"]
      resources = local.dedicated_crossplane_vpc_endpoint_arns

      condition {
        test     = "StringEquals"
        variable = "ec2:ResourceTag/cp9-owner"
        values   = [local.owner]
      }

      condition {
        test     = "StringEquals"
        variable = "ec2:ResourceTag/cp9-session"
        values   = [local.session_name]
      }

      condition {
        test     = "StringEquals"
        variable = "ec2:VpceServiceName"
        values   = [local.dedicated_crossplane_s3_vpc_endpoint_service_name]
      }

      condition {
        test     = "StringEquals"
        variable = "ec2:Region"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid     = "DedicatedCrossplaneS3VPCEndpointModify"
      effect  = "Allow"
      actions = ["ec2:ModifyVpcEndpoint"]
      resources = concat(
        local.dedicated_crossplane_vpc_endpoint_arns,
        local.dedicated_crossplane_route_table_arns,
      )

      condition {
        test     = "StringEquals"
        variable = "ec2:ResourceTag/cp9-owner"
        values   = [local.owner]
      }

      condition {
        test     = "StringEquals"
        variable = "ec2:ResourceTag/cp9-session"
        values   = [local.session_name]
      }

      condition {
        test     = "StringEquals"
        variable = "ec2:Region"
        values   = [local.target_region]
      }
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid     = "DedicatedCrossplaneInstanceProfileCreate"
      effect  = "Allow"
      actions = ["iam:CreateInstanceProfile"]
      resources = concat(
        local.dedicated_crossplane_iam_instance_profile_arns,
        local.dedicated_crossplane_ticdc_instance_profile_arns,
      )

      # The community provider creates the node instance profiles without tags
      # (no cp9-owner/cp9-session), and iam:CreateInstanceProfile is a global
      # action with no region context, so a RequestTag / aws:TagKeys guard cannot
      # apply. Scope by the specific instance-profile resource ARNs; session
      # isolation comes from the per-session workload role.
    }
  }

  dynamic "statement" {
    for_each = local.dedicated_crossplane_enabled ? [1] : []

    content {
      sid     = "DedicatedCrossplaneInstanceProfileTag"
      effect  = "Allow"
      actions = ["iam:TagInstanceProfile"]
      resources = concat(
        local.dedicated_crossplane_iam_instance_profile_arns,
        local.dedicated_crossplane_ticdc_instance_profile_arns,
      )

      # Scoped by the specific instance-profile resource ARNs (the profiles carry
      # no cp9-owner/cp9-session tags and iam:TagInstanceProfile is global); the
      # per-session workload role provides session isolation.
    }
  }
}

data "aws_iam_policy_document" "dataplane_manager_assume_role" {
  count = local.premium_dataplane_role_enabled ? 1 : 0

  statement {
    sid     = "TrustedCp9DataplaneSourcePrincipals"
    effect  = "Allow"
    actions = ["sts:AssumeRole"]

    principals {
      type        = "AWS"
      identifiers = local.trusted_principal_arns
    }
  }

  statement {
    sid     = "TrustedCp9DataplaneTagSessionPrincipals"
    effect  = "Allow"
    actions = ["sts:TagSession"]

    principals {
      type        = "AWS"
      identifiers = local.trusted_principal_arns
    }
  }

  dynamic "statement" {
    for_each = local.remote_ec2_k3d_enabled ? [aws_iam_role.remote_runtime[0].arn] : []

    content {
      sid     = "TrustedRemoteRuntimePrincipal"
      effect  = "Allow"
      actions = ["sts:AssumeRole", "sts:TagSession"]

      principals {
        type        = "AWS"
        identifiers = [statement.value]
      }
    }
  }
}

data "aws_iam_policy_document" "dataplane_manager" {
  count = local.premium_dataplane_role_enabled ? 1 : 0

  statement {
    sid       = "PremiumMeteringReadWriteAssume"
    effect    = "Allow"
    actions   = ["sts:AssumeRole"]
    resources = [local.metering_read_write_role_arn]
  }

  statement {
    sid    = "PremiumDataplaneEC2Lifecycle"
    effect = "Allow"
    actions = [
      "ec2:AcceptVpcPeeringConnection",
      "ec2:AllocateAddress",
      "ec2:AssociateAddress",
      "ec2:AssociateRouteTable",
      "ec2:AttachInternetGateway",
      "ec2:AuthorizeSecurityGroupEgress",
      "ec2:AuthorizeSecurityGroupIngress",
      "ec2:CreateInternetGateway",
      "ec2:CreateLaunchTemplate",
      "ec2:CreateNatGateway",
      "ec2:CreateNetworkInterface",
      "ec2:CreateRoute",
      "ec2:CreateRouteTable",
      "ec2:CreateSecurityGroup",
      "ec2:CreateSubnet",
      "ec2:CreateTags",
      "ec2:CreateVpc",
      "ec2:CreateVpcEndpoint",
      "ec2:CreateVpcEndpointServiceConfiguration",
      "ec2:CreateVpcPeeringConnection",
      "ec2:DeleteInternetGateway",
      "ec2:DeleteLaunchTemplate",
      "ec2:DeleteNatGateway",
      "ec2:DeleteNetworkInterface",
      "ec2:DeleteRoute",
      "ec2:DeleteRouteTable",
      "ec2:DeleteSecurityGroup",
      "ec2:DeleteSubnet",
      "ec2:DeleteTags",
      "ec2:DeleteVpc",
      "ec2:DeleteVpcEndpoints",
      "ec2:DeleteVpcEndpointServiceConfigurations",
      "ec2:DeleteVolume",
      "ec2:Describe*",
      "ec2:DetachInternetGateway",
      "ec2:DisassociateAddress",
      "ec2:DisassociateRouteTable",
      "ec2:ModifyLaunchTemplate",
      "ec2:ModifySubnetAttribute",
      "ec2:ModifyVpcAttribute",
      "ec2:ModifyVpcEndpoint",
      "ec2:ModifyVpcEndpointServiceConfiguration",
      "ec2:ReleaseAddress",
      "ec2:RevokeSecurityGroupEgress",
      "ec2:RevokeSecurityGroupIngress",
      "ec2:RunInstances",
      "ec2:TerminateInstances",
    ]
    resources = ["*"]
  }

  statement {
    sid    = "PremiumDataplaneEKSLifecycle"
    effect = "Allow"
    actions = [
      "eks:AssociateAccessPolicy",
      "eks:CreateAccessEntry",
      "eks:CreateAddon",
      "eks:CreateCluster",
      "eks:CreateNodegroup",
      "eks:CreatePodIdentityAssociation",
      "eks:DeleteAccessEntry",
      "eks:DeleteAddon",
      "eks:DeleteCluster",
      "eks:DeleteNodegroup",
      "eks:DeletePodIdentityAssociation",
      "eks:Describe*",
      "eks:DisassociateAccessPolicy",
      "eks:List*",
      "eks:TagResource",
      "eks:UntagResource",
      "eks:UpdateAccessEntry",
      "eks:UpdateAddon",
      "eks:UpdateClusterConfig",
      "eks:UpdateClusterVersion",
      "eks:UpdateNodegroupConfig",
      "eks:UpdateNodegroupVersion",
    ]
    resources = ["*"]
  }

  statement {
    sid    = "PremiumDataplaneMSKVpcConnectionLifecycle"
    effect = "Allow"
    actions = [
      "kafka:CreateVpcConnection",
      "kafka:DeleteVpcConnection",
      "kafka:DescribeCluster",
      "kafka:DescribeVpcConnection",
      "kafka:GetBootstrapBrokers",
      "kafka:ListClientVpcConnections",
    ]
    resources = ["*"]
  }

  statement {
    sid    = "PremiumDataplaneLoadBalancerLifecycle"
    effect = "Allow"
    actions = [
      "elasticloadbalancing:AddTags",
      "elasticloadbalancing:CreateListener",
      "elasticloadbalancing:CreateLoadBalancer",
      "elasticloadbalancing:CreateTargetGroup",
      "elasticloadbalancing:DeleteListener",
      "elasticloadbalancing:DeleteLoadBalancer",
      "elasticloadbalancing:DeleteTargetGroup",
      "elasticloadbalancing:Describe*",
      "elasticloadbalancing:Modify*",
      "elasticloadbalancing:RemoveTags",
      "elasticloadbalancing:SetIpAddressType",
      "elasticloadbalancing:SetSecurityGroups",
      "elasticloadbalancing:SetSubnets",
    ]
    resources = ["*"]
  }

  statement {
    sid    = "PremiumDataplaneAutoscalingLifecycle"
    effect = "Allow"
    actions = [
      "autoscaling:CreateAutoScalingGroup",
      "autoscaling:CreateLaunchConfiguration",
      "autoscaling:CreateOrUpdateTags",
      "autoscaling:DeleteAutoScalingGroup",
      "autoscaling:DeleteLaunchConfiguration",
      "autoscaling:DeleteTags",
      "autoscaling:Describe*",
      "autoscaling:SetDesiredCapacity",
      "autoscaling:UpdateAutoScalingGroup",
    ]
    resources = ["*"]
  }

  statement {
    sid    = "PremiumDataplaneIAMLifecycle"
    effect = "Allow"
    actions = [
      "iam:AddRoleToInstanceProfile",
      "iam:AttachRolePolicy",
      "iam:CreateInstanceProfile",
      "iam:CreatePolicy",
      "iam:CreatePolicyVersion",
      "iam:CreateRole",
      "iam:DeleteOpenIDConnectProvider",
      "iam:DeleteInstanceProfile",
      "iam:DeletePolicy",
      "iam:DeletePolicyVersion",
      "iam:DeleteRole",
      "iam:DeleteRolePolicy",
      "iam:DetachRolePolicy",
      "iam:GetInstanceProfile",
      "iam:GetOpenIDConnectProvider",
      "iam:GetPolicy",
      "iam:GetPolicyVersion",
      "iam:GetRole",
      "iam:GetRolePolicy",
      "iam:ListAttachedRolePolicies",
      "iam:ListEntitiesForPolicy",
      "iam:ListInstanceProfilesForRole",
      "iam:ListOpenIDConnectProviderTags",
      "iam:ListOpenIDConnectProviders",
      "iam:ListPolicyTags",
      "iam:ListPolicyVersions",
      "iam:ListRolePolicies",
      "iam:ListRoleTags",
      "iam:PassRole",
      "iam:PutRolePermissionsBoundary",
      "iam:PutRolePolicy",
      "iam:RemoveRoleFromInstanceProfile",
      "iam:SetDefaultPolicyVersion",
      "iam:TagOpenIDConnectProvider",
      "iam:TagInstanceProfile",
      "iam:TagPolicy",
      "iam:TagRole",
      "iam:UntagOpenIDConnectProvider",
      "iam:UntagPolicy",
      "iam:UntagRole",
      "iam:UpdateAssumeRolePolicy",
    ]
    resources = [
      "arn:aws:iam::${local.target_account}:instance-profile/cp9-*",
      "arn:aws:iam::${local.target_account}:instance-profile/tidbcloud-*",
      "arn:aws:iam::${local.target_account}:oidc-provider/*",
      "arn:aws:iam::${local.target_account}:policy/cp9-*",
      "arn:aws:iam::${local.target_account}:policy/tidbcloud-*",
      local.premium_eks_nodegroup_service_role_name_arn,
      local.premium_eks_nodegroup_service_role_arn,
      "arn:aws:iam::${local.target_account}:role/aws-service-role/*",
      "arn:aws:iam::${local.target_account}:role/cp9-*",
      "arn:aws:iam::${local.target_account}:role/tidbcloud-*",
    ]
  }

  statement {
    sid       = "PremiumDataplaneOIDCProviderCreate"
    effect    = "Allow"
    actions   = ["iam:CreateOpenIDConnectProvider"]
    resources = ["*"]
  }

  statement {
    sid       = "PremiumDataplaneServiceLinkedRoleCreate"
    effect    = "Allow"
    actions   = ["iam:CreateServiceLinkedRole"]
    resources = ["*"]

    condition {
      test     = "StringLike"
      variable = "iam:AWSServiceName"
      values = [
        "autoscaling.amazonaws.com",
        "elasticloadbalancing.amazonaws.com",
        "eks.amazonaws.com",
        "eks-nodegroup.amazonaws.com",
        "spot.amazonaws.com",
      ]
    }
  }

  statement {
    sid       = "PremiumDataplaneAccountAliasRead"
    effect    = "Allow"
    actions   = ["iam:ListAccountAliases"]
    resources = ["*"]
  }

  statement {
    sid    = "PremiumDataplaneSharedServices"
    effect = "Allow"
    actions = [
      "cloudformation:*",
      "kms:CreateGrant",
      "kms:DescribeKey",
      "kms:ListAliases",
      "kms:ListGrants",
      "kms:RevokeGrant",
      "route53:AssociateVPCWithHostedZone",
      "route53:ChangeResourceRecordSets",
      "route53:GetChange",
      "route53:GetHostedZone",
      "route53:ListHostedZones",
      "route53:ListResourceRecordSets",
      "route53:ListTagsForResource",
      "s3:*",
      "secretsmanager:*",
      "tag:GetResources",
      "tag:GetTagKeys",
      "tag:GetTagValues",
    ]
    resources = ["*"]
  }
}

resource "aws_iam_role" "workload" {
  count = local.iam_enabled ? 1 : 0

  name                 = local.workload_role_name
  description          = "cp9 ${local.session_name} workload role for Premium Reserved"
  assume_role_policy   = data.aws_iam_policy_document.workload_assume_role[0].json
  permissions_boundary = local.developer_boundary_policy_arn

  tags = merge(local.common_tags, {
    Name = local.workload_role_name
  })
}

resource "aws_iam_policy" "workload" {
  count = local.iam_enabled ? 1 : 0

  name        = local.workload_policy_name
  description = "cp9 ${local.session_name} workload policy for Premium Reserved"
  policy      = data.aws_iam_policy_document.workload[0].json

  tags = merge(local.common_tags, {
    Name = local.workload_policy_name
  })
}

resource "aws_iam_role_policy_attachment" "workload" {
  count = local.iam_enabled ? 1 : 0

  role       = aws_iam_role.workload[0].name
  policy_arn = aws_iam_policy.workload[0].arn
}

resource "aws_iam_role_policy_attachment" "workload_remote_power" {
  count = local.iam_enabled && local.remote_ec2_k3d_enabled ? 1 : 0

  role       = aws_iam_role.workload[0].name
  policy_arn = local.developer_power_policy_arn
}

resource "aws_iam_policy" "workload_dedicated_scoped" {
  count = local.iam_enabled && local.dedicated_crossplane_enabled ? 1 : 0

  name        = local.dedicated_scoped_policy_name
  description = "cp9 ${local.session_name} Dedicated scoped workload policy"
  policy      = data.aws_iam_policy_document.workload_dedicated_scoped[0].json

  tags = merge(local.common_tags, {
    Name = local.dedicated_scoped_policy_name
  })
}

resource "aws_iam_role_policy_attachment" "workload_dedicated_scoped" {
  count = local.iam_enabled && local.dedicated_crossplane_enabled ? 1 : 0

  role       = aws_iam_role.workload[0].name
  policy_arn = aws_iam_policy.workload_dedicated_scoped[0].arn
}

resource "aws_iam_policy" "workload_dedicated_core_a" {
  count = local.iam_enabled && local.dedicated_crossplane_enabled ? 1 : 0

  name        = "${local.workload_policy_name}-dedicated-core-a"
  description = "cp9 ${local.session_name} Dedicated core workload policy A"
  policy      = data.aws_iam_policy_document.workload_dedicated_core_a[0].json

  tags = merge(local.common_tags, {
    Name = "${local.workload_policy_name}-dedicated-core-a"
  })
}

resource "aws_iam_role_policy_attachment" "workload_dedicated_core_a" {
  count = local.iam_enabled && local.dedicated_crossplane_enabled ? 1 : 0

  role       = aws_iam_role.workload[0].name
  policy_arn = aws_iam_policy.workload_dedicated_core_a[0].arn
}

resource "aws_iam_policy" "workload_dedicated_core_b" {
  count = local.iam_enabled && local.dedicated_crossplane_enabled ? 1 : 0

  name        = "${local.workload_policy_name}-dedicated-core-b"
  description = "cp9 ${local.session_name} Dedicated core workload policy B"
  policy      = data.aws_iam_policy_document.workload_dedicated_core_b[0].json

  tags = merge(local.common_tags, {
    Name = "${local.workload_policy_name}-dedicated-core-b"
  })
}

resource "aws_iam_role_policy_attachment" "workload_dedicated_core_b" {
  count = local.iam_enabled && local.dedicated_crossplane_enabled ? 1 : 0

  role       = aws_iam_role.workload[0].name
  policy_arn = aws_iam_policy.workload_dedicated_core_b[0].arn
}

resource "aws_iam_role" "dataplane_manager" {
  count = local.premium_dataplane_role_enabled ? 1 : 0

  name                 = local.dataplane_manager_role_name
  description          = "cp9 ${local.session_name} dataplane manager role for Premium"
  assume_role_policy   = data.aws_iam_policy_document.dataplane_manager_assume_role[0].json
  permissions_boundary = local.remote_ec2_k3d_enabled ? local.developer_boundary_policy_arn : null

  tags = merge(local.common_tags, {
    Name = local.dataplane_manager_role_name
  })
}

resource "aws_iam_policy" "dataplane_manager" {
  count = local.premium_dataplane_role_enabled ? 1 : 0

  name        = local.dataplane_manager_policy_name
  description = "cp9 ${local.session_name} dataplane manager policy for Premium"
  policy      = data.aws_iam_policy_document.dataplane_manager[0].json

  tags = merge(local.common_tags, {
    Name = local.dataplane_manager_policy_name
  })
}

resource "aws_iam_role_policy_attachment" "dataplane_manager" {
  count = local.premium_dataplane_role_enabled ? 1 : 0

  role       = aws_iam_role.dataplane_manager[0].name
  policy_arn = aws_iam_policy.dataplane_manager[0].arn
}

resource "aws_iam_role_policy_attachment" "dataplane_manager_remote_power" {
  count = local.premium_dataplane_role_enabled && local.remote_ec2_k3d_enabled ? 1 : 0

  role       = aws_iam_role.dataplane_manager[0].name
  policy_arn = local.developer_power_policy_arn
}