优化

internal/biz/task.go

@@ -42,6 +42,13 @@ const (
 	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 TaskMethod struct {
 	Name        string
 	Description string
@@ -77,6 +84,17 @@ type TaskRepo interface {
 	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() chan []byte
+	Unsubscribe(chan []byte)
+}
+
 type TaskUsecase struct{ TaskRepo }
 
 func NewTaskUsecase(repo TaskRepo) *TaskUsecase { return &TaskUsecase{TaskRepo: repo} }
@@ -149,3 +167,69 @@ func (uc *TaskUsecase) Update(ctx context.Context, value *TimedTask) error {
 	}
 	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) Create(ctx context.Context, value *TimedTask) error {
+	if err := uc.tasks.Create(ctx, value); err != nil {
+		return err
+	}
+	if err := uc.runtime.ScheduleID(ctx, value.ID); err != nil {
+		return &TaskScheduleError{Err: err}
+	}
+	return nil
+}
+
+func (uc *TaskApplicationUsecase) Update(ctx context.Context, value *TimedTask) error {
+	if err := uc.tasks.Update(ctx, value); err != nil {
+		return err
+	}
+	if err := uc.runtime.ScheduleID(ctx, value.ID); err != nil {
+		return &TaskScheduleError{Err: err}
+	}
+	return nil
+}
+
+func (uc *TaskApplicationUsecase) Delete(ctx context.Context, id uint) error {
+	uc.runtime.Remove(id)
+	return uc.tasks.DeleteTask(ctx, id)
+}
+
+func (uc *TaskApplicationUsecase) Toggle(ctx context.Context, id uint, enabled bool) error {
+	if err := uc.tasks.ToggleTask(ctx, id, enabled); err != nil {
+		return err
+	}
+	if err := uc.runtime.ScheduleID(ctx, id); err != nil {
+		return &TaskScheduleError{Err: err}
+	}
+	return nil
+}
+
+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() chan []byte { return uc.runtime.Subscribe() }
+func (uc *TaskApplicationUsecase) Unsubscribe(events chan []byte) {
+	uc.runtime.Unsubscribe(events)
+}