"""Optional Stripe billing primitives with no import-time network activity.""" from __future__ import annotations from collections import Counter from dataclasses import dataclass from decimal import InvalidOperation, ROUND_CEILING, Decimal from pathlib import Path from typing import Any, Mapping from urllib.parse import urlsplit ENV_NAMES = ( "SYNDERESIS_STRIPE_SECRET_KEY", "SYNDERESIS_STRIPE_WEBHOOK_SECRET", "SYNDERESIS_STRIPE_FIXED_PRICE_ID", "SYNDERESIS_STRIPE_METERED_PRICE_ID", "SYNDERESIS_STRIPE_METER_EVENT_NAME", "SYNDERESIS_STRIPE_PUBLIC_ORIGIN", "SYNDERESIS_STRIPE_INCLUDED_RETAIL_MICRO_USD", "SYNDERESIS_STRIPE_FIXED_MONTHLY_CENTS", "SYNDERESIS_STRIPE_EXPECTED_LIVEMODE", "SYNDERESIS_STRIPE_REQUIRE_PERSISTENT_LEDGER", ) BYOK_PLATFORM_FEE_ENV = "SYNDERESIS_STRIPE_BYOK_PLATFORM_FEE_RATE" DEFAULT_BYOK_PLATFORM_FEE_RATE = Decimal("0.25") MAX_INCLUDED_RETAIL_MICRO_USD = 10**15 MAX_FIXED_MONTHLY_CENTS = 10**11 def _boolean(value: str) -> bool: if value.lower() not in {"true", "false"}: raise ValueError("billing boolean configuration must be true or false") return value.lower() == "true" def _positive_int(value: str, *, upper_bound: int) -> int: try: parsed = int(value) except (TypeError, ValueError) as exc: raise ValueError("billing numeric configuration must be an integer") from exc if parsed <= 0 or parsed > upper_bound: raise ValueError("billing numeric configuration is out of range") return parsed def _rate( value: Any, *, name: str, allow_zero: bool = True, ) -> Decimal: try: parsed = Decimal(str(value)) except (InvalidOperation, TypeError, ValueError) as exc: raise ValueError(f"{name} must be a finite decimal rate") from exc if ( not parsed.is_finite() or parsed < 0 or (not allow_zero and parsed == 0) or parsed >= 1 ): lower_bound = "greater than zero" if not allow_zero else "at least zero" raise ValueError(f"{name} must be {lower_bound} and less than one") return parsed @dataclass(frozen=True) class StripeBillingConfig: enabled: bool = False secret_key: str = "" webhook_secret: str = "" fixed_price_id: str = "" metered_price_id: str = "" meter_event_name: str = "" public_origin: str = "" included_retail_micro_usd: int = 0 fixed_monthly_cents: int = 0 expected_livemode: bool = False require_persistent_ledger: bool = False byok_platform_fee_rate: Decimal = DEFAULT_BYOK_PLATFORM_FEE_RATE db_path: Path = Path() @classmethod def from_env( cls, env: Mapping[str, str], *, db_path: Path ) -> "StripeBillingConfig": values = {name: str(env.get(name, "")).strip() for name in ENV_NAMES} present = {name for name, value in values.items() if value} byok_fee_value = str(env.get(BYOK_PLATFORM_FEE_ENV, "")).strip() if not present: if byok_fee_value: raise ValueError( "BYOK platform fee configuration requires Stripe billing" ) return cls(db_path=db_path) if present != set(ENV_NAMES): raise ValueError("Stripe billing configuration is incomplete") byok_platform_fee_rate = ( _rate( byok_fee_value, name="BYOK platform fee rate", allow_zero=False, ) if byok_fee_value else DEFAULT_BYOK_PLATFORM_FEE_RATE ) included = _positive_int( values["SYNDERESIS_STRIPE_INCLUDED_RETAIL_MICRO_USD"], upper_bound=MAX_INCLUDED_RETAIL_MICRO_USD, ) fixed = _positive_int( values["SYNDERESIS_STRIPE_FIXED_MONTHLY_CENTS"], upper_bound=MAX_FIXED_MONTHLY_CENTS, ) if fixed != included // 10_000 or included % 10_000: raise ValueError("fixed price must equal the included retail allowance") persistent = _boolean( values["SYNDERESIS_STRIPE_REQUIRE_PERSISTENT_LEDGER"] ) resolved = db_path.resolve() expected_livemode = _boolean( values["SYNDERESIS_STRIPE_EXPECTED_LIVEMODE"] ) if (persistent or expected_livemode) and not resolved.is_relative_to( Path("/data") ): raise ValueError("Stripe billing requires a persistent /data ledger") origin = values["SYNDERESIS_STRIPE_PUBLIC_ORIGIN"].rstrip("/") parsed_origin = urlsplit(origin) if ( parsed_origin.scheme != "https" or not parsed_origin.hostname or parsed_origin.username is not None or parsed_origin.password is not None or parsed_origin.query or parsed_origin.fragment or parsed_origin.path not in {"", "/"} ): raise ValueError("Stripe public origin must be a credential-free HTTPS origin") return cls( enabled=True, secret_key=values["SYNDERESIS_STRIPE_SECRET_KEY"], webhook_secret=values["SYNDERESIS_STRIPE_WEBHOOK_SECRET"], fixed_price_id=values["SYNDERESIS_STRIPE_FIXED_PRICE_ID"], metered_price_id=values["SYNDERESIS_STRIPE_METERED_PRICE_ID"], meter_event_name=values["SYNDERESIS_STRIPE_METER_EVENT_NAME"], public_origin=origin, included_retail_micro_usd=included, fixed_monthly_cents=fixed, expected_livemode=expected_livemode, require_persistent_ledger=persistent, byok_platform_fee_rate=byok_platform_fee_rate, db_path=db_path, ) def _finite_decimal(value: Any) -> Decimal: try: parsed = Decimal(str(value)) except (InvalidOperation, TypeError, ValueError) as exc: raise ValueError("provider cost must be a finite decimal string") from exc if not parsed.is_finite(): raise ValueError("provider cost must be finite") return parsed def loaded_provider_cost(details: list[dict[str, Any]]) -> Decimal: total = Decimal(0) for detail in details: route = detail.get("route") funding = detail.get("funding") if route not in {"openrouter", "direct"} or funding not in { "synderesis", "customer", }: raise ValueError("trusted cost detail requires an explicit route and funding") if ( not isinstance(detail.get("kind"), str) or not detail["kind"].strip() ): raise ValueError("trusted cost detail requires a kind") if funding == "customer": continue if detail.get("cost_source") not in { "provider_reported", "committed_model_price", }: raise ValueError("trusted cost detail requires an explicit cost source") cost = _finite_decimal(detail.get("cost_usd")) if cost < 0: raise ValueError("provider cost cannot be negative") total += cost * (Decimal("1.055") if route == "openrouter" else Decimal(1)) return total def _trusted_byok_reference(detail: Mapping[str, Any]) -> tuple[Decimal, str]: """Return customer-funded reference cost without treating it as our cost.""" legacy_cost = detail.get("cost_usd") if legacy_cost is not None and legacy_cost != "": raise ValueError( "BYOK reference cost must use the provider_cost_usd field" ) raw_provider_cost = detail.get("provider_cost_usd") if raw_provider_cost is not None and raw_provider_cost != "": if detail.get("cost_source") != "provider_reported": raise ValueError( "BYOK provider cost requires provider_reported cost source" ) provider_cost = _finite_decimal(raw_provider_cost) if provider_cost < 0: raise ValueError("BYOK reference cost cannot be negative") return provider_cost, "provider_reported" if detail.get("cost_source") != "committed_model_price": raise ValueError( "BYOK fallback requires committed_model_price cost source" ) if detail.get("token_source") != "provider_reported": raise ValueError("BYOK fallback requires provider-reported tokens") model = detail.get("model") if not isinstance(model, str) or not model.strip(): raise ValueError("BYOK fallback requires a committed model identity") def token_count(name: str) -> int: value = detail.get(name) if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise ValueError("BYOK fallback token counts must be non-negative integers") return value prompt_tokens = token_count("prompt_tokens") completion_tokens = token_count("completion_tokens") prompt_price = _finite_decimal(detail.get("prompt_price_usd_per_token")) completion_price = _finite_decimal( detail.get("completion_price_usd_per_token") ) if prompt_price < 0 or completion_price < 0: raise ValueError("BYOK fallback model prices cannot be negative") source = detail.get("price_source") updated_at = detail.get("price_updated_at") if ( not isinstance(source, str) or not source.strip() or not isinstance(updated_at, str) or not updated_at.strip() ): raise ValueError("BYOK fallback requires committed price provenance") return ( Decimal(prompt_tokens) * prompt_price + Decimal(completion_tokens) * completion_price, "committed_model_price", ) @dataclass(frozen=True) class PricingBreakdown: """Exact pre-rounding components retained as private billing evidence.""" managed_margin_rate: Decimal byok_platform_fee_rate: Decimal synderesis_provider_cost_usd: Decimal managed_retail_usd: Decimal byok_reference_cost_usd: Decimal byok_platform_fee_usd: Decimal unrounded_retail_usd: Decimal retail_micro_usd: int byok_reference_source_counts: tuple[tuple[str, int], ...] def evidence(self) -> dict[str, Any]: """Return canonical JSON-safe values without binary floating point.""" return { "version": 1, "managed_margin_rate": str(self.managed_margin_rate), "byok_platform_fee_rate": str(self.byok_platform_fee_rate), "synderesis_provider_cost_usd": str( self.synderesis_provider_cost_usd ), "byok_synderesis_provider_cost_usd": "0", "managed_retail_usd": str(self.managed_retail_usd), "byok_reference_cost_usd": str(self.byok_reference_cost_usd), "byok_platform_fee_usd": str(self.byok_platform_fee_usd), "unrounded_retail_usd": str(self.unrounded_retail_usd), "retail_micro_usd": self.retail_micro_usd, "byok_reference_source_counts": dict( self.byok_reference_source_counts ), } def pricing_breakdown( details: list[dict[str, Any]], *, margin: Decimal = Decimal("0.50"), byok_platform_fee_rate: Decimal = DEFAULT_BYOK_PLATFORM_FEE_RATE, ) -> PricingBreakdown: """Price managed cost and BYOK references, then round exactly once.""" managed_margin_rate = _rate(margin, name="margin") fee_rate = _rate( byok_platform_fee_rate, name="BYOK platform fee rate", allow_zero=False, ) managed_loaded = loaded_provider_cost(details) byok_reference = Decimal(0) reference_sources: Counter[str] = Counter() for detail in details: if detail.get("funding") != "customer": continue reference_cost, reference_source = _trusted_byok_reference(detail) byok_reference += reference_cost reference_sources[reference_source] += 1 managed_retail = managed_loaded / (Decimal(1) - managed_margin_rate) byok_fee = byok_reference * fee_rate unrounded_retail = managed_retail + byok_fee units = ( 0 if not unrounded_retail else int( (unrounded_retail * Decimal(1_000_000)).quantize( Decimal(1), rounding=ROUND_CEILING, ) ) ) return PricingBreakdown( managed_margin_rate=managed_margin_rate, byok_platform_fee_rate=fee_rate, synderesis_provider_cost_usd=managed_loaded, managed_retail_usd=managed_retail, byok_reference_cost_usd=byok_reference, byok_platform_fee_usd=byok_fee, unrounded_retail_usd=unrounded_retail, retail_micro_usd=units, byok_reference_source_counts=tuple(sorted(reference_sources.items())), ) def retail_micro_usd( details: list[dict[str, Any]], margin: Decimal = Decimal("0.50"), *, byok_platform_fee_rate: Decimal = DEFAULT_BYOK_PLATFORM_FEE_RATE, ) -> int: return pricing_breakdown( details, margin=margin, byok_platform_fee_rate=byok_platform_fee_rate, ).retail_micro_usd @dataclass(frozen=True) class SubscriptionEntitlement: subscription_id: str stripe_customer_id: str status: str period_start: int period_end: int metered_item_id: str def validate_subscription_catalog( config: StripeBillingConfig, subscription: Mapping[str, Any], *, prices: Mapping[str, Mapping[str, Any]], meter: Mapping[str, Any], ) -> SubscriptionEntitlement: if bool(subscription.get("livemode")) != config.expected_livemode: raise ValueError("subscription livemode mismatch") items = list(subscription.get("items", {}).get("data", [])) if len(items) != 2: raise ValueError("subscription must contain exactly two items") price_ids = [item.get("price", {}).get("id") for item in items] if len(set(price_ids)) != len(price_ids): raise ValueError("subscription contains duplicate prices") by_price = {item.get("price", {}).get("id"): item for item in items} expected_prices = {config.fixed_price_id, config.metered_price_id} if set(by_price) != expected_prices or set(prices) != expected_prices: raise ValueError("subscription price bundle mismatch") fixed_item = by_price[config.fixed_price_id] metered_item = by_price[config.metered_price_id] fixed_price = prices[config.fixed_price_id] metered_price = prices[config.metered_price_id] for price in (fixed_price, metered_price): recurring = price.get("recurring") or {} if ( not price.get("active") or bool(price.get("livemode")) != config.expected_livemode or price.get("currency") != "usd" or recurring.get("interval") != "month" or int(recurring.get("interval_count", 0)) != 1 ): raise ValueError("subscription item currency or interval mismatch") if fixed_price["recurring"].get("usage_type") != "licensed": raise ValueError("fixed item must be licensed") if ( fixed_item.get("quantity") != 1 or fixed_price.get("unit_amount") != config.fixed_monthly_cents ): raise ValueError("fixed item quantity or amount mismatch") if metered_price["recurring"].get("usage_type") != "metered": raise ValueError("usage item must be metered") if metered_item.get("quantity") is not None: raise ValueError("metered item quantity must be omitted") if ( metered_price.get("billing_scheme") != "tiered" or metered_price.get("tiers_mode") != "graduated" ): raise ValueError("metered price must use graduated tiers") tiers = list(metered_price.get("tiers") or []) if len(tiers) != 2: raise ValueError("metered price must contain exactly two tiers") first, final = tiers if ( int(first.get("up_to", -1)) != config.included_retail_micro_usd or int(first.get("unit_amount", -1)) != 0 or int(first.get("flat_amount", -1)) != 0 or final.get("up_to") != "inf" or _finite_decimal(final.get("unit_amount_decimal")) != Decimal("0.0001") or int(final.get("flat_amount", -1)) != 0 ): raise ValueError("metered price tiers mismatch") meter_id = metered_price.get("recurring", {}).get("meter") if ( not meter_id or str(meter.get("id", "")) != str(meter_id) or not meter.get("active", True) or str(meter.get("status", "active")) != "active" or bool(meter.get("livemode")) != config.expected_livemode or str(meter.get("event_name", "")) != config.meter_event_name or (meter.get("default_aggregation") or {}).get("formula") != "sum" or (meter.get("customer_mapping") or {}).get("type") != "by_id" or (meter.get("customer_mapping") or {}).get("event_payload_key") != "stripe_customer_id" or (meter.get("value_settings") or {}).get("event_payload_key") != "value" ): raise ValueError("meter catalog mismatch") start = int(metered_item.get("current_period_start", 0)) end = int(metered_item.get("current_period_end", 0)) fixed_start = int(fixed_item.get("current_period_start", 0)) fixed_end = int(fixed_item.get("current_period_end", 0)) if ( not start or end <= start or fixed_start != start or fixed_end != end ): raise ValueError("subscription items have no matching valid period") return SubscriptionEntitlement( str(subscription["id"]), str(subscription["customer"]), str(subscription["status"]), start, end, str(metered_item["id"]), ) def validate_subscription( config: StripeBillingConfig, subscription: Mapping[str, Any], *, prices: Mapping[str, Mapping[str, Any]], meter: Mapping[str, Any], ) -> SubscriptionEntitlement: """Validate the exact canonical catalog and require a paid-access status.""" if subscription.get("status") not in {"active", "trialing"}: raise ValueError("subscription is not entitled") return validate_subscription_catalog( config, subscription, prices=prices, meter=meter, ) def verify_webhook( secret: str, body: bytes, signature: str, tolerance: int = 300 ) -> dict[str, Any]: import stripe event = stripe.Webhook.construct_event( body, signature, secret, tolerance=tolerance ) if hasattr(event, "to_dict"): return event.to_dict(recursive=True) return dict(event) class StripeGateway: def __init__(self, client: Any, config: StripeBillingConfig): self.client, self.config = client, config def checkout( self, customer_id: str, stripe_customer_id: str = "", *, checkout_attempt_id: str = "", idempotency_key: str = "", ) -> Any: metadata = {"customer_id": customer_id} if checkout_attempt_id: metadata["checkout_attempt_id"] = checkout_attempt_id params: dict[str, Any] = { "mode": "subscription", "line_items": [ {"price": self.config.fixed_price_id, "quantity": 1}, {"price": self.config.metered_price_id}, ], "metadata": metadata, "subscription_data": {"metadata": metadata}, "success_url": f"{self.config.public_origin}/account/?billing=success", "cancel_url": f"{self.config.public_origin}/account/?billing=cancel", } if stripe_customer_id: params["customer"] = stripe_customer_id return self.client.v1.checkout.sessions.create( params=params, options={ "idempotency_key": ( idempotency_key or f"checkout-{customer_id}-{self.config.fixed_price_id}" ) }, ) def portal(self, stripe_customer_id: str) -> Any: if not stripe_customer_id: raise ValueError("billing portal requires a mapped Stripe customer") return self.client.v1.billing_portal.sessions.create( params={ "customer": stripe_customer_id, "return_url": f"{self.config.public_origin}/account/", }, options=None, ) def meter( self, identifier: str, customer: str, units: int, *, timestamp: int ) -> Any: return self.client.v1.billing.meter_events.create( params={ "event_name": self.config.meter_event_name, "identifier": identifier, "payload": { "stripe_customer_id": customer, "value": str(units), }, "timestamp": timestamp, }, options={"idempotency_key": identifier}, )