9c3c258f16
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 9m27s
- 每卡 Redis 采集队列（默认 N=3、跨天清空、回退容差 1 MB），攒满且不下降才写最后一个点 - 测试环境全量切换日快照计量，下线灰度名单、切换水位与卡表基线扣费；全部卡置待建起点 - 读数区分没给值与 0；有套餐的卡读数 0 也写快照与日用量 0 行；重置日比较基准为 0 照常防旧值 - 新增采集记录（按月分区保留 90 天）、人工冲正流水与后台接口、采集告警 - 扣减按采集时刻有效套餐查找，只对生效中套餐做状态迁移，查不到套餐按成功跳过 - 换货新卡（新设备下的卡）打待建起点；采集间隔上限只校验启用中的轮询配置 Change: add-card-traffic-queue-and-observed-deduction Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
131 lines
4.7 KiB
Go
131 lines
4.7 KiB
Go
package cardobservation
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"strconv"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
|
||
cardapp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation"
|
||
"github.com/break/junhong_cmp_fiber/internal/domain/cardflow"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
)
|
||
|
||
// trafficQueueScript 原子更新一张卡的采集队列：跨天清空（新点日期晚于队尾日期时）、丢弃过期点、
|
||
// 按采集时间插入并对同一时间去重、只保留最后 N 个、设置过期，返回队列与新点是否在队尾。
|
||
// 新点日期早于队尾日期（迟到的前一天点）时不改队列，按乱序处理。日期由调用方按上海时区算好传入。
|
||
//
|
||
// KEYS[1] 队列 key；ARGV: 采集时间毫秒、读数、日期(YYYYMMDD)、N、是否跨天清空(1/0)、最长点龄毫秒(0 不限)、过期毫秒。
|
||
var trafficQueueScript = redis.NewScript(`
|
||
local raw = redis.call('GET', KEYS[1])
|
||
local pts = {}
|
||
if raw then pts = cjson.decode(raw) end
|
||
local t = tonumber(ARGV[1])
|
||
local v = tonumber(ARGV[2])
|
||
local d = ARGV[3]
|
||
local n = tonumber(ARGV[4])
|
||
local clear = ARGV[5] == '1'
|
||
local maxAge = tonumber(ARGV[6])
|
||
local ttl = tonumber(ARGV[7])
|
||
if clear and #pts > 0 then
|
||
local lastDay = pts[#pts].d
|
||
if d < lastDay then
|
||
return {cjson.encode(pts), 0}
|
||
end
|
||
if d > lastDay then pts = {} end
|
||
end
|
||
if maxAge > 0 then
|
||
local kept = {}
|
||
for _, p in ipairs(pts) do
|
||
if t - p.t <= maxAge then table.insert(kept, p) end
|
||
end
|
||
pts = kept
|
||
end
|
||
local out = {}
|
||
local inserted = false
|
||
for _, p in ipairs(pts) do
|
||
if not inserted and t <= p.t then
|
||
table.insert(out, {t = t, v = v, d = d})
|
||
inserted = true
|
||
end
|
||
if p.t ~= t then table.insert(out, p) end
|
||
end
|
||
if not inserted then table.insert(out, {t = t, v = v, d = d}) end
|
||
while #out > n do table.remove(out, 1) end
|
||
local tail = 0
|
||
if #out > 0 and out[#out].t == t then tail = 1 end
|
||
redis.call('SET', KEYS[1], cjson.encode(out), 'PX', ttl)
|
||
return {cjson.encode(out), tail}
|
||
`)
|
||
|
||
// TrafficQueue 是基于 Redis 的卡流量采集队列。队列只是写入前的过滤器：key 丢失时从空开始，只晚写不错账。
|
||
type TrafficQueue struct {
|
||
redis *redis.Client
|
||
}
|
||
|
||
// NewTrafficQueue 创建卡流量采集队列。
|
||
func NewTrafficQueue(redisClient *redis.Client) *TrafficQueue {
|
||
return &TrafficQueue{redis: redisClient}
|
||
}
|
||
|
||
type queuedPoint struct {
|
||
T int64 `json:"t"`
|
||
V float64 `json:"v"`
|
||
D string `json:"d"`
|
||
}
|
||
|
||
// Push 把新点放入卡的采集队列，返回入队后的队列（按采集时间升序）与新点是否在队尾。
|
||
func (q *TrafficQueue) Push(ctx context.Context, cardID uint, point cardflow.QueuePoint, settings cardapp.TrafficQueueSettings) ([]cardflow.QueuePoint, bool, error) {
|
||
if q == nil || q.redis == nil {
|
||
return nil, false, errors.New(errors.CodeInternalError, "卡流量采集队列未配置")
|
||
}
|
||
day := cardflow.ShanghaiDate(point.ObservedAt)
|
||
clear := "0"
|
||
if settings.ClearOnNewDay {
|
||
clear = "1"
|
||
}
|
||
maxAgeMS := int64(settings.MaxPointAgeHours) * int64(time.Hour/time.Millisecond)
|
||
result, err := trafficQueueScript.Run(ctx, q.redis, []string{constants.RedisCardTrafficQueueKey(cardID)},
|
||
point.ObservedAt.UnixMilli(), strconv.FormatFloat(point.ReadingMB, 'f', -1, 64), day.Format("20060102"),
|
||
settings.Size, clear, maxAgeMS, queueTTL(point.ObservedAt, settings).Milliseconds(),
|
||
).Slice()
|
||
if err != nil {
|
||
return nil, false, errors.Wrap(errors.CodeRedisError, err, "更新卡流量采集队列失败")
|
||
}
|
||
if len(result) != 2 {
|
||
return nil, false, errors.New(errors.CodeRedisError, "卡流量采集队列返回格式异常")
|
||
}
|
||
encoded, _ := result[0].(string)
|
||
var stored []queuedPoint
|
||
if encoded != "" && encoded != "{}" {
|
||
if err := json.Unmarshal([]byte(encoded), &stored); err != nil {
|
||
return nil, false, errors.Wrap(errors.CodeRedisError, err, "解析卡流量采集队列失败")
|
||
}
|
||
}
|
||
points := make([]cardflow.QueuePoint, 0, len(stored))
|
||
for _, p := range stored {
|
||
points = append(points, cardflow.QueuePoint{ObservedAt: time.UnixMilli(p.T), ReadingMB: p.V})
|
||
}
|
||
tail, _ := result[1].(int64)
|
||
return points, tail == 1, nil
|
||
}
|
||
|
||
// queueTTL 返回队列 key 的过期时长：跨天清空开启时到次日 01:00（上海时间），关闭时为最长点龄（不限时 7 天）。
|
||
func queueTTL(observedAt time.Time, settings cardapp.TrafficQueueSettings) time.Duration {
|
||
if settings.ClearOnNewDay {
|
||
expireAt := cardflow.ShanghaiDate(observedAt).AddDate(0, 0, 1).Add(time.Hour)
|
||
if ttl := time.Until(expireAt); ttl > time.Minute {
|
||
return ttl
|
||
}
|
||
return time.Hour
|
||
}
|
||
hours := settings.MaxPointAgeHours
|
||
if hours <= 0 {
|
||
hours = constants.TrafficQueueUnlimitedTTLHours
|
||
}
|
||
return time.Duration(hours) * time.Hour
|
||
}
|