"""Stripe Billing routes — Checkout, Webhooks, Portal, Subscription Status.
Handles:
- POST /api/billing/checkout-session — creates Stripe Checkout session
- GET /api/billing/subscription — returns current subscription status
- POST /api/billing/portal-session — creates Stripe Customer Portal session
- POST /api/billing/webhook — handles Stripe events
"""
import os
import stripe
from datetime import datetime, timezone
from flask import Blueprint, request, jsonify
from flask_login import current_user, login_required
import logging
from app.models import db, Company, Subscription, UserCompany, StripeWebhookEvent
from app.utils.csrf import require_csrf
logger = logging.getLogger(__name__)
billing_bp = Blueprint('billing', __name__)
# ─── Stripe Config ──────────────────────────────────────────────────────────
STRIPE_SECRET_KEY = os.environ.get('STRIPE_SECRET_KEY', '')
STRIPE_WEBHOOK_SECRET = os.environ.get('STRIPE_WEBHOOK_SECRET', '')
STRIPE_PUBLISHABLE_KEY = os.environ.get('STRIPE_PUBLISHABLE_KEY', '')
FRONTEND_URL = os.environ.get('FRONTEND_URL', 'http://localhost:5003')
# Map plan IDs to Stripe Price IDs
# These MUST match the Price IDs created in your Stripe Dashboard
# Update these after creating the prices in Stripe
PRICE_MAP = {
'launch': os.environ.get('STRIPE_PRICE_LAUNCH', 'price_launch'),
'growth': os.environ.get('STRIPE_PRICE_GROWTH', 'price_growth'),
'command': os.environ.get('STRIPE_PRICE_COMMAND', 'price_command'),
}
stripe.api_key = STRIPE_SECRET_KEY
# ─── Publishable Key ────────────────────────────────────────────────────────
@billing_bp.route('/api/billing/config', methods=['GET'])
@login_required
def billing_config():
"""Return Stripe publishable key, plan info, and seat usage for the frontend."""
from app.utils.seats import get_company_seat_usage
seat_usage = None
if current_user.is_authenticated:
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if uc:
seat_usage = get_company_seat_usage(uc.company_id)
return jsonify({
'publishableKey': STRIPE_PUBLISHABLE_KEY,
'plans': {
'launch': {'name': 'Launch', 'price': '$49/mo', 'seats': 5},
'growth': {'name': 'Growth', 'price': '$99/mo', 'seats': 25},
'command': {'name': 'Command', 'price': '$199/mo', 'seats': 100},
},
'seatUsage': seat_usage,
})
# ─── Tier / Feature Gate ────────────────────────────────────────────────────
@billing_bp.route('/api/billing/tier', methods=['GET'])
@login_required
def billing_tier():
"""Return the current user's effective tier, unlocked features, and limits.
Consumed by the frontend useFeatureGate hook.
"""
from app.utils.feature_gate import (
get_user_tier, TIER_FEATURES, get_tier_limits,
)
tier = get_user_tier()
return jsonify({
'tier': tier,
'features': sorted(TIER_FEATURES[tier]),
'limits': get_tier_limits(tier),
})
# ─── Helpers ────────────────────────────────────────────────────────────────
def get_user_company():
"""Return the current user's primary company or None."""
if not current_user.is_authenticated:
return None
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
return uc.company if uc else None
def ensure_stripe_customer(company):
"""Create or return Stripe Customer for the company."""
if company.stripe_customer_id:
return stripe.Customer.retrieve(company.stripe_customer_id)
customer = stripe.Customer.create(
name=company.name,
email=current_user.email,
metadata={'company_id': company.id},
)
company.stripe_customer_id = customer.id
db.session.commit()
return customer
# ─── Checkout Session ───────────────────────────────────────────────────────
@billing_bp.route('/api/billing/checkout-session', methods=['POST'])
@login_required
@require_csrf
def create_checkout_session():
"""Create a Stripe Checkout session for the requested plan.
Body: { "plan": "launch" | "growth" | "command" }
"""
data = request.get_json()
if not data or 'plan' not in data:
return jsonify({'error': 'Plan ID is required'}), 400
plan_id = data['plan']
if plan_id not in PRICE_MAP:
return jsonify({'error': f'Invalid plan: {plan_id}'}), 400
company = get_user_company()
if not company:
return jsonify({'error': 'No company found for user'}), 400
customer = ensure_stripe_customer(company)
line_items = [{
'price': PRICE_MAP[plan_id],
'quantity': 1,
}]
# H3: prevent trial abuse — no free trial if the company ever had a subscription
existing_sub = Subscription.query.filter_by(company_id=company.id).first()
trial_days = 0 if existing_sub else 7
subscription_data = {
'metadata': {'plan': plan_id, 'company_id': company.id},
}
if trial_days:
subscription_data['trial_period_days'] = trial_days
checkout_session = stripe.checkout.Session.create(
customer=customer.id,
payment_method_types=['card'],
line_items=line_items,
mode='subscription',
# expand line_items so we can read price.id in the webhook handler
expand=['line_items'],
subscription_data=subscription_data,
success_url=f'{FRONTEND_URL}/app?session_id={{CHECKOUT_SESSION_ID}}&status=success',
cancel_url=f'{FRONTEND_URL}/app?status=canceled',
metadata={
'plan': plan_id,
'company_id': company.id,
},
)
return jsonify({
'url': checkout_session.url,
'sessionId': checkout_session.id,
})
# ─── Subscription Status ────────────────────────────────────────────────────
@billing_bp.route('/api/billing/subscription', methods=['GET'])
@login_required
def get_subscription():
"""Return current subscription status for the user's company."""
company = get_user_company()
if not company:
return jsonify({'subscription': None})
subscription = Subscription.query.filter_by(company_id=company.id).first()
if not subscription:
return jsonify({
'subscription': {
'status': 'none',
'plan': 'Starter',
'is_active': False,
'current_period_end': None,
'cancel_at_period_end': False,
'trial_end': None,
}
})
return jsonify({
'subscription': {
'id': subscription.id,
'status': subscription.stripe_status,
'plan': subscription.plan_id,
'is_active': subscription.is_active,
'current_period_start': subscription.current_period_start.isoformat() if subscription.current_period_start else None,
'current_period_end': subscription.current_period_end.isoformat() if subscription.current_period_end else None,
'cancel_at_period_end': subscription.cancel_at_period_end,
'trial_end': subscription.trial_end.isoformat() if subscription.trial_end else None,
}
})
# ─── Customer Portal ────────────────────────────────────────────────────────
@billing_bp.route('/api/billing/portal-session', methods=['POST'])
@login_required
@require_csrf
def create_portal_session():
"""Create a Stripe Customer Portal session for managing billing."""
company = get_user_company()
if not company or not company.stripe_customer_id:
return jsonify({'error': 'No Stripe customer found'}), 400
portal_session = stripe.billing_portal.Session.create(
customer=company.stripe_customer_id,
return_url=f'{FRONTEND_URL}/app/billing',
)
return jsonify({
'url': portal_session.url,
})
# ─── Webhook ────────────────────────────────────────────────────────────────
@billing_bp.route('/api/billing/webhook', methods=['POST'])
def stripe_webhook():
"""Handle Stripe webhook events.
Verifies signature, then processes:
- checkout.session.completed → create subscription record
- customer.subscription.updated → update subscription record
- customer.subscription.deleted → mark subscription as canceled
"""
payload = request.data
sig_header = request.headers.get('Stripe-Signature', '')
try:
event = stripe.Webhook.construct_event(
payload, sig_header, STRIPE_WEBHOOK_SECRET
)
except (ValueError, stripe.error.SignatureVerificationError):
return jsonify({'error': 'Webhook signature verification failed'}), 400
# ─── Idempotency: skip events we've already processed ───────────
if db.session.get(StripeWebhookEvent, event.id):
return jsonify({'received': True, 'duplicate': True}), 200
try:
_process_stripe_event(event)
db.session.add(StripeWebhookEvent(id=event.id, event_type=event.type))
db.session.commit()
except Exception:
db.session.rollback()
logger.exception('Stripe webhook processing failed for event %s', event.id)
return jsonify({'error': 'Webhook processing failed'}), 500
return jsonify({'received': True}), 200
def _is_stale_event(sub, event) -> bool:
"""Ordering guard: True if this event is older than the last one applied.
Stripe delivers webhooks at-least-once and out of order. The event-ID
dedupe table catches exact retries; this catches *different* events that
arrive late (e.g. an old subscription.updated after a newer one).
"""
created = getattr(event, 'created', None)
if created is None or sub is None or sub.last_event_created is None:
return False
return int(created) < int(sub.last_event_created)
def _mark_event_applied(sub, event) -> None:
created = getattr(event, 'created', None)
if created is not None and sub is not None:
sub.last_event_created = int(created)
def _process_stripe_event(event):
"""Process a verified Stripe event. Raises on failure (caller rolls back)."""
# ─── Checkout session completed ─────────────────────────────────
if event.type == 'checkout.session.completed':
session_data = event.data.object
metadata = session_data.get('metadata', {})
company_id = metadata.get('company_id')
plan = metadata.get('plan', 'launch')
# Never trust metadata for tier assignment beyond known plans
if plan not in PRICE_MAP:
logger.warning('checkout.session.completed with unknown plan %r — defaulting to launch', plan)
plan = 'launch'
if company_id:
company = db.session.get(Company, company_id)
if company:
if not company.stripe_customer_id:
company.stripe_customer_id = session_data.customer
db.session.commit()
# ── Capture the Stripe price_id from the checkout line items ──
price_id = ''
try:
expanded = stripe.checkout.Session.retrieve(
session_data.id,
expand=['line_items'],
)
items = expanded.line_items.data if expanded.line_items else []
if items:
price_id = items[0].price.id
except Exception:
# Fall back to metadata plan if expansion fails
price_id = ''
sub = Subscription.query.filter_by(
company_id=company_id
).first()
if not sub:
sub = Subscription(
company_id=company_id,
stripe_subscription_id=session_data.subscription,
stripe_price_id=price_id,
stripe_status='trialing',
)
db.session.add(sub)
else:
# Idempotency: if this checkout was already applied
# (same Stripe subscription), don't re-apply tier changes.
if _is_stale_event(sub, event):
logger.info('Skipping stale checkout.session.completed for company %s', company_id)
return
if price_id:
sub.stripe_price_id = price_id
if session_data.subscription:
sub.stripe_subscription_id = session_data.subscription
# Sync company tier with the plan from metadata
settings = company.settings_json or {}
settings['tier'] = plan
company.settings_json = settings
_mark_event_applied(sub, event)
db.session.commit()
# ─── Subscription updated ───────────────────────────────────────
elif event.type == 'customer.subscription.updated':
sub_data = event.data.object
sub = Subscription.query.filter_by(
stripe_subscription_id=sub_data.id
).first()
if sub:
if _is_stale_event(sub, event):
logger.info('Skipping stale customer.subscription.updated for sub %s', sub_data.id)
return
sub.stripe_status = sub_data.status
sub.cancel_at_period_end = sub_data.cancel_at_period_end
sub.current_period_start = datetime.fromtimestamp(
sub_data.current_period_start, tz=timezone.utc
)
sub.current_period_end = datetime.fromtimestamp(
sub_data.current_period_end, tz=timezone.utc
)
if sub_data.get('trial_end'):
new_trial_end = datetime.fromtimestamp(
sub_data.trial_end, tz=timezone.utc
)
# Trial-manipulation guard: a trial may end earlier or stay
# the same, but never be extended past what we first recorded.
existing = sub.trial_end
if existing is not None and existing.tzinfo is None:
existing = existing.replace(tzinfo=timezone.utc)
if existing is not None and new_trial_end > existing:
logger.warning(
'Ignoring trial_end extension for sub %s (%s -> %s)',
sub_data.id, existing.isoformat(), new_trial_end.isoformat(),
)
else:
sub.trial_end = new_trial_end
if sub_data.get('items', {}).get('data'):
item = sub_data.items.data[0]
sub.stripe_price_id = item.price.id
_mark_event_applied(sub, event)
db.session.commit()
# ─── Subscription deleted ───────────────────────────────────────
elif event.type == 'customer.subscription.deleted':
sub_data = event.data.object
sub = Subscription.query.filter_by(
stripe_subscription_id=sub_data.id
).first()
if sub:
if _is_stale_event(sub, event):
logger.info('Skipping stale customer.subscription.deleted for sub %s', sub_data.id)
return
sub.stripe_status = 'canceled'
sub.current_period_end = datetime.fromtimestamp(
sub_data.current_period_end, tz=timezone.utc
)
_mark_event_applied(sub, event)
db.session.commit()
# ─── Invoice payment failed ─────────────────────────────────────
elif event.type == 'invoice.payment_failed':
invoice = event.data.object
sub_id = invoice.get('subscription')
if sub_id:
sub = Subscription.query.filter_by(
stripe_subscription_id=sub_id
).first()
if sub:
sub.stripe_status = 'past_due'
db.session.commit()
# ─── Invoice payment succeeded ──────────────────────────────────
elif event.type == 'invoice.payment_succeeded':
pass