diff options
Diffstat (limited to '')
| -rw-r--r-- | internal/billing/stripe.go | 203 |
1 files changed, 203 insertions, 0 deletions
diff --git a/internal/billing/stripe.go b/internal/billing/stripe.go new file mode 100644 index 0000000..b06eb6a --- /dev/null +++ b/internal/billing/stripe.go @@ -0,0 +1,203 @@ +package billing + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + + "github.com/jackc/pgx/v5" + "github.com/stripe/stripe-go/v86" + "github.com/stripe/stripe-go/v86/webhook" +) + +const maxWebhookBodyBytes = 1 << 20 + +type stripeCheckoutCreator func(context.Context, *stripe.CheckoutSessionCreateParams) (*stripe.CheckoutSession, error) + +func newStripeCheckoutCreator(apiKey string) stripeCheckoutCreator { + client := stripe.NewClient(apiKey) + return client.V1CheckoutSessions.Create +} + +func (s *Service) CreateCheckout(ctx context.Context, input CheckoutInput) (CheckoutResult, error) { + orderID, _, err := s.createTopUpOrder(ctx, input) + if err != nil { + return CheckoutResult{}, err + } + params := &stripe.CheckoutSessionCreateParams{ + Mode: stripe.String("payment"), + ClientReferenceID: stripe.String(orderID), + IntegrationIdentifier: stripe.String(s.integrationIdentifier), + SuccessURL: stripe.String(s.stripeSuccessURL), + CancelURL: stripe.String(s.stripeCancelURL), + Metadata: map[string]string{ + "aigw_topup_order_id": orderID, + "aigw_tenant_id": strings.TrimSpace(input.TenantID), + }, + LineItems: []*stripe.CheckoutSessionCreateLineItemParams{{ + Quantity: stripe.Int64(1), + PriceData: &stripe.CheckoutSessionCreateLineItemPriceDataParams{ + Currency: stripe.String(s.currency), + UnitAmount: stripe.Int64(input.AmountMinor), + ProductData: &stripe.CheckoutSessionCreateLineItemPriceDataProductDataParams{ + Name: stripe.String("AIGW prepaid balance"), + Description: stripe.String("Prepaid API usage credit"), + }, + }, + }}, + } + params.SetIdempotencyKey("aigw_topup_" + orderID) + session, err := s.createStripeCheckout(ctx, params) + if err != nil { + _, _ = s.db.Exec(ctx, `UPDATE topup_orders SET status = 'failed' WHERE id = $1 AND status = 'pending'`, orderID) + return CheckoutResult{}, fmt.Errorf("create Stripe Checkout Session: %w", err) + } + if session.ID == "" || session.URL == "" { + return CheckoutResult{}, errors.New("Stripe returned an incomplete Checkout Session") + } + if _, err := s.db.Exec(ctx, ` + UPDATE topup_orders SET stripe_session_id = $2, checkout_url = $3 + WHERE id = $1`, orderID, session.ID, session.URL); err != nil { + return CheckoutResult{}, fmt.Errorf("persist Stripe Checkout Session: %w", err) + } + return CheckoutResult{OrderID: orderID, SessionID: session.ID, URL: session.URL}, nil +} + +func (s *Service) WebhookHandler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + body, err := io.ReadAll(io.LimitReader(r.Body, maxWebhookBodyBytes+1)) + if err != nil || len(body) > maxWebhookBodyBytes { + http.Error(w, "invalid webhook body", http.StatusBadRequest) + return + } + event, err := webhook.ConstructEvent(body, r.Header.Get("Stripe-Signature"), s.stripeWebhookSecret) + if err != nil { + http.Error(w, "invalid webhook signature", http.StatusBadRequest) + return + } + if err := s.processStripeEvent(r.Context(), event); err != nil { + if errors.Is(err, ErrInvalidAmount) || isNotFound(err) { + http.Error(w, "invalid checkout event", http.StatusBadRequest) + return + } + http.Error(w, "webhook processing failed", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"received":true}`+"\n") + }) +} + +func (s *Service) processStripeEvent(ctx context.Context, event stripe.Event) error { + typeName := string(event.Type) + switch event.Type { + case stripe.EventTypeCheckoutSessionCompleted, + stripe.EventTypeCheckoutSessionAsyncPaymentSucceeded, + stripe.EventTypeCheckoutSessionAsyncPaymentFailed, + stripe.EventTypeCheckoutSessionExpired: + default: + return nil + } + if event.Data == nil { + return ErrInvalidAmount + } + var session stripe.CheckoutSession + if err := json.Unmarshal(event.Data.Raw, &session); err != nil { + return ErrInvalidAmount + } + if event.ID == "" || session.ID == "" || session.ClientReferenceID == "" { + return ErrInvalidAmount + } + tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted}) + if err != nil { + return err + } + defer tx.Rollback(ctx) + tag, err := tx.Exec(ctx, ` + INSERT INTO stripe_webhook_events (event_id, event_type) VALUES ($1,$2) + ON CONFLICT (event_id) DO NOTHING`, event.ID, typeName) + if err != nil { + return fmt.Errorf("record Stripe event: %w", err) + } + if tag.RowsAffected() == 0 { + return tx.Commit(ctx) + } + + var tenantID, currency, status string + var amountMinor, amountMicros int64 + var storedSessionID *string + err = tx.QueryRow(ctx, ` + SELECT tenant_id::text, amount_minor, amount_micros, currency, status, stripe_session_id + FROM topup_orders WHERE id = $1 FOR UPDATE`, session.ClientReferenceID, + ).Scan(&tenantID, &amountMinor, &amountMicros, ¤cy, &status, &storedSessionID) + if err != nil { + return err + } + if (storedSessionID != nil && *storedSessionID != session.ID) || amountMinor != session.AmountTotal || currency != string(session.Currency) { + return ErrInvalidAmount + } + if event.Type == stripe.EventTypeCheckoutSessionAsyncPaymentFailed || event.Type == stripe.EventTypeCheckoutSessionExpired { + orderStatus := "failed" + if event.Type == stripe.EventTypeCheckoutSessionExpired { + orderStatus = "expired" + } + if _, err := tx.Exec(ctx, `UPDATE topup_orders SET status = $2, stripe_session_id = COALESCE(stripe_session_id, $3) WHERE id = $1 AND status = 'pending'`, session.ClientReferenceID, orderStatus, session.ID); err != nil { + return err + } + _, err = tx.Exec(ctx, `UPDATE stripe_webhook_events SET processed_at = now() WHERE event_id = $1`, event.ID) + if err != nil { + return err + } + return tx.Commit(ctx) + } + if session.PaymentStatus != stripe.CheckoutSessionPaymentStatusPaid { + _, err = tx.Exec(ctx, `UPDATE stripe_webhook_events SET processed_at = now() WHERE event_id = $1`, event.ID) + if err != nil { + return err + } + return tx.Commit(ctx) + } + if status != "paid" { + if _, err := tx.Exec(ctx, ` + INSERT INTO tenant_wallets (tenant_id, currency) VALUES ($1,$2) + ON CONFLICT (tenant_id) DO NOTHING`, tenantID, currency); err != nil { + return err + } + var walletCurrency string + var balance int64 + if err := tx.QueryRow(ctx, `SELECT currency, balance_micros FROM tenant_wallets WHERE tenant_id = $1 FOR UPDATE`, tenantID).Scan(&walletCurrency, &balance); err != nil { + return err + } + if walletCurrency != currency || amountMicros <= 0 || balance > int64(^uint64(0)>>1)-amountMicros { + return ErrInvalidAmount + } + newBalance := balance + amountMicros + if _, err := tx.Exec(ctx, `UPDATE tenant_wallets SET balance_micros = $2, updated_at = now() WHERE tenant_id = $1`, tenantID, newBalance); err != nil { + return err + } + if _, err := tx.Exec(ctx, ` + INSERT INTO billing_ledger (tenant_id, currency, amount_micros, balance_after_micros, kind, source_type, source_id, description) + VALUES ($1,$2,$3,$4,'topup','stripe_checkout',$5,'Stripe balance top-up') + ON CONFLICT (source_type, source_id) DO NOTHING`, tenantID, currency, amountMicros, newBalance, session.ID); err != nil { + return err + } + if _, err := tx.Exec(ctx, ` + UPDATE topup_orders SET status = 'paid', stripe_session_id = COALESCE(stripe_session_id, $2), paid_at = now() + WHERE id = $1`, session.ClientReferenceID, session.ID); err != nil { + return err + } + } + if _, err := tx.Exec(ctx, `UPDATE stripe_webhook_events SET processed_at = now() WHERE event_id = $1`, event.ID); err != nil { + return err + } + return tx.Commit(ctx) +} |