mayfly-go/server/internal/db/api/db_data_sync.go

package api

import (
	"mayfly-go/internal/db/api/form"
	"mayfly-go/internal/db/api/vo"
	"mayfly-go/internal/db/application"
	"mayfly-go/internal/db/domain/entity"
	"mayfly-go/internal/db/imsg"
	"mayfly-go/pkg/biz"
	"mayfly-go/pkg/req"
	"mayfly-go/pkg/utils/cryptox"
	"mayfly-go/pkg/utils/stringx"
	"strings"

	"github.com/may-fly/cast"
)

type DataSyncTask struct {
	dataSyncTaskApp application.DataSyncTask `inject:"T"`
}

func (d *DataSyncTask) ReqConfs() *req.Confs {
	reqs := [...]*req.Conf{
		// 获取任务列表 /datasync
		req.NewGet("", d.Tasks),

		req.NewGet(":taskId/logs", d.Logs).RequiredPermissionCode("db:sync:log"),

		// 保存任务 /datasync/save
		req.NewPost("save", d.SaveTask).Log(req.NewLogSaveI(imsg.LogDataSyncSave)).RequiredPermissionCode("db:sync:save"),

		// 获取单个详情 /datasync/:taskId
		req.NewGet(":taskId", d.GetTask),

		// 删除任务 /datasync/:taskId/del
		req.NewDelete(":taskId/del", d.DeleteTask).Log(req.NewLogSaveI(imsg.LogDataSyncDelete)).RequiredPermissionCode("db:sync:del"),

		// 启停用任务 /datasync/status
		req.NewPost(":taskId/status", d.ChangeStatus).Log(req.NewLogSaveI(imsg.LogDataSyncChangeStatus)).RequiredPermissionCode("db:sync:status"),

		// 立即执行任务 /datasync/run
		req.NewPost(":taskId/run", d.Run),

		// 停止正在执行中的任务
		req.NewPost(":taskId/stop", d.Stop),
	}

	return req.NewConfs("/datasync/tasks", reqs[:]...)
}

func (d *DataSyncTask) Tasks(rc *req.Ctx) {
	queryCond, page := req.BindQueryAndPage[*entity.DataSyncTaskQuery](rc, new(entity.DataSyncTaskQuery))
	res, err := d.dataSyncTaskApp.GetPageList(queryCond, page, new([]vo.DataSyncTaskListVO))
	biz.ErrIsNil(err)
	rc.ResData = res
}

func (d *DataSyncTask) Logs(rc *req.Ctx) {
	queryCond, page := req.BindQueryAndPage[*entity.DataSyncLogQuery](rc, new(entity.DataSyncLogQuery))
	res, err := d.dataSyncTaskApp.GetTaskLogList(queryCond, page, new([]vo.DataSyncLogListVO))
	biz.ErrIsNil(err)
	rc.ResData = res
}

func (d *DataSyncTask) SaveTask(rc *req.Ctx) {
	form := &form.DataSyncTaskForm{}
	task := req.BindJsonAndCopyTo[*entity.DataSyncTask](rc, form, new(entity.DataSyncTask))

	// 解码base64 sql
	sqlStr, err := cryptox.AesDecryptByLa(task.DataSql, rc.GetLoginAccount())
	biz.ErrIsNilAppendErr(err, "sql decoding failure: %s")
	sql := stringx.TrimSpaceAndBr(sqlStr)
	task.DataSql = sql
	form.DataSql = sql

	rc.ReqParam = form
	biz.ErrIsNil(d.dataSyncTaskApp.Save(rc.MetaCtx, task))
}

func (d *DataSyncTask) DeleteTask(rc *req.Ctx) {
	taskId := rc.PathParam("taskId")
	rc.ReqParam = taskId
	ids := strings.Split(taskId, ",")

	for _, v := range ids {
		biz.ErrIsNil(d.dataSyncTaskApp.Delete(rc.MetaCtx, cast.ToUint64(v)))
	}
}

func (d *DataSyncTask) ChangeStatus(rc *req.Ctx) {
	form := &form.DataSyncTaskStatusForm{}
	task := req.BindJsonAndCopyTo[*entity.DataSyncTask](rc, form, new(entity.DataSyncTask))
	_ = d.dataSyncTaskApp.UpdateById(rc.MetaCtx, task)

	if task.Status == entity.DataSyncTaskStatusEnable {
		task, err := d.dataSyncTaskApp.GetById(task.Id)
		biz.ErrIsNil(err, "task not found")
		d.dataSyncTaskApp.AddCronJob(rc.MetaCtx, task)
	} else {
		d.dataSyncTaskApp.RemoveCronJobById(task.Id)
	}
	// 记录请求日志
	rc.ReqParam = form
}

func (d *DataSyncTask) Run(rc *req.Ctx) {
	taskId := d.getTaskId(rc)
	rc.ReqParam = taskId
	_ = d.dataSyncTaskApp.RunCronJob(rc.MetaCtx, taskId)
}

func (d *DataSyncTask) Stop(rc *req.Ctx) {
	taskId := d.getTaskId(rc)
	rc.ReqParam = taskId

	task := new(entity.DataSyncTask)
	task.Id = taskId
	task.RunningState = entity.DataSyncTaskRunStateStop
	_ = d.dataSyncTaskApp.UpdateById(rc.MetaCtx, task)
}

func (d *DataSyncTask) GetTask(rc *req.Ctx) {
	taskId := d.getTaskId(rc)
	dbEntity, _ := d.dataSyncTaskApp.GetById(taskId)
	rc.ResData = dbEntity
}

func (d *DataSyncTask) getTaskId(rc *req.Ctx) uint64 {
	instanceId := rc.PathParamInt("taskId")
	biz.IsTrue(instanceId > 0, "instanceId error")
	return uint64(instanceId)
}