fix: or-2893 wip, amend please

Author: koenmetsu

src/OrganisationRegistry.ElasticSearch.Projections/BaseRunner.cs

@@ -8,6 +8,7 @@ namespace OrganisationRegistry.ElasticSearch.Projections;
 using System.Threading.Tasks;
 using Client;
 using Configuration;
+using ElasticSearch.Organisations;
 using Infrastructure;
 using Infrastructure.Change;
 using Microsoft.Extensions.Logging;
@@ -128,21 +129,29 @@ public async Task Run()
             _metrics.NumberOfEnvelopesHandledGauge = envelopes.Count;
             _metrics.NumberOfEnvelopesHandledCounter = envelopes.Count;
         }
-        catch (ElasticsearchOrganisationNotFoundException organisationNotFoundException)
+        catch (ElasticsearchAggregateNotFoundException organisationNotFoundException)
         {
             await using var organisationRegistryContext = _contextFactory.Create();
+
+            // TODO: switch over document type, place line in right xxxToRebuild table
+            switch (typeof(T).Name)
+            {
+                case nameof(OrganisationDocument):
+                    break;
+            }
+
             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;
         }

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

@@ -2,34 +2,27 @@ namespace OrganisationRegistry.ElasticSearch.Projections.Infrastructure.Change;
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
+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 =>
             {
+                try{
                 change(doc);
                 return Task.CompletedTask;
+                }catch (InvalidOperationException)
+                {
+                    return Task.FromException(new ElasticsearchPerDocumentChangeException());
+                }
             }
         }};
     }
 
-    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 +36,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,23 @@ 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 string AggregateId { get; }
+
+    public ElasticsearchPerDocumentChangeException(string aggregateId)
+    {
+        AggregateId = aggregateId;
     }
 }