package service import ( "context" "log/slog" "sync" "time" "git.lzcwy.cn/zhengchiliu/zt-closet-backend/global" "git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/mysqldao" "git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/mysqlmodel" "git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/redisdao" "git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/redismodel" "git.lzcwy.cn/zhengchiliu/zt-closet-backend/pkg/wechat" ) const ( // virtualConfirmLookbackSeconds 查单补偿的回看窗口:只处理近 2 小时内的待支付/已过期虚拟订单。 // 该窗口覆盖推送重试耗尽的时间(官方最多重试 15 次、间隔 2/4/8/16… 秒),又避免全表扫描。 virtualConfirmLookbackSeconds int64 = 2 * 60 * 60 // virtualConfirmBatchSize 单轮最多取回的候选订单数。 virtualConfirmBatchSize = 200 // virtualConfirmWorkers 单轮并发查询的 worker 数。 // 必须并发:串行查询在积压时会撞上 ctx 超时,导致尾部订单被静默跳过(看似处理了全部,实则只查了前一部分)。 virtualConfirmWorkers = 5 // virtualConfirmUnpaidTTL 「已确认未支付」负缓存的存活时间。 // 命中即跳过本轮查询,使积压订单让出扫描名额——这是修复「积压订单饿死已支付订单」的关键; // 同时把单个订单的查询频率从每分钟一次降到每 TTL 一次,控制微信接口调用量。 // 代价:用户支付后最坏要等一个 TTL 才被补偿发现,需要更快收敛就把该值调小。 virtualConfirmUnpaidTTL = 10 * time.Minute // virtualConfirmLockTTL 单轮任务的分布式锁 TTL,必须大于单轮最大耗时,否则锁提前过期会让多实例并发执行。 virtualConfirmLockTTL = 5 * time.Minute // virtualConfirmTimeout 单轮任务的上下文超时,需给「批量查询 + 并发反查」留足余量。 virtualConfirmTimeout = 10 * time.Minute ) // StartVirtualPayConfirmScheduler 启动虚拟支付查单补偿定时任务(官方「道具直购流程」的发货轮询分支)。 // // 官方要求「发货推送」与「发货轮询」至少实现一个、两者结合更可靠:推送可能丢失 // (回调地址未生效、网络抖动、重试 15 次耗尽、服务重启窗口),且 wx.requestVirtualPayment // 的 success 回调也会丢失。本任务按窗口轮询 query_order,把「微信侧已支付但本地仍为 // 待支付/已过期」的订单条件流转为已支付并补发货,再回写 notify_provide_goods, // 保证用户付款后权益一定到账、微信侧订单也能正常推进到已发货。 func StartVirtualPayConfirmScheduler() { go func() { slog.Info("virtual pay confirm scheduler started") for { func() { defer func() { if r := recover(); r != nil { slog.Error("virtual pay confirm scheduler panic", "recover", r) } }() runVirtualPayConfirm() }() time.Sleep(time.Minute) } }() } // runVirtualPayConfirm 扫描待确认的虚拟支付订单并并发向微信反查,单实例通过 Redis 分布式锁防重。 // // 扫描策略(避免积压饿死): // 1. 按 id 升序取窗口内候选订单,用「已确认未支付」负缓存批量过滤掉最近查过的订单, // 被过滤的订单不再占名额,后续新订单自然会被覆盖到; // 2. 剩余候选用固定 worker 并发反查,避免串行耗时撞上 ctx 超时; // 3. 候选数打满上限时打 warning,提示可能存在积压。 func runVirtualPayConfirm() { // 虚拟支付未开通时直接返回,避免无谓的数据库扫描。 if !wechat.IsVirtualPayEnabled() { return } ctx, cancel := context.WithTimeout(context.Background(), virtualConfirmTimeout) defer cancel() ctx, span := global.Tracer.Start(ctx, "scheduler.VirtualPayConfirm") defer span.End() lockKey := redismodel.KeyLockVirtualConfirm lockVal := genUUID() ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, virtualConfirmLockTTL) if lerr != nil { slog.ErrorContext(ctx, "virtual confirm: acquire lock failed", "error", lerr) return } 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(). OrderListVirtualUnconfirmed(ctx, now-virtualConfirmLookbackSeconds, now, virtualConfirmBatchSize) if err != nil { slog.ErrorContext(ctx, "list unconfirmed virtual orders failed", "error", err) return } if len(orders) == 0 { slog.DebugContext(ctx, "virtual confirm: no unconfirmed orders in window") return } if len(orders) >= virtualConfirmBatchSize { slog.WarnContext(ctx, "virtual confirm backlog detected, candidates reached limit", "candidates", len(orders), "limit", virtualConfirmBatchSize, "lookback_seconds", virtualConfirmLookbackSeconds) } // 批量过滤负缓存:一次 MGET,避免逐单往返 Redis。 orderNos := make([]string, 0, len(orders)) for i := range orders { orderNos = append(orderNos, orders[i].OrderNo) } unpaid, merr := redisdao.VirtualPayUnpaidBatchGet(ctx, orderNos) if merr != nil { // 负缓存不可用不阻断补偿:本轮退化为「逐单都查」(最坏情况等价于旧行为)。 slog.WarnContext(ctx, "get virtual pay unpaid cache failed, fallback to query all", "error", merr) unpaid = map[string]bool{} } targets := make([]*mysqlmodel.Order, 0, len(orders)) for i := range orders { if unpaid[orders[i].OrderNo] { continue // 最近已确认未支付,让出名额给其它订单。 } targets = append(targets, &orders[i]) } if len(targets) == 0 { slog.DebugContext(ctx, "virtual confirm: all candidates skipped by unpaid cache", "candidates", len(orders)) return } s := NewOrderService() var ( mu sync.Mutex wg sync.WaitGroup confirmed int ) sem := make(chan struct{}, virtualConfirmWorkers) for _, order := range targets { wg.Add(1) sem <- struct{}{} go func(o *mysqlmodel.Order) { defer wg.Done() defer func() { <-sem }() defer func() { if r := recover(); r != nil { slog.ErrorContext(ctx, "virtual confirm worker panic", "order_no", o.OrderNo, "recover", r) } }() if s.confirmVirtualOrder(ctx, o) { mu.Lock() confirmed++ mu.Unlock() } }(order) } wg.Wait() 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)) } } // confirmVirtualOrder 向微信反查单个虚拟支付订单,确认已支付则条件流转并补发货,返回是否完成补发。 // 对应官方「道具直购流程」的 9.1(轮询 query_order)→ 9.2(发货)→ 9.3(notify_provide_goods 回写)。 // 与推送路径共用同一套幂等保护(条件流转 + grantOrderBenefits 内部幂等),重复执行不会重复发放。 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 { 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 { // 查询失败不等于「未支付」,不写负缓存,下一轮继续重试。 slog.ErrorContext(ctx, "virtual confirm: query order failed", "order_no", order.OrderNo, "error", qerr) return false } if !wechat.IsOrderPaidStatus(info.Order.Status) { // 未支付:用户放弃或仍在支付中,保持原状态等待过期。 // 写入负缓存让该订单在 TTL 内让出扫描名额,避免大量未支付订单把已支付订单挤出扫描窗口。 if serr := redisdao.VirtualPayUnpaidSet(ctx, order.OrderNo, virtualConfirmUnpaidTTL); serr != nil { slog.WarnContext(ctx, "virtual confirm: set unpaid cache failed", "order_no", order.OrderNo, "error", serr) } return false } // 金额必须与下单快照一致,防止订单号错配导致错发。 if info.Order.PaidFee != order.PayAmount { slog.ErrorContext(ctx, "virtual confirm: amount mismatch", "order_no", order.OrderNo, "query_amount", info.Order.PaidFee, "order_amount", order.PayAmount) return false } // 待支付 / 已过期 -> 已支付(条件流转天然幂等,多实例并发无害)。 for _, from := range []int8{OrderStatusPending, OrderStatusExpired} { ok, uerr := s.orderDao.OrderUpdateStatusByOrderNo(ctx, order.OrderNo, from, OrderStatusPaid) if uerr != nil { slog.ErrorContext(ctx, "virtual confirm: update status failed", "order_no", order.OrderNo, "from_status", from, "error", uerr) return false } if ok { slog.InfoContext(ctx, "virtual confirm: order compensated", "order_no", order.OrderNo, "from_status", from) s.finalizeVirtualPaidOrder(ctx, order, info.Order.PaidTime, info.Order.WxpayOrderID, "") // 9.3 通知微信发货完成:查单分支没有推送应答的机会,必须主动回写, // 否则微信侧订单会长期停留在「已支付待发货」,影响结算推进。 // 权益此时已发放,回写失败只影响微信侧状态,记 error 告警,不回滚发货。 if nerr := wechat.NotifyProvideGoods(ctx, order.OrderNo); nerr != nil { slog.ErrorContext(ctx, "virtual confirm: notify provide goods failed", "order_no", order.OrderNo, "error", nerr) } // 订单已补发,清掉可能残留的负缓存。 if derr := redisdao.VirtualPayUnpaidDel(ctx, order.OrderNo); derr != nil { slog.WarnContext(ctx, "virtual confirm: clear unpaid cache failed", "order_no", order.OrderNo, "error", derr) } return true } } return false }