Add support for Updating aggregates in addition to GetById in repositories (#81)

* Changes to CorrelatedStreamStoreRepository
- Add a ctor overload to inject both an IRepository and a cache factory.
- Fix a bug in the Dispose(bool) method that could cause an exception to be thrown.
- Update repositories and caches to include update semantics

Author: joshkempner

src/ReactiveDomain.Core/IEventSource.cs

@@ -27,7 +27,16 @@ public interface IEventSource
         /// <exception cref="System.ArgumentNullException">Thrown when <paramref name="events"/> is <c>null</c>.</exception>
         /// <exception cref="System.InvalidOperationException">Thrown when this instance has already recorded events.</exception>
         void RestoreFromEvents(IEnumerable<object> events);
-        
+
+        /// <summary>
+        /// Updates this instance with historical events.
+        /// </summary>
+        /// <param name="events">The events to apply.</param>
+        /// <param name="expectedVersion">The expected version prior to applying events</param>
+        /// <exception cref="System.ArgumentNullException">Thrown when <paramref name="events"/> is <c>null</c>.</exception>
+        /// <exception cref="System.InvalidOperationException">Thrown when this instance does not have historical events or expected version mismatch</exception>
+        void UpdateWithEvents(IEnumerable<object> events, long expectedVersion);
+
         /// <summary>
         /// Takes the recorded history of events from this instance (CQS violation, beware).
         /// </summary>

src/ReactiveDomain.Foundation/Domain/EventDrivenStateMachine.cs

@@ -44,6 +44,20 @@ public void RestoreFromEvents(IEnumerable<object> events) {
                 _router.Route(@event);
             }
         }
+        public void UpdateWithEvents(IEnumerable<object> events, long expectedVersion) {
+            if (events == null)
+                throw new ArgumentNullException(nameof(events));
+            if (_version < 0)
+                throw new InvalidOperationException("Updating with events is not possible when an instance has no historical events.");
+            if ( _version != expectedVersion) {
+                throw  new InvalidOperationException("Expected version mismatch when updating ");
+            }
+
+            foreach (var @event in events) {
+                _version++;
+                _router.Route(@event);
+            }
+        }
 
         /// <summary>
         /// Returns all events from the EventRecorder since state was loaded or the last time TakeEvents was called

src/ReactiveDomain.Foundation/IRepository.cs

@@ -3,11 +3,9 @@ namespace ReactiveDomain.Foundation
 {
     public interface IRepository
 	{
-        bool TryGetById<TAggregate>(Guid id, out TAggregate aggregate) where TAggregate : class, IEventSource;
-        bool TryGetById<TAggregate>(Guid id, int version, out TAggregate aggregate) where TAggregate : class, IEventSource;
-        TAggregate GetById<TAggregate>(Guid id) where TAggregate : class, IEventSource;
-		TAggregate GetById<TAggregate>(Guid id, int version) where TAggregate : class, IEventSource;
-		void Save(IEventSource aggregate);
-        void UpdateToCurrent(IEventSource aggregate);
+        bool TryGetById<TAggregate>(Guid id,  out TAggregate aggregate, int version = int.MaxValue) where TAggregate : class, IEventSource;
+        TAggregate GetById<TAggregate>(Guid id, int version = int.MaxValue) where TAggregate : class, IEventSource;
+		void Update<TAggregate>(ref TAggregate aggregate, int version = int.MaxValue) where TAggregate : class, IEventSource;
+        void Save(IEventSource aggregate);
     }
 }

src/ReactiveDomain.Foundation/StreamStore/CachingRepository.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Generic;
 
 namespace ReactiveDomain.Foundation.StreamStore {
     /// <summary>
@@ -21,24 +20,15 @@ public CachingRepository(IRepository baseRepository, Func<IRepository,IAggregate
             if (cacheFactory == null) {
                 cacheFactory = repo => new ReadThroughAggregateCache(repo);
             }
-
             _cache = cacheFactory(baseRepository);
         }
-
-
-        public bool TryGetById<TAggregate>(Guid id, out TAggregate aggregate) where TAggregate : class, IEventSource
-        {
-            return _cache.GetById(id, out aggregate);
-        }
-        public TAggregate GetById<TAggregate>(Guid id) where TAggregate : class, IEventSource
-        {
-            if (!_cache.GetById(id, out TAggregate aggregate)) {
-                throw new AggregateNotFoundException(id,typeof(TAggregate));
-            }
-            return aggregate;
+        public bool TryGetById<TAggregate>(Guid id, out TAggregate aggregate) where TAggregate : class, IEventSource {
+            return _cache.TryGetById(id, out aggregate);
+        }
+        public TAggregate GetById<TAggregate>(Guid id) where TAggregate : class, IEventSource {
+            return _cache.GetById<TAggregate>(id);
         }
-        public void Save(IEventSource aggregate)
-        {
+        public void Save(IEventSource aggregate) {
             _cache.Save(aggregate);
         }
         public bool ClearCache(Guid id) {

src/ReactiveDomain.Foundation/StreamStore/CorrelatedStreamStoreRepository.cs

@@ -3,33 +3,39 @@
 using ReactiveDomain.Messaging;
 
 // ReSharper disable once CheckNamespace
-namespace ReactiveDomain.Foundation{
-    public class CorrelatedStreamStoreRepository:ICorrelatedRepository, IDisposable
+namespace ReactiveDomain.Foundation {
+    public class CorrelatedStreamStoreRepository : ICorrelatedRepository, IDisposable
     {
         private readonly IRepository _repository;
-        private readonly IAggregateCache _cache = null;
-        public CorrelatedStreamStoreRepository(IRepository repository) {
+        private readonly IAggregateCache _cache;
+        public CorrelatedStreamStoreRepository(
+            IRepository repository,
+            Func<IRepository, IAggregateCache> cacheFactory = null) {
             _repository = repository;
+            if (cacheFactory != null) {
+                _cache = cacheFactory(_repository);
+            }
         }
+
         public CorrelatedStreamStoreRepository(
             IStreamNameBuilder streamNameBuilder,
             IStreamStoreConnection streamStoreConnection,
             IEventSerializer eventSerializer,
-            Func<IRepository,IAggregateCache> cacheFactory = null) {
-            _repository = new StreamStoreRepository(streamNameBuilder,streamStoreConnection,eventSerializer);
+            Func<IRepository, IAggregateCache> cacheFactory = null) {
+            _repository = new StreamStoreRepository(streamNameBuilder, streamStoreConnection, eventSerializer);
             if (cacheFactory != null) {
                 _cache = cacheFactory(_repository);
             }
         }
-        
+
         public bool TryGetById<TAggregate>(Guid id, out TAggregate aggregate, ICorrelatedMessage source) where TAggregate : AggregateRoot, IEventSource {
             return TryGetById(id, int.MaxValue, out aggregate, source);
         }
 
         public TAggregate GetById<TAggregate>(Guid id, ICorrelatedMessage source) where TAggregate : AggregateRoot, IEventSource {
-           return GetById<TAggregate>(id, int.MaxValue, source);
+            return GetById<TAggregate>(id, int.MaxValue, source);
         }
-        
+
         public bool TryGetById<TAggregate>(Guid id, int version, out TAggregate aggregate, ICorrelatedMessage source) where TAggregate : AggregateRoot, IEventSource {
             try {
                 aggregate = GetById<TAggregate>(id, version, source);
@@ -40,33 +46,38 @@ public bool TryGetById<TAggregate>(Guid id, int version, out TAggregate aggregat
                 return false;
             }
         }
-       
-        public TAggregate GetById<TAggregate>(Guid id, int version, ICorrelatedMessage source) where TAggregate : AggregateRoot, IEventSource
-        {
-            TAggregate agg = null;
-            _cache?.GetById(id, out agg);
 
-            if(agg == null || agg.Version > version) {
+        public TAggregate GetById<TAggregate>(Guid id, int version, ICorrelatedMessage source) where TAggregate : AggregateRoot, IEventSource {
+            TAggregate agg = _cache?.GetById<TAggregate>(id, version);
+
+            if (agg == null || agg.Version > version) {
                 agg = _repository.GetById<TAggregate>(id, version);
+                if (agg != null) {
+                    _cache?.Save(agg);
+                }
+            }
+
+            if (agg != null) {
+                ((ICorrelatedEventSource) agg).Source = source;
             }
-            
-            ((ICorrelatedEventSource)agg).Source = source;
             return agg;
         }
 
         public void Save(IEventSource aggregate) {
             if (_cache != null) {
-                _cache.Save( aggregate);
+                _cache.Save(aggregate);
             }
             else {
                 _repository.Save(aggregate);
             }
         }
+
         protected virtual void Dispose(bool disposing) {
             if (disposing) {
-                _cache.Dispose();
+                _cache?.Dispose();
             }
         }
+
         public void Dispose() {
             Dispose(true);
             GC.SuppressFinalize(this);

src/ReactiveDomain.Foundation/StreamStore/IAggregateCache.cs

@@ -5,10 +5,8 @@ namespace ReactiveDomain.Foundation.StreamStore {
     /// While it might seem more natural to save and restore the event set behind the aggregate,
     /// this cache stores only the collapsed state in the aggregate  
     /// </summary>
-    public interface IAggregateCache: IDisposable
+    public interface IAggregateCache:IRepository, IDisposable
     {
-        bool GetById<TAggregate>(Guid id, out TAggregate aggregate) where TAggregate : class, IEventSource;
-        bool Save(IEventSource aggregate);
         bool Remove(Guid id);
         void Clear();
     }

src/ReactiveDomain.Foundation/StreamStore/ReadThroughAggregateCache.cs

@@ -1,7 +1,9 @@
 using System;
+using System.Collections;
 using System.Collections.Generic;
 
-namespace ReactiveDomain.Foundation.StreamStore {
+namespace ReactiveDomain.Foundation.StreamStore
+{
     /// <summary>
     /// This implementation assumes that the cached aggregate may have changes since it was last seen
     ///
@@ -14,47 +16,81 @@ namespace ReactiveDomain.Foundation.StreamStore {
     ///
     /// Save failures will clear the aggregate from the cache and return false
     /// </summary>
-    public class ReadThroughAggregateCache :IAggregateCache, IDisposable {
+    public class ReadThroughAggregateCache : IAggregateCache, IDisposable
+    {
 
         private readonly IRepository _baseRepository;
         private readonly Dictionary<Guid, IEventSource> _knownAggregates = new Dictionary<Guid, IEventSource>();
         public ReadThroughAggregateCache(IRepository baseRepository) {
             _baseRepository = baseRepository;
         }
-
-
-        public bool GetById<TAggregate>(Guid id, out TAggregate aggregate) where TAggregate : class, IEventSource {
+        public bool TryGetById<TAggregate>(Guid id, out TAggregate aggregate, int version = int.MaxValue) where TAggregate : class, IEventSource {
             try {
-                aggregate = GetById<TAggregate>(id);
+                aggregate = GetById<TAggregate>(id, version);
                 return true;
             }
             catch (Exception) {
                 aggregate = null;
+                Remove(id);
                 return false;
             }
         }
-        private TAggregate GetById<TAggregate>(Guid id) where TAggregate : class, IEventSource {
-            if (_knownAggregates.TryGetValue(id, out var aggregate)) {
-                try {
-                    _baseRepository.UpdateToCurrent(aggregate);
-                }
-                catch(InvalidOperationException ex) {
-                    _knownAggregates.Remove(aggregate.Id);
-                    throw new Exception("Persisted version changed with recorded events in aggregate.", ex);
-                }
-                catch (AggregateVersionException ex) {
-                    _knownAggregates.Remove(aggregate.Id);
-                    throw new Exception("Persisted version mismatch.", ex);
-                }
-                return aggregate as TAggregate;
-            }
 
-            aggregate = _baseRepository.GetById<TAggregate>(id);
+        public TAggregate GetById<TAggregate>(Guid id, int version = int.MaxValue) where TAggregate : class, IEventSource {
+            if (_knownAggregates.TryGetValue(id, out var cached)) {
+                var agg = (TAggregate) cached;
+                Update(ref agg, version);
+                return (TAggregate)cached;
+            }
+            
+            var aggregate = _baseRepository.GetById<TAggregate>(id, version);
             _knownAggregates.Add(id, aggregate);
+            return aggregate;
+        }
+
+        public void Update<TAggregate>(ref TAggregate aggregate, int version = int.MaxValue) where TAggregate : class, IEventSource {
+            if (aggregate == null) throw new ArgumentNullException(nameof(aggregate));
+            if (aggregate.ExpectedVersion == version) return;
+
+            _knownAggregates.TryGetValue(aggregate.Id, out var cached);
+            
+            if (cached == null) {
+                _baseRepository.Update(ref aggregate, version);
+                UpdateCache(aggregate);
+                return;
+            }
+
+            if (cached.ExpectedVersion == version) {
+                aggregate = (TAggregate)cached;
+                return;
+            }
 
-            return (TAggregate)aggregate;
+            if (cached.ExpectedVersion > version) { 
+                //cache is ahead of requested version, we need to get it fresh from repo
+                _baseRepository.Update(ref aggregate, version);
+                //don't regress the cache, just return
+                return;
+            }
+            //cache is ahead of item, but behind requested version
+            aggregate = (TAggregate)cached;
+            _baseRepository.Update(ref aggregate, version);
+            UpdateCache(aggregate);
         }
-        public bool Save(IEventSource aggregate) {
+
+
+
+        private void UpdateCache(IEventSource aggregate) {
+            if (_knownAggregates.TryGetValue(aggregate.Id, out var cached)) {
+                if (cached.ExpectedVersion < aggregate.ExpectedVersion) {
+                    _knownAggregates[aggregate.Id] = aggregate;
+                }
+            }
+            else {
+                _knownAggregates.Add(aggregate.Id, aggregate);
+            }
+        }
+
+        public void Save(IEventSource aggregate) {
             try {
                 _baseRepository.Save(aggregate);
                 if (!_knownAggregates.ContainsKey(aggregate.Id)) {
@@ -63,15 +99,12 @@ public bool Save(IEventSource aggregate) {
                 else {
                     _knownAggregates[aggregate.Id] = aggregate;
                 }
-
-                return true;
             }
             catch {
                 _knownAggregates.Remove(aggregate.Id);
-                return false;
             }
 
-            
+
         }
         public bool Remove(Guid id) {
             return _knownAggregates.Remove(id);