All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 6m58s
- 新增 expires_at 字段和复合索引,待支付订单 30 分钟超时自动取消 - 实现 cancelOrder/unfreezeWalletForCancel 钱包余额解冻逻辑 - 创建 Asynq 定时任务(order_expire/alert_check/data_cleanup) - 将原有 time.Ticker 轮询迁移至 Asynq Scheduler 统一调度 - 同步 delta specs 到 main specs 并归档变更
6.1 KiB
6.1 KiB
订单超时自动取消功能
功能概述
为待支付订单(微信/支付宝)添加 30 分钟超时自动取消机制。超时后自动取消订单并解冻钱包余额(如有冻结)。
核心设计
超时流程
用户下单(微信/支付宝)
├── 设置 expires_at = 当前时间 + 30 分钟
├── 订单状态: payment_status = 1(待支付)
│
├── 场景 1: 用户在 30 分钟内支付
│ ├── 支付成功 → 清除 expires_at(设为 NULL)
│ └── 订单正常完成
│
└── 场景 2: 超过 30 分钟未支付
├── Asynq Scheduler 每分钟触发扫描
├── 查询 expires_at <= NOW() AND payment_status = 1
├── 取消订单 → payment_status = 5(已取消)
├── 清除 expires_at
└── 解冻钱包余额(如有)
不设置超时的场景
- 钱包支付:立即扣款,无需超时
- 线下支付:管理员手动确认,无需超时
- 混合支付:需要在线支付部分才设置超时
技术实现
数据库变更
-- 迁移文件: migrations/000069_add_order_expiration.up.sql
ALTER TABLE tb_order ADD COLUMN expires_at TIMESTAMPTZ;
-- 部分索引: 仅索引待支付订单,减少索引大小
CREATE INDEX idx_order_expires ON tb_order (expires_at, payment_status)
WHERE expires_at IS NOT NULL AND payment_status = 1;
涉及文件
| 层级 | 文件 | 变更说明 |
|---|---|---|
| 迁移 | migrations/000069_add_order_expiration.up.sql |
添加 expires_at 字段和索引 |
| 迁移 | migrations/000069_add_order_expiration.down.sql |
回滚脚本 |
| 常量 | pkg/constants/constants.go |
添加任务类型和超时参数 |
| 模型 | internal/model/order.go |
添加 ExpiresAt 字段 |
| DTO | internal/model/dto/order_dto.go |
添加 ExpiresAt、IsExpired 响应字段 |
| Store | internal/store/postgres/order_store.go |
添加 FindExpiredOrders、is_expired 过滤 |
| Service | internal/service/order/service.go |
创建订单设置超时、取消逻辑、批量取消 |
| 任务 | internal/task/order_expire.go |
订单超时任务处理器 |
| 任务 | internal/task/alert_check.go |
告警检查任务处理器(从 ticker 迁移) |
| 任务 | internal/task/data_cleanup.go |
数据清理任务处理器(从 ticker 迁移) |
| 队列 | pkg/queue/types.go |
添加 OrderExpirer 接口和 WorkerStores/Services 字段 |
| 队列 | pkg/queue/handler.go |
注册 3 个新任务处理器 |
| Bootstrap | internal/bootstrap/worker_stores.go |
添加 CardWallet Store |
| Bootstrap | internal/bootstrap/worker_services.go |
添加 OrderService 初始化 |
| Worker | cmd/worker/main.go |
替换 ticker 为 Asynq Scheduler |
常量定义
// pkg/constants/constants.go
TaskTypeOrderExpire = "order:expire" // 订单超时任务
TaskTypeAlertCheck = "alert:check" // 告警检查任务
TaskTypeDataCleanup = "data:cleanup" // 数据清理任务
OrderExpireTimeout = 30 * time.Minute // 订单超时时间
OrderExpireBatchSize = 100 // 每次批量取消数量
接口变更
订单列表查询新增过滤参数
GET /api/admin/orders?is_expired=true
GET /api/h5/orders?is_expired=true
is_expired=true: 仅返回已超时的订单is_expired=false: 仅返回未超时的订单
订单响应新增字段
{
"expires_at": "2025-02-28T12:30:00+08:00",
"is_expired": false
}
expires_at: 超时时间,null表示无超时(钱包/线下支付)is_expired: 是否已超时(计算字段)
定时任务调度器重构
变更前(time.Ticker)
// cmd/worker/main.go 中的 goroutine
alertChecker := startAlertChecker(ctx, ...) // time.Ticker 每分钟
cleanupChecker := startCleanupScheduler(ctx, ...) // time.Timer 每天凌晨 2 点
问题:
- 单点运行,无法分布式
- 无重试机制
- 无任务状态监控
变更后(Asynq Scheduler)
// Asynq Scheduler 统一管理
asynqScheduler.Register("@every 1m", asynq.NewTask("order:expire", nil))
asynqScheduler.Register("@every 1m", asynq.NewTask("alert:check", nil))
asynqScheduler.Register("0 2 * * *", asynq.NewTask("data:cleanup", nil))
优势:
- 通过 Redis 实现分布式调度
- 自动重试失败任务
- 可通过 Asynq Dashboard 监控
- 统一的任务处理模式
调度规则
| 任务 | 调度表达式 | 说明 |
|---|---|---|
| 订单超时取消 | @every 1m |
每分钟扫描一次 |
| 告警检查 | @every 1m |
每分钟检查一次 |
| 数据清理 | 0 2 * * * |
每天凌晨 2 点执行 |
钱包解冻逻辑
取消订单时的解冻流程
cancelOrder(ctx, order)
├── 幂等更新: WHERE payment_status = 1 → 5
├── 清除 expires_at
│
├── 如果是代理钱包支付 (payment_method = wallet, buyer_type = agent)
│ └── AgentWalletStore.UnfreezeBalanceWithTx(tx, shopID, amount)
│
└── 如果是卡钱包支付 (payment_method = wallet/mixed, buyer_type != agent)
└── 直接更新 frozen_balance -= amount (WHERE frozen_balance >= amount)
幂等性保障
- 使用
WHERE payment_status = 1条件更新,确保只取消待支付订单 RowsAffected == 0说明订单已被处理(已支付或已取消),直接跳过- 批量取消时,单个订单失败不影响其他订单
循环依赖解决方案
internal/service/order 导入 pkg/queue(使用 queue.Client),而 pkg/queue/types.go 需要引用 OrderService。
解决方案:在 pkg/queue/types.go 定义 OrderExpirer 接口,internal/task/order_expire.go 定义同名局部接口。Go 的结构化类型系统使 order.Service 自动满足两个接口,无需显式声明。
// pkg/queue/types.go
type OrderExpirer interface {
CancelExpiredOrders(ctx context.Context) (int, error)
}
// WorkerServices 中使用接口类型
OrderExpirer OrderExpirer
// internal/task/order_expire.go(局部接口,避免导入 pkg/queue)
type OrderExpirer interface {
CancelExpiredOrders(ctx context.Context) (int, error)
}