refactor(infra): auto-configure MassTransit endpoints via ConfigureEndpoints

Replace ~240 lines of hand-written per-endpoint wiring with MassTransit's
ConfigureEndpoints + KebabCaseEndpointNameFormatter, and apply retry + EF
outbox globally through AddConfigureEndpointsCallback. Adding a consumer or
activity is now a single AddConsumer/AddActivity line.

- Normalize the 4 mismatched activity endpoint names to kebab-case so the
  routing-slip addressing keeps resolving (update-play-list-picture,
  delete/mark/publish play-list variants); cleanup consumer queue is now
  routing-slip-clean-up
- Drop the now-unused QueueName consts from event consumers and trim
  EndpointHelper / delete the manual endpoint-config partial

Verified end-to-end: booted the Worker against RabbitMQ in Docker and
confirmed all 38 endpoints are configured with the exact names the builders
address (19 execute queues + compensations + consumers + routing-slip-clean-up).

Build: 0 warnings, 0 errors.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-05 14:22:26 +02:00
co-authored by Claude Opus 4.8
parent d08d82a483
commit 3545d71016
17 changed files with 29 additions and 242 deletions
@@ -11,7 +11,7 @@ namespace Musify.Infrastructure.MassTransit.Activities.Pictures
ILogger<UpdatePlayListPictureActivity> logger) ILogger<UpdatePlayListPictureActivity> logger)
: IActivity<UpdatePlayListPictureArguments, UpdatePlayListPictureLog> : IActivity<UpdatePlayListPictureArguments, UpdatePlayListPictureLog>
{ {
public const string ExecuteEndpointName = "update-playlist-picture"; public const string ExecuteEndpointName = "update-play-list-picture";
public async Task<ExecutionResult> Execute(ExecuteContext<UpdatePlayListPictureArguments> executeContext) public async Task<ExecutionResult> Execute(ExecuteContext<UpdatePlayListPictureArguments> executeContext)
{ {
@@ -11,7 +11,7 @@ namespace Musify.Infrastructure.MassTransit.Activities.PlayLists
ILogger<DeletePlayListFromDbActivity> logger) ILogger<DeletePlayListFromDbActivity> logger)
: IExecuteActivity<DeletePlayListFromDbArguments> : IExecuteActivity<DeletePlayListFromDbArguments>
{ {
public const string ExecuteEndpointName = "delete-playList-from-db"; public const string ExecuteEndpointName = "delete-play-list-from-db";
public async Task<ExecutionResult> Execute(ExecuteContext<DeletePlayListFromDbArguments> executeContext) public async Task<ExecutionResult> Execute(ExecuteContext<DeletePlayListFromDbArguments> executeContext)
{ {
@@ -12,7 +12,7 @@ namespace Musify.Infrastructure.MassTransit.Activities.PlayLists
ILogger<MarkPlayListAsRemovingActivity> logger) ILogger<MarkPlayListAsRemovingActivity> logger)
: IExecuteActivity<MarkPlayListAsRemovingArguments> : IExecuteActivity<MarkPlayListAsRemovingArguments>
{ {
public const string ExecuteEndpointName = "mark-playList-as-removing"; public const string ExecuteEndpointName = "mark-play-list-as-removing";
public async Task<ExecutionResult> Execute(ExecuteContext<MarkPlayListAsRemovingArguments> executeContext) public async Task<ExecutionResult> Execute(ExecuteContext<MarkPlayListAsRemovingArguments> executeContext)
{ {
@@ -10,7 +10,7 @@ namespace Musify.Infrastructure.MassTransit.Activities.PlayLists
ILogger<PublishPlayListPictureProcessingEventActivity> logger) ILogger<PublishPlayListPictureProcessingEventActivity> logger)
: IExecuteActivity<PublishPlayListPictureProcessingEventArguments> : IExecuteActivity<PublishPlayListPictureProcessingEventArguments>
{ {
public const string ExecuteEndpointName = "publish-playList-picture-processing-event"; public const string ExecuteEndpointName = "publish-play-list-picture-processing-event";
public async Task<ExecutionResult> Execute(ExecuteContext<PublishPlayListPictureProcessingEventArguments> executeContext) public async Task<ExecutionResult> Execute(ExecuteContext<PublishPlayListPictureProcessingEventArguments> executeContext)
{ {
@@ -9,7 +9,6 @@ namespace Musify.Infrastructure.MassTransit.Consumers
PlayListPictureSourceRoutingSlipBuilder routingSlipBuilder) PlayListPictureSourceRoutingSlipBuilder routingSlipBuilder)
: IConsumer<CreatePlayListResourcesEvent> : IConsumer<CreatePlayListResourcesEvent>
{ {
public const string QueueName = "create-playList";
public async Task Consume(ConsumeContext<CreatePlayListResourcesEvent> consumeContext) public async Task Consume(ConsumeContext<CreatePlayListResourcesEvent> consumeContext)
{ {
@@ -9,7 +9,6 @@ namespace Musify.Infrastructure.MassTransit.Consumers
CreateTrackRoutingSlipBuilder routingSlipBuilder) CreateTrackRoutingSlipBuilder routingSlipBuilder)
: IConsumer<CreateTrackResourcesEvent> : IConsumer<CreateTrackResourcesEvent>
{ {
public const string QueueName = "create-track";
public async Task Consume(ConsumeContext<CreateTrackResourcesEvent> consumeContext) public async Task Consume(ConsumeContext<CreateTrackResourcesEvent> consumeContext)
{ {
@@ -9,7 +9,6 @@ namespace Musify.Infrastructure.MassTransit.Consumers
DeletePlayListRoutingSlipBuilder routingSlipBuilder) DeletePlayListRoutingSlipBuilder routingSlipBuilder)
: IConsumer<DeletePlayListEvent> : IConsumer<DeletePlayListEvent>
{ {
public const string QueueName = "delete-playList";
public async Task Consume(ConsumeContext<DeletePlayListEvent> consumeContext) public async Task Consume(ConsumeContext<DeletePlayListEvent> consumeContext)
{ {
@@ -9,7 +9,6 @@ namespace Musify.Infrastructure.MassTransit.Consumers
DeleteTrackRoutingSlipBuilder routingSlipBuilder) DeleteTrackRoutingSlipBuilder routingSlipBuilder)
: IConsumer<DeleteTrackEvent> : IConsumer<DeleteTrackEvent>
{ {
public const string QueueName = "delete-track";
public async Task Consume(ConsumeContext<DeleteTrackEvent> consumeContext) public async Task Consume(ConsumeContext<DeleteTrackEvent> consumeContext)
{ {
@@ -2,13 +2,7 @@ namespace Musify.Infrastructure.MassTransit.Consumers
{ {
internal static class EndpointHelper internal static class EndpointHelper
{ {
internal static string ExecuteQueueName(string endpointName) => $"{endpointName}_execute"; internal static Uri BuildExecuteActivityUri(string endpointName) => new($"queue:{endpointName}_execute");
internal static string CompensateQueueName(string endpointName) => $"{endpointName}_compensate";
internal static Uri BuildExecuteActivityUri(string endpointName) => new($"queue:{ExecuteQueueName(endpointName)}");
internal static Uri BuildCompensateActivityUri(string endpointName) => new($"queue:{CompensateQueueName(endpointName)}");
internal static Uri BuildConsumerUri(string endpointName) => new($"queue:{endpointName}"); internal static Uri BuildConsumerUri(string endpointName) => new($"queue:{endpointName}");
} }
@@ -9,7 +9,7 @@ namespace Musify.Infrastructure.MassTransit.Consumers
IConsumer<RoutingSlipCompleted>, IConsumer<RoutingSlipCompleted>,
IConsumer<RoutingSlipFaulted> IConsumer<RoutingSlipFaulted>
{ {
public const string QueueName = "routing-slip-events"; public const string QueueName = "routing-slip-clean-up";
public Task Consume(ConsumeContext<RoutingSlipCompleted> consumeContext) public Task Consume(ConsumeContext<RoutingSlipCompleted> consumeContext)
=> DeleteFolder(consumeContext.Message.Variables); => DeleteFolder(consumeContext.Message.Variables);
@@ -9,7 +9,6 @@ namespace Musify.Infrastructure.MassTransit.Consumers
PictureWorkflowRoutingSlipBuilder routingSlipBuilder) PictureWorkflowRoutingSlipBuilder routingSlipBuilder)
: IConsumer<UpdatePlayListPictureEvent> : IConsumer<UpdatePlayListPictureEvent>
{ {
public const string QueueName = "update-playList-picture";
public async Task Consume(ConsumeContext<UpdatePlayListPictureEvent> consumeContext) public async Task Consume(ConsumeContext<UpdatePlayListPictureEvent> consumeContext)
{ {
@@ -9,7 +9,6 @@ namespace Musify.Infrastructure.MassTransit.Consumers
PlayListPictureSourceRoutingSlipBuilder routingSlipBuilder) PlayListPictureSourceRoutingSlipBuilder routingSlipBuilder)
: IConsumer<UpdatePlayListPictureSourceEvent> : IConsumer<UpdatePlayListPictureSourceEvent>
{ {
public const string QueueName = "update-playList-picture-source";
public async Task Consume(ConsumeContext<UpdatePlayListPictureSourceEvent> consumeContext) public async Task Consume(ConsumeContext<UpdatePlayListPictureSourceEvent> consumeContext)
{ {
@@ -9,7 +9,6 @@ namespace Musify.Infrastructure.MassTransit.Consumers
AudioWorkflowRoutingSlipBuilder routingSlipBuilder) AudioWorkflowRoutingSlipBuilder routingSlipBuilder)
: IConsumer<UpdateTrackAudioEvent> : IConsumer<UpdateTrackAudioEvent>
{ {
public const string QueueName = "update-track-audio";
public async Task Consume(ConsumeContext<UpdateTrackAudioEvent> consumeContext) public async Task Consume(ConsumeContext<UpdateTrackAudioEvent> consumeContext)
{ {
@@ -9,7 +9,6 @@ namespace Musify.Infrastructure.MassTransit.Consumers
PictureWorkflowRoutingSlipBuilder routingSlipBuilder) PictureWorkflowRoutingSlipBuilder routingSlipBuilder)
: IConsumer<UpdateTrackPictureEvent> : IConsumer<UpdateTrackPictureEvent>
{ {
public const string QueueName = "update-track-picture";
public async Task Consume(ConsumeContext<UpdateTrackPictureEvent> consumeContext) public async Task Consume(ConsumeContext<UpdateTrackPictureEvent> consumeContext)
{ {
@@ -1,71 +0,0 @@
using MassTransit;
using Musify.Infrastructure.MassTransit.Consumers;
using Musify.Infrastructure.Persistence;
namespace Musify.Infrastructure.MassTransit
{
public static partial class MassTransitDependencyInjection
{
static void UseStandardRetry(IReceiveEndpointConfigurator endpointConfigurator) =>
endpointConfigurator.UseMessageRetry(retryConfigurator =>
retryConfigurator.Exponential(
retryLimit: 5,
minInterval: TimeSpan.FromSeconds(1),
maxInterval: TimeSpan.FromSeconds(30),
intervalDelta: TimeSpan.FromSeconds(2)));
static void ConfigureConsumerEndpoint<TConsumer>(
IRabbitMqBusFactoryConfigurator busFactoryConfigurator,
IBusRegistrationContext busRegistrationContext,
string queueName)
where TConsumer : class, IConsumer
{
busFactoryConfigurator.ReceiveEndpoint(queueName, endpointConfigurator =>
{
UseStandardRetry(endpointConfigurator);
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.ConfigureConsumer<TConsumer>(busRegistrationContext);
});
}
static void ConfigureExecuteActivityEndpoint<TActivity, TArguments>(
IRabbitMqBusFactoryConfigurator busFactoryConfigurator,
IBusRegistrationContext busRegistrationContext,
string endpointName)
where TActivity : class, IExecuteActivity<TArguments>
where TArguments : class
{
busFactoryConfigurator.ReceiveEndpoint(EndpointHelper.ExecuteQueueName(endpointName), endpointConfigurator =>
{
UseStandardRetry(endpointConfigurator);
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(busRegistrationContext);
});
}
static void ConfigureActivityEndpoint<TActivity, TArguments, TLog>(
IRabbitMqBusFactoryConfigurator busFactoryConfigurator,
IBusRegistrationContext busRegistrationContext,
string endpointName)
where TActivity : class, IActivity<TArguments, TLog>
where TArguments : class
where TLog : class
{
busFactoryConfigurator.ReceiveEndpoint(EndpointHelper.ExecuteQueueName(endpointName), endpointConfigurator =>
{
UseStandardRetry(endpointConfigurator);
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(
EndpointHelper.BuildCompensateActivityUri(endpointName),
busRegistrationContext);
});
busFactoryConfigurator.ReceiveEndpoint(EndpointHelper.CompensateQueueName(endpointName), endpointConfigurator =>
{
UseStandardRetry(endpointConfigurator);
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.CompensateActivityHost<TActivity, TLog>(busRegistrationContext);
});
}
}
}
@@ -1,6 +1,5 @@
using MassTransit; using MassTransit;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.MassTransit.Activities.Pictures; using Musify.Infrastructure.MassTransit.Activities.Pictures;
using Musify.Infrastructure.MassTransit.Activities.Audio; using Musify.Infrastructure.MassTransit.Activities.Audio;
using Musify.Infrastructure.MassTransit.Activities.Files; using Musify.Infrastructure.MassTransit.Activities.Files;
@@ -59,154 +58,5 @@ namespace Musify.Infrastructure.MassTransit
options.AddActivity<CopyFileInBucketActivity, CopyFileInBucketArguments, CopyFileInBucketLog>(); options.AddActivity<CopyFileInBucketActivity, CopyFileInBucketArguments, CopyFileInBucketLog>();
options.AddActivity<ConsumeUploadIntentsActivity, ConsumeUploadIntentsArguments, ConsumeUploadIntentsLog>(); options.AddActivity<ConsumeUploadIntentsActivity, ConsumeUploadIntentsArguments, ConsumeUploadIntentsLog>();
} }
static void ConfigureRabbitMqWorkerEndpoints(
IBusRegistrationContext busRegistrationContext,
IRabbitMqBusFactoryConfigurator busFactoryConfigurator)
{
ConfigureRabbitMqHost(
busFactoryConfigurator,
busRegistrationContext.GetRequiredService<MassTransitConfiguration>());
ConfigureConsumerEndpoint<UpdatePlayListPictureConsumer>(
busFactoryConfigurator,
busRegistrationContext,
UpdatePlayListPictureConsumer.QueueName);
ConfigureConsumerEndpoint<UpdateTrackAudioConsumer>(
busFactoryConfigurator,
busRegistrationContext,
UpdateTrackAudioConsumer.QueueName);
ConfigureConsumerEndpoint<UpdateTrackPictureConsumer>(
busFactoryConfigurator,
busRegistrationContext,
UpdateTrackPictureConsumer.QueueName);
ConfigureConsumerEndpoint<DeleteTrackConsumer>(
busFactoryConfigurator,
busRegistrationContext,
DeleteTrackConsumer.QueueName);
ConfigureConsumerEndpoint<DeletePlayListConsumer>(
busFactoryConfigurator,
busRegistrationContext,
DeletePlayListConsumer.QueueName);
ConfigureConsumerEndpoint<CreateTrackConsumer>(
busFactoryConfigurator,
busRegistrationContext,
CreateTrackConsumer.QueueName);
ConfigureConsumerEndpoint<CreatePlayListConsumer>(
busFactoryConfigurator,
busRegistrationContext,
CreatePlayListConsumer.QueueName);
ConfigureConsumerEndpoint<UpdatePlayListPictureSourceConsumer>(
busFactoryConfigurator,
busRegistrationContext,
UpdatePlayListPictureSourceConsumer.QueueName);
ConfigureConsumerEndpoint<RoutingSlipCleanUpConsumer>(
busFactoryConfigurator,
busRegistrationContext,
RoutingSlipCleanUpConsumer.QueueName);
ConfigureExecuteActivityEndpoint<RemoveFileFromBucketActivity, RemoveFileFromBucketArguments>(
busFactoryConfigurator,
busRegistrationContext,
RemoveFileFromBucketActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<GenerateAudioWorkflowPathsActivity, GenerateAudioWorkflowPathsArguments>(
busFactoryConfigurator,
busRegistrationContext,
GenerateAudioWorkflowPathsActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<GeneratePictureWorkflowPathsActivity, GeneratePictureWorkflowPathsArguments>(
busFactoryConfigurator,
busRegistrationContext,
GeneratePictureWorkflowPathsActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<MarkTrackAsRemovingActivity, MarkTrackAsRemovingArguments>(
busFactoryConfigurator,
busRegistrationContext,
MarkTrackAsRemovingActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<DeleteTrackFromDbActivity, DeleteTrackFromDbArguments>(
busFactoryConfigurator,
busRegistrationContext,
DeleteTrackFromDbActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<MarkPlayListAsRemovingActivity, MarkPlayListAsRemovingArguments>(
busFactoryConfigurator,
busRegistrationContext,
MarkPlayListAsRemovingActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<DeletePlayListFromDbActivity, DeletePlayListFromDbArguments>(
busFactoryConfigurator,
busRegistrationContext,
DeletePlayListFromDbActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<PublishTrackProcessingEventsActivity, PublishTrackProcessingEventsArguments>(
busFactoryConfigurator,
busRegistrationContext,
PublishTrackProcessingEventsActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<PublishPlayListPictureProcessingEventActivity, PublishPlayListPictureProcessingEventArguments>(
busFactoryConfigurator,
busRegistrationContext,
PublishPlayListPictureProcessingEventActivity.ExecuteEndpointName);
ConfigureActivityEndpoint<ResizePictureActivity, ResizePictureLocalArguments, ResizePictureLog>(
busFactoryConfigurator,
busRegistrationContext,
ResizePictureActivity.ExecuteEndpointName);
ConfigureActivityEndpoint<UpdatePlayListPictureActivity, UpdatePlayListPictureArguments, UpdatePlayListPictureLog>(
busFactoryConfigurator,
busRegistrationContext,
UpdatePlayListPictureActivity.ExecuteEndpointName);
ConfigureActivityEndpoint<UpdateTrackPictureActivity, UpdateTrackPictureArguments, UpdateTrackPictureLog>(
busFactoryConfigurator,
busRegistrationContext,
UpdateTrackPictureActivity.ExecuteEndpointName);
ConfigureActivityEndpoint<DownloadFileFromBucketActivity, DownloadFileFromBucketArguments, DownloadFileFromBucketLog>(
busFactoryConfigurator,
busRegistrationContext,
DownloadFileFromBucketActivity.ExecuteEndpointName);
ConfigureActivityEndpoint<TranscodeAudioActivity, TranscodeAudioArguments, TranscodeAudioLog>(
busFactoryConfigurator,
busRegistrationContext,
TranscodeAudioActivity.ExecuteEndpointName);
ConfigureActivityEndpoint<TransferFilesToBucketActivity, TransferFilesToBucketArguments, TransferFilesToBucketLog>(
busFactoryConfigurator,
busRegistrationContext,
TransferFilesToBucketActivity.ExecuteEndpointName);
ConfigureActivityEndpoint<UpdateTrackAudioActivity, UpdateTrackAudioArguments, UpdateTrackAudioLog>(
busFactoryConfigurator,
busRegistrationContext,
UpdateTrackAudioActivity.ExecuteEndpointName);
ConfigureActivityEndpoint<UploadFileToBucketActivity, UploadFileToBucketArguments, UploadFileToBucketLog>(
busFactoryConfigurator,
busRegistrationContext,
UploadFileToBucketActivity.ExecuteEndpointName);
ConfigureActivityEndpoint<CopyFileInBucketActivity, CopyFileInBucketArguments, CopyFileInBucketLog>(
busFactoryConfigurator,
busRegistrationContext,
CopyFileInBucketActivity.ExecuteEndpointName);
ConfigureActivityEndpoint<ConsumeUploadIntentsActivity, ConsumeUploadIntentsArguments, ConsumeUploadIntentsLog>(
busFactoryConfigurator,
busRegistrationContext,
ConsumeUploadIntentsActivity.ExecuteEndpointName);
}
} }
} }
@@ -62,15 +62,37 @@ namespace Musify.Infrastructure.MassTransit
outbox.DuplicateDetectionWindow = TimeSpan.FromMinutes(30); outbox.DuplicateDetectionWindow = TimeSpan.FromMinutes(30);
}); });
options.SetKebabCaseEndpointNameFormatter();
RegisterConsumersAndActivities(options); RegisterConsumersAndActivities(options);
options.AddConfigureEndpointsCallback((registrationContext, _, endpointConfigurator) =>
{
UseStandardRetry(endpointConfigurator);
endpointConfigurator.UseEntityFrameworkOutbox<Database>(registrationContext);
});
options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) => options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) =>
ConfigureRabbitMqWorkerEndpoints(busRegistrationContext, busFactoryConfigurator)); {
ConfigureRabbitMqHost(
busFactoryConfigurator,
busRegistrationContext.GetRequiredService<MassTransitConfiguration>());
busFactoryConfigurator.ConfigureEndpoints(busRegistrationContext);
});
} ); } );
return serviceDescriptors; return serviceDescriptors;
} }
static void UseStandardRetry(IReceiveEndpointConfigurator endpointConfigurator) =>
endpointConfigurator.UseMessageRetry(retryConfigurator =>
retryConfigurator.Exponential(
retryLimit: 5,
minInterval: TimeSpan.FromSeconds(1),
maxInterval: TimeSpan.FromSeconds(30),
intervalDelta: TimeSpan.FromSeconds(2)));
static void ConfigureRabbitMqHost( static void ConfigureRabbitMqHost(
IRabbitMqBusFactoryConfigurator busFactoryConfigurator, IRabbitMqBusFactoryConfigurator busFactoryConfigurator,
MassTransitConfiguration massTransitConfiguration) MassTransitConfiguration massTransitConfiguration)