1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
import clickhouse from '@/lib/clickhouse';
import { DATA_TYPE } from '@/lib/constants';
import { uuid } from '@/lib/crypto';
import { flattenJSON, getStringValue } from '@/lib/data';
import { CLICKHOUSE, PRISMA, runQuery } from '@/lib/db';
import kafka from '@/lib/kafka';
import prisma from '@/lib/prisma';
import type { DynamicData } from '@/lib/types';
export interface SaveEventDataArgs {
websiteId: string;
eventId: string;
sessionId?: string;
urlPath?: string;
eventName?: string;
eventData: DynamicData;
createdAt?: Date;
}
export async function saveEventData(data: SaveEventDataArgs) {
return runQuery({
[PRISMA]: () => relationalQuery(data),
[CLICKHOUSE]: () => clickhouseQuery(data),
});
}
async function relationalQuery(data: SaveEventDataArgs) {
const { websiteId, eventId, eventData, createdAt } = data;
const jsonKeys = flattenJSON(eventData);
// id, websiteEventId, eventStringValue
const flattenedData = jsonKeys.map(a => ({
id: uuid(),
websiteEventId: eventId,
websiteId,
dataKey: a.key,
stringValue: getStringValue(a.value, a.dataType),
numberValue: a.dataType === DATA_TYPE.number ? a.value : null,
dateValue: a.dataType === DATA_TYPE.date ? new Date(a.value) : null,
dataType: a.dataType,
createdAt,
}));
await prisma.client.eventData.createMany({
data: flattenedData,
});
}
async function clickhouseQuery(data: SaveEventDataArgs) {
const { websiteId, sessionId, eventId, urlPath, eventName, eventData, createdAt } = data;
const { insert, getUTCString } = clickhouse;
const { sendMessage } = kafka;
const jsonKeys = flattenJSON(eventData);
const messages = jsonKeys.map(({ key, value, dataType }) => {
return {
website_id: websiteId,
session_id: sessionId,
event_id: eventId,
url_path: urlPath,
event_name: eventName,
data_key: key,
data_type: dataType,
string_value: getStringValue(value, dataType),
number_value: dataType === DATA_TYPE.number ? value : null,
date_value: dataType === DATA_TYPE.date ? getUTCString(value) : null,
created_at: getUTCString(createdAt),
};
});
if (kafka.enabled) {
await sendMessage('event_data', messages);
} else {
await insert('event_data', messages);
}
}
|