Merge pull request #1226 from thunderstore-io/feature/TS-2985/kafka-profile-export-event

Profile Export Events to Kafka

Author: Roffenlund

django/thunderstore/modpacks/api/experimental/tests/test_legacyprofile.py

@@ -1,14 +1,17 @@
 import io
 import json
+from unittest.mock import patch
 
 import pytest
 import requests
 from django.core.files import File
+from django.db import connection
 from django.urls import reverse
 from rest_framework.test import APIClient
 
 from thunderstore.modpacks.models import LegacyProfile
 from thunderstore.modpacks.models.legacyprofile import LEGACYPROFILE_STORAGE_CAP
+from thunderstore.ts_analytics.kafka import KafkaTopic
 
 
 @pytest.mark.django_db
@@ -107,3 +110,41 @@ def test_experimental_api_legacyprofile_retrieve(api_client: APIClient) -> None:
     response = requests.get(response["Location"])
     assert response.status_code == 200
     assert response.content == test_content
+
+
+@pytest.mark.django_db(transaction=True)
+def test_experimental_api_legacyprofile_create_sends_kafka_event(
+    api_client: APIClient,
+) -> None:
+    with patch(
+        "thunderstore.modpacks.api.experimental.views.legacyprofile.send_kafka_message"
+    ) as mock_send_kafka_message:
+        assert LegacyProfile.objects.count() == 0
+        test_content = b"test profile data"
+
+        response = api_client.post(
+            reverse("api:experimental:legacyprofile.create"),
+            data=test_content,
+            content_type="application/octet-stream",
+        )
+
+        assert response.status_code == 200
+
+        while connection.run_on_commit:
+            sids, func = connection.run_on_commit.pop(0)
+            func()
+
+        result = response.json()
+        profile_key = result["key"]
+
+        mock_send_kafka_message.delay.assert_called_once()
+        call_args = mock_send_kafka_message.delay.call_args
+
+        assert call_args.kwargs["topic"] == KafkaTopic.A_LEGACY_PROFILE_EXPORT_V1
+
+        payload_string = call_args.kwargs["payload_string"]
+        payload = json.loads(payload_string)
+
+        assert payload["id"] == profile_key
+        assert "timestamp" in payload
+        assert payload["file_size_bytes"] == len(test_content)

django/thunderstore/modpacks/api/experimental/views/legacyprofile.py

@@ -1,7 +1,12 @@
+import logging
+from datetime import datetime
+
+from django.db import transaction
 from django.shortcuts import redirect
 from django.utils import timezone
 from drf_yasg.openapi import TYPE_FILE, Schema
 from drf_yasg.utils import swagger_auto_schema
+from pydantic import BaseModel
 from rest_framework import serializers, status
 from rest_framework.exceptions import NotFound, ValidationError
 from rest_framework.parsers import FileUploadParser
@@ -11,6 +16,10 @@
 
 from thunderstore.core.utils import replace_cdn
 from thunderstore.modpacks.models import LegacyProfile
+from thunderstore.ts_analytics.kafka import KafkaTopic
+from thunderstore.ts_analytics.tasks import send_kafka_message
+
+logger = logging.getLogger(__name__)
 
 
 class LegacyProfileCreateResponseSerializer(serializers.Serializer):
@@ -29,6 +38,12 @@ def get_rate(self) -> str:
         return "6/m"
 
 
+class AnalyticsEventLegacyProfileExport(BaseModel):
+    id: str
+    timestamp: datetime
+    file_size_bytes: int
+
+
 class LegacyProfileCreateApiView(APIView):
     permission_classes = []
     parser_classes = [LegacyProfileFileUploadParser]
@@ -43,9 +58,23 @@ class LegacyProfileCreateApiView(APIView):
     def post(self, request, *args, **kwargs):
         if "file" not in self.request.data or not self.request.data["file"]:
             raise ValidationError(detail="Request body was empty")
-        key = LegacyProfile.objects.get_or_create_from_upload(
-            content=self.request.data["file"]
+
+        file_obj = self.request.data["file"]
+        file_size = file_obj.size
+
+        key = LegacyProfile.objects.get_or_create_from_upload(content=file_obj)
+
+        transaction.on_commit(
+            lambda: send_kafka_message.delay(
+                topic=KafkaTopic.A_LEGACY_PROFILE_EXPORT_V1,
+                payload_string=AnalyticsEventLegacyProfileExport(
+                    id=str(key),
+                    timestamp=timezone.now(),
+                    file_size_bytes=file_size,
+                ).json(),
+            )
         )
+
         serializer = LegacyProfileCreateResponseSerializer({"key": key})
         return Response(
             serializer.data,

django/thunderstore/ts_analytics/kafka.py

@@ -17,6 +17,7 @@ class KafkaTopic(str, Enum):
     # 'analytics' prefixed events are events that aren't necessarily bound to
     # any specific data structure
     A_PACKAGE_DOWNLOAD_V1 = "analytics.package_download.v1"
+    A_LEGACY_PROFILE_EXPORT_V1 = "analytics.legacy_profile_export.v1"
 
     # 'model' prefixed events are generally events that are explicitly bound to
     # models. Use Django's db table naming convention.