CCM-13519 Add idempotency layer in front of upsert

Author: stevebux

infrastructure/terraform/components/api/README.md

@@ -58,6 +58,7 @@ No requirements.
 | <a name="module_ddb_alarms_letters"></a> [ddb\_alarms\_letters](#module\_ddb\_alarms\_letters) | ../../modules/alarms-ddb | n/a |
 | <a name="module_ddb_alarms_mi"></a> [ddb\_alarms\_mi](#module\_ddb\_alarms\_mi) | ../../modules/alarms-ddb | n/a |
 | <a name="module_ddb_alarms_suppliers"></a> [ddb\_alarms\_suppliers](#module\_ddb\_alarms\_suppliers) | ../../modules/alarms-ddb | n/a |
+| <a name="module_ddb_alarms_upsert_idempotency"></a> [ddb\_alarms\_upsert\_idempotency](#module\_ddb\_alarms\_upsert\_idempotency) | ../../modules/alarms-ddb | n/a |
 | <a name="module_domain_truststore"></a> [domain\_truststore](#module\_domain\_truststore) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/3.0.6/terraform-s3bucket.zip | n/a |
 | <a name="module_eventpub"></a> [eventpub](#module\_eventpub) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/3.0.6/terraform-eventpub.zip | n/a |
 | <a name="module_eventsub"></a> [eventsub](#module\_eventsub) | ../../modules/eventsub | n/a |

infrastructure/terraform/components/api/ddb_table_upsert_idempotency.tf

@@ -0,0 +1,24 @@
+resource "aws_dynamodb_table" "upsert_idempotency" {
+  name         = "${local.csi}-upsert-idempotency"
+  billing_mode = "PAY_PER_REQUEST"
+  hash_key     = "id"
+  attribute {
+    name = "id"
+    type = "S"
+  }
+  ttl {
+    attribute_name = "expiration"
+    enabled        = true
+  }
+
+  point_in_time_recovery {
+    enabled = true
+  }
+
+  tags = merge(
+    local.default_tags,
+    {
+      NHSE-Enable-Dynamo-Backup-Acct = "True"
+    }
+  )
+}

infrastructure/terraform/components/api/module_ddb_alarms_upsert_idempotency.tf

@@ -0,0 +1,7 @@
+module "ddb_alarms_upsert_idempotency" {
+  count        = local.alarms_enabled ? 1 : 0
+  source       = "../../modules/alarms-ddb"
+  alarm_prefix = local.csi
+  table_name   = aws_dynamodb_table.upsert_idempotency.name
+  tags         = local.default_tags
+}

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
   })
 }
 
@@ -67,6 +68,7 @@ data "aws_iam_policy_document" "upsert_letter_lambda" {
 
     resources = [
       aws_dynamodb_table.letters.arn,
+      aws_dynamodb_table.upsert_idempotency.arn,
       "${aws_dynamodb_table.letters.arn}/index/supplierStatus-index"
     ]
   }

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) => "supplier1");
+
+    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

@@ -1,4 +1,4 @@
-import { SQSBatchItemFailure, SQSEvent, SQSHandler } from "aws-lambda";
+import { Context, SQSBatchItemFailure, SQSEvent, SQSHandler } from "aws-lambda";
 import { $LetterRequestPreparedEvent } from "@nhsdigital/nhs-notify-event-schemas-letter-rendering-v1";
 import {
   InsertLetter,
@@ -11,6 +11,10 @@ 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 { Deps } from "../config/deps";
 import {
   PreparedEvents,
@@ -20,6 +24,10 @@ import {
   UpsertOperation,
 } from "./schemas";
 
+const idempotencyConfig = new IdempotencyConfig({
+  eventKeyJmesPath: "id",
+});
+
 function getOperationFromType(type: string): UpsertOperation {
   if (
     type.startsWith("uk.nhs.notify.letter-rendering.letter-request.prepared")
@@ -180,8 +188,13 @@ function parseQueueMessage(queueMessage: string): QueueMessage {
 }
 
 export default function createUpsertLetterHandler(deps: Deps): SQSHandler {
+  const processRecordIdempotently = makeIdempotent(processRecord, {
+    persistenceStore: deps.idempotencyLayer,
+    config: idempotencyConfig,
+  });
+
   return metricScope((metrics: MetricsLogger) => {
-    return async (event: SQSEvent) => {
+    return async (event: SQSEvent, context: Context) => {
       const batchItemFailures: SQSBatchItemFailure[] = [];
       const perSupplierSuccess: Map<string, number> = new Map<string, number>();
       const perSupplierFailure: Map<string, number> = new Map<string, number>();
@@ -216,29 +229,13 @@ 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,
+          idempotencyConfig.registerLambdaContext(context);
+          supplier = await processRecordIdempotently(
             letterEvent,
-            supplierSpec ?? {
-              supplierId: "unknown",
-              specId: "unknown",
-              priority: 10,
-              billingId: "unknown",
-            },
+            supplierSpec,
+            perSupplierSuccess,
             deps,
           );
-
-          perSupplierSuccess.set(
-            supplier,
-            (perSupplierSuccess.get(supplier) || 0) + 1,
-          );
         } catch (error) {
           deps.logger.error({
             description: "Error processing upsert of record",
@@ -261,3 +258,32 @@ export default function createUpsertLetterHandler(deps: Deps): SQSHandler {
     };
   });
 }
+
+async function processRecord(
+  letterEvent: LetterStatusChangeEvent | PreparedEvents,
+  supplierSpec: SupplierSpec | undefined,
+  perSupplierSuccess: Map<string, number>,
+  deps: Deps,
+) {
+  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;
+}