CCM-12745: initial send event code

Author: Ian-Hodges

.tool-versions

@@ -3,6 +3,7 @@ gitleaks 8.24.0
 jq 1.6
 nodejs 22.11.0
 pre-commit 3.6.0
+python 3.13.2
 terraform 1.10.1
 terraform-docs 0.19.0
 trivy 0.61.0

lambdas/ttl-create-lambda/src/infra/types.ts

@@ -1,15 +1,4 @@
-export type CloudEvent = {
-  id: string;
-  source: string;
-  specversion: string;
-  type: string;
-  plane: string;
-  subject: string;
-  time: string; // ISO 8601 datetime string
-  datacontenttype: string;
-  dataschema: string;
-  dataschemaversion: string;
-};
+import { CloudEvent } from 'utils';
 
 export type TtlItemEvent = CloudEvent & {
   data: {

package-lock.json

@@ -1,6 +1,7 @@
 {
   "dependencies": {
     "@aws-sdk/client-dynamodb": "^3.914.0",
+    "@aws-sdk/client-eventbridge": "^3.918.0",
     "@aws-sdk/client-lambda": "^3.914.0",
     "@aws-sdk/client-s3": "^3.914.0",
     "@aws-sdk/client-sqs": "^3.914.0",
@@ -10,7 +11,7 @@
     "async-mutex": "^0.4.0",
     "date-fns": "^4.1.0",
     "winston": "^3.17.0",
-    "zod": "^3.24.2"
+    "zod": "^4.1.12"
   },
   "devDependencies": {
     "@aws-sdk/types": "^3.914.0",

utils/utils/package.json

@@ -0,0 +1,326 @@
+import { SQSClient, SendMessageBatchCommand } from '@aws-sdk/client-sqs';
+import {
+  EventBridgeClient,
+  PutEventsCommand,
+} from '@aws-sdk/client-eventbridge';
+import { mockClient } from 'aws-sdk-client-mock';
+import { CloudEvent } from 'types';
+
+import { EventPublishConfig, sendEvents } from 'event-publish';
+
+process.env.EVENT_BUS_ARN =
+  'arn:aws:events:us-east-1:123456789012:event-bus/test-bus';
+process.env.DLQ_URL =
+  'https://sqs.us-east-1.amazonaws.com/123456789012/test-dlq';
+
+const eventBridgeMock = mockClient(EventBridgeClient);
+const sqsMock = mockClient(SQSClient);
+
+const testConfig: EventPublishConfig = {
+  eventBusArn: 'arn:aws:events:us-east-1:123456789012:event-bus/test-bus',
+  dlqUrl: 'https://sqs.us-east-1.amazonaws.com/123456789012/test-dlq',
+};
+
+const validCloudEvent: CloudEvent = {
+  id: '1',
+  source: 'test-source',
+  specversion: '1.0',
+  type: 'test-type',
+  plane: 'control',
+  subject: 'test-subject',
+  time: '2023-01-01T00:00:00Z',
+  datacontenttype: 'application/json',
+  dataschema: 'https://example.com/schema',
+  dataschemaversion: '1.0',
+};
+
+const validCloudEvent2: CloudEvent = {
+  ...validCloudEvent,
+  id: '2',
+};
+
+const invalidCloudEvent = {
+  type: 'data',
+  id: 'missing-source',
+};
+
+const validEvents = [validCloudEvent, validCloudEvent2];
+const invalidEvents = [invalidCloudEvent as unknown as CloudEvent];
+const mixedEvents = [
+  validCloudEvent,
+  invalidCloudEvent as unknown as CloudEvent,
+];
+
+describe('Event Publishing', () => {
+  beforeEach(() => {
+    eventBridgeMock.reset();
+    sqsMock.reset();
+  });
+
+  test('should return empty array when no events provided', async () => {
+    const result = await sendEvents([], testConfig);
+
+    expect(result).toEqual([]);
+    expect(eventBridgeMock.calls()).toHaveLength(0);
+    expect(sqsMock.calls()).toHaveLength(0);
+  });
+
+  test('should send valid events to EventBridge', async () => {
+    eventBridgeMock.on(PutEventsCommand).resolves({
+      FailedEntryCount: 0,
+      Entries: [{ EventId: 'event-1' }],
+    });
+
+    const result = await sendEvents(validEvents, testConfig);
+
+    expect(result).toEqual([]);
+    expect(eventBridgeMock.calls()).toHaveLength(1);
+    expect(sqsMock.calls()).toHaveLength(0);
+
+    const eventBridgeCall = eventBridgeMock.calls()[0];
+    expect(eventBridgeCall.args[0].input).toEqual({
+      Entries: [
+        {
+          Source: 'custom.event',
+          DetailType: 'test-type',
+          Detail: JSON.stringify(validCloudEvent),
+          EventBusName:
+            'arn:aws:events:us-east-1:123456789012:event-bus/test-bus',
+        },
+        {
+          Source: 'custom.event',
+          DetailType: 'test-type',
+          Detail: JSON.stringify(validCloudEvent2),
+          EventBusName:
+            'arn:aws:events:us-east-1:123456789012:event-bus/test-bus',
+        },
+      ],
+    });
+  });
+
+  test('should send invalid events directly to DLQ', async () => {
+    sqsMock.on(SendMessageBatchCommand).resolves({
+      Successful: [
+        { Id: 'msg-1', MessageId: 'success-1', MD5OfMessageBody: 'hash1' },
+      ],
+    });
+
+    const result = await sendEvents(invalidEvents, testConfig);
+
+    expect(result).toEqual([]);
+    expect(eventBridgeMock.calls()).toHaveLength(0);
+    expect(sqsMock.calls()).toHaveLength(1);
+
+    const sqsCall = sqsMock.calls()[0];
+    const sqsInput = sqsCall.args[0].input as any;
+    expect(sqsInput.QueueUrl).toBe(
+      'https://sqs.us-east-1.amazonaws.com/123456789012/test-dlq',
+    );
+    expect(sqsInput.Entries).toHaveLength(1);
+    expect(sqsInput.Entries[0].MessageBody).toBe(
+      JSON.stringify(invalidCloudEvent),
+    );
+    expect(sqsInput.Entries[0].Id).toBeDefined();
+  });
+
+  test('should handle mixed valid and invalid events', async () => {
+    eventBridgeMock.on(PutEventsCommand).resolves({
+      FailedEntryCount: 0,
+      Entries: [{ EventId: 'event-1' }],
+    });
+    sqsMock.on(SendMessageBatchCommand).resolves({
+      Successful: [
+        { Id: 'msg-1', MessageId: 'success-1', MD5OfMessageBody: 'hash1' },
+      ],
+    });
+
+    const result = await sendEvents(mixedEvents, testConfig);
+
+    expect(result).toEqual([]);
+    expect(eventBridgeMock.calls()).toHaveLength(1);
+    expect(sqsMock.calls()).toHaveLength(1);
+
+    // Verify EventBridge only gets valid events
+    const eventBridgeCall = eventBridgeMock.calls()[0];
+    expect(eventBridgeCall.args[0].input).toEqual({
+      Entries: [
+        {
+          Source: 'custom.event',
+          DetailType: 'test-type',
+          Detail: JSON.stringify(validCloudEvent),
+          EventBusName:
+            'arn:aws:events:us-east-1:123456789012:event-bus/test-bus',
+        },
+      ],
+    });
+
+    // Verify DLQ only gets invalid events
+    const sqsCall = sqsMock.calls()[0];
+    const sqsInput = sqsCall.args[0].input as any;
+    expect(sqsInput.Entries).toHaveLength(1);
+    expect(sqsInput.Entries[0].MessageBody).toBe(
+      JSON.stringify(invalidCloudEvent),
+    );
+  });
+
+  test('should send failed EventBridge events to DLQ', async () => {
+    eventBridgeMock.on(PutEventsCommand).resolves({
+      FailedEntryCount: 1,
+      Entries: [
+        { ErrorCode: 'InternalFailure', ErrorMessage: 'Internal error' },
+        { EventId: 'event-2' },
+      ],
+    });
+    sqsMock.on(SendMessageBatchCommand).resolves({
+      Successful: [
+        { Id: 'msg-1', MessageId: 'success-1', MD5OfMessageBody: 'hash1' },
+      ],
+    });
+
+    const result = await sendEvents(validEvents, testConfig);
+
+    expect(result).toEqual([]);
+    expect(eventBridgeMock.calls()).toHaveLength(1);
+    expect(sqsMock.calls()).toHaveLength(1);
+
+    // Verify EventBridge was called with both events
+    const eventBridgeCall = eventBridgeMock.calls()[0];
+    const eventBridgeInput = eventBridgeCall.args[0].input as any;
+    expect(eventBridgeInput.Entries).toHaveLength(2);
+
+    // Verify DLQ gets the failed event (first one)
+    const sqsCall = sqsMock.calls()[0];
+    const sqsInput = sqsCall.args[0].input as any;
+    expect(sqsInput.Entries).toHaveLength(1);
+    expect(sqsInput.Entries[0].MessageBody).toBe(
+      JSON.stringify(validCloudEvent),
+    );
+  });
+
+  test('should handle EventBridge send error and send all events to DLQ', async () => {
+    eventBridgeMock
+      .on(PutEventsCommand)
+      .rejects(new Error('EventBridge error'));
+    sqsMock.on(SendMessageBatchCommand).resolves({
+      Successful: [
+        { Id: 'msg-1', MessageId: 'success-1', MD5OfMessageBody: 'hash1' },
+      ],
+    });
+
+    const result = await sendEvents(validEvents, testConfig);
+
+    expect(result).toEqual([]);
+    expect(eventBridgeMock.calls()).toHaveLength(1);
+    expect(sqsMock.calls()).toHaveLength(1);
+  });
+
+  test('should return failed events when DLQ also fails', async () => {
+    sqsMock.on(SendMessageBatchCommand).callsFake((params) => {
+      const firstEntryId = params.Entries[0].Id;
+      return Promise.resolve({
+        Failed: [
+          {
+            Id: firstEntryId,
+            Code: 'SenderFault',
+            Message: 'Invalid message',
+            SenderFault: true,
+          },
+        ],
+      });
+    });
+
+    const result = await sendEvents(invalidEvents, testConfig);
+
+    expect(result).toHaveLength(1);
+    expect(eventBridgeMock.calls()).toHaveLength(0);
+    expect(sqsMock.calls()).toHaveLength(1);
+  });
+
+  test('should handle DLQ send error and return all events as failed', async () => {
+    sqsMock.on(SendMessageBatchCommand).rejects(new Error('DLQ error'));
+
+    const result = await sendEvents(invalidEvents, testConfig);
+
+    expect(result).toEqual(invalidEvents);
+    expect(eventBridgeMock.calls()).toHaveLength(0);
+    expect(sqsMock.calls()).toHaveLength(1);
+  });
+
+  test('should process multiple batches for large event arrays', async () => {
+    const largeEventArray = Array.from({ length: 25 })
+      .fill(null)
+      .map((_, i) => ({
+        ...validCloudEvent,
+        id: `event-${i}`,
+      }));
+
+    eventBridgeMock.on(PutEventsCommand).resolves({
+      FailedEntryCount: 0,
+      Entries: [{ EventId: 'success' }],
+    });
+
+    const result = await sendEvents(largeEventArray, testConfig);
+
+    expect(result).toEqual([]);
+    expect(eventBridgeMock.calls()).toHaveLength(3);
+
+    // Verify batch sizes: 10, 10, 5
+    const calls = eventBridgeMock.calls();
+    const firstBatchInput = calls[0].args[0].input as any;
+    const secondBatchInput = calls[1].args[0].input as any;
+    const thirdBatchInput = calls[2].args[0].input as any;
+
+    expect(firstBatchInput.Entries).toHaveLength(10);
+    expect(secondBatchInput.Entries).toHaveLength(10);
+    expect(thirdBatchInput.Entries).toHaveLength(5);
+
+    // Verify all use the same EventBusName
+    expect(firstBatchInput.Entries[0].EventBusName).toBe(
+      'arn:aws:events:us-east-1:123456789012:event-bus/test-bus',
+    );
+    expect(secondBatchInput.Entries[0].EventBusName).toBe(
+      'arn:aws:events:us-east-1:123456789012:event-bus/test-bus',
+    );
+    expect(thirdBatchInput.Entries[0].EventBusName).toBe(
+      'arn:aws:events:us-east-1:123456789012:event-bus/test-bus',
+    );
+  });
+
+  test('should preserve CloudEvent structure in EventBridge Detail field', async () => {
+    eventBridgeMock.on(PutEventsCommand).resolves({
+      FailedEntryCount: 0,
+      Entries: [{ EventId: 'event-1' }],
+    });
+
+    const customEvent: CloudEvent = {
+      ...validCloudEvent,
+      id: 'custom-event-123',
+      source: 'nhs.notify.digital-letters',
+      type: 'letter.created',
+      subject: 'Patient/12345',
+    };
+
+    await sendEvents([customEvent], testConfig);
+
+    const eventBridgeCall = eventBridgeMock.calls()[0];
+    const entry = (eventBridgeCall.args[0].input as any).Entries[0];
+
+    // Verify the CloudEvent is preserved as-is in the Detail field
+    expect(entry.Detail).toBe(JSON.stringify(customEvent));
+
+    // Verify EventBridge-specific fields are correctly mapped
+    expect(entry.Source).toBe('custom.event');
+    expect(entry.DetailType).toBe('letter.created');
+    expect(entry.EventBusName).toBe(
+      'arn:aws:events:us-east-1:123456789012:event-bus/test-bus',
+    );
+
+    // Verify the original CloudEvent structure is intact in Detail
+    const detailObject = JSON.parse(entry.Detail);
+    expect(detailObject.id).toBe('custom-event-123');
+    expect(detailObject.source).toBe('nhs.notify.digital-letters');
+    expect(detailObject.type).toBe('letter.created');
+    expect(detailObject.subject).toBe('Patient/12345');
+    expect(detailObject.specversion).toBe('1.0');
+  });
+});