Cosmos db registered to DI.

src/Squidex.Infrastructure.Azure/Diagnostics/CosmosDbHealthCheck.cs

@@ -0,0 +1,32 @@
+// ==========================================================================
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex UG (haftungsbeschraenkt)
+//  All rights reserved. Licensed under the MIT license.
+// ==========================================================================
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Azure.Documents.Client;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+
+namespace Squidex.Infrastructure.Diagnostics
+{
+    public sealed class CosmosDbHealthCheck : IHealthCheck
+    {
+        private readonly DocumentClient documentClient;
+
+        public CosmosDbHealthCheck(Uri uri, string masterKey)
+        {
+            documentClient = new DocumentClient(uri, masterKey);
+        }
+
+        public async Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default)
+        {
+            await documentClient.ReadDatabaseFeedAsync();
+
+            return HealthCheckResult.Healthy("Application must query data from CosmosDB.");
+        }
+    }
+}

src/Squidex.Infrastructure.Azure/EventSourcing/CosmosDbEventStore.cs

@@ -15,12 +15,11 @@ using Newtonsoft.Json;
 
 namespace Squidex.Infrastructure.EventSourcing
 {
-    public sealed partial class CosmosDbEventStore : IEventStore, IInitializable
+    public sealed partial class CosmosDbEventStore : DisposableObjectBase, IEventStore, IInitializable
     {
         private readonly DocumentClient documentClient;
-        private readonly Uri databaseUri;
         private readonly Uri collectionUri;
-        private readonly Uri serviceUri;
+        private readonly Uri databaseUri;
         private readonly string masterKey;
         private readonly string databaseId;
         private readonly JsonSerializerSettings serializerSettings;
@@ -42,30 +41,36 @@ namespace Squidex.Infrastructure.EventSourcing
 
         public Uri ServiceUri
         {
-            get { return serviceUri; }
+            get { return documentClient.ServiceEndpoint; }
         }
 
-        public CosmosDbEventStore(Uri uri, string masterKey, JsonSerializerSettings serializerSettings, string database)
+        public CosmosDbEventStore(DocumentClient documentClient, string masterKey, string database, JsonSerializerSettings serializerSettings)
         {
-            Guard.NotNull(uri, nameof(uri));
+            Guard.NotNull(documentClient, nameof(documentClient));
             Guard.NotNull(serializerSettings, nameof(serializerSettings));
             Guard.NotNullOrEmpty(masterKey, nameof(masterKey));
             Guard.NotNullOrEmpty(database, nameof(database));
 
-            documentClient = new DocumentClient(uri, masterKey, serializerSettings);
+            this.documentClient = documentClient;
 
             databaseUri = UriFactory.CreateDatabaseUri(database);
             databaseId = database;
 
             collectionUri = UriFactory.CreateDocumentCollectionUri(database, Constants.Collection);
 
-            serviceUri = uri;
-
             this.masterKey = masterKey;
 
             this.serializerSettings = serializerSettings;
         }
 
+        protected override void DisposeObject(bool disposing)
+        {
+            if (disposing)
+            {
+                documentClient.Dispose();
+            }
+        }
+
         public async Task InitializeAsync(CancellationToken ct = default)
         {
             await documentClient.CreateDatabaseIfNotExistsAsync(new Database { Id = databaseId });

src/Squidex.Infrastructure.Azure/EventSourcing/CosmosDbEventStore_Reader.cs

@@ -19,20 +19,30 @@ namespace Squidex.Infrastructure.EventSourcing
 
     public partial class CosmosDbEventStore : IEventStore, IInitializable
     {
-        public IEventSubscription CreateSubscription(IEventSubscriber subscriber, string streamFilter, string position = null)
+        public IEventSubscription CreateSubscription(IEventSubscriber subscriber, string streamFilter = null, string position = null)
         {
             Guard.NotNull(subscriber, nameof(subscriber));
 
+            ThrowIfDisposed();
+
             return new CosmosDbSubscription(this, subscriber, streamFilter, position);
         }
 
         public Task CreateIndexAsync(string property)
         {
+            Guard.NotNullOrEmpty(property, nameof(property));
+
+            ThrowIfDisposed();
+
             return Task.CompletedTask;
         }
 
         public async Task<IReadOnlyList<StoredEvent>> QueryAsync(string streamName, long streamPosition = 0)
         {
+            Guard.NotNullOrEmpty(streamName, nameof(streamName));
+
+            ThrowIfDisposed();
+
             using (Profiler.TraceMethod<CosmosDbEventStore>())
             {
                 var query = FilterBuilder.ByStreamName(streamName, streamPosition - MaxCommitSize);
@@ -69,6 +79,10 @@ namespace Squidex.Infrastructure.EventSourcing
         public Task QueryAsync(Func<StoredEvent, Task> callback, string property, object value, string position = null, CancellationToken ct = default)
         {
             Guard.NotNull(callback, nameof(callback));
+            Guard.NotNullOrEmpty(property, nameof(property));
+            Guard.NotNull(value, nameof(value));
+
+            ThrowIfDisposed();
 
             StreamPosition lastPosition = position;
 
@@ -82,6 +96,8 @@ namespace Squidex.Infrastructure.EventSourcing
         {
             Guard.NotNull(callback, nameof(callback));
 
+            ThrowIfDisposed();
+
             StreamPosition lastPosition = position;
 
             var filterDefinition = FilterBuilder.CreateByFilter(streamFilter, lastPosition);

src/Squidex.Infrastructure.Azure/EventSourcing/CosmosDbEventStore_Writer.cs

@@ -24,6 +24,10 @@ namespace Squidex.Infrastructure.EventSourcing
 
         public Task DeleteStreamAsync(string streamName)
         {
+            Guard.NotNullOrEmpty(streamName, nameof(streamName));
+
+            ThrowIfDisposed();
+
             var query = FilterBuilder.AllIds(streamName);
 
             return documentClient.QueryAsync(collectionUri, query, commit =>
@@ -41,11 +45,13 @@ namespace Squidex.Infrastructure.EventSourcing
 
         public async Task AppendAsync(Guid commitId, string streamName, long expectedVersion, ICollection<EventData> events)
         {
-            Guard.GreaterEquals(expectedVersion, EtagVersion.Any, nameof(expectedVersion));
+            Guard.NotEmpty(commitId, nameof(commitId));
             Guard.NotNullOrEmpty(streamName, nameof(streamName));
             Guard.NotNull(events, nameof(events));
             Guard.LessThan(events.Count, MaxCommitSize, "events.Count");
 
+            ThrowIfDisposed();
+
             using (Profiler.TraceMethod<CosmosDbEventStore>())
             {
                 if (events.Count == 0)

src/Squidex.Infrastructure.Azure/EventSourcing/CosmosDbSubscription.cs

@@ -32,6 +32,8 @@ namespace Squidex.Infrastructure.EventSourcing
 
         public CosmosDbSubscription(CosmosDbEventStore store, IEventSubscriber subscriber, string streamFilter, string position = null)
         {
+            this.store = store;
+
             var fromBeginning = string.IsNullOrWhiteSpace(position);
 
             if (fromBeginning)
@@ -48,8 +50,6 @@ namespace Squidex.Infrastructure.EventSourcing
                 regex = new Regex(streamFilter);
             }
 
-            this.store = store;
-
             this.subscriber = subscriber;
 
             processorTask = Task.Run(async () =>
@@ -73,7 +73,7 @@ namespace Squidex.Infrastructure.EventSourcing
                             .WithFeedCollection(CreateCollection(Constants.Collection))
                             .WithLeaseCollection(CreateCollection(Constants.LeaseCollection))
                             .WithHostName(hostName)
-                            .WithProcessorOptions(new Options { StartFromBeginning = fromBeginning })
+                            .WithProcessorOptions(new Options { StartFromBeginning = fromBeginning, LeasePrefix = hostName })
                             .WithObserverFactory(this)
                             .BuildAsync();
 

src/Squidex.Infrastructure.GetEventStore/EventSourcing/GetEventStore.cs

@@ -53,18 +53,26 @@ namespace Squidex.Infrastructure.EventSourcing
             await projectionClient.ConnectAsync(ct);
         }
 
-        public IEventSubscription CreateSubscription(IEventSubscriber subscriber, string streamFilter, string position = null)
+        public IEventSubscription CreateSubscription(IEventSubscriber subscriber, string streamFilter = null, string position = null)
         {
+            Guard.NotNull(streamFilter, nameof(streamFilter));
+
             return new GetEventStoreSubscription(connection, subscriber, serializer, projectionClient, position, prefix, streamFilter);
         }
 
         public Task CreateIndexAsync(string property)
         {
+            Guard.NotNullOrEmpty(property, nameof(property));
+
             return projectionClient.CreateProjectionAsync(property, string.Empty);
         }
 
         public async Task QueryAsync(Func<StoredEvent, Task> callback, string property, object value, string position = null, CancellationToken ct = default)
         {
+            Guard.NotNull(callback, nameof(callback));
+            Guard.NotNullOrEmpty(property, nameof(property));
+            Guard.NotNull(value, nameof(value));
+
             using (Profiler.TraceMethod<GetEventStore>())
             {
                 var streamName = await projectionClient.CreateProjectionAsync(property, value);
@@ -77,6 +85,8 @@ namespace Squidex.Infrastructure.EventSourcing
 
         public async Task QueryAsync(Func<StoredEvent, Task> callback, string streamFilter = null, string position = null, CancellationToken ct = default)
         {
+            Guard.NotNull(callback, nameof(callback));
+
             using (Profiler.TraceMethod<GetEventStore>())
             {
                 var streamName = await projectionClient.CreateProjectionAsync(streamFilter);
@@ -111,6 +121,8 @@ namespace Squidex.Infrastructure.EventSourcing
 
         public async Task<IReadOnlyList<StoredEvent>> QueryAsync(string streamName, long streamPosition = 0)
         {
+            Guard.NotNullOrEmpty(streamName, nameof(streamName));
+
             using (Profiler.TraceMethod<GetEventStore>())
             {
                 var result = new List<StoredEvent>();
@@ -142,6 +154,8 @@ namespace Squidex.Infrastructure.EventSourcing
 
         public Task DeleteStreamAsync(string streamName)
         {
+            Guard.NotNullOrEmpty(streamName, nameof(streamName));
+
             return connection.DeleteStreamAsync(GetStreamName(streamName), ExpectedVersion.Any);
         }
 
@@ -159,11 +173,11 @@ namespace Squidex.Infrastructure.EventSourcing
 
         private async Task AppendEventsInternalAsync(string streamName, long expectedVersion, ICollection<EventData> events)
         {
+            Guard.NotNullOrEmpty(streamName, nameof(streamName));
+            Guard.NotNull(events, nameof(events));
+
             using (Profiler.TraceMethod<GetEventStore>(nameof(AppendAsync)))
             {
-                Guard.NotNullOrEmpty(streamName, nameof(streamName));
-                Guard.NotNull(events, nameof(events));
-
                 if (events.Count == 0)
                 {
                     return;

src/Squidex.Infrastructure.GetEventStore/EventSourcing/GetEventStoreSubscription.cs

@@ -31,8 +31,6 @@ namespace Squidex.Infrastructure.EventSourcing
             string prefix,
             string streamFilter)
         {
-            Guard.NotNull(subscriber, nameof(subscriber));
-
             this.connection = connection;
 
             this.position = projectionClient.ParsePositionOrNull(position);

src/Squidex.Infrastructure.MongoDb/EventSourcing/MongoEventStore_Reader.cs

@@ -23,20 +23,23 @@ namespace Squidex.Infrastructure.EventSourcing
     {
         public Task CreateIndexAsync(string property)
         {
+            Guard.NotNullOrEmpty(property, nameof(property));
+
             return Collection.Indexes.CreateOneAsync(
                 new CreateIndexModel<MongoEventCommit>(Index.Ascending(CreateIndexPath(property))));
         }
 
-        public IEventSubscription CreateSubscription(IEventSubscriber subscriber, string streamFilter, string position = null)
+        public IEventSubscription CreateSubscription(IEventSubscriber subscriber, string streamFilter = null, string position = null)
         {
             Guard.NotNull(subscriber, nameof(subscriber));
-            Guard.NotNullOrEmpty(streamFilter, nameof(streamFilter));
 
             return new PollingSubscription(this, subscriber, streamFilter, position);
         }
 
         public async Task<IReadOnlyList<StoredEvent>> QueryAsync(string streamName, long streamPosition = 0)
         {
+            Guard.NotNullOrEmpty(streamName, nameof(streamName));
+
             using (Profiler.TraceMethod<MongoEventStore>())
             {
                 var commits =
@@ -76,6 +79,8 @@ namespace Squidex.Infrastructure.EventSourcing
         public Task QueryAsync(Func<StoredEvent, Task> callback, string property, object value, string position = null, CancellationToken ct = default)
         {
             Guard.NotNull(callback, nameof(callback));
+            Guard.NotNullOrEmpty(property, nameof(property));
+            Guard.NotNull(value, nameof(value));
 
             StreamPosition lastPosition = position;
 

src/Squidex.Infrastructure.MongoDb/EventSourcing/MongoEventStore_Writer.cs

@@ -22,6 +22,8 @@ namespace Squidex.Infrastructure.EventSourcing
 
         public Task DeleteStreamAsync(string streamName)
         {
+            Guard.NotNullOrEmpty(streamName, nameof(streamName));
+
             return Collection.DeleteManyAsync(x => x.EventStream == streamName);
         }
 
@@ -32,7 +34,7 @@ namespace Squidex.Infrastructure.EventSourcing
 
         public async Task AppendAsync(Guid commitId, string streamName, long expectedVersion, ICollection<EventData> events)
         {
-            Guard.GreaterEquals(expectedVersion, EtagVersion.Any, nameof(expectedVersion));
+            Guard.NotEmpty(commitId, nameof(commitId));
             Guard.NotNullOrEmpty(streamName, nameof(streamName));
             Guard.NotNull(events, nameof(events));
             Guard.LessThan(events.Count, MaxCommitSize, "events.Count");

src/Squidex.Infrastructure.RabbitMq/CQRS/Events/RabbitMqEventConsumer.cs

@@ -49,8 +49,8 @@ namespace Squidex.Infrastructure.CQRS.Events
 
             this.exchange = exchange;
             this.eventsFilter = eventsFilter;
-            this.jsonSerializer = jsonSerializer;
             this.eventPublisherName = eventPublisherName;
+            this.jsonSerializer = jsonSerializer;
         }
 
         protected override void DisposeObject(bool disposing)

src/Squidex.Infrastructure/EventSourcing/IEventStore.cs

@@ -28,6 +28,6 @@ namespace Squidex.Infrastructure.EventSourcing
 
         Task DeleteStreamAsync(string streamName);
 
-        IEventSubscription CreateSubscription(IEventSubscriber subscriber, string streamFilter, string position = null);
+        IEventSubscription CreateSubscription(IEventSubscriber subscriber, string streamFilter = null, string position = null);
     }
 }

src/Squidex/Config/Domain/EventStoreServices.cs

@@ -5,11 +5,14 @@
 //  All rights reserved. Licensed under the MIT license.
 // ==========================================================================
 
+using System;
 using System.Linq;
 using EventStore.ClientAPI;
+using Microsoft.Azure.Documents.Client;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using MongoDB.Driver;
+using Newtonsoft.Json;
 using Squidex.Infrastructure;
 using Squidex.Infrastructure.Diagnostics;
 using Squidex.Infrastructure.EventSourcing;
@@ -39,6 +42,25 @@ namespace Squidex.Config.Domain
                         })
                         .AsOptional<IEventStore>();
                 },
+                ["CosmosDb"] = () =>
+                {
+                    var cosmosDbConfiguration = config.GetRequiredValue("eventStore:cosmosDB:configuration");
+                    var cosmosDbMasterKey = config.GetRequiredValue("eventStore:cosmosDB:masterKey");
+                    var cosmosDbDatabase = config.GetRequiredValue("eventStore:cosmosDB:database");
+
+                    services.AddSingletonAs(c => new DocumentClient(new Uri(cosmosDbConfiguration), cosmosDbMasterKey, c.GetRequiredService<JsonSerializerSettings>()))
+                        .AsSelf();
+
+                    services.AddSingletonAs(c => new CosmosDbEventStore(
+                            c.GetRequiredService<DocumentClient>(),
+                            cosmosDbMasterKey,
+                            cosmosDbDatabase,
+                            c.GetRequiredService<JsonSerializerSettings>()))
+                        .AsOptional<IEventStore>();
+
+                    services.AddHealthChecks()
+                        .AddCheck<CosmosDbHealthCheck>("CosmosDB", tags: new[] { "node" });
+                },
                 ["GetEventStore"] = () =>
                 {
                     var eventStoreConfiguration = config.GetRequiredValue("eventStore:getEventStore:configuration");

src/Squidex/appsettings.json

@@ -29,8 +29,8 @@
     /*
      * Set to true, to use strong etags.
      */
-  "strong": false
-},
+    "strong": false
+  },
 
   "ui": {
     /*
@@ -199,7 +199,7 @@
     /*
      * Define the type of the event store. 
      * 
-     * Supported: MongoDb, GetEventStore
+     * Supported: MongoDb, GetEventStore, CosmosDb
      */
     "type": "MongoDb",
     "mongoDb": {
@@ -230,6 +230,20 @@
        */
       "prefix": "squidex"
     }
+    "cosmosDb": {
+      /*
+       * The connection string to your CosmosDB instance.
+       */
+      "configuration": "https://localhost:8081",
+      /*
+       * The primary access key.
+       */
+      "masterKey": "C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==",
+      /*
+       * The name of the event store database.
+       */
+      "database": "Squidex"
+    }
   },
 
   "eventPublishers": {

tests/Squidex.Infrastructure.Tests/EventSourcing/CosmosDbEventStoreFixture.cs

@@ -21,9 +21,9 @@ namespace Squidex.Infrastructure.EventSourcing
 
         public CosmosDbEventStoreFixture()
         {
-            client = new DocumentClient(new Uri(EmulatorUri), EmulatorKey);
+            client = new DocumentClient(new Uri(EmulatorUri), EmulatorKey, JsonHelper.DefaultSettings());
 
-            EventStore = new CosmosDbEventStore(new Uri(EmulatorUri), EmulatorKey, JsonHelper.DefaultSettings(), "Test");
+            EventStore = new CosmosDbEventStore(client, EmulatorKey, "Test", JsonHelper.DefaultSettings());
             EventStore.InitializeAsync().Wait();
         }
 

tests/Squidex.Infrastructure.Tests/EventSourcing/EventStoreTests.cs

@@ -19,11 +19,14 @@ namespace Squidex.Infrastructure.EventSourcing
     public abstract class EventStoreTests<T> where T : IEventStore
     {
         private readonly Lazy<T> sut;
+        private string subscriptionPosition;
 
         public sealed class EventSubscriber : IEventSubscriber
         {
             public List<StoredEvent> Events { get; } = new List<StoredEvent>();
 
+            public string LastPosition { get; set; }
+
             public Task OnErrorAsync(IEventSubscription subscription, Exception exception)
             {
                 throw new NotSupportedException();
@@ -31,6 +34,8 @@ namespace Squidex.Infrastructure.EventSourcing
 
             public Task OnEventAsync(IEventSubscription subscription, StoredEvent storedEvent)
             {
+                LastPosition = storedEvent.EventPosition;
+
                 Events.Add(storedEvent);
 
                 return TaskHelper.Done;
@@ -132,6 +137,54 @@ namespace Squidex.Infrastructure.EventSourcing
             ShouldBeEquivalentTo(readEvents, expected);
         }
 
+        [Fact]
+        public async Task Should_subscribe_to_next_events()
+        {
+            var streamName = $"test-{Guid.NewGuid()}";
+
+            var events1 = new EventData[]
+            {
+                new EventData("Type1", new EnvelopeHeaders(), "1"),
+                new EventData("Type2", new EnvelopeHeaders(), "2"),
+            };
+
+            await QueryWithSubscriptionAsync(streamName, async () =>
+            {
+                await Sut.AppendAsync(Guid.NewGuid(), streamName, events1);
+            });
+
+            var events2 = new EventData[]
+            {
+                new EventData("Type1", new EnvelopeHeaders(), "1"),
+                new EventData("Type2", new EnvelopeHeaders(), "2"),
+            };
+
+            var readEventsFromPosition = await QueryWithSubscriptionAsync(streamName, async () =>
+            {
+                await Sut.AppendAsync(Guid.NewGuid(), streamName, events2);
+            });
+
+            var expectedFromPosition = new StoredEvent[]
+            {
+                new StoredEvent(streamName, "Position", 2, events2[0]),
+                new StoredEvent(streamName, "Position", 3, events2[1])
+            };
+
+            var readEventsFromBeginning = await QueryWithSubscriptionAsync(streamName, fromBeginning: true);
+
+            var expectedFromBeginning = new StoredEvent[]
+            {
+                new StoredEvent(streamName, "Position", 0, events1[0]),
+                new StoredEvent(streamName, "Position", 1, events1[1]),
+                new StoredEvent(streamName, "Position", 2, events2[0]),
+                new StoredEvent(streamName, "Position", 3, events2[1])
+            };
+
+            ShouldBeEquivalentTo(readEventsFromPosition, expectedFromPosition);
+
+            ShouldBeEquivalentTo(readEventsFromBeginning, expectedFromBeginning);
+        }
+
         [Fact]
         public async Task Should_read_events_from_offset()
         {
@@ -272,16 +325,19 @@ namespace Squidex.Infrastructure.EventSourcing
             }
         }
 
-        private async Task<IReadOnlyList<StoredEvent>> QueryWithSubscriptionAsync(string streamFilter, Func<Task> action)
+        private async Task<IReadOnlyList<StoredEvent>> QueryWithSubscriptionAsync(string streamFilter, Func<Task> action = null, bool fromBeginning = false)
         {
             var subscriber = new EventSubscriber();
 
             IEventSubscription subscription = null;
             try
             {
-                subscription = Sut.CreateSubscription(subscriber, streamFilter);
+                subscription = Sut.CreateSubscription(subscriber, streamFilter, fromBeginning ? null : subscriptionPosition);
 
-                await action();
+                if (action != null)
+                {
+                    await action();
+                }
 
                 using (var cts = new CancellationTokenSource(30000))
                 {
@@ -293,6 +349,8 @@ namespace Squidex.Infrastructure.EventSourcing
 
                         if (subscriber.Events.Count > 0)
                         {
+                            subscriptionPosition = subscriber.LastPosition;
+
                             return subscriber.Events;
                         }
                     }