feat: Participant Audit Log Table

Directory.Packages.props

@@ -20,6 +20,7 @@
     <PackageVersion Include="Azure.Security.KeyVault.Secrets" Version="4.6.0" />
     <PackageVersion Include="Azure.Storage.Blobs" Version="12.20.0" />
     <PackageVersion Include="Azure.Storage.Queues" Version="12.20.1" />
+    <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Storage.Queues" Version="5.5.0" />
     <PackageVersion Include="Azure.Messaging.EventGrid" Version="4.28.0" />
     <PackageVersion Include="Microsoft.ApplicationInsights.DependencyCollector" Version="2.23.0" />
     <PackageVersion Include="Microsoft.ApplicationInsights.WorkerService" Version="2.23.0" />

application/CohortManager/Set-up/service-bus/config.json

@@ -3,6 +3,21 @@
     "Namespaces": [
       {
         "Name": "sbemulatorns",
+        "Queues": [
+          {
+            "Name": "participant-audit-queue",
+            "Properties": {
+              "DeadLetteringOnMessageExpiration": false,
+              "DefaultMessageTimeToLive": "PT1H",
+              "DuplicateDetectionHistoryTimeWindow": "PT20S",
+              "ForwardDeadLetteredMessagesTo": "",
+              "LockDuration": "PT1M",
+              "MaxDeliveryCount": 10,
+              "RequiresDuplicateDetection": false,
+              "RequiresSession": false
+            }
+          }
+        ],
           "Topics": [
             {
               "Name": "create-exception-topic",

application/CohortManager/compose.core.yaml

@@ -183,6 +183,22 @@ services:
       - CohortDistributionDataServiceURL=http://cohort-distribution-data-service:7992/api/CohortDistributionDataService/
       - AcceptableLatencyThresholdMs=500
 
+  # Audit Services
+  audit-writer:
+    container_name: audit-writer
+    image: cohort-manager-audit-writer
+    networks: [cohman-network]
+    build:
+      context: ./src/Functions/
+      dockerfile: AuditServices/AuditWriter/Dockerfile
+      args:
+        BASE_IMAGE: ${FUNCTION_BASE_IMAGE}
+    environment:
+      - AzureWebJobsStorage=${AZURITE_CONNECTION_STRING}
+      - DtOsDatabaseConnectionString=Server=db,1433;Database=${DB_NAME};User Id=SA;Password=${PASSWORD};TrustServerCertificate=True
+      - ServiceBusConnectionString=Endpoint=sb://service-bus;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;
+      - AuditQueueName=participant-audit-queue
+
 
   # Screening Data Service
   create-exception:

application/CohortManager/src/Functions/AuditServices/AuditWriter/AuditWriter.cs

@@ -0,0 +1,107 @@
+namespace NHS.CohortManager.AuditServices;
+
+using System.Text.Json;
+using DataServices.Database;
+using Microsoft.Azure.Functions.Worker;
+using Microsoft.Extensions.Logging;
+using Model;
+using Common;
+
+public class AuditWriter
+{
+    private const string AuditBlobContainer = "participant-audit";
+    private readonly DataServicesContext _dbContext;
+    private readonly ILogger<AuditWriter> _logger;
+    private readonly IBlobStorageHelper _blobStorageHelper;
+    private readonly AuditWriterConfig _config;
+
+    private static readonly JsonSerializerOptions JsonOptions = new()
+    {
+        PropertyNameCaseInsensitive = true
+    };
+
+    public AuditWriter(DataServicesContext dbContext, ILogger<AuditWriter> logger, IBlobStorageHelper blobStorageHelper, AuditWriterConfig config)
+    {
+        _dbContext = dbContext;
+        _logger = logger;
+        _blobStorageHelper = blobStorageHelper;
+        _config = config;
+    }
+
+    [Function(nameof(AuditWriter))]
+    public async Task Run(
+        [ServiceBusTrigger("%AuditQueueName%", Connection = "ServiceBusConnectionString")] string messageText, FunctionContext context)
+    {
+        ParticipantAuditMessage? audit;
+        try
+        {
+            audit = JsonSerializer.Deserialize<ParticipantAuditMessage>(messageText, JsonOptions);
+        }
+        catch (JsonException ex)
+        {
+            throw new InvalidOperationException("Failed to deserialise audit message.", ex);
+        }
+
+        if (audit is null)
+        {
+            _logger.LogError("Failed to deserialise audit message");
+            throw new JsonException("Audit message deserialised to null.");
+        }
+
+        audit.RawDataRef = await WriteSnapshotToBlobAsync(audit);
+
+        var auditLog = new ParticipantAuditLog
+        {
+            CorrelationId = audit.CorrelationId,
+            NhsNumber = audit.NhsNumber,
+            BatchId = audit.BatchId,
+            CreatedDatetime = audit.CreatedDatetime,
+            RecordSource = (int)audit.Source,
+            RecordSourceDesc = audit.RecordSourceDesc,
+            CreatedBy = audit.CreatedBy,
+            ScreeningId = audit.ScreeningId,
+            RawDataRef = audit.RawDataRef
+        };
+
+        _dbContext.Set<ParticipantAuditLog>().Add(auditLog);
+        await _dbContext.SaveChangesAsync();
+
+        _logger.LogInformation("Audit written | Source: {Source} | Correlation: {CorrelationId}",
+            audit.Source, audit.CorrelationId);
+    }
+
+    private async Task<string?> WriteSnapshotToBlobAsync(ParticipantAuditMessage message)
+    {
+        try
+        {
+            var blobPath = $"{message.CreatedDatetime:dd-MM-yyyy}/{message.CorrelationId}.json";
+
+            var payload = message.RequestSnapshot is not null
+                ? JsonSerializer.SerializeToUtf8Bytes(message.RequestSnapshot)
+                : JsonSerializer.SerializeToUtf8Bytes(message);
+
+            var blobFile = new BlobFile(payload, blobPath);
+            var uri = await _blobStorageHelper.UploadFileToBlobStorageAndGetUri(
+                _config.AzureWebJobsStorage, AuditBlobContainer, blobFile, overwrite: true);
+
+            if (uri is null)
+            {
+                _logger.LogError(
+                    "Failed to write audit snapshot to blob for CorrelationId {CorrelationId}. " +
+                    "Audit will be persisted without a blob reference.",
+                    message.CorrelationId);
+                return null;
+            }
+
+            return uri;
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex,
+                "Failed to write audit snapshot to blob for CorrelationId {CorrelationId}. " +
+                "Audit will be persisted without a blob reference.",
+                message.CorrelationId);
+            return null;
+        }
+    }
+}

application/CohortManager/src/Functions/AuditServices/AuditWriter/AuditWriter.csproj

@@ -0,0 +1,36 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <ProjectGuid>{A1B2C3D4-E5F6-7890-ABCD-EF1234567890}</ProjectGuid>
+    <TargetFramework>net8.0</TargetFramework>
+    <AzureFunctionsVersion>v4</AzureFunctionsVersion>
+    <OutputType>Exe</OutputType>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Azure.Functions.Worker" />
+    <PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.ServiceBus" />
+    <PackageReference Include="Microsoft.Azure.Functions.Worker.Sdk" />
+    <PackageReference Include="Microsoft.Extensions.Diagnostics.HealthChecks" />
+    <PackageReference Include="Azure.Storage.Blobs" />
+    <PackageReference Include="Azure.Messaging.ServiceBus" />
+  </ItemGroup>
+  <ItemGroup>
+    <None Update="host.json">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </None>
+    <None Update="local.settings.json">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+      <CopyToPublishDirectory>Never</CopyToPublishDirectory>
+    </None>
+  </ItemGroup>
+  <ItemGroup>
+    <Using Include="System.Threading.ExecutionContext" Alias="ExecutionContext" />
+  </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\..\Shared\DataServices.Database\DataServices.Database.csproj" />
+    <ProjectReference Include="..\..\Shared\DataServices.Core\DataServices.Core.csproj" />
+    <ProjectReference Include="..\..\Shared\Common\Common.csproj" />
+    <ProjectReference Include="..\..\Shared\HealthChecks\HealthChecks.csproj" />
+  </ItemGroup>
+</Project>

application/CohortManager/src/Functions/AuditServices/AuditWriter/AuditWriterConfig.cs

@@ -0,0 +1,11 @@
+namespace NHS.CohortManager.AuditServices;
+
+using System.ComponentModel.DataAnnotations;
+
+public class AuditWriterConfig
+{
+    [Required]
+    public required string ServiceBusConnectionString { get; set; }
+    [Required]
+    public required string AzureWebJobsStorage { get; set; }
+}

application/CohortManager/src/Functions/AuditServices/AuditWriter/Dockerfile

@@ -0,0 +1,16 @@
+ARG BASE_IMAGE
+FROM $BASE_IMAGE AS function
+
+COPY ./AuditServices/AuditWriter /app/src/dotnet-function-app
+WORKDIR /app/src/dotnet-function-app
+
+RUN --mount=type=cache,target=/root/.nuget/packages \
+    dotnet publish ./*.csproj --output /home/site/wwwroot
+
+# To enable ssh & remote debugging on app service change the base image to the one below
+# FROM mcr.microsoft.com/azure-functions/dotnet-isolated:4-dotnet-isolated8.0-appservice
+FROM mcr.microsoft.com/azure-functions/dotnet-isolated:4-dotnet-isolated8.0
+ENV AzureWebJobsScriptRoot=/home/site/wwwroot \
+    AzureFunctionsJobHost__Logging__Console__IsEnabled=true
+
+COPY --from=function ["/home/site/wwwroot", "/home/site/wwwroot"]

application/CohortManager/src/Functions/AuditServices/AuditWriter/Program.cs

@@ -0,0 +1,23 @@
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.DependencyInjection;
+using DataServices.Core;
+using DataServices.Database;
+using HealthChecks.Extensions;
+using Common;
+using NHS.CohortManager.AuditServices;
+
+var host = new HostBuilder()
+    .ConfigureFunctionsWorkerDefaults()
+    .AddConfiguration<AuditWriterConfig>(out AuditWriterConfig config)
+    .AddDataServicesHandler<DataServicesContext>()
+    .AddServiceBusClient(config.ServiceBusConnectionString)
+    .ConfigureServices(services =>
+    {
+        services.AddDatabaseHealthCheck("AuditWriter");
+        services.AddSingleton(config);
+        services.AddTransient<IBlobStorageHelper, BlobStorageHelper>();
+    })
+    .AddTelemetry()
+    .Build();
+
+await host.RunAsync();

application/CohortManager/src/Functions/CaasIntegration/receiveCaasFile/ProcessFileClasses/CallDurableDemographicFunc.cs

@@ -63,8 +63,13 @@ public async Task<bool> PostDemographicDataAsync(List<ParticipantDemographic> pa
         {
             var content = JsonSerializer.Serialize(participants);
             var response = await _httpClientFunction.SendPost(DemographicFunctionURI, content);
+            if (!response.IsSuccessStatusCode || response.Headers.Location is null)
+            {
+                _logger.LogError("DurableDemographicFunction returned {StatusCode} with no Location header", response.StatusCode);
+                return false;
+            }
+            responseContent = response.Headers.Location.ToString();
 
-            responseContent = response.Headers.Location!.ToString();
             // This is not retrying the function if it fails but checking if it has done yet.
             var retryPolicy = Policy
                 .HandleResult<WorkFlowStatus>(status => status != WorkFlowStatus.Completed && status != WorkFlowStatus.Failed)

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

@@ -1,4 +1,3 @@
-using Microsoft.Azure.Functions.Worker;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.DependencyInjection;
 using Common;
@@ -9,7 +8,6 @@
 using Model;
 using DataServices.Client;
 using HealthChecks.Extensions;
-using Microsoft.Extensions.Options;
 
 
 var loggerFactory = LoggerFactory.Create(builder => builder.AddConsole());

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

@@ -3,6 +3,7 @@ namespace NHS.Screening.ReceiveCaasFile;
 using Microsoft.Azure.Functions.Worker;
 using Microsoft.Extensions.Logging;
 using Model;
+using Model.Enums;
 using System;
 using System.IO;
 using ParquetSharp.RowOriented;
@@ -19,14 +20,16 @@ public class ReceiveCaasFile
     private readonly ReceiveCaasFileConfig _config;
     private readonly IBlobStorageHelper _blobStorageHelper;
     private readonly IExceptionHandler _exceptionHandler;
+    private readonly IQueueClient _queueClient;
 
     public ReceiveCaasFile(
         ILogger<ReceiveCaasFile> logger,
         IProcessCaasFile processCaasFile,
         IDataServiceClient<ScreeningLkp> screeningLkpClient,
         IOptions<ReceiveCaasFileConfig> receiveCaasFileConfig,
         IBlobStorageHelper blobStorageHelper,
-        IExceptionHandler exceptionHandler
+        IExceptionHandler exceptionHandler,
+        IQueueClient queueClient
         )
     {
         _logger = logger;
@@ -35,6 +38,7 @@ IExceptionHandler exceptionHandler
         _config = receiveCaasFileConfig.Value;
         _blobStorageHelper = blobStorageHelper;
         _exceptionHandler = exceptionHandler;
+        _queueClient = queueClient;
     }
 
     [Function(nameof(ReceiveCaasFile))]
@@ -56,15 +60,17 @@ public async Task Run([BlobTrigger("inbound/{name}", Connection = "caasfolder_ST
             downloadFilePath = Path.Combine(Path.GetTempPath(), name);
 
             _logger.LogInformation("Downloading file from the blob, file: {Name}.", name);
-            
-            // In order to use the parquet file we need to download it 
+
+            // In order to use the parquet file we need to download it
             await using (var fileStream = File.Create(downloadFilePath))
             {
                 await blobStream.CopyToAsync(fileStream);
             }
 
             var options = new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount };
 
+            var batchId = Guid.NewGuid();
+
             using (var rowReader = ParquetFile.CreateRowReader<ParticipantsParquetMap>(downloadFilePath))
             {
                 // A Parquet file is divided into one or more row groups. Each row group contains a specific number of rows.
@@ -74,6 +80,7 @@ public async Task Run([BlobTrigger("inbound/{name}", Connection = "caasfolder_ST
                     var listOfAllValues = values.ToList();
                     var allTasks = new List<Task>();
 
+                    await EnqueueAuditMessagesAsync(listOfAllValues, name, batchId, (int)screeningService.ScreeningId);
                     //split list of all into N amount of chunks to be processed as batches.
                     var chunks = listOfAllValues.Chunk(BatchSize).ToList();
 
@@ -125,4 +132,29 @@ public async Task<ScreeningLkp> GetScreeningService(FileNameParser fileNameParse
 
         return screeningService;
     }
+
+    private async Task EnqueueAuditMessagesAsync(
+        List<ParticipantsParquetMap> participants,
+        string fileName,
+        Guid batchId,
+        int screeningId)
+    {
+        var auditMessages = participants
+            .Where(w => w.NhsNumber.HasValue)
+            .Select(participant => new ParticipantAuditMessage
+            {
+                NhsNumber = participant.NhsNumber!.Value.ToString(),
+                Source = AuditSource.ParquetFile,
+                BatchId = batchId,
+                RecordSourceDesc = $"Parquet file: {fileName}",
+                CreatedBy = nameof(ReceiveCaasFile),
+                ScreeningId = screeningId,
+            });
+
+        var failCount = await _queueClient.AddBatchAsync(auditMessages, _config.AuditQueueName);
+        if (failCount != 0)
+        {
+            _logger.LogWarning("Audit enqueue failed for {FailCount} participants in batch {BatchId}", failCount, batchId);
+        }
+    }
 }

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

@@ -25,4 +25,6 @@ public class ReceiveCaasFileConfig
     public required string GetOrchestrationStatusURL { get; set; }
     [Required]
     public required string ParticipantManagementTopic { get; set; }
+    [Required]
+    public required string AuditQueueName { get; set; }
 }