yvan/kra-oa
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

优化结构

Browse Source
This commit is contained in:
Yvan
2026-08-22 12:13:16 +08:00
parent d987244d0e
commit b27a66e9d3
200 changed files with 1794 additions and 1715 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/worker/task_executor.go
+16 -16
View File
@@ -7,6 +7,7 @@ import (
"errors"
"fmt"
"io"
"kra/internal/biz/system"
"net"
"net/http"
"net/url"
@@ -14,24 +15,23 @@ import (
"syscall"
"time"
"kra/internal/biz"
"kra/internal/conf"
)
type TaskExecutor struct {
tasks *biz.TaskUsecase
media *biz.MediaUsecase
tasks *system.TaskUsecase
media *system.MediaUsecase
runtime *conf.Runtime
methods biz.TaskMethodRegistry
methods system.TaskMethodRegistry
}
func NewTaskExecutor(tasks *biz.TaskUsecase, media *biz.MediaUsecase, runtime *conf.Runtime) *TaskExecutor {
return NewTaskExecutorWithRegistry(tasks, media, runtime, biz.DefaultTaskMethodRegistry())
func NewTaskExecutor(tasks *system.TaskUsecase, media *system.MediaUsecase, runtime *conf.Runtime) *TaskExecutor {
return NewTaskExecutorWithRegistry(tasks, media, runtime, system.DefaultTaskMethodRegistry())
}
func NewTaskExecutorWithRegistry(tasks *biz.TaskUsecase, media *biz.MediaUsecase, runtime *conf.Runtime, methods biz.TaskMethodRegistry) *TaskExecutor {
func NewTaskExecutorWithRegistry(tasks *system.TaskUsecase, media *system.MediaUsecase, runtime *conf.Runtime, methods system.TaskMethodRegistry) *TaskExecutor {
if methods == nil {
methods = biz.DefaultTaskMethodRegistry()
methods = system.DefaultTaskMethodRegistry()
}
return &TaskExecutor{tasks: tasks, media: media, runtime: runtime, methods: methods}
}
@@ -61,7 +61,7 @@ func taskHTTPClient(allowPrivate bool) *http.Client {
return &http.Client{Timeout: 30 * time.Second, Transport: &http.Transport{Proxy: nil, DialContext: dialer.DialContext}}
}
func (e *TaskExecutor) runHTTP(ctx context.Context, task *biz.TimedTask) (string, error) {
func (e *TaskExecutor) runHTTP(ctx context.Context, task *system.TimedTask) (string, error) {
parsed, err := url.Parse(task.HTTPURL)
if err != nil {
return "", fmt.Errorf("URL 非法: %w", err)
@@ -109,13 +109,13 @@ func truncateTaskText(value string) string {
return value[:limit] + "...(截断)"
}
func (e *TaskExecutor) runMethod(ctx context.Context, task *biz.TimedTask) error {
var method biz.TaskMethodFunc
func (e *TaskExecutor) runMethod(ctx context.Context, task *system.TimedTask) error {
var method system.TaskMethodFunc
var ok bool
if e.methods != nil {
method, ok = e.methods.Lookup(task.MethodName)
} else {
method, ok = biz.TaskMethodByName(task.MethodName)
method, ok = system.TaskMethodByName(task.MethodName)
}
if !ok {
return fmt.Errorf("方法 %s 未注册(需通过 platform/task.Registry 注册)", task.MethodName)
@@ -145,12 +145,12 @@ func (e *TaskExecutor) runMethod(ctx context.Context, task *biz.TimedTask) error
}
}
func (e *TaskExecutor) Run(ctx context.Context, task *biz.TimedTask, trigger string) (log *biz.TimedTaskLog) {
func (e *TaskExecutor) Run(ctx context.Context, task *system.TimedTask, trigger string) (log *system.TimedTaskLog) {
if ctx == nil {
ctx = context.Background()
}
started := time.Now()
log = &biz.TimedTaskLog{TaskID: task.ID, TaskName: task.Name, TriggerType: trigger, StartedAt: started, Status: "success"}
log = &system.TimedTaskLog{TaskID: task.ID, TaskName: task.Name, TriggerType: trigger, StartedAt: started, Status: "success"}
defer func() {
if recovered := recover(); recovered != nil {
log.Status = "fail"
@@ -165,9 +165,9 @@ func (e *TaskExecutor) Run(ctx context.Context, task *biz.TimedTask, trigger str
}()
var err error
switch task.ExecutorType {
case biz.TaskExecutorMethod:
case system.TaskExecutorMethod:
err = e.runMethod(ctx, task)
case biz.TaskExecutorHTTP:
case system.TaskExecutorHTTP:
runCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
log.Output, err = e.runHTTP(runCtx, task)
cancel()
internal/worker/task_executor_test.go
+9 -10
View File
@@ -4,12 +4,11 @@ import (
"context"
"encoding/json"
"errors"
"kra/internal/biz/system"
"net"
"strings"
"testing"
"time"
"kra/internal/biz"
)
func TestPrivateIP(t *testing.T) {
@@ -25,25 +24,25 @@ func TestPrivateIP(t *testing.T) {
func TestRunMethodReportsParentDeadlineAsTimeout(t *testing.T) {
const methodName = "worker-test-parent-deadline"
biz.RegisterTaskMethod(methodName, "test", func(ctx context.Context, _ json.RawMessage) error {
system.RegisterTaskMethod(methodName, "test", func(ctx context.Context, _ json.RawMessage) error {
<-ctx.Done()
return ctx.Err()
})
executor := &TaskExecutor{}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
if err := executor.runMethod(ctx, &biz.TimedTask{MethodName: methodName}); !errors.Is(err, errTaskTimeout) {
if err := executor.runMethod(ctx, &system.TimedTask{MethodName: methodName}); !errors.Is(err, errTaskTimeout) {
t.Fatalf("runMethod() error = %v, want errTaskTimeout", err)
}
}
func TestRunMethodConvertsPanicToError(t *testing.T) {
const methodName = "worker-test-panic"
biz.RegisterTaskMethod(methodName, "test", func(context.Context, json.RawMessage) error {
system.RegisterTaskMethod(methodName, "test", func(context.Context, json.RawMessage) error {
panic("boom")
})
executor := &TaskExecutor{}
err := executor.runMethod(context.Background(), &biz.TimedTask{MethodName: methodName})
err := executor.runMethod(context.Background(), &system.TimedTask{MethodName: methodName})
if err == nil || !strings.Contains(err.Error(), "panic: boom") {
t.Fatalf("runMethod() error = %v, want recovered panic", err)
}
@@ -52,7 +51,7 @@ func TestRunMethodConvertsPanicToError(t *testing.T) {
func TestRunMethodUsesParentContext(t *testing.T) {
const methodName = "worker-test-parent-context"
started := make(chan struct{})
biz.RegisterTaskMethod(methodName, "test", func(ctx context.Context, _ json.RawMessage) error {
system.RegisterTaskMethod(methodName, "test", func(ctx context.Context, _ json.RawMessage) error {
close(started)
<-ctx.Done()
return ctx.Err()
@@ -61,7 +60,7 @@ func TestRunMethodUsesParentContext(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() {
done <- executor.runMethod(ctx, &biz.TimedTask{MethodName: methodName})
done <- executor.runMethod(ctx, &system.TimedTask{MethodName: methodName})
}()
select {
case <-started:
@@ -82,14 +81,14 @@ func TestRunMethodUsesParentContext(t *testing.T) {
func TestRunMethodHonorsDeadlineWhenMethodIgnoresContext(t *testing.T) {
const methodName = "worker-test-ignores-context"
release := make(chan struct{})
biz.RegisterTaskMethod(methodName, "test", func(context.Context, json.RawMessage) error {
system.RegisterTaskMethod(methodName, "test", func(context.Context, json.RawMessage) error {
<-release
return nil
})
executor := &TaskExecutor{}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
if err := executor.runMethod(ctx, &biz.TimedTask{MethodName: methodName}); !errors.Is(err, errTaskTimeout) {
if err := executor.runMethod(ctx, &system.TimedTask{MethodName: methodName}); !errors.Is(err, errTaskTimeout) {
close(release)
t.Fatalf("runMethod() error = %v, want errTaskTimeout", err)
}
internal/worker/task_registry.go
+6 -6
View File
@@ -3,8 +3,8 @@ package worker
import (
"context"
"encoding/json"
"kra/internal/biz/system"
"kra/internal/biz"
"kra/internal/conf"
platformtask "kra/pkg/task"
)
@@ -12,12 +12,12 @@ import (
// TaskMethods is the system module's dependency-bearing task contribution.
// Other modules provide their own contributor instead of editing this file.
type TaskMethods struct {
tasks *biz.TaskUsecase
media *biz.MediaUsecase
tasks *system.TaskUsecase
media *system.MediaUsecase
runtime *conf.Runtime
}
func NewTaskMethods(tasks *biz.TaskUsecase, media *biz.MediaUsecase, runtime *conf.Runtime) *TaskMethods {
func NewTaskMethods(tasks *system.TaskUsecase, media *system.MediaUsecase, runtime *conf.Runtime) *TaskMethods {
return &TaskMethods{tasks: tasks, media: media, runtime: runtime}
}
@@ -28,12 +28,12 @@ func (methods *TaskMethods) RegisterTasks(registry *platformtask.Registry) {
return
}
registry.Register(platformtask.Method{
Name: biz.TaskMethodClearDB, Description: "清理数据库过期日志(操作记录/JWT黑名单/定时任务执行日志)", Run: func(ctx context.Context, _ json.RawMessage) error {
Name: system.TaskMethodClearDB, Description: "清理数据库过期日志(操作记录/JWT黑名单/定时任务执行日志)", Run: func(ctx context.Context, _ json.RawMessage) error {
return methods.tasks.CleanupLogs(ctx)
},
})
registry.Register(platformtask.Method{
Name: biz.TaskMethodUploads, Description: "清理过期大文件上传会话", Run: func(ctx context.Context, _ json.RawMessage) error {
Name: system.TaskMethodUploads, Description: "清理过期大文件上传会话", Run: func(ctx context.Context, _ json.RawMessage) error {
ttl := 24
config := methods.runtime.Admin()
if config != nil && config.Media != nil && config.Media.SessionTtl > 0 {
internal/worker/task_runtime_test.go
+23 -24
View File
@@ -5,31 +5,30 @@ import (
"encoding/json"
"errors"
"io"
"kra/internal/biz/system"
"log/slog"
"sync"
"sync/atomic"
"testing"
"time"
"kra/internal/biz"
)
type workerTaskRepo struct {
mu sync.Mutex
items []*biz.TimedTask
items []*system.TimedTask
listErr error
logs chan *biz.TimedTaskLog
logs chan *system.TimedTaskLog
}
func (r *workerTaskRepo) CreateTask(context.Context, *biz.TimedTask) error { return nil }
func (r *workerTaskRepo) UpdateTask(context.Context, *biz.TimedTask) error { return nil }
func (r *workerTaskRepo) DeleteTask(context.Context, uint) error { return nil }
func (r *workerTaskRepo) ToggleTask(context.Context, uint, bool) error { return nil }
func (r *workerTaskRepo) CleanupLogs(context.Context) error { return nil }
func (r *workerTaskRepo) CreateTask(context.Context, *system.TimedTask) error { return nil }
func (r *workerTaskRepo) UpdateTask(context.Context, *system.TimedTask) error { return nil }
func (r *workerTaskRepo) DeleteTask(context.Context, uint) error { return nil }
func (r *workerTaskRepo) ToggleTask(context.Context, uint, bool) error { return nil }
func (r *workerTaskRepo) CleanupLogs(context.Context) error { return nil }
func (r *workerTaskRepo) TaskNameExists(context.Context, string, uint) (bool, error) {
return false, nil
}
func (r *workerTaskRepo) FindTask(_ context.Context, id uint) (*biz.TimedTask, error) {
func (r *workerTaskRepo) FindTask(_ context.Context, id uint) (*system.TimedTask, error) {
r.mu.Lock()
defer r.mu.Unlock()
for _, item := range r.items {
@@ -39,31 +38,31 @@ func (r *workerTaskRepo) FindTask(_ context.Context, id uint) (*biz.TimedTask, e
}
return nil, errors.New("task not found")
}
func (r *workerTaskRepo) ListTasks(context.Context, int, int, *biz.TimedTask) ([]*biz.TimedTask, int64, error) {
func (r *workerTaskRepo) ListTasks(context.Context, int, int, *system.TimedTask) ([]*system.TimedTask, int64, error) {
r.mu.Lock()
defer r.mu.Unlock()
if r.listErr != nil {
return nil, 0, r.listErr
}
items := make([]*biz.TimedTask, 0, len(r.items))
items := make([]*system.TimedTask, 0, len(r.items))
for _, item := range r.items {
items = append(items, cloneTimedTask(item))
}
return items, int64(len(items)), nil
}
func (r *workerTaskRepo) RecordTaskLog(_ context.Context, value *biz.TimedTaskLog) error {
func (r *workerTaskRepo) RecordTaskLog(_ context.Context, value *system.TimedTaskLog) error {
if r.logs != nil {
copy := *value
r.logs <- &copy
}
return nil
}
func (r *workerTaskRepo) ListTaskLogs(context.Context, int, int, uint, string) ([]*biz.TimedTaskLog, int64, error) {
func (r *workerTaskRepo) ListTaskLogs(context.Context, int, int, uint, string) ([]*system.TimedTaskLog, int64, error) {
return nil, 0, nil
}
func newTestTaskScheduler(repo *workerTaskRepo) *TaskScheduler {
tasks := biz.NewTaskUsecase(repo)
tasks := system.NewTaskUsecase(repo)
executor := &TaskExecutor{tasks: tasks}
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
return NewTaskScheduler(tasks, nil, executor, logger)
@@ -72,7 +71,7 @@ func newTestTaskScheduler(repo *workerTaskRepo) *TaskScheduler {
func TestReloadPreservesExistingScheduleOnPreparationFailure(t *testing.T) {
repo := &workerTaskRepo{}
scheduler := newTestTaskScheduler(repo)
old := &biz.TimedTask{ID: 1, Name: "old", Spec: "0 0 * * *", Enabled: true}
old := &system.TimedTask{ID: 1, Name: "old", Spec: "0 0 * * *", Enabled: true}
if err := scheduler.Schedule(old); err != nil {
t.Fatalf("schedule old task: %v", err)
}
@@ -90,7 +89,7 @@ func TestReloadPreservesExistingScheduleOnPreparationFailure(t *testing.T) {
t.Run("invalid cron", func(t *testing.T) {
repo.mu.Lock()
repo.listErr = nil
repo.items = []*biz.TimedTask{{ID: 2, Name: "invalid", Spec: "bad cron", Enabled: true}}
repo.items = []*system.TimedTask{{ID: 2, Name: "invalid", Spec: "bad cron", Enabled: true}}
repo.mu.Unlock()
if err := scheduler.Reload(context.Background()); err == nil {
t.Fatal("Reload() error = nil, want cron error")
@@ -122,7 +121,7 @@ func TestConcurrentScheduleKeepsSingleEntry(t *testing.T) {
wg.Add(1)
go func() {
defer wg.Done()
if err := scheduler.Schedule(&biz.TimedTask{ID: 7, Name: "same", Spec: "0 0 * * *", Enabled: true}); err != nil {
if err := scheduler.Schedule(&system.TimedTask{ID: 7, Name: "same", Spec: "0 0 * * *", Enabled: true}); err != nil {
t.Errorf("Schedule() error = %v", err)
}
}()
@@ -136,15 +135,15 @@ func TestManualTriggerSkipsOverlappingExecution(t *testing.T) {
started := make(chan struct{}, 2)
release := make(chan struct{})
var calls atomic.Int32
biz.RegisterTaskMethod(methodName, "test", func(context.Context, json.RawMessage) error {
system.RegisterTaskMethod(methodName, "test", func(context.Context, json.RawMessage) error {
calls.Add(1)
started <- struct{}{}
<-release
return nil
})
task := &biz.TimedTask{ID: 9, Name: "single", ExecutorType: biz.TaskExecutorMethod, MethodName: methodName}
repo := &workerTaskRepo{items: []*biz.TimedTask{task}, logs: make(chan *biz.TimedTaskLog, 2)}
task := &system.TimedTask{ID: 9, Name: "single", ExecutorType: system.TaskExecutorMethod, MethodName: methodName}
repo := &workerTaskRepo{items: []*system.TimedTask{task}, logs: make(chan *system.TimedTaskLog, 2)}
scheduler := newTestTaskScheduler(repo)
if err := scheduler.TriggerID(context.Background(), task.ID); err != nil {
t.Fatalf("first manual trigger failed: %v", err)
@@ -193,15 +192,15 @@ func TestAutomaticAndManualTriggersShareOverlapGate(t *testing.T) {
started := make(chan struct{}, 1)
release := make(chan struct{})
var calls atomic.Int32
biz.RegisterTaskMethod(methodName, "test", func(context.Context, json.RawMessage) error {
system.RegisterTaskMethod(methodName, "test", func(context.Context, json.RawMessage) error {
calls.Add(1)
started <- struct{}{}
<-release
return nil
})
task := &biz.TimedTask{ID: 10, Name: "shared-gate", ExecutorType: biz.TaskExecutorMethod, MethodName: methodName}
repo := &workerTaskRepo{logs: make(chan *biz.TimedTaskLog, 1)}
task := &system.TimedTask{ID: 10, Name: "shared-gate", ExecutorType: system.TaskExecutorMethod, MethodName: methodName}
repo := &workerTaskRepo{logs: make(chan *system.TimedTaskLog, 1)}
scheduler := newTestTaskScheduler(repo)
if !scheduler.dispatch(task, "auto", true) {
t.Fatal("automatic dispatch was rejected while idle")
internal/worker/task_scheduler.go
+13 -13
View File
@@ -5,17 +5,17 @@ import (
"encoding/json"
"errors"
"fmt"
"kra/internal/biz/system"
"log/slog"
"sync"
"time"
"github.com/robfig/cron/v3"
"kra/internal/biz"
)
type TaskScheduler struct {
tasks *biz.TaskUsecase
authorities *biz.AuthorityUsecase
tasks *system.TaskUsecase
authorities *system.AuthorityUsecase
executor *TaskExecutor
logger *slog.Logger
standard *cron.Cron
@@ -38,13 +38,13 @@ type scheduledEntry struct {
entry cron.EntryID
}
func NewTaskScheduler(tasks *biz.TaskUsecase, authorities *biz.AuthorityUsecase, executor *TaskExecutor, logger *slog.Logger) *TaskScheduler {
func NewTaskScheduler(tasks *system.TaskUsecase, authorities *system.AuthorityUsecase, executor *TaskExecutor, logger *slog.Logger) *TaskScheduler {
idle := make(chan struct{})
close(idle)
return &TaskScheduler{tasks: tasks, authorities: authorities, executor: executor, logger: logger.With("mod", "timedTask"), standard: cron.New(), seconds: cron.New(cron.WithSeconds()), entries: map[uint]scheduledEntry{}, running: map[uint]struct{}{}, idle: idle, subscribers: map[uint]map[chan []byte]struct{}{}}
}
func NewTaskRuntime(scheduler *TaskScheduler) biz.TaskRuntime { return scheduler }
func NewTaskRuntime(scheduler *TaskScheduler) system.TaskRuntime { return scheduler }
func (s *TaskScheduler) Start(ctx context.Context) error {
runContext, cancel := context.WithCancel(ctx)
@@ -116,7 +116,7 @@ func (s *TaskScheduler) Reload(ctx context.Context) error {
return err
}
type preparedTask struct {
task *biz.TimedTask
task *system.TimedTask
schedule cron.Schedule
}
prepared := make([]preparedTask, 0, len(items))
@@ -181,7 +181,7 @@ func (s *TaskScheduler) finishRun(id uint) {
}
}
func (s *TaskScheduler) dispatch(task *biz.TimedTask, trigger string, async bool) bool {
func (s *TaskScheduler) dispatch(task *system.TimedTask, trigger string, async bool) bool {
ctx, ok := s.beginRun(task.ID)
if !ok {
s.logger.Warn("timed task skipped because it is already running or the scheduler is stopping", "task_id", task.ID, "task_name", task.Name, "trigger_type", trigger)
@@ -199,7 +199,7 @@ func (s *TaskScheduler) dispatch(task *biz.TimedTask, trigger string, async bool
return true
}
func (s *TaskScheduler) run(ctx context.Context, task *biz.TimedTask, trigger string) {
func (s *TaskScheduler) run(ctx context.Context, task *system.TimedTask, trigger string) {
log := s.executor.Run(ctx, task, trigger)
attributes := []any{"task_id", log.TaskID, "task_name", log.TaskName, "trigger_type", log.TriggerType, "status", log.Status, "duration_ms", log.DurationMS, "started_at", log.StartedAt, "finished_at", log.FinishedAt}
if log.ErrorMsg != "" {
@@ -222,7 +222,7 @@ func (s *TaskScheduler) run(ctx context.Context, task *biz.TimedTask, trigger st
}
}
func (s *TaskScheduler) Schedule(task *biz.TimedTask) error {
func (s *TaskScheduler) Schedule(task *system.TimedTask) error {
if task == nil {
return errors.New("定时任务不能为空")
}
@@ -244,21 +244,21 @@ func (s *TaskScheduler) Schedule(task *biz.TimedTask) error {
return nil
}
func parseTaskSchedule(task *biz.TimedTask) (cron.Schedule, error) {
func parseTaskSchedule(task *system.TimedTask) (cron.Schedule, error) {
if task.WithSeconds {
return cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor).Parse(task.Spec)
}
return cron.ParseStandard(task.Spec)
}
func cloneTimedTask(task *biz.TimedTask) *biz.TimedTask {
func cloneTimedTask(task *system.TimedTask) *system.TimedTask {
copy := *task
copy.Params = append([]byte(nil), task.Params...)
copy.HTTPHeader = append([]byte(nil), task.HTTPHeader...)
return &copy
}
func (s *TaskScheduler) scheduleLocked(task *biz.TimedTask, schedule cron.Schedule) {
func (s *TaskScheduler) scheduleLocked(task *system.TimedTask, schedule cron.Schedule) {
run := func() {
s.dispatch(task, "auto", false)
}
@@ -317,7 +317,7 @@ func (s *TaskScheduler) NextRuns() map[uint]time.Time {
return out
}
func (s *TaskScheduler) Trigger(task *biz.TimedTask) bool {
func (s *TaskScheduler) Trigger(task *system.TimedTask) bool {
if task == nil {
return false
}