fix(canvas): keep uploaded resources in asset library (#388)
This commit is contained in:
1 parent
58defbc3b9
commit
16913ee72c
22 files changed
+1154
-34
No files matched your search
@@ -989,9 +989,17 @@ func (r *Repository) UserStoredFileBytes(userID string) (int64, error) {
|
||||
var total int64
|
||||
err := r.db.Raw(`
|
||||
SELECT
|
||||
(SELECT COALESCE(SUM(size), 0) FROM resources WHERE user_id = ?)
|
||||
COALESCE((
|
||||
SELECT SUM(physical_resources.size)
|
||||
FROM (
|
||||
SELECT MAX(size) AS size
|
||||
FROM resources
|
||||
WHERE user_id = ? AND status = ?
|
||||
GROUP BY COALESCE(NULLIF(provider, ''), 'local'), endpoint, bucket, object_key
|
||||
) AS physical_resources
|
||||
), 0)
|
||||
+ (SELECT COALESCE(SUM(size), 0) FROM session_files WHERE user_id = ?)
|
||||
`, userID, userID).Scan(&total).Error
|
||||
`, userID, model.ResourceStatusReady, userID).Scan(&total).Error
|
||||
return total, err
|
||||
}
|
||||
|
||||
@@ -1053,6 +1061,19 @@ func (r *Repository) Resources(userID string, limit int) ([]model.Resource, erro
|
||||
return resources, err
|
||||
}
|
||||
|
||||
func (r *Repository) ResourceCleanupCandidates(incompleteBefore time.Time, readyBefore time.Time, limit int) ([]model.Resource, error) {
|
||||
var resources []model.Resource
|
||||
if limit <= 0 || limit > 500 {
|
||||
limit = 100
|
||||
}
|
||||
err := r.db.Where(
|
||||
"(status IN ? AND updated_at <= ?) OR (status = ? AND created_at <= ?)",
|
||||
[]model.ResourceStatus{model.ResourceStatusPending, model.ResourceStatusFailed}, incompleteBefore,
|
||||
model.ResourceStatusReady, readyBefore,
|
||||
).Order("created_at asc, id asc").Limit(limit).Find(&resources).Error
|
||||
return resources, err
|
||||
}
|
||||
|
||||
func (r *Repository) Assets(userID string) ([]model.Asset, error) {
|
||||
var assets []model.Asset
|
||||
err := r.db.Order("updated_at desc").Find(&assets, "user_id = ?", userID).Error
|
||||
@@ -1073,6 +1094,15 @@ func (r *Repository) AssetForUser(userID string, id string) (*model.Asset, error
|
||||
return &asset, nil
|
||||
}
|
||||
|
||||
func (r *Repository) AssetsForUserIDs(userID string, ids []string) ([]model.Asset, error) {
|
||||
if len(ids) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
var assets []model.Asset
|
||||
err := r.db.Find(&assets, "user_id = ? AND id IN ?", userID, ids).Error
|
||||
return assets, err
|
||||
}
|
||||
|
||||
func (r *Repository) UpsertAsset(asset *model.Asset) error {
|
||||
result := r.db.Model(&model.Asset{}).
|
||||
Where("id = ? AND user_id = ?", asset.ID, asset.UserID).
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
)
|
||||
|
||||
var ErrResourceCleanupSetChanged = errors.New("resource cleanup set changed")
|
||||
var ErrResourceCleanupStillReferenced = errors.New("resource cleanup resource is still directly referenced")
|
||||
|
||||
// DeleteDetachedResources removes Resource rows and enqueues physical deletion in one transaction.
|
||||
// JSON references are checked by the service before this call; direct foreign-key-like references
|
||||
// are checked again while the candidate rows are locked to avoid deleting a newly attached object.
|
||||
func (r *Repository) DeleteDetachedResources(resources []model.Resource, deletionJobs []model.ResourceDeletionJob) error {
|
||||
if len(resources) == 0 {
|
||||
return nil
|
||||
}
|
||||
resourceIDs := make([]string, 0, len(resources))
|
||||
for _, resource := range resources {
|
||||
resourceIDs = append(resourceIDs, resource.ID)
|
||||
}
|
||||
return r.db.Transaction(func(tx *gorm.DB) error {
|
||||
var current []model.Resource
|
||||
query := tx.Where("id IN ?", resourceIDs)
|
||||
if r.Dialect() == "postgres" {
|
||||
query = query.Clauses(clause.Locking{Strength: "UPDATE"})
|
||||
}
|
||||
if err := query.Find(¤t).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if len(current) != len(resources) {
|
||||
return ErrResourceCleanupSetChanged
|
||||
}
|
||||
// The service performs exact JSON parsing before entering this transaction.
|
||||
// This conservative second check closes the common race where an Asset or
|
||||
// canvas is attached after that snapshot but before Resource deletion.
|
||||
documentsByUser := map[string][]string{}
|
||||
for _, resource := range current {
|
||||
if _, loaded := documentsByUser[resource.UserID]; loaded {
|
||||
continue
|
||||
}
|
||||
var assetDocuments []string
|
||||
if err := tx.Model(&model.Asset{}).Where("user_id = ?", resource.UserID).Pluck("payload_json", &assetDocuments).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
var canvasDocuments []string
|
||||
if err := tx.Model(&model.CanvasProject{}).Where("user_id = ?", resource.UserID).Pluck("payload_json", &canvasDocuments).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
documentsByUser[resource.UserID] = append(assetDocuments, canvasDocuments...)
|
||||
}
|
||||
for _, resource := range current {
|
||||
storageKey := "resource:" + resource.ID + `"`
|
||||
fileURL := "/api/resources/" + resource.ID + "/"
|
||||
for _, document := range documentsByUser[resource.UserID] {
|
||||
if strings.Contains(document, storageKey) || strings.Contains(document, fileURL) {
|
||||
return ErrResourceCleanupStillReferenced
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, check := range []struct {
|
||||
model any
|
||||
query string
|
||||
}{
|
||||
{&model.Announcement{}, "image_resource_id IN ?"},
|
||||
{&model.AnnouncementImageDraft{}, "resource_id IN ?"},
|
||||
{&model.AssetRepresentation{}, "resource_id IN ?"},
|
||||
{&model.VoiceProfile{}, "sample_resource_id IN ?"},
|
||||
{&model.ShotArtifact{}, "resource_id IN ?"},
|
||||
} {
|
||||
var count int64
|
||||
if err := tx.Model(check.model).Where(check.query, resourceIDs).Count(&count).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if count > 0 {
|
||||
return ErrResourceCleanupStillReferenced
|
||||
}
|
||||
}
|
||||
if err := tx.Where("resource_id IN ?", resourceIDs).Delete(&model.ArkPrivateAssetBinding{}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if len(deletionJobs) > 0 {
|
||||
if err := tx.Create(&deletionJobs).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
deleted := tx.Where("id IN ?", resourceIDs).Delete(&model.Resource{})
|
||||
if deleted.Error != nil {
|
||||
return deleted.Error
|
||||
}
|
||||
if deleted.RowsAffected != int64(len(resourceIDs)) {
|
||||
return ErrResourceCleanupSetChanged
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
@@ -103,6 +103,46 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s
|
||||
snapshot.Documents = append(snapshot.Documents, ResourceReferenceDocument{Kind: "画布", ID: canvas.ID, Title: canvas.Title, PrimaryJSON: canvas.PayloadJSON})
|
||||
}
|
||||
|
||||
var tasks []model.Task
|
||||
if err := r.db.Select("id", "prompt", "input_json", "result_json").Where("user_id = ?", userID).Find(&tasks).Error; err != nil {
|
||||
return snapshot, err
|
||||
}
|
||||
for _, task := range tasks {
|
||||
snapshot.Documents = append(snapshot.Documents, ResourceReferenceDocument{Kind: "任务", ID: task.ID, Title: task.Prompt, PrimaryJSON: task.InputJSON, SecondaryJSON: task.ResultJSON})
|
||||
}
|
||||
|
||||
var sessions []model.Session
|
||||
if err := r.db.Select("id", "prompt", "canvas_snapshot_json", "canvas_ops_json").Where("user_id = ?", userID).Find(&sessions).Error; err != nil {
|
||||
return snapshot, err
|
||||
}
|
||||
for _, session := range sessions {
|
||||
snapshot.Documents = append(snapshot.Documents, ResourceReferenceDocument{Kind: "会话", ID: session.ID, Title: session.Prompt, PrimaryJSON: session.CanvasSnapshotJSON, SecondaryJSON: session.CanvasOpsJSON})
|
||||
}
|
||||
|
||||
var messages []model.Message
|
||||
if err := r.db.Select("id", "content", "payload").Where("user_id = ?", userID).Find(&messages).Error; err != nil {
|
||||
return snapshot, err
|
||||
}
|
||||
for _, message := range messages {
|
||||
snapshot.Documents = append(snapshot.Documents, ResourceReferenceDocument{Kind: "会话消息", ID: message.ID, Title: message.Content, PrimaryJSON: message.Payload})
|
||||
}
|
||||
|
||||
var taskLogs []model.TaskLog
|
||||
if err := r.db.Select("id", "message", "payload").Where("user_id = ?", userID).Find(&taskLogs).Error; err != nil {
|
||||
return snapshot, err
|
||||
}
|
||||
for _, taskLog := range taskLogs {
|
||||
snapshot.Documents = append(snapshot.Documents, ResourceReferenceDocument{Kind: "任务日志", ID: taskLog.ID, Title: taskLog.Message, PrimaryJSON: taskLog.Payload})
|
||||
}
|
||||
|
||||
var results []model.Result
|
||||
if err := r.db.Select("id", "kind", "url", "payload").Where("user_id = ?", userID).Find(&results).Error; err != nil {
|
||||
return snapshot, err
|
||||
}
|
||||
for _, result := range results {
|
||||
snapshot.Documents = append(snapshot.Documents, ResourceReferenceDocument{Kind: "任务结果", ID: result.ID, Title: result.Kind, PrimaryJSON: result.URL, SecondaryJSON: result.Payload})
|
||||
}
|
||||
|
||||
var projects []model.Project
|
||||
if err := r.db.Where("user_id = ?", userID).Find(&projects).Error; err != nil {
|
||||
return snapshot, err
|
||||
@@ -221,6 +261,22 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s
|
||||
for _, artifact := range artifacts {
|
||||
snapshot.Direct = append(snapshot.Direct, ResourceDirectReference{Kind: "镜头产物", ID: artifact.ID, Title: artifact.Title, ResourceID: artifact.ResourceID})
|
||||
}
|
||||
|
||||
var announcements []model.Announcement
|
||||
if err := r.db.Where("created_by = ? AND image_resource_id IN ?", userID, resourceIDs).Find(&announcements).Error; err != nil {
|
||||
return snapshot, err
|
||||
}
|
||||
for _, announcement := range announcements {
|
||||
snapshot.Direct = append(snapshot.Direct, ResourceDirectReference{Kind: "公告", ID: announcement.ID, Title: announcement.Title, ResourceID: announcement.ImageResourceID})
|
||||
}
|
||||
|
||||
var announcementDrafts []model.AnnouncementImageDraft
|
||||
if err := r.db.Where("user_id = ? AND resource_id IN ?", userID, resourceIDs).Find(&announcementDrafts).Error; err != nil {
|
||||
return snapshot, err
|
||||
}
|
||||
for _, draft := range announcementDrafts {
|
||||
snapshot.Direct = append(snapshot.Direct, ResourceDirectReference{Kind: "公告草稿", ID: draft.ResourceID, ResourceID: draft.ResourceID})
|
||||
}
|
||||
return snapshot, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,201 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
)
|
||||
|
||||
type canvasMediaAssetReference struct {
|
||||
AssetID string
|
||||
ResourceID string
|
||||
}
|
||||
|
||||
// validateCanvasMediaAssets is the final server-side invariant for canvas sync:
|
||||
// a persisted canvas may only point at an uploaded Resource through an Asset
|
||||
// owned by the same user. The client writes Assets before canvases, so rejecting
|
||||
// an incomplete pair prevents a durable Resource-only (ghost capacity) state.
|
||||
func (s *Service) validateCanvasMediaAssets(userID string, raw json.RawMessage) error {
|
||||
references, err := canvasMediaAssetReferences(raw)
|
||||
if err != nil {
|
||||
return BadAuthRequest("画布媒体数据格式错误")
|
||||
}
|
||||
if len(references) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
assetIDSet := make(map[string]struct{}, len(references))
|
||||
resourceIDSet := make(map[string]struct{}, len(references))
|
||||
for _, reference := range references {
|
||||
if reference.AssetID == "" {
|
||||
return BadAuthRequest("画布媒体尚未进入素材库,请等待同步完成后重试")
|
||||
}
|
||||
assetIDSet[reference.AssetID] = struct{}{}
|
||||
resourceIDSet[reference.ResourceID] = struct{}{}
|
||||
}
|
||||
|
||||
assets, err := s.repo.AssetsForUserIDs(userID, sortedReferenceIDs(assetIDSet))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
assetResources := make(map[string]map[string]struct{}, len(assets))
|
||||
for _, asset := range assets {
|
||||
assetResources[asset.ID] = documentReferencedResourceIDs(asset.PayloadJSON, resourceIDSet)
|
||||
}
|
||||
|
||||
resources, err := s.repo.ResourcesForUserIDs(userID, sortedReferenceIDs(resourceIDSet))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
readyResources := make(map[string]struct{}, len(resources))
|
||||
for _, resource := range resources {
|
||||
if resource.Status == model.ResourceStatusReady {
|
||||
readyResources[resource.ID] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
for _, reference := range references {
|
||||
if _, exists := readyResources[reference.ResourceID]; !exists {
|
||||
return BadAuthRequest("画布媒体对应的云端资源不存在或尚未就绪,请重新上传")
|
||||
}
|
||||
resourceIDs, assetExists := assetResources[reference.AssetID]
|
||||
if !assetExists {
|
||||
return BadAuthRequest("画布媒体尚未进入素材库,请等待同步完成后重试")
|
||||
}
|
||||
if _, matches := resourceIDs[reference.ResourceID]; !matches {
|
||||
return BadAuthRequest("画布媒体与素材库记录不一致,请重新同步")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// 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 {
|
||||
canvases, err := s.repo.CanvasProjects(userID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, canvas := range canvases {
|
||||
references, parseErr := canvasMediaAssetReferences(json.RawMessage(canvas.PayloadJSON))
|
||||
if parseErr != nil {
|
||||
return BadAuthRequest("已有画布媒体数据无法解析,已停止修改素材")
|
||||
}
|
||||
for _, reference := range references {
|
||||
if reference.AssetID != asset.ID {
|
||||
continue
|
||||
}
|
||||
candidate := map[string]struct{}{reference.ResourceID: {}}
|
||||
if !documentReferencesResources(asset.PayloadJSON, candidate) {
|
||||
return BadAuthRequest("素材仍被画布引用,不能替换为其他云端资源")
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// validateAssetReplacementCanvasReferences applies the same invariant to the
|
||||
// legacy full-replacement endpoint, which otherwise could silently remove an
|
||||
// Asset that a canvas still needs.
|
||||
func (s *Service) validateAssetReplacementCanvasReferences(userID string, assets []model.Asset) error {
|
||||
assetByID := make(map[string]model.Asset, len(assets))
|
||||
for _, asset := range assets {
|
||||
assetByID[asset.ID] = asset
|
||||
}
|
||||
canvases, err := s.repo.CanvasProjects(userID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, canvas := range canvases {
|
||||
references, parseErr := canvasMediaAssetReferences(json.RawMessage(canvas.PayloadJSON))
|
||||
if parseErr != nil {
|
||||
return BadAuthRequest("已有画布媒体数据无法解析,已停止替换素材库")
|
||||
}
|
||||
for _, reference := range references {
|
||||
asset, exists := assetByID[reference.AssetID]
|
||||
if !exists {
|
||||
return BadAuthRequest("素材仍被画布引用,不能从素材库移除")
|
||||
}
|
||||
candidate := map[string]struct{}{reference.ResourceID: {}}
|
||||
if !documentReferencesResources(asset.PayloadJSON, candidate) {
|
||||
return BadAuthRequest("画布媒体与替换后的素材库记录不一致")
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func canvasMediaAssetReferences(raw json.RawMessage) ([]canvasMediaAssetReference, error) {
|
||||
var payload struct {
|
||||
Nodes []struct {
|
||||
Type string `json:"type"`
|
||||
Metadata struct {
|
||||
AssetID string `json:"assetId"`
|
||||
StorageKey string `json:"storageKey"`
|
||||
Content string `json:"content"`
|
||||
} `json:"metadata"`
|
||||
} `json:"nodes"`
|
||||
Timeline struct {
|
||||
Clips []struct {
|
||||
DirectMedia *struct {
|
||||
Kind string `json:"kind"`
|
||||
AssetID string `json:"assetId"`
|
||||
StorageKey string `json:"storageKey"`
|
||||
URL string `json:"url"`
|
||||
DataURL string `json:"dataUrl"`
|
||||
Content string `json:"content"`
|
||||
} `json:"directMedia"`
|
||||
} `json:"clips"`
|
||||
} `json:"timeline"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &payload); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
references := make([]canvasMediaAssetReference, 0)
|
||||
for _, node := range payload.Nodes {
|
||||
if !isCanvasMediaKind(node.Type) {
|
||||
continue
|
||||
}
|
||||
resourceID := firstCanvasResourceID(node.Metadata.StorageKey, node.Metadata.Content)
|
||||
if resourceID == "" {
|
||||
continue
|
||||
}
|
||||
references = append(references, canvasMediaAssetReference{
|
||||
AssetID: strings.TrimSpace(node.Metadata.AssetID), ResourceID: resourceID,
|
||||
})
|
||||
}
|
||||
for _, clip := range payload.Timeline.Clips {
|
||||
media := clip.DirectMedia
|
||||
if media == nil || !isCanvasMediaKind(media.Kind) {
|
||||
continue
|
||||
}
|
||||
resourceID := firstCanvasResourceID(media.StorageKey, media.URL, media.DataURL, media.Content)
|
||||
if resourceID == "" {
|
||||
continue
|
||||
}
|
||||
references = append(references, canvasMediaAssetReference{
|
||||
AssetID: strings.TrimSpace(media.AssetID), ResourceID: resourceID,
|
||||
})
|
||||
}
|
||||
return references, nil
|
||||
}
|
||||
|
||||
func firstCanvasResourceID(values ...string) string {
|
||||
for _, value := range values {
|
||||
if resourceID := canvasResourceID(value); resourceID != "" {
|
||||
return resourceID
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func isCanvasMediaKind(kind string) bool {
|
||||
switch strings.ToLower(strings.TrimSpace(kind)) {
|
||||
case "image", "video", "audio":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
@@ -312,6 +312,98 @@ func TestExpiredArchivedAssetCleanupRespectsCanvasReferencesAndUsesDeletionOutbo
|
||||
}
|
||||
}
|
||||
|
||||
func TestDetachedResourceCleanupRemovesOrphanAndKeepsAssetBackedResource(t *testing.T) {
|
||||
svc, db, _ := newResourceDeletionTestService(t)
|
||||
old := time.Now().Add(-48 * time.Hour)
|
||||
orphan := model.Resource{
|
||||
ID: "resource-detached", UserID: "user-1", Status: model.ResourceStatusReady,
|
||||
Provider: "unsupported-test-provider", ObjectKey: "users/user-1/image/detached.png",
|
||||
CreatedAt: old, UpdatedAt: old,
|
||||
}
|
||||
backed := model.Resource{
|
||||
ID: "resource-backed", UserID: "user-1", Status: model.ResourceStatusReady,
|
||||
Provider: "local", ObjectKey: "users/user-1/image/backed.png",
|
||||
CreatedAt: old, UpdatedAt: old,
|
||||
}
|
||||
asset := model.Asset{
|
||||
ID: "asset-backed", UserID: "user-1", Title: "素材库图片",
|
||||
PayloadJSON: `{"data":{"storageKey":"resource:resource-backed"}}`, CreatedAt: old, UpdatedAt: old,
|
||||
}
|
||||
for _, item := range []any{&orphan, &backed, &asset} {
|
||||
if err := db.Create(item).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := svc.cleanupDetachedUserResources("user-1", []model.Resource{orphan, backed}); err != nil {
|
||||
t.Fatalf("cleanupDetachedUserResources() error = %v", err)
|
||||
}
|
||||
|
||||
var orphanCount, backedCount int64
|
||||
if err := db.Model(&model.Resource{}).Where("id = ?", orphan.ID).Count(&orphanCount).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&model.Resource{}).Where("id = ?", backed.ID).Count(&backedCount).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if orphanCount != 0 || backedCount != 1 {
|
||||
t.Fatalf("cleanup result: orphan=%d backed=%d", orphanCount, backedCount)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResourceCleanupCandidatesUseStatusSpecificRetention(t *testing.T) {
|
||||
_, db, _ := newResourceDeletionTestService(t)
|
||||
now := time.Now()
|
||||
resources := []model.Resource{
|
||||
{ID: "pending-old", UserID: "user-1", Status: model.ResourceStatusPending, CreatedAt: now, UpdatedAt: now.Add(-2 * time.Hour)},
|
||||
{ID: "pending-fresh", UserID: "user-1", Status: model.ResourceStatusPending, CreatedAt: now, UpdatedAt: now.Add(-30 * time.Minute)},
|
||||
{ID: "ready-old", UserID: "user-1", Status: model.ResourceStatusReady, CreatedAt: now.Add(-25 * time.Hour), UpdatedAt: now},
|
||||
{ID: "ready-fresh", UserID: "user-1", Status: model.ResourceStatusReady, CreatedAt: now.Add(-23 * time.Hour), UpdatedAt: now},
|
||||
}
|
||||
for index := range resources {
|
||||
if err := db.Create(&resources[index]).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
candidates, err := repository.New(db).ResourceCleanupCandidates(now.Add(-time.Hour), now.Add(-24*time.Hour), 100)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ids := map[string]bool{}
|
||||
for _, resource := range candidates {
|
||||
ids[resource.ID] = true
|
||||
}
|
||||
if !ids["pending-old"] || !ids["ready-old"] || ids["pending-fresh"] || ids["ready-fresh"] {
|
||||
t.Fatalf("cleanup candidates = %#v", ids)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDetachedFailedResourceWithoutObjectKeyNeedsNoDeletionJob(t *testing.T) {
|
||||
svc, db, _ := newResourceDeletionTestService(t)
|
||||
resource := model.Resource{
|
||||
ID: "resource-failed-empty", UserID: "user-1", Status: model.ResourceStatusFailed,
|
||||
CreatedAt: time.Now().Add(-2 * time.Hour), UpdatedAt: time.Now().Add(-2 * time.Hour),
|
||||
}
|
||||
if err := db.Create(&resource).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := svc.cleanupDetachedUserResources(resource.UserID, []model.Resource{resource}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var resourceCount, jobCount int64
|
||||
if err := db.Model(&model.Resource{}).Where("id = ?", resource.ID).Count(&resourceCount).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&model.ResourceDeletionJob{}).Where("resource_id = ?", resource.ID).Count(&jobCount).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if resourceCount != 0 || jobCount != 0 {
|
||||
t.Fatalf("cleanup result: resource=%d jobs=%d", resourceCount, jobCount)
|
||||
}
|
||||
}
|
||||
|
||||
func newResourceDeletionTestService(t *testing.T) (*Service, *gorm.DB, string) {
|
||||
t.Helper()
|
||||
db, err := gorm.Open(sqlite.Open("file:"+newID()+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
|
||||
@@ -4,18 +4,22 @@ import (
|
||||
"context"
|
||||
"log"
|
||||
"math"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
)
|
||||
|
||||
const resourceDeletionLease = 2 * time.Minute
|
||||
const incompleteResourceRetention = time.Hour
|
||||
const detachedReadyResourceRetention = 24 * time.Hour
|
||||
|
||||
func (s *Service) startResourceDeletionWorker(ctx context.Context) {
|
||||
s.runWorkerLoop(func(ctx context.Context) {
|
||||
s.drainResourceDeletionJobs(32)
|
||||
s.cleanupStaleAnnouncementImageDrafts()
|
||||
s.cleanupExpiredArchivedAssets()
|
||||
s.cleanupDetachedResources()
|
||||
ticker := time.NewTicker(15 * time.Second)
|
||||
defer ticker.Stop()
|
||||
lastPeriodicCleanup := time.Now()
|
||||
@@ -28,6 +32,7 @@ func (s *Service) startResourceDeletionWorker(ctx context.Context) {
|
||||
if time.Since(lastPeriodicCleanup) >= time.Hour {
|
||||
s.cleanupStaleAnnouncementImageDrafts()
|
||||
s.cleanupExpiredArchivedAssets()
|
||||
s.cleanupDetachedResources()
|
||||
lastPeriodicCleanup = time.Now()
|
||||
}
|
||||
}
|
||||
@@ -35,6 +40,93 @@ func (s *Service) startResourceDeletionWorker(ctx context.Context) {
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Service) cleanupDetachedResources() {
|
||||
now := time.Now()
|
||||
candidates, err := s.repo.ResourceCleanupCandidates(
|
||||
now.Add(-incompleteResourceRetention),
|
||||
now.Add(-detachedReadyResourceRetention),
|
||||
500,
|
||||
)
|
||||
if err != nil {
|
||||
log.Printf("detached resource cleanup query failed: %v", err)
|
||||
return
|
||||
}
|
||||
byUser := map[string][]model.Resource{}
|
||||
for _, resource := range candidates {
|
||||
byUser[resource.UserID] = append(byUser[resource.UserID], resource)
|
||||
}
|
||||
for userID, resources := range byUser {
|
||||
if err := s.cleanupDetachedUserResources(userID, resources); err != nil {
|
||||
log.Printf("detached resource cleanup failed for user %s: %v", userID, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) cleanupDetachedUserResources(userID string, candidates []model.Resource) error {
|
||||
s.storageMu.Lock()
|
||||
defer s.storageMu.Unlock()
|
||||
resourceIDs := make([]string, 0, len(candidates))
|
||||
candidateSet := make(map[string]struct{}, len(candidates))
|
||||
for _, resource := range candidates {
|
||||
resourceIDs = append(resourceIDs, resource.ID)
|
||||
candidateSet[resource.ID] = struct{}{}
|
||||
}
|
||||
snapshot, err := s.repo.ResourceReferenceSnapshot(userID, "", resourceIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
referenced := map[string]struct{}{}
|
||||
for _, reference := range snapshot.Direct {
|
||||
if _, exists := candidateSet[reference.ResourceID]; exists {
|
||||
referenced[reference.ResourceID] = struct{}{}
|
||||
}
|
||||
}
|
||||
for _, document := range snapshot.Documents {
|
||||
for resourceID := range documentReferencedResourceIDs(document.PrimaryJSON, candidateSet) {
|
||||
referenced[resourceID] = struct{}{}
|
||||
}
|
||||
for resourceID := range documentReferencedResourceIDs(document.SecondaryJSON, candidateSet) {
|
||||
referenced[resourceID] = struct{}{}
|
||||
}
|
||||
}
|
||||
detached := make([]model.Resource, 0, len(candidates))
|
||||
for _, resource := range candidates {
|
||||
if _, exists := referenced[resource.ID]; !exists {
|
||||
detached = append(detached, resource)
|
||||
}
|
||||
}
|
||||
if len(detached) == 0 {
|
||||
return nil
|
||||
}
|
||||
detachedIDs := make([]string, 0, len(detached))
|
||||
for _, resource := range detached {
|
||||
detachedIDs = append(detachedIDs, resource.ID)
|
||||
}
|
||||
physicalObjects := map[string]*model.Resource{}
|
||||
for index := range detached {
|
||||
resource := &detached[index]
|
||||
if strings.TrimSpace(resource.ObjectKey) == "" {
|
||||
continue
|
||||
}
|
||||
sharedCount, countErr := s.repo.ResourceStorageReferenceCount(resource, detachedIDs)
|
||||
if countErr != nil {
|
||||
return countErr
|
||||
}
|
||||
if sharedCount == 0 {
|
||||
physicalObjects[resourceStorageIdentity(resource)] = resource
|
||||
}
|
||||
}
|
||||
deletionJobs := resourceDeletionJobs(userID, physicalObjects)
|
||||
if err := s.repo.DeleteDetachedResources(detached, deletionJobs); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("detached resource cleanup: removed %d resource rows for user %s", len(detached), userID)
|
||||
if len(deletionJobs) > 0 {
|
||||
go s.drainResourceDeletionJobs(len(deletionJobs))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) cleanupExpiredArchivedAssets() {
|
||||
policy, err := s.RuntimePolicy()
|
||||
if err != nil {
|
||||
|
||||
@@ -70,7 +70,16 @@ func TestReserveUserUploadQuotaRejectsTotalStoredFilesAtLimit(t *testing.T) {
|
||||
|
||||
func TestAccountFileStorageUsageUsesStoredFilePolicy(t *testing.T) {
|
||||
svc := newResourceTestService(t)
|
||||
if err := svc.repo.Create(&model.Resource{ID: "resource-1", UserID: "user-1", Status: model.ResourceStatusReady, Size: 3 << 20}); err != nil {
|
||||
if err := svc.repo.Create(&model.Resource{ID: "resource-1", UserID: "user-1", Status: model.ResourceStatusReady, Provider: "local", ObjectKey: "ready.png", Size: 3 << 20}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := svc.repo.Create(&model.Resource{ID: "resource-duplicate", UserID: "user-1", Status: model.ResourceStatusReady, Provider: "", ObjectKey: "ready.png", Size: 3 << 20}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := svc.repo.Create(&model.Resource{ID: "resource-failed", UserID: "user-1", Status: model.ResourceStatusFailed, Provider: "local", ObjectKey: "failed.png", Size: 7 << 20}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := svc.repo.Create(&model.Resource{ID: "resource-pending", UserID: "user-1", Status: model.ResourceStatusPending, Provider: "local", ObjectKey: "pending.png", Size: 11 << 20}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := svc.repo.Create(&model.SessionFile{ID: "session-file-1", UserID: "user-1", SessionID: "session-1", Size: 2 << 20}); err != nil {
|
||||
|
||||
@@ -82,6 +82,11 @@ func (s *Service) UpsertUserAsset(userID string, raw json.RawMessage) (UserDataS
|
||||
if existingErr != nil && !errors.Is(existingErr, gorm.ErrRecordNotFound) {
|
||||
return UserDataSummary{}, existingErr
|
||||
}
|
||||
if existing != nil && existing.PayloadJSON != asset.PayloadJSON {
|
||||
if err := s.validateAssetCanvasReferences(userID, asset); err != nil {
|
||||
return UserDataSummary{}, err
|
||||
}
|
||||
}
|
||||
existingBytes := int64(0)
|
||||
if existing != nil {
|
||||
existingBytes = int64(len([]byte(existing.PayloadJSON)))
|
||||
@@ -139,6 +144,9 @@ func (s *Service) ReplaceUserAssets(userID string, req AssetsSyncRequest) ([]jso
|
||||
}
|
||||
s.storageMu.Lock()
|
||||
defer s.storageMu.Unlock()
|
||||
if err := s.validateAssetReplacementCanvasReferences(userID, assets); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
usage, err := s.repo.UserStorageUsage(userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -200,6 +208,9 @@ func (s *Service) UpsertUserCanvasProject(userID string, raw json.RawMessage) (U
|
||||
}
|
||||
s.storageMu.Lock()
|
||||
defer s.storageMu.Unlock()
|
||||
if err := s.validateCanvasMediaAssets(userID, raw); err != nil {
|
||||
return UserDataSummary{}, err
|
||||
}
|
||||
existing, existingErr := s.repo.CanvasProjectForUser(userID, project.ID)
|
||||
if existingErr != nil && !errors.Is(existingErr, gorm.ErrRecordNotFound) {
|
||||
return UserDataSummary{}, existingErr
|
||||
@@ -245,6 +256,11 @@ func (s *Service) ReplaceUserCanvasProjects(userID string, req CanvasProjectsSyn
|
||||
}
|
||||
s.storageMu.Lock()
|
||||
defer s.storageMu.Unlock()
|
||||
for _, raw := range req.Projects {
|
||||
if err := s.validateCanvasMediaAssets(userID, raw); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
usage, err := s.repo.UserStorageUsage(userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -117,3 +117,105 @@ func TestValidateSyncedPayloadRejectsNestedInlineMedia(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCanvasMediaAssetReferencesCollectsNodesAndTimeline(t *testing.T) {
|
||||
raw := json.RawMessage(`{
|
||||
"nodes":[
|
||||
{"type":"image","metadata":{"assetId":"asset-image","storageKey":"resource:resource-image"}},
|
||||
{"type":"text","metadata":{"assetId":"asset-text","content":"/api/resources/not-media/file"}},
|
||||
{"type":"video","metadata":{"content":"https://cdn.example.com/external.mp4"}}
|
||||
],
|
||||
"timeline":{"clips":[
|
||||
{"directMedia":{"kind":"audio","assetId":"asset-audio","url":"/api/resources/resource-audio/file"}},
|
||||
{"directMedia":{"kind":"text","content":"/api/resources/not-timeline-media/file"}}
|
||||
]}
|
||||
}`)
|
||||
|
||||
references, err := canvasMediaAssetReferences(raw)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(references) != 2 {
|
||||
t.Fatalf("references = %#v", references)
|
||||
}
|
||||
if references[0].AssetID != "asset-image" || references[0].ResourceID != "resource-image" {
|
||||
t.Fatalf("node reference = %#v", references[0])
|
||||
}
|
||||
if references[1].AssetID != "asset-audio" || references[1].ResourceID != "resource-audio" {
|
||||
t.Fatalf("timeline reference = %#v", references[1])
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateCanvasMediaAssetsRejectsResourceWithoutAsset(t *testing.T) {
|
||||
svc, db, _ := newResourceDeletionTestService(t)
|
||||
resource := model.Resource{
|
||||
ID: "resource-canvas-orphan", UserID: "user-1", Status: model.ResourceStatusReady,
|
||||
Provider: "local", ObjectKey: "users/user-1/image/orphan.png",
|
||||
}
|
||||
if err := db.Create(&resource).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw := json.RawMessage(`{
|
||||
"id":"canvas-1",
|
||||
"nodes":[{"id":"node-1","type":"image","metadata":{"storageKey":"resource:resource-canvas-orphan","content":"/api/resources/resource-canvas-orphan/file"}}]
|
||||
}`)
|
||||
|
||||
err := svc.validateCanvasMediaAssets("user-1", raw)
|
||||
if err == nil || !strings.Contains(err.Error(), "尚未进入素材库") {
|
||||
t.Fatalf("validateCanvasMediaAssets() error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateCanvasMediaAssetsAcceptsMatchingNodeAndTimelineAssets(t *testing.T) {
|
||||
svc, db, _ := newResourceDeletionTestService(t)
|
||||
resources := []model.Resource{
|
||||
{ID: "resource-node", UserID: "user-1", Status: model.ResourceStatusReady, Provider: "local", ObjectKey: "users/user-1/image/node.png"},
|
||||
{ID: "resource-timeline", UserID: "user-1", Status: model.ResourceStatusReady, Provider: "local", ObjectKey: "users/user-1/video/timeline.mp4"},
|
||||
}
|
||||
assets := []model.Asset{
|
||||
{ID: "asset-node", UserID: "user-1", PayloadJSON: `{"id":"asset-node","data":{"storageKey":"resource:resource-node"}}`},
|
||||
{ID: "asset-timeline", UserID: "user-1", PayloadJSON: `{"id":"asset-timeline","data":{"url":"/api/resources/resource-timeline/file"}}`},
|
||||
}
|
||||
for index := range resources {
|
||||
if err := db.Create(&resources[index]).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for index := range assets {
|
||||
if err := db.Create(&assets[index]).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
raw := json.RawMessage(`{
|
||||
"id":"canvas-1",
|
||||
"nodes":[{"id":"node-1","type":"image","metadata":{"assetId":"asset-node","storageKey":"resource:resource-node"}}],
|
||||
"timeline":{"clips":[{"id":"clip-1","directMedia":{"kind":"video","assetId":"asset-timeline","storageKey":"resource:resource-timeline"}}]}
|
||||
}`)
|
||||
|
||||
if err := svc.validateCanvasMediaAssets("user-1", raw); err != nil {
|
||||
t.Fatalf("validateCanvasMediaAssets() error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateCanvasMediaAssetsRejectsMismatchedAsset(t *testing.T) {
|
||||
svc, db, _ := newResourceDeletionTestService(t)
|
||||
resources := []model.Resource{
|
||||
{ID: "resource-canvas", UserID: "user-1", Status: model.ResourceStatusReady, Provider: "local", ObjectKey: "users/user-1/image/canvas.png"},
|
||||
{ID: "resource-other", UserID: "user-1", Status: model.ResourceStatusReady, Provider: "local", ObjectKey: "users/user-1/image/other.png"},
|
||||
}
|
||||
for index := range resources {
|
||||
if err := db.Create(&resources[index]).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
asset := model.Asset{ID: "asset-wrong", UserID: "user-1", PayloadJSON: `{"data":{"storageKey":"resource:resource-other"}}`}
|
||||
if err := db.Create(&asset).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw := json.RawMessage(`{"nodes":[{"type":"image","metadata":{"assetId":"asset-wrong","storageKey":"resource:resource-canvas"}}]}`)
|
||||
|
||||
err := svc.validateCanvasMediaAssets("user-1", raw)
|
||||
if err == nil || !strings.Contains(err.Error(), "不一致") {
|
||||
t.Fatalf("validateCanvasMediaAssets() error = %v", err)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user