[PM-17562] Add feature flag for event-based organization integrations

src/Api/AdminConsole/Controllers/OrganizationIntegrationConfigurationController.cs

@@ -1,13 +1,16 @@
 using Bit.Api.AdminConsole.Models.Request.Organizations;
 using Bit.Api.AdminConsole.Models.Response.Organizations;
+using Bit.Core;
 using Bit.Core.Context;
 using Bit.Core.Exceptions;
 using Bit.Core.Repositories;
+using Bit.Core.Utilities;
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Mvc;
 
 namespace Bit.Api.AdminConsole.Controllers;
 
+[RequireFeature(FeatureFlagKeys.EventBasedOrganizationIntegrations)]
 [Route("organizations/{organizationId:guid}/integrations/{integrationId:guid}/configurations")]
 [Authorize("Application")]
 public class OrganizationIntegrationConfigurationController(

src/Api/AdminConsole/Controllers/OrganizationIntegrationController.cs

@@ -1,15 +1,18 @@
 using Bit.Api.AdminConsole.Models.Request.Organizations;
 using Bit.Api.AdminConsole.Models.Response.Organizations;
+using Bit.Core;
 using Bit.Core.Context;
 using Bit.Core.Exceptions;
 using Bit.Core.Repositories;
+using Bit.Core.Utilities;
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Mvc;
 
 #nullable enable
 
 namespace Bit.Api.AdminConsole.Controllers;
 
+[RequireFeature(FeatureFlagKeys.EventBasedOrganizationIntegrations)]
 [Route("organizations/{organizationId:guid}/integrations")]
 [Authorize("Application")]
 public class OrganizationIntegrationController(

src/Api/AdminConsole/Controllers/SlackIntegrationController.cs

@@ -1,17 +1,20 @@
 using System.Text.Json;
 using Bit.Api.AdminConsole.Models.Response.Organizations;
+using Bit.Core;
 using Bit.Core.AdminConsole.Entities;
 using Bit.Core.Context;
 using Bit.Core.Enums;
 using Bit.Core.Exceptions;
 using Bit.Core.Models.Data.Integrations;
 using Bit.Core.Repositories;
 using Bit.Core.Services;
+using Bit.Core.Utilities;
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Mvc;
 
 namespace Bit.Api.AdminConsole.Controllers;
 
+[RequireFeature(FeatureFlagKeys.EventBasedOrganizationIntegrations)]
 [Route("organizations/{organizationId:guid}/integrations/slack")]
 [Authorize("Application")]
 public class SlackIntegrationController(

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

@@ -0,0 +1,34 @@
+using Bit.Core.Models.Data;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Bit.Core.Services;
+
+public class EventRouteService(
+    [FromKeyedServices("broadcast")] IEventWriteService broadcastEventWriteService,
+    [FromKeyedServices("storage")] IEventWriteService storageEventWriteService,
+    IFeatureService _featureService) : IEventWriteService
+{
+    public async Task CreateAsync(IEvent e)
+    {
+        if (_featureService.IsEnabled(FeatureFlagKeys.EventBasedOrganizationIntegrations))
+        {
+            await broadcastEventWriteService.CreateAsync(e);
+        }
+        else
+        {
+            await storageEventWriteService.CreateAsync(e);
+        }
+    }
+
+    public async Task CreateManyAsync(IEnumerable<IEvent> e)
+    {
+        if (_featureService.IsEnabled(FeatureFlagKeys.EventBasedOrganizationIntegrations))
+        {
+            await broadcastEventWriteService.CreateManyAsync(e);
+        }
+        else
+        {
+            await storageEventWriteService.CreateManyAsync(e);
+        }
+    }
+}

src/Core/Constants.cs

@@ -108,6 +108,7 @@ public static class FeatureFlagKeys
     public const string PolicyRequirements = "pm-14439-policy-requirements";
     public const string SsoExternalIdVisibility = "pm-18630-sso-external-id-visibility";
     public const string ScimInviteUserOptimization = "pm-16811-optimize-invite-user-flow-to-fail-fast";
+    public const string EventBasedOrganizationIntegrations = "event-based-organization-integrations";
 
     /* Auth Team */
     public const string PM9112DeviceApprovalPersistence = "pm-9112-device-approval-persistence";

src/Events/Startup.cs

@@ -62,33 +62,39 @@
         {
             services.AddSingleton<IApplicationCacheService, InMemoryApplicationCacheService>();
         }
-        services.AddScoped<IEventService, EventService>();
+
         if (!globalSettings.SelfHosted && CoreHelpers.SettingHasValue(globalSettings.Events.ConnectionString))
         {
+            services.AddKeyedSingleton<IEventWriteService, AzureQueueEventWriteService>("storage");
+
             if (CoreHelpers.SettingHasValue(globalSettings.EventLogging.AzureServiceBus.ConnectionString) &&
                 CoreHelpers.SettingHasValue(globalSettings.EventLogging.AzureServiceBus.TopicName))
             {
-                services.AddSingleton<IEventWriteService, AzureServiceBusEventWriteService>();
+                services.AddKeyedSingleton<IEventWriteService, AzureServiceBusEventWriteService>("broadcast");
             }
             else
             {
-                services.AddSingleton<IEventWriteService, AzureQueueEventWriteService>();
+                services.AddKeyedSingleton<IEventWriteService, NoopEventWriteService>("broadcast");
             }
         }
         else
         {
+            services.AddKeyedSingleton<IEventWriteService, RepositoryEventWriteService>("storage");
+
             if (CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.HostName) &&
                 CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.Username) &&
                 CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.Password) &&
                 CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.ExchangeName))
             {
-                services.AddSingleton<IEventWriteService, RabbitMqEventWriteService>();
+                services.AddKeyedSingleton<IEventWriteService, RabbitMqEventWriteService>("broadcast");
             }
             else
             {
-                services.AddSingleton<IEventWriteService, RepositoryEventWriteService>();
+                services.AddKeyedSingleton<IEventWriteService, NoopEventWriteService>("broadcast");
             }
         }
+        services.AddScoped<IEventWriteService, EventRouteService>();
+        services.AddScoped<IEventService, EventService>();
 
         services.AddOptionality();
 

src/SharedWeb/Utilities/ServiceCollectionExtensions.cs

@@ -332,34 +332,40 @@
 
         if (!globalSettings.SelfHosted && CoreHelpers.SettingHasValue(globalSettings.Events.ConnectionString))
         {
+            services.AddKeyedSingleton<IEventWriteService, AzureQueueEventWriteService>("storage");
+
             if (CoreHelpers.SettingHasValue(globalSettings.EventLogging.AzureServiceBus.ConnectionString) &&
                 CoreHelpers.SettingHasValue(globalSettings.EventLogging.AzureServiceBus.TopicName))
             {
-                services.AddSingleton<IEventWriteService, AzureServiceBusEventWriteService>();
+                services.AddKeyedSingleton<IEventWriteService, AzureServiceBusEventWriteService>("broadcast");
             }
             else
             {
-                services.AddSingleton<IEventWriteService, AzureQueueEventWriteService>();
+                services.AddKeyedSingleton<IEventWriteService, NoopEventWriteService>("broadcast");
             }
         }
         else if (globalSettings.SelfHosted)
         {
+            services.AddKeyedSingleton<IEventWriteService, RepositoryEventWriteService>("storage");
+
             if (CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.HostName) &&
                 CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.Username) &&
                 CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.Password) &&
                 CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.ExchangeName))
             {
-                services.AddSingleton<IEventWriteService, RabbitMqEventWriteService>();
+                services.AddKeyedSingleton<IEventWriteService, RabbitMqEventWriteService>("broadcast");
             }
             else
             {
-                services.AddSingleton<IEventWriteService, RepositoryEventWriteService>();
+                services.AddKeyedSingleton<IEventWriteService, NoopEventWriteService>("broadcast");
             }
         }
         else
         {
-            services.AddSingleton<IEventWriteService, NoopEventWriteService>();
+            services.AddKeyedSingleton<IEventWriteService, NoopEventWriteService>("storage");
+            services.AddKeyedSingleton<IEventWriteService, NoopEventWriteService>("broadcast");
         }
+        services.AddScoped<IEventWriteService, EventRouteService>();
 
         if (CoreHelpers.SettingHasValue(globalSettings.Attachment.ConnectionString))
         {

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

@@ -0,0 +1,65 @@
+using Bit.Core.Models.Data;
+using Bit.Core.Services;
+using Bit.Test.Common.AutoFixture.Attributes;
+using NSubstitute;
+using Xunit;
+
+namespace Bit.Core.Test.Services;
+
+[SutProviderCustomize]
+public class EventRouteServiceTests
+{
+    private readonly IEventWriteService _broadcastEventWriteService = Substitute.For<IEventWriteService>();
+    private readonly IEventWriteService _storageEventWriteService = Substitute.For<IEventWriteService>();
+    private readonly IFeatureService _featureService = Substitute.For<IFeatureService>();
+    private readonly EventRouteService Subject;
+
+    public EventRouteServiceTests()
+    {
+        Subject = new EventRouteService(_broadcastEventWriteService, _storageEventWriteService, _featureService);
+    }
+
+    [Theory, BitAutoData]
+    public async Task CreateAsync_FlagDisabled_EventSentToStorageService(EventMessage eventMessage)
+    {
+        _featureService.IsEnabled(FeatureFlagKeys.EventBasedOrganizationIntegrations).Returns(false);
+
+        await Subject.CreateAsync(eventMessage);
+
+        _broadcastEventWriteService.DidNotReceiveWithAnyArgs().CreateAsync(Arg.Any<EventMessage>());
+        _storageEventWriteService.Received(1).CreateAsync(eventMessage);
+    }
+
+    [Theory, BitAutoData]
+    public async Task CreateAsync_FlagEnabled_EventSentToBroadcastService(EventMessage eventMessage)
+    {
+        _featureService.IsEnabled(FeatureFlagKeys.EventBasedOrganizationIntegrations).Returns(true);
+
+        await Subject.CreateAsync(eventMessage);
+
+        _broadcastEventWriteService.Received(1).CreateAsync(eventMessage);
+        _storageEventWriteService.DidNotReceiveWithAnyArgs().CreateAsync(Arg.Any<EventMessage>());
+    }
+
+    [Theory, BitAutoData]
+    public async Task CreateManyAsync_FlagDisabled_EventsSentToStorageService(IEnumerable<EventMessage> eventMessages)
+    {
+        _featureService.IsEnabled(FeatureFlagKeys.EventBasedOrganizationIntegrations).Returns(false);
+
+        await Subject.CreateManyAsync(eventMessages);
+
+        _broadcastEventWriteService.DidNotReceiveWithAnyArgs().CreateManyAsync(Arg.Any<IEnumerable<EventMessage>>());
+        _storageEventWriteService.Received(1).CreateManyAsync(eventMessages);
+    }
+
+    [Theory, BitAutoData]
+    public async Task CreateManyAsync_FlagEnabled_EventsSentToBroadcastService(IEnumerable<EventMessage> eventMessages)
+    {
+        _featureService.IsEnabled(FeatureFlagKeys.EventBasedOrganizationIntegrations).Returns(true);
+
+        await Subject.CreateManyAsync(eventMessages);
+
+        _broadcastEventWriteService.Received(1).CreateManyAsync(eventMessages);
+        _storageEventWriteService.DidNotReceiveWithAnyArgs().CreateManyAsync(Arg.Any<IEnumerable<EventMessage>>());
+    }
+}