package task import ( "context" "fmt" "sync" "time" notify_domain "git.huangwc.com/pig/pig-farm-controller/internal/domain/notify" "git.huangwc.com/pig/pig-farm-controller/internal/domain/plan" "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" "git.huangwc.com/pig/pig-farm-controller/internal/infra/models" "git.huangwc.com/pig/pig-farm-controller/internal/infra/notify" "git.huangwc.com/pig/pig-farm-controller/internal/infra/repository" ) // AlarmNotificationTaskParams 定义了 AlarmNotificationTask 的参数结构 // 如果用户没有指定某个等级的配置, 则默认为该等级消息只发送一次 type AlarmNotificationTaskParams struct { // NotificationIntervals 告警通知的发送间隔时间,键为告警等级,值为时间间隔(分钟) NotificationIntervals map[models.SeverityLevel]uint32 `json:"notification_intervals"` } // AlarmNotificationTask 告警通知发送任务 type AlarmNotificationTask struct { ctx context.Context taskLog *models.TaskExecutionLog // alarmNotificationTaskParams 是任务配置 alarmNotificationTaskParams AlarmNotificationTaskParams onceParse sync.Once // 保证解析参数只执行一次 notificationService notify_domain.Service alarmRepository repository.AlarmRepository } // NewAlarmNotificationTask 创建一个新的告警通知发送任务实例 func NewAlarmNotificationTask(ctx context.Context, taskLog *models.TaskExecutionLog, service notify_domain.Service, alarmRepository repository.AlarmRepository) plan.Task { return &AlarmNotificationTask{ ctx: ctx, taskLog: taskLog, alarmRepository: alarmRepository, notificationService: service, } } // Execute 执行告警通知发送任务 func (t *AlarmNotificationTask) Execute(ctx context.Context) error { taskCtx, logger := logs.Trace(ctx, t.ctx, "Execute") logger.Infof("开始执行告警通知发送任务, 任务ID: %d", t.taskLog.TaskID) if err := t.parseParameters(taskCtx); err != nil { return err } // 获取是否有待发送告警通知, 用于优化性能 alarmsCount, err := t.alarmRepository.CountAlarmsForNotification(taskCtx, t.alarmNotificationTaskParams.NotificationIntervals) if err != nil { logger.Errorf("任务 %v: 获取告警数量失败: %v", t.taskLog.TaskID, err) return err } if alarmsCount == 0 { logger.Debugf("没有待发送的告警通知, 跳过任务, 任务ID: %d", t.taskLog.TaskID) return nil } // 获取所有待发送的告警通知 alarms, err := t.alarmRepository.ListAlarmsForNotification(taskCtx, t.alarmNotificationTaskParams.NotificationIntervals) if err != nil { logger.Errorf("任务 %v: 获取告警列表失败: %v", t.taskLog.TaskID, err) return err } // 发送通知 for _, alarm := range alarms { // TODO 因为还没做权限管理, 所以暂时通过广播形式发给所有用户 err = t.notificationService.BroadcastAlarm(taskCtx, notify.AlarmContent{ Title: alarm.AlarmSummary, Message: alarm.AlarmDetails, Level: alarm.Level, Timestamp: time.Now(), }) if err != nil { // 非致命错误 logger.Errorf("任务 %v: 发送告警通知失败: %v", t.taskLog.TaskID, err) continue } // 能发送通知的告警要么是忽略期已过且到达触发时间, 要么是不忽略且到达触发时间, 二者都应该取消忽略并刷新最后一次发送时间 err = t.alarmRepository.UpdateAlarmNotificationStatus(taskCtx, alarm.ID, time.Now(), false, nil) if err != nil { // 非致命错误, 没有必要因为更新失败影响后续消息发送 logger.Errorf("任务 %v: 更新告警通知状态失败: %v", t.taskLog.TaskID, err) } } logger.Infof("告警通知发送任务执行完成, 任务ID: %d", t.taskLog.TaskID) return nil } // OnFailure 告警通知发送任务失败时的处理逻辑 func (t *AlarmNotificationTask) OnFailure(ctx context.Context, executeErr error) { logger := logs.TraceLogger(ctx, t.ctx, "OnFailure") logger.Errorf("告警通知发送任务执行失败, 任务ID: %d, 错误: %v", t.taskLog.TaskID, executeErr) } // ResolveDeviceIDs 从任务配置中解析并返回所有关联的设备ID列表 func (t *AlarmNotificationTask) ResolveDeviceIDs(ctx context.Context) ([]uint32, error) { // 告警通知任务与设备无关 return []uint32{}, nil } // parseParameters 解析任务参数 func (t *AlarmNotificationTask) parseParameters(ctx context.Context) error { logger := logs.TraceLogger(ctx, t.ctx, "parseParameters") var err error t.onceParse.Do(func() { if t.taskLog.Task.Parameters == nil { // 填充一个默认配置 t.alarmNotificationTaskParams = AlarmNotificationTaskParams{NotificationIntervals: make(map[models.SeverityLevel]uint32)} return } params := AlarmNotificationTaskParams{ NotificationIntervals: make(map[models.SeverityLevel]uint32), } err = t.taskLog.Task.ParseParameters(¶ms.NotificationIntervals) if err != nil { logger.Errorf("任务 %v: 解析参数失败: %v", t.taskLog.TaskID, err) err = fmt.Errorf("任务 %v: 解析参数失败: %v", t.taskLog.TaskID, err) return } t.alarmNotificationTaskParams = params }) return err }