resource "aws_dynamodb_table" "starter_infra_cluster" {
  count = local.starter_infra_cluster_enabled ? 1 : 0

  name             = local.starter_infra_cluster_table_name
  billing_mode     = "PAY_PER_REQUEST"
  hash_key         = "namespace"
  range_key        = "name"
  stream_enabled   = true
  stream_view_type = "NEW_AND_OLD_IMAGES"

  attribute {
    name = "namespace"
    type = "S"
  }

  attribute {
    name = "name"
    type = "S"
  }

  attribute {
    name = "status.keyspace_id"
    type = "N"
  }

  attribute {
    name = "status.keyspace_token"
    type = "S"
  }

  attribute {
    name = "status.zone_redundant_enabled"
    type = "S"
  }

  global_secondary_index {
    name            = "name-index"
    hash_key        = "name"
    projection_type = "KEYS_ONLY"
  }

  global_secondary_index {
    name            = "status.keyspace_id-index"
    hash_key        = "status.keyspace_id"
    projection_type = "KEYS_ONLY"
  }

  global_secondary_index {
    name            = "status.keyspace_token-index"
    hash_key        = "status.keyspace_token"
    projection_type = "KEYS_ONLY"
  }

  global_secondary_index {
    name            = "status.zone_redundant_enabled-index"
    hash_key        = "status.zone_redundant_enabled"
    projection_type = "KEYS_ONLY"
  }

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

resource "aws_sqs_queue" "starter_cluster" {
  count = local.starter_infra_cluster_enabled ? 1 : 0

  name = "cp9-${local.session_name}-starter-cluster"

  tags = merge(local.common_tags, {
    Name = "cp9-${local.session_name}-starter-cluster"
  })
}

data "archive_file" "starter_infra_cluster_stream_to_sqs" {
  count = local.starter_infra_cluster_enabled ? 1 : 0

  type        = "zip"
  source_file = "${path.module}/scripts/starter_infra_cluster_stream_to_sqs.py"
  output_path = "${path.module}/.terraform/cp9-starter-infra-cluster-stream-to-sqs.zip"
}

data "aws_iam_policy_document" "starter_infra_cluster_stream_to_sqs_assume" {
  count = local.starter_infra_cluster_enabled ? 1 : 0

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

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

resource "aws_iam_role" "starter_infra_cluster_stream_to_sqs" {
  count = local.starter_infra_cluster_enabled ? 1 : 0

  name                 = local.starter_infra_cluster_event_role_name
  assume_role_policy   = data.aws_iam_policy_document.starter_infra_cluster_stream_to_sqs_assume[0].json
  permissions_boundary = local.developer_boundary_policy_arn

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

data "aws_iam_policy_document" "starter_infra_cluster_stream_to_sqs" {
  count = local.starter_infra_cluster_enabled ? 1 : 0

  statement {
    effect = "Allow"
    actions = [
      "dynamodb:DescribeStream",
      "dynamodb:GetRecords",
      "dynamodb:GetShardIterator",
      "dynamodb:ListStreams",
    ]
    resources = [aws_dynamodb_table.starter_infra_cluster[0].stream_arn]
  }

  statement {
    effect    = "Allow"
    actions   = ["sqs:GetQueueUrl", "sqs:SendMessage"]
    resources = [aws_sqs_queue.starter_cluster[0].arn]
  }

  statement {
    effect = "Allow"
    actions = [
      "logs:CreateLogGroup",
      "logs:CreateLogStream",
      "logs:PutLogEvents",
    ]
    resources = [
      "arn:aws:logs:${local.target_region}:${local.target_account}:log-group:/aws/lambda/${local.starter_infra_cluster_event_function_name}",
      "arn:aws:logs:${local.target_region}:${local.target_account}:log-group:/aws/lambda/${local.starter_infra_cluster_event_function_name}:*",
    ]
  }

}

resource "aws_iam_role_policy" "starter_infra_cluster_stream_to_sqs" {
  count = local.starter_infra_cluster_enabled ? 1 : 0

  name   = "cp9-starter-cluster-events"
  role   = aws_iam_role.starter_infra_cluster_stream_to_sqs[0].id
  policy = data.aws_iam_policy_document.starter_infra_cluster_stream_to_sqs[0].json
}

resource "aws_lambda_function" "starter_infra_cluster_stream_to_sqs" {
  count = local.starter_infra_cluster_enabled ? 1 : 0

  function_name    = local.starter_infra_cluster_event_function_name
  role             = aws_iam_role.starter_infra_cluster_stream_to_sqs[0].arn
  handler          = "starter_infra_cluster_stream_to_sqs.handler"
  runtime          = "python3.12"
  filename         = data.archive_file.starter_infra_cluster_stream_to_sqs[0].output_path
  source_code_hash = data.archive_file.starter_infra_cluster_stream_to_sqs[0].output_base64sha256
  timeout          = 10

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

  depends_on = [aws_iam_role_policy.starter_infra_cluster_stream_to_sqs]
}

resource "aws_lambda_event_source_mapping" "starter_infra_cluster_stream_to_sqs" {
  count = local.starter_infra_cluster_enabled ? 1 : 0

  event_source_arn  = aws_dynamodb_table.starter_infra_cluster[0].stream_arn
  function_name     = aws_lambda_function.starter_infra_cluster_stream_to_sqs[0].arn
  starting_position = "LATEST"
  batch_size        = 10
}

data "aws_iam_policy_document" "starter_remote_node_runtime" {
  count = local.starter_remote_eks_enabled ? 1 : 0

  statement {
    sid    = "StarterInfraClusterTableRuntime"
    effect = "Allow"
    actions = [
      "dynamodb:DeleteItem",
      "dynamodb:DescribeTable",
      "dynamodb:GetItem",
      "dynamodb:PutItem",
      "dynamodb:Query",
      "dynamodb:Scan",
      "dynamodb:UpdateItem",
    ]
    resources = [
      aws_dynamodb_table.starter_infra_cluster[0].arn,
      "${aws_dynamodb_table.starter_infra_cluster[0].arn}/index/*",
    ]
  }

  statement {
    sid    = "StarterClusterQueueRuntime"
    effect = "Allow"
    actions = [
      "sqs:DeleteMessage",
      "sqs:GetQueueAttributes",
      "sqs:GetQueueUrl",
      "sqs:ReceiveMessage",
      "sqs:SendMessage",
    ]
    resources = [
      aws_sqs_queue.starter_cluster[0].arn,
    ]
  }

  statement {
    sid    = "StarterServerlessStorageList"
    effect = "Allow"
    actions = [
      "s3:ListBucket",
      "s3:ListBucketMultipartUploads",
    ]
    resources = [
      aws_s3_bucket.starter_dm_storage[0].arn,
      aws_s3_bucket.starter_stashing_area[0].arn,
      aws_s3_bucket.starter_serverless_storage[0].arn,
    ]
  }

  statement {
    sid    = "StarterServerlessStorageObjects"
    effect = "Allow"
    actions = [
      "s3:AbortMultipartUpload",
      "s3:DeleteObject*",
      "s3:GetObject*",
      "s3:ListMultipartUploadParts",
      "s3:PutObject*",
    ]
    resources = [
      "${aws_s3_bucket.starter_dm_storage[0].arn}/*",
      "${aws_s3_bucket.starter_stashing_area[0].arn}/*",
      "${aws_s3_bucket.starter_serverless_storage[0].arn}/*",
    ]
  }

  statement {
    sid    = "StarterPrivateLinkHostedZoneDiscovery"
    effect = "Allow"
    actions = [
      "route53:AssociateVPCWithHostedZone",
      "route53:ChangeResourceRecordSets",
      "route53:GetChange",
      "route53:GetHostedZone",
      "route53:ListHostedZones",
      "route53:ListHostedZonesByName",
      "route53:ListResourceRecordSets",
      "route53:ListTagsForResource",
    ]
    resources = ["*"]
  }

  statement {
    sid    = "StarterPrivateLinkClientVpcConnection"
    effect = "Allow"
    actions = [
      "ec2:CreateTags",
      "ec2:CreateVpcEndpoint",
      "ec2:DeleteVpcEndpoints",
      "ec2:DescribeAvailabilityZones",
      "ec2:DescribeSecurityGroups",
      "ec2:DescribeSubnets",
      "ec2:DescribeVpcAttribute",
      "ec2:DescribeVpcEndpoints",
      "ec2:DescribeVpcEndpointServices",
      "ec2:DescribeVpcs",
      "eks:DescribeCluster",
      "eks:DescribeNodegroup",
      "eks:ListNodegroups",
      "kafka:CreateVpcConnection",
      "kafka:DeleteVpcConnection",
      "kafka:DescribeCluster",
      "kafka:DescribeClusterV2",
      "kafka:DescribeVpcConnection",
      "kafka:GetBootstrapBrokers",
      "kafka:ListClientVpcConnections",
    ]
    resources = ["*"]
  }

  statement {
    sid    = "StarterMetaDatabaseSecret"
    effect = "Allow"
    actions = [
      "secretsmanager:GetSecretValue",
    ]
    resources = [
      aws_secretsmanager_secret.starter_meta_database["audit"].arn,
      aws_secretsmanager_secret.starter_meta_database["infra"].arn,
    ]
  }

  statement {
    sid    = "StarterAuditLogKey"
    effect = "Allow"
    actions = [
      "kms:Decrypt",
      "kms:DescribeKey",
    ]
    resources = [
      aws_kms_key.starter_audit_log[0].arn,
    ]
  }

  statement {
    sid    = "StarterCSEMasterKey"
    effect = "Allow"
    actions = [
      "kms:Decrypt",
      "kms:DescribeKey",
    ]
    resources = [
      aws_kms_key.starter_cse[0].arn,
    ]
  }
}

resource "aws_iam_role_policy" "starter_remote_node_runtime" {
  count = local.starter_remote_eks_enabled ? 1 : 0

  name   = "cp9-starter-runtime"
  role   = aws_iam_role.starter_remote_node[0].id
  policy = data.aws_iam_policy_document.starter_remote_node_runtime[0].json
}