feat: DTOSS-10479 add nems poison container and logic

application/CohortManager/compose.core.yaml

@@ -53,8 +53,9 @@ services:
     profiles: [non-essential]
     environment:
       - ASPNETCORE_URLS=http://*:9083
-      - caasfolder_STORAGE=${AZURITE_CONNECTION_STRING}
-      - NemsMessages="nems-messages"
+      - nemsmeshfolder_STORAGE=${AZURITE_CONNECTION_STRING}
+      - NemsMessages=nems-updates
+      - NemsPoisonContainer=nems-poison
       - ExceptionFunctionURL=http://create-exception:7070/api/CreateException
       - RetrievePdsDemographicURL=http://retrieve-pds-demographic:8082/api/RetrievePDSDemographic
       - UnsubscribeNemsSubscriptionUrl=http://manage-nems-subscription:9081/api/Unsubscribe

application/CohortManager/src/Functions/Functions.sln

@@ -233,6 +233,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "PdsProcesserTests", "PdsPro
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "PdsProcessorTests", "..\..\..\..\tests\UnitTests\PdsProcessorTests\PdsProcessorTests.csproj", "{392B3D99-C5C5-DB9F-4DCA-F389E679C7C0}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "BlobStorageHelperTests", "..\..\..\..\tests\UnitTests\SharedTests\BlobStorageHelperTests\BlobStorageHelperTests.csproj", "{BFA68329-98DD-4039-B009-C6DF23146765}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -1371,6 +1373,30 @@ Global
 		{59CBDBE5-29BE-F38C-80E6-40843F2F8AF6}.Release|x64.Build.0 = Release|Any CPU
 		{59CBDBE5-29BE-F38C-80E6-40843F2F8AF6}.Release|x86.ActiveCfg = Release|Any CPU
 		{59CBDBE5-29BE-F38C-80E6-40843F2F8AF6}.Release|x86.Build.0 = Release|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Debug|x64.Build.0 = Debug|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Debug|x86.Build.0 = Debug|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Release|Any CPU.Build.0 = Release|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Release|x64.ActiveCfg = Release|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Release|x64.Build.0 = Release|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Release|x86.ActiveCfg = Release|Any CPU
+		{FC22C311-57DD-B069-4041-AD2AC8F80B5D}.Release|x86.Build.0 = Release|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Debug|x64.Build.0 = Debug|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Debug|x86.Build.0 = Debug|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Release|Any CPU.Build.0 = Release|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Release|x64.ActiveCfg = Release|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Release|x64.Build.0 = Release|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Release|x86.ActiveCfg = Release|Any CPU
+		{BFA68329-98DD-4039-B009-C6DF23146765}.Release|x86.Build.0 = Release|Any CPU
 		{392B3D99-C5C5-DB9F-4DCA-F389E679C7C0}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{392B3D99-C5C5-DB9F-4DCA-F389E679C7C0}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{392B3D99-C5C5-DB9F-4DCA-F389E679C7C0}.Debug|x64.ActiveCfg = Debug|Any CPU

application/CohortManager/src/Functions/NemsSubscriptionService/ProcessNemsUpdate/ProcessNemsUpdate.cs

@@ -13,7 +13,6 @@
 using Model;
 using DataServices.Client;
 using System.Net;
-using FluentValidation.Validators;
 
 public class ProcessNemsUpdate
 {
@@ -24,6 +23,7 @@ public class ProcessNemsUpdate
     private readonly IHttpClientFunction _httpClientFunction;
     private readonly IExceptionHandler _exceptionHandler;
     private readonly IDataServiceClient<ParticipantDemographic> _participantDemographic;
+    private readonly IBlobStorageHelper _blobStorageHelper;
     private readonly ProcessNemsUpdateConfig _config;
     private long nhsNumberLong;
 
@@ -35,7 +35,8 @@ public ProcessNemsUpdate(
         IHttpClientFunction httpClientFunction,
         IExceptionHandler exceptionHandler,
         IDataServiceClient<ParticipantDemographic> participantDemographic,
-        IOptions<ProcessNemsUpdateConfig> processNemsUpdateConfig)
+        IOptions<ProcessNemsUpdateConfig> processNemsUpdateConfig,
+        IBlobStorageHelper blobStorageHelper)
     {
         _logger = logger;
         _fhirPatientDemographicMapper = fhirPatientDemographicMapper;
@@ -45,6 +46,7 @@ public ProcessNemsUpdate(
         _exceptionHandler = exceptionHandler;
         _participantDemographic = participantDemographic;
         _config = processNemsUpdateConfig.Value;
+        _blobStorageHelper = blobStorageHelper;
     }
 
     /// <summary>
@@ -66,18 +68,26 @@ public async Task Run([BlobTrigger("nems-updates/{name}", Connection = "nemsmesh
         {
             var nhsNumber = await GetNhsNumberFromFile(blobStream, name);
 
-            if (!ValidationHelper.ValidateNHSNumber(nhsNumber!))
+            if (nhsNumber == null)
             {
-                _logger.LogError("There was a problem parsing the NHS number from blob store in the ProcessNemsUpdate function");
-                throw new InvalidDataException("Invalid NHS Number");
+                _logger.LogError("No NHS number found in file {FileName}. Moving to poison container.", name);
+                await CopyToPoisonContainer(name);
+                return;
+            }
+
+            if (!ValidationHelper.ValidateNHSNumber(nhsNumber))
+            {
+                _logger.LogError("There was a problem validating the NHS number from blob store in the ProcessNemsUpdate function for file {FileName}. Moving to poison container.", name);
+                await CopyToPoisonContainer(name);
+                return;
             }
             nhsNumberLong = long.Parse(nhsNumber!);
 
-            var pdsResponse = await RetrievePdsRecord(nhsNumber!);
+            var pdsResponse = await RetrievePdsRecord(nhsNumber);
             if (pdsResponse!.StatusCode == HttpStatusCode.NotFound)
             {
-                _logger.LogError("the PDS function has returned a 404 error. function now stopping processing");
-                // we can stop processing here as we know that not found means the participant ether needed an update or they were actually not found
+                _logger.LogError("the PDS function has returned a 404 error for file {FileName}. Moving file to poison container.", name);
+                await CopyToPoisonContainer(name);
                 return;
             }
 
@@ -92,15 +102,27 @@ public async Task Run([BlobTrigger("nems-updates/{name}", Connection = "nemsmesh
             }
             else
             {
-                await UnsubscribeFromNems(nhsNumber!, retrievedPdsRecord!);
+                await UnsubscribeFromNems(nhsNumber, retrievedPdsRecord!);
             }
-
         }
         catch (Exception ex)
         {
-            _logger.LogError(ex, "There was an error processing NEMS update.");
+            _logger.LogError(ex, "There was an error processing NEMS update for file {FileName}. Moving to poison container.", name);
+            try
+            {
+                await CopyToPoisonContainer(name);
+            }
+            catch (Exception poisonEx)
+            {
+                _logger.LogError(poisonEx, "Failed to copy NEMS file {FileName} to poison container. Manual intervention required.", name);
+            }
         }
+    }
 
+    private async Task CopyToPoisonContainer(string fileName)
+    {
+        await _blobStorageHelper.CopyFileToPoisonAsync(_config.nemsmeshfolder_STORAGE, fileName, _config.NemsMessages, _config.NemsPoisonContainer, addTimestamp: true);
+        _logger.LogInformation("Copied failed NEMS file {FileName} to poison container with timestamp.", fileName);
     }
 
     private async Task UnsubscribeFromNems(string nhsNumber, PdsDemographic retrievedPdsRecord)

application/CohortManager/src/Functions/NemsSubscriptionService/ProcessNemsUpdate/ProcessNemsUpdateConfig.cs

@@ -17,4 +17,7 @@ public class ProcessNemsUpdateConfig
     [Required]
     public required string DemographicDataServiceURL { get; set; }
 
+    [Required]
+    public required string nemsmeshfolder_STORAGE { get; set; }
+    public string NemsPoisonContainer { get; set; } = "nems-poison";
 }

application/CohortManager/src/Functions/NemsSubscriptionService/ProcessNemsUpdate/Program.cs

@@ -18,6 +18,7 @@
         services.AddSingleton<IFhirPatientDemographicMapper, FhirPatientDemographicMapper>();
         services.AddScoped<ICreateBasicParticipantData, CreateBasicParticipantData>();
         services.AddScoped<IAddBatchToQueue, AddBatchToQueue>();
+        services.AddScoped<IBlobStorageHelper, BlobStorageHelper>();
         services.AddBlobStorageHealthCheck("ProcessNemsUpdate");
     })
     .AddTelemetry()

application/CohortManager/src/Functions/Shared/Common/BlobstorageHelper.cs

@@ -15,6 +15,13 @@ public BlobStorageHelper(ILogger<BlobStorageHelper> logger)
         _logger = logger;
     }
     public async Task CopyFileToPoisonAsync(string connectionString, string fileName, string containerName)
+    {
+        // Delegate to the extended overload to avoid duplication; preserve env var behaviour
+        var poisonContainerName = Environment.GetEnvironmentVariable("fileExceptions");
+        await CopyFileToPoisonAsync(connectionString, fileName, containerName, poisonContainerName, addTimestamp: false);
+    }
+
+    public async Task CopyFileToPoisonAsync(string connectionString, string fileName, string containerName, string poisonContainerName, bool addTimestamp = false)
     {
         var sourceBlobServiceClient = new BlobServiceClient(connectionString);
         var sourceContainerClient = sourceBlobServiceClient.GetBlobContainerClient(containerName);
@@ -23,8 +30,19 @@ public async Task CopyFileToPoisonAsync(string connectionString, string fileName
         BlobLeaseClient sourceBlobLease = new(sourceBlobClient);
 
         var destinationBlobServiceClient = new BlobServiceClient(connectionString);
-        var destinationContainerClient = destinationBlobServiceClient.GetBlobContainerClient(Environment.GetEnvironmentVariable("fileExceptions"));
-        var destinationBlobClient = destinationContainerClient.GetBlobClient(fileName);
+        var destinationContainerClient = destinationBlobServiceClient.GetBlobContainerClient(poisonContainerName);
+
+        // Conditionally add timestamp to prevent collisions and maintain audit trail
+        var destinationFileName = fileName;
+        if (addTimestamp)
+        {
+            var timestamp = DateTime.UtcNow.ToString("yyyyMMdd_HHmmss");
+            var fileExtension = Path.GetExtension(fileName);
+            var fileNameWithoutExtension = Path.GetFileNameWithoutExtension(fileName);
+            destinationFileName = $"{fileNameWithoutExtension}_{timestamp}{fileExtension}";
+        }
+
+        var destinationBlobClient = destinationContainerClient.GetBlobClient(destinationFileName);
 
         await destinationContainerClient.CreateIfNotExistsAsync(PublicAccessType.None);
 
@@ -36,7 +54,7 @@ public async Task CopyFileToPoisonAsync(string connectionString, string fileName
         catch (RequestFailedException ex)
         {
             _logger.LogError(ex, "There has been a problem while copying the file: {Message}", ex.Message);
-            throw;
+            throw new InvalidOperationException($"Failed to copy file '{fileName}' from container '{containerName}' to poison container as '{destinationFileName}'.", ex);
         }
         finally
         {

application/CohortManager/src/Functions/Shared/Common/Interfaces/IBlobstorageHelper.cs

@@ -5,6 +5,7 @@ namespace Common;
 public interface IBlobStorageHelper
 {
     Task CopyFileToPoisonAsync(string connectionString, string fileName, string containerName);
+    Task CopyFileToPoisonAsync(string connectionString, string fileName, string containerName, string poisonContainerName, bool addTimestamp = false);
 
     Task<bool> UploadFileToBlobStorage(string connectionString, string containerName, BlobFile blobFile, bool overwrite = false);
 

infrastructure/tf-core/environments/development.tfvars

@@ -438,6 +438,10 @@ function_apps = {
         {
           env_var_name   = "NemsMessages"
           container_name = "nems-updates"
+        },
+        {
+          env_var_name   = "NemsPoisonContainer"
+          container_name = "nems-poison"
         }
       ]
       env_vars_static = {
@@ -1348,6 +1352,9 @@ storage_accounts = {
       nems-config = {
         container_name = "nems-config"
       }
+      nems-poison = {
+        container_name = "nems-poison"
+      }
     }
   }
 }

infrastructure/tf-core/environments/integration.tfvars

@@ -344,6 +344,10 @@ function_apps = {
         {
           env_var_name   = "NemsMessages"
           container_name = "nems-updates"
+        },
+        {
+          env_var_name   = "NemsPoisonContainer"
+          container_name = "nems-poison"
         }
       ]
       env_vars_static = {
@@ -1253,6 +1257,9 @@ storage_accounts = {
       nems-config = {
         container_name = "nems-config"
       }
+      nems-poison = {
+        container_name = "nems-poison"
+      }
     }
   }
 }

infrastructure/tf-core/environments/nft.tfvars

@@ -437,6 +437,10 @@ function_apps = {
         {
           env_var_name   = "NemsMessages"
           container_name = "nems-updates"
+        },
+        {
+          env_var_name   = "NemsPoisonContainer"
+          container_name = "nems-poison"
         }
       ]
       env_vars_static = {
@@ -1346,6 +1350,9 @@ storage_accounts = {
       nems-config = {
         container_name = "nems-config"
       }
+      nems-poison = {
+        container_name = "nems-poison"
+      }
     }
   }
 }

infrastructure/tf-core/environments/preprod.tfvars

@@ -350,6 +350,10 @@ function_apps = {
         {
           env_var_name   = "NemsMessages"
           container_name = "nems-updates"
+        },
+        {
+          env_var_name   = "NemsPoisonContainer"
+          container_name = "nems-poison"
         }
       ]
       env_vars_static = {
@@ -1267,6 +1271,9 @@ storage_accounts = {
       nems-config = {
         container_name = "nems-config"
       }
+      nems-poison = {
+        container_name = "nems-poison"
+      }
     }
   }
 }

infrastructure/tf-core/environments/production.tfvars

@@ -335,6 +335,10 @@ function_apps = {
         {
           env_var_name   = "NemsMessages"
           container_name = "nems-updates"
+        },
+        {
+          env_var_name   = "NemsPoisonContainer"
+          container_name = "nems-poison"
         }
       ]
       env_vars_static = {
@@ -1249,6 +1253,9 @@ storage_accounts = {
       nems-config = {
         container_name = "nems-config"
       }
+      nems-poison = {
+        container_name = "nems-poison"
+      }
     }
   }
 }

infrastructure/tf-core/environments/sandbox.tfvars

@@ -358,6 +358,10 @@ function_apps = {
         {
           env_var_name   = "NemsMessages"
           container_name = "nems-updates"
+        },
+        {
+          env_var_name   = "NemsPoisonContainer"
+          container_name = "nems-poison"
         }
       ]
       env_vars_static = {
@@ -1547,6 +1551,9 @@ storage_accounts = {
       nems-config = {
         container_name = "nems-config"
       }
+      nems-poison = {
+        container_name = "nems-poison"
+      }
     }
   }
 }