From ab388e6ad6e9d9b944a57ef3578769a00ad06623 Mon Sep 17 00:00:00 2001 From: XShine555 Date: Thu, 24 Sep 2026 05:02:04 +0200 Subject: [PATCH] Fix processing pipeline: report album failures, fault handling on create slips, config key names, audio duration --- .../Albums/RequestAlbumPictureUpload.cs | 2 +- .../UploadIntentConfiguration.cs | 6 ++--- .../PlayLists/RequestPlayListPictureUpload.cs | 2 +- .../Tracks/RequestTrackUploadUrls.cs | 4 ++-- .../Jobs/TemporalUploadsCleanupJob.cs | 10 ++++---- .../Audio/TranscodeAudioActivity.cs | 2 +- .../Audio/UpdateTrackAudioActivity.cs | 6 +---- .../Consumers/ProcessingSlipFaultConsumer.cs | 22 +++++++++++++++++ .../AlbumPictureSourceRoutingSlipBuilder.cs | 17 ++++++++++--- .../Builders/CreateTrackRoutingSlipBuilder.cs | 9 +++++++ ...PlayListPictureSourceRoutingSlipBuilder.cs | 17 ++++++++++--- .../RoutingSlip/RoutingSlipVariableNames.cs | 6 +++++ .../Sagas/AlbumProcessingStateMachine.cs | 12 ++++++++-- .../Sagas/PlayListProcessingStateMachine.cs | 12 ++++++++-- .../Sagas/TrackProcessingStateMachine.cs | 24 +++++++++++++++---- 15 files changed, 119 insertions(+), 32 deletions(-) diff --git a/backend/Core/Musify.Application/Albums/RequestAlbumPictureUpload.cs b/backend/Core/Musify.Application/Albums/RequestAlbumPictureUpload.cs index c273345..09c3244 100644 --- a/backend/Core/Musify.Application/Albums/RequestAlbumPictureUpload.cs +++ b/backend/Core/Musify.Application/Albums/RequestAlbumPictureUpload.cs @@ -45,7 +45,7 @@ public class RequestAlbumPictureUploadCommandHandler( var objectName = $"{Guid.NewGuid()}.{request.FileType.TrimStart('.').ToLowerInvariant()}"; var tempKey = albumConfiguration.Routes.BuildTempPicturePath( - uploadIntentConfiguration.TemporalRootPrefix, request.UserId, objectName); + uploadIntentConfiguration.TempRootPrefix, request.UserId, objectName); try { diff --git a/backend/Core/Musify.Application/Configuration/UploadIntentConfiguration.cs b/backend/Core/Musify.Application/Configuration/UploadIntentConfiguration.cs index 15463ea..eb05dee 100644 --- a/backend/Core/Musify.Application/Configuration/UploadIntentConfiguration.cs +++ b/backend/Core/Musify.Application/Configuration/UploadIntentConfiguration.cs @@ -31,11 +31,11 @@ public class UploadIntentConfiguration public int ExpiredIntentsRetentionDays { get; set; } = 7; [Required, MinLength(1)] - public string TemporalRootPrefix { get; set; } = "temporal"; + public string TempRootPrefix { get; set; } = "temp"; [Range(1, 30)] - public int TemporalUploadsRetentionDays { get; set; } = 3; + public int TempUploadsRetentionDays { get; set; } = 3; [Range(60, 86400)] - public int TemporalCleanUpJobIntervalSeconds { get; set; } = 3600; + public int TempCleanupJobIntervalSeconds { get; set; } = 3600; } diff --git a/backend/Core/Musify.Application/PlayLists/RequestPlayListPictureUpload.cs b/backend/Core/Musify.Application/PlayLists/RequestPlayListPictureUpload.cs index 256ea6d..38c3a0f 100644 --- a/backend/Core/Musify.Application/PlayLists/RequestPlayListPictureUpload.cs +++ b/backend/Core/Musify.Application/PlayLists/RequestPlayListPictureUpload.cs @@ -45,7 +45,7 @@ public class RequestPlayListPictureUploadCommandHandler( var objectName = $"{Guid.NewGuid()}.{request.FileType.TrimStart('.').ToLowerInvariant()}"; var tempKey = playListConfiguration.Routes.BuildTempPicturePath( - uploadIntentConfiguration.TemporalRootPrefix, request.UserId, objectName); + uploadIntentConfiguration.TempRootPrefix, request.UserId, objectName); try { diff --git a/backend/Core/Musify.Application/Tracks/RequestTrackUploadUrls.cs b/backend/Core/Musify.Application/Tracks/RequestTrackUploadUrls.cs index fa077db..4d2fa1d 100644 --- a/backend/Core/Musify.Application/Tracks/RequestTrackUploadUrls.cs +++ b/backend/Core/Musify.Application/Tracks/RequestTrackUploadUrls.cs @@ -51,9 +51,9 @@ public class RequestTrackUploadUrlsCommandHandler( var audioObjectName = $"{Guid.NewGuid()}.{request.AudioFileType.TrimStart('.').ToLowerInvariant()}"; var tempPictureKey = trackConfiguration.Routes.BuildTempPicturePath( - uploadIntentConfiguration.TemporalRootPrefix, request.UserId, pictureObjectName); + uploadIntentConfiguration.TempRootPrefix, request.UserId, pictureObjectName); var tempAudioKey = trackConfiguration.Routes.BuildTempAudioPath( - uploadIntentConfiguration.TemporalRootPrefix, request.UserId, audioObjectName); + uploadIntentConfiguration.TempRootPrefix, request.UserId, audioObjectName); try { diff --git a/backend/Core/Musify.Infrastructure/Jobs/TemporalUploadsCleanupJob.cs b/backend/Core/Musify.Infrastructure/Jobs/TemporalUploadsCleanupJob.cs index 3f17c5c..d866c1f 100644 --- a/backend/Core/Musify.Infrastructure/Jobs/TemporalUploadsCleanupJob.cs +++ b/backend/Core/Musify.Infrastructure/Jobs/TemporalUploadsCleanupJob.cs @@ -16,10 +16,10 @@ public class TemporalUploadsCleanUpJob( protected override async Task ExecuteAsync(CancellationToken stoppingToken) { logger.LogInformation("TemporalUploadsCleanUpJob started (interval: {Interval}s, prefix: {Prefix})", - uploadIntentConfiguration.TemporalCleanUpJobIntervalSeconds, - uploadIntentConfiguration.TemporalRootPrefix); + uploadIntentConfiguration.TempCleanupJobIntervalSeconds, + uploadIntentConfiguration.TempRootPrefix); - using var timer = new PeriodicTimer(TimeSpan.FromSeconds(uploadIntentConfiguration.TemporalCleanUpJobIntervalSeconds)); + using var timer = new PeriodicTimer(TimeSpan.FromSeconds(uploadIntentConfiguration.TempCleanupJobIntervalSeconds)); do { @@ -40,13 +40,13 @@ public class TemporalUploadsCleanUpJob( await using var scope = scopeFactory.CreateAsyncScope(); var storageService = scope.ServiceProvider.GetRequiredService(); - var cutOff = DateTime.UtcNow.AddDays(-uploadIntentConfiguration.TemporalUploadsRetentionDays); + var cutOff = DateTime.UtcNow.AddDays(-uploadIntentConfiguration.TempUploadsRetentionDays); var deleted = 0; var errors = 0; await foreach (var (key, lastModified) in storageService.ListObjectsAsync( storageConfiguration.Bucket, - uploadIntentConfiguration.TemporalRootPrefix, + uploadIntentConfiguration.TempRootPrefix, cancellationToken)) { if (lastModified >= cutOff) diff --git a/backend/Core/Musify.Infrastructure/MassTransit/Activities/Audio/TranscodeAudioActivity.cs b/backend/Core/Musify.Infrastructure/MassTransit/Activities/Audio/TranscodeAudioActivity.cs index 6cf7d56..2c674b6 100644 --- a/backend/Core/Musify.Infrastructure/MassTransit/Activities/Audio/TranscodeAudioActivity.cs +++ b/backend/Core/Musify.Infrastructure/MassTransit/Activities/Audio/TranscodeAudioActivity.cs @@ -55,7 +55,7 @@ internal class TranscodeAudioActivity( var log = new TranscodeAudioLog(workingDirectory); return executeContext.CompletedWithVariables(log, new Dictionary { - [RoutingSlipVariableNames.Audio.DurationSeconds] = (int)Math.Round(transcodeResult.Duration.TotalSeconds) + [RoutingSlipVariableNames.Audio.DurationSeconds] = transcodeResult.Duration.TotalSeconds }); } catch (Exception exception) diff --git a/backend/Core/Musify.Infrastructure/MassTransit/Activities/Audio/UpdateTrackAudioActivity.cs b/backend/Core/Musify.Infrastructure/MassTransit/Activities/Audio/UpdateTrackAudioActivity.cs index 06b72ab..ef79453 100644 --- a/backend/Core/Musify.Infrastructure/MassTransit/Activities/Audio/UpdateTrackAudioActivity.cs +++ b/backend/Core/Musify.Infrastructure/MassTransit/Activities/Audio/UpdateTrackAudioActivity.cs @@ -38,11 +38,7 @@ internal class UpdateTrackAudioActivity( if (string.IsNullOrWhiteSpace(audioFolderKey)) { - logger.LogWarning("Audio folder key is empty for track {TrackId}; skipping update", track.Id); - await publishEndpoint.Publish( - new TrackAudioProcessed(executeContext.Arguments.TrackId), - executeContext.CancellationToken); - return executeContext.Completed(); + throw new InvalidOperationException($"Audio folder key is empty for track {track.Id}"); } try diff --git a/backend/Core/Musify.Infrastructure/MassTransit/Consumers/ProcessingSlipFaultConsumer.cs b/backend/Core/Musify.Infrastructure/MassTransit/Consumers/ProcessingSlipFaultConsumer.cs index e796383..8c1dca6 100644 --- a/backend/Core/Musify.Infrastructure/MassTransit/Consumers/ProcessingSlipFaultConsumer.cs +++ b/backend/Core/Musify.Infrastructure/MassTransit/Consumers/ProcessingSlipFaultConsumer.cs @@ -32,6 +32,28 @@ public class ProcessingSlipFaultConsumer : IConsumer case RoutingSlipVariableNames.ProcessKinds.PlayListPicture: await context.Publish(new PlayListPictureProcessingFailed(subjectId)); break; + case RoutingSlipVariableNames.ProcessKinds.AlbumPicture: + await context.Publish(new AlbumPictureProcessingFailed(subjectId)); + break; + case RoutingSlipVariableNames.ProcessKinds.TrackCreation: + await context.Publish(new TrackProcessingFailed( + subjectId, Variable(variables, RoutingSlipVariableNames.Workflow.Bucket), + Variable(variables, RoutingSlipVariableNames.Workflow.PictureKey), + Variable(variables, RoutingSlipVariableNames.Workflow.AudioKey))); + break; + case RoutingSlipVariableNames.ProcessKinds.AlbumCreation: + await context.Publish(new AlbumProcessingFailed( + subjectId, Variable(variables, RoutingSlipVariableNames.Workflow.Bucket), + Variable(variables, RoutingSlipVariableNames.Workflow.PictureKey))); + break; + case RoutingSlipVariableNames.ProcessKinds.PlayListCreation: + await context.Publish(new PlayListProcessingFailed( + subjectId, Variable(variables, RoutingSlipVariableNames.Workflow.Bucket), + Variable(variables, RoutingSlipVariableNames.Workflow.PictureKey))); + break; } } + + private static string? Variable(IDictionary variables, string name) => + variables.TryGetValue(name, out var value) ? value?.ToString() : null; } diff --git a/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/AlbumPictureSourceRoutingSlipBuilder.cs b/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/AlbumPictureSourceRoutingSlipBuilder.cs index 73ca7bf..866dbca 100644 --- a/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/AlbumPictureSourceRoutingSlipBuilder.cs +++ b/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/AlbumPictureSourceRoutingSlipBuilder.cs @@ -22,7 +22,8 @@ public class AlbumPictureSourceRoutingSlipBuilder message.Small, message.Medium, message.Large, - correlationId); + correlationId, + RoutingSlipVariableNames.ProcessKinds.AlbumCreation); public RoutingSlipBuilder Build(UpdateAlbumPictureSourceEvent message, Guid? correlationId) => BuildSlip( @@ -34,7 +35,8 @@ public class AlbumPictureSourceRoutingSlipBuilder message.Small, message.Medium, message.Large, - correlationId); + correlationId, + RoutingSlipVariableNames.ProcessKinds.AlbumPicture); private static RoutingSlipBuilder BuildSlip( Guid albumId, @@ -45,7 +47,8 @@ public class AlbumPictureSourceRoutingSlipBuilder ImageSize small, ImageSize medium, ImageSize large, - Guid? correlationId) + Guid? correlationId, + string processKind) { var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid()); routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty); @@ -79,6 +82,14 @@ public class AlbumPictureSourceRoutingSlipBuilder medium, large)); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, albumId); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, processKind); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, bucket); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, pictureDestinationKey); + routingSlipBuilder.AddSubscription( + EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName), + RoutingSlipEvents.Faulted); + return routingSlipBuilder; } } diff --git a/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/CreateTrackRoutingSlipBuilder.cs b/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/CreateTrackRoutingSlipBuilder.cs index e2aa438..426f544 100644 --- a/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/CreateTrackRoutingSlipBuilder.cs +++ b/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/CreateTrackRoutingSlipBuilder.cs @@ -57,6 +57,15 @@ public class CreateTrackRoutingSlipBuilder message.Medium, message.Large)); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, message.TrackId); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, RoutingSlipVariableNames.ProcessKinds.TrackCreation); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, message.Bucket); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, message.PictureDestinationKey); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.AudioKey, message.AudioDestinationKey); + routingSlipBuilder.AddSubscription( + EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName), + RoutingSlipEvents.Faulted); + return routingSlipBuilder; } } diff --git a/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/PlayListPictureSourceRoutingSlipBuilder.cs b/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/PlayListPictureSourceRoutingSlipBuilder.cs index 27bdaa5..6e79a3b 100644 --- a/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/PlayListPictureSourceRoutingSlipBuilder.cs +++ b/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/Builders/PlayListPictureSourceRoutingSlipBuilder.cs @@ -22,7 +22,8 @@ public class PlayListPictureSourceRoutingSlipBuilder message.Small, message.Medium, message.Large, - correlationId); + correlationId, + RoutingSlipVariableNames.ProcessKinds.PlayListCreation); public RoutingSlipBuilder Build(UpdatePlayListPictureSourceEvent message, Guid? correlationId) => BuildSlip( @@ -34,7 +35,8 @@ public class PlayListPictureSourceRoutingSlipBuilder message.Small, message.Medium, message.Large, - correlationId); + correlationId, + RoutingSlipVariableNames.ProcessKinds.PlayListPicture); private static RoutingSlipBuilder BuildSlip( Guid playListId, @@ -45,7 +47,8 @@ public class PlayListPictureSourceRoutingSlipBuilder ImageSize small, ImageSize medium, ImageSize large, - Guid? correlationId) + Guid? correlationId, + string processKind) { var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid()); routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty); @@ -79,6 +82,14 @@ public class PlayListPictureSourceRoutingSlipBuilder medium, large)); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, playListId); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, processKind); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, bucket); + routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, pictureDestinationKey); + routingSlipBuilder.AddSubscription( + EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName), + RoutingSlipEvents.Faulted); + return routingSlipBuilder; } } diff --git a/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/RoutingSlipVariableNames.cs b/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/RoutingSlipVariableNames.cs index 71fd347..6d49e2e 100644 --- a/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/RoutingSlipVariableNames.cs +++ b/backend/Core/Musify.Infrastructure/MassTransit/RoutingSlip/RoutingSlipVariableNames.cs @@ -8,6 +8,9 @@ internal static class RoutingSlipVariableNames internal const string CorrelationId = "Workflow.CorrelationId"; internal const string SubjectId = "Workflow.SubjectId"; internal const string ProcessKind = "Workflow.ProcessKind"; + internal const string Bucket = "Workflow.Bucket"; + internal const string PictureKey = "Workflow.PictureKey"; + internal const string AudioKey = "Workflow.AudioKey"; } internal static class ProcessKinds @@ -16,6 +19,9 @@ internal static class RoutingSlipVariableNames internal const string TrackAudio = "TrackAudio"; internal const string PlayListPicture = "PlayListPicture"; internal const string AlbumPicture = "AlbumPicture"; + internal const string TrackCreation = "TrackCreation"; + internal const string AlbumCreation = "AlbumCreation"; + internal const string PlayListCreation = "PlayListCreation"; } internal static class Audio diff --git a/backend/Core/Musify.Infrastructure/MassTransit/Sagas/AlbumProcessingStateMachine.cs b/backend/Core/Musify.Infrastructure/MassTransit/Sagas/AlbumProcessingStateMachine.cs index 4275aa4..3eaa300 100644 --- a/backend/Core/Musify.Infrastructure/MassTransit/Sagas/AlbumProcessingStateMachine.cs +++ b/backend/Core/Musify.Infrastructure/MassTransit/Sagas/AlbumProcessingStateMachine.cs @@ -17,8 +17,16 @@ public sealed class AlbumProcessingStateMachine : MassTransitStateMachine state.CurrentState); Event(() => ProcessingStarted, config => config.CorrelateById(context => context.Message.AlbumId)); - Event(() => PictureProcessed, config => config.CorrelateById(context => context.Message.AlbumId)); - Event(() => PictureFailed, config => config.CorrelateById(context => context.Message.AlbumId)); + Event(() => PictureProcessed, config => + { + config.CorrelateById(context => context.Message.AlbumId); + config.OnMissingInstance(instance => instance.Discard()); + }); + Event(() => PictureFailed, config => + { + config.CorrelateById(context => context.Message.AlbumId); + config.OnMissingInstance(instance => instance.Discard()); + }); Initially( When(ProcessingStarted) diff --git a/backend/Core/Musify.Infrastructure/MassTransit/Sagas/PlayListProcessingStateMachine.cs b/backend/Core/Musify.Infrastructure/MassTransit/Sagas/PlayListProcessingStateMachine.cs index c9eaec4..3ab7e6c 100644 --- a/backend/Core/Musify.Infrastructure/MassTransit/Sagas/PlayListProcessingStateMachine.cs +++ b/backend/Core/Musify.Infrastructure/MassTransit/Sagas/PlayListProcessingStateMachine.cs @@ -17,8 +17,16 @@ public sealed class PlayListProcessingStateMachine : MassTransitStateMachine state.CurrentState); Event(() => ProcessingStarted, config => config.CorrelateById(context => context.Message.PlayListId)); - Event(() => PictureProcessed, config => config.CorrelateById(context => context.Message.PlayListId)); - Event(() => PictureFailed, config => config.CorrelateById(context => context.Message.PlayListId)); + Event(() => PictureProcessed, config => + { + config.CorrelateById(context => context.Message.PlayListId); + config.OnMissingInstance(instance => instance.Discard()); + }); + Event(() => PictureFailed, config => + { + config.CorrelateById(context => context.Message.PlayListId); + config.OnMissingInstance(instance => instance.Discard()); + }); Initially( When(ProcessingStarted) diff --git a/backend/Core/Musify.Infrastructure/MassTransit/Sagas/TrackProcessingStateMachine.cs b/backend/Core/Musify.Infrastructure/MassTransit/Sagas/TrackProcessingStateMachine.cs index c01565e..5fdc6de 100644 --- a/backend/Core/Musify.Infrastructure/MassTransit/Sagas/TrackProcessingStateMachine.cs +++ b/backend/Core/Musify.Infrastructure/MassTransit/Sagas/TrackProcessingStateMachine.cs @@ -19,10 +19,26 @@ public sealed class TrackProcessingStateMachine : MassTransitStateMachine state.CurrentState); Event(() => ProcessingStarted, config => config.CorrelateById(context => context.Message.TrackId)); - Event(() => PictureProcessed, config => config.CorrelateById(context => context.Message.TrackId)); - Event(() => AudioProcessed, config => config.CorrelateById(context => context.Message.TrackId)); - Event(() => PictureFailed, config => config.CorrelateById(context => context.Message.TrackId)); - Event(() => AudioFailed, config => config.CorrelateById(context => context.Message.TrackId)); + Event(() => PictureProcessed, config => + { + config.CorrelateById(context => context.Message.TrackId); + config.OnMissingInstance(instance => instance.Discard()); + }); + Event(() => AudioProcessed, config => + { + config.CorrelateById(context => context.Message.TrackId); + config.OnMissingInstance(instance => instance.Discard()); + }); + Event(() => PictureFailed, config => + { + config.CorrelateById(context => context.Message.TrackId); + config.OnMissingInstance(instance => instance.Discard()); + }); + Event(() => AudioFailed, config => + { + config.CorrelateById(context => context.Message.TrackId); + config.OnMissingInstance(instance => instance.Discard()); + }); Initially( When(ProcessingStarted)