huang
cebcada950
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m16s
核心改造： - 增量算法：流量计算从覆盖式改为增量累加（gateway - lastReading），支持上游运营商重置检测 - 日流量缓冲：insertDataUsageRecord 改为 Redis INCRBYFLOAT，每日凌晨落盘到 tb_card_daily_usage - 运营商：新增 data_reset_day 字段（联通=27，其余=1） - IoT卡：新增 last_gateway_reading_mb 字段存储上次网关读数 - 查询层：新建 TrafficQueryService 合并 Redis（今日）+ DB（历史）数据源 - 清理：删除 DataUsageRecord model/store，移除 polling_handler 旧引用 迁移：000094-000097（carrier字段、iot_card字段、数据初始化、日流量表）
92 lines
2.8 KiB
Go
92 lines
2.8 KiB
Go
// Package traffic 提供流量数据查询服务
|
||
// 合并 Redis 缓冲（今日）和 DB（历史）两段数据源，对外提供统一的日粒度流量查询
|
||
package traffic
|
||
|
||
import (
|
||
"context"
|
||
"sort"
|
||
"strconv"
|
||
"time"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/redis/go-redis/v9"
|
||
)
|
||
|
||
// DailyUsageItem 日粒度流量记录
|
||
type DailyUsageItem struct {
|
||
Date string `json:"date"` // 日期，格式 YYYY-MM-DD
|
||
UsageMB float64 `json:"usage_mb"` // 当天累计使用流量(MB)
|
||
}
|
||
|
||
// QueryService 流量查询服务
|
||
// 合并 Redis（当天实时）和 DB（历史落盘）两段数据源
|
||
type QueryService struct {
|
||
redis *redis.Client
|
||
cardDailyUsageStore *postgres.CardDailyUsageStore
|
||
}
|
||
|
||
// NewQueryService 创建流量查询服务实例
|
||
func NewQueryService(redisClient *redis.Client, cardDailyUsageStore *postgres.CardDailyUsageStore) *QueryService {
|
||
return &QueryService{
|
||
redis: redisClient,
|
||
cardDailyUsageStore: cardDailyUsageStore,
|
||
}
|
||
}
|
||
|
||
// GetDailyUsage 查询卡在指定日期范围内的日粒度流量
|
||
// 今天的数据从 Redis 缓冲读取（可能尚未落盘），历史数据从 DB 读取，合并后按日期 ASC 排序返回
|
||
func (s *QueryService) GetDailyUsage(ctx context.Context, cardID uint, startDate, endDate time.Time) ([]*DailyUsageItem, error) {
|
||
now := time.Now()
|
||
todayStr := now.Format("2006-01-02")
|
||
|
||
// 判断结束日期是否包含今天
|
||
endDateStr := endDate.Format("2006-01-02")
|
||
includeToday := endDateStr >= todayStr
|
||
|
||
// 历史部分：从 DB 查询（排除今天，因为今天的数据可能还在 Redis 中）
|
||
dbEndDate := endDate
|
||
if includeToday {
|
||
// DB 只查到昨天，今天从 Redis 取
|
||
dbEndDate = now.AddDate(0, 0, -1)
|
||
}
|
||
|
||
var items []*DailyUsageItem
|
||
|
||
// 只有 startDate <= dbEndDate 时才查 DB
|
||
if !startDate.After(dbEndDate) {
|
||
records, err := s.cardDailyUsageStore.ListByCardIDAndDateRange(ctx, cardID, startDate, dbEndDate)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
for _, r := range records {
|
||
items = append(items, &DailyUsageItem{
|
||
Date: r.Date.Format("2006-01-02"),
|
||
UsageMB: r.UsageMB,
|
||
})
|
||
}
|
||
}
|
||
|
||
// 今天部分：从 Redis 读取
|
||
if includeToday && !startDate.After(now) {
|
||
key := constants.RedisCardDailyTrafficKey(cardID, todayStr)
|
||
val, err := s.redis.Get(ctx, key).Result()
|
||
if err == nil && val != "" {
|
||
if usageMB, parseErr := strconv.ParseFloat(val, 64); parseErr == nil && usageMB > 0 {
|
||
items = append(items, &DailyUsageItem{
|
||
Date: todayStr,
|
||
UsageMB: usageMB,
|
||
})
|
||
}
|
||
}
|
||
// Redis key 不存在（redis.Nil）时忽略，表示今天暂无流量
|
||
}
|
||
|
||
// 按日期 ASC 排序
|
||
sort.Slice(items, func(i, j int) bool {
|
||
return items[i].Date < items[j].Date
|
||
})
|
||
|
||
return items, nil
|
||
}
|