package service import ( "context" "errors" "fmt" "log" "time" "affiliate_dash/internal/model" "affiliate_dash/internal/pkg/timeutil" "gorm.io/gorm" "gorm.io/gorm/clause" ) func (s *FulfillmentService) MarkProcessingTimeouts(timeout time.Duration, limit int) (int, error) { if timeout <= 0 { return 0, nil } if limit <= 0 || limit > 100 { limit = 50 } now := time.Now() cutoff := now.Add(-timeout) var ids []uint if err := s.db.Model(&model.FulfillmentOrder{}). Where("order_status = ? AND updated_at < ?", model.OrderStatusDelivering, cutoff). Order("updated_at ASC, id ASC"). Limit(limit). Pluck("id", &ids).Error; err != nil { return 0, err } changed := 0 for _, id := range ids { updated, err := s.markProcessingTimeout(id, timeout, now) if err != nil { return changed, err } if updated { changed++ } } return changed, nil } func (s *FulfillmentService) markProcessingTimeout(id uint, timeout time.Duration, now time.Time) (bool, error) { returned := false err := s.db.Transaction(func(tx *gorm.DB) error { var order model.FulfillmentOrder if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&order, id).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil } return err } if !processingTimedOut(&order, timeout, now) { return nil } if err := validateOrderStatusTransition(&order, model.OrderStatusShipFailed, fulfillmentTransitionTimeout); err != nil { return nil } submittedUpstream := deliverySubmittedUpstream(&order) var reason string if submittedUpstream { reason = fmt.Sprintf("发货超时:订单已提交上游但超过 %d 分钟未回传结果,可能仍在处理;请勿直接重试,先在上游确认订单状态", int(timeout.Minutes())) } else { reason = fmt.Sprintf("发货超时:发货提交中断,请重新提交(已停留 delivering 超过 %d 分钟)", int(timeout.Minutes())) } updates := map[string]interface{}{"order_status": model.OrderStatusShipFailed} updates["failure_reason"] = reason updates["result_data"] = buildProcessingTimeoutResultData(order.ResultData, timeout, now, reason) if err := tx.Model(&order).Updates(updates).Error; err != nil { return err } var out model.FulfillmentOrder if err := tx.First(&out, order.ID).Error; err != nil { return err } if err := writeAudit(tx, &order.MerchantID, nil, nil, "fulfillment.timeout", "fulfillment_order", order.OrderNo, map[string]interface{}{ "from": normalizeOrderStatus(&order), "to": model.OrderStatusShipFailed, "timeout_minutes": int(timeout.Minutes()), "reason": reason, }); err != nil { return err } if s.callbacks != nil { if err := s.callbacks.Enqueue(tx, order.MerchantID, "order.shipping.updated", orderCallbackData(&out)); err != nil { return err } } returned = true return nil }) return returned, err } func (s *FulfillmentService) RunProcessingTimeoutMonitor(ctx context.Context, timeout, interval time.Duration) { if timeout <= 0 { return } if interval <= 0 { interval = time.Minute } ticker := time.NewTicker(interval) defer ticker.Stop() for { changed, err := s.MarkProcessingTimeouts(timeout, 50) if err != nil { log.Printf("[fulfillment] timeout scan error: %v", err) } else if changed > 0 { log.Printf("[fulfillment] timeout scan marked failed count=%d", changed) } select { case <-ctx.Done(): return case <-ticker.C: } } } func buildProcessingTimeoutResultData(existing string, timeout time.Duration, now time.Time, reason string) string { return mergeResultData(existing, map[string]interface{}{ "timeout": true, "timeout_minutes": int(timeout.Minutes()), "timeout_at": timeutil.FormatAPITime(now), "ship_status": "failed", "fail_reason": reason, }) }