LAB40_AWS_Resilient Queue Processing: SQS + Lambda with Concurrency Controls and DLQ

variable "aws_region"           { type = string; default = "us-east-1" }
variable "queue_name"           { type = string; default = "payment-processing" }
variable "lambda_timeout"       { type = number; default = 30; description = "Lambda timeout in seconds." }
variable "max_concurrency"      { type = number; default = 50; description = "Max concurrent Lambda executions." }
variable "alert_email"          { type = string; description = "Email for DLQ alarm notifications." }
variable "environment"          { type = string; default = "prod" }

provider "aws" { region = var.aws_region }
locals { tags = { Environment = var.environment, ManagedBy = "terraform" } }

# Dead-Letter Queue — receives messages that fail 3 times.
resource "aws_sqs_queue" "dlq" {
  name                        = "${var.queue_name}-dlq-${var.environment}.fifo"
  fifo_queue                  = true
  message_retention_seconds   = 1209600  # 14 days
  content_based_deduplication = true
  tags                        = local.tags
}

# Main FIFO queue — ensures ordered, exactly-once processing.
resource "aws_sqs_queue" "main" {
  name                        = "${var.queue_name}-${var.environment}.fifo"
  fifo_queue                  = true
  content_based_deduplication = true

  # visibility_timeout must be >= 6 * lambda_timeout to prevent premature redelivery.
  visibility_timeout_seconds = var.lambda_timeout * 6

  # Redrive policy: after 3 receive attempts, send to DLQ.
  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.dlq.arn
    maxReceiveCount     = 3
  })

  tags = local.tags
}

# Lambda execution role.
data "aws_iam_policy_document" "assume" {
  statement { effect = "Allow"; actions = ["sts:AssumeRole"]
    principals { type = "Service"; identifiers = ["lambda.amazonaws.com"] } }
}
resource "aws_iam_role" "processor" {
  name               = "${var.queue_name}-processor-role"
  assume_role_policy = data.aws_iam_policy_document.assume.json
}
resource "aws_iam_role_policy_attachment" "logs" {
  role       = aws_iam_role.processor.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
}
resource "aws_iam_role_policy" "sqs" {
  name = "sqs-consume"
  role = aws_iam_role.processor.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect   = "Allow"
      Action   = ["sqs:ReceiveMessage","sqs:DeleteMessage","sqs:GetQueueAttributes","sqs:ChangeMessageVisibility"]
      Resource = [aws_sqs_queue.main.arn, aws_sqs_queue.dlq.arn]
    }]
  })
}

resource "aws_cloudwatch_log_group" "processor" {
  name              = "/aws/lambda/${var.queue_name}-processor-${var.environment}"
  retention_in_days = 30
}

# Lambda processor — reserved_concurrent_executions throttles to 50 max.
resource "aws_lambda_function" "processor" {
  function_name = "${var.queue_name}-processor-${var.environment}"
  runtime       = "python3.12"
  handler       = "processor.lambda_handler"
  role          = aws_iam_role.processor.arn
  filename      = "${path.module}/processor.zip"
  timeout       = var.lambda_timeout

  # Cap concurrency — prevents overwhelming downstream payment gateway.
  reserved_concurrent_executions = var.max_concurrency

  depends_on = [aws_cloudwatch_log_group.processor]
  tags       = local.tags
}

# SQS event source mapping — triggers Lambda from the queue.
resource "aws_lambda_event_source_mapping" "sqs" {
  event_source_arn = aws_sqs_queue.main.arn
  function_name    = aws_lambda_function.processor.arn
  batch_size       = 10
  enabled          = true

  # Allow Lambda to report which specific messages in a batch failed.
  # Only failed messages become visible again — others are deleted.
  function_response_types = ["ReportBatchItemFailures"]

  # Scale down Lambda polling when the queue is low (saves cost).
  scaling_config { maximum_concurrency = var.max_concurrency }
}

# SNS topic for DLQ alarms.
resource "aws_sns_topic" "dlq_alarm" {
  name = "${var.queue_name}-dlq-alert-${var.environment}"
  tags = local.tags
}
resource "aws_sns_topic_subscription" "email" {
  topic_arn = aws_sns_topic.dlq_alarm.arn
  protocol  = "email"
  endpoint  = var.alert_email
}

# CloudWatch alarm — fires immediately when any message lands in the DLQ.
resource "aws_cloudwatch_metric_alarm" "dlq_depth" {
  alarm_name          = "${var.queue_name}-dlq-not-empty-${var.environment}"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  metric_name         = "ApproximateNumberOfMessagesVisible"
  namespace           = "AWS/SQS"
  period              = 60
  statistic           = "Sum"
  threshold           = 0
  alarm_description   = "Payment messages are failing and landing in the DLQ"
  dimensions          = { QueueName = aws_sqs_queue.dlq.name }
  alarm_actions       = [aws_sns_topic.dlq_alarm.arn]
  tags                = local.tags
}

output "queue_url"     { value = aws_sqs_queue.main.url }
output "dlq_url"       { value = aws_sqs_queue.dlq.url }
output "lambda_arn"    { value = aws_lambda_function.processor.arn }
output "sns_topic_arn" { value = aws_sns_topic.dlq_alarm.arn }

terraform init
terraform apply -var="[email protected]"

# Send a test payment message to the queue
aws sqs send-message \
  --queue-url $(terraform output -raw queue_url) \
  --message-body '{"payment_id":"PAY-001","amount":99.99,"currency":"USD"}' \
  --message-group-id "payment-001" \
  --message-deduplication-id "PAY-001-$(date +%s)"

# Check DLQ depth
aws sqs get-queue-attributes \
  --queue-url $(terraform output -raw dlq_url) \
  --attribute-names ApproximateNumberOfMessages

# View Lambda concurrency metrics
aws cloudwatch get-metric-statistics \
  --namespace AWS/Lambda \
  --metric-name ConcurrentExecutions \
  --dimensions Name=FunctionName,Value=payment-processing-processor-prod \
  --start-time $(date -u -v-1H +%Y-%m-%dT%H:%M:%SZ) \
  --end-time $(date -u +%Y-%m-%dT%H:%M:%SZ) \
  --period 60 --statistics Maximum

terraform destroy -var="alert_email=..."