优化结构
This commit is contained in:
Yvan
@@ -0,0 +1,9 @@
|
||||
package task
|
||||
|
||||
import "github.com/google/wire"
|
||||
|
||||
// ProviderSet wires timed-task usecases independently from the system domain.
|
||||
var ProviderSet = wire.NewSet(
|
||||
NewTaskUsecaseWithRegistry,
|
||||
NewTaskApplicationUsecase,
|
||||
)
|
||||
@@ -0,0 +1,240 @@
|
||||
package task
|
||||
|
||||
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)
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package task
|
||||
|
||||
import platformtask "kra/pkg/task"
|
||||
|
||||
type TaskMethodFunc = platformtask.MethodFunc
|
||||
type TaskMethod = platformtask.Method
|
||||
|
||||
type TaskMethodRegistry interface {
|
||||
Register(platformtask.Method)
|
||||
Lookup(string) (TaskMethodFunc, bool)
|
||||
List() []TaskMethod
|
||||
}
|
||||
|
||||
var defaultTaskMethods = platformtask.NewRegistry()
|
||||
|
||||
// RegisterTaskMethod remains as a compatibility helper for tests and callers
|
||||
// that have not moved to constructor injection yet.
|
||||
func RegisterTaskMethod(name, description string, fn TaskMethodFunc) {
|
||||
defaultTaskMethods.Register(platformtask.Method{Name: name, Description: description, Run: fn})
|
||||
}
|
||||
|
||||
func TaskMethodByName(name string) (TaskMethodFunc, bool) {
|
||||
return defaultTaskMethods.Lookup(name)
|
||||
}
|
||||
|
||||
func RegisteredTaskMethods() []TaskMethod { return defaultTaskMethods.List() }
|
||||
|
||||
func DefaultTaskMethodRegistry() TaskMethodRegistry { return defaultTaskMethods }
|
||||
@@ -0,0 +1,197 @@
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type applicationTaskRepo struct {
|
||||
mu sync.Mutex
|
||||
deleted bool
|
||||
deleteErr error
|
||||
createdID uint
|
||||
toggleCalls int
|
||||
updateCalls int
|
||||
createdCalls int
|
||||
nameExists bool
|
||||
nameExistsErr error
|
||||
deleteObserved chan struct{}
|
||||
}
|
||||
|
||||
func (r *applicationTaskRepo) CreateTask(_ context.Context, value *TimedTask) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.createdCalls++
|
||||
if value.ID == 0 {
|
||||
value.ID = r.createdID
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (r *applicationTaskRepo) UpdateTask(context.Context, *TimedTask) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.updateCalls++
|
||||
return nil
|
||||
}
|
||||
func (r *applicationTaskRepo) DeleteTask(context.Context, uint) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.deleteErr != nil {
|
||||
return r.deleteErr
|
||||
}
|
||||
r.deleted = true
|
||||
if r.deleteObserved != nil {
|
||||
close(r.deleteObserved)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (r *applicationTaskRepo) FindTask(context.Context, uint) (*TimedTask, error) {
|
||||
return nil, errors.New("not implemented")
|
||||
}
|
||||
func (r *applicationTaskRepo) ListTasks(context.Context, int, int, *TimedTask) ([]*TimedTask, int64, error) {
|
||||
return nil, 0, nil
|
||||
}
|
||||
func (r *applicationTaskRepo) ToggleTask(context.Context, uint, bool) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.toggleCalls++
|
||||
return nil
|
||||
}
|
||||
func (r *applicationTaskRepo) RecordTaskLog(context.Context, *TimedTaskLog) error { return nil }
|
||||
func (r *applicationTaskRepo) ListTaskLogs(context.Context, int, int, uint, string) ([]*TimedTaskLog, int64, error) {
|
||||
return nil, 0, nil
|
||||
}
|
||||
func (r *applicationTaskRepo) CleanupLogs(context.Context) error { return nil }
|
||||
func (r *applicationTaskRepo) TaskNameExists(context.Context, string, uint) (bool, error) {
|
||||
return r.nameExists, r.nameExistsErr
|
||||
}
|
||||
|
||||
type applicationTaskRuntime struct {
|
||||
mu sync.Mutex
|
||||
scheduleErr error
|
||||
reloadErr error
|
||||
scheduleCalls int
|
||||
reloadCalls int
|
||||
removeCalls int
|
||||
removeBeforeDelete bool
|
||||
repo *applicationTaskRepo
|
||||
reloadContextActive bool
|
||||
}
|
||||
|
||||
func (r *applicationTaskRuntime) ScheduleID(context.Context, uint) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.scheduleCalls++
|
||||
return r.scheduleErr
|
||||
}
|
||||
func (r *applicationTaskRuntime) Remove(uint) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.removeCalls++
|
||||
if r.repo != nil {
|
||||
r.repo.mu.Lock()
|
||||
r.removeBeforeDelete = !r.repo.deleted
|
||||
r.repo.mu.Unlock()
|
||||
}
|
||||
}
|
||||
func (r *applicationTaskRuntime) TriggerID(context.Context, uint) error { return nil }
|
||||
func (r *applicationTaskRuntime) NextRuns() map[uint]time.Time { return nil }
|
||||
func (r *applicationTaskRuntime) Reload(ctx context.Context) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.reloadCalls++
|
||||
r.reloadContextActive = ctx.Err() == nil
|
||||
return r.reloadErr
|
||||
}
|
||||
func (r *applicationTaskRuntime) Subscribe(uint) chan []byte { return make(chan []byte) }
|
||||
func (r *applicationTaskRuntime) Unsubscribe(uint, chan []byte) {}
|
||||
|
||||
func validApplicationTask(id uint) *TimedTask {
|
||||
const methodName = "biz-test-application-task"
|
||||
RegisterTaskMethod(methodName, "test", func(context.Context, json.RawMessage) error { return nil })
|
||||
return &TimedTask{ID: id, Name: "test", Spec: "0 0 * * *", ExecutorType: TaskExecutorMethod, MethodName: methodName, Enabled: true}
|
||||
}
|
||||
|
||||
func TestTaskApplicationRepairsRuntimeAfterScheduleFailure(t *testing.T) {
|
||||
repo := &applicationTaskRepo{createdID: 41}
|
||||
runtime := &applicationTaskRuntime{scheduleErr: errors.New("schedule failed")}
|
||||
uc := NewTaskApplicationUsecase(NewTaskUsecase(repo), runtime)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
if err := uc.Create(ctx, validApplicationTask(0)); err != nil {
|
||||
t.Fatalf("Create() error = %v, want successful reload compensation", err)
|
||||
}
|
||||
if runtime.scheduleCalls != 1 || runtime.reloadCalls != 1 {
|
||||
t.Fatalf("runtime calls = schedule:%d reload:%d, want 1 and 1", runtime.scheduleCalls, runtime.reloadCalls)
|
||||
}
|
||||
if !runtime.reloadContextActive {
|
||||
t.Fatal("reload compensation inherited the canceled request context")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskApplicationReturnsScheduleErrorWhenRepairFails(t *testing.T) {
|
||||
repo := &applicationTaskRepo{createdID: 42}
|
||||
runtime := &applicationTaskRuntime{scheduleErr: errors.New("schedule failed"), reloadErr: errors.New("reload failed")}
|
||||
uc := NewTaskApplicationUsecase(NewTaskUsecase(repo), runtime)
|
||||
|
||||
err := uc.Create(context.Background(), validApplicationTask(0))
|
||||
if !errors.Is(err, ErrTaskSchedule) {
|
||||
t.Fatalf("Create() error = %v, want ErrTaskSchedule", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskApplicationUpdateAndToggleUseRuntimeRepair(t *testing.T) {
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
run func(*TaskApplicationUsecase) error
|
||||
}{
|
||||
{name: "update", run: func(uc *TaskApplicationUsecase) error {
|
||||
return uc.Update(context.Background(), validApplicationTask(7))
|
||||
}},
|
||||
{name: "toggle", run: func(uc *TaskApplicationUsecase) error { return uc.Toggle(context.Background(), 7, true) }},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
repo := &applicationTaskRepo{}
|
||||
runtime := &applicationTaskRuntime{scheduleErr: errors.New("schedule failed")}
|
||||
uc := NewTaskApplicationUsecase(NewTaskUsecase(repo), runtime)
|
||||
if err := test.run(uc); err != nil {
|
||||
t.Fatalf("operation error = %v", err)
|
||||
}
|
||||
if runtime.scheduleCalls != 1 || runtime.reloadCalls != 1 {
|
||||
t.Fatalf("runtime calls = schedule:%d reload:%d, want 1 and 1", runtime.scheduleCalls, runtime.reloadCalls)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskApplicationDeleteRemovesRuntimeAfterDatabase(t *testing.T) {
|
||||
repo := &applicationTaskRepo{}
|
||||
runtime := &applicationTaskRuntime{repo: repo}
|
||||
uc := NewTaskApplicationUsecase(NewTaskUsecase(repo), runtime)
|
||||
if err := uc.Delete(context.Background(), 8); err != nil {
|
||||
t.Fatalf("Delete() error = %v", err)
|
||||
}
|
||||
if runtime.removeBeforeDelete {
|
||||
t.Fatal("runtime entry was removed before the database row")
|
||||
}
|
||||
if runtime.removeCalls != 1 {
|
||||
t.Fatalf("Remove() calls = %d, want 1", runtime.removeCalls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskApplicationDeleteKeepsRuntimeWhenDatabaseFails(t *testing.T) {
|
||||
repo := &applicationTaskRepo{deleteErr: errors.New("delete failed")}
|
||||
runtime := &applicationTaskRuntime{repo: repo}
|
||||
uc := NewTaskApplicationUsecase(NewTaskUsecase(repo), runtime)
|
||||
if err := uc.Delete(context.Background(), 8); err == nil {
|
||||
t.Fatal("Delete() error = nil, want database error")
|
||||
}
|
||||
if runtime.removeCalls != 0 {
|
||||
t.Fatalf("Remove() calls = %d, want 0", runtime.removeCalls)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user