fix(canvas): 画布版本 - 合并 PR #528 防止旧内容覆盖并支持历史恢复
将画布 revision CAS 与历史快照接到 schema v23,保留当前顶栏积分弹窗和已有菜单入口。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
commit
aae7b15ccc
69 files changed
+3571
-293
No files matched your search
@@ -159,7 +159,7 @@ func (s *Service) DeleteAdminResources(actor *model.User, req AdminResourceDelet
|
||||
audits = append(audits, *event)
|
||||
}
|
||||
if err := s.repo.DeleteAdminResources(deletable, deletionJobs, audits); err != nil {
|
||||
if errors.Is(err, repository.ErrAdminResourceDeleteChanged) || errors.Is(err, repository.ErrAdminResourceStillReferenced) {
|
||||
if errors.Is(err, repository.ErrAdminResourceDeleteChanged) || errors.Is(err, repository.ErrAdminResourceStillReferenced) || errors.Is(err, repository.ErrCanvasHistoryResourceReferenced) {
|
||||
return nil, BadAuthRequest("资源状态或引用已变化,请刷新后重试")
|
||||
}
|
||||
return nil, err
|
||||
|
||||
@@ -2,23 +2,26 @@ package app
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"time"
|
||||
|
||||
"infinite-canvas/backend/internal/assets"
|
||||
"infinite-canvas/backend/internal/canvas"
|
||||
"infinite-canvas/backend/internal/model"
|
||||
"infinite-canvas/backend/internal/repository"
|
||||
)
|
||||
|
||||
type (
|
||||
CanvasShareRequest = canvas.CanvasShareRequest
|
||||
CanvasShareStatus = canvas.CanvasShareStatus
|
||||
PublicCanvasShare = canvas.PublicCanvasShare
|
||||
AssetsSyncRequest = canvas.AssetsSyncRequest
|
||||
CanvasProjectsSyncRequest = canvas.CanvasProjectsSyncRequest
|
||||
UserDataSummary = canvas.UserDataSummary
|
||||
UserDataSnapshot = canvas.UserDataSnapshot
|
||||
CanvasLibrarySummary = canvas.CanvasLibrarySummary
|
||||
CanvasLibraryPage = canvas.CanvasLibraryPage
|
||||
CanvasShareRequest = canvas.CanvasShareRequest
|
||||
CanvasShareStatus = canvas.CanvasShareStatus
|
||||
PublicCanvasShare = canvas.PublicCanvasShare
|
||||
AssetsSyncRequest = canvas.AssetsSyncRequest
|
||||
CanvasHistoryList = canvas.CanvasHistoryList
|
||||
UserDataSummary = canvas.UserDataSummary
|
||||
UserDataSnapshot = canvas.UserDataSnapshot
|
||||
CanvasLibrarySummary = canvas.CanvasLibrarySummary
|
||||
CanvasLibraryPage = canvas.CanvasLibraryPage
|
||||
)
|
||||
|
||||
type canvasHost struct {
|
||||
@@ -207,8 +210,32 @@ func (s *Service) DeleteUserCanvasProject(userID string, id string) error {
|
||||
return s.canvasDomain().DeleteUserCanvasProject(userID, id)
|
||||
}
|
||||
|
||||
func (s *Service) ReplaceUserCanvasProjects(userID string, req CanvasProjectsSyncRequest) ([]json.RawMessage, error) {
|
||||
return s.canvasDomain().ReplaceUserCanvasProjects(userID, req)
|
||||
func (s *Service) CanvasHistory(userID, canvasID string) (CanvasHistoryList, error) {
|
||||
return s.canvasDomain().CanvasHistory(userID, canvasID)
|
||||
}
|
||||
|
||||
func (s *Service) CanvasHistorySnapshot(userID, canvasID, snapshotID string) (*model.CanvasSnapshot, error) {
|
||||
return s.canvasDomain().CanvasHistorySnapshot(userID, canvasID, snapshotID)
|
||||
}
|
||||
|
||||
func (s *Service) RestoreCanvasHistory(userID, canvasID, snapshotID string, revision *int64) (UserDataSummary, error) {
|
||||
return s.canvasDomain().RestoreCanvasHistory(userID, canvasID, snapshotID, revision)
|
||||
}
|
||||
|
||||
func saveCreationCanvasWithHistory(repo *repository.Repository, project *model.CanvasProject, previous string) error {
|
||||
before, err := repo.CanvasProjectForUser(project.UserID, project.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if before.PayloadJSON != previous || before.Revision != project.Revision {
|
||||
return repository.ErrCreationConflict
|
||||
}
|
||||
project.UpdatedAt = time.Now().UTC()
|
||||
err = canvas.SaveDocumentWithHistory(repo, before, project, "automatic")
|
||||
if errors.Is(err, repository.ErrCanvasRevisionConflict) {
|
||||
return repository.ErrCreationConflict
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Service) UserAssetsByIDs(userID string, ids []string) ([]json.RawMessage, error) {
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"infinite-canvas/backend/internal/model"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestCanvasHistoryProtectsMediaAndDeletionWorker(t *testing.T) {
|
||||
svc, db, dataDir := newResourceDeletionTestService(t)
|
||||
key := "users/user-1/video/history.mp4"
|
||||
file := filepath.Join(dataDir, "resources", filepath.FromSlash(key))
|
||||
if err := os.MkdirAll(filepath.Dir(file), 0o750); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(file, []byte("video"), 0o640); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
resource := model.Resource{ID: "historical-video", UserID: "user-1", Provider: "local", ObjectKey: key, Status: model.ResourceStatusReady}
|
||||
asset := model.Asset{ID: "history-asset", UserID: "user-1", Title: "video", PayloadJSON: `{"data":{"storageKey":"resource:historical-video"}}`}
|
||||
snapshot := model.CanvasSnapshot{ID: "history-snapshot", CanvasID: "canvas", UserID: "user-1", Revision: 1, Title: "wedding", PayloadJSON: `{}`, CreatedAt: time.Now()}
|
||||
ref := model.CanvasSnapshotResource{SnapshotID: snapshot.ID, ResourceID: resource.ID}
|
||||
for _, item := range []any{&resource, &asset, &snapshot, &ref} {
|
||||
if err := db.Create(item).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := svc.DeleteUserAsset("user-1", asset.ID); err == nil || !strings.Contains(err.Error(), "画布历史版本") {
|
||||
t.Fatalf("history reference ignored: %v", err)
|
||||
}
|
||||
// A previously queued physical deletion must also honor a later history reference.
|
||||
job := resourceDeletionJobs("user-1", map[string]*model.Resource{key: &resource})[0]
|
||||
job.ResourceID = "deleted-resource-alias"
|
||||
if err := db.Create(&job).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
svc.drainResourceDeletionJobs(1)
|
||||
if _, err := os.Stat(file); err != nil {
|
||||
t.Fatalf("history media removed: %v", err)
|
||||
}
|
||||
var pending model.ResourceDeletionJob
|
||||
if err := db.First(&pending, "id = ?", job.ID).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if pending.Attempts < 1 || pending.LastError == "" {
|
||||
t.Fatalf("protected deletion was not deferred: %#v", pending)
|
||||
}
|
||||
if err := db.Delete(&ref).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Delete(&asset).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Delete(&resource).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&pending).Update("next_attempt_at", time.Now().Add(-time.Minute)).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
svc.drainResourceDeletionJobs(1)
|
||||
if _, err := os.Stat(file); !errors.Is(err, os.ErrNotExist) {
|
||||
t.Fatalf("unreferenced media not cleaned: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -47,7 +47,9 @@ func emitCloudAgentCanvasChange(repo *repository.Repository, runID string, state
|
||||
}
|
||||
nodes := cloudAgentObjectChanges(before["nodes"], after["nodes"])
|
||||
edges := cloudAgentObjectChanges(before["connections"], after["connections"])
|
||||
if len(nodes) == 0 && len(edges) == 0 {
|
||||
if !cloudAgentPatchCoversDocument(before, after) {
|
||||
// A partial delta must never acknowledge a revision containing omitted edits.
|
||||
state.event(runID, "canvas_updated", map[string]any{"canvasId": input.CanvasID, "operation": input.Operation, "requiresRefresh": true})
|
||||
return nil
|
||||
}
|
||||
actions := []map[string]any{}
|
||||
@@ -126,7 +128,7 @@ func emitCloudAgentCanvasChange(repo *repository.Repository, runID string, state
|
||||
}
|
||||
payload := map[string]any{
|
||||
"canvasId": input.CanvasID, "operation": input.Operation, "actions": actions,
|
||||
"canvasPatch": map[string]any{"canvasId": input.CanvasID, "updatedAt": after["updatedAt"], "nodes": nodes, "connections": edges},
|
||||
"canvasPatch": map[string]any{"canvasId": input.CanvasID, "baseRevision": canvas.Revision - 1, "revision": canvas.Revision, "updatedAt": after["updatedAt"], "nodes": nodes, "connections": edges},
|
||||
}
|
||||
if input.Preview != nil {
|
||||
payload["preview"] = input.Preview
|
||||
@@ -138,3 +140,28 @@ func emitCloudAgentCanvasChange(repo *repository.Repository, runID string, state
|
||||
state.event(runID, "canvas_updated", payload)
|
||||
return nil
|
||||
}
|
||||
|
||||
func cloudAgentPatchCoversDocument(before, after map[string]any) bool {
|
||||
for _, key := range []string{"nodes", "connections"} {
|
||||
old, next := creationMaps(before[key]), creationMaps(after[key])
|
||||
if len(next) < len(old) {
|
||||
return false
|
||||
}
|
||||
// The delta merger updates existing items in place and appends new items.
|
||||
for i, item := range old {
|
||||
if item["id"] != next[i]["id"] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
}
|
||||
content := func(doc map[string]any) map[string]any {
|
||||
result := map[string]any{}
|
||||
for key, value := range doc {
|
||||
if key != "nodes" && key != "connections" && key != "updatedAt" && key != "viewport" {
|
||||
result[key] = value
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
return reflect.DeepEqual(content(before), content(after))
|
||||
}
|
||||
@@ -84,6 +84,22 @@ func TestCloudAgentCanvasPatchesPersistDraftSubmissionAndAllTerminalStates(t *te
|
||||
t.Fatalf("submission failed to lock the existing draft: %+v", submission)
|
||||
}
|
||||
completion := agentCanvasPatchForOperation(t, state, "generate_media_complete")["canvasPatch"].(map[string]any)
|
||||
previousRevision := patch["baseRevision"]
|
||||
for _, delta := range []map[string]any{patch, submission, completion} {
|
||||
if delta["baseRevision"] != previousRevision {
|
||||
t.Fatalf("discontinuous Agent revision: %+v", delta)
|
||||
}
|
||||
base, ok := delta["baseRevision"].(float64)
|
||||
next, nextOK := delta["revision"].(float64)
|
||||
if !ok || !nextOK || next != base+1 {
|
||||
t.Fatalf("invalid Agent revisions: %+v", delta)
|
||||
}
|
||||
previousRevision = delta["revision"]
|
||||
}
|
||||
history, err := s.CanvasHistory("user", "agent-canvas")
|
||||
if err != nil || len(history.Snapshots) == 0 || float64(history.CurrentRevision) != previousRevision {
|
||||
t.Fatalf("Agent save did not retain history/revision: %+v %v", history, err)
|
||||
}
|
||||
changes = creationMaps(completion["nodes"])
|
||||
if len(changes) != 1 || len(creationMaps(completion["connections"])) != 0 {
|
||||
t.Fatalf("completion includes unrelated canvas data: %+v", completion)
|
||||
@@ -127,6 +143,34 @@ func TestCloudAgentMediaImageSourceHasActionableCorrection(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCloudAgentDeletionInvalidatesTheWholeCanvasRevision(t *testing.T) {
|
||||
s, _, _ := agentMediaFixture(t)
|
||||
canvas, err := s.repo.CanvasProjectForUser("user", "agent-canvas")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
previous := canvas.PayloadJSON
|
||||
doc, err := creationDocument(previous)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
nodes := creationMaps(doc["nodes"])
|
||||
doc["nodes"] = nodes[1:]
|
||||
nodes[1]["title"] = "Also updated"
|
||||
raw, _ := json.Marshal(doc)
|
||||
canvas.PayloadJSON = string(raw)
|
||||
if err := saveCreationCanvasWithHistory(s.repo, canvas, previous); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
state := cloudAgentRuntime{}
|
||||
if err := emitCloudAgentCanvasChange(s.repo, "run", &state, cloudAgentMutationInput{UserID: "user", CanvasID: canvas.ID, BeforeJSON: previous}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(state.Events) != 1 || state.Events[0].Payload["requiresRefresh"] != true || state.Events[0].Payload["canvasPatch"] != nil {
|
||||
t.Fatalf("incomplete delta would acknowledge a deleted node: %+v", state.Events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCloudAgentCanvasOperationTraceSharesToolCallID(t *testing.T) {
|
||||
s, _, args := agentMediaFixture(t)
|
||||
run, state := agentMediaRun(t, s, args, "auto")
|
||||
|
||||
@@ -539,7 +539,7 @@ func saveCloudAgentDocument(repo *repository.Repository, canvas *model.CanvasPro
|
||||
}
|
||||
before := canvas.PayloadJSON
|
||||
canvas.PayloadJSON = string(raw)
|
||||
return repo.CompareSaveCreationCanvas(canvas, before)
|
||||
return saveCreationCanvasWithHistory(repo, canvas, before)
|
||||
}
|
||||
|
||||
// Called inside the same transaction as the task, charge reservation and Agent checkpoint.
|
||||
|
||||
@@ -296,7 +296,7 @@ func TestCloudAgentStoryboardRejectsUnsafeOrStaleMutations(t *testing.T) {
|
||||
stored, _ := s.repo.CanvasProjectForUser("user", canvas.ID)
|
||||
before := stored.PayloadJSON
|
||||
stored.PayloadJSON = string(raw)
|
||||
if err := s.repo.CompareSaveCreationCanvas(stored, before); err != nil {
|
||||
if err := saveCreationCanvasWithHistory(s.repo, stored, before); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wrongType := cloudAgentStoryboardCall(t, "canvas_edit_storyboard", "wrong-type", map[string]any{
|
||||
|
||||
@@ -100,7 +100,7 @@ func (s *Service) UndoCloudAgentCanvas(userID, runID, stepID, expectedSnapshotHa
|
||||
}
|
||||
previous := canvas.PayloadJSON
|
||||
canvas.PayloadJSON = mutation.BeforeJSON
|
||||
if err := repo.CompareSaveCreationCanvas(canvas, previous); err != nil {
|
||||
if err := saveCreationCanvasWithHistory(repo, canvas, previous); err != nil {
|
||||
if errors.Is(err, repository.ErrCreationConflict) {
|
||||
return creationConflict("画布已发生后续变化,未执行撤销")
|
||||
}
|
||||
|
||||
@@ -111,13 +111,13 @@ func TestUndoCloudAgentCanvasRejectsChangedCanvasAndSubmittedTask(t *testing.T)
|
||||
afterHash := cloudAgentCanvasHash(changedDoc)
|
||||
// A user edit after the Agent mutation must make the compare-and-restore fail.
|
||||
changed.PayloadJSON = `{"nodes":[{"id":"user-node","type":"text","title":"用户编辑","metadata":{"content":"后来修改"}}]}`
|
||||
if err := s.repo.CompareSaveCreationCanvas(changed, changed.PayloadJSON); err == nil {
|
||||
if err := saveCreationCanvasWithHistory(s.repo, changed, changed.PayloadJSON); err == nil {
|
||||
t.Fatal("test setup unexpectedly saved with the new value as previous snapshot")
|
||||
}
|
||||
latest, _ := s.repo.CanvasProjectForUser("user", "agent-canvas")
|
||||
previous := latest.PayloadJSON
|
||||
latest.PayloadJSON = `{"nodes":[{"id":"user-node","type":"text","title":"用户编辑","metadata":{"content":"后来修改"}}]}`
|
||||
if err := s.repo.CompareSaveCreationCanvas(latest, previous); err != nil {
|
||||
if err := saveCreationCanvasWithHistory(s.repo, latest, previous); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = s.UndoCloudAgentCanvas("user", run.ID, call.ID, afterHash, "冲突")
|
||||
|
||||
@@ -246,7 +246,7 @@ func (s *Service) CommitCreationCanvas(userID, id string, req CreationRequest) (
|
||||
if err = validateStructuredStorageQuotaWithPolicy(usage, "canvas", false, int64(len(req.Document)-len(previous)), policy.Resource); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = repo.CompareSaveCreationCanvas(canvas, previous); err != nil {
|
||||
if err = saveCreationCanvasWithHistory(repo, canvas, previous); err != nil {
|
||||
return err
|
||||
}
|
||||
out = map[string]any{"snapshotHash": creationHash(doc)}
|
||||
|
||||
@@ -609,7 +609,7 @@ func (s *Service) UnlinkCanvasProject(userID string, projectID string, canvasID
|
||||
return err
|
||||
}
|
||||
// 关系列、同步快照和更新时间必须原子更新,否则浏览器会用旧 projectId 把关系重新写回。
|
||||
return s.repo.UnassignCanvasFromProject(userID, projectID, canvas.ID, payloadJSON, now)
|
||||
return s.repo.UnassignCanvasFromProject(userID, projectID, canvas.ID, payloadJSON, now, canvas.Revision)
|
||||
}
|
||||
|
||||
func canvasPayloadWithoutProject(payloadJSON string, updatedAt time.Time) (string, error) {
|
||||
|
||||
@@ -22,6 +22,8 @@ func newProjectDeleteTestService(t *testing.T) (*Service, *gorm.DB) {
|
||||
&model.Project{},
|
||||
&model.ProjectUnit{},
|
||||
&model.CanvasProject{},
|
||||
&model.CanvasSnapshot{},
|
||||
&model.CanvasSnapshotResource{},
|
||||
&model.CanvasShare{},
|
||||
&model.CanvasUnitLink{},
|
||||
&model.ProjectAssetLink{},
|
||||
|
||||
@@ -20,6 +20,7 @@ func newProjectSettingsTestService(t *testing.T) (*Service, *gorm.DB) {
|
||||
if err := db.AutoMigrate(
|
||||
&model.Project{}, &model.ProjectUnit{}, &model.Resource{}, &model.Asset{}, &model.AssetVersion{}, &model.AssetRepresentation{},
|
||||
&model.CanvasProject{}, &model.StyleProfile{}, &model.ProjectAssetCandidate{}, &model.WorkflowInstance{}, &model.WorkflowStepInstance{},
|
||||
&model.CanvasSnapshot{}, &model.CanvasSnapshotResource{},
|
||||
&model.Shot{}, &model.ShotArtifact{}, &model.VoiceProfile{},
|
||||
&model.Task{}, &model.TaskLog{}, &model.Result{},
|
||||
&model.CreationRun{}, &model.CreationSubmission{},
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
|
||||
"infinite-canvas/backend/internal/assets"
|
||||
"infinite-canvas/backend/internal/model"
|
||||
"infinite-canvas/backend/internal/repository"
|
||||
|
||||
qiniuAuth "github.com/qiniu/go-sdk/v7/auth"
|
||||
qiniuStorage "github.com/qiniu/go-sdk/v7/storage"
|
||||
@@ -139,6 +140,9 @@ func (s *Service) deleteUserAssetWithResources(userID string, assetID string) er
|
||||
// 业务记录和 Outbox 必须在同一事务提交。事务失败时物理文件完全不动;
|
||||
// 提交成功后由幂等 worker 清理,进程退出或对象存储暂时失败都可继续重试。
|
||||
if err := s.repo.DeleteAssetAndResources(userID, assetID, ownedIDs, deletionJobs); err != nil {
|
||||
if errors.Is(err, repository.ErrCanvasHistoryResourceReferenced) {
|
||||
return BadAuthRequest("素材仍被画布历史版本引用,已保留文件")
|
||||
}
|
||||
return fmt.Errorf("素材记录删除失败,请重试:%w", err)
|
||||
}
|
||||
if len(deletionJobs) > 0 {
|
||||
@@ -234,6 +238,13 @@ func (s *Service) deleteStoredResourceObject(userID string, resource *model.Reso
|
||||
if strings.TrimSpace(resource.ObjectKey) == "" {
|
||||
return fmt.Errorf("资源 %s 的存储路径为空", resource.ID)
|
||||
}
|
||||
protected, err := s.repo.CanvasHistoryReferencesObject(resource)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if protected {
|
||||
return errors.New("资源仍被画布历史版本引用")
|
||||
}
|
||||
switch strings.ToLower(strings.TrimSpace(resource.Provider)) {
|
||||
case "", "local":
|
||||
return s.deleteLocalResourceObject(resource.ObjectKey)
|
||||
|
||||
Reference in new issue
Block a user