[PRMP-585] Create ReviewProcessor lambda logic

.github/workflows/base-lambdas-reusable-deploy-all.yml

@@ -696,7 +696,7 @@ jobs:
       lambda_layer_names: "core_lambda_layer"
     secrets:
       AWS_ASSUME_ROLE: ${{ secrets.AWS_ASSUME_ROLE }}
-    
+
   deploy_get_document_reference_by_id_lambda:
     name: Deploy get_document_reference_lambda
     uses: ./.github/workflows/base-lambdas-reusable-deploy.yml
@@ -710,3 +710,17 @@ jobs:
       lambda_layer_names: "core_lambda_layer"
     secrets:
       AWS_ASSUME_ROLE: ${{ secrets.AWS_ASSUME_ROLE }}
+
+  deploy_review_processor_lambda:
+    name: Deploy Review Processor Lambda
+    uses: ./.github/workflows/base-lambdas-reusable-deploy.yml
+    with:
+      environment: ${{ inputs.environment }}
+      python_version: ${{ inputs.python_version }}
+      build_branch: ${{ inputs.build_branch }}
+      sandbox: ${{ inputs.sandbox }}
+      lambda_handler_name: document_review_processor_handler
+      lambda_aws_name: DocumentReviewProcessor
+      lambda_layer_names: "core_lambda_layer"
+    secrets:
+      AWS_ASSUME_ROLE: ${{ secrets.AWS_ASSUME_ROLE }}

lambdas/handlers/document_review_processor_handler.py

@@ -0,0 +1,59 @@
+from models.sqs.review_message_body import ReviewMessageBody
+from pydantic import ValidationError
+from services.document_review_processor_service import ReviewProcessorService
+from utils.audit_logging_setup import LoggingService
+from utils.decorators.ensure_env_var import ensure_environment_variables
+from utils.decorators.override_error_check import override_error_check
+from utils.decorators.set_audit_arg import set_request_context_for_logging
+from utils.request_context import request_context
+
+logger = LoggingService(__name__)
+
+
+@set_request_context_for_logging
+@ensure_environment_variables(
+    names=[
+        "DOCUMENT_REVIEW_DYNAMODB_NAME",
+        "STAGING_STORE_BUCKET_NAME",
+        "PENDING_REVIEW_BUCKET_NAME",
+    ]
+)
+@override_error_check
+def lambda_handler(event, context):
+    """
+    This handler consumes SQS messages from the document review queue, creates DynamoDB
+    records in the DocumentReview table, and moves files from the staging bucket
+    to the pending review bucket.
+
+    Args:
+        event: Lambda event containing SQS Event
+        context: Lambda context
+
+    Returns:
+        None
+    """
+    logger.info("Starting review processor Lambda")
+
+    sqs_messages = event.get("Records", [])
+    review_service = ReviewProcessorService()
+
+    for sqs_message in sqs_messages:
+        try:
+            message = ReviewMessageBody.model_validate_json(sqs_message["body"])
+
+            review_service.process_review_message(message)
+
+        except ValidationError as error:
+            logger.error("Malformed review message")
+            logger.error(error)
+            raise error
+
+        except Exception as error:
+            logger.error(
+                f"Failed to process review message: {str(error)}",
+                {"Result": "Review processing failed"},
+            )
+            raise error
+
+        request_context.patient_nhs_no = ""
+        logger.info("Continuing to next message.")

lambdas/models/document_review.py

@@ -1,5 +1,4 @@
 import uuid
-from typing import Optional
 
 from enums.document_review_status import DocumentReviewStatus
 from enums.metadata_field_names import DocumentReferenceMetadataFields
@@ -45,5 +44,5 @@ class DocumentUploadReviewReference(BaseModel):
     ttl: int | None = Field(
         alias=str(DocumentReferenceMetadataFields.TTL.value), default=None
     )
-    document_reference_id: str = Field(default=None)
+    document_reference_id: str | None = Field(default=None)
     document_snomed_code_type: str = Field(default=SnomedCodes.LLOYD_GEORGE.value.code)

lambdas/models/sqs/review_message_body.py

@@ -0,0 +1,21 @@
+from pydantic import BaseModel, Field
+
+
+class ReviewMessageFile(BaseModel):
+    """Model for individual file in SQS message body from the document review queue."""
+
+    file_name: str
+    file_path: str = Field(description="Location in the staging bucket")
+    """Location in the staging bucket"""
+
+
+class ReviewMessageBody(BaseModel):
+    """Model for SQS message body from the document review queue."""
+
+    upload_id: str
+    files: list[ReviewMessageFile]
+    nhs_number: str
+    failure_reason: str
+    upload_date: str
+    uploader_ods: str
+    current_gp: str

lambdas/requirements/layers/requirements_core_lambda_layer.txt

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

lambdas/services/base/dynamo_service.py

@@ -141,11 +141,27 @@ def query_table(
 
         return items
 
-    def create_item(self, table_name, item):
+    def create_item(self, table_name, item, key_name: str | None = None):
+        """
+        Put an item into the specified DynamoDB table with a condition on the existence of the key.
+        Args:
+            table_name: Name of the DynamoDB table
+            item: The item to be inserted (as a dictionary)
+            key_name: The name of the key field to check existance for conditional put
+        Returns:
+            Response from the DynamoDB put_item operation
+        Raises:
+            ClientError: For AWS service errors (DynamoDB)
+        """
         try:
             table = self.get_table(table_name)
             logger.info(f"Writing item to table: {table_name}")
-            table.put_item(Item=item)
+            if key_name:
+                return table.put_item(
+                    Item=item, ConditionExpression=f"attribute_not_exists({key_name})"
+                )
+            else:
+                return table.put_item(Item=item)
         except ClientError as e:
             logger.error(
                 str(e), {"Result": f"Unable to write item to table: {table_name}"}

lambdas/services/base/s3_service.py

@@ -116,19 +116,32 @@ def copy_across_bucket(
         source_file_key: str,
         dest_bucket: str,
         dest_file_key: str,
+        if_none_match: str | None = None,
     ):
+        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",
         )
 
-    def delete_object(self, s3_bucket_name: str, file_key: str, version_id: str | None = None):
+    def delete_object(
+        self, s3_bucket_name: str, file_key: str, version_id: str | None = None
+    ):
         if version_id is None:
             return self.client.delete_object(Bucket=s3_bucket_name, Key=file_key)
-
-        return self.client.delete_object(Bucket=s3_bucket_name, Key=file_key, VersionId=version_id)
+
+        return self.client.delete_object(
+            Bucket=s3_bucket_name, Key=file_key, VersionId=version_id
+        )
 
     def create_object_tag(
         self, s3_bucket_name: str, file_key: str, tag_key: str, tag_value: str

lambdas/services/document_review_processor_service.py

@@ -0,0 +1,150 @@
+import os
+from datetime import datetime, timezone
+
+from botocore.exceptions import ClientError
+from enums.document_review_status import DocumentReviewStatus
+from models.document_reference import DocumentReferenceMetadataFields
+from models.document_review import (
+    DocumentReviewFileDetails,
+    DocumentUploadReviewReference,
+)
+from models.sqs.review_message_body import ReviewMessageBody
+from services.base.dynamo_service import DynamoDBService
+from services.base.s3_service import S3Service
+from utils.audit_logging_setup import LoggingService
+from utils.request_context import request_context
+
+logger = LoggingService(__name__)
+
+
+class ReviewProcessorService:
+    """
+    Service for processing single SQS messages from the document review queue.
+    """
+
+    def __init__(self):
+        """Initialize the review processor service with required AWS services."""
+        self.dynamo_service = DynamoDBService()
+        self.s3_service = S3Service()
+
+        self.review_table_name = os.environ["DOCUMENT_REVIEW_DYNAMODB_NAME"]
+        self.staging_bucket_name = os.environ["STAGING_STORE_BUCKET_NAME"]
+        self.review_bucket_name = os.environ["PENDING_REVIEW_BUCKET_NAME"]
+
+    def process_review_message(self, review_message: ReviewMessageBody) -> None:
+        """
+        Process a single SQS message from the review queue.
+
+        Args:
+            sqs_message: SQS message record containing file and failure information
+
+        Raises:
+            InvalidMessageException: If message format is invalid or required fields missing
+            S3FileNotFoundException: If file doesn't exist in staging bucket
+            ClientError: For AWS service errors (DynamoDB, S3)
+        """
+
+        logger.info("Processing review queue message")
+
+        request_context.patient_nhs_no = review_message.nhs_number
+
+        review_id = review_message.upload_id
+        review_files = self._move_files_to_review_bucket(review_message, review_id)
+        document_upload_review = self._build_review_record(
+            review_message, review_id, review_files
+        )
+        try:
+            self.dynamo_service.create_item(
+                table_name=self.review_table_name,
+                item=document_upload_review.model_dump(
+                    by_alias=True, exclude_none=True
+                ),
+                key_name=DocumentReferenceMetadataFields.ID.value,
+            )
+
+            logger.info(f"Created review record {document_upload_review.id}")
+        except ClientError as e:
+            if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
+                logger.info("Entry already exists on Document Review table")
+            else:
+                raise e
+
+        self._delete_files_from_staging(review_message)
+
+    def _build_review_record(
+        self,
+        message_data: ReviewMessageBody,
+        review_id: str,
+        review_files: list[DocumentReviewFileDetails],
+    ) -> DocumentUploadReviewReference:
+        return DocumentUploadReviewReference(
+            id=review_id,
+            nhs_number=message_data.nhs_number,
+            review_status=DocumentReviewStatus.PENDING_REVIEW,
+            review_reason=message_data.failure_reason,
+            author=message_data.uploader_ods,
+            custodian=message_data.current_gp,
+            files=review_files,
+            upload_date=int(datetime.now(tz=timezone.utc).timestamp()),
+        )
+
+    def _move_files_to_review_bucket(
+        self, message_data: ReviewMessageBody, review_record_id: str
+    ) -> list[DocumentReviewFileDetails]:
+        """
+        Move file from staging to review bucket.
+
+        Args:
+            message_data: Review queue message data
+            review_record_id: ID of the review record being created
+
+        Returns:
+            List of DocumentReviewFileDetails with new file locations in review bucket
+        """
+        new_file_keys: list[DocumentReviewFileDetails] = []
+
+        for file in message_data.files:
+            new_file_key = (
+                f"{message_data.nhs_number}/{review_record_id}/{file.file_name}"
+            )
+
+            logger.info(
+                f"Copying file from ({file.file_path}) in staging to review bucket: {new_file_key}"
+            )
+            try:
+
+                self.s3_service.copy_across_bucket(
+                    source_bucket=self.staging_bucket_name,
+                    source_file_key=file.file_path,
+                    dest_bucket=self.review_bucket_name,
+                    dest_file_key=new_file_key,
+                    if_none_match="*",
+                )
+                logger.info("File successfully copied to review bucket")
+                logger.info(f"Successfully moved file to: {new_file_key}")
+
+            except ClientError as e:
+                if e.response["Error"]["Code"] == "PreconditionFailed":
+                    logger.info("File already exists in the Review Bucket")
+                else:
+                    raise e
+
+            new_file_keys.append(
+                DocumentReviewFileDetails(
+                    file_name=file.file_name,
+                    file_location=new_file_key,
+                )
+            )
+
+        return new_file_keys
+
+    def _delete_files_from_staging(self, message_data: ReviewMessageBody) -> None:
+        for file in message_data.files:
+            try:
+                logger.info(f"Deleting file from staging bucket: {file.file_path}")
+                self.s3_service.delete_object(
+                    s3_bucket_name=self.staging_bucket_name, file_key=file.file_path
+                )
+            except Exception as e:
+                logger.error(f"Error deleting files from staging: {str(e)}")
+                # Continue processing as files

lambdas/services/document_upload_review_service.py

@@ -17,11 +17,13 @@
 
 class DocumentUploadReviewService(DocumentService):
     """Service for handling DocumentUploadReviewReference operations."""
+
     DEFAULT_QUERY_LIMIT = 50
+
     def __init__(self):
         super().__init__()
         self._table_name = os.environ.get("DOCUMENT_REVIEW_DYNAMODB_NAME")
-        self._s3_bucket = os.environ.get("DOCUMENT_REVIEW_S3_BUCKET_NAME")
+        self._s3_bucket = os.environ.get("PENDING_REVIEW_BUCKET_NAME")
 
     @property
     def table_name(self) -> str:

lambdas/tests/unit/conftest.py

@@ -138,10 +138,9 @@
 MOCK_ALERTING_SLACK_CHANNEL_ID = "slack_channel_id"
 MOCK_DOCUMENT_REVIEW_TABLE = "test_document_review"
 MOCK_DOCUMENT_REVIEW_BUCKET = "test_document_review_bucket"
-MOCK_EDGE_TABLE = "test_edge_reference_table"
-
 MOCK_EDGE_REFERENCE_TABLE = "test_edge_reference_table"
 
+
 @pytest.fixture
 def set_env(monkeypatch):
     monkeypatch.setenv("AWS_DEFAULT_REGION", REGION_NAME)
@@ -230,12 +229,12 @@ def set_env(monkeypatch):
     monkeypatch.setenv("SLACK_CHANNEL_ID", MOCK_ALERTING_SLACK_CHANNEL_ID)
     monkeypatch.setenv("ITOC_TESTING_ODS_CODES", MOCK_ITOC_ODS_CODES)
     monkeypatch.setenv("DOCUMENT_REVIEW_DYNAMODB_NAME", MOCK_DOCUMENT_REVIEW_TABLE)
-    monkeypatch.setenv("DOCUMENT_REVIEW_S3_BUCKET_NAME", MOCK_DOCUMENT_REVIEW_BUCKET)
-    monkeypatch.setenv("EDGE_REFERENCE_TABLE", MOCK_EDGE_TABLE)
+    monkeypatch.setenv("PENDING_REVIEW_BUCKET_NAME", MOCK_DOCUMENT_REVIEW_BUCKET)
     monkeypatch.setenv("STAGING_STORE_BUCKET_NAME", MOCK_STAGING_STORE_BUCKET)
     monkeypatch.setenv("METADATA_SQS_QUEUE_URL", MOCK_LG_METADATA_SQS_QUEUE)
     monkeypatch.setenv("EDGE_REFERENCE_TABLE", MOCK_EDGE_REFERENCE_TABLE)
 
+
 EXPECTED_PARSED_PATIENT_BASE_CASE = PatientDetails(
     givenName=["Jane"],
     familyName="Smith",