[PM-17562] Refactor existing RabbitMq implementation

src/Core/AdminConsole/Services/IEventMessageHandler.cs

@@ -0,0 +1,8 @@
+using Bit.Core.Models.Data;
+
+namespace Bit.Core.Services;
+
+public interface IEventMessageHandler
+{
+    Task HandleEventAsync(EventMessage eventMessage);
+}

src/Core/AdminConsole/Services/Implementations/EventRepositoryHandler.cs

@@ -0,0 +1,14 @@
+using Bit.Core.Models.Data;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Bit.Core.Services;
+
+public class EventRepositoryHandler(
+    [FromKeyedServices("persistent")] IEventWriteService eventWriteService)
+    : IEventMessageHandler
+{
+    public Task HandleEventAsync(EventMessage eventMessage)
+    {
+        return eventWriteService.CreateAsync(eventMessage);
+    }
+}

src/Core/AdminConsole/Services/Implementations/RabbitMqEventHttpPostListener.cs → src/Core/AdminConsole/Services/Implementations/HttpPostEventHandler.cs

@@ -1,32 +1,25 @@
 using System.Net.Http.Json;
 using Bit.Core.Models.Data;
 using Bit.Core.Settings;
-using Microsoft.Extensions.Logging;
 
 namespace Bit.Core.Services;
 
-public class RabbitMqEventHttpPostListener : RabbitMqEventListenerBase
+public class HttpPostEventHandler : IEventMessageHandler
 {
     private readonly HttpClient _httpClient;
     private readonly string _httpPostUrl;
-    private readonly string _queueName;
 
-    protected override string QueueName => _queueName;
+    public const string HttpClientName = "HttpPostEventHandlerHttpClient";
 
-    public const string HttpClientName = "EventHttpPostListenerHttpClient";
-
-    public RabbitMqEventHttpPostListener(
+    public HttpPostEventHandler(
         IHttpClientFactory httpClientFactory,
-        ILogger<RabbitMqEventListenerBase> logger,
         GlobalSettings globalSettings)
-        : base(logger, globalSettings)
     {
         _httpClient = httpClientFactory.CreateClient(HttpClientName);
         _httpPostUrl = globalSettings.EventLogging.RabbitMq.HttpPostUrl;
-        _queueName = globalSettings.EventLogging.RabbitMq.HttpPostQueueName;
     }
 
-    protected override async Task HandleMessageAsync(EventMessage eventMessage)
+    public async Task HandleEventAsync(EventMessage eventMessage)
     {
         var content = JsonContent.Create(eventMessage);
         var response = await _httpClient.PostAsync(_httpPostUrl, content);

src/Core/Services/EventLoggingListenerService.cs

@@ -0,0 +1,13 @@
+using Microsoft.Extensions.Hosting;
+
+namespace Bit.Core.Services;
+
+public abstract class EventLoggingListenerService : BackgroundService
+{
+    protected readonly IEventMessageHandler _handler;
+
+    protected EventLoggingListenerService(IEventMessageHandler handler)
+    {
+        _handler = handler ?? throw new ArgumentNullException(nameof(handler));
+    }
+}

src/Core/AdminConsole/Services/Implementations/RabbitMqEventListenerBase.cs → src/Core/Services/Implementations/RabbitMqEventListenerService.cs

@@ -1,26 +1,26 @@
 using System.Text.Json;
 using Bit.Core.Models.Data;
 using Bit.Core.Settings;
-using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
 
 namespace Bit.Core.Services;
 
-public abstract class RabbitMqEventListenerBase : BackgroundService
+public class RabbitMqEventListenerService : EventLoggingListenerService
 {
     private IChannel _channel;
     private IConnection _connection;
     private readonly string _exchangeName;
     private readonly ConnectionFactory _factory;
-    private readonly ILogger<RabbitMqEventListenerBase> _logger;
+    private readonly ILogger<RabbitMqEventListenerService> _logger;
+    private readonly string _queueName;
 
-    protected abstract string QueueName { get; }
-
-    protected RabbitMqEventListenerBase(
-        ILogger<RabbitMqEventListenerBase> logger,
-        GlobalSettings globalSettings)
+    public RabbitMqEventListenerService(
+        IEventMessageHandler handler,
+        ILogger<RabbitMqEventListenerService> logger,
+        GlobalSettings globalSettings,
+        string queueName) : base(handler)
     {
         _factory = new ConnectionFactory
         {
@@ -30,6 +30,7 @@
         };
         _exchangeName = globalSettings.EventLogging.RabbitMq.ExchangeName;
         _logger = logger;
+        _queueName = queueName;
     }
 
     public override async Task StartAsync(CancellationToken cancellationToken)
@@ -38,13 +39,13 @@
         _channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
 
         await _channel.ExchangeDeclareAsync(exchange: _exchangeName, type: ExchangeType.Fanout, durable: true);
-        await _channel.QueueDeclareAsync(queue: QueueName,
+        await _channel.QueueDeclareAsync(queue: _queueName,
                                          durable: true,
                                          exclusive: false,
                                          autoDelete: false,
                                          arguments: null,
                                          cancellationToken: cancellationToken);
-        await _channel.QueueBindAsync(queue: QueueName,
+        await _channel.QueueBindAsync(queue: _queueName,
                                       exchange: _exchangeName,
                                       routingKey: string.Empty,
                                       cancellationToken: cancellationToken);
@@ -59,15 +60,15 @@
             try
             {
                 var eventMessage = JsonSerializer.Deserialize<EventMessage>(eventArgs.Body.Span);
-                await HandleMessageAsync(eventMessage);
+                await _handler.HandleEventAsync(eventMessage);
             }
             catch (Exception ex)
             {
                 _logger.LogError(ex, "An error occurred while processing the message");
             }
         };
 
-        await _channel.BasicConsumeAsync(QueueName, autoAck: true, consumer: consumer, cancellationToken: stoppingToken);
+        await _channel.BasicConsumeAsync(_queueName, autoAck: true, consumer: consumer, cancellationToken: stoppingToken);
 
         while (!stoppingToken.IsCancellationRequested)
         {
@@ -88,6 +89,4 @@
         _connection.Dispose();
         base.Dispose();
     }
-
-    protected abstract Task HandleMessageAsync(EventMessage eventMessage);
 }

src/Core/Settings/IGlobalSettings.cs

@@ -27,4 +27,5 @@ public interface IGlobalSettings
     string DatabaseProvider { get; set; }
     GlobalSettings.SqlSettings SqlServer { get; set; }
     string DevelopmentDirectory { get; set; }
+    GlobalSettings.EventLoggingSettings EventLogging { get; set; }
 }

src/Events/Startup.cs

@@ -89,13 +89,26 @@
             CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.Password) &&
             CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.ExchangeName))
         {
+            services.AddSingleton<EventRepositoryHandler>();
             services.AddKeyedSingleton<IEventWriteService, RepositoryEventWriteService>("persistent");
-            services.AddHostedService<RabbitMqEventRepositoryListener>();
+            services.AddSingleton<IHostedService>(provider =>
+                new RabbitMqEventListenerService(
+                    provider.GetRequiredService<EventRepositoryHandler>(),
+                    provider.GetRequiredService<ILogger<RabbitMqEventListenerService>>(),
+                    provider.GetRequiredService<GlobalSettings>(),
+                    globalSettings.EventLogging.RabbitMq.EventRepositoryQueueName));
 
             if (CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.HttpPostUrl))
             {
-                services.AddHttpClient(RabbitMqEventHttpPostListener.HttpClientName);
-                services.AddHostedService<RabbitMqEventHttpPostListener>();
+                services.AddSingleton<HttpPostEventHandler>();
+                services.AddHttpClient(HttpPostEventHandler.HttpClientName);
+
+                services.AddSingleton<IHostedService>(provider =>
+                    new RabbitMqEventListenerService(
+                        provider.GetRequiredService<HttpPostEventHandler>(),
+                        provider.GetRequiredService<ILogger<RabbitMqEventListenerService>>(),
+                        provider.GetRequiredService<GlobalSettings>(),
+                        globalSettings.EventLogging.RabbitMq.HttpPostQueueName));
             }
         }
     }

test/Common/MockedHttpClient/MockedHttpMessageHandler.cs

@@ -8,6 +8,8 @@ public class MockedHttpMessageHandler : HttpMessageHandler
 {
     private readonly List<IHttpRequestMatcher> _matchers = new();
 
+    public List<HttpRequestMessage> CapturedRequests { get; } = new List<HttpRequestMessage>();
+
     /// <summary>
     /// The fallback handler to use when the request does not match any of the provided matchers.
     /// </summary>
@@ -16,6 +18,7 @@ public class MockedHttpMessageHandler : HttpMessageHandler
 
     protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
     {
+        CapturedRequests.Add(request);
         var matcher = _matchers.FirstOrDefault(x => x.Matches(request));
         if (matcher == null)
         {

test/Core.Test/AdminConsole/Services/EventRepositoryHandlerTests.cs

@@ -0,0 +1,24 @@
+using Bit.Core.Models.Data;
+using Bit.Core.Services;
+using Bit.Test.Common.AutoFixture;
+using Bit.Test.Common.AutoFixture.Attributes;
+using Bit.Test.Common.Helpers;
+using NSubstitute;
+using Xunit;
+
+namespace Bit.Core.Test.Services;
+
+[SutProviderCustomize]
+public class EventRepositoryHandlerTests
+{
+    [Theory, BitAutoData]
+    public async Task HandleEventAsync_WritesEventToIEventWriteService(
+        EventMessage eventMessage,
+        SutProvider<EventRepositoryHandler> sutProvider)
+    {
+        await sutProvider.Sut.HandleEventAsync(eventMessage);
+        await sutProvider.GetDependency<IEventWriteService>().Received(1).CreateAsync(
+            Arg.Is(AssertHelper.AssertPropertyEqual<IEvent>(eventMessage))
+        );
+    }
+}

test/Core.Test/AdminConsole/Services/HttpPostEventHandlerTests.cs

@@ -0,0 +1,66 @@
+using System.Net;
+using System.Net.Http.Json;
+using Bit.Core.Models.Data;
+using Bit.Core.Services;
+using Bit.Test.Common.AutoFixture;
+using Bit.Test.Common.AutoFixture.Attributes;
+using Bit.Test.Common.Helpers;
+using Bit.Test.Common.MockedHttpClient;
+using NSubstitute;
+using Xunit;
+using GlobalSettings = Bit.Core.Settings.GlobalSettings;
+
+namespace Bit.Core.Test.Services;
+
+[SutProviderCustomize]
+public class HttpPostEventHandlerTests
+{
+    private readonly MockedHttpMessageHandler _handler;
+    private HttpClient _httpClient;
+
+    private const string _httpPostUrl = "http://localhost/test/event";
+
+    public HttpPostEventHandlerTests()
+    {
+        _handler = new MockedHttpMessageHandler();
+        _handler.Fallback
+            .WithStatusCode(HttpStatusCode.OK)
+            .WithContent(new StringContent("<html><head><title>test</title></head><body>test</body></html>"));
+        _httpClient = _handler.ToHttpClient();
+    }
+
+    public SutProvider<HttpPostEventHandler> GetSutProvider()
+    {
+        var clientFactory = Substitute.For<IHttpClientFactory>();
+        clientFactory.CreateClient(HttpPostEventHandler.HttpClientName).Returns(_httpClient);
+
+        var globalSettings = new GlobalSettings();
+        globalSettings.EventLogging.RabbitMq.HttpPostUrl = _httpPostUrl;
+
+        return new SutProvider<HttpPostEventHandler>()
+            .SetDependency(globalSettings)
+            .SetDependency(clientFactory)
+            .Create();
+    }
+
+    [Theory, BitAutoData]
+    public async Task HandleEventAsync_PostsEventsToUrl(EventMessage eventMessage)
+    {
+        var sutProvider = GetSutProvider();
+        var content = JsonContent.Create(eventMessage);
+
+        await sutProvider.Sut.HandleEventAsync(eventMessage);
+        sutProvider.GetDependency<IHttpClientFactory>().Received(1).CreateClient(
+            Arg.Is(AssertHelper.AssertPropertyEqual<string>(HttpPostEventHandler.HttpClientName))
+        );
+
+        Assert.Single(_handler.CapturedRequests);
+        var request = _handler.CapturedRequests[0];
+        Assert.NotNull(request);
+        var returned = await request.Content.ReadFromJsonAsync<EventMessage>();
+
+        Assert.Equal(HttpMethod.Post, request.Method);
+        Assert.Equal(_httpPostUrl, request.RequestUri.ToString());
+        AssertHelper.AssertPropertyEqual(eventMessage, returned, new[] { "IdempotencyId" });
+    }
+}