schon/engine/vibes_auth/emailing/views.py

from uuid import UUID

from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import OpenApiParameter, extend_schema
from rest_framework import status
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from rest_framework.views import APIView

from engine.vibes_auth.models import User


class UnsubscribeView(APIView):
    """
    Public endpoint for one-click unsubscribe from email campaigns.

    Supports both GET (for email client compatibility) and POST (RFC 8058).
    """

    permission_classes = [AllowAny]
    authentication_classes = []

    @extend_schema(
        summary="Unsubscribe from email campaigns",
        description="Unsubscribe a user from email campaigns using their unsubscribe token.",
        parameters=[
            OpenApiParameter(
                name="token",
                description="Unsubscribe token from the email",
                required=True,
                type=str,
            ),
        ],
        responses={
            200: {"description": "Successfully unsubscribed"},
            400: {"description": "Invalid or missing token"},
            404: {"description": "User not found"},
        },
    )
    def get(self, request):
        """Handle GET request for unsubscribe (email link click)."""
        return self._process_unsubscribe(request)

    @extend_schema(
        summary="Unsubscribe from email campaigns (One-Click)",
        description="RFC 8058 compliant one-click unsubscribe endpoint.",
        parameters=[
            OpenApiParameter(
                name="token",
                description="Unsubscribe token from the email",
                required=True,
                type=str,
            ),
        ],
        responses={
            200: {"description": "Successfully unsubscribed"},
            400: {"description": "Invalid or missing token"},
            404: {"description": "User not found"},
        },
    )
    def post(self, request):
        """Handle POST request for one-click unsubscribe (RFC 8058)."""
        return self._process_unsubscribe(request)

    def _process_unsubscribe(self, request) -> Response:
        """Process the unsubscribe request."""
        token = request.query_params.get("token") or request.data.get("token")

        if not token:
            return Response(
                {"detail": _("Unsubscribe token is required.")},
                status=status.HTTP_400_BAD_REQUEST,
            )

        try:
            token_uuid = UUID(token)
        except (ValueError, TypeError):
            return Response(
                {"detail": _("Invalid unsubscribe token format.")},
                status=status.HTTP_400_BAD_REQUEST,
            )

        try:
            user = User.objects.get(unsubscribe_token=token_uuid)
        except User.DoesNotExist:
            return Response(
                {"detail": _("User not found.")},
                status=status.HTTP_404_NOT_FOUND,
            )

        if not user.is_subscribed:
            return Response(
                {"detail": _("You are already unsubscribed.")},
                status=status.HTTP_200_OK,
            )

        user.is_subscribed = False
        user.save(update_fields=["is_subscribed", "modified"])

        return Response(
            {"detail": _("You have been successfully unsubscribed from our emails.")},
            status=status.HTTP_200_OK,
        )


class TrackingView(APIView):
    """
    Endpoint for tracking email opens and clicks.

    This is optional - can be used to track engagement metrics.
    """

    permission_classes = [AllowAny]
    authentication_classes = []

    @extend_schema(
        summary="Track email open",
        description="Track when a campaign email is opened.",
        parameters=[
            OpenApiParameter(
                name="tid",
                description="Tracking ID from the email",
                required=True,
                type=str,
            ),
        ],
        responses={
            200: {"description": "Tracking recorded"},
            404: {"description": "Invalid tracking ID"},
        },
    )
    def get(self, request):
        """Track email open via tracking pixel."""
        from django.utils import timezone

        from engine.vibes_auth.emailing.choices import RecipientStatus
        from engine.vibes_auth.emailing.models import CampaignRecipient

        tracking_id = request.query_params.get("tid")

        if not tracking_id:
            return Response(status=status.HTTP_404_NOT_FOUND)

        try:
            tracking_uuid = UUID(tracking_id)
            recipient = CampaignRecipient.objects.get(tracking_id=tracking_uuid)

            if not recipient.opened_at:
                recipient.opened_at = timezone.now()
                recipient.status = RecipientStatus.OPENED
                recipient.save(update_fields=["opened_at", "status", "modified"])

                # Update campaign opened count
                campaign = recipient.campaign
                campaign.opened_count = campaign.recipients.filter(
                    status__in=(RecipientStatus.OPENED, RecipientStatus.CLICKED)
                ).count()
                campaign.save(update_fields=["opened_count", "modified"])

        except (ValueError, TypeError, CampaignRecipient.DoesNotExist):
            pass  # Silently ignore invalid tracking IDs

        # Return a 1x1 transparent GIF
        gif_data = (
            b"GIF89a\x01\x00\x01\x00\x80\x00\x00\xff\xff\xff"
            b"\x00\x00\x00!\xf9\x04\x01\x00\x00\x00\x00,\x00"
            b"\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02D\x01\x00;"
        )
        return Response(
            gif_data,
            status=status.HTTP_200_OK,
            content_type="image/gif",
        )