Or 2472 restructure

.localstack/init/ready.d/01_sqs-init.redrive.json

@@ -1,5 +1,3 @@
 {
-  "ContentBasedDeduplication": "true",
-  "DeduplicationScope": "messageGroup",
   "RedrivePolicy": "{\"maxReceiveCount\": \"3\", \"deadLetterTargetArn\": \"arn:aws:sqs:eu-west-1:000000000000:verenigingsregister-addresssmatch-dlq\"}"
 }

.localstack/init/ready.d/02_sqs-init.redrive.json

@@ -1,5 +1,3 @@
 {
-  "ContentBasedDeduplication": "true",
-  "DeduplicationScope": "messageGroup",
   "RedrivePolicy": "{\"maxReceiveCount\": \"3\", \"deadLetterTargetArn\": \"arn:aws:sqs:eu-west-1:000000000000:verenigingsregister-grarsync-dlq\"}"
 }

paket.lock

@@ -171,7 +171,7 @@ NUGET
       Autofac.Extensions.DependencyInjection (>= 9.0)
       Be.Vlaanderen.Basisregisters.EventHandling (5.0)
     Be.Vlaanderen.Basisregisters.Generators.Guid.Deterministic (4.0)
-    Be.Vlaanderen.Basisregisters.GrAr.Contracts (21.7)
+    Be.Vlaanderen.Basisregisters.GrAr.Contracts (21.15)
     Be.Vlaanderen.Basisregisters.Middleware.AddProblemJsonHeader (3.0)
     Be.Vlaanderen.Basisregisters.ProblemDetails (9.0)
       System.Reflection.Metadata (>= 8.0)

src/AssociationRegistry.Admin.Api/AssociationRegistry.Admin.Api.csproj

@@ -65,6 +65,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <Folder Include="GrarConsumer\Handlers\Fusies\" />
     <Folder Include="Infrastructure\ConfigurationBindings\" />
     <Folder Include="Verenigingen\Contactgegevens\FeitelijkeVereniging\" />
     <Folder Include="Verenigingen\Vertegenwoordigers\FeitelijkeVereniging\" />

src/AssociationRegistry.Admin.Api/GrarConsumer/Finders/LocatieFinder.cs

@@ -0,0 +1,52 @@
+namespace AssociationRegistry.Admin.Api.GrarConsumer.Finders;
+
+using Grar.GrarUpdates.LocatieFinder;
+using Schema.Detail;
+using Marten;
+
+public class LocatieFinder : ILocatieFinder
+{
+    private readonly IDocumentStore _documentStore;
+
+    public LocatieFinder(IDocumentStore documentStore)
+    {
+        _documentStore = documentStore;
+    }
+
+    public async Task<LocatiesPerVCodeCollection> FindLocaties(params int[] adresIds)
+    {
+        return await FindLocaties(
+            adresIds
+               .Select(x => x.ToString())
+               .ToArray());
+    }
+
+    public async Task<LocatiesPerVCodeCollection> FindLocaties(params string[] adresIds)
+    {
+        var locatieIdsGroupedByVCode =
+            GroupByVCode(
+                await FindLocatieLookupDocuments(adresIds)
+            );
+
+        return LocatiesPerVCodeCollection.FromLocatiesPerVCode(locatieIdsGroupedByVCode);
+    }
+
+    public async Task<IEnumerable<LocatieLookupData>> FindLocatieLookupDocuments(string[] adresIds)
+    {
+        await using var session = _documentStore.LightweightSession();
+
+        return session.Query<LocatieLookupDocument>()
+                      .Where(x => adresIds.Contains(x.AdresId))
+                      .Select(x => new LocatieLookupData(x.LocatieId, x.AdresId, x.VCode));
+    }
+
+    private static Dictionary<string, LocatieLookupData[]> GroupByVCode(IEnumerable<LocatieLookupData> locaties)
+    {
+        return locaties
+              .GroupBy(x => x.VCode)
+              .ToDictionary(grouping => grouping.Key,
+                            documents => documents.Select(x => x)
+                                                  .ToArray());
+    }
+}
+

src/AssociationRegistry.Admin.Api/GrarConsumer/Finders/LocatieMetVCode.cs

@@ -0,0 +1,3 @@
+namespace AssociationRegistry.Admin.Api.GrarConsumer.Finders;
+
+public record LocatieMetVCode(string VCode, int LocatieId);

src/AssociationRegistry.Admin.Api/GrarSync/AddressKafkaConfiguration.cs → src/AssociationRegistry.Admin.Api/GrarConsumer/Kafka/AddressKafkaConfiguration.cs

@@ -1,7 +1,7 @@
-namespace AssociationRegistry.Admin.Api.GrarSync;
+namespace AssociationRegistry.Admin.Api.GrarConsumer.Kafka;
 
-using Confluent.Kafka;
 using Hosts.Configuration;
+using Confluent.Kafka;
 
 public class AddressKafkaConfiguration : ConsumerConfig
 {

src/AssociationRegistry.Admin.Api/GrarSync/AddressKafkaConsumer.cs → src/AssociationRegistry.Admin.Api/GrarConsumer/Kafka/AddressKafkaConsumer.cs

@@ -1,36 +1,39 @@
-namespace AssociationRegistry.Admin.Api.GrarSync;
+namespace AssociationRegistry.Admin.Api.GrarConsumer.Kafka;
 
+using Infrastructure.AWS;
+using Infrastructure.Extensions;
+using Notifications.Messages;
 using AssociationRegistry.Notifications;
 using Be.Vlaanderen.Basisregisters.GrAr.Contracts.AddressRegistry;
 using Confluent.Kafka;
+using Grar.GrarUpdates.Fusies;
+using Grar.GrarUpdates.Hernummering;
 using IdentityModel;
-using Infrastructure.AWS;
-using Infrastructure.Extensions;
 using Marten;
-using Notifications.Messages;
 using Polly;
+using StraatHernummering;
 using System.Text;
 
 public class AddressKafkaConsumer : BackgroundService
 {
     private readonly AddressKafkaConfiguration _configuration;
-    private readonly TeHeradresserenLocatiesMapper _teHeradresserenLocatiesMapper;
-    private readonly SqsClientWrapper _sqsClientWrapper;
+    private readonly IFusieEventProcessor _fusieEventProcessor;
+    private readonly ITeHernummerenStraatEventProcessor _teHernummerenStraatEventProcessor;
     private readonly IDocumentStore _store;
     private readonly INotifier _notifier;
     private readonly ILogger<AddressKafkaConsumer> _logger;
 
     public AddressKafkaConsumer(
         AddressKafkaConfiguration configuration,
-        TeHeradresserenLocatiesMapper teHeradresserenLocatiesMapper,
-        SqsClientWrapper sqsClientWrapper,
+        IFusieEventProcessor fusieEventProcessor,
+        ITeHernummerenStraatEventProcessor teHernummerenStraatEventProcessor,
         IDocumentStore store,
         INotifier notifier,
         ILogger<AddressKafkaConsumer> logger)
     {
         _configuration = configuration;
-        _teHeradresserenLocatiesMapper = teHeradresserenLocatiesMapper;
-        _sqsClientWrapper = sqsClientWrapper;
+        _fusieEventProcessor = fusieEventProcessor;
+        _teHernummerenStraatEventProcessor = teHernummerenStraatEventProcessor;
         _store = store;
         _notifier = notifier;
         _logger = logger;
@@ -80,17 +83,25 @@ private async Task Consume(CancellationToken stoppingToken)
 
                 switch (message)
                 {
+                    case AddressWasRetiredBecauseOfMunicipalityMerger addressWasRetiredBecauseOfMunicipalityMerger:
+                        await _fusieEventProcessor.Process(
+                            addressWasRetiredBecauseOfMunicipalityMerger.AddressPersistentLocalId,
+                            addressWasRetiredBecauseOfMunicipalityMerger.NewAddressPersistentLocalId,
+                            idempotenceKey);
+                        break;
+
+                    case AddressWasRejectedBecauseOfMunicipalityMerger addressWasRejectedBecauseOfMunicipalityMerger:
+                        await _fusieEventProcessor.Process(
+                            addressWasRejectedBecauseOfMunicipalityMerger.AddressPersistentLocalId,
+                            addressWasRejectedBecauseOfMunicipalityMerger.NewAddressPersistentLocalId,
+                            idempotenceKey);
+                        break;
+
                     case StreetNameWasReaddressed streetNameWasReaddressed:
                         _logger.LogInformation($"{nameof(StreetNameWasReaddressed)} found! Offset: {result.Offset}");
 
-                        var teHeradresserenLocatiesMessages =
-                            await _teHeradresserenLocatiesMapper.ForAddress(streetNameWasReaddressed.ReaddressedHouseNumbers,
-                                                                            idempotenceKey);
-
-                        foreach (var teHeradresserenLocatiesMessage in teHeradresserenLocatiesMessages)
-                        {
-                            await _sqsClientWrapper.QueueReaddressMessage(teHeradresserenLocatiesMessage);
-                        }
+                        await _teHernummerenStraatEventProcessor.Process(
+                            TeHernummerenStraatFactory.From(streetNameWasReaddressed), idempotenceKey);
 
                         break;
 

src/AssociationRegistry.Admin.Api/GrarSync/AddressKafkaConsumerOffset.cs → src/AssociationRegistry.Admin.Api/GrarConsumer/Kafka/AddressKafkaConsumerOffset.cs

@@ -1,4 +1,4 @@
-namespace AssociationRegistry.Admin.Api.GrarSync;
+namespace AssociationRegistry.Admin.Api.GrarConsumer.Kafka;
 
 using Marten.Schema;
 

src/AssociationRegistry.Admin.Api/GrarConsumer/Kafka/StraatHernummering/TeHernummerenAdres.cs

@@ -0,0 +1,26 @@
+namespace AssociationRegistry.Admin.Api.GrarConsumer.Kafka.StraatHernummering;
+
+using Be.Vlaanderen.Basisregisters.GrAr.Contracts.AddressRegistry;
+using Grar.GrarUpdates.Hernummering;
+
+public static class TeHernummerenStraatFactory
+{
+    public static TeHernummerenStraat From(StreetNameWasReaddressed streetNameWasReaddressed)
+    {
+        var result = new List<TeHernummerenAdres>();
+
+        foreach (var readdressed in streetNameWasReaddressed.ReaddressedHouseNumbers)
+        {
+            result.AddRange(readdressed.ReaddressedBoxNumbers.Select(From));
+            result.Add(From(readdressed.ReaddressedHouseNumber));
+        }
+
+        return new(result);
+    }
+
+    private static TeHernummerenAdres From(ReaddressedAddressData readdressedAddressData)
+        => new(
+            readdressedAddressData.SourceAddressPersistentLocalId,
+            readdressedAddressData.DestinationAddressPersistentLocalId);
+}
+

src/AssociationRegistry.Admin.Api/Infrastructure/AWS/SqsClientWrapper.cs

@@ -1,13 +1,17 @@
 namespace AssociationRegistry.Admin.Api.Infrastructure.AWS;
 
+using Acties.GrarConsumer.HeradresseerLocaties;
 using Amazon.SQS;
-using Grar.HeradresseerLocaties;
+using Framework;
+using Grar.GrarUpdates.Fusies.TeHeradresserenLocaties;
 using Hosts.Configuration;
 using Hosts.Configuration.ConfigurationBindings;
 using Kbo;
 using System.Text.Json;
 
-public class SqsClientWrapper
+
+
+public class SqsClientWrapper : ISqsClientWrapper
 {
     private readonly IAmazonSQS _sqsClient;
     private readonly string _kboSyncQueueUrl;
@@ -20,7 +24,12 @@ public SqsClientWrapper(IAmazonSQS sqsClient, AppSettings appSettings, GrarOptio
         _readdressQueueUrl = grarOptions.Sqs.GrarSyncQueueUrl;
     }
 
-    public async Task QueueReaddressMessage(TeHeradresserenLocatiesMessage message)
+    public async Task QueueReaddressMessage(HeradresseerLocatiesMessage message)
+    {
+        await QueueMessage(message);
+    }
+
+    public async Task QueueMessage<TMessage>(TMessage message)
     {
         await _sqsClient.SendMessageAsync(
             _readdressQueueUrl,
@@ -29,9 +38,6 @@ await _sqsClient.SendMessageAsync(
 
     public async Task QueueKboNummerToSynchronise(string kboNummer)
     {
-        await _sqsClient.SendMessageAsync(
-            _kboSyncQueueUrl,
-            JsonSerializer.Serialize(
-                new TeSynchroniserenKboNummerMessage(kboNummer)));
+        await QueueMessage(new TeSynchroniserenKboNummerMessage(kboNummer));
     }
 }

src/AssociationRegistry.Admin.Api/Infrastructure/Extensions/MartenExtensions.cs

@@ -1,7 +1,8 @@
 namespace AssociationRegistry.Admin.Api.Infrastructure.Extensions;
 
 using Formats;
-using GrarSync;
+using GrarConsumer;
+using GrarConsumer.Kafka;
 using Hosts.Configuration.ConfigurationBindings;
 using JasperFx.CodeGeneration;
 using Json;