feat(Solution): Implemented an internal cloud event bus, allowing solution components to correlate internally emitted events

Closes #419

Signed-off-by: Charles d'Avernas <[email protected]>

Author: cdavernas

src/api/Synapse.Api.Application/Services/CloudEventPublisher.cs

@@ -17,6 +17,7 @@
 using Neuroglia.Eventing.CloudEvents;
 using Neuroglia.Serialization;
 using Polly;
+using StackExchange.Redis;
 using Synapse.Api.Application.Configuration;
 using System.Text;
 using System.Threading.Channels;
@@ -27,10 +28,11 @@ namespace Synapse.Api.Application.Services;
 /// Represents the default implementation of the <see cref="ICloudEventPublisher"/> interface
 /// </summary>
 /// <param name="logger">The service used to perform logging</param>
+/// <param name="connectionMultiplexer">The service used to connect to the Redis API</param>
 /// <param name="options">The service used to access the current <see cref="ApiServerOptions"/></param>
 /// <param name="jsonSerializer">The service used to serialize/deserialize data to/from JSON</param>
 /// <param name="httpClient">The service used to perform HTTP requests</param>
-public class CloudEventPublisher(ILogger<CloudEventPublisher> logger, IJsonSerializer jsonSerializer, IOptions<ApiServerOptions> options, HttpClient httpClient)
+public class CloudEventPublisher(ILogger<CloudEventPublisher> logger, IConnectionMultiplexer connectionMultiplexer, IJsonSerializer jsonSerializer, IOptions<ApiServerOptions> options, HttpClient httpClient)
     : IHostedService, ICloudEventPublisher, IDisposable, IAsyncDisposable
 {
 
@@ -41,6 +43,16 @@ public class CloudEventPublisher(ILogger<CloudEventPublisher> logger, IJsonSeria
     /// </summary>
     protected ILogger Logger { get; } = logger;
 
+    /// <summary>
+    /// Gets the service used to connect to the Redis API
+    /// </summary>
+    protected IConnectionMultiplexer ConnectionMultiplexer { get; } = connectionMultiplexer;
+
+    /// <summary>
+    /// Gets the Redis database to use
+    /// </summary>
+    protected StackExchange.Redis.IDatabase Database { get; } = connectionMultiplexer.GetDatabase();
+
     /// <summary>
     /// Gets the service used to serialize/deserialize data to/from JSON
     /// </summary>
@@ -66,11 +78,28 @@ public class CloudEventPublisher(ILogger<CloudEventPublisher> logger, IJsonSeria
     /// </summary>
     protected Channel<CloudEvent> Channel { get; } = System.Threading.Channels.Channel.CreateUnbounded<CloudEvent>();
 
+    /// <summary>
+    /// Gets a boolean indicating whether or not the Redis version used by the cloud event publisher supports streaming commands
+    /// </summary>
+    protected bool SupportsStreaming { get; private set; }
+
     /// <inheritdoc/>
     public virtual Task StartAsync(CancellationToken cancellationToken)
     {
         if (options.Value.CloudEvents.Endpoint == null) logger.LogWarning("No endpoint configured for cloud events. Events will not be published.");
         else _ = this.PublishEnqueuedEventsAsync();
+        var version = ((string)(this.Database.Execute("INFO", "server"))!).Split('\n').FirstOrDefault(line => line.StartsWith("redis_version:"))?[14..]?.Trim() ?? "undetermined";
+        try
+        {
+            this.Database.StreamInfo(SynapseDefaults.CloudEvents.Bus.StreamName);
+            this.SupportsStreaming = true;
+            this.Logger.LogInformation("Redis server version '{version}' supports streaming commands. Streaming feature is enabled", version);
+        }
+        catch (RedisServerException ex) when (ex.Message.StartsWith("ERR unknown command"))
+        {
+            this.SupportsStreaming = false;
+            this.Logger.LogInformation("Redis server version '{version}' does not support streaming commands. Streaming feature is emulated using lists", version);
+        }
         return Task.CompletedTask;
     }
 
@@ -79,6 +108,13 @@ public virtual async Task PublishAsync(CloudEvent e, CancellationToken cancellat
     {
         ArgumentNullException.ThrowIfNull(e);
         await this.Channel.Writer.WriteAsync(e, cancellationToken).ConfigureAwait(false);
+        var json = this.JsonSerializer.SerializeToText(e);
+        if (this.SupportsStreaming) await this.Database.StreamAddAsync(SynapseDefaults.CloudEvents.Bus.StreamName, SynapseDefaults.CloudEvents.Bus.EventFieldName, json).ConfigureAwait(false);
+        else
+        {
+            await this.Database.ListLeftPushAsync(SynapseDefaults.CloudEvents.Bus.StreamName, json).ConfigureAwait(false);
+            await this.Database.ScriptEvaluateAsync(SynapseDefaults.CloudEvents.Bus.MessageDistributionScript, [SynapseDefaults.CloudEvents.Bus.StreamName, SynapseDefaults.CloudEvents.Bus.ConsumerGroupListKey]).ConfigureAwait(false);
+        }
     }
 
     /// <inheritdoc/>

src/core/Synapse.Core.Infrastructure/Services/DatabaseInitializer.cs

@@ -12,8 +12,10 @@
 // limitations under the License.
 
 using Microsoft.Extensions.Logging;
+using Neuroglia;
 using Neuroglia.Data.Infrastructure.ResourceOriented;
 using Synapse.Resources;
+using System.Net;
 using System.Security.Cryptography;
 
 namespace Synapse.Core.Infrastructure.Services;
@@ -29,28 +31,33 @@ public class DatabaseInitializer(ILoggerFactory loggerFactory, IServiceProvider
     /// <inheritdoc/>
     protected override async Task SeedAsync(CancellationToken cancellationToken)
     {
-        foreach (var definition in SynapseDefaults.Resources.Definitions.AsEnumerable()) await this.Database.CreateResourceAsync(definition, cancellationToken: cancellationToken).ConfigureAwait(false);
+        foreach (var definition in SynapseDefaults.Resources.Definitions.AsEnumerable())
+        {
+            try { await this.Database.CreateResourceAsync(definition, cancellationToken: cancellationToken).ConfigureAwait(false); }
+            catch (ProblemDetailsException ex) when (ex.Problem.Status == (int)HttpStatusCode.Conflict || (ex.Problem.Status == (int)HttpStatusCode.BadRequest && ex.Problem.Title == "Conflict")) { continue; }
+        }
         var keyBytes = new byte[64];
         using var rng = RandomNumberGenerator.Create();
         rng.GetBytes(keyBytes);
         var key = Convert.ToBase64String(keyBytes);
-        await this.Database.CreateResourceAsync(new ServiceAccount()
+        try
         {
-            Metadata = new()
-            {
-                Namespace = Namespace.DefaultNamespaceName,
-                Name = ServiceAccount.DefaultServiceAccountName
-            },
-            Spec = new()
+            await this.Database.CreateResourceAsync(new ServiceAccount()
             {
-                Key = key,
-                Claims = new Dictionary<string, string>()
+                Metadata = new()
                 {
-
+                    Namespace = Namespace.DefaultNamespaceName,
+                    Name = ServiceAccount.DefaultServiceAccountName
+                },
+                Spec = new()
+                {
+                    Key = key,
+                    Claims = new Dictionary<string, string>() { }
                 }
-            }
-        }, false, cancellationToken).ConfigureAwait(false);
-        await this.Database.InitializeAsync(cancellationToken);
+            }, false, cancellationToken).ConfigureAwait(false);
+            await this.Database.InitializeAsync(cancellationToken);
+        }
+        catch (ProblemDetailsException ex) when (ex.Problem.Status == (int)HttpStatusCode.Conflict || (ex.Problem.Status == (int)HttpStatusCode.BadRequest && ex.Problem.Title == "Conflict")) { continue; }
     }
 
 }

src/core/Synapse.Core/SynapseDefaults.cs

@@ -53,6 +53,40 @@ public static class CloudEvents
         /// </summary>
         public const string TypePrefix = "io.synapse-wfms.events.";
 
+        /// <summary>
+        /// Exposes constants about the Synapse cloud event bus
+        /// </summary>
+        public static class Bus
+        {
+
+            /// <summary>
+            /// Gets the name of the stream used to observe cloud events published to the Synapse cloud event bus
+            /// </summary>
+            public const string StreamName = "cloud-events";
+            /// <summary>
+            /// Gets the name of the field used to store the serialized cloud event
+            /// </summary>
+            public const string EventFieldName = "event";
+            /// <summary>
+            /// Gets the key of the list used to store all existing consumer groups
+            /// </summary>
+            public const string ConsumerGroupListKey = "cloud-events-consumer-groups";
+            /// <summary>
+            /// Gets the LUA script used to distribute cloud event bus messages amongst consumer groups
+            /// </summary>
+            public static string MessageDistributionScript = @"
+local message = redis.call('RPOP', KEYS[1]);
+if message then 
+    local consumerGroups = redis.call('SMEMBERS', KEYS[2]); 
+    for _, group in ipairs(consumerGroups) do 
+        redis.call('LPUSH', group, message); 
+    end; 
+end; 
+return message;
+";
+
+        }
+
         /// <summary>
         /// Exposes constants about workflow-related cloud events
         /// </summary>
@@ -497,6 +531,24 @@ public static class Correlator
             /// </summary>
             public const string Name = Prefix + "NAME";
 
+            /// <summary>
+            /// Exposes constants about the cloud events related environment variables of a correlator
+            /// </summary>
+            public static class Events
+            {
+
+                /// <summary>
+                /// Gets the prefix for all correlator related environment variables
+                /// </summary>
+                public const string Prefix = Correlator.Prefix + "EVENTS_";
+
+                /// <summary>
+                /// Gets the name of the consumer group the correlator to configure belongs to, if any
+                /// </summary>
+                public const string ConsumerGroup = Prefix + "CONSUMER";
+
+            }
+
         }
 
         /// <summary>

src/correlator/Synapse.Correlator/Configuration/CloudEventConsumptionOptions.cs

@@ -0,0 +1,27 @@
+// Copyright © 2024-Present The Synapse Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License"),
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+namespace Synapse.Correlator.Configuration;
+
+/// <summary>
+/// Represents the options used to configure the way a Synapse Correlator consumes cloud events
+/// </summary>
+public class CloudEventConsumptionOptions
+{
+
+    /// <summary>
+    /// Gets/sets the name of the correlator's consumer group, if any. If not set, defaults to a concatenation of the correlators name and namespace.
+    /// </summary>
+    public virtual string? ConsumerGroup { get; set; }
+
+}

src/correlator/Synapse.Correlator/Configuration/CorrelatorOptions.cs

@@ -24,8 +24,14 @@ public class CorrelatorOptions
     /// </summary>
     public CorrelatorOptions()
     {
-        Namespace = Environment.GetEnvironmentVariable(SynapseDefaults.EnvironmentVariables.Correlator.Namespace)!;
-        Name = Environment.GetEnvironmentVariable(SynapseDefaults.EnvironmentVariables.Correlator.Name)!;
+        this.Namespace = Environment.GetEnvironmentVariable(SynapseDefaults.EnvironmentVariables.Correlator.Namespace)!;
+        this.Name = Environment.GetEnvironmentVariable(SynapseDefaults.EnvironmentVariables.Correlator.Name)!;
+        var env = Environment.GetEnvironmentVariable(SynapseDefaults.EnvironmentVariables.Correlator.Events.ConsumerGroup);
+        if (!string.IsNullOrWhiteSpace(env))
+        {
+            this.Events ??= new();
+            this.Events.ConsumerGroup = env;
+        }
     }
 
     /// <summary>
@@ -38,4 +44,9 @@ public CorrelatorOptions()
     /// </summary>
     public virtual string Name { get; set; }
 
+    /// <summary>
+    /// Gets/sets the options used to configure the way a Synapse Correlator consumes cloud events
+    /// </summary>
+    public virtual CloudEventConsumptionOptions? Events { get; set; }
+
 }

src/correlator/Synapse.Correlator/Program.cs

@@ -31,6 +31,9 @@
 builder.Services.AddCloudEventBus();
 builder.Services.AddTransient<IRequestHandler<IngestCloudEventCommand, IOperationResult>, IngestCloudEventCommandHandler>();
 
+builder.Services.AddSingleton<RedisCloudEventIngestor>();
+builder.Services.AddHostedService(provider => provider.GetRequiredService<RedisCloudEventIngestor>());
+
 builder.Services.AddScoped<CorrelatorController>();
 builder.Services.AddScoped<ICorrelatorController>(provider => provider.GetRequiredService<CorrelatorController>());