From c677d64529351bd585af63178503f2aa5d14264a Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Fri, 10 Jul 2026 18:06:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(pay-v2):=20P8=20Task4=20=E7=BB=AD=E8=B4=B9?= =?UTF-8?q?=E5=85=A5=E8=B4=A6=20invoice.paid=E2=86=92renewal=20order=20+?= =?UTF-8?q?=20subscription.renewed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/gateway/settle.go | 55 ++++++++- internal/gateway/subscription_test.go | 141 ++++++++++++++++++++++++ internal/provider/stripe/stripe.go | 46 ++++++-- internal/provider/stripe/stripe_test.go | 50 +++++++++ internal/store/order.go | 30 +++++ 5 files changed, 308 insertions(+), 14 deletions(-) diff --git a/internal/gateway/settle.go b/internal/gateway/settle.go index a4659b0..99a6b23 100644 --- a/internal/gateway/settle.go +++ b/internal/gateway/settle.go @@ -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) { diff --git a/internal/gateway/subscription_test.go b/internal/gateway/subscription_test.go index 55def36..f510114 100644 --- a/internal/gateway/subscription_test.go +++ b/internal/gateway/subscription_test.go @@ -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) + } +} diff --git a/internal/provider/stripe/stripe.go b/internal/provider/stripe/stripe.go index 6a70166..aec98f3 100644 --- a/internal/provider/stripe/stripe.go +++ b/internal/provider/stripe/stripe.go @@ -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) { diff --git a/internal/provider/stripe/stripe_test.go b/internal/provider/stripe/stripe_test.go index 12bf8bb..e6113b2 100644 --- a/internal/provider/stripe/stripe_test.go +++ b/internal/provider/stripe/stripe_test.go @@ -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=,v1=hex(HMAC-SHA256(secret, ".")) func signStripe(payload, secret string, ts int64) string { mac := hmac.New(sha256.New, []byte(secret)) diff --git a/internal/store/order.go b/internal/store/order.go index 58d783f..d10d761 100644 --- a/internal/store/order.go +++ b/internal/store/order.go @@ -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).