Implement pipe for sqs -> sns

Author: simonrw

cdk/quiz_app/quiz_app_stack.py

@@ -16,7 +16,7 @@
     custom_resources as cr,
     CfnOutput as Output,
 )
-from constructs import Construct
+from constructs import Construct, ConstructOrder
 
 
 class QuizAppStack(Stack):
@@ -64,7 +64,7 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
             write_capacity=5,
         )
 
-        dlq_submission_queue = sqs.Queue(self, "QuizSubmissionQueueDLQ")
+        dlq_submission_queue = sqs.Queue(self, "QuizSubmissionDLQ")
         submission_queue = sqs.Queue(
             self,
             "QuizSubmissionQueue",
@@ -214,6 +214,48 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
             )
         )
 
+        # eventbridge pipe
+        policy_document = iam.PolicyDocument.from_json(
+            {
+                "Version": "2012-10-17",
+                "Statement": [
+                    {
+                        "Effect": "Allow",
+                        "Action": [
+                            "sqs:ReceiveMessage",
+                            "sqs:DeleteMessage",
+                            "sqs:GetQueueAttributes",
+                            "sqs:GetQueueUrl",
+                        ],
+                        "Resource": dlq_submission_queue.queue_arn,
+                    },
+                    {
+                        "Effect": "Allow",
+                        "Action": "sns:Publish",
+                        "Resource": dlq_alarm_topic.topic_arn,
+                    },
+                ],
+            }
+        )
+        policy = iam.ManagedPolicy(
+            self,
+            "PipesPolicy",
+            document=policy_document,
+        )
+        pipes_role = iam.Role(
+            self,
+            f"PipeRole",
+            assumed_by=iam.ServicePrincipal("pipes.amazonaws.com"),
+            managed_policies=[policy],
+        )
+        pipe = pipes.CfnPipe(
+            self,
+            "DLQToSNSPipe",
+            source=dlq_submission_queue.queue_arn,
+            target=dlq_alarm_topic.topic_arn,
+            role_arn=pipes_role.role_arn,
+        )
+
     @staticmethod
     def read_policy_file(file_path: str) -> dict:
         """Reads a JSON policy file and returns it as a dictionary."""