Introduce idempotency: replayed update with same ID will be ignored

Author: stevebux

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

@@ -14,10 +14,12 @@ function createLetter(
   supplierId: string,
   letterId: string,
   status: Letter["status"] = "PENDING",
+  eventId?: string,
 ): InsertLetter {
   const now = new Date().toISOString();
   return {
     id: letterId,
+    eventId,
     supplierId,
     specificationId: "specification1",
     groupId: "group1",
@@ -168,6 +170,7 @@ describe("LetterRepository", () => {
 
     const updateLetter: UpdateLetter = {
       id: "letter1",
+      eventId: "event1",
       supplierId: "supplier1",
       status: "REJECTED",
       reasonCode: "R01",
@@ -199,6 +202,7 @@ describe("LetterRepository", () => {
     jest.setSystemTime(new Date(2020, 1, 2));
     const letterDto: UpdateLetter = {
       id: "letter1",
+      eventId: "event1",
       supplierId: "supplier1",
       status: "DELIVERED",
     };
@@ -215,6 +219,7 @@ describe("LetterRepository", () => {
   test("can't update a letter that does not exist", async () => {
     const updateLetter: UpdateLetter = {
       id: "letter1",
+      eventId: "event1",
       supplierId: "supplier1",
       status: "DELIVERED",
     };
@@ -233,6 +238,7 @@ describe("LetterRepository", () => {
 
     const updateLetter: UpdateLetter = {
       id: "letter1",
+      eventId: "event1",
       supplierId: "supplier1",
       status: "DELIVERED",
     };
@@ -241,6 +247,52 @@ describe("LetterRepository", () => {
     ).rejects.toThrow("Cannot do operations on a non-existent table");
   });
 
+  test("does not update a letter if the same eventId is used", async () => {
+    const letter = createLetter("supplier1", "letter1", "DELIVERED", "event1");
+    await letterRepository.putLetter(letter);
+
+    const duplicateUpdate: UpdateLetter = {
+      id: "letter1",
+      eventId: "event1",
+      supplierId: "supplier1",
+      status: "REJECTED",
+      reasonCode: "R01",
+    };
+    const result = await letterRepository.updateLetterStatus(duplicateUpdate);
+
+    expect(result).toBeUndefined();
+    const unchangedLetter = await letterRepository.getLetterById(
+      "supplier1",
+      "letter1",
+    );
+    expect(unchangedLetter.status).toBe("DELIVERED");
+    expect(unchangedLetter.eventId).toBe("event1");
+    expect(unchangedLetter.reasonCode).toBeUndefined();
+  });
+
+  test("updates a letter if a different eventId is used", async () => {
+    const letter = createLetter("supplier1", "letter1", "DELIVERED", "event1");
+    await letterRepository.putLetter(letter);
+
+    const duplicateUpdate: UpdateLetter = {
+      id: "letter1",
+      eventId: "event2",
+      supplierId: "supplier1",
+      status: "REJECTED",
+      reasonCode: "R01",
+    };
+    const result = await letterRepository.updateLetterStatus(duplicateUpdate);
+
+    expect(result).toBeDefined();
+    const changedLetter = await letterRepository.getLetterById(
+      "supplier1",
+      "letter1",
+    );
+    expect(changedLetter.status).toBe("REJECTED");
+    expect(changedLetter.eventId).toBe("event2");
+    expect(changedLetter.reasonCode).toBe("R01");
+  });
+
   test("should return a list of letters matching status", async () => {
     await letterRepository.putLetter(createLetter("supplier1", "letter1"));
     await letterRepository.putLetter(createLetter("supplier1", "letter2"));
@@ -278,6 +330,7 @@ describe("LetterRepository", () => {
 
     const updateLetter: UpdateLetter = {
       id: "letter1",
+      eventId: "event1",
       supplierId: "supplier1",
       status: "DELIVERED",
     };

internal/datastore/src/letter-repository.ts

@@ -7,6 +7,7 @@ import {
   UpdateCommand,
   UpdateCommandOutput,
 } from "@aws-sdk/lib-dynamodb";
+import { ConditionalCheckFailedException } from "@aws-sdk/client-dynamodb";
 import { Logger } from "pino";
 import { z } from "zod";
 import {
@@ -163,32 +164,16 @@ export class LetterRepository {
     };
   }
 
-  async updateLetterStatus(letterToUpdate: UpdateLetter): Promise<Letter> {
+  async updateLetterStatus(
+    letterToUpdate: UpdateLetter,
+  ): Promise<Letter | undefined> {
     this.log.debug(
       `Updating letter ${letterToUpdate.id} to status ${letterToUpdate.status}`,
     );
     let result: UpdateCommandOutput;
     try {
-      let updateExpression =
-        "set #status = :status, updatedAt = :updatedAt, supplierStatus = :supplierStatus, #ttl = :ttl";
-      const expressionAttributeValues: Record<string, any> = {
-        ":status": letterToUpdate.status,
-        ":updatedAt": new Date().toISOString(),
-        ":supplierStatus": `${letterToUpdate.supplierId}#${letterToUpdate.status}`,
-        ":ttl": Math.floor(
-          Date.now() / 1000 + 60 * 60 * this.config.lettersTtlHours,
-        ),
-      };
-
-      if (letterToUpdate.reasonCode) {
-        updateExpression += ", reasonCode = :reasonCode";
-        expressionAttributeValues[":reasonCode"] = letterToUpdate.reasonCode;
-      }
-
-      if (letterToUpdate.reasonText) {
-        updateExpression += ", reasonText = :reasonText";
-        expressionAttributeValues[":reasonText"] = letterToUpdate.reasonText;
-      }
+      const { expressionAttributeValues, updateExpression } =
+        this.buildUpdateExpression(letterToUpdate);
 
       result = await this.ddbClient.send(
         new UpdateCommand({
@@ -198,31 +183,61 @@ export class LetterRepository {
             supplierId: letterToUpdate.supplierId,
           },
           UpdateExpression: updateExpression,
-          ConditionExpression: "attribute_exists(id)", // Ensure letter exists
+          ConditionExpression:
+            "attribute_exists(id) AND (attribute_not_exists(eventId) OR eventId <> :eventId)",
           ExpressionAttributeNames: {
             "#status": "status",
             "#ttl": "ttl",
           },
           ExpressionAttributeValues: expressionAttributeValues,
           ReturnValues: "ALL_NEW",
+          ReturnValuesOnConditionCheckFailure: "ALL_OLD",
         }),
       );
+
+      this.log.debug(
+        `Updated letter ${letterToUpdate.id} to status ${letterToUpdate.status}`,
+      );
+      return LetterSchema.parse(result.Attributes);
     } catch (error) {
-      if (
-        error instanceof Error &&
-        error.name === "ConditionalCheckFailedException"
-      ) {
+      if (error instanceof ConditionalCheckFailedException) {
+        if (error.Item?.eventId.S === letterToUpdate.eventId) {
+          this.log.warn(
+            `Skipping update for letter ${letterToUpdate.id}: eventId ${letterToUpdate.eventId} already processed`,
+          );
+          return undefined;
+        }
         throw new Error(
           `Letter with id ${letterToUpdate.id} not found for supplier ${letterToUpdate.supplierId}`,
         );
       }
       throw error;
     }
+  }
 
-    this.log.debug(
-      `Updated letter ${letterToUpdate.id} to status ${letterToUpdate.status}`,
-    );
-    return LetterSchema.parse(result.Attributes);
+  private buildUpdateExpression(letterToUpdate: UpdateLetter) {
+    let updateExpression =
+      "set #status = :status, updatedAt = :updatedAt, supplierStatus = :supplierStatus, #ttl = :ttl, eventId = :eventId";
+    const expressionAttributeValues: Record<string, any> = {
+      ":status": letterToUpdate.status,
+      ":updatedAt": new Date().toISOString(),
+      ":supplierStatus": `${letterToUpdate.supplierId}#${letterToUpdate.status}`,
+      ":ttl": Math.floor(
+        Date.now() / 1000 + 60 * 60 * this.config.lettersTtlHours,
+      ),
+      ":eventId": letterToUpdate.eventId,
+    };
+
+    if (letterToUpdate.reasonCode) {
+      updateExpression += ", reasonCode = :reasonCode";
+      expressionAttributeValues[":reasonCode"] = letterToUpdate.reasonCode;
+    }
+
+    if (letterToUpdate.reasonText) {
+      updateExpression += ", reasonText = :reasonText";
+      expressionAttributeValues[":reasonText"] = letterToUpdate.reasonText;
+    }
+    return { updateExpression, expressionAttributeValues };
   }
 
   async getLettersBySupplier(

internal/datastore/src/types.ts

@@ -42,6 +42,7 @@ export const LetterSchemaBase = z.object({
 
 export const LetterSchema = LetterSchemaBase.extend({
   supplierId: idRef(SupplierSchema, "id"),
+  eventId: z.string().optional(),
   url: z.url(),
   createdAt: z.string(),
   updatedAt: z.string(),
@@ -67,6 +68,7 @@ export type InsertLetter = Omit<
 >;
 export type UpdateLetter = {
   id: string;
+  eventId: string;
   supplierId: string;
   status: Letter["status"];
   reasonCode?: string;

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

@@ -237,6 +237,7 @@ describe("createUpsertLetterHandler", () => {
     const firstArg = (mockedDeps.letterRepo.putLetter as jest.Mock).mock
       .calls[0][0];
     expect(firstArg.id).toBe("letter1");
+    expect(firstArg.eventId).toBe("7b9a03ca-342a-4150-b56b-989109c45613");
     expect(firstArg.supplierId).toBe("supplier1");
     expect(firstArg.specificationId).toBe("spec1");
     expect(firstArg.url).toBe("s3://letterDataBucket/letter1.pdf");

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

@@ -74,6 +74,7 @@ function mapToInsertLetter(
   const now = new Date().toISOString();
   return {
     id: upsertRequest.data.domainId,
+    eventId: upsertRequest.id,
     supplierId: supplier,
     status: "PENDING",
     specificationId: spec,
@@ -93,6 +94,7 @@ function mapToInsertLetter(
 function mapToUpdateLetter(upsertRequest: LetterEvent): UpdateLetter {
   return {
     id: upsertRequest.data.domainId,
+    eventId: upsertRequest.id,
     supplierId: upsertRequest.data.supplierId,
     status: upsertRequest.data.status,
     reasonCode: upsertRequest.data.reasonCode,