wangjia
300 lines
15 KiB
Go
300 lines
15 KiB
Go
package gateway
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log"
|
|
"time"
|
|
|
|
"github.com/wangjia/pay/internal/model"
|
|
"github.com/wangjia/pay/internal/provider"
|
|
"github.com/wangjia/pay/internal/store"
|
|
)
|
|
|
|
type SettleResult string
|
|
|
|
const (
|
|
SettleIgnored SettleResult = "ignored" // 非成功状态(pending/failed)
|
|
SettleNotFound SettleResult = "not_found" // provider_ref 无对应 attempt
|
|
SettleAmountMismatch SettleResult = "amount_mismatch" // 币种不符 / 少付
|
|
SettleDuplicate SettleResult = "duplicate" // 订单已非 pending(幂等 no-op)
|
|
SettleProcessed SettleResult = "processed" // 本次真正翻转为 paid
|
|
SettleFailed SettleResult = "failed" // 暂时性失败(outbox 入队失败等):未翻转,渠道应重投
|
|
)
|
|
|
|
var ErrAmountMismatch = errors.New("gateway: paid amount/currency mismatch")
|
|
|
|
// Settle normalizes a PaidEvent into activation: locate order via provider_ref →
|
|
// attempt → order (设计 §4「归一化 PaidEvent」), reconcile currency+amount, then
|
|
// enqueue-before-flip:先幂等入队 payment.succeeded(unique 键,重复 no-op),
|
|
// 再经 P1 MarkAttemptPaid 幂等翻转(order=pending 原子守卫)。
|
|
//
|
|
// 顺序不变量(资金命脉):**订单为 paid ⇒ outbox 行必已存在**。
|
|
// - 先翻转后入队:两步间崩溃 → 已收钱但业务方永不知情,且无任何机制重试 → 客诉才发现。
|
|
// - 先入队后翻转:两步间崩溃 → outbox 里躺着一条"单还没付"的行;Notifier 投递前有
|
|
// "订单已付"门禁(Task 6),不会把未付单通知出去。渠道因拿不到 200 会重投回调
|
|
// (查单兜底同样收敛),重投时入队/翻转都幂等,自愈。
|
|
// - 入队本身失败:返回 SettleFailed 且不翻转,同样交给渠道重投恢复。
|
|
func (g *Gateway) Settle(ctx context.Context, ev *provider.PaidEvent) (SettleResult, error) {
|
|
if ev.Status != provider.PaidSucceeded {
|
|
return SettleIgnored, nil // 非成功状态:确认收到即可
|
|
}
|
|
att, err := g.orders.AttemptByProviderRef(ev.ProviderRef)
|
|
if err != nil {
|
|
if errors.Is(err, store.ErrAttemptNotFound) {
|
|
return SettleNotFound, nil
|
|
}
|
|
// 读库瞬时失败是可重试态,不能与"查无此单"终态混淆。
|
|
return SettleFailed, err
|
|
}
|
|
// 金额/币种核对:币种须一致,实付须 ≥ 应收(允许 crypto 多付,拒少付)。
|
|
if ev.PaidCurrency != att.Currency || ev.PaidAmountMinor < att.AmountMinor {
|
|
return SettleAmountMismatch, ErrAmountMismatch
|
|
}
|
|
// paid_at 优先用渠道报的支付时间(对账时与渠道流水对得上),渠道不报才落收到时间。
|
|
paidAt := time.Now()
|
|
if ev.PaidAt != nil {
|
|
paidAt = *ev.PaidAt
|
|
}
|
|
|
|
if err := g.enqueuePaymentSucceeded(att, paidAt); err != nil {
|
|
return SettleFailed, err // 未入队绝不翻转;渠道重投时幂等恢复
|
|
}
|
|
flipped, err := g.orders.MarkAttemptPaid(att.OutTradeNo, att.Channel, ev.ProviderRef, paidAt)
|
|
if err != nil {
|
|
return SettleFailed, err
|
|
}
|
|
if !flipped {
|
|
return SettleDuplicate, nil // 已处理过 / 已取消 / 已过期 → 幂等 no-op
|
|
}
|
|
return SettleProcessed, nil
|
|
}
|
|
|
|
// enqueuePaymentSucceeded 组 webhook 领域 payload 并幂等入队。仅订单仍 pending 时入队:
|
|
// 订单已 paid 说明翻转已发生,而翻转严格发生在成功入队之后(顺序不变量),行必已存在;
|
|
// 订单已 canceled 则不该通知(晚到支付走 P4 退款/P6 对账,不自动开通)。
|
|
func (g *Gateway) enqueuePaymentSucceeded(att *model.Attempt, paidAt time.Time) error {
|
|
o, err := g.orders.GetOrder(att.OutTradeNo)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if o.BizSystem == "" || o.Status != model.OrderPendingV2 {
|
|
return nil // 独立收款无业务方回调;或已翻转/已取消
|
|
}
|
|
data := map[string]any{
|
|
"event_type": "payment.succeeded",
|
|
"out_trade_no": o.OutTradeNo,
|
|
"biz_system": o.BizSystem,
|
|
"biz_ref": o.BizRef,
|
|
"product_biz_code": o.BizCode, // 设计 §5:业务方按套餐码映射权益(时长/档位),不硬编码 product_id
|
|
"amount_minor": o.AmountMinor,
|
|
"currency": o.Currency,
|
|
"channel": att.Channel,
|
|
"paid_at": paidAt.Format(time.RFC3339),
|
|
}
|
|
return g.webhook.Enqueue(o.OutTradeNo, o.BizSystem, "payment.succeeded", "", data)
|
|
}
|
|
|
|
// HandleCallback runs a channel's raw callback through its Provider.VerifyCallback
|
|
// (验签/解析封死在渠道内) then dispatches the normalized event by Kind(设计 §5):
|
|
// 一次性/首期支付走既有 Settle(零 fork);订阅续费/催收/取消/拒付各有专属处理器
|
|
// (Task 4/5/6,本 Task 先占 stub 保证独立可编译)。
|
|
func (g *Gateway) HandleCallback(ctx context.Context, method string, in provider.CallbackInput) (SettleResult, error) {
|
|
prov, err := g.providers.Get(method)
|
|
if err != nil {
|
|
return SettleNotFound, err
|
|
}
|
|
ev, err := prov.VerifyCallback(ctx, in)
|
|
if err != nil {
|
|
return SettleNotFound, err
|
|
}
|
|
switch ev.Kind {
|
|
case provider.EventSubscriptionRenewal:
|
|
return g.settleRenewal(ctx, method, ev) // Task 4
|
|
case provider.EventSubscriptionPastDue:
|
|
return g.markSubscriptionPastDue(ctx, method, ev) // Task 5
|
|
case provider.EventSubscriptionCanceled:
|
|
return g.settleSubscriptionCanceled(ctx, method, ev) // Task 5
|
|
case provider.EventChargeback:
|
|
return g.recordChargeback(ctx, method, ev) // Task 6
|
|
default: // EventPayment:一次性 / 订阅首期
|
|
res, serr := g.Settle(ctx, ev)
|
|
// 门必须按结算结果开,不能只看 serr==nil:Settle 对非成功状态(如 Stripe
|
|
// checkout.session.completed 的 payment_status=unpaid 异步支付)返回
|
|
// (SettleIgnored, nil)——serr 为 nil 但订单并未结算,不能诞生订阅。
|
|
// Duplicate = 成功结算的重放(已 paid/已终态),安全且保证重投幂等路径仍能补建订阅行。
|
|
if (res == SettleProcessed || res == SettleDuplicate) && ev.SubscriptionRef != "" {
|
|
if aerr := g.onSubscriptionActivated(ctx, ev); aerr != nil {
|
|
return SettleFailed, aerr // 诞生订阅失败可重试(Stripe 重投)
|
|
}
|
|
}
|
|
return res, serr
|
|
}
|
|
}
|
|
|
|
// --- Task 6 处理器占位(Task 4 续费/Task 5 past_due+取消已替换实现,见 subscription.go)。
|
|
|
|
// settleRenewal 处理续费 invoice.paid(设计 §5/§4 决策记录):每期铸独立 renewal OrderV2
|
|
// (out_trade_no = 首购单号 + "-r-" + invoice id),建即 paid(续费不经收银台,无 pending 中间态)。
|
|
// 幂等靠 renewal attempt 的 (channel,provider_ref=invoice.ID) 唯一索引 + renewal order 的
|
|
// out_trade_no 唯一索引双保险。
|
|
//
|
|
// 入队纪律(与文件头 Settle 的顺序不变量同根同源,但形态不同——续费建单是"单步已 paid",
|
|
// 没有 Settle 那种"翻转严格发生在成功入队之后"的两段式可依赖):**duplicate 分支(created=
|
|
// false)也必须尝试入队**,不能像旧实现那样直接 return。理由:首次入队若瞬时失败,订单/attempt
|
|
// 已在同一事务内落为 paid(不回滚,幂等键护着),而 outbox 从未有行;Stripe 拿不到 200 会重投
|
|
// 同一 invoice.paid,此时 created 必为 false——若 duplicate 分支不入队,该续费通知永久丢失
|
|
// (SyncPendingAttempts 也救不了:续费 attempt 生来就是 AttemptPaid,不在 pending 轮询范围)。
|
|
// outbox 唯一键 (out_trade_no,event_type,refund_id) 上 ON CONFLICT DO NOTHING 天然幂等:行已
|
|
// 存在则本次 no-op,行曾丢失则本次补建——重投即自愈,任何一分支入队失败都仍返回 SettleFailed
|
|
// 交渠道再重投。
|
|
func (g *Gateway) settleRenewal(ctx context.Context, method string, ev *provider.PaidEvent) (SettleResult, error) {
|
|
// channel = 触发本次回调的 method(与 markSubscriptionPastDue/recordChargeback 同式);
|
|
// 订阅诞生时 Subscription.Channel 落的正是 att.Channel=method,查询须对齐,不能硬编码字面量
|
|
// "stripe"(测试固定用 "substripe" 注册子供应商,生产 Stripe 适配器 Method()="stripe")。
|
|
sub, err := g.subs.GetByProviderRef(method, ev.SubscriptionRef)
|
|
if err != nil {
|
|
if errors.Is(err, store.ErrSubNotFound) {
|
|
log.Printf("[renewal] 未知订阅 provider_ref=%s(未诞生/已清理),忽略", ev.SubscriptionRef)
|
|
return SettleIgnored, nil
|
|
}
|
|
return SettleFailed, err
|
|
}
|
|
paidAt := time.Now()
|
|
if ev.PaidAt != nil {
|
|
paidAt = *ev.PaidAt
|
|
}
|
|
renewalNo := sub.OutTradeNo + "-r-" + ev.InvoiceRef
|
|
created, err := g.orders.CreateRenewalPaid(
|
|
&model.OrderV2{
|
|
OutTradeNo: renewalNo, BizSystem: sub.BizSystem, BizRef: sub.BizRef, BizCode: sub.BizCode,
|
|
Subject: "续费", AmountMinor: ev.PaidAmountMinor, Currency: ev.PaidCurrency,
|
|
Status: model.OrderPaidV2, PaidAt: &paidAt,
|
|
},
|
|
&model.Attempt{
|
|
OutTradeNo: renewalNo, Channel: sub.Channel, Provider: sub.Channel, ProviderRef: ev.InvoiceRef,
|
|
AmountMinor: ev.PaidAmountMinor, Currency: ev.PaidCurrency, Status: model.AttemptPaid, PaidAt: &paidAt,
|
|
})
|
|
if err != nil {
|
|
return SettleFailed, err
|
|
}
|
|
// 续费成功即恢复/维持 active,刷新续费锚点(period_end 最小可行取 paidAt+30d;精确值后续从 invoice.period_end 下发)。
|
|
nextEnd := paidAt.Add(30 * 24 * time.Hour)
|
|
if _, err := g.subs.Activate(sub.SubID, &nextEnd); err != nil {
|
|
return SettleFailed, err
|
|
}
|
|
// 无论 created 与否都尝试入队(见函数注释的入队纪律);独立收款(BizSystem=="")无下游可发,跳过。
|
|
if sub.BizSystem != "" {
|
|
if err := g.enqueueRenewed(sub, renewalNo, ev, paidAt); err != nil {
|
|
return SettleFailed, err // 已建单的事实不回滚(幂等键护着);渠道重投时补入队
|
|
}
|
|
}
|
|
if !created {
|
|
return SettleDuplicate, nil // 重投:订单/attempt 已在(幂等 no-op),outbox 已补齐
|
|
}
|
|
return SettleProcessed, nil // 首过(含无业务方的独立收款,对齐 enqueuePaymentSucceeded 语义)
|
|
}
|
|
|
|
// enqueueRenewed 组 subscription.renewed 领域 payload 并幂等入队(抽出供 settleRenewal 的
|
|
// created/duplicate 两分支共用,避免复制两份 payload 构造)。
|
|
func (g *Gateway) enqueueRenewed(sub *model.Subscription, renewalNo string, ev *provider.PaidEvent, paidAt time.Time) error {
|
|
return g.webhook.Enqueue(renewalNo, sub.BizSystem, EvtSubscriptionRenewed, "", map[string]any{
|
|
"event_type": EvtSubscriptionRenewed, "out_trade_no": renewalNo, "sub_id": sub.SubID,
|
|
"biz_system": sub.BizSystem, "biz_ref": sub.BizRef, "product_biz_code": sub.BizCode,
|
|
"amount_minor": ev.PaidAmountMinor, "currency": ev.PaidCurrency, "channel": sub.Channel,
|
|
"paid_at": paidAt.Format(time.RFC3339),
|
|
})
|
|
}
|
|
|
|
// recordChargeback 处理入站 charge.dispute.created(P8 Task6,设计 §6「钱到账不可逆」的
|
|
// 例外形态)。只 Stripe(卡)有此语义;alipay/微信本轮无拒付流,crypto 收款永无
|
|
// chargeback——三者均不产出 EventChargeback,本函数只会被 stripe adapter 触发。
|
|
//
|
|
// 决策记录:不自动回收权益(§6,业务方裁量),只做三件事——①落 Chargeback(幂等 by
|
|
// dispute_ref,DisputeRef 重投 no-op,不双记)②给原订单打 Disputed 标(不改状态机)
|
|
// ③入队 chargeback.received 给业务方自行冲正。out_trade_no 解析失败(订阅拒付/查单失败)
|
|
// 时仍落 Chargeback 留痕 + log 告警,但不阻断、不转发(无法定位业务单)。
|
|
//
|
|
// 入队不按 created 分叉(与 settleRenewal/finalizeCanceled/markSubscriptionPastDue 同型的
|
|
// P8 反纪律修法):Chargeback.Create 与 Enqueue 是两次独立写,不在同一事务——若首次 Create
|
|
// 成功但 Enqueue 瞬时失败,调用方(HandleCallback)拿到 err 后 Stripe 会重投同一
|
|
// charge.dispute.created,此时 created 必为 false;若像"created=false→直接 return
|
|
// SettleDuplicate"那样跳过下面的打标+入队,chargeback.received 通知永久丢失、Disputed
|
|
// 标也永远打不上。改为无论 created 与否都走完打标+入队,outbox 唯一键 ON CONFLICT DO
|
|
// NOTHING + MarkDisputed 条件 UPDATE 天然双重幂等——重投即自愈,不会双记/双发。
|
|
func (g *Gateway) recordChargeback(ctx context.Context, method string, ev *provider.PaidEvent) (SettleResult, error) {
|
|
created, err := g.chargebacks.Create(&model.Chargeback{
|
|
DisputeRef: ev.DisputeRef, OutTradeNo: ev.OutTradeNo, Channel: method,
|
|
ProviderPaymentRef: ev.ProviderPaymentRef, AmountMinor: ev.PaidAmountMinor,
|
|
Currency: ev.PaidCurrency, Reason: ev.Reason, Status: "received",
|
|
})
|
|
if err != nil {
|
|
return SettleFailed, err
|
|
}
|
|
result := SettleProcessed
|
|
if !created {
|
|
result = SettleDuplicate // 拒付重投:Chargeback 已记录过,但仍需补齐下面的打标/入队(见函数注释)
|
|
}
|
|
if ev.OutTradeNo == "" {
|
|
log.Printf("[chargeback] dispute=%s 无法定位业务单(订阅/无 metadata),已记录未转发", ev.DisputeRef)
|
|
return result, nil
|
|
}
|
|
o, err := g.orders.GetOrder(ev.OutTradeNo)
|
|
if err != nil {
|
|
log.Printf("[chargeback] dispute=%s out_trade_no=%s 查单失败: %v", ev.DisputeRef, ev.OutTradeNo, err)
|
|
return result, nil // 已记录 chargeback;定位失败不阻断
|
|
}
|
|
if _, err := g.orders.MarkDisputed(o.OutTradeNo); err != nil { // 打标不改状态机;条件 UPDATE 幂等
|
|
return SettleFailed, err
|
|
}
|
|
if o.BizSystem == "" {
|
|
return result, nil
|
|
}
|
|
if err := g.webhook.Enqueue(o.OutTradeNo, o.BizSystem, EvtChargebackReceived, "", map[string]any{
|
|
"event_type": EvtChargebackReceived, "out_trade_no": o.OutTradeNo, "dispute_ref": ev.DisputeRef,
|
|
"biz_system": o.BizSystem, "biz_ref": o.BizRef, "product_biz_code": o.BizCode,
|
|
"amount_minor": ev.PaidAmountMinor, "currency": ev.PaidCurrency, "reason": ev.Reason,
|
|
"received_at": time.Now().Format(time.RFC3339),
|
|
}); err != nil {
|
|
return SettleFailed, err
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
// SyncPendingAttempts polls every pending attempt via its Provider.Query and
|
|
// settles hits — the query-based backstop for lost webhooks (设计 §8 对账优先).
|
|
// Returns how many attempts were newly settled to paid.
|
|
func (g *Gateway) SyncPendingAttempts(ctx context.Context, limit int) (int, error) {
|
|
atts, err := g.orders.ListAttemptsByStatus(model.AttemptPending, limit)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
settled := 0
|
|
for i := range atts {
|
|
att := &atts[i]
|
|
prov, err := g.providers.Get(att.Channel)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
// 查单带尝试完整上下文(crypto 需要地址/金额/时间窗才能扫链核对)。
|
|
ev, err := prov.Query(ctx, provider.QueryRequest{
|
|
ProviderRef: att.ProviderRef, OutTradeNo: att.OutTradeNo, AccountID: att.AccountID,
|
|
AmountMinor: att.AmountMinor, Currency: att.Currency,
|
|
CreatedAt: att.CreatedAt, ExpiresAt: att.ExpiresAt,
|
|
})
|
|
if err != nil || ev == nil {
|
|
continue
|
|
}
|
|
res, serr := g.Settle(ctx, ev)
|
|
switch res {
|
|
case SettleProcessed:
|
|
settled++
|
|
case SettleIgnored, SettleDuplicate:
|
|
// 常态:未付/已处理,不刷日志
|
|
default: // not_found / amount_mismatch / failed —— 对账兜底的盲区,必须可见
|
|
log.Printf("[settle-sync] attempt=%s channel=%s result=%s err=%v", att.ProviderRef, att.Channel, res, serr)
|
|
}
|
|
}
|
|
return settled, nil
|
|
}
|