From 4ec4b50c407d177d8934b2ca1f0810b9bcfc5dc1 Mon Sep 17 00:00:00 2001 From: XShine555 Date: Mon, 23 Mar 2026 03:13:08 +0100 Subject: [PATCH] Changes but failing --- Application/Events/UpdateTrackPictureEvent.cs | 4 +- .../Handler/CreateTrackCommandHandler.cs | 16 ++-- .../Arguments/ResizePictureLocalArguments.cs | 8 ++ .../DownloadFileFromBucketActivity.cs | 75 ++++++++++++----- .../RemoveFileFromBucketActivity.cs | 8 +- .../Activities/ResizePictureActivity.cs | 54 ++++++------ .../Activities/TranscodeDashAudioActivity.cs | 40 ++++++--- .../Activities/TransferFilesToBucket.cs | 58 ++++++++++--- .../UpdatePlayListPictureActivity.cs | 21 ++++- .../Activities/UpdateTrackPictureActivity.cs | 21 ++++- .../Messaging/Consumers/MessagingHelper.cs | 7 ++ .../Messaging/Consumers/RemoveFileConsumer.cs | 4 +- .../TranscodeAudioFromTrackConsumer.cs | 28 +++---- .../UpdatePlayListPictureConsumer.cs | 59 +++++++++---- .../Consumers/UpdateTrackPictureConsumer.cs | 82 +++++++++++++------ .../Messaging/MessagingDependencyInjection.cs | 37 +++++---- 16 files changed, 362 insertions(+), 160 deletions(-) create mode 100644 Infrastructure/Messaging/Activities/Arguments/ResizePictureLocalArguments.cs create mode 100644 Infrastructure/Messaging/Consumers/MessagingHelper.cs diff --git a/Application/Events/UpdateTrackPictureEvent.cs b/Application/Events/UpdateTrackPictureEvent.cs index ada8d60..1ba00be 100644 --- a/Application/Events/UpdateTrackPictureEvent.cs +++ b/Application/Events/UpdateTrackPictureEvent.cs @@ -2,8 +2,8 @@ { public record UpdateTrackPictureEvent( Guid TrackId, - string BucketName, - string OriginalPictureKeyName, + string SourceBucketName, + string SourceKeyName, string SmallPictureKeyName, int SmallPictureWidth, int SmallPictureHeight, diff --git a/Application/Tracks/Handler/CreateTrackCommandHandler.cs b/Application/Tracks/Handler/CreateTrackCommandHandler.cs index eeed857..6b0b7b1 100644 --- a/Application/Tracks/Handler/CreateTrackCommandHandler.cs +++ b/Application/Tracks/Handler/CreateTrackCommandHandler.cs @@ -42,6 +42,14 @@ namespace Musify.Application.Tracks.Handler return Result.Error($"Failed to upload picture for track {request.Title}"); } + var audioKeyName = Path.Combine(trackConfiguration.Routes.Audios, $"{Guid.NewGuid() }.{request.Audio.FileType}"); + var audioUploadResult = await UploadFile(request.Audio, audioKeyName, cancellationToken); + if (!audioUploadResult.IsSuccess) + { + logger.LogError("Failed to upload audio for track {TrackTitle}", request.Title); + return Result.Error($"Failed to upload audio for track {request.Title}"); + } + try { await PublishUpdateEvent(originalPictureKeyName, track, cancellationToken); @@ -52,14 +60,6 @@ namespace Musify.Application.Tracks.Handler return Result.Error($"Failed to publish resize picture event for Track with id {track.Id}."); } - var audioKeyName = Path.Combine(trackConfiguration.Routes.Audios, $"{Guid.NewGuid() }.{request.Audio.FileType}"); - var audioUploadResult = await UploadFile(request.Audio, audioKeyName, cancellationToken); - if (!audioUploadResult.IsSuccess) - { - logger.LogError("Failed to upload audio for track {TrackTitle}", request.Title); - return Result.Error($"Failed to upload audio for track {request.Title}"); - } - try { await PublishTranscodeEvent(track.Id, audioUploadResult, cancellationToken); diff --git a/Infrastructure/Messaging/Activities/Arguments/ResizePictureLocalArguments.cs b/Infrastructure/Messaging/Activities/Arguments/ResizePictureLocalArguments.cs new file mode 100644 index 0000000..3e7ed42 --- /dev/null +++ b/Infrastructure/Messaging/Activities/Arguments/ResizePictureLocalArguments.cs @@ -0,0 +1,8 @@ +namespace Musify.Infrastructure.Messaging.Activities.Arguments +{ + public record ResizePictureLocalArguments( + string SourceFilePath, + string DestinationFilePath, + int Width, + int Height); +} diff --git a/Infrastructure/Messaging/Activities/DownloadFileFromBucketActivity.cs b/Infrastructure/Messaging/Activities/DownloadFileFromBucketActivity.cs index 37bf996..b0f33b3 100644 --- a/Infrastructure/Messaging/Activities/DownloadFileFromBucketActivity.cs +++ b/Infrastructure/Messaging/Activities/DownloadFileFromBucketActivity.cs @@ -1,41 +1,76 @@ using MassTransit; using Microsoft.Extensions.Logging; using Musify.Application.Contracts.Infrastructure; +using Musify.Domain.Entities; +using Musify.Infrastructure.Configuration; using Musify.Infrastructure.Messaging.Activities.Arguments; -using Musify.Infrastructure.Messaging.Activities.Logs; namespace Musify.Infrastructure.Messaging.Activities { - public class DownloadFileFromBucketActivity(IStorageHandler storageHandler, ILogger logger) + public class DownloadFileFromBucketActivity( + IStorageHandler storageHandler, + ILogger logger, + IProcessTrackingStore processTrackingStore, + AudioTranscoderConfiguration audioTranscoderConfiguration) : IExecuteActivity { public const string ExecuteEndpointName = "Download-File-From-Bucket"; public async Task Execute(ExecuteContext executeContext) { - var getFile = await storageHandler.GetFileAsync( - executeContext.Arguments.BucketName, - executeContext.Arguments.KeyName, + var processId = await processTrackingStore.GetOrCreateProcessAsync( + "RoutingSlip", + executeContext.CorrelationId ?? executeContext.TrackingNumber, + executeContext.ConversationId, + executeContext.MessageId, executeContext.CancellationToken); - if (!getFile.IsSuccess) + var stepId = await processTrackingStore.StartStepAsync( + processId, + nameof(DownloadFileFromBucketActivity), + ProcessStepComponentType.Activity, + 0, + executeContext.CancellationToken); + + try { - var error = string.Join("; ", getFile.Errors); - logger.LogError("Failed to download file from bucket. Bucket: {BucketName}, Key: {KeyName}", + var getFile = await storageHandler.GetFileAsync( executeContext.Arguments.BucketName, - executeContext.Arguments.KeyName); - throw new Exception(error); + executeContext.Arguments.KeyName, + executeContext.CancellationToken); + + if (!getFile.IsSuccess) + { + var errorMessage = string.Join("; ", getFile.Errors); + logger.LogError("Failed to download file from bucket. Bucket: {BucketName}, Key: {KeyName}, Errors: {Errors}", + executeContext.Arguments.BucketName, + executeContext.Arguments.KeyName, + errorMessage); + throw new Exception(errorMessage); + } + + using var fileStream = getFile.Value; + if (fileStream.CanSeek) + fileStream.Position = 0; + + var destinationPath = Path.Combine( + audioTranscoderConfiguration.Routes.WorkingDirectory, + Path.GetFileName(executeContext.Arguments.KeyName)); + + Directory.CreateDirectory(Path.GetDirectoryName(destinationPath)!); + + using var destinationStream = File.Create(destinationPath); + await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken); + + await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken); + return executeContext.Completed(); + } + catch (Exception exception) + { + logger.LogError(exception, "Error downloading file from bucket"); + await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken); + throw; } - - using var fileStream = getFile.Value; - if (fileStream.CanSeek) - fileStream.Position = 0; - - var fileName = Path.GetFileName(executeContext.Arguments.KeyName); - using var destinationStream = File.Create($"D:\\tempsFilesDev\\{fileName}"); - await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken); - - return executeContext.Completed(); } } } \ No newline at end of file diff --git a/Infrastructure/Messaging/Activities/RemoveFileFromBucketActivity.cs b/Infrastructure/Messaging/Activities/RemoveFileFromBucketActivity.cs index 185260e..28ff4f4 100644 --- a/Infrastructure/Messaging/Activities/RemoveFileFromBucketActivity.cs +++ b/Infrastructure/Messaging/Activities/RemoveFileFromBucketActivity.cs @@ -39,9 +39,8 @@ namespace Musify.Infrastructure.Messaging.Activities if (!removeFile.IsSuccess) { - string errorMessage = string.Join(";", removeFile.Errors); - logger.LogWarning( - "Failed to remove file {KeyName} from bucket {BucketName}: {ErrorMessage}", + var errorMessage = string.Join("; ", removeFile.Errors); + logger.LogWarning("Failed to remove file {KeyName} from bucket {BucketName}. Errors: {Errors}", executeContext.Arguments.KeyName, executeContext.Arguments.BucketName, errorMessage); @@ -52,6 +51,9 @@ namespace Musify.Infrastructure.Messaging.Activities } catch (Exception exception) { + logger.LogError(exception, "Error removing file {KeyName} from bucket {BucketName}", + executeContext.Arguments.KeyName, + executeContext.Arguments.BucketName); await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken); throw; } diff --git a/Infrastructure/Messaging/Activities/ResizePictureActivity.cs b/Infrastructure/Messaging/Activities/ResizePictureActivity.cs index d28292e..8e0d0e6 100644 --- a/Infrastructure/Messaging/Activities/ResizePictureActivity.cs +++ b/Infrastructure/Messaging/Activities/ResizePictureActivity.cs @@ -1,4 +1,5 @@ using MassTransit; +using Microsoft.Extensions.Logging; using Musify.Application.Contracts.Infrastructure; using Musify.Domain.Entities; using Musify.Infrastructure.Messaging.Activities.Arguments; @@ -6,14 +7,14 @@ using Musify.Infrastructure.Messaging.Activities.Arguments; namespace Musify.Infrastructure.Messaging.Activities { public class ResizePictureActivity( - IStorageHandler storageHandler, IPictureHandler pictureHandler, + ILogger logger, IProcessTrackingStore processTrackingStore) - : IExecuteActivity + : IExecuteActivity { public const string ExecuteEndpointName = "Resize-Picture"; - public async Task Execute(ExecuteContext executeContext) + public async Task Execute(ExecuteContext executeContext) { var processId = await processTrackingStore.GetOrCreateProcessAsync( "RoutingSlip", @@ -31,47 +32,54 @@ namespace Musify.Infrastructure.Messaging.Activities try { - var file = await storageHandler.GetFileAsync( - executeContext.Arguments.OriginalBucketName, - executeContext.Arguments.OriginalKeyName, - executeContext.CancellationToken); - - if (!file.IsSuccess) + if (!File.Exists(executeContext.Arguments.SourceFilePath)) { - string errorMessages = string.Join(", ", file.Errors); - throw new Exception(errorMessages); + logger.LogError("Source file not found at {SourceFilePath}", + executeContext.Arguments.SourceFilePath); + throw new FileNotFoundException($"Source file not found: {executeContext.Arguments.SourceFilePath}"); } + await using var fileStream = File.OpenRead(executeContext.Arguments.SourceFilePath); + var resizedPicture = await pictureHandler.ResizePictureAsync( - file.Value, + fileStream, executeContext.Arguments.Width, executeContext.Arguments.Height, executeContext.CancellationToken); if (!resizedPicture.IsSuccess) { - string errorMessage = string.Join(", ", resizedPicture.Errors); + var errorMessage = string.Join("; ", resizedPicture.Errors); + logger.LogError("Failed to resize picture {SourceFilePath} to {Width}x{Height}. Errors: {Errors}", + executeContext.Arguments.SourceFilePath, + executeContext.Arguments.Width, + executeContext.Arguments.Height, + errorMessage); throw new Exception(errorMessage); } - var uploadFile = await storageHandler.UploadFileAsync( - resizedPicture.Value, - pictureHandler.ContentType, - executeContext.Arguments.DestinationBucketName, - executeContext.Arguments.DestinationKeyName, - executeContext.CancellationToken); - - if (!uploadFile.IsSuccess) + var destinationDirectory = Path.GetDirectoryName(executeContext.Arguments.DestinationFilePath); + if (!string.IsNullOrEmpty(destinationDirectory)) { - var errorMessage = string.Join(", ", uploadFile.Errors); - throw new Exception(errorMessage); + Directory.CreateDirectory(destinationDirectory); } + await using var destinationStream = File.Create(executeContext.Arguments.DestinationFilePath); + await resizedPicture.Value.CopyToAsync(destinationStream, executeContext.CancellationToken); + + logger.LogInformation("Picture resized successfully from {SourceFilePath} to {DestinationFilePath} ( {Width}x{Height} )", + executeContext.Arguments.SourceFilePath, + executeContext.Arguments.DestinationFilePath, + executeContext.Arguments.Width, + executeContext.Arguments.Height); + await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken); return executeContext.Completed(); } catch (Exception exception) { + logger.LogError(exception, "Error resizing picture {SourceFilePath}", + executeContext.Arguments.SourceFilePath); await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken); throw; } diff --git a/Infrastructure/Messaging/Activities/TranscodeDashAudioActivity.cs b/Infrastructure/Messaging/Activities/TranscodeDashAudioActivity.cs index 659afbb..ae8ba5e 100644 --- a/Infrastructure/Messaging/Activities/TranscodeDashAudioActivity.cs +++ b/Infrastructure/Messaging/Activities/TranscodeDashAudioActivity.cs @@ -1,13 +1,16 @@ using MassTransit; using Microsoft.Extensions.Logging; using Musify.Application.Contracts.Infrastructure; +using Musify.Domain.Entities; using Musify.Infrastructure.Configuration; using Musify.Infrastructure.Messaging.Activities.Arguments; -using Musify.Infrastructure.Messaging.Activities.Logs; namespace Musify.Infrastructure.Messaging.Activities { - public class TranscodeDashAudioActivity(IAudioTranscoder audioTranscoder, ILogger logger, + public class TranscodeDashAudioActivity( + IAudioTranscoder audioTranscoder, + ILogger logger, + IProcessTrackingStore processTrackingStore, AudioTranscoderConfiguration audioTranscoderConfiguration) : IExecuteActivity { @@ -15,12 +18,26 @@ namespace Musify.Infrastructure.Messaging.Activities public async Task Execute(ExecuteContext executeContext) { - var workingDirectory = Path.Combine( - audioTranscoderConfiguration.Routes.WorkingDirectory, - executeContext.Arguments.DestinationFolderName); + var processId = await processTrackingStore.GetOrCreateProcessAsync( + "RoutingSlip", + executeContext.CorrelationId ?? executeContext.TrackingNumber, + executeContext.ConversationId, + executeContext.MessageId, + executeContext.CancellationToken); + + var stepId = await processTrackingStore.StartStepAsync( + processId, + nameof(TranscodeDashAudioActivity), + ProcessStepComponentType.Activity, + 0, + executeContext.CancellationToken); try { + var workingDirectory = Path.Combine( + audioTranscoderConfiguration.Routes.WorkingDirectory, + executeContext.Arguments.DestinationFolderName); + await using var fileStream = File.OpenRead(executeContext.Arguments.SourceFilePath); var result = await audioTranscoder.TranscodeToDashAsync( @@ -31,17 +48,20 @@ namespace Musify.Infrastructure.Messaging.Activities if (!result.IsSuccess) { - var errors = string.Join("; ", result.Errors); - logger.LogError("transcoding failed for file {SourceFilePath}. Error: {ErrorMessage}", - executeContext.Arguments.SourceFilePath, errors); - throw new Exception(errors); + var errorMessage = string.Join("; ", result.Errors); + logger.LogError("Transcoding failed for file {SourceFilePath}. Errors: {ErrorMessage}", + executeContext.Arguments.SourceFilePath, errorMessage); + throw new Exception(errorMessage); } + await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken); return executeContext.Completed(); } catch (Exception exception) { - logger.LogError(exception, "Error during transcoding activity."); + logger.LogError(exception, "Error during transcoding activity for file {SourceFilePath}", + executeContext.Arguments.SourceFilePath); + await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken); throw; } } diff --git a/Infrastructure/Messaging/Activities/TransferFilesToBucket.cs b/Infrastructure/Messaging/Activities/TransferFilesToBucket.cs index 76c97a8..b3f1ea9 100644 --- a/Infrastructure/Messaging/Activities/TransferFilesToBucket.cs +++ b/Infrastructure/Messaging/Activities/TransferFilesToBucket.cs @@ -1,31 +1,65 @@ using MassTransit; using Microsoft.Extensions.Logging; using Musify.Application.Contracts.Infrastructure; +using Musify.Domain.Entities; using Musify.Infrastructure.Messaging.Activities.Arguments; namespace Musify.Infrastructure.Messaging.Activities { - public class TransferFilesToBucket(IStorageHandler storageHandler, ILogger logger) + public class TransferFilesToBucket( + IStorageHandler storageHandler, + ILogger logger, + IProcessTrackingStore processTrackingStore) : IExecuteActivity { public const string ExecuteEndpointName = "Transfer-Files-To-Bucket"; public async Task Execute(ExecuteContext executeContext) { - var transferFilesResult = await storageHandler.TransferFilesAsync( - executeContext.Arguments.FolderPath, - executeContext.Arguments.DestinationBucketName, - executeContext.Arguments.DestinationKeyName, + var processId = await processTrackingStore.GetOrCreateProcessAsync( + "RoutingSlip", + executeContext.CorrelationId ?? executeContext.TrackingNumber, + executeContext.ConversationId, + executeContext.MessageId, executeContext.CancellationToken); - if (!transferFilesResult.IsSuccess) - { - var errors = string.Join("; ", transferFilesResult.Errors); - logger.LogError($"An error occurred while transferring files to bucket: {errors}"); - throw new Exception(errors); - } + var stepId = await processTrackingStore.StartStepAsync( + processId, + nameof(TransferFilesToBucket), + ProcessStepComponentType.Activity, + 0, + executeContext.CancellationToken); - return executeContext.Completed(); + try + { + var transferFilesResult = await storageHandler.TransferFilesAsync( + executeContext.Arguments.FolderPath, + executeContext.Arguments.DestinationBucketName, + executeContext.Arguments.DestinationKeyName, + executeContext.CancellationToken); + + if (!transferFilesResult.IsSuccess) + { + var errorMessage = string.Join("; ", transferFilesResult.Errors); + logger.LogError("Failed to transfer files from {FolderPath} to bucket {DestinationBucketName}, key {DestinationKeyName}. Errors: {Errors}", + executeContext.Arguments.FolderPath, + executeContext.Arguments.DestinationBucketName, + executeContext.Arguments.DestinationKeyName, + errorMessage); + throw new Exception(errorMessage); + } + + await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken); + return executeContext.Completed(); + } + catch (Exception exception) + { + logger.LogError(exception, "Error transferring files from {FolderPath} to bucket {DestinationBucketName}", + executeContext.Arguments.FolderPath, + executeContext.Arguments.DestinationBucketName); + await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken); + throw; + } } } } \ No newline at end of file diff --git a/Infrastructure/Messaging/Activities/UpdatePlayListPictureActivity.cs b/Infrastructure/Messaging/Activities/UpdatePlayListPictureActivity.cs index cbd4c1b..82ae1c3 100644 --- a/Infrastructure/Messaging/Activities/UpdatePlayListPictureActivity.cs +++ b/Infrastructure/Messaging/Activities/UpdatePlayListPictureActivity.cs @@ -1,11 +1,15 @@ using MassTransit; +using Microsoft.Extensions.Logging; using Musify.Application.Contracts.Infrastructure; using Musify.Domain.Entities; using Musify.Infrastructure.Messaging.Activities.Arguments; namespace Musify.Infrastructure.Messaging.Activities { - public class UpdatePlayListPictureActivity(IDatabase database, IProcessTrackingStore processTrackingStore) + public class UpdatePlayListPictureActivity( + IDatabase database, + ILogger logger, + IProcessTrackingStore processTrackingStore) : IExecuteActivity { public const string ExecuteEndpointName = "Update-PlayList-Picture"; @@ -28,10 +32,16 @@ namespace Musify.Infrastructure.Messaging.Activities try { - var playList = await database.PlayLists.FindAsync(executeContext.Arguments.PlayListId, executeContext.CancellationToken); + var playList = await database.PlayLists.FindAsync( + new object[] { executeContext.Arguments.PlayListId }, + cancellationToken: executeContext.CancellationToken); if (playList is null) - throw new Exception($"PlayList with id {executeContext.Arguments.PlayListId} not found."); + { + logger.LogError("PlayList with id {PlayListId} not found", + executeContext.Arguments.PlayListId); + throw new InvalidOperationException($"PlayList with id {executeContext.Arguments.PlayListId} not found"); + } playList.SmallPictureKeyName = executeContext.Arguments.SmallPictureKeyName; playList.MediumPictureKeyName = executeContext.Arguments.MediumPictureKeyName; @@ -40,11 +50,16 @@ namespace Musify.Infrastructure.Messaging.Activities database.PlayLists.Update(playList); await database.SaveChangesAsync(executeContext.CancellationToken); + logger.LogInformation("Updated PlayList {PlayListId} pictures", + executeContext.Arguments.PlayListId); + await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken); return executeContext.Completed(); } catch (Exception exception) { + logger.LogError(exception, "Error updating PlayList {PlayListId} pictures", + executeContext.Arguments.PlayListId); await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken); throw; } diff --git a/Infrastructure/Messaging/Activities/UpdateTrackPictureActivity.cs b/Infrastructure/Messaging/Activities/UpdateTrackPictureActivity.cs index 63671de..103407e 100644 --- a/Infrastructure/Messaging/Activities/UpdateTrackPictureActivity.cs +++ b/Infrastructure/Messaging/Activities/UpdateTrackPictureActivity.cs @@ -1,11 +1,15 @@ using MassTransit; +using Microsoft.Extensions.Logging; using Musify.Application.Contracts.Infrastructure; using Musify.Domain.Entities; using Musify.Infrastructure.Messaging.Activities.Arguments; namespace Musify.Infrastructure.Messaging.Activities { - public class UpdateTrackPictureActivity(IDatabase database, IProcessTrackingStore processTrackingStore) + public class UpdateTrackPictureActivity( + IDatabase database, + ILogger logger, + IProcessTrackingStore processTrackingStore) : IExecuteActivity { public const string ExecuteEndpointName = "Update-Track-Picture"; @@ -28,10 +32,16 @@ namespace Musify.Infrastructure.Messaging.Activities try { - var track = await database.Tracks.FindAsync(executeContext.Arguments.TrackId, executeContext.CancellationToken); + var track = await database.Tracks.FindAsync( + new object[] { executeContext.Arguments.TrackId }, + cancellationToken: executeContext.CancellationToken); if (track is null) - throw new Exception($"Track with id {executeContext.Arguments.TrackId} not found."); + { + logger.LogError("Track with id {TrackId} not found", + executeContext.Arguments.TrackId); + throw new InvalidOperationException($"Track with id {executeContext.Arguments.TrackId} not found"); + } track.SmallPictureKeyName = executeContext.Arguments.SmallPictureKeyName; track.MediumPictureKeyName = executeContext.Arguments.MediumPictureKeyName; @@ -40,11 +50,16 @@ namespace Musify.Infrastructure.Messaging.Activities database.Tracks.Update(track); await database.SaveChangesAsync(executeContext.CancellationToken); + logger.LogInformation("Updated Track {TrackId} pictures", + executeContext.Arguments.TrackId); + await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken); return executeContext.Completed(); } catch (Exception exception) { + logger.LogError(exception, "Error updating Track {TrackId} pictures", + executeContext.Arguments.TrackId); await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken); throw; } diff --git a/Infrastructure/Messaging/Consumers/MessagingHelper.cs b/Infrastructure/Messaging/Consumers/MessagingHelper.cs new file mode 100644 index 0000000..5fd9cbb --- /dev/null +++ b/Infrastructure/Messaging/Consumers/MessagingHelper.cs @@ -0,0 +1,7 @@ +namespace Musify.Infrastructure.Messaging.Consumers +{ + internal static class MessagingHelper + { + internal static Uri BuildExecuteActivityUri(string endpointName) => new($"queue:{endpointName}_execute"); + } +} \ No newline at end of file diff --git a/Infrastructure/Messaging/Consumers/RemoveFileConsumer.cs b/Infrastructure/Messaging/Consumers/RemoveFileConsumer.cs index 79a2650..8a78a1a 100644 --- a/Infrastructure/Messaging/Consumers/RemoveFileConsumer.cs +++ b/Infrastructure/Messaging/Consumers/RemoveFileConsumer.cs @@ -16,7 +16,7 @@ namespace Musify.Infrastructure.Messaging.Consumers routingSlipBuilder.AddActivity( "RemoveFile", - BuildExecuteUri(RemoveFileFromBucketActivity.ExecuteEndpointName), + MessagingHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName), new RemoveFileFromBucketArguments( consumeContext.Message.BucketName, consumeContext.Message.KeyName)); @@ -24,7 +24,5 @@ namespace Musify.Infrastructure.Messaging.Consumers var routingSlip = routingSlipBuilder.Build(); await bus.Execute(routingSlip); } - - static Uri BuildExecuteUri(string endpointName) => new($"queue:{endpointName}_execute"); } } diff --git a/Infrastructure/Messaging/Consumers/TranscodeAudioFromTrackConsumer.cs b/Infrastructure/Messaging/Consumers/TranscodeAudioFromTrackConsumer.cs index a50204b..41bfc49 100644 --- a/Infrastructure/Messaging/Consumers/TranscodeAudioFromTrackConsumer.cs +++ b/Infrastructure/Messaging/Consumers/TranscodeAudioFromTrackConsumer.cs @@ -6,7 +6,9 @@ using Musify.Infrastructure.Messaging.Activities.Arguments; namespace Musify.Infrastructure.Messaging.Consumers { - public class TranscodeAudioFromTrackConsumer(IBus bus, AudioTranscoderConfiguration audioTranscoderConfiguration) + public class TranscodeAudioFromTrackConsumer( + IBus bus, + AudioTranscoderConfiguration audioTranscoderConfiguration) : IConsumer { public const string QueueName = "Transcode-Audio-From-Track"; @@ -14,41 +16,35 @@ namespace Musify.Infrastructure.Messaging.Consumers public async Task Consume(ConsumeContext consumeContext) { var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid()); - + var fileName = Path.GetFileName(consumeContext.Message.SourceKeyName); var folderName = Guid.NewGuid().ToString(); - var workingDirectory = Path.Combine("D:\\tempsFilesDev\\"); + var workingDirectory = Path.Combine(audioTranscoderConfiguration.Routes.WorkingDirectory, folderName); routingSlipBuilder.AddActivity( "DownloadFile", - BuildExecuteUri(DownloadFileFromBucketActivity.ExecuteEndpointName), + MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName), new DownloadFileFromBucketArguments( consumeContext.Message.SourceBucketName, consumeContext.Message.SourceKeyName, - workingDirectory) - ); - var fileName = Path.GetFileName(consumeContext.Message.SourceKeyName); + workingDirectory)); routingSlipBuilder.AddActivity( "TranscodeAudio", - BuildExecuteUri(TranscodeDashAudioActivity.ExecuteEndpointName), + MessagingHelper.BuildExecuteActivityUri(TranscodeDashAudioActivity.ExecuteEndpointName), new TranscodeDashAudioArguments( - $"D:\\tempsFilesDev\\{fileName}", - "D:\\tempsFilesDev\\") - ); + Path.Combine(workingDirectory, fileName), + folderName)); routingSlipBuilder.AddActivity( "UploadFile", - BuildExecuteUri(TransferFilesToBucket.ExecuteEndpointName), + MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName), new TransferFilesToBucketArguments( workingDirectory, consumeContext.Message.DestinationBucketName, - consumeContext.Message.DestinationKeyName) - ); + consumeContext.Message.DestinationKeyName)); var routingSlip = routingSlipBuilder.Build(); await bus.Execute(routingSlip); } - - static Uri BuildExecuteUri(string endpointName) => new($"queue:{endpointName}_execute"); } } \ No newline at end of file diff --git a/Infrastructure/Messaging/Consumers/UpdatePlayListPictureConsumer.cs b/Infrastructure/Messaging/Consumers/UpdatePlayListPictureConsumer.cs index b9271a1..539b2aa 100644 --- a/Infrastructure/Messaging/Consumers/UpdatePlayListPictureConsumer.cs +++ b/Infrastructure/Messaging/Consumers/UpdatePlayListPictureConsumer.cs @@ -1,11 +1,14 @@ using MassTransit; using Musify.Application.Events; +using Musify.Infrastructure.Configuration; using Musify.Infrastructure.Messaging.Activities; using Musify.Infrastructure.Messaging.Activities.Arguments; namespace Musify.Infrastructure.Messaging.Consumers { - public class UpdatePlayListPictureConsumer(IBus bus) + public class UpdatePlayListPictureConsumer( + IBus bus, + AudioTranscoderConfiguration audioTranscoderConfiguration) : IConsumer { public const string QueueName = "Update-PlayList-Picture"; @@ -13,6 +16,8 @@ namespace Musify.Infrastructure.Messaging.Consumers public async Task Consume(ConsumeContext consumeContext) { var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid()); + var folderName = Guid.NewGuid().ToString(); + var workingDirectory = Path.Combine(audioTranscoderConfiguration.Routes.WorkingDirectory, folderName); AddResizeActivity( routingSlipBuilder, @@ -21,7 +26,8 @@ namespace Musify.Infrastructure.Messaging.Consumers consumeContext.Message.SmallPictureWidth, consumeContext.Message.SmallPictureHeight, consumeContext.Message.OriginalPictureKeyName, - consumeContext.Message.BucketName); + consumeContext.Message.BucketName, + workingDirectory); AddResizeActivity( routingSlipBuilder, @@ -30,7 +36,8 @@ namespace Musify.Infrastructure.Messaging.Consumers consumeContext.Message.MediumPictureWidth, consumeContext.Message.MediumPictureHeight, consumeContext.Message.OriginalPictureKeyName, - consumeContext.Message.BucketName); + consumeContext.Message.BucketName, + workingDirectory); AddResizeActivity( routingSlipBuilder, @@ -39,19 +46,27 @@ namespace Musify.Infrastructure.Messaging.Consumers consumeContext.Message.LargePictureWidth, consumeContext.Message.LargePictureHeight, consumeContext.Message.OriginalPictureKeyName, - consumeContext.Message.BucketName); + consumeContext.Message.BucketName, + workingDirectory); routingSlipBuilder.AddActivity( "DeleteOriginal", - BuildExecuteUri(RemoveFileFromBucketActivity.ExecuteEndpointName), + MessagingHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName), new RemoveFileFromBucketArguments( consumeContext.Message.BucketName, - consumeContext.Message.OriginalPictureKeyName) - ); + consumeContext.Message.OriginalPictureKeyName)); + + routingSlipBuilder.AddActivity( + "UploadFiles", + MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName), + new TransferFilesToBucketArguments( + workingDirectory, + consumeContext.Message.BucketName, + folderName)); routingSlipBuilder.AddActivity( "UpdatePlayListPicture", - BuildExecuteUri(UpdatePlayListPictureActivity.ExecuteEndpointName), + MessagingHelper.BuildExecuteActivityUri(UpdatePlayListPictureActivity.ExecuteEndpointName), new UpdatePlayListPictureArguments( consumeContext.Message.PlayListId, consumeContext.Message.SmallPictureKeyName, @@ -69,20 +84,32 @@ namespace Musify.Infrastructure.Messaging.Consumers int width, int height, string originalPictureKeyName, - string bucketName) + string bucketName, + string workingDirectory) { + var fileName = Path.GetFileName(originalPictureKeyName); + var sourceFilePath = Path.Combine(workingDirectory, fileName); + var destinationFileName = Path.GetFileNameWithoutExtension(destinationKeyName) + Path.GetExtension(fileName); + var destinationFilePath = Path.Combine(workingDirectory, destinationFileName); + + // First: Download original file routingSlipBuilder.AddActivity( - activityName, - BuildExecuteUri(ResizePictureActivity.ExecuteEndpointName), - new ResizePictureArgument( + $"{activityName}_Download", + MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName), + new DownloadFileFromBucketArguments( bucketName, originalPictureKeyName, - bucketName, - destinationKeyName, + workingDirectory)); + + // Second: Resize picture + routingSlipBuilder.AddActivity( + activityName, + MessagingHelper.BuildExecuteActivityUri(ResizePictureActivity.ExecuteEndpointName), + new ResizePictureLocalArguments( + sourceFilePath, + destinationFilePath, width, height)); } - - static Uri BuildExecuteUri(string endpointName) => new($"queue:{endpointName}_execute"); } } \ No newline at end of file diff --git a/Infrastructure/Messaging/Consumers/UpdateTrackPictureConsumer.cs b/Infrastructure/Messaging/Consumers/UpdateTrackPictureConsumer.cs index 74a4fdd..8133f6d 100644 --- a/Infrastructure/Messaging/Consumers/UpdateTrackPictureConsumer.cs +++ b/Infrastructure/Messaging/Consumers/UpdateTrackPictureConsumer.cs @@ -1,18 +1,31 @@ using MassTransit; using Musify.Application.Events; +using Musify.Infrastructure.Configuration; using Musify.Infrastructure.Messaging.Activities; using Musify.Infrastructure.Messaging.Activities.Arguments; namespace Musify.Infrastructure.Messaging.Consumers { - public class UpdateTrackPictureConsumer(IBus bus) + public class UpdateTrackPictureConsumer( + IBus bus, + AudioTranscoderConfiguration audioTranscoderConfiguration) : IConsumer { - public const string QueueName = "update-track-picture"; + public const string QueueName = "Update-Track-Picture"; public async Task Consume(ConsumeContext consumeContext) { var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid()); + var folderName = Guid.NewGuid().ToString(); + var workingDirectory = Path.Combine(audioTranscoderConfiguration.Routes.WorkingDirectory, folderName); + + routingSlipBuilder.AddActivity( + "DownloadFile", + MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName), + new DownloadFileFromBucketArguments( + consumeContext.Message.SourceBucketName, + consumeContext.Message.SourceKeyName, + workingDirectory)); AddResizeActivity( routingSlipBuilder, @@ -20,38 +33,41 @@ namespace Musify.Infrastructure.Messaging.Consumers consumeContext.Message.SmallPictureKeyName, consumeContext.Message.SmallPictureWidth, consumeContext.Message.SmallPictureHeight, - consumeContext.Message.OriginalPictureKeyName, - consumeContext.Message.BucketName); + consumeContext.Message.SourceKeyName, + consumeContext.Message.SourceBucketName, + workingDirectory); AddResizeActivity( routingSlipBuilder, "ResizeMedium", - consumeContext.Message.SmallPictureKeyName, + consumeContext.Message.MediumPictureKeyName, consumeContext.Message.MediumPictureWidth, consumeContext.Message.MediumPictureHeight, - consumeContext.Message.OriginalPictureKeyName, - consumeContext.Message.BucketName); + consumeContext.Message.SourceKeyName, + consumeContext.Message.SourceBucketName, + workingDirectory); AddResizeActivity( routingSlipBuilder, - "ResizeMedium", - consumeContext.Message.SmallPictureKeyName, + "ResizeLarge", + consumeContext.Message.LargePictureKeyName, consumeContext.Message.LargePictureWidth, consumeContext.Message.LargePictureHeight, - consumeContext.Message.OriginalPictureKeyName, - consumeContext.Message.BucketName); + consumeContext.Message.SourceKeyName, + consumeContext.Message.SourceBucketName, + workingDirectory); routingSlipBuilder.AddActivity( - "DeleteOriginal", - BuildExecuteUri(RemoveFileFromBucketActivity.ExecuteEndpointName), - new RemoveFileFromBucketArguments( - consumeContext.Message.BucketName, - consumeContext.Message.OriginalPictureKeyName) - ); + "UploadFiles", + MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName), + new TransferFilesToBucketArguments( + workingDirectory, + consumeContext.Message.SourceBucketName, + folderName)); routingSlipBuilder.AddActivity( - "UpdateTrackPictureActivity", - BuildExecuteUri(UpdateTrackPictureActivity.ExecuteEndpointName), + "UpdateTrackPicture", + MessagingHelper.BuildExecuteActivityUri(UpdateTrackPictureActivity.ExecuteEndpointName), new UpdateTrackPictureArguments( consumeContext.Message.TrackId, consumeContext.Message.SmallPictureKeyName, @@ -69,20 +85,32 @@ namespace Musify.Infrastructure.Messaging.Consumers int width, int height, string originalPictureKeyName, - string bucketName) + string bucketName, + string workingDirectory) { + var fileName = Path.GetFileName(originalPictureKeyName); + var sourceFilePath = Path.Combine(workingDirectory, fileName); + var destinationFileName = Path.GetFileNameWithoutExtension(destinationKeyName) + Path.GetExtension(fileName); + var destinationFilePath = Path.Combine(workingDirectory, destinationFileName); + + // First: Download original file routingSlipBuilder.AddActivity( - activityName, - BuildExecuteUri(ResizePictureActivity.ExecuteEndpointName), - new ResizePictureArgument( + $"{activityName}_Download", + MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName), + new DownloadFileFromBucketArguments( bucketName, originalPictureKeyName, - bucketName, - destinationKeyName, + workingDirectory)); + + // Second: Resize picture + routingSlipBuilder.AddActivity( + activityName, + MessagingHelper.BuildExecuteActivityUri(ResizePictureActivity.ExecuteEndpointName), + new ResizePictureLocalArguments( + sourceFilePath, + destinationFilePath, width, height)); } - - static Uri BuildExecuteUri(string endpointName) => new($"queue:{endpointName}_execute"); } } \ No newline at end of file diff --git a/Infrastructure/Messaging/MessagingDependencyInjection.cs b/Infrastructure/Messaging/MessagingDependencyInjection.cs index d43bc7e..4c2324f 100644 --- a/Infrastructure/Messaging/MessagingDependencyInjection.cs +++ b/Infrastructure/Messaging/MessagingDependencyInjection.cs @@ -3,7 +3,6 @@ using Microsoft.Extensions.DependencyInjection; using Musify.Infrastructure.Configuration; using Musify.Infrastructure.Messaging.Activities; using Musify.Infrastructure.Messaging.Activities.Arguments; -using Musify.Infrastructure.Messaging.Activities.Logs; using Musify.Infrastructure.Messaging.Consumers; using Musify.Infrastructure.Messaging.Filters; @@ -19,7 +18,7 @@ namespace Musify.Infrastructure.Messaging ConfigureRabbitMqHost( busFactoryConfigurator, busRegistrationContext.GetRequiredService())); - } ); + }); return serviceDescriptors; } @@ -28,17 +27,20 @@ namespace Musify.Infrastructure.Messaging { serviceDescriptors.AddMassTransit(options => { + // Register consumers options.AddConsumer(); options.AddConsumer(); options.AddConsumer(); options.AddConsumer(); - options.AddExecuteActivity(); + // Register execute activities + options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddExecuteActivity(); + options.AddExecuteActivity(); options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) => { @@ -48,27 +50,29 @@ namespace Musify.Infrastructure.Messaging busFactoryConfigurator.UseConsumeFilter(typeof(ProcessTrackingConsumeFilter<>), busRegistrationContext); + // Consumer endpoints busFactoryConfigurator.ReceiveEndpoint(UpdatePlayListPictureConsumer.QueueName, endpointConfigurator => { endpointConfigurator.ConfigureConsumer(busRegistrationContext); - } ); + }); busFactoryConfigurator.ReceiveEndpoint(RemoveFileConsumer.QueueName, endpointConfigurator => { endpointConfigurator.ConfigureConsumer(busRegistrationContext); - } ); + }); busFactoryConfigurator.ReceiveEndpoint(TranscodeAudioFromTrackConsumer.QueueName, endpointConfigurator => { endpointConfigurator.ConfigureConsumer(busRegistrationContext); - } ); + }); busFactoryConfigurator.ReceiveEndpoint(UpdateTrackPictureConsumer.QueueName, endpointConfigurator => { endpointConfigurator.ConfigureConsumer(busRegistrationContext); - } ); + }); - ConfigureExecuteActivityEndpoint( + // Activity endpoints + ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, ResizePictureActivity.ExecuteEndpointName); @@ -97,8 +101,13 @@ namespace Musify.Infrastructure.Messaging busFactoryConfigurator, busRegistrationContext, TranscodeDashAudioActivity.ExecuteEndpointName); - } ); - } ); + + ConfigureExecuteActivityEndpoint( + busFactoryConfigurator, + busRegistrationContext, + TransferFilesToBucket.ExecuteEndpointName); + }); + }); return serviceDescriptors; } @@ -111,7 +120,7 @@ namespace Musify.Infrastructure.Messaging { options.Username(massTransitConfiguration.Username); options.Password(massTransitConfiguration.Password); - } ); + }); } static void ConfigureExecuteActivityEndpoint( @@ -124,7 +133,7 @@ namespace Musify.Infrastructure.Messaging busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_execute", endpointConfigurator => { endpointConfigurator.ExecuteActivityHost(busRegistrationContext); - } ); + }); } static void ConfigureActivityEndpoints( @@ -138,12 +147,12 @@ namespace Musify.Infrastructure.Messaging busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_execute", endpointConfigurator => { endpointConfigurator.ExecuteActivityHost(busRegistrationContext); - } ); + }); busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_compensate", endpointConfigurator => { endpointConfigurator.CompensateActivityHost(busRegistrationContext); - } ); + }); } } } \ No newline at end of file