Release of new minor version v1.20

CHANGELOG.md

@@ -23,6 +23,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [1.17.0] - 2025-03-21
 
+### Changed
+
+- Changed `ICommandProcessorFactory`, `ICommandHandlerFactory` and `CommandProcessor<>` to be 
+ registered as transient rather than singleton. This allows `CommandHandler` implementations to use 
+ dependencies registered as scoped.
+
+- Reintroduce `IProjectionFactory` in a slightly modified version to allow consumers to make additional "initialization" of projections.
+
+## [1.17.0] - 2025-03-21
+
 ### Fixed
 
 -   Ensure that a projection can run in its own scope. This is important as we do not want state to leak from one projection to the other.

Directory.Build.props

@@ -42,7 +42,7 @@
   <ItemGroup Label="Code Analyzers">
     <PackageReference Include="SecurityCodeScan.VS2019" Version="5.6.7" PrivateAssets="all" />
     <PackageReference Include="StyleCop.Analyzers" Version="1.2.0-beta.321" PrivateAssets="All" />
-    <PackageReference Include="SonarAnalyzer.CSharp" Version="9.0.0.68202" PrivateAssets="all" />
+    <PackageReference Include="SonarAnalyzer.CSharp" Version="10.7.0.110445" PrivateAssets="all" />
   </ItemGroup>
 
 </Project>

src/Atc.Cosmos.EventStore.Cqrs/DependencyInjection/EventStoreOptionsBuilderExtensions.cs

@@ -28,6 +28,7 @@ public static EventStoreOptionsBuilder UseCQRS(
         builder.Services.AddSingleton<IProjectionOptionsFactory, ProjectionOptionsFactory>();
 
         builder.Services.AddSingleton(typeof(ProjectionMetadata<>), typeof(ProjectionMetadata<>));
+        builder.Services.AddTransient<IProjectionFactory, DefaultProjectionFactory>();
 
         builder.Services.TryAddSingleton<IProjectionTelemetry, ProjectionTelemetry>();
 

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

@@ -0,0 +1,40 @@
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Atc.Cosmos.EventStore.Cqrs;
+
+/// <summary>
+/// Responsible for creating <see cref="IProjection"/> instances.
+/// </summary>
+public interface IProjectionFactory
+{
+    /// <summary>
+    /// Creates a projection of type <typeparamref name="TProjection"/> for the event stream
+    /// identified by <paramref name="streamId"/>.
+    /// </summary>
+    /// <param name="streamId">ID of the stream being projected.</param>
+    /// <param name="cancellationToken">Cancellation.</param>
+    /// <typeparam name="TProjection">Type of projection to create.</typeparam>
+    /// <returns>The created projection.</returns>
+    public Task<IProjection> CreateAsync<TProjection>(EventStreamId streamId, CancellationToken cancellationToken)
+        where TProjection : IProjection;
+}
+
+/// <summary>
+/// The default projection factory which just creates projections by
+/// getting them from the DI-container.
+/// </summary>
+internal sealed class DefaultProjectionFactory : IProjectionFactory
+{
+    private readonly IServiceProvider serviceProvider;
+
+    public DefaultProjectionFactory(IServiceProvider serviceProvider)
+    {
+        this.serviceProvider = serviceProvider;
+    }
+
+    public Task<IProjection> CreateAsync<TProjection>(EventStreamId streamId, CancellationToken cancellationToken)
+        where TProjection : IProjection
+    {
+        return Task.FromResult<IProjection>(serviceProvider.GetRequiredService<TProjection>());
+    }
+}

src/Atc.Cosmos.EventStore.Cqrs/Projections/ProjectionFilter.cs

@@ -15,11 +15,7 @@ public ProjectionFilter(string filter)
                 ? CreateEvaluateAll()
                 : CreateEvaluation(p))
             .ToArray();
-        endsOnAcceptAll = filter
-            .Split(
-                new[] { EventStreamId.PartSeperator },
-                StringSplitOptions.RemoveEmptyEntries)
-            .Last() == "**";
+        endsOnAcceptAll = filter.EndsWith("**");
     }
 
     public bool Evaluate(StreamId streamId)

src/Atc.Cosmos.EventStore.Cqrs/Projections/ProjectionProcessor.cs

@@ -10,18 +10,18 @@ internal class ProjectionProcessor<TProjection> : IProjectionProcessor<TProjecti
     private readonly IReadOnlyCollection<ProjectionFilter> filters;
     private readonly IProjectionTelemetry telemetry;
     private readonly ProjectionMetadata<TProjection> projectionMetadata;
-    private readonly IServiceProvider serviceProvider;
+    private readonly IServiceScopeFactory serviceScopeFactory;
     private readonly string projectionName;
 
     public ProjectionProcessor(
         IProjectionOptionsFactory optionsFactory,
         IProjectionTelemetry telemetry,
         ProjectionMetadata<TProjection> projectionMetadata,
-        IServiceProvider serviceProvider)
+        IServiceScopeFactory serviceScopeFactory)
     {
         this.telemetry = telemetry;
         this.projectionMetadata = projectionMetadata;
-        this.serviceProvider = serviceProvider;
+        this.serviceScopeFactory = serviceScopeFactory;
         filters = optionsFactory
             .GetOptions<TProjection>()
             .Filters;
@@ -49,10 +49,6 @@ public async Task<ProjectionAction> ProcessBatchAsync(
 
         foreach (var events in groupedEvents)
         {
-            await using var scope = serviceProvider.CreateAsyncScope();
-
-            var projection = scope.ServiceProvider.GetRequiredService<TProjection>();
-
             using var operation = batchTelemetry.StartProjection(events.Key);
 
             if (!projectionMetadata.CanConsumeOneOrMoreEvents(events))
@@ -62,11 +58,16 @@ public async Task<ProjectionAction> ProcessBatchAsync(
                 continue;
             }
 
+            var eventStreamId = EventStreamId.FromStreamId(events.Key);
+            await using var scope = serviceScopeFactory.CreateAsyncScope();
+            var projectionFactory = scope.ServiceProvider.GetRequiredService<IProjectionFactory>();
+            var projection = await projectionFactory.CreateAsync<TProjection>(eventStreamId, cancellationToken);
+
             try
             {
                 await projection
                     .InitializeAsync(
-                        events.Key,
+                        eventStreamId,
                         cancellationToken)
                     .ConfigureAwait(false);
 

src/Atc.Cosmos.EventStore/InMemory/InMemoryStore.cs

@@ -84,8 +84,9 @@ public Task WriteAsync(
             .GetOrAdd(streamId, new ConcurrentDictionary<string, CheckpointDocument>())
             .AddOrUpdate(
                 name,
-                key => new CheckpointDocument(name, streamId, streamVersion, dateTimeProvider.GetDateTime(), state),
-                (key, doc) => new CheckpointDocument(name, streamId, streamVersion, dateTimeProvider.GetDateTime(), state));
+                static (key, arg) => new CheckpointDocument(key, arg.streamId, arg.streamVersion, arg.currentTime, arg.state),
+                static (key, doc, arg) => new CheckpointDocument(key, arg.streamId, arg.streamVersion, arg.currentTime, arg),
+                (streamId, streamVersion, state, currentTime: dateTimeProvider.GetDateTime()));
 
         return Task.CompletedTask;
     }

test/Atc.Cosmos.EventStore.Cqrs.Tests/Atc.Cosmos.EventStore.Cqrs.Tests.csproj

@@ -3,11 +3,13 @@
   <PropertyGroup>
     <TargetFramework>net9.0</TargetFramework>
     <IsPackable>false</IsPackable>
+
   </PropertyGroup>
 
   <ItemGroup>
     <PackageReference Include="Atc.Test" Version="1.1.4" />
     <PackageReference Include="FluentAssertions" Version="6.12.1" />
+    <PackageReference Include="Microsoft.AspNetCore.TestHost" Version="9.0.3" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.11.1" />
     <PackageReference Include="xunit" Version="2.9.2" />
     <PackageReference Include="xunit.runner.visualstudio" Version="2.8.2">
@@ -18,6 +20,9 @@
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
+
+    <!-- Get the shared Microsoft.AspNetCore.App framework instead of referencing a bunch of MS.Ext.Hosting etc packages -->
+    <FrameworkReference Include="Microsoft.AspNetCore.App"/>
   </ItemGroup>
 
   <ItemGroup>

test/Atc.Cosmos.EventStore.Cqrs.Tests/Functional/CommandHandlerTests.cs

@@ -0,0 +1,85 @@
+using Microsoft.Extensions.DependencyInjection;
+using Xunit;
+
+namespace Atc.Cosmos.EventStore.Cqrs.Tests.Functional;
+
+#nullable enable
+
+[Trait("Category", "Functional")]
+public class CommandHandlerTests : IAsyncLifetime
+{
+    private CqrsTestHost host = null!;
+
+    [Fact]
+    public async Task Result_from_CommandHandler_must_be_returned()
+    {
+        var tick = DateTime.UtcNow.Ticks;
+        var commandResult = await host.Services.GetRequiredService<ICommandProcessorFactory>()
+            .Create<MakeTimeTickCommand>()
+            .ExecuteAsync(new MakeTimeTickCommand(tick), CancellationToken.None);
+
+        Assert.NotNull(commandResult);
+        Assert.NotNull(commandResult.Response);
+        Assert.Equal(tick, commandResult.Response);
+    }
+
+    [Fact]
+    public async Task CommandHandler_can_consume_existing_events()
+    {
+        // Produce some events by making time tick
+        await MakeTimeTick(host);
+        await MakeTimeTick(host);
+        await MakeTimeTick(host);
+
+        // Query events
+        var result = await host.Services.GetRequiredService<ICommandProcessorFactory>()
+            .Create<QueryTimeTickCommand>()
+            .ExecuteAsync(new QueryTimeTickCommand(), CancellationToken.None);
+
+        var events = Assert.IsType<List<(TimeTickedEvent Evt, EventMetadata Metadata)>>(result.Response);
+        Assert.Equal(3, events.Count);
+
+        static async Task MakeTimeTick(CqrsTestHost host)
+        {
+            var tick = DateTime.UtcNow.Ticks;
+            var commandProcessorFactory = host.Services.GetRequiredService<ICommandProcessorFactory>();
+            _ = await commandProcessorFactory
+                .Create<MakeTimeTickCommand>()
+                .ExecuteAsync(new MakeTimeTickCommand(tick), CancellationToken.None);
+        }
+    }
+
+    [Fact]
+    public async Task CommandHandler_that_consumes_events_works_when_no_existing_events_are_present()
+    {
+        // Query events - none exists
+        var result = await host.Services.GetRequiredService<ICommandProcessorFactory>()
+            .Create<QueryTimeTickCommand>()
+            .ExecuteAsync(new QueryTimeTickCommand(), CancellationToken.None);
+
+        var events = Assert.IsType<List<(TimeTickedEvent Evt, EventMetadata Metadata)>>(result.Response);
+        Assert.Empty(events);
+    }
+
+    [Fact]
+    public async Task Command_that_uses_RequiredVersion_Exists_must_result_in_NotFound_when_no_existing_events_are_present()
+    {
+        // Query events - must fail as non exists
+        var result = await host.Services.GetRequiredService<ICommandProcessorFactory>()
+            .Create<QueryExistingTimeTickCommand>()
+            .ExecuteAsync(new QueryExistingTimeTickCommand(), CancellationToken.None);
+
+        Assert.Equal(ResultType.NotFound, result.Result);
+    }
+
+    public async Task InitializeAsync()
+    {
+        host = new CqrsTestHost();
+        await host.StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await host.DisposeAsync();
+    }
+}

test/Atc.Cosmos.EventStore.Cqrs.Tests/Functional/CqrsTestHost.cs

@@ -0,0 +1,73 @@
+using Atc.Cosmos.EventStore.Streams;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.TestHost;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+
+namespace Atc.Cosmos.EventStore.Cqrs.Tests.Functional;
+
+#nullable enable
+
+public class CqrsTestHost : IAsyncDisposable
+{
+    private readonly WebApplication host;
+
+    public CqrsTestHost()
+    {
+        host = CreateHostBuilder().Build();
+    }
+
+    public IServiceProvider Services => host.Services;
+
+    public async Task StartAsync()
+    {
+        await host.StartAsync();
+    }
+
+    public async Task StopAsync()
+    {
+        await host.StopAsync();
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await host.StopAsync();
+        await host.DisposeAsync();
+    }
+
+    private static WebApplicationBuilder CreateHostBuilder()
+    {
+        // Build a host with Atc EventStore
+        var webApplicationBuilder = WebApplication.CreateEmptyBuilder(new WebApplicationOptions()
+        {
+            ApplicationName = "Atc.Cosmos.EventStore.Cqrs.Tests",
+        });
+
+        webApplicationBuilder.WebHost.UseTestServer();
+
+        // Configure EventStore
+        webApplicationBuilder.Services.AddEventStore(eventStoreBuilder =>
+        {
+            eventStoreBuilder.UseEvents(c => c.FromAssembly<CqrsTestHost>());
+            eventStoreBuilder.UseCQRS(c =>
+            {
+                c.AddCommandsFromAssembly<CqrsTestHost>();
+                c.AddProjectionJob<TimeProjection>("TimeProjection");
+            });
+        });
+
+        // Use InMemoryEventStoreClient which actually works
+        webApplicationBuilder.Services.Replace(
+            ServiceDescriptor.Singleton<IEventStoreClient, InMemoryEventStoreClient>());
+
+        // Remove unused registrations
+        webApplicationBuilder.Services.RemoveAll<IStreamInfoReader>();
+        webApplicationBuilder.Services.RemoveAll<IStreamMetadataReader>();
+        webApplicationBuilder.Services.RemoveAll<IStreamReader>();
+        webApplicationBuilder.Services.RemoveAll<IStreamWriter>();
+
+        webApplicationBuilder.Services.AddSingleton<FakeDatabase>();
+
+        return webApplicationBuilder;
+    }
+}

test/Atc.Cosmos.EventStore.Cqrs.Tests/Functional/FakeDatabase.cs

@@ -0,0 +1,18 @@
+#nullable enable
+namespace Atc.Cosmos.EventStore.Cqrs.Tests.Functional;
+
+internal class FakeDatabase
+{
+    private readonly Dictionary<string, object> storage = new();
+
+    public void Save(string key, object value)
+    {
+        storage[key] = value;
+    }
+
+    public object? Load(string key)
+    {
+        storage.TryGetValue(key, out var value);
+        return value;
+    }
+}