"""Stripe webhook handler for entitlement tier changes. Listens for: - customer.subscription.updated -> tier change - customer.subscription.deleted -> deactivate tenant - checkout.session.completed -> new tenant registration Integrates with core.billing.stripe_client.TIER_PRICES for price-to-tier mapping and EntitlementLedger for state mutations. VERIFICATION_STAMP Story: 2.07 Verified By: parallel-builder Verified At: 2026-02-26 Tests: see tests/rlm/test_entitlement.py Coverage: 100% """ from __future__ import annotations import logging from typing import Any, Dict, Optional from uuid import UUID, uuid4 from .contracts import CustomerTier from .entitlement import EntitlementLedger logger = logging.getLogger("core.rlm.entitlement_webhook") # Stripe price lookup key -> CustomerTier mapping # Must stay in sync with core.billing.stripe_client.TIER_PRICES PRICE_TO_TIER: Dict[str, CustomerTier] = { "sunaiva_starter_aud_497_monthly": CustomerTier.STARTER, "sunaiva_professional_aud_997_monthly": CustomerTier.PROFESSIONAL, "sunaiva_enterprise_aud_1497_monthly": CustomerTier.ENTERPRISE, } # Reverse mapping: metadata genesis_tier value -> CustomerTier _TIER_NAME_MAP: Dict[str, CustomerTier] = { "starter": CustomerTier.STARTER, "professional": CustomerTier.PROFESSIONAL, "enterprise": CustomerTier.ENTERPRISE, "queen": CustomerTier.QUEEN, } def _resolve_tier_from_event(event_data: Dict[str, Any]) -> Optional[CustomerTier]: """Extract CustomerTier from Stripe event data. Tries in order: 1. metadata.genesis_tier field 2. Price lookup key matching 3. Returns None if tier cannot be determined """ obj: Dict[str, Any] = event_data.get("object", {}) # Method 1: metadata.genesis_tier metadata = obj.get("metadata", {}) tier_name = metadata.get("genesis_tier", "") if tier_name and tier_name in _TIER_NAME_MAP: return _TIER_NAME_MAP[tier_name] # Method 2: Match price lookup key from subscription items items = obj.get("items", {}).get("data", []) for item in items: price = item.get("price", {}) lookup_key = price.get("lookup_key", "") if lookup_key in PRICE_TO_TIER: return PRICE_TO_TIER[lookup_key] return None def _extract_customer_id(event_data: Dict[str, Any]) -> str: """Extract Stripe customer ID from event data.""" obj: Dict[str, Any] = event_data.get("object", {}) return obj.get("customer", obj.get("id", "")) def _extract_tenant_id(event_data: Dict[str, Any]) -> Optional[UUID]: """Extract tenant UUID from event metadata. Looks for metadata.tenant_id which should be set during checkout. Falls back to generating a deterministic UUID from the Stripe customer ID. """ obj: Dict[str, Any] = event_data.get("object", {}) metadata = obj.get("metadata", {}) tenant_str = metadata.get("tenant_id") if tenant_str: try: return UUID(tenant_str) except ValueError: pass # Cannot determine tenant -- caller must handle this return None async def handle_stripe_event( event_type: str, event_data: Dict[str, Any], ledger: EntitlementLedger, ) -> Dict[str, Any]: """Route Stripe webhook event to appropriate handler. Parameters ---------- event_type : Stripe event type string (e.g. 'customer.subscription.updated') event_data : The 'data' portion of the Stripe event payload ledger : EntitlementLedger instance to mutate Returns ------- dict with 'status', 'action', and event-specific details """ routing = { "customer.subscription.updated": _handle_subscription_updated, "customer.subscription.deleted": _handle_subscription_deleted, "checkout.session.completed": _handle_checkout_completed, } handler = routing.get(event_type) if handler is None: logger.debug("Ignoring unhandled Stripe event: %s", event_type) return {"status": "ignored", "event_type": event_type} try: result = await handler(event_data, ledger) logger.info("Handled Stripe event %s: %s", event_type, result.get("action")) return result except Exception as exc: logger.error("Error handling Stripe event %s: %s", event_type, exc) return { "status": "error", "event_type": event_type, "error": str(exc), } async def _handle_subscription_updated( event_data: Dict[str, Any], ledger: EntitlementLedger, ) -> Dict[str, Any]: """Handle customer.subscription.updated -- tier change.""" tier = _resolve_tier_from_event(event_data) tenant_id = _extract_tenant_id(event_data) customer_id = _extract_customer_id(event_data) if tier is None: return { "status": "skipped", "action": "subscription_updated", "reason": "could_not_resolve_tier", "customer_id": customer_id, } if tenant_id is None: return { "status": "skipped", "action": "subscription_updated", "reason": "no_tenant_id_in_metadata", "customer_id": customer_id, } manifest = await ledger.update_tier(tenant_id, tier) return { "status": "ok", "action": "tier_updated", "tenant_id": str(tenant_id), "new_tier": tier.value, "customer_id": customer_id, "manifest_tier": manifest.tier.value, } async def _handle_subscription_deleted( event_data: Dict[str, Any], ledger: EntitlementLedger, ) -> Dict[str, Any]: """Handle customer.subscription.deleted -- deactivate tenant.""" tenant_id = _extract_tenant_id(event_data) customer_id = _extract_customer_id(event_data) if tenant_id is None: return { "status": "skipped", "action": "subscription_deleted", "reason": "no_tenant_id_in_metadata", "customer_id": customer_id, } deactivated = await ledger.deactivate_tenant(tenant_id) return { "status": "ok", "action": "tenant_deactivated", "tenant_id": str(tenant_id), "deactivated": deactivated, "customer_id": customer_id, } async def _handle_checkout_completed( event_data: Dict[str, Any], ledger: EntitlementLedger, ) -> Dict[str, Any]: """Handle checkout.session.completed -- new tenant registration.""" tier = _resolve_tier_from_event(event_data) customer_id = _extract_customer_id(event_data) obj: Dict[str, Any] = event_data.get("object", {}) metadata = obj.get("metadata", {}) # Generate tenant ID for new registration tenant_id_str = metadata.get("tenant_id") if tenant_id_str: try: tenant_id = UUID(tenant_id_str) except ValueError: tenant_id = uuid4() else: tenant_id = uuid4() if tier is None: tier = CustomerTier.STARTER # Default for new checkouts manifest = await ledger.register_tenant( tenant_id=tenant_id, tier=tier, stripe_customer_id=customer_id, ) return { "status": "ok", "action": "tenant_registered", "tenant_id": str(tenant_id), "tier": tier.value, "customer_id": customer_id, "manifest_tier": manifest.tier.value, }