Rework jobs

2026-03-20 21:40:05 +01:00
parent 93a975e9b5
commit b34a711d2d
22 changed files with 449 additions and 157 deletions
@@ -21,7 +21,8 @@ namespace Musify.Infrastructure
                .AddStorageHandler()
                .AddPictureHandler()
                .AddMassTransitClient(configuration)
-                .AddScoped<IEventBus, MassTransitEventBus>();
+                .AddScoped<IEventBus, MassTransitEventBus>()
+                .AddScoped<IProcessTrackingStore, ProcessTrackingStore>();

            return serviceDescriptors;
        }
@@ -34,7 +35,8 @@ namespace Musify.Infrastructure
                .AddStorageHandler()
                .AddPictureHandler()
                .AddMassTransitConsumers(configuration)
-                .AddScoped<IEventBus, MassTransitEventBus>();
+                .AddScoped<IEventBus, MassTransitEventBus>()
+                .AddScoped<IProcessTrackingStore, ProcessTrackingStore>();

            return serviceDescriptors;
        }
@@ -1,33 +1,60 @@
 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 RemoveFilesActivity(IStorageHandler storageHandler, ILogger<RemoveFilesActivity> logger)
+    public class RemoveFilesActivity(
+        IStorageHandler storageHandler,
+        ILogger<RemoveFilesActivity> logger,
+        IProcessTrackingStore processTrackingStore)
        : IExecuteActivity<RemoveFileArguments>
    {
        public const string ExecuteEndpointName = "Remove-File";

        public async Task<ExecutionResult> Execute(ExecuteContext<RemoveFileArguments> executeContext)
        {
-            var removeFile = await storageHandler.RemoveFileAsync(
-                executeContext.Arguments.BucketName,
-                executeContext.Arguments.KeyName,
+            var processId = await processTrackingStore.GetOrCreateProcessAsync(
+                "RoutingSlip",
+                executeContext.CorrelationId ?? executeContext.TrackingNumber,
+                executeContext.ConversationId,
+                executeContext.MessageId,
                executeContext.CancellationToken);

-            if (!removeFile.IsSuccess)
-            {
-                string errorMessage = string.Join(";", removeFile.Errors);
-                logger.LogWarning(
-                    "Failed to remove file {KeyName} from bucket {BucketName}: {ErrorMessage}",
-                    executeContext.Arguments.KeyName,
-                    executeContext.Arguments.BucketName,
-                    errorMessage);
-            }
+            var stepId = await processTrackingStore.StartStepAsync(
+                processId,
+                nameof(RemoveFilesActivity),
+                ProcessStepComponentType.Activity,
+                0,
+                executeContext.CancellationToken);

-            return executeContext.Completed();
+            try
+            {
+                var removeFile = await storageHandler.RemoveFileAsync(
+                    executeContext.Arguments.BucketName,
+                    executeContext.Arguments.KeyName,
+                    executeContext.CancellationToken);
+
+                if (!removeFile.IsSuccess)
+                {
+                    string errorMessage = string.Join(";", removeFile.Errors);
+                    logger.LogWarning(
+                        "Failed to remove file {KeyName} from bucket {BucketName}: {ErrorMessage}",
+                        executeContext.Arguments.KeyName,
+                        executeContext.Arguments.BucketName,
+                        errorMessage);
+                }
+
+                await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
+                return executeContext.Completed();
+            }
+            catch (Exception exception)
+            {
+                await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
+                throw;
+            }
        }
    }
 }
@@ -1,53 +1,80 @@
 using MassTransit;
 using Musify.Application.Contracts.Infrastructure;
+using Musify.Domain.Entities;
 using Musify.Infrastructure.Messaging.Activities.Arguments;

 namespace Musify.Infrastructure.Messaging.Activities
 {
-    public class ResizePictureActivity(IStorageHandler storageHandler, IPictureHandler pictureHandler)
+    public class ResizePictureActivity(
+        IStorageHandler storageHandler,
+        IPictureHandler pictureHandler,
+        IProcessTrackingStore processTrackingStore)
        : IExecuteActivity<ResizePictureArgument>
    {
        public const string ExecuteEndpointName = "Resize-Picture";

        public async Task<ExecutionResult> Execute(ExecuteContext<ResizePictureArgument> executeContext)
        {
-            var file = await storageHandler.GetFileAsync(
-                executeContext.Arguments.OriginalBucketName,
-                executeContext.Arguments.OriginalKeyName,
+            var processId = await processTrackingStore.GetOrCreateProcessAsync(
+                "RoutingSlip",
+                executeContext.CorrelationId ?? executeContext.TrackingNumber,
+                executeContext.ConversationId,
+                executeContext.MessageId,
                executeContext.CancellationToken);

-            if (!file.IsSuccess)
-            {
-                string errorMessages = string.Join(", ", file.Errors);
-                throw new Exception(errorMessages);
-            }
-
-            var resizedPicture = await pictureHandler.ResizePictureAsync(
-                file.Value,
-                executeContext.Arguments.Width,
-                executeContext.Arguments.Height,
+            var stepId = await processTrackingStore.StartStepAsync(
+                processId,
+                nameof(ResizePictureActivity),
+                ProcessStepComponentType.Activity,
+                0,
                executeContext.CancellationToken);

-            if (!resizedPicture.IsSuccess)
+            try
            {
-                string errorMessage = string.Join(", ", resizedPicture.Errors);
-                throw new Exception(errorMessage);
+                var file = await storageHandler.GetFileAsync(
+                    executeContext.Arguments.OriginalBucketName,
+                    executeContext.Arguments.OriginalKeyName,
+                    executeContext.CancellationToken);
+
+                if (!file.IsSuccess)
+                {
+                    string errorMessages = string.Join(", ", file.Errors);
+                    throw new Exception(errorMessages);
+                }
+
+                var resizedPicture = await pictureHandler.ResizePictureAsync(
+                    file.Value,
+                    executeContext.Arguments.Width,
+                    executeContext.Arguments.Height,
+                    executeContext.CancellationToken);
+
+                if (!resizedPicture.IsSuccess)
+                {
+                    string errorMessage = string.Join(", ", resizedPicture.Errors);
+                    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 errorMessage = string.Join(", ", uploadFile.Errors);
+                    throw new Exception(errorMessage);
+                }
+
+                await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
+                return executeContext.Completed();
            }
-
-            var uploadFile = await storageHandler.UploadFileAsync(
-                resizedPicture.Value,
-                pictureHandler.ContentType,
-                executeContext.Arguments.DestinationBucketName,
-                executeContext.Arguments.DestinationKeyName,
-                executeContext.CancellationToken);
-
-            if (!uploadFile.IsSuccess)
+            catch (Exception exception)
            {
-                var errorMessage = string.Join(", ", uploadFile.Errors);
-                throw new Exception(errorMessage);
+                await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
+                throw;
            }
-
-            return executeContext.Completed();
        }
    }
 }
@@ -1,31 +1,53 @@
 using MassTransit;
-using Microsoft.EntityFrameworkCore;
 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)
+    public class UpdatePlayListPictureActivity(IDatabase database, IProcessTrackingStore processTrackingStore)
        : IExecuteActivity<UpdatePlayListPictureArguments>
    {
        public const string ExecuteEndpointName = "Update-PlayList-Picture";

        public async Task<ExecutionResult> Execute(ExecuteContext<UpdatePlayListPictureArguments> executeContext)
        {
-            var playList = await database.PlayLists.SingleOrDefaultAsync(p => p.Id == executeContext.Arguments.PlayListId,
+            var processId = await processTrackingStore.GetOrCreateProcessAsync(
+                "RoutingSlip",
+                executeContext.CorrelationId ?? executeContext.TrackingNumber,
+                executeContext.ConversationId,
+                executeContext.MessageId,
                executeContext.CancellationToken);

-            if (playList is null)
-                throw new Exception($"PlayList with id {executeContext.Arguments.PlayListId} not found.");
+            var stepId = await processTrackingStore.StartStepAsync(
+                processId,
+                nameof(UpdatePlayListPictureActivity),
+                ProcessStepComponentType.Activity,
+                0,
+                executeContext.CancellationToken);

-            playList.SmallPictureKeyName = executeContext.Arguments.SmallPictureKeyName;
-            playList.MediumPictureKeyName = executeContext.Arguments.MediumPictureKeyName;
-            playList.LargePictureKeyName = executeContext.Arguments.LargePictureKeyName;
+            try
+            {
+                var playList = await database.PlayLists.FindAsync(executeContext.Arguments.PlayListId, executeContext.CancellationToken);

-            database.PlayLists.Update(playList);
-            await database.SaveChangesAsync(executeContext.CancellationToken);
+                if (playList is null)
+                    throw new Exception($"PlayList with id {executeContext.Arguments.PlayListId} not found.");

-            return executeContext.Completed();
+                playList.SmallPictureKeyName = executeContext.Arguments.SmallPictureKeyName;
+                playList.MediumPictureKeyName = executeContext.Arguments.MediumPictureKeyName;
+                playList.LargePictureKeyName = executeContext.Arguments.LargePictureKeyName;
+
+                database.PlayLists.Update(playList);
+                await database.SaveChangesAsync(executeContext.CancellationToken);
+
+                await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
+                return executeContext.Completed();
+            }
+            catch (Exception exception)
+            {
+                await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
+                throw;
+            }
        }
    }
 }
@@ -0,0 +1,48 @@
+using MassTransit;
+using Musify.Application.Contracts.Infrastructure;
+using Musify.Domain.Entities;
+
+namespace Musify.Infrastructure.Messaging.Filters
+{
+    public class ProcessTrackingConsumeFilter<TMessage>(IProcessTrackingStore processTrackingStore)
+        : IFilter<ConsumeContext<TMessage>>
+        where TMessage : class
+    {
+        public async Task Send(ConsumeContext<TMessage> context, IPipe<ConsumeContext<TMessage>> next)
+        {
+            var processId = await processTrackingStore.GetOrCreateProcessAsync(
+                typeof(TMessage).Name,
+                context.CorrelationId ?? context.MessageId,
+                context.ConversationId,
+                context.MessageId,
+                context.CancellationToken);
+
+            var stepId = await processTrackingStore.StartStepAsync(
+                processId,
+                $"{typeof(TMessage).Name}.Consume",
+                ProcessStepComponentType.Consumer,
+                0,
+                context.CancellationToken);
+
+            try
+            {
+                await next.Send(context);
+                await processTrackingStore.CompleteStepAsync(processId, stepId, context.CancellationToken);
+            }
+            catch (Exception exception)
+            {
+                await processTrackingStore.FailStepAsync(
+                    processId,
+                    stepId,
+                    exception.Message,
+                    context.CancellationToken);
+                throw;
+            }
+        }
+
+        public void Probe(ProbeContext context)
+        {
+            context.CreateFilterScope(nameof(ProcessTrackingConsumeFilter<TMessage>));
+        }
+    }
+}
@@ -4,6 +4,7 @@ using Microsoft.Extensions.DependencyInjection;
 using Musify.Infrastructure.Messaging.Activities;
 using Musify.Infrastructure.Messaging.Activities.Arguments;
 using Musify.Infrastructure.Messaging.Consumers;
+using Musify.Infrastructure.Messaging.Filters;

 namespace Musify.Infrastructure.Messaging
 {
@@ -34,6 +35,8 @@ namespace Musify.Infrastructure.Messaging
                {
                    ConfigureRabbitMqHost(busFactoryConfigurator, configuration);

+                    busFactoryConfigurator.UseConsumeFilter(typeof(ProcessTrackingConsumeFilter<>), busRegistrationContext);
+
                    busFactoryConfigurator.ReceiveEndpoint(UpdatePlayListPictureConsumer.QueueName, endpointConfigurator =>
                    {
                        endpointConfigurator.ConfigureConsumer<UpdatePlayListPictureConsumer>(busRegistrationContext);
@@ -23,8 +23,25 @@ namespace Musify.Infrastructure.Persistence

        public DbSet<Upload> Uploads => Set<Upload>();

-        public DbSet<Job> Jobs => Set<Job>();
+        public DbSet<ProcessExecution> ProcessExecutions => Set<ProcessExecution>();

-        public DbSet<JobOperation> JobOperations => Set<JobOperation>();
+        public DbSet<ProcessStepExecution> ProcessStepExecutions => Set<ProcessStepExecution>();
+
+        protected override void OnModelCreating(ModelBuilder modelBuilder)
+        {
+            modelBuilder.Entity<ProcessExecution>()
+                .HasIndex(process => new { process.CorrelationId, process.ProcessName });
+
+            modelBuilder.Entity<ProcessStepExecution>()
+                .HasIndex(step => new { step.ProcessExecutionId, step.StepName });
+
+            modelBuilder.Entity<ProcessExecution>()
+                .HasMany(process => process.Steps)
+                .WithOne(step => step.ProcessExecution)
+                .HasForeignKey(step => step.ProcessExecutionId)
+                .OnDelete(DeleteBehavior.Cascade);
+
+            base.OnModelCreating(modelBuilder);
+        }
    }
 }
@@ -0,0 +1,122 @@
+using Microsoft.EntityFrameworkCore;
+using Musify.Application.Contracts.Infrastructure;
+using Musify.Domain.Entities;
+
+namespace Musify.Infrastructure.Persistence
+{
+    public class ProcessTrackingStore(IDatabase database)
+        : IProcessTrackingStore
+    {
+        public async Task<Guid> GetOrCreateProcessAsync(
+            string processName,
+            Guid? correlationId,
+            Guid? conversationId,
+            Guid? messageId,
+            CancellationToken cancellationToken)
+        {
+            ProcessExecution? existingProcess = null;
+
+            if (correlationId.HasValue)
+            {
+                existingProcess = await database.ProcessExecutions
+                    .OrderByDescending(process => process.StartedDateTime)
+                    .FirstOrDefaultAsync(
+                        process => process.CorrelationId == correlationId && process.ProcessName == processName,
+                        cancellationToken);
+            }
+
+            if (existingProcess is not null)
+            {
+                return existingProcess.Id;
+            }
+
+            var newProcess = new ProcessExecution
+            {
+                ProcessName = processName,
+                CorrelationId = correlationId,
+                ConversationId = conversationId,
+                MessageId = messageId,
+                Status = ProcessExecutionStatus.Running,
+                StartedDateTime = DateTime.UtcNow,
+            };
+
+            await database.ProcessExecutions.AddAsync(newProcess, cancellationToken);
+            await database.SaveChangesAsync(cancellationToken);
+            return newProcess.Id;
+        }
+
+        public async Task<Guid> StartStepAsync(
+            Guid processExecutionId,
+            string stepName,
+            ProcessStepComponentType componentType,
+            int attempt,
+            CancellationToken cancellationToken)
+        {
+            var step = new ProcessStepExecution
+            {
+                ProcessExecutionId = processExecutionId,
+                StepName = stepName,
+                ComponentType = componentType,
+                Attempt = attempt,
+                Status = ProcessExecutionStatus.Running,
+                StartedDateTime = DateTime.UtcNow,
+            };
+
+            await database.ProcessStepExecutions.AddAsync(step, cancellationToken);
+            await database.SaveChangesAsync(cancellationToken);
+            return step.Id;
+        }
+
+        public async Task CompleteStepAsync(Guid processExecutionId, Guid stepExecutionId, CancellationToken cancellationToken)
+        {
+            var step = await database.ProcessStepExecutions
+                .SingleAsync(currentStep => currentStep.Id == stepExecutionId, cancellationToken);
+
+            step.Status = ProcessExecutionStatus.Succeeded;
+            step.FinishedDateTime = DateTime.UtcNow;
+
+            var process = await database.ProcessExecutions
+                .SingleAsync(currentProcess => currentProcess.Id == processExecutionId, cancellationToken);
+
+            var hasRunningSteps = await database.ProcessStepExecutions
+                .AnyAsync(
+                    currentStep => currentStep.ProcessExecutionId == processExecutionId
+                                   && currentStep.Status == ProcessExecutionStatus.Running,
+                    cancellationToken);
+
+            var hasFailedSteps = await database.ProcessStepExecutions
+                .AnyAsync(
+                    currentStep => currentStep.ProcessExecutionId == processExecutionId
+                                   && currentStep.Status == ProcessExecutionStatus.Failed,
+                    cancellationToken);
+
+            if (!hasRunningSteps && !hasFailedSteps)
+            {
+                process.Status = ProcessExecutionStatus.Succeeded;
+                process.FinishedDateTime = DateTime.UtcNow;
+                process.ErrorMessage = string.Empty;
+            }
+
+            await database.SaveChangesAsync(cancellationToken);
+        }
+
+        public async Task FailStepAsync(Guid processExecutionId, Guid stepExecutionId, string errorMessage, CancellationToken cancellationToken)
+        {
+            var step = await database.ProcessStepExecutions
+                .SingleAsync(currentStep => currentStep.Id == stepExecutionId, cancellationToken);
+
+            step.Status = ProcessExecutionStatus.Failed;
+            step.FinishedDateTime = DateTime.UtcNow;
+            step.ErrorMessage = errorMessage;
+
+            var process = await database.ProcessExecutions
+                .SingleAsync(currentProcess => currentProcess.Id == processExecutionId, cancellationToken);
+
+            process.Status = ProcessExecutionStatus.Failed;
+            process.FinishedDateTime = DateTime.UtcNow;
+            process.ErrorMessage = errorMessage;
+
+            await database.SaveChangesAsync(cancellationToken);
+        }
+    }
+}