Add connect methods to libraries.

lib/clickhouse.js

@@ -14,6 +14,9 @@ export const CLICKHOUSE_DATE_FORMATS = {
 
 const log = debug('umami:clickhouse');
 
+let clickhouse;
+const enabled = Boolean(process.env.CLICKHOUSE_URL);
+
 function getClient() {
   const {
     hostname,
@@ -144,6 +147,8 @@ async function rawQuery(query, params = []) {
     log(formattedQuery);
   }
 
+  await connect();
+
   return clickhouse.query(formattedQuery).toPromise();
 }
 
@@ -159,12 +164,19 @@ async function findFirst(data) {
   return data[0] ?? null;
 }
 
-// Initialization
-const clickhouse = process.env.CLICKHOUSE_URL && (global[CLICKHOUSE] || getClient());
+async function connect() {
+  if (!clickhouse) {
+    clickhouse = process.env.CLICKHOUSE_URL && (global[CLICKHOUSE] || getClient());
+  }
+
+  return clickhouse;
+}
 
 export default {
+  enabled,
   client: clickhouse,
   log,
+  connect,
   getDateStringQuery,
   getDateQuery,
   getDateFormat,

lib/kafka.js

@@ -5,6 +5,10 @@ import { KAFKA, KAFKA_PRODUCER } from 'lib/db';
 
 const log = debug('umami:kafka');
 
+let kafka;
+let producer;
+const enabled = Boolean(process.env.KAFKA_URL && process.env.KAFKA_BROKER);
+
 function getClient() {
   const { username, password } = new URL(process.env.KAFKA_URL);
   const brokers = process.env.KAFKA_BROKER.split(',');
@@ -61,7 +65,7 @@ function getDateFormat(date) {
 }
 
 async function sendMessage(params, topic) {
-  await getKafka();
+  await connect();
 
   await producer.send({
     topic,
@@ -74,7 +78,7 @@ async function sendMessage(params, topic) {
   });
 }
 
-async function getKafka() {
+async function connect() {
   if (!kafka) {
     kafka = process.env.KAFKA_URL && process.env.KAFKA_BROKER && (global[KAFKA] || getClient());
 
@@ -86,14 +90,12 @@ async function getKafka() {
   return kafka;
 }
 
-// Initialization
-let kafka;
-let producer;
-
 export default {
+  enabled,
   client: kafka,
   producer,
   log,
+  connect,
   getDateFormat,
   sendMessage,
 };

lib/redis.js

@@ -8,6 +8,9 @@ const log = debug('umami:redis');
 const INITIALIZED = 'redis:initialized';
 export const DELETED = 'deleted';
 
+let redis;
+const enabled = Boolean(process.env.REDIS_URL);
+
 function getClient() {
   if (!process.env.REDIS_URL) {
     return null;
@@ -40,26 +43,41 @@ async function stageData() {
     return { key: `website:${a.website_uuid}`, value: Number(a.website_id) };
   });
 
-  await addRedis(sessionUuids);
-  await addRedis(websiteIds);
+  await addSet(sessionUuids);
+  await addSet(websiteIds);
 
   await redis.set(INITIALIZED, 1);
 }
 
-async function addRedis(ids) {
+async function addSet(ids) {
   for (let i = 0; i < ids.length; i++) {
     const { key, value } = ids[i];
     await redis.set(key, value);
   }
 }
 
-// Initialization
-const redis = process.env.REDIS_URL && (global[REDIS] || getClient());
+async function get(key) {
+  await connect();
 
-(async () => {
-  if (redis && !(await redis.get(INITIALIZED))) {
-    await stageData();
+  return redis.get(key);
+}
+
+async function set(key, value) {
+  await connect();
+
+  return redis.set(key, value);
+}
+
+async function connect() {
+  if (!redis) {
+    process.env.REDIS_URL && (global[REDIS] || getClient());
+
+    if (!(await redis.get(INITIALIZED))) {
+      await stageData();
+    }
   }
-})();
 
-export default { client: redis, stageData, log };
+  return redis;
+}
+
+export default { enabled, client: redis, log, connect, get, set, stageData };

lib/session.js

@@ -1,14 +1,13 @@
 import { parseToken } from 'next-basics';
 import { validate } from 'uuid';
-import { uuid } from 'lib/crypto';
+import { secret, uuid } from 'lib/crypto';
 import redis, { DELETED } from 'lib/redis';
+import clickhouse from 'lib/clickhouse';
 import { getClientInfo, getJsonBody } from 'lib/request';
 import { createSession, getSessionByUuid, getWebsiteByUuid } from 'queries';
 
 export async function getSession(req) {
   const { payload } = getJsonBody(req);
-  const hasRedis = process.env.REDIS_URL;
-  const hasClickhouse = process.env.CLICKHOUSE_URL;
 
   if (!payload) {
     throw new Error('Invalid request');
@@ -17,7 +16,7 @@ export async function getSession(req) {
   const cache = req.headers['x-umami-cache'];
 
   if (cache) {
-    const result = await parseToken(cache);
+    const result = await parseToken(cache, secret());
 
     if (result) {
       return result;
@@ -33,8 +32,8 @@ export async function getSession(req) {
   let websiteId = null;
 
   // Check if website exists
-  if (hasRedis) {
-    websiteId = Number(await redis.client.get(`website:${website_uuid}`));
+  if (redis.enabled) {
+    websiteId = Number(await redis.get(`website:${website_uuid}`));
   }
 
   // Check database if does not exists in Redis
@@ -53,10 +52,10 @@ export async function getSession(req) {
   let sessionId = null;
   let session = null;
 
-  if (!hasClickhouse) {
+  if (!clickhouse.enabled) {
     // Check if session exists
-    if (hasRedis) {
-      sessionId = Number(await redis.client.get(`session:${session_uuid}`));
+    if (redis.enabled) {
+      sessionId = Number(await redis.get(`session:${session_uuid}`));
     }
 
     // Check database if does not exists in Redis

queries/admin/account/deleteAccount.js

@@ -41,7 +41,7 @@ export async function deleteAccount(user_id) {
     .then(async res => {
       if (redis.client) {
         for (let i = 0; i < websiteUuids.length; i++) {
-          await redis.client.set(`website:${websiteUuids[i]}`, DELETED);
+          await redis.set(`website:${websiteUuids[i]}`, DELETED);
         }
       }