csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复批量订购

Browse Source
This commit is contained in:
break
2026-08-26 14:54:26 +08:00
parent 5797fd0e94
commit 62f3d25e81
22 changed files with 478 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
internal/application/auditarchive/integration.go
View File

@@ -77,20 +77,20 @@ func (s *Service) archiveIntegrationDate(ctx context.Context, archiveDate time.T
if err != nil {
return err
}
if final && run.Status == constants.ArchiveStatusSuccess && run.IsFinal && run.CleanedAt != nil {
terminalCount, countErr := s.integrationTerminalCount(ctx, start, end)
if countErr != nil {
return countErr
}
if terminalCount == 0 {
return nil
}
}
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 {
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) {
@@ -146,7 +146,8 @@ func (s *Service) acquireIntegrationRun(ctx context.Context, run *model.LogArchi
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,
"attempt_count": gorm.Expr("attempt_count + 1"), "error_summary": "", "completed_at": nil,
"cleanup_started_at": nil, "cleaned_at": nil, "updated_at": now,
})
if result.Error != nil {
return false, fmt.Errorf("锁定 Integration Log 归档任务失败: %w", result.Error)
@@ -301,16 +302,6 @@ func (s *Service) integrationRecordCount(ctx context.Context, start, end time.Ti
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)

172
internal/application/auditarchive/retention.go
View File

@@ -7,7 +7,6 @@ import (
"io"
"os"
"strconv"
"strings"
"time"
"github.com/bytedance/sonic"
@@ -50,11 +49,6 @@ func (e *RetentionBlockedError) Error() string {
}
func (e *RetentionBlockedError) Unwrap() error { return e.Err }
type retentionRuns struct {
audit *model.LogArchiveRun
integration *model.LogArchiveRun
}
// RetainPendingDays 从最早仍在线的日期连续处理至昨天。cleanup 为 false 时只读校验。
func (s *Service) RetainPendingDays(ctx context.Context, cleanup bool) ([]RetentionResult, error) {
if s.db == nil || s.store == nil {
@@ -65,19 +59,39 @@ func (s *Service) RetainPendingDays(ctx context.Context, cleanup bool) ([]Retent
return nil, err
}
results := make([]RetentionResult, 0)
var auditBlocked, integrationBlocked *RetentionBlockedError
for date := start; date.Before(end); date = date.AddDate(0, 0, 1) {
result, err := s.retainDate(ctx, date, cleanup)
if err != nil {
source := "retention"
if strings.HasPrefix(err.Error(), "Audit") {
source = constants.AuditArchiveSource
startedAt := time.Now()
result := RetentionResult{ArchiveDate: date.Format(time.DateOnly)}
if auditBlocked == nil {
auditResult, auditErr := s.retainAuditDate(ctx, date, cleanup)
if auditErr != nil {
auditBlocked = &RetentionBlockedError{ArchiveDate: result.ArchiveDate, Source: constants.AuditArchiveSource, Err: auditErr}
} else {
result.EventCount, result.ResourceCount = auditResult.EventCount, auditResult.ResourceCount
result.ManifestKeys = append(result.ManifestKeys, auditResult.ManifestKeys...)
}
if strings.HasPrefix(err.Error(), "Integration") {
source = constants.IntegrationArchiveSource
}
return results, &RetentionBlockedError{ArchiveDate: date.Format(time.DateOnly), Source: source, Err: err}
}
results = append(results, result)
if integrationBlocked == nil {
integrationResult, integrationErr := s.retainIntegrationDate(ctx, date, cleanup)
if integrationErr != nil {
integrationBlocked = &RetentionBlockedError{ArchiveDate: result.ArchiveDate, Source: constants.IntegrationArchiveSource, Err: integrationErr}
} else {
result.IntegrationCount = integrationResult.IntegrationCount
result.ManifestKeys = append(result.ManifestKeys, integrationResult.ManifestKeys...)
}
}
result.EstimatedBatches = estimatedRetentionBatches(result)
result.Duration = time.Since(startedAt)
if auditBlocked == nil || integrationBlocked == nil {
results = append(results, result)
}
}
if auditBlocked != nil {
return results, auditBlocked
}
if integrationBlocked != nil {
return results, integrationBlocked
}
return results, nil
}
@@ -91,9 +105,16 @@ func (s *Service) pendingRetentionRange(ctx context.Context) (time.Time, time.Ti
today := time.Now().In(s.location)
end := time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, s.location)
var earliest *time.Time
for _, item := range []struct{ table, column string }{{"tb_audit_event", "created_at"}, {"tb_integration_log", "created_at"}} {
for _, item := range []struct{ table, column, condition string }{
{"tb_audit_event", "created_at", ""},
{"tb_integration_log", "created_at", "result <> 'pending'"},
} {
query := s.db.WithContext(ctx).Table(item.table).Select("MIN(" + item.column + ")")
if item.condition != "" {
query = query.Where(item.condition)
}
var value *time.Time
if err := s.db.WithContext(ctx).Table(item.table).Select("MIN(" + item.column + ")").Scan(&value).Error; err != nil {
if err := query.Scan(&value).Error; err != nil {
return time.Time{}, time.Time{}, fmt.Errorf("查询日留存起点失败: %w", err)
}
if value != nil && (earliest == nil || value.Before(*earliest)) {
@@ -110,20 +131,52 @@ func (s *Service) pendingRetentionRange(ctx context.Context) (time.Time, time.Ti
func (s *Service) retainDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) {
startedAt := time.Now()
var existing []model.LogArchiveRun
if err := s.db.WithContext(ctx).Where("source IN ? AND archive_date = ? AND instance_id = ?", []string{constants.AuditArchiveSource, constants.IntegrationArchiveSource}, date.Format(time.DateOnly), s.instanceID).Find(&existing).Error; err != nil {
return RetentionResult{}, fmt.Errorf("读取日归档账本失败: %w", err)
auditResult, err := s.retainAuditDate(ctx, date, cleanup)
if err != nil {
return RetentionResult{}, err
}
statuses := map[string]string{}
for _, run := range existing {
statuses[run.Source] = run.Status
integrationResult, err := s.retainIntegrationDate(ctx, date, cleanup)
if err != nil {
return RetentionResult{}, err
}
if statuses[constants.AuditArchiveSource] != constants.ArchiveStatusSuccess {
result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), EventCount: auditResult.EventCount, ResourceCount: auditResult.ResourceCount, IntegrationCount: integrationResult.IntegrationCount, ManifestKeys: append(auditResult.ManifestKeys, integrationResult.ManifestKeys...)}
result.EstimatedBatches = estimatedRetentionBatches(result)
result.Duration = time.Since(startedAt)
return result, nil
}
func (s *Service) retainAuditDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) {
run, err := s.retentionRun(ctx, date, constants.AuditArchiveSource)
if err != nil {
return RetentionResult{}, err
}
if run == nil || run.Status != constants.ArchiveStatusSuccess {
if err := s.ArchiveDate(ctx, date); err != nil {
return RetentionResult{}, fmt.Errorf("Audit 归档失败: %w", err)
}
run, err = s.retentionRun(ctx, date, constants.AuditArchiveSource)
if err != nil {
return RetentionResult{}, err
}
}
if statuses[constants.IntegrationArchiveSource] != constants.ArchiveStatusSuccess {
if err := s.validateAuditRetentionDay(ctx, date, run); err != nil {
return RetentionResult{}, fmt.Errorf("Audit 完整性校验失败: %w", err)
}
result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), EventCount: run.EventCount, ResourceCount: run.ResourceCount, ManifestKeys: []string{run.ManifestKey}}
if cleanup {
if err := s.cleanupAuditDate(ctx, date, run); err != nil {
return result, err
}
}
return result, nil
}
func (s *Service) retainIntegrationDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) {
run, err := s.retentionRun(ctx, date, constants.IntegrationArchiveSource)
if err != nil {
return RetentionResult{}, err
}
if run == nil || run.Status != constants.ArchiveStatusSuccess {
if err := s.ArchiveIntegrationDate(ctx, date); err != nil {
return RetentionResult{}, fmt.Errorf("Integration Log 归档失败: %w", err)
}
@@ -131,48 +184,31 @@ func (s *Service) retainDate(ctx context.Context, date time.Time, cleanup bool)
if err := s.FinalizeIntegrationDate(ctx, date); err != nil {
return RetentionResult{}, fmt.Errorf("Integration Log 最终归档失败: %w", err)
}
runs, err := s.loadRetentionRuns(ctx, date)
run, err = s.retentionRun(ctx, date, constants.IntegrationArchiveSource)
if err != nil {
return RetentionResult{}, err
}
if err := s.validateAuditRetentionDay(ctx, date, runs.audit); err != nil {
return RetentionResult{}, fmt.Errorf("Audit 完整性校验失败: %w", err)
}
if err := s.validateIntegrationRetentionDay(ctx, date, runs.integration); err != nil {
if err := s.validateIntegrationRetentionDay(ctx, date, run); err != nil {
return RetentionResult{}, fmt.Errorf("Integration Log 完整性校验失败: %w", err)
}
result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), EventCount: runs.audit.EventCount, ResourceCount: runs.audit.ResourceCount, IntegrationCount: runs.integration.RecordCount, ManifestKeys: []string{runs.audit.ManifestKey, runs.integration.ManifestKey}}
result.EstimatedBatches = estimatedRetentionBatches(result)
result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), IntegrationCount: run.RecordCount, ManifestKeys: []string{run.ManifestKey}}
if cleanup {
if err := s.cleanupAuditDate(ctx, date, runs.audit); err != nil {
return result, err
}
if err := s.cleanupIntegrationDate(ctx, date, runs.integration); err != nil {
if err := s.cleanupIntegrationDate(ctx, date, run); err != nil {
return result, err
}
}
result.Duration = time.Since(startedAt)
return result, nil
}
func (s *Service) loadRetentionRuns(ctx context.Context, date time.Time) (retentionRuns, error) {
var rows []model.LogArchiveRun
if err := s.db.WithContext(ctx).Where("source IN ? AND archive_date = ? AND instance_id = ?", []string{constants.AuditArchiveSource, constants.IntegrationArchiveSource}, date.Format(time.DateOnly), s.instanceID).Find(&rows).Error; err != nil {
return retentionRuns{}, fmt.Errorf("读取日归档账本失败: %w", err)
func (s *Service) retentionRun(ctx context.Context, date time.Time, source string) (*model.LogArchiveRun, error) {
var runs []model.LogArchiveRun
if err := s.db.WithContext(ctx).Where("source = ? AND archive_date = ? AND instance_id = ?", source, date.Format(time.DateOnly), s.instanceID).Find(&runs).Error; err != nil {
return nil, fmt.Errorf("读取日归档账本失败: %w", err)
}
runs := retentionRuns{}
for index := range rows {
switch rows[index].Source {
case constants.AuditArchiveSource:
runs.audit = &rows[index]
case constants.IntegrationArchiveSource:
runs.integration = &rows[index]
}
if len(runs) == 0 {
return nil, nil
}
if runs.audit == nil || runs.integration == nil {
return retentionRuns{}, fmt.Errorf("Audit 或 Integration Log 归档账本缺失")
}
return runs, nil
return &runs[0], nil
}
func (s *Service) validateAuditRetentionDay(ctx context.Context, date time.Time, run *model.LogArchiveRun) error {
@@ -200,13 +236,17 @@ func (s *Service) validateIntegrationRetentionDay(ctx context.Context, date time
if err != nil {
return err
}
if run.CleanedAt != nil && count != 0 {
return fmt.Errorf("已标记清理完成但数据库仍有 %d 条记录", count)
terminalCount, err := s.integrationTerminalCount(ctx, run.RangeStart, run.RangeEnd)
if err != nil {
return err
}
if run.CleanedAt != nil && terminalCount != 0 {
return fmt.Errorf("已标记清理完成但数据库仍有 %d 条终态记录", terminalCount)
}
if run.CleanupStartedAt != nil && count > run.RecordCount {
return fmt.Errorf("续跑窗口记录数超过最终归档数量")
}
if run.CleanupStartedAt == nil {
if terminalCount > 0 && run.CleanupStartedAt == nil {
file, err := s.buildIntegrationArchiveFile(ctx, run.RangeStart, run.RangeEnd)
if err != nil {
return err
@@ -348,7 +388,11 @@ func (s *Service) cleanupAuditDate(ctx context.Context, date time.Time, run *mod
return s.markCleaned(ctx, constants.AuditArchiveSource, date)
}
func (s *Service) cleanupIntegrationDate(ctx context.Context, date time.Time, run *model.LogArchiveRun) error {
if run.CleanedAt != nil {
terminalCount, err := s.integrationTerminalCount(ctx, date, date.AddDate(0, 0, 1))
if err != nil {
return err
}
if run.CleanedAt != nil && terminalCount == 0 {
return nil
}
if err := s.markCleanupStarted(ctx, constants.IntegrationArchiveSource, date); err != nil {
@@ -356,7 +400,7 @@ func (s *Service) cleanupIntegrationDate(ctx context.Context, date time.Time, ru
}
end := date.AddDate(0, 0, 1)
for {
subquery := s.db.Model(&model.IntegrationLog{}).Select("id").Where("created_at >= ? AND created_at < ?", date, end).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize)
subquery := s.db.Model(&model.IntegrationLog{}).Select("id").Where("created_at >= ? AND created_at < ? AND result <> ?", date, end, constants.IntegrationResultPending).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)
@@ -367,6 +411,16 @@ func (s *Service) cleanupIntegrationDate(ctx context.Context, date time.Time, ru
}
return s.markCleaned(ctx, constants.IntegrationArchiveSource, date)
}
func (s *Service) integrationTerminalCount(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("统计终态 Integration Log 失败: %w", err)
}
return count, nil
}
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)

10
internal/task/asset_package_batch_order.go
View File

@@ -121,19 +121,13 @@ func (h *AssetPackageBatchOrderHandler) finishBatchOrderTask(ctx context.Context
return err
}
rootID := audit.TaskEventID(constants.AuditResourceAssetPackageBatchOrderTask, taskRecord.ID, "completed")
var childCount int64
if err := tx.WithContext(ctx).Model(&model.AuditEvent{}).
Where("parent_event_id = ? AND action_code = ?", rootID, constants.AuditActionOrderCreated).
Count(&childCount).Error; err != nil {
return err
}
result := batchAuditResult(int(childCount), failCount)
result := batchAuditResult(successCount, failCount)
return h.auditWriter.WriteTask(ctx, tx, audit.TaskInput{
EventID: rootID, ActionCode: constants.AuditActionAssetPackageBatchOrderTaskCompleted,
Summary: "完成资产套餐批量订购任务", TaskID: taskRecord.ID, TaskNo: taskRecord.TaskNo,
Result: result, CorrelationID: taskRecord.TaskNo,
ParentEventID: audit.TaskEventID(constants.AuditResourceAssetPackageBatchOrderTask, taskRecord.ID, "created"),
BatchTotal: len(items), SuccessCount: int(childCount), FailCount: failCount,
BatchTotal: len(items), SuccessCount: successCount, FailCount: failCount,
IdentitySnapshot: map[string]any{
"id": taskRecord.ID, "task_no": taskRecord.TaskNo, "file_name": taskRecord.FileName,
"package_id": taskRecord.PackageID, "package_code": taskRecord.PackageCode,

8
internal/task/polling_carddata_handler.go
View File

@@ -67,11 +67,15 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e
}
result, err := h.gateway.QueryFlow(ctx, &gateway.FlowQueryReq{CardNo: card.ICCID})
if err != nil {
_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 流量 Integration Log 失败", logErr)
}
return h.failAndRequeue(ctx, cardID, startedAt, "查询流量失败", err)
}
if strings.TrimSpace(result.ICCID) == "" {
_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 流量 Integration Log 失败", logErr)
}
return h.failAndRequeue(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil)
}
if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil {

8
internal/task/polling_cardstatus_handler.go
View File

@@ -66,11 +66,15 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task)
}
result, err := h.gateway.QueryCardStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID})
if err != nil {
_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 网络 Integration Log 失败", logErr)
}
return h.failAndRequeue(ctx, cardID, startedAt, "查询卡状态失败", err)
}
if strings.TrimSpace(result.ICCID) == "" {
_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 网络 Integration Log 失败", logErr)
}
return h.failAndRequeue(ctx, cardID, startedAt, "卡状态查询响应缺少 ICCID", nil)
}
if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil {

4
internal/task/polling_integration_log.go
View File

@@ -38,11 +38,13 @@ func completeGatewayAttempt(ctx context.Context, repository *integrationlog.Repo
if repository == nil || attempt == nil {
return errors.New(errors.CodeInternalError, "Gateway Integration Log 尝试不存在")
}
completionCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
defer cancel()
result := constants.IntegrationResultFailed
if success {
result = constants.IntegrationResultSuccess
}
_, err := repository.Complete(ctx, attempt.IntegrationID, integrationlog.Completion{
_, err := repository.Complete(completionCtx, attempt.IntegrationID, integrationlog.Completion{
Result: result, DurationMS: time.Since(startedAt).Milliseconds(), StateChanged: false,
ResponseSummary: map[string]any{"success": success},
})

8
internal/task/polling_realname_handler.go
View File

@@ -58,11 +58,15 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) e
}
result, err := h.gateway.QueryRealnameStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID})
if err != nil {
_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 实名 Integration Log 失败", logErr)
}
return h.failAndRequeue(ctx, cardID, startedAt, "查询实名状态失败", err)
}
if strings.TrimSpace(result.ICCID) == "" {
_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 实名 Integration Log 失败", logErr)
}
return h.failAndRequeue(ctx, cardID, startedAt, "实名查询响应缺少 ICCID", nil)
}
if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil {