存储引擎模块实现教程

// 文件: internal/biz/storage.go

// Storage 定义了文件存储操作的接口
// biz 层仅依赖此接口——不依赖任何具体的存储实现
type Storage interface {
    // Upload 从读取器存储文件并返回存储路径
    // 返回的 storagePath 用于后续操作中标识文件
    Upload(ctx context.Context, filePath string, reader io.Reader) (storagePath string, err error)

    // Download 返回给定存储路径的文件的 ReadCloser
    // 调用方使用完毕后必须 Close 返回的 ReadCloser，避免文件描述符泄漏
    Download(ctx context.Context, storagePath string) (io.ReadCloser, error)

    // Delete 从存储中永久删除文件
    // 文件不存在时实现层应返回 nil，保证幂等性
    Delete(ctx context.Context, storagePath string) error

    // InitMultipartUpload 初始化分片上传会话
    // 使用调用方传入的 uploadID 作为分片目录标识，确保与 biz 层会话 ID 一致
    // 这样设计避免了 storage 层自己生成 ID 与 biz 层不一致导致的孤儿空目录
    InitMultipartUpload(ctx context.Context, uploadID, fileName string) error

    // UploadPart 上传分片上传的单个分片
    // 返回已上传分片的元数据（PartNumber / Hash / Size）
    UploadPart(ctx context.Context, uploadID string, partNumber int, reader io.Reader) (PartInfo, error)

    // CompleteMultipartUpload 完成分片上传并合并所有分片
    // 返回合并后文件的最终存储路径
    CompleteMultipartUpload(ctx context.Context, uploadID string, parts []PartInfo) (storagePath string, err error)

    // AbortMultipartUpload 取消分片上传并清理临时分片
    AbortMultipartUpload(ctx context.Context, uploadID string) error

    // CleanupOrphanChunks 清理孤儿分片目录
    // 扫描 chunks 目录，删除修改时间早于 maxAge 的子目录
    // 返回被清理的 uploadID 列表，便于上层记录日志或回调
    CleanupOrphanChunks(ctx context.Context, maxAge time.Duration) ([]string, error)

    // GetFileURL 返回可用于文件预览或分享的访问 URL
    // 本地存储返回本地路径；OSS 返回签名 URL
    GetFileURL(ctx context.Context, storagePath string, expire time.Duration) (string, error)
}

// PartInfo 保存已上传分片的元数据
// CompleteMultipartUpload 时会收到 []PartInfo 用于按序合并
type PartInfo struct {
    PartNumber int    // 分片编号，从 1 开始，决定合并顺序
    Hash       string // 分片哈希 / ETag，用于校验完整性
    Size       int64  // 分片字节数
}

// 常见的存储错误
// 所有实现层共用这套错误，便于 biz 层用 errors.Is 精确判断
var (
    ErrStorageNotImplemented = errors.New("storage: not implemented") // OSS stub 默认返回此错误
    ErrStorageFileNotFound   = errors.New("storage: file not found")  // 下载 / 删除时文件不存在
    ErrUploadNotFound        = errors.New("storage: upload session not found")
    ErrPartAlreadyUploaded   = errors.New("storage: part already uploaded") // 同一分片重复上传
)

// 文件: internal/data/storage/storage.go

// ProviderSet 是存储引擎的 Wire 提供集合。
// Wire 在编译期生成依赖注入代码时，会从此集合中找到 NewStorage 并注入。
var ProviderSet = wire.NewSet(NewStorage)

// NewStorage 根据配置中的 driver 字段选择对应的存储实现。
// 支持 "local" 和 "minio" 两种驱动。
// 后续扩展 OSS 时，只需在此 switch 增加 case "oss": return NewOssStorage(c.Oss)
func NewStorage(c *conf.Storage) (biz.Storage, error) {
    if c == nil {
        return nil, fmt.Errorf("storage: 存储配置为空")
    }

    switch c.Driver {
    case "minio":
        return NewMinioStorage(c.Minio)
    case "local", "":
        return NewLocalStorage(c.Local), nil
    default:
        return nil, fmt.Errorf("storage: 不支持的存储驱动: %s", c.Driver)
    }
}

// 文件: internal/data/storage/local.go

// 编译期确保 localStorage 实现了 biz.Storage 接口。
// 若接口新增方法而 localStorage 未实现，编译阶段就会报错，避免运行时才发现遗漏。
var _ biz.Storage = (*localStorage)(nil)

// localStorage 使用本地文件系统实现 Storage 接口。
type localStorage struct {
    uploadDir   string // 上传根目录，例如 ./data/uploads
    maxFileSize int64  // 单文件最大字节数
}

// NewLocalStorage 创建一个新的本地文件系统存储处理器。
// c 为空时使用默认值 ./data/uploads 与 100MB
func NewLocalStorage(c *conf.Storage_Local) biz.Storage {
    dir := "./data/uploads"
    var maxSize int64 = 100 * 1024 * 1024 // 默认 100MB
    if c != nil {
        if c.UploadDir != "" {
            dir = c.UploadDir
        }
        if c.MaxFileSize != "" {
            fmt.Sscanf(c.MaxFileSize, "%d", &maxSize) // 从字符串解析为 int64
        }
    }

    // 确保上传目录存在，0755 表示 owner 可读写执行，其他人可读可执行
    os.MkdirAll(dir, 0755)

    return &localStorage{
        uploadDir:   dir,
        maxFileSize: maxSize,
    }
}

// datePath 返回按日期分割的路径：uploadDir/2026/08/04/
// 使用 filepath.Join 保证跨平台路径分隔符正确（Windows 用 \，Linux/Mac 用 /）
func (s *localStorage) datePath() string {
    now := time.Now()
    return filepath.Join(s.uploadDir,
        fmt.Sprintf("%d", now.Year()),        // 2026
        fmt.Sprintf("%02d", now.Month()),     // 08
        fmt.Sprintf("%02d", now.Day()),       // 04
    )
}

// Upload 从读取器存储文件到本地文件系统。
// filePath 通常为 "uuid.extension"，作为当天目录内的文件名
func (s *localStorage) Upload(ctx context.Context, filePath string, reader io.Reader) (string, error) {
    dir := s.datePath()
    // 递归创建日期目录，已存在不会报错
    if err := os.MkdirAll(dir, 0755); err != nil {
        return "", fmt.Errorf("storage: failed to create upload dir: %w", err)
    }

    // 使用 filePath 作为日期目录内的相对文件名
    // filePath 通常为 "uuid.extension"
    destPath := filepath.Join(dir, filePath)

    // os.Create 若文件已存在会截断，否则新建
    f, err := os.Create(destPath)
    if err != nil {
        return "", fmt.Errorf("storage: failed to create file: %w", err)
    }
    defer f.Close()

    // io.Copy 内部维护 32KB 缓冲区，从 reader 读一块写一块
    // 整个过程不需要把文件全部加载到内存，适合大文件
    if _, err := io.Copy(f, reader); err != nil {
        return "", fmt.Errorf("storage: failed to write file: %w", err)
    }

    return destPath, nil
}

// Download 返回给定存储路径的文件的 ReadCloser。
// 调用方必须 Close 返回值，否则文件描述符会泄漏
func (s *localStorage) Download(ctx context.Context, storagePath string) (io.ReadCloser, error) {
    f, err := os.Open(storagePath)
    if err != nil {
        if os.IsNotExist(err) {
            // 文件不存在时返回业务错误，便于上层用 errors.Is 判断
            return nil, biz.ErrStorageFileNotFound
        }
        return nil, fmt.Errorf("storage: failed to open file: %w", err)
    }
    return f, nil
}

// Delete 从文件系统中删除文件。
// 文件不存在直接返回 nil，保证删除接口幂等
func (s *localStorage) Delete(ctx context.Context, storagePath string) error {
    if err := os.Remove(storagePath); err != nil {
        if os.IsNotExist(err) {
            return nil // 已删除，不算错误
        }
        return fmt.Errorf("storage: failed to delete file: %w", err)
    }
    return nil
}

// GetFileURL 返回包装为 file:// URL 的本地文件路径。
// 本地存储没有 HTTP 服务，因此用 file:// 协议让浏览器或客户端直接访问本地路径
func (s *localStorage) GetFileURL(ctx context.Context, storagePath string, expire time.Duration) (string, error) {
    if storagePath == "" {
        return "", biz.ErrFileNotFound
    }
    // 转为绝对路径，避免相对路径在不同工作目录下失效
    absPath, err := filepath.Abs(storagePath)
    if err != nil {
        return "", err
    }
    return "file://" + absPath, nil
}

// ChunkPath 返回特定分片文件的路径。
// 此方法非接口方法，提供给需要直接读取分片的内部逻辑使用
func (s *localStorage) ChunkPath(uploadID string, partNumber int) string {
    // %05d 表示左补零至 5 位，例如 part_00001、part_00012
    // 这样字典序与数字序一致，便于按文件名排序
    return filepath.Join(s.uploadDir, "chunks", uploadID, fmt.Sprintf("part_%05d", partNumber))
}

// 文件: internal/data/storage/minio.go

// 编译期确保 minioStorage 实现了 biz.Storage 接口。
var _ biz.Storage = (*minioStorage)(nil)

// minioStorage 使用 MinIO 对象存储实现 Storage 接口。
type minioStorage struct {
    client   *minio.Client // MinIO / S3 客户端
    bucket   string        // bucket 名称，例如 cloud-disk
    location string        // region，例如 us-east-1
}

// NewMinioStorage 创建一个新的 MinIO 存储处理器。
// 启动时自动检查并创建 bucket，避免部署时手动准备
func NewMinioStorage(c *conf.Storage_Minio) (biz.Storage, error) {
    if c == nil {
        return nil, fmt.Errorf("storage: minio 配置为空")
    }

    useSSL := c.UseSsl
    // credentials.NewStaticV4 使用静态 AK/SK 签名
    client, err := minio.New(c.Endpoint, &minio.Options{
        Creds:  credentials.NewStaticV4(c.AccessKey, c.SecretKey, ""),
        Secure: useSSL,
        Region: c.Region,
    })
    if err != nil {
        return nil, fmt.Errorf("storage: 创建 minio 客户端失败: %w", err)
    }

    bucket := c.BucketName
    if bucket == "" {
        bucket = "cloud-disk" // 默认 bucket
    }
    location := c.Region
    if location == "" {
        location = "us-east-1" // S3 默认 region
    }

    // 启动时确保 bucket 存在，避免运行时才发现配置错误
    ctx := context.Background()
    exists, err := client.BucketExists(ctx, bucket)
    if err != nil {
        return nil, fmt.Errorf("storage: 检查 minio bucket 失败: %w", err)
    }
    if !exists {
        if err := client.MakeBucket(ctx, bucket, minio.MakeBucketOptions{Region: location}); err != nil {
            return nil, fmt.Errorf("storage: 创建 minio bucket 失败: %w", err)
        }
        slog.Info("minio bucket 创建成功", "bucket", bucket)
    }

    slog.Info("minio 存储初始化成功", "endpoint", c.Endpoint, "bucket", bucket)

    return &minioStorage{
        client:   client,
        bucket:   bucket,
        location: location,
    }, nil
}

// dateKey 返回按日期分割的对象前缀：2026/08/04/
// 注意 MinIO 用的是 / 分隔的字符串，而非 filepath.Join（避免 Windows 平台 \ 污染）
func (s *minioStorage) dateKey() string {
    now := time.Now()
    return fmt.Sprintf("%d/%02d/%02d", now.Year(), int(now.Month()), now.Day())
}

// Upload 从读取器存储文件到 MinIO。
// objectKey 格式：2026/08/04/uuid.ext
// size 传 -1 表示未知大小，MinIO SDK 会分块上传
func (s *minioStorage) Upload(ctx context.Context, filePath string, reader io.Reader) (string, error) {
    objectKey := fmt.Sprintf("%s/%s", s.dateKey(), filePath)
    _, err := s.client.PutObject(ctx, s.bucket, objectKey, reader, -1, minio.PutObjectOptions{
        ContentType: "application/octet-stream", // 二进制流默认类型
    })
    if err != nil {
        return "", fmt.Errorf("storage: minio 上传文件失败: %w", err)
    }
    return objectKey, nil
}

// Download 返回给定存储路径的文件的 ReadCloser。
// 注意：GetObject 返回的 obj 在实际读取前不会报错，必须调用 Stat() 验证
func (s *minioStorage) Download(ctx context.Context, storagePath string) (io.ReadCloser, error) {
    obj, err := s.client.GetObject(ctx, s.bucket, storagePath, minio.GetObjectOptions{})
    if err != nil {
        return nil, fmt.Errorf("storage: minio 获取对象失败: %w", err)
    }
    // 尝试读取以验证对象是否存在
    if _, err := obj.Stat(); err != nil {
        obj.Close()
        resp := minio.ToErrorResponse(err)
        if resp.Code == "NoSuchKey" {
            return nil, biz.ErrStorageFileNotFound // 转为业务错误
        }
        return nil, fmt.Errorf("storage: minio 获取对象信息失败: %w", err)
    }
    return obj, nil
}

// Delete 从 MinIO 中删除文件。
// NoSuchKey 视为成功，保证幂等
func (s *minioStorage) Delete(ctx context.Context, storagePath string) error {
    err := s.client.RemoveObject(ctx, s.bucket, storagePath, minio.RemoveObjectOptions{})
    if err != nil {
        resp := minio.ToErrorResponse(err)
        if resp.Code == "NoSuchKey" {
            return nil
        }
        return fmt.Errorf("storage: minio 删除对象失败: %w", err)
    }
    return nil
}

// InitMultipartUpload 初始化分片上传会话。
// MinIO 方式下使用对象前缀模拟分片目录，无需额外初始化。
// 放入一个标记对象 .init 以标识会话存在，便于排查
func (s *minioStorage) InitMultipartUpload(ctx context.Context, uploadID, fileName string) error {
    markerKey := fmt.Sprintf("chunks/%s/.init", uploadID)
    // 写入 uploadID 字符串作为标记内容，便于后续排查
    _, err := s.client.PutObject(ctx, s.bucket, markerKey, strings.NewReader(uploadID), int64(len(uploadID)), minio.PutObjectOptions{})
    if err != nil {
        return fmt.Errorf("storage: minio 初始化分片上传失败: %w", err)
    }
    return nil
}

// UploadPart 上传分片上传的单个分片。
// 每个分片是一个独立对象：chunks/{uploadID}/part_00001
func (s *minioStorage) UploadPart(ctx context.Context, uploadID string, partNumber int, reader io.Reader) (biz.PartInfo, error) {
    partKey := fmt.Sprintf("chunks/%s/part_%05d", uploadID, partNumber)

    // 检查分片是否已存在（断点续传时同一分片可能重复上传）
    _, err := s.client.StatObject(ctx, s.bucket, partKey, minio.StatObjectOptions{})
    if err == nil {
        return biz.PartInfo{}, biz.ErrPartAlreadyUploaded // 已上传，拒绝重复写
    }

    // info.ETag 是 MinIO 返回的分片哈希，合并时可作为校验依据
    info, err := s.client.PutObject(ctx, s.bucket, partKey, reader, -1, minio.PutObjectOptions{
        ContentType: "application/octet-stream",
    })
    if err != nil {
        return biz.PartInfo{}, fmt.Errorf("storage: minio 上传分片失败: %w", err)
    }

    return biz.PartInfo{
        PartNumber: partNumber,
        Hash:       info.ETag, // S3 协议中 ETag 通常即分片 MD5
        Size:       info.Size,
    }, nil
}

// CompleteMultipartUpload 将所有已上传的分片合并为最终文件。
// 使用 MinIO 的 ComposeObject API 将多个分片对象在服务端合并为一个，
// 整个过程不经过应用服务器内存，是真正的「服务端合并」
func (s *minioStorage) CompleteMultipartUpload(ctx context.Context, uploadID string, parts []biz.PartInfo) (string, error) {
    if len(parts) == 0 {
        return "", fmt.Errorf("storage: 没有分片可合并")
    }

    // 按分片编号排序，保证合并后的字节顺序正确
    sort.Slice(parts, func(i, j int) bool {
        return parts[i].PartNumber < parts[j].PartNumber
    })

    // 构建最终对象路径：2026/08/04/{uploadID}.merged
    finalKey := fmt.Sprintf("%s/%s.merged", s.dateKey(), uploadID)

    // 构建 Compose 源：每个分片对象作为一个 CopySrcOptions
    sources := make([]minio.CopySrcOptions, len(parts))
    for i, part := range parts {
        sources[i] = minio.CopySrcOptions{
            Bucket: s.bucket,
            Object: fmt.Sprintf("chunks/%s/part_%05d", uploadID, part.PartNumber),
        }
    }

    // ComposeObject 在 MinIO 服务端把多个对象顺序拼接成一个新对象
    dst := minio.CopyDestOptions{
        Bucket: s.bucket,
        Object: finalKey,
    }
    _, err := s.client.ComposeObject(ctx, dst, sources...)
    if err != nil {
        return "", fmt.Errorf("storage: minio 合并分片失败: %w", err)
    }

    // 清理分片对象 — 使用独立后台 context 避免请求取消后清理中断
    // 例如客户端取消请求后请求 ctx 会被 cancel，若用请求 ctx 清理会半途中断
    go s.cleanupChunks(context.Background(), uploadID)

    return finalKey, nil
}

// AbortMultipartUpload 取消分片上传并清理临时分片。
func (s *minioStorage) AbortMultipartUpload(ctx context.Context, uploadID string) error {
    return s.cleanupChunks(ctx, uploadID)
}

// cleanupChunks 清理指定 uploadID 的所有分片对象。
// 使用 RemoveObjects 批量接口，避免逐个删除的网络往返开销
func (s *minioStorage) cleanupChunks(ctx context.Context, uploadID string) error {
    prefix := fmt.Sprintf("chunks/%s/", uploadID)
    // ListObjects 返回一个 channel，流式列举，适合海量对象
    objectsCh := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{
        Prefix:    prefix,
        Recursive: true, // 递归列举子「目录」
    })

    // RemoveObjects 接收对象 channel，批量删除并返回错误 channel
    for err := range s.client.RemoveObjects(ctx, s.bucket, objectsCh, minio.RemoveObjectsOptions{}) {
        if err.Err != nil {
            slog.Warn("清理 MinIO 分片对象失败", "uploadID", uploadID, "err", err.Err)
        }
    }
    return nil
}

// GetFileURL 返回 MinIO 预签名 URL，用于文件预览或分享。
// 预签名 URL 内嵌签名与过期时间，持有者无需 AK/SK 即可下载
func (s *minioStorage) GetFileURL(ctx context.Context, storagePath string, expire time.Duration) (string, error) {
    if storagePath == "" {
        return "", biz.ErrFileNotFound
    }
    // PresignedGetObject 生成 GET 方法的预签名 URL
    presignedURL, err := s.client.PresignedGetObject(ctx, s.bucket, storagePath, expire, nil)
    if err != nil {
        return "", fmt.Errorf("storage: minio 生成预签名 URL 失败: %w", err)
    }
    return presignedURL.String(), nil
}

// 文件: internal/data/storage/oss_stub.go

package storage

import (
    "context"
    "io"
    "time"

    "cloud-disk/internal/biz"
)

// Ensure ossStorage implements biz.Storage at compile time.
// 编译期断言：确保 ossStorage 实现了 biz.Storage 接口
// 即便所有方法都返回 ErrStorageNotImplemented，也必须实现全部方法签名
var _ biz.Storage = (*ossStorage)(nil)

// ossStorage provides a stub implementation of the Storage interface for Aliyun OSS.
// TODO(oss): Replace stub with real implementation using github.com/aliyun/aliyun-oss-go-sdk
//
// Required dependencies (add to go.mod when implementing):
//   github.com/aliyun/aliyun-oss-go-sdk/v3
//
// Configuration fields (from conf.Storage_Oss):
//   - Endpoint:        OSS endpoint URL (e.g., "oss-cn-hangzhou.aliyuncs.com")
//   - AccessKeyId:     Aliyun RAM access key ID
//   - AccessKeySecret: Aliyun RAM access key secret
//   - BucketName:      OSS bucket name
//   - Region:          OSS region (e.g., "cn-hangzhou")
//
// Key implementation notes:
//   - Upload:   Use bucket.PutObject() with io.Reader
//   - Download: Use bucket.GetObject() returning io.ReadCloser
//   - Delete:   Use bucket.DeleteObject()
//   - InitMultipartUpload:   Use bucket.InitiateMultipartUpload()
//   - UploadPart:            Use bucket.UploadPart() with upload ID and part number
//   - CompleteMultipartUpload: Use bucket.CompleteMultipartUpload() with all ETags
//   - AbortMultipartUpload: Use bucket.AbortMultipartUpload()
//   - GetFileURL: Use bucket.SignURL() with the given expiration duration
type ossStorage struct {
    // TODO(oss): Add fields:
    //   client    *oss.Client
    //   bucket    *oss.Bucket
    //   bucketName string
}

// NewOssStorage creates a new OSS storage handler stub.
// TODO(oss): Initialize the OSS client with credentials from conf.
// 后续实现时签名应改为 NewOssStorage(c *conf.Storage_Oss) (biz.Storage, error)
func NewOssStorage() biz.Storage {
    return &ossStorage{}
}

// Upload — TODO: bucket.PutObject(filePath, reader)
func (s *ossStorage) Upload(ctx context.Context, filePath string, reader io.Reader) (string, error) {
    return "", biz.ErrStorageNotImplemented
}

// Download — TODO: return bucket.GetObject(storagePath)
func (s *ossStorage) Download(ctx context.Context, storagePath string) (io.ReadCloser, error) {
    return nil, biz.ErrStorageNotImplemented
}

// Delete — TODO: bucket.DeleteObject(storagePath)
func (s *ossStorage) Delete(ctx context.Context, storagePath string) error {
    return biz.ErrStorageNotImplemented
}

// InitMultipartUpload — TODO: result, err := bucket.InitiateMultipartUpload(fileName)
// 使用 biz 层传入的 uploadID 作为本地会话标识
func (s *ossStorage) InitMultipartUpload(ctx context.Context, uploadID, fileName string) error {
    return biz.ErrStorageNotImplemented
}

// UploadPart — TODO: part, err := bucket.UploadPart(initResult, reader, partSize, partNumber)
// return biz.PartInfo{PartNumber: partNumber, Hash: part.ETag, Size: partSize}, nil
func (s *ossStorage) UploadPart(ctx context.Context, uploadID string, partNumber int, reader io.Reader) (biz.PartInfo, error) {
    return biz.PartInfo{}, biz.ErrStorageNotImplemented
}

// CompleteMultipartUpload — TODO: Build oss.Parts from parts,
// call bucket.CompleteMultipartUpload(initResult, ossParts), return storagePath, nil
func (s *ossStorage) CompleteMultipartUpload(ctx context.Context, uploadID string, parts []biz.PartInfo) (string, error) {
    return "", biz.ErrStorageNotImplemented
}

// AbortMultipartUpload — TODO: bucket.AbortMultipartUpload(initResult)
func (s *ossStorage) AbortMultipartUpload(ctx context.Context, uploadID string) error {
    return biz.ErrStorageNotImplemented
}

// CleanupOrphanChunks — TODO: 列举并清理 OSS 上的孤儿分片对象（前缀 chunks/），
// 可通过 ListObjectsV2 + DeleteObjects 实现。
func (s *ossStorage) CleanupOrphanChunks(ctx context.Context, maxAge time.Duration) ([]string, error) {
    return nil, biz.ErrStorageNotImplemented
}

// GetFileURL — TODO: signedURL, err := bucket.SignURL(storagePath, oss.HTTPGet, expire)
// return signedURL, nil
func (s *ossStorage) GetFileURL(ctx context.Context, storagePath string, expire time.Duration) (string, error) {
    return "", biz.ErrStorageNotImplemented
}

// 文件: internal/conf/conf.proto

message Storage {
  string driver = 1;        // 存储驱动：local / minio / oss
  message Local {
    string upload_dir = 1;
    string max_file_size = 2;
    int32 concurrent_upload = 3;
  }
  message Oss {              // 阿里云 OSS 配置
    string endpoint = 1;            // 例如 oss-cn-hangzhou.aliyuncs.com
    string access_key_id = 2;
    string access_key_secret = 3;
    string bucket_name = 4;
    string region = 5;              // 例如 cn-hangzhou
  }
  message Minio {
    string endpoint = 1;
    string access_key = 2;
    string secret_key = 3;
    string bucket_name = 4;
    string region = 5;
    bool use_ssl = 6;
  }
  Local local = 2;
  Oss oss = 3;
  Minio minio = 4;
}

// 文件: internal/data/storage/local.go

// CleanupOrphanChunks 扫描 chunks 目录并删除修改时间早于 maxAge 的子目录。
// 这些残留目录通常由服务异常崩溃或客户端断开未调用取消接口导致。
// 返回被清理的 uploadID 列表。
func (s *localStorage) CleanupOrphanChunks(ctx context.Context, maxAge time.Duration) ([]string, error) {
    chunksRoot := filepath.Join(s.uploadDir, "chunks")

    // os.ReadDir 是 Go 1.16+ 引入的高效列举方法，
    // 返回 []os.DirEntry 而非 []os.FileInfo，不会预读取每个条目的完整信息
    entries, err := os.ReadDir(chunksRoot)
    if err != nil {
        if os.IsNotExist(err) {
            // chunks 目录不存在，说明没有任何分片上传发生过
            return nil, nil
        }
        return nil, fmt.Errorf("storage: failed to read chunks dir: %w", err)
    }

    now := time.Now()
    // 预分配容量，减少 append 时的扩容拷贝
    removed := make([]string, 0, len(entries))

    for _, entry := range entries {
        // 检查上下文是否被取消，支持长扫描过程中调用方中止
        if ctx.Err() != nil {
            return removed, ctx.Err()
        }
        // 只处理目录，跳过可能存在的杂散文件
        if !entry.IsDir() {
            continue
        }

        uploadID := entry.Name()
        dirPath := filepath.Join(chunksRoot, uploadID)

        // DirEntry.Info() 按需获取条目信息，比传统 ReadDir 更省资源
        info, err := entry.Info()
        if err != nil {
            // 无法获取信息，跳过（不删除以避免误操作）
            continue
        }

        // 使用目录的 ModTime 判断；超过 maxAge 视为孤儿分片
        // ModTime 是目录最后被修改的时间（如分片写入会更新它）
        if now.Sub(info.ModTime()) > maxAge {
            if err := os.RemoveAll(dirPath); err != nil {
                // 记录错误但继续处理其他目录，避免一个失败影响整体清理
                fmt.Printf("storage: failed to remove orphan chunk dir %s: %v\n", uploadID, err)
                continue
            }
            removed = append(removed, uploadID)
        }
    }

    return removed, nil
}

// 文件: internal/data/storage/minio.go

// CleanupOrphanChunks 扫描 chunks/ 前缀下超过 maxAge 的分片对象并删除。
func (s *minioStorage) CleanupOrphanChunks(ctx context.Context, maxAge time.Duration) ([]string, error) {
    // 列举所有 chunks/ 下的对象
    objectCh := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{
        Prefix:    "chunks/",
        Recursive: true, // 递归列举所有层级
    })

    now := time.Now()
    uploadIDs := make(map[string]bool)           // 用 map 去重 uploadID
    var objectsToDelete []minio.ObjectInfo       // 收集待删除对象

    for obj := range objectCh {
        if ctx.Err() != nil {
            return nil, ctx.Err() // 支持取消
        }
        if obj.Err != nil {
            continue // 列举错误跳过单个对象
        }
        // LastModified 是对象在 MinIO 上的最后修改时间
        if now.Sub(obj.LastModified) > maxAge {
            objectsToDelete = append(objectsToDelete, obj)
            // 从对象路径提取 uploadID: chunks/{uploadID}/part_xxx
            // SplitN 切 3 段：["chunks", "{uploadID}", "part_xxx"]
            parts := strings.SplitN(obj.Key, "/", 3)
            if len(parts) >= 2 {
                uploadIDs[parts[1]] = true
            }
        }
    }

    // 批量删除过期对象
    // RemoveObjects 接收一个 channel，内部并发删除，效率远高于循环 RemoveObject
    if len(objectsToDelete) > 0 {
        delCh := make(chan minio.ObjectInfo, len(objectsToDelete))
        for _, obj := range objectsToDelete {
            delCh <- obj
        }
        close(delCh) // 必须关闭，RemoveObjects 才能知道输入结束

        for err := range s.client.RemoveObjects(ctx, s.bucket, delCh, minio.RemoveObjectsOptions{}) {
            if err.Err != nil {
                slog.Warn("清理 MinIO 孤儿分片失败", "err", err.Err)
            }
        }
    }

    // 把 map 的 key 转为 slice 返回
    result := make([]string, 0, len(uploadIDs))
    for id := range uploadIDs {
        result = append(result, id)
    }
    return result, nil
}

// 文件: internal/data/storage/local.go

// CompleteMultipartUpload 将所有已上传的分片合并为最终文件。
func (s *localStorage) CompleteMultipartUpload(ctx context.Context, uploadID string, parts []biz.PartInfo) (string, error) {
    chunkDir := filepath.Join(s.uploadDir, "chunks", uploadID)
    dir := s.datePath()
    if err := os.MkdirAll(dir, 0755); err != nil {
        return "", fmt.Errorf("storage: failed to create final dir: %w", err)
    }

    // 最终文件名用 uploadID + .merged 后缀，避免与普通上传文件混淆
    finalPath := filepath.Join(dir, uploadID+".merged")

    // 按分片编号排序，确保合并后字节顺序正确
    sort.Slice(parts, func(i, j int) bool {
        return parts[i].PartNumber < parts[j].PartNumber
    })

    dest, err := os.Create(finalPath)
    if err != nil {
        return "", fmt.Errorf("storage: failed to create final file: %w", err)
    }
    defer dest.Close()

    // 使用带缓冲的 Writer 减少 write 系统调用次数
    // 缓冲区 1MB，对于大文件合并可显著减少系统调用
    bufWriter := bufio.NewWriterSize(dest, 1024*1024)
    defer bufWriter.Flush()

    // 复用 io.Copy 的缓冲区，避免每次复制都分配新内存
    // 1MB 缓冲区在内存占用与拷贝效率之间取得平衡
    copyBuf := make([]byte, 1024*1024)

    // 按顺序合并分片
    for _, part := range parts {
        partPath := filepath.Join(chunkDir, fmt.Sprintf("part_%05d", part.PartNumber))
        src, err := os.Open(partPath)
        if err != nil {
            return "", fmt.Errorf("storage: failed to open part %d: %w", part.PartNumber, err)
        }
        // io.CopyBuffer 使用调用方提供的缓冲区，避免内部分配
        if _, err := io.CopyBuffer(bufWriter, src, copyBuf); err != nil {
            src.Close()
            return "", fmt.Errorf("storage: failed to merge part %d: %w", part.PartNumber, err)
        }
        src.Close()
    }

    // 刷新缓冲区并确保数据落盘
    // Flush 把 bufio 缓冲区的数据写到 os.File；Sync 调用 fsync 强制刷盘
    if err := bufWriter.Flush(); err != nil {
        return "", fmt.Errorf("storage: failed to flush merged file: %w", err)
    }
    if err := dest.Sync(); err != nil {
        return "", fmt.Errorf("storage: failed to sync merged file: %w", err)
    }

    // 清理分片目录 — 合并成功后分片已无用处
    os.RemoveAll(chunkDir)

    return finalPath, nil
}

// 文件: internal/data/model/user.go
type User struct {
    // ...
    StorageQuota int64 `gorm:"column:storage_quota;default:10737418240;not null"` // bytes, default 10GB
    TotalStorage int64 `gorm:"column:total_storage;default:10737418240;not null"` // bytes, default 10GB
    UsedStorage  int64 `gorm:"column:used_storage;default:0;not null"`
    // ...
}

// 文件: internal/data/user.go（教学版，含缺陷）
func (r *userRepo) UpdateUsedStorageAtomic(ctx context.Context, userID uint64, delta int64) error {
    if delta < 0 {
        // 减少：带 WHERE used_storage >= ? 防超卖 —— 正确
        result := r.db.WithContext(ctx).Model(&model.User{}).
            Where("id = ? AND used_storage >= ?", userID, -delta).
            UpdateColumn("used_storage", gorm.Expr("used_storage + ?", delta))
        if result.Error != nil {
            return result.Error
        }
        if result.RowsAffected == 0 {
            var count int64
            r.db.WithContext(ctx).Model(&model.User{}).Where("id = ?", userID).Count(&count)
            if count == 0 {
                return biz.ErrUserNotFound
            }
            return biz.ErrStorageInsufficient
        }
        return nil
    }

    // 增加（上传）：仅 WHERE id = ?，无上界 —— 缺陷！
    result := r.db.WithContext(ctx).Model(&model.User{}).
        Where("id = ?", userID).
        UpdateColumn("used_storage", gorm.Expr("used_storage + ?", delta))
    if result.Error != nil {
        return result.Error
    }
    if result.RowsAffected == 0 {
        return biz.ErrUserNotFound // 仅判断用户不存在，漏了"配额不足"
    }
    return nil
}

// 文件: internal/data/user.go（商用修正版）
func (r *userRepo) UpdateUsedStorageAtomic(ctx context.Context, userID uint64, delta int64) error {
    if delta < 0 {
        // 减少：used_storage >= -delta 防扣成负数（超卖反向）
        result := r.db.WithContext(ctx).Model(&model.User{}).
            Where("id = ? AND used_storage >= ?", userID, -delta).
            UpdateColumn("used_storage", gorm.Expr("used_storage + ?", delta))
        if result.Error != nil {
            return result.Error
        }
        if result.RowsAffected == 0 {
            return r.classifyZero(ctx, userID) // 区分"用户不存在"与"空间不足"
        }
        return nil
    }

    // 修正点：增加也必须带 DB 上界，防止并发超卖
    // 条件等价于 used_storage + delta <= total_storage
    result := r.db.WithContext(ctx).Model(&model.User{}).
        Where("id = ? AND used_storage <= total_storage - ?", userID, delta).
        UpdateColumn("used_storage", gorm.Expr("used_storage + ?", delta))
    if result.Error != nil {
        return result.Error
    }
    if result.RowsAffected == 0 {
        return r.classifyZero(ctx, userID) // 同样要区分两种 0 行情况
    }
    return nil
}

// classifyZero 在 RowsAffected==0 时区分"用户不存在"与"配额不足"
func (r *userRepo) classifyZero(ctx context.Context, userID uint64) error {
    var count int64
    r.db.WithContext(ctx).Model(&model.User{}).Where("id = ?", userID).Count(&count)
    if count == 0 {
        return biz.ErrUserNotFound
    }
    return biz.ErrStorageInsufficient
}

// 文件: internal/biz/storage.go
func (uc *UserUsecase) CheckStorageAvailable(ctx context.Context, userID uint64, fileSize int64) error {
    info, err := uc.GetStorageInfo(ctx, userID) // 读缓存（可能陈旧）
    if err != nil {
        return err
    }
    if info.Available < fileSize {
        return ErrStorageInsufficient // 413，前端提示"空间不足"
    }
    return nil
}

// 文件: internal/biz/file.go（Upload 关键片段）
func (uc *FileUsecase) Upload(ctx context.Context, userID uint64, name, fileType string, parentID *uint64, data []byte) (*File, error) {
    fileSize := int64(len(data))

    // 1) 上传前预检配额（基于缓存余量，UX 层快速拒绝）
    if uc.userUC != nil {
        if err := uc.userUC.CheckStorageAvailable(ctx, userID, fileSize); err != nil {
            return nil, err
        }
    }

    // 2) 写对象存储
    storagePath, err := uc.storage.Upload(ctx, fileName, bytesToReader(data))
    if err != nil {
        return nil, err
    }

    // 3) 写文件元数据
    created, err := uc.fileRepo.Create(ctx, file)
    if err != nil {
        uc.storage.Delete(ctx, storagePath) // 回滚孤儿物理文件
        return nil, err
    }

    // 4) 落库成功后再真实记账
    if uc.userUC != nil {
        if err := uc.userUC.AddUsedStorage(ctx, userID, fileSize); err != nil {
            // 修正点：AddUsedStorage 失败应回滚（删文件 + 删元数据），否则产生"文件在但没扣额度"的漂移
            log.Error("file: failed to add used storage after upload", "userID", userID, "err", err)
        }
    }
    return created, nil
}

// 文件: internal/biz/user.go（教学版）
const (
    cacheKeyUserStorage = "user:storage:%d"   // 修正点：缺命名空间前缀
    cacheTTLUserStorage = 300 * time.Second    // 固定 5 分钟，有雪崩风险
)

func (uc *UserUsecase) GetStorageInfo(ctx context.Context, userID uint64) (*StorageInfo, error) {
    if uc.cache != nil {
        if val, err := uc.cache.Get(ctx, fmt.Sprintf(cacheKeyUserStorage, userID)); err == nil && val != "" {
            var info StorageInfo
            if json.Unmarshal([]byte(val), &info) == nil {
                return &info, nil // 命中直接返回
            }
        }
    }
    // 未命中查库
    total, used, err := uc.repo.GetUserStorage(ctx, userID)
    if err != nil {
        return nil, err
    }
    available := total - used
    if available < 0 {
        available = 0
    }
    var percent float64
    if total > 0 {
        percent = float64(used) / float64(total) * 100
    }
    info := &StorageInfo{Total: total, Used: used, Available: available, Percent: percent}
    if uc.cache != nil {
        if b, err := json.Marshal(info); err == nil {
            _ = uc.cache.Set(ctx, fmt.Sprintf(cacheKeyUserStorage, userID), string(b), cacheTTLUserStorage)
            // 修正点：set 失败被忽略；且未缓存"空值"
        }
    }
    return info, nil
}

// 文件: internal/biz/storage.go
func (uc *UserUsecase) updateStorageWithLock(ctx context.Context, userID uint64, delta int64) error {
    if delta == 0 {
        return nil
    }
    lockKey := fmt.Sprintf(storageLockKey, userID)
    if uc.locker != nil {
        if err := uc.locker.Lock(ctx, lockKey); err != nil {
            return err
        }
        defer func() { _ = uc.locker.Unlock(ctx, lockKey) }()
    }
    if err := uc.repo.UpdateUsedStorageAtomic(ctx, userID, delta); err != nil {
        return err
    }
    // 记账成功后失效缓存，下次读取回源得到新值
    if uc.cache != nil {
        _ = uc.cache.Delete(ctx, fmt.Sprintf(cacheKeyUserStorage, userID))
    }
    return nil
}

// 文件: internal/biz/storage.go
const (
    storageLockKey   = "user:storage🔒%d"
    calibrateLockKey = "storage:calibrate:lock"
)

// 文件: internal/data/user.go
func (r *userRepo) GetActiveFileTotalSize(ctx context.Context, userID uint64) (int64, error) {
    var total int64
    err := r.db.WithContext(ctx).Model(&model.File{}).
        Where("user_id = ? AND status = 0", userID).
        Select("COALESCE(SUM(size), 0)").Scan(&total).Error
    return total, err
}

// 文件: internal/biz/storage.go
func (uc *UserUsecase) RecalibrateStorage(ctx context.Context, userID uint64) error {
    if uc.locker != nil {
        uc.locker.Lock(ctx, calibrateLockKey)
        defer func() { _ = uc.locker.Unlock(ctx, calibrateLockKey) }()
    }
    actualUsed, err := uc.repo.GetActiveFileTotalSize(ctx, userID)
    if err != nil {
        return err
    }
    _, used, err := uc.repo.GetUserStorage(ctx, userID)
    if err != nil {
        return err
    }
    delta := actualUsed - used
    if delta != 0 {
        uc.repo.UpdateUsedStorageAtomic(ctx, userID, delta) // 用原子更新纠正，避免覆盖并发改动
    }
    if uc.cache != nil {
        _ = uc.cache.Delete(ctx, fmt.Sprintf(cacheKeyUserStorage, userID))
    }
    return nil
}