feat(infra): playlist processing saga + generalize fault handling (phase 3)

Adds the PlayList picture-processing lifecycle saga (correlated by PlayListId) mirroring the track one, and generalizes the failure plumbing: - Rename the track fault translator to ProcessingSlipFaultConsumer and use a generic Workflow.SubjectId variable + ProcessKind; it now emits typed Track{Picture,Audio}/PlayListPicture ProcessingFailed events for all slips - PlayListProcessingStateMachine (+ EF-persisted PlayListProcessingState, migration): starts on CreatePlayListResourcesEvent, finalizes on PlayListPictureProcessed, and on failure marks the playlist Failed and removes the orphaned final picture (PlayListProcessingFailedConsumer + MarkPlayListAsFailedActivity) - UpdatePlayListPictureActivity publishes PlayListPictureProcessed on success - Picture slip carries SubjectId + PlayListPicture kind + fault subscription Verified end-to-end in Docker (RabbitMQ + migrated Postgres): both saga endpoints, the fault translator and cleanup consumers configure, and an addressing cross-check confirms all 21 activity execute queues the builders address exist (caught + fixed a kebab mismatch on mark-play-list-as-failed). Build: 0 warnings, 0 errors. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-05 14:55:40 +02:00
parent 8664bfb800
commit a5305b8ed8
20 changed files with 913 additions and 12 deletions
@@ -1,6 +1,7 @@
 using MassTransit;
 using Microsoft.Extensions.Logging;
 using Musify.Application.Contracts;
+using Musify.Application.Events;
 using Musify.Infrastructure.MassTransit.Arguments;
 using Musify.Infrastructure.MassTransit.Logs;

@@ -8,6 +9,7 @@ namespace Musify.Infrastructure.MassTransit.Activities.Pictures
 {
    internal class UpdatePlayListPictureActivity(
        IDatabase database,
+        IPublishEndpoint publishEndpoint,
        ILogger<UpdatePlayListPictureActivity> logger)
        : IActivity<UpdatePlayListPictureArguments, UpdatePlayListPictureLog>
    {
@@ -52,6 +54,11 @@ namespace Musify.Infrastructure.MassTransit.Activities.Pictures

                logger.LogInformation("Updated playlist {PlayListId} pictures",
                    executeContext.Arguments.PlayListId);
+
+                await publishEndpoint.Publish(
+                    new PlayListPictureProcessed(executeContext.Arguments.PlayListId),
+                    executeContext.CancellationToken);
+
                return executeContext.Completed();
            }
            catch (Exception exception)
@@ -0,0 +1,32 @@
+using MassTransit;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging;
+using Musify.Application.Contracts;
+using Musify.Domain.ValueObjects;
+using Musify.Infrastructure.MassTransit.Arguments;
+
+namespace Musify.Infrastructure.MassTransit.Activities.PlayLists
+{
+    internal class MarkPlayListAsFailedActivity(
+        IDatabase database,
+        ILogger<MarkPlayListAsFailedActivity> logger)
+        : IExecuteActivity<MarkPlayListAsFailedArguments>
+    {
+        public const string ExecuteEndpointName = "mark-play-list-as-failed";
+
+        public async Task<ExecutionResult> Execute(ExecuteContext<MarkPlayListAsFailedArguments> executeContext)
+        {
+            var playList = await database.PlayLists
+                .SingleOrDefaultAsync(p => p.Id == executeContext.Arguments.PlayListId, executeContext.CancellationToken);
+
+            if (playList is not null)
+            {
+                playList.LifeCycleStatus = LifeCycleStatus.Failed;
+                await database.SaveChangesAsync(executeContext.CancellationToken);
+                logger.LogInformation("Marked playlist {PlayListId} as failed", executeContext.Arguments.PlayListId);
+            }
+
+            return executeContext.Completed();
+        }
+    }
+}
@@ -0,0 +1,4 @@
+namespace Musify.Infrastructure.MassTransit.Arguments
+{
+    public record MarkPlayListAsFailedArguments(Guid PlayListId);
+}
@@ -0,0 +1,36 @@
+using MassTransit;
+using MassTransit.Courier.Contracts;
+using Musify.Application.Events;
+using Musify.Infrastructure.MassTransit.Activities.Files;
+using Musify.Infrastructure.MassTransit.Activities.PlayLists;
+using Musify.Infrastructure.MassTransit.Arguments;
+using Musify.Infrastructure.MassTransit.RoutingSlip;
+
+namespace Musify.Infrastructure.MassTransit.Consumers
+{
+    public class PlayListProcessingFailedConsumer(IBus bus) : IConsumer<PlayListProcessingFailed>
+    {
+        public async Task Consume(ConsumeContext<PlayListProcessingFailed> context)
+        {
+            var message = context.Message;
+
+            var builder = new RoutingSlipBuilder(NewId.NextGuid());
+            builder.AddSubscription(
+                EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
+                RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
+
+            if (!string.IsNullOrWhiteSpace(message.Bucket) && !string.IsNullOrWhiteSpace(message.PictureKey))
+                builder.AddActivity(
+                    ActivityNames.RemovePlayListOriginalPicture,
+                    EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
+                    new RemoveFileFromBucketArguments(message.Bucket, message.PictureKey));
+
+            builder.AddActivity(
+                ActivityNames.MarkPlayListAsFailed,
+                EndpointHelper.BuildExecuteActivityUri(MarkPlayListAsFailedActivity.ExecuteEndpointName),
+                new MarkPlayListAsFailedArguments(message.PlayListId));
+
+            await bus.Execute(builder.Build());
+        }
+    }
+}
@@ -4,9 +4,9 @@ using Musify.Application.Events;

 namespace Musify.Infrastructure.MassTransit.Consumers
 {
-    public class TrackProcessingSlipFaultConsumer : IConsumer<RoutingSlipFaulted>
+    public class ProcessingSlipFaultConsumer : IConsumer<RoutingSlipFaulted>
    {
-        public const string QueueName = "track-processing-slip-fault";
+        public const string QueueName = "processing-slip-fault";

        public async Task Consume(ConsumeContext<RoutingSlipFaulted> context)
        {
@@ -15,17 +15,20 @@ namespace Musify.Infrastructure.MassTransit.Consumers
            if (!variables.TryGetValue(RoutingSlipVariableNames.Workflow.ProcessKind, out var kindObject))
                return;

-            if (!variables.TryGetValue(RoutingSlipVariableNames.Workflow.TrackId, out var trackObject)
-                || !Guid.TryParse(trackObject?.ToString(), out var trackId))
+            if (!variables.TryGetValue(RoutingSlipVariableNames.Workflow.SubjectId, out var subjectObject)
+                || !Guid.TryParse(subjectObject?.ToString(), out var subjectId))
                return;

            switch (kindObject?.ToString())
            {
                case RoutingSlipVariableNames.ProcessKinds.TrackPicture:
-                    await context.Publish(new TrackPictureProcessingFailed(trackId));
+                    await context.Publish(new TrackPictureProcessingFailed(subjectId));
                    break;
                case RoutingSlipVariableNames.ProcessKinds.TrackAudio:
-                    await context.Publish(new TrackAudioProcessingFailed(trackId));
+                    await context.Publish(new TrackAudioProcessingFailed(subjectId));
+                    break;
+                case RoutingSlipVariableNames.ProcessKinds.PlayListPicture:
+                    await context.Publish(new PlayListPictureProcessingFailed(subjectId));
                    break;
            }
        }
@@ -36,8 +36,9 @@ namespace Musify.Infrastructure.MassTransit
            options.AddConsumer<CreateTrackConsumer>();
            options.AddConsumer<CreatePlayListConsumer>();
            options.AddConsumer<UpdatePlayListPictureSourceConsumer>();
-            options.AddConsumer<TrackProcessingSlipFaultConsumer>();
+            options.AddConsumer<ProcessingSlipFaultConsumer>();
            options.AddConsumer<TrackProcessingFailedConsumer>();
+            options.AddConsumer<PlayListProcessingFailedConsumer>();

            options.AddExecuteActivity<RemoveFileFromBucketActivity, RemoveFileFromBucketArguments>();
            options.AddExecuteActivity<GenerateAudioWorkflowPathsActivity, GenerateAudioWorkflowPathsArguments>();
@@ -49,6 +50,7 @@ namespace Musify.Infrastructure.MassTransit
            options.AddExecuteActivity<PublishTrackProcessingEventsActivity, PublishTrackProcessingEventsArguments>();
            options.AddExecuteActivity<PublishPlayListPictureProcessingEventActivity, PublishPlayListPictureProcessingEventArguments>();
            options.AddExecuteActivity<MarkTrackAsFailedActivity, MarkTrackAsFailedArguments>();
+            options.AddExecuteActivity<MarkPlayListAsFailedActivity, MarkPlayListAsFailedArguments>();

            options.AddActivity<ResizePictureActivity, ResizePictureLocalArguments, ResizePictureLog>();
            options.AddActivity<UpdatePlayListPictureActivity, UpdatePlayListPictureArguments, UpdatePlayListPictureLog>();
@@ -75,6 +75,14 @@ namespace Musify.Infrastructure.MassTransit
                        repository.UsePostgres();
                    });

+                options.AddSagaStateMachine<PlayListProcessingStateMachine, PlayListProcessingState>()
+                    .EntityFrameworkRepository(repository =>
+                    {
+                        repository.ConcurrencyMode = ConcurrencyMode.Pessimistic;
+                        repository.ExistingDbContext<Database>();
+                        repository.UsePostgres();
+                    });
+
                options.AddConfigureEndpointsCallback((registrationContext, _, endpointConfigurator) =>
                {
                    UseStandardRetry(endpointConfigurator);
@@ -6,6 +6,7 @@ namespace Musify.Infrastructure.MassTransit.RoutingSlip
        internal const string MarkTrackAsFailed = "MarkTrackAsFailed";
        internal const string DeleteTrackFromDb = "DeleteTrackFromDb";
        internal const string MarkPlayListAsRemoving = "MarkPlayListAsRemoving";
+        internal const string MarkPlayListAsFailed = "MarkPlayListAsFailed";
        internal const string DeletePlayListFromDb = "DeletePlayListFromDb";

        internal const string GenerateAudioWorkflowPaths = "GenerateAudioWorkflowPaths";
@@ -60,10 +60,10 @@ namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders
                    message.TrackId,
                    message.DestinationFolderKey));

-            routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.TrackId, message.TrackId);
+            routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, message.TrackId);
            routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, RoutingSlipVariableNames.ProcessKinds.TrackAudio);
            routingSlipBuilder.AddSubscription(
-                EndpointHelper.BuildConsumerUri(TrackProcessingSlipFaultConsumer.QueueName),
+                EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
                RoutingSlipEvents.Faulted);

            return routingSlipBuilder;
@@ -32,10 +32,10 @@ namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders
                    RoutingSlipVariableNames.Picture.MediumResizedFilePath,
                    RoutingSlipVariableNames.Picture.LargeResizedFilePath));

-            routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.TrackId, message.TrackId);
+            routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, message.TrackId);
            routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, RoutingSlipVariableNames.ProcessKinds.TrackPicture);
            routingSlipBuilder.AddSubscription(
-                EndpointHelper.BuildConsumerUri(TrackProcessingSlipFaultConsumer.QueueName),
+                EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
                RoutingSlipEvents.Faulted);

            return routingSlipBuilder;
@@ -61,6 +61,12 @@ namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders
                    RoutingSlipVariableNames.Picture.MediumResizedFilePath,
                    RoutingSlipVariableNames.Picture.LargeResizedFilePath));

+            routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, message.PlayListId);
+            routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, RoutingSlipVariableNames.ProcessKinds.PlayListPicture);
+            routingSlipBuilder.AddSubscription(
+                EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
+                RoutingSlipEvents.Faulted);
+
            return routingSlipBuilder;
        }

@@ -6,7 +6,7 @@ namespace Musify.Infrastructure.MassTransit
        {
            internal const string TemporalDirectory = "Workflow.TemporalDirectory";
            internal const string CorrelationId = "Workflow.CorrelationId";
-            internal const string TrackId = "Workflow.TrackId";
+            internal const string SubjectId = "Workflow.SubjectId";
            internal const string ProcessKind = "Workflow.ProcessKind";
        }

@@ -14,6 +14,7 @@ namespace Musify.Infrastructure.MassTransit
        {
            internal const string TrackPicture = "TrackPicture";
            internal const string TrackAudio = "TrackAudio";
+            internal const string PlayListPicture = "PlayListPicture";
        }

        internal static class Audio
@@ -0,0 +1,19 @@
+using MassTransit;
+
+namespace Musify.Infrastructure.MassTransit.Sagas
+{
+    public class PlayListProcessingState : SagaStateMachineInstance
+    {
+        public Guid CorrelationId { get; set; }
+
+        public string CurrentState { get; set; } = null!;
+
+        public string? Bucket { get; set; }
+
+        public string? PictureKey { get; set; }
+
+        public DateTime CreatedAt { get; set; }
+
+        public DateTime UpdatedAt { get; set; }
+    }
+}
@@ -0,0 +1,51 @@
+using MassTransit;
+using Musify.Application.Events;
+
+namespace Musify.Infrastructure.MassTransit.Sagas
+{
+    public sealed class PlayListProcessingStateMachine : MassTransitStateMachine<PlayListProcessingState>
+    {
+        public State Processing { get; private set; } = null!;
+        public State Failed { get; private set; } = null!;
+
+        public Event<CreatePlayListResourcesEvent> ProcessingStarted { get; private set; } = null!;
+        public Event<PlayListPictureProcessed> PictureProcessed { get; private set; } = null!;
+        public Event<PlayListPictureProcessingFailed> PictureFailed { get; private set; } = null!;
+
+        public PlayListProcessingStateMachine()
+        {
+            InstanceState(state => 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));
+
+            Initially(
+                When(ProcessingStarted)
+                    .Then(context =>
+                    {
+                        context.Saga.CreatedAt = DateTime.UtcNow;
+                        context.Saga.UpdatedAt = DateTime.UtcNow;
+                        context.Saga.Bucket = context.Message.Bucket;
+                        context.Saga.PictureKey = context.Message.PictureDestinationKey;
+                    })
+                    .TransitionTo(Processing));
+
+            During(Processing,
+                When(PictureProcessed).Finalize(),
+                When(PictureFailed)
+                    .Then(context => context.Saga.UpdatedAt = DateTime.UtcNow)
+                    .TransitionTo(Failed)
+                    .Publish(context => new PlayListProcessingFailed(
+                        context.Saga.CorrelationId,
+                        context.Saga.Bucket,
+                        context.Saga.PictureKey)));
+
+            During(Failed,
+                Ignore(PictureProcessed),
+                Ignore(PictureFailed));
+
+            SetCompletedWhenFinalized();
+        }
+    }
+}