Merge branch 'main' into Fix/retrieve-pds-demographic-service-bus-connection-string-explicit

Author: SamAinsworth-NHS

application/CohortManager/compose.core.yaml

@@ -140,6 +140,9 @@ services:
       - ParticipantManagementUrl=http://participant-management-data-service:7994/api/ParticipantManagementDataService
       - ParticipantDemographicDataServiceURL=http://participant-demographic-data-service:7993/api/ParticipantDemographicDataService/
       - ExceptionFunctionURL=http://create-exception:7070/api/CreateException
+      - ManageNemsSubscriptionUnsubscribeURL=http://manage-nems-subscription:9081/api/Unsubscribe
+      - ManageNemsSubscriptionSubscribeURL=http://manage-nems-subscription:9081/api/ManageNemsSubscriptionSubscribeURL
+      - RetrievePdsDemographicURL=http://etrieve-pds-demographic:8082/api/RetrievePdsDemographic
 
   delete-participant:
     container_name: delete-participant

application/CohortManager/src/Functions/CaasIntegration/receiveCaasFile/receiveCaasFileConfig.cs

@@ -19,7 +19,9 @@ public class ReceiveCaasFileConfig
     public string caasfolder_STORAGE { get; set; }
     [Required]
     public string inboundBlobName { get; set; }
+    [Required]
     public string ServiceBusConnectionString_client_internal { get; set; }
     public string GetOrchestrationStatusURL { get; set; }
+    [Required]
     public string ParticipantManagementTopic { get; set; }
 }

application/CohortManager/src/Functions/DemographicServices/ManageNemsSubscription/Extensions/CertificateExtensions.cs

@@ -1,9 +1,11 @@
+namespace NHS.CohortManager.DemographicServices;
+
 using System.Security.Cryptography.X509Certificates;
 using Azure.Identity;
 using Azure.Security.KeyVault.Certificates;
 using Microsoft.Extensions.Logging;
 
-namespace NHS.CohortManager.DemographicServices;
+
 
 public static class CertificateExtensions
 {
@@ -26,15 +28,16 @@ public static async Task<X509Certificate2> LoadNemsCertificateAsync(this ManageN
             var certResult = await certClient.DownloadCertificateAsync(config.NemsKeyName);
             return certResult.Value;
         }
-        
+
         if (!string.IsNullOrEmpty(config.NemsLocalCertPath))
         {
             logger.LogInformation("Loading NEMS certificate from local file");
             return !string.IsNullOrEmpty(config.NemsLocalCertPassword)
                 ? new X509Certificate2(config.NemsLocalCertPath, config.NemsLocalCertPassword)
                 : new X509Certificate2(config.NemsLocalCertPath);
+
         }
-        
+
         throw new InvalidOperationException("No certificate configuration found. Please configure either KeyVaultConnectionString or NemsLocalCertPath.");
     }
-}
+}

application/CohortManager/src/Functions/DemographicServices/ManageNemsSubscription/ManageNemsSubscriptionConfig.cs

@@ -83,14 +83,6 @@ public class ManageNemsSubscriptionConfig
     /// </summary>
     public bool NemsBypassServerCertificateValidation { get; set; } = false;
 
-    /// <summary>
-    /// Default event types to subscribe to
-    /// </summary>
-    public string[] NemsDefaultEventTypes { get; set; } = new[]
-    {
-        "pds-record-change-1"
-    };
-
     /// <summary>
     /// HTTP client timeout in seconds for NEMS API requests
     /// Default: 300 seconds (5 minutes)
@@ -101,5 +93,9 @@ public class ManageNemsSubscriptionConfig
     /// Custom validation to ensure either KeyVault or local cert is configured
     /// </summary>
     public bool IsValid => !string.IsNullOrEmpty(KeyVaultConnectionString) || !string.IsNullOrEmpty(NemsLocalCertPath);
+    /// <summary>
+    /// Bool to set the function to be in stubbed mode. Simulated responses from NEMS
+    /// </summary>
+    public bool IsStubbed { get; set; } = false;
 }
 

application/CohortManager/src/Functions/DemographicServices/ManageNemsSubscription/Program.cs

@@ -6,28 +6,37 @@
 using NHS.CohortManager.DemographicServices;
 using DataServices.Database;
 using Microsoft.Extensions.Logging;
+using System.Security.Cryptography.X509Certificates;
 
 var loggerFactory = LoggerFactory.Create(builder => builder.AddConsole());
 var logger = loggerFactory.CreateLogger("Program");
 
 var host = new HostBuilder();
 
+
+logger.LogInformation("Application Has Started");
 // Load configuration
 host.AddConfiguration<ManageNemsSubscriptionConfig>(out ManageNemsSubscriptionConfig config);
 
 var nemsConfig = config;
 
+X509Certificate2 nemsCertificate;
+
+
 // Load NEMS certificate up-front and inject into DI
-var nemsCertificate = await nemsConfig.LoadNemsCertificateAsync(logger);
+
+nemsCertificate = await nemsConfig.LoadNemsCertificateAsync(logger);
+
 
 host.ConfigureFunctionsWebApplication();
 host.AddHttpClient()
-    .AddNemsHttpClient()
+    .AddNemsHttpClient(nemsConfig.IsStubbed)
     .ConfigureServices(services =>
     {
         // Register NEMS certificate
         services.AddSingleton(nemsCertificate);
 
+
         // Register NEMS subscription manager
         services.AddScoped<NemsSubscriptionManager>();
 

application/CohortManager/src/Functions/DemographicServices/RetrievePDSDemographic/IPdsProcessor.cs

@@ -0,0 +1,10 @@
+namespace NHS.CohortManager.DemographicServices;
+
+using Model;
+
+public interface IPdsProcessor
+{
+    Task ProcessPdsNotFoundResponse(HttpResponseMessage pdsResponse, string nhsNumber);
+    Task ProcessRecord(Participant participant);
+    Task<bool> UpsertDemographicRecordFromPDS(ParticipantDemographic participantDemographic);
+}

application/CohortManager/src/Functions/DemographicServices/RetrievePDSDemographic/PdsProcessor.cs

@@ -0,0 +1,124 @@
+namespace NHS.CohortManager.DemographicServices;
+
+using System.Collections.Concurrent;
+using System.Net.Http.Json;
+using Common;
+using DataServices.Client;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Model;
+
+public class PdsProcessor : IPdsProcessor
+{
+    private readonly ILogger<PdsProcessor> _logger;
+
+    private readonly IDataServiceClient<ParticipantDemographic> _participantDemographicClient;
+    private readonly ICreateBasicParticipantData _createBasicParticipantData;
+    private readonly RetrievePDSDemographicConfig _config;
+    private readonly IAddBatchToQueue _addBatchToQueue;
+
+
+    public PdsProcessor(
+        ILogger<PdsProcessor> logger,
+        ICreateBasicParticipantData createBasicParticipantData,
+        IDataServiceClient<ParticipantDemographic> participantDemographicClient,
+        IAddBatchToQueue addBatchToQueue,
+        IOptions<RetrievePDSDemographicConfig> retrievePDSDemographicConfig)
+    {
+        _logger = logger;
+        _participantDemographicClient = participantDemographicClient;
+        _createBasicParticipantData = createBasicParticipantData;
+        _addBatchToQueue = addBatchToQueue;
+        _config = retrievePDSDemographicConfig.Value;
+    }
+
+    /// <summary>
+    /// processes pds error responses. Sends a record to distribute participant via service bus
+    /// </summary>
+    /// <param name="pdsResponse"></param>
+    /// <param name="nhsNumber"></param>
+    /// <returns></returns>
+    public async Task ProcessPdsNotFoundResponse(HttpResponseMessage pdsResponse, string nhsNumber)
+    {
+        var errorResponse = await pdsResponse!.Content.ReadFromJsonAsync<PdsErrorResponse>();
+        // we now create a record as an update record and send to the manage participant function. Reason for removal for date should be today and the reason for remove of ORR
+        if (errorResponse!.issue!.FirstOrDefault()!.details!.coding!.FirstOrDefault()!.code == PdsConstants.InvalidatedResourceCode)
+        {
+            var pdsDemographic = new PdsDemographic()
+            {
+                NhsNumber = nhsNumber,
+                PrimaryCareProvider = null,
+                ReasonForRemoval = PdsConstants.OrrRemovalReason,
+                RemovalEffectiveFromDate = DateTime.UtcNow.Date.ToString("yyyyMMdd")
+            };
+            var participant = new Participant(pdsDemographic);
+            participant.RecordType = Actions.Removed;
+            //sends record for an update
+            await ProcessRecord(participant);
+            return;
+        }
+        _logger.LogError("the PDS function has returned a 404 error. function now stopping processing");
+    }
+
+    /// <summary>
+    /// sends a participant record to the distribute service bus topic 
+    /// </summary>
+    /// <param name="participant"></param>
+    /// <returns></returns>
+    public async Task ProcessRecord(Participant participant)
+    {
+        var updateRecord = new ConcurrentQueue<BasicParticipantCsvRecord>();
+        participant.RecordType = Actions.Removed;
+
+        var basicParticipantCsvRecord = new BasicParticipantCsvRecord
+        {
+            BasicParticipantData = _createBasicParticipantData.BasicParticipantData(participant),
+            FileName = PdsConstants.DefaultFileName,
+            Participant = participant
+        };
+
+        updateRecord.Enqueue(basicParticipantCsvRecord);
+
+        _logger.LogInformation("Sending record to the update queue.");
+        await _addBatchToQueue.ProcessBatch(updateRecord, _config.ParticipantManagementTopic);
+    }
+
+    /// <summary>
+    /// adds or updates a demographic record depending on if an record already exists in the database
+    /// </summary>
+    /// <param name="participantDemographic"></param>
+    /// <returns></returns>
+    public async Task<bool> UpsertDemographicRecordFromPDS(ParticipantDemographic participantDemographic)
+    {
+        ParticipantDemographic oldParticipantDemographic = await _participantDemographicClient.GetSingleByFilter(i => i.NhsNumber == participantDemographic.NhsNumber);
+
+        if (oldParticipantDemographic == null)
+        {
+            _logger.LogInformation("Participant Demographic record not found, attemping to add Participant Demographic.");
+            bool addSuccess = await _participantDemographicClient.Add(participantDemographic);
+
+            if (addSuccess)
+            {
+                _logger.LogInformation("Successfully added Participant Demographic.");
+                return true;
+            }
+
+            _logger.LogError("Failed to add Participant Demographic.");
+            return false;
+        }
+
+        _logger.LogInformation("Participant Demographic record found, attempting to update Participant Demographic.");
+        participantDemographic.ParticipantId = oldParticipantDemographic.ParticipantId;
+        bool updateSuccess = await _participantDemographicClient.Update(participantDemographic);
+
+        if (updateSuccess)
+        {
+            _logger.LogInformation("Successfully updated Participant Demographic.");
+            return true;
+        }
+
+        _logger.LogError("Failed to update Participant Demographic.");
+        return false;
+    }
+
+}

application/CohortManager/src/Functions/DemographicServices/RetrievePDSDemographic/Program.cs

@@ -19,6 +19,7 @@
         services.AddSingleton<IHttpParserHelper, HttpParserHelper>();
         services.AddSingleton<IFhirPatientDemographicMapper, FhirPatientDemographicMapper>();
         services.AddSingleton<IAddBatchToQueue, AddBatchToQueue>();
+        services.AddSingleton<IPdsProcessor, PdsProcessor>();
         services.AddSingleton<ICreateBasicParticipantData, CreateBasicParticipantData>();
         // Register health checks
         services.AddBasicHealthCheck("RetrievePdsDemographic");