mirrors/posthog
0
0
Fork 0
mirror of https://github.com/PostHog/posthog.git synced 2024-11-24 09:14:46 +01:00
Code Issues Releases Activity

chore: make the timestamp field optional (looks like it's coming in with a variety of types) (#25368)

Browse Source
This commit is contained in:
James Greenhill 2024-10-03 09:31:50 -07:00 committed by GitHub
parent 8fd812cc5a
commit 33d28034bb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 30 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
livestream/filter.go
View File

@ -63,7 +63,7 @@ func convertToResponseGeoEvent(event PostHogEvent) *ResponseGeoEvent {
func convertToResponsePostHogEvent(event PostHogEvent, teamId int) *ResponsePostHogEvent {
return &ResponsePostHogEvent{
Uuid: event.Uuid,
Timestamp: event.Timestamp,
Timestamp: event.Timestamp.Value,
DistinctId: event.DistinctId,
PersonId: uuidFromDistinctId(teamId, event.DistinctId),
Event: event.Event,

6
livestream/filter_test.go
View File

@ -64,9 +64,10 @@ func TestConvertToResponseGeoEvent(t *testing.T) {
}
func TestConvertToResponsePostHogEvent(t *testing.T) {
timestamp := "2023-01-01T00:00:00Z"
event := PostHogEvent{
Uuid: "123",
Timestamp: "2023-01-01T00:00:00Z",
Timestamp: Timestamp{Value: timestamp},
DistinctId: "user1",
Event: "pageview",
Properties: map[string]interface{}{"url": "https://example.com"},
@ -108,9 +109,10 @@ func TestFilterRun(t *testing.T) {
time.Sleep(10 * time.Millisecond)
// Test event filtering
timestamp := "2023-01-01T00:00:00Z"
event := PostHogEvent{
Uuid: "123",
Timestamp: "2023-01-01T00:00:00Z",
Timestamp: Timestamp{Value: timestamp},
DistinctId: "user1",
Token: "token1",
Event: "pageview",

29
livestream/kafka.go
View File

@ -16,11 +16,32 @@ type PostHogEventWrapper struct {
Data string `json:"data"`
}
type Timestamp struct {
Value string
Raw string
}
func (t *Timestamp) UnmarshalJSON(data []byte) error {
t.Raw = string(data)
var s string
if err := json.Unmarshal(data, &s); err == nil {
t.Value = s
return nil
}
log.Printf("Unable to unmarshal timestamp to string. Raw value: %s", t.Raw)
// Set Default to empty string
t.Value = ""
return nil
}
type PostHogEvent struct {
Token string `json:"api_key,omitempty"`
Event string `json:"event"`
Properties map[string]interface{} `json:"properties"`
Timestamp string `json:"timestamp,omitempty"`
Timestamp Timestamp `json:"timestamp,omitempty"`
Uuid string
DistinctId string
@ -96,15 +117,15 @@ func (c *PostHogKafkaConsumer) Consume() {
var phEvent PostHogEvent
err = json.Unmarshal([]byte(wrapperMessage.Data), &phEvent)
if err != nil {
sentry.CaptureException(err)
log.Printf("Error decoding JSON: %v", err)
sentry.CaptureException(err)
continue
}
phEvent.Uuid = wrapperMessage.Uuid
phEvent.DistinctId = wrapperMessage.DistinctId
if phEvent.Timestamp == "" {
phEvent.Timestamp = time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
if phEvent.Timestamp.Value == "" {
phEvent.Timestamp.Value = time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
}
if phEvent.Token == "" {
if tokenValue, ok := phEvent.Properties["token"].(string); ok {