75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
"""Payment service — Stripe payment intent creation and webhook handling."""
|
|
from __future__ import annotations
|
|
|
|
import structlog
|
|
import stripe
|
|
|
|
from config import settings
|
|
from core.storage import PACK_CREDITS
|
|
import core.pocketbase as pb
|
|
|
|
log = structlog.get_logger()
|
|
|
|
PACK_PRICES: dict[str, int] = {
|
|
'starter': 1900, # $19.00 in cents
|
|
'studio': 5900, # $59.00
|
|
'agency': 14900, # $149.00
|
|
}
|
|
|
|
|
|
class PaymentService:
|
|
def __init__(self) -> None:
|
|
stripe.api_key = settings.stripe_secret_key
|
|
|
|
def create_payment_intent(self, pack: str, user_id: str) -> dict:
|
|
"""Create a Stripe PaymentIntent for the selected credit pack.
|
|
|
|
Returns dict with client_secret and amount.
|
|
Raises ValueError for unknown pack names.
|
|
"""
|
|
if pack not in PACK_PRICES:
|
|
raise ValueError(f'Unknown pack: {pack}')
|
|
|
|
amount = PACK_PRICES[pack]
|
|
credits = PACK_CREDITS[pack]
|
|
|
|
intent = stripe.PaymentIntent.create(
|
|
amount=amount,
|
|
currency='usd',
|
|
metadata={'user_id': user_id, 'pack': pack, 'credits': credits},
|
|
automatic_payment_methods={'enabled': True},
|
|
)
|
|
log.info('payment_intent_created', user_id=user_id, pack=pack, amount=amount)
|
|
return {
|
|
'client_secret': intent.client_secret,
|
|
'amount': amount,
|
|
'credits': credits,
|
|
'pack': pack,
|
|
}
|
|
|
|
async def handle_webhook(self, payload: bytes, sig_header: str, admin_token: str) -> None:
|
|
"""Verify Stripe webhook signature and process payment_intent.succeeded.
|
|
|
|
On success: creates a purchase record in PocketBase (triggers credit top-up hook).
|
|
Raises stripe.error.SignatureVerificationError if signature is invalid.
|
|
"""
|
|
event = stripe.Webhook.construct_event(
|
|
payload, sig_header, settings.stripe_webhook_secret
|
|
)
|
|
|
|
if event['type'] != 'payment_intent.succeeded':
|
|
return # Ignore other events
|
|
|
|
intent = event['data']['object']
|
|
user_id = intent['metadata'].get('user_id')
|
|
pack = intent['metadata'].get('pack')
|
|
credits = PACK_CREDITS.get(pack, 0)
|
|
|
|
if not user_id or not credits:
|
|
log.warning('webhook_missing_metadata', intent_id=intent['id'])
|
|
return
|
|
|
|
await pb.create_purchase(user_id, credits, intent['id'], admin_token)
|
|
log.info('purchase_credited',
|
|
user_id=user_id, pack=pack, credits=credits, intent_id=intent['id'])
|