"""Billing domain - business logic. Handles: 1. Customer creation/linking (user + workspace -> Stripe customer) 2. Subscription management (checkout, status tracking) 3. One-time usage credit purchases 4. Usage metering (record events, aggregate, report to Stripe) 5. Webhook processing (idempotent, event-type dispatched) """ from __future__ import annotations import structlog from datetime import datetime, timezone from typing import Any, Awaitable, Callable, Optional from uuid import UUID from src.core.exceptions import NotFoundError, ValidationError from src.domain.billing.constants import ENTERPRISE_PLAN_SLUG, FREE_PLAN_SLUG, STAFF_PLAN_SLUG from src.domain.billing.campaign_service import ( CampaignReservation, PlanOffer, PricingCampaignService, workspace_subject_key, ) from src.domain.billing.catalog_service import BillingCatalogService from src.domain.billing.interfaces import StripeClient from src.domain.billing.normalization import is_paid_billing_status from src.domain.billing.repository import ( InvoiceRepository, StripeCustomerRepository, StripeEventRepository, SubscriptionRepository, UsageCreditPurchaseRepository, UsageLedgerRepository, WorkspaceCreditLedgerRepository, WorkspaceUsageAllowanceRepository, ) from src.domain.workspaces.repository import WorkspaceRepository logger = structlog.get_logger() def _subscription_period_bounds(obj: dict, recurring_items: list[dict]) -> tuple[Any, Any]: """Read the subscription period from root fields or Clover item fields.""" period_start = obj.get("current_period_start") period_end = obj.get("current_period_end") if period_start is not None and period_end is not None: return period_start, period_end for item in recurring_items: if period_start is None: period_start = item.get("current_period_start") period_end = period_end if period_end is not None else item.get("current_period_end") if period_start is not None and period_end is not None: break return period_start, period_end def campaign_checkout_metadata(reservation: Optional[CampaignReservation]) -> dict: """Stripe session metadata that lets the webhook confirm a claimed campaign slot.""" if not reservation: return {} return { "campaign_slug": reservation.campaign_slug, "campaign_redemption_id": str(reservation.redemption_id), } def _campaign_idempotency_key( key: Optional[str], reservation: Optional[CampaignReservation] ) -> Optional[str]: """Keep a client's idempotency key from replaying a session created at another price.""" if key and reservation: return f"{key}:{reservation.stripe_price_id}" return key class BillingService: def __init__( self, customer_repo: StripeCustomerRepository, subscription_repo: SubscriptionRepository, invoice_repo: InvoiceRepository, event_repo: StripeEventRepository, usage_repo: UsageLedgerRepository, purchase_repo: UsageCreditPurchaseRepository, ledger_repo: WorkspaceCreditLedgerRepository, catalog: BillingCatalogService, stripe_client: Optional[StripeClient] = None, enqueue_billing_email: Optional[Callable[..., Awaitable[bool]]] = None, workspace_repo: Optional[WorkspaceRepository] = None, enqueue_pre_registration: Optional[Callable[..., Awaitable[bool]]] = None, allowance_repo: Optional[WorkspaceUsageAllowanceRepository] = None, campaign_service: Optional[PricingCampaignService] = None, ): self._customer_repo = customer_repo self._subscription_repo = subscription_repo self._invoice_repo = invoice_repo self._event_repo = event_repo self._usage_repo = usage_repo self._purchase_repo = purchase_repo self._ledger_repo = ledger_repo self._catalog = catalog self._stripe = stripe_client self._enqueue = enqueue_billing_email self._workspace_repo = workspace_repo self._enqueue_pre_reg = enqueue_pre_registration self._allowance_repo = allowance_repo self._campaigns = campaign_service async def reserve_campaign_price( self, plan, interval: str, subject_key: str, *, workspace_id: Optional[UUID] = None, ) -> Optional[CampaignReservation]: """Claim a limited-slot campaign price for a new subscription checkout, if one applies.""" if not self._campaigns: return None return await self._campaigns.reserve( plan=plan, interval=interval, subject_key=subject_key, workspace_id=workspace_id ) async def get_plan_offers(self, workspace_id: UUID, plans: list) -> dict[UUID, PlanOffer]: """Campaign prices to display for a workspace's self-serve plan list. Only shown where checkout would actually apply them: a workspace that already has a paid subscription changes plans through the portal at catalog prices. """ if not self._campaigns: return {} existing = await self._subscription_repo.get_latest_for_workspace(workspace_id) if ( existing and existing.stripe_subscription_id and existing.status not in ("canceled", "incomplete_expired") ): return {} return await self._campaigns.offers_for_plans( plans, subject_key=workspace_subject_key(workspace_id) ) async def ensure_customer(self, workspace_id: UUID, user_id: UUID, email: str, workspace_name: str) -> str: """Ensure workspace has a linked Stripe customer. Creates one if not.""" existing = await self._customer_repo.get_by_workspace(workspace_id) if existing: return existing.stripe_customer_id if not self._stripe: raise NotFoundError("Stripe client not configured") stripe_id = await self._stripe.create_customer(email, workspace_name) await self._customer_repo.create(workspace_id, stripe_id) logger.info("billing_customer_created", workspace_id=str(workspace_id), stripe_id=stripe_id) return stripe_id async def start_subscription_checkout( self, workspace_id: UUID, user_id: UUID, email: str, workspace_name: str, plan_slug: str, interval: str, success_url: str, cancel_url: str, idempotency_key: Optional[str] = None, ) -> dict: """Create Stripe checkout for subscription - driven by internal plan_slug.""" plan = await self._catalog.get_plan_by_slug(plan_slug) if plan.slug.lower() == STAFF_PLAN_SLUG: raise ValidationError("Staff plans cannot be selected through self-service billing") if plan.slug == ENTERPRISE_PLAN_SLUG: raise ValidationError( "Enterprise plans are custom per customer; use contact sales, not self-serve checkout." ) existing = await self._subscription_repo.get_latest_for_workspace(workspace_id) if ( existing and existing.stripe_subscription_id and existing.status not in ("canceled", "incomplete_expired") ): if plan.slug.lower() == FREE_PLAN_SLUG: raise ValidationError( "Cancel your paid subscription in the billing portal before selecting Free" ) if not self._stripe: raise NotFoundError("Stripe client not configured") price_id = self._catalog.resolve_stripe_price_id_for_plan(plan, interval) customer = await self._customer_repo.get_by_workspace(workspace_id) if not customer: raise NotFoundError("No billing customer found") url = await self._stripe.create_subscription_update_portal( customer.stripe_customer_id, existing.stripe_subscription_id, price_id, cancel_url, ) return {"checkout_url": url, "workspace_id": workspace_id, "free_plan": False} if plan.slug.lower() == FREE_PLAN_SLUG: if self._workspace_repo: await self._workspace_repo.update( workspace_id, selected_plan_slug=plan.slug.lower(), onboarding_status="plan_selected", ) await self._workspace_repo.clear_billing_checkout_tracking(workspace_id) # Create an internal subscription record so subscription_id is never null existing_sub = await self._subscription_repo.get_latest_for_workspace(workspace_id) if not existing_sub or existing_sub.status != "complimentary": sub = await self._subscription_repo.create_for_no_charge_plan(workspace_id, plan.slug.lower()) else: sub = existing_sub logger.info( "no_charge_plan_selected_no_checkout", workspace_id=str(workspace_id), user_id=str(user_id), plan_slug=plan.slug.lower(), subscription_id=str(sub.id), ) return { "checkout_url": None, "session_id": None, "workspace_id": workspace_id, "free_plan": True, "subscription_id": str(sub.id), } price_id = self._catalog.resolve_stripe_price_id_for_plan(plan, interval) customer_id = await self.ensure_customer(workspace_id, user_id, email, workspace_name) if not self._stripe: raise NotFoundError("Stripe client not configured") reservation = await self.reserve_campaign_price( plan, interval, workspace_subject_key(workspace_id), workspace_id=workspace_id ) if reservation: price_id = reservation.stripe_price_id metadata = { "workspace_id": str(workspace_id), "user_id": str(user_id), "plan_slug": plan_slug, "interval": interval, "purchase_type": "subscription", "onboarding_intent": "workspace_subscription", **campaign_checkout_metadata(reservation), } url, session_id = await self._stripe.create_subscription_checkout( customer_id, price_id, success_url, cancel_url, metadata, _campaign_idempotency_key(idempotency_key, reservation), ) if reservation and self._campaigns: await self._campaigns.attach_session(reservation.redemption_id, session_id) if self._workspace_repo: await self._workspace_repo.update_billing_checkout_tracking( workspace_id, session_id=session_id, checkout_at=datetime.now(timezone.utc), checkout_return="", ) logger.info( "checkout_session_created", workspace_id=str(workspace_id), user_id=str(user_id), plan_slug=plan_slug, interval=interval, stripe_session_id=session_id, ) return { "checkout_url": url, "session_id": session_id, "workspace_id": workspace_id, "free_plan": False, } async def start_usage_topup_checkout( self, workspace_id: UUID, user_id: UUID, email: str, workspace_name: str, package_slug: str, success_url: str, cancel_url: str, idempotency_key: Optional[str] = None, ) -> dict: """Create Stripe checkout for one-time usage credit purchase.""" pkg = await self._catalog.get_package_by_slug(package_slug) if not pkg.stripe_price_id: raise NotFoundError(f"Usage package '{package_slug}' not synced to Stripe") customer_id = await self.ensure_customer(workspace_id, user_id, email, workspace_name) if not self._stripe: raise NotFoundError("Stripe client not configured") metadata = { "workspace_id": str(workspace_id), "package_slug": package_slug, "purchase_type": "usage_topup", } url, session_id = await self._stripe.create_payment_checkout( customer_id, pkg.stripe_price_id, success_url, cancel_url, metadata, idempotency_key ) return {"checkout_url": url, "session_id": session_id, "workspace_id": workspace_id} async def create_customer_portal_session( self, workspace_id: UUID, return_url: Optional[str] = None ) -> str: customer = await self._customer_repo.get_by_workspace(workspace_id) if not customer or not self._stripe: raise NotFoundError("No billing customer found") return await self._stripe.create_customer_portal( customer.stripe_customer_id, return_url=return_url ) async def get_subscription(self, workspace_id: UUID): sub = await self._subscription_repo.get_by_workspace(workspace_id) if not sub: raise NotFoundError("No active subscription") return sub async def get_subscription_optional(self, workspace_id: UUID): """Return subscription or None if workspace has none. For plans UI context.""" return await self._subscription_repo.get_by_workspace(workspace_id) async def list_invoices(self, workspace_id: UUID): return await self._invoice_repo.list_for_workspace(workspace_id) async def record_usage( self, workspace_id: UUID, event_type: str, quantity: int = 1, unit_label: Optional[str] = None, project_id: Optional[UUID] = None, metadata: Optional[dict] = None, ): return await self._usage_repo.record( workspace_id, event_type, quantity, unit_label, project_id, metadata, ) async def get_usage_summary(self, workspace_id: UUID): return await self._usage_repo.summarize(workspace_id) async def report_usage_to_stripe(self, workspace_id: UUID) -> int: """Push unreported usage to Stripe via Meter Events. Returns count reported.""" from src.domain.billing.stripe_sync_service import _meter_event_name sub = await self._subscription_repo.get_by_workspace(workspace_id) if not sub or not self._stripe: return 0 unreported = await self._usage_repo.get_unreported(workspace_id) if not unreported: return 0 # Resolve event_name from the plan (catalog already loaded in service). event_name: Optional[str] = None if sub.plan_slug: try: plan = await self._catalog.get_plan_by_slug(sub.plan_slug) if plan.includes_metered_billing: event_name = _meter_event_name(plan.slug) except Exception: pass if not event_name or not sub.stripe_customer_id: logger.warning( "billing_usage_report_skipped", workspace_id=str(workspace_id), reason="no_meter_event_name_or_customer", ) return 0 total = sum(e.quantity for e in unreported) await self._stripe.report_usage(event_name, sub.stripe_customer_id, total) await self._usage_repo.mark_reported([e.id for e in unreported]) logger.info("billing_usage_reported", workspace_id=str(workspace_id), quantity=total, event_name=event_name) return len(unreported) async def build_checkout_status_payload( self, workspace_id: UUID, billing_last_checkout_return: Optional[str], query_checkout_return: Optional[str], ) -> dict: """Frontend polling after Stripe redirect — DB-backed, never trusts redirect alone.""" from src.domain.billing.enums import CheckoutReturnState from src.domain.billing.normalization import subscription_row_to_checkout_poll_status sub = await self._subscription_repo.get_latest_for_workspace(workspace_id) api_status, is_paid, has_sub = subscription_row_to_checkout_poll_status( sub=sub, billing_last_checkout_return=billing_last_checkout_return, checkout_return_query=query_checkout_return, ) if query_checkout_return == CheckoutReturnState.SUCCESS_RETURN.value: last_st = CheckoutReturnState.SUCCESS_RETURN.value elif query_checkout_return == CheckoutReturnState.CANCELED_RETURN.value: last_st = CheckoutReturnState.CANCELED_RETURN.value elif billing_last_checkout_return in ( CheckoutReturnState.SUCCESS_RETURN.value, CheckoutReturnState.CANCELED_RETURN.value, ): last_st = billing_last_checkout_return else: last_st = CheckoutReturnState.UNKNOWN.value plan_slug = getattr(sub, "plan_slug", None) if sub else None period_end = getattr(sub, "current_period_end", None) if sub else None return { "workspace_id": workspace_id, "has_subscription": has_sub, "status": api_status, "is_paid": is_paid, "requires_payment": not is_paid, "plan_slug": plan_slug, "current_period_end": period_end, "last_checkout_state": last_st, } async def handle_webhook_event(self, event_id: str, event_type: str, data: dict) -> dict: """Process Stripe webhook event (idempotent).""" obj = data.get("object") obj_id = obj.get("id", "") if isinstance(obj, dict) else "" stage = "deduplicate" try: if await self._event_repo.already_processed(event_id): logger.info( "stripe_webhook_duplicate", stripe_event_id=event_id, event_type=event_type, object_id=obj_id, ) return {"processed": False, "reason": "duplicate"} stage = "record" await self._event_repo.record(event_id, event_type, data) stage = "dispatch" handler = STRIPE_EVENT_HANDLERS.get(event_type) if handler: await handler(self, data, event_type) logger.info( "stripe_webhook_processed", stripe_event_id=event_id, event_type=event_type, object_id=obj_id, success=True, ) else: logger.info( "stripe_webhook_noop", stripe_event_id=event_id, event_type=event_type, object_id=obj_id, reason="unhandled_type", ) return {"processed": True, "event_type": event_type} except Exception as exc: logger.error( "stripe_webhook_error", stripe_event_id=event_id, event_type=event_type, object_id=obj_id, stage=stage, error=str(exc), exc_info=True, ) raise async def _sync_workspace_selected_plan_from_subscription( service: BillingService, *, workspace_id: UUID, sub, ) -> None: """Keep workspace.selected_plan_slug aligned with the latest resolved subscription plan.""" if ( service._allowance_repo and sub and sub.plan_slug and is_paid_billing_status(sub.status) ): plan = await service._catalog.get_plan_by_slug(sub.plan_slug) limits = plan.limits or {} included = limits.get("included_ai_units", limits.get("ai_minutes", 0)) or 0 await service._ledger_repo.acquire_consumption_lock(workspace_id) await service._allowance_repo.reconcile_plan_limit(workspace_id, included) if not service._workspace_repo or not sub: return plan_slug = (getattr(sub, "plan_slug", None) or "").strip().lower() if not plan_slug: return ws = await service._workspace_repo.get_by_id(workspace_id) if not ws: return current_plan_slug = ((getattr(ws, "selected_plan_slug", None) or "").strip().lower()) or None if current_plan_slug == plan_slug: return await service._workspace_repo.update( workspace_id, selected_plan_slug=plan_slug, ) logger.info( "workspace_selected_plan_synced_from_subscription", workspace_id=str(workspace_id), selected_plan_slug=plan_slug, ) async def _handle_checkout_completed(service: BillingService, data: dict, event_type: str = ""): """checkout.session.completed - subscription or one-time usage top-up.""" from datetime import datetime, timezone from uuid import UUID from sqlalchemy.exc import IntegrityError obj = data.get("object", data) metadata = dict(obj.get("metadata") or {}) purchase_type = metadata.get("purchase_type") if purchase_type == "domain_registration": from uuid import UUID from src.domain.domains.billing import fulfill_domain_registration order_id = metadata.get("domain_order_id") if order_id: from src.domain.domains.models import DomainOrder from sqlalchemy import select order = await service._event_repo._db.scalar(select(DomainOrder).where(DomainOrder.id == UUID(order_id))) if order: order.stripe_subscription_id = obj.get("subscription") order.stripe_checkout_session_id = obj.get("id") if obj.get("payment_status") == "paid": await fulfill_domain_registration(service._event_repo._db, UUID(order_id)) return if purchase_type == "agent_credit_purchase": # Agent purchases reconcile by retrieving Stripe objects in a durable worker. # This separate fulfillment path must never issue a second ledger credit. return workspace_id_str = metadata.get("workspace_id") subscription_id = obj.get("subscription") customer_id = obj.get("customer") await _confirm_campaign_slot(service, obj, metadata, subscription_id) customer = await service._customer_repo.get_by_stripe_id(customer_id) if not customer: # Check if this is a pre-registration checkout (from the home site signup flow) pending_registration_id = metadata.get("pending_registration_id") if pending_registration_id and service._enqueue_pre_reg: await service._enqueue_pre_reg( pending_registration_id=pending_registration_id, stripe_customer_id=customer_id, stripe_subscription_id=subscription_id, ) logger.info( "pre_registration_checkout_completed", pending_registration_id=pending_registration_id, stripe_customer_id=customer_id, ) return if purchase_type == "cloud_credit_topup" and workspace_id_str: if obj.get("payment_status") != "paid": return from src.domain.cloud_credits.service import CloudCreditService from src.domain.hosting.public_serving_gate import HostingPublicServingGate session_id = obj.get("id") package_slug = metadata.get("package_slug") if not session_id or not package_slug: logger.warning("cloud_credit_checkout_missing_fields", session_id=session_id) return cloud = CloudCreditService( db=service._event_repo._db, stripe_client=service._stripe, serving_gate=HostingPublicServingGate(), ) await cloud.fulfill_paid_session( workspace_id=UUID(workspace_id_str), stripe_customer_id=customer_id, session_id=session_id, payment_intent_id=obj.get("payment_intent"), package_slug=package_slug, ) return if purchase_type == "runtime_ai_topup" and workspace_id_str: if obj.get("payment_status") != "paid": return from src.domain.runtime_ai.service import RuntimeAiService session_id = obj.get("id") package_slug = metadata.get("package_slug") if not session_id or not package_slug: logger.warning("runtime_ai_checkout_missing_fields", session_id=session_id) return runtime = RuntimeAiService(db=service._event_repo._db, stripe_client=service._stripe) await runtime.fulfill_paid_session( workspace_id=UUID(workspace_id_str), stripe_customer_id=customer_id, session_id=session_id, payment_intent_id=obj.get("payment_intent"), package_slug=package_slug, ) return if purchase_type == "usage_topup" and workspace_id_str: # --- Top-up fulfillment state machine (exactly-once at business level) --- # States: not_fulfilled -> (create purchase + ledger + enqueue) -> fulfilled # Guards: (1) get_by_stripe_session_id before create; (2) unique on session_id; (3) savepoint+IntegrityError # Invariants: never two purchases, never two ledger credits, never two notifications per recipient # Enqueue policy: log on failure, do NOT fail webhook; reconciliation can retry via scripts session_id = obj.get("id") if obj.get("payment_status") == "paid": # Idempotency: first check by stripe_checkout_session_id existing = await service._purchase_repo.get_by_stripe_session_id(session_id) if existing and existing.status == "paid": logger.info( "usage_topup_already_fulfilled", session_id=session_id, workspace_id=workspace_id_str, ) return package_slug = metadata.get("package_slug") if not package_slug: logger.warning( "checkout_topup_no_package_slug", session_id=session_id, workspace_id=workspace_id_str, ) return try: pkg = await service._catalog.get_package_by_slug(package_slug) except NotFoundError: logger.warning( "checkout_topup_package_not_found", slug=package_slug, session_id=session_id, ) return workspace_id = UUID(workspace_id_str) db = service._purchase_repo._db purchase = None try: with db.begin_nested(): purchase = await service._purchase_repo.create( workspace_id=workspace_id, stripe_customer_id=customer_id, stripe_checkout_session_id=session_id, stripe_payment_intent_id=obj.get("payment_intent"), usage_credit_package_id=pkg.id, package_slug=pkg.slug, units_purchased=pkg.units, amount_minor=pkg.price_minor, currency=pkg.currency, status="paid", paid_at=datetime.now(timezone.utc), ) except IntegrityError: # Race: another webhook already created; unique on session_id. Savepoint rolled back. existing = await service._purchase_repo.get_by_stripe_session_id(session_id) if existing and existing.status == "paid": logger.info( "usage_topup_fulfilled_race_resolved", session_id=session_id, workspace_id=workspace_id_str, existing_purchase_id=str(existing.id), ) return raise # Add ledger credit entry (only when we created purchase) balance_before = await service._ledger_repo.sum_balance(workspace_id) balance_after = balance_before + pkg.units await service._ledger_repo.add_entry( workspace_id=workspace_id, entry_type="purchase", units_delta=pkg.units, balance_after=balance_after, source_bucket="purchased_topup", source_type="stripe_checkout", source_id=obj.get("id"), usage_credit_purchase_id=purchase.id, description=f"Purchased {pkg.units} units ({pkg.name})", ) if service._enqueue: await service._enqueue( "send_usage_topup_confirmation_email", workspace_id=workspace_id_str, stripe_checkout_session_id=session_id, package_name=pkg.name, units=pkg.units, amount_minor=pkg.price_minor, currency=pkg.currency, ) logger.info( "usage_topup_fulfilled", workspace_id=workspace_id_str, session_id=session_id, units=pkg.units, ) return # Subscription checkout if not subscription_id or not service._stripe: return sub_data = await service._stripe.get_subscription(subscription_id) # Resolve plan_slug from stripe_price_id plan_slug = metadata.get("plan_slug") if not plan_slug and sub_data.get("plan_id"): try: plan = await service._catalog.get_plan_by_stripe_price_id(sub_data["plan_id"]) plan_slug = plan.slug except NotFoundError: pass if plan_slug: sub_data["plan_slug"] = plan_slug if sub_data.get("stripe_product_id"): sub_data["stripe_product_id"] = sub_data["stripe_product_id"] sub_data["stripe_price_id"] = sub_data.get("plan_id") sub_data["cancel_at_period_end"] = sub_data.get("cancel_at_period_end", False) await service._subscription_repo.upsert_from_stripe(customer.workspace_id, sub_data) synced = await service._subscription_repo.get_by_stripe_id(subscription_id) if synced: await _sync_workspace_selected_plan_from_subscription( service, workspace_id=customer.workspace_id, sub=synced, ) if synced and is_paid_billing_status(synced.status) and service._workspace_repo: await service._workspace_repo.clear_billing_checkout_return(customer.workspace_id) async def _confirm_campaign_slot( service: BillingService, obj: dict, metadata: dict, subscription_id: Optional[str] ) -> None: """Make a founding/campaign slot permanent once its subscription checkout is paid.""" redemption_raw = metadata.get("campaign_redemption_id") if not redemption_raw or not service._campaigns: return if obj.get("payment_status") not in ("paid", "no_payment_required"): return try: redemption_id = UUID(redemption_raw) except ValueError: logger.warning("campaign_redemption_id_invalid", value=redemption_raw) return workspace_raw = metadata.get("workspace_id") await service._campaigns.confirm( redemption_id=redemption_id, session_id=obj.get("id"), subscription_id=subscription_id, workspace_id=UUID(workspace_raw) if workspace_raw else None, ) async def _handle_checkout_expired(service: BillingService, data: dict, event_type: str = ""): """checkout.session.expired - free the campaign slot an abandoned checkout was holding.""" obj = data.get("object", data) session_id = obj.get("id") if service._campaigns and session_id: await service._campaigns.release_session(session_id) async def _handle_subscription_updated(service: BillingService, data: dict, event_type: str = ""): """customer.subscription.updated/created - sync subscription state with plan_slug resolution.""" obj = data.get("object", data) if (obj.get("metadata") or {}).get("purchase_type") == "domain_registration": return sub_id = obj.get("id", "") customer_id = obj.get("customer", "") customer = await service._customer_repo.get_by_stripe_id(customer_id) if not customer: return items = obj.get("items", {}).get("data", []) recurring_items = [i for i in items if i.get("price", {}).get("recurring")] period_start, period_end = _subscription_period_bounds(obj, recurring_items) plan_price = recurring_items[0]["price"] if recurring_items else None plan_id = plan_price["id"] if plan_price else None stripe_product_id = plan_price.get("product") if plan_price else None interval = plan_price.get("recurring", {}).get("interval") if plan_price else None metered_item = next( ( item["id"] for item in items if item.get("price", {}).get("recurring", {}).get("usage_type") == "metered" ), None, ) plan_slug = None if plan_id: try: plan = await service._catalog.get_plan_by_stripe_price_id(plan_id) plan_slug = plan.slug except NotFoundError: pass await service._subscription_repo.upsert_from_stripe( customer.workspace_id, { "id": sub_id, "customer": customer_id, "status": obj.get("status"), "plan_id": plan_id, "stripe_price_id": plan_id, "stripe_product_id": stripe_product_id, "billing_interval": interval, "current_period_start": period_start, "current_period_end": period_end, "cancel_at_period_end": obj.get("cancel_at_period_end", False), "canceled_at": obj.get("canceled_at"), "trial_start": obj.get("trial_start"), "trial_end": obj.get("trial_end"), "metered_item_id": metered_item, "plan_slug": plan_slug, }, ) synced = await service._subscription_repo.get_by_stripe_id(sub_id) if synced: await _sync_workspace_selected_plan_from_subscription( service, workspace_id=customer.workspace_id, sub=synced, ) if synced and is_paid_billing_status(synced.status) and service._workspace_repo: await service._workspace_repo.clear_billing_checkout_return(customer.workspace_id) logger.info( "subscription_state_transition", workspace_id=str(customer.workspace_id), stripe_subscription_id=sub_id, status=obj.get("status"), event_type=event_type, ) async def _handle_subscription_deleted(service: BillingService, data: dict, event_type: str = ""): """customer.subscription.deleted - mark canceled, enqueue notification.""" from datetime import datetime, timezone obj = data.get("object", data) sub_id = obj.get("id", "") if (obj.get("metadata") or {}).get("purchase_type") == "domain_registration": from src.domain.domains.models import DomainOrder from sqlalchemy import select order = await service._event_repo._db.scalar(select(DomainOrder).where(DomainOrder.stripe_subscription_id == sub_id)) if order: order.payment_status = "canceled" order.fulfillment_status = "canceled" return existing = await service._subscription_repo.get_by_stripe_id(sub_id) if existing: existing.status = "canceled" existing.canceled_at = datetime.now(timezone.utc) if service._enqueue: period_end = existing.current_period_end period_end_iso = ( period_end.isoformat() if hasattr(period_end, "isoformat") else str(period_end) ) if period_end else "" await service._enqueue( "send_subscription_canceled_email", workspace_id=str(existing.workspace_id), stripe_subscription_id=sub_id, plan_slug=existing.plan_slug or "unknown", period_end_iso=period_end_iso, ) async def _handle_invoice_event(service: BillingService, data: dict, event_type: str = ""): """invoice.paid / invoice.payment_failed - sync invoice, enqueue emails.""" obj = data.get("object", data) customer_id = obj.get("customer", "") if obj.get("subscription"): from src.domain.domains.models import DomainOrder from sqlalchemy import select order = await service._event_repo._db.scalar(select(DomainOrder).where(DomainOrder.stripe_subscription_id == obj.get("subscription"))) if order: if event_type == "invoice.paid": from src.domain.domains.billing import fulfill_domain_registration order.payment_status = "paid" order.renewal_failure_count = 0 await fulfill_domain_registration(service._event_repo._db, order.id) if order.domain_id: from src.domain.domains.models import WorkspaceDomain domain = await service._event_repo._db.scalar(select(WorkspaceDomain).where(WorkspaceDomain.id == order.domain_id)) if domain: domain.renewal_authorized = True elif event_type == "invoice.payment_failed": order.renewal_failure_count = (order.renewal_failure_count or 0) + 1 order.payment_status = "past_due" # Stripe performs its configured progressive retries. After three # failures Core withdraws registrar renewal authorization; the # scheduled worker disables Cloudflare auto-renew in the lead window. if order.renewal_failure_count >= 3 and order.domain_id: from src.domain.domains.models import WorkspaceDomain domain = await service._event_repo._db.scalar(select(WorkspaceDomain).where(WorkspaceDomain.id == order.domain_id)) if domain: domain.renewal_authorized = False customer = await service._customer_repo.get_by_stripe_id(customer_id) if not customer: return await service._invoice_repo.upsert_from_stripe(customer.workspace_id, { "id": obj.get("id"), "customer": customer_id, "status": obj.get("status"), "amount_due": obj.get("amount_due", 0), "amount_paid": obj.get("amount_paid", 0), "currency": obj.get("currency", "usd"), "hosted_invoice_url": obj.get("hosted_invoice_url"), "invoice_pdf": obj.get("invoice_pdf"), "period_start": obj.get("period_start"), "period_end": obj.get("period_end"), }) if service._enqueue: ws_id = str(customer.workspace_id) inv_id = obj.get("id", "") inv_url = obj.get("hosted_invoice_url") or obj.get("invoice_pdf") if event_type == "invoice.paid": await service._enqueue( "send_invoice_paid_email", workspace_id=ws_id, stripe_invoice_id=inv_id, invoice_url=inv_url, amount_paid_minor=obj.get("amount_paid", 0), currency=obj.get("currency", "usd"), ) elif event_type == "invoice.payment_failed": await service._enqueue( "send_payment_failed_email", workspace_id=ws_id, stripe_invoice_id=inv_id, invoice_url=inv_url, ) STRIPE_EVENT_HANDLERS = { "checkout.session.completed": _handle_checkout_completed, "checkout.session.expired": _handle_checkout_expired, "customer.subscription.created": _handle_subscription_updated, "customer.subscription.updated": _handle_subscription_updated, "customer.subscription.deleted": _handle_subscription_deleted, "invoice.paid": _handle_invoice_event, "invoice.payment_failed": _handle_invoice_event, }