chat_backend/llm_be/finance/services/webhooks.py

"""Idempotent Stripe webhook handlers that upsert Invoice / Payment rows."""

from __future__ import annotations

import logging
from datetime import datetime, timezone as dt_timezone
from typing import Any

from django.contrib.auth import get_user_model
from django.db import transaction
from django.utils import timezone

from finance.models import Invoice, Payment
from finance.services.plans import assign_founders_from_stripe

logger = logging.getLogger(__name__)
User = get_user_model()


def _ts_to_dt(value: int | None):
    if not value:
        return None
    return datetime.fromtimestamp(value, tz=dt_timezone.utc)


def _resolve_user(*, user_id: str | None = None, customer_email: str | None = None):
    if user_id:
        try:
            return User.objects.get(pk=int(user_id))
        except (User.DoesNotExist, TypeError, ValueError):
            logger.warning("Webhook: user_id=%s not found", user_id)
    if customer_email:
        user = User.objects.filter(email__iexact=customer_email).first()
        if user:
            return user
        logger.warning("Webhook: email=%s not found", customer_email)
    return None


def _user_from_metadata(metadata: dict | None, *, email: str | None = None):
    metadata = metadata or {}
    return _resolve_user(
        user_id=metadata.get("user_id") or metadata.get("client_reference_id"),
        customer_email=email,
    )


@transaction.atomic
def upsert_invoice_from_stripe(
    *,
    user,
    stripe_invoice: dict[str, Any] | None = None,
    stripe_checkout_session_id: str | None = None,
    stripe_subscription_id: str | None = None,
    stripe_customer_id: str | None = None,
    status: str,
    amount_due: int = 0,
    amount_paid: int = 0,
    currency: str = "usd",
    period_start=None,
    period_end=None,
    hosted_invoice_url: str = "",
    description: str = "",
) -> Invoice:
    stripe_invoice_id = None
    if stripe_invoice:
        stripe_invoice_id = stripe_invoice.get("id")
        stripe_subscription_id = (
            stripe_subscription_id or stripe_invoice.get("subscription") or None
        )
        stripe_customer_id = (
            stripe_customer_id or stripe_invoice.get("customer") or None
        )
        amount_due = int(stripe_invoice.get("amount_due") or amount_due or 0)
        amount_paid = int(stripe_invoice.get("amount_paid") or amount_paid or 0)
        currency = (stripe_invoice.get("currency") or currency or "usd").lower()
        period_start = period_start or _ts_to_dt(
            (stripe_invoice.get("period_start") or stripe_invoice.get("created"))
        )
        period_end = period_end or _ts_to_dt(stripe_invoice.get("period_end"))
        hosted_invoice_url = (
            hosted_invoice_url or stripe_invoice.get("hosted_invoice_url") or ""
        )
        description = description or stripe_invoice.get("description") or ""

    lookup: dict[str, Any] = {}
    if stripe_invoice_id:
        lookup["stripe_invoice_id"] = stripe_invoice_id
    elif stripe_checkout_session_id:
        lookup["stripe_checkout_session_id"] = stripe_checkout_session_id
    else:
        raise ValueError("Need stripe_invoice_id or stripe_checkout_session_id")

    defaults = {
        "user": user,
        "company": getattr(user, "company", None),
        "provider": Invoice.Provider.STRIPE,
        "status": status,
        "currency": currency,
        "amount_due": amount_due,
        "amount_paid": amount_paid,
        "period_start": period_start,
        "period_end": period_end,
        "stripe_subscription_id": stripe_subscription_id or None,
        "stripe_customer_id": stripe_customer_id or "",
        "hosted_invoice_url": hosted_invoice_url or "",
        "description": description or "",
    }
    if stripe_invoice_id:
        defaults["stripe_invoice_id"] = stripe_invoice_id
    if stripe_checkout_session_id:
        defaults["stripe_checkout_session_id"] = stripe_checkout_session_id

    invoice, _created = Invoice.objects.update_or_create(
        **lookup,
        defaults=defaults,
    )
    if stripe_invoice_id and invoice.stripe_invoice_id != stripe_invoice_id:
        invoice.stripe_invoice_id = stripe_invoice_id
        invoice.save(update_fields=["stripe_invoice_id", "last_modified"])
    return invoice


@transaction.atomic
def upsert_payment_from_stripe(
    *,
    user,
    invoice: Invoice | None,
    amount: int,
    currency: str = "usd",
    status: str,
    stripe_payment_intent_id: str | None = None,
    stripe_charge_id: str | None = None,
    paid_at=None,
    failure_message: str = "",
) -> Payment:
    if not stripe_payment_intent_id and not stripe_charge_id:
        raise ValueError("Need stripe_payment_intent_id or stripe_charge_id")

    lookup: dict[str, Any] = {}
    if stripe_payment_intent_id:
        lookup["stripe_payment_intent_id"] = stripe_payment_intent_id
    else:
        lookup["stripe_charge_id"] = stripe_charge_id

    defaults = {
        "user": user,
        "company": getattr(user, "company", None),
        "invoice": invoice,
        "provider": Payment.Provider.STRIPE,
        "status": status,
        "currency": (currency or "usd").lower(),
        "amount": int(amount or 0),
        "paid_at": paid_at,
        "failure_message": failure_message or "",
    }
    if stripe_payment_intent_id:
        defaults["stripe_payment_intent_id"] = stripe_payment_intent_id
    if stripe_charge_id:
        defaults["stripe_charge_id"] = stripe_charge_id

    payment, _created = Payment.objects.update_or_create(
        **lookup,
        defaults=defaults,
    )
    return payment


def handle_checkout_session_completed(session: dict[str, Any]) -> Invoice | None:
    metadata = session.get("metadata") or {}
    customer_details = session.get("customer_details") or {}
    user = _user_from_metadata(
        metadata,
        email=customer_details.get("email") or session.get("customer_email"),
    )
    if user is None and session.get("client_reference_id"):
        user = _resolve_user(user_id=session.get("client_reference_id"))
    if user is None:
        logger.error(
            "checkout.session.completed: cannot resolve user for session %s",
            session.get("id"),
        )
        return None

    amount_total = int(session.get("amount_total") or 0)
    invoice = upsert_invoice_from_stripe(
        user=user,
        stripe_checkout_session_id=session.get("id"),
        stripe_subscription_id=session.get("subscription") or None,
        stripe_customer_id=session.get("customer") or None,
        status=(
            Invoice.Status.PAID
            if session.get("payment_status") == "paid"
            else Invoice.Status.OPEN
        ),
        amount_due=amount_total,
        amount_paid=amount_total if session.get("payment_status") == "paid" else 0,
        currency=(session.get("currency") or "usd").lower(),
        description="Subscription checkout",
    )

    payment_intent = session.get("payment_intent")
    if payment_intent and session.get("payment_status") == "paid":
        upsert_payment_from_stripe(
            user=user,
            invoice=invoice,
            amount=amount_total,
            currency=(session.get("currency") or "usd").lower(),
            status=Payment.Status.SUCCEEDED,
            stripe_payment_intent_id=(
                payment_intent if isinstance(payment_intent, str) else None
            ),
            paid_at=timezone.now(),
        )
    if session.get("payment_status") == "paid" or session.get("subscription"):
        assign_founders_from_stripe(
            user,
            stripe_subscription_id=session.get("subscription") or "",
        )
    return invoice


def handle_invoice_paid(stripe_invoice: dict[str, Any]) -> Invoice | None:
    metadata = stripe_invoice.get("metadata") or {}
    user = _user_from_metadata(
        metadata,
        email=stripe_invoice.get("customer_email"),
    )
    if user is None:
        existing = None
        if stripe_invoice.get("id"):
            existing = (
                Invoice.objects.filter(stripe_invoice_id=stripe_invoice["id"])
                .select_related("user")
                .first()
            )
        if existing is None and stripe_invoice.get("subscription"):
            existing = (
                Invoice.objects.filter(
                    stripe_subscription_id=stripe_invoice["subscription"]
                )
                .select_related("user")
                .order_by("-created")
                .first()
            )
        if existing:
            user = existing.user
    if user is None:
        logger.error(
            "invoice.paid: cannot resolve user for invoice %s",
            stripe_invoice.get("id"),
        )
        return None

    invoice = upsert_invoice_from_stripe(
        user=user,
        stripe_invoice=stripe_invoice,
        status=Invoice.Status.PAID,
    )

    payment_intent = stripe_invoice.get("payment_intent")
    charge = stripe_invoice.get("charge")
    if payment_intent or charge:
        paid_at = _ts_to_dt(
            (stripe_invoice.get("status_transitions") or {}).get("paid_at")
        ) or timezone.now()
        upsert_payment_from_stripe(
            user=user,
            invoice=invoice,
            amount=int(stripe_invoice.get("amount_paid") or 0),
            currency=(stripe_invoice.get("currency") or "usd").lower(),
            status=Payment.Status.SUCCEEDED,
            stripe_payment_intent_id=(
                payment_intent if isinstance(payment_intent, str) else None
            ),
            stripe_charge_id=charge if isinstance(charge, str) else None,
            paid_at=paid_at,
        )
    assign_founders_from_stripe(
        user,
        stripe_subscription_id=stripe_invoice.get("subscription") or "",
    )
    return invoice


def handle_invoice_payment_failed(stripe_invoice: dict[str, Any]) -> Invoice | None:
    metadata = stripe_invoice.get("metadata") or {}
    user = _user_from_metadata(
        metadata,
        email=stripe_invoice.get("customer_email"),
    )
    if user is None:
        existing = (
            Invoice.objects.filter(stripe_invoice_id=stripe_invoice.get("id"))
            .select_related("user")
            .first()
        )
        if existing:
            user = existing.user
    if user is None:
        logger.error(
            "invoice.payment_failed: cannot resolve user for invoice %s",
            stripe_invoice.get("id"),
        )
        return None

    invoice = upsert_invoice_from_stripe(
        user=user,
        stripe_invoice=stripe_invoice,
        status=Invoice.Status.PAYMENT_FAILED,
    )

    payment_intent = stripe_invoice.get("payment_intent")
    if payment_intent:
        upsert_payment_from_stripe(
            user=user,
            invoice=invoice,
            amount=int(stripe_invoice.get("amount_due") or 0),
            currency=(stripe_invoice.get("currency") or "usd").lower(),
            status=Payment.Status.FAILED,
            stripe_payment_intent_id=(
                payment_intent if isinstance(payment_intent, str) else None
            ),
            failure_message="Stripe invoice payment failed",
        )
    return invoice


def dispatch_stripe_event(event: dict[str, Any]):
    """Route a verified Stripe event to the appropriate handler."""
    event_type = event.get("type")
    data_object = (event.get("data") or {}).get("object") or {}

    if event_type == "checkout.session.completed":
        return handle_checkout_session_completed(data_object)
    if event_type == "invoice.paid":
        return handle_invoice_paid(data_object)
    if event_type == "invoice.payment_failed":
        return handle_invoice_payment_failed(data_object)

    logger.info("Ignoring unhandled Stripe event type: %s", event_type)
    return None