Improved batching of migration jobs (#1238)

* Handle duplicate Dialogporten migration jobs

* Batch the batching. It takes about a minute to create 10k jobs.

* Increase max cache entry size

* Error handling in sync

* Maybe actually scheduling it is a good idea too

* Ai

* Delete file

Author: Ceredron

Test/Altinn.Correspondence.Tests/TestingHandler/MigrateCorrespondenceHandlerTests.cs

@@ -122,7 +122,7 @@ static void ConfigureServices(IServiceCollection services, IConfiguration config
         options.InstanceName = "redisCache";
     });
 #pragma warning disable EXTEXP0018
-    services.AddHybridCache();
+    services.AddHybridCache(options => options.MaximumPayloadBytes = 1000 * 1000 * 10L);
 #pragma warning restore EXTEXP0018
     services.AddSingleton<IHybridCacheWrapper, HybridCacheWrapper>();
     services.ConfigureAuthentication(config, hostEnvironment);

src/Altinn.Correspondence.API/Program.cs

@@ -10,4 +10,5 @@ public class MakeCorrespondenceAvailableRequest
     public List<Guid>? CorrespondenceIds { get; set; }
     public bool AsyncProcessing { get; set; }
     public int? BatchSize { get; set; }
+    public int? BatchOffset { get; set; }
 }

src/Altinn.Correspondence.Application/MigrateCorrespondence/MakeCorrespondenceAvailableRequest.cs

@@ -111,17 +111,53 @@ public async Task<OneOf<MakeCorrespondenceAvailableResponse, Error>> MakeCorresp
         }
         else if (request.BatchSize is not null)
         {
-            var correspondences = await correspondenceRepository.GetCandidatesForMigrationToDialogporten(request.BatchSize ?? 0, cancellationToken);
-            foreach(var correspondence in correspondences)
+            if (!request.AsyncProcessing)
             {
-                if (request.AsyncProcessing)
+                var correspondences = await correspondenceRepository.GetCandidatesForMigrationToDialogporten(request.BatchSize ?? 0, request.BatchOffset ?? 0, cancellationToken);
+                foreach (var correspondence in correspondences)
                 {
-                    backgroundJobClient.Enqueue<MigrateCorrespondenceHandler>("migration", (handler) => handler.MakeCorrespondenceAvailableInDialogportenAndApi(correspondence.Id));
-                } 
-                else
+                    try
+                    {
+                        dialogId = await MakeCorrespondenceAvailableInDialogportenAndApi(correspondence.Id, cancellationToken);
+                        response.Statuses.Add(new(correspondence.Id, null, dialogId, true));
+                    }
+                    catch (Exception ex)
+                    {
+                        response.Statuses.Add(new(correspondence.Id, ex.ToString()));
+                    }
+                }
+                return response;
+            }
+            var batchLimit = 10000;
+            if (request.BatchSize > batchLimit)
+            {
+                var remainingCount = request.BatchSize;
+                var alreadyAdded = 0;
+                while(remainingCount > 0)
+                {
+                    int currentBatch = batchLimit;
+                    remainingCount -= batchLimit;
+                    if (remainingCount < 0)
+                    {
+                        currentBatch = (int)(remainingCount + batchLimit);
+                    }
+                    var migrateRequest = new MakeCorrespondenceAvailableRequest()
+                    {
+                        AsyncProcessing = true,
+                        BatchOffset = alreadyAdded,
+                        BatchSize = currentBatch,
+                        CreateEvents = request.CreateEvents
+                    };
+                    backgroundJobClient.Enqueue<MigrateCorrespondenceHandler>((handler) => handler.MakeCorrespondenceAvailable(migrateRequest, CancellationToken.None));
+                    alreadyAdded += currentBatch;
+                }
+            } 
+            else
+            {
+                var correspondences = await correspondenceRepository.GetCandidatesForMigrationToDialogporten(request.BatchSize ?? 0, request.BatchOffset ?? 0, cancellationToken);
+                foreach(var correspondence in correspondences)
                 {
-                    dialogId = await MakeCorrespondenceAvailableInDialogportenAndApi(correspondence.Id, cancellationToken);
-                    response.Statuses.Add(new(correspondence.Id, null, dialogId, true));
+                    backgroundJobClient.Enqueue<MigrateCorrespondenceHandler>(handler => handler.MakeCorrespondenceAvailableInDialogportenAndApi(correspondence.Id));
                 }
             }
         }

src/Altinn.Correspondence.Application/MigrateCorrespondence/MigrateCorrespondenceHandler.cs

@@ -54,7 +54,7 @@ Task<CorrespondenceEntity> GetCorrespondenceByAltinn2Id(
         Task UpdatePublished(Guid correspondenceId, DateTimeOffset published, CancellationToken cancellationToken);
         Task UpdateIsMigrating(Guid correspondenceId, bool isMigrating, CancellationToken cancellationToken);
         Task<bool> AreAllAttachmentsPublished(Guid correspondenceId, CancellationToken cancellationToken = default);
-        Task<List<CorrespondenceEntity>> GetCandidatesForMigrationToDialogporten(int batchSize, CancellationToken cancellationToken = default);
+        Task<List<CorrespondenceEntity>> GetCandidatesForMigrationToDialogporten(int batchSize, int offset, CancellationToken cancellationToken = default);
         Task<List<CorrespondenceEntity>> GetPurgedCorrespondencesWithDialogsAfter(
             int limit,
             DateTimeOffset? lastCreated,

src/Altinn.Correspondence.Core/Repositories/ICorrespondenceRepostitory.cs

@@ -5,6 +5,7 @@
 using Microsoft.Extensions.Options;
 using Altinn.Correspondence.Core.Repositories;
 using Altinn.Correspondence.Core.Models.Notifications;
+using System.Text.Json;
 
 namespace Altinn.Correspondence.Integrations.Altinn.Notifications;
 
@@ -42,6 +43,8 @@ public AltinnNotificationService(HttpClient httpClient, IOptions<AltinnOptions>
     {
         _logger.LogInformation("Creating notification in Altinn Notification v2");
         var response = await _httpClient.PostAsJsonAsync("notifications/api/v1/future/orders", notificationRequest, cancellationToken);
+        var jsonRequest = JsonSerializer.Serialize(notificationRequest);
+        var responseJson = await response.Content.ReadAsStringAsync(cancellationToken);
         if (!response.IsSuccessStatusCode)
         {
             _logger.LogError("Failed to create notification in Altinn Notification v2. Status code: {StatusCode}", response.StatusCode);

src/Altinn.Correspondence.Integrations/Altinn/Notifications/AltinnNotificationService.cs

@@ -435,6 +435,11 @@ public async Task<string> CreateCorrespondenceDialogForMigratedCorrespondence(Gu
         {
             throw new ArgumentException($"Correspondence with id {correspondenceId} not found", nameof(correspondenceId));
         }
+        if (correspondence.ExternalReferences.Any(reference => reference.ReferenceType == ReferenceType.DialogportenDialogId))
+        {
+            logger.LogWarning($"Duplicate job for correspondence {correspondenceId}");
+            return correspondence.ExternalReferences.FirstOrDefault(reference => reference.ReferenceType == ReferenceType.DialogportenDialogId)?.ReferenceValue ?? string.Empty;
+        }
 
         await CreateIdempotencyKeysForCorrespondence(correspondence, cancellationToken);
 

src/Altinn.Correspondence.Integrations/Dialogporten/DialogportenService.cs

@@ -185,11 +185,13 @@ public async Task<bool> AreAllAttachmentsPublished(Guid correspondenceId, Cancel
                 .SingleOrDefaultAsync(cancellationToken);
         }
 
-        public Task<List<CorrespondenceEntity>> GetCandidatesForMigrationToDialogporten(int batchSize, CancellationToken cancellationToken = default)
+        public Task<List<CorrespondenceEntity>> GetCandidatesForMigrationToDialogporten(int batchSize, int offset, CancellationToken cancellationToken = default)
         {
             return _context.Correspondences
                 .Where(c => c.Altinn2CorrespondenceId != null && c.IsMigrating) // Only include correspondences that are not already migrated 
                 .OrderByDescending(c => c.Created)
+                .ThenBy(c => c.Id)
+                .Skip(offset)
                 .Take(batchSize)
                 .ToListAsync(cancellationToken);
         }