CCM-13857 Skip replayed letter event (#495)

* Skip replayed letter event

* Fix tests

* Revert change to PR

* Fix order of arguments

Author: stevebux

internal/datastore/src/__test__/letter-repository.test.ts

@@ -9,6 +9,7 @@ import {
 import { LetterRepository } from "../letter-repository";
 import { InsertLetter, Letter, UpdateLetter } from "../types";
 import { LogStream, createTestLogger } from "./logs";
+import { LetterAlreadyExistsError } from "../letter-already-exists-error";
 
 function createLetter(
   supplierId: string,
@@ -149,9 +150,7 @@ describe("LetterRepository", () => {
     await letterRepository.putLetter(createLetter("supplier1", "letter1"));
     await expect(
       letterRepository.putLetter(createLetter("supplier1", "letter1")),
-    ).rejects.toThrow(
-      "Letter with id letter1 already exists for supplier supplier1",
-    );
+    ).rejects.toThrow(LetterAlreadyExistsError);
   });
 
   test("rethrows errors from DynamoDB when creating a letter", async () => {

internal/datastore/src/letter-repository.ts

@@ -18,6 +18,7 @@ import {
   LetterSchemaBase,
   UpdateLetter,
 } from "./types";
+import { LetterAlreadyExistsError } from "./letter-already-exists-error";
 
 export type PagingOptions = Partial<{
   exclusiveStartKey: Record<string, any>;
@@ -62,9 +63,7 @@ export class LetterRepository {
         error instanceof Error &&
         error.name === "ConditionalCheckFailedException"
       ) {
-        throw new Error(
-          `Letter with id ${letter.id} already exists for supplier ${letter.supplierId}`,
-        );
+        throw new LetterAlreadyExistsError(letter.supplierId, letter.id);
       }
       throw error;
     }

lambdas/upsert-letter/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,

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

@@ -1,6 +1,9 @@
 import { SQSEvent, SQSRecord } from "aws-lambda";
 import pino from "pino";
-import { LetterRepository } from "internal/datastore/src";
+import {
+  LetterAlreadyExistsError,
+  LetterRepository,
+} 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 {
@@ -199,7 +202,11 @@ describe("createUpsertLetterHandler", () => {
       putLetter: jest.fn(),
       updateLetterStatus: jest.fn(),
     } as unknown as LetterRepository,
-    logger: { error: jest.fn(), info: jest.fn() } as unknown as pino.Logger,
+    logger: {
+      error: jest.fn(),
+      warn: jest.fn(),
+      info: jest.fn(),
+    } as unknown as pino.Logger,
     env: {
       LETTERS_TABLE_NAME: "LETTERS_TABLE_NAME",
       LETTER_TTL_HOURS: 12_960,
@@ -301,6 +308,31 @@ describe("createUpsertLetterHandler", () => {
     );
   });
 
+  it("does not treat a replayed insert as a failure", async () => {
+    const v1message = {
+      letterEvent: createPreparedV1Event(),
+      supplierSpec: {
+        supplierId: "supplier1",
+        specId: "spec1",
+        priority: 10,
+        billingId: "billing1",
+      },
+    };
+    const evt: SQSEvent = createSQSEvent([
+      createSqsRecord("msg2", JSON.stringify(v1message)),
+    ]);
+    (mockedDeps.letterRepo.putLetter as jest.Mock).mockRejectedValue(
+      new LetterAlreadyExistsError("supplier1", "letter1"),
+    );
+
+    const result = await createUpsertLetterHandler(mockedDeps)(
+      evt,
+      {} as any,
+      {} as any,
+    );
+    expect(result!.batchItemFailures).toEqual([]);
+  });
+
   test("unknown supplier has metric emitted with 'unknown' supplier dimension", async () => {
     const letterEvent = createSupplierStatusChangeEventWithoutSupplier();
 

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

@@ -1,5 +1,9 @@
 import { SQSBatchItemFailure, SQSEvent, SQSHandler } from "aws-lambda";
-import { InsertLetter, UpdateLetter } from "@internal/datastore";
+import {
+  InsertLetter,
+  LetterAlreadyExistsError,
+  UpdateLetter,
+} from "@internal/datastore";
 import {
   $LetterRequestPreparedEvent,
   LetterRequestPreparedEvent,
@@ -69,14 +73,26 @@ function getOperationFromType(type: string): UpsertOperation {
           supplierSpec.priority,
           supplierSpec.billingId, // use billingId for now
         );
-        await deps.letterRepo.putLetter(letterToInsert);
+        try {
+          await deps.letterRepo.putLetter(letterToInsert);
 
-        deps.logger.info({
-          description: "Inserted letter",
-          eventId: preparedRequest.id,
-          letterId: letterToInsert.id,
-          supplierId: letterToInsert.supplierId,
-        });
+          deps.logger.info({
+            description: "Inserted letter",
+            eventId: preparedRequest.id,
+            letterId: letterToInsert.id,
+            supplierId: letterToInsert.supplierId,
+          });
+        } catch (error) {
+          if (error instanceof LetterAlreadyExistsError) {
+            deps.logger.warn({
+              description: "Letter already exists",
+              supplierId: letterToInsert.supplierId,
+              letterId: letterToInsert.id,
+            });
+          } else {
+            throw error;
+          }
+        }
       },
     };
   }

tests/component-tests/events-tests/event-subscription.spec.ts

@@ -9,6 +9,7 @@ import { SUPPLIER_LETTERS, envName } from "tests/constants/api-constants";
 import {
   pollSupplierAllocatorLogForResolvedSpec,
   pollUpsertLetterLogForError,
+  pollUpsertLetterLogForWarning,
 } from "tests/helpers/aws-cloudwatch-helper";
 import { supplierDataSetup } from "tests/helpers/suppliers-setup-helper";
 import { pollForLetterStatus } from "tests/helpers/poll-for-letters-helper";
@@ -133,8 +134,6 @@ test.describe("Event Subscription SNS Tests", () => {
     expect(duplicateResponse.MessageId).toBeTruthy();
 
     // poll supplier upsert to check if duplicate event was processed
-    await pollUpsertLetterLogForError(
-      `Letter with id ${domainId} already exists for supplier ${supplierId}"`,
-    );
+    await pollUpsertLetterLogForWarning("Letter already exists", domainId);
   });
 });

tests/component-tests/letterQueue-tests/queue-operations.spec.ts

@@ -7,7 +7,7 @@ import {
 import { logger } from "tests/helpers/pino-logger";
 import { sendSnsBatchEvent, sendSnsEvent } from "tests/helpers/send-sns-event";
 import {
-  pollUpsertLetterLogForError,
+  pollUpsertLetterLogForWarning,
   supplierIdFromSupplierAllocatorLog,
 } from "tests/helpers/aws-cloudwatch-helper";
 import getRestApiGatewayBaseUrl from "tests/helpers/aws-gateway-helper";
@@ -102,9 +102,6 @@ test.describe("Letter Queue Tests", () => {
     expect(letterExists).toBe(true);
     expect(itemCount).toBe(1);
 
-    await pollUpsertLetterLogForError(
-      `Letter with id ${letterId} already exists for supplier ${supplierId}"`,
-      letterId,
-    );
+    await pollUpsertLetterLogForWarning("Letter already exists", letterId);
   });
 });

tests/helpers/aws-cloudwatch-helper.ts

@@ -10,15 +10,17 @@ const sleep = (ms: number) =>
     setTimeout(resolve, ms);
   });
 
-export async function pollSupplierAllocatorLogForResolvedSpec(
-  domainId: string,
+async function pollLambdaLog(
+  lambdaName: string,
+  filterPatterns: string[],
+  extraPatterns?: string[],
 ): Promise<string> {
   const intervalMs = 5000;
   const startTimeMs = Date.now() - 5 * 60_000;
   const timeoutMs = 120_000;
 
   const client = new CloudWatchLogsClient({ region: AWS_REGION });
-  const logGroupName = `/aws/lambda/nhs-${envName}-supapi-supplier-allocator`;
+  const logGroupName = `/aws/lambda/nhs-${envName}-supapi-${lambdaName}`;
   const deadline = Date.now() + timeoutMs;
 
   while (Date.now() < deadline) {
@@ -28,18 +30,15 @@ export async function pollSupplierAllocatorLogForResolvedSpec(
         startTime: startTimeMs,
         interleaved: true,
         limit: 100,
-        filterPattern: `"Sending message to upsert letter queue" "${domainId}"`,
+        filterPattern: filterPatterns.join(" "),
       }),
     );
 
     const foundEvent = (response.events ?? []).find((event) => {
       const message = event.message ?? "";
-      return (
-        message.includes(
-          '"description":"Sending message to upsert letter queue"',
-        ) &&
-        (!domainId || message.includes(domainId))
-      );
+      return extraPatterns
+        ? extraPatterns.some((pattern) => message.includes(pattern))
+        : true;
     });
     if (foundEvent?.message) {
       return foundEvent.message;
@@ -48,55 +47,38 @@ export async function pollSupplierAllocatorLogForResolvedSpec(
     await sleep(intervalMs);
   }
 
-  throw new Error(
-    `Timed out waiting for resolved supplier spec log in ${logGroupName}`,
-  );
+  throw new Error(`Timed out waiting for resolved log in ${logGroupName}`);
+}
+
+export async function pollSupplierAllocatorLogForResolvedSpec(
+  domainId: string,
+): Promise<string> {
+  return pollLambdaLog("supplier-allocator", [
+    '"Sending message to upsert letter queue"',
+    `"${domainId}"`,
+  ]);
 }
 
 export async function pollUpsertLetterLogForError(
   msgToCheck: string,
   domainId?: string,
 ): Promise<string> {
-  const intervalMs = 5000;
-  const startTimeMs = Date.now() - 5 * 60_000;
-  const timeoutMs = 120_000;
-
-  const client = new CloudWatchLogsClient({ region: AWS_REGION });
-  const logGroupName = `/aws/lambda/nhs-${envName}-supapi-upsertletter`;
-  const deadline = Date.now() + timeoutMs;
-
-  while (Date.now() < deadline) {
-    const response = await client.send(
-      new FilterLogEventsCommand({
-        logGroupName,
-        startTime: startTimeMs,
-        interleaved: true,
-        limit: 100,
-        filterPattern: domainId
-          ? `"Error processing upsert of record" "${domainId}"`
-          : `"Error processing upsert of record"`,
-      }),
-    );
-
-    const foundEvent = (response.events ?? []).find((event) => {
-      const message = event.message ?? "";
-      return (
-        message.includes('"description":"Error processing upsert of record"') &&
-        (message.includes(`"message":"${msgToCheck}`) ||
-          message.includes(`"message": "${msgToCheck}`))
-      );
-    });
-
-    if (foundEvent?.message) {
-      return foundEvent.message;
-    }
-
-    await sleep(intervalMs);
+  const filterPatterns = ['"Error processing upsert of record"'];
+  if (domainId) {
+    filterPatterns.push(`"${domainId}"`);
   }
+  return pollLambdaLog("upsertletter", filterPatterns, [
+    `"message": "${msgToCheck}`,
+    `"message":"${msgToCheck}`,
+  ]);
+}
 
-  throw new Error(
-    `Timed out waiting for upsert letter error log in ${logGroupName}`,
-  );
+export async function pollUpsertLetterLogForWarning(
+  description: string,
+  domainId: string,
+): Promise<string> {
+  const filterPatterns = ['"WARN"', `"${domainId}"`, `"${description}"`];
+  return pollLambdaLog("upsertletter", filterPatterns);
 }
 
 export async function supplierIdFromSupplierAllocatorLog(