Uploads

Infrastructure/Infrastructure.csproj

@@ -10,6 +10,7 @@
   <ItemGroup>
     <PackageReference Include="AWSSDK.S3" Version="4.0.19" />
     <PackageReference Include="Keycloak.Net.Core" Version="1.0.38" />
+    <PackageReference Include="MassTransit" Version="9.0.1" />
     <PackageReference Include="Microsoft.EntityFrameworkCore" Version="10.0.5" />
     <PackageReference Include="SixLabors.ImageSharp" Version="3.1.12" />
   </ItemGroup>

Infrastructure/Messaging/Consumers/ImageResizedConsumer.cs

@@ -0,0 +1,99 @@
+using Ardalis.Result;
+using MassTransit;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging;
+using Musify.Application.Contracts.Infrastructure;
+using Musify.Application.Pictures.Events;
+using Musify.Domain.Entities;
+
+namespace Musify.Infrastructure.Messaging.Consumers
+{
+    public class ImageResizedConsumer(IStorageHandler storageHandler, IDatabase database, IPictureHandler pictureHandler, ILogger<ImageResizedConsumer> logger)
+        : IConsumer<ImageResizeEvent>
+    {
+        public async Task Consume(ConsumeContext<ImageResizeEvent> consumeContext)
+        {
+            try
+            {
+                var upload = await database.Uploads.AsNoTracking().SingleOrDefaultAsync(u => u.Id == consumeContext.Message.UploadId);
+
+                if (upload is null)
+                {
+                    logger.LogWarning("Upload not found in database: UploadId={UploadId}", consumeContext.Message.UploadId);
+                    return;
+                }
+
+                if (upload.State != UploadState.Pending)
+                {
+                    logger.LogInformation("Upload is not in a failed state, skipping processing: UploadId={UploadId}, State={State}", consumeContext.Message.UploadId, upload.State);
+                    return;
+                }
+
+                await UpdateState(database, upload, UploadState.Processing, consumeContext.CancellationToken);
+
+                var file = await storageHandler.GetFileAsync(
+                    upload.BucketName,
+                    upload.KeyName,
+                    consumeContext.CancellationToken);
+
+                if (file.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;
+                }
+
+                var pictureResult = await pictureHandler.ResizePictureAsync(
+                    file.Value,
+                    consumeContext.Message.Width,
+                    consumeContext.Message.Height,
+                    consumeContext.CancellationToken);
+
+                if (!pictureResult.IsSuccess)
+                {
+                    string errors = string.Join("; ", pictureResult.Errors);
+                    logger.LogError("Failed to resize picture: {ErrorMessage}", errors);
+                    
+                    await UpdateState(database, upload, UploadState.Failed, consumeContext.CancellationToken);
+
+                    return;
+                }
+
+                var saveResult = await storageHandler.UploadFileAsync(
+                    pictureResult.Value,
+                    "image/webp",
+                    upload.BucketName,
+                    $"{consumeContext.Message.SaveRoute}/{upload.Id}.webp",
+                    consumeContext.CancellationToken);
+
+                if (!saveResult.IsSuccess)
+                {
+                    string errors = string.Join("; ", saveResult.Errors);
+                    logger.LogError("Failed to upload resized picture: {ErrorMessage}", errors);
+
+                    await UpdateState(database, upload, UploadState.Failed, consumeContext.CancellationToken);
+
+                    return;
+                }
+
+                await UpdateState(database, upload, UploadState.Completed, consumeContext.CancellationToken);
+            }
+            catch (Exception exception)
+            {
+                logger.LogError(exception, "An error occurred while consuming the ImageResizedConsumer message.");
+                throw;
+            }
+        }
+
+        async Task UpdateState(IDatabase database, Upload upload, UploadState uploadState, CancellationToken cancellationToken)
+        {
+            upload.State = uploadState;
+
+            database.Uploads.Update(upload);
+
+            await database.SaveChangesAsync(cancellationToken);
+        }
+    }
+}

Infrastructure/Messaging/MassTransitEventBus.cs

@@ -0,0 +1,16 @@
+using MassTransit;
+using Musify.Application.Contracts.Infrastructure;
+
+namespace Musify.Infrastructure.Messaging
+{
+    public class MassTransitEventBus(IPublishEndpoint publishEndpoint)
+        : IEventBus
+    {
+        public async Task PublishAsync<T>(T message, CancellationToken cancellationToken)
+        {
+            ArgumentNullException.ThrowIfNull(message, nameof(message));
+
+            await publishEndpoint.Publish(message, cancellationToken);
+        }
+    }
+}

Infrastructure/Persistence/Database.cs

@@ -14,5 +14,7 @@ namespace Musify.Infrastructure.Persistence
         public DbSet<PlayList> PlayLists => Set<PlayList>();
 
         public DbSet<PlayListHasTrack> PlayListHasTracks => Set<PlayListHasTrack>();
+
+        public DbSet<Upload> Uploads => Set<Upload>();
     }
 }

Infrastructure/Pictures/PictureHandler.cs

@@ -11,7 +11,7 @@ namespace Musify.Infrastructure.Pictures
     {
         public async Task<Result<Stream>> ResizePictureAsync(Stream pictureStream, int width, int height, CancellationToken cancellationToken)
         {
-            try 
+            try
             {
                 if (pictureStream.CanSeek && pictureStream.Position > 0)
                     pictureStream.Position = 0;

Infrastructure/Storage/StorageHandler.cs

@@ -10,7 +10,7 @@ namespace Musify.Infrastructure.Storage
     public class StorageHandler(IAmazonS3 amazonS3, ILogger<StorageHandler> logger)
         : IStorageHandler
     {
-        public async Task<Result<Stream>> GetFile(string bucketName, string keyName, CancellationToken cancellationToken)
+        public async Task<Result<Stream>> GetFileAsync(string bucketName, string keyName, CancellationToken cancellationToken)
         {
             var request = new GetObjectRequest
             {
@@ -34,7 +34,7 @@ namespace Musify.Infrastructure.Storage
             }
         }
 
-        public async Task<string> GetUrl(string bucketName, string keyName, TimeSpan expirationTime, CancellationToken cancellationToken)
+        public async Task<string> GetUrlAsync(string bucketName, string keyName, TimeSpan expirationTime, CancellationToken cancellationToken)
         {
             var request = new GetPreSignedUrlRequest
             {
@@ -50,7 +50,7 @@ namespace Musify.Infrastructure.Storage
             return url;
         }
 
-        public async Task<Result> RemoveFile(string bucketName, string keyName, CancellationToken cancellationToken)
+        public async Task<Result> RemoveFileAsync(string bucketName, string keyName, CancellationToken cancellationToken)
         {
             var request = new DeleteObjectRequest
             {
@@ -64,7 +64,7 @@ namespace Musify.Infrastructure.Storage
             return Result.NoContent();
         }
 
-        public async Task<Result> TransferFiles(string sourceDirectory, string bucketName, string route, CancellationToken cancellationToken)
+        public async Task<Result> TransferFilesAsync(string sourceDirectory, string bucketName, string route, CancellationToken cancellationToken)
         {
             var trasnsferUtility = new TransferUtility(amazonS3);
 
@@ -89,7 +89,7 @@ namespace Musify.Infrastructure.Storage
             return Result.Success();
         }
 
-        public async Task<Result<string>> UploadFile(Stream sourceStream, string contentType, string bucketName, string keyName, CancellationToken cancellationToken)
+        public async Task<Result<string>> UploadFileAsync(Stream sourceStream, string contentType, string bucketName, string keyName, CancellationToken cancellationToken)
         {
             var request = new PutObjectRequest
             {