junhong_cmp_fiber/internal/query/retention/retention.go

// Package retention 提供在线审计查询的统一留存边界。
package retention

import (
	"context"
	"database/sql"
	"time"

	"gorm.io/gorm"

	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/errors"
)

type Source string

const (
	SourceAudit       Source = constants.AuditArchiveSource
	SourceIntegration Source = constants.IntegrationArchiveSource
)

type Info struct {
	OnlineFrom     time.Time  `json:"online_from" description:"当前可在线查询的最早时间"`
	ArchivedBefore *time.Time `json:"archived_before" description:"早于该时间的数据已归档；尚未清理时为空"`
	Timezone       string     `json:"timezone" description:"留存自然日时区"`
}

// 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, "加载审计留存时区失败")
	}
	boundaries := make([]sourceRetention, 0, len(sources))
	for _, source := range sources {
		boundary, err := sourceBoundary(ctx, db, source, location)
		if err != nil {
			return Info{}, err
		}
		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 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
}

type sourceRetention struct {
	onlineFrom time.Time
	cleaned    bool
}

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"
	}
	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 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, "查询审计留存账本边界失败")
	}
	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
}

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) {
		return nil, nil, archivedError(info)
	}
	if to != nil && info.ArchivedBefore != nil && !to.After(info.OnlineFrom) {
		return nil, nil, archivedError(info)
	}
	if from == nil {
		value := info.OnlineFrom
		from = &value
	}
	if to == nil {
		value := time.Now()
		to = &value
	}
	if len(maxRange) > 0 && maxRange[0] > 0 && to.Sub(*from) > maxRange[0] {
		if explicitFrom {
			return nil, nil, errors.New(errors.CodeInvalidParam)
		}
		value := to.Add(-maxRange[0])
		from = &value
	}
	if !from.Before(*to) {
		return nil, nil, errors.New(errors.CodeInvalidParam)
	}
	return from, to, nil
}
func archivedError(info Info) error {
	return errors.NewWithData(errors.CodeAuditDataArchived, map[string]any{"retention": info})
}