package auditarchive import ( "context" "crypto/sha256" "fmt" "io" "os" "strconv" "strings" "time" "github.com/bytedance/sonic" "gorm.io/gorm" "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/pkg/constants" ) const maxManifestBytes = 1024 * 1024 // RetentionAudit 描述月度物理清理的统一审计事实。 type RetentionAudit struct { EventID string Month string Summary string Result string ErrorSummary string RangeStart time.Time RangeEnd time.Time EventCount int64 ResourceCount int64 IntegrationCount int64 ManifestKeys []string DurationMS int64 } // RetentionResult 是月度留存清理的结构化执行结果。 type RetentionResult struct { Month string EventCount int64 ResourceCount int64 IntegrationCount int64 EstimatedBatches int64 ManifestKeys []string Duration time.Duration } type retentionRuns struct { audit []*model.LogArchiveRun integration []*model.LogArchiveRun } // CleanupPreviousMonth 校验并物理清理上一个完整自然月的在线审计日志。 func (s *Service) CleanupPreviousMonth(ctx context.Context) (RetentionResult, error) { now := time.Now().In(s.location) return s.CleanupMonth(ctx, now.AddDate(0, -1, 0)) } // ValidatePreviousMonth 只读校验上一个完整自然月的归档与清理门禁。 func (s *Service) ValidatePreviousMonth(ctx context.Context) (RetentionResult, error) { now := time.Now().In(s.location) return s.ValidateMonth(ctx, now.AddDate(0, -1, 0)) } // ValidateMonth 只读校验指定完整自然月,不写清理断点且不删除在线数据。 func (s *Service) ValidateMonth(ctx context.Context, month time.Time) (result RetentionResult, err error) { if s.db == nil || s.store == nil { return result, fmt.Errorf("日志留存演练数据库或对象存储未配置") } start, end, err := s.retentionMonthRange(month) if err != nil { return result, err } startedAt := time.Now() result.Month = start.Format("2006-01") runs, err := s.loadRetentionRuns(ctx, start, end) if err != nil { return result, err } if err := s.validateRetentionRuns(ctx, start, end, runs); err != nil { return result, err } summarizeRetentionRuns(runs, &result) result.EstimatedBatches = estimatedRetentionBatches(result) result.Duration = time.Since(startedAt) return result, nil } // CleanupMonth 校验归档硬门禁后按固定顺序物理清理指定完整自然月。 func (s *Service) CleanupMonth(ctx context.Context, month time.Time) (result RetentionResult, cleanupErr error) { if s.db == nil || s.store == nil || s.audit == nil { return result, fmt.Errorf("日志留存清理数据库、对象存储或审计 Writer 未配置") } start, end, err := s.retentionMonthRange(month) if err != nil { return result, err } startedAt := time.Now() result.Month = start.Format("2006-01") cleanupErr = s.executeRetention(ctx, start, end, &result) result.Duration = time.Since(startedAt) if auditErr := s.recordRetentionAudit(ctx, start, end, result, cleanupErr); auditErr != nil { if cleanupErr != nil { return result, fmt.Errorf("%w;记录留存清理失败审计失败: %v", cleanupErr, auditErr) } return result, fmt.Errorf("记录留存清理成功审计失败: %w", auditErr) } return result, cleanupErr } func (s *Service) retentionMonthRange(month time.Time) (time.Time, time.Time, error) { start := time.Date(month.In(s.location).Year(), month.In(s.location).Month(), 1, 0, 0, 0, 0, s.location) end := start.AddDate(0, 1, 0) now := time.Now().In(s.location) currentMonth := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, s.location) if !end.Before(currentMonth) && !end.Equal(currentMonth) { return time.Time{}, time.Time{}, fmt.Errorf("只能清理已经结束的完整自然月") } return start, end, nil } func (s *Service) executeRetention(ctx context.Context, start, end time.Time, result *RetentionResult) error { started, err := s.retentionCleanupStarted(ctx, start, end) if err != nil { return err } if !started { lastDay := end.AddDate(0, 0, -1) if err := s.ArchiveDate(ctx, lastDay); err != nil { return fmt.Errorf("完成上月最后一天 Audit 归档失败: %w", err) } if err := s.ArchiveIntegrationDate(ctx, lastDay); err != nil { return fmt.Errorf("完成上月最后一天 Integration Log 归档失败: %w", err) } if err := s.FinalizeIntegrationMonth(ctx, start); err != nil { return err } } runs, err := s.loadRetentionRuns(ctx, start, end) if err != nil { return err } if err := s.validateRetentionRuns(ctx, start, end, runs); err != nil { return err } summarizeRetentionRuns(runs, result) if err := s.cleanupAuditMonth(ctx, start, end, runs.audit); err != nil { return err } return s.cleanupIntegrationMonth(ctx, start, end, runs.integration) } func (s *Service) retentionCleanupStarted(ctx context.Context, start, end time.Time) (bool, error) { var count int64 err := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}). Where("archive_date >= ? AND archive_date < ? AND instance_id = ? AND cleanup_started_at IS NOT NULL", start.Format(time.DateOnly), end.Format(time.DateOnly), s.instanceID). Count(&count).Error if err != nil { return false, fmt.Errorf("读取月度清理断点失败: %w", err) } return count > 0, nil } func (s *Service) loadRetentionRuns(ctx context.Context, start, end time.Time) (retentionRuns, error) { var rows []model.LogArchiveRun err := s.db.WithContext(ctx).Where( "source IN ? AND archive_date >= ? AND archive_date < ? AND instance_id = ?", []string{constants.AuditArchiveSource, constants.IntegrationArchiveSource}, start.Format(time.DateOnly), end.Format(time.DateOnly), s.instanceID, ).Order("archive_date ASC, source ASC").Find(&rows).Error if err != nil { return retentionRuns{}, fmt.Errorf("读取月度归档账本失败: %w", err) } days := int(end.Sub(start).Hours() / 24) if len(rows) != days*2 { return retentionRuns{}, fmt.Errorf("月度归档账本缺日:期望 %d 条,实际 %d 条", days*2, len(rows)) } runs := retentionRuns{audit: make([]*model.LogArchiveRun, 0, days), integration: make([]*model.LogArchiveRun, 0, days)} for index := range rows { run := &rows[index] switch run.Source { case constants.AuditArchiveSource: runs.audit = append(runs.audit, run) case constants.IntegrationArchiveSource: runs.integration = append(runs.integration, run) } } if len(runs.audit) != days || len(runs.integration) != days { return retentionRuns{}, fmt.Errorf("月度 Audit 或 Integration 归档账本不完整") } return runs, nil } func (s *Service) validateRetentionRuns(ctx context.Context, start, end time.Time, runs retentionRuns) error { if err := validateCleanupLedgerState(runs.audit); err != nil { return fmt.Errorf("Audit 清理断点非法: %w", err) } if err := validateCleanupLedgerState(runs.integration); err != nil { return fmt.Errorf("Integration 清理断点非法: %w", err) } for index := range runs.audit { date := start.AddDate(0, 0, index) if err := s.validateAuditRetentionDay(ctx, date, runs.audit[index]); err != nil { return fmt.Errorf("%s Audit 清理门禁失败: %w", date.Format(time.DateOnly), err) } if err := s.validateIntegrationRetentionDay(ctx, date, runs.integration[index]); err != nil { return fmt.Errorf("%s Integration 清理门禁失败: %w", date.Format(time.DateOnly), err) } } return nil } func validateCleanupLedgerState(runs []*model.LogArchiveRun) error { started, cleaned := 0, 0 for _, run := range runs { if run.CleanupStartedAt != nil { started++ } if run.CleanedAt != nil { cleaned++ } } if started != 0 && started != len(runs) { return fmt.Errorf("清理开始断点不是整月原子状态") } if cleaned != 0 && cleaned != len(runs) { return fmt.Errorf("清理完成断点不是整月原子状态") } if cleaned > 0 && started == 0 { return fmt.Errorf("清理完成但缺少开始断点") } return nil } func (s *Service) validateAuditRetentionDay(ctx context.Context, date time.Time, run *model.LogArchiveRun) error { if err := validateRunBase(run, date, constants.AuditArchiveSchemaVersion, false); err != nil { return err } if err := s.validateAuditManifest(ctx, run); err != nil { return err } events, resources, err := s.databaseCounts(ctx, run.RangeStart, run.RangeEnd) if err != nil { return err } return validateRemainingCounts(run, events, resources) } func (s *Service) validateIntegrationRetentionDay(ctx context.Context, date time.Time, run *model.LogArchiveRun) error { if err := validateRunBase(run, date, constants.IntegrationArchiveSchemaVersion, true); err != nil { return err } if err := s.validateIntegrationManifest(ctx, run); err != nil { return err } count, err := s.integrationRecordCount(ctx, run.RangeStart, run.RangeEnd) if err != nil { return err } if run.CleanedAt != nil { if count != 0 { return fmt.Errorf("已标记清理完成但数据库仍有 %d 条记录", count) } return nil } if run.CleanupStartedAt != nil { if count > run.RecordCount { return fmt.Errorf("续跑窗口记录数超过最终归档数量") } return nil } file, err := s.buildIntegrationArchiveFile(ctx, run.RangeStart, run.RangeEnd) if err != nil { return err } defer os.Remove(file.path) if file.recordCount != run.RecordCount || file.sha256 != run.SHA256 { return fmt.Errorf("数据库当前 Integration 内容与最终 revision 不一致") } return nil } func validateRunBase(run *model.LogArchiveRun, date time.Time, schema string, final bool) error { if run.Status != constants.ArchiveStatusSuccess || run.SchemaVersion != schema { return fmt.Errorf("归档状态或 schema version 不符合清理要求") } if run.ArchiveDate.Format(time.DateOnly) != date.Format(time.DateOnly) || !run.RangeStart.Equal(date) || !run.RangeEnd.Equal(date.AddDate(0, 0, 1)) { return fmt.Errorf("归档日期或半开时间范围不一致") } if final && !run.IsFinal { return fmt.Errorf("Integration 最终 revision 尚未形成") } if run.ObjectKey == "" || run.ManifestKey == "" || run.SHA256 == "" { return fmt.Errorf("归档对象、manifest 或 SHA-256 缺失") } return nil } func validateRemainingCounts(run *model.LogArchiveRun, events, resources int64) error { if run.CleanedAt != nil { if events != 0 || resources != 0 { return fmt.Errorf("已标记清理完成但数据库仍有事件或资源") } return nil } if run.CleanupStartedAt != nil { if events > run.EventCount || resources > run.ResourceCount { return fmt.Errorf("续跑窗口数量超过已归档数量") } return nil } if events != run.EventCount || resources != run.ResourceCount { return fmt.Errorf("数据库事件或资源数量与 manifest 不一致") } return nil } func (s *Service) validateAuditManifest(ctx context.Context, run *model.LogArchiveRun) error { var manifest Manifest if err := s.readManifest(ctx, run.ManifestKey, &manifest); err != nil { return err } if manifest.Source != run.Source || manifest.SchemaVersion != run.SchemaVersion || manifest.Status != constants.ArchiveStatusSuccess || manifest.ArchiveDate != run.ArchiveDate.Format(time.DateOnly) || manifest.Timezone != constants.AuditArchiveTimezone || manifest.InstanceID != run.InstanceID || !manifest.RangeStart.Equal(run.RangeStart) || !manifest.RangeEnd.Equal(run.RangeEnd) || manifest.EventCount != run.EventCount || manifest.ResourceCount != run.ResourceCount || manifest.CompressedBytes != run.CompressedBytes || manifest.ObjectKey != run.ObjectKey || manifest.SHA256 != run.SHA256 || manifest.Revision != run.Revision { return fmt.Errorf("Audit manifest 与 ledger 不一致") } if err := s.verifyObject(ctx, run.ManifestKey, -1, map[string]string{ "source": constants.AuditArchiveSource, "data-sha256": run.SHA256, "revision": strconv.Itoa(run.Revision), }); err != nil { return err } return s.verifyRetentionObject(ctx, run, false) } func (s *Service) validateIntegrationManifest(ctx context.Context, run *model.LogArchiveRun) error { var manifest integrationArchiveManifest if err := s.readManifest(ctx, run.ManifestKey, &manifest); err != nil { return err } if manifest.Source != run.Source || manifest.SchemaVersion != run.SchemaVersion || manifest.Status != constants.ArchiveStatusSuccess || !manifest.Final || manifest.ArchiveDate != run.ArchiveDate.Format(time.DateOnly) || manifest.Timezone != constants.AuditArchiveTimezone || manifest.InstanceID != run.InstanceID || !manifest.RangeStart.Equal(run.RangeStart) || !manifest.RangeEnd.Equal(run.RangeEnd) || manifest.RecordCount != run.RecordCount || manifest.CompressedBytes != run.CompressedBytes || manifest.ObjectKey != run.ObjectKey || manifest.SHA256 != run.SHA256 || manifest.Revision != run.Revision { return fmt.Errorf("Integration manifest 与最终 ledger 不一致") } if err := s.verifyObject(ctx, run.ManifestKey, -1, map[string]string{ "source": constants.IntegrationArchiveSource, "data-sha256": run.SHA256, "revision": strconv.Itoa(run.Revision), "final": "true", }); err != nil { return err } return s.verifyRetentionObject(ctx, run, true) } func (s *Service) readManifest(ctx context.Context, key string, target any) error { object, err := s.store.Stat(ctx, key) if err != nil { return fmt.Errorf("读取 manifest metadata 失败: %w", err) } reader, err := s.store.Download(ctx, key) if err != nil { return fmt.Errorf("下载 manifest 失败: %w", err) } data, readErr := io.ReadAll(io.LimitReader(reader, maxManifestBytes+1)) closeErr := reader.Close() if readErr != nil { return fmt.Errorf("读取 manifest 失败: %w", readErr) } if closeErr != nil { return fmt.Errorf("关闭 manifest 对象失败: %w", closeErr) } if len(data) > maxManifestBytes || int64(len(data)) != object.Size { return fmt.Errorf("manifest 大小非法或不完整") } if err := sonic.Unmarshal(data, target); err != nil { return fmt.Errorf("解析 manifest 失败: %w", err) } return nil } func (s *Service) verifyRetentionObject(ctx context.Context, run *model.LogArchiveRun, final bool) error { metadata := map[string]string{ "schema-version": run.SchemaVersion, "source": run.Source, "archive-date": run.RangeStart.Format(time.DateOnly), "timezone": constants.AuditArchiveTimezone, "sha256": run.SHA256, "revision": strconv.Itoa(run.Revision), } if run.Source == constants.AuditArchiveSource { metadata["event-count"] = strconv.FormatInt(run.EventCount, 10) metadata["resource-count"] = strconv.FormatInt(run.ResourceCount, 10) } else { metadata["record-count"] = strconv.FormatInt(run.RecordCount, 10) metadata["final"] = strconv.FormatBool(final) } if err := s.verifyObject(ctx, run.ObjectKey, run.CompressedBytes, metadata); err != nil { return err } reader, err := s.store.Download(ctx, run.ObjectKey) if err != nil { return fmt.Errorf("下载归档对象复核 SHA-256 失败: %w", err) } hasher := sha256.New() written, copyErr := io.Copy(hasher, reader) closeErr := reader.Close() if copyErr != nil { return fmt.Errorf("读取归档对象复核 SHA-256 失败: %w", copyErr) } if closeErr != nil { return fmt.Errorf("关闭归档对象失败: %w", closeErr) } if written != run.CompressedBytes || fmt.Sprintf("%x", hasher.Sum(nil)) != run.SHA256 { return fmt.Errorf("归档对象大小或 SHA-256 复核失败") } return nil } func summarizeRetentionRuns(runs retentionRuns, result *RetentionResult) { result.ManifestKeys = make([]string, 0, len(runs.audit)+len(runs.integration)) for _, run := range runs.audit { result.EventCount += run.EventCount result.ResourceCount += run.ResourceCount result.ManifestKeys = append(result.ManifestKeys, run.ManifestKey) } for _, run := range runs.integration { result.IntegrationCount += run.RecordCount result.ManifestKeys = append(result.ManifestKeys, run.ManifestKey) } } func estimatedRetentionBatches(result RetentionResult) int64 { batchSize := int64(constants.AuditRetentionDeleteBatchSize) return (result.EventCount+batchSize-1)/batchSize + (result.ResourceCount+batchSize-1)/batchSize + (result.IntegrationCount+batchSize-1)/batchSize } func (s *Service) cleanupAuditMonth(ctx context.Context, start, end time.Time, runs []*model.LogArchiveRun) error { if allRunsCleaned(runs) { return nil } if err := s.markCleanupStarted(ctx, constants.AuditArchiveSource, start, end); err != nil { return err } if err := s.deleteAuditResources(ctx, start, end); err != nil { return err } if err := s.deleteAuditEvents(ctx, start, end); err != nil { return err } return s.markCleaned(ctx, constants.AuditArchiveSource, start, end) } func (s *Service) cleanupIntegrationMonth(ctx context.Context, start, end time.Time, runs []*model.LogArchiveRun) error { if allRunsCleaned(runs) { return nil } if err := s.markCleanupStarted(ctx, constants.IntegrationArchiveSource, start, end); err != nil { return err } for { subquery := s.db.Model(&model.IntegrationLog{}).Select("id"). Where("created_at >= ? AND created_at < ?", start, end).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize) deleted := s.db.WithContext(ctx).Where("id IN (?)", subquery).Delete(&model.IntegrationLog{}) if deleted.Error != nil { return fmt.Errorf("分批物理删除 Integration Log 失败: %w", deleted.Error) } if deleted.RowsAffected == 0 { break } } return s.markCleaned(ctx, constants.IntegrationArchiveSource, start, end) } func (s *Service) deleteAuditResources(ctx context.Context, start, end time.Time) error { for { subquery := s.db.Model(&model.AuditEventResource{}).Select("tb_audit_event_resource.id"). Joins("JOIN tb_audit_event ON tb_audit_event.id = tb_audit_event_resource.audit_event_id"). Where("tb_audit_event.created_at >= ? AND tb_audit_event.created_at < ?", start, end). Order("tb_audit_event_resource.id ASC").Limit(constants.AuditRetentionDeleteBatchSize) deleted := s.db.WithContext(ctx).Where("id IN (?)", subquery).Delete(&model.AuditEventResource{}) if deleted.Error != nil { return fmt.Errorf("分批物理删除 Audit Event Resource 失败: %w", deleted.Error) } if deleted.RowsAffected == 0 { return nil } } } func (s *Service) deleteAuditEvents(ctx context.Context, start, end time.Time) error { for { subquery := s.db.Model(&model.AuditEvent{}).Select("id"). Where("created_at >= ? AND created_at < ?", start, end).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize) deleted := s.db.WithContext(ctx).Where("id IN (?)", subquery).Delete(&model.AuditEvent{}) if deleted.Error != nil { return fmt.Errorf("分批物理删除 Audit Event 失败: %w", deleted.Error) } if deleted.RowsAffected == 0 { return nil } } } func (s *Service) markCleanupStarted(ctx context.Context, source string, start, end time.Time) error { now := time.Now() result := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}). Where("source = ? AND archive_date >= ? AND archive_date < ? AND instance_id = ? AND cleanup_started_at IS NULL", source, start.Format(time.DateOnly), end.Format(time.DateOnly), s.instanceID). Updates(map[string]any{"cleanup_started_at": now, "updated_at": now}) if result.Error != nil { return fmt.Errorf("记录月度清理开始断点失败: %w", result.Error) } return s.validateCleanupMarkerCount(ctx, source, start, end, "cleanup_started_at IS NOT NULL", "开始") } func (s *Service) markCleaned(ctx context.Context, source string, start, end time.Time) error { now := time.Now() result := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}). Where("source = ? AND archive_date >= ? AND archive_date < ? AND instance_id = ? AND cleanup_started_at IS NOT NULL", source, start.Format(time.DateOnly), end.Format(time.DateOnly), s.instanceID). Updates(map[string]any{"cleaned_at": now, "updated_at": now}) if result.Error != nil { return fmt.Errorf("记录月度清理完成断点失败: %w", result.Error) } return s.validateCleanupMarkerCount(ctx, source, start, end, "cleaned_at IS NOT NULL", "完成") } func (s *Service) validateCleanupMarkerCount(ctx context.Context, source string, start, end time.Time, marker, label string) error { var count int64 err := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}). Where("source = ? AND archive_date >= ? AND archive_date < ? AND instance_id = ? AND "+marker, source, start.Format(time.DateOnly), end.Format(time.DateOnly), s.instanceID). Count(&count).Error if err != nil { return fmt.Errorf("复核月度清理%s断点失败: %w", label, err) } expected := int64(end.Sub(start).Hours() / 24) if count != expected { return fmt.Errorf("月度清理%s断点不完整:期望 %d 条,实际 %d 条", label, expected, count) } return nil } func allRunsCleaned(runs []*model.LogArchiveRun) bool { return len(runs) > 0 && runs[0].CleanedAt != nil } func (s *Service) recordRetentionAudit(ctx context.Context, start, end time.Time, result RetentionResult, cleanupErr error) error { audit := RetentionAudit{ Month: result.Month, RangeStart: start, RangeEnd: end, EventCount: result.EventCount, ResourceCount: result.ResourceCount, IntegrationCount: result.IntegrationCount, ManifestKeys: result.ManifestKeys, DurationMS: result.Duration.Milliseconds(), Result: constants.AuditResultSuccess, Summary: "完成已归档在线日志月度物理清理", EventID: "evt_retention_" + strings.ReplaceAll(result.Month, "-", "_"), } if cleanupErr != nil { audit.EventID = "" audit.Result = constants.AuditResultFailed audit.Summary = "已归档在线日志月度物理清理失败" audit.ErrorSummary = truncateRetentionError(cleanupErr) } return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { return s.audit.WriteRetentionCleanup(ctx, tx, audit) }) } func truncateRetentionError(err error) string { value := []rune(err.Error()) if len(value) > 500 { value = value[:500] } return string(value) }