jobs and others

2026-03-19 01:09:02 +01:00
parent 574b0a05e2
commit 34c213ef5a
28 changed files with 334 additions and 316 deletions
@@ -1,138 +0,0 @@
-using Ardalis.Result;
-using MassTransit;
-using Microsoft.Extensions.Logging;
-using Musify.Application.Contracts.Infrastructure;
-using Musify.Application.Pictures.Contracts;
-using Musify.Application.Pictures.Events;
-using Musify.Domain.Entities;
-
-namespace Musify.Infrastructure.Messaging.Consumers
-{
-    public class PictureResizerConsumer(IStorageHandler storageHandler, IDatabase database, IPictureHandler pictureHandler, ILogger<PictureResizerConsumer> logger)
-        : IConsumer<PictureResizeEvent>
-    {
-        public const string QueueName = "picture-resize-queue";
-
-        public async Task Consume(ConsumeContext<PictureResizeEvent> consumeContext)
-        {
-            try
-            {
-                var upload = await database.Uploads.FindAsync(consumeContext.Message.UploadId);
-
-                if (upload is null)
-                {
-                    logger.LogWarning("Upload not found in database: UploadId={UploadId}", consumeContext.Message.UploadId);
-                    return;
-                }
-
-                if (upload.State != UploadState.Completed)
-                {
-                    logger.LogInformation("Upload is not in a Completed state, skipping processing: UploadId={UploadId}, State={State}", consumeContext.Message.UploadId, upload.State);
-                    return;
-                }
-
-                await UpdateState(database, upload, UploadState.Processing, consumeContext.CancellationToken);
-
-                var fileResult = await storageHandler.GetFileAsync(
-                    upload.BucketName,
-                    upload.KeyName,
-                    consumeContext.CancellationToken);
-
-                if (fileResult.IsNotFound())
-                {
-                    logger.LogWarning("File not found in storage: BucketName={BucketName}, KeyName={KeyName}", upload.BucketName, upload.KeyName);
-                    await UpdateState(database, upload, UploadState.Failed, consumeContext.CancellationToken);
-                    return;
-                }
-
-                using var fileStream = fileResult.Value;
-
-                using var memoryStream = new MemoryStream();
-                await fileStream.CopyToAsync(memoryStream, consumeContext.CancellationToken);
-
-                foreach (var resize in consumeContext.Message.ImageResizes)
-                {
-                    var newUploadId = Guid.NewGuid();
-
-                    var newUpload = new Upload
-                    {
-                        Id = newUploadId,
-                        BucketName = upload.BucketName,
-                        KeyName = $"{resize.SaveOnRoute}/{newUploadId}.webp",
-                        ContentType = "image/webp",
-                        State = UploadState.Processing,
-                        EntityId = upload.EntityId,
-                        EntityType = upload.EntityType
-                    };
-
-                    await database.Uploads.AddAsync(newUpload, consumeContext.CancellationToken);
-
-                    await database.SaveChangesAsync(consumeContext.CancellationToken);
-
-                    try
-                    {
-                        await ResizePicture(upload, newUpload, memoryStream, resize, consumeContext.CancellationToken);
-                    }
-                    catch (Exception exception)
-                    {
-                        logger.LogError(exception, "An error occurred while resizing picture for UploadId={UploadId} and Resize={Resize}", upload.Id, resize);
-                        await UpdateState(database, newUpload, UploadState.Failed, consumeContext.CancellationToken);
-                        continue;
-                    }
-                }
-
-                await UpdateState(database, upload, UploadState.Completed, consumeContext.CancellationToken);
-            }
-            catch (Exception exception)
-            {
-                logger.LogError(exception, "An error occurred while consuming the ImageResizedConsumer message.");
-                throw;
-            }
-            finally 
-            {
-                logger.LogInformation("Saving changes to database after processing ImageResizedConsumer message.");
-                await database.SaveChangesAsync(consumeContext.CancellationToken);
-            }
-        }
-        
-        async Task UpdateState(IDatabase database, Upload upload, UploadState uploadState, CancellationToken cancellationToken)
-        {
-            upload.State = uploadState;
-            database.Uploads.Update(upload);
-        }
-
-        async Task ResizePicture(Upload originalUpload, Upload newUpload, MemoryStream memoryStream, PictureResize pictureResize, CancellationToken cancellationToken)
-        {
-            memoryStream.Position = 0;
-
-            var pictureResult = await pictureHandler.ResizePictureAsync(
-                memoryStream,
-                pictureResize.Width,
-                pictureResize.Height,
-                cancellationToken);
-
-            if (!pictureResult.IsSuccess)
-            {
-                logger.LogError("Failed to resize picture: {ErrorMessage}", string.Join("; ", pictureResult.Errors));
-                await UpdateState(database, newUpload, UploadState.Failed, cancellationToken);
-                return;
-            }
-
-            var saveResult = await storageHandler.UploadFileAsync(
-                pictureResult.Value,
-                "image/webp",
-                originalUpload.BucketName,
-                $"{pictureResize.SaveOnRoute}/{newUpload.Id}.webp",
-                cancellationToken);
-
-            if (!saveResult.IsSuccess)
-            {
-                logger.LogError("Failed to upload resized picture: {ErrorMessage}", string.Join("; ", saveResult.Errors));
-                await UpdateState(database, newUpload, UploadState.Failed, cancellationToken);
-                return;
-            }
-
-            await UpdateState(database, newUpload, UploadState.Completed, cancellationToken);
-        }
-    }
-}
@@ -0,0 +1,14 @@
+using MassTransit;
+using Musify.Application.Events;
+
+namespace Musify.Infrastructure.Messaging.Consumers
+{
+    public class RemoveFileConsumer
+        : IConsumer<RemoveFileEvent>
+    {
+        public Task Consume(ConsumeContext<RemoveFileEvent> context)
+        {
+            throw new NotImplementedException();
+        }
+    }
+}
@@ -0,0 +1,138 @@
+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>
+    {
+        public const string QueueName = "picture-resize-queue";
+
+        public async Task Consume(ConsumeContext<ResizePictureEvent> consumeContext)
+        {
+            try
+            {
+                Dictionary<ResizePictureItems, Job> jobs = new Dictionary<ResizePictureItems, Job>();
+
+                foreach (var item in consumeContext.Message.Items)
+                {
+                    var payloadJson = JsonSerializer.Serialize(item);
+
+                    var job = await database.Jobs
+                        .SingleOrDefaultAsync(job => job.JobType == JobType.ResizePicture && job.Payload == payloadJson, consumeContext.CancellationToken);
+
+                    if (job is null)
+                    {
+                        job = new Job
+                        {
+                            JobType = JobType.ResizePicture,
+                            Payload = payloadJson,
+                        };
+                        await database.Jobs.AddAsync(job, consumeContext.CancellationToken);
+                    }
+                    else
+                    {
+                        job.RetryCount++;
+                        job.JobState = JobState.InProgress;
+                        database.Jobs.Update(job);
+                    }
+
+                    jobs.Add(item, job);
+                }
+
+                var fileResult = await storageHandler.GetFileAsync(
+                    consumeContext.Message.BucketName,
+                    consumeContext.Message.KeyName,
+                    consumeContext.CancellationToken);
+
+                if (fileResult.IsNotFound())
+                {
+                    logger.LogWarning("File not found in storage: BucketName={BucketName}, KeyName={KeyName}",
+                        consumeContext.Message.BucketName, consumeContext.Message.KeyName);
+
+                    foreach (var job in jobs)
+                    {
+                        job.Value.JobState = JobState.Failed;
+                    }
+                    await database.SaveChangesAsync(consumeContext.CancellationToken);
+                    return;
+                }
+
+                using var fileStream = fileResult.Value;
+                using var memoryStream = new MemoryStream();
+                await fileStream.CopyToAsync(memoryStream, consumeContext.CancellationToken);
+
+                foreach (var item in consumeContext.Message.Items)
+                {
+                    var job = jobs[item];
+
+                    try
+                    {
+                        await ResizePicture(
+                            job,
+                            consumeContext.Message.BucketName,
+                            memoryStream,
+                            item,
+                            consumeContext.CancellationToken);
+                    }
+                    catch (Exception exception)
+                    {
+                        logger.LogError(exception, "An error occurred while resizing picture for UploadId={UploadId} and Resize={Resize}", job.Id, item);
+                        continue;
+                    }
+                }
+            }
+            catch (Exception exception)
+            {
+                logger.LogError(exception, "An error occurred while consuming the ImageResizedConsumer message.");
+                throw;
+            }
+            finally 
+            {
+                logger.LogInformation("Saving changes to database after processing ImageResizedConsumer message.");
+                await database.SaveChangesAsync(consumeContext.CancellationToken);
+            }
+        }
+
+        async Task ResizePicture(Job job, string bucketName, MemoryStream memoryStream, ResizePictureItems items, CancellationToken cancellationToken)
+        {
+            memoryStream.Position = 0;
+            job.JobState = JobState.InProgress;
+
+            var pictureResult = await pictureHandler.ResizePictureAsync(
+                memoryStream,
+                items.Width,
+                items.Height,
+                cancellationToken);
+
+            if (!pictureResult.IsSuccess)
+            {
+                logger.LogError("Failed to resize picture: {ErrorMessage}", string.Join("; ", pictureResult.Errors));
+                job.JobState = JobState.Failed;
+                return;
+            }
+
+            var saveResult = await storageHandler.UploadFileAsync(
+                pictureResult.Value,
+                "image/webp",
+                bucketName,
+                $"{items.SaveOnRoute}/{Guid.NewGuid() }.webp",
+                cancellationToken);
+
+            if (!saveResult.IsSuccess)
+            {
+                logger.LogError("Failed to upload resized picture: {ErrorMessage}", string.Join("; ", saveResult.Errors));
+                job.JobState = JobState.Failed;
+                return;
+            }
+
+            job.JobState = JobState.Completed;
+        }
+    }
+}
@@ -37,7 +37,7 @@ namespace Musify.Infrastructure.Messaging
        {
            serviceDescriptors.AddMassTransit(options =>
            {
-                options.AddConsumer<PictureResizerConsumer>();
+                options.AddConsumer<ResizePictureConsumer>();

                options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) =>
                {
@@ -58,9 +58,9 @@ namespace Musify.Infrastructure.Messaging
                        options.Password(password);
                    } );

-                    busFactoryConfigurator.ReceiveEndpoint(PictureResizerConsumer.QueueName, endpointConfigurator =>
+                    busFactoryConfigurator.ReceiveEndpoint(ResizePictureConsumer.QueueName, endpointConfigurator =>
                    {
-                        endpointConfigurator.ConfigureConsumer<PictureResizerConsumer>(busRegistrationContext);
+                        endpointConfigurator.ConfigureConsumer<ResizePictureConsumer>(busRegistrationContext);
                    } );
                } );
            } );
@@ -22,5 +22,7 @@ namespace Musify.Infrastructure.Persistence
        public DbSet<PlayListHasTrack> PlayListHasTracks => Set<PlayListHasTrack>();

        public DbSet<Upload> Uploads => Set<Upload>();
+
+        public DbSet<Job> Jobs => Set<Job>();
    }
 }
@@ -4,10 +4,11 @@ using Amazon.S3.Transfer;
 using Ardalis.Result;
 using Microsoft.Extensions.Logging;
 using Musify.Application.Contracts.Infrastructure;
+using Musify.Domain.Entities;

 namespace Musify.Infrastructure.Storage
 {
-    public class StorageHandler(IAmazonS3 amazonS3, ILogger<StorageHandler> logger)
+    public class StorageHandler(IDatabase database, IAmazonS3 amazonS3, ILogger<StorageHandler> logger)
        : IStorageHandler
    {
        public async Task<Result<Stream>> GetFileAsync(string bucketName, string keyName, CancellationToken cancellationToken)
@@ -75,18 +76,31 @@ namespace Musify.Infrastructure.Storage
                string fileName = Path.GetFileName(file);
                string filePath = Path.Combine(sourceDirectory, file);

+                var upload = new Upload
+                {
+                    Id = Guid.NewGuid(),
+                    BucketName = bucketName,
+                    KeyName = $"{route}/{fileName}",
+                    ContentType = "Application/Octet-Stream"
+                };
+                await database.Uploads.AddAsync(upload, cancellationToken);
+
                try
                {
                    await trasnsferUtility.UploadAsync(filePath, bucketName, fileName, cancellationToken);
+                    upload.State = UploadState.Successful;
                    logger.LogInformation("Successfully transferred file {FileName} to S3 with bucket name {BucketName} and route {Route}", fileName, bucketName, route);
                }
                catch
                {
                    logger.LogError("Failed to transfer file {FileName} to S3 with bucket name {BucketName} and route {Route}", fileName, bucketName, route);
-                    return Result.Error($"Failed to transfer file {fileName} to S3 with bucket name {bucketName} and route {route}");
+                    upload.State = UploadState.Failed;
                }
            }

+            await database.SaveChangesAsync(cancellationToken);
+            logger.LogInformation("Successfully saved upload records to database for all files transferred to S3 with bucket name {BucketName} and route {Route}", bucketName, route);
+
            return Result.Success();
        }

@@ -100,6 +114,15 @@ namespace Musify.Infrastructure.Storage
                Key = keyName
            };

+            var upload = new Upload
+            {
+                Id = Guid.NewGuid(),
+                BucketName = bucketName,
+                KeyName = keyName,
+                ContentType = contentType
+            };
+            await database.Uploads.AddAsync(upload, cancellationToken);
+
            try
            {
                var response = await amazonS3.PutObjectAsync(request, cancellationToken);
@@ -108,9 +131,14 @@ namespace Musify.Infrastructure.Storage
            catch
            {
                logger.LogError("Failed to upload file to S3 with bucket name {BucketName} and key name {KeyName}", bucketName, keyName);
+                upload.State = UploadState.Failed;
                return Result.Error($"Failed to upload file to S3 with bucket name {bucketName} and key name {keyName}");
            }

+            upload.State = UploadState.Successful;
+            await database.SaveChangesAsync(cancellationToken);
+            logger.LogInformation("Successfully saved upload record to database with bucket name {BucketName} and key name {KeyName}", bucketName, keyName);
+
            return Result.Success(keyName);
        }
    }