This commit is contained in:
2026-03-20 11:57:16 +01:00
parent a10c8238f8
commit e02e1c5bc3
5 changed files with 46 additions and 46 deletions
@@ -17,7 +17,7 @@ namespace Musify.Application.Contracts.Infrastructure
DbSet<Job> Jobs { get; } DbSet<Job> Jobs { get; }
DbSet<JobExecution> JobExecutions { get; } DbSet<JobOperation> JobOperations { get; }
Task<int> SaveChangesAsync(CancellationToken cancellationToken); Task<int> SaveChangesAsync(CancellationToken cancellationToken);
} }
@@ -5,7 +5,7 @@ namespace Musify.Domain.Entities
{ {
#pragma warning disable CS8618 #pragma warning disable CS8618
[Table("JobExecution")] [Table("JobExecution")]
public class JobExecution public class JobOperation
{ {
[Key] [Key]
public Guid Id { get; set; } = Guid.NewGuid(); public Guid Id { get; set; } = Guid.NewGuid();
@@ -31,11 +31,11 @@ namespace Musify.Infrastructure.Messaging.Consumers
logger.LogInformation("Created new job with id {JobId} for removing file {BucketName}/{KeyName}", job.Id, consumeContext.Message.BucketName, consumeContext.Message.KeyName); logger.LogInformation("Created new job with id {JobId} for removing file {BucketName}/{KeyName}", job.Id, consumeContext.Message.BucketName, consumeContext.Message.KeyName);
} }
var jobExecution = new JobExecution var jobExecution = new JobOperation
{ {
JobId = job.Id JobId = job.Id
}; };
await database.JobExecutions.AddAsync(jobExecution, consumeContext.CancellationToken); await database.JobOperations.AddAsync(jobExecution, consumeContext.CancellationToken);
var removeFile = await storageHandler.RemoveFileAsync( var removeFile = await storageHandler.RemoveFileAsync(
consumeContext.Message.BucketName, consumeContext.Message.BucketName,
@@ -12,7 +12,7 @@ namespace Musify.Infrastructure.Messaging.Consumers
public class ResizePictureConsumer(IStorageHandler storageHandler, IPictureHandler pictureHandler, IDatabase database, ILogger<ResizePictureConsumer> logger) public class ResizePictureConsumer(IStorageHandler storageHandler, IPictureHandler pictureHandler, IDatabase database, ILogger<ResizePictureConsumer> logger)
: IConsumer<ResizePictureEvent> : IConsumer<ResizePictureEvent>
{ {
private record JobExecutionItem(ResizePictureItems Item, JobExecution JobExecution); private record JobOperationItem(ResizePictureItems Item, JobOperation JobOperation);
public const string QueueName = "picture-resize-queue"; public const string QueueName = "picture-resize-queue";
@@ -24,24 +24,24 @@ namespace Musify.Infrastructure.Messaging.Consumers
var cancellationToken = consumeContext.CancellationToken; var cancellationToken = consumeContext.CancellationToken;
var job = await GetOrCreateJobAsync(message, cancellationToken); var job = await GetOrCreateJobAsync(message, cancellationToken);
var jobExecutions = await CreateJobExecutionsAsync(message.Items, job.Id, cancellationToken); var jobOperations = await CreateJobOperationAsync(message.Items, job.Id, cancellationToken);
await database.SaveChangesAsync(cancellationToken); await database.SaveChangesAsync(cancellationToken);
var sourceImageResult = await GetSourceImageAsync(message.BucketName, message.KeyName, cancellationToken); var sourceImageResult = await GetSourceImageAsync(message.BucketName, message.KeyName, cancellationToken);
if (!sourceImageResult.IsSuccess) if (!sourceImageResult.IsSuccess)
{ {
MarkAllAsFailed(jobExecutions); MarkAllAsFailed(jobOperations);
await database.SaveChangesAsync(cancellationToken); await database.SaveChangesAsync(cancellationToken);
return; return;
} }
using var sourceImage = sourceImageResult.Value; using var sourceImage = sourceImageResult.Value;
foreach (var executionItem in jobExecutions) foreach (var operationItem in jobOperations)
{ {
await ProcessExecutionAsync( await ProcessOperationAsync(
executionItem, operationItem,
message.BucketName, message.BucketName,
sourceImage, sourceImage,
cancellationToken); cancellationToken);
@@ -78,25 +78,25 @@ namespace Musify.Infrastructure.Messaging.Consumers
return job; return job;
} }
private async Task<List<JobExecutionItem>> CreateJobExecutionsAsync( private async Task<List<JobOperationItem>> CreateJobOperationAsync(
IReadOnlyCollection<ResizePictureItems> items, IReadOnlyCollection<ResizePictureItems> resizeItems,
Guid jobId, Guid jobId,
CancellationToken cancellationToken) CancellationToken cancellationToken)
{ {
var jobExecutions = new List<JobExecutionItem>(items.Count); var jobOperations = new List<JobOperationItem>(resizeItems.Count);
foreach (var item in items) foreach (var resizeItem in resizeItems)
{ {
var jobExecution = new JobExecution var jobOperation = new JobOperation
{ {
JobId = jobId JobId = jobId
}; };
await database.JobExecutions.AddAsync(jobExecution, cancellationToken); await database.JobOperations.AddAsync(jobOperation, cancellationToken);
jobExecutions.Add(new JobExecutionItem(item, jobExecution)); jobOperations.Add(new JobOperationItem(resizeItem, jobOperation));
} }
return jobExecutions; return jobOperations;
} }
private async Task<Result<MemoryStream>> GetSourceImageAsync(string bucketName, string keyName, CancellationToken cancellationToken) private async Task<Result<MemoryStream>> GetSourceImageAsync(string bucketName, string keyName, CancellationToken cancellationToken)
@@ -126,32 +126,32 @@ namespace Musify.Infrastructure.Messaging.Consumers
return Result.Success(memoryStream); return Result.Success(memoryStream);
} }
private async Task ProcessExecutionAsync( private async Task ProcessOperationAsync(
JobExecutionItem executionItem, JobOperationItem operationItem,
string bucketName, string bucketName,
MemoryStream sourceImage, MemoryStream sourceImage,
CancellationToken cancellationToken) CancellationToken cancellationToken)
{ {
var jobExecution = executionItem.JobExecution; var jobOperation = operationItem.JobOperation;
try try
{ {
jobExecution.JobState = JobState.InProgress; jobOperation.JobState = JobState.InProgress;
sourceImage.Position = 0; sourceImage.Position = 0;
var resizedPictureResult = await pictureHandler.ResizePictureAsync( var resizedPictureResult = await pictureHandler.ResizePictureAsync(
sourceImage, sourceImage,
executionItem.Item.Width, operationItem.Item.Width,
executionItem.Item.Height, operationItem.Item.Height,
cancellationToken); cancellationToken);
if (!resizedPictureResult.IsSuccess) if (!resizedPictureResult.IsSuccess)
{ {
logger.LogError("Failed to resize picture for JobExecutionId={JobExecutionId}. Error: {ErrorMessage}", logger.LogError("Failed to resize picture for JobOperationId={JobOperationId}. Error: {ErrorMessage}",
jobExecution.Id, jobOperation.Id,
string.Join("; ", resizedPictureResult.Errors)); string.Join("; ", resizedPictureResult.Errors));
MarkAsFailed(jobExecution); MarkAsFailed(jobOperation);
return; return;
} }
@@ -159,47 +159,47 @@ namespace Musify.Infrastructure.Messaging.Consumers
resizedPictureResult.Value, resizedPictureResult.Value,
"image/webp", "image/webp",
bucketName, bucketName,
$"{executionItem.Item.SaveOnRoute}/{Guid.NewGuid()}.webp", $"{operationItem.Item.SaveOnRoute}/{Guid.NewGuid()}.webp",
cancellationToken); cancellationToken);
if (!saveResult.IsSuccess) if (!saveResult.IsSuccess)
{ {
logger.LogError("Failed to upload resized picture for JobExecutionId={JobExecutionId}. Error: {ErrorMessage}", logger.LogError("Failed to upload resized picture for JobOperationId={JobOperationId}. Error: {ErrorMessage}",
jobExecution.Id, jobOperation.Id,
string.Join("; ", saveResult.Errors)); string.Join("; ", saveResult.Errors));
MarkAsFailed(jobExecution); MarkAsFailed(jobOperation);
return; return;
} }
jobExecution.JobState = JobState.Completed; jobOperation.JobState = JobState.Completed;
jobExecution.FinishedAt = DateTime.UtcNow; jobOperation.FinishedAt = DateTime.UtcNow;
} }
catch (Exception exception) catch (Exception exception)
{ {
logger.LogError(exception, logger.LogError(exception,
"An error occurred while processing resize for JobExecutionId={JobExecutionId}, Width={Width}, Height={Height}, SaveOnRoute={SaveOnRoute}", "An error occurred while processing resize for JobOperationId={JobOperationId}, Width={Width}, Height={Height}, SaveOnRoute={SaveOnRoute}",
jobExecution.Id, jobOperation.Id,
executionItem.Item.Width, operationItem.Item.Width,
executionItem.Item.Height, operationItem.Item.Height,
executionItem.Item.SaveOnRoute); operationItem.Item.SaveOnRoute);
MarkAsFailed(jobExecution); MarkAsFailed(jobOperation);
} }
} }
private static void MarkAllAsFailed(IEnumerable<JobExecutionItem> jobExecutions) private static void MarkAllAsFailed(IEnumerable<JobOperationItem> jobOperations)
{ {
foreach (var executionItem in jobExecutions) foreach (var operationItem in jobOperations)
{ {
MarkAsFailed(executionItem.JobExecution); MarkAsFailed(operationItem.JobOperation);
} }
} }
private static void MarkAsFailed(JobExecution jobExecution) private static void MarkAsFailed(JobOperation jobOperation)
{ {
jobExecution.JobState = JobState.Failed; jobOperation.JobState = JobState.Failed;
jobExecution.FinishedAt = DateTime.UtcNow; jobOperation.FinishedAt = DateTime.UtcNow;
} }
} }
} }
+1 -1
View File
@@ -25,6 +25,6 @@ namespace Musify.Infrastructure.Persistence
public DbSet<Job> Jobs => Set<Job>(); public DbSet<Job> Jobs => Set<Job>();
public DbSet<JobExecution> JobExecutions => Set<JobExecution>(); public DbSet<JobOperation> JobOperations => Set<JobOperation>();
} }
} }