Migrate process_event shared functions to be public (#1625)

* Migrate process_event shared functions to be public

* mypy and adjust

* exclude tests that are tagged 'skip_on_multitenancy'

.github/workflows/ci-backend.yml

@@ -143,6 +143,15 @@ jobs:
         runs-on: ubuntu-latest
 
         services:
+            redis:
+                # Docker Hub image
+                image: redis
+                # Set health checks to wait until redis has started
+                options: >-
+                    --health-cmd "redis-cli ping"
+                    --health-interval 10s
+                    --health-timeout 5s
+                    --health-retries 5
             postgres:
                 image: postgres:12
                 env:
@@ -209,4 +218,4 @@ jobs:
                   touch frontend/dist/index.html
                   touch frontend/dist/layout.html
                   touch frontend/dist/shared_dashboard.html
-                  python manage.py test posthog --keepdb -v 2
+                  python manage.py test posthog --keepdb -v 2 --exclude-tag=skip_on_multitenancy

posthog/tasks/process_event.py

@@ -61,7 +61,7 @@ def _alias(previous_distinct_id: str, distinct_id: str, team_id: int, retry_if_f
         new_person.merge_people([old_person])
 
 
-def _store_names_and_properties(team: Team, event: str, properties: Dict) -> None:
+def store_names_and_properties(team: Team, event: str, properties: Dict) -> None:
     # In _capture we only prefetch a couple of fields in Team to avoid fetching too much data
     save = False
     if event not in team.event_names:
@@ -130,14 +130,22 @@ def _capture(
         **({"timestamp": timestamp} if timestamp else {}),
         **({"elements": elements_list} if elements_list else {})
     )
-    _store_names_and_properties(team=team, event=event, properties=properties)
+    store_names_and_properties(team=team, event=event, properties=properties)
+    check_and_create_person(team_id=team_id, distinct_id=distinct_id)
 
+
+def check_and_create_person(team_id: int, distinct_id: str) -> Optional[Person]:
+    person: Optional[Person]
     if not Person.objects.distinct_ids_exist(team_id=team_id, distinct_ids=[str(distinct_id)]):
         # Catch race condition where in between getting and creating, another request already created this user.
         try:
-            Person.objects.create(team_id=team_id, distinct_ids=[str(distinct_id)])
+            person = Person.objects.create(team_id=team_id, distinct_ids=[str(distinct_id)])
         except IntegrityError:
-            pass
+            return None
+        return person
+    else:
+        person = Person.objects.get(team_id=team_id, persondistinctid__distinct_id=str(distinct_id))
+        return person
 
 
 def _update_person_properties(team_id: int, distinct_id: str, properties: Dict) -> None:
@@ -167,7 +175,7 @@ def _set_is_identified(team_id: int, distinct_id: str, is_identified: bool = Tru
         person.save()
 
 
-def _handle_timestamp(data: dict, now: str, sent_at: Optional[str]) -> Union[datetime.datetime, str]:
+def handle_timestamp(data: dict, now: str, sent_at: Optional[str]) -> Union[datetime.datetime, str]:
     if data.get("timestamp"):
         if sent_at:
             # sent_at - timestamp == now - x
@@ -212,5 +220,5 @@ def process_event(
         event=data["event"],
         distinct_id=distinct_id,
         properties=properties,
-        timestamp=_handle_timestamp(data, now, sent_at),
+        timestamp=handle_timestamp(data, now, sent_at),
     )