From 79a633e5e004444462fac6f0ad3e60e65c820d9d Mon Sep 17 00:00:00 2001 From: XShine555 Date: Fri, 20 Mar 2026 12:10:54 +0100 Subject: [PATCH] Refactor --- .../Messaging/Consumers/RemoveFileConsumer.cs | 120 +++++++++++++----- .../Consumers/ResizePictureConsumer.cs | 16 +-- 2 files changed, 98 insertions(+), 38 deletions(-) diff --git a/Infrastructure/Messaging/Consumers/RemoveFileConsumer.cs b/Infrastructure/Messaging/Consumers/RemoveFileConsumer.cs index 1f71d37..ed33dbb 100644 --- a/Infrastructure/Messaging/Consumers/RemoveFileConsumer.cs +++ b/Infrastructure/Messaging/Consumers/RemoveFileConsumer.cs @@ -15,47 +15,109 @@ namespace Musify.Infrastructure.Messaging.Consumers public async Task Consume(ConsumeContext consumeContext) { - var payload = JsonSerializer.Serialize(consumeContext.Message); + var message = consumeContext.Message; + var cancellationToken = consumeContext.CancellationToken; - var job = await database.Jobs - .SingleOrDefaultAsync(j => j.Id == consumeContext.Message.JobId); + var job = await GetOrCreateJobAsync(message, cancellationToken); + var jobOperation = await CreateJobOperationAsync(job.Id, cancellationToken); + await database.SaveChangesAsync(cancellationToken); - if (job is null) + try { - job = new Job + jobOperation.JobState = JobState.InProgress; + await database.SaveChangesAsync(cancellationToken); + + var removeResult = await storageHandler.RemoveFileAsync( + message.BucketName, + message.KeyName, + cancellationToken); + + if (!removeResult.IsSuccess) { - JobType = JobType.RemoveFile, - Payload = payload - }; - await database.Jobs.AddAsync(job, consumeContext.CancellationToken); - logger.LogInformation("Created new job with id {JobId} for removing file {BucketName}/{KeyName}", job.Id, consumeContext.Message.BucketName, consumeContext.Message.KeyName); + MarkAsFailed(jobOperation, string.Join("; ", removeResult.Errors)); + + logger.LogError( + "Failed to remove file {BucketName}/{KeyName} for JobId={JobId}, JobOperationId={JobOperationId}. Error: {ErrorMessage}", + message.BucketName, + message.KeyName, + job.Id, + jobOperation.Id, + jobOperation.ErrorMessage); + + await database.SaveChangesAsync(cancellationToken); + return; + } + + jobOperation.JobState = JobState.Completed; + jobOperation.FinishedAt = DateTime.UtcNow; + + logger.LogInformation( + "Successfully removed file {BucketName}/{KeyName} for JobId={JobId}, JobOperationId={JobOperationId}.", + message.BucketName, + message.KeyName, + job.Id, + jobOperation.Id); + + await database.SaveChangesAsync(cancellationToken); + } + catch (Exception exception) + { + logger.LogError(exception, + "An error occurred while consuming RemoveFileConsumer message for JobId={JobId}, BucketName={BucketName}, KeyName={KeyName}", + message.JobId, + message.BucketName, + message.KeyName); + MarkAsFailed(jobOperation, exception.Message); + await database.SaveChangesAsync(cancellationToken); + + throw; + } + } + + async Task GetOrCreateJobAsync(RemoveFileEvent message, CancellationToken cancellationToken) + { + var job = await database.Jobs + .SingleOrDefaultAsync(j => j.Id == message.JobId, cancellationToken); + + if (job is not null) + { + return job; } - var jobExecution = new JobOperation + job = new Job { - JobId = job.Id + Id = message.JobId, + JobType = JobType.RemoveFile, + Payload = JsonSerializer.Serialize(message) }; - await database.JobOperations.AddAsync(jobExecution, consumeContext.CancellationToken); - var removeFile = await storageHandler.RemoveFileAsync( - consumeContext.Message.BucketName, - consumeContext.Message.KeyName, - consumeContext.CancellationToken); + await database.Jobs.AddAsync(job, cancellationToken); - if (!removeFile.IsSuccess) + logger.LogInformation( + "Created new job with id {JobId} for removing file {BucketName}/{KeyName}", + job.Id, + message.BucketName, + message.KeyName); + + return job; + } + + async Task CreateJobOperationAsync(Guid jobId, CancellationToken cancellationToken) + { + var jobOperation = new JobOperation { - jobExecution.JobState = JobState.Failed; - logger.LogInformation("Failed to remove file {BucketName}/{KeyName} for jobId={JobId}, JobExecutionId={jobExecutionId}. Error: {ErrorMessage}", - consumeContext.Message.BucketName, consumeContext.Message.KeyName, job.Id, jobExecution.Id, string.Join("; ", removeFile.Errors)); - } - else - { - logger.LogInformation("Successfully removed file {BucketName}/{KeyName} for jobId={JobId}, JobExecutionId={jobExecutionId}.", - consumeContext.Message.BucketName, consumeContext.Message.KeyName, job.Id, jobExecution.Id); - } + JobId = jobId + }; - jobExecution.FinishedAt = DateTime.UtcNow; - await database.SaveChangesAsync(consumeContext.CancellationToken); + await database.JobOperations.AddAsync(jobOperation, cancellationToken); + return jobOperation; + } + + void MarkAsFailed(JobOperation jobOperation, string errorMessage) + { + jobOperation.JobState = JobState.Failed; + jobOperation.ErrorMessage = errorMessage; + jobOperation.FinishedAt = DateTime.UtcNow; } } } \ No newline at end of file diff --git a/Infrastructure/Messaging/Consumers/ResizePictureConsumer.cs b/Infrastructure/Messaging/Consumers/ResizePictureConsumer.cs index bc261f9..af9a40b 100644 --- a/Infrastructure/Messaging/Consumers/ResizePictureConsumer.cs +++ b/Infrastructure/Messaging/Consumers/ResizePictureConsumer.cs @@ -56,7 +56,7 @@ namespace Musify.Infrastructure.Messaging.Consumers } } - private async Task GetOrCreateJobAsync(ResizePictureEvent message, CancellationToken cancellationToken) + async Task GetOrCreateJobAsync(ResizePictureEvent message, CancellationToken cancellationToken) { var job = await database.Jobs .SingleOrDefaultAsync(j => j.Id == message.JobId, cancellationToken); @@ -78,9 +78,7 @@ namespace Musify.Infrastructure.Messaging.Consumers return job; } - private async Task> CreateJobOperationAsync( - IReadOnlyCollection resizeItems, - Guid jobId, + async Task> CreateJobOperationAsync(IReadOnlyCollection resizeItems, Guid jobId, CancellationToken cancellationToken) { var jobOperations = new List(resizeItems.Count); @@ -99,7 +97,7 @@ namespace Musify.Infrastructure.Messaging.Consumers return jobOperations; } - private async Task> GetSourceImageAsync(string bucketName, string keyName, CancellationToken cancellationToken) + async Task> GetSourceImageAsync(string bucketName, string keyName, CancellationToken cancellationToken) { var fileResult = await storageHandler.GetFileAsync(bucketName, keyName, cancellationToken); @@ -126,7 +124,7 @@ namespace Musify.Infrastructure.Messaging.Consumers return Result.Success(memoryStream); } - private async Task ProcessOperationAsync( + async Task ProcessOperationAsync( JobOperationItem operationItem, string bucketName, MemoryStream sourceImage, @@ -160,7 +158,7 @@ namespace Musify.Infrastructure.Messaging.Consumers resizedPictureResult.Value, "image/webp", bucketName, - $"{operationItem.Item.SaveOnRoute}/{Guid.NewGuid()}.webp", + $"{operationItem.Item.SaveOnRoute}/{Guid.NewGuid() }.webp", cancellationToken); if (!saveResult.IsSuccess) @@ -190,7 +188,7 @@ namespace Musify.Infrastructure.Messaging.Consumers } } - private static void MarkAllAsFailed(IEnumerable jobOperations, string errorMessage) + static void MarkAllAsFailed(IEnumerable jobOperations, string errorMessage) { foreach (var operationItem in jobOperations) { @@ -198,7 +196,7 @@ namespace Musify.Infrastructure.Messaging.Consumers } } - private static void MarkAsFailed(JobOperation jobOperation, string errorMessage) + static void MarkAsFailed(JobOperation jobOperation, string errorMessage) { jobOperation.JobState = JobState.Failed; jobOperation.FinishedAt = DateTime.UtcNow;