scripts/stripe_billing.py

"""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},
        )