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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
|
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 SaveSessionDataArgs {
websiteId: string;
sessionId: string;
sessionData: DynamicData;
distinctId?: string;
createdAt?: Date;
}
export async function saveSessionData(data: SaveSessionDataArgs) {
return runQuery({
[PRISMA]: () => relationalQuery(data),
[CLICKHOUSE]: () => clickhouseQuery(data),
});
}
export async function relationalQuery({
websiteId,
sessionId,
sessionData,
distinctId,
createdAt,
}: SaveSessionDataArgs) {
const { client } = prisma;
const jsonKeys = flattenJSON(sessionData);
const flattenedData = jsonKeys.map(a => ({
id: uuid(),
websiteId,
sessionId,
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,
distinctId,
createdAt,
}));
const existing = await client.sessionData.findMany({
where: {
sessionId,
},
select: {
id: true,
sessionId: true,
dataKey: true,
},
});
for (const data of flattenedData) {
const { sessionId, dataKey, ...props } = data;
const record = existing.find(e => e.sessionId === sessionId && e.dataKey === dataKey);
if (record) {
await client.sessionData.update({
where: {
id: record.id,
},
data: {
...props,
},
});
} else {
await client.sessionData.create({
data,
});
}
}
}
async function clickhouseQuery({
websiteId,
sessionId,
sessionData,
distinctId,
createdAt,
}: SaveSessionDataArgs) {
const { insert, getUTCString } = clickhouse;
const { sendMessage } = kafka;
const jsonKeys = flattenJSON(sessionData);
const messages = jsonKeys.map(({ key, value, dataType }) => {
return {
website_id: websiteId,
session_id: sessionId,
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,
distinct_id: distinctId,
created_at: getUTCString(createdAt),
};
});
if (kafka.enabled) {
await sendMessage('session_data', messages);
} else {
await insert('session_data', messages);
}
}
|