Denormalizer management

src/Squidex.Infrastructure.MongoDb/MongoEventConsumerInfo.cs

@@ -0,0 +1,33 @@
+// ==========================================================================
+//  MongoEventConsumerInfo.cs
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex Group
+//  All rights reserved.
+// ==========================================================================
+
+using MongoDB.Bson;
+using MongoDB.Bson.Serialization.Attributes;
+using Squidex.Infrastructure.CQRS.Events;
+
+namespace Squidex.Infrastructure.MongoDb
+{
+    public sealed class MongoEventConsumerInfo : IEventConsumerInfo
+    {
+        [BsonId]
+        [BsonRepresentation(BsonType.String)]
+        public string Name { get; set; }
+
+        [BsonElement]
+        [BsonIgnoreIfDefault]
+        public bool IsStopped { get; set; }
+
+        [BsonElement]
+        [BsonIgnoreIfDefault]
+        public bool IsResetting { get; set; }
+
+        [BsonElement]
+        [BsonRequired]
+        public long LastHandledEventNumber { get; set; }
+    }
+}

src/Squidex.Infrastructure.MongoDb/MongoEventConsumerInfoRepository.cs

@@ -0,0 +1,82 @@
+// ==========================================================================
+//  MongoEventConsumerInfoRepository.cs
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex Group
+//  All rights reserved.
+// ==========================================================================
+
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using MongoDB.Bson;
+using MongoDB.Driver;
+using Squidex.Infrastructure.CQRS.Events;
+
+namespace Squidex.Infrastructure.MongoDb
+{
+    public sealed class MongoEventConsumerInfoRepository : MongoRepositoryBase<MongoEventConsumerInfo>, IEventConsumerInfoRepository
+    {
+        public MongoEventConsumerInfoRepository(IMongoDatabase database) 
+            : base(database)
+        {
+        }
+
+        protected override string CollectionName()
+        {
+            return "EventPositions";
+        }
+
+        public async Task<IReadOnlyList<IEventConsumerInfo>> QueryAsync()
+        {
+            var entities = await Collection.Find(new BsonDocument()).SortBy(x => x.Name).ToListAsync();
+
+            return entities.OfType<IEventConsumerInfo>().ToList();
+        }
+
+        public async Task<IEventConsumerInfo> FindAsync(string consumerName)
+        {
+            var entity = await Collection.Find(x => x.Name == consumerName).FirstOrDefaultAsync();
+
+            return entity;
+        }
+
+        public async Task CreateAsync(string consumerName)
+        {
+            if (await Collection.CountAsync(x => x.Name == consumerName) == 0)
+            {
+                try
+                {
+                    await Collection.InsertOneAsync(new MongoEventConsumerInfo { Name = consumerName, LastHandledEventNumber = -1 });
+                }
+                catch (MongoWriteException ex)
+                {
+                    if (ex.WriteError?.Category != ServerErrorCategory.DuplicateKey)
+                    {
+                        throw;
+                    }
+                }
+            }
+        }
+
+        public Task StartAsync(string consumerName)
+        {
+            return Collection.UpdateOneAsync(x => x.Name == consumerName, Update.Unset(x => x.IsStopped));
+        }
+
+        public Task StopAsync(string consumerName)
+        {
+            return Collection.UpdateOneAsync(x => x.Name == consumerName, Update.Set(x => x.IsStopped, true));
+        }
+
+        public Task SetLastHandledEventNumberAsync(string consumerName, long eventNumber)
+        {
+            return Collection.UpdateOneAsync(x => x.Name == consumerName, Update.Set(x => x.LastHandledEventNumber, eventNumber).Unset(x => x.IsResetting).Unset(x => x.IsStopped));
+        }
+
+        public Task ResetAsync(string consumerName)
+        {
+            return Collection.UpdateOneAsync(x => x.Name == consumerName, Update.Set(x => x.IsResetting, true));
+        }
+    }
+}

src/Squidex.Infrastructure.MongoDb/MongoRepositoryBase.cs

@@ -9,6 +9,7 @@
 using System;
 using System.Globalization;
 using System.Threading.Tasks;
+using MongoDB.Bson;
 using MongoDB.Driver;
 
 namespace Squidex.Infrastructure.MongoDb
@@ -128,6 +129,11 @@ namespace Squidex.Infrastructure.MongoDb
             return Task.FromResult(true);
         }
 
+        public virtual Task ClearAsync()
+        {
+            return Collection.DeleteManyAsync(new BsonDocument());
+        }
+
         public async Task<bool> TryDropCollectionAsync()
         {
             try

src/Squidex.Infrastructure/CQRS/Events/EventReceiver.cs

@@ -22,6 +22,7 @@ namespace Squidex.Infrastructure.CQRS.Events
         private readonly EventDataFormatter formatter;
         private readonly IEventStore eventStore;
         private readonly IEventNotifier eventNotifier;
+        private readonly IEventConsumerInfoRepository eventConsumerInfoRepository;
         private readonly ILogger<EventReceiver> logger;
         private CompletionTimer timer;
 
@@ -29,17 +30,20 @@ namespace Squidex.Infrastructure.CQRS.Events
             EventDataFormatter formatter,
             IEventStore eventStore, 
             IEventNotifier eventNotifier,
+            IEventConsumerInfoRepository eventConsumerInfoRepository,
             ILogger<EventReceiver> logger)
         {
             Guard.NotNull(logger, nameof(logger));
             Guard.NotNull(formatter, nameof(formatter));
             Guard.NotNull(eventStore, nameof(eventStore));
             Guard.NotNull(eventNotifier, nameof(eventNotifier));
+            Guard.NotNull(eventConsumerInfoRepository, nameof(eventConsumerInfoRepository));
 
             this.logger = logger;
             this.formatter = formatter;
             this.eventStore = eventStore;
             this.eventNotifier = eventNotifier;
+            this.eventConsumerInfoRepository = eventConsumerInfoRepository;
         }
 
         protected override void DisposeObject(bool disposing)
@@ -57,7 +61,12 @@ namespace Squidex.Infrastructure.CQRS.Events
             }
         }
 
-        public void Subscribe(IEventCatchConsumer eventConsumer, int delay = 5000)
+        public void Trigger()
+        {
+            timer?.Trigger();
+        }
+
+        public void Subscribe(IEventConsumer eventConsumer, int delay = 5000)
         {
             Guard.NotNull(eventConsumer, nameof(eventConsumer));
 
@@ -66,26 +75,40 @@ namespace Squidex.Infrastructure.CQRS.Events
                 return;
             }
 
-            var lastReceivedPosition = long.MinValue;
+            var consumerName = eventConsumer.GetType().Name;
+            var consumerStarted = false;
             
             timer = new CompletionTimer(delay, async ct =>
             {
-                if (lastReceivedPosition == long.MinValue)
+                if (!consumerStarted)
                 {
-                    lastReceivedPosition = await eventConsumer.GetLastHandledEventNumber();
+                    await eventConsumerInfoRepository.CreateAsync(consumerName);
+
+                    consumerStarted = true;
                 }
 
-                var tcs = new TaskCompletionSource<bool>();
+                try
+                {
+                    var status = await eventConsumerInfoRepository.FindAsync(consumerName);
+
+                    var lastHandledEventNumber = status.LastHandledEventNumber;
 
-                eventStore.GetEventsAsync(lastReceivedPosition).Subscribe(storedEvent =>
+                    if (status.IsResetting)
                     {
-                    var @event = ParseEvent(storedEvent.Data);
+                        await ResetAsync(eventConsumer, consumerName);
 
-                    @event.SetEventNumber(storedEvent.EventNumber);
+                        lastHandledEventNumber = -1;
+                    }
+                    else if (status.IsStopped)
+                    {
+                        return;
+                    }
 
-                    DispatchConsumer(@event, eventConsumer, storedEvent.EventNumber).Wait();
+                    var tcs = new TaskCompletionSource<bool>();
                     
-                    lastReceivedPosition++;
+                    eventStore.GetEventsAsync(lastHandledEventNumber).Subscribe(storedEvent =>
+                    {
+                        HandleEventAsync(eventConsumer, storedEvent, consumerName).Wait();
                     }, ex =>
                     {
                         tcs.SetException(ex);
@@ -95,38 +118,78 @@ namespace Squidex.Infrastructure.CQRS.Events
                     }, ct);
 
                     await tcs.Task;
+                }
+                catch (Exception ex)
+                {
+                    logger.LogError(InfrastructureErrors.EventHandlingFailed, ex, "Failed to handle events");
+
+                    await eventConsumerInfoRepository.StopAsync(consumerName);
+
+                    throw;
+                }
             });
 
             eventNotifier.Subscribe(timer.Trigger);
         }
 
-        private async Task DispatchConsumer(Envelope<IEvent> @event, IEventCatchConsumer consumer, long eventNumber)
+        private async Task HandleEventAsync(IEventConsumer eventConsumer, StoredEvent storedEvent, string consumerName)
+        {
+            var @event = ParseEvent(storedEvent);
+
+            await DispatchConsumer(@event, eventConsumer);
+
+            await eventConsumerInfoRepository.SetLastHandledEventNumberAsync(consumerName, storedEvent.EventNumber);
+        }
+
+        private async Task ResetAsync(IEventConsumer eventConsumer, string consumerName)
         {
             try
             {
-                await consumer.On(@event, eventNumber);
 
-                logger.LogDebug("[{0}]: Handled event {1} ({2})", consumer, @event.Payload, @event.Headers.EventId());
+                logger.LogDebug("[{0}]: Reset started", eventConsumer);
+
+                await eventConsumer.ClearAsync();
+                await eventConsumerInfoRepository.SetLastHandledEventNumberAsync(consumerName, -1);
+
+                logger.LogDebug("[{0}]: Reset completed", eventConsumer);
             }
             catch (Exception ex)
             {
-                logger.LogError(InfrastructureErrors.EventHandlingFailed, ex, "[{0}]: Failed to handle event {1} ({2})", consumer, @event.Payload, @event.Headers.EventId());
+                logger.LogError(InfrastructureErrors.EventResetFailed, ex, "[{0}]: Reset failed", eventConsumer);
 
                 throw;
             }
         }
 
-        private Envelope<IEvent> ParseEvent(EventData eventData)
+        private async Task DispatchConsumer(Envelope<IEvent> @event, IEventConsumer eventConsumer)
         {
             try
             {
-                var @event = formatter.Parse(eventData);
+                await eventConsumer.On(@event);
+
+                logger.LogDebug("[{0}]: Handled event {1} ({2})", eventConsumer, @event.Payload, @event.Headers.EventId());
+            }
+            catch (Exception ex)
+            {
+                logger.LogError(InfrastructureErrors.EventHandlingFailed, ex, "[{0}]: Failed to handle event {1} ({2})", eventConsumer, @event.Payload, @event.Headers.EventId());
+
+                throw;
+            }
+        }
+
+        private Envelope<IEvent> ParseEvent(StoredEvent storedEvent)
+        {
+            try
+            {
+                var @event = formatter.Parse(storedEvent.Data);
+
+                @event.SetEventNumber(storedEvent.EventNumber);
 
                 return @event;
             }
             catch (Exception ex)
             {
-                logger.LogError(InfrastructureErrors.EventDeserializationFailed, ex, "Failed to parse event {0}", eventData.EventId);
+                logger.LogError(InfrastructureErrors.EventDeserializationFailed, ex, "Failed to parse event {0}", storedEvent.Data.EventId);
 
                 throw;
             }

src/Squidex.Infrastructure/CQRS/Events/IEvent.cs

@@ -5,6 +5,7 @@
 //  Copyright (c) Squidex Group
 //  All rights reserved.
 // ==========================================================================
+
 namespace Squidex.Infrastructure.CQRS.Events
 {
     public interface IEvent

src/Squidex.Infrastructure/CQRS/Events/IEventConsumer.cs

@@ -12,6 +12,8 @@ namespace Squidex.Infrastructure.CQRS.Events
 {
     public interface IEventConsumer
     {
+        Task ClearAsync();
+
         Task On(Envelope<IEvent> @event);
     }
 }

src/Squidex.Infrastructure/CQRS/Events/IEventCatchConsumer.cs → src/Squidex.Infrastructure/CQRS/Events/IEventConsumerInfo.cs

@@ -1,19 +1,21 @@
 // ==========================================================================
-//  IEventCatchConsumer.cs
+//  IEventCatchConsumerInfo.cs
 //  Squidex Headless CMS
 // ==========================================================================
 //  Copyright (c) Squidex Group
 //  All rights reserved.
 // ==========================================================================
 
-using System.Threading.Tasks;
-
 namespace Squidex.Infrastructure.CQRS.Events
 {
-    public interface IEventCatchConsumer
+    public interface IEventConsumerInfo
     {
-        Task<long> GetLastHandledEventNumber();
+        long LastHandledEventNumber { get; }
+
+        bool IsStopped { get; }
+
+        bool IsResetting { get; }
 
-        Task On(Envelope<IEvent> @event, long eventNumber);
+        string Name { get; }
     }
 }

src/Squidex.Infrastructure/CQRS/Events/IEventConsumerInfoRepository.cs

@@ -0,0 +1,30 @@
+// ==========================================================================
+//  IEventCatchConsumerControlStore.cs
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex Group
+//  All rights reserved.
+// ==========================================================================
+
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace Squidex.Infrastructure.CQRS.Events
+{
+    public interface IEventConsumerInfoRepository
+    {
+        Task<IReadOnlyList<IEventConsumerInfo>> QueryAsync();
+
+        Task<IEventConsumerInfo> FindAsync(string consumerName);
+
+        Task CreateAsync(string consumerName);
+
+        Task StartAsync(string consumerName);
+
+        Task StopAsync(string consumerName);
+
+        Task ResetAsync(string consumerName);
+
+        Task SetLastHandledEventNumberAsync(string consumerName, long eventNumber);
+    }
+}

src/Squidex.Infrastructure/CQRS/Events/StoredEvent.cs

@@ -5,21 +5,31 @@
 //  Copyright (c) Squidex Group
 //  All rights reserved.
 // ==========================================================================
+
 namespace Squidex.Infrastructure.CQRS.Events
 {
     public sealed class StoredEvent
     {
-        public long EventNumber { get; }
+        private readonly long eventNumber;
+        private readonly EventData data;
+
+        public long EventNumber
+        {
+            get { return eventNumber; }
+        }
 
-        public EventData Data { get; }
+        public EventData Data
+        {
+            get { return data; }
+        }
 
         public StoredEvent(long eventNumber, EventData data)
         {
             Guard.NotNull(data, nameof(data));
 
-            EventNumber = eventNumber;
+            this.data = data;
 
-            Data = data;
+            this.eventNumber = eventNumber;
         }
     }
 }

src/Squidex.Infrastructure/InfrastructureErrors.cs

@@ -16,6 +16,8 @@ namespace Squidex.Infrastructure
 
         public static readonly EventId CommandFailed = new EventId(20001, "CommandFailed");
 
+        public static readonly EventId EventResetFailed = new EventId(10000, "EventResetFailed");
+
         public static readonly EventId EventHandlingFailed = new EventId(10001, "EventHandlingFailed");
 
         public static readonly EventId EventDeserializationFailed = new EventId(10002, "EventDeserializationFailed");

src/Squidex.Read.MongoDb/History/MongoHistoryEventRepository.cs

@@ -59,11 +59,6 @@ namespace Squidex.Read.MongoDb.History
                 collection.Indexes.CreateOneAsync(IndexKeys.Ascending(x => x.Created), new CreateIndexOptions { ExpireAfter = TimeSpan.FromDays(365) }));
         }
 
-        public Task ClearAsync()
-        {
-            return TryDropCollectionAsync();
-        }
-
         public async Task<List<IHistoryEventEntity>> QueryEventsByChannel(Guid appId, string channelPrefix, int count)
         {
             var entities =

src/Squidex.Read.MongoDb/Schemas/MongoSchemaRepository.cs

@@ -47,11 +47,6 @@ namespace Squidex.Read.MongoDb.Schemas
             return collection.Indexes.CreateOneAsync(IndexKeys.Ascending(x => x.Name));
         }
 
-        public Task ClearAsync()
-        {
-            return TryDropCollectionAsync();
-        }
-
         public async Task<IReadOnlyList<ISchemaEntity>> QueryAllAsync(Guid appId)
         {
             var entities = await Collection.Find(s => s.AppId == appId).ToListAsync();

src/Squidex.Read.MongoDb/Utils/MongoDbConsumerWrapper.cs

@@ -1,74 +0,0 @@
-// ==========================================================================
-//  MongoDbStore.cs
-//  Squidex Headless CMS
-// ==========================================================================
-//  Copyright (c) Squidex Group
-//  All rights reserved.
-// ==========================================================================
-
-using System.Threading.Tasks;
-using MongoDB.Bson;
-using MongoDB.Bson.Serialization.Attributes;
-using MongoDB.Driver;
-using Squidex.Infrastructure;
-using Squidex.Infrastructure.CQRS;
-using Squidex.Infrastructure.CQRS.Events;
-using Squidex.Infrastructure.MongoDb;
-
-namespace Squidex.Read.MongoDb.Utils
-{
-    public sealed class EventPosition
-    {
-        [BsonId]
-        [BsonRepresentation(BsonType.String)]
-        public string Name { get; set; }
-
-        [BsonElement]
-        [BsonRequired]
-        public long EventNumber { get; set; }
-    }
-
-    public sealed class MongoDbConsumerWrapper : MongoRepositoryBase<EventPosition>, IEventCatchConsumer
-    {
-        private static readonly UpdateOptions upsert = new UpdateOptions { IsUpsert = true };
-        private readonly IEventConsumer eventConsumer;
-        private readonly string eventStoreName;
-
-        public MongoDbConsumerWrapper(IMongoDatabase database, IEventConsumer eventConsumer) 
-            : base(database)
-        {
-            Guard.NotNull(eventConsumer, nameof(eventConsumer));
-
-            this.eventConsumer = eventConsumer;
-
-            eventStoreName = eventConsumer.GetType().Name;
-        }
-
-        protected override string CollectionName()
-        {
-            return "EventPositions";
-        }
-
-        public async Task On(Envelope<IEvent> @event, long eventNumber)
-        {
-            await eventConsumer.On(@event);
-
-            await SetLastHandledEventNumber(eventNumber);
-        }
-
-        private Task SetLastHandledEventNumber(long eventNumber)
-        {
-            return Collection.ReplaceOneAsync(x => x.Name == eventStoreName, new EventPosition { Name = eventStoreName, EventNumber = eventNumber }, upsert);
-        }
-
-        public async Task<long> GetLastHandledEventNumber()
-        {
-            var collectionPosition =
-                await Collection
-                    .Find(x => x.Name == eventStoreName).SortByDescending(x => x.EventNumber).Limit(1)
-                    .FirstOrDefaultAsync();
-                    
-            return collectionPosition?.EventNumber ?? -1;
-        }
-    }
-}

src/Squidex/Config/Constants.cs

@@ -14,6 +14,8 @@ namespace Squidex.Config
 
         public const string ApiScope = "squidex-api";
 
+        public const string RoleScope = "role";
+
         public const string ProfileScope = "squidex-profile";
 
         public const string FrontendClient = "squidex-frontend";

src/Squidex/Config/Domain/StoreMongoDbModule.cs

@@ -15,6 +15,7 @@ using Microsoft.Extensions.Configuration;
 using MongoDB.Driver;
 using Squidex.Infrastructure;
 using Squidex.Infrastructure.CQRS.Events;
+using Squidex.Infrastructure.MongoDb;
 using Squidex.Read.Apps.Repositories;
 using Squidex.Read.Contents.Repositories;
 using Squidex.Read.History.Repositories;
@@ -24,7 +25,6 @@ using Squidex.Read.MongoDb.History;
 using Squidex.Read.MongoDb.Infrastructure;
 using Squidex.Read.MongoDb.Schemas;
 using Squidex.Read.MongoDb.Users;
-using Squidex.Read.MongoDb.Utils;
 using Squidex.Read.Schemas.Repositories;
 using Squidex.Read.Users.Repositories;
 
@@ -93,15 +93,23 @@ namespace Squidex.Config.Domain
                 .As<IPersistedGrantStore>()
                 .SingleInstance();
 
+            builder.RegisterType<MongoEventConsumerInfoRepository>()
+                .WithParameter(ResolvedParameter.ForNamed<IMongoDatabase>(MongoDatabaseName))
+                .As<IEventConsumerInfoRepository>()
+                .AsSelf()
+                .SingleInstance();
+
             builder.RegisterType<MongoContentRepository>()
                 .WithParameter(ResolvedParameter.ForNamed<IMongoDatabase>(MongoDatabaseName))
                 .As<IContentRepository>()
+                .As<IEventConsumer>()
                 .AsSelf()
                 .SingleInstance();
 
             builder.RegisterType<MongoHistoryEventRepository>()
                 .WithParameter(ResolvedParameter.ForNamed<IMongoDatabase>(MongoDatabaseName))
                 .As<IHistoryEventRepository>()
+                .As<IEventConsumer>()
                 .As<IExternalSystem>()
                 .AsSelf()
                 .SingleInstance();
@@ -109,6 +117,7 @@ namespace Squidex.Config.Domain
             builder.RegisterType<MongoSchemaRepository>()
                 .WithParameter(ResolvedParameter.ForNamed<IMongoDatabase>(MongoDatabaseName))
                 .As<ISchemaRepository>()
+                .As<IEventConsumer>()
                 .As<IExternalSystem>()
                 .AsSelf()
                 .SingleInstance();
@@ -116,37 +125,10 @@ namespace Squidex.Config.Domain
             builder.RegisterType<MongoAppRepository>()
                 .WithParameter(ResolvedParameter.ForNamed<IMongoDatabase>(MongoDatabaseName))
                 .As<IAppRepository>()
+                .As<IEventConsumer>()
                 .As<IExternalSystem>()
                 .AsSelf()
                 .SingleInstance();
-
-            builder.Register(c =>
-                new MongoDbConsumerWrapper(
-                    c.ResolveNamed<IMongoDatabase>(MongoDatabaseName),
-                    c.Resolve<MongoAppRepository>()))
-                .As<IEventCatchConsumer>()
-                .SingleInstance();
-            
-            builder.Register(c =>
-                new MongoDbConsumerWrapper(
-                    c.ResolveNamed<IMongoDatabase>(MongoDatabaseName),
-                    c.Resolve<MongoContentRepository>()))
-                .As<IEventCatchConsumer>()
-                .SingleInstance();
-
-            builder.Register(c =>
-                new MongoDbConsumerWrapper(
-                    c.ResolveNamed<IMongoDatabase>(MongoDatabaseName),
-                    c.Resolve<MongoSchemaRepository>()))
-                .As<IEventCatchConsumer>()
-                .SingleInstance();
-
-            builder.Register(c =>
-                new MongoDbConsumerWrapper(
-                    c.ResolveNamed<IMongoDatabase>(MongoDatabaseName),
-                    c.Resolve<MongoHistoryEventRepository>()))
-                .As<IEventCatchConsumer>()
-                .SingleInstance();
         }
     }
 }

src/Squidex/Config/Domain/Usages.cs

@@ -22,7 +22,7 @@ namespace Squidex.Config.Domain
     {
         public static IApplicationBuilder UseMyEventStore(this IApplicationBuilder app)
         {
-            var catchConsumers = app.ApplicationServices.GetServices<IEventCatchConsumer>();
+            var catchConsumers = app.ApplicationServices.GetServices<IEventConsumer>();
 
             foreach (var catchConsumer in catchConsumers)
             {

src/Squidex/Config/Identity/IdentityServices.cs

@@ -11,6 +11,7 @@ using System.Collections.Generic;
 using System.IO;
 using System.Reflection;
 using System.Security.Cryptography.X509Certificates;
+using IdentityModel;
 using IdentityServer4.Models;
 using IdentityServer4.Stores;
 using Microsoft.AspNetCore.DataProtection;
@@ -107,13 +108,25 @@ namespace Squidex.Config.Identity
 
         private static IEnumerable<ApiResource> GetApiResources()
         {
-            yield return new ApiResource(Constants.ApiScope);
+            yield return new ApiResource(Constants.ApiScope)
+            {
+                UserClaims = new List<string>
+                {
+                    JwtClaimTypes.Role
+                }
+            };
         }
 
         private static IEnumerable<IdentityResource> GetIdentityResources()
         {
             yield return new IdentityResources.OpenId();
             yield return new IdentityResources.Profile();
+            yield return new IdentityResources.Profile();
+            yield return new IdentityResource(Constants.RoleScope,
+                new[]
+                {
+                    JwtClaimTypes.Role
+                });
             yield return new IdentityResource(Constants.ProfileScope,
                 new[]
                 {

src/Squidex/Config/Identity/IdentityUsage.cs

@@ -140,8 +140,7 @@ namespace Squidex.Config.Identity
                 {
                     var apiRequestUri = new Uri($"https://www.googleapis.com/oauth2/v2/userinfo?access_token={context.AccessToken}");
 
-                    var jsonReponseString =
-                        await HttpClient.GetStringAsync(apiRequestUri);
+                    var jsonReponseString = await HttpClient.GetStringAsync(apiRequestUri);
                     var jsonResponse = JToken.Parse(jsonReponseString);
 
                     var pictureUrl = jsonResponse["picture"]?.Value<string>();

src/Squidex/Config/Identity/LazyClientStore.cs

@@ -77,7 +77,8 @@ namespace Squidex.Config.Identity
                 AllowedGrantTypes = GrantTypes.ClientCredentials,
                 AllowedScopes = new List<string>
                 {
-                    Constants.ApiScope
+                    Constants.ApiScope,
+                    Constants.RoleScope
                 }
             };
         }
@@ -115,7 +116,8 @@ namespace Squidex.Config.Identity
                     IdentityServerConstants.StandardScopes.OpenId,
                     IdentityServerConstants.StandardScopes.Profile,
                     Constants.ApiScope,
-                    Constants.ProfileScope
+                    Constants.ProfileScope,
+                    Constants.RoleScope
                 },
                 RequireConsent = false
             };

src/Squidex/Controllers/Api/EventConsumers/EventConsumersController.cs

@@ -0,0 +1,71 @@
+// ==========================================================================
+//  EventConsumersController.cs
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex Group
+//  All rights reserved.
+// ==========================================================================
+
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Mvc;
+using NSwag.Annotations;
+using Squidex.Controllers.Api.EventConsumers.Models;
+using Squidex.Infrastructure.CQRS.Events;
+using Squidex.Infrastructure.Reflection;
+using Squidex.Pipeline;
+
+namespace Squidex.Controllers.Api.EventConsumers
+{
+    [ApiExceptionFilter]
+    [Authorize(Roles = "administrator")]
+    [SwaggerIgnore]
+    public sealed class EventConsumersController : Controller
+    {
+        private readonly IEventConsumerInfoRepository eventConsumerRepository;
+
+        public EventConsumersController(IEventConsumerInfoRepository eventConsumerRepository)
+        {
+            this.eventConsumerRepository = eventConsumerRepository;
+        }
+
+        [HttpGet]
+        [Route("event-consumers/")]
+        public async Task<IActionResult> GetEventConsumers()
+        {
+            var entities = await eventConsumerRepository.QueryAsync();
+
+            var models = entities.Select(x => SimpleMapper.Map(x, new EventConsumerDto())).ToList();
+
+            return Ok(models);
+        }
+
+        [HttpPut]
+        [Route("event-consumers/{name}/start")]
+        public async Task<IActionResult> Start(string name)
+        {
+            await eventConsumerRepository.StartAsync(name);
+
+            return NoContent();
+        }
+
+        [HttpPut]
+        [Route("event-consumers/{name}/stop")]
+        public async Task<IActionResult> Stop(string name)
+        {
+            await eventConsumerRepository.StopAsync(name);
+
+            return NoContent();
+        }
+
+        [HttpPut]
+        [Route("event-consumers/{name}/reset")]
+        public async Task<IActionResult> Reset(string name)
+        {
+            await eventConsumerRepository.ResetAsync(name);
+
+            return NoContent();
+        }
+    }
+}

src/Squidex/Controllers/Api/EventConsumers/Models/EventConsumerDto.cs

@@ -0,0 +1,21 @@
+// ==========================================================================
+//  EventConsumerDto.cs
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex Group
+//  All rights reserved.
+// ==========================================================================
+
+namespace Squidex.Controllers.Api.EventConsumers.Models
+{
+    public sealed class EventConsumerDto
+    {
+        public long LastHandledEventNumber { get; set; }
+
+        public bool IsStopped { get; set; }
+
+        public bool IsResetting { get; set; }
+
+        public string Name { get; set; }
+    }
+}

src/Squidex/Pipeline/ApiExceptionFilterAttribute.cs

@@ -34,14 +34,24 @@ namespace Squidex.Pipeline
 
         static ApiExceptionFilterAttribute()
         {
-            AddHandler<DomainObjectNotFoundException>(ex =>
-                new NotFoundResult());
+            AddHandler<DomainObjectNotFoundException>(OnDomainObjectNotFoundException);
+            AddHandler<DomainException>(OnDomainException);
+            AddHandler<ValidationException>(OnValidationException);
+        }
 
-            AddHandler<DomainException>(ex =>
-                new BadRequestObjectResult(new ErrorDto { Message = ex.Message }));
+        private static IActionResult OnDomainObjectNotFoundException(DomainObjectNotFoundException ex)
+        {
+            return new NotFoundResult();
+        }
 
-            AddHandler<ValidationException>(ex =>
-                new BadRequestObjectResult(new ErrorDto { Message = ex.Message, Details = ex.Errors.Select(e => e.Message).ToArray() }));
+        private static IActionResult OnDomainException(DomainException ex)
+        {
+            return new BadRequestObjectResult(new ErrorDto { Message = ex.Message });
+        }
+
+        private static IActionResult OnValidationException(ValidationException ex)
+        {
+            return new BadRequestObjectResult(new ErrorDto { Message = ex.Message, Details = ex.Errors.Select(e => e.Message).ToArray() });
         }
 
         public override void OnActionExecuting(ActionExecutingContext context)

src/Squidex/app/app.routes.ts

@@ -36,6 +36,9 @@ export const routes: Routes = [
             {
                 path: '',
                 loadChildren: './features/apps/module#SqxFeatureAppsModule'
+            }, {
+                path: 'administration',
+                loadChildren: './features/administration/module#SqxFeatureAdministrationModule'
             }, {
                 path: ':appName',
                 component: AppAreaComponent,

src/Squidex/app/features/administration/administration-area.component.html

@@ -0,0 +1,13 @@
+<div class="sidebar">
+    <ul class="nav flex-column">
+        <li class="nav-item">
+            <a class="nav-link" routerLink="event-consumers" routerLinkActive="active">
+                <i class="nav-icon icon-time"></i> <div class="nav-text">Event Consumers</div>
+            </a>
+        </li>
+    </ul>
+</div>
+
+<div class="panel-container">
+    <router-outlet></router-outlet>
+</div>

src/Squidex/app/features/administration/administration-area.component.scss

@@ -0,0 +1,46 @@
+@import '_vars';
+@import '_mixins';
+
+.sidebar {
+    @include fixed($size-navbar-height, auto, 0, 0);
+    @include box-shadow-colored(2px, 0, 0, $color-dark1-border2);
+    min-width: $size-sidebar-width;
+    max-width: $size-sidebar-width;
+    border-right: 1px solid $color-dark1-border1;
+    background: $color-dark1-background;
+    z-index: 100;
+}
+
+.nav {
+    &-icon {
+        font-size: 2rem;
+    }
+
+    &-text {
+        font-size: .9rem;
+    }
+
+    &-link {
+        & {
+            @include transition(color .3s ease);
+            padding: 1.25rem;
+            display: block;
+            text-align: center;
+            text-decoration: none;
+            color: $color-dark1-foreground;
+        }
+
+        &:hover,
+        &.active {
+            color: $color-dark1-focus-foreground;
+
+            .nav-icon {
+                color: $color-theme-blue;
+            }
+        }
+
+        &.active {
+            background: $color-dark1-active-background;
+        }
+    }
+}

src/Squidex/app/features/administration/administration-area.component.ts

@@ -0,0 +1,17 @@
+/*
+ * Squidex Headless CMS
+ *
+ * @license
+ * Copyright (c) Sebastian Stehle. All rights reserved
+ */
+
+import { Component } from '@angular/core';
+
+@Component({
+    selector: 'sqx-administration-area',
+    styleUrls: ['./administration-area.component.scss'],
+    templateUrl: './administration-area.component.html'
+})
+export class AdministrationAreaComponent {
+
+}

src/Squidex/app/features/administration/declarations.ts

@@ -0,0 +1,10 @@
+/*
+ * Squidex Headless CMS
+ *
+ * @license
+ * Copyright (c) Sebastian Stehle. All rights reserved
+ */
+
+export * from './pages/event-consumers/event-consumers-page.component';
+
+export * from './administration-area.component';

src/Squidex/app/features/administration/module.ts

@@ -0,0 +1,48 @@
+/*
+ * Squidex Headless CMS
+ *
+ * @license
+ * Copyright (c) Sebastian Stehle. All rights reserved
+ */
+
+import { NgModule } from '@angular/core';
+import { RouterModule, Routes } from '@angular/router';
+
+import {
+    SqxFrameworkModule,
+    SqxSharedModule
+} from 'shared';
+
+import {
+    AdministrationAreaComponent,
+    EventConsumersPage
+} from './declarations';
+
+const routes: Routes = [
+    {
+        path: '',
+        component: AdministrationAreaComponent,
+        children: [
+            {
+                path: '',
+                children: [{
+                    path: 'event-consumers',
+                    component: EventConsumersPage
+                }]
+            }
+        ]
+    }
+];
+
+@NgModule({
+    imports: [
+        SqxFrameworkModule,
+        SqxSharedModule,
+        RouterModule.forChild(routes)
+    ],
+    declarations: [
+        AdministrationAreaComponent,
+        EventConsumersPage
+    ]
+})
+export class SqxFeatureAdministrationModule { }

src/Squidex/app/features/administration/pages/event-consumers/event-consumers-page.component.html

@@ -0,0 +1,64 @@
+<sqx-title message="Event Consumers"></sqx-title>
+
+<sqx-panel theme="light" panelWidth="50rem">
+    <div class="panel-header">
+        <div class="panel-title-row">
+            <h3 class="panel-title">EventConsumers</h3>
+        </div>
+
+        <a class="panel-close" routerLink="['..']">
+            <i class="icon-close"></i>
+        </a>
+    </div>
+
+    <div class="panel-main">
+        <div class="panel-content panel-content-scroll">
+            <table class="table table-items table-fixed">
+                <colgroup>
+                    <col style="width: 100%" />
+                    <col style="width: 160px" />
+                    <col style="width: 160px" />
+                </colgroup>
+
+                <thead>
+                    <tr>
+                        <th>
+                            Name
+                        </th>
+                        <th class="col-right">
+                            Event Number
+                        </th>
+                        <th class="col-right">
+                            Options
+                        </th>
+                    </tr>
+                </thead>
+
+                <tbody>
+                    <template ngFor let-eventConsumer [ngForOf]="eventConsumers">
+                        <tr>
+                            <td>
+                                <span class="truncate">{{eventConsumer.name}}</span>
+                            </td>
+                            <td class="col-right">
+                                <span>{{eventConsumer.lastHandledEventNumber}}</span>
+                            </td>
+                            <td class="col-right">
+                                <button class="btn btn-simple" (click)="reset(eventConsumer.name)" *ngIf="!eventConsumer.isResetting">
+                                    <i class="icon icon-reset"></i>
+                                </button>
+                                <button class="btn btn-simple" (click)="start(eventConsumer.name)" *ngIf="eventConsumer.isStopped">
+                                    <i class="icon icon-play"></i>
+                                </button>
+                                <button class="btn btn-simple" (click)="stop(eventConsumer.name)" *ngIf="!eventConsumer.isStopped">
+                                    <i class="icon icon-pause"></i>
+                                </button>
+                            </td>
+                        </tr>
+                        <tr class="spacer"></tr>
+                    </template>
+                </tbody>
+            </table>
+        </div>
+    </div>
+</sqx-panel>

src/Squidex/app/features/administration/pages/event-consumers/event-consumers-page.component.scss

@@ -0,0 +1,10 @@
+@import '_vars';
+@import '_mixins';
+
+button {
+    display: inline-block;
+}
+
+.col-right {
+    text-align: right;
+}

src/Squidex/app/features/administration/pages/event-consumers/event-consumers-page.component.ts

@@ -0,0 +1,84 @@
+/*
+ * Squidex Headless CMS
+ *
+ * @license
+ * Copyright (c) Sebastian Stehle. All rights reserved
+ */
+
+import { Component, OnDestroy, OnInit } from '@angular/core';
+import { Observable, Subscription } from 'rxjs';
+
+import {
+    EventConsumerDto,
+    EventConsumersService,
+    ImmutableArray
+} from 'shared';
+
+@Component({
+    selector: 'sqx-event-consumers-page',
+    styleUrls: ['./event-consumers-page.component.scss'],
+    templateUrl: './event-consumers-page.component.html'
+})
+export class EventConsumersPage implements OnInit, OnDestroy {
+    private subscription: Subscription;
+
+    public eventConsumers = ImmutableArray.empty<EventConsumerDto>();
+
+    constructor(
+        private readonly eventConsumersService: EventConsumersService
+    ) {
+    }
+
+    public ngOnInit() {
+        this.subscription =
+            Observable.timer(0, 4000)
+                .switchMap(_ => this.eventConsumersService.getEventConsumers())
+                .subscribe(dtos => {
+                    this.eventConsumers = ImmutableArray.of(dtos);
+                });
+    }
+
+    public ngOnDestroy() {
+        this.subscription.unsubscribe();
+    }
+
+    public start(name: string) {
+        this.eventConsumersService.startEventConsumer(name)
+            .subscribe(() => {
+                this.eventConsumers = this.eventConsumers.map(e => {
+                    if (e.name === name) {
+                        return new EventConsumerDto(name, e.lastHandledEventNumber, false, e.isResetting);
+                    } else {
+                        return e;
+                    }
+                });
+            });
+    }
+
+    public stop(name: string) {
+        this.eventConsumersService.stopEventConsumer(name)
+            .subscribe(() => {
+                this.eventConsumers = this.eventConsumers.map(e => {
+                    if (e.name === name) {
+                        return new EventConsumerDto(name, e.lastHandledEventNumber, true, e.isResetting);
+                    } else {
+                        return e;
+                    }
+                });
+            });
+    }
+
+    public reset(name: string) {
+        this.eventConsumersService.resetEventConsumer(name)
+            .subscribe(() => {
+                this.eventConsumers = this.eventConsumers.map(e => {
+                    if (e.name === name) {
+                        return new EventConsumerDto(name, e.lastHandledEventNumber, e.isStopped, true);
+                    } else {
+                        return e;
+                    }
+                });
+            });
+    }
+}
+

src/Squidex/app/shared/declarations.ts

@@ -26,6 +26,7 @@ export * from './services/apps-store.service';
 export * from './services/apps.service';
 export * from './services/auth.service';
 export * from './services/contents.service';
+export * from './services/event-consumers.service';
 export * from './services/history.service';
 export * from './services/languages.service';
 export * from './services/schemas.service';

src/Squidex/app/shared/module.ts

@@ -20,6 +20,7 @@ import {
     AuthService,
     ContentsService,
     DashboardLinkDirective,
+    EventConsumersService,
     HistoryComponent,
     HistoryService,
     LanguageSelectorComponent,
@@ -65,6 +66,7 @@ export class SqxSharedModule {
                 AppMustExistGuard,
                 AuthService,
                 ContentsService,
+                EventConsumersService,
                 HistoryService,
                 LanguageService,
                 MustBeAuthenticatedGuard,

src/Squidex/app/shared/services/auth.service.ts

@@ -31,6 +31,10 @@ export class Profile {
         return this.user.profile['urn:squidex:picture'];
     }
 
+    public get isAdmin(): boolean {
+        return this.user.profile['role'] === 'administrator';
+    }
+
     public get token(): string {
         return `subject:${this.id}`;
     }
@@ -74,7 +78,7 @@ export class AuthService {
 
         this.userManager = new UserManager({
                        client_id: 'squidex-frontend',
-                           scope: 'squidex-api openid profile squidex-profile',
+                           scope: 'squidex-api openid profile squidex-profile role',
                    response_type: 'id_token token',
                     redirect_uri: apiUrl.buildUrl('login;'),
         post_logout_redirect_uri: apiUrl.buildUrl('logout'),

src/Squidex/app/shared/services/event-consumers.service.ts

@@ -0,0 +1,76 @@
+/*
+ * Squidex Headless CMS
+ *
+ * @license
+ * Copyright (c) Sebastian Stehle. All rights reserved
+ */
+
+import { Injectable } from '@angular/core';
+import { Observable } from 'rxjs';
+
+import 'framework/angular/http-extensions';
+
+import { ApiUrlConfig } from 'framework';
+import { AuthService } from './auth.service';
+
+export class EventConsumerDto {
+    constructor(
+        public readonly name: string,
+        public readonly lastHandledEventNumber: number,
+        public readonly isStopped: boolean,
+        public readonly isResetting: boolean
+    ) {
+    }
+}
+
+@Injectable()
+export class EventConsumersService {
+    constructor(
+        private readonly authService: AuthService,
+        private readonly apiUrl: ApiUrlConfig
+    ) {
+    }
+
+    public getEventConsumers(): Observable<EventConsumerDto[]> {
+        const url = this.apiUrl.buildUrl('/api/event-consumers');
+
+        return this.authService.authGet(url)
+                .map(response => response.json())
+                .map(response => {
+                    const items: any[] = response;
+
+                    return items.map(item => {
+                        return new EventConsumerDto(
+                            item.name,
+                            item.lastHandledEventNumber,
+                            item.isStopped,
+                            item.isResetting);
+                    });
+                })
+                .catchError('Failed to load event consumers. Please reload.');
+    }
+
+    public startEventConsumer(name: string): Observable<any> {
+        const url = this.apiUrl.buildUrl(`api/event-consumers/${name}/start`);
+
+        return this.authService.authPut(url, {})
+                .map(response => response.json())
+                .catchError('Failed to start event consumer. Please reload.');
+    }
+
+    public stopEventConsumer(name: string): Observable<any> {
+        const url = this.apiUrl.buildUrl(`api/event-consumers/${name}/stop`);
+
+        return this.authService.authPut(url, {})
+                .map(response => response.json())
+                .catchError('Failed to stop event consumer. Please reload.');
+    }
+
+    public resetEventConsumer(name: string): Observable<any> {
+        const url = this.apiUrl.buildUrl(`api/event-consumers/${name}/reset`);
+
+        return this.authService.authPut(url, {})
+                .map(response => response.json())
+                .catchError('Failed to reset event consumer. Please reload.');
+    }
+}

src/Squidex/app/shell/pages/internal/profile-menu.component.html

@@ -7,7 +7,13 @@
         </span>
 
         <div class="dropdown-menu" *sqxModalView="modalMenu" closeAlways="true" [@fade]>
-            <a class="dropdown-item" (click)="logout()">Logout</a>
+            <a class="dropdown-item"  [routerLink]="['/app/administration']" *ngIf="isAdmin">
+                Administration
+            </a>
+
+            <a class="dropdown-item" (click)="logout()">
+                Logout
+            </a>
         </div>
     </li>
 </ul>

src/Squidex/app/shell/pages/internal/profile-menu.component.ts

@@ -30,6 +30,8 @@ export class ProfileMenuComponent implements OnInit, OnDestroy {
     public profileDisplayName = '';
     public profilePictureUrl = '';
 
+    public isAdmin = false;
+
     constructor(
         private readonly auth: AuthService
     ) {
@@ -41,12 +43,14 @@ export class ProfileMenuComponent implements OnInit, OnDestroy {
 
     public ngOnInit() {
         this.authenticationSubscription =
-            this.auth.isAuthenticated.subscribe(() => {
+            this.auth.isAuthenticated.take(1).subscribe(() => {
                 const user = this.auth.user;
 
                 if (user) {
                     this.profilePictureUrl = user.pictureUrl;
                     this.profileDisplayName = user.displayName;
+
+                    this.isAdmin = user.isAdmin;
                 }
             });
     }

src/Squidex/app/theme/icomoon/fonts/icomoon.svg

@@ -53,4 +53,7 @@
 <glyph unicode="&#xe92b;" glyph-name="caret-up" horiz-adv-x="585" d="M585.143 256c0-20-16.571-36.571-36.571-36.571h-512c-20 0-36.571 16.571-36.571 36.571 0 9.714 4 18.857 10.857 25.714l256 256c6.857 6.857 16 10.857 25.714 10.857s18.857-4 25.714-10.857l256-256c6.857-6.857 10.857-16 10.857-25.714z" />
 <glyph unicode="&#xe92c;" glyph-name="caret-down" horiz-adv-x="585" d="M585.143 548.571c0-9.714-4-18.857-10.857-25.714l-256-256c-6.857-6.857-16-10.857-25.714-10.857s-18.857 4-25.714 10.857l-256 256c-6.857 6.857-10.857 16-10.857 25.714 0 20 16.571 36.571 36.571 36.571h512c20 0 36.571-16.571 36.571-36.571z" />
 <glyph unicode="&#xe92d;" glyph-name="settings2" d="M933.79 349.75c-53.726 93.054-21.416 212.304 72.152 266.488l-100.626 174.292c-28.75-16.854-62.176-26.518-97.846-26.518-107.536 0-194.708 87.746-194.708 195.99h-201.258c0.266-33.41-8.074-67.282-25.958-98.252-53.724-93.056-173.156-124.702-266.862-70.758l-100.624-174.292c28.97-16.472 54.050-40.588 71.886-71.478 53.638-92.908 21.512-211.92-71.708-266.224l100.626-174.292c28.65 16.696 61.916 26.254 97.4 26.254 107.196 0 194.144-87.192 194.7-194.958h201.254c-0.086 33.074 8.272 66.57 25.966 97.218 53.636 92.906 172.776 124.594 266.414 71.012l100.626 174.29c-28.78 16.466-53.692 40.498-71.434 71.228zM512 240.668c-114.508 0-207.336 92.824-207.336 207.334 0 114.508 92.826 207.334 207.336 207.334 114.508 0 207.332-92.826 207.332-207.334-0.002-114.51-92.824-207.334-207.332-207.334z" />
+<glyph unicode="&#xe92e;" glyph-name="reset" d="M889.68 793.68c-93.608 102.216-228.154 166.32-377.68 166.32-282.77 0-512-229.23-512-512h96c0 229.75 186.25 416 416 416 123.020 0 233.542-53.418 309.696-138.306l-149.696-149.694h352v352l-134.32-134.32zM928 448c0-229.75-186.25-416-416-416-123.020 0-233.542 53.418-309.694 138.306l149.694 149.694h-352v-352l134.32 134.32c93.608-102.216 228.154-166.32 377.68-166.32 282.77 0 512 229.23 512 512h-96z" />
+<glyph unicode="&#xe92f;" glyph-name="pause" d="M128 832h320v-768h-320zM576 832h320v-768h-320z" />
+<glyph unicode="&#xe930;" glyph-name="play" d="M192 832l640-384-640-384z" />
 </font></defs></svg>

src/Squidex/app/theme/icomoon/selection.json

@@ -1,6 +1,105 @@
 {
   "IcoMoonType": "selection",
   "icons": [
+    {
+      "icon": {
+        "paths": [
+          "M889.68 166.32c-93.608-102.216-228.154-166.32-377.68-166.32-282.77 0-512 229.23-512 512h96c0-229.75 186.25-416 416-416 123.020 0 233.542 53.418 309.696 138.306l-149.696 149.694h352v-352l-134.32 134.32z",
+          "M928 512c0 229.75-186.25 416-416 416-123.020 0-233.542-53.418-309.694-138.306l149.694-149.694h-352v352l134.32-134.32c93.608 102.216 228.154 166.32 377.68 166.32 282.77 0 512-229.23 512-512h-96z"
+        ],
+        "attrs": [
+          {},
+          {}
+        ],
+        "isMulticolor": false,
+        "isMulticolor2": false,
+        "tags": [
+          "loop",
+          "repeat",
+          "player",
+          "reload",
+          "refresh",
+          "update",
+          "synchronize",
+          "arrows"
+        ],
+        "grid": 16
+      },
+      "attrs": [
+        {},
+        {}
+      ],
+      "properties": {
+        "order": 1,
+        "id": 2,
+        "prevSize": 32,
+        "code": 59694,
+        "name": "reset"
+      },
+      "setIdx": 0,
+      "setId": 2,
+      "iconIdx": 0
+    },
+    {
+      "icon": {
+        "paths": [
+          "M128 128h320v768h-320zM576 128h320v768h-320z"
+        ],
+        "attrs": [
+          {}
+        ],
+        "isMulticolor": false,
+        "isMulticolor2": false,
+        "tags": [
+          "pause",
+          "player"
+        ],
+        "grid": 16
+      },
+      "attrs": [
+        {}
+      ],
+      "properties": {
+        "order": 2,
+        "id": 1,
+        "prevSize": 32,
+        "code": 59695,
+        "name": "pause"
+      },
+      "setIdx": 0,
+      "setId": 2,
+      "iconIdx": 1
+    },
+    {
+      "icon": {
+        "paths": [
+          "M192 128l640 384-640 384z"
+        ],
+        "attrs": [
+          {}
+        ],
+        "isMulticolor": false,
+        "isMulticolor2": false,
+        "tags": [
+          "play",
+          "player"
+        ],
+        "grid": 16
+      },
+      "attrs": [
+        {}
+      ],
+      "properties": {
+        "order": 3,
+        "id": 0,
+        "prevSize": 32,
+        "code": 59696,
+        "name": "play"
+      },
+      "setIdx": 0,
+      "setId": 2,
+      "iconIdx": 2
+    },
     {
       "icon": {
         "paths": [
@@ -32,8 +131,8 @@
         "code": 59693,
         "name": "settings2"
       },
-      "setIdx": 0,
-      "setId": 2,
+      "setIdx": 1,
+      "setId": 1,
       "iconIdx": 0
     },
     {
@@ -69,8 +168,8 @@
         "prevSize": 32,
         "code": 59650
       },
-      "setIdx": 0,
-      "setId": 2,
+      "setIdx": 1,
+      "setId": 1,
       "iconIdx": 1
     },
     {
@@ -100,7 +199,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 0
+      "iconIdx": 2
     },
     {
       "icon": {
@@ -129,7 +228,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 1
+      "iconIdx": 3
     },
     {
       "icon": {
@@ -158,7 +257,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 2
+      "iconIdx": 4
     },
     {
       "icon": {
@@ -187,7 +286,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 3
+      "iconIdx": 5
     },
     {
       "icon": {
@@ -216,7 +315,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 4
+      "iconIdx": 6
     },
     {
       "icon": {
@@ -245,7 +344,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 5
+      "iconIdx": 7
     },
     {
       "icon": {
@@ -274,7 +373,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 6
+      "iconIdx": 8
     },
     {
       "icon": {
@@ -303,7 +402,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 7
+      "iconIdx": 9
     },
     {
       "icon": {
@@ -332,7 +431,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 8
+      "iconIdx": 10
     },
     {
       "icon": {
@@ -361,7 +460,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 9
+      "iconIdx": 11
     },
     {
       "icon": {
@@ -390,7 +489,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 10
+      "iconIdx": 12
     },
     {
       "icon": {
@@ -419,7 +518,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 11
+      "iconIdx": 13
     },
     {
       "icon": {
@@ -448,7 +547,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 12
+      "iconIdx": 14
     },
     {
       "icon": {
@@ -477,7 +576,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 13
+      "iconIdx": 15
     },
     {
       "icon": {
@@ -506,7 +605,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 14
+      "iconIdx": 16
     },
     {
       "icon": {
@@ -535,7 +634,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 15
+      "iconIdx": 17
     },
     {
       "icon": {
@@ -564,7 +663,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 16
+      "iconIdx": 18
     },
     {
       "icon": {
@@ -593,7 +692,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 17
+      "iconIdx": 19
     },
     {
       "icon": {
@@ -622,7 +721,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 18
+      "iconIdx": 20
     },
     {
       "icon": {
@@ -651,7 +750,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 19
+      "iconIdx": 21
     },
     {
       "icon": {
@@ -680,7 +779,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 20
+      "iconIdx": 22
     },
     {
       "icon": {
@@ -709,7 +808,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 21
+      "iconIdx": 23
     },
     {
       "icon": {
@@ -738,7 +837,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 22
+      "iconIdx": 24
     },
     {
       "icon": {
@@ -767,7 +866,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 23
+      "iconIdx": 25
     },
     {
       "icon": {
@@ -796,7 +895,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 24
+      "iconIdx": 26
     },
     {
       "icon": {
@@ -825,7 +924,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 25
+      "iconIdx": 27
     },
     {
       "icon": {
@@ -854,7 +953,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 26
+      "iconIdx": 28
     },
     {
       "icon": {
@@ -883,7 +982,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 27
+      "iconIdx": 29
     },
     {
       "icon": {
@@ -912,7 +1011,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 28
+      "iconIdx": 30
     },
     {
       "icon": {
@@ -941,7 +1040,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 29
+      "iconIdx": 31
     },
     {
       "icon": {
@@ -970,7 +1069,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 30
+      "iconIdx": 32
     },
     {
       "icon": {
@@ -999,7 +1098,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 31
+      "iconIdx": 33
     },
     {
       "icon": {
@@ -1028,7 +1127,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 32
+      "iconIdx": 34
     },
     {
       "icon": {
@@ -1057,7 +1156,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 33
+      "iconIdx": 35
     },
     {
       "icon": {
@@ -1086,7 +1185,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 34
+      "iconIdx": 36
     },
     {
       "icon": {
@@ -1115,7 +1214,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 35
+      "iconIdx": 37
     },
     {
       "icon": {
@@ -1144,7 +1243,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 36
+      "iconIdx": 38
     },
     {
       "icon": {
@@ -1174,7 +1273,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 37
+      "iconIdx": 39
     },
     {
       "icon": {
@@ -1204,7 +1303,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 38
+      "iconIdx": 40
     },
     {
       "icon": {
@@ -1234,7 +1333,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 39
+      "iconIdx": 41
     },
     {
       "icon": {
@@ -1264,7 +1363,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 40
+      "iconIdx": 42
     },
     {
       "icon": {
@@ -1294,7 +1393,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 41
+      "iconIdx": 43
     },
     {
       "icon": {
@@ -1324,7 +1423,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 42
+      "iconIdx": 44
     },
     {
       "icon": {
@@ -1354,7 +1453,7 @@
       },
       "setIdx": 1,
       "setId": 1,
-      "iconIdx": 43
+      "iconIdx": 45
     }
   ],
   "height": 1024,

src/Squidex/app/theme/icomoon/style.css

@@ -1,10 +1,10 @@
 @font-face {
   font-family: 'icomoon';
-  src:  url('fonts/icomoon.eot?yajqm9');
-  src:  url('fonts/icomoon.eot?yajqm9#iefix') format('embedded-opentype'),
-    url('fonts/icomoon.ttf?yajqm9') format('truetype'),
-    url('fonts/icomoon.woff?yajqm9') format('woff'),
-    url('fonts/icomoon.svg?yajqm9#icomoon') format('svg');
+  src:  url('fonts/icomoon.eot?oo54rp');
+  src:  url('fonts/icomoon.eot?oo54rp#iefix') format('embedded-opentype'),
+    url('fonts/icomoon.ttf?oo54rp') format('truetype'),
+    url('fonts/icomoon.woff?oo54rp') format('woff'),
+    url('fonts/icomoon.svg?oo54rp#icomoon') format('svg');
   font-weight: normal;
   font-style: normal;
 }
@@ -24,6 +24,15 @@
   -moz-osx-font-smoothing: grayscale;
 }
 
+.icon-reset:before {
+  content: "\e92e";
+}
+.icon-pause:before {
+  content: "\e92f";
+}
+.icon-play:before {
+  content: "\e930";
+}
 .icon-settings2:before {
   content: "\e92d";
 }

src/Squidex/wwwroot/index.html

@@ -11,6 +11,7 @@
     <style>
         body {
             background: #F4F8F9;
+            margin: 0;
             padding-top: 3.25rem;
             padding-left: 7rem;
             line-height: 1.5;
@@ -57,9 +58,6 @@
     </noscript>
 </head>
 <body>
-    <noscript>
-        You must enable Javascript to use the Squidex Portal
-    </noscript>
 
     <sqx-app>
         <div class="loading">

tests/Squidex.Infrastructure.Tests/CQRS/Events/EventReceiverTests.cs

@@ -14,14 +14,27 @@ using Microsoft.Extensions.Logging;
 using Moq;
 using Xunit;
 
+// ReSharper disable UnusedAutoPropertyAccessor.Local
+
 namespace Squidex.Infrastructure.CQRS.Events
 {
-    public class EventReceiverTests
+    public class EventReceiverTests : IDisposable
     {
         public sealed class MyEvent : IEvent
         {
         }
 
+        private sealed class MyEventConsumerInfo : IEventConsumerInfo
+        {
+            public long LastHandledEventNumber { get; set; }
+
+            public bool IsStopped { get; set; }
+
+            public bool IsResetting { get; set; }
+
+            public string Name { get; set; }
+        }
+
         private sealed class MyLogger : ILogger<EventReceiver>
         {
             public Dictionary<LogLevel, int> LogCount { get; } = new Dictionary<LogLevel, int>();
@@ -44,45 +57,48 @@ namespace Squidex.Infrastructure.CQRS.Events
             }
         }
 
-        private readonly Mock<IEventCatchConsumer> eventConsumer = new Mock<IEventCatchConsumer>();
+        private readonly Mock<IEventConsumerInfoRepository> eventConsumerInfoRepository = new Mock<IEventConsumerInfoRepository>();
+        private readonly Mock<IEventConsumer> eventConsumer = new Mock<IEventConsumer>();
         private readonly Mock<IEventNotifier> eventNotifier = new Mock<IEventNotifier>();
         private readonly Mock<IEventStore> eventStore = new Mock<IEventStore>();
         private readonly Mock<EventDataFormatter> formatter = new Mock<EventDataFormatter>(new TypeNameRegistry(), null);
         private readonly EventData eventData1 = new EventData();
         private readonly EventData eventData2 = new EventData();
         private readonly EventData eventData3 = new EventData();
-        private readonly EventData eventData4 = new EventData();
         private readonly Envelope<IEvent> envelope1 = new Envelope<IEvent>(new MyEvent());
         private readonly Envelope<IEvent> envelope2 = new Envelope<IEvent>(new MyEvent());
         private readonly Envelope<IEvent> envelope3 = new Envelope<IEvent>(new MyEvent());
-        private readonly Envelope<IEvent> envelope4 = new Envelope<IEvent>(new MyEvent());
         private readonly EventReceiver sut;
         private readonly MyLogger logger = new MyLogger();
-        private readonly StoredEvent[][] events;
+        private readonly StoredEvent[] events;
+        private readonly MyEventConsumerInfo consumerInfo = new MyEventConsumerInfo();
+        private readonly string consumerName;
 
         public EventReceiverTests()
         {
             events = new[]
-            {
-                new []
             {
                 new StoredEvent(3, eventData1),
-                    new StoredEvent(4, eventData1)
-                },
-                new[]
-                {
-                    new StoredEvent(5, eventData1),
-                    new StoredEvent(6, eventData1)
-                }
+                new StoredEvent(4, eventData2),
+                new StoredEvent(4, eventData3)
             };
 
+            consumerName = eventConsumer.Object.GetType().Name;
+
+            eventStore.Setup(x => x.GetEventsAsync(2)).Returns(events.ToObservable());
+
+            eventConsumerInfoRepository.Setup(x => x.FindAsync(consumerName)).Returns(Task.FromResult<IEventConsumerInfo>(consumerInfo));
 
             formatter.Setup(x => x.Parse(eventData1)).Returns(envelope1);
             formatter.Setup(x => x.Parse(eventData2)).Returns(envelope2);
             formatter.Setup(x => x.Parse(eventData3)).Returns(envelope3);
-            formatter.Setup(x => x.Parse(eventData4)).Returns(envelope4);
 
-            sut = new EventReceiver(formatter.Object, eventStore.Object, eventNotifier.Object, logger);
+            sut = new EventReceiver(formatter.Object, eventStore.Object, eventNotifier.Object, eventConsumerInfoRepository.Object, logger);
+        }
+
+        public void Dispose()
+        {
+            sut.Dispose();
         }
 
         [Fact]
@@ -91,70 +107,91 @@ namespace Squidex.Infrastructure.CQRS.Events
             sut.Subscribe(eventConsumer.Object);
             sut.Subscribe(eventConsumer.Object);
 
-            eventConsumer.Verify(x => x.GetLastHandledEventNumber(), Times.Once());
+            eventConsumerInfoRepository.Verify(x => x.CreateAsync(consumerName), Times.Once());
         }
 
         [Fact]
-        public void Should_subscribe_to_consumers_and_handle_events()
+        public async Task Should_subscribe_to_consumers_and_handle_events()
         {
-            eventConsumer.Setup(x => x.GetLastHandledEventNumber()).Returns(Task.FromResult(2L));
-            eventConsumer.Setup(x => x.On(It.IsAny<Envelope<IEvent>>(), It.IsAny<long>())).Returns(Task.FromResult(true));
-
-            eventStore.Setup(x => x.GetEventsAsync(2)).Returns(events[0].ToObservable());
-            eventStore.Setup(x => x.GetEventsAsync(4)).Returns(events[1].ToObservable());
-
-            eventStore.Setup(x => x.GetEventsAsync(It.Is<long>(l => l != 2 && l != 4))).Returns(Observable.Empty<StoredEvent>());
+            consumerInfo.LastHandledEventNumber = 2L;
             
-            sut.Subscribe(eventConsumer.Object, 20);
+            sut.Subscribe(eventConsumer.Object);
 
-            Task.Delay(400).ContinueWith(x => sut.Dispose()).Wait();
+            await Task.Delay(20);
 
             Assert.Equal(1, logger.LogCount.Count);
-            Assert.Equal(4, logger.LogCount[LogLevel.Debug]);
+            Assert.Equal(3, logger.LogCount[LogLevel.Debug]);
 
-            eventConsumer.Verify(x => x.On(It.IsAny<Envelope<IEvent>>(), It.IsAny<long>()), Times.Exactly(4));
+            eventConsumer.Verify(x => x.On(envelope1), Times.Once());
+            eventConsumer.Verify(x => x.On(envelope2), Times.Once());
+            eventConsumer.Verify(x => x.On(envelope3), Times.Once());
         }
 
         [Fact]
-        public void Should_abort_if_handling_failed()
+        public async Task Should_abort_if_handling_failed()
         {
-            eventConsumer.Setup(x => x.GetLastHandledEventNumber()).Returns(Task.FromResult(2L));
-            eventConsumer.Setup(x => x.On(It.IsAny<Envelope<IEvent>>(), It.IsAny<long>())).Throws<InvalidOperationException>();
+            consumerInfo.LastHandledEventNumber = 2L;
 
-            eventStore.Setup(x => x.GetEventsAsync(2)).Returns(events[0].ToObservable());
-            eventStore.Setup(x => x.GetEventsAsync(It.Is<long>(l => l != 2 && l != 4))).Returns(Observable.Empty<StoredEvent>());
+            eventConsumer.Setup(x => x.On(envelope1)).Returns(Task.FromResult(true));
+            eventConsumer.Setup(x => x.On(envelope2)).Throws(new InvalidOperationException());
 
-            sut.Subscribe(eventConsumer.Object, 20);
+            sut.Subscribe(eventConsumer.Object);
 
-            Task.Delay(400).ContinueWith(x => sut.Dispose()).Wait();
+            await Task.Delay(20);
 
             Assert.Equal(2, logger.LogCount.Count);
-            Assert.Equal(1, logger.LogCount[LogLevel.Error]);
-            Assert.Equal(1, logger.LogCount[LogLevel.Critical]);
+            Assert.Equal(2, logger.LogCount[LogLevel.Error]);
+            Assert.Equal(1, logger.LogCount[LogLevel.Debug]);
+
+            eventConsumer.Verify(x => x.On(envelope1), Times.Once());
+            eventConsumer.Verify(x => x.On(envelope2), Times.Once());
+            eventConsumer.Verify(x => x.On(envelope3), Times.Never());
 
-            eventConsumer.Verify(x => x.On(It.IsAny<Envelope<IEvent>>(), It.IsAny<long>()), Times.Exactly(1));
-            eventStore.Verify(x => x.GetEventsAsync(It.IsAny<long>()), Times.Exactly(1));
+            eventConsumerInfoRepository.Verify(x => x.StopAsync(consumerName), Times.Once());
         }
 
         [Fact]
-        public void Should_abort_if_serialization_failed()
+        public async Task Should_abort_if_serialization_failed()
         {
-            eventConsumer.Setup(x => x.GetLastHandledEventNumber()).Returns(Task.FromResult(2L));
-            eventConsumer.Setup(x => x.On(It.IsAny<Envelope<IEvent>>(), It.IsAny<long>())).Throws<InvalidOperationException>();
+            consumerInfo.LastHandledEventNumber = 2L;
 
-            eventStore.Setup(x => x.GetEventsAsync(2)).Returns(events[0].ToObservable());
-            eventStore.Setup(x => x.GetEventsAsync(It.Is<long>(l => l != 2 && l != 4))).Returns(Observable.Empty<StoredEvent>());
+            formatter.Setup(x => x.Parse(eventData2)).Throws(new InvalidOperationException());
 
-            sut.Subscribe(eventConsumer.Object, 20);
+            sut.Subscribe(eventConsumer.Object);
 
-            Task.Delay(400).ContinueWith(x => sut.Dispose()).Wait();
+            await Task.Delay(20);
 
             Assert.Equal(2, logger.LogCount.Count);
-            Assert.Equal(1, logger.LogCount[LogLevel.Error]);
-            Assert.Equal(1, logger.LogCount[LogLevel.Critical]);
+            Assert.Equal(2, logger.LogCount[LogLevel.Error]);
+            Assert.Equal(1, logger.LogCount[LogLevel.Debug]);
+
+            eventConsumer.Verify(x => x.On(envelope1), Times.Once());
+            eventConsumer.Verify(x => x.On(envelope2), Times.Never());
+            eventConsumer.Verify(x => x.On(envelope3), Times.Never());
+
+            eventConsumerInfoRepository.Verify(x => x.StopAsync(consumerName), Times.Once());
+        }
+
+        [Fact]
+        public async Task Should_reset_if_requested()
+        {
+            consumerInfo.IsResetting = true;
+            consumerInfo.LastHandledEventNumber = 2L;
+            
+            eventStore.Setup(x => x.GetEventsAsync(-1)).Returns(events.ToObservable());
+
+            sut.Subscribe(eventConsumer.Object);
+
+            await Task.Delay(20);
+
+            Assert.Equal(1, logger.LogCount.Count);
+            Assert.Equal(5, logger.LogCount[LogLevel.Debug]);
+
+            eventConsumer.Verify(x => x.On(envelope1), Times.Once());
+            eventConsumer.Verify(x => x.On(envelope2), Times.Once());
+            eventConsumer.Verify(x => x.On(envelope3), Times.Once());
 
-            eventConsumer.Verify(x => x.On(It.IsAny<Envelope<IEvent>>(), It.IsAny<long>()), Times.Exactly(1));
-            eventStore.Verify(x => x.GetEventsAsync(It.IsAny<long>()), Times.Exactly(1));
+            eventConsumer.Verify(x => x.ClearAsync(), Times.Once());
         }
     }
 }