feat: Add promo code activation and history endpoints

This commit introduces two new endpoints for user interactions with promo codes:

- `POST /api/user/promo/{id}/activate`: Allows an authenticated user to activate a specific promo code. The activation is subject to several validation checks, including
      promo targeting rules (age, country), promo activity status, and a new anti-fraud service verification.

- `GET /api/user/promo/history`: Retrieves a paginated list of promo codes that the current user has previously activated.

Key changes include:
- A `PromoActivateView` that handles the logic for activating a promo code, including validation and atomic updates to the promo's usage count or unique code status.
- A `PromoHistoryView` to display the user's activation history.
- A new `antifraud_service` module to communicate with an external anti-fraud system, complete with caching and retry logic.
- A `PromoActivationHistory` model to log each successful activation, linking users to the promos they've activated.
 - Updates to `UserPromoDetailSerializer` to include an `is_activated_by_user` field, indicating if the user has already activated the promo.
- New URL patterns in `user/urls.py` to route to the new views.

Author: RandomProgramm3r

promo_code/business/validators.py

@@ -42,6 +42,7 @@ def _get_full_data(self):
                     'max_count': self.instance.max_count,
                     'active_from': self.instance.active_from,
                     'active_until': self.instance.active_until,
+                    'used_count': self.instance.used_count,
                     'target': self.instance.target
                     if self.instance.target
                     else {},
@@ -56,6 +57,7 @@ def validate(self):
         promo_common = self.full_data.get('promo_common')
         promo_unique = self.full_data.get('promo_unique')
         max_count = self.full_data.get('max_count')
+        used_count = self.full_data.get('used_count')
 
         if mode not in [
             business.constants.PROMO_MODE_COMMON,
@@ -65,6 +67,11 @@ def validate(self):
                 {'mode': 'Invalid mode.'},
             )
 
+        if used_count and used_count > max_count:
+            raise rest_framework.exceptions.ValidationError(
+                {'mode': 'Invalid max_count.'},
+            )
+
         if mode == business.constants.PROMO_MODE_COMMON:
             if not promo_common:
                 raise rest_framework.exceptions.ValidationError(

promo_code/promo_code/settings.py

@@ -153,6 +153,7 @@ def load_bool(name, default):
 
 ANTIFRAUD_ADDRESS = f'{os.getenv("ANTIFRAUD_ADDRESS")}'
 ANTIFRAUD_VALIDATE_URL = f'{ANTIFRAUD_ADDRESS}/api/validate'
+ANTIFRAUD_SET_DELAY_URL = f'{ANTIFRAUD_ADDRESS}/internal/set_delay'
 ANTIFRAUD_UPDATE_USER_VERDICT_URL = (
     f'{ANTIFRAUD_ADDRESS}/internal/update_user_verdict'
 )

promo_code/user/antifraud_service.py

@@ -0,0 +1,100 @@
+import datetime
+import json
+import typing
+
+import django.conf
+import django.core.cache
+import requests
+import requests.exceptions
+
+
+class AntiFraudService:
+    """
+    A service class to interact with the anti-fraud system.
+
+    Encapsulates caching, HTTP requests, and error handling.
+    """
+
+    def __init__(
+        self,
+        base_url: str = django.conf.settings.ANTIFRAUD_VALIDATE_URL,
+        timeout: int = 5,
+        max_retries: int = 2,
+    ):
+        self.base_url = base_url
+        self.timeout = timeout
+        self.max_retries = max_retries
+
+    def get_verdict(self, user_email: str, promo_id: str) -> typing.Dict:
+        """
+        Retrieves the anti-fraud verdict for a given user and promo.
+
+        1. Checks the cache.
+        2. If not in cache, fetches from the anti-fraud service.
+        3. Caches the result if the service provides a 'cache_until' value.
+        """
+        cache_key = f'antifraud_verdict_{user_email}'
+
+        if cached_verdict := django.core.cache.cache.get(cache_key):
+            return cached_verdict
+
+        verdict = self._fetch_from_service(user_email, promo_id)
+
+        if verdict.get('ok'):
+            timeout_seconds = self._calculate_cache_timeout(
+                verdict.get('cache_until'),
+            )
+            if timeout_seconds:
+                django.core.cache.cache.set(
+                    cache_key, verdict, timeout=timeout_seconds,
+                )
+
+        return verdict
+
+    def _fetch_from_service(
+        self, user_email: str, promo_id: str,
+    ) -> typing.Dict:
+        """
+        Performs the actual HTTP request with a retry mechanism.
+        """
+        payload = {'user_email': user_email, 'promo_id': promo_id}
+
+        for _ in range(self.max_retries):
+            try:
+                response = requests.post(
+                    self.base_url,
+                    json=payload,
+                    timeout=self.timeout,
+                )
+                response.raise_for_status()
+                return response.json()
+            except (
+                requests.exceptions.RequestException,
+                json.JSONDecodeError,
+            ):
+                continue
+
+        return {'ok': False, 'error': 'Anti-fraud service unavailable'}
+
+    @staticmethod
+    def _calculate_cache_timeout(
+        cache_until_str: typing.Optional[str],
+    ) -> typing.Optional[int]:
+        """
+        Safely parses an ISO format date string and returns a cache TTL in seconds.
+        """
+        if not cache_until_str:
+            return None
+
+        try:
+            naive_dt = datetime.datetime.fromisoformat(cache_until_str)
+            aware_dt = naive_dt.replace(tzinfo=datetime.timezone.utc)
+            now = datetime.datetime.now(datetime.timezone.utc)
+
+            timeout_seconds = (aware_dt - now).total_seconds()
+            return int(timeout_seconds) if timeout_seconds > 0 else None
+        except (ValueError, TypeError):
+            return None
+
+
+antifraud_service = AntiFraudService()

promo_code/user/migrations/0004_promoactivationhistory.py

@@ -0,0 +1,52 @@
+# Generated by Django 5.2 on 2025-07-01 11:16
+
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("business", "0004_promo_comment_count"),
+        ("user", "0003_promocomment"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="PromoActivationHistory",
+            fields=[
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="UUID",
+                    ),
+                ),
+                ("activated_at", models.DateTimeField(auto_now_add=True)),
+                (
+                    "promo",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="activations_history",
+                        to="business.promo",
+                    ),
+                ),
+                (
+                    "user",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="promo_activations",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            options={
+                "ordering": ["-activated_at"],
+            },
+        ),
+    ]

promo_code/user/models.py

@@ -148,3 +148,29 @@ class Meta:
 
     def __str__(self):
         return f'Comment by {self.author.email} on promo {self.promo.id}'
+
+
+class PromoActivationHistory(django.db.models.Model):
+    id = django.db.models.UUIDField(
+        'UUID',
+        primary_key=True,
+        default=uuid.uuid4,
+        editable=False,
+    )
+    user = django.db.models.ForeignKey(
+        User,
+        on_delete=django.db.models.CASCADE,
+        related_name='promo_activations',
+    )
+    promo = django.db.models.ForeignKey(
+        business.models.Promo,
+        on_delete=django.db.models.CASCADE,
+        related_name='activations_history',
+    )
+    activated_at = django.db.models.DateTimeField(auto_now_add=True)
+
+    class Meta:
+        ordering = ['-activated_at']
+
+    def __str__(self):
+        return f'{self.user} activated {self.promo.id} at {self.activated_at}'

promo_code/user/serializers.py

@@ -449,9 +449,22 @@ def get_is_liked_by_user(self, obj: business.models.Promo) -> bool:
             ).exists()
         return False
 
-    def get_is_activated_by_user(self, obj) -> bool:
-        # TODO:
-        return False
+    def get_is_activated_by_user(self, obj: business.models.Promo) -> bool:
+        """
+        Checks whether the current user has activated this promo code.
+        """
+        request = self.context.get('request')
+        if not (
+            request
+            and hasattr(request, 'user')
+            and request.user.is_authenticated
+        ):
+            return False
+
+        return user.models.PromoActivationHistory.objects.filter(
+            promo=obj,
+            user=request.user,
+        ).exists()
 
 
 class UserAuthorSerializer(rest_framework.serializers.ModelSerializer):
@@ -514,3 +527,11 @@ class CommentUpdateSerializer(rest_framework.serializers.ModelSerializer):
     class Meta:
         model = user.models.PromoComment
         fields = ('text',)
+
+
+class PromoActivationSerializer(rest_framework.serializers.Serializer):
+    """
+    Serializer for the response upon successful activation.
+    """
+
+    promo = rest_framework.serializers.CharField()

promo_code/user/urls.py

@@ -52,4 +52,14 @@
         user.views.PromoCommentDetailView.as_view(),
         name='user-promo-comment-detail',
     ),
+    django.urls.path(
+        'promo/<uuid:id>/activate',
+        user.views.PromoActivateView.as_view(),
+        name='user-promo-activate',
+    ),
+    django.urls.path(
+        'promo/history',
+        user.views.PromoHistoryView.as_view(),
+        name='user-promo-history',
+    ),
 ]