kra-new/internal/biz/task/task.go

package task

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"time"

	"github.com/robfig/cron/v3"
	platformtask "kra/pkg/task"
)

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)
	CleanupTaskLogs(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 NewTaskUsecaseWithRegistry(repo TaskRepo, methods TaskMethodRegistry) *TaskUsecase {
	if methods == nil {
		methods = platformtask.NewRegistry()
	}
	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)
}