# Managed Streaming for Kafka (MSK) event mapping

resource "aws_lambda_permission" "allow_execution_from_msk_cluster" {
  count         = var.msk_event_enabled ? 1 : 0
  statement_id  = "${var.environment}-${replace(var.event_source_mapping_msk_cluster_name, ".", "-")}-lambda-AllowMSKPermissions"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.function.arn
  principal     = "kafka.amazonaws.com"
  source_arn    = local.event_source_mapping_arn["msk"]
}

# Attach MSK policy to Lambda role
resource "aws_iam_role_policy_attachment" "lambda_msk_policy_attachment" {
  count      = var.msk_event_enabled ? 1 : 0
  role       = aws_iam_role.lambda_execution_role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaMSKExecutionRole"
}

resource "aws_lambda_event_source_mapping" "msk_event_source_mapping" {
  count                              = var.msk_event_enabled ? length(var.kafka_topics) : 0
  batch_size                         = var.event_source_mapping_batch_size
  event_source_arn                   = local.event_source_mapping_arn["msk"]
  function_name                      = local.lambda_function_arn
  topics                             = [var.kafka_topics[count.index]] # base terraform module topics param is array of only length 1
  starting_position                  = var.event_source_mapping_starting_position
  maximum_batching_window_in_seconds = var.event_source_mapping_batch_window

  dynamic "destination_config" {
    for_each = var.event_source_mapping_failure_arn != null ? [""] : []
    content {
      on_failure {
        destination_arn = var.event_source_mapping_failure_arn
      }
    }
  }

  dynamic "amazon_managed_kafka_event_source_config" {
    for_each = var.kafka_use_lambda_name_as_consumer_id ? [var.lambda_name] : []
    content {
      consumer_group_id = "${var.lambda_name}-${var.kafka_topics[count.index]}"
    }
  }

  dynamic "filter_criteria" {
    for_each = var.event_source_mapping_filter_criteria_pattern != "" ? [""] : []
    content {
      filter {
        pattern = var.event_source_mapping_filter_criteria_pattern
      }
    }
  }
}