"""Stripe sync services - DB to Stripe reconciliation for plans and usage packages.""" from __future__ import annotations import json from typing import Any, Optional import structlog from src.domain.admin.models import SubscriptionPlan from src.domain.admin.repository import PlanPriceVersionRepository, SubscriptionPlanRepository from src.domain.billing.interfaces import StripeSyncCapable from src.domain.billing.models import UsageCreditPackage from src.domain.billing.repository import UsageCreditPackageRepository logger = structlog.get_logger() _STRIPE_METADATA_VALUE_MAX = 500 def _meter_event_name(plan_slug: str) -> str: """Deterministic Stripe Meter event name derived from plan slug (alphanumeric + underscores).""" import re sanitized = re.sub(r"[^a-z0-9_]", "_", plan_slug.lower()) return f"{sanitized}_usage" def _metadata_json(value: Any) -> Optional[str]: if value is None: return None try: s = json.dumps(value, separators=(",", ":"), default=str) except (TypeError, ValueError): return None if len(s) <= _STRIPE_METADATA_VALUE_MAX: return s logger.warning("stripe_metadata_value_too_long", length=len(s), max=_STRIPE_METADATA_VALUE_MAX) return None class PlanStripeSyncService: """ Syncs subscription plans from DB to Stripe. - Creates Stripe product/prices if missing; updates DB with Stripe ids. - When DB amounts differ from existing Stripe Price objects, creates new prices, deactivates old ones, and records plan_price_versions (when repo is provided). - Pushes non-price fields (name, description, active, metadata for features/limits) to the Stripe Product without creating new prices. """ def __init__( self, plan_repo: SubscriptionPlanRepository, stripe_client: Optional[StripeSyncCapable], price_version_repo: Optional[PlanPriceVersionRepository] = None, ): self._plan_repo = plan_repo self._stripe = stripe_client self._price_version_repo = price_version_repo async def sync_plan(self, plan: SubscriptionPlan) -> SubscriptionPlan: if not self._stripe: logger.warning("plan_sync_skipped_no_stripe", slug=plan.slug) return plan updates: dict[str, Any] = {} if not plan.stripe_product_id: product = await self._stripe.create_product( name=plan.name, description=plan.description or "", metadata=self._plan_product_metadata(plan), ) updates["stripe_product_id"] = product["id"] logger.info("plan_synced_product_created", slug=plan.slug, product_id=product["id"]) product_id = plan.stripe_product_id or updates.get("stripe_product_id") if not product_id: return plan await self._push_plan_product_fields(plan, product_id) if updates: await self._plan_repo.update(plan.id, **updates) for k, v in updates.items(): setattr(plan, k, v) # Recurring prices (amount/currency drift vs Stripe) await self._sync_recurring_price( plan, product_id, amount=plan.price_monthly, stripe_price_id_attr="stripe_monthly_price_id", interval="month", nickname_suffix="Monthly", ) if plan.price_yearly and plan.price_yearly > 0: await self._sync_recurring_price( plan, product_id, amount=plan.price_yearly, stripe_price_id_attr="stripe_yearly_price_id", interval="year", nickname_suffix="Yearly", ) if plan.includes_metered_billing and not plan.metered_stripe_price_id: # Stripe API 2025-03-31.basil+: metered prices require a Meter object first. meter_id = getattr(plan, "metered_stripe_meter_id", None) if not meter_id: event_name = _meter_event_name(plan.slug) meter = await self._stripe.create_meter( event_name=event_name, display_name=f"{plan.name} Usage", ) meter_id = meter["id"] await self._plan_repo.update(plan.id, metered_stripe_meter_id=meter_id) plan.metered_stripe_meter_id = meter_id logger.info("plan_meter_created", slug=plan.slug, meter_id=meter_id, event_name=event_name) price = await self._stripe.create_metered_price( product_id=product_id, meter_id=meter_id, currency=plan.currency, nickname=f"{plan.name} - Usage", ) await self._plan_repo.update(plan.id, metered_stripe_price_id=price["id"]) plan.metered_stripe_price_id = price["id"] return plan def _plan_product_metadata(self, plan: SubscriptionPlan) -> dict[str, str]: meta: dict[str, str] = {"p2hs_slug": plan.slug} feat = _metadata_json(plan.features) if feat: meta["p2hs_features"] = feat lim = _metadata_json(plan.limits) if lim: meta["p2hs_limits"] = lim return meta async def _push_plan_product_fields(self, plan: SubscriptionPlan, product_id: str) -> None: """Update Stripe Product from DB (no new Price objects).""" if not self._stripe: return await self._stripe.update_product( product_id, name=plan.name, description=plan.description or "", metadata=self._plan_product_metadata(plan), active=plan.is_active, ) logger.info("plan_product_fields_pushed", slug=plan.slug, product_id=product_id) async def _sync_recurring_price( self, plan: SubscriptionPlan, product_id: str, amount: int, stripe_price_id_attr: str, interval: str, nickname_suffix: str, ) -> None: if not self._stripe: return current_id: Optional[str] = getattr(plan, stripe_price_id_attr) if amount <= 0: return need_new = False if not current_id: need_new = True else: remote = await self._stripe.retrieve_price(current_id) if not remote: need_new = True elif remote.get("unit_amount") != amount: need_new = True elif remote.get("currency", "").lower() != plan.currency.lower(): need_new = True else: rec = remote.get("recurring") or {} if rec.get("usage_type") == "metered": need_new = True elif rec.get("interval") != interval: need_new = True if not need_new: return new_price = await self._stripe.create_price( product_id=product_id, unit_amount=amount, currency=plan.currency, interval=interval, nickname=f"{plan.name} - {nickname_suffix}", ) old_id = current_id await self._plan_repo.update(plan.id, **{stripe_price_id_attr: new_price["id"]}) setattr(plan, stripe_price_id_attr, new_price["id"]) if old_id: await self._stripe.deactivate_price(old_id) await self._record_price_version(plan, interval, new_price["id"], amount, product_id) logger.info( "plan_recurring_price_rotated", slug=plan.slug, interval=interval, old_price_id=old_id, new_price_id=new_price["id"], ) async def _record_price_version( self, plan: SubscriptionPlan, interval: str, stripe_price_id: str, price_minor: int, stripe_product_id: str, ) -> None: if not self._price_version_repo: return v = await self._price_version_repo.create( subscription_plan_id=plan.id, interval=interval, stripe_price_id=stripe_price_id, price_minor=price_minor, currency=plan.currency, stripe_product_id=stripe_product_id, ) await self._price_version_repo.set_current_exclusive(plan.id, interval, v.id) field = { "month": "current_monthly_price_version_id", "year": "current_yearly_price_version_id", }.get(interval) if field: await self._plan_repo.update(plan.id, **{field: v.id}) setattr(plan, field, v.id) async def sync_all_plans(self, active_only: bool = True) -> list[SubscriptionPlan]: plans = await self._plan_repo.list_all(active_only=active_only) for plan in plans: await self.sync_plan(plan) return plans class UsagePackageStripeSyncService: """ Syncs usage credit packages from DB to Stripe. Creates Stripe product + one-time price if missing. Rotates one-time price when DB amount differs from Stripe. Updates product fields from DB without creating new prices when only metadata changes. """ def __init__( self, package_repo: UsageCreditPackageRepository, stripe_client: Optional[StripeSyncCapable], ): self._package_repo = package_repo self._stripe = stripe_client def _package_metadata(self, pkg: UsageCreditPackage) -> dict[str, str]: meta = {"p2hs_slug": pkg.slug, "p2hs_units": str(pkg.units)} extra = _metadata_json(pkg.metadata_json) if extra: meta["p2hs_metadata"] = extra return meta async def sync_package(self, pkg: UsageCreditPackage) -> UsageCreditPackage: if not self._stripe: logger.warning("usage_package_sync_skipped_no_stripe", slug=pkg.slug) return pkg updates: dict[str, Any] = {} if not pkg.stripe_product_id: product = await self._stripe.create_product( name=pkg.name, description=pkg.description or "", metadata=self._package_metadata(pkg), ) updates["stripe_product_id"] = product["id"] logger.info("usage_package_synced_product_created", slug=pkg.slug, product_id=product["id"]) product_id = pkg.stripe_product_id or updates.get("stripe_product_id") if not product_id: return pkg await self._stripe.update_product( product_id, name=pkg.name, description=pkg.description or "", metadata=self._package_metadata(pkg), active=pkg.is_active, ) if updates: await self._package_repo.update(pkg.id, **updates) for k, v in updates.items(): setattr(pkg, k, v) need_new = False if not pkg.stripe_price_id: need_new = True else: remote = await self._stripe.retrieve_price(pkg.stripe_price_id) if not remote: need_new = True elif remote.get("unit_amount") != pkg.price_minor: need_new = True elif remote.get("currency", "").lower() != pkg.currency.lower(): need_new = True elif remote.get("recurring"): need_new = True if not need_new: return pkg new_price = await self._stripe.create_one_time_price( product_id=product_id, unit_amount=pkg.price_minor, currency=pkg.currency, nickname=f"{pkg.name} - {pkg.units} units", ) old_id = pkg.stripe_price_id await self._package_repo.update(pkg.id, stripe_price_id=new_price["id"]) pkg.stripe_price_id = new_price["id"] if old_id: await self._stripe.deactivate_price(old_id) logger.info( "usage_package_price_rotated", slug=pkg.slug, old_price_id=old_id, new_price_id=new_price["id"], ) return pkg async def sync_all_packages(self, active_only: bool = True) -> list[UsageCreditPackage]: packages = await self._package_repo.list_all(active_only=active_only) for pkg in packages: await self.sync_package(pkg) return packages