diff --git a/backend/internal/handler/user_data.go b/backend/internal/handler/user_data.go index fe7458ddf..ffa8c4409 100644 --- a/backend/internal/handler/user_data.go +++ b/backend/internal/handler/user_data.go @@ -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 diff --git a/backend/internal/model/models_project.go b/backend/internal/model/models_project.go index 620b821aa..986584583 100644 --- a/backend/internal/model/models_project.go +++ b/backend/internal/model/models_project.go @@ -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 diff --git a/backend/internal/repository/repository.go b/backend/internal/repository/repository.go index 13453e931..4c393a5f8 100644 --- a/backend/internal/repository/repository.go +++ b/backend/internal/repository/repository.go @@ -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 } diff --git a/backend/internal/service/resource.go b/backend/internal/service/resource.go index 551b29032..f35e2d046 100644 --- a/backend/internal/service/resource.go +++ b/backend/internal/service/resource.go @@ -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,98 @@ 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() + 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 { + 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 { + 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 +673,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))) diff --git a/backend/internal/service/resource_test.go b/backend/internal/service/resource_test.go index 93e8f5858..3d011c58c 100644 --- a/backend/internal/service/resource_test.go +++ b/backend/internal/service/resource_test.go @@ -831,6 +831,59 @@ 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)) + } +} + func TestLegacyMediaMigrationSkipsInvalidDataURL(t *testing.T) { svc := &Service{} input := map[string]interface{}{ diff --git a/docs/content/docs/backend/backend-database.mdx b/docs/content/docs/backend/backend-database.mdx index 6f92e552e..13da5e499 100644 --- a/docs/content/docs/backend/backend-database.mdx +++ b/docs/content/docs/backend/backend-database.mdx @@ -128,8 +128,10 @@ description: 影策后端核心持久化表与状态边界 | 表 | 用途 | 关键边界 | | --- | --- | --- | -| `resources` | 保存资源元数据和实际存储位置引用 | S3 资源必须通过 `storage_setting_id` 固定绑定创建时的位置；浏览器默认经同源接口读取 | +| `resources` | 保存资源元数据和实际存储位置引用 | `user_id + upload_key` 唯一约束保证逻辑上传幂等；S3 资源必须通过 `storage_setting_id` 固定绑定创建时的位置；浏览器默认经同源接口读取 | | `storage_locations` | 保存平台或个人 S3 位置版本、加密凭据和连接测试摘要 | Endpoint、Bucket、Region、PathStyle、PathPrefix、Provider 组成不可变位置；同位置只轮换凭据，位置变化创建新记录 | | `resource_deletion_jobs` | 在业务记录删除后可靠清理物理对象 | 固化资源的存储位置与对象键；删除失败保留任务并重试，S3 404 按幂等成功处理 | `storage_locations.value_json` 中的 Secret Key 和 Session Token 使用后端设置密钥加密。启用 S3 前，关键位置字段和凭据摘要必须与最近一次 Put、Range Get、Delete 连接测试结果一致；历史资源继续按绑定的位置读取和删除，不随当前配置切换。 + +`resources.upload_key` 保存客户端 `X-Idempotency-Key` 的 SHA-256 摘要。相同用户重复提交同一逻辑上传时复用已有资源；失败重试继续使用原对象路径，避免对象存储已经接收请求但响应丢失时产生第二份物理对象。未提供幂等标识的后台生成资源保持 `NULL`，不参与该唯一约束。 diff --git a/docs/content/docs/progress/pending-test.mdx b/docs/content/docs/progress/pending-test.mdx index 26b673b09..146fd4e52 100644 --- a/docs/content/docs/progress/pending-test.mdx +++ b/docs/content/docs/progress/pending-test.mdx @@ -626,6 +626,16 @@ description: 影策已实现功能的手工验证清单与未确认边界 - [ ] 创建两条共享同一物理对象的资源记录，只删除其中一条，确认另一条仍能预览且物理对象未清理；两条同时删除时确认只产生一个物理删除任务。 - [ ] 模拟对象存储暂时失败，确认资源记录事务仍成功、删除任务保留并重试，恢复后物理对象最终清理。 +## 画布素材上传幂等 + +画布上传、远端用户数据同步与后端资源写入已使用同一逻辑上传身份，避免腾讯云 COS 等对象存储因自动同步、并发请求或响应丢失产生重复对象： + +- [x] 前端回归测试补充“内嵌图片上传期间继续编辑”，要求同一 `coverUrl/dataUrl` 跨同步批次只调用一次资源上传。 +- [x] 后端专项测试覆盖相同上传身份复用就绪资源，以及失败重试保持原资源 ID 和对象路径。 +- [ ] 在腾讯云 COS 测试桶上传图片后连续移动节点、修改标题并刷新页面，确认后台始终只有一条资源记录和一个物理对象。 +- [ ] 在 COS 已接收上传后中断浏览器响应，再恢复网络触发自动同步，确认返回原资源 ID，用户存储用量不增加。 +- [ ] 使用两个标签页并发提交相同 `X-Idempotency-Key`，确认一个请求完成上传，另一个复用结果或返回可重试冲突，不生成第二个对象路径。 + ## 短剧分镜工作流工作台 短剧工作流已按镜头生产重构为单层项目工作台，并将资产绑定、镜头编辑、生成预览和整集镜头轨道放入同一视口；以下场景仍需用户在真实项目数据与模型渠道下确认： diff --git a/web/src/services/api/resources.ts b/web/src/services/api/resources.ts index fb2910ed4..bad4f4c9a 100644 --- a/web/src/services/api/resources.ts +++ b/web/src/services/api/resources.ts @@ -62,6 +62,14 @@ export type ArkPrivateAssetSync = { status: "active" | string; }; +type ResourceUploadMeta = { + width?: number; + height?: number; + durationMs?: number; + fileName?: string; + idempotencyKey?: string; +}; + const api = apiClient; const resourceCache = new Map(); const resourceRequests = new Map>(); @@ -103,7 +111,7 @@ export function isResourceUrl(url?: string) { return path.startsWith(`${base}/resources/`) && path.endsWith("/file"); } -export async function uploadResourceFile(file: Blob, kind: "image" | "video" | "audio" | "file", meta?: { width?: number; height?: number; durationMs?: number; fileName?: string }) { +export async function uploadResourceFile(file: Blob, kind: "image" | "video" | "audio" | "file", meta?: ResourceUploadMeta) { const formData = new FormData(); const name = meta?.fileName || (file instanceof File ? file.name : `${kind}.${extensionFromMime(file.type, kind)}`); formData.append("kind", kind); @@ -111,17 +119,22 @@ export async function uploadResourceFile(file: Blob, kind: "image" | "video" | " if (meta?.width) formData.append("width", String(Math.round(meta.width))); if (meta?.height) formData.append("height", String(Math.round(meta.height))); if (meta?.durationMs) formData.append("durationMs", String(Math.round(meta.durationMs))); - const data = await request<{ resource: RemoteResource }>(api.post("/resources", formData)); + const data = await request<{ resource: RemoteResource }>(api.post("/resources", formData, uploadRequestConfig(meta?.idempotencyKey))); resourceCache.set(resourceCacheKey(data.resource.id), data.resource); return data.resource; } -export async function importResourceFromUrl(url: string, kind: "image" | "video" | "audio" | "file", meta?: { width?: number; height?: number; durationMs?: number }) { - const data = await request<{ resource: RemoteResource }>(api.post("/resources/import", { url, kind, width: meta?.width, height: meta?.height, durationMs: meta?.durationMs })); +export async function importResourceFromUrl(url: string, kind: "image" | "video" | "audio" | "file", meta?: Omit) { + const data = await request<{ resource: RemoteResource }>(api.post("/resources/import", { url, kind, width: meta?.width, height: meta?.height, durationMs: meta?.durationMs }, uploadRequestConfig(meta?.idempotencyKey))); resourceCache.set(resourceCacheKey(data.resource.id), data.resource); return data.resource; } +function uploadRequestConfig(idempotencyKey?: string) { + const value = idempotencyKey?.trim(); + return value ? { headers: { "X-Idempotency-Key": value } } : undefined; +} + export function getResource(id: string): Promise { const cacheKey = resourceCacheKey(id); const cached = resourceCache.get(cacheKey); diff --git a/web/src/services/file-storage.ts b/web/src/services/file-storage.ts index 681faabaa..0d34ff438 100644 --- a/web/src/services/file-storage.ts +++ b/web/src/services/file-storage.ts @@ -13,6 +13,8 @@ const store = localforage.createInstance({ name: "infinite-canvas", storeName: " const objectUrls = new Map(); export async function uploadMediaFile(input: string | Blob, prefix = "file"): Promise { + // 直传和失败后的本地同步必须复用同一上传身份，避免响应丢失后创建第二个对象。 + const storageKey = `${prefix}:${getActiveUserScope()}:${nanoid()}`; const blob = typeof input === "string" ? await (await fetch(input)).blob() : input; const previewUrl = URL.createObjectURL(blob); const captured = blob.type.startsWith("video/") ? await captureVideoPoster(previewUrl).catch(() => undefined) : undefined; @@ -29,14 +31,13 @@ export async function uploadMediaFile(input: string | Blob, prefix = "file"): Pr const poster = captured?.poster ? await uploadImage(captured.poster).catch(() => undefined) : undefined; try { const kind = blob.type.startsWith("video/") ? "video" : blob.type.startsWith("audio/") ? "audio" : "file"; - const resource = await uploadResourceFile(blob, kind, { ...meta, fileName: input instanceof File ? input.name : undefined }); + const resource = await uploadResourceFile(blob, kind, { ...meta, fileName: input instanceof File ? input.name : undefined, idempotencyKey: storageKey }); await primeResourceBlobCache(resourceStorageKey(resource.id), blob).catch(() => ""); URL.revokeObjectURL(previewUrl); return { url: resource.publicUrl || resourceFileUrl(resource.id), storageKey: resourceStorageKey(resource.id), bytes: resource.size || blob.size, mimeType: resource.mimeType || blob.type || "application/octet-stream", width: resource.width || meta.width, height: resource.height || meta.height, durationMs: resource.durationMs || meta.durationMs, hasAudio: meta.hasAudio, preview: poster }; } catch { // OSS is optional during local/self-hosted setup. Keep the existing local fallback. } - const storageKey = `${prefix}:${getActiveUserScope()}:${nanoid()}`; await store.setItem(storageKey, blob); const url = previewUrl; objectUrls.set(storageKey, url); diff --git a/web/src/services/image-storage.ts b/web/src/services/image-storage.ts index d8e3b7618..27c72e9ee 100644 --- a/web/src/services/image-storage.ts +++ b/web/src/services/image-storage.ts @@ -19,9 +19,12 @@ const store = localforage.createInstance({ name: "infinite-canvas", storeName: " const objectUrls = new Map(); export async function uploadImage(input: string | Blob): Promise { + // 同一个逻辑上传在直传失败后会退回 IndexedDB，并由云端数据同步再次提交。 + // 提前生成本地 key，确保两条路径向后端发送相同的幂等标识。 + const storageKey = `image:${getActiveUserScope()}:${nanoid()}`; if (typeof input === "string" && shouldImportRemoteImage(input)) { try { - const resource = await importResourceFromUrl(input, "image"); + const resource = await importResourceFromUrl(input, "image", { idempotencyKey: storageKey }); return { url: resource.publicUrl || resourceFileUrl(resource.id), storageKey: resourceStorageKey(resource.id), @@ -38,7 +41,7 @@ export async function uploadImage(input: string | Blob): Promise const previewUrl = URL.createObjectURL(blob); const meta = await readImageMeta(previewUrl); try { - const resource = await uploadResourceFile(blob, "image", { width: meta.width, height: meta.height, fileName: input instanceof File ? input.name : undefined }); + const resource = await uploadResourceFile(blob, "image", { width: meta.width, height: meta.height, fileName: input instanceof File ? input.name : undefined, idempotencyKey: storageKey }); await primeResourceBlobCache(resourceStorageKey(resource.id), blob).catch(() => ""); URL.revokeObjectURL(previewUrl); return { @@ -52,7 +55,6 @@ export async function uploadImage(input: string | Blob): Promise } catch { // OSS is optional during local/self-hosted setup. Keep the existing local fallback. } - const storageKey = `image:${getActiveUserScope()}:${nanoid()}`; await store.setItem(storageKey, blob); const url = previewUrl; objectUrls.set(storageKey, url); diff --git a/web/src/services/user-data-sync.ts b/web/src/services/user-data-sync.ts index b26f63988..8c5dd2c5d 100644 --- a/web/src/services/user-data-sync.ts +++ b/web/src/services/user-data-sync.ts @@ -19,6 +19,7 @@ let remoteOperationTail: Promise = Promise.resolve(); let subscriptionsInstalled = false; let acknowledgedAssets = new Map(); let acknowledgedProjects = new Map(); +const resolvedResourceReferences = new Map(); const LOCAL_STORAGE_KEY_PATTERN = /^(image|video|audio|file|video-reference|audio-reference):/; @@ -27,6 +28,7 @@ export async function syncRemoteUserData(userId?: string | null) { activeRemoteUserId = userId || ""; acknowledgedProjects.clear(); acknowledgedAssets.clear(); + resolvedResourceReferences.clear(); if (!activeRemoteUserId) { remoteUserDataPhase = "inactive"; return; @@ -67,6 +69,7 @@ export function resetRemoteUserDataSync() { remoteUserDataPhase = "inactive"; acknowledgedAssets.clear(); acknowledgedProjects.clear(); + resolvedResourceReferences.clear(); if (syncTimer) { window.clearTimeout(syncTimer); syncTimer = null; @@ -219,13 +222,18 @@ async function ensureRemoteResourceReferences(value: T, uploaded = new Map) { return ""; } -async function uploadInlineDataUrl(dataUrl: string) { +async function uploadInlineDataUrl(dataUrl: string, identity: string) { const response = await fetch(dataUrl); if (!response.ok) throw new Error("内嵌媒体读取失败"); const blob = await response.blob(); const kind: "image" | "video" | "audio" | "file" = blob.type.startsWith("image/") ? "image" : blob.type.startsWith("video/") ? "video" : blob.type.startsWith("audio/") ? "audio" : "file"; - const resource = await uploadResourceFile(blob, kind); + const resource = await uploadResourceFile(blob, kind, { idempotencyKey: identity }); return resourceStorageKey(resource.id); } +async function inlineMediaUploadIdentity(dataUrl: string) { + const digest = await globalThis.crypto.subtle.digest("SHA-256", new TextEncoder().encode(dataUrl)); + return `inline:${Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join("")}`; +} + async function uploadLocalStorageKey(storageKey: string, payload: Record) { const blob = storageKey.startsWith("image:") ? await getImageBlob(storageKey) : await getMediaBlob(storageKey); if (!blob) throw new Error(`本地媒体不存在：${storageKey}`); @@ -263,6 +276,7 @@ async function uploadLocalStorageKey(storageKey: string, payload: Record = {}): AiConfig { + const modelName = model.includes("::") ? model.slice(model.indexOf("::") + 2) : model; + return { + ...defaultConfig, + ...overrides, + model, + channels: [ + { + id: "default", + name: "Backend test channel", + baseUrl: "https://api.example.com", + apiKey: "test-key", + apiFormat: "openai", + interfaceType: "newapi-channel-2", + models: [modelName], + scope: "user", + }, + ], + }; +} + test("Dreamina submit failure categories have bounded user-facing messages", () => { const cases = [ ["dreamina_submit_spawn_failed", "无法启动官方即梦 CLI，任务尚未提交。"], @@ -1286,7 +1307,7 @@ test("remote provider keeps Create resolution semantics and still creates one Ba { mode: "video", prompt: "A remote test clip", - config: { ...defaultConfig, model: "default::grok-imagine-video", vquality: "720", quality: "auto" }, + config: remoteBackendConfig("default::grok-imagine-video", { vquality: "720", quality: "auto" }), }, { createTask: async (input) => { @@ -1354,7 +1375,7 @@ test("remote image video and audio references keep Backend parity without Dreami { mode: item.mode, prompt: "Remote parity fixture", - config: { ...defaultConfig, model: "default::provider-neutral-model" }, + config: remoteBackendConfig("default::provider-neutral-model"), ...item.references, }, { @@ -1673,7 +1694,7 @@ test("Create audio upload converts, previews, removes, and submits through the s projectId: running.projectId, mode: "video", prompt: running.prompt, - config: { ...defaultConfig, model: "remote-video-audio-fixture", videoModel: "remote-video-audio-fixture" }, + config: remoteBackendConfig("remote-video-audio-fixture", { videoModel: "remote-video-audio-fixture" }), ...references, } as Parameters[0], { diff --git a/web/test/user-scoped-generation-persistence.test.ts b/web/test/user-scoped-generation-persistence.test.ts index 7a1f49a33..d1a8ede27 100644 --- a/web/test/user-scoped-generation-persistence.test.ts +++ b/web/test/user-scoped-generation-persistence.test.ts @@ -4251,6 +4251,7 @@ test("remote resource preparation never overwrites an edit made while upload is const previousAssets = useAssetStore.getState().assets; const indexedValues = new Map(); const remoteWrites: Asset[] = []; + let resourceUploads = 0; let releaseUpload!: () => void; const uploadReleased = new Promise((resolve) => { releaseUpload = resolve; @@ -4280,6 +4281,7 @@ test("remote resource preparation never overwrites an edit made while upload is return { data: { code: 0, data: { projects: [], assets: [] }, msg: "" }, status: 200, statusText: "OK", headers: {}, config }; } if (method === "post" && url === "/resources") { + resourceUploads += 1; uploadStartedResolve(); await uploadReleased; return { @@ -4343,6 +4345,7 @@ test("remote resource preparation never overwrites an edit made while upload is expect(useAssetStore.getState().assets.find((item) => item.id === asset.id)?.title).toBe("edited during upload"); expect(remoteWrites.at(-1)?.title).toBe("edited during upload"); + expect(resourceUploads).toBe(1); } finally { releaseUpload(); resetRemoteUserDataSync();