aiops/backend/controller/alert_engine.go

package controller

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"ezops/model"
	ezredis "ezops/pkg/redis"

	"gorm.io/gorm"
)

// AlertEngine 告警引擎 — 定时扫描 Agent 上报数据并触发告警
// 支持分布式锁：多副本部署时仅 Leader 执行检查，避免重复告警
type AlertEngine struct {
	db         *gorm.DB
	cache      ezredis.Cache // 可选 Redis 缓存（分布式锁）
	instanceID string        // 当前实例唯一标识
}

const (
	alertEngineLockKey = "aiops:alert_engine:lock"
	alertEngineLockTTL = 60 * time.Second // 锁 TTL：2x tick 间隔，保证处理时间内不释放
)

// NewAlertEngine 创建告警引擎实例
// cache 参数用于分布式锁（多副本部署）。nil 时为单实例模式，不竞争锁。
func NewAlertEngine(db *gorm.DB, cache ezredis.Cache) *AlertEngine {
	// 生成唯一实例 ID（随机 hex，防止多副本 PID/hostname 碰撞）
	idBytes := make([]byte, 8)
	_, _ = rand.Read(idBytes)
	instanceID := fmt.Sprintf("ae-%s", hex.EncodeToString(idBytes))
	return &AlertEngine{
		db:         db,
		cache:      cache,
		instanceID: instanceID,
	}
}

// Check 每 30 秒调用一次。多副本模式下通过分布式锁保证仅 Leader 执行。
// 无 cache（单实例/缓存初始化失败）→ 直接执行
// 有 cache（Docker/K8s 多副本）→ SetNX 原子抢锁
// Redis 故障 → fail-open（继续执行，避免告警遗漏）
func (e *AlertEngine) Check() {
	if !e.tryAcquireLock() {
		log.Printf("[AlertEngine] 非 Leader 副本（%s），跳过本轮检查", e.instanceID)
		return
	}
	defer e.releaseLock()

	e.checkResourceAlerts()
	e.checkOfflineAlerts()
	e.checkPortAlerts()
	e.autoResolveAlerts()
}

// tryAcquireLock 尝试获取分布式锁，返回是否获得 Leader 角色
func (e *AlertEngine) tryAcquireLock() bool {
	if e.cache == nil {
		return true // 单实例模式
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	acquired, err := e.cache.SetNX(ctx, alertEngineLockKey, e.instanceID, alertEngineLockTTL)
	if err != nil {
		log.Printf("[AlertEngine] 获取分布式锁失败（%s），fail-open 继续: %v", e.instanceID, err)
		return true
	}
	return acquired
}

// releaseLock 释放分布式锁（仅当当前实例持有锁时才释放，防止误删其他副本的锁）
func (e *AlertEngine) releaseLock() {
	if e.cache == nil {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	val, err := e.cache.Get(ctx, alertEngineLockKey)
	if err != nil {
		return
	}
	if val == e.instanceID {
		_ = e.cache.Del(ctx, alertEngineLockKey)
	}
}

// GetEngineStatus 返回告警引擎运行状态（用于运维监控）
func (e *AlertEngine) GetEngineStatus() map[string]interface{} {
	isLeader := true
	lockHolder := ""
	if e.cache != nil {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if val, err := e.cache.Get(ctx, alertEngineLockKey); err == nil {
			lockHolder = val
			isLeader = lockHolder == "" || lockHolder == e.instanceID
		}
	}
	return map[string]interface{}{
		"is_leader":     isLeader,
		"instance_id":   e.instanceID,
		"lock_holder":   lockHolder,
		"distributed":   e.cache != nil,
		"tick_interval": 30,
		"lock_ttl_sec":  int(alertEngineLockTTL / time.Second),
	}
}

// 检查 CPU/内存/磁盘/进程/TCP连接/端口 阈值
func (e *AlertEngine) checkResourceAlerts() {
	var rules []model.AgentRule
	e.db.Where("enabled = ? AND metric IN ?", true, []string{"cpu", "mem", "disk", "service_process", "tcp_time_wait", "tcp_close_wait", "fd"}).Find(&rules)

	for _, rule := range rules {
		// 获取目标主机最新上报
		var report model.AgentReport
		query := e.db.Model(&model.AgentReport{})
		if rule.Hostname != "" {
			query = query.Where("hostname = ?", rule.Hostname)
		}
		query.Order("id DESC").First(&report)
		if report.ID == 0 {
			continue
		}

		var value float64
		switch rule.Metric {
		case "cpu":
			value = report.CPUUsage
		case "mem":
			value = report.MemUsage
		case "disk":
			var disks []model.DiskMetric
			json.Unmarshal([]byte(report.DiskInfo), &disks)
			for _, d := range disks {
				if d.UsagePercent > value {
					value = d.UsagePercent
				}
			}
		case "tcp_time_wait":
			var netStat model.NetworkStatModel
			json.Unmarshal([]byte(report.NetInfo), &netStat)
			value = float64(netStat.TCPTimeWait)
		case "tcp_close_wait":
			var netStat model.NetworkStatModel
			json.Unmarshal([]byte(report.NetInfo), &netStat)
			value = float64(netStat.TCPCloseWait)
		case "service_process":
			// 进程不存在告警 — 检查 process_top_info 中是否包含指定进程名
			var processes []model.ProcessMetric
			json.Unmarshal([]byte(report.ProcessTopInfo), &processes)
			found := false
			for _, p := range processes {
				if p.Name == rule.Hostname || rule.Hostname == "" {
					found = true
					break
				}
			}
			if found {
				continue // 进程存在，不触发
			}
			value = 0 // 进程不存在时触发
		case "fd":
			var fdStat model.FDMetric
			json.Unmarshal([]byte(report.FDInfo), &fdStat)
			value = fdStat.UsagePct
		}

		// 检查是否超阈值
		if !e.compareOperator(rule.Operator, value, rule.Threshold) {
			continue
		}

		// 避免重复告警：如果最近 10 分钟已有同规则未恢复告警，不重复
		var existing model.AgentAlert
		tenMinAgo := time.Now().Add(-10 * time.Minute)
		e.db.Where("rule_name = ? AND hostname = ? AND resolved = ? AND created_at > ?",
			rule.Name, report.Hostname, false, tenMinAgo).First(&existing)
		if existing.ID > 0 {
			continue
		}

		// 检查静默规则 — 匹配则不创建告警、不发送通知
		if e.isSilenced(rule.Name, report.Hostname, rule.ID) {
			continue
		}

		// 创建告警
		alert := model.AgentAlert{
			RuleName:  rule.Name,
			Hostname:  report.Hostname,
			Metric:    rule.Metric,
			Severity:  "warning",
			Message:   fmt.Sprintf("%s: %s %.1f%% %s %.1f%%", report.Hostname, rule.Metric, value, rule.Operator, rule.Threshold),
			Value:     value,
			Threshold: rule.Threshold,
		}
		if value > rule.Threshold*1.5 {
			alert.Severity = "critical"
		}
		e.db.Create(&alert)

		// 发送告警通知
		e.notifyAlert(&rule, report.Hostname, alert)

		// 更新规则最后触发时间
		now := time.Now()
		e.db.Model(&rule).Update("last_triggered", now)
	}
}

// 检查主机离线（超过 3 分钟无上报）— 内置规则，无需手动创建。
// 以 servers 表为中心：遍历所有服务器，检查其 IP 上是否有 Agent 在最近 3 分钟内上报。
// 如果同一 IP 下有多个 Agent hostname，只要有一个在活跃上报，该服务器就算在线。
func (e *AlertEngine) checkOfflineAlerts() {
	offlineThreshold := 3 * time.Minute

	// 获取所有已登记的主机（去重 IP）
	var servers []model.Server
	e.db.Select("ip, hostname").Group("ip").Find(&servers)

	// 收集所有 agent_reports 中活跃上报的主机名（最近 3 分钟内有上报）
	threeMinAgo := time.Now().Add(-offlineThreshold)

	var activeHostnames []string
	e.db.Model(&model.AgentReport{}).
		Where("report_time >= ?", threeMinAgo).
		Distinct("hostname").
		Pluck("hostname", &activeHostnames)

	// 对于每个已登记的服务器，检查是否有活跃 Agent
	for _, s := range servers {
		if s.IP == "" {
			continue
		}

		// 检查该 IP 对应的服务器是否有活跃 Agent 上报
		hasActiveAgent := false
		for _, h := range activeHostnames {
			// 通过 hostname 精确匹配或 servers 表 IP 匹配
			if h == s.Hostname {
				hasActiveAgent = true
				break
			}
			// 检查该 hostname 的最新上报——其源 IP 是否匹配此 server IP
			// （通过 servers 表中查找 hostname 对应的 IP）
			var matchedServer model.Server
			e.db.Where("hostname = ?", h).First(&matchedServer)
			if matchedServer.ID > 0 && matchedServer.IP == s.IP {
				hasActiveAgent = true
				break
			}
		}

		if hasActiveAgent {
			continue // 有活跃 Agent，在线
		}

		// 该服务器离线：检查是否已有未恢复的离线告警
		var existing model.AgentAlert
		e.db.Where("hostname = ? AND metric = ? AND resolved = ?",
			s.Hostname, "offline", false).First(&existing)

		if existing.ID == 0 {
			// 获取最后一次上报时间计算离线时长
			var lastReport model.AgentReport
			e.db.Where("hostname = ?", s.Hostname).Order("id DESC").First(&lastReport)
			offlineMinutes := 3.0
			var lastReportTime time.Time
			if lastReport.ID > 0 {
				offlineMinutes = time.Since(lastReport.ReportTime).Minutes()
				lastReportTime = lastReport.ReportTime
			}

			// 检查静默规则
			if e.isSilenced("主机离线", s.Hostname, 0) {
				continue
			}

			alert := model.AgentAlert{
				RuleName:  "主机离线",
				Hostname:  s.Hostname,
				Metric:    "offline",
				Severity:  "critical",
				Message:   fmt.Sprintf("主机 %s (%s) 已超过 3 分钟无数据上报，可能已离线", s.Hostname, s.IP),
				Value:     offlineMinutes,
				Threshold: 3,
			}
			e.db.Create(&alert)

			// 发送离线告警通知
			e.notifyAlertRule("主机离线", s.Hostname, "offline", "critical",
				alert.Message, offlineMinutes, 3)

			// 记录离线上报时间用于恢复通知
			if !lastReportTime.IsZero() {
				e.db.Model(&alert).Update("created_at", time.Now())
			}
		}
	}
}

// 检查端口消失（之前有的监听端口现在没了）
func (e *AlertEngine) checkPortAlerts() {
	var rules []model.AgentRule
	e.db.Where("enabled = ? AND metric = ?", true, "port_down").Find(&rules)

	for _, rule := range rules {
		if rule.Port == 0 {
			continue
		}

		// 获取目标主机最新上报
		var report model.AgentReport
		query := e.db.Model(&model.AgentReport{})
		if rule.Hostname != "" {
			query = query.Where("hostname = ?", rule.Hostname)
		}
		query.Order("id DESC").First(&report)
		if report.ID == 0 {
			continue
		}

		// 检查端口是否在最新上报中
		var ports []model.PortMetric
		json.Unmarshal([]byte(report.PortInfo), &ports)

		portUp := false
		for _, p := range ports {
			if p.Port == rule.Port {
				portUp = true
				break
			}
		}

		if portUp {
			continue // 端口正常
		}

		// 端口消失了，检查是否已有未恢复告警
		var existing model.AgentAlert
		e.db.Where("rule_name = ? AND hostname = ? AND resolved = ?",
			rule.Name, report.Hostname, false).First(&existing)
		if existing.ID > 0 {
			continue
		}

		// 检查静默规则
		if e.isSilenced(rule.Name, report.Hostname, rule.ID) {
			continue
		}

		alert := model.AgentAlert{
			RuleName:  rule.Name,
			Hostname:  report.Hostname,
			Metric:    "port_down",
			Severity:  "critical",
			Message:   fmt.Sprintf("主机 %s 端口 %d 服务疑似下线", report.Hostname, rule.Port),
			Value:     float64(rule.Port),
			Threshold: float64(rule.Port),
		}
		e.db.Create(&alert)

		// 发送端口Down告警通知
		e.notifyAlertRule(rule.Name, report.Hostname, "port_down", "critical",
			alert.Message, float64(rule.Port), float64(rule.Port))

		now := time.Now()
		e.db.Model(&rule).Update("last_triggered", now)
	}
}

// 自动恢复告警（指标恢复正常）
func (e *AlertEngine) autoResolveAlerts() {
	var alerts []model.AgentAlert
	e.db.Where("resolved = ?", false).Find(&alerts)

	for _, alert := range alerts {
		resolved := false

		switch alert.Metric {
		case "offline":
			// 主机重新上报了
			var report model.AgentReport
			e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report)
			if report.ID > 0 && time.Since(report.ReportTime) < 3*time.Minute {
				resolved = true
			}
		case "cpu", "mem", "disk", "tcp_time_wait", "tcp_close_wait", "fd":
			// 获取对应规则，检查是否恢复正常
			var rule model.AgentRule
			e.db.Where("name = ?", alert.RuleName).First(&rule)
			if rule.ID == 0 {
				continue
			}
			var report model.AgentReport
			if rule.Hostname != "" {
				e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report)
			} else {
				e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report)
			}
			if report.ID == 0 {
				continue
			}
			var value float64
			switch alert.Metric {
			case "cpu":
				value = report.CPUUsage
			case "mem":
				value = report.MemUsage
			case "disk":
				var disks []model.DiskMetric
				json.Unmarshal([]byte(report.DiskInfo), &disks)
				for _, d := range disks {
					if d.UsagePercent > value {
						value = d.UsagePercent
					}
				}
			case "tcp_time_wait":
				var netStat model.NetworkStatModel
				json.Unmarshal([]byte(report.NetInfo), &netStat)
				value = float64(netStat.TCPTimeWait)
			case "tcp_close_wait":
				var netStat model.NetworkStatModel
				json.Unmarshal([]byte(report.NetInfo), &netStat)
				value = float64(netStat.TCPCloseWait)
			case "fd":
				var fdStat model.FDMetric
				json.Unmarshal([]byte(report.FDInfo), &fdStat)
				value = fdStat.UsagePct
			}
			if !e.compareOperator(rule.Operator, value, rule.Threshold) {
				resolved = true
			}
		case "port_down":
			var rule model.AgentRule
			e.db.Where("name = ?", alert.RuleName).First(&rule)
			if rule.ID > 0 {
				var report model.AgentReport
				e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report)
				var ports []model.PortMetric
				json.Unmarshal([]byte(report.PortInfo), &ports)
				for _, p := range ports {
					if p.Port == rule.Port {
						resolved = true
						break
					}
				}
			}
		}

		if resolved {
			now := time.Now()
			e.db.Model(&alert).Updates(map[string]interface{}{
				"resolved":    true,
				"resolved_at": now,
			})

			// 发送恢复通知
			e.notifyRecovery(alert.RuleName, alert.Hostname, alert.Metric, alert.Message)
		}
	}
}

func (e *AlertEngine) compareOperator(op string, val, threshold float64) bool {
	switch op {
	case ">":
		return val > threshold
	case ">=":
		return val >= threshold
	case "<":
		return val < threshold
	case "<=":
		return val <= threshold
	case "==":
		return val == threshold
	default:
		return val > threshold
	}
}

// notifyAlert 发送告警通知（通过 AgentRule 对应的 notify_channels）
func (e *AlertEngine) notifyAlert(rule *model.AgentRule, hostname string, alert model.AgentAlert) {
	e.notifyAlertRule(rule.Name, hostname, alert.Metric, alert.Severity,
		alert.Message, alert.Value, alert.Threshold)
}

// notifyAlertRule 按规则发送告警通知到所有启用的飞书渠道
func (e *AlertEngine) notifyAlertRule(ruleName, hostname, metric, severity, message string, value, threshold float64) {
	var channels []model.NotifyChannel
	e.db.Where("enabled = ? AND type = ?", true, "feishu").Find(&channels)

	for _, ch := range channels {
		title := fmt.Sprintf("[%s] %s — %s (%s)", severity, ruleName, hostname, metric)
		content := fmt.Sprintf(
			"**规则名称：** %s\\n**目标主机：** %s\\n**指标类型：** %s\\n**告警级别：** %s\\n**当前值：** %.2f\\n**阈值：** %.2f\\n**详细信息：** %s\\n**触发时间：** %s",
			ruleName, hostname, metric, severity, value, threshold, message, time.Now().Format("2006-01-02 15:04:05"),
		)
		if err := sendFeishuAlert(ch.Config, title, content, severity); err != nil {
			fmt.Printf("[AlertEngine] 飞书通知失败 (渠道:%s): %v\n", ch.Name, err)
		}
	}
}

// notifyRecovery 发送恢复通知到所有启用的飞书渠道
func (e *AlertEngine) notifyRecovery(ruleName, hostname, metric, message string) {
	var channels []model.NotifyChannel
	e.db.Where("enabled = ? AND type = ?", true, "feishu").Find(&channels)

	for _, ch := range channels {
		title := fmt.Sprintf("%s — 已恢复正常", ruleName)
		content := fmt.Sprintf("**告警已恢复**\\\\n\\\\n主机：%s\\\\n指标：%s\\\\n详情：%s\\\\n恢复时间：%s",
			hostname, metric, message, time.Now().Format("2006-01-02 15:04:05"))
		if err := sendFeishuRecovery(ch.Config, title, content); err != nil {
			fmt.Printf("[AlertEngine] 飞书恢复通知失败 (渠道:%s): %v\n", ch.Name, err)
		}
	}
}

// isSilenced 检查给定告警是否被静默规则匹配
// 匹配逻辑：status=1 且 当前时间在静默窗口内
// 匹配维度：rule_name（精确匹配）、rule_id（ID 匹配）、hostname（通过 target_ids 中的 server IP 间接匹配）
// silence_type：onetime=一次性窗口，daily=每天指定时段，weekly=每周指定星期+时段
func (e *AlertEngine) isSilenced(ruleName, hostname string, ruleID uint) bool {
	var silences []model.AlertSilence
	e.db.Where("status = ?", 1).Find(&silences)

	now := time.Now()
	for _, s := range silences {
		// 检查规则名匹配 — 通过 rule_ids JSON 数组
		if s.RuleIDs != "" && s.RuleIDs != "null" {
			var ruleIDs []uint
			json.Unmarshal([]byte(s.RuleIDs), &ruleIDs)
			matched := false
			for _, rid := range ruleIDs {
				if rid == ruleID {
					matched = true
					break
				}
			}
			if !matched {
				continue // 该静默规则不匹配此 rule_id
			}
		}

		// 检查目标主机匹配 — 通过 target_ids（server IDs）间接匹配
		if s.TargetIDs != "" && s.TargetIDs != "null" {
			var targetIDs []uint
			json.Unmarshal([]byte(s.TargetIDs), &targetIDs)
			if len(targetIDs) > 0 {
				matched := false
				for _, tid := range targetIDs {
					var server model.Server
					e.db.Select("hostname").First(&server, tid)
					if server.Hostname == hostname {
						matched = true
						break
					}
				}
				if !matched {
					continue
				}
			}
		}

		// 检查时间窗口
		switch s.SilenceType {
		case model.AlertSilenceOneTime:
			// 一次性：start_time <= now <= end_time
			if now.After(s.StartTime) && now.Before(s.EndTime) {
				return true
			}
		case model.AlertSilenceDaily:
			// 每天：检查当前时分是否在窗口内
			nowMinutes := now.Hour()*60 + now.Minute()
			startMinutes := s.StartTime.Hour()*60 + s.StartTime.Minute()
			endMinutes := s.EndTime.Hour()*60 + s.EndTime.Minute()
			if startMinutes <= endMinutes {
				if nowMinutes >= startMinutes && nowMinutes <= endMinutes {
					return true
				}
			} else {
				// 跨天（如 23:00-06:00）
				if nowMinutes >= startMinutes || nowMinutes <= endMinutes {
					return true
				}
			}
		case model.AlertSilenceWeekly:
			// 每周：检查当前星期是否在 weekdays 列表中，且时分在窗口内
			if s.Weekdays != "" && s.Weekdays != "null" {
				var weekdays []int
				json.Unmarshal([]byte(s.Weekdays), &weekdays)
				weekdayMatched := false
				todayWeekday := int(now.Weekday())
				for _, wd := range weekdays {
					if wd == todayWeekday {
						weekdayMatched = true
						break
					}
				}
				if !weekdayMatched {
					continue
				}
			}
			nowMinutes := now.Hour()*60 + now.Minute()
			startMinutes := s.StartTime.Hour()*60 + s.StartTime.Minute()
			endMinutes := s.EndTime.Hour()*60 + s.EndTime.Minute()
			if startMinutes <= endMinutes {
				if nowMinutes >= startMinutes && nowMinutes <= endMinutes {
					return true
				}
			} else {
				if nowMinutes >= startMinutes || nowMinutes <= endMinutes {
					return true
				}
			}
		default:
			// 默认按 onetime 处理
			if now.After(s.StartTime) && now.Before(s.EndTime) {
				return true
			}
		}
	}
	return false
}