158 lines
5.6 KiB
Go
158 lines
5.6 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"time"
|
||
|
||
carddomain "github.com/break/junhong_cmp_fiber/internal/domain/cardobservation"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
)
|
||
|
||
// 入库消息的观测类型。
|
||
const (
|
||
ingestKindTraffic = "traffic"
|
||
ingestKindNetwork = "network"
|
||
ingestKindRealname = "realname"
|
||
)
|
||
|
||
// ingestRawResponseLimit 是入库消息携带原始响应的上限：比采集记录保存上限多 1 字节，使"超过 4KB 截断"判断不变，同时控制消息大小。
|
||
const ingestRawResponseLimit = 4096 + 1
|
||
|
||
// IngestMessage 是一次轮询观测的入库消息（topic card.ingest）。采集时间取网关返回时刻，入库全程按它归属日期与判断新旧。
|
||
type IngestMessage struct {
|
||
Kind string `json:"kind"`
|
||
CardID uint `json:"card_id"`
|
||
CarrierID uint `json:"carrier_id"`
|
||
ICCID string `json:"iccid"`
|
||
ObservationID string `json:"observation_id"`
|
||
ObservedAt time.Time `json:"observed_at"`
|
||
Source string `json:"source"`
|
||
Scene string `json:"scene"`
|
||
CorrelationID string `json:"correlation_id"`
|
||
// TaskType 是发起观测的轮询任务类型，入库时用于审计上下文与流量评估标记。
|
||
TaskType string `json:"task_type"`
|
||
Traffic *IngestTrafficPayload `json:"traffic,omitempty"`
|
||
Network *IngestNetworkPayload `json:"network,omitempty"`
|
||
Realname *IngestRealnamePayload `json:"realname,omitempty"`
|
||
Raw *carddomain.GatewayRaw `json:"raw,omitempty"`
|
||
PublishedAt time.Time `json:"published_at"`
|
||
}
|
||
|
||
// IngestTrafficPayload 是流量观测内容。
|
||
type IngestTrafficPayload struct {
|
||
ReadingMB float64 `json:"reading_mb"`
|
||
ReadingResult string `json:"reading_result"`
|
||
ResetDay int `json:"reset_day"`
|
||
DurationMS int `json:"duration_ms"`
|
||
}
|
||
|
||
// IngestNetworkPayload 是网络状态观测内容。
|
||
type IngestNetworkPayload struct {
|
||
Status string `json:"status"`
|
||
Extend string `json:"extend"`
|
||
IMEI string `json:"imei"`
|
||
}
|
||
|
||
// IngestRealnamePayload 是实名观测内容。
|
||
type IngestRealnamePayload struct {
|
||
Verified bool `json:"verified"`
|
||
UpstreamRealnameAt *time.Time `json:"upstream_realname_at,omitempty"`
|
||
}
|
||
|
||
// metadata 还原观测元数据。
|
||
func (m IngestMessage) metadata() carddomain.ObservationMetadata {
|
||
return carddomain.ObservationMetadata{
|
||
ObservationID: m.ObservationID, Source: m.Source, Scene: m.Scene,
|
||
ObservedAt: m.ObservedAt, CorrelationID: m.CorrelationID,
|
||
}
|
||
}
|
||
|
||
// valid 校验消息必需字段。
|
||
func (m IngestMessage) valid() bool {
|
||
if m.CardID == 0 || m.ObservationID == "" || m.ObservedAt.IsZero() {
|
||
return false
|
||
}
|
||
switch m.Kind {
|
||
case ingestKindTraffic:
|
||
return m.Traffic != nil
|
||
case ingestKindNetwork:
|
||
return m.Network != nil
|
||
case ingestKindRealname:
|
||
return m.Realname != nil
|
||
}
|
||
return false
|
||
}
|
||
|
||
// CollectionMessage 是一条采集记录消息（topic card.collection），成功记录与失败记录二选一。
|
||
type CollectionMessage struct {
|
||
Success *model.CardTrafficCollection `json:"success,omitempty"`
|
||
Failure *model.CardTrafficCollectionFailure `json:"failure,omitempty"`
|
||
}
|
||
|
||
// MessagePublisher 发布 NSQ 消息。
|
||
type MessagePublisher interface {
|
||
Publish(topic string, body []byte) error
|
||
}
|
||
|
||
// IngestPublisher 发布入库消息与采集记录消息。
|
||
type IngestPublisher struct {
|
||
producer MessagePublisher
|
||
}
|
||
|
||
// NewIngestPublisher 创建入库消息发布器；producer 为 nil 时所有发布都返回错误，调用方降级同步入库。
|
||
func NewIngestPublisher(producer MessagePublisher) *IngestPublisher {
|
||
return &IngestPublisher{producer: producer}
|
||
}
|
||
|
||
// PublishIngest 发布一条入库消息。
|
||
func (p *IngestPublisher) PublishIngest(_ context.Context, msg IngestMessage) error {
|
||
if p == nil || p.producer == nil {
|
||
return errors.New(errors.CodeInternalError, "入库消息发布器未配置")
|
||
}
|
||
if msg.Raw != nil && len(msg.Raw.Response) > ingestRawResponseLimit {
|
||
trimmed := *msg.Raw
|
||
trimmed.Response = trimmed.Response[:ingestRawResponseLimit]
|
||
msg.Raw = &trimmed
|
||
}
|
||
msg.PublishedAt = time.Now()
|
||
body, err := json.Marshal(msg)
|
||
if err != nil {
|
||
return errors.Wrap(errors.CodeInternalError, err, "编码入库消息失败")
|
||
}
|
||
if err := p.producer.Publish(constants.NSQCardIngestTopic, body); err != nil {
|
||
return errors.Wrap(errors.CodeInternalError, err, "发布入库消息失败")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// PublishCollection 发布一条采集记录消息，实现卡观测服务的 CollectionSink。
|
||
func (p *IngestPublisher) PublishCollection(_ context.Context, success *model.CardTrafficCollection, failure *model.CardTrafficCollectionFailure) error {
|
||
if p == nil || p.producer == nil {
|
||
return errors.New(errors.CodeInternalError, "采集记录消息发布器未配置")
|
||
}
|
||
body, err := json.Marshal(CollectionMessage{Success: success, Failure: failure})
|
||
if err != nil {
|
||
return errors.Wrap(errors.CodeInternalError, err, "编码采集记录消息失败")
|
||
}
|
||
if err := p.producer.Publish(constants.NSQCardCollectionTopic, body); err != nil {
|
||
return errors.Wrap(errors.CodeInternalError, err, "发布采集记录消息失败")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ingestRetryDelay 返回第 attempts 次投递失败后的重投延迟；超过序列长度取最后一项，不设次数上限。
|
||
func ingestRetryDelay(attempts uint16) time.Duration {
|
||
delays := constants.PollingIngestRetryDelays
|
||
index := int(attempts) - 1
|
||
if index < 0 {
|
||
index = 0
|
||
}
|
||
if index >= len(delays) {
|
||
index = len(delays) - 1
|
||
}
|
||
return delays[index]
|
||
}
|