Skip to content

Commit e7acea3

Browse files
Ian-HodgesIan-Hodges
authored
CCM-12745 Shared event publishing library (#91)
1 parent 55609ef commit e7acea3

30 files changed

+2296
-418
lines changed

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,3 +24,4 @@ dist
2424
.reports
2525
**/playwright-report
2626
**/test-results
27+
plugin-cache

‎infrastructure/terraform/components/dl/README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ No requirements.
3434
| <a name="module_kms"></a> [kms](#module\_kms) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.24/terraform-kms.zip | n/a |
3535
| <a name="module_mesh_poll"></a> [mesh\_poll](#module\_mesh\_poll) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.24/terraform-lambda.zip | n/a |
3636
| <a name="module_s3bucket_letters"></a> [s3bucket\_letters](#module\_s3bucket\_letters) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.24/terraform-s3bucket.zip | n/a |
37+
| <a name="module_sqs_event_publisher_errors"></a> [sqs\_event\_publisher\_errors](#module\_sqs\_event\_publisher\_errors) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.24/terraform-sqs.zip | n/a |
3738
| <a name="module_sqs_ttl"></a> [sqs\_ttl](#module\_sqs\_ttl) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.24/terraform-sqs.zip | n/a |
3839
| <a name="module_ttl_create"></a> [ttl\_create](#module\_ttl\_create) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.24/terraform-lambda.zip | n/a |
3940
| <a name="module_ttl_poll"></a> [ttl\_poll](#module\_ttl\_poll) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.24/terraform-lambda.zip | n/a |

‎infrastructure/terraform/components/dl/module_lambda_ttl_create.tf‎

Lines changed: 33 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -36,9 +36,11 @@ module "ttl_create" {
3636
log_subscription_role_arn = local.acct.log_subscription_role_arn
3737

3838
lambda_env_vars = {
39-
"TTL_TABLE_NAME" = aws_dynamodb_table.ttl.name
40-
"TTL_WAIT_TIME_HOURS" = 24
41-
"TTL_SHARD_COUNT" = local.ttl_shard_count
39+
"TTL_TABLE_NAME" = aws_dynamodb_table.ttl.name
40+
"TTL_WAIT_TIME_HOURS" = 24
41+
"TTL_SHARD_COUNT" = local.ttl_shard_count
42+
"EVENT_PUBLISHER_EVENT_BUS_ARN" = aws_cloudwatch_event_bus.main.arn
43+
"EVENT_PUBLISHER_DLQ_URL" = module.sqs_event_publisher_errors.sqs_queue_url
4244
}
4345
}
4446

@@ -78,11 +80,38 @@ data "aws_iam_policy_document" "ttl_create_lambda" {
7880
"sqs:ReceiveMessage",
7981
"sqs:DeleteMessage",
8082
"sqs:GetQueueAttributes",
81-
"sqs:GetQueueUrl"
83+
"sqs:GetQueueUrl",
8284
]
8385

8486
resources = [
8587
module.sqs_ttl.sqs_queue_arn,
8688
]
8789
}
90+
91+
statement {
92+
sid = "PutEvents"
93+
effect = "Allow"
94+
95+
actions = [
96+
"events:PutEvents",
97+
]
98+
99+
resources = [
100+
aws_cloudwatch_event_bus.main.arn,
101+
]
102+
}
103+
104+
statement {
105+
sid = "SQSPermissionsEventPublisherDLQ"
106+
effect = "Allow"
107+
108+
actions = [
109+
"sqs:SendMessage",
110+
"sqs:SendMessageBatch",
111+
]
112+
113+
resources = [
114+
module.sqs_event_publisher_errors.sqs_queue_arn,
115+
]
116+
}
88117
}

‎infrastructure/terraform/components/dl/module_sqs_event_publisher_errors.tf‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
module "sqs_event_publisher_errors" {
2+
source = "https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.24/terraform-sqs.zip"
3+
4+
aws_account_id = var.aws_account_id
5+
component = local.component
6+
environment = var.environment
7+
project = var.project
8+
region = var.region
9+
name = "event-publisher-errors"
10+
11+
sqs_kms_key_arn = module.kms.key_arn
12+
13+
visibility_timeout_seconds = 60
14+
}

‎lambdas/ttl-create-lambda/src/__tests__/apis/sqs-trigger-lambda.test.ts‎

Lines changed: 169 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,35 +1,47 @@
11
import { createHandler } from 'apis/sqs-trigger-lambda';
22
import type { SQSEvent } from 'aws-lambda';
3-
import { $TtlItem } from 'app/ttl-item-validator';
3+
import { $TtlItemEvent, TtlItemEvent } from 'utils';
44

55
describe('createHandler', () => {
66
let createTtl: any;
7+
let eventPublisher: any;
78
let logger: any;
89
let handler: any;
910

10-
const validItem = {
11-
id: '1',
12-
source: 'src',
13-
specversion: '1',
14-
type: 't',
15-
plane: 'p',
16-
subject: 's',
17-
time: 'now',
18-
datacontenttype: 'json',
19-
dataschema: 'sch',
20-
dataschemaversion: '1',
21-
data: { uri: 'uri' },
11+
const validItem: TtlItemEvent = {
12+
profileversion: '1.0.0',
13+
profilepublished: '2025-10',
14+
id: '550e8400-e29b-41d4-a716-446655440001',
15+
specversion: '1.0',
16+
source: '/nhs/england/notify/production/primary/data-plane/digital-letters',
17+
subject:
18+
'customer/920fca11-596a-4eca-9c47-99f624614658/recipient/769acdd4-6a47-496f-999f-76a6fd2c3959',
19+
type: 'uk.nhs.notify.digital.letters.sent.v1',
20+
time: '2023-06-20T12:00:00Z',
21+
recordedtime: '2023-06-20T12:00:00.250Z',
22+
severitynumber: 2,
23+
traceparent: '00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01',
24+
datacontenttype: 'application/json',
25+
dataschema:
26+
'https://notify.nhs.uk/schemas/events/digital-letters/2025-10/digital-letters.schema.json',
27+
dataschemaversion: '1.0',
28+
severitytext: 'INFO',
29+
data: {
30+
uri: 'https://example.com/ttl/resource',
31+
'digital-letter-id': '123e4567-e89b-12d3-a456-426614174000',
32+
},
2233
};
2334

2435
beforeEach(() => {
2536
createTtl = { send: jest.fn() };
26-
logger = { error: jest.fn(), info: jest.fn() };
27-
handler = createHandler({ createTtl, logger });
37+
eventPublisher = { sendEvents: jest.fn().mockResolvedValue([]) };
38+
logger = { error: jest.fn(), info: jest.fn(), warn: jest.fn() };
39+
handler = createHandler({ createTtl, eventPublisher, logger });
2840
});
2941

3042
it('processes a valid SQS event and returns success', async () => {
3143
jest
32-
.spyOn($TtlItem, 'safeParse')
44+
.spyOn($TtlItemEvent, 'safeParse')
3345
.mockReturnValue({ success: true, data: validItem });
3446
createTtl.send.mockResolvedValue('sent');
3547
const event: SQSEvent = {
@@ -40,6 +52,7 @@ describe('createHandler', () => {
4052

4153
expect(res.batchItemFailures).toEqual([]);
4254
expect(createTtl.send).toHaveBeenCalledWith(validItem);
55+
expect(eventPublisher.sendEvents).toHaveBeenCalledWith([validItem]);
4356
expect(logger.info).toHaveBeenCalledWith({
4457
description: 'Processed SQS Event.',
4558
failed: 0,
@@ -51,7 +64,7 @@ describe('createHandler', () => {
5164
it('handles parse failure and logs error', async () => {
5265
const zodError = { errors: [] } as any;
5366
jest
54-
.spyOn($TtlItem, 'safeParse')
67+
.spyOn($TtlItemEvent, 'safeParse')
5568
.mockReturnValue({ success: false, error: zodError });
5669
const event: SQSEvent = {
5770
Records: [{ body: '{}', messageId: 'msg2' }],
@@ -75,7 +88,7 @@ describe('createHandler', () => {
7588

7689
it('handles createTtl.send failure', async () => {
7790
jest
78-
.spyOn($TtlItem, 'safeParse')
91+
.spyOn($TtlItemEvent, 'safeParse')
7992
.mockReturnValue({ success: true, data: validItem });
8093
createTtl.send.mockResolvedValue('failed');
8194
const event: SQSEvent = {
@@ -94,7 +107,7 @@ describe('createHandler', () => {
94107
});
95108

96109
it('handles thrown error and logs', async () => {
97-
jest.spyOn($TtlItem, 'safeParse').mockImplementation(() => {
110+
jest.spyOn($TtlItemEvent, 'safeParse').mockImplementation(() => {
98111
throw new Error('bad json');
99112
});
100113
const event: SQSEvent = {
@@ -143,4 +156,141 @@ describe('createHandler', () => {
143156

144157
Promise.allSettled = originalAllSettled;
145158
});
159+
160+
it('processes multiple successful events and sends them as a batch', async () => {
161+
jest
162+
.spyOn($TtlItemEvent, 'safeParse')
163+
.mockReturnValue({ success: true, data: validItem });
164+
createTtl.send.mockResolvedValue('sent');
165+
const event: SQSEvent = {
166+
Records: [
167+
{ body: JSON.stringify(validItem), messageId: 'msg1' },
168+
{ body: JSON.stringify(validItem), messageId: 'msg2' },
169+
{ body: JSON.stringify(validItem), messageId: 'msg3' },
170+
],
171+
} as any;
172+
173+
const res = await handler(event);
174+
175+
expect(res.batchItemFailures).toEqual([]);
176+
expect(createTtl.send).toHaveBeenCalledTimes(3);
177+
expect(eventPublisher.sendEvents).toHaveBeenCalledWith([
178+
validItem,
179+
validItem,
180+
validItem,
181+
]);
182+
expect(logger.info).toHaveBeenCalledWith({
183+
description: 'Processed SQS Event.',
184+
failed: 0,
185+
retrieved: 3,
186+
sent: 3,
187+
});
188+
});
189+
190+
it('handles partial event publishing failures and logs warning', async () => {
191+
jest
192+
.spyOn($TtlItemEvent, 'safeParse')
193+
.mockReturnValue({ success: true, data: validItem });
194+
createTtl.send.mockResolvedValue('sent');
195+
const failedEvents = [validItem];
196+
eventPublisher.sendEvents.mockResolvedValue(failedEvents);
197+
198+
const event: SQSEvent = {
199+
Records: [
200+
{ body: JSON.stringify(validItem), messageId: 'msg1' },
201+
{ body: JSON.stringify(validItem), messageId: 'msg2' },
202+
],
203+
} as any;
204+
205+
const res = await handler(event);
206+
207+
expect(res.batchItemFailures).toEqual([]);
208+
expect(eventPublisher.sendEvents).toHaveBeenCalledWith([
209+
validItem,
210+
validItem,
211+
]);
212+
expect(logger.warn).toHaveBeenCalledWith({
213+
description: 'Some events failed to publish',
214+
failedCount: 1,
215+
totalAttempted: 2,
216+
});
217+
});
218+
219+
it('handles event publishing exception and logs warning', async () => {
220+
jest
221+
.spyOn($TtlItemEvent, 'safeParse')
222+
.mockReturnValue({ success: true, data: validItem });
223+
createTtl.send.mockResolvedValue('sent');
224+
const publishError = new Error('EventBridge error');
225+
eventPublisher.sendEvents.mockRejectedValue(publishError);
226+
227+
const event: SQSEvent = {
228+
Records: [{ body: JSON.stringify(validItem), messageId: 'msg1' }],
229+
} as any;
230+
231+
const res = await handler(event);
232+
233+
expect(res.batchItemFailures).toEqual([]);
234+
expect(eventPublisher.sendEvents).toHaveBeenCalledWith([validItem]);
235+
expect(logger.warn).toHaveBeenCalledWith({
236+
err: publishError,
237+
description: 'Failed to send events to EventBridge',
238+
eventCount: 1,
239+
});
240+
});
241+
242+
it('does not call eventPublisher when no successful events', async () => {
243+
jest
244+
.spyOn($TtlItemEvent, 'safeParse')
245+
.mockReturnValue({ success: true, data: validItem });
246+
createTtl.send.mockResolvedValue('failed');
247+
248+
const event: SQSEvent = {
249+
Records: [{ body: JSON.stringify(validItem), messageId: 'msg1' }],
250+
} as any;
251+
252+
const res = await handler(event);
253+
254+
expect(res.batchItemFailures).toEqual([{ itemIdentifier: 'msg1' }]);
255+
expect(eventPublisher.sendEvents).not.toHaveBeenCalled();
256+
expect(logger.info).toHaveBeenCalledWith({
257+
description: 'Processed SQS Event.',
258+
failed: 1,
259+
retrieved: 1,
260+
sent: 0,
261+
});
262+
});
263+
264+
it('handles mixed success and failure scenarios', async () => {
265+
jest
266+
.spyOn($TtlItemEvent, 'safeParse')
267+
.mockReturnValueOnce({ success: true, data: validItem })
268+
.mockReturnValueOnce({ success: false, error: { errors: [] } as any })
269+
.mockReturnValueOnce({ success: true, data: validItem });
270+
createTtl.send
271+
.mockResolvedValueOnce('sent')
272+
.mockResolvedValueOnce('failed');
273+
274+
const event: SQSEvent = {
275+
Records: [
276+
{ body: JSON.stringify(validItem), messageId: 'msg1' },
277+
{ body: '{}', messageId: 'msg2' },
278+
{ body: JSON.stringify(validItem), messageId: 'msg3' },
279+
],
280+
} as any;
281+
282+
const res = await handler(event);
283+
284+
expect(res.batchItemFailures).toEqual([
285+
{ itemIdentifier: 'msg2' },
286+
{ itemIdentifier: 'msg3' },
287+
]);
288+
expect(eventPublisher.sendEvents).toHaveBeenCalledWith([validItem]);
289+
expect(logger.info).toHaveBeenCalledWith({
290+
description: 'Processed SQS Event.',
291+
failed: 2,
292+
retrieved: 3,
293+
sent: 1,
294+
});
295+
});
146296
});

‎lambdas/ttl-create-lambda/src/__tests__/app/create-ttl.test.ts‎

Lines changed: 22 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,33 @@
11
import { CreateTtl } from 'app/create-ttl';
22
import { TtlRepository } from 'infra/ttl-repository';
3-
import { TtlItemEvent } from 'infra/types';
3+
import { TtlItemEvent } from 'utils';
44

55
describe('CreateTtl', () => {
66
let repo: jest.Mocked<TtlRepository>;
77
let logger: any;
88
let createTtl: CreateTtl;
99
const item: TtlItemEvent = {
10-
id: 'id',
11-
source: 'src',
12-
specversion: '1',
13-
type: 't',
14-
plane: 'p',
15-
subject: 's',
16-
time: 'now',
17-
datacontenttype: 'json',
18-
dataschema: 'sch',
19-
dataschemaversion: '1',
20-
data: { uri: 'uri' },
10+
profileversion: '1.0.0',
11+
profilepublished: '2025-10',
12+
id: '550e8400-e29b-41d4-a716-446655440001',
13+
specversion: '1.0',
14+
source: '/nhs/england/notify/production/primary/data-plane/digital-letters',
15+
subject:
16+
'customer/920fca11-596a-4eca-9c47-99f624614658/recipient/769acdd4-6a47-496f-999f-76a6fd2c3959',
17+
type: 'uk.nhs.notify.digital.letters.sent.v1',
18+
time: '2023-06-20T12:00:00Z',
19+
recordedtime: '2023-06-20T12:00:00.250Z',
20+
severitynumber: 2,
21+
traceparent: '00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01',
22+
datacontenttype: 'application/json',
23+
dataschema:
24+
'https://notify.nhs.uk/schemas/events/digital-letters/2025-10/digital-letters.schema.json',
25+
dataschemaversion: '1.0',
26+
severitytext: 'INFO',
27+
data: {
28+
uri: 'https://example.com/ttl/resource',
29+
'digital-letter-id': '123e4567-e89b-12d3-a456-426614174000',
30+
},
2131
};
2232

2333
beforeEach(() => {

0 commit comments

Comments
 (0)