Added support for saving multiple aggregates at once

Tacta.EventStore/Domain/DomainEvent.cs

@@ -6,7 +6,7 @@ namespace Tacta.EventStore.Domain
     public abstract class DomainEvent : IDomainEvent
     {
         public Guid Id { get; }
-
+        
         public string AggregateId { get; }
 
         [JsonIgnore] public int Version { get; private set; }

Tacta.EventStore/Domain/IDomainEvent.cs

@@ -4,6 +4,8 @@ namespace Tacta.EventStore.Domain
 {
     public interface IDomainEvent
     {
+        Guid Id { get; }
+
         int Sequence { get; }
 
         int Version { get; }

Tacta.EventStore/Repository/Aggregate.cs

@@ -0,0 +1,57 @@
+using System.Collections.Generic;
+using System.Linq;
+using Tacta.EventStore.Domain;
+using Tacta.EventStore.Repository.Exceptions;
+
+namespace Tacta.EventStore.Repository
+{
+    public sealed class Aggregate
+    {
+        public AggregateRecord AggregateRecord { get; }
+        public IReadOnlyCollection<EventRecord<IDomainEvent>> EventRecords { get; }
+
+        public Aggregate(AggregateRecord aggregateRecord, IReadOnlyCollection<EventRecord<IDomainEvent>> eventRecords)
+        {
+            if (aggregateRecord == null) throw new InvalidAggregateRecordException("Aggregate record cannot be null");
+            if (eventRecords == null) throw new InvalidEventRecordException("Event records cannot be null");
+
+            AggregateRecord = aggregateRecord;
+            EventRecords = eventRecords;
+        }
+
+        public Aggregate(IAggregateRoot<IEntityId> aggregateRoot)
+        {
+            var aggregateRecord = new AggregateRecord(
+                aggregateRoot.Id.ToString(),
+                aggregateRoot.GetType().Name,
+                aggregateRoot.Version);
+            var eventRecords = aggregateRoot.DomainEvents
+                .Select(@event => new EventRecord<IDomainEvent>(
+                                        @event.Id,
+                                        @event.CreatedAt,
+                                        @event))
+                .ToList()
+                .AsReadOnly();
+
+            AggregateRecord = aggregateRecord;
+            EventRecords = eventRecords;
+        }
+
+        internal IEnumerable<StoredEvent> ToStoredEvents()
+        {
+            var version = AggregateRecord.Version;
+
+            return EventRecords.Select(eventRecord =>
+                new StoredEvent
+                {
+                    AggregateId = AggregateRecord.Id,
+                    Aggregate = AggregateRecord.Name,
+                    Version = ++version,
+                    CreatedAt = eventRecord.CreatedAt,
+                    Payload = PayloadSerializer.Serialize(eventRecord.Event),
+                    Id = eventRecord.Id,
+                    Name = eventRecord.Event.GetType().Name
+                });
+        }
+    }
+}

Tacta.EventStore/Repository/EventStoreRepository.cs

@@ -6,6 +6,7 @@
 using Dapper;
 using Newtonsoft.Json;
 using Tacta.Connection;
+using Tacta.EventStore.Domain;
 using Tacta.EventStore.Repository.Exceptions;
 
 namespace Tacta.EventStore.Repository
@@ -53,7 +54,7 @@ public async Task SaveAsync<T>(AggregateRecord aggregateRecord,
                 Aggregate = aggregateRecord.Name,
                 Version = ++version,
                 CreatedAt = eventRecord.CreatedAt,
-                Payload = JsonConvert.SerializeObject(eventRecord.Event, Formatting.Indented, _jsonSerializerSettings),
+                Payload = PayloadSerializer.Serialize(eventRecord.Event),
                 Id = eventRecord.Id,
                 Name = eventRecord.Event.GetType().Name
             });
@@ -75,6 +76,44 @@ await _sqlConnectionFactory.ExecuteWithTransactionIfExists(async (connection, tr
             }
         }
 
+        public async Task SaveAsync(IReadOnlyCollection<Aggregate> aggregates,
+            CancellationToken cancellationToken = default)
+        {
+            if (aggregates == null || !aggregates.Any()) return;
+
+            var records = aggregates.SelectMany(aggregate => aggregate.ToStoredEvents());
+
+            try
+            {
+                await _sqlConnectionFactory.ExecuteWithTransactionIfExists(async (connection, transaction) =>
+                {
+                    await connection.ExecuteAsync(StoredEvent.InsertQuery, records, transaction).ConfigureAwait(false);
+                }, cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception e)
+            {
+                if ((e is System.Data.SqlClient.SqlException || e is Microsoft.Data.SqlClient.SqlException) &&
+                    e.Message.Contains("ConcurrencyCheckIndex"))
+                    throw new ConcurrencyCheckException(e.Message);
+
+                throw;
+            }
+        }
+
+        public async Task SaveAsync<T, TId>(T aggregateRoot) where T : AggregateRoot<TId> where TId : EntityId
+        {
+            var aggregate = new Aggregate(aggregateRoot);
+
+            await SaveAsync(aggregate.AggregateRecord, aggregate.EventRecords).ConfigureAwait(false);
+        }
+
+        public async Task SaveAsync<T, TId>(IEnumerable<T> aggregateRoots) where T : AggregateRoot<TId> where TId : EntityId
+        {
+            var aggregates = aggregateRoots.Select(ar => new Aggregate(ar)).ToList().AsReadOnly();
+
+            await SaveAsync(aggregates).ConfigureAwait(false);
+        }
+
         public async Task<IReadOnlyCollection<EventStoreRecord<T>>> GetAsync<T>(string aggregateId, CancellationToken cancellationToken = default)
         {
             if (string.IsNullOrWhiteSpace(aggregateId))
@@ -149,7 +188,7 @@ public async Task<IReadOnlyCollection<EventStoreRecord<T>>> GetAsync<T>(string q
 
                 return storedEvents.Select(@event => new EventStoreRecord<T>
                 {
-                    Event = JsonConvert.DeserializeObject<T>(@event.Payload, _jsonSerializerSettings),
+                    Event = PayloadSerializer.Deserialize<T>(@event),
                     AggregateId = @event.AggregateId,
                     CreatedAt = @event.CreatedAt,
                     Id = @event.Id,

Tacta.EventStore/Repository/PayloadSerializer.cs

@@ -0,0 +1,24 @@
+using Newtonsoft.Json;
+
+namespace Tacta.EventStore.Repository
+{
+    public static class PayloadSerializer
+    {
+        private static readonly JsonSerializerSettings JsonSerializerSettings = new JsonSerializerSettings
+        {
+            TypeNameHandling = TypeNameHandling.All,
+            NullValueHandling = NullValueHandling.Ignore,
+            MetadataPropertyHandling = MetadataPropertyHandling.ReadAhead
+        };
+
+        public static string Serialize<T>(T @event)
+        {
+            return JsonConvert.SerializeObject(@event, Formatting.Indented, JsonSerializerSettings);
+        }
+
+        internal static T Deserialize<T>(StoredEvent @event)
+        {
+            return JsonConvert.DeserializeObject<T>(@event.Payload, JsonSerializerSettings);
+        }
+    }
+}

Tacta.EventStore/Tacta.EventStore.csproj

@@ -12,10 +12,10 @@
     <PackageIcon></PackageIcon>
     <PackageProjectUrl>https://tacta.io/</PackageProjectUrl>
     <PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>
-    <Version>1.5.1</Version>
-    <PackageVersion>1.5.1</PackageVersion>
+    <Version>1.5.2</Version>
+    <PackageVersion>1.5.2</PackageVersion>
     <Title>Tacta EventStore Library</Title>
-    <PackageReleaseNotes>Projections Status Monitoring Page</PackageReleaseNotes>
+    <PackageReleaseNotes>Saving multiple aggregates at once</PackageReleaseNotes>
     <Configurations>Debug;Release</Configurations>
     <Platforms>AnyCPU</Platforms>
   </PropertyGroup>