Upload file activity

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)
             {

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<UploadFileToBucketActivity> logger,
+        IProcessTrackingStore processTrackingStore)
+        : IActivity<UploadFileToBucketArguments, UploadFileToBucketLog>
+    {
+        public const string ExecuteEndpointName = "upload-file-to-bucket";
+
+        public async Task<ExecutionResult> Execute(ExecuteContext<UploadFileToBucketArguments> 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<CompensationResult> Compensate(CompensateContext<UploadFileToBucketLog> 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);
+            }
+        }
+    }
+}

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);
+}

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<TranscodeDashAudioActivity, TranscodeDashAudioArguments, TranscodeDashAudioLog>();
                 options.AddActivity<TransferFilesToBucketActivity, TransferFilesToBucketArguments, TransferFilesToBucketLog>();
                 options.AddActivity<UpdateTrackAudioActivity, UpdateTrackAudioArguments, UpdateTrackAudioLog>();
+                options.AddActivity<UploadFileToBucketActivity, UploadFileToBucketArguments, UploadFileToBucketLog>();
 
                 options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) =>
                 {
@@ -131,6 +133,16 @@ namespace Musify.Infrastructure.MassTransit
                         busFactoryConfigurator,
                         busRegistrationContext,
                         TransferFilesToBucketActivity.ExecuteEndpointName);
+
+                    ConfigureActivityEndpoint<UpdateTrackAudioActivity, UpdateTrackAudioArguments, UpdateTrackAudioLog>(
+                        busFactoryConfigurator,
+                        busRegistrationContext,
+                        UpdateTrackAudioActivity.ExecuteEndpointName);
+
+                    ConfigureActivityEndpoint<UploadFileToBucketActivity, UploadFileToBucketArguments, UploadFileToBucketLog>(
+                        busFactoryConfigurator,
+                        busRegistrationContext,
+                        UploadFileToBucketActivity.ExecuteEndpointName);
                 } );
             } );