remove the deep cloning and update linking to use the changeset

Btms.Business.Tests/PreProcessing/MovementPreProcessingTests.cs

@@ -0,0 +1,39 @@
+using Btms.Backend.Data.InMemory;
+using Btms.Business.Pipelines.PreProcessing;
+using Btms.Types.Alvs;
+using FluentAssertions;
+using Microsoft.Extensions.Logging.Abstractions;
+using TestDataGenerator;
+using Xunit;
+
+namespace Btms.Business.Tests.PreProcessing;
+
+public class MovementPreProcessingTests
+{
+    [Fact]
+    public async Task WhenNotificationNotExists_ThenShouldBeCreated()
+    {
+        // ARRANGE
+        var clearanceRequest = CreateAlvsClearanceRequest();
+        var dbContext = new MemoryMongoDbContext();
+        var preProcessor = new MovementPreProcessor(dbContext, NullLogger<MovementPreProcessor>.Instance);
+
+
+        // ACT
+        var preProcessingResult = await preProcessor.Process(
+            new PreProcessingContext<AlvsClearanceRequest>(clearanceRequest, "TestMessageId"));
+
+        // ASSERT
+        preProcessingResult.Outcome.Should().Be(PreProcessingOutcome.New);
+        var savedMovement = await dbContext.Movements.Find(clearanceRequest!.Header!.EntryReference!);
+        savedMovement.Should().NotBeNull();
+        savedMovement.AuditEntries.Count.Should().Be(1);
+        savedMovement.AuditEntries[0].Status.Should().Be("Created");
+    }
+
+    private static AlvsClearanceRequest CreateAlvsClearanceRequest()
+    {
+        return ClearanceRequestBuilder.Default()
+            .WithValidDocumentReferenceNumbers().Build();
+    }
+}

Btms.Business.Tests/PreProcessing/NotificationsPreProcessingTests.cs

@@ -0,0 +1,71 @@
+using Btms.Backend.Data.InMemory;
+using Btms.Business.Pipelines.PreProcessing;
+using Btms.Types.Ipaffs;
+using Btms.Types.Ipaffs.Mapping;
+using FluentAssertions;
+using Microsoft.Extensions.Logging.Abstractions;
+using TestDataGenerator;
+using Xunit;
+
+namespace Btms.Business.Tests.PreProcessing
+{
+    public class NotificationsPreProcessingTests 
+    {
+        [Fact]
+        public async Task WhenNotificationNotExists_ThenShouldBeCreated()
+        {
+            // ARRANGE
+            var notification = CreateImportNotification();
+            var dbContext = new MemoryMongoDbContext();
+            var preProcessor = new ImportNotificationPreProcessor(dbContext, NullLogger<ImportNotificationPreProcessor>.Instance);
+
+            // ACT
+            var preProcessingResult = await preProcessor.Process(
+                new PreProcessingContext<ImportNotification>(notification, "TestMessageId"));
+
+            // ASSERT
+            preProcessingResult.Outcome.Should().Be(PreProcessingOutcome.New);
+            var savedNotification = await dbContext.Notifications.Find(notification!.ReferenceNumber!);
+            savedNotification.Should().NotBeNull();
+            savedNotification.AuditEntries.Count.Should().Be(1);
+            savedNotification.AuditEntries[0].Status.Should().Be("Created");
+        }
+
+        [Fact]
+        public async Task WhenNotificationExists_AndLastUpdatedIsNewer_ThenShouldBeUpdated()
+        {
+            // ARRANGE
+            var notification = CreateImportNotification();
+            var dbContext = new MemoryMongoDbContext();
+            await dbContext.Notifications.Insert(notification.MapWithTransform());
+            notification.LastUpdated = notification.LastUpdated?.AddHours(1);
+            var preProcessor = new ImportNotificationPreProcessor(dbContext, NullLogger<ImportNotificationPreProcessor>.Instance);
+
+
+            // ACT
+            var preProcessingResult = await preProcessor.Process(
+                new PreProcessingContext<ImportNotification>(notification, "TestMessageId"));
+
+            // ASSERT
+            preProcessingResult.Outcome.Should().Be(PreProcessingOutcome.Changed);
+            var savedNotification = await dbContext.Notifications.Find(notification!.ReferenceNumber!);
+            savedNotification.Should().NotBeNull();
+            savedNotification.AuditEntries.Count.Should().Be(1);
+            savedNotification.AuditEntries[0].Status.Should().Be("Updated");
+        }
+
+        private static ImportNotification CreateImportNotification()
+        {
+            return ImportNotificationBuilder.Default()
+                .WithReferenceNumber(ImportNotificationTypeEnum.Chedpp, 1, DateTime.UtcNow, 1)
+                .WithRandomCommodities(1, 2)
+                .Do(x =>
+                {
+                    foreach (var parameterSet in x.PartOne?.Commodities?.ComplementParameterSets!)
+                    {
+                        parameterSet.KeyDataPairs = null;
+                    }
+                }).Build();
+        }
+    }
+}

Btms.Business.Tests/Services/LinkingServiceTests.cs

@@ -2,14 +2,15 @@
 using Btms.Backend.Data.InMemory;
 using Btms.Business.Services;
 using Btms.Metrics;
-using Btms.Model;
 using Btms.Model.Alvs;
+using Btms.Model.ChangeLog;
 using Btms.Model.Ipaffs;
 using FluentAssertions;
 using Microsoft.Extensions.Logging.Abstractions;
 using Xunit;
 using Document = Btms.Model.Alvs.Document;
 using Items = Btms.Model.Alvs.Items;
+using Movement = Btms.Model.Movement;
 
 namespace Btms.Business.Tests.Services;
 
@@ -49,7 +50,7 @@ public async Task LinkMovement_ExistingRequest_IncludesFieldsOfInterest_Matching
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.Linked);
+        linkResult.Outcome.Should().Be(LinkOutcome.Linked);
         linkResult.Notifications.Count.Should().Be(1);
         linkResult.Movements.Count.Should().Be(1);
     }
@@ -68,7 +69,7 @@ public async Task LinkMovement_ExistingRequest_IncludesFieldsOfInterest_NoMatchi
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.NotLinked);
+        linkResult.Outcome.Should().Be(LinkOutcome.NotLinked);
         linkResult.Notifications.Count.Should().Be(0);
         linkResult.Movements.Count.Should().Be(1);
     }
@@ -87,7 +88,7 @@ public async Task LinkMovement_ExistingRequest_NoFieldsOfInterest_NoMatchingTrig
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.NotLinked);
+        linkResult.Outcome.Should().Be(LinkOutcome.NotLinked);
         linkResult.Notifications.Count.Should().Be(0);
         linkResult.Movements.Count.Should().Be(0);
     }
@@ -106,7 +107,7 @@ public async Task LinkMovement_NewRequest_MatchingCHED_AddsAllToLinkResult()
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.Linked);
+        linkResult.Outcome.Should().Be(LinkOutcome.Linked);
         linkResult.Notifications.Count.Should().Be(1);
         linkResult.Movements.Count.Should().Be(1);
     }
@@ -125,7 +126,7 @@ public async Task LinkMovement_NewRequest_MultipleMatchingCHEDs_AddsAllToLinkRes
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.Linked);
+        linkResult.Outcome.Should().Be(LinkOutcome.Linked);
         linkResult.Notifications.Count.Should().Be(2);
         linkResult.Movements.Count.Should().Be(1);
     }
@@ -142,7 +143,7 @@ public async Task LinkMovement_NewRequest_NoMatchingCHEDs_NoMatchingTriggered()
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.NotLinked);
+        linkResult.Outcome.Should().Be(LinkOutcome.NotLinked);
         linkResult.Notifications.Count.Should().Be(0);
         linkResult.Movements.Count.Should().Be(1);
     }
@@ -161,7 +162,7 @@ public async Task LinkNotification_ExistingNotification_IncludesFieldsOfInterest
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.Linked);
+        linkResult.Outcome.Should().Be(LinkOutcome.Linked);
         linkResult.Notifications.Count.Should().Be(1);
         linkResult.Movements.Count.Should().Be(1);
     }
@@ -180,7 +181,7 @@ public async Task LinkNotification_ExistingNotification_IncludesFieldsOfInterest
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.Linked);
+        linkResult.Outcome.Should().Be(LinkOutcome.Linked);
         linkResult.Notifications.Count.Should().Be(1);
         linkResult.Movements.Count.Should().Be(2);
     }
@@ -199,7 +200,7 @@ public async Task LinkNotification_ExistingNotification_IncludesFieldsOfInterest
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.NotLinked);
+        linkResult.Outcome.Should().Be(LinkOutcome.NotLinked);
         linkResult.Notifications.Count.Should().Be(1);
         linkResult.Movements.Count.Should().Be(0);
     }
@@ -218,7 +219,7 @@ public async Task LinkNotification_ExistingNotification_NoFieldsOfInterest_NoMat
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.NotLinked);
+        linkResult.Outcome.Should().Be(LinkOutcome.NotLinked);
         linkResult.Notifications.Count.Should().Be(0);
         linkResult.Movements.Count.Should().Be(0);
     }
@@ -237,7 +238,7 @@ public async Task LinkNotification_NewNotification_MatchingMRN_AddsAllToLinkResu
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.Linked);
+        linkResult.Outcome.Should().Be(LinkOutcome.Linked);
         linkResult.Notifications.Count.Should().Be(1);
         linkResult.Movements.Count.Should().Be(4);
     }
@@ -256,7 +257,7 @@ public async Task LinkNotification_NewNotification_MultipleMatchingMRNs_AddsAllT
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.Linked);
+        linkResult.Outcome.Should().Be(LinkOutcome.Linked);
         linkResult.Notifications.Count.Should().Be(1);
         linkResult.Movements.Count.Should().Be(1);
     }
@@ -273,7 +274,7 @@ public async Task LinkNotification_NewNotification_NoMatchingMRNs_NoMatchingTrig
 
         // Assert
         linkResult.Should().NotBeNull();
-        linkResult.State.Should().Be(LinkState.NotLinked);
+        linkResult.Outcome.Should().Be(LinkOutcome.NotLinked);
         linkResult.Notifications.Count.Should().Be(1);
         linkResult.Movements.Count.Should().Be(0);
     }
@@ -309,8 +310,9 @@ private MovementLinkContext CreateMovementContext(Movement? movement, List<Impor
                         : [ new Document { DocumentReference = GenerateDocumentReference(x) } ]
                 }).ToList()
             } : null;
-
-        var output = LinkContext.ForMovement(mov, existingMovement);
+
+        var changeSet = mov.GenerateChangeSet(existingMovement);
+        var output = LinkContext.ForMovement(mov, createExistingMovement ? changeSet : null);
 
         return output;
     }
@@ -348,7 +350,8 @@ private ImportNotificationLinkContext CreateNotificationContext(int chedReferenc
             }
             : null;
 
-        var output = LinkContext.ForImportNotification(notification, existingNotification);
+        var changeSet = notification.GenerateChangeSet(existingNotification);
+        var output = LinkContext.ForImportNotification(notification, createExistingNotification ? changeSet : null);
 
         return output;
     }

Btms.Business/Commands/SyncHandler.cs

@@ -155,7 +155,7 @@ private async Task SyncBlob<TRequest>(string path, string topic, IBlobItem item,
                         var message = sensitiveDataSerializer.Deserialize<TRequest>(blobContent, _ => { })!;
                         var headers = new Dictionary<string, object>()
                         {
-                            { "messageId", item.Name }, { "jobId", job.JobId }
+                            { "messageId", item.Name.TrimStart(path.ToCharArray()) }, { "jobId", job.JobId }
                         };
                         if (BtmsDiagnostics.ActivitySource.HasListeners())
                         {

Btms.Business/Extensions/ServiceCollectionExtensions.cs

@@ -9,9 +9,12 @@
 using Btms.Common.Extensions;
 using Btms.Metrics.Extensions;
 using Btms.SensitiveData;
+using Btms.Types.Ipaffs;
 using MediatR;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
+using Btms.Business.Pipelines.PreProcessing;
+using Btms.Types.Alvs;
 
 namespace Btms.Business.Extensions
 {
@@ -59,6 +62,9 @@ public static IServiceCollection AddBusinessServices(this IServiceCollection ser
 
             services.AddScoped<ILinkingService, LinkingService>();
 
+            services.AddScoped<IPreProcessor<ImportNotification, Model.Ipaffs.ImportNotification>, ImportNotificationPreProcessor>();
+            services.AddScoped<IPreProcessor<AlvsClearanceRequest, Model.Movement>, MovementPreProcessor>();
+
             return services;
         }
     }

Btms.Business/Pipelines/PreProcessing/IPreProcessor.cs

@@ -0,0 +1,8 @@
+using Btms.Model.Data;
+
+namespace Btms.Business.Pipelines.PreProcessing;
+
+public interface IPreProcessor<TInput, TOutput> where TOutput : IAuditable
+{
+    Task<PreProcessingResult<TOutput>> Process(PreProcessingContext<TInput> preProcessingContext);
+}

Btms.Business/Pipelines/PreProcessing/ImportNotificationPreProcessor.cs

@@ -0,0 +1,50 @@
+using Btms.Backend.Data;
+using Btms.Common.Extensions;
+using Btms.Model.ChangeLog;
+using Btms.Types.Ipaffs;
+using Btms.Types.Ipaffs.Mapping;
+using Microsoft.Extensions.Logging;
+
+namespace Btms.Business.Pipelines.PreProcessing;
+
+public class ImportNotificationPreProcessor(IMongoDbContext dbContext, ILogger<ImportNotificationPreProcessor> logger) : IPreProcessor<ImportNotification, Model.Ipaffs.ImportNotification>
+{
+    public async Task<PreProcessingResult<Model.Ipaffs.ImportNotification>> Process(PreProcessingContext<ImportNotification> preProcessingContext)
+    {
+        var internalNotification = preProcessingContext.Message.MapWithTransform();
+        var existingNotification =
+            await dbContext.Notifications.Find(preProcessingContext.Message.ReferenceNumber!);
+
+        if (existingNotification is null)
+        {
+            internalNotification.Create(preProcessingContext.MessageId);
+            await dbContext.Notifications.Insert(internalNotification);
+            return PreProcessResult.New(internalNotification);
+        }
+
+
+        if (internalNotification.UpdatedSource.TrimMicroseconds() >
+            existingNotification.UpdatedSource.TrimMicroseconds())
+        {
+            internalNotification.AuditEntries = existingNotification.AuditEntries;
+            internalNotification.CreatedSource = existingNotification.CreatedSource;
+
+            var changeSet = internalNotification.GenerateChangeSet(existingNotification);
+
+            internalNotification.Update(preProcessingContext.MessageId, changeSet);
+            await dbContext.Notifications.Update(internalNotification, existingNotification._Etag);
+
+            return PreProcessResult.Changed(internalNotification, changeSet);
+        }
+
+        if (internalNotification.UpdatedSource.TrimMicroseconds() ==
+                 existingNotification.UpdatedSource.TrimMicroseconds())
+        {
+            return PreProcessResult.AlreadyProcessed(existingNotification);
+        }
+
+        logger.MessageSkipped(preProcessingContext.MessageId!, preProcessingContext.Message.ReferenceNumber!);
+        return PreProcessResult.Skipped(existingNotification);
+
+    }
+}