tt/ttstd-infinite-canvas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(storage): 资源上传幂等 - 合并 PR #371 并修复重试配额

Browse Source
This commit is contained in:
ddCat
2026-09-01 19:09:15 +08:00
parent 8ae169bcbf 2a33a20628
commit 56e6ecb836
14 changed files with 419 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
backend/internal/handler/user_data.go
View File

@@ -177,7 +177,7 @@ func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) {
width, _ := strconv.Atoi(c.PostForm("width"))
height, _ := strconv.Atoi(c.PostForm("height"))
durationMs, _ := strconv.ParseInt(c.PostForm("durationMs"), 10, 64)
resource, err := svc.UploadResource(user.ID, file, c.PostForm("kind"), width, height, durationMs)
resource, err := svc.UploadResource(user.ID, file, c.PostForm("kind"), width, height, durationMs, c.GetHeader("X-Idempotency-Key"))
if err != nil {
failService(c, err)
return
@@ -206,7 +206,7 @@ func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) {
fail(c, http.StatusBadRequest, err)
return
}
resource, err := svc.ImportResourceURL(user.ID, req.URL, req.Kind, req.Width, req.Height, req.DurationMs)
resource, err := svc.ImportResourceURL(user.ID, req.URL, req.Kind, req.Width, req.Height, req.DurationMs, c.GetHeader("X-Idempotency-Key"))
if err != nil {
failService(c, err)
return

28
backend/internal/model/models_project.go
View File

@@ -19,25 +19,27 @@ const AssetIDMaxLength = 80
type Resource struct {
ID string `json:"id" gorm:"primaryKey;size:36"`
UserID string `json:"userId" gorm:"index;size:36;index:idx_resources_user_created,priority:1"`
UserID string `json:"userId" gorm:"index;size:36;index:idx_resources_user_created,priority:1;uniqueIndex:idx_resources_user_upload_key,priority:1"`
Kind string `json:"kind" gorm:"index;size:24"`
Status ResourceStatus `json:"status" gorm:"index;size:24"`
Provider string `json:"provider" gorm:"size:24"`
Endpoint string `json:"endpoint"`
Bucket string `json:"bucket" gorm:"size:160"`
// 用户 OSS 每次修改都会生成新版本，资源固定引用创建时的存储与密钥；只有同一存储位置才可复用当前 CDN。
StorageSettingID string `json:"-" gorm:"index;size:36"`
ObjectKey string `json:"objectKey" gorm:"index"`
PublicURL string `json:"publicUrl"`
MimeType string `json:"mimeType" gorm:"size:120"`
Size int64 `json:"size"`
Width int `json:"width"`
Height int `json:"height"`
DurationMs int64 `json:"durationMs"`
ETag string `json:"etag" gorm:"size:160"`
Error string `json:"error"`
CreatedAt time.Time `json:"createdAt" gorm:"index:idx_resources_user_created,priority:2"`
UpdatedAt time.Time `json:"updatedAt"`
StorageSettingID string `json:"-" gorm:"index;size:36"`
ObjectKey string `json:"objectKey" gorm:"index"`
PublicURL string `json:"publicUrl"`
MimeType string `json:"mimeType" gorm:"size:120"`
Size int64 `json:"size"`
Width int `json:"width"`
Height int `json:"height"`
DurationMs int64 `json:"durationMs"`
ETag string `json:"etag" gorm:"size:160"`
// UploadKey 是客户端逻辑上传身份的摘要；NULL 表示不参与幂等约束。
UploadKey *string `json:"-" gorm:"size:64;uniqueIndex:idx_resources_user_upload_key,priority:2"`
Error string `json:"error"`
CreatedAt time.Time `json:"createdAt" gorm:"index:idx_resources_user_created,priority:2"`
UpdatedAt time.Time `json:"updatedAt"`
}
// ResourceDeletionJob is the durable handoff between database deletion and

15
backend/internal/repository/repository.go
View File

@@ -1009,6 +1009,21 @@ func (r *Repository) SaveResource(resource *model.Resource) error {
return r.db.Save(resource).Error
}
func (r *Repository) ResourceByUploadKey(userID string, uploadKey string) (*model.Resource, error) {
var resource model.Resource
if err := r.db.First(&resource, "user_id = ? AND upload_key = ?", userID, uploadKey).Error; err != nil {
return nil, err
}
return &resource, nil
}
func (r *Repository) ClaimFailedResourceUpload(userID string, id string) (bool, error) {
result := r.db.Model(&model.Resource{}).
Where("id = ? AND user_id = ? AND status = ?", id, userID, model.ResourceStatusFailed).
Updates(map[string]any{"status": model.ResourceStatusPending, "error": "", "updated_at": time.Now()})
return result.RowsAffected == 1, result.Error
}
func (r *Repository) DeleteResource(userID string, id string) error {
return r.db.Delete(&model.Resource{}, "id = ? AND user_id = ?", id, userID).Error
}

213
backend/internal/service/resource.go
View File

@@ -7,6 +7,7 @@ import (
"crypto/sha1"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
@@ -266,28 +267,43 @@ func validatePublicResourceBaseURL(raw string) (*url.URL, error) {
return parsed, nil
}
func (s *Service) UploadResource(userID string, header *multipart.FileHeader, kind string, width int, height int, durationMs int64) (*model.Resource, error) {
func (s *Service) UploadResource(userID string, header *multipart.FileHeader, kind string, width int, height int, durationMs int64, uploadIdentity ...string) (*model.Resource, error) {
if header == nil {
return nil, BadAuthRequest("请选择要上传的文件")
}
day, err := s.reserveUserUploadQuota(userID, header.Size)
uploadKey := normalizedResourceUploadKey(uploadIdentity)
existing, err := s.resourceForUploadKey(userID, uploadKey)
if err != nil {
return nil, err
}
if existing != nil && existing.Status == model.ResourceStatusReady {
return existing, nil
}
if existing != nil && existing.Status == model.ResourceStatusPending {
return nil, resourceUploadInProgress()
}
file, err := header.Open()
if err != nil {
s.releaseUserUploadQuota(userID, day, header.Size)
return nil, err
}
defer file.Close()
mimeType := strings.TrimSpace(header.Header.Get("Content-Type"))
mimeType = detectUploadedMimeType(file, header.Filename, mimeType)
resource, err := s.storeResource(userID, kind, header.Filename, mimeType, header.Size, width, height, durationMs, file)
if existing != nil {
return s.retryStoredResource(userID, existing, kind, mimeType, header.Size, file)
}
day, err := s.reserveUserUploadQuota(userID, header.Size)
if err != nil {
return nil, err
}
resource, stored, err := s.storeResource(userID, kind, header.Filename, mimeType, header.Size, width, height, durationMs, file, uploadKey)
if err != nil {
s.releaseUserUploadQuota(userID, day, header.Size)
} else {
} else if stored {
s.commitUserUploadQuota(userID, header.Size)
} else {
s.releaseUserUploadQuota(userID, day, header.Size)
}
return resource, err
}
@@ -309,7 +325,18 @@ func detectUploadedMimeType(file multipart.File, fileName string, declared strin
return "application/octet-stream"
}
func (s *Service) ImportResourceURL(userID string, rawURL string, kind string, width int, height int, durationMs int64) (*model.Resource, error) {
func (s *Service) ImportResourceURL(userID string, rawURL string, kind string, width int, height int, durationMs int64, uploadIdentity ...string) (*model.Resource, error) {
uploadKey := normalizedResourceUploadKey(uploadIdentity)
existing, err := s.resourceForUploadKey(userID, uploadKey)
if err != nil {
return nil, err
}
if existing != nil && existing.Status == model.ResourceStatusReady {
return existing, nil
}
if existing != nil && existing.Status == model.ResourceStatusPending {
return nil, resourceUploadInProgress()
}
policy, err := s.RuntimePolicy()
if err != nil {
return nil, err
@@ -326,19 +353,50 @@ func (s *Service) ImportResourceURL(userID string, rawURL string, kind string, w
}
}
size := int64(len(payload.data))
if existing != nil {
return s.retryStoredResource(userID, existing, kind, payload.mimeType, size, bytes.NewReader(payload.data))
}
day, err := s.reserveUserUploadQuota(userID, size)
if err != nil {
return nil, err
}
resource, err := s.storeResource(userID, kind, payload.fileName, payload.mimeType, size, width, height, durationMs, bytes.NewReader(payload.data))
resource, stored, err := s.storeResource(userID, kind, payload.fileName, payload.mimeType, size, width, height, durationMs, bytes.NewReader(payload.data), uploadKey)
if err != nil {
s.releaseUserUploadQuota(userID, day, size)
} else {
} else if stored {
s.commitUserUploadQuota(userID, size)
} else {
s.releaseUserUploadQuota(userID, day, size)
}
return resource, err
}
func normalizedResourceUploadKey(values []string) *string {
if len(values) == 0 || strings.TrimSpace(values[0]) == "" {
return nil
}
digest := sha256.Sum256([]byte(strings.TrimSpace(values[0])))
value := hex.EncodeToString(digest[:])
return &value
}
func (s *Service) resourceForUploadKey(userID string, uploadKey *string) (*model.Resource, error) {
if uploadKey == nil {
return nil, nil
}
resource, err := s.repo.ResourceByUploadKey(userID, *uploadKey)
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
return resource, err
}
func resourceUploadInProgress() *AppError {
err := NewAppError(http.StatusConflict, "相同素材正在上传，请稍后重试")
err.Retryable = true
return err
}
func (s *Service) OpenResource(userID string, id string) (*model.Resource, io.ReadCloser, error) {
stream, err := s.OpenResourceRange(userID, id, "")
if err != nil {
@@ -397,16 +455,24 @@ func (s *Service) openResourceRange(userID string, resource *model.Resource, ran
return &ResourceStream{Resource: resource, Body: stream.body, StatusCode: stream.statusCode, ContentLength: stream.contentLength, ContentRange: stream.contentRange, AcceptRanges: stream.acceptRanges}, nil
}
func (s *Service) storeResource(userID string, kind string, fileName string, mimeType string, size int64, width int, height int, durationMs int64, body io.Reader) (*model.Resource, error) {
func (s *Service) storeResource(userID string, kind string, fileName string, mimeType string, size int64, width int, height int, durationMs int64, body io.Reader, uploadKey *string) (*model.Resource, bool, error) {
if existing, err := s.resourceForUploadKey(userID, uploadKey); err != nil {
return nil, false, err
} else if existing != nil {
if existing.Status == model.ResourceStatusReady {
return existing, false, nil
}
return nil, false, resourceUploadInProgress()
}
now := time.Now()
kind = normalizeResourceKind(kind, mimeType)
setting, storageSettingID, useOSS, err := s.activeResourceOSSSetting(userID)
if err != nil {
return nil, err
return nil, false, err
}
provider := "local"
objectKey := localObjectKey(userID, kind, fileName, mimeType, now)
resource := model.Resource{ID: newID(), UserID: userID, Kind: kind, Status: model.ResourceStatusPending, Provider: provider, ObjectKey: objectKey, MimeType: mimeType, Size: size, Width: width, Height: height, DurationMs: durationMs, CreatedAt: now, UpdatedAt: now}
resource := model.Resource{ID: newID(), UserID: userID, Kind: kind, Status: model.ResourceStatusPending, Provider: provider, ObjectKey: objectKey, MimeType: mimeType, Size: size, Width: width, Height: height, DurationMs: durationMs, UploadKey: uploadKey, CreatedAt: now, UpdatedAt: now}
if useOSS {
provider = setting.Provider
objectKey = ossObjectKey(setting, userID, kind, fileName, mimeType, now)
@@ -417,22 +483,18 @@ func (s *Service) storeResource(userID string, kind string, fileName string, mim
resource.ObjectKey = objectKey
}
if err := s.repo.CreateResource(&resource); err != nil {
return nil, err
if existing, lookupErr := s.resourceForUploadKey(userID, uploadKey); lookupErr == nil && existing != nil {
if existing.Status == model.ResourceStatusReady {
return existing, false, nil
}
return nil, false, resourceUploadInProgress()
}
return nil, false, err
}
var etag string
if provider == "local" {
filePath := filepath.Join(s.dataDir, "resources", filepath.FromSlash(objectKey))
if err = os.MkdirAll(filepath.Dir(filePath), 0o750); err == nil {
var file *os.File
file, err = os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o640)
if err == nil {
_, err = io.Copy(file, body)
closeErr := file.Close()
if err == nil {
err = closeErr
}
}
}
err = writeLocalResourceObject(filePath, body)
} else {
etag, err = putOSSObject(setting, objectKey, mimeType, size, body)
}
@@ -441,9 +503,9 @@ func (s *Service) storeResource(userID string, kind string, fileName string, mim
resource.Status = model.ResourceStatusFailed
resource.Error = err.Error()
if saveErr := s.repo.SaveResource(&resource); saveErr != nil {
return nil, errors.Join(err, fmt.Errorf("记录资源失败状态失败：%w", saveErr))
return nil, true, errors.Join(err, fmt.Errorf("记录资源失败状态失败：%w", saveErr))
}
return nil, err
return nil, true, err
}
resource.Status = model.ResourceStatusReady
resource.ETag = etag
@@ -451,21 +513,110 @@ func (s *Service) storeResource(userID string, kind string, fileName string, mim
cleanupErr := s.deleteStoredResourceObject(userID, &resource)
if cleanupErr == nil {
if deleteErr := s.repo.DeleteResource(userID, resource.ID); deleteErr != nil {
return nil, errors.Join(err, fmt.Errorf("清理资源记录失败：%w", deleteErr))
return nil, true, errors.Join(err, fmt.Errorf("清理资源记录失败：%w", deleteErr))
}
return nil, fmt.Errorf("保存资源就绪状态失败：%w", err)
return nil, true, fmt.Errorf("保存资源就绪状态失败：%w", err)
}
resource.Status = model.ResourceStatusFailed
resource.Error = fmt.Sprintf("保存资源就绪状态失败，物理对象清理失败：%v", cleanupErr)
statusErr := s.repo.SaveResource(&resource)
if statusErr != nil {
return nil, errors.Join(err, cleanupErr, fmt.Errorf("记录资源失败状态失败：%w", statusErr))
return nil, true, errors.Join(err, cleanupErr, fmt.Errorf("记录资源失败状态失败：%w", statusErr))
}
return nil, errors.Join(err, fmt.Errorf("清理已上传资源对象失败：%w", cleanupErr))
return nil, true, errors.Join(err, fmt.Errorf("清理已上传资源对象失败：%w", cleanupErr))
}
s.recordActivity(userID, "resource", 1)
return &resource, nil
return &resource, true, nil
}
func writeLocalResourceObject(filePath string, body io.Reader) error {
if err := os.MkdirAll(filepath.Dir(filePath), 0o750); err != nil {
return err
}
file, err := os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o640)
if err != nil {
return err
}
_, copyErr := io.Copy(file, body)
closeErr := file.Close()
if copyErr != nil {
return copyErr
}
return closeErr
}
func (s *Service) retryStoredResource(userID string, resource *model.Resource, kind string, mimeType string, size int64, body io.Reader) (*model.Resource, error) {
if resource == nil {
return nil, errors.New("资源不存在")
}
if resource.Status == model.ResourceStatusReady {
return resource, nil
}
if resource.Status != model.ResourceStatusFailed {
return nil, resourceUploadInProgress()
}
kind = normalizeResourceKind(kind, mimeType)
if resource.Size != size || resource.Kind != kind || (resource.MimeType != "" && mimeType != "" && resource.MimeType != mimeType) {
return nil, NewAppError(http.StatusConflict, "上传幂等标识已用于其他文件")
}
claimed, err := s.repo.ClaimFailedResourceUpload(userID, resource.ID)
if err != nil {
return nil, err
}
if !claimed {
latest, latestErr := s.repo.ResourceForUser(userID, resource.ID)
if latestErr == nil && latest.Status == model.ResourceStatusReady {
return latest, nil
}
return nil, resourceUploadInProgress()
}
resource.Status = model.ResourceStatusPending
resource.Error = ""
resource.UpdatedAt = time.Now()
day, err := s.reserveRetryUploadQuota(userID, size)
if err != nil {
resource.Status = model.ResourceStatusFailed
resource.Error = err.Error()
resource.UpdatedAt = time.Now()
if saveErr := s.repo.SaveResource(resource); saveErr != nil {
return nil, errors.Join(err, fmt.Errorf("恢复资源重试失败状态失败：%w", saveErr))
}
return nil, err
}
var etag string
if resource.Provider == "local" {
err = writeLocalResourceObject(filepath.Join(s.dataDir, "resources", filepath.FromSlash(resource.ObjectKey)), body)
} else {
var setting ossSettingValue
setting, err = s.ossSettingForResource(userID, resource)
if err == nil {
etag, err = putOSSObject(setting, resource.ObjectKey, resource.MimeType, resource.Size, body)
}
}
resource.UpdatedAt = time.Now()
if err != nil {
s.releaseRetryUploadQuota(userID, day, size)
resource.Status = model.ResourceStatusFailed
resource.Error = err.Error()
if saveErr := s.repo.SaveResource(resource); saveErr != nil {
return nil, errors.Join(err, fmt.Errorf("记录资源重试失败状态失败：%w", saveErr))
}
return nil, err
}
resource.Status = model.ResourceStatusReady
resource.ETag = etag
if err := s.repo.SaveResource(resource); err != nil {
s.releaseRetryUploadQuota(userID, day, size)
resource.Status = model.ResourceStatusFailed
resource.Error = "保存资源重试就绪状态失败"
if saveErr := s.repo.SaveResource(resource); saveErr != nil {
return nil, errors.Join(err, fmt.Errorf("记录资源重试失败状态失败：%w", saveErr))
}
return nil, fmt.Errorf("保存资源重试就绪状态失败：%w", err)
}
s.recordActivity(userID, "resource", 1)
return resource, nil
}
func localObjectKey(userID string, kind string, fileName string, mimeType string, now time.Time) string {
@@ -534,7 +685,7 @@ func (s *Service) persistGeneratedMediaValueMode(userID string, value interface{
return nil, err
}
}
resource, err := s.storeResource(userID, kind, "generated."+extensionFromMimeType(mimeType), mimeType, int64(len(data)), width, height, int64(intValue(item["durationMs"])), bytes.NewReader(data))
resource, _, err := s.storeResource(userID, kind, "generated."+extensionFromMimeType(mimeType), mimeType, int64(len(data)), width, height, int64(intValue(item["durationMs"])), bytes.NewReader(data), nil)
if err != nil {
if enforceQuota {
s.releaseUserUploadQuota(userID, quotaDay, int64(len(data)))

88
backend/internal/service/resource_test.go
View File

@@ -3,6 +3,7 @@ package service
import (
"bytes"
"encoding/json"
"errors"
"hash/crc64"
"io"
"net/http"
@@ -11,6 +12,7 @@ import (
"strconv"
"strings"
"testing"
"testing/iotest"
"time"
"infinite-canvas/backend/internal/model"
@@ -831,6 +833,92 @@ func newResourceTestService(t *testing.T) *Service {
return &Service{repo: repository.New(db), dataDir: t.TempDir()}
}
func TestStoreResourceReusesReadyUploadIdentity(t *testing.T) {
svc := newResourceTestService(t)
uploadKey := normalizedResourceUploadKey([]string{"image:user-1:logical-upload"})
first, stored, err := svc.storeResource("user-1", "image", "first.png", "image/png", 7, 1, 1, 0, bytes.NewReader([]byte("payload")), uploadKey)
if err != nil {
t.Fatal(err)
}
if !stored {
t.Fatal("first upload was not stored")
}
second, stored, err := svc.storeResource("user-1", "image", "second.png", "image/png", 7, 1, 1, 0, bytes.NewReader([]byte("payload")), uploadKey)
if err != nil {
t.Fatal(err)
}
if stored || second.ID != first.ID || second.ObjectKey != first.ObjectKey {
t.Fatalf("idempotent upload = %#v, stored=%v; first=%#v", second, stored, first)
}
resources, err := svc.repo.Resources("user-1", 10)
if err != nil {
t.Fatal(err)
}
if len(resources) != 1 {
t.Fatalf("resource count = %d, want 1", len(resources))
}
}
func TestRetryStoredResourceKeepsOriginalObjectKey(t *testing.T) {
svc := newResourceTestService(t)
uploadKey := normalizedResourceUploadKey([]string{"image:user-1:retry-upload"})
failed := &model.Resource{
ID: "resource-failed", UserID: "user-1", Kind: "image", Status: model.ResourceStatusFailed,
Provider: "local", ObjectKey: "users/user-1/image/fixed.png", MimeType: "image/png", Size: 7,
UploadKey: uploadKey, CreatedAt: time.Now(), UpdatedAt: time.Now(),
}
if err := svc.repo.CreateResource(failed); err != nil {
t.Fatal(err)
}
retried, err := svc.retryStoredResource("user-1", failed, "image", "image/png", 7, bytes.NewReader([]byte("payload")))
if err != nil {
t.Fatal(err)
}
if retried.ID != failed.ID || retried.ObjectKey != "users/user-1/image/fixed.png" || retried.Status != model.ResourceStatusReady {
t.Fatalf("retried resource = %#v", retried)
}
resources, err := svc.repo.Resources("user-1", 10)
if err != nil {
t.Fatal(err)
}
if len(resources) != 1 {
t.Fatalf("resource count = %d, want 1", len(resources))
}
day := time.Now().UTC().Format("2006-01-02")
usage, err := svc.repo.DailyUploadBytes("user-1", day)
if err != nil {
t.Fatal(err)
}
if usage != 7 {
t.Fatalf("daily upload usage = %d, want 7", usage)
}
}
func TestRetryStoredResourceReleasesDailyQuotaAfterFailure(t *testing.T) {
svc := newResourceTestService(t)
uploadKey := normalizedResourceUploadKey([]string{"image:user-1:failed-retry"})
failed := &model.Resource{
ID: "resource-failed-retry", UserID: "user-1", Kind: "image", Status: model.ResourceStatusFailed,
Provider: "local", ObjectKey: "users/user-1/image/failed.png", MimeType: "image/png", Size: 7,
UploadKey: uploadKey, CreatedAt: time.Now(), UpdatedAt: time.Now(),
}
if err := svc.repo.CreateResource(failed); err != nil {
t.Fatal(err)
}
_, err := svc.retryStoredResource("user-1", failed, "image", "image/png", 7, iotest.ErrReader(errors.New("write failed")))
if err == nil || !strings.Contains(err.Error(), "write failed") {
t.Fatalf("retryStoredResource() error = %v", err)
}
day := time.Now().UTC().Format("2006-01-02")
usage, usageErr := svc.repo.DailyUploadBytes("user-1", day)
if usageErr != nil {
t.Fatal(usageErr)
}
if usage != 0 {
t.Fatalf("daily upload usage = %d, want 0", usage)
}
}
func TestLegacyMediaMigrationSkipsInvalidDataURL(t *testing.T) {
svc := &Service{}
input := map[string]interface{}{

35
backend/internal/service/upload_quota.go
View File

@@ -34,6 +34,30 @@ func (s *Service) reserveGeneratedResourceQuota(userID string, size int64) (stri
return s.reserveUserStoredFileQuota(userID, size, megabytes(policy.Resource.GeneratedFileMB)+1, megabytes(policy.Resource.DailyUploadMB), gigabytes(policy.Resource.StoredFileGB), fmt.Sprintf("单个生成文件不能超过 %dMB", policy.Resource.GeneratedFileMB))
}
// 失败资源的记录已经计入账号存储用量；重试只重新预留当日上传额度，避免重复计算存储容量。
func (s *Service) reserveRetryUploadQuota(userID string, size int64) (string, error) {
policy, err := s.RuntimePolicy()
if err != nil {
return "", err
}
if size <= 0 {
return "", BadAuthRequest("上传文件不能为空")
}
if size >= megabytes(policy.Resource.ResourceUploadMB) {
return "", BadAuthRequest(fmt.Sprintf("单个上传文件必须小于 %dMB", policy.Resource.ResourceUploadMB))
}
day := time.Now().UTC().Format("2006-01-02")
s.storageMu.Lock()
defer s.storageMu.Unlock()
if err := s.repo.ReserveDailyUpload(userID, day, size, megabytes(policy.Resource.DailyUploadMB)); err != nil {
if errors.Is(err, repository.ErrDailyUploadLimitExceeded) {
return "", BadAuthRequest(fmt.Sprintf("每个账号 UTC 自然日上传总量必须小于 %s", formatStorageLimit(megabytes(policy.Resource.DailyUploadMB))))
}
return "", err
}
return day, nil
}
func (s *Service) reserveUserStoredFileQuota(userID string, size int64, exclusiveSingleFileLimit int64, dailyLimit int64, storedLimit int64, singleFileMessage string) (string, error) {
if size <= 0 {
return "", BadAuthRequest("上传文件不能为空")
@@ -84,6 +108,17 @@ func (s *Service) releaseUserUploadQuota(userID string, day string, size int64)
}
}
func (s *Service) releaseRetryUploadQuota(userID string, day string, size int64) {
if day == "" || size <= 0 {
return
}
s.storageMu.Lock()
defer s.storageMu.Unlock()
if err := s.repo.ReleaseDailyUpload(userID, day, size); err != nil {
log.Printf("release retry upload quota failed: user=%s day=%s size=%d error=%v", userID, day, size, err)
}
}
func (s *Service) commitUserUploadQuota(userID string, size int64) {
if size <= 0 {
return