break
59b3df868a
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 12m59s
69 lines
2.6 KiB
Go
69 lines
2.6 KiB
Go
package carrierthreshold
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/hibiken/asynq"
|
|
|
|
carrierthresholdapp "github.com/break/junhong_cmp_fiber/internal/application/carrierthreshold"
|
|
"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
|
|
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
|
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
|
)
|
|
|
|
// CycleTaskHandler 执行运营商通道流量阈值周期处理任务。
|
|
type CycleTaskHandler struct {
|
|
service *carrierthresholdapp.Service
|
|
}
|
|
|
|
// NewCycleTaskHandler 创建通道阈值周期处理任务 Handler。
|
|
func NewCycleTaskHandler(service *carrierthresholdapp.Service) *CycleTaskHandler {
|
|
return &CycleTaskHandler{service: service}
|
|
}
|
|
|
|
// Handle 扫描已跨期的持锁，认领解锁并在条件满足时写复机事件。
|
|
// 审计上下文固定为计划任务来源，使解锁与复机事件都可按计划任务维度追溯。
|
|
func (h *CycleTaskHandler) Handle(ctx context.Context, task *asynq.Task) error {
|
|
if h == nil || h.service == nil {
|
|
return errors.New(errors.CodeServiceUnavailable, "通道流量阈值周期处理任务未配置")
|
|
}
|
|
taskType := constants.TaskTypeCarrierThresholdCycle
|
|
if task != nil && task.Type() != "" {
|
|
taskType = task.Type()
|
|
}
|
|
ctx = auditcontext.With(ctx, auditcontext.Context{
|
|
ActorKind: constants.AuditActorScheduledJob, ActorID: taskType,
|
|
ActorName: "运营商通道流量阈值周期处理计划任务", Source: constants.AuditSourceScheduler,
|
|
})
|
|
_, err := h.service.ProcessDueLocks(ctx, time.Now().UTC())
|
|
return err
|
|
}
|
|
|
|
// RecoveryTaskHandler 执行运营商通道流量阈值停复机结果恢复任务。
|
|
type RecoveryTaskHandler struct {
|
|
service *carrierthresholdapp.Service
|
|
}
|
|
|
|
// NewRecoveryTaskHandler 创建通道阈值停复机结果恢复任务 Handler。
|
|
func NewRecoveryTaskHandler(service *carrierthresholdapp.Service) *RecoveryTaskHandler {
|
|
return &RecoveryTaskHandler{service: service}
|
|
}
|
|
|
|
// Handle 扫描已提交的停复机子任务并只查询运营商状态回填，绝不重复发起停复机。
|
|
func (h *RecoveryTaskHandler) Handle(ctx context.Context, task *asynq.Task) error {
|
|
if h == nil || h.service == nil {
|
|
return errors.New(errors.CodeServiceUnavailable, "通道流量阈值恢复扫描任务未配置")
|
|
}
|
|
taskType := constants.TaskTypeCarrierThresholdRecovery
|
|
if task != nil && task.Type() != "" {
|
|
taskType = task.Type()
|
|
}
|
|
ctx = auditcontext.With(ctx, auditcontext.Context{
|
|
ActorKind: constants.AuditActorScheduledJob, ActorID: taskType,
|
|
ActorName: "运营商通道流量阈值结果恢复计划任务", Source: constants.AuditSourceScheduler,
|
|
})
|
|
_, err := h.service.RecoverSubmitted(ctx, time.Now().UTC())
|
|
return err
|
|
}
|