241 lines
7.0 KiB
Go
241 lines
7.0 KiB
Go
package system
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"time"
|
|
|
|
"github.com/robfig/cron/v3"
|
|
)
|
|
|
|
type TimedTask struct {
|
|
ID uint
|
|
CreatedAt time.Time
|
|
UpdatedAt time.Time
|
|
Name, Description, Spec string
|
|
WithSeconds bool
|
|
ExecutorType, MethodName string
|
|
Params []byte
|
|
HTTPURL, HTTPMethod string
|
|
HTTPHeader []byte
|
|
HTTPBody string
|
|
HTTPAllowPrivate, Enabled bool
|
|
EnabledFilter *bool
|
|
}
|
|
type TimedTaskLog struct {
|
|
ID uint
|
|
CreatedAt, UpdatedAt time.Time
|
|
TaskID uint
|
|
TaskName, TriggerType string
|
|
StartedAt, FinishedAt time.Time
|
|
DurationMS int64
|
|
Status, ErrorMsg, Output string
|
|
}
|
|
|
|
const (
|
|
TaskExecutorMethod = "method"
|
|
TaskExecutorHTTP = "http"
|
|
TaskMethodClearDB = "ClearDB"
|
|
TaskMethodUploads = "CleanStaleUploads"
|
|
)
|
|
|
|
var ErrTaskSchedule = errors.New("task schedule failed")
|
|
|
|
type TaskScheduleError struct{ Err error }
|
|
|
|
func (e *TaskScheduleError) Error() string { return e.Err.Error() }
|
|
func (e *TaskScheduleError) Is(target error) bool { return target == ErrTaskSchedule }
|
|
|
|
type TaskRepo interface {
|
|
CreateTask(context.Context, *TimedTask) error
|
|
UpdateTask(context.Context, *TimedTask) error
|
|
DeleteTask(context.Context, uint) error
|
|
FindTask(context.Context, uint) (*TimedTask, error)
|
|
ListTasks(context.Context, int, int, *TimedTask) ([]*TimedTask, int64, error)
|
|
ToggleTask(context.Context, uint, bool) error
|
|
RecordTaskLog(context.Context, *TimedTaskLog) error
|
|
ListTaskLogs(context.Context, int, int, uint, string) ([]*TimedTaskLog, int64, error)
|
|
CleanupLogs(context.Context) error
|
|
TaskNameExists(context.Context, string, uint) (bool, error)
|
|
}
|
|
|
|
type TaskRuntime interface {
|
|
ScheduleID(context.Context, uint) error
|
|
Remove(uint)
|
|
TriggerID(context.Context, uint) error
|
|
NextRuns() map[uint]time.Time
|
|
Reload(context.Context) error
|
|
Subscribe(uint) chan []byte
|
|
Unsubscribe(uint, chan []byte)
|
|
}
|
|
|
|
type TaskUsecase struct {
|
|
TaskRepo
|
|
methods TaskMethodRegistry
|
|
}
|
|
|
|
func NewTaskUsecase(repo TaskRepo) *TaskUsecase {
|
|
return NewTaskUsecaseWithRegistry(repo, DefaultTaskMethodRegistry())
|
|
}
|
|
|
|
func NewTaskUsecaseWithRegistry(repo TaskRepo, methods TaskMethodRegistry) *TaskUsecase {
|
|
if methods == nil {
|
|
methods = DefaultTaskMethodRegistry()
|
|
}
|
|
return &TaskUsecase{TaskRepo: repo, methods: methods}
|
|
}
|
|
|
|
func (uc *TaskUsecase) RegisteredMethods() []TaskMethod { return uc.methods.List() }
|
|
|
|
func (uc *TaskUsecase) Validate(value *TimedTask) error {
|
|
if value.Name == "" {
|
|
return errors.New("任务名不能为空")
|
|
}
|
|
var err error
|
|
if value.WithSeconds {
|
|
_, err = cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor).Parse(value.Spec)
|
|
} else {
|
|
_, err = cron.ParseStandard(value.Spec)
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("cron 表达式非法: %w", err)
|
|
}
|
|
switch value.ExecutorType {
|
|
case TaskExecutorMethod:
|
|
if _, ok := uc.methods.Lookup(value.MethodName); !ok {
|
|
return fmt.Errorf("方法 %s 未注册", value.MethodName)
|
|
}
|
|
if len(value.Params) > 0 && !json.Valid(value.Params) {
|
|
return errors.New("params 必须是合法 JSON")
|
|
}
|
|
case TaskExecutorHTTP:
|
|
parsed, parseErr := url.Parse(value.HTTPURL)
|
|
if parseErr != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" {
|
|
return errors.New("httpUrl 必须是合法的 http/https 地址")
|
|
}
|
|
if len(value.HTTPHeader) > 0 {
|
|
headers := map[string]string{}
|
|
if json.Unmarshal(value.HTTPHeader, &headers) != nil {
|
|
return errors.New(`httpHeader 必须是 {"Key":"Value"} 形式的 JSON 对象`)
|
|
}
|
|
}
|
|
default:
|
|
return errors.New("executorType 必须为 method 或 http")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (uc *TaskUsecase) Create(ctx context.Context, value *TimedTask) error {
|
|
if err := uc.Validate(value); err != nil {
|
|
return err
|
|
}
|
|
exists, err := uc.TaskNameExists(ctx, value.Name, 0)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if exists {
|
|
return fmt.Errorf("任务名 %s 已存在", value.Name)
|
|
}
|
|
return uc.CreateTask(ctx, value)
|
|
}
|
|
|
|
func (uc *TaskUsecase) Update(ctx context.Context, value *TimedTask) error {
|
|
if value.ID == 0 {
|
|
return errors.New("缺少任务 ID")
|
|
}
|
|
if err := uc.Validate(value); err != nil {
|
|
return err
|
|
}
|
|
exists, err := uc.TaskNameExists(ctx, value.Name, value.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if exists {
|
|
return fmt.Errorf("任务名 %s 已存在", value.Name)
|
|
}
|
|
return uc.UpdateTask(ctx, value)
|
|
}
|
|
|
|
type TaskApplicationUsecase struct {
|
|
tasks *TaskUsecase
|
|
runtime TaskRuntime
|
|
}
|
|
|
|
func NewTaskApplicationUsecase(tasks *TaskUsecase, runtime TaskRuntime) *TaskApplicationUsecase {
|
|
return &TaskApplicationUsecase{tasks: tasks, runtime: runtime}
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) RegisteredMethods() []TaskMethod {
|
|
return uc.tasks.RegisteredMethods()
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) syncRuntime(ctx context.Context, id uint) error {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
if err := uc.runtime.ScheduleID(ctx, id); err != nil {
|
|
repairCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
|
defer cancel()
|
|
if reloadErr := uc.runtime.Reload(repairCtx); reloadErr != nil {
|
|
return &TaskScheduleError{Err: errors.Join(err, fmt.Errorf("重载任务运行时失败: %w", reloadErr))}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) Create(ctx context.Context, value *TimedTask) error {
|
|
if err := uc.tasks.Create(ctx, value); err != nil {
|
|
return err
|
|
}
|
|
return uc.syncRuntime(ctx, value.ID)
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) Update(ctx context.Context, value *TimedTask) error {
|
|
if err := uc.tasks.Update(ctx, value); err != nil {
|
|
return err
|
|
}
|
|
return uc.syncRuntime(ctx, value.ID)
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) Delete(ctx context.Context, id uint) error {
|
|
if err := uc.tasks.DeleteTask(ctx, id); err != nil {
|
|
return err
|
|
}
|
|
uc.runtime.Remove(id)
|
|
return nil
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) Toggle(ctx context.Context, id uint, enabled bool) error {
|
|
if err := uc.tasks.ToggleTask(ctx, id, enabled); err != nil {
|
|
return err
|
|
}
|
|
return uc.syncRuntime(ctx, id)
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) Trigger(ctx context.Context, id uint) error {
|
|
return uc.runtime.TriggerID(ctx, id)
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) List(ctx context.Context, page, size int, filter *TimedTask) ([]*TimedTask, int64, map[uint]time.Time, error) {
|
|
items, total, err := uc.tasks.ListTasks(ctx, page, size, filter)
|
|
return items, total, uc.runtime.NextRuns(), err
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) Logs(ctx context.Context, page, size int, taskID uint, status string) ([]*TimedTaskLog, int64, error) {
|
|
return uc.tasks.ListTaskLogs(ctx, page, size, taskID, status)
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) Reload(ctx context.Context) error {
|
|
return uc.runtime.Reload(ctx)
|
|
}
|
|
|
|
func (uc *TaskApplicationUsecase) Subscribe(userID uint) chan []byte {
|
|
return uc.runtime.Subscribe(userID)
|
|
}
|
|
func (uc *TaskApplicationUsecase) Unsubscribe(userID uint, events chan []byte) {
|
|
uc.runtime.Unsubscribe(userID, events)
|
|
}
|