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.course.consts import (
    COURSE_VERSICHERUNGSVERMITTLERIN_ID,
    COURSE_VERSICHERUNGSVERMITTLERIN_FR_ID,
    COURSE_VERSICHERUNGSVERMITTLERIN_IT_ID,
)
from vbv_lernwelt.course.models import CourseSession, CourseSessionUser
from vbv_lernwelt.shop.const import (
    VV_DE_PRODUCT_SKU,
    VV_FR_PRODUCT_SKU,
    VV_IT_PRODUCT_SKU,
)
from vbv_lernwelt.shop.models import (
    BillingAddress,
    CheckoutInformation,
    CheckoutState,
    Product,
)
from vbv_lernwelt.shop.serializers import BillingAddressSerializer
from vbv_lernwelt.shop.services import (
    datatrans_state_to_checkout_state,
    get_payment_url,
    init_transaction,
    InitTransactionException,
    is_signature_valid,
)

logger = structlog.get_logger(__name__)

PRODUCT_SKU_TO_COURSE = {
    VV_DE_PRODUCT_SKU: COURSE_VERSICHERUNGSVERMITTLERIN_ID,
    VV_FR_PRODUCT_SKU: COURSE_VERSICHERUNGSVERMITTLERIN_FR_ID,
    VV_IT_PRODUCT_SKU: COURSE_VERSICHERUNGSVERMITTLERIN_IT_ID,
}


@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"]

    # keep webhook history (for debugging)
    checkout_info = CheckoutInformation.objects.get(transaction_id=transaction_id)
    checkout_info.webhook_history.append(transaction)
    checkout_info.save(update_fields=["webhook_history"])

    # update checkout state
    checkout_state = datatrans_state_to_checkout_state(transaction["status"])
    update_checkout_state(checkout_info=checkout_info, state=checkout_state)

    # handle paid
    if checkout_state == CheckoutState.PAID:
        create_vv_course_session_user(checkout_info=checkout_info)

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


@api_view(["POST"])
@permission_classes([IsAuthenticated])
def checkout_vv(request):
    """
    Checkout for the Versicherungsvermittler products (vv-de, vv-fr, vv-it)
    """
    sku = request.data["product"]
    base_redirect_url = request.data["redirect_url"]

    logger.info(f"Checkout requested: sku={sku}", user_id=request.user.id)

    try:
        product = Product.objects.get(sku=sku)
    except Product.DoesNotExist:
        return next_step_response(
            url=checkout_error_url(
                base_url=base_redirect_url,
                message=f"{sku}_product_sku_does_not_exist_needs_to_be_created_first",
            ),
        )

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

    # already paid (successfully)-> redirect to home
    if checkouts.filter(state__in=[CheckoutState.PAID]).exists():
        return next_step_response(url="/")

    # already initialized -> redirect to payment page again
    if checkout := checkouts.filter(state=CheckoutState.INITIALIZED).first():
        return next_step_response(url=get_payment_url(checkout.transaction_id))

    # not yet initialized at all, or canceled/failed
    # -> create new transaction and checkout
    try:
        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),
        )
    except InitTransactionException as e:
        return next_step_response(
            url=checkout_error_url(
                base_url=base_redirect_url,
            ),
        )

    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 next_step_response(url=get_payment_url(transaction_id))


def update_checkout_state(checkout_info: CheckoutInformation, state: CheckoutState):
    checkout_info.state = state.value
    checkout_info.save(update_fields=["state"])


def create_vv_course_session_user(checkout_info: CheckoutInformation):
    logger.info("Creating VV course session user", user_id=checkout_info.user_id)

    CourseSessionUser.objects.get_or_create(
        user=checkout_info.user,
        role=CourseSessionUser.Role.MEMBER,
        course_session=CourseSession.objects.filter(
            course_id=PRODUCT_SKU_TO_COURSE[checkout_info.product_sku]
        ).first(),
    )


def next_step_response(
    url: str,
) -> JsonResponse:
    return JsonResponse(
        {
            "next_step_url": url,
        },
    )


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"