feat(infra): add Track processing saga (state machine, phase 1)

Introduce a MassTransit saga state machine that correlates by TrackId and tracks the two parallel processing slips (picture + audio): - TrackProcessingStateMachine + EF-persisted TrackProcessingState (pessimistic concurrency, EntityFrameworkRepository), migration added - Starts on CreateTrackResourcesEvent; UpdateTrack{Picture,Audio}Activity now publish TrackPictureProcessed / TrackAudioProcessed on success; when both arrive the saga finalizes (Ready). Failure transitions to a Failed state (cleanup wired in the next phase) - Add LifeCycleStatus.Failed and the four outcome events Verified: booted the Worker against RabbitMQ + migrated Postgres in Docker; the saga endpoint 'track-processing-state' is configured and the bus starts clean. Build: 0 warnings, 0 errors.
2026-07-05 14:41:19 +02:00
parent ee3e457dde
commit 97855d9094
14 changed files with 809 additions and 1 deletions
@@ -0,0 +1,19 @@
+using MassTransit;
+
+namespace Musify.Infrastructure.MassTransit.Sagas
+{
+    public class TrackProcessingState : SagaStateMachineInstance
+    {
+        public Guid CorrelationId { get; set; }
+
+        public string CurrentState { get; set; } = null!;
+
+        public bool PictureProcessed { get; set; }
+
+        public bool AudioProcessed { get; set; }
+
+        public DateTime CreatedAt { get; set; }
+
+        public DateTime UpdatedAt { get; set; }
+    }
+}
@@ -0,0 +1,67 @@
+using MassTransit;
+using Musify.Application.Events;
+
+namespace Musify.Infrastructure.MassTransit.Sagas
+{
+    public sealed class TrackProcessingStateMachine : MassTransitStateMachine<TrackProcessingState>
+    {
+        public State Processing { get; private set; } = null!;
+        public State Failed { get; private set; } = null!;
+
+        public Event<CreateTrackResourcesEvent> ProcessingStarted { get; private set; } = null!;
+        public Event<TrackPictureProcessed> PictureProcessed { get; private set; } = null!;
+        public Event<TrackAudioProcessed> AudioProcessed { get; private set; } = null!;
+        public Event<TrackPictureProcessingFailed> PictureFailed { get; private set; } = null!;
+        public Event<TrackAudioProcessingFailed> AudioFailed { get; private set; } = null!;
+
+        public TrackProcessingStateMachine()
+        {
+            InstanceState(state => 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));
+
+            Initially(
+                When(ProcessingStarted)
+                    .Then(context =>
+                    {
+                        context.Saga.CreatedAt = DateTime.UtcNow;
+                        context.Saga.UpdatedAt = DateTime.UtcNow;
+                    })
+                    .TransitionTo(Processing));
+
+            During(Processing,
+                When(PictureProcessed)
+                    .Then(context =>
+                    {
+                        context.Saga.PictureProcessed = true;
+                        context.Saga.UpdatedAt = DateTime.UtcNow;
+                    })
+                    .If(context => context.Saga.AudioProcessed, binder => binder.Finalize()),
+                When(AudioProcessed)
+                    .Then(context =>
+                    {
+                        context.Saga.AudioProcessed = true;
+                        context.Saga.UpdatedAt = DateTime.UtcNow;
+                    })
+                    .If(context => context.Saga.PictureProcessed, binder => binder.Finalize()),
+                When(PictureFailed)
+                    .Then(context => context.Saga.UpdatedAt = DateTime.UtcNow)
+                    .TransitionTo(Failed),
+                When(AudioFailed)
+                    .Then(context => context.Saga.UpdatedAt = DateTime.UtcNow)
+                    .TransitionTo(Failed));
+
+            During(Failed,
+                Ignore(PictureProcessed),
+                Ignore(AudioProcessed),
+                Ignore(PictureFailed),
+                Ignore(AudioFailed));
+
+            SetCompletedWhenFinalized();
+        }
+    }
+}