201 lines
8.0 KiB
Go
201 lines
8.0 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"
|
|
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/redisdao"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
// StartBenefitCompensationScheduler 启动权益发放补偿定时任务：周期性重试发放失败的权益。
|
|
func StartBenefitCompensationScheduler() {
|
|
go func() {
|
|
slog.Info("benefit compensation scheduler started")
|
|
for {
|
|
func() {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
slog.Error("benefit compensation scheduler panic", "recover", r)
|
|
}
|
|
}()
|
|
runBenefitCompensation()
|
|
}()
|
|
time.Sleep(time.Minute)
|
|
}
|
|
}()
|
|
}
|
|
|
|
// runBenefitCompensation 扫描待发放补偿任务并重试，单实例通过 Redis 分布式锁防重。
|
|
func runBenefitCompensation() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
|
|
defer cancel()
|
|
|
|
ctx, span := global.Tracer.Start(ctx, "scheduler.BenefitCompensation")
|
|
defer span.End()
|
|
|
|
lockKey := "ztcloset:lock:benefit:compensation"
|
|
lockVal := genUUID()
|
|
ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, time.Minute)
|
|
if lerr != nil {
|
|
slog.ErrorContext(ctx, "benefit compensation: acquire lock failed", "error", lerr)
|
|
return
|
|
}
|
|
if !ok {
|
|
slog.DebugContext(ctx, "benefit compensation: another instance running, skip")
|
|
return
|
|
}
|
|
defer func() {
|
|
if uerr := redisdao.Unlock(context.Background(), lockKey, lockVal); uerr != nil {
|
|
slog.WarnContext(ctx, "benefit compensation: release lock failed", "error", uerr)
|
|
}
|
|
}()
|
|
|
|
taskDao := mysqldao.NewBenefitGrantTaskDao()
|
|
s := NewOrderService()
|
|
// 对账已支付但任务未完整落库的订单，覆盖支付状态提交后进程崩溃/数据库瞬时错误。
|
|
if orders, err := mysqldao.NewOrderDao().OrderListPaidMissingBenefitTasks(ctx, 100); err == nil {
|
|
for i := range orders {
|
|
s.grantOrderBenefits(ctx, &orders[i])
|
|
}
|
|
} else {
|
|
slog.ErrorContext(ctx, "reconcile paid order benefits failed", "error", err)
|
|
}
|
|
|
|
// 恢复超时的「发放中」任务（认领后崩溃的场景），使其回到待发放继续重试。
|
|
recovered, rerr := taskDao.BenefitGrantTaskRecoverStale(ctx, nowUnix()-int64((10*time.Minute).Seconds()))
|
|
if rerr != nil {
|
|
slog.ErrorContext(ctx, "benefit compensation: recover stale grant tasks failed", "error", rerr)
|
|
} else if recovered > 0 {
|
|
slog.InfoContext(ctx, "benefit compensation: recovered stale grant tasks", "count", recovered)
|
|
}
|
|
|
|
tasks, err := taskDao.BenefitGrantTaskListPending(ctx, 100)
|
|
if err != nil {
|
|
slog.ErrorContext(ctx, "list pending benefit tasks failed", "error", err)
|
|
return
|
|
}
|
|
for i := range tasks {
|
|
s.retryBenefitTask(ctx, &tasks[i])
|
|
}
|
|
|
|
revokeDao := mysqldao.NewBenefitRevokeTaskDao()
|
|
if rerr := revokeDao.RecoverStale(ctx, nowUnix()-int64((10*time.Minute).Seconds())); rerr != nil {
|
|
slog.ErrorContext(ctx, "benefit compensation: recover stale revoke tasks failed", "error", rerr)
|
|
}
|
|
if revokeTasks, err := revokeDao.ListPending(ctx, 100); err == nil {
|
|
for i := range revokeTasks {
|
|
s.executeRevokeTask(ctx, &revokeTasks[i])
|
|
}
|
|
} else {
|
|
slog.ErrorContext(ctx, "list pending benefit revoke tasks failed", "error", err)
|
|
}
|
|
}
|
|
|
|
// retryBenefitTask 重试单个补偿任务（幂等：先查发放流水避免重复发放）。
|
|
func (s *OrderService) retryBenefitTask(ctx context.Context, task *mysqlmodel.BenefitGrantTask) {
|
|
taskDao := mysqldao.NewBenefitGrantTaskDao()
|
|
|
|
// 重试次数超限，标记失败并告警，等待人工介入。
|
|
if task.RetryCount >= task.MaxRetry {
|
|
if uerr := taskDao.BenefitGrantTaskUpdateByID(ctx, task.ID, map[string]any{"status": BenefitTaskFailed}); uerr != nil {
|
|
slog.ErrorContext(ctx, "benefit compensation: mark exhausted task failed status failed",
|
|
"order_no", task.OrderNo, "benefit_type", task.BenefitType, "error", uerr)
|
|
}
|
|
slog.ErrorContext(ctx, "benefit grant task exhausted retries",
|
|
"order_no", task.OrderNo, "benefit_type", task.BenefitType, "retry_count", task.RetryCount)
|
|
return
|
|
}
|
|
|
|
// 幂等判断：若发放流水已存在（上次发放成功但未置 success 的崩溃窗口），直接置成功。
|
|
if granted, err := s.benefitAlreadyGranted(ctx, task); err == nil && granted {
|
|
if uerr := taskDao.BenefitGrantTaskUpdateByID(ctx, task.ID, map[string]any{"status": BenefitTaskSuccess}); uerr != nil {
|
|
slog.ErrorContext(ctx, "benefit compensation: mark already-granted task success failed",
|
|
"order_no", task.OrderNo, "benefit_type", task.BenefitType, "error", uerr)
|
|
}
|
|
slog.InfoContext(ctx, "benefit already granted, task marked success",
|
|
"order_no", task.OrderNo, "benefit_type", task.BenefitType)
|
|
return
|
|
} else if err != nil {
|
|
slog.ErrorContext(ctx, "check benefit granted failed",
|
|
"order_no", task.OrderNo, "benefit_type", task.BenefitType, "error", err)
|
|
return
|
|
}
|
|
|
|
// 执行发放（claim 防并发，内部处理 retry_count 原子自增与状态流转）。
|
|
s.executeGrant(ctx, task)
|
|
}
|
|
|
|
// benefitAlreadyGranted 判断该权益是否已发放（用于崩溃恢复幂等）。
|
|
// VIP 查 VIP 流水；额度类查额度凭证（quota_grants）。
|
|
func (s *OrderService) benefitAlreadyGranted(ctx context.Context, task *mysqlmodel.BenefitGrantTask) (bool, error) {
|
|
grantDao := mysqldao.NewQuotaGrantDao()
|
|
switch task.BenefitType {
|
|
case BenefitTypeVIP:
|
|
return s.vipDao.VIPFlowLogExistsByOrderNo(ctx, task.OrderNo)
|
|
case BenefitTypeCos:
|
|
return grantDao.QuotaGrantExistsByRefIDAndType(ctx, task.OrderNo, QuotaTypeCos)
|
|
case BenefitTypeAI:
|
|
return grantDao.QuotaGrantExistsByRefIDAndType(ctx, task.OrderNo, QuotaTypeAI)
|
|
case BenefitTypeClosetShare:
|
|
return grantDao.QuotaGrantExistsByRefIDAndType(ctx, task.OrderNo, QuotaTypeClosetShare)
|
|
case BenefitTypeOutfitShare:
|
|
return grantDao.QuotaGrantExistsByRefIDAndType(ctx, task.OrderNo, QuotaTypeOutfitShare)
|
|
default:
|
|
return false, nil
|
|
}
|
|
}
|
|
|
|
func (s *OrderService) executeRevokeTaskByOrder(ctx context.Context, orderNo string) {
|
|
task, err := mysqldao.NewBenefitRevokeTaskDao().GetByOrderNo(ctx, orderNo)
|
|
if err != nil {
|
|
if isNotFound(err) {
|
|
return // 无回收任务（如未退款订单），属正常路径。
|
|
}
|
|
slog.ErrorContext(ctx, "get benefit revoke task failed", "order_no", orderNo, "error", err)
|
|
return
|
|
}
|
|
s.executeRevokeTask(ctx, task)
|
|
}
|
|
|
|
func (s *OrderService) executeRevokeTask(ctx context.Context, task *mysqlmodel.BenefitRevokeTask) {
|
|
dao := mysqldao.NewBenefitRevokeTaskDao()
|
|
if task.Status == BenefitTaskSuccess {
|
|
return
|
|
}
|
|
if task.RetryCount >= task.MaxRetry {
|
|
slog.ErrorContext(ctx, "benefit revoke task exceeded normal retry threshold; continuing for eventual consistency", "order_no", task.OrderNo, "retry_count", task.RetryCount)
|
|
}
|
|
claimed, err := dao.Claim(ctx, task.ID)
|
|
if err != nil {
|
|
slog.ErrorContext(ctx, "claim benefit revoke task failed", "order_no", task.OrderNo, "error", err)
|
|
return
|
|
}
|
|
if !claimed {
|
|
slog.DebugContext(ctx, "benefit revoke task already claimed, skip", "order_no", task.OrderNo)
|
|
return
|
|
}
|
|
order, oerr := s.orderDao.OrderGetByOrderNo(ctx, task.OrderNo)
|
|
if oerr != nil {
|
|
_ = dao.Update(ctx, task.ID, map[string]any{"status": BenefitTaskPending, "retry_count": gorm.Expr("retry_count + 1"), "last_error": oerr.Error()})
|
|
slog.ErrorContext(ctx, "revoke refunded order benefits failed: get order failed",
|
|
"order_no", task.OrderNo, "retry_count", task.RetryCount, "error", oerr)
|
|
return
|
|
}
|
|
if rerr := s.revokeOrderBenefits(ctx, order); rerr != nil {
|
|
_ = dao.Update(ctx, task.ID, map[string]any{"status": BenefitTaskPending, "retry_count": gorm.Expr("retry_count + 1"), "last_error": rerr.Error()})
|
|
slog.ErrorContext(ctx, "revoke refunded order benefits failed",
|
|
"order_no", task.OrderNo, "retry_count", task.RetryCount, "error", rerr)
|
|
return
|
|
}
|
|
if uerr := dao.Update(ctx, task.ID, map[string]any{"status": BenefitTaskSuccess, "last_error": ""}); uerr != nil {
|
|
slog.ErrorContext(ctx, "mark benefit revoke task success failed", "order_no", task.OrderNo, "error", uerr)
|
|
return
|
|
}
|
|
slog.InfoContext(ctx, "refunded order benefits revoked", "order_no", task.OrderNo, "user_id", order.UserID)
|
|
}
|