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 }