Refactor

Infrastructure/Messaging/Consumers/RemoveFileConsumer.cs

@@ -15,47 +15,109 @@ namespace Musify.Infrastructure.Messaging.Consumers
 
         public async Task Consume(ConsumeContext<RemoveFileEvent> consumeContext)
         {
-            var payload = JsonSerializer.Serialize(consumeContext.Message);
+            var message = consumeContext.Message;
+            var cancellationToken = consumeContext.CancellationToken;
 
-            var job = await database.Jobs
+            var job = await GetOrCreateJobAsync(message, cancellationToken);
-                .SingleOrDefaultAsync(j => j.Id == consumeContext.Message.JobId);
+            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,
+                    MarkAsFailed(jobOperation, string.Join("; ", removeResult.Errors));
-                    Payload = payload
+
-                };
+                    logger.LogError(
-                await database.Jobs.AddAsync(job, consumeContext.CancellationToken);
+                        "Failed to remove file {BucketName}/{KeyName} for JobId={JobId}, JobOperationId={JobOperationId}. Error: {ErrorMessage}",
-                logger.LogInformation("Created new job with id {JobId} for removing file {BucketName}/{KeyName}", job.Id, consumeContext.Message.BucketName, consumeContext.Message.KeyName);
+                        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<Job> 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(
+            await database.Jobs.AddAsync(job, cancellationToken);
-                consumeContext.Message.BucketName,
-                consumeContext.Message.KeyName,
-                consumeContext.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<JobOperation> CreateJobOperationAsync(Guid jobId, CancellationToken cancellationToken)
+        {
+            var jobOperation = new JobOperation
             {
-                jobExecution.JobState = JobState.Failed;
+                JobId = jobId
-                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);
-            }
 
-            jobExecution.FinishedAt = DateTime.UtcNow;
+            await database.JobOperations.AddAsync(jobOperation, cancellationToken);
-            await database.SaveChangesAsync(consumeContext.CancellationToken);
+            return jobOperation;
+        }
+
+        void MarkAsFailed(JobOperation jobOperation, string errorMessage)
+        {
+            jobOperation.JobState = JobState.Failed;
+            jobOperation.ErrorMessage = errorMessage;
+            jobOperation.FinishedAt = DateTime.UtcNow;
         }
     }
 }

Infrastructure/Messaging/Consumers/ResizePictureConsumer.cs

@@ -56,7 +56,7 @@ namespace Musify.Infrastructure.Messaging.Consumers
             }
         }
 
-        private async Task<Job> GetOrCreateJobAsync(ResizePictureEvent message, CancellationToken cancellationToken)
+        async Task<Job> 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<List<JobOperationItem>> CreateJobOperationAsync(
+        async Task<List<JobOperationItem>> CreateJobOperationAsync(IReadOnlyCollection<ResizePictureItems> resizeItems, Guid jobId,
-            IReadOnlyCollection<ResizePictureItems> resizeItems,
-            Guid jobId,
             CancellationToken cancellationToken)
         {
             var jobOperations = new List<JobOperationItem>(resizeItems.Count);
@@ -99,7 +97,7 @@ namespace Musify.Infrastructure.Messaging.Consumers
             return jobOperations;
         }
 
-        private async Task<Result<MemoryStream>> GetSourceImageAsync(string bucketName, string keyName, CancellationToken cancellationToken)
+        async Task<Result<MemoryStream>> 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<JobOperationItem> jobOperations, string errorMessage)
+        static void MarkAllAsFailed(IEnumerable<JobOperationItem> 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;