Put SQS queue between SNS topic and lambda

stevebux · stevebux · commit 210ca88426c5 · 2026-01-09T16:30:39.000Z
diff --git a/infrastructure/terraform/components/api/README.md b/infrastructure/terraform/components/api/README.md
@@ -58,6 +58,7 @@ No requirements.
 | <a name="module_post_mi"></a> [post\_mi](#module\_post\_mi) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.29/terraform-lambda.zip | n/a |
 | <a name="module_s3bucket_test_letters"></a> [s3bucket\_test\_letters](#module\_s3bucket\_test\_letters) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.26/terraform-s3bucket.zip | n/a |
 | <a name="module_supplier_events_forwarder_lambda"></a> [supplier\_events\_forwarder\_lambda](#module\_supplier\_events\_forwarder\_lambda) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.26/terraform-lambda.zip | n/a |
+| <a name="module_supplier_events_queue"></a> [supplier\_events\_queue](#module\_supplier\_events\_queue) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.26/terraform-sqs.zip | n/a |
 | <a name="module_supplier_requests_queue"></a> [supplier\_requests\_queue](#module\_supplier\_requests\_queue) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.24/terraform-sqs.zip | n/a |
 | <a name="module_supplier_ssl"></a> [supplier\_ssl](#module\_supplier\_ssl) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.26/terraform-ssl.zip | n/a |
 | <a name="module_upsert_letter"></a> [upsert\_letter](#module\_upsert\_letter) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.29/terraform-lambda.zip | n/a |
diff --git a/infrastructure/terraform/components/api/module_lambda_supplier_events_forwarder.tf b/infrastructure/terraform/components/api/module_lambda_supplier_events_forwarder.tf
@@ -68,4 +68,19 @@ data "aws_iam_policy_document" "supplier_events_forwarder_lambda" {
       module.eventsub.firehose_delivery_stream.arn,
     ]
   }
+
+  statement {
+    sid    = "SQSPermissions"
+    effect = "Allow"
+
+    actions = [
+      "sqs:ReceiveMessage",
+      "sqs:DeleteMessage",
+      "sqs:GetQueueAttributes",
+    ]
+
+    resources = [
+      module.supplier_events_queue.sqs_queue_arn,
+    ]
+  }
 }
diff --git a/infrastructure/terraform/components/api/module_sqs_supplier_events_queue.tf b/infrastructure/terraform/components/api/module_sqs_supplier_events_queue.tf
@@ -0,0 +1,74 @@
+module "supplier_events_queue" {
+  source = "https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.26/terraform-sqs.zip"
+
+  aws_account_id = var.aws_account_id
+  component      = var.component
+  environment    = var.environment
+  project        = var.project
+  region         = var.region
+  name           = "${local.csi}-supplier-events-queue"
+
+  fifo_queue                  = true
+  content_based_deduplication = true
+
+  sqs_kms_key_arn = module.kms.key_arn
+
+  visibility_timeout_seconds = 60
+
+  create_dlq          = true
+  sqs_policy_overload = data.aws_iam_policy_document.supplier_events_queue_policy.json
+}
+
+data "aws_iam_policy_document" "supplier_events_queue_policy" {
+  version = "2012-10-17"
+  statement {
+    sid    = "AllowSNSToSendMessage"
+    effect = "Allow"
+
+    principals {
+      type        = "Service"
+      identifiers = ["sns.amazonaws.com"]
+    }
+
+    actions = [
+      "sqs:SendMessage"
+    ]
+
+    resources = [
+      "arn:aws:sqs:${var.region}:${var.aws_account_id}:${local.csi}-supplier-events-queue.fifo"
+    ]
+
+    condition {
+      test     = "ArnEquals"
+      variable = "aws:SourceArn"
+      values   = [module.eventsub.sns_topic_supplier.arn]
+    }
+  }
+
+  statement {
+    sid    = "AllowSNSPermissions"
+    effect = "Allow"
+
+    principals {
+      type        = "Service"
+      identifiers = ["sns.amazonaws.com"]
+    }
+
+    actions = [
+      "sqs:SendMessage",
+      "sqs:ListQueueTags",
+      "sqs:GetQueueUrl",
+      "sqs:GetQueueAttributes",
+    ]
+
+    resources = [
+      "arn:aws:sqs:${var.region}:${var.aws_account_id}:${local.csi}-supplier-events-queue.fifo"
+    ]
+
+    condition {
+      test     = "ArnEquals"
+      variable = "aws:SourceArn"
+      values   = [module.eventsub.sns_topic_supplier.arn]
+    }
+  }
+}
diff --git a/infrastructure/terraform/components/api/sns_topic_subscription_supplier_events_forwarder_lambda.tf b/infrastructure/terraform/components/api/sns_topic_subscription_supplier_events_forwarder_lambda.tf
@@ -1,13 +1,19 @@
-resource "aws_sns_topic_subscription" "supplier_events_forwarder_lambda" {
-  topic_arn = module.eventsub.sns_topic_supplier.arn
-  protocol  = "lambda"
-  endpoint  = module.supplier_events_forwarder_lambda.function_arn
+resource "aws_sns_topic_subscription" "supplier_events_queue" {
+  topic_arn            = module.eventsub.sns_topic_supplier.arn
+  protocol             = "sqs"
+  endpoint             = module.supplier_events_queue.sqs_queue_arn
+  raw_message_delivery = false
 }
 
-resource "aws_lambda_permission" "supplier_events_forwarder_lambda_sns" {
-  statement_id  = "AllowExecutionFromSNS"
-  action        = "lambda:InvokeFunction"
-  function_name = module.supplier_events_forwarder_lambda.function_name
-  principal     = "sns.amazonaws.com"
-  source_arn    = module.eventsub.sns_topic_supplier.arn
+resource "aws_lambda_event_source_mapping" "supplier_events_forwarder" {
+  event_source_arn                   = module.supplier_events_queue.sqs_queue_arn
+  function_name                      = module.supplier_events_forwarder_lambda.function_arn
+  batch_size                         = 10
+  maximum_batching_window_in_seconds = 1
+  scaling_config { maximum_concurrency = 10 }
+
+  depends_on = [
+    module.supplier_events_queue,
+    module.supplier_events_forwarder_lambda
+  ]
 }
diff --git a/lambdas/supplier-events-forwarder/src/__tests__/forwarder.test.ts b/lambdas/supplier-events-forwarder/src/__tests__/forwarder.test.ts
@@ -1,56 +1,69 @@
-import { Context, SNSEvent, SNSEventRecord } from "aws-lambda";
+import { Context, SQSEvent, SQSRecord } from "aws-lambda";
 import { FirehoseClient, PutRecordCommand } from "@aws-sdk/client-firehose";
 import { mockDeep } from "jest-mock-extended";
 import pino from "pino";
 import createForwarder from "../forwarder";
 import { Deps } from "../deps";
 
-function createSNSEvent(records: SNSEventRecord[]): SNSEvent {
+function createSQSEvent(records: SQSRecord[]): SQSEvent {
   return {
     Records: records,
   };
 }
 
-function createSNSEventRecord(
-  message: string,
-  overrides: Partial<SNSEventRecord["Sns"]> = {},
-): SNSEventRecord {
+/**
+ * Creates an SQS record with a body containing the SNS notification wrapper.
+ * This simulates what SNS delivers to SQS when raw_message_delivery is false.
+ */
+function createSQSRecord(
+  body: string,
+  messageId = "test-sqs-msg-id",
+): SQSRecord {
   return {
-    EventVersion: "1.0",
-    EventSubscriptionArn:
-      "arn:aws:sns:eu-west-2:123456789012:topic:subscription",
-    EventSource: "aws:sns",
-    Sns: {
-      SignatureVersion: "1",
-      Timestamp: "2026-01-09T12:00:00.000Z",
-      Signature: "test-signature",
-      SigningCertUrl: "https://sns.eu-west-2.amazonaws.com/cert.pem",
-      MessageId: "test-message-id",
-      Message: message,
-      MessageAttributes: {},
-      Type: "Notification",
-      UnsubscribeUrl: "https://sns.eu-west-2.amazonaws.com/unsubscribe",
-      TopicArn: "arn:aws:sns:eu-west-2:123456789012:test-topic",
-      Subject: "Test Subject",
-      ...overrides,
+    messageId,
+    receiptHandle: "test-receipt-handle",
+    body,
+    attributes: {
+      ApproximateReceiveCount: "1",
+      SentTimestamp: "1704801600000",
+      SenderId: "123456789012",
+      ApproximateFirstReceiveTimestamp: "1704801600000",
     },
+    messageAttributes: {},
+    md5OfBody: "test-md5",
+    eventSource: "aws:sqs",
+    eventSourceARN: "arn:aws:sqs:eu-west-2:123456789012:test-queue.fifo",
+    awsRegion: "eu-west-2",
   };
 }
 
-function buildExpectedSnsWrapper(record: SNSEventRecord): object {
-  return {
+/**
+ * Creates an SNS notification wrapper as it would appear in the SQS message body
+ * when raw_message_delivery is false.
+ */
+function createSnsNotificationWrapper(
+  message: string,
+  overrides: Partial<{
+    MessageId: string;
+    TopicArn: string;
+    Subject: string;
+  }> = {},
+): string {
+  return JSON.stringify({
     Type: "Notification",
-    MessageId: record.Sns.MessageId,
-    TopicArn: record.Sns.TopicArn,
-    Subject: record.Sns.Subject,
-    Message: record.Sns.Message,
-    Timestamp: record.Sns.Timestamp,
-    SignatureVersion: record.Sns.SignatureVersion,
-    Signature: record.Sns.Signature,
-    SigningCertUrl: record.Sns.SigningCertUrl,
-    UnsubscribeUrl: record.Sns.UnsubscribeUrl,
-    MessageAttributes: record.Sns.MessageAttributes,
-  };
+    MessageId: overrides.MessageId ?? "test-sns-message-id",
+    TopicArn:
+      overrides.TopicArn ??
+      "arn:aws:sns:eu-west-2:123456789012:test-topic.fifo",
+    Subject: overrides.Subject ?? "Test Subject",
+    Message: message,
+    Timestamp: "2026-01-09T12:00:00.000Z",
+    SignatureVersion: "1",
+    Signature: "test-signature",
+    SigningCertUrl: "https://sns.eu-west-2.amazonaws.com/cert.pem",
+    UnsubscribeUrl: "https://sns.eu-west-2.amazonaws.com/unsubscribe",
+    MessageAttributes: {},
+  });
 }
 
 describe("forwarder", () => {
@@ -74,13 +87,14 @@ describe("forwarder", () => {
   });
 
   describe("createForwarder", () => {
-    it("should process a single SNS record and send to Firehose", async () => {
+    it("should process a single SQS record and send to Firehose", async () => {
       const message = JSON.stringify({ eventType: "test", data: "value" });
-      const snsRecord = createSNSEventRecord(message);
-      const snsEvent = createSNSEvent([snsRecord]);
+      const snsWrapper = createSnsNotificationWrapper(message);
+      const sqsRecord = createSQSRecord(snsWrapper);
+      const sqsEvent = createSQSEvent([sqsRecord]);
 
       const handler = createForwarder(mockDeps);
-      await handler(snsEvent, mockDeep<Context>(), jest.fn());
+      await handler(sqsEvent, mockDeep<Context>(), jest.fn());
 
       expect(mockFirehoseClient.send).toHaveBeenCalledTimes(1);
       expect(mockFirehoseClient.send).toHaveBeenCalledWith(
@@ -89,89 +103,87 @@ describe("forwarder", () => {
 
       const sentCommand = mockFirehoseClient.send.mock
         .calls[0][0] as PutRecordCommand;
-      const expectedWrapper = buildExpectedSnsWrapper(snsRecord);
       expect(sentCommand.input).toEqual({
         DeliveryStreamName: mockDeliveryStreamName,
         Record: {
-          Data: Buffer.from(`${JSON.stringify(expectedWrapper)}\n`),
+          Data: Buffer.from(`${snsWrapper}\n`),
         },
       });
     });
 
-    it("should process multiple SNS records and send to Firehose", async () => {
+    it("should process multiple SQS records and send to Firehose", async () => {
       const message1 = JSON.stringify({ eventType: "test1" });
       const message2 = JSON.stringify({ eventType: "test2" });
       const message3 = JSON.stringify({ eventType: "test3" });
 
-      const snsRecord1 = createSNSEventRecord(message1, {
+      const snsWrapper1 = createSnsNotificationWrapper(message1, {
         MessageId: "msg-1",
       });
-      const snsRecord2 = createSNSEventRecord(message2, {
+      const snsWrapper2 = createSnsNotificationWrapper(message2, {
         MessageId: "msg-2",
       });
-      const snsRecord3 = createSNSEventRecord(message3, {
+      const snsWrapper3 = createSnsNotificationWrapper(message3, {
         MessageId: "msg-3",
       });
 
-      const snsEvent = createSNSEvent([snsRecord1, snsRecord2, snsRecord3]);
+      const sqsEvent = createSQSEvent([
+        createSQSRecord(snsWrapper1, "sqs-1"),
+        createSQSRecord(snsWrapper2, "sqs-2"),
+        createSQSRecord(snsWrapper3, "sqs-3"),
+      ]);
 
       const handler = createForwarder(mockDeps);
-      await handler(snsEvent, mockDeep<Context>(), jest.fn());
+      await handler(sqsEvent, mockDeep<Context>(), jest.fn());
 
       expect(mockFirehoseClient.send).toHaveBeenCalledTimes(3);
 
       const sentCommands = mockFirehoseClient.send.mock.calls.map(
-        (call: [PutRecordCommand]) => call[0],
+        (call) => call[0] as PutRecordCommand,
       );
 
       expect(sentCommands[0].input).toEqual({
         DeliveryStreamName: mockDeliveryStreamName,
         Record: {
-          Data: Buffer.from(
-            `${JSON.stringify(buildExpectedSnsWrapper(snsRecord1))}\n`,
-          ),
+          Data: Buffer.from(`${snsWrapper1}\n`),
         },
       });
 
       expect(sentCommands[1].input).toEqual({
         DeliveryStreamName: mockDeliveryStreamName,
         Record: {
-          Data: Buffer.from(
-            `${JSON.stringify(buildExpectedSnsWrapper(snsRecord2))}\n`,
-          ),
+          Data: Buffer.from(`${snsWrapper2}\n`),
         },
       });
 
       expect(sentCommands[2].input).toEqual({
         DeliveryStreamName: mockDeliveryStreamName,
         Record: {
-          Data: Buffer.from(
-            `${JSON.stringify(buildExpectedSnsWrapper(snsRecord3))}\n`,
-          ),
+          Data: Buffer.from(`${snsWrapper3}\n`),
         },
       });
     });
 
-    it("should handle empty SNS event with no records", async () => {
-      const snsEvent = createSNSEvent([]);
+    it("should handle empty SQS event with no records", async () => {
+      const sqsEvent = createSQSEvent([]);
 
       const handler = createForwarder(mockDeps);
-      await handler(snsEvent, mockDeep<Context>(), jest.fn());
+      await handler(sqsEvent, mockDeep<Context>(), jest.fn());
 
       expect(mockFirehoseClient.send).not.toHaveBeenCalled();
     });
 
-    it("should wrap message in SNS notification format", async () => {
+    it("should forward the SNS notification wrapper from SQS body to Firehose", async () => {
       const message = JSON.stringify({ key: "value" });
-      const snsRecord = createSNSEventRecord(message, {
+      const snsWrapper = createSnsNotificationWrapper(message, {
         MessageId: "unique-msg-id",
-        TopicArn: "arn:aws:sns:eu-west-2:123456789012:my-topic",
+        TopicArn: "arn:aws:sns:eu-west-2:123456789012:my-topic.fifo",
         Subject: "My Subject",
       });
-      const snsEvent = createSNSEvent([snsRecord]);
+      const sqsRecord = createSQSRecord(snsWrapper);
+      const sqsEvent = createSQSEvent([sqsRecord]);
 
       const handler = createForwarder(mockDeps);
-      await handler(snsEvent, mockDeep<Context>(), jest.fn());
+      await handler(sqsEvent, mockDeep<Context>(), jest.fn());
 
       const sentCommand = mockFirehoseClient.send.mock
         .calls[0][0] as PutRecordCommand;
@@ -181,7 +193,7 @@ describe("forwarder", () => {
       expect(parsedData).toEqual({
         Type: "Notification",
         MessageId: "unique-msg-id",
-        TopicArn: "arn:aws:sns:eu-west-2:123456789012:my-topic",
+        TopicArn: "arn:aws:sns:eu-west-2:123456789012:my-topic.fifo",
         Subject: "My Subject",
         Message: message,
         Timestamp: "2026-01-09T12:00:00.000Z",
@@ -193,13 +205,14 @@ describe("forwarder", () => {
       });
     });
 
-    it("should append newline to wrapped message for JSON Lines format", async () => {
+    it("should append newline to message for JSON Lines format", async () => {
       const message = JSON.stringify({ key: "value" });
-      const snsRecord = createSNSEventRecord(message);
-      const snsEvent = createSNSEvent([snsRecord]);
+      const snsWrapper = createSnsNotificationWrapper(message);
+      const sqsRecord = createSQSRecord(snsWrapper);
+      const sqsEvent = createSQSEvent([sqsRecord]);
 
       const handler = createForwarder(mockDeps);
-      await handler(snsEvent, mockDeep<Context>(), jest.fn());
+      await handler(sqsEvent, mockDeep<Context>(), jest.fn());
 
       const sentCommand = mockFirehoseClient.send.mock
         .calls[0][0] as PutRecordCommand;
diff --git a/lambdas/supplier-events-forwarder/src/forwarder.ts b/lambdas/supplier-events-forwarder/src/forwarder.ts
