Monetization app + RevenueCat webhooks (store IAP ledger) (#69)
## Summary - Rename `finance` → **`monetization`** Django app (keep `finance_*` tables via `label = "finance"`) - Add `services/stripe.py` + `services/revenuecat.py`; RevenueCat webhook upserts **subscription + Invoice/Payment** (billing history parity with Stripe) - Mount `/api/monetization/` + keep `/api/finance/` alias - Extend `Source`/`Provider` with `revenuecat`; product→plan mapping via `revenuecat_product_id` / `REVENUECAT_PRODUCT_PLAN_MAP` Closes #68. Companion to [chat_web_app#100](ai_ml_operations/chat_web_app#100). ## Test plan - [x] `manage.py test monetization.tests` (54 OK) - [x] Smoke `chat_backend.tests.test_views_documents` + `test_oauth` - [ ] Deploy: set `REVENUECAT_WEBHOOK_SECRET`; point RC webhook at `/api/finance/webhooks/revenuecat/` - [ ] Map store product IDs on `SubscriptionPlan.revenuecat_product_id` (or env JSON map) - [ ] Sandbox INITIAL_PURCHASE → subscription `source=revenuecat` + invoice in `/finance/invoices/`Reviewed-on: #69
This commit was merged in pull request #69.
This commit is contained in:
@@ -0,0 +1,465 @@
|
||||
"""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 monetization.models import Invoice, Payment, UserSubscription
|
||||
from monetization.services.plans import (
|
||||
assign_plan_from_stripe,
|
||||
get_or_create_user_subscription,
|
||||
log_subscription_auth_event,
|
||||
resolve_plan_from_stripe_price,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
def _stripe_status_to_local(stripe_status: str | None) -> str:
|
||||
mapping = {
|
||||
"active": UserSubscription.Status.ACTIVE,
|
||||
"trialing": UserSubscription.Status.ACTIVE,
|
||||
"past_due": UserSubscription.Status.PAST_DUE,
|
||||
"unpaid": UserSubscription.Status.PAST_DUE,
|
||||
"canceled": UserSubscription.Status.CANCELED,
|
||||
"incomplete_expired": UserSubscription.Status.CANCELED,
|
||||
}
|
||||
return mapping.get((stripe_status or "").lower(), UserSubscription.Status.NONE)
|
||||
|
||||
|
||||
def _plan_slug_from_subscription(subscription: dict[str, Any]) -> str | None:
|
||||
metadata = subscription.get("metadata") or {}
|
||||
if metadata.get("plan_slug"):
|
||||
return metadata.get("plan_slug")
|
||||
items = (subscription.get("items") or {}).get("data") or []
|
||||
if not items:
|
||||
return None
|
||||
price = (items[0] or {}).get("price") or {}
|
||||
price_id = price.get("id") if isinstance(price, dict) else None
|
||||
plan = resolve_plan_from_stripe_price(price_id)
|
||||
return plan.slug if plan else None
|
||||
|
||||
|
||||
def _user_from_subscription(subscription: dict[str, Any]):
|
||||
metadata = subscription.get("metadata") or {}
|
||||
user = _user_from_metadata(metadata)
|
||||
if user is not None:
|
||||
return user
|
||||
sub_id = subscription.get("id")
|
||||
if sub_id:
|
||||
existing = (
|
||||
Invoice.objects.filter(stripe_subscription_id=sub_id)
|
||||
.select_related("user")
|
||||
.order_by("-created")
|
||||
.first()
|
||||
)
|
||||
if existing:
|
||||
return existing.user
|
||||
local_sub = (
|
||||
UserSubscription.objects.filter(stripe_subscription_id=sub_id)
|
||||
.select_related("user")
|
||||
.first()
|
||||
)
|
||||
if local_sub:
|
||||
return local_sub.user
|
||||
return None
|
||||
|
||||
|
||||
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_plan_from_stripe(
|
||||
user,
|
||||
plan_slug=metadata.get("plan_slug"),
|
||||
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_plan_from_stripe(
|
||||
user,
|
||||
plan_slug=metadata.get("plan_slug"),
|
||||
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 handle_customer_subscription_updated(subscription: dict[str, Any]):
|
||||
"""Sync local UserSubscription after portal plan change / cancel schedule."""
|
||||
user = _user_from_subscription(subscription)
|
||||
if user is None:
|
||||
logger.error(
|
||||
"customer.subscription.updated: cannot resolve user for %s",
|
||||
subscription.get("id"),
|
||||
)
|
||||
return None
|
||||
|
||||
local_status = _stripe_status_to_local(subscription.get("status"))
|
||||
if subscription.get("cancel_at_period_end") and local_status == (
|
||||
UserSubscription.Status.ACTIVE
|
||||
):
|
||||
# Still active until period end; keep ACTIVE and surface cancel flag.
|
||||
pass
|
||||
|
||||
return assign_plan_from_stripe(
|
||||
user,
|
||||
plan_slug=_plan_slug_from_subscription(subscription),
|
||||
stripe_subscription_id=subscription.get("id") or "",
|
||||
status=local_status or UserSubscription.Status.ACTIVE,
|
||||
cancel_at_period_end=bool(subscription.get("cancel_at_period_end")),
|
||||
current_period_end=_ts_to_dt(subscription.get("current_period_end")),
|
||||
keep_existing_plan_if_unknown=True,
|
||||
)
|
||||
|
||||
|
||||
def handle_customer_subscription_deleted(subscription: dict[str, Any]):
|
||||
"""Mark local subscription canceled when Stripe subscription ends."""
|
||||
user = _user_from_subscription(subscription)
|
||||
if user is None:
|
||||
logger.error(
|
||||
"customer.subscription.deleted: cannot resolve user for %s",
|
||||
subscription.get("id"),
|
||||
)
|
||||
return None
|
||||
|
||||
sub = get_or_create_user_subscription(user)
|
||||
prev_status = sub.status
|
||||
sub.status = UserSubscription.Status.CANCELED
|
||||
sub.cancel_at_period_end = False
|
||||
sub.current_period_end = _ts_to_dt(subscription.get("current_period_end"))
|
||||
if subscription.get("id"):
|
||||
sub.stripe_subscription_id = subscription["id"]
|
||||
# Preserve plan so UI can show what ended; source stays stripe.
|
||||
if sub.source == UserSubscription.Source.NONE:
|
||||
sub.source = UserSubscription.Source.STRIPE
|
||||
sub.save()
|
||||
if prev_status != UserSubscription.Status.CANCELED:
|
||||
log_subscription_auth_event(
|
||||
user,
|
||||
started=False,
|
||||
detail=(
|
||||
f"plan={sub.plan.slug if sub.plan_id else 'none'} "
|
||||
f"source={sub.source} status={sub.status} "
|
||||
f"stripe_subscription_id={sub.stripe_subscription_id}"
|
||||
),
|
||||
)
|
||||
return sub
|
||||
|
||||
|
||||
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)
|
||||
if event_type == "customer.subscription.updated":
|
||||
return handle_customer_subscription_updated(data_object)
|
||||
if event_type == "customer.subscription.deleted":
|
||||
return handle_customer_subscription_deleted(data_object)
|
||||
|
||||
logger.info("Ignoring unhandled Stripe event type: %s", event_type)
|
||||
return None
|
||||
Reference in New Issue
Block a user