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

FileService.Application/EventHandler/UploadTaskCompleteEventHandler.cs

@@ -0,0 +1,45 @@
+using FileService.Application.Ports;
+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<UploadTaskCompleteEvent>
+    {
+        private readonly IObjectStorageRouter router;
+        private readonly IStorageRedisCache cache;
+
+        public UploadTaskCompleteEventHandler(IObjectStorageRouter router, IStorageRedisCache cache)
+        {
+            this.router = router;
+            this.cache = cache;
+        }
+
+        public async Task Consume(ConsumeContext<UploadTaskCompleteEvent> context)
+        {
+            var @event = context.Message;
+            var storage = router.Route(@event.ProviderCode);
+            if(@event.ProviderCode == "Local")
+            {
+                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);
+            }
+        }
+    }
+}

FileService.Application/UploadFileTask/UploadFileTaskService.cs

@@ -4,7 +4,9 @@ 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;
@@ -17,13 +19,16 @@ namespace FileService.Application.UploadFileTask
 {
     public class UploadFileTaskService(IUploadTaskReposity reposity, 
         IMapper mapper, IObjectStorageRouter router, 
-        IOptions<StorageOptions> options, IStorageRedisCache redis)
+        IOptions<StorageOptions> options, IStorageRedisCache redis,
+        IPublishEndpoint endpoint
+        )
     {
         private readonly IUploadTaskReposity reposity = reposity;
         private readonly IMapper mapper = mapper;
         private readonly IObjectStorageRouter router = router;
         private readonly IOptions<StorageOptions> options = options;
         private readonly IStorageRedisCache redis = redis;
+        private readonly IPublishEndpoint endpoint = endpoint;
         private readonly IObjectStoragePort storage = router.Route(options.Value.DefaultProviderCode);
 
         public async Task<Result<TaskInitResponse>> InitTaskAsync(UploadTaskInitCommand command)
@@ -34,9 +39,9 @@ namespace FileService.Application.UploadFileTask
             var storageOption = options.Value.Providers[options.Value.DefaultProviderCode];
             var storage = router.Route(storageOption.ProviderCode);
             var initRes = await storage.InitUploadAsync(new StorageContracts.InitiateUploadCommand(
-                ProviderCode: options.Value.ProviderCode,
+                ProviderCode: storageOption.ProviderCode,
-                Bucket: options.Value.Bucket,
+                Bucket: storageOption.Bucket,
-                ObjectKey: options.Value.Endpoint,
+                ObjectKey: storageOption.Endpoint,
                 ContentType:task.ContentType.Value,
                 ContentLength: command.FileSize,
                 null
@@ -95,20 +100,35 @@ namespace FileService.Application.UploadFileTask
             }
 
             var task = await reposity.FindByIdAsync(Guid.Parse(taskCache.TaskId));
-            var res = await storage.CompleteUploadAsync(new CompleteUploadCommand(
+            //var res = await storage.CompleteUploadAsync(new CompleteUploadCommand(
-                ProviderCode: taskCache.ProviderCode,
+            //    ProviderCode: taskCache.ProviderCode,
-                Bucket: taskCache.Bucket,
+            //    Bucket: taskCache.Bucket,
-                Region: taskCache.Region,
+            //    Region: taskCache.Region,
-                ObjectKey: taskCache.ObjectKey,
+            //    ObjectKey: taskCache.ObjectKey,
-                UploadSessionId: taskCache.UploadSessionId,
+            //    UploadSessionId: taskCache.UploadSessionId,
-                Parts: command.Parts
+            //    Parts: command.Parts
-                ), cancellationToken);
+            //    ), cancellationToken);
 
             task.CompleteUpload(new Domain.ValueObjects.StorageLocation(
                     taskCache.ProviderCode, taskCache.Bucket,
                     taskCache.ObjectKey, taskCache.Region
                 ));
 
+            await endpoint.Publish(new UploadTaskCompleteEvent()
+            {
+                Bucket = taskCache.Bucket,
+                FileName = task.FileName.ToString(),
+                ObjectKey = taskCache.ObjectKey,
+                Parts = command.Parts.Select(s => 
+                new IM.Commons.IntegrationEvents.UploadPart(
+                    s.PartNumber, s.ETag, s.Size, s.Checksum)
+                ).ToList(),
+                ProviderCode = taskCache.ProviderCode,
+                Region = taskCache.Region,
+                SessionId = command.UploadSessionId,
+                TaskId = task.Id
+            });
+
             return Result.Success(mapper.Map<UploadTaskResponse>(task));
         }
     }

FileService.Infrastructure/ModuleInit.cs

@@ -14,6 +14,7 @@ namespace FileService.Infrastructure
             services.AddScoped<IUploadFileReposity, UploadFileReposity>();
             services.AddScoped<IUploadTaskReposity, UploadTaskReposity>();
             services.AddScoped<IObjectStoragePort, LocalStorageAdapter>();
+            services.AddScoped<IRedisService, RedisCacheService>();
             services.AddScoped<IObjectStorageRouter, ObjectStorageRouter>();
             services.AddScoped<IStorageRedisCache,StorageCacheService>();
         }

FileService.Infrastructure/Storage/LocalStorageAdapter.cs

@@ -3,27 +3,73 @@ using FileService.Application.StorageContracts;
 using FileService.Domain.ValueObjects;
 using IM.Commons;
 using IM.InitCommon;
-using MassTransit;
-using MassTransit.Configuration;
 using Microsoft.Extensions.Options;
-using System;
+using StackExchange.Redis;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 
 namespace FileService.Infrastructure.Storage
 {
-    public class LocalStorageAdapter(IRedisService redis, IOptions<StorageOptions> options) : IObjectStoragePort
+    public class LocalStorageAdapter(IStorageRedisCache redis, IOptions<StorageOptions> options) : IObjectStoragePort
     {
-        private readonly IRedisService redis = redis;
+        private readonly IStorageRedisCache redis = redis;
         private readonly IOptions<StorageOptions> options = options;
 
         public string ProviderCode => "Local";
 
-        public Task<CompleteUploadResult> CompleteUploadAsync(CompleteUploadCommand command, CancellationToken token)
+        public async Task<CompleteUploadResult> CompleteUploadAsync(CompleteUploadCommand command, CancellationToken token)
         {
-            throw new NotImplementedException();
+            var res = await MergeAsync(command.UploadSessionId, command.ObjectKey, command.Parts);
+            return new CompleteUploadResult(new StorageLocation(
+                    storageProvider: command.ProviderCode,
+                    bucket: command.Bucket,
+                    objectKey: command.ObjectKey,
+                    region: command.Region
+                ), null, command.Parts.Sum(x => x.Size).Value);
+        }
+        public async Task<Result<object>> MergeAsync(string sessionId, string objectKey, IReadOnlyList<UploadPart> parts)
+        {
+            var rootPath = options.Value.Providers[options.Value.DefaultProviderCode].LocalRootPath;
+            var tempPath = Path.Combine(rootPath, "temp"); // 项目根目录下 uploads                                                                               // 最终文件存储路径（这里可以用你之前 ObjectNameGenerator 生成的名字）
+            var finalPath = Path.Combine(rootPath, objectKey);
+            var finalDir = Path.GetDirectoryName(finalPath);
+            Directory.CreateDirectory(finalDir);
+
+            var storageCache = await redis.GetAsync(sessionId);
+            var totalChunks = storageCache.TotalPartCount;
+            try
+            {
+                using (var finalStream = new FileStream(finalPath, FileMode.Create))
+                {
+                    for (var i = 1; i <= totalChunks; i++)
+                    {
+                        var progress = (i * 100.0 / totalChunks);
+                        if (i % 5 == 0 || i == totalChunks)
+                        {
+                            //await _redis.HashSetAsync(RedisKeys.MergeStatus(taskId), new HashEntry[]
+                            //{
+                            //    new("status", "processing"),
+                            //    new("progress", progress.ToString("F2"))
+                            //});
+                        }
+                        var chunkPath = Path.Combine(tempPath, $"{i}.part.tmp");
+                        if (!File.Exists(chunkPath))
+                            return Result.Fail(ResultCode.CHUNK_NOT_FOUND);
+                        using (var chunkStream = new FileStream(chunkPath, FileMode.Open))
+                        {
+                            await chunkStream.CopyToAsync(finalStream);
+                        }
+                    }
+                    Directory.Delete(tempPath, true);
+                    await redis.DeleteAsync(sessionId);
+                }
+
+                return Result.Success();
+            }
+            catch (Exception e)
+            {
+                //_logger.LogError(e, e.Message);
+                throw;
+
+            }
         }
 
         public async Task<PresignedUrl> GenerateUploadUrlAsync(GenerateUploadUrlCommand command, CancellationToken token)

IM.Commons/IntegrationEvents/UploadTaskCompleteEvent.cs

@@ -0,0 +1,25 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace IM.Commons.IntegrationEvents
+{
+    public class UploadTaskCompleteEvent
+    {
+        public Guid TaskId { get; set; }
+        public string SessionId { get; set; }
+        public string FileName { get; set; }
+        public string ProviderCode { get; set; }
+        public string Bucket { get; set; }
+        public string Region { get; set; }
+        public string ObjectKey { get; set; }
+        public IReadOnlyList<UploadPart> Parts { get; set; }
+    }
+    public sealed record UploadPart(
+        int PartNumber,
+        string ETag,
+        long? Size = null,
+        string? Checksum = null);
+}

MessageService.WebApi/Application/Message/MessageService.cs

@@ -73,8 +73,6 @@ namespace MessageService.WebApi.Application.Message
             {
             // 2. 处理引用逻辑（如果传了 QuoteMessageId）
             QuoteInfo? quote = null;
-            if (command.QuoteMessageId.HasValue)
-            {
                 var originMsg = await reposity.FindByIdAsync(command.QuoteMessageId.Value);
                 if (originMsg != null)
                 {
@@ -89,9 +87,6 @@ namespace MessageService.WebApi.Application.Message
                 }
                 message.WithQuote(quote);
             }
-
-            
-
             reposity.Create(message);
 
             return Result.Success(mapper.Map<MessageResponse>(message));