优化结构

internal/data/data_scope.go

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 	"errors"
 	"kra/internal/biz/system"
+	datasystem "kra/internal/data/system"
 	"log/slog"
 	"reflect"
 	"strings"
@@ -13,6 +14,10 @@ import (
 	"gorm.io/gorm/schema"
 )
 
+// dataAccessLogPO is the infrastructure-side write model used by GORM
+// callbacks. The system module owns the query repository for the same table.
+type dataAccessLogPO = datasystem.DataAccessLogPO
+
 // registerDataScopeCallbacks installs the global GORM data-scope engine.
 // System tables are deliberately excluded: their access is controlled by
 // Casbin, while ownership columns on business tables are row-level scope.

internal/data/data_scope_record.go

@@ -1,9 +0,0 @@
-package data
-
-import (
-	datasystem "kra/internal/data/system"
-)
-
-// dataAccessLogPO is the infrastructure-side write model used by GORM
-// callbacks. The system module owns the query repository for the same table.
-type dataAccessLogPO = datasystem.DataAccessLogPO

internal/data/system/media_upload.go

@@ -90,11 +90,17 @@ func (r *mediaRepo) FailUploadSession(ctx context.Context, id uint) error {
 	return r.data.DB().WithContext(ctx).Model(&uploadSessionPO{}).Where("id = ?", id).Update("status", "failed").Error
 }
 func (r *mediaRepo) CompleteUploadSession(ctx context.Context, id uint, key string, mediaID uint) error {
-	return r.data.DB().WithContext(ctx).Model(&uploadSessionPO{}).Where("id = ?", id).Updates(map[string]any{"status": "completed", "storage_key": key, "media_id": mediaID}).Error
+	result := r.data.DB().WithContext(ctx).Model(&uploadSessionPO{}).Where("id = ? AND status = ?", id, "merging").Updates(map[string]any{"status": "completed", "storage_key": key, "media_id": mediaID})
+	if result.Error != nil {
+		return result.Error
+	}
+	if result.RowsAffected != 1 {
+		return system.ErrUploadSessionNotFound
+	}
+	return nil
 }
 func (r *mediaRepo) DeleteUploadSession(ctx context.Context, id uint) error {
-	// The compatible flow uses GORM's normal Delete here, retaining the soft-deleted session
-	// for audit/recovery rather than physically removing it.
+	// Upload sessions are staging records; cleanup physically removes them and their chunks.
 	return r.data.DB().WithContext(ctx).Unscoped().Delete(&uploadSessionPO{}, id).Error
 }
 func (r *mediaRepo) UpsertChunk(ctx context.Context, uploadID uint, v *system.UploadChunk) error {

internal/data/task/provider.go

@@ -1,13 +1,5 @@
 package task
 
-import (
-	dataprovider "kra/internal/data/provider"
-
-	"github.com/google/wire"
-)
-
-// Provider is the minimal database seam required by the task tables.
-// Task persistence must not depend on the full data runtime or system repos.
-type Provider = dataprovider.Database
+import "github.com/google/wire"
 
 var ProviderSet = wire.NewSet(NewTaskRepo)

internal/data/task/task.go

@@ -3,6 +3,7 @@ package task
 import (
 	"context"
 	taskbiz "kra/internal/biz/task"
+	dataprovider "kra/internal/data/provider"
 	"kra/pkg/database/gormkit"
 	"kra/pkg/database/pagination"
 	"time"
@@ -44,9 +45,9 @@ type taskLogPO struct {
 
 func (taskLogPO) TableName() string { return "sys_timed_task_logs" }
 
-type taskRepo struct{ data Provider }
+type taskRepo struct{ data dataprovider.Database }
 
-func NewTaskRepo(data Provider) taskbiz.TaskRepo { return &taskRepo{data: data} }
+func NewTaskRepo(data dataprovider.Database) taskbiz.TaskRepo { return &taskRepo{data: data} }
 func taskToPO(v *taskbiz.TimedTask) taskPO {
 	return taskPO{ID: v.ID, Name: v.Name, Description: v.Description, Spec: v.Spec, WithSeconds: v.WithSeconds, ExecutorType: v.ExecutorType, MethodName: v.MethodName, Params: gormkit.JSON(v.Params), HTTPURL: v.HTTPURL, HTTPMethod: v.HTTPMethod, HTTPHeader: gormkit.JSON(v.HTTPHeader), HTTPBody: v.HTTPBody, HTTPAllowPrivate: v.HTTPAllowPrivate, Enabled: v.Enabled}
 }