yml2213andClaude Opus 4.8
8192344027
- 号主交接超时(owner_timeout)后允许补提交交接说明，避免临时延误导致订单卡死 - 后台新增"重置交接"动作，可将超时订单恢复为待号主交接并刷新计时 - 交接超时改以进入待交接时刻为基准计算，新增 handoff_started_at 字段 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
532 lines
15 KiB
Go
532 lines
15 KiB
Go
package ordertimeout
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strconv"
|
|
"time"
|
|
|
|
"hfb_sys/backend/internal/model"
|
|
"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"
|
|
|
|
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
|
|
}
|
|
|
|
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,
|
|
}
|
|
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,
|
|
}
|
|
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",
|
|
}).
|
|
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
|
|
}
|
|
}
|
|
return cfg, nil
|
|
}
|
|
|
|
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) 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
|
|
}
|