364 lines
13 KiB
Python
364 lines
13 KiB
Python
"""Billing service: Stripe integration layer.
|
|
|
|
All Stripe interactions go through this module. Routes stay thin.
|
|
|
|
Key rules:
|
|
- Stripe is source of truth; local DB mirrors via webhooks.
|
|
- Money in integer cents always.
|
|
- Webhook handlers are idempotent (safe to replay).
|
|
- stripe_enabled() guard lets the app run with no keys in dev.
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
from app.utils.time import utcnow
|
|
import stripe
|
|
from flask import current_app
|
|
from app.extensions import db
|
|
from app.models.user import User
|
|
from app.models.plan import Plan
|
|
from app.models.listing import Listing
|
|
from app.models.payments import Subscription, Transaction, Boost
|
|
from app.models.enums import Role
|
|
|
|
|
|
# ── Stripe client init ─────────────────────────────────────────────────────
|
|
|
|
def stripe_enabled() -> bool:
|
|
return bool(current_app.config.get("STRIPE_SECRET_KEY"))
|
|
|
|
|
|
def _stripe():
|
|
key = current_app.config.get("STRIPE_SECRET_KEY")
|
|
if not key:
|
|
raise RuntimeError("Stripe not configured — set STRIPE_SECRET_KEY in .env")
|
|
stripe.api_key = key
|
|
return stripe
|
|
|
|
|
|
# ── Customer ───────────────────────────────────────────────────────────────
|
|
|
|
def get_or_create_customer(user) -> str:
|
|
"""Return existing Stripe customer ID or create one."""
|
|
sub = Subscription.query.filter_by(user_id=user.id).first()
|
|
if sub and sub.stripe_customer_id:
|
|
return sub.stripe_customer_id
|
|
s = _stripe()
|
|
customer = s.Customer.create(
|
|
email=user.email,
|
|
name=user.display_name,
|
|
metadata={"user_id": user.id},
|
|
)
|
|
return customer.id
|
|
|
|
|
|
# ── Subscription checkout ──────────────────────────────────────────────────
|
|
|
|
def create_subscription_checkout(user, plan, success_url, cancel_url) -> str:
|
|
"""Create a Stripe Checkout Session for a subscription. Returns URL."""
|
|
if not plan.stripe_price_id:
|
|
raise ValueError(f"Plan '{plan.slug}' has no stripe_price_id configured")
|
|
s = _stripe()
|
|
customer_id = get_or_create_customer(user)
|
|
session = s.checkout.Session.create(
|
|
customer=customer_id,
|
|
mode="subscription",
|
|
line_items=[{"price": plan.stripe_price_id, "quantity": 1}],
|
|
automatic_tax={"enabled": True},
|
|
success_url=success_url,
|
|
cancel_url=cancel_url,
|
|
metadata={"user_id": user.id, "plan_id": plan.id},
|
|
allow_promotion_codes=True,
|
|
)
|
|
return session.url
|
|
|
|
|
|
def create_customer_portal(user, return_url) -> str:
|
|
"""Create a Stripe Customer Portal session. Returns URL."""
|
|
s = _stripe()
|
|
customer_id = get_or_create_customer(user)
|
|
session = s.billing_portal.Session.create(
|
|
customer=customer_id,
|
|
return_url=return_url,
|
|
)
|
|
return session.url
|
|
|
|
|
|
# ── Boost checkout ─────────────────────────────────────────────────────────
|
|
|
|
def create_boost_checkout(user, listing, boost_type, success_url, cancel_url) -> str:
|
|
"""Create a Stripe Checkout Session for a one-off boost. Returns URL."""
|
|
info = Boost.TYPES.get(boost_type)
|
|
if not info:
|
|
raise ValueError(f"Unknown boost type: {boost_type}")
|
|
s = _stripe()
|
|
customer_id = get_or_create_customer(user)
|
|
session = s.checkout.Session.create(
|
|
customer=customer_id,
|
|
mode="payment",
|
|
line_items=[{
|
|
"price_data": {
|
|
"currency": "usd",
|
|
"unit_amount": info["cents"],
|
|
"product_data": {"name": info["label"],
|
|
"description": f'"{listing.title[:60]}"'},
|
|
},
|
|
"quantity": 1,
|
|
}],
|
|
automatic_tax={"enabled": True},
|
|
success_url=success_url,
|
|
cancel_url=cancel_url,
|
|
metadata={
|
|
"user_id": user.id,
|
|
"listing_id": listing.id,
|
|
"boost_type": boost_type,
|
|
},
|
|
)
|
|
return session.url
|
|
|
|
|
|
# ── Boost activation (called after successful payment) ────────────────────
|
|
|
|
def activate_boost(user_id: int, listing_id: int, boost_type: str,
|
|
stripe_payment_intent_id: str = None,
|
|
amount_cents: int = None) -> Boost:
|
|
"""Write Transaction + Boost rows. Idempotent on stripe_payment_intent_id."""
|
|
# idempotency: skip if transaction already recorded
|
|
if stripe_payment_intent_id:
|
|
existing = Transaction.query.filter_by(
|
|
stripe_object_id=stripe_payment_intent_id).first()
|
|
if existing:
|
|
return Boost.query.filter_by(
|
|
transaction_id=existing.id).first()
|
|
|
|
info = Boost.TYPES[boost_type]
|
|
txn = Transaction(
|
|
user_id=user_id,
|
|
type="boost",
|
|
amount_cents=amount_cents or info["cents"],
|
|
stripe_object_id=stripe_payment_intent_id,
|
|
status="succeeded",
|
|
meta={"boost_type": boost_type, "listing_id": listing_id},
|
|
)
|
|
db.session.add(txn)
|
|
db.session.flush()
|
|
|
|
expires_at = utcnow() + timedelta(days=info["days"])
|
|
boost = Boost(listing_id=listing_id, user_id=user_id,
|
|
type=boost_type, expires_at=expires_at,
|
|
transaction_id=txn.id)
|
|
db.session.add(boost)
|
|
|
|
# apply listing effects
|
|
listing = db.session.get(Listing, listing_id)
|
|
if listing:
|
|
if boost_type == "featured":
|
|
listing.is_featured = True
|
|
if boost_type == "bump":
|
|
listing.bump_at = utcnow()
|
|
|
|
db.session.commit()
|
|
return boost
|
|
|
|
|
|
# ── Tier sync (writes to user + subscription table) ───────────────────────
|
|
|
|
def sync_subscription(user_id: int, stripe_sub_obj) -> Subscription:
|
|
"""Upsert local Subscription from a Stripe subscription object."""
|
|
plan = Plan.query.filter_by(
|
|
stripe_price_id=stripe_sub_obj["items"]["data"][0]["price"]["id"]
|
|
).first()
|
|
|
|
user = db.session.get(User, user_id)
|
|
sub = Subscription.query.filter_by(user_id=user_id).first()
|
|
if sub is None:
|
|
sub = Subscription(user_id=user_id)
|
|
db.session.add(sub)
|
|
|
|
sub.stripe_customer_id = stripe_sub_obj["customer"]
|
|
sub.stripe_sub_id = stripe_sub_obj["id"]
|
|
sub.status = stripe_sub_obj["status"]
|
|
sub.cancel_at_period_end = stripe_sub_obj["cancel_at_period_end"]
|
|
if stripe_sub_obj.get("current_period_end"):
|
|
sub.current_period_end = datetime.utcfromtimestamp(
|
|
stripe_sub_obj["current_period_end"])
|
|
if plan:
|
|
sub.plan_id = plan.id
|
|
user.tier_id = plan.id
|
|
user.role = Role.subscriber
|
|
db.session.commit()
|
|
return sub
|
|
|
|
|
|
def downgrade_to_free(user_id: int):
|
|
"""Called on subscription canceled/expired. Revert user to free tier."""
|
|
user = db.session.get(User, user_id)
|
|
if user is None:
|
|
return
|
|
free_plan = Plan.query.filter_by(slug="free").first()
|
|
user.role = Role.free
|
|
user.tier_id = free_plan.id if free_plan else None
|
|
sub = Subscription.query.filter_by(user_id=user_id).first()
|
|
if sub:
|
|
sub.status = "canceled"
|
|
db.session.commit()
|
|
|
|
|
|
# ── Webhook event handlers ─────────────────────────────────────────────────
|
|
|
|
def handle_webhook(payload: bytes, sig_header: str):
|
|
"""Verify signature, dispatch to handler. Raises on bad sig."""
|
|
s = _stripe()
|
|
secret = current_app.config["STRIPE_WEBHOOK_SECRET"]
|
|
try:
|
|
event = s.Webhook.construct_event(payload, sig_header, secret)
|
|
except stripe.error.SignatureVerificationError as e:
|
|
raise ValueError(f"Invalid Stripe signature: {e}") from e
|
|
|
|
etype = event["type"]
|
|
data = event["data"]["object"]
|
|
|
|
if etype == "checkout.session.completed":
|
|
_on_checkout_completed(data)
|
|
elif etype in ("customer.subscription.updated",
|
|
"customer.subscription.created"):
|
|
_on_subscription_updated(data)
|
|
elif etype == "customer.subscription.deleted":
|
|
_on_subscription_deleted(data)
|
|
elif etype == "invoice.payment_failed":
|
|
_on_payment_failed(data)
|
|
else:
|
|
current_app.logger.debug("Unhandled Stripe event: %s", etype)
|
|
|
|
return event
|
|
|
|
|
|
def _on_checkout_completed(session):
|
|
meta = session.get("metadata", {})
|
|
mode = session.get("mode")
|
|
|
|
if mode == "subscription":
|
|
user_id = int(meta.get("user_id", 0))
|
|
if not user_id:
|
|
return
|
|
# Fetch full subscription object from Stripe
|
|
s = _stripe()
|
|
stripe_sub = s.Subscription.retrieve(session["subscription"])
|
|
sync_subscription(user_id, stripe_sub)
|
|
# Record transaction
|
|
_record_sub_transaction(user_id, session)
|
|
|
|
elif mode == "payment":
|
|
# boost purchase
|
|
user_id = int(meta.get("user_id", 0))
|
|
listing_id = int(meta.get("listing_id", 0))
|
|
boost_type = meta.get("boost_type")
|
|
if user_id and listing_id and boost_type:
|
|
activate_boost(
|
|
user_id=user_id,
|
|
listing_id=listing_id,
|
|
boost_type=boost_type,
|
|
stripe_payment_intent_id=session.get("payment_intent"),
|
|
amount_cents=session.get("amount_total"),
|
|
)
|
|
|
|
|
|
def _on_subscription_updated(stripe_sub):
|
|
customer_id = stripe_sub["customer"]
|
|
sub = Subscription.query.filter_by(
|
|
stripe_customer_id=customer_id).first()
|
|
if sub:
|
|
sync_subscription(sub.user_id, stripe_sub)
|
|
|
|
|
|
def _on_subscription_deleted(stripe_sub):
|
|
customer_id = stripe_sub["customer"]
|
|
sub = Subscription.query.filter_by(
|
|
stripe_customer_id=customer_id).first()
|
|
if sub:
|
|
downgrade_to_free(sub.user_id)
|
|
|
|
|
|
def _on_payment_failed(invoice):
|
|
customer_id = invoice["customer"]
|
|
sub = Subscription.query.filter_by(
|
|
stripe_customer_id=customer_id).first()
|
|
if sub:
|
|
sub.status = "past_due"
|
|
db.session.commit()
|
|
# email notification handled by email service caller (route layer)
|
|
|
|
|
|
def _record_sub_transaction(user_id, session):
|
|
existing = Transaction.query.filter_by(
|
|
stripe_object_id=session.get("payment_intent") or session["id"]).first()
|
|
if existing:
|
|
return
|
|
txn = Transaction(
|
|
user_id=user_id,
|
|
type="subscription",
|
|
amount_cents=session.get("amount_total", 0),
|
|
stripe_object_id=session.get("payment_intent") or session["id"],
|
|
status="succeeded",
|
|
meta={"session_id": session["id"]},
|
|
)
|
|
db.session.add(txn)
|
|
db.session.commit()
|
|
|
|
|
|
# ── Boost expiry sweep ─────────────────────────────────────────────────────
|
|
|
|
def expire_boosts() -> int:
|
|
"""Clear expired boosts and revert listing effects. Returns count."""
|
|
now = utcnow()
|
|
expired = Boost.query.filter(Boost.expires_at <= now).all()
|
|
n = 0
|
|
for boost in expired:
|
|
listing = db.session.get(Listing, boost.listing_id)
|
|
if listing and boost.type == "featured":
|
|
# only un-feature if no other active featured boost exists
|
|
other = Boost.query.filter(
|
|
Boost.listing_id == boost.listing_id,
|
|
Boost.type == "featured",
|
|
Boost.expires_at > now,
|
|
Boost.id != boost.id,
|
|
).first()
|
|
if not other:
|
|
listing.is_featured = False
|
|
db.session.delete(boost)
|
|
n += 1
|
|
if n:
|
|
db.session.commit()
|
|
return n
|
|
|
|
|
|
# ── Nightly reconcile ──────────────────────────────────────────────────────
|
|
|
|
def reconcile_subscriptions():
|
|
"""Compare local active subscriptions vs Stripe. Fix mismatches.
|
|
Catches webhooks that were missed/failed. Returns (checked, fixed) counts.
|
|
"""
|
|
if not stripe_enabled():
|
|
return 0, 0
|
|
s = _stripe()
|
|
subs = Subscription.query.filter(
|
|
Subscription.stripe_sub_id.isnot(None),
|
|
Subscription.status.in_(["active", "trialing", "past_due"]),
|
|
).all()
|
|
checked, fixed = 0, 0
|
|
for sub in subs:
|
|
try:
|
|
stripe_sub = s.Subscription.retrieve(sub.stripe_sub_id)
|
|
checked += 1
|
|
if stripe_sub["status"] != sub.status:
|
|
current_app.logger.info(
|
|
"Reconcile: sub %s local=%s stripe=%s — fixing",
|
|
sub.stripe_sub_id, sub.status, stripe_sub["status"])
|
|
if stripe_sub["status"] == "canceled":
|
|
downgrade_to_free(sub.user_id)
|
|
else:
|
|
sync_subscription(sub.user_id, stripe_sub)
|
|
fixed += 1
|
|
except stripe.error.StripeError as e:
|
|
current_app.logger.error("Reconcile error sub %s: %s",
|
|
sub.stripe_sub_id, e)
|
|
return checked, fixed
|