Merge branch 'brian/uc-33-kafka-optimizations' into dev

lib/session.js

@@ -1,8 +1,7 @@
-import { getWebsiteByUuid, getSessionByUuid, createSession } from 'queries';
+import { isValidUuid, parseToken, uuid } from 'lib/crypto';
-import { getJsonBody, getClientInfo } from 'lib/request';
+import redis from 'lib/redis';
-import { uuid, isValidUuid, parseToken } from 'lib/crypto';
+import { getClientInfo, getJsonBody } from 'lib/request';
-import { runAnalyticsQuery } from 'lib/db';
+import { createSession, getSessionByUuid, getWebsiteByUuid } from 'queries';
-import { RELATIONAL, CLICKHOUSE } from 'lib/constants';
 
 export async function getSession(req) {
   const { payload } = getJsonBody(req);
@@ -27,28 +26,42 @@ export async function getSession(req) {
     throw new Error(`Invalid website: ${website_uuid}`);
   }
 
-  const website = await getWebsiteByUuid(website_uuid);
+  let websiteId = null;
 
-  if (!website) {
+  // Check if website exists
+  if (process.env.REDIS_URL) {
+    websiteId = await redis.get(`website:${website_uuid}`);
+  } else {
+    const { website_id } = await getWebsiteByUuid(website_uuid);
+    websiteId = website_id;
+  }
+
+  if (!websiteId) {
     throw new Error(`Website not found: ${website_uuid}`);
   }
 
   const { userAgent, browser, os, ip, country, device } = await getClientInfo(req, payload);
 
-  const { website_id } = website;
+  const session_uuid = uuid(websiteId, hostname, ip, userAgent);
-  const session_uuid = uuid(website_id, hostname, ip, userAgent);
 
-  // logic placeholder for redis
+  let sessionCreated = false;
-  runAnalyticsQuery({
+  let sessionId = null;
-    [RELATIONAL]: () => {},
+  let session = null;
-    [CLICKHOUSE]: () => {},
-  });
 
-  let session = await getSessionByUuid(session_uuid);
+  // Check if session exists
+  if (process.env.REDIS_URL) {
+    sessionCreated = (await redis.get(`session:${session_uuid}`)) !== null;
+  } else {
+    console.log('test');
+    session = await getSessionByUuid(session_uuid);
+    sessionCreated = !!session;
+    sessionId = session ? session.session_id : null;
+  }
 
-  if (!session) {
+  if (!sessionCreated) {
     try {
-      session = await createSession(website_id, {
+      console.log('test2');
+      session = await createSession(websiteId, {
         session_uuid,
         hostname,
         browser,
@@ -58,6 +71,8 @@ export async function getSession(req) {
         country,
         device,
       });
+
+      sessionId = session ? session.session_id : null;
     } catch (e) {
       if (!e.message.toLowerCase().includes('unique constraint')) {
         throw e;
@@ -65,15 +80,9 @@ export async function getSession(req) {
     }
   }
 
-  if (!session) {
-    return null;
-  }
-
-  const { session_id } = session;
-
   return {
-    website_id,
+    website_id: websiteId,
-    session_id,
+    session_id: sessionId,
     session_uuid,
   };
 }

pages/api/heartbeat.js

@@ -1,16 +1,5 @@
-import { redis } from 'lib/redis';
 import { ok } from 'lib/response';
 
 export default async (req, res) => {
-  console.log('------------------------------------------------------------------------------');
+  return ok(res, 'nice');
-  // const redis = new createClient({
-  //   url: process.env.REDIS_URL,
-  // });
-
-  //console.log(redis.get);
-
-  const value = await redis.get('session:77c0b9de-677a-5268-8543-6018f0776a81');
-  console.log('complete');
-
-  return ok(res, value);
 };

queries/analytics/event/saveEvent.js

@@ -62,5 +62,5 @@ async function kafkaQuery(website_id, { event_uuid, session_uuid, url, event_nam
     event_name: event_name?.substr(0, 50),
   };
 
-  await kafka.sendKafkaMessage(params, 'event');
+  await kafka.sendMessage(params, 'event');
 }

queries/analytics/pageview/savePageView.js

@@ -50,5 +50,5 @@ async function kafkaQuery(website_id, { session_uuid, url, referrer }) {
     referrer: referrer?.substr(0, URL_LENGTH),
   };
 
-  await kafka.sendKafkaMessage(params, 'pageview');
+  await kafka.sendMessage(params, 'pageview');
 }

queries/analytics/session/createSession.js

@@ -73,7 +73,7 @@ async function kafkaQuery(
     country: country ? country : null,
   };
 
-  await kafka.sendKafkaMessage(params, 'session');
+  await kafka.sendMessage(params, 'session');
 
   await redis.set(`session:${session_uuid}`, '');
 }