This commit is contained in:
350
internal/application/auditarchive/integration.go
Normal file
350
internal/application/auditarchive/integration.go
Normal file
@@ -0,0 +1,350 @@
|
||||
package auditarchive
|
||||
|
||||
import (
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/bytedance/sonic"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||||
)
|
||||
|
||||
type integrationArchiveFile struct {
|
||||
path string
|
||||
recordCount int64
|
||||
uncompressedBytes int64
|
||||
compressedBytes int64
|
||||
sha256 string
|
||||
}
|
||||
|
||||
type integrationArchiveManifest struct {
|
||||
SchemaVersion string `json:"schema_version"`
|
||||
Source string `json:"source"`
|
||||
ArchiveDate string `json:"archive_date"`
|
||||
Timezone string `json:"timezone"`
|
||||
RangeStart time.Time `json:"range_start"`
|
||||
RangeEnd time.Time `json:"range_end"`
|
||||
InstanceID string `json:"instance_id"`
|
||||
RecordCount int64 `json:"record_count"`
|
||||
UncompressedBytes int64 `json:"uncompressed_bytes"`
|
||||
CompressedBytes int64 `json:"compressed_bytes"`
|
||||
ObjectKey string `json:"object_key"`
|
||||
SHA256 string `json:"sha256"`
|
||||
Revision int `json:"revision"`
|
||||
GeneratedAt time.Time `json:"generated_at"`
|
||||
Status string `json:"status"`
|
||||
Final bool `json:"final"`
|
||||
}
|
||||
|
||||
// ArchivePreviousIntegrationDay 归档 Asia/Shanghai 前一完整自然日的 Integration Log 创建日快照。
|
||||
func (s *Service) ArchivePreviousIntegrationDay(ctx context.Context) error {
|
||||
now := time.Now().In(s.location)
|
||||
return s.ArchiveIntegrationDate(ctx, now.AddDate(0, 0, -1))
|
||||
}
|
||||
|
||||
// ArchiveIntegrationDate 归档指定 Asia/Shanghai 自然日的 Integration Log 创建日快照。
|
||||
func (s *Service) ArchiveIntegrationDate(ctx context.Context, archiveDate time.Time) error {
|
||||
return s.archiveIntegrationDate(ctx, archiveDate, false)
|
||||
}
|
||||
|
||||
// FinalizePreviousIntegrationMonth 复核并终结上一个完整自然月的 Integration Log 归档。
|
||||
func (s *Service) FinalizePreviousIntegrationMonth(ctx context.Context) error {
|
||||
now := time.Now().In(s.location)
|
||||
return s.FinalizeIntegrationMonth(ctx, now.AddDate(0, -1, 0))
|
||||
}
|
||||
|
||||
// FinalizeIntegrationMonth 逐日复核指定完整自然月，并为变化内容创建最终 revision。
|
||||
func (s *Service) FinalizeIntegrationMonth(ctx context.Context, month time.Time) error {
|
||||
monthStart := time.Date(month.In(s.location).Year(), month.In(s.location).Month(), 1, 0, 0, 0, 0, s.location)
|
||||
currentMonth := time.Now().In(s.location)
|
||||
currentMonthStart := time.Date(currentMonth.Year(), currentMonth.Month(), 1, 0, 0, 0, 0, s.location)
|
||||
if !monthStart.Before(currentMonthStart) {
|
||||
return fmt.Errorf("只能终结已经结束的 Integration Log 完整自然月")
|
||||
}
|
||||
for date := monthStart; date.Before(monthStart.AddDate(0, 1, 0)); date = date.AddDate(0, 0, 1) {
|
||||
if err := s.archiveIntegrationDate(ctx, date, true); err != nil {
|
||||
return fmt.Errorf("终结 %s Integration Log 归档失败: %w", date.Format(time.DateOnly), err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) archiveIntegrationDate(ctx context.Context, archiveDate time.Time, final bool) error {
|
||||
if s.db == nil || s.store == nil {
|
||||
return fmt.Errorf("Integration Log 归档数据库或对象存储未配置")
|
||||
}
|
||||
start := time.Date(archiveDate.In(s.location).Year(), archiveDate.In(s.location).Month(), archiveDate.In(s.location).Day(), 0, 0, 0, 0, s.location)
|
||||
end := start.AddDate(0, 0, 1)
|
||||
today := time.Now().In(s.location)
|
||||
if end.After(time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, s.location)) {
|
||||
return fmt.Errorf("Integration Log 只能归档已经结束的完整自然日")
|
||||
}
|
||||
|
||||
run, err := s.ensureIntegrationRun(ctx, start, end)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
file, err := s.buildIntegrationArchiveFile(ctx, start, end)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer os.Remove(file.path)
|
||||
if final {
|
||||
pending, pendingErr := s.integrationPendingCount(ctx, start, end)
|
||||
if pendingErr != nil {
|
||||
return pendingErr
|
||||
}
|
||||
if pending > 0 {
|
||||
_ = s.db.WithContext(ctx).Model(&model.LogArchiveRun{}).Where("id = ?", run.ID).
|
||||
Updates(map[string]any{"is_final": false, "error_summary": "存在 pending Integration Log，无法形成最终归档", "updated_at": time.Now()}).Error
|
||||
return fmt.Errorf("仍有 %d 条 pending Integration Log，无法形成最终归档", pending)
|
||||
}
|
||||
}
|
||||
if run.Status == constants.ArchiveStatusSuccess && run.RecordCount == file.recordCount && run.SHA256 == file.sha256 {
|
||||
valid, validateErr := s.validateIntegrationRun(ctx, run)
|
||||
if validateErr == nil && valid && (!final || run.IsFinal) {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
acquired, err := s.acquireIntegrationRun(ctx, run)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !acquired {
|
||||
return fmt.Errorf("Integration Log 归档任务正在执行")
|
||||
}
|
||||
if err := s.uploadIntegrationArchive(ctx, run, file, final); err != nil {
|
||||
s.markFailed(ctx, run.ID, err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) ensureIntegrationRun(ctx context.Context, start, end time.Time) (*model.LogArchiveRun, error) {
|
||||
run := model.LogArchiveRun{
|
||||
Source: constants.IntegrationArchiveSource, ArchiveDate: start, InstanceID: s.instanceID,
|
||||
SchemaVersion: constants.IntegrationArchiveSchemaVersion, Revision: 1,
|
||||
Status: constants.ArchiveStatusPending, RangeStart: start, RangeEnd: end,
|
||||
}
|
||||
result := s.db.WithContext(ctx).Clauses(clause.OnConflict{
|
||||
Columns: []clause.Column{{Name: "source"}, {Name: "archive_date"}, {Name: "instance_id"}, {Name: "schema_version"}},
|
||||
DoNothing: true,
|
||||
}).Create(&run)
|
||||
if result.Error != nil {
|
||||
return nil, fmt.Errorf("创建 Integration Log 归档账本失败: %w", result.Error)
|
||||
}
|
||||
if result.RowsAffected == 0 {
|
||||
if err := s.db.WithContext(ctx).Where(
|
||||
"source = ? AND archive_date = ? AND instance_id = ? AND schema_version = ?",
|
||||
constants.IntegrationArchiveSource, start, s.instanceID, constants.IntegrationArchiveSchemaVersion,
|
||||
).First(&run).Error; err != nil {
|
||||
return nil, fmt.Errorf("读取 Integration Log 归档账本失败: %w", err)
|
||||
}
|
||||
}
|
||||
return &run, nil
|
||||
}
|
||||
|
||||
func (s *Service) acquireIntegrationRun(ctx context.Context, run *model.LogArchiveRun) (bool, error) {
|
||||
revision := run.Revision
|
||||
if run.Status != constants.ArchiveStatusPending {
|
||||
revision++
|
||||
}
|
||||
now := time.Now()
|
||||
result := s.db.WithContext(ctx).Model(&model.LogArchiveRun{}).
|
||||
Where("id = ? AND (status <> ? OR updated_at < ?)", run.ID, constants.ArchiveStatusRunning, now.Add(-3*time.Hour)).
|
||||
Updates(map[string]any{
|
||||
"status": constants.ArchiveStatusRunning, "revision": revision, "is_final": false,
|
||||
"attempt_count": gorm.Expr("attempt_count + 1"), "error_summary": "", "completed_at": nil, "updated_at": now,
|
||||
})
|
||||
if result.Error != nil {
|
||||
return false, fmt.Errorf("锁定 Integration Log 归档任务失败: %w", result.Error)
|
||||
}
|
||||
if result.RowsAffected == 0 {
|
||||
return false, nil
|
||||
}
|
||||
run.Revision = revision
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (s *Service) buildIntegrationArchiveFile(ctx context.Context, start, end time.Time) (*integrationArchiveFile, error) {
|
||||
temp, err := os.CreateTemp("", "integration-logs-*.jsonl.gz")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("创建 Integration Log 归档临时文件失败: %w", err)
|
||||
}
|
||||
path := temp.Name()
|
||||
failed := true
|
||||
defer func() {
|
||||
_ = temp.Close()
|
||||
if failed {
|
||||
_ = os.Remove(path)
|
||||
}
|
||||
}()
|
||||
|
||||
hasher := sha256.New()
|
||||
gzipWriter := gzip.NewWriter(io.MultiWriter(temp, hasher))
|
||||
result := &integrationArchiveFile{path: path}
|
||||
var lastID uint
|
||||
for {
|
||||
var logs []model.IntegrationLog
|
||||
if err := s.db.WithContext(ctx).Where("created_at >= ? AND created_at < ? AND id > ?", start, end, lastID).
|
||||
Order("id ASC").Limit(archivePageSize).Find(&logs).Error; err != nil {
|
||||
return nil, fmt.Errorf("读取 Integration Log 归档记录失败: %w", err)
|
||||
}
|
||||
if len(logs) == 0 {
|
||||
break
|
||||
}
|
||||
for i := range logs {
|
||||
line, marshalErr := sonic.Marshal(logs[i])
|
||||
if marshalErr != nil {
|
||||
return nil, fmt.Errorf("序列化 Integration Log 归档记录失败: %w", marshalErr)
|
||||
}
|
||||
line = append(line, '\n')
|
||||
if _, writeErr := gzipWriter.Write(line); writeErr != nil {
|
||||
return nil, fmt.Errorf("写入 Integration Log 归档压缩流失败: %w", writeErr)
|
||||
}
|
||||
result.recordCount++
|
||||
result.uncompressedBytes += int64(len(line))
|
||||
}
|
||||
lastID = logs[len(logs)-1].ID
|
||||
}
|
||||
if err := gzipWriter.Close(); err != nil {
|
||||
return nil, fmt.Errorf("关闭 Integration Log 归档压缩流失败: %w", err)
|
||||
}
|
||||
if err := temp.Close(); err != nil {
|
||||
return nil, fmt.Errorf("关闭 Integration Log 归档临时文件失败: %w", err)
|
||||
}
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取 Integration Log 归档临时文件信息失败: %w", err)
|
||||
}
|
||||
result.compressedBytes = info.Size()
|
||||
result.sha256 = fmt.Sprintf("%x", hasher.Sum(nil))
|
||||
failed = false
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (s *Service) uploadIntegrationArchive(ctx context.Context, run *model.LogArchiveRun, file *integrationArchiveFile, final bool) error {
|
||||
count, err := s.integrationRecordCount(ctx, run.RangeStart, run.RangeEnd)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if count != file.recordCount {
|
||||
return fmt.Errorf("Integration Log 归档生成期间记录数量发生变化")
|
||||
}
|
||||
objectKey, manifestKey := integrationObjectKeys(run.RangeStart, run.Revision)
|
||||
metadata := integrationArchiveMetadata(file, run, final)
|
||||
reader, err := os.Open(file.path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("打开 Integration Log 归档临时文件失败: %w", err)
|
||||
}
|
||||
uploadErr := s.store.UploadWithMetadata(ctx, objectKey, reader, "application/gzip", metadata)
|
||||
closeErr := reader.Close()
|
||||
if uploadErr != nil {
|
||||
return fmt.Errorf("上传 Integration Log 归档对象失败: %w", uploadErr)
|
||||
}
|
||||
if closeErr != nil {
|
||||
return fmt.Errorf("关闭 Integration Log 归档临时文件失败: %w", closeErr)
|
||||
}
|
||||
if err := s.verifyObject(ctx, objectKey, file.compressedBytes, metadata); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
generatedAt := time.Now().In(s.location)
|
||||
manifest := integrationArchiveManifest{
|
||||
SchemaVersion: constants.IntegrationArchiveSchemaVersion, Source: constants.IntegrationArchiveSource,
|
||||
ArchiveDate: run.RangeStart.In(s.location).Format(time.DateOnly), Timezone: constants.AuditArchiveTimezone,
|
||||
RangeStart: run.RangeStart, RangeEnd: run.RangeEnd, InstanceID: s.instanceID,
|
||||
RecordCount: file.recordCount, UncompressedBytes: file.uncompressedBytes, CompressedBytes: file.compressedBytes,
|
||||
ObjectKey: objectKey, SHA256: file.sha256, Revision: run.Revision, GeneratedAt: generatedAt,
|
||||
Status: constants.ArchiveStatusSuccess, Final: final,
|
||||
}
|
||||
manifestBytes, err := sonic.Marshal(manifest)
|
||||
if err != nil {
|
||||
return fmt.Errorf("序列化 Integration Log 归档清单失败: %w", err)
|
||||
}
|
||||
manifestMetadata := map[string]string{
|
||||
"source": constants.IntegrationArchiveSource, "data-sha256": file.sha256,
|
||||
"revision": strconv.Itoa(run.Revision), "final": strconv.FormatBool(final),
|
||||
}
|
||||
if err := s.store.UploadWithMetadata(ctx, manifestKey, strings.NewReader(string(manifestBytes)), "application/json", manifestMetadata); err != nil {
|
||||
return fmt.Errorf("上传 Integration Log 归档清单失败: %w", err)
|
||||
}
|
||||
if err := s.verifyObject(ctx, manifestKey, int64(len(manifestBytes)), manifestMetadata); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
completedAt := time.Now()
|
||||
return s.db.WithContext(ctx).Model(&model.LogArchiveRun{}).Where("id = ?", run.ID).Updates(map[string]any{
|
||||
"status": constants.ArchiveStatusSuccess, "is_final": final,
|
||||
"object_key": objectKey, "manifest_key": manifestKey, "record_count": file.recordCount,
|
||||
"uncompressed_bytes": file.uncompressedBytes, "compressed_bytes": file.compressedBytes,
|
||||
"sha256": file.sha256, "generated_at": generatedAt, "completed_at": completedAt, "updated_at": completedAt,
|
||||
}).Error
|
||||
}
|
||||
|
||||
func (s *Service) validateIntegrationRun(ctx context.Context, run *model.LogArchiveRun) (bool, error) {
|
||||
metadata := map[string]string{
|
||||
"sha256": run.SHA256, "record-count": strconv.FormatInt(run.RecordCount, 10),
|
||||
"revision": strconv.Itoa(run.Revision), "final": strconv.FormatBool(run.IsFinal),
|
||||
}
|
||||
if err := s.verifyObject(ctx, run.ObjectKey, run.CompressedBytes, metadata); err != nil {
|
||||
return false, nil
|
||||
}
|
||||
manifestMetadata := map[string]string{
|
||||
"source": constants.IntegrationArchiveSource, "data-sha256": run.SHA256,
|
||||
"revision": strconv.Itoa(run.Revision), "final": strconv.FormatBool(run.IsFinal),
|
||||
}
|
||||
if err := s.verifyObject(ctx, run.ManifestKey, -1, manifestMetadata); err != nil {
|
||||
return false, nil
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (s *Service) integrationRecordCount(ctx context.Context, start, end time.Time) (int64, error) {
|
||||
var count int64
|
||||
if err := s.db.WithContext(ctx).Model(&model.IntegrationLog{}).
|
||||
Where("created_at >= ? AND created_at < ?", start, end).Count(&count).Error; err != nil {
|
||||
return 0, fmt.Errorf("统计 Integration Log 归档记录失败: %w", err)
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
func (s *Service) integrationPendingCount(ctx context.Context, start, end time.Time) (int64, error) {
|
||||
var count int64
|
||||
if err := s.db.WithContext(ctx).Model(&model.IntegrationLog{}).
|
||||
Where("created_at >= ? AND created_at < ? AND result = ?", start, end, constants.IntegrationResultPending).
|
||||
Count(&count).Error; err != nil {
|
||||
return 0, fmt.Errorf("统计 pending Integration Log 失败: %w", err)
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
func integrationObjectKeys(date time.Time, revision int) (string, string) {
|
||||
prefix := fmt.Sprintf("audit-archive/v1/%04d/%02d/%02d", date.Year(), date.Month(), date.Day())
|
||||
name := fmt.Sprintf("integration-logs-%s-r%d", date.Format(time.DateOnly), revision)
|
||||
return prefix + "/" + name + ".jsonl.gz", prefix + "/" + name + ".manifest.json"
|
||||
}
|
||||
|
||||
func integrationArchiveMetadata(file *integrationArchiveFile, run *model.LogArchiveRun, final bool) map[string]string {
|
||||
return map[string]string{
|
||||
"schema-version": constants.IntegrationArchiveSchemaVersion,
|
||||
"source": constants.IntegrationArchiveSource,
|
||||
"archive-date": run.RangeStart.Format(time.DateOnly),
|
||||
"timezone": constants.AuditArchiveTimezone,
|
||||
"record-count": strconv.FormatInt(file.recordCount, 10),
|
||||
"sha256": file.sha256,
|
||||
"revision": strconv.Itoa(run.Revision),
|
||||
"final": strconv.FormatBool(final),
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user