Merge branch 'master' of https://gitea.nxsir.cn/nanxun/IM_NEW

FileService.Application/EventHandler/UploadTaskCompleteEventHandler.cs

@@ -1,4 +1,6 @@
 using FileService.Application.Ports;
+using FileService.Domain.IReposities;
+using IM.Application.Abstractions;
 using IM.Commons.IntegrationEvents;
 using MassTransit;
 using System;
@@ -12,18 +14,23 @@ namespace FileService.Application.EventHandler
     public class UploadTaskCompleteEventHandler : IConsumer<UploadTaskCompleteEvent>
     {
         private readonly IObjectStorageRouter router;
-        private readonly IStorageRedisCache cache;
+        private readonly IUploadFileReposity uploadFile;
+        private readonly IUnitOfWork uwork;
+        private readonly IStorageRedisCache storageCache;
 
-        public UploadTaskCompleteEventHandler(IObjectStorageRouter router, IStorageRedisCache cache)
+        public UploadTaskCompleteEventHandler(IObjectStorageRouter router, IUploadFileReposity uploadFile, IUnitOfWork uwork, IStorageRedisCache storageCache)
         {
             this.router = router;
-            this.cache = cache;
+            this.uploadFile = uploadFile;
+            this.uwork = uwork;
+            this.storageCache = storageCache;
         }
 
         public async Task Consume(ConsumeContext<UploadTaskCompleteEvent> context)
         {
             var @event = context.Message;
             var storage = router.Route(@event.ProviderCode);
+            var taskCache = await storageCache.GetAsync(@event.SessionId);
             if(@event.ProviderCode == "Local")
             {
                 await storage.CompleteUploadAsync(new StorageContracts.CompleteUploadCommand(
@@ -39,6 +46,14 @@ namespace FileService.Application.EventHandler
                             Checksum: s.Checksum
                         )).ToList()
                     ), context.CancellationToken);
+                uploadFile.Create(new Domain.Entities.UploadFile(
+                    ownerId: @event.OperatorId,
+                    fileName: @event.FileName,
+                    fileSize: taskCache.FileSize,
+                    contentType: @event.ContentType,
+                    new Domain.ValueObjects.StorageLocation(taskCache.ProviderCode, taskCache.Bucket, taskCache.ObjectKey, taskCache.Region),
+                    checkSum: new Domain.ValueObjects.CheckSum("md5", @event.CheckSun)
+                    ));
             }
         }
     }

FileService.Application/FileService.Application.csproj

@@ -12,6 +12,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\FileService.Domain\FileService.Domain.csproj" />
+    <ProjectReference Include="..\IM.Application\IM.Application.csproj" />
     <ProjectReference Include="..\IM.Commons\IM.Commons.csproj" />
     <ProjectReference Include="..\IM.InitCommon\IM.InitCommon.csproj" />
   </ItemGroup>

FileService.Application/ModueInit.cs

@@ -1,4 +1,5 @@
 using FileService.Application.UploadFile;
+using FileService.Application.UploadFileTask;
 using IM.Commons;
 using Microsoft.Extensions.DependencyInjection;
 
@@ -9,6 +10,7 @@ namespace FileService.Application
         public void Initialize(IServiceCollection services)
         {
             services.AddScoped<UploadFileService>();
+            services.AddScoped<UploadFileTaskService>();
         }
     }
 }

FileService.Application/Ports/ILocalChunckStorage.cs

@@ -0,0 +1,10 @@
+using FileService.Application.StorageContracts;
+
+namespace FileService.Application.Ports
+{
+    public interface ILocalChunkStorage
+    {
+        Task SavePartAsync(
+            SaveLocalPartCommand command);
+    }
+}

FileService.Application/StorageContracts/SaveLocalPartCommand.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace FileService.Application.StorageContracts
+{
+    public record SaveLocalPartCommand(
+        string UploadSessionId,
+    int PartNumber,
+    Stream Stream,
+    long ContentLength,
+    string? Checksum = null);
+}

FileService.Application/StorageContracts/UploadRuntimeCache.cs

@@ -25,14 +25,14 @@ namespace FileService.Application.StorageContracts
 
         public int TotalPartCount { get; init; }
 
-        public long UploadedBytes { get; private set; }
+        public long UploadedBytes { get; set; }
 
         public DateTimeOffset CreatedAt { get; init; }
 
         public DateTimeOffset ExpireAt { get; init; }
 
         public Dictionary<int, UploadPart> Parts { get; init; } = new();
-
+        public UploadRuntimeCache() { }
         public UploadRuntimeCache(string taskId, string providerCode, 
             string uploadSessionId, string bucket, string region, 
             string objectKey, long fileSize, int totalPartCount, 

FileService.Application/UploadFileTask/UploadFileTaskService.cs

@@ -1,26 +1,19 @@
 using AutoMapper;
 using FileService.Application.Ports;
 using FileService.Application.StorageContracts;
-using FileService.Domain.Entities;
 using FileService.Domain.IReposities;
 using IM.Commons;
 using IM.Commons.IntegrationEvents;
 using IM.InitCommon;
 using MassTransit;
-using MassTransit.Internals;
 using Microsoft.Extensions.Options;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 
 namespace FileService.Application.UploadFileTask
 {
     public class UploadFileTaskService(IUploadTaskReposity reposity, 
         IMapper mapper, IObjectStorageRouter router, 
         IOptions<StorageOptions> options, IStorageRedisCache redis,
-        IPublishEndpoint endpoint
+        IPublishEndpoint endpoint, ILocalChunkStorage localChunkStorage
         )
     {
         private readonly IUploadTaskReposity reposity = reposity;
@@ -29,23 +22,24 @@ namespace FileService.Application.UploadFileTask
         private readonly IOptions<StorageOptions> options = options;
         private readonly IStorageRedisCache redis = redis;
         private readonly IPublishEndpoint endpoint = endpoint;
+        private readonly ILocalChunkStorage localChunkStorage = localChunkStorage;
         private readonly IObjectStoragePort storage = router.Route(options.Value.DefaultProviderCode);
 
         public async Task<Result<TaskInitResponse>> InitTaskAsync(UploadTaskInitCommand command)
         {
             CancellationToken cancellationToken = CancellationToken.None;
             var task = command.ToUploadTask();
-
+            var date = DateTime.Now;
             var storageOption = options.Value.Providers[options.Value.DefaultProviderCode];
             var storage = router.Route(storageOption.ProviderCode);
-            var initRes = await storage.InitUploadAsync(new StorageContracts.InitiateUploadCommand(
+            var initUpdateCommand = new StorageContracts.InitiateUploadCommand(
                 ProviderCode: storageOption.ProviderCode,
                 Bucket: storageOption.Bucket,
-                ObjectKey: storageOption.Endpoint,
-                ContentType:task.ContentType.Value,
+                ObjectKey: $"{storageOption.LocalRootPath}\\{date.Year}\\{date.Month}\\{date.Day}\\{command.FileName}",
+                ContentType: task.ContentType.Value,
                 ContentLength: command.FileSize,
-                null
-                ), cancellationToken);
+                null);
+            var initRes = await storage.InitUploadAsync(initUpdateCommand, cancellationToken);
 
             var res = mapper.Map<TaskInitResponse>(initRes);
             res.TaskId = task.Id;
@@ -59,7 +53,7 @@ namespace FileService.Application.UploadFileTask
                     uploadSessionId: res.UploadSessionId,
                     bucket: storageOption.Bucket,
                     region: storageOption.Region,
-                    objectKey: storageOption.Endpoint,
+                    objectKey: initUpdateCommand.ObjectKey,
                     fileSize: task.FileSize,
                     totalPartCount: (int)(task.FileSize % storageOption.DefaultPartSizeBytes > 0 ?
                         (task.FileSize / storageOption.DefaultPartSizeBytes) + 1 : 
@@ -78,6 +72,11 @@ namespace FileService.Application.UploadFileTask
                 return Result.Fail<PresignedUrl>(ResultCode.CHUNK_NOT_FOUND);
             }
 
+            if(taskCache.TotalPartCount < partNum)
+            {
+                return Result.Fail<PresignedUrl>(ResultCode.CHUNK_NOT_FOUND);
+            }
+
             var presignUrl = await storage.GenerateUploadUrlAsync(new GenerateUploadUrlCommand(
                 ProviderCode: taskCache.ProviderCode,
                 Bucket: taskCache.Bucket,
@@ -99,6 +98,12 @@ namespace FileService.Application.UploadFileTask
                 return Result.Fail<UploadTaskResponse>(ResultCode.CHUNK_NOT_FOUND);
             }
 
+
+            if(taskCache.Parts.Count < taskCache.TotalPartCount)
+            {
+                return Result.Fail<UploadTaskResponse>(ResultCode.CHUNK_COMBINE_FAIL);
+            }
+
             var task = await reposity.FindByIdAsync(Guid.Parse(taskCache.TaskId));
             //var res = await storage.CompleteUploadAsync(new CompleteUploadCommand(
             //    ProviderCode: taskCache.ProviderCode,
@@ -116,6 +121,7 @@ namespace FileService.Application.UploadFileTask
 
             await endpoint.Publish(new UploadTaskCompleteEvent()
             {
+                OperatorId = command.userId,
                 Bucket = taskCache.Bucket,
                 FileName = task.FileName.ToString(),
                 ObjectKey = taskCache.ObjectKey,
@@ -126,10 +132,39 @@ namespace FileService.Application.UploadFileTask
                 ProviderCode = taskCache.ProviderCode,
                 Region = taskCache.Region,
                 SessionId = command.UploadSessionId,
-                TaskId = task.Id
-            });
+                TaskId = task.Id,
+                FileSize = task.FileSize,
+                ContentType = task.ContentType.ToString(),
+                CheckSun = task.CheckSum.Value
+
+            }, cancellationToken);
 
             return Result.Success(mapper.Map<UploadTaskResponse>(task));
         }
+
+        public async Task<Result<CompleteUploadResult>> UploadPartAsync(UploadPartCommand command)
+        {
+
+            var taskCache = await redis.GetAsync(command.SessionId);
+            if(taskCache is null)
+            {
+                return Result.Fail<CompleteUploadResult>(ResultCode.CHUNK_NOT_FOUND);
+            }
+            await localChunkStorage.SavePartAsync(new SaveLocalPartCommand(
+                UploadSessionId: command.SessionId,
+                PartNumber: command.PartNum,
+                Stream: command.Stream,
+                ContentLength: command.ContentLength
+                ));
+            taskCache.AddOrUpdatePart(new StorageContracts.UploadPart(command.PartNum, command.PartNum.ToString(), command.ContentLength));
+            await redis.SetAsync(taskCache);
+            var location = new Domain.ValueObjects.StorageLocation(
+                storageProvider: taskCache.ProviderCode,
+                bucket: taskCache.Bucket,
+                objectKey: taskCache.ObjectKey,
+                region: taskCache.Region
+                );
+            return Result.Success(new CompleteUploadResult(location, command.PartNum.ToString(), command.ContentLength));
+        }
     }
 }

FileService.Application/UploadFileTask/UploadPartCommand.cs

@@ -0,0 +1,10 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace FileService.Application.UploadFileTask
+{
+    public record UploadPartCommand(Stream Stream, string SessionId, int PartNum, long ContentLength);
+}