Add positions to the result

Author: alexeyzimarev

src/Eventuous.EventStoreDB/EsdbEventStore.cs

@@ -16,15 +16,15 @@ public class EsdbEventStore : IEventStore {
         public EsdbEventStore(EventStoreClientSettings clientSettings)
             : this(new EventStoreClient(Ensure.NotNull(clientSettings, nameof(clientSettings)))) { }
 
-        public async Task AppendEvents(
+        public async Task<AppendEventsResult> AppendEvents(
             string                           stream,
             ExpectedStreamVersion            expectedVersion,
             IReadOnlyCollection<StreamEvent> events,
             CancellationToken                cancellationToken
         ) {
             var proposedEvents = events.Select(ToEventData);
 
-            Task resultTask;
+            Task<IWriteResult> resultTask;
 
             if (expectedVersion == ExpectedStreamVersion.NoStream)
                 resultTask = _client.AppendToStreamAsync(stream, StreamState.NoStream, proposedEvents, cancellationToken: cancellationToken);
@@ -38,7 +38,12 @@ CancellationToken                cancellationToken
                     cancellationToken: cancellationToken
                 );
 
-            await resultTask.Ignore();
+            var result = await resultTask.Ignore();
+
+            return new AppendEventsResult(
+                result.LogPosition.CommitPosition,
+                result.NextExpectedStreamRevision.ToInt64()
+            );
 
             static EventData ToEventData(StreamEvent streamEvent)
                 => new(

src/Eventuous/AggregateStore.cs

@@ -15,22 +15,24 @@ public AggregateStore(IEventStore eventStore, IEventSerializer serializer) {
             _serializer = Ensure.NotNull(serializer, nameof(serializer));
         }
 
-        public async Task Store<T>(T aggregate, CancellationToken cancellationToken)
+        public async Task<AppendEventsResult> Store<T>(T aggregate, CancellationToken cancellationToken)
             where T : Aggregate {
             Ensure.NotNull(aggregate, nameof(aggregate));
 
-            if (aggregate.Changes.Count == 0) return;
+            if (aggregate.Changes.Count == 0) return AppendEventsResult.NoOp;
 
             var stream          = StreamName.For<T>(aggregate.GetId());
             var expectedVersion = new ExpectedStreamVersion(aggregate.OriginalVersion);
 
-            await _eventStore.AppendEvents(
+            var result = await _eventStore.AppendEvents(
                 stream,
                 expectedVersion,
                 aggregate.Changes.Select(ToStreamEvent).ToArray(),
                 cancellationToken
             ).Ignore();
 
+            return result;
+
             StreamEvent ToStreamEvent(object evt)
                 => new(TypeMap.GetTypeName(evt), _serializer.Serialize(evt), null, _serializer.ContentType);
         }

src/Eventuous/AppendEventsResult.cs

@@ -0,0 +1,5 @@
+namespace Eventuous {
+    public record AppendEventsResult(ulong GlobalPosition, long NextExpectedVersion) {
+        public static AppendEventsResult NoOp = new(0, -1);
+    }
+}

src/Eventuous/ApplicationService.cs

@@ -232,9 +232,9 @@ CancellationToken cancellationToken
 
             var result = await registeredHandler.Handler(aggregate, command, cancellationToken).Ignore();
 
-            await Store.Store(result, cancellationToken).Ignore();
+            var storeResult = await Store.Store(result, cancellationToken).Ignore();
 
-            return new OkResult<T, TState, TId>(result.State, result.Changes);
+            return new OkResult<T, TState, TId>(result.State, result.Changes, storeResult.GlobalPosition);
 
             Task<T> Load() => Store.Load<T, TState, TId>(id, cancellationToken);
 

src/Eventuous/IAggregateStore.cs

@@ -16,7 +16,7 @@ public interface IAggregateStore {
         /// <param name="cancellationToken">Cancellation token</param>
         /// <typeparam name="T">Aggregate type</typeparam>
         /// <returns></returns>
-        Task Store<T>(T aggregate, CancellationToken cancellationToken) where T : Aggregate;
+        Task<AppendEventsResult> Store<T>(T aggregate, CancellationToken cancellationToken) where T : Aggregate;
 
         /// <summary>
         /// Load the aggregate from the store for a given id

src/Eventuous/IEventStore.cs

@@ -2,15 +2,36 @@
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
+using JetBrains.Annotations;
 
 namespace Eventuous {
+    /// <summary>
+    /// Event Store is a place where events are stored. It is used by <see cref="AggregateStore"/> and
+    /// <seealso cref="StateStore"/>
+    /// </summary>
+    [PublicAPI]
     public interface IEventStore {
-        Task AppendEvents(string stream, ExpectedStreamVersion expectedVersion, IReadOnlyCollection<StreamEvent> events, CancellationToken cancellationToken);
-        
-        Task<StreamEvent[]> ReadEvents(string stream, StreamReadPosition start, int count, CancellationToken cancellationToken);
-        
+        Task<AppendEventsResult> AppendEvents(
+            string                           stream,
+            ExpectedStreamVersion            expectedVersion,
+            IReadOnlyCollection<StreamEvent> events,
+            CancellationToken                cancellationToken
+        );
+
+        Task<StreamEvent[]> ReadEvents(
+            string             stream,
+            StreamReadPosition start,
+            int                count,
+            CancellationToken  cancellationToken
+        );
+
         Task<StreamEvent[]> ReadEventsBackwards(string stream, int count, CancellationToken cancellationToken);
-        
-        Task ReadStream(string stream, StreamReadPosition start, Action<StreamEvent> callback, CancellationToken cancellationToken);
+
+        Task ReadStream(
+            string              stream,
+            StreamReadPosition  start,
+            Action<StreamEvent> callback,
+            CancellationToken   cancellationToken
+        );
     }
 }

src/Eventuous/Result.cs

@@ -6,13 +6,14 @@ public abstract record Result<T, TState, TId>(TState State, IEnumerable<object>?
         where TState : AggregateState<TState, TId>, new()
         where TId : AggregateId;
 
-    public record OkResult<T, TState, TId>(TState State, IEnumerable<object> Changes) : Result<T, TState, TId>(State, Changes) 
-        where T : Aggregate<TState, TId> 
+    public record OkResult<T, TState, TId>(TState State, IEnumerable<object> Changes, ulong StreamPosition)
+        : Result<T, TState, TId>(State, Changes)
+        where T : Aggregate<TState, TId>
         where TState : AggregateState<TState, TId>, new()
         where TId : AggregateId;
-        
-    public record ErrorResult<T, TState, TId>() : Result<T, TState, TId>(new TState()) 
-        where T : Aggregate<TState, TId> 
+
+    public record ErrorResult<T, TState, TId>() : Result<T, TState, TId>(new TState())
+        where T : Aggregate<TState, TId>
         where TState : AggregateState<TState, TId>, new()
         where TId : AggregateId;
-}
+}

test/Eventuous.Tests/Fakes/InMemoryEventStore.cs

@@ -7,30 +7,37 @@
 namespace Eventuous.Tests.Fakes {
     public class InMemoryEventStore : IEventStore {
         readonly Dictionary<string, List<StreamEvent>> _storage = new();
+        readonly List<StreamEvent>                     _global  = new();
 
-        public Task AppendEvents(
+        public Task<AppendEventsResult> AppendEvents(
             string                           stream,
             ExpectedStreamVersion            expectedVersion,
             IReadOnlyCollection<StreamEvent> events,
             CancellationToken                cancellationToken
         ) {
-            return _storage.TryGetValue(stream, out var existing) ? AddToExisting() : AddToNew();
+            if (_storage.TryGetValue(stream, out var existing)) AddToExisting();
+            else AddToNew();
 
-            Task AddToExisting() {
+            _global.AddRange(events);
+
+            return Task.FromResult(new AppendEventsResult((ulong) (_global.Count - 1), _storage[stream].Count));
+
+            void AddToExisting() {
                 if (existing.Count >= expectedVersion.Value)
                     throw new WrongVersion(expectedVersion, existing.Count - 1);
 
                 existing.AddRange(events);
-                return Task.CompletedTask;
             }
 
-            Task AddToNew() {
-                _storage[stream] = events.ToList();
-                return Task.CompletedTask;
-            }
+            void AddToNew() => _storage[stream] = events.ToList();
         }
 
-        public Task<StreamEvent[]> ReadEvents(string stream, StreamReadPosition start, int count, CancellationToken cancellationToken)
+        public Task<StreamEvent[]> ReadEvents(
+            string             stream,
+            StreamReadPosition start,
+            int                count,
+            CancellationToken  cancellationToken
+        )
             => Task.FromResult(FindStream(stream).Take(count).ToArray());
 
         public Task<StreamEvent[]> ReadEventsBackwards(string stream, int count, CancellationToken cancellationToken) {
@@ -40,7 +47,12 @@ public Task<StreamEvent[]> ReadEventsBackwards(string stream, int count, Cancell
             return Task.FromResult(reversed.Take(count).ToArray());
         }
 
-        public Task ReadStream(string stream, StreamReadPosition start, Action<StreamEvent> callback, CancellationToken cancellationToken) {
+        public Task ReadStream(
+            string              stream,
+            StreamReadPosition  start,
+            Action<StreamEvent> callback,
+            CancellationToken   cancellationToken
+        ) {
             foreach (var streamEvent in FindStream(stream)) {
                 callback(streamEvent);
             }