Auto clean old event consumers from MongoDB

src/Squidex.Infrastructure.MongoDb/CQRS/Events/MongoEventConsumerInfoRepository.cs

@@ -65,6 +65,11 @@ namespace Squidex.Infrastructure.CQRS.Events
             }
         }
 
+        public Task ClearAsync(IEnumerable<string> currentConsumerNames)
+        {
+            return Collection.DeleteManyAsync(Filter.Not(Filter.In(NameField, currentConsumerNames)));
+        }
+
         public Task StartAsync(string consumerName)
         {
             var filter = Filter.Eq(NameField, consumerName);

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

@@ -0,0 +1,36 @@
+// ==========================================================================
+//  EventConsumerCleaner.cs
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex Group
+//  All rights reserved.
+// ==========================================================================
+
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace Squidex.Infrastructure.CQRS.Events
+{
+    public sealed class EventConsumerCleaner
+    {
+        private readonly IEnumerable<IEventConsumer> eventConsumers;
+        private readonly IEventConsumerInfoRepository eventConsumerInfoRepository;
+
+        public EventConsumerCleaner(IEnumerable<IEventConsumer> eventConsumers, IEventConsumerInfoRepository eventConsumerInfoRepository)
+        {
+            Guard.NotNull(eventConsumers, nameof(eventConsumers));
+            Guard.NotNull(eventConsumerInfoRepository, nameof(eventConsumerInfoRepository));
+
+            this.eventConsumers = eventConsumers;
+            this.eventConsumerInfoRepository = eventConsumerInfoRepository;
+        }
+
+        public Task CleanAsync()
+        {
+            var names = eventConsumers.Select(x => x.Name).ToArray();
+
+            return eventConsumerInfoRepository.ClearAsync(names);
+        }
+    }
+}

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

@@ -17,6 +17,8 @@ namespace Squidex.Infrastructure.CQRS.Events
 
         Task<IEventConsumerInfo> FindAsync(string consumerName);
 
+        Task ClearAsync(IEnumerable<string> currentConsumerNames);
+
         Task CreateAsync(string consumerName);
 
         Task StartAsync(string consumerName);

src/Squidex/Config/Domain/InfrastructureModule.cs

@@ -131,6 +131,10 @@ namespace Squidex.Config.Domain
                 .As<IAssetThumbnailGenerator>()
                 .SingleInstance();
 
+            builder.RegisterType<EventConsumerCleaner>()
+                .AsSelf()
+                .SingleInstance();
+
             builder.RegisterType<EventDataFormatter>()
                 .AsSelf()
                 .SingleInstance();

src/Squidex/Config/Domain/Usages.cs

@@ -18,6 +18,8 @@ namespace Squidex.Config.Domain
     {
         public static IApplicationBuilder UseMyEventStore(this IApplicationBuilder app)
         {
+            app.ApplicationServices.GetService<EventConsumerCleaner>().CleanAsync().Wait();
+
             var catchConsumers = app.ApplicationServices.GetServices<IEventConsumer>();
 
             foreach (var catchConsumer in catchConsumers)

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

@@ -0,0 +1,36 @@
+// ==========================================================================
+//  EventConsumerCleanerTests.cs
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex Group
+//  All rights reserved.
+// ==========================================================================
+
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using FakeItEasy;
+using Xunit;
+
+namespace Squidex.Infrastructure.CQRS.Events
+{
+    public class EventConsumerCleanerTests
+    {
+        [Fact]
+        public async Task Should_call_repository_with_all_names()
+        {
+            var eventConsumer1 = A.Fake<IEventConsumer>();
+            var eventConsumer2 = A.Fake<IEventConsumer>();
+
+            A.CallTo(() => eventConsumer1.Name).Returns("consumer1");
+            A.CallTo(() => eventConsumer2.Name).Returns("consumer2");
+
+            var repository = A.Fake<IEventConsumerInfoRepository>();
+
+            var sut = new EventConsumerCleaner(new[] { eventConsumer1, eventConsumer2 }, repository);
+
+            await sut.CleanAsync();
+
+            A.CallTo(() => repository.ClearAsync(A<IEnumerable<string>>.That.IsSameSequenceAs(new string[] { "consumer1", "consumer2" }))).MustHaveHappened();
+        }
+    }
+}