247 lines
9.2 KiB
Go
247 lines
9.2 KiB
Go
package repository
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
|
||
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
|
||
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
|
||
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// ActiveAlarmListOptions 定义了查询活跃告警列表时的可选参数
|
||
type ActiveAlarmListOptions struct {
|
||
SourceType *models.AlarmSourceType // 按告警来源类型过滤
|
||
SourceID *uint // 按告警来源ID过滤
|
||
Level *models.SeverityLevel // 按告警严重性等级过滤
|
||
IsIgnored *bool // 按是否被忽略过滤
|
||
TriggerTime *time.Time // 告警触发时间范围 - 开始时间
|
||
EndTime *time.Time // 告警触发时间范围 - 结束时间
|
||
OrderBy string // 排序字段，例如 "trigger_time DESC"
|
||
}
|
||
|
||
// AlarmRepository 定义了对告警模型的数据库操作接口
|
||
type AlarmRepository interface {
|
||
// ListActiveAlarms 支持分页和过滤的活跃告警列表查询。
|
||
// 返回活跃告警列表、总记录数和错误。
|
||
ListActiveAlarms(ctx context.Context, opts ActiveAlarmListOptions, page, pageSize int) ([]models.ActiveAlarm, int64, error)
|
||
|
||
// <-- 下列两个方法是为了性能做出的架构妥协, 业务逻辑入侵仓库层带来的收益远大于通过业务层进行数据筛选 -->
|
||
|
||
// ListAlarmsForNotification 查询满足发送告警消息条件的活跃告警列表。
|
||
// 返回活跃告警列表和错误。
|
||
// intervalByLevel: key=SeverityLevel, value=interval_in_minutes
|
||
ListAlarmsForNotification(ctx context.Context, intervalByLevel map[models.SeverityLevel]int) ([]models.ActiveAlarm, error)
|
||
// 查询满足发送告警消息条件的记录总数
|
||
CountAlarmsForNotification(ctx context.Context, intervalByLevel map[models.SeverityLevel]int) (int64, error)
|
||
}
|
||
|
||
// gormAlarmRepository 是 AlarmRepository 的 GORM 实现。
|
||
type gormAlarmRepository struct {
|
||
ctx context.Context
|
||
db *gorm.DB
|
||
}
|
||
|
||
// NewGormAlarmRepository 创建一个新的 AlarmRepository GORM 实现实例。
|
||
func NewGormAlarmRepository(ctx context.Context, db *gorm.DB) AlarmRepository {
|
||
return &gormAlarmRepository{
|
||
ctx: ctx,
|
||
db: db,
|
||
}
|
||
}
|
||
|
||
// ListActiveAlarms 实现了分页和过滤查询活跃告警记录的功能
|
||
func (r *gormAlarmRepository) ListActiveAlarms(ctx context.Context, opts ActiveAlarmListOptions, page, pageSize int) ([]models.ActiveAlarm, int64, error) {
|
||
repoCtx := logs.AddFuncName(ctx, r.ctx, "ListActiveAlarms")
|
||
// --- 校验分页参数 ---
|
||
if page <= 0 || pageSize <= 0 {
|
||
return nil, 0, ErrInvalidPagination // 复用已定义的错误
|
||
}
|
||
|
||
var results []models.ActiveAlarm
|
||
var total int64
|
||
|
||
query := r.db.WithContext(repoCtx).Model(&models.ActiveAlarm{})
|
||
|
||
// --- 应用过滤条件 ---
|
||
if opts.SourceType != nil {
|
||
query = query.Where("source_type = ?", *opts.SourceType)
|
||
}
|
||
if opts.SourceID != nil {
|
||
query = query.Where("source_id = ?", *opts.SourceID)
|
||
}
|
||
if opts.Level != nil {
|
||
query = query.Where("level = ?", *opts.Level)
|
||
}
|
||
if opts.IsIgnored != nil {
|
||
query = query.Where("is_ignored = ?", *opts.IsIgnored)
|
||
}
|
||
if opts.TriggerTime != nil {
|
||
query = query.Where("trigger_time >= ?", *opts.TriggerTime)
|
||
}
|
||
if opts.EndTime != nil {
|
||
query = query.Where("trigger_time <= ?", *opts.EndTime)
|
||
}
|
||
|
||
// --- 计算总数 ---
|
||
if err := query.Count(&total).Error; err != nil {
|
||
return nil, 0, err
|
||
}
|
||
|
||
// --- 应用排序条件 ---
|
||
orderBy := "trigger_time DESC" // 默认按触发时间倒序
|
||
if opts.OrderBy != "" {
|
||
orderBy = opts.OrderBy
|
||
}
|
||
query = query.Order(orderBy)
|
||
|
||
// --- 分页 ---
|
||
offset := (page - 1) * pageSize
|
||
err := query.Limit(pageSize).Offset(offset).Find(&results).Error
|
||
|
||
return results, total, err
|
||
}
|
||
|
||
// CountAlarmsForNotification 查询满足发送告警消息条件的记录总数
|
||
func (r *gormAlarmRepository) CountAlarmsForNotification(ctx context.Context, intervalByLevel map[models.SeverityLevel]int) (int64, error) {
|
||
repoCtx := logs.AddFuncName(ctx, r.ctx, "CountAlarmsForNotification")
|
||
var total int64
|
||
|
||
// 1. 构造基础查询对象 (包含 Context 和 Model)
|
||
baseTx := r.db.WithContext(repoCtx).Model(&models.ActiveAlarm{})
|
||
|
||
// 2. 传递给辅助函数应用所有 WHERE 逻辑
|
||
query := r.buildNotificationBaseQuery(baseTx, intervalByLevel)
|
||
|
||
// 3. 只执行 Count
|
||
err := query.Count(&total).Error
|
||
|
||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return 0, err
|
||
}
|
||
|
||
return total, nil
|
||
}
|
||
|
||
// ListAlarmsForNotification 查询满足发送告警消息条件的活跃告警列表
|
||
func (r *gormAlarmRepository) ListAlarmsForNotification(ctx context.Context, intervalByLevel map[models.SeverityLevel]int) ([]models.ActiveAlarm, error) {
|
||
repoCtx := logs.AddFuncName(ctx, r.ctx, "ListAlarmsForNotification")
|
||
var results []models.ActiveAlarm
|
||
|
||
// 1. 构造基础查询对象 (包含 Context 和 Model)
|
||
baseTx := r.db.WithContext(repoCtx).Model(&models.ActiveAlarm{})
|
||
|
||
// 2. 传递给辅助函数应用所有 WHERE 逻辑
|
||
query := r.buildNotificationBaseQuery(baseTx, intervalByLevel)
|
||
|
||
// 3. 执行 Find (不排序，高性能)
|
||
err := query.Find(&results).Error
|
||
|
||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return nil, err
|
||
}
|
||
|
||
return results, nil
|
||
}
|
||
|
||
// buildNotificationBaseQuery 负责组合 Group A 和 Group B 的逻辑
|
||
func (r *gormAlarmRepository) buildNotificationBaseQuery(tx *gorm.DB, intervalByLevel map[models.SeverityLevel]int) *gorm.DB {
|
||
|
||
// 1. 获取所有配置的 Level 列表
|
||
configuredLevels := make([]models.SeverityLevel, 0, len(intervalByLevel))
|
||
for level := range intervalByLevel {
|
||
configuredLevels = append(configuredLevels, level)
|
||
}
|
||
|
||
// 2. 构造 Group A (只发送一次)
|
||
// Group A 是一个独立的 GORM SubQuery，用于构建 OR 关系
|
||
groupAQuery := r.buildGroupAClause(tx.Session(&gorm.Session{}), configuredLevels)
|
||
|
||
// 3. 构造 Group B (间隔发送)
|
||
// Group B 也是一个独立的 GORM SubQuery
|
||
groupBQuery := r.buildGroupBClause(tx.Session(&gorm.Session{}), intervalByLevel, configuredLevels)
|
||
|
||
// 4. 最终组合：Group A OR Group B
|
||
|
||
// 核心逻辑：利用 GORM 的 Where(SubQuery) OR Where(SubQuery) 特性。
|
||
// GORM 允许将 WHERE 或 OR 的参数写成 func(db *gorm.DB) *gorm.DB
|
||
// 这样可以确保子查询的括号被正确处理，实现 (A) OR (B) 结构。
|
||
|
||
// 注意：我们必须检查配置，因为如果 Group B 配置为空，我们不应该将其添加到 OR 关系中。
|
||
if len(configuredLevels) == 0 {
|
||
// 只有 Group A 存在（即 Level NOT IN 的条件是 1=1）
|
||
return tx.Where(groupAQuery)
|
||
}
|
||
|
||
// 存在 Group A 和 Group B，用 OR 连接
|
||
return tx.Where(groupAQuery).Or(groupBQuery)
|
||
}
|
||
|
||
// buildGroupAClause 构造 Group A 的 WHERE 语句和参数列表。
|
||
// 针对 Level 缺失配置（或所有 Level）的告警，使用“只发送一次”逻辑：LastNotifiedAt IS NULL
|
||
// 参数 configuredLevels: 用于构建 Level NOT IN (?) 子句。
|
||
func (r *gormAlarmRepository) buildGroupAClause(tx *gorm.DB, configuredLevels []models.SeverityLevel) *gorm.DB {
|
||
now := time.Now()
|
||
|
||
// A.1. 构造 Level 范围检查子句 (Level NOT IN 或 1=1)
|
||
if len(configuredLevels) > 0 {
|
||
tx = tx.Where("level NOT IN (?)", configuredLevels)
|
||
} else {
|
||
// 如果配置列表为空，则所有 Level 都符合，使用 1=1
|
||
tx = tx.Where("1 = 1")
|
||
}
|
||
|
||
// A.2. 构造 Group A 核心逻辑 (LastNotifiedAt IS NULL 且满足忽略条件)
|
||
|
||
// C_A_Ignored: 被忽略但忽略期结束 且 仅发送一次
|
||
ignoredQuery := tx.Where("is_ignored = ? AND ignored_until <= ? AND last_notified_at IS NULL", true, now)
|
||
|
||
// C_A_NotIgnored: 未被忽略 且 仅发送一次
|
||
notIgnoredQuery := tx.Where("is_ignored = ? AND last_notified_at IS NULL", false)
|
||
|
||
// A.3. 组合 Group A 核心逻辑: (C_A_Ignored OR C_A_NotIgnored)
|
||
return tx.Where(ignoredQuery).Or(notIgnoredQuery)
|
||
}
|
||
|
||
// buildGroupBClause 构造 Group B 的 WHERE 语句和参数列表。
|
||
// 针对 Level 存在配置的告警，使用“间隔发送”逻辑。
|
||
func (r *gormAlarmRepository) buildGroupBClause(tx *gorm.DB, intervalByLevel map[models.SeverityLevel]int, configuredLevels []models.SeverityLevel) *gorm.DB {
|
||
now := time.Now()
|
||
|
||
// B.1. 构造 Level IN 子句
|
||
tx = tx.Where("level IN (?)", configuredLevels)
|
||
|
||
// B.2. 构造 Level-Based 间隔检查 (OR 部分)
|
||
// 核心思想：利用 GORM 的 Or 链式调用构建 Level 间隔检查子句
|
||
|
||
// 初始化 Level 间隔检查查询 (ICC)
|
||
iccQuery := tx.Session(&gorm.Session{}) // 创建一个干净的子查询对象来构建 ICC
|
||
|
||
// 动态添加 Level 间隔检查 OR 条件
|
||
for level, minutes := range intervalByLevel {
|
||
// PostgreSQL 语法: last_notified_at + (5 * interval '1 minute') <= ?
|
||
sql := fmt.Sprintf("level = ? AND last_notified_at + (%d * interval '1 minute') <= ?", minutes)
|
||
|
||
// 每次使用 Or 叠加新的 Level 检查
|
||
iccQuery = iccQuery.Or(sql, level, now)
|
||
}
|
||
|
||
// B.3. 组合 Group B 核心逻辑: (last_notified_at IS NULL OR [ICC])
|
||
|
||
// C_B_NotIgnored: 未被忽略
|
||
notIgnoredQuery := tx.Where("is_ignored = ?", false).Where(
|
||
tx.Where("last_notified_at IS NULL").Or(iccQuery), // LastNotifiedAt IS NULL OR ICC
|
||
)
|
||
|
||
// C_B_Ignored: 被忽略但忽略期结束
|
||
ignoredQuery := tx.Where("is_ignored = ? AND ignored_until <= ?", true, now).Where(
|
||
tx.Where("last_notified_at IS NULL").Or(iccQuery), // LastNotifiedAt IS NULL OR ICC
|
||
)
|
||
|
||
// B.4. 组合 Group B 核心逻辑: (C_B_NotIgnored OR C_B_Ignored)
|
||
return tx.Where(notIgnoredQuery).Or(ignoredQuery)
|
||
}
|