Merge branch 'master' into ved-1252-list-store-fix

Author: Thomas-Boyle

.github/pull_request_template.md

@@ -1,25 +1,7 @@
-## Summary
+## PR Description
 
-- Routine Change
-- :exclamation: Breaking Change
-- :robot: Operational or Infrastructure Change
-- :sparkles: New Feature
-- :warning: Potential issues that might be caused by this change
+Description of the changes made.
 
-Add any other relevant notes or explanations here. **Remove this line if you have nothing to add.**
+## How were the changes tested
 
-## Reviews Required
-
-- [x] Dev
-- [ ] Test
-- [ ] Tech Author
-- [ ] Product Owner
-
-## Review Checklist
-
-:information_source: This section is to be filled in by the **reviewer**.
-
-- [ ] I have reviewed the changes in this PR and they fill all of the acceptance criteria of the ticket.
-- [ ] If there were infrastructure, operational, or build changes, I have made sure there is sufficient evidence that the changes will work.
-- [ ] If there were changes that are outside of the regular release processes e.g. account infrastructure to setup, manual setup for external API integrations, secrets to set, then I have checked that the developer has flagged this to the Tech Lead as release steps.
-- [ ] I have checked that no Personal Identifiable Data (PID) is logged as part of the changes.
+Describe how the changes were tested

.github/workflows/deploy-lambda-artifact.yml

@@ -239,7 +239,7 @@ jobs:
 
       - name: Set up Docker Buildx
         if: ${{ steps.decide.outputs.deployment_mode == 'build' && !steps.build-check.outputs.existing_image_digest }}
-        uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f
+        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd
 
       - name: Build and publish image with layer caching
         id: build-image

.github/workflows/quality-checks.yml

@@ -257,7 +257,7 @@ jobs:
           fi
 
       - name: SonarCloud Scan
-        uses: SonarSource/sonarqube-scan-action@299e4b793aaa83bf2aba7c9c14bedbb485688ec4
+        uses: SonarSource/sonarqube-scan-action@55e44800a8f495208cce6e4e82f5dedb45fcf0ef
         env:
           GITHUB_TOKEN: ${{ github.token }} # Needed to get PR information, if any
           SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}

docs/architecture/system-overview.md

@@ -0,0 +1,76 @@
+# System Overview
+
+This page gives a high-level view of the Immunisation FHIR API runtime architecture.
+
+It focuses on the API path, batch ingestion path, outbound notification flow, runtime configuration, and NHS number change handling.
+
+## High-Level Diagram
+
+```mermaid
+flowchart LR
+    subgraph Ingress[Ingress and API]
+        Suppliers[Supplier systems] --> Apigee[Apigee proxy\nOAuth, rate limiting, supplier header]
+        Apigee --> ApiGw[AWS API Gateway HTTP API]
+        ApiGw --> Backend[Backend API Lambdas\nCRUD, search, status]
+        Backend --> IEDS[(IEDS DynamoDB\nImmunisation event store)]
+    end
+
+    subgraph Batch[Batch ingestion]
+        SupplierFiles[Supplier batch files in S3] --> Filename[Filename Processor Lambda]
+        Mesh[MESH mailbox bucket] --> MeshProc[Mesh Processor Lambda]
+        MeshProc --> Filename
+        Filename --> BatchCreated[SQS FIFO\nbatch-file-created]
+        BatchCreated --> BatchFilter[Batch Processor Filter Lambda]
+        BatchFilter --> SupplierQueue[SQS FIFO\nsupplier metadata queue]
+        SupplierQueue --> BatchPipe[EventBridge Pipe]
+        BatchPipe --> RecordProcessor[ECS Fargate Record Processor]
+        RecordProcessor --> Kinesis[Kinesis data stream]
+        Kinesis --> Forwarder[Record Forwarder Lambda]
+        Forwarder --> IEDS
+        Forwarder --> AckQueue[SQS FIFO\nack metadata queue]
+        AckQueue --> Ack[Ack Backend Lambda]
+    end
+
+    subgraph Outbound[Outbound notifications]
+        IEDS -->|DynamoDB stream| Delta[Delta Lambda]
+        Delta --> DeltaTable[(Delta DynamoDB)]
+        DeltaTable -->|DynamoDB stream| MnsPipe[EventBridge Pipe]
+        MnsPipe --> MnsQueue[SQS\nmns-outbound-events]
+        MnsQueue --> MnsPublisher[MNS Publisher Lambda]
+        MnsPublisher --> Subscribers[MNS subscribers]
+    end
+
+    subgraph Config[Runtime config]
+        ConfigBucket[S3 config bucket] --> RedisSync[Redis Sync Lambda]
+        RedisSync --> Redis[(Redis cache\npermissions, disease mappings, config)]
+        Redis --> Backend
+        Redis --> Filename
+        Redis --> RecordProcessor
+        Redis --> Forwarder
+    end
+
+    subgraph IdSync[Identity sync]
+        MnsIdEvent[MNS NHS number change event] --> IdQueue[SQS\nid-sync-queue]
+        IdQueue --> IdSyncLambda[ID Sync Lambda]
+        IdSyncLambda --> IEDS
+    end
+```
+
+## Key Runtime Stores
+
+| Store          | Purpose                                                             |
+| -------------- | ------------------------------------------------------------------- |
+| IEDS DynamoDB  | System of record for immunisation events                            |
+| Delta DynamoDB | Outbound change store derived from IEDS stream events               |
+| Redis          | Runtime cache for permissions, disease mappings, and related config |
+| Audit table    | Batch-processing control state, deduplication, and status tracking  |
+
+## Design Notes
+
+- The filename processor is the batch entry point for files placed in the source bucket.
+- The audit table is for deduplication, processing state, and ordering decisions.
+- The batch processor filter ensures only one event is processed at a time for a given supplier and vaccine-type combination.
+- The supplier metadata FIFO queue preserves ordering before work is dispatched to ECS through EventBridge Pipe.
+- ECS is used for record processing because batch row processing can be long-running.
+- The record forwarder is the component that applies processed batch changes to IEDS.
+- ACK creation is part of the batch lifecycle.

infrastructure/account/.terraform.lock.hcl

@@ -2,4 +2,4 @@
 
 GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE = "Unable to process request. Issue may be transient."
 # Maximum response size for an AWS Lambda function
-MAX_RESPONSE_SIZE_BYTES = 6 * 1024 * 1024
+MAX_SEARCH_RESPONSE_SIZE_BYTES = 1 * 1024 * 1024

infrastructure/instance/.terraform.lock.hcl

@@ -11,7 +11,7 @@
 from fhir.resources.R4B.identifier import Identifier
 
 from common.get_service_url import get_service_url
-from constants import MAX_RESPONSE_SIZE_BYTES
+from constants import MAX_SEARCH_RESPONSE_SIZE_BYTES
 from controller.aws_apig_event_utils import (
     get_multi_value_query_params,
     get_path_parameter,
@@ -89,12 +89,17 @@ def create_immunization(self, aws_event: APIGatewayProxyEventV1) -> dict:
         except JSONDecodeError as e:
             raise InvalidJsonError(message=str(f"Request's body contains malformed JSON: {e}"))
 
-        created_resource_id = self.fhir_service.create_immunization(immunisation, supplier_system)
+        created_resource_id, created_resource_version = self.fhir_service.create_immunization(
+            immunisation, supplier_system
+        )
 
         return create_response(
             status_code=201,
             body=None,
-            headers={"Location": f"{self._API_SERVICE_URL}/Immunization/{created_resource_id}", "E-Tag": "1"},
+            headers={
+                "Location": f"{self._API_SERVICE_URL}/Immunization/{created_resource_id}",
+                "E-Tag": str(created_resource_version),
+            },
         )
 
     @fhir_api_exception_handler
@@ -208,7 +213,7 @@ def _search_immunizations_by_target_disease(self, search_params: dict[str, list[
     def _create_search_response(self, search_bundle: Bundle) -> dict:
         search_response_json = search_bundle.json(use_decimal=True)
 
-        if len(search_response_json) > MAX_RESPONSE_SIZE_BYTES:
+        if len(search_response_json) > MAX_SEARCH_RESPONSE_SIZE_BYTES:
             raise TooManyResultsError("Search returned too many results. Please narrow down the search")
 
         prepared_search_bundle = self._prepare_search_bundle(search_response_json)

infrastructure/mesh/.terraform.lock.hcl

@@ -3,7 +3,7 @@
 import logging
 import os
 import uuid
-from typing import Any, cast
+from typing import Any
 from uuid import uuid4
 
 from fhir.resources.R4B.bundle import (
@@ -145,7 +145,7 @@ def get_immunization_and_version_by_id(self, imms_id: str, supplier_system: str)
 
         return Immunization.parse_obj(resource), str(immunization_metadata.resource_version)
 
-    def create_immunization(self, immunization: dict, supplier_system: str) -> Id:
+    def create_immunization(self, immunization: dict, supplier_system: str) -> tuple[Id, int]:
         if immunization.get("id") is not None:
             raise CustomValidationError("id field must not be present for CREATE operation")
 
@@ -156,16 +156,32 @@ def create_immunization(self, immunization: dict, supplier_system: str) -> Id:
         if not self.authoriser.authorise(supplier_system, ApiOperationCode.CREATE, {vaccination_type}):
             raise UnauthorizedVaxError()
 
-        # Set ID for the requested new record
-        immunization["id"] = str(uuid.uuid4())
+        identifier = Identifier.parse_obj(immunization["identifier"][0])
+        duplicate_identifier = f"{identifier.system}#{identifier.value}"
 
-        immunization_fhir_entity = Immunization.parse_obj(immunization)
-        identifier = cast(Identifier, immunization_fhir_entity.identifier[0])
+        existing_immunization_resource, existing_immunization_meta = (
+            self.immunization_repo.get_immunization_by_identifier(identifier)
+        )
+        if existing_immunization_resource:
+            if not existing_immunization_meta.is_deleted:
+                raise IdentifierDuplicationError(identifier=duplicate_identifier)
+
+            immunization_id = existing_immunization_resource["id"]
+            immunization["id"] = immunization_id
+            immunization_fhir_entity = Immunization.parse_obj(immunization)
+            updated_version = self.immunization_repo.update_immunization(
+                immunization_id,
+                immunization_fhir_entity,
+                existing_immunization_meta,
+                supplier_system,
+            )
+            return immunization_id, updated_version
 
-        if self.immunization_repo.check_immunization_identifier_exists(identifier.system, identifier.value):
-            raise IdentifierDuplicationError(identifier=f"{identifier.system}#{identifier.value}")
+        immunization["id"] = str(uuid.uuid4())
+        immunization_fhir_entity = Immunization.parse_obj(immunization)
 
-        return self.immunization_repo.create_immunization(immunization_fhir_entity, supplier_system)
+        created_id = self.immunization_repo.create_immunization(immunization_fhir_entity, supplier_system)
+        return created_id, 1
 
     def update_immunization(self, imms_id: str, immunization: dict, supplier_system: str, resource_version: int) -> int:
         self._validate_immunization(immunization)