refactor(*): 数据同步与资源删除 - 收敛服务端真相并保证删除可恢复

backend/internal/service/resource_delete.go

@@ -89,7 +89,7 @@ func (s *Service) deleteUserAssetWithResources(userID string, assetID string) er
 		return BadAuthRequest(message)
 	}
 
-	// 所有引用校验必须先完成；任何资源仍被占用时，一个物理文件都不会开始删除。
+	// 所有引用校验必须先完成；仍被其他资源记录共享的物理对象不会进入删除队列。
 	physicalObjects := map[string]*model.Resource{}
 	for index := range resources {
 		resource := &resources[index]
@@ -102,23 +102,38 @@ func (s *Service) deleteUserAssetWithResources(userID string, assetID string) er
 		}
 		physicalObjects[resourceStorageIdentity(resource)] = resource
 	}
-	physicalKeys := make([]string, 0, len(physicalObjects))
-	for key := range physicalObjects {
-		physicalKeys = append(physicalKeys, key)
-	}
-	sort.Strings(physicalKeys)
-	for _, key := range physicalKeys {
-		resource := physicalObjects[key]
-		if err := s.deleteStoredResourceObject(userID, resource); err != nil {
-			return fmt.Errorf("素材文件删除失败，素材记录已保留：%w", err)
-		}
-	}
-	if err := s.repo.DeleteAssetAndResources(userID, assetID, ownedIDs); err != nil {
+	deletionJobs := resourceDeletionJobs(userID, physicalObjects)
+	// 业务记录和 Outbox 必须在同一事务提交。事务失败时物理文件完全不动；
+	// 提交成功后由幂等 worker 清理，进程退出或对象存储暂时失败都可继续重试。
+	if err := s.repo.DeleteAssetAndResources(userID, assetID, ownedIDs, deletionJobs); err != nil {
 		return fmt.Errorf("素材记录删除失败，请重试：%w", err)
 	}
+	if len(deletionJobs) > 0 {
+		go s.drainResourceDeletionJobs(len(deletionJobs))
+	}
 	return nil
 }
 
+func resourceDeletionJobs(userID string, physicalObjects map[string]*model.Resource) []model.ResourceDeletionJob {
+	keys := make([]string, 0, len(physicalObjects))
+	for key := range physicalObjects {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	now := time.Now()
+	jobs := make([]model.ResourceDeletionJob, 0, len(keys))
+	for _, key := range keys {
+		resource := physicalObjects[key]
+		jobs = append(jobs, model.ResourceDeletionJob{
+			ID: newID(), UserID: userID, ResourceID: resource.ID,
+			Provider: resource.Provider, Endpoint: resource.Endpoint, Bucket: resource.Bucket,
+			StorageSettingID: resource.StorageSettingID, ObjectKey: resource.ObjectKey,
+			Status: model.ResourceDeletionStatusPending, NextAttemptAt: now,
+		})
+	}
+	return jobs
+}
+
 type resourceUsage struct {
 	Kind  string
 	ID    string
@@ -161,6 +176,14 @@ func collectOwnedAssetDocumentReferences(raw string, resourceIDs map[string]stru
 	if err := json.Unmarshal([]byte(raw), &value); err != nil {
 		return err
 	}
+	// A scalar JSON document is an explicit URL/storage-key field. Nested scalar
+	// values must carry a registered field name; free text in arrays is not a reference.
+	if scalar, ok := value.(string); ok {
+		if resourceID := canvasResourceID(scalar); resourceID != "" {
+			resourceIDs[resourceID] = struct{}{}
+		}
+		return nil
+	}
 	walkReferenceDocument(value, "", resourceIDs)
 	return nil
 }
@@ -176,13 +199,12 @@ func walkReferenceDocument(value any, parentKey string, resourceIDs map[string]s
 			walkReferenceDocument(child, parentKey, resourceIDs)
 		}
 	case string:
-		key := normalizeReferenceKey(parentKey)
-		if isResourceLocatorKey(key) {
+		if isResourceLocatorField(parentKey) {
 			if resourceID := canvasResourceID(item); resourceID != "" {
 				resourceIDs[resourceID] = struct{}{}
 			}
 		}
-		if isBareResourceIDKey(key) {
+		if isBareResourceIDField(parentKey) {
 			if resourceID := validCanvasResourceID(item); resourceID != "" {
 				resourceIDs[resourceID] = struct{}{}
 			}
@@ -190,21 +212,20 @@ func walkReferenceDocument(value any, parentKey string, resourceIDs map[string]s
 	}
 }
 
-func normalizeReferenceKey(key string) string {
-	key = strings.ToLower(strings.TrimSpace(key))
-	return strings.NewReplacer("_", "", "-", "").Replace(key)
-}
-
-func isBareResourceIDKey(key string) bool {
-	return key == "resourceid" || key == "resourceids" || key == "sampleresourceid" || key == "referenceresourceid" || key == "referenceresourceids"
-}
-
-func isResourceLocatorKey(key string) bool {
-	if key == "" || strings.HasSuffix(key, "url") || strings.HasSuffix(key, "urls") || strings.HasSuffix(key, "storagekey") {
+func isBareResourceIDField(field string) bool {
+	switch field {
+	case "resourceId", "resourceIds", "sampleResourceId", "referenceResourceId", "referenceResourceIds":
 		return true
+	default:
+		return false
 	}
-	switch key {
-	case "content", "data", "image", "images", "video", "videos", "audio", "audios", "media", "references", "result", "output", "artifactref", "providerartifactref":
+}
+
+// Resource locator fields are a schema contract, not a naming heuristic.
+// Adding or renaming a persisted field requires updating this registry and its tests.
+func isResourceLocatorField(field string) bool {
+	switch field {
+	case "storageKey", "content", "url", "dataUrl", "coverUrl", "imageUrl", "videoUrl", "audioUrl", "referenceUrl", "referenceUrls", "artifactRef", "providerArtifactRef":
 		return true
 	default:
 		return false
@@ -219,7 +240,13 @@ func documentReferencesResources(raw string, resourceIDs map[string]struct{}) bo
 	var value any
 	if err := json.Unmarshal([]byte(raw), &value); err == nil {
 		found := map[string]struct{}{}
-		walkReferenceDocument(value, "", found)
+		if scalar, ok := value.(string); ok {
+			if resourceID := canvasResourceID(scalar); resourceID != "" {
+				found[resourceID] = struct{}{}
+			}
+		} else {
+			walkReferenceDocument(value, "", found)
+		}
 		for resourceID := range found {
 			if _, exists := resourceIDs[resourceID]; exists {
 				return true
@@ -227,11 +254,10 @@ func documentReferencesResources(raw string, resourceIDs map[string]struct{}) bo
 		}
 		return false
 	}
-	// cover_url 等字段可以直接保存资源 URL，而不是 JSON。候选记录已经包含完整 ID，保守阻止删除。
-	for resourceID := range resourceIDs {
-		if strings.Contains(raw, resourceID) {
-			return true
-		}
+	// cover_url 等数据库列可以直接保存一个资源 URL，而不是 JSON。
+	if resourceID := canvasResourceID(raw); resourceID != "" {
+		_, exists := resourceIDs[resourceID]
+		return exists
 	}
 	return false
 }

backend/internal/service/resource_delete_test.go

@@ -5,6 +5,14 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
+
+	"infinite-canvas/backend/internal/database"
+	"infinite-canvas/backend/internal/model"
+	"infinite-canvas/backend/internal/repository"
+
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
 )
 
 func TestCollectOwnedAssetDocumentReferences(t *testing.T) {
@@ -26,6 +34,38 @@ func TestCollectOwnedAssetDocumentReferences(t *testing.T) {
 	}
 }
 
+func TestResourceReferenceFieldsUseExplicitSchema(t *testing.T) {
+	resources := map[string]struct{}{}
+	raw := `{
+		"storage_key":"resource:legacy-snake-case",
+		"resourceStorageKey":"resource:guessed-suffix",
+		"errorMessage":"failed to load /api/resources/diagnostic-only/file",
+		"artifacts":[{"id":"bare-id"}],
+		"referenceResourceIds":["explicit-id"],
+		"providerArtifactRef":"resource:explicit-artifact"
+	}`
+	if err := collectOwnedAssetDocumentReferences(raw, resources); err != nil {
+		t.Fatal(err)
+	}
+	for _, resourceID := range []string{"explicit-id", "explicit-artifact"} {
+		if _, exists := resources[resourceID]; !exists {
+			t.Fatalf("explicit resource field %q was not collected: %#v", resourceID, resources)
+		}
+	}
+	for _, resourceID := range []string{"legacy-snake-case", "guessed-suffix", "diagnostic-only", "bare-id"} {
+		if _, exists := resources[resourceID]; exists {
+			t.Fatalf("unregistered field inferred resource %q: %#v", resourceID, resources)
+		}
+	}
+	arrayResources := map[string]struct{}{}
+	if err := collectOwnedAssetDocumentReferences(`{"artifacts":["resource:array-text"]}`, arrayResources); err != nil {
+		t.Fatal(err)
+	}
+	if _, exists := arrayResources["array-text"]; exists {
+		t.Fatalf("unnamed array values must not become resource references: %#v", arrayResources)
+	}
+}
+
 func TestDocumentReferencesResourcesUsesExactResourceID(t *testing.T) {
 	candidates := map[string]struct{}{"resource-1": {}}
 	if documentReferencesResources(`{"storageKey":"resource:resource-10"}`, candidates) {
@@ -34,6 +74,12 @@ func TestDocumentReferencesResourcesUsesExactResourceID(t *testing.T) {
 	if !documentReferencesResources(`{"storageKey":"resource:resource-1"}`, candidates) {
 		t.Fatal("exact resource ID should match")
 	}
+	if documentReferencesResources(`generation failed for resource-1`, candidates) {
+		t.Fatal("free-form diagnostic text must not be treated as a resource reference")
+	}
+	if !documentReferencesResources(`/api/resources/resource-1/file`, candidates) {
+		t.Fatal("a direct resource locator should match")
+	}
 }
 
 func TestResourceOccupiedMessageNamesBusinessRecord(t *testing.T) {
@@ -74,3 +120,91 @@ func TestDeleteLocalResourceObjectRemovesOnlyResourceDirectoryFile(t *testing.T)
 		t.Fatalf("outside file was changed: %v", err)
 	}
 }
+
+func TestDeleteAssetDatabaseFailureLeavesPhysicalObjectAndNoOutbox(t *testing.T) {
+	svc, db, dataDir := newResourceDeletionTestService(t)
+	objectKey := "users/user-1/image/asset.png"
+	resourcePath := filepath.Join(dataDir, "resources", filepath.FromSlash(objectKey))
+	if err := os.MkdirAll(filepath.Dir(resourcePath), 0o750); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(resourcePath, []byte("image"), 0o640); err != nil {
+		t.Fatal(err)
+	}
+	resource := model.Resource{ID: "resource-1", UserID: "user-1", Provider: "local", ObjectKey: objectKey, Status: model.ResourceStatusReady}
+	asset := model.Asset{ID: "asset-1", UserID: "user-1", Title: "test", PayloadJSON: `{"data":{"storageKey":"resource:resource-1"}}`}
+	if err := db.Create(&resource).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&asset).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Exec("CREATE TRIGGER fail_asset_delete BEFORE DELETE ON assets BEGIN SELECT RAISE(ABORT, 'forced asset delete failure'); END;").Error; err != nil {
+		t.Fatal(err)
+	}
+
+	if err := svc.deleteUserAssetWithResources("user-1", "asset-1"); err == nil {
+		t.Fatal("forced database failure should be returned")
+	}
+	if _, err := os.Stat(resourcePath); err != nil {
+		t.Fatalf("physical object changed before the database transaction committed: %v", err)
+	}
+	var assetCount, resourceCount, jobCount int64
+	if err := db.Model(&model.Asset{}).Where("id = ?", asset.ID).Count(&assetCount).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Model(&model.Resource{}).Where("id = ?", resource.ID).Count(&resourceCount).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Model(&model.ResourceDeletionJob{}).Count(&jobCount).Error; err != nil {
+		t.Fatal(err)
+	}
+	if assetCount != 1 || resourceCount != 1 || jobCount != 0 {
+		t.Fatalf("transaction was not rolled back: asset=%d resource=%d jobs=%d", assetCount, resourceCount, jobCount)
+	}
+}
+
+func TestResourceDeletionWorkerRemovesObjectAndCompletesOutbox(t *testing.T) {
+	svc, db, dataDir := newResourceDeletionTestService(t)
+	objectKey := "users/user-1/image/queued.png"
+	resourcePath := filepath.Join(dataDir, "resources", filepath.FromSlash(objectKey))
+	if err := os.MkdirAll(filepath.Dir(resourcePath), 0o750); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(resourcePath, []byte("image"), 0o640); err != nil {
+		t.Fatal(err)
+	}
+	job := model.ResourceDeletionJob{
+		ID: "deletion-1", UserID: "user-1", ResourceID: "resource-1",
+		Provider: "local", ObjectKey: objectKey,
+		Status: model.ResourceDeletionStatusPending, NextAttemptAt: time.Now().Add(-time.Second),
+	}
+	if err := db.Create(&job).Error; err != nil {
+		t.Fatal(err)
+	}
+
+	svc.drainResourceDeletionJobs(1)
+	if _, err := os.Stat(resourcePath); !os.IsNotExist(err) {
+		t.Fatalf("queued physical object was not deleted: %v", err)
+	}
+	var count int64
+	if err := db.Model(&model.ResourceDeletionJob{}).Where("id = ?", job.ID).Count(&count).Error; err != nil {
+		t.Fatal(err)
+	}
+	if count != 0 {
+		t.Fatal("completed deletion job was not removed")
+	}
+}
+
+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{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := database.MigrateSchema(db); err != nil {
+		t.Fatal(err)
+	}
+	dataDir := t.TempDir()
+	return New(repository.New(db), dataDir), db, dataDir
+}

backend/internal/service/resource_deletion_worker.go

@@ -0,0 +1,59 @@
+package service
+
+import (
+	"log"
+	"math"
+	"time"
+
+	"infinite-canvas/backend/internal/model"
+)
+
+const resourceDeletionLease = 2 * time.Minute
+
+func (s *Service) startResourceDeletionWorker() {
+	go func() {
+		s.drainResourceDeletionJobs(32)
+		ticker := time.NewTicker(15 * time.Second)
+		defer ticker.Stop()
+		for range ticker.C {
+			s.drainResourceDeletionJobs(32)
+		}
+	}()
+}
+
+func (s *Service) drainResourceDeletionJobs(limit int) {
+	owner := s.workerID
+	if owner == "" {
+		owner = newID()
+	}
+	for index := 0; index < limit; index++ {
+		job, err := s.repo.ClaimNextResourceDeletionJob(owner, resourceDeletionLease)
+		if err != nil {
+			log.Printf("resource deletion worker claim failed: %v", err)
+			return
+		}
+		if job == nil {
+			return
+		}
+		resource := &model.Resource{
+			ID: job.ResourceID, UserID: job.UserID, Provider: job.Provider,
+			Endpoint: job.Endpoint, Bucket: job.Bucket, StorageSettingID: job.StorageSettingID,
+			ObjectKey: job.ObjectKey,
+		}
+		if err := s.deleteStoredResourceObject(job.UserID, resource); err != nil {
+			delay := resourceDeletionRetryDelay(job.Attempts)
+			if retryErr := s.repo.RetryResourceDeletionJob(job.ID, owner, err.Error(), time.Now().Add(delay)); retryErr != nil {
+				log.Printf("resource deletion worker retry update failed for %s: %v", job.ID, retryErr)
+			}
+			continue
+		}
+		if err := s.repo.CompleteResourceDeletionJob(job.ID, owner); err != nil {
+			log.Printf("resource deletion worker completion failed for %s: %v", job.ID, err)
+		}
+	}
+}
+
+func resourceDeletionRetryDelay(attempts int) time.Duration {
+	exponent := math.Min(float64(max(attempts-1, 0)), 8)
+	return time.Duration(math.Pow(2, exponent)) * 15 * time.Second
+}

backend/internal/service/service.go

@@ -123,6 +123,7 @@ func (s *Service) taskBilling() *taskBillingCoordinator {
 
 func (s *Service) StartWorker() {
 	s.taskWorker().start()
+	s.startResourceDeletionWorker()
 }
 
 func (s *Service) CreateSession(userID string, req CreateSessionRequest) (*SessionDetail, error) {