CDMS-238 skeleton code to redact and replicate data to SND data lake

Btms.Analytics/MovementExceptions.cs

@@ -20,7 +20,8 @@ private static string FormatUnmatched(DecisionStatusEnum decisionStatus, Movemen
 
         return matched.ToString();
     }
-    //Returns a summary of the exceptions or a list
+
+    // Returns a summary of the exceptions or a list
     // Means we can share the same anonymous / query code without needing to create loads
     // of classes
     public (SingleSeriesDataset summary, List<ExceptionResult>) GetAllExceptions(DateTime from, DateTime to, bool finalisedOnly, bool summary, ImportNotificationTypeEnum[]? chedTypes = null, string? country = null)

Btms.Backend/Endpoints/SyncEndpoints.cs

@@ -38,7 +38,7 @@ public static void UseSyncEndpoints(this IEndpointRouteBuilder app, IOptions<Api
         app.MapPost(BaseRoute + "/generate-download", GenerateDownload).AllowAnonymous();
         app.MapGet(BaseRoute + "/download/{id}", DownloadNotifications).AllowAnonymous();
 
-
+        app.MapGet(BaseRoute + "/replicate", ReplicateGet).AllowAnonymous();
         app.MapGet(BaseRoute + "/queue-counts/", GetQueueCounts).AllowAnonymous();
         app.MapGet(BaseRoute + "/jobs/", GetAllSyncJobs).AllowAnonymous();
         app.MapGet(BaseRoute + "/jobs/clear", ClearSyncJobs).AllowAnonymous();
@@ -72,6 +72,15 @@ private static async Task<IResult> GenerateDownload([FromServices] IBtmsMediator
         return Results.Ok(command.JobId);
     }
 
+    private static async Task<IResult> ReplicateGet(
+        [FromServices] IBtmsMediator mediator,
+        SyncPeriod syncPeriod)
+    {
+        ReplicateCommand command = new() { SyncPeriod = syncPeriod };
+        await mediator.SendJob(command);
+        return Results.Ok(command.JobId);
+    }
+
     private static Task<IResult> GetAllSyncJobs([FromServices] ISyncJobStore store)
     {
         return Task.FromResult(Results.Ok(new { jobs = store.GetJobs() }));

Btms.Business/BusinessOptions.cs

@@ -1,16 +1,28 @@
 using System.ComponentModel.DataAnnotations;
+using Btms.Azure;
 using Btms.Business.Commands;
 
 namespace Btms.Business;
 
+public class ReplicationOptions : IAzureConfig
+{
+    public const string SectionName = nameof(ReplicationOptions);
+
+    public string? AzureClientId { get; set; }
+    public string? AzureTenantId { get; set; }
+    public string? AzureClientSecret { get; set; }
+
+    public bool Enabled { get; set; } = false;
+
+}
 public class BusinessOptions
 {
     public const string SectionName = nameof(BusinessOptions);
 
     private readonly int defaultDegreeOfParallelism = Math.Max(Environment.ProcessorCount / 4, 1);
 
     [Required] public string DmpBlobRootFolder { get; set; } = "RAW";
-
+    
     public Dictionary<string, Dictionary<Feature, int>> ConcurrencyConfiguration { get; set; }
 
     public enum Feature

Btms.Business/Commands/DownloadCommand.cs

@@ -26,6 +26,18 @@ public class DownloadCommand : IRequest, ISyncJob
 
     public string? RootFolder { get; set; }
 
+    public static List<(string path, Type dataType)> BlobFolders = new List<(string path, Type dataType)>()
+    {
+        ("IPAFFS/CHEDA", typeof(ImportNotification)),
+        ("IPAFFS/CHEDD", typeof(ImportNotification)),
+        ("IPAFFS/CHEDP", typeof(ImportNotification)),
+        ("IPAFFS/CHEDPP", typeof(ImportNotification)),
+        ("IPAFFS/ALVS", typeof(AlvsClearanceRequest)),
+        ("IPAFFS/GVMSAPIRESPONSE", typeof(SearchGmrsForDeclarationIdsResponse)),
+        ("IPAFFS/DECISIONS", typeof(AlvsClearanceRequest)),
+        ("IPAFFS/FINALISATION", typeof(Finalisation))
+    };
+
     internal class Handler(IBlobService blobService, ISensitiveDataSerializer sensitiveDataSerializer, IHostEnvironment env, IOptions<BusinessOptions> businessOptions) : IRequestHandler<DownloadCommand>
     {
 
@@ -55,18 +67,10 @@ public async Task Handle(DownloadCommand request, CancellationToken cancellation
             }
             else
             {
-                await Download(request, rootFolder, $"{blobContainer}/IPAFFS/CHEDA", typeof(ImportNotification), null, cancellationToken);
-                await Download(request, rootFolder, $"{blobContainer}/IPAFFS/CHEDD", typeof(ImportNotification), null, cancellationToken);
-                await Download(request, rootFolder, $"{blobContainer}/IPAFFS/CHEDP", typeof(ImportNotification), null, cancellationToken);
-                await Download(request, rootFolder, $"{blobContainer}/IPAFFS/CHEDPP", typeof(ImportNotification), null, cancellationToken);
-
-                await Download(request, rootFolder, $"{blobContainer}/ALVS", typeof(AlvsClearanceRequest), null, cancellationToken);
-
-                await Download(request, rootFolder, $"{blobContainer}/GVMSAPIRESPONSE", typeof(SearchGmrsForDeclarationIdsResponse), null, cancellationToken);
-
-                await Download(request, rootFolder, $"{blobContainer}/DECISIONS", typeof(AlvsClearanceRequest), null, cancellationToken);
-
-                await Download(request, rootFolder, $"{blobContainer}/FINALISATION", typeof(Finalisation), null, cancellationToken);
+                BlobFolders.ForEach(async f =>
+                {
+                    await Download(request, rootFolder, $"{blobContainer}/{f.path}", f.dataType, null, cancellationToken);
+                });
             }
 
 

Btms.Business/Commands/ReplicateCommand.cs

@@ -0,0 +1,54 @@
+using System.IO.Compression;
+using System.Text.Json.Serialization;
+using Btms.BlobService;
+using Btms.Common.Extensions;
+using MediatR;
+using Btms.SensitiveData;
+using Btms.Types.Ipaffs;
+using Btms.SyncJob;
+using Btms.Types.Alvs;
+using Btms.Types.Gvms;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+namespace Btms.Business.Commands;
+
+public class ReplicateCommand() : IRequest, ISyncJob
+{
+    [JsonConverter(typeof(JsonStringEnumConverter<SyncPeriod>))]
+    public SyncPeriod SyncPeriod { get; set; }
+
+    public Guid JobId { get; } = Guid.NewGuid();
+    public string Timespan { get; } = null!;
+    public string Resource { get; } = null!;
+
+    public string? RootFolder { get; set; }
+    // IBlobService blobService, ISensitiveDataSerializer sensitiveDataSerializer, IHostEnvironment env, 
+    internal class Handler(ILogger<ReplicateCommand> logger, IOptions<BusinessOptions> businessOptions, IOptions<ReplicationOptions> replicationOptions) : IRequestHandler<ReplicateCommand>
+    {
+        public async Task Handle(ReplicateCommand request, CancellationToken cancellationToken)
+        {
+            var blobContainer = string.IsNullOrEmpty(request.RootFolder)
+                ? businessOptions.Value.DmpBlobRootFolder
+                : request.RootFolder;
+
+            if (replicationOptions.Value.Enabled)
+            {
+                logger.LogInformation("Replicating from {BlobContainer} to {DestinationContainer}", blobContainer, blobContainer);
+
+                DownloadCommand.BlobFolders.ForEach(f =>
+                {
+                    logger.LogInformation("Replicate {Path}", f.path);
+                });
+            }
+            else
+            {
+                logger.LogWarning("ReplicateCommand called but not replication not enabled in config.");
+            }
+
+            await Task.CompletedTask;
+        }
+
+    }
+}

Btms.Business/Commands/SyncHandler.cs

@@ -15,6 +15,7 @@ namespace Btms.Business.Commands;
 public enum SyncPeriod
 {
     Today,
+    Yesterday,
     LastMonth,
     ThisMonth,
     From202411,

Btms.Business/Commands/SyncPeriodExtensions.cs

@@ -20,6 +20,10 @@ public static string[] GetPeriodPaths(this SyncPeriod period)
         {
             return [DateTime.Today.ToString("/yyyy/MM/dd/")];
         }
+        else if (period == SyncPeriod.Yesterday)
+        {
+            return [DateTime.Today.AddDays(-1).ToString("/yyyy/MM/dd/")];
+        }
         else if (period == SyncPeriod.From202411)
         {
             return DateTime.Today

Btms.Business/Extensions/ServiceCollectionExtensions.cs

@@ -30,7 +30,8 @@ public static IServiceCollection AddBusinessServices(this IServiceCollection ser
         services.AddBtmsMetrics();
         services.BtmsAddOptions<SensitiveDataOptions>(configuration, SensitiveDataOptions.SectionName);
         services.BtmsAddOptions<BusinessOptions>(configuration, BusinessOptions.SectionName);
-
+        services.BtmsAddOptions<ReplicationOptions>(configuration, ReplicationOptions.SectionName);
+
         services.AddMongoDbContext(configuration);
         services.AddBlobStorage(configuration);
         services.AddSingleton<IBlobServiceClientFactory, BlobServiceClientFactory>();