1

Infrastructure/MassTransit/Activities/Audio/GenerateAudioWorkflowPathsActivity.cs

@@ -45,7 +45,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
                 return executeContext.CompletedWithVariables(new Dictionary<string, object>
                 {
                     [RoutingSlipVariableNames.Workflow.TemporalDirectory] = workingDirectory,
-                    [RoutingSlipVariableNames.Audio.WorkingDirectory] = workingDirectory,
                     [RoutingSlipVariableNames.Audio.SourceFilePath] = sourceFilePath,
                     [RoutingSlipVariableNames.Audio.TranscodedDirectory] = workingDirectory
                 } );

Infrastructure/MassTransit/Activities/Files/CopyFileInBucketActivity.cs

@@ -0,0 +1,61 @@
+using MassTransit;
+using Microsoft.Extensions.Logging;
+using Musify.Application.Contracts.Infrastructure;
+using Musify.Infrastructure.MassTransit.Arguments;
+using Musify.Infrastructure.MassTransit.Logs;
+
+namespace Musify.Infrastructure.MassTransit.Activities.Files
+{
+    internal class CopyFileInBucketActivity(
+        IStorageService storageHandler,
+        ILogger<CopyFileInBucketActivity> logger)
+        : IActivity<CopyFileInBucketArguments, CopyFileInBucketLog>
+    {
+        public const string ExecuteEndpointName = "copy-file-in-bucket";
+
+        public async Task<ExecutionResult> Execute(ExecuteContext<CopyFileInBucketArguments> executeContext)
+        {
+            try
+            {
+                await storageHandler.CopyFileAsync(
+                    executeContext.Arguments.SourceBucket,
+                    executeContext.Arguments.SourceKey,
+                    executeContext.Arguments.DestinationBucket,
+                    executeContext.Arguments.DestinationKey,
+                    executeContext.CancellationToken);
+
+                return executeContext.Completed(new CopyFileInBucketLog(
+                    executeContext.Arguments.DestinationBucket,
+                    executeContext.Arguments.DestinationKey));
+            }
+            catch (Exception exception)
+            {
+                logger.LogError(exception, "Failed to copy {SourceBucket}/{SourceKey} to {DestinationBucket}/{DestinationKey}",
+                    executeContext.Arguments.SourceBucket,
+                    executeContext.Arguments.SourceKey,
+                    executeContext.Arguments.DestinationBucket,
+                    executeContext.Arguments.DestinationKey);
+                throw;
+            }
+        }
+
+        public async Task<CompensationResult> Compensate(CompensateContext<CopyFileInBucketLog> compensateContext)
+        {
+            try
+            {
+                await storageHandler.RemoveFileAsync(
+                    compensateContext.Log.DestinationBucket,
+                    compensateContext.Log.DestinationKey,
+                    compensateContext.CancellationToken);
+                return compensateContext.Compensated();
+            }
+            catch (Exception exception)
+            {
+                logger.LogError(exception, "Failed to compensate copied file {Bucket}/{Key}",
+                    compensateContext.Log.DestinationBucket,
+                    compensateContext.Log.DestinationKey);
+                return compensateContext.Failed(exception);
+            }
+        }
+    }
+}

Infrastructure/MassTransit/Activities/Files/RemoveFolderFromBucketActivity.cs

@@ -1,42 +0,0 @@
-using MassTransit;
-using Microsoft.Extensions.Logging;
-using Musify.Application.Contracts.Infrastructure;
-using Musify.Infrastructure.MassTransit.Arguments;
-
-namespace Musify.Infrastructure.MassTransit.Activities.Files
-{
-    internal class RemoveFolderFromBucketActivity(
-        IStorageService storageHandler,
-        ILogger<RemoveFolderFromBucketActivity> logger)
-        : IExecuteActivity<RemoveFolderFromBucketArguments>
-    {
-        public const string ExecuteEndpointName = "remove-folder-from-bucket";
-
-        public async Task<ExecutionResult> Execute(ExecuteContext<RemoveFolderFromBucketArguments> executeContext)
-        {
-            var folderKey = executeContext.GetVariable<string>(executeContext.Arguments.FolderKeyVariable);
-            ArgumentException.ThrowIfNullOrEmpty(folderKey, nameof(folderKey));
-
-            try
-            {
-                await storageHandler.RemoveFolderAsync(
-                    executeContext.Arguments.Bucket,
-                    folderKey,
-                    executeContext.CancellationToken);
-
-                logger.LogInformation("Removed folder {FolderKey} from bucket {Bucket}",
-                    folderKey,
-                    executeContext.Arguments.Bucket);
-
-                return executeContext.Completed();
-            }
-            catch (Exception exception)
-            {
-                logger.LogError(exception, "Failed to remove folder {FolderKey} from bucket {Bucket}",
-                    folderKey,
-                    executeContext.Arguments.Bucket);
-                throw;
-            }
-        }
-    }
-}

Infrastructure/MassTransit/Activities/Pictures/GeneratePictureWorkflowPathsActivity.cs

@@ -16,8 +16,7 @@ namespace Musify.Infrastructure.MassTransit.Activities
         {
             try
             {
-                var destinationFolderName = Guid.NewGuid().ToString();
-                var workingDirectory = Path.Combine(executeContext.Arguments.TemporaryRootDirectory, destinationFolderName);
+                var workingDirectory = Path.Combine(executeContext.Arguments.TemporaryRootDirectory, Guid.NewGuid().ToString());
 
                 var sourceFilePath = Path.Combine(
                     workingDirectory, Guid.NewGuid().ToString() + Path.GetExtension(executeContext.Arguments.SourceKey));
@@ -38,8 +37,7 @@ namespace Musify.Infrastructure.MassTransit.Activities
                     [RoutingSlipVariableNames.Picture.OriginalFilePath] = sourceFilePath,
                     [RoutingSlipVariableNames.Picture.SmallResizedFilePath] = smallPictureFilePath,
                     [RoutingSlipVariableNames.Picture.MediumResizedFilePath] = mediumPictureFilePath,
-                    [RoutingSlipVariableNames.Picture.LargeResizedFilePath] = largePictureFilePath,
-                    [RoutingSlipVariableNames.Picture.DestinationFolderName] = destinationFolderName
+                    [RoutingSlipVariableNames.Picture.LargeResizedFilePath] = largePictureFilePath
                 } );
             }
             catch (Exception exception)

Infrastructure/MassTransit/Activities/PlayLists/PublishPlayListPictureProcessingEventActivity.cs

@@ -0,0 +1,34 @@
+using MassTransit;
+using Microsoft.Extensions.Logging;
+using Musify.Application.Events;
+using Musify.Infrastructure.MassTransit.Arguments;
+
+namespace Musify.Infrastructure.MassTransit.Activities.PlayLists
+{
+    internal class PublishPlayListPictureProcessingEventActivity(
+        IPublishEndpoint publishEndpoint,
+        ILogger<PublishPlayListPictureProcessingEventActivity> logger)
+        : IExecuteActivity<PublishPlayListPictureProcessingEventArguments>
+    {
+        public const string ExecuteEndpointName = "publish-playList-picture-processing-event";
+
+        public async Task<ExecutionResult> Execute(ExecuteContext<PublishPlayListPictureProcessingEventArguments> executeContext)
+        {
+            var arguments = executeContext.Arguments;
+
+            await publishEndpoint.Publish(
+                new UpdatePlayListPictureEvent(
+                    arguments.PlayListId,
+                    arguments.Bucket,
+                    arguments.FinalPictureKey,
+                    arguments.Small,
+                    arguments.Medium,
+                    arguments.Large),
+                executeContext.CancellationToken);
+
+            logger.LogInformation("Published picture processing event for playlist {PlayListId}", arguments.PlayListId);
+
+            return executeContext.Completed();
+        }
+    }
+}

Infrastructure/MassTransit/Activities/Tracks/PublishTrackProcessingEventsActivity.cs

@@ -0,0 +1,43 @@
+using MassTransit;
+using Microsoft.Extensions.Logging;
+using Musify.Application.Events;
+using Musify.Infrastructure.MassTransit.Arguments;
+
+namespace Musify.Infrastructure.MassTransit.Activities.Tracks
+{
+    internal class PublishTrackProcessingEventsActivity(
+        IPublishEndpoint publishEndpoint,
+        ILogger<PublishTrackProcessingEventsActivity> logger)
+        : IExecuteActivity<PublishTrackProcessingEventsArguments>
+    {
+        public const string ExecuteEndpointName = "publish-track-processing-events";
+
+        public async Task<ExecutionResult> Execute(ExecuteContext<PublishTrackProcessingEventsArguments> executeContext)
+        {
+            var arguments = executeContext.Arguments;
+
+            await publishEndpoint.Publish(
+                new UpdateTrackPictureEvent(
+                    arguments.TrackId,
+                    arguments.Bucket,
+                    arguments.FinalPictureKey,
+                    arguments.Small,
+                    arguments.Medium,
+                    arguments.Large),
+                executeContext.CancellationToken);
+
+            await publishEndpoint.Publish(
+                new UpdateTrackAudioEvent(
+                    arguments.TrackId,
+                    arguments.Bucket,
+                    arguments.FinalAudioKey,
+                    arguments.Bucket,
+                    arguments.AudioDestinationFolderKey),
+                executeContext.CancellationToken);
+
+            logger.LogInformation("Published processing events for track {TrackId}", arguments.TrackId);
+
+            return executeContext.Completed();
+        }
+    }
+}

Infrastructure/MassTransit/Activities/UploadIntents/ConsumeUploadIntentsActivity.cs

@@ -0,0 +1,62 @@
+using MassTransit;
+using Microsoft.EntityFrameworkCore;
+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.UploadIntents
+{
+    internal class ConsumeUploadIntentsActivity(
+        IDatabase database,
+        ILogger<ConsumeUploadIntentsActivity> logger)
+        : IActivity<ConsumeUploadIntentsArguments, ConsumeUploadIntentsLog>
+    {
+        public const string ExecuteEndpointName = "consume-upload-intents";
+
+        public async Task<ExecutionResult> Execute(ExecuteContext<ConsumeUploadIntentsArguments> executeContext)
+        {
+            var intentIds = executeContext.Arguments.IntentIds;
+
+            var intents = await database.UploadIntents
+                .Where(i => intentIds.Contains(i.Id))
+                .ToListAsync(executeContext.CancellationToken);
+
+            foreach (var intent in intents)
+                intent.Status = UploadIntentStatus.Consumed;
+
+            await database.SaveChangesAsync(executeContext.CancellationToken);
+
+            logger.LogInformation("Marked {Count} upload intent(s) as consumed", intents.Count);
+
+            return executeContext.Completed(new ConsumeUploadIntentsLog(intentIds));
+        }
+
+        public async Task<CompensationResult> Compensate(CompensateContext<ConsumeUploadIntentsLog> compensateContext)
+        {
+            try
+            {
+                var intentIds = compensateContext.Log.IntentIds;
+
+                var intents = await database.UploadIntents
+                    .Where(i => intentIds.Contains(i.Id))
+                    .ToListAsync(compensateContext.CancellationToken);
+
+                foreach (var intent in intents)
+                    intent.Status = UploadIntentStatus.Issued;
+
+                await database.SaveChangesAsync(compensateContext.CancellationToken);
+
+                logger.LogInformation("Reverted {Count} upload intent(s) back to Issued", intents.Count);
+
+                return compensateContext.Compensated();
+            }
+            catch (Exception exception)
+            {
+                logger.LogError(exception, "Failed to compensate upload intent consumption");
+                return compensateContext.Failed(exception);
+            }
+        }
+    }
+}