"""Author RAG — Stripe Billing Service. Full Stripe integration — checkout, portal, webhooks, referral coupons. """ from __future__ import annotations import secrets import structlog from datetime import datetime, timedelta, timezone try: import stripe except ImportError: # pragma: no cover stripe = None # type: ignore[assignment] from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.config import get_settings from app.repositories.access_repo import AccessRepository from app.repositories.user_repo import UserRepository from app.services.auth_service import _hash_password logger = structlog.get_logger(__name__) def _refresh_stripe_key() -> None: """Load the current Stripe secret key (supports SuperAdmin hot-reload).""" if stripe is None: raise RuntimeError("Stripe SDK not installed") key = get_settings().STRIPE_SECRET_KEY if not key: raise ValueError("Stripe is not configured on this server") stripe.api_key = key def _stripe_resource_id(value: object | None) -> str: """Normalize Stripe customer/subscription fields (id string or expanded object).""" if value is None: raise ValueError("Stripe checkout session is missing billing details. Please contact support.") if isinstance(value, str): return value resource_id = getattr(value, "id", None) if not resource_id: raise ValueError("Stripe checkout session is incomplete. Please contact support.") return str(resource_id) def _customer_email_from_session(session: object) -> str: """Extract payer email from a Stripe Checkout session.""" details = getattr(session, "customer_details", None) email = getattr(details, "email", None) if details else None if not email: raise ValueError( "No email on this payment session. Use the same email in Stripe Checkout or contact support." ) return str(email).lower().strip() def _stripe_metadata_value(metadata: object | None, key: str, default: str = "starter") -> str: """Read metadata from Stripe StripeObject or plain dict.""" if not metadata: return default if isinstance(metadata, dict): value = metadata.get(key) return str(value) if value else default try: value = metadata[key] # StripeObject supports __getitem__ return str(value) if value else default except (KeyError, TypeError): value = getattr(metadata, key, None) return str(value) if value else default # Configure Stripe SDK when available. Refreshed again before each API call. if stripe and get_settings().STRIPE_SECRET_KEY: stripe.api_key = get_settings().STRIPE_SECRET_KEY # ── Webhook Signature Verification ─────────────────────────────────────────── def verify_webhook_signature(payload: bytes, sig_header: str): """Verify Stripe webhook signature and return the parsed event.""" if stripe is None: raise RuntimeError("Stripe SDK not installed") try: event = stripe.Webhook.construct_event( payload=payload, sig_header=sig_header, secret=get_settings().STRIPE_WEBHOOK_SECRET, ) return event except stripe.SignatureVerificationError as e: logger.warning("Invalid Stripe webhook signature", error=str(e)) raise except ValueError as e: logger.warning("Malformed Stripe webhook payload", error=str(e)) raise # ── Checkout ────────────────────────────────────────────────────────────────── async def create_checkout_session( plan_id: str, stripe_price_id: str, success_url: str, cancel_url: str, customer_email: str | None = None, existing_customer_id: str | None = None, coupon_code: str | None = None, author_id: str | None = None, ) -> str: """Create a Stripe Checkout session and return the redirect URL. Card always required upfront — no free access without payment. Args: plan_id: Our internal plan ID (e.g. "starter"). stripe_price_id: The Stripe Price ID for the selected plan/billing. success_url: Redirect after payment (must include {CHECKOUT_SESSION_ID}). cancel_url: Redirect if user cancels. customer_email: Pre-fill email in Stripe Checkout. existing_customer_id: For upgrades — attach to existing Stripe customer. coupon_code: Referral/affiliate coupon to apply automatically. author_id: Logged-in author id (metadata + client_reference_id). Returns: Stripe Checkout URL — redirect the browser here. """ metadata: dict[str, str] = {"plan_id": plan_id} if author_id: metadata["author_id"] = author_id params: dict = { "mode": "subscription", "line_items": [{"price": stripe_price_id, "quantity": 1}], "success_url": success_url, "cancel_url": cancel_url, "payment_method_collection": "always", # Card required upfront "metadata": metadata, "subscription_data": {"metadata": metadata}, "allow_promotion_codes": True, # Enable referral coupons } if author_id: params["client_reference_id"] = author_id if coupon_code: params["discounts"] = [{"coupon": coupon_code}] params.pop("allow_promotion_codes", None) if existing_customer_id: params["customer"] = existing_customer_id elif customer_email: params["customer_email"] = customer_email try: _refresh_stripe_key() session = stripe.checkout.Session.create(**params) logger.info("Stripe checkout session created", session_id=session.id, plan_id=plan_id) return session.url except stripe.StripeError as e: logger.error("Stripe checkout failed", error=str(e), plan_id=plan_id) raise async def create_addon_checkout_session( *, addon_id: str, price_usd: float, display_name: str, author_id: str, success_url: str, cancel_url: str, customer_email: str | None = None, existing_customer_id: str | None = None, description: str | None = None, extra_metadata: dict[str, str] | None = None, ) -> str: """Create a one-time Stripe Checkout session for a platform add-on (R-202). Uses price_data so SuperAdmin can change USD without syncing Stripe Price IDs. """ unit_amount = int(round(float(price_usd) * 100)) if unit_amount < 50: raise ValueError("Add-on price must be at least $0.50 for Stripe Checkout") metadata: dict[str, str] = { "addon": addon_id, "author_id": author_id, } if extra_metadata: metadata.update({k: str(v) for k, v in extra_metadata.items() if v is not None}) params: dict = { "mode": "payment", "line_items": [ { "quantity": 1, "price_data": { "currency": "usd", "unit_amount": unit_amount, "product_data": { "name": display_name or "Website URL unlock", "description": description or "One-time unlock credit for a locked embed website URL", }, }, } ], "success_url": success_url, "cancel_url": cancel_url, "metadata": metadata, "payment_intent_data": { "metadata": metadata, }, } if existing_customer_id: params["customer"] = existing_customer_id elif customer_email: params["customer_email"] = customer_email try: _refresh_stripe_key() session = stripe.checkout.Session.create(**params) logger.info( "Stripe addon checkout created", session_id=session.id, addon=addon_id, author_id=author_id, ) return session.url except stripe.StripeError as e: logger.error("Stripe addon checkout failed", error=str(e), addon=addon_id) raise async def get_checkout_session(session_id: str) -> stripe.checkout.Session: """Retrieve and verify a completed Stripe Checkout session. Called by POST /api/auth/activate after the browser returns from Stripe. """ try: _refresh_stripe_key() session = stripe.checkout.Session.retrieve( session_id, expand=["customer", "subscription", "subscription.items.data.price"], ) logger.info( "Stripe session retrieved", session_id=session_id, payment_status=session.payment_status, ) return session except stripe.StripeError as e: logger.error("Failed to retrieve Stripe session", session_id=session_id, error=str(e)) raise # ── Customer Portal ─────────────────────────────────────────────────────────── async def create_portal_session(stripe_customer_id: str, return_url: str) -> str: """Create a Stripe Customer Portal session for self-service billing. Authors can: update card, view invoices, cancel, upgrade/downgrade. """ try: session = stripe.billing_portal.Session.create( customer=stripe_customer_id, return_url=return_url, ) logger.info("Stripe portal session created", customer_id=stripe_customer_id) return session.url except stripe.StripeError as e: logger.error("Failed to create portal session", customer_id=stripe_customer_id, error=str(e)) raise # ── Subscription Management ─────────────────────────────────────────────────── async def get_subscription(subscription_id: str) -> stripe.Subscription: """Retrieve full subscription details from Stripe.""" return stripe.Subscription.retrieve( subscription_id, expand=["items.data.price", "customer"], ) async def cancel_subscription(subscription_id: str, at_period_end: bool = True) -> None: """Cancel a Stripe subscription. Args: at_period_end: True = access continues to end of paid period (recommended). False = immediate cancellation with proration. """ try: stripe.Subscription.modify(subscription_id, cancel_at_period_end=at_period_end) logger.info("Subscription cancel scheduled", sub_id=subscription_id, at_end=at_period_end) except stripe.StripeError as e: logger.error("Failed to cancel subscription", sub_id=subscription_id, error=str(e)) raise async def upgrade_subscription( subscription_id: str, new_stripe_price_id: str, new_plan_id: str | None = None, ) -> None: """Upgrade or downgrade a subscription. Stripe handles proration automatically.""" try: _refresh_stripe_key() sub = stripe.Subscription.retrieve(subscription_id) modify_kwargs: dict = { "items": [{"id": sub["items"]["data"][0]["id"], "price": new_stripe_price_id}], "proration_behavior": "create_prorations", } if new_plan_id: modify_kwargs["metadata"] = {"plan_id": new_plan_id} stripe.Subscription.modify(subscription_id, **modify_kwargs) logger.info( "Subscription upgraded", sub_id=subscription_id, new_price=new_stripe_price_id, plan_id=new_plan_id, ) except stripe.StripeError as e: logger.error("Failed to upgrade subscription", sub_id=subscription_id, error=str(e)) raise # ── Account Activation (Post-Payment) ──────────────────────────────────────── class BillingService: """Orchestrates account creation and access provisioning after Stripe payment.""" def __init__(self, db: AsyncSession) -> None: self._db = db self._access = AccessRepository(db) self._users = UserRepository(db) @staticmethod def _generate_temp_password() -> str: """Generate a secure temporary password for webhook auto-provision.""" alphabet = "abcdefghijkmnopqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789" return "".join(secrets.choice(alphabet) for _ in range(14)) async def _resolve_plan(self, plan_id: str) -> tuple[str, int]: """Return plan_id and token budget from PricingPlan or defaults.""" from app.models.pricing_plan import PricingPlan from sqlalchemy import select plan_result = await self._db.execute( select(PricingPlan).where(PricingPlan.id == plan_id) ) plan = plan_result.scalar_one_or_none() if not plan: logger.warning("Unknown plan_id in Stripe session", plan_id=plan_id) return "starter", 500_000 return plan_id, plan.max_token_budget async def _find_user_for_stripe_activation( self, email: str, stripe_session_id: str, stripe_customer_id: str, stripe_sub_id: str | None = None, ): """Locate an author by email or Stripe checkout identifiers.""" from sqlalchemy import select from app.models.client_access import ClientAccess from app.models.user import User user = await self._users.get_by_email(email) if user: return user for column, value in ( (User.stripe_checkout_session_id, stripe_session_id), (User.stripe_customer_id, stripe_customer_id), (User.stripe_subscription_id, stripe_sub_id), ): if not value: continue result = await self._db.execute(select(User).where(column == value)) found = result.scalar_one_or_none() if found: return found if stripe_sub_id: access_result = await self._db.execute( select(ClientAccess).where(ClientAccess.stripe_subscription_id == stripe_sub_id) ) access = access_result.scalar_one_or_none() if access: return await self._users.get_by_id(access.author_id) return None async def _create_author_for_paid_session( self, *, email: str, password_hash: str, full_name: str, plan_id: str, token_budget: int, stripe_customer_id: str, stripe_sub_id: str, stripe_session_id: str, ): """Create author + access; commit user first so payment is never orphaned.""" user_data: dict = { "email": email, "password_hash": password_hash, "full_name": full_name or None, "role": "author", "is_active": True, "stripe_customer_id": stripe_customer_id, "stripe_subscription_id": stripe_sub_id, "stripe_plan_id": plan_id, "stripe_checkout_session_id": stripe_session_id, "activation_status": "active", "provisioned_via": "stripe_activate", "password_change_required": False, } try: user = await self._users.create(user_data) await self._db.commit() except IntegrityError: await self._db.rollback() existing = await self._find_user_for_stripe_activation( email, stripe_session_id, stripe_customer_id, stripe_sub_id ) if existing: return existing minimal = {k: v for k, v in user_data.items() if k not in ( "activation_status", "provisioned_via", "password_change_required", "stripe_checkout_session_id", )} user = await self._users.create(minimal) await self._db.commit() ok = await self._ensure_stripe_client_access( author_id=user.id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) if not ok: logger.error("access_grant_failed_after_user_create", user_id=user.id) return user async def _resolve_access_record(self, author_id: str, stripe_sub_id: str): """Find the ClientAccess row to update — active, revoked, or by subscription id.""" from sqlalchemy import select from app.models.client_access import ClientAccess if stripe_sub_id: result = await self._db.execute( select(ClientAccess).where(ClientAccess.stripe_subscription_id == stripe_sub_id) ) by_sub = result.scalar_one_or_none() if by_sub: if by_sub.author_id != author_id: user = await self._users.get_by_id(author_id) if user and user.stripe_subscription_id == stripe_sub_id: logger.warning( "stripe_access_reclaimed", sub_id=stripe_sub_id, from_author=by_sub.author_id, to_author=author_id, ) by_sub.author_id = author_id if by_sub.author_id == author_id: return by_sub access = await self._access.get_active_for_author(author_id) if access: return access result = await self._db.execute( select(ClientAccess) .where(ClientAccess.author_id == author_id) .order_by(ClientAccess.created_at.desc()) .limit(1) ) return result.scalar_one_or_none() async def _reclaim_access_for_subscriber( self, *, author_id: str, stripe_sub_id: str, plan_id: str, token_budget: int, ): """Attach the subscription access row to the paying author.""" from sqlalchemy import select from app.models.client_access import ClientAccess if not stripe_sub_id: return None result = await self._db.execute( select(ClientAccess).where(ClientAccess.stripe_subscription_id == stripe_sub_id) ) access = result.scalar_one_or_none() if not access: return None if access.author_id != author_id: logger.warning( "stripe_access_reclaimed", sub_id=stripe_sub_id, from_author=access.author_id, to_author=author_id, ) access.author_id = author_id now = datetime.now(timezone.utc) access.plan = plan_id access.plan_id = plan_id access.token_budget = token_budget access.is_revoked = False access.auto_renew = True exp = access.expires_at compare_now = now.replace(tzinfo=None) if (exp and exp.tzinfo is None) else now if exp is None or exp < compare_now: access.expires_at = now + timedelta(days=365 * 10) return access async def _sync_user_stripe_fields( self, author_id: str, *, plan_id: str, stripe_sub_id: str, stripe_customer_id: str | None = None, ) -> None: """Keep User Stripe columns aligned with the paid subscription.""" from sqlalchemy import update from app.models.user import User values: dict = { "stripe_plan_id": plan_id, "stripe_subscription_id": stripe_sub_id, "chatbot_is_active": True, "activation_status": "active", } if stripe_customer_id: values["stripe_customer_id"] = stripe_customer_id await self._db.execute( update(User).where(User.id == author_id).values(**values) ) async def _purge_conflicting_subscription_rows( self, author_id: str, stripe_sub_id: str, ) -> None: """Remove orphan grants on other authors that block this subscription id.""" from sqlalchemy import delete from app.models.client_access import ClientAccess if not stripe_sub_id: return await self._db.execute( delete(ClientAccess).where( ClientAccess.stripe_subscription_id == stripe_sub_id, ClientAccess.author_id != author_id, ) ) await self._db.flush() async def _sync_from_stored_checkout_session(self, user_id: str) -> bool: """Reconcile grant from stripe_checkout_session_id when User Stripe fields are incomplete.""" user = await self._users.get_by_id(user_id) if not user or not user.stripe_checkout_session_id: return False try: session = await get_checkout_session(user.stripe_checkout_session_id) except Exception as exc: logger.warning("checkout_session_sync_failed", user_id=user_id, error=str(exc)[:120]) return False if session.payment_status != "paid": return False stripe_customer_id = _stripe_resource_id(session.customer) stripe_sub_id = _stripe_resource_id(session.subscription) plan_id, token_budget = await self._resolve_plan( _stripe_metadata_value(session.metadata, "plan_id", user.stripe_plan_id or "starter") ) return await self._ensure_stripe_client_access( author_id=user_id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) async def _upsert_author_access_row( self, *, author_id: str, plan_id: str, token_budget: int, stripe_sub_id: str, ): """Update an existing author grant or create a fresh one.""" from sqlalchemy import select, update from app.core.access.token_crypto import create_subscription_token, hash_subscription_token from app.models.base import generate_uuid from app.models.client_access import ClientAccess now = datetime.now(timezone.utc) expires = now + timedelta(days=365 * 10) await self._purge_conflicting_subscription_rows(author_id, stripe_sub_id) reclaimed = await self._reclaim_access_for_subscriber( author_id=author_id, stripe_sub_id=stripe_sub_id, plan_id=plan_id, token_budget=token_budget, ) if reclaimed: return reclaimed access = await self._resolve_access_record(author_id, stripe_sub_id) if access: await self._apply_access_grant_updates( access, author_id=author_id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, ) return access result = await self._db.execute( select(ClientAccess) .where(ClientAccess.author_id == author_id) .order_by(ClientAccess.created_at.desc()) .limit(1) ) existing = result.scalar_one_or_none() if existing: await self._apply_access_grant_updates( existing, author_id=author_id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, ) return existing grant_id = generate_uuid() token = create_subscription_token( author_id=author_id, grant_id=grant_id, granted_at=now, expires_at=expires, ) return await self._access.create({ "id": grant_id, "author_id": author_id, "granted_by": author_id, # self-provisioned via Stripe (DB may have NOT NULL) "plan": plan_id, "plan_id": plan_id, "stripe_subscription_id": stripe_sub_id, "granted_at": now, "expires_at": expires, "token_hash": hash_subscription_token(token), "token_budget": token_budget, "auto_renew": True, }) async def _ensure_stripe_client_access( self, *, author_id: str, plan_id: str, token_budget: int, stripe_sub_id: str, stripe_customer_id: str | None = None, ) -> bool: """Idempotent: create or reactivate ClientAccess for a paid Stripe author.""" await self._sync_user_stripe_fields( author_id, plan_id=plan_id, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) try: await self._upsert_author_access_row( author_id=author_id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, ) await self._db.commit() except IntegrityError as exc: await self._db.rollback() logger.warning( "stripe_access_upsert_retry", author_id=author_id, sub_id=stripe_sub_id, error=str(exc)[:120], ) try: await self._upsert_author_access_row( author_id=author_id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, ) await self._db.commit() except IntegrityError: await self._db.rollback() logger.error( "stripe_access_create_failed", author_id=author_id, sub_id=stripe_sub_id, ) return False active = await self._access.get_active_for_author(author_id) if active: return True logger.error( "stripe_access_inactive_after_upsert", author_id=author_id, sub_id=stripe_sub_id, ) return False async def _sync_subscription_from_stripe(self, user_id: str) -> bool: """Fetch live Stripe subscription for a customer and ensure local grant.""" user = await self._users.get_by_id(user_id) if not user: return False stripe_sub_id = user.stripe_subscription_id plan_id = user.stripe_plan_id or "starter" stripe_customer_id = user.stripe_customer_id if user.stripe_customer_id and not stripe_sub_id: try: _refresh_stripe_key() subs = stripe.Subscription.list( customer=user.stripe_customer_id, status="active", limit=1, ) if subs.data: stripe_sub_id = subs.data[0].id meta = subs.data[0].metadata or {} plan_id = _stripe_metadata_value(meta, "plan_id", plan_id) except Exception as exc: logger.warning("stripe_subscription_sync_failed", user_id=user_id, error=str(exc)[:120]) if not stripe_sub_id: return False _, token_budget = await self._resolve_plan(plan_id) return await self._ensure_stripe_client_access( author_id=user_id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) async def require_active_subscription(self, author_id: str) -> bool: """Repair and verify an author has an active ClientAccess grant.""" if await self._access.get_active_for_author(author_id): return True if await self.repair_missing_access_for_user(author_id): return True return await self._sync_subscription_from_stripe(author_id) async def repair_missing_access_for_user(self, user_id: str) -> bool: """Backfill ClientAccess when User has Stripe fields but no active grant.""" user = await self._users.get_by_id(user_id) if not user: return False if not user.stripe_subscription_id: if await self._sync_from_stored_checkout_session(user_id): return True if await self._sync_subscription_from_stripe(user_id): return True return False plan_id = user.stripe_plan_id or "starter" stripe_sub_id = user.stripe_subscription_id _, token_budget = await self._resolve_plan(plan_id) ok = await self._ensure_stripe_client_access( author_id=user_id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=user.stripe_customer_id, ) if ok: return True return await self._access.get_active_for_author(user_id) is not None async def _apply_access_grant_updates( self, access, *, author_id: str, plan_id: str, token_budget: int, stripe_sub_id: str, ) -> None: """Update an existing ClientAccess row for a Stripe subscriber.""" now = datetime.now(timezone.utc) access.plan = plan_id access.plan_id = plan_id access.token_budget = token_budget access.is_revoked = False access.auto_renew = True if stripe_sub_id and not access.stripe_subscription_id: access.stripe_subscription_id = stripe_sub_id exp = access.expires_at compare_now = now.replace(tzinfo=None) if (exp and exp.tzinfo is None) else now if exp is None or exp < compare_now: access.expires_at = now + timedelta(days=365 * 10) if access.author_id != author_id: access.author_id = author_id async def _force_password_login( self, *, customer_email: str, password_hash: str, full_name: str | None, stripe_session_id: str, stripe_customer_id: str, stripe_sub_id: str, plan_id: str, token_budget: int = 500_000, full_name_for_create: str = "", ) -> dict: """Guaranteed post-payment path: set password and return login payload.""" user = await self._find_user_for_stripe_activation( customer_email, stripe_session_id, stripe_customer_id, stripe_sub_id ) if not user: user = await self._users.get_by_email(customer_email) if not user: user = await self._create_author_for_paid_session( email=customer_email, password_hash=password_hash, full_name=full_name_for_create or full_name or customer_email, plan_id=plan_id, token_budget=token_budget, stripe_customer_id=stripe_customer_id, stripe_sub_id=stripe_sub_id, stripe_session_id=stripe_session_id, ) return { "user_id": user.id, "email": user.email, "plan_id": plan_id, "already_activated": False, } user.password_hash = password_hash user.password_change_required = False user.activation_status = "active" user.is_active = True user.provisioned_via = "stripe_activate" user.stripe_customer_id = stripe_customer_id user.stripe_subscription_id = stripe_sub_id user.stripe_plan_id = plan_id if full_name: user.full_name = full_name if not user.stripe_checkout_session_id: user.stripe_checkout_session_id = stripe_session_id try: await self._db.commit() except IntegrityError: await self._db.rollback() user = await self._users.get_by_email(customer_email) if not user: raise user.password_hash = password_hash user.password_change_required = False user.activation_status = "active" user.is_active = True if full_name: user.full_name = full_name await self._db.commit() user_id = user.id user_email = user.email ok = await self._ensure_stripe_client_access( author_id=user_id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) if not ok: logger.error("access_grant_failed_force_login", user_id=user_id) logger.info("Author password set after Stripe payment", user_id=user_id, email=user_email) return { "user_id": user_id, "email": user_email, "plan_id": plan_id, "already_activated": True, } async def _complete_existing_user_activation( self, user, *, stripe_session_id: str, password_hash: str, full_name: str | None, plan_id: str, token_budget: int, stripe_customer_id: str, stripe_sub_id: str, ) -> dict: """Attach Stripe payment to an existing author and apply their chosen password.""" now = datetime.now(timezone.utc) was_same_session = user.stripe_checkout_session_id == stripe_session_id user.password_hash = password_hash user.password_change_required = False user.provisioned_via = "stripe_activate" user.activation_status = "active" user.stripe_checkout_session_id = stripe_session_id user.stripe_customer_id = stripe_customer_id user.stripe_subscription_id = stripe_sub_id user.stripe_plan_id = plan_id if full_name: user.full_name = full_name user.credentials_email_sent_at = now try: ok = await self._ensure_stripe_client_access( author_id=user.id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) if not ok: raise RuntimeError("subscription_grant_failed") except (IntegrityError, RuntimeError): await self._db.rollback() return await self._force_password_login( customer_email=user.email, password_hash=password_hash, full_name=full_name, stripe_session_id=stripe_session_id, stripe_customer_id=stripe_customer_id, stripe_sub_id=stripe_sub_id, plan_id=plan_id, token_budget=token_budget, full_name_for_create=user.full_name or user.email, ) self._queue_credentials_email( email=user.email, name=user.full_name or user.email, temp_password=None, user_chose_password=True, ) logger.info( "Existing author linked to Stripe checkout", user_id=user.id, email=user.email, plan_id=plan_id, same_session=was_same_session, ) return { "user_id": user.id, "email": user.email, "plan_id": plan_id, "already_activated": was_same_session, } async def _provision_stripe_author( self, *, email: str, password_hash: str, full_name: str, plan_id: str, token_budget: int, stripe_customer_id: str, stripe_sub_id: str, stripe_session_id: str, provisioned_via: str, activation_status: str, password_change_required: bool, ) -> dict: """Create User + ClientAccess for a paid Stripe checkout.""" from app.models.base import generate_uuid from app.core.access.token_crypto import create_subscription_token, hash_subscription_token now = datetime.now(timezone.utc) user = await self._users.create({ "email": email, "password_hash": password_hash, "full_name": full_name or None, "role": "author", "is_active": True, "stripe_customer_id": stripe_customer_id, "stripe_subscription_id": stripe_sub_id, "stripe_plan_id": plan_id, "stripe_checkout_session_id": stripe_session_id, "activation_status": activation_status, "provisioned_via": provisioned_via, "password_change_required": password_change_required, }) grant_id = generate_uuid() token = create_subscription_token( author_id=user.id, grant_id=grant_id, granted_at=now, expires_at=now + timedelta(days=365 * 10), ) await self._access.create({ "id": grant_id, "author_id": user.id, "granted_by": user.id, # self-provisioned via Stripe (DB may have NOT NULL) "plan": plan_id, "plan_id": plan_id, "stripe_subscription_id": stripe_sub_id, "granted_at": now, "expires_at": now + timedelta(days=365 * 10), "token_hash": hash_subscription_token(token), "token_budget": token_budget, "auto_renew": True, }) await self._db.commit() return { "user_id": user.id, "email": email, "plan_id": plan_id, "subscription_token": token, "already_activated": False, "provisioned_via": provisioned_via, } def _queue_credentials_email( self, *, email: str, name: str, temp_password: str | None, user_chose_password: bool, ) -> None: """Queue credentials email via Celery (non-blocking).""" from app.services.author_onboarding_service import login_url from app.tasks.email_task import send_account_credentials as send_task try: send_task.delay({ "to": email, "name": name or email, "login_url": login_url(), "temp_password": temp_password, "user_chose_password": user_chose_password, }) except Exception as exc: logger.warning("credentials_email_queue_failed", email=email, error=str(exc)[:200]) async def _verify_post_activation( self, user_id: str, *, plan_id: str, token_budget: int, stripe_sub_id: str, stripe_customer_id: str | None = None, ) -> None: """Fail closed when a paid author has no active ClientAccess grant.""" if not await self.require_active_subscription(user_id): await self._ensure_stripe_client_access( author_id=user_id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) if not await self._access.get_active_for_author(user_id): raise RuntimeError("Subscription grant missing after Stripe payment") async def activate_from_session( self, stripe_session_id: str, password_hash: str, full_name: str | None = None, ) -> dict: """Create or upgrade User + ClientAccess from a verified Stripe Checkout session. Paid checkout always ends with the author able to log in with their chosen password. """ session = await get_checkout_session(stripe_session_id) if session.payment_status != "paid": raise ValueError(f"Payment not completed. Status: {session.payment_status}") customer_email = _customer_email_from_session(session) stripe_customer_id = _stripe_resource_id(session.customer) stripe_sub_id = _stripe_resource_id(session.subscription) plan_id, token_budget = await self._resolve_plan( _stripe_metadata_value(session.metadata, "plan_id", "starter") ) customer_name = full_name or getattr(session.customer_details, "name", "") or "" existing_user = await self._find_user_for_stripe_activation( customer_email, stripe_session_id, stripe_customer_id, stripe_sub_id, ) try: if existing_user: result = await self._complete_existing_user_activation( existing_user, stripe_session_id=stripe_session_id, password_hash=password_hash, full_name=full_name, plan_id=plan_id, token_budget=token_budget, stripe_customer_id=stripe_customer_id, stripe_sub_id=stripe_sub_id, ) await self._verify_post_activation( result["user_id"], plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) return result user = await self._create_author_for_paid_session( email=customer_email, password_hash=password_hash, full_name=customer_name, plan_id=plan_id, token_budget=token_budget, stripe_customer_id=stripe_customer_id, stripe_sub_id=stripe_sub_id, stripe_session_id=stripe_session_id, ) self._queue_credentials_email( email=customer_email, name=customer_name or customer_email, temp_password=None, user_chose_password=True, ) logger.info("Author account activated from Stripe", user_id=user.id, plan_id=plan_id) result = { "user_id": user.id, "email": customer_email, "plan_id": plan_id, "already_activated": False, } await self._verify_post_activation( result["user_id"], plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) return result except IntegrityError: await self._db.rollback() logger.warning("Stripe activation race — linking existing account", email=customer_email) existing_user = await self._find_user_for_stripe_activation( customer_email, stripe_session_id, stripe_customer_id, stripe_sub_id, ) if existing_user: result = await self._complete_existing_user_activation( existing_user, stripe_session_id=stripe_session_id, password_hash=password_hash, full_name=full_name, plan_id=plan_id, token_budget=token_budget, stripe_customer_id=stripe_customer_id, stripe_sub_id=stripe_sub_id, ) else: result = await self._force_password_login( customer_email=customer_email, password_hash=password_hash, full_name=full_name, stripe_session_id=stripe_session_id, stripe_customer_id=stripe_customer_id, stripe_sub_id=stripe_sub_id, plan_id=plan_id, token_budget=token_budget, full_name_for_create=customer_name, ) await self._verify_post_activation( result["user_id"], plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) return result except Exception as exc: await self._db.rollback() logger.warning("Stripe activation fallback", email=customer_email, error=str(exc)[:200]) result = await self._force_password_login( customer_email=customer_email, password_hash=password_hash, full_name=full_name, stripe_session_id=stripe_session_id, stripe_customer_id=stripe_customer_id, stripe_sub_id=stripe_sub_id, plan_id=plan_id, token_budget=token_budget, full_name_for_create=customer_name, ) await self._verify_post_activation( result["user_id"], plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, stripe_customer_id=stripe_customer_id, ) return result async def handle_embed_unlock_payment(self, stripe_session_id: str) -> dict | None: """Grant one embed unlock credit after paid one-time Checkout (R-202). Idempotent by Stripe session id — safe for webhook + browser confirm. """ from datetime import datetime, timezone from sqlalchemy.exc import IntegrityError from app.models.stripe_event import StripeEvent from app.models.user import User marker_id = f"embed_unlock:{stripe_session_id}" async def _credits_for(author_id: str) -> dict: user = await self._db.get(User, author_id) if author_id else None return { "author_id": author_id, "embed_unlock_credits": int(user.embed_unlock_credits) if user else 0, "already_processed": True, } existing = await self._db.get(StripeEvent, marker_id) if existing and existing.status == "processed": author_id = "" try: import json payload = json.loads(existing.raw_payload or "{}") author_id = str(payload.get("author_id") or "") except Exception: pass return await _credits_for(author_id) session = await get_checkout_session(stripe_session_id) if session.payment_status != "paid": logger.warning("embed_unlock skipped — not paid", session_id=stripe_session_id) return None mode = getattr(session, "mode", None) or "" if mode and mode != "payment": logger.warning("embed_unlock skipped — wrong mode", mode=mode) return None addon = _stripe_metadata_value(session.metadata, "addon", "") if addon != "embed_unlock": return None author_id = _stripe_metadata_value(session.metadata, "author_id", "") if not author_id: logger.warning("embed_unlock missing author_id", session_id=stripe_session_id) return None # Claim idempotency lock BEFORE granting (webhook + browser confirm race) import json if not existing: try: self._db.add( StripeEvent( id=marker_id, event_type="embed_unlock.payment", related_object_id=stripe_session_id, related_object_type="checkout.session", received_at=datetime.now(timezone.utc), processed_at=datetime.now(timezone.utc), status="processed", raw_payload=json.dumps( {"author_id": author_id, "addon": "embed_unlock"} ), ) ) await self._db.flush() except IntegrityError: await self._db.rollback() return await _credits_for(author_id) from app.services.embed_domain_service import EmbedDomainService total = await EmbedDomainService(self._db).apply_paid_unlock_credit(author_id) return {"author_id": author_id, "embed_unlock_credits": total} async def provision_from_checkout_webhook(self, stripe_session_id: str) -> dict | None: """Webhook fallback: auto-provision if user never completes /success page.""" session = await get_checkout_session(stripe_session_id) if session.payment_status != "paid": logger.warning("Webhook provision skipped — not paid", session_id=stripe_session_id) return None addon = _stripe_metadata_value(session.metadata, "addon", "") if addon == "embed_unlock": return await self.handle_embed_unlock_payment(stripe_session_id) customer_email = _customer_email_from_session(session) existing = await self._users.get_by_email(customer_email) stripe_customer_id = _stripe_resource_id(session.customer) stripe_sub_id = _stripe_resource_id(session.subscription) plan_id, token_budget = await self._resolve_plan( _stripe_metadata_value(session.metadata, "plan_id", "starter") ) if existing: existing.stripe_customer_id = existing.stripe_customer_id or stripe_customer_id existing.stripe_subscription_id = existing.stripe_subscription_id or stripe_sub_id existing.stripe_plan_id = existing.stripe_plan_id or plan_id if not existing.stripe_checkout_session_id: existing.stripe_checkout_session_id = stripe_session_id await self._ensure_stripe_client_access( author_id=existing.id, plan_id=plan_id, token_budget=token_budget, stripe_sub_id=stripe_sub_id, ) logger.info("Webhook ensured access for existing user", email=customer_email) return {"skipped": True, "user_id": existing.id, "access_ensured": True} temp_password = self._generate_temp_password() customer_name = getattr(session.customer_details, "name", "") or "" result = await self._provision_stripe_author( email=customer_email, password_hash=_hash_password(temp_password), full_name=customer_name, plan_id=plan_id, token_budget=token_budget, stripe_customer_id=stripe_customer_id, stripe_sub_id=stripe_sub_id, stripe_session_id=stripe_session_id, provisioned_via="stripe_webhook", activation_status="pending", password_change_required=True, ) user = await self._users.get_by_id(result["user_id"]) if user: user.credentials_email_sent_at = datetime.now(timezone.utc) await self._db.commit() self._queue_credentials_email( email=customer_email, name=customer_name or customer_email, temp_password=temp_password, user_chose_password=False, ) logger.info("Author auto-provisioned via webhook", user_id=result["user_id"]) return result async def handle_payment_succeeded(self, stripe_subscription_id: str) -> None: """Reset monthly token counter on successful invoice payment. Called by: invoice.payment_succeeded webhook. """ from sqlalchemy import select, update from app.models.client_access import ClientAccess result = await self._db.execute( select(ClientAccess).where( ClientAccess.stripe_subscription_id == stripe_subscription_id ) ) access = result.scalar_one_or_none() if not access: logger.warning("No ClientAccess for subscription", sub_id=stripe_subscription_id) return # Reset token counter + clear any grace period on the User access.tokens_used = 0 access.is_revoked = False from app.models.user import User user_result = await self._db.execute( select(User).where( User.stripe_subscription_id == stripe_subscription_id ) ) user = user_result.scalar_one_or_none() if user: user.payment_grace_until = None await self._db.commit() logger.info("Token counter reset on payment success", sub_id=stripe_subscription_id) async def handle_payment_failed( self, stripe_subscription_id: str, grace_days: int = 7, ) -> None: """Set grace period on payment failure — access continues for grace_days. Called by: invoice.payment_failed webhook. The dunning email is sent by the webhook handler separately. """ from sqlalchemy import select from app.models.user import User result = await self._db.execute( select(User).where(User.stripe_subscription_id == stripe_subscription_id) ) user = result.scalar_one_or_none() if user: user.payment_grace_until = datetime.now(timezone.utc) + timedelta(days=grace_days) await self._db.commit() logger.info( "Payment grace period set", user_id=user.id, grace_until=user.payment_grace_until, ) async def handle_subscription_cancelled(self, stripe_subscription_id: str) -> None: """Deactivate access when Stripe subscription is fully cancelled. Called by: customer.subscription.deleted webhook. """ from sqlalchemy import select from app.models.client_access import ClientAccess result = await self._db.execute( select(ClientAccess).where( ClientAccess.stripe_subscription_id == stripe_subscription_id ) ) access = result.scalar_one_or_none() if access: access.is_revoked = True access.revoked_at = datetime.now(timezone.utc) access.revoke_reason = "stripe_subscription_cancelled" await self._db.commit() logger.info("Access revoked on subscription cancellation", sub_id=stripe_subscription_id) async def handle_plan_changed( self, stripe_subscription_id: str, new_plan_id: str, ) -> None: """Update plan limits when author upgrades or downgrades. Called by: customer.subscription.updated webhook. """ from sqlalchemy import select from app.models.client_access import ClientAccess from app.models.pricing_plan import PricingPlan from app.models.user import User # Get new plan limits plan_result = await self._db.execute( select(PricingPlan).where(PricingPlan.id == new_plan_id) ) plan = plan_result.scalar_one_or_none() if not plan: logger.warning("Unknown plan on subscription update", plan_id=new_plan_id) return # Update ClientAccess budget access_result = await self._db.execute( select(ClientAccess).where( ClientAccess.stripe_subscription_id == stripe_subscription_id ) ) access = access_result.scalar_one_or_none() if access: access.plan = new_plan_id access.plan_id = new_plan_id access.token_budget = plan.max_token_budget # Update User plan reference user_result = await self._db.execute( select(User).where(User.stripe_subscription_id == stripe_subscription_id) ) user = user_result.scalar_one_or_none() if user: user.stripe_plan_id = new_plan_id await self._db.commit() logger.info( "Plan updated on subscription change", sub_id=stripe_subscription_id, new_plan=new_plan_id, ) async def _load_active_plan(self, plan_id: str): """Load an active PricingPlan or raise ValueError.""" from sqlalchemy import select from app.models.pricing_plan import PricingPlan result = await self._db.execute( select(PricingPlan).where( PricingPlan.id == plan_id, PricingPlan.is_active == True, # noqa: E712 ) ) plan = result.scalar_one_or_none() if not plan: raise ValueError(f"Plan '{plan_id}' not found or inactive") return plan def _stripe_price_for_billing(self, plan, billing: str) -> str: """Return Stripe Price ID for monthly/annual billing.""" price_id = ( plan.stripe_price_id_monthly if billing == "monthly" else plan.stripe_price_id_annual ) if not price_id: raise ValueError( f"Plan '{plan.id}' is not configured for {billing} billing yet." ) return price_id async def _stripe_subscription_is_active(self, subscription_id: str) -> bool: """True when Stripe reports the subscription as active or trialing.""" try: _refresh_stripe_key() sub = stripe.Subscription.retrieve(subscription_id) status = getattr(sub, "status", None) or "" return status in ("active", "trialing", "past_due") except Exception as exc: logger.warning( "stripe_subscription_status_check_failed", sub_id=subscription_id, error=str(exc)[:120], ) return False async def start_author_plan_action( self, author, plan_id: str, billing: str = "monthly", renew: bool = False, redis=None, ) -> dict: """Subscribe, renew/refill, or change plan for a logged-in author (R-184–R-188). Returns either a Checkout redirect or an in-place upgrade result. """ from app.core.public_url import resolve_api_public_base_url from app.services.token_budget import percent_used, tokens_remaining plan = await self._load_active_plan(plan_id) stripe_price_id = self._stripe_price_for_billing(plan, billing) access = await self._access.get_active_for_author(author.id) current_plan = ( (author.stripe_plan_id or "") or (getattr(access, "plan_id", None) or "") or (getattr(access, "plan", None) or "") ).strip().lower() target = plan_id.strip().lower() same_plan = bool(current_plan) and ( current_plan == target or current_plan == plan.display_name.strip().lower() ) needs_refill = False if access: needs_refill = tokens_remaining(access) <= 0 or percent_used(access) >= 80.0 if renew or (same_plan and needs_refill): return await self._start_plan_renew_refill_checkout( author=author, plan=plan, billing=billing, ) sub_id = (author.stripe_subscription_id or "").strip() if sub_id and await self._stripe_subscription_is_active(sub_id): if same_plan: raise ValueError( "You are already on this plan. When your token budget runs low, " "use Renew & refill tokens on Billing." ) await upgrade_subscription(sub_id, stripe_price_id, new_plan_id=plan_id) await self.handle_plan_changed(sub_id, plan_id) return { "action": "upgraded", "plan_id": plan_id, "plan_name": plan.display_name, "billing": billing, } base = resolve_api_public_base_url() success_url = ( f"{base}/author?page=billing&checkout=success" f"&session_id={{CHECKOUT_SESSION_ID}}" ) cancel_url = f"{base}/author?page=billing&checkout=cancel" checkout_url = await create_checkout_session( plan_id=plan_id, stripe_price_id=stripe_price_id, success_url=success_url, cancel_url=cancel_url, customer_email=None if author.stripe_customer_id else author.email, existing_customer_id=author.stripe_customer_id, author_id=author.id, ) return { "action": "checkout", "checkout_url": checkout_url, "plan_id": plan_id, "plan_name": plan.display_name, "billing": billing, } def _renew_refill_price_usd(self, plan, billing: str) -> float: """USD charge for one renew/refill cycle.""" if billing == "annual": # annual_price_usd is per-month when billed annually return float(plan.annual_price_usd or 0) * 12 return float(plan.monthly_price_usd or 0) async def _start_plan_renew_refill_checkout( self, author, plan, billing: str = "monthly", ) -> dict: """One-time Checkout to refill token budget for the current (or chosen) plan.""" from app.core.public_url import resolve_api_public_base_url price_usd = self._renew_refill_price_usd(plan, billing) if price_usd < 0.5: raise ValueError("This plan has no renew price configured yet.") base = resolve_api_public_base_url() success_url = ( f"{base}/author?page=billing&checkout=success" f"&session_id={{CHECKOUT_SESSION_ID}}" ) cancel_url = f"{base}/author?page=billing&checkout=cancel" period = "annual" if billing == "annual" else "monthly" checkout_url = await create_addon_checkout_session( addon_id="plan_renew_refill", price_usd=price_usd, display_name=f"{plan.display_name} — renew & refill tokens", author_id=author.id, success_url=success_url, cancel_url=cancel_url, customer_email=None if author.stripe_customer_id else author.email, existing_customer_id=author.stripe_customer_id, description=( f"Refills your {plan.display_name} token budget for one {period} period." ), extra_metadata={"plan_id": plan.id, "billing": billing}, ) return { "action": "checkout", "checkout_url": checkout_url, "plan_id": plan.id, "plan_name": plan.display_name, "billing": billing, "renew_refill": True, } async def apply_plan_renew_refill( self, author_id: str, plan_id: str, billing: str = "monthly", redis=None, ) -> dict: """Reset token usage and extend access after a paid renew/refill (R-188).""" from datetime import timedelta from app.services.token_budget import clear_budget_exhausted plan = await self._load_active_plan(plan_id) access = await self._access.get_active_for_author(author_id) if not access: from sqlalchemy import select from app.models.client_access import ClientAccess result = await self._db.execute( select(ClientAccess) .where(ClientAccess.author_id == author_id) .order_by(ClientAccess.granted_at.desc()) .limit(1) ) access = result.scalar_one_or_none() if not access: raise ValueError("No subscription found to refill. Subscribe to a plan first.") now = datetime.now(timezone.utc) days = 365 if billing == "annual" else 30 exp = access.expires_at if exp is not None and exp.tzinfo is None: exp = exp.replace(tzinfo=timezone.utc) base = exp if exp and exp > now else now access.tokens_used = 0 access.is_revoked = False access.revoked_at = None access.revoke_reason = None access.plan = plan_id access.plan_id = plan_id access.token_budget = plan.max_token_budget access.expires_at = base + timedelta(days=days) user = await self._users.get_by_id(author_id) if user: user.stripe_plan_id = plan_id user.payment_grace_until = None user.chatbot_is_active = True await self._db.commit() if redis is not None: try: await clear_budget_exhausted(redis, author_id) except Exception as exc: logger.warning( "clear_budget_exhausted_failed", author_id=author_id, error=str(exc)[:120], ) logger.info( "Plan renew refill applied", author_id=author_id, plan_id=plan_id, billing=billing, ) return { "ok": True, "plan_id": plan_id, "tokens_remaining": plan.max_token_budget, "expires_at": access.expires_at.isoformat() if access.expires_at else None, } async def handle_plan_renew_refill_payment(self, stripe_session_id: str, redis=None) -> dict | None: """Apply renew/refill after paid one-time Checkout (webhook or confirm).""" session = await get_checkout_session(stripe_session_id) if getattr(session, "payment_status", None) != "paid": return None meta = getattr(session, "metadata", None) addon = _stripe_metadata_value(meta, "addon", "") if addon != "plan_renew_refill": return None author_id = _stripe_metadata_value(meta, "author_id", "") plan_id = _stripe_metadata_value(meta, "plan_id", "starter") billing = _stripe_metadata_value(meta, "billing", "monthly") if not author_id: logger.warning("plan_renew_refill missing author_id", session_id=stripe_session_id) return None return await self.apply_plan_renew_refill( author_id=author_id, plan_id=plan_id, billing=billing if billing in ("monthly", "annual") else "monthly", redis=redis, ) async def confirm_author_checkout(self, author, session_id: str, redis=None) -> dict: """Apply a paid Checkout session to the logged-in author and repair grant.""" session = await get_checkout_session(session_id) if getattr(session, "payment_status", None) != "paid": raise ValueError("Payment not completed yet. Please wait a moment and refresh.") meta_author = _stripe_metadata_value(session.metadata, "author_id", "") if meta_author and meta_author != author.id: raise ValueError("This payment does not belong to your account.") addon = _stripe_metadata_value(session.metadata, "addon", "") if addon == "plan_renew_refill": result = await self.handle_plan_renew_refill_payment(session_id, redis=redis) if not result: raise ValueError("Could not apply renew payment.") access = await self._access.get_active_for_author(author.id) fresh = await self._users.get_by_id(author.id) return { "ok": True, "has_active_subscription": access is not None, "plan": access.plan if access else (fresh.stripe_plan_id if fresh else None), "stripe_customer_id": fresh.stripe_customer_id if fresh else None, "renew_refill": True, } customer_email = "" try: customer_email = _customer_email_from_session(session) except ValueError: customer_email = "" if customer_email and customer_email.lower() != (author.email or "").lower(): cust_id = "" try: cust_id = _stripe_resource_id(session.customer) except ValueError: cust_id = "" if not meta_author and cust_id != (author.stripe_customer_id or ""): raise ValueError("This payment email does not match your account.") await self.provision_from_checkout_webhook(session_id) await self.require_active_subscription(author.id) access = await self._access.get_active_for_author(author.id) fresh = await self._users.get_by_id(author.id) return { "ok": True, "has_active_subscription": access is not None, "plan": access.plan if access else (fresh.stripe_plan_id if fresh else None), "stripe_customer_id": fresh.stripe_customer_id if fresh else None, }