break
88cc5e96ec
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 8m32s
114 lines
3.8 KiB
Go
114 lines
3.8 KiB
Go
package task
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/bytedance/sonic"
|
|
"github.com/hibiken/asynq"
|
|
"go.uber.org/zap"
|
|
|
|
"github.com/break/junhong_cmp_fiber/internal/application/auditarchive"
|
|
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
|
)
|
|
|
|
// IntegrationDailyArchivePayload 是人工补档时可选的任务载荷。
|
|
type IntegrationDailyArchivePayload struct {
|
|
ArchiveDate string `json:"archive_date"`
|
|
}
|
|
|
|
// IntegrationMonthlyFinalizePayload 是人工月度复核时可选的任务载荷。
|
|
type IntegrationMonthlyFinalizePayload struct {
|
|
ArchiveMonth string `json:"archive_month"`
|
|
}
|
|
|
|
// IntegrationArchiveHandler 处理 Integration Log 每日归档与月度最终复核。
|
|
type IntegrationArchiveHandler struct {
|
|
service *auditarchive.Service
|
|
logger *zap.Logger
|
|
}
|
|
|
|
// NewIntegrationArchiveHandler 创建 Integration Log 归档任务处理器。
|
|
func NewIntegrationArchiveHandler(service *auditarchive.Service, logger *zap.Logger) *IntegrationArchiveHandler {
|
|
return &IntegrationArchiveHandler{service: service, logger: logger}
|
|
}
|
|
|
|
// HandleDaily 执行前一完整自然日归档，或按任务载荷补档指定自然日。
|
|
func (h *IntegrationArchiveHandler) HandleDaily(ctx context.Context, task *asynq.Task) error {
|
|
if h.service == nil {
|
|
return fmt.Errorf("Integration Log 归档服务未配置")
|
|
}
|
|
var err error
|
|
if len(task.Payload()) == 0 {
|
|
err = h.service.ArchivePreviousIntegrationDay(ctx)
|
|
} else {
|
|
var payload IntegrationDailyArchivePayload
|
|
if unmarshalErr := sonic.Unmarshal(task.Payload(), &payload); unmarshalErr != nil {
|
|
return fmt.Errorf("解析 Integration Log 每日归档任务载荷失败: %w", unmarshalErr)
|
|
}
|
|
date, parseErr := parseArchiveDate(payload.ArchiveDate)
|
|
if parseErr != nil {
|
|
return parseErr
|
|
}
|
|
err = h.service.ArchiveIntegrationDate(ctx, date)
|
|
}
|
|
if err != nil {
|
|
h.logger.Error("Integration Log 每日冷归档失败", zap.Error(err))
|
|
return err
|
|
}
|
|
h.logger.Info("Integration Log 每日冷归档完成")
|
|
return nil
|
|
}
|
|
|
|
// HandleMonthlyFinalize 执行上一个完整自然月复核，或按任务载荷复核指定月份。
|
|
func (h *IntegrationArchiveHandler) HandleMonthlyFinalize(ctx context.Context, task *asynq.Task) error {
|
|
if h.service == nil {
|
|
return fmt.Errorf("Integration Log 归档服务未配置")
|
|
}
|
|
var err error
|
|
if len(task.Payload()) == 0 {
|
|
err = h.service.FinalizePreviousIntegrationMonth(ctx)
|
|
} else {
|
|
var payload IntegrationMonthlyFinalizePayload
|
|
if unmarshalErr := sonic.Unmarshal(task.Payload(), &payload); unmarshalErr != nil {
|
|
return fmt.Errorf("解析 Integration Log 月度复核任务载荷失败: %w", unmarshalErr)
|
|
}
|
|
month, parseErr := parseArchiveMonth(payload.ArchiveMonth)
|
|
if parseErr != nil {
|
|
return parseErr
|
|
}
|
|
err = h.service.FinalizeIntegrationMonth(ctx, month)
|
|
}
|
|
if err != nil {
|
|
h.logger.Error("Integration Log 月度最终版本复核失败，后续清理必须阻止", zap.Error(err))
|
|
return err
|
|
}
|
|
h.logger.Info("Integration Log 月度最终版本复核完成")
|
|
return nil
|
|
}
|
|
|
|
func parseArchiveDate(value string) (time.Time, error) {
|
|
location, err := time.LoadLocation(constants.AuditArchiveTimezone)
|
|
if err != nil {
|
|
return time.Time{}, fmt.Errorf("加载 Integration Log 归档时区失败: %w", err)
|
|
}
|
|
date, err := time.ParseInLocation(time.DateOnly, value, location)
|
|
if err != nil {
|
|
return time.Time{}, fmt.Errorf("解析 Integration Log 归档日期失败: %w", err)
|
|
}
|
|
return date, nil
|
|
}
|
|
|
|
func parseArchiveMonth(value string) (time.Time, error) {
|
|
location, err := time.LoadLocation(constants.AuditArchiveTimezone)
|
|
if err != nil {
|
|
return time.Time{}, fmt.Errorf("加载 Integration Log 归档时区失败: %w", err)
|
|
}
|
|
month, err := time.ParseInLocation("2006-01", value, location)
|
|
if err != nil {
|
|
return time.Time{}, fmt.Errorf("解析 Integration Log 归档月份失败: %w", err)
|
|
}
|
|
return month, nil
|
|
}
|