Added ElasticSearch left fold projections

Cleaned EventStoreDB README links
Minor updates in other places

Author: oskardudycz

Core.ElasticSearch/Config.cs

@@ -0,0 +1,32 @@
+using System;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Nest;
+
+namespace Core.ElasticSearch
+{
+    public class ElasticSearchConfig
+    {
+        public string Url { get; set; } = default!;
+        public string DefaultIndex { get; set; } = default!;
+    }
+    
+    public static class ElasticSearchConfigExtensions
+    {
+        private const string DefaultConfigKey = "ElasticSearch";
+        public static void AddElasticsearch(
+            this IServiceCollection services, IConfiguration configuration, Action<ConnectionSettings>? config = null)
+        {
+            var elasticSearchConfig = configuration.GetSection(DefaultConfigKey).Get<ElasticSearchConfig>();
+            
+            var settings = new ConnectionSettings(new Uri(elasticSearchConfig.Url))
+                .DefaultIndex(elasticSearchConfig.DefaultIndex);
+
+            config?.Invoke(settings);
+
+            var client = new ElasticClient(settings);
+
+            services.AddSingleton<IElasticClient>(client);
+        }
+    }
+}

Core.ElasticSearch/Core.ElasticSearch.csproj

@@ -10,6 +10,7 @@
     <ItemGroup>
         <PackageReference Include="NEST" Version="7.13.0" />
         <PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="5.0.0" />
+        <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="5.0.0" />
         <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="5.0.1" />
         <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
     </ItemGroup>

Core.ElasticSearch/ElasticSearchConfig.cs

@@ -0,0 +1,40 @@
+using System;
+using System.Collections.Concurrent;
+using System.Linq;
+
+namespace Core.ElasticSearch.Indices
+{
+    public class IndexNameMapper
+    {
+        private static readonly IndexNameMapper Instance = new();
+
+        private readonly ConcurrentDictionary<Type, string> typeNameMap = new();
+
+        public static void AddCustomMap<TStream>(string mappedStreamName) =>
+            AddCustomMap(typeof(TStream), mappedStreamName);
+
+        public static void AddCustomMap(Type streamType, string mappedStreamName)
+        {
+            Instance.typeNameMap.AddOrUpdate(streamType, mappedStreamName, (_, _) => mappedStreamName);
+        }
+
+        public static string ToIndexPrefix<TStream>() => ToIndexPrefix(typeof(TStream));
+
+        public static string ToIndexPrefix(Type streamType) => Instance.typeNameMap.GetOrAdd(streamType, (_) =>
+        {
+            var modulePrefix = streamType.Namespace!.Split(".").First();
+            return $"{modulePrefix}-{streamType.Name}".ToLower();
+        });
+
+        public static string ToIndexName<TStream>(object? tenantId = null) =>
+            ToIndexName(typeof(TStream));
+
+        public static string ToIndexName(Type streamType, object? tenantId = null)
+        {
+            var tenantPrefix = tenantId != null ? $"{tenantId}-"  : "";
+
+            return $"{tenantPrefix}{ToIndexPrefix(streamType)}".ToLower();
+        }
+
+    }
+}

Core.ElasticSearch/Indices/IndexNameMapper.cs

@@ -0,0 +1,62 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Core.ElasticSearch.Indices;
+using Core.Events;
+using Core.Projections;
+using MediatR;
+using Microsoft.Extensions.DependencyInjection;
+using Nest;
+
+namespace Core.ElasticSearch.Projections
+{
+    public class ElasticSearchProjection<TEvent, TView> : IEventHandler<TEvent>
+        where TView : class, IProjection
+        where TEvent : IEvent
+    {
+        private readonly IElasticClient elasticClient;
+        private readonly Func<TEvent, string> getId;
+
+        public ElasticSearchProjection(
+            IElasticClient elasticClient,
+            Func<TEvent, string> getId
+        )
+        {
+            this.elasticClient = elasticClient ?? throw new ArgumentNullException(nameof(elasticClient));
+            this.getId = getId ?? throw new ArgumentNullException(nameof(getId));
+        }
+
+        public async Task Handle(TEvent @event, CancellationToken ct)
+        {
+            string id = getId(@event);
+
+            var entity = (await elasticClient.GetAsync<TView>(id, ct: ct))?.Source
+                         ?? (TView) Activator.CreateInstance(typeof(TView), true)!;
+
+            entity.When(@event);
+
+            var result = await elasticClient.UpdateAsync<TView>(id,
+                u => u.Doc(entity).Upsert(entity).Index(IndexNameMapper.ToIndexName<TView>()),
+                ct
+            );
+        }
+    }
+
+    public static class ElasticSearchProjectionConfig
+    {
+        public static IServiceCollection Project<TEvent, TView>(this IServiceCollection services,
+            Func<TEvent, string> getId)
+            where TView : class, IProjection
+            where TEvent : IEvent
+        {
+            services.AddTransient<INotificationHandler<TEvent>>(sp =>
+            {
+                var session = sp.GetRequiredService<IElasticClient>();
+
+                return new ElasticSearchProjection<TEvent, TView>(session, getId);
+            });
+
+            return services;
+        }
+    }
+}

Core.ElasticSearch/Projections/ElasticSearchProjection.cs

@@ -1,19 +1,20 @@
 using System;
 using System.Threading;
 using System.Threading.Tasks;
-using Core.Aggregates;
+using Core.ElasticSearch.Indices;
 using Core.Events;
-using Core.Repositories;
+using Nest;
+using IAggregate = Core.Aggregates.IAggregate;
 
 namespace Core.ElasticSearch.Repository
 {
-    public class ElasticSearchRepository<T>: IRepository<T> where T : class, IAggregate, new()
+    public class ElasticSearchRepository<T>: Repositories.IRepository<T> where T : class, IAggregate, new()
     {
-        private readonly Nest.IElasticClient elasticClient;
+        private readonly IElasticClient elasticClient;
         private readonly IEventBus eventBus;
 
         public ElasticSearchRepository(
-            Nest.IElasticClient elasticClient,
+            IElasticClient elasticClient,
             IEventBus eventBus
         )
         {
@@ -29,12 +30,12 @@ IEventBus eventBus
 
         public Task Add(T aggregate, CancellationToken cancellationToken)
         {
-            return elasticClient.IndexAsync(aggregate, i => i.Id(aggregate.Id), cancellationToken);
+            return elasticClient.IndexAsync(aggregate, i => i.Id(aggregate.Id).Index(IndexNameMapper.ToIndexName<T>()), cancellationToken);
         }
 
         public Task Update(T aggregate, CancellationToken cancellationToken)
         {
-            return elasticClient.UpdateAsync<T>(aggregate.Id, i => i.Doc(aggregate), cancellationToken);
+            return elasticClient.UpdateAsync<T>(aggregate.Id, i => i.Doc(aggregate).Index(IndexNameMapper.ToIndexName<T>()), cancellationToken);
         }
 
         public Task Delete(T aggregate, CancellationToken cancellationToken)

Core.ElasticSearch/Repository/ElasticSearchRepository.cs

@@ -1,6 +1,5 @@
 using System;
 using Core.EventStoreDB.Subscriptions;
-using Core.Subscriptions;
 using EventStore.Client;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Configuration;

Core.EventStoreDB/Config.cs

@@ -1,6 +1,5 @@
 using System.Text;
 using Core.Events;
-using Core.Reflection;
 using EventStore.Client;
 using Newtonsoft.Json;
 

Core.EventStoreDB/Serialization/EventStoreDBSerializer.cs

@@ -4,7 +4,6 @@
 using System.Threading.Tasks;
 using Core.Events;
 using Core.EventStoreDB.Serialization;
-using Core.Subscriptions;
 using EventStore.Client;
 
 namespace Core.EventStoreDB.Subscriptions

Core.EventStoreDB/Subscriptions/EventStoreDBSubscriptionCheckpointRepository.cs

@@ -1,7 +1,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 
-namespace Core.Subscriptions
+namespace Core.EventStoreDB.Subscriptions
 {
     public interface ISubscriptionCheckpointRepository
     {