Update: [AEA-4021] - Reject duplicate requests (#287)

## Summary

Dispensing systems will send updatePrescriptionStatus requests to the
PSU API to provide status updates on prescription items which are being
processed by pharmacies using the system. The status updates will
include a last updated time stamp and the NPPT service is only
interested in the tracking status with the most recent time stamp.

Although all status updates will be recorded in the data store,
duplicate updates must be rejected.

https://nhsd-jira.digital.nhs.uk/browse/AEA-4021

- ✨ New Feature

packages/updatePrescriptionStatus/src/updatePrescriptionStatus.ts

@@ -14,10 +14,12 @@ import {
   accepted,
   badRequest,
   bundleWrap,
+  conflictDuplicate,
   createSuccessResponseEntries,
   serverError,
   timeoutResponse
 } from "./utils/responses"
+import {TransactionCanceledException} from "@aws-sdk/client-dynamodb"
 
 const LAMBDA_TIMEOUT_MS = 9500
 const logger = new Logger({serviceName: "updatePrescriptionStatus"})
@@ -73,22 +75,31 @@ const lambdaHandler = async (event: APIGatewayProxyEvent): Promise<APIGatewayPro
   }
 
   const dataItems = buildDataItems(requestEntries, xRequestID, applicationName)
-  const persistSuccess = persistDataItems(dataItems)
-  const persistResponse = await jobWithTimeout(LAMBDA_TIMEOUT_MS, persistSuccess)
 
-  if (hasTimedOut(persistResponse)) {
-    responseEntries = [timeoutResponse()]
-    logger.info("DynamoDB operation timed out.")
-    return response(504, responseEntries)
-  }
+  try {
+    const persistSuccess = persistDataItems(dataItems)
+    const persistResponse = await jobWithTimeout(LAMBDA_TIMEOUT_MS, persistSuccess)
 
-  if (!persistResponse) {
-    responseEntries = [serverError()]
-    return response(500, responseEntries)
-  }
+    if (hasTimedOut(persistResponse)) {
+      responseEntries = [timeoutResponse()]
+      logger.info("DynamoDB operation timed out.")
+      return response(504, responseEntries)
+    }
+
+    if (!persistResponse) {
+      responseEntries = [serverError()]
+      return response(500, responseEntries)
+    }
+
+    responseEntries = createSuccessResponseEntries(requestEntries)
+    logger.info("Event processed successfully.")
+  } catch (e) {
+    if (e instanceof TransactionCanceledException) {
+      handleTransactionCancelledException(e, responseEntries)
 
-  responseEntries = createSuccessResponseEntries(requestEntries)
-  logger.info("Event processed successfully.")
+      return response(409, responseEntries)
+    }
+  }
   return response(201, responseEntries)
 }
 
@@ -140,6 +151,18 @@ export function validateEntries(requestEntries: Array<BundleEntry>, responseEntr
   return valid
 }
 
+export function handleTransactionCancelledException(
+  e: TransactionCanceledException,
+  responseEntries: Array<BundleEntry>
+): void {
+  e.CancellationReasons?.forEach((reason) => {
+    if (reason.Item?.TaskID?.S) {
+      const taskId = reason.Item.TaskID.S
+      responseEntries.push(conflictDuplicate(taskId))
+    }
+  })
+}
+
 export function buildDataItems(
   requestEntries: Array<BundleEntry>,
   xRequestID: string,

packages/updatePrescriptionStatus/src/utils/databaseClient.ts

@@ -1,11 +1,16 @@
 import {Logger} from "@aws-lambda-powertools/logger"
-import {DynamoDBClient, TransactWriteItem, TransactWriteItemsCommand} from "@aws-sdk/client-dynamodb"
+import {
+  DynamoDBClient,
+  TransactWriteItem,
+  TransactWriteItemsCommand,
+  TransactionCanceledException
+} from "@aws-sdk/client-dynamodb"
 import {marshall} from "@aws-sdk/util-dynamodb"
 
 import {DataItem} from "../updatePrescriptionStatus"
 import {Timeout} from "./timeoutUtils"
 
-const logger = new Logger({serviceName: "databaseClient"})
+export const logger = new Logger({serviceName: "databaseClient"})
 const client = new DynamoDBClient()
 const tableName = process.env.TABLE_NAME ?? "PrescriptionStatusUpdates"
 
@@ -15,7 +20,9 @@ function createTransactionCommand(dataItems: Array<DataItem>): TransactWriteItem
     return {
       Put: {
         TableName: tableName,
-        Item: marshall(d)
+        Item: marshall(d),
+        ConditionExpression: "attribute_not_exists(TaskID) AND attribute_not_exists(PrescriptionID)",
+        ReturnValuesOnConditionCheckFailure: "ALL_OLD"
       }
     }
   })
@@ -30,6 +37,10 @@ export async function persistDataItems(dataItems: Array<DataItem>): Promise<bool
     logger.info("TransactWriteItemsCommand sent to DynamoDB successfully.", {command: transactionCommand})
     return true
   } catch (e) {
+    if (e instanceof TransactionCanceledException) {
+      logger.error("DynamoDB transaction cancelled due to conditional check failure.", {reasons: e.CancellationReasons})
+      throw e
+    }
     logger.error("Error sending TransactWriteItemsCommand to DynamoDB.", {error: e})
     return false
   }

packages/updatePrescriptionStatus/src/utils/responses.ts

@@ -119,6 +119,39 @@ export function serverError(): BundleEntry {
   }
 }
 
+export function conflictDuplicate(taskId: string): BundleEntry {
+  return {
+    response: {
+      status: "409 Conflict",
+      location: `Task/${taskId}`,
+      lastModified: new Date().toISOString(),
+      outcome: {
+        resourceType: "OperationOutcome",
+        meta: {
+          lastUpdated: new Date().toISOString()
+        },
+        issue: [
+          {
+            code: "duplicate",
+            severity: "error",
+            details: {
+              coding: [
+                {
+                  system: "https://fhir.nhs.uk/CodeSystem/http-error-codes",
+                  code: "REC_CONFLICT",
+                  display: "409: The Receiver identified a conflict."
+                }
+              ]
+            },
+            diagnostics:
+              "Request contains a task id and prescription id identical to a record already in the data store."
+          }
+        ]
+      }
+    }
+  }
+}
+
 export function createSuccessResponseEntries(entries: Array<BundleEntry>) {
   return entries.map((e) => created(e.fullUrl!))
 }

packages/updatePrescriptionStatus/tests/testDatabaseClient.test.ts

@@ -0,0 +1,68 @@
+import {
+  expect,
+  describe,
+  it,
+  jest
+} from "@jest/globals"
+
+import {TransactionCanceledException} from "@aws-sdk/client-dynamodb"
+import {mockDynamoDBClient} from "./utils/testUtils"
+
+const {mockSend} = mockDynamoDBClient()
+const {persistDataItems, logger} = await import("../src/utils/databaseClient")
+
+describe("Unit test persistDataItems", () => {
+  beforeEach(() => {
+    jest.resetModules()
+    jest.clearAllMocks()
+  })
+  it("when the conditional check fails, an error is thrown", async () => {
+    const dataItems = [
+      {
+        LastModified: "2023-01-01T00:00:00Z",
+        LineItemID: "LineItemID_1",
+        PatientNHSNumber: "PatientNHSNumber_1",
+        PharmacyODSCode: "PharmacyODSCode_1",
+        PrescriptionID: "PrescriptionID_1",
+        RequestID: "RequestID_1",
+        Status: "Status_1",
+        TaskID: "TaskID_1",
+        TerminalStatus: "TerminalStatus_1",
+        ApplicationName: "name"
+      },
+      {
+        LastModified: "2023-01-02T00:00:00Z",
+        LineItemID: "LineItemID_2",
+        PatientNHSNumber: "PatientNHSNumber_2",
+        PharmacyODSCode: "PharmacyODSCode_2",
+        PrescriptionID: "PrescriptionID_1",
+        RequestID: "RequestID_2",
+        Status: "Status_2",
+        TaskID: "TaskID_1",
+        TerminalStatus: "TerminalStatus_2",
+        ApplicationName: "name"
+      }
+    ]
+
+    mockSend.mockRejectedValue(
+      new TransactionCanceledException({
+        $metadata: {},
+        message: "DynamoDB transaction cancelled due to conditional check failure.",
+        CancellationReasons: [{Code: "ConditionalCheckFailedException"}]
+      }) as never
+    )
+    const loggerSpy = jest.spyOn(logger, "error")
+
+    await expect(persistDataItems(dataItems)).rejects.toThrow(
+      new TransactionCanceledException({
+        $metadata: {},
+        message: "DynamoDB transaction cancelled due to conditional check failure.",
+        CancellationReasons: [{Code: "ConditionalCheckFailedException"}]
+      }) as never
+    )
+
+    expect(loggerSpy).toHaveBeenCalledWith("DynamoDB transaction cancelled due to conditional check failure.", {
+      reasons: [{Code: "ConditionalCheckFailedException"}]
+    })
+  })
+})

packages/updatePrescriptionStatus/tests/testHandler.test.ts

@@ -30,6 +30,7 @@ import {
   serverError,
   timeoutResponse
 } from "../src/utils/responses"
+import {TransactionCanceledException} from "@aws-sdk/client-dynamodb"
 
 const {mockSend, mockTransact} = mockDynamoDBClient()
 const {handler} = await import("../src/updatePrescriptionStatus")
@@ -185,7 +186,6 @@ describe("Integration tests for updatePrescriptionStatus handler", () => {
   })
 
   it("when x-request-id header is mixed case, expect it to work", async () => {
-
     const body = generateBody()
     const event: APIGatewayProxyEvent = generateMockEvent(body)
     delete event.headers["x-request-id"]
@@ -197,6 +197,22 @@ describe("Integration tests for updatePrescriptionStatus handler", () => {
     const response: APIGatewayProxyResult = await handler(event, {})
 
     expect(response.statusCode).toEqual(201)
+  })
+
+  it("when duplicates are introduced, expect 409 status code and message indicating duplicate fields", async () => {
+    const body = generateBody()
+    const mockEvent: APIGatewayProxyEvent = generateMockEvent(body)
+
+    mockSend.mockRejectedValue(
+      new TransactionCanceledException({
+        $metadata: {},
+        message: "DynamoDB transaction cancelled due to conditional check failure.",
+        CancellationReasons: [{Code: "ConditionalCheckFailedException"}]
+      }) as never
+    )
+
+    const result: APIGatewayProxyResult = await handler(mockEvent, {})
 
+    expect(result.statusCode).toBe(409)
   })
 })

packages/updatePrescriptionStatus/tests/testUpdatePrescriptionStatus.test.ts

@@ -8,13 +8,16 @@ import {
 
 import {BundleEntry} from "fhir/r4"
 
-import {badRequest} from "../src/utils/responses"
+import {badRequest, conflictDuplicate} from "../src/utils/responses"
 import {DEFAULT_DATE, X_REQUEST_ID, mockInternalDependency} from "./utils/testUtils"
 import {APIGatewayProxyEvent} from "aws-lambda"
 
 import * as content from "../src/validation/content"
+import {TransactionCanceledException} from "@aws-sdk/client-dynamodb"
 const mockValidateEntry = mockInternalDependency("../src/validation/content", content, "validateEntry")
-const {castEventBody, getXRequestID, validateEntries} = await import("../src/updatePrescriptionStatus")
+const {castEventBody, getXRequestID, validateEntries, handleTransactionCancelledException} = await import(
+  "../src/updatePrescriptionStatus"
+)
 
 describe("Unit test getXRequestID", () => {
   beforeAll(() => {
@@ -64,7 +67,9 @@ describe("Unit test castEventBody", () => {
 
     expect(result).toEqual(undefined)
     expect(responseEntries.length).toEqual(1)
-    expect(responseEntries[0]).toEqual(badRequest("Request body does not have resourceType of 'Bundle' and type of 'transaction'."))
+    expect(responseEntries[0]).toEqual(
+      badRequest("Request body does not have resourceType of 'Bundle' and type of 'transaction'.")
+    )
   })
 
   it("when body has correct resourceType and type, return bundle and no response entries", async () => {
@@ -123,3 +128,31 @@ describe("Unit test validateEntries", () => {
     expect(inValidResponseEntry.response?.status).toEqual("400 Bad Request")
   })
 })
+
+describe("handleTransactionCancelledException", () => {
+  it("should add conflictDuplicate entries to responseEntries", () => {
+    const responseEntries: Array<any> = []
+    const mockException: TransactionCanceledException = {
+      name: "TransactionCanceledException",
+      message: "DynamoDB transaction cancelled due to conditional check failure.",
+      $fault: "client",
+      $metadata: {},
+      CancellationReasons: [
+        {
+          Code: "ConditionalCheckFailed",
+          Item: {
+            TaskID: {S: "d70678c-81e4-6665-8c67-17596fd0aa46"}
+          },
+          Message: "The conditional request failed"
+        }
+      ]
+    }
+
+    handleTransactionCancelledException(mockException, responseEntries)
+    const validResponseEntry = responseEntries[0]
+
+    expect(responseEntries).toHaveLength(1)
+    expect(validResponseEntry).toEqual(conflictDuplicate("d70678c-81e4-6665-8c67-17596fd0aa46"))
+    expect(validResponseEntry.response?.status).toEqual("409 Conflict")
+  })
+})

packages/updatePrescriptionStatus/tests/utils/testUtils.ts

@@ -2,6 +2,7 @@
 
 import {APIGatewayProxyEvent} from "aws-lambda"
 import {jest} from "@jest/globals"
+import * as dynamo from "@aws-sdk/client-dynamodb"
 
 import {
   LINE_ITEM_ID_CODESYSTEM,
@@ -146,6 +147,7 @@ export function mockDynamoDBClient() {
   const mockTransact = jest.fn()
   jest.unstable_mockModule("@aws-sdk/client-dynamodb", () => {
     return {
+      ...dynamo,
       DynamoDBClient: jest.fn().mockImplementation(() => ({
         send: mockSend
       })),