From 0384de792baf647fcc177809a1f68e19fb560d2e Mon Sep 17 00:00:00 2001 From: NotS1mp Date: Thu, 26 Mar 2026 18:23:59 +0100 Subject: [PATCH] Upload file activity --- .../Files/DownloadFileFromBucketActivity.cs | 5 +- .../Files/UploadFileToBucketActivity.cs | 95 +++++++++++++++++++ .../Arguments/UploadFileToBucketArguments.cs | 8 ++ .../MessagingDependencyInjection.cs | 12 +++ 4 files changed, 119 insertions(+), 1 deletion(-) create mode 100644 Infrastructure/MassTransit/Activities/Files/UploadFileToBucketActivity.cs create mode 100644 Infrastructure/MassTransit/Arguments/UploadFileToBucketArguments.cs diff --git a/Infrastructure/MassTransit/Activities/Files/DownloadFileFromBucketActivity.cs b/Infrastructure/MassTransit/Activities/Files/DownloadFileFromBucketActivity.cs index 8ff1571..901c7d6 100644 --- a/Infrastructure/MassTransit/Activities/Files/DownloadFileFromBucketActivity.cs +++ b/Infrastructure/MassTransit/Activities/Files/DownloadFileFromBucketActivity.cs @@ -65,7 +65,10 @@ namespace Musify.Infrastructure.MassTransit.Activities await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken); await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken); - return executeContext.Completed(); + return executeContext.Completed(new DownloadFileFromBucketLog( + executeContext.Arguments.BucketName, + executeContext.Arguments.KeyName, + destinationPath)); } catch (Exception exception) { diff --git a/Infrastructure/MassTransit/Activities/Files/UploadFileToBucketActivity.cs b/Infrastructure/MassTransit/Activities/Files/UploadFileToBucketActivity.cs new file mode 100644 index 0000000..62356ae --- /dev/null +++ b/Infrastructure/MassTransit/Activities/Files/UploadFileToBucketActivity.cs @@ -0,0 +1,95 @@ +using MassTransit; +using Microsoft.Extensions.Logging; +using Musify.Application.Contracts.Infrastructure; +using Musify.Domain.Entities; +using Musify.Infrastructure.MassTransit.Arguments; +using Musify.Infrastructure.MassTransit.Logs; + +namespace Musify.Infrastructure.MassTransit.Activities.Files +{ + public class UploadFileToBucketActivity(IStorageHandler storageHandler, + ILogger logger, + IProcessTrackingStore processTrackingStore) + : IActivity + { + public const string ExecuteEndpointName = "upload-file-to-bucket"; + + public async Task Execute(ExecuteContext executeContext) + { + var processId = await processTrackingStore.GetOrCreateProcessAsync( + "RoutingSlip", + executeContext.CorrelationId ?? executeContext.TrackingNumber, + executeContext.ConversationId, + executeContext.MessageId, + executeContext.CancellationToken); + + var stepId = await processTrackingStore.StartStepAsync( + processId, + nameof(UploadFileToBucketActivity), + ProcessStepComponentType.Activity, + 0, + executeContext.CancellationToken); + + try + { + using var fileStream = File.OpenRead(executeContext.Arguments.SourceFilePath); + + var uploadResult = await storageHandler.UploadFileAsync( + fileStream, + executeContext.Arguments.ContentType, + executeContext.Arguments.DestinationBucketName, + executeContext.Arguments.DestinationKeyName, + executeContext.CancellationToken); + + if (!uploadResult.IsSuccess) + { + var errors = string.Join("; ", uploadResult.Errors); + logger.LogWarning("Failed to upload file to bucket. Bucket: {BucketName}, Key: {KeyName}, Errors: {Errors}", + executeContext.Arguments.DestinationBucketName, + executeContext.Arguments.DestinationKeyName, + errors); + throw new Exception(errors); + } + + await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken); + return executeContext.Completed(new UploadFileToBucketLog( + executeContext.Arguments.DestinationBucketName, + executeContext.Arguments.DestinationKeyName)); + } + catch (Exception exception) + { + logger.LogError(exception, "An error occurred while uploading file to bucket for process {ProcessId} at step {StepId}", processId, stepId); + await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken); + throw; + } + } + + public async Task Compensate(CompensateContext compensateContext) + { + try + { + var uploadResult = await storageHandler.RemoveFileAsync( + compensateContext.Log.DestinationBucketName, + compensateContext.Log.DestinationKeyName, + compensateContext.CancellationToken); + + if (!uploadResult.IsSuccess) + { + var errors = string.Join("; ", uploadResult.Errors); + var errorMessage = $"Failed to remove file from bucket during compensation. Bucket: {compensateContext.Log.DestinationBucketName}, Key: {compensateContext.Log.DestinationKeyName}, Errors: {errors}"; + + logger.LogError(errorMessage); + return compensateContext.Failed(new Exception(errorMessage)); + } + return compensateContext.Compensated(); + } + catch (Exception exception) + { + logger.LogError(exception, "An error occurred while compensating upload file to bucket activity for bucket {BucketName} and key {KeyName}", + compensateContext.Log.DestinationBucketName, + compensateContext.Log.DestinationKeyName); + return compensateContext.Failed(exception); + } + } + } +} \ No newline at end of file diff --git a/Infrastructure/MassTransit/Arguments/UploadFileToBucketArguments.cs b/Infrastructure/MassTransit/Arguments/UploadFileToBucketArguments.cs new file mode 100644 index 0000000..7f58fcd --- /dev/null +++ b/Infrastructure/MassTransit/Arguments/UploadFileToBucketArguments.cs @@ -0,0 +1,8 @@ +namespace Musify.Infrastructure.MassTransit.Arguments +{ + public record UploadFileToBucketArguments( + string SourceFilePath, + string ContentType, + string DestinationBucketName, + string DestinationKeyName); +} \ No newline at end of file diff --git a/Infrastructure/MassTransit/DependencyInjection/MessagingDependencyInjection.cs b/Infrastructure/MassTransit/DependencyInjection/MessagingDependencyInjection.cs index 7a10f39..af7b44c 100644 --- a/Infrastructure/MassTransit/DependencyInjection/MessagingDependencyInjection.cs +++ b/Infrastructure/MassTransit/DependencyInjection/MessagingDependencyInjection.cs @@ -10,6 +10,7 @@ using Musify.Infrastructure.MassTransit.Activities.Arguments; using Musify.Infrastructure.MassTransit.Activities.Logs; using Musify.Infrastructure.MassTransit.Filters; using Musify.Infrastructure.MassTransit.RoutingSlip.Builders; +using Musify.Infrastructure.MassTransit.Activities.Files; namespace Musify.Infrastructure.MassTransit { @@ -53,6 +54,7 @@ namespace Musify.Infrastructure.MassTransit options.AddActivity(); options.AddActivity(); options.AddActivity(); + options.AddActivity(); options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) => { @@ -131,6 +133,16 @@ namespace Musify.Infrastructure.MassTransit busFactoryConfigurator, busRegistrationContext, TransferFilesToBucketActivity.ExecuteEndpointName); + + ConfigureActivityEndpoint( + busFactoryConfigurator, + busRegistrationContext, + UpdateTrackAudioActivity.ExecuteEndpointName); + + ConfigureActivityEndpoint( + busFactoryConfigurator, + busRegistrationContext, + UploadFileToBucketActivity.ExecuteEndpointName); } ); } );