fileupload/service.go

package fileupload

import (
	"crypto/md5"
	"errors"
	"fmt"
	"io"
	"mime/multipart"
	"os"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"time"
)

// service 核心业务，持 Options + Repository。
type service struct {
	opts Options
	repo Repository
}

func newService(opts Options, repo Repository) *service {
	return &service{opts: opts, repo: repo}
}

func (s *service) chunkDir(uploadID string) string {
	return filepath.Join(s.opts.ChunkRoot, uploadID)
}

func (s *service) fileURL(storagePath string) string {
	return s.opts.URLPrefix + "/" + storagePath
}

// ── init ────────────────────────────────────────────────────────────

type InitReq struct {
	FileName   string `json:"fileName" binding:"required,max=255"`
	FileSize   int64  `json:"fileSize" binding:"min=0"`
	FileMD5    string `json:"fileMD5" binding:"required,len=32"`
	MimeType   string `json:"mimeType" binding:"max=128"`
	ChunkSize  int64  `json:"chunkSize" binding:"required,min=1"`
	TotalChunk int    `json:"totalChunk" binding:"required,min=1"`
	// TargetDir 目标目录（相对 CloudDir，/ 分隔），空=根目录（固定路径模式）。
	TargetDir string `json:"targetDir" binding:"max=500"`
}

func (s *service) init(req *InitReq, operator string, tenantID int64) (any, error) {
	if req.ChunkSize <= 0 || req.ChunkSize > s.opts.maxChunkBytes() {
		return nil, fmt.Errorf("chunkSize 超出范围 (0, %d]", s.opts.maxChunkBytes())
	}
	if req.FileSize < 0 {
		return nil, errors.New("fileSize 非法")
	}
	if len(req.FileMD5) != 32 {
		return nil, errors.New("fileMD5 必须为 32 位十六进制")
	}
	if expect := expectChunks(req.FileSize, req.ChunkSize); req.TotalChunk != expect {
		return nil, fmt.Errorf("totalChunk(%d) 与 fileSize(%d)/chunkSize(%d) 不匹配，应为 %d",
			req.TotalChunk, req.FileSize, req.ChunkSize, expect)
	}
	if s.opts.MaxTotalSizeMB > 0 && req.FileSize > s.opts.MaxTotalSizeMB*1024*1024 {
		return nil, fmt.Errorf("文件超过总大小上限 %dMB", s.opts.MaxTotalSizeMB)
	}
	targetDir, err := SanitizeDir(req.TargetDir)
	if err != nil {
		return nil, err
	}

	// 1. 秒传：同 MD5 已完成任务
	done, err := s.repo.FindDoneByMD5(req.FileMD5)
	if err != nil {
		return nil, err
	}
	if done != nil && done.StoragePath != "" {
		if hit, ok := s.instantHit(done, targetDir); ok {
			if s.opts.OnInstantHit != nil {
				if err := s.opts.OnInstantHit(hit); err != nil {
					return nil, err
				}
			}
			return map[string]any{
				"uploadId":   "",
				"skipUpload": true,
				"fileUrl":    s.fileURL(hit.StoragePath),
			}, nil
		}
		// 秒传不可用（旧物理文件缺失/复制失败）→ 落入正常上传流程
	}

	// 2. 断点续传：同 MD5 进行中任务（目标目录不同则不复用，另起新任务）
	resuming, err := s.repo.FindUploadingByMD5(req.FileMD5)
	if err != nil {
		return nil, err
	}
	if resuming != nil && resuming.TargetDir == targetDir {
		indexes, err := s.repo.ListChunkIndexes(resuming.UploadId)
		if err != nil {
			return nil, err
		}
		return map[string]any{
			"uploadId":       resuming.UploadId,
			"skipUpload":     false,
			"uploadedChunks": indexes,
		}, nil
	}

	// 3. 新建任务
	name := sanitizeFileName(req.FileName)
	if name == "" {
		return nil, errors.New("文件名非法")
	}
	task := TaskInfo{
		UploadId:   newUploadID(),
		FileName:   name,
		FileSize:   req.FileSize,
		FileMd5:    req.FileMD5,
		MimeType:   req.MimeType,
		ChunkSize:  req.ChunkSize,
		TotalChunk: int32(req.TotalChunk),
		Status:     StatusUploading,
		TargetDir:  targetDir,
		Operator:   operator,
		TenantId:   tenantID,
	}
	if err := s.repo.CreateTask(&task); err != nil {
		return nil, err
	}
	return map[string]any{
		"uploadId":       task.UploadId,
		"skipUpload":     false,
		"uploadedChunks": []int32{},
	}, nil
}

// instantHit 处理秒传命中：
//   - 目标目录为空（固定模式）或与旧任务一致 → 直接复用旧 storagePath，行为与 v1.0.x 一致；
//   - 跨目录 → 把旧物理文件复制进目标目录，返回携带新 storagePath/TargetDir 的命中视图；
//
// ok=false 表示无法秒传（旧物理文件缺失/复制失败），调用方回退正常分片上传。
func (s *service) instantHit(done *TaskInfo, targetDir string) (TaskInfo, bool) {
	hit := *done
	if targetDir == "" || targetDir == done.TargetDir {
		return hit, true
	}
	src := s.physicalPath(done.StoragePath)
	if _, err := os.Stat(src); err != nil {
		return TaskInfo{}, false
	}
	name := path.Base(strings.ReplaceAll(done.StoragePath, "\\", "/"))
	dstDir := filepath.Join(s.opts.CloudDir, filepath.FromSlash(targetDir))
	if err := os.MkdirAll(dstDir, 0o766); err != nil {
		return TaskInfo{}, false
	}
	if err := copyFile(src, filepath.Join(dstDir, name)); err != nil {
		return TaskInfo{}, false
	}
	hit.StoragePath = buildStoragePath(s.opts.WebPathPrefix, targetDir, name)
	hit.TargetDir = targetDir
	return hit, true
}

// ── chunk ───────────────────────────────────────────────────────────

func (s *service) chunk(uploadID string, chunkIndex int, fh *multipart.FileHeader) (any, error) {
	if uploadID == "" {
		return nil, errors.New("缺少 uploadId")
	}
	task, err := s.repo.GetTaskByUploadID(uploadID)
	if err != nil {
		return nil, err
	}
	if task == nil {
		return nil, errors.New("上传任务不存在")
	}
	if task.Status != StatusUploading {
		return nil, fmt.Errorf("任务状态异常: %s", task.Status)
	}
	if chunkIndex < 0 || int32(chunkIndex) >= task.TotalChunk {
		return nil, fmt.Errorf("chunkIndex %d 超出范围 [0, %d)", chunkIndex, task.TotalChunk)
	}
	if fh.Size > s.opts.maxChunkBytes() {
		return nil, fmt.Errorf("分片超过单片上限 %d MB", s.opts.maxChunkBytes()/1024/1024)
	}

	// 幂等：已确认的分片直接成功
	exists, err := s.repo.ChunkExists(uploadID, int32(chunkIndex))
	if err != nil {
		return nil, err
	}
	if exists {
		return map[string]any{"chunkIndex": chunkIndex}, nil
	}

	dir := s.chunkDir(uploadID)
	if err := os.MkdirAll(dir, 0o766); err != nil {
		return nil, err
	}
	chunkPath := filepath.Join(dir, strconv.Itoa(chunkIndex))
	dst, err := os.Create(chunkPath)
	if err != nil {
		return nil, err
	}
	src, err := fh.Open()
	if err != nil {
		_ = dst.Close()
		return nil, err
	}
	hash := md5.New()
	written, copyErr := io.Copy(io.MultiWriter(dst, hash), src)
	closeErr := dst.Close()
	_ = src.Close()
	if copyErr != nil {
		return nil, copyErr
	}
	if closeErr != nil {
		return nil, closeErr
	}

	if err := s.repo.CreateChunk(&ChunkInfo{
		UploadId:   uploadID,
		ChunkIndex: int32(chunkIndex),
		ChunkSize:  written,
		ChunkMd5:   fmt.Sprintf("%x", hash.Sum(nil)),
	}); err != nil {
		// 并发下另一请求刚写入同分片：唯一冲突视为幂等成功
		again, e2 := s.repo.ChunkExists(uploadID, int32(chunkIndex))
		if e2 == nil && again {
			return map[string]any{"chunkIndex": chunkIndex}, nil
		}
		return nil, err
	}
	return map[string]any{"chunkIndex": chunkIndex}, nil
}

// ── merge ───────────────────────────────────────────────────────────

type MergeReq struct {
	UploadId string `json:"uploadId" binding:"required"`
}

func (s *service) merge(uploadID string) (any, error) {
	task, err := s.repo.GetTaskByUploadID(uploadID)
	if err != nil {
		return nil, err
	}
	if task == nil {
		return nil, errors.New("上传任务不存在")
	}
	// 目标目录在 init 时已 sanitize，此处防御性重校验（防宿主存脏数据）
	targetDir, err := SanitizeDir(task.TargetDir)
	if err != nil {
		return nil, err
	}

	// 幂等：已完成直接返回原地址；合并中拒绝重复触发
	if task.Status == StatusDone && task.StoragePath != "" {
		return map[string]any{"fileUrl": s.fileURL(task.StoragePath)}, nil
	}
	if task.Status == StatusMerging {
		return nil, errors.New("任务正在合并中，请勿重复提交")
	}

	count, err := s.repo.CountChunks(uploadID)
	if err != nil {
		return nil, err
	}
	if count < int(task.TotalChunk) {
		return nil, fmt.Errorf("分片不完整，已上传 %d/%d", count, task.TotalChunk)
	}

	_ = s.repo.SetTaskStatus(task.Id, StatusMerging)

	// 先写 .part，全部成功后改名，避免半成品文件残留
	partPath := filepath.Join(s.chunkDir(uploadID), "merged.part")
	part, err := os.Create(partPath)
	if err != nil {
		_ = s.repo.SetTaskStatus(task.Id, StatusError)
		return nil, err
	}
	hash := md5.New()
	for i := int32(0); i < task.TotalChunk; i++ {
		src, err := os.Open(filepath.Join(s.chunkDir(uploadID), strconv.Itoa(int(i))))
		if err != nil {
			_ = part.Close()
			_ = os.Remove(partPath)
			_ = s.repo.SetTaskStatus(task.Id, StatusError)
			return nil, fmt.Errorf("读取分片 %d 失败: %w", i, err)
		}
		_, copyErr := io.Copy(part, io.TeeReader(src, hash))
		_ = src.Close()
		if copyErr != nil {
			_ = part.Close()
			_ = os.Remove(partPath)
			_ = s.repo.SetTaskStatus(task.Id, StatusError)
			return nil, fmt.Errorf("合并分片 %d 失败: %w", i, copyErr)
		}
	}
	if err := part.Close(); err != nil {
		_ = os.Remove(partPath)
		_ = s.repo.SetTaskStatus(task.Id, StatusError)
		return nil, err
	}

	// 整文件 MD5 校验：保证秒传可信
	if actual := fmt.Sprintf("%x", hash.Sum(nil)); actual != task.FileMd5 {
		_ = os.Remove(partPath)
		_ = s.repo.SetTaskStatus(task.Id, StatusError)
		return nil, fmt.Errorf("合并结果校验失败：实际 MD5 %s 与声称的 %s 不一致", actual, task.FileMd5)
	}

	// 改名进宿主产物目录（targetDir 是 CloudDir 下的真实子目录，空=根目录）
	finalDir := s.opts.CloudDir
	if targetDir != "" {
		finalDir = filepath.Join(s.opts.CloudDir, filepath.FromSlash(targetDir))
	}
	if err := os.MkdirAll(finalDir, 0o766); err != nil {
		_ = os.Remove(partPath)
		_ = s.repo.SetTaskStatus(task.Id, StatusError)
		return nil, err
	}
	storageName := strconv.FormatInt(time.Now().UnixNano(), 10) + "_" + task.FileName
	// storagePath 是 Web 相对路径（WebPathPrefix + targetDir + 产物名），
	// 前缀只是 URL/拼接约定，targetDir 才是物理子目录
	finalPath := filepath.Join(finalDir, storageName)
	storagePath := buildStoragePath(s.opts.WebPathPrefix, targetDir, storageName)
	if err := os.Rename(partPath, finalPath); err != nil {
		_ = os.Remove(partPath)
		_ = s.repo.SetTaskStatus(task.Id, StatusError)
		return nil, err
	}

	task.Status = StatusDone
	task.StoragePath = storagePath
	if err := s.repo.UpdateTaskMerged(task.Id, storagePath); err != nil {
		return nil, err
	}

	// 宿主业务衔接（如写云盘记录行）；失败则 merge 整体失败
	if s.opts.OnMerged != nil {
		if err := s.opts.OnMerged(*task); err != nil {
			return nil, err
		}
	}

	// 分片目录异步清理
	go func(dir string) {
		_ = os.RemoveAll(dir)
	}(s.chunkDir(uploadID))

	return map[string]any{"fileUrl": s.fileURL(storagePath)}, nil
}

// ── check / task / abort ────────────────────────────────────────────

// check 秒传/续传预检，targetDir 与 init 同源：
//   - done 命中照常返回（init 会按目录做 copy-on-hit）；
//   - uploading 命中仅在目标目录一致时返回（目录不同时 init 会另起新任务，此处报 none）。
func (s *service) check(fileMD5, targetDir string) (any, error) {
	if len(fileMD5) != 32 {
		return map[string]any{"status": "none"}, nil
	}
	dir, err := SanitizeDir(targetDir)
	if err != nil {
		return map[string]any{"status": "none"}, nil
	}
	done, err := s.repo.FindDoneByMD5(fileMD5)
	if err != nil {
		return nil, err
	}
	if done != nil && done.StoragePath != "" {
		return map[string]any{
			"status":  StatusDone,
			"fileUrl": s.fileURL(done.StoragePath),
		}, nil
	}
	resuming, err := s.repo.FindUploadingByMD5(fileMD5)
	if err != nil {
		return nil, err
	}
	if resuming != nil && resuming.TargetDir == dir {
		indexes, err := s.repo.ListChunkIndexes(resuming.UploadId)
		if err != nil {
			return nil, err
		}
		return map[string]any{
			"status":         StatusUploading,
			"uploadId":       resuming.UploadId,
			"uploadedChunks": indexes,
		}, nil
	}
	return map[string]any{"status": "none"}, nil
}

func (s *service) taskInfo(uploadID string) (any, error) {
	task, err := s.repo.GetTaskByUploadID(uploadID)
	if err != nil {
		return nil, err
	}
	if task == nil {
		return nil, errors.New("任务不存在")
	}
	indexes, err := s.repo.ListChunkIndexes(uploadID)
	if err != nil {
		return nil, err
	}
	return map[string]any{"task": task, "uploadedChunks": indexes}, nil
}

func (s *service) abort(uploadID string) (any, error) {
	task, err := s.repo.GetTaskByUploadID(uploadID)
	if err != nil {
		return nil, err
	}
	if task == nil {
		return nil, errors.New("上传任务不存在")
	}
	if task.Status == StatusDone {
		return nil, errors.New("任务已完成，无需取消")
	}
	if task.Status == StatusMerging {
		return nil, errors.New("任务正在合并中，无法取消")
	}
	if err := s.repo.DeleteChunks(uploadID); err != nil {
		return nil, err
	}
	if err := s.repo.DeleteTaskByUploadID(uploadID); err != nil {
		return nil, err
	}
	go func(dir string) {
		_ = os.RemoveAll(dir)
	}(s.chunkDir(uploadID))
	return nil, nil
}