""" Stripe billing integration for ModuleDesk. Handles subscription creation/updates and webhook event processing. All Stripe calls are optional at runtime — if STRIPE_SECRET_KEY is not set, billing operations are no-ops and the app degrades gracefully. """ from __future__ import annotations import datetime as dt import json import logging import os from typing import Optional logger = logging.getLogger(__name__) STRIPE_SECRET_KEY = os.getenv("STRIPE_SECRET_KEY") STRIPE_WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET") # Stripe price IDs — set via env vars so they can differ between test and live STRIPE_PRICE_BASIC = os.getenv("STRIPE_PRICE_BASIC") # recurring per-module Basic price STRIPE_PRICE_PRO = os.getenv("STRIPE_PRICE_PRO") # recurring per-module Pro price def _stripe(): """Lazily import stripe and configure the API key.""" import stripe as _s _s.api_key = STRIPE_SECRET_KEY return _s def create_stripe_customer(org) -> Optional[str]: """ Create a Stripe customer for the org. Returns the Stripe customer ID. Stores it back on org.stripe_customer_id (caller must commit). No-op if STRIPE_SECRET_KEY is not configured. """ if not STRIPE_SECRET_KEY: logger.warning("STRIPE_SECRET_KEY not set; skipping Stripe customer creation") return None try: s = _stripe() customer = s.Customer.create( email=_get_org_email(org), name=org.name, metadata={"org_id": str(org.id), "slug": org.slug}, ) org.stripe_customer_id = customer.id logger.info("Created Stripe customer %s for org %s", customer.id, org.id) return customer.id except Exception as exc: logger.error("Failed to create Stripe customer for org %s: %s", org.id, exc) return None def create_subscription(org, plan: str, module_count: int) -> Optional[str]: """ Create a Stripe subscription for the org. plan: 'basic' or 'pro' module_count: number of billable modules (Stripe subscription quantity). Returns the Stripe subscription ID. """ if not STRIPE_SECRET_KEY: return None price_id = STRIPE_PRICE_BASIC if plan == "basic" else STRIPE_PRICE_PRO if not price_id: logger.error("STRIPE_PRICE_%s not configured", plan.upper()) return None if not org.stripe_customer_id: create_stripe_customer(org) try: s = _stripe() sub = s.Subscription.create( customer=org.stripe_customer_id, items=[{"price": price_id, "quantity": max(1, module_count)}], metadata={"org_id": str(org.id), "plan": plan}, expand=["latest_invoice.payment_intent"], ) org.stripe_subscription_id = sub.id org.plan = plan logger.info("Created Stripe subscription %s for org %s (plan=%s, qty=%d)", sub.id, org.id, plan, module_count) return sub.id except Exception as exc: logger.error("Failed to create Stripe subscription for org %s: %s", org.id, exc) return None def update_subscription_module_count(org, module_count: int) -> bool: """ Update the subscription quantity to reflect the current billable module count. Called when a tenant toggles billing_active on a product. """ if not STRIPE_SECRET_KEY or not org.stripe_subscription_id: return False try: s = _stripe() sub = s.Subscription.retrieve(org.stripe_subscription_id) item_id = sub["items"]["data"][0]["id"] s.SubscriptionItem.modify(item_id, quantity=max(1, module_count)) logger.info("Updated subscription %s quantity to %d", org.stripe_subscription_id, module_count) return True except Exception as exc: logger.error("Failed to update subscription quantity for org %s: %s", org.id, exc) return False def handle_webhook(payload: bytes, sig_header: str) -> dict: """ Process an incoming Stripe webhook event. Returns a dict with 'status' and optional 'detail'. Caller (the Flask route) should return HTTP 200 regardless of our processing outcome to prevent Stripe from retrying, unless signature verification fails. """ if not STRIPE_WEBHOOK_SECRET: logger.warning("STRIPE_WEBHOOK_SECRET not set; skipping webhook verification") return {"status": "skip", "detail": "webhook secret not configured"} try: s = _stripe() event = s.Webhook.construct_event(payload, sig_header, STRIPE_WEBHOOK_SECRET) except Exception as exc: logger.error("Stripe webhook signature verification failed: %s", exc) return {"status": "invalid_signature"} event_type = event["type"] data = event["data"]["object"] logger.info("Stripe webhook received: %s", event_type) if event_type == "invoice.paid": return _handle_invoice_paid(data) elif event_type == "customer.subscription.deleted": return _handle_subscription_deleted(data) elif event_type == "customer.subscription.updated": return _handle_subscription_updated(data) return {"status": "ignored", "event_type": event_type} def _handle_invoice_paid(invoice: dict) -> dict: """Reset AI credits for the org when an invoice is paid.""" subscription_id = invoice.get("subscription") if not subscription_id: return {"status": "skip", "detail": "no subscription_id on invoice"} from ..db import session_scope from ..public_models import Organization from sqlalchemy import text with session_scope() as session: org = session.query(Organization).filter_by( stripe_subscription_id=subscription_id ).one_or_none() if not org: logger.warning("No org found for subscription %s", subscription_id) return {"status": "skip", "detail": "org not found"} org.ai_credits_used = 0 org.ai_credits_reset_at = dt.datetime.utcnow() logger.info("Reset AI credits for org %s (invoice paid)", org.id) return {"status": "ok", "action": "credits_reset", "org_id": org.id} def _handle_subscription_deleted(sub: dict) -> dict: """Downgrade org to free when subscription is cancelled.""" sub_id = sub.get("id") from ..db import session_scope from ..public_models import Organization with session_scope() as session: org = session.query(Organization).filter_by( stripe_subscription_id=sub_id ).one_or_none() if not org: return {"status": "skip", "detail": "org not found"} org.plan = "free" org.stripe_subscription_id = None logger.info("Downgraded org %s to free (subscription deleted)", org.id) return {"status": "ok", "action": "downgraded_to_free", "org_id": org.id} def _handle_subscription_updated(sub: dict) -> dict: """Sync plan field when subscription plan changes.""" sub_id = sub.get("id") metadata = sub.get("metadata", {}) new_plan = metadata.get("plan") if not new_plan: return {"status": "skip", "detail": "no plan in subscription metadata"} from ..db import session_scope from ..public_models import Organization with session_scope() as session: org = session.query(Organization).filter_by( stripe_subscription_id=sub_id ).one_or_none() if not org: return {"status": "skip", "detail": "org not found"} org.plan = new_plan logger.info("Updated org %s plan to %s", org.id, new_plan) return {"status": "ok", "action": "plan_updated", "plan": new_plan, "org_id": org.id} def _get_org_email(org) -> str: """Get the billing email for an org (the first admin user's email).""" try: from ..db import session_scope from ..public_models import User with session_scope() as session: admin = ( session.query(User) .filter_by(org_id=org.id, role="admin") .order_by(User.created_at) .first() ) return admin.email if admin else f"org-{org.id}@moduledesk.app" except Exception: return f"org-{org.id}@moduledesk.app"