vbv/server/vbv_lernwelt/shop/views.py

import structlog
from django.http import JsonResponse
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response

from vbv_lernwelt.core.admin import User
from vbv_lernwelt.course.models import CourseSession, CourseSessionUser
from vbv_lernwelt.shop.models import (
    BillingAddress,
    CheckoutInformation,
    CheckoutState,
    Product,
    VV_PRODUCT_SKU,
)
from vbv_lernwelt.shop.serializers import BillingAddressSerializer
from vbv_lernwelt.shop.services import (
    get_payment_url,
    init_transaction,
    is_signature_valid,
)

logger = structlog.get_logger(__name__)


@api_view(["GET"])
@permission_classes([IsAuthenticated])
def get_billing_address(request):
    try:
        billing_address = BillingAddress.objects.get(user=request.user)
        data = BillingAddressSerializer(billing_address).data
    except BillingAddress.DoesNotExist:
        data = BillingAddressSerializer().data
        data["first_name"] = request.user.first_name  # noqa
        data["last_name"] = request.user.last_name  # noqa

    return Response(data)


@api_view(["PUT"])
@permission_classes([IsAuthenticated])
def update_billing_address(request):
    try:
        billing_address = BillingAddress.objects.get(user=request.user)
    except BillingAddress.DoesNotExist:
        billing_address = None

    serializer = BillingAddressSerializer(
        billing_address, data=request.data, partial=True
    )

    if serializer.is_valid():
        serializer.save(user=request.user)
        return Response(serializer.data)

    return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)


@api_view(["POST"])
def transaction_webhook(request):
    logger.info("Webhook: Datatrans called transaction webhook", body=request.body)

    if not is_signature_valid(
        signature=request.headers.get("Datatrans-Signature", ""),
        payload=request.body,
    ):
        logger.warning("Invalid signature")
        return JsonResponse({"status": "invalid signature"}, status=400)

    transaction = request.data
    transaction_id = transaction["transactionId"]
    transaction_status = transaction["status"]

    checkout_info = CheckoutInformation.objects.get(transaction_id=transaction_id)
    checkout_info.state = transaction_status
    checkout_info.webhook_history.append(transaction)
    checkout_info.save(update_fields=["state", "webhook_history"])

    if transaction_status in ["settled", "transmitted"]:
        logger.info(
            "Webhook: Create course session user", transaction_id=transaction_id
        )

        # FIXME make this "better"
        cs_vv = CourseSession.objects.filter(
            title="Versicherungsvermittler/-in"
        ).first()

        CourseSessionUser.objects.get_or_create(
            course_session=cs_vv,
            user=User.objects.get(id=checkout_info.user_id),
            role=CourseSessionUser.Role.MEMBER,
        )

    return JsonResponse({"status": "ok"})


@api_view(["POST"])
@permission_classes([IsAuthenticated])
def checkout_vv(request):
    """
    Checkout for the Versicherungsvermittler product (VV).

    VV_PRODUCT_SKU: The one and only product, thus hardcoded
        Expected to be created in the admin interface first.
    """
    sku = VV_PRODUCT_SKU
    logger.info(f"Checkout requested: sku={sku}", user_id=request.user.id)
    base_redirect_url = request.data["redirect_url"]

    try:
        product = Product.objects.get(sku=sku)
    except Product.DoesNotExist:
        return JsonResponse(
            {
                "next_step_url": checkout_error_url(
                    base_url=base_redirect_url,
                    message="vv_product_does_not_exist_needs_to_be_created",
                )
            },
        )

    checkouts = CheckoutInformation.objects.filter(
        user=request.user,
        product_sku=sku,
    )

    # already purchased (settled or transmitted)
    if checkouts.filter(
        state__in=[CheckoutState.SETTLED, CheckoutState.TRANSMITTED]
    ).exists():
        return JsonResponse({"next_step_url": "/"})

    # already initialized -> redirect to payment page again
    if checkout := checkouts.filter(state=CheckoutState.INITIALIZED).first():
        return JsonResponse({"next_step_url": get_payment_url(checkout.transaction_id)})

    # not yet initialized at all, or canceled/failed
    transaction_id = init_transaction(
        user=request.user,
        amount_chf_centimes=product.price,
        redirect_url_success=checkout_success_url(base_redirect_url),
        redirect_url_error=checkout_error_url(base_redirect_url),
        redirect_url_cancel=checkout_cancel_url(base_redirect_url),
        webhook_url=webhook_url(base_redirect_url),
    )

    # immutable snapshot of data at time of purchase
    CheckoutInformation.objects.create(
        user=request.user,
        state="initialized",
        transaction_id=transaction_id,
        # product
        product_sku=sku,
        product_price=product.price,
        product_name=product.name,
        product_description=product.description,
        # address
        **request.data["address"],
    )

    return JsonResponse({"next_step_url": get_payment_url(transaction_id)})


def webhook_url(base_url: str) -> str:
    return f"{base_url}/api/shop/transaction/webhook/"


def checkout_error_url(base_url: str, message: str | None = None) -> str:
    url = f"{base_url}/onboarding/vv/checkout/address?error"

    if message:
        url += f"&message={message}"

    return url


def checkout_cancel_url(base_url: str) -> str:
    return f"{base_url}/"


def checkout_success_url(base_url: str = "") -> str:
    return f"{base_url}/onboarding/vv/checkout/complete"