Stage 159 (β2): Stripe Checkout for plan upgrades

infra/api/app.py

@@ -57,6 +57,7 @@ from .schemas import (
     AdminRecalibrateBody,
     AdminTriggerRunBody,
     ApiKeyMintBody,
+    BillingCheckoutBody,
     CalibrateBody,
     ConnectorTestBody,
     DecisionPatchBody,
@@ -1196,6 +1197,39 @@ def create_app(db_path: Optional[str] = None,
         except KeyError as e:
             raise ApiError("not_found", str(e), status=404)
 
+    @app.post("/tenants/{tenant_id}/billing/checkout",
+               tags=["tenants"])
+    async def billing_checkout(
+        tenant_id: str,
+        body: BillingCheckoutBody,
+        key: ApiKey = Depends(auth_dep),
+    ):
+        """Stage 159 (β2) — kick off a Stripe Checkout session for a
+        plan upgrade. admin tier — only the tenant's admin can move
+        the plan. Returns {checkout_url, session_id, plan}; the
+        dashboard redirects to checkout_url.
+        Returns 503 when the deployment has no Stripe API key OR no
+        price_id env var for the requested plan, so the UI can
+        switch to a "contact sales" affordance gracefully.
+        """
+        require_tenant_access(key, tenant_id)
+        require_role(key, ROLE_ADMIN)
+        from infra.billing_stripe import StripeUnavailable, create_checkout
+        tenant = svc.get_tenant(tenant_id)
+        if tenant is None:
+            raise ApiError("not_found",
+                            f"unknown tenant {tenant_id!r}", status=404)
+        try:
+            return create_checkout(
+                tenant_id,
+                body.plan,
+                success_url=body.success_url,
+                cancel_url=body.cancel_url,
+                stripe_customer_id=tenant.get("stripe_customer_id"),
+            )
+        except StripeUnavailable as e:
+            raise ApiError("stripe_unavailable", str(e), status=503)
+
     @app.get("/tenants/{tenant_id}/invoices/{period}",
               tags=["tenants"])
     async def get_tenant_invoice(

infra/api/schemas.py

@@ -257,6 +257,22 @@ class ConnectorTestBody(_Body):
     entity_type: Optional[str] = Field(default=None)
 
 
+# --- billing checkout (Stage 159 β2) ----------------------------------
+
+class BillingCheckoutBody(_Body):
+    """Kick off a Stripe Checkout session for a plan upgrade. Both
+    URLs are absolute — the dashboard passes its own /billing pages
+    so success / cancel land back inside the app, not on Stripe."""
+    model_config = _with_example({
+        "plan": "pro",
+        "success_url": "https://orgstate.example.com/billing?upgraded=1",
+        "cancel_url":  "https://orgstate.example.com/billing",
+    })
+    plan: str = Field(..., min_length=1)
+    success_url: str = Field(..., min_length=1)
+    cancel_url:  str = Field(..., min_length=1)
+
+
 # --- decision triage (Stage 156) --------------------------------------
 
 class DecisionPatchBody(_Body):

infra/billing_stripe.py

@@ -313,6 +313,12 @@ DEFAULT_SIG_TOLERANCE_SECONDS = 300
 HANDLED_EVENT_TYPES = frozenset([
     "invoice.paid",
     "invoice.payment_failed",
+    # Stage 159 (β2) — Stripe Checkout for plan upgrades. Stripe
+    # delivers this after the customer completes the Checkout
+    # flow we kick off via create_checkout(). We use the event's
+    # client_reference_id (= tenant_id we set at session-create
+    # time) and metadata.orgstate_plan to bump the tenant's plan.
+    "checkout.session.completed",
 ])
 
 
@@ -443,6 +449,12 @@ def handle_event(svc, event: dict) -> dict:
     if event_type not in HANDLED_EVENT_TYPES:
         return {"status": "ignored", "type": event_type,
                  "event_id": event_id}
+    # Stage 159 (β2) — checkout.session.completed has a different
+    # data shape than invoice.* events; branch before the invoice
+    # lookup below so we don't try to resolve a Checkout session id
+    # in the invoices table.
+    if event_type == "checkout.session.completed":
+        return _handle_checkout_completed(svc, event)
     external_invoice_id = _extract_event_invoice_id(event)
     if external_invoice_id is None:
         raise StripeWebhookError(
@@ -470,3 +482,148 @@ def handle_event(svc, event: dict) -> dict:
     return svc.apply_stripe_webhook_event(
         invoice_row=row, event=event,
     )
+
+
+# ---- Stage 159 (β2): Stripe Checkout for plan upgrades -----------------
+
+# Plan → env var holding the Stripe price_id. The operator sets
+# STRIPE_PRICE_PRO and STRIPE_PRICE_ENTERPRISE on the deployment.
+# When unset, create_checkout returns a clear "plan not configured"
+# error so the dashboard can fall back to "contact sales" gracefully.
+_PLAN_PRICE_ENV: dict = {
+    "pro":        "STRIPE_PRICE_PRO",
+    "enterprise": "STRIPE_PRICE_ENTERPRISE",
+}
+
+
+class _CheckoutTransport:
+    """Minimal Stripe Checkout transport — separated from
+    _RealStripeTransport so the invoice + checkout call sites can be
+    tested independently."""
+
+    def __init__(self, api_key: str):
+        import stripe  # type: ignore
+        self._stripe = stripe
+        stripe.api_key = api_key
+
+    def create_checkout_session(self, *, price_id: str,
+                                 success_url: str, cancel_url: str,
+                                 customer_id: Optional[str] = None,
+                                 client_reference_id: Optional[str] = None,
+                                 metadata: Optional[dict] = None,
+                                 mode: str = "subscription") -> dict:
+        kwargs: dict = {
+            "mode": mode,
+            "line_items": [{"price": price_id, "quantity": 1}],
+            "success_url": success_url,
+            "cancel_url": cancel_url,
+        }
+        if customer_id:
+            kwargs["customer"] = customer_id
+        if client_reference_id:
+            kwargs["client_reference_id"] = client_reference_id
+        if metadata:
+            kwargs["metadata"] = metadata
+        session = self._stripe.checkout.Session.create(**kwargs)
+        return {"id": session.id, "url": session.url}
+
+
+def create_checkout(tenant_id: str, plan: str, *,
+                    success_url: str, cancel_url: str,
+                    stripe_customer_id: Optional[str] = None,
+                    transport: Any = None,
+                    api_key: Optional[str] = None) -> dict:
+    """Create a Stripe Checkout session for ``tenant_id`` upgrading to
+    ``plan`` ('pro' or 'enterprise'). Returns
+    ``{checkout_url, session_id, plan}``.
+
+    The session carries client_reference_id=tenant_id + metadata.
+    orgstate_plan=plan so the eventual checkout.session.completed
+    webhook can route the upgrade back to the right tenant.
+
+    Raises StripeUnavailable on missing api_key or unconfigured plan.
+    """
+    import os
+    if plan not in _PLAN_PRICE_ENV:
+        raise StripeUnavailable(
+            f"unknown plan {plan!r}; available: "
+            f"{sorted(_PLAN_PRICE_ENV.keys())}"
+        )
+    price_id = os.environ.get(_PLAN_PRICE_ENV[plan], "").strip()
+    if not price_id:
+        raise StripeUnavailable(
+            f"plan {plan!r} is not configured for this deployment — "
+            f"operator must set {_PLAN_PRICE_ENV[plan]} env var to a "
+            f"Stripe price_id (e.g. 'price_1Ab...')"
+        )
+    if transport is None:
+        if not api_key:
+            api_key = os.environ.get("STRIPE_API_KEY", "").strip()
+        if not api_key:
+            raise StripeUnavailable(
+                "STRIPE_API_KEY env var is unset — cannot create "
+                "Checkout session without credentials"
+            )
+        if not is_stripe_available():
+            raise StripeUnavailable(
+                "Stripe Checkout needs the `stripe` package — "
+                "`pip install stripe`."
+            )
+        transport = _CheckoutTransport(api_key)
+    session = transport.create_checkout_session(
+        price_id=price_id,
+        success_url=success_url,
+        cancel_url=cancel_url,
+        customer_id=stripe_customer_id or None,
+        client_reference_id=tenant_id,
+        metadata={
+            "orgstate_tenant_id": tenant_id,
+            "orgstate_plan": plan,
+        },
+    )
+    return {
+        "checkout_url": session["url"],
+        "session_id": session["id"],
+        "plan": plan,
+    }
+
+
+def _handle_checkout_completed(svc, event: dict) -> dict:
+    """Apply a verified checkout.session.completed event to the DB.
+
+    Pulls tenant_id from client_reference_id (preferred) or
+    metadata.orgstate_tenant_id (fallback for sessions created
+    outside our flow). Pulls plan from metadata.orgstate_plan. Calls
+    svc.set_tenant_billing_config to bump plan_name +
+    stripe_customer_id atomically. Returns a status dict for ops
+    debugging; route layer always 200s so Stripe stops retrying.
+    """
+    event_id = event.get("id", "")
+    data_object = (event.get("data") or {}).get("object") or {}
+    tenant_id = (data_object.get("client_reference_id")
+                  or (data_object.get("metadata") or {})
+                       .get("orgstate_tenant_id"))
+    plan = (data_object.get("metadata") or {}).get("orgstate_plan")
+    customer_id = data_object.get("customer")
+    if not tenant_id or not plan:
+        return {"status": "skipped_no_tenant_or_plan",
+                "event_id": event_id,
+                "session_id": data_object.get("id")}
+    if plan not in _PLAN_PRICE_ENV:
+        return {"status": "skipped_unknown_plan",
+                "plan": plan, "event_id": event_id}
+    if svc.get_tenant(tenant_id) is None:
+        return {"status": "skipped_unknown_tenant",
+                "tenant_id": tenant_id, "event_id": event_id}
+    svc.set_tenant_billing_config(
+        tenant_id,
+        plan_name=plan,
+        stripe_customer_id=customer_id if customer_id else None,
+    )
+    return {
+        "status": "plan_upgraded",
+        "tenant_id": tenant_id,
+        "plan": plan,
+        "stripe_customer_id": customer_id,
+        "event_id": event_id,
+    }