SKYMirror
84df9996cb
CI 测试 / go vet + go test (push) Successful in 49s
- 全项目版本号统一：v3.x 迭代号（26.9/26.9/26.9/26.9 及裸 v2/v3）→ 26.9， 覆盖 Go 注释 / 文档 / openapi.yaml / README×4 / 前端源码（80+ 处） - v31_test.go 更名 custom_code_test.go；TestV2AccessorDefaults → TestKVAccessorDefaults - docs/api/00-overview.md 更新日志合并为单条 26.9 条目（修复错位拼接） - .goreleaser.yaml 头部注释与实际一致（Pro 2.18.1 / GITEA_TOKEN / semver tag 要求） - CI：release-image.yml → ci.yml，仅保留 vet+test 门禁； 镜像发布移交 GoReleaser Pro（原 build-push 的 tag 校验与 26.9 版本方案冲突，历史 9 次失败） - 前端重建：server/web/dist 与 web-embed 同步（docs 文案嵌入更新）
668 lines
22 KiB
Go
668 lines
22 KiB
Go
package api
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"mime/multipart"
|
||
"net/http"
|
||
"strconv"
|
||
"time"
|
||
|
||
"github.com/gin-gonic/gin"
|
||
"gorm.io/gorm"
|
||
|
||
"fileshare/internal/middleware"
|
||
"fileshare/internal/model"
|
||
"fileshare/internal/response"
|
||
"fileshare/internal/storage"
|
||
)
|
||
|
||
// chunkExpireTTL 分片会话保留时长（M5：预留窗口由 24h 缩短为 2h；
|
||
// 会话本身保留 24h 支持断点续传，见 janitor 的清理周期）。
|
||
const chunkExpireTTL = 2 * time.Hour
|
||
|
||
// maxChunkSizeBytes 单分片大小上限 32MB（M3：限制 io.ReadAll 内存占用）。
|
||
const maxChunkSizeBytes = 32 * 1024 * 1024
|
||
|
||
// ============ POST /chunk/upload/init 初始化分片会话 ============
|
||
|
||
// requireChunkEnabled L4：enableChunk 开关后端强制（此前仅前端隐藏入口，
|
||
// 开关关闭后 /chunk/* 接口仍可直接调用）。
|
||
func (d *Deps) requireChunkEnabled(c *gin.Context) bool {
|
||
if d.Cfg.EnableChunk() {
|
||
return true
|
||
}
|
||
auditRecordFailed(c, d.AuditSvc, "分片上传未启用")
|
||
response.Fail(c, http.StatusForbidden, "分片上传未启用")
|
||
return false
|
||
}
|
||
|
||
// chunkInitRequest init 请求体（JSON 或表单）。
|
||
type chunkInitRequest struct {
|
||
FileName string `json:"file_name" form:"file_name"`
|
||
ChunkSize int64 `json:"chunk_size" form:"chunk_size"`
|
||
FileSize int64 `json:"file_size" form:"file_size"`
|
||
FileHash string `json:"file_hash" form:"file_hash"`
|
||
}
|
||
|
||
// chunkInit 创建分片上传会话（对齐参考 init_chunk_upload）：
|
||
// 支持断点续传（相同 hash/大小/文件名的未完成会话直接续传）。
|
||
func (d *Deps) chunkInit(c *gin.Context) {
|
||
if !d.requireChunkEnabled(c) {
|
||
return
|
||
}
|
||
if !d.requireShareLogin(c) {
|
||
return
|
||
}
|
||
if !requireUploadLimit(c, d.Limiter) {
|
||
return
|
||
}
|
||
var req chunkInitRequest
|
||
if err := bindJSONOrForm(c, &req); err != nil {
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
safeName := storage.SanitizeFileName(req.FileName)
|
||
if safeName == "" {
|
||
auditRecordFailed(c, d.AuditSvc, "文件名非法")
|
||
response.Fail(c, http.StatusBadRequest, "文件名非法")
|
||
return
|
||
}
|
||
// 文件类型白名单（无内容可校验，仅名称）
|
||
if err := validateFileMagic(d.Cfg, safeName, "", nil); err != nil {
|
||
auditUploadEntry(c, "", safeName, req.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "文件类型被拒绝")
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
chunkSize := req.ChunkSize
|
||
if chunkSize <= 0 {
|
||
chunkSize = 5 * 1024 * 1024 // 默认 5MB（对齐参考 InitChunkUploadModel）
|
||
}
|
||
// M3：单片全部读入内存后再落存储，必须限制单片大小（客户端声明的
|
||
// chunk_size 上界受策略约束，但策略允许至 10GiB → 显式封顶 32MB）。
|
||
if chunkSize > maxChunkSizeBytes {
|
||
auditRecordFailed(c, d.AuditSvc, "chunk_size 超过上限")
|
||
response.Fail(c, http.StatusBadRequest, fmt.Sprintf("chunk_size 过大，最大为 %d MB", maxChunkSizeBytes>>20))
|
||
return
|
||
}
|
||
if req.FileSize <= 0 {
|
||
auditRecordFailed(c, d.AuditSvc, "file_size 非法")
|
||
response.Fail(c, http.StatusBadRequest, "file_size 必须大于 0")
|
||
return
|
||
}
|
||
// 服务端按分片数上限校验总大小（防分片声明绕过）
|
||
totalChunks := (req.FileSize + chunkSize - 1) / chunkSize
|
||
maxPossible := totalChunks * chunkSize
|
||
// 26.9 需求 ④⑩：动态策略校验（max_file_size，0=回落 uploadSize）
|
||
if err := d.CurrentUploadPolicy().CheckSize(maxPossible); err != nil {
|
||
auditUploadEntry(c, "", safeName, req.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "文件大小超过限制")
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
|
||
ctx := c.Request.Context()
|
||
// 断点续传：查找相同 hash+大小+文件名的未完成会话（chunk_index=-1 为会话头）
|
||
var existing model.UploadChunk
|
||
err := d.DB.WithContext(ctx).
|
||
Where("chunk_hash = ? AND chunk_index = -1 AND file_size = ? AND file_name = ?",
|
||
req.FileHash, req.FileSize, safeName).
|
||
First(&existing).Error
|
||
if err == nil {
|
||
if existing.SavePath == "" {
|
||
// 脏会话：清理后按新建处理
|
||
_ = d.DB.WithContext(ctx).
|
||
Where("upload_id = ?", existing.UploadID).
|
||
Delete(&model.UploadChunk{}).Error
|
||
releaseStorage(ctx, d.DB, "chunk:"+existing.UploadID)
|
||
} else {
|
||
if err := reserveStorage(ctx, d.DB, d.Cfg, "chunk:"+existing.UploadID, existing.FileSize, chunkExpireTTL); err != nil {
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
uploaded := d.uploadedChunkIndexes(ctx, existing.UploadID)
|
||
auditUploadEntry(c, existing.UploadID, safeName, req.FileSize, 0)
|
||
auditRecordSuccess(c, d.AuditSvc)
|
||
response.OK(c, gin.H{
|
||
"existed": false,
|
||
"upload_id": existing.UploadID,
|
||
"chunk_size": existing.ChunkSize,
|
||
"total_chunks": existing.TotalChunks,
|
||
"uploaded_chunks": uploaded,
|
||
})
|
||
return
|
||
}
|
||
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
||
respondError(c, errInternal("查询上传会话失败: "+err.Error()))
|
||
return
|
||
}
|
||
|
||
// 新建会话
|
||
uploadID := uuidHex()
|
||
resToken := "chunk:" + uploadID
|
||
if err := reserveStorage(ctx, d.DB, d.Cfg, resToken, req.FileSize, chunkExpireTTL); err != nil {
|
||
auditUploadEntry(c, "", safeName, req.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "容量预留失败")
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
// M5：init 即计入上传限流（此前仅 complete 成功时计数，
|
||
// 恶意客户端可无限创建会话占用容量预留）
|
||
d.Limiter.Add(c, middleware.LimitUpload)
|
||
_, _, _, _, savePath := buildSavePath(d.Cfg, safeName, uploadID)
|
||
session := model.UploadChunk{
|
||
UploadID: uploadID,
|
||
ChunkIndex: -1,
|
||
TotalChunks: int(totalChunks),
|
||
FileSize: req.FileSize,
|
||
ChunkSize: int(chunkSize),
|
||
ChunkHash: req.FileHash,
|
||
FileName: safeName,
|
||
SavePath: savePath,
|
||
Engine: d.Store.CurrentName(), // 26.9：会话归属引擎（分片/合并全程走同一引擎）
|
||
}
|
||
if err := d.DB.WithContext(ctx).Create(&session).Error; err != nil {
|
||
releaseStorage(ctx, d.DB, resToken)
|
||
auditUploadEntry(c, "", safeName, req.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "会话创建失败")
|
||
respondError(c, errInternal("创建上传会话失败: "+err.Error()))
|
||
return
|
||
}
|
||
auditUploadEntry(c, uploadID, safeName, req.FileSize, 0)
|
||
auditRecordSuccess(c, d.AuditSvc)
|
||
response.OK(c, gin.H{
|
||
"existed": false,
|
||
"upload_id": uploadID,
|
||
"chunk_size": chunkSize,
|
||
"total_chunks": totalChunks,
|
||
"uploaded_chunks": []int{},
|
||
})
|
||
}
|
||
|
||
// uploadedChunkIndexes 查询会话中已完成分片的索引列表。
|
||
func (d *Deps) uploadedChunkIndexes(ctx context.Context, uploadID string) []int {
|
||
var rows []model.UploadChunk
|
||
if err := d.DB.WithContext(ctx).
|
||
Where("upload_id = ? AND completed = ?", uploadID, true).
|
||
Order("chunk_index ASC").Find(&rows).Error; err != nil {
|
||
return []int{}
|
||
}
|
||
out := make([]int, 0, len(rows))
|
||
for _, r := range rows {
|
||
out = append(out, r.ChunkIndex)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// ============ POST /chunk/upload/{uploadID}/{index}（及扁平兼容）============
|
||
|
||
// chunkUploadFlat 扁平模式：POST /chunk/upload，upload_id/chunk_index 走表单或 query。
|
||
// 多文件字段（chunk/chunks）时按 base_chunk_index 顺序批量接收。
|
||
func (d *Deps) chunkUploadFlat(c *gin.Context) {
|
||
c.Params = append(c.Params, gin.Param{Key: "uploadID", Value: resolveUploadID(c)})
|
||
c.Params = append(c.Params, gin.Param{Key: "chunkIndex", Value: resolveChunkIndex(c)})
|
||
d.chunkUpload(c)
|
||
}
|
||
|
||
// resolveUploadID 解析 upload_id：路径参数 → multipart 表单 → query。
|
||
func resolveUploadID(c *gin.Context) string {
|
||
if v := c.Param("uploadID"); v != "" {
|
||
return v
|
||
}
|
||
if v := c.PostForm("upload_id"); v != "" {
|
||
return v
|
||
}
|
||
return c.Query("upload_id")
|
||
}
|
||
|
||
// resolveChunkIndex 解析 chunk_index：路径参数 → multipart 表单 → query。
|
||
func resolveChunkIndex(c *gin.Context) string {
|
||
if v := c.Param("chunkIndex"); v != "" {
|
||
return v
|
||
}
|
||
if v := c.PostForm("chunk_index"); v != "" {
|
||
return v
|
||
}
|
||
return c.Query("chunk_index")
|
||
}
|
||
|
||
// chunkUpload 上传单个（或批量）分片（对齐参考 upload_chunk）。
|
||
// multipart 文件字段：chunk（主）或 file（回退）；批量用 chunk[]/chunks 数组 + chunk_index 为起始索引。
|
||
func (d *Deps) chunkUpload(c *gin.Context) {
|
||
if !d.requireChunkEnabled(c) {
|
||
return
|
||
}
|
||
if !d.requireShareLogin(c) {
|
||
return
|
||
}
|
||
uploadID := resolveUploadID(c)
|
||
ctx := c.Request.Context()
|
||
|
||
var session model.UploadChunk
|
||
if err := d.DB.WithContext(ctx).
|
||
Where("upload_id = ? AND chunk_index = -1", uploadID).
|
||
First(&session).Error; err != nil {
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
auditUploadEntry(c, uploadID, "", 0, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "上传会话不存在")
|
||
response.Fail(c, http.StatusNotFound, "上传会话不存在")
|
||
return
|
||
}
|
||
respondError(c, errInternal("查询上传会话失败: "+err.Error()))
|
||
return
|
||
}
|
||
if err := reserveStorage(ctx, d.DB, d.Cfg, "chunk:"+uploadID, session.FileSize, chunkExpireTTL); err != nil {
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
|
||
// 收集分片文件：chunk（单）→ file（回退）→ chunk[]/chunks（批量）
|
||
form, err := c.MultipartForm()
|
||
if err != nil {
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "multipart 解析失败")
|
||
response.Fail(c, http.StatusBadRequest, "multipart 表单解析失败")
|
||
return
|
||
}
|
||
files := form.File["chunk"]
|
||
single := len(files) == 0
|
||
if single {
|
||
files = form.File["file"]
|
||
}
|
||
if len(files) == 0 {
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "缺少 chunk 分片字段")
|
||
response.Fail(c, http.StatusBadRequest, "缺少分片文件字段 chunk")
|
||
return
|
||
}
|
||
|
||
baseIndex, err := strconv.Atoi(resolveChunkIndex(c))
|
||
if err != nil {
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "无效的分片索引")
|
||
response.Fail(c, http.StatusBadRequest, "无效的分片索引")
|
||
return
|
||
}
|
||
|
||
results := make([]gin.H, 0, len(files))
|
||
for i, fh := range files {
|
||
// 单分片模式严格使用请求索引；批量模式从 base 递增
|
||
idx := baseIndex
|
||
if !single && len(files) > 1 {
|
||
idx = baseIndex + i
|
||
}
|
||
res, status, msg := d.saveOneChunk(c, ctx, &session, idx, fh)
|
||
if status != 0 {
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, msg)
|
||
response.Fail(c, status, msg)
|
||
return
|
||
}
|
||
results = append(results, res)
|
||
}
|
||
// 审计：传输字节数为本次请求分片总和
|
||
var transferred int64
|
||
for _, fh := range files {
|
||
transferred += fh.Size
|
||
}
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, transferred)
|
||
auditRecordSuccess(c, d.AuditSvc)
|
||
if len(results) == 1 {
|
||
response.OK(c, results[0])
|
||
return
|
||
}
|
||
response.OK(c, gin.H{"chunks": results})
|
||
}
|
||
|
||
// saveOneChunk 保存一个分片：查重→读数据→校验→存储→记录。
|
||
// 返回 (响应体, HTTP错误状态码, 错误信息)；成功时状态码为 0。
|
||
func (d *Deps) saveOneChunk(c *gin.Context, ctx context.Context, session *model.UploadChunk, idx int, fh *multipart.FileHeader) (gin.H, int, string) {
|
||
if idx < 0 || idx >= session.TotalChunks {
|
||
return nil, http.StatusBadRequest, "无效的分片索引"
|
||
}
|
||
// 已上传分片：断点续传直接跳过
|
||
var existing model.UploadChunk
|
||
err := d.DB.WithContext(ctx).
|
||
Where("upload_id = ? AND chunk_index = ? AND completed = ?", session.UploadID, idx, true).
|
||
First(&existing).Error
|
||
if err == nil {
|
||
return gin.H{"chunk_hash": existing.ChunkHash, "skipped": true, "chunk_index": idx}, 0, ""
|
||
}
|
||
if !errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return nil, http.StatusInternalServerError, "查询分片记录失败"
|
||
}
|
||
|
||
f, err := fh.Open()
|
||
if err != nil {
|
||
return nil, http.StatusBadRequest, "分片数据读取失败"
|
||
}
|
||
defer func() { _ = f.Close() }()
|
||
data, err := io.ReadAll(io.LimitReader(f, int64(session.ChunkSize)+1))
|
||
if err != nil {
|
||
return nil, http.StatusBadRequest, "分片数据读取失败"
|
||
}
|
||
// 校验分片大小不超过声明值
|
||
if int64(len(data)) > int64(session.ChunkSize) {
|
||
return nil, http.StatusBadRequest,
|
||
"分片大小超过声明值: 最大 " + strconv.Itoa(session.ChunkSize) + ", 实际 " + strconv.Itoa(len(data))
|
||
}
|
||
// 累计大小校验（已传分片数×chunk_size + 当前分片；动态策略上限）
|
||
var uploadedCount int64
|
||
_ = d.DB.WithContext(ctx).Model(&model.UploadChunk{}).
|
||
Where("upload_id = ? AND completed = ?", session.UploadID, true).
|
||
Count(&uploadedCount).Error
|
||
if err := d.CurrentUploadPolicy().CheckSize(uploadedCount*int64(session.ChunkSize) + int64(len(data))); err != nil {
|
||
return nil, http.StatusForbidden, err.Error()
|
||
}
|
||
// 首分片做 magic bytes 防伪造
|
||
if idx == 0 {
|
||
head := data
|
||
if len(head) > 64 {
|
||
head = head[:64]
|
||
}
|
||
if err := validateFileMagic(d.Cfg, session.FileName, "", head); err != nil {
|
||
return nil, http.StatusForbidden, "文件内容校验失败：" + err.Error()
|
||
}
|
||
}
|
||
|
||
sum := sha256.Sum256(data)
|
||
chunkHash := hex.EncodeToString(sum[:])
|
||
if _, err := d.Store.SaveChunk(ctx, session.UploadID, idx, bytes.NewReader(data), session.SavePath); err != nil {
|
||
return nil, http.StatusInternalServerError, "分片保存失败: " + err.Error()
|
||
}
|
||
// 保存成功后再记录（对齐参考：先存储后落库）。
|
||
// 注意：不能用结构体 Where 条件（GORM 会忽略零值字段，chunk_index=0 会被
|
||
// 丢弃从而误匹配 -1 会话行），必须用字符串条件 + 完整目标结构体。
|
||
rec := model.UploadChunk{
|
||
UploadID: session.UploadID,
|
||
ChunkIndex: idx,
|
||
ChunkHash: chunkHash,
|
||
Completed: true,
|
||
FileSize: session.FileSize,
|
||
TotalChunks: session.TotalChunks,
|
||
ChunkSize: session.ChunkSize,
|
||
FileName: session.FileName,
|
||
SavePath: session.SavePath,
|
||
Engine: session.Engine, // 26.9：继承会话引擎
|
||
}
|
||
if err := d.DB.WithContext(ctx).
|
||
Where("upload_id = ? AND chunk_index = ?", session.UploadID, idx).
|
||
FirstOrCreate(&rec).Error; err != nil {
|
||
return nil, http.StatusInternalServerError, "分片记录写入失败"
|
||
}
|
||
return gin.H{"chunk_hash": chunkHash, "chunk_index": idx}, 0, ""
|
||
}
|
||
|
||
// ============ GET /chunk/upload/status/{uploadID} ============
|
||
|
||
// chunkStatus 查询上传进度（对齐参考 get_upload_status）。
|
||
func (d *Deps) chunkStatus(c *gin.Context) {
|
||
if !d.requireShareLogin(c) {
|
||
return
|
||
}
|
||
uploadID := c.Param("uploadID")
|
||
if uploadID == "" {
|
||
uploadID = c.Query("upload_id")
|
||
}
|
||
ctx := c.Request.Context()
|
||
var session model.UploadChunk
|
||
if err := d.DB.WithContext(ctx).
|
||
Where("upload_id = ? AND chunk_index = -1", uploadID).
|
||
First(&session).Error; err != nil {
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
response.Fail(c, http.StatusNotFound, "上传会话不存在")
|
||
return
|
||
}
|
||
respondError(c, errInternal("查询上传会话失败: "+err.Error()))
|
||
return
|
||
}
|
||
uploaded := d.uploadedChunkIndexes(ctx, uploadID)
|
||
var progress float64
|
||
if session.TotalChunks > 0 {
|
||
progress = float64(len(uploaded)) / float64(session.TotalChunks) * 100
|
||
}
|
||
response.OK(c, gin.H{
|
||
"upload_id": uploadID,
|
||
"file_name": session.FileName,
|
||
"file_size": session.FileSize,
|
||
"chunk_size": session.ChunkSize,
|
||
"total_chunks": session.TotalChunks,
|
||
"uploaded_chunks": uploaded,
|
||
"progress": progress,
|
||
})
|
||
}
|
||
|
||
// ============ POST /chunk/upload/complete/{uploadID} ============
|
||
|
||
// chunkCompleteRequest complete 请求体。
|
||
type chunkCompleteRequest struct {
|
||
ExpireValue int `json:"expire_value" form:"expire_value"`
|
||
ExpireStyle string `json:"expire_style" form:"expire_style"`
|
||
Code string `json:"code" form:"code"` // 26.9：自定义提取码（4-8 位字母数字，空=随机）
|
||
}
|
||
|
||
// chunkComplete 合并分片并创建分享（对齐参考 complete_upload）。
|
||
func (d *Deps) chunkComplete(c *gin.Context) {
|
||
if !d.requireChunkEnabled(c) {
|
||
return
|
||
}
|
||
if !d.requireShareLogin(c) {
|
||
return
|
||
}
|
||
if !requireUploadLimit(c, d.Limiter) {
|
||
return
|
||
}
|
||
uploadID := c.Param("uploadID")
|
||
if uploadID == "" {
|
||
uploadID = resolveUploadID(c)
|
||
}
|
||
ctx := c.Request.Context()
|
||
|
||
var session model.UploadChunk
|
||
if err := d.DB.WithContext(ctx).
|
||
Where("upload_id = ? AND chunk_index = -1", uploadID).
|
||
First(&session).Error; err != nil {
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
auditUploadEntry(c, uploadID, "", 0, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "上传会话不存在")
|
||
response.Fail(c, http.StatusNotFound, "上传会话不存在")
|
||
return
|
||
}
|
||
respondError(c, errInternal("查询上传会话失败: "+err.Error()))
|
||
return
|
||
}
|
||
var req chunkCompleteRequest
|
||
if err := bindJSONOrForm(c, &req); err != nil {
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
exp, err := resolveExpire(d.Cfg, req.ExpireValue, req.ExpireStyle)
|
||
if err != nil {
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "过期策略非法")
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
// 26.9：自定义提取码（合并前校验，失败快速返回）
|
||
if err := validatePickupCode(req.Code); err != nil {
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "提取码非法")
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
if err := reserveStorage(ctx, d.DB, d.Cfg, "chunk:"+uploadID, session.FileSize, chunkExpireTTL); err != nil {
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
|
||
// 分片完整性校验（chunk_index >= 0；-1 为会话头，completed 恒为 false）
|
||
var completed []model.UploadChunk
|
||
if err := d.DB.WithContext(ctx).
|
||
Where("upload_id = ? AND completed = ? AND chunk_index >= 0", uploadID, true).
|
||
Find(&completed).Error; err != nil {
|
||
respondError(c, errInternal("查询分片记录失败: "+err.Error()))
|
||
return
|
||
}
|
||
if len(completed) != session.TotalChunks {
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "分片不完整")
|
||
response.Fail(c, http.StatusBadRequest, "分片不完整")
|
||
return
|
||
}
|
||
// 累计大小上限校验（超限清理会话，对齐参考；动态策略上限）
|
||
if err := d.CurrentUploadPolicy().CheckSize(int64(len(completed)) * int64(session.ChunkSize)); err != nil {
|
||
if cs, ce := d.storeFor(session.Engine); ce == nil {
|
||
_ = cs.CleanChunks(ctx, uploadID, session.SavePath)
|
||
}
|
||
_ = d.DB.WithContext(ctx).Where("upload_id = ?", uploadID).Delete(&model.UploadChunk{}).Error
|
||
releaseStorage(ctx, d.DB, "chunk:"+uploadID)
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, 0)
|
||
auditRecordFailed(c, d.AuditSvc, "实际上传大小超过限制")
|
||
respondError(c, err)
|
||
return
|
||
}
|
||
|
||
// 合并（引擎负责按索引有序合并+SHA256 校验）
|
||
verifyHash := func(index int) (string, error) {
|
||
var rec model.UploadChunk
|
||
err := d.DB.WithContext(ctx).
|
||
Where("upload_id = ? AND chunk_index = ?", uploadID, index).
|
||
First(&rec).Error
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
return rec.ChunkHash, nil
|
||
}
|
||
// 26.9：合并走会话归属引擎（会话创建时的引擎，即使中途热切换也不受影响）
|
||
mergeStore, sErr := d.storeFor(session.Engine)
|
||
if sErr != nil {
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, session.FileSize)
|
||
auditRecordFailed(c, d.AuditSvc, "存储引擎不可用: "+sErr.Error())
|
||
respondError(c, mapStorageError(sErr))
|
||
return
|
||
}
|
||
size, fileHash, err := mergeStore.MergeChunks(ctx, uploadID, session.TotalChunks, verifyHash, session.SavePath)
|
||
if err != nil {
|
||
_ = mergeStore.CleanChunks(ctx, uploadID, session.SavePath)
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, session.FileSize)
|
||
auditRecordFailed(c, d.AuditSvc, "文件合并失败")
|
||
respondError(c, mapStorageError(err))
|
||
return
|
||
}
|
||
|
||
// 创建分享记录（26.9：支持自定义提取码）
|
||
code, err := pickCustomCode(ctx, d.DB, d.Cfg, req.Code)
|
||
if err == nil {
|
||
fc := model.FileCodes{
|
||
Code: code,
|
||
FileHash: &fileHash,
|
||
IsChunked: true,
|
||
UploadID: &uploadID,
|
||
Size: session.FileSize,
|
||
ExpiredAt: exp.ExpiredAt,
|
||
ExpiredCount: exp.ExpiredCount,
|
||
UsedCount: exp.UsedCount,
|
||
Engine: session.Engine, // 26.9：归属引擎戳
|
||
}
|
||
// 拆分路径与文件名（对齐参考：path=dirname(save_path), uuid=basename）
|
||
dir, name := splitDirBase(session.SavePath)
|
||
ext := baseExt(name)
|
||
fc.FilePath = &dir
|
||
fc.UUIDFileName = &name
|
||
fc.Prefix = trimExt(name)
|
||
fc.Suffix = ext
|
||
err = d.DB.WithContext(ctx).Create(&fc).Error
|
||
err = mapCodeConflict(err) // 26.9
|
||
}
|
||
if err == nil {
|
||
// 成功：清理分片与记录（走归属引擎）
|
||
_ = mergeStore.CleanChunks(ctx, uploadID, session.SavePath)
|
||
_ = d.DB.WithContext(ctx).Where("upload_id = ?", uploadID).Delete(&model.UploadChunk{}).Error
|
||
}
|
||
releaseStorage(ctx, d.DB, "chunk:"+uploadID)
|
||
if err != nil {
|
||
auditUploadEntry(c, uploadID, session.FileName, session.FileSize, size)
|
||
auditRecordFailed(c, d.AuditSvc, "创建分享失败")
|
||
respondError(c, errInternal("创建分享失败: "+err.Error()))
|
||
return
|
||
}
|
||
|
||
d.Limiter.Add(c, middleware.LimitUpload)
|
||
auditUploadEntry(c, code, session.FileName, session.FileSize, size)
|
||
auditRecordSuccess(c, d.AuditSvc)
|
||
response.OK(c, gin.H{"code": code, "name": session.FileName})
|
||
}
|
||
|
||
// splitDirBase 拆分相对路径为目录与文件名。
|
||
func splitDirBase(p string) (dir, base string) {
|
||
for i := len(p) - 1; i >= 0; i-- {
|
||
if p[i] == '/' {
|
||
return p[:i], p[i+1:]
|
||
}
|
||
}
|
||
return "", p
|
||
}
|
||
|
||
// baseExt 提取扩展名（含点）。
|
||
func baseExt(name string) string {
|
||
for i := len(name) - 1; i >= 0; i-- {
|
||
if name[i] == '.' {
|
||
return name[i:]
|
||
}
|
||
if name[i] == '/' {
|
||
break
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// trimExt 去除扩展名。
|
||
func trimExt(name string) string {
|
||
ext := baseExt(name)
|
||
return name[:len(name)-len(ext)]
|
||
}
|
||
|
||
// ============ DELETE /chunk/upload/{uploadID} 取消上传 ============
|
||
|
||
// chunkCancel 取消上传并清理临时文件（对齐参考 cancel_upload）。
|
||
func (d *Deps) chunkCancel(c *gin.Context) {
|
||
if !d.requireShareLogin(c) {
|
||
return
|
||
}
|
||
uploadID := c.Param("uploadID")
|
||
if uploadID == "" {
|
||
uploadID = c.Query("upload_id")
|
||
}
|
||
ctx := c.Request.Context()
|
||
var session model.UploadChunk
|
||
if err := d.DB.WithContext(ctx).
|
||
Where("upload_id = ? AND chunk_index = -1", uploadID).
|
||
First(&session).Error; err != nil {
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
response.Fail(c, http.StatusNotFound, "上传会话不存在")
|
||
return
|
||
}
|
||
respondError(c, errInternal("查询上传会话失败: "+err.Error()))
|
||
return
|
||
}
|
||
if session.SavePath != "" {
|
||
if cs, ce := d.storeFor(session.Engine); ce == nil {
|
||
_ = cs.CleanChunks(ctx, uploadID, session.SavePath)
|
||
}
|
||
}
|
||
if err := d.DB.WithContext(ctx).
|
||
Where("upload_id = ?", uploadID).
|
||
Delete(&model.UploadChunk{}).Error; err != nil {
|
||
respondError(c, errInternal("取消上传失败: "+err.Error()))
|
||
return
|
||
}
|
||
releaseStorage(ctx, d.DB, "chunk:"+uploadID)
|
||
response.OK(c, gin.H{"message": "上传已取消"})
|
||
}
|