diff --git a/src/lib/constants.ts b/src/lib/constants.ts index 35917802d..406f119ba 100644 --- a/src/lib/constants.ts +++ b/src/lib/constants.ts @@ -143,6 +143,7 @@ export const REPORT_PARAMETERS = { export const KAFKA_TOPIC = { event: 'event', eventData: 'event_data', + eventDataBlob: 'event_data_blob', } as const; export const ROLES = { diff --git a/src/lib/data.ts b/src/lib/data.ts index cf2722b5c..68e8bc758 100644 --- a/src/lib/data.ts +++ b/src/lib/data.ts @@ -92,3 +92,37 @@ function getKeyName(key: string, parentKey: string) { export function objectToArray(obj: object) { return Object.keys(obj).map(key => obj[key]); } + +export function flattenDynamicData( + data: { key: string; value: any; dataType: DynamicDataType }[], +): { + blobs: string[]; + doubles: number[]; +} { + const blobs: string[] = []; + const doubles: number[] = []; + + data.forEach(({ value, dataType }) => { + switch (dataType) { + case DATA_TYPE.string: + blobs.push(value); + break; + case DATA_TYPE.number: + doubles.push(value); + break; + case DATA_TYPE.date: + doubles.push(new Date(value).getTime()); + break; + case DATA_TYPE.boolean: + doubles.push(value ? 1 : 0); + break; + case DATA_TYPE.array: + blobs.push(JSON.stringify(value)); + break; + default: + break; + } + }); + + return { blobs, doubles }; +} diff --git a/src/queries/analytics/eventData/saveEventData.ts b/src/queries/analytics/eventData/saveEventData.ts index 0cbc750e7..e18021c9f 100644 --- a/src/queries/analytics/eventData/saveEventData.ts +++ b/src/queries/analytics/eventData/saveEventData.ts @@ -2,7 +2,7 @@ import { Prisma } from '@prisma/client'; import { DATA_TYPE } from 'lib/constants'; import { uuid } from 'lib/crypto'; import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db'; -import { flattenJSON, getStringValue } from 'lib/data'; +import { flattenDynamicData, flattenJSON, getStringValue } from 'lib/data'; import kafka from 'lib/kafka'; import prisma from 'lib/prisma'; import { DynamicData } from 'lib/types'; @@ -61,7 +61,7 @@ async function clickhouseQuery(data: { }) { const { websiteId, sessionId, visitId, eventId, urlPath, eventName, eventData, createdAt } = data; - const { getDateFormat, sendMessages } = kafka; + const { getDateFormat, sendMessages, sendMessage } = kafka; const jsonKeys = flattenJSON(eventData); @@ -84,5 +84,22 @@ async function clickhouseQuery(data: { await sendMessages(messages, 'event_data'); + const jsonBlobs = flattenDynamicData(jsonKeys); + const message: { [key: string]: string | number } = { + website_id: websiteId, + session_id: sessionId, + event_id: eventId, + visitId: visitId, + event_name: eventName, + }; + jsonBlobs.blobs.forEach((blob, i) => { + message[`blob${i + 1}`] = blob; + }); + jsonBlobs.doubles.forEach((double, i) => { + message[`double${i + 1}`] = double; + }); + + await sendMessage(message, 'event_data_blob'); + return data; }