feat: 迁移预警处理审核与备注附件流程
This commit is contained in:
Yvan
@@ -0,0 +1,329 @@
|
||||
package business
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
"kra/internal/biz/business"
|
||||
)
|
||||
|
||||
type alertRecordRepo struct{ data Provider }
|
||||
|
||||
func NewAlertRecordRepo(data Provider) business.AlertRecordRepo { return &alertRecordRepo{data: data} }
|
||||
|
||||
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}
|
||||
}
|
||||
|
||||
func alertBindingQuery(db *gorm.DB, scope business.UserScope) *gorm.DB {
|
||||
q := db.Table("sys_user_bind_platform b").Where("b.status = ? AND b.source_db IN ? AND b.platform_busier_id IS NOT NULL", "0", []string{"lx", "hw", "yq"})
|
||||
if !scope.All {
|
||||
q = q.Joins("JOIN sys_users u ON u.id=b.user_id AND u.deleted_at IS NULL")
|
||||
q = applyUserScope(q, scope, "u.")
|
||||
}
|
||||
return q
|
||||
}
|
||||
|
||||
func alertChannelScopes(scope business.AlertRecordScope) []business.UserScope {
|
||||
var scopes []business.UserScope
|
||||
if !scope.Viewer.All {
|
||||
scopes = append(scopes, scope.Viewer)
|
||||
}
|
||||
if scope.Selected != nil {
|
||||
scopes = append(scopes, *scope.Selected)
|
||||
}
|
||||
return scopes
|
||||
}
|
||||
|
||||
// 先判断没有绑定的确定空集，再检查渠道表；缺表不能放宽权限或伪造统计。
|
||||
func prepareAlertScope(db *gorm.DB, scope business.AlertRecordScope) (empty, available bool, err error) {
|
||||
if scope.Owner.Empty {
|
||||
return true, true, nil
|
||||
}
|
||||
scopes := alertChannelScopes(scope)
|
||||
for _, s := range scopes {
|
||||
if s.Empty {
|
||||
return true, true, nil
|
||||
}
|
||||
var count int64
|
||||
if err := alertBindingQuery(db, s).Count(&count).Error; err != nil {
|
||||
return false, false, business.ErrStorage
|
||||
}
|
||||
if count == 0 {
|
||||
return true, true, nil
|
||||
}
|
||||
}
|
||||
if len(scopes) == 0 {
|
||||
return false, true, nil
|
||||
}
|
||||
available, err = businessTableAvailable(db, "tab_promote")
|
||||
return false, available, err
|
||||
}
|
||||
|
||||
func alertScopedQuery(db *gorm.DB, scope business.AlertRecordScope) *gorm.DB {
|
||||
q := applyOwnerScope(db.Table("sys_alert_record r"), scope.Owner, "r.business_id")
|
||||
for _, s := range alertChannelScopes(scope) {
|
||||
bindings := alertBindingQuery(db, s).Select("1").Where("b.source_db=p.source_db AND b.platform_busier_id=p.busier_id")
|
||||
// 两个范围独立取交集；只限制一级渠道，停用渠道的历史预警仍可见。
|
||||
channels := db.Table("tab_promote p").Select("1").Where("p.id=r.target_promote_id AND p.source_db=r.platform_code AND p.promote_level = ?", 1).Where("EXISTS (?)", bindings)
|
||||
q = q.Where("EXISTS (?)", channels)
|
||||
}
|
||||
return q
|
||||
}
|
||||
|
||||
func (r *alertRecordRepo) List(ctx context.Context, scope business.AlertRecordScope, filter business.AlertRecordFilter, options ...business.ListOption) (*business.AlertRecordPage, error) {
|
||||
db := r.data.DB().WithContext(ctx)
|
||||
page := &business.AlertRecordPage{Rows: []*business.AlertRecord{}, StatusCount: map[int]int64{}, Available: true}
|
||||
empty, available, err := prepareAlertScope(db, scope)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
page.Available = available
|
||||
if empty || !available {
|
||||
return page, nil
|
||||
}
|
||||
base := func() *gorm.DB {
|
||||
q := alertScopedQuery(db, scope)
|
||||
if len(filter.PlatformCodes) > 0 {
|
||||
q = q.Where("r.platform_code IN ?", filter.PlatformCodes)
|
||||
}
|
||||
if filter.TargetPromoteID != nil {
|
||||
q = q.Where("r.target_promote_id = ?", *filter.TargetPromoteID)
|
||||
}
|
||||
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+"%")
|
||||
}
|
||||
}
|
||||
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.Status != nil {
|
||||
q = q.Where("r.status = ?", *filter.Status)
|
||||
}
|
||||
if err := q.Count(&page.Total).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
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 {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
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
|
||||
}
|
||||
return page, nil
|
||||
}
|
||||
|
||||
func findScopedAlert(db *gorm.DB, scope business.AlertRecordScope, id uint64, lock bool) (*alertRecordPO, error) {
|
||||
empty, available, err := prepareAlertScope(db, scope)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if empty {
|
||||
return nil, business.ErrAlertRecordScope
|
||||
}
|
||||
if !available {
|
||||
return nil, business.ErrAlertChannelsUnavailable
|
||||
}
|
||||
// 锁定读取本身带权限条件，避免先读旧快照的权限再锁到已换归属的记录。
|
||||
q := alertScopedQuery(db, scope).Select("r.*")
|
||||
if lock {
|
||||
q = q.Clauses(clause.Locking{Strength: "UPDATE"})
|
||||
}
|
||||
var po alertRecordPO
|
||||
if err := q.Where("r.id = ?", id).Take(&po).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
var count int64
|
||||
if err := db.Model(&alertRecordPO{}).Where("id = ?", id).Count(&count).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
if count > 0 {
|
||||
return nil, business.ErrAlertRecordScope
|
||||
}
|
||||
return nil, business.ErrAlertRecordNotFound
|
||||
}
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
return &po, nil
|
||||
}
|
||||
|
||||
func (r *alertRecordRepo) Detail(ctx context.Context, scope business.AlertRecordScope, id uint64) (*business.AlertRecordDetail, error) {
|
||||
var detail *business.AlertRecordDetail
|
||||
err := r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
po, err := findScopedAlert(tx, scope, id, false)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
logs, err := alertLogs(tx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
detail = &business.AlertRecordDetail{Record: toBizAlertRecord(*po), Logs: logs}
|
||||
return nil
|
||||
})
|
||||
return detail, storageError(err)
|
||||
}
|
||||
|
||||
func (r *alertRecordRepo) WithRecord(ctx context.Context, scope business.AlertRecordScope, id uint64, fn func(*business.AlertRecord) ([]*business.AlertLog, error)) error {
|
||||
return storageError(r.data.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
po, err := findScopedAlert(tx, scope, id, true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
record := toBizAlertRecord(*po)
|
||||
logs, err := fn(record)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
now := time.Now()
|
||||
if record.Status != po.Status {
|
||||
result := tx.Model(&alertRecordPO{}).Where("id = ? AND status = ?", id, po.Status).Updates(map[string]any{"status": record.Status, "updated_at": now})
|
||||
if result.Error != nil {
|
||||
return result.Error
|
||||
}
|
||||
if result.RowsAffected != 1 {
|
||||
return business.ErrAlertRecordChanged
|
||||
}
|
||||
}
|
||||
for _, log := range logs {
|
||||
attachments := encodeAlertAttachments(log.AttachmentIDs)
|
||||
row := alertLogPO{AlertID: id, OperatorID: log.OperatorID, OperatorName: log.OperatorName, LogType: log.LogType, ActionDetail: log.ActionDetail, NoteContent: log.NoteContent, AttachmentIDs: attachments, CreatedAt: now}
|
||||
if err := tx.Create(&row).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}))
|
||||
}
|
||||
|
||||
func encodeAlertAttachments(values []string) *string {
|
||||
if len(values) == 0 {
|
||||
return nil
|
||||
}
|
||||
ids := make([]int64, 0, len(values))
|
||||
for _, value := range values {
|
||||
id, err := strconv.ParseInt(value, 10, 64)
|
||||
if err != nil {
|
||||
ids = nil
|
||||
break
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
var raw []byte
|
||||
if ids != nil {
|
||||
raw, _ = json.Marshal(ids)
|
||||
} else {
|
||||
raw, _ = json.Marshal(values)
|
||||
}
|
||||
value := string(raw)
|
||||
return &value
|
||||
}
|
||||
|
||||
func decodeAlertAttachments(raw *string) (values []string, ids []int64) {
|
||||
if raw == nil {
|
||||
return nil, nil
|
||||
}
|
||||
var entries []json.RawMessage
|
||||
if json.Unmarshal([]byte(*raw), &entries) != nil {
|
||||
return nil, nil
|
||||
}
|
||||
allNumeric := true
|
||||
for _, entry := range entries {
|
||||
var value string
|
||||
if json.Unmarshal(entry, &value) != nil {
|
||||
var n json.Number
|
||||
if json.Unmarshal(entry, &n) != nil || n == "" {
|
||||
return nil, nil
|
||||
}
|
||||
value = n.String()
|
||||
}
|
||||
values = append(values, value)
|
||||
id, err := strconv.ParseInt(value, 10, 64)
|
||||
if err != nil {
|
||||
allNumeric = false
|
||||
} else {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
}
|
||||
if !allNumeric {
|
||||
ids = nil
|
||||
}
|
||||
return values, ids
|
||||
}
|
||||
|
||||
func alertLogs(db *gorm.DB, alertID uint64) ([]*business.AlertLog, error) {
|
||||
var pos []alertLogPO
|
||||
if err := db.Where("alert_id = ?", alertID).Order("created_at DESC,id DESC").Find(&pos).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
logs := make([]*business.AlertLog, 0, len(pos))
|
||||
idsByLog := map[uint64][]int64{}
|
||||
ids := map[int64]bool{}
|
||||
for _, po := range pos {
|
||||
values, numeric := decodeAlertAttachments(po.AttachmentIDs)
|
||||
log := &business.AlertLog{ID: po.ID, AlertID: po.AlertID, OperatorID: po.OperatorID, OperatorName: po.OperatorName, LogType: po.LogType, ActionDetail: po.ActionDetail, NoteContent: po.NoteContent, CreatedAt: po.CreatedAt, Attachments: []string{}}
|
||||
if len(numeric) > 0 {
|
||||
idsByLog[po.ID] = numeric
|
||||
for _, id := range numeric {
|
||||
ids[id] = true
|
||||
}
|
||||
} else if len(values) > 0 {
|
||||
log.Attachments = values
|
||||
}
|
||||
logs = append(logs, log)
|
||||
}
|
||||
if len(ids) == 0 {
|
||||
return logs, nil
|
||||
}
|
||||
keys := make([]int64, 0, len(ids))
|
||||
for id := range ids {
|
||||
keys = append(keys, id)
|
||||
}
|
||||
// 一次批量查询代替旧实现逐日志、逐附件查询。
|
||||
urls := map[int64]string{}
|
||||
for start := 0; start < len(keys); start += 500 {
|
||||
var files []struct {
|
||||
ID int64
|
||||
URL string
|
||||
}
|
||||
if err := db.Table("media_file_upload_and_downloads").Select("id,url").Where("id IN ? AND deleted_at IS NULL", keys[start:min(start+500, len(keys))]).Scan(&files).Error; err != nil {
|
||||
return nil, business.ErrStorage
|
||||
}
|
||||
for _, file := range files {
|
||||
urls[file.ID] = file.URL
|
||||
}
|
||||
}
|
||||
for _, log := range logs {
|
||||
seen := map[int64]bool{}
|
||||
for _, id := range idsByLog[log.ID] {
|
||||
if url, exists := urls[id]; exists && !seen[id] {
|
||||
log.Attachments = append(log.Attachments, url)
|
||||
seen[id] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
return logs, nil
|
||||
}
|
||||
Reference in New Issue
Block a user