Remove processingStatus

This commit is contained in:
2026-05-13 21:56:34 +02:00
parent d0e6eaaee8
commit 7f0266d914
17 changed files with 7 additions and 445 deletions
@@ -17,10 +17,6 @@ namespace Musify.Application.Contracts.Infrastructure
DbSet<Upload> Uploads { get; } DbSet<Upload> Uploads { get; }
DbSet<ProcessExecution> ProcessExecutions { get; }
DbSet<ProcessStepExecution> ProcessStepExecutions { get; }
Task<int> SaveChangesAsync(CancellationToken cancellationToken); Task<int> SaveChangesAsync(CancellationToken cancellationToken);
} }
} }
@@ -1,25 +0,0 @@
using Musify.Domain.Entities;
namespace Musify.Application.Contracts.Infrastructure
{
public interface IProcessTrackingStore
{
Task<Guid> GetOrCreateProcessAsync(
string processName,
Guid? correlationId,
Guid? conversationId,
Guid? messageId,
CancellationToken cancellationToken);
Task<Guid> 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);
}
}
-36
View File
@@ -1,36 +0,0 @@
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<ProcessStepExecution> Steps { get; set; } = new List<ProcessStepExecution>();
}
}
-10
View File
@@ -1,10 +0,0 @@
namespace Musify.Domain.Entities
{
public enum ProcessExecutionStatus
{
Running,
Succeeded,
Failed,
Compensated
}
}
@@ -1,8 +0,0 @@
namespace Musify.Domain.Entities
{
public enum ProcessStepComponentType
{
Consumer,
Activity
}
}
-40
View File
@@ -1,40 +0,0 @@
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; }
}
}
@@ -1,7 +1,6 @@
using MassTransit; using MassTransit;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure; using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.MassTransit.Activities.Arguments; using Musify.Infrastructure.MassTransit.Activities.Arguments;
using Musify.Infrastructure.MassTransit.Activities.Logs; using Musify.Infrastructure.MassTransit.Activities.Logs;
@@ -9,28 +8,13 @@ namespace Musify.Infrastructure.MassTransit.Activities.Files
{ {
internal class DownloadFileFromBucketActivity( internal class DownloadFileFromBucketActivity(
IStorageService storageHandler, IStorageService storageHandler,
ILogger<DownloadFileFromBucketActivity> logger, ILogger<DownloadFileFromBucketActivity> logger)
IProcessTrackingStore processTrackingStore)
: IActivity<DownloadFileFromBucketArguments, DownloadFileFromBucketLog> : IActivity<DownloadFileFromBucketArguments, DownloadFileFromBucketLog>
{ {
public const string ExecuteEndpointName = "download-file-from-bucket"; public const string ExecuteEndpointName = "download-file-from-bucket";
public async Task<ExecutionResult> Execute(ExecuteContext<DownloadFileFromBucketArguments> executeContext) public async Task<ExecutionResult> Execute(ExecuteContext<DownloadFileFromBucketArguments> executeContext)
{ {
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(DownloadFileFromBucketActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
var destinationPath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable); var destinationPath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable);
ArgumentNullException.ThrowIfNull(destinationPath, nameof(destinationPath)); ArgumentNullException.ThrowIfNull(destinationPath, nameof(destinationPath));
@@ -50,8 +34,6 @@ namespace Musify.Infrastructure.MassTransit.Activities.Files
using var destinationStream = File.Create(destinationPath); using var destinationStream = File.Create(destinationPath);
await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken); await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed(new DownloadFileFromBucketLog( return executeContext.Completed(new DownloadFileFromBucketLog(
executeContext.Arguments.Bucket, executeContext.Arguments.Bucket,
executeContext.Arguments.Key, executeContext.Arguments.Key,
@@ -60,7 +42,6 @@ namespace Musify.Infrastructure.MassTransit.Activities.Files
catch (Exception exception) catch (Exception exception)
{ {
logger.LogError(exception, "Failed to download file from bucket"); logger.LogError(exception, "Failed to download file from bucket");
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw; throw;
} }
} }
@@ -1,7 +1,6 @@
using MassTransit; using MassTransit;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure; using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.MassTransit.Activities.Arguments; using Musify.Infrastructure.MassTransit.Activities.Arguments;
using Musify.Infrastructure.MassTransit.Activities.Logs; using Musify.Infrastructure.MassTransit.Activities.Logs;
@@ -9,28 +8,13 @@ namespace Musify.Infrastructure.MassTransit.Activities
{ {
public class TransferFilesToBucketActivity( public class TransferFilesToBucketActivity(
IStorageService storageHandler, IStorageService storageHandler,
ILogger<TransferFilesToBucketActivity> logger, ILogger<TransferFilesToBucketActivity> logger)
IProcessTrackingStore processTrackingStore)
: IActivity<TransferFilesToBucketArguments, TransferFilesToBucketLog> : IActivity<TransferFilesToBucketArguments, TransferFilesToBucketLog>
{ {
public const string ExecuteEndpointName = "transfer-files-to-bucket"; public const string ExecuteEndpointName = "transfer-files-to-bucket";
public async Task<ExecutionResult> Execute(ExecuteContext<TransferFilesToBucketArguments> executeContext) public async Task<ExecutionResult> Execute(ExecuteContext<TransferFilesToBucketArguments> executeContext)
{ {
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(TransferFilesToBucketActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
var folderPath = executeContext.GetVariable<string>(executeContext.Arguments.SourceDirectoryVariable); var folderPath = executeContext.GetVariable<string>(executeContext.Arguments.SourceDirectoryVariable);
ArgumentNullException.ThrowIfNull(folderPath, nameof(folderPath)); ArgumentNullException.ThrowIfNull(folderPath, nameof(folderPath));
var destinationKey = executeContext.Arguments.DestinationKey; var destinationKey = executeContext.Arguments.DestinationKey;
@@ -47,8 +31,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
executeContext.Arguments.DestinationBucket, executeContext.Arguments.DestinationBucket,
destinationKey, destinationKey,
executeContext.CancellationToken); executeContext.CancellationToken);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed(); return executeContext.Completed();
} }
catch (Exception exception) catch (Exception exception)
@@ -56,7 +38,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
logger.LogError(exception, "Failed to transfer files from {FolderPath} to {DestinationBucket}", logger.LogError(exception, "Failed to transfer files from {FolderPath} to {DestinationBucket}",
folderPath, folderPath,
executeContext.Arguments.DestinationBucket); executeContext.Arguments.DestinationBucket);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw; throw;
} }
} }
@@ -2,35 +2,19 @@ using MassTransit;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using MimeMapping; using MimeMapping;
using Musify.Application.Contracts.Infrastructure; using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs; using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Files namespace Musify.Infrastructure.MassTransit.Activities.Files
{ {
public class UploadFileToBucketActivity(IStorageService storageHandler, public class UploadFileToBucketActivity(IStorageService storageHandler,
ILogger<UploadFileToBucketActivity> logger, ILogger<UploadFileToBucketActivity> logger)
IProcessTrackingStore processTrackingStore)
: IActivity<UploadFileToBucketArguments, UploadFileToBucketLog> : IActivity<UploadFileToBucketArguments, UploadFileToBucketLog>
{ {
public const string ExecuteEndpointName = "upload-file-to-bucket"; public const string ExecuteEndpointName = "upload-file-to-bucket";
public async Task<ExecutionResult> Execute(ExecuteContext<UploadFileToBucketArguments> executeContext) public async Task<ExecutionResult> Execute(ExecuteContext<UploadFileToBucketArguments> executeContext)
{ {
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(UploadFileToBucketActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.FilePathVariable); var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.FilePathVariable);
ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath)); ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath));
var fileName = Path.GetFileName(sourceFilePath); var fileName = Path.GetFileName(sourceFilePath);
@@ -42,16 +26,13 @@ namespace Musify.Infrastructure.MassTransit.Activities.Files
executeContext.Arguments.DestinationBucket, executeContext.Arguments.DestinationBucket,
Path.Combine(executeContext.Arguments.DestinationRoute, fileName), Path.Combine(executeContext.Arguments.DestinationRoute, fileName),
executeContext.CancellationToken); executeContext.CancellationToken);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.CompletedWithVariables(new UploadFileToBucketLog( return executeContext.CompletedWithVariables(new UploadFileToBucketLog(
executeContext.Arguments.DestinationBucket, executeContext.Arguments.DestinationBucket,
executeContext.Arguments.DestinationRoute)); executeContext.Arguments.DestinationRoute));
} }
catch (Exception exception) catch (Exception exception)
{ {
logger.LogError(exception, "Failed to upload file to bucket for process {ProcessId} step {StepId}", processId, stepId); logger.LogError(exception, "Failed to upload file to bucket");
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw; throw;
} }
} }
@@ -1,14 +1,11 @@
using MassTransit; using MassTransit;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.MassTransit.Activities.Arguments; using Musify.Infrastructure.MassTransit.Activities.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities namespace Musify.Infrastructure.MassTransit.Activities
{ {
public class GeneratePictureWorkflowPathsActivity( public class GeneratePictureWorkflowPathsActivity(
ILogger<GeneratePictureWorkflowPathsActivity> logger, ILogger<GeneratePictureWorkflowPathsActivity> logger)
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<GeneratePictureWorkflowPathsArguments> : IExecuteActivity<GeneratePictureWorkflowPathsArguments>
{ {
public const string ExecuteEndpointName = "generate-picture-workflow-paths"; public const string ExecuteEndpointName = "generate-picture-workflow-paths";
@@ -17,20 +14,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
public async Task<ExecutionResult> Execute(ExecuteContext<GeneratePictureWorkflowPathsArguments> executeContext) public async Task<ExecutionResult> Execute(ExecuteContext<GeneratePictureWorkflowPathsArguments> executeContext)
{ {
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(GeneratePictureWorkflowPathsActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
try try
{ {
var destinationFolderName = Guid.NewGuid().ToString(); var destinationFolderName = Guid.NewGuid().ToString();
@@ -49,8 +32,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
logger.LogDebug("Generated picture workflow paths in {WorkingDirectory}", logger.LogDebug("Generated picture workflow paths in {WorkingDirectory}",
workingDirectory); workingDirectory);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.CompletedWithVariables(new Dictionary<string, object> return executeContext.CompletedWithVariables(new Dictionary<string, object>
{ {
[RoutingSlipVariableNames.Workflow.TemporalDirectory] = workingDirectory, [RoutingSlipVariableNames.Workflow.TemporalDirectory] = workingDirectory,
@@ -65,7 +46,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
{ {
logger.LogError(exception, "Failed to generate picture workflow paths for {SourceKey}", logger.LogError(exception, "Failed to generate picture workflow paths for {SourceKey}",
executeContext.Arguments.SourceKey); executeContext.Arguments.SourceKey);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw; throw;
} }
} }
@@ -1,7 +1,6 @@
using MassTransit; using MassTransit;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure; using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.MassTransit.Activities.Arguments; using Musify.Infrastructure.MassTransit.Activities.Arguments;
using Musify.Infrastructure.MassTransit.Activities.Logs; using Musify.Infrastructure.MassTransit.Activities.Logs;
using Musify.Infrastructure.MassTransit.RoutingSlip.Builders; using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
@@ -10,28 +9,13 @@ namespace Musify.Infrastructure.MassTransit.Activities
{ {
public class ResizePictureActivity( public class ResizePictureActivity(
IPictureService pictureHandler, IPictureService pictureHandler,
ILogger<ResizePictureActivity> logger, ILogger<ResizePictureActivity> logger)
IProcessTrackingStore processTrackingStore)
: IActivity<ResizePictureLocalArguments, ResizePictureLog> : IActivity<ResizePictureLocalArguments, ResizePictureLog>
{ {
public const string ExecuteEndpointName = "resize-picture"; public const string ExecuteEndpointName = "resize-picture";
public async Task<ExecutionResult> Execute(ExecuteContext<ResizePictureLocalArguments> executeContext) public async Task<ExecutionResult> Execute(ExecuteContext<ResizePictureLocalArguments> executeContext)
{ {
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(ResizePictureActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.SourceFilePathVariable); var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.SourceFilePathVariable);
ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath)); ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath));
var destinationFilePath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable); var destinationFilePath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable);
@@ -63,14 +47,11 @@ namespace Musify.Infrastructure.MassTransit.Activities
logger.LogDebug("Resized picture from {SourceFilePath} to {DestinationFilePath}", logger.LogDebug("Resized picture from {SourceFilePath} to {DestinationFilePath}",
sourceFilePath, sourceFilePath,
destinationFilePath); destinationFilePath);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed(new ResizePictureLog(destinationFilePath)); return executeContext.Completed(new ResizePictureLog(destinationFilePath));
} }
catch (Exception exception) catch (Exception exception)
{ {
logger.LogError(exception, "Failed to resize picture {SourceFilePath}", sourceFilePath); logger.LogError(exception, "Failed to resize picture {SourceFilePath}", sourceFilePath);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw; throw;
} }
} }
@@ -1,7 +1,6 @@
using MassTransit; using MassTransit;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure; using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.MassTransit.Activities.Arguments; using Musify.Infrastructure.MassTransit.Activities.Arguments;
using Musify.Infrastructure.MassTransit.Activities.Logs; using Musify.Infrastructure.MassTransit.Activities.Logs;
@@ -9,28 +8,13 @@ namespace Musify.Infrastructure.MassTransit.Activities
{ {
public class UpdatePlayListPictureActivity( public class UpdatePlayListPictureActivity(
IDatabase database, IDatabase database,
ILogger<UpdatePlayListPictureActivity> logger, ILogger<UpdatePlayListPictureActivity> logger)
IProcessTrackingStore processTrackingStore)
: IActivity<UpdatePlayListPictureArguments, UpdatePlayListPictureLog> : IActivity<UpdatePlayListPictureArguments, UpdatePlayListPictureLog>
{ {
public const string ExecuteEndpointName = "update-playlist-picture"; public const string ExecuteEndpointName = "update-playlist-picture";
public async Task<ExecutionResult> Execute(ExecuteContext<UpdatePlayListPictureArguments> executeContext) public async Task<ExecutionResult> Execute(ExecuteContext<UpdatePlayListPictureArguments> executeContext)
{ {
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(UpdatePlayListPictureActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
var smallResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.SmallPictureVariable); var smallResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.SmallPictureVariable);
ArgumentNullException.ThrowIfNull(smallResizedVariable, nameof(smallResizedVariable)); ArgumentNullException.ThrowIfNull(smallResizedVariable, nameof(smallResizedVariable));
var mediumResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.MediumPictureVariable); var mediumResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.MediumPictureVariable);
@@ -68,15 +52,12 @@ namespace Musify.Infrastructure.MassTransit.Activities
logger.LogInformation("Updated playlist {PlayListId} pictures", logger.LogInformation("Updated playlist {PlayListId} pictures",
executeContext.Arguments.PlayListId); executeContext.Arguments.PlayListId);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed(); return executeContext.Completed();
} }
catch (Exception exception) catch (Exception exception)
{ {
logger.LogError(exception, "Failed to update playlist {PlayListId} pictures", logger.LogError(exception, "Failed to update playlist {PlayListId} pictures",
executeContext.Arguments.PlayListId); executeContext.Arguments.PlayListId);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw; throw;
} }
} }
@@ -11,7 +11,6 @@ using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.Logs; using Musify.Infrastructure.MassTransit.Activities.Logs;
using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers; using Musify.Infrastructure.MassTransit.Consumers;
using Musify.Infrastructure.MassTransit.Filters;
using Musify.Infrastructure.MassTransit.Logs; using Musify.Infrastructure.MassTransit.Logs;
using Musify.Infrastructure.MassTransit.RoutingSlip.Builders; using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
using Musify.Infrastructure.MassTransit.Activities.Tracks; using Musify.Infrastructure.MassTransit.Activities.Tracks;
@@ -92,8 +91,6 @@ namespace Musify.Infrastructure.MassTransit
busFactoryConfigurator, busFactoryConfigurator,
busRegistrationContext.GetRequiredService<MassTransitConfiguration>()); busRegistrationContext.GetRequiredService<MassTransitConfiguration>());
busFactoryConfigurator.UseConsumeFilter(typeof(ProcessTrackingConsumeFilter<>), busRegistrationContext);
busFactoryConfigurator.ReceiveEndpoint(UpdatePlayListPictureConsumer.QueueName, endpointConfigurator => busFactoryConfigurator.ReceiveEndpoint(UpdatePlayListPictureConsumer.QueueName, endpointConfigurator =>
endpointConfigurator.ConfigureConsumer<UpdatePlayListPictureConsumer>(busRegistrationContext)); endpointConfigurator.ConfigureConsumer<UpdatePlayListPictureConsumer>(busRegistrationContext));
@@ -1,48 +0,0 @@
using MassTransit;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
namespace Musify.Infrastructure.MassTransit.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<>));
}
}
}
-21
View File
@@ -24,26 +24,5 @@ namespace Musify.Infrastructure.Persistence
public DbSet<PlayListHasTrack> PlayListHasTracks => Set<PlayListHasTrack>(); public DbSet<PlayListHasTrack> PlayListHasTracks => Set<PlayListHasTrack>();
public DbSet<Upload> Uploads => Set<Upload>(); public DbSet<Upload> Uploads => Set<Upload>();
public DbSet<ProcessExecution> ProcessExecutions => Set<ProcessExecution>();
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);
}
} }
} }
@@ -23,11 +23,5 @@ namespace Musify.Infrastructure.Persistence
return serviceDescriptors; return serviceDescriptors;
} }
public static IServiceCollection AddProcessTrackingStore(this IServiceCollection serviceDescriptors)
{
serviceDescriptors.AddScoped<IProcessTrackingStore, ProcessTrackingStore>();
return serviceDescriptors;
}
} }
} }
@@ -1,122 +0,0 @@
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);
}
}
}