amadan
abono: webhook and daily reconcile
Summary
Adds ServeWebhook (signature check, 1 MiB cap, idempotent events, refetch and sync with checkout or metadata hints) and ServeSync/Reconcile (sync every subscription, refresh prices, run OnChange for every subject holding anything).
Agent notes
OnChange is documented as idempotent because Reconcile calls it for a superset of subjects. Reconcile runs its three steps independently and joins errors so a Stripe outage does not stop the comp sweep. The Stripe double gained a subscription list route.
6 files changed,
+847
−1
abono.go+6 −1reconcile.go+81 −0reconcile_test.go+156 −0stripe_double_test.go+16 −0webhook.go+209 −0webhook_test.go+379 −0
CI passed — run details
CI log
=== vet === go vet ./... === fmt === === test === go test ./... -count=1 ok amadan.net/rastrillo/abono 0.549s ok amadan.net/rastrillo/abono/internal/stripeapi 0.005s === race === CGO_ENABLED=1 go test ./... -race -count=1 ok amadan.net/rastrillo/abono 4.942s ok amadan.net/rastrillo/abono/internal/stripeapi 1.016s === browser === browser: no browser test file yet, nothing to run
Intent
diff --git a/abono.go b/abono.go| index 180f28b..a9d13f2 100644 |
| --- a/abono.go |
| +++ b/abono.go |
| @@ -29,6 +29,10 @@ type HolderFunc func(ctx context.Context, tx *gorm.DB, email, name string) (subj |
| // transaction as the write, so the app's own state never disagrees with |
| // abono's, and its error rolls the write back. Use tx, never the app's own |
| // handle. |
| +// |
| +// It must be idempotent: the daily Reconcile calls it for every subject that |
| +// holds anything, whether or not their answer moved, and a retried webhook |
| +// may run it again for one that did. |
| type ChangeFunc func(ctx context.Context, tx *gorm.DB, subject string) error |
| // Config configures New. DB is required. Key and WebhookSecrets go |
| @@ -50,7 +54,8 @@ type Config struct { |
| // Holder says whose a payment is. Optional. |
| Holder HolderFunc |
| - // OnChange is told when a subject's holdings change. Optional. |
| + // OnChange is told when a subject's holdings change, and by Reconcile for |
| + // every subject holding anything, so it must be idempotent. Optional. |
| OnChange ChangeFunc |
| // Sandbox runs against a local fake of Stripe instead of the real one. |
diff --git a/reconcile.go b/reconcile.go| new file mode 100644 |
| index 0000000..01ee2ee |
| --- /dev/null |
| +++ b/reconcile.go |
| @@ -0,0 +1,81 @@ |
| +package abono |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + |
| + "gorm.io/gorm" |
| + |
| + "amadan.net/rastrillo/abono/internal/stripeapi" |
| +) |
| + |
| +// Report is what a Reconcile looked at. |
| +type Report struct { |
| + // Subscriptions is how many Stripe subscriptions were synced. |
| + Subscriptions int |
| + // Subjects is how many subjects OnChange was run for. |
| + Subjects int |
| +} |
| + |
| +// Reconcile is the daily tick's work. It lists every subscription on the |
| +// Stripe account and syncs each with no hint, which makes up for a webhook |
| +// Stripe gave up on; it refreshes every price's interval, amount and |
| +// currency; and it runs OnChange for every distinct subject that holds any |
| +// subscription row or comp, each in its own transaction. That last step is |
| +// how a comp or a paid period that ran out is told to the app, which no |
| +// event announces. |
| +// |
| +// OnChange must therefore be idempotent: Reconcile calls it for subjects |
| +// whose answer has not moved, a deliberate superset of the ones whose has. |
| +// |
| +// The three steps run independently, so a Stripe outage does not stop the |
| +// sweep for comps that need no Stripe. Their errors are joined. Off (no |
| +// key), only the sweep runs. |
| +func (s *Service) Reconcile(ctx context.Context) (Report, error) { |
| + var rep Report |
| + var errs []error |
| + if s.On() { |
| + err := s.stripe.ListSubscriptions(ctx, func(sub stripeapi.Subscription) error { |
| + rep.Subscriptions++ |
| + return s.sync(ctx, sub, Hint{}) |
| + }) |
| + if err != nil { |
| + s.noteRefusedKey(err) |
| + errs = append(errs, fmt.Errorf("abono: reconcile subscriptions: %w", err)) |
| + } |
| + errs = append(errs, s.refreshPrices(ctx)...) |
| + } |
| + |
| + var subjects []string |
| + if err := s.cfg.DB.WithContext(ctx).Raw(`SELECT subject FROM abono_subscriptions WHERE subject <> '' |
| + UNION SELECT subject FROM abono_comps WHERE subject <> '' ORDER BY subject`).Scan(&subjects).Error; err != nil { |
| + return rep, errors.Join(append(errs, fmt.Errorf("abono: reconcile subjects: %w", err))...) |
| + } |
| + for _, subject := range subjects { |
| + err := s.tx(ctx, func(tx *gorm.DB) error { return s.changed(ctx, tx, subject) }) |
| + if err != nil { |
| + errs = append(errs, fmt.Errorf("abono: reconcile: telling %q: %w", subject, err)) |
| + continue |
| + } |
| + rep.Subjects++ |
| + } |
| + return rep, errors.Join(errs...) |
| +} |
| + |
| +// refreshPrices reads every known price again. A price that cannot be read |
| +// is reported and the rest carry on. |
| +func (s *Service) refreshPrices(ctx context.Context) []error { |
| + var ids []string |
| + if err := s.cfg.DB.WithContext(ctx).Raw(`SELECT stripe_price_id FROM abono_prices ORDER BY stripe_price_id`).Scan(&ids).Error; err != nil { |
| + return []error{fmt.Errorf("abono: reconcile prices: %w", err)} |
| + } |
| + var errs []error |
| + for _, id := range ids { |
| + if err := s.refreshPrice(ctx, id); err != nil { |
| + s.noteRefusedKey(err) |
| + errs = append(errs, err) |
| + } |
| + } |
| + return errs |
| +} |
diff --git a/reconcile_test.go b/reconcile_test.go| new file mode 100644 |
| index 0000000..b7e91bd |
| --- /dev/null |
| +++ b/reconcile_test.go |
| @@ -0,0 +1,156 @@ |
| +package abono_test |
| + |
| +import ( |
| + "context" |
| + "testing" |
| + "time" |
| + |
| + "amadan.net/rastrillo/abono" |
| +) |
| + |
| +func TestReconcileEndsAnExpiredCompThroughOnChange(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + rec := &recorder{} |
| + s, _ := d.service(t, &now, rec.config()) |
| + seedPlans(t, s) |
| + ctx := context.Background() |
| + |
| + ends := t0.Add(2 * time.Hour) |
| + if _, err := s.Comp(ctx, "ana", "annual", &ends, "press"); err != nil { |
| + t.Fatalf("Comp: %v", err) |
| + } |
| + rec.take() |
| + if !hasPlan(t, s, "ana", "annual") { |
| + t.Fatal("the comp does not grant before it runs out") |
| + } |
| + |
| + now = t0.Add(3 * time.Hour) |
| + rep, err := s.Reconcile(ctx) |
| + if err != nil { |
| + t.Fatalf("Reconcile: %v", err) |
| + } |
| + if rep.Subjects != 1 { |
| + t.Errorf("Report = %+v, want 1 subject", rep) |
| + } |
| + if _, changes := rec.take(); len(changes) != 1 || changes[0] != "ana" { |
| + t.Errorf("OnChange ran %v, want once for ana", changes) |
| + } |
| + if hasPlan(t, s, "ana", "annual") { |
| + t.Error("the expired comp still grants") |
| + } |
| +} |
| + |
| +func TestReconcileTellsEverySubjectHoldingAnything(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + rec := &recorder{} |
| + s, _ := d.service(t, &now, rec.config()) |
| + seedPlans(t, s) |
| + ctx := context.Background() |
| + |
| + if _, err := s.Comp(ctx, "cy", "annual", nil, ""); err != nil { |
| + t.Fatalf("Comp: %v", err) |
| + } |
| + // ana and bo hold subscriptions (one on a price nobody sells); a third |
| + // is unclaimed and has nobody to tell. |
| + d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", Email: "ana@example.com", End: t0.Add(30 * day), |
| + Metadata: map[string]string{"abono_subject": "ana"}}) |
| + d.put(t, subSpec{ID: "sub_2", Status: "active", Price: "price_Z", Customer: "cus_2", Email: "bo@example.com", End: t0.Add(30 * day)}) |
| + d.put(t, subSpec{ID: "sub_3", Status: "active", Price: "price_M", Customer: "cus_3", Email: "di@example.com", End: t0.Add(30 * day)}) |
| + for _, id := range []string{"sub_1", "sub_2"} { |
| + if err := s.Sync(ctx, asSubscription(t, d.subs[id]), abono.Hint{Subject: map[string]string{"sub_1": "ana", "sub_2": "bo"}[id]}); err != nil { |
| + t.Fatalf("Sync %s: %v", id, err) |
| + } |
| + } |
| + rec.take() |
| + |
| + rep, err := s.Reconcile(ctx) |
| + if err != nil { |
| + t.Fatalf("Reconcile: %v", err) |
| + } |
| + if rep.Subscriptions != 3 || rep.Subjects != 3 { |
| + t.Errorf("Report = %+v, want 3 subscriptions and 3 subjects", rep) |
| + } |
| + _, changes := rec.take() |
| + got := map[string]int{} |
| + for _, c := range changes { |
| + got[c]++ |
| + } |
| + // sub_3's sync makes its row (unclaimed: Holder answers ""), and ana, bo |
| + // and cy are each told once by the sweep. |
| + for _, who := range []string{"ana", "bo", "cy"} { |
| + if got[who] != 1 { |
| + t.Errorf("%s was told %d times, want 1 (all: %v)", who, got[who], changes) |
| + } |
| + } |
| + if got[""] != 0 || len(got) != 3 { |
| + t.Errorf("OnChange ran for %v", changes) |
| + } |
| +} |
| + |
| +func TestReconcileRefreshesPrices(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + s, db := d.service(t, &now, abono.Config{}) |
| + seedPlans(t, s) |
| + const q = `SELECT stripe_price_id, interval, amount, currency FROM abono_prices ORDER BY stripe_price_id` |
| + |
| + // Stripe's prices change under the account (a new currency, say). |
| + d.price("price_A", "year", 7200, "usd") |
| + d.price("price_M", "month", 700, "usd") |
| + d.price("price_old", "year", 4500, "usd") |
| + |
| + rep, err := s.Reconcile(context.Background()) |
| + if err != nil { |
| + t.Fatalf("Reconcile: %v", err) |
| + } |
| + if rep.Subscriptions != 0 || rep.Subjects != 0 { |
| + t.Errorf("Report = %+v, want nothing", rep) |
| + } |
| + want := "price_A|year|7200|usd\nprice_M|month|700|usd\nprice_old|year|4500|usd\n" |
| + if got := db.dump(q); got != want { |
| + t.Errorf("prices:\n%s\nwant:\n%s", got, want) |
| + } |
| +} |
| + |
| +func TestReconcileRecoversALostSubscription(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + rec := &recorder{subject: "ana"} |
| + s, db := d.service(t, &now, rec.config()) |
| + seedPlans(t, s) |
| + d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", Email: "ana@example.com", End: t0.Add(30 * day)}) |
| + d.put(t, subSpec{ID: "sub_0", Status: "canceled", Price: "price_A", Customer: "cus_0", Email: "old@example.com", End: t0.Add(-day)}) |
| + |
| + rep, err := s.Reconcile(context.Background()) |
| + if err != nil { |
| + t.Fatalf("Reconcile: %v", err) |
| + } |
| + if rep.Subscriptions != 2 || rep.Subjects != 1 { |
| + t.Errorf("Report = %+v, want 2 subscriptions and 1 subject", rep) |
| + } |
| + if got := db.dump(subsQuery); got != "sub_1|ana|annual|active|ana@example.com\n" { |
| + t.Errorf("subscriptions:\n%s", got) |
| + } |
| +} |
| + |
| +func TestReconcileWithStripeOffStillSweepsComps(t *testing.T) { |
| + now := t0 |
| + rec := &recorder{} |
| + s, _ := newService(t, &now, rec.config()) |
| + if _, err := s.Comp(context.Background(), "ana", "annual", nil, ""); err != nil { |
| + t.Fatalf("Comp: %v", err) |
| + } |
| + rec.take() |
| + rep, err := s.Reconcile(context.Background()) |
| + if err != nil { |
| + t.Fatalf("Reconcile: %v", err) |
| + } |
| + if rep.Subjects != 1 { |
| + t.Errorf("Report = %+v", rep) |
| + } |
| + if _, changes := rec.take(); len(changes) != 1 { |
| + t.Errorf("OnChange ran %v", changes) |
| + } |
| +} |
diff --git a/stripe_double_test.go b/stripe_double_test.go| index 90c4b57..687cc27 100644 |
| --- a/stripe_double_test.go |
| +++ b/stripe_double_test.go |
| @@ -7,6 +7,7 @@ import ( |
| "net/http" |
| "net/http/httptest" |
| "net/url" |
| + "sort" |
| "sync" |
| "testing" |
| "time" |
| @@ -53,6 +54,21 @@ func newStripeDouble(t *testing.T) *stripeDouble { |
| d.mu.Unlock() |
| d.reply(w, d.prices, r.PathValue("id")) |
| }) |
| + mux.HandleFunc("GET /v1/subscriptions", func(w http.ResponseWriter, r *http.Request) { |
| + // Every subscription in one page, whatever its status, oldest id first. |
| + d.mu.Lock() |
| + ids := make([]string, 0, len(d.subs)) |
| + for id := range d.subs { |
| + ids = append(ids, id) |
| + } |
| + sort.Strings(ids) |
| + data := make([]any, 0, len(ids)) |
| + for _, id := range ids { |
| + data = append(data, d.subs[id]) |
| + } |
| + d.mu.Unlock() |
| + writeJSON(w, http.StatusOK, map[string]any{"data": data, "has_more": false}) |
| + }) |
| mux.HandleFunc("GET /v1/subscriptions/{id}", func(w http.ResponseWriter, r *http.Request) { |
| d.reply(w, d.subs, r.PathValue("id")) |
| if d.fetched != nil { |
diff --git a/webhook.go b/webhook.go| new file mode 100644 |
| index 0000000..8f568a3 |
| --- /dev/null |
| +++ b/webhook.go |
| @@ -0,0 +1,209 @@ |
| +package abono |
| + |
| +import ( |
| + "encoding/json" |
| + "errors" |
| + "io" |
| + "net/http" |
| + "time" |
| + |
| + "gorm.io/gorm/clause" |
| + |
| + "amadan.net/rastrillo/rastrillo/carlos" |
| + |
| + "amadan.net/rastrillo/abono/internal/stripeapi" |
| +) |
| + |
| +// The default addresses of the two handlers that answer machines rather |
| +// than browsers. An app may mount them elsewhere; it passes whatever paths |
| +// it chose to csrf.Exempt, because neither carries a browser's Origin: the |
| +// webhook checks Stripe's signature and the sync checks carlos's bearer. |
| +const ( |
| + WebhookPath = "/abono/webhook" |
| + SyncPath = "/jobs/abono-sync" |
| +) |
| + |
| +// maxWebhookBody is the largest delivery read. Stripe's events are a few |
| +// kilobytes; a megabyte is generous. |
| +const maxWebhookBody = 1 << 20 |
| + |
| +// event is what the handler reads of a Stripe event: which object it is |
| +// about. The object's fields are in the endpoint's API version, so only |
| +// ids and the checkout hint are taken from it, and the subscription is |
| +// fetched again. |
| +type event struct { |
| + ID string `json:"id"` |
| + Type string `json:"type"` |
| + Data struct { |
| + Object struct { |
| + ID string `json:"id"` |
| + Subscription json.RawMessage `json:"subscription"` |
| + ClientReferenceID string `json:"client_reference_id"` |
| + CustomerDetails struct { |
| + Email string `json:"email"` |
| + } `json:"customer_details"` |
| + // An invoice's subscription, from 2025-03-31.basil on. |
| + Parent struct { |
| + SubscriptionDetails struct { |
| + Subscription string `json:"subscription"` |
| + } `json:"subscription_details"` |
| + } `json:"parent"` |
| + } `json:"object"` |
| + } `json:"data"` |
| +} |
| + |
| +// isCheckout reports whether the event is about a Checkout session, which |
| +// carries the buyer's own hint. |
| +func (e event) isCheckout() bool { |
| + return e.Type == "checkout.session.completed" || e.Type == "checkout.session.async_payment_succeeded" |
| +} |
| + |
| +// subscriptionID is the event's subscription: the object itself, or the one |
| +// a checkout session or invoice is for. Events abono does not read have none. |
| +func (e event) subscriptionID() string { |
| + o := e.Data.Object |
| + switch e.Type { |
| + case "customer.subscription.created", "customer.subscription.updated", "customer.subscription.deleted", |
| + "customer.subscription.paused", "customer.subscription.resumed": |
| + return o.ID |
| + case "checkout.session.completed", "checkout.session.async_payment_succeeded", |
| + "invoice.paid", "invoice.payment_failed", "invoice.payment_succeeded": |
| + var id string |
| + if len(o.Subscription) > 0 && o.Subscription[0] == '"' { |
| + _ = json.Unmarshal(o.Subscription, &id) |
| + } else if len(o.Subscription) > 0 { |
| + var obj struct { |
| + ID string `json:"id"` |
| + } |
| + _ = json.Unmarshal(o.Subscription, &obj) |
| + id = obj.ID |
| + } |
| + if id == "" { |
| + id = o.Parent.SubscriptionDetails.Subscription |
| + } |
| + return id |
| + } |
| + return "" |
| +} |
| + |
| +// hint is what the event and the re-fetched subscription say of whose it |
| +// is: a Checkout session's client_reference_id and typed address, or, for |
| +// the subscription's and its invoices' events, the abono_subject Checkout |
| +// put in the subscription's metadata and the customer's address. |
| +func (e event) hint(sub stripeapi.Subscription) Hint { |
| + if e.isCheckout() { |
| + return Hint{Subject: e.Data.Object.ClientReferenceID, Email: e.Data.Object.CustomerDetails.Email} |
| + } |
| + return Hint{Subject: sub.Metadata["abono_subject"], Email: sub.Customer.Email} |
| +} |
| + |
| +type eventRow struct { |
| + ID string |
| + Type string |
| + ReceivedAt time.Time |
| +} |
| + |
| +func (eventRow) TableName() string { return "abono_events" } |
| + |
| +// ServeWebhook is the Stripe endpoint, POST WebhookPath. A delivery whose |
| +// signature does not check, or which is over 1 MiB, is 400. A failure to |
| +// sync, or to reach Stripe, is 500, which Stripe retries for three days. |
| +// Anything else is 200, including events abono does not read and a |
| +// subscription Stripe says no longer exists. Off (no key), it is 404. |
| +// |
| +// The event is read for ids only. The subscription is fetched again, so a |
| +// delivery that arrives late or out of order cannot move a row backwards, |
| +// and a created event that was lost is made up by the next one. |
| +func (s *Service) ServeWebhook(w http.ResponseWriter, r *http.Request) { |
| + if !s.On() { |
| + http.NotFound(w, r) |
| + return |
| + } |
| + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxWebhookBody)) |
| + if err != nil { |
| + http.Error(w, "unreadable body", http.StatusBadRequest) |
| + return |
| + } |
| + if err := stripeapi.VerifySignature(r.Header.Get("Stripe-Signature"), body, s.cfg.WebhookSecrets, s.now()); err != nil { |
| + http.Error(w, "bad signature", http.StatusBadRequest) |
| + return |
| + } |
| + var e event |
| + if err := json.Unmarshal(body, &e); err != nil || e.ID == "" { |
| + http.Error(w, "not an event", http.StatusBadRequest) |
| + return |
| + } |
| + ctx := r.Context() |
| + |
| + var seen int64 |
| + if err := s.cfg.DB.WithContext(ctx).Table("abono_events").Where("id = ?", e.ID).Count(&seen).Error; err != nil { |
| + s.fail(w, "webhook: read events", err, "event", e.ID) |
| + return |
| + } |
| + if seen > 0 { |
| + w.WriteHeader(http.StatusOK) |
| + return |
| + } |
| + |
| + if id := e.subscriptionID(); id != "" { |
| + sub, err := s.stripe.GetSubscription(ctx, id) |
| + switch { |
| + case stripeapi.NotFound(err): |
| + // Gone from Stripe (deleted with its account data, or the |
| + // delivery is from another mode). Retrying cannot help. |
| + s.cfg.Logger.Warn("abono: webhook: subscription no longer exists", "event", e.ID, "type", e.Type, "subscription", id) |
| + case err != nil: |
| + s.fail(w, "webhook: fetch subscription", err, "event", e.ID, "subscription", id) |
| + return |
| + default: |
| + if err := s.sync(ctx, sub, e.hint(sub)); err != nil { |
| + s.fail(w, "webhook: sync", err, "event", e.ID, "subscription", id) |
| + return |
| + } |
| + s.cfg.Logger.Info("abono: synced", "event", e.ID, "type", e.Type, "subscription", id, "status", sub.Status) |
| + } |
| + } |
| + |
| + // Recorded after the work, so a failure above is retried in full. Two |
| + // deliveries of one event at once both sync, which is harmless. |
| + if err := s.cfg.DB.WithContext(ctx).Clauses(clause.OnConflict{DoNothing: true}). |
| + Create(&eventRow{ID: e.ID, Type: e.Type, ReceivedAt: s.now()}).Error; err != nil { |
| + s.cfg.Logger.Warn("abono: record event", "event", e.ID, "err", err) |
| + } |
| + w.WriteHeader(http.StatusOK) |
| +} |
| + |
| +// ServeSync is the daily tick, POST SyncPath: Reconcile, inside the request |
| +// so the instance stays awake for it. Only carlos's bearer token gets in |
| +// (403 otherwise). It answers 204, or 500 when Reconcile failed. It is |
| +// idempotent, so a retried occurrence needs no dedupe. |
| +func (s *Service) ServeSync(w http.ResponseWriter, r *http.Request) { |
| + if !carlos.Tick(r) { |
| + http.Error(w, "forbidden", http.StatusForbidden) |
| + return |
| + } |
| + rep, err := s.Reconcile(r.Context()) |
| + if err != nil { |
| + s.fail(w, "reconcile", err, "subscriptions", rep.Subscriptions, "subjects", rep.Subjects) |
| + return |
| + } |
| + s.cfg.Logger.Info("abono: reconciled", "subscriptions", rep.Subscriptions, "subjects", rep.Subjects) |
| + w.WriteHeader(http.StatusNoContent) |
| +} |
| + |
| +// fail logs err and answers 500. A Stripe 401 is a revoked or wrong key: |
| +// the log says so, and the answer is still 500, so Stripe retries once the |
| +// key is fixed. |
| +func (s *Service) fail(w http.ResponseWriter, msg string, err error, kv ...any) { |
| + s.cfg.Logger.Error("abono: "+msg, append(kv, "err", err)...) |
| + s.noteRefusedKey(err) |
| + http.Error(w, "abono failed", http.StatusInternalServerError) |
| +} |
| + |
| +// noteRefusedKey logs when err is Stripe refusing the API key. |
| +func (s *Service) noteRefusedKey(err error) { |
| + var se *stripeapi.Error |
| + if errors.As(err, &se) && se.Status == http.StatusUnauthorized { |
| + s.cfg.Logger.Error("abono: Stripe refused the API key (Config.Key): it is wrong, expired or revoked") |
| + } |
| +} |
diff --git a/webhook_test.go b/webhook_test.go| new file mode 100644 |
| index 0000000..bf27476 |
| --- /dev/null |
| +++ b/webhook_test.go |
| @@ -0,0 +1,379 @@ |
| +package abono_test |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "crypto/hmac" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "log/slog" |
| + "net/http" |
| + "net/http/httptest" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "gorm.io/gorm" |
| + |
| + "amadan.net/rastrillo/abono" |
| +) |
| + |
| +// event is a Stripe event as a delivery carries it. |
| +func event(id, typ string, object map[string]any) []byte { |
| + b, err := json.Marshal(map[string]any{"id": id, "type": typ, "data": map[string]any{"object": object}}) |
| + if err != nil { |
| + panic(err) |
| + } |
| + return b |
| +} |
| + |
| +// sign is a Stripe-Signature header for body, made at at. |
| +func sign(secret string, body []byte, at time.Time) string { |
| + mac := hmac.New(sha256.New, []byte(secret)) |
| + fmt.Fprintf(mac, "%d.", at.Unix()) |
| + mac.Write(body) |
| + return fmt.Sprintf("t=%d,v1=%s", at.Unix(), hex.EncodeToString(mac.Sum(nil))) |
| +} |
| + |
| +// deliver posts a body signed with the double's secret at t0. |
| +func deliver(t *testing.T, s *abono.Service, body []byte) *httptest.ResponseRecorder { |
| + t.Helper() |
| + return deliverWith(s, body, sign("whsec_double", body, t0)) |
| +} |
| + |
| +func deliverWith(s *abono.Service, body []byte, signature string) *httptest.ResponseRecorder { |
| + r := httptest.NewRequest(http.MethodPost, abono.WebhookPath, bytes.NewReader(body)) |
| + r.Header.Set("Stripe-Signature", signature) |
| + w := httptest.NewRecorder() |
| + s.ServeWebhook(w, r) |
| + return w |
| +} |
| + |
| +func wantStatus(t *testing.T, w *httptest.ResponseRecorder, want int) { |
| + t.Helper() |
| + if w.Code != want { |
| + t.Fatalf("status = %d, want %d (body %q)", w.Code, want, w.Body.String()) |
| + } |
| +} |
| + |
| +const ( |
| + subsQuery = `SELECT stripe_subscription_id, subject, plan, status, email FROM abono_subscriptions ORDER BY stripe_subscription_id` |
| + eventsQuery = `SELECT id, type FROM abono_events ORDER BY id` |
| +) |
| + |
| +func TestWebhookRefusesBadSignature(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + s, db := d.service(t, &now, abono.Config{}) |
| + seedPlans(t, s) |
| + d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", End: t0.Add(30 * day)}) |
| + body := event("evt_1", "customer.subscription.created", map[string]any{"id": "sub_1"}) |
| + |
| + for name, header := range map[string]string{ |
| + "none": "", |
| + "wrong secret": sign("whsec_other", body, t0), |
| + "stale": sign("whsec_double", body, t0.Add(-time.Hour)), |
| + "other body": sign("whsec_double", []byte("{}"), t0), |
| + } { |
| + wantStatus(t, deliverWith(s, body, header), http.StatusBadRequest) |
| + if got := db.dump(eventsQuery); got != "" { |
| + t.Fatalf("%s: events recorded:\n%s", name, got) |
| + } |
| + if got := db.dump(subsQuery); got != "" { |
| + t.Fatalf("%s: a subscription was synced:\n%s", name, got) |
| + } |
| + } |
| +} |
| + |
| +func TestWebhookRefusesAnOversizeBody(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + s, db := d.service(t, &now, abono.Config{}) |
| + body := []byte(`{"id":"evt_big","type":"customer.created","pad":"` + strings.Repeat("x", 1<<20) + `"}`) |
| + wantStatus(t, deliver(t, s, body), http.StatusBadRequest) |
| + if got := db.dump(eventsQuery); got != "" { |
| + t.Fatalf("events recorded:\n%s", got) |
| + } |
| +} |
| + |
| +func TestWebhookRefusesWhatIsNotAnEvent(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + s, _ := d.service(t, &now, abono.Config{}) |
| + for _, body := range []string{"not json", `{"type":"invoice.paid"}`} { |
| + wantStatus(t, deliver(t, s, []byte(body)), http.StatusBadRequest) |
| + } |
| +} |
| + |
| +func TestWebhookIsIdempotent(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + rec := &recorder{} |
| + s, db := d.service(t, &now, rec.config()) |
| + seedPlans(t, s) |
| + d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", Email: "ana@example.com", End: t0.Add(30 * day), |
| + Metadata: map[string]string{"abono_subject": "ana"}}) |
| + var mu sync.Mutex |
| + fetches := 0 |
| + d.fetched = func() { mu.Lock(); fetches++; mu.Unlock() } |
| + |
| + body := event("evt_1", "customer.subscription.created", map[string]any{"id": "sub_1"}) |
| + wantStatus(t, deliver(t, s, body), http.StatusOK) |
| + wantStatus(t, deliver(t, s, body), http.StatusOK) |
| + |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if fetches != 1 { |
| + t.Errorf("the subscription was read %d times, want 1", fetches) |
| + } |
| + if _, changes := rec.take(); len(changes) != 1 { |
| + t.Errorf("OnChange ran %v, want once", changes) |
| + } |
| + if got := db.dump(eventsQuery); got != "evt_1|customer.subscription.created\n" { |
| + t.Errorf("events:\n%s", got) |
| + } |
| + if got := db.dump(subsQuery); got != "sub_1|ana|annual|active|ana@example.com\n" { |
| + t.Errorf("subscriptions:\n%s", got) |
| + } |
| +} |
| + |
| +func TestWebhookUnreadEventIs200(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + s, db := d.service(t, &now, abono.Config{}) |
| + fetched := false |
| + d.fetched = func() { fetched = true } |
| + |
| + wantStatus(t, deliver(t, s, event("evt_1", "customer.created", map[string]any{"id": "cus_1"})), http.StatusOK) |
| + // A checkout session that is not a subscription has none to read. |
| + wantStatus(t, deliver(t, s, event("evt_2", "checkout.session.completed", map[string]any{"id": "cs_1", "mode": "payment"})), http.StatusOK) |
| + if fetched { |
| + t.Error("a subscription was read for an event that names none") |
| + } |
| + if got := db.dump(eventsQuery); got != "evt_1|customer.created\nevt_2|checkout.session.completed\n" { |
| + t.Errorf("events:\n%s", got) |
| + } |
| +} |
| + |
| +func TestWebhookVanishedSubscriptionIs200(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + s, db := d.service(t, &now, abono.Config{}) |
| + // The subscription is not in the double: Stripe says it does not exist. |
| + wantStatus(t, deliver(t, s, event("evt_1", "customer.subscription.deleted", map[string]any{"id": "sub_gone"})), http.StatusOK) |
| + if got := db.dump(eventsQuery); got != "evt_1|customer.subscription.deleted\n" { |
| + t.Errorf("events:\n%s", got) |
| + } |
| + if got := db.dump(subsQuery); got != "" { |
| + t.Errorf("subscriptions:\n%s", got) |
| + } |
| +} |
| + |
| +func TestWebhookSyncFailureIs500AndNotRecorded(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + var mu sync.Mutex |
| + fail := true |
| + cfg := abono.Config{OnChange: func(_ context.Context, _ *gorm.DB, _ string) error { |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if fail { |
| + return errors.New("the app is down") |
| + } |
| + return nil |
| + }} |
| + s, db := d.service(t, &now, cfg) |
| + seedPlans(t, s) |
| + d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", End: t0.Add(30 * day), |
| + Metadata: map[string]string{"abono_subject": "ana"}}) |
| + body := event("evt_1", "customer.subscription.created", map[string]any{"id": "sub_1"}) |
| + |
| + wantStatus(t, deliver(t, s, body), http.StatusInternalServerError) |
| + if got := db.dump(eventsQuery); got != "" { |
| + t.Errorf("a failed delivery was recorded:\n%s", got) |
| + } |
| + if got := db.dump(subsQuery); got != "" { |
| + t.Errorf("a failed sync left its row:\n%s", got) |
| + } |
| + |
| + // Stripe retries; the app is back. |
| + mu.Lock() |
| + fail = false |
| + mu.Unlock() |
| + wantStatus(t, deliver(t, s, body), http.StatusOK) |
| + if got := db.dump(subsQuery); got != "sub_1|ana|annual|active|\n" { |
| + t.Errorf("subscriptions:\n%s", got) |
| + } |
| + if got := db.dump(eventsQuery); got != "evt_1|customer.subscription.created\n" { |
| + t.Errorf("events:\n%s", got) |
| + } |
| +} |
| + |
| +func TestWebhookStripeRefusingTheKeyIs500(t *testing.T) { |
| + now := t0 |
| + refuse := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { |
| + writeJSON(w, http.StatusUnauthorized, map[string]any{"error": map[string]any{"type": "invalid_request_error", "message": "Invalid API Key"}}) |
| + })) |
| + defer refuse.Close() |
| + var logs bytes.Buffer |
| + s, db := newService(t, &now, abono.Config{ |
| + Key: "rk_test_revoked", WebhookSecrets: []string{"whsec_double"}, StripeBase: refuse.URL, |
| + Origin: "https://site.example", Logger: slog.New(slog.NewTextHandler(&logs, nil)), |
| + }) |
| + wantStatus(t, deliver(t, s, event("evt_1", "customer.subscription.updated", map[string]any{"id": "sub_1"})), http.StatusInternalServerError) |
| + if !strings.Contains(logs.String(), "refused the API key") { |
| + t.Errorf("the log does not say the key was refused:\n%s", logs.String()) |
| + } |
| + if got := db.dump(eventsQuery); got != "" { |
| + t.Errorf("events:\n%s", got) |
| + } |
| +} |
| + |
| +func TestWebhookOffIs404(t *testing.T) { |
| + now := t0 |
| + s, db := newService(t, &now, abono.Config{}) |
| + if s.On() { |
| + t.Fatal("service is On with no key") |
| + } |
| + body := event("evt_1", "customer.subscription.created", map[string]any{"id": "sub_1"}) |
| + wantStatus(t, deliver(t, s, body), http.StatusNotFound) |
| + if got := db.dump(eventsQuery); got != "" { |
| + t.Errorf("events:\n%s", got) |
| + } |
| +} |
| + |
| +// An updated event for an ended subscription this service never saw |
| +// creates nothing (R18): no resurrection. |
| +func TestWebhookUpdatedForAnEndedUnseenSubscriptionCreatesNothing(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + s, db := d.service(t, &now, abono.Config{}) |
| + seedPlans(t, s) |
| + d.put(t, subSpec{ID: "sub_old", Status: "canceled", Price: "price_A", Customer: "cus_1", Email: "ana@example.com", End: t0.Add(-day)}) |
| + |
| + wantStatus(t, deliver(t, s, event("evt_1", "customer.subscription.updated", map[string]any{"id": "sub_old"})), http.StatusOK) |
| + if got := db.dump(subsQuery); got != "" { |
| + t.Errorf("an ended subscription first seen was recorded:\n%s", got) |
| + } |
| + if got := db.dump(eventsQuery); got != "evt_1|customer.subscription.updated\n" { |
| + t.Errorf("events:\n%s", got) |
| + } |
| +} |
| + |
| +// A lost created event recovers: the next event for the subscription makes |
| +// its row (R16). |
| +func TestWebhookUpdatedForAGrantingUnseenSubscriptionCreatesARow(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + rec := &recorder{subject: "ana"} |
| + s, db := d.service(t, &now, rec.config()) |
| + seedPlans(t, s) |
| + d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", Email: "ana@example.com", End: t0.Add(30 * day)}) |
| + |
| + wantStatus(t, deliver(t, s, event("evt_1", "customer.subscription.updated", map[string]any{"id": "sub_1"})), http.StatusOK) |
| + if got := db.dump(subsQuery); got != "sub_1|ana|annual|active|ana@example.com\n" { |
| + t.Errorf("subscriptions:\n%s", got) |
| + } |
| + if _, changes := rec.take(); len(changes) != 1 || changes[0] != "ana" { |
| + t.Errorf("OnChange ran %v, want once for ana", changes) |
| + } |
| +} |
| + |
| +func TestWebhookCheckoutCompletedCarriesItsHint(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + rec := &recorder{} |
| + s, db := d.service(t, &now, rec.config()) |
| + seedPlans(t, s) |
| + // The customer has no address on file: only the session knows it. |
| + d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", End: t0.Add(30 * day)}) |
| + |
| + body := event("evt_1", "checkout.session.completed", map[string]any{ |
| + "id": "cs_1", "subscription": "sub_1", "client_reference_id": "ana", |
| + "customer_details": map[string]any{"email": "Ana@Example.com"}, |
| + }) |
| + wantStatus(t, deliver(t, s, body), http.StatusOK) |
| + if got := db.dump(subsQuery); got != "sub_1|ana|annual|active|ana@example.com\n" { |
| + t.Errorf("subscriptions:\n%s", got) |
| + } |
| + |
| + // An unpaid session's later payment is a second event for the same one. |
| + body = event("evt_2", "checkout.session.async_payment_succeeded", map[string]any{"id": "cs_1", "subscription": "sub_1", "client_reference_id": "ana"}) |
| + wantStatus(t, deliver(t, s, body), http.StatusOK) |
| +} |
| + |
| +func TestWebhookInvoiceNamesItsSubscriptionTwoWays(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + s, db := d.service(t, &now, abono.Config{}) |
| + seedPlans(t, s) |
| + d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", Email: "ana@example.com", End: t0.Add(30 * day), |
| + Metadata: map[string]string{"abono_subject": "ana"}}) |
| + d.put(t, subSpec{ID: "sub_2", Status: "active", Price: "price_M", Customer: "cus_2", Email: "bo@example.com", End: t0.Add(30 * day), |
| + Metadata: map[string]string{"abono_subject": "bo"}}) |
| + |
| + // Before 2025-03-31.basil: "subscription" on the invoice. |
| + wantStatus(t, deliver(t, s, event("evt_1", "invoice.paid", map[string]any{"id": "in_1", "subscription": "sub_1"})), http.StatusOK) |
| + // From it: parent.subscription_details. |
| + wantStatus(t, deliver(t, s, event("evt_2", "invoice.payment_failed", map[string]any{ |
| + "id": "in_2", "parent": map[string]any{"subscription_details": map[string]any{"subscription": "sub_2"}}})), http.StatusOK) |
| + |
| + want := "sub_1|ana|annual|active|ana@example.com\nsub_2|bo|monthly|active|bo@example.com\n" |
| + if got := db.dump(subsQuery); got != want { |
| + t.Errorf("subscriptions:\n%s\nwant:\n%s", got, want) |
| + } |
| +} |
| + |
| +func TestSyncNeedsTick(t *testing.T) { |
| + t.Setenv("CARLOS_ADMIN_TOKEN", "secret-token") |
| + now := t0 |
| + d := newStripeDouble(t) |
| + s, db := d.service(t, &now, abono.Config{}) |
| + seedPlans(t, s) |
| + d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", Email: "ana@example.com", End: t0.Add(30 * day)}) |
| + |
| + call := func(auth string) int { |
| + r := httptest.NewRequest(http.MethodPost, abono.SyncPath, nil) |
| + if auth != "" { |
| + r.Header.Set("Authorization", auth) |
| + } |
| + w := httptest.NewRecorder() |
| + s.ServeSync(w, r) |
| + return w.Code |
| + } |
| + for _, auth := range []string{"", "Bearer wrong"} { |
| + if got := call(auth); got != http.StatusForbidden { |
| + t.Errorf("Authorization %q: status %d, want 403", auth, got) |
| + } |
| + } |
| + if got := db.dump(subsQuery); got != "" { |
| + t.Fatalf("a refused tick reconciled:\n%s", got) |
| + } |
| + if got := call("Bearer secret-token"); got != http.StatusNoContent { |
| + t.Fatalf("with the token: status %d, want 204", got) |
| + } |
| + if got := db.dump(subsQuery); got != "sub_1||annual|active|ana@example.com\n" { |
| + t.Errorf("subscriptions after the tick:\n%s", got) |
| + } |
| +} |
| + |
| +func TestSyncFailureIs500(t *testing.T) { |
| + t.Setenv("CARLOS_ADMIN_TOKEN", "secret-token") |
| + now := t0 |
| + broken := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { |
| + writeJSON(w, http.StatusInternalServerError, map[string]any{"error": map[string]any{"message": "down"}}) |
| + })) |
| + defer broken.Close() |
| + s, _ := newService(t, &now, abono.Config{ |
| + Key: "rk_test_x", WebhookSecrets: []string{"whsec_double"}, StripeBase: broken.URL, Origin: "https://site.example", |
| + }) |
| + r := httptest.NewRequest(http.MethodPost, abono.SyncPath, nil) |
| + r.Header.Set("Authorization", "Bearer secret-token") |
| + w := httptest.NewRecorder() |
| + s.ServeSync(w, r) |
| + wantStatus(t, w, http.StatusInternalServerError) |
| +} |