affiliate_dash/backend/internal/service/fulfillment_timeout.go

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 {
		if s.upstreamSync != nil {
			var ord model.FulfillmentOrder
			if err := s.db.First(&ord, id).Error; err == nil {
				if synced, _ := s.upstreamSync(context.Background(), &ord); synced {
					continue
				}
			}
		}
		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,
	})
}