feat: implement V2 alert center and scoped alert governance
This commit is contained in:
Yvan
@@ -0,0 +1,290 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
"kra/internal/biz/business"
|
||||
)
|
||||
|
||||
type alertGapPO struct {
|
||||
OwnerID uint `gorm:"primaryKey;autoIncrement:false"`
|
||||
Enabled bool
|
||||
Days int
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
func (alertGapPO) TableName() string { return "sys_alert_gap" }
|
||||
|
||||
type alertExemptionPO struct {
|
||||
ID uint64 `gorm:"primaryKey"`
|
||||
SourceDB string `gorm:"size:50;index:idx_alert_exempt_channel,priority:1"`
|
||||
PromoteID uint64 `gorm:"index:idx_alert_exempt_channel,priority:2"`
|
||||
ChannelName string `gorm:"size:100"`
|
||||
OwnerID uint `gorm:"index:idx_alert_exempt_owner,priority:1"`
|
||||
AddedBy uint
|
||||
AddedName string `gorm:"size:100"`
|
||||
Reason string `gorm:"size:500"`
|
||||
AddedAt time.Time
|
||||
RemovedBy uint
|
||||
RemovedName string `gorm:"size:100"`
|
||||
RemoveReason string `gorm:"size:100"`
|
||||
RemovedAt *time.Time `gorm:"index:idx_alert_exempt_owner,priority:2;index:idx_alert_exempt_channel,priority:3"`
|
||||
}
|
||||
|
||||
func (alertExemptionPO) TableName() string { return "sys_alert_exemption" }
|
||||
|
||||
// 所有渠道豁免变更和预警生成共用渠道锁，防止已加入豁免后仍插入新预警。
|
||||
type alertChannelLockPO struct {
|
||||
SourceDB string `gorm:"primaryKey;size:50"`
|
||||
PromoteID uint64 `gorm:"primaryKey;autoIncrement:false"`
|
||||
}
|
||||
|
||||
func (alertChannelLockPO) TableName() string { return "sys_alert_channel_lock" }
|
||||
func lockAlertChannel(tx *gorm.DB, source string, id uint64) error {
|
||||
key := alertChannelLockPO{SourceDB: source, PromoteID: id}
|
||||
if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&key).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("source_db=? AND promote_id=?", source, id).Take(&key).Error
|
||||
}
|
||||
func toBizAlertExemption(p alertExemptionPO) *business.AlertExemption {
|
||||
return &business.AlertExemption{ID: p.ID, ChannelKey: business.ChannelKey{SourceDB: p.SourceDB, ID: p.PromoteID}, ChannelName: p.ChannelName, OwnerID: p.OwnerID, AddedBy: p.AddedBy, AddedName: p.AddedName, Reason: p.Reason, AddedAt: p.AddedAt, RemovedBy: p.RemovedBy, RemovedName: p.RemovedName, RemoveReason: p.RemoveReason, RemovedAt: p.RemovedAt}
|
||||
}
|
||||
|
||||
type alertGovernanceRepo struct{ data Provider }
|
||||
|
||||
func NewAlertGovernanceRepo(data Provider) business.AlertGovernanceRepo {
|
||||
return &alertGovernanceRepo{data}
|
||||
}
|
||||
|
||||
func (r *alertGovernanceRepo) Candidates(ctx context.Context, scope business.UserScope, keyword string, limit int) ([]*business.AlertChannelOption, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
bindings := db.Table("sys_user_bind_platform b").Select("1").Joins("JOIN sys_users u ON u.id=b.user_id AND u.deleted_at IS NULL").Where("b.status='0' AND b.source_db=p.source_db AND b.platform_busier_id=p.busier_id")
|
||||
bindings = applyUserScope(bindings, scope, "u.")
|
||||
q := db.Table("tab_promote p").Where("p.promote_level=1 AND p.source_db IN ?", []string{"lx", "hw", "yq"}).Where("EXISTS (?)", bindings)
|
||||
if keyword != "" {
|
||||
q = q.Where("p.account LIKE ?", "%"+keyword+"%")
|
||||
}
|
||||
var rows []struct {
|
||||
ID uint64
|
||||
SourceDB, Name string
|
||||
}
|
||||
if err := q.Select("p.id,p.source_db,p.account AS name").Order("p.source_db,p.account,p.id").Limit(limit).Scan(&rows).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
result := make([]*business.AlertChannelOption, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
result = append(result, &business.AlertChannelOption{ChannelKey: business.ChannelKey{SourceDB: row.SourceDB, ID: row.ID}, Name: row.Name})
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
func (r *alertGovernanceRepo) Gap(ctx context.Context, id uint) (*business.AlertGap, error) {
|
||||
var p alertGapPO
|
||||
err := r.data.DB().WithContext(ctx).Where("owner_id=?", id).Take(&p).Error
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return &business.AlertGap{OwnerID: id, Days: 1}, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
return &business.AlertGap{OwnerID: p.OwnerID, Enabled: p.Enabled, Days: p.Days}, nil
|
||||
}
|
||||
func (r *alertGovernanceRepo) SaveGap(ctx context.Context, gap *business.AlertGap) error {
|
||||
p := alertGapPO{OwnerID: gap.OwnerID, Enabled: gap.Enabled, Days: gap.Days, UpdatedAt: time.Now()}
|
||||
return storageError(r.data.DB().WithContext(ctx).Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "owner_id"}}, DoUpdates: clause.AssignmentColumns([]string{"enabled", "days", "updated_at"})}).Create(&p).Error)
|
||||
}
|
||||
func (r *alertGovernanceRepo) Exemptions(ctx context.Context, scope business.LevelOneScope, filter business.AlertExemptionFilter, opts ...business.ListOption) ([]*business.AlertExemption, int64, error) {
|
||||
q := applyOwnerScope(r.data.DB().WithContext(ctx).Model(&alertExemptionPO{}), scope, "owner_id")
|
||||
if filter.Name != "" {
|
||||
q = q.Where("channel_name LIKE ?", "%"+filter.Name+"%")
|
||||
}
|
||||
if filter.Active != nil {
|
||||
if *filter.Active {
|
||||
q = q.Where("removed_at IS NULL")
|
||||
} else {
|
||||
q = q.Where("removed_at IS NOT NULL")
|
||||
}
|
||||
}
|
||||
var total int64
|
||||
if err := q.Count(&total).Error; err != nil {
|
||||
return nil, 0, business.ErrStorage
|
||||
}
|
||||
o := business.NewListOptions(opts...)
|
||||
var pos []alertExemptionPO
|
||||
if err := q.Order("id DESC").Offset(o.Offset).Limit(o.Limit).Find(&pos).Error; err != nil {
|
||||
return nil, 0, business.ErrStorage
|
||||
}
|
||||
rows := make([]*business.AlertExemption, 0, len(pos))
|
||||
for _, p := range pos {
|
||||
rows = append(rows, toBizAlertExemption(p))
|
||||
}
|
||||
return rows, total, nil
|
||||
}
|
||||
func alertExemptionTarget(db *gorm.DB, scope business.UserScope, key business.ChannelKey) (*business.AlertExemptionTarget, error) {
|
||||
q := db.Table("tab_promote p").Joins("JOIN sys_user_bind_platform b ON b.source_db=p.source_db AND b.platform_busier_id=p.busier_id AND b.status='0'").Joins("JOIN sys_users u ON u.id=b.user_id AND u.deleted_at IS NULL").Where("p.source_db=? AND p.id=? AND p.promote_level=1", key.SourceDB, key.ID)
|
||||
allowed := applyUserScope(db.Table("sys_users viewer").Select("1").Where("viewer.id=u.id"), scope, "viewer.")
|
||||
var pos []struct {
|
||||
Name string
|
||||
UserID uint
|
||||
Level int
|
||||
IDPath string
|
||||
Allowed bool
|
||||
}
|
||||
if err := q.Clauses(clause.Locking{Strength: "UPDATE"}).Select("p.account AS name,u.id AS user_id,u.level,u.id_path,EXISTS (?) AS allowed", allowed).Distinct().Limit(2).Scan(&pos).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
if len(pos) != 1 || !pos[0].Allowed {
|
||||
return nil, business.ErrAlertRecordScope
|
||||
}
|
||||
p := pos[0]
|
||||
return &business.AlertExemptionTarget{Name: p.Name, Account: &business.Account{ID: p.UserID, Level: p.Level, IDPath: p.IDPath}}, nil
|
||||
}
|
||||
func (r *alertGovernanceRepo) AddExemption(ctx context.Context, scope business.UserScope, key business.ChannelKey, build func(*business.AlertExemptionTarget) (*business.AlertExemption, error)) error {
|
||||
return storageError(r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := lockAlertChannel(tx, key.SourceDB, key.ID); err != nil {
|
||||
return err
|
||||
}
|
||||
target, err := alertExemptionTarget(tx, scope, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
value, err := build(target)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var count int64
|
||||
if err := tx.Model(&alertExemptionPO{}).Where("source_db=? AND promote_id=? AND owner_id=? AND removed_at IS NULL", value.SourceDB, value.ChannelKey.ID, value.OwnerID).Count(&count).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if count > 0 {
|
||||
return business.ErrAlertExemptExists
|
||||
}
|
||||
p := alertExemptionPO{SourceDB: value.SourceDB, PromoteID: value.ChannelKey.ID, ChannelName: value.ChannelName, OwnerID: value.OwnerID, AddedBy: value.AddedBy, AddedName: value.AddedName, Reason: value.Reason, AddedAt: value.AddedAt}
|
||||
return tx.Create(&p).Error
|
||||
}))
|
||||
}
|
||||
func (r *alertGovernanceRepo) RemoveExemption(ctx context.Context, scope business.LevelOneScope, id uint64, operator uint, name, reason string, now time.Time) error {
|
||||
return storageError(r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
var p alertExemptionPO
|
||||
if err := applyOwnerScope(tx.Model(&alertExemptionPO{}), scope, "owner_id").Where("id=? AND removed_at IS NULL", id).Take(&p).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return business.ErrAlertExemptNotFound
|
||||
}
|
||||
return err
|
||||
}
|
||||
if err := lockAlertChannel(tx, p.SourceDB, p.PromoteID); err != nil {
|
||||
return err
|
||||
}
|
||||
result := tx.Model(&alertExemptionPO{}).Where("id=? AND removed_at IS NULL", id).Updates(map[string]any{"removed_at": now, "removed_by": operator, "removed_name": name, "remove_reason": reason})
|
||||
if result.Error != nil {
|
||||
return result.Error
|
||||
}
|
||||
if result.RowsAffected != 1 {
|
||||
return business.ErrAlertExemptNotFound
|
||||
}
|
||||
return nil
|
||||
}))
|
||||
}
|
||||
|
||||
// 按渠道等值、成功支付及时间范围索引读取，避免为每条预警全表统计历史流水。
|
||||
func alertLostQuery(db *gorm.DB, sourceColumn, idColumn string, window business.AlertLossWindow) *gorm.DB {
|
||||
pay := db.Table("tab_spend ls").Select("COALESCE(SUM(ls.pay_amount),0)").Where("ls.source_db=lp.source_db AND ls.promote_id=lp.id AND ls.pay_status=1 AND ls.pay_time>=? AND ls.pay_time<?", window.PayBegin, window.PayEnd)
|
||||
return db.Table("tab_promote lp").Select("1").Where("lp.source_db="+sourceColumn+" AND lp.id="+idColumn).
|
||||
Where("(COALESCE(NULLIF(lp.last_login_time,0),lp.create_time)>0 AND COALESCE(NULLIF(lp.last_login_time,0),lp.create_time)<?) OR (?) < CAST(? AS DECIMAL(20,2))", window.LoginBefore, pay, window.MinPay)
|
||||
}
|
||||
func (r *alertGovernanceRepo) LostExemptions(ctx context.Context, window business.AlertLossWindow) ([]*business.AlertExemption, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
var pos []alertExemptionPO
|
||||
// 没有在名单渠道时无需读取平台明细。
|
||||
if err := db.Select("id").Where("removed_at IS NULL").Limit(1).Find(&pos).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
if len(pos) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
available, err := businessTablesAvailable(db, "tab_promote", "tab_spend")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !available {
|
||||
return nil, business.ErrAlertChannelsUnavailable
|
||||
}
|
||||
result := []*business.AlertExemption{}
|
||||
var cursor uint64
|
||||
for {
|
||||
var batch []alertExemptionPO
|
||||
if err := db.Where("removed_at IS NULL AND id>?", cursor).Order("id").Limit(500).Find(&batch).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
if len(batch) == 0 {
|
||||
break
|
||||
}
|
||||
keys := make([]business.ChannelKey, 0, len(batch))
|
||||
for _, p := range batch {
|
||||
keys = append(keys, business.ChannelKey{SourceDB: p.SourceDB, ID: p.PromoteID})
|
||||
}
|
||||
states, err := loadAlertLoss(db, keys, window)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, p := range batch {
|
||||
if lost := states[business.ChannelKey{SourceDB: p.SourceDB, ID: p.PromoteID}]; lost != nil && *lost {
|
||||
result = append(result, toBizAlertExemption(p))
|
||||
}
|
||||
}
|
||||
cursor = batch[len(batch)-1].ID
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// 同一页的重复渠道只计算一次；按固定平台、渠道集合和时间范围扫描覆盖索引。
|
||||
func loadAlertLoss(db *gorm.DB, keys []business.ChannelKey, window business.AlertLossWindow) (map[business.ChannelKey]*bool, error) {
|
||||
bySource := map[string][]uint64{}
|
||||
seen := map[business.ChannelKey]bool{}
|
||||
for _, key := range keys {
|
||||
if !seen[key] {
|
||||
bySource[key.SourceDB] = append(bySource[key.SourceDB], key.ID)
|
||||
seen[key] = true
|
||||
}
|
||||
}
|
||||
result := map[business.ChannelKey]*bool{}
|
||||
for source, ids := range bySource {
|
||||
for start := 0; start < len(ids); start += 500 {
|
||||
batch := ids[start:min(start+500, len(ids))]
|
||||
var channels []struct {
|
||||
ID uint64
|
||||
LastLoginTime, CreateTime int64
|
||||
}
|
||||
if err := db.Table("tab_promote").Select("id,last_login_time,create_time").Where("source_db=? AND id IN ?", source, batch).Scan(&channels).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
var amounts []struct {
|
||||
PromoteID uint64
|
||||
RecentAmount string
|
||||
}
|
||||
if err := db.Table("tab_spend").Select("promote_id,COALESCE(SUM(pay_amount),0) AS recent_amount").Where("source_db=? AND promote_id IN ? AND pay_status=1 AND pay_time>=? AND pay_time<?", source, batch, window.PayBegin, window.PayEnd).Group("promote_id").Scan(&amounts).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
payments := map[uint64]string{}
|
||||
for _, amount := range amounts {
|
||||
payments[amount.PromoteID] = amount.RecentAmount
|
||||
}
|
||||
for _, channel := range channels {
|
||||
amount := payments[channel.ID]
|
||||
if amount == "" {
|
||||
amount = "0"
|
||||
}
|
||||
lost, err := window.IsLost(channel.LastLoginTime, channel.CreateTime, amount)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result[business.ChannelKey{SourceDB: source, ID: channel.ID}] = &lost
|
||||
}
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -0,0 +1,195 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"gorm.io/gorm"
|
||||
"kra/internal/biz/business"
|
||||
"math/big"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
func alertTree(db *gorm.DB, roots []business.ChannelKey) *gorm.DB {
|
||||
return db.Table("tab_promote root").Joins("JOIN tab_promote p ON p.source_db=root.source_db AND (p.id=root.id OR p.parent_id=root.id OR p.top_promote_id=root.id)").
|
||||
Where("(root.source_db,root.id) IN ?", channelPairs(roots))
|
||||
}
|
||||
func alertMetricSpec(db *gorm.DB, metric int) (table, stamp, condition, amount string) {
|
||||
if metric == 2 {
|
||||
return "tab_spend", "m.pay_time", "m.pay_status=1", "m.pay_amount"
|
||||
}
|
||||
stamp = "CAST(m.register_time AS INTEGER)"
|
||||
if db.Dialector.Name() == "mysql" {
|
||||
stamp = "m.sw_register_time_unix"
|
||||
}
|
||||
return "tab_user", stamp, "(m.puid=0 OR m.puid IS NULL)", "1"
|
||||
}
|
||||
|
||||
func (r *alertTriggerRepo) Comparison(ctx context.Context, metric int, channels []business.ChannelKey, today time.Time, days int) (map[business.ChannelKey]business.AlertComparison, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
if metric != 1 && metric != 2 {
|
||||
return nil, business.ErrAlertRuleInput
|
||||
}
|
||||
table, stamp, condition, amount := alertMetricSpec(db, metric)
|
||||
available, err := businessTablesAvailable(db, table, "tab_promote")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !available {
|
||||
return nil, business.ErrReportBuildUnavailable
|
||||
}
|
||||
result := map[business.ChannelKey]business.AlertComparison{}
|
||||
current, previous := today.AddDate(0, 0, -days).Unix(), today.AddDate(0, 0, -2*days).Unix()
|
||||
// 先取小规模渠道映射，再按固定平台及渠道 ID 批量聚合；避免 MySQL 5.7
|
||||
// 将动态 JOIN 的时间范围退化为 ref 扫描，读取每个渠道的全部历史明细。
|
||||
owners := map[business.ChannelKey][]business.ChannelKey{}
|
||||
bySource := map[string][]uint64{}
|
||||
for start := 0; start < len(channels); start += 100 {
|
||||
var pairs []struct {
|
||||
SourceDB string
|
||||
RootID, PromoteID uint64
|
||||
}
|
||||
if err := alertTree(db, channels[start:min(start+100, len(channels))]).Select("root.source_db,root.id AS root_id,p.id AS promote_id").Scan(&pairs).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, p := range pairs {
|
||||
leaf := business.ChannelKey{SourceDB: p.SourceDB, ID: p.PromoteID}
|
||||
root := business.ChannelKey{SourceDB: p.SourceDB, ID: p.RootID}
|
||||
if len(owners[leaf]) == 0 {
|
||||
bySource[p.SourceDB] = append(bySource[p.SourceDB], p.PromoteID)
|
||||
}
|
||||
owners[leaf] = append(owners[leaf], root)
|
||||
result[root] = business.AlertComparison{Current: "0", Previous: "0"}
|
||||
}
|
||||
}
|
||||
for source, ids := range bySource {
|
||||
for start := 0; start < len(ids); start += 300 {
|
||||
batch := ids[start:min(start+300, len(ids))]
|
||||
firstQueries := []string{}
|
||||
args := []any{}
|
||||
for _, id := range batch {
|
||||
// MIN 的三个等值前缀命中覆盖索引，MySQL 可直接读取首个索引项。
|
||||
firstQueries = append(firstQueries, "SELECT ? AS promote_id,(SELECT MIN("+stamp+") FROM "+table+" m WHERE m.source_db=? AND m.promote_id=? AND "+condition+" AND "+stamp+">0) AS first_unix")
|
||||
args = append(args, id, source, id)
|
||||
}
|
||||
var first []struct {
|
||||
PromoteID uint64
|
||||
FirstUnix *int64
|
||||
}
|
||||
if err := db.Raw(strings.Join(firstQueries, " UNION ALL "), args...).Scan(&first).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, row := range first {
|
||||
if row.FirstUnix == nil {
|
||||
continue
|
||||
}
|
||||
for _, root := range owners[business.ChannelKey{SourceDB: source, ID: row.PromoteID}] {
|
||||
v := result[root]
|
||||
if v.FirstUnix == 0 || *row.FirstUnix < v.FirstUnix {
|
||||
v.FirstUnix = *row.FirstUnix
|
||||
}
|
||||
result[root] = v
|
||||
}
|
||||
}
|
||||
var rows []struct {
|
||||
PromoteID uint64
|
||||
CurrentValue, PreviousValue string
|
||||
}
|
||||
selection := "m.promote_id,COALESCE(SUM(CASE WHEN " + stamp + ">=? THEN " + amount + " ELSE 0 END),0) AS current_value,COALESCE(SUM(CASE WHEN " + stamp + "<? THEN " + amount + " ELSE 0 END),0) AS previous_value"
|
||||
q := db.Table(table+" m").Where("m.source_db=? AND m.promote_id IN ?", source, batch).Where(condition).Where(stamp+">=? AND "+stamp+"<?", previous, today.Unix())
|
||||
if err := q.Select(selection, current, current).Group("m.promote_id").Scan(&rows).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
scale := 0
|
||||
if metric == 2 {
|
||||
scale = 2
|
||||
}
|
||||
for _, row := range rows {
|
||||
for _, root := range owners[business.ChannelKey{SourceDB: source, ID: row.PromoteID}] {
|
||||
v := result[root]
|
||||
var err error
|
||||
v.Current, err = addAlertMetric(v.Current, row.CurrentValue, scale)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
v.Previous, err = addAlertMetric(v.Previous, row.PreviousValue, scale)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result[root] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func addAlertMetric(a, b string, scale int) (string, error) {
|
||||
left, ok := new(big.Rat).SetString(a)
|
||||
if !ok {
|
||||
return "", business.ErrStorage
|
||||
}
|
||||
right, ok := new(big.Rat).SetString(b)
|
||||
if !ok {
|
||||
return "", business.ErrStorage
|
||||
}
|
||||
return left.Add(left, right).FloatString(scale), nil
|
||||
}
|
||||
|
||||
func (r *alertTriggerRepo) Snapshots(ctx context.Context, channels []business.ChannelKey, now time.Time, dedup bool) (map[business.ChannelKey]business.AlertSnapshot, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
available, err := businessTablesAvailable(db, "tab_user", "tab_spend", "tab_promote")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !available {
|
||||
return nil, business.ErrReportBuildUnavailable
|
||||
}
|
||||
result := map[business.ChannelKey]business.AlertSnapshot{}
|
||||
for _, key := range channels {
|
||||
result[key] = business.AlertSnapshot{Pay: "0.00"}
|
||||
}
|
||||
for start := 0; start < len(channels); start += 100 {
|
||||
batch := channels[start:min(start+100, len(channels))]
|
||||
var players []struct {
|
||||
SourceDB string
|
||||
PromoteID uint64
|
||||
Players int64
|
||||
}
|
||||
_, stamp, condition, _ := alertMetricSpec(db, 1)
|
||||
identity := "CAST(m.id AS TEXT)"
|
||||
if db.Dialector.Name() == "mysql" {
|
||||
identity = "CAST(m.id AS CHAR)"
|
||||
}
|
||||
if dedup {
|
||||
identity = "CASE WHEN TRIM(COALESCE(m.idcard,''))<>'' THEN 'card:' || TRIM(m.idcard) ELSE 'user:' || " + identity + " END"
|
||||
if db.Dialector.Name() == "mysql" {
|
||||
identity = "CASE WHEN TRIM(COALESCE(m.idcard,''))<>'' THEN CONCAT('card:',TRIM(m.idcard)) ELSE CONCAT('user:',m.id) END"
|
||||
}
|
||||
}
|
||||
q := alertTree(db, batch).Joins("JOIN tab_user m ON m.source_db=p.source_db AND m.promote_id=p.id").Where(condition).Where(stamp+"<=?", now.Unix())
|
||||
if err := q.Select("root.source_db,root.id AS promote_id,COUNT(DISTINCT " + identity + ") AS players").Group("root.source_db,root.id").Scan(&players).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, row := range players {
|
||||
key := business.ChannelKey{SourceDB: row.SourceDB, ID: row.PromoteID}
|
||||
v := result[key]
|
||||
v.Players = row.Players
|
||||
result[key] = v
|
||||
}
|
||||
var payments []struct {
|
||||
SourceDB string
|
||||
PromoteID uint64
|
||||
Amount string
|
||||
}
|
||||
if err := alertTree(db, batch).Joins("JOIN tab_spend m ON m.source_db=p.source_db AND m.promote_id=p.id").Where("m.pay_status=1 AND m.pay_time<=?", now.Unix()).Select("root.source_db,root.id AS promote_id,COALESCE(SUM(m.pay_amount),0) AS amount").Group("root.source_db,root.id").Scan(&payments).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, row := range payments {
|
||||
key := business.ChannelKey{SourceDB: row.SourceDB, ID: row.PromoteID}
|
||||
v := result[key]
|
||||
v.Pay = row.Amount
|
||||
result[key] = v
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -3,6 +3,9 @@ package business
|
||||
import "time"
|
||||
|
||||
type alertRulePO struct {
|
||||
ConditionMode int `gorm:"not null;default:1"`
|
||||
AmountDirection int `gorm:"not null;default:1"`
|
||||
AmountThreshold *string `gorm:"type:decimal(14,2)"`
|
||||
ID uint64 `gorm:"primaryKey"`
|
||||
RuleName string `gorm:"size:100;not null"`
|
||||
RuleType string `gorm:"size:50;not null"`
|
||||
@@ -22,8 +25,12 @@ type alertRulePO struct {
|
||||
func (alertRulePO) TableName() string { return "sys_alert_rule" }
|
||||
|
||||
type alertRecordPO struct {
|
||||
ID uint64 `gorm:"primaryKey"`
|
||||
PlatformCode, PlatformName string `gorm:"size:50"`
|
||||
IndependentRule bool `gorm:"not null;default:0"`
|
||||
TotalPlayers *int64
|
||||
TotalPay *string `gorm:"type:decimal(20,2)"`
|
||||
WarningColor *string `gorm:"size:20"`
|
||||
ID uint64 `gorm:"primaryKey"`
|
||||
PlatformCode, PlatformName string `gorm:"size:50"`
|
||||
RuleID uint64
|
||||
RuleName string `gorm:"size:100"`
|
||||
RuleType string `gorm:"size:50"`
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/logger"
|
||||
"kra/internal/biz/business"
|
||||
)
|
||||
|
||||
type alertSQLCapture struct {
|
||||
logger.Interface
|
||||
queries []string
|
||||
elapsed []time.Duration
|
||||
}
|
||||
|
||||
func (l *alertSQLCapture) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) {
|
||||
sql, _ := fc()
|
||||
if strings.Contains(sql, "AS current_value") || strings.Contains(sql, "AS first_unix") || strings.Contains(sql, "AS recent_amount") {
|
||||
l.queries = append(l.queries, sql)
|
||||
l.elapsed = append(l.elapsed, time.Since(begin))
|
||||
}
|
||||
l.Interface.Trace(ctx, begin, fc, err)
|
||||
}
|
||||
|
||||
func TestMySQLAlertV2MigrationAndRangePlans(t *testing.T) {
|
||||
db := openRehearsalMySQL(t)
|
||||
now := prepareAlertV2(t, db)
|
||||
// 模拟尚未升级的真实旧表，覆盖普通列合并迁移路径。
|
||||
for _, sql := range []string{
|
||||
"ALTER TABLE sys_alert_record DROP INDEX uk_alert_record_v2_open",
|
||||
"ALTER TABLE sys_alert_record DROP COLUMN v2_open_key",
|
||||
"ALTER TABLE sys_alert_record DROP COLUMN independent_rule, DROP COLUMN total_players, DROP COLUMN total_pay, DROP COLUMN warning_color",
|
||||
"ALTER TABLE sys_alert_rule DROP COLUMN condition_mode, DROP COLUMN amount_direction, DROP COLUMN amount_threshold",
|
||||
} {
|
||||
if err := db.Exec(sql).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := db.Exec("ALTER TABLE sys_alert_record ADD COLUMN unclosed_flag TINYINT GENERATED ALWAYS AS (CASE WHEN status=5 THEN NULL ELSE 1 END) STORED, ADD UNIQUE INDEX uk_alert_record_unclosed(platform_code,target_promote_id,rule_type,unclosed_flag)").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := migrateAlertV2(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := migrateAlertV2(db); err != nil {
|
||||
t.Fatalf("重复迁移: %v", err)
|
||||
}
|
||||
if db.Migrator().HasIndex("sys_alert_record", "uk_alert_record_unclosed") {
|
||||
t.Fatal("旧唯一键阻挡已忽略或独立绝对值规则")
|
||||
}
|
||||
if err := db.Exec("ALTER TABLE sys_alert_record DROP INDEX uk_alert_record_v2_open, ADD UNIQUE INDEX uk_alert_record_v2_open(platform_code,target_promote_id,rule_type,v2_open_key)").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := migrateAlertV2UniqueKey(db); err != nil {
|
||||
t.Fatalf("已有 V2 索引升级失败: %v", err)
|
||||
}
|
||||
rows := make([]map[string]any, 10000)
|
||||
for i := range rows {
|
||||
rows[i] = map[string]any{"id": 100 + i, "source_db": "lx", "promote_id": 1, "pay_status": 1, "pay_time": now.AddDate(-1, 0, 0).Unix(), "pay_amount": "1.00"}
|
||||
}
|
||||
if err := db.Table("tab_spend").CreateInBatches(rows, 500).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Exec("ANALYZE TABLE tab_spend,tab_promote").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
capture := &alertSQLCapture{Interface: logger.Default.LogMode(logger.Silent)}
|
||||
r := NewAlertTriggerRepo(testData{db.Session(&gorm.Session{Logger: capture})})
|
||||
key := business.ChannelKey{SourceDB: "lx", ID: 1}
|
||||
values, err := r.Comparison(context.Background(), 2, []business.ChannelKey{key}, now, 1)
|
||||
if err != nil || values[key].Current != "3000.00" || values[key].Previous != "100.00" {
|
||||
t.Fatalf("MySQL 周期计算: %+v %v", values, err)
|
||||
}
|
||||
if _, err := loadAlertLoss(db.Session(&gorm.Session{Logger: capture}), []business.ChannelKey{key, key}, business.AlertLossAt(now)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i, sql := range capture.queries {
|
||||
var plan []struct {
|
||||
Table string
|
||||
Type string
|
||||
Key string
|
||||
Rows int64
|
||||
Extra string
|
||||
}
|
||||
if err := db.Raw("EXPLAIN " + sql).Scan(&plan).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Logf("query %d elapsed=%v plan=%+v", i, capture.elapsed[i], plan)
|
||||
for _, step := range plan {
|
||||
if (strings.Contains(sql, "AS current_value") || strings.Contains(sql, "AS recent_amount")) && (step.Table == "m" || step.Table == "tab_spend") && (step.Type != "range" || step.Rows > 100) {
|
||||
t.Fatalf("预警周期统计必须按时间范围读取，不能扫描整个渠道历史: %+v", plan)
|
||||
}
|
||||
}
|
||||
}
|
||||
hit := &business.AlertRecord{PlatformCode: "lx", TargetPromoteID: 1, RuleID: 1, RuleType: "充值", IndependentRule: true, GapDays: 7, Status: 1, CreatedAt: now, UpdatedAt: now}
|
||||
decide := func(r *business.AlertRecord) bool { return r == nil || r.Status == 1 }
|
||||
for _, id := range []uint64{1, 2} {
|
||||
hit.RuleID = id
|
||||
if ok, err := r.StoreHit(context.Background(), hit, []int{1, 2, 3, 4}, decide); err != nil || !ok {
|
||||
t.Fatalf("独立规则被唯一键挡住: %v %v", ok, err)
|
||||
}
|
||||
}
|
||||
if err := db.Model(&alertRecordPO{}).Where("rule_id=1").Update("status", 6).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
hit.RuleID = 1
|
||||
hit.CreatedAt = now.AddDate(0, 0, 7)
|
||||
if ok, err := r.StoreHit(context.Background(), hit, []int{1, 2, 3, 4}, decide); err != nil || !ok {
|
||||
t.Fatalf("已忽略仍占用唯一键: %v %v", ok, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMySQLAlertMigrationKeepsLegacyConstraintOnUnexpectedIndex(t *testing.T) {
|
||||
db := openRehearsalMySQL(t)
|
||||
for _, sql := range []string{
|
||||
"ALTER TABLE sys_alert_record DROP INDEX uk_alert_record_v2_open, ADD INDEX uk_alert_record_v2_open(platform_code), ADD UNIQUE INDEX uk_alert_record_daily(platform_code,target_promote_id,rule_type,alert_date)",
|
||||
} {
|
||||
if err := db.Exec(sql).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := migrateAlertV2UniqueKey(db); err == nil {
|
||||
t.Fatal("意外的现有索引定义不应被当成迁移成功")
|
||||
}
|
||||
if !db.Migrator().HasIndex("sys_alert_record", "uk_alert_record_daily") {
|
||||
t.Fatal("验证失败前不能移除旧约束")
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sort"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
@@ -16,8 +17,17 @@ type alertRecordRepo struct{ data Provider }
|
||||
|
||||
func NewAlertRecordRepo(data Provider) business.AlertRecordRepo { return &alertRecordRepo{data: data} }
|
||||
|
||||
func (r *alertRecordRepo) Platforms(ctx context.Context, scope business.AlertRecordScope) ([]string, error) {
|
||||
rows := []string{}
|
||||
if scope.Owner.Empty {
|
||||
return rows, nil
|
||||
}
|
||||
err := alertBindingQuery(r.data.DB().WithContext(ctx), scope.Viewer).Distinct().Order("b.source_db").Pluck("b.source_db", &rows).Error
|
||||
return rows, storageError(err)
|
||||
}
|
||||
|
||||
func toBizAlertRecord(po alertRecordPO) *business.AlertRecord {
|
||||
return &business.AlertRecord{ID: po.ID, RuleID: po.RuleID, PlatformCode: po.PlatformCode, PlatformName: po.PlatformName, RuleName: po.RuleName, RuleType: po.RuleType, MonitorMetric: po.MonitorMetric, TargetPromoteID: po.TargetPromoteID, TargetPromoteName: po.TargetPromoteName, BusinessID: po.BusinessID, BusinessName: po.BusinessName, AlertContent: po.AlertContent, Status: po.Status, CreatedAt: po.CreatedAt, UpdatedAt: po.UpdatedAt}
|
||||
return &business.AlertRecord{ID: po.ID, TotalPlayers: po.TotalPlayers, TotalPay: po.TotalPay, WarningColor: po.WarningColor, RuleID: po.RuleID, PlatformCode: po.PlatformCode, PlatformName: po.PlatformName, RuleName: po.RuleName, RuleType: po.RuleType, MonitorMetric: po.MonitorMetric, TargetPromoteID: po.TargetPromoteID, TargetPromoteName: po.TargetPromoteName, BusinessID: po.BusinessID, BusinessName: po.BusinessName, AlertContent: po.AlertContent, Status: po.Status, CreatedAt: po.CreatedAt, UpdatedAt: po.UpdatedAt}
|
||||
}
|
||||
|
||||
func alertBindingQuery(db *gorm.DB, scope business.UserScope) *gorm.DB {
|
||||
@@ -50,11 +60,11 @@ func prepareAlertScope(db *gorm.DB, scope business.AlertRecordScope) (empty, ava
|
||||
if s.Empty {
|
||||
return true, true, nil
|
||||
}
|
||||
var count int64
|
||||
if err := alertBindingQuery(db, s).Count(&count).Error; err != nil {
|
||||
var bindings []struct{ UserID uint }
|
||||
if err := alertBindingQuery(db, s).Select("b.user_id").Limit(1).Scan(&bindings).Error; err != nil {
|
||||
return false, false, business.ErrStorage
|
||||
}
|
||||
if count == 0 {
|
||||
if len(bindings) == 0 {
|
||||
return true, true, nil
|
||||
}
|
||||
}
|
||||
@@ -97,6 +107,14 @@ func (r *alertRecordRepo) List(ctx context.Context, scope business.AlertRecordSc
|
||||
return page, nil
|
||||
}
|
||||
}
|
||||
lossAvailable, err := businessTablesAvailable(db, "tab_promote", "tab_spend")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if filter.Lost != nil && !lossAvailable {
|
||||
return nil, business.ErrAlertChannelsUnavailable
|
||||
}
|
||||
lossWindow := business.AlertLossAt(time.Now())
|
||||
base := func() *gorm.DB {
|
||||
q := alertScopedQuery(db, scope)
|
||||
if len(filter.PlatformCodes) > 0 {
|
||||
@@ -110,24 +128,49 @@ func (r *alertRecordRepo) List(ctx context.Context, scope business.AlertRecordSc
|
||||
}
|
||||
return q
|
||||
}
|
||||
q := base()
|
||||
for _, f := range []struct{ column, value string }{{"target_promote_name", filter.TargetPromoteName}, {"rule_name", filter.RuleName}, {"business_name", filter.BusinessName}} {
|
||||
if f.value != "" {
|
||||
q = q.Where("r."+f.column+" LIKE ?", "%"+f.value+"%")
|
||||
filtered := func() *gorm.DB {
|
||||
q := base()
|
||||
for _, f := range []struct{ column, value string }{{"target_promote_name", filter.TargetPromoteName}, {"rule_name", filter.RuleName}, {"business_name", filter.BusinessName}} {
|
||||
if f.value != "" {
|
||||
q = q.Where("r."+f.column+" LIKE ?", "%"+f.value+"%")
|
||||
}
|
||||
}
|
||||
if filter.RuleType != "" {
|
||||
q = q.Where("r.rule_type = ?", filter.RuleType)
|
||||
}
|
||||
if filter.MonitorMetric != nil {
|
||||
q = q.Where("r.monitor_metric = ?", *filter.MonitorMetric)
|
||||
}
|
||||
if filter.AlertContent != "" {
|
||||
q = q.Where("r.alert_content LIKE ?", "%"+filter.AlertContent+"%")
|
||||
}
|
||||
if filter.Lost != nil {
|
||||
q = q.Where("EXISTS (?)", db.Table("tab_promote known").Select("1").Where("known.source_db=r.platform_code AND known.id=r.target_promote_id"))
|
||||
q = q.Where("EXISTS (?) = ?", alertLostQuery(db, "r.platform_code", "r.target_promote_id", lossWindow), *filter.Lost)
|
||||
}
|
||||
return q
|
||||
}
|
||||
if filter.RuleType != "" {
|
||||
q = q.Where("r.rule_type = ?", filter.RuleType)
|
||||
}
|
||||
if filter.MonitorMetric != nil {
|
||||
q = q.Where("r.monitor_metric = ?", *filter.MonitorMetric)
|
||||
}
|
||||
q := filtered()
|
||||
if filter.Status != nil {
|
||||
q = q.Where("r.status = ?", *filter.Status)
|
||||
}
|
||||
if err := q.Count(&page.Total).Error; err != nil {
|
||||
|
||||
// 六态卡片继承业务筛选，忽略状态自身，使切换状态时其他计数仍可见。
|
||||
var counts []struct {
|
||||
Status int
|
||||
Count int64
|
||||
}
|
||||
if err := filtered().Select("r.status,COUNT(*) AS count").Group("r.status").Scan(&counts).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, c := range counts {
|
||||
page.StatusCount[c.Status] = c.Count
|
||||
}
|
||||
for status, count := range page.StatusCount {
|
||||
if filter.Status == nil || status == *filter.Status {
|
||||
page.Total += count
|
||||
}
|
||||
}
|
||||
o := business.NewListOptions(options...)
|
||||
var pos []alertRecordPO
|
||||
if err := q.Select("r.*").Order("r.created_at DESC,r.id DESC").Offset(o.Offset).Limit(o.Limit).Scan(&pos).Error; err != nil {
|
||||
@@ -136,20 +179,68 @@ func (r *alertRecordRepo) List(ctx context.Context, scope business.AlertRecordSc
|
||||
for _, po := range pos {
|
||||
page.Rows = append(page.Rows, toBizAlertRecord(po))
|
||||
}
|
||||
// 顶部汇总忽略名称、类型、指标与状态筛选，只保留数据范围、平台和指定渠道。
|
||||
var counts []struct {
|
||||
Status int
|
||||
Count int64
|
||||
}
|
||||
if err := base().Select("r.status,COUNT(*) AS count").Group("r.status").Scan(&counts).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, c := range counts {
|
||||
page.StatusCount[c.Status] = c.Count
|
||||
if lossAvailable && len(pos) > 0 {
|
||||
keys := make([]business.ChannelKey, 0, len(pos))
|
||||
for _, p := range pos {
|
||||
keys = append(keys, business.ChannelKey{SourceDB: p.PlatformCode, ID: p.TargetPromoteID})
|
||||
}
|
||||
states, err := loadAlertLoss(db, keys, lossWindow)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, row := range page.Rows {
|
||||
row.Lost = states[business.ChannelKey{SourceDB: row.PlatformCode, ID: row.TargetPromoteID}]
|
||||
}
|
||||
}
|
||||
|
||||
return page, nil
|
||||
}
|
||||
|
||||
func (r *alertRecordRepo) Ignore(ctx context.Context, scope business.AlertRecordScope, ids []uint64, log *business.AlertLog) error {
|
||||
ids = append([]uint64(nil), ids...)
|
||||
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
|
||||
return storageError(r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
empty, available, err := prepareAlertScope(tx, scope)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if empty {
|
||||
return business.ErrAlertRecordScope
|
||||
}
|
||||
if !available {
|
||||
return business.ErrAlertChannelsUnavailable
|
||||
}
|
||||
var rows []alertRecordPO
|
||||
if err := alertScopedQuery(tx, scope).Select("r.*").Where("r.id IN ?", ids).Order("r.id").Clauses(clause.Locking{Strength: "UPDATE"}).Scan(&rows).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
unique := map[uint64]bool{}
|
||||
for _, id := range ids {
|
||||
unique[id] = true
|
||||
}
|
||||
if len(rows) != len(unique) {
|
||||
return business.ErrAlertRecordScope
|
||||
}
|
||||
now := time.Now()
|
||||
logs := []alertLogPO{}
|
||||
changed := []uint64{}
|
||||
for _, row := range rows {
|
||||
if row.Status == business.AlertIgnored {
|
||||
continue
|
||||
}
|
||||
changed = append(changed, row.ID)
|
||||
logs = append(logs, alertLogPO{AlertID: row.ID, OperatorID: log.OperatorID, OperatorName: log.OperatorName, LogType: log.LogType, ActionDetail: log.ActionDetail, CreatedAt: now})
|
||||
}
|
||||
if len(changed) == 0 {
|
||||
return nil
|
||||
}
|
||||
if err := tx.Model(&alertRecordPO{}).Where("id IN ?", changed).Updates(map[string]any{"status": business.AlertIgnored, "updated_at": now}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.CreateInBatches(logs, 100).Error
|
||||
}))
|
||||
}
|
||||
|
||||
func findScopedAlert(db *gorm.DB, scope business.AlertRecordScope, id uint64, lock bool) (*alertRecordPO, error) {
|
||||
empty, available, err := prepareAlertScope(db, scope)
|
||||
if err != nil {
|
||||
|
||||
@@ -58,8 +58,8 @@ func TestAlertRecordsScopeAndSummaryUseDifferentFilters(t *testing.T) {
|
||||
scope := alertTestScope(7)
|
||||
status := 1
|
||||
page, err := repo.List(ctx, scope, business.AlertRecordFilter{RuleName: "注册", Status: &status})
|
||||
if err != nil || page.Total != 1 || len(page.StatusCount) != 3 || page.StatusCount[3] != 1 || page.StatusCount[2] != 1 {
|
||||
t.Fatalf("汇总错误继承名称或状态筛选: %+v %v", page, err)
|
||||
if err != nil || page.Total != 1 || len(page.StatusCount) != 1 || page.StatusCount[1] != 1 {
|
||||
t.Fatalf("汇总应继承名称筛选并忽略状态自身: %+v %v", page, err)
|
||||
}
|
||||
selected := alertTestScope(8).Viewer
|
||||
scope.Selected = &selected
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"gorm.io/gorm"
|
||||
"kra/internal/biz/business"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestAlertWritesRecheckCurrentChannelOwnerAndRejectConflictingBindings(t *testing.T) {
|
||||
exerciseAlertOwnerRecheck(t, openTestDB(t))
|
||||
}
|
||||
func TestMySQLAlertWritesRecheckCurrentChannelOwner(t *testing.T) {
|
||||
exerciseAlertOwnerRecheck(t, openRehearsalMySQL(t))
|
||||
}
|
||||
func exerciseAlertOwnerRecheck(t *testing.T, db *gorm.DB) {
|
||||
t.Helper()
|
||||
now := prepareAlertV2(t, db)
|
||||
if err := db.AutoMigrate(&accountPO{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, sql := range []string{
|
||||
"INSERT INTO sys_users(id,username,level,id_path) VALUES(7,'原商务',1,'0'),(8,'新商务',1,'0')",
|
||||
"INSERT INTO sys_user_bind_platform(user_id,source_db,platform_busier_id,status) VALUES(7,'lx',7,'0')",
|
||||
} {
|
||||
if err := db.Exec(sql).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
ctx := context.Background()
|
||||
scope := business.UserScope{SelfID: 7, IncludeSelf: true, DescendantPath: "0,7"}
|
||||
hit := &business.AlertRecord{OwnerScope: &scope, PlatformCode: "lx", TargetPromoteID: 1, BusinessID: 7, RuleID: 1, RuleType: "充值", Status: 1, AlertContent: "换绑前", CreatedAt: now, UpdatedAt: now}
|
||||
trigger := NewAlertTriggerRepo(testData{db})
|
||||
write := func(want bool) {
|
||||
t.Helper()
|
||||
got, err := trigger.StoreHit(ctx, hit, []int{1, 2, 3, 4}, func(*business.AlertRecord) bool { return true })
|
||||
if err != nil || got != want {
|
||||
t.Fatalf("written=%v want=%v err=%v", got, want, err)
|
||||
}
|
||||
}
|
||||
write(true)
|
||||
oldExemption := &business.AlertExemption{ChannelKey: business.ChannelKey{SourceDB: "lx", ID: 1}, OwnerID: 7, AddedBy: 7, AddedName: "原商务", Reason: "原商务的豁免", AddedAt: now}
|
||||
if err := NewAlertGovernanceRepo(testData{db}).AddExemption(ctx, scope, oldExemption.ChannelKey, func(*business.AlertExemptionTarget) (*business.AlertExemption, error) { return oldExemption, nil }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := db.Exec("UPDATE sys_user_bind_platform SET user_id=8 WHERE source_db='lx'").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
hit.AlertContent = "换绑后的私有数据"
|
||||
write(false)
|
||||
var old alertRecordPO
|
||||
if err := db.First(&old).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if old.AlertContent != "换绑前" {
|
||||
t.Fatal("旧商务预警被新归属数据刷新")
|
||||
}
|
||||
newHit := *hit
|
||||
newScope := business.UserScope{SelfID: 8, IncludeSelf: true, DescendantPath: "0,8"}
|
||||
newHit.OwnerScope = &newScope
|
||||
newHit.BusinessID = 8
|
||||
newHit.RuleID = 2
|
||||
if ok, err := trigger.StoreHit(ctx, &newHit, []int{1, 2, 3, 4}, func(*business.AlertRecord) bool { return true }); err != nil || !ok {
|
||||
t.Fatalf("新商务应产生独立预警: %v %v", ok, err)
|
||||
}
|
||||
var histories []alertRecordPO
|
||||
if err := db.Order("id").Find(&histories).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(histories) != 2 || histories[0].BusinessID != 7 || histories[0].AlertContent != "换绑前" || histories[1].BusinessID != 8 {
|
||||
t.Fatalf("旧商务预警不应被转移给新商务: %+v", histories)
|
||||
}
|
||||
governance := NewAlertGovernanceRepo(testData{db})
|
||||
called := false
|
||||
build := func(*business.AlertExemptionTarget) (*business.AlertExemption, error) {
|
||||
called = true
|
||||
return nil, business.ErrAlertRecordInput
|
||||
}
|
||||
if err := governance.AddExemption(ctx, scope, business.ChannelKey{SourceDB: "lx", ID: 1}, build); err != business.ErrAlertRecordScope || called {
|
||||
t.Fatalf("换绑后仍允许免预警: %v called=%v", err, called)
|
||||
}
|
||||
if err := db.Exec("INSERT INTO sys_user_bind_platform(user_id,source_db,platform_busier_id,status) VALUES(7,'lx',7,'0')").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := governance.AddExemption(ctx, scope, business.ChannelKey{SourceDB: "lx", ID: 1}, build); err != business.ErrAlertRecordScope || called {
|
||||
t.Fatal("不能忽略范围外的冲突绑定")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAlertLossMissingChannelRemainsUnknown(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
now := prepareAlertV2(t, db)
|
||||
if err := db.Create(&alertRecordPO{ID: 1, PlatformCode: "lx", TargetPromoteID: 999, Status: 1, CreatedAt: now}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
scope := business.AlertRecordScope{Owner: business.LevelOneScope{All: true}, Viewer: business.UserScope{All: true}}
|
||||
r := NewAlertRecordRepo(testData{db})
|
||||
page, err := r.List(context.Background(), scope, business.AlertRecordFilter{})
|
||||
if err != nil || page.Total != 1 || page.Rows[0].Lost != nil {
|
||||
t.Fatalf("缺失渠道不应被展示为未流失: %+v %v", page, err)
|
||||
}
|
||||
lost := false
|
||||
page, err = r.List(context.Background(), scope, business.AlertRecordFilter{Lost: &lost})
|
||||
if err != nil || page.Total != 0 {
|
||||
t.Fatalf("未流失筛选应排除未知渠道: %+v %v", page, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAlertComparisonIgnoresInvalidHistoricalTimestamps(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
now := prepareAlertV2(t, db)
|
||||
if err := db.Exec("INSERT INTO tab_user VALUES(99,'lx',1,0,'','unknown')").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key := business.ChannelKey{SourceDB: "lx", ID: 1}
|
||||
values, err := NewAlertTriggerRepo(testData{db}).Comparison(context.Background(), 1, []business.ChannelKey{key}, now, 1)
|
||||
if err != nil || values[key].FirstUnix != now.Add(-48*time.Hour).Unix() {
|
||||
t.Fatalf("无效时间不应阻止正常渠道触发: %+v %v", values, err)
|
||||
}
|
||||
}
|
||||
@@ -12,10 +12,10 @@ type alertRuleRepo struct{ data Provider }
|
||||
|
||||
func NewAlertRuleRepo(data Provider) business.AlertRuleRepo { return &alertRuleRepo{data: data} }
|
||||
func toBizAlertRule(po alertRulePO) *business.AlertRule {
|
||||
return &business.AlertRule{ID: po.ID, Name: po.RuleName, Type: po.RuleType, Metric: po.MonitorMetric, StatPeriodDays: po.StatPeriodDays, MomDirection: po.MomDirection, MomThreshold: po.MomThreshold, UnloginDays: po.UnloginDays, MonitorScope: po.MonitorScope, WarningColor: po.WarningColor, CreatorID: po.CreatorID, Enabled: po.Enabled, IsDeleted: po.IsDeleted, CreatedAt: po.CreatedAt, UpdatedAt: po.UpdatedAt}
|
||||
return &business.AlertRule{ID: po.ID, Name: po.RuleName, Type: po.RuleType, Metric: po.MonitorMetric, ConditionMode: po.ConditionMode, AmountDirection: po.AmountDirection, AmountThreshold: po.AmountThreshold, StatPeriodDays: po.StatPeriodDays, MomDirection: po.MomDirection, MomThreshold: po.MomThreshold, UnloginDays: po.UnloginDays, MonitorScope: po.MonitorScope, WarningColor: po.WarningColor, CreatorID: po.CreatorID, Enabled: po.Enabled, IsDeleted: po.IsDeleted, CreatedAt: po.CreatedAt, UpdatedAt: po.UpdatedAt}
|
||||
}
|
||||
func newAlertRule(r *business.AlertRule) alertRulePO {
|
||||
return alertRulePO{ID: r.ID, RuleName: r.Name, RuleType: r.Type, MonitorMetric: r.Metric, StatPeriodDays: r.StatPeriodDays, MomDirection: r.MomDirection, MomThreshold: r.MomThreshold, UnloginDays: r.UnloginDays, MonitorScope: r.MonitorScope, WarningColor: r.WarningColor, CreatorID: r.CreatorID, Enabled: r.Enabled, IsDeleted: r.IsDeleted}
|
||||
return alertRulePO{ID: r.ID, RuleName: r.Name, RuleType: r.Type, MonitorMetric: r.Metric, ConditionMode: r.ConditionMode, AmountDirection: r.AmountDirection, AmountThreshold: r.AmountThreshold, StatPeriodDays: r.StatPeriodDays, MomDirection: r.MomDirection, MomThreshold: r.MomThreshold, UnloginDays: r.UnloginDays, MonitorScope: r.MonitorScope, WarningColor: r.WarningColor, CreatorID: r.CreatorID, Enabled: r.Enabled, IsDeleted: r.IsDeleted}
|
||||
}
|
||||
func applyOwnerScope(db *gorm.DB, scope business.LevelOneScope, column string) *gorm.DB {
|
||||
if scope.All {
|
||||
@@ -59,7 +59,7 @@ func (r *alertRuleRepo) List(ctx context.Context, scope business.LevelOneScope,
|
||||
}
|
||||
func (r *alertRuleRepo) Find(ctx context.Context, id uint64) (*business.AlertRule, error) {
|
||||
var po alertRulePO
|
||||
err := r.data.DB().WithContext(ctx).First(&po, "id = ?", id).Error
|
||||
err := r.data.DB().WithContext(ctx).First(&po, "id = ? AND is_deleted = 0", id).Error
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, business.ErrAlertRuleNotFound
|
||||
}
|
||||
@@ -82,7 +82,7 @@ func (r *alertRuleRepo) Create(ctx context.Context, rule *business.AlertRule) er
|
||||
}
|
||||
func (r *alertRuleRepo) Update(ctx context.Context, rule *business.AlertRule) error {
|
||||
// 编辑不覆盖启停、删除和创建人字段，避免并发启停被旧配置覆盖。
|
||||
return storageError(r.data.DB().WithContext(ctx).Model(&alertRulePO{}).Where("id = ?", rule.ID).Updates(map[string]any{"rule_name": rule.Name, "rule_type": rule.Type, "monitor_metric": rule.Metric, "stat_period_days": rule.StatPeriodDays, "mom_direction": rule.MomDirection, "mom_threshold": rule.MomThreshold, "unlogin_days": rule.UnloginDays, "monitor_scope": 1, "warning_color": rule.WarningColor, "updated_at": time.Now()}).Error)
|
||||
return storageError(r.data.DB().WithContext(ctx).Model(&alertRulePO{}).Where("id = ?", rule.ID).Updates(map[string]any{"condition_mode": rule.ConditionMode, "amount_direction": rule.AmountDirection, "amount_threshold": rule.AmountThreshold, "rule_name": rule.Name, "rule_type": rule.Type, "monitor_metric": rule.Metric, "stat_period_days": rule.StatPeriodDays, "mom_direction": rule.MomDirection, "mom_threshold": rule.MomThreshold, "unlogin_days": rule.UnloginDays, "monitor_scope": 1, "warning_color": rule.WarningColor, "updated_at": time.Now()}).Error)
|
||||
}
|
||||
func (r *alertRuleRepo) SetDeleted(ctx context.Context, id uint64) error {
|
||||
return storageError(r.data.DB().WithContext(ctx).Model(&alertRulePO{}).Where("id = ?", id).Updates(map[string]any{"is_deleted": 1, "updated_at": time.Now()}).Error)
|
||||
|
||||
@@ -3,6 +3,7 @@ package business
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
@@ -24,9 +25,82 @@ type alertTriggerLockPO struct {
|
||||
|
||||
func (alertTriggerLockPO) TableName() string { return "sys_business_alert_locks" }
|
||||
|
||||
type alertHitKey struct {
|
||||
OwnerID uint
|
||||
Source, Type string
|
||||
PromoteID, RuleID uint64
|
||||
Independent bool
|
||||
}
|
||||
|
||||
func hitKey(source, kind string, promote, rule uint64, independent bool, owner uint) alertHitKey {
|
||||
if !independent {
|
||||
rule = 0
|
||||
}
|
||||
return alertHitKey{OwnerID: owner, Source: strings.ToLower(source), Type: strings.ToLower(kind), PromoteID: promote, RuleID: rule, Independent: independent}
|
||||
}
|
||||
|
||||
// 先批量排除在途记录，避免为不会更新的预警重复扫描累计玩家和流水。
|
||||
// 这是性能预筛选；落库时仍在事务及行锁内复查当前状态。
|
||||
func (r *alertTriggerRepo) FilterWritable(ctx context.Context, hits []*business.AlertRecord, open []int, canWrite func(*business.AlertRecord) bool) ([]*business.AlertRecord, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
current := map[alertHitKey]*business.AlertRecord{}
|
||||
for start := 0; start < len(hits); start += 100 {
|
||||
var condition *gorm.DB
|
||||
for _, hit := range hits[start:min(start+100, len(hits))] {
|
||||
q := db.Where("platform_code=? AND target_promote_id=? AND rule_type=? AND independent_rule=? AND business_id=?", hit.PlatformCode, hit.TargetPromoteID, hit.RuleType, hit.IndependentRule, hit.BusinessID)
|
||||
if hit.IndependentRule {
|
||||
q = q.Where("rule_id=?", hit.RuleID)
|
||||
}
|
||||
if condition == nil {
|
||||
condition = q
|
||||
} else {
|
||||
condition = condition.Or(q)
|
||||
}
|
||||
}
|
||||
var rows []alertRecordPO
|
||||
if err := db.Select("id,platform_code,target_promote_id,rule_type,rule_id,independent_rule,status,business_id").Where(condition).Where("status IN ?", open).Order("id").Find(&rows).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, row := range rows {
|
||||
key := hitKey(row.PlatformCode, row.RuleType, row.TargetPromoteID, row.RuleID, row.IndependentRule, row.BusinessID)
|
||||
if current[key] == nil {
|
||||
current[key] = toBizAlertRecord(row)
|
||||
}
|
||||
}
|
||||
}
|
||||
result := make([]*business.AlertRecord, 0, len(hits))
|
||||
for _, hit := range hits {
|
||||
if canWrite(current[hitKey(hit.PlatformCode, hit.RuleType, hit.TargetPromoteID, hit.RuleID, hit.IndependentRule, hit.BusinessID)]) {
|
||||
result = append(result, hit)
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (r *alertTriggerRepo) Input(ctx context.Context) (*business.AlertTriggerInput, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
out := &business.AlertTriggerInput{}
|
||||
out := &business.AlertTriggerInput{Exemptions: map[business.AlertExemptionKey]bool{}, Gaps: map[uint]*business.AlertGap{}, Dedup: map[uint]bool{}}
|
||||
var exemptions []alertExemptionPO
|
||||
if err := db.Select("source_db,promote_id,owner_id").Where("removed_at IS NULL").Find(&exemptions).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, e := range exemptions {
|
||||
out.Exemptions[business.AlertExemptionKey{ChannelKey: business.ChannelKey{SourceDB: e.SourceDB, ID: e.PromoteID}, OwnerID: e.OwnerID}] = true
|
||||
}
|
||||
var gaps []alertGapPO
|
||||
if err := db.Find(&gaps).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, g := range gaps {
|
||||
out.Gaps[g.OwnerID] = &business.AlertGap{OwnerID: g.OwnerID, Enabled: g.Enabled, Days: g.Days}
|
||||
}
|
||||
var prefs []preferencePO
|
||||
if err := db.Find(&prefs).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, p := range prefs {
|
||||
out.Dedup[p.UserID] = p.IDCardDedup == nil || *p.IDCardDedup != 0
|
||||
}
|
||||
var rules []alertRulePO
|
||||
if err := db.Where("enabled=1 AND is_deleted=0").Order("id").Find(&rules).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
@@ -71,6 +145,30 @@ func (r *alertTriggerRepo) Input(ctx context.Context) (*business.AlertTriggerInp
|
||||
if err := db.Table("tab_promote p").Select("p.id,p.source_db,p.account,p.busier_id,p.last_login_time,p.create_time").Where("p.promote_level=1 AND p.status=1").Where("EXISTS (?)", bound).Order("p.source_db,p.id").Scan(&channels).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
if len(channels) > 0 {
|
||||
keys := []business.ChannelKey{}
|
||||
for _, row := range channels {
|
||||
keys = append(keys, business.ChannelKey{SourceDB: row.SourceDB, ID: row.ID})
|
||||
}
|
||||
latest := map[business.ChannelKey]int64{}
|
||||
for start := 0; start < len(keys); start += 100 {
|
||||
var times []struct {
|
||||
SourceDB string
|
||||
PromoteID uint64
|
||||
Stamp int64
|
||||
}
|
||||
if err := alertTree(db, keys[start:min(start+100, len(keys))]).Select("root.source_db,root.id AS promote_id,MAX(COALESCE(NULLIF(p.last_login_time,0),p.create_time,0)) AS stamp").Group("root.source_db,root.id").Scan(×).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, v := range times {
|
||||
latest[business.ChannelKey{SourceDB: v.SourceDB, ID: v.PromoteID}] = v.Stamp
|
||||
}
|
||||
}
|
||||
for i := range channels {
|
||||
v := latest[business.ChannelKey{SourceDB: channels[i].SourceDB, ID: channels[i].ID}]
|
||||
channels[i].LastLoginTime = &v
|
||||
}
|
||||
}
|
||||
for _, row := range channels {
|
||||
channel := &business.AlertTriggerChannel{ChannelKey: business.ChannelKey{SourceDB: row.SourceDB, ID: row.ID}, Account: row.Account, BusierID: row.BusierID}
|
||||
if row.LastLoginTime != nil {
|
||||
@@ -84,51 +182,28 @@ func (r *alertTriggerRepo) Input(ctx context.Context) (*business.AlertTriggerInp
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r *alertTriggerRepo) Comparison(ctx context.Context, metric int, channels []business.ChannelKey, today time.Time, days int) (map[business.ChannelKey]business.AlertComparison, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
table, stamp, amount, condition := "tab_user", "register_time", "1", "(puid=0 OR puid IS NULL)"
|
||||
if metric == 2 {
|
||||
table, stamp, amount, condition = "tab_spend", "pay_time", "pay_amount", "pay_status=1"
|
||||
} else if metric != 1 {
|
||||
return nil, business.ErrAlertRuleInput
|
||||
}
|
||||
available, err := businessTableAvailable(db, table)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !available {
|
||||
return nil, business.ErrReportBuildUnavailable
|
||||
}
|
||||
result := map[business.ChannelKey]business.AlertComparison{}
|
||||
currentStart, previousStart := today.AddDate(0, 0, -days).Unix(), today.AddDate(0, 0, -2*days).Unix()
|
||||
for start := 0; start < len(channels); start += 500 {
|
||||
var rows []struct {
|
||||
SourceDB string `gorm:"column:source_db"`
|
||||
PromoteID uint64
|
||||
FirstUnix *int64
|
||||
CurrentValue, PreviousValue string
|
||||
}
|
||||
// 原始主账号与成功支付口径，不扩展下级渠道，也不使用身份证偏好。
|
||||
selection := "source_db,promote_id,MIN(" + stamp + ") AS first_unix," +
|
||||
"COALESCE(SUM(CASE WHEN " + stamp + ">=? AND " + stamp + "<? THEN " + amount + " ELSE 0 END),0) AS current_value," +
|
||||
"COALESCE(SUM(CASE WHEN " + stamp + ">=? AND " + stamp + "<? THEN " + amount + " ELSE 0 END),0) AS previous_value"
|
||||
if err := db.Table(table).Select(selection, currentStart, today.Unix(), previousStart, currentStart).Where(condition).Where("(source_db,promote_id) IN ?", channelPairs(channels[start:min(start+500, len(channels))])).Group("source_db,promote_id").Scan(&rows).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, row := range rows {
|
||||
value := business.AlertComparison{Current: row.CurrentValue, Previous: row.PreviousValue}
|
||||
if row.FirstUnix != nil {
|
||||
value.FirstUnix = *row.FirstUnix
|
||||
}
|
||||
result[business.ChannelKey{SourceDB: row.SourceDB, ID: row.PromoteID}] = value
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (r *alertTriggerRepo) StoreHit(ctx context.Context, hit *business.AlertRecord, open []int, shouldWrite func(*business.AlertRecord) bool) (bool, error) {
|
||||
written := false
|
||||
err := r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := lockAlertChannel(tx, hit.PlatformCode, hit.TargetPromoteID); err != nil {
|
||||
return business.ErrStorage
|
||||
}
|
||||
if hit.OwnerScope != nil {
|
||||
_, err := alertExemptionTarget(tx, *hit.OwnerScope, business.ChannelKey{SourceDB: hit.PlatformCode, ID: hit.TargetPromoteID})
|
||||
if errors.Is(err, business.ErrAlertRecordScope) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
var exempt int64
|
||||
if err := tx.Model(&alertExemptionPO{}).Where("source_db=? AND promote_id=? AND owner_id=? AND removed_at IS NULL", hit.PlatformCode, hit.TargetPromoteID, hit.BusinessID).Count(&exempt).Error; err != nil {
|
||||
return business.ErrStorage
|
||||
}
|
||||
if exempt > 0 {
|
||||
return nil
|
||||
}
|
||||
key := alertTriggerLockPO{SourceDB: hit.PlatformCode, PromoteID: hit.TargetPromoteID, RuleType: hit.RuleType}
|
||||
if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&key).Error; err != nil {
|
||||
return business.ErrStorage
|
||||
@@ -136,8 +211,30 @@ func (r *alertTriggerRepo) StoreHit(ctx context.Context, hit *business.AlertReco
|
||||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("source_db=? AND promote_id=? AND rule_type=?", key.SourceDB, key.PromoteID, key.RuleType).Take(&key).Error; err != nil {
|
||||
return business.ErrStorage
|
||||
}
|
||||
if hit.GapDays > 0 {
|
||||
var latest alertRecordPO
|
||||
gapQuery := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("business_id=? AND platform_code=? AND target_promote_id=? AND rule_type=?", hit.BusinessID, key.SourceDB, key.PromoteID, key.RuleType)
|
||||
if hit.IndependentRule {
|
||||
gapQuery = gapQuery.Where("rule_id=? AND independent_rule=1", hit.RuleID)
|
||||
} else {
|
||||
gapQuery = gapQuery.Where("independent_rule=0")
|
||||
}
|
||||
err := gapQuery.Order("created_at DESC,id DESC").Take(&latest).Error
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return business.ErrStorage
|
||||
}
|
||||
if err == nil && hit.CreatedAt.Unix()-latest.CreatedAt.Unix() < int64(hit.GapDays)*86400 {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
var previous alertRecordPO
|
||||
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("platform_code=? AND target_promote_id=? AND rule_type=? AND status IN ?", key.SourceDB, key.PromoteID, key.RuleType, open).Order("id").Take(&previous).Error
|
||||
q := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("business_id=? AND platform_code=? AND target_promote_id=? AND rule_type=? AND status IN ?", hit.BusinessID, key.SourceDB, key.PromoteID, key.RuleType, open)
|
||||
if hit.IndependentRule {
|
||||
q = q.Where("rule_id=? AND independent_rule=1", hit.RuleID)
|
||||
} else {
|
||||
q = q.Where("independent_rule=0")
|
||||
}
|
||||
err := q.Order("id").Take(&previous).Error
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return business.ErrStorage
|
||||
}
|
||||
@@ -149,14 +246,14 @@ func (r *alertTriggerRepo) StoreHit(ctx context.Context, hit *business.AlertReco
|
||||
return nil
|
||||
}
|
||||
day := time.Date(hit.CreatedAt.Year(), hit.CreatedAt.Month(), hit.CreatedAt.Day(), 0, 0, 0, 0, hit.CreatedAt.Location())
|
||||
row := alertRecordPO{PlatformCode: hit.PlatformCode, PlatformName: hit.PlatformName, RuleID: hit.RuleID, RuleName: hit.RuleName, RuleType: hit.RuleType,
|
||||
row := alertRecordPO{IndependentRule: hit.IndependentRule, TotalPlayers: hit.TotalPlayers, TotalPay: hit.TotalPay, WarningColor: hit.WarningColor, PlatformCode: hit.PlatformCode, PlatformName: hit.PlatformName, RuleID: hit.RuleID, RuleName: hit.RuleName, RuleType: hit.RuleType,
|
||||
MonitorMetric: hit.MonitorMetric, TargetPromoteID: hit.TargetPromoteID, TargetPromoteName: hit.TargetPromoteName, BusinessID: hit.BusinessID, BusinessName: hit.BusinessName,
|
||||
AlertContent: hit.AlertContent, Status: hit.Status, AlertDate: &day, CreatedAt: hit.CreatedAt, UpdatedAt: hit.UpdatedAt}
|
||||
if current == nil {
|
||||
err = tx.Create(&row).Error
|
||||
} else {
|
||||
row.ID = previous.ID
|
||||
err = tx.Model(&alertRecordPO{}).Where("id=?", previous.ID).Select("platform_name", "rule_id", "rule_name", "rule_type", "monitor_metric", "target_promote_name", "business_id", "business_name", "alert_content", "status", "alert_date", "created_at", "updated_at").Updates(&row).Error
|
||||
err = tx.Model(&alertRecordPO{}).Where("id=?", previous.ID).Select("independent_rule", "total_players", "total_pay", "warning_color", "platform_name", "rule_id", "rule_name", "rule_type", "monitor_metric", "target_promote_name", "business_id", "business_name", "alert_content", "status", "alert_date", "created_at", "updated_at").Updates(&row).Error
|
||||
}
|
||||
if err != nil {
|
||||
return business.ErrStorage
|
||||
|
||||
@@ -15,19 +15,19 @@ func TestAlertTriggerStorageUsesRawMetricsAndPreservesWorkflow(t *testing.T) {
|
||||
}
|
||||
now := time.Date(2026, 11, 4, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600))
|
||||
for _, sql := range []string{
|
||||
"CREATE TABLE tab_promote(id INTEGER,source_db TEXT,busier_id INTEGER,account TEXT,promote_level INTEGER,status INTEGER,last_login_time INTEGER,create_time INTEGER)",
|
||||
"CREATE TABLE tab_user(id INTEGER,source_db TEXT,promote_id INTEGER,puid INTEGER,register_time TEXT)",
|
||||
"CREATE TABLE tab_promote(id INTEGER,source_db TEXT,busier_id INTEGER,account TEXT,promote_level INTEGER,status INTEGER,last_login_time INTEGER,create_time INTEGER,parent_id INTEGER DEFAULT 0,top_promote_id INTEGER DEFAULT 0)",
|
||||
"CREATE TABLE tab_user(id INTEGER,source_db TEXT,promote_id INTEGER,puid INTEGER,register_time TEXT,idcard TEXT)",
|
||||
"CREATE TABLE tab_spend(source_db TEXT,promote_id INTEGER,pay_time INTEGER,pay_status INTEGER,pay_amount DECIMAL(20,2))",
|
||||
"INSERT INTO sys_users(id,username,level,parent_id,id_path,authority_id) VALUES(10,'一级',1,0,'0',901),(20,'二级',2,10,'0,10',902),(30,'三级',3,20,'0,10,20',903)",
|
||||
"INSERT INTO sys_user_bind_platform(user_id,source_db,platform_busier_id,platform_busier_account,status) VALUES(30,'lx',7,'绑定','0')",
|
||||
"INSERT INTO tab_promote VALUES(10,'lx',7,'正常渠道',1,1,0,1),(11,'lx',7,'停用渠道',1,0,0,1),(12,'lx',7,'下级渠道',2,1,0,1)",
|
||||
"INSERT INTO tab_promote(id,source_db,busier_id,account,promote_level,status,last_login_time,create_time) VALUES(10,'lx',7,'正常渠道',1,1,0,1),(11,'lx',7,'停用渠道',1,0,0,1),(12,'lx',7,'下级渠道',2,1,0,1)",
|
||||
} {
|
||||
if err := db.Exec(sql).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
today := time.Date(2026, 11, 4, 0, 0, 0, 0, now.Location())
|
||||
if err := db.Exec("INSERT INTO tab_user VALUES(1,'lx',10,0,?),(2,'lx',10,0,?),(3,'lx',10,NULL,?),(4,'lx',10,1,?),(5,'hw',10,0,?)", today.AddDate(0, 0, -2).Unix(), today.AddDate(0, 0, -1).Unix(), today.AddDate(0, 0, -1).Unix(), today.AddDate(0, 0, -1).Unix(), today.AddDate(0, 0, -1).Unix()).Error; err != nil {
|
||||
if err := db.Exec("INSERT INTO tab_user(id,source_db,promote_id,puid,register_time) VALUES(1,'lx',10,0,?),(2,'lx',10,0,?),(3,'lx',10,NULL,?),(4,'lx',10,1,?),(5,'hw',10,0,?)", today.AddDate(0, 0, -2).Unix(), today.AddDate(0, 0, -1).Unix(), today.AddDate(0, 0, -1).Unix(), today.AddDate(0, 0, -1).Unix(), today.AddDate(0, 0, -1).Unix()).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Exec("INSERT INTO tab_spend VALUES('lx',10,?,1,100),('lx',10,?,1,110),('lx',10,?,0,999),('hw',10,?,1,999)", today.AddDate(0, 0, -2).Unix(), today.AddDate(0, 0, -1).Unix(), today.AddDate(0, 0, -1).Unix(), today.AddDate(0, 0, -1).Unix()).Error; err != nil {
|
||||
@@ -48,7 +48,7 @@ func TestAlertTriggerStorageUsesRawMetricsAndPreservesWorkflow(t *testing.T) {
|
||||
if err != nil || metrics[business.ChannelKey{SourceDB: "lx", ID: 10}].Current != "2" {
|
||||
t.Fatalf("主账号、NULL父账号或跨平台口径错误: %+v %v", metrics, err)
|
||||
}
|
||||
uc := business.NewAlertTriggerUsecase(r)
|
||||
uc := business.NewAlertTriggerUsecase(r, nil)
|
||||
result, err := uc.Detect(context.Background(), now)
|
||||
if err != nil || result.Written != 1 {
|
||||
t.Fatalf("首次命中: %+v %v", result, err)
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"gorm.io/gorm"
|
||||
"kra/pkg/database/migration"
|
||||
"strings"
|
||||
)
|
||||
|
||||
func migrateAlertV2(db *gorm.DB) error {
|
||||
if err := migration.CreateMissingTables(db, &alertGapPO{}, &alertExemptionPO{}, &alertChannelLockPO{}, &alertRulePO{}, &alertRecordPO{}); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, item := range []struct {
|
||||
model any
|
||||
fields []string
|
||||
}{
|
||||
{&alertRulePO{}, []string{"ConditionMode", "AmountDirection", "AmountThreshold"}},
|
||||
{&alertRecordPO{}, []string{"TotalPlayers", "TotalPay", "WarningColor", "IndependentRule"}},
|
||||
} {
|
||||
if db.Dialector.Name() == "mysql" {
|
||||
statement := &gorm.Statement{DB: db}
|
||||
if err := statement.Parse(item.model); err != nil {
|
||||
return err
|
||||
}
|
||||
columns := []string{}
|
||||
args := []any{}
|
||||
for _, name := range item.fields {
|
||||
if !db.Migrator().HasColumn(item.model, name) {
|
||||
field := statement.Schema.LookUpField(name)
|
||||
definition := db.Migrator().FullDataTypeOf(field)
|
||||
columns = append(columns, "ADD COLUMN `"+field.DBName+"` "+definition.SQL)
|
||||
args = append(args, definition.Vars...)
|
||||
}
|
||||
}
|
||||
// MySQL 5.7 添加普通列会重建表；合并执行，避免每个字段重建一次。
|
||||
if len(columns) > 0 {
|
||||
if err := db.Exec("ALTER TABLE `"+statement.Schema.Table+"` "+strings.Join(columns, ", ")+", ALGORITHM=INPLACE, LOCK=NONE", args...).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
for _, field := range item.fields {
|
||||
if !db.Migrator().HasColumn(item.model, field) {
|
||||
if err := db.Migrator().AddColumn(item.model, field); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := migrateRegistrationIndex(db); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := migrateAlertV2UniqueKey(db); err != nil {
|
||||
return err
|
||||
}
|
||||
return migrateQueryIndexes(db, []businessQueryIndex{
|
||||
{"sys_alert_record", "idx_alert_owner_status_time", []string{"business_id", "status", "created_at", "id"}},
|
||||
{"sys_alert_record", "idx_alert_channel_type_time", []string{"platform_code", "target_promote_id", "rule_type", "created_at", "id"}},
|
||||
{"sys_alert_record", "idx_alert_owner_time", []string{"business_id", "created_at", "id"}},
|
||||
{"sys_alert_record", "idx_alert_created", []string{"created_at", "id"}},
|
||||
{"sys_alert_operation_log", "idx_alert_log_time", []string{"alert_id", "created_at", "id"}},
|
||||
{"tab_spend", "idx_alert_spend_window", []string{"source_db", "promote_id", "pay_status", "pay_time", "pay_amount"}},
|
||||
{"tab_user", "idx_alert_user_window", []string{"source_db", "promote_id", "sw_register_time_unix", "puid"}},
|
||||
{"tab_promote", "idx_alert_promote_tree", []string{"source_db", "top_promote_id", "id"}},
|
||||
})
|
||||
}
|
||||
|
||||
func migrateAlertV2UniqueKey(db *gorm.DB) error {
|
||||
if db.Dialector.Name() != "mysql" {
|
||||
return nil
|
||||
}
|
||||
// 旧唯一键将已忽略视为未闭环，也无法容纳同类的两条绝对值规则。
|
||||
// 保留旧生成列，原子替换唯一键；不改写、合并或删除历史记录。
|
||||
clauses := []string{}
|
||||
if !db.Migrator().HasColumn("sys_alert_record", "v2_open_key") {
|
||||
// MySQL 5.7 的在线虚拟列变更不能与索引变更放在同一 ALTER 中。
|
||||
if err := db.Exec("ALTER TABLE sys_alert_record ADD COLUMN v2_open_key BIGINT UNSIGNED GENERATED ALWAYS AS (CASE WHEN status IN (1,2,3,4) THEN CASE WHEN independent_rule=1 THEN rule_id ELSE 0 END ELSE NULL END) VIRTUAL, ALGORITHM=INPLACE, LOCK=NONE").Error; err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
var column struct {
|
||||
GenerationExpression string
|
||||
Extra string
|
||||
}
|
||||
if err := db.Raw("SELECT generation_expression,extra FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='sys_alert_record' AND column_name='v2_open_key'").Scan(&column).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
normalize := strings.NewReplacer("`", "", " ", "", "\n", "", "\t", "", "(", "", ")", "")
|
||||
if normalize.Replace(strings.ToLower(column.GenerationExpression)) != "casewhenstatusin1,2,3,4thencasewhenindependent_rule=1thenrule_idelse0endelsenullend" || !strings.Contains(strings.ToUpper(column.Extra), "VIRTUAL GENERATED") {
|
||||
return fmt.Errorf("sys_alert_record.v2_open_key 的现有定义不符，未修改原唯一索引")
|
||||
}
|
||||
}
|
||||
var existing []struct {
|
||||
ColumnName string
|
||||
NonUnique int
|
||||
SubPart *int
|
||||
}
|
||||
if err := db.Raw("SELECT column_name,non_unique,sub_part FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='sys_alert_record' AND index_name='uk_alert_record_v2_open' ORDER BY seq_in_index").Scan(&existing).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
replaceV2 := false
|
||||
if len(existing) > 0 {
|
||||
columns := []string{}
|
||||
for _, column := range existing {
|
||||
if column.NonUnique != 0 || column.SubPart != nil {
|
||||
return fmt.Errorf("sys_alert_record.uk_alert_record_v2_open 必须是完整字段的唯一索引")
|
||||
}
|
||||
columns = append(columns, column.ColumnName)
|
||||
}
|
||||
layout := strings.Join(columns, ",")
|
||||
replaceV2 = layout == "platform_code,target_promote_id,rule_type,v2_open_key"
|
||||
if layout != "business_id,platform_code,target_promote_id,rule_type,v2_open_key" && !replaceV2 {
|
||||
return fmt.Errorf("sys_alert_record.uk_alert_record_v2_open 的现有索引列不符，未修改原唯一索引")
|
||||
}
|
||||
}
|
||||
for _, name := range []string{"uk_alert_record_unclosed", "uk_alert_record_daily"} {
|
||||
if db.Migrator().HasIndex("sys_alert_record", name) {
|
||||
clauses = append(clauses, "DROP INDEX `"+name+"`")
|
||||
}
|
||||
}
|
||||
if replaceV2 {
|
||||
clauses = append(clauses, "DROP INDEX uk_alert_record_v2_open")
|
||||
}
|
||||
if len(existing) == 0 || replaceV2 {
|
||||
clauses = append(clauses, "ADD UNIQUE INDEX uk_alert_record_v2_open(business_id,platform_code,target_promote_id,rule_type,v2_open_key)")
|
||||
}
|
||||
if len(clauses) == 0 {
|
||||
return nil
|
||||
}
|
||||
return db.Exec("ALTER TABLE sys_alert_record " + strings.Join(clauses, ", ") + ", ALGORITHM=INPLACE, LOCK=NONE").Error
|
||||
}
|
||||
@@ -0,0 +1,171 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"kra/internal/biz/business"
|
||||
)
|
||||
|
||||
func prepareAlertV2(t *testing.T, db *gorm.DB) time.Time {
|
||||
t.Helper()
|
||||
now := time.Date(2026, 11, 5, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600))
|
||||
for _, sql := range []string{
|
||||
"CREATE TABLE tab_promote(id BIGINT,source_db VARCHAR(20),busier_id BIGINT,account VARCHAR(100),promote_level INT,status INT,last_login_time BIGINT,create_time BIGINT,parent_id BIGINT,top_promote_id BIGINT,PRIMARY KEY(source_db,id))",
|
||||
"CREATE TABLE tab_user(id BIGINT,source_db VARCHAR(20),promote_id BIGINT,puid BIGINT,register_time VARCHAR(30),idcard VARCHAR(100),PRIMARY KEY(source_db,id))",
|
||||
"CREATE TABLE tab_spend(id BIGINT PRIMARY KEY,source_db VARCHAR(20),promote_id BIGINT,pay_status INT,pay_time BIGINT,pay_amount DECIMAL(20,2))",
|
||||
"INSERT INTO tab_promote VALUES(1,'lx',7,'父渠道',1,1,0,1,0,0),(2,'lx',7,'子渠道',2,1,0,1,1,1),(3,'lx',7,'孙渠道',3,1,0,1,2,1),(1,'hw',7,'同号渠道',1,1,0,1,0,0)",
|
||||
} {
|
||||
if err := db.Exec(sql).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := db.Exec("UPDATE tab_promote SET last_login_time=?", now.Unix()).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Exec("INSERT INTO tab_user VALUES(1,'lx',1,0,?,'card-a'),(2,'lx',2,0,?,'card-a'),(3,'lx',3,0,?,'card-b'),(4,'lx',3,1,?,'card-c'),(1,'hw',1,0,?,'other')", now.AddDate(0, 0, -2).Unix(), now.AddDate(0, 0, -1).Unix(), now.AddDate(0, 0, -1).Unix(), now.AddDate(0, 0, -1).Unix(), now.AddDate(0, 0, -1).Unix()).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Exec("INSERT INTO tab_spend VALUES(1,'lx',1,1,?,100),(2,'lx',2,1,?,2000),(3,'lx',3,1,?,1000),(4,'lx',3,0,?,999999),(5,'hw',1,1,?,999999)", now.AddDate(0, 0, -2).Unix(), now.AddDate(0, 0, -1).Unix(), now.AddDate(0, 0, -1).Unix(), now.AddDate(0, 0, -1).Unix(), now.AddDate(0, 0, -1).Unix()).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return now
|
||||
}
|
||||
|
||||
func TestAlertV2SnapshotsTreeAndBoundedComparison(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
now := prepareAlertV2(t, db)
|
||||
repo := NewAlertTriggerRepo(testData{db})
|
||||
key := business.ChannelKey{SourceDB: "lx", ID: 1}
|
||||
sample, err := repo.Comparison(context.Background(), 2, []business.ChannelKey{key}, now, 1)
|
||||
if err != nil || sample[key].Current != "3000.00" || sample[key].Previous != "100.00" {
|
||||
t.Fatalf("子渠道汇总或平台隔离错误: %+v %v", sample, err)
|
||||
}
|
||||
for _, dedup := range []bool{true, false} {
|
||||
totals, err := repo.Snapshots(context.Background(), []business.ChannelKey{key}, now, dedup)
|
||||
want := int64(3)
|
||||
if dedup {
|
||||
want = 2
|
||||
}
|
||||
if err != nil || totals[key].Players != want || totals[key].Pay != "3100" {
|
||||
t.Fatalf("累计快照或去重错误: %+v %v", totals, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAlertV2ExemptionHistoryAndAutomaticRemoval(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
now := prepareAlertV2(t, db)
|
||||
if err := db.AutoMigrate(&accountPO{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Exec("INSERT INTO sys_users(id,username,level,id_path) VALUES(7,'owner',1,'0')").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Exec("INSERT INTO sys_user_bind_platform(user_id,source_db,platform_busier_id,status) VALUES(7,'lx',7,'0')").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r := NewAlertGovernanceRepo(testData{db})
|
||||
ctx := context.Background()
|
||||
entry := &business.AlertExemption{ChannelKey: business.ChannelKey{SourceDB: "lx", ID: 1}, ChannelName: "父渠道", OwnerID: 7, AddedBy: 7, AddedName: "一级商务", AddedAt: now, Reason: "长期合作"}
|
||||
if err := r.AddExemption(ctx, business.UserScope{All: true}, entry.ChannelKey, func(*business.AlertExemptionTarget) (*business.AlertExemption, error) { return entry, nil }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := r.AddExemption(ctx, business.UserScope{All: true}, entry.ChannelKey, func(*business.AlertExemptionTarget) (*business.AlertExemption, error) { return entry, nil }); err != business.ErrAlertExemptExists {
|
||||
t.Fatalf("重复加入: %v", err)
|
||||
}
|
||||
rows, total, err := r.Exemptions(ctx, business.LevelOneScope{UserID: 7}, business.AlertExemptionFilter{})
|
||||
if err != nil || total != 1 {
|
||||
t.Fatalf("名单: %+v %v", rows, err)
|
||||
}
|
||||
if err := r.RemoveExemption(ctx, business.LevelOneScope{UserID: 8}, rows[0].ID, 8, "越权", "手动移出", now); err != business.ErrAlertExemptNotFound {
|
||||
t.Fatalf("跨商务移出: %v", err)
|
||||
}
|
||||
hit := &business.AlertRecord{BusinessID: 7, PlatformCode: "lx", TargetPromoteID: 1, RuleID: 1, RuleType: "充值", Status: 1, CreatedAt: now, UpdatedAt: now}
|
||||
written, err := NewAlertTriggerRepo(testData{db}).StoreHit(ctx, hit, []int{1, 2, 3, 4}, func(*business.AlertRecord) bool { return true })
|
||||
if err != nil || written {
|
||||
t.Fatalf("免预警仍生成: %v %v", written, err)
|
||||
}
|
||||
// 父渠道自身近 30 天流水仅 100 元，满足激活池条件；子渠道流水不冒充它自身的流水。
|
||||
uc := business.NewAlertGovernanceUsecase(r, nil)
|
||||
if err := uc.Reconcile(ctx, now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rows, _, _ = r.Exemptions(ctx, business.LevelOneScope{All: true}, business.AlertExemptionFilter{})
|
||||
if rows[0].RemovedAt == nil || rows[0].RemovedName != "系统（自动移出）" || rows[0].RemoveReason != "进入激活池 · 自动移出" {
|
||||
t.Fatalf("自动移出未留痕: %+v", rows[0])
|
||||
}
|
||||
if err := r.AddExemption(ctx, business.UserScope{All: true}, entry.ChannelKey, func(*business.AlertExemptionTarget) (*business.AlertExemption, error) { return entry, nil }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, total, err = r.Exemptions(ctx, business.LevelOneScope{All: true}, business.AlertExemptionFilter{})
|
||||
if err != nil || total != 2 {
|
||||
t.Fatal("重新加入不得覆盖历史")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAlertV2GapSeparatesAbsoluteRulesAndIgnoredRecords(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
r := NewAlertTriggerRepo(testData{db})
|
||||
now := time.Now().Truncate(time.Second)
|
||||
hit := &business.AlertRecord{PlatformCode: "lx", TargetPromoteID: 1, RuleID: 1, RuleType: "充值", Status: 1, CreatedAt: now, UpdatedAt: now, GapDays: 7, IndependentRule: true}
|
||||
decide := func(current *business.AlertRecord) bool {
|
||||
return current == nil || current.Status == business.AlertUnprocessed
|
||||
}
|
||||
write := func(want bool) {
|
||||
t.Helper()
|
||||
got, err := r.StoreHit(context.Background(), hit, []int{1, 2, 3, 4}, decide)
|
||||
if err != nil || got != want {
|
||||
t.Fatalf("written=%v want=%v err=%v", got, want, err)
|
||||
}
|
||||
}
|
||||
write(true)
|
||||
hit.RuleID = 2
|
||||
write(true)
|
||||
hit.RuleID = 1
|
||||
write(false)
|
||||
if err := db.Model(&alertRecordPO{}).Where("rule_id=1").Update("status", business.AlertIgnored).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
hit.CreatedAt = now.AddDate(0, 0, 6)
|
||||
write(false)
|
||||
hit.CreatedAt = now.AddDate(0, 0, 7)
|
||||
write(true)
|
||||
var rows []alertRecordPO
|
||||
if err := db.Order("id").Find(&rows).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(rows) != 3 || rows[0].Status != business.AlertIgnored {
|
||||
t.Fatal("已忽略记录被复活或独立规则被覆盖")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAlertV2LiveLossDoesNotRewriteSnapshots(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
now := prepareAlertV2(t, db)
|
||||
// 流失检测使用父渠道的当前值，快照保持触发时的值。
|
||||
players, pay := int64(2), "3100.00"
|
||||
if err := db.Create(&alertRecordPO{ID: 1, PlatformCode: "lx", TargetPromoteID: 1, BusinessID: 7, Status: 1, AlertContent: "满2000", TotalPlayers: &players, TotalPay: &pay, CreatedAt: now}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
scope := business.AlertRecordScope{Owner: business.LevelOneScope{All: true}, Viewer: business.UserScope{All: true}}
|
||||
r := NewAlertRecordRepo(testData{db})
|
||||
lost := true
|
||||
page, err := r.List(context.Background(), scope, business.AlertRecordFilter{Lost: &lost, AlertContent: "2000"})
|
||||
if err != nil || page.Total != 1 || page.Rows[0].Lost == nil || !*page.Rows[0].Lost {
|
||||
t.Fatalf("实时流失筛选: %+v %v", page, err)
|
||||
}
|
||||
if err := db.Exec("UPDATE tab_spend SET pay_amount=1000 WHERE id=1").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Exec("UPDATE tab_promote SET last_login_time=?", time.Now().Unix()).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
lost = false
|
||||
page, err = r.List(context.Background(), scope, business.AlertRecordFilter{Lost: &lost})
|
||||
if err != nil || page.Total != 1 || *page.Rows[0].TotalPlayers != 2 || *page.Rows[0].TotalPay != "3100" {
|
||||
t.Fatalf("实时状态或冻结快照错误: %+v %v", page, err)
|
||||
}
|
||||
}
|
||||
@@ -6,7 +6,7 @@ import (
|
||||
)
|
||||
|
||||
func Migrations() []migration.Step {
|
||||
return []migration.Step{{ID: "202611040017_business_groups_goals_effectiveness", Migrate: migrateBusinessManagement}, {ID: "202611040018_business_effectiveness_query_index", Migrate: migrateEffectivenessQueryIndex}, {ID: "202611040015_business_approved_game_index", Migrate: migrateApprovedGameIndexes}, {ID: "202611040014_business_report_build_indexes", Migrate: migrateReportBuildQueryIndexes}, {ID: "202611040013_business_query_indexes", Migrate: migrateBusinessQueryIndexes}, {ID: "202611040012_business_server_report_index", Migrate: migrateServerReportIndex}, {ID: "202611040011_business_registration_index", Migrate: migrateRegistrationIndex}, {ID: "202611040010_business_alert_trigger_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &alertTriggerLockPO{}) }}, {ID: "202611040009_business_menu_icons", Migrate: migrateBusinessMenuIcons}, {ID: "202611040008_business_report_build_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &reportBuildLockPO{}) }}, {ID: "202611030007_business_alert_attachments", Migrate: migrateAlertAttachments}, {ID: "202611030006_business_alert_schema", Migrate: migrateAlertSchema}, {ID: "202611030005_business_password_reset", Migrate: func(db *gorm.DB) error {
|
||||
return []migration.Step{{ID: "202611050020_business_alert_v2", Migrate: migrateAlertV2}, {ID: "202611040017_business_groups_goals_effectiveness", Migrate: migrateBusinessManagement}, {ID: "202611040018_business_effectiveness_query_index", Migrate: migrateEffectivenessQueryIndex}, {ID: "202611040015_business_approved_game_index", Migrate: migrateApprovedGameIndexes}, {ID: "202611040014_business_report_build_indexes", Migrate: migrateReportBuildQueryIndexes}, {ID: "202611040013_business_query_indexes", Migrate: migrateBusinessQueryIndexes}, {ID: "202611040012_business_server_report_index", Migrate: migrateServerReportIndex}, {ID: "202611040011_business_registration_index", Migrate: migrateRegistrationIndex}, {ID: "202611040010_business_alert_trigger_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &alertTriggerLockPO{}) }}, {ID: "202611040009_business_menu_icons", Migrate: migrateBusinessMenuIcons}, {ID: "202611040008_business_report_build_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &reportBuildLockPO{}) }}, {ID: "202611030007_business_alert_attachments", Migrate: migrateAlertAttachments}, {ID: "202611030006_business_alert_schema", Migrate: migrateAlertSchema}, {ID: "202611030005_business_password_reset", Migrate: func(db *gorm.DB) error {
|
||||
return migration.CreateMissingTables(db, &resetCodePO{}, &resetTokenPO{})
|
||||
}}, {ID: "202611030004_business_binding_locks", Migrate: func(db *gorm.DB) error {
|
||||
return migration.CreateMissingTables(db, &bindingLockPO{})
|
||||
|
||||
@@ -70,6 +70,9 @@ func TestMySQLAlertTriggerConcurrentDeduplication(t *testing.T) {
|
||||
if err := db.Exec("ALTER TABLE sys_alert_record ADD COLUMN unclosed_flag TINYINT GENERATED ALWAYS AS (CASE WHEN status=5 THEN NULL ELSE 1 END) VIRTUAL, ADD UNIQUE INDEX uk_alert_record_unclosed(platform_code,target_promote_id,rule_type,unclosed_flag)").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := migrateAlertV2UniqueKey(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r := NewAlertTriggerRepo(testData{db})
|
||||
now := time.Now().Truncate(time.Millisecond)
|
||||
hit := &business.AlertRecord{RuleID: 1, PlatformCode: "lx", PlatformName: "龙翔", TargetPromoteID: 10, RuleType: "注册", Status: 1, BusinessID: 10, CreatedAt: now, UpdatedAt: now}
|
||||
|
||||
@@ -40,6 +40,7 @@ var ProviderSet = wire.NewSet(
|
||||
databusiness.NewPlatformAccountRepo,
|
||||
databusiness.NewPasswordResetRepo,
|
||||
databusiness.NewBusinessPromoteRepo,
|
||||
databusiness.NewAlertGovernanceRepo,
|
||||
databusiness.NewAlertRuleRepo,
|
||||
databusiness.NewAlertRecordRepo,
|
||||
databusiness.NewActivationRepo,
|
||||
|
||||
Reference in New Issue
Block a user