break
5e552d99bc
Constraint: 在线热修前必须保存当前迭代分支全部有效代码进展 Confidence: medium Scope-risk: broad Directive: 后续修改需保持审计事件与业务事务边界一致 Tested: git diff --cached --check Not-tested: 未运行全量测试，提交用于切换分支前保存既有工作
371 lines
13 KiB
Go
371 lines
13 KiB
Go
package export_task
|
|
|
|
import (
|
|
"context"
|
|
stderrors "errors"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/bytedance/sonic"
|
|
"github.com/hibiken/asynq"
|
|
"gorm.io/datatypes"
|
|
"gorm.io/gorm"
|
|
|
|
"github.com/break/junhong_cmp_fiber/internal/exporter"
|
|
"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
|
|
"github.com/break/junhong_cmp_fiber/internal/model"
|
|
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
|
"github.com/break/junhong_cmp_fiber/internal/store"
|
|
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
|
"github.com/break/junhong_cmp_fiber/pkg/auditfailure"
|
|
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
|
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
|
"github.com/break/junhong_cmp_fiber/pkg/middleware"
|
|
"github.com/break/junhong_cmp_fiber/pkg/queue"
|
|
"github.com/break/junhong_cmp_fiber/pkg/storage"
|
|
)
|
|
|
|
// Service 导出任务服务。
|
|
type Service struct {
|
|
db *gorm.DB
|
|
taskStore *postgres.ExportTaskStore
|
|
queueClient *queue.Client
|
|
storageSvc *storage.Service
|
|
sceneRegistry *exporter.Registry
|
|
auditWriter *audit.Writer
|
|
}
|
|
|
|
type dispatchPayload struct {
|
|
TaskID uint `json:"task_id"`
|
|
}
|
|
|
|
// New 创建导出任务服务。
|
|
func New(db *gorm.DB, taskStore *postgres.ExportTaskStore, queueClient *queue.Client, storageSvc *storage.Service, auditWriters ...*audit.Writer) *Service {
|
|
service := &Service{
|
|
db: db,
|
|
taskStore: taskStore,
|
|
queueClient: queueClient,
|
|
storageSvc: storageSvc,
|
|
sceneRegistry: exporter.NewDefaultRegistry(db),
|
|
}
|
|
if len(auditWriters) > 0 {
|
|
service.auditWriter = auditWriters[0]
|
|
}
|
|
return service
|
|
}
|
|
|
|
// CreateTask 创建导出任务并入队 dispatch。
|
|
func (s *Service) CreateTask(ctx context.Context, req *dto.CreateExportTaskRequest) (*dto.CreateExportTaskResponse, error) {
|
|
userID := middleware.GetUserIDFromContext(ctx)
|
|
if userID == 0 {
|
|
return nil, errors.New(errors.CodeUnauthorized)
|
|
}
|
|
|
|
userType := middleware.GetUserTypeFromContext(ctx)
|
|
if userType != constants.UserTypeSuperAdmin && userType != constants.UserTypePlatform && userType != constants.UserTypeAgent {
|
|
return nil, errors.New(errors.CodeForbidden, "当前账号无权限创建导出任务")
|
|
}
|
|
|
|
if req.Scene == "" || !s.sceneRegistry.IsSupported(req.Scene) {
|
|
return nil, errors.New(errors.CodeInvalidParam, "导出场景不支持")
|
|
}
|
|
if req.Format != constants.ExportTaskFormatXLSX && req.Format != constants.ExportTaskFormatCSV {
|
|
return nil, errors.New(errors.CodeInvalidParam, "导出格式不支持")
|
|
}
|
|
|
|
queryJSON := datatypes.JSON("{}")
|
|
if req.Query != nil {
|
|
raw, err := sonic.Marshal(req.Query)
|
|
if err != nil {
|
|
return nil, errors.Wrap(errors.CodeInvalidParam, err, "导出筛选参数格式不正确")
|
|
}
|
|
queryJSON = datatypes.JSON(raw)
|
|
}
|
|
|
|
shopID := middleware.GetShopIDFromContext(ctx)
|
|
enterpriseID := middleware.GetEnterpriseIDFromContext(ctx)
|
|
|
|
var (
|
|
creatorShopID *uint
|
|
creatorEnterpriseID *uint
|
|
)
|
|
if shopID > 0 {
|
|
creatorShopID = &shopID
|
|
}
|
|
if enterpriseID > 0 {
|
|
creatorEnterpriseID = &enterpriseID
|
|
}
|
|
|
|
scopeShopIDs := model.UIntListJSON{}
|
|
if userType == constants.UserTypeAgent {
|
|
subordinateShopIDs := middleware.GetSubordinateShopIDs(ctx)
|
|
if subordinateShopIDs == nil {
|
|
if shopID == 0 {
|
|
return nil, errors.New(errors.CodeForbidden, "代理账号缺少店铺信息")
|
|
}
|
|
subordinateShopIDs = []uint{shopID}
|
|
}
|
|
if len(subordinateShopIDs) == 0 {
|
|
return nil, errors.New(errors.CodeForbidden, "当前账号无可导出的数据范围")
|
|
}
|
|
scopeShopIDs = subordinateShopIDs
|
|
}
|
|
|
|
task := &model.ExportTask{
|
|
TaskNo: s.taskStore.GenerateTaskNo(),
|
|
Scene: req.Scene,
|
|
Format: req.Format,
|
|
Status: constants.ExportTaskStatusPending,
|
|
Progress: 0,
|
|
QueryJSON: queryJSON,
|
|
ScopeShopIDs: scopeShopIDs,
|
|
CreatorUserID: userID,
|
|
CreatorUserType: userType,
|
|
CreatorShopID: creatorShopID,
|
|
CreatorEnterpriseID: creatorEnterpriseID,
|
|
}
|
|
task.Creator = userID
|
|
task.Updater = userID
|
|
|
|
if s.auditWriter == nil {
|
|
return nil, errors.New(errors.CodeInvalidStatus, "导出任务统一审计接缝未配置")
|
|
}
|
|
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
if err := s.taskStore.WithTx(tx).Create(ctx, task); err != nil {
|
|
return err
|
|
}
|
|
return s.writeTaskAudit(ctx, tx, constants.AuditActionExportTaskCreated, "创建业务导出任务", task, nil, exportTaskState(task), constants.AuditResultSuccess, "created", "", "")
|
|
}); err != nil {
|
|
s.recordTaskAudit(ctx, constants.AuditActionExportTaskCreated, "创建业务导出任务失败", task, nil, exportTaskState(task), constants.AuditResultFailed, "create_failed", errors.CodeDatabaseError)
|
|
return nil, errors.Wrap(errors.CodeDatabaseError, err, "创建导出任务失败")
|
|
}
|
|
|
|
if err := s.queueClient.EnqueueTask(
|
|
ctx,
|
|
constants.TaskTypeExportDispatch,
|
|
dispatchPayload{TaskID: task.ID},
|
|
asynq.MaxRetry(constants.ExportDispatchRetryMax),
|
|
asynq.Timeout(constants.ExportDispatchTaskTimeout),
|
|
asynq.Queue(constants.QueueForTaskType(constants.TaskTypeExportDispatch)),
|
|
); err != nil {
|
|
secondaryErr := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
if err := s.taskStore.WithTx(tx).MarkFailed(ctx, task.ID, userID, "导出任务入队失败"); err != nil {
|
|
return err
|
|
}
|
|
before := exportTaskState(task)
|
|
task.Status = constants.ExportTaskStatusFailed
|
|
task.ErrorMessage = "导出任务入队失败"
|
|
return s.writeTaskAudit(ctx, tx, constants.AuditActionExportTaskCreated, "导出任务入队失败", task, before, exportTaskState(task), constants.AuditResultFailed, "enqueue_failed", strconv.Itoa(errors.CodeTaskQueueError), "导出任务入队失败")
|
|
})
|
|
if secondaryErr != nil {
|
|
auditfailure.RecordSecondaryWriteFailure(constants.AuditActionExportTaskCreated, task.TaskNo, "", task.TaskNo, strconv.Itoa(errors.CodeTaskQueueError), secondaryErr)
|
|
}
|
|
return nil, errors.Wrap(errors.CodeTaskQueueError, err, "导出任务入队失败")
|
|
}
|
|
|
|
return &dto.CreateExportTaskResponse{
|
|
TaskID: task.ID,
|
|
TaskNo: task.TaskNo,
|
|
Status: task.Status,
|
|
StatusName: constants.GetExportTaskStatusName(task.Status),
|
|
Message: "导出任务创建成功，系统将异步处理",
|
|
}, nil
|
|
}
|
|
|
|
// ListTasks 查询导出任务列表。
|
|
func (s *Service) ListTasks(ctx context.Context, req *dto.ListExportTaskRequest) (*dto.ListExportTaskResponse, error) {
|
|
page := req.Page
|
|
if page <= 0 {
|
|
page = 1
|
|
}
|
|
|
|
pageSize := req.PageSize
|
|
if pageSize <= 0 {
|
|
pageSize = constants.DefaultPageSize
|
|
}
|
|
if pageSize > constants.MaxPageSize {
|
|
pageSize = constants.MaxPageSize
|
|
}
|
|
|
|
filters := make(map[string]any)
|
|
if req.Scene != "" {
|
|
filters["scene"] = req.Scene
|
|
}
|
|
if req.Status != nil {
|
|
filters["status"] = *req.Status
|
|
}
|
|
if req.StartTime != nil {
|
|
filters["start_time"] = *req.StartTime
|
|
}
|
|
if req.EndTime != nil {
|
|
filters["end_time"] = *req.EndTime
|
|
}
|
|
|
|
items, total, err := s.taskStore.List(ctx, &store.QueryOptions{
|
|
Page: page,
|
|
PageSize: pageSize,
|
|
OrderBy: "created_at DESC",
|
|
}, filters)
|
|
if err != nil {
|
|
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询导出任务列表失败")
|
|
}
|
|
|
|
result := make([]*dto.ExportTaskItem, 0, len(items))
|
|
for _, item := range items {
|
|
result = append(result, toTaskItemDTO(item))
|
|
}
|
|
|
|
return &dto.ListExportTaskResponse{
|
|
List: result,
|
|
Total: total,
|
|
Page: page,
|
|
PageSize: pageSize,
|
|
}, nil
|
|
}
|
|
|
|
// GetTaskDetail 查询导出任务详情。
|
|
func (s *Service) GetTaskDetail(ctx context.Context, id uint) (*dto.ExportTaskDetailResponse, error) {
|
|
task, err := s.taskStore.GetByID(ctx, id)
|
|
if err != nil {
|
|
if err == gorm.ErrRecordNotFound {
|
|
return nil, errors.New(errors.CodeForbidden, "无权限操作该资源或资源不存在")
|
|
}
|
|
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询导出任务详情失败")
|
|
}
|
|
|
|
resp := &dto.ExportTaskDetailResponse{ExportTaskItem: *toTaskItemDTO(task)}
|
|
if task.Status == constants.ExportTaskStatusCompleted && task.FileKey != "" && s.storageSvc != nil && s.storageSvc.Provider() != nil {
|
|
url, err := s.storageSvc.Provider().GetDownloadURL(ctx, task.FileKey, constants.ExportDownloadURLExpire)
|
|
if err != nil {
|
|
return nil, errors.Wrap(errors.CodeInternalError, err, "生成下载链接失败")
|
|
}
|
|
expiresAt := time.Now().Add(constants.ExportDownloadURLExpire)
|
|
resp.DownloadURL = url
|
|
resp.DownloadExpiresAt = &expiresAt
|
|
}
|
|
|
|
return resp, nil
|
|
}
|
|
|
|
// CancelTask 取消导出任务。
|
|
func (s *Service) CancelTask(ctx context.Context, id uint) (*dto.CancelExportTaskResponse, error) {
|
|
userID := middleware.GetUserIDFromContext(ctx)
|
|
if userID == 0 {
|
|
return nil, errors.New(errors.CodeUnauthorized)
|
|
}
|
|
|
|
task, err := s.taskStore.GetByID(ctx, id)
|
|
if err != nil {
|
|
if err == gorm.ErrRecordNotFound {
|
|
return nil, errors.New(errors.CodeForbidden, "无权限操作该资源或资源不存在")
|
|
}
|
|
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询导出任务失败")
|
|
}
|
|
|
|
if s.auditWriter == nil {
|
|
return nil, errors.New(errors.CodeInvalidStatus, "导出任务统一审计接缝未配置")
|
|
}
|
|
message := "取消请求已提交"
|
|
before := exportTaskState(task)
|
|
changed := false
|
|
err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
txStore := s.taskStore.WithTx(tx)
|
|
switch task.Status {
|
|
case constants.ExportTaskStatusPending:
|
|
ok, updateErr := txStore.CancelPendingTask(ctx, id, userID)
|
|
if updateErr != nil {
|
|
return updateErr
|
|
}
|
|
if !ok {
|
|
return errors.New(errors.CodeInvalidStatus, "当前状态不支持取消")
|
|
}
|
|
task.Status, task.CancelRequested, task.Progress = constants.ExportTaskStatusCancelled, true, 100
|
|
message, changed = "任务已取消", true
|
|
case constants.ExportTaskStatusProcessing:
|
|
if task.CancelRequested {
|
|
message = "取消请求已提交，请稍后刷新状态"
|
|
return nil
|
|
}
|
|
ok, updateErr := txStore.SetCancelRequested(ctx, id, userID)
|
|
if updateErr != nil {
|
|
return updateErr
|
|
}
|
|
if !ok {
|
|
return errors.New(errors.CodeInvalidStatus, "当前状态不支持取消")
|
|
}
|
|
task.CancelRequested, changed = true, true
|
|
case constants.ExportTaskStatusCompleted, constants.ExportTaskStatusFailed, constants.ExportTaskStatusCancelled:
|
|
return errors.New(errors.CodeInvalidStatus, "当前状态不支持取消")
|
|
default:
|
|
return errors.New(errors.CodeInvalidStatus, "当前状态不支持取消")
|
|
}
|
|
if !changed {
|
|
return nil
|
|
}
|
|
return s.writeTaskAudit(ctx, tx, constants.AuditActionExportTaskCancelled, message, task, before, exportTaskState(task), constants.AuditResultSuccess, "cancelled", "", "")
|
|
})
|
|
if err != nil {
|
|
result := constants.AuditResultFailed
|
|
errorCode := errors.CodeDatabaseError
|
|
var appErr *errors.AppError
|
|
if stderrors.As(err, &appErr) && appErr.Code == errors.CodeInvalidStatus {
|
|
result = constants.AuditResultDenied
|
|
errorCode = appErr.Code
|
|
}
|
|
s.recordTaskAudit(ctx, constants.AuditActionExportTaskCancelled, "取消业务导出任务失败", task, before, exportTaskState(task), result, "", errorCode)
|
|
if appErr != nil {
|
|
return nil, appErr
|
|
}
|
|
return nil, errors.Wrap(errors.CodeDatabaseError, err, "取消导出任务失败")
|
|
}
|
|
|
|
return &dto.CancelExportTaskResponse{
|
|
TaskID: task.ID,
|
|
Status: task.Status,
|
|
StatusName: constants.GetExportTaskStatusName(task.Status),
|
|
CancelRequested: task.CancelRequested,
|
|
Message: message,
|
|
}, nil
|
|
}
|
|
|
|
func toTaskItemDTO(task *model.ExportTask) *dto.ExportTaskItem {
|
|
return &dto.ExportTaskItem{
|
|
ID: task.ID,
|
|
TaskID: task.ID,
|
|
TaskNo: task.TaskNo,
|
|
Scene: task.Scene,
|
|
Format: task.Format,
|
|
Status: task.Status,
|
|
StatusName: constants.GetExportTaskStatusName(task.Status),
|
|
Progress: task.Progress,
|
|
TotalRows: task.TotalRows,
|
|
ProcessedRows: task.ProcessedRows,
|
|
TotalShards: task.TotalShards,
|
|
SuccessShards: task.SuccessShards,
|
|
FailedShards: task.FailedShards,
|
|
TotalCount: task.TotalShards,
|
|
SuccessCount: task.SuccessShards,
|
|
FailedCount: task.FailedShards,
|
|
CancelRequested: task.CancelRequested,
|
|
FileKey: task.FileKey,
|
|
ErrorMessage: task.ErrorMessage,
|
|
ErrorCode: exportTaskErrorCode(task),
|
|
ErrorSummary: task.ErrorMessage,
|
|
CreatedAt: task.CreatedAt,
|
|
UpdatedAt: task.UpdatedAt,
|
|
StartedAt: task.StartedAt,
|
|
CompletedAt: task.CompletedAt,
|
|
CreatorUserID: task.CreatorUserID,
|
|
CreatorUserType: task.CreatorUserType,
|
|
CreatorShopID: task.CreatorShopID,
|
|
CreatorEnterpriseID: task.CreatorEnterpriseID,
|
|
}
|
|
}
|
|
|
|
func exportTaskErrorCode(task *model.ExportTask) string {
|
|
if task.ErrorMessage == "" {
|
|
return ""
|
|
}
|
|
return "EXPORT_TASK_FAILED"
|
|
}
|