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

fix(storage): prevent duplicate COS resource uploads

Browse Source
This commit is contained in:
MasterChiefCN
2026-09-01 17:10:55 +08:00
parent e1f2f16e85
commit 2a33a20628
13 changed files with 339 additions and 64 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
}

201
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,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)))

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

@@ -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{}{

4
docs/content/docs/backend/backend-database.mdx
View File

@@ -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`，不参与该唯一约束。

10
docs/content/docs/progress/pending-test.mdx
View File

@@ -626,6 +626,16 @@ description: 影策已实现功能的手工验证清单与未确认边界
- [ ] 创建两条共享同一物理对象的资源记录，只删除其中一条，确认另一条仍能预览且物理对象未清理；两条同时删除时确认只产生一个物理删除任务。
- [ ] 模拟对象存储暂时失败，确认资源记录事务仍成功、删除任务保留并重试，恢复后物理对象最终清理。
## 画布素材上传幂等
画布上传、远端用户数据同步与后端资源写入已使用同一逻辑上传身份，避免腾讯云 COS 等对象存储因自动同步、并发请求或响应丢失产生重复对象：
- [x] 前端回归测试补充“内嵌图片上传期间继续编辑”，要求同一 `coverUrl/dataUrl` 跨同步批次只调用一次资源上传。
- [x] 后端专项测试覆盖相同上传身份复用就绪资源，以及失败重试保持原资源 ID 和对象路径。
- [ ] 在腾讯云 COS 测试桶上传图片后连续移动节点、修改标题并刷新页面，确认后台始终只有一条资源记录和一个物理对象。
- [ ] 在 COS 已接收上传后中断浏览器响应，再恢复网络触发自动同步，确认返回原资源 ID，用户存储用量不增加。
- [ ] 使用两个标签页并发提交相同 `X-Idempotency-Key`，确认一个请求完成上传，另一个复用结果或返回可重试冲突，不生成第二个对象路径。
## 短剧分镜工作流工作台
短剧工作流已按镜头生产重构为单层项目工作台，并将资产绑定、镜头编辑、生成预览和整集镜头轨道放入同一视口；以下场景仍需用户在真实项目数据与模型渠道下确认：

21
web/src/services/api/resources.ts
View File

@@ -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<string, RemoteResource>();
const resourceRequests = new Map<string, Promise<RemoteResource>>();
@@ -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<ResourceUploadMeta, "fileName">) {
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<RemoteResource> {
const cacheKey = resourceCacheKey(id);
const cached = resourceCache.get(cacheKey);

5
web/src/services/file-storage.ts
View File

@@ -13,6 +13,8 @@ const store = localforage.createInstance({ name: "infinite-canvas", storeName: "
const objectUrls = new Map<string, string>();
export async function uploadMediaFile(input: string | Blob, prefix = "file"): Promise<UploadedFile> {
// 直传和失败后的本地同步必须复用同一上传身份，避免响应丢失后创建第二个对象。
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);

8
web/src/services/image-storage.ts
View File

@@ -19,9 +19,12 @@ const store = localforage.createInstance({ name: "infinite-canvas", storeName: "
const objectUrls = new Map<string, string>();
export async function uploadImage(input: string | Blob): Promise<UploadedImage> {
// 同一个逻辑上传在直传失败后会退回 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<UploadedImage>
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<UploadedImage>
} 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);

22
web/src/services/user-data-sync.ts
View File

@@ -19,6 +19,7 @@ let remoteOperationTail: Promise<void> = Promise.resolve();
let subscriptionsInstalled = false;
let acknowledgedAssets = new Map<string, Asset>();
let acknowledgedProjects = new Map<string, CanvasProject>();
const resolvedResourceReferences = new Map<string, string>();
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<T>(value: T, uploaded = new Map<st
if (!isLocalStorageKey(storageKey)) {
const inline = inlineMediaDataUrl(next);
if (!inline) return next as T;
const resourceStorage = await uploadInlineDataUrl(inline);
const identity = await inlineMediaUploadIdentity(inline);
const cached = uploaded.get(identity) || resolvedResourceReferences.get(identity);
const resourceStorage = cached || (await uploadInlineDataUrl(inline, identity));
uploaded.set(identity, resourceStorage);
resolvedResourceReferences.set(identity, resourceStorage);
return applyResourceReference(next, resourceStorage) as T;
}
const cached = uploaded.get(storageKey);
const cached = uploaded.get(storageKey) || resolvedResourceReferences.get(storageKey);
const resourceStorage = cached || (await uploadLocalStorageKey(storageKey, next));
uploaded.set(storageKey, resourceStorage);
resolvedResourceReferences.set(storageKey, resourceStorage);
return applyResourceReference(next, resourceStorage) as T;
}
@@ -246,15 +254,20 @@ function inlineMediaDataUrl(payload: Record<string, unknown>) {
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<string, unknown>) {
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<string,
width: numberValue(payload.naturalWidth) || numberValue(payload.width),
height: numberValue(payload.naturalHeight) || numberValue(payload.height),
durationMs: numberValue(payload.durationMs),
idempotencyKey: storageKey,
});
return resourceStorageKey(resource.id);
}

29
web/test/generation-task-local.test.ts
View File

@@ -1,6 +1,6 @@
import { expect, test } from "bun:test";
import { defaultConfig } from "../src/stores/use-config-store";
import { defaultConfig, type AiConfig } from "../src/stores/use-config-store";
import { runBackendGenerationTask, runBackendGenerationTaskBatch } from "../src/services/api/generation-task";
import { deleteGenerationTask, formatTaskLog, listGenerationTasks, projectBackendSafeTaskLog, splitGenerationTaskObservationIds, type GenerationTask } from "../src/services/api/task-center";
import { isLocalDreaminaBackgroundTask, localDreaminaCancellationCopy, localDreaminaDetachOutcome, projectLocalDreaminaTask } from "../src/services/local-dreamina-task-projection";
@@ -24,6 +24,27 @@ function sourceSection(source: string, startMarker: string, endMarker: string) {
return compactSource(source.slice(start, end));
}
function remoteBackendConfig(model: string, overrides: Partial<AiConfig> = {}): 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<typeof runBackendGenerationTask>[0],
{

3
web/test/user-scoped-generation-persistence.test.ts
View File

@@ -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<string, string>();
const remoteWrites: Asset[] = [];
let resourceUploads = 0;
let releaseUpload!: () => void;
const uploadReleased = new Promise<void>((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();