"""Cloud wallet checkout, daily settlement, serving policy, and dashboard snapshot.""" from __future__ import annotations from datetime import datetime, timezone from secrets import token_urlsafe from typing import Optional from uuid import UUID import structlog from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from src.core.exceptions import NotFoundError from src.domain.billing.interfaces import StripeClient, StripeSyncCapable from src.domain.cloud_credits.conversion import ( DISPLAY_SCALE, credits_from_millicredits, millicredits_from_metrics, ) from src.domain.cloud_credits.interfaces import PublicServingGate from src.domain.cloud_credits.repository import ( CloudCreditLedgerRepository, CloudCreditPackageRepository, CloudCreditPurchaseRepository, CloudUsageConversionRateRepository, WorkspaceCloudCreditPolicyRepository, ) from src.domain.cloud_credits.wallet import CloudCreditWallet from src.domain.projects.repository import ProjectRepository logger = structlog.get_logger() PURCHASE_TYPE = "cloud_credit_topup" UNAVAILABLE_MESSAGE = ( "The Web App is not available if you are the owner of the site please contact support" ) class CloudCreditService: def __init__( self, *, db: AsyncSession, stripe_client: Optional[StripeClient] = None, ensure_customer=None, serving_gate: PublicServingGate | None = None, ) -> None: self._db = db self._packages = CloudCreditPackageRepository(db) self._purchases = CloudCreditPurchaseRepository(db) self._rates = CloudUsageConversionRateRepository(db) self._policy = WorkspaceCloudCreditPolicyRepository(db) self._wallet = CloudCreditWallet(CloudCreditLedgerRepository(db)) self._projects = ProjectRepository(db) self._stripe = stripe_client self._ensure_customer = ensure_customer self._gate = serving_gate async def list_packages(self): return await self._packages.list_active() async def balance_millicredits(self, workspace_id: UUID) -> int: return await self._wallet.balance_millicredits(workspace_id) async def jobs_allowed(self, workspace_id: UUID) -> bool: """False only when a Cloud cap is enforced and the wallet is at or below zero.""" policy = await self._policy.get(workspace_id) if not policy or not policy.enforcement_enabled or policy.cap_credits is None: return True return await self._wallet.balance_millicredits(workspace_id) > 0 async def public_serving_allowed(self, workspace_id: UUID) -> bool: return await self.jobs_allowed(workspace_id) async def settle_daily_usage( self, workspace_id: UUID, snapshot_date: str, metrics: dict, *, millicredits_override: Optional[int] = None, ) -> int: """Idempotent daily wallet debit (see CloudCreditWallet.settle_usage). Pass millicredits_override when the caller already applied a per-project markup/fixed-fee (apply_project_markup_and_fee in conversion.py) on top of metered usage — otherwise millicredits are computed from `metrics` and the active DB conversion rates alone. """ millicredits = millicredits_override if millicredits is None: rates = await self._rates.list_active() millicredits = millicredits_from_metrics(metrics, rates) settlement = await self._wallet.settle_usage( workspace_id=workspace_id, snapshot_date=snapshot_date, millicredits=millicredits, metrics=metrics, ) await self._sync_serving_state(workspace_id) return settlement.millicredits async def start_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: pkg = await self._packages.get_by_slug(package_slug) if not pkg or not pkg.is_active: raise NotFoundError(f"Cloud credit package '{package_slug}' not found") if not pkg.stripe_price_id: raise NotFoundError(f"Cloud credit package '{package_slug}' is not synced to Stripe") if not self._stripe or not self._ensure_customer: raise NotFoundError("Stripe client not configured") customer_id = await self._ensure_customer(workspace_id, user_id, email, workspace_name) metadata = { "workspace_id": str(workspace_id), "package_slug": package_slug, "purchase_type": PURCHASE_TYPE, } 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 fulfill_paid_session( self, *, workspace_id: UUID, stripe_customer_id: str, session_id: str, payment_intent_id: Optional[str], package_slug: str, ) -> None: existing = await self._purchases.get_by_session_id(session_id) if existing and existing.status == "paid": await self._sync_serving_state(workspace_id) return pkg = await self._packages.get_by_slug(package_slug) if not pkg: logger.warning("cloud_credit_topup_package_missing", slug=package_slug, session_id=session_id) return millicredits = pkg.credits * DISPLAY_SCALE purchase = None try: with self._db.begin_nested(): purchase = await self._purchases.create( workspace_id=workspace_id, stripe_customer_id=stripe_customer_id, stripe_checkout_session_id=session_id, stripe_payment_intent_id=payment_intent_id, package_id=pkg.id, package_slug=pkg.slug, credits_purchased=pkg.credits, millicredits_purchased=millicredits, amount_minor=pkg.price_minor, currency=pkg.currency, status="paid", paid_at=datetime.now(timezone.utc), ) except IntegrityError: existing = await self._purchases.get_by_session_id(session_id) if existing and existing.status == "paid": await self._sync_serving_state(workspace_id) return raise await self._wallet.credit_purchase( workspace_id=workspace_id, millicredits=millicredits, purchase_id=purchase.id, source_id=session_id, description=f"Purchased {pkg.credits} Cloud credits ({pkg.name})", ) await self._sync_serving_state(workspace_id) logger.info( "cloud_credit_topup_fulfilled", workspace_id=str(workspace_id), session_id=session_id, credits=pkg.credits, ) async def issue_owner_bypass(self, workspace_id: UUID) -> str: token = token_urlsafe(24) if self._gate: projects = await self._projects.list_for_workspace(workspace_id) await self._gate.set_owner_bypass([p.id for p in projects], token) return token async def snapshot( self, workspace_id: UUID, *, period_start: datetime | None, period_metrics: dict[str, float], ) -> dict: millicredits = await self._wallet.balance_millicredits(workspace_id) consumed = 0 if period_start: consumed = abs( await CloudCreditLedgerRepository(self._db).sum_consumption_since(workspace_id, period_start) ) policy = await self._policy.get(workspace_id) enforcement = bool(policy and policy.enforcement_enabled and policy.cap_credits) cap = policy.cap_credits if policy else None used_pct = None alert = None if enforcement and cap: used_pct = round((credits_from_millicredits(consumed) / cap) * 100, 2) if used_pct >= 100: alert = "exhausted" elif used_pct >= 80: alert = "warning" packages = [ { "slug": p.slug, "name": p.name, "description": p.description, "credits": p.credits, "price_minor": p.price_minor, "currency": p.currency, "stripe_ready": bool(p.stripe_price_id), } for p in await self.list_packages() ] return { "millicredits": millicredits, "credits": credits_from_millicredits(millicredits), "consumed_credits_period": credits_from_millicredits(consumed), "fair_usage": not enforcement, "enforcement_enabled": enforcement, "cap_credits": cap, "used_pct": used_pct, "alert_level": alert, "public_serving": await self.public_serving_allowed(workspace_id), "period_metrics": period_metrics, "packages": packages, } async def sync_packages_to_stripe(self, stripe: StripeSyncCapable) -> None: for pkg in await self._packages.list_active(): updates: dict = {} if not pkg.stripe_product_id: product = await stripe.create_product( name=pkg.name, description=pkg.description or "", metadata={"p2hs_product": "cloud_credits", "p2hs_slug": pkg.slug, "p2hs_credits": str(pkg.credits)}, ) updates["stripe_product_id"] = product["id"] product_id = pkg.stripe_product_id or updates.get("stripe_product_id") if not product_id: continue if not pkg.stripe_price_id: price = await stripe.create_one_time_price( product_id=product_id, unit_amount=pkg.price_minor, currency=pkg.currency, nickname=pkg.name, ) updates["stripe_price_id"] = price["id"] if updates: await self._packages.update(pkg.id, **updates) async def _sync_serving_state(self, workspace_id: UUID) -> None: allowed = await self.public_serving_allowed(workspace_id) await self._policy.mark_public_blocked(workspace_id, blocked=not allowed) if not self._gate: return projects = await self._projects.list_for_workspace(workspace_id) await self._gate.set_public_blocked([p.id for p in projects], blocked=not allowed)