diff --git a/infrastructure/terraform/components/api/module_lambda_supplier_allocator.tf b/infrastructure/terraform/components/api/module_lambda_supplier_allocator.tf index b0354bd88..3ab822ecc 100644 --- a/infrastructure/terraform/components/api/module_lambda_supplier_allocator.tf +++ b/infrastructure/terraform/components/api/module_lambda_supplier_allocator.tf @@ -35,8 +35,9 @@ module "supplier_allocator" { log_subscription_role_arn = local.acct.log_subscription_role_arn lambda_env_vars = merge(local.common_lambda_env_vars, { - UPSERT_LETTERS_QUEUE_URL = module.sqs_letter_updates.sqs_queue_url, - IDEMPOTENCY_TABLE_NAME = aws_dynamodb_table.idempotency.name + UPSERT_LETTERS_QUEUE_URL = module.sqs_letter_updates.sqs_queue_url, + SUPPLIER_ALLOCATOR_DLQ_URL = module.sqs_supplier_allocator.sqs_dlq_url, + IDEMPOTENCY_TABLE_NAME = aws_dynamodb_table.idempotency.name }) } @@ -83,6 +84,19 @@ data "aws_iam_policy_document" "supplier_allocator_lambda" { ] } + statement { + sid = "AllowSupplierAllocatorDLQWrite" + effect = "Allow" + + actions = [ + "sqs:SendMessage" + ] + + resources = [ + module.sqs_supplier_allocator.sqs_dlq_arn + ] + } + statement { sid = "AllowConfigDynamoDBAccess" effect = "Allow" diff --git a/internal/datastore/src/errors/missing-supplier-config-error.ts b/internal/datastore/src/errors/missing-supplier-config-error.ts new file mode 100644 index 000000000..2dac18d4a --- /dev/null +++ b/internal/datastore/src/errors/missing-supplier-config-error.ts @@ -0,0 +1,9 @@ +/** + * Error thrown when a supplier cannot be allocated due to missing supplier config + */ +export default class MissingSupplierConfigError extends Error { + constructor(public readonly message: string) { + super(message); + this.name = "MissingSupplierConfigError"; + } +} diff --git a/internal/datastore/src/index.ts b/internal/datastore/src/index.ts index 10e255fbb..48e963112 100644 --- a/internal/datastore/src/index.ts +++ b/internal/datastore/src/index.ts @@ -9,3 +9,4 @@ export { default as DBHealthcheck } from "./healthcheck"; export { default as LetterAlreadyExistsError } from "./errors/letter-already-exists-error"; export { default as LetterNotFoundError } from "./errors/letter-not-found-error"; export { default as MiNotFoundError } from "./errors/mi-not-found-error"; +export { default as MissingSupplierConfigError } from "./errors/missing-supplier-config-error"; diff --git a/internal/datastore/src/supplier-config-repository.ts b/internal/datastore/src/supplier-config-repository.ts index 0d0492bab..0a8ecf6fc 100644 --- a/internal/datastore/src/supplier-config-repository.ts +++ b/internal/datastore/src/supplier-config-repository.ts @@ -19,6 +19,7 @@ import { VolumeGroup, } from "@nhsdigital/nhs-notify-event-schemas-supplier-config"; import { SupplierConfigEntity } from "./types"; +import MissingSupplierConfigError from "./errors/missing-supplier-config-error"; export type SupplierConfigRepositoryConfig = { supplierConfigTableName: string; @@ -40,7 +41,9 @@ export class SupplierConfigRepository { }), ); if (!result.Item) { - throw new Error(`No letter variant details found for id ${variantId}`); + throw new MissingSupplierConfigError( + `No letter variant details found for id ${variantId}`, + ); } return $LetterVariant.parse(result.Item); @@ -54,7 +57,9 @@ export class SupplierConfigRepository { }), ); if (!result.Item) { - throw new Error(`No volume group details found for id ${groupId}`); + throw new MissingSupplierConfigError( + `No volume group details found for id ${groupId}`, + ); } return $VolumeGroup.parse(result.Item); } @@ -81,7 +86,7 @@ export class SupplierConfigRepository { }), ); if (!result.Items || result.Items.length === 0) { - throw new Error( + throw new MissingSupplierConfigError( `No active supplier allocations found for volume group id ${groupId}`, ); } @@ -99,7 +104,9 @@ export class SupplierConfigRepository { }), ); if (!result.Item) { - throw new Error(`Supplier with id ${supplierId} not found`); + throw new MissingSupplierConfigError( + `Supplier with id ${supplierId} not found`, + ); } suppliers.push($Supplier.parse(result.Item)); } @@ -140,7 +147,9 @@ export class SupplierConfigRepository { }), ); if (!result.Item) { - throw new Error(`No pack specification found for id ${packSpecId}`); + throw new MissingSupplierConfigError( + `No pack specification found for id ${packSpecId}`, + ); } return $PackSpecification.parse(result.Item); } diff --git a/lambdas/supplier-allocator/README.md b/lambdas/supplier-allocator/README.md index f33e1517c..a11189ae9 100644 --- a/lambdas/supplier-allocator/README.md +++ b/lambdas/supplier-allocator/README.md @@ -12,21 +12,25 @@ Consumes `LetterRequestPrepared` events (v1 and v2) from an SQS queue, chooses a 2. Each record body is parsed and validated as either `$LetterRequestPreparedEventV2` or `$LetterRequestPreparedEvent` (v1 fallback). 3. The allocator loads the relevant supplier configuration from `SUPPLIER_CONFIG_TABLE`, including the letter variant, active volume group, candidate suppliers, and compatible pack details. 4. Candidate suppliers are filtered using pack support and daily capacity, then ranked using quota data from `SUPPLIER_QUOTAS_TABLE`. -5. On success, the handler produces an allocation with `allocationStatus.status = "PENDING"`. If allocation cannot be completed, it produces a REJECTED allocation with a failure reason instead of dropping the message. -6. Each record produces a `{ letterEvent, allocationDetails }` message sent to `UPSERT_LETTERS_QUEUE_URL`. -7. After the batch completes, allocation counters are written back to `SUPPLIER_QUOTAS_TABLE`, and only genuine processing failures are returned as `batchItemFailures`. +5. On success, the handler produces an allocation with `allocationStatus.status = "PENDING"` and sends `{ letterEvent, allocationDetails }` to `UPSERT_LETTERS_QUEUE_URL`. +6. If a failure occurs due to missing or invalid supplier config, the original record is sent directly to `SUPPLIER_ALLOCATOR_DLQ_URL` and acknowledged so it is not retried. +7. If supplier config is valid, but the letter request violates the contraints of all the available supplier packs (for example, too many pages), the handler produces a REJECTED allocation with a failure reason instead of dropping the message. +8. Any other processing error is returned in `batchItemFailures` so SQS retries the record based on the source queue redrive policy. +9. After the batch completes, allocation counters are written back to `SUPPLIER_QUOTAS_TABLE`. ## Key Integration Points -- **SQS**: Input from EventSub, output to the upsert-letter queue (`UPSERT_LETTERS_QUEUE`). +- **SQS**: Input from EventSub, output to the upsert-letter queue (`UPSERT_LETTERS_QUEUE`), and direct publish to the allocator DLQ (`SUPPLIER_ALLOCATOR_DLQ_URL`) for `SupplierConfigValidationError` or `MissingSupplierConfigError`. - **`SupplierConfigRepository`** from `@internal/datastore` (`SUPPLIER_CONFIG_TABLE`): reads letter variants, volume groups, supplier allocations, pack specifications, and supplier packs. - **`SupplierQuotasRepository`** from `@internal/datastore` (`SUPPLIER_QUOTAS_TABLE`): reads and writes daily and overall allocation counts per volume group and supplier. - **Event schemas**: `@nhsdigital/nhs-notify-event-schemas-letter-rendering` (v2) and `@nhsdigital/nhs-notify-event-schemas-letter-rendering-v1` (v1). -- **Downstream consumer**: `upsert-letter` receives `{ letterEvent, allocationDetails }` and persists either a PENDING or REJECTED letter. +- **Downstream consumer**: `upsert-letter` receives `{ letterEvent, allocationDetails }` and persists either PENDING or REJECTED letters. ## Nuances and Peculiarities -- **Failed allocations produce REJECTED letters, not dropped messages.** If the config lookup chain fails for any reason, the handler still sends a message to the upsert queue with `allocationStatus.status = "REJECTED"` and `supplierId = "unknown"`. No letters are silently lost. +- **`SupplierConfigValidationError` and `MissingSupplierConfigError` are treated as terminal for retries.** The handler sends the original message directly to the allocator DLQ and acknowledges the source record. +- **Otherwise failed allocations produce REJECTED letters, not dropped messages.** If no valid supplier packs are found, the handler sends a message to the upsert queue with `allocationStatus.status = "REJECTED"` and `supplierId = "unknown"`. +- **All other failures retain normal retry semantics.** Records that fail for non-supplier-config-related reasons are returned in `batchItemFailures` and retried according to queue configuration. - **The factor algorithm is a running weighted average across the lifetime of the system, not per-batch.** The `overallAllocation` table accumulates counts since deployment. A supplier that handled a disproportionate share yesterday will have a high factor today and be deprioritised, allowing others to catch up to their target percentage. - **Daily capacity check uses London timezone.** `format(toZonedTime(new Date(), "Europe/London"), "yyyy-MM-dd")` determines the date key. Capacity resets at midnight London time, not UTC. - **Quota updates happen after the entire batch completes**, not per-record. This optimisation means concurrent Lambda invocations can transiently over-allocate a supplier before quotas are reconciled. diff --git a/lambdas/supplier-allocator/jest.config.ts b/lambdas/supplier-allocator/jest.config.ts index 872794514..174e7f7f9 100644 --- a/lambdas/supplier-allocator/jest.config.ts +++ b/lambdas/supplier-allocator/jest.config.ts @@ -9,6 +9,9 @@ export const baseJestConfig = { }, ], }, + transformIgnorePatterns: [ + "node_modules/(?!(@nhsdigital/nhs-notify-event-schemas-supplier-config)/)", + ], // Automatically clear mock calls, instances, contexts and results before every test clearMocks: true, diff --git a/lambdas/supplier-allocator/src/errors/rejected-error.ts b/lambdas/supplier-allocator/src/errors/rejected-error.ts new file mode 100644 index 000000000..022bade8c --- /dev/null +++ b/lambdas/supplier-allocator/src/errors/rejected-error.ts @@ -0,0 +1,9 @@ +/** + * Error thrown when a letter is rejected due to violating pack constraints for all possible supplier packs. + */ +export default class RejectedError extends Error { + constructor(public readonly message: string) { + super(message); + this.name = "RejectedError"; + } +} diff --git a/lambdas/supplier-allocator/src/errors/supplier-config-validation-error.ts b/lambdas/supplier-allocator/src/errors/supplier-config-validation-error.ts new file mode 100644 index 000000000..8bde4b622 --- /dev/null +++ b/lambdas/supplier-allocator/src/errors/supplier-config-validation-error.ts @@ -0,0 +1,9 @@ +/** + * Error thrown when a supplier cannot be allocated due to incorrect supplier config + */ +export default class SupplierConfigValidationError extends Error { + constructor(public readonly message: string) { + super(message); + this.name = "SupplierConfigValidationError"; + } +} diff --git a/lambdas/supplier-allocator/src/handler/__tests__/allocate-handler.test.ts b/lambdas/supplier-allocator/src/handler/__tests__/allocate-handler.test.ts index 97d6e391a..ba541c3ec 100644 --- a/lambdas/supplier-allocator/src/handler/__tests__/allocate-handler.test.ts +++ b/lambdas/supplier-allocator/src/handler/__tests__/allocate-handler.test.ts @@ -8,12 +8,15 @@ import { LetterStatusChangeEvent, } from "@nhsdigital/nhs-notify-event-schemas-supplier-api/src/events/letter-events"; import { makeIdempotent } from "@aws-lambda-powertools/idempotency"; +import { MissingSupplierConfigError } from "@internal/datastore"; import createSupplierAllocatorHandler from "../allocate-handler"; import * as supplierConfig from "../../services/supplier-config"; import * as supplierQuotas from "../../services/supplier-quotas"; import * as allocationConfig from "../allocation-config"; import { Deps } from "../../config/deps"; +import SupplierConfigValidationError from "../../errors/supplier-config-validation-error"; import packageJson from "../../../package.json"; +import RejectedError from "../../errors/rejected-error"; const renderingSchemaVersion: string = packageJson.dependencies[ @@ -218,6 +221,9 @@ describe("createSupplierAllocatorHandler", () => { } as unknown as Deps; beforeEach(() => { + process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; + process.env.SUPPLIER_ALLOCATOR_DLQ_URL = + "https://sqs.test.queue/supplier-allocator-dlq"; jest.clearAllMocks(); }); @@ -228,7 +234,6 @@ describe("createSupplierAllocatorHandler", () => { ]); setupDefaultMocks(); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; const handler = createSupplierAllocatorHandler(mockedDeps); const result = await handler(evt, {} as any, {} as any); @@ -265,7 +270,6 @@ describe("createSupplierAllocatorHandler", () => { ]); setupDefaultMocks(); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; const handler = createSupplierAllocatorHandler(mockedDeps); const result = await handler(evt, {} as any, {} as any); @@ -299,7 +303,6 @@ describe("createSupplierAllocatorHandler", () => { createSqsRecord("msg1", JSON.stringify(preparedEvent)), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; setupDefaultMocks(); const handler = createSupplierAllocatorHandler(mockedDeps); const result = await handler(evt, {} as any, {} as any); @@ -330,8 +333,6 @@ describe("createSupplierAllocatorHandler", () => { createSqsRecord("invalid-event", JSON.stringify(preparedEvent)), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - const handler = createSupplierAllocatorHandler(mockedDeps); const result = await handler(evt, {} as any, {} as any); @@ -350,8 +351,6 @@ describe("createSupplierAllocatorHandler", () => { createSqsRecord("msg1", JSON.stringify(preparedEvent)), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - const handler = createSupplierAllocatorHandler(mockedDeps); await handler(evt, {} as any, {} as any); @@ -370,8 +369,6 @@ describe("createSupplierAllocatorHandler", () => { createSqsRecord("msg1", JSON.stringify(preparedEvent)), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - const handler = createSupplierAllocatorHandler(mockedDeps); const result = await handler(evt, {} as any, {} as any); @@ -395,8 +392,6 @@ describe("createSupplierAllocatorHandler", () => { ), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - const handler = createSupplierAllocatorHandler(mockedDeps); const result = await handler(evt, {} as any, {} as any); @@ -412,8 +407,6 @@ describe("createSupplierAllocatorHandler", () => { createSqsRecord("bad-json", "this-is-not-json"), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - const handler = createSupplierAllocatorHandler(mockedDeps); const result = await handler(evt, {} as any, {} as any); @@ -432,8 +425,6 @@ describe("createSupplierAllocatorHandler", () => { createSqsRecord("no-type", JSON.stringify(event)), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - const handler = createSupplierAllocatorHandler(mockedDeps); const result = await handler(evt, {} as any, {} as any); if (!result) throw new Error("expected BatchResponse, got void"); @@ -449,6 +440,7 @@ describe("createSupplierAllocatorHandler", () => { createSqsRecord("msg1", JSON.stringify(preparedEvent)), ]); + setupDefaultMocks(); delete process.env.UPSERT_LETTERS_QUEUE_URL; const handler = createSupplierAllocatorHandler(mockedDeps); @@ -471,8 +463,6 @@ describe("createSupplierAllocatorHandler", () => { createSqsRecord("msg1", JSON.stringify(preparedEvent)), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - const sqsError = new Error("SQS send failed"); (mockedDeps.sqsClient.send as jest.Mock).mockRejectedValueOnce(sqsError); @@ -498,8 +488,6 @@ describe("createSupplierAllocatorHandler", () => { ), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - const handler = createSupplierAllocatorHandler(mockedDeps); const result = await handler(evt, {} as any, {} as any); if (!result) throw new Error("expected BatchResponse, got void"); @@ -527,56 +515,13 @@ describe("createSupplierAllocatorHandler", () => { expect(sendCall.input.QueueUrl).toBe(queueUrl); }); - test("logs error when supplier config retrieval fails", async () => { - const preparedEvent = createPreparedV2Event(); - - const evt: SQSEvent = createSQSEvent([ - createSqsRecord("msg1", JSON.stringify(preparedEvent)), - ]); - - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - const configError = new Error("Failed to retrieve supplier config"); - (supplierConfig.getVariantDetails as jest.Mock).mockRejectedValueOnce( - configError, - ); - - const handler = createSupplierAllocatorHandler(mockedDeps); - const result = await handler(evt, {} as any, {} as any); - if (!result) throw new Error("expected BatchResponse, got void"); - expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(1); - expect((mockedDeps.logger.error as jest.Mock).mock.calls[0][0]).toEqual( - expect.objectContaining({ - description: "Error fetching supplier from config", - err: configError, - variantId: "lv1", - }), - ); - expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(1); - const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock.calls[0][0]; - expect(sendCall).toBeInstanceOf(SendMessageCommand); - - const messageBody = JSON.parse(sendCall.input.MessageBody); - expect(messageBody.letterEvent).toEqual(preparedEvent); - expect(messageBody.allocationDetails.supplierSpec).toEqual({ - supplierId: "unknown", - specId: "unknown", - priority: 0, - billingId: "unknown", - }); - expect(messageBody.allocationDetails.allocationStatus).toEqual({ - status: "REJECTED", - reasonCode: "NO_SUPPLIERS_AVAILABLE", - reasonText: "Failed to retrieve supplier config", - }); - }); - const rejectWith = (mock: jest.Mock, errorMessage: string) => mock.mockRejectedValueOnce(new Error(errorMessage)); const throwAny = (mock: jest.Mock) => mock.mockRejectedValueOnce("anything that is not an Error"); - const supplierConfigErrorCases = [ + const nonSupplierConfigValidationErrorCases = [ { name: "getVolumeGroupDetails", errorMessage: "Volume group retrieval failed", @@ -639,15 +584,14 @@ describe("createSupplierAllocatorHandler", () => { }, ]; - test.each(supplierConfigErrorCases)( - "logs error when %s rejects during supplier config resolution", + test.each(nonSupplierConfigValidationErrorCases)( + "returns batch failure when %s rejects with a non-SupplierConfigValidationError", async ({ errorMessage, setup }) => { const preparedEvent = createPreparedV2Event(); const evt: SQSEvent = createSQSEvent([ createSqsRecord("msg1", JSON.stringify(preparedEvent)), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; setup(); const handler = createSupplierAllocatorHandler(mockedDeps); @@ -657,42 +601,26 @@ describe("createSupplierAllocatorHandler", () => { expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(1); expect((mockedDeps.logger.error as jest.Mock).mock.calls[0][0]).toEqual( expect.objectContaining({ - description: "Error fetching supplier from config", - variantId: "lv1", + description: "Error processing allocation of record", }), ); - expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(1); - const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock - .calls[0][0]; - expect(sendCall).toBeInstanceOf(SendMessageCommand); + expect(result.batchItemFailures).toHaveLength(1); + expect(result.batchItemFailures[0].itemIdentifier).toBe("msg1"); + expect(mockedDeps.sqsClient.send).not.toHaveBeenCalled(); - const messageBody = JSON.parse(sendCall.input.MessageBody); - expect(messageBody.letterEvent).toEqual(preparedEvent); - expect(messageBody.allocationDetails.supplierSpec).toEqual({ - supplierId: "unknown", - specId: "unknown", - priority: 0, - billingId: "unknown", - }); - expect(messageBody.allocationDetails.allocationStatus).toEqual({ - status: "REJECTED", - reasonCode: "NO_SUPPLIERS_AVAILABLE", - reasonText: errorMessage, - }); + expect(errorMessage).toBeDefined(); }, ); - test("returns batch failure when no suppliers are found for pack specification", async () => { + it("rejects the letter when a RejectedError is thrown", async () => { const preparedEvent = createPreparedV2Event(); const evt: SQSEvent = createSQSEvent([ createSqsRecord("msg1", JSON.stringify(preparedEvent)), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - setupDefaultMocks(); - (allocationConfig.suppliersWithValidPack as jest.Mock).mockResolvedValue( - [], + (allocationConfig.preferredSupplierPack as jest.Mock).mockRejectedValueOnce( + new RejectedError("No eligible packs found"), ); const handler = createSupplierAllocatorHandler(mockedDeps); @@ -702,8 +630,7 @@ describe("createSupplierAllocatorHandler", () => { expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(1); expect((mockedDeps.logger.error as jest.Mock).mock.calls[0][0]).toEqual( expect.objectContaining({ - description: "Error fetching supplier from config", - variantId: "lv1", + description: "Letter request rejected", }), ); expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(1); @@ -721,7 +648,130 @@ describe("createSupplierAllocatorHandler", () => { expect(messageBody.allocationDetails.allocationStatus).toEqual({ status: "REJECTED", reasonCode: "NO_SUPPLIERS_AVAILABLE", - reasonText: "No suppliers found for pack specification spec1", + reasonText: "No eligible packs found", + }); + }); + + describe("Dead letter queue", () => { + it("places the record on the DLQ when no suppliers are found for pack specification", async () => { + const preparedEvent = createPreparedV2Event(); + const evt: SQSEvent = createSQSEvent([ + createSqsRecord("msg1", JSON.stringify(preparedEvent)), + ]); + + setupDefaultMocks(); + (allocationConfig.suppliersWithValidPack as jest.Mock).mockResolvedValue( + [], + ); + + const handler = createSupplierAllocatorHandler(mockedDeps); + const result = await handler(evt, {} as any, {} as any); + if (!result) throw new Error("expected BatchResponse, got void"); + + expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(1); + expect((mockedDeps.logger.error as jest.Mock).mock.calls[0][0]).toEqual( + expect.objectContaining({ + description: "Error processing allocation of record", + }), + ); + + expect(result.batchItemFailures).toHaveLength(0); + expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(1); + + const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock + .calls[0][0]; + expect(sendCall).toBeInstanceOf(SendMessageCommand); + expect(sendCall.input.QueueUrl).toBe( + "https://sqs.test.queue/supplier-allocator-dlq", + ); + expect(sendCall.input.MessageBody).toBe(JSON.stringify(preparedEvent)); + }); + + test.each([ + new SupplierConfigValidationError("Failed to retrieve supplier config"), + new MissingSupplierConfigError("Failed to retrieve supplier config"), + ])( + "places the record on the DLQ when supplier config retrieval fails", + async (configError: Error) => { + const preparedEvent = createPreparedV2Event(); + + const evt: SQSEvent = createSQSEvent([ + createSqsRecord("msg1", JSON.stringify(preparedEvent)), + ]); + + (supplierConfig.getVariantDetails as jest.Mock).mockRejectedValueOnce( + configError, + ); + + const handler = createSupplierAllocatorHandler(mockedDeps); + const result = await handler(evt, {} as any, {} as any); + if (!result) throw new Error("expected BatchResponse, got void"); + expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength( + 1, + ); + expect((mockedDeps.logger.error as jest.Mock).mock.calls[0][0]).toEqual( + expect.objectContaining({ + description: "Error processing allocation of record", + err: configError, + }), + ); + expect(result.batchItemFailures).toHaveLength(0); + expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(1); + const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock + .calls[0][0]; + expect(sendCall.input.QueueUrl).toBe( + "https://sqs.test.queue/supplier-allocator-dlq", + ); + expect(sendCall.input.MessageBody).toBe(JSON.stringify(preparedEvent)); + }, + ); + + it("returns batch failure when DLQ not configured", async () => { + const preparedEvent = createPreparedV2Event(); + const evt: SQSEvent = createSQSEvent([ + createSqsRecord("msg1", JSON.stringify(preparedEvent)), + ]); + + const configError = new SupplierConfigValidationError( + "Failed to retrieve supplier config", + ); + (supplierConfig.getVariantDetails as jest.Mock).mockRejectedValueOnce( + configError, + ); + + setupDefaultMocks(); + delete process.env.SUPPLIER_ALLOCATOR_DLQ_URL; + + const handler = createSupplierAllocatorHandler(mockedDeps); + const result = await handler(evt, {} as any, {} as any); + if (!result) throw new Error("expected BatchResponse, got void"); + + expect(result.batchItemFailures).toHaveLength(1); + expect(result.batchItemFailures[0].itemIdentifier).toBe("msg1"); + expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(2); + }); + + it("returns batch failure when sending to DLQ fails", async () => { + const preparedEvent = createPreparedV2Event(); + const evt: SQSEvent = createSQSEvent([ + createSqsRecord("msg1", JSON.stringify(preparedEvent)), + ]); + + setupDefaultMocks(); + (allocationConfig.suppliersWithValidPack as jest.Mock).mockResolvedValue( + [], + ); + (mockedDeps.sqsClient.send as jest.Mock).mockRejectedValueOnce( + new Error("DLQ send failed"), + ); + + const handler = createSupplierAllocatorHandler(mockedDeps); + const result = await handler(evt, {} as any, {} as any); + if (!result) throw new Error("expected BatchResponse, got void"); + + expect(result.batchItemFailures).toHaveLength(1); + expect(result.batchItemFailures[0].itemIdentifier).toBe("msg1"); + expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(2); }); }); @@ -730,7 +780,6 @@ describe("createSupplierAllocatorHandler", () => { ( allocationConfig.filterSuppliersWithCapacity as jest.Mock ).mockResolvedValue([]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; const evt: SQSEvent = createSQSEvent([ createSqsRecord("msg1", JSON.stringify(createPreparedV2Event())), @@ -754,7 +803,6 @@ describe("createSupplierAllocatorHandler", () => { (allocationConfig.selectSupplierByFactor as jest.Mock) .mockResolvedValueOnce(null) .mockResolvedValueOnce("supplier1"); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; const evt: SQSEvent = createSQSEvent([ createSqsRecord("msg1", JSON.stringify(createPreparedV2Event())), @@ -774,8 +822,6 @@ describe("createSupplierAllocatorHandler", () => { createSqsRecord("msg1", JSON.stringify(preparedEvent)), ]); - process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue"; - setupDefaultMocks(); (makeIdempotent as jest.Mock).mockImplementationOnce((_fn) => "supplier1"); diff --git a/lambdas/supplier-allocator/src/handler/__tests__/allocation-config.test.ts b/lambdas/supplier-allocator/src/handler/__tests__/allocation-config.test.ts index ef3e576f2..eca009e65 100644 --- a/lambdas/supplier-allocator/src/handler/__tests__/allocation-config.test.ts +++ b/lambdas/supplier-allocator/src/handler/__tests__/allocation-config.test.ts @@ -21,6 +21,15 @@ import * as supplierQuotasService from "../../services/supplier-quotas"; jest.mock("../../services/supplier-config"); jest.mock("../../services/supplier-quotas"); +async function expectSupplierConfigValidationError( + promise: Promise, + message: string | RegExp, +): Promise { + await expect(promise).rejects.toThrow(message); + await expect(promise).rejects.toMatchObject({ + name: "SupplierConfigValidationError", + }); +} describe("eligibleSuppliers", () => { let mockDeps: jest.Mocked; let mockVolumeGroup: VolumeGroup; @@ -970,14 +979,14 @@ describe("selectSupplierByFactor", () => { } as SupplierAllocation, ]; - await expect( + await expectSupplierConfigValidationError( selectSupplierByFactor( mockSuppliers, zeroAllocations, domainId, mockDeps, ), - ).rejects.toThrow( + "No valid supplier allocations found for suppliers with valid pack", ); }); @@ -1155,13 +1164,14 @@ describe("selectSupplierByFactor", () => { supplierQuotasService.calculateSupplierAllocatedFactor as jest.Mock ).mockResolvedValue([]); - await expect( + await expectSupplierConfigValidationError( selectSupplierByFactor( mockSuppliers, mockSupplierAllocations, domainId, mockDeps, ), - ).rejects.toThrow("No supplier factors could be calculated for allocation"); + "No supplier factors could be calculated for allocation", + ); }); }); diff --git a/lambdas/supplier-allocator/src/handler/allocate-handler.ts b/lambdas/supplier-allocator/src/handler/allocate-handler.ts index 795814937..a4dd059f0 100644 --- a/lambdas/supplier-allocator/src/handler/allocate-handler.ts +++ b/lambdas/supplier-allocator/src/handler/allocate-handler.ts @@ -1,4 +1,10 @@ -import { Context, SQSBatchItemFailure, SQSEvent, SQSHandler } from "aws-lambda"; +import { + Context, + SQSBatchItemFailure, + SQSEvent, + SQSHandler, + SQSRecord, +} from "aws-lambda"; import { SendMessageCommand } from "@aws-sdk/client-sqs"; import { LetterVariant, @@ -17,6 +23,7 @@ import { IdempotencyConfig, makeIdempotent, } from "@aws-lambda-powertools/idempotency"; +import { MissingSupplierConfigError } from "@internal/datastore"; import { getVariantDetails, getVolumeGroupDetails, @@ -31,6 +38,8 @@ import { } from "./allocation-config"; import { Deps } from "../config/deps"; import { PreparedEventSchema, PreparedEvents, SupplierDetails } from "./types"; +import SupplierConfigValidationError from "../errors/supplier-config-validation-error"; +import RejectedError from "../errors/rejected-error"; const idempotencyConfig = new IdempotencyConfig({ eventKeyJmesPath: "data.domainId", @@ -81,20 +90,20 @@ async function getSupplierFromConfig( letterEvent: PreparedEvents, deps: Deps, ): Promise { - try { - const letterVariant: LetterVariant = await getVariantDetails( - letterEvent.data.letterVariantId, - deps, - ); + const letterVariant: LetterVariant = await getVariantDetails( + letterEvent.data.letterVariantId, + deps, + ); - const volumeGroup: VolumeGroup = await getVolumeGroupDetails( - letterVariant.volumeGroupId, - deps, - ); + const volumeGroup: VolumeGroup = await getVolumeGroupDetails( + letterVariant.volumeGroupId, + deps, + ); - const { supplierAllocations, suppliers: allocatedSuppliers } = - await eligibleSuppliers(volumeGroup, deps, letterVariant.supplierId); + const { supplierAllocations, suppliers: allocatedSuppliers } = + await eligibleSuppliers(volumeGroup, deps, letterVariant.supplierId); + try { const preferredPack: PackSpecification = await preferredSupplierPack( letterEvent, allocatedSuppliers, @@ -109,7 +118,7 @@ async function getSupplierFromConfig( ); if (allSuppliersForPack.length === 0) { - throw new Error( + throw new SupplierConfigValidationError( `No suppliers found for pack specification ${preferredPack.id}`, ); } @@ -158,21 +167,24 @@ async function getSupplierFromConfig( volumeGroup.id, ); } catch (error) { - deps.logger.error({ - description: "Error fetching supplier from config", - err: error, - variantId: letterEvent.data.letterVariantId, - }); - return buildSupplierDetails( - "unknown", - "unknown", - "unknown", - 0, - "REJECTED", - "unknown", - "NO_SUPPLIERS_AVAILABLE", - error instanceof Error ? error.message : "Unknown error", - ); + if (error instanceof RejectedError) { + deps.logger.error({ + description: "Letter request rejected", + err: error, + variantId: letterEvent.data.letterVariantId, + }); + return buildSupplierDetails( + "unknown", + "unknown", + "unknown", + 0, + "REJECTED", + "unknown", + "NO_SUPPLIERS_AVAILABLE", + error.message, + ); + } + throw error; } } @@ -278,7 +290,6 @@ async function processSupplierAllocation( letterEvent: PreparedEvents, deps: Deps, perAllocationSuccess: AllocationMetrics, - perAllocationFailure: AllocationMetrics, volumeGroupAllocations: VolumeGroupAllocation, ): Promise { const supplierDetails: SupplierDetails = await getSupplierFromConfig( @@ -294,20 +305,16 @@ async function processSupplierAllocation( const supplier = supplierSpec.supplierId; const priority = String(supplierSpec.priority); - if (supplierDetails.allocationDetails.allocationStatus.status === "PENDING") { - incrementMetric(perAllocationSuccess, supplier, priority); - emitDataMetrics(letterEvent, supplier, "extra_data_dimensions", deps); + incrementMetric(perAllocationSuccess, supplier, priority); + emitDataMetrics(letterEvent, supplier, "extra_data_dimensions", deps); - incrementAllocation( - volumeGroupAllocations, - supplierDetails.volumeGroupId, - supplier, - 1, - deps, - ); - } else { - incrementMetric(perAllocationFailure, supplier, priority); - } + incrementAllocation( + volumeGroupAllocations, + supplierDetails.volumeGroupId, + supplier, + 1, + deps, + ); // Send to allocated letters queue const queueUrl = process.env.UPSERT_LETTERS_QUEUE_URL; @@ -338,10 +345,29 @@ async function processSupplierAllocation( }; } +async function placeOnDeadLetterQueue(record: SQSRecord, deps: Deps) { + const deadLetterQueueUrl = process.env.SUPPLIER_ALLOCATOR_DLQ_URL; + if (!deadLetterQueueUrl) { + throw new Error("SUPPLIER_ALLOCATOR_DLQ_URL not configured"); + } + + deps.logger.info({ + description: "Sending record to supplier allocator DLQ", + messageId: record.messageId, + deadLetterQueueUrl, + }); + + await deps.sqsClient.send( + new SendMessageCommand({ + QueueUrl: deadLetterQueueUrl, + MessageBody: record.body, + }), + ); +} + export default function createSupplierAllocatorHandler(deps: Deps): SQSHandler { const createGetSupplierIdempotently = ( perAllocationSuccess: AllocationMetrics, - perAllocationFailure: AllocationMetrics, volumeGroupAllocations: VolumeGroupAllocation, ) => { return makeIdempotent( @@ -350,7 +376,6 @@ export default function createSupplierAllocatorHandler(deps: Deps): SQSHandler { letterEvent, depsInner, perAllocationSuccess, - perAllocationFailure, volumeGroupAllocations, ), { @@ -368,7 +393,6 @@ export default function createSupplierAllocatorHandler(deps: Deps): SQSHandler { // create an idempotent function bound to this handler's global variables to track metrics and allocations const getSupplierIdempotently = createGetSupplierIdempotently( perAllocationSuccess, - perAllocationFailure, volumeGroupAllocations, ); @@ -400,7 +424,24 @@ export default function createSupplierAllocatorHandler(deps: Deps): SQSHandler { message: record.body, }); incrementMetric(perAllocationFailure, supplier, priority); - batchItemFailures.push({ itemIdentifier: record.messageId }); + if ( + error instanceof SupplierConfigValidationError || + error instanceof MissingSupplierConfigError + ) { + try { + await placeOnDeadLetterQueue(record, deps); + } catch (dlqError) { + deps.logger.error({ + description: "Failed to send record to supplier allocator DLQ", + err: dlqError, + messageId: record.messageId, + message: record.body, + }); + batchItemFailures.push({ itemIdentifier: record.messageId }); + } + } else { + batchItemFailures.push({ itemIdentifier: record.messageId }); + } } }); diff --git a/lambdas/supplier-allocator/src/handler/allocation-config.ts b/lambdas/supplier-allocator/src/handler/allocation-config.ts index f9e7f92d1..6a31b54dd 100644 --- a/lambdas/supplier-allocator/src/handler/allocation-config.ts +++ b/lambdas/supplier-allocator/src/handler/allocation-config.ts @@ -19,6 +19,7 @@ import { calculateSupplierAllocatedFactor } from "../services/supplier-quotas"; import { Deps } from "../config/deps"; import { PreparedEvents } from "./types"; +import SupplierConfigValidationError from "../errors/supplier-config-validation-error"; export async function eligibleSuppliers( volumeGroup: VolumeGroup, @@ -163,7 +164,7 @@ export async function selectSupplierByFactor( return suppliers.some((supplier) => supplier.id === alloc.supplier); }); if (supplierAllocationsForPack.length === 0) { - throw new Error( + throw new SupplierConfigValidationError( "No valid supplier allocations found for suppliers with valid pack", ); } @@ -171,7 +172,9 @@ export async function selectSupplierByFactor( await calculateSupplierAllocatedFactor(supplierAllocationsForPack, deps); if (supplierFactors.length === 0) { - throw new Error("No supplier factors could be calculated for allocation"); + throw new SupplierConfigValidationError( + "No supplier factors could be calculated for allocation", + ); } deps.logger.info({ diff --git a/lambdas/supplier-allocator/src/services/__tests__/supplier-config.test.ts b/lambdas/supplier-allocator/src/services/__tests__/supplier-config.test.ts index 1f6807dba..1bd5fb9c3 100644 --- a/lambdas/supplier-allocator/src/services/__tests__/supplier-config.test.ts +++ b/lambdas/supplier-allocator/src/services/__tests__/supplier-config.test.ts @@ -31,6 +31,26 @@ function makeDeps(overrides: Partial = {}): Deps { return { ...(base as Deps), ...overrides }; } +async function expectSupplierConfigValidationError( + promise: Promise, + message: string | RegExp, +): Promise { + await expect(promise).rejects.toThrow(message); + await expect(promise).rejects.toMatchObject({ + name: "SupplierConfigValidationError", + }); +} + +async function expectRejectedError( + promise: Promise, + message: string | RegExp, +): Promise { + await expect(promise).rejects.toThrow(message); + await expect(promise).rejects.toMatchObject({ + name: "RejectedError", + }); +} + describe("supplier-config service", () => { afterEach(() => jest.resetAllMocks()); @@ -88,7 +108,8 @@ describe("supplier-config service", () => { .fn() .mockResolvedValue(group); - await expect(getVolumeGroupDetails("g2", deps)).rejects.toThrow( + await expectSupplierConfigValidationError( + getVolumeGroupDetails("g2", deps), /not active/, ); expect(deps.logger.error).toHaveBeenCalled(); @@ -102,7 +123,8 @@ describe("supplier-config service", () => { .fn() .mockResolvedValue(group); - await expect(getVolumeGroupDetails("g3", deps)).rejects.toThrow( + await expectSupplierConfigValidationError( + getVolumeGroupDetails("g3", deps), /not active/, ); expect(deps.logger.error).toHaveBeenCalled(); @@ -121,7 +143,8 @@ describe("supplier-config service", () => { .fn() .mockResolvedValue(group); - await expect(getVolumeGroupDetails("g3", deps)).rejects.toThrow( + await expectSupplierConfigValidationError( + getVolumeGroupDetails("g3", deps), /not active/, ); expect(deps.logger.error).toHaveBeenCalled(); @@ -183,9 +206,10 @@ describe("supplier-config service", () => { .fn() .mockResolvedValue(allocations); - await expect( + await expectSupplierConfigValidationError( getSupplierAllocationsForVolumeGroup("g1", deps, "missing"), - ).rejects.toThrow(/No supplier allocations found/); + /No supplier allocations found/, + ); expect(deps.logger.error).toHaveBeenCalled(); }); }); @@ -218,7 +242,8 @@ describe("supplier-config service", () => { .fn() .mockResolvedValue([]); - await expect(getSupplierDetails(supplierIds, deps)).rejects.toThrow( + await expectSupplierConfigValidationError( + getSupplierDetails(supplierIds, deps), /No supplier details found/, ); }); @@ -293,7 +318,8 @@ describe("supplier-config service", () => { .fn() .mockResolvedValue(suppliers); - await expect(getSupplierDetails(supplierIds, deps)).rejects.toThrow( + await expectSupplierConfigValidationError( + getSupplierDetails(supplierIds, deps), /No active suppliers found/, ); expect(deps.logger.error).toHaveBeenCalledWith( @@ -360,9 +386,10 @@ describe("supplier-config service", () => { .fn() .mockResolvedValue([]); - await expect( + await expectSupplierConfigValidationError( getPreferredSupplierPacks(["spec1"], suppliers, deps), - ).rejects.toThrow(/No preferred supplier packs found/); + /No preferred supplier packs found/, + ); expect(deps.logger.error).toHaveBeenCalledWith( expect.objectContaining({ description: @@ -409,9 +436,10 @@ describe("supplier-config service", () => { .fn() .mockResolvedValue(supplierPacks); - await expect( + await expectSupplierConfigValidationError( getPreferredSupplierPacks(["spec1"], suppliers, deps), - ).rejects.toThrow(/No preferred supplier packs found/); + /No preferred supplier packs found/, + ); expect(deps.logger.error).toHaveBeenCalledWith( expect.objectContaining({ description: @@ -449,7 +477,8 @@ describe("supplier-config service", () => { .fn() .mockResolvedValue(packSpec); - await expect(getPackSpecification("spec2", deps)).rejects.toThrow( + await expectSupplierConfigValidationError( + getPackSpecification("spec2", deps), /not active/, ); expect(deps.logger.error).toHaveBeenCalledWith( @@ -500,9 +529,8 @@ describe("supplier-config service", () => { }, } as any; - await expect( + await expectRejectedError( filterPacksForLetter(letterEvent, ["spec1"], deps), - ).rejects.toThrow( "No eligible pack specifications found for letter variant id undefined and pack specification ids spec1", ); expect(deps.logger.info).toHaveBeenCalledWith({ @@ -571,9 +599,10 @@ describe("supplier-config service", () => { }, } as any; - await expect( + await expectRejectedError( filterPacksForLetter(letterEvent, ["spec1"], deps), - ).rejects.toThrow(/No eligible pack specifications found/); + /No eligible pack specifications found/, + ); expect(deps.logger.info).toHaveBeenCalledWith({ description: @@ -611,9 +640,10 @@ describe("supplier-config service", () => { }, } as any; - await expect( + await expectRejectedError( filterPacksForLetter(letterEvent, ["spec1"], deps), - ).rejects.toThrow(/No eligible pack specifications found/); + /No eligible pack specifications found/, + ); expect(deps.logger.info).toHaveBeenCalledWith({ description: @@ -731,9 +761,8 @@ describe("supplier-config service", () => { }, } as any; - await expect( + await expectSupplierConfigValidationError( filterPacksForLetter(letterEvent, ["spec1"], deps), - ).rejects.toThrow( "Unsupported operator UNSUPPORTED_OP in pack specification constraints", ); }); diff --git a/lambdas/supplier-allocator/src/services/supplier-config.ts b/lambdas/supplier-allocator/src/services/supplier-config.ts index 7ed2509bd..dcb71abc1 100644 --- a/lambdas/supplier-allocator/src/services/supplier-config.ts +++ b/lambdas/supplier-allocator/src/services/supplier-config.ts @@ -9,6 +9,8 @@ import { import { Deps } from "../config/deps"; import { PreparedEvents } from "../handler/types"; +import SupplierConfigValidationError from "../errors/supplier-config-validation-error"; +import RejectedError from "../errors/rejected-error"; export async function getVariantDetails( variantId: string, @@ -46,7 +48,9 @@ export async function getVolumeGroupDetails( startDate: groupDetails.startDate, endDate: groupDetails.endDate, }); - throw new Error(`Volume group with id ${groupId} is not active`); + throw new SupplierConfigValidationError( + `Volume group with id ${groupId} is not active`, + ); } export async function getSupplierAllocationsForVolumeGroup( @@ -68,7 +72,7 @@ export async function getSupplierAllocationsForVolumeGroup( groupId, supplierId, }); - throw new Error( + throw new SupplierConfigValidationError( `No supplier allocations found for variant supplier id ${supplierId} in volume group ${groupId}`, ); } @@ -90,7 +94,7 @@ export async function getSupplierDetails( description: "No supplier details found for supplier allocations", supplierIds, }); - throw new Error( + throw new SupplierConfigValidationError( `No supplier details found for supplier ids ${supplierIds.join(", ")}`, ); } @@ -113,7 +117,7 @@ export async function getSupplierDetails( description: "No active suppliers found for supplier allocations", supplierIds, }); - throw new Error( + throw new SupplierConfigValidationError( `No active suppliers found for supplier ids ${supplierIds.join(", ")}`, ); } @@ -145,7 +149,7 @@ export async function getPreferredSupplierPacks( packSpecificationIds, supplierIds: suppliers.map((s) => s.id), }); - throw new Error( + throw new SupplierConfigValidationError( `No preferred supplier packs found for pack specification ids ${packSpecificationIds.join(", ")} and suppliers ${suppliers.map((s) => s.id).join(", ")}`, ); } @@ -162,7 +166,9 @@ export async function getPackSpecification( packSpecId, status: packSpec.status, }); - throw new Error(`Pack specification with id ${packSpecId} is not active`); + throw new SupplierConfigValidationError( + `Pack specification with id ${packSpecId} is not active`, + ); } return packSpec; } @@ -203,7 +209,7 @@ function evaluateContraint( return actualValue <= constraintValue; } default: { - throw new Error( + throw new SupplierConfigValidationError( `Unsupported operator ${operator} in pack specification constraints`, ); } @@ -301,7 +307,7 @@ export async function filterPacksForLetter( if (violatedConstraints.length > 0) { deps.logger.info({ description: `Pack specification filtered out based on pageCount constraints`, - dommainId: letterEvent.data.domainId, + domainId: letterEvent.data.domainId, packSpecId, pageCount, violatedConstraints, @@ -322,7 +328,7 @@ export async function filterPacksForLetter( letterVariantId: letterEvent.data.letterVariantId, packSpecificationIds, }); - throw new Error( + throw new RejectedError( `No eligible pack specifications found for letter variant id ${letterEvent.data.letterVariantId} and pack specification ids ${packSpecificationIds.join(", ")}`, ); } diff --git a/package-lock.json b/package-lock.json index ec6854450..c2c3f6e7d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -24617,6 +24617,7 @@ "@aws-sdk/client-kinesis": "^3.964.0", "@aws-sdk/client-lambda": "^3.986.0", "@aws-sdk/client-sns": "^3.1044.0", + "@aws-sdk/client-sqs": "^3.1044.0", "@aws-sdk/lib-dynamodb": "^3.1044.0", "@nhsdigital/nhs-notify-event-schemas-letter-rendering": "^2.0.2", "@nhsdigital/notify-digital-letters-consumer-contracts": "^1.0.1", @@ -24629,6 +24630,7 @@ "graphql": "^16.11.0", "graphql-tag": "^2.12.6", "md5": "^2.3.0", + "nhs-notify-supplier-api-allocate-letter": "^0.0.1", "nhs-notify-supplier-api-upsert-letter": "^0.0.1", "openapi-response-validator": "^12.1.3", "pino": "^10.3.0", diff --git a/tests/component-tests/allocation-tests/allocation-target-percentage.spec.ts b/tests/component-tests/allocation-tests/allocation-target-percentage.spec.ts index 65bc5d6db..0a627bb4d 100644 --- a/tests/component-tests/allocation-tests/allocation-target-percentage.spec.ts +++ b/tests/component-tests/allocation-tests/allocation-target-percentage.spec.ts @@ -8,6 +8,7 @@ import { updateSupplierAllocation, updateVolumeGroupData, } from "tests/helpers/allocation-helper"; +import { pollQueueForLetterEvent } from "tests/helpers/aws-queue-helper"; import { createPreparedV1Event } from "tests/helpers/event-fixtures"; import { getLettersFromSupplierTable } from "tests/helpers/generate-fetch-test-data"; import { sendSnsEvent } from "tests/helpers/send-sns-event"; @@ -44,21 +45,7 @@ test.describe("Allocation Target Percentage Tests", () => { const response = await sendSnsEvent(preparedEvent); expect(response.MessageId).toBeTruthy(); - const allocationLog = await getAllocationLogForDomainId(domainId); - const lettersInDb = await getLettersFromSupplierTable( - "unknown", - domainId, - "REJECTED", - ); - - expect(lettersInDb.status).toBe("REJECTED"); - expect(lettersInDb.supplierId).toBe( - allocationLog.msg?.allocationDetails?.supplierSpec?.supplierId, - ); - expect(lettersInDb.reasonCode).toBe("NO_SUPPLIERS_AVAILABLE"); - expect(lettersInDb.reasonText).toBe( - `No valid supplier allocations found for suppliers with valid pack`, - ); + await pollQueueForLetterEvent("supplier-allocator-dlq", domainId); }); test("Verify that supplier with less than 100 target percentage is handled correctly", async () => { diff --git a/tests/component-tests/allocation-tests/letter-allocation-rejected.spec.ts b/tests/component-tests/allocation-tests/letter-allocation-rejected.spec.ts index bcf4a6182..3220a348d 100644 --- a/tests/component-tests/allocation-tests/letter-allocation-rejected.spec.ts +++ b/tests/component-tests/allocation-tests/letter-allocation-rejected.spec.ts @@ -9,6 +9,7 @@ import { updateLetterVariantPackSpecs, updateVolumeGroupData, } from "tests/helpers/allocation-helper"; +import { pollQueueForLetterEvent } from "tests/helpers/aws-queue-helper"; import { createPreparedV1Event } from "tests/helpers/event-fixtures"; import { getLettersFromSupplierTable } from "tests/helpers/generate-fetch-test-data"; import { logger } from "tests/helpers/pino-logger"; @@ -16,116 +17,93 @@ import { sendSnsEvent } from "tests/helpers/send-sns-event"; test.describe("Allocator Rejected Allocation Tests", () => { test.setTimeout(180_000); // 3 minutes for long running polling - for (const { - domainIdName, - expectedError, - letterVariantMapping, - pageCount, - testCase, - testName, - } of [ - { - testCase: 1, - testName: - "Verify that the letters are REJECTED when no pack specification is eligible", - letterVariantMapping: 1, - domainIdName: "NoEligiblePackSpecs", + + test("Verify that the letters are REJECTED when no pack specification is eligible", async () => { + const letterVariant = getVariantsForAllocation(1); + const domainId = `NoEligiblePackSpecs-${randomUUID()}`; + const preparedEvent = createPreparedV1Event({ + domainId, + letterVariantId: letterVariant, pageCount: 100, // high page count to ensure pack specifications are filtered out based on constraints - expectedError: "No eligible pack specifications found for letter", - }, - { - testCase: 2, - testName: - "Verify that the letters are REJECTED when no supplier pack are found for selected pack", - letterVariantMapping: 6, - domainIdName: "NoSupplierPacksFound", - pageCount: 2, - expectedError: - "No preferred supplier packs found for pack specification ids and suppliers", - }, - { - testCase: 3, - testName: - "Verify that the letters are REJECTED when no pack specification found for letter variant", - letterVariantMapping: 7, - domainIdName: "NoPackSpecificationFound", + }); + + const response = await sendSnsEvent(preparedEvent); + expect(response.MessageId).toBeTruthy(); + + const supplierAllocatorLog = await getAllocationLog( + "No eligible pack specifications found for letter", + ); + + const allocationLog = await getAllocationLogForDomainId(domainId); + const lettersInDb = await getLettersFromSupplierTable( + "unknown", + domainId, + "REJECTED", + ); + + expect(lettersInDb.status).toBe("REJECTED"); + expect(lettersInDb.supplierId).toBe( + allocationLog.msg?.allocationDetails?.supplierSpec?.supplierId, + ); + + const { packSpecificationIds } = supplierAllocatorLog; + expect(packSpecificationIds).toBeTruthy(); + }); + + test("Verify that the letters are placed on a DLQ when no supplier packs are found", async () => { + const letterVariant = getVariantsForAllocation(6); + const domainId = `NoSupplierPacksFound-${randomUUID()}`; + const preparedEvent = createPreparedV1Event({ + domainId, + letterVariantId: letterVariant, pageCount: 2, - expectedError: "No pack specification found for id", - }, - ]) { - test(testName, async () => { - const letterVariant = getVariantsForAllocation(letterVariantMapping); - const domainId = `${domainIdName}-${randomUUID()}`; - const preparedEvent = createPreparedV1Event({ - domainId, - letterVariantId: letterVariant, - pageCount, - }); + }); - if (letterVariantMapping === 7) { - await updateLetterVariantPackSpecs(letterVariant, [""]); - } + const response = await sendSnsEvent(preparedEvent); + expect(response.MessageId).toBeTruthy(); - const response = await sendSnsEvent(preparedEvent); - expect(response.MessageId).toBeTruthy(); + const supplierAllocatorLog = await getAllocationLog( + "No preferred supplier packs found for pack specification ids and suppliers", + ); - const supplierAllocatorLog = - await getAllocationLog(expectedError); + await pollQueueForLetterEvent("supplier-allocator-dlq", domainId); - const allocationLog = await getAllocationLogForDomainId(domainId); - const lettersInDb = await getLettersFromSupplierTable( - "unknown", - domainId, - "REJECTED", - ); + const { packSpecificationIds } = supplierAllocatorLog; + expect(packSpecificationIds).toBeTruthy(); + }); - expect(lettersInDb.status).toBe("REJECTED"); - expect(lettersInDb.supplierId).toBe( - allocationLog.msg?.allocationDetails?.supplierSpec?.supplierId, - ); - switch (testCase) { - case 1: { - const { packSpecificationIds } = supplierAllocatorLog; - expect(packSpecificationIds).toBeTruthy(); - expect(lettersInDb.reasonText).toBe( - `No eligible pack specifications found for letter variant id ${letterVariant} and pack specification ids ${packSpecificationIds?.join(", ")}`, - ); - break; - } - case 2: { - const { packSpecificationIds } = supplierAllocatorLog; - expect(packSpecificationIds).toBeTruthy(); - expect(lettersInDb.reasonText).toContain( - `No preferred supplier packs found for pack specification ids ${packSpecificationIds?.join(", ")} and suppliers`, - ); - break; - } - case 3: { - expect(lettersInDb.reasonText).toContain( - `No pack specification found for id`, - ); - await updateLetterVariantPackSpecs(letterVariant, [ - "notify-c5-colour", - ]); // update back to valid config for other tests - break; - } - default: { - throw new Error(`Unexpected test case ${testCase}`); - } - } + test("Verify that the letters are placed on a DLQ when no pack specification found for letter variant", async () => { + const letterVariant = getVariantsForAllocation(7); + const domainId = `NoPackSpecificationFound-${randomUUID()}`; + + await updateLetterVariantPackSpecs(letterVariant, [""]); + + const preparedEvent = createPreparedV1Event({ + domainId, + letterVariantId: letterVariant, + pageCount: 2, }); - } + + const response = await sendSnsEvent(preparedEvent); + expect(response.MessageId).toBeTruthy(); + + await getAllocationLog("No pack specification found for id"); + + await pollQueueForLetterEvent("supplier-allocator-dlq", domainId); + + await updateLetterVariantPackSpecs(letterVariant, ["notify-c5-colour"]); // update back to valid config for other tests + }); for (const { fieldToUpdate, testName, volumeGroupId } of [ { testName: - "Verify that letters are rejected when volumeGroup is not active", + "Verify that letters are placed on a DLQ when volumeGroup is not active", volumeGroupId: "volumeGroup-test2", fieldToUpdate: "startDate", }, { testName: - "Verify that letters are rejected when volumeGroup is no longer active", + "Verify that letters are placed on a DLQ when volumeGroup is no longer active", volumeGroupId: "volumeGroup-test2", fieldToUpdate: "endDate", }, @@ -163,20 +141,7 @@ test.describe("Allocator Rejected Allocation Tests", () => { const response = await sendSnsEvent(preparedEvent); expect(response.MessageId).toBeTruthy(); - const allocationLog = await getAllocationLogForDomainId(domainId); - const lettersInDb = await getLettersFromSupplierTable( - "unknown", - domainId, - "REJECTED", - ); - - expect(lettersInDb.status).toBe("REJECTED"); - expect(lettersInDb.supplierId).toBe( - allocationLog.msg?.allocationDetails?.supplierSpec?.supplierId, - ); - expect(lettersInDb.reasonText).toContain( - `Volume group with id ${volumeGroupId} is not active`, - ); + await pollQueueForLetterEvent("supplier-allocator-dlq", domainId); const resolvedOriginalEndDate = originalEndDate ?? diff --git a/tests/component-tests/allocation-tests/supplier-allocation.spec.ts b/tests/component-tests/allocation-tests/supplier-allocation.spec.ts index fc1f03c27..c71ad873e 100644 --- a/tests/component-tests/allocation-tests/supplier-allocation.spec.ts +++ b/tests/component-tests/allocation-tests/supplier-allocation.spec.ts @@ -13,6 +13,7 @@ import { toZonedTime } from "date-fns-tz"; import { randomUUID } from "node:crypto"; import { createPreparedV1Event } from "tests/helpers/event-fixtures"; import { sendSnsBatchEvent } from "tests/helpers/send-sns-event"; +import { pollQueueForLetterEvent } from "tests/helpers/aws-queue-helper"; test.describe("Supplier Allocation Tests", () => { test("Verify that successful supplier allocation emits a PENDING event for the allocated supplier", async () => { @@ -41,7 +42,7 @@ test.describe("Supplier Allocation Tests", () => { ); }); - test("Verify that supplier allocator emits a rejected request for an unknown letter variant", async () => { + test("Verify that supplier allocator places the event on a DLQ for an unknown letter variant", async () => { test.setTimeout(180_000); // 3 minutes for long running polling const domainId = randomUUID(); logger.info( @@ -55,15 +56,8 @@ test.describe("Supplier Allocation Tests", () => { { id: preparedEvent.id, message: preparedEvent }, ]); expect(response.Successful).toHaveLength(1); - const allocationDetails = - await pollSupplierAllocatorForAllocationDetails(domainId); - - const supplierId = allocationDetails?.supplierSpec?.supplierId; - - const status = allocationDetails?.allocationStatus?.status; - expect(supplierId).toBe("unknown"); - expect(status).toBe("REJECTED"); + await pollQueueForLetterEvent("supplier-allocator-dlq", domainId); }); test("Verify that supplier allocations are correctly updated only once for a volume group for multiple messages", async () => { diff --git a/tests/helpers/aws-queue-helper.ts b/tests/helpers/aws-queue-helper.ts new file mode 100644 index 000000000..6c75347db --- /dev/null +++ b/tests/helpers/aws-queue-helper.ts @@ -0,0 +1,87 @@ +import { + DeleteMessageCommand, + Message, + ReceiveMessageCommand, + SQSClient, +} from "@aws-sdk/client-sqs"; +import { PreparedEventSchema } from "lambdas/supplier-allocator/src/handler/types"; +import { + AWS_ACCOUNT_ID, + AWS_REGION, + envName, +} from "tests/constants/api-constants"; +import { setTimeout } from "node:timers/promises"; + +function messageMatchesDomainId(message: Message, domainId: string): boolean { + if (!message.Body) { + return false; + } + try { + const letterEvent = PreparedEventSchema.parse(JSON.parse(message.Body)); + return letterEvent.data.domainId === domainId; + } catch { + // Allow for (and ignore) malformed messages on DLQ + return false; + } +} + +async function doPoll( + client: SQSClient, + queueUrl: string, + domainId: string, + options: { abortSignal: AbortSignal }, +) { + let matchingMessage; + do { + const response = await client.send( + new ReceiveMessageCommand({ + QueueUrl: queueUrl, + VisibilityTimeout: 0, + WaitTimeSeconds: 5, + }), + options, + ); + + matchingMessage = (response.Messages || []).find((message) => + messageMatchesDomainId(message, domainId), + ); + } while (!matchingMessage && !options.abortSignal.aborted); + + if (matchingMessage) { + await client.send( + new DeleteMessageCommand({ + QueueUrl: queueUrl, + ReceiptHandle: matchingMessage.ReceiptHandle, + }), + options, + ); + } else { + throw new Error("Timed out polling queue"); + } +} + +export async function pollQueueForLetterEvent( + queueName: string, + domainId: string, +) { + const queueUrl = `https://sqs.${AWS_REGION}.amazonaws.com/${AWS_ACCOUNT_ID}/nhs-${envName}-supapi-${queueName}`; + const client = new SQSClient({ region: AWS_REGION }); + + const cancelTimeout = new AbortController(); + const cancelPolling = new AbortController(); + + const timeoutPromise = setTimeout(60_000, undefined, { + signal: cancelTimeout.signal, + }).then(() => { + cancelPolling.abort(); + throw new Error("Timed out polling queue"); + }); + + const pollPromise = doPoll(client, queueUrl, domainId, { + abortSignal: cancelPolling.signal, + }).finally(() => { + cancelTimeout.abort(); + }); + + await Promise.race([timeoutPromise, pollPromise]); +} diff --git a/tests/package.json b/tests/package.json index d2fab44c1..2482f5228 100644 --- a/tests/package.json +++ b/tests/package.json @@ -6,6 +6,7 @@ "@aws-sdk/client-kinesis": "^3.964.0", "@aws-sdk/client-lambda": "^3.986.0", "@aws-sdk/client-sns": "^3.1044.0", + "@aws-sdk/client-sqs": "^3.1044.0", "@aws-sdk/lib-dynamodb": "^3.1044.0", "@nhsdigital/nhs-notify-event-schemas-letter-rendering": "^2.0.2", "@nhsdigital/notify-digital-letters-consumer-contracts": "^1.0.1", @@ -18,6 +19,7 @@ "graphql": "^16.11.0", "graphql-tag": "^2.12.6", "md5": "^2.3.0", + "nhs-notify-supplier-api-allocate-letter": "^0.0.1", "nhs-notify-supplier-api-upsert-letter": "^0.0.1", "openapi-response-validator": "^12.1.3", "pino": "^10.3.0",