feat(admin): 存储管理 - 支持批量引用校验与事务化删除

Adapted from ddcat-ai/open-ai-canvas#339

Original commit: fcdbe2aa833f31f683f709b7f409a0534ed78e20

Refs #339

Co-authored-by: 宇熙 <101980758+yuxiwl@users.noreply.github.com>
This commit is contained in:
ddCat
2026-08-29 15:40:40 +08:00
parent 464bf6e352
commit 7e20041835
7 changed files with 649 additions and 8 deletions

View File

@@ -9,6 +9,25 @@ import (
)
func RegisterAdminStorageRoutes(r *gin.RouterGroup, svc *service.Service) {
r.POST("/admin/resources/delete", func(c *gin.Context) {
user, err := currentUser(c, svc)
if err != nil {
failService(c, err)
return
}
var req service.AdminResourceDeleteRequest
if err := c.ShouldBindJSON(&req); err != nil {
failService(c, service.BadAuthRequest("删除资源请求无效"))
return
}
result, err := svc.DeleteAdminResources(user, req)
if err != nil {
failService(c, err)
return
}
ok(c, result)
})
r.GET("/admin/storage/stats", func(c *gin.Context) {
user, err := currentUser(c, svc)
if err != nil {

View File

@@ -0,0 +1,105 @@
package repository
import (
"errors"
"infinite-canvas/backend/internal/model"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
var ErrAdminResourceDeleteChanged = errors.New("admin resource delete set changed")
var ErrAdminResourceStillReferenced = errors.New("admin resource is still directly referenced")
var ErrAdminResourceAuditMismatch = errors.New("admin resource delete audit mismatch")
func (r *Repository) AdminResourcesByIDs(ids []string) ([]model.Resource, error) {
if len(ids) == 0 {
return []model.Resource{}, nil
}
var resources []model.Resource
err := r.db.Where("id IN ?", ids).Find(&resources).Error
return resources, err
}
func (r *Repository) AnnouncementResourceReferences(resourceIDs []string) ([]ResourceDirectReference, error) {
if len(resourceIDs) == 0 {
return []ResourceDirectReference{}, nil
}
var announcements []model.Announcement
if err := r.db.Where("image_resource_id IN ?", resourceIDs).Find(&announcements).Error; err != nil {
return nil, err
}
result := make([]ResourceDirectReference, 0, len(announcements))
for _, announcement := range announcements {
result = append(result, ResourceDirectReference{Kind: "公告", ID: announcement.ID, Title: announcement.Title, ResourceID: announcement.ImageResourceID})
}
return result, nil
}
func (r *Repository) DeleteAdminResources(resources []model.Resource, deletionJobs []model.ResourceDeletionJob, audits []model.AdminAuditEvent) error {
if len(resources) == 0 {
return nil
}
if len(audits) != len(resources) {
return ErrAdminResourceAuditMismatch
}
resourceIDs := make([]string, 0, len(resources))
for _, resource := range resources {
resourceIDs = append(resourceIDs, resource.ID)
}
return r.db.Transaction(func(tx *gorm.DB) error {
var current []model.Resource
query := tx.Where("id IN ?", resourceIDs)
if r.Dialect() == "postgres" {
query = query.Clauses(clause.Locking{Strength: "UPDATE"})
}
if err := query.Find(&current).Error; err != nil {
return err
}
if len(current) != len(resources) {
return ErrAdminResourceDeleteChanged
}
for _, check := range []struct {
model any
query string
}{
{&model.Announcement{}, "image_resource_id IN ?"},
{&model.AssetRepresentation{}, "resource_id IN ?"},
{&model.VoiceProfile{}, "sample_resource_id IN ?"},
{&model.ShotArtifact{}, "resource_id IN ?"},
} {
var count int64
if err := tx.Model(check.model).Where(check.query, resourceIDs).Count(&count).Error; err != nil {
return err
}
if count > 0 {
return ErrAdminResourceStillReferenced
}
}
if err := tx.Where("resource_id IN ?", resourceIDs).Delete(&model.ArkPrivateAssetBinding{}).Error; err != nil {
return err
}
if err := tx.Where("resource_id IN ?", resourceIDs).Delete(&model.AnnouncementImageDraft{}).Error; err != nil {
return err
}
if len(deletionJobs) > 0 {
if err := tx.Create(&deletionJobs).Error; err != nil {
return err
}
}
deleted := tx.Where("id IN ?", resourceIDs).Delete(&model.Resource{})
if deleted.Error != nil {
return deleted.Error
}
if deleted.RowsAffected != int64(len(resourceIDs)) {
return ErrAdminResourceDeleteChanged
}
if len(audits) > 0 {
if err := tx.Create(&audits).Error; err != nil {
return err
}
}
return nil
})
}

View File

@@ -0,0 +1,262 @@
package service
import (
"errors"
"sort"
"strings"
"infinite-canvas/backend/internal/model"
"infinite-canvas/backend/internal/repository"
)
const maxAdminResourceDeleteCount = 100
type AdminResourceDeleteRequest struct {
ResourceIDs []string `json:"resourceIds"`
}
type AdminResourceReferenceView struct {
Kind string `json:"kind"`
ID string `json:"id"`
Title string `json:"title"`
}
type AdminResourceDeleteBlocked struct {
ID string `json:"id"`
Reason string `json:"reason"`
References []AdminResourceReferenceView `json:"references"`
}
type AdminResourceDeleteResult struct {
Deleted []string `json:"deleted"`
Blocked []AdminResourceDeleteBlocked `json:"blocked"`
}
func (s *Service) DeleteAdminResources(actor *model.User, req AdminResourceDeleteRequest) (*AdminResourceDeleteResult, error) {
if err := s.RequireAdmin(actor); err != nil {
return nil, err
}
resourceIDs, err := normalizeAdminResourceDeleteIDs(req.ResourceIDs)
if err != nil {
return nil, err
}
s.storageMu.Lock()
defer s.storageMu.Unlock()
resources, err := s.repo.AdminResourcesByIDs(resourceIDs)
if err != nil {
return nil, err
}
resourcesByID := make(map[string]model.Resource, len(resources))
resourcesByUser := make(map[string][]model.Resource)
for _, resource := range resources {
resourcesByID[resource.ID] = resource
resourcesByUser[resource.UserID] = append(resourcesByUser[resource.UserID], resource)
}
blockedByID := make(map[string]AdminResourceDeleteBlocked)
for _, resourceID := range resourceIDs {
if _, exists := resourcesByID[resourceID]; !exists {
blockedByID[resourceID] = AdminResourceDeleteBlocked{ID: resourceID, Reason: "资源不存在", References: []AdminResourceReferenceView{}}
}
}
for userID, userResources := range resourcesByUser {
userResourceIDs := make([]string, 0, len(userResources))
for _, resource := range userResources {
userResourceIDs = append(userResourceIDs, resource.ID)
}
snapshot, snapshotErr := s.repo.ResourceReferenceSnapshot(userID, "", userResourceIDs)
if snapshotErr != nil {
return nil, snapshotErr
}
for resourceID, references := range adminResourceReferences(snapshot, userResources) {
blockedByID[resourceID] = AdminResourceDeleteBlocked{ID: resourceID, Reason: "资源仍被业务数据引用", References: references}
}
}
announcementReferences, err := s.repo.AnnouncementResourceReferences(resourceIDs)
if err != nil {
return nil, err
}
for _, reference := range announcementReferences {
blocked := blockedByID[reference.ResourceID]
blocked.ID = reference.ResourceID
blocked.Reason = "资源仍被业务数据引用"
blocked.References = appendUniqueAdminResourceReference(blocked.References, AdminResourceReferenceView{Kind: reference.Kind, ID: reference.ID, Title: reference.Title})
blockedByID[reference.ResourceID] = blocked
}
deletable := make([]model.Resource, 0, len(resources))
deletableIDs := make([]string, 0, len(resources))
for _, resourceID := range resourceIDs {
resource, exists := resourcesByID[resourceID]
if !exists {
continue
}
if _, blocked := blockedByID[resourceID]; blocked {
continue
}
if !supportedResourceDeleteProvider(resource.Provider) {
blockedByID[resourceID] = AdminResourceDeleteBlocked{ID: resourceID, Reason: "资源使用了不支持的存储类型,无法安全删除", References: []AdminResourceReferenceView{}}
continue
}
deletable = append(deletable, resource)
deletableIDs = append(deletableIDs, resource.ID)
}
physicalObjects := map[string]*model.Resource{}
checkedPhysical := map[string]bool{}
for index := range deletable {
resource := &deletable[index]
if strings.TrimSpace(resource.ObjectKey) == "" {
continue
}
identity := resourceStorageIdentity(resource)
if checkedPhysical[identity] {
continue
}
checkedPhysical[identity] = true
sharedCount, countErr := s.repo.ResourceStorageReferenceCount(resource, deletableIDs)
if countErr != nil {
return nil, countErr
}
if sharedCount == 0 {
physicalObjects[identity] = resource
}
}
deletionJobs := resourceDeletionJobsForResources(physicalObjects)
queuedResourceIDs := make(map[string]bool, len(deletionJobs))
for _, job := range deletionJobs {
queuedResourceIDs[job.ResourceID] = true
}
audits := make([]model.AdminAuditEvent, 0, len(deletable))
for _, resource := range deletable {
event, eventErr := newAdminAuditEvent(actor, "resource.delete", "resource", resource.ID, "管理员删除存储资源", map[string]any{
"userId": resource.UserID, "kind": resource.Kind, "provider": normalizedResourceProvider(resource.Provider),
"objectKey": resource.ObjectKey, "physicalDeleteQueued": queuedResourceIDs[resource.ID],
})
if eventErr != nil {
return nil, eventErr
}
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) {
return nil, BadAuthRequest("资源状态或引用已变化,请刷新后重试")
}
return nil, err
}
if len(deletionJobs) > 0 {
go s.drainResourceDeletionJobs(len(deletionJobs))
}
result := &AdminResourceDeleteResult{Deleted: []string{}, Blocked: []AdminResourceDeleteBlocked{}}
deletedSet := make(map[string]struct{}, len(deletable))
for _, resource := range deletable {
deletedSet[resource.ID] = struct{}{}
}
for _, resourceID := range resourceIDs {
if _, deleted := deletedSet[resourceID]; deleted {
result.Deleted = append(result.Deleted, resourceID)
} else if blocked, exists := blockedByID[resourceID]; exists {
result.Blocked = append(result.Blocked, blocked)
}
}
return result, nil
}
func normalizeAdminResourceDeleteIDs(values []string) ([]string, error) {
if len(values) == 0 {
return nil, BadAuthRequest("请选择要删除的资源")
}
result := make([]string, 0, len(values))
seen := make(map[string]struct{}, len(values))
for _, value := range values {
value = strings.TrimSpace(value)
if value == "" || len(value) > 80 {
return nil, BadAuthRequest("资源 ID 无效")
}
if _, exists := seen[value]; exists {
continue
}
seen[value] = struct{}{}
result = append(result, value)
}
if len(result) == 0 || len(result) > maxAdminResourceDeleteCount {
return nil, BadAuthRequest("单次最多删除 100 个资源")
}
return result, nil
}
func adminResourceReferences(snapshot repository.ResourceReferenceSnapshot, resources []model.Resource) map[string][]AdminResourceReferenceView {
result := make(map[string][]AdminResourceReferenceView)
seen := make(map[string]map[string]struct{})
for _, reference := range snapshot.Direct {
appendAdminResourceReference(result, seen, reference.ResourceID, AdminResourceReferenceView{Kind: reference.Kind, ID: reference.ID, Title: reference.Title})
}
for _, resource := range resources {
candidate := map[string]struct{}{resource.ID: {}}
for _, document := range snapshot.Documents {
if documentReferencesResources(document.PrimaryJSON, candidate) || documentReferencesResources(document.SecondaryJSON, candidate) {
appendAdminResourceReference(result, seen, resource.ID, AdminResourceReferenceView{Kind: document.Kind, ID: document.ID, Title: document.Title})
}
}
}
for resourceID := range result {
sort.Slice(result[resourceID], func(i, j int) bool {
left, right := result[resourceID][i], result[resourceID][j]
if left.Kind != right.Kind {
return left.Kind < right.Kind
}
return left.ID < right.ID
})
}
return result
}
func appendAdminResourceReference(result map[string][]AdminResourceReferenceView, seen map[string]map[string]struct{}, resourceID string, reference AdminResourceReferenceView) {
if resourceID == "" {
return
}
if seen[resourceID] == nil {
seen[resourceID] = map[string]struct{}{}
}
key := reference.Kind + "\x00" + reference.ID
if _, exists := seen[resourceID][key]; exists {
return
}
seen[resourceID][key] = struct{}{}
result[resourceID] = append(result[resourceID], reference)
}
func appendUniqueAdminResourceReference(references []AdminResourceReferenceView, reference AdminResourceReferenceView) []AdminResourceReferenceView {
for _, existing := range references {
if existing.Kind == reference.Kind && existing.ID == reference.ID {
return references
}
}
return append(references, reference)
}
func supportedResourceDeleteProvider(provider string) bool {
provider = normalizedResourceProvider(provider)
return oneOf(provider, "local", aliyunOSSProvider, tencentCOSProvider, qiniuKodoProvider, s3Provider)
}
func resourceDeletionJobsForResources(physicalObjects map[string]*model.Resource) []model.ResourceDeletionJob {
resourcesByUser := make(map[string]map[string]*model.Resource)
for identity, resource := range physicalObjects {
if resourcesByUser[resource.UserID] == nil {
resourcesByUser[resource.UserID] = map[string]*model.Resource{}
}
resourcesByUser[resource.UserID][identity] = resource
}
userIDs := make([]string, 0, len(resourcesByUser))
for userID := range resourcesByUser {
userIDs = append(userIDs, userID)
}
sort.Strings(userIDs)
jobs := []model.ResourceDeletionJob{}
for _, userID := range userIDs {
jobs = append(jobs, resourceDeletionJobs(userID, resourcesByUser[userID])...)
}
return jobs
}

View File

@@ -0,0 +1,170 @@
package service
import (
"os"
"path/filepath"
"strings"
"testing"
"infinite-canvas/backend/internal/database"
"infinite-canvas/backend/internal/model"
"infinite-canvas/backend/internal/repository"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
func TestDeleteAdminResourcesSupportsPartialSuccessAndBatchedReferences(t *testing.T) {
svc, db, dataDir, admin := newAdminStorageDeleteTestService(t)
resources := []model.Resource{
{ID: "resource-free", UserID: "user-1", Kind: "image", Status: model.ResourceStatusReady, Provider: "local", ObjectKey: "users/user-1/image/delete-directory"},
{ID: "resource-blocked", UserID: "user-1", Kind: "image", Status: model.ResourceStatusReady, Provider: "local", ObjectKey: "users/user-1/image/blocked.png"},
}
if err := db.Create(&resources).Error; err != nil {
t.Fatal(err)
}
canvas := model.CanvasProject{ID: "canvas-1", UserID: "user-1", Title: "广告分镜", PayloadJSON: `{"nodes":[{"data":{"storageKey":"resource:resource-blocked"}}]}`}
if err := db.Create(&canvas).Error; err != nil {
t.Fatal(err)
}
directoryPath := filepath.Join(dataDir, "resources", filepath.FromSlash(resources[0].ObjectKey))
if err := os.MkdirAll(directoryPath, 0o750); err != nil {
t.Fatal(err)
}
result, err := svc.DeleteAdminResources(admin, AdminResourceDeleteRequest{ResourceIDs: []string{"resource-free", "resource-blocked", "resource-free"}})
if err != nil {
t.Fatal(err)
}
if len(result.Deleted) != 1 || result.Deleted[0] != "resource-free" {
t.Fatalf("deleted = %#v", result.Deleted)
}
if len(result.Blocked) != 1 || result.Blocked[0].ID != "resource-blocked" || len(result.Blocked[0].References) != 1 || result.Blocked[0].References[0].Kind != "画布" {
t.Fatalf("blocked = %#v", result.Blocked)
}
assertModelCount(t, db, &model.Resource{}, "id = ?", 0, "resource-free")
assertModelCount(t, db, &model.Resource{}, "id = ?", 1, "resource-blocked")
assertModelCount(t, db, &model.ResourceDeletionJob{}, "resource_id = ?", 1, "resource-free")
assertModelCount(t, db, &model.AdminAuditEvent{}, "action = ? AND target_id = ?", 1, "resource.delete", "resource-free")
}
func TestDeleteAdminResourcesBlocksAnnouncementImage(t *testing.T) {
svc, db, _, admin := newAdminStorageDeleteTestService(t)
resource := model.Resource{ID: "announcement-image", UserID: admin.ID, Kind: "image", Status: model.ResourceStatusReady, Provider: "local", ObjectKey: "announcements/cover.png"}
announcement := model.Announcement{ID: "announcement-1", Title: "维护通知", Content: "content", ImageResourceID: resource.ID, CreatedBy: admin.ID, Status: model.AnnouncementStatusActive}
if err := db.Create(&resource).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&announcement).Error; err != nil {
t.Fatal(err)
}
result, err := svc.DeleteAdminResources(admin, AdminResourceDeleteRequest{ResourceIDs: []string{resource.ID}})
if err != nil {
t.Fatal(err)
}
if len(result.Deleted) != 0 || len(result.Blocked) != 1 || result.Blocked[0].References[0].Kind != "公告" {
t.Fatalf("result = %#v", result)
}
assertModelCount(t, db, &model.Resource{}, "id = ?", 1, resource.ID)
}
func TestDeleteAdminResourcesKeepsSharedPhysicalObject(t *testing.T) {
svc, db, _, admin := newAdminStorageDeleteTestService(t)
resources := []model.Resource{
{ID: "resource-delete", UserID: "user-1", Kind: "image", Status: model.ResourceStatusReady, Provider: "", ObjectKey: "shared/object.png"},
{ID: "resource-keep", UserID: "user-2", Kind: "image", Status: model.ResourceStatusReady, Provider: "local", ObjectKey: "shared/object.png"},
}
if err := db.Create(&resources).Error; err != nil {
t.Fatal(err)
}
result, err := svc.DeleteAdminResources(admin, AdminResourceDeleteRequest{ResourceIDs: []string{"resource-delete"}})
if err != nil {
t.Fatal(err)
}
if len(result.Deleted) != 1 {
t.Fatalf("result = %#v", result)
}
assertModelCount(t, db, &model.Resource{}, "id = ?", 0, "resource-delete")
assertModelCount(t, db, &model.Resource{}, "id = ?", 1, "resource-keep")
assertModelCount(t, db, &model.ResourceDeletionJob{}, "1 = 1", 0)
var audit model.AdminAuditEvent
if err := db.First(&audit, "target_id = ?", "resource-delete").Error; err != nil {
t.Fatal(err)
}
if !strings.Contains(audit.MetadataJSON, `"physicalDeleteQueued":false`) {
t.Fatalf("audit metadata = %s", audit.MetadataJSON)
}
}
func TestDeleteAdminResourcesRollsBackWhenAuditInsertFails(t *testing.T) {
svc, db, _, admin := newAdminStorageDeleteTestService(t)
resource := model.Resource{ID: "resource-rollback", UserID: admin.ID, Kind: "image", Status: model.ResourceStatusReady, Provider: "local"}
binding := model.ArkPrivateAssetBinding{ID: "binding-1", UserID: admin.ID, ResourceID: resource.ID, ProjectName: "project", Status: "active"}
draft := model.AnnouncementImageDraft{ResourceID: resource.ID, UserID: admin.ID}
if err := db.Create(&resource).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&binding).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&draft).Error; err != nil {
t.Fatal(err)
}
if err := db.Exec("CREATE TRIGGER fail_resource_audit BEFORE INSERT ON admin_audit_events WHEN NEW.action = 'resource.delete' BEGIN SELECT RAISE(ABORT, 'forced audit failure'); END;").Error; err != nil {
t.Fatal(err)
}
if _, err := svc.DeleteAdminResources(admin, AdminResourceDeleteRequest{ResourceIDs: []string{resource.ID}}); err == nil {
t.Fatal("expected audit failure")
}
assertModelCount(t, db, &model.Resource{}, "id = ?", 1, resource.ID)
assertModelCount(t, db, &model.ArkPrivateAssetBinding{}, "resource_id = ?", 1, resource.ID)
assertModelCount(t, db, &model.AnnouncementImageDraft{}, "resource_id = ?", 1, resource.ID)
assertModelCount(t, db, &model.ResourceDeletionJob{}, "1 = 1", 0)
}
func TestDeleteAdminResourcesRejectsNonAdminAndOversizedBatch(t *testing.T) {
svc := &Service{}
user := &model.User{ID: "user", Role: model.UserRoleUser, Status: model.UserStatusActive}
if _, err := svc.DeleteAdminResources(user, AdminResourceDeleteRequest{ResourceIDs: []string{"resource-1"}}); err == nil {
t.Fatal("expected non-admin rejection")
}
ids := make([]string, maxAdminResourceDeleteCount+1)
for index := range ids {
ids[index] = newID()
}
if _, err := normalizeAdminResourceDeleteIDs(ids); err == nil {
t.Fatal("expected oversized batch rejection")
}
}
func newAdminStorageDeleteTestService(t *testing.T) (*Service, *gorm.DB, string, *model.User) {
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)
}
admin := &model.User{ID: "admin", Username: "admin", Role: model.UserRoleAdmin, Status: model.UserStatusActive}
if err := db.Create(admin).Error; err != nil {
t.Fatal(err)
}
dataDir := t.TempDir()
return New(repository.New(db), dataDir), db, dataDir, admin
}
func assertModelCount(t *testing.T, db *gorm.DB, value any, query string, expected int64, args ...any) {
t.Helper()
var count int64
dbQuery := db.Model(value).Where(query, args...)
if err := dbQuery.Count(&count).Error; err != nil {
t.Fatal(err)
}
if count != expected {
t.Fatalf("count = %d, want %d", count, expected)
}
}