using FileService.Application.Ports; using FileService.Domain.IReposities; using IM.Application.Abstractions; using IM.Commons.IntegrationEvents; using MassTransit; using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; namespace FileService.Application.EventHandler { public class UploadTaskCompleteEventHandler : IConsumer { private readonly IObjectStorageRouter router; private readonly IUploadFileReposity uploadFile; private readonly IUnitOfWork uwork; private readonly IUploadTaskReposity uploadTask; public UploadTaskCompleteEventHandler(IObjectStorageRouter router, IUploadFileReposity uploadFile, IUploadTaskReposity uploadTask, IUnitOfWork uwork) { this.router = router; this.uploadFile = uploadFile; this.uwork = uwork; this.uploadTask = uploadTask; } public async Task Consume(ConsumeContext context) { var @event = context.Message; var existingFile = await uploadFile.FindBySourceTaskIdAsync(@event.TaskId); if (existingFile != null) { return; } var task = await uploadTask.FindByIdAsync(@event.TaskId); if (task is null) { throw new InvalidOperationException($"Upload task {@event.TaskId} has not been committed yet."); } var storage = router.Route(@event.ProviderCode); try { await storage.CompleteUploadAsync(new StorageContracts.CompleteUploadCommand( ProviderCode: @event.ProviderCode, Bucket: @event.Bucket, Region: @event.Region, ObjectKey: @event.ObjectKey, UploadSessionId: @event.SessionId, Parts: @event.Parts.Select(s => new StorageContracts.UploadPart( PartNumber: s.PartNumber, ETag: s.ETag, Size: s.Size, Checksum: s.Checksum )).ToList() ), context.CancellationToken); var file = new Domain.Entities.UploadFile( ownerId: @event.OperatorId, fileName: @event.FileName, fileSize: @event.FileSize, contentType: @event.ContentType, new Domain.ValueObjects.StorageLocation(@event.ProviderCode, @event.Bucket, @event.ObjectKey, @event.Region), checkSum: new Domain.ValueObjects.CheckSum("md5", @event.CheckSun), sourceTaskId: @event.TaskId, chatType: @event.ChatType, targetId: @event.TargetId, isPublic: false ); uploadFile.Create(file); task.CompleteUpload(file.Id); await uwork.SaveChangesAsync(context.CancellationToken); } catch (Exception ex) { task.Fail(ex.Message); await uwork.SaveChangesAsync(context.CancellationToken); throw; } } } }