fix: or-2893 ensure bodies rebuild on failure

Author: koenmetsu

src/OrganisationRegistry.ElasticSearch.Projections/BaseRunner.cs

@@ -8,6 +8,8 @@ namespace OrganisationRegistry.ElasticSearch.Projections;
 using System.Threading.Tasks;
 using Client;
 using Configuration;
+using ElasticSearch.Bodies;
+using ElasticSearch.Organisations;
 using Infrastructure;
 using Infrastructure.Change;
 using Microsoft.Extensions.Logging;
@@ -128,21 +130,22 @@ public async Task Run()
             _metrics.NumberOfEnvelopesHandledGauge = envelopes.Count;
             _metrics.NumberOfEnvelopesHandledCounter = envelopes.Count;
         }
-        catch (ElasticsearchOrganisationNotFoundException organisationNotFoundException)
+        catch (ElasticsearchAggregateNotFoundException organisationNotFoundException)
         {
             await using var organisationRegistryContext = _contextFactory.Create();
+
             organisationRegistryContext.OrganisationsToRebuild.Add(
                 new OrganisationToRebuild
                 {
-                    OrganisationId = Guid.Parse(organisationNotFoundException.OrganisationId)
+                    OrganisationId = Guid.Parse(organisationNotFoundException.AggregateId)
                 });
             await organisationRegistryContext.SaveChangesAsync();
             _logger.LogWarning(
                 0,
                 organisationNotFoundException,
                 "[{ProjectionName}] Could not find {OrganisationId} in ES while processing envelope #{EnvelopeNumber}, adding it to organisations to rebuild",
                 ProjectionName,
-                organisationNotFoundException.OrganisationId,
+                organisationNotFoundException.AggregateId,
                 newLastProcessedEventNumber);
             throw;
         }
@@ -168,26 +171,7 @@ private async Task ProcessChange(IElasticChange? changeSetChange, Dictionary<Gui
             }
             case ElasticPerDocumentChange<T> perDocumentChange:
             {
-                foreach (var documentChange in perDocumentChange.Changes)
-                {
-                    T? document;
-
-                    if (!documentCache.ContainsKey(documentChange.Key))
-                    {
-                        document = (await _elastic.TryGetAsync(async () =>
-                                (await _elastic.WriteClient.GetAsync<T>(documentChange.Key))
-                                .ThrowOnFailure()))
-                            .Source;
-
-                        documentCache.Add(documentChange.Key, document);
-                    }
-                    else
-                    {
-                        document = documentCache[documentChange.Key];
-                    }
-
-                    await documentChange.Value(document);
-                }
+                await HandlePerDocumentChange(documentCache, perDocumentChange);
 
                 break;
             }
@@ -206,6 +190,76 @@ await _elastic.TryGetAsync(async () =>
         }
     }
 
+    private async Task HandlePerDocumentChange(Dictionary<Guid, T> documentCache, ElasticPerDocumentChange<T> perDocumentChange)
+    {
+        try
+        {
+            foreach (var documentChange in perDocumentChange.Changes)
+            {
+                T? document;
+
+                if (!documentCache.ContainsKey(documentChange.Key))
+                {
+                    document = (await _elastic.TryGetAsync(async () =>
+                            (await _elastic.WriteClient.GetAsync<T>(documentChange.Key))
+                            .ThrowOnFailure()))
+                        .Source;
+
+                    documentCache.Add(documentChange.Key, document);
+                }
+                else
+                {
+                    document = documentCache[documentChange.Key];
+                }
+
+                await documentChange.Value(document);
+            }
+        }
+        catch (ElasticsearchPerDocumentChangeException e)
+        {
+            await using var organisationRegistryContext = _contextFactory.Create();
+
+            switch (perDocumentChange)
+            {
+                case ElasticPerDocumentChange<OrganisationDocument>:
+                    organisationRegistryContext.OrganisationsToRebuild.Add(
+                        new OrganisationToRebuild
+                        {
+                            OrganisationId = e.AggregateId,
+                        });
+                    await organisationRegistryContext.SaveChangesAsync();
+                    _logger.LogWarning(
+                        0,
+                        e,
+                        "[{ProjectionName}] Error occured for {AggregateId} in ES while processing envelope #{EnvelopeNumber}, adding it to entities to rebuild",
+                        ProjectionName,
+                        e.AggregateId,
+                        e.EnvelopeNumber);
+
+                    break;
+
+                case ElasticPerDocumentChange<BodyDocument>:
+                    organisationRegistryContext.BodiesToRebuild.Add(
+                        new BodyToRebuild()
+                        {
+                            BodyId = e.AggregateId,
+                        });
+                    await organisationRegistryContext.SaveChangesAsync();
+                    _logger.LogWarning(
+                        0,
+                        e,
+                        "[{ProjectionName}] Error occured for {AggregateId} in ES while processing envelope #{EnvelopeNumber}, adding it to entities to rebuild",
+                        ProjectionName,
+                        e.AggregateId,
+                        e.EnvelopeNumber);
+
+                    break;
+            }
+
+            throw;
+        }
+    }
+
     private async Task FlushDocuments(Dictionary<Guid, T> documentCache)
     {
         if (documentCache.Any())

src/OrganisationRegistry.ElasticSearch.Projections/Infrastructure/Change/ElasticPerDocumentChange.cs

@@ -2,34 +2,29 @@ namespace OrganisationRegistry.ElasticSearch.Projections.Infrastructure.Change;
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
+using Be.Vlaanderen.Basisregisters.Utilities;
+using Client;
+using Humanizer;
 
 public class ElasticPerDocumentChange<T> : IElasticChange where T: IDocument
 {
     public ElasticPerDocumentChange(Guid id, Action<T> change)
     {
         Changes = new Dictionary<Guid, Func<T, Task>> {{id, doc =>
             {
-                change(doc);
-                return Task.CompletedTask;
+                try{
+                    change(doc);
+                    return Task.CompletedTask;
+                }catch (InvalidOperationException)
+                {
+                    return Task.FromException(new ElasticsearchPerDocumentChangeException(doc.Id, doc.ChangeId));
+                }
             }
         }};
     }
 
-    public ElasticPerDocumentChange(Dictionary<Guid, Action<T>> changes)
-    {
-        Changes = new Dictionary<Guid, Func<T, Task>>();
-        foreach (var change in changes)
-        {
-            Func<T, Task> changeAction = doc =>
-            {
-                change.Value(doc);
-                return Task.CompletedTask;
-            };
-
-            Changes.Add(change.Key, changeAction);
-        }
-    }
 
     public ElasticPerDocumentChange(Guid id, Func<T, Task> change)
     {
@@ -43,3 +38,4 @@ public ElasticPerDocumentChange(Dictionary<Guid, Func<T, Task>> changes)
 
     public Dictionary<Guid, Func<T, Task>> Changes { get; init; }
 }
+

src/OrganisationRegistry.ElasticSearch/Client/ThrowOnFailureExtension.cs

@@ -22,8 +22,8 @@ public static IGetResponse<T> ThrowOnFailure<T>(this IGetResponse<T> response) w
     {
         if (response.IsValid) return response;
 
-        if (response.ApiCall.HttpStatusCode == 404 && typeof(T) == typeof(OrganisationDocument))
-            throw new ElasticsearchOrganisationNotFoundException(((GetResponse<OrganisationDocument>)response).Id);
+        if (response.ApiCall.HttpStatusCode == 404)
+            throw new ElasticsearchAggregateNotFoundException(((GetResponse<T>)response).Id);
 
         throw new ElasticsearchException(response.DebugInformation, response.OriginalException);
     }
@@ -76,12 +76,25 @@ public static void ThrowOnFailure(this BulkResponse response)
     }
 }
 
-public class ElasticsearchOrganisationNotFoundException : ElasticsearchException
+public class ElasticsearchAggregateNotFoundException : ElasticsearchException
 {
-    public string OrganisationId { get; }
+    public string AggregateId { get; }
 
-    public ElasticsearchOrganisationNotFoundException(string organisationId)
+    public ElasticsearchAggregateNotFoundException(string aggregateId)
     {
-        OrganisationId = organisationId;
+        AggregateId = aggregateId;
+    }
+}
+
+
+public class ElasticsearchPerDocumentChangeException : ElasticsearchException
+{
+    public Guid AggregateId { get; }
+    public int EnvelopeNumber { get; }
+
+    public ElasticsearchPerDocumentChangeException(Guid aggregateId, int envelopeNumber)
+    {
+        AggregateId = aggregateId;
+        EnvelopeNumber = envelopeNumber;
     }
 }