Protobufize events to protect from malformed JSON (#2085)

* Protobuf all the things

* oops

* Protobufize events to protect from malformed JSON

* format the generated files (will need to remember this for future)

* format

* clean up kafka produce serializer

* fixes

ee/clickhouse/models/event.py

@@ -2,14 +2,15 @@
 import uuid
 from typing import Dict, List, Optional, Tuple, Union
 
+import pytz
 from dateutil.parser import isoparse
 from django.utils import timezone
 from rest_framework import serializers
 
 from ee.clickhouse.client import sync_execute
 from ee.clickhouse.models.element import chain_to_elements, elements_to_string
-from ee.clickhouse.models.util import cast_timestamp_or_now
 from ee.clickhouse.sql.events import GET_EVENTS_BY_TEAM_SQL, GET_EVENTS_SQL, INSERT_EVENT_SQL
+from ee.idl.gen import events_pb2  # type: ignore
 from ee.kafka_client.client import ClickhouseProducer
 from ee.kafka_client.topics import KAFKA_EVENTS
 from posthog.models.element import Element
@@ -26,24 +27,33 @@ def create_event(
     properties: Optional[Dict] = {},
     elements: Optional[List[Element]] = None,
 ) -> str:
-    timestamp = cast_timestamp_or_now(timestamp)
+
+    if not timestamp:
+        timestamp = timezone.now()
+    assert timestamp is not None
+
+    # clickhouse specific formatting
+    if isinstance(timestamp, str):
+        timestamp = isoparse(timestamp)
+    else:
+        timestamp = timestamp.astimezone(pytz.utc)
 
     elements_chain = ""
     if elements and len(elements) > 0:
         elements_chain = elements_to_string(elements=elements)
 
-    data = {
-        "uuid": str(event_uuid),
-        "event": event,
-        "properties": json.dumps(properties),
-        "timestamp": timestamp,
-        "team_id": team.pk,
-        "distinct_id": distinct_id,
-        "created_at": timestamp,
-        "elements_chain": elements_chain,
-    }
+    pb_event = events_pb2.Event()
+    pb_event.uuid = str(event_uuid)
+    pb_event.event = event
+    pb_event.properties = json.dumps(properties)
+    pb_event.timestamp = timestamp.strftime("%Y-%m-%d %H:%M:%S.%f")
+    pb_event.team_id = team.pk
+    pb_event.distinct_id = str(distinct_id)
+    pb_event.elements_chain = elements_chain
+
     p = ClickhouseProducer()
-    p.produce(sql=INSERT_EVENT_SQL, topic=KAFKA_EVENTS, data=data)
+
+    p.produce_proto(sql=INSERT_EVENT_SQL, topic=KAFKA_EVENTS, data=pb_event)
     return str(event_uuid)
 
 

ee/idl/element.proto

@@ -0,0 +1,19 @@
+syntax = "proto3";
+import "google/protobuf/timestamp.proto";
+
+message Element {
+  string uuid = 1;
+  string event_uuid = 2;
+  string text = 3;
+  string tag_name = 4;
+  string href = 5;
+  string attr_id = 6;
+  repeated string attr_class = 7;
+  uint64 nth_child = 8;
+  uint64 nth_of_type = 9;
+  string attributes = 10;
+  uint64 order = 11;
+  uint64 team_id = 12;
+  google.protobuf.Timestamp created_at = 13;
+  string elements_hash = 14;
+}

ee/idl/events.proto

@@ -0,0 +1,15 @@
+syntax = "proto3";
+import "google/protobuf/timestamp.proto";
+
+message Event {
+  string uuid = 1;
+  string event = 2;
+  string properties = 3;
+  string timestamp = 4;
+  uint64 team_id = 5;
+  string distinct_id = 6;
+  string created_at = 7;
+  string elements_chain = 8;
+  google.protobuf.Timestamp proto_created_at = 9;
+  google.protobuf.Timestamp proto_timestamp = 10;
+}