121 lines
3.9 KiB
Go
121 lines
3.9 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/global"
|
|
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/mysqldao"
|
|
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/mysqlmodel"
|
|
)
|
|
|
|
// 额度凭证的「到期」原本只是惰性判断（读取时用 grantExpired 现算），既不落流水也不做任何动作，
|
|
// 导致用户看到的额度明细对不上账：+3 之后某个时刻剩余无故减少，却没有任何对应记录。
|
|
//
|
|
// 本任务为已过期的「固定有效期」凭证补记一条 change_type=7（到期失效）流水，让明细闭环。
|
|
// 注意：只补流水，不删除凭证 —— 凭证本身保留，读取时依旧由 grantExpired 过滤。
|
|
const (
|
|
// quotaExpireScanInterval 任务执行间隔。
|
|
quotaExpireScanInterval = time.Hour
|
|
|
|
// quotaExpireScanBatch 单批处理条数。
|
|
quotaExpireScanBatch = 500
|
|
|
|
// quotaExpireScanMaxRounds 单次执行最大批次轮数。
|
|
quotaExpireScanMaxRounds = 20
|
|
)
|
|
|
|
// StartQuotaExpireScheduler 启动额度到期补记流水任务（启动即执行一次，之后每小时一次）。
|
|
func StartQuotaExpireScheduler() {
|
|
go func() {
|
|
slog.Info("quota expire scheduler started", "interval", quotaExpireScanInterval.String())
|
|
for {
|
|
func() {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
slog.Error("quota expire scheduler panic", "recover", r)
|
|
}
|
|
}()
|
|
runQuotaExpire()
|
|
}()
|
|
time.Sleep(quotaExpireScanInterval)
|
|
}
|
|
}()
|
|
}
|
|
|
|
// runQuotaExpire 执行一次到期流水补记。
|
|
// 天然幂等（靠流水唯一索引 uk_quota_flow_idem 去重），多实例并发执行无害，无需分布式锁。
|
|
func runQuotaExpire() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
|
|
ctx, span := global.Tracer.Start(ctx, "scheduler.QuotaExpire")
|
|
defer span.End()
|
|
|
|
grantDao := mysqldao.NewQuotaGrantDao()
|
|
now := nowUnix()
|
|
|
|
var total int64
|
|
for round := 0; round < quotaExpireScanMaxRounds; round++ {
|
|
grants, err := grantDao.QuotaGrantListExpiredFixed(ctx, now, quotaExpireScanBatch)
|
|
if err != nil {
|
|
slog.ErrorContext(ctx, "list expired quota grants failed", "error", err, "logged", total)
|
|
return
|
|
}
|
|
if len(grants) == 0 {
|
|
break
|
|
}
|
|
|
|
for i := range grants {
|
|
logOneExpireFlow(ctx, grantDao, &grants[i])
|
|
}
|
|
|
|
total += int64(len(grants))
|
|
if len(grants) < quotaExpireScanBatch {
|
|
break
|
|
}
|
|
}
|
|
|
|
if total > 0 {
|
|
slog.InfoContext(ctx, "logged expired quota flows", "count", total)
|
|
} else {
|
|
slog.DebugContext(ctx, "quota expire: nothing to log")
|
|
}
|
|
}
|
|
|
|
// logOneExpireFlow 为单条已过期凭证补记「到期失效」流水。
|
|
func logOneExpireFlow(ctx context.Context, grantDao *mysqldao.QuotaGrantDao, g *mysqlmodel.QuotaGrant) {
|
|
// 只有「没用完」的部分才是真正失效损失的额度；已消耗的不再变化。
|
|
expireAmount := g.Amount - g.Used
|
|
if expireAmount < 0 {
|
|
expireAmount = 0
|
|
}
|
|
|
|
// 凭证此刻已过期，getEffectiveQuota 返回的正是「失效后」的真实剩余；
|
|
// 用它反推变更前剩余，保证明细里满足「变更前 - 变更量 = 变更后」。
|
|
newValue := int64(0)
|
|
if eff, qerr := getEffectiveQuota(ctx, g.UserID, g.QuotaType); qerr == nil {
|
|
newValue = eff.Remain
|
|
}
|
|
|
|
flow := &mysqlmodel.QuotaFlowLog{
|
|
QuotaFlowUUID: genUUID(),
|
|
UserID: g.UserID,
|
|
ChangeTarget: QuotaChangeTargetAdjust,
|
|
ChangeType: QuotaChangeTypeExpire,
|
|
ChangeNum: -expireAmount,
|
|
OldValue: newValue + expireAmount,
|
|
NewValue: newValue,
|
|
// 与 quota_type、change_type 一起命中 uk_quota_flow_idem，天然幂等，无需额外标记字段。
|
|
OrderNo: g.RefID,
|
|
QuotaType: g.QuotaType,
|
|
Reason: "额度凭证到期失效",
|
|
}
|
|
|
|
if _, err := grantDao.QuotaFlowLogCreateIgnoreDup(ctx, flow); err != nil {
|
|
slog.WarnContext(ctx, "write quota expire flow failed",
|
|
"grant_id", g.ID, "user_id", g.UserID, "quota_type", g.QuotaType, "error", err)
|
|
}
|
|
}
|