chat_backend/llm_be/monetization/services/stripe.py

"""Stripe Checkout, Billing Portal, and webhook dispatch."""

from __future__ import annotations

from typing import Any

import stripe
from django.conf import settings

from monetization.models import Invoice, SubscriptionPlan
from monetization.services.plans import get_plan, seed_subscription_plans
from monetization.services.webhooks import dispatch_stripe_event

__all__ = [
    "StripeNotConfiguredError",
    "configure_stripe",
    "resolve_checkout_plan",
    "subscription_line_items",
    "create_checkout_session",
    "resolve_stripe_customer_id",
    "create_billing_portal_session",
    "dispatch_stripe_event",
]


class StripeNotConfiguredError(RuntimeError):
    """Raised when Stripe secret key is missing."""


def configure_stripe() -> str:
    secret = settings.STRIPE_SECRET_KEY
    if not secret:
        raise StripeNotConfiguredError(
            "STRIPE_SECRET_KEY is not configured. Set it in the environment."
        )
    stripe.api_key = secret
    return secret


def resolve_checkout_plan(plan_slug: str | None = None) -> SubscriptionPlan:
    """Return the plan for Checkout (defaults to public Founders)."""
    seed_subscription_plans(update_existing=False)
    slug = (plan_slug or SubscriptionPlan.Slug.FOUNDERS).strip().lower()
    plan = get_plan(slug)
    if plan is None:
        raise ValueError(f"Unknown plan: {slug}")
    if not plan.is_selectable:
        raise ValueError(f"Plan '{plan.slug}' is not available for checkout.")
    return plan


def subscription_line_items(plan: SubscriptionPlan) -> list[dict[str, Any]]:
    """Build Checkout line_items from a SubscriptionPlan (or legacy settings)."""
    price_id = (plan.stripe_price_id or "").strip() or (
        settings.STRIPE_PRICE_ID if plan.slug == SubscriptionPlan.Slug.FOUNDERS else ""
    )
    if price_id:
        return [{"price": price_id, "quantity": 1}]

    # Founders without a plan stripe_price_id may still use legacy env amount.
    unit_amount = plan.price_cents
    product_name = plan.name
    currency = plan.currency or settings.SUBSCRIPTION_PRICE_CURRENCY
    interval = plan.interval or settings.SUBSCRIPTION_PRICE_INTERVAL
    if plan.slug == SubscriptionPlan.Slug.FOUNDERS and not plan.stripe_price_id:
        # Keep env overrides working for the live Founders price.
        unit_amount = int(
            getattr(settings, "SUBSCRIPTION_PRICE_AMOUNT_CENTS", None) or unit_amount
        )
        product_name = (
            getattr(settings, "SUBSCRIPTION_PRODUCT_NAME", None) or product_name
        )

    return [
        {
            "price_data": {
                "currency": currency,
                "unit_amount": unit_amount,
                "recurring": {"interval": interval},
                "product_data": {"name": product_name},
            },
            "quantity": 1,
        }
    ]


def create_checkout_session(
    *,
    user,
    success_url: str | None = None,
    cancel_url: str | None = None,
    plan_slug: str | None = None,
):
    """Create a Stripe Checkout Session for a selectable subscription plan."""
    configure_stripe()
    plan = resolve_checkout_plan(plan_slug)

    metadata = {
        "user_id": str(user.pk),
        "company_id": str(user.company_id) if user.company_id else "",
        "plan_slug": plan.slug,
    }
    customer_email = getattr(user, "email", None) or None

    session = stripe.checkout.Session.create(
        mode="subscription",
        line_items=subscription_line_items(plan),
        success_url=success_url or settings.STRIPE_CHECKOUT_SUCCESS_URL,
        cancel_url=cancel_url or settings.STRIPE_CHECKOUT_CANCEL_URL,
        customer_email=customer_email,
        client_reference_id=str(user.pk),
        metadata=metadata,
        subscription_data={"metadata": metadata},
    )
    return session, plan


def resolve_stripe_customer_id(*, user) -> str | None:
    """Return the most recent Stripe customer id stored on the user's invoices."""
    return (
        Invoice.objects.filter(user=user)
        .exclude(stripe_customer_id="")
        .order_by("-created")
        .values_list("stripe_customer_id", flat=True)
        .first()
    )


def create_billing_portal_session(
    *,
    customer_id: str,
    return_url: str | None = None,
):
    """Create a Stripe Customer Portal session for plan/payment/cancel management."""
    configure_stripe()
    return stripe.billing_portal.Session.create(
        customer=customer_id,
        return_url=return_url or settings.STRIPE_PORTAL_RETURN_URL,
    )