Add store item logic to EventStore

.github/workflows/build.yml

@@ -7,7 +7,7 @@ on:
     branches: [ "main" ]
 
 env:
-  PACKAGE_VERSION: 0.1.2
+  PACKAGE_VERSION: 0.1.3
 
 jobs:
 

CloudFabric.EventSourcing.AspNet/CloudFabric.EventSourcing.AspNet.CosmosDb/Extensions/ServiceCollectionExtensions.cs

@@ -14,7 +14,8 @@ public static IEventSourcingBuilder AddCosmosDbEventStore(
             string connectionString,
             CosmosClientOptions cosmosClientOptions,
             string databaseId,
-            string containerId,
+            string eventsContainerId,
+            string itemsContainerId,
             CosmosClient leaseClient,
             string leaseDatabaseId,
             string leaseContainerId,
@@ -23,13 +24,13 @@ string processorName
         {
             var cosmosClient = new CosmosClient(connectionString, cosmosClientOptions);
 
-            var eventStore = new CosmosDbEventStore(cosmosClient, databaseId, containerId);
+            var eventStore = new CosmosDbEventStore(cosmosClient, databaseId, eventsContainerId, itemsContainerId);
             eventStore.Initialize().Wait();
 
             var eventStoreObserver = new CosmosDbEventStoreChangeFeedObserver(
                 cosmosClient,
                 databaseId,
-                containerId,
+                eventsContainerId,
                 leaseClient,
                 leaseDatabaseId,
                 leaseContainerId,
@@ -52,10 +53,11 @@ public static IEventSourcingBuilder AddCosmosDbEventStore(
             this IServiceCollection services,
             CosmosClient client,
             string databaseId,
-            string containerId
+            string eventsContainerId,
+            string itemsContainerId
         )
         {
-            var eventStore = new CosmosDbEventStore(client, databaseId, containerId);
+            var eventStore = new CosmosDbEventStore(client, databaseId, eventsContainerId, itemsContainerId);
             eventStore.Initialize().Wait();
 
             return new EventSourcingBuilder

CloudFabric.EventSourcing.AspNet/CloudFabric.EventSourcing.AspNet.InMemory/Extensions/ServiceCollectionExtensions.cs

@@ -11,10 +11,11 @@ public static class ServiceCollectionExtensions
     {
         public static IEventSourcingBuilder AddInMemoryEventStore(
             this IServiceCollection services,
-            Dictionary<(Guid, string), List<string>> eventsContainer
+            Dictionary<(Guid, string), List<string>> eventsContainer,
+            Dictionary<(string, string), string> itemsContainer
         )
         {
-            var eventStore = new InMemoryEventStore(eventsContainer);
+            var eventStore = new InMemoryEventStore(eventsContainer, itemsContainer);
             eventStore.Initialize().Wait();
 
             // add events observer for projections

CloudFabric.EventSourcing.AspNet/CloudFabric.EventSourcing.AspNet.Postgresql/Extensions/ServiceCollectionExtensions.cs

@@ -12,10 +12,11 @@ public static class ServiceCollectionExtensions
         public static IEventSourcingBuilder AddPostgresqlEventStore(
             this IServiceCollection services,
             string eventsConnectionString,
-            string tableName
+            string eventTableName,
+            string itemTableName
         )
         {
-            var eventStore = new PostgresqlEventStore(eventsConnectionString, tableName);
+            var eventStore = new PostgresqlEventStore(eventsConnectionString, eventTableName, itemTableName);
             eventStore.Initialize().Wait();
 
             // add events observer for projections

CloudFabric.EventSourcing.EventStore/EventSerializerOptions.cs → CloudFabric.EventSourcing.EventStore/EventStoreSerializerOptions.cs

@@ -2,7 +2,7 @@
 
 namespace CloudFabric.EventSourcing.EventStore;
 
-public static class EventSerializerOptions
+public static class EventStoreSerializerOptions
 {
     public static JsonSerializerOptions Options
     {

CloudFabric.EventSourcing.EventStore/IEventStore.cs

@@ -23,4 +23,8 @@ Task<bool> AppendToStreamAsync(
     Task DeleteAll(CancellationToken cancellationToken = default);
 
     Task<bool> HardDeleteAsync(Guid streamId, string partitionKey, CancellationToken cancellationToken = default);
+
+    Task UpsertItem<T>(string id, string partitionKey, T item, CancellationToken cancellationToken = default);
+
+    Task<T?> LoadItem<T>(string id, string partitionKey, CancellationToken cancellationToken = default);
 }

CloudFabric.EventSourcing.EventStore/Persistence/EventWrapper.cs

@@ -36,7 +36,7 @@ public IEvent GetEvent()
                 throw new Exception("Couldn't find event type. Make sure it's in correct namespace.");
             }
 
-            var e = (IEvent?)EventData.Deserialize(eventType, EventSerializerOptions.Options);
+            var e = (IEvent?)EventData.Deserialize(eventType, EventStoreSerializerOptions.Options);
 
             if (e == null)
             {

CloudFabric.EventSourcing.EventStore/Persistence/ItemWrapper.cs

@@ -0,0 +1,15 @@
+using System.Text.Json.Serialization;
+
+namespace CloudFabric.EventSourcing.EventStore.Persistence;
+
+public class ItemWrapper
+{
+    [JsonPropertyName("id")]
+    public string? Id { get; set; }
+
+    [JsonPropertyName("partitionKey")]
+    public string? PartitionKey { get; set; }
+
+    [JsonPropertyName("item_data")]
+    public string? ItemData { get; set; }
+}

CloudFabric.EventSourcing.Tests/Domain/Item.cs

@@ -0,0 +1,19 @@
+namespace CloudFabric.EventSourcing.Tests.Domain;
+
+public class Item
+{
+    public Guid Id { get; set; }
+
+    public string Name { get; set; }
+
+    public Dictionary<string, NestedItemClass> Properties { get; set; }
+}
+
+public class NestedItemClass
+{
+    public Guid Id { get; set; } = Guid.NewGuid();
+
+    public string Name { get; set; } = nameof(Name).ToLowerInvariant();
+
+    public DateTime date { get; } = DateTime.MinValue;
+}

CloudFabric.EventSourcing.Tests/ItemTests.cs

@@ -0,0 +1,104 @@
+using CloudFabric.EventSourcing.EventStore;
+using CloudFabric.EventSourcing.Tests.Domain;
+using FluentAssertions;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace CloudFabric.EventSourcing.Tests;
+public abstract class ItemTests
+{
+    protected abstract Task<IEventStore> GetEventStore();
+    private IEventStore _eventStore;
+
+    protected TimeSpan ProjectionsUpdateDelay { get; set; } = TimeSpan.FromMilliseconds(1000);
+
+    [TestInitialize]
+    public async Task Initialize()
+    {
+        _eventStore = await GetEventStore();
+    }
+
+    [TestCleanup]
+    public async Task Cleanup()
+    {
+        await _eventStore.DeleteAll();
+    }
+
+    [TestMethod]
+    public async Task SaveItem()
+    {
+        var item = new Item
+        {
+            Id = Guid.NewGuid(),
+            Name = "Item1",
+            Properties = new Dictionary<string, NestedItemClass>
+            {
+                { Guid.NewGuid().ToString(), new NestedItemClass() },
+                { Guid.NewGuid().ToString(), new NestedItemClass() }
+            }
+        };
+
+        Func<Task> action = async () => await _eventStore.UpsertItem($"{item.Id}{item.Name}", $"{item.Id}{item.Name}", item);
+        await action.Should().NotThrowAsync();
+    }
+
+    [TestMethod]
+    public async Task LoadItem()
+    {
+        var item = new Item
+        {
+            Id = Guid.NewGuid(),
+            Name = "Item1",
+            Properties = new Dictionary<string, NestedItemClass>
+            {
+                { Guid.NewGuid().ToString(), new NestedItemClass() }
+            }
+        };
+
+        await _eventStore.UpsertItem($"{item.Id}{item.Name}", $"{item.Id}{item.Name}", item);
+
+        var loadedItem = await _eventStore.LoadItem<Item>($"{item.Id}{item.Name}", $"{item.Id}{item.Name}");
+
+        loadedItem.Id.Should().Be(item.Id);
+        loadedItem.Name.Should().Be(item.Name);
+        loadedItem.Properties.Keys.Should().BeEquivalentTo(item.Properties.Keys);
+        loadedItem.Properties.Values.Should().BeEquivalentTo(item.Properties.Values);
+    }
+
+    [TestMethod]
+    public async Task LoadItem_NullIfNotFound()
+    {
+        var loadedItem = await _eventStore.LoadItem<Item>($"{Guid.NewGuid()}", $"{Guid.NewGuid()}");
+
+        loadedItem.Should().BeNull();
+    }
+
+    [TestMethod]
+    public async Task UpdateItem()
+    {
+        var item = new Item
+        {
+            Id = Guid.NewGuid(),
+            Name = "Item1",
+            Properties = new Dictionary<string, NestedItemClass>
+            {
+                { Guid.NewGuid().ToString(), new NestedItemClass() },
+                { Guid.NewGuid().ToString(), new NestedItemClass() }
+            }
+        };
+
+        await _eventStore.UpsertItem($"{item.Id}{item.Name}", $"{item.Id}{item.Name}", item);
+
+        string propertyGuid = Guid.NewGuid().ToString();
+
+        item.Properties = new()
+        {
+            { propertyGuid, new NestedItemClass() }
+        };
+
+        Func<Task> action = async () => await _eventStore.UpsertItem($"{item.Id}{item.Name}", $"{item.Id}{item.Name}", item);
+        await action.Should().NotThrowAsync();
+
+        var updatedItem = await _eventStore.LoadItem<Item>($"{item.Id}{item.Name}", $"{item.Id}{item.Name}");
+        updatedItem.Properties.ContainsKey(propertyGuid).Should().BeTrue();
+    }
+}

Demos/ToDoList.Api/Program.cs

@@ -49,7 +49,7 @@
 
 #region User Accounts Projections
 
-var userEventSourcingBuilder = builder.Services.AddPostgresqlEventStore(builder.Configuration.GetConnectionString("Default"), "user-events")
+var userEventSourcingBuilder = builder.Services.AddPostgresqlEventStore(builder.Configuration.GetConnectionString("Default"), "user-events", "user-items")
     .AddRepository<AggregateRepository<UserAccount>>()
     .AddRepository<AggregateRepository<UserAccountEmailAddress>>()
     .AddPostgresqlProjections(
@@ -61,7 +61,7 @@
 
 #region Task Lists Projections
 
-var taskListEventSourcingBuilder = builder.Services.AddPostgresqlEventStore(builder.Configuration.GetConnectionString("Default"), "task-list-events")
+var taskListEventSourcingBuilder = builder.Services.AddPostgresqlEventStore(builder.Configuration.GetConnectionString("Default"), "task-list-events", "task-list-items")
     .AddRepository<AggregateRepository<TaskList>>()
     .AddPostgresqlProjections(
         builder.Configuration.GetConnectionString("Default"),
@@ -72,7 +72,7 @@
 
 #region Task Projections
 
-var taskEventSourcingBuilder = builder.Services.AddPostgresqlEventStore(builder.Configuration.GetConnectionString("Default"), "task-events")
+var taskEventSourcingBuilder = builder.Services.AddPostgresqlEventStore(builder.Configuration.GetConnectionString("Default"), "task-events", "task-items")
     .AddRepository<AggregateRepository<ToDoList.Domain.Task>>()
     .AddPostgresqlProjections(
         builder.Configuration.GetConnectionString("Default"),