AWS Asynchronous Messaging with SNS and SQS

AWSTemplateFormatVersion: '2010-09-09'
Description: Provisions an SNS topic, SQS queue with subscription, and dead-letter queue
Resources:
  MyDLQ:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: MyDeadLetterQueue
      Tags:
        - Key: Environment
          Value: production
  MySQSQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: MyMessageQueue
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt MyDLQ.Arn
        maxReceiveCount: 5
      Tags:
        - Key: Environment
          Value: production
  MySNSTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: MyMessageTopic
      Tags:
        - Key: Environment
          Value: production
  SQSSubscription:
    Type: AWS::SNS::Subscription
    Properties:
      Protocol: sqs
      TopicArn: !Ref MySNSTopic
      Endpoint: !GetAtt MySQSQueue.Arn
      FilterPolicy:
        eventType:
          - order_created
          - payment_processed
  SQSQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues:
        - !Ref MySQSQueue
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: sns.amazonaws.com
            Action: sqs:SendMessage
            Resource: !GetAtt MySQSQueue.Arn
            Condition:
              ArnEquals:
                aws:SourceArn: !Ref MySNSTopic
Outputs:
  TopicArn:
    Value: !Ref MySNSTopic
  QueueUrl:
    Value: !Ref MySQSQueue
  DLQUrl:
    Value: !Ref MyDLQ
                

from aws_cdk import (
    core,
    aws_sns as sns,
    aws_sqs as sqs,
    aws_lambda as lambda_,
    aws_sns_subscriptions as subscriptions,
    aws_iam as iam
)

class SNSSQSStack(core.Stack):
    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # Define SQS dead-letter queue
        dlq = sqs.Queue(
            self, "MyDeadLetterQueue",
            queue_name="MyDeadLetterQueue",
            retention_period=core.Duration.days(14)
        )

        # Define SQS queue with DLQ
        queue = sqs.Queue(
            self, "MyMessageQueue",
            queue_name="MyMessageQueue",
            visibility_timeout=core.Duration.seconds(30),
            dead_letter_queue=sqs.DeadLetterQueue(
                max_receive_count=5,
                queue=dlq
            )
        )

        # Define SNS topic
        topic = sns.Topic(
            self, "MyMessageTopic",
            topic_name="MyMessageTopic"
        )

        # Subscribe SQS to SNS
        topic.add_subscription(
            subscriptions.SqsSubscription(
                queue,
                filter_policy={
                    "eventType": sns.SubscriptionFilter.string_filter(
                        allowlist=["order_created", "payment_processed"]
                    )
                }
            )
        )

        # Define Lambda consumer
        lambda_consumer = lambda_.Function(
            self, "SQSConsumerLambda",
            runtime=lambda_.Runtime.PYTHON_3_8,
            handler="lambda_function.lambda_handler",
            code=lambda_.Code.from_inline("""
import json

def lambda_handler(event, context):
    for record in event['Records']:
        print("Received SQS message:", json.loads(record['body']))
    return {"statusCode": 200}
"""),
            timeout=core.Duration.seconds(30),
            environment={
                "QUEUE_URL": queue.queue_url
            }
        )

        # Grant Lambda permissions to read from SQS
        queue.grant_consume_messages(lambda_consumer)

        # Grant SNS permissions to write to SQS
        queue.add_to_resource_policy(
            iam.PolicyStatement(
                effect=iam.Effect.ALLOW,
                principals=[iam.ServicePrincipal("sns.amazonaws.com")],
                actions=["sqs:SendMessage"],
                resources=[queue.queue_arn],
                conditions={
                    "ArnEquals": {"aws:SourceArn": topic.topic_arn}
                }
            )
        )

        # Add tags
        core.Tags.of(queue).add("Environment", "production")
        core.Tags.of(topic).add("Environment", "production")
        core.Tags.of(dlq).add("Environment", "production")

app = core.App()
SNSSQSStack(app, "SNSSQSStack")
app.synth()
                

provider "aws" {
  region = "us-west-2"
}

resource "aws_sns_topic" "message_topic" {
  name = "MyMessageTopic"
  tags = {
    Environment = "production"
  }
}

resource "aws_sqs_queue" "dead_letter_queue" {
  name = "MyDeadLetterQueue"
  tags = {
    Environment = "production"
  }
}

resource "aws_sqs_queue" "message_queue" {
  name = "MyMessageQueue"
  visibility_timeout_seconds = 30
  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.dead_letter_queue.arn
    maxReceiveCount     = 5
  })
  tags = {
    Environment = "production"
  }
}

resource "aws_sns_topic_subscription" "sqs_subscription" {
  topic_arn = aws_sns_topic.message_topic.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.message_queue.arn
  filter_policy = jsonencode({
    eventType = ["order_created", "payment_processed"]
  })
}

resource "aws_sqs_queue_policy" "queue_policy" {
  queue_url = aws_sqs_queue.message_queue.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Principal = {
          Service = "sns.amazonaws.com"
        }
        Action = "sqs:SendMessage"
        Resource = aws_sqs_queue.message_queue.arn
        Condition = {
          ArnEquals = {
            "aws:SourceArn" = aws_sns_topic.message_topic.arn
          }
        }
      }
    ]
  })
}

resource "aws_ecs_cluster" "consumer_cluster" {
  name = "MessageConsumerCluster"
  tags = {
    Environment = "production"
  }
}

resource "aws_ecs_task_definition" "consumer_task" {
  family = "MessageConsumerTask"
  network_mode = "awsvpc"
  requires_compatibilities = ["FARGATE"]
  cpu = "256"
  memory = "512"
  execution_role_arn = aws_iam_role.ecs_execution_role.arn
  container_definitions = jsonencode([
    {
      name = "consumer-container"
      image = "amazon/amazon-ecs-sample"
      essential = true
      environment = [
        {
          name = "QUEUE_URL"
          value = aws_sqs_queue.message_queue.id
        }
      ]
      logConfiguration = {
        logDriver = "awslogs"
        options = {
          awslogs-group = "/ecs/message-consumer"
          awslogs-region = "us-west-2"
          awslogs-stream-prefix = "consumer"
        }
      }
    }
  ])
  tags = {
    Environment = "production"
  }
}

resource "aws_iam_role" "ecs_execution_role" {
  name = "ecs-execution-role"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "ecs-tasks.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy" "ecs_sqs_policy" {
  name = "ecs-sqs-policy"
  role = aws_iam_role.ecs_execution_role.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "sqs:ReceiveMessage",
          "sqs:DeleteMessage",
          "sqs:GetQueueAttributes"
        ]
        Resource = aws_sqs_queue.message_queue.arn
      },
      {
        Effect = "Allow"
        Action = [
          "logs:CreateLogStream",
          "logs:PutLogEvents"
        ]
        Resource = "*"
      }
    ]
  })
}