| 1 | package abono |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "io" |
| 8 | "net/http" |
| 9 | "time" |
| 10 | |
| 11 | "gorm.io/gorm/clause" |
| 12 | |
| 13 | "amadan.net/rastrillo/rastrillo/carlos" |
| 14 | |
| 15 | "amadan.net/rastrillo/abono/internal/stripeapi" |
| 16 | ) |
| 17 | |
| 18 | // The default addresses of the two handlers that answer machines rather |
| 19 | // than browsers. An app may mount them elsewhere; it passes whatever paths |
| 20 | // it chose to csrf.Exempt, because neither carries a browser's Origin: the |
| 21 | // webhook checks Stripe's signature and the sync checks carlos's bearer. |
| 22 | const ( |
| 23 | WebhookPath = "/abono/webhook" |
| 24 | SyncPath = "/jobs/abono-sync" |
| 25 | ) |
| 26 | |
| 27 | // maxWebhookBody is the largest delivery read. Stripe's events are a few |
| 28 | // kilobytes; a megabyte is generous. |
| 29 | const maxWebhookBody = 1 << 20 |
| 30 | |
| 31 | // event is what the handler reads of a Stripe event: which object it is |
| 32 | // about. The object's fields are in the endpoint's API version, so only |
| 33 | // ids and the checkout hint are taken from it, and the subscription is |
| 34 | // fetched again. |
| 35 | type event struct { |
| 36 | ID string `json:"id"` |
| 37 | Type string `json:"type"` |
| 38 | Data struct { |
| 39 | Object struct { |
| 40 | ID string `json:"id"` |
| 41 | Subscription json.RawMessage `json:"subscription"` |
| 42 | ClientReferenceID string `json:"client_reference_id"` |
| 43 | CustomerDetails struct { |
| 44 | Email string `json:"email"` |
| 45 | } `json:"customer_details"` |
| 46 | // An invoice's subscription, from 2025-03-31.basil on. |
| 47 | Parent struct { |
| 48 | SubscriptionDetails struct { |
| 49 | Subscription string `json:"subscription"` |
| 50 | } `json:"subscription_details"` |
| 51 | } `json:"parent"` |
| 52 | } `json:"object"` |
| 53 | } `json:"data"` |
| 54 | } |
| 55 | |
| 56 | // isCheckout reports whether the event is about a Checkout session, which |
| 57 | // carries the buyer's own hint. |
| 58 | func (e event) isCheckout() bool { |
| 59 | return e.Type == "checkout.session.completed" || e.Type == "checkout.session.async_payment_succeeded" |
| 60 | } |
| 61 | |
| 62 | // subscriptionID is the event's subscription: the object itself, or the one |
| 63 | // a checkout session or invoice is for. Events abono does not read have none. |
| 64 | func (e event) subscriptionID() string { |
| 65 | o := e.Data.Object |
| 66 | switch e.Type { |
| 67 | case "customer.subscription.created", "customer.subscription.updated", "customer.subscription.deleted", |
| 68 | "customer.subscription.paused", "customer.subscription.resumed": |
| 69 | return o.ID |
| 70 | case "checkout.session.completed", "checkout.session.async_payment_succeeded", |
| 71 | "invoice.paid", "invoice.payment_failed", "invoice.payment_succeeded": |
| 72 | var id string |
| 73 | if len(o.Subscription) > 0 && o.Subscription[0] == '"' { |
| 74 | _ = json.Unmarshal(o.Subscription, &id) |
| 75 | } else if len(o.Subscription) > 0 { |
| 76 | var obj struct { |
| 77 | ID string `json:"id"` |
| 78 | } |
| 79 | _ = json.Unmarshal(o.Subscription, &obj) |
| 80 | id = obj.ID |
| 81 | } |
| 82 | if id == "" { |
| 83 | id = o.Parent.SubscriptionDetails.Subscription |
| 84 | } |
| 85 | return id |
| 86 | } |
| 87 | return "" |
| 88 | } |
| 89 | |
| 90 | // hint is what the event and the re-fetched subscription say of whose it |
| 91 | // is: a Checkout session's client_reference_id and typed address, or, for |
| 92 | // the subscription's and its invoices' events, the abono_subject Checkout |
| 93 | // put in the subscription's metadata and the customer's address. |
| 94 | func (e event) hint(sub stripeapi.Subscription) Hint { |
| 95 | if e.isCheckout() { |
| 96 | return Hint{Subject: e.Data.Object.ClientReferenceID, Email: e.Data.Object.CustomerDetails.Email} |
| 97 | } |
| 98 | return subscriptionHint(sub) |
| 99 | } |
| 100 | |
| 101 | // subscriptionHint is what a subscription says of whose it is: the |
| 102 | // abono_subject Checkout put in its metadata for a signed-in buyer, and its |
| 103 | // customer's address. The webhook's subscription and invoice events and |
| 104 | // the reconcile both use it, so a reconcile that recovers lost events ties |
| 105 | // the subscription to whoever bought it, not to whoever owns the address |
| 106 | // it was paid with (ruling R34). |
| 107 | func subscriptionHint(sub stripeapi.Subscription) Hint { |
| 108 | return Hint{Subject: sub.Metadata["abono_subject"], Email: sub.Customer.Email} |
| 109 | } |
| 110 | |
| 111 | type eventRow struct { |
| 112 | ID string |
| 113 | Type string |
| 114 | ReceivedAt time.Time |
| 115 | } |
| 116 | |
| 117 | func (eventRow) TableName() string { return "abono_events" } |
| 118 | |
| 119 | // ServeWebhook is the Stripe endpoint, POST WebhookPath. A delivery whose |
| 120 | // signature does not check, or which is over 1 MiB, is 400. A failure to |
| 121 | // sync, or to reach Stripe, is 500, which Stripe retries for three days. |
| 122 | // Anything else is 200, including events abono does not read and a |
| 123 | // subscription Stripe says no longer exists. Off (no key), it is 404. |
| 124 | // |
| 125 | // The event is read for ids only. The subscription is fetched again, so a |
| 126 | // delivery that arrives late or out of order cannot move a row backwards, |
| 127 | // and a created event that was lost is made up by the next one. |
| 128 | func (s *Service) ServeWebhook(w http.ResponseWriter, r *http.Request) { |
| 129 | if !s.On() { |
| 130 | http.NotFound(w, r) |
| 131 | return |
| 132 | } |
| 133 | body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxWebhookBody)) |
| 134 | if err != nil { |
| 135 | http.Error(w, "unreadable body", http.StatusBadRequest) |
| 136 | return |
| 137 | } |
| 138 | if err := stripeapi.VerifySignature(r.Header.Get("Stripe-Signature"), body, s.cfg.WebhookSecrets, s.now()); err != nil { |
| 139 | http.Error(w, "bad signature", http.StatusBadRequest) |
| 140 | return |
| 141 | } |
| 142 | if status, msg := s.receive(r.Context(), body); status != http.StatusOK { |
| 143 | http.Error(w, msg, status) |
| 144 | return |
| 145 | } |
| 146 | w.WriteHeader(http.StatusOK) |
| 147 | } |
| 148 | |
| 149 | // receive does what an event says, once: the work of ServeWebhook after |
| 150 | // the signature check, and all of it for the sandbox, whose fake hands |
| 151 | // events straight in. It answers the status the webhook should, with a |
| 152 | // message for anything but 200; a failure is logged here. |
| 153 | func (s *Service) receive(ctx context.Context, body []byte) (int, string) { |
| 154 | var e event |
| 155 | if err := json.Unmarshal(body, &e); err != nil || e.ID == "" { |
| 156 | return http.StatusBadRequest, "not an event" |
| 157 | } |
| 158 | |
| 159 | var seen int64 |
| 160 | if err := s.cfg.DB.WithContext(ctx).Table("abono_events").Where("id = ?", e.ID).Count(&seen).Error; err != nil { |
| 161 | return s.fail("webhook: read events", err, "event", e.ID) |
| 162 | } |
| 163 | if seen > 0 { |
| 164 | return http.StatusOK, "" |
| 165 | } |
| 166 | |
| 167 | if id := e.subscriptionID(); id != "" { |
| 168 | if status, msg := s.receiveSubscription(ctx, e, id); status != http.StatusOK { |
| 169 | return status, msg |
| 170 | } |
| 171 | } |
| 172 | |
| 173 | // Recorded after the work, so a failure above is retried in full. Two |
| 174 | // deliveries of one event at once both sync, which is harmless. A record |
| 175 | // that cannot be saved is a failure too: answering 200 would stop Stripe |
| 176 | // retrying with the deduplication unwritten. Sync is idempotent, so the |
| 177 | // retry redoing the work is safe. |
| 178 | if err := s.cfg.DB.WithContext(ctx).Clauses(clause.OnConflict{DoNothing: true}). |
| 179 | Create(&eventRow{ID: e.ID, Type: e.Type, ReceivedAt: s.now()}).Error; err != nil { |
| 180 | return s.fail("webhook: record event", err, "event", e.ID) |
| 181 | } |
| 182 | return http.StatusOK, "" |
| 183 | } |
| 184 | |
| 185 | // receiveSubscription fetches subscription id again and syncs it, holding |
| 186 | // its lock from the fetch to the commit (ruling R35). |
| 187 | func (s *Service) receiveSubscription(ctx context.Context, e event, id string) (int, string) { |
| 188 | unlock, err := s.subs.lock(ctx, id) |
| 189 | if err != nil { |
| 190 | return s.fail("webhook: wait for subscription", err, "event", e.ID, "subscription", id) |
| 191 | } |
| 192 | defer unlock() |
| 193 | sub, err := s.stripe.GetSubscription(ctx, id) |
| 194 | switch { |
| 195 | case stripeapi.NotFound(err): |
| 196 | // Gone from Stripe (deleted with its account data, or the |
| 197 | // delivery is from another mode). Retrying cannot help. |
| 198 | s.cfg.Logger.Warn("abono: webhook: subscription no longer exists", "event", e.ID, "type", e.Type, "subscription", id) |
| 199 | case err != nil: |
| 200 | return s.fail("webhook: fetch subscription", err, "event", e.ID, "subscription", id) |
| 201 | default: |
| 202 | if err := s.sync(ctx, sub, e.hint(sub)); err != nil { |
| 203 | return s.fail("webhook: sync", err, "event", e.ID, "subscription", id) |
| 204 | } |
| 205 | s.cfg.Logger.Info("abono: synced", "event", e.ID, "type", e.Type, "subscription", id, "status", sub.Status) |
| 206 | } |
| 207 | return http.StatusOK, "" |
| 208 | } |
| 209 | |
| 210 | // ServeSync is the daily tick, POST SyncPath: Reconcile, inside the request |
| 211 | // so the instance stays awake for it. Only carlos's bearer token gets in |
| 212 | // (403 otherwise). It answers 204, or 500 when Reconcile failed. It is |
| 213 | // idempotent, so a retried occurrence needs no dedupe. |
| 214 | func (s *Service) ServeSync(w http.ResponseWriter, r *http.Request) { |
| 215 | if !carlos.Tick(r) { |
| 216 | http.Error(w, "forbidden", http.StatusForbidden) |
| 217 | return |
| 218 | } |
| 219 | rep, err := s.Reconcile(r.Context()) |
| 220 | if err != nil { |
| 221 | status, msg := s.fail("reconcile", err, "subscriptions", rep.Subscriptions, "subjects", rep.Subjects) |
| 222 | http.Error(w, msg, status) |
| 223 | return |
| 224 | } |
| 225 | s.cfg.Logger.Info("abono: reconciled", "subscriptions", rep.Subscriptions, "subjects", rep.Subjects) |
| 226 | w.WriteHeader(http.StatusNoContent) |
| 227 | } |
| 228 | |
| 229 | // fail logs err and returns the 500 to answer. A Stripe 401 is a revoked |
| 230 | // or wrong key: the log says so, and the answer is still 500, so Stripe |
| 231 | // retries once the key is fixed. |
| 232 | func (s *Service) fail(msg string, err error, kv ...any) (int, string) { |
| 233 | s.cfg.Logger.Error("abono: "+msg, append(kv, "err", err)...) |
| 234 | s.noteRefusedKey(err) |
| 235 | return http.StatusInternalServerError, "abono failed" |
| 236 | } |
| 237 | |
| 238 | // noteRefusedKey logs when err is Stripe refusing the API key. |
| 239 | func (s *Service) noteRefusedKey(err error) { |
| 240 | var se *stripeapi.Error |
| 241 | if errors.As(err, &se) && se.Status == http.StatusUnauthorized { |
| 242 | s.cfg.Logger.Error("abono: Stripe refused the API key (Config.Key): it is wrong, expired or revoked") |
| 243 | } |
| 244 | } |
| 245 | |
amadan