CCM-19812: Fix eventpub queue names

infrastructure/terraform/modules/eventpub/README.md

@@ -46,7 +46,6 @@
 
 | Name | Description |
 |------|-------------|
-| <a name="output_dlq"></a> [dlq](#output\_dlq) | EventPub DLQ name and ARN |
 | <a name="output_publishing_anomaly_alarm"></a> [publishing\_anomaly\_alarm](#output\_publishing\_anomaly\_alarm) | CloudWatch anomaly detection alarm details for SNS publishing |
 | <a name="output_s3_bucket_event_cache"></a> [s3\_bucket\_event\_cache](#output\_s3\_bucket\_event\_cache) | S3 Bucket ARN and Name for event cache |
 | <a name="output_sns_topic"></a> [sns\_topic](#output\_sns\_topic) | SNS Topic ARN and Name |

infrastructure/terraform/modules/eventpub/cloudwatch_metric_alarm_dlq_alarm.tf

@@ -12,6 +12,6 @@ resource "aws_cloudwatch_metric_alarm" "dlq_alarm" {
   treat_missing_data  = "notBreaching"
 
   dimensions = {
-    QueueName = aws_sqs_queue.dlq.name
+    QueueName = module.sqs_queue.sqs_dlq_name
   }
 }

infrastructure/terraform/modules/eventpub/iam_role_lambda.tf

@@ -68,19 +68,6 @@ data "aws_iam_policy_document" "lambda" {
     ]
   }
 
-  statement {
-    sid    = "DLQPutMessage"
-    effect = "Allow"
-
-    actions = [
-      "sqs:SendMessage",
-    ]
-
-    resources = [
-      aws_sqs_queue.dlq.arn
-    ]
-  }
-
   statement {
     sid    = "AllowSQSInput"
     effect = "Allow"

infrastructure/terraform/modules/eventpub/lambda/eventpub/package.json

@@ -19,7 +19,6 @@
   "license": "ISC",
   "dependencies": {
     "@aws-sdk/client-eventbridge": "^3.1004.0",
-    "@aws-sdk/client-sqs": "^3.1004.0",
     "aws-sdk-client-mock": "^4.1.0"
   }
 }

infrastructure/terraform/modules/eventpub/lambda/eventpub/src/__tests__/index.test.js

@@ -1,10 +1,8 @@
 const { handler } = require('../index.js');
 const { mockClient } = require('aws-sdk-client-mock');
-const { SQSClient, SendMessageCommand } = require('@aws-sdk/client-sqs');
 const { EventBridgeClient, PutEventsCommand } = require('@aws-sdk/client-eventbridge');
 
 const eventBridgeMock = mockClient(EventBridgeClient);
-const sqsMock = mockClient(SQSClient);
 
 const validCloudEvent = {
   id: "123e4567-e89b-12d3-a456-426614174000",
@@ -34,72 +32,83 @@ const invalidCloudEvent = {
   data: {}
 };
 
-const snsEvent = {
+const sqsEvent = {
   Records: [
-    { Sns: { Message: JSON.stringify(validCloudEvent) } }
+    {
+      messageId: 'msg-1',
+      body: JSON.stringify({ Message: JSON.stringify(validCloudEvent) })
+    }
   ]
 };
 
-const snsEventInvalid = {
+const sqsEventInvalid = {
   Records: [
-    { Sns: { Message: JSON.stringify(invalidCloudEvent) } }
+    {
+      messageId: 'msg-1',
+      body: JSON.stringify({ Message: JSON.stringify(invalidCloudEvent) })
+    }
   ]
 };
 
-describe('SNS to EventBridge Lambda', () => {
+describe('SQS to EventBridge Lambda', () => {
     beforeEach(() => {
         eventBridgeMock.reset();
-        sqsMock.reset();
     });
 
     test('Valid event is sent to the correct EventBridge bus', async () => {
         eventBridgeMock.on(PutEventsCommand).resolves({ FailedEntryCount: 0, Entries: [{}] });
 
-        await handler(snsEvent);
+        const result = await handler(sqsEvent);
 
         expect(eventBridgeMock.calls()).toHaveLength(1);
+        expect(result.batchItemFailures).toHaveLength(0);
     });
 
-    test('Invalid event is sent to DLQ', async () => {
-        sqsMock.on(SendMessageCommand).resolves({ MessageId: '123' });
+    test('Invalid event is reported as a batch item failure', async () => {
+        const result = await handler(sqsEventInvalid);
 
-        await handler(snsEventInvalid);
-
-        expect(sqsMock.calls()).toHaveLength(1);
+        expect(eventBridgeMock.calls()).toHaveLength(0);
+        expect(result.batchItemFailures).toHaveLength(1);
+        expect(result.batchItemFailures[0].itemIdentifier).toBe('msg-1');
     });
 
-    test('Event with unknown plane field is sent to DLQ', async () => {
+    test('Event with unknown plane field is reported as a batch item failure', async () => {
         const eventUnknownPlane = { ...validCloudEvent, plane: "unknown" };
 
-        const snsEventUnknownPlane = {
+        const sqsEventUnknownPlane = {
             Records: [
-                { Sns: { Message: JSON.stringify(eventUnknownPlane) } }
+                {
+                    messageId: 'msg-1',
+                    body: JSON.stringify({ Message: JSON.stringify(eventUnknownPlane) })
+                }
             ]
         };
 
-        await handler(snsEventUnknownPlane);
+        const result = await handler(sqsEventUnknownPlane);
 
-        expect(sqsMock.calls()).toHaveLength(1);
+        expect(result.batchItemFailures).toHaveLength(1);
+        expect(result.batchItemFailures[0].itemIdentifier).toBe('msg-1');
     });
 
-    test('Retries on EventBridge failure and sends failed events to DLQ', async () => {
+    test('Retries on EventBridge failure and reports failed events as batch item failures', async () => {
       eventBridgeMock
           .on(PutEventsCommand)
           .rejectsOnce(Object.assign(new Error('Rate limit exceeded'), { retryable: true }))
           .resolves({ FailedEntryCount: 1, Entries: [{ ErrorCode: 'InternalFailure' }] });
-      sqsMock.on(SendMessageCommand).resolves({ MessageId: '123' });
 
-      await handler(snsEvent);
+      const result = await handler(sqsEvent);
 
       expect(eventBridgeMock.calls()).toHaveLength(2);
-      expect(sqsMock.calls()).toHaveLength(1);
+      expect(result.batchItemFailures).toHaveLength(1);
     });
 
     test('Throttling delays event processing', async () => {
       process.env.THROTTLE_DELAY_MS = '500';
       jest.useFakeTimers();
 
-      const handlerPromise = handler(snsEvent);
+      eventBridgeMock.on(PutEventsCommand).resolves({ FailedEntryCount: 0, Entries: [{}] });
+
+      const handlerPromise = handler(sqsEvent);
       jest.advanceTimersByTime(500);
       await handlerPromise;
 

infrastructure/terraform/modules/eventpub/lambda/eventpub/src/index.js

@@ -1,12 +1,9 @@
 const { EventBridgeClient, PutEventsCommand } = require('@aws-sdk/client-eventbridge');
-const { SQSClient, SendMessageCommand } = require('@aws-sdk/client-sqs');
 
 const eventBridge = new EventBridgeClient({});
-const sqs = new SQSClient({});
 
 const DATA_PLANE_EVENT_BUS_ARN = process.env.DATA_PLANE_EVENT_BUS_ARN;
 const CONTROL_PLANE_EVENT_BUS_ARN = process.env.CONTROL_PLANE_EVENT_BUS_ARN;
-const DLQ_URL = process.env.DLQ_URL;
 const THROTTLE_DELAY_MS = parseInt(process.env.THROTTLE_DELAY_MS || '0', 10);
 const MAX_RETRIES = 3;
 const EVENTBRIDGE_MAX_BATCH_SIZE = 10;
@@ -44,8 +41,8 @@ async function sendToEventBridge(events, eventBusArn) {
         const batch = events.slice(i, i + EVENTBRIDGE_MAX_BATCH_SIZE);
         const entries = batch.map(event => ({
             Source: 'custom.event',
-            DetailType: event.type,
-            Detail: JSON.stringify(event),
+            DetailType: event.message.type,
+            Detail: JSON.stringify(event.message),
             EventBusName: eventBusArn
         }));
 
@@ -70,8 +67,8 @@ async function sendToEventBridge(events, eventBusArn) {
                     await new Promise(res => setTimeout(res, 2 ** attempts * 100));
                     attempts++;
                 } else {
-                    console.error(`Non-retryable error encountered. Moving ${batch.length} events to DLQ`);
-                    failedEvents.push(...batch);
+                    console.error(`Non-retryable error encountered. Reporting ${batch.length} as batch item failures`);
+                    failedEvents.push(...batch.map(event => ({ itemIdentifier: event.itemIdentifier })));
                     break;
                 }
             }
@@ -80,21 +77,6 @@ async function sendToEventBridge(events, eventBusArn) {
     return failedEvents;
 }
 
-async function sendToDLQ(events) {
-  if (events.length === 0) return;
-
-  console.warn(`Sending ${events.length} failed event(s) to DLQ: ${DLQ_URL}`);
-
-  for (const event of events) {
-        try {
-            await sqs.send(new SendMessageCommand({ QueueUrl: DLQ_URL, MessageBody: JSON.stringify(event) }));
-            console.debug(`Successfully sent event ${event.id} to DLQ`);
-        } catch (error) {
-            console.error(`Failed to send event ${event.id} to DLQ - Name: ${error.name}, Message: ${error.message}, Code: ${error.$metadata?.httpStatusCode}, RequestId: ${error.$metadata?.requestId}`);
-        }
-    }
-}
-
 exports.handler = async (sqsEvent) => {
     console.debug(`Received SQS event with ${sqsEvent.Records.length} records.`);
 
@@ -103,28 +85,36 @@ exports.handler = async (sqsEvent) => {
         await new Promise(res => setTimeout(res, THROTTLE_DELAY_MS));
     }
 
-    const records = sqsEvent.Records
-        .map(record => record.body)
-        .map(JSON.parse)
-        .map(record => record.Message)
-        .map(JSON.parse);
-    const validEvents = records.filter(validateEvent);
-    const invalidEvents = records.filter(event => !validateEvent(event));
-
-    console.debug(`Valid events: ${validEvents.length}, Invalid events: ${invalidEvents.length}`);
-    if (invalidEvents.length) {
-        console.warn(`${invalidEvents.length} event(s) failed validation and will be sent to DLQ`);
-        await sendToDLQ(invalidEvents);
-    }
+    const dataPlaneEvents = [];
+    const controlPlaneEvents = [];
+    const batchItemFailures = [];
 
-    const dataEvents = validEvents.filter(event => event.plane === 'data');
-    const controlEvents = validEvents.filter(event => event.plane === 'control');
-    const unknownEvents = validEvents.filter(event => event.plane !== 'data' && event.plane !== 'control');
+    for(const sqsRecord of sqsEvent.Records) {
+        const record = JSON.parse(sqsRecord.body);
+        const message = JSON.parse(record.Message);
+
+        if (!validateEvent(message)) {
+            console.error(`Invalid event received. EventID: ${message.id || 'unknown'}, EventType: ${message.type || 'unknown'}`);
+            batchItemFailures.push({ itemIdentifier: sqsRecord.messageId });
+        }
+        else if (message.plane === 'data') {
+            dataPlaneEvents.push({ itemIdentifier: sqsRecord.messageId, message: message });
+        }
+        else if (message.plane === 'control') {
+            controlPlaneEvents.push({ itemIdentifier: sqsRecord.messageId, message: message });
+        }
+        else {
+            console.error(`Unknown plane type received: ${message.plane}. EventID: ${message.id || 'unknown'}, EventType: ${message.type || 'unknown'}`);
+            batchItemFailures.push({ itemIdentifier: sqsRecord.messageId });
+        }
+    }
 
-    console.debug(`Data events: ${dataEvents.length}, Control events: ${controlEvents.length}, Unknown events: ${unknownEvents.length}`);
+    console.debug(`Data events: ${dataPlaneEvents.length}, Control events: ${controlPlaneEvents.length}, Invalid/Unknown events: ${batchItemFailures.length}`);
 
-    const failedDataEvents = await sendToEventBridge(dataEvents, DATA_PLANE_EVENT_BUS_ARN);
-    const failedControlEvents = await sendToEventBridge(controlEvents, CONTROL_PLANE_EVENT_BUS_ARN);
+    batchItemFailures.push(...await sendToEventBridge(dataPlaneEvents, DATA_PLANE_EVENT_BUS_ARN));
+    batchItemFailures.push(...await sendToEventBridge(controlPlaneEvents, CONTROL_PLANE_EVENT_BUS_ARN));
 
-    await sendToDLQ([...failedDataEvents, ...failedControlEvents, ...unknownEvents]);
+    return {
+        batchItemFailures: batchItemFailures
+    };
 };

infrastructure/terraform/modules/eventpub/lambda_function.tf

@@ -23,7 +23,6 @@ resource "aws_lambda_function" "main" {
     variables = {
       DATA_PLANE_EVENT_BUS_ARN    = var.data_plane_bus_arn
       CONTROL_PLANE_EVENT_BUS_ARN = var.control_plane_bus_arn
-      DLQ_URL                     = aws_sqs_queue.dlq.url
       THROTTLE_DELAY_MS           = "0"
     }
   }