Merge pull request #1397 from Cratis/fix/rehydration

Fix/rehydration

Integration/Orleans.InProcess/AggregateRoots/Concepts/UserId.cs

@@ -0,0 +1,10 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Concepts;
+
+public record UserId(Guid Value) : ConceptAs<Guid>(Value)
+{
+    public static implicit operator Guid(UserId value) => value.Value;
+    public static implicit operator UserId(Guid value) => new(value);
+}

Integration/Orleans.InProcess/AggregateRoots/Concepts/UserName.cs

@@ -0,0 +1,10 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Concepts;
+
+public record UserName(string Value) : ConceptAs<string>(Value)
+{
+    public static implicit operator string(UserName value) => value.Value;
+    public static implicit operator UserName(string value) => new(value);
+}

Integration/Orleans.InProcess/AggregateRoots/Domain.Interfaces/IUser.cs

@@ -0,0 +1,15 @@
+using Cratis.Chronicle.Aggregates;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Concepts;
+using IAggregateRoot = Cratis.Chronicle.Orleans.Aggregates.IAggregateRoot;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Domain.Interfaces;
+
+public record UserInternalState(StateProperty<UserName> Name, StateProperty<bool> Deleted);
+
+public interface IUser : IIntegrationTestAggregateRoot<UserInternalState>
+{
+    Task Onboard(UserName name);
+    Task Delete();
+    Task ChangeUserName(UserName newName);
+    Task<bool> Exists();
+}

Integration/Orleans.InProcess/AggregateRoots/Domain/User.cs

@@ -0,0 +1,42 @@
+using Cratis.Chronicle.Aggregates;
+using Cratis.Chronicle.Concepts.Events;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Concepts;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Domain.Interfaces;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Events;
+using AggregateRoot = Cratis.Chronicle.Orleans.Aggregates.AggregateRoot;
+using IAggregateRootFactory = Cratis.Chronicle.Orleans.Aggregates.IAggregateRootFactory;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Domain;
+
+public class User(IAggregateRootFactory aggregateRootFactory) : AggregateRoot, IUser
+{
+    public class UserDeleted(EventSourceId id) : InvalidOperationException($"User {id} is deleted");
+
+    public StateProperty<UserName> Name = StateProperty<UserName>.Empty;
+    public StateProperty<bool> Deleted = StateProperty<bool>.Empty;
+
+    public Task Onboard(UserName name) => ApplyIfNotDeleted(new UserOnBoarded(name));
+    public Task Delete() => ApplyIfNotDeleted(new Events.UserDeleted());
+    public Task ChangeUserName(UserName newName) =>
+        Name.Value == newName ? Task.CompletedTask : ApplyIfNotDeleted(new UserNameChanged(newName));
+
+    public Task<bool> Exists() => Task.FromResult(!IsNew && !Deleted.Value);
+    public Task<CorrelationId> GetCorrelationId() => Task.FromResult(Context!.UnitOfWOrk.CorrelationId);
+    public Task<bool> GetIsNew() => Task.FromResult(IsNew);
+
+    Task ApplyIfNotDeleted(object evt) => Deleted.Value ? throw new UserDeleted(IdentityString) : Apply(evt);
+
+    public Task On(UserOnBoarded evt)
+    {
+        Name = Name.New(evt.Name);
+        return Task.CompletedTask;
+    }
+
+    public Task On(UserDeleted evt)
+    {
+        Deleted = Deleted.New(true);
+        return Task.CompletedTask;
+    }
+
+    public Task<UserInternalState> GetState() => Task.FromResult(new UserInternalState(Name, Deleted));
+}

Integration/Orleans.InProcess/AggregateRoots/EventAndEventSourceId.cs

@@ -0,0 +1,8 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using Cratis.Chronicle.Events;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots;
+
+public record EventAndEventSourceId(EventSourceId EventSourceId, object Event);

Integration/Orleans.InProcess/AggregateRoots/Events/UserDeleted.cs

@@ -0,0 +1,6 @@
+using Cratis.Chronicle.Events;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Events;
+
+[EventType]
+public record UserDeleted;

Integration/Orleans.InProcess/AggregateRoots/Events/UserNameChanged.cs

@@ -0,0 +1,9 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using Cratis.Chronicle.Events;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Concepts;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Events;
+[EventType]
+public record UserNameChanged(UserName NewName);

Integration/Orleans.InProcess/AggregateRoots/Events/UserOnboarded.cs

@@ -0,0 +1,7 @@
+using Cratis.Chronicle.Events;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Concepts;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Events;
+
+[EventType]
+public record UserOnBoarded(UserName Name);

Integration/Orleans.InProcess/AggregateRoots/IIntegrationTestAggregateRoot.cs

@@ -0,0 +1,14 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using Cratis.Chronicle.Orleans.Aggregates;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots;
+
+public interface IIntegrationTestAggregateRoot<TInternalState> : IAggregateRoot 
+    where TInternalState : class
+{
+    Task<TInternalState> GetState();
+    Task<CorrelationId> GetCorrelationId();
+    Task<bool> GetIsNew();
+}

Integration/Orleans.InProcess/AggregateRoots/Scenarios/given/context_for_aggregate_root.cs

@@ -0,0 +1,62 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using Cratis.Chronicle.Events;
+using Cratis.Chronicle.Integration.Base;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Domain;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Domain.Interfaces;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Events;
+using Cratis.Chronicle.Orleans.Aggregates;
+using Cratis.Chronicle.Transactions;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Scenarios.given;
+
+public class context_for_aggregate_root<TAggregate, TInternalState>(GlobalFixture globalFixture) : IntegrationSpecificationContext(globalFixture)
+    where TAggregate : IIntegrationTestAggregateRoot<TInternalState>
+    where TInternalState : class
+{
+#pragma warning disable CA2213 // Disposable fields should be disposed
+    protected GlobalFixture _globalFixture = globalFixture;
+#pragma warning restore CA2213 // Disposable fields should be disposed
+
+    public TInternalState ResultState;
+    public bool IsNew;
+    public IUnitOfWork UnitOfWork;
+
+    public override IEnumerable<Type> AggregateRoots => [typeof(User)];
+    public override IEnumerable<Type> EventTypes => [typeof(UserOnBoarded), typeof(UserDeleted), typeof(UserNameChanged)];
+
+    protected List<EventAndEventSourceId> EventsWithEventSourceIdToAppend = [];
+    protected IAggregateRootFactory AggregateRootFactory => Services.GetRequiredService<IAggregateRootFactory>();
+    protected IUnitOfWorkManager UnitOfWorkManager => Services.GetRequiredService<IUnitOfWorkManager>();
+
+    protected override void ConfigureServices(IServiceCollection services)
+    {
+    }
+
+    protected async Task DoOnAggregate(EventSourceId eventSourceId, Func<TAggregate, Task> action, bool commitUnitOfWork = true)
+    {
+        var user = await AggregateRootFactory.Get<TAggregate>(eventSourceId);
+        IsNew = await user.GetIsNew();
+        await action(user);
+        var correlationId = await user.GetCorrelationId();
+        ResultState = await user.GetState();
+        UnitOfWorkManager.TryGetFor(correlationId, out UnitOfWork);
+        if (commitUnitOfWork)
+        {
+            await UnitOfWork!.Commit();
+        }
+    }
+
+    void Establish()
+    {
+    }
+
+    async Task Because()
+    {
+        foreach (var @event in EventsWithEventSourceIdToAppend)
+        {
+            await EventStore.EventLog.Append(@event.EventSourceId, @event.Event);
+        }
+    }
+}

Integration/Orleans.InProcess/AggregateRoots/Scenarios/when_there_are_events_to_rehydrate/and_performing_no_action_on_aggregate.cs

@@ -0,0 +1,53 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using Cratis.Chronicle.Aggregates;
+using Cratis.Chronicle.Events;
+using Cratis.Chronicle.Integration.Base;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Concepts;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Domain;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Domain.Interfaces;
+using Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Events;
+using Cratis.Chronicle.Transactions;
+using context = Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Scenarios.when_there_are_events_to_rehydrate.and_performing_no_action_on_aggregate.context;
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots.Scenarios.when_there_are_events_to_rehydrate;
+
+[Collection(GlobalCollection.Name)]
+public class and_performing_no_action_on_aggregate(context context) : Given<context>(context)
+{
+    public class context(GlobalFixture globalFixture) : given.context_for_aggregate_root<IUser, UserInternalState>(globalFixture)
+    {
+        UserId _userId;
+        public UserName UserName;
+
+        public bool UserExists;
+
+        void Establish()
+        {
+            _userId = Guid.NewGuid();
+            UserName = "some name";
+            EventsWithEventSourceIdToAppend.Add(new EventAndEventSourceId(_userId.Value, new UserOnBoarded(UserName)));
+        }
+
+        Task Because() => DoOnAggregate(_userId.Value, async user => UserExists = await user.Exists());
+    }
+
+    [Fact]
+    void should_not_be_new_aggregate() => Context.IsNew.ShouldBeFalse();
+
+    [Fact]
+    void should_return_that_user_exists() => Context.UserExists.ShouldBeTrue();
+
+    [Fact]
+    void should_commit_unit_of_work_successfully() => Context.UnitOfWork.IsSuccess.ShouldBeTrue();
+
+    [Fact]
+    void should_not_commit_any_events() => Context.UnitOfWork.GetEvents().ShouldBeEmpty();
+
+    [Fact]
+    void should_not_be_deleted() => Context.ResultState.Deleted.ShouldEqual(new StateProperty<bool>(false, 0));
+
+    [Fact]
+    void should_have_assigned_the_username_once() => Context.ResultState.Name.ShouldEqual(new StateProperty<UserName>(Context.UserName, 1));
+}

Integration/Orleans.InProcess/AggregateRoots/StateProperty.cs

@@ -0,0 +1,11 @@
+
+namespace Cratis.Chronicle.Integration.Orleans.InProcess.AggregateRoots;
+
+public record StateProperty<TValue>(TValue Value, int CallCount)
+{
+    public static StateProperty<TValue> Empty => new(default, 0);
+
+    public StateProperty<TValue> New(TValue value) => IncrementCallCount() with { Value = value };
+
+    public StateProperty<TValue> IncrementCallCount() => this with { CallCount = CallCount + 1 };
+}

Integration/Orleans.InProcess/Orleans.InProcess.csproj

@@ -18,6 +18,9 @@
     </ItemGroup>
 
     <ItemGroup>
+      <Folder Include="AggregateRoots\Domain\" />
+      <Folder Include="AggregateRoots\Scenarios\given\" />
+      <Folder Include="AggregateRoots\Scenarios\when_there_are_events_to_rehydrate\" />
       <Folder Include="for_Observers\when_appending_event\" />
     </ItemGroup>
 

Source/Clients/AspNetCore/Transactions/UnitOfWorkActionFilter.cs

@@ -4,22 +4,32 @@
 using Cratis.Chronicle.Events.Constraints;
 using Cratis.Chronicle.Transactions;
 using Microsoft.AspNetCore.Mvc.Filters;
+using Microsoft.Extensions.Logging;
 
 namespace Cratis.Chronicle.AspNetCore.Transactions;
 
 /// <summary>
 /// Represents an implementation of <see cref="IAsyncActionFilter"/> for managing units of work and errors.
 /// </summary>
 /// <param name="unitOfWorkManager">The <see cref="IUnitOfWorkManager"/> to use.</param>
-public class UnitOfWorkActionFilter(IUnitOfWorkManager unitOfWorkManager) : IAsyncActionFilter
+/// <param name="logger">The <see cref="ILogger{TCategoryName}"/>.</param>
+public class UnitOfWorkActionFilter(IUnitOfWorkManager unitOfWorkManager, ILogger<UnitOfWorkActionFilter> logger) : IAsyncActionFilter
 {
     /// <inheritdoc/>
     public async Task OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next)
     {
         await next();
 
         var unitOfWork = unitOfWorkManager.Current;
-        await unitOfWork.Commit();
+
+        if (!unitOfWork.IsCompleted)
+        {
+            await unitOfWork.Commit();
+        }
+        else
+        {
+            logger.AlreadyCompletedManually(unitOfWork.CorrelationId);
+        }
         if (!unitOfWork.IsSuccess)
         {
             foreach (var violation in unitOfWork.GetConstraintViolations())
@@ -29,4 +39,4 @@ public async Task OnActionExecutionAsync(ActionExecutingContext context, ActionE
             }
         }
     }
-}
+}

Source/Clients/AspNetCore/Transactions/UnitOfWorkActionFilterLogging.cs

@@ -0,0 +1,13 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using Cratis.Execution;
+using Microsoft.Extensions.Logging;
+
+namespace Cratis.Chronicle.AspNetCore.Transactions;
+
+internal static partial class UnitOfWorkActionFilterLogging
+{
+    [LoggerMessage(LogLevel.Warning, "The unit of work for correlation {CorrelationId} has already been completed manually either by Commit or Rollback. It is recommended to not complete the unit of work manually in this context.")]
+    internal static partial void AlreadyCompletedManually(this ILogger<UnitOfWorkActionFilter> logger, CorrelationId correlationId);
+}

Source/Clients/AspNetCore/Transactions/UnitOfWorkMiddleware.cs

@@ -21,7 +21,18 @@ public class UnitOfWorkMiddleware(IUnitOfWorkManager unitOfWorkManager, RequestD
     /// <returns>Awaitable task.</returns>
     public async Task InvokeAsync(HttpContext context)
     {
-        unitOfWorkManager.Begin(CorrelationId.New());
-        await next(context);
+        var unitOfWork = unitOfWorkManager.Begin(CorrelationId.New());
+        try
+        {
+            await next(context);
+        }
+        catch
+        {
+            if (!unitOfWork.IsCompleted)
+            {
+                unitOfWork.Dispose();
+            }
+            throw;
+        }
     }
 }

Source/Clients/DotNET.Specs/Aggregates/for_AggregateRoot/given/a_stateful_aggregate_root.cs

@@ -21,7 +21,7 @@ void Establish()
 
         _unitOfWork = Substitute.For<IUnitOfWork>();
 
-        _aggregateRootContext = new AggregateRootContext(_eventSourceId, _eventSequence, _aggregateRoot, _unitOfWork);
+        _aggregateRootContext = new AggregateRootContext(_eventSourceId, _eventSequence, _aggregateRoot, _unitOfWork, EventSequenceNumber.First);
         _aggregateRoot._context = _aggregateRootContext;
         _aggregateRoot._mutation = _mutation;
     }

Source/Clients/DotNET.Specs/Aggregates/for_AggregateRoot/given/a_stateless_aggregate_root.cs

@@ -18,7 +18,7 @@ void Establish()
         _aggregateRoot = new();
         _eventSourceId = Guid.NewGuid().ToString();
         _unitOfWork = Substitute.For<IUnitOfWork>();
-        _aggregateRootContext = new AggregateRootContext(_eventSourceId, _eventSequence, _aggregateRoot, _unitOfWork);
+        _aggregateRootContext = new AggregateRootContext(_eventSourceId, _eventSequence, _aggregateRoot, _unitOfWork, EventSequenceNumber.First);
 
         _aggregateRoot._context = _aggregateRootContext;
         _aggregateRoot._mutation = _mutation;

Source/Clients/DotNET.Specs/Aggregates/for_AggregateRootMutatorFactory/when_creating_for_stateful_aggregate_root.cs

@@ -16,7 +16,8 @@ public class when_creating_for_stateful_aggregate_root : given.an_aggregate_root
         EventSourceId.New(),
         Substitute.For<IEventSequence>(),
         new StatefulAggregateRoot(),
-        Substitute.For<IUnitOfWork>());
+        Substitute.For<IUnitOfWork>(),
+        EventSequenceNumber.First);
 
     async Task Because() => _result = await _factory.Create<StatefulAggregateRoot>(_context);
 

Source/Clients/DotNET.Specs/Aggregates/for_AggregateRootMutatorFactory/when_creating_for_stateless_aggregate_root.cs

@@ -16,7 +16,8 @@ public class when_creating_for_stateless_aggregate_root : given.an_aggregate_roo
         EventSourceId.New(),
         Substitute.For<IEventSequence>(),
         new StatelessAggregateRoot(),
-        Substitute.For<IUnitOfWork>());
+        Substitute.For<IUnitOfWork>(),
+        EventSequenceNumber.First);
 
     async Task Because() => _result = await _factory.Create<StatelessAggregateRoot>(_context);