Скилл payments: интеграция Stripe, биллинг, безопасность

--- name: payments description: Implements Stripe payment integration including one-time charges, subscription billing, webhook handling, and failed payment recovery. Use when adding any payment functionality — checkout flows, subscription plans, refunds, invoices, or proration. Trigger when the user mentions "Stripe", "payments", "subscriptions", "billing", "checkout", "invoice", or "credit card". This skill covers the full payment lifecycle including the security-critical webhook verification step that vibe-coders almost always skip. --- # Payments Engineering (Stripe) ## Purpose Payments are the most consequential code in any SaaS. A bug here is not a UX problem — it is a financial, legal, and trust problem. This skill does not treat Stripe as a simple API call. It treats payment flows as state machines with explicit error handling, idempotency, and webhook-first architecture. ## Payment Type Decision Before writing any code, identify which Stripe flow applies: | User Need | Stripe Flow | |---|---| | User pays once (product, service, credit) | Payment Intent + Checkout Session | | User pays monthly/yearly (SaaS) | Subscription + Prices + Customer | | Cancel and get money back | Refunds API | | Custom invoice / usage-based | Invoice API + Meter | ## SOP: Stripe Integration ### Step 1 - Setup and Keys ```bash uv pip install stripe ``` ```python # config.py (via pydantic-settings) class Settings(BaseSettings): STRIPE_SECRET_KEY: str # sk_live_... or sk_test_... STRIPE_PUBLISHABLE_KEY: str # pk_live_... STRIPE_WEBHOOK_SECRET: str # whsec_... (from Stripe Dashboard > Webhooks) ``` **Never hardcode keys. Keys live in environment variables only.** ### Step 2 - Stripe Client (Singleton) ```python # integrations/stripe_client.py import stripe from src.config import settings stripe.api_key = settings.STRIPE_SECRET_KEY stripe.api_version = "2024-06-20" # pin the API version - never use "latest" # Re-export so callers import from here, not directly from stripe __all__ = ["stripe"] ``` ### Step 3 - Checkout Session (One-Time Payment) ```python # services/payment_service.py from integrations.stripe_client import stripe from uuid import UUID async def create_checkout_session(order_id: UUID, amount_cents: int, user_email: str) -> str: session = stripe.checkout.Session.create( payment_method_types=["card"], line_items=[{ "price_data": { "currency": "usd", "unit_amount": amount_cents, "product_data": {"name": "Order Payment"}, }, "quantity": 1, }], mode="payment", success_url=f"{settings.FRONTEND_URL}/orders/{order_id}/success?session_id={{CHECKOUT_SESSION_ID}}", cancel_url=f"{settings.FRONTEND_URL}/orders/{order_id}/cancel", customer_email=user_email, metadata={"order_id": str(order_id)}, # CRITICAL: link Stripe session to your DB record idempotency_key=str(order_id), # safe to retry if the request fails ) return session.url ``` ### Step 4 - Subscription Billing ```python async def create_subscription(user: User, price_id: str, db: AsyncSession) -> str: """price_id is the Stripe Price ID (e.g., price_1ABC...) from your dashboard.""" # Create or retrieve Stripe Customer if not user.stripe_customer_id: customer = stripe.Customer.create(email=user.email, metadata={"user_id": str(user.id)}) await user_repo.set_stripe_customer_id(db, user.id, customer.id) customer_id = customer.id else: customer_id = user.stripe_customer_id # Create Checkout Session for subscription session = stripe.checkout.Session.create( customer=customer_id, payment_method_types=["card"], line_items=[{"price": price_id, "quantity": 1}], mode="subscription", success_url=f"{settings.FRONTEND_URL}/billing/success", cancel_url=f"{settings.FRONTEND_URL}/billing", subscription_data={"metadata": {"user_id": str(user.id)}}, ) return session.url ``` ### Step 5 - Webhook Handler (The Critical Part) Webhooks are how Stripe tells your server that a payment succeeded. If you only check the redirect URL after checkout, you have a race condition. **The canonical source of truth is the webhook, not the redirect.** ```python # api/v1/webhooks.py from fastapi import APIRouter, Request, HTTPException from integrations.stripe_client import stripe from src.config import settings router = APIRouter() @router.post("/webhooks/stripe", include_in_schema=False) async def stripe_webhook(request: Request, db: AsyncSession = Depends(get_db)): payload = await request.body() sig_header = request.headers.get("stripe-signature") # Step 1: Verify the webhook came from Stripe (NEVER skip this) try: event = stripe.Webhook.construct_event(payload, sig_header, settings.STRIPE_WEBHOOK_SECRET) except stripe.error.SignatureVerificationError: raise HTTPException(status_code=400, detail="Invalid Stripe signature") # Step 2: Use event type to route to the correct handler event_type = event["type"] data = event["data"]["object"] # Step 3: Idempotency - always check if you've already processed this event if await webhook_event_repo.exists(db, event["id"]): return {"status": "already_processed"} # Stripe retries on non-2xx - return 200 match event_type: case "checkout.session.completed": order_id = data["metadata"]["order_id"] await order_service.mark_paid(UUID(order_id), db) case "customer.subscription.created" | "customer.subscription.updated": user_id = data["metadata"]["user_id"] await subscription_service.sync_subscription(UUID(user_id), data, db) case "customer.subscription.deleted": user_id = data["metadata"]["user_id"] await subscription_service.cancel_subscription(UUID(user_id), db) case "invoice.payment_failed": # Notify the user, optionally trigger dunning logic customer_id = data["customer"] await email_service.send_payment_failed_notice(customer_id, db) await webhook_event_repo.create(db, event["id"], event_type) return {"status": "ok"} ``` **Webhook best practices:** - Register the route BEFORE any auth middleware (Stripe doesn't send auth headers) - Return `200 OK` even for events you don't handle - otherwise Stripe retries forever - Store processed event IDs to prevent duplicate processing - In `docker-compose.yml` for local dev, use [Stripe CLI]: `stripe listen --forward-to localhost:8000/api/v1/webhooks/stripe` ### Step 6 - Refunds ```python async def create_refund(payment_intent_id: str, amount_cents: int | None = None) -> stripe.Refund: """amount_cents=None refunds the full amount.""" return stripe.Refund.create( payment_intent=payment_intent_id, amount=amount_cents, # partial refund if specified reason="requested_by_customer", ) # Then update your DB order status to 'refunded' via webhook: charge.refunded ``` ### Step 7 - Frontend (Next.js Checkout Button) ```tsx // app/checkout/page.tsx (client component) "use client"; export function CheckoutButton({ orderId }: { orderId: string }) { const [loading, setLoading] = useState(false); async function handleCheckout() { setLoading(true); const res = await fetch(`/api/orders/${orderId}/checkout`, { method: "POST" }); const { checkout_url } = await res.json(); window.location.href = checkout_url; // redirect to Stripe-hosted checkout } return <Button onClick={handleCheckout} disabled={loading}> {loading ? "Redirecting..." : "Pay Now"} </Button>; } ``` ### DB Schema Additions Required ```python # Add to User model: stripe_customer_id: Mapped[str | None] = mapped_column(String, nullable=True, unique=True) # New table: class WebhookEvent(Base): __tablename__ = "webhook_events" id: Mapped[str] = mapped_column(String, primary_key=True) # Stripe event ID event_type: Mapped[str] = mapped_column(String, nullable=False) processed_at: Mapped[datetime] = mapped_column(default=func.now()) ```