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>
95 lines
4.7 KiB
Go
95 lines
4.7 KiB
Go
package postgres
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"gorm.io/gorm"
|
||
"gorm.io/gorm/clause"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
// effectivePackageStatuses 是"采集时刻有效套餐"认可的套餐状态：状态可能在采集之后才变为已用完或已过期，
|
||
// 是否在采集时刻有效由激活时间与到期时间判断。
|
||
var effectivePackageStatuses = []int{
|
||
constants.PackageUsageStatusActive, constants.PackageUsageStatusDepleted, constants.PackageUsageStatusExpired,
|
||
}
|
||
|
||
// EffectivePackages 是一次"采集时刻有效套餐"查询的结果。
|
||
type EffectivePackages struct {
|
||
// CarrierType 与 CarrierID 是实际扣减载体：卡自身没有有效套餐时回退到卡当前绑定的设备。
|
||
CarrierType string
|
||
CarrierID uint
|
||
// Packages 按扣减优先级排序：加油包优先，再按 priority、id。
|
||
Packages []*model.PackageUsage
|
||
}
|
||
|
||
// FindEffectivePackagesAt 按"有套餐"共用口径查询卡在采集时刻的有效套餐：状态为生效中、已用完或已过期，
|
||
// 激活时间不晚于采集时间（为空按已生效），到期时间为空或晚于采集时间，未退款；
|
||
// 卡自身一个都没有时看卡当前绑定设备的套餐。lock 为 true 时对查到的套餐行加 FOR UPDATE。
|
||
// 采集侧（写入范围、是否发事件、采集记录、日用量 0 行）与扣费侧共用本函数，避免两边口径不一致。
|
||
func FindEffectivePackagesAt(ctx context.Context, db *gorm.DB, cardID uint, observedAt time.Time, lock bool) (EffectivePackages, error) {
|
||
own, err := queryEffectivePackages(ctx, db, "iot_card_id = ?", cardID, observedAt, lock)
|
||
if err != nil || len(own) > 0 {
|
||
return EffectivePackages{CarrierType: constants.AssetTypeIotCard, CarrierID: cardID, Packages: own}, err
|
||
}
|
||
var deviceIDs []uint
|
||
if err := db.WithContext(ctx).Model(&model.DeviceSimBinding{}).
|
||
Where("iot_card_id = ? AND bind_status = ?", cardID, constants.BindStatusBound).
|
||
Limit(1).Pluck("device_id", &deviceIDs).Error; err != nil {
|
||
return EffectivePackages{}, err
|
||
}
|
||
if len(deviceIDs) == 0 {
|
||
return EffectivePackages{CarrierType: constants.AssetTypeIotCard, CarrierID: cardID}, nil
|
||
}
|
||
devicePackages, err := queryEffectivePackages(ctx, db, "device_id = ?", deviceIDs[0], observedAt, lock)
|
||
if err != nil {
|
||
return EffectivePackages{}, err
|
||
}
|
||
if len(devicePackages) == 0 {
|
||
return EffectivePackages{CarrierType: constants.AssetTypeIotCard, CarrierID: cardID}, nil
|
||
}
|
||
return EffectivePackages{CarrierType: constants.AssetTypeDevice, CarrierID: deviceIDs[0], Packages: devicePackages}, nil
|
||
}
|
||
|
||
func queryEffectivePackages(ctx context.Context, db *gorm.DB, carrierCondition string, carrierID uint, observedAt time.Time, lock bool) ([]*model.PackageUsage, error) {
|
||
query := db.WithContext(ctx).Model(&model.PackageUsage{}).
|
||
Where(carrierCondition, carrierID).
|
||
Where("status IN ?", effectivePackageStatuses).
|
||
Where("activated_at IS NULL OR activated_at <= ?", observedAt).
|
||
Where("expires_at IS NULL OR expires_at > ?", observedAt).
|
||
Where("refund_id IS NULL")
|
||
if lock {
|
||
query = query.Clauses(clause.Locking{Strength: "UPDATE"})
|
||
}
|
||
var packages []*model.PackageUsage
|
||
err := query.Order("CASE WHEN master_usage_id IS NOT NULL THEN 0 ELSE 1 END, priority ASC, id ASC").Find(&packages).Error
|
||
return packages, err
|
||
}
|
||
|
||
// EnsureDailyUsageRows 为每个套餐确保 date 当天一行套餐日用量（没有则写 0，已有不动）。
|
||
func EnsureDailyUsageRows(ctx context.Context, db *gorm.DB, packageUsageIDs []uint, date time.Time, cumulative map[uint]int64) error {
|
||
for _, id := range packageUsageIDs {
|
||
if err := db.WithContext(ctx).Exec(`
|
||
INSERT INTO tb_package_usage_daily_record (package_usage_id, date, daily_usage_mb, cumulative_usage_mb, created_at, updated_at)
|
||
VALUES (?, ?::date, 0, ?, NOW(), NOW())
|
||
ON CONFLICT (package_usage_id, date) DO NOTHING`, id, date.Format(cardFlowDateLayout), cumulative[id]).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// AddDailyUsage 原子累加套餐 date 当天用量并覆盖累计用量；当天没有行时新建。
|
||
func AddDailyUsage(ctx context.Context, db *gorm.DB, packageUsageID uint, date time.Time, dailyMB int64, cumulativeMB int64) error {
|
||
return db.WithContext(ctx).Exec(`
|
||
INSERT INTO tb_package_usage_daily_record (package_usage_id, date, daily_usage_mb, cumulative_usage_mb, created_at, updated_at)
|
||
VALUES (?, ?::date, ?, ?, NOW(), NOW())
|
||
ON CONFLICT (package_usage_id, date) DO UPDATE SET
|
||
daily_usage_mb = tb_package_usage_daily_record.daily_usage_mb + EXCLUDED.daily_usage_mb,
|
||
cumulative_usage_mb = EXCLUDED.cumulative_usage_mb,
|
||
updated_at = NOW()`, packageUsageID, date.Format(cardFlowDateLayout), dailyMB, cumulativeMB).Error
|
||
}
|