"""Billing routes — upgrade page, Stripe webhook handler."""
import os
from flask import Blueprint, current_app, flash, jsonify, redirect, render_template, request, url_for
from app.crypto import hash_value
from app.models import (
STRIPE_PRICE_IDS,
STRIPE_WEBHOOK_SECRET,
TIERS,
get_user,
update_stripe_customer,
update_user_tier,
)
from app.routes.auth import current_user, login_required
billing_bp = Blueprint("billing", __name__, url_prefix="/billing")
@billing_bp.route("/upgrade")
@login_required
def upgrade():
"""Upgrade page — shows tier comparison and upgrade options."""
user = current_user()
# Build tier comparison
tiers = []
for key, config in TIERS.items():
tier_info = {
"key": key,
"name": config["name"],
"price": config["price"],
"max_sites": config["max_sites"],
"max_submissions": config["max_submissions"],
"is_current": key == user["tier"],
"can_upgrade": key != user["tier"] and STRIPE_PRICE_IDS.get(key),
"stripe_price_id": STRIPE_PRICE_IDS.get(key, ""),
}
# Add team-specific fields
if "max_members" in config:
tier_info["max_members"] = config["max_members"]
tiers.append(tier_info)
return render_template(
"upgrade.html",
user=user,
tiers=tiers,
stripe_configured=bool(STRIPE_PRICE_IDS.get("starter")),
)
@billing_bp.route("/stripe/webhook", methods=["POST"])
def stripe_webhook():
"""Handle Stripe webhook events.
Expects Stripe signature header for verification.
Idempotent — duplicate events are silently ignored via event_id dedup.
"""
if not STRIPE_WEBHOOK_SECRET:
return jsonify({"error": "Webhook secret not configured"}), 500
payload = request.get_data()
sig_header = request.headers.get("Stripe-Signature", "")
current_app.logger.info(
"stripe_webhook: received payload, len=%d, has_sig=%s",
len(payload),
bool(sig_header),
)
import stripe
stripe.api_key = os.environ.get("STRIPE_SECRET_KEY", "")
try:
event = stripe.Webhook.construct_event(payload, sig_header, STRIPE_WEBHOOK_SECRET)
except ValueError:
current_app.logger.warning(
"stripe_webhook: invalid payload, preview=%s",
payload[:200],
)
return jsonify({"error": "Invalid payload"}), 400
except stripe.error.SignatureVerificationError as e:
current_app.logger.warning(
"stripe_webhook: signature verification failed, error=%s, sig_present=%s",
str(e),
bool(sig_header),
)
return jsonify({"error": "Invalid signature"}), 400
current_app.logger.info(
"stripe_webhook: verified event, id=%s, type=%s, created=%s",
event.id,
event.type,
event.created,
)
# ─── Idempotency: skip already-processed events ────
if not _record_webhook_event(event.id):
current_app.logger.info(
"stripe_webhook: duplicate event, skipping, id=%s",
event.id,
)
return jsonify({"received": True, "dedup": True}), 200
# Handle events
if event.type == "checkout.session.completed":
session_data = event.data.object
_handle_checkout_completed(session_data)
elif event.type == "invoice.payment_succeeded":
invoice_data = event.data.object
_handle_invoice_payment_succeeded(invoice_data)
elif event.type == "customer.subscription.updated":
sub_data = event.data.object
_handle_subscription_updated(sub_data)
elif event.type == "customer.subscription.deleted":
sub_data = event.data.object
_handle_subscription_deleted(sub_data)
else:
current_app.logger.info(
"stripe_webhook: unhandled event type=%s, id=%s",
event.type,
event.id,
)
current_app.logger.info(
"stripe_webhook: event processed successfully, id=%s, type=%s",
event.id,
event.type,
)
return jsonify({"received": True}), 200
def _record_webhook_event(event_id):
"""Track processed Stripe event IDs to prevent duplicate processing.
Returns True if event was newly recorded, False if already seen.
"""
import time
from app.db import write_lock
from app.models import get_db
conn = None
with write_lock():
try:
conn = get_db()
# Create dedup table if needed
conn.execute("""
CREATE TABLE IF NOT EXISTS stripe_webhook_events (
event_id TEXT PRIMARY KEY,
processed_at INTEGER NOT NULL
)
""")
result = conn.execute(
"INSERT OR IGNORE INTO stripe_webhook_events (event_id, processed_at) VALUES (?, ?)",
(event_id, int(time.time())),
)
conn.commit()
return result.rowcount > 0
except Exception as e:
current_app.logger.warning("stripe_webhook: failed to record event %s: %s", event_id, str(e))
# On DB error, allow processing to avoid false negatives
return True
finally:
if conn:
conn.close()
def _handle_checkout_completed(session_data):
"""Process completed checkout session."""
from app.models import get_db
customer_email = session_data.get("customer_details", {}).get("email", "")
subscription_id = session_data.get("subscription")
price_id = session_data.get("line_items", {}).get("data", [{}])[0].get("price", {}).get("id", "")
current_app.logger.info(
"stripe_webhook: checkout.completed, email=%s, subscription=%s, price=%s",
customer_email,
subscription_id,
price_id,
)
# Handle document invoice payments (one-time checkout)
metadata = session_data.get("metadata", {})
document_id = metadata.get("document_id")
if document_id:
from app.models import update_document_status, update_document_stripe_payment
payment_intent = session_data.get("payment_intent", "unknown")
update_document_status(document_id, "paid")
update_document_stripe_payment(document_id, payment_intent, "succeeded")
current_app.logger.info(
"stripe_webhook: document payment processed, doc=%s, intent=%s",
document_id,
payment_intent,
)
return
# Handle subscription-based tier upgrades
# Reverse-lookup tier from price ID
tier = None
for key, price_id_str in STRIPE_PRICE_IDS.items():
if price_id_str == price_id:
tier = key
break
if not tier:
current_app.logger.warning(
"stripe_webhook: checkout.completed — price_id %s not mapped to any tier",
price_id,
)
if not customer_email:
current_app.logger.warning(
"stripe_webhook: checkout.completed — no customer email found"
)
if not tier or not customer_email:
return
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT id FROM users WHERE email_hash = ? AND stripe_customer_id IS NOT NULL",
(hash_value(customer_email.lower().strip()),),
).fetchone()
finally:
if conn:
conn.close()
if row and tier:
update_user_tier(row["id"], tier, subscription_id)
current_app.logger.info(
"stripe_webhook: user upgraded, id=%s, tier=%s, subscription=%s",
row["id"],
tier,
subscription_id,
)
elif row and not tier:
current_app.logger.warning(
"stripe_webhook: user found but tier not mapped, id=%s",
row["id"],
)
else:
current_app.logger.warning(
"stripe_webhook: user not found or missing stripe_customer_id, email=%s",
customer_email,
)
def _handle_subscription_updated(sub_data):
"""Handle subscription updates (upgrades/downgrades)."""
from app.models import get_db
customer_id = sub_data.get("customer")
status = sub_data.get("status")
current_app.logger.info(
"stripe_webhook: subscription.updated, customer=%s, status=%s",
customer_id,
status,
)
if status != "active" or not customer_id:
current_app.logger.info(
"stripe_webhook: subscription.updated — skipping (status=%s)",
status,
)
return
# Find price ID from subscription items
price_id = sub_data.get("items", {}).get("data", [{}])[0].get("price", {}).get("id", "")
tier = None
for key, price_id_str in STRIPE_PRICE_IDS.items():
if price_id_str == price_id:
tier = key
break
if not tier:
current_app.logger.warning(
"stripe_webhook: subscription.updated — price_id %s not mapped to any tier",
price_id,
)
return
conn = None
try:
conn = get_db()
row = conn.execute("SELECT id FROM users WHERE stripe_customer_id = ?", (customer_id,)).fetchone()
finally:
if conn:
conn.close()
if row:
update_user_tier(row["id"], tier, sub_data.get("id"))
current_app.logger.info(
"stripe_webhook: user tier updated via subscription, id=%s, tier=%s",
row["id"],
tier,
)
else:
current_app.logger.warning(
"stripe_webhook: subscription.updated — user not found for customer %s",
customer_id,
)
def _handle_subscription_deleted(sub_data):
"""Handle subscription cancellation — downgrade to free."""
from app.models import get_db
customer_id = sub_data.get("customer")
if not customer_id:
current_app.logger.warning("stripe_webhook: subscription.deleted — no customer_id")
return
current_app.logger.info(
"stripe_webhook: subscription.deleted, customer=%s",
customer_id,
)
conn = None
try:
conn = get_db()
row = conn.execute("SELECT id FROM users WHERE stripe_customer_id = ?", (customer_id,)).fetchone()
finally:
if conn:
conn.close()
if row:
update_user_tier(row["id"], "free")
current_app.logger.info(
"stripe_webhook: user downgraded to free, id=%s",
row["id"],
)
else:
current_app.logger.warning(
"stripe_webhook: subscription.deleted — user not found for customer %s",
customer_id,
)
def _handle_invoice_payment_succeeded(invoice_data):
"""Handle successful invoice payment — update document status to 'paid'."""
from app.models import get_db, update_document_status, update_document_stripe_payment
payment_intent = invoice_data.get("payment_intent")
customer = invoice_data.get("customer")
current_app.logger.info(
"stripe_webhook: invoice.payment_succeeded, intent=%s, customer=%s",
payment_intent,
customer,
)
if not payment_intent:
current_app.logger.warning("stripe_webhook: invoice.payment_succeeded — no payment_intent")
return
# Find documents with this payment intent reference
conn = None
try:
conn = get_db()
rows = conn.execute(
"SELECT document_id FROM documents WHERE stripe_payment_intent = ? OR stripe_payment_intent LIKE ?",
(payment_intent, f"%{payment_intent}%"),
).fetchall()
finally:
if conn:
conn.close()
if not rows:
# Also check via customer email — find documents belonging to this customer
if customer:
# Look up the customer email from Stripe
import stripe
stripe.api_key = os.environ.get("STRIPE_SECRET_KEY", "")
try:
customer_obj = stripe.Customer.retrieve(customer)
email = customer_obj.get("email", "")
if email:
from app.models import list_documents_by_email
docs = list_documents_by_email(email)
for doc in docs:
if doc.get("status") not in ("paid", "void"):
update_document_status(doc["document_id"], "paid")
update_document_stripe_payment(
doc["document_id"],
payment_intent,
"succeeded",
)
current_app.logger.info(
"stripe_webhook: document marked paid via customer email, doc=%s, email=%s",
doc["document_id"],
email,
)
except Exception as e:
current_app.logger.warning(
"stripe_webhook: invoice.payment_succeeded — Stripe customer lookup failed: %s",
str(e),
)
else:
current_app.logger.info(
"stripe_webhook: invoice.payment_succeeded — no matching documents, no customer to look up"
)
return
# Update each matching document
for row in rows:
update_document_status(row["document_id"], "paid")
update_document_stripe_payment(
row["document_id"],
payment_intent,
"succeeded",
)
current_app.logger.info(
"stripe_webhook: document marked paid via payment_intent, doc=%s",
row["document_id"],
)
@billing_bp.route("/create-checkout", methods=["POST"])
@login_required
def create_checkout():
"""Create a Stripe Checkout session for upgrading."""
import stripe
stripe.api_key = os.environ.get("STRIPE_SECRET_KEY", "")
user = current_user()
target_tier = request.form.get("tier", "").strip()
if target_tier not in STRIPE_PRICE_IDS or not STRIPE_PRICE_IDS[target_tier]:
flash("That tier is not available or Stripe is not configured.", "error")
return redirect(url_for("billing.upgrade"))
price_id = STRIPE_PRICE_IDS[target_tier]
# Create or reuse Stripe customer
if not user.get("stripe_customer_id"):
customer = stripe.Customer.create(
email=user["email"],
name=user.get("name", "") or user["email"],
metadata={"user_id": user["id"]},
)
update_stripe_customer(user["id"], customer.id)
checkout_session = stripe.checkout.Session.create(
customer=user["stripe_customer_id"],
line_items=[{"price": price_id, "quantity": 1}],
mode="subscription",
success_url=os.environ.get("APP_URL", "https://agentforms.io") + "/auth/dashboard",
cancel_url=os.environ.get("APP_URL", "https://agentforms.io") + "/billing/upgrade",
)
return redirect(checkout_session.url, code=303)