271 lines
8.9 KiB
Go
271 lines
8.9 KiB
Go
package business
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log/slog"
|
||
"math/big"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
type AlertTriggerChannel struct {
|
||
ChannelKey
|
||
Account string
|
||
BusierID uint64
|
||
LastLoginUnix, CreatedUnix int64
|
||
}
|
||
|
||
type AlertTriggerInput struct {
|
||
Rules []*AlertRule
|
||
Accounts []*Account
|
||
Bindings []*Binding
|
||
Channels []*AlertTriggerChannel
|
||
}
|
||
|
||
type AlertComparison struct {
|
||
FirstUnix int64
|
||
Current, Previous string
|
||
}
|
||
|
||
type AlertTriggerRepo interface {
|
||
Input(context.Context) (*AlertTriggerInput, error)
|
||
Comparison(context.Context, int, []ChannelKey, time.Time, int) (map[ChannelKey]AlertComparison, error)
|
||
// 仓储在渠道/分类键锁及事务内读取未关闭记录，由领域决定是否刷新。
|
||
StoreHit(context.Context, *AlertRecord, []int, func(*AlertRecord) bool) (bool, error)
|
||
}
|
||
|
||
type AlertTriggerResult struct{ Rules, Written, Skipped int }
|
||
|
||
type AlertTriggerUsecase struct{ repo AlertTriggerRepo }
|
||
|
||
func NewAlertTriggerUsecase(repo AlertTriggerRepo) *AlertTriggerUsecase {
|
||
return &AlertTriggerUsecase{repo: repo}
|
||
}
|
||
|
||
func (uc *AlertTriggerUsecase) Run(ctx context.Context) error {
|
||
_, err := uc.Detect(ctx, time.Now())
|
||
return err
|
||
}
|
||
|
||
func validTriggerRule(rule *AlertRule) bool {
|
||
if rule == nil || rule.CreatorID == 0 || strings.TrimSpace(rule.Type) == "" {
|
||
return false
|
||
}
|
||
if rule.Metric == 3 {
|
||
return rule.UnloginDays != nil && *rule.UnloginDays > 0
|
||
}
|
||
if rule.Metric != 1 && rule.Metric != 2 {
|
||
return false
|
||
}
|
||
if rule.StatPeriodDays == nil || *rule.StatPeriodDays <= 0 || rule.MomDirection == nil || (*rule.MomDirection != 1 && *rule.MomDirection != 2) || rule.MomThreshold == nil {
|
||
return false
|
||
}
|
||
threshold, ok := new(big.Rat).SetString(*rule.MomThreshold)
|
||
// 历史规则检测允许零阈值，保持与旧任务一致；新增规则另有边界校验。
|
||
return ok && threshold.Sign() >= 0
|
||
}
|
||
|
||
func triggerOwners(input *AlertTriggerInput) (map[ChannelKey]*Account, map[uint]*Account) {
|
||
accounts := map[uint]*Account{}
|
||
for _, account := range input.Accounts {
|
||
if account != nil {
|
||
accounts[account.ID] = account
|
||
}
|
||
}
|
||
bindings := map[PlatformKey]uint{}
|
||
conflicts := map[PlatformKey]bool{}
|
||
for _, binding := range input.Bindings {
|
||
if binding == nil || binding.Status != "0" || binding.UserID == 0 || binding.BusierID == 0 {
|
||
continue
|
||
}
|
||
source, err := NormalizePlatform(binding.SourceDB)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
key := PlatformKey{SourceDB: source, BusierID: binding.BusierID}
|
||
if existing, ok := bindings[key]; ok && existing != binding.UserID {
|
||
conflicts[key] = true
|
||
}
|
||
bindings[key] = binding.UserID
|
||
}
|
||
owners := map[ChannelKey]*Account{}
|
||
for _, channel := range input.Channels {
|
||
if channel == nil || channel.ID == 0 {
|
||
continue
|
||
}
|
||
source, err := NormalizePlatform(channel.SourceDB)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
channel.SourceDB = source
|
||
key := PlatformKey{SourceDB: source, BusierID: channel.BusierID}
|
||
account := accounts[bindings[key]]
|
||
if conflicts[key] || account == nil || account.Level != 3 {
|
||
continue
|
||
}
|
||
path := strings.Split(account.IDPath, ",")
|
||
if len(path) < 2 {
|
||
continue
|
||
}
|
||
id, err := strconv.ParseUint(path[1], 10, 64)
|
||
owner := accounts[uint(id)]
|
||
if err != nil || owner == nil || owner.Level != 1 {
|
||
continue
|
||
}
|
||
owners[channel.ChannelKey] = owner
|
||
}
|
||
return owners, accounts
|
||
}
|
||
|
||
// 一次执行固定上海自然日，避免跨午夜时同一任务的各规则使用不同的比较窗口。
|
||
func (uc *AlertTriggerUsecase) Detect(ctx context.Context, now time.Time) (*AlertTriggerResult, error) {
|
||
result := &AlertTriggerResult{}
|
||
input, err := uc.repo.Input(ctx)
|
||
if err != nil {
|
||
return result, err
|
||
}
|
||
if input == nil {
|
||
return result, ErrStorage
|
||
}
|
||
zone := time.FixedZone("Asia/Shanghai", 8*3600)
|
||
now = now.In(zone)
|
||
today := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, zone)
|
||
owners, accounts := triggerOwners(input)
|
||
type snapshotKey struct {
|
||
creator uint
|
||
metric, days int
|
||
}
|
||
snapshots := map[snapshotKey]map[ChannelKey]AlertComparison{}
|
||
var failures []string
|
||
for _, rule := range input.Rules {
|
||
if err := ctx.Err(); err != nil {
|
||
return result, err
|
||
}
|
||
if !validTriggerRule(rule) {
|
||
result.Skipped++
|
||
continue
|
||
}
|
||
creator := accounts[rule.CreatorID]
|
||
if creator == nil || (creator.AuthorityID != 888 && creator.Level != 1) {
|
||
result.Skipped++
|
||
continue
|
||
}
|
||
channels := []*AlertTriggerChannel{}
|
||
keys := []ChannelKey{}
|
||
for _, channel := range input.Channels {
|
||
if channel == nil {
|
||
continue
|
||
}
|
||
owner := owners[channel.ChannelKey]
|
||
if owner != nil && (creator.AuthorityID == 888 || owner.ID == creator.ID) {
|
||
channels = append(channels, channel)
|
||
keys = append(keys, channel.ChannelKey)
|
||
}
|
||
}
|
||
if len(channels) == 0 {
|
||
continue
|
||
}
|
||
result.Rules++
|
||
var snapshot map[ChannelKey]AlertComparison
|
||
if rule.Metric != 3 {
|
||
key := snapshotKey{creator.ID, rule.Metric, *rule.StatPeriodDays}
|
||
var exists bool
|
||
snapshot, exists = snapshots[key]
|
||
if !exists {
|
||
snapshot, err = uc.repo.Comparison(ctx, rule.Metric, keys, today, key.days)
|
||
if err != nil {
|
||
failures = append(failures, fmt.Sprintf("规则 %d 读取指标失败: %v", rule.ID, err))
|
||
continue
|
||
}
|
||
snapshots[key] = snapshot
|
||
}
|
||
}
|
||
for _, channel := range channels {
|
||
if err := ctx.Err(); err != nil {
|
||
return result, err
|
||
}
|
||
content, hit, err := triggerContent(rule, channel, snapshot[channel.ChannelKey], today)
|
||
if err != nil {
|
||
failures = append(failures, fmt.Sprintf("规则 %d 渠道 %s/%d: %v", rule.ID, channel.SourceDB, channel.ID, err))
|
||
continue
|
||
}
|
||
if !hit {
|
||
continue
|
||
}
|
||
owner := owners[channel.ChannelKey]
|
||
record := &AlertRecord{RuleID: rule.ID, RuleName: rule.Name, RuleType: strings.TrimSpace(rule.Type), MonitorMetric: &rule.Metric,
|
||
PlatformCode: channel.SourceDB, PlatformName: PlatformName(channel.SourceDB), TargetPromoteID: channel.ID, TargetPromoteName: channel.Account,
|
||
BusinessID: owner.ID, BusinessName: owner.Username, AlertContent: content, Status: AlertUnprocessed, CreatedAt: now, UpdatedAt: now}
|
||
written, err := uc.repo.StoreHit(ctx, record, []int{AlertUnprocessed, AlertProcessing, AlertPendingAudit, AlertAuditRejected}, func(current *AlertRecord) bool { return current == nil || current.Status == AlertUnprocessed })
|
||
if err != nil {
|
||
failures = append(failures, fmt.Sprintf("规则 %d 渠道 %s/%d 写入失败: %v", rule.ID, channel.SourceDB, channel.ID, err))
|
||
continue
|
||
}
|
||
if written {
|
||
result.Written++
|
||
}
|
||
}
|
||
}
|
||
if len(failures) > 0 {
|
||
for _, failure := range failures {
|
||
slog.ErrorContext(ctx, "预警检测失败", "detail", failure)
|
||
}
|
||
return result, fmt.Errorf("预警规则检测存在 %d 项失败：%s", len(failures), strings.Join(failures[:min(5, len(failures))], "；"))
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
func triggerContent(rule *AlertRule, channel *AlertTriggerChannel, sample AlertComparison, today time.Time) (string, bool, error) {
|
||
if rule.Metric == 3 {
|
||
stamp := channel.LastLoginUnix
|
||
if stamp <= 0 {
|
||
stamp = channel.CreatedUnix
|
||
}
|
||
if stamp <= 0 {
|
||
return "", false, nil
|
||
}
|
||
day := time.Unix(stamp, 0).In(today.Location())
|
||
day = time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, today.Location())
|
||
days := int(today.Sub(day) / (24 * time.Hour))
|
||
if days < *rule.UnloginDays {
|
||
return "", false, nil
|
||
}
|
||
return fmt.Sprintf("连续%d天未登录", days), true, nil
|
||
}
|
||
days := *rule.StatPeriodDays
|
||
if sample.FirstUnix <= 0 || time.Unix(sample.FirstUnix, 0).In(today.Location()).Format("2006-01-02") > today.AddDate(0, 0, -2*days).Format("2006-01-02") {
|
||
return "", false, nil
|
||
}
|
||
previous, err := portraitNumber(sample.Previous)
|
||
if err != nil {
|
||
return "", false, err
|
||
}
|
||
if previous.Sign() <= 0 {
|
||
return "", false, nil
|
||
}
|
||
current, err := portraitNumber(sample.Current)
|
||
if err != nil {
|
||
return "", false, err
|
||
}
|
||
rate := new(big.Rat).Quo(new(big.Rat).Sub(current, previous), previous)
|
||
// 先对比值保留四位再转百分比，与旧 BigDecimal HALF_UP 的阈值边界一致。
|
||
rate, _ = new(big.Rat).SetString(portraitDecimal(rate, 4))
|
||
rate.Mul(rate, big.NewRat(100, 1))
|
||
threshold, _ := new(big.Rat).SetString(*rule.MomThreshold)
|
||
if (*rule.MomDirection == 1 && rate.Cmp(threshold) < 0) || (*rule.MomDirection == 2 && rate.Cmp(new(big.Rat).Neg(threshold)) > 0) {
|
||
return "", false, nil
|
||
}
|
||
direction := "上升"
|
||
if rate.Sign() < 0 {
|
||
direction = "下降"
|
||
}
|
||
percent := portraitDecimal(new(big.Rat).Abs(rate), 2)
|
||
if rule.Metric == 1 {
|
||
return fmt.Sprintf("截至昨日%d个完整自然日注册%s人，较此前%d个完整自然日（%s人）环比%s%s%%", days, portraitDecimal(current, 0), days, portraitDecimal(previous, 0), direction, percent), true, nil
|
||
}
|
||
yuan := big.NewRat(10000, 1)
|
||
return fmt.Sprintf("截至昨日%d个完整自然日充值¥%s万，较此前%d个完整自然日（¥%s万）环比%s%s%%", days, portraitDecimal(new(big.Rat).Quo(current, yuan), 2), days, portraitDecimal(new(big.Rat).Quo(previous, yuan), 2), direction, percent), true, nil
|
||
}
|