import json import logging import os import stripe from django.conf import settings from django.contrib.auth.decorators import login_required from django.db import IntegrityError, transaction from django.http import HttpResponse from django.shortcuts import redirect, render from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_POST from home.models import CoursePage from purchase.models import CoursePurchase, PurchasableProduct logger = logging.getLogger(__name__) def mock_purchase_course(request, course_id): course = CoursePage.objects.get(id=course_id) course.mock_purchase(request.user) return redirect(course.url) def mock_refund_purchase(request, purchase_id): purchase = CoursePurchase.objects.get(id=purchase_id) purchase.mock_refund() return redirect(purchase.course.url) def purchase_success(request): return render(request, "success.html") @login_required @require_POST def create_checkout_session(request, purchasable_id): """Create a Stripe Checkout Session for the given PurchasableProduct and include the local user id in metadata. This view requires an authenticated POST request and will redirect the user to Stripe Checkout. It will also create or get a pending CoursePurchase linked to the checkout session. """ stripe.api_key = getattr( settings, "STRIPE_SECRET_KEY", os.getenv("STRIPE_SECRET_KEY") ) try: purch = PurchasableProduct.objects.get(pk=purchasable_id) except PurchasableProduct.DoesNotExist: return HttpResponse("Purchasable product not found", status=404) if not purch.stripe_price_id: return HttpResponse("Product not configured for Stripe", status=400) try: session = stripe.checkout.Session.create( payment_method_types=["card", "blik", "p24"], line_items=[{"price": purch.stripe_price_id, "quantity": 1}], customer_email=request.user.email or None, mode="payment", client_reference_id=str(request.user.id), metadata={"user_id": str(request.user.id), "purchasable_id": str(purch.id)}, payment_intent_data={ "metadata": { "user_id": str(request.user.id), "purchasable_id": str(purch.id), "client_reference_id": str(request.user.id), } }, success_url=getattr( settings, "STRIPE_SUCCESS_URL", "https://example.com/success" ), cancel_url=getattr( settings, "STRIPE_CANCEL_URL", "https://example.com/cancel" ), ) # Create or get a pending CoursePurchase tied to this session (idempotent) try: with transaction.atomic(): purchase, created = CoursePurchase.objects.get_or_create( user=request.user, course=purch.course, refunded=False, defaults={ "status": CoursePurchase.Status.PENDING, "stripe_checkout_session_id": session.id, "stripe_payment_intent_id": session.payment_intent, }, ) except IntegrityError: # Race: another worker created the purchase concurrently. Re-fetch. purchase = CoursePurchase.objects.filter( user=request.user, course=purch.course ).first() if not purchase: purchase = CoursePurchase.objects.filter( stripe_checkout_session_id=session.id ).first() if purchase: if purchase.status == CoursePurchase.Status.PAID and not purchase.refunded: return redirect(purch.course.url) update_fields = [] if purchase.status != CoursePurchase.Status.PENDING: purchase.status = CoursePurchase.Status.PENDING update_fields.append("status") if purchase.refunded: purchase.refunded = False update_fields.append("refunded") if purchase.stripe_checkout_session_id != session.id: purchase.stripe_checkout_session_id = session.id update_fields.append("stripe_checkout_session_id") if purchase.stripe_payment_intent_id is not None: purchase.stripe_payment_intent_id = None update_fields.append("stripe_payment_intent_id") if purchase.stripe_charge_id is not None: purchase.stripe_charge_id = None update_fields.append("stripe_charge_id") if update_fields: purchase.save(update_fields=update_fields) # Redirect to Stripe Checkout return redirect(session.url) except Exception as e: logger.exception( f"Failed to create checkout session for purchasable {purchasable_id}: {e}" ) raise e @csrf_exempt def stripe_webhook(request): stripe.api_key = getattr( settings, "STRIPE_SECRET_KEY", os.getenv("STRIPE_SECRET_KEY") ) webhook_secret = getattr( settings, "STRIPE_WEBHOOK_SECRET", os.getenv("STRIPE_WEBHOOK_SECRET") ) payload = request.body sig_header = request.META.get("HTTP_STRIPE_SIGNATURE") event = None try: if webhook_secret: event = stripe.Webhook.construct_event(payload, sig_header, webhook_secret) else: # fallback: parse payload as JSON event = stripe.Event.construct_from( json.loads(payload.decode("utf-8")), stripe.api_key ) except ValueError as e: logger.error(f"Invalid payload: {e}") raise e except stripe.error.SignatureVerificationError as e: logger.error(f"Webhook signature verification failed: {e}") raise e # Helper to safely index Stripe objects / nested dict-like objects using [] semantics def _s(obj, key): try: return obj[key] except Exception: return None # Handle the event logger.info(f"Received Stripe event: {event['type']}") # Helper to mark a purchase refunded by Stripe identifiers def _mark_purchase_refunded(purchase): if not purchase: return purchase.refunded = True purchase.status = CoursePurchase.Status.REFUNDED # Save normally so our CoursePurchase.save() logic runs (this will remove groups/gitea membership) purchase.save() logger.info( f"Marked CoursePurchase {purchase.id} as REFUNDED due to Stripe event" ) def _find_purchase_from_payment_intent(pi): pi_id = _s(pi, "id") metadata = _s(pi, "metadata") or {} purchase = None session_id = ( _s(metadata, "checkout_session") or _s(metadata, "session_id") or _s(metadata, "stripe_checkout_session_id") ) if session_id: purchase = CoursePurchase.objects.filter( stripe_checkout_session_id=session_id ).first() if not purchase and pi_id: purchase = CoursePurchase.objects.filter( stripe_payment_intent_id=pi_id ).first() from django.contrib.auth import get_user_model User = get_user_model() user = None user_id = _s(metadata, "user_id") if user_id: try: user = User.objects.get(pk=int(user_id)) except Exception: user = None if not user: client_ref = _s(metadata, "client_reference_id") or _s( metadata, "client_id" ) if client_ref: try: user = User.objects.get(pk=int(client_ref)) except Exception: user = None if not user: email = _s(pi, "receipt_email") if not email: charges = _s(_s(pi, "charges") or {}, "data") or [] if charges: billing = _s(charges[0], "billing_details") or {} email = _s(billing, "email") if email: user = User.objects.filter(email=email).first() course = None purchasable_id = _s(metadata, "purchasable_id") if purchasable_id: try: purch = PurchasableProduct.objects.get(pk=int(purchasable_id)) course = purch.course except Exception: course = None if not purchase and user and course: purchase = ( CoursePurchase.objects.filter(user=user, course=course, refunded=False) .order_by("-id") .first() ) return purchase # Handle checkout session completion: ensure purchase exists and store PaymentIntent / Charge ids if event["type"] == "checkout.session.completed": session = event["data"]["object"] session_id = session["id"] payment_status = _s(session, "payment_status") logger.info(f"Checkout session completed: {session}") # First try to find existing CoursePurchase by session id purchase = CoursePurchase.objects.filter( stripe_checkout_session_id=session_id ).first() if purchase: updated = False if ( payment_status == "paid" and purchase.status != CoursePurchase.Status.PAID ): purchase.status = CoursePurchase.Status.PAID updated = True if purchase.refunded: purchase.refunded = False updated = True if updated: purchase.save(update_fields=["status", "refunded"]) if payment_status == "paid": logger.info(f"Marked CoursePurchase {purchase.id} as PAID.") else: logger.info( "Processed checkout.session.completed for CoursePurchase %s without changing status (payment_status=%s).", purchase.id, payment_status, ) else: # No existing purchase — attempt to create one by inspecting the Stripe session try: session_obj = stripe.checkout.Session.retrieve( session_id, expand=[ "line_items.data.price.product", "customer", "customer_details", ], ) # Determine user from session metadata, client_reference_id or customer details from django.contrib.auth import get_user_model User = get_user_model() user = None metadata = _s(session_obj, "metadata") or {} # Prefer explicit user_id in metadata user_id = _s(metadata, "user_id") if user_id: try: user = User.objects.get(pk=int(user_id)) except Exception: user = None # Fallback to client_reference_id (often used to pass local user id) if not user: client_ref = _s(session_obj, "client_reference_id") if client_ref: try: user = User.objects.get(pk=int(client_ref)) except Exception: user = None # Fallback to customer email if not user: email = None cust_details = _s(session_obj, "customer_details") or {} email = _s(cust_details, "email") or _s( session_obj, "customer_email" ) if not email: customer = _s(session_obj, "customer") if isinstance(customer, dict): email = _s(customer, "email") if email: user = User.objects.filter(email=email).first() # Find the purchasable product from line items using product.metadata.local_id course = None line_items = _s(_s(session_obj, "line_items") or {}, "data") or [] for item in line_items: price = _s(item, "price") or {} product = _s(price, "product") if isinstance(product, dict): local_id = _s(_s(product, "metadata") or {}, "local_id") if local_id: try: purch = PurchasableProduct.objects.get(pk=int(local_id)) course = purch.course break except Exception: continue if user and course: try: with transaction.atomic(): purchase, created = CoursePurchase.objects.get_or_create( user=user, course=course, defaults={ "status": ( CoursePurchase.Status.PAID if payment_status == "paid" else CoursePurchase.Status.PENDING ), "refunded": False, "stripe_checkout_session_id": session_id, }, ) except IntegrityError: # Race: another worker created the purchase concurrently. Re-fetch. purchase = CoursePurchase.objects.filter( user=user, course=course ).first() if not purchase: purchase = CoursePurchase.objects.filter( stripe_checkout_session_id=session_id ).first() if purchase: # Ensure fields are up to date idempotently update_fields = [] if ( payment_status == "paid" and purchase.status != CoursePurchase.Status.PAID ): purchase.status = CoursePurchase.Status.PAID update_fields.append("status") if purchase.refunded: purchase.refunded = False update_fields.append("refunded") if not purchase.stripe_checkout_session_id: purchase.stripe_checkout_session_id = session_id update_fields.append("stripe_checkout_session_id") # Store PaymentIntent and Charge (if present) for later refund handling pi = _s(session_obj, "payment_intent") if pi and not purchase.stripe_payment_intent_id: purchase.stripe_payment_intent_id = pi update_fields.append("stripe_payment_intent_id") # Try to obtain charge id from PaymentIntent try: pi_obj = stripe.PaymentIntent.retrieve( pi, expand=["charges.data"] ) charges = _s(_s(pi_obj, "charges") or {}, "data") or [] if charges: charge_id = _s(charges[0], "id") if charge_id and not purchase.stripe_charge_id: purchase.stripe_charge_id = charge_id update_fields.append("stripe_charge_id") except Exception: logger.exception( "Failed to retrieve PaymentIntent/charges for storing charge id" ) if update_fields: purchase.save(update_fields=update_fields) logger.info( "Updated CoursePurchase %s (fields: %s).", purchase.id, update_fields, ) else: logger.info( "CoursePurchase %s already up-to-date.", purchase.id ) if created: logger.info( "Created CoursePurchase %s for user %s and course %s.", purchase.id, user, course, ) else: logger.warning( "Could not create CoursePurchase for session %s: user=%s, course=%s", session_id, user, course, ) else: logger.warning( "Could not create CoursePurchase for session %s: user=%s, course=%s", session_id, user, course, ) except Exception as e: logger.exception( f"Failed to create CoursePurchase for session {session_id}: {e}" ) # Refund-related events: charge.refunded, refund.created, refund.updated elif event["type"] in ("charge.refunded", "refund.created", "refund.updated"): obj = event["data"]["object"] # obj may be a Charge (for charge.refunded) or a Refund (for refund.*) obj_object = _s(obj, "object") if obj_object == "charge": charge_id = _s(obj, "id") else: charge_id = _s(obj, "charge") payment_intent_id = _s(obj, "payment_intent") # Try to find the CoursePurchase by charge id or payment_intent purchase = None if charge_id: purchase = CoursePurchase.objects.filter(stripe_charge_id=charge_id).first() if not purchase and payment_intent_id: purchase = CoursePurchase.objects.filter( stripe_payment_intent_id=payment_intent_id ).first() if purchase: _mark_purchase_refunded(purchase) else: logger.warning( "Received refund webhook but could not find CoursePurchase for charge=%s pi=%s", charge_id, payment_intent_id, ) elif event["type"] == "checkout.session.expired": session = event["data"]["object"] session_id = session["id"] try: purchase = CoursePurchase.objects.get(stripe_checkout_session_id=session_id) purchase.status = CoursePurchase.Status.FAILED purchase.save(update_fields=["status"]) logger.info(f"Marked CoursePurchase {purchase.id} as FAILED (expired).") except CoursePurchase.DoesNotExist: logger.warning(f"No CoursePurchase found for expired session {session_id}") elif event["type"] == "checkout.session.async_payment_failed": session = event["data"]["object"] session_id = session["id"] try: purchase = CoursePurchase.objects.get(stripe_checkout_session_id=session_id) purchase.status = CoursePurchase.Status.FAILED purchase.save(update_fields=["status"]) logger.info( f"Marked CoursePurchase {purchase.id} as FAILED (async payment failed)." ) except CoursePurchase.DoesNotExist: logger.warning(f"No CoursePurchase found for failed session {session_id}") elif event["type"] == "checkout.session.async_payment_succeeded": session = event["data"]["object"] session_id = session["id"] try: purchase = CoursePurchase.objects.get(stripe_checkout_session_id=session_id) purchase.status = CoursePurchase.Status.PAID purchase.refunded = False purchase.save(update_fields=["status", "refunded"]) logger.info( f"Marked CoursePurchase {purchase.id} as PAID (async succeeded)." ) except CoursePurchase.DoesNotExist: logger.warning( f"No CoursePurchase found for async succeeded session {session_id}" ) elif event["type"] == "payment_intent.created": pi = event["data"]["object"] pi_id = pi["id"] purchase = None try: # Try to find CoursePurchase by checkout session id present in PaymentIntent metadata metadata = _s(pi, "metadata") or {} session_id = ( _s(metadata, "checkout_session") or _s(metadata, "session_id") or _s(metadata, "stripe_checkout_session_id") ) if session_id: purchase = CoursePurchase.objects.filter( stripe_checkout_session_id=session_id ).first() # Fallback: try client_reference_id (may be copied into metadata) if not purchase: client_ref = _s(metadata, "client_reference_id") or _s( metadata, "client_id" ) if client_ref: try: purchase = ( CoursePurchase.objects.filter( user__pk=int(client_ref), refunded=False ) .order_by("-id") .first() ) except Exception: purchase = None # Fallback: try receipt email / billing details if not purchase: email = _s(pi, "receipt_email") if not email: charges = _s(_s(pi, "charges") or {}, "data") or [] if charges: billing = _s(charges[0], "billing_details") or {} email = _s(billing, "email") if email: purchase = ( CoursePurchase.objects.filter(user__email=email, refunded=False) .order_by("-id") .first() ) if purchase: update_fields = [] if not purchase.stripe_payment_intent_id and pi_id: purchase.stripe_payment_intent_id = pi_id update_fields.append("stripe_payment_intent_id") # If charge id is present in the PaymentIntent payload, store it too charges = _s(_s(pi, "charges") or {}, "data") or [] if charges: charge_id = _s(charges[0], "id") if charge_id and not purchase.stripe_charge_id: purchase.stripe_charge_id = charge_id update_fields.append("stripe_charge_id") if update_fields: purchase.save(update_fields=update_fields) logger.info( "Updated CoursePurchase %s with fields %s from payment_intent.created", purchase.id, update_fields, ) else: logger.debug( "payment_intent.created: CoursePurchase %s already has payment fields set", purchase.id, ) else: logger.info( "payment_intent.created: no matching CoursePurchase for pi=%s", pi_id, ) logger.debug("payment_intent.created payload: %s", pi) except Exception: logger.exception("Failed processing payment_intent.created") elif event["type"] == "payment_intent.payment_failed": pi = event["data"]["object"] pi_id = pi["id"] try: purchase = _find_purchase_from_payment_intent(pi) if purchase: if purchase.status != CoursePurchase.Status.FAILED: purchase.status = CoursePurchase.Status.FAILED purchase.save(update_fields=["status"]) logger.info( f"Marked CoursePurchase {purchase.id} as FAILED (payment intent failed)." ) else: raise CoursePurchase.DoesNotExist except CoursePurchase.DoesNotExist: logger.warning(f"No CoursePurchase found for failed payment intent {pi_id}") return HttpResponse(status=200)