A lot of changes

This commit is contained in:
2026-05-14 14:32:19 +02:00
parent 7160266763
commit f2febb898e
43 changed files with 2941 additions and 444 deletions
@@ -17,17 +17,20 @@ namespace Musify.Infrastructure.MassTransit.Activities.Files
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.FilePathVariable);
ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath));
var fileName = Path.GetFileName(sourceFilePath);
var destinationKey = string.Join('/', new[] { executeContext.Arguments.DestinationRoute, fileName }
.Where(static s => !string.IsNullOrWhiteSpace(s))
.Select(static s => s.Trim().Trim('/', '\\')));
try
{
await storageHandler.UploadFileAsync(
sourceFilePath,
executeContext.Arguments.DestinationBucket,
Path.Combine(executeContext.Arguments.DestinationRoute, fileName),
destinationKey,
executeContext.CancellationToken);
return executeContext.CompletedWithVariables(new UploadFileToBucketLog(
executeContext.Arguments.DestinationBucket,
executeContext.Arguments.DestinationRoute));
destinationKey));
}
catch (Exception exception)
{
@@ -33,11 +33,11 @@ namespace Musify.Infrastructure.MassTransit.Activities.Tracks
try
{
await RemovePicturesAsync(track, executeContext.CancellationToken);
await RemovePicturesAsync(track, executeContext.Arguments.UserId, executeContext.CancellationToken);
track.PicturesProcessingStatus = ProcessingStatus.Completed;
await database.SaveChangesAsync(executeContext.CancellationToken);
await RemoveAudiosAsync(track, executeContext.CancellationToken);
await RemoveAudiosAsync(track, executeContext.Arguments.UserId, executeContext.CancellationToken);
track.AudioTranscodeProcessingStatus = ProcessingStatus.Completed;
await database.SaveChangesAsync(executeContext.CancellationToken);
@@ -66,9 +66,9 @@ namespace Musify.Infrastructure.MassTransit.Activities.Tracks
}
}
async Task RemovePicturesAsync(Track track, CancellationToken cancellationToken)
async Task RemovePicturesAsync(Track track, Guid userId, CancellationToken cancellationToken)
{
await RemoveFileAsync(trackConfiguration.Routes.BuildOriginalPicturePath(track.OriginalPictureName), cancellationToken);
await RemoveFileAsync(trackConfiguration.Routes.BuildOriginalPicturePath(userId, track.OriginalPictureName), cancellationToken);
if (!track.IsPicturesProcessed)
{
@@ -83,9 +83,9 @@ namespace Musify.Infrastructure.MassTransit.Activities.Tracks
await RemoveFileAsync(trackConfiguration.Routes.BuildLargePicturePath(track.LargePictureName), cancellationToken);
}
async Task RemoveAudiosAsync(Track track, CancellationToken cancellationToken)
async Task RemoveAudiosAsync(Track track, Guid userId, CancellationToken cancellationToken)
{
await RemoveFileAsync(trackConfiguration.Routes.BuildOriginalAudioPath(track.OriginalAudioName), cancellationToken);
await RemoveFileAsync(trackConfiguration.Routes.BuildOriginalAudioPath(userId, track.OriginalAudioName), cancellationToken);
if (!track.IsAudioProcessed)
{
@@ -1,5 +1,6 @@
using MassTransit;
using Musify.Infrastructure.MassTransit.Consumers;
using Musify.Infrastructure.Persistence;
namespace Musify.Infrastructure.MassTransit
{
@@ -12,7 +13,18 @@ namespace Musify.Infrastructure.MassTransit
where TConsumer : class, IConsumer
{
busFactoryConfigurator.ReceiveEndpoint(queueName, endpointConfigurator =>
endpointConfigurator.ConfigureConsumer<TConsumer>(busRegistrationContext));
{
endpointConfigurator.UseMessageRetry(retryConfigurator =>
retryConfigurator.Exponential(
retryLimit: 5,
minInterval: TimeSpan.FromSeconds(1),
maxInterval: TimeSpan.FromSeconds(30),
intervalDelta: TimeSpan.FromSeconds(2)));
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.ConfigureConsumer<TConsumer>(busRegistrationContext);
});
}
static void ConfigureExecuteActivityEndpoint<TActivity, TArguments>(
@@ -24,6 +36,15 @@ namespace Musify.Infrastructure.MassTransit
{
busFactoryConfigurator.ReceiveEndpoint(MassTransitEndpointNames.ExecuteQueue(endpointName), endpointConfigurator =>
{
endpointConfigurator.UseMessageRetry(retryConfigurator =>
retryConfigurator.Exponential(
retryLimit: 5,
minInterval: TimeSpan.FromSeconds(1),
maxInterval: TimeSpan.FromSeconds(30),
intervalDelta: TimeSpan.FromSeconds(2)));
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(busRegistrationContext);
});
}
@@ -37,14 +58,34 @@ namespace Musify.Infrastructure.MassTransit
where TLog : class
{
busFactoryConfigurator.ReceiveEndpoint(MassTransitEndpointNames.ExecuteQueue(endpointName), endpointConfigurator =>
{
endpointConfigurator.UseMessageRetry(retryConfigurator =>
retryConfigurator.Exponential(
retryLimit: 5,
minInterval: TimeSpan.FromSeconds(1),
maxInterval: TimeSpan.FromSeconds(30),
intervalDelta: TimeSpan.FromSeconds(2)));
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(
EndpointHelper.BuildCompensateActivityUri(endpointName),
busRegistrationContext)
);
busRegistrationContext);
});
busFactoryConfigurator.ReceiveEndpoint(MassTransitEndpointNames.CompensateQueue(endpointName), endpointConfigurator =>
endpointConfigurator.CompensateActivityHost<TActivity, TLog>(busRegistrationContext)
);
{
endpointConfigurator.UseMessageRetry(retryConfigurator =>
retryConfigurator.Exponential(
retryLimit: 5,
minInterval: TimeSpan.FromSeconds(1),
maxInterval: TimeSpan.FromSeconds(30),
intervalDelta: TimeSpan.FromSeconds(2)));
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
endpointConfigurator.CompensateActivityHost<TActivity, TLog>(busRegistrationContext);
});
}
}
}
@@ -3,6 +3,7 @@ using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Musify.Application.Abstractions.Infrastructure;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.Persistence;
namespace Musify.Infrastructure.MassTransit
{
@@ -18,6 +19,15 @@ namespace Musify.Infrastructure.MassTransit
serviceDescriptors.AddScoped<IEventBus, MassTransitEventBus>();
serviceDescriptors.AddMassTransit(options =>
{
options.AddEntityFrameworkOutbox<Database>(outbox =>
{
outbox.UsePostgres();
outbox.UseBusOutbox();
outbox.QueryDelay = TimeSpan.FromSeconds(1);
outbox.DuplicateDetectionWindow = TimeSpan.FromMinutes(30);
});
options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) =>
ConfigureRabbitMqHost(
busFactoryConfigurator,
@@ -43,6 +53,15 @@ namespace Musify.Infrastructure.MassTransit
serviceDescriptors.AddMassTransit(options =>
{
options.AddEntityFrameworkOutbox<Database>(outbox =>
{
outbox.UsePostgres();
outbox.UseBusOutbox();
outbox.QueryDelay = TimeSpan.FromSeconds(1);
outbox.DuplicateDetectionWindow = TimeSpan.FromMinutes(30);
});
RegisterConsumersAndActivities(options);
options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) =>
@@ -6,7 +6,7 @@ namespace Musify.Infrastructure.MassTransit
public class MassTransitEventBus(IPublishEndpoint publishEndpoint)
: IEventBus
{
public async Task PublishAsync<T>(T message, CancellationToken cancellationToken)
public async Task PublishAsync<T>(T message, CancellationToken cancellationToken) where T : class
{
ArgumentNullException.ThrowIfNull(message, nameof(message));
@@ -99,7 +99,7 @@ namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders
AddResizeActivity(
routingSlipBuilder,
ActivityNames.ResizeMedium,
ActivityNames.ResizeMedium,
ActivityNames.UploadMedium,
bucket,
RoutingSlipVariableNames.Picture.MediumResizedFilePath,
medium);