first commit

2026-09-11 01:41:52 +08:00
commit 9f22e6ccae
263 changed files with 28935 additions and 0 deletions
@@ -0,0 +1,112 @@
+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"
+)
+
+// 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 || !ok {
+		return
+	}
+	defer redisdao.Unlock(context.Background(), lockKey, lockVal)
+
+	taskDao := mysqldao.NewBenefitGrantTaskDao()
+
+	// 恢复超时的「发放中」任务（认领后崩溃的场景），使其回到待发放继续重试。
+	_, _ = taskDao.BenefitGrantTaskRecoverStale(ctx, nowUnix()-int64((10*time.Minute).Seconds()))
+
+	tasks, err := taskDao.BenefitGrantTaskListPending(ctx, 100)
+	if err != nil {
+		slog.ErrorContext(ctx, "list pending benefit tasks failed", "error", err)
+		return
+	}
+	if len(tasks) == 0 {
+		return
+	}
+
+	s := NewOrderService()
+	for i := range tasks {
+		s.retryBenefitTask(ctx, &tasks[i])
+	}
+}
+
+// retryBenefitTask 重试单个补偿任务（幂等：先查发放流水避免重复发放）。
+func (s *OrderService) retryBenefitTask(ctx context.Context, task *mysqlmodel.BenefitGrantTask) {
+	taskDao := mysqldao.NewBenefitGrantTaskDao()
+
+	// 重试次数超限，标记失败并告警，等待人工介入。
+	if task.RetryCount >= task.MaxRetry {
+		_ = taskDao.BenefitGrantTaskUpdateByID(ctx, task.ID, map[string]any{"status": BenefitTaskFailed})
+		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 {
+		_ = taskDao.BenefitGrantTaskUpdateByID(ctx, task.ID, map[string]any{"status": BenefitTaskSuccess})
+		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
+	}
+}