178 lines
6.8 KiB
Go
178 lines
6.8 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
stderrors "errors"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"go.uber.org/zap"
|
||
|
||
cardapp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation"
|
||
carddomain "github.com/break/junhong_cmp_fiber/internal/domain/cardobservation"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
)
|
||
|
||
// IngestExecutor 消费轮询观测入库消息（topic card.ingest）：复用卡观测应用服务写库，按采集 ID 去重；
|
||
// 可重试错误按退避延迟重投且不设次数上限，业务性错误记录后丢弃。
|
||
type IngestExecutor struct {
|
||
base *PollingBase
|
||
observation *cardapp.Service
|
||
timeout time.Duration
|
||
// lastLagMS 是最近处理消息的发布到开始处理的等待时长，供积压监控读取。
|
||
lastLagMS atomic.Int64
|
||
}
|
||
|
||
// NewIngestExecutor 创建入库消费者。
|
||
func NewIngestExecutor(base *PollingBase, observation *cardapp.Service, timeout time.Duration) *IngestExecutor {
|
||
if timeout <= 0 {
|
||
timeout = constants.PollingIngestDefaultTimeout
|
||
}
|
||
return &IngestExecutor{base: base, observation: observation, timeout: timeout}
|
||
}
|
||
|
||
// LastLag 返回最近处理的入库消息从发布到开始处理的等待时长。
|
||
func (e *IngestExecutor) LastLag() time.Duration {
|
||
if e == nil {
|
||
return 0
|
||
}
|
||
return time.Duration(e.lastLagMS.Load()) * time.Millisecond
|
||
}
|
||
|
||
// Handle 处理一条入库消息，返回 0 表示确认，正数表示按该延迟重投。
|
||
func (e *IngestExecutor) Handle(ctx context.Context, body []byte, attempts uint16, _ func()) time.Duration {
|
||
var msg IngestMessage
|
||
if err := json.Unmarshal(body, &msg); err != nil || !msg.valid() {
|
||
e.base.logger.Warn("入库消息无效，丢弃", zap.ByteString("body", body))
|
||
return 0
|
||
}
|
||
if !msg.PublishedAt.IsZero() {
|
||
e.lastLagMS.Store(time.Since(msg.PublishedAt).Milliseconds())
|
||
}
|
||
if e.alreadyDone(ctx, msg.ObservationID) {
|
||
return 0
|
||
}
|
||
taskCtx, cancel := context.WithTimeout(ctx, e.timeout)
|
||
defer cancel()
|
||
taskCtx = withPollingWorkerAuditContext(taskCtx, msg.TaskType, pollingPriorityTaskName(msg.TaskType), msg.ObservationID)
|
||
changed, err := e.apply(taskCtx, msg)
|
||
if err != nil {
|
||
return e.onFailure(msg, attempts, err)
|
||
}
|
||
e.markDone(msg.ObservationID)
|
||
if changed && msg.Kind == ingestKindTraffic {
|
||
e.markFlowEvaluated(msg.CardID)
|
||
}
|
||
return 0
|
||
}
|
||
|
||
// apply 按观测类型调用卡观测应用服务，返回流量是否产生增量或跨月。
|
||
func (e *IngestExecutor) apply(ctx context.Context, msg IngestMessage) (bool, error) {
|
||
if e.observation == nil {
|
||
return false, errors.New(errors.CodeInternalError, "卡观测能力未配置")
|
||
}
|
||
switch msg.Kind {
|
||
case ingestKindTraffic:
|
||
decision, err := e.observation.ApplyTrafficObservation(ctx, carddomain.TrafficObservation{
|
||
CardID: msg.CardID, GatewayReadingMB: msg.Traffic.ReadingMB, ReadingResult: msg.Traffic.ReadingResult,
|
||
ResetDay: msg.Traffic.ResetDay, Raw: msg.Raw, ICCID: msg.ICCID, CarrierID: msg.CarrierID,
|
||
Failure: &carddomain.CollectionFailure{DurationMS: msg.Traffic.DurationMS},
|
||
Metadata: msg.metadata(),
|
||
})
|
||
return decision.IncrementMB > 0 || decision.CrossMonth, err
|
||
case ingestKindNetwork:
|
||
_, err := e.observation.ApplyNetworkObservation(ctx, carddomain.NetworkObservation{
|
||
CardID: msg.CardID, GatewayStatus: msg.Network.Status, GatewayExtend: msg.Network.Extend,
|
||
GatewayIMEI: msg.Network.IMEI, Metadata: msg.metadata(),
|
||
})
|
||
return false, err
|
||
case ingestKindRealname:
|
||
_, err := e.observation.ApplyCardObservation(ctx, carddomain.RealnameObservation{
|
||
CardID: msg.CardID, Verified: msg.Realname.Verified, UpstreamRealnameAt: msg.Realname.UpstreamRealnameAt, Metadata: msg.metadata(),
|
||
})
|
||
return false, err
|
||
}
|
||
return false, errors.New(errors.CodeInvalidParam, "未知的入库观测类型")
|
||
}
|
||
|
||
// onFailure 业务性错误丢弃，其余按退避重投；第 10 次起每次重投记 ERROR。
|
||
func (e *IngestExecutor) onFailure(msg IngestMessage, attempts uint16, err error) time.Duration {
|
||
fields := []zap.Field{zap.String("kind", msg.Kind), zap.Uint("card_id", msg.CardID),
|
||
zap.String("observation_id", msg.ObservationID), zap.Uint16("attempts", attempts), zap.Error(err)}
|
||
if ingestBusinessError(err) {
|
||
e.base.logger.Warn("入库遇到业务性错误，丢弃消息", fields...)
|
||
return 0
|
||
}
|
||
delay := ingestRetryDelay(attempts)
|
||
fields = append(fields, zap.Duration("retry_after", delay))
|
||
if attempts >= constants.PollingIngestErrorLogFromAttempt {
|
||
e.base.logger.Error("入库持续失败，继续重投", fields...)
|
||
} else {
|
||
e.base.logger.Warn("入库失败，延迟重投", fields...)
|
||
}
|
||
return delay
|
||
}
|
||
|
||
// ingestBusinessError 判断错误是否为重试也不会成功的业务性错误（卡不存在、观测数据校验失败）。
|
||
func ingestBusinessError(err error) bool {
|
||
var appErr *errors.AppError
|
||
if !stderrors.As(err, &appErr) {
|
||
return false
|
||
}
|
||
switch appErr.Code {
|
||
case errors.CodeNotFound, errors.CodeInvalidParam, errors.CodeInvalidStatus:
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
func (e *IngestExecutor) alreadyDone(ctx context.Context, observationID string) bool {
|
||
if e.base.redis == nil {
|
||
return false
|
||
}
|
||
exists, err := e.base.redis.Exists(ctx, constants.RedisPollingIngestDoneKey(observationID)).Result()
|
||
return err == nil && exists > 0
|
||
}
|
||
|
||
func (e *IngestExecutor) markDone(observationID string) {
|
||
if e.base.redis == nil {
|
||
return
|
||
}
|
||
markCtx, cancel := pollingFallbackContext()
|
||
defer cancel()
|
||
if err := e.base.redis.Set(markCtx, constants.RedisPollingIngestDoneKey(observationID), "1", constants.PollingIngestDoneTTL).Err(); err != nil {
|
||
e.base.logger.Warn("写入库已处理标记失败，重投时由业务幂等兜底", zap.String("observation_id", observationID), zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// markFlowEvaluated 流量入库产生增量或跨月时写评估标记，有效期取该卡套餐检查间隔；套餐步骤据此跳过。
|
||
func (e *IngestExecutor) markFlowEvaluated(cardID uint) {
|
||
if e.base.redis == nil {
|
||
return
|
||
}
|
||
markCtx, cancel := pollingFallbackContext()
|
||
defer cancel()
|
||
ttl := constants.PollingFlowEvaluatedFallbackTTL
|
||
if card, err := e.base.getCardWithCache(markCtx, cardID); err == nil && e.base.configMgr != nil {
|
||
info, ok := e.base.configMgr.MergedTaskIntervals(card)[constants.TaskTypePollingPackage]
|
||
if !ok || info.Interval <= 0 {
|
||
return
|
||
}
|
||
ttl = time.Duration(info.Interval) * time.Second
|
||
}
|
||
if err := e.base.redis.Set(markCtx, constants.RedisPollingFlowEvaluatedKey(cardID), "1", ttl).Err(); err != nil {
|
||
e.base.logger.Debug("写流量评估标记失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// flowEvaluatedRecently 判断入库段是否在套餐检查间隔内写过流量评估标记。
|
||
func (b *PollingBase) flowEvaluatedRecently(ctx context.Context, cardID uint) bool {
|
||
if b == nil || b.redis == nil {
|
||
return false
|
||
}
|
||
exists, err := b.redis.Exists(ctx, constants.RedisPollingFlowEvaluatedKey(cardID)).Result()
|
||
return err == nil && exists > 0
|
||
}
|