Feat/um 305 unique session ch (#2065)

* Add session_data / session redis to CH.

* Add mysql migration.

queries/analytics/session/createSession.ts

@@ -1,23 +1,8 @@
-import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
-import kafka from 'lib/kafka';
-import prisma from 'lib/prisma';
-import cache from 'lib/cache';
 import { Prisma } from '@prisma/client';
+import cache from 'lib/cache';
+import prisma from 'lib/prisma';
 
-export async function createSession(args: Prisma.SessionCreateInput) {
-  return runQuery({
-    [PRISMA]: () => relationalQuery(args),
-    [CLICKHOUSE]: () => clickhouseQuery(args),
-  }).then(async data => {
-    if (cache.enabled) {
-      await cache.storeSession(data);
-    }
-
-    return data;
-  });
-}
-
-async function relationalQuery(data: Prisma.SessionCreateInput) {
+export async function createSession(data: Prisma.SessionCreateInput) {
   const {
     id,
     websiteId,
@@ -33,71 +18,28 @@ async function relationalQuery(data: Prisma.SessionCreateInput) {
     city,
   } = data;
 
-  return prisma.client.session.create({
-    data: {
-      id,
-      websiteId,
-      hostname,
-      browser,
-      os,
-      device,
-      screen,
-      language,
-      country,
-      subdivision1: country && subdivision1 ? `${country}-${subdivision1}` : null,
-      subdivision2,
-      city,
-    },
-  });
-}
-
-async function clickhouseQuery(data: {
-  id: string;
-  websiteId: string;
-  hostname?: string;
-  browser?: string;
-  os?: string;
-  device?: string;
-  screen?: string;
-  language?: string;
-  country?: string;
-  subdivision1?: string;
-  subdivision2?: string;
-  city?: string;
-}) {
-  const {
-    id,
-    websiteId,
-    hostname,
-    browser,
-    os,
-    device,
-    screen,
-    language,
-    country,
-    subdivision1,
-    subdivision2,
-    city,
-  } = data;
-  const { getDateFormat, sendMessage } = kafka;
-
-  const msg = {
-    session_id: id,
-    website_id: websiteId,
-    hostname,
-    browser,
-    os,
-    device,
-    screen,
-    language,
-    country,
-    subdivision1,
-    subdivision2,
-    city,
-    created_at: getDateFormat(new Date()),
-  };
-
-  await sendMessage(msg, 'event');
-
-  return data;
+  return prisma.client.session
+    .create({
+      data: {
+        id,
+        websiteId,
+        hostname,
+        browser,
+        os,
+        device,
+        screen,
+        language,
+        country,
+        subdivision1: country && subdivision1 ? `${country}-${subdivision1}` : null,
+        subdivision2,
+        city,
+      },
+    })
+    .then(async data => {
+      if (cache.enabled) {
+        await cache.storeSession(data);
+      }
+
+      return data;
+    });
 }

queries/analytics/session/getSession.ts

@@ -1,43 +1,8 @@
-import clickhouse from 'lib/clickhouse';
-import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
-import prisma from 'lib/prisma';
 import { Prisma } from '@prisma/client';
+import prisma from 'lib/prisma';
 
-export async function getSession(args: { id: string }) {
-  return runQuery({
-    [PRISMA]: () => relationalQuery(args),
-    [CLICKHOUSE]: () => clickhouseQuery(args),
-  });
-}
-
-async function relationalQuery(where: Prisma.SessionWhereUniqueInput) {
+export async function getSession(where: Prisma.SessionWhereUniqueInput) {
   return prisma.client.session.findUnique({
     where,
   });
 }
-
-async function clickhouseQuery({ id: sessionId }: { id: string }) {
-  const { rawQuery, findFirst } = clickhouse;
-  const params = { sessionId };
-
-  return rawQuery(
-    `select
-      session_id, 
-      website_id, 
-      created_at, 
-      hostname, 
-      browser, 
-      os, 
-      device, 
-      screen,
-      language, 
-      country,
-      subdivision1,
-      subdivision2,
-      city
-    from website_event
-      where session_id = {sessionId:UUID}
-    limit 1`,
-    params,
-  ).then(result => findFirst(result));
-}

queries/analytics/session/saveSessionData.ts

@@ -0,0 +1,43 @@
+import { DYNAMIC_DATA_TYPE } from 'lib/constants';
+import { uuid } from 'lib/crypto';
+import { flattenJSON } from 'lib/dynamicData';
+import prisma from 'lib/prisma';
+import { DynamicData } from 'lib/types';
+
+export async function saveSessionData(data: {
+  websiteId: string;
+  sessionId: string;
+  sessionData: DynamicData;
+}) {
+  const { client, transaction } = prisma;
+  const { websiteId, sessionId, sessionData } = data;
+
+  const jsonKeys = flattenJSON(sessionData);
+
+  const flattendData = jsonKeys.map(a => ({
+    id: uuid(),
+    websiteId,
+    sessionId,
+    key: a.key,
+    stringValue:
+      a.dynamicDataType === DYNAMIC_DATA_TYPE.string ||
+      a.dynamicDataType === DYNAMIC_DATA_TYPE.boolean ||
+      a.dynamicDataType === DYNAMIC_DATA_TYPE.array
+        ? a.value
+        : null,
+    numericValue: a.dynamicDataType === DYNAMIC_DATA_TYPE.number ? a.value : null,
+    dateValue: a.dynamicDataType === DYNAMIC_DATA_TYPE.date ? new Date(a.value) : null,
+    dataType: a.dynamicDataType,
+  }));
+
+  return transaction([
+    client.sessionData.deleteMany({
+      where: {
+        sessionId,
+      },
+    }),
+    client.sessionData.createMany({
+      data: flattendData,
+    }),
+  ]);
+}