add event tracking

commcare_connect/events/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class EventsAppConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "commcare_connect.events"

commcare_connect/events/migrations/0001_initial.py

@@ -0,0 +1,53 @@
+# Generated by Django 4.2.5 on 2024-07-13 14:45
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("opportunity", "0044_opportunityverificationflags"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="Event",
+            fields=[
+                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("date_created", models.DateTimeField(auto_now_add=True, db_index=True)),
+                (
+                    "event_type",
+                    models.CharField(
+                        choices=[
+                            ("invite_sent", "Invite Sent"),
+                            ("records_approved", "Records Approved"),
+                            ("records_flagged", "Records Flagged"),
+                            ("records_rejected", "Records Rejected"),
+                            ("payment_approved", "Payment Approved"),
+                            ("payment_accrued", "Payment Accrued"),
+                            ("payment_transferred", "Payment Transferred"),
+                            ("notifications_sent", "Notifications Sent"),
+                            ("additional_budget_added", "Additional Budget Added"),
+                        ],
+                        max_length=40,
+                    ),
+                ),
+                (
+                    "opportunity",
+                    models.ForeignKey(
+                        null=True, on_delete=django.db.models.deletion.PROTECT, to="opportunity.opportunity"
+                    ),
+                ),
+                (
+                    "user",
+                    models.ForeignKey(
+                        null=True, on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL
+                    ),
+                ),
+            ],
+        ),
+    ]

commcare_connect/events/models.py

@@ -0,0 +1,144 @@
+from abc import ABCMeta, abstractproperty
+from dataclasses import dataclass
+from datetime import datetime
+
+from django.db import models
+from django.utils.translation import gettext as _
+
+from commcare_connect.users.models import User
+
+from . import types
+
+
+class Event(models.Model):
+    from commcare_connect.opportunity.models import Opportunity
+
+    # this allows referring to event types in this style: Event.Type.INVITE_SENT
+    Type = types
+
+    date_created = models.DateTimeField(auto_now_add=True, db_index=True)
+    event_type = models.CharField(max_length=40, choices=types.EVENT_TYPE_CHOICES)
+    user = models.ForeignKey(User, on_delete=models.CASCADE, null=True)
+    opportunity = models.ForeignKey(Opportunity, on_delete=models.PROTECT, null=True)
+
+    def track(self, use_async=True):
+        """
+        To track an event, instantiate the object and call this method,
+        instead of calling save directly.
+
+        If use_async is True, the event is queued in Redis and saved
+        via celery, otherwise it's saved directly.
+        """
+        from commcare_connect.events.tasks import track_event
+
+        track_event(self, use_async=use_async)
+
+
+@dataclass
+class InferredEvent:
+    from commcare_connect.opportunity.models import Opportunity
+
+    user: User
+    opportunity: Opportunity
+    date_created: datetime
+    event_type: str
+
+
+class InferredEventSpec(metaclass=ABCMeta):
+    """
+    Use this to define an Event that can be inferred
+    based on other models. See RecordsFlagged for example
+    """
+
+    @abstractproperty
+    def model_cls(self):
+        """
+        The source model class to infer the event from
+        for e.g. UserVisit
+        """
+        raise NotImplementedError
+
+    @abstractproperty
+    def event_type(self):
+        """
+        Should be a tuple to indicate the name
+        for e.g. "RECORDS_FLAGGED", gettext("Records Flagged")
+        """
+        raise NotImplementedError
+
+    @abstractproperty
+    def event_filters(self):
+        """
+        Should be a dict of the queryset filters
+        for e.g. {'flagged': True} for RecordsFlagged for UserVisit
+        """
+        raise NotImplementedError
+
+    @abstractproperty
+    def user(self):
+        """
+        The field corresponding to user on source model.
+        """
+        raise NotImplementedError
+
+    @abstractproperty
+    def date_created(self):
+        """
+        The field corresponding to user date_created.
+        """
+        raise NotImplementedError
+
+    @abstractproperty
+    def opportunity(self):
+        """
+        The field corresponding to user opportunity, could be None.
+        """
+        raise NotImplementedError
+
+    def get_events(self, user=None, from_date=None, to_date=None):
+        filters = {}
+        filters.update(self.event_filters)
+
+        if user:
+            filters.update({self.user: user})
+        if from_date:
+            filters.update({f"{self.date_created}__gte": from_date})
+        if to_date:
+            filters.update({f"{self.date_created}__lte": to_date})
+
+        events = (
+            self.model_cls.objects.filter(**filters).values(self.user, self.opportunity, self.date_created).iterator()
+        )
+        for event in events:
+            yield InferredEvent(
+                event_type=self.event_type[0],
+                user=event[self.user],
+                date_created=event[self.date_created],
+                opportunity=event.get(self.opportunity, None),
+            )
+
+
+class RecordsFlagged(InferredEventSpec):
+    event_type = (types.RECORDS_FLAGGED, _("Records Flagged"))
+    model_cls = "UserVisit"
+    event_filters = {"flagged": True}
+    user = "user"
+    date_created = "visit_date"
+    opportunity = "opportunity"
+
+
+INFERRED_EVENT_SPECS = [RecordsFlagged()]
+
+
+def get_events(user=None, from_date=None, to_date=None):
+    filters = {
+        "user": user,
+        "date_created__gte": from_date,
+        "date_created__lte": to_date,
+    }
+    filters = {k: v for k, v in filters.items() if v is not None}
+    raw_events = Event.objects.filter(**filters).all()
+    inferred_events = []
+    for event_spec in INFERRED_EVENT_SPECS:
+        inferred_events += list(event_spec.get_events(user=user, from_date=from_date, to_date=to_date))
+    return list(raw_events) + inferred_events

commcare_connect/events/tasks.py

@@ -0,0 +1,47 @@
+import pickle
+from datetime import datetime
+
+from django.db import transaction
+from django_redis import get_redis_connection
+
+from commcare_connect.events.models import Event
+from config import celery_app
+
+REDIS_EVENTS_QUEUE = "events_queue"
+
+
+class EventQueue:
+    def __init__(self):
+        self.redis_conn = get_redis_connection("default")
+
+    def push(self, event_obj):
+        serialized_event = pickle.dumps(event_obj)
+        self.redis_conn.rpush(REDIS_EVENTS_QUEUE, serialized_event)
+
+    def pop(self):
+        events = [pickle.loads(event) for event in self.redis_conn.lrange(REDIS_EVENTS_QUEUE, 0, -1)]
+        self.redis_conn.ltrim(REDIS_EVENTS_QUEUE, len(events), -1)
+        return events
+
+
+@celery_app.task
+def process_events_batch():
+    event_queue = EventQueue()
+    events = event_queue.pop()
+    if not events:
+        return
+    try:
+        with transaction.atomic():
+            Event.objects.bulk_create(events)
+    except Exception as e:
+        for event in events:
+            event_queue.push(event)
+        raise e
+
+
+def track_event(event_obj, use_async=True):
+    event_obj.date_created = datetime.utcnow()
+    if use_async:
+        EventQueue().push(event_obj)
+    else:
+        event_obj.save()

commcare_connect/events/tests.py

@@ -0,0 +1,60 @@
+from rest_framework.test import APIClient
+
+from commcare_connect.opportunity.models import Opportunity
+from commcare_connect.users.models import User
+
+from .models import Event
+from .tasks import EventQueue, process_events_batch
+
+
+def test_post_events(mobile_user_with_connect_link: User, api_client: APIClient, opportunity: Opportunity):
+    api_client.force_authenticate(mobile_user_with_connect_link)
+    response = api_client.post(
+        "/api/events/",
+        data=[
+            {
+                "event_type": "invalid_event_name",
+                "user": mobile_user_with_connect_link.pk,
+                "opportunity": opportunity.pk,
+            }
+        ],
+        format="json",
+    )
+    assert response.status_code == 400
+    assert Event.objects.count() == 0
+    response = api_client.post(
+        "/api/events/",
+        data=[
+            {
+                "event_type": Event.Type.INVITE_SENT,
+                "user": mobile_user_with_connect_link.pk,
+                "opportunity": opportunity.pk,
+            },
+            {
+                "event_type": Event.Type.RECORDS_APPROVED,
+                "user": mobile_user_with_connect_link.pk,
+                "opportunity": opportunity.pk,
+            },
+        ],
+        format="json",
+    )
+    assert response.status_code == 201
+    assert Event.objects.count() == 2
+
+
+def test_event_queue(mobile_user_with_connect_link: User, opportunity: Opportunity):
+    event_queue = EventQueue()
+    assert event_queue.pop() == []
+
+    # queue the event
+    event = Event(event_type=Event.Type.INVITE_SENT, user=mobile_user_with_connect_link, opportunity=opportunity)
+    event.track()
+    queued_events = event_queue.pop()
+    process_events_batch()
+    assert len(queued_events) == 1
+    assert Event.objects.count() == 0
+    # process the batch
+    event.track()
+    process_events_batch()
+    assert Event.objects.count() == 1
+    assert Event.objects.first().user == event.user

commcare_connect/events/types.py

@@ -0,0 +1,28 @@
+from django.utils.translation import gettext as _
+
+# Server/Web events
+INVITE_SENT = "invite_sent"
+RECORDS_APPROVED = "records_approved"
+RECORDS_REJECTED = "records_rejected"
+PAYMENT_APPROVED = "payment_approved"
+PAYMENT_ACCRUED = "payment_accrued"
+PAYMENT_TRANSFERRED = "payment_transferred"
+NOTIFICATIONS_SENT = "notifications_sent"
+ADDITIONAL_BUDGET_ADDED = "additional_budget_added"
+
+EVENT_TYPES = {
+    INVITE_SENT: _("Invite Sent"),
+    RECORDS_APPROVED: _("Records Approved"),
+    RECORDS_REJECTED: _("Records Rejected"),
+    PAYMENT_APPROVED: _("Payment Approved"),
+    PAYMENT_ACCRUED: _("Payment Accrued"),
+    PAYMENT_TRANSFERRED: _("Payment Transferred"),
+    NOTIFICATIONS_SENT: _("Notifications Sent"),
+    ADDITIONAL_BUDGET_ADDED: _("Additional Budget Added"),
+}
+
+EVENT_TYPE_CHOICES = list(EVENT_TYPES.items())
+
+
+# Inferred Events
+RECORDS_FLAGGED = "records_flagged"

commcare_connect/events/views.py

@@ -0,0 +1,34 @@
+from django.utils.decorators import method_decorator
+from django.views.decorators.csrf import csrf_exempt
+from rest_framework import serializers, status
+from rest_framework.generics import ListCreateAPIView
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.response import Response
+
+from .models import Event
+
+
+class EventSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Event
+        fields = ["date_created", "event_type", "user", "opportunity"]
+
+
+@method_decorator(csrf_exempt, name="dispatch")
+class EventListCreateView(ListCreateAPIView):
+    queryset = Event.objects.all()
+    serializer_class = EventSerializer
+    permission_classes = [IsAuthenticated]
+
+    def create(self, request, *args, **kwargs):
+        if not isinstance(request.data, list):
+            return Response({"error": "Expected a list of items"}, status=status.HTTP_400_BAD_REQUEST)
+
+        serializer = self.get_serializer(data=request.data, many=True)
+        serializer.is_valid(raise_exception=True)
+
+        event_objects = [Event(**item) for item in serializer.validated_data]
+        Event.objects.bulk_create(event_objects)
+
+        headers = self.get_success_headers(serializer.data)
+        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)

commcare_connect/opportunity/tasks.py

@@ -101,6 +101,12 @@ def invite_user(user_id, opportunity_access_id):
         ),
     )
     send_message(message)
+    from commcare_connect.events.models import Event
+
+    Event(event_type=Event.Type.INVITE_SENT, user=user, opportunity=opportunity_access.opportunity).track(
+        # this is already in async worker, so user_async is False
+        use_async=False
+    )
 
 
 @celery_app.task()