Fixed the issue with Any expected state for func services

src/Core/src/Eventuous.Application/AggregateService/CommandHandlerBuilder.cs

@@ -1,6 +1,8 @@
 // Copyright (C) Ubiquitous AS.All rights reserved
 // Licensed under the Apache License, Version 2.0.
 
+using static Eventuous.CommandServiceDelegates;
+
 namespace Eventuous;
 
 public abstract class CommandHandlerBuilder<TAggregate, TState, TId>
@@ -12,7 +14,7 @@
 /// Builds a command handler for a specific command type. You would not need to instantiate this class directly,
 /// use <see cref="CommandService{TAggregate,TState,TId}.On{TCommand}" /> function.
 /// </summary>
 /// <param name="store">Default aggregate store instance for the command service</param>
 /// <typeparam name="TCommand">Command type</typeparam>
 /// <typeparam name="TAggregate">Aggregate type</typeparam>
 /// <typeparam name="TState">State of the aggregate type</typeparam>
@@ -90,7 +92,7 @@
     /// </summary>
     /// <param name="resolveStore"></param>
     /// <returns></returns>
     public CommandHandlerBuilder<TCommand, TAggregate, TState, TId> ResolveStore(ResolveStore<TCommand>? resolveStore) {
         _resolveStore = resolveStore;
 
         return this;

src/Core/src/Eventuous.Application/AggregateService/CommandHandlersMap.cs

@@ -2,28 +2,12 @@
 // Licensed under the Apache License, Version 2.0.
 
 using System.Reflection;
+using static Eventuous.CommandServiceDelegates;
 
 namespace Eventuous;
 
 using static Diagnostics.ApplicationEventSource;
 
-public delegate Task ActOnAggregateAsync<in TAggregate, in TCommand>(TAggregate aggregate, TCommand command, CancellationToken cancellationToken)
-    where TAggregate : Aggregate;
-
-public delegate void ActOnAggregate<in TAggregate, in TCommand>(TAggregate aggregate, TCommand command) where TAggregate : Aggregate;
-
-delegate ValueTask<T> HandleUntypedCommand<T>(T aggregate, object command, CancellationToken cancellationToken) where T : Aggregate;
-
-public delegate Task<TId> GetIdFromCommandAsync<TId, in TCommand>(TCommand command, CancellationToken cancellationToken) where TId : Id where TCommand : class;
-
-public delegate TId GetIdFromCommand<out TId, in TCommand>(TCommand command) where TId : Id where TCommand : class;
-
-delegate ValueTask<TId> GetIdFromUntypedCommand<TId>(object command, CancellationToken cancellationToken) where TId : Id;
-
-public delegate IAggregateStore ResolveStore<in TCommand>(TCommand command) where TCommand : class;
-
-delegate IAggregateStore ResolveStoreFromCommand(object command);
-
 record RegisteredHandler<T, TId>(
         ExpectedState                ExpectedState,
         GetIdFromUntypedCommand<TId> GetId,

src/Core/src/Eventuous.Application/AggregateService/CommandHandlingDelegateExtensions.cs

@@ -1,6 +1,8 @@
 // Copyright (C) Ubiquitous AS.All rights reserved
 // Licensed under the Apache License, Version 2.0.
 
+using static Eventuous.CommandServiceDelegates;
+
 namespace Eventuous;
 
 static class CommandHandlingDelegateExtensions {

src/Core/src/Eventuous.Application/AggregateService/CommandService.Async.cs

@@ -1,6 +1,8 @@
 // Copyright (C) Ubiquitous AS. All rights reserved
 // Licensed under the Apache License, Version 2.0.
 
+using static Eventuous.CommandServiceDelegates;
+
 namespace Eventuous;
 
 public abstract partial class CommandService<TAggregate, TState, TId> {
@@ -12,7 +14,7 @@
     /// <param name="resolveStore">Resolve aggregate store from the command</param>
     /// <typeparam name="TCommand">Command type</typeparam>
     [Obsolete("Use On<TCommand>().InState(ExpectedState.New).GetId(...).ActAsync(...).ResolveStore(...) instead")]
     protected void OnNewAsync<TCommand>(
             GetIdFromCommand<TId, TCommand>           getId,
             ActOnAggregateAsync<TAggregate, TCommand> action,
             ResolveStore<TCommand>?                   resolveStore = null

src/Core/src/Eventuous.Application/AggregateService/CommandService.Sync.cs

@@ -1,6 +1,8 @@
 // Copyright (C) Ubiquitous AS. All rights reserved
 // Licensed under the Apache License, Version 2.0.
 
+using static Eventuous.CommandServiceDelegates;
+
 namespace Eventuous;
 
 public abstract partial class CommandService<TAggregate, TState, TId> {
@@ -42,7 +44,7 @@
     /// <param name="resolveStore">Resolve aggregate store from the command</param>
     /// <typeparam name="TCommand">Command type</typeparam>
     [Obsolete("Use On<TCommand>().InState(ExpectedState.Any).GetId(...).Act(...).ResolveStore(...) instead")]
     protected void OnAny<TCommand>(
             GetIdFromCommand<TId, TCommand>      getId,
             ActOnAggregate<TAggregate, TCommand> action,
             ResolveStore<TCommand>?              resolveStore = null

src/Core/src/Eventuous.Application/AggregateService/CommandServiceDelegates.cs

@@ -0,0 +1,24 @@
+// Copyright (C) Ubiquitous AS.All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
+namespace Eventuous;
+
+public static class CommandServiceDelegates {
+    public delegate Task ActOnAggregateAsync<in TAggregate, in TCommand>(TAggregate aggregate, TCommand command, CancellationToken cancellationToken)
+        where TAggregate : Aggregate;
+
+    public delegate void ActOnAggregate<in TAggregate, in TCommand>(TAggregate aggregate, TCommand command) where TAggregate : Aggregate;
+
+    internal delegate ValueTask<T> HandleUntypedCommand<T>(T aggregate, object command, CancellationToken cancellationToken) where T : Aggregate;
+
+    public delegate Task<TId> GetIdFromCommandAsync<TId, in TCommand>(TCommand command, CancellationToken cancellationToken)
+        where TId : Id where TCommand : class;
+
+    public delegate TId GetIdFromCommand<out TId, in TCommand>(TCommand command) where TId : Id where TCommand : class;
+
+    internal delegate ValueTask<TId> GetIdFromUntypedCommand<TId>(object command, CancellationToken cancellationToken) where TId : Id;
+
+    public delegate IAggregateStore ResolveStore<in TCommand>(TCommand command) where TCommand : class;
+
+    internal delegate IAggregateStore ResolveStoreFromCommand(object command);
+}

src/Core/src/Eventuous.Application/FunctionalService/FuncHandlersMap.cs

@@ -35,23 +35,5 @@ void AddHandlerInternal<TCommand>(RegisteredFuncHandler<TState> handler) where T
         }
     }
 
-    public void AddHandler<TCommand>(
-            ExpectedState                      expectedState,
-            GetStreamNameFromCommand<TCommand> getStreamName,
-            ExecuteCommand<TState, TCommand>   action,
-            ResolveReaderFromCommand<TCommand> resolveReaderFromCommand,
-            ResolveWriterFromCommand<TCommand> resolveWriterFromCommand
-        )
-        where TCommand : class
-        => AddHandlerInternal<TCommand>(
-            new RegisteredFuncHandler<TState>(
-                expectedState,
-                getStreamName.AsGetStream(),
-                action.AsExecute(),
-                resolveReaderFromCommand.AsResolveReader(),
-                resolveWriterFromCommand.AsResolveWriter()
-            )
-        );
-
     public bool TryGet<TCommand>([NotNullWhen(true)] out RegisteredFuncHandler<TState>? handler) => _typeMap.TryGetValue<TCommand>(out handler);
 }

src/Core/src/Eventuous.Application/FunctionalService/FunctionalCommandService.cs

@@ -9,11 +9,6 @@
 
 public abstract class FunctionalCommandService<TState>(IEventReader reader, IEventWriter writer, TypeMapper? typeMap = null)
     : IFuncCommandService<TState>, IStateCommandService<TState> where TState : State<TState>, new() {
-    [PublicAPI]
-    protected IEventReader Reader { get; } = reader;
-    [PublicAPI]
-    protected IEventWriter Writer { get; } = writer;
-
     readonly TypeMapper              _typeMap  = typeMap ?? TypeMap.Instance;
     readonly FuncHandlersMap<TState> _handlers = new();
 
@@ -28,7 +23,7 @@
     /// <typeparam name="TCommand">Command type</typeparam>
     /// <returns></returns>
     protected FuncCommandHandlerBuilder<TCommand, TState> On<TCommand>() where TCommand : class {
-        var builder = new FuncCommandHandlerBuilder<TCommand, TState>(Reader, Writer);
+        var builder = new FuncCommandHandlerBuilder<TCommand, TState>(reader, writer);
         _builders.Add(typeof(TCommand), builder);
 
         return builder;
@@ -39,11 +34,11 @@
         => On<TCommand>().InState(ExpectedState.New).GetStream(getStreamName).Act(action);
 
     [Obsolete("Use On<TCommand>().InState(ExpectedState.Existing).GetStream(...).Act(...) instead")]
     protected void OnExisting<TCommand>(GetStreamNameFromCommand<TCommand> getStreamName, ExecuteCommand<TState, TCommand> action) where TCommand : class
         => On<TCommand>().InState(ExpectedState.Existing).GetStream(getStreamName).Act(action);
 
     [Obsolete("Use On<TCommand>().InState(ExpectedState.Any).GetStream(...).Act(...) instead")]
     protected void OnAny<TCommand>(GetStreamNameFromCommand<TCommand> getStreamName, ExecuteCommand<TState, TCommand> action) where TCommand : class
         => On<TCommand>().InState(ExpectedState.Any).GetStream(getStreamName).Act(action);
 
     public async Task<Result<TState>> Handle<TCommand>(TCommand command, CancellationToken cancellationToken) where TCommand : class {
@@ -57,11 +52,13 @@
         }
 
         var streamName = await registeredHandler.GetStream(command, cancellationToken).NoContext();
+        var reader     = registeredHandler.ResolveReaderFromCommand(command);
+        var writer     = registeredHandler.ResolveWriterFromCommand(command);
 
         try {
             var loadedState = registeredHandler.ExpectedState switch {
-                ExpectedState.Any      => await Reader.LoadStateOrNew<TState>(streamName, cancellationToken).NoContext(),
-                ExpectedState.Existing => await Reader.LoadState<TState>(streamName, cancellationToken).NoContext(),
+                ExpectedState.Any      => await reader.LoadStateOrNew<TState>(streamName, cancellationToken).NoContext(),
+                ExpectedState.Existing => await reader.LoadState<TState>(streamName, cancellationToken).NoContext(),
                 ExpectedState.New      => new FoldedEventStream<TState>(streamName, ExpectedStreamVersion.NoStream, Array.Empty<object>()),
                 _                      => throw new ArgumentOutOfRangeException(nameof(registeredHandler.ExpectedState), "Unknown expected state")
             };
@@ -76,7 +73,7 @@
             // Zero in the global position would mean nothing, so the receiver need to check the Changes.Length
             if (newEvents.Length == 0) return new OkResult<TState>(newState, Array.Empty<Change>(), 0);
 
-            var storeResult = await Writer.Store(streamName, (int)loadedState.StreamVersion.Value, newEvents, static e => e, cancellationToken).NoContext();
+            var storeResult = await writer.Store(streamName, (int)loadedState.StreamVersion.Value, newEvents, static e => e, cancellationToken).NoContext();
             var changes     = newEvents.Select(x => new Change(x, _typeMap.GetTypeName(x)));
             Log.CommandHandled<TCommand>();
 

src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs

@@ -7,13 +7,13 @@ namespace Eventuous;
 
 public static class StoreFunctions {
     public static async Task<AppendEventsResult> Store(
-        this IEventWriter              eventWriter,
-        StreamName                     streamName,
-        int                            originalVersion,
-        IReadOnlyCollection<object>    changes,
-        Func<StreamEvent, StreamEvent> amendEvent,
-        CancellationToken              cancellationToken
-    ) {
+            this IEventWriter              eventWriter,
+            StreamName                     streamName,
+            int                            originalVersion,
+            IReadOnlyCollection<object>    changes,
+            Func<StreamEvent, StreamEvent> amendEvent,
+            CancellationToken              cancellationToken
+        ) {
         Ensure.NotNull(changes);
 
         if (changes.Count == 0) return AppendEventsResult.NoOp;
@@ -30,44 +30,44 @@ CancellationToken              cancellationToken
                 .NoContext();
 
             return result;
-        }
-        catch (Exception e) {
+        } catch (Exception e) {
             throw e.InnerException?.Message.Contains("WrongExpectedVersion") == true
                 ? new OptimisticConcurrencyException(streamName, e)
                 : e;
         }
 
         StreamEvent ToStreamEvent(object evt, int position) {
             var streamEvent = new StreamEvent(Guid.NewGuid(), evt, new Metadata(), "", position);
+
             return amendEvent(streamEvent);
         }
     }
 
     public static async Task<AppendEventsResult> Store<T>(
-        this IEventWriter              eventWriter,
-        StreamName                     streamName,
-        T                              aggregate,
-        Func<StreamEvent, StreamEvent> amendEvent,
-        CancellationToken              cancellationToken
-    ) where T : Aggregate {
+            this IEventWriter              eventWriter,
+            StreamName                     streamName,
+            T                              aggregate,
+            Func<StreamEvent, StreamEvent> amendEvent,
+            CancellationToken              cancellationToken
+        ) where T : Aggregate {
         Ensure.NotNull(aggregate);
 
         try {
             return await eventWriter.Store(streamName, aggregate.OriginalVersion, aggregate.Changes, amendEvent, cancellationToken).NoContext();
-        }
-        catch (OptimisticConcurrencyException e) {
+        } catch (OptimisticConcurrencyException e) {
             Log.UnableToStoreAggregate<T>(streamName, e);
+
             throw new OptimisticConcurrencyException<T>(streamName, e.InnerException!);
         }
     }
 
     public static async Task<StreamEvent[]> ReadStream(
-        this IEventReader  eventReader,
-        StreamName         streamName,
-        StreamReadPosition start,
-        bool               failIfNotFound,
-        CancellationToken  cancellationToken
-    ) {
+            this IEventReader  eventReader,
+            StreamName         streamName,
+            StreamReadPosition start,
+            bool               failIfNotFound,
+            CancellationToken  cancellationToken
+        ) {
         const int pageSize = 500;
 
         var streamEvents = new List<StreamEvent>();
@@ -83,8 +83,7 @@ CancellationToken  cancellationToken
 
                 position = new StreamReadPosition(position.Value + events.Length);
             }
-        }
-        catch (StreamNotFound) when (!failIfNotFound) {
+        } catch (StreamNotFound) when (!failIfNotFound) {
             return Array.Empty<StreamEvent>();
         }
 

src/Core/src/Eventuous.Persistence/StateStore/StateStoreFunctions.cs

@@ -10,7 +10,7 @@
         where T : State<T>, new()
         => reader.LoadEventsInternal<T>(streamName, true, cancellationToken);
 
     public static async Task<FoldedEventStream<T>> LoadState<T, TId>(this IEventReader reader, StreamNameMap streamNameMap, TId id, CancellationToken cancellationToken)
         where T : State<T>, new() where TId : Id {
         var foldedStream = await reader.LoadEventsInternal<T>(streamNameMap.GetStreamName(id), true, cancellationToken);
         return foldedStream with { State = foldedStream.State.WithId(id) };
@@ -20,7 +20,7 @@
         where T : State<T>, new()
         => reader.LoadEventsInternal<T>(streamName, false, cancellationToken);
 
     public static async Task<FoldedEventStream<T>> LoadStateOrNew<T, TId>(this IEventReader reader, StreamNameMap streamNameMap, TId id, CancellationToken cancellationToken)
         where T : State<T>, new()
         where TId : Id {
         var foldedStream = await reader.LoadEventsInternal<T>(streamNameMap.GetStreamName(id), false, cancellationToken);
@@ -36,7 +36,10 @@
         try {
             var streamEvents = await reader.ReadStream(streamName, StreamReadPosition.Start, failIfNotFound, cancellationToken).NoContext();
             var events       = streamEvents.Select(x => x.Payload!).ToArray();
-            return (new FoldedEventStream<T>(streamName, new ExpectedStreamVersion(streamEvents.Last().Position), events));
+            var expectedVersion = events.Length == 0
+                ? ExpectedStreamVersion.NoStream
+                : new ExpectedStreamVersion(streamEvents.Last().Position);
+            return (new FoldedEventStream<T>(streamName, expectedVersion, events));
         }
         catch (StreamNotFound) when (!failIfNotFound) {
             return new FoldedEventStream<T>(streamName, ExpectedStreamVersion.NoStream, Array.Empty<object>());

src/Core/test/Eventuous.Tests.Application/BookingFuncService.cs

@@ -14,6 +14,7 @@ public BookingFuncService(IEventStore store, TypeMapper? typeMap = null)
         OnNew<BookRoom>(cmd => GetStream(cmd.BookingId), BookRoom);
 #pragma warning restore CS0618 // Type or member is obsolete
         On<RecordPayment>().InState(ExpectedState.Existing).GetStream(cmd => GetStream(cmd.BookingId)).Act(RecordPayment);
+        On<ImportBooking>().InState(ExpectedState.Any).GetStream(cmd => GetStream(cmd.BookingId)).Act(ImportBooking);
 
         return;
 
@@ -24,6 +25,10 @@ static IEnumerable<object> BookRoom(BookRoom cmd) {
             yield return new RoomBooked(cmd.RoomId, cmd.CheckIn, cmd.CheckOut, cmd.Price);
         }
 
+        static IEnumerable<object> ImportBooking(BookingState state, object[] events, ImportBooking cmd) {
+            yield return new BookingImported(cmd.RoomId, cmd.Price, cmd.CheckIn, cmd.CheckOut);
+        }
+
         static IEnumerable<object> RecordPayment(BookingState state, object[] originalEvents, RecordPayment cmd) {
             if (state.HasPayment(cmd.PaymentId)) yield break;
 

src/Core/test/Eventuous.Tests.Application/FunctionalServiceTests.cs

@@ -39,11 +39,9 @@
 
     [Fact]
     public async Task ExecuteOnExistingStream() {
-        var bookRoom = await Seed();
-
+        var bookRoom    = await Seed();
         var paymentTime = DateTimeOffset.Now;
-
-        var cmd = new Commands.RecordPayment(new BookingId(bookRoom.BookingId), "444", new Money(bookRoom.Price), paymentTime);
+        var cmd         = new Commands.RecordPayment(new BookingId(bookRoom.BookingId), "444", new Money(bookRoom.Price), paymentTime);
 
         var result = await _service.Handle(cmd, default);
 
@@ -57,12 +55,34 @@
         newEvents.Should().BeEquivalentTo(expectedResult);
     }
 
-    async Task<Commands.BookRoom> Seed() {
+    [Fact]
+    public async Task ExecuteOnAnyForNewStream() {
+        var bookRoom    = GetBookRoom();
+        var paymentTime = DateTimeOffset.Now;
+
+        var cmd = new Commands.ImportBooking {
+            BookingId = "dummy",
+            Price     = bookRoom.Price,
+            CheckIn   = bookRoom.CheckIn,
+            CheckOut  = bookRoom.CheckOut,
+            RoomId    = bookRoom.RoomId
+        };
+        var result = await _service.Handle(cmd, default);
+        result.Success.Should().BeTrue();
+        result.Changes.Should().HaveCount(1);
+    }
+
+    static Commands.BookRoom GetBookRoom() {
         var checkIn  = LocalDate.FromDateTime(DateTime.Today);
         var checkOut = checkIn.PlusDays(1);
-        var cmd      = new Commands.BookRoom("123", "234", checkIn, checkOut, 100);
 
+        return new Commands.BookRoom("123", "234", checkIn, checkOut, 100);
+    }
+
+    async Task<Commands.BookRoom> Seed() {
+        var cmd = GetBookRoom();
         await _service.Handle(cmd, default);
+
         return cmd;
     }
 

test/Eventuous.TestHelpers/Fakes/InMemoryEventStore.cs

@@ -42,12 +42,12 @@
     )
         => Task.FromResult(FindStream(stream).GetEventsBackwards(count).ToArray());
 
     public Task<long> ReadStream(
         StreamName          stream,
         StreamReadPosition  start,
         int                 count,
         Action<StreamEvent> callback,
         CancellationToken   cancellationToken
     ) {
         var readCount = 0L;
 
@@ -82,12 +82,10 @@
 
     // ReSharper disable once ReturnTypeCanBeEnumerable.Local
     InMemoryStream FindStream(StreamName stream) {
-        if (!_storage.TryGetValue(stream, out var existing)) throw new NotFound(stream);
+        if (!_storage.TryGetValue(stream, out var existing)) throw new StreamNotFound(stream);
 
         return existing;
     }
-
-    class NotFound(StreamName stream) : Exception($"Stream not found: {stream}");
 }
 
 record StoredEvent(StreamEvent Event, int Position);
@@ -95,7 +93,7 @@
 class InMemoryStream {
     public int Version { get; private set; } = -1;
 
     StreamName                 _name;
     readonly List<StoredEvent> _events = new();
 
     public InMemoryStream(StreamName name)