pig-farm-controller/internal/domain/task/alarm_notification_task.go

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(&params.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
}