feat(pay-v2): P8 Task4 续费入账 invoice.paid→renewal order + subscription.renewed

This commit is contained in:
wangjia
2026-07-10 18:06:38 +08:00
parent 4c242607a7
commit c677d64529
5 changed files with 308 additions and 14 deletions
+52 -3
View File
@@ -111,7 +111,7 @@ func (g *Gateway) HandleCallback(ctx context.Context, method string, in provider
}
switch ev.Kind {
case provider.EventSubscriptionRenewal:
return g.settleRenewal(ctx, ev) // Task 4
return g.settleRenewal(ctx, method, ev) // Task 4
case provider.EventSubscriptionPastDue:
return g.markSubscriptionPastDue(ctx, method, ev) // Task 5
case provider.EventSubscriptionCanceled:
@@ -132,8 +132,57 @@ func (g *Gateway) HandleCallback(ctx context.Context, method string, in provider
// --- Task 4/5/6 处理器占位(本 Task 只需 default 分支可用 + onSubscriptionActivated)。
// 保持本 Task 独立可编译;后续 Task 各自替换实现 + 补测试。
func (g *Gateway) settleRenewal(ctx context.Context, ev *provider.PaidEvent) (SettleResult, error) {
return SettleFailed, fmt.Errorf("not implemented: %s", ev.Kind)
// 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 唯一索引双保险;出问题让 Stripe 重投,重投时 created=false 不重复入队,自愈。
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
}
if !created || sub.BizSystem == "" {
return SettleDuplicate, nil // 重投 / 独立收款
}
if err := 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),
}); err != nil {
return SettleFailed, err
}
return SettleProcessed, nil
}
func (g *Gateway) markSubscriptionPastDue(ctx context.Context, method string, ev *provider.PaidEvent) (SettleResult, error) {
+141
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"strings"
"testing"
"github.com/wangjia/pay/config"
@@ -187,3 +188,143 @@ func TestSubscriptionActivationOnFirstPayment(t *testing.T) {
t.Fatalf("subscription duplicated on replay: %+v vs %+v", subAfter, sub)
}
}
// TestSettleRenewal 先经首期激活诞生订阅(provider_sub_ref=sub_new),再喂
// EventSubscriptionRenewal:①新增 renewal OrderV2(paid,out_trade_no 含 -r-)
// ②webhook 收到 subscription.renewed ③重投幂等(不双铸/不双发) ④订阅刷新 current_period_end、
// 从 past_due 恢复 active。
func TestSettleRenewal(t *testing.T) {
g, fp, spy, orders, subs := newSubGateway(t)
ctx := context.Background()
res, err := g.CreateSubscription(ctx, gateway.CreateSubscriptionInput{
SKU: "pro_monthly", Method: "substripe", BizSystem: "pangolin", BizRef: "u-1",
})
if err != nil {
t.Fatalf("CreateSubscription: %v", err)
}
activateRaw, err := json.Marshal(provider.PaidEvent{
Kind: provider.EventPayment, ProviderRef: fp.sessionRef, Status: provider.PaidSucceeded,
PaidAmountMinor: 2999, PaidCurrency: "USD", SubscriptionRef: "sub_new",
})
if err != nil {
t.Fatalf("marshal activate event: %v", err)
}
if _, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: activateRaw}); err != nil {
t.Fatalf("activate: %v", err)
}
spy.calls = nil // 只看续费产生的 webhook
// 续费成功前先把订阅打成 past_due,验证续费成功后能恢复 active(路径复用 Activate)。
if ok, err := subs.MarkPastDue("substripe", "sub_new"); err != nil || !ok {
t.Fatalf("MarkPastDue: ok=%v err=%v", ok, err)
}
renewalRaw, err := json.Marshal(provider.PaidEvent{
Kind: provider.EventSubscriptionRenewal, SubscriptionRef: "sub_new", InvoiceRef: "in_123",
PaidAmountMinor: 2999, PaidCurrency: "USD", Status: provider.PaidSucceeded,
})
if err != nil {
t.Fatalf("marshal renewal event: %v", err)
}
result, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: renewalRaw})
if err != nil || result != gateway.SettleProcessed {
t.Fatalf("HandleCallback renewal = %v, %v", result, err)
}
allOrders, err := orders.ListOrders("pangolin", "u-1", 10)
if err != nil {
t.Fatalf("ListOrders: %v", err)
}
var renewalOrder *model.OrderV2
for i := range allOrders {
if strings.Contains(allOrders[i].OutTradeNo, "-r-") {
renewalOrder = &allOrders[i]
}
}
if renewalOrder == nil {
t.Fatalf("renewal order not found: %+v", allOrders)
}
if renewalOrder.Status != model.OrderPaidV2 || renewalOrder.AmountMinor != 2999 || renewalOrder.Currency != "USD" {
t.Fatalf("renewal order = %+v", renewalOrder)
}
if !strings.HasSuffix(renewalOrder.OutTradeNo, "-r-in_123") {
t.Fatalf("renewal out_trade_no = %s, want suffix -r-in_123", renewalOrder.OutTradeNo)
}
if len(spy.calls) != 1 {
t.Fatalf("webhook calls = %d, want 1: %+v", len(spy.calls), spy.calls)
}
renewedCall := spy.calls[0]
if renewedCall["event_type"] != gateway.EvtSubscriptionRenewed {
t.Fatalf("event_type = %v, want %s", renewedCall["event_type"], gateway.EvtSubscriptionRenewed)
}
if renewedCall["sub_id"] != res.SubID {
t.Fatalf("sub_id = %v, want %s", renewedCall["sub_id"], res.SubID)
}
if renewedCall["out_trade_no"] != renewalOrder.OutTradeNo {
t.Fatalf("out_trade_no = %v, want %s", renewedCall["out_trade_no"], renewalOrder.OutTradeNo)
}
if renewedCall["amount_minor"] != int64(2999) {
t.Fatalf("amount_minor = %v, want 2999", renewedCall["amount_minor"])
}
sub, err := subs.GetByProviderRef("substripe", "sub_new")
if err != nil {
t.Fatalf("GetByProviderRef: %v", err)
}
if sub.Status != model.SubActive {
t.Fatalf("subscription status after renewal = %v, want active(recovered from past_due)", sub.Status)
}
if sub.CurrentPeriodEnd == nil {
t.Fatalf("current_period_end not refreshed")
}
// 重投同一 invoice → 不双铸 renewal order、不双发 webhook。
result2, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: renewalRaw})
if err != nil {
t.Fatalf("HandleCallback renewal replay: %v", err)
}
if result2 != gateway.SettleDuplicate {
t.Fatalf("replay result = %v, want duplicate", result2)
}
ordersAfterReplay, err := orders.ListOrders("pangolin", "u-1", 10)
if err != nil {
t.Fatalf("ListOrders after replay: %v", err)
}
renewalCount := 0
for i := range ordersAfterReplay {
if strings.Contains(ordersAfterReplay[i].OutTradeNo, "-r-") {
renewalCount++
}
}
if renewalCount != 1 {
t.Fatalf("renewal order count after replay = %d, want 1", renewalCount)
}
if len(spy.calls) != 1 {
t.Fatalf("webhook calls after replay = %d, want still 1: %+v", len(spy.calls), spy.calls)
}
}
// TestSettleRenewalUnknownSubscription 未知 provider_sub_ref(订阅未诞生/已清理)→ 忽略,不报错。
func TestSettleRenewalUnknownSubscription(t *testing.T) {
g, _, spy, _, _ := newSubGateway(t)
ctx := context.Background()
raw, err := json.Marshal(provider.PaidEvent{
Kind: provider.EventSubscriptionRenewal, SubscriptionRef: "sub_ghost", InvoiceRef: "in_ghost",
PaidAmountMinor: 2999, PaidCurrency: "USD", Status: provider.PaidSucceeded,
})
if err != nil {
t.Fatalf("marshal: %v", err)
}
result, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: raw})
if err != nil {
t.Fatalf("HandleCallback: %v", err)
}
if result != gateway.SettleIgnored {
t.Fatalf("result = %v, want ignored", result)
}
if len(spy.calls) != 0 {
t.Fatalf("webhook calls = %d, want 0: %+v", len(spy.calls), spy.calls)
}
}
+35 -11
View File
@@ -180,20 +180,44 @@ func (p *Provider) VerifyCallback(_ context.Context, in provider.CallbackInput)
if err != nil {
return nil, fmt.Errorf("stripe: webhook 验签失败: %w", err)
}
if event.Type != "checkout.session.completed" {
switch event.Type {
case "checkout.session.completed":
var sess gostripe.CheckoutSession
if err := json.Unmarshal(event.Data.Raw, &sess); err != nil {
return nil, fmt.Errorf("stripe: 解析 session 失败: %w", err)
}
ev := sessionToEvent(&sess, in.Raw)
if event.Created > 0 {
paidAt := unixToTime(event.Created)
ev.PaidAt = &paidAt
}
return ev, nil
case "invoice.paid":
var inv gostripe.Invoice
if err := json.Unmarshal(event.Data.Raw, &inv); err != nil {
return nil, fmt.Errorf("stripe: 解析 invoice 失败: %w", err)
}
if inv.BillingReason != gostripe.InvoiceBillingReasonSubscriptionCycle {
// 首期(subscription_create)由 checkout.session.completed 入账;其余非续费忽略。
return &provider.PaidEvent{Kind: provider.EventPayment, Status: provider.PaidPending, Raw: string(in.Raw)}, nil
}
ev := &provider.PaidEvent{
Kind: provider.EventSubscriptionRenewal, Status: provider.PaidSucceeded,
InvoiceRef: inv.ID, PaidAmountMinor: inv.Total, PaidCurrency: strings.ToUpper(string(inv.Currency)),
Raw: string(in.Raw),
}
if inv.Subscription != nil {
ev.SubscriptionRef = inv.Subscription.ID
}
if event.Created > 0 {
t := unixToTime(event.Created)
ev.PaidAt = &t
}
return ev, nil
default:
// 其它事件此阶段不处理:归一化 pending(管线 Settle 视为 ignored)。
return &provider.PaidEvent{Status: provider.PaidPending, Raw: string(in.Raw)}, nil
}
var sess gostripe.CheckoutSession
if err := json.Unmarshal(event.Data.Raw, &sess); err != nil {
return nil, fmt.Errorf("stripe: 解析 session 失败: %w", err)
}
ev := sessionToEvent(&sess, in.Raw)
if event.Created > 0 {
paidAt := unixToTime(event.Created)
ev.PaidAt = &paidAt
}
return ev, nil
}
func (p *Provider) Query(_ context.Context, req provider.QueryRequest) (*provider.PaidEvent, error) {
+50
View File
@@ -167,6 +167,56 @@ func TestVerifyWebhookWrongSecretFails(t *testing.T) {
}
}
// invoice.paid + billing_reason=subscription_cycle → 归一化为 EventSubscriptionRenewal,
// 携带 invoice/subscription/金额,供 gateway.settleRenewal 铸 renewal order。
func TestVerifyInvoicePaidRenewal(t *testing.T) {
ts := fakeStripeAPI(t)
defer ts.Close()
p := newStripe(t, ts)
payload := `{"id":"evt_r","object":"event","type":"invoice.paid","created":1700000000,"data":{"object":{"id":"in_123","object":"invoice","billing_reason":"subscription_cycle","total":2999,"currency":"usd","subscription":{"id":"sub_new","object":"subscription"}}}}`
sig := signStripe(payload, whSecret, time.Now().Unix())
ev, err := p.VerifyCallback(context.Background(), provider.CallbackInput{
Raw: []byte(payload),
Headers: map[string]string{"Stripe-Signature": sig},
})
if err != nil {
t.Fatalf("verify: %v", err)
}
if ev.Kind != provider.EventSubscriptionRenewal {
t.Fatalf("kind = %v, want EventSubscriptionRenewal", ev.Kind)
}
if ev.InvoiceRef != "in_123" || ev.SubscriptionRef != "sub_new" || ev.PaidAmountMinor != 2999 || ev.Status != provider.PaidSucceeded {
t.Fatalf("event = %+v", ev)
}
if ev.PaidCurrency != "USD" {
t.Fatalf("currency = %s, want USD", ev.PaidCurrency)
}
}
// invoice.paid 但 billing_reason=subscription_create 是首期发票,与 checkout.session.completed
// 是同一笔钱——由后者入账,这里必须跳过(归一化为 EventPayment,不触发续费铸单)。
func TestVerifyInvoicePaidFirstPeriodSkipped(t *testing.T) {
ts := fakeStripeAPI(t)
defer ts.Close()
p := newStripe(t, ts)
payload := `{"id":"evt_r2","object":"event","type":"invoice.paid","created":1700000000,"data":{"object":{"id":"in_first","object":"invoice","billing_reason":"subscription_create","total":2999,"currency":"usd","subscription":{"id":"sub_new","object":"subscription"}}}}`
sig := signStripe(payload, whSecret, time.Now().Unix())
ev, err := p.VerifyCallback(context.Background(), provider.CallbackInput{
Raw: []byte(payload),
Headers: map[string]string{"Stripe-Signature": sig},
})
if err != nil {
t.Fatalf("verify: %v", err)
}
if ev.Kind != provider.EventPayment {
t.Fatalf("kind = %v, want EventPayment(skip)", ev.Kind)
}
}
// signStripe 复刻 Stripe webhook 签名头: t=<ts>,v1=hex(HMAC-SHA256(secret, "<ts>.<payload>"))
func signStripe(payload, secret string, ts int64) string {
mac := hmac.New(sha256.New, []byte(secret))
+30
View File
@@ -8,6 +8,7 @@ import (
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"github.com/wangjia/pay/internal/model"
)
@@ -65,6 +66,35 @@ func (s *OrderStore) MarkAttemptPaid(outTradeNo, channel, providerRef string, at
return flipped, nil
}
// CreateRenewalPaid 幂等建一张已付 renewal order + attempt(续费不经收银台,建即 paid)。
// 重复(invoice 重投)→ created=false。renewal order/attempt 均带唯一约束,ON CONFLICT DO NOTHING。
func (s *OrderStore) CreateRenewalPaid(order *model.OrderV2, att *model.Attempt) (bool, error) {
var created bool
err := s.db.Transaction(func(tx *gorm.DB) error {
ores := tx.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "out_trade_no"}}, DoNothing: true,
}).Create(order)
if ores.Error != nil {
return ores.Error
}
if ores.RowsAffected == 0 {
return nil // 已建过 → 幂等 no-op
}
ares := tx.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "channel"}, {Name: "provider_ref"}}, DoNothing: true,
}).Create(att)
if ares.Error != nil {
return ares.Error
}
created = ares.RowsAffected > 0
return nil
})
if err != nil {
return false, fmt.Errorf("store.CreateRenewalPaid: %w", err)
}
return created, nil
}
func (s *OrderStore) CancelOrder(outTradeNo string) (bool, error) {
res := s.db.Model(&model.OrderV2{}).
Where("out_trade_no = ? AND status = ?", outTradeNo, model.OrderPendingV2).