CCM-12614: add some basic event handling

Author: Ian-Hodges

infrastructure/terraform/components/dl/README.md

@@ -28,7 +28,7 @@ No requirements.
 | <a name="input_parent_acct_environment"></a> [parent\_acct\_environment](#input\_parent\_acct\_environment) | Name of the environment responsible for the acct resources used, affects things like DNS zone. Useful for named dev environments | `string` | `"main"` | no |
 | <a name="input_project"></a> [project](#input\_project) | The name of the tfscaffold project | `string` | n/a | yes |
 | <a name="input_queue_batch_size"></a> [queue\_batch\_size](#input\_queue\_batch\_size) | maximum number of queue items to process | `number` | `10` | no |
-| <a name="input_queue_batch_window_seconds"></a> [queue\_batch\_window\_seconds](#input\_queue\_batch\_window\_seconds) | maximum time in seconds between processing events | `number` | `10` | no |
+| <a name="input_queue_batch_window_seconds"></a> [queue\_batch\_window\_seconds](#input\_queue\_batch\_window\_seconds) | maximum time in seconds between processing events | `number` | `1` | no |
 | <a name="input_region"></a> [region](#input\_region) | The AWS Region | `string` | n/a | yes |
 | <a name="input_shared_infra_account_id"></a> [shared\_infra\_account\_id](#input\_shared\_infra\_account\_id) | The AWS Shared Infra Account ID (numeric) | `string` | n/a | yes |
 | <a name="input_ttl_poll_schedule"></a> [ttl\_poll\_schedule](#input\_ttl\_poll\_schedule) | Schedule to poll for any overdue TTL records | `string` | `"rate(10 minutes)"` | no |

infrastructure/terraform/components/dl/cloudwatch_event_rule_pdm_resource_unavailable.tf

@@ -0,0 +1,20 @@
+resource "aws_cloudwatch_event_rule" "pdm_resource_unavailable" {
+  name           = "${local.csi}-pdm-resource-unavailable"
+  description    = "PDM resource unavailable event rule"
+  event_bus_name = aws_cloudwatch_event_bus.main.name
+
+  event_pattern = jsonencode({
+    "detail" : {
+      "type" : [
+        "uk.nhs.notify.digital.letters.pdm.resource.unavailable.v1"
+      ]
+    }
+  })
+}
+
+resource "aws_cloudwatch_event_target" "pdm_resource_unavailable" {
+  rule           = aws_cloudwatch_event_rule.pdm_resource_unavailable.name
+  arn            = module.sqs_pdm_poll.sqs_queue_arn
+  target_id      = "pdm-resource-unavailable-target"
+  event_bus_name = aws_cloudwatch_event_bus.main.name
+}

infrastructure/terraform/components/dl/variables.tf

@@ -101,7 +101,7 @@ variable "queue_batch_size" {
 variable "queue_batch_window_seconds" {
   type        = number
   description = "maximum time in seconds between processing events"
-  default     = 10
+  default     = 1
 }
 
 variable "enable_dynamodb_delete_protection" {

lambdas/pdm-poll-lambda/src/__tests__/apis/sqs-handler.test.ts

@@ -1,36 +1,30 @@
-import { createHandler } from 'apis/sqs-handler';
-import { Logger } from 'utils';
-import { SQSEvent, SQSRecord } from 'aws-lambda';
 import { mock } from 'jest-mock-extended';
+import { randomUUID } from 'node:crypto';
+import { createHandler } from 'apis/sqs-handler';
+import { EventPublisher, Logger } from 'utils';
+import { Pdm } from 'app/pdm';
+import {
+  pdmResourceSubmittedEvent,
+  pdmResourceUnavailableEvent,
+  recordEvent,
+} from '__tests__/test-data';
 
 const logger = mock<Logger>();
+const eventPublisher = mock<EventPublisher>();
+const pdm = mock<Pdm>();
+
+jest.mock('node:crypto', () => ({
+  randomUUID: jest.fn(),
+}));
 
-const event = {
-  sourceEventId: 'test-event-id',
-};
-
-const sqsRecord1: SQSRecord = {
-  messageId: '1',
-  receiptHandle: 'abc',
-  body: JSON.stringify(event),
-  attributes: {
-    ApproximateReceiveCount: '1',
-    SentTimestamp: '2025-07-03T14:23:30Z',
-    SenderId: 'sender-id',
-    ApproximateFirstReceiveTimestamp: '2025-07-03T14:23:30Z',
-  },
-  messageAttributes: {},
-  md5OfBody: '',
-  eventSource: 'aws:sqs',
-  eventSourceARN: '',
-  awsRegion: '',
-};
-
-const singleRecordEvent: SQSEvent = {
-  Records: [sqsRecord1],
-};
+const mockRandomUUID = randomUUID as jest.MockedFunction<typeof randomUUID>;
+const mockDate = jest.spyOn(Date.prototype, 'toISOString');
+mockRandomUUID.mockReturnValue('550e8400-e29b-41d4-a716-446655440001');
+mockDate.mockReturnValue('2023-06-20T12:00:00.250Z');
 
 const handler = createHandler({
+  eventPublisher,
+  pdm,
   logger,
 });
 
@@ -39,35 +33,169 @@ describe('SQS Handler', () => {
     jest.clearAllMocks();
   });
 
-  it('processes a single record', async () => {
-    const response = await handler(singleRecordEvent);
+  describe('pdm.resource.submitted', () => {
+    it('should send pdm.resource.available event when the document is ready', async () => {
+      pdm.poll.mockResolvedValueOnce('available');
+
+      const response = await handler(recordEvent([pdmResourceSubmittedEvent]));
+
+      expect(eventPublisher.sendEvents).toHaveBeenCalledWith([
+        {
+          ...pdmResourceSubmittedEvent,
+          id: '550e8400-e29b-41d4-a716-446655440001',
+          time: '2023-06-20T12:00:00.250Z',
+          recordedtime: '2023-06-20T12:00:00.250Z',
+          type: 'uk.nhs.notify.digital.letters.pdm.resource.available.v1',
+        },
+      ]);
+      expect(logger.info).toHaveBeenCalledWith(
+        'Received SQS Event of 1 record(s)',
+      );
+      expect(logger.info).toHaveBeenCalledWith(
+        '1 of 1 records processed successfully',
+      );
+      expect(response).toEqual({ batchItemFailures: [] });
+    });
 
-    expect(logger.info).toHaveBeenCalledWith(
-      'Received SQS Event of 1 record(s)',
-    );
-    expect(logger.info).toHaveBeenCalledWith(
-      '1 of 1 records processed successfully',
-    );
-    expect(response).toEqual({ batchItemFailures: [] });
+    it('should send pdm.resource.unavailable event when the document is not ready', async () => {
+      pdm.poll.mockResolvedValueOnce('unavailable');
+
+      const response = await handler(recordEvent([pdmResourceSubmittedEvent]));
+
+      expect(eventPublisher.sendEvents).toHaveBeenCalledWith([
+        {
+          ...pdmResourceSubmittedEvent,
+          id: '550e8400-e29b-41d4-a716-446655440001',
+          time: '2023-06-20T12:00:00.250Z',
+          recordedtime: '2023-06-20T12:00:00.250Z',
+          type: 'uk.nhs.notify.digital.letters.pdm.resource.unavailable.v1',
+          data: {
+            ...pdmResourceSubmittedEvent.data,
+            retryCount: 1,
+          },
+        },
+      ]);
+      expect(logger.info).toHaveBeenCalledWith(
+        'Received SQS Event of 1 record(s)',
+      );
+      expect(logger.info).toHaveBeenCalledWith(
+        '1 of 1 records processed successfully',
+      );
+      expect(response).toEqual({ batchItemFailures: [] });
+    });
   });
 
-  it('should return failed items to the queue if an error occurs while processing them', async () => {
-    singleRecordEvent.Records[0].body = 'not-json';
+  describe('pdm.resource.unavailable', () => {
+    it('should send pdm.resource.available event when the document is ready', async () => {
+      pdm.poll.mockResolvedValueOnce('available');
+
+      const response = await handler(
+        recordEvent([pdmResourceUnavailableEvent]),
+      );
+
+      expect(eventPublisher.sendEvents).toHaveBeenCalledWith([
+        {
+          ...pdmResourceUnavailableEvent,
+          id: '550e8400-e29b-41d4-a716-446655440001',
+          time: '2023-06-20T12:00:00.250Z',
+          recordedtime: '2023-06-20T12:00:00.250Z',
+          type: 'uk.nhs.notify.digital.letters.pdm.resource.available.v1',
+        },
+      ]);
+      expect(logger.info).toHaveBeenCalledWith(
+        'Received SQS Event of 1 record(s)',
+      );
+      expect(logger.info).toHaveBeenCalledWith(
+        '1 of 1 records processed successfully',
+      );
+      expect(response).toEqual({ batchItemFailures: [] });
+    });
+
+    it('should send pdm.resource.unavailable event when the document is not ready', async () => {
+      pdm.poll.mockResolvedValueOnce('unavailable');
 
-    const result = await handler(singleRecordEvent);
+      const response = await handler(
+        recordEvent([pdmResourceUnavailableEvent]),
+      );
 
-    expect(logger.warn).toHaveBeenCalledWith({
-      error: `Unexpected token 'o', "not-json" is not valid JSON`,
-      description: 'Failed processing message',
-      messageId: '1',
+      expect(eventPublisher.sendEvents).toHaveBeenCalledWith([
+        {
+          ...pdmResourceUnavailableEvent,
+          id: '550e8400-e29b-41d4-a716-446655440001',
+          time: '2023-06-20T12:00:00.250Z',
+          recordedtime: '2023-06-20T12:00:00.250Z',
+          type: 'uk.nhs.notify.digital.letters.pdm.resource.unavailable.v1',
+          data: {
+            ...pdmResourceSubmittedEvent.data,
+            retryCount: 2,
+          },
+        },
+      ]);
+      expect(logger.info).toHaveBeenCalledWith(
+        'Received SQS Event of 1 record(s)',
+      );
+      expect(logger.info).toHaveBeenCalledWith(
+        '1 of 1 records processed successfully',
+      );
+      expect(response).toEqual({ batchItemFailures: [] });
     });
 
-    expect(logger.info).toHaveBeenCalledWith(
-      '0 of 1 records processed successfully',
-    );
+    it('should send pdm.resource.retries.exceeded event when the document is not ready after 10 retries', async () => {
+      pdm.poll.mockResolvedValueOnce('unavailable');
+
+      const testEvent = {
+        ...pdmResourceUnavailableEvent,
+        data: {
+          ...pdmResourceUnavailableEvent.data,
+          retryCount: 9,
+        },
+      };
+
+      const response = await handler(recordEvent([testEvent]));
+
+      expect(eventPublisher.sendEvents).toHaveBeenCalledWith([
+        {
+          ...pdmResourceUnavailableEvent,
+          id: '550e8400-e29b-41d4-a716-446655440001',
+          time: '2023-06-20T12:00:00.250Z',
+          recordedtime: '2023-06-20T12:00:00.250Z',
+          type: 'uk.nhs.notify.digital.letters.pdm.resource.retries.exceeded.v1',
+          data: {
+            ...pdmResourceSubmittedEvent.data,
+            retryCount: 10,
+          },
+        },
+      ]);
+      expect(logger.info).toHaveBeenCalledWith(
+        'Received SQS Event of 1 record(s)',
+      );
+      expect(logger.info).toHaveBeenCalledWith(
+        '1 of 1 records processed successfully',
+      );
+      expect(response).toEqual({ batchItemFailures: [] });
+    });
+  });
+
+  describe('errors', () => {
+    it('should return failed items to the queue if an error occurs while processing them', async () => {
+      const event = recordEvent([pdmResourceSubmittedEvent]);
+      event.Records[0].body = 'not-json';
+
+      const result = await handler(event);
+
+      expect(logger.warn).toHaveBeenCalledWith({
+        error: `Unexpected token 'o', "not-json" is not valid JSON`,
+        description: 'Failed processing message',
+        messageId: '1',
+      });
+
+      expect(logger.info).toHaveBeenCalledWith(
+        '0 of 1 records processed successfully',
+      );
 
-    expect(result).toEqual({
-      batchItemFailures: [{ itemIdentifier: '1' }],
+      expect(result).toEqual({
+        batchItemFailures: [{ itemIdentifier: '1' }],
+      });
     });
   });
 });

lambdas/pdm-poll-lambda/src/__tests__/app/pdm.test.ts

@@ -0,0 +1,77 @@
+import { mock } from 'jest-mock-extended';
+import { Logger } from 'utils';
+import { Pdm, PdmDependencies } from 'app/pdm';
+import { pdmResourceSubmittedEvent } from '__tests__/test-data';
+
+const logger = mock<Logger>();
+const validConfig = (): PdmDependencies => ({
+  pdmUrl: 'https://example.com/pdm',
+  logger,
+});
+
+describe('Pdm', () => {
+  describe('constructor', () => {
+    it('is created when required deps are provided', () => {
+      const cfg = validConfig();
+      expect(() => new Pdm(cfg)).not.toThrow();
+    });
+
+    it('throws if pdmUrl is not provided', () => {
+      const cfg = {
+        logger,
+      } as unknown as PdmDependencies;
+
+      expect(() => new Pdm(cfg)).toThrow('pdmUrl has not been specified');
+    });
+
+    it('throws if logger is not provided', () => {
+      const cfg = {
+        pdmUrl: 'https://example.com/pdm',
+      } as PdmDependencies;
+
+      expect(() => new Pdm(cfg)).toThrow('logger has not been provided');
+    });
+  });
+
+  describe('poll', () => {
+    it('returns available when the document is ready', async () => {
+      const cfg = validConfig();
+      const pdm = new Pdm(cfg);
+
+      const result = await pdm.poll(pdmResourceSubmittedEvent);
+
+      expect(result).toBe('available');
+    });
+
+    it('returns unavailable when the document is not ready', async () => {
+      const cfg = validConfig();
+      const pdm = new Pdm(cfg);
+
+      pdmResourceSubmittedEvent.data.messageReference = 'ref2';
+
+      const result = await pdm.poll(pdmResourceSubmittedEvent);
+
+      expect(result).toBe('unavailable');
+    });
+
+    it('returns failed and logs error when logger.info throws', async () => {
+      const cfg = validConfig();
+      const thrown = new Error('logger failure');
+      cfg.logger.info = jest.fn(() => {
+        throw thrown;
+      });
+
+      const pdm = new Pdm(cfg);
+
+      await expect(pdm.poll(pdmResourceSubmittedEvent)).rejects.toThrow(thrown);
+
+      expect(logger.error).toHaveBeenCalledTimes(1);
+      expect(logger.error).toHaveBeenCalledWith(
+        expect.objectContaining({
+          description: 'Error getting document resource from PDM',
+          err: thrown,
+        }),
+      );
+    });
+  });
+});