Projects saved

src/PinkParrot/Configurations/InfrastructureDependencies.cs

@@ -10,9 +10,12 @@ using System.Net;
 using Autofac;
 using EventStore.ClientAPI;
 using EventStore.ClientAPI.SystemData;
+using Microsoft.AspNetCore.Builder;
+using MongoDB.Driver;
 using PinkParrot.Infrastructure.CQRS.Autofac;
 using PinkParrot.Infrastructure.CQRS.Commands;
 using PinkParrot.Infrastructure.CQRS.EventStore;
+using PinkParrot.Read.Services.Implementations;
 
 namespace PinkParrot.Configurations
 {
@@ -29,9 +32,21 @@ namespace PinkParrot.Configurations
                        .KeepRetrying(),
                    new IPEndPoint(IPAddress.Loopback, 1113));
 
+            var mongoDbClient = new MongoClient("mongodb://localhost");
+            var mongoDatabase = mongoDbClient.GetDatabase("PinkParrot");
+
             eventStore.ConnectAsync().Wait();
 
             builder.RegisterInstance(new UserCredentials("admin", "changeit"))
+                .AsSelf()
+                .SingleInstance();
+
+            builder.RegisterInstance(mongoDatabase)
+                .As<IMongoDatabase>()
+                .SingleInstance();
+
+            builder.RegisterType<MongoStreamPositionsStorage>()
+                .As<IStreamPositionStorage>()
                 .SingleInstance();
 
             builder.RegisterType<AutofacDomainObjectFactory>()
@@ -53,6 +68,18 @@ namespace PinkParrot.Configurations
             builder.RegisterType<InMemoryCommandBus>()
                 .As<ICommandBus>()
                 .SingleInstance();
+
+            builder.RegisterType<EventStoreBus>()
+                .AsSelf()
+                .SingleInstance();
+        }
+    }
+
+    public static class InfrastructureDependencie
+    {
+        public static void UseAppEventBus(this IApplicationBuilder app)
+        {
+            app.ApplicationServices.GetService(typeof(EventStoreBus));
         }
     }
 }

src/PinkParrot/Configurations/ReadDependencies.cs

@@ -7,6 +7,9 @@
 // ==========================================================================
 
 using Autofac;
+using PinkParrot.Infrastructure.CQRS.Events;
+using PinkParrot.Read.Repositories;
+using PinkParrot.Read.Repositories.Implementations;
 using PinkParrot.Read.Services;
 using PinkParrot.Read.Services.Implementations;
 
@@ -16,8 +19,17 @@ namespace PinkParrot.Configurations
     {
         protected override void Load(ContainerBuilder builder)
         {
-            builder.RegisterType<SchemaProvider>()
+            builder.RegisterType<TenantProvider>()
-                .As<ISchemaProvider>()
+                .As<ITenantProvider>()
+                .SingleInstance();
+
+            builder.RegisterType<ModelSchemaProvider>()
+                .As<IModelSchemaProvider>()
+                .SingleInstance();
+
+            builder.RegisterType<MongoModelSchemaRepository>()
+                .As<IModelSchemaRepository>()
+                .As<ICatchEventConsumer>()
                 .SingleInstance();
         }
     }

src/PinkParrot/Configurations/Serializers.cs

@@ -7,12 +7,15 @@
 // ==========================================================================
 
 using System.Reflection;
-using Microsoft.Extensions.DependencyInjection;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Serialization;
 using PinkParrot.Core.Schema;
 using PinkParrot.Infrastructure.CQRS.EventStore;
 using PinkParrot.Infrastructure.Json;
+using IMvcBuilder = Microsoft.Extensions.DependencyInjection.IMvcBuilder;
+using IServiceCollection = Microsoft.Extensions.DependencyInjection.IServiceCollection;
+using MvcJsonMvcBuilderExtensions = Microsoft.Extensions.DependencyInjection.MvcJsonMvcBuilderExtensions;
+using ServiceCollectionServiceExtensions = Microsoft.Extensions.DependencyInjection.ServiceCollectionServiceExtensions;
 
 namespace PinkParrot.Configurations
 {
@@ -41,14 +44,14 @@ namespace PinkParrot.Configurations
                 new ModelFieldFactory()
                     .AddFactory<NumberFieldProperties>(id => new NumberField(id));
 
-            services.AddSingleton(t => CreateSettings());
+            ServiceCollectionServiceExtensions.AddSingleton(services, t => CreateSettings());
-            services.AddSingleton(fieldFactory);
+            ServiceCollectionServiceExtensions.AddSingleton(services, fieldFactory);
-            services.AddSingleton<EventStoreParser>();
+            ServiceCollectionServiceExtensions.AddSingleton<EventStoreFormatter>(services);
         }
 
         public static void AddAppSerializers(this IMvcBuilder mvc)
         {
-            mvc.AddJsonOptions(options =>
+            MvcJsonMvcBuilderExtensions.AddJsonOptions(mvc, options =>
             {
                 ConfigureJson(options.SerializerSettings);
             });

src/PinkParrot/Configurations/WriteDependencies.cs

@@ -8,6 +8,7 @@
 
 using Autofac;
 using PinkParrot.Infrastructure.CQRS.Commands;
+using PinkParrot.Pipeline.CommandHandlers;
 using PinkParrot.Write.Schema;
 
 namespace PinkParrot.Configurations
@@ -16,11 +17,20 @@ namespace PinkParrot.Configurations
     {
         protected override void Load(ContainerBuilder builder)
         {
+            builder.RegisterType<EnrichWithAggregateIdHandler>()
+                .As<ICommandHandler>()
+                .SingleInstance();
+
+            builder.RegisterType<EnrichWithTenantIdHandler>()
+                .As<ICommandHandler>()
+                .SingleInstance();
+
             builder.RegisterType<ModelSchemaCommandHandler>()
                 .As<ICommandHandler>()
                 .SingleInstance();
 
             builder.RegisterType<ModelSchemaDomainObject>()
+                .AsSelf()
                 .InstancePerDependency();
         }
     }

src/PinkParrot/Pipeline/CommandHandlers/EnrichWithAggregateIdHandler.cs

@@ -0,0 +1,50 @@
+// ==========================================================================
+//  EnrichWithAggregateIdHandler.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Mvc.Infrastructure;
+using PinkParrot.Infrastructure.CQRS.Commands;
+using PinkParrot.Read.Services;
+
+// ReSharper disable InvertIf
+
+namespace PinkParrot.Pipeline.CommandHandlers
+{
+    public sealed class EnrichWithAggregateIdHandler : ICommandHandler
+    {
+        private readonly IModelSchemaProvider modelSchemaProvider;
+        private readonly IActionContextAccessor actionContextAccessor;
+
+        public EnrichWithAggregateIdHandler(IModelSchemaProvider modelSchemaProvider, IActionContextAccessor actionContextAccessor)
+        {
+            this.modelSchemaProvider = modelSchemaProvider;
+
+            this.actionContextAccessor = actionContextAccessor;
+        }
+
+        public async Task<bool> HandleAsync(CommandContext context)
+        {
+            var aggregateCommand = context.Command as IAggregateCommand;
+
+            if (aggregateCommand != null && aggregateCommand.AggregateId == Guid.Empty)
+            {
+                var routeValues = actionContextAccessor.ActionContext.RouteData.Values;
+
+                if (routeValues.ContainsKey("name"))
+                {
+                    var schemeName = routeValues["name"];
+
+                    aggregateCommand.AggregateId = await modelSchemaProvider.FindSchemaIdByNameAsync(schemeName.ToString());
+                }
+            }
+
+            return false;
+        }
+    }
+}

src/PinkParrot/Pipeline/CommandHandlers/EnrichWithTenantIdHandler.cs

@@ -0,0 +1,45 @@
+// ==========================================================================
+//  EnrichWithTenantIdHandler.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Http;
+using PinkParrot.Infrastructure.CQRS.Commands;
+using PinkParrot.Read.Services;
+using PinkParrot.Write;
+
+// ReSharper disable InvertIf
+
+namespace PinkParrot.Pipeline.CommandHandlers
+{
+    public sealed class EnrichWithTenantIdHandler : ICommandHandler
+    {
+        private readonly ITenantProvider tenantProvider;
+        private readonly IHttpContextAccessor httpContextAccessor;
+
+        public EnrichWithTenantIdHandler(ITenantProvider tenantProvider, IHttpContextAccessor httpContextAccessor)
+        {
+            this.tenantProvider = tenantProvider;
+
+            this.httpContextAccessor = httpContextAccessor;
+        }
+
+        public async Task<bool> HandleAsync(CommandContext context)
+        {
+            var tenantCommand = context.Command as ITenantCommand;
+
+            if (tenantCommand != null)
+            {
+                var domain = httpContextAccessor.HttpContext.Request.Host.ToString();
+
+                tenantCommand.TenantId = await tenantProvider.ProvideTenantIdByDomainAsync(domain);
+            }
+
+            return false;
+        }
+    }
+}

src/PinkParrot/Pipeline/CommandHandlers/LogExceptionHandler.cs

@@ -0,0 +1,40 @@
+// ==========================================================================
+//  LogExceptionHandler.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using PinkParrot.Infrastructure.CQRS.Commands;
+
+// ReSharper disable InvertIf
+
+namespace PinkParrot.Pipeline.CommandHandlers
+{
+    public sealed class LogExceptionHandler : ICommandHandler
+    {
+        private readonly ILogger<LogExceptionHandler> logger;
+
+        public LogExceptionHandler(ILogger<LogExceptionHandler> logger)
+        {
+            this.logger = logger;
+        }
+
+        public Task<bool> HandleAsync(CommandContext context)
+        {
+            var exception = context.Exception;
+
+            if (exception != null)
+            {
+                var eventId = new EventId(9999, "CommandFailed");
+
+                logger.LogError(eventId, exception, "Handling {0} command failed", context.Command);
+            }
+
+            return Task.FromResult(false);
+        }
+    }
+}

src/PinkParrot/Pipeline/CommandHandlers/LogExecutingHandler.cs

@@ -0,0 +1,31 @@
+// ==========================================================================
+//  LogExecutingHandler.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using PinkParrot.Infrastructure.CQRS.Commands;
+
+namespace PinkParrot.Pipeline.CommandHandlers
+{
+    public sealed class LogExecutingHandler : ICommandHandler
+    {
+        private readonly ILogger<LogExecutingHandler> logger;
+
+        public LogExecutingHandler(ILogger<LogExecutingHandler> logger)
+        {
+            this.logger = logger;
+        }
+
+        public Task<bool> HandleAsync(CommandContext context)
+        {
+            logger.LogError("Handling {0} command", context.Command);
+
+            return Task.FromResult(false);
+        }
+    }
+}

src/PinkParrot/Startup.cs

@@ -44,6 +44,7 @@ namespace PinkParrot
             app.UseMvc();
             app.UseStaticFiles();
             app.UseAppSwagger();
+            app.UseAppEventBus();
 
             if (env.IsDevelopment())
             {

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/CommonHeaders.cs

@@ -12,6 +12,7 @@ namespace PinkParrot.Infrastructure.CQRS
         public const string AggregateId = "AggregateId";
         public const string CommitId = "CommitId";
         public const string Timestamp = "Timestamp";
+        public const string TenantId = "TenantId";
         public const string EventId = "EventId";
         public const string EventType = "EventType";
         public const string EventNumber = "EventNumber";

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/DomainObject.cs

@@ -58,7 +58,7 @@ namespace PinkParrot.Infrastructure.CQRS
         {
             Guard.NotNull(@event, nameof(@event));
 
-            var envelopeToAdd = EnvelopeFactory.ForEvent(@event, id);
+            var envelopeToAdd = EnvelopeFactory.ForEvent(@event, this);
 
             uncomittedEvents.Add(envelopeToAdd);
 

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/Envelope.cs

@@ -5,7 +5,6 @@
 //  Copyright (c) PinkParrot Group
 //  All rights reserved.
 // ==========================================================================
-
 namespace PinkParrot.Infrastructure.CQRS
 {
     public class Envelope<TPayload> where TPayload : class

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/EnvelopeExtensions.cs

@@ -62,6 +62,18 @@ namespace PinkParrot.Infrastructure.CQRS
             return envelope;
         }
 
+        public static Guid TenantId(this EnvelopeHeaders headers)
+        {
+            return headers[CommonHeaders.TenantId].ToGuid(CultureInfo.InvariantCulture);
+        }
+
+        public static Envelope<T> SetTenantId<T>(this Envelope<T> envelope, Guid value) where T : class
+        {
+            envelope.Headers.Set(CommonHeaders.TenantId, value);
+
+            return envelope;
+        }
+
         public static Instant Timestamp(this EnvelopeHeaders headers)
         {
             return headers[CommonHeaders.Timestamp].ToInstant(CultureInfo.InvariantCulture);

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/EnvelopeFactory.cs

@@ -14,12 +14,24 @@ namespace PinkParrot.Infrastructure.CQRS
 {
     public static class EnvelopeFactory
     {
-        public static Envelope<IEvent> ForEvent(IEvent @event, Guid aggregateId)
+        public static Envelope<IEvent> ForEvent(IEvent @event, IAggregate aggregate)
         {
-            return new Envelope<IEvent>(@event)
+            var eventId = Guid.NewGuid();
-                .SetAggregateId(aggregateId)
+
-                .SetEventId(aggregateId)
+            var envelope =
-                .SetTimestamp(SystemClock.Instance.GetCurrentInstant());
+                new Envelope<IEvent>(@event)
+                    .SetAggregateId(aggregate.Id)
+                    .SetEventId(eventId)
+                    .SetTimestamp(SystemClock.Instance.GetCurrentInstant());
+
+            var tenantAggregate = aggregate as ITenantAggregate;
+
+            if (tenantAggregate != null)
+            {
+                envelope = envelope.SetTenantId(tenantAggregate.TenantId);
+            }
+
+            return envelope;
         }
     }
 }

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/EventStore/EventStoreBus.cs

@@ -0,0 +1,128 @@
+// ==========================================================================
+//  EventStoreBus.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using EventStore.ClientAPI;
+using EventStore.ClientAPI.SystemData;
+using Microsoft.Extensions.Logging;
+using PinkParrot.Infrastructure.CQRS.Events;
+
+namespace PinkParrot.Infrastructure.CQRS.EventStore
+{
+    public sealed class EventStoreBus
+    {
+        private readonly IEventStoreConnection connection;
+        private readonly UserCredentials credentials;
+        private readonly EventStoreFormatter formatter;
+        private readonly IEnumerable<ILiveEventConsumer> liveConsumers;
+        private readonly IEnumerable<ICatchEventConsumer> catchConsumers;
+        private readonly ILogger<EventStoreBus> logger;
+        private readonly IStreamPositionStorage positions;
+        private EventStoreAllCatchUpSubscription catchSubscription;
+
+        public EventStoreBus(
+            ILogger<EventStoreBus> logger,
+            IEnumerable<ILiveEventConsumer> liveConsumers,
+            IEnumerable<ICatchEventConsumer> catchConsumers,
+            IStreamPositionStorage positions,
+            IEventStoreConnection connection,
+            UserCredentials credentials,
+            EventStoreFormatter formatter)
+        {
+            Guard.NotNull(logger, nameof(logger));
+            Guard.NotNull(formatter, nameof(formatter));
+            Guard.NotNull(positions, nameof(positions));
+            Guard.NotNull(connection, nameof(connection));
+            Guard.NotNull(credentials, nameof(credentials));
+            Guard.NotNull(liveConsumers, nameof(liveConsumers));
+            Guard.NotNull(catchConsumers, nameof(catchConsumers));
+
+            this.logger = logger;
+            this.formatter = formatter;
+            this.positions = positions;
+            this.connection = connection;
+            this.credentials = credentials;
+            this.liveConsumers = liveConsumers;
+            this.catchConsumers = catchConsumers;
+
+            Subscribe();
+        }
+
+        private void Subscribe()
+        {
+            var position = positions.ReadPosition();
+
+            var now = DateTime.UtcNow;
+
+            logger.LogInformation($"Subscribing from: {0}", position);
+
+            var settings =
+                new CatchUpSubscriptionSettings(
+                    int.MaxValue, 4096,
+                    true,
+                    true);
+
+            catchSubscription = connection.SubscribeToAllFrom(position, settings, (s, resolvedEvent) =>
+            {
+                var requireUpdate = false;
+
+                Debug.WriteLine($"Last Position: {catchSubscription.LastProcessedPosition}");
+                try
+                {
+                    if (resolvedEvent.OriginalEvent.EventStreamId.StartsWith("$", StringComparison.OrdinalIgnoreCase))
+                    {
+                        return;
+                    }
+
+                    if (liveConsumers.Any() || catchConsumers.Any())
+                    {
+                        requireUpdate = true;
+
+                        var @event = formatter.Parse(resolvedEvent);
+
+                        if (resolvedEvent.Event.Created > now)
+                        {
+                            Dispatch(liveConsumers, @event);
+                        }
+
+                        Dispatch(catchConsumers, @event);
+                    }
+
+                    requireUpdate = requireUpdate || catchSubscription.LastProcessedPosition.CommitPosition % 2 == 0;
+                }
+                finally
+                {
+                    if (requireUpdate)
+                    {
+                        positions.WritePosition(catchSubscription.LastProcessedPosition);
+                    }
+                }
+            }, userCredentials: credentials);
+        }
+
+        private void Dispatch(IEnumerable<IEventConsumer> consumers, Envelope<IEvent> @event)
+        {
+            foreach (var consumer in consumers)
+            {
+                try
+                {
+                    consumer.On(@event);
+                }
+                catch (Exception ex)
+                {
+                    var eventId = new EventId(10001, "EventConsumeFailed");
+
+                    logger.LogError(eventId, ex, "'{0}' failed to handle event {1} ({2})", consumer, @event.Payload, @event.Headers.EventId());
+                }
+            }
+        }
+    }
+}

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/EventStore/EventStoreDomainObjectRepository.cs

@@ -1,5 +1,5 @@
 // ==========================================================================
-//  GetEventStoreDomainObjectRepository.cs
+//  EventStoreDomainObjectRepository.cs
 //  PinkParrot Headless CMS
 // ==========================================================================
 //  Copyright (c) PinkParrot Group
@@ -14,8 +14,8 @@ using System.Threading.Tasks;
 using EventStore.ClientAPI;
 using EventStore.ClientAPI.SystemData;
 using PinkParrot.Infrastructure.CQRS.Commands;
-// ReSharper disable RedundantAssignment
 
+// ReSharper disable RedundantAssignment
 // ReSharper disable ConvertIfStatementToSwitchStatement
 // ReSharper disable TooWideLocalVariableScope
 
@@ -29,14 +29,14 @@ namespace PinkParrot.Infrastructure.CQRS.EventStore
         private readonly IStreamNameResolver nameResolver;
         private readonly IDomainObjectFactory factory;
         private readonly UserCredentials credentials;
-        private readonly EventStoreParser formatter;
+        private readonly EventStoreFormatter formatter;
 
         public EventStoreDomainObjectRepository(
             IDomainObjectFactory factory, 
             IStreamNameResolver nameResolver,
             IEventStoreConnection connection,
             UserCredentials credentials,
-            EventStoreParser formatter)
+            EventStoreFormatter formatter)
         {
             Guard.NotNull(factory, nameof(factory));
             Guard.NotNull(formatter, nameof(formatter));
@@ -106,28 +106,28 @@ namespace PinkParrot.Infrastructure.CQRS.EventStore
 
             var newEvents = domainObject.GetUncomittedEvents();
 
-            var currVersion = domainObject.Version;
+            var versionCurrent = domainObject.Version;
-            var prevVersion = currVersion - newEvents.Count;
+            var versionPrevious = versionCurrent - newEvents.Count;
-            var exptVersion = prevVersion == 0 ? ExpectedVersion.NoStream : prevVersion - 1;
+            var versionExpected = versionPrevious == 0 ? ExpectedVersion.NoStream : versionPrevious - 1;
 
             var eventsToSave = newEvents.Select(x => formatter.ToEventData(x, commitId)).ToList();
 
-            await InsertEventsAsync(streamName, exptVersion, eventsToSave);
+            await InsertEventsAsync(streamName, versionExpected, eventsToSave);
 
             domainObject.ClearUncommittedEvents();
         }
 
-        private async Task InsertEventsAsync(string streamName, int exptVersion, IReadOnlyCollection<EventData> eventsToSave)
+        private async Task InsertEventsAsync(string streamName, int expectedVersion, IReadOnlyCollection<EventData> eventsToSave)
         {
             if (eventsToSave.Count > 0)
             {
                 if (eventsToSave.Count < WritePageSize)
                 {
-                    await connection.AppendToStreamAsync(streamName, exptVersion, eventsToSave, credentials);
+                    await connection.AppendToStreamAsync(streamName, expectedVersion, eventsToSave, credentials);
                 }
                 else
                 {
-                    var transaction = await connection.StartTransactionAsync(streamName, exptVersion, credentials);
+                    var transaction = await connection.StartTransactionAsync(streamName, expectedVersion, credentials);
 
                     try
                     {
@@ -146,7 +146,7 @@ namespace PinkParrot.Infrastructure.CQRS.EventStore
             }
             else
             {
-                Debug.WriteLine(string.Format("No events to insert for: {0}", streamName), "GetEventStoreRepository");
+                Debug.WriteLine($"No events to insert for: {streamName}", "GetEventStoreRepository");
             }
         }
     }

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/EventStore/EventStoreFormatter.cs

@@ -16,11 +16,11 @@ using PinkParrot.Infrastructure.CQRS.Events;
 
 namespace PinkParrot.Infrastructure.CQRS.EventStore
 {
-    public class EventStoreParser
+    public class EventStoreFormatter
     {
         private readonly JsonSerializerSettings serializerSettings;
 
-        public EventStoreParser(JsonSerializerSettings serializerSettings = null)
+        public EventStoreFormatter(JsonSerializerSettings serializerSettings = null)
         {
             this.serializerSettings = serializerSettings ?? new JsonSerializerSettings();
         }

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/EventStore/IStreamPositionStorage.cs

@@ -0,0 +1,19 @@
+// ==========================================================================
+//  IStreamPositionStorage.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using EventStore.ClientAPI;
+
+namespace PinkParrot.Infrastructure.CQRS.EventStore
+{
+    public interface IStreamPositionStorage
+    {
+        Position? ReadPosition();
+
+        void WritePosition(Position position);
+    }
+}

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/Events/ICatchEventConsumer.cs

@@ -0,0 +1,13 @@
+// ==========================================================================
+//  ICatchEventConsumer.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+namespace PinkParrot.Infrastructure.CQRS.Events
+{
+    public interface ICatchEventConsumer : IEventConsumer
+    {
+    }
+}

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/Events/IEventConsumer.cs

@@ -1,4 +1,4 @@
-// ==========================================================================
+// ==========================================================================
 //  IEventConsumer.cs
 //  PinkParrot Headless CMS
 // ==========================================================================
@@ -11,4 +11,4 @@ namespace PinkParrot.Infrastructure.CQRS.Events
     {
         void On(Envelope<IEvent> @event);
     }
-}
+}

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/Events/ILiveEventConsumer.cs

@@ -0,0 +1,13 @@
+// ==========================================================================
+//  ILiveEventConsumer.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+namespace PinkParrot.Infrastructure.CQRS.Events
+{
+    public interface ILiveEventConsumer : IEventConsumer
+    {
+    }
+}

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/CQRS/ITenantAggregate.cs

@@ -0,0 +1,17 @@
+// ==========================================================================
+//  ITenantAggregate.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System;
+
+namespace PinkParrot.Infrastructure.CQRS
+{
+    public interface ITenantAggregate : IAggregate
+    {
+        Guid TenantId { get; }
+    }
+}

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/Dispatching/ActionContextDispatcher.cs

@@ -25,7 +25,7 @@ namespace PinkParrot.Infrastructure.Dispatching
                     .Where(Helper.HasRightName)
                     .Where(Helper.HasRightParameters<TIn, TContext>)
                     .Select(ActionContextDispatcherFactory.CreateActionHandler<TTarget, TContext>)
-                    .ToDictionary(h => h.Item1, h => h.Item2);
+                    .ToDictionary<Tuple<Type, Action<TTarget, object, TContext>>, Type, Action<TTarget, object, TContext>>(h => h.Item1, h => h.Item2);
         }
 
         public static bool Dispatch(TTarget target, TIn input, TContext context)

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/Dispatching/ActionDispatcher.cs

@@ -25,7 +25,7 @@ namespace PinkParrot.Infrastructure.Dispatching
                     .Where(Helper.HasRightName)
                     .Where(Helper.HasRightParameters<TIn>)
                     .Select(ActionDispatcherFactory.CreateActionHandler<TTarget>)
-                    .ToDictionary(h => h.Item1, h => h.Item2);
+                    .ToDictionary<Tuple<Type, Action<TTarget, object>>, Type, Action<TTarget, object>>(h => h.Item1, h => h.Item2);
         }
 
         public static bool Dispatch(TTarget target, TIn item)

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/Dispatching/FuncContextDispatcher.cs

@@ -26,7 +26,7 @@ namespace PinkParrot.Infrastructure.Dispatching
                     .Where(Helper.HasRightParameters<TIn, TContext>)
                     .Where(Helper.HasRightReturnType<TOut>)
                     .Select(FuncContextDispatcherFactory.CreateFuncHandler<TTarget, TContext, TOut >)
-                    .ToDictionary(h => h.Item1, h => h.Item2);
+                    .ToDictionary<Tuple<Type, Func<TTarget, object, TContext, TOut>>, Type, Func<TTarget, object, TContext, TOut>>(h => h.Item1, h => h.Item2);
         }
 
         public static TOut Dispatch(TTarget target, TIn item, TContext context)

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/Dispatching/FuncDispatcher.cs

@@ -26,7 +26,7 @@ namespace PinkParrot.Infrastructure.Dispatching
                     .Where(Helper.HasRightParameters<TIn>)
                     .Where(Helper.HasRightReturnType<TOut>)
                     .Select(FuncDispatcherFactory.CreateFuncHandler<TTarget, TOut>)
-                    .ToDictionary(h => h.Item1, h => h.Item2);
+                    .ToDictionary<Tuple<Type, Func<TTarget, object, TOut>>, Type, Func<TTarget, object, TOut>>(h => h.Item1, h => h.Item2);
         }
 
         public static TOut Dispatch(TTarget target, TIn item)

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/DomainObjectNotFoundException.cs

@@ -1,9 +1,9 @@
 // ==========================================================================
-// DomainObjectNotFoundException.cs
+//  DomainObjectNotFoundException.cs
-// Green Parrot Framework
+//  PinkParrot Headless CMS
 // ==========================================================================
-// Copyright (c) Sebastian Stehle
+//  Copyright (c) PinkParrot Group
-// All rights reserved.
+//  All rights reserved.
 // ==========================================================================
 
 using System;

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/MongoDb/BaseMongoDbRepository.cs

@@ -0,0 +1,122 @@
+// ==========================================================================
+//  BaseMongoDbRepository.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System.Globalization;
+using System.Threading.Tasks;
+using MongoDB.Driver;
+
+namespace PinkParrot.Infrastructure.MongoDb
+{
+    public abstract class BaseMongoDbRepository<TEntity>
+    {
+        private const string CollectionFormat = "{0}Set";
+        private readonly IMongoCollection<TEntity> mongoCollection;
+        private readonly IMongoDatabase mongoDatabase;
+        private readonly string typeName;
+
+        protected string TypeName
+        {
+            get
+            {
+                return typeName;
+            }
+        }
+
+        protected ProjectionDefinitionBuilder<TEntity> Projection
+        {
+            get
+            {
+                return Builders<TEntity>.Projection;
+            }
+        }
+
+        protected SortDefinitionBuilder<TEntity> Sort
+        {
+            get
+            {
+                return Builders<TEntity>.Sort;
+            }
+        }
+
+        protected UpdateDefinitionBuilder<TEntity> Update
+        {
+            get
+            {
+                return Builders<TEntity>.Update;
+            }
+        }
+
+        protected FilterDefinitionBuilder<TEntity> Filter
+        {
+            get
+            {
+                return Builders<TEntity>.Filter;
+            }
+        }
+
+        protected IndexKeysDefinitionBuilder<TEntity> IndexKeys
+        {
+            get
+            {
+                return Builders<TEntity>.IndexKeys;
+            }
+        }
+
+        protected IMongoCollection<TEntity> Collection
+        {
+            get
+            {
+                return mongoCollection;
+            }
+        }
+
+        protected IMongoDatabase Database
+        {
+            get
+            {
+                return mongoDatabase;
+            }
+        }
+
+        protected BaseMongoDbRepository(IMongoDatabase database)
+        {
+            Guard.NotNull(database, nameof(database));
+
+            mongoDatabase = database;
+            mongoCollection = CreateCollection();
+
+            typeName = GetType().Name;
+        }
+
+        protected virtual MongoCollectionSettings CollectionSettings()
+        {
+            return new MongoCollectionSettings();
+        }
+
+        protected virtual string CollectionName()
+        {
+            return string.Format(CultureInfo.InvariantCulture, CollectionFormat, typeof(TEntity).Name);
+        }
+
+        private IMongoCollection<TEntity> CreateCollection()
+        {
+            var databaseCollection = mongoDatabase.GetCollection<TEntity>(
+                CollectionName(),
+                CollectionSettings() ?? new MongoCollectionSettings());
+
+            SetupCollectionAsync(databaseCollection).Wait();
+
+            return databaseCollection;
+        }
+
+        protected virtual Task SetupCollectionAsync(IMongoCollection<TEntity> collection)
+        {
+            return Task.FromResult(true);
+        }
+    }
+}

src/pinkparrot_infrastructure/PinkParrot.Infrastructure/project.json

@@ -3,6 +3,8 @@
   "dependencies": {
     "Autofac": "4.1.0",
     "EventStore.ClientAPI.DotNetCore": "1.0.0",
+    "Microsoft.Extensions.Logging": "1.0.0",
+    "MongoDB.Driver": "2.3.0-rc1",
     "NETStandard.Library": "1.6.0",
     "Newtonsoft.Json": "9.0.1",
     "NodaTime": "2.0.0-alpha20160729",

src/pinkparrot_read/PinkParrot.Read/Models/ModelSchemaRM.cs

@@ -8,6 +8,7 @@
 
 using System;
 using System.ComponentModel.DataAnnotations;
+using MongoDB.Bson.Serialization.Attributes;
 using PinkParrot.Infrastructure;
 
 namespace PinkParrot.Read.Models
@@ -15,12 +16,29 @@ namespace PinkParrot.Read.Models
     public sealed class ModelSchemaRM
     {
         [Hide]
+        [BsonId]
         public string Id { get; set; }
 
         [Required]
+        [BsonElement]
+        public Guid SchemaId { get; set; }
+
+        [Required]
+        [BsonElement]
         public string Name { get; set; }
 
         [Required]
-        public Guid SchemaId { get; set; }
+        [BsonElement]
+        public DateTime Created { get; set; }
+
+        [Required]
+        [BsonElement]
+        public DateTime Modified { get; set; }
+
+        [BsonElement]
+        public string Label { get; set; }
+
+        [BsonElement]
+        public string Hints { get; set; }
     }
 }

src/pinkparrot_read/PinkParrot.Read/Repositories/Implementations/MongoModelSchemaRepository.cs

@@ -0,0 +1,67 @@
+// ==========================================================================
+//  MongoModelSchemaRepository.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using MongoDB.Driver;
+using PinkParrot.Events.Schema;
+using PinkParrot.Infrastructure.CQRS;
+using PinkParrot.Infrastructure.CQRS.Events;
+using PinkParrot.Infrastructure.Dispatching;
+using PinkParrot.Infrastructure.MongoDb;
+using PinkParrot.Read.Models;
+
+namespace PinkParrot.Read.Repositories.Implementations
+{
+    public sealed class MongoModelSchemaRepository : BaseMongoDbRepository<ModelSchemaRM>, IModelSchemaRepository, ICatchEventConsumer
+    {
+        public MongoModelSchemaRepository(IMongoDatabase database)
+            : base(database)
+        {
+        }
+
+        protected override Task SetupCollectionAsync(IMongoCollection<ModelSchemaRM> collection)
+        {
+            return Collection.Indexes.CreateOneAsync(IndexKeys.Ascending(x => x.SchemaId));
+        }
+
+        public IQueryable<ModelSchemaRM> QuerySchemas()
+        {
+            return Collection.AsQueryable();
+        }
+
+        public Task<List<ModelSchemaRM>> QueryAllAsync()
+        {
+            return Collection.Find(s => true).ToListAsync();
+        }
+
+        public async void On(ModelSchemaCreated @event, EnvelopeHeaders headers)
+        {
+            var now = DateTime.UtcNow;
+
+            var entity = new ModelSchemaRM
+            {
+                SchemaId = headers.AggregateId(),
+                Created = now,
+                Modified = now,
+                Name = @event.Properties.Name,
+                Hints = @event.Properties.Hints,
+                Label = @event.Properties.Label,
+            };
+
+            await Collection.InsertOneAsync(entity);
+        }
+
+        public void On(Envelope<IEvent> @event)
+        {
+            this.DispatchAction(@event.Payload, @event.Headers);
+        }
+    }
+}

src/pinkparrot_read/PinkParrot.Read/Repositories/MongoDb/MongoDbModelSchemaRepository.cs

@@ -1,41 +0,0 @@
-// ==========================================================================
-//  MongoDbModelSchemaRepository.cs
-//  PinkParrot Headless CMS
-// ==========================================================================
-//  Copyright (c) PinkParrot Group
-//  All rights reserved.
-// ==========================================================================
-
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading.Tasks;
-using MongoDB.Driver;
-using PinkParrot.Read.Models;
-using PinkParrot.Read.Repositories.MongoDb.Utils;
-
-namespace PinkParrot.Read.Repositories.MongoDb
-{
-    public sealed class MongoDbModelSchemaRepository : BaseRepository<ModelSchemaRM>, IModelSchemaRepository
-    {
-        public MongoDbModelSchemaRepository(IMongoDatabase database)
-            : base(database, "ModelSchemas")
-        {
-            CreateIndicesAsync().Wait();
-        }
-
-        private async Task CreateIndicesAsync()
-        {
-            await Collection.Indexes.CreateOneAsync(IndexKeys.Ascending(x => x.SchemaId));
-        }
-
-        public IQueryable<ModelSchemaRM> QuerySchemas()
-        {
-            return Collection.AsQueryable();
-        }
-
-        public Task<List<ModelSchemaRM>> QueryAllAsync()
-        {
-            return Collection.Find(s => true).ToListAsync();
-        }
-    }
-}

src/pinkparrot_read/PinkParrot.Read/Repositories/MongoDb/Utils/BaseRepository.cs

@@ -1,37 +0,0 @@
-// ==========================================================================
-//  BaseRepository.cs
-//  PinkParrot Headless CMS
-// ==========================================================================
-//  Copyright (c) PinkParrot Group
-//  All rights reserved.
-// ==========================================================================
-
-using MongoDB.Driver;
-using PinkParrot.Infrastructure;
-
-namespace PinkParrot.Read.Repositories.MongoDb.Utils
-{
-    public abstract class BaseRepository<T>
-    {
-        private readonly IMongoCollection<T> collection;
-        private readonly IndexKeysDefinitionBuilder<T> indexKeys = new IndexKeysDefinitionBuilder<T>();
-
-        protected IMongoCollection<T> Collection
-        {
-            get { return collection; }
-        }
-
-        protected IndexKeysDefinitionBuilder<T> IndexKeys
-        {
-            get { return indexKeys; }
-        }
-
-        protected BaseRepository(IMongoDatabase database, string collectioName)
-        {
-            Guard.NotNull(database, nameof(database));
-            Guard.NotNullOrEmpty(collectioName, nameof(collectioName));
-
-            collection = database.GetCollection<T>(collectioName);
-        }
-    }
-}

src/pinkparrot_read/PinkParrot.Read/Services/ISchemaProvider.cs → src/pinkparrot_read/PinkParrot.Read/Services/IModelSchemaProvider.cs

@@ -1,5 +1,5 @@
 // ==========================================================================
-//  ISchemaProvider.cs
+//  IModelSchemaProvider.cs
 //  PinkParrot Headless CMS
 // ==========================================================================
 //  Copyright (c) PinkParrot Group
@@ -11,7 +11,7 @@ using System.Threading.Tasks;
 
 namespace PinkParrot.Read.Services
 {
-    public interface ISchemaProvider
+    public interface IModelSchemaProvider
     {
         Task<Guid> FindSchemaIdByNameAsync(string name);
     }

src/pinkparrot_read/PinkParrot.Read/Services/ITenantProvider.cs

@@ -0,0 +1,18 @@
+// ==========================================================================
+//  ITenantProvider.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System;
+using System.Threading.Tasks;
+
+namespace PinkParrot.Read.Services
+{
+    public interface ITenantProvider
+    {
+        Task<Guid> ProvideTenantIdByDomainAsync(string domain);
+    }
+}

src/pinkparrot_read/PinkParrot.Read/Services/Implementations/SchemaProvider.cs → src/pinkparrot_read/PinkParrot.Read/Services/Implementations/ModelSchemaProvider.cs

@@ -1,5 +1,5 @@
 // ==========================================================================
-//  SchemaProvider.cs
+//  ModelSchemaProvider.cs
 //  PinkParrot Headless CMS
 // ==========================================================================
 //  Copyright (c) PinkParrot Group
@@ -11,7 +11,7 @@ using System.Threading.Tasks;
 
 namespace PinkParrot.Read.Services.Implementations
 {
-    public class SchemaProvider : ISchemaProvider
+    public class ModelSchemaProvider : IModelSchemaProvider
     {
         public Task<Guid> FindSchemaIdByNameAsync(string name)
         {

src/pinkparrot_read/PinkParrot.Read/Services/Implementations/MongoPositions.cs

@@ -0,0 +1,27 @@
+// ==========================================================================
+//  MongoPositions.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System.Runtime.Serialization;
+using MongoDB.Bson;
+using MongoDB.Bson.Serialization.Attributes;
+
+namespace PinkParrot.Read.Services.Implementations
+{
+    [DataContract]
+    public class MongoPosition
+    {
+        [BsonId]
+        public ObjectId Id { get; set; }
+
+        [BsonElement]
+        public long CommitPosition { get; set; }
+
+        [BsonElement]
+        public long PreparePosition { get; set; }
+    }
+}

src/pinkparrot_read/PinkParrot.Read/Services/Implementations/MongoStreamPositionsStorage.cs

@@ -0,0 +1,60 @@
+// ==========================================================================
+//  MongoStreamPositionsStorage.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using EventStore.ClientAPI;
+using MongoDB.Bson;
+using PinkParrot.Infrastructure.CQRS.EventStore;
+using PinkParrot.Infrastructure.MongoDb;
+using IFindFluentExtensions = MongoDB.Driver.IFindFluentExtensions;
+using IMongoDatabase = MongoDB.Driver.IMongoDatabase;
+
+//// ReSharper disable once ConvertIfStatementToNullCoalescingExpression
+
+namespace PinkParrot.Read.Services.Implementations
+{
+    public sealed class MongoStreamPositionsStorage : BaseMongoDbRepository<MongoPosition>, IStreamPositionStorage
+    {
+        private static readonly ObjectId Id = new ObjectId("507f1f77bcf86cd799439011");
+
+        public MongoStreamPositionsStorage(IMongoDatabase database)
+            : base(database)
+        {
+        }
+
+        public Position? ReadPosition()
+        {
+            var document = IFindFluentExtensions.FirstOrDefault<MongoPosition, MongoPosition>(Collection.Find(t => t.Id == Id));
+
+            return document != null ? new Position(document.CommitPosition, document.PreparePosition) : Position.Start;
+        }
+
+        public void WritePosition(Position position)
+        {
+            var document = IFindFluentExtensions.FirstOrDefault<MongoPosition, MongoPosition>(Collection.Find(t => t.Id == Id));
+
+            var isFound = document != null;
+
+            if (document == null)
+            {
+                document = new MongoPosition { Id = Id };
+            }
+
+            document.CommitPosition = position.CommitPosition;
+            document.PreparePosition = position.PreparePosition;
+
+            if (isFound)
+            {
+                Collection.ReplaceOne(t => t.Id == Id, document);
+            }
+            else
+            {
+                Collection.InsertOne(document);
+            }
+        }
+    }
+}

src/pinkparrot_read/PinkParrot.Read/Services/Implementations/TenantProvider.cs

@@ -0,0 +1,21 @@
+// ==========================================================================
+//  TenantProvider.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System;
+using System.Threading.Tasks;
+
+namespace PinkParrot.Read.Services.Implementations
+{
+    public sealed class TenantProvider : ITenantProvider
+    {
+        public Task<Guid> ProvideTenantIdByDomainAsync(string domain)
+        {
+            return Task.FromResult(Guid.Empty);
+        }
+    }
+}

src/pinkparrot_write/PinkParrot.Write/ITenantCommand.cs

@@ -0,0 +1,18 @@
+// ==========================================================================
+//  ITenantCommand.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System;
+using PinkParrot.Infrastructure.CQRS.Commands;
+
+namespace PinkParrot.Write
+{
+    public interface ITenantCommand : ICommand
+    {
+        Guid TenantId { get; set; }
+    }
+}

src/pinkparrot_write/PinkParrot.Write/Schema/Commands/CreateModelSchema.cs

@@ -7,11 +7,10 @@
 // ==========================================================================
 
 using PinkParrot.Core.Schema;
-using PinkParrot.Infrastructure.CQRS.Commands;
 
 namespace PinkParrot.Write.Schema.Commands
 {
-    public class CreateModelSchema : AggregateCommand
+    public class CreateModelSchema : TenantCommand
     {
         public ModelSchemaProperties Properties { get; set; }
     }

src/pinkparrot_write/PinkParrot.Write/Schema/Commands/DeleteModelSchema.cs

@@ -6,13 +6,11 @@
 //  All rights reserved.
 // ==========================================================================
 
-using System;
 using PinkParrot.Infrastructure.CQRS.Commands;
 
 namespace PinkParrot.Write.Schema.Commands
 {
-    public class DeleteModelSchema : IAggregateCommand
+    public class DeleteModelSchema : AggregateCommand
     {
-        public Guid AggregateId { get; set; }
     }
 }

src/pinkparrot_write/PinkParrot.Write/Schema/ModelSchemaDomainObject.cs

@@ -16,9 +16,10 @@ using PinkParrot.Write.Schema.Commands;
 
 namespace PinkParrot.Write.Schema
 {
-    public class ModelSchemaDomainObject : DomainObject
+    public class ModelSchemaDomainObject : DomainObject, IAggregate
     {
         private readonly ModelFieldFactory fieldFactory;
+        private Guid tenantId;
         private bool isDeleted;
         private long totalFields;
         private ModelSchema schema;
@@ -28,6 +29,11 @@ namespace PinkParrot.Write.Schema
             get { return schema; }
         }
 
+        public Guid TenantId
+        {
+            get { return tenantId; }
+        }
+
         public bool IsDeleted
         {
             get { return isDeleted; }
@@ -48,6 +54,8 @@ namespace PinkParrot.Write.Schema
 
         protected void Apply(ModelSchemaCreated @event)
         {
+            tenantId = @event.TenantId;
+
             schema = ModelSchema.Create(@event.Properties);
         }
 
@@ -106,6 +114,8 @@ namespace PinkParrot.Write.Schema
         {
             VerifyNotCreated();
 
+            tenantId = command.TenantId;
+
             schema = ModelSchema.Create(command.Properties);
 
             RaiseEvent(new ModelSchemaCreated { Properties = command.Properties }, true);

src/pinkparrot_write/PinkParrot.Write/TenantCommand.cs

@@ -0,0 +1,18 @@
+// ==========================================================================
+//  TenantCommand.cs
+//  PinkParrot Headless CMS
+// ==========================================================================
+//  Copyright (c) PinkParrot Group
+//  All rights reserved.
+// ==========================================================================
+
+using System;
+using PinkParrot.Infrastructure.CQRS.Commands;
+
+namespace PinkParrot.Write
+{
+    public abstract class TenantCommand : AggregateCommand
+    {
+        public Guid TenantId { get; set; }
+    }
+}

src/pinkparrot_write/PinkParrot.Write/project.json

@@ -6,7 +6,8 @@
     "NodaTime": "2.0.0-alpha20160729",
     "PinkParrot.Core": "1.0.0-*",
     "PinkParrot.Events": "1.0.0-*",
-    "PinkParrot.Infrastructure": "1.0.0-*"
+    "PinkParrot.Infrastructure": "1.0.0-*",
+    "PinkParrot.Read": "1.0.0-*"
   },
 
   "frameworks": {