Refactored redis usage. Added lib/cache.

This commit is contained in:
Mike Cao 2022-11-07 22:35:51 -08:00
parent 3485b6268b
commit f118bc95c1
22 changed files with 236 additions and 221 deletions

View File

@ -1,6 +1,6 @@
import React from 'react';
import { useSpring, animated } from 'react-spring';
import { formatNumber } from '../../lib/format';
import { formatNumber } from 'lib/format';
import styles from './MetricCard.module.css';
const MetricCard = ({

View File

@ -5,7 +5,7 @@ import Button from 'components/common/Button';
import useLocale from 'hooks/useLocale';
import { DEFAULT_LOCALE } from 'lib/constants';
import styles from './TimezoneSetting.module.css';
import { languages } from '../../lib/lang';
import { languages } from 'lib/lang';
export default function LanguageSetting() {
const { locale, saveLocale } = useLocale();

View File

@ -0,0 +1,84 @@
import { getWebsite, getUser, getSession } from '../queries';
import redis, { DELETED } from 'lib/redis';
async function fetchObject(key, query) {
const obj = await redis.get(key);
if (!obj) {
return query().then(async data => {
if (data) {
await redis.set(key, data);
}
return data;
});
}
return obj;
}
async function storeObject(key, data) {
return redis.set(key, data);
}
async function deleteObject(key) {
return redis.set(key, DELETED);
}
async function fetchWebsite(id) {
return fetchObject(`website:${id}`, () => getWebsite({ id }));
}
async function storeWebsite(data) {
const { id } = data;
const key = `website:${id}`;
return storeObject(key, data);
}
async function deleteWebsite(id) {
return deleteObject(`website:${id}`);
}
async function fetchUser(id) {
return fetchObject(`user:${id}`, () => getUser({ id }));
}
async function storeUser(data) {
const { id } = data;
const key = `user:${id}`;
return storeObject(key, data);
}
async function deleteUser(id) {
return deleteObject(`user:${id}`);
}
async function fetchSession(id) {
return fetchObject(`session:${id}`, () => getSession({ id }));
}
async function storeSession(data) {
const { id } = data;
const key = `session:${id}`;
return storeObject(key, data);
}
async function deleteSession(id) {
return deleteObject(`session:${id}`);
}
export default {
fetchWebsite,
storeWebsite,
deleteWebsite,
fetchUser,
storeUser,
deleteUser,
fetchSession,
storeSession,
deleteSession,
enabled: redis.enabled,
};

View File

@ -1,23 +1,19 @@
import { createMiddleware, unauthorized, badRequest, serverError } from 'next-basics';
import { createMiddleware, unauthorized, badRequest } from 'next-basics';
import debug from 'debug';
import cors from 'cors';
import { getSession } from './session';
import { parseAuthToken, parseShareToken } from './auth';
import { findSession } from 'lib/session';
import { parseAuthToken, parseShareToken } from 'lib/auth';
import redis from 'lib/redis';
const log = debug('umami:middleware');
export const useCors = createMiddleware(cors());
export const useSession = createMiddleware(async (req, res, next) => {
let session;
try {
session = await getSession(req);
} catch (e) {
// eslint-disable-next-line no-console
console.error(e);
return serverError(res, e.message);
}
const session = await findSession(req);
if (!session) {
log('useSession:session-not-found');
return badRequest(res);
}
@ -29,10 +25,14 @@ export const useAuth = createMiddleware(async (req, res, next) => {
const token = await parseAuthToken(req);
const shareToken = await parseShareToken(req);
if (!token && !shareToken) {
const key = `auth:${token?.authKey}`;
const data = redis.enabled ? await redis.get(key) : token;
if (!data && !shareToken) {
log('useAuth:user-not-authorized');
return unauthorized(res);
}
req.auth = { ...token, shareToken };
req.auth = { ...data, shareToken };
next();
});

View File

@ -32,13 +32,13 @@ function getClient() {
async function get(key) {
await connect();
return redis.get(key);
return JSON.parse(await redis.get(key));
}
async function set(key, value) {
await connect();
return redis.set(key, value);
return redis.set(key, JSON.stringify(value));
}
async function del(key) {

View File

@ -1,102 +1,80 @@
import { parseToken } from 'next-basics';
import { validate } from 'uuid';
import { secret, uuid } from 'lib/crypto';
import redis, { DELETED } from 'lib/redis';
import clickhouse from 'lib/clickhouse';
import cache from 'lib/cache';
import { getClientInfo, getJsonBody } from 'lib/request';
import { createSession, getSession as getSessionPrisma, getWebsite } from 'queries';
import { createSession, getSession, getWebsite } from 'queries';
export async function getSession(req) {
export async function findSession(req) {
const { payload } = getJsonBody(req);
if (!payload) {
throw new Error('Invalid request');
return null;
}
const cache = req.headers['x-umami-cache'];
// Check if cache token is passed
const cacheToken = req.headers['x-umami-cache'];
if (cache) {
const result = await parseToken(cache, secret());
if (cacheToken) {
const result = await parseToken(cacheToken, secret());
if (result) {
return result;
}
}
// Verify payload
const { website: websiteId, hostname, screen, language } = payload;
if (!validate(websiteId)) {
return null;
}
let isValidWebsite = null;
// Find website
let website;
// Check if website exists
if (redis.enabled) {
isValidWebsite = await redis.get(`website:${websiteId}`);
if (cache.enabled) {
website = await cache.fetchWebsite(websiteId);
} else {
website = await getWebsite({ id: websiteId });
}
// Check database if does not exists in Redis
if (!isValidWebsite) {
const website = await getWebsite({ id: websiteId });
isValidWebsite = !!website;
}
if (!isValidWebsite || isValidWebsite === DELETED) {
if (!website || website.isDeleted) {
throw new Error(`Website not found: ${websiteId}`);
}
const { userAgent, browser, os, ip, country, device } = await getClientInfo(req, payload);
const sessionId = uuid(websiteId, hostname, ip, userAgent);
let isValidSession = null;
let session = null;
// Find session
let session;
if (!clickhouse.enabled) {
// Check if session exists
if (redis.enabled) {
isValidSession = await redis.get(`session:${sessionId}`);
}
// Check database if does not exists in Redis
if (!isValidSession) {
session = await getSessionPrisma({ id: sessionId });
isValidSession = !!session;
}
if (!isValidSession) {
try {
session = await createSession(websiteId, {
id: sessionId,
hostname,
browser,
os,
screen,
language,
country,
device,
});
} catch (e) {
if (!e.message.toLowerCase().includes('unique constraint')) {
throw e;
}
}
}
if (cache.enabled) {
session = await cache.fetchSession(sessionId);
} else {
session = {
id: sessionId,
hostname,
browser,
os,
screen,
language,
country,
device,
};
session = await getSession({ id: sessionId });
}
return {
websiteId,
session,
};
// Create a session if not found
if (!session) {
try {
session = await createSession({
id: sessionId,
websiteId,
hostname,
browser,
os,
device,
screen,
language,
country,
});
} catch (e) {
if (!e.message.toLowerCase().includes('unique constraint')) {
throw e;
}
}
}
return session;
}

View File

@ -84,7 +84,7 @@
"maxmind": "^4.3.6",
"moment-timezone": "^0.5.35",
"next": "^12.3.1",
"next-basics": "^0.20.0",
"next-basics": "^0.23.0",
"node-fetch": "^3.2.8",
"npm-run-all": "^4.1.5",
"prop-types": "^15.7.2",

View File

@ -5,7 +5,7 @@ import { createToken, unauthorized, send, badRequest, forbidden } from 'next-bas
import { savePageView, saveEvent } from 'queries';
import { useCors, useSession } from 'lib/middleware';
import { getJsonBody, getIpAddress } from 'lib/request';
import { secret, uuid } from 'lib/crypto';
import { secret } from 'lib/crypto';
export default async (req, res) => {
await useCors(req, res);
@ -58,9 +58,10 @@ export default async (req, res) => {
await useSession(req, res);
const { websiteId, session } = req.session;
const session = req.session;
const { type, payload } = getJsonBody(req);
const { websiteId } = session;
let { url, referrer, event_name: eventName, event_data: eventData } = payload;
@ -69,10 +70,9 @@ export default async (req, res) => {
}
if (type === 'pageview') {
await savePageView(websiteId, { pageViewId: uuid(), session, url, referrer });
await savePageView(websiteId, { session, url, referrer });
} else if (type === 'event') {
await saveEvent(websiteId, {
eventId: uuid(),
session,
url,
eventName,
@ -82,13 +82,7 @@ export default async (req, res) => {
return badRequest(res);
}
const token = createToken(
{
websiteId,
session,
},
secret(),
);
const token = createToken(session, secret());
return send(res, token);
};

View File

@ -1,5 +1,5 @@
import prisma from 'lib/prisma';
import redis, { DELETED } from 'lib/redis';
import cache from 'lib/cache';
export async function deleteUser(userId) {
const { client } = prisma;
@ -8,12 +8,6 @@ export async function deleteUser(userId) {
where: { userId },
});
let websiteIds = [];
if (websites.length > 0) {
websiteIds = websites.map(a => a.id);
}
return client
.$transaction([
client.pageview.deleteMany({
@ -37,13 +31,15 @@ export async function deleteUser(userId) {
},
}),
])
.then(async res => {
if (redis.enabled) {
for (let i = 0; i < websiteIds.length; i++) {
await redis.set(`website:${websiteIds[i]}`, DELETED);
.then(async data => {
if (cache.enabled) {
const ids = websites.map(a => a.id);
for (let i = 0; i < ids.length; i++) {
await cache.deleteWebsite(`website:${ids[i]}`);
}
}
return res;
return data;
});
}

View File

@ -1,5 +1,5 @@
import prisma from 'lib/prisma';
import redis from 'lib/redis';
import cache from 'lib/cache';
export async function createWebsite(userId, data) {
return prisma.client.website
@ -13,11 +13,11 @@ export async function createWebsite(userId, data) {
...data,
},
})
.then(async res => {
if (redis.enabled && res) {
await redis.set(`website:${res.id}`, res);
.then(async data => {
if (cache.enabled) {
await cache.storeWebsite(data);
}
return res;
return data;
});
}

View File

@ -1,5 +1,5 @@
import prisma from 'lib/prisma';
import redis, { DELETED } from 'lib/redis';
import cache from 'lib/cache';
export async function deleteWebsite(id) {
const { client, transaction } = prisma;
@ -20,11 +20,11 @@ export async function deleteWebsite(id) {
client.website.delete({
where: { id },
}),
]).then(async res => {
if (redis.enabled) {
await redis.set(`website:${id}`, DELETED);
]).then(async data => {
if (cache.enabled) {
await cache.deleteWebsite(id);
}
return res;
return data;
});
}

View File

@ -1,16 +1,7 @@
import prisma from 'lib/prisma';
import redis from 'lib/redis';
export async function getWebsite(where) {
return prisma.client.website
.findUnique({
where,
})
.then(async data => {
if (redis.enabled && data) {
await redis.set(`website:${data.id}`, data);
}
return data;
});
return prisma.client.website.findUnique({
where,
});
}

View File

@ -1,7 +1,7 @@
import clickhouse from 'lib/clickhouse';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import prisma from 'lib/prisma';
import redis from 'lib/redis';
import cache from 'lib/cache';
export async function getEventData(...args) {
return runQuery({
@ -41,7 +41,7 @@ async function relationalQuery(websiteId, { startDate, endDate, event_name, colu
async function clickhouseQuery(websiteId, { startDate, endDate, event_name, columns, filters }) {
const { rawQuery, getBetweenDates, getEventDataColumnsQuery, getEventDataFilterQuery } =
clickhouse;
const website = await redis.get(`website:${websiteId}`);
const website = await cache.fetchWebsite(websiteId);
const params = [websiteId, website?.revId || 0];
return rawQuery(

View File

@ -1,7 +1,7 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db';
import redis from 'lib/redis';
import cache from 'lib/cache';
export async function getEventMetrics(...args) {
return runQuery({
@ -47,7 +47,7 @@ async function clickhouseQuery(
filters = {},
) {
const { rawQuery, getDateQuery, getBetweenDates, getFilterQuery } = clickhouse;
const website = await redis.get(`website:${websiteId}`);
const website = await cache.fetchWebsite(websiteId);
const params = [websiteId, website?.revId || 0];
return rawQuery(

View File

@ -3,7 +3,7 @@ import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import kafka from 'lib/kafka';
import prisma from 'lib/prisma';
import { uuid } from 'lib/crypto';
import redis from 'lib/redis';
import cache from 'lib/cache';
export async function saveEvent(...args) {
return runQuery({
@ -12,52 +12,49 @@ export async function saveEvent(...args) {
});
}
async function relationalQuery(
websiteId,
{ eventId, session: { id: sessionId }, eventUuid, url, eventName, eventData },
) {
const data = {
async function relationalQuery(data) {
const { websiteId, sessionId, url, eventName, eventData } = data;
const eventId = uuid();
const params = {
id: eventId,
websiteId,
sessionId,
url: url?.substring(0, URL_LENGTH),
eventName: eventName?.substring(0, EVENT_NAME_LENGTH),
eventUuid,
};
if (eventData) {
data.eventData = {
params.eventData = {
create: {
id: eventId,
eventData: eventData,
id: uuid(),
},
};
}
return prisma.client.event.create({
data,
data: params,
});
}
async function clickhouseQuery(
websiteId,
{ session: { country, sessionUuid, ...sessionArgs }, eventUuid, url, eventName, eventData },
) {
async function clickhouseQuery(data) {
const { websiteId, sessionId, url, eventName, eventData } = data;
const { getDateFormat, sendMessage } = kafka;
const website = await redis.get(`website:${websiteId}`);
const website = await cache.fetchWebsite(websiteId);
const params = {
session_id: sessionUuid,
event_id: eventUuid,
website_id: websiteId,
rev_id: website?.revId || 0,
created_at: getDateFormat(new Date()),
session_id: sessionId,
event_id: uuid(),
url: url?.substring(0, URL_LENGTH),
event_name: eventName?.substring(0, EVENT_NAME_LENGTH),
event_data: eventData ? JSON.stringify(eventData) : null,
...sessionArgs,
country: country ? country : null,
rev_id: website?.revId || 0,
created_at: getDateFormat(new Date()),
};
await sendMessage(params, 'event');
return data;
}

View File

@ -1,7 +1,7 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db';
import redis from 'lib/redis';
import cache from 'lib/cache';
export async function getPageviewMetrics(...args) {
return runQuery({
@ -38,7 +38,7 @@ async function relationalQuery(websiteId, { startDate, endDate, column, table, f
async function clickhouseQuery(websiteId, { startDate, endDate, column, filters = {} }) {
const { rawQuery, parseFilters, getBetweenDates } = clickhouse;
const website = await redis.get(`website:${websiteId}`);
const website = await cache.fetchWebsite(websiteId);
const params = [websiteId, website?.revId || 0];
const { pageviewQuery, sessionQuery, eventQuery } = parseFilters(column, filters, params);

View File

@ -1,7 +1,7 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db';
import redis from 'lib/redis';
import cache from 'lib/cache';
export async function getPageviewStats(...args) {
return runQuery({
@ -52,7 +52,7 @@ async function clickhouseQuery(
{ start_at, end_at, timezone = 'UTC', unit = 'day', count = '*', filters = {} },
) {
const { parseFilters, rawQuery, getDateStringQuery, getDateQuery, getBetweenDates } = clickhouse;
const website = await redis.get(`website:${websiteId}`);
const website = await cache.fetchWebsite(websiteId);
const params = [websiteId, website?.revId || 0];
const { pageviewQuery, sessionQuery } = parseFilters(null, filters, params);

View File

@ -2,7 +2,8 @@ import { URL_LENGTH } from 'lib/constants';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import kafka from 'lib/kafka';
import prisma from 'lib/prisma';
import redis from 'lib/redis';
import cache from 'lib/cache';
import { uuid } from 'lib/crypto';
export async function savePageView(...args) {
return runQuery({
@ -11,13 +12,11 @@ export async function savePageView(...args) {
});
}
async function relationalQuery(
websiteId,
{ pageViewId, session: { id: sessionId }, url, referrer },
) {
async function relationalQuery(data) {
const { websiteId, sessionId, url, referrer } = data;
return prisma.client.pageview.create({
data: {
id: pageViewId,
id: uuid(),
websiteId,
sessionId,
url: url?.substring(0, URL_LENGTH),
@ -26,22 +25,21 @@ async function relationalQuery(
});
}
async function clickhouseQuery(
websiteId,
{ session: { country, id: sessionId, ...sessionArgs }, url, referrer },
) {
const website = await redis.get(`website:${websiteId}`);
async function clickhouseQuery(data) {
const { websiteId, sessionId, url, referrer } = data;
const website = await cache.fetchWebsite(websiteId);
const { getDateFormat, sendMessage } = kafka;
const params = {
const msg = {
session_id: sessionId,
website_id: websiteId,
created_at: getDateFormat(new Date()),
url: url?.substring(0, URL_LENGTH),
referrer: referrer?.substring(0, URL_LENGTH),
rev_id: website?.revId || 0,
...sessionArgs,
country: country ? country : null,
created_at: getDateFormat(new Date()),
};
await sendMessage(params, 'event');
await sendMessage(msg, 'event');
return data;
}

View File

@ -1,62 +1,45 @@
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import kafka from 'lib/kafka';
import prisma from 'lib/prisma';
import redis from 'lib/redis';
import cache from 'lib/cache';
export async function createSession(...args) {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
}).then(async data => {
if (redis.enabled && data) {
await redis.set(`session:${data.id}`, data);
if (cache.enabled) {
await cache.storeSession(data);
}
return data;
});
}
async function relationalQuery(websiteId, data) {
return prisma.client.session.create({
data: {
websiteId,
...data,
},
select: {
id: true,
hostname: true,
browser: true,
os: true,
screen: true,
language: true,
country: true,
device: true,
},
});
async function relationalQuery(data) {
return prisma.client.session.create({ data });
}
async function clickhouseQuery(
websiteId,
{ sessionId, hostname, browser, os, screen, language, country, device },
) {
async function clickhouseQuery(data) {
const { id, websiteId, hostname, browser, os, device, screen, language, country } = data;
const { getDateFormat, sendMessage } = kafka;
const website = await redis.get(`website:${websiteId}`);
const website = await cache.fetchWebsite(websiteId);
const data = {
sessionId,
const msg = {
session_id: id,
website_id: websiteId,
rev_id: website?.revId || 0,
created_at: getDateFormat(new Date()),
hostname,
browser,
os,
device,
screen,
language,
country: country ? country : null,
country,
rev_id: website?.revId || 0,
created_at: getDateFormat(new Date()),
};
await sendMessage(data, 'event');
await sendMessage(msg, 'event');
return data;
}

View File

@ -1,18 +1,11 @@
import clickhouse from 'lib/clickhouse';
import { CLICKHOUSE, PRISMA, runQuery } from 'lib/db';
import prisma from 'lib/prisma';
import redis from 'lib/redis';
export async function getSession(...args) {
return runQuery({
[PRISMA]: () => relationalQuery(...args),
[CLICKHOUSE]: () => clickhouseQuery(...args),
}).then(async data => {
if (redis.enabled && data) {
await redis.set(`session:${data.id}`, data);
}
return data;
});
}
@ -27,7 +20,7 @@ async function clickhouseQuery(sessionId) {
const params = [sessionId];
return rawQuery(
`select distinct
`select
session_id,
website_id,
created_at,
@ -39,7 +32,8 @@ async function clickhouseQuery(sessionId) {
language,
country
from event
where session_id = $1`,
where session_id = $1
limit 1`,
params,
).then(result => findFirst(result));
}

View File

@ -1,7 +1,7 @@
import prisma from 'lib/prisma';
import clickhouse from 'lib/clickhouse';
import { runQuery, CLICKHOUSE, PRISMA } from 'lib/db';
import redis from 'lib/redis';
import cache from 'lib/cache';
export async function getSessionMetrics(...args) {
return runQuery({
@ -37,7 +37,7 @@ async function relationalQuery(websiteId, { startDate, endDate, field, filters =
async function clickhouseQuery(websiteId, { startDate, endDate, field, filters = {} }) {
const { parseFilters, getBetweenDates, rawQuery } = clickhouse;
const website = await redis.get(`website:${websiteId}`);
const website = await cache.fetchWebsite(websiteId);
const params = [websiteId, website?.revId || 0];
const { pageviewQuery, sessionQuery } = parseFilters(null, filters, params);

View File

@ -5344,10 +5344,10 @@ natural-compare@^1.4.0:
resolved "https://registry.npmjs.org/natural-compare/-/natural-compare-1.4.0.tgz"
integrity sha1-Sr6/7tdUHywnrPspvbvRXI1bpPc=
next-basics@^0.20.0:
version "0.20.0"
resolved "https://registry.yarnpkg.com/next-basics/-/next-basics-0.20.0.tgz#f88791dcc538438b5044695e37d58b4fdccc6dd8"
integrity sha512-KPqVVSzkKUvu9shvZt5Bp7Xv1nZ2xJRRqwqQ+a6a5JjsdE10Q3p07VLrT2ykl+v/CvR4sz98c0n+MaWpgO3Ckw==
next-basics@^0.23.0:
version "0.23.0"
resolved "https://registry.yarnpkg.com/next-basics/-/next-basics-0.23.0.tgz#6ccc4b5f3cd87550b07d280c2a3bdee26f557912"
integrity sha512-AwKipMFkxS84ZKFDVSo8ZHXoOVzF2sPZp21pgtGlqf9aUmj48+QtxdM6ZnJVRg8Savf48ENjeYzeiu5I8Z4l/w==
dependencies:
base-x "^4.0.0"
bcryptjs "^2.4.3"