Projection snapshot - initial framework

Author: Duncan-at-Hedgefacts

src/EventSourcingOnAzureFunctions.Common/EventSourcing/Implementation/AzureStorage/AppendBlob/BlobEventStreamBase.cs

@@ -21,9 +21,17 @@ public abstract class BlobEventStreamBase
         public const string METADATA_DATE_CREATED = "DATECREATED";
         public const string METADATA_CORRELATION_ID = "CORRELATIONIDENTIFIER";
 
-        // Named specific sub-folders
+        /// <summary>
+        /// The default folder where uncateggorised entities are stored
+        /// </summary>
         public const string ORPHANS_FOLDER = "uncategorised";
+        /// <summary>
+        /// The default subfolder where the event streams are stored
+        /// </summary>
         public const string EVENTSTREAM_FOLDER = "eventstreams";
+        /// <summary>
+        /// The default subfolder where projection snapshots are stored
+        /// </summary>
         public const string SNAPSHOTS_FOLDER = "snapshots";
 
         private readonly CloudBlobContainer _blobBasePath;

src/EventSourcingOnAzureFunctions.Common/EventSourcing/Implementation/AzureStorage/Blob/BlobProjectionSnapshotBase.cs

@@ -0,0 +1,46 @@
+using EventSourcingOnAzureFunctions.Common.EventSourcing.Implementation.AzureStorage.AppendBlob;
+using EventSourcingOnAzureFunctions.Common.EventSourcing.Interfaces;
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace EventSourcingOnAzureFunctions.Common.EventSourcing.Implementation.AzureStorage.Blob
+{
+    public class BlobSnapshotBase
+    {
+
+        /// <summary>
+        /// Turn the given snapshot definition into an unique filename
+        /// </summary>
+        /// <param name="snapshot">
+        /// The snapshot definition to be saved or read
+        /// </param>
+        /// <remarks>
+        /// The filename is the sequence number with the rest of the snapshot definition
+        /// going into the folder path
+        /// </remarks>
+        public static string MakeSnapshotFilename(ISnapshot snapshot)
+        {
+            if (snapshot != null)
+            {
+                return BlobEventStreamBase.MakeValidStorageFolderName($"{snapshot.CurrentSequenceNumber}");
+            }
+            return string.Empty;
+        }
+
+        /// <summary>
+        /// Make a folder path in which a snapshot for a projection can be saved
+        /// </summary>
+        /// <param name="snapshot"></param>
+        /// <returns></returns>
+        public static string MakeSnapshotFolder(ISnapshot snapshot)
+        {
+            if (snapshot != null)
+            {
+                return BlobEventStreamBase.MakeValidStorageFolderName($"{snapshot.DomainName}/{snapshot.EntityTypeName}/{snapshot.InstanceKey}");
+            }
+            return string.Empty;
+        }
+
+    }
+}

src/EventSourcingOnAzureFunctions.Common/EventSourcing/Implementation/AzureStorage/Blob/BlobProjectionSnapshotReader.cs

@@ -0,0 +1,47 @@
+using EventSourcingOnAzureFunctions.Common.EventSourcing.Interfaces;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace EventSourcingOnAzureFunctions.Common.EventSourcing.Implementation.AzureStorage.Blob
+{
+    /// <summary>
+    /// Class to read a projection snapshot from an Azure storage blob
+    /// </summary>
+    public sealed class BlobProjectionSnapshotReader
+        : BlobSnapshotBase,
+          IProjectionSnapshotReader
+    {
+
+        /// <summary>
+        /// Loads the snapshot record closes to the requested as-of sequence number 
+        /// from which to start running a projection
+        /// </summary>
+        /// <typeparam name="TProjection">
+        /// The projection into which to load the initial snapshot
+        /// </typeparam>
+        /// <param name="snapshot">
+        /// The snapshot definition from which to load
+        /// </param>
+        public async Task<TProjection> LoadProjection<TProjection>(ISnapshot snapshot) where TProjection : IProjectionWithSnapshots, new()
+        {
+            if (snapshot != null)
+            {
+                if (snapshot.CurrentSequenceNumber <= 0)
+                {
+                    // We are just asking for the latest available snapshot
+
+                }
+                // Create a filename from the snapshot
+                string filename = MakeSnapshotFilename(snapshot);
+                string filepath = MakeSnapshotFolder(snapshot);
+
+                TProjection projectionToLoad = new TProjection();
+                
+                return projectionToLoad;
+            }
+            throw new NotImplementedException();
+        }
+    }
+}

src/EventSourcingOnAzureFunctions.Common/EventSourcing/Implementation/AzureStorage/Blob/BlobProjectionSnapshotWriter.cs

@@ -0,0 +1,37 @@
+using EventSourcingOnAzureFunctions.Common.EventSourcing.Interfaces;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace EventSourcingOnAzureFunctions.Common.EventSourcing.Implementation.AzureStorage.Blob
+{
+    /// <summary>
+    /// Writer to save projection snapshots in an Azure storage blob
+    /// </summary>
+    public sealed class BlobProjectionSnapshotWriter
+        : BlobSnapshotBase, 
+          IProjectionSnapshotWriter
+    {
+
+
+        public async Task WriteSnapshot<TProjection>(ISnapshot snapshot, 
+            TProjection state) where TProjection : IProjectionWithSnapshots
+        {
+            if (snapshot != null)
+            {
+                if (state != null)
+                {
+                    // Create a filename from the snapshot
+                    string filename = MakeSnapshotFilename(snapshot);
+                    string filepath = MakeSnapshotFolder(snapshot);
+
+
+                }
+            }
+            throw new NotImplementedException();
+        }
+
+
+    }
+}

src/EventSourcingOnAzureFunctions.Common/EventSourcing/Interfaces/IProjectionSnapshotReader.cs

@@ -22,7 +22,7 @@ public interface IProjectionSnapshotReader
         /// The entity instance and sequence number to get the snapshot from
         /// (If no sequence number is specified, get the latest snapshot)
         /// </param>
-        Task<TProjection> LoadProjection<TProjection>(ISnapshot snapshot) where TProjection : IProjection, new();
+        Task<TProjection> LoadProjection<TProjection>(ISnapshot snapshot) where TProjection : IProjectionWithSnapshots, new();
 
     }
 }

src/EventSourcingOnAzureFunctions.Common/EventSourcing/Interfaces/IProjectionSnapshotWriter.cs

@@ -23,7 +23,7 @@ public interface IProjectionSnapshotWriter
         /// <param name="state">
         /// The projection state as at that point
         /// </param>
-        Task WriteSnapshot<TProjection>(ISnapshot snapshot, TProjection state) where TProjection : IProjection;
+        Task WriteSnapshot<TProjection>(ISnapshot snapshot, TProjection state) where TProjection : IProjectionWithSnapshots;
 
     }
 }

src/EventSourcingOnAzureFunctions.Common/EventSourcing/Interfaces/IProjectionWithSnapshots.cs

@@ -0,0 +1,36 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace EventSourcingOnAzureFunctions.Common.EventSourcing.Interfaces
+{
+    /// <summary>
+    /// A projection which supports saving a state to, and reading a state from, a snapshot
+    /// </summary>
+    public interface IProjectionWithSnapshots
+        : IProjection 
+    {
+
+        /// <summary>
+        /// Read the current state of a projection from a snapshot 
+        /// </summary>
+        /// <param name="sequenceNumber">
+        /// The sequence number at which the snapshot was taken
+        /// </param>
+        /// <param name="asOfDate">
+        /// The as-of date at which the snapshot was taken
+        /// </param>
+        /// <param name="snapshot">
+        /// The raw data object of the state as at when the snapshot was taken
+        /// </param>
+        void FromSnapshot(int sequenceNumber,
+            Nullable<DateTime> asOfDate,
+            object snapshot);
+
+        /// <summary>
+        /// Turn the current state of this projection into a snapshot object to be saved
+        /// </summary>
+        object ToSnapshot();
+
+    }
+}

src/EventSourcingOnAzureFunctions.Common/EventSourcing/Projection.cs

@@ -8,14 +8,18 @@
 
 namespace EventSourcingOnAzureFunctions.Common.EventSourcing
 {
+    /// <summary>
+    /// A projection is a piece of code that runs over the event stream for an entity in order to derive some state information about that entity.
+    /// For each event in the stream it needs to decide (a) am I interested in this kind of event and if so (b) what do I do with it.
+    /// </summary>
     public class Projection
          : IEventStreamIdentity
     {
 
         private readonly IEventStreamSettings _settings = null;
         private readonly IProjectionProcessor _projectionProcessor = null;
         private readonly INotificationDispatcher _notificationDispatcher = null;
-        private readonly IProjectionSnapshotWriter _snapshortWriter = null;
+        private readonly IProjectionSnapshotWriter _snapshotWriter = null;
         private readonly IProjectionSnapshotReader _snapshotReader = null;
 
         private readonly string _domainName;
@@ -56,10 +60,11 @@ public string InstanceKey
             }
         }
 
+        
+        private readonly string _projectionTypeName;
         /// <summary>
         /// The type of the projection we are going to run 
         /// </summary>
-        private readonly string _projectionTypeName;
         public string ProjectionTypeName
         {
             get
@@ -69,13 +74,42 @@ public string ProjectionTypeName
         }
 
 
+        /// <summary>
+        /// Process the projection and return the results
+        /// </summary>
+        /// <typeparam name="TProjection">
+        /// The type of the projection to be executed
+        /// </typeparam>
+        /// <param name="asOfDate">
+        /// The date up to which to run the projection. 
+        /// If this is not specified then run to the end of the event stream.
+        /// </param>
+        /// <returns>
+        /// A task which is used to execute the projection
+        /// </returns>
         public async Task<TProjection> Process<TProjection>(Nullable<DateTime> asOfDate = null) where TProjection : IProjection, new()
         {
             TProjection projectionToRun = new TProjection();
             return await Process(projectionToRun, asOfDate);
         }
 
-
+        /// <summary>
+        /// Process the projection and return the results
+        /// </summary>
+        /// <typeparam name="TProjection">
+        /// The type of the projection to be executed
+        /// </typeparam>
+        /// <param name="projectionToRun">
+        /// The instance of the projection to start processing from
+        /// (This allows a projection to start from a given initial state)
+        /// </param>
+        /// <param name="asOfDate">
+        /// The date up to which to run the projection. 
+        /// If this is not specified then run to the end of the event stream.
+        /// </param>
+        /// <returns>
+        /// A task which is used to execute the projection
+        /// </returns>
         public async Task<TProjection> Process<TProjection>(TProjection projectionToRun, Nullable<DateTime> asOfDate = null) where TProjection : IProjection
         {
             if (null != _projectionProcessor)
@@ -111,16 +145,23 @@ await _notificationDispatcher.ProjectionCompleted(this,
         /// <param name="state">
         /// The state of the projection when snapshotted
         /// </param>
-        public async Task WriteSnapshot<TProjection>(ISnapshot snapshot, TProjection state) where TProjection : IProjection
+        public async Task WriteSnapshot<TProjection>(ISnapshot snapshot, TProjection state) where TProjection : IProjectionWithSnapshots
         {
-            // If there is a snapshot writer, use it...
-            if (null != _snapshortWriter)
+            if (state.SupportsSnapshots)
             {
-                await _snapshortWriter.WriteSnapshot(snapshot, state); 
+                // If there is a snapshot writer, use it...
+                if (null != _snapshotWriter)
+                {
+                    await _snapshotWriter.WriteSnapshot(snapshot, state);
+                }
             }
         }
 
         private readonly string _connectionStringName;
+
+        /// <summary>
+        /// The name of the connection string used to run this projection
+        /// </summary>
         public string ConnectionStringName
         {
             get
@@ -193,7 +234,7 @@ public Projection(ProjectionAttribute attribute,
 
             if (null != snapshotWriter )
             {
-                _snapshortWriter = snapshotWriter;
+                _snapshotWriter = snapshotWriter;
             }
 
         }