zhengchiliu/zt-closet-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

补齐相关日志日志

Browse Source
This commit is contained in:
zhengchiliu
2026-09-16 00:23:49 +08:00
parent 8c2ea84939
commit b2e5eef3c8
39 changed files with 1576 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/service/ai_outfit_cleanup.go
+2
View File
@@ -78,5 +78,7 @@ func runAIOutfitCleanup() {
"count", total,
"retention_hours", int(aiUnsavedOutfitRetention.Hours()),
)
} else {
slog.DebugContext(ctx, "ai outfit cleanup: nothing to clean")
}
}
internal/service/airecommend.go
+44 -8
View File
@@ -67,13 +67,17 @@ func (s *AIRecommendService) Submit(ctx context.Context, userID int64, request *
// 1. 抢占活跃位：抢不到说明该用户已有任务在跑，幂等返回它。
acquired, runningTaskID, err := redisdao.AIRecommendTaskAcquire(ctx, userID, taskID, aiRecommendSlotTTL)
if err != nil {
slog.ErrorContext(ctx, "ai recommend submit: acquire slot failed", "user_id", userID, "error", err)
return nil, errcode.CacheError
}
if !acquired {
if runningTaskID == "" {
// 活跃位恰好在两次调用之间过期，让前端稍后重试即可。
slog.InfoContext(ctx, "ai recommend submit: slot expired between calls, ask retry", "user_id", userID)
return nil, errcode.TooManyRequests
}
slog.InfoContext(ctx, "ai recommend submit: task already running, return existing",
"user_id", userID, "task_id", runningTaskID)
return &dto.AIRecommendSubmitResponse{
TaskID: runningTaskID,
Status: AIRecommendTaskPending,
@@ -84,8 +88,10 @@ func (s *AIRecommendService) Submit(ctx context.Context, userID int64, request *
if qerr := consumeDailyQuota(ctx, userID, QuotaTypeAI, 1, "AI 推荐消耗"); qerr != nil {
s.releaseSlot(context.WithoutCancel(ctx), userID, taskID)
if qerr == errcode.QuotaInsufficient {
slog.InfoContext(ctx, "ai recommend submit: quota insufficient", "user_id", userID)
return nil, errcode.AIRecommendQuotaLimit
}
slog.ErrorContext(ctx, "ai recommend submit: consume quota failed", "user_id", userID, "error", qerr)
return nil, qerr
}
@@ -96,6 +102,8 @@ func (s *AIRecommendService) Submit(ctx context.Context, userID int64, request *
Status: AIRecommendTaskPending,
CreateTime: nowUnix(),
}); serr != nil {
slog.ErrorContext(ctx, "ai recommend submit: save task snapshot failed",
"user_id", userID, "task_id", taskID, "error", serr)
cleanupCtx := context.WithoutCancel(ctx)
refundDailyQuota(cleanupCtx, userID, QuotaTypeAI, 1)
s.releaseSlot(cleanupCtx, userID, taskID)
@@ -106,6 +114,8 @@ func (s *AIRecommendService) Submit(ctx context.Context, userID int64, request *
reqCopy := *request
go s.runTask(taskID, userID, &reqCopy)
slog.InfoContext(ctx, "ai recommend submitted", "user_id", userID, "task_id", taskID)
return &dto.AIRecommendSubmitResponse{
TaskID: taskID,
Status: AIRecommendTaskPending,
@@ -153,26 +163,39 @@ func (s *AIRecommendService) Save(ctx context.Context, userID int64, request *dt
outfit, err := s.outfitDao.OutfitGetByUUID(ctx, request.OutfitUUID)
if err != nil {
if isNotFound(err) {
slog.WarnContext(ctx, "ai recommend save: outfit not found",
"user_id", userID, "outfit_uuid", request.OutfitUUID)
return errcode.OutfitNotFound
}
slog.ErrorContext(ctx, "ai recommend save: query outfit failed",
"user_id", userID, "outfit_uuid", request.OutfitUUID, "error", err)
return errcode.DatabaseError
}
if outfit.UserID != userID {
slog.InfoContext(ctx, "ai recommend save: permission denied",
"user_id", userID, "outfit_uuid", request.OutfitUUID)
return errcode.OutfitNoPermission
}
// 仅待保存状态可保存；已正常则幂等返回。
if outfit.Status == OutfitStatusNormal {
slog.DebugContext(ctx, "ai recommend save: outfit already saved, skip",
"user_id", userID, "outfit_uuid", request.OutfitUUID)
return nil
}
if outfit.Status != OutfitStatusUnsaved {
slog.WarnContext(ctx, "ai recommend save: outfit status not saveable",
"user_id", userID, "outfit_uuid", request.OutfitUUID, "status", outfit.Status)
return errcode.OutfitUpdateFailed
}
if uerr := s.outfitDao.OutfitUpdateByUUID(ctx, request.OutfitUUID, map[string]any{
"status": OutfitStatusNormal,
}); uerr != nil {
slog.ErrorContext(ctx, "ai recommend save: update outfit failed",
"user_id", userID, "outfit_uuid", request.OutfitUUID, "error", uerr)
return errcode.OutfitUpdateFailed
}
slog.InfoContext(ctx, "ai recommend outfit saved", "user_id", userID, "outfit_uuid", request.OutfitUUID)
return nil
}
@@ -196,7 +219,7 @@ func (s *AIRecommendService) runTask(taskID string, userID int64, request *dto.A
s.releaseSlot(context.WithoutCancel(ctx), userID, taskID)
}()
out, bizErr := s.generate(ctx, userID, request)
out, bizErr := s.generate(ctx, taskID, userID, request)
if bizErr != nil {
s.finishFailed(ctx, taskID, userID, bizErr.Code(), formatBizErr(bizErr))
return
@@ -206,11 +229,11 @@ func (s *AIRecommendService) runTask(taskID string, userID int64, request *dto.A
TaskID: taskID,
UserID: userID,
Status: AIRecommendTaskSuccess,
Result: mustMarshal(out),
Result: mustMarshal(ctx, out),
CreateTime: nowUnix(),
}); serr != nil {
// 结果已落库为主，快照写失败只影响前端轮询展示，记录即可。
slog.ErrorContext(ctx, "save ai recommend task result failed",
// 结果已落库为主，快照写失败只影响前端轮询展示，降为 Warn。
slog.WarnContext(ctx, "save ai recommend task result failed",
"task_id", taskID, "user_id", userID, "error", serr)
}
}
@@ -249,13 +272,16 @@ func (s *AIRecommendService) releaseSlot(ctx context.Context, userID int64, task
}
// generate 执行一次完整的推荐生成：查衣橱 → 取运势 → 调大模型 → 校验 → 落库。
func (s *AIRecommendService) generate(ctx context.Context, userID int64, request *dto.AIRecommendCreateRequest) (*dto.AIRecommendCreateResponse, *errcode.Error) {
func (s *AIRecommendService) generate(ctx context.Context, taskID string, userID int64, request *dto.AIRecommendCreateRequest) (*dto.AIRecommendCreateResponse, *errcode.Error) {
// 1. 加载用户可用衣物（已成功处理）。
clothes, cerr := s.processedClothes(ctx, userID)
if cerr != nil {
slog.ErrorContext(ctx, "ai recommend generate: load clothes failed",
"user_id", userID, "task_id", taskID, "error", cerr)
return nil, errcode.DatabaseError
}
if len(clothes) == 0 {
slog.InfoContext(ctx, "ai recommend generate: no usable clothes", "user_id", userID, "task_id", taskID)
return nil, errcode.AIRecommendFailed.WithDetails("衣橱暂无可用衣物")
}
@@ -267,11 +293,13 @@ func (s *AIRecommendService) generate(ctx context.Context, userID int64, request
client := tokenhub.NewClient()
result, gerr := client.GenerateOutfitRecommendation(ctx, zodiac, fortune, clothes, request.Scene)
if gerr != nil {
slog.ErrorContext(ctx, "generate outfit recommendation failed", "user_id", userID, "error", gerr)
slog.ErrorContext(ctx, "generate outfit recommendation failed", "user_id", userID, "task_id", taskID, "error", gerr)
return nil, errcode.AIRecommendFailed.WithDetails("大模型调用失败")
}
result.ValidateClothesIDs(clothes)
if len(result.Items) == 0 {
slog.WarnContext(ctx, "ai recommend generate: no valid items after validation",
"user_id", userID, "task_id", taskID)
return nil, errcode.AIRecommendFailed.WithDetails("未生成有效搭配")
}
@@ -302,6 +330,8 @@ func (s *AIRecommendService) generate(ctx context.Context, userID int64, request
}
return tx.Create(&items).Error
}); err != nil {
slog.ErrorContext(ctx, "ai recommend generate: save outfit failed",
"user_id", userID, "task_id", taskID, "error", err)
return nil, errcode.OutfitCreateFailed
}
@@ -365,6 +395,9 @@ func (s *AIRecommendService) userFortune(ctx context.Context, userID int64) (str
zodiac := ""
if user, err := s.userDao.UserGetByID(ctx, userID); err == nil && user.BirthDay != 0 {
zodiac = zodiacByUnix(user.BirthDay)
} else if err != nil {
slog.WarnContext(ctx, "ai recommend generate: get user for fortune failed, degrade",
"user_id", userID, "error", err)
}
fortune := &tokenhub.FortuneItem{Zodiac: zodiac}
@@ -380,6 +413,9 @@ func (s *AIRecommendService) userFortune(ctx context.Context, userID int64) (str
LuckyNumber: h.LuckyNumber,
Description: h.DailyText,
}
} else if err != nil {
slog.WarnContext(ctx, "ai recommend generate: get horoscope failed, degrade",
"user_id", userID, "zodiac", zodiac, "error", err)
}
}
return zodiac, fortune
@@ -411,10 +447,10 @@ func formatBizErr(e *errcode.Error) string {
}
// mustMarshal 序列化任务结果；失败时返回 nil（任务仍标记成功，前端可去「我的搭配」查看）。
func mustMarshal(v any) json.RawMessage {
func mustMarshal(ctx context.Context, v any) json.RawMessage {
data, err := json.Marshal(v)
if err != nil {
slog.Error("marshal ai recommend task result failed", "error", err)
slog.ErrorContext(ctx, "marshal ai recommend task result failed", "error", err)
return nil
}
return data
internal/service/benefit_compensation.go
+60 -18
View File
@@ -41,10 +41,19 @@ func runBenefitCompensation() {
lockKey := "ztcloset:lock:benefit:compensation"
lockVal := genUUID()
ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, time.Minute)
if lerr != nil || !ok {
if lerr != nil {
slog.ErrorContext(ctx, "benefit compensation: acquire lock failed", "error", lerr)
return
}
defer redisdao.Unlock(context.Background(), lockKey, lockVal)
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()
@@ -58,7 +67,12 @@ func runBenefitCompensation() {
}
// 恢复超时的「发放中」任务（认领后崩溃的场景），使其回到待发放继续重试。
_, _ = taskDao.BenefitGrantTaskRecoverStale(ctx, nowUnix()-int64((10*time.Minute).Seconds()))
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 {
@@ -70,7 +84,9 @@ func runBenefitCompensation() {
}
revokeDao := mysqldao.NewBenefitRevokeTaskDao()
_ = revokeDao.RecoverStale(ctx, nowUnix()-int64((10*time.Minute).Seconds()))
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])
@@ -86,7 +102,10 @@ func (s *OrderService) retryBenefitTask(ctx context.Context, task *mysqlmodel.Be
// 重试次数超限，标记失败并告警，等待人工介入。
if task.RetryCount >= task.MaxRetry {
_ = taskDao.BenefitGrantTaskUpdateByID(ctx, task.ID, map[string]any{"status": BenefitTaskFailed})
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
@@ -94,7 +113,12 @@ func (s *OrderService) retryBenefitTask(ctx context.Context, task *mysqlmodel.Be
// 幂等判断：若发放流水已存在（上次发放成功但未置 success 的崩溃窗口），直接置成功。
if granted, err := s.benefitAlreadyGranted(ctx, task); err == nil && granted {
_ = taskDao.BenefitGrantTaskUpdateByID(ctx, task.ID, map[string]any{"status": BenefitTaskSuccess})
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",
@@ -128,9 +152,14 @@ func (s *OrderService) benefitAlreadyGranted(ctx context.Context, task *mysqlmod
func (s *OrderService) executeRevokeTaskByOrder(ctx context.Context, orderNo string) {
task, err := mysqldao.NewBenefitRevokeTaskDao().GetByOrderNo(ctx, orderNo)
if err == nil {
s.executeRevokeTask(ctx, task)
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) {
@@ -142,17 +171,30 @@ func (s *OrderService) executeRevokeTask(ctx context.Context, task *mysqlmodel.B
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 || !claimed {
return
}
order, err := s.orderDao.OrderGetByOrderNo(ctx, task.OrderNo)
if err == nil {
err = s.revokeOrderBenefits(ctx, order)
}
if err != nil {
_ = dao.Update(ctx, task.ID, map[string]any{"status": BenefitTaskPending, "retry_count": gorm.Expr("retry_count + 1"), "last_error": err.Error()})
slog.ErrorContext(ctx, "revoke refunded order benefits failed", "order_no", task.OrderNo, "error", err)
slog.ErrorContext(ctx, "claim benefit revoke task failed", "order_no", task.OrderNo, "error", err)
return
}
_ = dao.Update(ctx, task.ID, map[string]any{"status": BenefitTaskSuccess, "last_error": ""})
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)
}
internal/service/closet.go
+31 -3
View File
@@ -61,6 +61,7 @@ func (s *ClosetShareService) Create(ctx context.Context, userID int64, request *
case ShareTypeSnapshot:
ids, err := s.allUserClothesIDs(ctx, userID)
if err != nil {
slog.ErrorContext(ctx, "closet share create: list user clothes failed", "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
clothesIDs = ids
@@ -70,6 +71,11 @@ func (s *ClosetShareService) Create(ctx context.Context, userID int64, request *
// 2. 消耗衣柜分享每日额度。
if qerr := consumeDailyQuota(ctx, userID, QuotaTypeClosetShare, 1, "衣柜分享消耗"); qerr != nil {
if qerr == errcode.QuotaInsufficient {
slog.InfoContext(ctx, "closet share create: quota insufficient", "user_id", userID)
} else {
slog.ErrorContext(ctx, "closet share create: consume quota failed", "user_id", userID, "error", qerr)
}
return nil, qerr
}
@@ -89,6 +95,8 @@ func (s *ClosetShareService) Create(ctx context.Context, userID int64, request *
}
return nil
}); err != nil {
slog.ErrorContext(ctx, "closet share create: transaction failed",
"user_id", userID, "share_uuid", shareUUID, "error", err)
refundDailyQuota(ctx, userID, QuotaTypeClosetShare, 1)
return nil, errcode.DatabaseError
}
@@ -99,9 +107,13 @@ func (s *ClosetShareService) Create(ctx context.Context, userID int64, request *
AuditResult: AuditResultPass,
AuditTime: nowUnix(),
}); err != nil {
slog.WarnContext(ctx, "write closet audit log failed", "share_id", share.ID, "error", err)
slog.WarnContext(ctx, "write closet audit log failed",
"share_id", share.ID, "share_uuid", shareUUID, "error", err)
}
slog.InfoContext(ctx, "closet share created",
"user_id", userID, "share_uuid", shareUUID, "share_type", request.ShareType, "clothes_count", len(clothesIDs))
return &dto.ClosetShareCreateResponse{ShareUUID: shareUUID}, nil
}
@@ -115,6 +127,7 @@ func (s *ClosetShareService) Detail(ctx context.Context, userID int64, shareUUID
if isNotFound(err) {
return nil, errcode.ShareNotFound
}
slog.ErrorContext(ctx, "closet share detail: query share failed", "share_uuid", shareUUID, "error", err)
return nil, errcode.DatabaseError
}
@@ -129,7 +142,9 @@ func (s *ClosetShareService) Detail(ctx context.Context, userID int64, shareUUID
}
// 时间过期校验。
if share.ExpireTime > 0 && nowUnix() > share.ExpireTime {
_ = s.shareDao.ClosetShareUpdateStatus(ctx, shareUUID, ShareStatusExpired)
if uerr := s.shareDao.ClosetShareUpdateStatus(ctx, shareUUID, ShareStatusExpired); uerr != nil {
slog.WarnContext(ctx, "closet share detail: mark expired failed", "share_uuid", shareUUID, "error", uerr)
}
return nil, errcode.ShareExpired
}
@@ -137,7 +152,9 @@ func (s *ClosetShareService) Detail(ctx context.Context, userID int64, shareUUID
// 访问计数（访客计入）。
if !isOwner {
_, _ = redisdao.ClosetShareViewIncr(ctx, shareUUID)
if _, verr := redisdao.ClosetShareViewIncr(ctx, shareUUID); verr != nil {
slog.DebugContext(ctx, "closet share detail: incr view count failed", "share_uuid", shareUUID, "error", verr)
}
}
// 加载分享的衣物。
@@ -170,6 +187,9 @@ func (s *ClosetShareService) Detail(ctx context.Context, userID int64, shareUUID
if owner, oerr := s.userDao.UserGetByID(ctx, share.UserID); oerr == nil {
ownerName = owner.Username
ownerAvatar = upload.AvatarURL(owner.Avatar)
} else {
slog.DebugContext(ctx, "closet share detail: get owner info failed",
"share_uuid", shareUUID, "owner_id", share.UserID, "error", oerr)
}
return &dto.ClosetShareDetailResponse{
@@ -294,14 +314,21 @@ func (s *ClosetShareService) Cancel(ctx context.Context, userID int64, request *
if isNotFound(err) {
return nil, errcode.ShareNotFound
}
slog.ErrorContext(ctx, "closet share cancel: query share failed",
"share_uuid", request.ShareUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
if share.UserID != userID {
slog.InfoContext(ctx, "closet share cancel: permission denied",
"share_uuid", request.ShareUUID, "user_id", userID, "owner_id", share.UserID)
return nil, errcode.Forbidden
}
if err := s.shareDao.ClosetShareUpdateStatus(ctx, request.ShareUUID, ShareStatusCancelled); err != nil {
slog.ErrorContext(ctx, "closet share cancel: update status failed",
"share_uuid", request.ShareUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
slog.InfoContext(ctx, "closet share cancelled", "share_uuid", request.ShareUUID, "user_id", userID)
return &dto.CollectResponse{}, nil
}
@@ -321,6 +348,7 @@ func (s *ClosetShareService) resolveOwnedClothesIDs(ctx context.Context, userID
clothesList, err := s.clothesDao.ClothesGetByUUIDs(ctx, dedup)
if err != nil {
slog.ErrorContext(ctx, "closet share: query clothes failed", "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
byUUID := make(map[string]mysqlmodel.Clothes, len(clothesList))
internal/service/clothes.go
+78 -18
View File
@@ -53,36 +53,47 @@ func (s *ClothesService) Upload(ctx context.Context, userID int64, fileHeader *m
defer span.End()
if fileHeader == nil || fileHeader.Size == 0 {
slog.DebugContext(ctx, "clothes upload: empty file", "user_id", userID)
return nil, errcode.UploadFileEmpty
}
// 1. 校验扩展名与大小。
ext := upload.FileGetExt(fileHeader.Filename)
if !upload.FileCheckAllowExt(ext, global.UploadSetting.AllowedExtensions) {
slog.DebugContext(ctx, "clothes upload: extension not allowed", "user_id", userID, "ext", ext)
return nil, errcode.UploadFileExtNotAllow
}
if !upload.FileCheckMaxSize(fileHeader.Size, int64(global.UploadSetting.MaxSizeMB)) {
slog.DebugContext(ctx, "clothes upload: file too large", "user_id", userID, "size", fileHeader.Size)
return nil, errcode.UploadFileTooLarge
}
// 2. 读取原图字节（必须同步读取，请求返回后 multipart 临时文件会被清理）。
src, oerr := fileHeader.Open()
if oerr != nil {
slog.ErrorContext(ctx, "clothes upload: open file failed", "user_id", userID, "error", oerr)
return nil, errcode.UploadFileSaveFailed
}
imageData, rerr := io.ReadAll(src)
_ = src.Close()
if rerr != nil {
slog.ErrorContext(ctx, "clothes upload: read file failed", "user_id", userID, "error", rerr)
return nil, errcode.UploadFileSaveFailed
}
// 3. 校验文件内容为真实图片（magic bytes），防止伪造扩展名上传恶意文件。
if !upload.FileCheckImageMagic(imageData) {
slog.WarnContext(ctx, "clothes upload: magic bytes check failed", "user_id", userID, "size", fileHeader.Size)
return nil, errcode.UploadFileExtNotAllow
}
// 4. 扣减 COS 上传额度（权威扣减，先到期凭证先用，再扣免费额度）。
if qerr := consumeCosQuota(ctx, userID, 1, 0, "上传衣物消耗"); qerr != nil {
if qerr == errcode.QuotaInsufficient {
slog.InfoContext(ctx, "clothes upload: quota insufficient", "user_id", userID)
} else {
slog.ErrorContext(ctx, "clothes upload: consume quota failed", "user_id", userID, "error", qerr)
}
return nil, qerr
}
@@ -91,6 +102,7 @@ func (s *ClothesService) Upload(ctx context.Context, userID int64, fileHeader *m
originalName := clothesUUID + ext
originalPath, serr := upload.FileSaveTmpBytes(ctx, imageData, originalName, fmt.Sprintf("pending/%d", userID))
if serr != nil {
slog.ErrorContext(ctx, "clothes upload: save original image failed", "user_id", userID, "error", serr)
refundCosQuota(ctx, userID, 1, 0, "原图持久化失败回退")
return nil, errcode.UploadFileSaveFailed
}
@@ -102,6 +114,8 @@ func (s *ClothesService) Upload(ctx context.Context, userID int64, fileHeader *m
OriginalImagePath: originalPath,
}
if err := s.clothesDao.ClothesCreate(ctx, clothes); err != nil {
slog.ErrorContext(ctx, "clothes upload: create clothes record failed",
"user_id", userID, "clothes_uuid", clothesUUID, "error", err)
_ = os.Remove(originalPath)
// 建记录失败，回退已扣额度。
refundCosQuota(ctx, userID, 1, 0, "上传失败回退")
@@ -111,6 +125,8 @@ func (s *ClothesService) Upload(ctx context.Context, userID int64, fileHeader *m
// 6. 异步抠图 + 分类。
go s.processImageAsync(ctx, userID, clothesUUID, imageData, originalPath)
slog.InfoContext(ctx, "clothes uploaded", "user_id", userID, "clothes_uuid", clothesUUID)
return &dto.ClothesUploadResponse{
ClothesUUID: clothesUUID,
ImageProcessStatus: clothes.ImageProcessStatus,
@@ -140,11 +156,13 @@ func (s *ClothesService) processImageAsync(parentCtx context.Context, userID int
dao := s.clothesDao
// 1. 抠图。
_ = dao.ClothesUpdateByUUID(ctx, clothesUUID, map[string]any{"image_process_status": ImageProcessMatting})
if uerr := dao.ClothesUpdateByUUID(ctx, clothesUUID, map[string]any{"image_process_status": ImageProcessMatting}); uerr != nil {
slog.DebugContext(ctx, "clothes process: mark matting status failed", "clothes_uuid", clothesUUID, "error", uerr)
}
mattingClient := matting.NewClient()
processedData, err := retryCall(ctx, global.MattingModelSetting.MaxRetry, "matting", func() ([]byte, error) {
return mattingClient.Matting(ctx, imageData)
})
}, "clothes_uuid", clothesUUID)
if err != nil {
slog.ErrorContext(ctx, "matting failed", "clothes_uuid", clothesUUID, "error", err)
s.markProcessFailed(ctx, userID, clothesUUID, "图片抠图失败")
@@ -156,7 +174,7 @@ func (s *ClothesService) processImageAsync(parentCtx context.Context, userID int
// 获取用户 UUID，作为 COS 对象键第一层（uuid/userid/图片）。
user, gerr := s.userDao.UserGetByID(ctx, userID)
if gerr != nil {
slog.ErrorContext(ctx, "get user for cos key failed", "clothes_uuid", clothesUUID, "error", gerr)
slog.ErrorContext(ctx, "get user for cos key failed", "clothes_uuid", clothesUUID, "user_id", userID, "error", gerr)
s.markProcessFailed(ctx, userID, clothesUUID, "获取用户信息失败")
return
}
@@ -186,15 +204,19 @@ func (s *ClothesService) processImageAsync(parentCtx context.Context, userID int
}
// 3. 分类。
_ = dao.ClothesUpdateByUUID(ctx, clothesUUID, map[string]any{"image_process_status": ImageProcessClassifying})
if uerr := dao.ClothesUpdateByUUID(ctx, clothesUUID, map[string]any{"image_process_status": ImageProcessClassifying}); uerr != nil {
slog.DebugContext(ctx, "clothes process: mark classifying status failed", "clothes_uuid", clothesUUID, "error", uerr)
}
classifyClient := classification.NewClient()
result, cerr := retryCall(ctx, global.ClassificationModelSetting.MaxRetry, "classify", func() (*classification.ClassifyResult, error) {
return classifyClient.Classify(ctx, processedData)
})
}, "clothes_uuid", clothesUUID)
if cerr != nil {
slog.ErrorContext(ctx, "classify failed", "clothes_uuid", clothesUUID, "error", cerr)
// 分类失败仍保留已抠图结果，但条件标记失败并回退以便用户手动补全。
_ = dao.ClothesUpdateByUUID(ctx, clothesUUID, map[string]any{"processed_image": cosKey})
if uerr := dao.ClothesUpdateByUUID(ctx, clothesUUID, map[string]any{"processed_image": cosKey}); uerr != nil {
slog.WarnContext(ctx, "clothes process: keep processed image failed", "clothes_uuid", clothesUUID, "error", uerr)
}
s.markProcessFailed(ctx, userID, clothesUUID, "图片分类失败")
return
}
@@ -260,7 +282,7 @@ func buildDefaultClothesName(result *classification.ClassifyResult) string {
// retryCall 对 fn 进行最多 maxRetry 次额外重试（总尝试 maxRetry+1 次）。
// 重试间隔 1s；ctx 到期则提前终止并返回 ctx.Err()。
func retryCall[T any](ctx context.Context, maxRetry int, name string, fn func() (T, error)) (T, error) {
func retryCall[T any](ctx context.Context, maxRetry int, name string, fn func() (T, error), attrs ...any) (T, error) {
var zero T
var err error
for attempt := 0; attempt <= maxRetry; attempt++ {
@@ -272,7 +294,8 @@ func retryCall[T any](ctx context.Context, maxRetry int, name string, fn func()
if attempt >= maxRetry {
break
}
slog.WarnContext(ctx, "model call failed, retrying", "name", name, "attempt", attempt+1, "error", e)
args := append([]any{"name", name, "attempt", attempt + 1, "error", e}, attrs...)
slog.WarnContext(ctx, "model call failed, retrying", args...)
select {
case <-ctx.Done():
return zero, ctx.Err()
@@ -287,10 +310,12 @@ func retryCall[T any](ctx context.Context, maxRetry int, name string, fn func()
// 两路径串行化保证「处理失败回退」与「删除回退」恰好执行一次，消除漏退/双退窗口。
func (s *ClothesService) markProcessFailed(ctx context.Context, userID int64, clothesUUID, reason string) {
// 调用方的三分钟处理 context 可能已超时；清理与额度回退必须使用独立短 context。
cleanupCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
// 用 WithoutCancel 派生：既不受调用方取消影响，又保留 ctx 中已有的 span/trace 信息。
cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 20*time.Second)
defer cancel()
vipExpire := getUserVIPExpire(cleanupCtx, userID)
var originalPath string
failPhase := ""
err := global.MysqlEngine.WithContext(cleanupCtx).Transaction(func(tx *gorm.DB) error {
var c mysqlmodel.Clothes
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
@@ -312,16 +337,24 @@ func (s *ClothesService) markProcessFailed(ctx context.Context, userID int64, cl
"fail_reason": reason,
"original_image_path": "",
}).Error; err != nil {
failPhase = "mark_failed"
return err
}
return refundCosQuotaTx(tx, userID, vipExpire, 1, 0, "衣物处理失败回退")
if err := refundCosQuotaTx(cleanupCtx, tx, userID, vipExpire, 1, 0, "衣物处理失败回退"); err != nil {
failPhase = "refund_quota"
return err
}
return nil
})
if err != nil {
slog.ErrorContext(ctx, "mark process failed transaction error", "clothes_uuid", clothesUUID, "error", err)
slog.ErrorContext(ctx, "mark process failed transaction error",
"clothes_uuid", clothesUUID, "user_id", userID, "phase", failPhase, "error", err)
return
}
if originalPath != "" {
_ = os.Remove(originalPath)
if rerr := os.Remove(originalPath); rerr != nil {
slog.DebugContext(ctx, "clothes process: remove original image failed", "clothes_uuid", clothesUUID, "error", rerr)
}
}
}
@@ -452,9 +485,13 @@ func (s *ClothesService) DeleteByUUID(ctx context.Context, userID int64, request
// 也与接口文档约定的 50001「衣服已被搭配使用，无法删除」保持一致。
refCount, rerr := s.clothesDao.ClothesReferencedCount(ctx, request.ClothesUUID)
if rerr != nil {
slog.ErrorContext(ctx, "delete clothes: count references failed",
"clothes_uuid", request.ClothesUUID, "user_id", userID, "error", rerr)
return nil, errcode.DatabaseError
}
if refCount > 0 {
slog.InfoContext(ctx, "delete clothes: clothes in use",
"clothes_uuid", request.ClothesUUID, "user_id", userID)
return nil, errcode.ClothesInUse
}
@@ -480,18 +517,23 @@ func (s *ClothesService) DeleteByUUID(ctx context.Context, userID int64, request
return nil
}
// 同事务回退额度（优先免费额度，不足回退凭证）。
return refundCosQuotaTx(tx, userID, vipExpire, 1, 0, "删除衣物回退")
return refundCosQuotaTx(ctx, tx, userID, vipExpire, 1, 0, "删除衣物回退")
})
if err != nil {
if isNotFound(err) {
return nil, errcode.ClothesNotFound
}
slog.ErrorContext(ctx, "delete clothes: transaction failed",
"clothes_uuid", request.ClothesUUID, "user_id", userID, "error", err)
return nil, errcode.ClothesDeleteFailed
}
// 事务提交成功后清理 COS 图片（尽力而为，失败仅记录日志，不影响删除结果）。
if originalImagePath != "" {
_ = os.Remove(originalImagePath)
if rerr := os.Remove(originalImagePath); rerr != nil {
slog.DebugContext(ctx, "delete clothes: remove original image failed",
"clothes_uuid", request.ClothesUUID, "error", rerr)
}
}
if processedImage != "" {
if derr := upload.FileDeleteCos(ctx, processedImage); derr != nil {
@@ -499,6 +541,8 @@ func (s *ClothesService) DeleteByUUID(ctx context.Context, userID int64, request
}
}
slog.InfoContext(ctx, "clothes deleted", "clothes_uuid", request.ClothesUUID, "user_id", userID)
return &dto.ClothesDeleteByUUIDResponse{}, nil
}
@@ -604,17 +648,33 @@ func runClothesProcessingRecovery() {
slog.ErrorContext(ctx, "list stale clothes processing tasks failed", "error", err)
return
}
if len(list) == 0 {
slog.DebugContext(ctx, "clothes recovery: no stale processing tasks")
return
}
service := NewClothesService()
recovered := 0
for i := range list {
claimed, err := dao.ClothesClaimStaleProcessing(ctx, list[i].ClothesUUID, before)
if err != nil || !claimed {
claimed, cerr := dao.ClothesClaimStaleProcessing(ctx, list[i].ClothesUUID, before)
if cerr != nil {
slog.ErrorContext(ctx, "clothes recovery: claim stale task failed",
"clothes_uuid", list[i].ClothesUUID, "error", cerr)
continue
}
imageData, err := os.ReadFile(list[i].OriginalImagePath)
if err != nil || !upload.FileCheckImageMagic(imageData) {
if !claimed {
slog.DebugContext(ctx, "clothes recovery: task already claimed by another instance",
"clothes_uuid", list[i].ClothesUUID)
continue
}
imageData, rerr := os.ReadFile(list[i].OriginalImagePath)
if rerr != nil || !upload.FileCheckImageMagic(imageData) {
slog.WarnContext(ctx, "clothes recovery: original image missing or corrupted",
"clothes_uuid", list[i].ClothesUUID, "user_id", list[i].UserID, "error", rerr)
service.markProcessFailed(ctx, list[i].UserID, list[i].ClothesUUID, "原图丢失或损坏，任务无法恢复")
continue
}
recovered++
go service.processImageAsync(context.Background(), list[i].UserID, list[i].ClothesUUID, imageData, list[i].OriginalImagePath)
}
slog.InfoContext(ctx, "clothes recovery: stale tasks recovered", "scanned", len(list), "recovered", recovered)
}
internal/service/collect.go
+18 -2
View File
@@ -40,6 +40,7 @@ func (s *CollectService) Collect(ctx context.Context, userID int64, request *dto
if isNotFound(err) {
return nil, errcode.TargetNotFound
}
slog.ErrorContext(ctx, "collect: query outfit failed", "outfit_uuid", request.TargetUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
// 非本人仅可收藏公开可见的穿搭。
@@ -54,6 +55,7 @@ func (s *CollectService) Collect(ctx context.Context, userID int64, request *dto
if isNotFound(err) {
notExist = true
} else {
slog.ErrorContext(ctx, "collect: query record failed", "target_id", targetID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
}
@@ -66,16 +68,24 @@ func (s *CollectService) Collect(ctx context.Context, userID int64, request *dto
UserID: userID,
Status: InteractStatusOn,
}); cerr != nil {
slog.ErrorContext(ctx, "collect: upsert record failed", "target_id", targetID, "user_id", userID, "error", cerr)
return nil, errcode.CollectFailed
}
_ = redisdao.CollectAdd(ctx, CollectTypeOutfit, targetID, userID)
if rerr := redisdao.CollectAdd(ctx, CollectTypeOutfit, targetID, userID); rerr != nil {
slog.WarnContext(ctx, "collect: redis add failed, counter may undercount",
"target_id", targetID, "user_id", userID, "error", rerr)
}
case ActionCancel: // 取消收藏
if !notExist && existing.Status == InteractStatusOn {
if uerr := s.collectDao.CollectUpdateStatus(ctx, existing.ID, InteractStatusOff); uerr != nil {
slog.ErrorContext(ctx, "collect: update status failed", "target_id", targetID, "user_id", userID, "error", uerr)
return nil, errcode.CollectFailed
}
_ = redisdao.CollectRemove(ctx, CollectTypeOutfit, targetID, userID)
if rerr := redisdao.CollectRemove(ctx, CollectTypeOutfit, targetID, userID); rerr != nil {
slog.WarnContext(ctx, "collect: redis remove failed, counter may overcount",
"target_id", targetID, "user_id", userID, "error", rerr)
}
}
}
@@ -131,6 +141,7 @@ func (s *CollectService) GetByTarget(ctx context.Context, userID int64, request
if isNotFound(err) {
return nil, errcode.TargetNotFound
}
slog.ErrorContext(ctx, "collect: query outfit for status failed", "outfit_uuid", request.TargetUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
// 非本人仅可查询公开可见穿搭的收藏状态。
@@ -150,6 +161,9 @@ func (s *CollectService) GetByTarget(ctx context.Context, userID int64, request
"target_id", targetID, "user_id", userID, "error", derr)
if v, rerr := redisdao.CollectIsMember(ctx, CollectTypeOutfit, targetID, userID); rerr == nil {
isCollected = v
} else {
slog.DebugContext(ctx, "collect: redis fallback read failed",
"target_id", targetID, "user_id", userID, "error", rerr)
}
}
@@ -160,6 +174,8 @@ func (s *CollectService) GetByTarget(ctx context.Context, userID int64, request
"target_id", targetID, "error", derr)
if rcnt, rerr := redisdao.CollectCount(ctx, CollectTypeOutfit, targetID); rerr == nil {
count = rcnt
} else {
slog.DebugContext(ctx, "collect: redis fallback count failed", "target_id", targetID, "error", rerr)
}
}
internal/service/constants.go
+7 -4
View File
@@ -16,7 +16,7 @@ const (
VIPTypeNone int8 = 1 // 非会员
VIPTypeMonthly int8 = 2 // 月度会员
VIPTypeYearly int8 = 3 // 年度会员
VIPTypeTrial int8 = 4 // 体验会员（新用户注册赠送，天数由 UserDefaultQuota.GiftVIPDay 配置）
VIPTypeTrial int8 = 4 // 体验会员（新用户注册赠送，天数由 NewUserGift.GiftVIPDay 配置）
)
// ===== 配额 Quota =====
@@ -27,9 +27,10 @@ const (
QuotaTypeClosetShare int8 = 4 // 衣柜分享
// 额度凭证来源（quota_grants.source）
QuotaGrantSourceMember int8 = 1 // 会员包（跟随 VIP 有效期）
QuotaGrantSourceMember int8 = 1 // 会员包（跟随 VIP 有效期，续期自动延长）
QuotaGrantSourceAddon int8 = 2 // 加量包（独立有效期）
QuotaGrantSourceAdReward int8 = 3 // 广告激励（当天有效）
QuotaGrantSourceTrial int8 = 4 // 体验会员赠送（新用户注册；固定有效期=体验期结束，不随续费延长）
// ChangeTarget 调整额度类型
QuotaChangeTargetAdjust int8 = 1 // 配额调整
@@ -42,6 +43,7 @@ const (
QuotaChangeTypeRefund int8 = 4 // 删除回退
QuotaChangeTypeManual int8 = 5 // 人工调整
QuotaChangeTypeRefundRevoke int8 = 6 // 退款回收
QuotaChangeTypeExpire int8 = 7 // 到期失效（固定有效期的额度凭证过期，由定时任务补记）
)
// 新用户默认免费额度（内置兜底值）。
@@ -55,8 +57,9 @@ const (
)
// 新用户注册赠送体验会员的内置兜底天数。
// 优先使用 UserDefaultQuota.GiftVIPDay 的配置值（配置为 0 时生效，表示不赠送）；
// 仅在该项未配置（指针为 nil）或配置为负数时回落此常量。
// 优先使用 NewUserGift.GiftVIPDay 的配置值（配置为 0 时生效，表示不赠送）；
// 该项未配置时回落旧的 UserDefaultQuota.GiftVIPDay（兼容既有配置）；
// 两者都未配置（指针为 nil）或配置为负数时回落此常量。
const DefaultVipDays int64 = 3
// ===== 衣物 Clothes 图片处理状态 =====
internal/service/horoscope.go
+17 -2
View File
@@ -3,6 +3,7 @@ package service
import (
"context"
"encoding/json"
"log/slog"
"time"
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/global"
@@ -36,11 +37,14 @@ func (s *HoroscopeService) GetToday(ctx context.Context, userID int64) (*dto.Hor
user, err := s.userDao.UserGetByID(ctx, userID)
if err != nil {
if isNotFound(err) {
slog.WarnContext(ctx, "horoscope: user not found", "user_id", userID)
return nil, errcode.UserNotFound
}
slog.ErrorContext(ctx, "horoscope: query user failed", "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
if user.BirthDay == 0 {
slog.InfoContext(ctx, "horoscope: birth day missing", "user_id", userID)
return nil, errcode.InvalidParams.WithDetails("请先完善生日信息")
}
@@ -50,9 +54,14 @@ func (s *HoroscopeService) GetToday(ctx context.Context, userID int64) (*dto.Hor
// 1. Redis 优先。
if data, cerr := redisdao.HoroscopeGet(ctx, zodiac, statDate); cerr == nil && len(data) > 0 {
var response dto.HoroscopeResponse
if json.Unmarshal(data, &response) == nil {
if uerr := json.Unmarshal(data, &response); uerr == nil {
return &response, nil
} else {
slog.DebugContext(ctx, "horoscope: unmarshal cached data failed, fallback to db",
"zodiac", zodiac, "error", uerr)
}
} else if cerr != nil {
slog.DebugContext(ctx, "horoscope: cache read failed, fallback to db", "zodiac", zodiac, "error", cerr)
}
// 2. 回落 MySQL。
@@ -60,9 +69,11 @@ func (s *HoroscopeService) GetToday(ctx context.Context, userID int64) (*dto.Hor
if derr != nil {
if isNotFound(derr) {
// 今日运势尚未生成，异步触发兜底生成，本次返回未生成。
slog.InfoContext(ctx, "horoscope: today data missing, trigger fallback generation", "zodiac", zodiac)
tokenhub.EnsureTodayFortunes()
return nil, errcode.HoroscopeNotFound
}
slog.ErrorContext(ctx, "horoscope: query db failed", "zodiac", zodiac, "error", derr)
return nil, errcode.DatabaseError
}
@@ -70,7 +81,11 @@ func (s *HoroscopeService) GetToday(ctx context.Context, userID int64) (*dto.Hor
// 回填缓存，TTL 至次日凌晨。
if data, merr := json.Marshal(response); merr == nil {
_ = redisdao.HoroscopeSet(ctx, zodiac, statDate, data, ttlUntilNextMidnight())
if serr := redisdao.HoroscopeSet(ctx, zodiac, statDate, data, ttlUntilNextMidnight()); serr != nil {
slog.DebugContext(ctx, "horoscope: cache backfill failed", "zodiac", zodiac, "error", serr)
}
} else {
slog.DebugContext(ctx, "horoscope: marshal response failed", "zodiac", zodiac, "error", merr)
}
return &response, nil
internal/service/like.go
+15 -2
View File
@@ -43,6 +43,7 @@ func (s *LikeService) LikeCreate(ctx context.Context, userID int64, request *dto
if isNotFound(err) {
notExist = true
} else {
slog.ErrorContext(ctx, "like: query record failed", "target_id", targetID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
}
@@ -55,9 +56,13 @@ func (s *LikeService) LikeCreate(ctx context.Context, userID int64, request *dto
UserID: userID,
Status: InteractStatusOn,
}); cerr != nil {
slog.ErrorContext(ctx, "like: upsert record failed", "target_id", targetID, "user_id", userID, "error", cerr)
return nil, errcode.LikeFailed
}
_ = redisdao.LikeAdd(ctx, LikeTypeOutfit, targetID, userID)
if rerr := redisdao.LikeAdd(ctx, LikeTypeOutfit, targetID, userID); rerr != nil {
slog.WarnContext(ctx, "like: redis add failed, counter may undercount",
"target_id", targetID, "user_id", userID, "error", rerr)
}
// 仅在「从取消/无记录变为点赞」时发通知，避免并发/重复点赞重复通知。
wasLiked := !notExist && existing.Status == InteractStatusOn
@@ -68,9 +73,13 @@ func (s *LikeService) LikeCreate(ctx context.Context, userID int64, request *dto
case ActionCancel: // 取消点赞
if !notExist && existing.Status == InteractStatusOn {
if uerr := s.likeDao.LikeUpdateStatus(ctx, existing.ID, InteractStatusOff); uerr != nil {
slog.ErrorContext(ctx, "like: update status failed", "target_id", targetID, "user_id", userID, "error", uerr)
return nil, errcode.LikeFailed
}
_ = redisdao.LikeRemove(ctx, LikeTypeOutfit, targetID, userID)
if rerr := redisdao.LikeRemove(ctx, LikeTypeOutfit, targetID, userID); rerr != nil {
slog.WarnContext(ctx, "like: redis remove failed, counter may overcount",
"target_id", targetID, "user_id", userID, "error", rerr)
}
}
}
@@ -99,6 +108,9 @@ func (s *LikeService) GetByTargetID(ctx context.Context, userID int64, request *
"target_id", targetID, "user_id", userID, "error", derr)
if v, rerr := redisdao.LikeIsMember(ctx, LikeTypeOutfit, targetID, userID); rerr == nil {
isLiked = v
} else {
slog.DebugContext(ctx, "like: redis fallback read failed",
"target_id", targetID, "user_id", userID, "error", rerr)
}
}
@@ -125,6 +137,7 @@ func (s *LikeService) getPublicOutfit(ctx context.Context, userID int64, outfitU
if isNotFound(err) {
return nil, errcode.TargetNotFound
}
slog.ErrorContext(ctx, "like: query outfit failed", "outfit_uuid", outfitUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
if outfit.UserID != userID && !isOutfitPublicVisible(outfit) {
internal/service/notification.go
+10 -2
View File
@@ -76,6 +76,8 @@ func (s *NotificationService) Read(ctx context.Context, userID int64, request *d
if isNotFound(err) {
return nil, errcode.NotificationNotFound
}
slog.ErrorContext(ctx, "read notification: query notification failed",
"notification_uuid", request.NotificationUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
if n.UserID != userID {
@@ -93,6 +95,8 @@ func (s *NotificationService) Read(ctx context.Context, userID int64, request *d
// 条件标记已读（仅未读 -> 已读），防并发重复标记。
if _, err := s.notificationDao.NotificationUpdateReadByUUIDIfUnread(ctx, request.NotificationUUID); err != nil {
slog.ErrorContext(ctx, "mark notification read failed",
"notification_uuid", request.NotificationUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
// 无论本次是否真的标记成功，都按 DB 重新统计并回填缓存：
@@ -127,8 +131,11 @@ func (s *NotificationService) unreadCount(ctx context.Context, userID int64) (in
slog.WarnContext(ctx, "count unread from db failed, fallback to redis",
"user_id", userID, "error", err)
return rcnt, nil
} else {
slog.ErrorContext(ctx, "count unread failed: db and redis both unavailable",
"user_id", userID, "db_error", err, "redis_error", rerr)
return 0, errcode.DatabaseError
}
return 0, errcode.DatabaseError
}
if serr := redisdao.NotificationUnreadSet(ctx, userID, cnt); serr != nil {
slog.WarnContext(ctx, "backfill notification unread cache failed",
@@ -167,6 +174,7 @@ func sendNotification(ctx context.Context, recipientUserID, senderID int64, msgT
return
}
if _, err := redisdao.NotificationUnreadIncr(ctx, recipientUserID); err != nil {
slog.ErrorContext(ctx, "notification unread incr failed", "user_id", recipientUserID, "error", err)
// 未读数是 best-effort 缓存：失败不影响通知已落库，级别与「通知创建失败」保持一致。
slog.WarnContext(ctx, "notification unread incr failed", "user_id", recipientUserID, "error", err)
}
}
internal/service/order.go
+101 -16
View File
@@ -47,11 +47,14 @@ func (s *OrderService) Create(ctx context.Context, userID int64, request *dto.Or
goods, err := s.goodsDao.GoodsGetByCode(ctx, request.GoodsCode)
if err != nil {
if isNotFound(err) {
slog.InfoContext(ctx, "create order: goods not found", "user_id", userID, "goods_code", request.GoodsCode)
return nil, errcode.GoodsNotFound
}
slog.ErrorContext(ctx, "create order: query goods failed", "goods_code", request.GoodsCode, "error", err)
return nil, errcode.DatabaseError
}
if goods.Status != GoodsStatusOnShelf {
slog.InfoContext(ctx, "create order: goods off shelf", "user_id", userID, "goods_code", request.GoodsCode)
return nil, errcode.GoodsOffShelf
}
@@ -59,11 +62,14 @@ func (s *OrderService) Create(ctx context.Context, userID int64, request *dto.Or
user, uerr := s.userDao.UserGetByID(ctx, userID)
if uerr != nil {
if isNotFound(uerr) {
slog.WarnContext(ctx, "create order: user not found", "user_id", userID)
return nil, errcode.UserNotFound
}
slog.ErrorContext(ctx, "create order: query user failed", "user_id", userID, "error", uerr)
return nil, errcode.DatabaseError
}
if user.WxOpenID == "" {
slog.WarnContext(ctx, "create order: user openid missing", "user_id", userID)
return nil, errcode.OrderCreateFailed.WithDetails("缺少微信 openid")
}
@@ -93,14 +99,20 @@ func (s *OrderService) Create(ctx context.Context, userID int64, request *dto.Or
// 先落库再微信下单：微信下单失败时本地订单保留为待支付并超时自动过期，
// 避免「先微信下单、落库失败」产生无本地记录的孤儿预支付单。
if cerr := s.orderDao.OrderCreate(ctx, order); cerr != nil {
slog.ErrorContext(ctx, "create order: save order failed",
"order_no", orderNo, "user_id", userID, "goods_code", request.GoodsCode, "error", cerr)
return nil, errcode.OrderCreateFailed
}
payParams, perr := wechat.JsapiPrepay(ctx, user.WxOpenID, orderNo, goods.GoodsName, payAmount)
if perr != nil {
slog.ErrorContext(ctx, "create order: jsapi prepay failed", "order_no", orderNo, "user_id", userID, "error", perr)
return nil, errcode.OrderCreateFailed
}
slog.InfoContext(ctx, "order created",
"order_no", orderNo, "user_id", userID, "goods_code", request.GoodsCode, "pay_amount", payAmount)
return &dto.OrderCreateResponse{
OrderNo: orderNo,
PayAmount: payAmount,
@@ -122,16 +134,25 @@ func (s *OrderService) PayNotify(ctx context.Context, request *http.Request) *er
data, err := wechat.VerifyAndDecryptPayNotify(ctx, request)
if err != nil {
slog.ErrorContext(ctx, "pay notify: verify or decrypt failed", "remote_addr", request.RemoteAddr, "error", err)
return errcode.RPCError
}
if data.TradeState != "SUCCESS" {
slog.InfoContext(ctx, "pay notify: ignored non-success trade state",
"out_trade_no", data.OutTradeNo, "trade_state", data.TradeState)
return nil // 非支付成功，忽略。
}
// 幂等锁。
lockKey := fmtLockKey(redismodel.KeyLockPayCallback, data.OutTradeNo)
lockVal := genUUID()
if ok, _ := redisdao.TryLock(ctx, lockKey, lockVal, 30*time.Second); !ok {
ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, 30*time.Second)
if lerr != nil {
slog.ErrorContext(ctx, "pay notify: acquire lock failed", "out_trade_no", data.OutTradeNo, "error", lerr)
return nil // 锁不可用（Redis 故障）：不重复处理，交由微信重试/持锁者。
}
if !ok {
slog.DebugContext(ctx, "pay notify: concurrent callback, skip", "out_trade_no", data.OutTradeNo)
return nil // 并发回调，交由持锁者处理。
}
defer func() { _ = redisdao.Unlock(ctx, lockKey, lockVal) }()
@@ -143,6 +164,7 @@ func (s *OrderService) PayNotify(ctx context.Context, request *http.Request) *er
slog.ErrorContext(ctx, "pay notify: order not found", "out_trade_no", data.OutTradeNo)
return nil
}
slog.ErrorContext(ctx, "pay notify: query order failed", "out_trade_no", data.OutTradeNo, "error", gerr)
return errcode.DatabaseError
}
if data.Amount.Total != order.PayAmount {
@@ -166,9 +188,12 @@ func (s *OrderService) PayNotify(ctx context.Context, request *http.Request) *er
// 条件流转：待支付 -> 已支付，防重复回调。
updated, uerr := s.orderDao.OrderUpdateStatusByOrderNo(ctx, data.OutTradeNo, OrderStatusPending, OrderStatusPaid)
if uerr != nil {
slog.ErrorContext(ctx, "pay notify: update order status failed", "out_trade_no", data.OutTradeNo, "error", uerr)
return errcode.DatabaseError
}
if updated {
slog.InfoContext(ctx, "pay notify: order paid",
"order_no", order.OrderNo, "user_id", order.UserID, "pay_amount", order.PayAmount, "transaction_id", data.TransactionID)
s.finalizePaidOrder(ctx, order, data)
return nil
}
@@ -177,8 +202,10 @@ func (s *OrderService) PayNotify(ctx context.Context, request *http.Request) *er
current, gerr := s.orderDao.OrderGetByOrderNo(ctx, data.OutTradeNo)
if gerr != nil {
if isNotFound(gerr) {
slog.WarnContext(ctx, "pay notify: order disappeared after status conflict", "out_trade_no", data.OutTradeNo)
return nil
}
slog.ErrorContext(ctx, "pay notify: reload order failed", "out_trade_no", data.OutTradeNo, "error", gerr)
return errcode.DatabaseError
}
@@ -186,22 +213,28 @@ func (s *OrderService) PayNotify(ctx context.Context, request *http.Request) *er
case OrderStatusPaid:
// 已被并发回调处理。兜底确保支付信息与权益发放已完成（均幂等），
// 覆盖「订单已流转为 Paid 但权益发放前崩溃导致权益丢失」的场景。
_ = s.orderDao.OrderUpdateByOrderNo(ctx, current.OrderNo, map[string]any{
if uerr := s.orderDao.OrderUpdateByOrderNo(ctx, current.OrderNo, map[string]any{
"pay_time": nowUnix(),
"transaction_id": data.TransactionID,
"callback_raw": data.Raw,
})
}); uerr != nil {
slog.ErrorContext(ctx, "pay notify: reconcile paid order pay info failed", "order_no", current.OrderNo, "error", uerr)
}
slog.InfoContext(ctx, "pay notify: order already paid, reconcile benefits", "order_no", current.OrderNo, "user_id", current.UserID)
s.grantOrderBenefits(ctx, current)
return nil
case OrderStatusExpired:
// 订单被系统过期但用户已付款：过期 -> 已支付并补发货，避免用户付款却不到账。
ok, uerr := s.orderDao.OrderUpdateStatusByOrderNo(ctx, data.OutTradeNo, OrderStatusExpired, OrderStatusPaid)
if uerr != nil {
slog.ErrorContext(ctx, "pay notify: compensate expired order failed", "out_trade_no", data.OutTradeNo, "error", uerr)
return errcode.DatabaseError
}
if !ok {
slog.DebugContext(ctx, "pay notify: expired order already handled by another callback", "out_trade_no", data.OutTradeNo)
return nil // 已被其它回调处理。
}
slog.InfoContext(ctx, "pay notify: expired order compensated", "order_no", current.OrderNo, "user_id", current.UserID)
s.finalizePaidOrder(ctx, current, data)
return nil
case OrderStatusCancelled:
@@ -209,6 +242,7 @@ func (s *OrderService) PayNotify(ctx context.Context, request *http.Request) *er
s.autoRefundCancelledOrder(ctx, current, data)
return nil
default:
slog.DebugContext(ctx, "pay notify: unhandled order status", "out_trade_no", data.OutTradeNo, "order_status", current.OrderStatus)
return nil
}
}
@@ -353,12 +387,19 @@ func (s *OrderService) RefundNotify(ctx context.Context, request *http.Request)
data, err := wechat.VerifyAndDecryptRefundNotify(ctx, request)
if err != nil {
slog.ErrorContext(ctx, "refund notify: verify or decrypt failed", "remote_addr", request.RemoteAddr, "error", err)
return errcode.RPCError
}
lockKey := fmtLockKey(redismodel.KeyLockRefundCallback, data.OutRefundNo)
lockVal := genUUID()
if ok, _ := redisdao.TryLock(ctx, lockKey, lockVal, 30*time.Second); !ok {
ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, 30*time.Second)
if lerr != nil {
slog.ErrorContext(ctx, "refund notify: acquire lock failed", "out_refund_no", data.OutRefundNo, "error", lerr)
return nil // 锁不可用（Redis 故障）：不重复处理，交由微信重试/持锁者。
}
if !ok {
slog.DebugContext(ctx, "refund notify: concurrent callback, skip", "out_refund_no", data.OutRefundNo)
return nil
}
defer func() { _ = redisdao.Unlock(ctx, lockKey, lockVal) }()
@@ -366,11 +407,14 @@ func (s *OrderService) RefundNotify(ctx context.Context, request *http.Request)
flow, ferr := s.refundDao.RefundFlowLogGetByRefundNo(ctx, data.OutRefundNo)
if ferr != nil {
if isNotFound(ferr) {
slog.WarnContext(ctx, "refund notify: refund flow not found", "out_refund_no", data.OutRefundNo)
return nil
}
slog.ErrorContext(ctx, "refund notify: get refund flow failed", "out_refund_no", data.OutRefundNo, "error", ferr)
return errcode.DatabaseError
}
if flow.Status == RefundStatusFailed {
slog.DebugContext(ctx, "refund notify: refund flow already failed, skip", "out_refund_no", data.OutRefundNo)
return nil
}
if data.Amount.Refund != flow.RefundAmount {
@@ -381,21 +425,30 @@ func (s *OrderService) RefundNotify(ctx context.Context, request *http.Request)
if flow.Status != RefundStatusSuccess {
ok, uerr := s.refundDao.RefundFlowLogUpdateStatusByFrom(ctx, data.OutRefundNo, flow.Status, map[string]any{"status": RefundStatusSuccess, "success_time": nowUnix(), "callback_raw": data.Raw})
if uerr != nil {
slog.ErrorContext(ctx, "refund notify: update refund flow status failed", "out_refund_no", data.OutRefundNo, "error", uerr)
return errcode.DatabaseError
}
if !ok {
slog.WarnContext(ctx, "refund notify: refund flow status changed concurrently", "out_refund_no", data.OutRefundNo)
return errcode.DatabaseError
}
}
order, gerr := s.orderDao.OrderGetByOrderNo(ctx, data.OutTradeNo)
if gerr != nil {
slog.ErrorContext(ctx, "refund notify: get order failed", "out_trade_no", data.OutTradeNo, "error", gerr)
return errcode.DatabaseError
}
if err := s.settleRefundSuccess(ctx, order, flow.RefundAmount); err != nil {
slog.ErrorContext(ctx, "refund notify: settle refund success failed",
"out_refund_no", data.OutRefundNo, "order_no", order.OrderNo, "error", err)
return errcode.DatabaseError
}
slog.InfoContext(ctx, "refund notify: refund success settled",
"order_no", order.OrderNo, "out_refund_no", data.OutRefundNo, "refund_amount", flow.RefundAmount)
} else if flow.Status != RefundStatusSuccess {
if _, err := s.refundDao.RefundFlowLogUpdateStatusByFrom(ctx, data.OutRefundNo, flow.Status, map[string]any{"status": RefundStatusFailed, "fail_reason": "微信退款未成功:" + data.RefundStatus, "callback_raw": data.Raw}); err != nil {
slog.ErrorContext(ctx, "refund notify: mark refund failed status failed",
"out_refund_no", data.OutRefundNo, "refund_status", data.RefundStatus, "error", err)
return errcode.DatabaseError
}
}
@@ -407,15 +460,20 @@ func (s *OrderService) RefundNotify(ctx context.Context, request *http.Request)
func (s *OrderService) autoRefundCancelledOrder(ctx context.Context, order *mysqlmodel.Order, data *wechat.PayNotifyData) {
// 幂等：已有退款流水则跳过，避免重复生成待审核退款单。
if existing, err := s.refundDao.RefundFlowLogGetByOrderNo(ctx, order.OrderNo); err == nil && existing != nil {
slog.InfoContext(ctx, "auto refund: refund flow already exists, skip", "order_no", order.OrderNo)
return
} else if err != nil && !isNotFound(err) {
slog.ErrorContext(ctx, "auto refund: query existing refund flow failed", "order_no", order.OrderNo, "error", err)
}
// 补记支付信息（订单已取消，此前未记录），便于管理员审核退款时核对。
_ = s.orderDao.OrderUpdateByOrderNo(ctx, order.OrderNo, map[string]any{
if uerr := s.orderDao.OrderUpdateByOrderNo(ctx, order.OrderNo, map[string]any{
"pay_time": nowUnix(),
"transaction_id": data.TransactionID,
"callback_raw": data.Raw,
})
}); uerr != nil {
slog.ErrorContext(ctx, "auto refund: record pay info failed", "order_no", order.OrderNo, "error", uerr)
}
refundNo := genRefundNo()
flow := &mysqlmodel.RefundFlowLog{
@@ -435,8 +493,8 @@ func (s *OrderService) autoRefundCancelledOrder(ctx context.Context, order *mysq
slog.ErrorContext(ctx, "auto refund: create refund flow failed", "order_no", order.OrderNo, "error", err)
return
}
slog.InfoContext(ctx, "cancelled order refund flow created, waiting manual review",
"order_no", order.OrderNo, "refund_no", refundNo)
slog.WarnContext(ctx, "cancelled order refund flow created, waiting manual review",
"order_no", order.OrderNo, "user_id", order.UserID, "refund_no", refundNo, "refund_amount", order.PayAmount)
}
// ---- 内部辅助 ----
@@ -458,11 +516,13 @@ func (s *OrderService) getOwnedOrder(ctx context.Context, orderNo string, userID
// finalizePaidOrder 记录支付信息、发放权益并发送通知（支付成功或过期补发货后调用）。
func (s *OrderService) finalizePaidOrder(ctx context.Context, order *mysqlmodel.Order, data *wechat.PayNotifyData) {
_ = s.orderDao.OrderUpdateByOrderNo(ctx, order.OrderNo, map[string]any{
if uerr := s.orderDao.OrderUpdateByOrderNo(ctx, order.OrderNo, map[string]any{
"pay_time": nowUnix(),
"transaction_id": data.TransactionID,
"callback_raw": data.Raw,
})
}); uerr != nil {
slog.ErrorContext(ctx, "finalize paid order: update pay info failed", "order_no", order.OrderNo, "error", uerr)
}
s.grantOrderBenefits(ctx, order)
@@ -522,7 +582,7 @@ func (s *OrderService) grantBenefitIdempotent(ctx context.Context, order *mysqlm
// 曾失败过：冷却期内不重置（防微信回调重试风暴无限重置任务绕过重试上限），
// 冷却期外重置为待发放并尝试发放（发放本身有幂等兜底，重置失败仅告警）。
if nowUnix()-existing.UpdatedAt < benefitRetryCooldownSeconds {
slog.WarnContext(ctx, "benefit grant task in cooldown, skip retry",
slog.InfoContext(ctx, "benefit grant task in cooldown, skip retry",
"order_no", order.OrderNo, "benefit_type", benefitType, "task_id", existing.ID)
return
}
@@ -543,28 +603,43 @@ func (s *OrderService) grantBenefitIdempotent(ctx context.Context, order *mysqlm
// executeGrant 认领并发放单个权益任务（claim 保证同一任务仅发放一次，防并发重复发放）。
func (s *OrderService) executeGrant(ctx context.Context, task *mysqlmodel.BenefitGrantTask) {
var grantErr error
var granted bool
err := global.MysqlEngine.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var order mysqlmodel.Order
if err := tx.Select("id", "order_status").Clauses(clause.Locking{Strength: "UPDATE"}).Where("order_no = ?", task.OrderNo).First(&order).Error; err != nil {
return err
}
if order.OrderStatus != OrderStatusPaid {
slog.InfoContext(ctx, "grant benefit skipped: order not paid",
"order_no", task.OrderNo, "benefit_type", task.BenefitType)
return tx.Model(&mysqlmodel.BenefitGrantTask{}).Where("id = ? AND status IN ?", task.ID, []int8{BenefitTaskPending, BenefitTaskFailed}).Update("status", BenefitTaskCancelled).Error
}
res := tx.Model(&mysqlmodel.BenefitGrantTask{}).Where("id = ? AND status IN ?", task.ID, []int8{BenefitTaskPending, BenefitTaskFailed}).Update("status", BenefitTaskProcessing)
if res.Error != nil || res.RowsAffected == 0 {
if res.Error == nil {
slog.InfoContext(ctx, "grant benefit skipped: task already claimed or status changed",
"order_no", task.OrderNo, "benefit_type", task.BenefitType, "task_id", task.ID)
}
return res.Error
}
if e := s.doGrantBenefit(ctx, task); e != nil {
grantErr = e
return tx.Model(&mysqlmodel.BenefitGrantTask{}).Where("id = ?", task.ID).Updates(map[string]any{"status": BenefitTaskPending, "retry_count": gorm.Expr("retry_count + 1"), "last_error": e.Error()}).Error
}
granted = true
return tx.Model(&mysqlmodel.BenefitGrantTask{}).Where("id = ?", task.ID).Updates(map[string]any{"status": BenefitTaskSuccess, "last_error": ""}).Error
})
if err != nil {
slog.ErrorContext(ctx, "grant benefit task update failed", "order_no", task.OrderNo, "benefit_type", task.BenefitType, "error", err)
slog.ErrorContext(ctx, "grant benefit task update failed",
"order_no", task.OrderNo, "benefit_type", task.BenefitType, "user_id", task.UserID,
"retry_count", task.RetryCount, "amount", task.Amount, "error", err)
} else if grantErr != nil {
slog.ErrorContext(ctx, "grant benefit failed", "order_no", task.OrderNo, "benefit_type", task.BenefitType, "error", grantErr)
slog.ErrorContext(ctx, "grant benefit failed",
"order_no", task.OrderNo, "benefit_type", task.BenefitType, "user_id", task.UserID,
"retry_count", task.RetryCount, "amount", task.Amount, "error", grantErr)
} else if granted {
slog.InfoContext(ctx, "benefit granted",
"order_no", task.OrderNo, "benefit_type", task.BenefitType, "user_id", task.UserID, "amount", task.Amount)
}
}
@@ -697,7 +772,9 @@ func (s *OrderService) grantVIP(ctx context.Context, userID, memberDays int64, o
return err
}
// 用户信息已变更，失效缓存。
_ = redisdao.UserInfoDel(ctx, userID)
if derr := redisdao.UserInfoDel(ctx, userID); derr != nil {
slog.DebugContext(ctx, "grant vip: invalidate user cache failed", "user_id", userID, "order_no", orderNo, "error", derr)
}
return nil
}
@@ -710,7 +787,10 @@ func (s *OrderService) settleRefundSuccess(ctx context.Context, order *mysqlmode
return err
}
if settled {
slog.InfoContext(ctx, "refund settled", "order_no", order.OrderNo, "user_id", order.UserID, "refund_amount", refundAmount)
sendNotification(ctx, order.UserID, 0, NotificationTypeOrder, "退款成功", "订单「"+order.GoodsName+"」已退款")
} else {
slog.InfoContext(ctx, "refund settle skipped: order already settled or not in paid status", "order_no", order.OrderNo)
}
// 立即尝试；失败时任务仍为待处理，由定时任务恢复。
s.executeRevokeTaskByOrder(ctx, order.OrderNo)
@@ -738,7 +818,7 @@ func (s *OrderService) revokeOrderBenefits(ctx context.Context, order *mysqlmode
func (s *OrderService) buildQuotaRevokeFlows(ctx context.Context, order *mysqlmodel.Order) ([]*mysqlmodel.QuotaFlowLog, error) {
grants, err := mysqldao.NewQuotaGrantDao().QuotaGrantListByRefID(ctx, order.OrderNo)
if err != nil {
slog.WarnContext(ctx, "list quota grants for revoke failed", "order_no", order.OrderNo, "error", err)
slog.ErrorContext(ctx, "list quota grants for revoke failed", "order_no", order.OrderNo, "error", err)
return nil, err
}
@@ -755,6 +835,9 @@ func (s *OrderService) buildQuotaRevokeFlows(ctx context.Context, order *mysqlmo
if !computed[g.QuotaType] {
if eff, qerr := getEffectiveQuota(ctx, order.UserID, g.QuotaType); qerr == nil {
current[g.QuotaType] = eff.Remain
} else {
slog.WarnContext(ctx, "build quota revoke flows: calc effective quota failed",
"order_no", order.OrderNo, "user_id", order.UserID, "quota_type", g.QuotaType, "error", qerr)
}
computed[g.QuotaType] = true
}
@@ -786,7 +869,9 @@ func (s *OrderService) revokeVIP(ctx context.Context, userID, memberDays int64,
if err := s.userDao.VIPRevokeIdempotent(ctx, userID, nowUnix(), memberDays, flow); err != nil && !isDuplicateKey(err) {
return err
}
_ = redisdao.UserInfoDel(ctx, userID)
if derr := redisdao.UserInfoDel(ctx, userID); derr != nil {
slog.DebugContext(ctx, "revoke vip: invalidate user cache failed", "user_id", userID, "order_no", orderNo, "error", derr)
}
return nil
}
internal/service/order_expire.go
+2
View File
@@ -42,5 +42,7 @@ func runOrderExpire() {
}
if affected > 0 {
slog.InfoContext(ctx, "expired pending orders", "count", affected)
} else {
slog.DebugContext(ctx, "order expire: no pending orders expired")
}
}
internal/service/outfit.go
+32 -3
View File
@@ -79,9 +79,12 @@ func (s *OutfitService) Create(ctx context.Context, userID int64, request *dto.O
}
return nil
}); err != nil {
slog.ErrorContext(ctx, "create outfit: transaction failed", "user_id", userID, "error", err)
return nil, errcode.OutfitCreateFailed
}
slog.InfoContext(ctx, "outfit created", "user_id", userID, "outfit_uuid", outfit.OutfitUUID)
response := buildOutfitCreateResponse(outfit, items)
return &response, nil
}
@@ -152,6 +155,8 @@ func (s *OutfitService) UpdateByUUID(ctx context.Context, userID int64, request
return tx.Create(&items).Error
})
if rebuildErr != nil {
slog.ErrorContext(ctx, "update outfit: transaction failed",
"user_id", userID, "outfit_uuid", request.OutfitUUID, "error", rebuildErr)
return nil, errcode.OutfitUpdateFailed
}
@@ -220,8 +225,11 @@ func (s *OutfitService) Delete(ctx context.Context, userID int64, request *dto.O
}
return tx.Where("outfit_uuid = ?", request.OutfitUUID).Delete(&mysqlmodel.Outfit{}).Error
}); err != nil {
slog.ErrorContext(ctx, "delete outfit: transaction failed",
"user_id", userID, "outfit_uuid", request.OutfitUUID, "error", err)
return nil, errcode.OutfitDeleteFailed
}
slog.InfoContext(ctx, "outfit deleted", "user_id", userID, "outfit_uuid", request.OutfitUUID)
return &dto.OutfitDeleteResponse{}, nil
}
@@ -243,9 +251,14 @@ func (s *OutfitService) Public(ctx context.Context, userID int64, request *dto.O
slog.WarnContext(ctx, "acquire outfit public lock failed, proceed without lock",
"outfit_uuid", request.OutfitUUID, "error", lerr)
} else if !ok {
slog.DebugContext(ctx, "outfit public: locked by another request, reject", "outfit_uuid", request.OutfitUUID)
return nil, errcode.TooManyRequests
} else {
defer func() { _ = redisdao.Unlock(ctx, lockKey, lockVal) }()
defer func() {
if uerr := redisdao.Unlock(ctx, lockKey, lockVal); uerr != nil {
slog.WarnContext(ctx, "outfit public: release lock failed", "outfit_uuid", request.OutfitUUID, "error", uerr)
}
}()
}
// 取消公开：直接置为不公开。
@@ -254,8 +267,11 @@ func (s *OutfitService) Public(ctx context.Context, userID int64, request *dto.O
return &dto.OutfitPublicResponse{}, nil
}
if err := s.outfitDao.OutfitUpdateIsPublic(ctx, request.OutfitUUID, OutfitPrivate); err != nil {
slog.ErrorContext(ctx, "outfit public: set private failed",
"outfit_uuid", request.OutfitUUID, "user_id", userID, "error", err)
return nil, errcode.OutfitUpdateFailed
}
slog.InfoContext(ctx, "outfit set private", "outfit_uuid", request.OutfitUUID, "user_id", userID)
return &dto.OutfitPublicResponse{}, nil
}
@@ -267,16 +283,24 @@ func (s *OutfitService) Public(ctx context.Context, userID int64, request *dto.O
// 首次公开：先条件更新（私密 -> 公开+审核通过），成功才消耗额度，防并发重复扣减。
ok, perr := s.outfitDao.OutfitPublish(ctx, request.OutfitUUID)
if perr != nil {
slog.ErrorContext(ctx, "outfit public: condition publish failed",
"outfit_uuid", request.OutfitUUID, "user_id", userID, "error", perr)
return nil, errcode.OutfitUpdateFailed
}
if !ok {
// 已被并发公开，幂等返回。
slog.DebugContext(ctx, "outfit public: already published by another request", "outfit_uuid", request.OutfitUUID)
return &dto.OutfitPublicResponse{}, nil
}
if qerr := consumeDailyQuota(ctx, userID, QuotaTypeOutfitShare, 1, "穿搭社区分享消耗"); qerr != nil {
// 扣额度失败，回退公开状态。
_ = s.outfitDao.OutfitUpdateIsPublic(ctx, request.OutfitUUID, OutfitPrivate)
slog.WarnContext(ctx, "outfit public: consume share quota failed, revert to private",
"outfit_uuid", request.OutfitUUID, "user_id", userID, "error", qerr)
if rerr := s.outfitDao.OutfitUpdateIsPublic(ctx, request.OutfitUUID, OutfitPrivate); rerr != nil {
slog.ErrorContext(ctx, "outfit public: revert to private failed",
"outfit_uuid", request.OutfitUUID, "user_id", userID, "error", rerr)
}
return nil, qerr
}
@@ -286,11 +310,14 @@ func (s *OutfitService) Public(ctx context.Context, userID int64, request *dto.O
AuditResult: AuditResultPass,
AuditTime: nowUnix(),
}); err != nil {
slog.WarnContext(ctx, "write outfit audit log failed", "outfit_id", outfit.ID, "error", err)
slog.WarnContext(ctx, "write outfit audit log failed",
"outfit_id", outfit.ID, "outfit_uuid", outfit.OutfitUUID, "error", err)
}
// 记录当日社区分享计数（统计）。
slog.InfoContext(ctx, "outfit published", "outfit_uuid", outfit.OutfitUUID, "user_id", userID)
return &dto.OutfitPublicResponse{}, nil
}
@@ -303,6 +330,8 @@ func (s *OutfitService) getOwnedOutfit(ctx context.Context, outfitUUID string, u
if isNotFound(err) {
return nil, errcode.OutfitNotFound
}
slog.ErrorContext(ctx, "get owned outfit: query failed",
"outfit_uuid", outfitUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
if outfit.UserID != userID {
internal/service/outfit_share.go
+24
View File
@@ -46,6 +46,8 @@ func (s *OutfitShareService) Create(ctx context.Context, userID int64, request *
if isNotFound(err) {
return nil, errcode.OutfitNotFound
}
slog.ErrorContext(ctx, "outfit share create: query outfit failed",
"outfit_uuid", request.OutfitUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
if outfit.UserID != userID {
@@ -61,6 +63,11 @@ func (s *OutfitShareService) Create(ctx context.Context, userID int64, request *
charged := outfit.IsPublic == OutfitPrivate
if charged {
if qerr := consumeDailyQuota(ctx, userID, QuotaTypeClosetShare, 1, "私密穿搭分享消耗"); qerr != nil {
if qerr == errcode.QuotaInsufficient {
slog.InfoContext(ctx, "outfit share create: quota insufficient", "user_id", userID)
} else {
slog.ErrorContext(ctx, "outfit share create: consume quota failed", "user_id", userID, "error", qerr)
}
return nil, qerr
}
}
@@ -72,12 +79,16 @@ func (s *OutfitShareService) Create(ctx context.Context, userID int64, request *
Status: ShareStatusNormal,
}
if err := s.shareDao.OutfitShareCreate(ctx, share); err != nil {
slog.ErrorContext(ctx, "outfit share create: save share failed",
"user_id", userID, "outfit_uuid", request.OutfitUUID, "error", err)
// 分享落库失败：回退刚扣的额度，避免用户额度被白扣。
if charged {
refundDailyQuota(ctx, userID, QuotaTypeClosetShare, 1)
}
return nil, errcode.DatabaseError
}
slog.InfoContext(ctx, "outfit share created",
"user_id", userID, "share_uuid", share.ShareUUID, "outfit_uuid", request.OutfitUUID, "charged", charged)
return &dto.OutfitShareCreateResponse{ShareUUID: share.ShareUUID}, nil
}
@@ -91,6 +102,7 @@ func (s *OutfitShareService) Detail(ctx context.Context, userID int64, shareUUID
if isNotFound(err) {
return nil, errcode.OutfitShareNotFound
}
slog.ErrorContext(ctx, "outfit share detail: query share failed", "share_uuid", shareUUID, "error", err)
return nil, errcode.DatabaseError
}
if share.Status == ShareStatusCancelled {
@@ -129,6 +141,9 @@ func (s *OutfitShareService) Detail(ctx context.Context, userID int64, shareUUID
if owner, uerr := s.userDao.UserGetByID(ctx, share.UserID); uerr == nil {
ownerName = owner.Username
ownerAvatar = upload.AvatarURL(owner.Avatar)
} else {
slog.DebugContext(ctx, "outfit share detail: get owner info failed",
"share_uuid", shareUUID, "owner_id", share.UserID, "error", uerr)
}
return &dto.OutfitShareDetailResponse{
@@ -210,14 +225,21 @@ func (s *OutfitShareService) Cancel(ctx context.Context, userID int64, request *
if isNotFound(err) {
return nil, errcode.OutfitShareNotFound
}
slog.ErrorContext(ctx, "outfit share cancel: query share failed",
"share_uuid", request.ShareUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
if share.UserID != userID {
slog.InfoContext(ctx, "outfit share cancel: permission denied",
"share_uuid", request.ShareUUID, "user_id", userID, "owner_id", share.UserID)
return nil, errcode.Forbidden
}
if err := s.shareDao.OutfitShareUpdateStatus(ctx, request.ShareUUID, ShareStatusCancelled); err != nil {
slog.ErrorContext(ctx, "outfit share cancel: update status failed",
"share_uuid", request.ShareUUID, "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
slog.InfoContext(ctx, "outfit share cancelled", "share_uuid", request.ShareUUID, "user_id", userID)
return &dto.OutfitShareCancelResponse{}, nil
}
@@ -266,6 +288,8 @@ func (s *OutfitShareService) buildShareItems(ctx context.Context, list []mysqlmo
for i := range clothesList {
coverByClothesID[clothesList[i].ID] = clothesImageURL(clothesList[i].ProcessedImage)
}
} else {
slog.DebugContext(ctx, "outfit share list: load cover images failed", "error", cerr)
}
for _, sh := range list {
internal/service/quota.go
+3
View File
@@ -36,6 +36,8 @@ func (s *QuotaService) InfoGetBySelf(ctx context.Context, userID int64) (*dto.Qu
for _, quotaType := range types {
eff, err := getEffectiveQuota(ctx, userID, quotaType)
if err != nil {
slog.ErrorContext(ctx, "quota info: calc effective quota failed",
"user_id", userID, "quota_type", quotaType, "error", err)
return nil, err
}
items = append(items, dto.QuotaItemResponse{
@@ -58,6 +60,7 @@ func (s *QuotaService) FlowList(ctx context.Context, userID int64, request *dto.
list, total, err := s.flowLogDao.QuotaFlowLogListByUserID(ctx, userID, request.QuotaType, page, pageSize)
if err != nil {
slog.ErrorContext(ctx, "quota flow list: query failed", "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
internal/service/quota_expire.go
+120
View File
@@ -0,0 +1,120 @@
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)
}
}
internal/service/service.go
+20 -8
View File
@@ -103,6 +103,10 @@ type EffectiveQuota struct {
func getUserVIPExpire(ctx context.Context, userID int64) int64 {
user, err := mysqldao.NewUserDao().UserGetByID(ctx, userID)
if err != nil {
// 查询失败一律视为非会员（跟随 VIP 的凭证将按失效处理），必须留痕便于排查额度差异。
if !isNotFound(err) {
slog.WarnContext(ctx, "get user vip expire failed, treat as non-vip", "user_id", userID, "error", err)
}
return 0
}
return user.VIPExpire
@@ -140,6 +144,8 @@ func getEffectiveQuota(ctx context.Context, userID int64, quotaType int8) (*Effe
vipExpire := getUserVIPExpire(ctx, userID)
grants, err := listValidGrants(ctx, userID, quotaType, vipExpire, now)
if err != nil {
slog.ErrorContext(ctx, "get effective quota: list valid grants failed",
"user_id", userID, "quota_type", quotaType, "error", err)
return nil, errcode.DatabaseError
}
@@ -154,6 +160,9 @@ func getEffectiveQuota(ctx context.Context, userID int64, quotaType int8) (*Effe
freeUsed := int64(0)
if q, qerr := mysqldao.NewQuotaDao().QuotaGetByUserIDAndType(ctx, userID, quotaType); qerr == nil {
freeUsed = q.Used
} else if !isNotFound(qerr) {
slog.WarnContext(ctx, "get effective quota: query free quota failed, treat as 0",
"user_id", userID, "quota_type", quotaType, "error", qerr)
}
total := freeTotal + grantTotal
used := freeUsed + grantUsed
@@ -329,7 +338,7 @@ func consumeCosQuota(ctx context.Context, userID int64, n int64, relatedID int64
}
// 3. 写消费流水（old/new 记录 Cos 有效剩余）。
before := cosRemainTx(tx, userID, vipExpire, now)
before := cosRemainTx(ctx, tx, userID, vipExpire, now)
flow := &mysqlmodel.QuotaFlowLog{
QuotaFlowUUID: genUUID(),
UserID: userID,
@@ -351,13 +360,15 @@ func consumeCosQuota(ctx context.Context, userID int64, n int64, relatedID int64
if errors.Is(err, errcode.QuotaInsufficient) {
return errcode.QuotaInsufficient
}
slog.ErrorContext(ctx, "consume cos quota: transaction failed",
"user_id", userID, "amount", n, "related_id", relatedID, "error", err)
return errcode.DatabaseError
}
return nil
}
// cosRemainTx 在事务内计算 Cos 有效剩余（免费剩余 + 有效凭证剩余）。
func cosRemainTx(tx *gorm.DB, userID int64, vipExpire, now int64) int64 {
func cosRemainTx(ctx context.Context, tx *gorm.DB, userID int64, vipExpire, now int64) int64 {
freeRemain := int64(0)
var freeQ mysqlmodel.Quota
if err := tx.Where("user_id = ? AND quota_type = ?", userID, QuotaTypeCos).First(&freeQ).Error; err == nil {
@@ -372,7 +383,7 @@ func cosRemainTx(tx *gorm.DB, userID int64, vipExpire, now int64) int64 {
q = q.Where("expire_time > ?", now)
}
if err := q.Select("COALESCE(SUM(amount - used), 0)").Scan(&grantRemain).Error; err != nil {
slog.WarnContext(context.Background(), "calc cos grant remain failed", "user_id", userID, "error", err)
slog.WarnContext(ctx, "calc cos grant remain failed", "user_id", userID, "error", err)
}
if freeRemain < 0 {
freeRemain = 0
@@ -382,7 +393,7 @@ func cosRemainTx(tx *gorm.DB, userID int64, vipExpire, now int64) int64 {
// refundCosQuotaTx 在事务内回退 Cos 额度并写流水。
// 回退策略（后进先出）：优先回退免费额度（最后扣减的），不足再回退凭证（最后到期优先）。
func refundCosQuotaTx(tx *gorm.DB, userID int64, vipExpire int64, n int64, relatedID int64, reason string) error {
func refundCosQuotaTx(ctx context.Context, tx *gorm.DB, userID int64, vipExpire int64, n int64, relatedID int64, reason string) error {
now := nowUnix()
remain := n
@@ -431,10 +442,10 @@ func refundCosQuotaTx(tx *gorm.DB, userID int64, vipExpire int64, n int64, relat
flowReason := reason
if refunded < n {
flowReason = fmt.Sprintf("%s（实际回退 %d/%d，可用已退额度不足）", reason, refunded, n)
slog.WarnContext(context.Background(), "refund cos quota partially applied",
slog.WarnContext(ctx, "refund cos quota partially applied",
"user_id", userID, "want", n, "refunded", refunded, "related_id", relatedID)
}
after := cosRemainTx(tx, userID, vipExpire, now)
after := cosRemainTx(ctx, tx, userID, vipExpire, now)
flow := &mysqlmodel.QuotaFlowLog{
QuotaFlowUUID: genUUID(),
UserID: userID,
@@ -456,8 +467,9 @@ func refundCosQuotaTx(tx *gorm.DB, userID int64, vipExpire int64, n int64, relat
func refundCosQuota(ctx context.Context, userID int64, n int64, relatedID int64, reason string) {
vipExpire := getUserVIPExpire(ctx, userID)
if err := global.MysqlEngine.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
return refundCosQuotaTx(tx, userID, vipExpire, n, relatedID, reason)
return refundCosQuotaTx(ctx, tx, userID, vipExpire, n, relatedID, reason)
}); err != nil {
slog.WarnContext(ctx, "refund cos quota failed", "user_id", userID, "error", err)
slog.WarnContext(ctx, "refund cos quota failed",
"user_id", userID, "amount", n, "related_id", relatedID, "reason", reason, "error", err)
}
}
internal/service/user.go
+170 -16
View File
@@ -43,8 +43,10 @@ func (s *UserService) WxLogin(ctx context.Context, request *dto.UserWxLoginReque
if err != nil {
var codeErr *wechat.Code2SessionError
if errors.As(err, &codeErr) && codeErr.IsInvalidCode() {
slog.WarnContext(ctx, "wx login: invalid code", "error", err)
return nil, errcode.UserWxCodeInvalid
}
slog.ErrorContext(ctx, "wx login: code2session failed", "error", err)
return nil, errcode.UserWxLoginFailed
}
@@ -55,6 +57,7 @@ func (s *UserService) WxLogin(ctx context.Context, request *dto.UserWxLoginReque
user, err := s.userDao.UserGetByOpenID(ctx, sess.OpenID)
if err != nil {
if !isNotFound(err) {
slog.ErrorContext(ctx, "wx login: query user failed", "error", err)
return nil, errcode.DatabaseError
}
// 首次登录：注册新用户。
@@ -70,11 +73,14 @@ func (s *UserService) WxLogin(ctx context.Context, request *dto.UserWxLoginReque
if cerr := s.userDao.UserCreate(ctx, user); cerr != nil {
if isDuplicateKey(cerr) {
// 并发注册冲突：重查已存在的用户继续登录。
slog.InfoContext(ctx, "wx login: concurrent register conflict, reload user")
user, err = s.userDao.UserGetByOpenID(ctx, sess.OpenID)
if err != nil {
slog.ErrorContext(ctx, "wx login: reload user after register conflict failed", "error", err)
return nil, errcode.DatabaseError
}
} else {
slog.ErrorContext(ctx, "wx login: create user failed", "error", cerr)
return nil, errcode.UserCreateFailed
}
} else {
@@ -83,7 +89,8 @@ func (s *UserService) WxLogin(ctx context.Context, request *dto.UserWxLoginReque
ensureDefaultQuota(ctx, user.ID)
// 赠送体验会员（best-effort）；赠送成功后回读用户，
// 保证后续写入缓存与登录响应的是会员态，避免被内存中的非会员对象覆盖。
if s.ensureTrialVIP(ctx, user.ID) > 0 {
trialVIPExpire := s.ensureTrialVIP(ctx, user.ID)
if trialVIPExpire > 0 {
if fresh, ferr := s.userDao.UserGetByID(ctx, user.ID); ferr == nil {
user = fresh
} else {
@@ -92,25 +99,32 @@ func (s *UserService) WxLogin(ctx context.Context, request *dto.UserWxLoginReque
userCacheStale = true
}
}
// 赠送体验会员额度（best-effort）：有效期固定为体验期结束时刻（trialVIPExpire），
// 不跟随 VIP 延长 —— 否则用户后续充值会把体验额度一直「继承」下去。
s.ensureTrialVIPQuota(ctx, user.ID, trialVIPExpire)
}
}
// 非正常状态账号（临时/永久封禁、已注销）禁止登录。
if user.Status != UserStatusNormal {
slog.WarnContext(ctx, "wx login: user disabled", "user_id", user.ID, "status", user.Status)
return nil, errcode.UserDisabled
}
if !isNew {
_ = s.userDao.UserUpdateByID(ctx, user.ID, map[string]any{"last_login_time": nowUnix()})
if uerr := s.userDao.UserUpdateByID(ctx, user.ID, map[string]any{"last_login_time": nowUnix()}); uerr != nil {
slog.WarnContext(ctx, "wx login: update last login time failed", "user_id", user.ID, "error", uerr)
}
}
accessToken, refreshToken, terr := lzcwyjwt.GenerateTokens(user.ID)
if terr != nil {
slog.ErrorContext(ctx, "wx login: generate tokens failed", "user_id", user.ID, "error", terr)
return nil, errcode.UnauthorizedTokenGenerate
}
if aerr := redisdao.AuthTokenSet(ctx, user.ID, accessToken, refreshToken); aerr != nil {
// Redis 故障降级放行（与鉴权中间件 fail-open 策略一致），避免 Redis 抖动导致全员无法登录。
slog.WarnContext(ctx, "set auth token failed, allow by fallback", "error", aerr)
slog.WarnContext(ctx, "set auth token failed, allow by fallback", "user_id", user.ID, "error", aerr)
}
// 缓存 session_key（虚拟支付 signature 计算依赖；best-effort，失败不阻断登录，
@@ -123,12 +137,16 @@ func (s *UserService) WxLogin(ctx context.Context, request *dto.UserWxLoginReque
// 刷新缓存中的用户信息（best-effort）；对象已落后于数据库时不回写。
if !userCacheStale {
_ = redisdao.UserInfoSet(ctx, user, 0)
if cerr := redisdao.UserInfoSet(ctx, user, 0); cerr != nil {
slog.WarnContext(ctx, "wx login: cache user info failed", "user_id", user.ID, "error", cerr)
}
}
info := buildUserInfoResponse(user)
info.IsNewUser = isNew
slog.InfoContext(ctx, "user login", "user_id", user.ID, "is_new_user", isNew)
return &dto.UserWxLoginResponse{
AccessToken: accessToken,
RefreshToken: refreshToken,
@@ -148,16 +166,21 @@ func (s *UserService) InfoGetBySelf(ctx context.Context, userID int64) (*dto.Use
return &info, nil
} else if err != nil && !isCacheMiss(err) {
// 缓存异常不阻断，继续回落 DB。
slog.WarnContext(ctx, "get user info: cache unavailable, fallback to db", "user_id", userID, "error", err)
}
user, err := s.userDao.UserGetByID(ctx, userID)
if err != nil {
if isNotFound(err) {
slog.WarnContext(ctx, "get user info: user not found", "user_id", userID)
return nil, errcode.UserNotFound
}
slog.ErrorContext(ctx, "get user info: query user failed", "user_id", userID, "error", err)
return nil, errcode.DatabaseError
}
_ = redisdao.UserInfoSet(ctx, user, 0)
if cerr := redisdao.UserInfoSet(ctx, user, 0); cerr != nil {
slog.WarnContext(ctx, "get user info: backfill cache failed", "user_id", userID, "error", cerr)
}
info := buildUserInfoResponse(user)
s.consumeNewUserFlag(userID, &info)
@@ -216,10 +239,13 @@ func (s *UserService) ProfileUpdate(ctx context.Context, userID int64, request *
}
if err := s.userDao.UserUpdateByID(ctx, userID, updates); err != nil {
slog.ErrorContext(ctx, "profile update: update user failed", "user_id", userID, "error", err)
return nil, errcode.UserUpdateFailed
}
// 失效缓存，下次读取回填最新数据。
_ = redisdao.UserInfoDel(ctx, userID)
if derr := redisdao.UserInfoDel(ctx, userID); derr != nil {
slog.DebugContext(ctx, "profile update: invalidate user cache failed", "user_id", userID, "error", derr)
}
return &dto.UserProfileUpdateBySelfResponse{}, nil
}
@@ -231,31 +257,37 @@ func (s *UserService) AvatarUpload(ctx context.Context, userID int64, fileHeader
defer span.End()
if fileHeader == nil || fileHeader.Size == 0 {
slog.DebugContext(ctx, "avatar upload: empty file", "user_id", userID)
return nil, errcode.UploadFileEmpty
}
// 1. 校验扩展名与大小。
ext := upload.FileGetExt(fileHeader.Filename)
if !upload.FileCheckAllowExt(ext, global.UploadSetting.AllowedExtensions) {
slog.DebugContext(ctx, "avatar upload: extension not allowed", "user_id", userID, "ext", ext)
return nil, errcode.UploadFileExtNotAllow
}
if !upload.FileCheckMaxSize(fileHeader.Size, int64(global.UploadSetting.MaxSizeMB)) {
slog.DebugContext(ctx, "avatar upload: file too large", "user_id", userID, "size", fileHeader.Size)
return nil, errcode.UploadFileTooLarge
}
// 2. 读取原图字节（必须同步读取，请求返回后 multipart 临时文件会被清理）。
src, oerr := fileHeader.Open()
if oerr != nil {
slog.ErrorContext(ctx, "avatar upload: open file failed", "user_id", userID, "error", oerr)
return nil, errcode.UploadFileSaveFailed
}
imageData, rerr := io.ReadAll(src)
_ = src.Close()
if rerr != nil {
slog.ErrorContext(ctx, "avatar upload: read file failed", "user_id", userID, "error", rerr)
return nil, errcode.UploadFileSaveFailed
}
// 3. 校验文件内容为真实图片（magic bytes），防止伪造扩展名上传恶意文件。
if !upload.FileCheckImageMagic(imageData) {
slog.WarnContext(ctx, "avatar upload: magic bytes check failed", "user_id", userID, "size", fileHeader.Size)
return nil, errcode.UploadFileExtNotAllow
}
@@ -263,17 +295,21 @@ func (s *UserService) AvatarUpload(ctx context.Context, userID int64, fileHeader
user, uerr := s.userDao.UserGetByID(ctx, userID)
if uerr != nil {
if isNotFound(uerr) {
slog.WarnContext(ctx, "avatar upload: user not found", "user_id", userID)
return nil, errcode.UserNotFound
}
slog.ErrorContext(ctx, "avatar upload: query user failed", "user_id", userID, "error", uerr)
return nil, errcode.DatabaseError
}
fileName, nerr := upload.FileGenName(fileHeader.Filename)
if nerr != nil {
slog.ErrorContext(ctx, "avatar upload: generate file name failed", "user_id", userID, "error", nerr)
return nil, errcode.UploadFileSaveFailed
}
subDir := fmt.Sprintf("%s/%d/avatar", user.UserUUID, userID)
tmpPath, serr := upload.FileSaveTmpBytes(ctx, imageData, fileName, subDir)
if serr != nil {
slog.ErrorContext(ctx, "avatar upload: save temp file failed", "user_id", userID, "error", serr)
return nil, errcode.UploadFileSaveFailed
}
// 头像不消耗额度、上传接口也无限流，临时文件必须清理，
@@ -281,16 +317,20 @@ func (s *UserService) AvatarUpload(ctx context.Context, userID int64, fileHeader
defer func() { _ = os.Remove(tmpPath) }()
cosKey := fmt.Sprintf("%s/%s", subDir, fileName)
if cerr := upload.FileUploadCos(ctx, cosKey); cerr != nil {
slog.ErrorContext(ctx, "avatar upload: upload to cos failed", "user_id", userID, "cos_key", cosKey, "error", cerr)
return nil, errcode.UploadFileSaveFailed
}
// 5. 更新 avatar（存 COS key，读取时统一转 URL）。
if uerr := s.userDao.UserUpdateByID(ctx, userID, map[string]any{"avatar": cosKey}); uerr != nil {
slog.ErrorContext(ctx, "avatar upload: update user avatar failed", "user_id", userID, "error", uerr)
return nil, errcode.UserUpdateFailed
}
// 6. 失效缓存，下次读取回填最新头像。
_ = redisdao.UserInfoDel(ctx, userID)
if derr := redisdao.UserInfoDel(ctx, userID); derr != nil {
slog.DebugContext(ctx, "avatar upload: invalidate user cache failed", "user_id", userID, "error", derr)
}
return &dto.UserAvatarUploadResponse{Avatar: upload.GetCosUrl(cosKey)}, nil
}
@@ -303,8 +343,10 @@ func (s *UserService) RefreshToken(ctx context.Context, request *dto.UserRefresh
claims, err := lzcwyjwt.ParseRefreshToken(request.RefreshToken)
if err != nil {
if errors.Is(err, lzcwyjwt.ErrTokenExpired) {
slog.InfoContext(ctx, "refresh token: token expired")
return nil, errcode.UnauthorizedTokenTimeout
}
slog.WarnContext(ctx, "refresh token: parse token failed", "error", err)
return nil, errcode.UnauthorizedTokenError
}
@@ -312,35 +354,47 @@ func (s *UserService) RefreshToken(ctx context.Context, request *dto.UserRefresh
uid, aerr := redisdao.AuthUserIDByRefresh(ctx, request.RefreshToken)
if aerr != nil {
if isCacheMiss(aerr) {
slog.WarnContext(ctx, "refresh token: session not found", "user_id", claims.UserID)
return nil, errcode.UnauthorizedTokenError
}
slog.ErrorContext(ctx, "refresh token: get session failed", "user_id", claims.UserID, "error", aerr)
return nil, errcode.CacheError
}
if uid != claims.UserID {
slog.ErrorContext(ctx, "refresh token: session user mismatch", "user_id", claims.UserID, "session_user_id", uid)
return nil, errcode.UnauthorizedTokenError
}
accessToken, refreshToken, terr := lzcwyjwt.GenerateTokens(claims.UserID)
if terr != nil {
slog.ErrorContext(ctx, "refresh token: generate tokens failed", "user_id", claims.UserID, "error", terr)
return nil, errcode.UnauthorizedTokenGenerate
}
// 失效旧会话：强制携带旧 access token，精确吊销旧 access + 旧 refresh，防止旧 token 泄露后被继续使用。
if request.AccessToken == "" {
slog.InfoContext(ctx, "refresh token: old access token missing", "user_id", claims.UserID)
return nil, errcode.InvalidParams.WithDetails("缺少旧 access token")
}
oldUID, oerr := redisdao.AuthUserIDByAccess(ctx, request.AccessToken)
if oerr != nil && !isCacheMiss(oerr) {
slog.ErrorContext(ctx, "refresh token: get old access session failed", "user_id", claims.UserID, "error", oerr)
return nil, errcode.CacheError
}
if oerr == nil && oldUID != claims.UserID {
slog.ErrorContext(ctx, "refresh token: old access token user mismatch", "user_id", claims.UserID, "old_user_id", oldUID)
return nil, errcode.UnauthorizedTokenError
}
_ = redisdao.AuthTokenDel(ctx, claims.UserID, request.AccessToken, request.RefreshToken)
if derr := redisdao.AuthTokenDel(ctx, claims.UserID, request.AccessToken, request.RefreshToken); derr != nil {
slog.ErrorContext(ctx, "refresh token: revoke old tokens failed", "user_id", claims.UserID, "error", derr)
}
if serr := redisdao.AuthTokenSet(ctx, claims.UserID, accessToken, refreshToken); serr != nil {
slog.ErrorContext(ctx, "refresh token: save new tokens failed", "user_id", claims.UserID, "error", serr)
return nil, errcode.CacheError
}
slog.DebugContext(ctx, "refresh token: tokens rotated", "user_id", claims.UserID)
return &dto.UserRefreshTokenResponse{
AccessToken: accessToken,
RefreshToken: refreshToken,
@@ -349,14 +403,16 @@ func (s *UserService) RefreshToken(ctx context.Context, request *dto.UserRefresh
}
// defaultVipDays 返回新用户注册赠送的体验会员天数：
// 优先取 UserDefaultQuota.GiftVIPDay 的配置值，配置为 0 时生效（表示不赠送）；
// 未配置（整段缺失或该项缺失，指针为 nil）或配置为负数时，回落内置常量 DefaultVipDays。
// 优先取 NewUserGift.GiftVIPDay（新位置）；未配置该项时回落 UserDefaultQuota.GiftVIPDay（旧位置，兼容既有配置）；
// 两者都不可用时回落内置常量 DefaultVipDays。显式配置为 0 表示不赠送；负数视为非法配置，继续往后回落。
func defaultVipDays() int64 {
cfg := global.UserDefaultQuotaSetting
if cfg == nil || cfg.GiftVIPDay == nil || *cfg.GiftVIPDay < 0 {
return DefaultVipDays
if cfg := global.NewUserGiftSetting; cfg != nil && cfg.GiftVIPDay != nil && *cfg.GiftVIPDay >= 0 {
return *cfg.GiftVIPDay
}
return *cfg.GiftVIPDay
if cfg := global.UserDefaultQuotaSetting; cfg != nil && cfg.GiftVIPDay != nil && *cfg.GiftVIPDay >= 0 {
return *cfg.GiftVIPDay
}
return DefaultVipDays
}
// ensureTrialVIP 新用户注册赠送体验会员（best-effort：任何失败仅告警，不阻断注册）。
@@ -397,7 +453,10 @@ func (s *UserService) ensureTrialVIP(ctx context.Context, userID int64) int64 {
}
// 会员信息已变更，失效缓存（注册流程随后会用回读后的用户重新写入）。
_ = redisdao.UserInfoDel(ctx, userID)
if derr := redisdao.UserInfoDel(ctx, userID); derr != nil {
slog.DebugContext(ctx, "grant trial vip: invalidate user cache failed", "user_id", userID, "error", derr)
}
slog.InfoContext(ctx, "trial vip granted", "user_id", userID, "days", days, "vip_expire", flow.NewVIPExpire)
sendNotification(ctx, userID, 0, NotificationTypeSystem, "欢迎礼包已到账",
fmt.Sprintf("已为你开通 %d 天体验会员，有效期至 %s",
@@ -406,6 +465,96 @@ func (s *UserService) ensureTrialVIP(ctx context.Context, userID int64) int64 {
return flow.NewVIPExpire
}
// trialVIPQuotaValue 返回体验会员赠送的某类额度数量（NewUserGift 段配置）。
// 段缺失、该项未配置（指针为 nil）或非正数一律返回 0，表示不赠送该类额度。
func trialVIPQuotaValue(quotaType int8) int64 {
cfg := global.NewUserGiftSetting
if cfg == nil {
return 0
}
switch quotaType {
case QuotaTypeCos:
if cfg.Cos != nil && *cfg.Cos > 0 {
return *cfg.Cos
}
case QuotaTypeAI:
if cfg.AI != nil && *cfg.AI > 0 {
return *cfg.AI
}
case QuotaTypeOutfitShare:
if cfg.OutfitShare != nil && *cfg.OutfitShare > 0 {
return *cfg.OutfitShare
}
case QuotaTypeClosetShare:
if cfg.ClosetShare != nil && *cfg.ClosetShare > 0 {
return *cfg.ClosetShare
}
}
return 0
}
// ensureTrialVIPQuota 新用户注册赠送体验会员额度（best-effort：任何失败仅告警，不阻断注册）。
//
// expireAt 为体验期结束时刻（取自 ensureTrialVIP 的实际发放结果）。
//
// 这里刻意使用「固定到期时间」而不是 expire_time=0（跟随 VIP）：
// 跟随 VIP 时 grantExpired 判断的是用户「当前」的 vip_expire，用户后续一充值就会把
// vip_expire 推后，体验额度被续费「继承」下去、永远不失效，失去「体验」的语义。
//
// 幂等：ref_id 固定为 register_gift_<userID>，依赖 quota_grants 的 (ref_id, quota_type) 唯一索引，
// 重复注册或重试不会重复发放，也不会重复写发放流水。
func (s *UserService) ensureTrialVIPQuota(ctx context.Context, userID, expireAt int64) {
if expireAt <= 0 {
// 未赠送体验会员（GiftVIPDay<=0），或本次未实际发放（幂等命中）：
// 没有明确的体验期就不发额度，避免写入语义不清的凭证。
return
}
ctx, span := global.Tracer.Start(ctx, "svc.EnsureTrialVIPQuota")
defer span.End()
refID := fmt.Sprintf("register_gift_%d", userID)
grantDao := mysqldao.NewQuotaGrantDao()
for _, quotaType := range []int8{QuotaTypeCos, QuotaTypeAI, QuotaTypeOutfitShare, QuotaTypeClosetShare} {
amount := trialVIPQuotaValue(quotaType)
if amount <= 0 {
continue // 未配置或显式配 0：不赠送该类额度
}
grant := &mysqlmodel.QuotaGrant{
UserID: userID,
QuotaType: quotaType,
Amount: amount,
Source: QuotaGrantSourceTrial, // 体验会员赠送（与购买会员包 source=1 区分）
ExpireTime: expireAt, // 固定到期时间，不随后续续费延长
RefID: refID,
}
// 变更前后取「有效剩余」，仅用于额度明细展示（权威数值以凭证表为准），与购买发放路径保持一致。
before := int64(0)
if eff, qerr := getEffectiveQuota(ctx, userID, quotaType); qerr == nil {
before = eff.Remain
}
flow := &mysqlmodel.QuotaFlowLog{
QuotaFlowUUID: genUUID(),
UserID: userID,
ChangeTarget: QuotaChangeTargetAdjust,
ChangeType: QuotaChangeTypeActivity,
ChangeNum: amount,
OldValue: before,
NewValue: before + amount,
QuotaType: quotaType,
Reason: "新用户注册赠送体验会员额度",
}
if _, err := grantDao.QuotaGrantCreateWithFlow(ctx, grant, flow); err != nil {
slog.WarnContext(ctx, "grant trial vip quota failed",
"user_id", userID, "quota_type", quotaType, "amount", amount, "error", err)
}
}
}
// buildUserInfoResponse 将 User 模型映射为用户信息响应。
// 会员已过期（vip_expire <= now）时把 vip_type 归一为 1：项目没有会员到期归位任务，
// 过期会员的 vip_type 仍保留原值，归一可避免调用方仅凭 vip_type 误判为有效会员（只影响响应，不写库）。
@@ -440,10 +589,15 @@ func buildUserInfoResponse(user *mysqlmodel.User) dto.UserInfoGetBySelfResponse
func (s *UserService) resetNewUserFlag(userID int64) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
// 异步任务无请求 ctx，派生独立 span 保证该路径日志携带 trace_id。
ctx, span := global.Tracer.Start(ctx, "svc.ResetNewUserFlag")
defer span.End()
if err := s.userDao.UserUpdateByID(ctx, userID, map[string]any{"is_new_user": false}); err != nil {
slog.WarnContext(ctx, "reset is_new_user failed", "user_id", userID, "error", err)
return
}
_ = redisdao.UserInfoDel(ctx, userID)
if derr := redisdao.UserInfoDel(ctx, userID); derr != nil {
slog.DebugContext(ctx, "reset is_new_user: invalidate user cache failed", "user_id", userID, "error", derr)
}
}
internal/service/virtualpay.go
+84 -12
View File
@@ -24,25 +24,32 @@ func (s *OrderService) VirtualOrderCreate(ctx context.Context, userID int64, req
defer span.End()
if !wechat.IsVirtualPayEnabled() {
slog.WarnContext(ctx, "virtual order create: virtual pay not enabled", "user_id", userID)
return nil, errcode.VirtualPayNotEnabled
}
// 订单记录当前激活的虚拟支付环境（现网 0 / 沙箱 1），发货推送按此校验。
xpayEnv, verr := wechat.GetVirtualPayEnv()
if verr != nil {
slog.ErrorContext(ctx, "virtual order create: resolve pay env failed", "user_id", userID, "error", verr)
return nil, errcode.VirtualPayNotEnabled
}
goods, err := s.goodsDao.GoodsGetByCode(ctx, request.GoodsCode)
if err != nil {
if isNotFound(err) {
slog.InfoContext(ctx, "virtual order create: goods not found", "user_id", userID, "goods_code", request.GoodsCode)
return nil, errcode.GoodsNotFound
}
slog.ErrorContext(ctx, "virtual order create: query goods failed", "goods_code", request.GoodsCode, "error", err)
return nil, errcode.DatabaseError
}
if goods.Status != GoodsStatusOnShelf {
slog.InfoContext(ctx, "virtual order create: goods off shelf", "user_id", userID, "goods_code", request.GoodsCode)
return nil, errcode.GoodsOffShelf
}
if goods.ProductID == "" || goods.ProductStatus != GoodsProductPublished {
slog.WarnContext(ctx, "virtual order create: goods not published as xpay product",
"user_id", userID, "goods_code", request.GoodsCode, "product_id", goods.ProductID, "product_status", goods.ProductStatus)
return nil, errcode.VirtualPayGoodsNotPublished
}
// iOS 端微信限制最低支付 1 元（100 分），低于此金额调起支付必失败，提前拒绝以降低垃圾订单。
@@ -50,6 +57,8 @@ func (s *OrderService) VirtualOrderCreate(ctx context.Context, userID int64, req
// 此校验仅作体验优化，不构成资损防线——真正的资损防线是发货推送 verifyXPayOrderPaid
// 对微信侧订单真实性、金额、道具的反查。低金额商品在 iOS 上的垃圾订单靠过期清理兜底。
if request.Platform == XPayPlatformIOS && goods.DiscountPrice < 100 {
slog.InfoContext(ctx, "virtual order create: rejected low amount on ios",
"user_id", userID, "goods_code", request.GoodsCode, "pay_amount", goods.DiscountPrice)
return nil, errcode.VirtualPayAmountTooLow
}
@@ -57,8 +66,10 @@ func (s *OrderService) VirtualOrderCreate(ctx context.Context, userID int64, req
sessionKey, serr := redisdao.AuthSessionKeyGet(ctx, userID)
if serr != nil {
if serr == redis.Nil {
slog.WarnContext(ctx, "virtual order create: session key missing", "user_id", userID)
return nil, errcode.UnauthorizedTokenError.WithDetails("缺少 session_key，请重新登录")
}
slog.ErrorContext(ctx, "virtual order create: get session key failed", "user_id", userID, "error", serr)
return nil, errcode.CacheError
}
@@ -90,6 +101,8 @@ func (s *OrderService) VirtualOrderCreate(ctx context.Context, userID int64, req
XPayEnv: int8(xpayEnv),
}
if cerr := s.orderDao.OrderCreate(ctx, order); cerr != nil {
slog.ErrorContext(ctx, "virtual order create: save order failed",
"order_no", orderNo, "user_id", userID, "goods_code", request.GoodsCode, "error", cerr)
return nil, errcode.VirtualOrderCreateFailed
}
@@ -102,9 +115,13 @@ func (s *OrderService) VirtualOrderCreate(ctx context.Context, userID int64, req
payAmount,
)
if perr != nil {
slog.ErrorContext(ctx, "virtual order create: build payment sign failed", "order_no", orderNo, "user_id", userID, "error", perr)
return nil, errcode.VirtualOrderCreateFailed
}
slog.InfoContext(ctx, "virtual order created",
"order_no", orderNo, "user_id", userID, "goods_code", request.GoodsCode, "pay_amount", payAmount, "env", xpayEnv)
return &dto.VirtualOrderCreateResponse{
OrderNo: orderNo,
PayAmount: payAmount,
@@ -136,7 +153,8 @@ func (s *OrderService) XPayNotify(ctx context.Context, raw []byte) []byte {
return wechat.BuildNotifyResponse(false)
}
if e := s.handleGoodsDeliver(ctx, &deliver, string(raw)); e != nil {
slog.ErrorContext(ctx, "xpay deliver notify handle failed", "error", e)
// 内部失败分支已各自记录 Error 日志，这里仅保留调用链痕迹，避免重复告警。
slog.DebugContext(ctx, "xpay deliver notify handle failed", "error", e)
return wechat.BuildNotifyResponse(false)
}
return wechat.BuildNotifyResponse(true)
@@ -148,7 +166,8 @@ func (s *OrderService) XPayNotify(ctx context.Context, raw []byte) []byte {
return wechat.BuildNotifyResponse(false)
}
if e := s.handleRefundNotify(ctx, &refund); e != nil {
slog.ErrorContext(ctx, "xpay refund notify handle failed", "error", e)
// 内部失败分支已各自记录 Error 日志，这里仅保留调用链痕迹，避免重复告警。
slog.DebugContext(ctx, "xpay refund notify handle failed", "error", e)
return wechat.BuildNotifyResponse(false)
}
return wechat.BuildNotifyResponse(true)
@@ -241,7 +260,13 @@ func (s *OrderService) verifyXPayRefund(ctx context.Context, order *mysqlmodel.O
func (s *OrderService) handleGoodsDeliver(ctx context.Context, deliver *wechat.XPayGoodsDeliverNotify, raw string) *errcode.Error {
lockKey := fmtLockKey(redismodel.KeyLockPayCallback, deliver.OutTradeNo)
lockVal := genUUID()
if ok, _ := redisdao.TryLock(ctx, lockKey, lockVal, 30*time.Second); !ok {
ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, 30*time.Second)
if lerr != nil {
slog.ErrorContext(ctx, "xpay deliver: acquire lock failed", "out_trade_no", deliver.OutTradeNo, "error", lerr)
return nil // 锁不可用（Redis 故障）：不重复处理，交由微信重试/持锁者。
}
if !ok {
slog.DebugContext(ctx, "xpay deliver: concurrent callback, skip", "out_trade_no", deliver.OutTradeNo)
return nil // 并发回调，交由持锁者处理。
}
defer func() { _ = redisdao.Unlock(ctx, lockKey, lockVal) }()
@@ -252,6 +277,7 @@ func (s *OrderService) handleGoodsDeliver(ctx context.Context, deliver *wechat.X
slog.ErrorContext(ctx, "xpay deliver: order not found", "out_trade_no", deliver.OutTradeNo)
return errcode.OrderRealityCheckFailed
}
slog.ErrorContext(ctx, "xpay deliver: query order failed", "out_trade_no", deliver.OutTradeNo, "error", gerr)
return errcode.DatabaseError
}
@@ -287,9 +313,12 @@ func (s *OrderService) handleGoodsDeliver(ctx context.Context, deliver *wechat.X
// 条件流转：待支付 -> 已支付，防重复回调。
updated, uerr := s.orderDao.OrderUpdateStatusByOrderNo(ctx, deliver.OutTradeNo, OrderStatusPending, OrderStatusPaid)
if uerr != nil {
slog.ErrorContext(ctx, "xpay deliver: update order status failed", "out_trade_no", deliver.OutTradeNo, "error", uerr)
return errcode.DatabaseError
}
if updated {
slog.InfoContext(ctx, "xpay deliver: order paid",
"order_no", order.OrderNo, "user_id", order.UserID, "pay_amount", order.PayAmount)
s.finalizeVirtualPaidOrder(ctx, order, deliver.WeChatPayInfo.PaidTime, deliver.WeChatPayInfo.TransactionID, raw)
return nil
}
@@ -298,33 +327,46 @@ func (s *OrderService) handleGoodsDeliver(ctx context.Context, deliver *wechat.X
current, gerr := s.orderDao.OrderGetByOrderNo(ctx, deliver.OutTradeNo)
if gerr != nil {
if isNotFound(gerr) {
slog.WarnContext(ctx, "xpay deliver: order disappeared after status conflict", "out_trade_no", deliver.OutTradeNo)
return nil
}
slog.ErrorContext(ctx, "xpay deliver: reload order failed", "out_trade_no", deliver.OutTradeNo, "error", gerr)
return errcode.DatabaseError
}
switch current.OrderStatus {
case OrderStatusPaid:
// 已被并发回调处理，兜底补发（幂等）。
_ = s.orderDao.OrderUpdateByOrderNo(ctx, current.OrderNo, map[string]any{
if uerr := s.orderDao.OrderUpdateByOrderNo(ctx, current.OrderNo, map[string]any{
"pay_time": resolvePaidTime(deliver.WeChatPayInfo.PaidTime),
"transaction_id": deliver.WeChatPayInfo.TransactionID,
"callback_raw": raw,
})
}); uerr != nil {
slog.ErrorContext(ctx, "xpay deliver: reconcile paid order pay info failed", "order_no", current.OrderNo, "error", uerr)
}
slog.InfoContext(ctx, "xpay deliver: order already paid, reconcile benefits", "order_no", current.OrderNo, "user_id", current.UserID)
s.grantOrderBenefits(ctx, current)
case OrderStatusExpired:
// 订单被系统过期但用户已付款：过期 -> 已支付并补发货。
if ok, uerr := s.orderDao.OrderUpdateStatusByOrderNo(ctx, deliver.OutTradeNo, OrderStatusExpired, OrderStatusPaid); uerr != nil {
slog.ErrorContext(ctx, "xpay deliver: compensate expired order failed", "out_trade_no", deliver.OutTradeNo, "error", uerr)
return errcode.DatabaseError
} else if ok {
slog.InfoContext(ctx, "xpay deliver: expired order compensated", "order_no", current.OrderNo, "user_id", current.UserID)
s.finalizeVirtualPaidOrder(ctx, current, deliver.WeChatPayInfo.PaidTime, deliver.WeChatPayInfo.TransactionID, raw)
} else {
slog.DebugContext(ctx, "xpay deliver: expired order already handled by another callback", "out_trade_no", deliver.OutTradeNo)
}
case OrderStatusCancelled:
// 防御分支：历史上已取消的虚拟支付订单用户仍完成了支付，补发货避免「钱已扣、权益不到账」。
// 当前虚拟支付订单已禁止取消，此分支兜底存量数据与并发竞态。
if ok, uerr := s.orderDao.OrderUpdateStatusByOrderNo(ctx, deliver.OutTradeNo, OrderStatusCancelled, OrderStatusPaid); uerr != nil {
slog.ErrorContext(ctx, "xpay deliver: compensate cancelled order failed", "out_trade_no", deliver.OutTradeNo, "error", uerr)
return errcode.DatabaseError
} else if ok {
slog.InfoContext(ctx, "xpay deliver: cancelled order compensated", "order_no", current.OrderNo, "user_id", current.UserID)
s.finalizeVirtualPaidOrder(ctx, current, deliver.WeChatPayInfo.PaidTime, deliver.WeChatPayInfo.TransactionID, raw)
} else {
slog.DebugContext(ctx, "xpay deliver: cancelled order already handled by another callback", "out_trade_no", deliver.OutTradeNo)
}
}
return nil
@@ -334,7 +376,13 @@ func (s *OrderService) handleGoodsDeliver(ctx context.Context, deliver *wechat.X
func (s *OrderService) handleRefundNotify(ctx context.Context, refund *wechat.XPayRefundNotify) *errcode.Error {
lockKey := fmtLockKey(redismodel.KeyLockRefundCallback, refund.RefundOrderID)
lockVal := genUUID()
if ok, _ := redisdao.TryLock(ctx, lockKey, lockVal, 30*time.Second); !ok {
ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, 30*time.Second)
if lerr != nil {
slog.ErrorContext(ctx, "xpay refund: acquire lock failed", "out_refund_no", refund.RefundOrderID, "error", lerr)
return nil // 锁不可用（Redis 故障）：不重复处理，交由微信重试/持锁者。
}
if !ok {
slog.DebugContext(ctx, "xpay refund: concurrent callback, skip", "out_refund_no", refund.RefundOrderID)
return nil
}
defer func() { _ = redisdao.Unlock(ctx, lockKey, lockVal) }()
@@ -343,7 +391,8 @@ func (s *OrderService) handleRefundNotify(ctx context.Context, refund *wechat.XP
// （退款推送按官方文档在退款完成时触发，状态枚举联调时以官方为准）。
if refund.RefundStatus != wechat.XPayRefundStatusSuccess {
slog.WarnContext(ctx, "xpay refund notify not success, skip revoke",
"order_no", refund.OutTradeNo, "refund_status", refund.RefundStatus)
"order_no", refund.OutTradeNo, "refund_order_id", refund.RefundOrderID,
"refund_status", refund.RefundStatus, "refund_fee", refund.RefundFee)
return nil
}
@@ -353,10 +402,12 @@ func (s *OrderService) handleRefundNotify(ctx context.Context, refund *wechat.XP
slog.ErrorContext(ctx, "xpay refund: order not found", "out_trade_no", refund.OutTradeNo)
return nil
}
slog.ErrorContext(ctx, "xpay refund: query order failed", "out_trade_no", refund.OutTradeNo, "error", gerr)
return errcode.DatabaseError
}
// 已终态则幂等返回。
if order.OrderStatus == OrderStatusRefunded {
slog.DebugContext(ctx, "xpay refund: order already refunded, skip", "order_no", order.OrderNo)
return nil
}
@@ -367,7 +418,11 @@ func (s *OrderService) handleRefundNotify(ctx context.Context, refund *wechat.XP
}
// settleRefundSuccess 内部含条件流转（Paid -> Refunded），重复推送不会重复回退。
s.settleRefundSuccess(ctx, order, refund.RefundFee)
// 内部失败已由 settleRefundSuccess 的使用方日志覆盖：此处接收返回值并记录，避免静默。
if serr := s.settleRefundSuccess(ctx, order, refund.RefundFee); serr != nil {
slog.ErrorContext(ctx, "xpay refund: settle refund success failed",
"order_no", order.OrderNo, "refund_order_id", refund.RefundOrderID, "error", serr)
}
return nil
}
@@ -377,20 +432,29 @@ func (s *OrderService) handleRefundNotify(ctx context.Context, refund *wechat.XP
func (s *OrderService) handleIOSRefundQuery(ctx context.Context, query *wechat.XPayIOSRefundQueryNotify) []byte {
order, err := s.orderDao.OrderGetByOrderNo(ctx, query.OutTradeNo)
if err != nil {
slog.ErrorContext(ctx, "xpay ios refund query: order not found", "out_trade_no", query.OutTradeNo, "error", err)
if isNotFound(err) {
slog.ErrorContext(ctx, "xpay ios refund query: order not found", "out_trade_no", query.OutTradeNo)
} else {
slog.ErrorContext(ctx, "xpay ios refund query: query order failed", "out_trade_no", query.OutTradeNo, "error", err)
}
return wechat.BuildIOSRefundQueryResponse(1, "拒绝退款",
"订单号 "+query.OutTradeNo+" 在本地无记录，无法确认权益使用情况，拒绝退款")
}
// 已退款：款项已退回，再次建议退款会造成重复退款资损。
if order.OrderStatus == OrderStatusRefunded {
slog.InfoContext(ctx, "xpay ios refund query: reject, order already refunded", "order_no", order.OrderNo)
return wechat.BuildIOSRefundQueryResponse(1, "拒绝退款",
"订单 "+order.OrderNo+" 已处于已退款状态，不支持重复退款")
}
// 权益已消耗则拒绝退款，防止「已使用权益后白嫖退款」。
if s.iosRefundBenefitsConsumed(ctx, order) {
slog.InfoContext(ctx, "xpay ios refund query: reject, benefits consumed",
"order_no", order.OrderNo, "user_id", order.UserID)
return wechat.BuildIOSRefundQueryResponse(1, "拒绝退款",
"订单 "+order.OrderNo+" 发放的权益已被使用（上传额度已消耗或会员已生效），不支持退款")
}
slog.InfoContext(ctx, "xpay ios refund query: suggest refund, benefits unused",
"order_no", order.OrderNo, "user_id", order.UserID)
return wechat.BuildIOSRefundQueryResponse(0, "建议退款",
"订单 "+order.OrderNo+" 发放的权益尚未使用，建议退款")
}
@@ -401,7 +465,10 @@ func (s *OrderService) iosRefundBenefitsConsumed(ctx context.Context, order *mys
// 额度类：该订单发放的 Cos 凭证已消耗则视为已使用。
if order.CosQuota > 0 {
grants, err := mysqldao.NewQuotaGrantDao().QuotaGrantListByUserIDAndType(ctx, order.UserID, QuotaTypeCos)
if err == nil {
if err != nil {
slog.WarnContext(ctx, "xpay ios refund query: list cos grants failed, treat as not consumed",
"order_no", order.OrderNo, "user_id", order.UserID, "error", err)
} else {
for _, g := range grants {
if g.RefID == order.OrderNo && g.Used > 0 {
return true
@@ -415,6 +482,9 @@ func (s *OrderService) iosRefundBenefitsConsumed(ctx context.Context, order *mys
if user.VIPExpire > nowUnix() && user.VIPType != VIPTypeNone {
return true
}
} else {
slog.WarnContext(ctx, "xpay ios refund query: get user failed, treat vip as not consumed",
"order_no", order.OrderNo, "user_id", order.UserID, "error", err)
}
}
return false
@@ -423,11 +493,13 @@ func (s *OrderService) iosRefundBenefitsConsumed(ctx context.Context, order *mys
// finalizeVirtualPaidOrder 虚拟支付发货：记录支付信息、发放权益并发送通知。
// paidTime/transactionID 由调用方提供：推送路径取推送报文，查单补偿路径取 query_order 结果。
func (s *OrderService) finalizeVirtualPaidOrder(ctx context.Context, order *mysqlmodel.Order, paidTime int64, transactionID, raw string) {
_ = s.orderDao.OrderUpdateByOrderNo(ctx, order.OrderNo, map[string]any{
if uerr := s.orderDao.OrderUpdateByOrderNo(ctx, order.OrderNo, map[string]any{
"pay_time": resolvePaidTime(paidTime),
"transaction_id": transactionID,
"callback_raw": raw,
})
}); uerr != nil {
slog.ErrorContext(ctx, "finalize virtual paid order: update pay info failed", "order_no", order.OrderNo, "error", uerr)
}
s.grantOrderBenefits(ctx, order)
internal/service/virtualpay_compensate.go
+24 -5
View File
@@ -80,10 +80,19 @@ func runVirtualPayConfirm() {
lockKey := redismodel.KeyLockVirtualConfirm
lockVal := genUUID()
ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, virtualConfirmLockTTL)
if lerr != nil || !ok {
if lerr != nil {
slog.ErrorContext(ctx, "virtual confirm: acquire lock failed", "error", lerr)
return
}
defer redisdao.Unlock(context.Background(), lockKey, lockVal)
if !ok {
slog.DebugContext(ctx, "virtual confirm: another instance running, skip")
return
}
defer func() {
if uerr := redisdao.Unlock(context.Background(), lockKey, lockVal); uerr != nil {
slog.WarnContext(ctx, "virtual confirm: release lock failed", "error", uerr)
}
}()
now := nowUnix()
orders, err := mysqldao.NewOrderDao().
@@ -93,6 +102,7 @@ func runVirtualPayConfirm() {
return
}
if len(orders) == 0 {
slog.DebugContext(ctx, "virtual confirm: no unconfirmed orders in window")
return
}
if len(orders) >= virtualConfirmBatchSize {
@@ -121,6 +131,7 @@ func runVirtualPayConfirm() {
targets = append(targets, &orders[i])
}
if len(targets) == 0 {
slog.DebugContext(ctx, "virtual confirm: all candidates skipped by unpaid cache", "candidates", len(orders))
return
}
@@ -139,7 +150,7 @@ func runVirtualPayConfirm() {
defer func() { <-sem }()
defer func() {
if r := recover(); r != nil {
slog.Error("virtual confirm worker panic", "order_no", o.OrderNo, "recover", r)
slog.ErrorContext(ctx, "virtual confirm worker panic", "order_no", o.OrderNo, "recover", r)
}
}()
if s.confirmVirtualOrder(ctx, o) {
@@ -154,6 +165,9 @@ func runVirtualPayConfirm() {
if confirmed > 0 {
slog.InfoContext(ctx, "virtual pay confirm compensated orders",
"scanned", len(orders), "queried", len(targets), "confirmed", confirmed)
} else {
slog.DebugContext(ctx, "virtual pay confirm finished without compensation",
"scanned", len(orders), "queried", len(targets))
}
}
@@ -163,11 +177,16 @@ func runVirtualPayConfirm() {
func (s *OrderService) confirmVirtualOrder(ctx context.Context, order *mysqlmodel.Order) bool {
// query_order 需要 openid，订单表不冗余该字段，按 user_id 取。
user, err := s.userDao.UserGetByID(ctx, order.UserID)
if err != nil || user.WxOpenID == "" {
slog.WarnContext(ctx, "virtual confirm: user openid unavailable",
if err != nil {
slog.ErrorContext(ctx, "virtual confirm: get user failed",
"order_no", order.OrderNo, "user_id", order.UserID, "error", err)
return false
}
if user.WxOpenID == "" {
slog.WarnContext(ctx, "virtual confirm: user openid unavailable",
"order_no", order.OrderNo, "user_id", order.UserID)
return false
}
info, qerr := wechat.QueryOrder(ctx, user.WxOpenID, order.OrderNo, "")
if qerr != nil {