"""Author RAG — Public Billing API. Routes: GET /api/billing/plans → Public: active pricing plans (for Next.js website) POST /api/billing/checkout → Public: create Stripe Checkout URL POST /api/billing/portal → Auth: self-service billing portal link POST /api/billing/webhook → Public (Stripe-signed): handle Stripe events GET /api/billing/status → Auth: current subscription status """ import json from datetime import datetime, timezone import structlog from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import get_settings from app.dependencies import get_current_author, get_db, get_redis from app.models.pricing_plan import PricingPlan from app.models.stripe_event import StripeEvent from app.services import billing_service as bs logger = structlog.get_logger(__name__) cfg = get_settings() router = APIRouter(prefix="/billing", tags=["billing"]) # ── Schemas ─────────────────────────────────────────────────────────────────── class CheckoutRequest(BaseModel): plan_id: str = Field(..., description="Our internal plan ID, e.g. 'starter'") billing: str = Field("monthly", pattern="^(monthly|annual)$") coupon_code: str | None = Field(None, description="Referral/affiliate coupon code") class PortalRequest(BaseModel): return_url: str = Field(..., description="URL to redirect back to after portal visit") # ── Public: Plan Listing ────────────────────────────────────────────────────── @router.get("/plans") async def list_plans( billing: str = Query("monthly", pattern="^(monthly|annual)$"), db: AsyncSession = Depends(get_db), ): """Return all active pricing plans — consumed by the Next.js website's Pricing section. The website fetches this on page load so prices are always in sync with whatever the SuperAdmin has configured. """ result = await db.execute( select(PricingPlan) .where(PricingPlan.is_active == True) # noqa: E712 .order_by(PricingPlan.sort_order) ) plans = result.scalars().all() return {"plans": [p.to_public_dict(billing=billing) for p in plans]} # ── Public: Create Checkout Session ────────────────────────────────────────── @router.post("/checkout") async def create_checkout( body: CheckoutRequest, db: AsyncSession = Depends(get_db), ): """Create a Stripe Checkout session for the requested plan. Returns the Stripe Checkout URL — the client redirects the browser there. Card is always required upfront (no free access). """ # Fetch the plan and its Stripe Price ID result = await db.execute( select(PricingPlan).where( PricingPlan.id == body.plan_id, PricingPlan.is_active == True, # noqa: E712 ) ) plan = result.scalar_one_or_none() if not plan: raise HTTPException(404, f"Plan '{body.plan_id}' not found or inactive") stripe_price_id = ( plan.stripe_price_id_monthly if body.billing == "monthly" else plan.stripe_price_id_annual ) if not stripe_price_id: raise HTTPException( 503, f"Plan '{body.plan_id}' is not yet configured with a Stripe Price ID. " "Contact support or try again later.", ) # Build success/cancel URLs success_url = ( f"{cfg.FRONTEND_URL}/success?session_id={{CHECKOUT_SESSION_ID}}" ) cancel_url = f"{cfg.FRONTEND_URL}/pricing" try: checkout_url = await bs.create_checkout_session( plan_id=body.plan_id, stripe_price_id=stripe_price_id, success_url=success_url, cancel_url=cancel_url, coupon_code=body.coupon_code, ) except Exception as e: logger.error("Checkout session creation failed", error=str(e)) raise HTTPException(502, "Failed to create checkout session. Please try again.") return {"checkout_url": checkout_url} # ── Auth: Customer Portal ───────────────────────────────────────────────────── @router.post("/portal") async def create_portal( body: PortalRequest, author=Depends(get_current_author), ): """Create a Stripe Customer Portal session for the authenticated author. Authors can: update card, view invoices, cancel, upgrade/downgrade. """ if not author.stripe_customer_id: raise HTTPException(400, "No billing account found. Please contact support.") try: portal_url = await bs.create_portal_session( stripe_customer_id=author.stripe_customer_id, return_url=body.return_url or f"{cfg.FRONTEND_URL}/dashboard", ) except Exception as e: logger.error("Portal session creation failed", author_id=author.id, error=str(e)) raise HTTPException(502, "Failed to open billing portal. Please try again.") return {"portal_url": portal_url} # ── Auth: Subscription Status ───────────────────────────────────────────────── @router.get("/status") async def billing_status(author=Depends(get_current_author)): """Return the current author's subscription status.""" return { "plan_id": author.stripe_plan_id, "stripe_customer_id": author.stripe_customer_id, "stripe_subscription_id": author.stripe_subscription_id, "payment_grace_until": author.payment_grace_until, "in_grace_period": ( author.payment_grace_until is not None and author.payment_grace_until > datetime.now(timezone.utc) ), } # ── Public: Stripe Webhook ──────────────────────────────────────────────────── @router.post("/webhook", status_code=200) async def stripe_webhook( request: Request, stripe_signature: str = Header(..., alias="stripe-signature"), db: AsyncSession = Depends(get_db), ): """Handle Stripe webhook events. Security: - No JWT auth — Stripe signs the payload with STRIPE_WEBHOOK_SECRET - We verify the signature before any processing - Each event is logged to StripeEvent with its Stripe ID as PK so duplicate deliveries are safely ignored (idempotent) Supported events: checkout.session.completed → activate account (handled by auth router) customer.subscription.updated → plan change customer.subscription.deleted → cancel access invoice.payment_succeeded → reset token counter invoice.payment_failed → set grace period + send dunning email """ payload = await request.body() # 1. Verify Stripe signature try: event = bs.verify_webhook_signature(payload, stripe_signature) except Exception: raise HTTPException(400, "Invalid webhook signature") event_id = event["id"] event_type = event["type"] event_obj = event.get("data", {}).get("object", {}) # 2. Idempotency check — skip if already processed existing = await db.get(StripeEvent, event_id) if existing and existing.status == "processed": logger.info("Stripe event already processed, skipping", event_id=event_id) return {"received": True, "skipped": True} # 3. Log the event (upsert — handles retries) now = datetime.now(timezone.utc) if not existing: db.add(StripeEvent( id=event_id, event_type=event_type, related_object_id=event_obj.get("id"), related_object_type=event_obj.get("object"), received_at=now, status="pending", raw_payload=json.dumps(dict(event)), )) await db.flush() # 4. Route to handler billing_svc = bs.BillingService(db) error: str | None = None try: if event_type == "invoice.payment_succeeded": sub_id = event_obj.get("subscription") if sub_id: await billing_svc.handle_payment_succeeded(sub_id) elif event_type == "invoice.payment_failed": sub_id = event_obj.get("subscription") if sub_id: await billing_svc.handle_payment_failed(sub_id, grace_days=7) # TODO Phase 1 Week 4: trigger dunning email via email service elif event_type == "customer.subscription.deleted": sub_id = event_obj.get("id") if sub_id: await billing_svc.handle_subscription_cancelled(sub_id) # TODO Phase 1 Week 4: trigger cancellation email elif event_type == "customer.subscription.updated": sub_id = event_obj.get("id") # Extract the new plan_id from subscription metadata metadata = event_obj.get("metadata", {}) new_plan_id = metadata.get("plan_id") if sub_id and new_plan_id: await billing_svc.handle_plan_changed(sub_id, new_plan_id) # checkout.session.completed is handled by POST /api/auth/activate # (the Next.js success page POSTs the session_id there directly) else: logger.debug("Unhandled Stripe event type", event_type=event_type) except Exception as exc: error = str(exc)[:500] logger.error("Stripe webhook handler error", event_id=event_id, event_type=event_type, error=error) # 5. Update event log status event_record = await db.get(StripeEvent, event_id) if event_record: event_record.processed_at = datetime.now(timezone.utc) event_record.status = "failed" if error else "processed" event_record.processing_error = error await db.commit() # Always return 200 to Stripe — errors are logged, not returned # (returning 4xx/5xx causes Stripe to retry indefinitely) return {"received": True}