Add support for moving files to the new lambda function

Author: dlzhry2nhs

batch_processor_filter/src/batch_file_repository.py

@@ -0,0 +1,60 @@
+"""Module for the batch file repository"""
+from csv import writer
+from io import StringIO, BytesIO
+
+import boto3
+
+from batch_file_created_event import BatchFileCreatedEvent
+from constants import SOURCE_BUCKET_NAME, ACK_BUCKET_NAME
+
+
+class BatchFileRepository:
+    """Repository class to handle interactions with batch file interactions e.g. managing the source and ack files"""
+    _ARCHIVE_FILE_DIR: str = "archive"
+    _SOURCE_BUCKET_NAME: str = SOURCE_BUCKET_NAME
+    _ACK_BUCKET_NAME: str = ACK_BUCKET_NAME
+
+    def __init__(self):
+        self._s3_client = boto3.client('s3')
+
+    @staticmethod
+    def _create_ack_failure_data(batch_file_created_event: BatchFileCreatedEvent) -> dict:
+        return {
+            "MESSAGE_HEADER_ID": batch_file_created_event["message_id"],
+            "HEADER_RESPONSE_CODE": "Failure",
+            "ISSUE_SEVERITY": "Fatal",
+            "ISSUE_CODE": "Fatal Error",
+            "ISSUE_DETAILS_CODE": "10001",
+            "RESPONSE_TYPE": "Technical",
+            "RESPONSE_CODE": "10002",
+            "RESPONSE_DISPLAY": "Infrastructure Level Response Value - Processing Error",
+            "RECEIVED_TIME": batch_file_created_event["created_at_formatted_string"],
+            "MAILBOX_FROM": "",  # TODO: Leave blank for DPS, add mailbox if from mesh mailbox
+            "LOCAL_ID": "",  # TODO: Leave blank for DPS, add from ctl file if data picked up from MESH mailbox
+            "MESSAGE_DELIVERY": False,
+        }
+
+    def move_source_file_to_archive(self, file_key: str) -> None:
+        self._s3_client.copy_object(
+            Bucket=self._SOURCE_BUCKET_NAME,
+            CopySource={"Bucket": self._SOURCE_BUCKET_NAME, "Key": file_key},
+            Key=f"{self._ARCHIVE_FILE_DIR}/{file_key}"
+        )
+        self._s3_client.delete_object(Bucket=self._SOURCE_BUCKET_NAME, Key=file_key)
+
+    def upload_failure_ack(self, batch_file_created_event: BatchFileCreatedEvent) -> None:
+        ack_failure_data = self._create_ack_failure_data(batch_file_created_event)
+
+        ack_filename = ("ack/" + batch_file_created_event["filename"]
+                        .replace(".csv", f"_InfAck_{batch_file_created_event['created_at_formatted_string']}.csv"))
+
+        # Create CSV file with | delimiter, filetype .csv
+        csv_buffer = StringIO()
+        csv_writer = writer(csv_buffer, delimiter="|")
+        csv_writer.writerow(list(ack_failure_data.keys()))
+        csv_writer.writerow(list(ack_failure_data.values()))
+
+        # Upload the CSV file to S3
+        csv_buffer.seek(0)
+        csv_bytes = BytesIO(csv_buffer.getvalue().encode("utf-8"))
+        self._s3_client.upload_fileobj(csv_bytes, self._ACK_BUCKET_NAME, ack_filename)

batch_processor_filter/src/batch_processor_filter_service.py

@@ -5,6 +5,7 @@
 
 from batch_audit_repository import BatchAuditRepository
 from batch_file_created_event import BatchFileCreatedEvent
+from batch_file_repository import BatchFileRepository
 from constants import REGION_NAME, FileStatus, QUEUE_URL
 from exceptions import EventAlreadyProcessingForSupplierAndVaccTypeError
 from send_log_to_firehose import send_log_to_firehose
@@ -16,8 +17,13 @@
 
 class BatchProcessorFilterService:
     """Batch processor filter service class. Provides the business logic for the Lambda function"""
-    def __init__(self, audit_repo: BatchAuditRepository = BatchAuditRepository()):
+    def __init__(
+        self,
+        audit_repo: BatchAuditRepository = BatchAuditRepository(),
+        batch_file_repo: BatchFileRepository = BatchFileRepository()
+    ):
         self._batch_audit_repository = audit_repo
+        self._batch_file_repo = batch_file_repo
         self._queue_client = boto3.client('sqs', region_name=REGION_NAME)
 
     def _is_duplicate_file(self, file_key: str) -> bool:
@@ -34,6 +40,8 @@ def apply_filter(self, batch_file_created_event: BatchFileCreatedEvent) -> None:
             # Mark as processed and return without error so next event will be picked up from queue
             logger.info("A duplicate file has already been processed. Filename: %s", filename)
             self._batch_audit_repository.update_status(message_id, FileStatus.DUPLICATE)
+            self._batch_file_repo.upload_failure_ack(batch_file_created_event)
+            self._batch_file_repo.move_source_file_to_archive(filename)
             return
 
         if self._batch_audit_repository.is_event_processing_for_supplier_and_vacc_type(supplier, vaccine_type):
@@ -49,6 +57,6 @@ def apply_filter(self, batch_file_created_event: BatchFileCreatedEvent) -> None:
         )
         self._batch_audit_repository.update_status(message_id, FileStatus.PROCESSING)
 
-        successful_log_message = "File forwarded for processing by ECS"
+        successful_log_message = f"File forwarded for processing by ECS. Filename: {filename}"
         logger.info(successful_log_message)
         send_log_to_firehose({**batch_file_created_event, "message": successful_log_message})

batch_processor_filter/src/constants.py

@@ -7,6 +7,8 @@
 AUDIT_TABLE_QUEUE_NAME_GSI = os.getenv("QUEUE_NAME_GSI")
 QUEUE_URL = os.getenv("QUEUE_URL")
 SPLUNK_FIREHOSE_STREAM_NAME = os.getenv("SPLUNK_FIREHOSE_NAME")
+SOURCE_BUCKET_NAME = os.getenv("SOURCE_BUCKET_NAME")
+ACK_BUCKET_NAME = os.getenv("ACK_BUCKET_NAME")
 
 
 class FileStatus(StrEnum):

batch_processor_filter/tests/test_lambda_handler.py

@@ -5,7 +5,8 @@
 from unittest import TestCase
 from unittest.mock import patch
 
-from moto import mock_dynamodb, mock_sqs
+import botocore
+from moto import mock_dynamodb, mock_sqs, mock_s3
 
 from batch_file_created_event import BatchFileCreatedEvent
 from exceptions import InvalidBatchSizeError, EventAlreadyProcessingForSupplierAndVaccTypeError
@@ -18,10 +19,12 @@
 
 sqs_client = boto3.client("sqs", region_name=REGION_NAME)
 dynamodb_client = boto3.client("dynamodb", region_name=REGION_NAME)
+s3_client = boto3.client("s3", region_name=REGION_NAME)
 
 
 @mock_dynamodb
 @mock_sqs
+@mock_s3
 class TestLambdaHandler(TestCase):
     default_batch_file_event: BatchFileCreatedEvent = BatchFileCreatedEvent(
         message_id="df0b745c-b8cb-492c-ba84-8ea28d9f51d5",
@@ -32,6 +35,8 @@ class TestLambdaHandler(TestCase):
         created_at_formatted_string="20250826T14372600"
     )
     mock_queue_url = MOCK_ENVIRONMENT_DICT.get("QUEUE_URL")
+    mock_source_bucket = MOCK_ENVIRONMENT_DICT.get("SOURCE_BUCKET_NAME")
+    mock_ack_bucket = MOCK_ENVIRONMENT_DICT.get("ACK_BUCKET_NAME")
 
     def setUp(self):
         dynamodb_client.create_table(
@@ -66,6 +71,12 @@ def setUp(self):
             "FifoQueue": "true",
             "ContentBasedDeduplication": "true"
         })
+
+        for bucket_name in [self.mock_source_bucket, self.mock_ack_bucket]:
+            s3_client.create_bucket(
+                Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": REGION_NAME}
+            )
+
         self.logger_patcher = patch("batch_processor_filter_service.logger")
         self.mock_logger = self.logger_patcher.start()
         self.firehose_log_patcher = patch("batch_processor_filter_service.send_log_to_firehose")
@@ -74,9 +85,31 @@ def setUp(self):
     def tearDown(self):
         dynamodb_client.delete_table(TableName=AUDIT_TABLE_NAME)
         sqs_client.delete_queue(QueueUrl=self.mock_queue_url)
+
+        for bucket_name in [self.mock_source_bucket, self.mock_ack_bucket]:
+            for obj in s3_client.list_objects_v2(Bucket=bucket_name).get("Contents", []):
+                # Must delete objects before bucket can be deleted
+                s3_client.delete_object(Bucket=bucket_name, Key=obj["Key"])
+            s3_client.delete_bucket(Bucket=bucket_name)
+
         self.logger_patcher.stop()
         self.firehose_log_patcher.stop()
 
+    def _assert_source_file_moved(self, filename: str):
+        """Check used in the duplicate scenario to validate that the original uploaded file is moved"""
+        with self.assertRaises(botocore.exceptions.ClientError) as exc:
+            s3_client.get_object(Bucket=self.mock_source_bucket, Key=filename)
+
+        self.assertEqual(str(exc.exception), "An error occurred (NoSuchKey) when calling the GetObject "
+                                             "operation: The specified key does not exist.")
+        archived_object = s3_client.get_object(Bucket=self.mock_source_bucket, Key=f"archive/{filename}")
+        self.assertIsNotNone(archived_object)
+
+    def _assert_ack_file_created(self, ack_file_key: str):
+        """Check used in duplicate scenario to validate that the failure ack was created"""
+        ack_file = s3_client.get_object(Bucket=self.mock_ack_bucket, Key=f"ack/{ack_file_key}")
+        self.assertIsNotNone(ack_file)
+
     def test_lambda_handler_raises_error_when_empty_batch_received(self):
         with self.assertRaises(InvalidBatchSizeError) as exc:
             lambda_handler({"Records": []}, {})
@@ -98,21 +131,27 @@ def test_lambda_handler_handles_duplicate_file_scenario(self):
         add_entry_to_mock_table(dynamodb_client, AUDIT_TABLE_NAME, self.default_batch_file_event, FileStatus.PROCESSED)
         duplicate_file_event = copy.deepcopy(self.default_batch_file_event)
         duplicate_file_event["message_id"] = "fc9008b7-3865-4dcf-88b8-fc4abafff5f8"
+        test_file_name = duplicate_file_event["filename"]
 
         # Add the audit record for the incoming event
         add_entry_to_mock_table(dynamodb_client, AUDIT_TABLE_NAME, duplicate_file_event, FileStatus.QUEUED)
 
+        # Create the source file in S3
+        s3_client.put_object(Bucket=self.mock_source_bucket, Key=test_file_name)
+
         lambda_handler({"Records": [make_sqs_record(duplicate_file_event)]}, {})
 
         status = get_audit_entry_status_by_id(dynamodb_client, AUDIT_TABLE_NAME, duplicate_file_event["message_id"])
         self.assertEqual(status, "Not processed - duplicate")
 
         sqs_messages = sqs_client.receive_message(QueueUrl=self.mock_queue_url)
         self.assertEqual(sqs_messages.get("Messages", []), [])
+        self._assert_source_file_moved(test_file_name)
+        self._assert_ack_file_created("Menacwy_Vaccinations_v5_TEST_20250820T10210000_InfAck_20250826T14372600.csv")
 
         self.mock_logger.info.assert_called_once_with(
             "A duplicate file has already been processed. Filename: %s",
-            "Menacwy_Vaccinations_v5_TEST_20250820T10210000.csv"
+            test_file_name
         )
 
     def test_lambda_handler_raises_error_when_event_already_processing_for_supplier_and_vacc_type(self):
@@ -164,7 +203,8 @@ def test_lambda_handler_processes_event_successfully(self):
         self.assertEqual(len(sqs_messages.get("Messages", [])), 1)
         self.assertDictEqual(json.loads(sqs_messages["Messages"][0]["Body"]), dict(self.default_batch_file_event))
 
-        expected_log_message = "File forwarded for processing by ECS"
+        expected_log_message = (f"File forwarded for processing by ECS. Filename: "
+                                f"{self.default_batch_file_event['filename']}")
         self.mock_logger.info.assert_called_once_with(expected_log_message)
         self.mock_firehose_send_log.assert_called_once_with(
             {**self.default_batch_file_event, "message": expected_log_message}
@@ -194,7 +234,7 @@ def test_lambda_handler_processes_event_successfully_when_event_for_same_supplie
         self.assertEqual(len(sqs_messages.get("Messages", [])), 1)
         self.assertDictEqual(json.loads(sqs_messages["Messages"][0]["Body"]), dict(test_event))
 
-        expected_log_message = "File forwarded for processing by ECS"
+        expected_log_message = f"File forwarded for processing by ECS. Filename: {test_event['filename']}"
         self.mock_logger.info.assert_called_once_with(expected_log_message)
         self.mock_firehose_send_log.assert_called_once_with(
             {**test_event, "message": expected_log_message}

batch_processor_filter/tests/testing_utils.py

@@ -8,7 +8,9 @@
     "AUDIT_TABLE_NAME": "immunisation-batch-internal-dev-audit-table",
     "QUEUE_URL": "https://sqs.eu-west-2.amazonaws.com/123456789012/imms-batch-metadata-queue.fifo",
     "FILE_NAME_GSI": "filename_index",
-    "QUEUE_NAME_GSI": "queue_name_index"
+    "QUEUE_NAME_GSI": "queue_name_index",
+    "SOURCE_BUCKET_NAME": "immunisation-batch-internal-dev-data-sources",
+    "ACK_BUCKET_NAME": "immunisation-batch-internal-dev-data-destinations"
 }
 
 

terraform/batch_processor_filter_lambda.tf

@@ -122,6 +122,32 @@ resource "aws_iam_policy" "batch_processor_filter_lambda_exec_policy" {
           "firehose:PutRecordBatch"
         ],
         "Resource" : "arn:aws:firehose:*:*:deliverystream/${module.splunk.firehose_stream_name}"
+      },
+      {
+        Effect = "Allow"
+        Action = [
+          "s3:GetObject",
+          "s3:ListBucket",
+          "s3:PutObject",
+          "s3:CopyObject",
+          "s3:DeleteObject"
+        ]
+        Resource = [
+          aws_s3_bucket.batch_data_source_bucket.arn,
+          "${aws_s3_bucket.batch_data_source_bucket.arn}/*"
+        ]
+      },
+      {
+        Effect = "Allow"
+        Action = [
+          "s3:GetObject",
+          "s3:PutObject",
+          "s3:ListBucket"
+        ]
+        Resource = [
+          aws_s3_bucket.batch_data_destination_bucket.arn,
+          "${aws_s3_bucket.batch_data_destination_bucket.arn}/*"
+        ]
       }
     ]
   })
@@ -247,11 +273,13 @@ resource "aws_lambda_function" "batch_processor_filter_lambda" {
 
   environment {
     variables = {
-      QUEUE_URL                  = aws_sqs_queue.supplier_fifo_queue.url
-      SPLUNK_FIREHOSE_NAME       = module.splunk.firehose_stream_name
-      AUDIT_TABLE_NAME           = aws_dynamodb_table.audit-table.name
-      FILE_NAME_GSI              = "filename_index"
-      QUEUE_NAME_GSI             = "queue_name_index"
+      SOURCE_BUCKET_NAME   = aws_s3_bucket.batch_data_source_bucket.bucket
+      ACK_BUCKET_NAME      = aws_s3_bucket.batch_data_destination_bucket.bucket
+      QUEUE_URL            = aws_sqs_queue.supplier_fifo_queue.url
+      SPLUNK_FIREHOSE_NAME = module.splunk.firehose_stream_name
+      AUDIT_TABLE_NAME     = aws_dynamodb_table.audit-table.name
+      FILE_NAME_GSI        = "filename_index"
+      QUEUE_NAME_GSI       = "queue_name_index"
     }
   }
   kms_key_arn                    = data.aws_kms_key.existing_lambda_encryption_key.arn

terraform/sqs_batch_file_created.tf

@@ -3,5 +3,5 @@ resource "aws_sqs_queue" "batch_file_created" {
   name                        = "${local.short_prefix}-batch-file-created-queue.fifo"
   fifo_queue                  = true
   content_based_deduplication = true # Optional, helps with deduplication
-  visibility_timeout_seconds  = 900 # TODO - discuss and refine both this, max receives and DLQ
+  visibility_timeout_seconds  = 60 # TODO - discuss and refine this + DDB TTL
 }