CCM-12392: Allocation idempotency

infrastructure/terraform/components/api/module_lambda_supplier_allocator.tf

@@ -35,7 +35,8 @@ 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
+    UPSERT_LETTERS_QUEUE_URL = module.sqs_letter_updates.sqs_queue_url,
+    IDEMPOTENCY_TABLE_NAME   = aws_dynamodb_table.idempotency.name
   })
 }
 
@@ -110,6 +111,7 @@ data "aws_iam_policy_document" "supplier_allocator_lambda" {
 
     resources = [
       aws_dynamodb_table.supplier-quotas.arn,
+      aws_dynamodb_table.idempotency.arn,
       "${aws_dynamodb_table.supplier-quotas.arn}/index/*"
     ]
   }

lambdas/supplier-allocator/package.json

@@ -1,5 +1,6 @@
 {
   "dependencies": {
+    "@aws-lambda-powertools/idempotency": "^2.33.0",
     "@aws-sdk/client-dynamodb": "^3.858.0",
     "@aws-sdk/client-sqs": "^3.984.0",
     "@aws-sdk/lib-dynamodb": "^3.1008.0",

lambdas/supplier-allocator/src/config/__tests__/deps.test.ts

@@ -4,6 +4,7 @@ describe("createDependenciesContainer", () => {
   const env = {
     SUPPLIER_CONFIG_TABLE_NAME: "SupplierConfigTable",
     SUPPLIER_QUOTAS_TABLE_NAME: "SupplierQuotasTable",
+    IDEMPOTENCY_TABLE_NAME: "IdempotencyTable",
   };
 
   beforeEach(() => {
@@ -27,6 +28,10 @@ describe("createDependenciesContainer", () => {
       SupplierQuotasRepository: jest.fn(),
     }));
 
+    jest.mock("@aws-lambda-powertools/idempotency/dynamodb", () => ({
+      DynamoDBPersistenceLayer: jest.fn(),
+    }));
+
     // Env
     jest.mock("../env", () => ({ envVars: env }));
   });
@@ -40,6 +45,9 @@ describe("createDependenciesContainer", () => {
     const { SupplierQuotasRepository } = jest.requireMock(
       "@internal/datastore",
     );
+    const { DynamoDBPersistenceLayer } = jest.requireMock(
+      "@aws-lambda-powertools/idempotency/dynamodb",
+    );
     // eslint-disable-next-line @typescript-eslint/no-require-imports
     const { createDependenciesContainer } = require("../deps");
     const deps: Deps = createDependenciesContainer();
@@ -54,6 +62,11 @@ describe("createDependenciesContainer", () => {
     expect(supplierQuotasRepoCtorArgs[1]).toEqual({
       supplierQuotasTableName: "SupplierQuotasTable",
     });
+    expect(DynamoDBPersistenceLayer).toHaveBeenCalledTimes(1);
+    const idempotencyLayerCtorArgs = DynamoDBPersistenceLayer.mock.calls[0][0];
+    expect(idempotencyLayerCtorArgs).toEqual({
+      tableName: "IdempotencyTable",
+    });
     expect(deps.env).toEqual(env);
   });
 });

lambdas/supplier-allocator/src/config/__tests__/env.test.ts

@@ -16,12 +16,14 @@ describe("lambdaEnv", () => {
   it("should load all environment variables successfully", () => {
     process.env.SUPPLIER_CONFIG_TABLE_NAME = "SupplierConfigTable";
     process.env.SUPPLIER_QUOTAS_TABLE_NAME = "SupplierQuotasTable";
+    process.env.IDEMPOTENCY_TABLE_NAME = "IdempotencyTable";
 
     const { envVars } = require("../env");
 
     expect(envVars).toEqual({
       SUPPLIER_CONFIG_TABLE_NAME: "SupplierConfigTable",
       SUPPLIER_QUOTAS_TABLE_NAME: "SupplierQuotasTable",
+      IDEMPOTENCY_TABLE_NAME: "IdempotencyTable",
     });
   });
 });

lambdas/supplier-allocator/src/config/deps.ts

@@ -1,5 +1,6 @@
 import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
 import { DynamoDBDocumentClient } from "@aws-sdk/lib-dynamodb";
+import { DynamoDBPersistenceLayer } from "@aws-lambda-powertools/idempotency/dynamodb";
 import { SQSClient } from "@aws-sdk/client-sqs";
 import { Logger } from "pino";
 import { createLogger } from "@internal/helpers";
@@ -12,6 +13,7 @@ import { EnvVars, envVars } from "./env";
 export type Deps = {
   supplierConfigRepo: SupplierConfigRepository;
   supplierQuotasRepo: SupplierQuotasRepository;
+  idempotencyLayer: DynamoDBPersistenceLayer;
   logger: Logger;
   env: EnvVars;
   sqsClient: SQSClient;
@@ -22,36 +24,35 @@ function createDocumentClient(): DynamoDBDocumentClient {
   return DynamoDBDocumentClient.from(ddbClient);
 }
 
-function createSupplierConfigRepository(
-  log: Logger,
-  // eslint-disable-next-line @typescript-eslint/no-shadow
-  envVars: EnvVars,
-): SupplierConfigRepository {
+function createSupplierConfigRepository(): SupplierConfigRepository {
   const config = {
     supplierConfigTableName: envVars.SUPPLIER_CONFIG_TABLE_NAME,
   };
 
   return new SupplierConfigRepository(createDocumentClient(), config);
 }
 
-function createSupplierQuotasRepository(
-  log: Logger,
-  // eslint-disable-next-line @typescript-eslint/no-shadow
-  envVars: EnvVars,
-): SupplierQuotasRepository {
+function createSupplierQuotasRepository(): SupplierQuotasRepository {
   const config = {
     supplierQuotasTableName: envVars.SUPPLIER_QUOTAS_TABLE_NAME,
   };
 
   return new SupplierQuotasRepository(createDocumentClient(), config);
 }
 
+function createIdempotencyLayer(): DynamoDBPersistenceLayer {
+  return new DynamoDBPersistenceLayer({
+    tableName: envVars.IDEMPOTENCY_TABLE_NAME,
+  });
+}
+
 export function createDependenciesContainer(): Deps {
   const log = createLogger({ logLevel: envVars.PINO_LOG_LEVEL });
 
   return {
-    supplierConfigRepo: createSupplierConfigRepository(log, envVars),
-    supplierQuotasRepo: createSupplierQuotasRepository(log, envVars),
+    supplierConfigRepo: createSupplierConfigRepository(),
+    supplierQuotasRepo: createSupplierQuotasRepository(),
+    idempotencyLayer: createIdempotencyLayer(),
     logger: log,
     env: envVars,
     sqsClient: new SQSClient({}),

lambdas/supplier-allocator/src/config/env.ts

@@ -4,6 +4,7 @@ const EnvVarsSchema = z.object({
   SUPPLIER_CONFIG_TABLE_NAME: z.string(),
   SUPPLIER_QUOTAS_TABLE_NAME: z.string(),
   PINO_LOG_LEVEL: z.coerce.string().optional(),
+  IDEMPOTENCY_TABLE_NAME: z.string(),
 });
 
 export type EnvVars = z.infer<typeof EnvVarsSchema>;

lambdas/supplier-allocator/src/handler/__tests__/allocate-handler.test.ts

@@ -7,15 +7,11 @@ import {
   $LetterStatusChangeEvent,
   LetterStatusChangeEvent,
 } from "@nhsdigital/nhs-notify-event-schemas-supplier-api/src/events/letter-events";
-import {
-  SupplierConfigRepository,
-  SupplierQuotasRepository,
-} from "@internal/datastore";
+import { makeIdempotent } from "@aws-lambda-powertools/idempotency";
 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 packageJson from "../../../package.json";
 
@@ -28,6 +24,14 @@ jest.mock("../../services/supplier-config");
 jest.mock("../../services/supplier-quotas");
 jest.mock("../allocation-config");
 
+jest.mock("@aws-lambda-powertools/idempotency", () => {
+  const original = jest.requireActual("@aws-lambda-powertools/idempotency");
+  return {
+    ...original,
+    makeIdempotent: jest.fn((fn, _) => fn),
+  };
+});
+
 function createSQSEvent(records: SQSRecord[]): SQSEvent {
   return {
     Records: records,
@@ -185,16 +189,15 @@ function setupDefaultMocks() {
 }
 
 describe("createSupplierAllocatorHandler", () => {
-  let mockSqsClient: jest.Mocked<SQSClient>;
-  let mockedDeps: jest.Mocked<Deps>;
-  let mockedSupplierConfigRepo: jest.Mocked<SupplierConfigRepository>;
-  let mockedSupplierQuotasRepo: jest.Mocked<SupplierQuotasRepository>;
-  beforeEach(() => {
-    mockSqsClient = {
-      send: jest.fn(),
-    } as unknown as jest.Mocked<SQSClient>;
-
-    mockedSupplierConfigRepo = {
+  const mockedDeps: jest.Mocked<Deps> = {
+    logger: { error: jest.fn(), info: jest.fn() } as unknown as pino.Logger,
+    env: {
+      SUPPLIER_CONFIG_TABLE_NAME: "SupplierConfigTable",
+      SUPPLIER_QUOTAS_TABLE_NAME: "SupplierQuotasTable",
+      IDEMPOTENCY_TABLE_NAME: "IdempotencyTable",
+    },
+    sqsClient: { send: jest.fn() } as unknown as SQSClient,
+    supplierConfigRepo: {
       ddbClient: {} as any,
       config: {} as any,
       getLetterVariant: jest.fn(),
@@ -203,27 +206,18 @@ describe("createSupplierAllocatorHandler", () => {
       getSuppliersDetails: jest.fn(),
       getSupplierPacksForPackSpecification: jest.fn(),
       getPackSpecification: jest.fn(),
-    };
-
-    mockedSupplierQuotasRepo = {
+    },
+    supplierQuotasRepo: {
       ddbClient: {} as any,
       config: {} as any,
       getOverallAllocation: jest.fn(),
       updateOverallAllocation: jest.fn(),
       getDailyAllocation: jest.fn(),
       updateDailyAllocation: jest.fn(),
-    };
+    },
+  } as unknown as Deps;
 
-    mockedDeps = {
-      logger: { error: jest.fn(), info: jest.fn() } as unknown as pino.Logger,
-      env: {
-        SUPPLIER_CONFIG_TABLE_NAME: "SupplierConfigTable",
-        SUPPLIER_QUOTAS_TABLE_NAME: "SupplierQuotasTable",
-      },
-      sqsClient: mockSqsClient,
-      supplierConfigRepo: mockedSupplierConfigRepo,
-      supplierQuotasRepo: mockedSupplierQuotasRepo,
-    };
+  beforeEach(() => {
     jest.clearAllMocks();
   });
 
@@ -244,8 +238,8 @@ describe("createSupplierAllocatorHandler", () => {
 
     expect(result.batchItemFailures).toHaveLength(0);
 
-    expect(mockSqsClient.send).toHaveBeenCalledTimes(1);
-    const sendCall = (mockSqsClient.send as jest.Mock).mock.calls[0][0];
+    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);
@@ -281,8 +275,8 @@ describe("createSupplierAllocatorHandler", () => {
 
     expect(result.batchItemFailures).toHaveLength(0);
 
-    expect(mockSqsClient.send).toHaveBeenCalledTimes(1);
-    const sendCall = (mockSqsClient.send as jest.Mock).mock.calls[0][0];
+    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);
@@ -315,8 +309,8 @@ describe("createSupplierAllocatorHandler", () => {
 
     expect(result.batchItemFailures).toHaveLength(0);
 
-    expect(mockSqsClient.send).toHaveBeenCalledTimes(1);
-    const sendCall = (mockSqsClient.send as jest.Mock).mock.calls[0][0];
+    expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(1);
+    const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock.calls[0][0];
     const messageBody = JSON.parse(sendCall.input.MessageBody);
     expect(messageBody.allocationDetails.supplierSpec).toEqual({
       supplierId: "supplier1",
@@ -361,7 +355,7 @@ describe("createSupplierAllocatorHandler", () => {
     const handler = createSupplierAllocatorHandler(mockedDeps);
     await handler(evt, {} as any, {} as any);
 
-    const sendCall = (mockSqsClient.send as jest.Mock).mock.calls[0][0];
+    const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock.calls[0][0];
     const messageBody = JSON.parse(sendCall.input.MessageBody);
     expect(messageBody.letterEvent.data.domainId).toBe("letter-test");
   });
@@ -410,7 +404,7 @@ describe("createSupplierAllocatorHandler", () => {
     if (!result) throw new Error("expected BatchResponse, got void");
 
     expect(result.batchItemFailures).toHaveLength(0);
-    expect(mockSqsClient.send).toHaveBeenCalledTimes(2);
+    expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(2);
   });
 
   test("returns batch failure for invalid JSON", async () => {
@@ -480,7 +474,7 @@ describe("createSupplierAllocatorHandler", () => {
     process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue";
 
     const sqsError = new Error("SQS send failed");
-    (mockSqsClient.send as jest.Mock).mockRejectedValueOnce(sqsError);
+    (mockedDeps.sqsClient.send as jest.Mock).mockRejectedValueOnce(sqsError);
 
     const handler = createSupplierAllocatorHandler(mockedDeps);
     const result = await handler(evt, {} as any, {} as any);
@@ -513,7 +507,7 @@ describe("createSupplierAllocatorHandler", () => {
     expect(result.batchItemFailures).toHaveLength(1);
     expect(result.batchItemFailures[0].itemIdentifier).toBe("fail-msg");
 
-    expect(mockSqsClient.send).toHaveBeenCalledTimes(2);
+    expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(2);
   });
 
   test("sends correct queue URL in SQS message command", async () => {
@@ -529,7 +523,7 @@ describe("createSupplierAllocatorHandler", () => {
     const handler = createSupplierAllocatorHandler(mockedDeps);
     await handler(evt, {} as any, {} as any);
 
-    const sendCall = (mockSqsClient.send as jest.Mock).mock.calls[0][0];
+    const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock.calls[0][0];
     expect(sendCall.input.QueueUrl).toBe(queueUrl);
   });
 
@@ -557,8 +551,8 @@ describe("createSupplierAllocatorHandler", () => {
         variantId: "lv1",
       }),
     );
-    expect(mockSqsClient.send).toHaveBeenCalledTimes(1);
-    const sendCall = (mockSqsClient.send as jest.Mock).mock.calls[0][0];
+    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);
@@ -667,8 +661,9 @@ describe("createSupplierAllocatorHandler", () => {
           variantId: "lv1",
         }),
       );
-      expect(mockSqsClient.send).toHaveBeenCalledTimes(1);
-      const sendCall = (mockSqsClient.send as jest.Mock).mock.calls[0][0];
+      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);
@@ -711,8 +706,8 @@ describe("createSupplierAllocatorHandler", () => {
         variantId: "lv1",
       }),
     );
-    expect(mockSqsClient.send).toHaveBeenCalledTimes(1);
-    const sendCall = (mockSqsClient.send as jest.Mock).mock.calls[0][0];
+    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);
@@ -771,4 +766,22 @@ describe("createSupplierAllocatorHandler", () => {
     expect(result.batchItemFailures).toHaveLength(0);
     expect(allocationConfig.selectSupplierByFactor).toHaveBeenCalledTimes(2);
   });
+
+  test("does not process a message more than once due to idempotency wrapper", async () => {
+    const preparedEvent = createPreparedV2Event();
+    const evt: SQSEvent = createSQSEvent([
+      createSqsRecord("msg1", JSON.stringify(preparedEvent)),
+    ]);
+
+    process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue";
+
+    setupDefaultMocks();
+    (makeIdempotent as jest.Mock).mockImplementationOnce((_fn) => "supplier1");
+
+    const handler = createSupplierAllocatorHandler(mockedDeps);
+    await handler(evt, {} as any, {} as any);
+
+    expect(makeIdempotent).toHaveBeenCalledTimes(1);
+    expect(mockedDeps.sqsClient.send).not.toHaveBeenCalled();
+  });
 });