Add domain & integration events (#11)

Co-authored-by: Artem Leshchev <artem.leshchev@dxc.com>

src/Bss.Platform.Events.Abstractions/Bss.Platform.Events.Abstractions.csproj

@@ -0,0 +1,8 @@
+<Project Sdk="Microsoft.NET.Sdk">
+    <PropertyGroup>
+        <PackageId>Luxoft.Bss.Platform.Events.Abstractions</PackageId>
+    </PropertyGroup>
+    <ItemGroup>
+        <PackageReference Include="MediatR"/>
+    </ItemGroup>
+</Project>

src/Bss.Platform.Events.Abstractions/IDomainEvent.cs

@@ -0,0 +1,5 @@
+using MediatR;
+
+namespace Bss.Platform.Events.Abstractions;
+
+public interface IDomainEvent : INotification;

src/Bss.Platform.Events.Abstractions/IDomainEventPublisher.cs

@@ -0,0 +1,6 @@
+namespace Bss.Platform.Events.Abstractions;
+
+public interface IDomainEventPublisher
+{
+    Task PublishAsync(IDomainEvent @event, CancellationToken cancellationToken);
+}

src/Bss.Platform.Events.Abstractions/IIntegrationEvent.cs

@@ -0,0 +1,5 @@
+using MediatR;
+
+namespace Bss.Platform.Events.Abstractions;
+
+public interface IIntegrationEvent : INotification;

src/Bss.Platform.Events.Abstractions/IIntegrationEventPublisher.cs

@@ -0,0 +1,6 @@
+namespace Bss.Platform.Events.Abstractions;
+
+public interface IIntegrationEventPublisher
+{
+    Task PublishAsync(IIntegrationEvent @event, CancellationToken cancellationToken);
+}

src/Bss.Platform.Events/Bss.Platform.Events.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+    <PropertyGroup>
+        <PackageId>Luxoft.Bss.Platform.Events</PackageId>
+    </PropertyGroup>
+    <ItemGroup>
+      <ProjectReference Include="..\Bss.Platform.Events.Abstractions\Bss.Platform.Events.Abstractions.csproj" />
+    </ItemGroup>
+    <ItemGroup>
+      <PackageReference Include="DotNetCore.CAP.Dashboard" />
+      <PackageReference Include="DotNetCore.CAP.RabbitMQ" />
+      <PackageReference Include="DotNetCore.CAP.SqlServer" />
+      <PackageReference Include="Savorboard.CAP.InMemoryMessageQueue" />
+    </ItemGroup>
+</Project>

src/Bss.Platform.Events/CapConsumerExecutor.cs

@@ -0,0 +1,10 @@
+using Bss.Platform.Events.Abstractions;
+using Bss.Platform.Events.Interfaces;
+
+namespace Bss.Platform.Events;
+
+internal class CapConsumerExecutor<TEvent>(IIntegrationEventProcessor eventProcessor)
+    where TEvent : IIntegrationEvent
+{
+    public Task HandleAsync(TEvent @event, CancellationToken cancellationToken) => eventProcessor.ProcessAsync(@event, cancellationToken);
+}

src/Bss.Platform.Events/CapConsumerServiceSelector.cs

@@ -0,0 +1,53 @@
+using System.Reflection;
+
+using Bss.Platform.Events.Abstractions;
+
+using DotNetCore.CAP;
+using DotNetCore.CAP.Internal;
+
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+
+namespace Bss.Platform.Events;
+
+public class CapConsumerServiceSelector(IServiceProvider serviceProvider, Assembly assembly)
+    : ConsumerServiceSelector(serviceProvider)
+{
+    protected override IEnumerable<ConsumerExecutorDescriptor> FindConsumersFromControllerTypes() => [];
+
+    protected override IEnumerable<ConsumerExecutorDescriptor> FindConsumersFromInterfaceTypes(IServiceProvider provider)
+    {
+        var namePrefix = provider.GetRequiredService<IOptions<CapOptions>>().Value.TopicNamePrefix;
+
+        return assembly
+               .ExportedTypes
+               .Where(x => typeof(IIntegrationEvent).IsAssignableFrom(x) && x is { IsInterface: false, IsAbstract: false })
+               .Select(x => this.CreateExecutorDescriptor(typeof(CapConsumerExecutor<>).MakeGenericType(x), x, namePrefix));
+    }
+
+    private ConsumerExecutorDescriptor CreateExecutorDescriptor(Type executor, Type @event, string? namePrefix)
+    {
+        var subscribeAttribute = new CapSubscribeAttribute(@event.Name);
+        this.SetSubscribeAttribute(subscribeAttribute);
+
+        var methodInfo = executor
+                         .GetRuntimeMethods()
+                         .Single(x => x.Name.Contains(nameof(CapConsumerExecutor<IIntegrationEvent>.HandleAsync)));
+
+        var methodParameters = methodInfo.GetParameters();
+        return new ConsumerExecutorDescriptor
+               {
+                   Attribute = subscribeAttribute,
+                   ClassAttribute = null,
+                   MethodInfo = methodInfo,
+                   ImplTypeInfo = executor.GetTypeInfo(),
+                   ServiceTypeInfo = null,
+                   TopicNamePrefix = namePrefix,
+                   Parameters = new List<ParameterDescriptor>
+                                {
+                                    new() { ParameterType = methodParameters[0].ParameterType, IsFromCap = false },
+                                    new() { ParameterType = methodParameters[1].ParameterType, IsFromCap = true }
+                                }
+               };
+    }
+}

src/Bss.Platform.Events/DependencyInjection.cs

@@ -0,0 +1,80 @@
+using System.Data;
+using System.Reflection;
+
+using Bss.Platform.Events.Abstractions;
+using Bss.Platform.Events.Interfaces;
+using Bss.Platform.Events.Models;
+using Bss.Platform.Events.Publishers;
+
+using DotNetCore.CAP;
+using DotNetCore.CAP.Internal;
+
+using Microsoft.Extensions.DependencyInjection;
+
+using Savorboard.CAP.InMemoryMessageQueue;
+
+namespace Bss.Platform.Events;
+
+public static class DependencyInjection
+{
+    public static IServiceCollection AddPlatformDomainEvents(this IServiceCollection services) =>
+        services.AddScoped<IDomainEventPublisher, DomainEventPublisher>();
+
+    public static IServiceCollection AddPlatformIntegrationEvents<TEventProcessor>(
+        this IServiceCollection services,
+        Assembly eventsAssembly,
+        Action<IntegrationEventsOptions>? setup = null)
+        where TEventProcessor : class, IIntegrationEventProcessor
+    {
+        services
+            .AddSingleton<IIntegrationEventProcessor, TEventProcessor>()
+            .AddSingleton<IConsumerServiceSelector, CapConsumerServiceSelector>(x => new CapConsumerServiceSelector(x, eventsAssembly))
+            .AddScoped<IIntegrationEventPublisher, IntegrationEventPublisher>()
+            .AddScoped<ICapTransaction>(
+                serviceProvider =>
+                {
+                    var capTransaction = ActivatorUtilities.CreateInstance<SqlServerCapTransaction>(serviceProvider);
+                    capTransaction.DbTransaction = serviceProvider.GetRequiredService<IDbTransaction>();
+                    return capTransaction;
+                })
+            .AddCap(
+                x =>
+                {
+                    var eventsOptions = IntegrationEventsOptions.Default;
+                    setup?.Invoke(eventsOptions);
+
+                    x.FailedRetryCount = eventsOptions.FailedRetryCount;
+                    x.SucceedMessageExpiredAfter = (int)TimeSpan.FromDays(eventsOptions.RetentionDays).TotalSeconds;
+
+                    x.UseSqlServer(
+                        o =>
+                        {
+                            o.ConnectionString = eventsOptions.SqlServer.ConnectionString;
+                            o.Schema = eventsOptions.SqlServer.Schema;
+                        });
+
+                    x.UseDashboard(o => o.PathMatch = eventsOptions.DashboardPath);
+
+                    if (!eventsOptions.MessageQueue.Enable)
+                    {
+                        x.UseInMemoryMessageQueue();
+                        return;
+                    }
+
+                    x.DefaultGroupName = eventsOptions.MessageQueue.ExchangeName;
+                    x.UseRabbitMQ(
+                        o =>
+                        {
+                            o.HostName = eventsOptions.MessageQueue.Host;
+                            o.Port = eventsOptions.MessageQueue.Port;
+                            o.VirtualHost = eventsOptions.MessageQueue.VirtualHost;
+                            o.Password = eventsOptions.MessageQueue.Secret;
+                            o.UserName = eventsOptions.MessageQueue.UserName;
+                            o.ExchangeName = eventsOptions.MessageQueue.ExchangeName;
+                            o.BasicQosOptions = new RabbitMQOptions.BasicQos(1, true);
+                        });
+                });
+
+        return services;
+    }
+}

src/Bss.Platform.Events/Interfaces/IIntegrationEventProcessor.cs

@@ -0,0 +1,8 @@
+using Bss.Platform.Events.Abstractions;
+
+namespace Bss.Platform.Events.Interfaces;
+
+public interface IIntegrationEventProcessor
+{
+    Task ProcessAsync(IIntegrationEvent @event, CancellationToken token);
+}

src/Bss.Platform.Events/Models/IntegrationEventsMessageQueueOptions.cs

@@ -0,0 +1,18 @@
+namespace Bss.Platform.Events.Models;
+
+public class IntegrationEventsMessageQueueOptions
+{
+    public bool Enable { get; set; }
+
+    public string Host { get; set; } = default!;
+
+    public int Port { get; set; }
+
+    public string UserName { get; set; } = default!;
+
+    public string Secret { get; set; } = default!;
+
+    public string VirtualHost { get; set; } = default!;
+
+    public string ExchangeName { get; set; } = default!;
+}

src/Bss.Platform.Events/Models/IntegrationEventsOptions.cs

@@ -0,0 +1,24 @@
+namespace Bss.Platform.Events.Models;
+
+public class IntegrationEventsOptions
+{
+    public string DashboardPath { get; set; } = default!;
+
+    public int FailedRetryCount { get; set; }
+
+    public int RetentionDays { get; set; }
+
+    public IntegrationEventsSqlServerOptions SqlServer { get; set; } = default!;
+
+    public IntegrationEventsMessageQueueOptions MessageQueue { get; set; } = default!;
+
+    public static IntegrationEventsOptions Default =>
+        new()
+        {
+            DashboardPath = "/admin/events",
+            FailedRetryCount = 5,
+            RetentionDays = 15,
+            SqlServer = new IntegrationEventsSqlServerOptions { Schema = "events" },
+            MessageQueue = new IntegrationEventsMessageQueueOptions { Enable = true }
+        };
+}

src/Bss.Platform.Events/Models/IntegrationEventsSqlServerOptions.cs

@@ -0,0 +1,8 @@
+namespace Bss.Platform.Events.Models;
+
+public class IntegrationEventsSqlServerOptions
+{
+    public string ConnectionString { get; set; } = default!;
+
+    public string Schema { get; set; } = default!;
+}

src/Bss.Platform.Events/Publishers/DomainEventPublisher.cs

@@ -0,0 +1,10 @@
+using Bss.Platform.Events.Abstractions;
+
+using MediatR;
+
+namespace Bss.Platform.Events.Publishers;
+
+public class DomainEventPublisher(IMediator mediator) : IDomainEventPublisher
+{
+    public Task PublishAsync(IDomainEvent @event, CancellationToken cancellationToken) => mediator.Publish(@event, cancellationToken);
+}

src/Bss.Platform.Events/Publishers/IntegrationEventPublisher.cs

@@ -0,0 +1,19 @@
+using Bss.Platform.Events.Abstractions;
+
+using DotNetCore.CAP;
+
+namespace Bss.Platform.Events.Publishers;
+
+public class IntegrationEventPublisher(ICapPublisher capPublisher, ICapTransaction capTransaction) : IIntegrationEventPublisher
+{
+    public Task PublishAsync(IIntegrationEvent @event, CancellationToken cancellationToken)
+    {
+        if (capPublisher.Transaction is not null && capPublisher.Transaction != capTransaction)
+        {
+            throw new Exception("There cannot be different CAP transactions within the same scope");
+        }
+
+        capPublisher.Transaction = capTransaction;
+        return capPublisher.PublishAsync(@event.GetType().Name, @event, cancellationToken: cancellationToken);
+    }
+}

src/Bss.Platform.sln

@@ -28,6 +28,12 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bss.Platform.Kubernetes", "
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bss.Platform.Api.Middlewares", "Bss.Platform.Api.Middlewares\Bss.Platform.Api.Middlewares.csproj", "{285EBB7B-6B2A-453E-98F6-30AC38317A76}"
 EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Events", "Events", "{80AA8C01-842D-4A5F-BCBE-48E0361783A0}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bss.Platform.Events", "Bss.Platform.Events\Bss.Platform.Events.csproj", "{51668BE6-C4A4-4E62-9BEC-529C33A434DB}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bss.Platform.Events.Abstractions", "Bss.Platform.Events.Abstractions\Bss.Platform.Events.Abstractions.csproj", "{BE2E219F-3F3B-48E8-B13B-A71321BB6372}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -42,6 +48,8 @@ Global
 		{0005DAE1-3F66-4A98-A11B-DEFDF2BD7EA3} = {8F22FE4E-8FD4-4309-8B47-C33AA6E52052}
 		{1BD00077-8005-42E9-AEFE-62A21E057858} = {51F57F48-6CDC-43E2-90DA-B73C12EA2185}
 		{285EBB7B-6B2A-453E-98F6-30AC38317A76} = {8F22FE4E-8FD4-4309-8B47-C33AA6E52052}
+		{51668BE6-C4A4-4E62-9BEC-529C33A434DB} = {80AA8C01-842D-4A5F-BCBE-48E0361783A0}
+		{BE2E219F-3F3B-48E8-B13B-A71321BB6372} = {80AA8C01-842D-4A5F-BCBE-48E0361783A0}
 	EndGlobalSection
 	GlobalSection(ProjectConfigurationPlatforms) = postSolution
 		{860BFBD8-26EA-44F9-980E-21B828FC8F72}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
@@ -76,5 +84,13 @@ Global
 		{285EBB7B-6B2A-453E-98F6-30AC38317A76}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{285EBB7B-6B2A-453E-98F6-30AC38317A76}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{285EBB7B-6B2A-453E-98F6-30AC38317A76}.Release|Any CPU.Build.0 = Release|Any CPU
+		{51668BE6-C4A4-4E62-9BEC-529C33A434DB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{51668BE6-C4A4-4E62-9BEC-529C33A434DB}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{51668BE6-C4A4-4E62-9BEC-529C33A434DB}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{51668BE6-C4A4-4E62-9BEC-529C33A434DB}.Release|Any CPU.Build.0 = Release|Any CPU
+		{BE2E219F-3F3B-48E8-B13B-A71321BB6372}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{BE2E219F-3F3B-48E8-B13B-A71321BB6372}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{BE2E219F-3F3B-48E8-B13B-A71321BB6372}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{BE2E219F-3F3B-48E8-B13B-A71321BB6372}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 EndGlobal

src/Directory.Packages.props

@@ -4,6 +4,11 @@
   </PropertyGroup>
   <ItemGroup>
     <PackageVersion Include="AspNetCore.HealthChecks.SqlServer" Version="8.0.2" />
+    <PackageVersion Include="DotNetCore.CAP.Dashboard" Version="8.1.2" />
+    <PackageVersion Include="DotNetCore.CAP.RabbitMQ" Version="8.1.2" />
+    <PackageVersion Include="DotNetCore.CAP.SqlServer" Version="8.1.2" />
+    <PackageVersion Include="Savorboard.CAP.InMemoryMessageQueue" Version="8.0.0" />
+    <PackageVersion Include="MediatR" Version="12.2.0" />
     <PackageVersion Include="Microsoft.ApplicationInsights" Version="2.22.0" />
     <PackageVersion Include="Microsoft.ApplicationInsights.AspNetCore" Version="2.22.0" />
     <PackageVersion Include="Microsoft.ApplicationInsights.Kubernetes" Version="6.1.2" />

src/__SolutionItems/CommonAssemblyInfo.cs

@@ -4,9 +4,9 @@
 [assembly: AssemblyCompany("Luxoft")]
 [assembly: AssemblyCopyright("Copyright © Luxoft 2024")]
 
-[assembly: AssemblyVersion("1.3.1.0")]
-[assembly: AssemblyFileVersion("1.3.1.0")]
-[assembly: AssemblyInformationalVersion("1.3.1.0")]
+[assembly: AssemblyVersion("1.4.0.0")]
+[assembly: AssemblyFileVersion("1.4.0.0")]
+[assembly: AssemblyInformationalVersion("1.4.0.0")]
 
 #if DEBUG
 [assembly: AssemblyConfiguration("Debug")]