VED-294: Fix potential duplication in the delta during blue / green switchover.

Author: mfjarvis

delta_backend/src/delta.py

@@ -1,16 +1,17 @@
 import decimal
-
-import boto3
 import json
+import logging
 import os
 import time
-from datetime import datetime, timedelta
-import uuid
-import logging
+from datetime import datetime, timedelta, UTC
+
+import boto3
+from boto3.dynamodb.conditions import Attr
 from botocore.exceptions import ClientError
-from log_firehose import FirehoseLogger
-from converter import Converter
+
 from common.mappings import ActionFlag, Operation, EventName
+from converter import Converter
+from log_firehose import FirehoseLogger
 
 failure_queue_url = os.environ["AWS_SQS_QUEUE_URL"]
 delta_table_name = os.environ["DELTA_TABLE_NAME"]
@@ -75,33 +76,36 @@ def send_firehose(log_data):
 
 def process_record(record, log_data):
     ret = True
+    operation_outcome = {}
     try:
         start = time.time()
-        operation_outcome = {}
         error_records = []
         response = str()
         imms_id = str()
         operation = str()
-        approximate_creation_time = datetime.utcfromtimestamp(record["dynamodb"]["ApproximateCreationDateTime"])
+        approximate_creation_time = datetime.fromtimestamp(record["dynamodb"]["ApproximateCreationDateTime"], UTC)
         expiry_time = approximate_creation_time + timedelta(days=30)
         expiry_time_epoch = int(expiry_time.timestamp())
         delta_table = get_delta_table()
 
+        event_id = record["eventID"]
         if record["eventName"] != EventName.DELETE_PHYSICAL:
             new_image = record["dynamodb"]["NewImage"]
             imms_id = new_image["PK"]["S"].split("#")[1]
+            operation_outcome["record"] = imms_id
             vaccine_type = get_vaccine_type(new_image["PatientSK"]["S"])
             supplier_system = new_image["SupplierSystem"]["S"]
             if supplier_system not in ("DPSFULL", "DPSREDUCED"):
                 operation = new_image["Operation"]["S"]
+                operation_outcome["operation_type"] = operation
                 action_flag = ActionFlag.CREATE if operation == Operation.CREATE else operation
                 resource_json = json.loads(new_image["Resource"]["S"], parse_float=decimal.Decimal)
-                FHIRConverter = Converter(resource_json, action_flag=action_flag)
-                flat_json = FHIRConverter.run_conversion()
-                error_records = FHIRConverter.get_error_records()
+                fhir_converter = Converter(resource_json, action_flag=action_flag)
+                flat_json = fhir_converter.run_conversion()
+                error_records = fhir_converter.get_error_records()
                 response = delta_table.put_item(
                     Item={
-                        "PK": str(uuid.uuid4()),
+                        "PK": event_id,
                         "ImmsID": imms_id,
                         "Operation": operation,
                         "VaccineType": vaccine_type,
@@ -110,7 +114,8 @@ def process_record(record, log_data):
                         "Source": delta_source,
                         "Imms": flat_json,
                         "ExpiresAt": expiry_time_epoch,
-                    }
+                    },
+                    ConditionExpression=Attr("PK").not_exists(),
                 )
             else:
                 operation_outcome["statusCode"] = "200"
@@ -120,12 +125,14 @@ def process_record(record, log_data):
                 return True, log_data
         else:
             operation = Operation.DELETE_PHYSICAL
+            operation_outcome["operation_type"] = operation
             new_image = record["dynamodb"]["Keys"]
             logger.info(f"Record to delta:{new_image}")
             imms_id = new_image["PK"]["S"].split("#")[1]
+            operation_outcome["record"] = imms_id
             response = delta_table.put_item(
                 Item={
-                    "PK": str(uuid.uuid4()),
+                    "PK": event_id,
                     "ImmsID": imms_id,
                     "Operation": operation,
                     "VaccineType": "default",
@@ -134,11 +141,10 @@ def process_record(record, log_data):
                     "Source": delta_source,
                     "Imms": "",
                     "ExpiresAt": expiry_time_epoch,
-                }
+                },
+                ConditionExpression=Attr("PK").not_exists(),
             )
-        end = time.time()
-        log_data["time_taken"] = f"{round(end - start, 5)}s"
-        operation_outcome = {"record": imms_id, "operation_type": operation}
+
         if response["ResponseMetadata"]["HTTPStatusCode"] == 200:
             if error_records:
                 log = f"Partial success: successfully synced into delta, but issues found within record {imms_id}"
@@ -154,23 +160,29 @@ def process_record(record, log_data):
         else:
             log = f"Record NOT created for {imms_id}"
             operation_outcome["statusCode"] = "500"
-            operation_outcome["statusDesc"] = "Exception"
+            operation_outcome["statusDesc"] = "Failure response from DynamoDB"
             logger.warning(log)
             ret = False
     except Exception as e:
-        operation_outcome["statusCode"] = "500"
-        operation_outcome["statusDesc"] = "Exception"
-        logger.exception(f"Error processing record: {e}")
-        ret = False
+        if isinstance(e, ClientError) and e.response["Error"]["Code"] == "ConditionalCheckFailedException":
+            operation_outcome["statusCode"] = "200"
+            operation_outcome["statusDesc"] = "Skipped record already present in delta"
+            logger.info(f"Skipped record {event_id} already present in delta for {imms_id}")
+        else:
+            operation_outcome["statusCode"] = "500"
+            operation_outcome["statusDesc"] = "Exception"
+            logger.exception(f"Error processing record: {e}")
+            ret = False
 
+    end = time.time()
+    log_data["time_taken"] = f"{round(end - start, 5)}s"
     log_data["operation_outcome"] = operation_outcome
     return ret, log_data
 
-def handler(event, context):
+def handler(event, _context):
     ret = True
     logger.info("Starting Delta Handler")
     log_data = dict()
-    operation_outcome = dict()
     log_data["function_name"] = "delta_sync"
     try:
         for record in event["Records"]:
@@ -180,7 +192,7 @@ def handler(event, context):
             if not result:
                 ret = False
 
-    except Exception as e:
+    except Exception:
         ret = False
         operation_outcome = {
             "statusCode": "500",

delta_backend/tests/test_delta.py

@@ -16,7 +16,8 @@
 from delta import send_message, handler, process_record  # Import after setting environment variables
 
 SUCCESS_RESPONSE = {"ResponseMetadata": {"HTTPStatusCode": 200}}
-EXCEPTION_RESPONSE = ClientError({"Error": {"Code": "ConditionalCheckFailedException"}}, "PutItem")
+DUPLICATE_RESPONSE = ClientError({"Error": {"Code": "ConditionalCheckFailedException"}}, "PutItem")
+EXCEPTION_RESPONSE = ClientError({"Error": {"Code": "InternalServerError"}}, "PutItem")
 FAIL_RESPONSE = {"ResponseMetadata": {"HTTPStatusCode": 500}}
 
 class DeltaHandlerTestCase(unittest.TestCase):
@@ -374,6 +375,25 @@ def test_single_exception_in_multi(self):
         self.assertEqual(self.mock_delta_table.put_item.call_count, len(records_config))
         self.assertEqual(self.mock_firehose_logger.send_log.call_count, len(records_config))
 
+    def test_single_duplicate_in_multi(self):
+        # Arrange
+        self.mock_delta_table.put_item.side_effect = [SUCCESS_RESPONSE, DUPLICATE_RESPONSE, SUCCESS_RESPONSE]
+
+        records_config = [
+            RecordConfig(EventName.CREATE, Operation.CREATE, "ok-id2.1", ActionFlag.CREATE),
+            RecordConfig(EventName.UPDATE, Operation.UPDATE, "duplicate-id2.2", ActionFlag.UPDATE),
+            RecordConfig(EventName.DELETE_PHYSICAL, Operation.DELETE_PHYSICAL, "ok-id2.3"),
+        ]
+        event = ValuesForTests.get_multi_record_event(records_config)
+
+        # Act
+        result = handler(event, None)
+
+        # Assert
+        self.assertTrue(result)
+        self.assertEqual(self.mock_delta_table.put_item.call_count, len(records_config))
+        self.assertEqual(self.mock_firehose_logger.send_log.call_count, len(records_config))
+
     @patch("delta.process_record")
     @patch("delta.send_firehose")
     def test_handler_calls_process_record_for_each_event(self, mock_send_firehose, mock_process_record):

delta_backend/tests/utils_for_converter_tests.py

@@ -1,3 +1,4 @@
+import uuid
 from decimal import Decimal
 import json
 from common.mappings import EventName, Operation
@@ -171,7 +172,8 @@ def get_multi_record_event(records_config: List[RecordConfig]):
     def get_event_record(imms_id, event_name, operation, supplier="EMIS"):
         pk = f"covid#{imms_id}"
         if operation != Operation.DELETE_PHYSICAL:
-            return{
+            return {
+                "eventID": str(uuid.uuid4()),
                 "eventName": event_name,
                 "dynamodb": {
                     "ApproximateCreationDateTime": 1690896000,
@@ -187,6 +189,7 @@ def get_event_record(imms_id, event_name, operation, supplier="EMIS"):
             }
         else:
             return {
+                "eventID": str(uuid.uuid4()),
                 "eventName": event_name,
                 "dynamodb": {
                     "ApproximateCreationDateTime": 1690896000,