WIP

Author: stevebux

infrastructure/terraform/components/api/iam_role_upsert_idempotency.tf

@@ -16,8 +16,8 @@ resource "aws_iam_role" "upsert_idempotency_role" {
   })
 }
 
-resource "aws_iam_policy" "LambdaDynamoDBPolicy" {
-  name        = "LambdaDynamoDBPolicy"
+resource "aws_iam_policy" "upsert_idempotency_policy" {
+  name        = "${local.csi}-upsert-idempotency-policy"
   description = "IAM policy for Lambda function to access DynamoDB"
   policy = jsonencode({
     Version = "2012-10-17"
@@ -31,13 +31,13 @@ resource "aws_iam_policy" "LambdaDynamoDBPolicy" {
           "dynamodb:UpdateItem",
           "dynamodb:DeleteItem",
         ]
-        Resource = aws_dynamodb_table.IdempotencyTable.arn
+        Resource = aws_dynamodb_table.upsert_idempotency.arn
       },
     ]
   })
 }
 
-resource "aws_iam_role_policy_attachment" "IdempotencyFunctionRoleAttachment" {
-  role       = aws_iam_role.IdempotencyFunctionRole.name
-  policy_arn = aws_iam_policy.LambdaDynamoDBPolicy.arn
+resource "aws_iam_role_policy_attachment" "upsert_idempotency_role_attachment" {
+  role       = aws_iam_role.upsert_idempotency_role.name
+  policy_arn = aws_iam_policy.upsert_idempotency_policy.arn
 }

infrastructure/terraform/components/api/module_lambda_upsert_letter.tf

@@ -35,7 +35,8 @@ module "upsert_letter" {
   log_subscription_role_arn = local.acct.log_subscription_role_arn
 
   lambda_env_vars = merge(local.common_lambda_env_vars, {
-    VARIANT_MAP = jsonencode(var.letter_variant_map)
+    VARIANT_MAP                   = jsonencode(var.letter_variant_map),
+    UPSERT_IDEMPOTENCY_TABLE_NAME = aws_dynamodb_table.upsert_idempotency.name
   })
 }
 

lambdas/upsert-letter/package.json

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

lambdas/upsert-letter/src/config/__tests__/env.test.ts

@@ -16,12 +16,14 @@ describe("lambdaEnv", () => {
 
   it("should load all environment variables successfully", () => {
     process.env.LETTERS_TABLE_NAME = "letters-table";
+    process.env.UPSERT_IDEMPOTENCY_TABLE_NAME = "idempotency-table";
     process.env.LETTER_TTL_HOURS = "12960";
 
     const { envVars } = require("../env");
 
     expect(envVars).toEqual({
       LETTERS_TABLE_NAME: "letters-table",
+      UPSERT_IDEMPOTENCY_TABLE_NAME: "idempotency-table",
       LETTER_TTL_HOURS: 12_960,
     });
   });

lambdas/upsert-letter/src/config/deps.ts

@@ -1,12 +1,14 @@
 import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
 import { DynamoDBDocumentClient } from "@aws-sdk/lib-dynamodb";
+import { DynamoDBPersistenceLayer } from "@aws-lambda-powertools/idempotency/dynamodb";
 import { Logger } from "pino";
 import { LetterRepository } from "@internal/datastore";
 import { createLogger } from "@internal/helpers";
 import { EnvVars, envVars } from "./env";
 
 export type Deps = {
   letterRepo: LetterRepository;
+  idempotencyLayer: DynamoDBPersistenceLayer;
   logger: Logger;
   env: EnvVars;
 };
@@ -16,11 +18,7 @@ function createDocumentClient(): DynamoDBDocumentClient {
   return DynamoDBDocumentClient.from(ddbClient);
 }
 
-function createLetterRepository(
-  log: Logger,
-  // eslint-disable-next-line @typescript-eslint/no-shadow
-  envVars: EnvVars,
-): LetterRepository {
+function createLetterRepository(log: Logger): LetterRepository {
   const config = {
     lettersTableName: envVars.LETTERS_TABLE_NAME,
     lettersTtlHours: envVars.LETTER_TTL_HOURS,
@@ -29,11 +27,18 @@ function createLetterRepository(
   return new LetterRepository(createDocumentClient(), log, config);
 }
 
+function createIdempotencyLayer(): DynamoDBPersistenceLayer {
+  return new DynamoDBPersistenceLayer({
+    tableName: envVars.UPSERT_IDEMPOTENCY_TABLE_NAME,
+  });
+}
+
 export function createDependenciesContainer(): Deps {
   const log = createLogger({ logLevel: envVars.PINO_LOG_LEVEL });
 
   return {
-    letterRepo: createLetterRepository(log, envVars),
+    letterRepo: createLetterRepository(log),
+    idempotencyLayer: createIdempotencyLayer(),
     logger: log,
     env: envVars,
   };

lambdas/upsert-letter/src/config/env.ts

@@ -4,6 +4,7 @@ const EnvVarsSchema = z.object({
   LETTERS_TABLE_NAME: z.string(),
   LETTER_TTL_HOURS: z.coerce.number().int(),
   PINO_LOG_LEVEL: z.coerce.string().optional(),
+  UPSERT_IDEMPOTENCY_TABLE_NAME: z.string(),
 });
 
 export type EnvVars = z.infer<typeof EnvVarsSchema>;

lambdas/upsert-letter/src/handler/__tests__/upsert-handler.test.ts

@@ -6,6 +6,7 @@ import {
 } from "@internal/datastore";
 import { LetterRequestPreparedEventV2 } from "@nhsdigital/nhs-notify-event-schemas-letter-rendering";
 import { LetterRequestPreparedEvent } from "@nhsdigital/nhs-notify-event-schemas-letter-rendering-v1";
+import { makeIdempotent } from "@aws-lambda-powertools/idempotency";
 import {
   $LetterStatusChangeEvent,
   LetterStatusChangeEvent,
@@ -15,6 +16,14 @@ import { Deps } from "../../config/deps";
 import { EnvVars } from "../../config/env";
 import packageJson from "../../../package.json";
 
+jest.mock("@aws-lambda-powertools/idempotency", () => {
+  const original = jest.requireActual("@aws-lambda-powertools/idempotency");
+  return {
+    ...original,
+    makeIdempotent: jest.fn((fn, _) => fn),
+  };
+});
+
 const renderingSchemaVersion: string =
   packageJson.dependencies[
     "@nhsdigital/nhs-notify-event-schemas-letter-rendering"
@@ -313,7 +322,7 @@ describe("createUpsertLetterHandler", () => {
     );
   });
 
-  it("does not treat a replayed insert as a failure", async () => {
+  it("does not treat a second insert for the same letter as a failure", async () => {
     const v1message = {
       letterEvent: createPreparedV1Event(),
       supplierSpec: {
@@ -338,6 +347,26 @@ describe("createUpsertLetterHandler", () => {
     expect(result!.batchItemFailures).toEqual([]);
   });
 
+  it("does not insert a letter if the same message is replayed", async () => {
+    const v1message = {
+      letterEvent: createPreparedV1Event(),
+      supplierSpec: {
+        supplierId: "supplier1",
+        specId: "spec1",
+        priority: 10,
+        billingId: "billing1",
+      },
+    };
+    const evt: SQSEvent = createSQSEvent([
+      createSqsRecord("msg2", JSON.stringify(v1message)),
+    ]);
+    (makeIdempotent as jest.Mock).mockImplementationOnce((_fn) => {});
+
+    await createUpsertLetterHandler(mockedDeps)(evt, {} as any, {} as any);
+
+    expect(mockedDeps.letterRepo.putLetter).not.toHaveBeenCalled();
+  });
+
   test("unknown supplier has metric emitted with 'unknown' supplier dimension", async () => {
     const letterEvent = createSupplierStatusChangeEventWithoutSupplier();
 

lambdas/upsert-letter/src/handler/upsert-handler.ts

@@ -11,6 +11,11 @@ import {
 } from "@nhsdigital/nhs-notify-event-schemas-supplier-api/src/events/letter-events";
 import { $LetterRequestPreparedEventV2 } from "@nhsdigital/nhs-notify-event-schemas-letter-rendering";
 import { MetricsLogger, Unit, metricScope } from "aws-embedded-metrics";
+import {
+  IdempotencyConfig,
+  makeIdempotent,
+} from "@aws-lambda-powertools/idempotency";
+import { AnyFunction } from "@aws-lambda-powertools/idempotency/lib/cjs/types/IdempotencyOptions";
 import { Deps } from "../config/deps";
 import {
   PreparedEvents,
@@ -180,6 +185,7 @@ function parseQueueMessage(queueMessage: string): QueueMessage {
 }
 
 export default function createUpsertLetterHandler(deps: Deps): SQSHandler {
+  const processRecordIdempotently = makeIdempotentOnId(processRecord, deps);
   return metricScope((metrics: MetricsLogger) => {
     return async (event: SQSEvent) => {
       const batchItemFailures: SQSBatchItemFailure[] = [];
@@ -216,28 +222,11 @@ export default function createUpsertLetterHandler(deps: Deps): SQSHandler {
             supplier: supplierSpec,
           });
 
-          supplier =
-            !supplierSpec || !supplierSpec.supplierId
-              ? getSupplierIdFromEvent(letterEvent)
-              : supplierSpec.supplierId;
-
-          const operation = getOperationFromType(letterEvent.type);
-
-          await runUpsert(
-            operation,
-            letterEvent,
-            supplierSpec ?? {
-              supplierId: "unknown",
-              specId: "unknown",
-              priority: 10,
-              billingId: "unknown",
-            },
+          supplier = await processRecordIdempotently(
             deps,
-          );
-
-          perSupplierSuccess.set(
-            supplier,
-            (perSupplierSuccess.get(supplier) || 0) + 1,
+            letterEvent,
+            supplierSpec,
+            perSupplierSuccess,
           );
         } catch (error) {
           deps.logger.error({
@@ -261,3 +250,42 @@ export default function createUpsertLetterHandler(deps: Deps): SQSHandler {
     };
   });
 }
+
+async function processRecord(
+  deps: Deps,
+  letterEvent: LetterStatusChangeEvent | PreparedEvents,
+  supplierSpec: SupplierSpec | undefined,
+  perSupplierSuccess: Map<string, number>,
+) {
+  const supplier =
+    !supplierSpec || !supplierSpec.supplierId
+      ? getSupplierIdFromEvent(letterEvent)
+      : supplierSpec.supplierId;
+
+  const operation = getOperationFromType(letterEvent.type);
+
+  await runUpsert(
+    operation,
+    letterEvent,
+    supplierSpec ?? {
+      supplierId: "unknown",
+      specId: "unknown",
+      priority: 10,
+      billingId: "unknown",
+    },
+    deps,
+  );
+
+  perSupplierSuccess.set(supplier, (perSupplierSuccess.get(supplier) || 0) + 1);
+  return supplier;
+}
+
+function makeIdempotentOnId(fn: AnyFunction, deps: Deps) {
+  return makeIdempotent(fn, {
+    persistenceStore: deps.idempotencyLayer,
+    dataIndexArgument: 1,
+    config: new IdempotencyConfig({
+      eventKeyJmesPath: "id",
+    }),
+  });
+}