[PRMP-895] Add document review processing to virus scan handler (#923)

Author: NogaNHS

lambdas/handlers/document_reference_virus_scan_handler.py

@@ -1,8 +1,10 @@
 from enums.logging_app_interaction import LoggingAppInteraction
+from services.staged_document_review_processing_service import (
+    StagedDocumentReviewProcessingService,
+)
 from services.upload_document_reference_service import UploadDocumentReferenceService
 from utils.audit_logging_setup import LoggingService
 from utils.decorators.ensure_env_var import ensure_environment_variables
-from utils.decorators.handle_lambda_exceptions import handle_lambda_exceptions
 from utils.decorators.set_audit_arg import set_request_context_for_logging
 from utils.lambda_response import ApiGatewayResponse
 from utils.request_context import request_context
@@ -19,19 +21,26 @@
         "LLOYD_GEORGE_BUCKET_NAME",
         "PDM_BUCKET_NAME",
         "VIRUS_SCAN_STUB",
+        "DOCUMENT_REVIEW_DYNAMODB_NAME",
+        "PENDING_REVIEW_BUCKET_NAME"
     ]
 )
-@handle_lambda_exceptions
 def lambda_handler(event, context):
     request_context.app_interaction = LoggingAppInteraction.VIRUS_SCAN.value
     upload_document_reference_service = UploadDocumentReferenceService()
+    review_upload_document_reference_service = StagedDocumentReviewProcessingService()
 
     for s3_object in event.get("Records"):
         object_key = s3_object["s3"]["object"]["key"]
         object_size = s3_object["s3"]["object"]["size"]
-        upload_document_reference_service.handle_upload_document_reference_request(
-            object_key, object_size
-        )
+        if object_key.startswith("review/"):
+            logger.info("Using review document service")
+            service = review_upload_document_reference_service
+        else:
+            logger.info("Using upload document service")
+            service = upload_document_reference_service
+
+        service.handle_upload_document_reference_request(object_key, object_size)
 
     return ApiGatewayResponse(
         200, "Virus Scan was successful", "POST"

lambdas/requirements/layers/requirements_core_lambda_layer.txt

@@ -1,7 +1,7 @@
 PyJWT==2.8.0
 PyYAML==6.0.1
-boto3==1.40.71
-botocore==1.40.71
+boto3==1.42.1
+botocore==1.42.1
 charset-normalizer==3.2.0
 cryptography==44.0.1
 idna==3.7

lambdas/requirements/requirements_test.txt

@@ -1,7 +1,7 @@
 MarkupSafe==2.1.3
 black==24.3.0
 freezegun==1.5.4
-isort==5.13.0
+isort==7.0.0
 pip-audit==2.6.1
 pytest-cov==4.1.0
 pytest-mock==3.11.1

lambdas/services/base/dynamo_service.py

@@ -196,7 +196,7 @@ def update_item(
         table_name: str,
         key_pair: dict[str, str],
         updated_fields: dict,
-        condition_expression: str | None = None,
+        condition_expression: str | ConditionBase| None = None,
         expression_attribute_values: dict | None = None,
     ):
         table = self.get_table(table_name)

lambdas/services/base/s3_service.py

@@ -116,22 +116,25 @@ def copy_across_bucket(
         source_file_key: str,
         dest_bucket: str,
         dest_file_key: str,
-        if_none_match: str | None = None,
+        if_none_match: bool = False,
+        retry_on_conflict: bool = True,
     ):
-        if if_none_match is not None:
-            return self.client.copy_object(
-                Bucket=dest_bucket,
-                Key=dest_file_key,
-                CopySource={"Bucket": source_bucket, "Key": source_file_key},
-                IfNoneMatch=if_none_match,
-                StorageClass="INTELLIGENT_TIERING",
-            )
-        return self.client.copy_object(
-            Bucket=dest_bucket,
-            Key=dest_file_key,
-            CopySource={"Bucket": source_bucket, "Key": source_file_key},
-            StorageClass="INTELLIGENT_TIERING",
-        )
+        copy_source_params = {"Bucket": source_bucket, "Key": source_file_key}
+        copy_object_params = {"Bucket": dest_bucket, "Key": dest_file_key, "CopySource": copy_source_params, "StorageClass": "INTELLIGENT_TIERING"}
+        if if_none_match:
+            copy_object_params["IfNoneMatch"] = '*'
+        try:
+            return self.client.copy_object(**copy_object_params)
+        except ClientError as e:
+            if e.response["ResponseMetadata"]["HTTPStatusCode"] == 409:
+                logger.info(f"Copy failed due to conflict, retrying: {e}")
+                if retry_on_conflict:
+                    return self.copy_across_bucket(source_bucket, source_file_key, dest_bucket, dest_file_key, if_none_match, False)
+                else:
+                    raise e
+            else:
+                logger.error(f"Copy failed: {e}")
+                raise e
 
     def delete_object(
         self, s3_bucket_name: str, file_key: str, version_id: str | None = None

lambdas/services/document_service.py

@@ -142,6 +142,7 @@ def get_item(
             return document
 
         except ValidationError as e:
+            logger.error(f"Validation error on document: {response.get('Item')}")
             logger.error(f"{e}")
             return None
 
@@ -213,11 +214,11 @@ def delete_document_object(self, bucket: str, key: str):
     def update_document(
         self,
         table_name: str | None = None,
-        update_key: dict[str, str] | None = None,
         document: BaseModel = None,
         update_fields_name: set[str] | None = None,
         condition_expression: str | Attr | ConditionBase = None,
         expression_attribute_values: dict = None,
+        key_pair: dict | None = None
     ):
         """Update document in specified or configured table."""
         table_name = table_name or self.table_name
@@ -228,8 +229,8 @@ def update_document(
                 exclude_none=True, by_alias=True, include=update_fields_name
             ),
         }
-        if update_key:
-            update_kwargs["key_pair"] = update_key
+        if key_pair:
+            update_kwargs["key_pair"] = key_pair
         else:
             update_kwargs["key_pair"] = {
                 DocumentReferenceMetadataFields.ID.value: document.id

lambdas/services/document_upload_review_service.py

@@ -10,6 +10,7 @@
 from pydantic import ValidationError
 from services.document_service import DocumentService
 from utils.audit_logging_setup import LoggingService
+from utils.aws_transient_error_check import is_transient_error
 from utils.dynamo_query_filter_builder import DynamoQueryFilterBuilder
 from utils.dynamo_utils import build_transaction_item
 from utils.exceptions import DocumentReviewException
@@ -49,7 +50,7 @@ def query_docs_pending_review_by_custodian_with_limit(
     ) -> tuple[list[DocumentUploadReviewReference], dict | None]:
         logger.info(f"Getting review document references for custodian: {ods_code}")
 
-        filter_expression = self.build_review_query_filter(
+        filter_expression = self.build_review_dynamo_filter(
             nhs_number=nhs_number, uploader=uploader
         )
 
@@ -72,7 +73,7 @@ def query_docs_pending_review_by_custodian_with_limit(
 
         except ClientError as e:
             logger.error(e)
-            raise DocumentReviewException("Failed to query document reviews")
+            raise DocumentReviewException("Error querying document review references")
 
     def _validate_review_references(
         self, items: list[dict]
@@ -85,9 +86,7 @@ def _validate_review_references(
             return review_references
         except ValidationError as e:
             logger.error(e)
-            raise DocumentReviewException(
-                "Failed to validate document review references"
-            )
+            raise DocumentReviewException("Error validating document review references")
 
     def get_document(
         self, document_id: str, version: int | None
@@ -120,10 +119,29 @@ def update_document_review_custodian(
 
                 self.update_document(
                     document=review,
-                    update_key={"ID": review.id, "Version": review.version},
+                    key_pair={"ID": review.id, "Version": review.version},
                     update_fields_name=review_update_field,
                 )
 
+    def update_document_review_status(
+        self,
+        review_document: DocumentUploadReviewReference,
+        condition_expression: str | ConditionBase | None = None,
+    ):
+        review_update_field = {"review_status", "files"}
+        try:
+            self.update_document(
+                document=review_document,
+                key_pair={"ID": review_document.id, "Version": review_document.version},
+                update_fields_name=review_update_field,
+                condition_expression=condition_expression,
+            )
+        except ClientError as e:
+            logger.error(e)
+            if is_transient_error(e):
+                raise e
+            raise DocumentReviewException("Error updating document review status")
+
     def get_document_review_by_id(self, document_id: str, document_version: int):
         return self.get_item(document_id, {"Version": document_version})
 
@@ -169,7 +187,7 @@ def update_document_review_for_patient(
         try:
             return self.update_document(
                 document=review_update,
-                update_key={"ID": review_update.id, "Version": review_update.version},
+                key_pair={"ID": review_update.id, "Version": review_update.version},
                 update_fields_name=field_names,
                 condition_expression=condition_expression,
             )
@@ -273,13 +291,14 @@ def delete_document_review_files(
                 logger.warning(f"Skipping file deletion for {file.file_name}")
                 continue
 
-    def build_review_query_filter(
-        self, nhs_number: str | None = None, uploader: str | None = None
+    def build_review_dynamo_filter(
+        self, nhs_number: str | None = None, uploader: str | None = None, status: DocumentReviewStatus | None = DocumentReviewStatus.PENDING_REVIEW
     ) -> Attr | ConditionBase:
         filter_builder = DynamoQueryFilterBuilder()
-        filter_builder.add_condition(
-            "ReviewStatus", AttributeOperator.EQUAL, DocumentReviewStatus.PENDING_REVIEW
-        )
+        if status:
+            filter_builder.add_condition(
+                "ReviewStatus", AttributeOperator.EQUAL, status
+            )
 
         if nhs_number:
             filter_builder.add_condition(

lambdas/services/mock_virus_scan_service.py

@@ -19,7 +19,16 @@
 
 
 class MockVirusScanService:
+    _instance = None
+
+    def __new__(cls, *args, **kwargs):
+        if cls._instance is None:
+            cls._instance = super(MockVirusScanService, cls).__new__(cls)
+        return cls._instance
+
     def __init__(self):
+        if hasattr(self, "_initialized") and self._initialized:
+            return
         logger.info("Virus scan service is set to mock virus scan service")
 
         infected_nhs_numbers_str = os.getenv("INFECTED_NHS_NUMBERS")
@@ -30,6 +39,7 @@ def __init__(self):
         )
         if self.infected_nhs_numbers:
             logger.info(f"Infected NHS numbers are set to: {self.infected_nhs_numbers}")
+        self._initialized = True
 
     def scan_file(self, file_ref: str, *args, **kwargs) -> VirusScanResult:
         nhs_number = kwargs.get("nhs_number")