Extend CommandContext with the streams current StreamVersion

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ###  Added
 
 -   Implement `IEventStoreManagementClient.DeleteStreamAsync` using the newly released `DeleteAllItemsByPartitionKeyStreamAsync` method in the Cosmos SDK.
+-   Extend `CommandContext` with the current `StreamVersion` of the stream.
 
 ## [1.13.3] - 2024-04-21
 

src/Atc.Cosmos.EventStore.Cqrs/Commands/CommandContext.cs

@@ -5,10 +5,17 @@ namespace Atc.Cosmos.EventStore.Cqrs.Commands;
 
 internal class CommandContext : ICommandContext, ICommandContextInspector
 {
+    public StreamVersion StreamVersion { get; }
+
     public const int EventLimit = 10;
 
     private readonly List<object> appliedEvents = new();
 
+    public CommandContext(StreamVersion streamVersion)
+    {
+        StreamVersion = streamVersion;
+    }
+
     public IReadOnlyCollection<object> Events
         => appliedEvents;
 

src/Atc.Cosmos.EventStore.Cqrs/Commands/CommandProcessor.cs

@@ -60,7 +60,7 @@ private async ValueTask<CommandResult> SafeExecuteAsync(
                 .ConfigureAwait(false);
 
             // Execute command on aggregate.
-            var context = new CommandContext();
+            var context = new CommandContext(state.Version);
             await handler
                 .ExecuteAsync(command, context, cancellationToken)
                 .ConfigureAwait(false);

src/Atc.Cosmos.EventStore.Cqrs/ICommandContext.cs

@@ -2,6 +2,8 @@ namespace Atc.Cosmos.EventStore.Cqrs;
 
 public interface ICommandContext
 {
+    StreamVersion StreamVersion { get; }
+
     void AddEvent(object evt);
 
     object? ResponseObject { get; set; }

src/Atc.Cosmos.EventStore.Cqrs/Testing/CommandHandlerTester.cs

@@ -125,7 +125,7 @@ await handlerMetadata
                 .ConfigureAwait(false);
         }
 
-        var context = new CommandContext();
+        var context = new CommandContext(version);
 
         await handler
             .ExecuteAsync(command!, context, cancellationToken)

src/Atc.Cosmos.EventStore.Cqrs/Testing/ICommandContextInspector.cs

@@ -2,6 +2,8 @@
 
 public interface ICommandContextInspector
 {
+    StreamVersion StreamVersion { get; }
+
     IReadOnlyCollection<object> Events { get; }
 
     object? ResponseObject { get; }

test/Atc.Cosmos.EventStore.Cqrs.Tests/Commands/CommandProcessorTests.cs

@@ -0,0 +1,180 @@
+using System.Collections.ObjectModel;
+using Atc.Cosmos.EventStore.Cqrs.Commands;
+using Atc.Cosmos.EventStore.Cqrs.Tests.Mocks;
+using Atc.Test;
+using AutoFixture.Xunit2;
+using FluentAssertions;
+using NSubstitute;
+using Xunit;
+
+namespace Atc.Cosmos.EventStore.Cqrs.Tests.Commands;
+
+public class CommandProcessorTests
+{
+    [Theory, AutoNSubstituteData]
+    internal async Task Should_Exeute_State_Projector(
+        [Frozen] ICommandHandlerFactory commandHandlerFactory,
+        [Frozen] IStateProjector<MockCommand> stateProjector,
+        CommandProcessor<MockCommand> sut,
+        MockCommand command,
+        ICommandHandler<MockCommand> handler,
+        Atc.Cosmos.EventStore.Cqrs.Commands.StreamState streamState,
+        CancellationToken cancellationToken)
+    {
+        commandHandlerFactory.Create<MockCommand>().Returns(handler);
+        stateProjector.ProjectAsync(command, handler, cancellationToken).Returns(streamState);
+
+        await sut.ExecuteAsync(command, cancellationToken);
+
+        await stateProjector.Received(1).ProjectAsync(command, handler, cancellationToken);
+    }
+
+    [Theory, AutoNSubstituteData]
+    internal async Task Should_Execute_Command(
+        [Frozen] ICommandHandlerFactory commandHandlerFactory,
+        [Frozen] IStateProjector<MockCommand> stateProjector,
+        CommandProcessor<MockCommand> sut,
+        MockCommand command,
+        ICommandHandler<MockCommand> handler,
+        Atc.Cosmos.EventStore.Cqrs.Commands.StreamState streamState,
+        CancellationToken cancellationToken)
+    {
+        commandHandlerFactory.Create<MockCommand>().Returns(handler);
+        stateProjector.ProjectAsync(command, handler, cancellationToken).Returns(streamState);
+
+        await sut.ExecuteAsync(command, cancellationToken);
+
+        await handler.Received(1).ExecuteAsync(command, Arg.Any<CommandContext>(), cancellationToken);
+    }
+
+    [Theory, AutoNSubstituteData]
+    internal async Task Should_Set_Command_Context_StreamVersion(
+        [Frozen] ICommandHandlerFactory commandHandlerFactory,
+        [Frozen] IStateProjector<MockCommand> stateProjector,
+        CommandProcessor<MockCommand> sut,
+        MockCommand command,
+        ICommandHandler<MockCommand> handler,
+        Atc.Cosmos.EventStore.Cqrs.Commands.StreamState streamState,
+        CancellationToken cancellationToken)
+    {
+        commandHandlerFactory.Create<MockCommand>().Returns(handler);
+        stateProjector.ProjectAsync(command, handler, cancellationToken).Returns(streamState);
+
+        await sut.ExecuteAsync(command, cancellationToken);
+
+        var commandContext = handler.ReceivedCallWithArgument<CommandContext>();
+        commandContext.StreamVersion.Should().Be(streamState.Version);
+    }
+
+    [Theory, AutoNSubstituteData]
+    internal async Task Should_Return_NotModified_When_Command_Emits_No_Events(
+        [Frozen] ICommandHandlerFactory commandHandlerFactory,
+        [Frozen] IStateProjector<MockCommand> stateProjector,
+        CommandProcessor<MockCommand> sut,
+        MockCommand command,
+        MockCommandHandler handler,
+        Atc.Cosmos.EventStore.Cqrs.Commands.StreamState streamState,
+        CancellationToken cancellationToken)
+    {
+        commandHandlerFactory.Create<MockCommand>().Returns(handler);
+        stateProjector.ProjectAsync(command, handler, cancellationToken).Returns(streamState);
+
+        var result = await sut.ExecuteAsync(command, cancellationToken);
+
+        result.Result.Should().Be(ResultType.NotModified);
+    }
+
+    [Theory, AutoNSubstituteData]
+    internal async Task Should_Write_ResposeObject_To_CommandResult_When_Command_Emits_No_Events(
+        [Frozen] ICommandHandlerFactory commandHandlerFactory,
+        [Frozen] IStateProjector<MockCommand> stateProjector,
+        CommandProcessor<MockCommand> sut,
+        MockCommand command,
+        MockCommandHandler handler,
+        object responseObject,
+        Atc.Cosmos.EventStore.Cqrs.Commands.StreamState streamState,
+        CancellationToken cancellationToken)
+    {
+        commandHandlerFactory.Create<MockCommand>().Returns(handler);
+        stateProjector.ProjectAsync(command, handler, cancellationToken).Returns(streamState);
+        handler.ResponseObject = responseObject;
+
+        var result = await sut.ExecuteAsync(command, cancellationToken);
+
+        result.Response.Should().Be(responseObject);
+    }
+
+    [Theory, AutoNSubstituteData]
+    internal async Task Should_Call_StateWriter_With_Events__When_Command_Emits_Events(
+        [Frozen] ICommandHandlerFactory commandHandlerFactory,
+        [Frozen] IStateProjector<MockCommand> stateProjector,
+        [Frozen] IStateWriter<MockCommand> stateWriter,
+        CommandProcessor<MockCommand> sut,
+        MockCommand command,
+        CommandResult commandResult,
+        MockCommandHandler handler,
+        MockEvent[] events,
+        Atc.Cosmos.EventStore.Cqrs.Commands.StreamState streamState,
+        CancellationToken cancellationToken)
+    {
+        commandHandlerFactory.Create<MockCommand>().Returns(handler);
+        stateProjector.ProjectAsync(command, handler, cancellationToken).Returns(streamState);
+        stateWriter.WriteEventAsync(command, events, cancellationToken).ReturnsForAnyArgs(commandResult);
+        handler.AddEventsToEmit(events);
+
+        await sut.ExecuteAsync(command, cancellationToken);
+
+        await stateWriter.Received(1).WriteEventAsync(command, Arg.Any<IReadOnlyCollection<object>>(), cancellationToken);
+        var writtenEvents = stateWriter.ReceivedCallWithArgument<IReadOnlyCollection<object>>();
+        writtenEvents.Should().HaveSameCount(events);
+        writtenEvents.AsEnumerable().Should().BeEquivalentTo(events);
+    }
+
+    [Theory, AutoNSubstituteData]
+    internal async Task Should_Return_Changed_When_Command_Emits_Events(
+        [Frozen] ICommandHandlerFactory commandHandlerFactory,
+        [Frozen] IStateProjector<MockCommand> stateProjector,
+        [Frozen] IStateWriter<MockCommand> stateWriter,
+        CommandProcessor<MockCommand> sut,
+        MockCommand command,
+        MockCommandHandler handler,
+        CommandResult commandResult,
+        MockEvent[] events,
+        Atc.Cosmos.EventStore.Cqrs.Commands.StreamState streamState,
+        CancellationToken cancellationToken)
+    {
+        commandHandlerFactory.Create<MockCommand>().Returns(handler);
+        stateProjector.ProjectAsync(command, handler, cancellationToken).Returns(streamState);
+        stateWriter.WriteEventAsync(command, events, cancellationToken).ReturnsForAnyArgs(commandResult);
+        handler.AddEventsToEmit(events);
+
+        var result = await sut.ExecuteAsync(command, cancellationToken);
+
+        result.Result.Should().Be(ResultType.Changed);
+    }
+
+    [Theory, AutoNSubstituteData]
+    internal async Task Should_Write_ResposeObject_To_CommandResult_When_Command_Emits_Events(
+        [Frozen] ICommandHandlerFactory commandHandlerFactory,
+        [Frozen] IStateProjector<MockCommand> stateProjector,
+        [Frozen] IStateWriter<MockCommand> stateWriter,
+        CommandProcessor<MockCommand> sut,
+        MockCommand command,
+        MockCommandHandler handler,
+        CommandResult commandResult,
+        MockEvent[] events,
+        object responseObject,
+        Atc.Cosmos.EventStore.Cqrs.Commands.StreamState streamState,
+        CancellationToken cancellationToken)
+    {
+        commandHandlerFactory.Create<MockCommand>().Returns(handler);
+        stateProjector.ProjectAsync(command, handler, cancellationToken).Returns(streamState);
+        stateWriter.WriteEventAsync(command, events, cancellationToken).ReturnsForAnyArgs(commandResult);
+        handler.AddEventsToEmit(events);
+        handler.ResponseObject = responseObject;
+
+        var result = await sut.ExecuteAsync(command, cancellationToken);
+
+        result.Response.Should().Be(responseObject);
+    }
+}

test/Atc.Cosmos.EventStore.Cqrs.Tests/Mocks/MockCommandHandler.cs

@@ -0,0 +1,28 @@
+namespace Atc.Cosmos.EventStore.Cqrs.Tests.Mocks;
+
+public class MockCommandHandler : ICommandHandler<MockCommand>
+{
+    private List<IEvent> events = new();
+
+    public object ResponseObject { get; set; } = null;
+
+    public void AddEventsToEmit(params IEvent[] eventsToEmit)
+    {
+        events = events.Concat(eventsToEmit).ToList();
+    }
+
+    public ValueTask ExecuteAsync(
+        MockCommand command,
+        ICommandContext context,
+        CancellationToken cancellationToken)
+    {
+        foreach (var evt in events)
+        {
+            context.AddEvent(evt);
+        }
+
+        context.ResponseObject = ResponseObject;
+
+        return default;
+    }
+}