Coverage for backend / app / payments / webhooks.py: 96%

1"""Stripe webhooks module""" 

2 

3import datetime as dt 

4 

5from fastapi import HTTPException 

6from sqlalchemy.orm import Session 

7from starlette import status 

8 

9from app.emails.email_service import email_service 

10from app.models import User 

11from app.payments import logger, stripe 

12 

13 

14async def process_subscription_event( 

15 customer_id: str, 

16 event_type: str, 

17 db: Session, 

18 subscription_id: str | None = None, 

19 trial_end: float | None = None, 

20) -> None: 

21 """Process subscription event for a given user. 

22 :param customer_id: Stripe customer id 

23 :param event_type: Stripe event type 

24 :param db: Database session 

25 :param subscription_id: Stripe subscription id 

26 :param trial_end: Stripe trial end""" 

27 

28 user = db.query(User).filter(User.stripe_details.has(customer_id=customer_id)).first() 

29 logger.info(f"Received event: {event_type} for customer {customer_id}") 

30 

31 # If user not found (e.g., user was deleted), log and return early 

32 if not user: 

33 logger.warning(f"User not found for customer {customer_id}, skipping event {event_type}") 

34 return 

35 

36 # Handle subscription creation 

37 if event_type == "customer.subscription.created": 

38 user.stripe_details.subscription_id = subscription_id 

39 user.premium.is_active = True 

40 db.commit() 

41 logger.info(f"Subscription created: {subscription_id} for user {user.id}") 

42 

43 # Handle subscription deletion by setting premium as not active 

44 elif event_type == "customer.subscription.deleted": 

45 user.premium.is_active = False 

46 db.commit() 

47 logger.info(f"Subscription deleted for user {user.id}") 

48 

49 # Handle trial ending soon, send email to user 

50 elif event_type == "customer.subscription.trial_will_end": 

51 try: 

52 trial_end_date = dt.datetime.fromtimestamp(trial_end) 

53 email_service.send_trial_end_notification(user.email, trial_end_date) 

54 logger.info(f"Trial ending notification sent to user {user.id}") 

55 except Exception as e: 

56 logger.error(f"Failed to send trial ending email to user {user.id}: {e}") 

57 logger.info(f"Trial ending soon for user {user.id}") 

58 

59 elif event_type in ["billing_portal.session.created", "customer.created"]: 

60 pass 

61 

62 else: 

63 logger.error(f"Unhandled event type: {event_type}") 

64 

65 # Update the user's subscription status' 

66 subscription_status = await get_subscription_status(user) 

67 user.stripe_details.subscription_status = subscription_status["status"] 

68 user.stripe_details.trial_end_date = subscription_status["trial_end"] 

69 db.commit() 

70 

71 

72async def get_subscription_status(current_user: User) -> dict: 

73 """Get subscription status and trial information for the current user. 

74 :param current_user: Authenticated user from JWT token 

75 :return: dict with subscription status and remaining trial days""" 

76 

77 if not current_user.stripe_details.subscription_id: 

78 return {"status": None, "trial_end": None} 

79 try: 

80 subscription = await stripe.Subscription.retrieve_async(current_user.stripe_details.subscription_id) 

81 logger.info(f"Retrieved subscription status for user {current_user.id}. Trial end: {subscription.trial_end}") 

82 return {"status": subscription.status, "trial_end": subscription.trial_end} 

83 except stripe.error.StripeError as e: 

84 logger.error(f"Failed to retrieve subscription for user {current_user.id}: {e}") 

85 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))