Imporve UI and token
This commit is contained in:
Urtzi Alfaro
@@ -4,12 +4,17 @@ These endpoints receive events from payment providers like Stripe
|
||||
"""
|
||||
|
||||
import structlog
|
||||
import stripe
|
||||
from fastapi import APIRouter, Depends, HTTPException, status, Request
|
||||
from typing import Dict, Any
|
||||
from datetime import datetime
|
||||
|
||||
from app.services.payment_service import PaymentService
|
||||
from shared.auth.decorators import get_current_user_dep
|
||||
from shared.monitoring.metrics import track_endpoint_metrics
|
||||
from app.core.config import settings
|
||||
from app.core.database import get_db
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select
|
||||
from app.models.tenants import Subscription, Tenant
|
||||
|
||||
logger = structlog.get_logger()
|
||||
router = APIRouter()
|
||||
@@ -24,58 +29,295 @@ def get_payment_service():
|
||||
@router.post("/webhooks/stripe")
|
||||
async def stripe_webhook(
|
||||
request: Request,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
payment_service: PaymentService = Depends(get_payment_service)
|
||||
):
|
||||
"""
|
||||
Stripe webhook endpoint to handle payment events
|
||||
This endpoint verifies webhook signatures and processes Stripe events
|
||||
"""
|
||||
try:
|
||||
# Get the payload
|
||||
# Get the payload and signature
|
||||
payload = await request.body()
|
||||
sig_header = request.headers.get('stripe-signature')
|
||||
|
||||
# In a real implementation, you would verify the signature
|
||||
# using the webhook signing secret
|
||||
# event = stripe.Webhook.construct_event(
|
||||
# payload, sig_header, settings.STRIPE_WEBHOOK_SECRET
|
||||
# )
|
||||
|
||||
# For now, we'll just log the event
|
||||
logger.info("Received Stripe webhook", payload=payload.decode('utf-8'))
|
||||
|
||||
|
||||
if not sig_header:
|
||||
logger.error("Missing stripe-signature header")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Missing signature header"
|
||||
)
|
||||
|
||||
# Verify the webhook signature
|
||||
try:
|
||||
event = stripe.Webhook.construct_event(
|
||||
payload, sig_header, settings.STRIPE_WEBHOOK_SECRET
|
||||
)
|
||||
except stripe.error.SignatureVerificationError as e:
|
||||
logger.error("Invalid webhook signature", error=str(e))
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Invalid signature"
|
||||
)
|
||||
except ValueError as e:
|
||||
logger.error("Invalid payload", error=str(e))
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Invalid payload"
|
||||
)
|
||||
|
||||
# Get event type and data
|
||||
event_type = event['type']
|
||||
event_data = event['data']['object']
|
||||
|
||||
logger.info("Processing Stripe webhook event",
|
||||
event_type=event_type,
|
||||
event_id=event.get('id'))
|
||||
|
||||
# Process different types of events
|
||||
# event_type = event['type']
|
||||
# event_data = event['data']['object']
|
||||
|
||||
# Example processing for different event types:
|
||||
# if event_type == 'checkout.session.completed':
|
||||
# # Handle successful checkout
|
||||
# pass
|
||||
# elif event_type == 'customer.subscription.created':
|
||||
# # Handle new subscription
|
||||
# pass
|
||||
# elif event_type == 'customer.subscription.updated':
|
||||
# # Handle subscription update
|
||||
# pass
|
||||
# elif event_type == 'customer.subscription.deleted':
|
||||
# # Handle subscription cancellation
|
||||
# pass
|
||||
# elif event_type == 'invoice.payment_succeeded':
|
||||
# # Handle successful payment
|
||||
# pass
|
||||
# elif event_type == 'invoice.payment_failed':
|
||||
# # Handle failed payment
|
||||
# pass
|
||||
|
||||
return {"success": True}
|
||||
|
||||
if event_type == 'checkout.session.completed':
|
||||
# Handle successful checkout
|
||||
await handle_checkout_completed(event_data, db)
|
||||
|
||||
elif event_type == 'customer.subscription.created':
|
||||
# Handle new subscription
|
||||
await handle_subscription_created(event_data, db)
|
||||
|
||||
elif event_type == 'customer.subscription.updated':
|
||||
# Handle subscription update
|
||||
await handle_subscription_updated(event_data, db)
|
||||
|
||||
elif event_type == 'customer.subscription.deleted':
|
||||
# Handle subscription cancellation
|
||||
await handle_subscription_deleted(event_data, db)
|
||||
|
||||
elif event_type == 'invoice.payment_succeeded':
|
||||
# Handle successful payment
|
||||
await handle_payment_succeeded(event_data, db)
|
||||
|
||||
elif event_type == 'invoice.payment_failed':
|
||||
# Handle failed payment
|
||||
await handle_payment_failed(event_data, db)
|
||||
|
||||
elif event_type == 'customer.subscription.trial_will_end':
|
||||
# Handle trial ending soon (3 days before)
|
||||
await handle_trial_will_end(event_data, db)
|
||||
|
||||
else:
|
||||
logger.info("Unhandled webhook event type", event_type=event_type)
|
||||
|
||||
return {"success": True, "event_type": event_type}
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("Error processing Stripe webhook", error=str(e))
|
||||
logger.error("Error processing Stripe webhook", error=str(e), exc_info=True)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Webhook error"
|
||||
detail="Webhook processing error"
|
||||
)
|
||||
|
||||
|
||||
async def handle_checkout_completed(session: Dict[str, Any], db: AsyncSession):
|
||||
"""Handle successful checkout session completion"""
|
||||
logger.info("Processing checkout.session.completed",
|
||||
session_id=session.get('id'))
|
||||
|
||||
customer_id = session.get('customer')
|
||||
subscription_id = session.get('subscription')
|
||||
|
||||
if customer_id and subscription_id:
|
||||
# Update tenant with subscription info
|
||||
query = select(Tenant).where(Tenant.stripe_customer_id == customer_id)
|
||||
result = await db.execute(query)
|
||||
tenant = result.scalar_one_or_none()
|
||||
|
||||
if tenant:
|
||||
logger.info("Checkout completed for tenant",
|
||||
tenant_id=str(tenant.id),
|
||||
subscription_id=subscription_id)
|
||||
|
||||
|
||||
async def handle_subscription_created(subscription: Dict[str, Any], db: AsyncSession):
|
||||
"""Handle new subscription creation"""
|
||||
logger.info("Processing customer.subscription.created",
|
||||
subscription_id=subscription.get('id'))
|
||||
|
||||
customer_id = subscription.get('customer')
|
||||
subscription_id = subscription.get('id')
|
||||
status_value = subscription.get('status')
|
||||
|
||||
# Find tenant by customer ID
|
||||
query = select(Tenant).where(Tenant.stripe_customer_id == customer_id)
|
||||
result = await db.execute(query)
|
||||
tenant = result.scalar_one_or_none()
|
||||
|
||||
if tenant:
|
||||
logger.info("Subscription created for tenant",
|
||||
tenant_id=str(tenant.id),
|
||||
subscription_id=subscription_id,
|
||||
status=status_value)
|
||||
|
||||
|
||||
async def handle_subscription_updated(subscription: Dict[str, Any], db: AsyncSession):
|
||||
"""Handle subscription updates (status changes, plan changes, etc.)"""
|
||||
subscription_id = subscription.get('id')
|
||||
status_value = subscription.get('status')
|
||||
|
||||
logger.info("Processing customer.subscription.updated",
|
||||
subscription_id=subscription_id,
|
||||
new_status=status_value)
|
||||
|
||||
# Find subscription in database
|
||||
query = select(Subscription).where(Subscription.subscription_id == subscription_id)
|
||||
result = await db.execute(query)
|
||||
db_subscription = result.scalar_one_or_none()
|
||||
|
||||
if db_subscription:
|
||||
# Update subscription status
|
||||
db_subscription.status = status_value
|
||||
db_subscription.current_period_end = datetime.fromtimestamp(
|
||||
subscription.get('current_period_end')
|
||||
)
|
||||
|
||||
# Update active status based on Stripe status
|
||||
if status_value == 'active':
|
||||
db_subscription.is_active = True
|
||||
elif status_value in ['canceled', 'past_due', 'unpaid']:
|
||||
db_subscription.is_active = False
|
||||
|
||||
await db.commit()
|
||||
|
||||
# Invalidate cache
|
||||
try:
|
||||
from app.services.subscription_cache import get_subscription_cache_service
|
||||
import shared.redis_utils
|
||||
|
||||
redis_client = await shared.redis_utils.initialize_redis(settings.REDIS_URL)
|
||||
cache_service = get_subscription_cache_service(redis_client)
|
||||
await cache_service.invalidate_subscription_cache(str(db_subscription.tenant_id))
|
||||
except Exception as cache_error:
|
||||
logger.error("Failed to invalidate cache", error=str(cache_error))
|
||||
|
||||
logger.info("Subscription updated in database",
|
||||
subscription_id=subscription_id,
|
||||
tenant_id=str(db_subscription.tenant_id))
|
||||
|
||||
|
||||
async def handle_subscription_deleted(subscription: Dict[str, Any], db: AsyncSession):
|
||||
"""Handle subscription cancellation/deletion"""
|
||||
subscription_id = subscription.get('id')
|
||||
|
||||
logger.info("Processing customer.subscription.deleted",
|
||||
subscription_id=subscription_id)
|
||||
|
||||
# Find subscription in database
|
||||
query = select(Subscription).where(Subscription.subscription_id == subscription_id)
|
||||
result = await db.execute(query)
|
||||
db_subscription = result.scalar_one_or_none()
|
||||
|
||||
if db_subscription:
|
||||
db_subscription.status = 'canceled'
|
||||
db_subscription.is_active = False
|
||||
db_subscription.canceled_at = datetime.utcnow()
|
||||
|
||||
await db.commit()
|
||||
|
||||
# Invalidate cache
|
||||
try:
|
||||
from app.services.subscription_cache import get_subscription_cache_service
|
||||
import shared.redis_utils
|
||||
|
||||
redis_client = await shared.redis_utils.initialize_redis(settings.REDIS_URL)
|
||||
cache_service = get_subscription_cache_service(redis_client)
|
||||
await cache_service.invalidate_subscription_cache(str(db_subscription.tenant_id))
|
||||
except Exception as cache_error:
|
||||
logger.error("Failed to invalidate cache", error=str(cache_error))
|
||||
|
||||
logger.info("Subscription canceled in database",
|
||||
subscription_id=subscription_id,
|
||||
tenant_id=str(db_subscription.tenant_id))
|
||||
|
||||
|
||||
async def handle_payment_succeeded(invoice: Dict[str, Any], db: AsyncSession):
|
||||
"""Handle successful invoice payment"""
|
||||
invoice_id = invoice.get('id')
|
||||
subscription_id = invoice.get('subscription')
|
||||
|
||||
logger.info("Processing invoice.payment_succeeded",
|
||||
invoice_id=invoice_id,
|
||||
subscription_id=subscription_id)
|
||||
|
||||
if subscription_id:
|
||||
# Find subscription and ensure it's active
|
||||
query = select(Subscription).where(Subscription.subscription_id == subscription_id)
|
||||
result = await db.execute(query)
|
||||
db_subscription = result.scalar_one_or_none()
|
||||
|
||||
if db_subscription:
|
||||
db_subscription.status = 'active'
|
||||
db_subscription.is_active = True
|
||||
|
||||
await db.commit()
|
||||
|
||||
logger.info("Payment succeeded, subscription activated",
|
||||
subscription_id=subscription_id,
|
||||
tenant_id=str(db_subscription.tenant_id))
|
||||
|
||||
|
||||
async def handle_payment_failed(invoice: Dict[str, Any], db: AsyncSession):
|
||||
"""Handle failed invoice payment"""
|
||||
invoice_id = invoice.get('id')
|
||||
subscription_id = invoice.get('subscription')
|
||||
customer_id = invoice.get('customer')
|
||||
|
||||
logger.error("Processing invoice.payment_failed",
|
||||
invoice_id=invoice_id,
|
||||
subscription_id=subscription_id,
|
||||
customer_id=customer_id)
|
||||
|
||||
if subscription_id:
|
||||
# Find subscription and mark as past_due
|
||||
query = select(Subscription).where(Subscription.subscription_id == subscription_id)
|
||||
result = await db.execute(query)
|
||||
db_subscription = result.scalar_one_or_none()
|
||||
|
||||
if db_subscription:
|
||||
db_subscription.status = 'past_due'
|
||||
db_subscription.is_active = False
|
||||
|
||||
await db.commit()
|
||||
|
||||
logger.warning("Payment failed, subscription marked past_due",
|
||||
subscription_id=subscription_id,
|
||||
tenant_id=str(db_subscription.tenant_id))
|
||||
|
||||
# TODO: Send notification to user about payment failure
|
||||
# You can integrate with your notification service here
|
||||
|
||||
|
||||
async def handle_trial_will_end(subscription: Dict[str, Any], db: AsyncSession):
|
||||
"""Handle notification that trial will end in 3 days"""
|
||||
subscription_id = subscription.get('id')
|
||||
trial_end = subscription.get('trial_end')
|
||||
|
||||
logger.info("Processing customer.subscription.trial_will_end",
|
||||
subscription_id=subscription_id,
|
||||
trial_end_timestamp=trial_end)
|
||||
|
||||
# Find subscription
|
||||
query = select(Subscription).where(Subscription.subscription_id == subscription_id)
|
||||
result = await db.execute(query)
|
||||
db_subscription = result.scalar_one_or_none()
|
||||
|
||||
if db_subscription:
|
||||
logger.info("Trial ending soon for subscription",
|
||||
subscription_id=subscription_id,
|
||||
tenant_id=str(db_subscription.tenant_id))
|
||||
|
||||
# TODO: Send notification to user about trial ending soon
|
||||
# You can integrate with your notification service here
|
||||
|
||||
@router.post("/webhooks/generic")
|
||||
async def generic_webhook(
|
||||
request: Request,
|
||||
|
||||
@@ -6,6 +6,7 @@ This service abstracts payment provider interactions and makes the system paymen
|
||||
import structlog
|
||||
from typing import Dict, Any, Optional
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.config import settings
|
||||
@@ -250,3 +251,53 @@ class PaymentService:
|
||||
except Exception as e:
|
||||
logger.error("Failed to get subscription from payment provider", error=str(e))
|
||||
raise e
|
||||
|
||||
async def sync_subscription_status(self, subscription_id: str, db_session: Session) -> Subscription:
|
||||
"""
|
||||
Sync subscription status from payment provider to database
|
||||
This ensures our local subscription status matches the payment provider
|
||||
"""
|
||||
try:
|
||||
# Get current subscription from payment provider
|
||||
stripe_subscription = await self.payment_provider.get_subscription(subscription_id)
|
||||
|
||||
logger.info("Syncing subscription status",
|
||||
subscription_id=subscription_id,
|
||||
stripe_status=stripe_subscription.status)
|
||||
|
||||
# Update local database record
|
||||
self.subscription_repo.db_session = db_session
|
||||
local_subscription = await self.subscription_repo.get_by_stripe_id(subscription_id)
|
||||
|
||||
if local_subscription:
|
||||
# Update status and dates
|
||||
local_subscription.status = stripe_subscription.status
|
||||
local_subscription.current_period_end = stripe_subscription.current_period_end
|
||||
|
||||
# Handle status-specific logic
|
||||
if stripe_subscription.status == 'active':
|
||||
local_subscription.is_active = True
|
||||
local_subscription.canceled_at = None
|
||||
elif stripe_subscription.status == 'canceled':
|
||||
local_subscription.is_active = False
|
||||
local_subscription.canceled_at = datetime.utcnow()
|
||||
elif stripe_subscription.status == 'past_due':
|
||||
local_subscription.is_active = False
|
||||
elif stripe_subscription.status == 'unpaid':
|
||||
local_subscription.is_active = False
|
||||
|
||||
await self.subscription_repo.update(local_subscription)
|
||||
logger.info("Subscription status synced successfully",
|
||||
subscription_id=subscription_id,
|
||||
new_status=stripe_subscription.status)
|
||||
else:
|
||||
logger.warning("Local subscription not found for Stripe subscription",
|
||||
subscription_id=subscription_id)
|
||||
|
||||
return stripe_subscription
|
||||
|
||||
except Exception as e:
|
||||
logger.error("Failed to sync subscription status",
|
||||
error=str(e),
|
||||
subscription_id=subscription_id)
|
||||
raise e
|
||||
|
||||
@@ -23,7 +23,7 @@ opentelemetry-instrumentation-httpx==0.60b1
|
||||
opentelemetry-instrumentation-redis==0.60b1
|
||||
opentelemetry-instrumentation-sqlalchemy==0.60b1
|
||||
python-jose[cryptography]==3.3.0
|
||||
stripe==11.3.0
|
||||
stripe==14.1.0
|
||||
python-multipart==0.0.6
|
||||
cryptography==44.0.0
|
||||
email-validator==2.2.0
|
||||
|
||||
Reference in New Issue
Block a user