package ordertimeout import ( "context" "crypto/rand" "encoding/hex" "encoding/json" "fmt" "strconv" "time" "hfb_sys/backend/internal/model" "hfb_sys/backend/internal/modules/chat" "hfb_sys/backend/internal/modules/notification" "github.com/redis/go-redis/v9" "go.uber.org/zap" "gorm.io/datatypes" "gorm.io/gorm" "gorm.io/gorm/clause" ) const orderTimeoutLockKey = "hfb:job:ordertimeout:lock" const ( defaultRenterRetentionDaysAfterOrderEnd = 5 minRenterRetentionDaysAfterOrderEnd = 3 maxRenterRetentionDaysAfterOrderEnd = 7 ) type Job struct { db *gorm.DB redis *redis.Client logger *zap.Logger interval time.Duration instanceID string } type thresholds struct { PendingPaymentTimeoutMinutes int OwnerSubmitTimeoutMinutes int RenterConfirmTimeoutMinutes int ReturnOverdueGraceMinutes int OwnerReturnConfirmTimeoutMinutes int RenterRetentionDaysAfterOrderEnd int } func New(db *gorm.DB, redisClient *redis.Client, logger *zap.Logger) *Job { return &Job{ db: db, redis: redisClient, logger: logger, interval: time.Minute, instanceID: newInstanceID(), } } // acquireLock 通过 Redis 分布式锁确保同一时刻只有一个实例执行超时扫描。 // 未配置 Redis 时直接执行;Redis 出错时降级执行(事务内行锁与状态二次校验可兜底,不会写坏数据)。 func (j *Job) acquireLock(ctx context.Context) (func(), bool) { if j.redis == nil { return func() {}, true } ok, err := j.redis.SetNX(ctx, orderTimeoutLockKey, j.instanceID, j.lockTTL()).Result() if err != nil { j.logger.Warn("order timeout job acquire lock failed, run without lock", zap.Error(err)) return func() {}, true } if !ok { return nil, false } return j.releaseLock, true } // releaseLock 仅在锁仍归本实例时释放,避免误删其他实例已续上的锁。 func (j *Job) releaseLock() { if j.redis == nil { return } relCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() script := redis.NewScript(`if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`) if err := script.Run(relCtx, j.redis, []string{orderTimeoutLockKey}, j.instanceID).Err(); err != nil { j.logger.Warn("order timeout job release lock failed", zap.Error(err)) } } func (j *Job) lockTTL() time.Duration { return 2 * j.interval } func newInstanceID() string { b := make([]byte, 8) if _, err := rand.Read(b); err != nil { return fmt.Sprintf("inst-%d", time.Now().UnixNano()) } return hex.EncodeToString(b) } func (j *Job) Start(ctx context.Context) { if j == nil || j.db == nil { return } go j.loop(ctx) } func (j *Job) loop(ctx context.Context) { j.run(ctx) ticker := time.NewTicker(j.interval) defer ticker.Stop() for { select { case <-ctx.Done(): j.logger.Info("order timeout job stopped") return case <-ticker.C: j.run(ctx) } } } func (j *Job) run(ctx context.Context) { release, ok := j.acquireLock(ctx) if !ok { return } defer release() cfg, err := j.loadThresholds(ctx) if err != nil { j.logger.Warn("order timeout job config load failed", zap.Error(err)) return } now := time.Now() handlers := []func(context.Context, time.Time, thresholds) (int, error){ j.handlePendingPaymentTimeout, j.handleOwnerSubmitTimeout, j.handleRenterConfirmTimeout, j.handleReturnOverdue, j.handleOwnerReturnConfirmTimeout, j.handleEndedOrderRenterRetention, } total := 0 for _, handler := range handlers { count, err := handler(ctx, now, cfg) if err != nil { j.logger.Warn("order timeout handler failed", zap.Error(err)) continue } total += count } if total > 0 { j.logger.Info("order timeout job processed orders", zap.Int("count", total)) } } func (j *Job) loadThresholds(ctx context.Context) (thresholds, error) { cfg := thresholds{ PendingPaymentTimeoutMinutes: 15, OwnerSubmitTimeoutMinutes: 30, RenterConfirmTimeoutMinutes: 30, ReturnOverdueGraceMinutes: 10, OwnerReturnConfirmTimeoutMinutes: 120, RenterRetentionDaysAfterOrderEnd: defaultRenterRetentionDaysAfterOrderEnd, } var rows []model.SystemConfig err := j.db.WithContext(ctx). Where("`key` IN ?", []string{ "handoff.owner_submit_timeout_minutes", "handoff.renter_confirm_timeout_minutes", "order.pending_payment_timeout_minutes", "order.return_overdue_grace_minutes", "handoff.owner_return_confirm_timeout_minutes", "chat.renter_retention_days_after_order_end", }). Find(&rows).Error if err != nil { return cfg, err } for _, row := range rows { value, err := strconv.Atoi(row.Value) if err != nil || value < 0 { continue } switch row.Key { case "handoff.owner_submit_timeout_minutes": cfg.OwnerSubmitTimeoutMinutes = value case "handoff.renter_confirm_timeout_minutes": cfg.RenterConfirmTimeoutMinutes = value case "order.pending_payment_timeout_minutes": cfg.PendingPaymentTimeoutMinutes = value case "order.return_overdue_grace_minutes": cfg.ReturnOverdueGraceMinutes = value case "handoff.owner_return_confirm_timeout_minutes": cfg.OwnerReturnConfirmTimeoutMinutes = value case "chat.renter_retention_days_after_order_end": cfg.RenterRetentionDaysAfterOrderEnd = clampRenterRetentionDays(value) } } return cfg, nil } func clampRenterRetentionDays(value int) int { if value < minRenterRetentionDaysAfterOrderEnd { return minRenterRetentionDaysAfterOrderEnd } if value > maxRenterRetentionDaysAfterOrderEnd { return maxRenterRetentionDaysAfterOrderEnd } return value } func (j *Job) handlePendingPaymentTimeout(ctx context.Context, now time.Time, cfg thresholds) (int, error) { if cfg.PendingPaymentTimeoutMinutes <= 0 { return 0, nil } var rows []model.RentalOrder err := j.db.WithContext(ctx). Where("status = ? AND created_at <= ?", "pending_payment", now.Add(-time.Duration(cfg.PendingPaymentTimeoutMinutes)*time.Minute)). Order("id ASC"). Limit(100). Find(&rows).Error if err != nil { return 0, err } count := 0 for _, row := range rows { if err := j.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { var order model.RentalOrder if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&order, row.ID).Error; err != nil { return err } if order.Status != "pending_payment" { return nil } before := snapshot(&order) var listing model.RentalListing if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&listing, order.ListingID).Error; err != nil { return err } order.Status = "cancelled" order.HandoffStatus = "cancelled" listing.InTransaction = false orderID := order.ID if err := closePendingOrderPayments(tx, order.ID, "order_timeout"); err != nil { return err } if err := notification.Append(tx, notification.Entry{ UserID: order.RenterID, Type: "timeout", Title: "订单支付超时", Content: "订单未在规定时间内完成支付,已自动取消。", BizType: "order", BizID: &orderID, }); err != nil { return err } if err := tx.Save(&order).Error; err != nil { return err } if err := tx.Save(&listing).Error; err != nil { return err } return appendAuditLog(tx, "order.timeout.pending_payment", order.ID, map[string]any{ "order_id": order.ID, "order_no": order.OrderNo, "before": before, "after": snapshot(&order), }) }); err != nil { return count, err } count++ } return count, nil } // closePendingOrderPayments 在系统自动取消订单时同步关闭未完成支付单,保持订单和支付流水状态一致。 func closePendingOrderPayments(tx *gorm.DB, orderID uint64, source string) error { raw, err := json.Marshal(map[string]string{ "source": source, "reason": "订单已取消,关闭未完成支付单", }) if err != nil { return err } return tx.Model(&model.PaymentOrder{}). Where("order_id = ? AND biz_type = ? AND status IN ?", orderID, "order_pay", []string{"created", "paying"}). Updates(map[string]any{ "status": "closed", "raw_response": datatypes.JSON(raw), }).Error } func (j *Job) handleOwnerSubmitTimeout(ctx context.Context, now time.Time, cfg thresholds) (int, error) { if cfg.OwnerSubmitTimeoutMinutes <= 0 { return 0, nil } var rows []model.RentalOrder err := j.db.WithContext(ctx). Where("status = ? AND handoff_status = ? AND COALESCE(handoff_started_at, created_at) <= ?", "pending_handoff", "pending_owner", now.Add(-time.Duration(cfg.OwnerSubmitTimeoutMinutes)*time.Minute)). Order("id ASC"). Limit(100). Find(&rows).Error if err != nil { return 0, err } count := 0 for _, row := range rows { if err := j.updateOrder(ctx, row.ID, "order.timeout.owner_submit", func(tx *gorm.DB, order *model.RentalOrder) (string, error) { if order.Status != "pending_handoff" || order.HandoffStatus != "pending_owner" { return "", nil } before := snapshot(order) order.HandoffStatus = "owner_timeout" orderID := order.ID if err := notification.Append(tx, notification.Entry{ UserID: order.RenterID, Type: "timeout", Title: "号主交接超时", Content: "号主未在规定时间内提交交接说明,你可以取消订单或发起申诉。", BizType: "order", BizID: &orderID, }, notification.Entry{ UserID: order.OwnerID, Type: "timeout", Title: "订单交接已超时", Content: "你未在规定时间内提交交接说明,租客可取消订单或发起申诉。", BizType: "order", BizID: &orderID, }, ); err != nil { return "", err } return before, nil }); err != nil { return count, err } count++ } return count, nil } func (j *Job) handleRenterConfirmTimeout(ctx context.Context, now time.Time, cfg thresholds) (int, error) { if cfg.RenterConfirmTimeoutMinutes <= 0 { return 0, nil } var rows []model.RentalOrder err := j.db.WithContext(ctx). Where("status = ? AND handoff_status = ?", "pending_handoff", "pending_renter_confirm"). Order("id ASC"). Limit(100). Find(&rows).Error if err != nil { return 0, err } count := 0 deadline := now.Add(-time.Duration(cfg.RenterConfirmTimeoutMinutes) * time.Minute) for _, row := range rows { var handoff model.HandoffRecord err := j.db.WithContext(ctx). Where("order_id = ? AND type = ?", row.ID, "owner_handoff"). Order("id DESC"). First(&handoff).Error if err != nil || handoff.CreatedAt.After(deadline) { continue } if err := j.updateOrder(ctx, row.ID, "order.timeout.renter_confirm", func(tx *gorm.DB, order *model.RentalOrder) (string, error) { if order.Status != "pending_handoff" || order.HandoffStatus != "pending_renter_confirm" { return "", nil } before := snapshot(order) order.Status = "abnormal" order.HandoffStatus = "renter_confirm_timeout" orderID := order.ID if err := notification.Append(tx, notification.Entry{ UserID: order.RenterID, Type: "timeout", Title: "确认收号超时", Content: "你未在规定时间内确认收号,订单已进入客服介入状态。", BizType: "order", BizID: &orderID, }, notification.Entry{ UserID: order.OwnerID, Type: "timeout", Title: "租客确认收号超时", Content: "租客未在规定时间内确认收号,订单已进入客服介入状态。", BizType: "order", BizID: &orderID, }, ); err != nil { return "", err } return before, nil }); err != nil { return count, err } count++ } return count, nil } func (j *Job) handleReturnOverdue(ctx context.Context, now time.Time, cfg thresholds) (int, error) { var rows []model.RentalOrder overdueBefore := now.Add(-time.Duration(cfg.ReturnOverdueGraceMinutes) * time.Minute) err := j.db.WithContext(ctx). Where(`status = ? AND rented_at IS NOT NULL AND TIMESTAMPADD(HOUR, COALESCE(NULLIF(estimated_duration_hours, 0), 24), rented_at) <= ?`, "renting", overdueBefore). Order("id ASC"). Limit(100). Find(&rows).Error if err != nil { return 0, err } count := 0 for _, row := range rows { if err := j.updateOrder(ctx, row.ID, "order.timeout.return_overdue", func(tx *gorm.DB, order *model.RentalOrder) (string, error) { if order.Status != "renting" { return "", nil } before := snapshot(order) order.Status = "overdue" order.HandoffStatus = "return_overdue" orderID := order.ID if err := notification.Append(tx, notification.Entry{ UserID: order.RenterID, Type: "timeout", Title: "订单已逾期未结账", Content: "订单已超过预计截止时间,请尽快发起结账,避免进入申诉处理。", BizType: "order", BizID: &orderID, }, notification.Entry{ UserID: order.OwnerID, Type: "timeout", Title: "租客逾期未结账", Content: "租客未在预计截止后及时发起结账,你可以发起申诉或等待客服处理。", BizType: "order", BizID: &orderID, }, ); err != nil { return "", err } return before, nil }); err != nil { return count, err } count++ } return count, nil } func (j *Job) handleOwnerReturnConfirmTimeout(ctx context.Context, now time.Time, cfg thresholds) (int, error) { if cfg.OwnerReturnConfirmTimeoutMinutes <= 0 { return 0, nil } var rows []model.RentalOrder err := j.db.WithContext(ctx). Where("status = ? AND handoff_status = ? AND updated_at <= ?", "pending_checkout_confirm", "pending_owner_checkout", now.Add(-time.Duration(cfg.OwnerReturnConfirmTimeoutMinutes)*time.Minute)). Order("id ASC"). Limit(100). Find(&rows).Error if err != nil { return 0, err } count := 0 for _, row := range rows { if err := j.updateOrder(ctx, row.ID, "order.timeout.owner_checkout_confirm", func(tx *gorm.DB, order *model.RentalOrder) (string, error) { if order.Status != "pending_checkout_confirm" || order.HandoffStatus != "pending_owner_checkout" { return "", nil } before := snapshot(order) order.Status = "abnormal" order.HandoffStatus = "owner_checkout_confirm_timeout" orderID := order.ID if err := notification.Append(tx, notification.Entry{ UserID: order.RenterID, Type: "timeout", Title: "号主确认结账超时", Content: "号主未在规定时间内确认结账,订单已进入客服复核状态。", BizType: "order", BizID: &orderID, }, notification.Entry{ UserID: order.OwnerID, Type: "timeout", Title: "确认结账已超时", Content: "你未在规定时间内确认结账,订单已进入客服复核状态。", BizType: "order", BizID: &orderID, }, ); err != nil { return "", err } return before, nil }); err != nil { return count, err } count++ } return count, nil } func (j *Job) handleEndedOrderRenterRetention(ctx context.Context, now time.Time, cfg thresholds) (int, error) { retentionDays := clampRenterRetentionDays(cfg.RenterRetentionDaysAfterOrderEnd) deadline := now.AddDate(0, 0, -retentionDays) var rows []model.RentalOrder err := j.db.WithContext(ctx). Table("rental_orders AS o"). Select("o.*"). Joins("JOIN chat_conversations AS c ON c.listing_id = o.listing_id AND c.type = ?", chat.ConversationTypeListingGroup). Joins("JOIN chat_participants AS cp ON cp.conversation_id = c.id AND cp.participant_type = ? AND cp.participant_id = o.renter_id AND cp.role = ?", "user", "renter"). Where("o.status IN ?", []string{"completed", "cancelled", "closed"}). Where("o.settled_at IS NOT NULL AND o.settled_at <= ?", deadline). Order("o.id ASC"). Limit(100). Find(&rows).Error if err != nil { return 0, err } count := 0 for _, row := range rows { removed := false if err := j.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { var order model.RentalOrder if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&order, row.ID).Error; err != nil { return err } if !isEndedOrderStatus(order.Status) { return nil } if order.SettledAt == nil || order.SettledAt.After(deadline) { return nil } var err error removed, err = chat.RemoveRenterFromListingConversation(tx, order.ListingID, order.RenterID) if err != nil { return err } if !removed { return nil } return appendAuditLog(tx, "chat.renter_retention.remove", order.ID, map[string]any{ "order_id": order.ID, "order_no": order.OrderNo, "listing_id": order.ListingID, "renter_id": order.RenterID, "retention_days": retentionDays, }) }); err != nil { return count, err } if removed { count++ } } return count, nil } func isEndedOrderStatus(status string) bool { switch status { case "completed", "cancelled", "closed": return true default: return false } } func (j *Job) updateOrder(ctx context.Context, orderID uint64, action string, fn func(*gorm.DB, *model.RentalOrder) (string, error)) error { return j.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { var order model.RentalOrder if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&order, orderID).Error; err != nil { return err } before, err := fn(tx, &order) if err != nil || before == "" { return err } if err := tx.Save(&order).Error; err != nil { return err } return appendAuditLog(tx, action, order.ID, map[string]any{ "order_id": order.ID, "order_no": order.OrderNo, "before": before, "after": snapshot(&order), }) }) } func snapshot(order *model.RentalOrder) string { raw, _ := json.Marshal(map[string]any{ "status": order.Status, "handoff_status": order.HandoffStatus, "settlement_status": order.SettlementStatus, }) return string(raw) } func appendAuditLog(tx *gorm.DB, action string, bizID uint64, detail map[string]any) error { raw, err := json.Marshal(detail) if err != nil { return err } row := model.AuditLog{ ActorType: "system", ActorID: 0, Action: action, BizType: "order", BizID: &bizID, Detail: datatypes.JSON(raw), } return tx.Create(&row).Error }