feat(data_export): add workflow support for data export #784

internal/dms/biz/data_export_workflow.go

@@ -117,6 +117,7 @@ type WorkflowRepo interface {
 	GetDataExportWorkflowsByDBServices(ctx context.Context, dbUid []string) ([]string, error)
 	DeleteDataExportWorkflowsByIds(ctx context.Context, dataExportWorkflowUid []string) error
 	GetGlobalWorkflowsByParameterMap(ctx context.Context, data map[string]interface{}) ([]*Workflow, int64, error)
+	AuditWorkflowAndAdvanceStep(ctx context.Context, workflowRecordUid string, step *WorkflowStep, nextStepId uint64, operateId, reason string) error
 }
 
 type DataExportWorkflowUsecase struct {

internal/dms/service/data_export_workflow.go

@@ -147,7 +147,11 @@ func (d *DMSService) ListDataExportWorkflow(ctx context.Context, req *dmsV1.List
 			ret[i].Creater = creater[0]
 		}
 		// 结束时不显示当前步骤操作人，其他状态显示当前步骤操作人
-		if w.Status != string(dmsV1.StatusFinish) {
+		if w.Status == string(dmsV1.DataExportWorkflowStatusWaitForExport) || w.Status == string(dmsV1.DataExportWorkflowStatusWaitForExporting) {
+			// wait_for_export/wait_for_exporting 状态下，待操作人为工单创建者
+			ret[i].CurrentStepAssigneeUsers = convertBizUidWithName(d.UserUsecase.GetBizUserIncludeDeletedWithNameByUids(ctx, []string{w.CreateUserUID}))
+		} else if w.Status != string(dmsV1.StatusFinish) && w.WorkflowRecord.CurrentWorkflowStepId > 0 &&
+			int(w.WorkflowRecord.CurrentWorkflowStepId-1) < len(w.WorkflowRecord.WorkflowSteps) {
 			ret[i].CurrentStepAssigneeUsers = convertBizUidWithName(d.UserUsecase.GetBizUserIncludeDeletedWithNameByUids(ctx, w.WorkflowRecord.WorkflowSteps[w.WorkflowRecord.CurrentWorkflowStepId-1].Assignees))
 		}
 	}
@@ -192,7 +196,11 @@ func (d *DMSService) GetGlobalWorkflowsList(ctx context.Context, req *dmsV1.Filt
 			ret[i].Creater = creater[0]
 		}
 		// 结束时不显示当前步骤操作人，其他状态显示当前步骤操作人
-		if w.Status != string(dmsV1.StatusFinish) {
+		if w.Status == string(dmsV1.DataExportWorkflowStatusWaitForExport) || w.Status == string(dmsV1.DataExportWorkflowStatusWaitForExporting) {
+			// wait_for_export/wait_for_exporting 状态下，待操作人为工单创建者
+			ret[i].CurrentStepAssigneeUsers = convertBizUidWithName(d.UserUsecase.GetBizUserIncludeDeletedWithNameByUids(ctx, []string{w.CreateUserUID}))
+		} else if w.Status != string(dmsV1.StatusFinish) && w.WorkflowRecord.CurrentWorkflowStepId > 0 &&
+			int(w.WorkflowRecord.CurrentWorkflowStepId-1) < len(w.WorkflowRecord.WorkflowSteps) {
 			ret[i].CurrentStepAssigneeUsers = convertBizUidWithName(d.UserUsecase.GetBizUserIncludeDeletedWithNameByUids(ctx, w.WorkflowRecord.WorkflowSteps[w.WorkflowRecord.CurrentWorkflowStepId-1].Assignees))
 		}
 	}

internal/dms/storage/model/model.go

@@ -257,7 +257,7 @@ const (
 type ProxyTarget struct {
 	Name            string `json:"name" gorm:"primaryKey;size:200;not null;column:name"`
 	Url             string `json:"url" gorm:"size:255;column:url"`
-	Version         string `json:"version" gorm:"size:64;column:version"`
+	Version         string `json:"version" gorm:"size:512;column:version"`
 	ProxyUrlPrefixs string `json:"proxy_url_prefixs" gorm:"size:255;column:proxy_url_prefixs"`
 	Scenario        string `json:"scenario" gorm:"size:64;column:scenario;default:'internal_service'"`
 }

internal/dms/storage/workflow.go

@@ -285,6 +285,24 @@ func (d *WorkflowRepo) AuditWorkflow(ctx context.Context, dataExportWorkflowUid
 	})
 }
 
+func (d *WorkflowRepo) AuditWorkflowAndAdvanceStep(ctx context.Context, workflowRecordUid string, step *biz.WorkflowStep, nextStepId uint64, operateId, reason string) error {
+	return transaction(d.log, ctx, d.db, func(tx *gorm.DB) error {
+		// update current step state
+		operateTime := time.Now()
+		fields := map[string]interface{}{"operation_user_uid": operateId, "operate_at": operateTime, "reason": reason, "state": step.State}
+		if err := tx.WithContext(ctx).Model(&model.WorkflowStep{}).Where("step_id = ? and workflow_record_uid = ?", step.StepId, step.WorkflowRecordUid).Updates(fields).Error; err != nil {
+			return fmt.Errorf("failed to update current workflow step, err: %v", err)
+		}
+
+		// advance CurrentWorkflowStepId to next step
+		if err := tx.WithContext(ctx).Model(&model.WorkflowRecord{}).Where("uid = ?", workflowRecordUid).Update("current_workflow_step_id", nextStepId).Error; err != nil {
+			return fmt.Errorf("failed to advance workflow step, err: %v", err)
+		}
+
+		return nil
+	})
+}
+
 func (d *WorkflowRepo) UpdateWorkflowStatusById(ctx context.Context, dataExportWorkflowUid string, status biz.DataExportWorkflowStatus) error {
 	return transaction(d.log, ctx, d.db, func(tx *gorm.DB) error {
 		if err := tx.WithContext(ctx).Model(&model.WorkflowRecord{}).Where("uid = ?", dataExportWorkflowUid).Update("status", status).Error; err != nil {