fix(infra): repair TransferFiles compensation + tidy MassTransit consistency

Fix (recoverability bug):
- TransferFilesToBucketActivity returned Completed() with no log, so its
  compensation could never delete the transferred files (orphaned objects on
  fault). IStorageService.TransferFilesAsync now returns the real uploaded
  keys; the activity logs them so Compensate can remove them. Also drops the
  Path.Combine key recomputation (produced backslash keys on Windows).

Consistency / DRY:
- Normalize all courier activities to internal + folder-based namespaces
  (several were public and in the flat .Activities namespace)
- Extract the duplicated exponential retry into UseStandardRetry
- Unify endpoint-name logic into a single EndpointHelper (remove the parallel
  MassTransitEndpointNames helper)

Build: 0 warnings, 0 errors.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-05 04:58:27 +02:00
co-authored by Claude Opus 4.8
parent 423e06baf9
commit d08d82a483
15 changed files with 49 additions and 71 deletions
@@ -14,7 +14,7 @@ namespace Musify.Application.Contracts
Task UploadFileAsync(Stream sourceStream, string contentType, string bucket, string key, CancellationToken cancellationToken); Task UploadFileAsync(Stream sourceStream, string contentType, string bucket, string key, CancellationToken cancellationToken);
Task TransferFilesAsync(string sourceDirectory, string bucket, string route, CancellationToken cancellationToken); Task<IReadOnlyList<string>> TransferFilesAsync(string sourceDirectory, string bucket, string route, CancellationToken cancellationToken);
Task RemoveFileAsync(string bucket, string key, CancellationToken cancellationToken); Task RemoveFileAsync(string bucket, string key, CancellationToken cancellationToken);
@@ -6,7 +6,7 @@ using Musify.Domain.Entities;
using Musify.Domain.ValueObjects; using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities namespace Musify.Infrastructure.MassTransit.Activities.Audio
{ {
internal class GenerateAudioWorkflowPathsActivity( internal class GenerateAudioWorkflowPathsActivity(
IDatabase database, IDatabase database,
@@ -7,7 +7,7 @@ using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs; using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities namespace Musify.Infrastructure.MassTransit.Activities.Audio
{ {
internal class TranscodeAudioActivity( internal class TranscodeAudioActivity(
IDatabase database, IDatabase database,
@@ -4,9 +4,9 @@ using Musify.Application.Contracts;
using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs; using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities namespace Musify.Infrastructure.MassTransit.Activities.Files
{ {
public class TransferFilesToBucketActivity( internal class TransferFilesToBucketActivity(
IStorageService storageHandler, IStorageService storageHandler,
ILogger<TransferFilesToBucketActivity> logger) ILogger<TransferFilesToBucketActivity> logger)
: IActivity<TransferFilesToBucketArguments, TransferFilesToBucketLog> : IActivity<TransferFilesToBucketArguments, TransferFilesToBucketLog>
@@ -22,16 +22,15 @@ namespace Musify.Infrastructure.MassTransit.Activities
try try
{ {
var uploadedKeys = Directory.GetFiles(folderPath) var uploadedKeys = await storageHandler.TransferFilesAsync(
.Select(file => Path.Combine(destinationKey, Path.GetFileName(file)))
.ToArray();
await storageHandler.TransferFilesAsync(
folderPath, folderPath,
executeContext.Arguments.DestinationBucket, executeContext.Arguments.DestinationBucket,
destinationKey, destinationKey,
executeContext.CancellationToken); executeContext.CancellationToken);
return executeContext.Completed();
return executeContext.Completed(new TransferFilesToBucketLog(
executeContext.Arguments.DestinationBucket,
uploadedKeys.ToArray()));
} }
catch (Exception exception) catch (Exception exception)
{ {
@@ -6,7 +6,7 @@ using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Files namespace Musify.Infrastructure.MassTransit.Activities.Files
{ {
public class UploadFileToBucketActivity(IStorageService storageHandler, internal class UploadFileToBucketActivity(IStorageService storageHandler,
ILogger<UploadFileToBucketActivity> logger) ILogger<UploadFileToBucketActivity> logger)
: IActivity<UploadFileToBucketArguments, UploadFileToBucketLog> : IActivity<UploadFileToBucketArguments, UploadFileToBucketLog>
{ {
@@ -2,9 +2,9 @@ using MassTransit;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities namespace Musify.Infrastructure.MassTransit.Activities.Pictures
{ {
public class GeneratePictureWorkflowPathsActivity( internal class GeneratePictureWorkflowPathsActivity(
ILogger<GeneratePictureWorkflowPathsActivity> logger) ILogger<GeneratePictureWorkflowPathsActivity> logger)
: IExecuteActivity<GeneratePictureWorkflowPathsArguments> : IExecuteActivity<GeneratePictureWorkflowPathsArguments>
{ {
@@ -4,9 +4,9 @@ using Musify.Application.Contracts;
using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs; using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities namespace Musify.Infrastructure.MassTransit.Activities.Pictures
{ {
public class ResizePictureActivity( internal class ResizePictureActivity(
IPictureService pictureHandler, IPictureService pictureHandler,
ILogger<ResizePictureActivity> logger) ILogger<ResizePictureActivity> logger)
: IActivity<ResizePictureLocalArguments, ResizePictureLog> : IActivity<ResizePictureLocalArguments, ResizePictureLog>
@@ -4,9 +4,9 @@ using Musify.Application.Contracts;
using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs; using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities namespace Musify.Infrastructure.MassTransit.Activities.Pictures
{ {
public class UpdatePlayListPictureActivity( internal class UpdatePlayListPictureActivity(
IDatabase database, IDatabase database,
ILogger<UpdatePlayListPictureActivity> logger) ILogger<UpdatePlayListPictureActivity> logger)
: IActivity<UpdatePlayListPictureArguments, UpdatePlayListPictureLog> : IActivity<UpdatePlayListPictureArguments, UpdatePlayListPictureLog>
@@ -5,9 +5,9 @@ using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs; using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities namespace Musify.Infrastructure.MassTransit.Activities.Pictures
{ {
public class UpdateTrackPictureActivity( internal class UpdateTrackPictureActivity(
IDatabase database, IDatabase database,
ILogger<UpdateTrackPictureActivity> logger) ILogger<UpdateTrackPictureActivity> logger)
: IActivity<UpdateTrackPictureArguments, UpdateTrackPictureLog> : IActivity<UpdateTrackPictureArguments, UpdateTrackPictureLog>
@@ -2,9 +2,13 @@ namespace Musify.Infrastructure.MassTransit.Consumers
{ {
internal static class EndpointHelper internal static class EndpointHelper
{ {
internal static Uri BuildExecuteActivityUri(string endpointName) => new($"queue:{endpointName}_execute"); internal static string ExecuteQueueName(string endpointName) => $"{endpointName}_execute";
internal static Uri BuildCompensateActivityUri(string endpointName) => new($"queue:{endpointName}_compensate"); 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}");
} }
@@ -6,6 +6,14 @@ namespace Musify.Infrastructure.MassTransit
{ {
public static partial class MassTransitDependencyInjection 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>( static void ConfigureConsumerEndpoint<TConsumer>(
IRabbitMqBusFactoryConfigurator busFactoryConfigurator, IRabbitMqBusFactoryConfigurator busFactoryConfigurator,
IBusRegistrationContext busRegistrationContext, IBusRegistrationContext busRegistrationContext,
@@ -14,15 +22,8 @@ namespace Musify.Infrastructure.MassTransit
{ {
busFactoryConfigurator.ReceiveEndpoint(queueName, endpointConfigurator => busFactoryConfigurator.ReceiveEndpoint(queueName, endpointConfigurator =>
{ {
endpointConfigurator.UseMessageRetry(retryConfigurator => UseStandardRetry(endpointConfigurator);
retryConfigurator.Exponential(
retryLimit: 5,
minInterval: TimeSpan.FromSeconds(1),
maxInterval: TimeSpan.FromSeconds(30),
intervalDelta: TimeSpan.FromSeconds(2)));
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext); endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.ConfigureConsumer<TConsumer>(busRegistrationContext); endpointConfigurator.ConfigureConsumer<TConsumer>(busRegistrationContext);
}); });
} }
@@ -34,17 +35,10 @@ namespace Musify.Infrastructure.MassTransit
where TActivity : class, IExecuteActivity<TArguments> where TActivity : class, IExecuteActivity<TArguments>
where TArguments : class where TArguments : class
{ {
busFactoryConfigurator.ReceiveEndpoint(MassTransitEndpointNames.ExecuteQueue(endpointName), endpointConfigurator => busFactoryConfigurator.ReceiveEndpoint(EndpointHelper.ExecuteQueueName(endpointName), endpointConfigurator =>
{ {
endpointConfigurator.UseMessageRetry(retryConfigurator => UseStandardRetry(endpointConfigurator);
retryConfigurator.Exponential(
retryLimit: 5,
minInterval: TimeSpan.FromSeconds(1),
maxInterval: TimeSpan.FromSeconds(30),
intervalDelta: TimeSpan.FromSeconds(2)));
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext); endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(busRegistrationContext); endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(busRegistrationContext);
}); });
} }
@@ -57,33 +51,19 @@ namespace Musify.Infrastructure.MassTransit
where TArguments : class where TArguments : class
where TLog : class where TLog : class
{ {
busFactoryConfigurator.ReceiveEndpoint(MassTransitEndpointNames.ExecuteQueue(endpointName), endpointConfigurator => busFactoryConfigurator.ReceiveEndpoint(EndpointHelper.ExecuteQueueName(endpointName), endpointConfigurator =>
{ {
endpointConfigurator.UseMessageRetry(retryConfigurator => UseStandardRetry(endpointConfigurator);
retryConfigurator.Exponential(
retryLimit: 5,
minInterval: TimeSpan.FromSeconds(1),
maxInterval: TimeSpan.FromSeconds(30),
intervalDelta: TimeSpan.FromSeconds(2)));
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext); endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>( endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(
EndpointHelper.BuildCompensateActivityUri(endpointName), EndpointHelper.BuildCompensateActivityUri(endpointName),
busRegistrationContext); busRegistrationContext);
}); });
busFactoryConfigurator.ReceiveEndpoint(MassTransitEndpointNames.CompensateQueue(endpointName), endpointConfigurator => busFactoryConfigurator.ReceiveEndpoint(EndpointHelper.CompensateQueueName(endpointName), endpointConfigurator =>
{ {
endpointConfigurator.UseMessageRetry(retryConfigurator => UseStandardRetry(endpointConfigurator);
retryConfigurator.Exponential(
retryLimit: 5,
minInterval: TimeSpan.FromSeconds(1),
maxInterval: TimeSpan.FromSeconds(30),
intervalDelta: TimeSpan.FromSeconds(2)));
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext); endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.CompensateActivityHost<TActivity, TLog>(busRegistrationContext); endpointConfigurator.CompensateActivityHost<TActivity, TLog>(busRegistrationContext);
}); });
} }
@@ -1,7 +1,7 @@
using MassTransit; using MassTransit;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Musify.Infrastructure.Configuration; using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.MassTransit.Activities; 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;
using Musify.Infrastructure.MassTransit.Activities.PlayLists; using Musify.Infrastructure.MassTransit.Activities.PlayLists;
@@ -1,9 +0,0 @@
namespace Musify.Infrastructure.MassTransit
{
internal static class MassTransitEndpointNames
{
internal static string ExecuteQueue(string endpointName) => $"{endpointName}_execute";
internal static string CompensateQueue(string endpointName) => $"{endpointName}_compensate";
}
}
@@ -3,7 +3,7 @@ using MassTransit.Courier.Contracts;
using Musify.Application.Shared; using Musify.Application.Shared;
using Musify.Application.Events; using Musify.Application.Events;
using Musify.Infrastructure.Configuration; using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.MassTransit.Activities; using Musify.Infrastructure.MassTransit.Activities.Pictures;
using Musify.Infrastructure.MassTransit.Activities.Files; using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers; using Musify.Infrastructure.MassTransit.Consumers;
@@ -105,12 +105,13 @@ namespace Musify.Infrastructure.Services
await amazonS3.DeleteObjectAsync(request, cancellationToken); await amazonS3.DeleteObjectAsync(request, cancellationToken);
} }
public async Task TransferFilesAsync(string sourceDirectory, string bucket, string route, CancellationToken cancellationToken) public async Task<IReadOnlyList<string>> TransferFilesAsync(string sourceDirectory, string bucket, string route, CancellationToken cancellationToken)
{ {
var trasnsferUtility = new TransferUtility(amazonS3); var trasnsferUtility = new TransferUtility(amazonS3);
var files = Directory.GetFiles(sourceDirectory, "*"); var files = Directory.GetFiles(sourceDirectory, "*");
var uploadedKeys = new List<string>();
var successCount = 0; var successCount = 0;
var failedCount = 0; var failedCount = 0;
@@ -144,6 +145,7 @@ namespace Musify.Infrastructure.Services
}, },
cancellationToken); cancellationToken);
upload.State = UploadState.Successful; upload.State = UploadState.Successful;
uploadedKeys.Add(key);
successCount++; successCount++;
logger.LogDebug("Transferred file to S3 {Bucket}/{Key}", bucket, key); logger.LogDebug("Transferred file to S3 {Bucket}/{Key}", bucket, key);
} }
@@ -158,6 +160,8 @@ namespace Musify.Infrastructure.Services
await database.SaveChangesAsync(cancellationToken); await database.SaveChangesAsync(cancellationToken);
logger.LogInformation("File transfer completed to {Bucket}/{Route}. Success: {SuccessCount}, Failed: {FailedCount}", bucket, route, successCount, failedCount); logger.LogInformation("File transfer completed to {Bucket}/{Route}. Success: {SuccessCount}, Failed: {FailedCount}", bucket, route, successCount, failedCount);
return uploadedKeys;
} }
public async Task UploadFileAsync(string filePath, string bucket, string key, CancellationToken cancellationToken) public async Task UploadFileAsync(string filePath, string bucket, string key, CancellationToken cancellationToken)