IM_NEW/FileService.Application/UploadFileTask/UploadFileTaskService.cs

using AutoMapper;
using FileService.Application.Ports;
using FileService.Application.StorageContracts;
using FileService.Domain.IReposities;
using IM.Commons;
using IM.Commons.IntegrationEvents;
using IM.InitCommon;
using MassTransit;
using Microsoft.Extensions.Options;

namespace FileService.Application.UploadFileTask
{
    public class UploadFileTaskService(IUploadTaskReposity reposity,
        IMapper mapper, IObjectStorageRouter router,
        IOptions<StorageOptions> options, IStorageRedisCache redis,
        IPublishEndpoint endpoint, ILocalChunkStorage localChunkStorage,
        IUploadFileReposity uploadFileReposity
        )
    {
        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 ILocalChunkStorage localChunkStorage = localChunkStorage;
        private readonly IUploadFileReposity uploadFileReposity = uploadFileReposity;
        private readonly IObjectStoragePort storage = router.Route(options.Value.DefaultProviderCode);

        public async Task<Result<TaskInitResponse>> InitTaskAsync(UploadTaskInitCommand command)
        {
            CancellationToken cancellationToken = CancellationToken.None;

            // 秒传：相同 checksum 的文件若已存在于已完成文件表，直接返回已有记录
            var existingFile = await uploadFileReposity.FindByCheckSumGlobalAsync("md5", command.checkSum);
            if (existingFile != null && CanReuse(existingFile, command))
            {
                var storageForResponse = router.Route(existingFile.StorageLocation.StorageProvider);
                var fileResponse = mapper.Map<UploadFile.FileResponse>(existingFile);
                fileResponse.Url = storageForResponse.GetPublicUrl(existingFile.StorageLocation);
                fileResponse.IsPublic = fileResponse.IsPublic || fileResponse.Url != null;
                return Result.Success(new TaskInitResponse
                {
                    TaskId = existingFile.Id,
                    UploadSessionId = existingFile.Id.ToString(),
                    StorageLocation = existingFile.StorageLocation,
                    Instant = true,
                    UploadMode = "Instant",
                    TotalPartCount = 0,
                    PartSizeBytes = 0,
                    File = fileResponse
                });
            }

            // 文件大小上限校验
            var maxSize = options.Value.Providers[options.Value.DefaultProviderCode].MaxObjectSizeBytes;
            if (command.FileSize > maxSize)
            {
                return Result.Fail<TaskInitResponse>(ResultCode.FILE_TOO_LARGE);
            }

            var task = command.ToUploadTask();
            var date = DateTime.Now;
            var storageOption = options.Value.Providers[options.Value.DefaultProviderCode];
            var storage = router.Route(storageOption.ProviderCode);
            var initUpdateCommand = new StorageContracts.InitiateUploadCommand(
                ProviderCode: storageOption.ProviderCode,
                Bucket: storageOption.Bucket,
                ObjectKey: $"{storageOption.LocalRootPath}\\{date.Year}\\{date.Month}\\{date.Day}\\{command.FileName}",
                ContentType: task.ContentType.Value,
                ContentLength: command.FileSize,
                null);
            var initRes = await storage.InitUploadAsync(initUpdateCommand, cancellationToken);

            var totalPartCount = (int)(task.FileSize % storageOption.DefaultPartSizeBytes > 0 ?
                (task.FileSize / storageOption.DefaultPartSizeBytes) + 1 :
                task.FileSize / storageOption.DefaultPartSizeBytes);

            var res = mapper.Map<TaskInitResponse>(initRes);
            res.TaskId = task.Id;
            res = new TaskInitResponse
            {
                TaskId = task.Id,
                UploadSessionId = initRes.UploadSessionId,
                StorageLocation = initRes.Location,
                Instant = false,
                UploadMode = string.Equals(storage.ProviderCode, "Local", StringComparison.OrdinalIgnoreCase)
                    ? "LocalMultipart"
                    : "Presigned",
                TotalPartCount = totalPartCount,
                PartSizeBytes = storageOption.DefaultPartSizeBytes
            };

            task.StartUpload();
            reposity.Create(task);

            await redis.SetAsync(new StorageContracts.UploadRuntimeCache(
                    taskId: task.Id.ToString(),
                    providerCode: storage.ProviderCode,
                    uploadSessionId: res.UploadSessionId,
                    bucket: storageOption.Bucket,
                    region: storageOption.Region,
                    objectKey: initUpdateCommand.ObjectKey,
                    fileSize: task.FileSize,
                    totalPartCount: totalPartCount
                ));

            return Result.Success(res);

        }

        public async Task<Result<PresignedUrl>> GenerateUrlAsync(string sessionId, int partNum, Guid userId, CancellationToken token = default)
        {
            var taskCache = await redis.GetAsync(sessionId);
            if (taskCache is null)
            {
                return Result.Fail<PresignedUrl>(ResultCode.CHUNK_NOT_FOUND);
            }

            var task = await reposity.FindByIdAsync(Guid.Parse(taskCache.TaskId));
            if (task is null || task.UploaderId != userId)
            {
                return Result.Fail<PresignedUrl>(ResultCode.PERMISSION_DENIED);
            }

            if (taskCache.TotalPartCount < partNum || partNum < 1)
            {
                return Result.Fail<PresignedUrl>(ResultCode.INVALID_PART_NUMBER);
            }

            var presignUrl = await storage.GenerateUploadUrlAsync(new GenerateUploadUrlCommand(
                ProviderCode: taskCache.ProviderCode,
                Bucket: taskCache.Bucket,
                ObjectKey: taskCache.ObjectKey,
                UploadSessionId: taskCache.UploadSessionId,
                PartNumber: partNum,
                ExpiresIn: options.Value.Providers[options.Value.DefaultProviderCode].UploadUrlExpiresIn
                ), token);

            return Result.Success(presignUrl);
        }

        public async Task<Result<UploadTaskResponse>> CompleteTaskAsync(UploadTaskCompleteCommand command, CancellationToken cancellationToken = default)
        {
            var taskCache = await redis.GetAsync(command.UploadSessionId);

            if (taskCache is null)
            {
                return Result.Fail<UploadTaskResponse>(ResultCode.CHUNK_NOT_FOUND);
            }

            var task = await reposity.FindByIdAsync(Guid.Parse(taskCache.TaskId));
            if (task is null)
            {
                return Result.Fail<UploadTaskResponse>(ResultCode.CHUNK_NOT_FOUND);
            }
            if (task.UploaderId != command.userId)
            {
                return Result.Fail<UploadTaskResponse>(ResultCode.PERMISSION_DENIED);
            }

            // 校验分片数量必须匹配
            if (command.Parts.Count != taskCache.TotalPartCount)
            {
                return Result.Fail<UploadTaskResponse>(ResultCode.PART_COUNT_MISMATCH);
            }

            var expectedPartNumbers = Enumerable.Range(1, taskCache.TotalPartCount).ToHashSet();
            if (!expectedPartNumbers.SetEquals(command.Parts.Select(x => x.PartNumber)))
            {
                return Result.Fail<UploadTaskResponse>(ResultCode.INVALID_PART_NUMBER);
            }

            // 本地分片必须由本服务接收；预签名模式由对象存储在完成合并时校验 ETag。
            if (string.Equals(taskCache.ProviderCode, "Local", StringComparison.OrdinalIgnoreCase))
            {
                foreach (var part in command.Parts)
                {
                    if (!taskCache.Parts.TryGetValue(part.PartNumber, out _))
                    {
                        return Result.Fail<UploadTaskResponse>(ResultCode.CHUNK_NOT_FOUND);
                    }
                }
            }

            if (task.State == Domain.UploadTaskState.Completed)
            {
                var completedResponse = mapper.Map<UploadTaskResponse>(task);
                if (task.ResultFileId.HasValue)
                {
                    var file = await uploadFileReposity.FindByIdAsync(task.ResultFileId.Value);
                    if (file != null)
                    {
                        completedResponse.File = mapper.Map<UploadFile.FileResponse>(file);
                        completedResponse.File.Url = router.Route(file.StorageLocation.StorageProvider)
                            .GetPublicUrl(file.StorageLocation);
                        completedResponse.File.IsPublic = completedResponse.File.IsPublic || completedResponse.File.Url != null;
                    }
                }
                return Result.Success(completedResponse);
            }

            task.StartMerging(new Domain.ValueObjects.StorageLocation(
                    taskCache.ProviderCode, taskCache.Bucket,
                    taskCache.ObjectKey, taskCache.Region
                ));

            await endpoint.Publish(new UploadTaskCompleteEvent()
            {
                OperatorId = command.userId,
                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,
                FileSize = task.FileSize,
                ContentType = task.ContentType.ToString(),
                CheckSun = task.CheckSum.Value
                ,ChatType = task.ChatType
                ,TargetId = task.TargetId

            }, cancellationToken);

            return Result.Success(mapper.Map<UploadTaskResponse>(task));
        }

        public async Task<Result<CompleteUploadResult>> UploadPartAsync(UploadPartCommand command, Guid userId)
        {
            var taskCache = await redis.GetAsync(command.SessionId);
            if (taskCache is null)
            {
                return Result.Fail<CompleteUploadResult>(ResultCode.CHUNK_NOT_FOUND);
            }
            var task = await reposity.FindByIdAsync(Guid.Parse(taskCache.TaskId));
            if (task is null || task.UploaderId != userId)
            {
                return Result.Fail<CompleteUploadResult>(ResultCode.PERMISSION_DENIED);
            }

            var minPartSize = options.Value.Providers[options.Value.DefaultProviderCode].MinPartSizeBytes;

            // 最后一个分片豁免最小值校验（仅校验非最后一片）
            var isLastPart = command.PartNum == taskCache.TotalPartCount;
            if (!isLastPart && command.ContentLength < minPartSize)
            {
                return Result.Fail<CompleteUploadResult>(ResultCode.PART_TOO_SMALL,
                    $"分片 {command.PartNum} 大小为 {command.ContentLength} 字节，小于最小值 {minPartSize} 字节");
            }

            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));
        }

        /// <summary>
        /// 查询分片上传进度。
        /// </summary>
        public async Task<Result<UploadProgressResponse>> GetProgressAsync(string sessionId, Guid userId)
        {
            var taskCache = await redis.GetAsync(sessionId);
            if (taskCache is null)
            {
                return Result.Fail<UploadProgressResponse>(ResultCode.CHUNK_NOT_FOUND);
            }
            var task = await reposity.FindByIdAsync(Guid.Parse(taskCache.TaskId));
            if (task is null || task.UploaderId != userId)
            {
                return Result.Fail<UploadProgressResponse>(ResultCode.PERMISSION_DENIED);
            }

            var response = new UploadProgressResponse
            {
                SessionId = sessionId,
                TaskId = taskCache.TaskId,
                FileSize = taskCache.FileSize,
                TotalPartCount = taskCache.TotalPartCount,
                CompletedPartCount = taskCache.Parts.Count,
                UploadedBytes = taskCache.UploadedBytes,
                ProgressPercent = taskCache.FileSize > 0
                    ? (int)(taskCache.UploadedBytes * 100 / taskCache.FileSize)
                    : 0
            };

            return Result.Success(response);
        }

        private bool CanReuse(Domain.Entities.UploadFile file, UploadTaskInitCommand command)
        {
            var publicUrl = router.Route(file.StorageLocation.StorageProvider).GetPublicUrl(file.StorageLocation);
            if (file.IsPublic || publicUrl != null) return true;
            if (file.OwnerId == command.UploaderId)
            {
                return string.Equals(file.ChatType, command.ChatType, StringComparison.OrdinalIgnoreCase) &&
                    file.TargetId == command.TargetId;
            }
            if (string.Equals(file.ChatType, "GROUP", StringComparison.OrdinalIgnoreCase))
            {
                return string.Equals(command.ChatType, "GROUP", StringComparison.OrdinalIgnoreCase) &&
                    file.TargetId == command.TargetId;
            }
            if (string.Equals(file.ChatType, "PRIVATE", StringComparison.OrdinalIgnoreCase))
            {
                return string.Equals(command.ChatType, "PRIVATE", StringComparison.OrdinalIgnoreCase) &&
                    file.TargetId == command.UploaderId && command.TargetId == file.OwnerId;
            }
            return false;
        }

        public async Task<Result<UploadTaskResponse>> GetStatusAsync(Guid taskId, Guid userId)
        {
            var task = await reposity.FindByIdAsync(taskId);
            if (task is null)
            {
                return Result.Fail<UploadTaskResponse>(ResultCode.CHUNK_NOT_FOUND);
            }
            if (task.UploaderId != userId)
            {
                return Result.Fail<UploadTaskResponse>(ResultCode.PERMISSION_DENIED);
            }

            var response = mapper.Map<UploadTaskResponse>(task);
            if (task.ResultFileId.HasValue)
            {
                var file = await uploadFileReposity.FindByIdAsync(task.ResultFileId.Value);
                if (file != null)
                {
                    response.File = mapper.Map<UploadFile.FileResponse>(file);
                    response.File.Url = router.Route(file.StorageLocation.StorageProvider)
                        .GetPublicUrl(file.StorageLocation);
                    response.File.IsPublic = response.File.IsPublic || response.File.Url != null;
                }
            }
            return Result.Success(response);
        }
    }
}