Changes
This commit is contained in:
@@ -1,123 +0,0 @@
|
||||
using MassTransit;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Contracts.Infrastructure;
|
||||
using Musify.Application.Events;
|
||||
using Musify.Domain.Entities;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace Musify.Infrastructure.Messaging.Consumers
|
||||
{
|
||||
public class RemoveFileConsumer(IDatabase database, IStorageHandler storageHandler, ILogger<RemoveFileConsumer> logger)
|
||||
: IConsumer<RemoveFileEvent>
|
||||
{
|
||||
public const string QueueName = "file-remove-queue";
|
||||
|
||||
public async Task Consume(ConsumeContext<RemoveFileEvent> consumeContext)
|
||||
{
|
||||
var message = consumeContext.Message;
|
||||
var cancellationToken = consumeContext.CancellationToken;
|
||||
|
||||
var job = await GetOrCreateJobAsync(message, cancellationToken);
|
||||
var jobOperation = await CreateJobOperationAsync(job.Id, cancellationToken);
|
||||
await database.SaveChangesAsync(cancellationToken);
|
||||
|
||||
try
|
||||
{
|
||||
jobOperation.JobState = JobState.InProgress;
|
||||
await database.SaveChangesAsync(cancellationToken);
|
||||
|
||||
var removeResult = await storageHandler.RemoveFileAsync(
|
||||
message.BucketName,
|
||||
message.KeyName,
|
||||
cancellationToken);
|
||||
|
||||
if (!removeResult.IsSuccess)
|
||||
{
|
||||
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<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;
|
||||
}
|
||||
|
||||
job = new Job
|
||||
{
|
||||
Id = message.JobId,
|
||||
JobType = JobType.RemoveFile,
|
||||
Payload = JsonSerializer.Serialize(message)
|
||||
};
|
||||
|
||||
await database.Jobs.AddAsync(job, cancellationToken);
|
||||
|
||||
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
|
||||
{
|
||||
JobId = jobId
|
||||
};
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,206 +0,0 @@
|
||||
using Ardalis.Result;
|
||||
using MassTransit;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Contracts.Infrastructure;
|
||||
using Musify.Application.Events;
|
||||
using Musify.Domain.Entities;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace Musify.Infrastructure.Messaging.Consumers
|
||||
{
|
||||
public class ResizePictureConsumer(IStorageHandler storageHandler, IPictureHandler pictureHandler, IDatabase database, ILogger<ResizePictureConsumer> logger)
|
||||
: IConsumer<ResizePictureEvent>
|
||||
{
|
||||
private record JobOperationItem(ResizePictureArguments Item, JobOperation JobOperation);
|
||||
|
||||
public const string QueueName = "picture-resize-queue";
|
||||
|
||||
public async Task Consume(ConsumeContext<ResizePictureEvent> consumeContext)
|
||||
{
|
||||
try
|
||||
{
|
||||
var message = consumeContext.Message;
|
||||
var cancellationToken = consumeContext.CancellationToken;
|
||||
|
||||
var job = await GetOrCreateJobAsync(message, cancellationToken);
|
||||
var jobOperations = await CreateJobOperationAsync(message.Arguments, job.Id, cancellationToken);
|
||||
|
||||
await database.SaveChangesAsync(cancellationToken);
|
||||
|
||||
var sourceImageResult = await GetSourceImageAsync(message.BucketName, message.KeyName, cancellationToken);
|
||||
if (!sourceImageResult.IsSuccess)
|
||||
{
|
||||
MarkAllAsFailed(jobOperations, $"Failed to retrieve source image from storage, ResultState: {sourceImageResult.Status}");
|
||||
await database.SaveChangesAsync(cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
using var sourceImage = sourceImageResult.Value;
|
||||
|
||||
foreach (var operationItem in jobOperations)
|
||||
{
|
||||
await ProcessOperationAsync(
|
||||
operationItem,
|
||||
message.BucketName,
|
||||
sourceImage,
|
||||
cancellationToken);
|
||||
|
||||
await database.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "An error occurred while consuming the ResizePictureConsumer message.");
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
async Task<Job> GetOrCreateJobAsync(ResizePictureEvent message, CancellationToken cancellationToken)
|
||||
{
|
||||
var job = await database.Jobs
|
||||
.SingleOrDefaultAsync(j => j.Id == message.JobId, cancellationToken);
|
||||
|
||||
if (job is not null)
|
||||
{
|
||||
return job;
|
||||
}
|
||||
|
||||
var payloadJson = JsonSerializer.Serialize(message.Arguments);
|
||||
job = new Job
|
||||
{
|
||||
Id = message.JobId,
|
||||
JobType = JobType.ResizePicture,
|
||||
Payload = payloadJson
|
||||
};
|
||||
|
||||
await database.Jobs.AddAsync(job, cancellationToken);
|
||||
return job;
|
||||
}
|
||||
|
||||
async Task<List<JobOperationItem>> CreateJobOperationAsync(IReadOnlyCollection<ResizePictureArguments> resizeItems, Guid jobId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var jobOperations = new List<JobOperationItem>(resizeItems.Count);
|
||||
|
||||
foreach (var resizeItem in resizeItems)
|
||||
{
|
||||
var jobOperation = new JobOperation
|
||||
{
|
||||
JobId = jobId
|
||||
};
|
||||
|
||||
await database.JobOperations.AddAsync(jobOperation, cancellationToken);
|
||||
jobOperations.Add(new JobOperationItem(resizeItem, jobOperation));
|
||||
}
|
||||
|
||||
return jobOperations;
|
||||
}
|
||||
|
||||
async Task<Result<MemoryStream>> GetSourceImageAsync(string bucketName, string keyName, CancellationToken cancellationToken)
|
||||
{
|
||||
var fileResult = await storageHandler.GetFileAsync(bucketName, keyName, cancellationToken);
|
||||
|
||||
if (fileResult.IsNotFound())
|
||||
{
|
||||
logger.LogWarning("File not found in storage: BucketName={BucketName}, KeyName={KeyName}", bucketName, keyName);
|
||||
return Result.NotFound();
|
||||
}
|
||||
|
||||
if (!fileResult.IsSuccess)
|
||||
{
|
||||
logger.LogError("Failed to read source image from storage: BucketName={BucketName}, KeyName={KeyName}. Errors: {Errors}",
|
||||
bucketName,
|
||||
keyName,
|
||||
string.Join("; ", fileResult.Errors));
|
||||
return Result.Error(string.Join("; ", fileResult.Errors));
|
||||
}
|
||||
|
||||
using var fileStream = fileResult.Value;
|
||||
var memoryStream = new MemoryStream();
|
||||
await fileStream.CopyToAsync(memoryStream, cancellationToken);
|
||||
memoryStream.Position = 0;
|
||||
|
||||
return Result.Success(memoryStream);
|
||||
}
|
||||
|
||||
async Task ProcessOperationAsync(
|
||||
JobOperationItem operationItem,
|
||||
string bucketName,
|
||||
MemoryStream sourceImage,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var jobOperation = operationItem.JobOperation;
|
||||
|
||||
try
|
||||
{
|
||||
jobOperation.JobState = JobState.InProgress;
|
||||
sourceImage.Position = 0;
|
||||
|
||||
var resizedPictureResult = await pictureHandler.ResizePictureAsync(
|
||||
sourceImage,
|
||||
operationItem.Item.Width,
|
||||
operationItem.Item.Height,
|
||||
cancellationToken);
|
||||
|
||||
if (!resizedPictureResult.IsSuccess)
|
||||
{
|
||||
var errorMessage = string.Join("; ", resizedPictureResult.Errors);
|
||||
logger.LogError("Failed to resize picture for JobOperationId={JobOperationId}. Error: {ErrorMessage}",
|
||||
jobOperation.Id,
|
||||
errorMessage);
|
||||
|
||||
MarkAsFailed(jobOperation, errorMessage);
|
||||
return;
|
||||
}
|
||||
|
||||
var saveResult = await storageHandler.UploadFileAsync(
|
||||
resizedPictureResult.Value,
|
||||
"image/webp",
|
||||
bucketName,
|
||||
$"{operationItem.Item.SaveOnRoute}/{Guid.NewGuid() }.webp",
|
||||
cancellationToken);
|
||||
|
||||
if (!saveResult.IsSuccess)
|
||||
{
|
||||
var errorMessage = string.Join("; ", saveResult.Errors);
|
||||
logger.LogError("Failed to upload resized picture for JobOperationId={JobOperationId}. Error: {ErrorMessage}",
|
||||
jobOperation.Id,
|
||||
errorMessage);
|
||||
|
||||
MarkAsFailed(jobOperation, errorMessage);
|
||||
return;
|
||||
}
|
||||
|
||||
jobOperation.JobState = JobState.Completed;
|
||||
jobOperation.FinishedAt = DateTime.UtcNow;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception,
|
||||
"An error occurred while processing resize for JobOperationId={JobOperationId}, Width={Width}, Height={Height}, SaveOnRoute={SaveOnRoute}",
|
||||
jobOperation.Id,
|
||||
operationItem.Item.Width,
|
||||
operationItem.Item.Height,
|
||||
operationItem.Item.SaveOnRoute);
|
||||
|
||||
MarkAsFailed(jobOperation, exception.Message);
|
||||
}
|
||||
}
|
||||
|
||||
static void MarkAllAsFailed(IEnumerable<JobOperationItem> jobOperations, string errorMessage)
|
||||
{
|
||||
foreach (var operationItem in jobOperations)
|
||||
{
|
||||
MarkAsFailed(operationItem.JobOperation, errorMessage);
|
||||
}
|
||||
}
|
||||
|
||||
static void MarkAsFailed(JobOperation jobOperation, string errorMessage)
|
||||
{
|
||||
jobOperation.JobState = JobState.Failed;
|
||||
jobOperation.FinishedAt = DateTime.UtcNow;
|
||||
jobOperation.ErrorMessage = errorMessage;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
using MassTransit;
|
||||
using Musify.Application.Configuration;
|
||||
using Musify.Application.Contracts.Infrastructure;
|
||||
using Musify.Application.Events;
|
||||
using Musify.Infrastructure.Messaging.Activities.Arguments;
|
||||
|
||||
namespace Musify.Infrastructure.Messaging.Consumers
|
||||
{
|
||||
public class UpdatePlayListPictureConsumer(IBus bus, IPictureHandler pictureHandler,
|
||||
StorageSettings storageSettings, PlayListConfiguration playListConfiguration)
|
||||
: IConsumer<UpdatePlayListPictureEvent>
|
||||
{
|
||||
public const string QueueName = "Update-PlayList-Picture";
|
||||
|
||||
public async Task Consume(ConsumeContext<UpdatePlayListPictureEvent> consumeContext)
|
||||
{
|
||||
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
|
||||
|
||||
var smallPictureKeyName = Path.Combine(
|
||||
playListConfiguration.Routes.SmallPictures,
|
||||
$"{consumeContext.Message.PlayListId}_small.{pictureHandler.FileExtension}");
|
||||
var mediumPictureKeyName = Path.Combine(
|
||||
playListConfiguration.Routes.MediumPictures,
|
||||
$"{consumeContext.Message.PlayListId}_medium.{pictureHandler.FileExtension}");
|
||||
var largePictureKeyName = Path.Combine(
|
||||
playListConfiguration.Routes.LargePictures,
|
||||
$"{consumeContext.Message.PlayListId}_large.{pictureHandler.FileExtension}");
|
||||
|
||||
routingSlipBuilder.AddActivity(
|
||||
"ResizeSmall",
|
||||
new Uri("Queue:Resize-Picture_Execute"),
|
||||
new ResizePictureArgument(
|
||||
storageSettings.BucketName,
|
||||
consumeContext.Message.OriginalPictureKeyName,
|
||||
storageSettings.BucketName,
|
||||
playListConfiguration.Routes.SmallPictures,
|
||||
playListConfiguration.PicturesSizes.SmallPictureWidth,
|
||||
playListConfiguration.PicturesSizes.SmallPictureHeight));
|
||||
|
||||
routingSlipBuilder.AddActivity(
|
||||
"ResizeMedium",
|
||||
new Uri("Queue:Resize-Picture_Execute"),
|
||||
new ResizePictureArgument(
|
||||
storageSettings.BucketName,
|
||||
consumeContext.Message.OriginalPictureKeyName,
|
||||
storageSettings.BucketName,
|
||||
playListConfiguration.Routes.MediumPictures,
|
||||
playListConfiguration.PicturesSizes.MediumPictureWidth,
|
||||
playListConfiguration.PicturesSizes.MediumPictureHeight));
|
||||
|
||||
routingSlipBuilder.AddActivity(
|
||||
"ResizeLarge",
|
||||
new Uri("Queue:Resize-Picture_Execute"),
|
||||
new ResizePictureArgument(
|
||||
storageSettings.BucketName,
|
||||
consumeContext.Message.OriginalPictureKeyName,
|
||||
storageSettings.BucketName,
|
||||
playListConfiguration.Routes.LargePictures,
|
||||
playListConfiguration.PicturesSizes.LargePictureWidth,
|
||||
playListConfiguration.PicturesSizes.LargePictureHeight));
|
||||
|
||||
routingSlipBuilder.AddActivity(
|
||||
"DeleteOriginal",
|
||||
new Uri("Queue:Delete-Picture_Execute"),
|
||||
new RemoveFileArguments(
|
||||
storageSettings.BucketName,
|
||||
consumeContext.Message.OriginalPictureKeyName));
|
||||
|
||||
routingSlipBuilder.AddActivity(
|
||||
"UpdatePlayListPicture",
|
||||
new Uri("Queue:Update-PlayList-Picture_Execute"),
|
||||
new UpdatePlayListPictureArguments(
|
||||
consumeContext.Message.PlayListId,
|
||||
smallPictureKeyName,
|
||||
mediumPictureKeyName,
|
||||
largePictureKeyName));
|
||||
|
||||
var routingSlip = routingSlipBuilder.Build();
|
||||
await bus.Execute(routingSlip);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user