Files
pay/main.go
T
wangjia 669774d0f7 feat(dev): 本地联调 mock 渠道工具链(fake 注册/seed pangolin/模拟付款端点/可配投递间隔,MOCK_CHANNEL_ENABLED 门控)
让 pangolin(业务方)能在本地端到端测通:下单(method=fake)→显示假收款→手动
POST /api/v2/dev/orders/:order_no/mark-paid 模拟付款成功→pay 签名 webhook→pangolin
开通订阅。全部改动收在 env MOCK_CHANNEL_ENABLED 门控内,生产默认 false 零影响:

- providerbuild.BuildRegistry:门控下注册 fake provider,SettleCurrencies 覆盖成
  ["CNY","USDT"](CNY 排首位对齐 pangolin 实际下单币种),fake.go 加 SetSettleCurrencies
  测试缝,默认("USDT")不变、不破坏既有测试。
- main.go:门控下注入内存 fake 账户(picker 才选得到 method=fake)+ seedPangolin()
  幂等 upsert pro_month/pro_quarter/pro_year 三档商品 + 挂载 dev mark-paid 路由。
- config:新增 MOCK_CHANNEL_ENABLED / WEBHOOK_TICK_SECONDS(默认 60,本地可设 2 近实时
  投递)两个 env 开关;config.yaml 补 biz.pangolin 声明(secret 走 BIZ_PANGOLIN_SECRET env)。
- internal/handler/devmock.go:MarkPaid 按 out_trade_no 取 fake 渠道最新 attempt,
  原样组回调体走 gateway.HandleCallback("fake",...),与真实回调同一入口。

自测:go build/vet/test 全绿;起服务后签名下单(currency=CNY,与 pangolin 未传
currency 时的实际结算币种一致)→ mark-paid → 订单翻 paid → outbox ~2s 内尝试投递
(因本机未起 pangolin 收 404,预期内,证明 pay 侧链路已通)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013nMthbVEmQquxBRKb9Fj8u
2026-07-11 01:26:14 +08:00

385 lines
18 KiB
Go

package main
import (
"context"
"errors"
"log"
"strings"
"time"
"github.com/gin-gonic/gin"
"github.com/glebarez/sqlite"
"gorm.io/gorm"
"gorm.io/gorm/logger"
"github.com/wangjia/pay/config"
"github.com/wangjia/pay/internal/accounts"
"github.com/wangjia/pay/internal/channel"
"github.com/wangjia/pay/internal/gateway"
"github.com/wangjia/pay/internal/handler"
"github.com/wangjia/pay/internal/model"
"github.com/wangjia/pay/internal/providerbuild"
"github.com/wangjia/pay/internal/reconcile"
"github.com/wangjia/pay/internal/router"
"github.com/wangjia/pay/internal/store"
"github.com/wangjia/pay/internal/webhook"
)
func main() {
config.Load()
db := initDB()
autoMigrate(db)
seed(db)
seedWechat(db)
if config.C.MockChannelEnabled {
seedPangolin(db) // dev-only:仅 MOCK_CHANNEL_ENABLED=true 时 seed pangolin 三档商品
}
reg := channel.NewRegistry(db)
gin.SetMode(config.C.Server.Mode)
r := gin.New()
r.Use(gin.Logger(), gin.Recovery())
// 安全review发现:当前部署(阿里云单机,无 nginx/网关前置)不存在真实反向代理,
// 但 gin 默认 trustedProxies=0.0.0.0/0(信任任意来源的 X-Forwarded-For/X-Real-IP)。
// 不关掉的话,c.ClientIP() 会读客户端可控的请求头——攻击者每次伪造不同 XFF 即可
// 让限流中间件(internal/middleware/ratelimit.go)把每次请求都当"新 IP"的新桶,
// burst 永不触发,且伪造 IP 数量无上限还能撑爆 buckets map。
// SetTrustedProxies(nil) 让 engine.trustedCIDRs=nil → isTrustedProxy 恒 false →
// ClientIP() 退回 RemoteIP()(即 net/http 拿到的 socket 对端地址,请求头伪造不了)。
// 若未来加 nginx/API 网关反代,这里要改成该反代的真实出口 IP/CIDR 列表,
// 不能再传 nil(否则又会把加固之后合法的下游 IP 直接当客户端源用)。
if err := r.SetTrustedProxies(nil); err != nil {
log.Fatalf("SetTrustedProxies(nil) 失败: %v", err)
}
orderSvc := router.Setup(r, db, reg)
// v2 统一网关装配(P2 骨架,P3 配置驱动注册接线):provider 注册表 + gateway + webhook notifier。
orderStore := store.NewOrderStore(db)
refundStore := store.NewRefundStore(db)
subStore := store.NewSubscriptionStore(db)
chargebackStore := store.NewChargebackStore(db)
webhookStore := store.NewWebhookStore(db)
notifier := webhook.NewNotifier(webhookStore, config.C.BizByName, func(no string) (bool, error) {
o, err := orderStore.GetOrder(no) // 投递门禁:订单已结算(paid 及退款态)才放行
if err != nil {
return false, err
}
return o.Status.Settled(), nil
})
// WebhookTickSeconds 默认 60(config.Load 兜底);本地联调设 WEBHOOK_TICK_SECONDS=2
// 可让「模拟付款成功」后近实时投递,不必等最多 60s。
notifier.Start(time.Duration(config.C.WebhookTickSeconds) * time.Second)
productResolver := gateway.NewDBProductResolver(db) // 币种由下单渠道结算能力驱动(设计 §4.1)
if config.C.MockChannelEnabled {
// dev-only:fake 渠道要能被 acctPicker 选中才能下单成功(EnabledFor("fake",...)
// 否则候选为空 → ErrNoAccount)。注入一条内存 fake 账户,不落 config.yaml/DB。
// Region 留空 = 不按 region 过滤(accounts.Registry.EnabledFor 对空 Region 放行任意 region)。
config.C.Accounts = append(config.C.Accounts, config.AccountConfig{
AccountID: "fake-dev", Channel: "fake", Weight: 1, Enabled: true, Subject: "本地联调 mock 渠道",
})
}
acctReg := accounts.New(config.C.Accounts)
pReg := providerbuild.BuildRegistry(acctReg) // 配置驱动:有 enabled 账户才注册对应渠道(P3);fake 见 MOCK_CHANNEL_ENABLED 门控
// P5 多账户路由:按 config.routing.<channel> 选策略(缺省 round_robin)。
// limit_aware 用量数据源:P6 对账 Runner 周期 Refresh 的真实用量源(替 NopUsage)。
usage := reconcile.NewUsageSource(orderStore, time.Now)
acctPicker := accounts.NewRouter(acctReg, config.C.Routing, usage)
gw := gateway.New(orderStore, refundStore, pReg, acctPicker, productResolver, notifier, "cn", subStore, chargebackStore)
router.SetupV2(r, gw)
if config.C.MockChannelEnabled {
// dev-only:浏览器/curl 一个请求即可把某笔 fake 渠道订单标为已付,触发 Settle→webhook
// 入队,免去手工查库拼回调体。刻意不挂 stateChangeRateLimit(本机联调用,不设限)。
devH := handler.NewDevMockHandler(db, gw)
r.POST("/api/v2/dev/orders/:order_no/mark-paid", devH.MarkPaid)
log.Println("[dev] MOCK_CHANNEL_ENABLED=true:已挂载 POST /api/v2/dev/orders/:order_no/mark-paid")
}
// P6 后台守护 / 对账:订单过期清理 + 用量刷新 + 查单对账收敛 + 已付抽查 +
// 退款修复扫描/卡滞告警(P4 T3 review 追加义务,归到本任务一起装配)+
// crypto 预留冷启动 Warm + 孤儿扫描(Task 6,AddCryptoJobs)。
if config.C.Reconcile.Enabled {
rc := config.C.Reconcile
runner := reconcile.NewRunner()
// 纯本地(DB-only)任务用 AddLocal:会被启动预热 RunOnceLocal 同步跑一遍。
runner.AddLocal("order-expire", time.Duration(rc.ExpireEverySec)*time.Second,
reconcile.OrderExpirerTask(orderStore, time.Duration(rc.OrderTTLMin)*time.Minute, time.Now))
runner.AddLocal("usage-refresh", time.Duration(rc.UsageEverySec)*time.Second,
reconcile.RefreshUsageTask(usage))
runner.AddLocal("refund-apply-sweep", time.Duration(rc.RefundApplyEverySec)*time.Second,
reconcile.RefundApplyTask(orderStore, refundStore, time.Duration(rc.RefundApplyLookbackMin)*time.Minute, time.Now, 200))
runner.AddLocal("refund-stuck-alert", time.Duration(rc.RefundApplyEverySec)*time.Second,
reconcile.RefundStuckAlertTask(refundStore, time.Duration(rc.RefundStuckWarnMin)*time.Minute, time.Now))
// 网络型(出网 HTTP,10-15s 超时)任务用 Add:启动预热不跑,交各自 ticker 首跳。
runner.Add("sync-pending", time.Duration(rc.SyncEverySec)*time.Second,
reconcile.SyncPendingTask(gw, 100))
runner.Add("paid-spotcheck", time.Duration(rc.SpotCheckEverySec)*time.Second,
reconcile.PaidSpotCheckTask(orderStore, pReg, time.Duration(rc.SpotCheckWindowMin)*time.Minute, time.Now))
orphanStore := store.NewOrphanStore(db)
reconcile.AddCryptoJobs(runner, pReg, acctReg, orderStore, orphanStore, rc) // crypto 预留冷启动 Warm(直跑,不受影响)+ 孤儿扫描(网络型,交 ticker 首跳)
ctx := context.Background()
runner.RunOnceLocal(ctx) // 启动预热:只跑本地任务(usage 快照/过期清理/退款自愈立即生效);网络型任务(sync-pending/paid-spotcheck/crypto-orphan-scan)交各自 ticker 首跳,避免上游慢拖住 r.Run(addr) 前的启动
runner.Start(ctx)
log.Printf("[reconcile] 后台守护已启动(过期清理/用量刷新/查单对账/已付抽查/退款修复扫描/退款卡滞告警/crypto 孤儿扫描)")
}
if config.C.QuerySync.Enabled {
orderSvc.StartQuerySync(
time.Duration(config.C.QuerySync.IntervalSec)*time.Second,
time.Duration(config.C.QuerySync.MaxAgeMin)*time.Minute,
)
log.Printf("查单兜底已启动:每 %ds 一次", config.C.QuerySync.IntervalSec)
}
// 业务回调重试兜底(webhook 丢失/业务方短暂不可用时补推)。
orderSvc.StartBizNotifyRetry(60 * time.Second)
addr := ":" + config.C.Server.Port
log.Printf("支付服务启动 %s (mode=%s, base_url=%s)", addr, config.C.Server.Mode, config.C.Server.BaseURL)
if err := r.Run(addr); err != nil {
log.Fatalf("启动失败: %v", err)
}
}
func initDB() *gorm.DB {
logLevel := logger.Silent
if config.C.Server.Mode == "debug" {
logLevel = logger.Info
}
gormCfg := &gorm.Config{Logger: logger.Default.LogMode(logLevel), TranslateError: true}
switch config.C.Database.Driver {
case "sqlite", "":
db, err := gorm.Open(sqlite.Open(sqliteDSN(config.C.Database.DSN)), gormCfg)
if err != nil {
log.Fatalf("连接 sqlite 失败: %v", err)
}
return db
case "mysql":
log.Fatal("mysql 驱动尚未启用:go get gorm.io/driver/mysql 后在 initDB 接入")
return nil
default:
log.Fatalf("未知 database.driver: %s", config.C.Database.Driver)
return nil
}
}
// sqliteDSN 补 _txlock=immediate(BEGIN IMMEDIATE 立即抢库级写锁,串行化并发事务的
// 读阶段——RefundStore.CreateRefundGuarded 等守卫类事务依赖此语义)+
// busy_timeout(5000)(抢不到锁时等待重试而非立即报 SQLITE_BUSY)。呼应 pangolin
// 的 SQLite DSN 约定(server/internal/db/db.go)。config.yaml 里的 dsn 是纯文件路径
// (如 "pay.db"),这里补上查询串;若用户已经自带 "?",追加而不是覆盖。
func sqliteDSN(raw string) string {
extra := "_txlock=immediate&_pragma=busy_timeout(5000)"
if strings.Contains(raw, "?") {
return raw + "&" + extra
}
return raw + "?" + extra
}
func autoMigrate(db *gorm.DB) {
if err := db.AutoMigrate(
&model.Merchant{},
&model.Product{},
&model.ProductPrice{},
&model.Order{},
&model.NotifyLog{},
&model.BizNotifyLog{},
&model.OrderV2{}, &model.Attempt{}, &model.Account{}, &model.Refund{}, &model.WebhookDelivery{}, &model.OrphanPayment{}, // v2
&model.Subscription{}, // v2 recurring
&model.Chargeback{}, // v2 recurring — P8 Task6 拒付记录
); err != nil {
log.Fatalf("自动迁移失败: %v", err)
}
// AutoMigrate 只按名字判断索引是否存在,不会把老库同名的 2 列 uq_delivery
// 加宽到当前模型的 3 列(out_trade_no,event_type,refund_id);显式补一刀。
if err := model.UpgradeWebhookDeliveryIndex(db); err != nil {
log.Fatalf("升级 uq_delivery 索引失败: %v", err)
}
log.Println("AutoMigrate 完成")
}
// seed 据 config 的 alipay_sandbox upsert 一个支付宝商户,并为其补两个测试套餐。
func seed(db *gorm.DB) {
sb := config.C.AlipaySandbox
if !sb.Enabled {
log.Println("[seed] alipay_sandbox.enabled=false,跳过沙箱商户初始化")
return
}
var m model.Merchant
err := db.Where("code = ?", sb.MerchantCode).First(&m).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
m = model.Merchant{
Code: sb.MerchantCode,
Name: sb.MerchantName,
Channel: "alipay",
}
} else if err != nil {
log.Fatalf("[seed] 查询商户失败: %v", err)
}
m.Production = sb.Production
m.Enabled = true
m.AppID = sb.AppID
m.AppPrivateKey = sb.AppPrivateKey
m.AlipayPublicKey = sb.AlipayPublicKey
if err := db.Save(&m).Error; err != nil {
log.Fatalf("[seed] 保存商户失败: %v", err)
}
log.Printf("[seed] 支付宝沙箱商户就绪: code=%s app_id=%s", m.Code, m.AppID)
var cnt int64
db.Model(&model.Product{}).Where("merchant_id = ?", m.ID).Count(&cnt)
if cnt == 0 {
samples := []model.Product{
{MerchantID: m.ID, Name: "测试套餐 A", Description: "沙箱联调用", Price: "0.01", Active: true, Sort: 1},
{MerchantID: m.ID, Name: "测试套餐 B", Description: "沙箱联调用", Price: "0.02", Active: true, Sort: 2},
}
if err := db.Create(&samples).Error; err != nil {
log.Fatalf("[seed] 创建测试套餐失败: %v", err)
}
log.Printf("[seed] 已为商户 %s 创建 %d 个测试套餐", m.Code, len(samples))
}
seedPlans(db, m.ID)
}
// seedPlans 幂等 upsert 业务套餐(带 biz_code,供 jiu 等业务系统按 biz_code 查 product)。
// 价格权威在此表;权益(时长/档位/设备/功能)由业务方按 biz_code 映射。
func seedPlans(db *gorm.DB, merchantID uint64) {
// Name 会成为支付宝收银台的 subject(一行约 12-14 汉字可见):
// 带品牌前缀、短句式,避免截断(2026-07-04 用户反馈收银台信息太少/被截断)。
plans := []model.Product{
{Name: "岩美酒库·联调测试", Description: "1 分钱端到端联调用(标准权益)", Price: "0.01", BizCode: "test_liandiao", Sort: 0},
{Name: "岩美酒库·首月特惠30天", Description: "标准权益 · 30 天 · 每门店限一次(限次由 jiu 侧校验)", Price: "1.00", BizCode: "promo_first_month", Sort: 1},
{Name: "岩美酒库·标准版月付", Description: "单店/单仓库 · 2 客户端 · 千张图片分享", Price: "299.00", BizCode: "monthly_standard", Sort: 2},
{Name: "岩美酒库·高级版月付", Description: "单店/多仓库 · 5 客户端 · 万张照片 · 免费 AI 周/月度商业分析", Price: "599.00", BizCode: "monthly_pro", Sort: 3},
{Name: "岩美酒库·标准版年付", Description: "单店/单仓库 · 2 客户端 · 千张图片分享", Price: "2999.00", BizCode: "annual_standard", Sort: 4},
{Name: "岩美酒库·高级版年付", Description: "单店/多仓库 · 5 客户端 · 万张照片 · 免费 AI 周/月度商业分析", Price: "5999.00", BizCode: "annual_pro", Sort: 5},
}
for _, p := range plans {
var existing model.Product
err := db.Where("biz_code = ?", p.BizCode).First(&existing).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
p.MerchantID = merchantID
p.Active = true
if err := db.Create(&p).Error; err != nil {
log.Fatalf("[seed] 创建套餐 %s 失败: %v", p.BizCode, err)
}
continue
} else if err != nil {
log.Fatalf("[seed] 查询套餐 %s 失败: %v", p.BizCode, err)
}
// 已存在:更新价格/文案/归属(价格权威可从此改)。
db.Model(&existing).Updates(map[string]any{
"merchant_id": merchantID, "name": p.Name, "description": p.Description,
"price": p.Price, "active": true, "sort": p.Sort,
})
}
log.Printf("[seed] 业务套餐就绪:%d 个(含联调 0.01)", len(plans))
}
// seedWechat 据 config 的 wechat upsert 一个微信商户(V3 Native),并为其补两个测试套餐。
// 微信无沙箱:wechat.enabled=true 即视为正式(Production=true),只能用真实商户号做 1 分钱实测。
func seedWechat(db *gorm.DB) {
wc := config.C.Wechat
if !wc.Enabled {
log.Println("[seed] wechat.enabled=false,跳过微信商户初始化")
return
}
var m model.Merchant
err := db.Where("code = ?", wc.MerchantCode).First(&m).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
m = model.Merchant{
Code: wc.MerchantCode,
Name: wc.MerchantName,
Channel: "wechat",
}
} else if err != nil {
log.Fatalf("[seed] 查询微信商户失败: %v", err)
}
m.Production = true // 微信无沙箱,一律正式网关
m.Enabled = true
m.MchID = wc.MchID
m.WxAppID = wc.AppID
m.CertSerial = wc.CertSerial
m.WxPrivateKey = wc.PrivateKey
m.APIv3Key = wc.APIv3Key
if err := db.Save(&m).Error; err != nil {
log.Fatalf("[seed] 保存微信商户失败: %v", err)
}
log.Printf("[seed] 微信商户就绪: code=%s mch_id=%s", m.Code, m.MchID)
var cnt int64
db.Model(&model.Product{}).Where("merchant_id = ?", m.ID).Count(&cnt)
if cnt == 0 {
samples := []model.Product{
{MerchantID: m.ID, Name: "微信测试套餐 A", Description: "微信联调用", Price: "0.01", Active: true, Sort: 1},
{MerchantID: m.ID, Name: "微信测试套餐 B", Description: "微信联调用", Price: "0.02", Active: true, Sort: 2},
}
if err := db.Create(&samples).Error; err != nil {
log.Fatalf("[seed] 创建微信测试套餐失败: %v", err)
}
log.Printf("[seed] 已为微信商户 %s 创建 %d 个测试套餐", m.Code, len(samples))
}
}
// seedPangolin dev-only:仅 main() 在 MOCK_CHANNEL_ENABLED=true 时调用,幂等 upsert
// pangolin 业务方的三档订阅商品(biz_code=sku:pro_month/pro_quarter/pro_year)。
// 价格用小额占位;MerchantID 留 0——ProductResolver.Resolve 只按 biz_code 查,不看
// merchant 归属(与 seedPlans 场景不同,这里无需真实商户)。
// 币种与 providerbuild.BuildRegistry 给 fake 设的 SettleCurrencies=["CNY","USDT"]对齐:
// CNY 排首位,是 gateway.CreateOrder 实际会选用的结算币种(pangolin 下单不传
// currency),故用 v1 Product.Price(元 string)回退即可满足;顺带补一行 USDT
// ProductPrice,覆盖将来切换/联调其它币种的场景。
func seedPangolin(db *gorm.DB) {
type pangolinPlan struct {
BizCode string
Name string
CNYYuan string // Product.Price(CNY 回退价,元 string)
USDTMino int64 // ProductPrice(USDT, 6 位小数 minor units)占位
}
plans := []pangolinPlan{
{BizCode: "pro_month", Name: "Pangolin·Pro 月付(联调)", CNYYuan: "1.00", USDTMino: 1000000},
{BizCode: "pro_quarter", Name: "Pangolin·Pro 季付(联调)", CNYYuan: "1.00", USDTMino: 1000000},
{BizCode: "pro_year", Name: "Pangolin·Pro 年付(联调)", CNYYuan: "1.00", USDTMino: 1000000},
}
for i, p := range plans {
var existing model.Product
err := db.Where("biz_code = ?", p.BizCode).First(&existing).Error
switch {
case errors.Is(err, gorm.ErrRecordNotFound):
existing = model.Product{
Name: p.Name, Description: "pay 本地联调 mock 渠道占位商品(MOCK_CHANNEL_ENABLED)",
Price: p.CNYYuan, BizCode: p.BizCode, Active: true, Sort: i,
}
if err := db.Create(&existing).Error; err != nil {
log.Fatalf("[seed-pangolin] 创建套餐 %s 失败: %v", p.BizCode, err)
}
case err != nil:
log.Fatalf("[seed-pangolin] 查询套餐 %s 失败: %v", p.BizCode, err)
default:
db.Model(&existing).Updates(map[string]any{
"name": p.Name, "price": p.CNYYuan, "active": true, "sort": i,
})
}
var pp model.ProductPrice
err = db.Where("product_id = ? AND currency = ?", existing.ID, "USDT").First(&pp).Error
switch {
case errors.Is(err, gorm.ErrRecordNotFound):
pp = model.ProductPrice{ProductID: existing.ID, Currency: "USDT", AmountMinor: p.USDTMino}
if err := db.Create(&pp).Error; err != nil {
log.Fatalf("[seed-pangolin] 创建套餐 %s 的 USDT 价失败: %v", p.BizCode, err)
}
case err != nil:
log.Fatalf("[seed-pangolin] 查询套餐 %s 的 USDT 价失败: %v", p.BizCode, err)
default:
db.Model(&pp).Update("amount_minor", p.USDTMino)
}
}
log.Printf("[seed-pangolin] pangolin 业务套餐就绪:%d 个(pro_month/pro_quarter/pro_year,CNY+USDT)", len(plans))
}