add event tracking

commcare_connect/events/models..py

@@ -0,0 +1,151 @@
+from django.db import models
+
+from commcare_connect.users.models import User
+
+
+class Event(models.Model):
+
+	class Type(models.TextChoices):
+	    INVITE_SENT = "is", gettext("Invite Sent")
+	    RECORDS_APPROVED = "rp", gettext("Records Approved")
+	    RECORDS_FLAGGED = "rf", gettext("Records Flagged")
+	    RECORDS_REJECTED = "rj", gettext("Records Rejected")
+	    PAYMENT_APPROVED = "pp", gettext("Payment Approved")
+	    PAYMENT_ACCRUED = "pa", gettext("Payment Accrued")
+	    PAYMENT_TRANSFERRED = "pf", gettext("Payment Transferred")
+	    NOTIFICATIONS_SENT = "nf", gettext("Notification Sent")
+	    ADDITIONAL_BUDGET_ADDED = "ab", gettext("Additional Budget Added")
+
+    date_created = models.DateTimeField(auto_now_add=True, db_index=True)
+    event_type = models.CharField(max_length='2', choices=Type.choices)
+    user = models.ForeignKey(User, on_delete=models.CASCADE, null=True)
+    opportunity = models.ForeignKey(Opportunity, on_delete=models.PROTECT, null=True)
+
+    @classmethod
+    def track(cls, 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:
+    user: User
+    opportunity: Opportunity
+    date_created: datetime
+    event_type: str
+
+
+class InferredEventSpec(ABCMeta):
+	"""
+	Use this to define an Event that can be inferred
+	based on other models.
+	"""
+
+	@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(event_filters)
+
+		if user:
+			filters.update({self.user: user})
+		if from_date:
+			filters.update({f"{self.date_created}__gte": from_date})
+		if from_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 = "RECORDS_FLAGGED", gettext("Records Flagged")
+	model = UserVisit
+	event_filters = {"flagged": True}
+	field_mapping = {
+		"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 = Events.objects.filter(**filters).all()
+	inferred_events = []
+	for event_spec in INFERRED_EVENT_SPECS:
+		inferred_events += event_spec.get_events()
+	return raw_events + inferred_events

commcare_connect/events/tasks.py

@@ -0,0 +1,37 @@
+import pickle
+from datetime import datetime
+
+from django.db import transaction
+from django_redis import get_redis_connection
+
+from config import celery_app
+
+from .models import Event
+
+REDIS_EVENTS_QUEUE = "events_queue"
+
+
+@celery_app.task
+def process_events_batch():
+    redis_conn = get_redis_connection("default")
+    events = redis_conn.lrange(REDIS_EVENTS_QUEUE, 0, -1)
+    if not events:
+        return
+
+    with transaction.atomic():
+        event_objs = []
+        for event in events:
+            event_objs.append(pickle.loads(event))
+        Event.objects.bulk_create(event_objs)
+
+    redis_conn.ltrim(REDIS_EVENTS_QUEUE, len(events), -1)
+
+
+def track_event(event_obj, use_async=True):
+    event_obj.date_created = datetime.now()
+    if use_async:
+        redis_conn = get_redis_connection("default")
+        serialized_event = pickle.dumps(event_obj)
+        redis_conn.rpush(REDIS_EVENTS_QUEUE, serialized_event)
+    else:
+        event_obj.save()

commcare_connect/opportunity/tasks.py

@@ -14,6 +14,7 @@
 
 from commcare_connect.connect_id_client import fetch_users, send_message, send_message_bulk
 from commcare_connect.connect_id_client.models import Message
+from commcare_connect.events.models import Event
 from commcare_connect.opportunity.app_xml import get_connect_blocks_for_app, get_deliver_units_for_app
 from commcare_connect.opportunity.export import (
     export_deliver_status_table,
@@ -101,6 +102,9 @@ def invite_user(user_id, opportunity_access_id):
         ),
     )
     send_message(message)
+    Event(event_type=Event.Type.INVITE_SENT, user=user, opportunity=opportunity_access.opportunity).track(
+        use_async=False
+    )
 
 
 @celery_app.task()

commcare_connect/opportunity/views.py

@@ -20,6 +20,7 @@
 from django_tables2.export import TableExport
 from geopy import distance
 
+from commcare_connect.events.models import Event
 from commcare_connect.form_receiver.serializers import XFormSerializer
 from commcare_connect.opportunity.forms import (
     AddBudgetExistingUsersForm,
@@ -382,6 +383,7 @@ def add_budget_existing_users(request, org_slug=None, pk=None):
                 opportunity.total_budget += ocl.payment_unit.amount * additional_visits
             opportunity.save()
             return redirect("opportunity:detail", org_slug, pk)
+    Event(event_type=Event.Type.ADDITIONAL_BUDGET_ADDED, opportunity=opportunity).track()
 
     return render(
         request,
@@ -750,6 +752,7 @@ def approve_visit(request, org_slug=None, pk=None):
     opp_id = user_visit.opportunity_id
     access = OpportunityAccess.objects.get(user_id=user_visit.user_id, opportunity_id=opp_id)
     update_payment_accrued(opportunity=access.opportunity, users=[access.user])
+    Event(event_type=Event.Type.RECORDS_APPROVED, user=user_visit.user, opportunity=access.opportunity).track()
     return redirect("opportunity:user_visits_list", org_slug=org_slug, opp_id=user_visit.opportunity.id, pk=access.id)
 
 
@@ -760,6 +763,7 @@ def reject_visit(request, org_slug=None, pk=None):
     user_visit.save()
     access = OpportunityAccess.objects.get(user_id=user_visit.user_id, opportunity_id=user_visit.opportunity_id)
     update_payment_accrued(opportunity=access.opportunity, users=[access.user])
+    Event(event_type=Event.Type.RECORDS_REJECTED, user=user_visit.user, opportunity=access.opportunity).track()
     return redirect("opportunity:user_visits_list", org_slug=org_slug, opp_id=user_visit.opportunity_id, pk=access.id)
 
 

commcare_connect/opportunity/visit_import.py

@@ -7,6 +7,7 @@
 from django.db import transaction
 from tablib import Dataset
 
+from commcare_connect.events.models import Event
 from commcare_connect.opportunity.models import (
     CompletedWork,
     CompletedWorkStatus,
@@ -143,6 +144,7 @@ def update_payment_accrued(opportunity: Opportunity, users):
                 access.payment_accrued += approved_count * completed_work.payment_unit.amount
             completed_work.save()
         access.save()
+        Event(event_type=Event.Type.PAYMENT_ACCRUED, user=access.user, opportunity=access.opportunity).track()
 
 
 def get_status_by_visit_id(dataset) -> dict[int, VisitValidationStatus]:
@@ -238,6 +240,7 @@ def _bulk_update_payments(opportunity: Opportunity, imported_data: Dataset) -> P
             payment = Payment.objects.create(opportunity_access=access, amount=amount)
             seen_users.add(username)
             payment_ids.append(payment.pk)
+            Event(event_type=Event.Type.PAYMENT_TRANSFERRED, user=access.user, opportunity=opportunity).track()
     missing_users = set(usernames) - seen_users
     send_payment_notification.delay(opportunity.id, payment_ids)
     return PaymentImportStatus(seen_users, missing_users)

config/settings/base.py

@@ -69,6 +69,7 @@
 
 LOCAL_APPS = [
     "commcare_connect.commcarehq_provider",
+    "commcare_connect.events",
     "commcare_connect.form_receiver",
     "commcare_connect.opportunity",
     "commcare_connect.organization",