每日清理
This commit is contained in:
@@ -13,75 +13,108 @@ import (
|
||||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||||
)
|
||||
|
||||
// Source 表示受在线留存边界约束的数据源。
|
||||
type Source string
|
||||
|
||||
const (
|
||||
// SourceAudit 表示统一审计事件。
|
||||
SourceAudit Source = constants.AuditArchiveSource
|
||||
// SourceIntegration 表示外部交互日志。
|
||||
SourceAudit Source = constants.AuditArchiveSource
|
||||
SourceIntegration Source = constants.IntegrationArchiveSource
|
||||
)
|
||||
|
||||
// Info 是查询响应公开的在线留存边界。
|
||||
type Info struct {
|
||||
OnlineFrom time.Time `json:"online_from" description:"当前可在线查询的最早时间"`
|
||||
ArchivedBefore *time.Time `json:"archived_before" description:"早于该时间的数据已归档；尚未清理时为空"`
|
||||
Timezone string `json:"timezone" description:"留存自然日时区"`
|
||||
}
|
||||
|
||||
// Load 从归档账本读取已完成物理清理的数据边界。
|
||||
// Load 仅公开从最早在线日期开始连续完成物理清理的边界，绝不跨越清理空洞。
|
||||
func Load(ctx context.Context, db *gorm.DB, sources ...Source) (Info, error) {
|
||||
location, err := time.LoadLocation(constants.AuditArchiveTimezone)
|
||||
if err != nil {
|
||||
return Info{}, errors.Wrap(errors.CodeInternalError, err, "加载审计留存时区失败")
|
||||
}
|
||||
now := time.Now().In(location)
|
||||
info := Info{OnlineFrom: time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, location), Timezone: constants.AuditArchiveTimezone}
|
||||
boundaries := make([]sourceRetention, 0, len(sources))
|
||||
for _, source := range sources {
|
||||
boundary, cleaned, err := sourceBoundary(ctx, db, source, location)
|
||||
boundary, err := sourceBoundary(ctx, db, source, location)
|
||||
if err != nil {
|
||||
return Info{}, err
|
||||
}
|
||||
if boundary.Before(info.OnlineFrom) && info.ArchivedBefore == nil {
|
||||
info.OnlineFrom = boundary
|
||||
boundaries = append(boundaries, boundary)
|
||||
}
|
||||
now := time.Now().In(location)
|
||||
fallback := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, location)
|
||||
info := Info{OnlineFrom: fallback, Timezone: constants.AuditArchiveTimezone}
|
||||
if len(boundaries) == 0 {
|
||||
return info, nil
|
||||
}
|
||||
for _, boundary := range boundaries {
|
||||
if boundary.onlineFrom.Before(info.OnlineFrom) {
|
||||
info.OnlineFrom = boundary.onlineFrom
|
||||
}
|
||||
if cleaned && (info.ArchivedBefore == nil || boundary.After(*info.ArchivedBefore)) {
|
||||
value := boundary
|
||||
info.ArchivedBefore = &value
|
||||
info.OnlineFrom = boundary
|
||||
}
|
||||
if len(boundaries) == 1 && boundaries[0].cleaned {
|
||||
value := boundaries[0].onlineFrom
|
||||
info.ArchivedBefore = &value
|
||||
return info, nil
|
||||
}
|
||||
if len(boundaries) > 1 {
|
||||
common := boundaries[0].onlineFrom
|
||||
allCleaned := boundaries[0].cleaned
|
||||
for _, boundary := range boundaries[1:] {
|
||||
if boundary.onlineFrom.Before(common) {
|
||||
common = boundary.onlineFrom
|
||||
}
|
||||
allCleaned = allCleaned && boundary.cleaned
|
||||
}
|
||||
if allCleaned {
|
||||
info.OnlineFrom = common
|
||||
info.ArchivedBefore = &common
|
||||
}
|
||||
}
|
||||
return info, nil
|
||||
}
|
||||
|
||||
func sourceBoundary(ctx context.Context, db *gorm.DB, source Source, location *time.Location) (time.Time, bool, error) {
|
||||
var cleanedEnd sql.NullTime
|
||||
if err := db.WithContext(ctx).Model(&model.LogArchiveRun{}).
|
||||
Where("source = ? AND cleaned_at IS NOT NULL", source).
|
||||
Select("MAX(range_end)").Scan(&cleanedEnd).Error; err != nil {
|
||||
return time.Time{}, false, errors.Wrap(errors.CodeDatabaseError, err, "查询审计留存清理边界失败")
|
||||
}
|
||||
if cleanedEnd.Valid {
|
||||
return cleanedEnd.Time.In(location), true, nil
|
||||
}
|
||||
type sourceRetention struct {
|
||||
onlineFrom time.Time
|
||||
cleaned bool
|
||||
}
|
||||
|
||||
var earliest sql.NullTime
|
||||
table, column := "tb_audit_event", "occurred_at"
|
||||
func sourceBoundary(ctx context.Context, db *gorm.DB, source Source, location *time.Location) (sourceRetention, error) {
|
||||
table, column := "tb_audit_event", "created_at"
|
||||
if source == SourceIntegration {
|
||||
table, column = "tb_integration_log", "created_at"
|
||||
}
|
||||
if err := db.WithContext(ctx).Table(table).Select("MIN(" + column + ")").Scan(&earliest).Error; err != nil {
|
||||
return time.Time{}, false, errors.Wrap(errors.CodeDatabaseError, err, "查询审计在线数据边界失败")
|
||||
var earliestOnline, earliestLedger sql.NullTime
|
||||
if err := db.WithContext(ctx).Table(table).Select("MIN(" + column + ")").Scan(&earliestOnline).Error; err != nil {
|
||||
return sourceRetention{}, errors.Wrap(errors.CodeDatabaseError, err, "查询审计在线数据边界失败")
|
||||
}
|
||||
if earliest.Valid {
|
||||
return earliest.Time.In(location), false, nil
|
||||
if err := db.WithContext(ctx).Model(&model.LogArchiveRun{}).Where("source = ?", source).Select("MIN(archive_date)").Scan(&earliestLedger).Error; err != nil {
|
||||
return sourceRetention{}, errors.Wrap(errors.CodeDatabaseError, err, "查询审计留存账本边界失败")
|
||||
}
|
||||
now := time.Now().In(location)
|
||||
return time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, location), false, nil
|
||||
if !earliestOnline.Valid && !earliestLedger.Valid {
|
||||
now := time.Now().In(location)
|
||||
return sourceRetention{onlineFrom: time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, location)}, nil
|
||||
}
|
||||
start := earliestLedger.Time.In(location)
|
||||
if earliestOnline.Valid && (!earliestLedger.Valid || earliestOnline.Time.Before(earliestLedger.Time)) {
|
||||
start = earliestOnline.Time.In(location)
|
||||
}
|
||||
start = time.Date(start.Year(), start.Month(), start.Day(), 0, 0, 0, 0, location)
|
||||
var rows []model.LogArchiveRun
|
||||
if err := db.WithContext(ctx).Where("source = ? AND archive_date >= ?", source, start.Format(time.DateOnly)).Order("archive_date ASC").Find(&rows).Error; err != nil {
|
||||
return sourceRetention{}, errors.Wrap(errors.CodeDatabaseError, err, "查询审计留存清理边界失败")
|
||||
}
|
||||
expected := start
|
||||
for _, row := range rows {
|
||||
date := row.ArchiveDate
|
||||
date = time.Date(date.Year(), date.Month(), date.Day(), 0, 0, 0, 0, location)
|
||||
if !date.Equal(expected) || row.CleanedAt == nil {
|
||||
break
|
||||
}
|
||||
expected = expected.AddDate(0, 0, 1)
|
||||
}
|
||||
return sourceRetention{onlineFrom: expected, cleaned: !expected.Equal(start)}, nil
|
||||
}
|
||||
|
||||
// NormalizeRange 将缺省范围收敛到在线窗口，并拒绝归档或跨边界查询。
|
||||
func NormalizeRange(info Info, from, to *time.Time, maxRange ...time.Duration) (*time.Time, *time.Time, error) {
|
||||
explicitFrom := from != nil
|
||||
if from != nil && info.ArchivedBefore != nil && from.Before(info.OnlineFrom) {
|
||||
@@ -110,7 +143,6 @@ func NormalizeRange(info Info, from, to *time.Time, maxRange ...time.Duration) (
|
||||
}
|
||||
return from, to, nil
|
||||
}
|
||||
|
||||
func archivedError(info Info) error {
|
||||
return errors.NewWithData(errors.CodeAuditDataArchived, map[string]any{"retention": info})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user