fix(canvas): 资源引用 - 校验递归素材并修复断链
This commit is contained in:
@@ -17,13 +17,16 @@ func main() {
|
||||
command = strings.ToLower(strings.TrimSpace(os.Args[1]))
|
||||
}
|
||||
apply := false
|
||||
if command == "repair-asset-bytes" {
|
||||
if command == "repair-asset-bytes" || command == "repair-canvas-resources" {
|
||||
if len(os.Args) == 3 && os.Args[2] == "--apply" {
|
||||
apply = true
|
||||
} else if len(os.Args) > 2 {
|
||||
log.Fatal("用法:migrate-schema repair-asset-bytes [--apply];默认只检查")
|
||||
log.Fatalf("用法:migrate-schema %s [--apply];默认只检查", command)
|
||||
}
|
||||
}
|
||||
if command == "repair-canvas-resources" && strings.TrimSpace(os.Getenv("CANVAS_BACKEND_DATA_DIR")) == "" && strings.TrimSpace(os.Getenv("DATABASE_URL")) == "" {
|
||||
log.Fatal("资源修复必须显式设置 CANVAS_BACKEND_DATA_DIR 或 DATABASE_URL;拒绝使用默认数据库")
|
||||
}
|
||||
db, err := database.Open(database.Config{
|
||||
Driver: env("CANVAS_DATABASE_DRIVER", "sqlite"),
|
||||
DSN: os.Getenv("DATABASE_URL"),
|
||||
@@ -42,6 +45,23 @@ func main() {
|
||||
defer sqlDB.Close()
|
||||
|
||||
switch command {
|
||||
case "repair-canvas-resources":
|
||||
if err := database.RequireSchemaVersion(db); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
report, repairErr := canvas.RepairCanvasResources(db, apply)
|
||||
encoded, err := json.Marshal(report)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
fmt.Println(string(encoded))
|
||||
if repairErr != nil {
|
||||
log.Fatal(repairErr)
|
||||
}
|
||||
if report.Unresolved > 0 {
|
||||
log.Fatal("存在需人工处理的画布,请按报告恢复历史或重新上传;未删除主媒体")
|
||||
}
|
||||
return
|
||||
case "repair-asset-bytes":
|
||||
report, repairErr := canvas.RepairAssetBytes(db, apply)
|
||||
encoded, err := json.Marshal(report)
|
||||
@@ -66,7 +86,7 @@ func main() {
|
||||
log.Fatal(err)
|
||||
}
|
||||
default:
|
||||
log.Fatalf("未知命令 %q;可用命令:up、status、verify、repair-asset-bytes", command)
|
||||
log.Fatalf("未知命令 %q;可用命令:up、status、verify、repair-asset-bytes、repair-canvas-resources", command)
|
||||
}
|
||||
status, err := database.ReadSchemaStatus(db)
|
||||
if err != nil {
|
||||
|
||||
@@ -218,6 +218,10 @@ func (s *Service) UpsertUserCanvasProject(userID string, raw json.RawMessage) (U
|
||||
return s.canvasDomain().UpsertUserCanvasProject(userID, raw)
|
||||
}
|
||||
|
||||
func (s *Service) RepairUserCanvasProject(userID string, raw json.RawMessage) (UserDataSummary, error) {
|
||||
return s.canvasDomain().RepairUserCanvasProject(userID, raw)
|
||||
}
|
||||
|
||||
func (s *Service) DeleteUserCanvasProject(userID string, id string) error {
|
||||
return s.canvasDomain().DeleteUserCanvasProject(userID, id)
|
||||
}
|
||||
|
||||
@@ -67,7 +67,7 @@ func TestCanvasHistoryProtectsMediaAndDeletionWorker(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestArchivedAssetDeletionRemovesCanvasHistoryReference(t *testing.T) {
|
||||
func TestArchivedAssetDeletionPreservesCanvasHistoryReference(t *testing.T) {
|
||||
svc, db, _ := newResourceDeletionTestService(t)
|
||||
resource := model.Resource{
|
||||
ID: "archived-history-resource", UserID: "user-1", Provider: "unsupported-test-provider",
|
||||
@@ -88,18 +88,18 @@ func TestArchivedAssetDeletionRemovesCanvasHistoryReference(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
if err := svc.DeleteUserAsset("user-1", asset.ID); err != nil {
|
||||
t.Fatalf("archived asset deletion failed: %v", err)
|
||||
if err := svc.DeleteUserAsset("user-1", asset.ID); err == nil || !strings.Contains(err.Error(), "画布历史版本") {
|
||||
t.Fatalf("archived asset bypassed history protection: %v", err)
|
||||
}
|
||||
for _, check := range []struct {
|
||||
model any
|
||||
want int64
|
||||
}{
|
||||
{&model.Asset{}, 0},
|
||||
{&model.Resource{}, 0},
|
||||
{&model.CanvasSnapshotResource{}, 0},
|
||||
{&model.Asset{}, 1},
|
||||
{&model.Resource{}, 1},
|
||||
{&model.CanvasSnapshotResource{}, 1},
|
||||
{&model.CanvasSnapshot{}, 1},
|
||||
{&model.ResourceDeletionJob{}, 1},
|
||||
{&model.ResourceDeletionJob{}, 0},
|
||||
} {
|
||||
var count int64
|
||||
if err := db.Model(check.model).Count(&count).Error; err != nil {
|
||||
|
||||
@@ -36,11 +36,7 @@ func (s *Service) ResourceAccessBatch(userID string, requests []ResourceAccessRe
|
||||
results := make([]ResourceAccessResult, 0, len(requests))
|
||||
for _, request := range requests {
|
||||
item := ResourceAccessResult{ResourceID: request.ResourceID}
|
||||
// The caller has already been authenticated and each item is resolved through
|
||||
// ResourceForUser below. Provider input is therefore safe to expose here and
|
||||
// keeps browser-side model adapters on the same access contract as every
|
||||
// other resource consumer.
|
||||
options, err := assets.NormalizeAccessOptions(request.AccessOptions, true)
|
||||
options, err := assets.NormalizeAccessOptions(request.AccessOptions, false)
|
||||
if err == nil {
|
||||
var resource *model.Resource
|
||||
resource, err = s.repo.ResourceForUser(userID, request.ResourceID)
|
||||
|
||||
@@ -43,13 +43,10 @@ func (s *Service) deleteUserAssetsWithResources(userID string, ids []string, pur
|
||||
if len(assetRecords) != len(assetIDs) {
|
||||
return NotFound("素材不存在或无权删除")
|
||||
}
|
||||
deleteReferencedResources := purge
|
||||
if !purge {
|
||||
// 内部普通删除仍为单素材保护路径;用户显式删除统一走 purge。
|
||||
if len(assetRecords) != 1 {
|
||||
return BadAuthRequest("普通删除仅支持单个素材")
|
||||
}
|
||||
deleteReferencedResources = assetRecords[0].Status == model.AssetVersionStatusArchived
|
||||
}
|
||||
versions, representations, err := s.repo.AssetsResourceRecords(assetIDs)
|
||||
if err != nil {
|
||||
@@ -89,8 +86,8 @@ func (s *Service) deleteUserAssetsWithResources(userID string, ids []string, pur
|
||||
}
|
||||
|
||||
usages := make([]resourceUsage, 0)
|
||||
if !deleteReferencedResources {
|
||||
assetReferences, referenceErr := s.repo.AssetBusinessReferences(userID, assetIDs[0])
|
||||
for _, assetID := range assetIDs {
|
||||
assetReferences, referenceErr := s.repo.AssetBusinessReferences(userID, assetID)
|
||||
if referenceErr != nil {
|
||||
return referenceErr
|
||||
}
|
||||
@@ -99,13 +96,7 @@ func (s *Service) deleteUserAssetsWithResources(userID string, ids []string, pur
|
||||
}
|
||||
}
|
||||
if len(ownedIDs) > 0 {
|
||||
var snapshot repository.ResourceReferenceSnapshot
|
||||
var snapshotErr error
|
||||
if deleteReferencedResources {
|
||||
snapshot, snapshotErr = s.repo.OtherAssetResourceReferences(userID, assetIDs)
|
||||
} else {
|
||||
snapshot, snapshotErr = s.repo.ResourceReferenceSnapshot(userID, assetIDs[0], ownedIDs)
|
||||
}
|
||||
snapshot, snapshotErr := s.repo.ResourceReferenceSnapshotExcludingAssets(userID, assetIDs, ownedIDs)
|
||||
if snapshotErr != nil {
|
||||
return snapshotErr
|
||||
}
|
||||
@@ -116,9 +107,7 @@ func (s *Service) deleteUserAssetsWithResources(userID string, ids []string, pur
|
||||
sharedAssetResourceIDs[reference.ResourceID] = struct{}{}
|
||||
continue
|
||||
}
|
||||
if !deleteReferencedResources {
|
||||
usages = append(usages, resourceUsage{Kind: reference.Kind, ID: reference.ID, Title: reference.Title})
|
||||
}
|
||||
usages = append(usages, resourceUsage{Kind: reference.Kind, ID: reference.ID, Title: reference.Title})
|
||||
}
|
||||
}
|
||||
for _, document := range snapshot.Documents {
|
||||
@@ -144,9 +133,7 @@ func (s *Service) deleteUserAssetsWithResources(userID string, ids []string, pur
|
||||
}
|
||||
continue
|
||||
}
|
||||
if !deleteReferencedResources {
|
||||
usages = append(usages, resourceUsage{Kind: document.Kind, ID: document.ID, Title: document.Title})
|
||||
}
|
||||
usages = append(usages, resourceUsage{Kind: document.Kind, ID: document.ID, Title: document.Title})
|
||||
}
|
||||
}
|
||||
if len(sharedAssetResourceIDs) > 0 {
|
||||
@@ -179,7 +166,7 @@ func (s *Service) deleteUserAssetsWithResources(userID string, ids []string, pur
|
||||
deletionJobs := resourceDeletionJobs(userID, physicalObjects)
|
||||
// 业务记录和 Outbox 必须在同一事务提交。事务失败时物理文件完全不动;
|
||||
// 提交成功后由幂等 worker 清理,进程退出或对象存储暂时失败都可继续重试。
|
||||
if err := s.repo.DeleteAssetsAndResources(userID, assetIDs, ownedIDs, deletionJobs, deleteReferencedResources); err != nil {
|
||||
if err := s.repo.DeleteAssetsAndResources(userID, assetIDs, ownedIDs, deletionJobs, false); err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return NotFound("素材不存在或无权删除")
|
||||
}
|
||||
|
||||
@@ -298,7 +298,7 @@ func TestDeleteGeneratedAssetTaskReferences(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPurgeConfirmedAssetIgnoresTaskReferences(t *testing.T) {
|
||||
func TestPurgeConfirmedAssetPreservesTaskReferences(t *testing.T) {
|
||||
svc, db, _ := newResourceDeletionTestService(t)
|
||||
payload := `{"url":"/api/resources/purge-confirmed-resource/file"}`
|
||||
resource := model.Resource{
|
||||
@@ -319,16 +319,16 @@ func TestPurgeConfirmedAssetIgnoresTaskReferences(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
if err := svc.PurgeUserAsset("user-1", asset.ID); err != nil {
|
||||
t.Fatalf("PurgeUserAsset() error = %v", err)
|
||||
if err := svc.PurgeUserAsset("user-1", asset.ID); err == nil || !strings.Contains(err.Error(), "任务") {
|
||||
t.Fatalf("PurgeUserAsset() ignored active task: %v", err)
|
||||
}
|
||||
for _, check := range []struct {
|
||||
model any
|
||||
want int64
|
||||
}{
|
||||
{&model.Asset{}, 0},
|
||||
{&model.Resource{}, 0},
|
||||
{&model.ResourceDeletionJob{}, 1},
|
||||
{&model.Asset{}, 1},
|
||||
{&model.Resource{}, 1},
|
||||
{&model.ResourceDeletionJob{}, 0},
|
||||
{&model.Task{}, 1},
|
||||
} {
|
||||
var count int64
|
||||
@@ -373,7 +373,7 @@ func TestResourceDeletionWorkerRemovesObjectAndCompletesOutbox(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestExpiredArchivedAssetCleanupDeletesReferencedAssetAndUsesDeletionOutbox(t *testing.T) {
|
||||
func TestExpiredArchivedAssetCleanupPreservesReferencedAsset(t *testing.T) {
|
||||
svc, db, _ := newResourceDeletionTestService(t)
|
||||
old := time.Now().Add(-45 * 24 * time.Hour)
|
||||
resource := model.Resource{
|
||||
@@ -406,8 +406,8 @@ func TestExpiredArchivedAssetCleanupDeletesReferencedAssetAndUsesDeletionOutbox(
|
||||
if err := db.Model(&model.ResourceDeletionJob{}).Where("resource_id = ?", resource.ID).Count(&jobCount).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if assetCount != 0 || resourceCount != 0 || jobCount != 1 {
|
||||
t.Fatalf("referenced archived asset was not force deleted through outbox: asset=%d resource=%d jobs=%d", assetCount, resourceCount, jobCount)
|
||||
if assetCount != 1 || resourceCount != 1 || jobCount != 0 {
|
||||
t.Fatalf("referenced archived asset was removed: asset=%d resource=%d jobs=%d", assetCount, resourceCount, jobCount)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -2,58 +2,90 @@ package assets
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// DocumentResourceReference describes one durable resource locator in a JSON document.
|
||||
// Path is intentionally human-readable so it can be shown in save diagnostics.
|
||||
type DocumentResourceReference struct {
|
||||
ResourceID string `json:"resourceId"`
|
||||
Path string `json:"path"`
|
||||
NodeID string `json:"nodeId,omitempty"`
|
||||
ReferenceType string `json:"referenceType"`
|
||||
Source string `json:"source,omitempty"`
|
||||
}
|
||||
|
||||
func DocumentReferences(raw string, resourceIDs map[string]struct{}) bool {
|
||||
return len(DocumentReferencedIDs(raw, resourceIDs)) > 0
|
||||
}
|
||||
|
||||
func DocumentReferencedIDs(raw string, resourceIDs map[string]struct{}) map[string]struct{} {
|
||||
matched := map[string]struct{}{}
|
||||
raw = strings.TrimSpace(raw)
|
||||
if raw == "" || len(resourceIDs) == 0 {
|
||||
if len(resourceIDs) == 0 {
|
||||
return matched
|
||||
}
|
||||
found := map[string]struct{}{}
|
||||
var value any
|
||||
if err := json.Unmarshal([]byte(raw), &value); err == nil {
|
||||
if scalar, ok := value.(string); ok {
|
||||
if resourceID := ResourceID(scalar); resourceID != "" {
|
||||
found[resourceID] = struct{}{}
|
||||
references, err := CollectDocumentResourceReferences(raw)
|
||||
if err != nil {
|
||||
// A few read models store a URL directly rather than a JSON string.
|
||||
if id := ResourceID(strings.TrimSpace(raw)); id != "" {
|
||||
if _, exists := resourceIDs[id]; exists {
|
||||
matched[id] = struct{}{}
|
||||
}
|
||||
} else {
|
||||
walkReferenceDocument(value, "", found)
|
||||
}
|
||||
} else if resourceID := ResourceID(raw); resourceID != "" {
|
||||
found[resourceID] = struct{}{}
|
||||
return matched
|
||||
}
|
||||
for resourceID := range found {
|
||||
if _, exists := resourceIDs[resourceID]; exists {
|
||||
matched[resourceID] = struct{}{}
|
||||
for _, reference := range references {
|
||||
if _, exists := resourceIDs[reference.ResourceID]; exists {
|
||||
matched[reference.ResourceID] = struct{}{}
|
||||
}
|
||||
}
|
||||
return matched
|
||||
}
|
||||
|
||||
// CollectOwnedDocumentReferences is kept as the compatibility API used by the
|
||||
// history and deletion paths. The actual traversal lives in one collector so
|
||||
// nested fields such as metadata.videoPreview.storageKey cannot be skipped by
|
||||
// one of the callers.
|
||||
func CollectOwnedDocumentReferences(raw string, resourceIDs map[string]struct{}) error {
|
||||
references, err := CollectDocumentResourceReferences(raw)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, reference := range references {
|
||||
resourceIDs[reference.ResourceID] = struct{}{}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func CollectDocumentResourceReferences(raw string) ([]DocumentResourceReference, error) {
|
||||
raw = strings.TrimSpace(raw)
|
||||
if raw == "" {
|
||||
return nil
|
||||
return nil, nil
|
||||
}
|
||||
var value any
|
||||
if err := json.Unmarshal([]byte(raw), &value); err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
if scalar, ok := value.(string); ok {
|
||||
if resourceID := ResourceID(scalar); resourceID != "" {
|
||||
resourceIDs[resourceID] = struct{}{}
|
||||
return []DocumentResourceReference{{ResourceID: resourceID, Path: "$", ReferenceType: "value"}}, nil
|
||||
}
|
||||
return nil
|
||||
return nil, nil
|
||||
}
|
||||
walkReferenceDocument(value, "", resourceIDs)
|
||||
return nil
|
||||
references := make([]DocumentResourceReference, 0)
|
||||
walkReferenceDocument(value, "", "", "", &references)
|
||||
sort.Slice(references, func(i, j int) bool {
|
||||
if references[i].Path != references[j].Path {
|
||||
return references[i].Path < references[j].Path
|
||||
}
|
||||
if references[i].ResourceID != references[j].ResourceID {
|
||||
return references[i].ResourceID < references[j].ResourceID
|
||||
}
|
||||
return references[i].ReferenceType < references[j].ReferenceType
|
||||
})
|
||||
return references, nil
|
||||
}
|
||||
|
||||
func SortedIDs(values map[string]struct{}) []string {
|
||||
@@ -65,30 +97,75 @@ func SortedIDs(values map[string]struct{}) []string {
|
||||
return result
|
||||
}
|
||||
|
||||
func walkReferenceDocument(value any, parentKey string, resourceIDs map[string]struct{}) {
|
||||
func walkReferenceDocument(value any, parentKey, path, nodeID string, references *[]DocumentResourceReference) {
|
||||
switch item := value.(type) {
|
||||
case map[string]any:
|
||||
for key, child := range item {
|
||||
walkReferenceDocument(child, key, resourceIDs)
|
||||
currentNodeID := nodeID
|
||||
if candidate, ok := item["id"].(string); ok && parentKey == "nodes" {
|
||||
currentNodeID = strings.TrimSpace(candidate)
|
||||
}
|
||||
keys := make([]string, 0, len(item))
|
||||
for key := range item {
|
||||
keys = append(keys, key)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
for _, key := range keys {
|
||||
child := item[key]
|
||||
childPath := joinReferencePath(path, key)
|
||||
if text, ok := child.(string); ok {
|
||||
if resourceID := resourceIDForField(key, text); resourceID != "" {
|
||||
*references = append(*references, DocumentResourceReference{
|
||||
ResourceID: resourceID, Path: childPath, NodeID: currentNodeID, ReferenceType: key,
|
||||
})
|
||||
}
|
||||
continue
|
||||
}
|
||||
walkReferenceDocument(child, key, childPath, currentNodeID, references)
|
||||
}
|
||||
case []any:
|
||||
for _, child := range item {
|
||||
walkReferenceDocument(child, parentKey, resourceIDs)
|
||||
for index, child := range item {
|
||||
childPath := arrayReferencePath(path, index, child)
|
||||
walkReferenceDocument(child, parentKey, childPath, nodeID, references)
|
||||
}
|
||||
case string:
|
||||
if isResourceLocatorField(parentKey) {
|
||||
if resourceID := ResourceID(item); resourceID != "" {
|
||||
resourceIDs[resourceID] = struct{}{}
|
||||
}
|
||||
}
|
||||
if isBareResourceIDField(parentKey) {
|
||||
if resourceID := ValidID(item); resourceID != "" {
|
||||
resourceIDs[resourceID] = struct{}{}
|
||||
}
|
||||
if resourceID := resourceIDForField(parentKey, item); resourceID != "" {
|
||||
*references = append(*references, DocumentResourceReference{
|
||||
ResourceID: resourceID, Path: path, NodeID: nodeID, ReferenceType: parentKey,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func resourceIDForField(field, value string) string {
|
||||
if isResourceLocatorField(field) {
|
||||
return ResourceID(value)
|
||||
}
|
||||
if isBareResourceIDField(field) {
|
||||
return ValidID(value)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func joinReferencePath(parent, child string) string {
|
||||
if parent == "" {
|
||||
return child
|
||||
}
|
||||
return parent + "." + child
|
||||
}
|
||||
|
||||
func arrayReferencePath(parent string, index int, child any) string {
|
||||
if item, ok := child.(map[string]any); ok {
|
||||
if id, ok := item["id"].(string); ok && isCanvasCollectionPath(parent) && strings.TrimSpace(id) != "" {
|
||||
return fmt.Sprintf("%s[%s]", parent, strings.TrimSpace(id))
|
||||
}
|
||||
}
|
||||
return fmt.Sprintf("%s[%d]", parent, index)
|
||||
}
|
||||
|
||||
func isCanvasCollectionPath(path string) bool {
|
||||
return path == "nodes" || strings.HasSuffix(path, ".clips") || path == "clips"
|
||||
}
|
||||
|
||||
func isBareResourceIDField(field string) bool {
|
||||
switch field {
|
||||
case "resourceId", "resourceIds", "sampleResourceId", "referenceResourceId", "referenceResourceIds":
|
||||
|
||||
38
backend/internal/assets/references_test.go
Normal file
38
backend/internal/assets/references_test.go
Normal file
@@ -0,0 +1,38 @@
|
||||
package assets
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestCollectDocumentResourceReferencesIncludesNestedVideoPreview(t *testing.T) {
|
||||
raw := `{"nodes":[{"id":"node-video","metadata":{"videoPreview":{"storageKey":"resource:poster-1","source":{"id":"nested-not-node","url":"/api/resources/poster-2/file"}}}}]}`
|
||||
refs, err := CollectDocumentResourceReferences(raw)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got := map[string]DocumentResourceReference{}
|
||||
for _, ref := range refs {
|
||||
got[ref.ResourceID] = ref
|
||||
}
|
||||
if ref := got["poster-1"]; ref.Path != "nodes[node-video].metadata.videoPreview.storageKey" || ref.NodeID != "node-video" || ref.ReferenceType != "storageKey" {
|
||||
t.Fatalf("nested storageKey reference = %#v", ref)
|
||||
}
|
||||
if ref := got["poster-2"]; ref.Path != "nodes[node-video].metadata.videoPreview.source.url" || ref.NodeID != "node-video" {
|
||||
t.Fatalf("nested URL reference = %#v", ref)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCollectDocumentResourceReferencesUsesExplicitFieldsOnly(t *testing.T) {
|
||||
raw := `{"nodes":[{"id":"node-1","metadata":{"label":"resource:not-a-reference","videoPreview":{"resourceId":"poster-3"}}}],"log":"/api/resources/log-1/file"}`
|
||||
refs, err := CollectDocumentResourceReferences(raw)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(refs) != 1 || refs[0].ResourceID != "poster-3" || refs[0].NodeID != "node-1" || refs[0].ReferenceType != "resourceId" {
|
||||
t.Fatalf("references = %#v", refs)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCollectDocumentResourceReferencesReportsInvalidJSON(t *testing.T) {
|
||||
if _, err := CollectDocumentResourceReferences(`{"nodes":`); err == nil {
|
||||
t.Fatal("expected malformed document to be rejected")
|
||||
}
|
||||
}
|
||||
@@ -4,14 +4,19 @@ import (
|
||||
"encoding/json"
|
||||
"infinite-canvas/backend/internal/assets"
|
||||
"infinite-canvas/backend/internal/kernel"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
)
|
||||
|
||||
type MediaAssetReference struct {
|
||||
AssetID string
|
||||
ResourceID string
|
||||
AssetID string
|
||||
ResourceID string
|
||||
NodeID string
|
||||
Path string
|
||||
ReferenceType string
|
||||
}
|
||||
|
||||
// validateCanvasMediaAssets is the final server-side invariant for canvas sync:
|
||||
@@ -23,12 +28,16 @@ func (s *Service) ValidateCanvasMediaAssets(userID string, raw json.RawMessage)
|
||||
if err != nil {
|
||||
return kernel.BadAuthRequest("画布媒体数据格式错误")
|
||||
}
|
||||
if len(references) == 0 {
|
||||
allResourceReferences, err := assets.CollectDocumentResourceReferences(string(raw))
|
||||
if err != nil {
|
||||
return kernel.BadAuthRequest("画布媒体数据格式错误")
|
||||
}
|
||||
if len(references) == 0 && len(allResourceReferences) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
assetIDSet := make(map[string]struct{}, len(references))
|
||||
resourceIDSet := make(map[string]struct{}, len(references))
|
||||
resourceIDSet := make(map[string]struct{}, len(references)+len(allResourceReferences))
|
||||
for _, reference := range references {
|
||||
if reference.AssetID == "" {
|
||||
return kernel.BadAuthRequest("画布媒体尚未进入素材库,请等待同步完成后重试")
|
||||
@@ -36,6 +45,9 @@ func (s *Service) ValidateCanvasMediaAssets(userID string, raw json.RawMessage)
|
||||
assetIDSet[reference.AssetID] = struct{}{}
|
||||
resourceIDSet[reference.ResourceID] = struct{}{}
|
||||
}
|
||||
for _, reference := range allResourceReferences {
|
||||
resourceIDSet[reference.ResourceID] = struct{}{}
|
||||
}
|
||||
|
||||
ownedAssets, err := s.repo.AssetsForUserIDs(userID, assets.SortedIDs(assetIDSet))
|
||||
if err != nil {
|
||||
@@ -56,11 +68,18 @@ func (s *Service) ValidateCanvasMediaAssets(userID string, raw json.RawMessage)
|
||||
readyResources[resource.ID] = struct{}{}
|
||||
}
|
||||
}
|
||||
missing := make([]assets.DocumentResourceReference, 0)
|
||||
for _, reference := range allResourceReferences {
|
||||
if _, exists := readyResources[reference.ResourceID]; !exists {
|
||||
reference.Source = "current"
|
||||
missing = append(missing, reference)
|
||||
}
|
||||
}
|
||||
if len(missing) > 0 {
|
||||
return canvasResourcesMissingError(missing)
|
||||
}
|
||||
|
||||
for _, reference := range references {
|
||||
if _, exists := readyResources[reference.ResourceID]; !exists {
|
||||
return kernel.BadAuthRequest("画布媒体对应的云端资源不存在或尚未就绪,请重新上传")
|
||||
}
|
||||
resourceIDs, assetExists := assetResources[reference.AssetID]
|
||||
if !assetExists {
|
||||
return kernel.BadAuthRequest("画布媒体尚未进入素材库,请等待同步完成后重试")
|
||||
@@ -72,6 +91,18 @@ func (s *Service) ValidateCanvasMediaAssets(userID string, raw json.RawMessage)
|
||||
return nil
|
||||
}
|
||||
|
||||
func canvasResourcesMissingError(references []assets.DocumentResourceReference) *kernel.AppError {
|
||||
ids := make(map[string]struct{}, len(references))
|
||||
for _, reference := range references {
|
||||
ids[reference.ResourceID] = struct{}{}
|
||||
}
|
||||
err := kernel.NewAppError(http.StatusConflict, "画布引用的素材已变化,当前内容未被覆盖,请修复缺失素材后重试")
|
||||
err.Code = kernel.CodeCanvasResourcesMissing
|
||||
err.Reason = kernel.ReasonCanvasResourcesMissing
|
||||
err.Details = map[string]any{"resourceIds": assets.SortedIDs(ids), "missingResources": references}
|
||||
return err
|
||||
}
|
||||
|
||||
// validateAssetCanvasReferences prevents an Asset update from changing the
|
||||
// resource behind a canvas that already points at that Asset.
|
||||
func (s *Service) ValidateAssetCanvasReferences(userID string, asset model.Asset) error {
|
||||
@@ -131,6 +162,7 @@ func (s *Service) ValidateAssetReplacementCanvasReferences(userID string, replac
|
||||
func MediaAssetReferences(raw json.RawMessage) ([]MediaAssetReference, error) {
|
||||
var payload struct {
|
||||
Nodes []struct {
|
||||
ID string `json:"id"`
|
||||
Type string `json:"type"`
|
||||
Metadata struct {
|
||||
AssetID string `json:"assetId"`
|
||||
@@ -160,15 +192,16 @@ func MediaAssetReferences(raw json.RawMessage) ([]MediaAssetReference, error) {
|
||||
if !isCanvasMediaKind(node.Type) {
|
||||
continue
|
||||
}
|
||||
assetID := strings.TrimSpace(node.Metadata.AssetID)
|
||||
resourceID := firstCanvasResourceID(node.Metadata.StorageKey, node.Metadata.Content)
|
||||
if resourceID == "" {
|
||||
continue
|
||||
if resourceID != "" {
|
||||
references = append(references, MediaAssetReference{
|
||||
AssetID: assetID, ResourceID: resourceID, NodeID: node.ID,
|
||||
Path: "nodes[" + node.ID + "].metadata.storageKey", ReferenceType: "storageKey",
|
||||
})
|
||||
}
|
||||
references = append(references, MediaAssetReference{
|
||||
AssetID: strings.TrimSpace(node.Metadata.AssetID), ResourceID: resourceID,
|
||||
})
|
||||
}
|
||||
for _, clip := range payload.Timeline.Clips {
|
||||
for index, clip := range payload.Timeline.Clips {
|
||||
media := clip.DirectMedia
|
||||
if media == nil || !isCanvasMediaKind(media.Kind) {
|
||||
continue
|
||||
@@ -179,6 +212,7 @@ func MediaAssetReferences(raw json.RawMessage) ([]MediaAssetReference, error) {
|
||||
}
|
||||
references = append(references, MediaAssetReference{
|
||||
AssetID: strings.TrimSpace(media.AssetID), ResourceID: resourceID,
|
||||
Path: "timeline.clips[" + strconv.Itoa(index) + "].directMedia", ReferenceType: "directMedia",
|
||||
})
|
||||
}
|
||||
return references, nil
|
||||
|
||||
@@ -25,15 +25,17 @@ func SaveDocumentWithHistory(repo *repository.Repository, before *model.CanvasPr
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var restoredIDs []string
|
||||
if reason == "before_restore" {
|
||||
refs := map[string]struct{}{}
|
||||
if err := assets.CollectOwnedDocumentReferences(after.PayloadJSON, refs); err != nil {
|
||||
return err
|
||||
}
|
||||
restoredIDs = assets.SortedIDs(refs)
|
||||
refs := map[string]struct{}{}
|
||||
if err := assets.CollectOwnedDocumentReferences(after.PayloadJSON, refs); err != nil {
|
||||
return err
|
||||
}
|
||||
return repo.SaveCanvasWithSnapshot(after, snapshot, resourceIDs, restoredIDs, after.UpdatedAt.Add(-canvasHistoryInterval), canvasHistoryLimit, reason == "before_restore")
|
||||
return repo.SaveCanvasWithSnapshot(after, snapshot, resourceIDs, assets.SortedIDs(refs), after.UpdatedAt.Add(-canvasHistoryInterval), canvasHistoryLimit, reason == "before_restore" || reason == "before_resource_repair")
|
||||
}
|
||||
|
||||
// Explicit repair keeps the damaged preimage for audit/undo, but only protects
|
||||
// resources that still exist. The incoming document remains strictly validated.
|
||||
func (s *Service) RepairUserCanvasProject(userID string, raw json.RawMessage) (UserDataSummary, error) {
|
||||
return s.upsertUserCanvasProjectWithHistory(userID, raw, "before_resource_repair")
|
||||
}
|
||||
|
||||
func canvasRevisionConflict() error {
|
||||
|
||||
132
backend/internal/canvas/repair_resources.go
Normal file
132
backend/internal/canvas/repair_resources.go
Normal file
@@ -0,0 +1,132 @@
|
||||
package canvas
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
|
||||
"infinite-canvas/backend/internal/assets"
|
||||
"infinite-canvas/backend/internal/model"
|
||||
"infinite-canvas/backend/internal/repository"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/logger"
|
||||
)
|
||||
|
||||
type CanvasResourceRepairIssue struct {
|
||||
CanvasID string `json:"canvasId"`
|
||||
References []assets.DocumentResourceReference `json:"missingResources"`
|
||||
Repaired bool `json:"repaired"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
}
|
||||
|
||||
type CanvasResourceRepairReport struct {
|
||||
Apply bool `json:"apply"`
|
||||
Scanned int `json:"scanned"`
|
||||
Repairable int `json:"repairable"`
|
||||
Repaired int `json:"repaired"`
|
||||
Unresolved int `json:"unresolved"`
|
||||
Issues []CanvasResourceRepairIssue `json:"issues"`
|
||||
}
|
||||
|
||||
// RepairCanvasResources is an explicit maintenance operation. It only removes
|
||||
// invalid derived video previews, never original media, and retains the preimage.
|
||||
// Each canvas commits independently with the normal ownership/revision checks.
|
||||
func RepairCanvasResources(db *gorm.DB, apply bool) (CanvasResourceRepairReport, error) {
|
||||
report := CanvasResourceRepairReport{Apply: apply, Issues: []CanvasResourceRepairIssue{}}
|
||||
db = db.Session(&gorm.Session{Logger: logger.Default.LogMode(logger.Silent)})
|
||||
repo := repository.New(db)
|
||||
svc := New(repo, nil)
|
||||
after := ""
|
||||
for {
|
||||
var rows []model.CanvasProject
|
||||
if err := db.Where("id > ?", after).Order("id ASC").Limit(100).Find(&rows).Error; err != nil {
|
||||
return report, errors.New("读取画布失败;此前已提交的修复保留,请按报告重试")
|
||||
}
|
||||
if len(rows) == 0 {
|
||||
return report, nil
|
||||
}
|
||||
for _, project := range rows {
|
||||
after = project.ID
|
||||
report.Scanned++
|
||||
issue := CanvasResourceRepairIssue{CanvasID: project.ID}
|
||||
raw, err := canvasProjectPayload(project)
|
||||
if err != nil {
|
||||
issue.Reason = "画布 JSON 无效,需人工核查"
|
||||
report.Issues = append(report.Issues, issue)
|
||||
report.Unresolved++
|
||||
continue
|
||||
}
|
||||
refs, err := assets.CollectDocumentResourceReferences(string(raw))
|
||||
if err != nil {
|
||||
return report, errors.New("收集资源引用失败")
|
||||
}
|
||||
ids := map[string]struct{}{}
|
||||
for _, ref := range refs {
|
||||
ids[ref.ResourceID] = struct{}{}
|
||||
}
|
||||
resources, err := repo.ResourcesForUserIDs(project.UserID, assets.SortedIDs(ids))
|
||||
if err != nil {
|
||||
return report, errors.New("读取资源状态失败;未把查询失败视为缺失")
|
||||
}
|
||||
for _, resource := range resources {
|
||||
if resource.Status == model.ResourceStatusReady {
|
||||
delete(ids, resource.ID)
|
||||
}
|
||||
}
|
||||
if len(ids) == 0 {
|
||||
continue
|
||||
}
|
||||
for _, ref := range refs {
|
||||
if _, missing := ids[ref.ResourceID]; missing {
|
||||
ref.Source = "current"
|
||||
issue.References = append(issue.References, ref)
|
||||
}
|
||||
}
|
||||
var payload map[string]json.RawMessage
|
||||
var nodes []map[string]json.RawMessage
|
||||
if json.Unmarshal(raw, &payload) != nil || json.Unmarshal(payload["nodes"], &nodes) != nil {
|
||||
issue.Reason = "节点数据无效,需人工核查"
|
||||
} else {
|
||||
for _, node := range nodes {
|
||||
var metadata map[string]json.RawMessage
|
||||
if json.Unmarshal(node["metadata"], &metadata) != nil || metadata["videoPreview"] == nil {
|
||||
continue
|
||||
}
|
||||
previewRefs, parseErr := assets.CollectDocumentResourceReferences(string(metadata["videoPreview"]))
|
||||
if parseErr != nil {
|
||||
continue
|
||||
}
|
||||
for _, ref := range previewRefs {
|
||||
if _, missing := ids[ref.ResourceID]; missing {
|
||||
delete(metadata, "videoPreview")
|
||||
node["metadata"], _ = json.Marshal(metadata)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
payload["nodes"], _ = json.Marshal(nodes)
|
||||
repaired, marshalErr := json.Marshal(payload)
|
||||
if marshalErr != nil {
|
||||
return report, marshalErr
|
||||
}
|
||||
if err := svc.ValidateCanvasMediaAssets(project.UserID, repaired); err != nil {
|
||||
issue.Reason = "仍有主媒体缺失或素材绑定错误;未修改画布,请恢复有效历史或重新上传"
|
||||
} else {
|
||||
report.Repairable++
|
||||
if apply {
|
||||
if _, err := svc.RepairUserCanvasProject(project.UserID, repaired); err != nil {
|
||||
issue.Reason = "修复保存失败(版本冲突或资源再次变化);未覆盖,请重新扫描"
|
||||
} else {
|
||||
issue.Repaired = true
|
||||
report.Repaired++
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if issue.Reason != "" {
|
||||
report.Unresolved++
|
||||
}
|
||||
report.Issues = append(report.Issues, issue)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -293,8 +293,9 @@ func (s *Service) upsertUserCanvasProjectWithHistory(userID string, raw json.Raw
|
||||
if errors.Is(err, repository.ErrCanvasRevisionConflict) {
|
||||
return canvasRevisionConflict()
|
||||
}
|
||||
if errors.Is(err, repository.ErrCanvasHistoryResourceMissing) {
|
||||
return kernel.NewAppError(http.StatusConflict, "画布引用的素材已变化,当前内容未被覆盖,请保留草稿并重新加载")
|
||||
var missing *repository.CanvasHistoryResourceMissingError
|
||||
if errors.As(err, &missing) {
|
||||
return canvasResourcesMissingError(missing.References)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -67,7 +67,14 @@ func writeAppError(c *gin.Context, appErr *service.AppError) {
|
||||
}
|
||||
logHandlerError(c, appErr.Status, diagnosticErr)
|
||||
}
|
||||
writeFailure(c, appErr.Status, code, reason, message)
|
||||
body := gin.H{"code": code, "data": nil, "msg": message}
|
||||
if reason != "" {
|
||||
body["reason"] = reason
|
||||
}
|
||||
if len(appErr.Details) > 0 {
|
||||
body["details"] = appErr.Details
|
||||
}
|
||||
c.JSON(appErr.Status, body)
|
||||
}
|
||||
|
||||
// failInternal 保留真实 HTTP 状态,但绝不把未分类错误原文写入响应。
|
||||
|
||||
@@ -606,7 +606,8 @@ func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) {
|
||||
}
|
||||
c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 5<<20)
|
||||
var req struct {
|
||||
Project json.RawMessage `json:"project"`
|
||||
Project json.RawMessage `json:"project"`
|
||||
RepairMissingResources bool `json:"repairMissingResources"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
fail(c, http.StatusBadRequest, err)
|
||||
@@ -629,7 +630,12 @@ func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) {
|
||||
if audit.Revision != nil {
|
||||
baseRevision = *audit.Revision
|
||||
}
|
||||
project, err := svc.UpsertUserCanvasProject(user.ID, req.Project)
|
||||
var project service.UserDataSummary
|
||||
if req.RepairMissingResources {
|
||||
project, err = svc.RepairUserCanvasProject(user.ID, req.Project)
|
||||
} else {
|
||||
project, err = svc.UpsertUserCanvasProject(user.ID, req.Project)
|
||||
}
|
||||
defer func() {
|
||||
nodesBefore, nodesAfter := -1, len(audit.Nodes)
|
||||
if project.SaveAudit != nil {
|
||||
|
||||
@@ -16,28 +16,30 @@ const (
|
||||
CodeUnavailable = 503
|
||||
CodeTimeout = 504
|
||||
|
||||
CodeQuotaExceeded = 40301
|
||||
CodeIdempotencyConflict = 40901
|
||||
CodeRateLimited = 42901
|
||||
CodeQuotaExceeded = 40301
|
||||
CodeIdempotencyConflict = 40901
|
||||
CodeCanvasResourcesMissing = 40902
|
||||
CodeRateLimited = 42901
|
||||
)
|
||||
|
||||
// ErrorReason 是稳定机器可读原因,前端应判断 reason 而不是解析 msg。
|
||||
type ErrorReason string
|
||||
|
||||
const (
|
||||
ReasonInvalidArgument ErrorReason = "invalid_argument"
|
||||
ReasonUnauthorized ErrorReason = "unauthorized"
|
||||
ReasonForbidden ErrorReason = "forbidden"
|
||||
ReasonNotFound ErrorReason = "not_found"
|
||||
ReasonConflict ErrorReason = "conflict"
|
||||
ReasonFailedPrecondition ErrorReason = "failed_precondition"
|
||||
ReasonQuotaExceeded ErrorReason = "quota_exceeded"
|
||||
ReasonRateLimited ErrorReason = "rate_limited"
|
||||
ReasonUnavailable ErrorReason = "unavailable"
|
||||
ReasonTimeout ErrorReason = "timeout"
|
||||
ReasonInternal ErrorReason = "internal"
|
||||
ReasonBadGateway ErrorReason = "bad_gateway"
|
||||
ReasonUpstreamDNSFailed ErrorReason = "upstream_dns_failed"
|
||||
ReasonInvalidArgument ErrorReason = "invalid_argument"
|
||||
ReasonUnauthorized ErrorReason = "unauthorized"
|
||||
ReasonForbidden ErrorReason = "forbidden"
|
||||
ReasonNotFound ErrorReason = "not_found"
|
||||
ReasonConflict ErrorReason = "conflict"
|
||||
ReasonCanvasResourcesMissing ErrorReason = "canvas_history_resources_missing"
|
||||
ReasonFailedPrecondition ErrorReason = "failed_precondition"
|
||||
ReasonQuotaExceeded ErrorReason = "quota_exceeded"
|
||||
ReasonRateLimited ErrorReason = "rate_limited"
|
||||
ReasonUnavailable ErrorReason = "unavailable"
|
||||
ReasonTimeout ErrorReason = "timeout"
|
||||
ReasonInternal ErrorReason = "internal"
|
||||
ReasonBadGateway ErrorReason = "bad_gateway"
|
||||
ReasonUpstreamDNSFailed ErrorReason = "upstream_dns_failed"
|
||||
)
|
||||
|
||||
func ReasonForStatus(status int) ErrorReason {
|
||||
|
||||
@@ -9,6 +9,7 @@ type AppError struct {
|
||||
Message string
|
||||
Retryable bool
|
||||
Cause error
|
||||
Details map[string]any
|
||||
}
|
||||
|
||||
func (e *AppError) Error() string {
|
||||
|
||||
@@ -2,9 +2,11 @@ package repository
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
"infinite-canvas/backend/internal/assets"
|
||||
"infinite-canvas/backend/internal/model"
|
||||
|
||||
"gorm.io/gorm"
|
||||
@@ -14,6 +16,19 @@ import (
|
||||
var ErrCanvasHistoryResourceMissing = errors.New("canvas history resource missing")
|
||||
var ErrCanvasHistoryResourceReferenced = errors.New("resource referenced by canvas history")
|
||||
|
||||
type CanvasHistoryResourceMissingError struct {
|
||||
ResourceIDs []string
|
||||
References []assets.DocumentResourceReference
|
||||
}
|
||||
|
||||
func (e *CanvasHistoryResourceMissingError) Error() string {
|
||||
return fmt.Sprintf("canvas history resources missing: %v", e.ResourceIDs)
|
||||
}
|
||||
|
||||
func (e *CanvasHistoryResourceMissingError) Unwrap() error {
|
||||
return ErrCanvasHistoryResourceMissing
|
||||
}
|
||||
|
||||
const canvasSnapshotSummaryColumns = "id, canvas_id, user_id, revision, title, node_count, connection_count, payload_bytes, reason, content_updated_at, created_at"
|
||||
|
||||
func (r *Repository) CanvasProjectMetadata(userID, id string) (*model.CanvasProject, error) {
|
||||
@@ -37,22 +52,22 @@ func (r *Repository) CanvasSnapshot(userID, canvasID, id string) (*model.CanvasS
|
||||
|
||||
// The successful CAS takes the canvas row lock before inspecting the history.
|
||||
// Snapshot creation, reference protection and retention commit with the content.
|
||||
func (r *Repository) SaveCanvasWithSnapshot(project *model.CanvasProject, snapshot *model.CanvasSnapshot, resourceIDs, restoredResourceIDs []string, cutoff time.Time, limit int, force bool) error {
|
||||
func (r *Repository) SaveCanvasWithSnapshot(project *model.CanvasProject, snapshot *model.CanvasSnapshot, resourceIDs, currentResourceIDs []string, cutoff time.Time, limit int, force bool) error {
|
||||
next := *project
|
||||
err := r.db.Transaction(func(tx *gorm.DB) error {
|
||||
if err := New(tx).UpsertCanvasProject(&next); err != nil {
|
||||
return err
|
||||
}
|
||||
if snapshot == nil {
|
||||
return nil
|
||||
capture := false
|
||||
if snapshot != nil {
|
||||
var last model.CanvasSnapshot
|
||||
err := tx.Select("id", "revision", "created_at").Where("canvas_id = ?", project.ID).Order("revision DESC").First(&last).Error
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return err
|
||||
}
|
||||
capture = err != nil || (last.Revision != snapshot.Revision && (force || !last.CreatedAt.After(cutoff)))
|
||||
}
|
||||
var last model.CanvasSnapshot
|
||||
err := tx.Select("id", "revision", "created_at").Where("canvas_id = ?", project.ID).Order("revision DESC").First(&last).Error
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return err
|
||||
}
|
||||
capture := err != nil || (last.Revision != snapshot.Revision && (force || !last.CreatedAt.After(cutoff)))
|
||||
requiredIDs := append([]string{}, restoredResourceIDs...)
|
||||
requiredIDs := append([]string{}, currentResourceIDs...)
|
||||
if capture {
|
||||
requiredIDs = append(requiredIDs, resourceIDs...)
|
||||
}
|
||||
@@ -60,15 +75,61 @@ func (r *Repository) SaveCanvasWithSnapshot(project *model.CanvasProject, snapsh
|
||||
requiredIDs = slices.Compact(requiredIDs)
|
||||
if len(requiredIDs) > 0 {
|
||||
var resources []model.Resource
|
||||
query := tx.Select("id").Where("id IN ? AND status = ?", requiredIDs, model.ResourceStatusReady).Order("id")
|
||||
query := tx.Select("id").Where("id IN ? AND user_id = ? AND status = ?", requiredIDs, project.UserID, model.ResourceStatusReady).Order("id")
|
||||
if r.Dialect() == "postgres" {
|
||||
query = query.Clauses(clause.Locking{Strength: "UPDATE"})
|
||||
}
|
||||
if err := query.Find(&resources).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if len(resources) != len(requiredIDs) {
|
||||
return ErrCanvasHistoryResourceMissing
|
||||
found := make(map[string]struct{}, len(resources))
|
||||
for _, resource := range resources {
|
||||
found[resource.ID] = struct{}{}
|
||||
}
|
||||
missingIDs := make([]string, 0)
|
||||
allowDamagedPreimage := snapshot != nil && (snapshot.Reason == "before_resource_repair" || snapshot.Reason == "before_restore")
|
||||
for _, id := range requiredIDs {
|
||||
if _, exists := found[id]; !exists {
|
||||
if allowDamagedPreimage && !slices.Contains(currentResourceIDs, id) {
|
||||
continue
|
||||
}
|
||||
missingIDs = append(missingIDs, id)
|
||||
}
|
||||
}
|
||||
if len(missingIDs) > 0 {
|
||||
missingSet := make(map[string]struct{}, len(missingIDs))
|
||||
for _, id := range missingIDs {
|
||||
missingSet[id] = struct{}{}
|
||||
}
|
||||
missingRefs := make([]assets.DocumentResourceReference, 0)
|
||||
documents := map[string]string{"current": project.PayloadJSON}
|
||||
if capture {
|
||||
documents["history"] = snapshot.PayloadJSON
|
||||
}
|
||||
for _, source := range []string{"current", "history"} {
|
||||
refs, collectErr := assets.CollectDocumentResourceReferences(documents[source])
|
||||
if collectErr != nil {
|
||||
return collectErr
|
||||
}
|
||||
for _, ref := range refs {
|
||||
if _, missing := missingSet[ref.ResourceID]; missing {
|
||||
ref.Source = source
|
||||
missingRefs = append(missingRefs, ref)
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, id := range missingIDs {
|
||||
if !containsResourceReference(missingRefs, id) {
|
||||
missingRefs = append(missingRefs, assets.DocumentResourceReference{ResourceID: id, Path: "history", ReferenceType: "snapshot"})
|
||||
}
|
||||
}
|
||||
return &CanvasHistoryResourceMissingError{ResourceIDs: missingIDs, References: missingRefs}
|
||||
}
|
||||
if allowDamagedPreimage {
|
||||
resourceIDs = slices.DeleteFunc(slices.Clone(resourceIDs), func(id string) bool {
|
||||
_, exists := found[id]
|
||||
return !exists
|
||||
})
|
||||
}
|
||||
}
|
||||
if !capture {
|
||||
@@ -107,6 +168,15 @@ func (r *Repository) CanvasHistoryReferencesObject(resource *model.Resource) (bo
|
||||
return count > 0, err
|
||||
}
|
||||
|
||||
func containsResourceReference(references []assets.DocumentResourceReference, resourceID string) bool {
|
||||
for _, reference := range references {
|
||||
if reference.ResourceID == resourceID {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func deleteCanvasSnapshots(tx *gorm.DB, ids []string) error {
|
||||
if len(ids) == 0 {
|
||||
return nil
|
||||
|
||||
@@ -119,6 +119,10 @@ func (r *Repository) ResourceStorageReferenceCount(resource *model.Resource, exc
|
||||
}
|
||||
|
||||
func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID string, resourceIDs []string) (ResourceReferenceSnapshot, error) {
|
||||
return r.ResourceReferenceSnapshotExcludingAssets(userID, []string{excludingAssetID}, resourceIDs)
|
||||
}
|
||||
|
||||
func (r *Repository) ResourceReferenceSnapshotExcludingAssets(userID string, excludingAssetIDs []string, resourceIDs []string) (ResourceReferenceSnapshot, error) {
|
||||
snapshot := ResourceReferenceSnapshot{Documents: []ResourceReferenceDocument{}, Direct: []ResourceDirectReference{}}
|
||||
if len(resourceIDs) == 0 {
|
||||
return snapshot, nil
|
||||
@@ -154,7 +158,10 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s
|
||||
snapshot.Documents = append(snapshot.Documents, ResourceReferenceDocument{Kind: "工具", ID: strconv.FormatInt(tool.ID, 10), Title: tool.Label, PrimaryJSON: string(payload)})
|
||||
}
|
||||
var assets []model.Asset
|
||||
assetQuery := r.db.Where("user_id = ? AND id <> ?", userID, excludingAssetID)
|
||||
assetQuery := r.db.Where("user_id = ?", userID)
|
||||
if len(excludingAssetIDs) > 0 {
|
||||
assetQuery = assetQuery.Where("id NOT IN ?", excludingAssetIDs)
|
||||
}
|
||||
if err := assetQuery.Find(&assets).Error; err != nil {
|
||||
return snapshot, err
|
||||
}
|
||||
@@ -240,7 +247,10 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s
|
||||
versionQuery := r.db.Table("asset_versions").
|
||||
Select("asset_versions.id, assets.title, asset_versions.definition_json AS primary_json").
|
||||
Joins("JOIN assets ON assets.id = asset_versions.asset_id").
|
||||
Where("assets.user_id = ? AND assets.id <> ?", userID, excludingAssetID)
|
||||
Where("assets.user_id = ?", userID)
|
||||
if len(excludingAssetIDs) > 0 {
|
||||
versionQuery = versionQuery.Where("assets.id NOT IN ?", excludingAssetIDs)
|
||||
}
|
||||
if err := versionQuery.Scan(&versions).Error; err != nil {
|
||||
return snapshot, err
|
||||
}
|
||||
@@ -292,12 +302,15 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s
|
||||
ResourceID string
|
||||
}
|
||||
var representations []joinedRepresentation
|
||||
if err := r.db.Table("asset_representations").
|
||||
representationQuery := r.db.Table("asset_representations").
|
||||
Select("asset_representations.id, assets.title, asset_representations.resource_id").
|
||||
Joins("JOIN asset_versions ON asset_versions.id = asset_representations.asset_version_id").
|
||||
Joins("JOIN assets ON assets.id = asset_versions.asset_id").
|
||||
Where("assets.user_id = ? AND assets.id <> ? AND asset_representations.resource_id IN ?", userID, excludingAssetID, resourceIDs).
|
||||
Scan(&representations).Error; err != nil {
|
||||
Where("assets.user_id = ? AND asset_representations.resource_id IN ?", userID, resourceIDs)
|
||||
if len(excludingAssetIDs) > 0 {
|
||||
representationQuery = representationQuery.Where("assets.id NOT IN ?", excludingAssetIDs)
|
||||
}
|
||||
if err := representationQuery.Scan(&representations).Error; err != nil {
|
||||
return snapshot, err
|
||||
}
|
||||
for _, representation := range representations {
|
||||
@@ -403,23 +416,9 @@ func (r *Repository) DeleteAssetsAndResources(userID string, assetIDs []string,
|
||||
if len(assetIDs) == 0 || len(ownedAssets) != len(assetIDs) {
|
||||
return gorm.ErrRecordNotFound
|
||||
}
|
||||
if deleteReferencedResources {
|
||||
if len(resourceIDs) > 0 {
|
||||
// 与历史快照写入串行化,避免清除索引后又插入外键引用。
|
||||
if r.Dialect() == "postgres" {
|
||||
var resources []model.Resource
|
||||
if err := tx.Select("id").Where("id IN ?", resourceIDs).Order("id").Clauses(clause.Locking{Strength: "UPDATE"}).Find(&resources).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := tx.Where("resource_id IN ?", resourceIDs).Delete(&model.CanvasSnapshotResource{}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if err := New(tx).RequireNoCanvasHistoryReferences(resourceIDs); err != nil {
|
||||
return err
|
||||
}
|
||||
// Explicit deletion and archive expiry must never invalidate a snapshot.
|
||||
if err := New(tx).RequireNoCanvasHistoryReferences(resourceIDs); err != nil {
|
||||
return err
|
||||
}
|
||||
versionIDs := tx.Model(&model.AssetVersion{}).Select("id").Where("asset_id IN ?", assetIDs)
|
||||
if err := tx.Where("asset_version_id IN (?)", versionIDs).Delete(&model.ShotAssetReference{}).Error; err != nil {
|
||||
|
||||
@@ -13,6 +13,8 @@ import { imageSizeConfigWithPresets } from "@/lib/image-size-presets";
|
||||
import { useUserStore } from "@/stores/use-user-store";
|
||||
import { PLUGIN_STORE_KEY, usePluginStore } from "@/stores/use-plugin-store";
|
||||
import { initializeRemoteUserDataSession, installRemoteUserDataAutoSync, resetRemoteUserDataSync, withRemoteUserDataSyncExclusive } from "@/services/user-data-sync";
|
||||
import { clearResourceAccessCache } from "@/services/api/resources";
|
||||
import { clearResourceBlobCache } from "@/services/resource-blob-cache";
|
||||
import { withGenerationConsumersPaused } from "@/services/generation-consumer-lifecycle";
|
||||
|
||||
export async function switchUserStorageScope(userId?: string | null) {
|
||||
@@ -20,6 +22,8 @@ export async function switchUserStorageScope(userId?: string | null) {
|
||||
await withRemoteUserDataSyncExclusive(async () => {
|
||||
await Promise.all([flushCanvasStorePersistence(), flushAssetStorePersistence()]);
|
||||
resetRemoteUserDataSync();
|
||||
clearResourceAccessCache();
|
||||
clearResourceBlobCache();
|
||||
setActiveUserScope(userId);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -7,6 +7,7 @@ export type BackendEnvelope<T> = {
|
||||
data: T;
|
||||
msg: string;
|
||||
reason?: string;
|
||||
details?: Record<string, unknown>;
|
||||
};
|
||||
|
||||
/**
|
||||
@@ -18,16 +19,18 @@ export class ApiError extends Error {
|
||||
readonly status?: number;
|
||||
readonly code?: number;
|
||||
readonly reason?: string;
|
||||
readonly details?: Record<string, unknown>;
|
||||
readonly retryable: boolean;
|
||||
readonly retryAfterMs?: number;
|
||||
readonly cause?: unknown;
|
||||
|
||||
constructor(message: string, options: { status?: number; code?: number; reason?: string; retryable?: boolean; retryAfterMs?: number; cause?: unknown } = {}) {
|
||||
constructor(message: string, options: { status?: number; code?: number; reason?: string; details?: Record<string, unknown>; retryable?: boolean; retryAfterMs?: number; cause?: unknown } = {}) {
|
||||
super(message);
|
||||
this.name = "ApiError";
|
||||
this.status = options.status;
|
||||
this.code = options.code;
|
||||
this.reason = options.reason;
|
||||
this.details = options.details;
|
||||
this.retryable = options.retryable ?? isRetryableStatus(options.status ?? options.code);
|
||||
this.retryAfterMs = options.retryAfterMs;
|
||||
this.cause = options.cause;
|
||||
@@ -51,6 +54,7 @@ export async function request<T>(promise: Promise<{ data: BackendEnvelope<T>; st
|
||||
status: response.status,
|
||||
code: response.data.code,
|
||||
reason: response.data.reason,
|
||||
details: response.data.details,
|
||||
retryable: isRetryableStatus(response.status) || isRetryableStatus(response.data.code),
|
||||
retryAfterMs: retryAfterMilliseconds(response.headers),
|
||||
});
|
||||
@@ -75,6 +79,7 @@ function unwrapTransportError(error: unknown): never {
|
||||
status,
|
||||
code,
|
||||
reason: error.response?.data?.reason,
|
||||
details: error.response?.data?.details,
|
||||
retryable: isRetryableStatus(status) || isRetryableStatus(code),
|
||||
retryAfterMs: retryAfterMilliseconds(error.response?.headers),
|
||||
cause: error,
|
||||
|
||||
@@ -18,10 +18,10 @@ export type RemoteResource = {
|
||||
height?: number;
|
||||
durationMs?: number;
|
||||
etag?: string;
|
||||
playbackStatus?: string;
|
||||
playbackObjectKey?: string;
|
||||
playbackError?: string;
|
||||
error?: string;
|
||||
playbackStatus?: string;
|
||||
playbackObjectKey?: string;
|
||||
playbackError?: string;
|
||||
error?: string;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
};
|
||||
@@ -119,6 +119,21 @@ export type ResourceAccess = {
|
||||
|
||||
const accessCache = new Map<string, { value: ResourceAccess; expiresAt: number }>();
|
||||
const accessRequests = new Map<string, Promise<ResourceAccess>>();
|
||||
let accessGeneration = 0;
|
||||
|
||||
/**
|
||||
* Drop every in-memory access descriptor when the authenticated scope changes.
|
||||
* Signed URLs are credentials, so an old request must not be allowed to publish
|
||||
* its result into the next account's cache after a logout/login race.
|
||||
*/
|
||||
export function clearResourceAccessCache() {
|
||||
accessGeneration += 1;
|
||||
accessCache.clear();
|
||||
accessRequests.clear();
|
||||
resourceCache.clear();
|
||||
resourceRequests.clear();
|
||||
missingResourceIds.clear();
|
||||
}
|
||||
|
||||
export function resourceStorageKey(id: string) {
|
||||
return `resource:${id}`;
|
||||
@@ -160,12 +175,7 @@ export function isResourceUrl(url?: string) {
|
||||
const CHUNK_UPLOAD_THRESHOLD = 50 << 20;
|
||||
const CHUNK_UPLOAD_RETRIES = 2;
|
||||
|
||||
export async function uploadResourceFile(
|
||||
file: Blob,
|
||||
kind: "image" | "video" | "audio" | "file",
|
||||
meta?: ResourceUploadMeta,
|
||||
onProgress?: (uploadedBytes: number, totalBytes: number) => void,
|
||||
): Promise<RemoteResource> {
|
||||
export async function uploadResourceFile(file: Blob, kind: "image" | "video" | "audio" | "file", meta?: ResourceUploadMeta, onProgress?: (uploadedBytes: number, totalBytes: number) => void): Promise<RemoteResource> {
|
||||
const name = meta?.fileName || (file instanceof File ? file.name : `${kind}.${extensionFromMime(file.type, kind)}`);
|
||||
// 分片与 multipart 两条路径的失败都要归一成 ResourceUploadError,
|
||||
// 否则调用方只能靠文案猜测该重试还是该报错。
|
||||
@@ -183,9 +193,11 @@ export async function uploadResourceFile(
|
||||
if (meta?.durationMs) formData.append("durationMs", String(Math.round(meta.durationMs)));
|
||||
const data = await http.post<{ resource: RemoteResource }>("/resources", formData, {
|
||||
...uploadRequestConfig(meta?.idempotencyKey),
|
||||
onUploadProgress: onProgress ? ({ loaded, total }) => {
|
||||
if (total && total > 0) onProgress(Math.min(file.size, file.size * loaded / total), file.size);
|
||||
} : undefined,
|
||||
onUploadProgress: onProgress
|
||||
? ({ loaded, total }) => {
|
||||
if (total && total > 0) onProgress(Math.min(file.size, (file.size * loaded) / total), file.size);
|
||||
}
|
||||
: undefined,
|
||||
});
|
||||
resourceCache.set(resourceCacheKey(data.resource.id), data.resource);
|
||||
return data.resource;
|
||||
@@ -209,7 +221,11 @@ async function uploadFileInChunks(file: Blob, name: string, kind: "image" | "vid
|
||||
}
|
||||
|
||||
async function runChunkedUpload(file: Blob, name: string, kind: "image" | "video" | "audio" | "file", meta: ResourceUploadMeta | undefined, onProgress?: (uploadedBytes: number, totalBytes: number) => void) {
|
||||
const session = await http.post<{ uploadId: string; chunkSize: number; chunkCount: number }>("/resources/uploads", { fileName: name, kind, size: file.size, width: meta?.width, height: meta?.height, durationMs: meta?.durationMs }, uploadRequestConfig(meta?.idempotencyKey));
|
||||
const session = await http.post<{ uploadId: string; chunkSize: number; chunkCount: number }>(
|
||||
"/resources/uploads",
|
||||
{ fileName: name, kind, size: file.size, width: meta?.width, height: meta?.height, durationMs: meta?.durationMs },
|
||||
uploadRequestConfig(meta?.idempotencyKey),
|
||||
);
|
||||
for (let index = 0; index < session.chunkCount; index++) {
|
||||
const start = index * session.chunkSize;
|
||||
const end = Math.min(file.size, start + session.chunkSize);
|
||||
@@ -260,48 +276,69 @@ function uploadRequestConfig(idempotencyKey?: string) {
|
||||
}
|
||||
|
||||
export function getResource(id: string): Promise<RemoteResource> {
|
||||
const generation = accessGeneration;
|
||||
const scope = getActiveUserScope();
|
||||
const cacheKey = resourceCacheKey(id);
|
||||
const cached = resourceCache.get(cacheKey);
|
||||
if (cached) return Promise.resolve(cached);
|
||||
if (missingResourceIds.has(cacheKey)) return Promise.reject(new Error("资源不存在或已被删除"));
|
||||
const pending = resourceRequests.get(cacheKey);
|
||||
if (pending) return pending;
|
||||
const task = http.get<{ resource: RemoteResource }>(`/resources/${encodeURIComponent(id)}`)
|
||||
const task = http
|
||||
.get<{ resource: RemoteResource }>(`/resources/${encodeURIComponent(id)}`)
|
||||
.then((data) => {
|
||||
assertResourceRequestCurrent(generation, scope);
|
||||
resourceCache.set(cacheKey, data.resource);
|
||||
return data.resource;
|
||||
})
|
||||
.catch((error) => {
|
||||
assertResourceRequestCurrent(generation, scope);
|
||||
if (error instanceof ApiError && error.status === 404) missingResourceIds.add(cacheKey);
|
||||
throw error;
|
||||
})
|
||||
.finally(() => resourceRequests.delete(cacheKey));
|
||||
.finally(() => {
|
||||
if (resourceRequests.get(cacheKey) === task) resourceRequests.delete(cacheKey);
|
||||
});
|
||||
resourceRequests.set(cacheKey, task);
|
||||
return task;
|
||||
}
|
||||
|
||||
// refreshResource 绕过缓存强制拉取资源最新状态(转码副本就绪轮询用),并回写缓存。
|
||||
export function refreshResource(id: string): Promise<RemoteResource> {
|
||||
const generation = accessGeneration;
|
||||
const scope = getActiveUserScope();
|
||||
const cacheKey = resourceCacheKey(id);
|
||||
return http.get<{ resource: RemoteResource }>(`/resources/${encodeURIComponent(id)}`)
|
||||
.then((data) => {
|
||||
resourceCache.set(cacheKey, data.resource);
|
||||
missingResourceIds.delete(cacheKey);
|
||||
return data.resource;
|
||||
});
|
||||
return http.get<{ resource: RemoteResource }>(`/resources/${encodeURIComponent(id)}`).then((data) => {
|
||||
assertResourceRequestCurrent(generation, scope);
|
||||
resourceCache.set(cacheKey, data.resource);
|
||||
missingResourceIds.delete(cacheKey);
|
||||
return data.resource;
|
||||
});
|
||||
}
|
||||
|
||||
function assertResourceRequestCurrent(generation: number, scope: string) {
|
||||
if (generation !== accessGeneration || scope !== getActiveUserScope()) {
|
||||
throw new DOMException("资源请求已因账号切换失效", "AbortError");
|
||||
}
|
||||
}
|
||||
|
||||
export async function getResourceAccess(storageKey: string | undefined, purpose: ResourceAccessPurpose = "display", variant: ResourceAccessVariant = "original", downloadName = "") {
|
||||
const id = resourceIdFromStorageKey(storageKey);
|
||||
if (!id) throw new Error("当前媒体尚未上传到后端资源存储");
|
||||
const key = `${resourceCacheKey(id)}:${purpose}:${variant}:${downloadName}`;
|
||||
const scope = getActiveUserScope();
|
||||
const generation = accessGeneration;
|
||||
const key = `${scope}:${id}:${purpose}:${variant}:${downloadName}`;
|
||||
const cached = accessCache.get(key);
|
||||
if (cached && cached.expiresAt > Date.now()) return cached.value;
|
||||
const pending = accessRequests.get(key);
|
||||
if (pending) return pending;
|
||||
const request = (async () => {
|
||||
let request!: Promise<ResourceAccess>;
|
||||
request = (async () => {
|
||||
try {
|
||||
const data = await http.post<{ items: Array<{ resourceId: string; access?: ResourceAccess; error?: { msg?: string } }> }>("/resources/access", [{ resourceId: id, purpose, variant, ...(downloadName ? { downloadName } : {}) }]);
|
||||
if (generation !== accessGeneration || scope !== getActiveUserScope()) {
|
||||
throw new DOMException("资源访问请求已因账号切换失效", "AbortError");
|
||||
}
|
||||
const item = data.items?.[0];
|
||||
if (!item?.access?.url) throw new Error(item?.error?.msg || "后端未返回资源访问地址");
|
||||
const value = item.access;
|
||||
@@ -312,7 +349,7 @@ export async function getResourceAccess(storageKey: string | undefined, purpose:
|
||||
if (error instanceof ApiError) throw new Error(error.message || "获取对象存储地址失败");
|
||||
throw error;
|
||||
} finally {
|
||||
accessRequests.delete(key);
|
||||
if (accessRequests.get(key) === request) accessRequests.delete(key);
|
||||
}
|
||||
})();
|
||||
accessRequests.set(key, request);
|
||||
|
||||
@@ -121,9 +121,9 @@ export function getRemoteCanvasProject(id: string) {
|
||||
return http.get<{ project: CanvasProject }>(`/canvas-projects/${encodeURIComponent(id)}`);
|
||||
}
|
||||
|
||||
export function upsertRemoteCanvasProject(project: CanvasProject) {
|
||||
export function upsertRemoteCanvasProject(project: CanvasProject, options?: { repairMissingResources?: boolean }) {
|
||||
const { viewport: _viewport, remoteContentHash: _hash, ...content } = project;
|
||||
return http.put<{ project: RemoteUserDataSummary & { revision: number } }>(`/canvas-projects/${encodeURIComponent(project.id)}`, { project: content });
|
||||
return http.put<{ project: RemoteUserDataSummary & { revision: number } }>(`/canvas-projects/${encodeURIComponent(project.id)}`, { project: content, ...(options?.repairMissingResources ? { repairMissingResources: true } : {}) });
|
||||
}
|
||||
|
||||
export function deleteRemoteCanvasProject(id: string) {
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
import { canvasNodeToAsset } from "@/lib/canvas/canvas-node-asset";
|
||||
import { resourceIdFromStorageKey } from "@/services/api/resources";
|
||||
import { refreshResource, resourceIdFromStorageKey } from "@/services/api/resources";
|
||||
import { ApiError } from "@/services/api/request";
|
||||
import { useAssetStore, type Asset, type NewAsset } from "@/stores/use-asset-store";
|
||||
import { useCanvasStore, type CanvasProject } from "@/stores/canvas/use-canvas-store";
|
||||
import { CanvasNodeType, type CanvasNodeData } from "@/types/canvas";
|
||||
import type { TimelineDirectMedia } from "@/types/timeline";
|
||||
import { getActiveUserScope } from "@/lib/user-scope";
|
||||
|
||||
export type CanvasAssetRepairResult = {
|
||||
createdAssets: number;
|
||||
@@ -134,6 +136,67 @@ function assetStorageKey(asset: Asset) {
|
||||
return asset.kind === "text" || asset.kind === "entity" ? undefined : asset.data.storageKey;
|
||||
}
|
||||
|
||||
export type CanvasVideoPreviewRepairResult = {
|
||||
clearedPreviews: number;
|
||||
updatedProjects: number;
|
||||
};
|
||||
|
||||
/** 仅清理确认缺失的派生封面;网络/鉴权失败必须上抛,不能当作不存在。 */
|
||||
export async function repairMissingCanvasVideoPreviews(projectIds?: Set<string>, lookup = refreshResource): Promise<CanvasVideoPreviewRepairResult> {
|
||||
const scope = getActiveUserScope();
|
||||
const projects = useCanvasStore.getState().projects.filter((project) => !projectIds || projectIds.has(project.id));
|
||||
const previewResourceIds = new Set<string>();
|
||||
for (const project of projects) {
|
||||
for (const node of project.nodes) {
|
||||
for (const id of previewResourceIDs(node)) previewResourceIds.add(id);
|
||||
}
|
||||
}
|
||||
const invalidIds = new Set<string>();
|
||||
const ids = [...previewResourceIds];
|
||||
for (let offset = 0; offset < ids.length; offset += 8) {
|
||||
await Promise.all(ids.slice(offset, offset + 8).map(async (id) => {
|
||||
try {
|
||||
const resource = await lookup(id);
|
||||
if (resource.status !== "ready") invalidIds.add(id);
|
||||
} catch (error) {
|
||||
if (error instanceof ApiError && error.status === 404) invalidIds.add(id);
|
||||
else throw error;
|
||||
}
|
||||
}));
|
||||
}
|
||||
if (getActiveUserScope() !== scope) throw new Error("账号已切换,已停止修复视频封面");
|
||||
if (!invalidIds.size) return { clearedPreviews: 0, updatedProjects: 0 };
|
||||
|
||||
let clearedPreviews = 0;
|
||||
let updatedProjects = 0;
|
||||
const canvasStore = useCanvasStore.getState();
|
||||
// Re-read live projects after I/O; never replace edits made while diagnosing.
|
||||
for (const project of canvasStore.projects.filter((item) => !projectIds || projectIds.has(item.id))) {
|
||||
let changed = false;
|
||||
const nodes = project.nodes.map((node) => {
|
||||
if (!previewResourceIDs(node).some((id) => invalidIds.has(id)) || !node.metadata?.videoPreview) return node;
|
||||
const { videoPreview: _videoPreview, ...metadata } = node.metadata;
|
||||
changed = true;
|
||||
clearedPreviews += 1;
|
||||
return { ...node, metadata };
|
||||
});
|
||||
if (!changed) continue;
|
||||
canvasStore.updateProject(project.id, { nodes });
|
||||
updatedProjects += 1;
|
||||
}
|
||||
return { clearedPreviews, updatedProjects };
|
||||
}
|
||||
|
||||
function previewResourceIDs(node: CanvasNodeData): string[] {
|
||||
const preview = node.metadata?.videoPreview;
|
||||
if (!preview) return [];
|
||||
return Object.entries(preview).flatMap(([key, value]) => {
|
||||
if (!["storageKey", "url", "dataUrl", "content"].includes(key) || typeof value !== "string") return [];
|
||||
const id = resourceIdFromStorageKey(value) || value.match(/(?:^|\/)api\/resources\/([A-Za-z0-9_-]+)\/file(?:[?#]|$)/)?.[1];
|
||||
return id ? [id] : [];
|
||||
});
|
||||
}
|
||||
|
||||
export type CanvasAssetRebindResult = {
|
||||
reboundNodes: number;
|
||||
createdAssets: number;
|
||||
|
||||
@@ -18,12 +18,12 @@ const metaStore = localforage.createInstance({ name: "infinite-canvas", storeNam
|
||||
const objectUrls = new Map<string, string>();
|
||||
const sessionBlobs = new Map<string, Blob>();
|
||||
const inFlight = new Map<string, Promise<string>>();
|
||||
const scheduled = new Set<string>();
|
||||
const cacheMetaTouchWarnings = new Set<string>();
|
||||
const metaTouchedAt = new Map<string, number>();
|
||||
const downloadQueue: Array<() => void> = [];
|
||||
let activeDownloads = 0;
|
||||
let persistQueue: Promise<void> = Promise.resolve();
|
||||
let cacheGeneration = 0;
|
||||
const MAX_CACHE_BYTES = 2 * 1024 * 1024 * 1024;
|
||||
const FALLBACK_CACHE_BYTES = 512 * 1024 * 1024;
|
||||
const MIN_CACHE_BYTES = 64 * 1024 * 1024;
|
||||
@@ -33,6 +33,18 @@ const BUDGET_REFRESH_MS = 5 * 60 * 1000;
|
||||
// 现代浏览器对同源 HTTP/2 连接多路复用;上限给到 16 让大画布冷启动在 1~2 轮内完成并发拉取。
|
||||
const MAX_CONCURRENT_DOWNLOADS = 16;
|
||||
|
||||
/** Release session-only byte resources during account/logout transitions. */
|
||||
export function clearResourceBlobCache() {
|
||||
cacheGeneration += 1;
|
||||
objectUrls.forEach((url) => URL.revokeObjectURL(url));
|
||||
objectUrls.clear();
|
||||
sessionBlobs.clear();
|
||||
inFlight.clear();
|
||||
metaTouchedAt.clear();
|
||||
cacheMetaTouchWarnings.clear();
|
||||
cacheStats = null;
|
||||
}
|
||||
|
||||
export async function getCachedResourceObjectUrl(storageKey: string) {
|
||||
const target = await cacheTarget(storageKey);
|
||||
if (!target) return "";
|
||||
@@ -56,42 +68,25 @@ export async function cacheResourceObjectUrl(storageKey: string) {
|
||||
const pending = inFlight.get(target.key);
|
||||
if (pending) return pending;
|
||||
|
||||
const task = withDownloadSlot(() => downloadAndCacheResource(storageKey, target)).finally(() => inFlight.delete(target.key));
|
||||
const generation = cacheGeneration;
|
||||
let task: Promise<string>;
|
||||
task = withDownloadSlot(() => downloadAndCacheResource(storageKey, target, generation)).finally(() => {
|
||||
if (inFlight.get(target.key) === task) inFlight.delete(target.key);
|
||||
});
|
||||
inFlight.set(target.key, task);
|
||||
return task;
|
||||
}
|
||||
|
||||
/**
|
||||
* 播放器先使用支持 Range 的资源 URL 起播;确认用户实际播放后,再延迟下载完整 Blob。
|
||||
* 这样不会让 IndexedDB 缓存阻塞首帧,同时后续打开可直接复用本地 Object URL。
|
||||
*/
|
||||
export function scheduleResourceBlobCache(storageKey: string, delayMs = 4_000) {
|
||||
if (!resourceIdFromStorageKey(storageKey) || scheduled.has(storageKey)) return;
|
||||
scheduled.add(storageKey);
|
||||
const run = () => {
|
||||
void cacheResourceObjectUrl(storageKey)
|
||||
.catch((error) => {
|
||||
// 这是播放后的后台缓存优化,不应让播放器失败;但下载/持久化异常必须可观测。
|
||||
console.warn("后台缓存资源 Blob 失败", { storageKey, error });
|
||||
return "";
|
||||
})
|
||||
.finally(() => scheduled.delete(storageKey));
|
||||
};
|
||||
if (typeof window === "undefined") {
|
||||
run();
|
||||
return;
|
||||
}
|
||||
window.setTimeout(run, Math.max(0, delayMs));
|
||||
}
|
||||
|
||||
function withDownloadSlot<T>(task: () => Promise<T>) {
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
downloadQueue.push(() => {
|
||||
activeDownloads += 1;
|
||||
task().then(resolve, reject).finally(() => {
|
||||
activeDownloads -= 1;
|
||||
runDownloadQueue();
|
||||
});
|
||||
task()
|
||||
.then(resolve, reject)
|
||||
.finally(() => {
|
||||
activeDownloads -= 1;
|
||||
runDownloadQueue();
|
||||
});
|
||||
});
|
||||
runDownloadQueue();
|
||||
});
|
||||
@@ -123,12 +118,12 @@ export async function getCachedResourceBlob(storageKey: string) {
|
||||
const pending = inFlight.get(target.key);
|
||||
if (pending) {
|
||||
await pending.catch(() => "");
|
||||
const downloaded = sessionBlobs.get(target.key) || await blobStore.getItem<Blob>(target.key);
|
||||
const downloaded = sessionBlobs.get(target.key) || (await blobStore.getItem<Blob>(target.key));
|
||||
if (downloaded) return downloaded;
|
||||
return loadAndPersistResource(storageKey);
|
||||
}
|
||||
await cacheResourceObjectUrl(storageKey).catch(() => "");
|
||||
const downloaded = sessionBlobs.get(target.key) || await blobStore.getItem<Blob>(target.key);
|
||||
const downloaded = sessionBlobs.get(target.key) || (await blobStore.getItem<Blob>(target.key));
|
||||
if (downloaded) return downloaded;
|
||||
return loadAndPersistResource(storageKey);
|
||||
}
|
||||
@@ -139,9 +134,9 @@ async function loadAndPersistResource(storageKey: string) {
|
||||
return blob;
|
||||
}
|
||||
|
||||
async function downloadAndCacheResource(storageKey: string, target: ResourceCacheMeta) {
|
||||
async function downloadAndCacheResource(storageKey: string, target: ResourceCacheMeta, generation: number) {
|
||||
const blob = await downloadResourceBlob(storageKey, target);
|
||||
if (!blob) return "";
|
||||
if (!blob || generation !== cacheGeneration) return "";
|
||||
return objectUrl(target.key, blob);
|
||||
}
|
||||
|
||||
@@ -251,7 +246,6 @@ async function touchCacheMeta(target: ResourceCacheMeta) {
|
||||
await metaStore.setItem(target.key, { ...current, lastAccessedAt: now });
|
||||
}
|
||||
|
||||
|
||||
function touchCacheMetaSafely(target: ResourceCacheMeta) {
|
||||
void touchCacheMeta(target).catch((error) => {
|
||||
// 访问时间只服务于缓存淘汰,不影响资源读取;失败可降级,但不能无痕吞掉。
|
||||
|
||||
Reference in New Issue
Block a user