266 lines
11 KiB
Go
266 lines
11 KiB
Go
package business
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"strings"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
"kra/internal/biz/business"
|
|
)
|
|
|
|
type alertTriggerRepo struct{ data Provider }
|
|
|
|
func NewAlertTriggerRepo(data Provider) business.AlertTriggerRepo {
|
|
return &alertTriggerRepo{data: data}
|
|
}
|
|
|
|
type alertTriggerLockPO struct {
|
|
SourceDB string `gorm:"column:source_db;primaryKey;size:50"`
|
|
PromoteID uint64 `gorm:"primaryKey;autoIncrement:false"`
|
|
RuleType string `gorm:"primaryKey;size:50"`
|
|
}
|
|
|
|
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{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
|
|
}
|
|
for _, row := range rules {
|
|
out.Rules = append(out.Rules, toBizAlertRule(row))
|
|
}
|
|
if len(rules) == 0 {
|
|
return out, nil
|
|
}
|
|
var accounts []accountPO
|
|
if err := db.Select("id,username,level,parent_id,id_path,authority_id").Find(&accounts).Error; err != nil {
|
|
return nil, business.ErrStorage
|
|
}
|
|
for _, row := range accounts {
|
|
out.Accounts = append(out.Accounts, toBizAccount(row))
|
|
}
|
|
var bindings []bindingPO
|
|
if err := db.Where("status=?", "0").Find(&bindings).Error; err != nil {
|
|
return nil, business.ErrStorage
|
|
}
|
|
for _, row := range bindings {
|
|
out.Bindings = append(out.Bindings, toBizBinding(row))
|
|
}
|
|
if len(bindings) == 0 {
|
|
return out, nil
|
|
}
|
|
available, err := businessTableAvailable(db, "tab_promote")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !available {
|
|
return nil, business.ErrAlertChannelsUnavailable
|
|
}
|
|
var channels []struct {
|
|
ID, BusierID uint64
|
|
SourceDB string `gorm:"column:source_db"`
|
|
Account string
|
|
LastLoginTime, CreateTime *int64
|
|
}
|
|
bound := db.Table("sys_user_bind_platform b").Select("1").Where("b.status=? AND b.source_db=p.source_db AND b.platform_busier_id=p.busier_id", "0")
|
|
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 {
|
|
channel.LastLoginUnix = *row.LastLoginTime
|
|
}
|
|
if row.CreateTime != nil {
|
|
channel.CreatedUnix = *row.CreateTime
|
|
}
|
|
out.Channels = append(out.Channels, channel)
|
|
}
|
|
return out, 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
|
|
}
|
|
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
|
|
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
|
|
}
|
|
var current *business.AlertRecord
|
|
if err == nil {
|
|
current = toBizAlertRecord(previous)
|
|
}
|
|
if !shouldWrite(current) {
|
|
return nil
|
|
}
|
|
day := time.Date(hit.CreatedAt.Year(), hit.CreatedAt.Month(), hit.CreatedAt.Day(), 0, 0, 0, 0, hit.CreatedAt.Location())
|
|
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("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
|
|
}
|
|
written = true
|
|
return nil
|
|
})
|
|
return written, err
|
|
}
|