# Create a new dynamo table
resource "aws_dynamodb_table" "dynamodb_table" {
  # checkov:skip=CKV_AWS_119:DynamoDB is encrypted at rest using an AWS-managed KMS key by default, which is sufficient at this moment. We will add KMS CMK support when we have an established workflow on how different services can interact using a single CMK.
  # checkov:skip=CKV2_AWS_16:This check does not work correctly in conjunction with count. Autoscaling is correctly configured when using provisioned mode.

  lifecycle {
    ignore_changes = [
      read_capacity, write_capacity,
    ]
    precondition {
      condition = var.use_on_demand || alltrue([
        for gsi in var.global_secondary_index : (
          coalesce(gsi["read_capacity"], 0) > 0 && coalesce(gsi["write_capacity"], 0) > 0
        )
      ])
      error_message = "When using provisioned mode all global secondary indexes must have a non-zero read_capacity and write_capacity defined"
    }
  }
  count            = var.range_key_enabled ? 0 : 1
  name             = var.use_custom_table_name ? var.custom_table_name : local.generated_table_name
  billing_mode     = var.use_on_demand ? "PAY_PER_REQUEST" : "PROVISIONED"
  read_capacity    = var.use_on_demand ? 0 : var.read_capacity
  write_capacity   = var.use_on_demand ? 0 : var.write_capacity
  hash_key         = var.hash_key
  stream_enabled   = true
  stream_view_type = var.view_type

  point_in_time_recovery {
    enabled = var.point_in_time_recovery_enabled
  }

  server_side_encryption {
    enabled     = var.server_side_encryption_enabled #false here means use AWS owned keys, not KMS
    kms_key_arn = var.kms_key_arn                    # if not provided the default AWS KMS key alias/aws/dynamodb is used
  }

  ttl {
    attribute_name = var.ttl_attribute_name
    enabled        = var.ttl_enabled
  }

  dynamic "attribute" {
    for_each = var.attribute
    content {
      name = attribute.value.name
      type = attribute.value.type
    }
  }

  dynamic "replica" {
    for_each = var.global_table_replica_regions
    content {
      point_in_time_recovery = var.point_in_time_recovery_enabled
      propagate_tags         = true
      region_name            = replica.value
    }
  }

  dynamic "global_secondary_index" {
    for_each = var.global_secondary_index
    content {
      name            = global_secondary_index.value.name
      hash_key        = global_secondary_index.value.hash_key
      range_key       = global_secondary_index.value.range_key
      projection_type = global_secondary_index.value.projection_type
      write_capacity  = var.use_on_demand ? 0 : global_secondary_index.value.write_capacity
      read_capacity   = var.use_on_demand ? 0 : global_secondary_index.value.read_capacity
    }
  }

  tags = local.combined_resource_tags
}

resource "aws_dynamodb_table" "dynamodb_table_with_sort_key" {
  # checkov:skip=CKV_AWS_119:DynamoDB is encrypted at rest using an AWS-managed KMS key by default, which is sufficient at this moment. We will add KMS CMK support when we have an established workflow on how different services can interact using a single CMK.
  # checkov:skip=CKV2_AWS_16:This check does not work correctly in conjunction with count. Autoscaling is correctly configured when using provisioned mode.

  lifecycle {
    ignore_changes = [
      read_capacity, write_capacity,
    ]
    precondition {
      condition = var.use_on_demand || alltrue([
        for gsi in var.global_secondary_index : (
          coalesce(gsi["read_capacity"], 0) > 0 && coalesce(gsi["write_capacity"], 0) > 0
        )
      ])
      error_message = "When using provisioned mode all global secondary indexes must have a non-zero read_capacity and write_capacity defined"
    }
  }

  count            = var.range_key_enabled ? 1 : 0
  name             = var.use_custom_table_name ? var.custom_table_name : local.generated_table_name
  billing_mode     = var.use_on_demand ? "PAY_PER_REQUEST" : "PROVISIONED"
  read_capacity    = var.use_on_demand ? 0 : var.read_capacity
  write_capacity   = var.use_on_demand ? 0 : var.write_capacity
  hash_key         = var.hash_key
  range_key        = var.range_key
  stream_enabled   = true
  stream_view_type = var.view_type

  point_in_time_recovery {
    enabled = var.point_in_time_recovery_enabled
  }

  server_side_encryption {
    enabled     = var.server_side_encryption_enabled #false here means use AWS owned keys, not KMS
    kms_key_arn = var.kms_key_arn                    # if not provided the default AWS KMS key alias/aws/dynamodb is used
  }

  ttl {
    attribute_name = var.ttl_attribute_name
    enabled        = var.ttl_enabled
  }

  dynamic "attribute" {
    for_each = var.attribute
    content {
      name = attribute.value.name
      type = attribute.value.type
    }
  }

  dynamic "replica" {
    for_each = var.global_table_replica_regions
    content {
      point_in_time_recovery = var.point_in_time_recovery_enabled
      propagate_tags         = true
      region_name            = replica.value
    }
  }

  dynamic "global_secondary_index" {
    for_each = var.global_secondary_index
    content {
      name            = global_secondary_index.value.name
      hash_key        = global_secondary_index.value.hash_key
      range_key       = global_secondary_index.value.range_key
      projection_type = global_secondary_index.value.projection_type
      write_capacity  = var.use_on_demand ? 0 : global_secondary_index.value.write_capacity
      read_capacity   = var.use_on_demand ? 0 : global_secondary_index.value.read_capacity

    }
  }

  tags = local.combined_resource_tags
}

# Create an autoscaling read capacity target
resource "aws_appautoscaling_target" "dynamodb_table_read_target" {
  count              = !var.use_on_demand && !var.range_key_enabled ? 1 : 0
  max_capacity       = var.max_read_capacity
  min_capacity       = var.min_read_capacity
  resource_id        = "table/${aws_dynamodb_table.dynamodb_table[count.index].name}"
  scalable_dimension = "dynamodb:table:ReadCapacityUnits"
  service_namespace  = "dynamodb"
}

# Create an autoscaling read capacity target
resource "aws_appautoscaling_target" "dynamodb_table_read_target_with_sort_key" {
  count              = !var.use_on_demand && var.range_key_enabled ? 1 : 0
  max_capacity       = var.max_read_capacity
  min_capacity       = var.min_read_capacity
  resource_id        = "table/${aws_dynamodb_table.dynamodb_table_with_sort_key[count.index].name}"
  scalable_dimension = "dynamodb:table:ReadCapacityUnits"
  service_namespace  = "dynamodb"
}

# Create an autoscaling policy to adjust read capacity
resource "aws_appautoscaling_policy" "dynamodb_table_read_policy" {
  count              = !var.use_on_demand && !var.range_key_enabled ? 1 : 0
  name               = "DynamoDBReadCapacityUtilization:${aws_appautoscaling_target.dynamodb_table_read_target[count.index].resource_id}"
  policy_type        = "TargetTrackingScaling"
  resource_id        = aws_appautoscaling_target.dynamodb_table_read_target[count.index].resource_id
  scalable_dimension = aws_appautoscaling_target.dynamodb_table_read_target[count.index].scalable_dimension
  service_namespace  = aws_appautoscaling_target.dynamodb_table_read_target[count.index].service_namespace

  target_tracking_scaling_policy_configuration {
    predefined_metric_specification {
      predefined_metric_type = "DynamoDBReadCapacityUtilization"
    }

    target_value = var.read_capacity_utilization_percentage
  }
}

# Create an autoscaling policy to adjust read capacity
resource "aws_appautoscaling_policy" "dynamodb_table_read_policy_with_sort_key" {
  count              = !var.use_on_demand && var.range_key_enabled ? 1 : 0
  name               = "DynamoDBReadCapacityUtilization:${aws_appautoscaling_target.dynamodb_table_read_target_with_sort_key[count.index].resource_id}"
  policy_type        = "TargetTrackingScaling"
  resource_id        = aws_appautoscaling_target.dynamodb_table_read_target_with_sort_key[count.index].resource_id
  scalable_dimension = aws_appautoscaling_target.dynamodb_table_read_target_with_sort_key[count.index].scalable_dimension
  service_namespace  = aws_appautoscaling_target.dynamodb_table_read_target_with_sort_key[count.index].service_namespace

  target_tracking_scaling_policy_configuration {
    predefined_metric_specification {
      predefined_metric_type = "DynamoDBReadCapacityUtilization"
    }

    target_value = var.read_capacity_utilization_percentage
  }
}

# Create an autoscaling write capacity target
resource "aws_appautoscaling_target" "dynamodb_table_write_target" {
  count              = !var.use_on_demand && !var.range_key_enabled ? 1 : 0
  max_capacity       = var.max_write_capacity
  min_capacity       = var.min_write_capacity
  resource_id        = "table/${aws_dynamodb_table.dynamodb_table[count.index].name}"
  scalable_dimension = "dynamodb:table:WriteCapacityUnits"
  service_namespace  = "dynamodb"
}

# Create an autoscaling write capacity target
resource "aws_appautoscaling_target" "dynamodb_table_write_target_with_sort_key" {
  count              = !var.use_on_demand && var.range_key_enabled ? 1 : 0
  max_capacity       = var.max_write_capacity
  min_capacity       = var.min_write_capacity
  resource_id        = "table/${aws_dynamodb_table.dynamodb_table_with_sort_key[count.index].name}"
  scalable_dimension = "dynamodb:table:WriteCapacityUnits"
  service_namespace  = "dynamodb"
}

# Create an autoscaling policy to adjust write capacity
resource "aws_appautoscaling_policy" "dynamodb_table_write_policy" {
  count              = !var.use_on_demand && !var.range_key_enabled ? 1 : 0
  name               = "DynamoDBwriteCapacityUtilization:${aws_appautoscaling_target.dynamodb_table_write_target[count.index].resource_id}"
  policy_type        = "TargetTrackingScaling"
  resource_id        = aws_appautoscaling_target.dynamodb_table_write_target[count.index].resource_id
  scalable_dimension = aws_appautoscaling_target.dynamodb_table_write_target[count.index].scalable_dimension
  service_namespace  = aws_appautoscaling_target.dynamodb_table_write_target[count.index].service_namespace

  target_tracking_scaling_policy_configuration {
    predefined_metric_specification {
      predefined_metric_type = "DynamoDBWriteCapacityUtilization"
    }

    target_value = var.write_capacity_utilization_percentage
  }
}

# Create an autoscaling policy to adjust write capacity
resource "aws_appautoscaling_policy" "dynamodb_table_write_policy_with_sort_key" {
  count              = !var.use_on_demand && var.range_key_enabled ? 1 : 0
  name               = "DynamoDBwriteCapacityUtilization:${aws_appautoscaling_target.dynamodb_table_write_target_with_sort_key[count.index].resource_id}"
  policy_type        = "TargetTrackingScaling"
  resource_id        = aws_appautoscaling_target.dynamodb_table_write_target_with_sort_key[count.index].resource_id
  scalable_dimension = aws_appautoscaling_target.dynamodb_table_write_target_with_sort_key[count.index].scalable_dimension
  service_namespace  = aws_appautoscaling_target.dynamodb_table_write_target_with_sort_key[count.index].service_namespace

  target_tracking_scaling_policy_configuration {
    predefined_metric_specification {
      predefined_metric_type = "DynamoDBWriteCapacityUtilization"
    }

    target_value = var.write_capacity_utilization_percentage
  }
}