diff --git a/Application/Contracts/Application/IEvent.cs b/Application/Contracts/Application/IEvent.cs deleted file mode 100644 index 1d495fb..0000000 --- a/Application/Contracts/Application/IEvent.cs +++ /dev/null @@ -1,7 +0,0 @@ -namespace Musify.Application.Contracts.Application -{ - public interface IEvent - { - Guid JobId { get; } - } -} \ No newline at end of file diff --git a/Application/Contracts/Infrastructure/IDatabase.cs b/Application/Contracts/Infrastructure/IDatabase.cs index 5b3cdf2..ab56211 100644 --- a/Application/Contracts/Infrastructure/IDatabase.cs +++ b/Application/Contracts/Infrastructure/IDatabase.cs @@ -15,9 +15,9 @@ namespace Musify.Application.Contracts.Infrastructure DbSet Uploads { get; } - DbSet Jobs { get; } + DbSet ProcessExecutions { get; } - DbSet JobOperations { get; } + DbSet ProcessStepExecutions { get; } Task SaveChangesAsync(CancellationToken cancellationToken); } diff --git a/Application/Contracts/Infrastructure/IProcessTrackingStore.cs b/Application/Contracts/Infrastructure/IProcessTrackingStore.cs new file mode 100644 index 0000000..97921ca --- /dev/null +++ b/Application/Contracts/Infrastructure/IProcessTrackingStore.cs @@ -0,0 +1,25 @@ +using Musify.Domain.Entities; + +namespace Musify.Application.Contracts.Infrastructure +{ + public interface IProcessTrackingStore + { + Task GetOrCreateProcessAsync( + string processName, + Guid? correlationId, + Guid? conversationId, + Guid? messageId, + CancellationToken cancellationToken); + + Task StartStepAsync( + Guid processExecutionId, + string stepName, + ProcessStepComponentType componentType, + int attempt, + CancellationToken cancellationToken); + + Task CompleteStepAsync(Guid processExecutionId, Guid stepExecutionId, CancellationToken cancellationToken); + + Task FailStepAsync(Guid processExecutionId, Guid stepExecutionId, string errorMessage, CancellationToken cancellationToken); + } +} diff --git a/Application/Events/RemoveFileEvent.cs b/Application/Events/RemoveFileEvent.cs index 1106536..88cbd0e 100644 --- a/Application/Events/RemoveFileEvent.cs +++ b/Application/Events/RemoveFileEvent.cs @@ -3,7 +3,6 @@ namespace Musify.Application.Events { public record RemoveFileEvent( - Guid JobId, string BucketName, - string KeyName) : IEvent; + string KeyName); } \ No newline at end of file diff --git a/Application/PlayLists/Commands/DeletePlayList/DeletePlayListCommandHandler.cs b/Application/PlayLists/Commands/DeletePlayList/DeletePlayListCommandHandler.cs index 1148783..8853bdd 100644 --- a/Application/PlayLists/Commands/DeletePlayList/DeletePlayListCommandHandler.cs +++ b/Application/PlayLists/Commands/DeletePlayList/DeletePlayListCommandHandler.cs @@ -70,7 +70,6 @@ namespace Musify.Application.PlayLists.Commands.DeletePlayList try { await eventBus.PublishAsync(new RemoveFileEvent( - Guid.NewGuid(), storageConfiguration.BucketName, keyName), cancellationToken); diff --git a/Application/PlayLists/Commands/UpdatePlayList/UpdatePlayListCommandHandler.cs b/Application/PlayLists/Commands/UpdatePlayList/UpdatePlayListCommandHandler.cs index 5791661..8225c26 100644 --- a/Application/PlayLists/Commands/UpdatePlayList/UpdatePlayListCommandHandler.cs +++ b/Application/PlayLists/Commands/UpdatePlayList/UpdatePlayListCommandHandler.cs @@ -1,6 +1,5 @@ using Ardalis.Result; using DispatchR.Abstractions.Send; -using MassTransit; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; using Musify.Application.Configuration; @@ -54,7 +53,6 @@ namespace Musify.Application.PlayLists.Commands.UpdatePlayList try { await eventBus.PublishAsync(new RemoveFileEvent( - Guid.NewGuid(), storageConfiguration.BucketName, originalImageStorageKey), cancellationToken); } @@ -72,7 +70,6 @@ namespace Musify.Application.PlayLists.Commands.UpdatePlayList try { await eventBus.PublishAsync(new RemoveFileEvent( - Guid.NewGuid(), storageConfiguration.BucketName, playList.SmallPictureKeyName), cancellationToken); } @@ -88,7 +85,6 @@ namespace Musify.Application.PlayLists.Commands.UpdatePlayList try { await eventBus.PublishAsync(new RemoveFileEvent( - Guid.NewGuid(), storageConfiguration.BucketName, playList.MediumPictureKeyName), cancellationToken); } @@ -104,7 +100,6 @@ namespace Musify.Application.PlayLists.Commands.UpdatePlayList try { await eventBus.PublishAsync(new RemoveFileEvent( - Guid.NewGuid(), storageConfiguration.BucketName, playList.LargePictureKeyName), cancellationToken); } diff --git a/Domain/Entities/Job.cs b/Domain/Entities/Job.cs deleted file mode 100644 index 45f07dc..0000000 --- a/Domain/Entities/Job.cs +++ /dev/null @@ -1,22 +0,0 @@ -using System.ComponentModel.DataAnnotations; -using System.ComponentModel.DataAnnotations.Schema; - -namespace Musify.Domain.Entities -{ - [Table("Job")] - public class Job - { - [Key] - public Guid Id { get; set; } = Guid.NewGuid(); - - [Required] - public required JobType JobType { get; set; } - - [Required] - [MaxLength(512)] - public required string Payload { get; set; } - - [Required] - public DateTime CreateAt { get; set; } = DateTime.UtcNow; - } -} \ No newline at end of file diff --git a/Domain/Entities/JobOperation.cs b/Domain/Entities/JobOperation.cs deleted file mode 100644 index 3e14c26..0000000 --- a/Domain/Entities/JobOperation.cs +++ /dev/null @@ -1,41 +0,0 @@ -using System.ComponentModel.DataAnnotations; -using System.ComponentModel.DataAnnotations.Schema; - -namespace Musify.Domain.Entities -{ -#pragma warning disable CS8618 - [Table("JobExecution")] - public class JobOperation - { - [Key] - public Guid Id { get; set; } = Guid.NewGuid(); - - [Required] - public required Guid JobId { get; set; } - - [Required] - public JobState JobState { get; set; } = JobState.Pending; - - [NotMapped] - public bool IsCompleted => JobState == JobState.Completed; - - [NotMapped] - public bool IsFailed => JobState == JobState.Failed; - - [NotMapped] - public bool IsPending => JobState == JobState.Pending; - - [Required] - [MaxLength(512)] - public string ErrorMessage { get; set; } = string.Empty; - - [Required] - public DateTime CreatedAt { get; set; } = DateTime.UtcNow; - - [Required] - public DateTime FinishedAt { get; set; } - - [ForeignKey(nameof(JobId)) ] - public Job Job { get; set; } - } -} \ No newline at end of file diff --git a/Domain/Entities/JobState.cs b/Domain/Entities/JobState.cs deleted file mode 100644 index 9cc6cde..0000000 --- a/Domain/Entities/JobState.cs +++ /dev/null @@ -1,10 +0,0 @@ -namespace Musify.Domain.Entities -{ - public enum JobState - { - Pending, - Failed, - InProgress, - Completed - } -} \ No newline at end of file diff --git a/Domain/Entities/JobType.cs b/Domain/Entities/JobType.cs deleted file mode 100644 index c0ded2f..0000000 --- a/Domain/Entities/JobType.cs +++ /dev/null @@ -1,8 +0,0 @@ -namespace Musify.Domain.Entities -{ - public enum JobType - { - ResizePicture, - RemoveFile, - } -} \ No newline at end of file diff --git a/Domain/Entities/ProcessExecution.cs b/Domain/Entities/ProcessExecution.cs new file mode 100644 index 0000000..b0bbf21 --- /dev/null +++ b/Domain/Entities/ProcessExecution.cs @@ -0,0 +1,36 @@ +using System.ComponentModel.DataAnnotations; +using System.ComponentModel.DataAnnotations.Schema; + +namespace Musify.Domain.Entities +{ +#pragma warning disable CS8618 + [Table("ProcessExecution")] + public class ProcessExecution + { + [Key] + public Guid Id { get; set; } = Guid.NewGuid(); + + public Guid? CorrelationId { get; set; } + + public Guid? ConversationId { get; set; } + + public Guid? MessageId { get; set; } + + [Required] + [MaxLength(256)] + public required string ProcessName { get; set; } + + [Required] + public ProcessExecutionStatus Status { get; set; } = ProcessExecutionStatus.Running; + + [Required] + public DateTime StartedDateTime { get; set; } = DateTime.UtcNow; + + public DateTime FinishedDateTime { get; set; } + + [MaxLength(2048)] + public string ErrorMessage { get; set; } + + public ICollection Steps { get; set; } = new List(); + } +} diff --git a/Domain/Entities/ProcessExecutionStatus.cs b/Domain/Entities/ProcessExecutionStatus.cs new file mode 100644 index 0000000..b609694 --- /dev/null +++ b/Domain/Entities/ProcessExecutionStatus.cs @@ -0,0 +1,10 @@ +namespace Musify.Domain.Entities +{ + public enum ProcessExecutionStatus + { + Running, + Succeeded, + Failed, + Compensated + } +} diff --git a/Domain/Entities/ProcessStepComponentType.cs b/Domain/Entities/ProcessStepComponentType.cs new file mode 100644 index 0000000..43c4848 --- /dev/null +++ b/Domain/Entities/ProcessStepComponentType.cs @@ -0,0 +1,8 @@ +namespace Musify.Domain.Entities +{ + public enum ProcessStepComponentType + { + Consumer, + Activity + } +} diff --git a/Domain/Entities/ProcessStepExecution.cs b/Domain/Entities/ProcessStepExecution.cs new file mode 100644 index 0000000..a1d01cc --- /dev/null +++ b/Domain/Entities/ProcessStepExecution.cs @@ -0,0 +1,40 @@ +using System.ComponentModel.DataAnnotations; +using System.ComponentModel.DataAnnotations.Schema; + +namespace Musify.Domain.Entities +{ +#pragma warning disable CS8618 + [Table("ProcessStepExecution")] + public class ProcessStepExecution + { + [Key] + public Guid Id { get; set; } = Guid.NewGuid(); + + [Required] + public Guid ProcessExecutionId { get; set; } + + [Required] + [MaxLength(256)] + public required string StepName { get; set; } + + [Required] + public ProcessStepComponentType ComponentType { get; set; } + + [Required] + public ProcessExecutionStatus Status { get; set; } = ProcessExecutionStatus.Running; + + [Required] + public int Attempt { get; set; } + + [Required] + public DateTime StartedDateTime { get; set; } = DateTime.UtcNow; + + public DateTime FinishedDateTime { get; set; } + + [MaxLength(2048)] + public string ErrorMessage { get; set; } + + [ForeignKey(nameof(ProcessExecutionId)) ] + public ProcessExecution ProcessExecution { get; set; } + } +} diff --git a/Infrastructure/InfrastructureDependencyInjection.cs b/Infrastructure/InfrastructureDependencyInjection.cs index 5eddee6..6588275 100644 --- a/Infrastructure/InfrastructureDependencyInjection.cs +++ b/Infrastructure/InfrastructureDependencyInjection.cs @@ -21,7 +21,8 @@ namespace Musify.Infrastructure .AddStorageHandler() .AddPictureHandler() .AddMassTransitClient(configuration) - .AddScoped(); + .AddScoped() + .AddScoped(); return serviceDescriptors; } @@ -34,7 +35,8 @@ namespace Musify.Infrastructure .AddStorageHandler() .AddPictureHandler() .AddMassTransitConsumers(configuration) - .AddScoped(); + .AddScoped() + .AddScoped(); return serviceDescriptors; } diff --git a/Infrastructure/Messaging/Activities/RemoveFilesActivity.cs b/Infrastructure/Messaging/Activities/RemoveFilesActivity.cs index 4a4479d..fd0b6b8 100644 --- a/Infrastructure/Messaging/Activities/RemoveFilesActivity.cs +++ b/Infrastructure/Messaging/Activities/RemoveFilesActivity.cs @@ -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 logger) + public class RemoveFilesActivity( + IStorageHandler storageHandler, + ILogger logger, + IProcessTrackingStore processTrackingStore) : IExecuteActivity { public const string ExecuteEndpointName = "Remove-File"; public async Task Execute(ExecuteContext 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; + } } } } \ No newline at end of file diff --git a/Infrastructure/Messaging/Activities/ResizePictureActivity.cs b/Infrastructure/Messaging/Activities/ResizePictureActivity.cs index 5e75db4..d28292e 100644 --- a/Infrastructure/Messaging/Activities/ResizePictureActivity.cs +++ b/Infrastructure/Messaging/Activities/ResizePictureActivity.cs @@ -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 { public const string ExecuteEndpointName = "Resize-Picture"; public async Task Execute(ExecuteContext 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(); } } } \ No newline at end of file diff --git a/Infrastructure/Messaging/Activities/UpdatePlayListPictureActivity.cs b/Infrastructure/Messaging/Activities/UpdatePlayListPictureActivity.cs index 0d5ea4b..cbd4c1b 100644 --- a/Infrastructure/Messaging/Activities/UpdatePlayListPictureActivity.cs +++ b/Infrastructure/Messaging/Activities/UpdatePlayListPictureActivity.cs @@ -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 { public const string ExecuteEndpointName = "Update-PlayList-Picture"; public async Task Execute(ExecuteContext 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; + } } } } \ No newline at end of file diff --git a/Infrastructure/Messaging/Filters/ProcessTrackingConsumeFilter.cs b/Infrastructure/Messaging/Filters/ProcessTrackingConsumeFilter.cs new file mode 100644 index 0000000..3df0b38 --- /dev/null +++ b/Infrastructure/Messaging/Filters/ProcessTrackingConsumeFilter.cs @@ -0,0 +1,48 @@ +using MassTransit; +using Musify.Application.Contracts.Infrastructure; +using Musify.Domain.Entities; + +namespace Musify.Infrastructure.Messaging.Filters +{ + public class ProcessTrackingConsumeFilter(IProcessTrackingStore processTrackingStore) + : IFilter> + where TMessage : class + { + public async Task Send(ConsumeContext context, IPipe> 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)); + } + } +} diff --git a/Infrastructure/Messaging/MassTransitDependencyInjection.cs b/Infrastructure/Messaging/MassTransitDependencyInjection.cs index 55b85b1..ae5a99b 100644 --- a/Infrastructure/Messaging/MassTransitDependencyInjection.cs +++ b/Infrastructure/Messaging/MassTransitDependencyInjection.cs @@ -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(busRegistrationContext); diff --git a/Infrastructure/Persistence/Database.cs b/Infrastructure/Persistence/Database.cs index 4dd8f61..5676339 100644 --- a/Infrastructure/Persistence/Database.cs +++ b/Infrastructure/Persistence/Database.cs @@ -23,8 +23,25 @@ namespace Musify.Infrastructure.Persistence public DbSet Uploads => Set(); - public DbSet Jobs => Set(); + public DbSet ProcessExecutions => Set(); - public DbSet JobOperations => Set(); + public DbSet ProcessStepExecutions => Set(); + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.Entity() + .HasIndex(process => new { process.CorrelationId, process.ProcessName }); + + modelBuilder.Entity() + .HasIndex(step => new { step.ProcessExecutionId, step.StepName }); + + modelBuilder.Entity() + .HasMany(process => process.Steps) + .WithOne(step => step.ProcessExecution) + .HasForeignKey(step => step.ProcessExecutionId) + .OnDelete(DeleteBehavior.Cascade); + + base.OnModelCreating(modelBuilder); + } } } \ No newline at end of file diff --git a/Infrastructure/Persistence/ProcessTrackingStore.cs b/Infrastructure/Persistence/ProcessTrackingStore.cs new file mode 100644 index 0000000..3636609 --- /dev/null +++ b/Infrastructure/Persistence/ProcessTrackingStore.cs @@ -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 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 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); + } + } +}