优化

internal/worker/task_executor.go

@@ -56,7 +56,7 @@ func taskHTTPClient(allowPrivate bool) *http.Client {
 func (e *TaskExecutor) runHTTP(ctx context.Context, task *biz.TimedTask) (string, error) {
 	parsed, err := url.Parse(task.HTTPURL)
 	if err != nil {
-		return "", err
+		return "", fmt.Errorf("URL 非法: %w", err)
 	}
 	if parsed.Scheme != "http" && parsed.Scheme != "https" {
 		return "", fmt.Errorf("仅允许 http/https, 实际为 %q", parsed.Scheme)
@@ -70,7 +70,11 @@ func (e *TaskExecutor) runHTTP(ctx context.Context, task *biz.TimedTask) (string
 		return "", err
 	}
 	headers := map[string]string{}
-	_ = json.Unmarshal(task.HTTPHeader, &headers)
+	if len(task.HTTPHeader) > 0 {
+		if err = json.Unmarshal(task.HTTPHeader, &headers); err != nil {
+			return "", fmt.Errorf("http_header 必须是 JSON 对象: %w", err)
+		}
+	}
 	for key, value := range headers {
 		request.Header.Set(key, value)
 	}
@@ -118,6 +122,8 @@ func (e *TaskExecutor) runMethod(task *biz.TimedTask) error {
 				ttl = int(config.Media.SessionTtl)
 			}
 			err = e.media.CleanupStale(ctx, ttl)
+		default:
+			err = fmt.Errorf("方法 %s 未注册(需在 initialize/timer.go 经 task.Register 注册)", task.MethodName)
 		}
 		done <- err
 	}()

internal/worker/task_scheduler.go

@@ -14,6 +14,7 @@ import (
 
 type TaskScheduler struct {
 	tasks       *biz.TaskUsecase
+	authorities *biz.AuthorityUsecase
 	executor    *TaskExecutor
 	logger      *slog.Logger
 	standard    *cron.Cron
@@ -24,7 +25,7 @@ type TaskScheduler struct {
 	runContext  context.Context
 	cancel      context.CancelFunc
 	subMu       sync.RWMutex
-	subscribers map[chan []byte]struct{}
+	subscribers map[uint]map[chan []byte]struct{}
 }
 
 type scheduledEntry struct {
@@ -32,8 +33,8 @@ type scheduledEntry struct {
 	entry   cron.EntryID
 }
 
-func NewTaskScheduler(tasks *biz.TaskUsecase, executor *TaskExecutor, logger *slog.Logger) *TaskScheduler {
-	return &TaskScheduler{tasks: tasks, executor: executor, logger: logger, standard: cron.New(), seconds: cron.New(cron.WithSeconds()), entries: map[uint]scheduledEntry{}, subscribers: map[chan []byte]struct{}{}}
+func NewTaskScheduler(tasks *biz.TaskUsecase, authorities *biz.AuthorityUsecase, executor *TaskExecutor, logger *slog.Logger) *TaskScheduler {
+	return &TaskScheduler{tasks: tasks, authorities: authorities, executor: executor, logger: logger, standard: cron.New(), seconds: cron.New(cron.WithSeconds()), entries: map[uint]scheduledEntry{}, subscribers: map[uint]map[chan []byte]struct{}{}}
 }
 
 func NewTaskRuntime(scheduler *TaskScheduler) biz.TaskRuntime { return scheduler }
@@ -131,7 +132,12 @@ func (s *TaskScheduler) executionContext() context.Context {
 func (s *TaskScheduler) run(task *biz.TimedTask, trigger string) {
 	log := s.executor.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()})
+		ids, err := s.authorities.AuthorityUserIDs(context.Background(), 888)
+		if err != nil {
+			s.logger.Error("query timed task alert recipients failed", "error", err)
+			return
+		}
+		s.PublishToUsers(ids, map[string]any{"taskId": task.ID, "name": task.Name, "error": log.ErrorMsg, "time": time.Now().Format(time.RFC3339)})
 	}
 }
 
@@ -196,32 +202,44 @@ func (s *TaskScheduler) Trigger(task *biz.TimedTask) {
 	go s.run(&copy, "manual")
 }
 
-func (s *TaskScheduler) Subscribe() chan []byte {
+func (s *TaskScheduler) Subscribe(userID uint) chan []byte {
 	ch := make(chan []byte, 16)
 	s.subMu.Lock()
-	s.subscribers[ch] = struct{}{}
+	if s.subscribers[userID] == nil {
+		s.subscribers[userID] = map[chan []byte]struct{}{}
+	}
+	s.subscribers[userID][ch] = struct{}{}
 	s.subMu.Unlock()
 	return ch
 }
-func (s *TaskScheduler) Unsubscribe(ch chan []byte) {
+func (s *TaskScheduler) Unsubscribe(userID uint, ch chan []byte) {
 	s.subMu.Lock()
-	if _, ok := s.subscribers[ch]; ok {
-		delete(s.subscribers, ch)
+	if subscribers := s.subscribers[userID]; subscribers != nil {
+		if _, ok := subscribers[ch]; !ok {
+			s.subMu.Unlock()
+			return
+		}
+		delete(subscribers, ch)
+		if len(subscribers) == 0 {
+			delete(s.subscribers, userID)
+		}
 		close(ch)
 	}
 	s.subMu.Unlock()
 }
-func (s *TaskScheduler) Broadcast(value any) {
+func (s *TaskScheduler) PublishToUsers(userIDs []uint, value any) {
 	raw, err := json.Marshal(value)
 	if err != nil {
 		raw = []byte(fmt.Sprint(value))
 	}
 	s.subMu.RLock()
 	defer s.subMu.RUnlock()
-	for ch := range s.subscribers {
-		select {
-		case ch <- raw:
-		default:
+	for _, userID := range userIDs {
+		for ch := range s.subscribers[userID] {
+			select {
+			case ch <- raw:
+			default:
+			}
 		}
 	}
 }