Merge pull request #84 from DEFRA/feature/cleanse-report-optimisations

Added more mongo indexes

Author: krisddefra

src/KeeperData.Bridge.Worker/Configuration/ScheduledJobConfiguration.cs

@@ -3,6 +3,6 @@ namespace KeeperData.Bridge.Worker.Configuration;
 public class ScheduledJobConfiguration
 {
     public string JobType { get; set; } = string.Empty;
-    public bool Enabled { get; set; }
+    public bool IsEnabled { get; set; }
     public string CronSchedule { get; set; } = string.Empty;
 }

src/KeeperData.Bridge.Worker/Jobs/ImportDeltaFilesJob.cs

@@ -27,7 +27,7 @@ private static IServiceCollection AddQuartz(this IServiceCollection services, IC
         services.AddQuartz(q =>
         {
             var importBulkFilesConfig = scheduledJobConfiguration.FirstOrDefault(x => x.JobType == nameof(ImportBulkFilesJob));
-            if (importBulkFilesConfig?.CronSchedule != null)
+            if (!string.IsNullOrWhiteSpace(importBulkFilesConfig?.CronSchedule) && importBulkFilesConfig.IsEnabled)
             {
                 q.AddJob<ImportBulkFilesJob>(opts => opts.WithIdentity(importBulkFilesConfig.JobType));
 
@@ -37,19 +37,8 @@ private static IServiceCollection AddQuartz(this IServiceCollection services, IC
                     .WithCronSchedule(importBulkFilesConfig.CronSchedule));
             }
 
-            var importDeltaFilesConfig = scheduledJobConfiguration.FirstOrDefault(x => x.JobType == nameof(ImportDeltaFilesJob));
-            if (importDeltaFilesConfig?.CronSchedule != null)
-            {
-                q.AddJob<ImportDeltaFilesJob>(opts => opts.WithIdentity(importDeltaFilesConfig.JobType));
-
-                q.AddTrigger(opts => opts
-                    .ForJob(importDeltaFilesConfig.JobType)
-                    .WithIdentity($"{importDeltaFilesConfig.JobType}-trigger")
-                    .WithCronSchedule(importDeltaFilesConfig.CronSchedule));
-            }
-
             var cleanseReportConfig = scheduledJobConfiguration.FirstOrDefault(x => x.JobType == nameof(CleanseReportJob));
-            if (cleanseReportConfig?.CronSchedule != null)
+            if (!string.IsNullOrWhiteSpace(cleanseReportConfig?.CronSchedule) && cleanseReportConfig.IsEnabled)
             {
                 q.AddJob<CleanseReportJob>(opts => opts.WithIdentity(cleanseReportConfig.JobType));
 
@@ -71,7 +60,6 @@ private static IServiceCollection AddQuartz(this IServiceCollection services, IC
     private static IServiceCollection AddJobs(this IServiceCollection services)
     {
         services.AddScoped<ImportBulkFilesJob>();
-        services.AddScoped<ImportDeltaFilesJob>();
         services.AddScoped<CleanseReportJob>();
 
         return services;
@@ -80,7 +68,6 @@ private static IServiceCollection AddJobs(this IServiceCollection services)
     private static IServiceCollection AddTasks(this IServiceCollection services)
     {
         services.AddScoped<ITaskProcessBulkFiles, TaskProcessBulkFiles>();
-        services.AddScoped<ITaskProcessDeltaFiles, TaskProcessDeltaFiles>();
         services.AddScoped<ITaskRunCleanseReport, TaskRunCleanseReport>();
 
         return services;

src/KeeperData.Bridge.Worker/Setup/ServiceCollectionExtensions.cs

@@ -0,0 +1,26 @@
+using KeeperData.Bridge.Worker.Configuration;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+
+namespace KeeperData.Bridge.Setup;
+
+public class QuartzJobsHealthCheck(IConfiguration configuration) : IHealthCheck
+{
+    public Task<HealthCheckResult> CheckHealthAsync(
+        HealthCheckContext context,
+        CancellationToken cancellationToken = default)
+    {
+        var jobs = configuration.GetSection("Quartz:Jobs").Get<List<ScheduledJobConfiguration>>() ?? [];
+
+        var data = new Dictionary<string, object>();
+        foreach (var job in jobs)
+        {
+            data[job.JobType] = new
+            {
+                job.IsEnabled,
+                job.CronSchedule
+            };
+        }
+
+        return Task.FromResult(HealthCheckResult.Healthy("Quartz scheduled jobs", data));
+    }
+}

src/KeeperData.Bridge.Worker/Tasks/ITaskProcessDeltaFiles.cs

@@ -76,7 +76,8 @@ public static void ConfigureApi(this IServiceCollection services, IConfiguration
 
         private static void ConfigureHealthChecks(this IServiceCollection services)
         {
-            services.AddHealthChecks();
+            services.AddHealthChecks()
+                .AddCheck<QuartzJobsHealthCheck>("quartz_jobs", tags: ["quartz", "jobs"]);
             services.AddSingleton<IHealthCheckPublisher, HealthCheckMetricsPublisher>();
         }
 

src/KeeperData.Bridge.Worker/Tasks/Implementations/TaskProcessDeltaFiles.cs

@@ -73,17 +73,12 @@
     "Jobs": [
       {
         "JobType": "ImportBulkFilesJob",
-        "Enabled": true,
+        "IsEnabled": true,
         "CronSchedule": "0 0 8 * * ?"
       },
-      {
-        "JobType": "ImportDeltaFilesJob",
-        "Enabled": true,
-        "CronSchedule": "0 0 9 * * ?"
-      },
       {
         "JobType": "CleanseReportJob",
-        "Enabled": true,
+        "IsEnabled": false,
         "CronSchedule": "0 0 2 * * ?"
       }
     ]

src/KeeperData.Bridge/Setup/QuartzJobsHealthCheck.cs

@@ -1,20 +1,28 @@
+using System.Diagnostics;
 using KeeperData.Core.Reports.Cleanse.Analysis.Command.Domain;
 using KeeperData.Core.Reports.Domain;
 using KeeperData.Core.Reports.Issues.Command.Abstract;
 using KeeperData.Core.Reports.SamCtsHoldings.Query.Abstract;
 using KeeperData.Core.Reports.SamCtsHoldings.Query.Domain;
+using Microsoft.Extensions.Logging;
 
 namespace KeeperData.Core.Reports.Cleanse.Analysis.Command.Abstract;
 
-public abstract class CleanseAnalysisEngineBase(ICtsSamQueryService dataService, IIssueCommandService issueCommandService)
+public abstract class CleanseAnalysisEngineBase(ICtsSamQueryService dataService, IIssueCommandService issueCommandService, ILogger logger)
 {
-    private const int ThrottlingPumpDelayMs = 400;
-    private const int BatchSize = 70;
+    private const int ThrottlingPumpDelayMs = 150;
+    private const int BatchSize = 100;
     private const int ProgressUpdateInterval = 100;
     private const string DateTimeFormat = "yyyy-MM-dd HH:mm:ss";
 
     protected IIssueCommandService IssueCommandService { get; } = issueCommandService;
 
+    /// <summary>
+    /// In-memory lookup mapping CPH values to their full LID_FULL_IDENTIFIER strings.
+    /// Populated before the SAM pump runs so that CPH→LID resolution avoids regex queries on MongoDB.
+    /// </summary>
+    protected Dictionary<string, string> CphToLidLookup { get; private set; } = [];
+
     protected delegate Task RecordProcessor(string id,
         string operationId, AnalysisMetrics metrics, CancellationToken ct);
 
@@ -77,6 +85,10 @@ await PumpAsync(new PumpContext(totalRecords, operationId, metrics, progressCall
             dataService.ListCtsCphHoldingsAsync, ProcessCtsPrimaryRecordAsync,
             DataFields.CtsCphHoldingFields.LidFullIdentifier), ct);
 
+        // Pre-load all CTS LID_FULL_IDENTIFIER values so CPH→LID resolution
+        // can be done in-memory instead of via regex queries on MongoDB.
+        CphToLidLookup = await BuildCphToLidLookupAsync(ct);
+
         // iterate SAM CPH records
         await PumpAsync(new PumpContext(totalRecords, operationId, metrics, progressCallback,
             dataService.ListSamCphHoldingsAsync, ProcessSamPrimaryRecordAsync,
@@ -117,6 +129,47 @@ protected static bool IsCtsCphHoldingRecordActive(IDictionary<string, object?> r
         };
 
 
+    /// <summary>
+    /// Pages through all CTS CPH Holding records in throttled batches,
+    /// building a dictionary that maps each CPH value to its full LID_FULL_IDENTIFIER.
+    /// </summary>
+    private async Task<Dictionary<string, string>> BuildCphToLidLookupAsync(CancellationToken ct)
+    {
+        logger.LogInformation("Building CPH to LID lookup: starting");
+        var stopwatch = Stopwatch.StartNew();
+
+        var lookup = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
+        var skip = 0;
+
+        while (!ct.IsCancellationRequested)
+        {
+            var batch = await dataService.ListCtsCphHoldingsAsync(skip, BatchSize, ct);
+
+            if (batch.Data.Count == 0)
+            {
+                break;
+            }
+
+            foreach (var record in batch.Data)
+            {
+                var lid = LidFullIdentifier.TryParse(record[DataFields.CtsCphHoldingFields.LidFullIdentifier]?.ToString());
+                if (lid is not null)
+                {
+                    lookup.TryAdd(lid.Cph.Value, lid.Value);
+                }
+            }
+
+            skip += batch.Data.Count;
+            await Task.Delay(ThrottlingPumpDelayMs, ct);
+        }
+
+        stopwatch.Stop();
+        logger.LogInformation("Building CPH to LID lookup: completed. Records={RecordCount}, Duration={DurationMs}ms ({DurationSeconds}s)",
+            lookup.Count, stopwatch.ElapsedMilliseconds, stopwatch.Elapsed.TotalSeconds);
+
+        return lookup;
+    }
+
     protected static bool ShouldUpdateProgress(int recordsAnalyzed)
         => recordsAnalyzed % ProgressUpdateInterval == 0;
 }

src/KeeperData.Bridge/Setup/ServiceCollectionExtensions.cs

@@ -6,16 +6,17 @@
 using KeeperData.Core.Reports.Issues.Command.Abstract;
 using KeeperData.Core.Reports.SamCtsHoldings.Query.Abstract;
 using KeeperData.Core.Reports.SamCtsHoldings.Query.Domain;
+using Microsoft.Extensions.Logging;
 
 namespace KeeperData.Core.Reports.Cleanse.Analysis.Command.Impl;
 
-public class CleanseAnalysisEngine(ICtsSamQueryService dataService, IIssueCommandService issueCommandService) 
-    : CleanseAnalysisEngineBase(dataService, issueCommandService), ICleanseAnalysisEngine
+public class CleanseAnalysisEngine(ICtsSamQueryService dataService, IIssueCommandService issueCommandService, ILogger<CleanseAnalysisEngine> logger) 
+    : CleanseAnalysisEngineBase(dataService, issueCommandService, logger), ICleanseAnalysisEngine
 {
     private readonly RecordIdGenerator _recordIdGenerator = new();
     private readonly ICtsSamQueryService _dataService = dataService;
 
-    private const int ThrottleDelayMs = 100;
+    private const int ThrottleDelayMs = 50;
 
     private async Task ProcessCtsPrimaryRecordInternalAsync(LidFullIdentifier lidFullIdentifier, string operationId, AnalysisMetrics metrics, CancellationToken ct)
     {
@@ -146,8 +147,15 @@ private async Task ProcessSamPrimaryRecordInternalAsync(Cph cph, string operatio
     {
         var results = new List<RuleResult>();
 
-        // get the CTS CPH Holding...
-        var ctsCphHolding = await _dataService.GetCtsCphHoldingAsync(cph, ct);
+        // Resolve CPH to full LID_FULL_IDENTIFIER via the in-memory lookup
+        // so we can query by equals instead of a regex endswith on MongoDB.
+        CtsCphHoldingModel? ctsCphHolding = null;
+        if (CphToLidLookup.TryGetValue(cph.Value, out var lidValue))
+        {
+            var lid = LidFullIdentifier.Parse(lidValue);
+            ctsCphHolding = await _dataService.GetCtsCphHoldingAsync(lid, ct);
+        }
+
         if (ctsCphHolding is null) // does not exist
         {
             results.Add(RuleResult.Issue(RuleDescriptors.SamCphNotInCts, cph)); // PRIORITY 1B: RULE 2B - CPH present in SAM but missing in CTS