fix: event validation (#1)

* fix: event validation * chore: pipeline
atmina · Mar 15, 2024 · eebee43 · eebee43
1 parent 6c19bc5
commit eebee43
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 8 deletions.
diff --git a/src/Fluss.Testing/AggregateTestBed.cs b/src/Fluss.Testing/AggregateTestBed.cs
@@ -17,8 +17,8 @@ namespace Fluss.Testing;
     public AggregateTestBed()
     {
         var validator = new Mock<IRootValidator>();
-        validator.Setup(v => v.ValidateEvent(It.IsAny<EventEnvelope>()))
-            .Returns<EventEnvelope>(_ => Task.CompletedTask);
+        validator.Setup(v => v.ValidateEvent(It.IsAny<EventEnvelope>(), It.IsAny<IReadOnlyList<EventEnvelope>?>()))
+            .Returns<EventEnvelope, IReadOnlyList<EventEnvelope>?>((_, _) => Task.CompletedTask);
         validator.Setup(v => v.ValidateAggregate(It.IsAny<AggregateRoot>(), It.IsAny<UnitOfWork.UnitOfWork>()))
             .Returns<AggregateRoot, UnitOfWork.UnitOfWork>((_, _) => Task.CompletedTask);
 

diff --git a/src/Fluss.UnitTest/Core/UnitOfWork/UnitOfWorkTest.cs b/src/Fluss.UnitTest/Core/UnitOfWork/UnitOfWorkTest.cs
@@ -28,8 +28,8 @@ public UnitOfWorkTest()
         _policies = new List<Policy>();
 
         _validator = new Mock<IRootValidator>(MockBehavior.Strict);
-        _validator.Setup(v => v.ValidateEvent(It.IsAny<EventEnvelope>()))
-            .Returns<EventEnvelope>(_ => Task.CompletedTask);
+        _validator.Setup(v => v.ValidateEvent(It.IsAny<EventEnvelope>(), It.IsAny<IReadOnlyList<EventEnvelope>?>()))
+            .Returns<EventEnvelope, IReadOnlyList<EventEnvelope>?>((_, _) => Task.CompletedTask);
         _validator.Setup(v => v.ValidateAggregate(It.IsAny<AggregateRoot>(), It.IsAny<Fluss.UnitOfWork.UnitOfWork>()))
             .Returns<AggregateRoot, Fluss.UnitOfWork.UnitOfWork>((_, _) => Task.CompletedTask);
 

diff --git a/src/Fluss/UnitOfWork/UnitOfWork.Aggregates.cs b/src/Fluss/UnitOfWork/UnitOfWork.Aggregates.cs
@@ -93,7 +93,12 @@ internal async ValueTask CommitInternal()
     {
         using var activity = FlussActivitySource.Source.StartActivity();
 
-        await Task.WhenAll(PublishedEventEnvelopes.Select(envelope => _validator.ValidateEvent(envelope)));
+        var validatedEnvelopes = new List<EventEnvelope>();
+        foreach (var envelope in PublishedEventEnvelopes)
+        {
+            await _validator.ValidateEvent(envelope, validatedEnvelopes);
+            validatedEnvelopes.Add(envelope);
+        }
 
         await _eventRepository.Publish(PublishedEventEnvelopes);
         _consistentVersion += PublishedEventEnvelopes.Count;

diff --git a/src/Fluss/Validation/RootValidator.cs b/src/Fluss/Validation/RootValidator.cs
@@ -7,7 +7,7 @@ namespace Fluss.Core.Validation;
 
 public interface IRootValidator
 {
-    public Task ValidateEvent(EventEnvelope envelope);
+    public Task ValidateEvent(EventEnvelope envelope, IReadOnlyList<EventEnvelope>? PreviousEnvelopes = null);
     public Task ValidateAggregate(AggregateRoot aggregate, Fluss.UnitOfWork.UnitOfWork unitOfWork);
 }
 
@@ -54,11 +54,18 @@ private void CacheEventValidators(IEnumerable<EventValidator> validators)
         }
     }
 
-    public async Task ValidateEvent(EventEnvelope envelope)
+    public async Task ValidateEvent(EventEnvelope envelope, IReadOnlyList<EventEnvelope>? previousEnvelopes = null)
     {
         var unitOfWork = _serviceProvider.GetUserUnitOfWork(envelope.By ?? SystemUser.SystemUserGuid);
 
-        var versionedUnitOfWork = unitOfWork.WithPrefilledVersion(envelope.Version - 1);
+        var willBePublishedEnvelopes = previousEnvelopes ?? new List<EventEnvelope>();
+
+        var versionedUnitOfWork = unitOfWork.WithPrefilledVersion(envelope.Version - willBePublishedEnvelopes.Count - 1);
+        foreach (var willBePublishedEnvelope in willBePublishedEnvelopes)
+        {
+            versionedUnitOfWork.PublishedEventEnvelopes.Enqueue(willBePublishedEnvelope);
+        }
+
         var type = envelope.Event.GetType();
 
         if (!_eventValidators.ContainsKey(type)) return;