diff --git a/backend/internal/database/schema.go b/backend/internal/database/schema.go index 88a94673b..3f0d3d579 100644 --- a/backend/internal/database/schema.go +++ b/backend/internal/database/schema.go @@ -45,6 +45,7 @@ func Models() []any { &model.Skill{}, &model.UserSkillState{}, &model.Resource{}, + &model.ResourceDeletionJob{}, &model.Asset{}, &model.ProjectAssetLink{}, &model.ProjectAssetFolder{}, diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index d3d280e07..86f30968d 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -13,6 +13,7 @@ type ChannelScope string type ChannelInterfaceType string type ApiCallStatus string type ResourceStatus string +type ResourceDeletionStatus string type BillingStatus string type CreditLedgerType string type RedeemCodeStatus string @@ -91,6 +92,9 @@ const ( ResourceStatusFailed ResourceStatus = "failed" ResourceStatusDeleted ResourceStatus = "deleted" + ResourceDeletionStatusPending ResourceDeletionStatus = "pending" + ResourceDeletionStatusProcessing ResourceDeletionStatus = "processing" + BillingStatusReserved BillingStatus = "reserved" BillingStatusRunning BillingStatus = "running" BillingStatusSettled BillingStatus = "settled" diff --git a/backend/internal/model/models_project.go b/backend/internal/model/models_project.go index 414ea1daa..0c4d876c2 100644 --- a/backend/internal/model/models_project.go +++ b/backend/internal/model/models_project.go @@ -27,6 +27,28 @@ type Resource struct { UpdatedAt time.Time `json:"updatedAt"` } +// ResourceDeletionJob is the durable handoff between database deletion and +// physical object cleanup. Storage fields are frozen because the Resource row +// is removed in the same transaction that creates this job. +type ResourceDeletionJob struct { + ID string `json:"id" gorm:"primaryKey;size:36"` + UserID string `json:"userId" gorm:"index;size:36"` + ResourceID string `json:"resourceId" gorm:"index;size:36"` + Provider string `json:"provider" gorm:"size:24"` + Endpoint string `json:"endpoint"` + Bucket string `json:"bucket" gorm:"size:160"` + StorageSettingID string `json:"-" gorm:"index;size:36"` + ObjectKey string `json:"objectKey" gorm:"index"` + Status ResourceDeletionStatus `json:"status" gorm:"index:idx_resource_deletion_jobs_due,priority:1;size:24"` + Attempts int `json:"attempts"` + LastError string `json:"lastError" gorm:"type:text"` + NextAttemptAt time.Time `json:"nextAttemptAt" gorm:"index:idx_resource_deletion_jobs_due,priority:2"` + LeaseOwner string `json:"-" gorm:"index;size:120"` + LeaseExpiresAt *time.Time `json:"-" gorm:"index"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + type Asset struct { ID string `json:"id" gorm:"primaryKey;size:80"` UserID string `json:"userId" gorm:"index;size:36;index:idx_assets_user_updated,priority:1"` diff --git a/backend/internal/repository/repository.go b/backend/internal/repository/repository.go index b7c854a61..8c82a29a3 100644 --- a/backend/internal/repository/repository.go +++ b/backend/internal/repository/repository.go @@ -953,7 +953,7 @@ func (r *Repository) UpsertAsset(asset *model.Asset) error { } func (r *Repository) DeleteAsset(userID string, id string) error { - return r.DeleteAssetAndResources(userID, id, nil) + return r.DeleteAssetAndResources(userID, id, nil, nil) } func (r *Repository) ReplaceAssets(userID string, assets []model.Asset) error { diff --git a/backend/internal/repository/resource_deletion.go b/backend/internal/repository/resource_deletion.go new file mode 100644 index 000000000..24bd98460 --- /dev/null +++ b/backend/internal/repository/resource_deletion.go @@ -0,0 +1,67 @@ +package repository + +import ( + "time" + + "infinite-canvas/backend/internal/model" + + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +func (r *Repository) ClaimNextResourceDeletionJob(owner string, leaseDuration time.Duration) (*model.ResourceDeletionJob, error) { + var job model.ResourceDeletionJob + now := time.Now() + err := r.db.Transaction(func(tx *gorm.DB) error { + available := "(status = ? OR (status = ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?))) AND next_attempt_at <= ?" + query := tx.Where(available, model.ResourceDeletionStatusPending, model.ResourceDeletionStatusProcessing, now, now). + Order("next_attempt_at asc, created_at asc").Limit(1) + if r.Dialect() == "postgres" { + query = query.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}) + } + result := query.Find(&job) + if result.Error != nil || result.RowsAffected == 0 { + return result.Error + } + claim := tx.Model(&model.ResourceDeletionJob{}).Where("id = ?", job.ID) + if r.Dialect() != "postgres" { + claim = claim.Where(available, model.ResourceDeletionStatusPending, model.ResourceDeletionStatusProcessing, now, now) + } + updated := claim.Updates(map[string]any{ + "status": model.ResourceDeletionStatusProcessing, + "attempts": gorm.Expr("attempts + ?", 1), + "lease_owner": owner, + "lease_expires_at": now.Add(leaseDuration), + "updated_at": now, + }) + if updated.Error != nil { + return updated.Error + } + if updated.RowsAffected == 0 { + job = model.ResourceDeletionJob{} + return nil + } + return tx.First(&job, "id = ?", job.ID).Error + }) + if err != nil || job.ID == "" { + return nil, err + } + return &job, nil +} + +func (r *Repository) CompleteResourceDeletionJob(id string, owner string) error { + return r.db.Where("id = ? AND lease_owner = ?", id, owner).Delete(&model.ResourceDeletionJob{}).Error +} + +func (r *Repository) RetryResourceDeletionJob(id string, owner string, lastError string, nextAttemptAt time.Time) error { + return r.db.Model(&model.ResourceDeletionJob{}). + Where("id = ? AND lease_owner = ?", id, owner). + Updates(map[string]any{ + "status": model.ResourceDeletionStatusPending, + "last_error": lastError, + "next_attempt_at": nextAttemptAt, + "lease_owner": "", + "lease_expires_at": nil, + "updated_at": time.Now(), + }).Error +} diff --git a/backend/internal/repository/resource_reference.go b/backend/internal/repository/resource_reference.go index 1c596860b..ffd97fcf1 100644 --- a/backend/internal/repository/resource_reference.go +++ b/backend/internal/repository/resource_reference.go @@ -9,7 +9,7 @@ import ( ) // ResourceReferenceDocument 是资源删除校验使用的只读业务文档快照。 -// JSON 中的精确资源 ID 识别由 service 完成，repository 只负责缩小候选记录。 +// repository 只按用户范围读取记录；JSON 中的资源引用合同由 service 统一解释。 type ResourceReferenceDocument struct { Kind string ID string @@ -87,7 +87,7 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s var assets []model.Asset assetQuery := r.db.Where("user_id = ? AND id <> ?", userID, excludingAssetID) - if err := resourceTextQuery(assetQuery, []string{"payload_json"}, resourceIDs).Find(&assets).Error; err != nil { + if err := assetQuery.Find(&assets).Error; err != nil { return snapshot, err } for _, asset := range assets { @@ -95,7 +95,7 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s } var canvases []model.CanvasProject - if err := resourceTextQuery(r.db.Where("user_id = ?", userID), []string{"payload_json"}, resourceIDs).Find(&canvases).Error; err != nil { + if err := r.db.Where("user_id = ?", userID).Find(&canvases).Error; err != nil { return snapshot, err } for _, canvas := range canvases { @@ -103,7 +103,7 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s } var projects []model.Project - if err := resourceTextQuery(r.db.Where("user_id = ?", userID), []string{"style_profile_json"}, resourceIDs).Find(&projects).Error; err != nil { + if err := r.db.Where("user_id = ?", userID).Find(&projects).Error; err != nil { return snapshot, err } for _, project := range projects { @@ -111,7 +111,7 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s } var styles []model.StyleProfile - if err := resourceTextQuery(r.db.Where("user_id = ?", userID), []string{"cover_url", "profile_json"}, resourceIDs).Find(&styles).Error; err != nil { + if err := r.db.Where("user_id = ?", userID).Find(&styles).Error; err != nil { return snapshot, err } for _, style := range styles { @@ -129,7 +129,7 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s 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) - if err := resourceTextQuery(versionQuery, []string{"asset_versions.definition_json"}, resourceIDs).Scan(&versions).Error; err != nil { + if err := versionQuery.Scan(&versions).Error; err != nil { return snapshot, err } for _, version := range versions { @@ -141,7 +141,7 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s Select("project_asset_candidates.id, projects.name AS title, project_asset_candidates.details_json AS primary_json"). Joins("JOIN projects ON projects.id = project_asset_candidates.project_id"). Where("projects.user_id = ?", userID) - if err := resourceTextQuery(candidateQuery, []string{"project_asset_candidates.details_json"}, resourceIDs).Scan(&candidates).Error; err != nil { + if err := candidateQuery.Scan(&candidates).Error; err != nil { return snapshot, err } for _, candidate := range candidates { @@ -154,7 +154,7 @@ func (r *Repository) ResourceReferenceSnapshot(userID string, excludingAssetID s Joins("JOIN workflow_instances ON workflow_instances.id = workflow_step_instances.workflow_instance_id"). Joins("JOIN projects ON projects.id = workflow_instances.project_id"). Where("projects.user_id = ?", userID) - if err := resourceTextQuery(stepQuery, []string{"workflow_step_instances.input_json", "workflow_step_instances.output_json"}, resourceIDs).Scan(&steps).Error; err != nil { + if err := stepQuery.Scan(&steps).Error; err != nil { return snapshot, err } for _, step := range steps { @@ -227,7 +227,7 @@ func (r *Repository) AssetBusinessReferences(userID string, assetID string) ([]R return result, nil } -func (r *Repository) DeleteAssetAndResources(userID string, assetID string, resourceIDs []string) error { +func (r *Repository) DeleteAssetAndResources(userID string, assetID string, resourceIDs []string, deletionJobs []model.ResourceDeletionJob) error { return r.db.Transaction(func(tx *gorm.DB) error { versionIDs := tx.Model(&model.AssetVersion{}).Select("id").Where("asset_id = ?", assetID) if err := tx.Where("asset_version_id IN (?)", versionIDs).Delete(&model.ShotAssetReference{}).Error; err != nil { @@ -251,6 +251,11 @@ func (r *Repository) DeleteAssetAndResources(userID string, assetID string, reso if err := tx.Delete(&model.Asset{}, "id = ? AND user_id = ?", assetID, userID).Error; err != nil { return err } + if len(deletionJobs) > 0 { + if err := tx.Create(&deletionJobs).Error; err != nil { + return err + } + } if len(resourceIDs) == 0 { return nil } @@ -260,18 +265,3 @@ func (r *Repository) DeleteAssetAndResources(userID string, assetID string, reso return tx.Where("user_id = ? AND id IN ?", userID, resourceIDs).Delete(&model.Resource{}).Error }) } - -func resourceTextQuery(query *gorm.DB, columns []string, resourceIDs []string) *gorm.DB { - conditions := make([]string, 0, len(columns)*len(resourceIDs)) - args := make([]any, 0, len(columns)*len(resourceIDs)) - for _, column := range columns { - for _, resourceID := range resourceIDs { - conditions = append(conditions, column+" LIKE ?") - args = append(args, "%"+resourceID+"%") - } - } - if len(conditions) == 0 { - return query - } - return query.Where("("+strings.Join(conditions, " OR ")+")", args...) -} diff --git a/backend/internal/service/resource_delete.go b/backend/internal/service/resource_delete.go index 528f2320f..03b24eea7 100644 --- a/backend/internal/service/resource_delete.go +++ b/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 } diff --git a/backend/internal/service/resource_delete_test.go b/backend/internal/service/resource_delete_test.go index 7b08ed896..d8e63c133 100644 --- a/backend/internal/service/resource_delete_test.go +++ b/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 +} diff --git a/backend/internal/service/resource_deletion_worker.go b/backend/internal/service/resource_deletion_worker.go new file mode 100644 index 000000000..fa7ac520a --- /dev/null +++ b/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 +} diff --git a/backend/internal/service/service.go b/backend/internal/service/service.go index c9003eea7..7687f1ea2 100644 --- a/backend/internal/service/service.go +++ b/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) { diff --git a/docs/content/docs/progress/pending-test.mdx b/docs/content/docs/progress/pending-test.mdx index 126fd6968..afeaa6fa6 100644 --- a/docs/content/docs/progress/pending-test.mdx +++ b/docs/content/docs/progress/pending-test.mdx @@ -80,3 +80,13 @@ description: 影策已实现功能的手工验证清单与未确认边界 - [ ] 点击“重新加载资源”，确认只重新查询并消费原任务结果，不创建新的生成任务或重复扣费。 - [ ] 模拟视频结果落盘/OSS 上传失败后再次加载，确认恢复成功会清除错误状态并保留原任务信息。 - [ ] 点击“重新生成”，确认仍走原有生成重试流程；普通生成失败仍只显示“重新生成”。 + +## 数据同步与资源删除重构 + +- [x] 前端远端基线、登录 hydrated 屏障、上传并发编辑和删除持久化回归测试通过。 +- [x] Generation 失败只回滚生成字段，保留同节点并发标题编辑；重复 Generation 仍按明确 effect key 幂等。 +- [x] 资源引用识别改为显式字段合同，未登记的字段名和自由文本不会被推断为资源引用。 +- [x] 后端素材删除改为数据库事务 + `resource_deletion_jobs` Outbox，数据库失败时物理文件保持不动。 +- [x] Outbox Worker 本地文件删除、对象不存在幂等成功和失败退避测试通过。 +- [ ] 使用真实 OSS 配置验证对象删除成功、权限失败后的退避重试和多实例租约接管。 +- [ ] 浏览器真实环境验证登录快照慢响应期间工作区保持不可写，以及删除后立即刷新不复活素材。 diff --git a/web/src/components/canvas/canvas-assistant-panel.tsx b/web/src/components/canvas/canvas-assistant-panel.tsx index 56e159651..1424ec17c 100644 --- a/web/src/components/canvas/canvas-assistant-panel.tsx +++ b/web/src/components/canvas/canvas-assistant-panel.tsx @@ -631,14 +631,14 @@ export function CanvasAssistantPanel({ setLocalSessions(next); setLocalActiveSessionId(localActiveSessionId && ids.includes(localActiveSessionId) ? next[0].id : localActiveSessionId); } - cleanupImages({ sessions: next }); + void cleanupImages({ sessions: next }); }; const clearSessions = () => { const session = createSession(); setLocalSessions([session]); setLocalActiveSessionId(session.id); - cleanupImages({ sessions: [session] }); + void cleanupImages({ sessions: [session] }); }; const sendMessage = async (text: string, history: CanvasAssistantMessage[], savedReferences?: CanvasAssistantReference[]) => { diff --git a/web/src/components/canvas/canvas-delete-projects-dialog.tsx b/web/src/components/canvas/canvas-delete-projects-dialog.tsx index f13d2f79f..6f3863578 100644 --- a/web/src/components/canvas/canvas-delete-projects-dialog.tsx +++ b/web/src/components/canvas/canvas-delete-projects-dialog.tsx @@ -16,7 +16,7 @@ export function CanvasDeleteProjectsDialog() { setDeleting(true); try { await deleteCanvasProjectsWithRemoteSync(ids); - cleanupImages(); + void cleanupImages(); removeSelectedIds(ids); setDeleteIds([]); } catch (error) { diff --git a/web/src/lib/user-session.ts b/web/src/lib/user-session.ts index 8ad6e317b..3ef80a8dc 100644 --- a/web/src/lib/user-session.ts +++ b/web/src/lib/user-session.ts @@ -8,12 +8,12 @@ import { ASSET_STORE_KEY, flushAssetStorePersistence, useAssetStore } from "@/st import { CONFIG_STORE_KEY, PUBLIC_MODEL_CATALOG_ID, defaultConfig, normalizeConfigSnapshot, useConfigStore, type ModelCapability, type ModelChannel } from "@/stores/use-config-store"; import { defaultModelCapabilityConfig, STANDARD_IMAGE_SIZE_VALUES, type ModelCapabilityConfig } from "@/lib/model-capabilities"; import { useUserStore } from "@/stores/use-user-store"; -import { installRemoteUserDataAutoSync, resetRemoteUserDataSync, syncRemoteUserData, withRemoteUserDataSyncPaused } from "@/services/user-data-sync"; +import { installRemoteUserDataAutoSync, resetRemoteUserDataSync, syncRemoteUserData, withRemoteUserDataSyncExclusive } from "@/services/user-data-sync"; import { withGenerationConsumersPaused } from "@/services/generation-consumer-lifecycle"; export async function switchUserStorageScope(userId?: string | null) { await withGenerationConsumersPaused(async () => { - await withRemoteUserDataSyncPaused(async () => { + await withRemoteUserDataSyncExclusive(async () => { await Promise.all([flushCanvasStorePersistence(), flushAssetStorePersistence()]); resetRemoteUserDataSync(); setActiveUserScope(userId); @@ -69,8 +69,9 @@ export async function applyUserSession(payload: AuthSessionPayload) { } installRemoteUserDataAutoSync(); if (payload.user?.id) { - // 认证状态先完成，云端数据在后台合并；远端同步失败不能伪装成登录失败。 - void syncRemoteUserData(payload.user.id).catch((error) => console.warn("登录后云端数据同步失败，保留本地数据等待重试", error)); + // 登录后的服务端快照是实体基线；基线完成前不开放工作区写操作。 + // 拉取失败时保留本地缓存供只读降级，但远端写入口会明确拒绝，不能把旧缓存上传成真相。 + await syncRemoteUserData(payload.user.id).catch((error) => console.warn("登录后云端数据基线建立失败，已停止远端写入", error)); } else resetRemoteUserDataSync(); } finally { useUserStore.getState().setHydrated(true); diff --git a/web/src/services/canvas-generation-consumer.ts b/web/src/services/canvas-generation-consumer.ts index 1ead172d7..e3b537dee 100644 --- a/web/src/services/canvas-generation-consumer.ts +++ b/web/src/services/canvas-generation-consumer.ts @@ -15,6 +15,7 @@ import { rebasePendingCanvasStorePersistenceAfterGenerationCommitLocked, reconcileCanvasGenerationFailure, recordCanvasStorageDocument, + registerCanvasGenerationPersistenceAttempt, useCanvasStore, withCanvasStorePersistenceLock, withCanvasStorePersistenceSuppressed, @@ -445,9 +446,17 @@ export async function persistCanvasGenerationEffect(input: CanvasGenerationEffec const delta = generationProjectDelta(input, memoryProject); if (!delta.stamped) throw new Error("生成副作用缺少持久幂等标记"); - return withCanvasStorePersistenceLock( - scope, - async () => { + const unregisterAttempt = registerCanvasGenerationPersistenceAttempt(scope, input.projectId, input.effectKey, { + previousNodes: input.previousNodes, + nodes: input.nodes, + previousChatSessions: input.previousChatSessions, + chatSessions: input.chatSessions, + }); + + try { + return await withCanvasStorePersistenceLock( + scope, + async () => { let latestDurable: ReturnType | undefined; let generationCommittedProject: CanvasProject | undefined; let reconcileLiveOnFailure = false; @@ -519,7 +528,10 @@ export async function persistCanvasGenerationEffect(input: CanvasGenerationEffec } throw error; } - }, - { requireCrossRealmLock: true }, - ); + }, + { requireCrossRealmLock: true }, + ); + } finally { + unregisterAttempt(); + } } diff --git a/web/src/services/user-data-sync.ts b/web/src/services/user-data-sync.ts index 926530907..371783ed6 100644 --- a/web/src/services/user-data-sync.ts +++ b/web/src/services/user-data-sync.ts @@ -8,47 +8,46 @@ import type { CanvasProject } from "@/stores/canvas/use-canvas-store"; import { flushCanvasStorePersistence, useCanvasStore } from "@/stores/canvas/use-canvas-store"; let activeRemoteUserId = ""; -let applyingRemoteState = false; +type RemoteUserDataPhase = "inactive" | "hydrating" | "ready" | "failed"; + +let remoteUserDataPhase: RemoteUserDataPhase = "inactive"; let syncTimer: number | null = null; let syncPromise: Promise | null = null; let syncQueued = false; -let syncPauseCount = 0; -let syncPauseTail: Promise = Promise.resolve(); -let resumeSync: (() => void) | null = null; -let syncResumed: Promise | null = null; -const activeRemoteSyncOperations = new Set>(); +let remoteOperationTail: Promise = Promise.resolve(); let subscriptionsInstalled = false; -let remoteAssetVersions = new Map(); -let remoteProjectVersions = new Map(); +let acknowledgedAssets = new Map(); +let acknowledgedProjects = new Map(); const LOCAL_STORAGE_KEY_PATTERN = /^(image|video|audio|file|video-reference|audio-reference):/; export async function syncRemoteUserData(userId?: string | null) { - await runRemoteUserDataSyncOperation(async () => { + await withRemoteUserDataSyncExclusive(async () => { activeRemoteUserId = userId || ""; - if (!activeRemoteUserId) return; - applyingRemoteState = true; + acknowledgedProjects.clear(); + acknowledgedAssets.clear(); + if (!activeRemoteUserId) { + remoteUserDataPhase = "inactive"; + return; + } + remoteUserDataPhase = "hydrating"; try { // 登录只拉一次聚合快照。摘要列表再逐条请求详情会把 N 条数据放大成 2N+2 个请求， // 并且会在登录阶段同时触发大量媒体解析，任何一项失败都会污染登录结果。 const snapshot = await getRemoteUserDataSnapshot(); - remoteProjectVersions = versionMap(snapshot.projects); - remoteAssetVersions = versionMap(snapshot.assets); // 登录时服务端是实体真相。浏览器 IndexedDB 只作为首屏缓存，不能把服务端已删除的记录补回去。 // 这里只替换结构化记录，不在登录阶段解析图片/视频/音频 URL；媒体由实际使用方按需解析。 useCanvasStore.getState().replaceProjects(snapshot.projects); useAssetStore.getState().replaceAssets(snapshot.assets); - } finally { - applyingRemoteState = false; + await Promise.all([flushCanvasStorePersistence(), flushAssetStorePersistence()]); + acknowledgedProjects = new Map(snapshot.projects.map((project) => [project.id, project])); + acknowledgedAssets = new Map(snapshot.assets.map((asset) => [asset.id, asset])); + remoteUserDataPhase = "ready"; + } catch (error) { + remoteUserDataPhase = "failed"; + throw error; } }); - // 首次登录可能带有尚未创建到云端的本地画布；先完成一次 upsert，避免详情页保存/分享先于项目创建。 - try { - await saveRemoteUserDataNow(); - } catch (error) { - console.warn("登录后画布首次同步失败，保留本地项目等待重试", error); - } - scheduleRemoteUserDataSync(); } export function installRemoteUserDataAutoSync() { @@ -64,8 +63,9 @@ export function installRemoteUserDataAutoSync() { export function resetRemoteUserDataSync() { activeRemoteUserId = ""; - remoteAssetVersions.clear(); - remoteProjectVersions.clear(); + remoteUserDataPhase = "inactive"; + acknowledgedAssets.clear(); + acknowledgedProjects.clear(); if (syncTimer) { window.clearTimeout(syncTimer); syncTimer = null; @@ -73,73 +73,17 @@ export function resetRemoteUserDataSync() { syncQueued = false; } -function waitForRemoteUserDataSyncResume() { - if (!syncPauseCount) return Promise.resolve(); - if (!syncResumed) { - syncResumed = new Promise((resolve) => { - resumeSync = resolve; - }); - } - return syncResumed; -} - -async function runRemoteUserDataSyncOperation(operation: () => Promise): Promise { - while (syncPauseCount) await waitForRemoteUserDataSyncResume(); - let finish!: () => void; - const active = new Promise((resolve) => { - finish = resolve; - }); - activeRemoteSyncOperations.add(active); - try { - return await operation(); - } finally { - activeRemoteSyncOperations.delete(active); - finish(); - } -} - -export async function withRemoteUserDataSyncPaused(operation: () => Promise): Promise { - syncPauseCount += 1; - if (syncTimer) { - window.clearTimeout(syncTimer); - syncTimer = null; - syncQueued = true; - } - let releaseTurn!: () => void; - const previousTurn = syncPauseTail; - syncPauseTail = new Promise((resolve) => { - releaseTurn = resolve; - }); - await previousTurn; - try { - if (activeRemoteSyncOperations.size) { - await Promise.allSettled([...activeRemoteSyncOperations]); - } - if (syncPromise) await syncPromise; - return await operation(); - } finally { - releaseTurn(); - syncPauseCount -= 1; - if (!syncPauseCount) { - const resolve = resumeSync; - resumeSync = null; - syncResumed = null; - resolve?.(); - if (syncQueued) scheduleRemoteUserDataSync(); - } - } +export function withRemoteUserDataSyncExclusive(operation: () => Promise): Promise { + const pending = remoteOperationTail.catch(() => undefined).then(operation); + remoteOperationTail = pending.then( + () => undefined, + () => undefined, + ); + return pending; } export function scheduleRemoteUserDataSync() { - if (!activeRemoteUserId || applyingRemoteState) return; - if (syncPauseCount) { - syncQueued = true; - if (syncTimer) { - window.clearTimeout(syncTimer); - syncTimer = null; - } - return; - } + if (!activeRemoteUserId || remoteUserDataPhase !== "ready") return; if (syncPromise) { syncQueued = true; return; @@ -165,12 +109,15 @@ export async function createCanvasProjectWithRemoteSync(title: string, projectId } export async function deleteAssetWithRemoteSync(id: string) { - await runRemoteUserDataSyncOperation(async () => { + const assetId = id.trim(); + if (!assetId) throw new Error("素材 ID 不能为空"); + await withRemoteUserDataSyncExclusive(async () => { if (activeRemoteUserId) { - await deleteRemoteAsset(id); - remoteAssetVersions.delete(id); + requireRemoteUserDataBaseline(); + await deleteRemoteAsset(assetId); + acknowledgedAssets.delete(assetId); } - useAssetStore.getState().removeAsset(id); + await useAssetStore.getState().removeAsset(assetId); await flushAssetStorePersistence(); }); } @@ -178,31 +125,31 @@ export async function deleteAssetWithRemoteSync(id: string) { export async function deleteCanvasProjectsWithRemoteSync(ids: string[]) { const projectIds = [...new Set(ids.map((id) => id.trim()).filter(Boolean))]; if (!projectIds.length) return; - await withRemoteUserDataSyncPaused(async () => { - if (activeRemoteUserId) { - for (const id of projectIds) { + await withRemoteUserDataSyncExclusive(async () => { + if (activeRemoteUserId) requireRemoteUserDataBaseline(); + for (const id of projectIds) { + if (activeRemoteUserId) { await deleteRemoteCanvasProject(id); - remoteProjectVersions.delete(id); + acknowledgedProjects.delete(id); } + useCanvasStore.getState().deleteProjects([id]); + // 批量删除允许部分成功；每个已成功远端删除的实体都立即落实到本地 durable cache。 + await flushCanvasStorePersistence(); } - useCanvasStore.getState().deleteProjects(projectIds); - // 删除完成后再返回，确保刷新不会读到尚未写入 IndexedDB 的旧快照。 - await flushCanvasStorePersistence(); }); } export async function saveRemoteUserDataNow() { if (!activeRemoteUserId) return; - if (syncPauseCount) { - syncQueued = true; - await waitForRemoteUserDataSyncResume(); - return saveRemoteUserDataNow(); - } + requireRemoteUserDataBaseline(); if (syncPromise) { syncQueued = true; return syncPromise; } - syncPromise = drainRemoteUserDataChanges(); + syncPromise = withRemoteUserDataSyncExclusive(async () => { + requireRemoteUserDataBaseline(); + await drainRemoteUserDataChanges(); + }); try { await syncPromise; } finally { @@ -218,53 +165,37 @@ async function drainRemoteUserDataChanges() { } async function saveRemoteUserDataBatch() { - try { - const currentProjects = useCanvasStore.getState().projects; - const currentAssets = useAssetStore.getState().assets; - const dirtyProjects = currentProjects.filter((item) => !sameVersion(remoteProjectVersions.get(item.id), item.updatedAt)); - const dirtyAssets = currentAssets.filter((item) => !sameVersion(remoteAssetVersions.get(item.id), item.updatedAt)); - const deletedProjectIds = missingIds(remoteProjectVersions, currentProjects); - const deletedAssetIds = missingIds(remoteAssetVersions, currentAssets); - if (!dirtyProjects.length && !dirtyAssets.length && !deletedProjectIds.length && !deletedAssetIds.length) return; - const uploaded = new Map(); - const projects = await prepareRemoteCanvasProjects(dirtyProjects, uploaded); - const assets = await prepareRemoteAssets(dirtyAssets, uploaded); - applyingRemoteState = true; - if (projects.length) useCanvasStore.getState().replaceProjects(replaceById(currentProjects, projects)); - if (assets.length) useAssetStore.getState().replaceAssets(replaceById(currentAssets, assets)); - applyingRemoteState = false; - // SQLite 和接口频控都要求写入保持有界；逐项提交还能准确记录已完成版本。 - for (const project of projects) { - await upsertRemoteCanvasProject(project); - remoteProjectVersions.set(project.id, project.updatedAt); - } - for (const asset of assets) { - await upsertRemoteAsset(asset); - remoteAssetVersions.set(asset.id, asset.updatedAt); - } - for (const id of deletedProjectIds) { - await deleteRemoteCanvasProject(id); - remoteProjectVersions.delete(id); - } - for (const id of deletedAssetIds) { - await deleteRemoteAsset(id); - remoteAssetVersions.delete(id); - } - } finally { - applyingRemoteState = false; + const currentProjects = useCanvasStore.getState().projects; + const currentAssets = useAssetStore.getState().assets; + const dirtyProjects = currentProjects.filter((project) => !sameEntitySnapshot(acknowledgedProjects.get(project.id), project)); + const dirtyAssets = currentAssets.filter((asset) => !sameEntitySnapshot(acknowledgedAssets.get(asset.id), asset)); + const currentProjectIds = new Set(currentProjects.map((project) => project.id)); + const currentAssetIds = new Set(currentAssets.map((asset) => asset.id)); + const deletedProjectIds = [...acknowledgedProjects.keys()].filter((id) => !currentProjectIds.has(id)); + const deletedAssetIds = [...acknowledgedAssets.keys()].filter((id) => !currentAssetIds.has(id)); + if (!dirtyProjects.length && !dirtyAssets.length && !deletedProjectIds.length && !deletedAssetIds.length) return; + + const uploaded = new Map(); + // 转换后的 resource: 引用只属于发往服务端的 payload，不能反写整份实时 store。 + // 已确认快照记录的是本次上传所依据的本地实体；上传期间的新编辑会在下一轮继续提交。 + for (const source of dirtyProjects) { + const remotePayload = await ensureRemoteResourceReferences(source, uploaded); + await upsertRemoteCanvasProject(remotePayload); + acknowledgedProjects.set(source.id, source); + } + for (const source of dirtyAssets) { + const remotePayload = await ensureRemoteResourceReferences(source, uploaded); + await upsertRemoteAsset(remotePayload); + acknowledgedAssets.set(source.id, source); + } + for (const id of deletedProjectIds) { + await deleteRemoteCanvasProject(id); + acknowledgedProjects.delete(id); + } + for (const id of deletedAssetIds) { + await deleteRemoteAsset(id); + acknowledgedAssets.delete(id); } -} - -async function prepareRemoteAssets(assets: Asset[], uploaded: Map) { - const result: Asset[] = []; - for (const asset of assets) result.push(await ensureRemoteResourceReferences(asset, uploaded)); - return result; -} - -async function prepareRemoteCanvasProjects(projects: CanvasProject[], uploaded: Map) { - const result: CanvasProject[] = []; - for (const project of projects) result.push(await ensureRemoteResourceReferences(project, uploaded)); - return result; } async function ensureRemoteResourceReferences(value: T, uploaded = new Map()): Promise { @@ -287,13 +218,12 @@ async function ensureRemoteResourceReferences(value: T, uploaded = new Map ""); - return (resourceStorage ? applyResourceReference(next, resourceStorage) : next) as T; + const resourceStorage = await uploadInlineDataUrl(inline); + return applyResourceReference(next, resourceStorage) as T; } const cached = uploaded.get(storageKey); - const resourceStorage = cached || (await uploadLocalStorageKey(storageKey, next).catch(() => "")); - if (!resourceStorage) return next as T; + const resourceStorage = cached || (await uploadLocalStorageKey(storageKey, next)); uploaded.set(storageKey, resourceStorage); return applyResourceReference(next, resourceStorage) as T; } @@ -316,7 +246,9 @@ function inlineMediaDataUrl(payload: Record) { } async function uploadInlineDataUrl(dataUrl: string) { - const blob = await (await fetch(dataUrl)).blob(); + const response = await fetch(dataUrl); + if (!response.ok) throw new Error("内嵌媒体读取失败"); + const blob = await response.blob(); const kind: "image" | "video" | "audio" | "file" = blob.type.startsWith("image/") ? "image" : blob.type.startsWith("video/") ? "video" : blob.type.startsWith("audio/") ? "audio" : "file"; const resource = await uploadResourceFile(blob, kind); return resourceStorageKey(resource.id); @@ -324,7 +256,7 @@ async function uploadInlineDataUrl(dataUrl: string) { async function uploadLocalStorageKey(storageKey: string, payload: Record) { const blob = storageKey.startsWith("image:") ? await getImageBlob(storageKey) : await getMediaBlob(storageKey); - if (!blob) return ""; + if (!blob) throw new Error(`本地媒体不存在：${storageKey}`); const kind = blob.type.startsWith("image/") ? "image" : blob.type.startsWith("video/") ? "video" : blob.type.startsWith("audio/") ? "audio" : "file"; const resource = await uploadResourceFile(blob, kind, { width: numberValue(payload.naturalWidth) || numberValue(payload.width), @@ -334,27 +266,12 @@ async function uploadLocalStorageKey(storageKey: string, payload: Record) { - return new Map(items.map((item) => [item.id, item.updatedAt || ""])); +function requireRemoteUserDataBaseline() { + if (remoteUserDataPhase !== "ready") throw new Error("云端数据基线尚未建立，已停止写入"); } -function missingIds(remote: Map, local: T[]) { - const localIds = new Set(local.map((item) => item.id)); - return Array.from(remote.keys()).filter((id) => !localIds.has(id)); -} - -function replaceById(current: T[], changed: T[]) { - const changedById = new Map(changed.map((item) => [item.id, item])); - return current.map((item) => changedById.get(item.id) || item); -} - -function timeValue(value?: string) { - const time = value ? Date.parse(value) : 0; - return Number.isFinite(time) ? time : 0; -} - -function sameVersion(remote?: string, local?: string) { - return Boolean(remote && local) && timeValue(remote) === timeValue(local); +function sameEntitySnapshot(acknowledged: T | undefined, current: T) { + return acknowledged !== undefined && (acknowledged === current || JSON.stringify(acknowledged) === JSON.stringify(current)); } function isLocalStorageKey(value: string) { diff --git a/web/src/stores/canvas/use-canvas-store.ts b/web/src/stores/canvas/use-canvas-store.ts index 3c64293dd..a720c04df 100644 --- a/web/src/stores/canvas/use-canvas-store.ts +++ b/web/src/stores/canvas/use-canvas-store.ts @@ -62,6 +62,13 @@ const observedCanvasPersists = new Map(); const queuedCanvasPersists = new Map(); const canvasSaveTimers = new Map>(); const canvasPersistTokens = new Map(); +type CanvasGenerationPersistenceAttempt = { + previousNodes?: CanvasNodeData[]; + nodes?: CanvasNodeData[]; + previousChatSessions?: CanvasAssistantSession[]; + chatSessions?: CanvasAssistantSession[]; +}; +const pendingCanvasGenerationAttempts = new Map>(); type AsyncCanvasStorageLock = { request(name: string, callback: () => Promise): Promise; @@ -162,7 +169,7 @@ export function rebasePendingCanvasStorePersistenceAfterGenerationCommitLocked(s // Dedicated commit 已确认 generation stamp；用同 scope 最新内存重新投影队列，避免同节点普通编辑被旧 durable 快照替换。 queued.baseProjects = committed.state.projects; queued.baseRevision = committed.storageRevision; - queued.state = ordinaryCanvasPersistenceState({ projects: latestMemory }, committed.state.projects); + queued.state = ordinaryCanvasPersistenceState(scope, { projects: latestMemory }, committed.state.projects); } export function withCanvasStorePersistenceSuppressed(operation: () => T) { @@ -174,11 +181,81 @@ export function withCanvasStorePersistenceSuppressed(operation: () => T) { } } +export function registerCanvasGenerationPersistenceAttempt(scope: string, projectId: string, effectKey: string, attempt: CanvasGenerationPersistenceAttempt) { + const projectKey = `${scope}\0${projectId}`; + const attempts = pendingCanvasGenerationAttempts.get(projectKey) ?? new Map(); + attempts.set(effectKey, attempt); + pendingCanvasGenerationAttempts.set(projectKey, attempts); + const queued = queuedCanvasPersists.get(scope); + if (queued) { + const latestMemory = canvasMemoryStates.get(scope)?.projects ?? queued.state.projects; + const durableProjects = observedCanvasPersists.get(scope)?.projects ?? queued.baseProjects; + queued.state = ordinaryCanvasPersistenceState(scope, { projects: latestMemory }, durableProjects); + } + return () => { + if (attempts.get(effectKey) !== attempt) return; + attempts.delete(effectKey); + if (!attempts.size) pendingCanvasGenerationAttempts.delete(projectKey); + }; +} + function hasUnconfirmedGenerationKey(localKeys?: string[], durableKeys?: string[]) { return Boolean(localKeys?.some((key) => !durableKeys?.includes(key))); } -function ordinaryCanvasProjectSnapshot(project: CanvasProject, durableProject: CanvasProject | undefined) { +function samePersistenceValue(left: unknown, right: unknown) { + return left === right || JSON.stringify(left) === JSON.stringify(right); +} + +function rollbackGenerationValue(previous: unknown, attempted: unknown, live: unknown, durable: unknown): unknown { + if (samePersistenceValue(previous, attempted)) return live; + if (samePersistenceValue(live, attempted)) return durable; + if (!previous || !attempted || !live || !durable || Array.isArray(previous) || Array.isArray(attempted) || Array.isArray(live) || Array.isArray(durable) || typeof previous !== "object" || typeof attempted !== "object" || typeof live !== "object" || typeof durable !== "object") return live; + + const previousRecord = previous as Record; + const attemptedRecord = attempted as Record; + const liveRecord = live as Record; + const durableRecord = durable as Record; + const result: Record = { ...durableRecord }; + for (const key of new Set([...Object.keys(previousRecord), ...Object.keys(attemptedRecord), ...Object.keys(liveRecord)])) { + const previousHasKey = Object.hasOwn(previousRecord, key); + const attemptedHasKey = Object.hasOwn(attemptedRecord, key); + const liveHasKey = Object.hasOwn(liveRecord, key); + if (!previousHasKey && !attemptedHasKey) { + if (liveHasKey) result[key] = liveRecord[key]; + continue; + } + if (!previousHasKey && attemptedHasKey) { + if (!liveHasKey || samePersistenceValue(liveRecord[key], attemptedRecord[key])) delete result[key]; + else result[key] = liveRecord[key]; + continue; + } + if (previousHasKey && !attemptedHasKey) { + if (liveHasKey) result[key] = liveRecord[key]; + continue; + } + if (!liveHasKey) { + delete result[key]; + continue; + } + const value = rollbackGenerationValue(previousRecord[key], attemptedRecord[key], liveRecord[key], durableRecord[key]); + if (value === undefined) delete result[key]; + else result[key] = value; + } + return result; +} + +function pendingGenerationAttempt(scope: string, projectId: string, effectKeys?: string[]) { + const attempts = pendingCanvasGenerationAttempts.get(`${scope}\0${projectId}`); + if (!attempts) return undefined; + for (const effectKey of effectKeys || []) { + const attempt = attempts.get(effectKey); + if (attempt) return attempt; + } + return undefined; +} + +function ordinaryCanvasProjectSnapshot(scope: string, project: CanvasProject, durableProject: CanvasProject | undefined) { const durableNodes = new Map((durableProject?.nodes || []).map((node) => [node.id, node])); const durableSessions = new Map((durableProject?.chatSessions || []).map((session) => [session.id, session])); let changed = false; @@ -191,7 +268,20 @@ function ordinaryCanvasProjectSnapshot(project: CanvasProject, durableProject: C if (durableProject && hasUnconfirmedGenerationKey(localKeys, durableKeys)) { hasUnconfirmedGeneration = true; changed = true; - if (durableNode) nodes.push(durableNode); + if (durableNode) { + const attempt = pendingGenerationAttempt(scope, project.id, localKeys); + const previousNode = attempt?.previousNodes?.find((candidate) => candidate.id === node.id); + const attemptedNode = attempt?.nodes?.find((candidate) => candidate.id === node.id); + // durableNode comes from the observed storage snapshot. Never mutate that snapshot while + // rebuilding a failed generation, otherwise a later retry observes a partially rolled-back base. + const rolledBack = previousNode && attemptedNode + ? (rollbackGenerationValue(previousNode, attemptedNode, node, durableNode) as CanvasNodeData) + : { ...durableNode, metadata: durableNode.metadata ? { ...durableNode.metadata } : undefined }; + const metadata = { ...(rolledBack.metadata || {}) }; + if (durableKeys?.length) metadata.generationEffectKeys = [...durableKeys]; + else delete metadata.generationEffectKeys; + nodes.push({ ...rolledBack, metadata }); + } continue; } if (JSON.stringify(localKeys) === JSON.stringify(durableKeys)) { @@ -211,7 +301,17 @@ function ordinaryCanvasProjectSnapshot(project: CanvasProject, durableProject: C if (durableProject && hasUnconfirmedGenerationKey(session.generationEffectKeys, durableKeys)) { hasUnconfirmedGeneration = true; changed = true; - if (durableSession) chatSessions.push(durableSession); + if (durableSession) { + const attempt = pendingGenerationAttempt(scope, project.id, session.generationEffectKeys); + const previousSession = attempt?.previousChatSessions?.find((candidate) => candidate.id === session.id); + const attemptedSession = attempt?.chatSessions?.find((candidate) => candidate.id === session.id); + const rolledBack = previousSession && attemptedSession + ? (rollbackGenerationValue(previousSession, attemptedSession, session, durableSession) as CanvasAssistantSession) + : { ...durableSession, generationEffectKeys: durableSession.generationEffectKeys ? [...durableSession.generationEffectKeys] : undefined }; + if (durableKeys?.length) rolledBack.generationEffectKeys = [...durableKeys]; + else delete rolledBack.generationEffectKeys; + chatSessions.push(rolledBack); + } continue; } if (JSON.stringify(session.generationEffectKeys) === JSON.stringify(durableKeys)) { @@ -237,16 +337,16 @@ function ordinaryCanvasProjectSnapshot(project: CanvasProject, durableProject: C return changed ? { ...project, nodes, chatSessions } : project; } -function ordinaryCanvasPersistenceState(state: PersistedCanvasState, durableProjects: CanvasProject[]) { +function ordinaryCanvasPersistenceState(scope: string, state: PersistedCanvasState, durableProjects: CanvasProject[]) { const durableById = new Map(durableProjects.map((project) => [project.id, project])); return { - projects: state.projects.map((project) => ordinaryCanvasProjectSnapshot(project, durableById.get(project.id))), + projects: state.projects.map((project) => ordinaryCanvasProjectSnapshot(scope, project, durableById.get(project.id))), }; } export function reconcileCanvasGenerationFailure(scope: string, durableProjects: CanvasProject[]) { if (getActiveUserScope() !== scope) return; - const projects = ordinaryCanvasPersistenceState({ projects: useCanvasStore.getState().projects }, durableProjects).projects; + const projects = ordinaryCanvasPersistenceState(scope, { projects: useCanvasStore.getState().projects }, durableProjects).projects; withCanvasStorePersistenceSuppressed(() => { useCanvasStore.setState({ projects }); }); @@ -285,7 +385,7 @@ const canvasStorage: PersistStorage = { baseProjects, baseRevision: queued?.baseRevision ?? observed?.revision ?? 0, // generationEffectKeys 由专用 generation durable commit 管理；普通队列只能携带已确认 stamp。 - state: ordinaryCanvasPersistenceState(nextState, observed?.projects ?? baseProjects), + state: ordinaryCanvasPersistenceState(scope, nextState, observed?.projects ?? baseProjects), token, }); clearCanvasSaveTimer(scope); diff --git a/web/src/stores/use-asset-store.ts b/web/src/stores/use-asset-store.ts index f5cd8b627..170026312 100644 --- a/web/src/stores/use-asset-store.ts +++ b/web/src/stores/use-asset-store.ts @@ -52,9 +52,9 @@ type AssetStore = { addAsset: (asset: NewAsset) => string; addGenerationAsset: (effectKey: string, asset: NewAsset, signal?: AbortSignal) => Promise; updateAsset: (id: string, patch: Partial>) => void; - removeAsset: (id: string) => void; + removeAsset: (id: string) => Promise; replaceAssets: (assets: Asset[]) => void; - cleanupImages: (extra?: unknown) => void; + cleanupImages: (extra?: unknown) => Promise; }; export const ASSET_STORE_KEY = "infinite-canvas:asset_store"; @@ -355,42 +355,52 @@ export const useAssetStore = create()( set((state) => ({ assets: state.assets.map((asset) => (asset.id === id ? ({ ...asset, ...patch, updatedAt: new Date().toISOString() } as Asset) : asset)), })), - removeAsset: (id) => + removeAsset: async (id) => { + let remainingAssets: Asset[] = []; + let removedAsset: Asset | undefined; set((state) => { + removedAsset = state.assets.find((asset) => asset.id === id); const assets = state.assets.filter((asset) => asset.id !== id); - get().cleanupImages({ assets }); + remainingAssets = assets; return { assets }; - }), + }); + // 没有本地媒体定位时没有需要由该删除动作回收的 Blob；跳过全库扫描， + // 避免纯文本/远程资源删除依赖浏览器 IndexedDB 驱动。 + if (!removedAsset || (!collectImageStorageKeys(removedAsset).size && !collectMediaStorageKeys(removedAsset).size)) return; + await get().cleanupImages({ assets: remainingAssets }); + }, replaceAssets: (assets) => set({ assets }), - cleanupImages: (extra) => { + cleanupImages: async (extra) => { const scope = getActiveUserScope(); const frozenExtraImageKeys = collectImageStorageKeys(extra); const frozenExtraMediaKeys = collectMediaStorageKeys(extra); - window.setTimeout(async () => { - await withGenerationArtifactCommitLock(scope, async () => { - // 固定锁序：artifact -> Canvas（释放）-> Asset，避免跨 store 锁重入。 - const canvasProjects = await withCanvasStorePersistenceLock(scope, async () => { - await commitPendingCanvasStorePersistenceLocked(scope); - const durableCanvas = parseCanvasStorageDocument(await localForageStorageForScope(scope).getItem(CANVAS_STORE_KEY)); - return pendingCanvasStorePersistence(scope)?.projects ?? durableCanvas.state.projects; - }); - await withGenerationAssetStorageLock(scope, async () => { - await commitPendingAssetStorePersistenceLocked(scope); - const durableAssets = (await readPersistedAssetDocumentForScope(scope)).state.assets; - const references = { projects: canvasProjects, assets: durableAssets }; - const imageKeys = new Set([...frozenExtraImageKeys, ...collectImageStorageKeys(references)]); - const mediaKeys = new Set([...frozenExtraMediaKeys, ...collectMediaStorageKeys(references)]); - await cleanupUnusedImages( - [...imageKeys].map((storageKey) => ({ storageKey })), - scope, - ); - await cleanupUnusedMedia( - [...mediaKeys].map((storageKey) => ({ storageKey })), - scope, - ); - }); - }); - }, 0); + await new Promise((resolve, reject) => { + window.setTimeout(() => + withGenerationArtifactCommitLock(scope, async () => { + // 固定锁序：artifact -> Canvas（释放）-> Asset，避免跨 store 锁重入。 + const canvasProjects = await withCanvasStorePersistenceLock(scope, async () => { + await commitPendingCanvasStorePersistenceLocked(scope); + const durableCanvas = parseCanvasStorageDocument(await localForageStorageForScope(scope).getItem(CANVAS_STORE_KEY)); + return pendingCanvasStorePersistence(scope)?.projects ?? durableCanvas.state.projects; + }); + await withGenerationAssetStorageLock(scope, async () => { + await commitPendingAssetStorePersistenceLocked(scope); + const durableAssets = (await readPersistedAssetDocumentForScope(scope)).state.assets; + const references = { projects: canvasProjects, assets: durableAssets }; + const imageKeys = new Set([...frozenExtraImageKeys, ...collectImageStorageKeys(references)]); + const mediaKeys = new Set([...frozenExtraMediaKeys, ...collectMediaStorageKeys(references)]); + await cleanupUnusedImages( + [...imageKeys].map((storageKey) => ({ storageKey })), + scope, + ); + await cleanupUnusedMedia( + [...mediaKeys].map((storageKey) => ({ storageKey })), + scope, + ); + }); + }).then(resolve, reject), + ); + }); }, }), { diff --git a/web/test/user-scoped-generation-persistence.test.ts b/web/test/user-scoped-generation-persistence.test.ts index 908f2e21b..7a1f49a33 100644 --- a/web/test/user-scoped-generation-persistence.test.ts +++ b/web/test/user-scoped-generation-persistence.test.ts @@ -4,7 +4,7 @@ import localforage from "localforage"; import { rebaseCanvasProjects } from "../src/lib/canvas/canvas-storage-revision"; import { localForageStorageForScope } from "../src/lib/localforage-storage"; import { getActiveUserScope, setActiveUserScope } from "../src/lib/user-scope"; -import { switchUserStorageScope } from "../src/lib/user-session"; +import { applyUserSession, switchUserStorageScope } from "../src/lib/user-session"; import { canvasCinematicContinuationEntryAdapters } from "../src/components/canvas/canvas-assistant-panel"; import { activeGenerationConsumerController, beginGenerationConsumer, runGenerationConsumer } from "../src/services/generation-consumer-lifecycle"; import { createCanvasGenerationLiveProjectAdapter, persistCanvasCinematicSessionContinuationEffect, persistCanvasGenerationEffect, registerCanvasGenerationLiveProject } from "../src/services/canvas-generation-consumer"; @@ -14,8 +14,9 @@ import { ASSET_STORE_KEY, flushAssetStorePersistence, useAssetStore, type Asset, import { withGenerationAssetStorageLock } from "../src/services/generation-asset-repository"; import { CANVAS_STORE_KEY, flushCanvasStorePersistence, useCanvasStore, withCanvasStorePersistenceLock, withCanvasStorePersistenceSuppressed, type CanvasProject } from "../src/stores/canvas/use-canvas-store"; import { CanvasNodeType, type CanvasAssistantSession, type CanvasConnection, type CanvasNodeData } from "../src/types/canvas"; -import { deleteAssetWithRemoteSync, deleteCanvasProjectsWithRemoteSync, resetRemoteUserDataSync, syncRemoteUserData, withRemoteUserDataSyncPaused } from "../src/services/user-data-sync"; +import { deleteAssetWithRemoteSync, deleteCanvasProjectsWithRemoteSync, installRemoteUserDataAutoSync, resetRemoteUserDataSync, saveRemoteUserDataNow, syncRemoteUserData, withRemoteUserDataSyncExclusive } from "../src/services/user-data-sync"; import { apiClient } from "../src/services/api/request"; +import { useUserStore } from "../src/stores/use-user-store"; test("creation recovery observes streaming text tasks after reload", () => { const conversations = [{ @@ -271,7 +272,7 @@ test("ordinary Asset add update and remove survive a failed catalog flush and re failWrites = true; useAssetStore.getState().updateAsset("asset-keep", { title: "A1" }); const addedId = useAssetStore.getState().addAsset(generatedAsset("new asset")); - useAssetStore.getState().removeAsset("asset-remove"); + void useAssetStore.getState().removeAsset("asset-remove"); await expect(flushAssetStorePersistence()).rejects.toThrow("asset catalog write failed"); const afterFailure = JSON.parse(durableValues.get(key)!) as { state: { assets: Asset[] }; storageRevision: number }; expect(afterFailure.state.assets.map((asset) => asset.id).sort()).toEqual(["asset-keep", "asset-remove"]); @@ -2065,6 +2066,87 @@ test("dedicated Canvas conflict reconciles the mounted live page generation stat } }); +test("failed Canvas generation rolls back its fields without losing a concurrent edit on the same node", async () => { + const originalWindow = (globalThis as { window?: unknown }).window; + const originalGetItem = localforage.getItem.bind(localforage); + const originalSetItem = localforage.setItem.bind(localforage); + const previousScope = getActiveUserScope(); + const previousProjects = useCanvasStore.getState().projects; + const values = new Map(); + const localStorageValues = new Map(); + Object.defineProperty(globalThis, "window", { + configurable: true, + value: { + localStorage: { + getItem: (key: string) => localStorageValues.get(key) ?? null, + setItem: (key: string, value: string) => localStorageValues.set(key, value), + removeItem: (key: string) => localStorageValues.delete(key), + }, + }, + }); + localforage.getItem = (async (key: string) => values.get(key) ?? null) as typeof localforage.getItem; + localforage.setItem = (async (key: string, value: string) => { + values.set(key, value); + return value; + }) as typeof localforage.setItem; + + const scope = "canvas-generation-same-node-edit"; + const storageKey = `${CANVAS_STORE_KEY}:user:${scope}`; + const effectKey = "canvas-effect:same-node-edit"; + const node = (title: string, content: string, stamped = false): CanvasNodeData => ({ + id: "node-same-edit", + type: CanvasNodeType.Text, + title, + position: { x: 0, y: 0 }, + width: 320, + height: 180, + metadata: { content, ...(stamped ? { generationEffectKeys: [effectKey] } : {}) }, + }); + const project = (currentNode: CanvasNodeData): CanvasProject => ({ ...storedCanvasProject("canvas-same-node-edit", "same node edit"), nodes: [currentNode] }); + + try { + setActiveUserScope(scope); + const previousNode = node("原始标题", "原始内容"); + const attemptedNode = node("原始标题", "生成内容", true); + useCanvasStore.setState({ projects: [project(previousNode)] }); + await flushCanvasStorePersistence(); + + localforage.setItem = (async (key: string, value: string) => { + if (value.includes(effectKey)) throw new Error("forced generation durable failure"); + values.set(key, value); + return value; + }) as typeof localforage.setItem; + useCanvasStore.getState().updateProject("canvas-same-node-edit", { + nodes: [{ ...attemptedNode, title: "生成期间修改的标题" }], + }); + + await expect( + persistCanvasGenerationEffect({ + projectId: "canvas-same-node-edit", + effectKey, + previousNodes: [previousNode], + nodes: [attemptedNode], + }), + ).rejects.toThrow("forced generation durable failure"); + + const liveNode = useCanvasStore.getState().projects[0]?.nodes[0]; + expect(liveNode?.title).toBe("生成期间修改的标题"); + expect(liveNode?.metadata?.content).toBe("原始内容"); + expect(liveNode?.metadata?.generationEffectKeys).toBeUndefined(); + const durable = JSON.parse(values.get(storageKey)!) as { state: { projects: CanvasProject[] } }; + expect(durable.state.projects[0]?.nodes[0]?.title).toBe("生成期间修改的标题"); + expect(durable.state.projects[0]?.nodes[0]?.metadata?.content).toBe("原始内容"); + } finally { + setActiveUserScope(previousScope); + withCanvasStorePersistenceSuppressed(() => useCanvasStore.setState({ projects: previousProjects })); + await flushCanvasStorePersistence(); + localforage.getItem = originalGetItem; + localforage.setItem = originalSetItem; + if (originalWindow === undefined) delete (globalThis as { window?: unknown }).window; + else Object.defineProperty(globalThis, "window", { configurable: true, value: originalWindow }); + } +}); + test("cinematic continuation durable ack failure restores mounted pending sessions across all production entry modes", async () => { const originalWindow = (globalThis as { window?: unknown }).window; const originalGetItem = localforage.getItem.bind(localforage); @@ -3953,7 +4035,7 @@ test("account scope transition can hold remote user-data sync paused for its ful releaseResolve = resolve; }); let entered = false; - const transition = withRemoteUserDataSyncPaused(async () => { + const transition = withRemoteUserDataSyncExclusive(async () => { entered = true; await release; }); @@ -3974,6 +4056,7 @@ test("account scope transition drains an active remote deletion before entering const originalWindow = (globalThis as { window?: unknown }).window; const originalGetItem = localforage.getItem.bind(localforage); const originalSetItem = localforage.setItem.bind(localforage); + const originalCreateInstance = localforage.createInstance.bind(localforage); const indexedValues = new Map(); const localStorageValues = new Map(); const previousAdapter = apiClient.defaults.adapter; @@ -3998,7 +4081,10 @@ test("account scope transition drains an active remote deletion before entering Object.defineProperty(globalThis, "window", { configurable: true, value: { - setTimeout: () => 1, + setTimeout: (handler: () => void, delay = 0) => { + if (delay === 0) queueMicrotask(handler); + return 1; + }, clearTimeout: () => undefined, localStorage: { getItem: (key: string) => localStorageValues.get(key) ?? null, @@ -4012,6 +4098,12 @@ test("account scope transition drains an active remote deletion before entering indexedValues.set(key, value); return value; }) as typeof localforage.setItem; + localforage.createInstance = (() => ({ + iterate: async () => undefined, + removeItem: async () => undefined, + getItem: async () => null, + setItem: async (_key: string, value: unknown) => value, + })) as typeof localforage.createInstance; try { useAssetStore.getState().replaceAssets([]); await syncRemoteUserData("account-A"); @@ -4034,7 +4126,7 @@ test("account scope transition drains an active remote deletion before entering await new Promise((resolve) => queueMicrotask(resolve)); expect(deleteStarted).toBe(true); let transitionEntered = false; - const transition = withRemoteUserDataSyncPaused(async () => { + const transition = withRemoteUserDataSyncExclusive(async () => { transitionEntered = true; }); await new Promise((resolve) => queueMicrotask(resolve)); @@ -4052,6 +4144,7 @@ test("account scope transition drains an active remote deletion before entering await flushAssetStorePersistence(); localforage.getItem = originalGetItem; localforage.setItem = originalSetItem; + localforage.createInstance = originalCreateInstance; apiClient.defaults.adapter = previousAdapter; if (originalWindow === undefined) delete (globalThis as { window?: unknown }).window; else Object.defineProperty(globalThis, "window", { configurable: true, value: originalWindow }); @@ -4149,3 +4242,232 @@ test("login replaces stale local entities instead of resurrecting remote deletio else Object.defineProperty(globalThis, "window", { configurable: true, value: originalWindow }); } }); + +test("remote resource preparation never overwrites an edit made while upload is in flight", async () => { + const originalWindow = (globalThis as { window?: unknown }).window; + const originalGetItem = localforage.getItem.bind(localforage); + const originalSetItem = localforage.setItem.bind(localforage); + const previousAdapter = apiClient.defaults.adapter; + const previousAssets = useAssetStore.getState().assets; + const indexedValues = new Map(); + const remoteWrites: Asset[] = []; + let releaseUpload!: () => void; + const uploadReleased = new Promise((resolve) => { + releaseUpload = resolve; + }); + let uploadStartedResolve!: () => void; + const uploadStarted = new Promise((resolve) => { + uploadStartedResolve = resolve; + }); + + Object.defineProperty(globalThis, "window", { + configurable: true, + value: { + setTimeout: () => 1, + clearTimeout: () => undefined, + localStorage: { getItem: () => null, setItem: () => undefined, removeItem: () => undefined }, + }, + }); + localforage.getItem = (async (key: string) => indexedValues.get(key) ?? null) as typeof localforage.getItem; + localforage.setItem = (async (key: string, value: string) => { + indexedValues.set(key, value); + return value; + }) as typeof localforage.setItem; + apiClient.defaults.adapter = async (config) => { + const url = String(config.url || ""); + const method = String(config.method || "get").toLowerCase(); + if (url.includes("user-data/snapshot")) { + return { data: { code: 0, data: { projects: [], assets: [] }, msg: "" }, status: 200, statusText: "OK", headers: {}, config }; + } + if (method === "post" && url === "/resources") { + uploadStartedResolve(); + await uploadReleased; + return { + data: { + code: 0, + data: { + resource: { + id: "resource-uploaded", + userId: "account-sync-edit", + kind: "image", + status: "ready", + provider: "local", + endpoint: "", + bucket: "", + objectKey: "users/account-sync-edit/image.png", + publicUrl: "", + mimeType: "image/png", + size: 1, + createdAt: "2026-08-25T00:00:00.000Z", + updatedAt: "2026-08-25T00:00:00.000Z", + }, + }, + msg: "", + }, + status: 200, + statusText: "OK", + headers: {}, + config, + }; + } + if (method === "put" && url.startsWith("/assets/")) { + const body = typeof config.data === "string" ? JSON.parse(config.data) : config.data; + remoteWrites.push(body.asset as Asset); + return { data: { code: 0, data: { asset: body.asset }, msg: "" }, status: 200, statusText: "OK", headers: {}, config }; + } + throw new Error(`unexpected request: ${method} ${url}`); + }; + + try { + resetRemoteUserDataSync(); + installRemoteUserDataAutoSync(); + await syncRemoteUserData("account-sync-edit"); + const asset = { + ...storedAsset("asset-sync-edit", "before upload"), + coverUrl: "data:image/png;base64,iVBORw0KGgo=", + data: { + dataUrl: "data:image/png;base64,iVBORw0KGgo=", + width: 1, + height: 1, + bytes: 1, + mimeType: "image/png", + }, + } as Asset; + useAssetStore.getState().replaceAssets([asset]); + + const saving = saveRemoteUserDataNow(); + await uploadStarted; + useAssetStore.getState().updateAsset(asset.id, { title: "edited during upload" }); + releaseUpload(); + await saving; + + expect(useAssetStore.getState().assets.find((item) => item.id === asset.id)?.title).toBe("edited during upload"); + expect(remoteWrites.at(-1)?.title).toBe("edited during upload"); + } finally { + releaseUpload(); + resetRemoteUserDataSync(); + useAssetStore.getState().replaceAssets(previousAssets); + await flushAssetStorePersistence(); + localforage.getItem = originalGetItem; + localforage.setItem = originalSetItem; + apiClient.defaults.adapter = previousAdapter; + if (originalWindow === undefined) delete (globalThis as { window?: unknown }).window; + else Object.defineProperty(globalThis, "window", { configurable: true, value: originalWindow }); + } +}); + +test("failed remote baseline cannot upload stale local cache", async () => { + const originalWindow = (globalThis as { window?: unknown }).window; + const previousAdapter = apiClient.defaults.adapter; + const previousProjects = useCanvasStore.getState().projects; + const requests: string[] = []; + Object.defineProperty(globalThis, "window", { + configurable: true, + value: { + setTimeout: () => 1, + clearTimeout: () => undefined, + localStorage: { getItem: () => null, setItem: () => undefined, removeItem: () => undefined }, + }, + }); + apiClient.defaults.adapter = async (config) => { + const request = `${String(config.method || "get").toLowerCase()} ${String(config.url || "")}`; + requests.push(request); + if (String(config.url || "").includes("user-data/snapshot")) throw new Error("snapshot unavailable"); + return { data: { code: 0, data: {}, msg: "" }, status: 200, statusText: "OK", headers: {}, config }; + }; + + try { + resetRemoteUserDataSync(); + useCanvasStore.setState({ projects: [storedCanvasProject("stale-local-project", "stale cache")] }); + await expect(syncRemoteUserData("account-baseline-failed")).rejects.toThrow("snapshot unavailable"); + useCanvasStore.getState().renameProject("stale-local-project", "edited without baseline"); + + await expect(saveRemoteUserDataNow()).rejects.toThrow("云端数据基线尚未建立"); + expect(requests.some((request) => request.startsWith("put "))).toBe(false); + } finally { + resetRemoteUserDataSync(); + useCanvasStore.setState({ projects: previousProjects }); + apiClient.defaults.adapter = previousAdapter; + if (originalWindow === undefined) delete (globalThis as { window?: unknown }).window; + else Object.defineProperty(globalThis, "window", { configurable: true, value: originalWindow }); + } +}); + +test("user session stays unhydrated until the remote baseline is durable", async () => { + const originalWindow = (globalThis as { window?: unknown }).window; + const originalGetItem = localforage.getItem.bind(localforage); + const originalSetItem = localforage.setItem.bind(localforage); + const previousAdapter = apiClient.defaults.adapter; + const previousUserState = useUserStore.getState(); + const localValues = new Map(); + let releaseSnapshot!: () => void; + const snapshotReleased = new Promise((resolve) => { + releaseSnapshot = resolve; + }); + let snapshotStartedResolve!: () => void; + const snapshotStarted = new Promise((resolve) => { + snapshotStartedResolve = resolve; + }); + const localStorageValues = new Map(); + Object.defineProperty(globalThis, "window", { + configurable: true, + value: { + setTimeout: () => 1, + clearTimeout: () => undefined, + localStorage: { + getItem: (key: string) => localStorageValues.get(key) ?? null, + setItem: (key: string, value: string) => localStorageValues.set(key, value), + removeItem: (key: string) => localStorageValues.delete(key), + }, + }, + }); + localforage.getItem = (async (key: string) => localValues.get(key) ?? null) as typeof localforage.getItem; + localforage.setItem = (async (key: string, value: string) => { + localValues.set(key, value); + return value; + }) as typeof localforage.setItem; + apiClient.defaults.adapter = async (config) => { + const url = String(config.url || ""); + if (url === "/model-catalog") { + return { data: { code: 0, data: { source: "frontend", models: [] }, msg: "" }, status: 200, statusText: "OK", headers: {}, config }; + } + if (url.includes("user-data/snapshot")) { + snapshotStartedResolve(); + await snapshotReleased; + return { data: { code: 0, data: { projects: [], assets: [] }, msg: "" }, status: 200, statusText: "OK", headers: {}, config }; + } + throw new Error(`unexpected request: ${String(config.method || "get")} ${url}`); + }; + + let applying: Promise | undefined; + try { + resetRemoteUserDataSync(); + useUserStore.setState({ user: null, hydrated: true }); + applying = applyUserSession({ + user: { + id: "account-baseline-gate", + username: "baseline-gate", + displayName: "Baseline Gate", + role: "user", + status: "active", + createdAt: "2026-08-25T00:00:00.000Z", + updatedAt: "2026-08-25T00:00:00.000Z", + }, + }); + await snapshotStarted; + expect(useUserStore.getState().hydrated).toBe(false); + releaseSnapshot(); + await applying; + expect(useUserStore.getState().hydrated).toBe(true); + } finally { + releaseSnapshot(); + await applying?.catch(() => undefined); + resetRemoteUserDataSync(); + useUserStore.setState(previousUserState); + localforage.getItem = originalGetItem; + localforage.setItem = originalSetItem; + apiClient.defaults.adapter = previousAdapter; + if (originalWindow === undefined) delete (globalThis as { window?: unknown }).window; + else Object.defineProperty(globalThis, "window", { configurable: true, value: originalWindow }); + } +}); diff --git a/模型接口协议插件化改造调研报告.md b/模型接口协议插件化改造调研报告.md deleted file mode 100644 index 7b17eeff5..000000000 --- a/模型接口协议插件化改造调研报告.md +++ /dev/null @@ -1,1017 +0,0 @@ -# 模型接口协议插件化改造调研报告 - -## 一、现状分析 - -### 1.1 当前架构问题 - -通过代码审查发现以下核心问题： - -#### 问题1: 协议硬编码严重 -在 `backend/internal/service/provider.go` 中,所有协议处理逻辑都通过 `switch-case` 硬编码: - -```go -// 当前实现方式 -if input.Config.InterfaceType == string(model.ChannelInterfaceMiniMaxVideo) { - return runMiniMaxVideoTask(ctx, input) -} -if input.Config.InterfaceType == string(model.ChannelInterfaceVolcengineJiMengVideo) { - return runVolcengineJiMengVideoTask(ctx, input) -} -// ...更多硬编码判断 -``` - -**影响**: -- 每新增一个协议需要修改核心代码 -- 测试回归成本高 -- 代码耦合度高,难以维护 - -#### 问题2: 参数转换逻辑分散 -不同协议的参数转换函数散落在各处: -- `normalizeXAIVideoDuration()` - xAI专用 -- `normalizeMiniMaxDuration()` - MiniMax专用 -- `normalizeSeedanceDuration()` - Seedance专用 -- `normalizeNovitaVideoDuration()` - Novita专用 - -**影响**: -- 重复代码多(duration归一化逻辑至少4个版本) -- 新协议需要重新写一套转换逻辑 -- 缺乏统一的参数映射机制 - -#### 问题3: 轮询逻辑重复 -异步协议的轮询逻辑高度相似但各自实现: - -```go -// MiniMax轮询 -for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { - // 查询状态 - // 判断完成/失败 - // 下载结果 -} - -// Novita轮询 - 几乎相同的逻辑 -for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { - // 查询状态 - // 判断完成/失败 - // 下载结果 -} -``` - -**影响**: -- 代码重复率>70% -- 轮询策略调整需要修改多处 - -#### 问题4: 缺乏协议能力声明 -当前 `ChannelInterfaceType` 枚举定义协议类型,但: -- 没有协议元数据(支持的能力、参数映射规则) -- 无法动态扩展 -- 协议之间的差异性没有结构化描述 - -### 1.2 现有插件系统分析 - -项目已有初步的插件机制(`backend/internal/provider/interface.go`): - -```go -type ModelCatalogPlugin interface { - GetProviderID() string - Match(baseURL string, headers map[string]string) bool - AdditionalModels(config DiscoveryConfig) []Model - GetMetadata() ProviderMetadata -} -``` - -**现状**: -- 仅用于模型目录发现(Model Catalog) -- 未应用到API协议转换层 -- 缺少协议执行接口定义 - -### 1.3 业界最佳实践调研 - -#### A. OpenAI兼容层模式 (One API / New API) -**核心思想**: 统一入口协议 + 后端适配器 - -``` -用户请求(OpenAI格式) - ↓ -协议适配器路由 - ↓ -├─ OpenAI原生 → 直通 -├─ Anthropic → Claude适配器 -├─ Google → Gemini适配器 -└─ 自定义 → 插件适配器 -``` - -**优点**: -- 用户侧统一标准 -- 适配逻辑解耦 - -**缺点**: -- 强制OpenAI协议可能丢失厂商特性 -- 不适合平台内部多协议并存场景 - -#### B. Apache Camel 模式 (企业集成模式) -**核心思想**: 管道 + 转换器 - -``` -Source Protocol → Transformer → Unified Model → Transformer → Target Protocol -``` - -**优点**: -- 高度灵活 -- 支持复杂转换链 - -**缺点**: -- 对于AI API场景过于复杂 -- 性能开销大 - -#### C. Kubernetes CRI/CSI 插件模式 (推荐采用) -**核心思想**: 接口标准化 + 插件实现 - -```go -type RuntimeInterface interface { - CreateContainer(spec ContainerSpec) (Container, error) - // ...标准接口 -} - -// 不同运行时实现该接口 -type DockerRuntime implements RuntimeInterface -type ContainerdRuntime implements RuntimeInterface -``` - -**优点**: -- 接口清晰,易于实现 -- 插件间完全解耦 -- 易于测试和替换 - -**为什么适合我们**: -1. AI模型API操作有限且标准(生成、查询、下载) -2. 需要支持多种协议共存 -3. 需要动态加载能力 - -#### D. Terraform Provider 模式 -**核心思想**: Schema + CRUD + 生命周期管理 - -```go -type Provider struct { - Schema Schema // 参数定义 - ConfigureFunc func(d *ResourceData) error - ResourcesMap map[string]*Resource -} -``` - -**优点**: -- Schema驱动,自动验证 -- 声明式参数映射 - -**适用场景分析**: -- 适合资源管理场景 -- 对我们来说Schema机制可借鉴用于参数映射 - -## 二、插件化改造方案 - -### 2.1 设计原则 - -1. **最小侵入**: 现有代码逐步迁移,支持新旧协议共存 -2. **向后兼容**: 已有的硬编码协议仍可运行 -3. **渐进增强**: 先支持视频协议插件化,再扩展到图片/音频 -4. **开发友好**: 新增协议只需实现接口,不修改核心代码 - -### 2.2 核心架构设计 - -#### 2.2.1 统一内部协议(Internal Protocol Contract) - -定义平台内部标准请求/响应格式: - -```go -// backend/internal/protocol/types.go -package protocol - -// GenerationRequest 统一生成请求 -type GenerationRequest struct { - // 基础字段 - Model string - Prompt string - - // 媒体输入 - Images []MediaInput - Videos []MediaInput - Audios []MediaInput - - // 通用参数(标准化命名) - Duration int // 秒数 - AspectRatio string // 16:9 / 9:16 / 1:1 - Resolution string // 720p / 1080p / 4k - Quality string // low / medium / high - GenerateAudio bool // 是否生成音频 - Watermark bool // 是否添加水印 - - // 扩展参数 - Extra map[string]interface{} -} - -// GenerationResponse 统一生成响应 -type GenerationResponse struct { - TaskID string - Status TaskStatus - Result *GenerationResult - EstimatedTime int // 预计完成时间(秒) - Error *ProtocolError -} - -type GenerationResult struct { - Videos []MediaOutput - Images []MediaOutput - Audios []MediaOutput -} - -type TaskStatus string -const ( - TaskPending TaskStatus = "pending" - TaskProcessing TaskStatus = "processing" - TaskCompleted TaskStatus = "completed" - TaskFailed TaskStatus = "failed" -) -``` - -#### 2.2.2 协议适配器接口 - -```go -// backend/internal/protocol/adapter.go -package protocol - -// ProtocolAdapter 协议适配器接口 -type ProtocolAdapter interface { - // 元数据 - GetMetadata() AdapterMetadata - - // 参数映射 - MapRequest(internal GenerationRequest) (vendorRequest interface{}, err error) - MapResponse(vendorResponse interface{}) (internal GenerationResponse, err error) - - // 协议执行 - Submit(ctx context.Context, request GenerationRequest) (taskID string, err error) - Query(ctx context.Context, taskID string) (response GenerationResponse, err error) - Download(ctx context.Context, taskID string) (data []byte, mimeType string, err error) - - // 能力声明 - Capabilities() ProtocolCapabilities -} - -// AdapterMetadata 适配器元数据 -type AdapterMetadata struct { - ID string // 唯一标识,如 "minimax-video" - Name string // 显示名称 - Version string // 版本号 - Provider string // 提供商,如 "MiniMax" - Description string // 描述 - Author string // 作者 - Capabilities []string // 支持的能力: video, image, audio -} - -// ProtocolCapabilities 协议能力声明 -type ProtocolCapabilities struct { - SupportedModes []string // video, image, audio - AsyncMode bool // 是否异步 - SupportsWebhook bool // 是否支持Webhook - MaxMediaInputs MediaLimits // 媒体输入限制 - ParameterSchema ParameterSchema // 参数Schema -} - -type MediaLimits struct { - MaxImages int - MaxVideos int - MaxAudios int -} - -type ParameterSchema struct { - Duration ParameterDef - AspectRatio ParameterDef - Resolution ParameterDef - Quality ParameterDef - // ... -} - -type ParameterDef struct { - Supported bool - Required bool - DefaultValue string - AllowedValues []string - Mapper ParameterMapper // 参数映射器 -} - -// ParameterMapper 参数映射器 -type ParameterMapper interface { - // 将内部标准值映射为厂商值 - ToVendor(internal string) (vendor interface{}, err error) - // 将厂商值映射为内部标准值 - FromVendor(vendor interface{}) (internal string, err error) -} -``` - -#### 2.2.3 协议注册中心 - -```go -// backend/internal/protocol/registry.go -package protocol - -type Registry struct { - adapters map[string]ProtocolAdapter - mu sync.RWMutex -} - -var globalRegistry = &Registry{ - adapters: make(map[string]ProtocolAdapter), -} - -// Register 注册协议适配器 -func Register(adapter ProtocolAdapter) error { - metadata := adapter.GetMetadata() - if metadata.ID == "" { - return errors.New("adapter ID cannot be empty") - } - - globalRegistry.mu.Lock() - defer globalRegistry.mu.Unlock() - - if _, exists := globalRegistry.adapters[metadata.ID]; exists { - return fmt.Errorf("adapter %s already registered", metadata.ID) - } - - globalRegistry.adapters[metadata.ID] = adapter - return nil -} - -// Get 获取协议适配器 -func Get(adapterID string) (ProtocolAdapter, error) { - globalRegistry.mu.RLock() - defer globalRegistry.mu.RUnlock() - - adapter, exists := globalRegistry.adapters[adapterID] - if !exists { - return nil, fmt.Errorf("adapter %s not found", adapterID) - } - return adapter, nil -} - -// List 列出所有适配器 -func List() []AdapterMetadata { - globalRegistry.mu.RLock() - defer globalRegistry.mu.RUnlock() - - result := make([]AdapterMetadata, 0, len(globalRegistry.adapters)) - for _, adapter := range globalRegistry.adapters { - result = append(result, adapter.GetMetadata()) - } - return result -} -``` - -### 2.3 具体实现示例 - -#### 2.3.1 MiniMax协议适配器实现 - -```go -// backend/internal/protocol/adapters/minimax_video.go -package adapters - -import ( - "context" - "infinite-canvas/backend/internal/protocol" -) - -type MiniMaxVideoAdapter struct { - baseURL string - apiKey string - secretKey string -} - -func NewMiniMaxVideoAdapter() *MiniMaxVideoAdapter { - return &MiniMaxVideoAdapter{} -} - -func (a *MiniMaxVideoAdapter) GetMetadata() protocol.AdapterMetadata { - return protocol.AdapterMetadata{ - ID: "minimax-video-h3", - Name: "MiniMax Video H3", - Version: "1.0.0", - Provider: "MiniMax", - Description: "MiniMax H3 视频生成协议", - Author: "Yingce Platform", - Capabilities: []string{"video"}, - } -} - -func (a *MiniMaxVideoAdapter) Capabilities() protocol.ProtocolCapabilities { - return protocol.ProtocolCapabilities{ - SupportedModes: []string{"video"}, - AsyncMode: true, - SupportsWebhook: false, - MaxMediaInputs: protocol.MediaLimits{ - MaxImages: 9, - MaxVideos: 3, - MaxAudios: 3, - }, - ParameterSchema: protocol.ParameterSchema{ - Duration: protocol.ParameterDef{ - Supported: true, - Required: false, - DefaultValue: "5", - AllowedValues: []string{"4", "5", "6", "10", "15"}, - Mapper: &miniMaxDurationMapper{}, - }, - AspectRatio: protocol.ParameterDef{ - Supported: true, - AllowedValues: []string{"16:9", "9:16", "1:1", "4:3", "3:4", "21:9"}, - Mapper: &directMapper{}, - }, - Resolution: protocol.ParameterDef{ - Supported: true, - AllowedValues: []string{"768p", "1080p", "2k"}, - Mapper: &miniMaxResolutionMapper{}, - }, - }, - } -} - -func (a *MiniMaxVideoAdapter) MapRequest(req protocol.GenerationRequest) (interface{}, error) { - // 构建MiniMax请求格式 - content := []miniMaxVideoContent{{Type: "text", Text: req.Prompt}} - - // 映射参考图片 - for i, img := range req.Images { - role := "reference_image" - if len(req.Videos) == 0 && len(req.Audios) == 0 && len(req.Images) <= 2 { - if i == 0 { - role = "first_frame" - } else if i == 1 { - role = "last_frame" - } - } - content = append(content, miniMaxVideoContent{ - Type: "image_url", - ImageURL: &miniMaxMediaURL{URL: img.URL}, - Role: role, - }) - } - - // 参数映射 - resolution, _ := a.Capabilities().ParameterSchema.Resolution.Mapper.ToVendor(req.Resolution) - duration, _ := a.Capabilities().ParameterSchema.Duration.Mapper.ToVendor(fmt.Sprint(req.Duration)) - aspectRatio, _ := a.Capabilities().ParameterSchema.AspectRatio.Mapper.ToVendor(req.AspectRatio) - - return miniMaxVideoRequest{ - Model: req.Model, - Content: content, - Resolution: resolution.(string), - Duration: duration.(int), - Ratio: aspectRatio.(string), - AIGCWatermark: &req.Watermark, - }, nil -} - -func (a *MiniMaxVideoAdapter) Submit(ctx context.Context, req protocol.GenerationRequest) (string, error) { - vendorReq, err := a.MapRequest(req) - if err != nil { - return "", err - } - - // 调用MiniMax API - var created map[string]interface{} - if err := a.postJSON(ctx, "/v2/video_generation", vendorReq, &created); err != nil { - return "", err - } - - taskID := firstNonEmptyString(stringField(created, "task_id"), stringField(created, "id")) - if taskID == "" { - return "", errors.New("未返回task_id") - } - - return taskID, nil -} - -func (a *MiniMaxVideoAdapter) Query(ctx context.Context, taskID string) (protocol.GenerationResponse, error) { - var response map[string]interface{} - if err := a.getJSON(ctx, "/v2/query/video_generation/"+url.PathEscape(taskID), &response); err != nil { - return protocol.GenerationResponse{}, err - } - - return a.MapResponse(response) -} - -func (a *MiniMaxVideoAdapter) MapResponse(vendorResp interface{}) (protocol.GenerationResponse, error) { - resp := vendorResp.(map[string]interface{}) - task, _ := resp["task"].(map[string]interface{}) - - status := strings.ToLower(stringField(task, "status")) - - result := protocol.GenerationResponse{ - TaskID: stringField(task, "task_id"), - } - - switch status { - case "succeeded", "completed": - result.Status = protocol.TaskCompleted - contentValue, _ := task["content"].(map[string]interface{}) - videoURL := stringField(contentValue, "url") - if videoURL != "" { - result.Result = &protocol.GenerationResult{ - Videos: []protocol.MediaOutput{{URL: videoURL}}, - } - } - case "failed", "cancelled": - result.Status = protocol.TaskFailed - result.Error = &protocol.ProtocolError{ - Message: miniMaxTaskError(task), - } - default: - result.Status = protocol.TaskProcessing - } - - return result, nil -} - -// 参数映射器实现 -type miniMaxDurationMapper struct{} - -func (m *miniMaxDurationMapper) ToVendor(internal string) (interface{}, error) { - duration, err := strconv.Atoi(internal) - if err != nil { - return 5, nil - } - if duration < 4 { - return 4, nil - } - if duration > 15 { - return 15, nil - } - return duration, nil -} - -func (m *miniMaxDurationMapper) FromVendor(vendor interface{}) (string, error) { - return fmt.Sprint(vendor), nil -} - -type miniMaxResolutionMapper struct{} - -func (m *miniMaxResolutionMapper) ToVendor(internal string) (interface{}, error) { - switch strings.ToLower(internal) { - case "2k", "4k", "high", "1080p", "1440p", "2160p": - return "2K", nil - default: - return "768P", nil - } -} - -func (m *miniMaxResolutionMapper) FromVendor(vendor interface{}) (string, error) { - v := strings.ToUpper(fmt.Sprint(vendor)) - if v == "2K" { - return "1080p", nil - } - return "720p", nil -} - -func init() { - // 自动注册 - protocol.Register(NewMiniMaxVideoAdapter()) -} -``` - -### 2.4 插件中心设计 - -#### 2.4.1 插件元数据文件 - -```yaml -# plugins/minimax-video-h3/plugin.yaml -id: minimax-video-h3 -name: MiniMax Video H3 -version: 1.0.0 -provider: MiniMax -author: Yingce Platform -description: MiniMax H3 视频生成协议适配器 - -# 能力分类 -categories: - - video-generation - - text-to-video - - image-to-video - -# 支持的能力 -capabilities: - - video - -# 协议类型 -protocol_type: async_polling - -# 依赖 -dependencies: [] - -# 配置Schema -config_schema: - base_url: - type: string - required: true - description: API基础URL - api_key: - type: string - required: true - secret: true - description: API密钥 - -# 参数映射 -parameter_mappings: - duration: - supported: true - default: "5" - allowed_values: ["4", "5", "6", "10", "15"] - vendor_field: "duration" - vendor_type: integer - - aspect_ratio: - supported: true - allowed_values: ["16:9", "9:16", "1:1", "4:3", "3:4", "21:9", "adaptive"] - vendor_field: "ratio" - vendor_type: string - - resolution: - supported: true - allowed_values: ["768p", "1080p", "2k"] - vendor_field: "resolution" - vendor_type: string - mapping: - "720p": "768P" - "1080p": "2K" - "2k": "2K" - "4k": "2K" - -# 输入限制 -input_limits: - max_images: 9 - max_videos: 3 - max_audios: 3 - max_prompt_length: 2000 - -# 端点配置 -endpoints: - submit: /v2/video_generation - query: /v2/query/video_generation/{task_id} - -# 轮询配置 -polling: - enabled: true - interval: 2500ms - timeout: 30m - -# 状态映射 -status_mapping: - succeeded: completed - completed: completed - failed: failed - cancelled: failed - processing: processing - pending: pending -``` - -#### 2.4.2 插件管理API - -```go -// backend/internal/service/plugin_manager.go -package service - -type PluginManager struct { - pluginDir string - registry *protocol.Registry - db *gorm.DB -} - -// InstallPlugin 安装插件 -func (pm *PluginManager) InstallPlugin(pluginID string, source PluginSource) error { - // 1. 下载插件包 - // 2. 验证签名 - // 3. 解压到pluginDir - // 4. 加载插件元数据 - // 5. 注册到数据库 - // 6. 热加载适配器 -} - -// UninstallPlugin 卸载插件 -func (pm *PluginManager) UninstallPlugin(pluginID string) error { - // 1. 检查是否被使用 - // 2. 从注册中心移除 - // 3. 从数据库删除 - // 4. 删除插件文件 -} - -// ListInstalledPlugins 列出已安装插件 -func (pm *PluginManager) ListInstalledPlugins() ([]PluginInfo, error) { - // 从数据库查询 -} - -// UpdatePlugin 更新插件 -func (pm *PluginManager) UpdatePlugin(pluginID string, newVersion string) error { - // 版本兼容性检查 - // 下载新版本 - // 替换旧版本 - // 热重载 -} -``` - -### 2.5 数据库设计 - -```sql --- 插件表 -CREATE TABLE protocol_plugins ( - id VARCHAR(36) PRIMARY KEY, - plugin_id VARCHAR(100) UNIQUE NOT NULL, -- 如 "minimax-video-h3" - name VARCHAR(100) NOT NULL, - version VARCHAR(20) NOT NULL, - provider VARCHAR(50), - author VARCHAR(100), - description TEXT, - categories JSON, -- ["video-generation", "text-to-video"] - capabilities JSON, -- ["video"] - config_schema JSON, - parameter_mappings JSON, - enabled BOOLEAN DEFAULT true, - installed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP -); - --- 插件使用关联 -CREATE TABLE channel_protocol_plugins ( - id VARCHAR(36) PRIMARY KEY, - channel_id VARCHAR(36) NOT NULL, - plugin_id VARCHAR(100) NOT NULL, -- 关联 protocol_plugins.plugin_id - enabled BOOLEAN DEFAULT true, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (channel_id) REFERENCES model_channels(id), - UNIQUE(channel_id, plugin_id) -); - --- 索引 -CREATE INDEX idx_plugins_enabled ON protocol_plugins(enabled); -CREATE INDEX idx_plugins_categories ON protocol_plugins USING GIN(categories); -CREATE INDEX idx_channel_plugins_channel ON channel_protocol_plugins(channel_id); -``` - -### 2.6 前端UI改造 - -#### 2.6.1 插件中心页面 - -``` -管理后台 > 插件中心 > 协议插件 - -┌────────────────────────────────────────────┐ -│ 协议插件库 │ -├────────────────────────────────────────────┤ -│ 分类筛选: │ -│ [全部] [视频生成] [图片生成] [音频生成] │ -├────────────────────────────────────────────┤ -│ ┌──────────────────────────────────────┐ │ -│ │ MiniMax Video H3 [已安装] │ │ -│ │ 版本: 1.0.0 | 作者: Yingce │ │ -│ │ 支持能力: 文生视频、图生视频 │ │ -│ │ [查看详情] [卸载] [更新] │ │ -│ └──────────────────────────────────────┘ │ -│ │ -│ ┌──────────────────────────────────────┐ │ -│ │ Novita Video [未安装] │ │ -│ │ 版本: 1.0.0 | 作者: Yingce │ │ -│ │ 支持能力: 文生视频、图生视频 │ │ -│ │ [安装] [查看详情] │ │ -│ └──────────────────────────────────────┘ │ -└────────────────────────────────────────────┘ -``` - -#### 2.6.2 渠道模型配置增强 - -``` -系统渠道模型配置 - -模型: MiniMax-video-01 -请求协议: [下拉选择] - ├─ 视频生成协议 - │ ├─ MiniMax Video H3 ← 插件提供 - │ ├─ MiniMax Video (Legacy) - │ └─ OpenAI Videos 兼容 - ├─ 图片生成协议 - │ └─ ... - └─ 音频生成协议 - └─ ... - -[协议参数预览] -支持的参数: 时长、分辨率、宽高比 -输入限制: 最多9张图、3个视频、3个音频 -``` - -### 2.7 兼容性策略 - -#### 2.7.1 渐进迁移 - -**阶段1**: 插件系统与硬编码协议共存 -```go -func runVideoTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) { - // 先尝试插件 - if pluginAdapter, err := protocol.Get(input.Config.InterfaceType); err == nil { - return runPluginBasedVideoTask(ctx, input, pluginAdapter) - } - - // 回退到旧实现 - if input.Config.InterfaceType == string(model.ChannelInterfaceMiniMaxVideo) { - return runMiniMaxVideoTask(ctx, input) // 旧代码保留 - } - // ... -} -``` - -**阶段2**: 将旧协议包装为插件 -```go -// 为旧实现创建适配器包装 -type LegacyMiniMaxAdapter struct { - legacyFunc func(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) -} - -func (a *LegacyMiniMaxAdapter) Submit(ctx context.Context, req protocol.GenerationRequest) (string, error) { - // 转换为旧格式,调用旧函数 - legacyInput := convertToLegacyInput(req) - result, err := a.legacyFunc(ctx, legacyInput) - // 转换回新格式 - return extractTaskID(result), err -} -``` - -**阶段3**: 逐步移除旧代码 - -### 2.8 实施计划 - -#### 第一阶段: 基础设施 (2周) -- [ ] 设计并实现 `protocol.ProtocolAdapter` 接口 -- [ ] 实现协议注册中心 -- [ ] 定义统一内部协议格式 -- [ ] 设计插件元数据Schema -- [ ] 数据库表设计与迁移 - -#### 第二阶段: 首个插件实现 (1周) -- [ ] 选择MiniMax H3作为试点 -- [ ] 实现MiniMaxVideoAdapter -- [ ] 编写参数映射器 -- [ ] 单元测试覆盖 -- [ ] 集成测试 - -#### 第三阶段: 核心服务改造 (1周) -- [ ] 修改 `runVideoTask` 支持插件路由 -- [ ] 实现兼容层(插件优先,旧代码回退) -- [ ] 添加插件加载日志和监控 -- [ ] 性能测试 - -#### 第四阶段: 插件管理后台 (1周) -- [ ] 插件中心页面开发 -- [ ] 插件安装/卸载API -- [ ] 渠道模型配置UI增强 -- [ ] 协议选择下拉框改造 - -#### 第五阶段: 批量迁移 (2周) -- [ ] 将5个现有协议迁移为插件 - - Gemini Veo - - Novita Video - - xAI Video - - Seedance Videos - - 火山方舟视频 -- [ ] 回归测试 -- [ ] 文档更新 - -#### 第六阶段: 扩展到其他模态 (2周) -- [ ] 图片生成协议插件化 -- [ ] 音频生成协议插件化 -- [ ] 文本生成协议插件化 - -### 2.9 风险控制 - -| 风险 | 影响 | 应对措施 | -|------|------|----------| -| 插件bug影响生产 | 高 | 1. 插件沙箱隔离 2. 回退到旧实现 3. 熔断机制 | -| 性能下降 | 中 | 1. 适配器接口调用缓存 2. 参数映射预编译 3. 压测验证 | -| 插件版本冲突 | 中 | 1. 语义化版本管理 2. 兼容性矩阵 3. 自动化测试 | -| 用户配置迁移 | 中 | 1. 自动迁移脚本 2. 兼容性保留3个月 3. 提前通知 | - -## 三、收益评估 - -### 3.1 开发效率提升 - -**当前**: 新增一个协议需要: -1. 修改 `provider.go` (核心服务代码) -2. 添加参数转换函数 -3. 实现请求/轮询/下载逻辑 -4. 修改路由判断 -5. 更新枚举定义 -6. 回归测试所有协议 - -**预计工作量**: 2-3天 - -**插件化后**: 新增一个协议需要: -1. 创建插件目录 -2. 编写 `plugin.yaml` 元数据 -3. 实现 `ProtocolAdapter` 接口 -4. 单元测试插件 -5. 上传到插件中心 - -**预计工作量**: 0.5-1天 - -**效率提升**: 60-70% - -### 3.2 维护成本降低 - -- **代码行数**: 核心服务代码预计减少 40% -- **重复代码**: 消除 70% 的重复轮询/参数转换逻辑 -- **回归测试**: 插件独立测试,不影响其他协议 - -### 3.3 扩展性增强 - -- **社区贡献**: 第三方可贡献插件(需审核) -- **私有协议**: 企业客户可自行开发私有协议插件 -- **快速响应**: 厂商API更新后,可单独更新插件,不需要发版 - -### 3.4 用户体验改进 - -- **配置简化**: 协议选择界面分类清晰 -- **错误提示**: 插件级别的参数校验,错误提示更精准 -- **灵活性**: 用户可选择安装需要的协议,减少界面混乱 - -## 四、技术债务处理 - -### 4.1 参数命名标准化 - -建议统一内部参数命名: - -| 旧命名(不统一) | 新命名(标准) | 说明 | -|---------------|-------------|------| -| videoSeconds / duration / DurationSeconds | duration | 统一用秒数 | -| vquality / resolution / quality | resolution | 统一为分辨率 | -| size / ratio / aspect_ratio | aspectRatio | 统一为宽高比 | -| videoGenerateAudio / generate_audio | generateAudio | 驼峰命名 | - -### 4.2 错误处理标准化 - -当前错误处理分散,建议: - -```go -// protocol/errors.go -type ProtocolError struct { - Code string // 错误码: INVALID_PARAM, QUOTA_EXCEEDED, etc - Message string // 用户友好消息 - VendorCode string // 厂商原始错误码 - VendorMsg string // 厂商原始消息(不暴露给用户) - Retryable bool // 是否可重试 -} - -// 标准错误码 -const ( - ErrInvalidParameter = "INVALID_PARAMETER" - ErrQuotaExceeded = "QUOTA_EXCEEDED" - ErrModelNotFound = "MODEL_NOT_FOUND" - ErrContentBlocked = "CONTENT_BLOCKED" - ErrTimeout = "TIMEOUT" - ErrVendorError = "VENDOR_ERROR" -) -``` - -## 五、总结与建议 - -### 5.1 核心优势 - -1. **架构清晰**: 接口标准化,职责明确 -2. **易于扩展**: 新增协议零侵入核心代码 -3. **降低耦合**: 协议实现完全解耦 -4. **提升质量**: 插件级别独立测试 -5. **用户友好**: 配置界面分类清晰 - -### 5.2 立即行动项 - -**高优先级**: -1. 评审本方案,确定技术路线 -2. 设计详细的接口规范文档 -3. 搭建插件开发脚手架 -4. 选择1-2个协议作为试点 - -**中优先级**: -1. 制定插件审核规范 -2. 建立插件测试框架 -3. 编写插件开发者文档 - -**低优先级**: -1. 插件市场/插件仓库建设 -2. 第三方插件贡献通道 -3. 插件性能监控面板 - -### 5.3 下一步工作 - -1. **本周**: 技术方案评审会议 -2. **下周**: 开始第一阶段开发(基础设施) -3. **两周后**: 完成首个插件试点(MiniMax H3) -4. **一个月后**: 完成核心视频协议迁移 -5. **两个月后**: 扩展到图片/音频协议 - ---- - -**文档版本**: v1.0 -**创建日期**: 2026-08-23 -**作者**: AI Assistant -**状态**: 待评审 diff --git a/视频模型API协议改造方案.md b/视频模型API协议改造方案.md deleted file mode 100644 index 451b541cf..000000000 --- a/视频模型API协议改造方案.md +++ /dev/null @@ -1,473 +0,0 @@ -# 视频模型API协议改造方案 - -## 一、现状分析 - -### 1.1 当前协议混乱情况 - -根据系统截图，当前视频生成能力存在以下协议： - -| 服务商 | 协议名称 | 端点 | 问题 | -|--------|---------|------|------| -| OpenAI/NewAPI | OpenAI / NewAPI Videos | POST /v1/videos | NewAPI重复1 | -| NewAPI | NewAPI 媒体任务 | POST /v1/videos | NewAPI重复2，端点与上面相同 | -| NewAPI | NewAPI Video Generations | POST /v1/video/generations | NewAPI重复3 | -| xAI | xAI 官方视频 | POST /v1/videos/generations | - | -| 火山方舟 | 火山方舟视频 | POST /api/v3/contents/generation... | 路径过长 | -| 即梦 | 即梦官方视频 | POST CVSync2AsyncSubmitTask | RPC风格，非RESTful | -| Gemini | Gemini Veo | POST /v1beta/models/{model}:pred... | Google API风格 | -| Novita | Novita 视频 | POST /v3/video/create | - | -| MiniMax | MiniMax 视频 | POST /v2/video_generation | - | - -**核心问题：** -1. NewAPI同时存在3个不同的协议配置，造成混淆 -2. 端点路径不统一（/v1/videos vs /v1/video/generations vs /v2/video_generation） -3. 命名风格不一致（videos vs video vs video_generation） -4. 部分使用RPC风格，与RESTful体系不兼容 - -### 1.2 用户体验问题 - -- **选择困难**：面对3个NewAPI协议，用户不知道该选哪个 -- **迁移成本高**：不同协议间切换需要修改请求格式 -- **维护复杂**：每个协议需要单独的适配代码 -- **文档混乱**：难以为用户提供清晰的使用文档 - -## 二、主流协议标准调研 - -### 2.1 OpenAI协议标准（事实标准） - -OpenAI作为行业领导者，其API设计已成为事实标准： - -**文本生成：** -``` -POST /v1/chat/completions -POST /v1/completions -``` - -**图片生成：** -``` -POST /v1/images/generations -POST /v1/images/edits -POST /v1/images/variations -``` - -**视频生成（推测标准）：** -``` -POST /v1/videos/generations -``` - -**设计特点：** -- 使用复数名词（images, videos, completions） -- 动作使用名词形式（generations, edits, variations） -- 清晰的版本控制（/v1/） -- 一致的命名规范 - -### 2.2 其他主流厂商协议 - -**Google Gemini风格：** -``` -POST /v1beta/models/{model}:generateContent -POST /v1beta/models/{model}:predict -``` -- 特点：RPC风格，使用冒号分隔资源和动作 -- 适用场景：Google生态内部 - -**Anthropic Claude风格：** -``` -POST /v1/messages -``` -- 特点：极简设计，资源导向 -- 不太适合多模态场景 - -**业界通用RESTful风格：** -``` -POST /api/v{version}/{resource}/{action} -``` - -### 2.3 异步任务处理模式 - -视频生成通常是长时间任务，业界有两种主流处理模式： - -**模式1：同步等待（适合短视频）** -``` -POST /v1/videos/generations -Response: { - "id": "video-123", - "status": "completed", - "url": "https://..." -} -``` - -**模式2：异步轮询（主流）** -``` -# 提交任务 -POST /v1/videos/generations -Response: { - "id": "video-123", - "status": "processing" -} - -# 查询状态 -GET /v1/videos/{video_id} -Response: { - "id": "video-123", - "status": "completed", - "url": "https://..." -} -``` - -**模式3：Webhook回调** -``` -POST /v1/videos/generations -Request: { - "prompt": "...", - "webhook_url": "https://your-domain.com/callback" -} -``` - -## 三、问题诊断 - -### 3.1 NewAPI多协议分析 - -根据端点路径推测各协议用途： - -| 协议 | 端点 | 推测用途 | 建议 | -|------|------|----------|------| -| OpenAI / NewAPI Videos | /v1/videos | 兼容OpenAI官方API（如果有的话） | 保留作为主协议 | -| NewAPI 媒体任务 | /v1/videos | 可能是旧版本或别名 | 废弃，合并到主协议 | -| NewAPI Video Generations | /v1/video/generations | 更RESTful的命名 | 考虑作为标准协议 | - -### 3.2 命名不一致问题 - -| 问题 | 示例 | 影响 | -|------|------|------| -| 单复数混用 | videos vs video | 造成记忆负担 | -| 下划线vs连字符 | video_generation vs video-generation | 风格不统一 | -| 路径深度不一 | /v1/videos vs /api/v3/contents/generation | 增加适配复杂度 | - -## 四、改造方案 - -### 4.1 统一协议标准 - -**推荐：采用OpenAI风格作为统一标准** - -理由： -1. OpenAI是行业事实标准 -2. 大部分开发者熟悉此风格 -3. 利于生态兼容 -4. 语义清晰，易于理解 - -**标准端点设计：** -``` -POST /v1/videos/generations # 创建视频生成任务 -GET /v1/videos/{video_id} # 查询视频生成状态 -GET /v1/videos # 列出视频任务（可选） -DELETE /v1/videos/{video_id} # 取消任务（可选） -``` - -### 4.2 协议映射策略 - -将现有协议统一映射到标准协议： - -``` -旧协议 新协议（统一） -───────────────────────────────────────────────────────── -POST /v1/videos -> POST /v1/videos/generations -POST /v1/video/generations -> POST /v1/videos/generations -POST /v2/video_generation -> POST /v1/videos/generations -POST /api/v3/contents/generation -> POST /v1/videos/generations -POST CVSync2AsyncSubmitTask -> POST /v1/videos/generations -POST /v3/video/create -> POST /v1/videos/generations -``` - -### 4.3 请求/响应标准化 - -**标准请求格式：** -```json -{ - "model": "video-model-name", - "prompt": "视频描述文本", - "duration": 5, - "aspect_ratio": "16:9", - "fps": 30, - "resolution": "1080p", - "webhook_url": "https://callback-url.com/webhook" -} -``` - -**标准响应格式（提交时）：** -```json -{ - "id": "video-abc123", - "object": "video.generation", - "created": 1692800000, - "model": "video-model-name", - "status": "processing", - "estimated_time": 120 -} -``` - -**标准响应格式（完成后查询）：** -```json -{ - "id": "video-abc123", - "object": "video.generation", - "created": 1692800000, - "completed": 1692800120, - "model": "video-model-name", - "status": "completed", - "video": { - "url": "https://storage.example.com/video-abc123.mp4", - "duration": 5.2, - "width": 1920, - "height": 1080, - "fps": 30, - "format": "mp4" - } -} -``` - -**错误响应格式：** -```json -{ - "error": { - "type": "invalid_request_error", - "code": "invalid_prompt", - "message": "提示词包含不支持的内容", - "param": "prompt" - } -} -``` - -### 4.4 NewAPI协议整合方案 - -**方案A：保留一个，废弃其他（推荐）** - -- 保留：`POST /v1/videos/generations`（最符合OpenAI标准） -- 废弃：另外两个NewAPI协议 -- 过渡期：3个月内保留旧协议，显示弃用警告 - -**方案B：统一到通用适配器** - -- 设置一个"NewAPI 视频生成（统一）"协议 -- 后端自动检测NewAPI服务实际支持的端点 -- 用户无需选择具体协议 - -### 4.5 厂商特殊协议处理 - -对于特殊协议（如即梦的RPC风格），采用适配器模式： - -``` -用户请求（标准格式） - ↓ -协议适配层 - ↓ -厂商特定格式转换 - ↓ -实际API调用 -``` - -**实现示例：** -```python -class VideoAPIAdapter: - def generate_video(self, standard_request): - if self.provider == "jimeng": - # 转换为即梦的RPC格式 - rpc_request = { - "task_type": "video_generation", - "params": self._convert_to_jimeng_format(standard_request) - } - return self._call_jimeng_api(rpc_request) - elif self.provider == "volcano": - # 转换为火山方舟格式 - return self._call_volcano_api(standard_request) - # ... 其他厂商 -``` - -## 五、实施计划 - -### 5.1 阶段划分 - -**第一阶段：标准制定（1周）** -- [ ] 确定统一协议标准 -- [ ] 编写详细的协议规范文档 -- [ ] 设计请求/响应Schema -- [ ] 评审和确认 - -**第二阶段：适配器开发（2-3周）** -- [ ] 开发协议适配层 -- [ ] 实现各厂商协议转换器 -- [ ] 单元测试覆盖 -- [ ] 集成测试 - -**第三阶段：UI改造（1周）** -- [ ] 简化协议选择界面 -- [ ] NewAPI只显示一个统一选项 -- [ ] 添加协议说明文档链接 -- [ ] 优化用户体验 - -**第四阶段：兼容性处理（1周）** -- [ ] 保留旧协议的兼容性 -- [ ] 添加弃用警告 -- [ ] 数据迁移工具 -- [ ] 用户通知 - -**第五阶段：上线和监控（持续）** -- [ ] 灰度发布 -- [ ] 监控错误率 -- [ ] 收集用户反馈 -- [ ] 迭代优化 - -### 5.2 风险控制 - -| 风险 | 影响 | 应对措施 | -|------|------|----------| -| 现有用户业务中断 | 高 | 保留旧协议3个月，提前通知 | -| 新协议与厂商API不兼容 | 中 | 充分测试，保留特殊处理逻辑 | -| 性能下降 | 中 | 适配层做性能优化，缓存转换逻辑 | -| 文档更新不及时 | 低 | 自动生成API文档 | - -## 六、技术实现建议 - -### 6.1 配置文件结构 - -```yaml -video_providers: - # 统一配置格式 - - name: "newapi" - display_name: "NewAPI 视频生成" - protocol: "openai_compatible" - endpoint: "/v1/videos/generations" - auth_type: "bearer" - features: - - async_polling - - webhook - - - name: "jimeng" - display_name: "即梦官方视频" - protocol: "rpc" - endpoint: "/CVSync2AsyncSubmitTask" - adapter: "JimengVideoAdapter" - auth_type: "custom" - - # ... 其他厂商 -``` - -### 6.2 代码架构 - -``` -/src - /adapters - /base.py # 基础适配器接口 - /openai.py # OpenAI标准协议 - /jimeng.py # 即梦适配器 - /volcano.py # 火山方舟适配器 - /minimax.py # MiniMax适配器 - /models - /video_request.py # 标准请求模型 - /video_response.py # 标准响应模型 - /services - /video_service.py # 视频生成服务 - /api - /video_api.py # 统一API接口 -``` - -### 6.3 关键接口设计 - -```python -from abc import ABC, abstractmethod -from typing import Dict, Any - -class VideoProviderAdapter(ABC): - """视频服务提供商适配器基类""" - - @abstractmethod - def submit_generation(self, request: VideoGenerationRequest) -> VideoGenerationResponse: - """提交视频生成任务""" - pass - - @abstractmethod - def query_status(self, video_id: str) -> VideoStatusResponse: - """查询视频生成状态""" - pass - - @abstractmethod - def cancel_generation(self, video_id: str) -> bool: - """取消视频生成任务""" - pass - - def _convert_request(self, request: VideoGenerationRequest) -> Dict[str, Any]: - """将标准请求转换为厂商特定格式""" - pass - - def _convert_response(self, raw_response: Dict[str, Any]) -> VideoGenerationResponse: - """将厂商响应转换为标准格式""" - pass -``` - -## 七、配置界面优化建议 - -### 7.1 当前界面问题 - -- NewAPI出现3次，用户困惑 -- 协议列表过长，难以选择 -- 缺少协议说明和推荐 - -### 7.2 优化方案 - -**视觉分组：** -``` -请求协议 -├─ 推荐协议 -│ └─ OpenAI 标准协议（推荐） [✓] -│ -├─ 兼容协议 -│ ├─ NewAPI 视频生成 [ ] -│ ├─ Gemini Veo [ ] -│ └─ xAI 官方视频 [ ] -│ -└─ 特殊协议（需要特殊配置） - ├─ 火山方舟视频 [ ] - ├─ 即梦官方视频 [ ] - ├─ Novita 视频 [ ] - └─ MiniMax 视频 [ ] -``` - -**添加协议说明：** -- 鼠标悬停显示协议详情 -- 链接到完整文档 -- 显示支持的功能特性 - -**简化选择：** -- 默认推荐OpenAI标准协议 -- 自动检测厂商并推荐对应协议 -- 提供"快速配置"向导 - -## 八、总结与建议 - -### 8.1 核心建议 - -1. **立即行动**：NewAPI的3个重复协议问题需要尽快解决 -2. **采用标准**：统一采用OpenAI风格协议作为标准 -3. **平滑迁移**：保留兼容性，给用户足够的迁移时间 -4. **持续优化**：根据新厂商和新标准不断迭代 - -### 8.2 预期收益 - -- **开发效率**：减少70%的协议适配代码 -- **用户体验**：降低50%的配置复杂度 -- **维护成本**：统一标准后维护成本降低60% -- **扩展性**：新增厂商只需实现标准适配器接口 - -### 8.3 下一步行动 - -1. 与团队评审此方案 -2. 确定实施优先级 -3. 制定详细的开发计划 -4. 开始第一阶段工作 - ---- - -**文档版本**: v1.0 -**创建日期**: 2026-08-21 -**作者**: AI Assistant -**状态**: 待评审