""" Stripe Payment Service Handles subscription payments for pounce.ch Subscription Tiers: - Scout (Free): $0/month - Trader: $9/month - Tycoon: $29/month Environment Variables Required: - STRIPE_SECRET_KEY: Stripe API secret key - STRIPE_WEBHOOK_SECRET: Webhook signing secret - STRIPE_PRICE_TRADER: Price ID for Trader plan - STRIPE_PRICE_TYCOON: Price ID for Tycoon plan """ import logging import os from typing import Optional, Dict, Any from datetime import datetime import stripe from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.user import User from app.models.subscription import Subscription, SubscriptionTier, SubscriptionStatus logger = logging.getLogger(__name__) # Initialize Stripe with API key from environment stripe.api_key = os.getenv("STRIPE_SECRET_KEY") # Price IDs from Stripe Dashboard STRIPE_PRICES = { "trader": os.getenv("STRIPE_PRICE_TRADER"), "tycoon": os.getenv("STRIPE_PRICE_TYCOON"), } # Subscription tier features TIER_FEATURES = { "scout": { "name": "Scout", "price": 0, "currency": "USD", "max_domains": 5, "check_frequency": "daily", "portfolio_domains": 0, "features": ["Basic monitoring", "Daily checks", "Email alerts"], }, "trader": { "name": "Trader", "price": 9, "currency": "USD", "max_domains": 50, "check_frequency": "hourly", "portfolio_domains": 25, "features": [ "50 domain monitoring", "Hourly checks", "Portfolio tracking (25 domains)", "Domain valuation", "Market insights", "SMS/Telegram alerts", ], }, "tycoon": { "name": "Tycoon", "price": 29, "currency": "USD", "max_domains": 500, "check_frequency": "realtime", "portfolio_domains": -1, # Unlimited "features": [ "500+ domain monitoring", "10-minute checks", "Unlimited portfolio", "API access", "Bulk tools", "SEO metrics", "Priority support", "Webhooks", ], }, } class StripeService: """ Handles Stripe payment operations. Usage: 1. Create checkout session for user 2. User completes payment on Stripe 3. Webhook updates subscription status """ @staticmethod def is_configured() -> bool: """Check if Stripe is properly configured.""" return bool(stripe.api_key) @staticmethod async def create_checkout_session( user: User, plan: str, success_url: str, cancel_url: str, ) -> Dict[str, Any]: """ Create a Stripe Checkout session for subscription. Args: user: User subscribing plan: "trader" or "tycoon" success_url: URL to redirect after successful payment cancel_url: URL to redirect if user cancels Returns: Dict with checkout_url and session_id """ if not StripeService.is_configured(): raise ValueError("Stripe is not configured. Set STRIPE_SECRET_KEY environment variable.") if plan not in ["trader", "tycoon"]: raise ValueError(f"Invalid plan: {plan}. Must be 'trader' or 'tycoon'") price_id = STRIPE_PRICES.get(plan) if not price_id: raise ValueError(f"Price ID not configured for plan: {plan}") try: # Create or get Stripe customer if user.stripe_customer_id: customer_id = user.stripe_customer_id else: customer = stripe.Customer.create( email=user.email, name=user.name or user.email, metadata={ "user_id": str(user.id), "pounce_user": "true", } ) customer_id = customer.id # Note: Save customer_id to user in calling code # Create checkout session session = stripe.checkout.Session.create( customer=customer_id, payment_method_types=["card"], line_items=[ { "price": price_id, "quantity": 1, } ], mode="subscription", success_url=success_url, cancel_url=cancel_url, metadata={ "user_id": str(user.id), "plan": plan, }, subscription_data={ "metadata": { "user_id": str(user.id), "plan": plan, } }, ) return { "checkout_url": session.url, "session_id": session.id, "customer_id": customer_id, } except stripe.error.StripeError as e: logger.error(f"Stripe error creating checkout: {e}") raise @staticmethod async def create_portal_session( customer_id: str, return_url: str, ) -> str: """ Create a Stripe Customer Portal session for managing subscription. Users can: - Update payment method - View invoices - Cancel subscription """ if not StripeService.is_configured(): raise ValueError("Stripe is not configured") try: session = stripe.billing_portal.Session.create( customer=customer_id, return_url=return_url, ) return session.url except stripe.error.StripeError as e: logger.error(f"Stripe error creating portal session: {e}") raise @staticmethod async def handle_webhook( payload: bytes, sig_header: str, db: AsyncSession, ) -> Dict[str, Any]: """ Handle Stripe webhook events. Important events: - checkout.session.completed: Payment successful - customer.subscription.updated: Subscription changed - customer.subscription.deleted: Subscription cancelled - invoice.payment_failed: Payment failed """ webhook_secret = os.getenv("STRIPE_WEBHOOK_SECRET") if not webhook_secret: raise ValueError("STRIPE_WEBHOOK_SECRET not configured") try: event = stripe.Webhook.construct_event( payload, sig_header, webhook_secret ) except ValueError: raise ValueError("Invalid payload") except stripe.error.SignatureVerificationError: raise ValueError("Invalid signature") event_type = event["type"] data = event["data"]["object"] logger.info(f"Processing Stripe webhook: {event_type}") if event_type == "checkout.session.completed": await StripeService._handle_checkout_complete(data, db) elif event_type == "customer.subscription.updated": await StripeService._handle_subscription_updated(data, db) elif event_type == "customer.subscription.deleted": await StripeService._handle_subscription_cancelled(data, db) elif event_type == "invoice.payment_failed": await StripeService._handle_payment_failed(data, db) return {"status": "success", "event_type": event_type} @staticmethod async def _handle_checkout_complete(data: Dict, db: AsyncSession): """Handle successful checkout - activate subscription.""" user_id = data.get("metadata", {}).get("user_id") plan = data.get("metadata", {}).get("plan") # "trader" or "tycoon" customer_id = data.get("customer") subscription_id = data.get("subscription") if not user_id or not plan: logger.error("Missing user_id or plan in checkout metadata") return # Convert plan string to SubscriptionTier enum tier_map = { "trader": SubscriptionTier.TRADER, "tycoon": SubscriptionTier.TYCOON, "scout": SubscriptionTier.SCOUT, } tier_enum = tier_map.get(plan.lower(), SubscriptionTier.SCOUT) # Get user result = await db.execute( select(User).where(User.id == int(user_id)) ) user = result.scalar_one_or_none() if not user: logger.error(f"User {user_id} not found for checkout") return # Update user's Stripe customer ID user.stripe_customer_id = customer_id # Create or update subscription sub_result = await db.execute( select(Subscription).where(Subscription.user_id == user.id) ) subscription = sub_result.scalar_one_or_none() tier_info = TIER_FEATURES.get(plan.lower(), TIER_FEATURES["scout"]) if subscription: subscription.tier = tier_enum # Use enum, not string subscription.status = SubscriptionStatus.ACTIVE subscription.stripe_subscription_id = subscription_id subscription.max_domains = tier_info["max_domains"] subscription.check_frequency = tier_info["check_frequency"] subscription.updated_at = datetime.utcnow() else: subscription = Subscription( user_id=user.id, tier=tier_enum, # Use enum, not string status=SubscriptionStatus.ACTIVE, stripe_subscription_id=subscription_id, max_domains=tier_info["max_domains"], check_frequency=tier_info["check_frequency"], ) db.add(subscription) await db.commit() logger.info(f"Activated {plan} subscription for user {user_id}") @staticmethod async def _handle_subscription_updated(data: Dict, db: AsyncSession): """Handle subscription update (plan change, renewal, etc.).""" subscription_id = data.get("id") status = data.get("status") result = await db.execute( select(Subscription).where( Subscription.stripe_subscription_id == subscription_id ) ) subscription = result.scalar_one_or_none() if subscription: subscription.is_active = status == "active" subscription.updated_at = datetime.utcnow() await db.commit() logger.info(f"Updated subscription {subscription_id}: status={status}") @staticmethod async def _handle_subscription_cancelled(data: Dict, db: AsyncSession): """Handle subscription cancellation - downgrade to Scout.""" subscription_id = data.get("id") result = await db.execute( select(Subscription).where( Subscription.stripe_subscription_id == subscription_id ) ) subscription = result.scalar_one_or_none() if subscription: subscription.tier = SubscriptionTier.SCOUT # Use enum subscription.status = SubscriptionStatus.ACTIVE # Scout is still active subscription.stripe_subscription_id = None subscription.max_domains = TIER_FEATURES["scout"]["max_domains"] subscription.check_frequency = TIER_FEATURES["scout"]["check_frequency"] subscription.updated_at = datetime.utcnow() await db.commit() logger.info(f"Cancelled subscription, downgraded to Scout: {subscription_id}") @staticmethod async def _handle_payment_failed(data: Dict, db: AsyncSession): """Handle failed payment - send notification, eventually downgrade.""" subscription_id = data.get("subscription") # Log the failure - in production, send email notification logger.warning(f"Payment failed for subscription: {subscription_id}") # After multiple failures, Stripe will cancel the subscription # which triggers _handle_subscription_cancelled # Global instance stripe_service = StripeService()