Introduce event publishing management for integration components and adjust Azure Service Bus customization to install the topics during provisioning

src/Directory.Packages.props

@@ -34,7 +34,7 @@
     <PackageVersion Include="Microsoft-WindowsAPICodePack-Shell" Version="1.1.5" />
     <PackageVersion Include="Mindscape.Raygun4Net.NetCore" Version="11.2.5" />
     <PackageVersion Include="NLog.Extensions.Logging" Version="6.1.2" />
-    <PackageVersion Include="NServiceBus" Version="10.1.2" />
+    <PackageVersion Include="NServiceBus" Version="10.1.3" />
     <PackageVersion Include="NServiceBus.AcceptanceTesting" Version="10.1.2" />
     <PackageVersion Include="NServiceBus.AmazonSQS" Version="9.0.0" />
     <PackageVersion Include="NServiceBus.CustomChecks" Version="6.0.0" />

src/ServiceControl.Transports.ASBS/ASBSTransportCustomization.cs

@@ -1,7 +1,11 @@
 namespace ServiceControl.Transports.ASBS
 {
+    using System.Collections.Generic;
     using System.Linq;
     using System.Text.Json;
+    using System.Threading.Tasks;
+    using Azure.Messaging.ServiceBus;
+    using Azure.Messaging.ServiceBus.Administration;
     using BrokerThroughput;
     using Configuration;
     using Microsoft.Extensions.DependencyInjection;
@@ -71,18 +75,7 @@ protected override void AddTransportForPrimaryCore(IServiceCollection services,
                 {
                     TopicToPublishTo = connectionSettings.TopicName,
                     TopicToSubscribeOn = connectionSettings.TopicName,
-                    EventsToMigrateMap =
-                    [
-                        "ServiceControl.Contracts.CustomCheckFailed",
-                        "ServiceControl.Contracts.CustomCheckSucceeded",
-                        "ServiceControl.Contracts.HeartbeatRestored",
-                        "ServiceControl.Contracts.HeartbeatStopped",
-                        "ServiceControl.Contracts.FailedMessagesArchived",
-                        "ServiceControl.Contracts.FailedMessagesUnArchived",
-                        "ServiceControl.Contracts.MessageFailed",
-                        "ServiceControl.Contracts.MessageFailureResolvedByRetry",
-                        "ServiceControl.Contracts.MessageFailureResolvedManually"
-                    ]
+                    EventsToMigrateMap = [.. transportSettings.EventTypesPublished.Select(t => t.FullName)]
                 });
             }
             else if (SettingsReader.TryRead<string>(serviceBusRootNamespace, "Topology", out var topologyJson))
@@ -104,5 +97,47 @@ protected override void AddTransportForMonitoringCore(IServiceCollection service
             services.AddSingleton<IProvideQueueLength, QueueLengthProvider>();
             services.AddHostedService(provider => provider.GetRequiredService<IProvideQueueLength>());
         }
+
+        public override async Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<string> additionalQueues)
+        {
+            await base.ProvisionQueues(transportSettings, additionalQueues);
+
+            if (transportSettings.EventTypesPublished.Count == 0)
+            {
+                return;
+            }
+
+            var connectionSettings = ConnectionStringParser.Parse(transportSettings.ConnectionString);
+
+            var managementClient = connectionSettings.AuthenticationMethod.BuildManagementClient();
+
+            var creationTasks = new List<Task>(transportSettings.EventTypesPublished.Count);
+            foreach (var publishedTopic in transportSettings.EventTypesPublished)
+            {
+                creationTasks.Add(CreateTopic(publishedTopic.FullName));
+            }
+            await Task.WhenAll(creationTasks);
+
+            async Task CreateTopic(string publishedTopic)
+            {
+                var topicToPublishTo = new CreateTopicOptions(connectionSettings.HierarchyNamespace != null
+                    ? $"{connectionSettings.HierarchyNamespace}/{publishedTopic}"
+                    : publishedTopic)
+                {
+                    EnableBatchedOperations = true,
+                    MaxSizeInMegabytes = 5 * 1024, // we are currently not configuring this in the connection string so it uses the same default as the transport
+                    EnablePartitioning = connectionSettings.EnablePartitioning,
+                };
+
+                try
+                {
+                    await managementClient.CreateTopicAsync(topicToPublishTo).ConfigureAwait(false);
+                }
+                catch (ServiceBusException sbe) when (sbe.Reason == ServiceBusFailureReason.MessagingEntityAlreadyExists || sbe.IsTransient)
+                {
+                    // carry on
+                }
+            }
+        }
     }
 }

src/ServiceControl.Transports/TransportSettings.cs

@@ -1,6 +1,7 @@
 namespace ServiceControl.Transports
 {
     using System;
+    using System.Collections.Generic;
     using System.Runtime.Loader;
     using NServiceBus.Settings;
 
@@ -18,6 +19,8 @@ public class TransportSettings : SettingsHolder
 
         public bool RunCustomChecks { get; set; }
 
+        public IReadOnlySet<Type> EventTypesPublished { get; init; } = new HashSet<Type>();
+
         public string ErrorQueue
         {
             set;

src/ServiceControl/ComponentInstallationContext.cs

@@ -1,11 +1,17 @@
 namespace Particular.ServiceControl
 {
+    using System;
     using System.Collections.Generic;
 
-    class ComponentInstallationContext : IComponentInstallationContext
+    public class ComponentInstallationContext : IComponentInstallationContext
     {
-        public List<string> Queues { get; } = [];
+        public IReadOnlyCollection<string> Queues => queuesToCreate;
+        public IReadOnlySet<Type> EventTypesPublished => eventTypePublished;
 
-        public void CreateQueue(string queueName) => Queues.Add(queueName);
+        public void CreateQueue(string queueName) => queuesToCreate.Add(queueName);
+        public void AddEventPublished<TEvent>() => eventTypePublished.Add(typeof(TEvent));
+
+        readonly List<string> queuesToCreate = [];
+        readonly HashSet<Type> eventTypePublished = [];
     }
 }

src/ServiceControl/CustomChecks/CustomChecksComponent.cs

@@ -1,16 +1,28 @@
 namespace ServiceControl.CustomChecks
 {
     using Connection;
+    using Contracts;
     using ExternalIntegrations;
     using Microsoft.Extensions.DependencyInjection;
     using Microsoft.Extensions.Hosting;
+    using NServiceBus;
     using Particular.ServiceControl;
     using ServiceBus.Management.Infrastructure.Settings;
     using Transports;
 
     class CustomChecksComponent : ServiceControlComponent
     {
-        public override void Configure(Settings settings, ITransportCustomization transportCustomization, IHostApplicationBuilder hostBuilder)
+        public override void Setup(Settings settings, IComponentInstallationContext context, IHostApplicationBuilder hostBuilder)
+        {
+            // Integration Events
+            if (!settings.DisableExternalIntegrationsPublishing)
+            {
+                context.AddEventPublished<CustomCheckFailed>();
+                context.AddEventPublished<CustomCheckSucceeded>();
+            }
+        }
+
+        public override void Configure(Settings settings, EndpointConfiguration endpointConfiguration, ITransportCustomization transportCustomization, IHostApplicationBuilder hostBuilder)
         {
             hostBuilder.Services.AddIntegrationEventPublisher<CustomCheckFailedPublisher>();
             hostBuilder.Services.AddIntegrationEventPublisher<CustomCheckSucceededPublisher>();

src/ServiceControl/EventLog/EventLogComponent.cs

@@ -2,14 +2,15 @@
 {
     using Microsoft.Extensions.DependencyInjection;
     using Microsoft.Extensions.Hosting;
+    using NServiceBus;
     using Particular.ServiceControl;
     using ServiceBus.Management.Infrastructure.Settings;
     using ServiceControl.Infrastructure.DomainEvents;
     using Transports;
 
     class EventLogComponent : ServiceControlComponent
     {
-        public override void Configure(Settings settings, ITransportCustomization transportCustomization, IHostApplicationBuilder hostBuilder)
+        public override void Configure(Settings settings, EndpointConfiguration endpointConfiguration, ITransportCustomization transportCustomization, IHostApplicationBuilder hostBuilder)
         {
             var services = hostBuilder.Services;
             services.AddSingleton<EventLogMappings>();

src/ServiceControl/ExternalIntegrations/ExternalIntegrationsComponent.cs

@@ -1,16 +1,26 @@
 namespace ServiceControl.ExternalIntegrations
 {
+    using Infrastructure.DomainEvents;
+    using Microsoft.Extensions.DependencyInjection;
     using Microsoft.Extensions.Hosting;
+    using NServiceBus;
     using Particular.ServiceControl;
     using ServiceBus.Management.Infrastructure.Settings;
     using Transports;
 
     class ExternalIntegrationsComponent : ServiceControlComponent
     {
-        public override void Configure(Settings settings, ITransportCustomization transportCustomization, IHostApplicationBuilder hostBuilder)
+        public override void Configure(Settings settings, EndpointConfiguration endpointConfiguration, ITransportCustomization transportCustomization, IHostApplicationBuilder hostBuilder)
         {
             var services = hostBuilder.Services;
             services.AddEventLogMapping<ExternalIntegrationEventFailedToBePublishedDefinition>();
+
+            if (!settings.DisableExternalIntegrationsPublishing)
+            {
+                services.AddHostedService<EventDispatcherHostedService>();
+                services.AddDomainEventHandler<IntegrationEventWriter>();
+                endpointConfiguration.EnableFeature<ExternalIntegrationsFeature>();
+            }
         }
     }
 }

src/ServiceControl/HostApplicationBuilderExtensions.cs

@@ -4,7 +4,6 @@ namespace Particular.ServiceControl
     using System.Diagnostics;
     using System.Runtime.InteropServices;
     using global::ServiceControl.CustomChecks;
-    using global::ServiceControl.ExternalIntegrations;
     using global::ServiceControl.Hosting;
     using global::ServiceControl.Infrastructure;
     using global::ServiceControl.Infrastructure.BackgroundTasks;
@@ -45,8 +44,15 @@ public static void AddServiceControl(this IHostApplicationBuilder hostBuilder, S
             hostBuilder.Logging.ClearProviders();
             hostBuilder.Logging.ConfigureLogging(settings.LoggingSettings.LogLevel);
 
+            var componentSetupContext = new ComponentInstallationContext();
+            var serviceControlComponents = ServiceControlMainInstance.Components;
+            foreach (ServiceControlComponent component in serviceControlComponents)
+            {
+                component.Setup(settings, componentSetupContext, hostBuilder);
+            }
+
             var services = hostBuilder.Services;
-            var transportSettings = settings.ToTransportSettings();
+            var transportSettings = settings.ToTransportSettings(componentSetupContext);
             var transportCustomization = TransportFactory.Create(transportSettings);
             transportCustomization.AddTransportForPrimary(services, transportSettings);
 
@@ -81,11 +87,6 @@ public static void AddServiceControl(this IHostApplicationBuilder hostBuilder, S
             NServiceBusFactory.Configure(settings, transportCustomization, transportSettings, configuration);
             hostBuilder.UseNServiceBus(configuration);
 
-            if (!settings.DisableExternalIntegrationsPublishing)
-            {
-                hostBuilder.AddExternalIntegrationEvents();
-            }
-
             hostBuilder.AddServicePulseSignalRNotifier();
             hostBuilder.AddEmailNotifications();
             hostBuilder.AddAsyncTimer();
@@ -101,7 +102,7 @@ public static void AddServiceControl(this IHostApplicationBuilder hostBuilder, S
                 hostBuilder.AddWindowsServiceWithRequestTimeout();
             }
 
-            hostBuilder.AddServiceControlComponents(settings, transportCustomization, ServiceControlMainInstance.Components);
+            hostBuilder.AddServiceControlComponents(componentSetupContext, settings, configuration, transportCustomization, serviceControlComponents);
         }
 
         public static void AddServiceControlInstallers(this IHostApplicationBuilder hostApplicationBuilder, Settings settings)

src/ServiceControl/Hosting/Commands/SetupCommand.cs

@@ -40,7 +40,7 @@ public override async Task Execute(HostArguments args, Settings settings)
             }
             else
             {
-                var transportSettings = settings.ToTransportSettings();
+                var transportSettings = settings.ToTransportSettings(componentSetupContext);
                 transportSettings.RunCustomChecks = false;
                 var transportCustomization = TransportFactory.Create(transportSettings);
 

src/ServiceControl/HostingComponent.cs

@@ -5,11 +5,12 @@ namespace Particular.ServiceControl
     using global::ServiceControl.Transports;
     using Microsoft.Extensions.DependencyInjection;
     using Microsoft.Extensions.Hosting;
+    using NServiceBus;
     using ServiceBus.Management.Infrastructure.Settings;
 
     class HostingComponent : ServiceControlComponent
     {
-        public override void Configure(Settings settings, ITransportCustomization transportCustomization, IHostApplicationBuilder hostBuilder)
+        public override void Configure(Settings settings, EndpointConfiguration endpointConfiguration, ITransportCustomization transportCustomization, IHostApplicationBuilder hostBuilder)
         {
             var services = hostBuilder.Services;
             services.AddPlatformConnectionProvider<RemotePlatformConnectionDetailsProvider>();

src/ServiceControl/IComponentInstallationContext.cs

@@ -1,7 +1,9 @@
 namespace Particular.ServiceControl
 {
-    interface IComponentInstallationContext
+    public interface IComponentInstallationContext
     {
         void CreateQueue(string queueName);
+
+        void AddEventPublished<TEvent>();
     }
 }

src/ServiceControl/Infrastructure/NServiceBusFactory.cs

@@ -7,7 +7,6 @@ namespace ServiceBus.Management.Infrastructure
     using NServiceBus;
     using NServiceBus.Configuration.AdvancedExtensibility;
     using ServiceControl.Configuration;
-    using ServiceControl.ExternalIntegrations;
     using ServiceControl.Infrastructure;
     using ServiceControl.Infrastructure.Plugins;
     using ServiceControl.Infrastructure.Subscriptions;
@@ -38,11 +37,6 @@ public static void Configure(Settings.Settings settings, ITransportCustomization
             configuration.GetSettings().Set(settings.LoggingSettings);
             configuration.SetDiagnosticsPath(settings.LoggingSettings.LogPath);
 
-            if (!settings.DisableExternalIntegrationsPublishing)
-            {
-                configuration.EnableFeature<ExternalIntegrationsFeature>();
-            }
-
             var recoverability = configuration.Recoverability();
             recoverability.Immediate(c => c.NumberOfRetries(3));
             recoverability.Delayed(c => c.NumberOfRetries(0));

src/ServiceControl/Infrastructure/Settings/Settings.cs

@@ -9,6 +9,7 @@ namespace ServiceBus.Management.Infrastructure.Settings
     using Microsoft.Extensions.Logging;
     using NLog.Common;
     using NServiceBus.Transport;
+    using Particular.ServiceControl;
     using ServiceControl.Configuration;
     using ServiceControl.Infrastructure;
     using ServiceControl.Infrastructure.Settings;
@@ -238,7 +239,7 @@ public string GetConnectionString()
             return connectionStringSettings?.ConnectionString;
         }
 
-        public TransportSettings ToTransportSettings()
+        public TransportSettings ToTransportSettings(ComponentInstallationContext installationContext)
         {
             var transportSettings = new TransportSettings
             {
@@ -247,7 +248,8 @@ public TransportSettings ToTransportSettings()
                 MaxConcurrency = MaximumConcurrencyLevel,
                 RunCustomChecks = true,
                 TransportType = TransportType,
-                AssemblyLoadContextResolver = AssemblyLoadContextResolver
+                AssemblyLoadContextResolver = AssemblyLoadContextResolver,
+                EventTypesPublished = installationContext.EventTypesPublished,
             };
             return transportSettings;
         }

src/ServiceControl/Licensing/LicensingComponent.cs

@@ -4,13 +4,14 @@ namespace Particular.ServiceControl;
 using global::ServiceControl.LicenseManagement;
 using global::ServiceControl.Transports;
 using Microsoft.Extensions.Hosting;
+using NServiceBus;
 using Particular.LicensingComponent;
 using Particular.LicensingComponent.Shared;
 using ServiceBus.Management.Infrastructure.Settings;
 
 class LicensingComponent : ServiceControlComponent
 {
-    public override void Configure(Settings settings, ITransportCustomization transportCustomization,
+    public override void Configure(Settings settings, EndpointConfiguration endpointConfiguration, ITransportCustomization transportCustomization,
         IHostApplicationBuilder hostBuilder)
     {
         var licenseDetails = LicenseManager.FindLicense().Details;