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

fix: align admin behavior and Kratos integrations

Browse Source
This commit is contained in:
yvan
2026-08-14 14:14:51 +08:00
parent 8c4c6ba661
commit e0b0e1d8c8
123 changed files with 4334 additions and 1877 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/worker/task_scheduler.go
+37 -22
View File
@@ -21,6 +21,9 @@ type TaskScheduler struct {
seconds *cron.Cron
mu sync.Mutex
entries map[uint]scheduledEntry
ctxMu sync.RWMutex
runContext context.Context
cancel context.CancelFunc
subMu sync.RWMutex
subscribers map[chan []byte]struct{}
}
@@ -35,15 +38,18 @@ func NewTaskScheduler(service *service.TaskService, logger *slog.Logger) *TaskSc
}
func (s *TaskScheduler) Start(ctx context.Context) error {
runContext, cancel := context.WithCancel(ctx)
s.ctxMu.Lock()
s.runContext, s.cancel = runContext, cancel
s.ctxMu.Unlock()
s.standard.Start()
s.seconds.Start()
items, _, err := s.service.Tasks(ctx, 1, 100000, nil, nil)
items, err := s.service.ScheduledTasks(ctx)
if err == nil {
for _, raw := range items {
id, _ := raw["ID"].(uint)
if task, findErr := s.service.Task(ctx, id); findErr == nil && task.Enabled {
for _, task := range items {
if task.Enabled {
if scheduleErr := s.Schedule(task); scheduleErr != nil {
s.logger.ErrorContext(ctx, "restore timed task failed", "id", id, "error", scheduleErr)
s.logger.ErrorContext(ctx, "restore timed task failed", "id", task.ID, "error", scheduleErr)
}
}
}
@@ -55,6 +61,11 @@ func (s *TaskScheduler) Start(ctx context.Context) error {
}
func (s *TaskScheduler) Stop(ctx context.Context) error {
s.ctxMu.Lock()
if s.cancel != nil {
s.cancel()
}
s.ctxMu.Unlock()
standardDone, secondsDone := s.standard.Stop().Done(), s.seconds.Stop().Done()
select {
case <-standardDone:
@@ -93,16 +104,11 @@ func (s *TaskScheduler) Reload(ctx context.Context) error {
delete(s.entries, id)
}
s.mu.Unlock()
items, _, err := s.service.Tasks(ctx, 1, 100000, nil, nil)
items, err := s.service.ScheduledTasks(ctx)
if err != nil {
return err
}
for _, raw := range items {
id, _ := raw["ID"].(uint)
task, findErr := s.service.Task(ctx, id)
if findErr != nil {
return findErr
}
for _, task := range items {
if task.Enabled {
if err = s.Schedule(task); err != nil {
return err
@@ -112,6 +118,22 @@ func (s *TaskScheduler) Reload(ctx context.Context) error {
return nil
}
func (s *TaskScheduler) executionContext() context.Context {
s.ctxMu.RLock()
defer s.ctxMu.RUnlock()
if s.runContext != nil {
return s.runContext
}
return context.Background()
}
func (s *TaskScheduler) run(task *biz.TimedTask, trigger string) {
log := s.service.Run(s.executionContext(), task, trigger)
if log.Status != "success" {
s.Broadcast(map[string]any{"taskId": task.ID, "taskName": task.Name, "status": log.Status, "errorMsg": log.ErrorMsg, "time": time.Now()})
}
}
func (s *TaskScheduler) Schedule(task *biz.TimedTask) error {
s.Remove(task.ID)
if !task.Enabled {
@@ -119,10 +141,7 @@ func (s *TaskScheduler) Schedule(task *biz.TimedTask) error {
}
copy := *task
run := func() {
log := s.service.Run(context.Background(), &copy, "auto")
if log.Status != "success" {
s.Broadcast(map[string]any{"taskId": copy.ID, "taskName": copy.Name, "status": log.Status, "errorMsg": log.ErrorMsg, "time": time.Now()})
}
s.run(&copy, "auto")
}
var id cron.EntryID
var err error
@@ -172,12 +191,8 @@ func (s *TaskScheduler) NextRuns() map[uint]time.Time {
}
func (s *TaskScheduler) Trigger(task *biz.TimedTask) {
go func() {
log := s.service.Run(context.Background(), task, "manual")
if log.Status != "success" {
s.Broadcast(map[string]any{"taskId": task.ID, "taskName": task.Name, "status": log.Status, "errorMsg": log.ErrorMsg, "time": time.Now()})
}
}()
copy := *task
go s.run(&copy, "manual")
}
func (s *TaskScheduler) Subscribe() chan []byte {